from __future__ import annotations import unittest from pathlib import Path from uuid import uuid4 from pydantic import ValidationError from .contracts.outcome_trajectory import ( LongitudinalOutcomeInput, OutcomeAxisObservation, RelationshipMemoryEvent, SyntheticExpectedArc, ) from .services.outcome_trajectory import ( assess_longitudinal_outcome, build_role_safe_read_model, evaluate_trajectory_benchmark, load_trajectory_benchmark, project_relationship_memory, render_benchmark_report, ) BENCHMARK_PATH = ( Path(__file__).resolve().parent / "data" / "outcome_trajectory_benchmark_g2.v1.json" ) class OutcomeTrajectoryContractTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_trajectory_benchmark(BENCHMARK_PATH) def test_synthetic_arc_requires_all_three_axes_for_sessions_one_to_five(self) -> None: payload = self.pack.expected_arc.model_dump() payload["distributions"] = payload["distributions"][:-1] with self.assertRaisesRegex( ValidationError, "every outcome axis for sessions 1..5" ): SyntheticExpectedArc.model_validate(payload) def test_synthetic_arc_cannot_enable_clinical_claims(self) -> None: payload = self.pack.expected_arc.model_dump() payload["clinical_claim_allowed"] = True with self.assertRaises(ValidationError): SyntheticExpectedArc.model_validate(payload) def test_missing_observation_cannot_carry_imputed_value(self) -> None: with self.assertRaisesRegex( ValidationError, "missing/error outcomes must remain scoreless" ): OutcomeAxisObservation( axis="distress_load", status="missing", value=0.5, confidence=0.5, source_kind="simulated_state", instrument_id="test", instrument_version="1.0.0", missing_reason="not_collected", ) def test_outcome_observation_enforces_source_perspective_provenance(self) -> None: with self.assertRaisesRegex( ValidationError, "mixes source and perspective layers" ): OutcomeAxisObservation( axis="daily_functioning", value=0.5, confidence=0.8, source_kind="learner_reported", perspective="independent_observer", instrument_id="test", instrument_version="1.0.0", evidence_refs=("turn-1",), ) with self.assertRaisesRegex( ValidationError, "require model_run_id provenance" ): OutcomeAxisObservation( axis="daily_functioning", value=0.5, confidence=0.8, source_kind="model_inferred", perspective="independent_observer", instrument_id="test", instrument_version="1.0.0", evidence_refs=("turn-1",), ) valid = OutcomeAxisObservation( axis="daily_functioning", value=0.5, confidence=0.8, source_kind="model_inferred", perspective="independent_observer", instrument_id="test", instrument_version="1.0.0", model_run_id=uuid4(), evidence_refs=("turn-1",), ) self.assertIsNotNone(valid.model_run_id) def test_relationship_summary_keys_must_exactly_match_visible_roles(self) -> None: with self.assertRaisesRegex( ValidationError, "summaries must exactly match visible_to" ): RelationshipMemoryEvent( event_id="role-leak", session_no=1, event_type="unresolved_rupture", visible_to=("client",), summaries={ "client": "말하지 못한 부담이 남아 있다.", "counselor": "이 문장은 노출되면 안 된다.", }, evidence_refs=("turn-1",), ) class OutcomeTrajectoryAssessmentTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_trajectory_benchmark(BENCHMARK_PATH) def _result(self, case_index: int): case = self.pack.cases[case_index] return assess_longitudinal_outcome( LongitudinalOutcomeInput( expected_arc=self.pack.expected_arc, sessions=case.sessions, ) ) def test_sessions_one_to_five_cover_on_track_watch_and_deteriorating(self) -> None: result = self._result(1) self.assertEqual( [item.status for item in result.sessions], ["on_track", "watch", "deteriorating", "deteriorating", "deteriorating"], ) third = result.sessions[2] self.assertEqual( {item.status for item in third.axes}, {"deteriorating"}, ) self.assertTrue(third.next_check_questions) self.assertTrue( any("측정 시점" in question for question in third.next_check_questions) ) def test_single_session_large_deviation_is_off_track_not_a_fake_trend(self) -> None: case = self.pack.cases[0] first = case.sessions[0] axes = list(first.axes) axes[0] = axes[0].model_copy(update={"value": 0.90}) changed = first.model_copy(update={"axes": tuple(axes)}) result = assess_longitudinal_outcome( LongitudinalOutcomeInput( expected_arc=self.pack.expected_arc, sessions=(changed,), ) ) distress = result.sessions[0].axes[0] self.assertEqual(distress.status, "off_track") self.assertIn( "single_session_deviation_not_yet_a_worsening_trend", distress.counterevidence, ) def test_transient_watch_exposes_uncertainty_and_avoids_false_alert(self) -> None: result = self._result(2) second = result.sessions[1] self.assertEqual(second.status, "watch") self.assertTrue(all(item.uncertainty >= 0.28 for item in second.axes)) self.assertTrue( all("inside_off_track_threshold" in item.counterevidence for item in second.axes) ) self.assertEqual(result.sessions[2].status, "on_track") def test_missing_axis_stays_scoreless_and_does_not_poison_later_session(self) -> None: result = self._result(3) second = result.sessions[1] missing = next( item for item in second.axes if item.axis == "learning_engagement" ) self.assertEqual(second.status, "insufficient_evidence") self.assertEqual(second.missing_axes, ("learning_engagement",)) self.assertIsNone(missing.observed_value) self.assertIsNone(missing.adverse_z) self.assertEqual(missing.uncertainty, 1.0) self.assertIn("no_value_imputation", missing.decision_basis) self.assertEqual(result.sessions[2].status, "on_track") def test_gap_in_measurement_cannot_create_a_fake_deterioration_trend(self) -> None: case = self.pack.cases[3] third = case.sessions[2] axes = list(third.axes) axes[2] = axes[2].model_copy(update={"value": 0.25}) changed_sessions = list(case.sessions[:3]) changed_sessions[2] = third.model_copy(update={"axes": tuple(axes)}) result = assess_longitudinal_outcome( LongitudinalOutcomeInput( expected_arc=self.pack.expected_arc, sessions=tuple(changed_sessions), ) ) engagement = result.sessions[2].axes[2] self.assertEqual(engagement.status, "off_track") self.assertIsNone(engagement.adverse_z_change) self.assertIn( "single_session_deviation_not_yet_a_worsening_trend", engagement.counterevidence, ) def test_safety_signal_is_returned_but_never_changes_outcome_classification(self) -> None: case = self.pack.cases[0] with_safety = self._result(0) sessions_without_safety = tuple( item.model_copy(update={"safety_signals": ()}) for item in case.sessions ) without_safety = assess_longitudinal_outcome( LongitudinalOutcomeInput( expected_arc=self.pack.expected_arc, sessions=sessions_without_safety, ) ) self.assertEqual( [item.status for item in with_safety.sessions], [item.status for item in without_safety.sessions], ) self.assertEqual(len(with_safety.sessions[1].safety_signals), 1) decision_text = " ".join( reason for axis in with_safety.sessions[1].axes for reason in axis.decision_basis ) self.assertNotIn("safety", decision_text) def test_assessment_has_axis_rows_and_no_compensating_total_score(self) -> None: payload = self._result(1).model_dump() self.assertNotIn("total", payload) self.assertNotIn("score", payload) self.assertEqual( [axis["axis"] for axis in payload["sessions"][2]["axes"]], ["distress_load", "daily_functioning", "learning_engagement"], ) self.assertFalse(payload["clinical_claim_allowed"]) self.assertEqual(payload["data_classification"], "synthetic_educational") class OutcomeTrajectoryMemoryAndBenchmarkTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_trajectory_benchmark(BENCHMARK_PATH) cls.case = cls.pack.cases[0] cls.events = tuple( event for session in cls.case.sessions for event in session.relationship_events ) def test_role_safe_projection_does_not_leak_client_only_rupture_to_counselor(self) -> None: client = project_relationship_memory(self.events, view="client") counselor = project_relationship_memory(self.events, view="counselor") self.assertIn("b001-private-rupture", {item.event_id for item in client}) self.assertNotIn("b001-private-rupture", {item.event_id for item in counselor}) self.assertNotIn( "동의한 척했다", " ".join(item.summary for item in counselor), ) self.assertEqual( [item.session_no for item in counselor], sorted(item.session_no for item in counselor), ) def test_role_safe_read_model_keeps_outcome_safety_and_memory_separate(self) -> None: model = build_role_safe_read_model( LongitudinalOutcomeInput( expected_arc=self.pack.expected_arc, sessions=self.case.sessions, ), view="supervisor", ) self.assertEqual(len(model.assessment.sessions), 5) self.assertEqual(len(model.safety_signals), 1) self.assertEqual(len(model.relationship_memory), 3) self.assertEqual(model.assessment.sessions[1].status, "on_track") def test_deterministic_benchmark_meets_early_warning_and_false_alert_gate(self) -> None: report = evaluate_trajectory_benchmark(self.pack) self.assertEqual(report["case_count"], 4) self.assertEqual(report["session_count"], 20) self.assertEqual(report["early_warning_recall"], 1.0) self.assertEqual(report["false_alert_rate"], 0.0) self.assertEqual(report["status_accuracy"], 1.0) self.assertEqual( report["confusion"], { "true_positive": 3, "false_negative": 0, "false_positive": 0, "true_negative": 17, }, ) def test_benchmark_report_labels_synthetic_scope_and_uncertainty(self) -> None: report = evaluate_trajectory_benchmark(self.pack) rendered = render_benchmark_report(report) self.assertIn('"data_classification": "synthetic_educational"', rendered) self.assertIn('"clinical_claim_allowed": false', rendered) self.assertIn('"false_alert_counterevidence"', rendered) self.assertIn('"uncertainty"', rendered) if __name__ == "__main__": unittest.main()