"""DB-backed counseling session persistence with in-process fallback support."""

from __future__ import annotations

import hashlib
import json
import logging
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Iterable, Protocol

from .db import acquire, get_pool
from .deps import Principal
from .config import settings
from .persona_repository import (
    SEED_VERSION,
    card_from_row,
    seed_fallback_persona,
    seed_persona_id,
)
from .runtime_policy import require_runtime_fallback_allowed, runtime_fallback_allowed
from .runtime_schema import (
    REVIEW_SCHEMA_CONTRACT,
    runtime_schema_bootstrap_required,
    schema_contract_ready,
)
from .services import guardrail, memory, state_machine
from .services.evaluation_contract import PERSISTED_APPROPRIATENESS_SCORE_5PT
from .services.persona import PersonaCard
from .store import DEFAULT_TURN_VISIBLE_TO, InProcSession, TurnRecord

logger = logging.getLogger(__name__)

_EVALUATION_CACHE: dict[str, dict[str, Any]] = {}
_CASE_WORKSHEET_CACHE: dict[str, dict[str, Any]] = {}
_SESSION_REVIEW_STATUS_CACHE: dict[str, dict[str, Any]] = {}
_SESSION_SHARE_CACHE: dict[str, dict[str, Any]] = {}
_SESSION_SHARE_TOKEN_INDEX: dict[str, str] = {}
_LIVE_COACH_EVENT_CACHE: dict[str, list[dict[str, Any]]] = {}
_SESSION_ARCHIVE_CACHE: dict[str, dict[str, Any]] = {}
_SESSION_AUDIT_ROLES = {"teacher", "admin"}
_WORKSHEET_REVIEW_STATUS_VALUES = {
    "pending",
    "approved",
    "changes_requested",
    "rejected",
}
LIVE_COACH_INITIAL_CREDITS = 3
LIVE_COACH_MAX_CREDITS = 3
LIVE_COACH_USE_REASON = "AI 코칭 힌트 사용"
LIVE_COACH_RECHARGE_REASON = "좋은 발화로 내담자 변화 신호 확인"

SESSION_PERSONA_SELECT_COLUMNS_SQL = """
    s.id, s.runtime_case_id, s.case_id, s.learner_id, s.persona_code,
    s.session_no, s.theory_mode, s.started_at, s.ended_at,
    s.prev_rapport_credit, s.session_goals,
    COALESCE(
      NULLIF(learner.display_name, ''),
      NULLIF(learner.nickname, ''),
      NULLIF(learner.email, ''),
      s.learner_id::text
    ) AS learner_label,
    pc.persona_id AS card_persona_id,
    pc.code AS card_code,
    pc.version AS card_version,
    pc.status AS card_status,
    pc.display_name AS card_display_name,
    pc.difficulty AS card_difficulty,
    pc.theory_target AS card_theory_target,
    pc.demographics AS card_demographics,
    pc.presenting AS card_presenting,
    pc.history AS card_history,
    pc.big5 AS card_big5,
    pc.resistance AS card_resistance,
    pc.speech_style AS card_speech_style,
    pc.affect_baseline AS card_affect_baseline,
    pc.ccd AS card_ccd,
    pc.dsm5_dimensional AS card_dsm5_dimensional,
    pc.triggers AS card_triggers,
    pc.source_provenance AS card_source_provenance,
    pc.is_synthetic AS card_is_synthetic
"""

SESSION_PERSONA_JOINS_SQL = """
    LEFT JOIN app.persona_card pc
      ON pc.persona_id = s.persona_id
     AND pc.version = s.persona_version
    LEFT JOIN app.app_user learner ON learner.user_id = s.learner_id
"""

SESSION_STATE_SELECT_COLUMNS_SQL = """
    session_id, stage, turn_seq, effective_openness, rapport_credit, resistance,
    ideation_stage, turns_in_stage, affect_state
"""

SESSION_TURN_SELECT_COLUMNS_SQL = """
    session_id, id, seq, speaker, stage, text, text_masked, created_at,
    llm_provider, model, tokens_in, tokens_out, cost_usd,
    audio_ref, silence_ms, speech_rate, barge_in, provider_events, visible_to
"""


class LiveCoachCreditExhausted(RuntimeError):
    """Raised when a learner tries to use live coaching without credits."""


def _coerce_error_message(error: BaseException | str) -> str:
    if isinstance(error, BaseException):
        message = str(error).strip()
        return message or error.__class__.__name__
    return str(error).strip() or "unknown session evaluation error"


@dataclass(slots=True)
class CaseContext:
    case_id: str
    last_session_no: int


@dataclass(slots=True)
class SessionSummaryWrite:
    session_id: str
    case_id: str
    session_no: int
    end_state: dict[str, Any]
    rapport_delta: float
    digest: str
    open_threads: list[str]


class _SessionEvaluationResult(Protocol):
    scope: str
    stage: str
    error: str | None

    def to_dict(self) -> dict[str, Any]: ...


@dataclass(slots=True)
class SessionEvaluationWrite:
    session_id: str
    learner_id: str
    status: str
    source: str
    scope: str
    stage: str
    payload: dict[str, Any]
    error: str | None = None

    @classmethod
    def from_result(
        cls,
        *,
        session_id: str,
        learner_id: str,
        result: _SessionEvaluationResult,
        source: str = "engine",
    ) -> "SessionEvaluationWrite":
        return cls(
            session_id=session_id,
            learner_id=learner_id,
            status="error" if result.error else "ready",
            source=source,
            scope=result.scope,
            stage=result.stage,
            payload=_mask_json_text_values(result.to_dict()),
            error=_clean_masked_text(result.error),
        )

    @classmethod
    def from_error(
        cls,
        *,
        session_id: str,
        learner_id: str,
        scope: str,
        stage: str,
        error: BaseException | str,
        source: str = "engine",
    ) -> "SessionEvaluationWrite":
        return cls(
            session_id=session_id,
            learner_id=learner_id,
            status="error",
            source=source,
            scope=scope,
            stage=stage,
            payload={},
            error=_clean_masked_text(_coerce_error_message(error)),
        )

    def cache_record(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "source": self.source,
            "scope": self.scope,
            "stage": self.stage,
            "payload": self.payload,
            "error": self.error,
        }


_JOINED_CARD_COLUMNS = (
    "card_persona_id",
    "card_code",
    "card_version",
    "card_status",
    "card_display_name",
    "card_difficulty",
    "card_theory_target",
    "card_demographics",
    "card_presenting",
    "card_history",
    "card_big5",
    "card_resistance",
    "card_speech_style",
    "card_affect_baseline",
    "card_ccd",
    "card_dsm5_dimensional",
    "card_triggers",
    "card_source_provenance",
    "card_is_synthetic",
)


def _ts(value: datetime | None) -> float | None:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    return value.timestamp()


