diff --git a/.github/workflows/mrl-0604-0606-ruff-diagnostic.yml b/.github/workflows/mrl-0604-0606-ruff-diagnostic.yml new file mode 100644 index 00000000..a6a35554 --- /dev/null +++ b/.github/workflows/mrl-0604-0606-ruff-diagnostic.yml @@ -0,0 +1,21 @@ +name: MRL-0604-0606 Ruff diagnostic + +on: + pull_request: + branches: [main] + +permissions: + contents: read + +jobs: + ruff-format-diff: + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + persist-credentials: false + - uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d + - run: uv sync --frozen + - name: Ruff format diff diagnostic + run: | + uv run ruff format --diff src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py || true 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..14dc5ee7 --- /dev/null +++ b/src/medscale/mesc/_mrl_benchmark_derived_generation_v1.py @@ -0,0 +1,243 @@ +"""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 +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" + + +@dataclass(frozen=True, slots=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" + ) + + @property + def benchmark_derived_flag(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 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(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, + "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, + } + + @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: + 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..a3f288db --- /dev/null +++ b/src/medscale/mesc/_mrl_temporal_canary_fixture_workflow_v1.py @@ -0,0 +1,208 @@ +"""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. It +never reads canary content and cannot place a canary into training or research search. +""" + +from __future__ import annotations + +import re +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.""" + + +@dataclass(frozen=True, slots=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") + _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" + ) + + @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(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", + } + + @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 a sealed canary identity on the pure in-memory fixture surface only.""" + 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") + + try: + 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, + ) + candidate: FixtureCandidate = build_fixture_candidate(surface, parameter_values) + evaluation: FixtureEvaluation = evaluate_fixture_candidate( + surface, + evaluator, + 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.content_sha256, + evaluator_sha256=evaluator.content_sha256, + candidate_sha256=candidate.content_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..e7c43e5b --- /dev/null +++ b/tests/test_mesc_mrl_benchmark_derived_generation_v1.py @@ -0,0 +1,137 @@ +"""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_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_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..1ad45749 --- /dev/null +++ b/tests/test_mesc_mrl_temporal_canary_fixture_workflow_v1.py @@ -0,0 +1,93 @@ +"""MRL-0606 tests for the sealed temporal-canary fixture workflow.""" + +from __future__ import annotations + +import pytest + +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 TemporalCanarySourceKind +from test_mesc_mrl_fixture_research_surface_v1 import _evaluator, _surface, _values +from test_mesc_mrl_temporal_canary_manifest_v1 import _manifest + + +def test_fixture_canary_workflow_is_deterministic_and_identity_bound() -> None: + manifest = _manifest() + evaluator = _evaluator() + surface = _surface(evaluator) + + first = run_temporal_canary_fixture_workflow(manifest, surface, evaluator, _values()) + second = run_temporal_canary_fixture_workflow(manifest, surface, evaluator, _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.surface_sha256 == surface.content_sha256 + assert first.evaluator_sha256 == evaluator.content_sha256 + assert first.observed_score == 1 + assert first.observed_max_score == 2 + + +def test_hand_authored_fixture_canary_remains_r2_fixture_only() -> None: + manifest = _manifest(source_kind=TemporalCanarySourceKind.HAND_AUTHORED_FIXTURE) + evaluator = _evaluator() + surface = _surface(evaluator) + receipt = run_temporal_canary_fixture_workflow(manifest, surface, evaluator, _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: + manifest = _manifest() + evaluator = _evaluator() + surface = _surface(evaluator) + receipt = run_temporal_canary_fixture_workflow(manifest, surface, evaluator, _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_mutated_manifest_fails_closed() -> None: + manifest = _manifest() + object.__setattr__(manifest, "canary_artifact_sha256", "invalid") + evaluator = _evaluator() + surface = _surface(evaluator) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="canonical validation"): + run_temporal_canary_fixture_workflow(manifest, surface, evaluator, _values()) + + +def test_evaluator_surface_mismatch_fails_closed() -> None: + manifest = _manifest() + evaluator = _evaluator() + other_evaluator = _evaluator(targets=(_values()[0],)) + surface = _surface(evaluator) + + with pytest.raises(TemporalCanaryFixtureWorkflowError, match="canonical validation"): + run_temporal_canary_fixture_workflow( + manifest, + surface, + other_evaluator, + _values(), + ) + + +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(), + )