from __future__ import annotations import unittest from pydantic import ValidationError from .contracts.continuous_improvement import ( AgenticReleaseManifest, ContentBenchmarkQualification, ContentSourceArtifact, GeneratedContentDraft, IndependentRedTeamReview, ModelCalibrationSnapshot, OperationalIncident, RedTeamFinding, ) from .services.continuous_improvement import ( build_incident_regression_dag, decide_model_change, promote_content_to_catalog, release_allowed, ) def _source(status: str = "approved") -> ContentSourceArtifact: return ContentSourceArtifact( source_id="oas-g8-source-training-guide", version="1.0.0", content_sha256="a" * 64, provenance_uri="repo://data/source-pack/training-guide", usage_status=status, citation_label="합성 상담 훈련 source pack 1.0", ) def _draft(**overrides: object) -> GeneratedContentDraft: payload: dict[str, object] = { "draft_id": "oas-g8-draft-repair-case", "content_kind": "case", "source_refs": ("oas-g8-source-training-guide",), "generation_model": "content-agent-v1", "prompt_version": "1.0.0", "prompt_sha256": "b" * 64, "payload_sha256": "c" * 64, "synthetic_identity_id": "synthetic-identity-repair-case", "difficulty_level": 4, "hidden_answer_fingerprint": "d" * 64, "visible_answer_overlap_tokens": 0, "pii_findings": 0, "unsupported_clinical_claims": 0, } payload.update(overrides) return GeneratedContentDraft.model_validate(payload) def _reviews( *, finding: RedTeamFinding | None = None ) -> tuple[IndependentRedTeamReview, IndependentRedTeamReview]: dimensions_a = ("safety", "identity", "pii", "grounding") dimensions_b = ("answer_leakage", "cultural_bias", "difficulty") findings_a = (finding,) if finding and finding.dimension in dimensions_a else () findings_b = (finding,) if finding and finding.dimension in dimensions_b else () return ( IndependentRedTeamReview( review_id="oas-g8-review-safety-agent", draft_id="oas-g8-draft-repair-case", reviewer_agent_id="safety-red-team-agent", dimensions=dimensions_a, findings=findings_a, reviewed_payload_sha256="c" * 64, ), IndependentRedTeamReview( review_id="oas-g8-review-bias-agent", draft_id="oas-g8-draft-repair-case", reviewer_agent_id="bias-red-team-agent", dimensions=dimensions_b, findings=findings_b, reviewed_payload_sha256="c" * 64, ), ) def _benchmark(**overrides: object) -> ContentBenchmarkQualification: payload: dict[str, object] = { "benchmark_id": "oas-g8-benchmark-repair-case", "draft_id": "oas-g8-draft-repair-case", "variant_count": 8, "variant_pass_rate": 0.875, "answer_leakage_count": 0, "pii_finding_count": 0, "unsupported_claim_count": 0, "safety_failure_count": 0, "reward_hacking_count": 0, "evidence_refs": ("benchmark-report:repair-case",), } payload.update(overrides) return ContentBenchmarkQualification.model_validate(payload) def _snapshot(snapshot_id: str, **overrides: object) -> ModelCalibrationSnapshot: payload: dict[str, object] = { "snapshot_id": snapshot_id, "model": snapshot_id, "prompt_version": "1.0.0", "benchmark_version": "1.0.0", "task_accuracy": 0.9, "critical_miss_count": 0, "leakage_count": 0, "pii_count": 0, "calibration_error": 0.12, "subgroup_max_gap": 0.1, } payload.update(overrides) return ModelCalibrationSnapshot.model_validate(payload) class ContentPromotionTests(unittest.TestCase): def test_source_to_redteam_benchmark_catalog_closed_loop(self) -> None: entry = promote_content_to_catalog( draft=_draft(), sources=(_source(),), reviews=_reviews(), benchmark=_benchmark(), ) self.assertEqual(entry.status, "approved") self.assertFalse(entry.clinical_claim_allowed) self.assertEqual(len(entry.review_ids), 2) self.assertEqual(entry.payload_sha256, "c" * 64) def test_restricted_source_blocks_promotion(self) -> None: with self.assertRaisesRegex(ValueError, "approved source usage"): promote_content_to_catalog( draft=_draft(), sources=(_source("restricted"),), reviews=_reviews(), benchmark=_benchmark(), ) def test_answer_leakage_pii_and_unsupported_claims_each_block(self) -> None: for field in ( "visible_answer_overlap_tokens", "pii_findings", "unsupported_clinical_claims", ): with self.subTest(field=field): with self.assertRaises(ValueError): promote_content_to_catalog( draft=_draft(**{field: 1}), sources=(_source(),), reviews=_reviews(), benchmark=_benchmark(), ) def test_unresolved_redteam_finding_blocks_promotion(self) -> None: finding = RedTeamFinding( finding_id="oas-g8-finding-answer-leak", dimension="answer_leakage", severity="blocker", state="open", evidence_ref="visible-answer-overlap:4", ) with self.assertRaisesRegex(ValueError, "unresolved red-team"): promote_content_to_catalog( draft=_draft(), sources=(_source(),), reviews=_reviews(finding=finding), benchmark=_benchmark(), ) def test_resolved_finding_requires_remediation_evidence(self) -> None: with self.assertRaisesRegex(ValidationError, "remediation evidence"): RedTeamFinding( finding_id="oas-g8-finding-resolved-without-proof", dimension="grounding", severity="moderate", state="resolved", evidence_ref="claim:unsupported", ) def test_same_reviewer_cannot_satisfy_independence(self) -> None: reviews = list(_reviews()) payload = reviews[1].model_dump(mode="json") payload["reviewer_agent_id"] = reviews[0].reviewer_agent_id reviews[1] = IndependentRedTeamReview.model_validate(payload) with self.assertRaisesRegex(ValueError, "reviewers must be independent"): promote_content_to_catalog( draft=_draft(), sources=(_source(),), reviews=reviews, benchmark=_benchmark(), ) def test_unqualified_benchmark_blocks_promotion(self) -> None: benchmark = _benchmark(reward_hacking_count=1) self.assertFalse(benchmark.qualified) with self.assertRaisesRegex(ValueError, "qualified benchmark"): promote_content_to_catalog( draft=_draft(), sources=(_source(),), reviews=_reviews(), benchmark=benchmark, ) class ModelChangeGateTests(unittest.TestCase): def test_clean_candidate_is_promoted(self) -> None: result = decide_model_change( baseline=_snapshot("oas-g8-model-snapshot-baseline"), candidate=_snapshot( "oas-g8-model-snapshot-candidate", task_accuracy=0.92, calibration_error=0.10, subgroup_max_gap=0.08, ), ) self.assertEqual(result.decision, "promote") self.assertIsNone(result.rollback_target_snapshot_id) def test_critical_miss_or_leakage_rolls_back(self) -> None: baseline = _snapshot("oas-g8-model-snapshot-baseline") for field in ("critical_miss_count", "leakage_count", "pii_count"): with self.subTest(field=field): candidate = _snapshot("oas-g8-model-snapshot-candidate", **{field: 1}) result = decide_model_change( baseline=baseline, candidate=candidate, ) self.assertEqual(result.decision, "rollback") self.assertEqual( result.rollback_target_snapshot_id, baseline.snapshot_id ) def test_noncritical_calibration_regression_is_quarantined(self) -> None: result = decide_model_change( baseline=_snapshot("oas-g8-model-snapshot-baseline"), candidate=_snapshot( "oas-g8-model-snapshot-candidate", calibration_error=0.2 ), ) self.assertEqual(result.decision, "quarantine") self.assertIn("calibration_error_regression", result.reasons) class OperationalClosedLoopTests(unittest.TestCase): def test_incident_becomes_reproduction_to_runtime_dag(self) -> None: incident = OperationalIncident( incident_id="oas-g8-incident-voice-timeout", error_fingerprint="e" * 64, affected_contract="voice.turn.completion", evidence_refs=("audit://incident/voice-timeout",), ) dag = build_incident_regression_dag(incident) self.assertEqual( tuple(item.node_type for item in dag.nodes), ("reproduction_test", "implementation", "e2e", "runtime_proof"), ) self.assertEqual(dag.nodes[0].depends_on, ()) self.assertEqual(dag.nodes[-1].depends_on, (dag.nodes[-2].node_id,)) self.assertFalse(incident.pii_included) def test_release_requires_every_evidence_gate(self) -> None: manifest = AgenticReleaseManifest( release_id="oas-g8-release-candidate", red_green_passed=True, contract_passed=True, e2e_passed=True, runtime_proof_passed=True, public_proof_passed=True, ssot_synced=True, evidence_refs=("release://candidate/manifest",), ) self.assertTrue(release_allowed(manifest)) payload = manifest.model_dump(mode="json") payload["public_proof_passed"] = False self.assertFalse( release_allowed(AgenticReleaseManifest.model_validate(payload)) ) if __name__ == "__main__": unittest.main()