def share_token_hash(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def _share_record_from_row(row) -> dict[str, Any]:
    return {
        "session_id": str(row["session_id"]),
        "payload": dict(row["payload"] or {}),
        "created_at": _ts(row["created_at"]),
        "updated_at": _ts(row["updated_at"]),
        "revoked_at": _ts(row["revoked_at"]),
    }


def _archive_record_from_row(row) -> dict[str, Any]:
    return {
        "session_id": str(row["session_id"]),
        "learner_id": str(row["learner_id"]),
        "archived_at": _ts(row["archived_at"]),
        "updated_at": _ts(row["updated_at"]),
    }


def _model_payload(value: Any) -> dict[str, Any]:
    if hasattr(value, "model_dump"):
        return value.model_dump(mode="json")
    if isinstance(value, dict):
        return dict(value)
    return {}


def _json_object_payload(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return dict(value)
    if not isinstance(value, str):
        return {}
    text = value.strip()
    for _ in range(2):
        if not text:
            return {}
        try:
            parsed = json.loads(text)
        except json.JSONDecodeError:
            return {}
        if isinstance(parsed, dict):
            return parsed
        if not isinstance(parsed, str):
            return {}
        text = parsed.strip()
    return {}


def _json_list_payload(value: Any) -> list[Any]:
    if isinstance(value, list):
        return list(value)
    if not isinstance(value, str):
        return []
    text = value.strip()
    if not text:
        return []
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        return []
    return parsed if isinstance(parsed, list) else []


def _masked_excerpt(value: str | None, *, limit: int = 220) -> str | None:
    text = (value or "").strip()
    if not text:
        return None
    result = guardrail.mask_pii(text)
    masked = str(getattr(result, "text_masked", text)).strip()
    compact = " ".join(masked.split())
    if len(compact) <= limit:
        return compact
    return f"{compact[: max(0, limit - 1)].rstrip()}..."


def _live_coach_event_from_row(row) -> dict[str, Any]:
    return {
        "event_id": str(row["id"]),
        "session_id": str(row["session_id"]),
        "turn_seq": int(row["turn_seq"] or 1),
        "stage": str(row["stage"] or ""),
        "created_at": _iso_dt(row["created_at"]),
        "learner_text_excerpt": row["learner_text_excerpt"],
        "client_reply_excerpt": row["client_reply_excerpt"],
        "suggestion": dict(row["payload"] or {}),
    }


def _live_coach_remaining(value: int | float | None) -> int:
    try:
        parsed = int(value if value is not None else LIVE_COACH_INITIAL_CREDITS)
    except (TypeError, ValueError):
        parsed = LIVE_COACH_INITIAL_CREDITS
    return max(0, min(LIVE_COACH_MAX_CREDITS, parsed))


def _live_coach_quota_payload(remaining: int | float | None) -> dict[str, int]:
    return {
        "remaining": _live_coach_remaining(remaining),
        "max": LIVE_COACH_MAX_CREDITS,
    }


def _live_coach_delta_from_record(record: dict[str, Any]) -> int:
    if record.get("credit_delta") is not None:
        try:
            return int(record["credit_delta"])
        except (TypeError, ValueError):
            pass
    return 1 if record.get("event_type") == "recharge" else -1


def _live_coach_cached_remaining(session_id: str) -> int:
    delta = sum(
        _live_coach_delta_from_record(record)
        for record in _LIVE_COACH_EVENT_CACHE.get(session_id, [])
    )
    return _live_coach_remaining(LIVE_COACH_INITIAL_CREDITS + delta)


async def _live_coach_remaining_for_conn(conn: Any, session_id: str) -> int:
    delta = await conn.fetchval(
        """
        SELECT COALESCE(SUM(credit_delta), 0)
        FROM app.live_coach_events
        WHERE session_id = $1::uuid
        """,
        session_id,
    )
    return _live_coach_remaining(LIVE_COACH_INITIAL_CREDITS + int(delta or 0))


def _live_coach_credit_event_from_row(row) -> dict[str, Any]:
    event_type = str(_row_value(row, "event_type") or "use")
    if event_type not in {"use", "recharge"}:
        event_type = "use"
    delta = _row_value(row, "credit_delta")
    if delta is None:
        delta = 1 if event_type == "recharge" else -1
    balance = _row_value(row, "credit_balance")
    return {
        "event_id": str(row["id"]),
        "session_id": str(row["session_id"]),
        "turn_seq": int(row["turn_seq"] or 1),
        "stage": str(row["stage"] or ""),
        "event_type": event_type,
        "delta": int(delta),
        "balance": _live_coach_remaining(balance),
        "reason": str(
            _row_value(row, "reason")
            or (
                LIVE_COACH_RECHARGE_REASON
                if event_type == "recharge"
                else LIVE_COACH_USE_REASON
            )
        ),
        "created_at": _iso_dt(row["created_at"]),
    }


def _live_coach_credit_event_from_record(record: dict[str, Any]) -> dict[str, Any]:
    event_type = str(record.get("event_type") or "use")
    if event_type not in {"use", "recharge"}:
        event_type = "use"
    delta = _live_coach_delta_from_record(record)
    return {
        "event_id": str(record.get("event_id") or uuid.uuid4()),
        "session_id": str(record.get("session_id") or ""),
        "turn_seq": int(record.get("turn_seq") or 1),
        "stage": str(record.get("stage") or ""),
        "event_type": event_type,
        "delta": delta,
        "balance": _live_coach_remaining(record.get("credit_balance")),
        "reason": str(
            record.get("reason")
            or (
                LIVE_COACH_RECHARGE_REASON
                if event_type == "recharge"
                else LIVE_COACH_USE_REASON
            )
        ),
        "created_at": str(record.get("created_at") or ""),
    }


def _live_coach_cache_record(
    *,
    session_id: str,
    turn_seq: int,
    stage: str,
    learner_text: str,
    client_reply: str | None,
    suggestion: Any,
    credit_balance: int | None = None,
) -> dict[str, Any]:
    now = datetime.now(timezone.utc)
    balance = _live_coach_remaining(
        credit_balance
        if credit_balance is not None
        else _live_coach_cached_remaining(session_id) - 1
    )
    return {
        "event_id": str(uuid.uuid4()),
        "session_id": session_id,
        "turn_seq": int(turn_seq),
        "stage": stage,
        "event_type": "use",
        "credit_delta": -1,
        "credit_balance": balance,
        "reason": LIVE_COACH_USE_REASON,
        "created_at": now.isoformat().replace("+00:00", "Z"),
        "learner_text_excerpt": _masked_excerpt(learner_text),
        "client_reply_excerpt": _masked_excerpt(client_reply),
        "suggestion": _model_payload(suggestion),
    }


def _live_coach_recharge_cache_record(
    *,
    session_id: str,
    turn_seq: int,
    stage: str,
    credit_balance: int,
    reason: str,
) -> dict[str, Any]:
    now = datetime.now(timezone.utc)
    return {
        "event_id": str(uuid.uuid4()),
        "session_id": session_id,
        "turn_seq": int(turn_seq),
        "stage": stage,
        "event_type": "recharge",
        "credit_delta": 1,
        "credit_balance": _live_coach_remaining(credit_balance),
        "reason": reason,
        "created_at": now.isoformat().replace("+00:00", "Z"),
        "learner_text_excerpt": None,
        "client_reply_excerpt": None,
        "suggestion": {},
    }


def _live_coach_payload_from_record(record: dict[str, Any]) -> dict[str, Any]:
    payload = dict(record.get("suggestion") or {})
    return payload


def _row_value(row, key: str):
    try:
        return row[key]
    except Exception:
        return None


def _card_from_joined_session_row(row) -> PersonaCard | None:
    if _row_value(row, "card_persona_id") is None:
        return None
    card_row = {
        key.removeprefix("card_"): _row_value(row, key) for key in _JOINED_CARD_COLUMNS
    }
    return card_from_row(card_row)


def _stage(stage: object) -> str:
    return getattr(stage, "value", str(stage))


def _clean_text(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _clean_masked_text(value: Any) -> str | None:
    text = _clean_text(value)
    if text is None:
        return None
    masked = guardrail.mask_pii(text).text_masked.strip()
    return masked or None


def _mask_json_text_values(value: Any) -> Any:
    if isinstance(value, str):
        return _clean_masked_text(value) or ""
    if isinstance(value, dict):
        return {key: _mask_json_text_values(child) for key, child in value.items()}
    if isinstance(value, list):
        return [_mask_json_text_values(child) for child in value]
    if isinstance(value, tuple):
        return [_mask_json_text_values(child) for child in value]
    return value


def _safe_float(value: Any) -> float | None:
    if isinstance(value, (int, float)):
        return float(value)
    return None


def _safe_int(value: Any) -> int | None:
    if isinstance(value, bool):
        return int(value)
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    return None


def _evaluation_loop(evaluation: dict[str, Any]) -> str:
    loop = _clean_text(evaluation.get("loop")) or "fast"
    return loop if loop in {"fast", "deep"} else "fast"


def _dict_items(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    return [item for item in value if isinstance(item, dict)]


def _evaluation_feedback_rows(
    evaluation: dict[str, Any] | None,
) -> list[dict[str, Any]]:
    """Normalize scalar/rationale turn-evaluation fields into feedback_scores rows."""
    if not isinstance(evaluation, dict):
        return []
    loop = _evaluation_loop(evaluation)
    rows: list[dict[str, Any]] = []

    def add(
        dimension: str,
        *,
        score: float | None = None,
        rationale: str | None = None,
        top1_score: float | None = None,
    ) -> None:
        dim = _clean_text(dimension)
        if not dim:
            return
        rows.append(
            {
                "dimension": dim,
                "score": score,
                "rationale": rationale,
                "top1_score": top1_score,
                "loop": loop,
            }
        )

    appropriateness = _clean_text(evaluation.get("appropriateness")) or "neutral"
    if appropriateness not in PERSISTED_APPROPRIATENESS_SCORE_5PT:
        appropriateness = "neutral"
    add(
        "appropriateness",
        score=PERSISTED_APPROPRIATENESS_SCORE_5PT[appropriateness],
        rationale=_clean_masked_text(evaluation.get("appropriateness_note")),
    )

    rapport = _safe_float(evaluation.get("rapport_signal"))
    if rapport is not None:
        add("rapport_signal", score=max(-1.0, min(1.0, rapport)))

    theory_mode = _clean_text(evaluation.get("theory_mode"))
    if theory_mode:
        add("theory_mode", rationale=theory_mode)

    error = _clean_masked_text(evaluation.get("error"))
    if error:
        add("error", rationale=error)

    for tag in _dict_items(evaluation.get("techniques")):
        code = _clean_text(tag.get("code"))
        rationale = _clean_masked_text(tag.get("rationale"))
        if code and rationale:
            add(f"technique:{code}", rationale=rationale)

    for state in _dict_items(evaluation.get("client_state_read")):
        code = _clean_text(state.get("code"))
        rationale = _clean_masked_text(state.get("rationale"))
        if code and rationale:
            add(f"client_state:{code}", rationale=rationale)

    return rows


def _evaluation_technique_rows(
    evaluation: dict[str, Any] | None,
) -> list[dict[str, str]]:
    if not isinstance(evaluation, dict):
        return []
    rows: list[dict[str, str]] = []
    for tag in _dict_items(evaluation.get("techniques")):
        code = _clean_text(tag.get("code"))
        if not code:
            continue
        rows.append(
            {
                "code": code,
                "label_ko": _clean_text(tag.get("label_ko")) or code,
                "category": _clean_text(tag.get("category")) or "",
            }
        )
    return rows


def _evaluation_client_state_rows(
    evaluation: dict[str, Any] | None,
) -> list[dict[str, str]]:
    if not isinstance(evaluation, dict):
        return []
    rows: list[dict[str, str]] = []
    for state in _dict_items(evaluation.get("client_state_read")):
        code = _clean_text(state.get("code"))
        if not code:
            continue
        rows.append(
            {
                "code": code,
                "label_ko": _clean_text(state.get("label_ko")) or code,
            }
        )
    return rows


def _evaluation_comment_rows(evaluation: dict[str, Any] | None) -> list[dict[str, Any]]:
    if not isinstance(evaluation, dict):
        return []
    deviation = evaluation.get("intent_deviation")
    if not isinstance(deviation, dict):
        return []
    note = (
        _clean_masked_text(evaluation.get("appropriateness_note")) or "의도와 다른 부분"
    )
    return [
        {
            "kind": "critique",
            "text": note,
            "intent_deviation": _mask_json_text_values(deviation),
        }
    ]


def _evaluation_alternative_rows(
    evaluation: dict[str, Any] | None,
) -> list[dict[str, str | None]]:
    if not isinstance(evaluation, dict):
        return []
    alternatives = evaluation.get("alternative_utterances")
    if not isinstance(alternatives, list):
        return []
    rows: list[dict[str, str | None]] = []
    for item in alternatives:
        if isinstance(item, dict):
            suggestion = _clean_masked_text(item.get("suggestion") or item.get("text"))
            rationale = _clean_masked_text(item.get("rationale"))
        else:
            suggestion = _clean_masked_text(item)
            rationale = None
        if suggestion:
            rows.append({"suggestion": suggestion, "rationale": rationale})
    return rows


def _appropriateness_from_score(score: Any) -> str:
    value = _safe_float(score)
    if value is None:
        return "neutral"
    if value >= 4.0:
        return "pos"
    if value <= 2.0:
        return "warn"
    return "neutral"


def _base_turn_evaluation(turn_seq: int, stage: str) -> dict[str, Any]:
    return {
        "loop": "fast",
        "turn_seq": turn_seq,
        "stage": stage,
        "techniques": [],
        "client_state_read": [],
        "appropriateness": "neutral",
    }


def _rebuild_turn_evaluations(
    turn_refs: list[tuple[str, int, str]],
    *,
    feedback_rows: Iterable[Any],
    technique_rows: Iterable[Any],
    client_state_rows: Iterable[Any],
    comment_rows: Iterable[Any],
    alternative_rows: Iterable[Any] = (),
) -> dict[str, dict[str, Any]]:
    """Rehydrate normalized DB rows back into the TurnRecord.evaluation shape."""
    refs = {turn_id: (turn_seq, stage) for turn_id, turn_seq, stage in turn_refs}
    evaluations: dict[str, dict[str, Any]] = {}
    rationale_by_dimension: dict[str, dict[str, str]] = {}

    def ensure(turn_id: str) -> dict[str, Any]:
        if turn_id not in evaluations:
            turn_seq, stage = refs[turn_id]
            evaluations[turn_id] = _base_turn_evaluation(turn_seq, stage)
        return evaluations[turn_id]

    for row in feedback_rows:
        turn_id = str(row["turn_id"])
        if turn_id not in refs:
            continue
        ev = ensure(turn_id)
        loop = _clean_text(row["loop"])
        if loop in {"fast", "deep"}:
            ev["loop"] = loop
        dimension = _clean_text(row["dimension"]) or ""
        rationale = _clean_text(row["rationale"])
        if rationale:
            rationale_by_dimension.setdefault(turn_id, {})[dimension] = rationale
        if dimension == "appropriateness":
            ev["appropriateness"] = _appropriateness_from_score(row["score"])
            if rationale:
                ev["appropriateness_note"] = rationale
        elif dimension == "rapport_signal":
            score = _safe_float(row["score"])
            if score is not None:
                ev["rapport_signal"] = max(-1.0, min(1.0, score))
        elif dimension == "theory_mode" and rationale:
            ev["theory_mode"] = rationale
        elif dimension == "error" and rationale:
            ev["error"] = rationale

    for row in technique_rows:
        turn_id = str(row["turn_id"])
        if turn_id not in refs:
            continue
        code = _clean_text(row["code"])
        if not code:
            continue
        item = {
            "code": code,
            "label_ko": _clean_text(row["label_ko"]) or code,
            "category": _clean_text(row["category"]) or "",
        }
        rationale = rationale_by_dimension.get(turn_id, {}).get(f"technique:{code}")
        if rationale:
            item["rationale"] = rationale
        ensure(turn_id)["techniques"].append(item)

    for row in client_state_rows:
        turn_id = str(row["turn_id"])
        if turn_id not in refs:
            continue
        code = _clean_text(row["code"])
        if not code:
            continue
        item = {
            "code": code,
            "label_ko": _clean_text(row["label_ko"]) or code,
        }
        rationale = rationale_by_dimension.get(turn_id, {}).get(f"client_state:{code}")
        if rationale:
            item["rationale"] = rationale
        ensure(turn_id)["client_state_read"].append(item)

    for row in comment_rows:
        turn_id = str(row["turn_id"])
        if turn_id not in refs:
            continue
        deviation = row["intent_deviation"]
        if isinstance(deviation, dict):
            ensure(turn_id)["intent_deviation"] = deviation

    for row in alternative_rows:
        turn_id = str(row["turn_id"])
        if turn_id not in refs:
            continue
        suggestion = _clean_text(row["suggestion"])
        if suggestion:
            ensure(turn_id).setdefault("alternative_utterances", []).append(suggestion)

    return evaluations


async def _record_session_read_audit(
    conn: Any,
    principal: Principal,
    *,
    target_kind: str,
    target_id: str,
    detail: dict[str, Any],
) -> None:
    if principal.role.value not in _SESSION_AUDIT_ROLES:
        return
    await conn.execute(
        """
        INSERT INTO audit.audit_log (
          actor_uid, action, target_kind, target_id, detail
        )
        VALUES ($1::uuid, $2, $3, $4, $5::jsonb)
        """,
        principal.user_id,
        "read_session",
        target_kind,
        target_id,
        detail,
    )


async def record_llm_call_audit(payload: dict[str, Any]) -> bool:
    """Append provider/token/cost metadata for an external LLM call.

    The audit table intentionally stores no prompt or completion text. A DB outage
    must not block the counseling loop, so failures are reported as False.

    dev degraded(무DB) 기동에서는 audit 인프라 자체가 없는 것이 정상 폴백이므로,
    호출부가 이를 "기록 실패"로 보고 정상 LLM 응답을 강등하지 않도록 True 를 반환한다.
    durable 환경에서 풀이 없거나 INSERT 가 실패한 경우에만 False 다.
    """
    try:
        get_pool()
    except Exception:
        return runtime_fallback_allowed()

    try:
        async with acquire(ai_context=True, ai_view="evaluator") as conn:
            await conn.execute(
                """
                INSERT INTO audit.llm_call_log (
                  session_id, turn_id, provider, model,
                  tokens_in, tokens_out, cost_usd,
                  inference_geo, latency_ms
                )
                VALUES (
                  $1::uuid, $2::uuid, $3, $4,
                  $5, $6, $7,
                  $8, $9
                )
                """,
                _clean_text(payload.get("session_id")),
                _clean_text(payload.get("turn_id")),
                _clean_text(payload.get("provider")),
                _clean_text(payload.get("model")),
                _safe_int(payload.get("tokens_in")),
                _safe_int(payload.get("tokens_out")),
                _safe_float(payload.get("cost_usd")),
                _clean_text(payload.get("inference_geo")),
                _safe_int(payload.get("latency_ms")),
            )
        return True
    except Exception:
        return False


def _state_from_row(row, card: PersonaCard) -> state_machine.SessionState:
    if row is None:
        return state_machine.init_state(
            params=card.openness_params(),
        )
    return state_machine.SessionState(
        stage=state_machine.Stage(row["stage"]),
        turn_seq=int(row["turn_seq"]),
        effective_openness=float(row["effective_openness"]),
        rapport_credit=float(row["rapport_credit"]),
        resistance=float(row["resistance"]),
        ideation_stage=int(row["ideation_stage"]),
        turns_in_stage=int(row["turns_in_stage"] or 0),
        affect_state=dict(row["affect_state"] or {}),
    )


def _turn_from_row(row, evaluation: dict[str, Any] | None = None) -> TurnRecord:
    created_at = _ts(row["created_at"]) or time.time()
    text_masked = (
        _clean_text(row["text_masked"]) or _clean_masked_text(row["text"]) or ""
    )
    return TurnRecord(
        turn_seq=int(row["seq"]),
        speaker=row["speaker"],
        stage=row["stage"],
        text=text_masked,
        text_masked=text_masked,
        turn_id=str(_row_value(row, "id"))
        if _row_value(row, "id") is not None
        else None,
        created_at=created_at,
        llm_provider=_row_value(row, "llm_provider"),
        model=_row_value(row, "model"),
        tokens_in=_row_value(row, "tokens_in"),
        tokens_out=_row_value(row, "tokens_out"),
        cost_usd=_row_value(row, "cost_usd"),
        audio_ref=_row_value(row, "audio_ref"),
        silence_ms=_row_value(row, "silence_ms"),
        speech_rate=_row_value(row, "speech_rate"),
        barge_in=_row_value(row, "barge_in"),
        provider_events=_dict_items(_row_value(row, "provider_events")),
        evaluation=evaluation,
        visible_to=tuple(_row_value(row, "visible_to") or DEFAULT_TURN_VISIBLE_TO),
    )


async def _persist_turn_evaluation(
    conn: Any, turn_id: str, evaluation: dict[str, Any] | None
) -> None:
    if not isinstance(evaluation, dict):
        return
    await conn.execute("SELECT set_config('app.ai_context', '1', true)")
    await conn.execute("SELECT set_config('app.current_ai_view', 'evaluator', true)")
    await conn.execute("SELECT set_config('app.current_sens_max', '2', true)")

    for row in _evaluation_feedback_rows(evaluation):
        await conn.execute(
            """
            INSERT INTO app.feedback_scores (
              turn_id, dimension, score, rationale, top1_score, loop
            )
            VALUES ($1::uuid, $2, $3, $4, $5, $6)
            ON CONFLICT (turn_id, dimension) DO UPDATE SET
              score = EXCLUDED.score,
              rationale = EXCLUDED.rationale,
              top1_score = EXCLUDED.top1_score,
              loop = EXCLUDED.loop
            """,
            turn_id,
            row["dimension"],
            row["score"],
            row["rationale"],
            row["top1_score"],
            row["loop"],
        )

    for row in _evaluation_technique_rows(evaluation):
        label_id = await conn.fetchval(
            """
            INSERT INTO app.technique_label_def (code, display_name, category)
            VALUES ($1, $2, $3)
            ON CONFLICT (code, version) DO UPDATE SET
              display_name = EXCLUDED.display_name,
              category = EXCLUDED.category,
              is_active = TRUE
            RETURNING label_id
            """,
            row["code"],
            row["label_ko"],
            row["category"],
        )
        await conn.execute(
            """
            INSERT INTO app.turn_technique (turn_id, label_id)
            VALUES ($1::uuid, $2)
            ON CONFLICT DO NOTHING
            """,
            turn_id,
            label_id,
        )

    for row in _evaluation_client_state_rows(evaluation):
        label_id = await conn.fetchval(
            """
            INSERT INTO app.client_state_def (code, display_name)
            VALUES ($1, $2)
            ON CONFLICT (code, version) DO UPDATE SET
              display_name = EXCLUDED.display_name,
              is_active = TRUE
            RETURNING label_id
            """,
            row["code"],
            row["label_ko"],
        )
        await conn.execute(
            """
            INSERT INTO app.turn_client_state (turn_id, label_id)
            VALUES ($1::uuid, $2)
            ON CONFLICT DO NOTHING
            """,
            turn_id,
            label_id,
        )

    for row in _evaluation_comment_rows(evaluation):
        await conn.execute(
            """
            INSERT INTO app.supervisor_comment (
              turn_id, kind, text, intent_deviation
            )
            VALUES ($1::uuid, $2, $3, $4::jsonb)
            """,
            turn_id,
            row["kind"],
            row["text"],
            row["intent_deviation"],
        )

    await conn.execute(
        "DELETE FROM app.alternative_utterance WHERE turn_id = $1::uuid", turn_id
    )
    for row in _evaluation_alternative_rows(evaluation):
        await conn.execute(
            """
            INSERT INTO app.alternative_utterance (
              turn_id, suggestion, rationale
            )
            VALUES ($1::uuid, $2, $3)
            """,
            turn_id,
            row["suggestion"],
            row["rationale"],
        )


async def _replace_turn_evaluation(
    conn: Any, turn_id: str, evaluation: dict[str, Any] | None
) -> None:
    await conn.execute("SELECT set_config('app.ai_context', '1', true)")
    await conn.execute("SELECT set_config('app.current_ai_view', 'evaluator', true)")
    await conn.execute("SELECT set_config('app.current_sens_max', '2', true)")
    await conn.execute(
        "DELETE FROM app.feedback_scores WHERE turn_id = $1::uuid", turn_id
    )
    await conn.execute(
        "DELETE FROM app.turn_technique WHERE turn_id = $1::uuid", turn_id
    )
    await conn.execute(
        "DELETE FROM app.turn_client_state WHERE turn_id = $1::uuid", turn_id
    )
    await conn.execute(
        "DELETE FROM app.supervisor_comment WHERE turn_id = $1::uuid", turn_id
    )
    await conn.execute(
        "DELETE FROM app.alternative_utterance WHERE turn_id = $1::uuid", turn_id
    )
    await _persist_turn_evaluation(conn, turn_id, _mask_json_text_values(evaluation))


async def replace_turn_evaluation(
    *, turn_id: str, evaluation: dict[str, Any] | None
) -> bool:
    if not turn_id or not isinstance(evaluation, dict):
        return False
    try:
        get_pool()
        async with acquire(ai_context=True, ai_view="evaluator") as conn:
            await _replace_turn_evaluation(conn, turn_id, evaluation)
        return True
    except Exception:
        require_runtime_fallback_allowed("turn evaluation replacement")
        return False


async def _load_turn_evaluations(
    conn: Any,
    turn_refs: list[tuple[str, int, str]],
) -> dict[str, dict[str, Any]]:
    turn_ids = [turn_id for turn_id, _, _ in turn_refs]
    if not turn_ids:
        return {}
    feedback_rows = await conn.fetch(
        """
        SELECT turn_id::text AS turn_id, dimension, score, rationale, top1_score, loop
        FROM app.feedback_scores
        WHERE turn_id = ANY($1::uuid[])
        ORDER BY created_at, dimension
        """,
        turn_ids,
    )
    technique_rows = await conn.fetch(
        """
        SELECT
          tt.turn_id::text AS turn_id,
          d.code,
          d.display_name AS label_ko,
          d.category
        FROM app.turn_technique tt
        JOIN app.technique_label_def d ON d.label_id = tt.label_id
        WHERE tt.turn_id = ANY($1::uuid[])
        ORDER BY tt.turn_id, d.code
        """,
        turn_ids,
    )
    client_state_rows = await conn.fetch(
        """
        SELECT
          ts.turn_id::text AS turn_id,
          d.code,
          d.display_name AS label_ko
        FROM app.turn_client_state ts
        JOIN app.client_state_def d ON d.label_id = ts.label_id
        WHERE ts.turn_id = ANY($1::uuid[])
        ORDER BY ts.turn_id, d.code
        """,
        turn_ids,
    )
    comment_rows = await conn.fetch(
        """
        SELECT turn_id::text AS turn_id, intent_deviation
        FROM app.supervisor_comment
        WHERE turn_id = ANY($1::uuid[])
          AND intent_deviation IS NOT NULL
        ORDER BY created_at
        """,
        turn_ids,
    )
    alternative_rows = await conn.fetch(
        """
        SELECT turn_id::text AS turn_id, suggestion, rationale
        FROM app.alternative_utterance
        WHERE turn_id = ANY($1::uuid[])
        ORDER BY turn_id, created_at, id
        """,
        turn_ids,
    )
    return _rebuild_turn_evaluations(
        turn_refs,
        feedback_rows=feedback_rows,
        technique_rows=technique_rows,
        client_state_rows=client_state_rows,
        comment_rows=comment_rows,
        alternative_rows=alternative_rows,
    )


async def _hydrate_session_turn_evaluations(sess: InProcSession) -> None:
    await _hydrate_sessions_turn_evaluations([sess])


async def _hydrate_sessions_turn_evaluations(sessions: list[InProcSession]) -> None:
    """여러 세션의 턴 평가를 evaluator-view 연결 1회로 배치 하이드레이트(세션당 N+1 제거)."""
    turn_refs = [
        (turn.turn_id, turn.turn_seq, turn.stage)
        for sess in sessions
        for turn in sess.turns
        if turn.turn_id is not None
    ]
    if not turn_refs:
        return
    async with acquire(ai_view="evaluator") as conn:
        evaluations = await _load_turn_evaluations(conn, turn_refs)
    for sess in sessions:
        for turn in sess.turns:
            if turn.turn_id and turn.turn_id in evaluations:
                turn.evaluation = evaluations[turn.turn_id]


async def ensure_review_tables() -> None:
    """Verify the review schema, with DDL repair restricted to local development."""
    try:
        get_pool()
        async with acquire(role="admin") as conn:
            ready = await schema_contract_ready(conn, REVIEW_SCHEMA_CONTRACT)
            if not runtime_schema_bootstrap_required(
                REVIEW_SCHEMA_CONTRACT, ready=ready
            ):
                return
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.session_evaluation (
                  session_id UUID PRIMARY KEY REFERENCES app.sessions(id) ON DELETE CASCADE,
                  status TEXT NOT NULL CHECK (status IN ('ready','degraded','error')),
                  source TEXT NOT NULL,
                  scope TEXT NOT NULL,
                  stage TEXT NOT NULL,
                  payload JSONB NOT NULL DEFAULT '{}'::jsonb,
                  error TEXT,
                  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
                )
                """
            )
            await conn.execute(
                """
                ALTER TABLE app.session_evaluation ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_session_evaluation_select ON app.session_evaluation;
                DROP POLICY IF EXISTS p_session_evaluation_insert ON app.session_evaluation;
                DROP POLICY IF EXISTS p_session_evaluation_update ON app.session_evaluation;
                DROP POLICY IF EXISTS p_session_evaluation_delete ON app.session_evaluation;

                CREATE POLICY p_session_evaluation_select
                  ON app.session_evaluation FOR SELECT USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_evaluation.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_session_evaluation_insert
                  ON app.session_evaluation FOR INSERT WITH CHECK (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_evaluation.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_session_evaluation_update
                  ON app.session_evaluation FOR UPDATE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_evaluation.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  ) WITH CHECK (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_evaluation.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_session_evaluation_delete
                  ON app.session_evaluation FOR DELETE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_evaluation.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  )
                """
            )
            await conn.execute(
                """
                ALTER TABLE app.feedback_scores ENABLE ROW LEVEL SECURITY;
                ALTER TABLE app.turn_technique ENABLE ROW LEVEL SECURITY;
                ALTER TABLE app.turn_client_state ENABLE ROW LEVEL SECURITY;
                ALTER TABLE app.supervisor_comment ENABLE ROW LEVEL SECURITY;
                ALTER TABLE app.alternative_utterance ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_feedback_delete ON app.feedback_scores;
                DROP POLICY IF EXISTS p_turn_technique_delete ON app.turn_technique;
                DROP POLICY IF EXISTS p_turn_client_state_delete ON app.turn_client_state;
                DROP POLICY IF EXISTS p_supervisor_comment_delete ON app.supervisor_comment;
                DROP POLICY IF EXISTS p_alternative_utterance_delete ON app.alternative_utterance;

                CREATE POLICY p_feedback_delete
                  ON app.feedback_scores FOR DELETE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_turn_technique_delete
                  ON app.turn_technique FOR DELETE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_turn_client_state_delete
                  ON app.turn_client_state FOR DELETE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_supervisor_comment_delete
                  ON app.supervisor_comment FOR DELETE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_alternative_utterance_delete
                  ON app.alternative_utterance FOR DELETE USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                  )
                """
            )
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.case_worksheet (
                  session_id UUID PRIMARY KEY REFERENCES app.sessions(id) ON DELETE CASCADE,
                  payload JSONB NOT NULL DEFAULT '{}'::jsonb,
                  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
                )
                """
            )
            await conn.execute(
                """
                ALTER TABLE app.case_worksheet ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_case_worksheet_select ON app.case_worksheet;
                DROP POLICY IF EXISTS p_case_worksheet_insert ON app.case_worksheet;
                DROP POLICY IF EXISTS p_case_worksheet_update ON app.case_worksheet;
                DROP POLICY IF EXISTS p_case_worksheet_delete ON app.case_worksheet;

                CREATE POLICY p_case_worksheet_select
                  ON app.case_worksheet FOR SELECT USING (
                    app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.case_worksheet.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_case_worksheet_insert
                  ON app.case_worksheet FOR INSERT WITH CHECK (
                    EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.case_worksheet.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_case_worksheet_update
                  ON app.case_worksheet FOR UPDATE USING (
                    EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.case_worksheet.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  ) WITH CHECK (
                    EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.case_worksheet.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_case_worksheet_delete
                  ON app.case_worksheet FOR DELETE USING (
                    EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.case_worksheet.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  )
                """
            )
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.live_coach_events (
                  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                  session_id UUID NOT NULL REFERENCES app.sessions(id) ON DELETE CASCADE,
                  turn_seq INT NOT NULL CHECK (turn_seq >= 1),
                  stage TEXT NOT NULL,
                  event_type TEXT NOT NULL DEFAULT 'use',
                  credit_delta INT NOT NULL DEFAULT -1,
                  credit_balance INT,
                  reason TEXT,
                  learner_text_excerpt TEXT,
                  client_reply_excerpt TEXT,
                  payload JSONB NOT NULL DEFAULT '{}'::jsonb,
                  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
                );
                ALTER TABLE app.live_coach_events
                  ADD COLUMN IF NOT EXISTS event_type TEXT NOT NULL DEFAULT 'use';
                ALTER TABLE app.live_coach_events
                  ADD COLUMN IF NOT EXISTS credit_delta INT NOT NULL DEFAULT -1;
                ALTER TABLE app.live_coach_events
                  ADD COLUMN IF NOT EXISTS credit_balance INT;
                ALTER TABLE app.live_coach_events
                  ADD COLUMN IF NOT EXISTS reason TEXT;
                UPDATE app.live_coach_events
                   SET event_type = 'use'
                 WHERE event_type IS NULL OR event_type NOT IN ('use', 'recharge');
                UPDATE app.live_coach_events
                   SET credit_delta = CASE WHEN event_type = 'recharge' THEN 1 ELSE -1 END
                 WHERE credit_delta IS NULL;
                DO $$
                BEGIN
                  IF NOT EXISTS (
                    SELECT 1
                    FROM pg_constraint
                    WHERE conname = 'live_coach_events_event_type_check'
                  ) THEN
                    ALTER TABLE app.live_coach_events
                      ADD CONSTRAINT live_coach_events_event_type_check
                      CHECK (event_type IN ('use', 'recharge'));
                  END IF;
                END $$;
                CREATE INDEX IF NOT EXISTS idx_live_coach_events_session_turn
                  ON app.live_coach_events(session_id, turn_seq, created_at);
                CREATE INDEX IF NOT EXISTS idx_live_coach_events_session_type
                  ON app.live_coach_events(session_id, event_type, created_at);

                ALTER TABLE app.live_coach_events ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_live_coach_events_select ON app.live_coach_events;
                DROP POLICY IF EXISTS p_live_coach_events_insert ON app.live_coach_events;

                CREATE POLICY p_live_coach_events_select
                  ON app.live_coach_events FOR SELECT USING (
                    app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.live_coach_events.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_live_coach_events_insert
                  ON app.live_coach_events FOR INSERT WITH CHECK (
                    EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.live_coach_events.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  )
                """
            )
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.safety_events (
                  id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
                  session_id UUID REFERENCES app.sessions(id) ON DELETE CASCADE,
                  turn_id UUID REFERENCES app.turns(id) ON DELETE SET NULL,
                  trigger_type TEXT NOT NULL,
                  ko_risk_level SMALLINT,
                  escalated BOOLEAN NOT NULL DEFAULT FALSE,
                  latency_ms INT,
                  detail JSONB,
                  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
                );
                CREATE INDEX IF NOT EXISTS idx_safety_session
                  ON app.safety_events(session_id);

                ALTER TABLE app.safety_events ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_safety_events_select ON app.safety_events;
                DROP POLICY IF EXISTS p_safety_events_insert ON app.safety_events;

                CREATE POLICY p_safety_events_select
                  ON app.safety_events FOR SELECT USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.safety_events.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_safety_events_insert
                  ON app.safety_events FOR INSERT WITH CHECK (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.safety_events.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  )
                """
            )
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.session_review_status (
                  session_id UUID PRIMARY KEY REFERENCES app.sessions(id) ON DELETE CASCADE,
                  reviewer_id UUID REFERENCES app.app_user(user_id),
                  status TEXT NOT NULL DEFAULT 'pending'
                    CHECK (status IN ('pending','viewed','closed')),
                  note TEXT NOT NULL DEFAULT '',
                  worksheet_status TEXT NOT NULL DEFAULT 'pending',
                  worksheet_note TEXT NOT NULL DEFAULT '',
                  worksheet_reviewed_at TIMESTAMPTZ,
                  reviewed_at TIMESTAMPTZ,
                  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
                )
                """
            )
            await conn.execute(
                """
                ALTER TABLE app.session_review_status
                  ADD COLUMN IF NOT EXISTS worksheet_status TEXT NOT NULL DEFAULT 'pending';
                ALTER TABLE app.session_review_status
                  ADD COLUMN IF NOT EXISTS worksheet_note TEXT NOT NULL DEFAULT '';
                ALTER TABLE app.session_review_status
                  ADD COLUMN IF NOT EXISTS worksheet_reviewed_at TIMESTAMPTZ;
                """
            )
            await conn.execute(
                """
                ALTER TABLE app.session_review_status ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_session_review_status_select ON app.session_review_status;
                DROP POLICY IF EXISTS p_session_review_status_insert ON app.session_review_status;
                DROP POLICY IF EXISTS p_session_review_status_update ON app.session_review_status;
                DROP POLICY IF EXISTS p_session_review_status_delete ON app.session_review_status;

                CREATE POLICY p_session_review_status_select
                  ON app.session_review_status FOR SELECT USING (
                    app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_session_review_status_insert
                  ON app.session_review_status FOR INSERT WITH CHECK (
                    app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_session_review_status_update
                  ON app.session_review_status FOR UPDATE USING (
                    app.current_role_name() IN ('admin','instructor')
                  ) WITH CHECK (
                    app.current_role_name() IN ('admin','instructor')
                  );
                CREATE POLICY p_session_review_status_delete
                  ON app.session_review_status FOR DELETE USING (
                    app.current_role_name() IN ('admin','instructor')
                  )
                """
            )
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.session_share_link (
                  session_id UUID PRIMARY KEY REFERENCES app.sessions(id) ON DELETE CASCADE,
                  created_by UUID NOT NULL REFERENCES app.app_user(user_id),
                  token_hash TEXT NOT NULL UNIQUE,
                  payload JSONB NOT NULL,
                  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                  revoked_at TIMESTAMPTZ
                );
                CREATE INDEX IF NOT EXISTS idx_session_share_token_active
                  ON app.session_share_link(token_hash)
                  WHERE revoked_at IS NULL;
                ALTER TABLE app.session_share_link ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_session_share_select ON app.session_share_link;
                DROP POLICY IF EXISTS p_session_share_insert ON app.session_share_link;
                DROP POLICY IF EXISTS p_session_share_update ON app.session_share_link;
                DROP POLICY IF EXISTS p_session_share_delete ON app.session_share_link;

                CREATE POLICY p_session_share_select
                  ON app.session_share_link FOR SELECT USING (
                    app.is_ai_context()
                    OR app.current_role_name() IN ('admin','instructor')
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_share_link.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_session_share_insert
                  ON app.session_share_link FOR INSERT WITH CHECK (
                    app.is_ai_context()
                    OR (
                      created_by = app.current_uid()
                      AND EXISTS (
                        SELECT 1 FROM app.sessions s
                        WHERE s.id = app.session_share_link.session_id
                          AND s.learner_id = app.current_uid()
                      )
                    )
                  );
                CREATE POLICY p_session_share_update
                  ON app.session_share_link FOR UPDATE USING (
                    app.is_ai_context()
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_share_link.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  ) WITH CHECK (
                    app.is_ai_context()
                    OR (
                      created_by = app.current_uid()
                      AND EXISTS (
                        SELECT 1 FROM app.sessions s
                        WHERE s.id = app.session_share_link.session_id
                          AND s.learner_id = app.current_uid()
                      )
                    )
                  );
                CREATE POLICY p_session_share_delete
                  ON app.session_share_link FOR DELETE USING (
                    app.is_ai_context()
                    OR EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_share_link.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                """
            )
            await conn.execute(
                """
                CREATE TABLE IF NOT EXISTS app.session_archive_state (
                  session_id UUID PRIMARY KEY REFERENCES app.sessions(id) ON DELETE CASCADE,
                  learner_id UUID NOT NULL REFERENCES app.app_user(user_id),
                  archived_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
                );
                CREATE INDEX IF NOT EXISTS idx_session_archive_learner
                  ON app.session_archive_state(learner_id, archived_at DESC);

                ALTER TABLE app.session_archive_state ENABLE ROW LEVEL SECURITY;

                DROP POLICY IF EXISTS p_session_archive_select ON app.session_archive_state;
                DROP POLICY IF EXISTS p_session_archive_insert ON app.session_archive_state;
                DROP POLICY IF EXISTS p_session_archive_update ON app.session_archive_state;
                DROP POLICY IF EXISTS p_session_archive_delete ON app.session_archive_state;

                CREATE POLICY p_session_archive_select
                  ON app.session_archive_state FOR SELECT USING (
                    app.current_role_name() IN ('admin','instructor')
                    OR learner_id = app.current_uid()
                  );
                CREATE POLICY p_session_archive_insert
                  ON app.session_archive_state FOR INSERT WITH CHECK (
                    learner_id = app.current_uid()
                    AND EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_archive_state.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_session_archive_update
                  ON app.session_archive_state FOR UPDATE USING (
                    learner_id = app.current_uid()
                  ) WITH CHECK (
                    learner_id = app.current_uid()
                    AND EXISTS (
                      SELECT 1 FROM app.sessions s
                      WHERE s.id = app.session_archive_state.session_id
                        AND s.learner_id = app.current_uid()
                    )
                  );
                CREATE POLICY p_session_archive_delete
                  ON app.session_archive_state FOR DELETE USING (
                    learner_id = app.current_uid()
                  )
                """
            )
            if not await schema_contract_ready(conn, REVIEW_SCHEMA_CONTRACT):
                raise RuntimeError(
                    "review/evaluation development schema bootstrap did not satisfy readiness"
                )
    except Exception:
        if settings.environment != "dev":
            raise
        return


async def save_session_evaluation(write: SessionEvaluationWrite) -> bool:
    write.payload = _mask_json_text_values(write.payload)
    write.error = _clean_masked_text(write.error)
    record = write.cache_record()
    if runtime_fallback_allowed():
        _EVALUATION_CACHE[write.session_id] = record
    try:
        get_pool()
        async with acquire(role="learner", user_id=write.learner_id) as conn:
            await conn.execute(
                """
                INSERT INTO app.session_evaluation (
                  session_id, status, source, scope, stage, payload, error,
                  created_at, updated_at
                )
                VALUES ($1::uuid, $2, $3, $4, $5, $6::jsonb, $7, now(), now())
                ON CONFLICT (session_id) DO UPDATE SET
                  status = EXCLUDED.status,
                  source = EXCLUDED.source,
                  scope = EXCLUDED.scope,
                  stage = EXCLUDED.stage,
                  payload = EXCLUDED.payload,
                  error = EXCLUDED.error,
                  updated_at = now()
                """,
                write.session_id,
                write.status,
                write.source,
                write.scope,
                write.stage,
                write.payload,
                write.error,
            )
        return True
    except Exception:
        require_runtime_fallback_allowed("session evaluation")
        return False


async def load_session_evaluation(
    session_id: str,
    principal: Principal,
) -> tuple[dict[str, Any] | None, bool]:
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            row = await conn.fetchrow(
                """
                SELECT status, source, scope, stage, payload, error, updated_at
                FROM app.session_evaluation
                WHERE session_id = $1::uuid
                """,
                session_id,
            )
            if row is None:
                cached = (
                    _EVALUATION_CACHE.get(session_id)
                    if runtime_fallback_allowed()
                    else None
                )
                return cached, cached is None
            return {
                "status": row["status"],
                "source": row["source"],
                "scope": row["scope"],
                "stage": row["stage"],
                "payload": dict(row["payload"] or {}),
                "error": row["error"],
                "updated_at": _ts(row["updated_at"]),
            }, True
    except Exception:
        require_runtime_fallback_allowed("session evaluation")
        return _EVALUATION_CACHE.get(session_id), False


async def list_session_evaluations(
    session_ids: list[str],
    principal: Principal,
) -> tuple[dict[str, dict[str, Any]], bool]:
    if not session_ids:
        return {}, True
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                """
                SELECT session_id::text AS session_id, status, source, scope, stage, payload, error, updated_at
                FROM app.session_evaluation
                WHERE session_id = ANY($1::uuid[])
                """,
                session_ids,
            )
            records = {
                str(row["session_id"]): {
                    "status": row["status"],
                    "source": row["source"],
                    "scope": row["scope"],
                    "stage": row["stage"],
                    "payload": dict(row["payload"] or {}),
                    "error": row["error"],
                    "updated_at": _ts(row["updated_at"]),
                }
                for row in rows
            }
            cached_used = False
            if runtime_fallback_allowed():
                for session_id in session_ids:
                    if session_id not in records and session_id in _EVALUATION_CACHE:
                        records[session_id] = _EVALUATION_CACHE[session_id]
                        cached_used = True
            return records, not cached_used
    except Exception:
        require_runtime_fallback_allowed("session evaluation")
        return {
            session_id: _EVALUATION_CACHE[session_id]
            for session_id in session_ids
            if session_id in _EVALUATION_CACHE
        }, False


async def _fetch_session_runtime_rows(
    conn: Any,
    session_ids: list[str],
) -> tuple[dict[str, Any], dict[str, list[Any]]]:
    """세션 상태와 턴을 두 번의 배치 조회로 적재한다."""
    if not session_ids:
        return {}, {}
    state_rows = await conn.fetch(
        f"""
        SELECT {SESSION_STATE_SELECT_COLUMNS_SQL}
        FROM app.session_state
        WHERE session_id = ANY($1::uuid[])
        """,
        session_ids,
    )
    states_by_id = {
        str(state_row["session_id"]): state_row for state_row in state_rows
    }
    turn_rows = await conn.fetch(
        f"""
        SELECT {SESSION_TURN_SELECT_COLUMNS_SQL}
        FROM app.turns
        WHERE session_id = ANY($1::uuid[])
        ORDER BY session_id, seq
        """,
        session_ids,
    )
    turns_by_id: dict[str, list[Any]] = {}
    for turn_row in turn_rows:
        turns_by_id.setdefault(str(turn_row["session_id"]), []).append(turn_row)
    return states_by_id, turns_by_id


async def list_sessions_missing_session_evaluation(
    *,
    older_than_seconds: float,
    limit: int,
) -> tuple[list[InProcSession], bool]:
    if limit <= 0:
        return [], True
    try:
        get_pool()
        stale_seconds = max(float(older_than_seconds), 0.0)
        async with acquire(ai_context=True, ai_view="evaluator") as conn:
            rows = await conn.fetch(
                f"""
                SELECT {SESSION_PERSONA_SELECT_COLUMNS_SQL}
                FROM app.sessions s
                LEFT JOIN app.session_evaluation se ON se.session_id = s.id
                {SESSION_PERSONA_JOINS_SQL}
                WHERE s.ended_at IS NOT NULL
                  AND s.ended_at <= now() - ($1::double precision * interval '1 second')
                  AND se.session_id IS NULL
                  AND EXISTS (
                    SELECT 1
                    FROM app.turns t
                    WHERE t.session_id = s.id
                      AND 'client' = ANY(t.visible_to)
                  )
                ORDER BY s.ended_at ASC
                LIMIT $2
                """,
                stale_seconds,
                max(1, int(limit)),
            )
            session_ids = [str(row["id"]) for row in rows]
            states_by_id, turns_by_id = await _fetch_session_runtime_rows(
                conn, session_ids
            )
            sessions: list[InProcSession] = []
            for row in rows:
                session_id = str(row["id"])
                sess = _session_from_rows(
                    row,
                    states_by_id.get(session_id),
                    turns_by_id.get(session_id, []),
                )
                if sess is not None:
                    sessions.append(sess)
            return sessions, True
    except Exception:
        require_runtime_fallback_allowed("missing session evaluation recovery")
        return [], False


async def save_case_worksheet(
    *,
    session_id: str,
    learner_id: str,
    payload: dict[str, Any],
) -> bool:
    if runtime_fallback_allowed():
        _CASE_WORKSHEET_CACHE[session_id] = payload
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            await conn.execute(
                """
                INSERT INTO app.case_worksheet (session_id, payload, created_at, updated_at)
                VALUES ($1::uuid, $2::jsonb, now(), now())
                ON CONFLICT (session_id) DO UPDATE SET
                  payload = EXCLUDED.payload,
                  updated_at = now()
                """,
                session_id,
                payload,
            )
        return True
    except Exception:
        require_runtime_fallback_allowed("case worksheet save")
        return False


async def load_case_worksheet(
    session_id: str,
    principal: Principal,
) -> tuple[dict[str, Any] | None, bool]:
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            row = await conn.fetchrow(
                """
                SELECT payload, updated_at
                FROM app.case_worksheet
                WHERE session_id = $1::uuid
                """,
                session_id,
            )
            if row is None:
                return (
                    _CASE_WORKSHEET_CACHE.get(session_id)
                    if runtime_fallback_allowed()
                    else None
                ), False
            payload = dict(row["payload"] or {})
            payload.setdefault("savedAt", _iso_dt(row["updated_at"]))
            return payload, True
    except Exception:
        require_runtime_fallback_allowed("case worksheet load")
        return _CASE_WORKSHEET_CACHE.get(session_id), False


async def save_live_coach_event(
    *,
    session_id: str,
    learner_id: str,
    turn_seq: int,
    stage: str,
    learner_text: str,
    client_reply: str | None,
    suggestion: Any,
) -> tuple[dict[str, Any] | None, bool]:
    cached_remaining = _live_coach_cached_remaining(session_id)
    if cached_remaining <= 0:
        raise LiveCoachCreditExhausted("live coach credit exhausted")
    record = _live_coach_cache_record(
        session_id=session_id,
        turn_seq=turn_seq,
        stage=stage,
        learner_text=learner_text,
        client_reply=client_reply,
        suggestion=suggestion,
        credit_balance=cached_remaining - 1,
    )
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            remaining_before = await _live_coach_remaining_for_conn(conn, session_id)
            if remaining_before <= 0:
                raise LiveCoachCreditExhausted("live coach credit exhausted")
            remaining_after = _live_coach_remaining(remaining_before - 1)
            row = await conn.fetchrow(
                """
                INSERT INTO app.live_coach_events (
                  session_id, turn_seq, stage, event_type, credit_delta,
                  credit_balance, reason, learner_text_excerpt,
                  client_reply_excerpt, payload, created_at
                )
                VALUES ($1::uuid, $2, $3, 'use', -1, $4, $5, $6, $7, $8::jsonb, now())
                RETURNING id, session_id, turn_seq, stage, event_type, credit_delta,
                          credit_balance, reason, learner_text_excerpt,
                          client_reply_excerpt, payload, created_at
                """,
                session_id,
                int(turn_seq),
                stage,
                remaining_after,
                LIVE_COACH_USE_REASON,
                record.get("learner_text_excerpt"),
                record.get("client_reply_excerpt"),
                _model_payload(suggestion),
            )
            return (_live_coach_event_from_row(row) if row else None), True
    except LiveCoachCreditExhausted:
        raise
    except Exception:
        require_runtime_fallback_allowed("live coach event save")
        _LIVE_COACH_EVENT_CACHE.setdefault(session_id, []).append(record)
        return record, False


async def list_live_coach_events(
    session_id: str,
    principal: Principal,
) -> tuple[list[dict[str, Any]], bool]:
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                """
                SELECT id, session_id, turn_seq, stage, learner_text_excerpt,
                       client_reply_excerpt, payload, created_at
                FROM app.live_coach_events
                WHERE session_id = $1::uuid
                  AND event_type = 'use'
                ORDER BY created_at ASC, turn_seq ASC
                """,
                session_id,
            )
            return [_live_coach_event_from_row(row) for row in rows], True
    except Exception:
        require_runtime_fallback_allowed("live coach event list")
        return [
            dict(item)
            for item in _LIVE_COACH_EVENT_CACHE.get(session_id, [])
            if item.get("event_type", "use") == "use"
        ], False


async def get_live_coach_quota(
    session_id: str,
    principal: Principal,
) -> tuple[dict[str, int], bool]:
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            remaining = await _live_coach_remaining_for_conn(conn, session_id)
            return _live_coach_quota_payload(remaining), True
    except Exception:
        require_runtime_fallback_allowed("live coach quota")
        return _live_coach_quota_payload(
            _live_coach_cached_remaining(session_id)
        ), False


async def list_live_coach_credit_events(
    session_id: str,
    principal: Principal,
) -> tuple[list[dict[str, Any]], bool]:
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                """
                SELECT id, session_id, turn_seq, stage, event_type, credit_delta,
                       credit_balance, reason, created_at
                FROM app.live_coach_events
                WHERE session_id = $1::uuid
                ORDER BY created_at ASC, turn_seq ASC
                """,
                session_id,
            )
            return [_live_coach_credit_event_from_row(row) for row in rows], True
    except Exception:
        require_runtime_fallback_allowed("live coach credit event list")
        return [
            _live_coach_credit_event_from_record(item)
            for item in _LIVE_COACH_EVENT_CACHE.get(session_id, [])
        ], False


