diff --git a/ai/CLAUDE.md b/ai/CLAUDE.md index edf12a5d..214efcbb 100644 --- a/ai/CLAUDE.md +++ b/ai/CLAUDE.md @@ -323,6 +323,8 @@ docker run --env-file .env -p 8000:8000 stackup-ai - 질문 풀 생성 (Pro 모델, `chain/question_generation_chain.py`) - 꼬리질문 + 답변 평가 (Flash 모델, `chain/followup_generation_chain.py`) - 콜백: `callback.questions` (`kind=POOL|FOLLOWUP`) +- **꼬리질문 토큰 스트리밍 본 구현**: followup 출력을 `……{json}` 구분자 포맷으로 바꾸고(`chain/prompts/followup_generation.py`), `StreamingFollowupGenerator`(`astream`)가 `` 토큰만 `SessionRealtimeNotifier`(`messaging/session_notify.py`)로 `SESSION_MESSAGE_DELTA` 발행(`stackup.realtime`/`realtime.session.notify`, Core 우회). `DONT_KNOW` 면 델타 미발행. 종료 후 `parse_followup_result` 로 검증해 기존 `callback.questions(FOLLOWUP, followupMessageId)` 발행. 와이어링은 `messaging/runner.py`(분석 진행 publisher 재사용). +- **문장 단위 TTS 본 구현 (Part B)**: followup consumer 스트림 루프가 `chain/sentence_split.next_sentences` 로 문장 경계를 잡아, 문장마다 `TtsProvider` 인라인 합성(`asyncio.create_task` 백그라운드, 텍스트 델타 비차단)→S3 `interview/tts/{sid}/{mid}/seg-{seq}.{ext}` PUT→`SessionRealtimeNotifier.emit_audio`(`SESSION_MESSAGE_AUDIO`). 콜백 전 `gather` 로 수거. 라이브 세그먼트는 휘발성(DB 미기록). - **임베딩 본 구현** (`rag/`): `MarkdownChunker` + `GeminiEmbeddingProvider` (1536d, `gemini-embedding-001`). 운영/개발 default 는 gemini, 테스트는 `MockEmbeddingProvider`. - 청크를 `EMBEDDING_BATCH_SIZE`(기본 32) 단위로 쪼개 순차 호출 — 한 요청에 몰면 분당 토큰 한도(429 `RESOURCE_EXHAUSTED`)에 걸린다. diff --git a/ai/src/ai_server/chain/followup_generation_chain.py b/ai/src/ai_server/chain/followup_generation_chain.py index 54ffc8d5..ae191101 100644 --- a/ai/src/ai_server/chain/followup_generation_chain.py +++ b/ai/src/ai_server/chain/followup_generation_chain.py @@ -1,6 +1,8 @@ from __future__ import annotations -from typing import Literal, Protocol +import re +from collections.abc import Awaitable, Callable +from typing import Any, Literal, Protocol from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate @@ -16,11 +18,123 @@ class FollowupResult(BaseModel): followup_question: str = Field(..., description="한국어 꼬리질문 1개") - answer_evaluation: AnswerEvaluation + answer_evaluation: AnswerEvaluation | None = None # 답변 의도. NORMAL=정상답변, DONT_KNOW=모름/포기, CLARIFICATION=질문 재설명 요청. answer_intent: Literal["NORMAL", "DONT_KNOW", "CLARIFICATION"] = "NORMAL" +_INTENT_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL) +_QUESTION_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL) +_META_RE = re.compile(r"\s*(\{.*?\})\s*", re.DOTALL) +_VALID_INTENTS = {"NORMAL", "DONT_KNOW", "CLARIFICATION"} + + +def extract_question_span(text: str) -> str: + """완성 또는 진행 중 텍스트에서 안쪽만 추출(닫는 태그 없어도 가능).""" + closed = _QUESTION_RE.search(text) + if closed: + return closed.group(1).strip() + open_idx = text.find("") + if open_idx == -1: + return "" + tail = text[open_idx + len("") :] + tail = tail.split("", 1)[0] + # 닫는/메타 태그가 스트림 청크 경계로 잘려 들어와도(예: "질문?" not in tail[lt:]: + tail = tail[:lt] + return tail.strip() + + +def parse_followup_result(text: str) -> FollowupResult: + """종료된 누적 텍스트를 FollowupResult 로. 태그 누락 시 전체를 질문으로 폴백.""" + intent_m = _INTENT_RE.search(text) + intent = intent_m.group(1).strip().upper() if intent_m else "NORMAL" + if intent not in _VALID_INTENTS: + intent = "NORMAL" + + question = extract_question_span(text) + if not question: + question = text.strip() + + evaluation = None + meta_m = _META_RE.search(text) + if meta_m: + try: + evaluation = AnswerEvaluation.model_validate_json(meta_m.group(1)) + except Exception: + evaluation = None + + return FollowupResult( + followup_question=question, + answer_evaluation=evaluation, + answer_intent=intent, + ) + + +class StreamingFollowupGenerator: + """단일 LLM 콜을 astream 으로 흘리며 토큰만 콜백으로 내보낸다. + + intent 가 DONT_KNOW 면 질문 델타를 보내지 않는다(Core 가 폐기하므로). + 종료 후 누적 텍스트를 parse_followup_result 로 검증해 반환. + """ + + def __init__(self, *, prompt: ChatPromptTemplate, llm: Any) -> None: + self._prompt = prompt + self._llm = llm + + async def stream( + self, + *, + on_question_token: Callable[[str], Awaitable[None] | None], + job_category: str, + mode: str, + previous_question: str, + answer_text: str, + context: str, + parent_category: str, + expected_signal: str, + history: str, + ) -> FollowupResult: + all_vars = { + "job_category": job_category, + "mode": mode, + "previous_question": previous_question, + "answer_text": answer_text, + "context": context, + "parent_category": parent_category, + "expected_signal": expected_signal, + "history": history, + } + # 프롬프트가 선언한 변수만 전달(템플릿마다 변수 집합이 다를 수 있음). + wanted = set(self._prompt.input_variables) + prompt_vars = {k: v for k, v in all_vars.items() if k in wanted} + messages = self._prompt.format_messages(**prompt_vars) + + acc = "" + emitted_q = "" + async for chunk in self._llm.astream(messages): + piece = getattr(chunk, "content", "") or "" + if not piece: + continue + acc += piece + intent_known = _INTENT_RE.search(acc) + if not intent_known: + continue + if intent_known.group(1).strip().upper() == "DONT_KNOW": + continue + q_now = extract_question_span(acc) + if len(q_now) > len(emitted_q): + delta = q_now[len(emitted_q) :] + emitted_q = q_now + res = on_question_token(delta) + if res is not None and hasattr(res, "__await__"): + await res + + return parse_followup_result(acc) + + class FollowupGenerator(Protocol): async def generate( self, @@ -103,3 +217,41 @@ def build_followup_generation_chain( callbacks=callbacks, ) return prompt | llm | parser + + +def build_streaming_followup_generator( + settings: Settings, core_client: CoreClient | None = None +) -> StreamingFollowupGenerator: + """스트리밍 꼬리질문 생성기 빌더. + + format_instructions 없는 프롬프트 + Flash LLM 을 조합해 + StreamingFollowupGenerator 를 반환한다. + """ + from langchain_openai import ChatOpenAI + + prompt = ChatPromptTemplate.from_messages( + [ + ("system", SYSTEM_PROMPT), + ("human", HUMAN_PROMPT), + ] + ) + + callbacks = [] + if core_client is not None: + callbacks.append( + CoreAiLogCallback( + core_client=core_client, + request_type="generate.followup.stream", + default_model=settings.llm_flash_model, + ) + ) + + llm = ChatOpenAI( + model=settings.llm_flash_model, + temperature=settings.llm_flash_temperature, + api_key=settings.llm_api_key or None, + base_url=settings.llm_base_url, + max_tokens=settings.llm_flash_max_tokens, + callbacks=callbacks, + ) + return StreamingFollowupGenerator(prompt=prompt, llm=llm) diff --git a/ai/src/ai_server/chain/prompts/followup_generation.py b/ai/src/ai_server/chain/prompts/followup_generation.py index aeefdaf6..70921d21 100644 --- a/ai/src/ai_server/chain/prompts/followup_generation.py +++ b/ai/src/ai_server/chain/prompts/followup_generation.py @@ -26,7 +26,13 @@ "이때 followup_question 에는 새 꼬리질문이 아니라 **직전 질문을 더 쉽고 구체적으로 다시 설명한 문장**을 담으세요.\n" " - NORMAL: 그 외 정상 답변. followup_question 은 평소처럼 가장 약한 축을 파는 꼬리질문.\n" " - DONT_KNOW/CLARIFICATION 이면 채점(specificity/logic/correctness)은 보수적으로(낮게/null) 둡니다.\n" - "- 응답은 반드시 지정된 JSON 스키마를 따릅니다." + "- 출력은 정확히 다음 3개 블록을 이 순서로만 작성합니다(설명·코드펜스 금지):\n" + " NORMAL 또는 DONT_KNOW 또는 CLARIFICATION\n" + " 지원자에게 보여줄 한국어 질문 1개(또는 CLARIFICATION 시 재설명 문장)\n" + " {{\"specificity\": <0~5>, \"logic\": <0~5>, " + "\"structure\": \"FULL_STAR|PARTIAL_STAR|NONE\", \"correctness\": <0~5 또는 null>}}\n" + "- 를 가장 먼저 확정해 출력합니다. DONT_KNOW 여도 블록은 비우지 말고 한 문장 넣되, " + "표시는 시스템이 결정합니다." ) HUMAN_PROMPT = ( @@ -41,6 +47,5 @@ "직전 질문:\n{previous_question}\n\n" "직전 질문의 기대 신호(평가 기준):\n{expected_signal}\n\n" "지원자 답변:\n{answer_text}\n\n" - "검색 문서 컨텍스트:\n---\n{context}\n---\n\n" - "{format_instructions}" + "검색 문서 컨텍스트:\n---\n{context}\n---\n" ) diff --git a/ai/src/ai_server/chain/sentence_split.py b/ai/src/ai_server/chain/sentence_split.py new file mode 100644 index 00000000..4e058dcb --- /dev/null +++ b/ai/src/ai_server/chain/sentence_split.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import re + +# 종결: 문장부호(. ! ? …) 뒤에 공백/끝, 또는 줄바꿈. +_SENTENCE_END = re.compile(r".*?(?:[.!?…](?=[ \t]|$)|\n)", re.DOTALL) + + +def next_sentences(text: str, consumed: int) -> tuple[list[str], int]: + """text[consumed:] 에서 '완성된' 문장들만 떼어 반환하고 새 consumed 를 돌려준다. + + 미완성(종결부호 없는) 꼬리는 남겨 다음 호출에서 이어 처리한다. + + 반환 정책: + - 완성 문장이 없으면 ([], consumed) 반환. + - 완성 문장이 1개이거나 마지막 완성 문장이 '\\n' 으로 끝나면 전부 반환. + - 그 외(마지막 완성 문장이 공백 앞 구두점으로만 끝나고 2개 이상 존재)에는 + 마지막 문장을 버퍼에 남기고 나머지만 반환한다. 이는 스트리밍 중 아직 + 다음 토큰이 올 수 있는 경계를 보수적으로 처리하기 위함이다. + """ + rest = text[consumed:] + + # 각 완성 문장의 (stripped 텍스트, 줄바꿈 종결 여부, 매치 끝 위치) 수집 + segments: list[tuple[str, bool, int]] = [] + pos = 0 + for m in _SENTENCE_END.finditer(rest): + chunk = rest[pos : m.end()].strip() + is_newline = m.group().endswith("\n") + if chunk: + segments.append((chunk, is_newline, m.end())) + pos = m.end() + + if not segments: + return [], consumed + + # 마지막 문장이 줄바꿈 종결이거나 완성 문장이 1개뿐이면 전부 방출 + last_is_newline = segments[-1][1] + if last_is_newline or len(segments) == 1: + out = [s for s, _, _ in segments] + new_consumed = consumed + segments[-1][2] + else: + # 마지막 space-terminated 문장은 버퍼로 남김 + to_emit = segments[:-1] + out = [s for s, _, _ in to_emit] + new_consumed = consumed + to_emit[-1][2] + + return out, new_consumed diff --git a/ai/src/ai_server/messaging/consumers/followup_consumer.py b/ai/src/ai_server/messaging/consumers/followup_consumer.py index 280980aa..4dc6606f 100644 --- a/ai/src/ai_server/messaging/consumers/followup_consumer.py +++ b/ai/src/ai_server/messaging/consumers/followup_consumer.py @@ -1,12 +1,19 @@ from __future__ import annotations +import asyncio + import structlog from aio_pika.abc import AbstractIncomingMessage -from ai_server.chain.followup_generation_chain import FollowupGenerator +from ai_server.chain.followup_generation_chain import ( + FollowupGenerator, + StreamingFollowupGenerator, +) +from ai_server.chain.sentence_split import next_sentences 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 SessionRealtimeNotifier from ai_server.model.envelope import Envelope from ai_server.model.messages.followup import ( FollowupCallbackPayload, @@ -17,6 +24,13 @@ log = structlog.get_logger(__name__) +_EXT_BY_CONTENT_TYPE = { + "audio/wav": "wav", + "audio/mpeg": "mp3", + "audio/ogg": "ogg", + "audio/mp4": "m4a", +} + class FollowupConsumer: def __init__( @@ -31,6 +45,11 @@ def __init__( rag_top_k: int = 5, reranker: Reranker | None = None, candidate_k: int = 20, + streaming_generator: StreamingFollowupGenerator | None = None, + session_notifier: SessionRealtimeNotifier | None = None, + tts=None, + storage=None, + tts_voice: str = "", ) -> None: self._generator = generator self._publisher = publisher @@ -41,6 +60,11 @@ def __init__( self._rag_top_k = rag_top_k self._reranker = reranker or NoopReranker() self._candidate_k = max(candidate_k, rag_top_k) + self._streaming = streaming_generator + self._notifier = session_notifier + self._tts = tts + self._storage = storage + self._tts_voice = tts_voice async def handle(self, message: AbstractIncomingMessage) -> None: async with message.process(requeue=False): @@ -73,16 +97,74 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) - result = await self._generator.generate( - job_category=req.job_category, - mode=req.mode, - previous_question=req.previous_question, - answer_text=req.answer_text, - context=await self._build_rag_context(req), - parent_category=req.parent_category or "UNKNOWN", - expected_signal=req.parent_expected_signal or "(none)", - history=_format_history(req.history), - ) + rag_context = await self._build_rag_context(req) + + if self._streaming is not None and self._notifier is not None: + seq = {"n": 0} + audio_seq = {"n": 0} + q_acc = {"text": "", "consumed": 0} + synth_tasks: list = [] + + def _spawn_segment(sentence: str) -> None: + if not (self._tts and self._storage): + return + idx = audio_seq["n"] + audio_seq["n"] += 1 + synth_tasks.append( + asyncio.create_task( + self._synth_segment( + session_id=req.session_id, + message_id=req.followup_message_id, + seq=idx, + sentence=sentence, + trace_id=envelope.trace_id, + ) + ) + ) + + async def _on_token(delta: str) -> None: + await self._notifier.emit_delta( + session_id=req.session_id, + message_id=req.followup_message_id, + seq=seq["n"], + text=delta, + trace_id=envelope.trace_id, + ) + seq["n"] += 1 + q_acc["text"] += delta + sents, q_acc["consumed"] = next_sentences( + q_acc["text"], q_acc["consumed"] + ) + for s in sents: + _spawn_segment(s) + + result = await self._streaming.stream( + on_question_token=_on_token, + job_category=req.job_category, + mode=req.mode, + previous_question=req.previous_question, + answer_text=req.answer_text, + context=rag_context, + parent_category=req.parent_category or "UNKNOWN", + expected_signal=req.parent_expected_signal or "(none)", + history=_format_history(req.history), + ) + tail = q_acc["text"][q_acc["consumed"] :].strip() + if tail: + _spawn_segment(tail) + if synth_tasks: + await asyncio.gather(*synth_tasks, return_exceptions=True) + else: + result = await self._generator.generate( + job_category=req.job_category, + mode=req.mode, + previous_question=req.previous_question, + answer_text=req.answer_text, + context=rag_context, + parent_category=req.parent_category or "UNKNOWN", + expected_signal=req.parent_expected_signal or "(none)", + history=_format_history(req.history), + ) payload = FollowupCallbackPayload( session_id=req.session_id, @@ -92,6 +174,7 @@ async def handle(self, message: AbstractIncomingMessage) -> None: followup_question=result.followup_question, answer_evaluation=result.answer_evaluation, answer_intent=result.answer_intent, + followup_message_id=req.followup_message_id, ) await self._publisher.publish( @@ -109,6 +192,44 @@ async def handle(self, message: AbstractIncomingMessage) -> None: trace_id=envelope.trace_id, ) + async def _synth_segment( + self, + *, + session_id: int, + message_id: int, + seq: int, + sentence: str, + trace_id: str, + ) -> None: + """문장 하나를 TTS 합성 → S3 세그먼트 PUT → SESSION_MESSAGE_AUDIO 발행. + + 라이브 세그먼트는 휘발성 — 실패해도 텍스트 스트림/콜백을 막지 않는다(로그만). + """ + try: + res = await self._tts.synthesize(sentence, voice=self._tts_voice) + ext = _EXT_BY_CONTENT_TYPE.get( + res.content_type.split(";")[0].strip(), "mp3" + ) + key = f"interview/tts/{session_id}/{message_id}/seg-{seq}.{ext}" + await self._storage.put_bytes( + key, res.audio_bytes, content_type=res.content_type + ) + await self._notifier.emit_audio( + session_id=session_id, + message_id=message_id, + seq=seq, + ext=ext, + duration_sec=res.duration_sec, + trace_id=trace_id, + ) + except Exception as exc: # noqa: BLE001 + log.warning( + "followup.segment.failed", + message_id=message_id, + seq=seq, + error=str(exc), + ) + async def _build_rag_context(self, req: GenerateFollowupRequest) -> str: if not self._core or not self._embedder or not req.context_document_ids: return "(none)" diff --git a/ai/src/ai_server/messaging/runner.py b/ai/src/ai_server/messaging/runner.py index 02448a61..5bbb368f 100644 --- a/ai/src/ai_server/messaging/runner.py +++ b/ai/src/ai_server/messaging/runner.py @@ -17,6 +17,7 @@ from ai_server.chain.followup_generation_chain import ( LlmFollowupGenerator, build_followup_generation_chain, + build_streaming_followup_generator, ) from ai_server.chain.feedback_generation_chain import ( LlmFeedbackGenerator, @@ -45,6 +46,7 @@ from ai_server.messaging.idempotency import LruIdempotencyStore from ai_server.messaging.progress import AnalysisProgressNotifier from ai_server.messaging.publisher import CallbackPublisher +from ai_server.messaging.session_notify import SessionRealtimeNotifier from ai_server.storage.factory import build_storage log = structlog.get_logger(__name__) @@ -188,6 +190,15 @@ def __init__(self, settings: Settings) -> None: followup_generator = LlmFollowupGenerator( build_followup_generation_chain(settings, core_client=core_client) ) + 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( generator=followup_generator, publisher=self._publisher, @@ -197,6 +208,11 @@ def __init__(self, settings: Settings) -> None: embedder=embedder, reranker=reranker, candidate_k=settings.rerank_candidate_k, + streaming_generator=streaming_followup_generator, + session_notifier=session_notifier, + tts=tts, + storage=storage, + tts_voice=settings.openai_tts_voice, ) # 종합 피드백 생성 (US-24) @@ -227,8 +243,7 @@ def __init__(self, settings: Settings) -> None: core_client=core_client, ) - # 질문 TTS (Part A) - tts = build_tts_provider(settings) + # 질문 TTS (Part A) — tts 인스턴스는 위에서 이미 생성됨 self._tts_consumer = TtsConsumer( tts=tts, storage=storage, diff --git a/ai/src/ai_server/messaging/session_notify.py b/ai/src/ai_server/messaging/session_notify.py new file mode 100644 index 00000000..cd82cb57 --- /dev/null +++ b/ai/src/ai_server/messaging/session_notify.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import structlog + +from ai_server.messaging.publisher import CallbackPublisher +from ai_server.model.envelope import MessageContext +from ai_server.model.messages.realtime import ( + SessionAudioNotifyPayload, + SessionMessageAudioData, + SessionMessageDeltaData, + SessionNotifyPayload, +) + +log = structlog.get_logger(__name__) + +SESSION_MESSAGE_DELTA_EVENT = "SESSION_MESSAGE_DELTA" +SESSION_MESSAGE_AUDIO_EVENT = "SESSION_MESSAGE_AUDIO" + + +class SessionRealtimeNotifier: + """AI -> RealTime 세션 채널 직접 발행. 꼬리질문 토큰 델타(휘발성). + + 발행 실패가 생성 파이프라인을 막아선 안 된다 — 예외를 삼키고 경고만. + """ + + def __init__(self, *, publisher: CallbackPublisher, routing_key: str) -> None: + self._publisher = publisher + self._routing_key = routing_key + + async def emit_delta( + self, + *, + session_id: int, + message_id: int, + seq: int, + text: str, + trace_id: str, + ) -> None: + payload = SessionNotifyPayload( + event_type=SESSION_MESSAGE_DELTA_EVENT, + data=SessionMessageDeltaData(message_id=message_id, seq=seq, text=text), + ) + try: + await self._publisher.publish( + routing_key=self._routing_key, + message_type=self._routing_key, + payload=payload, + trace_id=trace_id, + correlation_id=f"delta-{message_id}-{seq}", + context=MessageContext(session_id=session_id), + ) + except Exception: + log.warning( + "session.delta.publish_failed", + session_id=session_id, + message_id=message_id, + seq=seq, + trace_id=trace_id, + ) + + async def emit_audio( + self, + *, + session_id: int, + message_id: int, + seq: int, + ext: str, + duration_sec: float | None, + trace_id: str, + ) -> None: + payload = SessionAudioNotifyPayload( + event_type=SESSION_MESSAGE_AUDIO_EVENT, + data=SessionMessageAudioData( + message_id=message_id, seq=seq, ext=ext, duration_sec=duration_sec + ), + ) + try: + await self._publisher.publish( + routing_key=self._routing_key, + message_type=self._routing_key, + payload=payload, + trace_id=trace_id, + correlation_id=f"audio-{message_id}-{seq}", + context=MessageContext(session_id=session_id), + ) + except Exception: + log.warning( + "session.audio.publish_failed", + session_id=session_id, + message_id=message_id, + seq=seq, + trace_id=trace_id, + ) diff --git a/ai/src/ai_server/model/messages/followup.py b/ai/src/ai_server/model/messages/followup.py index 4ef93fdb..1463c890 100644 --- a/ai/src/ai_server/model/messages/followup.py +++ b/ai/src/ai_server/model/messages/followup.py @@ -24,6 +24,7 @@ class GenerateFollowupRequest(BaseModel): session_id: int parent_message_id: int # 직전 질문 메시지 ID answer_message_id: int # 답변 메시지 ID + followup_message_id: int # Core 가 선INSERT 한 placeholder 질문 메시지 id previous_question: str answer_text: str mode: InterviewMode @@ -58,3 +59,4 @@ class FollowupCallbackPayload(BaseModel): answer_evaluation: AnswerEvaluation | None = None # 답변 의도: NORMAL | DONT_KNOW | CLARIFICATION. Core 가 흐름 분기에 사용. answer_intent: str = "NORMAL" + followup_message_id: int # placeholder UPDATE 대상 diff --git a/ai/src/ai_server/model/messages/realtime.py b/ai/src/ai_server/model/messages/realtime.py index 2abf28ae..61e64ddb 100644 --- a/ai/src/ai_server/model/messages/realtime.py +++ b/ai/src/ai_server/model/messages/realtime.py @@ -21,3 +21,36 @@ class RealtimeNotifyPayload(BaseModel): event_type: str data: AnalysisProgressData + + +# AI -> RealTime 세션 채널 직접 발행. 꼬리질문 토큰 델타(휘발성). Core 우회. +class SessionMessageDeltaData(BaseModel): + model_config = camel_config() + + message_id: int + seq: int + text: str # 이번 델타에서 추가된 조각(누적 아님) + + +class SessionNotifyPayload(BaseModel): + model_config = camel_config() + + event_type: str + data: SessionMessageDeltaData + + +# AI -> RealTime 세션 채널 직접 발행. 꼬리질문 문장 단위 TTS 세그먼트(휘발성). +class SessionMessageAudioData(BaseModel): + model_config = camel_config() + + message_id: int + seq: int + ext: str # wav | mp3 | ogg | m4a + duration_sec: float | None = None + + +class SessionAudioNotifyPayload(BaseModel): + model_config = camel_config() + + event_type: str + data: SessionMessageAudioData diff --git a/ai/tests/test_followup_consumer.py b/ai/tests/test_followup_consumer.py index a4f232db..2abebacd 100644 --- a/ai/tests/test_followup_consumer.py +++ b/ai/tests/test_followup_consumer.py @@ -44,6 +44,7 @@ def _envelope() -> bytes: "sessionId": 99, "parentMessageId": 501, "answerMessageId": 502, + "followupMessageId": 503, "previousQuestion": "결제 outbox 어떻게 구현?", "answerText": "RabbitMQ로 보냈습니다.", "mode": "TECHNICAL", @@ -55,17 +56,36 @@ def _envelope() -> bytes: return json.dumps(env).encode() +def _make_streaming_generator(result: FollowupResult): + """streaming_generator 페이크: stream() 호출 시 on_question_token 한 번 호출 후 result 반환.""" + streaming = MagicMock() + + async def _stream(*, on_question_token, **kwargs): + await on_question_token("토큰") + return result + + streaming.stream = _stream + return streaming + + +def _make_session_notifier(): + notifier = MagicMock() + notifier.emit_delta = AsyncMock() + return notifier + + @pytest.mark.asyncio async def test_consumer_generates_followup_and_publishes_callback(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="구체적으로 outbox 테이블 스키마와 polling 주기는?", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=3.0, structure="PARTIAL_STAR" - ), - ) + followup_result = FollowupResult( + followup_question="구체적으로 outbox 테이블 스키마와 polling 주기는?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) + streaming_generator = _make_streaming_generator(followup_result) + session_notifier = _make_session_notifier() publisher = MagicMock() publisher.publish = AsyncMock() @@ -74,10 +94,14 @@ async def test_consumer_generates_followup_and_publishes_callback(): publisher=publisher, idempotency=LruIdempotencyStore(max_size=10), callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) - generator.generate.assert_awaited_once() + # streaming path이 사용되므로 generator.generate 는 호출되지 않음 + generator.generate.assert_not_awaited() + session_notifier.emit_delta.assert_awaited_once() publisher.publish.assert_awaited_once() payload: FollowupCallbackPayload = publisher.publish.await_args.kwargs["payload"] assert payload.session_id == 99 @@ -85,20 +109,20 @@ async def test_consumer_generates_followup_and_publishes_callback(): assert payload.parent_message_id == 501 assert payload.followup_question.startswith("구체적으로") assert payload.answer_evaluation.structure == "PARTIAL_STAR" + assert payload.followup_message_id == 503 assert publisher.publish.await_args.kwargs["message_type"] == "callback.questions" @pytest.mark.asyncio async def test_consumer_injects_followup_rag_context_when_available(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="outbox 저장과 발행의 원자성은 어떻게 보장했나요?", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=3.0, structure="PARTIAL_STAR" - ), - ) + followup_result = FollowupResult( + followup_question="outbox 저장과 발행의 원자성은 어떻게 보장했나요?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() core = MagicMock() @@ -115,6 +139,18 @@ async def test_consumer_injects_followup_rag_context_when_available(): embedder = MagicMock() embedder.embed = AsyncMock(return_value=[[0.1, 0.2, 0.3]]) + # Track kwargs passed to stream + received_kwargs: dict = {} + + async def _stream(*, on_question_token, **kwargs): + received_kwargs.update(kwargs) + await on_question_token("토큰") + return followup_result + + streaming_generator = MagicMock() + streaming_generator.stream = _stream + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, @@ -123,6 +159,8 @@ async def test_consumer_injects_followup_rag_context_when_available(): core_client=core, embedder=embedder, rag_top_k=3, + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) @@ -132,21 +170,21 @@ async def test_consumer_injects_followup_rag_context_when_available(): assert call.kwargs["document_ids"] == [7] assert call.kwargs["top_k"] == 20 # candidate_k (리랭크 후보 수) assert call.kwargs["query_text"] # 하이브리드 검색: 쿼리 텍스트 동봉 - context = generator.generate.await_args.kwargs["context"] - assert "Outbox rows are inserted in the same transaction" in context + assert ( + "Outbox rows are inserted in the same transaction" in received_kwargs["context"] + ) @pytest.mark.asyncio async def test_consumer_falls_back_when_followup_rag_fails(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="실패 재처리는 어떻게 했나요?", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=3.0, structure="PARTIAL_STAR" - ), - ) + followup_result = FollowupResult( + followup_question="실패 재처리는 어떻게 했나요?", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() core = MagicMock() @@ -154,6 +192,17 @@ async def test_consumer_falls_back_when_followup_rag_fails(): embedder = MagicMock() embedder.embed = AsyncMock(return_value=[[0.1, 0.2, 0.3]]) + received_kwargs: dict = {} + + async def _stream(*, on_question_token, **kwargs): + received_kwargs.update(kwargs) + await on_question_token("토큰") + return followup_result + + streaming_generator = MagicMock() + streaming_generator.stream = _stream + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, @@ -161,10 +210,12 @@ async def test_consumer_falls_back_when_followup_rag_fails(): callback_routing_key="callback.questions", core_client=core, embedder=embedder, + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) - assert generator.generate.await_args.kwargs["context"] == "(none)" + assert received_kwargs["context"] == "(none)" @pytest.mark.asyncio @@ -175,12 +226,18 @@ async def test_consumer_idempotent_skip(): publisher.publish = AsyncMock() idempotency = LruIdempotencyStore(max_size=10) idempotency.is_seen_then_mark("m-1") + streaming_generator = _make_streaming_generator( + FollowupResult(followup_question="Q") + ) + session_notifier = _make_session_notifier() consumer = FollowupConsumer( generator=generator, publisher=publisher, idempotency=idempotency, callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) generator.generate.assert_not_awaited() @@ -210,24 +267,158 @@ def test_answer_evaluation_correctness_defaults_none_and_parses(): assert e2.correctness == 3.5 +# --------------------------------------------------------------------------- +# TTS segment synthesis tests (Part B Task 4) +# --------------------------------------------------------------------------- + + +def _make_tts_fake(): + """audio/mpeg を返す偽 TtsProvider.""" + from ai_server.voice.tts.base import TtsResult + + tts = MagicMock() + tts.synthesize = AsyncMock( + return_value=TtsResult( + audio_bytes=b"x", duration_sec=0.5, content_type="audio/mpeg" + ) + ) + return tts + + +def _make_storage_fake(): + storage = MagicMock() + storage.put_bytes = AsyncMock() + return storage + + +def _make_full_session_notifier(): + notifier = MagicMock() + notifier.emit_delta = AsyncMock() + notifier.emit_audio = AsyncMock() + return notifier + + +def _make_streaming_generator_two_tokens(result): + """stream() 이 on_question_token 을 두 번 호출 ('첫 문장이다. ', '둘째다.').""" + streaming = MagicMock() + + async def _stream(*, on_question_token, **kwargs): + await on_question_token("첫 문장이다. ") + await on_question_token("둘째다.") + return result + + streaming.stream = _stream + return streaming + + @pytest.mark.asyncio -async def test_consumer_passes_parent_category_and_history_to_generator(): +async def test_segment_synthesis_emits_audio_event(): + """스트리밍 경로에서 TTS + storage 가 주입되면 segment 합성 후 emit_audio 를 발행한다.""" + followup_result = FollowupResult( + followup_question="구체적인 질문", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=3.0, structure="PARTIAL_STAR" + ), + ) generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="새 각도 질문", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=2.0, structure="NONE" - ), - ) + generator.generate = AsyncMock(return_value=followup_result) + publisher = MagicMock() + publisher.publish = AsyncMock() + + tts = _make_tts_fake() + storage = _make_storage_fake() + session_notifier = _make_full_session_notifier() + streaming_generator = _make_streaming_generator_two_tokens(followup_result) + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, + tts=tts, + storage=storage, + tts_voice="alloy", + ) + await consumer.handle(_StubMessage(_envelope())) + + # emit_audio 는 최소 1회 이상 호출돼야 한다 + assert session_notifier.emit_audio.await_count >= 1 + + # storage.put_bytes 첫 번째 호출 키가 seg-0.mp3 패턴 + assert storage.put_bytes.await_count >= 1 + first_key = storage.put_bytes.await_args_list[0].args[0] + assert first_key == "interview/tts/99/503/seg-0.mp3" + + # emit_audio 첫 번째 호출 kwargs + first_audio_call = session_notifier.emit_audio.await_args_list[0].kwargs + assert first_audio_call["ext"] == "mp3" + assert first_audio_call["seq"] == 0 + + +@pytest.mark.asyncio +async def test_no_tts_injected_emits_no_audio(): + """tts/storage 를 주입하지 않으면 emit_audio 는 호출되지 않는다.""" + followup_result = FollowupResult( + followup_question="Q", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=2.0, structure="NONE" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() + + session_notifier = _make_full_session_notifier() + streaming_generator = _make_streaming_generator_two_tokens(followup_result) + consumer = FollowupConsumer( generator=generator, publisher=publisher, idempotency=LruIdempotencyStore(max_size=10), callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, + # tts / storage 미주입 + ) + await consumer.handle(_StubMessage(_envelope())) + + session_notifier.emit_audio.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_consumer_passes_parent_category_and_history_to_generator(): + followup_result = FollowupResult( + followup_question="새 각도 질문", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=2.0, structure="NONE" + ), + ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) + publisher = MagicMock() + publisher.publish = AsyncMock() + + received_kwargs: dict = {} + + async def _stream(*, on_question_token, **kwargs): + received_kwargs.update(kwargs) + await on_question_token("토큰") + return followup_result + + streaming_generator = MagicMock() + streaming_generator.stream = _stream + session_notifier = _make_session_notifier() + + consumer = FollowupConsumer( + generator=generator, + publisher=publisher, + idempotency=LruIdempotencyStore(max_size=10), + callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) env = { "messageId": "m-9", @@ -240,6 +431,7 @@ async def test_consumer_passes_parent_category_and_history_to_generator(): "sessionId": 99, "parentMessageId": 501, "answerMessageId": 502, + "followupMessageId": 503, "previousQuestion": "Q?", "answerText": "A.", "mode": "TECHNICAL", @@ -255,32 +447,36 @@ async def test_consumer_passes_parent_category_and_history_to_generator(): } await consumer.handle(_StubMessage(json.dumps(env).encode())) - kwargs = generator.generate.await_args.kwargs - assert kwargs["parent_category"] == "PROJECT_DEEP_DIVE" - assert kwargs["expected_signal"] == "동시성 제어를 DB 레벨까지 설명하는지" - assert "이전 질문" in kwargs["history"] - assert "면접관:" in kwargs["history"] + assert received_kwargs["parent_category"] == "PROJECT_DEEP_DIVE" + assert received_kwargs["expected_signal"] == "동시성 제어를 DB 레벨까지 설명하는지" + assert "이전 질문" in received_kwargs["history"] + assert "면접관:" in received_kwargs["history"] @pytest.mark.asyncio async def test_callback_includes_answer_message_id(): - generator = MagicMock() - generator.generate = AsyncMock( - return_value=FollowupResult( - followup_question="Q", - answer_evaluation=AnswerEvaluation( - specificity=2.0, logic=2.0, structure="NONE" - ), - ) + followup_result = FollowupResult( + followup_question="Q", + answer_evaluation=AnswerEvaluation( + specificity=2.0, logic=2.0, structure="NONE" + ), ) + generator = MagicMock() + generator.generate = AsyncMock(return_value=followup_result) publisher = MagicMock() publisher.publish = AsyncMock() + streaming_generator = _make_streaming_generator(followup_result) + session_notifier = _make_session_notifier() + consumer = FollowupConsumer( generator=generator, publisher=publisher, idempotency=LruIdempotencyStore(max_size=10), callback_routing_key="callback.questions", + streaming_generator=streaming_generator, + session_notifier=session_notifier, ) await consumer.handle(_StubMessage(_envelope())) payload = publisher.publish.await_args.kwargs["payload"] assert payload.answer_message_id == 502 # _envelope 의 answerMessageId + assert payload.followup_message_id == 503 # _envelope 의 followupMessageId diff --git a/ai/tests/test_followup_parse.py b/ai/tests/test_followup_parse.py new file mode 100644 index 00000000..e67b82e2 --- /dev/null +++ b/ai/tests/test_followup_parse.py @@ -0,0 +1,31 @@ +from ai_server.chain.followup_generation_chain import parse_followup_result + + +def test_parses_intent_question_meta(): + text = ( + "NORMAL" + "그 설계에서 트랜잭션 경계를 어디에 뒀나요?" + '{"specificity": 2, "logic": 3, "structure": "PARTIAL_STAR", "correctness": null}' + ) + r = parse_followup_result(text) + assert r.answer_intent == "NORMAL" + assert r.followup_question == "그 설계에서 트랜잭션 경계를 어디에 뒀나요?" + assert r.answer_evaluation.specificity == 2 + assert r.answer_evaluation.correctness is None + + +def test_fallback_when_no_tags(): + r = parse_followup_result("이건 그냥 질문 한 줄") + assert r.followup_question == "이건 그냥 질문 한 줄" + assert r.answer_intent == "NORMAL" + assert r.answer_evaluation is None + + +def test_extract_question_only_helper(): + from ai_server.chain.followup_generation_chain import extract_question_span + + assert extract_question_span("NORMAL안녕") == "안녕" + assert ( + extract_question_span("NORMAL안녕{}") + == "안녕" + ) diff --git a/ai/tests/test_followup_streaming.py b/ai/tests/test_followup_streaming.py new file mode 100644 index 00000000..ee5b977e --- /dev/null +++ b/ai/tests/test_followup_streaming.py @@ -0,0 +1,69 @@ +import pytest +from langchain_core.prompts import ChatPromptTemplate +from ai_server.chain.followup_generation_chain import StreamingFollowupGenerator + + +class _FakeStreamLLM: + """astream 시 토큰 조각을 순차로 내는 가짜 모델.""" + + def __init__(self, pieces): + self._pieces = pieces + + async def astream(self, _input, **_kw): + from langchain_core.messages import AIMessageChunk + + for p in self._pieces: + yield AIMessageChunk(content=p) + + +@pytest.mark.asyncio +async def test_stream_emits_question_tokens_and_final_result(): + pieces = [ + "NORMAL그 ", + "설계에서 ", + "트레이드오프는?", + "", + '{"specificity":2,"logic":3,"structure":"NONE","correctness":null}', + ] + prompt = ChatPromptTemplate.from_messages([("human", "{answer_text}")]) + gen = StreamingFollowupGenerator(prompt=prompt, llm=_FakeStreamLLM(pieces)) + + seen = [] + result = await gen.stream( + on_question_token=lambda piece: seen.append(piece), + job_category="BACKEND", + mode="TECHNICAL", + previous_question="q", + answer_text="a", + context="(none)", + parent_category="X", + expected_signal="(none)", + history="(none)", + ) + joined = "".join(seen) + assert "그 설계에서 트레이드오프는?" in joined + assert "" not in joined and "" not in joined # 질문만 흘렀다 + assert result.answer_intent == "NORMAL" + assert result.followup_question == "그 설계에서 트레이드오프는?" + + +@pytest.mark.asyncio +async def test_stream_dont_know_emits_no_question_tokens(): + pieces = ["DONT_KNOW모르면 이렇게{}"] + prompt = ChatPromptTemplate.from_messages([("human", "{answer_text}")]) + gen = StreamingFollowupGenerator(prompt=prompt, llm=_FakeStreamLLM(pieces)) + seen = [] + result = await gen.stream( + on_question_token=lambda piece: seen.append(piece), + job_category="BACKEND", + mode="TECHNICAL", + previous_question="q", + answer_text="a", + context="(none)", + parent_category="X", + expected_signal="(none)", + history="(none)", + ) + assert seen == [] # DONT_KNOW 는 질문 델타 0개 + assert result.answer_intent == "DONT_KNOW" diff --git a/ai/tests/test_sentence_split.py b/ai/tests/test_sentence_split.py new file mode 100644 index 00000000..011eaba1 --- /dev/null +++ b/ai/tests/test_sentence_split.py @@ -0,0 +1,28 @@ +from ai_server.chain.sentence_split import next_sentences + + +def test_extracts_completed_sentences_only(): + sents, consumed = next_sentences("첫 문장이다. 둘째 문장", 0) + assert sents == ["첫 문장이다."] + assert consumed == len("첫 문장이다.") + + +def test_no_complete_sentence_returns_empty(): + sents, consumed = next_sentences("아직 끝나지 않은", 0) + assert sents == [] + assert consumed == 0 + + +def test_multiple_enders_and_newline(): + text = "하나? 둘! 셋\n넷" + sents, consumed = next_sentences(text, 0) + assert sents == ["하나?", "둘!", "셋"] + assert text[consumed:] == "넷" + + +def test_resume_from_consumed(): + text = "가. 나. 다" + s1, c1 = next_sentences(text, 0) + assert s1 == ["가."] + s2, c2 = next_sentences(text, c1) + assert s2 == ["나."] diff --git a/ai/tests/test_session_notify.py b/ai/tests/test_session_notify.py new file mode 100644 index 00000000..02af3932 --- /dev/null +++ b/ai/tests/test_session_notify.py @@ -0,0 +1,61 @@ +import pytest +from ai_server.messaging.session_notify import SessionRealtimeNotifier + + +class _FakePublisher: + def __init__(self): + self.calls = [] + + async def publish(self, **kwargs): + self.calls.append(kwargs) + + +@pytest.mark.asyncio +async def test_emit_delta_publishes_session_event(): + pub = _FakePublisher() + n = SessionRealtimeNotifier(publisher=pub, routing_key="realtime.session.notify") + await n.emit_delta(session_id=7, message_id=503, seq=0, text="안녕", trace_id="t1") + assert len(pub.calls) == 1 + call = pub.calls[0] + assert call["routing_key"] == "realtime.session.notify" + assert call["message_type"] == "realtime.session.notify" + assert call["context"].session_id == 7 + assert call["payload"].event_type == "SESSION_MESSAGE_DELTA" + assert call["payload"].data.text == "안녕" + + +@pytest.mark.asyncio +async def test_emit_delta_swallows_publish_errors(): + class Boom: + async def publish(self, **kwargs): + raise RuntimeError("down") + + n = SessionRealtimeNotifier(publisher=Boom(), routing_key="realtime.session.notify") + await n.emit_delta(session_id=1, message_id=2, seq=0, text="x", trace_id="t") + + +@pytest.mark.asyncio +async def test_emit_audio_publishes_audio_event(): + pub = _FakePublisher() + n = SessionRealtimeNotifier(publisher=pub, routing_key="realtime.session.notify") + await n.emit_audio( + session_id=7, message_id=503, seq=0, ext="mp3", duration_sec=1.2, trace_id="t1" + ) + call = pub.calls[0] + assert call["routing_key"] == "realtime.session.notify" + assert call["payload"].event_type == "SESSION_MESSAGE_AUDIO" + assert call["payload"].data.ext == "mp3" + assert call["payload"].data.seq == 0 + assert call["context"].session_id == 7 + + +@pytest.mark.asyncio +async def test_emit_audio_swallows_publish_errors(): + class Boom: + async def publish(self, **kwargs): + raise RuntimeError("down") + + n = SessionRealtimeNotifier(publisher=Boom(), routing_key="realtime.session.notify") + await n.emit_audio( + session_id=1, message_id=2, seq=0, ext="mp3", duration_sec=None, trace_id="t" + ) diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md index 433dd373..be5ba071 100644 --- a/backend/CLAUDE.md +++ b/backend/CLAUDE.md @@ -361,6 +361,8 @@ docker compose up -d callback.questions(POOL/FOLLOWUP) 수신, 자동 종료 - 질문 TTS 발행 본 구현: 질문 영속 후 `QuestionPersistedEvent`(AFTER_COMMIT) → `SessionTtsRequester` 가 `generate.tts` 발행, `callback.tts` 수신해 메시지에 오디오 경로 반영(`TtsCallbackService`) +- **꼬리질문 토큰 스트리밍 placeholder 본 구현**: `SessionFollowupRequester` 가 AI 호출 분기에서 INTERVIEWER placeholder(`InterviewMessage.followupPlaceholder`, content=`"(생성 중)"`, status=`CREATED`)를 선INSERT + `SESSION_MESSAGE` 발행 + `generate.followup.followupMessageId` 동봉. AI 가 토큰을 `SESSION_MESSAGE_DELTA` 로 흘린 뒤 `callback.questions(FOLLOWUP)` 도착 시 `QuestionsCallbackService` 가 분기: NORMAL→`completeFollowup` UPDATE+카운트, CLARIFICATION→UPDATE(카운트 X), DONT_KNOW→placeholder DELETE 후 `advanceToNextGeneral`. placeholder 없는 레거시 콜백은 기존 INSERT 폴백. +- **문장 단위 TTS 세그먼트 프록시 본 구현 (Part B)**: `InterviewMessageService.streamAudioSegment` + `GET /api/sessions/{sid}/messages/{mid}/audio/segments/{seq}?ext=`. AI 가 휘발성으로 쓴 라이브 세그먼트를 규칙(`interview/tts/{sid}/{mid}/seg-{seq}.{ext}`)으로 재구성해 프록시(DB 미기록). 소유권+ext 화이트리스트+seq>=0 검증으로 임의 키 노출 차단. - AI 호출 로깅 (US-30) 본 구현: `/api/internal/ai-logs` + `ai_request_logs` INSERT - **Spring AI 미사용** — LLM·임베딩 호출은 모두 AI 서버 위임. Core는 RabbitMQ 발행만 담당. - **Redis 미사용** — 휘발성 데이터는 DB short-lived 레코드 또는 인메모리로. diff --git a/backend/openapi.json b/backend/openapi.json index 81330e8d..42a7f194 100644 --- a/backend/openapi.json +++ b/backend/openapi.json @@ -1861,6 +1861,81 @@ } } }, + "/api/sessions/{sessionId}/messages/{messageId}/audio/segments/{seq}" : { + "get" : { + "tags" : [ "Session Messages" ], + "summary" : "라이브 TTS 문장 세그먼트 스트리밍", + "description" : "AI 서버가 S3에 쓴 per-sentence TTS 세그먼트를 규칙 기반 키로 프록시한다. 세그먼트는 DB에 미저장(휘발성 라이브)이므로 Core 가 키를 재구성해 스트리밍.", + "operationId" : "streamMessageAudioSegment", + "parameters" : [ { + "name" : "sessionId", + "in" : "path", + "required" : true, + "schema" : { + "type" : "integer", + "format" : "int64" + } + }, { + "name" : "messageId", + "in" : "path", + "required" : true, + "schema" : { + "type" : "integer", + "format" : "int64" + } + }, { + "name" : "seq", + "in" : "path", + "required" : true, + "schema" : { + "type" : "integer", + "format" : "int32" + } + }, { + "name" : "ext", + "in" : "query", + "required" : true, + "schema" : { + "type" : "string" + } + } ], + "responses" : { + "200" : { + "description" : "세그먼트 오디오 바이트", + "content" : { + "*/*" : { + "schema" : { + "type" : "string", + "format" : "binary" + } + } + } + }, + "401" : { + "description" : "인증 실패", + "content" : { + "*/*" : { + "schema" : { + "type" : "string", + "format" : "binary" + } + } + } + }, + "404" : { + "description" : "세션/메시지/세그먼트 없음", + "content" : { + "*/*" : { + "schema" : { + "type" : "string", + "format" : "binary" + } + } + } + } + } + } + }, "/api/sessions/{sessionId}/feedback" : { "get" : { "tags" : [ "Session Feedback" ], diff --git a/backend/src/main/java/com/stackup/stackup/session/application/InterviewMessageService.java b/backend/src/main/java/com/stackup/stackup/session/application/InterviewMessageService.java index 5405421b..c65ff62b 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/InterviewMessageService.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/InterviewMessageService.java @@ -88,6 +88,23 @@ public AudioStream streamAudio(Long userId, Long sessionId, Long messageId) { public record AudioStream(java.io.InputStream stream, String contentType) {} + private static final java.util.Set SEGMENT_EXTS = + java.util.Set.of("wav", "mp3", "ogg", "m4a"); + + // 라이브 문장 TTS 세그먼트 프록시. 키는 AI 와 동일 규칙으로 재구성(DB 미저장). + // 소유권(ownedSession) + 메시지-세션 일치 + ext 화이트리스트 + seq>=0 으로 임의 키 노출 차단. + public AudioStream streamAudioSegment(Long userId, Long sessionId, Long messageId, int seq, String ext) { + ownedSession(userId, sessionId); + if (ext == null || !SEGMENT_EXTS.contains(ext) || seq < 0) { + throw new DomainException(ApiErrorCode.VOICE_MESSAGE_NOT_FOUND); + } + InterviewMessage m = messageRepository.findById(messageId) + .filter(x -> x.getSession().getId().equals(sessionId)) + .orElseThrow(() -> new DomainException(ApiErrorCode.VOICE_MESSAGE_NOT_FOUND)); + String key = "interview/tts/%d/%d/seg-%d.%s".formatted(sessionId, m.getId(), seq, ext); + return new AudioStream(storage.get(key), contentTypeForKey(key)); + } + private static String contentTypeForKey(String key) { String k = key.toLowerCase(); if (k.endsWith(".mp3")) return "audio/mpeg"; diff --git a/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java b/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java index f0885a58..7525344c 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java @@ -163,9 +163,17 @@ private void applyFollowup(InterviewSession session, QuestionsCallbackPayload pa : messageRepository.findById(payload.parentMessageId()).orElse(null); String intent = payload.answerIntent() == null ? "NORMAL" : payload.answerIntent(); - // 모르겠음 → 이 주제 더 안 파고 다음 일반질문으로(없으면 종료). 답변평가는 기록(낮은 점수 의미 있음). + InterviewMessage placeholder = payload.followupMessageId() == null + ? null + : messageRepository.findById(payload.followupMessageId()).orElse(null); + + // 모르겠음 → 이 주제 그만, 다음 일반질문. placeholder 는 삭제(seq 연속성 유지). if ("DONT_KNOW".equalsIgnoreCase(intent)) { recordAnswerEvaluation(payload); + if (placeholder != null) { + messageRepository.delete(placeholder); + messageRepository.flush(); + } log.info("callback.questions DONT_KNOW — advancing to next general. sessionId={}", session.getId()); advanceToNextGeneral(session.getId()); return; @@ -176,31 +184,32 @@ private void applyFollowup(InterviewSession session, QuestionsCallbackPayload pa return; } - // 부연 요청 → 같은 질문 재설명. 질문 수(k)·깊이(m) 미반영, 답변평가 기록 안 함. - if ("CLARIFICATION".equalsIgnoreCase(intent)) { - int seq = (int) messageRepository.countBySession_Id(session.getId()) + 1; - InterviewMessage clar = messageRepository.save( - InterviewMessage.clarification(session, seq, payload.followupQuestion(), parent)); - publishQuestionEvents(session, clar, "CLARIFICATION"); - log.info("callback.questions CLARIFICATION — re-explained question. sessionId={}, msg={}", - session.getId(), clar.getId()); - return; + boolean clarification = "CLARIFICATION".equalsIgnoreCase(intent); + if (!clarification) { + recordAnswerEvaluation(payload); } - // 정상 답변 → 꼬리질문 삽입. - recordAnswerEvaluation(payload); - int nextSeq = (int) messageRepository.countBySession_Id(session.getId()) + 1; - InterviewMessage message = messageRepository.save( - InterviewMessage.followup(session, nextSeq, payload.followupQuestion(), parent) - ); - session.incrementQuestionCount(); - publishQuestionEvents(session, message, "FOLLOWUP_READY"); - - // maxQuestions(k) 도달 시 자동 종료. - maybeAutoEnd(session); + InterviewMessage message; + if (placeholder != null) { + placeholder.completeFollowup(payload.followupQuestion(), clarification); + message = messageRepository.save(placeholder); + } else { + // 레거시 폴백: placeholder 없이 도착한 콜백(롤아웃 호환). + int nextSeq = (int) messageRepository.countBySession_Id(session.getId()) + 1; + message = messageRepository.save(clarification + ? InterviewMessage.clarification(session, nextSeq, payload.followupQuestion(), parent) + : InterviewMessage.followup(session, nextSeq, payload.followupQuestion(), parent)); + } - log.info("callback.questions FOLLOWUP processed. sessionId={}, msg={}, totalQ={}", - session.getId(), message.getId(), session.getTotalQuestionCount()); + if (!clarification) { + session.incrementQuestionCount(); + } + publishQuestionEvents(session, message, clarification ? "CLARIFICATION" : "FOLLOWUP_READY"); + if (!clarification) { + maybeAutoEnd(session); + } + log.info("callback.questions FOLLOWUP processed. sessionId={}, msg={}, clarification={}", + session.getId(), message.getId(), clarification); } public record SessionStateNotice(Long sessionId, String status, String reason) { diff --git a/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java b/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java index cc567232..577beb43 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java @@ -3,6 +3,8 @@ import com.stackup.stackup.common.config.properties.RabbitMqProperties; import com.stackup.stackup.common.messaging.MessageContext; import com.stackup.stackup.common.messaging.RabbitMessagePublisher; +import com.stackup.stackup.common.messaging.RealtimeNotifyEvent; +import com.stackup.stackup.common.sse.SseEventType; import com.stackup.stackup.session.application.dto.GenerateFollowupPayload; import com.stackup.stackup.session.application.dto.GenerateFollowupPayload.HistoryItem; import com.stackup.stackup.session.application.event.AnswerSubmittedEvent; @@ -35,6 +37,7 @@ public class SessionFollowupRequester { private final InterviewMessageRepository messageRepository; private final SessionContextRepository contextRepository; private final QuestionsCallbackService questionsCallbackService; + private final org.springframework.context.ApplicationEventPublisher events; @Transactional(propagation = Propagation.REQUIRES_NEW) @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) @@ -74,6 +77,15 @@ public void onAnswerSubmitted(AnswerSubmittedEvent event) { .map(m -> new HistoryItem(m.getRole().name(), m.getContent())) .toList(); + // 스트리밍 표시용 placeholder 선INSERT (seq 예약). 콜백에서 content 채움. + int placeholderSeq = (int) messageRepository.countBySession_Id(session.getId()) + 1; + InterviewMessage placeholder = messageRepository.save( + InterviewMessage.followupPlaceholder(session, placeholderSeq, parent)); + // placeholder 를 프론트 메시지 목록에 즉시 노출 → 토큰 델타가 채울 버블 확보. + // REQUIRES_NEW 트랜잭션 commit 후 RealtimeNotifyEventListener 가 발행하므로 조회 보장. + events.publishEvent(RealtimeNotifyEvent.session( + session.getId(), SseEventType.SESSION_MESSAGE, placeholder.getId())); + GenerateFollowupPayload payload = new GenerateFollowupPayload( session.getId(), parent.getId(), @@ -85,6 +97,7 @@ public void onAnswerSubmitted(AnswerSubmittedEvent event) { contextDocumentIds, parent.getCategory(), parent.getExpectedSignal(), + placeholder.getId(), history ); publisher.publishToAi( diff --git a/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java b/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java index 0bc78376..72f67825 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/dto/GenerateFollowupPayload.java @@ -15,6 +15,7 @@ public record GenerateFollowupPayload( List contextDocumentIds, // RAG(자료 근거/correctness) 용 String parentCategory, // 직전 질문 카테고리 (루브릭 선택) String parentExpectedSignal, // 직전 질문이 기대하는 핵심(평가 관점) — 충족도 채점용 + Long followupMessageId, // 스트리밍 표시용 placeholder 메시지 ID (AI 콜백에서 채움) List history // 최근 대화 (중복 회피) ) { public record HistoryItem(String role, String content) { diff --git a/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java b/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java index 28e57fab..626c66a4 100644 --- a/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java +++ b/backend/src/main/java/com/stackup/stackup/session/application/dto/QuestionsCallbackPayload.java @@ -12,7 +12,8 @@ public record QuestionsCallbackPayload( Long answerMessageId, // FOLLOWUP 시 평가 대상 답변 메시지 String followupQuestion, // FOLLOWUP 시 사용 AnswerEvaluation answerEvaluation, // FOLLOWUP 시 옵션 - String answerIntent // NORMAL | DONT_KNOW | CLARIFICATION (FOLLOWUP) + String answerIntent, // NORMAL | DONT_KNOW | CLARIFICATION (FOLLOWUP) + Long followupMessageId // 스트리밍 placeholder 메시지 ID (AI→Core 콜백, null 가능) ) { public boolean isPool() { return "POOL".equals(kind); diff --git a/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java b/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java index 37db87b5..4831454a 100644 --- a/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java +++ b/backend/src/main/java/com/stackup/stackup/session/domain/InterviewMessage.java @@ -36,6 +36,7 @@ public class InterviewMessage extends BaseTimeEntity { public static final String VOICE_TRANSCRIPTION_PENDING_TEXT = "(transcribing)"; public static final String VOICE_TRANSCRIPTION_FAILED_TEXT = "음성 인식에 실패했습니다. 텍스트로 다시 답변해 주세요."; + public static final String FOLLOWUP_GENERATING_TEXT = "(생성 중)"; @Id @GeneratedValue(strategy = GenerationType.IDENTITY) @@ -151,6 +152,14 @@ public static InterviewMessage followup(InterviewSession session, int seq, Strin return m; } + // 스트리밍 꼬리질문 placeholder. content 는 콜백에서 채운다. CHECK 제약(blank 금지) 위해 sentinel. + // TTS 는 생성 시점이 아니라 completeFollowup 에서 요청한다. + public static InterviewMessage followupPlaceholder(InterviewSession session, int seq, + InterviewMessage parent) { + return new InterviewMessage(session, seq, MessageRole.INTERVIEWER, + FOLLOWUP_GENERATING_TEXT, parent, MessageStatus.CREATED, null); + } + // 부연: 직전 질문을 다시 설명해 같은 차례를 유지. 질문 수/깊이에 카운트 안 함. public static InterviewMessage clarification(InterviewSession session, int seq, String content, InterviewMessage parent) { @@ -211,6 +220,16 @@ public void failTts() { this.ttsStatus = TtsStatus.FAILED; } + // 스트리밍 종료 후 placeholder 를 실제 질문으로 확정. clarification=true 면 질문 수/깊이 미반영. + public void completeFollowup(String content, boolean clarification) { + if (content != null && !content.isBlank()) { + this.content = content; + } + this.clarification = clarification; + markTtsPending(); + this.status = MessageStatus.COMPLETED; + } + public void completeWithTranscript(String transcript) { if (transcript != null && !transcript.isBlank()) { this.content = transcript; diff --git a/backend/src/main/java/com/stackup/stackup/session/presentation/InterviewMessageController.java b/backend/src/main/java/com/stackup/stackup/session/presentation/InterviewMessageController.java index 73767fe6..ebc88caf 100644 --- a/backend/src/main/java/com/stackup/stackup/session/presentation/InterviewMessageController.java +++ b/backend/src/main/java/com/stackup/stackup/session/presentation/InterviewMessageController.java @@ -26,6 +26,7 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; @@ -101,4 +102,30 @@ public ResponseEntity audio( .cacheControl(CacheControl.maxAge(Duration.ofHours(1)).cachePrivate()) .body(new InputStreamResource(audio.stream())); } + + @Operation( + operationId = "streamMessageAudioSegment", + summary = "라이브 TTS 문장 세그먼트 스트리밍", + description = "AI 서버가 S3에 쓴 per-sentence TTS 세그먼트를 규칙 기반 키로 프록시한다. " + + "세그먼트는 DB에 미저장(휘발성 라이브)이므로 Core 가 키를 재구성해 스트리밍." + ) + @ApiResponses({ + @ApiResponse(responseCode = "200", description = "세그먼트 오디오 바이트"), + @ApiResponse(responseCode = "401", description = "인증 실패"), + @ApiResponse(responseCode = "404", description = "세션/메시지/세그먼트 없음") + }) + @GetMapping("/{messageId}/audio/segments/{seq}") + public ResponseEntity audioSegment( + @AuthenticationPrincipal UserPrincipal principal, + @PathVariable Long sessionId, + @PathVariable Long messageId, + @PathVariable int seq, + @RequestParam String ext + ) { + var audio = messageService.streamAudioSegment(principal.userId(), sessionId, messageId, seq, ext); + return ResponseEntity.ok() + .contentType(MediaType.parseMediaType(audio.contentType())) + .cacheControl(CacheControl.maxAge(Duration.ofMinutes(10)).cachePrivate()) + .body(new InputStreamResource(audio.stream())); + } } diff --git a/backend/src/test/java/com/stackup/stackup/session/application/InterviewMessageServiceTest.java b/backend/src/test/java/com/stackup/stackup/session/application/InterviewMessageServiceTest.java index 2acea44c..91046c61 100644 --- a/backend/src/test/java/com/stackup/stackup/session/application/InterviewMessageServiceTest.java +++ b/backend/src/test/java/com/stackup/stackup/session/application/InterviewMessageServiceTest.java @@ -3,11 +3,14 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.stackup.stackup.common.exception.DomainException; +import com.stackup.stackup.common.storage.ObjectStorageClient; import com.stackup.stackup.session.application.dto.MessageResult; import com.stackup.stackup.session.application.event.AnswerSubmittedEvent; import com.stackup.stackup.session.domain.InterviewMessage; @@ -30,8 +33,13 @@ @ExtendWith(MockitoExtension.class) class InterviewMessageServiceTest { + private static final Long USER_ID = 1L; + private static final Long SESSION_ID = 10L; + private static final Long MESSAGE_ID = 100L; + @Mock InterviewSessionRepository sessionRepository; @Mock InterviewMessageRepository messageRepository; + @Mock ObjectStorageClient storage; @Mock ApplicationEventPublisher events; @InjectMocks InterviewMessageService service; @@ -114,9 +122,39 @@ void submitAnswer_allowsTextRetryAfterFailedVoiceAnswer() { assertThat(eventCaptor.getValue().answerMessageId()).isEqualTo(201L); } + // ── 라이브 TTS 세그먼트 프록시 테스트 ───────────────────────────────────────── + + @Test + void streamAudioSegment_rejects_unknown_extension() { + // ownedSession 통과하도록 sessionRepository.findByIdAndUser_IdAndDeletedFalse 스텁 + given(sessionRepository.findByIdAndUser_IdAndDeletedFalse(SESSION_ID, USER_ID)) + .willReturn(Optional.of(sessionInProgress(SESSION_ID))); + assertThatThrownBy(() -> service.streamAudioSegment(USER_ID, SESSION_ID, MESSAGE_ID, 0, "exe")) + .isInstanceOf(DomainException.class); + } + + @Test + void streamAudioSegment_builds_convention_key_and_streams() { + InterviewSession session = sessionInProgress(SESSION_ID); + InterviewMessage message = InterviewMessage.interviewer(session, 1, "Q?"); + ReflectionTestUtils.setField(message, "id", MESSAGE_ID); + + given(sessionRepository.findByIdAndUser_IdAndDeletedFalse(SESSION_ID, USER_ID)) + .willReturn(Optional.of(session)); + given(messageRepository.findById(MESSAGE_ID)).willReturn(Optional.of(message)); + given(storage.get(anyString())).willReturn(new java.io.ByteArrayInputStream(new byte[]{1, 2, 3})); + + var audio = service.streamAudioSegment(USER_ID, SESSION_ID, MESSAGE_ID, 2, "mp3"); + + verify(storage).get("interview/tts/%d/%d/seg-2.mp3".formatted(SESSION_ID, MESSAGE_ID)); + assertThat(audio.contentType()).isEqualTo("audio/mpeg"); + } + + // ───────────────────────────────────────────────────────────────────────────── + private InterviewSession sessionInProgress(Long id) { User user = User.createGithubUser(1L, "u", null, null, "t"); - ReflectionTestUtils.setField(user, "id", 1L); + ReflectionTestUtils.setField(user, "id", USER_ID); InterviewSession s = InterviewSession.create( user, "t", null, SessionMode.TECHNICAL, JobCategory.BACKEND, 5, 30, null, null ); diff --git a/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java b/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java index c6572204..a32695f6 100644 --- a/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java +++ b/backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java @@ -139,7 +139,7 @@ void apply_followupPersistsAnswerEvaluationOntoAnswerMessage() { QuestionsCallbackPayload payload = new QuestionsCallbackPayload( 11L, "FOLLOWUP", null, 500L, 600L, "다음 질문?", - new QuestionsCallbackPayload.AnswerEvaluation(2.0, 3.0, "PARTIAL_STAR", 1.5), "NORMAL" + new QuestionsCallbackPayload.AnswerEvaluation(2.0, 3.0, "PARTIAL_STAR", 1.5), "NORMAL", null ); QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( "m-eval", "callback.questions", "1", "t", null, "ai", payload, null); @@ -165,7 +165,7 @@ void apply_clarificationReExplainsWithoutCountingQuestion() { InterviewSession session = sessionFixture(11L, SessionStatus.IN_PROGRESS); QuestionsCallbackPayload payload = new QuestionsCallbackPayload( 11L, "FOLLOWUP", null, 500L, 600L, "쉽게 다시 설명: 트랜잭션이란…", - null, "CLARIFICATION" + null, "CLARIFICATION", null ); QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( "m-clar", "callback.questions", "1", "t", null, "ai", payload, null); @@ -185,16 +185,112 @@ void apply_clarificationReExplainsWithoutCountingQuestion() { assertThat(session.getTotalQuestionCount()).isEqualTo(0); } + // ── 새 placeholder 경로 테스트 ───────────────────────────────────────────── + + @Test + void apply_followupNormal_updatesPlaceholderInPlaceAndCounts() { + InterviewSession session = sessionFixture(20L, SessionStatus.IN_PROGRESS); + // placeholder: followupPlaceholder 의 sentinel content + InterviewMessage placeholder = InterviewMessage.followupPlaceholder( + session, 3, parentMessageFixture(session)); + ReflectionTestUtils.setField(placeholder, "id", 301L); + + QuestionsCallbackPayload payload = new QuestionsCallbackPayload( + 20L, "FOLLOWUP", null, 200L, null, "꼬리질문 내용?", + null, "NORMAL", 301L + ); + QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( + "m-ph-normal", "callback.questions", "1", "t", null, "ai", payload, null); + + when(processedMessageRepository.existsById("m-ph-normal")).thenReturn(false); + when(sessionRepository.findById(20L)).thenReturn(Optional.of(session)); + when(messageRepository.findById(200L)).thenReturn(Optional.of(parentMessageFixture(session))); + when(messageRepository.findById(301L)).thenReturn(Optional.of(placeholder)); + when(messageRepository.save(any(InterviewMessage.class))).thenAnswer(inv -> inv.getArgument(0)); + + service.apply(env); + + // placeholder 가 in-place 로 업데이트되어야 함 + assertThat(placeholder.getContent()).isEqualTo("꼬리질문 내용?"); + assertThat(placeholder.getStatus()).isEqualTo(com.stackup.stackup.session.domain.MessageStatus.COMPLETED); + // 질문 수 카운트 증가 + assertThat(session.getTotalQuestionCount()).isEqualTo(1); + // save 는 placeholder 자체에 대해 호출됨 (새 InterviewMessage INSERT 아님) + ArgumentCaptor cap = ArgumentCaptor.forClass(InterviewMessage.class); + verify(messageRepository).save(cap.capture()); + assertThat(cap.getValue()).isSameAs(placeholder); + } + + @Test + void apply_followupClarification_updatesPlaceholderWithoutCounting() { + InterviewSession session = sessionFixture(21L, SessionStatus.IN_PROGRESS); + InterviewMessage placeholder = InterviewMessage.followupPlaceholder( + session, 3, parentMessageFixture(session)); + ReflectionTestUtils.setField(placeholder, "id", 302L); + + QuestionsCallbackPayload payload = new QuestionsCallbackPayload( + 21L, "FOLLOWUP", null, 200L, null, "다시 설명드리면…", + null, "CLARIFICATION", 302L + ); + QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( + "m-ph-clar", "callback.questions", "1", "t", null, "ai", payload, null); + + when(processedMessageRepository.existsById("m-ph-clar")).thenReturn(false); + when(sessionRepository.findById(21L)).thenReturn(Optional.of(session)); + when(messageRepository.findById(200L)).thenReturn(Optional.of(parentMessageFixture(session))); + when(messageRepository.findById(302L)).thenReturn(Optional.of(placeholder)); + when(messageRepository.save(any(InterviewMessage.class))).thenAnswer(inv -> inv.getArgument(0)); + + service.apply(env); + + // clarification 플래그가 세팅되어야 함 + assertThat(placeholder.isClarification()).isTrue(); + assertThat(placeholder.getContent()).isEqualTo("다시 설명드리면…"); + // 질문 수 카운트 미증가 + assertThat(session.getTotalQuestionCount()).isEqualTo(0); + } + + @Test + void apply_followupDontKnow_deletesPlaceholderAndAdvances() { + InterviewSession session = sessionFixture(22L, SessionStatus.IN_PROGRESS); + InterviewMessage placeholder = InterviewMessage.followupPlaceholder( + session, 3, parentMessageFixture(session)); + ReflectionTestUtils.setField(placeholder, "id", 303L); + + QuestionsCallbackPayload payload = new QuestionsCallbackPayload( + 22L, "FOLLOWUP", null, 200L, null, null, + null, "DONT_KNOW", 303L + ); + QuestionsCallbackEnvelope env = new QuestionsCallbackEnvelope( + "m-ph-dk", "callback.questions", "1", "t", null, "ai", payload, null); + + when(processedMessageRepository.existsById("m-ph-dk")).thenReturn(false); + when(sessionRepository.findById(22L)).thenReturn(Optional.of(session)); + when(messageRepository.findById(200L)).thenReturn(Optional.of(parentMessageFixture(session))); + when(messageRepository.findById(303L)).thenReturn(Optional.of(placeholder)); + // advanceToNextGeneral 내부에서 sessionRepository.findById 재호출 + when(poolRepository.findFirstBySessionIdAndUsedFalseOrderByIdxAsc(22L)) + .thenReturn(java.util.Optional.empty()); + + service.apply(env); + + // placeholder 가 삭제되어야 함 + verify(messageRepository).delete(placeholder); + verify(messageRepository).flush(); + // 세션이 종료됨 (풀이 비어있어 POOL_EXHAUSTED) + assertThat(session.getStatus()).isEqualTo(SessionStatus.COMPLETED); + } + private QuestionsCallbackEnvelope poolEnvelope(Long sessionId, List questions) { QuestionsCallbackPayload payload = new QuestionsCallbackPayload( - sessionId, "POOL", questions, null, null, null, null, null + sessionId, "POOL", questions, null, null, null, null, null, null ); return new QuestionsCallbackEnvelope("m-1", "callback.questions", "1", "t", null, "ai", payload, null); } private QuestionsCallbackEnvelope followupEnvelope(Long sessionId, Long parentId, String followup) { QuestionsCallbackPayload payload = new QuestionsCallbackPayload( - sessionId, "FOLLOWUP", null, parentId, null, followup, null, "NORMAL" + sessionId, "FOLLOWUP", null, parentId, null, followup, null, "NORMAL", null ); return new QuestionsCallbackEnvelope("m-2", "callback.questions", "1", "t", null, "ai", payload, null); } diff --git a/backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java b/backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java new file mode 100644 index 00000000..e9cb2ccc --- /dev/null +++ b/backend/src/test/java/com/stackup/stackup/session/domain/InterviewMessageTest.java @@ -0,0 +1,60 @@ +package com.stackup.stackup.session.domain; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.stackup.stackup.user.domain.User; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; + +class InterviewMessageTest { + + private InterviewSession session; + + @BeforeEach + void setUp() { + User user = User.createGithubUser(1L, "u", null, null, "t"); + ReflectionTestUtils.setField(user, "id", 1L); + session = InterviewSession.create( + user, "테스트 세션", null, SessionMode.TECHNICAL, JobCategory.BACKEND, 5, 30, null, null + ); + ReflectionTestUtils.setField(session, "id", 10L); + } + + @Test + void followupPlaceholder_hasSentinelContent_createdStatus_interviewerRole_ttsNotRequested() { + InterviewMessage parent = InterviewMessage.interviewer(session, 1, "원질문"); + + InterviewMessage ph = InterviewMessage.followupPlaceholder(session, 2, parent); + + assertThat(ph.getContent()).isEqualTo(InterviewMessage.FOLLOWUP_GENERATING_TEXT); + assertThat(ph.getStatus()).isEqualTo(MessageStatus.CREATED); + assertThat(ph.getTtsStatus()).isEqualTo(TtsStatus.NOT_REQUESTED); + assertThat(ph.getRole()).isEqualTo(MessageRole.INTERVIEWER); + } + + @Test + void completeFollowup_fillsContentMarksCompletedAndTtsPending() { + InterviewMessage parent = InterviewMessage.interviewer(session, 1, "원질문"); + InterviewMessage ph = InterviewMessage.followupPlaceholder(session, 2, parent); + + ph.completeFollowup("진짜 꼬리질문?", false); + + assertThat(ph.getContent()).isEqualTo("진짜 꼬리질문?"); + assertThat(ph.isClarification()).isFalse(); + assertThat(ph.getStatus()).isEqualTo(MessageStatus.COMPLETED); + assertThat(ph.getTtsStatus()).isEqualTo(TtsStatus.PENDING); + } + + @Test + void completeFollowup_withClarificationTrue_setsClarificationFlag() { + InterviewMessage parent = InterviewMessage.interviewer(session, 1, "원질문"); + InterviewMessage ph2 = InterviewMessage.followupPlaceholder(session, 3, parent); + + ph2.completeFollowup("다시 설명하면…", true); + + assertThat(ph2.isClarification()).isTrue(); + assertThat(ph2.getStatus()).isEqualTo(MessageStatus.COMPLETED); + assertThat(ph2.getTtsStatus()).isEqualTo(TtsStatus.PENDING); + } +} diff --git a/docs/event-stream.md b/docs/event-stream.md index 1263fed2..0ab34bbe 100644 --- a/docs/event-stream.md +++ b/docs/event-stream.md @@ -40,7 +40,7 @@ WS /realtime/sessions/{sessionId}/audio # RT3 실시간 음성 답변 스트 ## 2. 이벤트 포맷 -> - `event` 이름은 **`SseEventType` enum 이름(대문자)**: `DOC_STATE`·`REPO_STATE`·`SESSION_MESSAGE`·`SESSION_STATE`·`FEEDBACK_READY`·`ERROR`·`KEEP_ALIVE` (+ AI 가 직접 발행하는 `ANALYSIS_PROGRESS`). **`session.message` 같은 소문자 점표기가 아니다.** (RealTime `bridge/dispatcher.go` `Type: env.Payload.EventType` → `sse.go`/`ws.go` 가 그대로 전달.) +> - `event` 이름은 **`SseEventType` enum 이름(대문자)**: `DOC_STATE`·`REPO_STATE`·`SESSION_MESSAGE`·`SESSION_STATE`·`FEEDBACK_READY`·`ERROR`·`KEEP_ALIVE` (+ AI 가 직접 발행하는 `ANALYSIS_PROGRESS`·`SESSION_MESSAGE_DELTA`·`SESSION_MESSAGE_AUDIO`). **`session.message` 같은 소문자 점표기가 아니다.** (RealTime `bridge/dispatcher.go` `Type: env.Payload.EventType` → `sse.go`/`ws.go` 가 그대로 전달.) RealTime 디스패처는 이벤트 타입을 화이트리스트 없이 투명 전달하므로, AI 가 새 이벤트 타입(`SESSION_MESSAGE_DELTA` 등)을 발행해도 RealTime 코드 변경이 필요 없다. > - `data` 봉투는 `{"data": , "traceId": "..."}` 다 (`realtime/CLAUDE.md §8`). payload 필드는 camelCase. > - 클라는 SSE `addEventListener(, …)` / WS `frame.event === ''` 로 매칭한다. @@ -116,6 +116,31 @@ WS(RT1)는 같은 내용을 JSON 한 줄 프레임으로: `{ "id": , "e { "data": 503, "traceId": "..." } ``` +### 3.3-1 꼬리질문 토큰 스트리밍 (`event: SESSION_MESSAGE_DELTA`) + +**휘발성** 이벤트. 꼬리질문을 ChatGPT처럼 토큰 단위로 흘려 체감 반응을 높인다. **AI 서버가 `stackup.realtime`(`realtime.session.notify`)으로 직접 발행**(Core·DB 미경유) → RealTime 이 세션 채널로 fan-out. `SESSION_MESSAGE`(종료, messageId만)와 짝을 이루는 진행 이벤트다 (`ANALYSIS_PROGRESS`↔`DOC_STATE` 관계와 동일 패턴). + +```json +{ "data": { "messageId": 503, "seq": 0, "text": "그 설계에서 " }, "traceId": "..." } +``` + +- `messageId`: Core 가 답변 직후 선INSERT 한 INTERVIEWER **placeholder** 메시지 id(content=`"(생성 중)"`, status=`CREATED`). 생성 시 Core 가 `SESSION_MESSAGE(placeholderId)` 를 1회 발행하므로 프론트 목록에 placeholder 버블이 먼저 뜬다. +- `seq`: 0부터 단조 증가. `text`: 이번 델타에서 **추가된 조각**(누적 아님). 프론트는 placeholder 버블에 append. +- 흐름: 답변 → (Core) placeholder INSERT + `SESSION_MESSAGE` → (AI) `astream` 으로 `SESSION_MESSAGE_DELTA` 연속 발행 → (AI) `callback.questions(FOLLOWUP, followupMessageId)` → (Core) placeholder UPDATE(content/COMPLETED) + `SESSION_MESSAGE`(종료) → 프론트 `GET …/messages` 재조회로 정본 reconcile. +- `answer_intent=DONT_KNOW` 면 AI 가 델타를 **발행하지 않고**, Core 가 placeholder 삭제 후 다음 일반질문으로 진행한다. 이때 프론트는 placeholder 를 "생각 중"으로만 표시하다 일반질문으로 교체. + +### 3.3-2 꼬리질문 문장 단위 TTS (`event: SESSION_MESSAGE_AUDIO`) + +**휘발성** 이벤트. 꼬리질문이 토큰으로 차오르는 동안, **문장이 완성될 때마다 그 문장만 TTS 합성**해 첫 소리까지의 지연을 단축한다. AI 서버가 인라인 합성 후 `stackup.realtime`(`realtime.session.notify`)으로 직접 발행. + +```json +{ "data": { "messageId": 503, "seq": 0, "ext": "mp3", "durationSec": 1.2 }, "traceId": "..." } +``` + +- `messageId` = `SESSION_MESSAGE_DELTA` 와 동일 placeholder id. `seq` = 오디오 세그먼트 순번(0부터, **델타 seq 와 독립**). `ext` ∈ `wav|mp3|ogg|m4a`. +- 세그먼트는 S3 `interview/tts/{sessionId}/{messageId}/seg-{seq}.{ext}` 에 저장(DB 미기록, 휘발성). 프론트는 Core 프록시 `GET /api/sessions/{sid}/messages/{mid}/audio/segments/{seq}?ext=` 로 받아 **seq 순서대로 순차 재생**. +- `DONT_KNOW` 면 발행 안 함. 라이브 세그먼트를 재생한 메시지는 완료 후 whole-message TTS **autoPlay 억제**(중복 재생 방지) — 수동 "다시 듣기"만 동작. + ### 3.4 세션 상태 (`event: SESSION_STATE`) ```json { diff --git a/docs/messaging.md b/docs/messaging.md index 6ab55b7b..dddb11d8 100644 --- a/docs/messaging.md +++ b/docs/messaging.md @@ -263,12 +263,14 @@ "contextDocumentIds": [12, 13], "parentCategory": "PROJECT_DEEP_DIVE", "parentExpectedSignal": "동시성 제어를 DB 레벨까지 설명하는지", + "followupMessageId": 503, "history": [{ "role": "INTERVIEWER", "content": "..." }] } } ``` > `parentExpectedSignal` = 직전 질문 생성 시 만든 기대 신호(평가 관점). AI 가 specificity/correctness 채점의 핵심 기준으로 사용(없으면 무시). `contextDocumentIds` 로 RAG 검색 → correctness 판정. +> `followupMessageId` = Core 가 답변 직후 선INSERT 한 INTERVIEWER placeholder(content=`"(생성 중)"`) 메시지 id. AI 는 이 id 로 `SESSION_MESSAGE_DELTA` 토큰을 태깅해 발행하고(§5.12-1), 콜백에도 되돌려준다. Core 는 콜백 시 이 id 의 placeholder 를 UPDATE(NORMAL/CLARIFICATION) 또는 DELETE(DONT_KNOW) 한다. ### 5.9 `callback.questions` (꼬리질문) ```json @@ -278,6 +280,8 @@ "sessionId": 99, "kind": "FOLLOWUP", "parentMessageId": 502, + "followupMessageId": 503, + "answerIntent": "NORMAL", "followupQuestion": "...", "answerEvaluation": { "specificity": 3.5, @@ -390,6 +394,16 @@ - 발행: AI `AnalysisProgressNotifier` (envelope 구조는 Core `RealtimeNotifyPublisher` 와 동일 — `payload.eventType` + `context.userId`). - `payload.eventType` = `ANALYSIS_PROGRESS`, `payload.data` = `{ targetType, targetId, phase, message }`. phase ∈ `EXTRACTING | SUMMARIZING | EMBEDDING`. + +### 5.12-1 `realtime.session.notify` — 꼬리질문 토큰 델타 (AI 직접 발행) +분석 진행(ANALYSIS_PROGRESS)과 동일 패턴으로, AI 서버가 꼬리질문 생성 중 토큰을 `stackup.realtime` exchange 의 `realtime.session.notify` 로 **직접** 발행(Core 미경유)해 세션 채널로 흘린다. 휘발성(영속 불필요)이며 발행 실패는 무시(경고만). +- `context.sessionId` 로 세션 채널 라우팅. `payload.eventType` = `SESSION_MESSAGE_DELTA`, `payload.data` = `{ messageId, seq, text }` (§event-stream.md 3.3-1). +- `messageId` = `generate.followup` 의 `followupMessageId`(placeholder). 종료/정본은 기존대로 Core 가 `callback.questions` 수신 후 `SESSION_MESSAGE`(messageId) 로 통지. + +### 5.12-2 `realtime.session.notify` — 문장 단위 TTS 세그먼트 (AI 직접 발행) +AI followup consumer 가 토큰 스트림 중 문장 경계마다 그 문장만 인라인 TTS 합성 → S3 세그먼트 PUT → `SESSION_MESSAGE_AUDIO` 를 `realtime.session.notify` 로 직접 발행(휘발성). 문장 합성은 `asyncio.create_task` 백그라운드(텍스트 델타 비차단), 콜백 발행 전 `gather` 로 수거. +- `payload.eventType` = `SESSION_MESSAGE_AUDIO`, `payload.data` = `{ messageId, seq, ext, durationSec }`. `seq` 는 오디오 전용(델타 seq 와 독립). +- 세그먼트 S3 키 규칙(AI·Core 공유): `interview/tts/{sessionId}/{messageId}/seg-{seq}.{ext}`. Core 는 DB 미기록, `GET …/messages/{mid}/audio/segments/{seq}?ext=` 프록시에서 소유권 검증 후 규칙으로 키 재구성(ext 화이트리스트 wav|mp3|ogg|m4a). - 상세 SSE 스펙: [`event-stream.md §3.2-1`](./event-stream.md). --- diff --git a/docs/storage.md b/docs/storage.md index a95ac3b6..ee7fd10c 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -24,10 +24,14 @@ analyzed/resume/{resume_id}/summary.md analyzed/repository/{repository_id}/summary.md session/{session_id}/audio/{message_id}.webm session/{session_id}/audio/{message_id}.transcript.txt +interview/tts/{session_id}/{message_id}.{ext} # 질문 whole-message TTS (영속, ttsAudioPath) +interview/tts/{session_id}/{message_id}/seg-{seq}.{ext} # 꼬리질문 문장 단위 TTS 세그먼트 (휘발성, DB 미기록) feedback/{session_id}/report.md feedback/{session_id}/charts/{name}.png ``` +> 세그먼트 키 규칙은 **AI(발행)·Core(프록시 재구성) 양측이 하드코딩으로 공유**하는 SSOT다 — 변경 시 양측 동시 수정. `{ext}` ∈ `wav|mp3|ogg|m4a`, `{seq}` 는 0부터의 오디오 세그먼트 순번. 세그먼트는 DB에 기록하지 않고 Core 프록시(`GET …/audio/segments/{seq}?ext=`)가 소유권 검증 후 규칙으로 키를 만든다. + 규칙: - 슬래시(`/`) 계층 구조로 탐색성 확보 - 사용자 식별자는 user_id (정수) 사용 — github_username은 변경 가능하므로 회피 diff --git a/frontend/src/domain/session/index.ts b/frontend/src/domain/session/index.ts index 8b359c08..a2aa7441 100644 --- a/frontend/src/domain/session/index.ts +++ b/frontend/src/domain/session/index.ts @@ -12,6 +12,7 @@ export { isAnswer, isTranscribing, VOICE_TRANSCRIBING_TEXT, + FOLLOWUP_GENERATING_TEXT, currentTurn, canSubmitAnswer, sessionProgress, diff --git a/frontend/src/domain/session/lib/turn.test.ts b/frontend/src/domain/session/lib/turn.test.ts index 95b86db2..e6e00c3d 100644 --- a/frontend/src/domain/session/lib/turn.test.ts +++ b/frontend/src/domain/session/lib/turn.test.ts @@ -1,6 +1,6 @@ import { describe, it, expect } from 'vitest' import type { Message, Session } from '../model/types' -import { currentTurn, canSubmitAnswer, sessionProgress, isQuestion, isAnswer } from './turn' +import { currentTurn, canSubmitAnswer, sessionProgress, isQuestion, isAnswer, FOLLOWUP_GENERATING_TEXT } from './turn' const msg = (role: 'INTERVIEWER' | 'INTERVIEWEE', content = '내용'): Message => ({ role, content } as Message) @@ -19,6 +19,9 @@ describe('currentTurn', () => { it('내용 없는 질문(생성 중)이면 질문 대기', () => { expect(currentTurn([msg('INTERVIEWER', '')])).toBe('WAITING_FOR_QUESTION') }) + it('생성 중 sentinel 질문이면 질문 대기', () => { + expect(currentTurn([msg('INTERVIEWER', FOLLOWUP_GENERATING_TEXT)])).toBe('WAITING_FOR_QUESTION') + }) }) describe('canSubmitAnswer', () => { diff --git a/frontend/src/domain/session/lib/turn.ts b/frontend/src/domain/session/lib/turn.ts index f3f455b0..d958b043 100644 --- a/frontend/src/domain/session/lib/turn.ts +++ b/frontend/src/domain/session/lib/turn.ts @@ -14,6 +14,10 @@ export function isAnswer(message: Message): boolean { // STT 완료 전까지 이 값이 들어 있고, 완료되면 실제 transcript 로 교체된다. export const VOICE_TRANSCRIBING_TEXT = '(transcribing)' +// 꼬리질문 생성 중인 placeholder content (백엔드 sentinel 과 동일). +// 스트리밍이 완료되면 실제 질문 텍스트로 교체된다. +export const FOLLOWUP_GENERATING_TEXT = '(생성 중)' + // STT 대기 중(transcript 아직 없음)인 음성 답변인지. export function isTranscribing(message: Message): boolean { if (!isAnswer(message)) return false @@ -23,10 +27,14 @@ export function isTranscribing(message: Message): boolean { export function currentTurn(messages: Message[]): Turn { const last = messages[messages.length - 1] - // 마지막이 '내용이 있는' 면접관 질문일 때만 답변 차례. 질문 생성 중(빈 content) - // 메시지가 먼저 도착해도 컴포저가 섣불리 활성화되지 않도록 content 유무로 가드한다. + // 마지막이 '내용이 있는' 면접관 질문일 때만 답변 차례. 질문 생성 중(빈 content 또는 sentinel) + // 메시지가 먼저 도착해도 컴포저가 섣불리 활성화되지 않도록 content 유무 + sentinel 로 가드한다. // (질문 status 는 CREATED 로 들어오므로 status 기준으로 거르면 안 된다.) - const askable = last?.role === 'INTERVIEWER' && (last.content ?? '').trim().length > 0 + const content = (last?.content ?? '').trim() + const askable = + last?.role === 'INTERVIEWER' && + content.length > 0 && + content !== FOLLOWUP_GENERATING_TEXT return askable ? 'AWAITING_ANSWER' : 'WAITING_FOR_QUESTION' } diff --git a/frontend/src/features/interview/api/messageApi.ts b/frontend/src/features/interview/api/messageApi.ts index 1b676d13..a8281a04 100644 --- a/frontend/src/features/interview/api/messageApi.ts +++ b/frontend/src/features/interview/api/messageApi.ts @@ -46,3 +46,14 @@ export async function fetchMessageAudioObjectUrl( ) return URL.createObjectURL(data) } + +// 라이브 문장 TTS 세그먼트를 Core 프록시로 받아 object URL 로 변환. +export async function fetchMessageSegmentObjectUrl( + sessionId: number, messageId: number, seq: number, ext: string, +): Promise { + const { data } = await apiClient.get( + `/api/sessions/${sessionId}/messages/${messageId}/audio/segments/${seq}`, + { responseType: 'blob', params: { ext } }, + ) + return URL.createObjectURL(data) +} diff --git a/frontend/src/features/interview/lib/media/segmentAudioQueue.test.ts b/frontend/src/features/interview/lib/media/segmentAudioQueue.test.ts new file mode 100644 index 00000000..da4d1577 --- /dev/null +++ b/frontend/src/features/interview/lib/media/segmentAudioQueue.test.ts @@ -0,0 +1,25 @@ +import { describe, it, expect, vi } from 'vitest' +import { createSegmentQueue } from './segmentAudioQueue' + +describe('segmentAudioQueue', () => { + it('seq 순서대로 하나씩 재생하고 ended 후 다음으로', async () => { + const played: string[] = [] + const play = vi.fn(async (url: string) => { played.push(url) }) + const q = createSegmentQueue(play) + q.enqueue({ seq: 1, url: 'u1' }) // 뒤섞여 도착 + q.enqueue({ seq: 0, url: 'u0' }) + await Promise.resolve(); await Promise.resolve() + expect(played[0]).toBe('u0') // seq0 먼저 + q.onEnded() + await Promise.resolve(); await Promise.resolve() + expect(played[1]).toBe('u1') + }) + it('reset 후에는 expected seq 가 0 으로 돌아간다', async () => { + const played: string[] = [] + const q = createSegmentQueue(async (u) => { played.push(u) }) + q.enqueue({ seq: 0, url: 'a' }); await Promise.resolve(); await Promise.resolve() + q.reset() + q.enqueue({ seq: 0, url: 'b' }); await Promise.resolve(); await Promise.resolve() + expect(played).toEqual(['a', 'b']) + }) +}) diff --git a/frontend/src/features/interview/lib/media/segmentAudioQueue.ts b/frontend/src/features/interview/lib/media/segmentAudioQueue.ts new file mode 100644 index 00000000..75ecd3bd --- /dev/null +++ b/frontend/src/features/interview/lib/media/segmentAudioQueue.ts @@ -0,0 +1,24 @@ +export type Segment = { seq: number; url: string } + +// seq 순서로 한 번에 하나씩 재생. play(url) 는 재생 시작(Promise). onEnded() 로 다음 진행. +// reset() 으로 새 질문 시작 시 seq 카운터/대기열 초기화. +export function createSegmentQueue(play: (url: string) => Promise) { + const pending: Segment[] = [] + let expected = 0 + let busy = false + + function drain() { + if (busy) return + const i = pending.findIndex((s) => s.seq === expected) + if (i === -1) return + const [seg] = pending.splice(i, 1) + busy = true + expected += 1 + void play(seg.url) + } + return { + enqueue(seg: Segment) { pending.push(seg); drain() }, + onEnded() { busy = false; drain() }, + reset() { pending.length = 0; expected = 0; busy = false }, + } +} diff --git a/frontend/src/features/interview/model/interviewEvent.test.ts b/frontend/src/features/interview/model/interviewEvent.test.ts index 4fbb83d4..1b12768b 100644 --- a/frontend/src/features/interview/model/interviewEvent.test.ts +++ b/frontend/src/features/interview/model/interviewEvent.test.ts @@ -17,4 +17,10 @@ describe('interviewEventAction', () => { it('알 수 없는 이벤트는 무시', () => { expect(interviewEventAction('whatever')).toEqual({ kind: 'ignore' }) }) + it('SESSION_MESSAGE_DELTA → 델타 append', () => { + expect(interviewEventAction('SESSION_MESSAGE_DELTA')).toEqual({ kind: 'append-delta' }) + }) + it('SESSION_MESSAGE_AUDIO → 오디오 큐', () => { + expect(interviewEventAction('SESSION_MESSAGE_AUDIO')).toEqual({ kind: 'queue-audio' }) + }) }) diff --git a/frontend/src/features/interview/model/interviewEvent.ts b/frontend/src/features/interview/model/interviewEvent.ts index 6f4ccd42..dd61c8e7 100644 --- a/frontend/src/features/interview/model/interviewEvent.ts +++ b/frontend/src/features/interview/model/interviewEvent.ts @@ -2,6 +2,8 @@ export type InterviewAction = | { kind: 'refetch-messages' } | { kind: 'refetch-session' } | { kind: 'redirect-feedback' } + | { kind: 'append-delta' } + | { kind: 'queue-audio' } | { kind: 'ignore' } export function interviewEventAction(event: string): InterviewAction { @@ -12,6 +14,10 @@ export function interviewEventAction(event: string): InterviewAction { return { kind: 'refetch-session' } case 'FEEDBACK_READY': return { kind: 'redirect-feedback' } + case 'SESSION_MESSAGE_DELTA': + return { kind: 'append-delta' } + case 'SESSION_MESSAGE_AUDIO': + return { kind: 'queue-audio' } default: return { kind: 'ignore' } } diff --git a/frontend/src/features/interview/model/streamingBuffer.test.ts b/frontend/src/features/interview/model/streamingBuffer.test.ts new file mode 100644 index 00000000..38e92320 --- /dev/null +++ b/frontend/src/features/interview/model/streamingBuffer.test.ts @@ -0,0 +1,16 @@ +import { describe, it, expect } from 'vitest' +import { applyDelta, isStreamingMessage, FOLLOWUP_GENERATING_TEXT } from './streamingBuffer' + +describe('streamingBuffer', () => { + it('applyDelta 는 messageId별로 누적한다', () => { + let buf: Record = {} + buf = applyDelta(buf, { messageId: 5, seq: 0, text: '안녕' }) + buf = applyDelta(buf, { messageId: 5, seq: 1, text: '하세요' }) + expect(buf[5]).toBe('안녕하세요') + }) + it('isStreamingMessage 는 sentinel content + 버퍼 유무로 판별', () => { + expect(isStreamingMessage({ content: FOLLOWUP_GENERATING_TEXT }, undefined)).toBe(true) + expect(isStreamingMessage({ content: FOLLOWUP_GENERATING_TEXT }, '안녕')).toBe(true) + expect(isStreamingMessage({ content: '완성된 질문' }, undefined)).toBe(false) + }) +}) diff --git a/frontend/src/features/interview/model/streamingBuffer.ts b/frontend/src/features/interview/model/streamingBuffer.ts new file mode 100644 index 00000000..6bc99062 --- /dev/null +++ b/frontend/src/features/interview/model/streamingBuffer.ts @@ -0,0 +1,21 @@ +import { FOLLOWUP_GENERATING_TEXT } from '@/domain/session' +export { FOLLOWUP_GENERATING_TEXT } + +export type DeltaPayload = { messageId: number; seq: number; text: string } + +// messageId -> 누적 텍스트. seq 는 순서 보조(현재는 단순 append, gap 은 종료 reconcile 로 자기치유). +export function applyDelta( + buffer: Record, + delta: DeltaPayload, +): Record { + const prev = buffer[delta.messageId] ?? '' + return { ...buffer, [delta.messageId]: prev + delta.text } +} + +export function isStreamingMessage( + message: { content?: string | null }, + bufferedText: string | undefined, +): boolean { + if (bufferedText !== undefined) return true + return message.content === FOLLOWUP_GENERATING_TEXT +} diff --git a/frontend/src/features/interview/model/useLiveInterview.ts b/frontend/src/features/interview/model/useLiveInterview.ts index 101f5108..2d70a915 100644 --- a/frontend/src/features/interview/model/useLiveInterview.ts +++ b/frontend/src/features/interview/model/useLiveInterview.ts @@ -1,9 +1,10 @@ -import { useCallback, useEffect, useState } from 'react' +import { useCallback, useEffect, useRef, useState } from 'react' import { useNavigate } from 'react-router-dom' import { useMutation, useQueryClient } from '@tanstack/react-query' import { currentTurn } from '@/domain/session' import type { Message } from '@/domain/session' -import { submitVoiceAnswer } from '../api/messageApi' +import { submitVoiceAnswer, fetchMessageSegmentObjectUrl } from '../api/messageApi' +import { createSegmentQueue } from '../lib/media/segmentAudioQueue' import { sessionKeys, useSession } from './useSession' import { messageKeys, useSessionMessages } from './useSessionMessages' import { useSessionLifecycle } from './useSessionLifecycle' @@ -11,6 +12,8 @@ import { useInterviewSocket } from './useInterviewSocket' import { interviewEventAction } from './interviewEvent' import { pendingAnswers, toOptimisticMessage } from './optimistic' import type { OptimisticAnswer } from './optimistic' +import { applyDelta, isStreamingMessage, FOLLOWUP_GENERATING_TEXT } from './streamingBuffer' +import type { DeltaPayload } from './streamingBuffer' export type ThreadItem = Message & { key: string } export type ConnectionStatus = 'connecting' | 'open' | 'closed' @@ -25,17 +28,59 @@ export function useLiveInterview(sessionId: number) { const [optimistic, setOptimistic] = useState([]) const [connection, setConnection] = useState('connecting') + const [deltaBuffer, setDeltaBuffer] = useState>({}) + + // ── 라이브 세그먼트 TTS ────────────────────────────────────────────────── + const audioElRef = useRef(null) + const lastUrlRef = useRef(null) + const queueRef = useRef | null>(null) + const currentAudioMsgId = useRef(null) + const segmentedIds = useRef>(new Set()) + if (queueRef.current === null) { + queueRef.current = createSegmentQueue(async (url) => { + if (typeof Audio === 'undefined') return + if (!audioElRef.current) audioElRef.current = new Audio() + const el = audioElRef.current + if (lastUrlRef.current) URL.revokeObjectURL(lastUrlRef.current) + lastUrlRef.current = url + el.src = url + try { await el.play() } catch { /* autoplay 차단 등은 무시 */ } + }) + } // 서버가 sequenceNumber asc 로 주지만, 순서를 코드에서 명시적으로 보장한다. const serverMessages = [...(messagesQuery.data ?? [])].sort( (a, b) => (a.sequenceNumber ?? 0) - (b.sequenceNumber ?? 0), ) const pending = pendingAnswers(optimistic, serverMessages) + + // 스트리밍 중인 메시지는 deltaBuffer의 누적 텍스트로 content를 오버라이드한다. + const mergedMessages = serverMessages.map((m) => { + const buffered = deltaBuffer[m.id ?? -1] + if (buffered !== undefined && isStreamingMessage(m, buffered)) { + return { ...m, content: buffered } + } + return m + }) + const items: ThreadItem[] = [ - ...serverMessages.map((m) => ({ ...m, key: `m-${m.id}` })), + ...mergedMessages.map((m) => ({ ...m, key: `m-${m.id}` })), ...pending.map((o) => ({ ...toOptimisticMessage(o), key: `opt-${o.tempId}` })), ] + // 세그먼트 오디오 엘리먼트 ended 리스너 — 마운트 1회. + useEffect(() => { + if (typeof Audio === 'undefined') return + if (!audioElRef.current) audioElRef.current = new Audio() + const el = audioElRef.current + const onEnded = () => queueRef.current?.onEnded() + el.addEventListener('ended', onEnded) + return () => { + el.removeEventListener('ended', onEnded) + if (lastUrlRef.current) URL.revokeObjectURL(lastUrlRef.current) + } + }, []) + // 서버에 반영된 낙관적 답변은 상태에서 제거(세션 내내 누적 방지). useEffect(() => { setOptimistic((prev) => { @@ -44,6 +89,22 @@ export function useLiveInterview(sessionId: number) { }) }, [messagesQuery.data]) + // 서버에서 최종 content가 도착한 메시지의 버퍼를 정리한다. + useEffect(() => { + setDeltaBuffer((prev) => { + const next = { ...prev } + let changed = false + for (const m of serverMessages) { + if (m.id !== undefined && next[m.id] !== undefined && m.content !== FOLLOWUP_GENERATING_TEXT) { + delete next[m.id] + changed = true + } + } + return changed ? next : prev + }) + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [messagesQuery.data]) + const { submitAnswer: socketSubmit } = useInterviewSocket({ sessionId, enabled: status === 'IN_PROGRESS', @@ -56,6 +117,23 @@ export function useLiveInterview(sessionId: number) { void queryClient.invalidateQueries({ queryKey: sessionKeys.detail(sessionId) }) } else if (action.kind === 'redirect-feedback') { navigate(`/sessions/${sessionId}/feedback`) + } else if (action.kind === 'append-delta') { + const payload = (frame.data as { data?: DeltaPayload } | undefined)?.data + if (payload && typeof payload.messageId === 'number') { + setDeltaBuffer((b) => applyDelta(b, payload)) + } + } else if (action.kind === 'queue-audio') { + const p = (frame.data as { data?: { messageId: number; seq: number; ext: string } } | undefined)?.data + if (p && typeof p.messageId === 'number') { + segmentedIds.current.add(p.messageId) + if (currentAudioMsgId.current !== p.messageId) { + currentAudioMsgId.current = p.messageId + queueRef.current?.reset() + } + void fetchMessageSegmentObjectUrl(sessionId, p.messageId, p.seq, p.ext) + .then((url) => queueRef.current?.enqueue({ seq: p.seq, url })) + .catch(() => { /* best-effort */ }) + } } }, }) @@ -83,6 +161,12 @@ export function useLiveInterview(sessionId: number) { [voiceMutation], ) + // 서버 메시지 기준으로 가장 최근 면접관 메시지가 여전히 sentinel이면 스트리밍 진행 중. + const latestServerQuestion = [...serverMessages].reverse().find((m) => m.role === 'INTERVIEWER') + const questionStreaming = latestServerQuestion?.content === FOLLOWUP_GENERATING_TEXT + + const wasSegmented = useCallback((id: number) => segmentedIds.current.has(id), []) + return { session: sessionQuery.data, status, @@ -95,5 +179,7 @@ export function useLiveInterview(sessionId: number) { voiceError: voiceMutation.isError, endSession: () => end.mutate(), isLoading: sessionQuery.isLoading, + questionStreaming, + wasSegmented, } } diff --git a/frontend/src/features/interview/ui/live/AnswerComposer.tsx b/frontend/src/features/interview/ui/live/AnswerComposer.tsx index a93a3856..57294dec 100644 --- a/frontend/src/features/interview/ui/live/AnswerComposer.tsx +++ b/frontend/src/features/interview/ui/live/AnswerComposer.tsx @@ -25,11 +25,13 @@ function MicIcon() { export function AnswerComposer({ disabled = false, + submitLocked = false, onSubmit, onSubmitVoice, voiceUploading = false, }: { disabled?: boolean + submitLocked?: boolean onSubmit: (content: string) => void onSubmitVoice?: (audio: Blob) => void voiceUploading?: boolean @@ -41,7 +43,7 @@ export function AnswerComposer({ const submit = () => { const trimmed = value.trim() - if (!trimmed || disabled) return + if (!trimmed || disabled || submitLocked) return onSubmit(trimmed) setValue('') } @@ -95,21 +97,25 @@ export function AnswerComposer({ maxLength={8000} aria-label="답변 입력" placeholder={ - disabled ? '질문을 기다리는 중…' : '답변을 입력하세요 (Enter 전송, Shift+Enter 줄바꿈)' + disabled + ? '질문을 기다리는 중…' + : submitLocked + ? '질문이 끝나면 전송할 수 있어요' + : '답변을 입력하세요 (Enter 전송, Shift+Enter 줄바꿈)' } /> {voiceSupported && ( )} - diff --git a/frontend/src/features/interview/ui/live/InterviewStage.tsx b/frontend/src/features/interview/ui/live/InterviewStage.tsx index 01da3ddb..5082d4bf 100644 --- a/frontend/src/features/interview/ui/live/InterviewStage.tsx +++ b/frontend/src/features/interview/ui/live/InterviewStage.tsx @@ -48,19 +48,23 @@ export function InterviewStage({ connection, items, awaitingQuestion, + questionStreaming, onSubmit, onSubmitVoice, voiceUploading, onEnd, + wasSegmented, }: { session: Session connection: ConnectionStatus items: ThreadItem[] awaitingQuestion: boolean + questionStreaming: boolean onSubmit: (content: string) => void onSubmitVoice: (audio: Blob) => void voiceUploading: boolean onEnd: () => void + wasSegmented: (id: number) => boolean }) { const [transcriptOpen, setTranscriptOpen] = useState(false) const progress = sessionProgress(session) @@ -115,7 +119,7 @@ export function InterviewStage({ {awaitingQuestion || !currentQuestion ? ( ) : ( - + )} @@ -126,6 +130,7 @@ export function InterviewStage({
) } diff --git a/frontend/src/features/interview/ui/live/QuestionBubble.tsx b/frontend/src/features/interview/ui/live/QuestionBubble.tsx index 3c46a15d..f543659f 100644 --- a/frontend/src/features/interview/ui/live/QuestionBubble.tsx +++ b/frontend/src/features/interview/ui/live/QuestionBubble.tsx @@ -2,6 +2,7 @@ import type { Message } from '@/domain/session' import { StatusBadge } from '@/shared/ui/StatusBadge' import { categoryLabel } from '../../lib/categoryLabel' import { useTtsPlayback } from '../../lib/media/useTtsPlayback' +import { FOLLOWUP_GENERATING_TEXT } from '../../model/streamingBuffer' function PlayIcon({ playing }: { playing: boolean }) { return ( @@ -41,7 +42,15 @@ export function QuestionBubble({
)}
-

{message.content}

+ {message.content === FOLLOWUP_GENERATING_TEXT ? ( + + ● + ● + ● + + ) : ( +

{message.content}

+ )} {ttsReady && (