fix(ai): ai_request_logs 에 세션 귀속 — 비용을 면접 단위로 볼 수 있게

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

@@ -119,7 +119,11 @@ async def consume_with_failure_signal(
             )
             raise
 
-        with trace_context(envelope.trace_id):
+        with trace_context(
+            envelope.trace_id,
+            session_id=getattr(envelope.payload, "session_id", None),
+            user_id=getattr(envelope.payload, "user_id", None),
+        ):
             if idempotency.is_seen_then_mark(envelope.message_id):
                 log.info(
                     f"{domain}.idempotent.skip",

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

@@ -82,7 +82,11 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 )
                 raise
 
-            with trace_context(envelope.trace_id):
+            with trace_context(
+                envelope.trace_id,
+                session_id=getattr(envelope.payload, "session_id", None),
+                user_id=getattr(envelope.payload, "user_id", None),
+            ):
                 if self._idempotency.is_seen_then_mark(envelope.message_id):
                     log.info("tts.idempotent.skip", message_id=envelope.message_id)
                     return

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

@@ -72,7 +72,11 @@ async def handle(self, message: AbstractIncomingMessage) -> None:
                 )
                 raise
 
-            with trace_context(envelope.trace_id):
+            with trace_context(
+                envelope.trace_id,
+                session_id=getattr(envelope.payload, "session_id", None),
+                user_id=getattr(envelope.payload, "user_id", None),
+            ):
                 if self._idempotency.is_seen_then_mark(envelope.message_id):
                     log.info("voice.idempotent.skip", message_id=envelope.message_id)
                     return

ai/src/ai_server/observability/ai_call_log.py

@@ -8,6 +8,7 @@
 import structlog
 
 from ai_server.core.client import CoreClient
+from ai_server.observability.trace import current_session_id, current_user_id
 
 log = structlog.get_logger(__name__)
 