async def record_live_coach_recharge(
    *,
    session_id: str,
    learner_id: str,
    turn_seq: int,
    stage: str,
    reason: str = LIVE_COACH_RECHARGE_REASON,
) -> tuple[dict[str, Any] | None, bool]:
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            existing = await conn.fetchval(
                """
                SELECT id
                FROM app.live_coach_events
                WHERE session_id = $1::uuid
                  AND turn_seq = $2
                  AND event_type = 'recharge'
                LIMIT 1
                """,
                session_id,
                int(turn_seq),
            )
            if existing:
                return None, True
            remaining_before = await _live_coach_remaining_for_conn(conn, session_id)
            if remaining_before >= LIVE_COACH_MAX_CREDITS:
                return None, True
            remaining_after = _live_coach_remaining(remaining_before + 1)
            row = await conn.fetchrow(
                """
                INSERT INTO app.live_coach_events (
                  session_id, turn_seq, stage, event_type, credit_delta,
                  credit_balance, reason, payload, created_at
                )
                VALUES ($1::uuid, $2, $3, 'recharge', 1, $4, $5, '{}'::jsonb, now())
                RETURNING id, session_id, turn_seq, stage, event_type, credit_delta,
                          credit_balance, reason, created_at
                """,
                session_id,
                int(turn_seq),
                stage,
                remaining_after,
                reason,
            )
            return (_live_coach_credit_event_from_row(row) if row else None), True
    except Exception:
        require_runtime_fallback_allowed("live coach recharge")
        events = _LIVE_COACH_EVENT_CACHE.setdefault(session_id, [])
        if any(
            event.get("event_type") == "recharge"
            and int(event.get("turn_seq") or 0) == int(turn_seq)
            for event in events
        ):
            return None, False
        remaining_before = _live_coach_cached_remaining(session_id)
        if remaining_before >= LIVE_COACH_MAX_CREDITS:
            return None, False
        record = _live_coach_recharge_cache_record(
            session_id=session_id,
            turn_seq=turn_seq,
            stage=stage,
            credit_balance=remaining_before + 1,
            reason=reason,
        )
        events.append(record)
        return _live_coach_credit_event_from_record(record), False


