From ea22ff69fec7a5d6d91f9f3cbd8640a1ef46c1fd Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:26:14 +0300 Subject: [PATCH 1/7] feat(mrl): port benchmark flags and canary fixture workflow --- .../_mrl_benchmark_derived_generation_v1.py | 318 ++++++++++++++++++ ...mrl_temporal_canary_fixture_workflow_v1.py | 302 +++++++++++++++++ ...esc_mrl_benchmark_derived_generation_v1.py | 240 +++++++++++++ ...mrl_temporal_canary_fixture_workflow_v1.py | 297 ++++++++++++++++ 4 files changed, 1157 insertions(+) create mode 100644 src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py create mode 100644 src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py create mode 100644 tests/test_mesc_mrl_benchmark_derived_generation_v1.py create mode 100644 tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py 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..0bee8650 --- /dev/null +++ b/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py @@ -0,0 +1,318 @@ +"""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 ( + ContaminationCheckEvidence, + ContaminationEvidenceReport, +) +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, + build_training_example_lineage, +) +from medscale.mesc._mrl_training_transformation_binding_v1 import TrainingTransformationBinding + +__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-derivation evidence to exact MRL-0601/0602/0603 identities.""" + 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: + lineage_snapshot = build_training_example_lineage(lineage.example) + except TrainingExampleLineageError as exc: + raise BenchmarkDerivedGenerationError( + "training lineage failed canonical revalidation" + ) from exc + if lineage_snapshot.content_sha256 != lineage.content_sha256: + raise BenchmarkDerivedGenerationError( + "training lineage identity changed after construction" + ) + + contamination_snapshot = _snapshot_contamination_report(contamination_report) + transformation_snapshot = _snapshot_transformation_binding(transformation_binding) + lineage_sha256 = lineage_snapshot.content_sha256 + 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 != lineage_snapshot.example.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 _snapshot_contamination_report( + report: ContaminationEvidenceReport, +) -> ContaminationEvidenceReport: + if type(report.checks) is not tuple: + raise BenchmarkDerivedGenerationError( + "contamination report checks must remain an exact tuple" + ) + if any(type(item) is not ContaminationCheckEvidence for item in report.checks): + raise BenchmarkDerivedGenerationError( + "contamination report checks contains an invalid item type" + ) + checks = tuple( + ContaminationCheckEvidence( + kind=item.kind, + detector_id=item.detector_id, + detector_artifact_sha256=item.detector_artifact_sha256, + evidence_artifact_sha256=item.evidence_artifact_sha256, + disposition=item.disposition, + similarity_decimal=item.similarity_decimal, + threshold_decimal=item.threshold_decimal, + ) + for item in report.checks + ) + return ContaminationEvidenceReport( + training_lineage_sha256=report.training_lineage_sha256, + checks=checks, + ) + + +def _snapshot_transformation_binding( + binding: TrainingTransformationBinding, +) -> TrainingTransformationBinding: + return TrainingTransformationBinding( + training_lineage_sha256=binding.training_lineage_sha256, + source_sha256=binding.source_sha256, + transformation_kind=binding.transformation_kind, + transformation_artifact_sha256=binding.transformation_artifact_sha256, + prompt_template_sha256=binding.prompt_template_sha256, + teacher_model_sha256=binding.teacher_model_sha256, + teacher_output_sha256=binding.teacher_output_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..5900402d --- /dev/null +++ b/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py @@ -0,0 +1,302 @@ +"""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: + # Pre-reconciliation MRL-0605 compatibility: once PR #299 becomes canonical, + # this must call the original manifest construction-bound validation path. + manifest_snapshot = TemporalCanaryManifest( + canary_id=manifest.canary_id, + source_kind=manifest.source_kind, + canary_artifact_sha256=manifest.canary_artifact_sha256, + temporal_boundary_at=manifest.temporal_boundary_at, + created_at=manifest.created_at, + evaluator_artifact_sha256=manifest.evaluator_artifact_sha256, + topic_tags=manifest.topic_tags, + ) + 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..7806559e --- /dev/null +++ b/tests/test_mesc_mrl_benchmark_derived_generation_v1.py @@ -0,0 +1,240 @@ +"""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(ValueError, match="64 lowercase hex"): + 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="exact tuple"): + 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..5578731c --- /dev/null +++ b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py @@ -0,0 +1,297 @@ +"""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 + ] = getattr(canary_module, "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_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(), + ) From 038bea966b6f4f5f58b6ad947699e5b84ac841b8 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:26:59 +0300 Subject: [PATCH 2/7] fix(mrl): bind benchmark flags to canonical upstream evidence --- .../_mrl_benchmark_derived_generation_v1.py | 74 ++++--------------- 1 file changed, 16 insertions(+), 58 deletions(-) diff --git a/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py b/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py index 0bee8650..9aa6bd03 100644 --- a/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py +++ b/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py @@ -15,8 +15,8 @@ from dataclasses import dataclass from medscale.mesc._mrl_contamination_interfaces_v1 import ( - ContaminationCheckEvidence, ContaminationEvidenceReport, + ContaminationInterfaceError, ) from medscale.mesc._mrl_content_identity_v1 import ( canonical_semantic_bytes, @@ -25,9 +25,11 @@ from medscale.mesc._mrl_training_example_lineage_v1 import ( TrainingExampleLineageContract, TrainingExampleLineageError, - build_training_example_lineage, ) -from medscale.mesc._mrl_training_transformation_binding_v1 import TrainingTransformationBinding +from medscale.mesc._mrl_training_transformation_binding_v1 import ( + TrainingTransformationBinding, + TrainingTransformationBindingError, +) __all__ = [ "BenchmarkDerivedGenerationClassification", @@ -214,7 +216,7 @@ def build_benchmark_derived_generation_flags( classification: BenchmarkDerivedGenerationClassification, benchmark_artifact_sha256: str | None = None, ) -> BenchmarkDerivedGenerationFlags: - """Bind supplied benchmark-derivation evidence to exact MRL-0601/0602/0603 identities.""" + """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" @@ -229,19 +231,18 @@ def build_benchmark_derived_generation_flags( ) try: - lineage_snapshot = build_training_example_lineage(lineage.example) - except TrainingExampleLineageError as exc: + 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( - "training lineage failed canonical revalidation" + "benchmark derivation evidence failed canonical revalidation" ) from exc - if lineage_snapshot.content_sha256 != lineage.content_sha256: - raise BenchmarkDerivedGenerationError( - "training lineage identity changed after construction" - ) - contamination_snapshot = _snapshot_contamination_report(contamination_report) - transformation_snapshot = _snapshot_transformation_binding(transformation_binding) - lineage_sha256 = lineage_snapshot.content_sha256 if contamination_snapshot.training_lineage_sha256 != lineage_sha256: raise BenchmarkDerivedGenerationError( "contamination report does not bind the supplied training lineage" @@ -250,7 +251,7 @@ def build_benchmark_derived_generation_flags( raise BenchmarkDerivedGenerationError( "transformation binding does not bind the supplied training lineage" ) - if transformation_snapshot.source_sha256 != lineage_snapshot.example.source_sha256: + if transformation_snapshot.source_sha256 != example_snapshot.source_sha256: raise BenchmarkDerivedGenerationError( "transformation source identity does not match the canonical lineage source" ) @@ -265,49 +266,6 @@ def build_benchmark_derived_generation_flags( ) -def _snapshot_contamination_report( - report: ContaminationEvidenceReport, -) -> ContaminationEvidenceReport: - if type(report.checks) is not tuple: - raise BenchmarkDerivedGenerationError( - "contamination report checks must remain an exact tuple" - ) - if any(type(item) is not ContaminationCheckEvidence for item in report.checks): - raise BenchmarkDerivedGenerationError( - "contamination report checks contains an invalid item type" - ) - checks = tuple( - ContaminationCheckEvidence( - kind=item.kind, - detector_id=item.detector_id, - detector_artifact_sha256=item.detector_artifact_sha256, - evidence_artifact_sha256=item.evidence_artifact_sha256, - disposition=item.disposition, - similarity_decimal=item.similarity_decimal, - threshold_decimal=item.threshold_decimal, - ) - for item in report.checks - ) - return ContaminationEvidenceReport( - training_lineage_sha256=report.training_lineage_sha256, - checks=checks, - ) - - -def _snapshot_transformation_binding( - binding: TrainingTransformationBinding, -) -> TrainingTransformationBinding: - return TrainingTransformationBinding( - training_lineage_sha256=binding.training_lineage_sha256, - source_sha256=binding.source_sha256, - transformation_kind=binding.transformation_kind, - transformation_artifact_sha256=binding.transformation_artifact_sha256, - prompt_template_sha256=binding.prompt_template_sha256, - teacher_model_sha256=binding.teacher_model_sha256, - teacher_output_sha256=binding.teacher_output_sha256, - ) - - def _require_optional_sha256(value: object, label: str) -> None: if value is not None: _require_sha256(value, label) From 9ab4f4f2b2b24d7753dcf413b388ab0e225c2778 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:27:33 +0300 Subject: [PATCH 3/7] fix(mrl): bind canary workflow to manifest construction identity --- .../mesc/_mrl_temporal_canary_fixture_workflow_v1.py | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py b/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py index 5900402d..c2469fcd 100644 --- a/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py +++ b/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py @@ -233,17 +233,7 @@ def run_temporal_canary_fixture_workflow( raise TemporalCanaryFixtureWorkflowError("parameter_values contains an invalid item type") try: - # Pre-reconciliation MRL-0605 compatibility: once PR #299 becomes canonical, - # this must call the original manifest construction-bound validation path. - manifest_snapshot = TemporalCanaryManifest( - canary_id=manifest.canary_id, - source_kind=manifest.source_kind, - canary_artifact_sha256=manifest.canary_artifact_sha256, - temporal_boundary_at=manifest.temporal_boundary_at, - created_at=manifest.created_at, - evaluator_artifact_sha256=manifest.evaluator_artifact_sha256, - topic_tags=manifest.topic_tags, - ) + manifest_snapshot = manifest._validated_snapshot() surface_snapshot = FixtureResearchSurface._validated_snapshot(surface) evaluator_snapshot = FixtureEvaluator._validated_snapshot(evaluator) parameter_snapshots = tuple( From f5f119d8fd76d07aa1d8eda26fefff420d6f2d56 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:29:03 +0300 Subject: [PATCH 4/7] test(mrl): reject upstream benchmark evidence identity drift --- ...esc_mrl_benchmark_derived_generation_v1.py | 46 ++++++++++++++++++- 1 file changed, 44 insertions(+), 2 deletions(-) diff --git a/tests/test_mesc_mrl_benchmark_derived_generation_v1.py b/tests/test_mesc_mrl_benchmark_derived_generation_v1.py index 7806559e..701b409e 100644 --- a/tests/test_mesc_mrl_benchmark_derived_generation_v1.py +++ b/tests/test_mesc_mrl_benchmark_derived_generation_v1.py @@ -149,7 +149,7 @@ 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(ValueError, match="64 lowercase hex"): + with pytest.raises(BenchmarkDerivedGenerationError, match="canonical revalidation"): build_benchmark_derived_generation_flags( lineage, contamination, @@ -163,7 +163,49 @@ 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="exact tuple"): + 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, From 8a9ed836eb5d95da82992d06cfb7e5e35af8b3c6 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:29:33 +0300 Subject: [PATCH 5/7] test(mrl): reject canary manifest identity drift --- ...sc_mrl_temporal_canary_fixture_workflow_v1.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py index 5578731c..edf29a19 100644 --- a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py +++ b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py @@ -189,6 +189,22 @@ def test_mutated_manifest_fails_closed() -> None: ) +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) From d2f509a2549f13e328b21e1a3e062e779b251266 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:31:15 +0300 Subject: [PATCH 6/7] test(mrl): use direct fixture builder reference --- tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py index edf29a19..cf3e084a 100644 --- a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py +++ b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py @@ -86,7 +86,7 @@ def test_workflow_uses_fixture_snapshots_if_live_evaluator_drifts_mid_call( original_metric_id = evaluator.metric_id original_build_candidate: Callable[ [FixtureResearchSurface, tuple[FixtureParameterValue, ...]], FixtureCandidate - ] = getattr(canary_module, "build_fixture_candidate") + ] = canary_module.build_fixture_candidate mutation_performed = False def mutate_live_evaluator_then_build_snapshot( From 30a5d8441684d1f474565b1f9080d13ab603819e Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Mon, 31 Aug 2026 08:32:20 +0300 Subject: [PATCH 7/7] test(mrl): bind original fixture builder explicitly --- tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py index cf3e084a..ec426560 100644 --- a/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py +++ b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py @@ -86,7 +86,7 @@ def test_workflow_uses_fixture_snapshots_if_live_evaluator_drifts_mid_call( original_metric_id = evaluator.metric_id original_build_candidate: Callable[ [FixtureResearchSurface, tuple[FixtureParameterValue, ...]], FixtureCandidate - ] = canary_module.build_fixture_candidate + ] = build_fixture_candidate mutation_performed = False def mutate_live_evaluator_then_build_snapshot(