From 72353ecd826570379dfa98c596856b8205707e95 Mon Sep 17 00:00:00 2001 From: Yun Chan Date: Tue, 1 Sep 2026 11:45:16 +0900 Subject: [PATCH] =?UTF-8?q?=ED=9A=8C=EA=B8=B0=20=EC=97=B0=EC=86=8D?= =?UTF-8?q?=EC=84=B1=EA=B3=BC=20=EB=A9=80=ED=8B=B0=20=EC=BC=80=EC=9D=B4?= =?UTF-8?q?=EC=8A=A4=20=EA=B3=84=EC=95=BD=EC=9D=84=20=EC=98=81=EC=86=8D?= =?UTF-8?q?=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/config.py | 5 +- apps/api/app/routes/eval.py | 50 ++- apps/api/app/routes/sessions.py | 322 ++++++++++++++-- apps/api/app/routes/teacher.py | 7 +- apps/api/app/services/evaluator.py | 3 +- apps/api/app/services/llm_audit.py | 10 +- apps/api/app/session_evaluation_timeout.py | 58 +++ apps/api/app/session_persistence.py | 294 +++++++++++++-- apps/api/app/session_read_model.py | 126 ++++++- apps/api/app/store.py | 22 +- apps/api/app/test_eval_routes.py | 63 +++- apps/api/app/test_evaluation_persistence.py | 40 ++ apps/api/app/test_evaluator_model_routing.py | 23 +- apps/api/app/test_notifications.py | 55 ++- apps/api/app/test_persona_session_contract.py | 10 +- apps/api/app/test_session_case_api.py | 357 ++++++++++++++++++ apps/api/app/test_session_continuity_guard.py | 326 ++++++++++++++++ apps/api/app/test_session_memory.py | 17 + apps/api/app/test_session_turn_persistence.py | 121 ++++-- apps/api/app/test_teacher_dashboard.py | 12 +- apps/api/engine_gateway/provider_registry.py | 37 +- .../engine_gateway/test_provider_registry.py | 83 +++- infra/db/init/02_schema.sql | 8 +- .../20_public_bootstrap_ticket_events.sql | 32 ++ infra/db/init/21_single_active_session.sql | 87 +++++ infra/db/init/22_case_profile_multi_case.sql | 129 +++++++ 26 files changed, 2170 insertions(+), 127 deletions(-) create mode 100644 apps/api/app/session_evaluation_timeout.py create mode 100644 apps/api/app/test_session_case_api.py create mode 100644 apps/api/app/test_session_continuity_guard.py create mode 100644 infra/db/init/20_public_bootstrap_ticket_events.sql create mode 100644 infra/db/init/21_single_active_session.sql create mode 100644 infra/db/init/22_case_profile_multi_case.sql diff --git a/apps/api/app/config.py b/apps/api/app/config.py index 5ac61e8..50e1395 100644 --- a/apps/api/app/config.py +++ b/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( diff --git a/apps/api/app/routes/eval.py b/apps/api/app/routes/eval.py index f3da643..50a004a 100644 --- a/apps/api/app/routes/eval.py +++ b/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 {}, diff --git a/apps/api/app/routes/sessions.py b/apps/api/app/routes/sessions.py index f6bce87..a17b2f0 100644 --- a/apps/api/app/routes/sessions.py +++ b/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 = ( - _session_evaluation_timeout_seconds() + MISSING_SESSION_EVALUATION_GRACE_SECONDS - ) + stale_after_seconds = session_evaluation_stale_after_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( - learner_id=principal.user_id, - persona_id=catalog_persona.persona_id, - ) - 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 + # The durable transaction chooses or creates the case only after active-case + # validation. Start with an empty recall here so a fresh request cannot see + # any legacy case before its own empty case exists. + recall = memory.build_recall_context() 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, ) diff --git a/apps/api/app/routes/teacher.py b/apps/api/app/routes/teacher.py index 50ccfb9..b5cbf7e 100644 --- a/apps/api/app/routes/teacher.py +++ b/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=( - str(summary_evaluation_record.get("error") or "") if summary_evaluation_record else None - ), + evaluation_error=teacher_evaluation_failure_message(evaluation_failure), ) diff --git a/apps/api/app/services/evaluator.py b/apps/api/app/services/evaluator.py index d9f6cb2..4a2b63d 100644 --- a/apps/api/app/services/evaluator.py +++ b/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 diff --git a/apps/api/app/services/llm_audit.py b/apps/api/app/services/llm_audit.py index 6af24db..ad22e07 100644 --- a/apps/api/app/services/llm_audit.py +++ b/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, diff --git a/apps/api/app/session_evaluation_timeout.py b/apps/api/app/session_evaluation_timeout.py new file mode 100644 index 0000000..f1984c3 --- /dev/null +++ b/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 + ) diff --git a/apps/api/app/session_persistence.py b/apps/api/app/session_persistence.py index 3ea6d72..6553e30 100644 --- a/apps/api/app/session_persistence.py +++ b/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( - """ - 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 - """, - persona_id, - learner_id, - ) + if case_id: + row = await conn.fetchrow( + """ + SELECT case_id, last_session_no + FROM app.case_profile + WHERE case_id = $1::uuid + AND persona_id = $2::uuid + AND learner_id = $3::uuid + """, + 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 - if stable_case_id is None: + # A case row cannot serialize two simultaneous first/fresh starts, + # 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: - session_no = int(counter_row["last_session_no"] or session_no) + if counter_row is None: + 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" + ) diff --git a/apps/api/app/session_read_model.py b/apps/api/app/session_read_model.py index d2ed949..b7de784 100644 --- a/apps/api/app/session_read_model.py +++ b/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: - configured = float(settings.session_evaluation_timeout or settings.engine_timeout) - return max(1.0, configured) +def teacher_evaluation_failure( + evaluation_record: dict[str, object] | None, +) -> 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, ) diff --git a/apps/api/app/store.py b/apps/api/app/store.py index c9ff3bf..39638d9 100644 --- a/apps/api/app/store.py +++ b/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: diff --git a/apps/api/app/test_eval_routes.py b/apps/api/app/test_eval_routes.py index 9e1cdd4..3e72350 100644 --- a/apps/api/app/test_eval_routes.py +++ b/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( diff --git a/apps/api/app/test_evaluation_persistence.py b/apps/api/app/test_evaluation_persistence.py index 3100734..38bc6e4 100644 --- a/apps/api/app/test_evaluation_persistence.py +++ b/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( diff --git a/apps/api/app/test_evaluator_model_routing.py b/apps/api/app/test_evaluator_model_routing.py index 1854de7..df292a7 100644 --- a/apps/api/app/test_evaluator_model_routing.py +++ b/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 = "" diff --git a/apps/api/app/test_notifications.py b/apps/api/app/test_notifications.py index 83257b6..e2557b0 100644 --- a/apps/api/app/test_notifications.py +++ b/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() diff --git a/apps/api/app/test_persona_session_contract.py b/apps/api/app/test_persona_session_contract.py index 4eaee3c..500fa7f 100644 --- a/apps/api/app/test_persona_session_contract.py +++ b/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}") diff --git a/apps/api/app/test_session_case_api.py b/apps/api/app/test_session_case_api.py new file mode 100644 index 0000000..f9ea53d --- /dev/null +++ b/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() diff --git a/apps/api/app/test_session_continuity_guard.py b/apps/api/app/test_session_continuity_guard.py new file mode 100644 index 0000000..4d85c9a --- /dev/null +++ b/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) diff --git a/apps/api/app/test_session_memory.py b/apps/api/app/test_session_memory.py index d4db2f2..b6d873c 100644 --- a/apps/api/app/test_session_memory.py +++ b/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 ] diff --git a/apps/api/app/test_session_turn_persistence.py b/apps/api/app/test_session_turn_persistence.py index a3b2aaf..c46274f 100644 --- a/apps/api/app/test_session_turn_persistence.py +++ b/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"], - session_no=int(kwargs["session_no"]), - prev_rapport_credit=float(kwargs["carry_rapport"]), + state=state, + session_no=2, + 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.assertGreater(kwargs["state"].rapport_credit, 0) + self.assertEqual(kwargs["start_mode"], "continue") + 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"], - session_no=kwargs["session_no"], - prev_rapport_credit=kwargs["carry_rapport"], + state=state, + session_no=2, + 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"], - session_no=kwargs["session_no"], - prev_rapport_credit=kwargs["carry_rapport"], + state=state, + session_no=2, + 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: diff --git a/apps/api/app/test_teacher_dashboard.py b/apps/api/app/test_teacher_dashboard.py index 08cf767..43eeade 100644 --- a/apps/api/app/test_teacher_dashboard.py +++ b/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() diff --git a/apps/api/engine_gateway/provider_registry.py b/apps/api/engine_gateway/provider_registry.py index 5594829..c8a066e 100644 --- a/apps/api/engine_gateway/provider_registry.py +++ b/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() diff --git a/apps/api/engine_gateway/test_provider_registry.py b/apps/api/engine_gateway/test_provider_registry.py index e0ae71c..d7c13b6 100644 --- a/apps/api/engine_gateway/test_provider_registry.py +++ b/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.assertEqual(print_index, len(args) - 2) - self.assertIn("[시스템 지침]", args[-1]) + self.assertIn("--input-format", args) + self.assertEqual(args[args.index("--input-format") + 1], "stream-json") 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( diff --git a/infra/db/init/02_schema.sql b/infra/db/init/02_schema.sql index 573fca3..d5764fe 100644 --- a/infra/db/init/02_schema.sql +++ b/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(), - UNIQUE (persona_id, learner_id) -- 핵심 복합키(§3.5) + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +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); diff --git a/infra/db/init/20_public_bootstrap_ticket_events.sql b/infra/db/init/20_public_bootstrap_ticket_events.sql new file mode 100644 index 0000000..f701b88 --- /dev/null +++ b/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'); diff --git a/infra/db/init/21_single_active_session.sql b/infra/db/init/21_single_active_session.sql new file mode 100644 index 0000000..11da77c --- /dev/null +++ b/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 +$$; diff --git a/infra/db/init/22_case_profile_multi_case.sql b/infra/db/init/22_case_profile_multi_case.sql new file mode 100644 index 0000000..ec9074f --- /dev/null +++ b/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 +$$;