def _review_status_from_row(row: Any) -> dict[str, Any]:
    return {
        "session_id": str(row["session_id"]),
        "reviewer_id": str(row["reviewer_id"] or ""),
        "status": str(row["status"] or "pending"),
        "note": str(row["note"] or ""),
        "worksheet_status": _worksheet_review_status(row["worksheet_status"]),
        "worksheet_note": str(row["worksheet_note"] or ""),
        "worksheet_reviewed_at": _iso_dt(row["worksheet_reviewed_at"]),
        "reviewed_at": _iso_dt(row["reviewed_at"]),
        "updated_at": _iso_dt(row["updated_at"]),
    }


def _worksheet_review_status(value: object) -> str:
    raw = str(value or "pending")
    if raw in _WORKSHEET_REVIEW_STATUS_VALUES:
        return raw
    return "pending"


def _review_status_cache_record(
    *,
    session_id: str,
    reviewer_id: str,
    status: str,
    note: str,
    worksheet_status: str | None = None,
    worksheet_note: str | None = None,
) -> dict[str, Any]:
    now = datetime.now(timezone.utc)
    previous = _SESSION_REVIEW_STATUS_CACHE.get(session_id) or {}
    reviewed_at = previous.get("reviewed_at")
    if status == "closed" and not reviewed_at:
        reviewed_at = now.isoformat().replace("+00:00", "Z")
    if status != "closed":
        reviewed_at = None
    worksheet_reviewed_at = previous.get("worksheet_reviewed_at")
    if worksheet_status is None:
        worksheet_status = _worksheet_review_status(previous.get("worksheet_status"))
        worksheet_note = str(previous.get("worksheet_note") or "")
    else:
        worksheet_status = _worksheet_review_status(worksheet_status)
        worksheet_note = str(worksheet_note or "").strip()
        if worksheet_status == "pending":
            worksheet_reviewed_at = None
        else:
            worksheet_reviewed_at = now.isoformat().replace("+00:00", "Z")
    return {
        "session_id": session_id,
        "reviewer_id": reviewer_id,
        "status": status,
        "note": note,
        "worksheet_status": worksheet_status,
        "worksheet_note": worksheet_note,
        "worksheet_reviewed_at": worksheet_reviewed_at,
        "reviewed_at": reviewed_at,
        "updated_at": now.isoformat().replace("+00:00", "Z"),
    }


