feat: 질문 풀·피드백 생성 진행 이벤트 (AI 직접 발행 + 프론트 표시)

ai/CLAUDE.md

@@ -397,4 +397,13 @@ docker run --env-file .env -p 8000:8000 stackup-ai
   하드 타임아웃을 추가하고, 모든 `ChatOpenAI` 호출에 `llm_pro_timeout_sec`(30s)/`llm_flash_timeout_sec`
   (10s) 요청 타임아웃을 명시했다(이전엔 미설정 — SDK 기본값까지 무기한 대기 가능).
 
+- **질문 풀·피드백 생성 진행 이벤트 본 구현 (B2)**: 스트리밍이 없는 두 블로킹 생성 경로(질문 풀 Pro ≤30s,
+  피드백 병렬 gather ≈2분 예산)가 진행 중 무통보였던 것을 고쳤다. `SessionRealtimeNotifier.emit_progress`
+  (`messaging/session_notify.py`)가 `realtime.session.notify` 로 `QUESTION_POOL_PROGRESS`/`FEEDBACK_PROGRESS`
+  를 직접 발행(휘발성, 실패는 경고만). `questions_consumer` 는 CONTEXT_BUILDING→GENERATING→FINALIZING 순차
+  3단계, `feedback_consumer` 는 최상위가 `asyncio.gather` 병렬이라 순차 phase 대신 **태스크 완료 카운터**
+  (SCORING `completed/total`, 각 세부 평가 완료 시 emit)로 표현한다. 기존 `AnalysisProgressNotifier`
+  (user 채널)는 무변경 — user 채널 경로 불변은 회귀 테스트(`tests/test_progress.py`)로 고정.
+  와이어링은 `runner.py` 에서 followup 과 동일 notifier 인스턴스 재사용. 스펙: `docs/messaging.md §5.12-3`.
+
 각 도입 시 본 문서 갱신.

ai/src/ai_server/CLAUDE.md

@@ -74,7 +74,7 @@ class QuestionPoolCallbackPayload(BaseModel):
   (resume/repository/web/cover_letter/questions/followup/feedback/voice/tts)
 - 조립·기동은 `runner.py` 의 `MessagingRuntime` (§3), 연결은 `connection.py`,
   콜백 발행은 `publisher.py`, 멱등은 `idempotency.py`(`LruIdempotencyStore`),
