세션 메모리와 비언어 이벤트 저장

apps/api/app/routes/voice.py

@@ -48,6 +48,56 @@ WS_CLOSE_UNAUTHORIZED = 1008
 
 # Per-utterance audio cap to avoid unbounded memory growth.
 _MAX_AUDIO_BYTES = 10 * 1024 * 1024
+_PROVIDER_EVENT_MAX_ITEMS = 12
+_PROVIDER_EVENT_MAX_STRING = 80
+_PROVIDER_EVENT_ALLOWED_KEYS = {
+    "category",
+    "event_type",
+    "type",
+    "kind",
+    "label",
+    "source",
+    "provider",
+    "start_ms",
+    "end_ms",
+    "duration_ms",
+    "confidence",
+    "score",
+    "is_final",
+}
+_PROVIDER_EVENT_TAXONOMY = {
+    "barge_in": ("barge_in", "turn_taking"),
+    "interrupt": ("barge_in", "turn_taking"),
+    "interruption": ("barge_in", "turn_taking"),
+    "overlap": ("barge_in", "turn_taking"),
+    "sigh": ("sigh", "paralinguistic"),
+    "sighing": ("sigh", "paralinguistic"),
+    "sob": ("cry", "paralinguistic"),
+    "cry": ("cry", "paralinguistic"),
+    "crying": ("cry", "paralinguistic"),
+    "weep": ("cry", "paralinguistic"),
+    "laugh": ("laugh", "paralinguistic"),
+    "laughter": ("laugh", "paralinguistic"),
+    "breath": ("breath", "paralinguistic"),
+    "breathing": ("breath", "paralinguistic"),
+    "voice_activity": ("voice_activity", "speech_activity"),
+    "vad": ("voice_activity", "speech_activity"),
+    "speech_start": ("speech_start", "speech_activity"),
+    "speech_end": ("speech_end", "speech_activity"),
+    "speech_final": ("speech_final", "speech_activity"),
+    "silence": ("silence", "timing"),
+    "pause": ("silence", "timing"),
+    "long_pause": ("silence", "timing"),
+    "speech_rate": ("speech_rate", "prosody"),
+    "fast_speech": ("speech_rate", "prosody"),
+    "slow_speech": ("speech_rate", "prosody"),
+    "pitch": ("pitch", "prosody"),
+    "intonation": ("intonation", "prosody"),
+    "prosody": ("prosody", "prosody"),
+    "noise": ("background_noise", "audio_quality"),
+    "background_noise": ("background_noise", "audio_quality"),
+}
+_PROVIDER_EVENT_TYPE_FIELDS = ("event_type", "type", "kind", "label")
 
 
 @router.get("/health")
@@ -187,6 +237,7 @@ async def voice_ws(websocket: WebSocket) -> None:
                     audio_ended_at=audio_ended_at,
                     silence_ms=silence_ms,
                     barge_in=_safe_bool(ctrl.get("barge_in")),
+                    provider_events=_safe_provider_events(ctrl.get("provider_events")),
                 )
                 last_audio_end_at = audio_ended_at
                 audio_started_at = None
@@ -232,6 +283,7 @@ async def _handle_utterance(
     audio_ended_at: float | None = None,
     silence_ms: int | None = None,
     barge_in: bool | None = None,
+    provider_events: list[dict[str, object]] | None = None,
 ) -> None:
     """Transcribe one utterance, generate the client reply, then synthesize TTS."""
     if not audio:
@@ -259,6 +311,7 @@ async def _handle_utterance(
     audio_ref = _voice_audio_ref(audio, fmt)
     duration_s = stt.duration or _elapsed_seconds(audio_started_at, audio_ended_at)
     speech_rate = _estimate_speech_rate(learner_text, duration_s)
+    provider_events = _merge_provider_events(provider_events, getattr(stt, "provider_events", []))
     await _safe_send_json(
         websocket,
         {"type": "transcript", "text": learner_text, "final": True, "speaker": "counselor"},
@@ -277,6 +330,7 @@ async def _handle_utterance(
         silence_ms=silence_ms,
         speech_rate=speech_rate,
         barge_in=barge_in,
+        provider_events=provider_events,
     )
 
 
@@ -291,6 +345,7 @@ async def _run_turn_and_speak(
     silence_ms: int | None = None,
     speech_rate: float | None = None,
     barge_in: bool | None = None,
+    provider_events: list[dict[str, object]] | None = None,
 ) -> None:
     """Run one counseling turn and stream synthesized client speech."""
     sess, err = await _load_voice_session(session_id, principal)
@@ -351,6 +406,7 @@ async def _run_turn_and_speak(
             silence_ms=silence_ms,
             speech_rate=speech_rate,
             barge_in=barge_in,
+            provider_events=provider_events or [],
             evaluation=result.evaluation,
         ),
     )
@@ -646,6 +702,71 @@ def _safe_bool(value: object) -> bool | None:
     return bool(value)
 
 
+def _provider_event_slug(value: object) -> str:
+    text = str(value or "").strip().lower()
+    if not text:
+        return ""
+    chars = []
+    previous_underscore = False
+    for char in text:
+        if char.isalnum():
+            chars.append(char)
+            previous_underscore = False
+        elif not previous_underscore:
+            chars.append("_")
+            previous_underscore = True
+    return "".join(chars).strip("_")[:_PROVIDER_EVENT_MAX_STRING]
+
+
+def _provider_event_taxonomy(event: dict[str, object]) -> tuple[str, str]:
+    for field in _PROVIDER_EVENT_TYPE_FIELDS:
+        slug = _provider_event_slug(event.get(field))
+        if not slug:
+            continue
+        canonical = _PROVIDER_EVENT_TAXONOMY.get(slug)
+        if canonical:
+            return canonical
+        return slug, "unknown"
+    return "", ""
+
+
+def _safe_provider_events(value: object) -> list[dict[str, object]]:
+    if not isinstance(value, list):
+        return []
+    events: list[dict[str, object]] = []
+    for item in value[:_PROVIDER_EVENT_MAX_ITEMS]:
+        if not isinstance(item, dict):
+            continue
+        safe: dict[str, object] = {}
+        for key in _PROVIDER_EVENT_ALLOWED_KEYS:
+            raw = item.get(key)
+            if isinstance(raw, bool):
+                safe[key] = raw
+            elif isinstance(raw, (int, float)):
+                safe[key] = raw
+            elif isinstance(raw, str):
+                text = raw.strip()
+                if text:
+                    safe[key] = text[:_PROVIDER_EVENT_MAX_STRING]
+        if safe:
+            event_type, category = _provider_event_taxonomy(safe)
+            if event_type:
+                safe["event_type"] = event_type
+            if category:
+                safe["category"] = category
+            events.append(safe)
+    return events
+
+
+def _merge_provider_events(*values: object) -> list[dict[str, object]]:
+    merged: list[dict[str, object]] = []
+    for value in values:
+        merged.extend(_safe_provider_events(value))
+        if len(merged) >= _PROVIDER_EVENT_MAX_ITEMS:
+            return merged[:_PROVIDER_EVENT_MAX_ITEMS]
+    return merged
+
+
 async def _safe_send_json(websocket: WebSocket, payload: dict) -> None:
     if websocket.client_state != WebSocketState.CONNECTED:
         return

apps/api/app/services/memory.py

@@ -18,12 +18,26 @@ DB 미가용(Docker off) 시에도 동작하도록 입력은 plain dict/list 로
 
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass, field
 from typing import Any, Callable, Optional
 
 from .state_machine import SessionState
 
 
+_SESSION_DIGEST_EXCERPT_CHARS = 90
+_CASE_DIGEST_MAX_ENTRIES = 12
+_RAPPORT_TRAJECTORY_MAX_ENTRIES = 24
+_PINNED_FACT_MAX_VALUE_CHARS = 160
+_COUNSELING_AGREEMENT_RE = re.compile(r"(상담|회기).*(주\s*\d+\s*회|매주|약속|계속|이어)|"
+                                      r"(주\s*\d+\s*회|매주).*(상담|회기)")
+_COUNSELING_AGREEMENT_WITHDRAWAL_RE = re.compile(
+    r"(상담|회기).*(그만|중단|취소|철회|안\s*하|하지\s*않|못\s*하|이어\s*가지\s*않|계속\s*하지\s*않)|"
+    r"(약속).*(취소|철회|못\s*지키|지키지\s*않|안\s*지키)|"
+    r"더\s*이상.*(상담|회기).*(안\s*하|하지\s*않|못\s*하)"
+)
+
+
 # ════════════════════════════════════════════════════════════════════════════
 # 회기 시작 — 회상 (큰그림 → 세부)
 # ════════════════════════════════════════════════════════════════════════════
@@ -114,6 +128,18 @@ class CompressionJob:
     open_threads: list[str] = field(default_factory=list)
 
 
+@dataclass(frozen=True, slots=True)
+class PinnedFactCandidate:
+    """Rule-derived client-visible fact candidate for app.pinned_fact."""
+
+    key: str
+    value: str
+    fact_type: str
+    status: str = "stable"
+    confidence: float = 0.7
+    source_turn_id: str | None = None
+
+
 def make_carry_over(
     *,
     state: SessionState,
@@ -170,6 +196,174 @@ def build_compression_messages(job: CompressionJob) -> list[dict[str, str]]:
     return [{"role": "system", "content": system}, {"role": "user", "content": user}]
 
 
+def _compact(value: Any, *, limit: int = _SESSION_DIGEST_EXCERPT_CHARS) -> str:
+    text = " ".join(str(value or "").split())
+    if len(text) <= limit:
+        return text
+    return text[: max(0, limit - 3)].rstrip() + "..."
+
+
+def build_fallback_session_digest(
+    *,
+    session_no: int,
+    masked_turns: list[dict[str, str]],
+    end_state: dict,
+) -> str:
+    """마스킹 축어록 기반 임시 회기 digest.
+
+    LLM 압축/embedding writer가 붙기 전에도 다음 회기 recall이 빈 문자열로 남지 않도록
+    client-visible 마스킹 발화와 결정론 상태 수치만 사용한다.
+    """
+    if not masked_turns:
+        return f"S{session_no}: 실제 발화가 없어 요약을 생성하지 않았다."
+
+    counselor_count = sum(1 for turn in masked_turns if turn.get("speaker") == "counselor")
+    client_turns = [turn for turn in masked_turns if turn.get("speaker") == "client"]
+    client_count = len(client_turns)
+    last_client = _compact(client_turns[-1].get("text") if client_turns else "")
+    stage = str(end_state.get("stage") or "미확인")
+    openness = end_state.get("effective_openness")
+    rapport = end_state.get("rapport_credit")
+    status_bits = [f"종료 단계 {stage}"]
+    if openness is not None:
+        status_bits.append(f"개방도 {openness}")
+    if rapport is not None:
+        status_bits.append(f"라포 {rapport}")
+    status = ", ".join(status_bits)
+    if last_client:
+        return (
+            f"S{session_no}: 마스킹 축어록 기준 상담자 {counselor_count}회, "
+            f"내담자 {client_count}회 발화. 마지막 내담자 반응은 \"{last_client}\". {status}."
+        )
+    return (
+        f"S{session_no}: 마스킹 축어록 기준 상담자 {counselor_count}회, "
+        f"내담자 {client_count}회 발화. {status}."
+    )
+
+
+def _fact_value(text: Any) -> str:
+    return _compact(text, limit=_PINNED_FACT_MAX_VALUE_CHARS)
+
+
+def extract_pinned_fact_candidates(
+    masked_turns: list[dict[str, Any]],
+) -> list[PinnedFactCandidate]:
+    """Extract conservative pinned facts from masked client-visible text.
+
+    The first pass intentionally avoids clinical inference. It only preserves
+    explicit facts already surfaced by the client AI and already masked for
+    learner visibility.
+    """
+    by_key: dict[str, PinnedFactCandidate] = {}
+    for turn in masked_turns:
+        if turn.get("speaker") != "client":
+            continue
+        text = _fact_value(turn.get("text"))
+        if not text:
+            continue
+        source_turn_id = turn.get("turn_id")
+        if "[NAME]" in text:
+            by_key["identity:name"] = PinnedFactCandidate(
+                key="identity:name",
+                value="[NAME]",
+                fact_type="identity",
+                confidence=0.85,
+                source_turn_id=str(source_turn_id) if source_turn_id else None,
+            )
+        if "[ORG]" in text:
+            by_key["identity:org"] = PinnedFactCandidate(
+                key="identity:org",
+                value="[ORG]",
+                fact_type="identity",
+                confidence=0.85,
+                source_turn_id=str(source_turn_id) if source_turn_id else None,
+            )
+        if _COUNSELING_AGREEMENT_WITHDRAWAL_RE.search(text):
+            by_key["agreement:counseling"] = PinnedFactCandidate(
+                key="agreement:counseling",
+                value=text,
+                fact_type="agreement",
+                status="contradicted",
+                confidence=0.8,
+                source_turn_id=str(source_turn_id) if source_turn_id else None,
+            )
+        elif _COUNSELING_AGREEMENT_RE.search(text):
+            by_key["agreement:counseling"] = PinnedFactCandidate(
+                key="agreement:counseling",
+                value=text,
+                fact_type="agreement",
+                confidence=0.75,
+                source_turn_id=str(source_turn_id) if source_turn_id else None,
+            )
+    return list(by_key.values())
+
+
+def merge_case_digest(
+    *,
+    existing_digest: str | None,
+    session_no: int,
+    session_digest: str,
+    max_entries: int = _CASE_DIGEST_MAX_ENTRIES,
+) -> str:
+    """case_profile.case_digest를 session_no 기준으로 idempotent append한다."""
+    prefix = f"S{session_no}:"
+    lines = [
+        line.strip()
+        for line in str(existing_digest or "").splitlines()
+        if line.strip() and not line.strip().startswith(prefix)
+    ]
+    next_line = session_digest.strip()
+    if next_line and not next_line.startswith(prefix):
+        next_line = f"{prefix} {next_line}"
+    if next_line:
+        lines.append(next_line)
+    return "\n".join(lines[-max_entries:])
+
+
+def rapport_trajectory_point(*, session_no: int, end_state: dict) -> dict[str, Any]:
+    """case_profile.rapport_trajectory에 저장할 최소 무손실 수치 포인트."""
+    return {
+        "session_no": int(session_no),
+        "stage": end_state.get("stage"),
+        "end_rapport": end_state.get("rapport_credit"),
+        "end_openness": end_state.get("effective_openness"),
+        "resistance": end_state.get("resistance"),
+    }
+
+
+def merge_rapport_trajectory(
+    existing: Any,
+    point: dict[str, Any],
+    *,
+    max_entries: int = _RAPPORT_TRAJECTORY_MAX_ENTRIES,
+) -> list[dict[str, Any]]:
+    """session_no 기준으로 trajectory를 덮어쓰기 가능하게 append한다."""
+    session_no = point.get("session_no")
+    merged: list[dict[str, Any]] = []
+    if isinstance(existing, list):
+        for item in existing:
+            if not isinstance(item, dict):
+                continue
+            if item.get("session_no") == session_no:
+                continue
+            merged.append(dict(item))
+    merged.append(dict(point))
+    return merged[-max_entries:]
+
+
+def update_alliance_level(previous: Any, end_rapport: Any) -> float:
+    """case_profile.alliance_level EWMA. 이전 값이 없으면 schema default 0.2 기준."""
+    try:
+        prev = float(previous)
+    except (TypeError, ValueError):
+        prev = 0.2
+    try:
+        rapport = float(end_rapport)
+    except (TypeError, ValueError):
+        rapport = prev
+    return round(max(0.0, min(1.0, prev * 0.7 + rapport * 0.3)), 4)
+
+
 __all__ = [
     "RecallContext",
     "build_recall_context",
@@ -177,4 +371,11 @@ __all__ = [
     "CompressionJob",
     "make_carry_over",
     "build_compression_messages",
+    "build_fallback_session_digest",
+    "PinnedFactCandidate",
+    "extract_pinned_fact_candidates",
+    "merge_case_digest",
+    "rapport_trajectory_point",
+    "merge_rapport_trajectory",
+    "update_alliance_level",
 ]

apps/api/app/services/voice.py

@@ -19,7 +19,7 @@ PRESET_TO_OPENAI_VOICE 테이블이 흡수. 새 preset 추가는 이 테이블
 from __future__ import annotations
 
 import re
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, AsyncIterator, Mapping, Optional
 
@@ -137,6 +137,7 @@ class TranscriptResult:
     language: Optional[str] = None
     model: str = STT_MODEL
     duration: Optional[float] = None
+    provider_events: list[dict[str, object]] = field(default_factory=list)
 
 
 @dataclass(frozen=True, slots=True)

apps/api/app/session_persistence.py

@@ -551,6 +551,7 @@ def _turn_from_row(row, evaluation: dict[str, Any] | None = None) -> TurnRecord:
         silence_ms=_row_value(row, "silence_ms"),
         speech_rate=_row_value(row, "speech_rate"),
         barge_in=_row_value(row, "barge_in"),
+        provider_events=_dict_items(_row_value(row, "provider_events")),
         evaluation=evaluation,
         visible_to=tuple(_row_value(row, "visible_to") or DEFAULT_TURN_VISIBLE_TO),
     )
@@ -1890,7 +1891,7 @@ async def load_session(
                 """
                 SELECT id, seq, speaker, stage, text, text_masked, created_at,
                        llm_provider, model, tokens_in, tokens_out, cost_usd,
-                       audio_ref, silence_ms, speech_rate, barge_in, visible_to
+                       audio_ref, silence_ms, speech_rate, barge_in, provider_events, visible_to
                 FROM app.turns
                 WHERE session_id = $1::uuid
                 ORDER BY seq
@@ -1945,12 +1946,12 @@ async def append_turn(
                 INSERT INTO app.turns (
                   session_id, seq, speaker, stage, text, text_masked, actor_kind,
                   llm_provider, model, tokens_in, tokens_out, cost_usd,
-                  audio_ref, silence_ms, speech_rate, barge_in, visible_to
+                  audio_ref, silence_ms, speech_rate, barge_in, provider_events, visible_to
                 )
                 VALUES (
                   $1::uuid, $2, $3, $4, $5, $6, $7,
                   $8, $9, $10, $11, $12,
-                  $13, $14, $15, $16, $17::text[]
+                  $13, $14, $15, $16, $17::jsonb, $18::text[]
                 )
                 ON CONFLICT (session_id, seq) DO NOTHING
                 RETURNING id
@@ -1971,6 +1972,7 @@ async def append_turn(
                 turn.silence_ms,
                 turn.speech_rate,
                 turn.barge_in,
+                turn.provider_events or [],
                 list(turn.visible_to or DEFAULT_TURN_VISIBLE_TO),
             )
             if inserted_turn_id is None:
@@ -2000,13 +2002,176 @@ async def update_state(
         return False
 
 
+async def _insert_pinned_fact_history(
+    conn: Any,
+    *,
+    fact_id: Any,
+    case_id: str,
+    old_value: Any,
+    new_value: Any,
+    reason: str,
+    session_no: int,
+    turn_id: str | None,
+) -> None:
+    await conn.execute(
+        """
+        INSERT INTO app.pinned_fact_history (
+          fact_id, case_id, old_value, new_value, reason, session_no, turn_id
+        )
+        VALUES ($1::uuid, $2::uuid, $3, $4, $5, $6, $7::uuid)
+        """,
+        fact_id,
+        case_id,
+        old_value,
+        new_value,
+        reason,
+        session_no,
+        turn_id,
+    )
+
+
+async def _upsert_pinned_fact_candidates(conn: Any, sess: InProcSession) -> None:
+    turn_rows = [
+        {
+            "speaker": turn.speaker,
+            "text": turn.text_masked,
+            "turn_id": turn.turn_id,
+        }
+        for turn in sess.turns_visible_to("client")
+    ]
+    candidates = memory.extract_pinned_fact_candidates(turn_rows)
+    for fact in candidates:
+        if fact.status == "contradicted":
+            row = await conn.fetchrow(
+                """
+                WITH existing AS (
+                  SELECT id, value
+                  FROM app.pinned_fact
+                  WHERE case_id = $1::uuid
+                    AND key = $2
+                    AND status <> 'locked'
+                  FOR UPDATE
+                ),
+                updated AS (
+                  UPDATE app.pinned_fact
+                  SET value = $3,
+                      fact_type = $4,
+                      status = 'contradicted',
+                      source_turn = COALESCE($5::uuid, app.pinned_fact.source_turn),
+                      confidence = GREATEST(app.pinned_fact.confidence, $6),
+                      version = app.pinned_fact.version + 1,
+                      updated_session_no = $7,
+                      visible_to = $8::text[],
+                      updated_at = now()
+                  FROM existing
+                  WHERE app.pinned_fact.id = existing.id
+                  RETURNING
+                    app.pinned_fact.id,
+                    existing.value AS old_value,
+                    app.pinned_fact.value AS new_value
+                )
+                SELECT id, old_value, new_value FROM updated
+                """,
+                sess.case_id,
+                fact.key,
+                fact.value,
+                fact.fact_type,
+                fact.source_turn_id,
+                fact.confidence,
+                sess.session_no,
+                ["evaluator"],
+            )
+            if not row:
+                continue
+            old_value = row["old_value"]
+            new_value = row["new_value"]
+            if old_value == new_value:
+                continue
+            await _insert_pinned_fact_history(
+                conn,
+                fact_id=row["id"],
+                case_id=sess.case_id,
+                old_value=old_value,
+                new_value=new_value,
+                reason="contradiction",
+                session_no=sess.session_no,
+                turn_id=fact.source_turn_id,
+            )
+            continue
+        row = await conn.fetchrow(
+            """
+            WITH existing AS (
+              SELECT id, value, status
+              FROM app.pinned_fact
+              WHERE case_id = $1::uuid AND key = $2
+              FOR UPDATE
+            ),
+            upserted AS (
+            INSERT INTO app.pinned_fact (
+              case_id, key, value, fact_type, status, source_turn,
+              confidence, updated_session_no, visible_to, updated_at
+            )
+            VALUES (
+              $1::uuid, $2, $3, $4, $5, $6::uuid,
+              $7, $8, $9::text[], now()
+            )
+            ON CONFLICT (case_id, key) DO UPDATE SET
+              value = EXCLUDED.value,
+              fact_type = EXCLUDED.fact_type,
+              status = EXCLUDED.status,
+              source_turn = COALESCE(EXCLUDED.source_turn, app.pinned_fact.source_turn),
+              confidence = GREATEST(app.pinned_fact.confidence, EXCLUDED.confidence),
+              version = CASE
+                WHEN app.pinned_fact.value IS DISTINCT FROM EXCLUDED.value
+                THEN app.pinned_fact.version + 1
+                ELSE app.pinned_fact.version
+              END,
+              updated_session_no = EXCLUDED.updated_session_no,
+              visible_to = EXCLUDED.visible_to,
+              updated_at = now()
+            WHERE app.pinned_fact.status <> 'locked'
+            RETURNING
+              app.pinned_fact.id,
+              (SELECT value FROM existing) AS old_value,
+              app.pinned_fact.value AS new_value
+            )
+            SELECT id, old_value, new_value FROM upserted
+            """,
+            sess.case_id,
+            fact.key,
+            fact.value,
+            fact.fact_type,
+            fact.status,
+            fact.source_turn_id,
+            fact.confidence,
+            sess.session_no,
+            ["client", "evaluator"],
+        )
+        if not row:
+            continue
+        old_value = row["old_value"]
+        new_value = row["new_value"]
+        if old_value is not None and old_value == new_value:
+            continue
+        reason = "progression" if old_value is None else "clarification"
+        await _insert_pinned_fact_history(
+            conn,
+            fact_id=row["id"],
+            case_id=sess.case_id,
+            old_value=old_value,
+            new_value=new_value,
+            reason=reason,
+            session_no=sess.session_no,
+            turn_id=fact.source_turn_id,
+        )
+
 async def end_session(sess: InProcSession, carry: memory.CarryOver) -> bool:
     try:
         get_pool()
-        digest = (
-            f"회기 축어록 {len(sess.turns)}개가 저장되었습니다. 정밀 리뷰는 생성 대기 중입니다."
-            if sess.turns
-            else "실제 발화가 없어 요약을 생성하지 않았습니다."
+        digest = memory.build_fallback_session_digest(
+            session_no=sess.session_no,
+            masked_turns=sess.masked_turns(visible_to="client"),
+            end_state=carry.end_state,
         )
         async with acquire(role="learner", user_id=sess.learner_id) as conn:
             await conn.execute(
@@ -2039,6 +2204,51 @@ async def end_session(sess: InProcSession, carry: memory.CarryOver) -> bool:
                 digest,
                 list(carry.compression_job.open_threads if carry.compression_job else []),
             )
+            case_row = await conn.fetchrow(
+                """
+                SELECT case_digest, rapport_trajectory, alliance_level
+                FROM app.case_profile
+                WHERE case_id = $1::uuid
+                  AND learner_id = $2::uuid
+                """,
+                sess.case_id,
+                sess.learner_id,
+            )
+            if case_row is not None:
+                trajectory_point = memory.rapport_trajectory_point(
+                    session_no=sess.session_no,
+                    end_state=carry.end_state,
+                )
+                case_digest = memory.merge_case_digest(
+                    existing_digest=case_row["case_digest"],
+                    session_no=sess.session_no,
+                    session_digest=digest,
+                )
+                rapport_trajectory = memory.merge_rapport_trajectory(
+                    case_row["rapport_trajectory"],
+                    trajectory_point,
+                )
+                alliance_level = memory.update_alliance_level(
+                    case_row["alliance_level"],
+                    trajectory_point.get("end_rapport"),
+                )
+                await conn.execute(
+                    """
+                    UPDATE app.case_profile
+                    SET case_digest = $3,
+                        rapport_trajectory = $4::jsonb,
+                        alliance_level = $5,
+                        updated_at = now()
+                    WHERE case_id = $1::uuid
+                      AND learner_id = $2::uuid
+                    """,
+                    sess.case_id,
+                    sess.learner_id,
+                    case_digest,
+                    rapport_trajectory,
+                    alliance_level,
+                )
+                await _upsert_pinned_fact_candidates(conn, sess)
         return True
     except Exception:
         require_runtime_fallback_allowed("session end")
@@ -2110,7 +2320,7 @@ async def list_sessions(
                     """
                     SELECT id, seq, speaker, stage, text, text_masked, created_at,
                            llm_provider, model, tokens_in, tokens_out, cost_usd,
-                           audio_ref, silence_ms, speech_rate, barge_in, visible_to
+                           audio_ref, silence_ms, speech_rate, barge_in, provider_events, visible_to
                     FROM app.turns
                     WHERE session_id = $1::uuid
                     ORDER BY seq

apps/api/app/store.py

@@ -42,6 +42,7 @@ class TurnRecord:
     silence_ms: int | None = None
     speech_rate: float | None = None
     barge_in: bool | None = None
+    provider_events: list[dict[str, object]] = field(default_factory=list)
     # fast-loop 턴 평가(TurnEvaluation.to_hook_dict). 학습자(상담자) 발화에 부착.
     evaluation: Optional[dict] = None
     visible_to: tuple[str, ...] = DEFAULT_TURN_VISIBLE_TO

apps/api/app/test_session_memory.py

@@ -0,0 +1,573 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from . import session_persistence
+from .routes import sessions
+from .services import memory, rag, state_machine
+from .services.persona import P1
+from .store import InProcSession, TurnRecord
+
+
+class SessionMemoryPureTest(unittest.TestCase):
+    def test_case_digest_merge_is_idempotent_by_session_no(self) -> None:
+        digest = memory.merge_case_digest(
+            existing_digest="S1: 이전 회기 요약\nS2: 오래된 요약",
+            session_no=2,
+            session_digest="S2: 새 요약",
+        )
+
+        self.assertEqual(digest, "S1: 이전 회기 요약\nS2: 새 요약")
+
+    def test_rapport_trajectory_merge_replaces_same_session(self) -> None:
+        merged = memory.merge_rapport_trajectory(
+            [{"session_no": 1, "end_rapport": 0.2}, {"session_no": 2, "end_rapport": 0.3}],
+            {"session_no": 2, "end_rapport": 0.7, "end_openness": 0.5},
+        )
+
+        self.assertEqual(len(merged), 2)
+        self.assertEqual(merged[-1]["session_no"], 2)
+        self.assertEqual(merged[-1]["end_rapport"], 0.7)
+
+    def test_fallback_session_digest_uses_masked_client_visible_text(self) -> None:
+        digest = memory.build_fallback_session_digest(
+            session_no=3,
+            masked_turns=[
+                {"speaker": "counselor", "text": "그때 마음이 어땠나요?"},
+                {"speaker": "client", "text": "저는 [NAME]이고 [ORG]에 다녀요."},
+            ],
+            end_state={"stage": "탐색", "effective_openness": 0.42, "rapport_credit": 0.31},
+        )
+
+        self.assertIn("S3:", digest)
+        self.assertIn("[NAME]", digest)
+        self.assertIn("[ORG]", digest)
+        self.assertNotIn("김서연", digest)
+
+    def test_extract_pinned_fact_candidates_is_conservative_and_masked(self) -> None:
+        facts = memory.extract_pinned_fact_candidates(
+            [
+                {"speaker": "counselor", "text": "이름을 말해줄 수 있나요?"},
+                {
+                    "speaker": "client",
+                    "text": "저는 [NAME]이고 [ORG]에 다녀요. 동생과 자주 싸워요. 주 1회 상담 약속은 지키고 싶어요.",
+                    "turn_id": "00000000-0000-0000-0000-000000000201",
+                },
+            ]
+        )
+
+        by_key = {fact.key: fact for fact in facts}
+        self.assertEqual(by_key["identity:name"].value, "[NAME]")
+        self.assertEqual(by_key["identity:org"].value, "[ORG]")
+        self.assertEqual(by_key["agreement:counseling"].fact_type, "agreement")
+        self.assertNotIn("relationship:sibling", by_key)
+        self.assertNotIn("김서연", " ".join(fact.value for fact in facts))
+
+    def test_extract_pinned_fact_candidates_skips_inferred_or_transient_content(self) -> None:
+        facts = memory.extract_pinned_fact_candidates(
+            [
+                {"speaker": "client", "text": "오늘은 그냥 기분이 좀 나빴어요."},
+                {"speaker": "client", "text": "동생과 자주 싸워요."},
+                {"speaker": "client", "text": "죽고 싶다는 생각이 스쳐갔어요."},
+            ]
+        )
+
+        self.assertEqual(facts, [])
+
+    def test_extract_pinned_fact_candidates_marks_explicit_agreement_withdrawal_only(self) -> None:
+        facts = memory.extract_pinned_fact_candidates(
+            [
+                {
+                    "speaker": "client",
+                    "text": "상담 약속은 이제 못 지키겠어요. 회기는 그만하고 싶어요.",
+                    "turn_id": "00000000-0000-0000-0000-000000000203",
+                },
+                {"speaker": "client", "text": "동생과 자주 싸워요."},
+            ]
+        )
+
+        by_key = {fact.key: fact for fact in facts}
+        self.assertEqual(list(by_key), ["agreement:counseling"])
+        self.assertEqual(by_key["agreement:counseling"].status, "contradicted")
+        self.assertEqual(by_key["agreement:counseling"].fact_type, "agreement")
+        self.assertIn("못 지키겠어요", by_key["agreement:counseling"].value)
+        self.assertEqual(
+            by_key["agreement:counseling"].source_turn_id,
+            "00000000-0000-0000-0000-000000000203",
+        )
+
+    def test_episodic_turn_inputs_use_masked_client_visible_client_turns_only(self) -> None:
+        inputs = rag.episodic_turn_inputs_from_records(
+            session_id="00000000-0000-0000-0000-00000000feed",
+            case_id="00000000-0000-0000-0000-00000000ca5e",
+            turns=[
+                TurnRecord(
+                    turn_seq=1,
+                    speaker="client",
+                    stage="라포",
+                    text="저는 김서연이고 한신대학교에 다녀요.",
+                    text_masked="저는 [NAME]이고 [ORG]에 다녀요.",
+                    turn_id="00000000-0000-0000-0000-000000000201",
+                ),
+                TurnRecord(
+                    turn_seq=2,
+                    speaker="counselor",
+                    stage="라포",
+                    text="상담자 발화",
+                    text_masked="상담자 발화",
+                    turn_id="00000000-0000-0000-0000-000000000202",
+                ),
+                TurnRecord(
+                    turn_seq=3,
+                    speaker="client",
+                    stage="라포",
+                    text="평가자만 볼 발화",
+                    text_masked="평가자만 볼 발화",
+                    turn_id="00000000-0000-0000-0000-000000000203",
+                    visible_to=("evaluator",),
+                ),
+                TurnRecord(
+                    turn_seq=4,
+                    speaker="client",
+                    stage="라포",
+                    text="DB turn id 없음",
+                    text_masked="DB turn id 없음",
+                ),
+            ],
+        )
+
+        self.assertEqual(len(inputs), 1)
+        self.assertEqual(inputs[0].turn_id, "00000000-0000-0000-0000-000000000201")
+        self.assertEqual(inputs[0].seq, 1)
+        self.assertEqual(inputs[0].text_masked, "저는 [NAME]이고 [ORG]에 다녀요.")
+        self.assertNotIn("김서연", inputs[0].text_masked)
+        self.assertNotIn("한신대학교", inputs[0].text_masked)
+
+
+class SessionMemoryPersistenceTest(unittest.IsolatedAsyncioTestCase):
+    async def test_write_persona_turn_embeddings_is_masked_and_idempotent(self) -> None:
+        class FakeConn:
+            def __init__(self) -> None:
+                self.executed: list[tuple[str, tuple[object, ...]]] = []
+
+            async def execute(self, query: str, *args: object) -> str:
+                self.executed.append((query, args))
+                return "INSERT 0 1"
+
+        conn = FakeConn()
+        turn = rag.EpisodicTurnInput(
+            turn_id="00000000-0000-0000-0000-000000000201",
+            case_id="00000000-0000-0000-0000-00000000ca5e",
+            session_id="00000000-0000-0000-0000-00000000feed",
+            seq=7,
+            text_masked="저는 [NAME]이고 [ORG]에 다녀요.",
+        )
+        captured_texts: list[str] = []
+
+        def fake_embed_query(text: str) -> rag.EmbeddedQuery:
+            captured_texts.append(text)
+            return rag.EmbeddedQuery(dense=[0.1] * rag.EMBED_DIM, sparse={"42": 0.7})
+
+        with patch.object(rag, "embed_query", fake_embed_query):
+            result = await rag.write_persona_turn_embeddings(conn, turns=[turn])
+
+        self.assertEqual(result.inserted, 1)
+        self.assertEqual(captured_texts, ["저는 [NAME]이고 [ORG]에 다녀요."])
+        self.assertEqual(len(conn.executed), 1)
+        query, args = conn.executed[0]
+        self.assertIn("INSERT INTO app.turn_embedding", query)
+        self.assertIn("ON CONFLICT (turn_id) DO NOTHING", query)
+        self.assertEqual(args[0], turn.turn_id)
+        self.assertEqual(args[1], turn.case_id)
+        self.assertEqual(args[2], turn.session_id)
+        self.assertEqual(args[3], 7)
+        self.assertIn("[0.1,0.1", str(args[4]))
+        self.assertEqual(args[5], '{"42": 0.7}')
+        self.assertNotIn("김서연", " ".join(str(arg) for arg in args))
+
+    async def test_end_persisted_session_schedules_episodic_embedding_writer(self) -> None:
+        scheduled: list[object] = []
+
+        def fake_create_task(coro):
+            scheduled.append(coro)
+            coro.close()
+            return None
+
+        sess = InProcSession(
+            session_id="00000000-0000-0000-0000-00000000feed",
+            case_id="00000000-0000-0000-0000-00000000ca5e",
+            learner_id="00000000-0000-0000-0000-000000000101",
+            persona_code=P1.code,
+            theory_mode="humanistic",
+            persona=P1,
+            state=state_machine.SessionState(),
+            session_no=1,
+        )
+        carry = memory.CarryOver(
+            end_state={},
+            rapport_delta=0.0,
+            compression_job=None,
+        )
+
+        with patch.object(session_persistence, "end_session", return_value=True), patch.object(
+            sessions.asyncio,
+            "create_task",
+            fake_create_task,
+        ):
+            await sessions._end_persisted_session(sess, carry)
+
+        self.assertTrue(sess.ended)
+        self.assertEqual(len(scheduled), 1)
+
+    async def test_seed_recall_loads_case_digest_and_client_visible_pinned_facts(self) -> None:
+        test_case = self
+
+        class FakeConn:
+            def __init__(self) -> None:
+                self.fetch_queries: list[tuple[str, tuple[object, ...]]] = []
+
+            async def fetchrow(self, query: str, *args: object):
+                self.fetch_queries.append((query, args))
+                if "FROM app.case_profile" in query:
+                    return {"case_digest": "S1: 케이스 큰그림"}
+                if "FROM app.session_summary" in query:
+                    return {
+                        "digest": "직전 회기 요약",
+                        "open_threads": ["가족 이야기를 이어가기"],
+                        "end_state": {"rapport_credit": 0.5},
+                    }
+                return None
+
+            async def fetch(self, query: str, *args: object):
+                self.fetch_queries.append((query, args))
+                test_case.assertIn("FROM app.pinned_fact", query)
+                test_case.assertIn("status IN ('stable', 'evolving', 'locked')", query)
+                test_case.assertIn("$2 = ANY(visible_to)", query)
+                return [{"value": "동생과의 갈등"}, {"value": "주 1회 상담 약속"}]
+
+        class FakeAcquire:
+            def __init__(self, conn: FakeConn) -> None:
+                self.conn = conn
+
+            async def __aenter__(self) -> FakeConn:
+                return self.conn
+
+            async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
+                return None
+
+        conn = FakeConn()
+        with patch.object(sessions.db, "get_pool", return_value=object()), patch.object(
+            sessions.db,
+            "acquire",
+            return_value=FakeAcquire(conn),
+        ):
+            recall = await sessions._build_seed_recall(
+                case_id="00000000-0000-0000-0000-00000000ca5e"
+            )
+
+        self.assertIn("[케이스 큰그림]", recall.recall_summary or "")
+        self.assertIn("S1: 케이스 큰그림", recall.recall_summary or "")
+        self.assertIn("직전 회기 요약", recall.recall_summary or "")
+        self.assertEqual(recall.pinned_facts, ["동생과의 갈등", "주 1회 상담 약속"])
+        self.assertEqual(recall.carry, {"rapport_credit": 0.5})
+
+    async def test_end_session_updates_session_summary_and_case_profile(self) -> None:
+        test_case = self
+
+        class FakeConn:
+            def __init__(self) -> None:
+                self.executed: list[tuple[str, tuple[object, ...]]] = []
+
+            async def execute(self, query: str, *args: object) -> str:
+                self.executed.append((query, args))
+                return "OK"
+
+            async def fetchrow(self, query: str, *args: object):
+                self.executed.append((query, args))
+                if "INSERT INTO app.pinned_fact" in query:
+                    return {
+                        "id": f"00000000-0000-0000-0000-00000000fa{len(self.executed):02d}",
+                        "old_value": None,
+                        "new_value": args[2],
+                    }
+                test_case.assertIn("FROM app.case_profile", query)
+                return {
+                    "case_digest": "S1: 이전 회기",
+                    "rapport_trajectory": [{"session_no": 1, "end_rapport": 0.2}],
+                    "alliance_level": 0.2,
+                }
+
+        class FakeAcquire:
+            def __init__(self, conn: FakeConn) -> None:
+                self.conn = conn
+
+            async def __aenter__(self) -> FakeConn:
+                return self.conn
+
+            async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
+                return None
+
+        learner_id = "00000000-0000-0000-0000-000000000101"
+        case_id = "00000000-0000-0000-0000-00000000ca5e"
+        state = state_machine.SessionState(
+            stage=state_machine.Stage.EXPLORE,
+            turn_seq=2,
+            effective_openness=0.42,
+            rapport_credit=0.31,
+            resistance=0.5,
+        )
+        sess = InProcSession(
+            session_id="00000000-0000-0000-0000-00000000feed",
+            case_id=case_id,
+            learner_id=learner_id,
+            persona_code=P1.code,
+            theory_mode="humanistic",
+            persona=P1,
+            state=state,
+            session_no=2,
+            prev_rapport_credit=0.1,
+            turns=[
+                TurnRecord(
+                    turn_seq=1,
+                    speaker="counselor",
+                    stage="라포",
+                    text="실명 질문",
+                    text_masked="이름을 말해줄 수 있나요?",
+                ),
+                TurnRecord(
+                    turn_seq=1,
+                    speaker="client",
+                    stage="라포",
+                    text="저는 김서연이고 한신대학교에 다녀요. 동생과 자주 싸워요. 주 1회 상담 약속은 지키고 싶어요.",
+                    text_masked="저는 [NAME]이고 [ORG]에 다녀요. 동생과 자주 싸워요. 주 1회 상담 약속은 지키고 싶어요.",
+                    turn_id="00000000-0000-0000-0000-000000000201",
+                ),
+            ],
+        )
+        carry = memory.make_carry_over(
+            state=state,
+            session_id=sess.session_id,
+            case_id=sess.case_id,
+            session_no=sess.session_no,
+            masked_turns=sess.masked_turns(),
+            prev_rapport_credit=sess.prev_rapport_credit,
+        )
+
+        conn = FakeConn()
+        with patch.object(session_persistence, "get_pool", return_value=object()), patch.object(
+            session_persistence,
+            "acquire",
+            return_value=FakeAcquire(conn),
+        ):
+            persisted = await session_persistence.end_session(sess, carry)
+
+        self.assertTrue(persisted)
+        case_updates = [
+            args for query, args in conn.executed if "UPDATE app.case_profile" in query
+        ]
+        self.assertEqual(len(case_updates), 1)
+        _, _, case_digest, trajectory, alliance_level = case_updates[0]
+        self.assertIn("S1: 이전 회기", case_digest)
+        self.assertIn("S2:", case_digest)
+        self.assertIn("[NAME]", case_digest)
+        self.assertNotIn("김서연", case_digest)
+        self.assertEqual(trajectory[-1]["session_no"], 2)
+        self.assertEqual(trajectory[-1]["end_rapport"], 0.31)
+        self.assertGreater(alliance_level, 0.2)
+        pinned_writes = [
+            args
+            for query, args in conn.executed
+            if "INSERT INTO app.pinned_fact (" in query
+        ]
+        self.assertEqual(len(pinned_writes), 3)
+        by_key = {args[1]: args for args in pinned_writes}
+        self.assertEqual(by_key["identity:name"][2], "[NAME]")
+        self.assertEqual(by_key["identity:org"][2], "[ORG]")
+        self.assertEqual(by_key["agreement:counseling"][3], "agreement")
+        self.assertNotIn("relationship:sibling", by_key)
+        self.assertEqual(by_key["identity:name"][5], "00000000-0000-0000-0000-000000000201")
+        self.assertEqual(by_key["identity:name"][7], 2)
+        self.assertEqual(by_key["identity:name"][8], ["client", "evaluator"])
+        self.assertNotIn("김서연", " ".join(str(args[2]) for args in pinned_writes))
+        history_writes = [
+            args for query, args in conn.executed if "INSERT INTO app.pinned_fact_history" in query
+        ]
+        self.assertEqual(len(history_writes), 3)
+        self.assertEqual(history_writes[0][1], case_id)
+        self.assertIsNone(history_writes[0][2])
+        self.assertEqual(history_writes[0][3], "[NAME]")
+        self.assertEqual(history_writes[0][4], "progression")
+        self.assertEqual(history_writes[0][5], 2)
+        self.assertEqual(
+            history_writes[0][6],
+            "00000000-0000-0000-0000-000000000201",
+        )
+        self.assertNotIn("김서연", " ".join(str(args[3]) for args in history_writes))
+
+    async def test_pinned_fact_history_skips_same_value_refresh(self) -> None:
+        class FakeConn:
+            def __init__(self) -> None:
+                self.executed: list[tuple[str, tuple[object, ...]]] = []
+
+            async def fetchrow(self, query: str, *args: object):
+                self.executed.append((query, args))
+                return {
+                    "id": "00000000-0000-0000-0000-00000000fa11",
+                    "old_value": args[2],
+                    "new_value": args[2],
+                }
+
+            async def execute(self, query: str, *args: object) -> str:
+                self.executed.append((query, args))
+                return "OK"
+
+        conn = FakeConn()
+        sess = InProcSession(
+            session_id="00000000-0000-0000-0000-00000000feed",
+            case_id="00000000-0000-0000-0000-00000000ca5e",
+            learner_id="00000000-0000-0000-0000-000000000101",
+            persona_code=P1.code,
+            theory_mode="humanistic",
+            persona=P1,
+            state=state_machine.SessionState(),
+            session_no=3,
+            turns=[
+                TurnRecord(
+                    turn_seq=1,
+                    speaker="client",
+                    stage="라포",
+                    text="저는 김서연입니다.",
+                    text_masked="저는 [NAME]입니다.",
+                    turn_id="00000000-0000-0000-0000-000000000201",
+                )
+            ],
+        )
+
+        await session_persistence._upsert_pinned_fact_candidates(conn, sess)
+
+        pinned_writes = [
+            args
+            for query, args in conn.executed
+            if "INSERT INTO app.pinned_fact (" in query
+        ]
+        history_writes = [
+            args for query, args in conn.executed if "INSERT INTO app.pinned_fact_history" in query
+        ]
+        self.assertEqual(len(pinned_writes), 1)
+        self.assertEqual(history_writes, [])
+
+    async def test_pinned_fact_history_records_value_change_as_clarification(self) -> None:
+        class FakeConn:
+            def __init__(self) -> None:
+                self.executed: list[tuple[str, tuple[object, ...]]] = []
+
+            async def fetchrow(self, query: str, *args: object):
+                self.executed.append((query, args))
+                return {
+                    "id": "00000000-0000-0000-0000-00000000fa22",
+                    "old_value": "예전에는 격주 상담 약속을 말함.",
+                    "new_value": args[2],
+                }
+
+            async def execute(self, query: str, *args: object) -> str:
+                self.executed.append((query, args))
+                return "OK"
+
+        conn = FakeConn()
+        sess = InProcSession(
+            session_id="00000000-0000-0000-0000-00000000feed",
+            case_id="00000000-0000-0000-0000-00000000ca5e",
+            learner_id="00000000-0000-0000-0000-000000000101",
+            persona_code=P1.code,
+            theory_mode="humanistic",
+            persona=P1,
+            state=state_machine.SessionState(),
+            session_no=4,
+            turns=[
+                TurnRecord(
+                    turn_seq=1,
+                    speaker="client",
+                    stage="라포",
+                    text="앞으로는 주 1회 상담 약속을 지키고 싶어요.",
+                    text_masked="앞으로는 주 1회 상담 약속을 지키고 싶어요.",
+                    turn_id="00000000-0000-0000-0000-000000000202",
+                )
+            ],
+        )
+
+        await session_persistence._upsert_pinned_fact_candidates(conn, sess)
+
+        history_writes = [
+            args for query, args in conn.executed if "INSERT INTO app.pinned_fact_history" in query
+        ]
+        self.assertEqual(len(history_writes), 1)
+        self.assertEqual(history_writes[0][2], "예전에는 격주 상담 약속을 말함.")
+        self.assertIn("주 1회 상담 약속", str(history_writes[0][3]))
+        self.assertEqual(history_writes[0][4], "clarification")
+        self.assertEqual(history_writes[0][5], 4)
+
+    async def test_pinned_fact_history_records_explicit_agreement_withdrawal_as_contradiction(self) -> None:
+        test_case = self
+
+        class FakeConn:
+            def __init__(self) -> None:
+                self.executed: list[tuple[str, tuple[object, ...]]] = []
+
+            async def fetchrow(self, query: str, *args: object):
+                self.executed.append((query, args))
+                test_case.assertIn("UPDATE app.pinned_fact", query)
+                test_case.assertIn("status = 'contradicted'", query)
+                return {
+                    "id": "00000000-0000-0000-0000-00000000fa33",
+                    "old_value": "주 1회 상담 약속은 지키고 싶어요.",
+                    "new_value": args[2],
+                }
+
+            async def execute(self, query: str, *args: object) -> str:
+                self.executed.append((query, args))
+                return "OK"
+
+        conn = FakeConn()
+        sess = InProcSession(
+            session_id="00000000-0000-0000-0000-00000000feed",
+            case_id="00000000-0000-0000-0000-00000000ca5e",
+            learner_id="00000000-0000-0000-0000-000000000101",
+            persona_code=P1.code,
+            theory_mode="humanistic",
+            persona=P1,
+            state=state_machine.SessionState(),
+            session_no=5,
+            turns=[
+                TurnRecord(
+                    turn_seq=1,
+                    speaker="client",
+                    stage="라포",
+                    text="상담 약속은 이제 못 지키겠어요. 회기는 그만하고 싶어요.",
+                    text_masked="상담 약속은 이제 못 지키겠어요. 회기는 그만하고 싶어요.",
+                    turn_id="00000000-0000-0000-0000-000000000203",
+                )
+            ],
+        )
+
+        await session_persistence._upsert_pinned_fact_candidates(conn, sess)
+
+        contradiction_updates = [
+            args for query, args in conn.executed if "UPDATE app.pinned_fact" in query
+        ]
+        self.assertEqual(len(contradiction_updates), 1)
+        self.assertEqual(contradiction_updates[0][1], "agreement:counseling")
+        self.assertIn("그만하고 싶어요", str(contradiction_updates[0][2]))
+        self.assertEqual(contradiction_updates[0][7], ["evaluator"])
+        history_writes = [
+            args for query, args in conn.executed if "INSERT INTO app.pinned_fact_history" in query
+        ]
+        self.assertEqual(len(history_writes), 1)
+        self.assertEqual(history_writes[0][2], "주 1회 상담 약속은 지키고 싶어요.")
+        self.assertIn("못 지키겠어요", str(history_writes[0][3]))
+        self.assertEqual(history_writes[0][4], "contradiction")
+        self.assertEqual(history_writes[0][5], 5)
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/api/app/test_session_turn_persistence.py

@@ -7,7 +7,8 @@ import unittest
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
-from . import turn_runtime
+from . import session_persistence, turn_runtime
+from .contracts.engine_gateway import EngineGatewaySseLineDecoder
 from .deps import Principal, Role
 from .engine_client import EngineError
 from .routes import sessions
@@ -17,6 +18,14 @@ from .services.voice import TTSChunk, TranscriptResult, VoicePreset
 from .store import InProcSession, TurnRecord, store
 
 
+async def _decoded_stream_packets(stream_engine, req):
+    decoder = EngineGatewaySseLineDecoder()
+    async for raw in stream_engine.stream(req):
+        packet = decoder.feed_line(raw)
+        if packet is not None:
+            yield packet
+
+
 def _principal() -> Principal:
     return Principal(
         user_id="00000000-0000-0000-0000-000000000101",
@@ -67,6 +76,65 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
     async def asyncTearDown(self) -> None:
         store._sessions.clear()
 
+    async def test_append_turn_writes_provider_events_to_db(self) -> None:
+        class FakeConn:
+            def __init__(self) -> None:
+                self.insert_query = ""
+                self.insert_args: tuple[object, ...] = ()
+
+            async def fetchval(self, query: str, *args: object) -> object:
+                if "SELECT id FROM app.sessions" in query:
+                    return "turn-persistence-session"
+                if "COALESCE(MAX(seq)" in query:
+                    return 1
+                if "INSERT INTO app.turns" in query:
+                    self.insert_query = query
+                    self.insert_args = args
+                    return "00000000-0000-0000-0000-000000009999"
+                return None
+
+        class FakeAcquire:
+            def __init__(self, conn: FakeConn) -> None:
+                self.conn = conn
+
+            async def __aenter__(self) -> FakeConn:
+                return self.conn
+
+            async def __aexit__(self, exc_type, exc, tb) -> None:
+                return None
+
+        conn = FakeConn()
+        turn = TurnRecord(
+            turn_seq=1,
+            speaker="counselor",
+            stage="rapport",
+            text="voice text",
+            text_masked="voice text",
+            audio_ref="voice:webm:sha256:test",
+            silence_ms=1234,
+            speech_rate=210.0,
+            barge_in=True,
+            provider_events=[{"type": "sigh", "confidence": 0.82}],
+        )
+
+        with patch.object(session_persistence, "get_pool", return_value=object()), patch.object(
+            session_persistence,
+            "acquire",
+            return_value=FakeAcquire(conn),
+        ):
+            ok = await session_persistence.append_turn(
+                session_id="turn-persistence-session",
+                learner_id=_principal().user_id,
+                turn=turn,
+            )
+
+        self.assertTrue(ok)
+        self.assertIn("provider_events", conn.insert_query)
+        self.assertIn("$17::jsonb", conn.insert_query)
+        self.assertEqual(conn.insert_args[16], [{"type": "sigh", "confidence": 0.82}])
+        self.assertEqual(conn.insert_args[17], list(turn.visible_to))
+        self.assertEqual(turn.turn_id, "00000000-0000-0000-0000-000000009999")
+
     async def test_generate_turn_engine_failure_does_not_append_learner_turn(self) -> None:
         principal = _principal()
         sess = _session(principal)
@@ -96,7 +164,7 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 turn_seq=ctx.state_after.turn_seq,
                 stage=ctx.state_after.stage.value,
                 effective_openness=ctx.state_after.effective_openness,
-                client_reply="괜찮아요. 천천히 말해볼게요.",
+                client_reply="저는 김서연 씨고 한신대학교 상담심리학과 학생이에요.",
                 safety_flagged=False,
                 state_after=ctx.state_after,
                 llm_provider="claude_cli",
@@ -113,10 +181,16 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 principal,
             )
 
-        self.assertEqual(response.client_reply, "괜찮아요. 천천히 말해볼게요.")
+        self.assertEqual(response.client_reply, "저는 김서연 씨고 한신대학교 상담심리학과 학생이에요.")
         self.assertEqual(len(sess.turns), 2)
         learner_turn, client_turn = sess.turns
         self.assertIsNone(learner_turn.llm_provider)
+        self.assertEqual(client_turn.text, "저는 김서연 씨고 한신대학교 상담심리학과 학생이에요.")
+        self.assertNotIn("김서연", client_turn.text_masked)
+        self.assertNotIn("한신대학교", client_turn.text_masked)
+        self.assertNotIn("상담심리학과", client_turn.text_masked)
+        self.assertIn("[NAME]", client_turn.text_masked)
+        self.assertIn("[ORG]", client_turn.text_masked)
         self.assertEqual(client_turn.llm_provider, "claude_cli")
         self.assertEqual(client_turn.model, "gateway-default")
         self.assertEqual(client_turn.tokens_in, 17)
@@ -502,6 +576,10 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                     '"tokens_in":5,"tokens_out":7,"cost_usd":0.034567}'
                 )
 
+            async def stream_packets(self, req):
+                async for packet in _decoded_stream_packets(self, req):
+                    yield packet
+
         principal = _principal()
         sess = _session(principal)
         ctx = orchestrator.prepare_turn(
@@ -537,6 +615,10 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 yield "event: error"
                 yield 'data: {"detail":"engine unavailable: gateway"}'
 
+            async def stream_packets(self, req):
+                async for packet in _decoded_stream_packets(self, req):
+                    yield packet
+
         principal = _principal()
         sess = _session(principal)
         ctx = orchestrator.prepare_turn(
@@ -655,7 +737,15 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
         with patch.object(
             voice_routes.voice_service,
             "transcribe",
-            AsyncMock(return_value=TranscriptResult(text="오늘은 좀 힘들었어요.", duration=2.0)),
+            AsyncMock(
+                return_value=TranscriptResult(
+                    text="오늘은 좀 힘들었어요.",
+                    duration=2.0,
+                    provider_events=[
+                        {"kind": "voice_activity", "start_ms": 10, "raw_text": "drop"},
+                    ],
+                )
+            ),
         ), patch.object(
             voice_routes.orchestrator,
             "run_turn_generate",
@@ -674,6 +764,7 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 fmt="webm",
                 silence_ms=1234,
                 barge_in=True,
+                provider_events=[{"type": "sigh", "confidence": 0.82, "text": "drop"}],
             )
 
         self.assertEqual(len(sess.turns), 2)
@@ -682,7 +773,25 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(learner_turn.silence_ms, 1234)
         self.assertGreater(learner_turn.speech_rate or 0, 0)
         self.assertTrue(learner_turn.barge_in)
+        self.assertEqual(
+            learner_turn.provider_events,
+            [
+                {
+                    "type": "sigh",
+                    "confidence": 0.82,
+                    "event_type": "sigh",
+                    "category": "paralinguistic",
+                },
+                {
+                    "kind": "voice_activity",
+                    "start_ms": 10,
+                    "event_type": "voice_activity",
+                    "category": "speech_activity",
+                },
+            ],
+        )
         self.assertIsNone(client_turn.audio_ref)
+        self.assertEqual(client_turn.provider_events, [])
         self.assertEqual(client_turn.llm_provider, "claude_cli")
         self.assertTrue(any(message.get("type") == "tts_end" for message in websocket.messages))
 
@@ -703,6 +812,26 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                     silence_ms=1234,
                     speech_rate=420.0,
                     barge_in=True,
+                    provider_events=[
+                        {
+                            "event_type": "sigh",
+                            "category": "paralinguistic",
+                            "confidence": 0.82,
+                            "provider": "stt-provider",
+                            "type": "raw_sigh",
+                        },
+                        {
+                            "event_type": "speech_start",
+                            "category": "speech_activity",
+                            "start_ms": 100,
+                        },
+                        {
+                            "event_type": "background_noise",
+                            "category": "audio_quality",
+                            "score": 77,
+                            "label": "busy cafe",
+                        },
+                    ],
                 ),
                 TurnRecord(
                     turn_seq=2,
@@ -715,6 +844,7 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                     silence_ms=2500,
                     speech_rate=180.0,
                     barge_in=True,
+                    provider_events=[{"event_type": "cry", "category": "paralinguistic"}],
                 ),
             ]
         )
@@ -723,10 +853,20 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual(len(response.turns), 2)
         learner_turn, client_turn = response.turns
-        self.assertEqual([event.kind for event in learner_turn.nonverbal], ["silence", "pace", "barge_in", "audio"])
+        self.assertEqual(
+            [event.kind for event in learner_turn.nonverbal],
+            ["silence", "pace", "barge_in", "audio", "paralinguistic", "audio_quality"],
+        )
         self.assertEqual(learner_turn.nonverbal[0].label, "침묵")
         self.assertEqual(learner_turn.nonverbal[0].detail, "1.2초")
         self.assertEqual(learner_turn.nonverbal[1].detail, "분당 420자")
+        self.assertEqual(learner_turn.nonverbal[4].label, "음성 단서")
+        self.assertEqual(learner_turn.nonverbal[4].detail, "한숨 감지 · 신뢰도 82%")
+        exposed_details = " ".join(event.detail for event in learner_turn.nonverbal)
+        self.assertNotIn("stt-provider", exposed_details)
+        self.assertNotIn("raw_sigh", exposed_details)
+        self.assertEqual(learner_turn.nonverbal[5].label, "오디오 품질")
+        self.assertEqual(learner_turn.nonverbal[5].detail, "배경 소음 · 신뢰도 77%")
         self.assertEqual(client_turn.nonverbal, [])
 
     async def test_review_includes_case_formulation_worksheet_draft(self) -> None:

apps/api/app/test_voice_ws.py

@@ -77,6 +77,50 @@ class VoiceWebSocketContractTest(unittest.IsolatedAsyncioTestCase):
             {"degraded": False, "persona_catalog_source": "session"},
         )
 