async def list_session_review_statuses(
    session_ids: list[str],
    principal: Principal,
) -> tuple[dict[str, dict[str, Any]], bool]:
    if not session_ids:
        return {}, True
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                """
                SELECT session_id, reviewer_id, status, note, worksheet_status,
                       worksheet_note, worksheet_reviewed_at, reviewed_at, updated_at
                FROM app.session_review_status
                WHERE session_id = ANY($1::uuid[])
                """,
                session_ids,
            )
            return {
                str(row["session_id"]): _review_status_from_row(row) for row in rows
            }, True
    except Exception:
        require_runtime_fallback_allowed("session review status list")
        return {
            session_id: dict(record)
            for session_id in session_ids
            if (record := _SESSION_REVIEW_STATUS_CACHE.get(session_id)) is not None
        }, False


async def load_session_review_status(
    session_id: str,
    principal: Principal,
) -> tuple[dict[str, Any] | None, bool]:
    statuses, durable = await list_session_review_statuses([session_id], principal)
    return statuses.get(session_id), durable


async def save_session_review_status(
    *,
    session_id: str,
    reviewer_id: str,
    status: str,
    note: str,
    principal: Principal,
    worksheet_status: str | None = None,
    worksheet_note: str | None = None,
) -> tuple[dict[str, Any] | None, bool]:
    note = note.strip()
    if worksheet_status is not None:
        worksheet_note = str(worksheet_note or "").strip()
    if runtime_fallback_allowed():
        _SESSION_REVIEW_STATUS_CACHE[session_id] = _review_status_cache_record(
            session_id=session_id,
            reviewer_id=reviewer_id,
            status=status,
            note=note,
            worksheet_status=worksheet_status,
            worksheet_note=worksheet_note,
        )
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            row = await conn.fetchrow(
                """
                INSERT INTO app.session_review_status (
                  session_id, reviewer_id, status, note, worksheet_status,
                  worksheet_note, worksheet_reviewed_at, reviewed_at, updated_at
                )
                VALUES (
                  $1::uuid, $2::uuid, $3, $4,
                  COALESCE($5, 'pending'),
                  CASE WHEN $5::text IS NULL THEN '' ELSE COALESCE($6, '') END,
                  CASE
                    WHEN $5::text IS NULL OR $5 = 'pending' THEN NULL
                    ELSE now()
                  END,
                  CASE WHEN $3 = 'closed' THEN now() ELSE NULL END,
                  now()
                )
                ON CONFLICT (session_id) DO UPDATE SET
                  reviewer_id = EXCLUDED.reviewer_id,
                  status = EXCLUDED.status,
                  note = EXCLUDED.note,
                  worksheet_status = CASE
                    WHEN $5::text IS NULL
                      THEN app.session_review_status.worksheet_status
                    ELSE $5
                  END,
                  worksheet_note = CASE
                    WHEN $5::text IS NULL
                      THEN app.session_review_status.worksheet_note
                    ELSE COALESCE($6, '')
                  END,
                  worksheet_reviewed_at = CASE
                    WHEN $5::text IS NULL
                      THEN app.session_review_status.worksheet_reviewed_at
                    WHEN $5 = 'pending'
                      THEN NULL
                    ELSE now()
                  END,
                  reviewed_at = CASE
                    WHEN EXCLUDED.status = 'closed'
                      THEN COALESCE(app.session_review_status.reviewed_at, now())
                    ELSE NULL
                  END,
                  updated_at = now()
                RETURNING session_id, reviewer_id, status, note, worksheet_status,
                          worksheet_note, worksheet_reviewed_at, reviewed_at, updated_at
                """,
                session_id,
                reviewer_id,
                status,
                note,
                worksheet_status,
                worksheet_note,
            )
            return (_review_status_from_row(row) if row else None), True
    except Exception:
        require_runtime_fallback_allowed("session review status save")
        return _SESSION_REVIEW_STATUS_CACHE.get(session_id), False


