yunchan/vignette
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

세션 평가·라이브코치·교수자 분석 라운드 마감 + 문서 정리 + 코드품질 리팩터

Browse source 
- 누적 작업트리 커밋: 회기 평가 복구·durable 저장, 라이브 코치 이력/근거, 교수자 학생분석, 음성 비언어 메타, PII 마스킹, 운영 티켓/헬스 등
- 문서: 완료 기록 docs/archive/ 냉동 보관, docs/ 단일 인덱스(docs/README.md)+통합 TODO(docs/TODO.md)로 정리
- 리팩터(행위 보존): Stage enum SSOT(taxonomy 소유·state_machine re-export), store recent/masked_turns 중복 제거, speaker_ko_label 단일 헬퍼, _list_sessions N+1 제거(state/turns 배치 + 턴평가 하이드레이션 배치)
- 검증: 백엔드 pytest 352 passed, _list_sessions E2E chromium-single-run 2 passed
This commit is contained in:
Yun Chan 2026-07-02 02:50:36 +09:00
parent 7c41c3ce79
commit 778e8526d4
108 changed files with 6457 additions and 455 deletions
Download patch file Download diff file Expand all files Collapse all files

698
apps/api/app/session_persistence.py
View file

@ -38,6 +38,14 @@ _APPROPRIATENESS_SCORE = {
"neutral": 3.0,
"pos": 5.0,
}
LIVE_COACH_INITIAL_CREDITS = 3
LIVE_COACH_MAX_CREDITS = 3
LIVE_COACH_USE_REASON = "AI 코칭 힌트 사용"
LIVE_COACH_RECHARGE_REASON = "좋은 발화로 내담자 변화 신호 확인"
class LiveCoachCreditExhausted(RuntimeError):
"""Raised when a learner tries to use live coaching without credits."""
def _coerce_error_message(error: BaseException | str) -> str:
@ -99,8 +107,8 @@ class SessionEvaluationWrite:
source=source,
scope=result.scope,
stage=result.stage,
payload=result.to_dict(),
error=result.error,
payload=_mask_json_text_values(result.to_dict()),
error=_clean_masked_text(result.error),
)
@classmethod
@ -122,7 +130,7 @@ class SessionEvaluationWrite:
scope=scope,
stage=stage,
payload={},
error=_coerce_error_message(error),
error=_clean_masked_text(_coerce_error_message(error)),
)
def cache_record(self) -> dict[str, Any]:
@ -244,6 +252,90 @@ def _live_coach_event_from_row(row) -> dict[str, Any]:
}
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,
@ -252,6 +344,35 @@ def _live_coach_cache_record(
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 {
@ -259,13 +380,22 @@ def _live_coach_cache_record(
"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": _masked_excerpt(learner_text),
"client_reply_excerpt": _masked_excerpt(client_reply),
"suggestion": _model_payload(suggestion),
"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]
@ -294,6 +424,26 @@ def _clean_text(value: Any) -> str | None:
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)
@ -354,7 +504,7 @@ def _evaluation_feedback_rows(evaluation: dict[str, Any] | None) -> list[dict[st
add(
"appropriateness",
score=_APPROPRIATENESS_SCORE[appropriateness],
rationale=_clean_text(evaluation.get("appropriateness_note")),
rationale=_clean_masked_text(evaluation.get("appropriateness_note")),
)
rapport = _safe_float(evaluation.get("rapport_signal"))
@ -365,19 +515,19 @@ def _evaluation_feedback_rows(evaluation: dict[str, Any] | None) -> list[dict[st
if theory_mode:
add("theory_mode", rationale=theory_mode)
error = _clean_text(evaluation.get("error"))
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_text(tag.get("rationale"))
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_text(state.get("rationale"))
rationale = _clean_masked_text(state.get("rationale"))
if code and rationale:
add(f"client_state:{code}", rationale=rationale)
@ -425,8 +575,8 @@ def _evaluation_comment_rows(evaluation: dict[str, Any] | None) -> list[dict[str
deviation = evaluation.get("intent_deviation")
if not isinstance(deviation, dict):
return []
note = _clean_text(evaluation.get("appropriateness_note")) or "의도와 다른 부분"
return [{"kind": "critique", "text": note, "intent_deviation": deviation}]
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]]:
@ -438,10 +588,10 @@ def _evaluation_alternative_rows(evaluation: dict[str, Any] | None) -> list[dict
rows: list[dict[str, str | None]] = []
for item in alternatives:
if isinstance(item, dict):
suggestion = _clean_text(item.get("suggestion") or item.get("text"))
rationale = _clean_text(item.get("rationale"))
suggestion = _clean_masked_text(item.get("suggestion") or item.get("text"))
rationale = _clean_masked_text(item.get("rationale"))
else:
suggestion = _clean_text(item)
suggestion = _clean_masked_text(item)
rationale = None
if suggestion:
rows.append({"suggestion": suggestion, "rationale": rationale})
@ -652,12 +802,13 @@ def _state_from_row(row, card: PersonaCard) -> state_machine.SessionState:
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=row["text"] or row["text_masked"] or "",
text_masked=row["text_masked"] or row["text"] or "",
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"),
@ -780,6 +931,31 @@ async def _persist_turn_evaluation(conn: Any, turn_id: str, evaluation: dict[str
)
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]],
@ -853,8 +1029,14 @@ async def _load_turn_evaluations(
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
]
@ -862,9 +1044,10 @@ async def _hydrate_session_turn_evaluations(sess: InProcSession) -> None:
return
async with acquire(ai_view="evaluator") as conn:
evaluations = await _load_turn_evaluations(conn, turn_refs)
for turn in sess.turns:
if turn.turn_id and turn.turn_id in evaluations:
turn.evaluation = evaluations[turn.turn_id]
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:
@ -946,6 +1129,47 @@ async def ensure_review_tables() -> None:
)
"""
)
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 (
@ -1013,13 +1237,45 @@ async def ensure_review_tables() -> None:
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;
@ -1270,6 +1526,8 @@ async def ensure_review_tables() -> None:
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
@ -1342,6 +1600,150 @@ async def load_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 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(
"""
SELECT
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,
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
FROM app.sessions s
LEFT JOIN app.session_evaluation se ON se.session_id = s.id
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
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)),
)
sessions: list[InProcSession] = []
for row in rows:
session_id = str(row["id"])
state_row = await conn.fetchrow(
"""
SELECT stage, turn_seq, effective_openness, rapport_credit, resistance,
ideation_stage, turns_in_stage, affect_state
FROM app.session_state
WHERE session_id = $1::uuid
""",
session_id,
)
turn_rows = await conn.fetch(
"""
SELECT id, seq, speaker, stage, text, text_masked, created_at,
llm_provider, model, tokens_in, tokens_out, cost_usd,
audio_ref, silence_ms, speech_rate, barge_in, provider_events, visible_to
FROM app.turns
WHERE session_id = $1::uuid
ORDER BY seq
""",
session_id,
)
sess = _session_from_rows(row, state_row, turn_rows)
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,
@ -1411,6 +1813,9 @@ async def save_live_coach_event(
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,
@ -1418,32 +1823,42 @@ async def save_live_coach_event(
learner_text=learner_text,
client_reply=client_reply,
suggestion=suggestion,
credit_balance=cached_remaining - 1,
)
if runtime_fallback_allowed():
_LIVE_COACH_EVENT_CACHE.setdefault(session_id, []).append(record)
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, learner_text_excerpt,
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, $4, $5, $6::jsonb, now())
RETURNING id, session_id, turn_seq, stage, learner_text_excerpt,
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
@ -1464,6 +1879,7 @@ async def list_live_coach_events(
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,
@ -1471,7 +1887,127 @@ async def list_live_coach_events(
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, [])], False
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]:
@ -1883,6 +2419,7 @@ def _session_from_rows(row, state_row, turn_rows: Iterable) -> InProcSession | N
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")),
)
@ -2063,6 +2600,12 @@ async def load_session(
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,
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,
@ -2086,6 +2629,7 @@ async def load_session(
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
WHERE s.id = $1::uuid
""",
session_id,
@ -2490,15 +3034,50 @@ async def end_session(sess: InProcSession, carry: memory.CarryOver) -> bool:
return False
async def list_sessions(
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 = [principal.user_id] 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,
@ -2510,6 +3089,12 @@ async def list_sessions(
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,
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,
@ -2533,47 +3118,58 @@ async def list_sessions(
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
{learner_filter}
ORDER BY s.started_at DESC
LIMIT 100
{limit_clause}
""",
*query_args,
)
sessions: list[InProcSession] = []
for row in rows:
session_id = str(row["id"])
state_row = await conn.fetchrow(
# N+1 제거: 세션별 state/turns fetch 루프(1+2N 왕복) 대신 id 집합으로 한 번씩 배치 조회.
session_ids = [str(row["id"]) for row in rows]
states_by_id: dict[str, Any] = {}
turns_by_id: dict[str, list] = {}
if session_ids:
state_rows = await conn.fetch(
"""
SELECT stage, turn_seq, effective_openness, rapport_credit, resistance,
SELECT session_id, stage, turn_seq, effective_openness, rapport_credit, resistance,
ideation_stage, turns_in_stage, affect_state
FROM app.session_state
WHERE session_id = $1::uuid
WHERE session_id = ANY($1::uuid[])
""",
session_id,
session_ids,
)
turn_rows = await conn.fetch(
states_by_id = {str(state_row["session_id"]): state_row for state_row in state_rows}
turn_rows_all = await conn.fetch(
"""
SELECT id, seq, speaker, stage, text, text_masked, created_at,
SELECT 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
FROM app.turns
WHERE session_id = $1::uuid
ORDER BY seq
WHERE session_id = ANY($1::uuid[])
ORDER BY session_id, seq
""",
session_id,
session_ids,
)
for turn_row in turn_rows_all:
turns_by_id.setdefault(str(turn_row["session_id"]), []).append(turn_row)
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, [])
)
sess = _session_from_rows(row, state_row, turn_rows)
if sess is not None:
if include_turn_evaluation:
await _hydrate_session_turn_evaluations(sess)
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": "list_sessions",
"access": audit_access,
"role": principal.role.value,
"result_count": len(sessions),
},
@ -2602,9 +3198,16 @@ async def list_safety_alerts(
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
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
@ -2616,7 +3219,8 @@ async def list_safety_alerts(
"id": str(row["id"]),
"session_id": str(row["session_id"]),
"learner_id": str(row["learner_id"] or ""),
"learner_label": _learner_label_from_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"),