+    def test_provider_events_get_internal_taxonomy_without_raw_payload(self) -> None:
+        events = voice_routes._safe_provider_events(
+            [
+                {
+                    "type": "SIGH",
+                    "confidence": 0.81,
+                    "text": "raw transcript must drop",
+                },
+                {
+                    "kind": "voice_activity",
+                    "start_ms": 10,
+                    "raw_text": "drop",
+                },
+                {
+                    "label": "vendor custom marker",
+                    "score": 0.44,
+                },
+            ]
+        )
+
+        self.assertEqual(
+            events,
+            [
+                {
+                    "type": "SIGH",
+                    "confidence": 0.81,
+                    "event_type": "sigh",
+                    "category": "paralinguistic",
+                },
+                {
+                    "kind": "voice_activity",
+                    "start_ms": 10,
+                    "event_type": "voice_activity",
+                    "category": "speech_activity",
+                },
+                {
+                    "label": "vendor custom marker",
+                    "score": 0.44,
+                    "event_type": "vendor_custom_marker",
+                    "category": "unknown",
+                },
+            ],
+        )
+
     async def test_audio_start_binary_chunks_audio_end_ping_close_contract(self) -> None:
         websocket = FakeWebSocket(
             [
@@ -90,6 +134,15 @@ class VoiceWebSocketContractTest(unittest.IsolatedAsyncioTestCase):
                         "format": "webm",
                         "silence_ms": "450",
                         "barge_in": "true",
+                        "provider_events": [
+                            {
+                                "type": "sigh",
+                                "confidence": 0.82,
+                                "text": "raw transcript must not persist",
+                            },
+                            {"kind": "noise", "label": "x" * 120},
+                            "invalid",
+                        ],
                     }
                 ),
                 _control({"type": "close"}),