async def save_session_share(
    *,
    session_id: str,
    learner_id: str,
    token_hash: str,
    payload: dict[str, Any],
) -> dict[str, Any] | None:
    record = {
        "session_id": session_id,
        "payload": payload,
        "created_at": time.time(),
        "updated_at": time.time(),
        "revoked_at": None,
    }
    if runtime_fallback_allowed():
        old = _SESSION_SHARE_CACHE.get(session_id)
        if old:
            _SESSION_SHARE_TOKEN_INDEX.pop(str(old.get("token_hash") or ""), None)
        _SESSION_SHARE_CACHE[session_id] = {**record, "token_hash": token_hash}
        _SESSION_SHARE_TOKEN_INDEX[token_hash] = session_id
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            row = await conn.fetchrow(
                """
                INSERT INTO app.session_share_link (
                  session_id, created_by, token_hash, payload, created_at, updated_at, revoked_at
                )
                VALUES ($1::uuid, $2::uuid, $3, $4::jsonb, now(), now(), NULL)
                ON CONFLICT (session_id) DO UPDATE SET
                  created_by = EXCLUDED.created_by,
                  token_hash = EXCLUDED.token_hash,
                  payload = EXCLUDED.payload,
                  updated_at = now(),
                  revoked_at = NULL
                RETURNING session_id, payload, created_at, updated_at, revoked_at
                """,
                session_id,
                learner_id,
                token_hash,
                payload,
            )
            return _share_record_from_row(row) if row is not None else None
    except Exception:
        require_runtime_fallback_allowed("session share save")
        return record


async def revoke_session_share(*, session_id: str, learner_id: str) -> bool:
    if runtime_fallback_allowed():
        old = _SESSION_SHARE_CACHE.pop(session_id, None)
        if old:
            _SESSION_SHARE_TOKEN_INDEX.pop(str(old.get("token_hash") or ""), None)
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            status = await conn.execute(
                """
                UPDATE app.session_share_link
                SET revoked_at = now(), updated_at = now()
                WHERE session_id = $1::uuid
                """,
                session_id,
            )
            return status != "UPDATE 0"
    except Exception:
        require_runtime_fallback_allowed("session share revoke")
        return True


async def load_public_session_share(token_hash: str) -> dict[str, Any] | None:
    try:
        get_pool()
        async with acquire(ai_context=True) as conn:
            row = await conn.fetchrow(
                """
                SELECT session_id, payload, created_at, updated_at, revoked_at
                FROM app.session_share_link
                WHERE token_hash = $1
                  AND revoked_at IS NULL
                """,
                token_hash,
            )
            return _share_record_from_row(row) if row is not None else None
    except Exception:
        require_runtime_fallback_allowed("session share public load")
        session_id = _SESSION_SHARE_TOKEN_INDEX.get(token_hash)
        if not session_id:
            return None
        cached = _SESSION_SHARE_CACHE.get(session_id)
        if not cached or cached.get("revoked_at") is not None:
            return None
        return {
            "session_id": str(cached["session_id"]),
            "payload": dict(cached.get("payload") or {}),
            "created_at": cached.get("created_at"),
            "updated_at": cached.get("updated_at"),
            "revoked_at": cached.get("revoked_at"),
        }


async def list_session_archives(
    session_ids: list[str],
    principal: Principal,
) -> tuple[dict[str, dict[str, Any]], bool]:
    """Return learner archive state keyed by session id."""
    if not session_ids:
        return {}, True
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                """
                SELECT session_id, learner_id, archived_at, updated_at
                FROM app.session_archive_state
                WHERE session_id = ANY($1::uuid[])
                """,
                session_ids,
            )
        return {
            str(row["session_id"]): _archive_record_from_row(row) for row in rows
        }, True
    except Exception:
        require_runtime_fallback_allowed("session archive list")
        return {
            session_id: dict(record)
            for session_id in session_ids
            if (record := _SESSION_ARCHIVE_CACHE.get(session_id)) is not None
        }, False


