"""Outcome & Alliance OS G0 측정 계약의 결정론적 회귀 검사.""" from __future__ import annotations import json import unittest from pathlib import Path from uuid import uuid4 from pydantic import ValidationError from .contracts.measurement import ( AI_VIEWS, INSTRUMENT_KINDS, MEASUREMENT_CONSTRUCTS, MEASUREMENT_PERSPECTIVES, MEASUREMENT_STATUSES, MODEL_RUN_STATUSES, SOURCE_KINDS, BenchmarkCase, MeasurementEvent, ) from .services.measurement_legacy import ( LEGACY_SIGNAL_INVENTORY, adapt_deep_evaluation, adapt_fast_evaluation, adapt_legacy_simulation_signals, adapt_phase3_metric, require_homogeneous_provenance, ) REPO_ROOT = Path(__file__).resolve().parents[3] BENCHMARK_PATH = ( REPO_ROOT / "apps" / "api" / "app" / "data" / "outcome_alliance_benchmark_g0.v1.json" ) CONTRACT_PATH = ( REPO_ROOT / "apps" / "api" / "app" / "contracts" / "measurement_contract.v1.json" ) SQL_PATH = REPO_ROOT / "infra" / "db" / "init" / "07_measurement_foundation.sql" def valid_event_payload(**overrides: object) -> dict[str, object]: payload: dict[str, object] = { "session_id": uuid4(), "construct": "working_alliance", "dimension": "goal", "perspective": "learner_self_report", "source_kind": "learner_reported", "instrument_id": "pilot-alliance-pulse", "instrument_version": "design-1", "value": 0.75, "scale_min": 0.0, "scale_max": 1.0, "visible_to": ("counselor", "evaluator"), } payload.update(overrides) return payload class MeasurementContractTest(unittest.TestCase): def test_external_contract_keeps_construct_alias_without_pydantic_collision(self) -> None: event = MeasurementEvent.model_validate(valid_event_payload()) self.assertEqual(event.construct_key, "working_alliance") self.assertEqual(event.model_dump(by_alias=True)["construct"], "working_alliance") schema = MeasurementEvent.model_json_schema() self.assertIn("construct", schema["properties"]) self.assertNotIn("construct_key", schema["properties"]) self.assertIn("construct", schema["required"]) def test_measurement_truth_guards_reject_invalid_layers_and_scores(self) -> None: invalid_payloads = ( valid_event_payload(perspective="independent_observer"), valid_event_payload(scale_min=1.0, scale_max=1.0), valid_event_payload(value=1.1), valid_event_payload(value=None, status="ready"), valid_event_payload(value=0.5, status="error", error_code="failed"), valid_event_payload(value=None, status="error", error_code=None), ) for payload in invalid_payloads: with self.subTest(payload=payload): with self.assertRaises(ValidationError): MeasurementEvent.model_validate(payload) def test_model_and_agent_measurements_require_model_run_provenance(self) -> None: for source_kind, perspective in ( ("model_inferred", "independent_observer"), ("agent_reported", "client_agent_report"), ): with self.subTest(source_kind=source_kind): with self.assertRaises(ValidationError): MeasurementEvent.model_validate( valid_event_payload( source_kind=source_kind, perspective=perspective, model_run_id=None, ) ) event = MeasurementEvent.model_validate( valid_event_payload( source_kind=source_kind, perspective=perspective, model_run_id=uuid4(), ) ) self.assertIsNotNone(event.model_run_id) def test_legacy_simulation_signals_never_become_clinical_alliance(self) -> None: session_id = uuid4() events = adapt_legacy_simulation_signals( session_id=session_id, rapport_credit=0.6, alliance_level=0.7, ) self.assertEqual(len(events), 2) for event in events: self.assertEqual(event.source_kind, "simulated_state") self.assertEqual(event.perspective, "client_simulation") self.assertEqual(event.construct_key, "simulation_progress") self.assertFalse(event.metadata["clinical_claim_allowed"]) inventory = {item.signal: item for item in LEGACY_SIGNAL_INVENTORY} self.assertFalse(inventory["session_state.rapport_credit"].clinical_claim_allowed) self.assertFalse(inventory["case_profile.alliance_level"].clinical_claim_allowed) def test_legacy_model_adapters_keep_model_and_evidence_provenance(self) -> None: session_id = uuid4() turn_id = uuid4() model_run_id = uuid4() fast = adapt_fast_evaluation( session_id=session_id, turn_id=turn_id, evaluation={"appropriateness": "pos", "rapport_signal": -0.5}, model_run_id=model_run_id, ) self.assertEqual([event.value for event in fast], [1.0, -0.5]) for event in fast: self.assertEqual(event.source_kind, "model_inferred") self.assertEqual(event.model_run_id, model_run_id) self.assertEqual(event.evidence_turn_ids, (turn_id,)) deep = adapt_deep_evaluation( session_id=session_id, evaluation={"distribution": {"total": 7}}, model_run_id=model_run_id, evidence_turn_ids=(turn_id,), ) self.assertEqual(deep.value, 7.0) self.assertEqual(deep.dimension, "technique_occurrence_count") def test_failed_legacy_model_runs_emit_error_event_without_score(self) -> None: event = adapt_deep_evaluation( session_id=uuid4(), evaluation={"error": "provider_timeout"}, model_run_id=uuid4(), ) self.assertEqual(event.status, "error") self.assertEqual(event.error_code, "provider_timeout") self.assertIsNone(event.value) def test_phase3_self_report_and_runtime_layers_are_separate(self) -> None: session_id = uuid4() learner = adapt_phase3_metric( session_id=session_id, metric_name="self_efficacy_prepost", value=0.8, ) runtime = adapt_phase3_metric( session_id=session_id, metric_name="completion_rate", value=0.9, ) self.assertEqual((learner.source_kind, learner.perspective), ( "learner_reported", "learner_self_report", )) self.assertEqual((runtime.source_kind, runtime.perspective), ( "observed_runtime", "runtime_observation", )) with self.assertRaisesRegex(ValueError, "heterogeneous measurement provenance"): require_homogeneous_provenance( (learner, runtime), operation="phase3_total_score", ) def test_benchmark_pack_has_exactly_eight_versioned_truth_cases(self) -> None: payload = json.loads(BENCHMARK_PATH.read_text(encoding="utf-8")) self.assertEqual(payload["schema"], "vignette.outcome_alliance_benchmark.v1") cases = tuple(BenchmarkCase.model_validate(item) for item in payload["cases"]) self.assertEqual(len(cases), 8) self.assertEqual(len({case.case_id for case in cases}), 8) self.assertEqual( {case.scene_type for case in cases}, { "goal_mismatch", "task_mismatch", "empathic_miss", "withdrawal", "confrontation", "successful_repair", "failed_repair", "warm_but_directionless", }, ) for case in cases: self.assertEqual(case.version, "1.0.0") self.assertTrue(case.forbidden_claims) for expectation in case.expected: self.assertLess( max(expectation.evidence_turn_indices), len(case.turns), ) def test_generated_contract_enums_match_python_ssot(self) -> None: contract = json.loads(CONTRACT_PATH.read_text(encoding="utf-8")) self.assertEqual(contract["enums"]["sourceKinds"], list(SOURCE_KINDS)) self.assertEqual(contract["enums"]["constructs"], list(MEASUREMENT_CONSTRUCTS)) self.assertEqual( contract["enums"]["perspectives"], list(MEASUREMENT_PERSPECTIVES) ) self.assertEqual( contract["enums"]["measurementStatuses"], list(MEASUREMENT_STATUSES) ) self.assertEqual(contract["enums"]["instrumentKinds"], list(INSTRUMENT_KINDS)) self.assertEqual(contract["enums"]["aiViews"], list(AI_VIEWS)) self.assertEqual( contract["enums"]["modelRunStatuses"], list(MODEL_RUN_STATUSES) ) def test_sql_ledger_is_append_only_rls_guarded_and_seeded(self) -> None: sql = SQL_PATH.read_text(encoding="utf-8") self.assertIn("BEFORE UPDATE OR DELETE ON app.measurement_event", sql) self.assertIn("BEFORE UPDATE OR DELETE ON audit.model_run", sql) self.assertIn("ALTER TABLE app.measurement_event ENABLE ROW LEVEL SECURITY", sql) self.assertIn("ALTER TABLE audit.model_run ENABLE ROW LEVEL SECURITY", sql) self.assertIn("CREATE POLICY p_measurement_event_select", sql) self.assertIn("CREATE POLICY p_measurement_event_insert", sql) self.assertNotIn("CREATE POLICY p_measurement_event_update", sql) self.assertNotIn("CREATE POLICY p_measurement_event_delete", sql) for instrument_id in ( "vignette-state-machine", "vignette-alliance-ewma", "vignette-fast-evaluator", "vignette-deep-evaluator", "phase3-prepost", "phase3-runtime-kpi", ): self.assertIn(f"'{instrument_id}'", sql) if __name__ == "__main__": unittest.main()