회기 연속성과 멀티 케이스 계약을 영속화

apps/api/app/config.py

@@ -119,7 +119,10 @@ class Settings(BaseSettings):
         validation_alias="EVALUATOR_SEMANTIC_CACHE_MAX_ENTRIES",
     )
     session_evaluation_timeout: float = Field(
-        default=0.0,
+        # deep-loop은 전체 축어록·구조화 출력을 생성하므로 실시간/일반 HTTP 120초와
+        # 분리한다. 이 값은 생성 deadline이고, 호출부가 gateway 초기화·응답 수신 여유를
+        # 별도로 더한다.
+        default=300.0,
         validation_alias="SESSION_EVALUATION_TIMEOUT",
     )
     session_evaluation_recovery_limit: int = Field(

apps/api/app/routes/eval.py

@@ -28,7 +28,15 @@ from ..deps import Principal, Role, require_role
 from ..engine_client import EngineError, engine_client
 from ..runtime_policy import runtime_fallback_allowed
 from ..session_evaluation_input import enriched_masked_turns
-from ..session_read_model import StageLabel, stage_label_or_none
+from ..session_evaluation_timeout import (
+    session_evaluation_transport_timeout_seconds,
+)
+from ..session_read_model import (
+    ReviewEvaluationFailure,
+    StageLabel,
+    stage_label_or_none,
+    teacher_evaluation_failure,
+)
 from ..services import evaluator
 from ..services.evaluator import SessionEvaluation, TurnEvaluation
 from ..store import InProcSession
@@ -51,12 +59,13 @@ class TurnReevaluateRequest(BaseModel):
 
 
 class EvaluationSummary(BaseModel):
-    """회기 평가 조회 응답(분포 + deep 결과 합본)."""
+    """회기 평가 조회 응답(분포 + deep 결과 합본, provider 오류 원문 제외)."""
 
     session_id: str
     stage: StageLabel | None = None
     status: str | None = None
     error: str | None = None
+    failure: ReviewEvaluationFailure | None = None
     durable: bool = False
     deep: Optional[dict[str, Any]] = None
     distribution: dict[str, Any] = Field(default_factory=dict)
@@ -108,6 +117,15 @@ def _session_evaluation_error_status(error: str) -> int:
     return status.HTTP_502_BAD_GATEWAY
 
 
+def _safe_session_evaluation_retry_detail() -> str:
+    """재시도 HTTP 응답에서는 provider 예외 원문을 내보내지 않는다.
+
+    원인은 durable evaluation record에 서버 전용으로 보존하고, 교수자 화면은 review의
+    안전 분류(evaluationFailure)로 다음 행동만 안내한다.
+    """
+    return "AI 평가를 완료하지 못했습니다. 최신 평가 상태를 확인해 주세요."
+
+
 # ════════════════════════════════════════════════════════════════════════════
 # 회기 deep-loop 재평가 트리거 (교수자/관리자)
 # ════════════════════════════════════════════════════════════════════════════
@@ -120,7 +138,8 @@ async def reevaluate_session(
     """회기 전체 deep-loop 재평가(슈퍼바이저 rationale/critique + 개선점 + 대안발화).
 
     저장된 마스킹 축어록을 evaluator.evaluate_session 으로 평가한다.
-    엔진 장애는 503 으로 변환(평가는 비치명적이지만 트리거는 사용자 명시 요청이라 에러 노출).
+    엔진 장애는 503으로 변환한다. provider 예외 원문은 durable 기록에만 남기고 HTTP에는
+    안전한 안내만 반환한다.
     """
     sess = await _load_session_or_404(session_id, principal)
     counselor_identity = getattr(sess, "learner_label", None)
@@ -144,6 +163,7 @@ async def reevaluate_session(
             theory_mode=_theory_mode_of(sess),
             scope=body.scope if body.scope in ("session_end", "stage_transition") else "session_end",
             audit_hook=session_persistence.record_llm_call_audit,
+            timeout=session_evaluation_transport_timeout_seconds(),
         )
     except EngineError as e:
         detail = f"engine unavailable: {e}"
@@ -163,7 +183,10 @@ async def reevaluate_session(
                 session_id,
                 write.error,
             )
-        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=detail)
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=_safe_session_evaluation_retry_detail(),
+        )
 
     write = session_persistence.SessionEvaluationWrite.from_result(
         session_id=session_id,
@@ -178,7 +201,10 @@ async def reevaluate_session(
         logger.error("%s: session_id=%s status=%s", detail, session_id, write.status)
         raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, detail=detail)
     if result.error:
-        raise HTTPException(_session_evaluation_error_status(result.error), detail=result.error)
+        raise HTTPException(
+            _session_evaluation_error_status(result.error),
+            detail=_safe_session_evaluation_retry_detail(),
+        )
     return result
 
 
@@ -256,7 +282,10 @@ async def reevaluate_turn(
         )
     learner.evaluation = result_payload
     if result.error:
-        raise HTTPException(_session_evaluation_error_status(result.error), detail=result.error)
+        raise HTTPException(
+            _session_evaluation_error_status(result.error),
+            detail=_safe_session_evaluation_retry_detail(),
+        )
     return result
 
 
@@ -285,13 +314,18 @@ async def get_session_evaluation(
             distribution={},
         )
     payload = record.get("payload")
-    deep = payload if isinstance(payload, dict) else {}
+    deep = dict(payload) if isinstance(payload, dict) else {}
+    # SessionEvaluation.to_dict()에는 server-side error가 함께 저장된다. deep 객체도
+    # 교수자 API 경계에서는 같은 원칙으로 제거한다.
+    deep.pop("error", None)
     distribution = deep.get("distribution")
+    failure = teacher_evaluation_failure(record)
     return EvaluationSummary(
         session_id=session_id,
         stage=_summary_stage(record.get("stage") or deep.get("stage")),
         status=str(record.get("status") or "") or None,
-        error=str(record.get("error") or "") or None,
+        error=_safe_session_evaluation_retry_detail() if failure is not None else None,
+        failure=failure,
         durable=durable,
         deep=deep,
         distribution=distribution if isinstance(distribution, dict) else {},

apps/api/app/routes/sessions.py

@@ -15,6 +15,7 @@ import secrets
 import time
 from datetime import datetime
 from typing import Literal, Optional
+from uuid import UUID
 
 from fastapi import APIRouter, HTTPException, Request, status
 from pydantic import BaseModel, Field, field_validator
@@ -28,6 +29,12 @@ from ..engine_client import EngineError, engine_client
 from ..persona_repository import get_catalog_persona
 from ..runtime_policy import require_runtime_fallback_allowed
 from ..session_evaluation_input import enriched_masked_turns
+from ..session_evaluation_timeout import (
+    session_evaluation_outer_timeout_seconds,
+    session_evaluation_timeout_seconds as _session_evaluation_timeout_seconds,
+    session_evaluation_stale_after_seconds,
+    session_evaluation_transport_timeout_seconds,
+)
 from ..services import (
     evaluator,
     feedback_policy,
@@ -44,11 +51,14 @@ from ..services import (
     state_machine,
 )
 from ..session_read_model import (
+    CaseMemoryPreview,
+    CaseProgressStats,
+    LearnerCaseListResponse,
+    LearnerCaseSummary,
     LearnerDashboardResponse,
     LearnerSessionsResponse,
     LearnerSessionSummary,
     LEARNER_VISIBLE_AI_ROLE,
-    MISSING_SESSION_EVALUATION_GRACE_SECONDS,
     ReviewCaseWorksheet,
     ReviewCaseWorksheetSaveRequest,
     ReviewWorksheetItem as ReviewWorksheetItem,
@@ -96,6 +106,10 @@ EndStateValue = str | int | float | bool | None | dict[str, float]
 class SessionStartRequest(BaseModel):
     persona_code: str = Field(..., examples=["P1"])
     theory_mode: TheoryMode = "humanistic"
+    # continue는 선택한 사례의 압축 기억을 이어 받고, fresh는 새 case_id/S1으로 시작한다.
+    # 구클라이언트는 기존 동작을 보존하도록 continue가 기본이다.
+    start_mode: Literal["continue", "fresh"] = "continue"
+    case_id: UUID | None = None
     # 이번 회기 목표 단계(2026-07-13 회의 P1). 회의 권장은 2개 수준이지만
     # 소유자 지시(2026-07-15)로 1~4개까지 자유 선택을 허용한다.
     # 빈 리스트는 구계약 클라이언트 호환용 — 준비 페이지는 항상 1개 이상을 보낸다.
@@ -127,6 +141,7 @@ class SessionStartResponse(BaseModel):
     duration_limit_seconds: int = 0
     warning_before_end_seconds: int = 0
     learner_feedback_enabled: bool = True
+    start_mode: Literal["continue", "fresh"] = "continue"
 
 
 class TurnRequest(BaseModel):
@@ -916,6 +931,7 @@ async def _generate_and_save_session_evaluation(sess: InProcSession) -> None:
         return
 
     timeout_seconds = _session_evaluation_timeout_seconds()
+    transport_timeout_seconds = session_evaluation_transport_timeout_seconds()
     enriched = enriched_masked_turns(
         sess.masked_turns(),
         counselor_identity=sess.learner_label,
@@ -933,8 +949,12 @@ async def _generate_and_save_session_evaluation(sess: InProcSession) -> None:
                 theory_mode=sess.theory_mode,
                 scope="session_end",
                 audit_hook=session_persistence.record_llm_call_audit,
+                timeout=transport_timeout_seconds,
             ),
-            timeout=timeout_seconds,
+            # gateway의 생성 deadline과 HTTP deadline을 같게 두면 요청 초기화·응답 수신
+            # 비용만으로 app이 먼저 취소될 수 있다. transport grace와 durable audit 기록
+            # 예산 뒤에 outer grace를 두어 정상 결과를 timeout error로 바꾸지 않는다.
+            timeout=session_evaluation_outer_timeout_seconds(),
         )
         write = session_persistence.SessionEvaluationWrite.from_result(
             session_id=sess.session_id,
@@ -1050,9 +1070,7 @@ async def recover_missing_session_evaluations(*, limit: int | None = None) -> in
     )
     if recovery_limit <= 0:
         return 0
-    stale_after_seconds = (
+    stale_after_seconds = session_evaluation_stale_after_seconds()
-        _session_evaluation_timeout_seconds() + MISSING_SESSION_EVALUATION_GRACE_SECONDS
-    )
     (
         candidates,
         durable,
@@ -1108,11 +1126,6 @@ def cancel_missing_session_evaluation_recovery() -> None:
         task.cancel()
 
 
-def _session_evaluation_timeout_seconds() -> float:
-    configured = float(settings.session_evaluation_timeout or settings.engine_timeout)
-    return max(configured, 1.0)
-
-
 async def _enqueue_session_review_ready_notification(session_id: str) -> None:
     try:
         await notifications.enqueue_session_review_ready(session_id=session_id)
@@ -1236,6 +1249,180 @@ async def list_learner_sessions(principal: CurrentPrincipal) -> LearnerSessionsR
     )
 
 
+def _preview_text(value: object, *, limit: int) -> str | None:
+    """Keep the learner foldout bounded even when a legacy digest is verbose."""
+    text = str(value or "").strip()
+    if not text:
+        return None
+    if len(text) <= limit:
+        return text
+    return f"{text[: max(1, limit - 1)].rstrip()}…"
+
+
+def _preview_items(value: object, *, limit: int, item_limit: int) -> list[str]:
+    if not isinstance(value, (list, tuple)):
+        return []
+    items: list[str] = []
+    for raw in value:
+        clipped = _preview_text(raw, limit=item_limit)
+        if clipped:
+            items.append(clipped)
+        if len(items) >= limit:
+            break
+    return items
+
+
+def _db_datetime_iso(value: object) -> str | None:
+    return value.isoformat() if isinstance(value, datetime) else None
+
+
+@router.get("/cases", response_model=LearnerCaseListResponse)
+async def list_learner_cases(
+    persona_code: str,
+    principal: CurrentPrincipal,
+) -> LearnerCaseListResponse:
+    """Return complete case-local progress for one NPC, not a capped history slice."""
+    principal = _ensure_learner(principal)
+    try:
+        catalog_persona = await get_catalog_persona(persona_code)
+    except Exception as exc:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="persona catalog database unavailable",
+        ) from exc
+    if catalog_persona is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND, detail=f"unknown persona {persona_code}"
+        )
+    try:
+        rows = await session_persistence.list_case_summaries(
+            learner_id=principal.user_id,
+            persona_id=catalog_persona.persona_id,
+        )
+    except session_persistence.CaseProgressUnavailableError as exc:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="case_progress_unavailable",
+        ) from exc
+
+    card = catalog_persona.card
+    return LearnerCaseListResponse(
+        cases=[
+            LearnerCaseSummary(
+                case_id=str(row["case_id"]),
+                persona_code=card.code,
+                persona_name=card.display_name,
+                last_session_no=int(row["last_session_no"] or 0),
+                progress=CaseProgressStats(
+                    total_sessions=int(row["total_sessions"] or 0),
+                    completed_sessions=int(row["completed_sessions"] or 0),
+                    total_turns=int(row["total_turns"] or 0),
+                    total_duration_seconds=int(row["total_duration_seconds"] or 0),
+                    active_session_id=(
+                        str(row["active_session_id"])
+                        if row.get("active_session_id") is not None
+                        else None
+                    ),
+                    active_session_no=(
+                        int(row["active_session_no"])
+                        if row.get("active_session_no") is not None
+                        else None
+                    ),
+                    active_started_at=_db_datetime_iso(row.get("active_started_at")),
+                    last_activity_at=_db_datetime_iso(row.get("last_activity_at")),
+                ),
+            )
+            for row in rows
+        ]
+    )
+
+
+@router.get("/cases/{case_id}/memory", response_model=CaseMemoryPreview)
+async def get_learner_case_memory_preview(
+    case_id: UUID,
+    principal: CurrentPrincipal,
+) -> CaseMemoryPreview:
+    """Load only the learner-safe compact memory when its foldout is opened."""
+    principal = _ensure_learner(principal)
+    case_key = str(case_id)
+    try:
+        async with db.acquire(role="learner", user_id=principal.user_id) as conn:
+            case_row = await conn.fetchrow(
+                """
+                SELECT case_digest
+                FROM app.case_profile
+                WHERE case_id = $1::uuid
+                  AND learner_id = $2::uuid
+                """,
+                case_key,
+                principal.user_id,
+            )
+            if case_row is None:
+                raise HTTPException(status.HTTP_404_NOT_FOUND, detail="case_not_found")
+            summary_row = await conn.fetchrow(
+                """
+                SELECT ss.digest, ss.open_threads
+                FROM app.session_summary AS ss
+                JOIN app.sessions AS s ON s.id = ss.session_id
+                WHERE ss.case_id = $1::uuid
+                  AND s.learner_id = $2::uuid
+                ORDER BY ss.session_no DESC, ss.created_at DESC
+                LIMIT 1
+                """,
+                case_key,
+                principal.user_id,
+            )
+            fact_rows = await conn.fetch(
+                """
+                SELECT LEFT(pf.value, 160) AS value
+                FROM app.pinned_fact AS pf
+                JOIN app.case_profile AS cp ON cp.case_id = pf.case_id
+                WHERE pf.case_id = $1::uuid
+                  AND cp.learner_id = $2::uuid
+                  AND pf.status IN ('stable', 'evolving', 'locked')
+                  AND 'client' = ANY(pf.visible_to)
+                ORDER BY pf.updated_at DESC
+                LIMIT 8
+                """,
+                case_key,
+                principal.user_id,
+            )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.exception("case memory preview read failed", extra={"case_id": case_key})
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="case_memory_unavailable",
+        ) from exc
+
+    case_digest = _preview_text(case_row["case_digest"], limit=600)
+    latest_session_digest = _preview_text(
+        summary_row["digest"] if summary_row is not None else None,
+        limit=600,
+    )
+    open_threads = _preview_items(
+        summary_row["open_threads"] if summary_row is not None else [],
+        limit=6,
+        item_limit=160,
+    )
+    pinned_facts = _preview_items(
+        [row["value"] for row in fact_rows],
+        limit=8,
+        item_limit=160,
+    )
+    return CaseMemoryPreview(
+        case_id=case_key,
+        memory_available=bool(
+            case_digest or latest_session_digest or open_threads or pinned_facts
+        ),
+        case_digest=case_digest,
+        latest_session_digest=latest_session_digest,
+        open_threads=open_threads,
+        pinned_facts=pinned_facts,
+    )
+
+
 @router.get("/dashboard", response_model=LearnerDashboardResponse)
 async def learner_dashboard(principal: CurrentPrincipal) -> LearnerDashboardResponse:
     """Return the current learner's real practice dashboard aggregates."""
