from __future__ import annotations import unittest from pathlib import Path from uuid import uuid4 from pydantic import ValidationError from .contracts.deliberate_practice import ( PRACTICE_MODES, CoachingCard, CompetencyDefinition, CompetencyGraph, CompetencyState, CriterionObservation, DifficultyLadderActivity, PracticeAttemptObservation, PracticeBenchmarkPack, PracticeEpisodeAssessment, ) from .services.deliberate_practice import ( apply_episode_to_competency_graph, assess_practice_episode, evaluate_practice_benchmark, load_practice_benchmark, prescribe_from_coaching_cards, render_practice_benchmark_report, select_next_practice, ) BENCHMARK_PATH = ( Path(__file__).resolve().parent / "data" / "deliberate_practice_benchmark_g4.v1.json" ) def _all_keys(value: object) -> set[str]: if isinstance(value, dict): return set(value) | set().union(*(_all_keys(item) for item in value.values())) if isinstance(value, (list, tuple)): return set().union(*(_all_keys(item) for item in value)) if value else set() return set() class DeliberatePracticeContractTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_practice_benchmark(BENCHMARK_PATH) def test_benchmark_is_version_fixed_synthetic_and_covers_all_modes(self) -> None: self.assertEqual(self.pack.version, "1.0.0") self.assertEqual(self.pack.data_classification, "synthetic_educational") self.assertFalse(self.pack.clinical_claim_allowed) modes = { target.activity.mode for case in self.pack.cases for card in case.coaching_cards for target in card.targets } self.assertEqual(modes, set(PRACTICE_MODES)) def test_ready_coaching_card_without_actionable_target_is_rejected(self) -> None: payload = self.pack.cases[0].coaching_cards[0].model_dump(mode="json") payload["targets"] = [] with self.assertRaises(ValidationError): CoachingCard.model_validate(payload) def test_coaching_card_rejects_duplicate_atomic_target(self) -> None: payload = self.pack.cases[0].coaching_cards[0].model_dump(mode="json") duplicate = dict(payload["targets"][0]) duplicate["prescription_id"] = "oas-g4-practice-reward-replay-copy" payload["targets"].append(duplicate) with self.assertRaisesRegex(ValidationError, "atomic and unique"): CoachingCard.model_validate(payload) def test_difficulty_ladder_requires_unseen_transfer_step(self) -> None: activity = self.pack.cases[1].coaching_cards[0].targets[1].activity payload = activity.model_dump(mode="json") for step in payload["steps"]: step["scenario_novelty"] = "familiar" with self.assertRaisesRegex(ValidationError, "unseen transfer step"): DifficultyLadderActivity.model_validate(payload) def test_model_observation_requires_model_run_provenance(self) -> None: with self.assertRaisesRegex(ValidationError, "requires model_run_id"): CriterionObservation( criterion_id="criterion.model-observed", status="observed", source_kind="model_inferred", perspective="independent_observer", evidence_refs=( { "ref_id": "model-evidence", "scene_id": "scene-model", "turn_index": 1, "actor": "observer", "kind": "evaluator_decision", }, ), uncertainty=0.2, ) value = CriterionObservation( criterion_id="criterion.model-observed", status="observed", source_kind="model_inferred", perspective="independent_observer", model_run_id=uuid4(), evidence_refs=( { "ref_id": "model-evidence", "scene_id": "scene-model", "turn_index": 1, "actor": "observer", "kind": "evaluator_decision", }, ), uncertainty=0.2, ) self.assertIsNotNone(value.model_run_id) def test_ready_attempt_requires_behavior_and_client_response_evidence(self) -> None: attempt = self.pack.cases[0].episodes[0].attempts[0] payload = attempt.model_dump(mode="json") payload["evidence_refs"] = [payload["evidence_refs"][0]] with self.assertRaisesRegex( ValidationError, "learner behavior and client response" ): PracticeAttemptObservation.model_validate(payload) def test_error_observation_is_never_imputed_as_success(self) -> None: with self.assertRaisesRegex(ValidationError, "maximum uncertainty"): CriterionObservation( criterion_id="criterion.error", status="error", source_kind="observed_runtime", perspective="runtime_observation", uncertainty=0.4, error_code="evaluator_timeout", ) def test_transfer_verified_state_requires_unseen_evidence(self) -> None: with self.assertRaisesRegex(ValidationError, "unseen transfer evidence"): CompetencyState( competency_id="competency.invalid.mastery", band="transfer_verified", forgetting_risk=0.2, uncertainty=0.2, attempt_count=4, familiar_demonstrations=4, unseen_transfer_demonstrations=0, highest_familiar_difficulty=5, ) def test_competency_graph_rejects_cycles(self) -> None: definitions = ( CompetencyDefinition( competency_id="competency.a", label_ko="A", description="순환 검증을 위한 첫 번째 합성 역량 정의다.", prerequisite_ids=("competency.b",), ), CompetencyDefinition( competency_id="competency.b", label_ko="B", description="순환 검증을 위한 두 번째 합성 역량 정의다.", prerequisite_ids=("competency.a",), ), ) states = tuple( CompetencyState( competency_id=item.competency_id, band="unassessed", forgetting_risk=0.5, uncertainty=1.0, attempt_count=0, familiar_demonstrations=0, unseen_transfer_demonstrations=0, highest_familiar_difficulty=0, ) for item in definitions ) with self.assertRaisesRegex(ValidationError, "acyclic"): CompetencyGraph(definitions=definitions, states=states) class DeliberatePracticePrescriptionAndEpisodeTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_practice_benchmark(BENCHMARK_PATH) def _prescriptions(self, case_index: int): return prescribe_from_coaching_cards(self.pack.cases[case_index].coaching_cards) def _assessment(self, case_index: int, episode_index: int = 0): case = self.pack.cases[case_index] prescriptions = { item.prescription_id: item for item in self._prescriptions(case_index) } episode = case.episodes[episode_index] return assess_practice_episode(prescriptions[episode.prescription_id], episode) def test_every_coaching_card_target_becomes_one_launchable_prescription( self, ) -> None: for case in self.pack.cases: prescriptions = prescribe_from_coaching_cards(case.coaching_cards) self.assertEqual( len(prescriptions), sum(len(card.targets) for card in case.coaching_cards), ) self.assertTrue(all(item.can_launch for item in prescriptions)) self.assertTrue(all(item.evidence_refs for item in prescriptions)) self.assertTrue(all(item.activity.launch_intent for item in prescriptions)) def test_self_claim_without_observed_effect_does_not_pass(self) -> None: assessment = self._assessment(0) self.assertEqual(assessment.progress, "practicing") self.assertFalse(assessment.mastery_allowed) self.assertEqual(assessment.attempts[0].outcome, "needs_retry") self.assertIn( "learner_success_claim_not_supported_by_attempt_evidence", assessment.counterevidence, ) def test_learner_report_cannot_independently_pass_target_criterion(self) -> None: case = self.pack.cases[3] prescription = self._prescriptions(3)[0] payload = case.episodes[0].model_dump(mode="json") payload["attempts"] = [payload["attempts"][0]] payload["attempts"][0]["criterion"].update( { "source_kind": "learner_reported", "perspective": "learner_self_report", "model_run_id": None, } ) episode = case.episodes[0].model_validate(payload) assessment = assess_practice_episode(prescription, episode) self.assertEqual(assessment.attempts[0].outcome, "insufficient_evidence") self.assertIn( "independent_observer_required", assessment.attempts[0].counterevidence, ) def test_unseen_transfer_can_use_prior_durable_familiar_demonstration(self) -> None: case = self.pack.cases[3] prescription = self._prescriptions(3)[0] payload = case.episodes[0].model_dump(mode="json") payload["episode_id"] = "oas-g4-episode-cross-session-transfer" payload["attempts"] = [payload["attempts"][-1]] payload["attempts"][0]["sequence_no"] = 1 payload["attempts"][0]["attempt_id"] = ( "oas-g4-attempt-cross-session-transfer" ) prior_state = case.graph.states[0].model_copy( update={ "band": "consistent_local", "attempt_count": 1, "familiar_demonstrations": 1, "highest_familiar_difficulty": 2, "evidence_refs": tuple( case.episodes[0].attempts[0].criterion.evidence_refs ), } ) assessment = assess_practice_episode( prescription, case.episodes[0].model_validate(payload), prior_state=prior_state, ) self.assertEqual(assessment.progress, "mastered") self.assertTrue(assessment.mastery_allowed) self.assertEqual(assessment.prior_familiar_demonstrations, 1) self.assertIn("transfer.verified", assessment.event_names) def test_before_after_comparison_preserves_both_evidence_sets(self) -> None: assessment = self._assessment(2) self.assertEqual(assessment.comparison.change, "improved") self.assertTrue(assessment.comparison.before_evidence_refs) self.assertTrue(assessment.comparison.after_evidence_refs) self.assertEqual( assessment.comparison.criterion_id, "criterion.acknowledge-impact-and-check", ) def test_memorized_phrase_blocks_transfer_mastery(self) -> None: assessment = self._assessment(2) self.assertEqual(assessment.progress, "transfer_pending") self.assertFalse(assessment.mastery_allowed) self.assertEqual(assessment.attempts[-1].outcome, "needs_retry") self.assertIn( "memorized_phrase_reused_in_transfer", assessment.mastery_blockers ) self.assertNotIn("transfer.verified", assessment.event_names) def test_genuine_unseen_voice_transfer_allows_mastery(self) -> None: assessment = self._assessment(3) self.assertEqual(assessment.progress, "mastered") self.assertTrue(assessment.mastery_allowed) self.assertEqual(assessment.attempts[-1].outcome, "passed") self.assertIn("transfer.verified", assessment.event_names) self.assertIn("practice.mastered", assessment.event_names) def test_voice_retry_cannot_pass_without_voice_feature_evidence(self) -> None: case = self.pack.cases[3] prescription = self._prescriptions(3)[0] episode_payload = case.episodes[0].model_dump(mode="json") attempt = episode_payload["attempts"][1] attempt["evidence_refs"] = [ item for item in attempt["evidence_refs"] if item["kind"] != "voice_feature" ] episode_payload["attempts"] = [attempt] episode_payload["attempts"][0]["sequence_no"] = 1 episode_payload["attempts"][0]["attempt_id"] = "oas-g4-attempt-voice-no-feature" result = assess_practice_episode( prescription, case.episodes[0].model_validate(episode_payload), ) self.assertEqual(result.attempts[0].outcome, "needs_retry") self.assertIn( "voice_retry_missing_voice_feature_evidence", result.attempts[0].counterevidence, ) def test_episode_assessment_rejects_compensating_total_score(self) -> None: payload = self._assessment(3).model_dump(mode="json") payload["total_score"] = 1.0 with self.assertRaises(ValidationError): PracticeEpisodeAssessment.model_validate(payload) def test_competency_update_records_attempt_evidence_but_only_verified_transfer_mastery( self, ) -> None: blocked = self._assessment(2) genuine = self._assessment(3) blocked_graph = apply_episode_to_competency_graph( self.pack.cases[2].graph, blocked ) genuine_graph = apply_episode_to_competency_graph( self.pack.cases[3].graph, genuine ) self.assertEqual(blocked_graph.states[0].band, "consistent_local") self.assertEqual(blocked_graph.states[0].unseen_transfer_demonstrations, 0) self.assertTrue(blocked_graph.states[0].evidence_refs) self.assertEqual(genuine_graph.states[0].band, "transfer_verified") self.assertEqual(genuine_graph.states[0].unseen_transfer_demonstrations, 1) def test_episode_payload_has_no_compensating_total_or_reward_field(self) -> None: keys = _all_keys(self._assessment(3).model_dump(mode="json")) self.assertNotIn("total", keys) self.assertNotIn("total_score", keys) self.assertNotIn("reward", keys) self.assertIn("uncertainty", keys) self.assertIn("evidence_refs", keys) self.assertIn("counterevidence", keys) class DeliberatePracticeCurriculumTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_practice_benchmark(BENCHMARK_PATH) def test_easy_familiar_repeat_is_blocked_in_favor_of_ladder(self) -> None: case = self.pack.cases[1] prescriptions = prescribe_from_coaching_cards(case.coaching_cards) decision = select_next_practice(case.graph, prescriptions) self.assertEqual( decision.selected_prescription_id, "oas-g4-practice-goal-ladder", ) self.assertEqual(decision.mode, "difficulty_ladder") self.assertTrue( any( "easy_repeat_blocked" in item for item in decision.blocked_prescription_reasons ) ) def test_same_weakness_band_uses_higher_forgetting_risk_deterministically( self, ) -> None: first = self.pack.cases[0] second = self.pack.cases[4] definitions = (first.graph.definitions[0], second.graph.definitions[0]) states = ( first.graph.states[0].model_copy( update={"band": "fragile", "attempt_count": 1, "forgetting_risk": 0.82} ), second.graph.states[0].model_copy( update={"band": "fragile", "attempt_count": 1, "forgetting_risk": 0.41} ), ) graph = CompetencyGraph(definitions=definitions, states=states) prescriptions = ( *prescribe_from_coaching_cards(first.coaching_cards), *prescribe_from_coaching_cards(second.coaching_cards), ) decision = select_next_practice(graph, prescriptions) self.assertEqual(decision.competency_id, "competency.empathy.reflection") self.assertEqual(decision.forgetting_risk, 0.82) self.assertEqual(decision.selection_basis[0], "weakest_available_band:fragile") class DeliberatePracticeBenchmarkTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.pack = load_practice_benchmark(BENCHMARK_PATH) cls.report = evaluate_practice_benchmark(cls.pack) def test_benchmark_meets_actionability_progress_and_selection_gates(self) -> None: self.assertEqual(self.report["case_count"], 5) self.assertEqual(self.report["actionable_prescription_coverage"], 1.0) self.assertEqual(self.report["episode_progress_accuracy"], 1.0) self.assertEqual(self.report["final_band_accuracy"], 1.0) self.assertEqual(self.report["curriculum_selection_accuracy"], 1.0) def test_benchmark_has_zero_reward_easy_repeat_or_phrase_hacking_regression( self, ) -> None: self.assertEqual(self.report["reward_hacking_regressions"], 0) self.assertEqual(self.report["easy_repeat_regressions"], 0) self.assertEqual(self.report["memorized_phrase_false_mastery"], 0) self.assertEqual(self.report["premature_mastery_count"], 0) def test_benchmark_report_preserves_uncertainty_evidence_and_counterevidence( self, ) -> None: rendered = render_practice_benchmark_report(self.report) phrase_row = next( item for item in self.report["rows"] if item["case_id"] == "oas-g4-bench-003" ) self.assertIn('"data_classification": "synthetic_educational"', rendered) self.assertIn('"clinical_claim_allowed": false', rendered) self.assertIsInstance(phrase_row["uncertainty"], float) self.assertTrue(phrase_row["evidence_refs"]) self.assertIn( "memorized_phrase_reused_in_transfer", phrase_row["counterevidence"] ) def test_benchmark_pack_rejects_missing_hacking_coverage(self) -> None: payload = self.pack.model_dump(mode="json") for case in payload["cases"]: case["tags"] = [tag for tag in case["tags"] if tag != "reward_hacking"] with self.assertRaisesRegex(ValidationError, "adversarial coverage"): PracticeBenchmarkPack.model_validate(payload) if __name__ == "__main__": unittest.main()