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 산출물은 커밋에서 제외했다.
469 lines
19 KiB
Python
469 lines
19 KiB
Python
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()
|