@@ -1391,15 +1578,16 @@ async def start_session(
             status.HTTP_404_NOT_FOUND, detail=f"unknown persona {body.persona_code}"
         )
     card = catalog_persona.card
+    if body.start_mode == "fresh" and body.case_id is not None:
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="fresh_start_must_not_select_case",
+        )
 
-    case_context = await session_persistence.get_case_context(
+    # The durable transaction chooses or creates the case only after active-case
-        learner_id=principal.user_id,
+    # validation. Start with an empty recall here so a fresh request cannot see
-        persona_id=catalog_persona.persona_id,
+    # any legacy case before its own empty case exists.
-    )
+    recall = memory.build_recall_context()
-    recall = await _build_seed_recall(
-        case_id=case_context.case_id if case_context else None
-    )
-    session_no = (case_context.last_session_no + 1) if case_context else 1
     st = state_machine.init_state(
         params=card.openness_params(),
         carry=recall.carry,
@@ -1408,20 +1596,51 @@ async def start_session(
     carry_rapport = st.rapport_credit
     goal_stages = [str(stage) for stage in body.goal_stages]
     learner_feedback_enabled = principal.learner_feedback_enabled
+
+    async def build_locked_start_state(
+        stable_case_id: str,
+        _session_no: int,
+    ) -> state_machine.SessionState:
+        nonlocal recall
+        recall = (
+            memory.build_recall_context()
+            if body.start_mode == "fresh"
+            else await _build_seed_recall(case_id=stable_case_id)
+        )
+        return state_machine.init_state(
+            params=card.openness_params(),
+            carry=recall.carry,
+        )
+
     try:
         sess = await session_persistence.create_session(
             learner_id=principal.user_id,
             card=card,
             theory_mode=body.theory_mode,
             state=st,
-            session_no=session_no,
+            session_no=1,
             carry_rapport=carry_rapport,
             persona_id=catalog_persona.persona_id,
             persona_version=catalog_persona.version,
-            case_id=case_context.case_id if case_context else None,
+            case_id=str(body.case_id) if body.case_id is not None else None,
+            start_mode=body.start_mode,
             goal_stages=goal_stages,
             learner_feedback_enabled=learner_feedback_enabled,
+            locked_state_factory=build_locked_start_state,
         )
+    except session_persistence.ActiveSessionExistsError as exc:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            detail={
+                "code": "active_session_exists",
+                "session_id": exc.session_id,
+            },
+        ) from exc
+    except session_persistence.CaseNotFoundError as exc:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            detail="case_not_found",
+        ) from exc
     except session_persistence.SessionCreationPersistenceError as exc:
         raise HTTPException(
             status.HTTP_503_SERVICE_UNAVAILABLE,
@@ -1430,6 +1649,64 @@ async def start_session(
     degraded = catalog_persona.degraded or sess is None
     if sess is None:
         require_runtime_fallback_allowed("session creation")
+        # Runtime fallback cannot prove durable case memory. Keep it empty rather
+        # than leaking a guessed legacy recall, while preserving a selected
+        # in-process case ID when one is available.
+        recall = memory.build_recall_context()
+        st = state_machine.init_state(
+            params=card.openness_params(),
+            carry=recall.carry,
+        )
+        carry_rapport = st.rapport_credit
+        active_session = store.find_active(
+            learner_id=principal.user_id,
+            persona_id=catalog_persona.persona_id,
+            persona_code=card.code,
+        )
+        if active_session is not None:
+            raise HTTPException(
+                status.HTTP_409_CONFLICT,
+                detail={
+                    "code": "active_session_exists",
+                    "session_id": active_session.session_id,
+                },
+            )
+        runtime_case_id: str | None = None
+        runtime_session_no = 1
+        if body.start_mode == "continue":
+            related = [
+                candidate
+                for candidate in store.list()
+                if candidate.learner_id == principal.user_id
+                and (
+                    candidate.persona_id == catalog_persona.persona_id
+                    or candidate.persona_code == card.code
+                )
+            ]
+            if body.case_id is not None:
+                runtime_case_id = str(body.case_id)
+                related = [
+                    candidate
+                    for candidate in related
+                    if candidate.case_id == runtime_case_id
+                ]
+                if not related:
+                    raise HTTPException(
+                        status.HTTP_404_NOT_FOUND,
+                        detail="case_not_found",
+                    )
+            elif related:
+                newest = max(related, key=lambda candidate: candidate.created_at)
+                runtime_case_id = newest.case_id
+                related = [
+                    candidate
+                    for candidate in related
+                    if candidate.case_id == runtime_case_id
+                ]
+            if related:
+                runtime_session_no = max(
+                    candidate.session_no for candidate in related
+                ) + 1
         sess = store.create(
             learner_id=principal.user_id,
             persona=card,
@@ -1437,12 +1714,14 @@ async def start_session(
             state=st,
             persona_id=catalog_persona.persona_id,
             persona_version=catalog_persona.version,
-            session_no=session_no,
+            case_id=runtime_case_id,
+            session_no=runtime_session_no,
             carry_rapport=carry_rapport,
             goal_stages=goal_stages,
             learner_feedback_enabled=learner_feedback_enabled,
         )
     else:
+        st = sess.state
         store.put(sess)
 
     # DB 재조회 전의 첫 턴과 runtime fallback에서도 인증된 학습자 표시명을
@@ -1470,6 +1749,7 @@ async def start_session(
         duration_limit_seconds=settings.session_duration_minutes * 60,
         warning_before_end_seconds=settings.session_warning_minutes * 60,
         learner_feedback_enabled=sess.learner_feedback_enabled,
+        start_mode=body.start_mode,
     )
 
 

apps/api/app/routes/teacher.py

@@ -16,6 +16,8 @@ from ..session_read_model import (
     learner_visible_turns,
     missing_session_evaluation_record,
     stage_label,
+    teacher_evaluation_failure,
+    teacher_evaluation_failure_message,
 )
 from ..services import session_metrics
 from ..stage_contract import STAGE_LABEL_VALUES
@@ -298,6 +300,7 @@ def _summary(
     learner_turns = sum(1 for turn in visible_turns if turn.speaker == "counselor")
     client_turns = sum(1 for turn in visible_turns if turn.speaker == "client")
     evaluation_status = _evaluation_status_value(summary_evaluation_record)
+    evaluation_failure = teacher_evaluation_failure(summary_evaluation_record)
     return TeacherSessionSummary(
         session_id=sess.session_id,
         learner_id=sess.learner_id,
@@ -322,9 +325,7 @@ def _summary(
             summary_evaluation_record,
             has_visible_turns=bool(visible_turns),
         ),
-        evaluation_error=(
+        evaluation_error=teacher_evaluation_failure_message(evaluation_failure),
-            str(summary_evaluation_record.get("error") or "") if summary_evaluation_record else None
-        ),
     )
 
 

apps/api/app/services/evaluator.py

@@ -820,6 +820,7 @@ async def evaluate_session(
     theory_mode: Optional[str] = None,
     scope: str = "session_end",
     audit_hook: Optional["LlmAuditHook"] = None,
+    timeout: float | None = None,
 ) -> SessionEvaluation:
     """deep-loop 정밀 평가 — 단계전환/회기말. 전체 축어록 + 코드 집계 분포 + LLM 정성 평가.
 
@@ -859,7 +860,7 @@ async def evaluate_session(
         cached = _evaluator_cache_get(cache_key)
         if cached is not None:
             return SessionEvaluation.model_validate(cached)
-        resp = await generate_with_audit(engine, req, audit_hook)
+        resp = await generate_with_audit(engine, req, audit_hook, timeout=timeout)
     except EngineError as e:
         base.error = f"engine_error: {e}"
         return base

apps/api/app/services/llm_audit.py

@@ -28,10 +28,18 @@ async def generate_with_audit(
     engine: EngineClient,
     request: GenerateRequest,
     audit_hook: LlmAuditHook | None,
+    *,
+    timeout: float | None = None,
 ) -> GenerateResponse:
     """비스트리밍 LLM 호출의 지연·토큰·비용 기록을 한 계약으로 고정한다."""
     started = time.perf_counter()
-    response = await engine.generate(request)
+    # deep-loop처럼 호출별 deadline이 필요한 경로가 audit 경계를 지나도 사라지지
+    # 않게 한다. 기본(None)은 기존 엔진 client deadline을 그대로 쓴다.
+    response = (
+        await engine.generate(request, timeout=timeout)
+        if timeout is not None
+        else await engine.generate(request)
+    )
     latency_ms = int((time.perf_counter() - started) * 1000)
     await record_llm_audit(
         audit_hook,

apps/api/app/session_evaluation_timeout.py

@@ -0,0 +1,58 @@
+"""deep-loop 평가의 실행 deadline과 HTTP transport deadline 계약."""
+
+from __future__ import annotations
+
+from .config import settings
+
+# 게이트웨이의 모델 생성 deadline 뒤에도 세션 초기화·요청 직렬화·응답 수신이 남는다.
+# 짧은 운영자 override는 빠르게 취소할 수 있도록 5초부터, 기본 300초에서는 30초까지만 준다.
+SESSION_EVALUATION_TRANSPORT_GRACE_MIN_SECONDS = 5.0
+SESSION_EVALUATION_TRANSPORT_GRACE_MAX_SECONDS = 30.0
+SESSION_EVALUATION_TRANSPORT_GRACE_RATIO = 0.10
+SESSION_EVALUATION_OUTER_GRACE_SECONDS = 5.0
+SESSION_EVALUATION_STALE_STABILITY_SECONDS = 5.0
+
+
+def session_evaluation_timeout_seconds() -> float:
+    """평가 엔진이 생성에 쓸 수 있는 의미상 deadline."""
+    configured = float(settings.session_evaluation_timeout or settings.engine_timeout)
+    return max(configured, 1.0)
+
+
+def session_evaluation_transport_timeout_seconds() -> float:
+    """HTTP 호출이 gateway의 생성 deadline보다 먼저 취소되지 않는 경계."""
+    generation_timeout = session_evaluation_timeout_seconds()
+    grace = min(
+        SESSION_EVALUATION_TRANSPORT_GRACE_MAX_SECONDS,
+        max(
+            SESSION_EVALUATION_TRANSPORT_GRACE_MIN_SECONDS,
+            generation_timeout * SESSION_EVALUATION_TRANSPORT_GRACE_RATIO,
+        ),
+    )
+    return generation_timeout + grace
+
+
+def session_evaluation_outer_timeout_seconds() -> float:
+    """background task가 생성 뒤 감사 원장 기록까지 기다리는 최종 deadline.
+
+    ``evaluate_session()``은 engine transport가 끝난 뒤에도 audit hook을 await한다.
+    그 hook은 durable DB에서는 ``db_command_timeout``까지 걸릴 수 있으므로,
+    transport grace만 더한 outer deadline으로 감싸면 정상 모델 결과를 timeout error로
+    잘못 저장할 수 있다. 결과 평가 원장 저장은 이 await 바깥에서 한 번 더 일어나며,
+    stale deadline이 그 별도 예산을 보유한다.
+    """
+    audit_persistence_timeout = max(float(settings.db_command_timeout), 1.0)
+    return (
+        session_evaluation_transport_timeout_seconds()
+        + audit_persistence_timeout
+        + SESSION_EVALUATION_OUTER_GRACE_SECONDS
+    )
+
+
+def session_evaluation_stale_after_seconds() -> float:
+    """실제 평가 task와 DB 저장이 모두 끝난 뒤에만 누락으로 판정하는 시점."""
+    return (
+        session_evaluation_outer_timeout_seconds()
+        + max(float(settings.db_command_timeout), 1.0)
+        + SESSION_EVALUATION_STALE_STABILITY_SECONDS
+    )

apps/api/app/session_persistence.py

@@ -9,7 +9,7 @@ import time
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any, Iterable, Protocol
+from typing import Any, Awaitable, Callable, Iterable, Literal, Protocol
 
 from .db import acquire, get_pool
 from .deps import Principal
@@ -38,6 +38,22 @@ class SessionCreationPersistenceError(RuntimeError):
     """fail-closed 환경에서 영속 세션 생성이 실패했다."""
 
 
+class ActiveSessionExistsError(RuntimeError):
+    """같은 learner-persona 전체에 미종료 회기가 이미 존재한다."""
+
+    def __init__(self, session_id: str) -> None:
+        self.session_id = session_id
+        super().__init__("active_session_exists")
+
+
+class CaseNotFoundError(RuntimeError):
+    """선택한 연속 사례가 이 learner-persona에 존재하지 않는다."""
+
+
+class CaseProgressUnavailableError(RuntimeError):
+    """DB 전체 집계가 필요한 사례 진행 수치를 안전하게 읽지 못했다."""
+
+
 _EVALUATION_CACHE: dict[str, dict[str, Any]] = {}
 _CASE_WORKSHEET_CACHE: dict[str, dict[str, Any]] = {}
 _SESSION_REVIEW_STATUS_CACHE: dict[str, dict[str, Any]] = {}
@@ -1768,7 +1784,9 @@ async def save_session_evaluation(write: SessionEvaluationWrite) -> bool:
     )
     record = write.cache_record()
     if runtime_fallback_allowed():
-        _EVALUATION_CACHE[write.session_id] = record
+        existing = _EVALUATION_CACHE.get(write.session_id)
+        if _should_replace_evaluation_record(existing, record):
+            _EVALUATION_CACHE[write.session_id] = record
     try:
         get_pool()
         async with acquire(role="learner", user_id=write.learner_id) as conn:
@@ -1787,6 +1805,8 @@ async def save_session_evaluation(write: SessionEvaluationWrite) -> bool:
                   payload = EXCLUDED.payload,
                   error = EXCLUDED.error,
                   updated_at = now()
+                WHERE app.session_evaluation.status <> 'ready'
+                   OR EXCLUDED.status = 'ready'
                 """,
                 write.session_id,
                 write.status,
@@ -2731,22 +2751,45 @@ async def get_case_context(
     *,
     learner_id: str,
     persona_id: str,
+    case_id: str | None = None,
 ) -> CaseContext | None:
-    """Return the stable learner-persona case row, creating it when possible."""
+    """Return one owned continuation case without creating or mutating a row.
+
+    Legacy callers without ``case_id`` receive the most recently active case.
+    Fresh starts must never call this helper: their case row is created only after
+    the global active-session check inside ``create_session`` succeeds.
+    """
     try:
         get_pool()
         async with acquire(role="learner", user_id=learner_id) as conn:
-            row = await conn.fetchrow(
+            if case_id:
-                """
+                row = await conn.fetchrow(
-                INSERT INTO app.case_profile (persona_id, learner_id)
+                    """
-                VALUES ($1::uuid, $2::uuid)
+                    SELECT case_id, last_session_no
-                ON CONFLICT (persona_id, learner_id) DO UPDATE SET
+                    FROM app.case_profile
-                  updated_at = app.case_profile.updated_at
+                    WHERE case_id = $1::uuid
-                RETURNING case_id, last_session_no
+                      AND persona_id = $2::uuid
-                """,
+                      AND learner_id = $3::uuid
-                persona_id,
+                    """,
-                learner_id,
+                    case_id,
-            )
+                    persona_id,
+                    learner_id,
+                )
+            else:
+                row = await conn.fetchrow(
+                    """
+                    SELECT case_id, last_session_no
+                    FROM app.case_profile
+                    WHERE persona_id = $1::uuid
+                      AND learner_id = $2::uuid
+                    ORDER BY updated_at DESC, case_id DESC
+                    LIMIT 1
+                    """,
+                    persona_id,
+                    learner_id,
+                )
+        if row is None:
+            return None
         return CaseContext(
             case_id=str(row["case_id"]),
             last_session_no=int(row["last_session_no"] or 0),
@@ -2755,6 +2798,97 @@ async def get_case_context(
         return None
 
 
+async def list_case_summaries(
+    *,
+    learner_id: str,
+    persona_id: str,
+) -> list[dict[str, Any]]:
+    """Read complete case-scoped progress without the recent-session list cap.
+
+    Duration and turn counts deliberately use different CTEs so joining turns
+    cannot multiply the duration of a session. This is DB-only: returning a
+    runtime approximation would make the learner choose a continuation on false
+    progress data.
+    """
+    try:
+        get_pool()
+        async with acquire(role="learner", user_id=learner_id) as conn:
+            rows = await conn.fetch(
+                """
+                WITH scoped_cases AS (
+                  SELECT cp.case_id, cp.last_session_no, cp.updated_at
+                  FROM app.case_profile AS cp
+                  WHERE cp.learner_id = $1::uuid
+                    AND cp.persona_id = $2::uuid
+                ),
+                session_stats AS (
+                  SELECT
+                    s.case_id,
+                    count(*)::int AS total_sessions,
+                    count(*) FILTER (WHERE s.ended_at IS NOT NULL)::int AS completed_sessions,
+                    COALESCE(
+                      sum(
+                        GREATEST(
+                          0,
+                          EXTRACT(EPOCH FROM (COALESCE(s.ended_at, now()) - s.started_at))
+                        )
+                      ),
+                      0
+                    )::bigint AS total_duration_seconds,
+                    max(COALESCE(s.ended_at, s.started_at)) AS last_activity_at
+                  FROM app.sessions AS s
+                  JOIN scoped_cases AS c ON c.case_id = s.case_id
+                  GROUP BY s.case_id
+                ),
+                turn_stats AS (
+                  SELECT s.case_id, count(t.id)::int AS total_turns
+                  FROM app.sessions AS s
+                  JOIN scoped_cases AS c ON c.case_id = s.case_id
+                  LEFT JOIN app.turns AS t
+                    ON t.session_id = s.id
+                   AND 'counselor' = ANY(t.visible_to)
+                  GROUP BY s.case_id
+                ),
+                active_session AS (
+                  SELECT DISTINCT ON (s.case_id)
+                    s.case_id,
+                    s.id AS active_session_id,
+                    s.session_no AS active_session_no,
+                    s.started_at AS active_started_at
+                  FROM app.sessions AS s
+                  JOIN scoped_cases AS c ON c.case_id = s.case_id
+                  WHERE s.ended_at IS NULL
+                  ORDER BY s.case_id, s.started_at DESC, s.id DESC
+                )
+                SELECT
+                  c.case_id,
+                  c.last_session_no,
+                  COALESCE(ss.total_sessions, 0)::int AS total_sessions,
+                  COALESCE(ss.completed_sessions, 0)::int AS completed_sessions,
+                  COALESCE(ts.total_turns, 0)::int AS total_turns,
+                  COALESCE(ss.total_duration_seconds, 0)::bigint AS total_duration_seconds,
+                  active.active_session_id,
+                  active.active_session_no,
+                  active.active_started_at,
+                  ss.last_activity_at
+                FROM scoped_cases AS c
+                LEFT JOIN session_stats AS ss USING (case_id)
+                LEFT JOIN turn_stats AS ts USING (case_id)
+                LEFT JOIN active_session AS active USING (case_id)
+                ORDER BY COALESCE(ss.last_activity_at, c.updated_at) DESC, c.case_id DESC
+                """,
+                learner_id,
+                persona_id,
+            )
+        return [dict(row) for row in rows]
+    except Exception as exc:
+        logger.exception(
+            "case progress read failed",
+            extra={"learner_id": learner_id, "persona_id": persona_id},
+        )
+        raise CaseProgressUnavailableError("case_progress_unavailable") from exc
+
+
 async def create_session(
     *,
     learner_id: str,
@@ -2766,10 +2900,25 @@ async def create_session(
     persona_id: str | None = None,
     persona_version: int | None = None,
     case_id: str | None = None,
+    start_mode: Literal["continue", "fresh"] = "continue",
     goal_stages: list[str] | None = None,
     learner_feedback_enabled: bool = True,
+    locked_state_factory: (
+        Callable[[str, int], Awaitable[state_machine.SessionState]] | None
+    ) = None,
 ) -> InProcSession | None:
-    """Create a DB-backed session, returning None when DB persistence is unavailable."""
+    """Create a DB-backed session, returning None when DB persistence is unavailable.
+
+    A fresh start creates a new case only after the global learner-persona active
+    session guard passes. A continuation selects the requested owned case (or the
+    most recent legacy case) in the same transaction. ``locked_state_factory``
+    therefore sees either the committed selected case or the empty new case, never
+    a cross-case or in-between memory state.
+    """
+    if start_mode not in ("continue", "fresh"):
+        raise ValueError(f"unsupported session start mode: {start_mode}")
+    if start_mode == "fresh" and case_id is not None:
+        raise ValueError("fresh start must not select an existing case")
     try:
         get_pool()
         runtime_case_id = str(uuid.uuid4())
@@ -2777,21 +2926,86 @@ async def create_session(
         pinned_persona_version = persona_version or SEED_VERSION
         async with acquire(role="learner", user_id=learner_id) as conn:
             async with conn.transaction():
-                stable_case_id = case_id
+                # A case row cannot serialize two simultaneous first/fresh starts,
-                if stable_case_id is None:
+                # so lock the learner-persona scope before inspecting active rows.
+                await conn.execute(
+                    "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))",
+                    f"{learner_id}:{pinned_persona_id}",
+                )
+                active_row = await conn.fetchrow(
+                    """
+                    SELECT id
+                    FROM app.sessions
+                    WHERE learner_id = $1::uuid
+                      AND persona_id = $2::uuid
+                      AND ended_at IS NULL
+                    LIMIT 1
+                    FOR UPDATE
+                    """,
+                    learner_id,
+                    pinned_persona_id,
+                )
+                if active_row is not None:
+                    raise ActiveSessionExistsError(str(active_row["id"]))
+
+                stable_case_id: str
+                if start_mode == "fresh":
                     case_row = await conn.fetchrow(
                         """
                         INSERT INTO app.case_profile (persona_id, learner_id)
                         VALUES ($1::uuid, $2::uuid)
-                        ON CONFLICT (persona_id, learner_id) DO UPDATE SET
-                          updated_at = app.case_profile.updated_at
                         RETURNING case_id, last_session_no
                         """,
                         pinned_persona_id,
                         learner_id,
                     )
                     stable_case_id = str(case_row["case_id"])
+                    session_no = 1
+                elif case_id is not None:
+                    case_row = await conn.fetchrow(
+                        """
+                        SELECT case_id, last_session_no
+                        FROM app.case_profile
+                        WHERE case_id = $1::uuid
+                          AND learner_id = $2::uuid
+                          AND persona_id = $3::uuid
+                        FOR UPDATE
+                        """,
+                        case_id,
+                        learner_id,
+                        pinned_persona_id,
+                    )
+                    if case_row is None:
+                        raise CaseNotFoundError("case_not_found")
+                    stable_case_id = str(case_row["case_id"])
                     session_no = int(case_row["last_session_no"] or 0) + 1
+                else:
+                    case_row = await conn.fetchrow(
+                        """
+                        SELECT case_id, last_session_no
+                        FROM app.case_profile
+                        WHERE persona_id = $1::uuid
+                          AND learner_id = $2::uuid
+                        ORDER BY updated_at DESC, case_id DESC
+                        LIMIT 1
+                        FOR UPDATE
+                        """,
+                        pinned_persona_id,
+                        learner_id,
+                    )
+                    if case_row is None:
+                        case_row = await conn.fetchrow(
+                            """
+                            INSERT INTO app.case_profile (persona_id, learner_id)
+                            VALUES ($1::uuid, $2::uuid)
+                            RETURNING case_id, last_session_no
+                            """,
+                            pinned_persona_id,
+                            learner_id,
+                        )
+                    stable_case_id = str(case_row["case_id"])
+                    session_no = int(case_row["last_session_no"] or 0) + 1
+
                 counter_row = await conn.fetchrow(
                     """
                     UPDATE app.case_profile
@@ -2805,8 +3019,16 @@ async def create_session(
                     session_no,
                     learner_id,
                 )
-                if counter_row is not None:
+                if counter_row is None:
-                    session_no = int(counter_row["last_session_no"] or session_no)
+                    raise CaseNotFoundError("case_not_found")
+                session_no = int(counter_row["last_session_no"] or session_no)
+                persisted_state = state
+                if locked_state_factory is not None:
+                    persisted_state = await locked_state_factory(
+                        stable_case_id,
+                        session_no,
+                    )
+                    carry_rapport = persisted_state.rapport_credit
                 row = await conn.fetchrow(
                     """
                     INSERT INTO app.sessions (
@@ -2837,7 +3059,7 @@ async def create_session(
                     list(goal_stages or []),
                     learner_feedback_enabled,
                 )
-                await _upsert_state(conn, str(row["id"]), state)
+                await _upsert_state(conn, str(row["id"]), persisted_state)
         return InProcSession(
             session_id=str(row["id"]),
             case_id=str(row["case_id"] or row["runtime_case_id"] or row["id"]),
@@ -2845,7 +3067,7 @@ async def create_session(
             persona_code=card.code,
             theory_mode=theory_mode,
             persona=card,
-            state=state,
+            state=persisted_state,
             persona_id=pinned_persona_id,
             persona_version=pinned_persona_version,
             session_no=int(row["session_no"] or session_no),
@@ -2857,6 +3079,8 @@ async def create_session(
             goal_stages=list(goal_stages or []),
             learner_feedback_enabled=learner_feedback_enabled,
         )
+    except (ActiveSessionExistsError, CaseNotFoundError):
+        raise
     except Exception as exc:
         logger.exception(
             "durable session creation failed",
@@ -3218,6 +3442,21 @@ async def end_session(sess: InProcSession, carry: memory.CarryOver) -> bool:
         get_pool()
         summary_write = _build_session_summary_write(sess, carry)
         async with acquire(role="learner", user_id=sess.learner_id) as conn:
+            # start_session/create_session also locks this stable case before it
+            # inspects active rows. Taking the case lock first gives end/start a
+            # single lock order: the next session either sees S1 still active or
+            # sees its committed summary and carry-over, never an in-between row.
+            await conn.fetchval(
+                """
+                SELECT case_id
+                FROM app.case_profile
+                WHERE case_id = $1::uuid
+                  AND learner_id = $2::uuid
+                FOR NO KEY UPDATE
+                """,
+                sess.case_id,
+                sess.learner_id,
+            )
             await conn.execute(
                 """
                 UPDATE app.sessions
@@ -3463,3 +3702,12 @@ def _iso_dt(value: datetime | None) -> str:
     if value is None:
         return ""
     return value.astimezone(timezone.utc).isoformat()
+def _should_replace_evaluation_record(
+    existing: dict[str, Any] | None,
+    replacement: dict[str, Any],
+) -> bool:
+    """늦게 도착한 실패가 이미 확정된 ready 평가를 덮지 못하게 한다."""
+    return not (
+        str((existing or {}).get("status") or "") == "ready"
+        and str(replacement.get("status") or "") != "ready"
+    )

apps/api/app/session_read_model.py

@@ -25,6 +25,7 @@ from .stage_contract import (
     stage_label_or_none as stage_label_or_none,
 )
 from .store import InProcSession, TurnRecord
+from .session_evaluation_timeout import session_evaluation_stale_after_seconds
 
 WorksheetSpeaker = Literal["learner", "client"]
 WorksheetItemSpec = tuple[str, str, list[str], WorksheetSpeaker | None]
@@ -32,7 +33,6 @@ WorksheetSectionSpec = tuple[str, str, list[WorksheetItemSpec]]
 
 LEARNER_VISIBLE_AI_ROLE = "counselor"
 KOREA_STANDARD_TIME = timezone(timedelta(hours=9), name="KST")
-MISSING_SESSION_EVALUATION_GRACE_SECONDS = 30.0
 MISSING_SESSION_EVALUATION_ERROR = (
     "회기말 평가가 제한 시간 이후에도 저장되지 않았습니다. AI 평가 재시도가 필요합니다."
 )
@@ -44,6 +44,8 @@ FIRST_SESSION_CHECKLIST_VERSION = "first-session-rapport-open-question.v1"
 
 class LearnerSessionSummary(BaseModel):
     session_id: str
+    # 같은 내담자도 새 사례에서 다시 S1이 될 수 있으므로, 이력 행의 연속체 경계를 함께 준다.
+    case_id: str = ""
     persona_code: str
     persona_name: str
     session_no: int
@@ -65,6 +67,44 @@ class LearnerSessionsResponse(BaseModel):
     sessions: list[LearnerSessionSummary] = Field(default_factory=list)
 
 
+class CaseProgressStats(BaseModel):
+    """한 사례 안에서만 누적한 learner-visible 진행 수치."""
+
+    total_sessions: int = 0
+    completed_sessions: int = 0
+    total_turns: int = 0
+    total_duration_seconds: int = 0
+    active_session_id: str | None = None
+    active_session_no: int | None = None
+    active_started_at: str | None = None
+    last_activity_at: str | None = None
+
+
+class LearnerCaseSummary(BaseModel):
+    case_id: str
+    persona_code: str
+    persona_name: str
+    last_session_no: int = 0
+    progress: CaseProgressStats = Field(default_factory=CaseProgressStats)
+
+
+class LearnerCaseListResponse(BaseModel):
+    # case별 누적 수치는 DB 전체 집계여야 한다. runtime cache 추정값은 반환하지 않는다.
+    source: Literal["database"] = "database"
+    cases: list[LearnerCaseSummary] = Field(default_factory=list)
+
+
+class CaseMemoryPreview(BaseModel):
+    """접힌 learner UI에만 쓰는 최소·마스킹된 사례 기억 투영."""
+
+    case_id: str
+    memory_available: bool = False
+    case_digest: str | None = None
+    latest_session_digest: str | None = None
+    open_threads: list[str] = Field(default_factory=list)
+    pinned_facts: list[str] = Field(default_factory=list)
+
+
 class LearnerDashboardOverview(BaseModel):
     total_sessions: int = 0
     completed_sessions: int = 0
@@ -472,6 +512,21 @@ class SessionTeacherReviewStatus(BaseModel):
     worksheetReviewedAt: str | None = None
 
 
+class ReviewEvaluationFailure(BaseModel):
+    """교수자용 deep-loop 실패 분류. 원문 예외·축어록은 절대 응답에 넣지 않는다."""
+
+    code: Literal[
+        "timeout",
+        "engine_unavailable",
+        "legacy_argv_limit",
+        "prompt_too_large",
+        "invalid_structured_output",
+        "missing_evaluation",
+        "unknown",
+    ]
+    retryable: bool
+
+
 class SessionReviewResponse(BaseModel):
     session_id: str
     sessionNo: int = Field(ge=1)
@@ -501,6 +556,7 @@ class SessionReviewResponse(BaseModel):
     pdfExportUrl: Optional[str] = None
     degraded: bool = True
     reviewReady: bool = False
+    evaluationFailure: ReviewEvaluationFailure | None = None
     learnerFeedbackEnabled: bool = True
     teacherReview: SessionTeacherReviewStatus | None = None
 
@@ -561,6 +617,7 @@ def learner_summary(
     client_turns = sum(1 for turn in turns if turn.speaker == "client")
     return LearnerSessionSummary(
         session_id=sess.session_id,
+        case_id=sess.case_id,
         persona_code=sess.persona_code,
         persona_name=sess.persona.display_name,
         session_no=sess.session_no,
@@ -1121,6 +1178,12 @@ def _review_summary_from_evaluation(
         return fallback
     status = str(evaluation_record.get("status") or "")
     if status != "ready":
+        failure = teacher_evaluation_failure(evaluation_record)
+        if failure is not None and not failure.retryable:
+            return (
+                "저장된 축어록은 확인했지만 deep-loop 평가 AI 산출물을 표시하지 못했습니다. "
+                "평가 입력 경로를 조정한 뒤 다시 생성해야 합니다."
+            )
         return (
             "저장된 축어록은 확인했지만 deep-loop 평가 AI 산출물을 표시하지 못했습니다. "
             "AI 평가 재시도가 필요합니다."
@@ -1133,9 +1196,56 @@ def _review_summary_from_evaluation(
     return prefix + (details if details else "아래 코칭 항목은 저장된 축어록과 평가 AI 결과를 기준으로 합니다.")
 
 
-def _session_evaluation_timeout_seconds() -> float:
+def teacher_evaluation_failure(
-    configured = float(settings.session_evaluation_timeout or settings.engine_timeout)
+    evaluation_record: dict[str, object] | None,
-    return max(1.0, configured)
+) -> ReviewEvaluationFailure | None:
+    """오류 원문을 노출하지 않고 교수자가 취할 다음 행동만 분류한다."""
+    if not evaluation_record or str(evaluation_record.get("status") or "") != "error":
+        return None
+    raw_error = str(evaluation_record.get("error") or "").strip().lower()
+    if raw_error == MISSING_SESSION_EVALUATION_ERROR.lower() or "저장되지 않았습니다" in raw_error:
+        return ReviewEvaluationFailure(code="missing_evaluation", retryable=True)
+    if (
+        ("24,000" in raw_error or "명령줄 안전 한도" in raw_error)
+        and "agy" in raw_error
+    ):
+        # 2026-08 이전 Windows Agy argv 경로의 durable 오류다. 현재 stdin 경로로는
+        # 재평가할 수 있으므로, 같은 오류처럼 보이더라도 복구 재시도를 열어 둔다.
+        return ReviewEvaluationFailure(code="legacy_argv_limit", retryable=True)
+    if "argv" in raw_error or "24000" in raw_error or "prompt too large" in raw_error:
+        return ReviewEvaluationFailure(code="prompt_too_large", retryable=False)
+    if "timeout" in raw_error or "timed out" in raw_error:
+        return ReviewEvaluationFailure(code="timeout", retryable=True)
+    if "no_structured_output" in raw_error or "parse_error" in raw_error:
+        return ReviewEvaluationFailure(code="invalid_structured_output", retryable=True)
+    if (
+        "engine_error" in raw_error
+        or "engine unavailable" in raw_error
+        or "transport error" in raw_error
+    ):
+        return ReviewEvaluationFailure(code="engine_unavailable", retryable=True)
+    return ReviewEvaluationFailure(code="unknown", retryable=True)
+
+
+def teacher_evaluation_failure_message(
+    failure: ReviewEvaluationFailure | None,
+) -> str | None:
+    """교수자 목록 API에서 provider 원문 대신 사용할 최소 행동 안내."""
+    if failure is None:
+        return None
+    if failure.code == "timeout":
+        return "AI 평가가 제한 시간 안에 끝나지 않았습니다. 다시 시도할 수 있습니다."
+    if failure.code == "engine_unavailable":
+        return "평가 엔진에 일시적으로 연결하지 못했습니다. 다시 시도할 수 있습니다."
+    if failure.code == "legacy_argv_limit":
+        return "이전 Windows 입력 한도에 걸린 평가입니다. 현재 입력 경로로 다시 시도할 수 있습니다."
+    if failure.code == "prompt_too_large":
+        return "평가 입력이 허용 크기를 넘어섰습니다. 입력 경로 조정이 필요합니다."
+    if failure.code == "invalid_structured_output":
+        return "평가 결과 형식이 검증되지 않았습니다. 다시 시도할 수 있습니다."
+    if failure.code == "missing_evaluation":
+        return "회기말 평가 기록이 아직 저장되지 않았습니다. 다시 시도할 수 있습니다."
+    return "AI 평가를 완료하지 못했습니다. 최신 상태를 확인해 주세요."
 
 
 def _missing_session_evaluation_record(
@@ -1146,7 +1256,7 @@ def _missing_session_evaluation_record(
 ) -> dict[str, object] | None:
     if not sess.ended or not has_visible_turns or sess.ended_at is None:
         return None
-    stale_after = _session_evaluation_timeout_seconds() + MISSING_SESSION_EVALUATION_GRACE_SECONDS
+    stale_after = session_evaluation_stale_after_seconds()
     if now_ts - sess.ended_at < stale_after:
         return None
     return {
@@ -1903,6 +2013,11 @@ def build_session_review(read_input: SessionReviewReadInput) -> SessionReviewRes
         else ""
     )
     evaluation_ready = not feedback_hidden and evaluation_status == "ready"
+    evaluation_failure = (
+        teacher_evaluation_failure(evaluation_record)
+        if read_input.include_teacher_review and not feedback_hidden
+        else None
+    )
 
     first_turn_ts = visible_turns[0].created_at if visible_turns else sess.created_at
     turns: list[ReviewTurn] = []
@@ -2083,6 +2198,7 @@ def build_session_review(read_input: SessionReviewReadInput) -> SessionReviewRes
         pdfExportUrl=None,
         degraded=review_degraded,
         reviewReady=evaluation_ready,
+        evaluationFailure=evaluation_failure,
         learnerFeedbackEnabled=read_input.learner_feedback_enabled,
         teacherReview=teacher_review,
     )

apps/api/app/store.py

@@ -104,16 +104,17 @@ class SessionStore:
         state: SessionState,
         persona_id: str | None = None,
         persona_version: int | None = None,
+        case_id: str | None = None,
         session_no: int = 1,
         carry_rapport: float = 0.0,
         goal_stages: list[str] | None = None,
         learner_feedback_enabled: bool = True,
     ) -> InProcSession:
         session_id = uuid4().hex
-        case_id = uuid4().hex
+        stable_case_id = case_id or uuid4().hex
         s = InProcSession(
             session_id=session_id,
-            case_id=case_id,
+            case_id=stable_case_id,
             learner_id=learner_id,
             persona_code=persona.code,
             theory_mode=theory_mode,
@@ -138,6 +139,23 @@ class SessionStore:
     def list(self) -> list[InProcSession]:
         return list(self._sessions.values())
 
+    def find_active(
+        self,
+        *,
+        learner_id: str,
+        persona_id: str | None,
+        persona_code: str,
+    ) -> Optional[InProcSession]:
+        """Return the active session for one learner-persona continuity case."""
+        for session in self._sessions.values():
+            if session.ended or session.learner_id != learner_id:
+                continue
+            if persona_id is not None and session.persona_id == persona_id:
+                return session
+            if session.persona_id is None and session.persona_code == persona_code:
+                return session
+        return None
+
     def append_turn(self, session_id: str, turn: TurnRecord) -> None:
         s = self._sessions.get(session_id)
         if s is not None:

apps/api/app/test_eval_routes.py

@@ -9,6 +9,7 @@ from unittest.mock import AsyncMock, patch
 
 from fastapi import HTTPException
 
+from .config import settings
 from .deps import Principal, Role
 from .services import evaluator
 
@@ -44,7 +45,11 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
                 )
 
         self.assertEqual(raised.exception.status_code, 502)
-        self.assertEqual(raised.exception.detail, "no_structured_output")
+        self.assertEqual(
+            raised.exception.detail,
+            "AI 평가를 완료하지 못했습니다. 최신 평가 상태를 확인해 주세요.",
+        )
+        self.assertNotIn("no_structured_output", raised.exception.detail)
         saved_write = save_evaluation.await_args.args[0]
         self.assertEqual(saved_write.status, "error")
         self.assertEqual(saved_write.error, "no_structured_output")
@@ -78,6 +83,39 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(raised.exception.status_code, 503)
         self.assertIn("could not be saved", raised.exception.detail)
 
+    async def test_reevaluate_session_redacts_engine_exception_from_http_response(self) -> None:
+        session_id = "00000000-0000-0000-0000-00000000e222"
+        sess = _session_stub()
+
+        with (
+            patch.object(eval_routes, "_load_session_or_404", AsyncMock(return_value=sess)),
+            patch.object(
+                eval_routes.evaluator,
+                "evaluate_session",
+                AsyncMock(side_effect=eval_routes.EngineError("provider body: [NAME]")),
+            ),
+            patch.object(
+                eval_routes.session_persistence,
+                "save_session_evaluation",
+                AsyncMock(return_value=True),
+            ) as save_evaluation,
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await eval_routes.reevaluate_session(
+                    session_id,
+                    eval_routes.ReevaluateRequest(scope="session_end"),
+                    _teacher_principal(),
+                )
+
+        self.assertEqual(raised.exception.status_code, 503)
+        self.assertEqual(
+            raised.exception.detail,
+            "AI 평가를 완료하지 못했습니다. 최신 평가 상태를 확인해 주세요.",
+        )
+        self.assertNotIn("[NAME]", raised.exception.detail)
+        saved_write = save_evaluation.await_args.args[0]
+        self.assertIn("provider body: [NAME]", saved_write.error)
+
     async def test_reevaluate_session_prefers_session_theory_mode(self) -> None:
         session_id = "00000000-0000-0000-0000-00000000e222"
         sess = _session_stub(theory_mode="cbt", theory_target=["humanistic"])
@@ -89,6 +127,7 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
         )
 
         with (
+            patch.object(settings, "session_evaluation_timeout", 300.0),
             patch.object(eval_routes, "_load_session_or_404", AsyncMock(return_value=sess)),
             patch.object(
                 eval_routes.evaluator,
@@ -108,6 +147,7 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
             )
 
         self.assertEqual(evaluate_session.await_args.kwargs["theory_mode"], "cbt")
+        self.assertEqual(evaluate_session.await_args.kwargs["timeout"], 330.0)
 
     async def test_reevaluate_session_uses_one_based_masked_turn_seq(self) -> None:
         session_id = "00000000-0000-0000-0000-00000000e222"
@@ -277,7 +317,10 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
                 )
 
         self.assertEqual(raised.exception.status_code, 503)
-        self.assertEqual(raised.exception.detail, "engine_error")
+        self.assertEqual(
+            raised.exception.detail,
+            "AI 평가를 완료하지 못했습니다. 최신 평가 상태를 확인해 주세요.",
+        )
         saved_payload = replace_turn_evaluation.await_args.kwargs["evaluation"]
         self.assertEqual(saved_payload["error"], "engine_error")
 
@@ -289,7 +332,10 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
             "source": "engine",
             "scope": "session_end",
             "stage": "정리",
-            "payload": {},
+            "payload": {
+                "error": "session evaluation timeout after 45s",
+                "distribution": {"total": 0},
+            },
             "error": "session evaluation timeout after 45s",
             "updated_at": "2026-07-01T00:00:00Z",
         }
@@ -305,9 +351,16 @@ class ReevaluateSessionRouteTest(unittest.IsolatedAsyncioTestCase):
             response = await eval_routes.get_session_evaluation(session_id, _teacher_principal())
 
         self.assertEqual(response.status, "error")
-        self.assertEqual(response.error, "session evaluation timeout after 45s")
+        self.assertEqual(
+            response.error,
+            "AI 평가를 완료하지 못했습니다. 최신 평가 상태를 확인해 주세요.",
+        )
+        self.assertIsNotNone(response.failure)
+        assert response.failure is not None
+        self.assertEqual(response.failure.code, "timeout")
+        self.assertNotIn("session evaluation timeout after 45s", response.model_dump_json())
         self.assertTrue(response.durable)
-        self.assertEqual(response.deep, {})
+        self.assertEqual(response.deep, {"distribution": {"total": 0}})
 
 
 def _session_stub(

apps/api/app/test_evaluation_persistence.py

@@ -429,6 +429,46 @@ class EvaluationPersistenceMappingTest(unittest.IsolatedAsyncioTestCase):
         self.assertNotIn("010-1234-5678", blob)
         self.assertIn("[NAME]", blob)
         self.assertIn("[PHONE]", blob)
+        self.assertIn(
+            "WHERE app.session_evaluation.status <> 'ready'",
+            conn.executed[0][0],
+        )
+
+    def test_late_error_never_replaces_ready_evaluation_record(self) -> None:
+        ready = {"status": "ready", "payload": {"strengths": ["완료"]}}
+        error = {"status": "error", "error": "late timeout"}
+
+        self.assertFalse(
+            session_persistence._should_replace_evaluation_record(ready, error)
+        )
+        self.assertTrue(
+            session_persistence._should_replace_evaluation_record(error, ready)
+        )
+        self.assertTrue(
+            session_persistence._should_replace_evaluation_record(ready, ready)
+        )
+
+    async def test_fallback_cache_keeps_ready_result_when_late_error_arrives(self) -> None:
+        session_id = "11111111-1111-1111-1111-111111111111"
+        session_persistence._EVALUATION_CACHE[session_id] = {"status": "ready"}
+        self.addCleanup(session_persistence._EVALUATION_CACHE.pop, session_id, None)
+        write = session_persistence.SessionEvaluationWrite.from_error(
+            session_id=session_id,
+            learner_id="22222222-2222-2222-2222-222222222222",
+            scope="session_end",
+            stage="정리",
+            error="late timeout",
+        )
+
+        with (
+            patch.object(session_persistence, "runtime_fallback_allowed", return_value=True),
+            patch.object(session_persistence, "get_pool", side_effect=RuntimeError("offline")),
+            patch.object(session_persistence, "require_runtime_fallback_allowed"),
+        ):
+            saved = await session_persistence.save_session_evaluation(write)
+
+        self.assertFalse(saved)
+        self.assertEqual(session_persistence._EVALUATION_CACHE[session_id]["status"], "ready")
 
     def test_rebuild_turn_evaluation_restores_review_shape(self) -> None:
         rebuilt = session_persistence._rebuild_turn_evaluations(

apps/api/app/test_evaluator_model_routing.py

@@ -30,9 +30,11 @@ def _turn_context() -> orchestrator.TurnContext:
 class CaptureEvaluatorEngine:
     def __init__(self) -> None:
         self.requests: list[Any] = []
+        self.timeouts: list[float | None] = []
 
-    async def generate(self, req: Any) -> GenerateResponse:
+    async def generate(self, req: Any, *, timeout: float | None = None) -> GenerateResponse:
         self.requests.append(req)
+        self.timeouts.append(timeout)
         loop = req.metadata.get("loop")
         if loop == "fast":
             structured = {
@@ -61,7 +63,7 @@ class FailingEvaluatorEngine:
     def __init__(self) -> None:
         self.requests: list[Any] = []
 
-    async def generate(self, req: Any) -> GenerateResponse:
+    async def generate(self, req: Any, *, timeout: float | None = None) -> GenerateResponse:
         self.requests.append(req)
         raise EngineError("synthetic evaluator failure")
 
@@ -122,6 +124,23 @@ class EvaluatorModelRoutingTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(engine.requests[0].ai_role, "evaluator")
         self.assertEqual(engine.requests[0].model, "cheap-deep")
 
+    async def test_deep_evaluator_forwards_explicit_timeout(self) -> None:
+        engine = CaptureEvaluatorEngine()
+
+        result = await evaluator.evaluate_session(
+            session_id="evaluator-model-session",
+            stage="라포",
+            masked_turns=[
+                {"speaker": "counselor", "text": "천천히 이야기해줘도 괜찮아요."},
+                {"speaker": "client", "text": "잘 모르겠어요."},
+            ],
+            engine=engine,  # type: ignore[arg-type]
+            timeout=300.0,
+        )
+
+        self.assertIsNone(result.error)
+        self.assertEqual(engine.timeouts, [300.0])
+
     async def test_blank_model_settings_keep_gateway_default_routing(self) -> None:
         settings.evaluator_fast_model = "   "
         settings.evaluator_deep_model = ""

apps/api/app/test_notifications.py

@@ -9,6 +9,7 @@ from typing import Any
 from unittest.mock import AsyncMock, patch
 
 from .config import settings
+from . import session_evaluation_timeout
 from .routes import sessions
 from .services import notifications, state_machine
 from .services.persona import P1
@@ -209,6 +210,39 @@ class NotificationTriggerTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(saved_write.status, "error")
         self.assertEqual(saved_write.error, "session evaluation timeout after 1s")
 
+    async def test_session_evaluation_forwards_transport_and_audit_budget_after_deep_deadline(self) -> None:
+        sess = _ended_session()
+        result = sessions.evaluator.SessionEvaluation(
+            session_id=sess.session_id,
+            stage=sess.state.stage.value,
+            scope="session_end",
+            turns_evaluated=len(sess.turns),
+        )
+
+        with (
+            patched_settings(session_evaluation_timeout=300.0, db_command_timeout=30.0),
+            patch.object(
+                sessions.evaluator,
+                "evaluate_session",
+                AsyncMock(return_value=result),
+            ) as evaluate_session,
+            patch.object(
+                sessions.session_persistence,
+                "save_session_evaluation",
+                AsyncMock(return_value=True),
+            ),
+            patch.object(
+                sessions.notifications,
+                "enqueue_session_review_ready",
+                AsyncMock(),
+            ),
+            patch.object(sessions.asyncio, "wait_for", wraps=asyncio.wait_for) as wait_for,
+        ):
+            await sessions._generate_and_save_session_evaluation(sess)
+
+        self.assertEqual(evaluate_session.await_args.kwargs["timeout"], 330.0)
+        self.assertEqual(wait_for.await_args.kwargs["timeout"], 365.0)
+
     async def test_scheduled_session_evaluation_task_crash_is_observed(self) -> None:
         sess = _ended_session()
 
@@ -245,11 +279,30 @@ class NotificationTriggerTest(unittest.IsolatedAsyncioTestCase):
             ) as schedule_evaluation,
         ):
             scheduled = await sessions.recover_missing_session_evaluations(limit=2)
+            expected_stale_after = session_evaluation_timeout.session_evaluation_stale_after_seconds()
 
         self.assertEqual(scheduled, 1)
-        list_missing.assert_awaited_once_with(older_than_seconds=40.0, limit=2)
+        list_missing.assert_awaited_once_with(
+            older_than_seconds=expected_stale_after,
+            limit=2,
+        )
         schedule_evaluation.assert_called_once_with(sess)
 
+    def test_stale_deadline_follows_outer_completion_and_db_save_budget(self) -> None:
+        with patched_settings(session_evaluation_timeout=300.0, db_command_timeout=30.0):
+            self.assertEqual(
+                session_evaluation_timeout.session_evaluation_transport_timeout_seconds(),
+                330.0,
+            )
+            self.assertEqual(
+                session_evaluation_timeout.session_evaluation_outer_timeout_seconds(),
+                365.0,
+            )
+            self.assertEqual(
+                session_evaluation_timeout.session_evaluation_stale_after_seconds(),
+                400.0,
+            )
+
     async def test_missing_session_evaluation_recovery_requires_durable_store(self) -> None:
         sess = _ended_session()
 

apps/api/app/test_persona_session_contract.py

@@ -119,10 +119,14 @@ class _ContractConnection:
             ]
             matches.sort(key=lambda row: int(row["version"]), reverse=True)
             return matches[0] if matches else None
+        if "FROM app.case_profile" in query and "FOR UPDATE" in query:
+            return None
         if "INSERT INTO app.case_profile" in query:
             return {"case_id": self.case_id, "last_session_no": 0}
         if "UPDATE app.case_profile" in query:
             return {"last_session_no": int(args[1])}
+        if "FROM app.sessions" in query and "ended_at IS NULL" in query:
+            return None
         if "INSERT INTO app.sessions" in query:
             self.session_insert_args = args
             return {
@@ -140,7 +144,11 @@ class _ContractConnection:
         raise AssertionError(f"unexpected fetchrow: {query}")
 
     async def execute(self, query: str, *args: Any) -> str:
-        if "audit.audit_log" in query or "app.session_state" in query:
+        if (
+            "audit.audit_log" in query
+            or "app.session_state" in query
+            or "pg_advisory_xact_lock" in query
+        ):
             return "INSERT 0 1"
         raise AssertionError(f"unexpected execute: {query}")
 

apps/api/app/test_session_case_api.py

@@ -0,0 +1,357 @@
+"""새 사례/이어가기 API의 learner 경계 계약 테스트."""
+
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, patch
+from uuid import UUID
+
+from fastapi import HTTPException
+
+from .deps import Principal, Role
+from .routes import sessions
+from .services import persona as persona_service
+from .store import InProcSession, store
+
+
+LEARNER_ID = "00000000-0000-0000-0000-000000000741"
+PERSONA_ID = "00000000-0000-0000-0000-000000000742"
+CASE_ID = "00000000-0000-0000-0000-00000000ca5e"
+ACTIVE_SESSION_ID = "00000000-0000-0000-0000-000000000743"
+
+
+def _principal() -> Principal:
+    return Principal(
+        user_id=LEARNER_ID,
+        role=Role.LEARNER,
+        cohort_ids=[],
+        email="case-api-test@hs.ac.kr",
+        display_name="Case API Test",
+        consent_at=1.0,
+        profile_completed_at=1.0,
+    )
+
+
+def _catalog_persona() -> SimpleNamespace:
+    return SimpleNamespace(
+        card=persona_service.P1,
+        persona_id=PERSONA_ID,
+        version=3,
+        degraded=False,
+    )
+
+
+class _Acquire:
+    def __init__(self, conn: "_MemoryConnection") -> None:
+        self.conn = conn
+
+    async def __aenter__(self) -> "_MemoryConnection":
+        return self.conn
+
+    async def __aexit__(
+        self,
+        exc_type: object,
+        exc: object,
+        tb: object,
+    ) -> bool:
+        return False
+
+
+class _MemoryConnection:
+    """foldout이 필요한 최소 learner-safe 행만 내는 DB 대역."""
+
+    def __init__(
+        self,
+        *,
+        case_row: dict[str, Any] | None,
+        summary_row: dict[str, Any] | None = None,
+        fact_rows: list[dict[str, Any]] | None = None,
+    ) -> None:
+        self.case_row = case_row
+        self.summary_row = summary_row
+        self.fact_rows = fact_rows or []
+        self.calls: list[tuple[str, tuple[Any, ...]]] = []
+
+    async def fetchrow(self, query: str, *args: Any) -> dict[str, Any] | None:
+        self.calls.append((query, args))
+        if "FROM app.case_profile" in query:
+            return self.case_row
+        if "FROM app.session_summary AS ss" in query:
+            return self.summary_row
+        raise AssertionError(f"unexpected fetchrow: {query}")
+
+    async def fetch(self, query: str, *args: Any) -> list[dict[str, Any]]:
+        self.calls.append((query, args))
+        if "FROM app.pinned_fact AS pf" not in query:
+            raise AssertionError(f"unexpected fetch: {query}")
+        return self.fact_rows
+
+
+class LearnerCaseListApiTest(unittest.IsolatedAsyncioTestCase):
+    async def test_case_list_projects_one_case_with_complete_case_local_stats(self) -> None:
+        principal = _principal()
+        catalog_persona = _catalog_persona()
+        active_started_at = datetime(2026, 8, 31, 9, 15, tzinfo=timezone.utc)
+        last_activity_at = datetime(2026, 8, 31, 10, 45, tzinfo=timezone.utc)
+        rows = [
+            {
+                "case_id": CASE_ID,
+                "last_session_no": 4,
+                "total_sessions": 4,
+                "completed_sessions": 3,
+                "total_turns": 18,
+                "total_duration_seconds": 5_400,
+                "active_session_id": ACTIVE_SESSION_ID,
+                "active_session_no": 4,
+                "active_started_at": active_started_at,
+                "last_activity_at": last_activity_at,
+            }
+        ]
+        list_summaries = AsyncMock(return_value=rows)
+
+        with (
+            patch.object(
+                sessions,
+                "get_catalog_persona",
+                AsyncMock(return_value=catalog_persona),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "list_case_summaries",
+                list_summaries,
+            ),
+        ):
+            response = await sessions.list_learner_cases(
+                persona_code=persona_service.P1.code,
+                principal=principal,
+            )
+
+        list_summaries.assert_awaited_once_with(
+            learner_id=principal.user_id,
+            persona_id=PERSONA_ID,
+        )
+        self.assertEqual(response.source, "database")
+        self.assertEqual(len(response.cases), 1)
+        case = response.cases[0]
+        self.assertEqual(case.case_id, CASE_ID)
+        self.assertEqual(case.persona_code, persona_service.P1.code)
+        self.assertEqual(case.persona_name, persona_service.P1.display_name)
+        self.assertEqual(case.last_session_no, 4)
+        self.assertEqual(case.progress.total_sessions, 4)
+        self.assertEqual(case.progress.completed_sessions, 3)
+        self.assertEqual(case.progress.total_turns, 18)
+        self.assertEqual(case.progress.total_duration_seconds, 5_400)
+        self.assertEqual(case.progress.active_session_id, ACTIVE_SESSION_ID)
+        self.assertEqual(case.progress.active_session_no, 4)
+        self.assertEqual(case.progress.active_started_at, active_started_at.isoformat())
+        self.assertEqual(case.progress.last_activity_at, last_activity_at.isoformat())
+
+    async def test_case_list_fails_closed_when_database_progress_is_unavailable(self) -> None:
+        principal = _principal()
+        with (
+            patch.object(
+                sessions,
+                "get_catalog_persona",
+                AsyncMock(return_value=_catalog_persona()),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "list_case_summaries",
+                AsyncMock(
+                    side_effect=sessions.session_persistence.CaseProgressUnavailableError(
+                        "database unavailable"
+                    )
+                ),
+            ),
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await sessions.list_learner_cases(
+                    persona_code=persona_service.P1.code,
+                    principal=principal,
+                )
+
+        self.assertEqual(raised.exception.status_code, 503)
+        self.assertEqual(raised.exception.detail, "case_progress_unavailable")
+
+
+class LearnerCaseMemoryPreviewApiTest(unittest.IsolatedAsyncioTestCase):
+    async def test_memory_preview_is_owner_scoped_and_never_selects_raw_session_data(self) -> None:
+        principal = _principal()
+        conn = _MemoryConnection(
+            case_row={"case_digest": "사례 요약 " + "가" * 700},
+            summary_row={
+                "digest": "직전 회기 요약 " + "나" * 700,
+                "open_threads": ["남은 주제 " + "다" * 200, "다음 질문"],
+            },
+            fact_rows=[{"value": "기억 항목 " + "라" * 200}],
+        )
+
+        with patch.object(
+            sessions.db,
+            "acquire",
+            return_value=_Acquire(conn),
+        ) as acquire:
+            response = await sessions.get_learner_case_memory_preview(
+                case_id=UUID(CASE_ID),
+                principal=principal,
+            )
+
+        acquire.assert_called_once_with(role="learner", user_id=principal.user_id)
+        self.assertEqual(response.case_id, CASE_ID)
+        self.assertTrue(response.memory_available)
+        self.assertEqual(len(response.case_digest or ""), 600)
+        self.assertEqual(len(response.latest_session_digest or ""), 600)
+        self.assertEqual(len(response.open_threads[0]), 160)
+        self.assertTrue(response.open_threads[0].endswith("…"))
+        self.assertEqual(response.open_threads[1], "다음 질문")
+        self.assertEqual(len(response.pinned_facts[0]), 160)
+        self.assertTrue(response.pinned_facts[0].endswith("…"))
+
+        queried_sql = "\n".join(query for query, _ in conn.calls).lower()
+        self.assertIn("and learner_id = $2::uuid", queried_sql)
+        self.assertIn("and s.learner_id = $2::uuid", queried_sql)
+        self.assertIn("and cp.learner_id = $2::uuid", queried_sql)
+        self.assertNotIn("app.turns", queried_sql)
+        self.assertNotIn("transcript", queried_sql)
+        self.assertNotIn("evaluator", queried_sql)
+        self.assertNotIn("end_state", queried_sql)
+        self.assertNotIn(" text", queried_sql)
+        self.assertEqual({args[1] for _, args in conn.calls}, {principal.user_id})
+
+        response_payload = response.model_dump()
+        self.assertNotIn("transcript", response_payload)
+        self.assertNotIn("evaluator", response_payload)
+        self.assertNotIn("end_state", response_payload)
+        self.assertEqual(
+            set(response_payload),
+            {
+                "case_id",
+                "memory_available",
+                "case_digest",
+                "latest_session_digest",
+                "open_threads",
+                "pinned_facts",
+            },
+        )
+
+    async def test_memory_preview_returns_404_before_loading_unowned_case_details(self) -> None:
+        principal = _principal()
+        conn = _MemoryConnection(case_row=None)
+
+        with patch.object(
+            sessions.db,
+            "acquire",
+            return_value=_Acquire(conn),
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await sessions.get_learner_case_memory_preview(
+                    case_id=UUID(CASE_ID),
+                    principal=principal,
+                )
+
+        self.assertEqual(raised.exception.status_code, 404)
+        self.assertEqual(raised.exception.detail, "case_not_found")
+        self.assertEqual(len(conn.calls), 1)
+        self.assertIn("FROM app.case_profile", conn.calls[0][0])
+        self.assertEqual(conn.calls[0][1], (CASE_ID, principal.user_id))
+
+
+class SessionStartCaseModeApiTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        store._sessions.clear()
+        sessions._RECALL_CACHE.clear()
+        sessions._KB_CUES_CACHE.clear()
+
+    async def asyncTearDown(self) -> None:
+        store._sessions.clear()
+        sessions._RECALL_CACHE.clear()
+        sessions._KB_CUES_CACHE.clear()
+
+    async def test_fresh_start_rejects_selected_existing_case(self) -> None:
+        principal = _principal()
+        create_session = AsyncMock()
+        with (
+            patch.object(
+                sessions,
+                "get_catalog_persona",
+                AsyncMock(return_value=_catalog_persona()),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "create_session",
+                create_session,
+            ),
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await sessions.start_session(
+                    sessions.SessionStartRequest(
+                        persona_code=persona_service.P1.code,
+                        start_mode="fresh",
+                        case_id=UUID(CASE_ID),
+                    ),
+                    principal,
+                )
+
+        self.assertEqual(raised.exception.status_code, 422)
+        self.assertEqual(raised.exception.detail, "fresh_start_must_not_select_case")
+        create_session.assert_not_awaited()
+
+    async def test_continue_start_forwards_the_selected_case_to_durable_creation(self) -> None:
+        principal = _principal()
+        catalog_persona = _catalog_persona()
+        captured: dict[str, Any] = {}
+
+        async def fake_create_session(**kwargs: Any) -> InProcSession:
+            captured.update(kwargs)
+            return InProcSession(
+                session_id="selected-case-session",
+                case_id=str(kwargs["case_id"]),
+                learner_id=principal.user_id,
+                persona_code=persona_service.P1.code,
+                theory_mode=str(kwargs["theory_mode"]),
+                persona=persona_service.P1,
+                state=kwargs["state"],
+                persona_id=PERSONA_ID,
+                persona_version=3,
+                session_no=7,
+                prev_rapport_credit=float(kwargs["carry_rapport"]),
+            )
+
+        def close_background(coro: Any) -> SimpleNamespace:
+            coro.close()
+            return SimpleNamespace()
+
+        with (
+            patch.object(
+                sessions,
+                "get_catalog_persona",
+                AsyncMock(return_value=catalog_persona),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "create_session",
+                AsyncMock(side_effect=fake_create_session),
+            ),
+            patch.object(sessions.asyncio, "create_task", close_background),
+        ):
+            response = await sessions.start_session(
+                sessions.SessionStartRequest(
+                    persona_code=persona_service.P1.code,
+                    start_mode="continue",
+                    case_id=UUID(CASE_ID),
+                ),
+                principal,
+            )
+
+        self.assertEqual(captured["case_id"], CASE_ID)
+        self.assertEqual(captured["start_mode"], "continue")
+        self.assertEqual(response.case_id, CASE_ID)
+        self.assertEqual(response.session_no, 7)
+        self.assertEqual(response.start_mode, "continue")
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/api/app/test_session_continuity_guard.py

@@ -0,0 +1,326 @@
+"""같은 내담자 케이스의 회기 순차성 회귀 테스트."""
+
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from fastapi import HTTPException
+
+from . import session_persistence
+from .deps import Principal, Role
+from .routes import sessions
+from .services import memory, persona as persona_service, state_machine
+from .store import store
+
+
+class _Transaction:
+    async def __aenter__(self) -> None:
+        return None
+
+    async def __aexit__(self, exc_type, exc, tb) -> bool:
+        return False
+
+
+class _Acquire:
+    def __init__(self, conn: "_ActiveSessionConnection") -> None:
+        self.conn = conn
+
+    async def __aenter__(self) -> "_ActiveSessionConnection":
+        return self.conn
+
+    async def __aexit__(self, exc_type, exc, tb) -> bool:
+        return False
+
+
+class _ActiveSessionConnection:
+    """새 row INSERT 전 active-case 확인 순서를 검증하는 최소 DB 대역."""
+
+    active_session_id = "00000000-0000-0000-0000-000000000777"
+
+    def __init__(self) -> None:
+        self.active_query = ""
+        self.insert_attempted = False
+
+    def transaction(self) -> _Transaction:
+        return _Transaction()
+
+    async def fetchrow(self, query: str, *args: object) -> dict[str, object] | None:
+        if "UPDATE app.case_profile" in query:
+            return {"last_session_no": 2}
+        if "FROM app.sessions" in query and "ended_at IS NULL" in query:
+            self.active_query = query
+            return {"id": self.active_session_id}
+        if "INSERT INTO app.sessions" in query:
+            self.insert_attempted = True
+            raise AssertionError("active session guard must run before INSERT")
+        raise AssertionError(f"unexpected query: {query}")
+
+    async def execute(self, query: str, *args: object) -> str:
+        if "pg_advisory_xact_lock" in query:
+            return "SELECT 1"
+        raise AssertionError(f"unexpected execute: {query}")
+
+
+def _principal() -> Principal:
+    return Principal(
+        user_id="00000000-0000-0000-0000-000000000101",
+        role=Role.LEARNER,
+        cohort_ids=[],
+        email="continuity-guard@example.test",
+        display_name="연속성 검증 학습자",
+        consent_at=1.0,
+        profile_completed_at=1.0,
+    )
+
+
+def _catalog_persona() -> SimpleNamespace:
+    return SimpleNamespace(
+        card=persona_service.P1,
+        persona_id="00000000-0000-0000-0000-0000000000a1",
+        version=1,
+        degraded=False,
+    )
+
+
+class SessionContinuityGuardTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        store._sessions.clear()
+        sessions._RECALL_CACHE.clear()
+
+    async def asyncTearDown(self) -> None:
+        store._sessions.clear()
+        sessions._RECALL_CACHE.clear()
+
+    async def test_persistence_rejects_a_second_active_session_before_insert(self) -> None:
+        conn = _ActiveSessionConnection()
+        principal = _principal()
+
+        with (
+            patch.object(session_persistence, "get_pool", return_value=object()),
+            patch.object(session_persistence, "acquire", return_value=_Acquire(conn)),
+        ):
+            with self.assertRaises(session_persistence.ActiveSessionExistsError) as raised:
+                await session_persistence.create_session(
+                    learner_id=principal.user_id,
+                    card=persona_service.P1,
+                    theory_mode="humanistic",
+                    state=state_machine.SessionState(),
+                    session_no=2,
+                    persona_id="00000000-0000-0000-0000-0000000000a1",
+                    persona_version=1,
+                    case_id="00000000-0000-0000-0000-00000000ca5e",
+                )
+
+        self.assertEqual(raised.exception.session_id, conn.active_session_id)
+        self.assertIn("FOR UPDATE", conn.active_query)
+        self.assertIn("persona_id = $2::uuid", conn.active_query)
+        self.assertNotIn("case_id = $1::uuid", conn.active_query)
+        self.assertFalse(conn.insert_attempted)
+
+    async def test_persistence_builds_start_state_after_case_lock_and_active_check(
+        self,
+    ) -> None:
+        class ReadyConnection:
+            def __init__(self) -> None:
+                self.events: list[str] = []
+
+            def transaction(self) -> _Transaction:
+                return _Transaction()
+
+            async def fetchrow(
+                self,
+                query: str,
+                *args: object,
+            ) -> dict[str, object] | None:
+                if "FROM app.case_profile" in query and "FOR UPDATE" in query:
+                    self.events.append("case_lock")
+                    return {"case_id": args[0], "last_session_no": 1}
+                if "UPDATE app.case_profile" in query:
+                    self.events.append("case_counter")
+                    return {"last_session_no": 2}
+                if "FROM app.sessions" in query and "ended_at IS NULL" in query:
+                    self.events.append("active_check")
+                    return None
+                if "INSERT INTO app.sessions" in query:
+                    self.events.append("session_insert")
+                    return {
+                        "id": "00000000-0000-0000-0000-000000000302",
+                        "runtime_case_id": args[0],
+                        "case_id": args[1],
+                        "learner_id": args[2],
+                        "persona_code": args[5],
+                        "session_no": args[8],
+                        "theory_mode": args[9],
+                        "started_at": session_persistence.datetime.fromtimestamp(
+                            1_000.0,
+                            tz=session_persistence.timezone.utc,
+                        ),
+                        "ended_at": None,
+                        "prev_rapport_credit": args[10],
+                    }
+                raise AssertionError(f"unexpected query: {query}")
+
+            async def execute(self, query: str, *args: object) -> str:
+                if "pg_advisory_xact_lock" in query:
+                    self.events.append("scope_lock")
+                    return "SELECT 1"
+                if "INSERT INTO app.session_state" not in query:
+                    raise AssertionError(f"unexpected execute: {query}")
+                self.events.append("state_insert")
+                return "INSERT 0 1"
+
+        conn = ReadyConnection()
+        principal = _principal()
+        recalled_state = state_machine.SessionState(rapport_credit=0.37)
+        factory_calls: list[tuple[str, int]] = []
+
+        async def locked_state_factory(
+            case_id: str,
+            session_no: int,
+        ) -> state_machine.SessionState:
+            self.assertEqual(
+                conn.events,
+                ["scope_lock", "active_check", "case_lock", "case_counter"],
+            )
+            factory_calls.append((case_id, session_no))
+            return recalled_state
+
+        with (
+            patch.object(session_persistence, "get_pool", return_value=object()),
+            patch.object(session_persistence, "acquire", return_value=_Acquire(conn)),
+        ):
+            created = await session_persistence.create_session(
+                learner_id=principal.user_id,
+                card=persona_service.P1,
+                theory_mode="humanistic",
+                state=state_machine.SessionState(),
+                session_no=2,
+                persona_id="00000000-0000-0000-0000-0000000000a1",
+                persona_version=1,
+                case_id="00000000-0000-0000-0000-00000000ca5e",
+                locked_state_factory=locked_state_factory,
+            )
+
+        self.assertEqual(
+            factory_calls,
+            [("00000000-0000-0000-0000-00000000ca5e", 2)],
+        )
+        self.assertEqual(
+            conn.events,
+            [
+                "scope_lock",
+                "active_check",
+                "case_lock",
+                "case_counter",
+                "session_insert",
+                "state_insert",
+            ],
+        )
+        self.assertIsNotNone(created)
+        assert created is not None
+        self.assertIs(created.state, recalled_state)
+        self.assertEqual(created.prev_rapport_credit, 0.37)
+
+    async def test_start_returns_resumable_conflict_for_persisted_active_session(self) -> None:
+        principal = _principal()
+        catalog_persona = _catalog_persona()
+        active_session_id = "00000000-0000-0000-0000-000000000777"
+
+        with (
+            patch.object(
+                sessions,
+                "get_catalog_persona",
+                AsyncMock(return_value=catalog_persona),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "get_case_context",
+                AsyncMock(
+                    return_value=session_persistence.CaseContext(
+                        case_id="00000000-0000-0000-0000-00000000ca5e",
+                        last_session_no=1,
+                    )
+                ),
+            ),
+            patch.object(
+                sessions,
+                "_build_seed_recall",
+                AsyncMock(return_value=memory.RecallContext()),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "create_session",
+                AsyncMock(
+                    side_effect=session_persistence.ActiveSessionExistsError(
+                        active_session_id
+                    )
+                ),
+            ),
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await sessions.start_session(
+                    sessions.SessionStartRequest(persona_code=persona_service.P1.code),
+                    principal,
+                )
+
+        self.assertEqual(raised.exception.status_code, 409)
+        self.assertEqual(
+            raised.exception.detail,
+            {
+                "code": "active_session_exists",
+                "session_id": active_session_id,
+            },
+        )
+
+    async def test_runtime_fallback_rejects_a_second_active_session(self) -> None:
+        principal = _principal()
+        catalog_persona = _catalog_persona()
+        existing = store.create(
+            learner_id=principal.user_id,
+            persona=persona_service.P1,
+            theory_mode="humanistic",
+            state=state_machine.SessionState(),
+            persona_id=catalog_persona.persona_id,
+            persona_version=catalog_persona.version,
+        )
+
+        with (
+            patch.object(
+                sessions,
+                "get_catalog_persona",
+                AsyncMock(return_value=catalog_persona),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "get_case_context",
+                AsyncMock(return_value=None),
+            ),
+            patch.object(
+                sessions,
+                "_build_seed_recall",
+                AsyncMock(return_value=memory.RecallContext()),
+            ),
+            patch.object(
+                sessions.session_persistence,
+                "create_session",
+                AsyncMock(return_value=None),
+            ),
+            patch.object(sessions, "require_runtime_fallback_allowed"),
+        ):
+            with self.assertRaises(HTTPException) as raised:
+                await sessions.start_session(
+                    sessions.SessionStartRequest(persona_code=persona_service.P1.code),
+                    principal,
+                )
+
+        self.assertEqual(raised.exception.status_code, 409)
+        self.assertEqual(
+            raised.exception.detail,
+            {
+                "code": "active_session_exists",
+                "session_id": existing.session_id,
+            },
+        )
+        self.assertEqual(len(store.list()), 1)

apps/api/app/test_session_memory.py

@@ -693,6 +693,12 @@ class SessionMemoryPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 self.executed.append((query, args))
                 return "OK"
 
+            async def fetchval(self, query: str, *args: object) -> object:
+                self.executed.append((query, args))
+                test_case.assertIn("FROM app.case_profile", query)
+                test_case.assertIn("FOR NO KEY UPDATE", query)
+                return args[0]
+
             async def fetchrow(self, query: str, *args: object):
                 self.executed.append((query, args))
                 if "INSERT INTO app.pinned_fact" in query:
@@ -773,6 +779,17 @@ class SessionMemoryPersistenceTest(unittest.IsolatedAsyncioTestCase):
             persisted = await session_persistence.end_session(sess, carry)
 
         self.assertTrue(persisted)
+        case_lock_index = next(
+            index
+            for index, (query, _) in enumerate(conn.executed)
+            if "SELECT case_id" in query and "FOR NO KEY UPDATE" in query
+        )
+        session_end_index = next(
+            index
+            for index, (query, _) in enumerate(conn.executed)
+            if "UPDATE app.sessions" in query and "ended_at" in query
+        )
+        self.assertLess(case_lock_index, session_end_index)
         summary_writes = [
             args for query, args in conn.executed if "INSERT INTO app.session_summary" in query
         ]

apps/api/app/test_session_turn_persistence.py

@@ -249,15 +249,16 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
             version=3,
             degraded=False,
         )
-        case_context = sessions.session_persistence.CaseContext(
-            case_id=case_id,
-            last_session_no=1,
-        )
         next_session_id = "00000000-0000-0000-0000-000000000702"
 
         async def create_next_session(**kwargs: object) -> InProcSession:
             self.assertEqual(kwargs["case_id"], case_id)
-            self.assertEqual(kwargs["session_no"], 2)
+            self.assertEqual(kwargs["start_mode"], "continue")
+            # The route now delegates durable case selection to the locked
+            # persistence transaction. It only provides a provisional first
+            # session number; the selected case counter makes this session #2.
+            self.assertEqual(kwargs["session_no"], 1)
+            state = await kwargs["locked_state_factory"](case_id, 2)
             created = InProcSession(
                 session_id=next_session_id,
                 case_id=case_id,
@@ -265,9 +266,9 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 persona_code=card.code,
                 theory_mode=str(kwargs["theory_mode"]),
                 persona=card,
-                state=kwargs["state"],
+                state=state,
-                session_no=int(kwargs["session_no"]),
+                session_no=2,
-                prev_rapport_credit=float(kwargs["carry_rapport"]),
+                prev_rapport_credit=state.rapport_credit,
             )
             durable_sessions[created.session_id] = created
             return created
@@ -283,11 +284,6 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 "get_catalog_persona",
                 AsyncMock(return_value=catalog_persona),
             ),
-            patch.object(
-                sessions.session_persistence,
-                "get_case_context",
-                AsyncMock(return_value=case_context),
-            ),
             patch.object(
                 sessions,
                 "_build_seed_recall",
@@ -301,7 +297,10 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
             patch.object(sessions.asyncio, "create_task", close_background),
         ):
             started = await sessions.start_session(
-                sessions.SessionStartRequest(persona_code=card.code),
+                sessions.SessionStartRequest(
+                    persona_code=card.code,
+                    case_id=case_id,
+                ),
                 principal,
             )
 
@@ -511,8 +510,12 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 return FakeTransaction()
 
             async def fetchrow(self, query: str, *args: object) -> dict[str, object]:
+                if "FROM app.case_profile" in query and "FOR UPDATE" in query:
+                    return {"case_id": args[0], "last_session_no": 1}
                 if "UPDATE app.case_profile" in query:
                     return {"last_session_no": 2}
+                if "FROM app.sessions" in query and "ended_at IS NULL" in query:
+                    return None
                 if "INSERT INTO app.sessions" in query:
                     self.session_insert_args = args
                     return {
@@ -532,6 +535,8 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 raise AssertionError(f"unexpected query: {query}")
 
             async def execute(self, query: str, *args: object) -> str:
+                if "pg_advisory_xact_lock" in query:
+                    return "SELECT 1"
                 self.assert_jsonb_object(query, args)
                 return "INSERT 0 1"
 
@@ -1811,8 +1816,13 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
 
         async def fake_create_session(**kwargs):
             self.assertEqual(kwargs["case_id"], case_context.case_id)
-            self.assertEqual(kwargs["session_no"], 2)
+            self.assertEqual(kwargs["start_mode"], "continue")
-            self.assertGreater(kwargs["state"].rapport_credit, 0)
+            self.assertEqual(kwargs["session_no"], 1)
+            state = await kwargs["locked_state_factory"](
+                case_context.case_id,
+                2,
+            )
+            self.assertGreater(state.rapport_credit, 0)
             return InProcSession(
                 session_id="stable-case-session",
                 case_id=kwargs["case_id"],
@@ -1820,9 +1830,9 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 persona_code=card.code,
                 theory_mode=kwargs["theory_mode"],
                 persona=card,
-                state=kwargs["state"],
+                state=state,
-                session_no=kwargs["session_no"],
+                session_no=2,
-                prev_rapport_credit=kwargs["carry_rapport"],
+                prev_rapport_credit=state.rapport_credit,
             )
 
         def close_background(coro):
@@ -1833,11 +1843,6 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
             patch.object(
                 sessions, "get_catalog_persona", AsyncMock(return_value=catalog_persona)
             ),
-            patch.object(
-                sessions.session_persistence,
-                "get_case_context",
-                AsyncMock(return_value=case_context),
-            ),
             patch.object(
                 sessions,
                 "_build_seed_recall",
@@ -1851,7 +1856,10 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
             patch.object(sessions.asyncio, "create_task", close_background),
         ):
             response = await sessions.start_session(
-                sessions.SessionStartRequest(persona_code=card.code),
+                sessions.SessionStartRequest(
+                    persona_code=card.code,
+                    case_id=case_context.case_id,
+                ),
                 principal,
             )
 
@@ -1911,16 +1919,23 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                 return None
 
         async def fake_create_session(**kwargs):
+            test_case.assertEqual(kwargs["case_id"], case_context.case_id)
+            test_case.assertEqual(kwargs["start_mode"], "continue")
+            test_case.assertEqual(kwargs["session_no"], 1)
+            state = await kwargs["locked_state_factory"](
+                case_context.case_id,
+                2,
+            )
             return InProcSession(
                 session_id="db-seed-recall-session",
-                case_id=kwargs["case_id"],
+                case_id=case_context.case_id,
                 learner_id=principal.user_id,
                 persona_code=card.code,
                 theory_mode=kwargs["theory_mode"],
                 persona=card,
-                state=kwargs["state"],
+                state=state,
-                session_no=kwargs["session_no"],
+                session_no=2,
-                prev_rapport_credit=kwargs["carry_rapport"],
+                prev_rapport_credit=state.rapport_credit,
             )
 
         def close_background(coro):
@@ -1931,11 +1946,6 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
             patch.object(
                 sessions, "get_catalog_persona", AsyncMock(return_value=catalog_persona)
             ),
-            patch.object(
-                sessions.session_persistence,
-                "get_case_context",
-                AsyncMock(return_value=case_context),
-            ),
             patch.object(sessions.db, "get_pool", return_value=object()),
             patch.object(
                 sessions.db,
@@ -1950,7 +1960,10 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
             patch.object(sessions.asyncio, "create_task", close_background),
         ):
             response = await sessions.start_session(
-                sessions.SessionStartRequest(persona_code=card.code),
+                sessions.SessionStartRequest(
+                    persona_code=card.code,
+                    case_id=case_context.case_id,
+                ),
                 principal,
             )
         started = store.get(response.session_id)
@@ -2743,10 +2756,43 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
         )
         self.assertIn("AI 평가 재시도가 필요합니다", response.summary)
         self.assertNotIn("session evaluation timeout after 45s", response.summary)
+        self.assertIsNone(response.evaluationFailure)
         self.assertEqual(response.rubric, [])
         self.assertEqual(response.goodMoments, [])
         self.assertEqual(response.growthPoints, [])
 
+        teacher_response = sessions.build_session_review(
+            sessions.SessionReviewReadInput(
+                session=sess,
+                evaluation_record=error_record,
+                evaluation_durable=True,
+                include_teacher_review=True,
+            )
+        )
+        self.assertIsNotNone(teacher_response.evaluationFailure)
+        assert teacher_response.evaluationFailure is not None
+        self.assertEqual(teacher_response.evaluationFailure.code, "timeout")
+        self.assertTrue(teacher_response.evaluationFailure.retryable)
+        self.assertNotIn("session evaluation timeout after 45s", teacher_response.model_dump_json())
+
+        oversized_input_record = {
+            **error_record,
+            "error": "engine_error: Agy CLI 프롬프트가 Windows 명령줄 안전 한도(24,000자)를 초과했습니다.",
+        }
+        oversized_response = sessions.build_session_review(
+            sessions.SessionReviewReadInput(
+                session=sess,
+                evaluation_record=oversized_input_record,
+                evaluation_durable=True,
+                include_teacher_review=True,
+            )
+        )
+        self.assertIsNotNone(oversized_response.evaluationFailure)
+        assert oversized_response.evaluationFailure is not None
+        self.assertEqual(oversized_response.evaluationFailure.code, "legacy_argv_limit")
+        self.assertTrue(oversized_response.evaluationFailure.retryable)
+        self.assertNotIn("24,000", oversized_response.model_dump_json())
+
     def test_review_marks_stale_missing_session_evaluation_as_failed(self) -> None:
         principal = _principal()
         sess = _session(principal)
@@ -2780,7 +2826,10 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase):
                     session=sess,
                     evaluation_record=None,
                     evaluation_durable=True,
-                    now_ts=(sess.ended_at or sess.created_at) + 41.0,
+                    # 평가 생성·감사 원장·결과 저장의 전체 예산을 넘긴 뒤에만 누락을 실패로 본다.
+                    now_ts=(sess.ended_at or sess.created_at)
+                    + sessions.session_evaluation_stale_after_seconds()
+                    + 1.0,
                 )
             )
         finally:

apps/api/app/test_teacher_dashboard.py

@@ -10,7 +10,6 @@ from fastapi import HTTPException
 from .config import settings
 from .deps import Principal, Role
 from .routes import teacher
-from .session_read_model import MISSING_SESSION_EVALUATION_ERROR
 from .services import state_machine
 from .services.persona import P1
 from .store import InProcSession, TurnRecord
@@ -424,7 +423,11 @@ class TeacherDashboardGrowthTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(failed.evaluation_status, "error")
         self.assertFalse(failed.review_ready)
         self.assertEqual(failed.supervisor_state, "평가 실패")
-        self.assertEqual(failed.evaluation_error, "session evaluation timeout after 45s")
+        self.assertEqual(
+            failed.evaluation_error,
+            "AI 평가가 제한 시간 안에 끝나지 않았습니다. 다시 시도할 수 있습니다.",
+        )
+        self.assertNotIn("session evaluation timeout after 45s", response.model_dump_json())
 
     async def test_dashboard_marks_stale_missing_session_evaluation_as_failed(self) -> None:
         principal = _principal()
@@ -468,7 +471,10 @@ class TeacherDashboardGrowthTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(failed.evaluation_status, "error")
         self.assertFalse(failed.review_ready)
         self.assertEqual(failed.supervisor_state, "평가 실패")
-        self.assertEqual(failed.evaluation_error, MISSING_SESSION_EVALUATION_ERROR)
+        self.assertEqual(
+            failed.evaluation_error,
+            "회기말 평가 기록이 아직 저장되지 않았습니다. 다시 시도할 수 있습니다.",
+        )
 
     async def test_dashboard_fails_closed_when_safety_alerts_are_not_durable_outside_dev(self) -> None:
         principal = _principal()

apps/api/engine_gateway/provider_registry.py

@@ -698,28 +698,29 @@ async def _stream_agy(
         raise ProviderError("Agy CLI를 찾을 수 없습니다.")
     model, effort = await _resolve_selection(req, "agy_cli")
     prompt = _cli_prompt(system_prompt, user_payload)
-    if os.name == "nt" and len(prompt) > 24_000:
-        raise ProviderError(
-            "Agy CLI 프롬프트가 Windows 명령줄 안전 한도(24,000자)를 초과했습니다."
-        )
     args = [binary, "--model", model, "--sandbox"]
     if effort:
         args += ["--effort", effort]
     args += [
         "--print-timeout",
         f"{int(CLI_TIMEOUT_SECONDS)}s",
+        # 긴 deep-loop 축어록을 Windows argv에 싣지 않는다. Agy의 공식 stream-json
+        # 입력 계약은 prompt를 stdin의 단일 user 이벤트로 받으므로 명령줄 길이 한계를
+        # 피하면서 전체 마스킹 근거를 그대로 보존한다.
+        "--input-format",
+        "stream-json",
         "--output-format",
         "stream-json",
-        "--print",
-        prompt,
     ]
     proc = await asyncio.create_subprocess_exec(
         *args,
         cwd=str(_cli_runtime_cwd()),
+        stdin=asyncio.subprocess.PIPE,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
         env=_cli_subprocess_env(),
     )
+    assert proc.stdin is not None
     assert proc.stdout is not None
     assert proc.stderr is not None
     stderr_task = asyncio.create_task(proc.stderr.read())
@@ -730,6 +731,30 @@ async def _stream_agy(
     cached_input_tokens = 0
     result_status = ""
     try:
+        # 공식 protocol: 한 줄에 한 user event. 마지막 turn 뒤 stdin을 닫아도 CLI는
+        # terminal result를 내보낸 뒤 종료한다. stdin 거절은 child와 stderr를 정리한 뒤
+        # provider 오류로 승격해 프로세스를 남기지 않는다.
+        try:
+            proc.stdin.write(
+                (
+                    json.dumps(
+                        {"event": "user", "message": {"content": prompt}},
+                        ensure_ascii=False,
+                    )
+                    + "\n"
+                ).encode("utf-8")
+            )
+            await proc.stdin.drain()
+        except (BrokenPipeError, ConnectionResetError) as exc:
+            raise ProviderError("Agy CLI가 stdin 평가 입력을 수락하지 않았습니다.") from exc
+        finally:
+            if not proc.stdin.is_closing():
+                proc.stdin.close()
+            try:
+                await proc.stdin.wait_closed()
+            except (BrokenPipeError, ConnectionResetError):
+                # 이미 종료된 CLI가 close 직후 EOF를 끊어도 finally가 child를 회수한다.
+                pass
         async with asyncio.timeout(CLI_TIMEOUT_SECONDS):
             while True:
                 raw = await proc.stdout.readline()

apps/api/engine_gateway/test_provider_registry.py

@@ -19,8 +19,30 @@ class _FakeStreamReader:
         return self.body
 
 
+class _FakeStreamWriter:
+    def __init__(self):
+        self.writes: list[bytes] = []
+        self.closed = False
+
+    def write(self, data: bytes) -> None:
+        self.writes.append(data)
+
+    async def drain(self) -> None:
+        return None
+
+    def is_closing(self) -> bool:
+        return self.closed
+
+    def close(self) -> None:
+        self.closed = True
+
+    async def wait_closed(self) -> None:
+        return None
+
+
 class _FakeAgyProcess:
     def __init__(self, events: list[dict]):
+        self.stdin = _FakeStreamWriter()
         self.stdout = _FakeStreamReader(
             [(json.dumps(event, ensure_ascii=False) + "\n").encode("utf-8") for event in events]
         )
@@ -102,6 +124,7 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase):
                 },
             ]
         }
+        long_payload = "x" * 24_001
         with (
             patch.object(provider_registry, "_binary", return_value="codex.exe"),
             patch.object(
@@ -321,6 +344,7 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase):
             captured.append(args)
             return process
 
+        long_payload = "x" * 24_001
         with (
             patch.object(provider_registry, "_binary", return_value="agy"),
             patch.object(
@@ -342,7 +366,7 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase):
             result = await provider_registry.generate_with_provider(
                 request,
                 system_prompt="system",
-                user_payload="hello",
+                user_payload=long_payload,
             )
 
         self.assertEqual(result.text, "OK")
@@ -350,10 +374,16 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(result.tokens_out, 2)
         self.assertEqual(result.cost_usd, 0.00001515)
         args = captured[0]
-        print_index = args.index("--print")
+        self.assertIn("--input-format", args)
-        self.assertEqual(print_index, len(args) - 2)
+        self.assertEqual(args[args.index("--input-format") + 1], "stream-json")
-        self.assertIn("[시스템 지침]", args[-1])
         self.assertEqual(args[args.index("--output-format") + 1], "stream-json")
+        self.assertNotIn("--print", args)
+        self.assertTrue(all(long_payload not in str(arg) for arg in args))
+        self.assertTrue(process.stdin.closed)
+        sent = json.loads(b"".join(process.stdin.writes).decode("utf-8"))
+        self.assertEqual(sent["event"], "user")
+        self.assertIn("[시스템 지침]", sent["message"]["content"])
+        self.assertIn(long_payload, sent["message"]["content"])
 
     async def test_agy_stream_forwards_live_deltas_without_repeating_final_response(self):
         capabilities = provider_registry.EngineCapabilitiesResponse(
@@ -451,9 +481,52 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(events[-1].result.tokens_out, 2)
         self.assertEqual(events[-1].result.cost_usd, 0.00001515)
         args = captured[0]
+        self.assertIn("--input-format", args)
+        self.assertEqual(args[args.index("--input-format") + 1], "stream-json")
         self.assertIn("--output-format", args)
         self.assertEqual(args[args.index("--output-format") + 1], "stream-json")
-        self.assertEqual(args.index("--print"), len(args) - 2)
+        self.assertNotIn("--print", args)
+
+    async def test_agy_stdin_rejection_reaps_child_process(self):
+        request = GenerateRequest(
+            provider="agy_cli",
+            model="gemini-3.6-flash-high",
+            reasoning_effort="high",
+            messages=[EngineMessage(role="user", content="hello")],
+        )
+        process = _FakeAgyProcess([])
+
+        async def broken_drain() -> None:
+            raise BrokenPipeError()
+
+        process.stdin.drain = broken_drain  # type: ignore[method-assign]
+
+        async def fake_create_subprocess_exec(*args, **kwargs):
+            return process
+
+        with (
+            patch.object(provider_registry, "_binary", return_value="agy.exe"),
+            patch.object(
+                provider_registry,
+                "_resolve_selection",
+                AsyncMock(return_value=("gemini-3.6-flash-high", "high")),
+            ),
+            patch.object(
+                provider_registry.asyncio,
+                "create_subprocess_exec",
+                fake_create_subprocess_exec,
+            ),
+        ):
+            with self.assertRaisesRegex(provider_registry.ProviderError, "stdin 평가 입력"):
+                async for _event in provider_registry._stream_agy(
+                    request,
+                    system_prompt="system",
+                    user_payload="hello",
+                ):
+                    pass
+
+        self.assertTrue(process.stdin.closed)
+        self.assertEqual(process.returncode, -9)
 
     async def test_generation_rejects_model_effort_not_returned_by_provider(self):
         capabilities = provider_registry.EngineCapabilitiesResponse(

infra/db/init/02_schema.sql

@@ -347,7 +347,8 @@ CREATE INDEX IF NOT EXISTS idx_sess_sum_emb ON app.session_summary
 
 -- =============================================================================
 -- 8. 메모리 ④ SEMANTIC (설계서 §3.5) — case_profile(evolving) + pinned_fact
---    케이스 = (persona_id 템플릿) × (learner_id 인스턴스). 학습자별 독립 연속체.
+--    케이스 = (persona_id 템플릿) × (learner_id 인스턴스) × 시작 시점.
+--    같은 내담자도 새 사례와 이어지는 사례를 분리해 학습자별 독립 연속체를 보존한다.
 -- =============================================================================
 CREATE TABLE IF NOT EXISTS app.case_profile (
   case_id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
@@ -360,9 +361,10 @@ CREATE TABLE IF NOT EXISTS app.case_profile (
   alliance_level     REAL DEFAULT 0.2,                  -- 치료동맹 누적(EWMA)
   case_digest        TEXT NOT NULL DEFAULT '',          -- 전체 궤적 8~12문장(큰그림)
   digest_embedding   vector(1024),
-  updated_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at         TIMESTAMPTZ NOT NULL DEFAULT now()
-  UNIQUE (persona_id, learner_id)                       -- 핵심 복합키(§3.5)
 );
+CREATE INDEX IF NOT EXISTS idx_case_profile_learner_persona_activity
+  ON app.case_profile (learner_id, persona_id, updated_at DESC, case_id);
 CREATE INDEX IF NOT EXISTS idx_case_learner ON app.case_profile(learner_id);
 CREATE INDEX IF NOT EXISTS idx_case_emb ON app.case_profile
   USING hnsw (digest_embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);

infra/db/init/20_public_bootstrap_ticket_events.sql

@@ -0,0 +1,32 @@
+-- 공개 부트스트랩 장애의 재시도 이력을 support_ticket 본문과 분리해 append-only로 보존한다.
+-- 브라우저가 보낼 수 있는 값은 API의 폐쇄 enum으로 더 제한하므로 이 테이블에는
+-- 응답 원문, URL, 쿠키, 계정 식별자 같은 고객 데이터가 들어가지 않는다.
+
+CREATE TABLE IF NOT EXISTS app.support_ticket_event (
+  id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+  ticket_id UUID NOT NULL
+    REFERENCES app.support_ticket(id) ON DELETE CASCADE,
+  observer_kind TEXT NOT NULL CHECK (observer_kind = 'public_browser'),
+  event_kind TEXT NOT NULL CHECK (
+    event_kind IN ('auth_restore_failure', 'boot_render_failure')
+  ),
+  status_code SMALLINT NOT NULL CHECK (status_code IN (500, 502, 503, 504)),
+  attempt TEXT NOT NULL CHECK (attempt IN ('automatic', 'retry')),
+  source_path TEXT NOT NULL CHECK (source_path = '/ops/public-bootstrap'),
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_support_ticket_event_ticket_latest
+  ON app.support_ticket_event(ticket_id, created_at DESC);
+
+ALTER TABLE app.support_ticket_event ENABLE ROW LEVEL SECURITY;
+
+DROP POLICY IF EXISTS p_support_ticket_event_select ON app.support_ticket_event;
+CREATE POLICY p_support_ticket_event_select
+  ON app.support_ticket_event FOR SELECT
+  USING (app.current_role_name() = 'admin');
+
+DROP POLICY IF EXISTS p_support_ticket_event_insert ON app.support_ticket_event;
+CREATE POLICY p_support_ticket_event_insert
+  ON app.support_ticket_event FOR INSERT
+  WITH CHECK (app.current_role_name() = 'admin');

infra/db/init/21_single_active_session.sql

@@ -0,0 +1,87 @@
+-- =============================================================================
+-- Vignette · migration 21 — learner-persona별 단일 활성 회기
+-- =============================================================================
+-- 기존 DB 적용 전 owner가 아래 읽기 전용 점검을 실행한다.
+-- 중복 또는 실패 후 남은 invalid index가 있으면 임의 종료·DROP 하지 말고
+-- 소유자가 보존·복구 방식을 결정해야 한다.
+--
+-- SELECT learner_id, persona_id, count(*)
+-- FROM app.sessions
+-- WHERE ended_at IS NULL AND persona_id IS NOT NULL
+-- GROUP BY learner_id, persona_id
+-- HAVING count(*) > 1;
+--
+-- CREATE INDEX CONCURRENTLY는 transaction block 안에서 실행할 수 없다.
+-- release agent가 online migration으로 owner psql에 전달한다.
+-- 이 파일의 전·후 condition은 IF NOT EXISTS가 invalid/wrong index를 조용히
+-- 건너뛰는 경우를 fail-closed로 막는다.
+
+DO $$
+DECLARE
+  target_index oid := to_regclass('app.uq_sessions_one_active_learner_persona');
+BEGIN
+  IF EXISTS (
+    SELECT 1
+    FROM app.sessions
+    WHERE ended_at IS NULL
+      AND persona_id IS NOT NULL
+    GROUP BY learner_id, persona_id
+    HAVING count(*) > 1
+  ) THEN
+    RAISE EXCEPTION
+      'migration 21 blocked: duplicate active learner-persona sessions exist';
+  END IF;
+
+  IF target_index IS NOT NULL AND NOT EXISTS (
+    SELECT 1
+    FROM pg_index AS index_meta
+    WHERE index_meta.indexrelid = target_index
+      AND index_meta.indrelid = 'app.sessions'::regclass
+      AND index_meta.indisvalid
+      AND index_meta.indisready
+      AND index_meta.indisunique
+      AND index_meta.indnkeyatts = 2
+      AND pg_get_indexdef(index_meta.indexrelid, 1, true) = 'learner_id'
+      AND pg_get_indexdef(index_meta.indexrelid, 2, true) = 'persona_id'
+      AND regexp_replace(
+        pg_get_expr(index_meta.indpred, index_meta.indrelid),
+        '[[:space:]()]',
+        '',
+        'g'
+      ) = 'ended_atISNULLANDpersona_idISNOTNULL'
+  ) THEN
+    RAISE EXCEPTION
+      'migration 21 blocked: target index exists but is invalid or has a different definition';
+  END IF;
+END
+$$;
+
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS uq_sessions_one_active_learner_persona
+  ON app.sessions (learner_id, persona_id)
+  WHERE ended_at IS NULL AND persona_id IS NOT NULL;
+
+DO $$
+BEGIN
+  IF NOT EXISTS (
+    SELECT 1
+    FROM pg_index AS index_meta
+    WHERE index_meta.indexrelid = 'app.uq_sessions_one_active_learner_persona'::regclass
+      AND index_meta.indrelid = 'app.sessions'::regclass
+      AND index_meta.indisvalid
+      AND index_meta.indisready
+      AND index_meta.indisunique
+      AND index_meta.indnkeyatts = 2
+      AND pg_get_indexdef(index_meta.indexrelid, 1, true) = 'learner_id'
+      AND pg_get_indexdef(index_meta.indexrelid, 2, true) = 'persona_id'
+      AND regexp_replace(
+        pg_get_expr(index_meta.indpred, index_meta.indrelid),
+        '[[:space:]()]',
+        '',
+        'g'
+      ) = 'ended_atISNULLANDpersona_idISNOTNULL'
+  ) THEN
+    RAISE EXCEPTION
+      'migration 21 failed: valid target unique index was not created';
+  END IF;
+END
+$$;

infra/db/init/22_case_profile_multi_case.sql

@@ -0,0 +1,129 @@
+-- =============================================================================
+-- Vignette · migration 22 — 동일 learner-persona의 새 사례 분리
+-- =============================================================================
+-- migration 21의 learner-persona 단일 활성 회기 제약은 그대로 유지한다.
+-- 이 migration은 끝난 기존 사례를 보존한 채, 같은 내담자에 새 사례(case_profile)를
+-- 만들 수 있도록 legacy pair unique만 제거한다. raw transcript·memory row는 수정하지 않는다.
+--
+-- CREATE INDEX CONCURRENTLY는 transaction block 안에서 실행할 수 없다.
+-- release agent가 online migration으로 owner psql에 전달한다.
+
+SET lock_timeout = '5s';
+SET statement_timeout = '15min';
+
+DO $$
+DECLARE
+  legacy_constraints text[];
+  target_index oid := to_regclass('app.idx_case_profile_learner_persona_activity');
+BEGIN
+  SELECT array_agg(candidate.conname ORDER BY candidate.conname)
+  INTO legacy_constraints
+  FROM (
+    SELECT constraint_meta.conname
+    FROM pg_constraint AS constraint_meta
+    WHERE constraint_meta.conrelid = 'app.case_profile'::regclass
+      AND constraint_meta.contype = 'u'
+      AND ARRAY(
+        SELECT attribute_meta.attname
+        FROM unnest(constraint_meta.conkey) WITH ORDINALITY AS key_column(attnum, ordinality)
+        JOIN pg_attribute AS attribute_meta
+          ON attribute_meta.attrelid = constraint_meta.conrelid
+         AND attribute_meta.attnum = key_column.attnum
+        ORDER BY key_column.ordinality
+      ) = ARRAY['persona_id', 'learner_id']::text[]
+  ) AS candidate;
+
+  IF COALESCE(array_length(legacy_constraints, 1), 0) > 1 THEN
+    RAISE EXCEPTION
+      'migration 22 blocked: multiple legacy case_profile persona-learner unique constraints exist';
+  END IF;
+
+  IF target_index IS NOT NULL AND NOT EXISTS (
+    SELECT 1
+    FROM pg_index AS index_meta
+    WHERE index_meta.indexrelid = target_index
+      AND index_meta.indrelid = 'app.case_profile'::regclass
+      AND index_meta.indisvalid
+      AND index_meta.indisready
+      AND NOT index_meta.indisunique
+      AND index_meta.indnkeyatts = 4
+      AND pg_get_indexdef(index_meta.indexrelid) LIKE
+        '%(learner_id, persona_id, updated_at DESC, case_id)%'
+  ) THEN
+    RAISE EXCEPTION
+      'migration 22 blocked: target index exists but is invalid or has a different definition';
+  END IF;
+END
+$$;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_case_profile_learner_persona_activity
+  ON app.case_profile (learner_id, persona_id, updated_at DESC, case_id);
+
+DO $$
+DECLARE
+  legacy_constraints text[];
+BEGIN
+  IF NOT EXISTS (
+    SELECT 1
+    FROM pg_index AS index_meta
+    WHERE index_meta.indexrelid = 'app.idx_case_profile_learner_persona_activity'::regclass
+      AND index_meta.indrelid = 'app.case_profile'::regclass
+      AND index_meta.indisvalid
+      AND index_meta.indisready
+      AND NOT index_meta.indisunique
+      AND index_meta.indnkeyatts = 4
+      AND pg_get_indexdef(index_meta.indexrelid) LIKE
+        '%(learner_id, persona_id, updated_at DESC, case_id)%'
+  ) THEN
+    RAISE EXCEPTION
+      'migration 22 failed: valid case activity index was not created';
+  END IF;
+
+  SELECT array_agg(candidate.conname ORDER BY candidate.conname)
+  INTO legacy_constraints
+  FROM (
+    SELECT constraint_meta.conname
+    FROM pg_constraint AS constraint_meta
+    WHERE constraint_meta.conrelid = 'app.case_profile'::regclass
+      AND constraint_meta.contype = 'u'
+      AND ARRAY(
+        SELECT attribute_meta.attname
+        FROM unnest(constraint_meta.conkey) WITH ORDINALITY AS key_column(attnum, ordinality)
+        JOIN pg_attribute AS attribute_meta
+          ON attribute_meta.attrelid = constraint_meta.conrelid
+         AND attribute_meta.attnum = key_column.attnum
+        ORDER BY key_column.ordinality
+      ) = ARRAY['persona_id', 'learner_id']::text[]
+  ) AS candidate;
+
+  IF COALESCE(array_length(legacy_constraints, 1), 0) > 1 THEN
+    RAISE EXCEPTION
+      'migration 22 blocked: multiple legacy case_profile persona-learner unique constraints exist';
+  END IF;
+
+  IF COALESCE(array_length(legacy_constraints, 1), 0) = 1 THEN
+    EXECUTE format(
+      'ALTER TABLE app.case_profile DROP CONSTRAINT %I',
+      legacy_constraints[1]
+    );
+  END IF;
+
+  IF EXISTS (
+    SELECT 1
+    FROM pg_constraint AS constraint_meta
+    WHERE constraint_meta.conrelid = 'app.case_profile'::regclass
+      AND constraint_meta.contype = 'u'
+      AND ARRAY(
+        SELECT attribute_meta.attname
+        FROM unnest(constraint_meta.conkey) WITH ORDINALITY AS key_column(attnum, ordinality)
+        JOIN pg_attribute AS attribute_meta
+          ON attribute_meta.attrelid = constraint_meta.conrelid
+         AND attribute_meta.attnum = key_column.attnum
+        ORDER BY key_column.ordinality
+      ) = ARRAY['persona_id', 'learner_id']::text[]
+  ) THEN
+    RAISE EXCEPTION
+      'migration 22 failed: legacy case_profile persona-learner unique constraint remains';
+  END IF;
+END
+$$;