@@ -141,6 +194,23 @@ class VoiceWebSocketContractTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(kwargs["audio_ended_at"], 12.0)
         self.assertEqual(kwargs["silence_ms"], 450)
         self.assertIs(kwargs["barge_in"], True)
+        self.assertEqual(
+            kwargs["provider_events"],
+            [
+                {
+                    "type": "sigh",
+                    "confidence": 0.82,
+                    "event_type": "sigh",
+                    "category": "paralinguistic",
+                },
+                {
+                    "kind": "noise",
+                    "label": "x" * 80,
+                    "event_type": "background_noise",
+                    "category": "audio_quality",
+                },
+            ],
+        )
 
     async def test_text_turn_strips_text_runs_turn_and_ping_close_still_work(self) -> None:
         websocket = FakeWebSocket(

infra/db/init/02_schema.sql

@@ -221,6 +221,7 @@ CREATE TABLE IF NOT EXISTS app.turns (
   silence_ms  INT,
   speech_rate REAL,
   barge_in    BOOLEAN,
+  provider_events JSONB NOT NULL DEFAULT '[]'::jsonb,
   created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
   UNIQUE (session_id, seq)
 );

infra/db/init/04_audit_eval_rls.sql

@@ -328,6 +328,31 @@ CREATE POLICY p_pinned_select ON app.pinned_fact FOR SELECT USING (
     AND current_setting('app.current_ai_view', true) = ANY(visible_to) )
   OR app.current_role_name() IN ('admin','instructor')
 );