async def set_session_archived(
    *,
    session_id: str,
    learner_id: str,
    archived: bool,
) -> tuple[dict[str, Any] | None, bool]:
    """Archive or restore a learner-owned session without deleting evidence."""
    now = datetime.now(timezone.utc)
    cache_record = {
        "session_id": session_id,
        "learner_id": learner_id,
        "archived_at": now.timestamp(),
        "updated_at": now.timestamp(),
    }
    if runtime_fallback_allowed():
        if archived:
            _SESSION_ARCHIVE_CACHE[session_id] = cache_record
        else:
            _SESSION_ARCHIVE_CACHE.pop(session_id, None)
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            if archived:
                row = await conn.fetchrow(
                    """
                    INSERT INTO app.session_archive_state (
                      session_id, learner_id, archived_at, updated_at
                    )
                    VALUES ($1::uuid, $2::uuid, now(), now())
                    ON CONFLICT (session_id) DO UPDATE SET
                      learner_id = EXCLUDED.learner_id,
                      archived_at = COALESCE(app.session_archive_state.archived_at, EXCLUDED.archived_at),
                      updated_at = now()
                    RETURNING session_id, learner_id, archived_at, updated_at
                    """,
                    session_id,
                    learner_id,
                )
                return _archive_record_from_row(row), True
            deleted = await conn.fetchrow(
                """
                DELETE FROM app.session_archive_state
                WHERE session_id = $1::uuid
                  AND learner_id = $2::uuid
                RETURNING session_id, learner_id, archived_at, updated_at
                """,
                session_id,
                learner_id,
            )
            return (
                _archive_record_from_row(deleted) if deleted is not None else None
            ), True
    except Exception:
        require_runtime_fallback_allowed("session archive update")
        return (_SESSION_ARCHIVE_CACHE.get(session_id) if archived else None), False


def _session_from_rows(row, state_row, turn_rows: Iterable) -> InProcSession | None:
    card = _card_from_joined_session_row(row)
    if card is None:
        if not settings.allow_seed_persona_fallback:
            return None
        persona_code = (row["persona_code"] or "").upper()
        legacy_entry = seed_fallback_persona(persona_code)
        if legacy_entry is None:
            return None
        card = legacy_entry.card
    persona_code = card.code

    ended_at = _ts(row["ended_at"])
    started_at = _ts(row["started_at"]) or time.time()
    return InProcSession(
        session_id=str(row["id"]),
        case_id=str(row["case_id"] or row["runtime_case_id"] or row["id"]),
        learner_id=str(row["learner_id"]),
        persona_code=persona_code,
        theory_mode=row["theory_mode"] or "humanistic",
        persona=card,
        state=_state_from_row(state_row, card),
        session_no=int(row["session_no"] or 1),
        created_at=started_at,
        ended_at=ended_at,
        turns=[_turn_from_row(turn_row) for turn_row in turn_rows],
        ended=ended_at is not None,
        prev_rapport_credit=float(row["prev_rapport_credit"] or 0.0),
        learner_label=_clean_text(_row_value(row, "learner_label")),
        goal_stages=[
            str(v) for v in _json_list_payload(_row_value(row, "session_goals"))
        ],
    )


async def _upsert_state(
    conn, session_id: str, state: state_machine.SessionState
) -> None:
    await conn.execute(
        """
        INSERT INTO app.session_state (
          session_id, stage, turn_seq, effective_openness, rapport_credit,
          resistance, ideation_stage, turns_in_stage, affect_state, updated_at
        )
        VALUES ($1::uuid, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, now())
        ON CONFLICT (session_id) DO UPDATE SET
          stage = EXCLUDED.stage,
          turn_seq = EXCLUDED.turn_seq,
          effective_openness = EXCLUDED.effective_openness,
          rapport_credit = EXCLUDED.rapport_credit,
          resistance = EXCLUDED.resistance,
          ideation_stage = EXCLUDED.ideation_stage,
          turns_in_stage = EXCLUDED.turns_in_stage,
          affect_state = EXCLUDED.affect_state,
          updated_at = now()
        """,
        session_id,
        _stage(state.stage),
        state.turn_seq,
        state.effective_openness,
        state.rapport_credit,
        state.resistance,
        state.ideation_stage,
        state.turns_in_stage,
        state.affect_state,
    )


async def get_case_context(
    *,
    learner_id: str,
    persona_id: str,
) -> CaseContext | None:
    """Return the stable learner-persona case row, creating it when possible."""
    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,
            )
        return CaseContext(
            case_id=str(row["case_id"]),
            last_session_no=int(row["last_session_no"] or 0),
        )
    except Exception:
        return None


async def create_session(
    *,
    learner_id: str,
    card: PersonaCard,
    theory_mode: str,
    state: state_machine.SessionState,
    session_no: int = 1,
    carry_rapport: float = 0.0,
    persona_id: str | None = None,
    persona_version: int | None = None,
    case_id: str | None = None,
    goal_stages: list[str] | None = None,
) -> InProcSession | None:
    """Create a DB-backed session, returning None when DB persistence is unavailable."""
    try:
        get_pool()
        runtime_case_id = str(uuid.uuid4())
        pinned_persona_id = persona_id or seed_persona_id(card.code)
        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:
                    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 = int(case_row["last_session_no"] or 0) + 1
                counter_row = await conn.fetchrow(
                    """
                    UPDATE app.case_profile
                    SET last_session_no = GREATEST(last_session_no + 1, $2),
                        updated_at = now()
                    WHERE case_id = $1::uuid
                      AND learner_id = $3::uuid
                    RETURNING last_session_no
                    """,
                    stable_case_id,
                    session_no,
                    learner_id,
                )
                if counter_row is not None:
                    session_no = int(counter_row["last_session_no"] or session_no)
                row = await conn.fetchrow(
                    """
                    INSERT INTO app.sessions (
                      runtime_case_id, case_id, learner_id, persona_id, persona_version,
                      persona_code, persona_display_name, persona_difficulty,
                      session_no, theory_mode, stage_path, prev_rapport_credit, session_goals
                    )
                    VALUES (
                      $1::uuid, $2::uuid, $3::uuid, $4::uuid, $5,
                      $6, $7, $8,
                      $9, $10, '[]'::jsonb, $11, $12::jsonb
                    )
                    RETURNING id, runtime_case_id, case_id, learner_id, persona_code,
                              session_no, theory_mode, started_at, ended_at, prev_rapport_credit
                    """,
                    runtime_case_id,
                    stable_case_id,
                    learner_id,
                    pinned_persona_id,
                    pinned_persona_version,
                    card.code,
                    card.display_name,
                    card.difficulty,
                    session_no,
                    theory_mode,
                    carry_rapport,
                    list(goal_stages or []),
                )
                await _upsert_state(conn, str(row["id"]), state)
        return InProcSession(
            session_id=str(row["id"]),
            case_id=str(row["case_id"] or row["runtime_case_id"] or row["id"]),
            learner_id=learner_id,
            persona_code=card.code,
            theory_mode=theory_mode,
            persona=card,
            state=state,
            session_no=int(row["session_no"] or session_no),
            created_at=_ts(row["started_at"]) or time.time(),
            ended_at=None,
            turns=[],
            ended=False,
            prev_rapport_credit=carry_rapport,
            goal_stages=list(goal_stages or []),
        )
    except Exception:
        require_runtime_fallback_allowed("session creation")
        return None


async def load_session(
    session_id: str,
    principal: Principal,
    *,
    allow_ended: bool = False,
    include_turn_evaluation: bool = False,
) -> InProcSession | None:
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            row = await conn.fetchrow(
                f"""
                SELECT {SESSION_PERSONA_SELECT_COLUMNS_SQL}
                FROM app.sessions s
                {SESSION_PERSONA_JOINS_SQL}
                WHERE s.id = $1::uuid
                """,
                session_id,
            )
            if row is None:
                return None
            if row["ended_at"] is not None and not allow_ended:
                return None
            states_by_id, turns_by_id = await _fetch_session_runtime_rows(
                conn, [session_id]
            )
            sess = _session_from_rows(
                row,
                states_by_id.get(session_id),
                turns_by_id.get(session_id, []),
            )
            if sess is not None:
                await _record_session_read_audit(
                    conn,
                    principal,
                    target_kind="session",
                    target_id=session_id,
                    detail={
                        "access": "load_session",
                        "role": principal.role.value,
                        "learner_id": sess.learner_id,
                    },
                )
        if sess is not None and include_turn_evaluation:
            await _hydrate_session_turn_evaluations(sess)
        return sess
    except Exception:
        require_runtime_fallback_allowed("session load")
        return None


async def append_turn(
    *,
    session_id: str,
    learner_id: str,
    turn: TurnRecord,
) -> bool:
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            locked = await conn.fetchval(
                "SELECT id FROM app.sessions WHERE id = $1::uuid FOR UPDATE",
                session_id,
            )
            if locked is None:
                return False
            seq = int(
                await conn.fetchval(
                    "SELECT COALESCE(MAX(seq), 0) + 1 FROM app.turns WHERE session_id = $1::uuid",
                    session_id,
                )
                or 1
            )
            inserted_turn_id = await conn.fetchval(
                """
                INSERT INTO app.turns (
                  session_id, seq, speaker, stage, text, text_masked, actor_kind,
                  llm_provider, model, tokens_in, tokens_out, cost_usd,
                  audio_ref, silence_ms, speech_rate, barge_in, provider_events, visible_to
                )
                VALUES (
                  $1::uuid, $2, $3, $4, $5, $6, $7,
                  $8, $9, $10, $11, $12,
                  $13, $14, $15, $16, $17::jsonb, $18::text[]
                )
                ON CONFLICT (session_id, seq) DO NOTHING
                RETURNING id
                """,
                session_id,
                seq,
                turn.speaker,
                turn.stage,
                turn.text_masked,
                turn.text_masked,
                "human_learner" if turn.speaker == "counselor" else "client_ai",
                turn.llm_provider,
                turn.model,
                turn.tokens_in,
                turn.tokens_out,
                turn.cost_usd,
                turn.audio_ref,
                turn.silence_ms,
                turn.speech_rate,
                turn.barge_in,
                turn.provider_events or [],
                list(turn.visible_to or DEFAULT_TURN_VISIBLE_TO),
            )
            if inserted_turn_id is None:
                return False
            turn.turn_id = str(inserted_turn_id)
            # 원시 평가 row는 학습자 축어록이 아니라 evaluator 전용 데이터다.
            await _persist_turn_evaluation(conn, turn.turn_id, turn.evaluation)
        return True
    except Exception:
        require_runtime_fallback_allowed("session turn append")
        return False


async def update_state(
    *,
    session_id: str,
    learner_id: str,
    state: state_machine.SessionState,
) -> bool:
    try:
        get_pool()
        async with acquire(role="learner", user_id=learner_id) as conn:
            await _upsert_state(conn, session_id, state)
        return True
    except Exception:
        require_runtime_fallback_allowed("session state update")
        return False


async def _insert_pinned_fact_history(
    conn: Any,
    *,
    fact_id: Any,
    case_id: str,
    old_value: Any,
    new_value: Any,
    reason: str,
    session_no: int,
    turn_id: str | None,
) -> None:
    await conn.execute(
        """
        INSERT INTO app.pinned_fact_history (
          fact_id, case_id, old_value, new_value, reason, session_no, turn_id
        )
        VALUES ($1::uuid, $2::uuid, $3, $4, $5, $6, $7::uuid)
        """,
        fact_id,
        case_id,
        old_value,
        new_value,
        reason,
        session_no,
        turn_id,
    )


