"""회기말 평가에서 G4 처방과 G5 독립 관찰을 파생하는 production worker. 평가 원장은 이미 커밋된 뒤 이 worker가 실행된다. 따라서 파생 원장 장애는 회기 평가 저장을 되돌리지 않는다. 카드/관찰은 durable turn UUID와 구조화된 evaluator 판정이 함께 있을 때만 만들며, 성공·mastery·transfer는 자동 추론하지 않는다. """ from __future__ import annotations import asyncio import hashlib import json import logging import re from dataclasses import dataclass from typing import Any, Mapping from uuid import UUID, uuid5 from .. import db from ..contracts.deliberate_practice import ( CoachingCard, CompetencyDefinition, CompetencyGraph, CompetencyState, PracticeEvidenceRef, PracticeTargetSpec, ReplayActivity, ) from . import calibration_transfer_store, deliberate_practice_store logger = logging.getLogger(__name__) _PRODUCER_NAMESPACE = UUID("20ae3e1e-4a36-5b22-9794-6cb0248b1740") _PRODUCER_VERSION = "session-learning-producer-v1" _CALIBRATION_INSTRUMENT_ID = "calibration-mirror-g5" _CALIBRATION_INSTRUMENT_VERSION = "1.0.0" _OBSERVATION_UNCERTAINTY = 0.5 _SEVERITY_RANK = {"major": 3, "moderate": 2, "minor": 1} def _canonical_hash(value: Any) -> str: encoded = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() _RULESET_HASH = _canonical_hash( { "version": _PRODUCER_VERSION, "ready_session_evaluation_required": True, "durable_turn_uuid_required": True, "fast_and_deep_dimension_agreement_required": True, "auto_observation_status": "failed_only", "auto_mastery": False, "auto_transfer": False, } ) @dataclass(frozen=True, slots=True) class CompetencySpec: competency_id: str criterion_id: str label_ko: str description: str observable_behavior: str @dataclass(frozen=True, slots=True) class DurableDeviation: turn_id: UUID turn_seq: int response_turn_id: UUID | None response_turn_seq: int | None dimension: str severity: str spec: CompetencySpec @property def evidence_turn_ids(self) -> tuple[UUID, ...]: if self.response_turn_id is None: return (self.turn_id,) return (self.turn_id, self.response_turn_id) _COMPETENCIES = { "empathic_reflection": CompetencySpec( competency_id="competency.empathic_reflection", criterion_id="criterion.reflect-and-check", label_ko="공감적 반영", description="내담자의 핵심 정서를 짧게 반영하고 실제로 맞게 이해했는지 확인하는 미세기술이다.", observable_behavior="핵심 정서를 한 문장으로 반영한 뒤 내담자에게 이해가 맞는지 확인한다.", ), "open_question": CompetencySpec( competency_id="competency.open_question", criterion_id="criterion.open-question-one-focus", label_ko="개방형 질문", description="한 번에 하나의 초점을 유지하며 내담자의 탐색을 넓히는 개방형 질문 기술이다.", observable_behavior="한 번에 하나의 초점만 담은 개방형 질문으로 내담자의 탐색을 이어간다.", ), "rupture_repair": CompetencySpec( competency_id="competency.rupture_repair", criterion_id="criterion.name-and-repair-rupture", label_ko="관계 균열 수선", description="관계의 긴장이나 단절 신호를 알아차리고 명시적으로 확인하여 다시 협력하는 기술이다.", observable_behavior="관계의 긴장 신호를 짚고 자신의 영향을 확인한 뒤 수선 질문을 한 번 제시한다.", ), "collaborative_goal": CompetencySpec( competency_id="competency.collaborative_goal", criterion_id="criterion.confirm-shared-goal", label_ko="협력적 목표 합의", description="상담자의 목표를 앞세우지 않고 내담자의 언어로 회기 목표를 함께 합의하는 기술이다.", observable_behavior="내담자의 표현을 사용해 이번 대화의 목표가 맞는지 명시적으로 합의한다.", ), } def _dimension_key(value: object) -> str: return re.sub(r"[^a-z0-9가-힣]+", "_", str(value or "").strip().lower()).strip("_") def _spec_for_dimension(value: object) -> CompetencySpec | None: key = _dimension_key(value) if not key: return None if any( token in key for token in ( "reflection", "empathy", "empathic", "공감", "정서반영", "감정반영", ) ): return _COMPETENCIES["empathic_reflection"] if any( token in key for token in ("open_question", "openquestion", "개방형질문", "열린질문") ): return _COMPETENCIES["open_question"] if any(token in key for token in ("rupture", "repair", "균열", "수선", "관계회복")): return _COMPETENCIES["rupture_repair"] if any( token in key for token in ( "collaborative_goal", "goal_collaboration", "goal_alignment", "공동목표", "협력적목표", "목표합의", ) ): return _COMPETENCIES["collaborative_goal"] return None def _value(row: Mapping[str, Any] | Any, key: str, default: Any = None) -> Any: try: return row[key] except (KeyError, TypeError): return getattr(row, key, default) def _deep_competency_ids(payload: Mapping[str, Any]) -> set[str]: identifiers: set[str] = set() deviations = payload.get("intent_deviations") if not isinstance(deviations, list): return identifiers for item in deviations: if not isinstance(item, Mapping): continue spec = _spec_for_dimension(item.get("dimension")) if spec is not None: identifiers.add(spec.competency_id) return identifiers async def _load_ready_source( conn: Any, *, session_id: UUID, ) -> tuple[Mapping[str, Any], tuple[DurableDeviation, ...]] | None: evaluation = await conn.fetchrow( """ SELECT e.status, e.scope, e.payload, s.learner_id FROM app.session_evaluation e JOIN app.sessions s ON s.id = e.session_id WHERE e.session_id = $1 """, session_id, ) if evaluation is None: return None if ( _value(evaluation, "status") != "ready" or _value(evaluation, "scope") != "session_end" ): return None payload = _value(evaluation, "payload", {}) if not isinstance(payload, Mapping): return None deep_competencies = _deep_competency_ids(payload) if not deep_competencies: return evaluation, () rows = await conn.fetch( """ SELECT t.id AS turn_id, t.seq AS turn_seq, c.intent_deviation, response.id AS response_turn_id, response.seq AS response_turn_seq FROM app.turns t JOIN LATERAL ( SELECT sc.intent_deviation FROM app.supervisor_comment sc WHERE sc.turn_id = t.id AND sc.intent_deviation IS NOT NULL ORDER BY sc.created_at DESC, sc.id DESC LIMIT 1 ) c ON TRUE LEFT JOIN LATERAL ( SELECT next_turn.id, next_turn.seq FROM app.turns next_turn WHERE next_turn.session_id = t.session_id AND next_turn.speaker = 'client' AND next_turn.seq > t.seq ORDER BY next_turn.seq LIMIT 1 ) response ON TRUE WHERE t.session_id = $1 AND t.speaker = 'counselor' ORDER BY t.seq """, session_id, ) signals: list[DurableDeviation] = [] for row in rows: deviation = _value(row, "intent_deviation", {}) if not isinstance(deviation, Mapping): continue spec = _spec_for_dimension(deviation.get("dimension")) if spec is None or spec.competency_id not in deep_competencies: continue try: turn_id = UUID(str(_value(row, "turn_id"))) response_value = _value(row, "response_turn_id") response_turn_id = UUID(str(response_value)) if response_value else None severity = str(deviation.get("severity") or "minor") if severity not in _SEVERITY_RANK: severity = "minor" signals.append( DurableDeviation( turn_id=turn_id, turn_seq=int(_value(row, "turn_seq")), response_turn_id=response_turn_id, response_turn_seq=( int(_value(row, "response_turn_seq")) if response_turn_id is not None else None ), dimension=str(deviation.get("dimension") or ""), severity=severity, spec=spec, ) ) except (TypeError, ValueError): continue signals.sort(key=lambda item: (-_SEVERITY_RANK[item.severity], -item.turn_seq)) return evaluation, tuple(signals) def _coaching_card( session_id: UUID, signal: DurableDeviation, *, difficulty_level: int, ) -> CoachingCard: token = hashlib.sha256( f"{session_id}:{signal.turn_id}:{signal.spec.competency_id}".encode("utf-8") ).hexdigest()[:20] scene_id = f"session-{session_id.hex}-turn-{signal.turn_seq}" evidence_refs = [ PracticeEvidenceRef( ref_id=str(signal.turn_id), scene_id=scene_id, turn_index=signal.turn_seq, actor="learner", kind="learner_behavior", ) ] if signal.response_turn_id is not None and signal.response_turn_seq is not None: evidence_refs.append( PracticeEvidenceRef( ref_id=str(signal.response_turn_id), scene_id=scene_id, turn_index=signal.response_turn_seq, actor="client", kind="client_response", ) ) return CoachingCard( card_id=f"oas-g4-card-auto-{token}", scene_id=scene_id, coach_claim=( f"{signal.spec.label_ko} 이탈이 확인된 장면을 다시 열어 " f"{signal.spec.observable_behavior}" ), evidence_refs=tuple(evidence_refs), source_refs=( f"session-evaluation:{session_id}:session_end", f"turn-evaluation:{signal.turn_id}", f"producer:{_PRODUCER_VERSION}", ), uncertainty=_OBSERVATION_UNCERTAINTY, counterevidence=( f"intent_deviation:{_dimension_key(signal.dimension)}:{signal.severity}", "unseen_transfer_not_verified", ), targets=( PracticeTargetSpec( prescription_id=f"oas-g4-practice-auto-{token}", competency_id=signal.spec.competency_id, criterion_id=signal.spec.criterion_id, observable_behavior=signal.spec.observable_behavior, activity=ReplayActivity( scenario_variant_id=f"session-{session_id.hex}-turn-{signal.turn_seq}-replay", scenario_novelty="familiar", difficulty_level=difficulty_level, pause_at_evidence_ref=str(signal.turn_id), ), ), ), ) def _initial_graph() -> CompetencyGraph: specs = tuple(_COMPETENCIES.values()) return CompetencyGraph( definitions=tuple( CompetencyDefinition( competency_id=spec.competency_id, label_ko=spec.label_ko, description=spec.description, ) for spec in specs ), states=tuple( CompetencyState( competency_id=spec.competency_id, band="unassessed", forgetting_risk=0.0, uncertainty=1.0, attempt_count=0, familiar_demonstrations=0, unseen_transfer_demonstrations=0, highest_familiar_difficulty=0, evidence_refs=(), counterevidence=("unseen_transfer_not_verified",), ) for spec in specs ), ) async def _produce_g4(conn: Any, *, session_id: UUID) -> dict[str, Any]: source = await _load_ready_source(conn, session_id=session_id) if source is None: return {"status": "skipped", "reason": "ready_session_evaluation_missing"} evaluation, signals = source if not signals: return {"status": "skipped", "reason": "durable_actionable_deviation_missing"} submission_id = uuid5(_PRODUCER_NAMESPACE, f"g4-prescription:{session_id}") existing_snapshot = await conn.fetchrow( """ SELECT graph_payload FROM app.competency_graph_snapshot WHERE source_prescription_submission_id = $1 """, submission_id, ) latest_snapshot = existing_snapshot if latest_snapshot is None: latest_snapshot = await conn.fetchrow( """ SELECT graph_payload FROM app.competency_graph_snapshot WHERE learner_id = $1 ORDER BY snapshot_no DESC LIMIT 1 """, UUID(str(_value(evaluation, "learner_id"))), ) graph = ( CompetencyGraph.model_validate(_value(latest_snapshot, "graph_payload")) if latest_snapshot is not None else None ) state_by_competency = None if graph is not None: state_by_competency = { state.competency_id: state for state in graph.states if state.band != "transfer_verified" and not ( state.familiar_demonstrations >= 2 and state.highest_familiar_difficulty >= 5 ) } signal = next( ( item for item in signals if state_by_competency is None or item.spec.competency_id in state_by_competency ), None, ) if signal is None: return { "status": "skipped", "reason": "compatible_unmastered_competency_missing", } if graph is None: graph = _initial_graph() state = next( item for item in graph.states if item.competency_id == signal.spec.competency_id ) difficulty_level = ( min(5, state.highest_familiar_difficulty + 1) if state.familiar_demonstrations >= 2 else 1 ) result = await deliberate_practice_store.append_prescription_submission( conn=conn, session_id=session_id, submission_id=submission_id, coaching_cards=( _coaching_card( session_id, signal, difficulty_level=difficulty_level, ), ), graph=graph, evidence_turn_ids=signal.evidence_turn_ids, ) return {"status": "ready", **result} async def _ensure_observation_model_run( conn: Any, *, session_id: UUID, history_id: UUID, signal: DurableDeviation, evaluation_payload: Mapping[str, Any], ) -> UUID: input_payload = { "session_id": str(session_id), "history_id": str(history_id), "evaluation_hash": _canonical_hash(evaluation_payload), "competency_id": signal.spec.competency_id, "dimension": _dimension_key(signal.dimension), "severity": signal.severity, "evidence_turn_ids": [str(item) for item in signal.evidence_turn_ids], } input_hash = _canonical_hash(input_payload) model_run_id = uuid5( _PRODUCER_NAMESPACE, f"g5-observation-model:{history_id}:{input_hash}", ) await conn.execute( """ INSERT INTO audit.model_run ( model_run_id, session_id, turn_id, agent_role, provider, model, prompt_bundle_id, prompt_bundle_version, prompt_bundle_hash, structured_schema_version, input_evidence_hash, status, metadata ) VALUES ( $1,$2,$3,'evaluator','vignette-runtime','session-evaluation-observation-adapter', 'session-learning-producer',$4,$5, 'calibration-performance-observation-1',$6,'ready',$7::jsonb ) ON CONFLICT (model_run_id) DO NOTHING """, model_run_id, session_id, signal.turn_id, _PRODUCER_VERSION, _RULESET_HASH, input_hash, { "source": "ready_session_evaluation_and_fast_turn_evaluation", "dimension": _dimension_key(signal.dimension), "severity": signal.severity, "auto_mastery": False, "auto_transfer": False, }, ) return model_run_id async def _produce_g5(conn: Any, *, session_id: UUID) -> dict[str, Any]: source = await _load_ready_source(conn, session_id=session_id) if source is None: return {"status": "skipped", "reason": "ready_session_evaluation_missing"} evaluation, signals = source if not signals: return {"status": "skipped", "reason": "durable_actionable_deviation_missing"} rows = await conn.fetch( """ SELECT h.history_id, h.competency_id, l.locked_sequence FROM app.calibration_prediction_history h JOIN app.calibration_prediction_lock l ON l.history_id = h.history_id LEFT JOIN app.calibration_performance_observation o ON o.history_id = h.history_id WHERE h.session_id = $1 AND o.history_id IS NULL ORDER BY h.created_at, h.history_id """, session_id, ) if not rows: return {"status": "skipped", "reason": "locked_prediction_missing"} produced: list[dict[str, Any]] = [] for row in rows: competency_id = str(_value(row, "competency_id")) signal = next( (item for item in signals if item.spec.competency_id == competency_id), None, ) if signal is None: continue history_id = UUID(str(_value(row, "history_id"))) model_run_id = await _ensure_observation_model_run( conn, session_id=session_id, history_id=history_id, signal=signal, evaluation_payload=_value(evaluation, "payload", {}), ) result = await calibration_transfer_store.append_performance_observation( conn=conn, submission_id=uuid5( _PRODUCER_NAMESPACE, f"g5-observation-submission:{history_id}" ), observation_id=uuid5(_PRODUCER_NAMESPACE, f"g5-observation:{history_id}"), history_id=history_id, status="failed", source_kind="model_inferred", perspective="independent_observer", model_run_id=model_run_id, instrument_id=_CALIBRATION_INSTRUMENT_ID, instrument_version=_CALIBRATION_INSTRUMENT_VERSION, uncertainty=_OBSERVATION_UNCERTAINTY, evidence_turn_ids=signal.evidence_turn_ids, counterevidence=( f"intent_deviation:{_dimension_key(signal.dimension)}:{signal.severity}", ), revealed_sequence=int(_value(row, "locked_sequence")) + 1, ) produced.append(result) if not produced: return {"status": "skipped", "reason": "locked_competency_evidence_mismatch"} return {"status": "ready", "observations": produced} async def produce_session_learning_artifacts(session_id: str | UUID) -> dict[str, Any]: """G4/G5를 독립 트랜잭션으로 실행해 한쪽 장애를 다른 쪽과 격리한다.""" session_uuid = UUID(str(session_id)) results: dict[str, Any] = {} for key, producer in (("g4", _produce_g4), ("g5", _produce_g5)): try: async with db.acquire(ai_view="evaluator", ai_context=True) as conn: results[key] = await producer(conn, session_id=session_uuid) except asyncio.CancelledError: raise except Exception as exc: logger.exception( "session learning producer failed: track=%s session_id=%s", key, session_uuid, ) results[key] = {"status": "failed", "error": type(exc).__name__} return results async def produce_locked_prediction_history(history_id: str | UUID) -> dict[str, Any]: """잠금이 평가보다 늦게 생기는 UI 흐름도 같은 session worker로 수렴시킨다.""" history_uuid = UUID(str(history_id)) try: async with db.acquire(ai_view="evaluator", ai_context=True) as conn: session_id = await conn.fetchval( "SELECT session_id FROM app.calibration_prediction_history WHERE history_id = $1", history_uuid, ) except asyncio.CancelledError: raise except Exception as exc: logger.exception( "locked prediction session lookup failed: history_id=%s", history_uuid ) return {"status": "failed", "error": type(exc).__name__} if session_id is None: return {"status": "skipped", "reason": "prediction_history_missing"} return await produce_session_learning_artifacts(UUID(str(session_id))) __all__ = [ "produce_locked_prediction_history", "produce_session_learning_artifacts", ]