+DROP POLICY IF EXISTS p_pinned_insert ON app.pinned_fact;
+CREATE POLICY p_pinned_insert ON app.pinned_fact FOR INSERT WITH CHECK (
+  app.current_role_name() IN ('admin','instructor')
+  OR EXISTS (
+    SELECT 1 FROM app.case_profile cp
+    WHERE cp.case_id = app.pinned_fact.case_id
+      AND cp.learner_id = app.current_uid()
+  )
+);
+DROP POLICY IF EXISTS p_pinned_update ON app.pinned_fact;
+CREATE POLICY p_pinned_update ON app.pinned_fact FOR UPDATE USING (
+  app.current_role_name() IN ('admin','instructor')
+  OR EXISTS (
+    SELECT 1 FROM app.case_profile cp
+    WHERE cp.case_id = app.pinned_fact.case_id
+      AND cp.learner_id = app.current_uid()
+  )
+) WITH CHECK (
+  app.current_role_name() IN ('admin','instructor')
+  OR EXISTS (
+    SELECT 1 FROM app.case_profile cp
+    WHERE cp.case_id = app.pinned_fact.case_id
+      AND cp.learner_id = app.current_uid()
+  )
+);
 
 -- ── learner_profile: 학습자=본인, persistent_gaps 는 응답단 필터 ──
 ALTER TABLE app.learner_profile ENABLE ROW LEVEL SECURITY;