#!/usr/bin/env python3 """Prove the G4/G5 production session-learning caller on PostgreSQL. The smoke persists a real ended session, durable counselor/client turn UUIDs, a ready session evaluation, and a learner prediction/lock. G4/G5 artifacts are created only by ``session_learning_producer`` (including the production lock callback), then audited for replay, provenance, and non-promotion invariants. Run only against an expendable development database. Unique fixture rows remain because the learning ledgers are intentionally append-only. """ from __future__ import annotations import argparse import asyncio import json import os import sys from pathlib import Path from typing import Any from uuid import UUID, uuid4 REPO_ROOT = Path(__file__).resolve().parents[1] API_ROOT = REPO_ROOT / "apps" / "api" API_ENV = API_ROOT / ".env" def _load_api_env() -> None: if not API_ENV.exists(): return for raw_line in API_ENV.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) _load_api_env() if str(API_ROOT) not in sys.path: sys.path.insert(0, str(API_ROOT)) from app import db, persona_repository, session_persistence # noqa: E402 from app.deps import Principal, Role # noqa: E402 from app.routes import calibration_transfer # noqa: E402 from app.routes.calibration_transfer import ( # noqa: E402 PredictionLockRequest, PredictionRevisionRequest, ) from app.services import ( # noqa: E402 calibration_transfer_store, deliberate_practice_store, memory, session_learning_producer, state_machine, ) from app.store import TurnRecord # noqa: E402 COHORT = "g4-g5-production-caller-smoke" COMPETENCY_ID = "competency.empathic_reflection" CALIBRATION_INSTRUMENT_ID = "calibration-mirror-g5" CALIBRATION_INSTRUMENT_VERSION = "1.0.0" class SmokeError(RuntimeError): pass def _principal(user_id: UUID, role: Role) -> Principal: return Principal( user_id=str(user_id), role=role, cohort_ids=[COHORT], consent_at=1.0, profile_completed_at=1.0, ) async def _seed_users(learner_id: UUID, teacher_id: UUID) -> None: async with db.acquire(role="admin") as conn: for user_id, role in ( (learner_id, "learner"), (teacher_id, "instructor"), ): await conn.execute( """ INSERT INTO app.app_user ( user_id, external_id, email, display_name, role, cohort, consent_at, profile_completed_at, terms_agreed_at, privacy_agreed_at ) VALUES ($1,$2,$3,$4,$5,$6,now(),now(),now(),now()) """, user_id, f"dev:e2e:session-learning-producer:{user_id}", f"{user_id}@session-learning-producer.invalid", f"G4 G5 synthetic {role} fixture", role, COHORT, ) def _fast_deviation() -> dict[str, Any]: return { "loop": "fast", "turn_seq": 1, "stage": "rapport", "appropriateness": "warn", "appropriateness_note": "정서 반영 전에 다음 질문으로 이동함", "intent_deviation": { "dimension": "reflection", "expected": "정서를 반영한 뒤 이해가 맞는지 확인한다", "actual": "정서 확인 없이 다음 질문으로 이동했다", "severity": "moderate", }, } def _ready_deep_evaluation(session_id: str) -> dict[str, Any]: return { "loop": "deep", "session_id": session_id, "stage": "rapport", "scope": "session_end", "turns_evaluated": 2, "intent_deviations": [ { "dimension": "reflection", "expected": "정서를 반영한 뒤 이해가 맞는지 확인한다", "actual": "정서 확인 없이 다음 질문으로 이동했다", "severity": "moderate", } ], "improvements": ["핵심 정서를 짧게 반영하고 이해가 맞는지 확인한다."], } async def _create_ready_session(learner: Principal) -> tuple[Any, UUID, UUID]: await persona_repository.materialize_seed_personas() catalog = await persona_repository.get_approved_persona("P1") if catalog is None: raise SmokeError("approved P1 persona is unavailable") session = await session_persistence.create_session( learner_id=learner.user_id, card=catalog.card, theory_mode="humanistic", state=state_machine.init_state(params=catalog.card.openness_params()), persona_id=catalog.persona_id, persona_version=catalog.version, goal_stages=["라포", "탐색"], ) if session is None: raise SmokeError("durable session creation fell back or failed") counselor_turn = TurnRecord( turn_seq=1, speaker="counselor", stage=session.state.stage.value, text="[MASKED]", text_masked="[MASKED]", evaluation=_fast_deviation(), ) if not await session_persistence.append_turn( session_id=session.session_id, learner_id=learner.user_id, turn=counselor_turn, ): raise SmokeError("durable counselor turn append failed") if counselor_turn.turn_id is None: raise SmokeError("counselor turn did not receive a durable UUID") session.turns.append(counselor_turn) client_turn = TurnRecord( turn_seq=2, speaker="client", stage=session.state.stage.value, text="[MASKED]", text_masked="[MASKED]", llm_provider="fixture", model="synthetic-client-response", tokens_in=0, tokens_out=0, cost_usd=0.0, ) if not await session_persistence.append_turn( session_id=session.session_id, learner_id=learner.user_id, turn=client_turn, ): raise SmokeError("durable client turn append failed") if client_turn.turn_id is None: raise SmokeError("client turn did not receive a durable UUID") session.turns.append(client_turn) carry = memory.make_carry_over( state=session.state, session_id=session.session_id, case_id=session.case_id, session_no=session.session_no, masked_turns=session.masked_turns(visible_to="client"), open_threads=[], ) if not await session_persistence.end_session(session, carry): raise SmokeError("durable session end failed") saved = await session_persistence.save_session_evaluation( session_persistence.SessionEvaluationWrite( session_id=session.session_id, learner_id=learner.user_id, status="ready", source="synthetic_fixture", scope="session_end", stage=session.state.stage.value, payload=_ready_deep_evaluation(session.session_id), ) ) if not saved: raise SmokeError("ready session evaluation persistence failed") return session, UUID(counselor_turn.turn_id), UUID(client_turn.turn_id) async def _counts(session_id: UUID) -> dict[str, int]: async with db.acquire(role="admin") as conn: row = await conn.fetchrow( """ SELECT (SELECT count(*) FROM app.practice_prescription_submission WHERE session_id = $1)::int AS g4_submissions, (SELECT count(*) FROM app.practice_coaching_card WHERE session_id = $1)::int AS g4_cards, (SELECT count(*) FROM app.practice_prescription WHERE session_id = $1)::int AS g4_prescriptions, (SELECT count(*) FROM app.competency_graph_snapshot WHERE session_id = $1)::int AS g4_snapshots, (SELECT count(*) FROM app.practice_curriculum_decision_event WHERE session_id = $1)::int AS g4_decisions, (SELECT count(*) FROM app.practice_episode_submission WHERE session_id = $1)::int AS g4_episodes, (SELECT count(*) FROM app.practice_attempt_evidence WHERE session_id = $1)::int AS g4_attempts, (SELECT count(*) FROM app.calibration_prediction_history WHERE session_id = $1)::int AS g5_histories, (SELECT count(*) FROM app.calibration_prediction_revision WHERE session_id = $1)::int AS g5_revisions, (SELECT count(*) FROM app.calibration_prediction_lock WHERE session_id = $1)::int AS g5_locks, (SELECT count(*) FROM app.calibration_performance_observation WHERE session_id = $1)::int AS g5_observations, (SELECT count(*) FROM audit.model_run WHERE session_id = $1 AND prompt_bundle_id = 'session-learning-producer')::int AS producer_model_runs, (SELECT count(*) FROM app.calibration_assessment_snapshot WHERE session_id = $1)::int AS g5_assessments, (SELECT count(*) FROM app.calibration_transfer_suite WHERE session_id = $1)::int AS g5_transfer_suites """, session_id, ) return dict(row or {}) async def _postgres_proof( *, session_id: UUID, counselor_turn_id: UUID, client_turn_id: UUID, history_id: UUID, ) -> dict[str, Any]: expected_evidence = {counselor_turn_id, client_turn_id} async with db.acquire(role="admin") as conn: source = await conn.fetchrow( """ SELECT e.status AS evaluation_status, e.scope, count(t.id)::int AS durable_turn_count, count(sc.id) FILTER (WHERE sc.intent_deviation IS NOT NULL)::int AS deviation_comment_count FROM app.session_evaluation e JOIN app.turns t ON t.session_id = e.session_id LEFT JOIN app.supervisor_comment sc ON sc.turn_id = t.id WHERE e.session_id = $1 GROUP BY e.status, e.scope """, session_id, ) g4 = await conn.fetchrow( """ SELECT ps.submission_id, c.coaching_card_record_id, c.card_key, c.evidence_turn_ids AS card_evidence_turn_ids, p.prescription_record_id, p.prescription_key, p.competency_id, p.scenario_novelty, p.evidence_turn_ids AS prescription_evidence_turn_ids, gs.snapshot_id, gs.graph_payload, d.decision_id FROM app.practice_prescription_submission ps JOIN app.practice_coaching_card c ON c.submission_id = ps.submission_id JOIN app.practice_prescription p ON p.coaching_card_record_id = c.coaching_card_record_id JOIN app.competency_graph_snapshot gs ON gs.source_prescription_submission_id = ps.submission_id JOIN app.practice_curriculum_decision_event d ON d.source_snapshot_id = gs.snapshot_id WHERE ps.session_id = $1 """, session_id, ) g5 = await conn.fetchrow( """ SELECT h.history_id, r.prediction_revision_id, l.lock_id, o.observation_id, o.status, o.source_kind, o.perspective, o.model_run_id, o.evidence_turn_ids, o.counterevidence, o.revealed_sequence, l.locked_sequence, mr.turn_id AS model_run_turn_id, mr.agent_role, mr.provider, mr.model, mr.status AS model_run_status, mr.prompt_bundle_id, mr.prompt_bundle_version, mr.input_evidence_hash, mr.metadata AS model_run_metadata FROM app.calibration_prediction_history h JOIN app.calibration_prediction_revision r ON r.history_id = h.history_id JOIN app.calibration_prediction_lock l ON l.history_id = h.history_id JOIN app.calibration_performance_observation o ON o.history_id = h.history_id JOIN audit.model_run mr ON mr.model_run_id = o.model_run_id WHERE h.history_id = $1 AND h.session_id = $2 """, history_id, session_id, ) forbidden = await conn.fetchrow( """ SELECT (SELECT count(*) FROM app.practice_episode_submission WHERE session_id = $1 AND (mastery_allowed OR progress = 'mastered'))::int AS auto_mastery_rows, (SELECT count(*) FROM app.competency_graph_snapshot WHERE session_id = $1 AND graph_payload::text LIKE '%transfer_verified%')::int AS transfer_verified_snapshots, (SELECT count(*) FROM app.calibration_performance_observation WHERE session_id = $1 AND status = 'passed')::int AS passed_observations, (SELECT count(*) FROM app.calibration_transfer_trial WHERE session_id = $1)::int AS transfer_trials, (SELECT count(*) FROM app.calibration_transfer_assessment WHERE session_id = $1)::int AS transfer_assessments, (SELECT count(*) FROM app.calibration_subgroup_drift_report WHERE session_id = $1)::int AS drift_reports """, session_id, ) source_payload = dict(source or {}) g4_payload = dict(g4 or {}) g5_payload = dict(g5 or {}) forbidden_payload = dict(forbidden or {}) if source_payload != { "evaluation_status": "ready", "scope": "session_end", "durable_turn_count": 2, "deviation_comment_count": 1, }: raise SmokeError(f"durable ready source differs: {source_payload}") if set(g4_payload.get("card_evidence_turn_ids") or []) != expected_evidence: raise SmokeError("G4 card did not preserve both durable turn UUIDs") if set(g4_payload.get("prescription_evidence_turn_ids") or []) != expected_evidence: raise SmokeError("G4 prescription did not preserve both durable turn UUIDs") if g4_payload.get("competency_id") != COMPETENCY_ID: raise SmokeError("G4 prescription competency differs from evaluation") if g4_payload.get("scenario_novelty") != "familiar": raise SmokeError("G4 caller claimed unseen transfer") graph = g4_payload.get("graph_payload") or {} states = graph.get("states") or [] if not states or not all(item.get("band") == "unassessed" for item in states): raise SmokeError("G4 caller promoted a competency band") if not all(item.get("attempt_count") == 0 for item in states): raise SmokeError("G4 caller invented practice attempts") if g5_payload.get("status") != "failed": raise SmokeError("G5 caller did not preserve failed-only status") if ( g5_payload.get("source_kind"), g5_payload.get("perspective"), ) != ("model_inferred", "independent_observer"): raise SmokeError("G5 independent observation provenance drifted") if set(g5_payload.get("evidence_turn_ids") or []) != expected_evidence: raise SmokeError("G5 observation did not preserve durable turn UUIDs") if g5_payload.get("model_run_turn_id") != counselor_turn_id: raise SmokeError("G5 model run is not anchored to the counselor turn") if g5_payload.get("revealed_sequence") != g5_payload.get("locked_sequence") + 1: raise SmokeError("G5 observation was not revealed after prediction lock") model_metadata = g5_payload.get("model_run_metadata") or {} if model_metadata.get("auto_mastery") is not False: raise SmokeError("G5 model run metadata allowed automatic mastery") if model_metadata.get("auto_transfer") is not False: raise SmokeError("G5 model run metadata allowed automatic transfer") if any(forbidden_payload.values()): raise SmokeError(f"caller auto-promoted forbidden evidence: {forbidden_payload}") return { "ready_source": source_payload, "g4": { "submission_id": g4_payload["submission_id"], "coaching_card_record_id": g4_payload["coaching_card_record_id"], "card_key": g4_payload["card_key"], "prescription_record_id": g4_payload["prescription_record_id"], "prescription_key": g4_payload["prescription_key"], "snapshot_id": g4_payload["snapshot_id"], "decision_id": g4_payload["decision_id"], "competency_id": g4_payload["competency_id"], "scenario_novelty": g4_payload["scenario_novelty"], "evidence_turn_ids": sorted(str(item) for item in expected_evidence), "all_competency_bands": sorted({item["band"] for item in states}), "attempt_count_total": sum(item["attempt_count"] for item in states), }, "g5": { "history_id": g5_payload["history_id"], "prediction_revision_id": g5_payload["prediction_revision_id"], "lock_id": g5_payload["lock_id"], "observation_id": g5_payload["observation_id"], "observation_status": g5_payload["status"], "source_kind": g5_payload["source_kind"], "perspective": g5_payload["perspective"], "model_run_id": g5_payload["model_run_id"], "model_run_turn_id": g5_payload["model_run_turn_id"], "model_run_status": g5_payload["model_run_status"], "agent_role": g5_payload["agent_role"], "provider": g5_payload["provider"], "model": g5_payload["model"], "prompt_bundle_id": g5_payload["prompt_bundle_id"], "prompt_bundle_version": g5_payload["prompt_bundle_version"], "input_evidence_hash": g5_payload["input_evidence_hash"], "revealed_after_lock": True, "evidence_turn_ids": sorted(str(item) for item in expected_evidence), "counterevidence_count": len(g5_payload["counterevidence"] or []), }, "non_promotion": forbidden_payload, } async def _projection_proof( *, learner: Principal, teacher: Principal, learner_id: UUID ) -> dict[str, Any]: learner_g4 = await deliberate_practice_store.read_deliberate_practice( principal=learner ) teacher_g4 = await deliberate_practice_store.read_deliberate_practice( principal=teacher, learner_id=learner_id ) learner_g5 = await calibration_transfer_store.read_calibration_transfer( principal=learner ) teacher_g5 = await calibration_transfer_store.read_calibration_transfer( principal=teacher, learner_id=learner_id ) for payload in (learner_g4, teacher_g4, learner_g5, teacher_g5): if payload.get("clinical_claim_allowed") is not False: raise SmokeError("a role projection allowed a clinical claim") if len(learner_g4["prescriptions"]) != 1 or learner_g4["episodes"]: raise SmokeError("learner G4 projection invented or omitted artifacts") if len(teacher_g4["prescriptions"]) != 1 or teacher_g4["episodes"]: raise SmokeError("teacher G4 projection differs from learner projection") learner_history = learner_g5["prediction_histories"] teacher_history = teacher_g5["prediction_histories"] if len(learner_history) != 1 or len(teacher_history) != 1: raise SmokeError("G5 role projections omitted prediction history") learner_observation = learner_history[0].get("external_observation") or {} teacher_observation = teacher_history[0].get("external_observation") or {} if learner_observation.get("status") != "failed": raise SmokeError("learner projection omitted failed observation") if teacher_observation.get("status") != "failed": raise SmokeError("teacher projection omitted failed observation") if learner_g5["calibration_assessments"] or learner_g5["transfer_suites"]: raise SmokeError("G5 projection auto-created assessment or transfer") return { "learner_teacher_g4_prescription_count": 1, "learner_teacher_g4_episode_count": 0, "learner_teacher_g5_history_count": 1, "learner_teacher_g5_observation_status": "failed", "calibration_assessment_count": 0, "transfer_suite_count": 0, "clinical_claim_allowed": False, } async def run() -> dict[str, Any]: await db.init_pool() try: learner_id = uuid4() teacher_id = uuid4() await _seed_users(learner_id, teacher_id) learner = _principal(learner_id, Role.LEARNER) teacher = _principal(teacher_id, Role.TEACHER) session, counselor_turn_id, client_turn_id = await _create_ready_session( learner ) session_id = UUID(session.session_id) before_lock = await session_learning_producer.produce_session_learning_artifacts( session_id ) if before_lock.get("g4", {}).get("status") != "ready": raise SmokeError(f"G4 was not produced from ready evaluation: {before_lock}") if before_lock.get("g4", {}).get("idempotent_replay") is not False: raise SmokeError("first G4 production was not a new append") if before_lock.get("g5") != { "status": "skipped", "reason": "locked_prediction_missing", }: raise SmokeError("G5 observation escaped the prediction lock") counts_before_lock = await _counts(session_id) if counts_before_lock["g5_observations"] != 0: raise SmokeError("G5 observation exists before prediction lock") suffix = learner_id.hex[:12] history_id = uuid4() revision_response = await calibration_transfer.create_prediction_revision( PredictionRevisionRequest( submission_id=uuid4(), prediction_revision_id=uuid4(), history_id=history_id, session_id=session_id, competency_id=COMPETENCY_ID, practice_block_id=f"oas-g5-block-auto-{suffix}", scenario_variant_id=f"auto-session-{session_id.hex}", phrase_family_id=f"auto-phrase-{suffix}", revision_no=1, predicted_success_probability=0.8, confidence=0.8, recorded_sequence=1, revision_reason="독립 평가 공개 전 자기예측", instrument_id=CALIBRATION_INSTRUMENT_ID, instrument_version=CALIBRATION_INSTRUMENT_VERSION, evidence_turn_ids=[counselor_turn_id], ), learner, ) lock_response = await calibration_transfer.lock_prediction_history( history_id, PredictionLockRequest( submission_id=uuid4(), lock_id=uuid4(), prediction_revision_id=revision_response.prediction_revision_id, locked_sequence=2, ), learner, ) if revision_response.idempotent_replay or lock_response.idempotent_replay: raise SmokeError("first prediction revision/lock unexpectedly replayed") counts_after_lock = await _counts(session_id) replay = await session_learning_producer.produce_locked_prediction_history( history_id ) if replay.get("g4", {}).get("idempotent_replay") is not True: raise SmokeError("G4 stable submission/card replay was not detected") if replay.get("g5") != { "status": "skipped", "reason": "locked_prediction_missing", }: raise SmokeError("G5 replay created or exposed a duplicate observation") counts_after_replay = await _counts(session_id) if counts_after_lock != counts_after_replay: raise SmokeError( "producer replay changed append-only row counts: " f"{counts_after_lock} -> {counts_after_replay}" ) expected_counts = { "g4_submissions": 1, "g4_cards": 1, "g4_prescriptions": 1, "g4_snapshots": 1, "g4_decisions": 1, "g4_episodes": 0, "g4_attempts": 0, "g5_histories": 1, "g5_revisions": 1, "g5_locks": 1, "g5_observations": 1, "producer_model_runs": 1, "g5_assessments": 0, "g5_transfer_suites": 0, } if counts_after_replay != expected_counts: raise SmokeError(f"production caller row counts differ: {counts_after_replay}") postgres = await _postgres_proof( session_id=session_id, counselor_turn_id=counselor_turn_id, client_turn_id=client_turn_id, history_id=history_id, ) projections = await _projection_proof( learner=learner, teacher=teacher, learner_id=learner_id, ) return { "ok": True, "fixture_classification": "synthetic_educational", "clinical_claim_allowed": False, "notice_ko": ( "production caller와 원장 연결을 검증하는 합성 교육 fixture이며 " "임상 효과나 실제 역량 숙달을 주장하지 않는다." ), "fixture_policy": ( "retained unique dev:e2e rows in expendable development DB; " "append-only ledger" ), "entrypoints": [ "session_learning_producer.produce_session_learning_artifacts", "calibration_transfer.lock_prediction_history -> " "session_learning_producer.produce_locked_prediction_history", ], "direct_g4_or_g5_artifact_insert_used": False, "external_model_call_used": False, "session_id": session_id, "learner_id": learner_id, "teacher_id": teacher_id, "durable_turn_ids": [counselor_turn_id, client_turn_id], "before_prediction_lock": { "g4_status": before_lock["g4"]["status"], "g4_idempotent_replay": before_lock["g4"]["idempotent_replay"], "g5_status": before_lock["g5"]["status"], "g5_reason": before_lock["g5"]["reason"], "g5_observation_count": counts_before_lock["g5_observations"], }, "prediction_lock": { "history_id": history_id, "prediction_revision_id": revision_response.prediction_revision_id, "lock_id": lock_response.lock_id, "revision_idempotent_replay": revision_response.idempotent_replay, "lock_idempotent_replay": lock_response.idempotent_replay, }, "replay": { "g4_idempotent_replay": replay["g4"]["idempotent_replay"], "g5_status": replay["g5"]["status"], "g5_reason": replay["g5"]["reason"], "row_counts_stable": True, "row_counts": counts_after_replay, }, "postgres": postgres, "role_projections": projections, } finally: await db.close_pool() def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--out", default="") args = parser.parse_args() result = asyncio.run(run()) output = json.dumps(result, ensure_ascii=False, indent=2, default=str) if args.out: path = Path(args.out) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(output + "\n", encoding="utf-8") print(output) if __name__ == "__main__": main()