async def _upsert_pinned_fact_candidates(conn: Any, sess: InProcSession) -> None:
    turn_rows = [
        {
            "speaker": turn.speaker,
            "text": turn.text_masked,
            "turn_id": turn.turn_id,
        }
        for turn in sess.turns_visible_to("client")
    ]
    candidates = memory.extract_pinned_fact_candidates(turn_rows)
    for fact in candidates:
        if fact.status == "contradicted":
            row = await conn.fetchrow(
                """
                WITH existing AS (
                  SELECT id, value
                  FROM app.pinned_fact
                  WHERE case_id = $1::uuid
                    AND key = $2
                    AND status <> 'locked'
                  FOR UPDATE
                ),
                updated AS (
                  UPDATE app.pinned_fact
                  SET value = $3,
                      fact_type = $4,
                      status = 'contradicted',
                      source_turn = COALESCE($5::uuid, app.pinned_fact.source_turn),
                      confidence = GREATEST(app.pinned_fact.confidence, $6),
                      version = app.pinned_fact.version + 1,
                      updated_session_no = $7,
                      visible_to = $8::text[],
                      updated_at = now()
                  FROM existing
                  WHERE app.pinned_fact.id = existing.id
                  RETURNING
                    app.pinned_fact.id,
                    existing.value AS old_value,
                    app.pinned_fact.value AS new_value
                )
                SELECT id, old_value, new_value FROM updated
                """,
                sess.case_id,
                fact.key,
                fact.value,
                fact.fact_type,
                fact.source_turn_id,
                fact.confidence,
                sess.session_no,
                ["evaluator"],
            )
            if not row:
                continue
            old_value = row["old_value"]
            new_value = row["new_value"]
            if old_value == new_value:
                continue
            await _insert_pinned_fact_history(
                conn,
                fact_id=row["id"],
                case_id=sess.case_id,
                old_value=old_value,
                new_value=new_value,
                reason="contradiction",
                session_no=sess.session_no,
                turn_id=fact.source_turn_id,
            )
            continue
        row = await conn.fetchrow(
            """
            WITH existing AS (
              SELECT id, value, status
              FROM app.pinned_fact
              WHERE case_id = $1::uuid AND key = $2
              FOR UPDATE
            ),
            upserted AS (
            INSERT INTO app.pinned_fact (
              case_id, key, value, fact_type, status, source_turn,
              confidence, updated_session_no, visible_to, updated_at
            )
            VALUES (
              $1::uuid, $2, $3, $4, $5, $6::uuid,
              $7, $8, $9::text[], now()
            )
            ON CONFLICT (case_id, key) DO UPDATE SET
              value = EXCLUDED.value,
              fact_type = EXCLUDED.fact_type,
              status = EXCLUDED.status,
              source_turn = COALESCE(EXCLUDED.source_turn, app.pinned_fact.source_turn),
              confidence = GREATEST(app.pinned_fact.confidence, EXCLUDED.confidence),
              version = CASE
                WHEN app.pinned_fact.value IS DISTINCT FROM EXCLUDED.value
                THEN app.pinned_fact.version + 1
                ELSE app.pinned_fact.version
              END,
              updated_session_no = EXCLUDED.updated_session_no,
              visible_to = EXCLUDED.visible_to,
              updated_at = now()
            WHERE app.pinned_fact.status <> 'locked'
            RETURNING
              app.pinned_fact.id,
              (SELECT value FROM existing) AS old_value,
              app.pinned_fact.value AS new_value
            )
            SELECT id, old_value, new_value FROM upserted
            """,
            sess.case_id,
            fact.key,
            fact.value,
            fact.fact_type,
            fact.status,
            fact.source_turn_id,
            fact.confidence,
            sess.session_no,
            ["client", "evaluator"],
        )
        if not row:
            continue
        old_value = row["old_value"]
        new_value = row["new_value"]
        if old_value is not None and old_value == new_value:
            continue
        reason = "progression" if old_value is None else "clarification"
        await _insert_pinned_fact_history(
            conn,
            fact_id=row["id"],
            case_id=sess.case_id,
            old_value=old_value,
            new_value=new_value,
            reason=reason,
            session_no=sess.session_no,
            turn_id=fact.source_turn_id,
        )


def _build_session_summary_write(
    sess: InProcSession, carry: memory.CarryOver
) -> SessionSummaryWrite:
    digest_input = memory.build_session_digest_input(
        session_id=sess.session_id,
        case_id=sess.case_id,
        session_no=sess.session_no,
        masked_turns=sess.masked_turns(visible_to="client"),
        open_threads=carry.compression_job.open_threads
        if carry.compression_job
        else [],
    )
    digest_result = memory.build_fallback_digest_result(
        digest_input, end_state=carry.end_state
    )
    return SessionSummaryWrite(
        session_id=sess.session_id,
        case_id=sess.case_id,
        session_no=sess.session_no,
        end_state=carry.end_state,
        rapport_delta=carry.rapport_delta,
        digest=digest_result.digest,
        open_threads=list(digest_result.open_threads),
    )


async def end_session(sess: InProcSession, carry: memory.CarryOver) -> bool:
    try:
        get_pool()
        summary_write = _build_session_summary_write(sess, carry)
        async with acquire(role="learner", user_id=sess.learner_id) as conn:
            await conn.execute(
                """
                UPDATE app.sessions
                SET ended_at = COALESCE(ended_at, now())
                WHERE id = $1::uuid
                """,
                sess.session_id,
            )
            await _upsert_state(conn, sess.session_id, sess.state)
            await conn.execute(
                """
                INSERT INTO app.session_summary (
                  session_id, case_id, session_no, end_state, rapport_delta,
                  digest, open_threads, created_at
                )
                VALUES ($1::uuid, $2::uuid, $3, $4::jsonb, $5, $6, $7::jsonb, now())
                ON CONFLICT (session_id) DO UPDATE SET
                  end_state = EXCLUDED.end_state,
                  rapport_delta = EXCLUDED.rapport_delta,
                  digest = EXCLUDED.digest,
                  open_threads = EXCLUDED.open_threads,
                  compressed_by = NULL,
                  token_count = NULL
                """,
                summary_write.session_id,
                summary_write.case_id,
                summary_write.session_no,
                summary_write.end_state,
                summary_write.rapport_delta,
                summary_write.digest,
                summary_write.open_threads,
            )
            case_row = await conn.fetchrow(
                """
                SELECT case_digest, rapport_trajectory, alliance_level
                FROM app.case_profile
                WHERE case_id = $1::uuid
                  AND learner_id = $2::uuid
                """,
                sess.case_id,
                sess.learner_id,
            )
            if case_row is not None:
                trajectory_point = memory.rapport_trajectory_point(
                    session_no=sess.session_no,
                    end_state=carry.end_state,
                )
                case_digest = memory.merge_case_digest(
                    existing_digest=case_row["case_digest"],
                    session_no=sess.session_no,
                    session_digest=summary_write.digest,
                )
                rapport_trajectory = memory.merge_rapport_trajectory(
                    case_row["rapport_trajectory"],
                    trajectory_point,
                )
                alliance_level = memory.update_alliance_level(
                    case_row["alliance_level"],
                    trajectory_point.get("end_rapport"),
                )
                await conn.execute(
                    """
                    UPDATE app.case_profile
                    SET case_digest = $3,
                        rapport_trajectory = $4::jsonb,
                        alliance_level = $5,
                        updated_at = now()
                    WHERE case_id = $1::uuid
                      AND learner_id = $2::uuid
                    """,
                    sess.case_id,
                    sess.learner_id,
                    case_digest,
                    rapport_trajectory,
                    alliance_level,
                )
                await _upsert_pinned_fact_candidates(conn, sess)
        return True
    except Exception:
        logger.exception("persisted session end failed session_id=%s", sess.session_id)
        require_runtime_fallback_allowed("session end")
        return False


RECENT_SESSION_LIST_LIMIT = 100


async def list_recent_sessions(
    principal: Principal,
    *,
    include_turn_evaluation: bool = False,
) -> tuple[list[InProcSession], bool]:
    return await _list_sessions(
        principal,
        include_turn_evaluation=include_turn_evaluation,
        session_limit=RECENT_SESSION_LIST_LIMIT,
        audit_access="list_recent_sessions",
    )


async def list_all_sessions(
    principal: Principal,
    *,
    include_turn_evaluation: bool = False,
) -> tuple[list[InProcSession], bool]:
    return await _list_sessions(
        principal,
        include_turn_evaluation=include_turn_evaluation,
        session_limit=None,
        audit_access="list_all_sessions",
    )


async def _list_sessions(
    principal: Principal,
    *,
    include_turn_evaluation: bool,
    session_limit: int | None,
    audit_access: str,
) -> tuple[list[InProcSession], bool]:
    try:
        get_pool()
        learner_filter = (
            "WHERE s.learner_id = $1::uuid" if principal.role.value == "learner" else ""
        )
        query_args: list[object] = (
            [principal.user_id] if principal.role.value == "learner" else []
        )
        limit_clause = ""
        if session_limit is not None:
            query_args.append(max(1, int(session_limit)))
            limit_clause = f"LIMIT ${len(query_args)}"
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                f"""
                SELECT {SESSION_PERSONA_SELECT_COLUMNS_SQL}
                FROM app.sessions s
                {SESSION_PERSONA_JOINS_SQL}
                {learner_filter}
                ORDER BY s.started_at DESC
                {limit_clause}
                """,
                *query_args,
            )
            # N+1 제거: 세션별 state/turns fetch 루프(1+2N 왕복) 대신 id 집합으로 한 번씩 배치 조회.
            session_ids = [str(row["id"]) for row in rows]
            states_by_id, turns_by_id = await _fetch_session_runtime_rows(
                conn, session_ids
            )
            sessions: list[InProcSession] = []
            for row in rows:
                session_id = str(row["id"])
                sess = _session_from_rows(
                    row, states_by_id.get(session_id), turns_by_id.get(session_id, [])
                )
                if sess is not None:
                    sessions.append(sess)
            if include_turn_evaluation and sessions:
                await _hydrate_sessions_turn_evaluations(sessions)
            await _record_session_read_audit(
                conn,
                principal,
                target_kind="session_list",
                target_id="sessions",
                detail={
                    "access": audit_access,
                    "role": principal.role.value,
                    "result_count": len(sessions),
                },
            )
            return sessions, True
    except Exception:
        require_runtime_fallback_allowed("session list")
        return [], False


async def list_safety_alerts(
    principal: Principal,
    *,
    limit: int = 20,
) -> tuple[list[dict[str, Any]], bool]:
    """Teacher/admin-visible crisis alerts from app.safety_events."""
    try:
        get_pool()
        async with acquire(
            role=principal.role.value,
            user_id=principal.user_id,
            cohort_ids=principal.cohort_ids,
        ) as conn:
            rows = await conn.fetch(
                """
                SELECT
                  se.id, se.session_id, se.trigger_type, se.ko_risk_level,
                  se.escalated, se.detail, se.created_at,
                  s.learner_id, s.persona_code, s.session_no,
                  COALESCE(
                    NULLIF(learner.display_name, ''),
                    NULLIF(learner.nickname, ''),
                    NULLIF(learner.email, ''),
                    s.learner_id::text
                  ) AS learner_label
                FROM app.safety_events se
                LEFT JOIN app.sessions s ON s.id = se.session_id
                LEFT JOIN app.app_user learner ON learner.user_id = s.learner_id
                WHERE se.escalated = TRUE
                ORDER BY se.created_at DESC
                LIMIT $1
                """,
                limit,
            )
            return [
                {
                    "id": str(row["id"]),
                    "session_id": str(row["session_id"]),
                    "learner_id": str(row["learner_id"] or ""),
                    "learner_label": _clean_text(row["learner_label"])
                    or _learner_label_from_id(str(row["learner_id"] or "")),
                    "persona_code": str(row["persona_code"] or ""),
                    "session_no": int(row["session_no"] or 0),
                    "trigger_type": str(row["trigger_type"] or "crisis"),
                    "ko_risk_level": int(row["ko_risk_level"] or 0),
                    "escalated": bool(row["escalated"]),
                    "detail": _json_object_payload(row["detail"]),
                    "created_at": _iso_dt(row["created_at"]),
                }
                for row in rows
            ], True
    except Exception:
        require_runtime_fallback_allowed("safety alert list")
        return [], False


def _learner_label_from_id(learner_id: str) -> str:
    suffix = learner_id[-6:] if len(learner_id) > 6 else learner_id
    return f"학습자 {suffix}" if suffix else "학습자"


def _iso_dt(value: datetime | None) -> str:
    if value is None:
        return ""
    return value.astimezone(timezone.utc).isoformat()