16e791e044
8월 7일까지 워킹트리에만 남아 있던 미커밋 작업을 커밋한다. 여러 사본 폴더(worktree·clone)에 흩어져 있던 중간 스냅샷을 정리하기 전에 원본을 git 이력으로 고정하는 것이 목적이다. - contracts/routes/services: measurement, outcome_trajectory, rupture_repair, deliberate_practice, calibration_transfer, supervision_research, multimodal_alliance, continuous_improvement 계열 신규 모듈과 테스트 - infra/db/init: 07~16 마이그레이션(측정 기반~calibration transfer 실행) - apps/web: 세션 리뷰 카드·관리 화면·E2E 스펙 추가 - docs/ops: G0~G8 라이브 통합·배포·롤백 증거 문서와 evidence JSON/PNG - scripts: smoke·ledger·릴리스 에이전트·NAS 프리뷰 운영 스크립트 engine.public 로그 .bak과 apps/web/test-results 산출물은 커밋에서 제외했다.
334 lines
12 KiB
Python
334 lines
12 KiB
Python
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()
|