diff --git a/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py b/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py new file mode 100644 index 00000000..9aa6bd03 --- /dev/null +++ b/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py @@ -0,0 +1,276 @@ +"""Benchmark-derived generation flags for MESC Research Loop V1. + +MRL-0604 binds one exact training lineage, contamination assessment, and transformation +provenance record to an explicit benchmark-derivation classification. It records supplied +provenance evidence only; it does not generate data, inspect benchmarks, execute detectors, +or authorize training or model promotion. +""" + +from __future__ import annotations + +import enum +import re +import weakref +from collections.abc import Callable +from dataclasses import dataclass + +from medscale.mesc._mrl_contamination_interfaces_v1 import ( + ContaminationEvidenceReport, + ContaminationInterfaceError, +) +from medscale.mesc._mrl_content_identity_v1 import ( + canonical_semantic_bytes, + derive_content_sha256, +) +from medscale.mesc._mrl_training_example_lineage_v1 import ( + TrainingExampleLineageContract, + TrainingExampleLineageError, +) +from medscale.mesc._mrl_training_transformation_binding_v1 import ( + TrainingTransformationBinding, + TrainingTransformationBindingError, +) + +__all__ = [ + "BenchmarkDerivedGenerationClassification", + "BenchmarkDerivedGenerationError", + "BenchmarkDerivedGenerationFlags", + "build_benchmark_derived_generation_flags", +] + +_SHA256 = re.compile(r"^[0-9a-f]{64}$") + + +class BenchmarkDerivedGenerationError(ValueError): + """Fail-closed validation error for benchmark-derived generation metadata.""" + + +class BenchmarkDerivedGenerationClassification(enum.Enum): + """Closed benchmark-derivation classifications for one training example.""" + + NOT_BENCHMARK_DERIVED = "NOT_BENCHMARK_DERIVED" + BENCHMARK_DERIVED = "BENCHMARK_DERIVED" + INDETERMINATE = "INDETERMINATE" + + +def _make_flags_identity_registry() -> tuple[ + Callable[[BenchmarkDerivedGenerationFlags, str], None], + Callable[[BenchmarkDerivedGenerationFlags], str], +]: + identities: dict[int, str] = {} + + def remove(key: int) -> None: + identities.pop(key, None) + + def store(value: BenchmarkDerivedGenerationFlags, content_sha256: str) -> None: + key = id(value) + if key in identities: + raise BenchmarkDerivedGenerationError( + "benchmark flags construction identity already exists" + ) + identities[key] = content_sha256 + weakref.finalize(value, remove, key) + + def load(value: BenchmarkDerivedGenerationFlags) -> str: + identity = identities.get(id(value)) + if identity is None: + raise BenchmarkDerivedGenerationError( + "benchmark flags construction identity is missing" + ) + return identity + + return store, load + + +_store_flags_identity, _load_flags_identity = _make_flags_identity_registry() + + +@dataclass(frozen=True, slots=True, weakref_slot=True) +class BenchmarkDerivedGenerationFlags: + """Immutable benchmark-derivation metadata bound to exact contamination lineage.""" + + training_lineage_sha256: str + contamination_report_sha256: str + transformation_binding_sha256: str + assessment_artifact_sha256: str + classification: BenchmarkDerivedGenerationClassification + benchmark_artifact_sha256: str | None = None + + def __post_init__(self) -> None: + _require_sha256(self.training_lineage_sha256, "training_lineage_sha256") + _require_sha256(self.contamination_report_sha256, "contamination_report_sha256") + _require_sha256(self.transformation_binding_sha256, "transformation_binding_sha256") + _require_sha256(self.assessment_artifact_sha256, "assessment_artifact_sha256") + if type(self.classification) is not BenchmarkDerivedGenerationClassification: + raise BenchmarkDerivedGenerationError( + "classification must be an exact BenchmarkDerivedGenerationClassification" + ) + _require_optional_sha256(self.benchmark_artifact_sha256, "benchmark_artifact_sha256") + if self.classification is BenchmarkDerivedGenerationClassification.BENCHMARK_DERIVED: + if self.benchmark_artifact_sha256 is None: + raise BenchmarkDerivedGenerationError( + "benchmark-derived classification requires an exact benchmark artifact identity" + ) + elif ( + self.classification is BenchmarkDerivedGenerationClassification.NOT_BENCHMARK_DERIVED + and self.benchmark_artifact_sha256 is not None + ): + raise BenchmarkDerivedGenerationError( + "not-benchmark-derived classification cannot claim a benchmark source artifact" + ) + _store_flags_identity( + self, + derive_content_sha256(self._semantic_dict_validated()), + ) + + def _validated_snapshot(self) -> BenchmarkDerivedGenerationFlags: + if type(self) is not BenchmarkDerivedGenerationFlags: + raise BenchmarkDerivedGenerationError( + "flags must be an exact BenchmarkDerivedGenerationFlags" + ) + bound_content_sha256 = _load_flags_identity(self) + _require_sha256(bound_content_sha256, "bound benchmark flags content_sha256") + snapshot = BenchmarkDerivedGenerationFlags( + training_lineage_sha256=self.training_lineage_sha256, + contamination_report_sha256=self.contamination_report_sha256, + transformation_binding_sha256=self.transformation_binding_sha256, + assessment_artifact_sha256=self.assessment_artifact_sha256, + classification=self.classification, + benchmark_artifact_sha256=self.benchmark_artifact_sha256, + ) + current_content_sha256 = derive_content_sha256(snapshot._semantic_dict_validated()) + if current_content_sha256 != bound_content_sha256: + raise BenchmarkDerivedGenerationError( + "benchmark flags identity changed after construction" + ) + return snapshot + + def _benchmark_derived_flag_validated(self) -> bool | None: + if self.classification is BenchmarkDerivedGenerationClassification.BENCHMARK_DERIVED: + return True + if self.classification is BenchmarkDerivedGenerationClassification.NOT_BENCHMARK_DERIVED: + return False + return None + + @property + def benchmark_derived_flag(self) -> bool | None: + snapshot = BenchmarkDerivedGenerationFlags._validated_snapshot(self) + return snapshot._benchmark_derived_flag_validated() + + @property + def can_generate_examples(self) -> bool: + return False + + @property + def can_access_benchmark(self) -> bool: + return False + + @property + def can_authorize_training(self) -> bool: + return False + + @property + def can_authorize_model_promotion(self) -> bool: + return False + + def _semantic_dict_validated(self) -> dict[str, object]: + return { + "assessment_artifact_sha256": self.assessment_artifact_sha256, + "benchmark_artifact_sha256": self.benchmark_artifact_sha256, + "benchmark_derived_flag": self._benchmark_derived_flag_validated(), + "can_access_benchmark": False, + "can_authorize_model_promotion": False, + "can_authorize_training": False, + "can_generate_examples": False, + "classification": self.classification.value, + "contamination_report_sha256": self.contamination_report_sha256, + "format": "MRL-BENCHMARK-DERIVED-GENERATION-FLAGS-V1", + "training_lineage_sha256": self.training_lineage_sha256, + "transformation_binding_sha256": self.transformation_binding_sha256, + } + + def semantic_dict(self) -> dict[str, object]: + snapshot = BenchmarkDerivedGenerationFlags._validated_snapshot(self) + return snapshot._semantic_dict_validated() + + @property + def semantic_bytes(self) -> bytes: + return canonical_semantic_bytes(self.semantic_dict()) + + @property + def content_sha256(self) -> str: + return derive_content_sha256(self.semantic_dict()) + + def to_dict(self) -> dict[str, object]: + data = self.semantic_dict() + data["content_sha256"] = derive_content_sha256(data) + return data + + +def build_benchmark_derived_generation_flags( + lineage: TrainingExampleLineageContract, + contamination_report: ContaminationEvidenceReport, + transformation_binding: TrainingTransformationBinding, + *, + assessment_artifact_sha256: str, + classification: BenchmarkDerivedGenerationClassification, + benchmark_artifact_sha256: str | None = None, +) -> BenchmarkDerivedGenerationFlags: + """Bind supplied benchmark evidence to construction-bound MRL-0601/0602/0603 inputs.""" + if type(lineage) is not TrainingExampleLineageContract: + raise BenchmarkDerivedGenerationError( + "lineage must be an exact TrainingExampleLineageContract" + ) + if type(contamination_report) is not ContaminationEvidenceReport: + raise BenchmarkDerivedGenerationError( + "contamination_report must be an exact ContaminationEvidenceReport" + ) + if type(transformation_binding) is not TrainingTransformationBinding: + raise BenchmarkDerivedGenerationError( + "transformation_binding must be an exact TrainingTransformationBinding" + ) + + try: + example_snapshot, lineage_sha256 = lineage._validated_example() + contamination_snapshot = contamination_report._validated_snapshot() + transformation_snapshot = transformation_binding._validated_snapshot() + except ( + TrainingExampleLineageError, + ContaminationInterfaceError, + TrainingTransformationBindingError, + ) as exc: + raise BenchmarkDerivedGenerationError( + "benchmark derivation evidence failed canonical revalidation" + ) from exc + + if contamination_snapshot.training_lineage_sha256 != lineage_sha256: + raise BenchmarkDerivedGenerationError( + "contamination report does not bind the supplied training lineage" + ) + if transformation_snapshot.training_lineage_sha256 != lineage_sha256: + raise BenchmarkDerivedGenerationError( + "transformation binding does not bind the supplied training lineage" + ) + if transformation_snapshot.source_sha256 != example_snapshot.source_sha256: + raise BenchmarkDerivedGenerationError( + "transformation source identity does not match the canonical lineage source" + ) + + return BenchmarkDerivedGenerationFlags( + training_lineage_sha256=lineage_sha256, + contamination_report_sha256=contamination_snapshot.content_sha256, + transformation_binding_sha256=transformation_snapshot.content_sha256, + assessment_artifact_sha256=assessment_artifact_sha256, + classification=classification, + benchmark_artifact_sha256=benchmark_artifact_sha256, + ) + + +def _require_optional_sha256(value: object, label: str) -> None: + if value is not None: + _require_sha256(value, label) + + +def _require_sha256(value: object, label: str) -> None: + if type(value) is not str or _SHA256.fullmatch(value) is None: + raise BenchmarkDerivedGenerationError(f"{label} must be 64 lowercase hex") diff --git a/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py b/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py new file mode 100644 index 00000000..c2469fcd --- /dev/null +++ b/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py @@ -0,0 +1,292 @@ +"""R2-compatible sealed temporal-canary fixture workflow for MRL V1. + +MRL-0606 binds one exact MRL-0605 temporal-canary manifest to the canonical pure in-memory +fixture evaluator and records only content identities plus aggregate fixture metrics. The +executed fixture candidate must exactly match the sealed canary artifact identity; receipts +never expose item-level canary content and cannot place a canary into training or search. +""" + +from __future__ import annotations + +import re +import weakref +from collections.abc import Callable +from dataclasses import dataclass + +from medscale.mesc._mrl_content_identity_v1 import ( + canonical_semantic_bytes, + derive_content_sha256, +) +from medscale.mesc._mrl_fixture_research_surface_v1 import ( + FixtureCandidate, + FixtureEvaluation, + FixtureEvaluator, + FixtureParameterValue, + FixtureResearchSurface, + FixtureResearchSurfaceError, + build_fixture_candidate, + evaluate_fixture_candidate, +) +from medscale.mesc._mrl_temporal_canary_manifest_v1 import ( + TemporalCanaryManifest, + TemporalCanaryManifestError, + TemporalCanarySourceKind, +) + +__all__ = [ + "TemporalCanaryFixtureReceipt", + "TemporalCanaryFixtureWorkflowError", + "run_temporal_canary_fixture_workflow", +] + +_SHA256 = re.compile(r"^[0-9a-f]{64}$") + + +class TemporalCanaryFixtureWorkflowError(ValueError): + """Fail-closed validation error for the R2 temporal-canary fixture workflow.""" + + +def _make_receipt_identity_registry() -> tuple[ + Callable[[TemporalCanaryFixtureReceipt, str], None], + Callable[[TemporalCanaryFixtureReceipt], str], +]: + identities: dict[int, str] = {} + + def remove(key: int) -> None: + identities.pop(key, None) + + def store(value: TemporalCanaryFixtureReceipt, content_sha256: str) -> None: + key = id(value) + if key in identities: + raise TemporalCanaryFixtureWorkflowError( + "temporal-canary receipt construction identity already exists" + ) + identities[key] = content_sha256 + weakref.finalize(value, remove, key) + + def load(value: TemporalCanaryFixtureReceipt) -> str: + identity = identities.get(id(value)) + if identity is None: + raise TemporalCanaryFixtureWorkflowError( + "temporal-canary receipt construction identity is missing" + ) + return identity + + return store, load + + +_store_receipt_identity, _load_receipt_identity = _make_receipt_identity_registry() + + +@dataclass(frozen=True, slots=True, weakref_slot=True) +class TemporalCanaryFixtureReceipt: + """Immutable aggregate-only receipt for one sealed fixture canary evaluation.""" + + manifest_sha256: str + canary_artifact_sha256: str + source_kind: TemporalCanarySourceKind + surface_sha256: str + evaluator_sha256: str + candidate_sha256: str + evaluation_sha256: str + metric_id: str + observed_score: int + observed_max_score: int + + def __post_init__(self) -> None: + _require_sha256(self.manifest_sha256, "manifest_sha256") + _require_sha256(self.canary_artifact_sha256, "canary_artifact_sha256") + if type(self.source_kind) is not TemporalCanarySourceKind: + raise TemporalCanaryFixtureWorkflowError( + "source_kind must be an exact TemporalCanarySourceKind" + ) + _require_sha256(self.surface_sha256, "surface_sha256") + _require_sha256(self.evaluator_sha256, "evaluator_sha256") + _require_sha256(self.candidate_sha256, "candidate_sha256") + if self.canary_artifact_sha256 != self.candidate_sha256: + raise TemporalCanaryFixtureWorkflowError( + "candidate identity must equal the sealed canary artifact identity" + ) + _require_sha256(self.evaluation_sha256, "evaluation_sha256") + _require_text(self.metric_id, "metric_id") + if type(self.observed_score) is not int or type(self.observed_max_score) is not int: + raise TemporalCanaryFixtureWorkflowError("observed scores must be exact integers") + if ( + self.observed_max_score <= 0 + or self.observed_score < 0 + or self.observed_score > self.observed_max_score + ): + raise TemporalCanaryFixtureWorkflowError( + "observed scores must satisfy 0 <= score <= max_score" + ) + _store_receipt_identity( + self, + derive_content_sha256(self._semantic_dict_validated()), + ) + + def _validated_snapshot(self) -> TemporalCanaryFixtureReceipt: + if type(self) is not TemporalCanaryFixtureReceipt: + raise TemporalCanaryFixtureWorkflowError( + "receipt must be an exact TemporalCanaryFixtureReceipt" + ) + bound_content_sha256 = _load_receipt_identity(self) + _require_sha256(bound_content_sha256, "bound receipt content_sha256") + snapshot = TemporalCanaryFixtureReceipt( + manifest_sha256=self.manifest_sha256, + canary_artifact_sha256=self.canary_artifact_sha256, + source_kind=self.source_kind, + surface_sha256=self.surface_sha256, + evaluator_sha256=self.evaluator_sha256, + candidate_sha256=self.candidate_sha256, + evaluation_sha256=self.evaluation_sha256, + metric_id=self.metric_id, + observed_score=self.observed_score, + observed_max_score=self.observed_max_score, + ) + current_content_sha256 = derive_content_sha256(snapshot._semantic_dict_validated()) + if current_content_sha256 != bound_content_sha256: + raise TemporalCanaryFixtureWorkflowError( + "temporal-canary receipt identity changed after construction" + ) + return snapshot + + @property + def sealed(self) -> bool: + return True + + @property + def fixture_only(self) -> bool: + return True + + @property + def exposes_canary_content(self) -> bool: + return False + + @property + def can_enter_training(self) -> bool: + return False + + @property + def can_enter_search(self) -> bool: + return False + + @property + def can_authorize(self) -> bool: + return False + + def _semantic_dict_validated(self) -> dict[str, object]: + return { + "can_authorize": False, + "can_enter_search": False, + "can_enter_training": False, + "canary_artifact_sha256": self.canary_artifact_sha256, + "candidate_sha256": self.candidate_sha256, + "evaluation_sha256": self.evaluation_sha256, + "evaluator_sha256": self.evaluator_sha256, + "exposes_canary_content": False, + "fixture_only": True, + "format": "MRL-TEMPORAL-CANARY-FIXTURE-RECEIPT-V1", + "manifest_sha256": self.manifest_sha256, + "metric_id": self.metric_id, + "observed_max_score": self.observed_max_score, + "observed_score": self.observed_score, + "sealed": True, + "source_kind": self.source_kind.value, + "surface_sha256": self.surface_sha256, + "workflow_mode": "R2_FIXTURE_ONLY", + } + + def semantic_dict(self) -> dict[str, object]: + snapshot = TemporalCanaryFixtureReceipt._validated_snapshot(self) + return snapshot._semantic_dict_validated() + + @property + def semantic_bytes(self) -> bytes: + return canonical_semantic_bytes(self.semantic_dict()) + + @property + def content_sha256(self) -> str: + return derive_content_sha256(self.semantic_dict()) + + def to_dict(self) -> dict[str, object]: + data = self.semantic_dict() + data["content_sha256"] = derive_content_sha256(data) + return data + + +def run_temporal_canary_fixture_workflow( + manifest: TemporalCanaryManifest, + surface: FixtureResearchSurface, + evaluator: FixtureEvaluator, + parameter_values: tuple[FixtureParameterValue, ...], +) -> TemporalCanaryFixtureReceipt: + """Evaluate one exact sealed-canary fixture candidate from coherent fixture snapshots.""" + if type(manifest) is not TemporalCanaryManifest: + raise TemporalCanaryFixtureWorkflowError("manifest must be an exact TemporalCanaryManifest") + if type(surface) is not FixtureResearchSurface: + raise TemporalCanaryFixtureWorkflowError("surface must be an exact FixtureResearchSurface") + if type(evaluator) is not FixtureEvaluator: + raise TemporalCanaryFixtureWorkflowError("evaluator must be an exact FixtureEvaluator") + if type(parameter_values) is not tuple: + raise TemporalCanaryFixtureWorkflowError("parameter_values must be an exact tuple") + if any(type(value) is not FixtureParameterValue for value in parameter_values): + raise TemporalCanaryFixtureWorkflowError("parameter_values contains an invalid item type") + + try: + manifest_snapshot = manifest._validated_snapshot() + surface_snapshot = FixtureResearchSurface._validated_snapshot(surface) + evaluator_snapshot = FixtureEvaluator._validated_snapshot(evaluator) + parameter_snapshots = tuple( + FixtureParameterValue(parameter_id=value.parameter_id, value=value.value) + for value in parameter_values + ) + evaluator_sha256 = evaluator_snapshot.content_sha256 + if manifest_snapshot.evaluator_artifact_sha256 != evaluator_sha256: + raise TemporalCanaryFixtureWorkflowError( + "temporal-canary manifest evaluator identity does not match " + "supplied fixture evaluator" + ) + candidate: FixtureCandidate = build_fixture_candidate( + surface_snapshot, + parameter_snapshots, + ) + candidate_sha256 = candidate.content_sha256 + if manifest_snapshot.canary_artifact_sha256 != candidate_sha256: + raise TemporalCanaryFixtureWorkflowError( + "temporal-canary manifest artifact identity does not match " + "supplied fixture candidate" + ) + evaluation: FixtureEvaluation = evaluate_fixture_candidate( + surface_snapshot, + evaluator_snapshot, + candidate, + ) + except (TemporalCanaryManifestError, FixtureResearchSurfaceError) as exc: + raise TemporalCanaryFixtureWorkflowError( + "temporal-canary fixture workflow failed canonical validation" + ) from exc + + return TemporalCanaryFixtureReceipt( + manifest_sha256=manifest_snapshot.content_sha256, + canary_artifact_sha256=manifest_snapshot.canary_artifact_sha256, + source_kind=manifest_snapshot.source_kind, + surface_sha256=surface_snapshot.content_sha256, + evaluator_sha256=evaluator_sha256, + candidate_sha256=candidate_sha256, + evaluation_sha256=evaluation.content_sha256, + metric_id=evaluation.metric_id, + observed_score=evaluation.score, + observed_max_score=evaluation.max_score, + ) + + +def _require_text(value: object, label: str) -> None: + if type(value) is not str or not value or value.strip() != value: + raise TemporalCanaryFixtureWorkflowError(f"{label} must be canonical non-empty text") + if any(character.isspace() for character in value): + raise TemporalCanaryFixtureWorkflowError(f"{label} cannot contain whitespace") + + +def _require_sha256(value: object, label: str) -> None: + if type(value) is not str or _SHA256.fullmatch(value) is None: + raise TemporalCanaryFixtureWorkflowError(f"{label} must be 64 lowercase hex") diff --git a/tests/test_mesc_mrl_benchmark_derived_generation_v1.py b/tests/test_mesc_mrl_benchmark_derived_generation_v1.py new file mode 100644 index 00000000..701b409e --- /dev/null +++ b/tests/test_mesc_mrl_benchmark_derived_generation_v1.py @@ -0,0 +1,282 @@ +"""MRL-0604 tests for benchmark-derived generation flags.""" + +from __future__ import annotations + +import pytest + +from medscale.mesc._mrl_benchmark_derived_generation_v1 import ( + BenchmarkDerivedGenerationClassification, + BenchmarkDerivedGenerationError, + BenchmarkDerivedGenerationFlags, + build_benchmark_derived_generation_flags, +) +from medscale.mesc._mrl_contamination_interfaces_v1 import ( + ContaminationEvidenceReport, + build_contamination_evidence_report, +) +from medscale.mesc._mrl_training_example_lineage_v1 import ( + TrainingExampleLineageContract, + build_training_example_lineage, +) +from medscale.mesc._mrl_training_transformation_binding_v1 import ( + TrainingTransformationBinding, + build_training_transformation_binding, +) +from test_mesc_mrl_contamination_interfaces_v1 import _checks +from test_mesc_mrl_training_example_lineage_v1 import _example + + +def _bound_inputs() -> tuple[ + TrainingExampleLineageContract, + ContaminationEvidenceReport, + TrainingTransformationBinding, +]: + lineage = build_training_example_lineage(_example()) + contamination = build_contamination_evidence_report(lineage, _checks()) + transformation = build_training_transformation_binding( + lineage, + transformation_kind="normalization", + transformation_artifact_sha256="7" * 64, + ) + return lineage, contamination, transformation + + +def test_not_derived_flags_are_deterministic_and_exactly_bound() -> None: + lineage, contamination, transformation = _bound_inputs() + + first = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.NOT_BENCHMARK_DERIVED, + ) + second = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.NOT_BENCHMARK_DERIVED, + ) + + assert first.semantic_bytes == second.semantic_bytes + assert first.content_sha256 == second.content_sha256 + assert first.training_lineage_sha256 == lineage.content_sha256 + assert first.contamination_report_sha256 == contamination.content_sha256 + assert first.transformation_binding_sha256 == transformation.content_sha256 + assert first.benchmark_derived_flag is False + + +def test_benchmark_derived_classification_requires_benchmark_artifact() -> None: + with pytest.raises(BenchmarkDerivedGenerationError, match="requires"): + BenchmarkDerivedGenerationFlags( + training_lineage_sha256="a" * 64, + contamination_report_sha256="b" * 64, + transformation_binding_sha256="c" * 64, + assessment_artifact_sha256="d" * 64, + classification=BenchmarkDerivedGenerationClassification.BENCHMARK_DERIVED, + ) + + +def test_benchmark_derived_flag_binds_exact_benchmark_identity() -> None: + lineage, contamination, transformation = _bound_inputs() + flags = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.BENCHMARK_DERIVED, + benchmark_artifact_sha256="9" * 64, + ) + + assert flags.benchmark_derived_flag is True + assert flags.benchmark_artifact_sha256 == "9" * 64 + + +def test_not_derived_classification_cannot_claim_benchmark_source() -> None: + with pytest.raises(BenchmarkDerivedGenerationError, match="cannot claim"): + BenchmarkDerivedGenerationFlags( + training_lineage_sha256="a" * 64, + contamination_report_sha256="b" * 64, + transformation_binding_sha256="c" * 64, + assessment_artifact_sha256="d" * 64, + classification=BenchmarkDerivedGenerationClassification.NOT_BENCHMARK_DERIVED, + benchmark_artifact_sha256="e" * 64, + ) + + +def test_contamination_report_from_another_lineage_fails_closed() -> None: + lineage, _, transformation = _bound_inputs() + other_lineage = build_training_example_lineage(_example(source_sha256="c" * 64)) + other_contamination = build_contamination_evidence_report(other_lineage, _checks()) + + with pytest.raises( + BenchmarkDerivedGenerationError, + match="contamination report does not bind", + ): + build_benchmark_derived_generation_flags( + lineage, + other_contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_transformation_from_another_lineage_fails_closed() -> None: + lineage, contamination, _ = _bound_inputs() + other_lineage = build_training_example_lineage(_example(source_sha256="c" * 64)) + other_transformation = build_training_transformation_binding( + other_lineage, + transformation_kind="normalization", + transformation_artifact_sha256="7" * 64, + ) + + with pytest.raises( + BenchmarkDerivedGenerationError, + match="transformation binding does not bind", + ): + build_benchmark_derived_generation_flags( + lineage, + contamination, + other_transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_mutated_contamination_evidence_fails_closed() -> None: + lineage, contamination, transformation = _bound_inputs() + object.__setattr__(contamination.checks[0], "detector_artifact_sha256", "invalid") + + with pytest.raises(BenchmarkDerivedGenerationError, match="canonical revalidation"): + build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_mutated_contamination_container_fails_closed() -> None: + lineage, contamination, transformation = _bound_inputs() + object.__setattr__(contamination, "checks", list(contamination.checks)) + + with pytest.raises(BenchmarkDerivedGenerationError, match="canonical revalidation"): + build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_valid_lineage_identity_mutation_fails_closed() -> None: + lineage, contamination, transformation = _bound_inputs() + object.__setattr__(lineage.example, "source_sha256", "f" * 64) + + with pytest.raises(BenchmarkDerivedGenerationError, match="canonical revalidation"): + build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_valid_contamination_identity_mutation_fails_closed() -> None: + lineage, contamination, transformation = _bound_inputs() + object.__setattr__(contamination.checks[0], "evidence_artifact_sha256", "f" * 64) + + with pytest.raises(BenchmarkDerivedGenerationError, match="canonical revalidation"): + build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_valid_transformation_identity_mutation_fails_closed() -> None: + lineage, contamination, transformation = _bound_inputs() + object.__setattr__(transformation, "transformation_artifact_sha256", "f" * 64) + + with pytest.raises(BenchmarkDerivedGenerationError, match="canonical revalidation"): + build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + +def test_mutated_flags_fail_closed_on_public_views() -> None: + lineage, contamination, transformation = _bound_inputs() + flags = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + object.__setattr__(flags, "assessment_artifact_sha256", "invalid") + + with pytest.raises(BenchmarkDerivedGenerationError, match="64 lowercase hex"): + flags.semantic_dict() + with pytest.raises(BenchmarkDerivedGenerationError, match="64 lowercase hex"): + _ = flags.content_sha256 + + +def test_valid_flags_identity_mutation_fails_closed() -> None: + lineage, contamination, transformation = _bound_inputs() + flags = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + object.__setattr__(flags, "assessment_artifact_sha256", "9" * 64) + + with pytest.raises(BenchmarkDerivedGenerationError, match="identity changed"): + flags.semantic_dict() + with pytest.raises(BenchmarkDerivedGenerationError, match="identity changed"): + _ = flags.content_sha256 + + +def test_mutated_classification_fails_closed_on_derived_flag() -> None: + lineage, contamination, transformation = _bound_inputs() + flags = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + object.__setattr__(flags, "classification", "INDETERMINATE") + + with pytest.raises(BenchmarkDerivedGenerationError, match="exact"): + _ = flags.benchmark_derived_flag + + +def test_flags_cannot_generate_access_or_authorize_training() -> None: + lineage, contamination, transformation = _bound_inputs() + flags = build_benchmark_derived_generation_flags( + lineage, + contamination, + transformation, + assessment_artifact_sha256="8" * 64, + classification=BenchmarkDerivedGenerationClassification.INDETERMINATE, + ) + + assert flags.benchmark_derived_flag is None + assert flags.can_generate_examples is False + assert flags.can_access_benchmark is False + assert flags.can_authorize_training is False + assert flags.can_authorize_model_promotion is False + assert b"PROMOTED" not in flags.semantic_bytes diff --git a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py new file mode 100644 index 00000000..ec426560 --- /dev/null +++ b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py @@ -0,0 +1,313 @@ +"""MRL-0606 tests for the sealed temporal-canary fixture workflow.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import replace + +import pytest + +import medscale.mesc._mrl_temporal_canary_fixture_workflow_v1 as canary_module +from medscale.mesc._mrl_fixture_research_surface_v1 import ( + FixtureCandidate, + FixtureEvaluator, + FixtureParameterValue, + FixtureResearchSurface, + build_fixture_candidate, +) +from medscale.mesc._mrl_temporal_canary_fixture_workflow_v1 import ( + TemporalCanaryFixtureWorkflowError, + run_temporal_canary_fixture_workflow, +) +from medscale.mesc._mrl_temporal_canary_manifest_v1 import ( + TemporalCanaryManifest, + TemporalCanarySourceKind, +) +from test_mesc_mrl_fixture_research_surface_v1 import _evaluator, _surface, _values +from test_mesc_mrl_temporal_canary_manifest_v1 import _manifest + + +def _bound_manifest( + evaluator: FixtureEvaluator, + surface: FixtureResearchSurface, + parameter_values: tuple[FixtureParameterValue, ...], + *, + source_kind: TemporalCanarySourceKind = TemporalCanarySourceKind.SYNTHETIC, +) -> TemporalCanaryManifest: + candidate = build_fixture_candidate(surface, parameter_values) + return replace( + _manifest( + source_kind=source_kind, + artifact=candidate.content_sha256, + ), + evaluator_artifact_sha256=evaluator.content_sha256, + ) + + +def test_fixture_canary_workflow_is_deterministic_and_identity_bound() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + + first = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + second = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + assert first.semantic_bytes == second.semantic_bytes + assert first.content_sha256 == second.content_sha256 + assert first.manifest_sha256 == manifest.content_sha256 + assert first.canary_artifact_sha256 == manifest.canary_artifact_sha256 + assert first.canary_artifact_sha256 == first.candidate_sha256 + assert first.surface_sha256 == surface.content_sha256 + assert first.evaluator_sha256 == evaluator.content_sha256 + assert first.evaluator_sha256 == manifest.evaluator_artifact_sha256 + assert first.observed_score == 1 + assert first.observed_max_score == 2 + + +def test_workflow_uses_fixture_snapshots_if_live_evaluator_drifts_mid_call( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + original_evaluator_sha256 = evaluator.content_sha256 + original_metric_id = evaluator.metric_id + original_build_candidate: Callable[ + [FixtureResearchSurface, tuple[FixtureParameterValue, ...]], FixtureCandidate + ] = build_fixture_candidate + mutation_performed = False + + def mutate_live_evaluator_then_build_snapshot( + surface_snapshot: FixtureResearchSurface, + values: tuple[FixtureParameterValue, ...], + ) -> FixtureCandidate: + nonlocal mutation_performed + if not mutation_performed: + mutation_performed = True + object.__setattr__(evaluator, "metric_id", "fixture-metric-mutated") + return original_build_candidate(surface_snapshot, values) + + monkeypatch.setattr( + canary_module, + "build_fixture_candidate", + mutate_live_evaluator_then_build_snapshot, + ) + + receipt = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + assert mutation_performed is True + assert receipt.evaluator_sha256 == original_evaluator_sha256 + assert receipt.metric_id == original_metric_id + assert evaluator.content_sha256 != original_evaluator_sha256 + + +def test_hand_authored_fixture_canary_remains_r2_fixture_only() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest( + evaluator, + surface, + parameter_values, + source_kind=TemporalCanarySourceKind.HAND_AUTHORED_FIXTURE, + ) + receipt = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + assert receipt.source_kind is TemporalCanarySourceKind.HAND_AUTHORED_FIXTURE + assert receipt.fixture_only is True + assert receipt.sealed is True + + +def test_receipt_never_exposes_or_recycles_canary_content() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + receipt = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + assert receipt.exposes_canary_content is False + assert receipt.can_enter_training is False + assert receipt.can_enter_search is False + assert receipt.can_authorize is False + assert receipt.semantic_dict()["workflow_mode"] == "R2_FIXTURE_ONLY" + + +def test_manifest_canary_artifact_must_match_executed_fixture_candidate() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + manifest = _bound_manifest(evaluator, surface, _values()) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="artifact identity"): + run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + _values(beta=10), + ) + + +def test_mutated_manifest_fails_closed() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + object.__setattr__(manifest, "canary_artifact_sha256", "invalid") + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="canonical validation"): + run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + +def test_valid_manifest_identity_mutation_fails_closed() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + object.__setattr__(manifest, "evaluator_artifact_sha256", "f" * 64) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="canonical validation"): + run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + +def test_manifest_evaluator_identity_mismatch_fails_closed() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = replace( + _bound_manifest(evaluator, surface, parameter_values), + evaluator_artifact_sha256="f" * 64, + ) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="manifest evaluator identity"): + run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + + +def test_evaluator_surface_mismatch_fails_closed() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + other_evaluator = _evaluator(targets=(parameter_values[0],)) + manifest = _bound_manifest( + other_evaluator, + surface, + parameter_values, + ) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="canonical validation"): + run_temporal_canary_fixture_workflow( + manifest, + surface, + other_evaluator, + parameter_values, + ) + + +def test_mutated_receipt_candidate_identity_fails_closed_on_public_views() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + receipt = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + object.__setattr__(receipt, "candidate_sha256", "f" * 64) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="candidate identity"): + receipt.semantic_dict() + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="candidate identity"): + _ = receipt.content_sha256 + + +def test_valid_receipt_identity_mutation_fails_closed() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + receipt = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + object.__setattr__(receipt, "evaluation_sha256", "f" * 64) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="identity changed"): + receipt.semantic_dict() + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="identity changed"): + _ = receipt.content_sha256 + + +def test_mutated_receipt_fails_closed_on_public_views() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + parameter_values = _values() + manifest = _bound_manifest(evaluator, surface, parameter_values) + receipt = run_temporal_canary_fixture_workflow( + manifest, + surface, + evaluator, + parameter_values, + ) + object.__setattr__(receipt, "evaluation_sha256", "invalid") + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="64 lowercase hex"): + receipt.semantic_dict() + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="64 lowercase hex"): + _ = receipt.content_sha256 + + +def test_wrong_manifest_type_fails_closed() -> None: + evaluator = _evaluator() + surface = _surface(evaluator) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="exact TemporalCanaryManifest"): + run_temporal_canary_fixture_workflow( + object(), # type: ignore[arg-type] + surface, + evaluator, + _values(), + )