-  RealTime 직접 발행은 `progress.py`(분석 진행)·`session_notify.py`(델타/오디오)
+  RealTime 직접 발행은 `progress.py`(분석 진행, user 채널)·`session_notify.py`(델타/오디오/질문 풀·피드백 생성 진행, 세션 채널)
 - 모든 consumer는 envelope parsing → trace_context → 비즈니스 핸들러 호출 패턴
 
 ```python

ai/src/ai_server/messaging/consumers/feedback_consumer.py

@@ -2,6 +2,8 @@
 
 import asyncio
 import re
+from collections.abc import Awaitable
+from typing import TypeVar
 
 import structlog
 from aio_pika.abc import AbstractIncomingMessage
@@ -26,6 +28,10 @@
 from ai_server.core.client import CoreClient
 from ai_server.messaging.idempotency import LruIdempotencyStore
 from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.messaging.session_notify import (
+    FEEDBACK_PROGRESS_EVENT,
+    SessionRealtimeNotifier,
+)
 from ai_server.model.envelope import Envelope
 from ai_server.model.messages.feedback import (
     AnswerCoachingItem,
@@ -39,6 +45,8 @@
 
 log = structlog.get_logger(__name__)
 
+T = TypeVar("T")
+
 _SELF_INTRO_CATEGORY = "SELF_INTRODUCTION"
 _JOB_TAILORED_MODE = "JOB_TAILORED"
 _BEHAVIORAL_CATEGORY = "BEHAVIORAL"
@@ -77,6 +85,7 @@ def __init__(
         answer_coach: AnswerCoach | None = None,
         coaching_max_answers: int = 30,
         coaching_concurrency: int = 5,
+        session_notifier: SessionRealtimeNotifier | None = None,
     ) -> None:
         self._generator = generator
         self._publisher = publisher
@@ -91,6 +100,7 @@ def __init__(
         self._answer_coach = answer_coach
         self._coaching_max_answers = coaching_max_answers
         self._coaching_concurrency = max(1, coaching_concurrency)
+        self._session_notifier = session_notifier
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -124,13 +134,47 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 trace_id=envelope.trace_id,
             )
 
+            await self._emit_progress(
+                session_id=req.session_id,
+                phase="PREPARING",
+                message="면접 기록을 정리하고 있어요.",
+                trace_id=envelope.trace_id,
+            )
             transcript = _build_transcript(req.messages)
             score_basis = _build_score_basis(req.messages)
             rag_context = await self._build_rag_context(req)
             voice_analysis_summary = _build_voice_analysis_summary(
                 req.voice_analysis_summary
             )
 
+            # 세부 평가 5개가 병렬(gather)이라 순차 phase 로는 진행을 표현할 수 없다 —
+            # 각 태스크 완료 시점에 completed/total 카운터로 emit 한다.
+            scoring_total = 5
+            scoring_done = 0
+
+            async def _tracked(coro: Awaitable[T]) -> T:
+                nonlocal scoring_done
+                task_result = await coro
+                scoring_done += 1
+                await self._emit_progress(
+                    session_id=req.session_id,
+                    phase="SCORING",
+                    message=f"세부 평가를 진행하고 있어요. ({scoring_done}/{scoring_total})",
+                    trace_id=envelope.trace_id,
+                    completed=scoring_done,
+                    total=scoring_total,
+                )
+                return task_result
+
+            await self._emit_progress(
+                session_id=req.session_id,
+                phase="SCORING",
+                message="평가위원들이 답변을 검토하고 있어요.",
+                trace_id=envelope.trace_id,
+                completed=0,
+                total=scoring_total,
+            )
+
             # 종합 피드백 + 자기소개 첫인상 + 직무 적합도(직무 맞춤 모드)를 병렬 실행.
             # 첫인상·직무 적합도는 종합 점수(overall)에 미포함 — generator 가 모른 채 계산한 뒤 표시용으로 덧붙인다.
             (
@@ -140,29 +184,37 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 personality_item,
                 answer_coaching,
             ) = await asyncio.gather(
-                self._generate_panel(
-                    job_category=req.job_category,
-                    mode=req.mode,
-                    total_question_count=req.total_question_count,
-                    end_reason=req.end_reason,
-                    transcript=transcript,
-                    score_basis=score_basis,
-                    rag_context=rag_context,
-                    voice_analysis_summary=voice_analysis_summary,
-                    domain_question_counts=req.domain_question_counts,
-                    session_id=req.session_id,
+                _tracked(
+                    self._generate_panel(
+                        job_category=req.job_category,
+                        mode=req.mode,
+                        total_question_count=req.total_question_count,
+                        end_reason=req.end_reason,
+                        transcript=transcript,
+                        score_basis=score_basis,
+                        rag_context=rag_context,
+                        voice_analysis_summary=voice_analysis_summary,
+                        domain_question_counts=req.domain_question_counts,
+                        session_id=req.session_id,
+                    )
                 ),
-                self._evaluate_self_intro(req, voice_analysis_summary),
-                self._evaluate_job_fit(req, transcript, rag_context),
-                self._evaluate_personality(req),
-                self._coach_answers(req),
+                _tracked(self._evaluate_self_intro(req, voice_analysis_summary)),
+                _tracked(self._evaluate_job_fit(req, transcript, rag_context)),
+                _tracked(self._evaluate_personality(req)),
+                _tracked(self._coach_answers(req)),
             )
             # 빈 평가위원 항목(점수·내용 모두 없음)은 표시하지 않는다 — LLM 부분 응답이 빈 패널로 새는 것 방지.
             extras = [self_intro_item, *job_fit_items, personality_item]
             result.panel_breakdown.extend(
                 e for e in extras if e is not None and _panel_has_content(e)
             )
 
+            await self._emit_progress(
+                session_id=req.session_id,
+                phase="FINALIZING",
+                message="피드백 리포트를 정리하고 있어요.",
+                trace_id=envelope.trace_id,
+            )
             payload = FeedbackCallbackPayload(
                 session_id=req.session_id,
                 overall_score=result.overall_score,
@@ -194,6 +246,28 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 trace_id=envelope.trace_id,
             )
 
+    async def _emit_progress(
+        self,
+        *,
+        session_id: int,
+        phase: str,
+        message: str,
+        trace_id: str,
+        completed: int | None = None,
+        total: int | None = None,
+    ) -> None:
+        if self._session_notifier is None:
+            return
+        await self._session_notifier.emit_progress(
+            event_type=FEEDBACK_PROGRESS_EVENT,
+            session_id=session_id,
+            phase=phase,
+            message=message,
+            trace_id=trace_id,
+            completed=completed,
+            total=total,
+        )
+
     async def _generate_panel(
         self,
         *,

ai/src/ai_server/messaging/consumers/questions_consumer.py

@@ -9,6 +9,10 @@
 from ai_server.core.client import CoreClient
 from ai_server.messaging.idempotency import LruIdempotencyStore
 from ai_server.messaging.publisher import CallbackPublisher
+from ai_server.messaging.session_notify import (
+    QUESTION_POOL_PROGRESS_EVENT,
+    SessionRealtimeNotifier,
+)
 from ai_server.model.envelope import Envelope
 from ai_server.model.messages.questions import (
     DocumentContext,
@@ -33,6 +37,7 @@ def __init__(
         embedder: EmbeddingProvider | None = None,
         rag_top_k: int = 5,
         rag_timeout_sec: float = 1.5,
+        session_notifier: SessionRealtimeNotifier | None = None,
     ) -> None:
         self._generator = generator
         self._publisher = publisher
@@ -45,6 +50,7 @@ def __init__(
         self._embedder = embedder
         self._rag_top_k = rag_top_k
         self._rag_timeout_sec = rag_timeout_sec
+        self._session_notifier = session_notifier
 
     async def handle(self, message: AbstractIncomingMessage) -> None:
         async with message.process(requeue=False):
@@ -83,10 +89,30 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 trace_id=envelope.trace_id,
             )
 
+            await self._emit_progress(
+                session_id=req.session_id,
+                phase="CONTEXT_BUILDING",
+                message="면접 자료를 정리하고 있어요.",
+                trace_id=envelope.trace_id,
+            )
             context_text = await self._build_context(req)
+            await self._emit_progress(
+                session_id=req.session_id,
+                phase="GENERATING",
+                message="자료를 바탕으로 첫 질문을 만들고 있어요.",
+                trace_id=envelope.trace_id,
+            )
             payload = await self._generate_pool_payload(
                 req, context_text, effective_pool_size, trace_id=envelope.trace_id
             )
+            # 생성 실패(FAILED 콜백) 직전에 "마무리하고 있어요" 가 스치면 오해를 부른다 — 성공시에만.
+            if payload.status != "FAILED":
+                await self._emit_progress(
+                    session_id=req.session_id,
+                    phase="FINALIZING",
+                    message="질문 준비를 마무리하고 있어요.",
+                    trace_id=envelope.trace_id,
+                )
 
             await self._publisher.publish(
                 routing_key=self._callback_routing_key,
@@ -105,6 +131,19 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 trace_id=envelope.trace_id,
             )
 
+    async def _emit_progress(
+        self, *, session_id: int, phase: str, message: str, trace_id: str
+    ) -> None:
+        if self._session_notifier is None:
+            return
+        await self._session_notifier.emit_progress(
+            event_type=QUESTION_POOL_PROGRESS_EVENT,
+            session_id=session_id,
+            phase=phase,
+            message=message,
+            trace_id=trace_id,
+        )
+
     async def _generate_pool_payload(
         self,
         req: GenerateQuestionsRequest,

ai/src/ai_server/messaging/runner.py

@@ -197,6 +197,12 @@ def __init__(self, settings: Settings) -> None:
             progress_notifier=self._progress_notifier,
         )
 
+        # 세션 채널 휘발성 이벤트 발행기 — 꼬리질문 델타/TTS 와 질문 풀·피드백 진행 이벤트가 공유.
+        session_notifier = SessionRealtimeNotifier(
+            publisher=self._realtime_publisher,
+            routing_key="realtime.session.notify",
+        )
+
         # 질문 풀 생성 (US-18)
         question_generator = LlmQuestionGenerator(
             build_question_generation_chain(settings, core_client=core_client)
@@ -210,6 +216,7 @@ def __init__(self, settings: Settings) -> None:
             core_client=core_client,
             embedder=embedder,
             rag_timeout_sec=settings.questions_rag_timeout_sec,
+            session_notifier=session_notifier,
         )
 
         # 꼬리질문 생성 (US-19)
@@ -219,10 +226,6 @@ def __init__(self, settings: Settings) -> None:
         streaming_followup_generator = build_streaming_followup_generator(
             settings, core_client=core_client
         )
-        session_notifier = SessionRealtimeNotifier(
-            publisher=self._realtime_publisher,
-            routing_key="realtime.session.notify",
-        )
         # TTS provider 는 꼬리질문 인라인 세그먼트 합성과 질문 TTS 양쪽에서 재사용한다.
         tts = build_tts_provider(settings)
         self._followup_consumer = FollowupConsumer(
@@ -272,6 +275,7 @@ def __init__(self, settings: Settings) -> None:
             answer_coach=LlmAnswerCoach(
                 build_answer_coaching_chain(settings, core_client=core_client)
             ),
+            session_notifier=session_notifier,
         )
 
         # 음성 답변 STT + 분석 (Phase 2)

ai/src/ai_server/messaging/session_notify.py

@@ -9,12 +9,16 @@
     SessionMessageAudioData,
     SessionMessageDeltaData,
     SessionNotifyPayload,
+    SessionProgressData,
+    SessionProgressNotifyPayload,
 )
 
 log = structlog.get_logger(__name__)
 
 SESSION_MESSAGE_DELTA_EVENT = "SESSION_MESSAGE_DELTA"
 SESSION_MESSAGE_AUDIO_EVENT = "SESSION_MESSAGE_AUDIO"
+QUESTION_POOL_PROGRESS_EVENT = "QUESTION_POOL_PROGRESS"
+FEEDBACK_PROGRESS_EVENT = "FEEDBACK_PROGRESS"
 
 
 class SessionRealtimeNotifier:
@@ -91,3 +95,42 @@ async def emit_audio(
                 seq=seq,
                 trace_id=trace_id,
             )
+
+    async def emit_progress(
+        self,
+        *,
+        event_type: str,
+        session_id: int,
+        phase: str,
+        message: str,
+        trace_id: str,
+        completed: int | None = None,
+        total: int | None = None,
+    ) -> None:
+        payload = SessionProgressNotifyPayload(
+            event_type=event_type,
+            data=SessionProgressData(
+                session_id=session_id,
+                phase=phase,
+                message=message,
+                completed=completed,
+                total=total,
+            ),
+        )
+        try:
+            await self._publisher.publish(
+                routing_key=self._routing_key,
+                message_type=self._routing_key,
+                payload=payload,
+                trace_id=trace_id,
+                correlation_id=f"progress-{session_id}-{phase}-{completed or 0}",
+                context=MessageContext(session_id=session_id),
+            )
+        except Exception:
+            log.warning(
+                "session.progress.publish_failed",
+                session_id=session_id,
+                event_type=event_type,
+                phase=phase,
+                trace_id=trace_id,
+            )