@@ -37,6 +38,11 @@ def record_ai_call(
     if core_client is None:
         return
 
+    # 호출부가 안 넘기면 컨텍스트에서 가져온다 — 체인 빌더 6곳의 시그니처를 오염시키지
+    # 않으면서 "어느 면접의 호출인가" 를 채운다.
+    session_id = session_id if session_id is not None else current_session_id()
+    user_id = user_id if user_id is not None else current_user_id()
+
     async def _do() -> None:
         try:
             await core_client.record_ai_log(

ai/src/ai_server/observability/llm_logging_callback.py

@@ -10,6 +10,7 @@
 from langchain_core.outputs import LLMResult
 
 from ai_server.core.client import CoreClient
+from ai_server.observability.trace import current_session_id, current_user_id
 
 log = structlog.get_logger(__name__)
 
@@ -71,6 +72,12 @@ def _latency_ms(self, run_id: UUID) -> int | None:
         return int((time.perf_counter() - started) * 1000)
 
     async def _fire(self, **kwargs: Any) -> None:
+        # 체인 빌더는 세션을 모른다(팩토리가 요청 밖에서 불릴 수 있다). 기록 시점에
+        # 컨텍스트에서 읽어 채운다 — 이게 없으면 generate.* / feedback.* 로그의
+        # session_id 가 전부 NULL 로 남는다(2026-08-06 이래 실제로 그랬다).
+        kwargs.setdefault("session_id", current_session_id())
+        kwargs.setdefault("user_id", current_user_id())
+
         # fire-and-forget: 본 핸들러 실패가 LLM 응답 흐름을 막지 않도록.
         async def _do() -> None:
             try:

ai/src/ai_server/observability/trace.py

@@ -18,22 +18,64 @@
 from __future__ import annotations
 
 import contextlib
+import contextvars
 import uuid
 from collections.abc import Iterator
 
 import structlog
 
+# 로그 바인딩과 별개로, `ai_request_logs` 기록부가 읽어 가는 값.
+#
+# `record_ai_call` 과 `CoreAiLogCallback` 은 둘 다 session_id 를 **받을 수** 있게 만들어
+# 놨는데 실제로 넘기는 호출부가 stt.transcribe 하나뿐이었다. 그래서 2026-08-06 이래
+# ai_request_logs 의 session_id 는 그 한 종류 말고 **전부 NULL** 이다 — "이 면접이 얼마나
+# 들었나", "어느 세션이 게이트웨이 크레딧을 태웠나" 를 답할 수 없다(2026-09-17 에 실제로
+# 크레딧이 소진돼 운영이 402 로 죽었을 때 정확히 이 질문을 못 했다).
+#
+# 체인 빌더마다 인자로 넘기면 LangChain 콜백 생성 지점 6곳 + 체인 팩토리 시그니처가
+# 전부 오염된다. 관측값은 횡단 관심사이므로 traceId 와 같은 방식으로 컨텍스트에 싣는다.
+_session_id: contextvars.ContextVar[int | None] = contextvars.ContextVar(
+    "ai_log_session_id", default=None
+)
+_user_id: contextvars.ContextVar[int | None] = contextvars.ContextVar(
+    "ai_log_user_id", default=None
+)
+
+
+def current_session_id() -> int | None:
+    """기록부가 명시적 인자 없이 읽어 가는 세션 id."""
+    return _session_id.get()
+
+
+def current_user_id() -> int | None:
+    return _user_id.get()
+
 
 @contextlib.contextmanager
-def trace_context(trace_id: str | None) -> Iterator[str]:
+def trace_context(
+    trace_id: str | None,
+    *,
+    session_id: int | None = None,
+    user_id: int | None = None,
+) -> Iterator[str]:
     """`trace_id` 를 이 구간의 모든 로그에 바인딩한다. 실제 쓰인 값을 돌려준다.
 
+    `session_id`/`user_id` 를 주면 같은 구간의 `ai_request_logs` 기록에도 자동으로 붙는다
+    (로그 줄에도 실린다 — 어느 면접의 호출인지 로그만 봐도 알 수 있게).
+
     비어 있으면 새로 만든다 — 없는 것보다 낫고, 적어도 그 처리 한 건은 묶인다.
     끝나면 반드시 되돌린다: 워커 태스크가 재사용되면 남은 값이 다음 메시지 로그에 묻는다.
     """
     effective = trace_id if trace_id else str(uuid.uuid4())
-    tokens = structlog.contextvars.bind_contextvars(trace_id=effective)
+    bound: dict[str, object] = {"trace_id": effective}
+    if session_id is not None:
+        bound["session_id"] = session_id
+    tokens = structlog.contextvars.bind_contextvars(**bound)
+    sid_token = _session_id.set(session_id)
+    uid_token = _user_id.set(user_id)
     try:
         yield effective
     finally:
+        _user_id.reset(uid_token)
+        _session_id.reset(sid_token)
         structlog.contextvars.reset_contextvars(**tokens)

ai/tests/test_trace_context.py

@@ -86,3 +86,131 @@ async def main() -> None:
 
     seen = {e["which"]: e["trace_id"] for e in cap.entries}
     assert seen == {"A": "A", "B": "B", "C": "C"}
+
+
+# ── ai_request_logs 세션 귀속 ────────────────────────────────────────────────
+#
+# record_ai_call 과 CoreAiLogCallback 은 둘 다 session_id 를 받을 수 있게 만들어 놨는데
+# 실제로 넘기는 호출부가 stt.transcribe 하나뿐이어서, 2026-08-06 이래 ai_request_logs 의
+# session_id 는 그 한 종류 말고 전부 NULL 이었다. "이 면접이 얼마나 들었나", "어느 세션이
+# 게이트웨이 크레딧을 태웠나" 를 답할 수 없다.
+
+
+class _RecordingCore:
+    def __init__(self) -> None:
+        self.calls: list[dict] = []
+
+    async def record_ai_log(self, **kwargs) -> None:
+        self.calls.append(kwargs)
+
+
+def test_record_ai_call_picks_up_session_from_context():
+    from ai_server.observability.ai_call_log import record_ai_call
+
+    core = _RecordingCore()
+
+    async def main() -> None:
+        with trace_context("t", session_id=102, user_id=7):
+            record_ai_call(
+                core,
+                request_type="tts.synthesize",
+                model_name="m",
+                latency_ms=10,
+                status="SUCCESS",
+            )
+        # fire-and-forget 이라 태스크가 돌 틈을 준다
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+    asyncio.run(main())
+
+    assert core.calls, "기록이 아예 안 됐다"
+    assert core.calls[0]["session_id"] == 102
+    assert core.calls[0]["user_id"] == 7
+
+
+def test_explicit_session_wins_over_context():
+    from ai_server.observability.ai_call_log import record_ai_call
+
+    core = _RecordingCore()
+
+    async def main() -> None:
+        with trace_context("t", session_id=102):
+            # 호출부가 명시하면 그게 이긴다(voice_consumer 가 그렇게 쓴다).
+            record_ai_call(
+                core,
+                request_type="stt.transcribe",
+                model_name="m",
+                latency_ms=10,
+                status="SUCCESS",
+                session_id=999,
+            )
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+    asyncio.run(main())
+
+    assert core.calls[0]["session_id"] == 999
+
+
+def test_no_session_context_records_none_not_crash():
+    from ai_server.observability.ai_call_log import record_ai_call
+
+    core = _RecordingCore()
+
+    async def main() -> None:
+        # 문서 분석처럼 세션이 없는 흐름 — 기록은 되고 session_id 만 비어야 한다.
+        with trace_context("t"):
+            record_ai_call(
+                core,
+                request_type="analyze.document",
+                model_name="m",
+                latency_ms=10,
+                status="SUCCESS",
+            )
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+    asyncio.run(main())
+
+    assert core.calls[0]["session_id"] is None
+
+
+def test_langchain_callback_picks_up_session_from_context():
+    from ai_server.observability.llm_logging_callback import CoreAiLogCallback
+
+    core = _RecordingCore()
+    cb = CoreAiLogCallback(
+        core_client=core, request_type="generate.feedback.panel", default_model="m"
+    )
+
+    async def main() -> None:
+        with trace_context("t", session_id=102, user_id=7):
+            await cb._fire(
+                request_type="generate.feedback.panel",
+                model_name="m",
+                input_tokens=1,
+                output_tokens=2,
+                latency_ms=3,
+                status="SUCCESS",
+                error_message=None,
+            )
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+    asyncio.run(main())
+
+    # 체인 빌더는 세션을 모른다 — 기록 시점에 컨텍스트에서 채워야 한다.
+    assert core.calls[0]["session_id"] == 102
+    assert core.calls[0]["user_id"] == 7
+
+
+def test_session_id_also_appears_in_logs():
+    cap = _capture()
+    log = structlog.get_logger("t")
+
+    with trace_context("t", session_id=102):
+        log.info("inside")
+
+    # 로그만 봐도 어느 면접의 호출인지 알 수 있어야 한다.
+    assert cap.entries[0]["session_id"] == 102

docs/observability.md

@@ -4,6 +4,26 @@
 
 ---
 
+## 0. AI 호출 비용 귀속 (`ai_request_logs.session_id`)
+
+**2026-09-28 이전에는 `stt.transcribe` 말고 전부 NULL 이었다.** 2026-08-06 이래 쌓인
+`generate.*`·`feedback.*`·`tts.synthesize`·`embedding.embed` 로그 전부가 세션을 모른 채
+기록됐다. `record_ai_call` 과 `CoreAiLogCallback` 은 둘 다 `session_id` 를 **받을 수 있게**
+만들어져 있었는데 실제로 넘기는 호출부가 voice consumer 하나뿐이었다.
+
+그래서 "이 면접이 얼마나 들었나", "어느 세션이 게이트웨이 크레딧을 태웠나" 를 답할 수 없었다 —
+2026-09-17 에 실제로 크레딧이 소진돼 운영 AI 가 402 로 죽었을 때 정확히 이 질문을 못 했다.
+
+체인 빌더마다 인자로 넘기면 LangChain 콜백 생성 지점 6곳과 체인 팩토리 시그니처가 전부
+오염된다. 관측값은 횡단 관심사이므로 **traceId 와 같은 방식으로 컨텍스트에 싣는다** —
+`trace_context(trace_id, session_id=…, user_id=…)` 가 컨슈머 진입점에서 묶고,
+`record_ai_call` 과 `CoreAiLogCallback` 이 명시 인자가 없을 때 거기서 읽는다.
+명시 인자가 있으면 그쪽이 이긴다.
+
+세션이 없는 흐름(문서 분석)은 그대로 NULL 이다 — 억지로 채우지 않는다.
+
+---
+
 ## 1. 분산 추적 (X-Trace-Id)
 
 ### 1.1 ID 생성