From 8f1926a0551560c3f03e43d72663f2a6d258ad51 Mon Sep 17 00:00:00 2001 From: CC <78262508+Wenbobobo@users.noreply.github.com> Date: Sat, 25 Jul 2026 22:53:50 +0800 Subject: [PATCH] feat(builder): isolate selected formal generation fields --- Builder/src/autolean_builder/__init__.py | 10 + .../src/autolean_builder/fidelity_harness.py | 417 ++++++++++++-- .../src/autolean_builder/source_harness.py | 4 +- Builder/tests/test_source_harness.py | 30 +- Builder/tests/test_workflow.py | 294 +++++++++- benchmarks/source_backed_oci_fixture.py | 30 +- .../tests/test_builder_prover_closed_loop.py | 30 +- docs/builder-fidelity-harness.md | 93 ++- docs/builder-prover-evidence-closure.md | 14 +- docs/phase-1-progress.md | 2 +- .../src/autolean_control_plane/service.py | 311 +++++++++- packages/control_plane/tests/test_service.py | 544 +++++++++++++++++- 12 files changed, 1660 insertions(+), 119 deletions(-) diff --git a/Builder/src/autolean_builder/__init__.py b/Builder/src/autolean_builder/__init__.py index aad0675..18029b5 100644 --- a/Builder/src/autolean_builder/__init__.py +++ b/Builder/src/autolean_builder/__init__.py @@ -19,6 +19,10 @@ from .fidelity_harness import ( AutomaticCheckResult, CandidateFormalization, + CandidateGenerationEnvelope, + CandidateGenerationObligation, + CandidateGenerationTask, + CandidateProposal, CandidateReviewVerdict, EvidenceAuthority, FidelityEvaluation, @@ -34,6 +38,7 @@ SourceClaimSpan, StatementFidelityHarness, TranslationAgent, + TranslationAgentV2, TranslationTask, ) from .pilot_harness import ( @@ -148,6 +153,10 @@ "CalibrationReviewEvidenceV1", "CalibrationRoleV1", "CandidateFormalization", + "CandidateGenerationEnvelope", + "CandidateGenerationObligation", + "CandidateGenerationTask", + "CandidateProposal", "CandidateReviewVerdict", "ChapterSourceSpan", "DownloadObservation", @@ -249,6 +258,7 @@ "TextbookAnchorV1", "TextbookReferenceBindingV1", "TranslationAgent", + "TranslationAgentV2", "TranslationTask", "VerifiedReference", "candidate_integrity_checksum_sha256", diff --git a/Builder/src/autolean_builder/fidelity_harness.py b/Builder/src/autolean_builder/fidelity_harness.py index 719b159..ec19bf9 100644 --- a/Builder/src/autolean_builder/fidelity_harness.py +++ b/Builder/src/autolean_builder/fidelity_harness.py @@ -13,7 +13,7 @@ from dataclasses import dataclass from datetime import UTC, datetime from enum import StrEnum -from typing import Protocol +from typing import Literal, Protocol from autolean_contracts import ( DecisionV1, @@ -22,6 +22,7 @@ FidelityCheckV1, FidelityReportV1, HashKindV1, + MathematicalSpecificationV1, MutationKindV1, MutationProbeV1, MutationResultV1, @@ -29,6 +30,7 @@ ReviewerSignoffV1, StableIdentifierV1, StatementContractV1, + TaskKindV1, canonical_json_bytes, digest_bytes, digest_text, @@ -146,6 +148,8 @@ def __post_init__(self) -> None: raise FidelityHarnessError("semantic obligation descriptions must not be empty") if not self.source_span_ids: raise FidelityHarnessError("semantic obligations must cite at least one source span") + if len(set(self.source_span_ids)) != len(self.source_span_ids): + raise FidelityHarnessError("semantic obligation source span identifiers must be unique") if not self.normalized_fragment.strip() or not self.lean_fragment.strip(): raise FidelityHarnessError( "semantic obligations require normalized and Lean target fragments" @@ -267,9 +271,196 @@ def payload(self) -> dict[str, object]: } +@dataclass(frozen=True, slots=True) +class CandidateGenerationObligation: + """The selected-formal-field-blind portion of an obligation visible to a translator. + + Lean fragments are deliberately retained only in :class:`SemanticObligation`, which is + supplied after candidate generation to mutation and semantic-review roles. This projection + keeps the translator input distinct from the later reviewer-visible obligation record. + """ + + obligation_id: str + kind: SemanticObligationKind + source_span_ids: tuple[StableIdentifierV1, ...] + normalized_fragment: str + + @classmethod + def from_semantic_obligation( + cls, + obligation: SemanticObligation, + ) -> CandidateGenerationObligation: + return cls( + obligation_id=obligation.obligation_id, + kind=obligation.kind, + source_span_ids=obligation.source_span_ids, + normalized_fragment=obligation.normalized_fragment, + ) + + def __post_init__(self) -> None: + if not self.obligation_id.strip(): + raise FidelityHarnessError("candidate-generation obligations require an id") + if not self.source_span_ids or not self.normalized_fragment.strip(): + raise FidelityHarnessError( + "candidate-generation obligations require source spans and normalized text" + ) + + def payload(self) -> dict[str, object]: + return { + "obligation_id": self.obligation_id, + "kind": self.kind.value, + "source_span_ids": [item.value for item in self.source_span_ids], + "normalized_fragment": self.normalized_fragment, + } + + +@dataclass(frozen=True, slots=True) +class CandidateGenerationEnvelope: + """Selected-formal-field-blind Lean context needed to render a declaration candidate.""" + + task_kind: TaskKindV1 + declaration_name: str + namespace: str + lean_version: str + mathlib_revision: str + imports_allowlist: tuple[str, ...] + axioms_allowlist: tuple[str, ...] + rendering_profile: Literal["autolean.full-declaration-exact.v1"] = ( + "autolean.full-declaration-exact.v1" + ) + + @classmethod + def from_contract(cls, contract: StatementContractV1) -> CandidateGenerationEnvelope: + return cls( + task_kind=contract.task_kind, + declaration_name=contract.formal.declaration_name, + namespace=contract.formal.namespace, + lean_version=contract.formal.environment.lean_version, + mathlib_revision=contract.formal.environment.mathlib_revision, + imports_allowlist=contract.formal.imports_allowlist, + axioms_allowlist=contract.formal.axioms_allowlist, + ) + + def __post_init__(self) -> None: + for label, value in ( + ("declaration_name", self.declaration_name), + ("namespace", self.namespace), + ("lean_version", self.lean_version), + ("mathlib_revision", self.mathlib_revision), + ): + if not value.strip() or value != value.strip(): + raise FidelityHarnessError(f"candidate-generation {label} must be trimmed text") + if self.rendering_profile != "autolean.full-declaration-exact.v1": + raise FidelityHarnessError("candidate-generation rendering profile is unsupported") + + def payload(self) -> dict[str, object]: + return { + "task_kind": self.task_kind.value, + "declaration_name": self.declaration_name, + "namespace": self.namespace, + "lean_version": self.lean_version, + "mathlib_revision": self.mathlib_revision, + "imports_allowlist": list(self.imports_allowlist), + "axioms_allowlist": list(self.axioms_allowlist), + "rendering_profile": self.rendering_profile, + } + + +@dataclass(frozen=True, slots=True) +class CandidateGenerationTask: + """The only task payload visible to a V2 translation agent. + + This selected-formal projection excludes all selected Lean statement bytes and hashes, every + ``SemanticObligation.lean_fragment``, and the target-dependent draft contract hash. The + Harness owns the full :class:`TranslationTask` and binds a returned proposal to it server-side + before any later role sees the candidate. + """ + + source_spans: tuple[SourceClaimSpan, ...] + mathematics: MathematicalSpecificationV1 + formalization: CandidateGenerationEnvelope + obligations: tuple[CandidateGenerationObligation, ...] + + @classmethod + def from_contract( + cls, + contract: StatementContractV1, + task: TranslationTask, + ) -> CandidateGenerationTask: + return cls( + source_spans=task.source_spans, + mathematics=contract.mathematics, + formalization=CandidateGenerationEnvelope.from_contract(contract), + obligations=tuple( + CandidateGenerationObligation.from_semantic_obligation(item) + for item in task.obligations + ), + ) + + def __post_init__(self) -> None: + if not self.source_spans: + raise FidelityHarnessError("candidate generation requires at least one source span") + if not self.obligations: + raise FidelityHarnessError("candidate generation requires semantic obligations") + obligation_ids = [item.obligation_id for item in self.obligations] + if len(set(obligation_ids)) != len(obligation_ids): + raise FidelityHarnessError("candidate-generation obligation identifiers must be unique") + known_spans = {item.span_id for item in self.source_spans} + if any(not set(item.source_span_ids) <= known_spans for item in self.obligations): + raise FidelityHarnessError( + "candidate-generation obligation cites an unknown source span" + ) + if any( + item.normalized_fragment not in self.mathematics.normalized_statement + for item in self.obligations + ): + raise FidelityHarnessError( + "candidate-generation obligation is absent from the normalized statement" + ) + + def payload(self) -> dict[str, object]: + """Serialize precisely the agent-visible payload for transport and auditing.""" + + return { + "source_spans": [item.payload() for item in self.source_spans], + "mathematics": self.mathematics.model_dump(mode="json"), + "formalization": self.formalization.payload(), + "obligations": [item.payload() for item in self.obligations], + } + + @property + def content_hash(self) -> DigestV1: + """Identify the canonical bytes visible to every registered translation agent.""" + + return digest_bytes(HashKindV1.PROMPT, canonical_json_bytes(self.payload())) + + +@dataclass(frozen=True, slots=True) +class CandidateProposal: + """The minimal answer returned by a translation agent before Harness binding.""" + + candidate_id: str + lean_statement_source: str + reverse_rendering: str + covered_obligation_ids: tuple[str, ...] + + def __post_init__(self) -> None: + for label, value in ( + ("candidate_id", self.candidate_id), + ("lean_statement_source", self.lean_statement_source), + ("reverse_rendering", self.reverse_rendering), + ): + if not value.strip(): + raise FidelityHarnessError(f"{label} must not be empty") + if not self.covered_obligation_ids: + raise FidelityHarnessError( + "candidate proposal must declare semantic obligation coverage" + ) + + @dataclass(frozen=True, slots=True) class CandidateFormalization: - """One independently generated, contract-bound translation candidate.""" + """One Harness-bound translation candidate from a declared agent registration.""" candidate_id: str actor_id: str @@ -279,6 +470,7 @@ class CandidateFormalization: draft_contract_hash: DigestV1 source_hash: DigestV1 normalized_statement_sha256: str + generation_task_hash: DigestV1 lean_statement_source: str reverse_rendering: str covered_obligation_ids: tuple[str, ...] @@ -321,18 +513,97 @@ def payload(self) -> dict[str, object]: "draft_contract_hash": self.draft_contract_hash.model_dump(mode="json"), "source_hash": self.source_hash.model_dump(mode="json"), "normalized_statement_sha256": self.normalized_statement_sha256, + "generation_task_hash": self.generation_task_hash.model_dump(mode="json"), "lean_statement_source": self.lean_statement_source, "reverse_rendering": self.reverse_rendering, "covered_obligation_ids": list(self.covered_obligation_ids), } -class TranslationAgent(Protocol): +class TranslationAgentV2(Protocol): + actor_id: str + independence_group: str + + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + """Return a proposal from the selected-formal-field-blind generation task.""" + + +# Deprecated pre-RC alias. New code must use TranslationAgentV2. +type TranslationAgent = TranslationAgentV2 + + +@dataclass(frozen=True, slots=True) +class _TranslationAgentRegistration: + agent: TranslationAgentV2 actor_id: str independence_group: str - def translate(self, task: TranslationTask) -> CandidateFormalization: - """Return a candidate bound to the supplied task.""" + +def _validated_role_identity(value: object, *, role: str, field: str) -> str: + if not isinstance(value, str) or not value.strip() or value != value.strip(): + raise FidelityHarnessError(f"{role} {field} must be trimmed text") + return value + + +def _register_translation_agents( + agents: tuple[TranslationAgentV2, ...], +) -> tuple[_TranslationAgentRegistration, ...]: + """Snapshot every declared identity before any untrusted translation call runs.""" + + if len(agents) < 2: + raise FidelityHarnessError("at least two translation agents are required") + registrations = tuple( + _TranslationAgentRegistration( + agent=agent, + actor_id=_validated_role_identity( + agent.actor_id, + role="translation agent", + field="actor_id", + ), + independence_group=_validated_role_identity( + agent.independence_group, + role="translation agent", + field="independence_group", + ), + ) + for agent in agents + ) + if len({item.actor_id for item in registrations}) != len(registrations): + raise FidelityHarnessError("translation agents must have distinct actor identities") + if len({item.independence_group for item in registrations}) < 2: + raise FidelityHarnessError("translation agents require two independence groups") + return registrations + + +def _bind_candidate_proposal( + task: TranslationTask, + generation_task: CandidateGenerationTask, + registration: _TranslationAgentRegistration, + proposal: CandidateProposal, +) -> CandidateFormalization: + """Attach authoritative bindings after the translation agent has returned. + + A proposal cannot nominate its contract revision, source hash, normalized statement hash, + actor identity, or independence group. Those values are set once here from the full, + server-owned task and the registered agent configuration. + """ + + if not isinstance(proposal, CandidateProposal): + raise FidelityHarnessError("translation agent must return a CandidateProposal") + return CandidateFormalization( + candidate_id=proposal.candidate_id, + actor_id=registration.actor_id, + independence_group=registration.independence_group, + contract_id=task.contract_id, + revision=task.revision, + draft_contract_hash=task.draft_contract_hash, + source_hash=task.source_hash, + normalized_statement_sha256=task.normalized_statement_sha256, + generation_task_hash=generation_task.content_hash, + lean_statement_source=proposal.lean_statement_source, + reverse_rendering=proposal.reverse_rendering, + covered_obligation_ids=proposal.covered_obligation_ids, + ) class MutationSuiteAgent(Protocol): @@ -346,6 +617,23 @@ def generate( """Return the complete adversarial mutation suite.""" +@dataclass(frozen=True, slots=True) +class _MutationAgentRegistration: + agent: MutationSuiteAgent + actor_id: str + + +def _register_mutation_agent(agent: MutationSuiteAgent) -> _MutationAgentRegistration: + return _MutationAgentRegistration( + agent=agent, + actor_id=_validated_role_identity( + agent.actor_id, + role="mutation agent", + field="actor_id", + ), + ) + + @dataclass(frozen=True, slots=True) class ObligationReviewVerdict: obligation_id: str @@ -471,6 +759,23 @@ def review(self, packet: SemanticReviewPacket) -> SemanticReviewVerdict: """Return explicit expert judgements; the harness does not infer them.""" +@dataclass(frozen=True, slots=True) +class _SemanticReviewerRegistration: + agent: SemanticReviewAgent + reviewer_id: str + + +def _register_semantic_reviewer(agent: SemanticReviewAgent) -> _SemanticReviewerRegistration: + return _SemanticReviewerRegistration( + agent=agent, + reviewer_id=_validated_role_identity( + agent.reviewer_id, + role="semantic reviewer", + field="reviewer_id", + ), + ) + + @dataclass(frozen=True, slots=True) class AutomaticCheckResult: check_name: str @@ -490,6 +795,7 @@ def payload(self) -> dict[str, object]: @dataclass(frozen=True, slots=True) class FidelityEvaluation: task: TranslationTask + generation_task: CandidateGenerationTask candidates: tuple[CandidateFormalization, ...] mutation_agent_id: str mutation_probes: tuple[MutationProbeV1, ...] @@ -504,6 +810,7 @@ def artifact_payload(self) -> dict[str, object]: return _evaluation_payload( self.task, + self.generation_task, self.candidates, self.mutation_agent_id, self.mutation_probes, @@ -529,8 +836,17 @@ def assert_binds(self, contract: StatementContractV1) -> None: raise FidelityHarnessError( "translation task does not reproduce from the draft contract" ) + expected_generation_task = CandidateGenerationTask.from_contract(contract, expected_task) + if self.generation_task != expected_generation_task: + raise FidelityHarnessError( + "candidate-generation task does not reproduce from the draft contract" + ) expected_automatic = ( - *_candidate_structure_checks(self.task, self.candidates), + *_candidate_structure_checks( + self.task, + self.generation_task, + self.candidates, + ), *_mutation_structure_checks( self.task, self.mutation_agent_id, @@ -555,6 +871,7 @@ def assert_binds(self, contract: StatementContractV1) -> None: ) expected = _evaluation_hash( self.task, + self.generation_task, self.candidates, self.mutation_agent_id, self.mutation_probes, @@ -601,7 +918,7 @@ def assert_binds(self, contract: StatementContractV1) -> None: ), FidelityCheckKindV1.INDEPENDENT_TRANSLATION: ( candidates_accepted and len(self.candidates) >= 2, - "expert compared independently generated candidates", + "expert compared candidates from separately declared registrations", ), FidelityCheckKindV1.POSITIVE_EXAMPLE: ( self.review.positive_example_valid, @@ -662,7 +979,7 @@ def assert_binds(self, contract: StatementContractV1) -> None: class StatementFidelityHarness: - """Run independent translation, mutation, semantic review, and evidence assembly.""" + """Run declared-group translation, mutation, semantic review, and evidence assembly.""" def __init__(self, *, clock: Callable[[], datetime] = utc_now) -> None: self._clock = clock @@ -673,37 +990,59 @@ def run( *, obligations: tuple[SemanticObligation, ...], source_claims: tuple[SourceClaimSpan, ...] | None = None, - translators: tuple[TranslationAgent, ...], + translators: tuple[TranslationAgentV2, ...], mutation_agent: MutationSuiteAgent, reviewer: SemanticReviewAgent, additional_signoffs: tuple[ReviewerSignoffV1, ...] = (), ) -> FidelityEvaluation: + registrations = _register_translation_agents(translators) + mutation_registration = _register_mutation_agent(mutation_agent) + reviewer_registration = _register_semantic_reviewer(reviewer) task = TranslationTask.from_contract( contract, obligations, source_claims=source_claims, ) - candidates = tuple(agent.translate(task) for agent in translators) - automatic_checks = self._validate_candidates(task, translators, candidates) - probes = mutation_agent.generate(task, candidates[0]) - automatic_checks += self._validate_mutations(task, mutation_agent, probes) + generation_task = CandidateGenerationTask.from_contract(contract, task) + candidates = tuple( + _bind_candidate_proposal( + task, + generation_task, + registration, + registration.agent.translate(generation_task), + ) + for registration in registrations + ) + automatic_checks = self._validate_candidates( + task, + generation_task, + registrations, + candidates, + ) + probes = mutation_registration.agent.generate(task, candidates[0]) + automatic_checks += self._validate_mutations( + task, + mutation_registration.actor_id, + probes, + ) packet = SemanticReviewPacket( task=task, candidates=candidates, mutation_probes=probes, ) - review = reviewer.review(packet) - self._validate_review_structure(packet, reviewer, review) + review = reviewer_registration.agent.review(packet) + self._validate_review_structure(packet, reviewer_registration.reviewer_id, review) _assert_declared_actor_separation( candidates, - mutation_agent.actor_id, + mutation_registration.actor_id, review, additional_signoffs, ) evidence_hash = _evaluation_hash( task, + generation_task, candidates, - mutation_agent.actor_id, + mutation_registration.actor_id, probes, review, automatic_checks, @@ -719,8 +1058,9 @@ def run( ) evaluation = FidelityEvaluation( task=task, + generation_task=generation_task, candidates=candidates, - mutation_agent_id=mutation_agent.actor_id, + mutation_agent_id=mutation_registration.actor_id, mutation_probes=probes, review=review, automatic_checks=automatic_checks, @@ -734,38 +1074,35 @@ def run( def _validate_candidates( self, task: TranslationTask, - translators: tuple[TranslationAgent, ...], + generation_task: CandidateGenerationTask, + registrations: tuple[_TranslationAgentRegistration, ...], candidates: tuple[CandidateFormalization, ...], ) -> tuple[AutomaticCheckResult, ...]: - if len(translators) < 2 or len(candidates) != len(translators): + if len(registrations) < 2 or len(candidates) != len(registrations): raise FidelityHarnessError("at least two translation agents are required") - if len({item.actor_id for item in translators}) != len(translators): - raise FidelityHarnessError("translation agents must have distinct actor identities") - if len({item.independence_group for item in translators}) < 2: - raise FidelityHarnessError("translation agents require two independence groups") - for agent, candidate in zip(translators, candidates, strict=True): + for registration, candidate in zip(registrations, candidates, strict=True): if ( - candidate.actor_id != agent.actor_id - or candidate.independence_group != agent.independence_group + candidate.actor_id != registration.actor_id + or candidate.independence_group != registration.independence_group ): raise FidelityHarnessError("candidate misrepresents its generating agent") - return _candidate_structure_checks(task, candidates) + return _candidate_structure_checks(task, generation_task, candidates) def _validate_mutations( self, task: TranslationTask, - mutation_agent: MutationSuiteAgent, + mutation_agent_id: str, probes: tuple[MutationProbeV1, ...], ) -> tuple[AutomaticCheckResult, ...]: - return _mutation_structure_checks(task, mutation_agent.actor_id, probes) + return _mutation_structure_checks(task, mutation_agent_id, probes) def _validate_review_structure( self, packet: SemanticReviewPacket, - reviewer: SemanticReviewAgent, + reviewer_id: str, review: SemanticReviewVerdict, ) -> None: - if review.reviewer_id != reviewer.reviewer_id: + if review.reviewer_id != reviewer_id: raise FidelityHarnessError("semantic verdict misrepresents its reviewer") if not review.independent: raise FidelityHarnessError("semantic review must be independently authored") @@ -824,7 +1161,9 @@ def _build_report( ), kind=FidelityCheckKindV1.INDEPENDENT_TRANSLATION, passed=candidates_accepted and len(candidates) >= 2, - evidence=f"{marker}; expert compared independently generated candidates", + evidence=( + f"{marker}; expert compared candidates from separately declared registrations" + ), reviewer_id=review.reviewer_id, independent=review.independent, ), @@ -904,6 +1243,7 @@ def _now(self) -> datetime: def _candidate_structure_checks( task: TranslationTask, + generation_task: CandidateGenerationTask, candidates: tuple[CandidateFormalization, ...], ) -> tuple[AutomaticCheckResult, ...]: if len(candidates) < 2: @@ -922,6 +1262,7 @@ def _candidate_structure_checks( or candidate.draft_contract_hash != task.draft_contract_hash or candidate.source_hash != task.source_hash or candidate.normalized_statement_sha256 != task.normalized_statement_sha256 + or candidate.generation_task_hash != generation_task.content_hash ): raise FidelityHarnessError("candidate is not bound to the translation task") if candidate.statement_hash != task.selected_statement_hash: @@ -936,7 +1277,10 @@ def _candidate_structure_checks( check_name="candidate_contract_bindings", authority=EvidenceAuthority.AUTOMATIC, passed=True, - evidence="all candidate hashes and revision bindings match the translation task", + evidence=( + "all candidate hashes and revision bindings match the translation task; " + f"generation_task_hash={generation_task.content_hash.value}" + ), ), AutomaticCheckResult( check_name="candidate_independence", @@ -1059,6 +1403,7 @@ def _text_sha256(value: str) -> str: def _evaluation_hash( task: TranslationTask, + generation_task: CandidateGenerationTask, candidates: tuple[CandidateFormalization, ...], mutation_agent_id: str, probes: tuple[MutationProbeV1, ...], @@ -1071,6 +1416,7 @@ def _evaluation_hash( canonical_json_bytes( _evaluation_payload( task, + generation_task, candidates, mutation_agent_id, probes, @@ -1084,6 +1430,7 @@ def _evaluation_hash( def _evaluation_payload( task: TranslationTask, + generation_task: CandidateGenerationTask, candidates: tuple[CandidateFormalization, ...], mutation_agent_id: str, probes: tuple[MutationProbeV1, ...], @@ -1094,6 +1441,8 @@ def _evaluation_payload( return { "schema_version": "autolean.builder-fidelity-evidence.v1", "task": task.payload(), + "generation_task": generation_task.payload(), + "generation_task_hash": generation_task.content_hash.model_dump(mode="json"), "candidates": [item.payload() for item in candidates], "mutation_agent_id": mutation_agent_id, "mutation_probes": [item.model_dump(mode="json") for item in probes], diff --git a/Builder/src/autolean_builder/source_harness.py b/Builder/src/autolean_builder/source_harness.py index afc9487..c6b8636 100644 --- a/Builder/src/autolean_builder/source_harness.py +++ b/Builder/src/autolean_builder/source_harness.py @@ -44,7 +44,7 @@ SemanticReviewAgent, SourceClaimSpan, StatementFidelityHarness, - TranslationAgent, + TranslationAgentV2, ) from .pilot_harness import ( PilotAdmissionReceiptV1, @@ -517,7 +517,7 @@ def run_fidelity( packet: StatementDraftPacket, *, obligations: tuple[SemanticObligation, ...], - translators: tuple[TranslationAgent, ...], + translators: tuple[TranslationAgentV2, ...], mutation_agent: MutationSuiteAgent, reviewer: SemanticReviewAgent, additional_signoffs: tuple[ReviewerSignoffV1, ...] = (), diff --git a/Builder/tests/test_source_harness.py b/Builder/tests/test_source_harness.py index bcd476f..39b45c4 100644 --- a/Builder/tests/test_source_harness.py +++ b/Builder/tests/test_source_harness.py @@ -17,6 +17,8 @@ import pytest from autolean_builder import ( CandidateFormalization, + CandidateGenerationTask, + CandidateProposal, CandidateReviewVerdict, ChapterSourceSpan, DownloadObservation, @@ -523,19 +525,13 @@ def test_real_source_harness_prepare_fidelity_and_freeze_path(tmp_path: Path) -> class Translator: actor_id: str independence_group: str + oracle_lean_statement: str - def translate(self, task: TranslationTask) -> CandidateFormalization: - return CandidateFormalization( + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + return CandidateProposal( candidate_id=f"{self.actor_id}-candidate", - actor_id=self.actor_id, - independence_group=self.independence_group, - contract_id=task.contract_id, - revision=task.revision, - draft_contract_hash=task.draft_contract_hash, - source_hash=task.source_hash, - normalized_statement_sha256=task.normalized_statement_sha256, - lean_statement_source=task.selected_lean_statement, - reverse_rendering=task.normalized_statement, + lean_statement_source=self.oracle_lean_statement, + reverse_rendering=task.mathematics.normalized_statement, covered_obligation_ids=tuple( obligation.obligation_id for obligation in task.obligations ), @@ -623,8 +619,16 @@ def review(self, review_packet: SemanticReviewPacket) -> SemanticReviewVerdict: packet, obligations=obligations, translators=( - Translator("source-translator-a", "source-team-a"), - Translator("source-translator-b", "source-team-b"), + Translator( + "source-translator-a", + "source-team-a", + packet.contract.formal.lean_statement_source, + ), + Translator( + "source-translator-b", + "source-team-b", + packet.contract.formal.lean_statement_source, + ), ), mutation_agent=MutationAgent(), reviewer=Reviewer(), diff --git a/Builder/tests/test_workflow.py b/Builder/tests/test_workflow.py index 1f7563f..8e6b637 100644 --- a/Builder/tests/test_workflow.py +++ b/Builder/tests/test_workflow.py @@ -12,6 +12,8 @@ BuilderError, BuilderStage, CandidateFormalization, + CandidateGenerationTask, + CandidateProposal, CandidateReviewVerdict, EvidenceAuthority, FidelityHarnessError, @@ -236,27 +238,33 @@ def _obligations(contract: StatementContractV1) -> tuple[SemanticObligation, ... ) +def test_semantic_obligation_rejects_duplicate_source_span_ids_at_construction() -> None: + obligation = _obligations(_contract())[0] + + with pytest.raises(FidelityHarnessError, match="source span identifiers must be unique"): + replace( + obligation, + source_span_ids=(obligation.source_span_ids[0], obligation.source_span_ids[0]), + ) + + @dataclass(frozen=True, slots=True) class FakeTranslationAgent: actor_id: str independence_group: str candidate_id: str - corrupt_normalized_binding: bool = False - - def translate(self, task: TranslationTask) -> CandidateFormalization: - normalized_hash = ( - "0" * 64 if self.corrupt_normalized_binding else task.normalized_statement_sha256 + oracle_lean_statement: str + return_mismatched_statement: bool = False + + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + lean_statement_source = ( + "theorem mismatched_fixture : True" + if self.return_mismatched_statement + else self.oracle_lean_statement ) - return CandidateFormalization( + return CandidateProposal( candidate_id=self.candidate_id, - actor_id=self.actor_id, - independence_group=self.independence_group, - contract_id=task.contract_id, - revision=task.revision, - draft_contract_hash=task.draft_contract_hash, - source_hash=task.source_hash, - normalized_statement_sha256=normalized_hash, - lean_statement_source=task.selected_lean_statement, + lean_statement_source=lean_statement_source, reverse_rendering=_golden()["reverse_rendering"], covered_obligation_ids=tuple(item.obligation_id for item in task.obligations), ) @@ -343,9 +351,10 @@ def review(self, packet: SemanticReviewPacket) -> SemanticReviewVerdict: def _translators() -> tuple[FakeTranslationAgent, ...]: + oracle_lean_statement = _golden()["lean_statement"] return ( - FakeTranslationAgent("translator-a", "team-a", "candidate-a"), - FakeTranslationAgent("translator-b", "team-b", "candidate-b"), + FakeTranslationAgent("translator-a", "team-a", "candidate-a", oracle_lean_statement), + FakeTranslationAgent("translator-b", "team-b", "candidate-b", oracle_lean_statement), ) @@ -521,25 +530,266 @@ def test_false_negative_mutation_golden_blocks_freeze(kind: MutationKindV1) -> N ) -def test_harness_rejects_candidate_bound_to_another_normalized_statement() -> None: +def test_harness_rejects_mismatched_candidate_before_reviewer_runs() -> None: contract = _contract() + + @dataclass(frozen=True, slots=True) + class UnreachableMutationAgent: + actor_id: str = "unreachable-mutation-agent" + + def generate( + self, + task: TranslationTask, + selected_candidate: CandidateFormalization, + ) -> tuple[MutationProbeV1, ...]: + del task, selected_candidate + raise AssertionError("mutation must not receive a candidate that fails exact matching") + + @dataclass(frozen=True, slots=True) + class UnreachableReviewer: + reviewer_id: str = "unreachable-reviewer" + + def review(self, packet: SemanticReviewPacket) -> SemanticReviewVerdict: + del packet + raise AssertionError("reviewer must not receive a candidate that fails exact matching") + translators = ( FakeTranslationAgent( "translator-a", "team-a", "candidate-a", - corrupt_normalized_binding=True, + _golden()["lean_statement"], + return_mismatched_statement=True, ), - FakeTranslationAgent("translator-b", "team-b", "candidate-b"), + FakeTranslationAgent("translator-b", "team-b", "candidate-b", _golden()["lean_statement"]), ) - with pytest.raises(FidelityHarnessError, match="not bound"): + with pytest.raises(FidelityHarnessError, match="exactly match"): StatementFidelityHarness().run( contract, obligations=_obligations(contract), translators=translators, - mutation_agent=FakeMutationSuiteAgent(), - reviewer=FakeSemanticReviewer(), + mutation_agent=UnreachableMutationAgent(), + reviewer=UnreachableReviewer(), + ) + + +def test_translation_agent_receives_selected_formal_field_blind_generation_payload() -> None: + contract = _contract() + base_obligations = _obligations(contract) + obligations = ( + replace( + base_obligations[0], + description=contract.formal.lean_statement_source, + ), + *base_obligations[1:], + ) + observed_tasks: list[CandidateGenerationTask] = [] + observed_payloads: list[dict[str, object]] = [] + + @dataclass(frozen=True, slots=True) + class SpyTranslator: + actor_id: str + independence_group: str + candidate_id: str + oracle_lean_statement: str + + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + assert isinstance(task, CandidateGenerationTask) + observed_tasks.append(task) + observed_payloads.append(task.payload()) + return CandidateProposal( + candidate_id=self.candidate_id, + lean_statement_source=self.oracle_lean_statement, + reverse_rendering=_golden()["reverse_rendering"], + covered_obligation_ids=tuple(item.obligation_id for item in task.obligations), + ) + + evaluation = StatementFidelityHarness().run( + contract, + obligations=obligations, + translators=( + SpyTranslator("spy-a", "team-a", "candidate-a", _golden()["lean_statement"]), + SpyTranslator("spy-b", "team-b", "candidate-b", _golden()["lean_statement"]), + ), + mutation_agent=FakeMutationSuiteAgent(), + reviewer=FakeSemanticReviewer(), + ) + + forbidden_keys = { + "selected_lean_statement", + "selected_statement_hash", + "draft_contract_hash", + "lean_fragment", + } + forbidden_values = { + contract.formal.lean_statement_source, + contract.formal.statement_source_hash.value, + contract.semantic_hash().value, + *(obligation.lean_fragment for obligation in obligations), + } + + def assert_selected_formal_fields_absent(value: object) -> None: + if isinstance(value, dict): + assert not (set(value) & forbidden_keys) + for nested in value.values(): + assert_selected_formal_fields_absent(nested) + elif isinstance(value, (list, tuple)): + for nested in value: + assert_selected_formal_fields_absent(nested) + elif isinstance(value, str): + assert all(forbidden not in value for forbidden in forbidden_values) + + assert len(observed_tasks) == 2 + for task in observed_tasks: + assert not hasattr(task, "selected_lean_statement") + assert not hasattr(task, "selected_statement_hash") + assert not hasattr(task, "draft_contract_hash") + assert all( + not hasattr(obligation, "lean_fragment") and not hasattr(obligation, "description") + for obligation in task.obligations ) + assert task.mathematics == contract.mathematics + assert task.formalization.task_kind is contract.task_kind + assert task.formalization.declaration_name == contract.formal.declaration_name + assert task.formalization.namespace == contract.formal.namespace + assert task.formalization.lean_version == contract.formal.environment.lean_version + assert task.formalization.mathlib_revision == contract.formal.environment.mathlib_revision + assert task.formalization.imports_allowlist == contract.formal.imports_allowlist + assert task.formalization.axioms_allowlist == contract.formal.axioms_allowlist + assert task.formalization.rendering_profile == "autolean.full-declaration-exact.v1" + assert len(observed_payloads) == 2 + for payload in observed_payloads: + assert_selected_formal_fields_absent(payload) + projected_obligations = payload["obligations"] + assert isinstance(projected_obligations, list) + assert all( + isinstance(obligation, dict) and "description" not in obligation + for obligation in projected_obligations + ) + assert all(candidate.contract_id == contract.contract_id for candidate in evaluation.candidates) + assert all(candidate.revision == contract.revision for candidate in evaluation.candidates) + assert all( + candidate.draft_contract_hash == contract.semantic_hash() + for candidate in evaluation.candidates + ) + assert all( + candidate.generation_task_hash == evaluation.generation_task.content_hash + for candidate in evaluation.candidates + ) + artifact = evaluation.artifact_payload() + assert artifact["generation_task"] == evaluation.generation_task.payload() + assert artifact["generation_task_hash"] == evaluation.generation_task.content_hash.model_dump( + mode="json" + ) + binding_check = next( + item + for item in evaluation.automatic_checks + if item.check_name == "candidate_contract_bindings" + ) + assert evaluation.generation_task.content_hash.value in binding_check.evidence + + tampered_candidate = replace( + evaluation.candidates[0], + generation_task_hash=digest_text(HashKindV1.PROMPT, "detached-generation-task"), + ) + tampered_evaluation = replace( + evaluation, + candidates=(tampered_candidate, *evaluation.candidates[1:]), + ) + with pytest.raises(FidelityHarnessError, match="candidate is not bound"): + tampered_evaluation.assert_binds(contract) + + +def test_harness_freezes_all_role_identities_before_untrusted_calls() -> None: + contract = _contract() + + @dataclass(slots=True) + class MutableMutationAgent: + actor_id: str = "registered-mutation-agent" + + def generate( + self, + task: TranslationTask, + selected_candidate: CandidateFormalization, + ) -> tuple[MutationProbeV1, ...]: + self.actor_id = "spoofed-mutation-agent" + return FakeMutationSuiteAgent().generate(task, selected_candidate) + + @dataclass(slots=True) + class MutableReviewer: + reviewer_id: str = "registered-semantic-reviewer" + + def review(self, packet: SemanticReviewPacket) -> SemanticReviewVerdict: + self.reviewer_id = "spoofed-semantic-reviewer" + return FakeSemanticReviewer(reviewer_id="registered-semantic-reviewer").review(packet) + + @dataclass(slots=True) + class MutableTranslator: + actor_id: str + independence_group: str + candidate_id: str + oracle_lean_statement: str + peer: MutableTranslator | None = None + mutation_agent: MutableMutationAgent | None = None + reviewer: MutableReviewer | None = None + + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + self.actor_id = f"spoofed-{self.candidate_id}" + self.independence_group = f"spoofed-group-{self.candidate_id}" + if self.peer is not None: + self.peer.actor_id = "spoofed-before-second-translate" + self.peer.independence_group = "spoofed-before-second-group" + if self.mutation_agent is not None: + self.mutation_agent.actor_id = "spoofed-before-mutation" + if self.reviewer is not None: + self.reviewer.reviewer_id = "spoofed-before-review" + return CandidateProposal( + candidate_id=self.candidate_id, + lean_statement_source=self.oracle_lean_statement, + reverse_rendering=_golden()["reverse_rendering"], + covered_obligation_ids=tuple(item.obligation_id for item in task.obligations), + ) + + mutation_agent = MutableMutationAgent() + reviewer = MutableReviewer() + second = MutableTranslator( + "registered-translator-b", + "registered-group-b", + "candidate-b", + _golden()["lean_statement"], + ) + first = MutableTranslator( + "registered-translator-a", + "registered-group-a", + "candidate-a", + _golden()["lean_statement"], + peer=second, + mutation_agent=mutation_agent, + reviewer=reviewer, + ) + + evaluation = StatementFidelityHarness().run( + contract, + obligations=_obligations(contract), + translators=(first, second), + mutation_agent=mutation_agent, + reviewer=reviewer, + ) + + assert tuple(candidate.actor_id for candidate in evaluation.candidates) == ( + "registered-translator-a", + "registered-translator-b", + ) + assert tuple(candidate.independence_group for candidate in evaluation.candidates) == ( + "registered-group-a", + "registered-group-b", + ) + assert evaluation.mutation_agent_id == "registered-mutation-agent" + assert evaluation.review.reviewer_id == "registered-semantic-reviewer" + assert first.actor_id == "spoofed-candidate-a" + assert second.actor_id == "spoofed-candidate-b" + assert mutation_agent.actor_id == "spoofed-mutation-agent" + assert reviewer.reviewer_id == "spoofed-semantic-reviewer" def test_harness_rejects_an_incomplete_mutation_suite() -> None: diff --git a/benchmarks/source_backed_oci_fixture.py b/benchmarks/source_backed_oci_fixture.py index e389454..db32711 100644 --- a/benchmarks/source_backed_oci_fixture.py +++ b/benchmarks/source_backed_oci_fixture.py @@ -16,6 +16,8 @@ from autolean_builder import ( CandidateFormalization, + CandidateGenerationTask, + CandidateProposal, CandidateReviewVerdict, ChapterSourceSpan, FidelityEvaluation, @@ -357,19 +359,13 @@ def _obligations(packet: StatementDraftPacket) -> tuple[SemanticObligation, ...] class _Translator: actor_id: str independence_group: str + oracle_lean_statement: str - def translate(self, task: TranslationTask) -> CandidateFormalization: - return CandidateFormalization( + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + return CandidateProposal( candidate_id=f"candidate-{self.actor_id}", - actor_id=self.actor_id, - independence_group=self.independence_group, - contract_id=task.contract_id, - revision=task.revision, - draft_contract_hash=task.draft_contract_hash, - source_hash=task.source_hash, - normalized_statement_sha256=task.normalized_statement_sha256, - lean_statement_source=task.selected_lean_statement, - reverse_rendering=task.normalized_statement, + lean_statement_source=self.oracle_lean_statement, + reverse_rendering=task.mathematics.normalized_statement, covered_obligation_ids=tuple(item.obligation_id for item in task.obligations), ) @@ -486,8 +482,16 @@ def build_source_backed_oci_fixture( packet, obligations=_obligations(packet), translators=( - _Translator("oci-source-backed-translator-a", "independence-a"), - _Translator("oci-source-backed-translator-b", "independence-b"), + _Translator( + "oci-source-backed-translator-a", + "independence-a", + packet.contract.formal.lean_statement_source, + ), + _Translator( + "oci-source-backed-translator-b", + "independence-b", + packet.contract.formal.lean_statement_source, + ), ), mutation_agent=_MutationAgent(), reviewer=_SemanticReviewer(), diff --git a/benchmarks/tests/test_builder_prover_closed_loop.py b/benchmarks/tests/test_builder_prover_closed_loop.py index 2687a07..8a59a46 100644 --- a/benchmarks/tests/test_builder_prover_closed_loop.py +++ b/benchmarks/tests/test_builder_prover_closed_loop.py @@ -11,6 +11,8 @@ import pytest from autolean_builder import ( CandidateFormalization, + CandidateGenerationTask, + CandidateProposal, CandidateReviewVerdict, ChapterSourceSpan, DownloadObservation, @@ -377,19 +379,13 @@ def _obligations(contract: StatementContractV1) -> tuple[SemanticObligation, ... class _Translator: actor_id: str independence_group: str + oracle_lean_statement: str - def translate(self, task: TranslationTask) -> CandidateFormalization: - return CandidateFormalization( + def translate(self, task: CandidateGenerationTask) -> CandidateProposal: + return CandidateProposal( candidate_id=f"candidate-{self.actor_id}", - actor_id=self.actor_id, - independence_group=self.independence_group, - contract_id=task.contract_id, - revision=task.revision, - draft_contract_hash=task.draft_contract_hash, - source_hash=task.source_hash, - normalized_statement_sha256=task.normalized_statement_sha256, - lean_statement_source=task.selected_lean_statement, - reverse_rendering=task.normalized_statement, + lean_statement_source=self.oracle_lean_statement, + reverse_rendering=task.mathematics.normalized_statement, covered_obligation_ids=tuple(item.obligation_id for item in task.obligations), ) @@ -500,8 +496,16 @@ def _evaluation( packet, obligations=_obligations(packet.contract), translators=( - _Translator("translator-a", "independence-a"), - _Translator("translator-b", "independence-b"), + _Translator( + "translator-a", + "independence-a", + packet.contract.formal.lean_statement_source, + ), + _Translator( + "translator-b", + "independence-b", + packet.contract.formal.lean_statement_source, + ), ), mutation_agent=_MutationAgent(), reviewer=_SemanticReviewer(), diff --git a/docs/builder-fidelity-harness.md b/docs/builder-fidelity-harness.md index eb8e2e2..0c000a3 100644 --- a/docs/builder-fidelity-harness.md +++ b/docs/builder-fidelity-harness.md @@ -27,21 +27,32 @@ to `StatementFidelityHarness` through a hash-checked private `SourceClaimSpan`. fidelity artifact remains rights-scoped, while the Builder-to-Prover bundle exposes only its digest and byte length. -## Closed loop +## Preformal evidence flow 1. `TranslationTask.from_contract` binds public source-span hashes to rights-scoped private excerpts, then snapshots the informal and normalized statements, selected Lean statement, - contract revision, and semantic obligations. -2. At least two `TranslationAgent` implementations produce candidates under distinct actor IDs and - independence groups. Each candidate binds the source, normalized text, contract revision, - selected Lean bytes, reverse rendering, and obligation coverage. -3. A `MutationSuiteAgent` emits the required adversarial statement variants. -4. An independent `SemanticReviewAgent` receives the complete packet. It returns per-candidate, + contract revision, and complete semantic obligations for the Harness. +2. Before calling a `TranslationAgentV2`, the Harness derives a selected-formal-field-blind + `CandidateGenerationTask`. It contains the rights-scoped source claims, the immutable + `MathematicalSpecificationV1`, and an obligation projection containing only ID, kind, source + span IDs, and a normalized-statement fragment. Caller-authored obligation descriptions and Lean + fragments are not projected. A selected-formal-field-blind formalization envelope supplies task + kind, declaration name, namespace, Lean and mathlib versions, import and axiom allowlists, and the fixed + `autolean.full-declaration-exact.v1` rendering profile. The task excludes selected Lean bytes, + selected-statement hashes, elaborated types, and the target-dependent draft-contract hash. +3. The Harness hashes the canonical agent-visible projection as `HashKindV1.PROMPT`. The agent + returns only a `CandidateProposal` with an identifier, proposed Lean source, reverse rendering, + and coverage. The Harness injects the generation-task hash plus the contract, revision, draft, + source, normalized-text, actor, and independence bindings server-side to form + `CandidateFormalization`. +4. A `MutationSuiteAgent` emits the required adversarial statement variants. +5. An independent `SemanticReviewAgent` receives the complete packet. It returns per-candidate, per-obligation, and per-mutation verdicts plus example and non-vacuity evidence. -5. `FidelityEvaluation` canonically serializes the task, candidates, mutations, structural checks, - expert verdict, and extra signoffs. Its SHA-256 is a typed field of `FidelityReportV1` and is - also embedded in every derived fidelity check, mutation result, and semantic signoff. -6. The internal freeze primitive accepts only a complete `FidelityEvaluation`. It recomputes its +6. `FidelityEvaluation` canonically serializes both task projections, candidates, mutations, + structural checks, expert verdict, and extra signoffs. Its SHA-256 is a typed field of + `FidelityReportV1` and is also embedded in every derived fidelity check, mutation result, and + semantic signoff. +7. The internal freeze primitive accepts only a complete `FidelityEvaluation`. It recomputes its artifact hash, checks every derived report field against the expert verdict, applies the risk-tier freeze gate, and only then produces the frozen contract. @@ -51,11 +62,20 @@ the bundle. The Builder attestation covers that reference through the handoff ha registration rehashes the bytes, checks their task/statement/source binding, and records the reference in the append-only registration event. +The source claims, mathematical specification, and formalization envelope are the preformal trust +boundary: the Harness copies them from an already prepared draft and does not prove that their +contents were authored from the source without contamination. The generation-task hash proves +content identity only. It does not prove execution chronology, process isolation, or independence +of the declared translation role. Identity strings for all +translation, mutation, and review roles are snapshotted before the first untrusted role call; this +prevents later attribute mutation from rewriting evidence but does not authenticate the people or +processes behind those strings. + ## Decision boundary | Check | Authority | What the Harness establishes | | --- | --- | --- | -| Source, revision, normalized-text, statement, and candidate hashes | Automatic | Exact byte and revision identity | +| Source, revision, generation-task, statement, and candidate hashes | Automatic | Exact content and revision identity | | Candidate actor IDs and declared independence groups | Automatic | Distinct declared origins; not epistemic independence | | Obligation coverage and source-span references | Automatic | Complete and well-formed trace declarations | | Mutation suite shape | Automatic | Required unique kinds exist and change statement bytes | @@ -95,8 +115,9 @@ The synthetic golden fixture is [`statement_fidelity_golden.json`](../Builder/tests/fixtures/statement_fidelity_golden.json). The tests run two fake translators, a fake mutation agent, and a fake expert reviewer. For every required mutation, a deliberately false-negative reviewer verdict reaches the evidence report but -is rejected by the freeze gate. This tests plumbing and policy only; it is not human fidelity -calibration and is not Lean compilation evidence. +is rejected by the freeze gate. This tests plumbing and policy only; it is not independently +isolated selected-formal-blind model evidence, human fidelity calibration, or Lean compilation +evidence. ## Operational requirements @@ -118,3 +139,47 @@ calibration and is not Lean compilation evidence. V1 intentionally requires both independent candidates to match the selected Lean statement bytes. Supporting syntactically different but propositionally equivalent candidates needs a separately audited equivalence certificate and is deferred rather than inferred from model agreement. + +An exact-match mismatch fails before mutation or reviewer execution. V1 records no inferred +semantic reconciliation for such a mismatch: a versioned disagreement artifact and its review +workflow are deliberately deferred to a later contract revision. The deterministic test fixtures +carry fixed oracle Lean source when they are constructed solely to exercise the plumbing; their +successful equality is not evidence that an independent agent generated the statement. + +This exact-byte rule is selected-formal-field-blind groundwork, not a usable blind-formalization +closed loop. It is deliberately low-liveness: a generator must reproduce the preselected full +declaration byte for byte, including formatting. No real model run currently establishes that +capability. Independently created preformal artifacts and chronology receipts remain separate +requirements. Acceptance by canonical elaborated-type identity, rather than selected source bytes, +is reserved for a separate audited change. + +## Pre-RC artifact cutoff + +`autolean.builder-fidelity-evidence.v1` remains the artifact schema before the first release +candidate; this is an intentional migration cutoff, not a silent `FormalizationTaskBundleV2` +union. New registration requires the V1 top-level shape to contain `generation_task`, its complete +typed `HashKindV1.PROMPT` digest, and the same typed digest on every candidate. The control plane +derives the only permitted generation projection from the retained task and frozen contract, +rebuilds every source claim from its frozen-span binding while preserving the already validated +claim order. The private excerpt remains artifact-resident for rights reasons, yet must rehash to +the frozen span's typed hash; its ID, locator, and hash cannot come from the artifact. Obligation +records remain private artifact evidence, but registration +requires their exact V1 shape, valid kind, unique nonempty IDs, references only to frozen spans, +and normalized fragments that occur in the frozen normalized statement. The generation projection +contains none of each obligation's description, Lean fragment, or authority. The control plane +rehashes the resulting projection with `canonical_json_bytes` and `digest_bytes`, and rejects any +shape, projection, kind, or value mismatch before recording `task.registered`. + +Old V1 source-backed artifacts without these fields remain historical evidence only. They can be +read through existing event replay but cannot be newly registered or used for promotion. The +explicit `allow_test_only_unreviewed_bundles` switch remains limited to a bundle with neither a +fidelity report nor an artifact reference; it is not an exception for legacy or malformed reviewed +V1 evidence. + +## Pre-RC API migration + +The translator signature changed before the first release candidate. New implementations must use +`TranslationAgentV2.translate(CandidateGenerationTask) -> CandidateProposal`. The exported +`TranslationAgent` name is only a deprecated pre-RC type alias and will be removed; it does not +preserve the old full-`TranslationTask` return contract. Mutation and semantic-review signatures +remain unchanged and continue to receive the full task only after candidate exact matching. diff --git a/docs/builder-prover-evidence-closure.md b/docs/builder-prover-evidence-closure.md index a3eafea..36ad982 100644 --- a/docs/builder-prover-evidence-closure.md +++ b/docs/builder-prover-evidence-closure.md @@ -10,8 +10,10 @@ The offline integration fixture in `FidelityEvidenceArtifactRefV1`; 3. only the reviewed draft can freeze, and only that frozen revision can produce a signed `FormalizationTaskBundleV1`; -4. the control plane rehashes and minimally cross-binds the canonical fidelity artifact, roots - its reference in `task.registered`, and only then permits a fenced claim; +4. the control plane requires the pre-RC V1 generation projection and typed prompt digests, + rebuilds each source claim from frozen source-span bindings while rehashing its private excerpt, + validates the private obligation projection against frozen spans and normalized text, roots the + artifact reference in `task.registered`, and only then permits a fenced claim; 5. a Prover-side failure can append a revision-bound `GapReportV1` and `ContractChangeRequestV1`, but neither command can mutate the registered bundle; 6. the next revision begins as a draft, rejects the old fidelity evaluation, and must run the @@ -28,8 +30,12 @@ the authoritative Linux/OCI verifier path. - `FidelityReportV1.evidence_hash`, the typed bundle reference, Builder handoff hash, retained bytes, and `task.registered` reference now form one content chain. Production registration - requires it. Synthetic legacy fixtures must opt into the conspicuously named - `allow_test_only_unreviewed_bundles` switch; that switch is not a migration or release mode. + requires it. Before the first RC, the artifact schema string remains V1 but new registration + fails closed unless the complete generation projection and all typed prompt digests validate. + Earlier source-backed V1 evidence without those fields is retained for historical replay only, + not new registration or promotion. Synthetic unreviewed fixtures may opt into the conspicuously + named `allow_test_only_unreviewed_bundles` switch only when they carry neither report nor + artifact; that switch is not a migration or release mode. - Registration now commits the `task.registered` event, Builder-attestation nonce, idempotency record, and immutable `(contract_id, revision) -> (bundle_id, handoff_hash)` projection in one SQLite transaction. Exact retries reuse the original event; conflicting legacy events make diff --git a/docs/phase-1-progress.md b/docs/phase-1-progress.md index c28a65a..948c695 100644 --- a/docs/phase-1-progress.md +++ b/docs/phase-1-progress.md @@ -48,7 +48,7 @@ These labels are deliberately non-interchangeable: | Builder source-to-freeze path | Verified, local/synthetic | Official parent/text hashes, exact UTF-8 byte spans, append-only SQLite CAS preparation, two independent candidates, complete mutation suite, semantic/library review, and a real `prepare -> fidelity -> freeze` test | Reviewer and freezer IDs are unauthenticated strings; no production Builder signer | | Public/private source boundary | Verified | Public source-backed contracts retain span IDs, locators, offsets, and typed hashes but no verbatim excerpt; exact text remains in the ignored source cache and private fidelity artifact; public-readiness rejects JSON with non-null `permitted_excerpt` | Not a forensic scanner and not protection from a hostile local administrator | | Builder experience retrieval | Verified, synthetic | Content-addressed success/negative/gap records, rights/endpoint filters, deterministic ranking and budgets, poison-text rejection, and validated retrieval skill | Advisory context cannot satisfy a freeze gate | -| Builder--Prover closure | Verified, local test-only | A synthetic source-backed statement passed the public Builder `prepare -> fidelity -> freeze -> bridge` path, register/claim, immutable workspace materialization, real pure-Lean exact-image OCI execution, independent gateway rerun, and terminal acceptance. Evidence SHA `a83e703add32f9c896b3dbcd5f81982dabcfe5d036799a70cf5b5f9cd500a62e`; handoff hash `770ad9dab1fabd15722aba6cdc938d2172de4fe7a9e166ff508aa219cbc71375`; source-preparation hash `312be270ddb575ce036a7b286199027a2995d867c3e31acd935d70d9911b685d` | The source and review identities are synthetic fixtures; execution is `test-only-local`, the result is `not_a_promotion`, and no production authority is established | +| Builder--Prover closure | Verified, local test-only | A post-cutoff WSL pure-Lean exact-image canary ran `prepare -> fidelity -> freeze -> bridge -> register -> claim -> OCI compile/query -> independent gateway -> terminal acceptance` with `builder_unreviewed_bypass=false`. Builder fidelity digest `d08a743f264db53c7a5b7c8c27e8a7387796559e218426883a17ed056cbe3b57`; handoff hash `2a7600baddfe5a2335b38ebdec5ee08a18261a3e6d3045c9055303fdfbe654c7`; verifier evidence SHA `55d53a7e92839badba32c4ef4287dd4895cfcfdcad73488bd96dec1ca77b50f8`; source-preparation hash `312be270ddb575ce036a7b286199027a2995d867c3e31acd935d70d9911b685d`; image `autolean/lean-worker@sha256:9a85f190bfaaf5cc79418abe3cee46cf5456b9aaaa0c78df5d3c1e380ee419e5`. Earlier V1 source-backed evidence without the generation projection remains historical only and cannot authorize new registration or promotion. | The source and review identities are synthetic fixtures; execution is `test-only-local`, the result is `not_a_promotion`, and no production authority is established. | | Event, lease, and artifact control plane | Verified, synthetic | SQLite WAL, command idempotency, fencing, append-only events, concurrent artifact create-if-absent, and immutable contract-revision projection | No power-loss or storage-device fault injection | | OS-process recovery | Verified, synthetic | 1,000 jobs; 1,000 expired leases/stale-fence rejections; 4,000 duplicate-delivery replays; 5,000 contiguous events; 4,000 artifact checks; no loss or duplicate terminal verdict. Report SHA `d80c2c0dbcbab22bfcbd0bea13f41e07c5f337a4a93d82203ef060e7765a2847` | No Lean, OCI, network, model, physical power loss, or mid-transaction kill | | Verifier evidence and signing gateway | Verified, local test-only | Lease/context/digest-bound request, evidence-artifact rehash and cross-binding, replay ledger, fail-closed outage, stale-fence rejection, and mandatory independently authenticated execution receipt before signing | Fixture HMAC proves the software boundary only; production mTLS/ACL and KMS/HSM custody are absent | diff --git a/packages/control_plane/src/autolean_control_plane/service.py b/packages/control_plane/src/autolean_control_plane/service.py index 1da5a78..d338228 100644 --- a/packages/control_plane/src/autolean_control_plane/service.py +++ b/packages/control_plane/src/autolean_control_plane/service.py @@ -28,6 +28,9 @@ VerificationSigningContextV1, VerificationSigningLeaseBindingV1, builder_attestation_payload, + canonical_json_bytes, + digest_bytes, + digest_text, proof_dependency_manifest_hash, verification_attestation_payload, verification_gateway_attestation_payload, @@ -59,6 +62,45 @@ _PLACEHOLDER_RE = re.compile(r"\b(?:sorry|admit)\b|sorryAx") +_FIDELITY_ARTIFACT_V1_FIELDS = frozenset( + { + "schema_version", + "task", + "generation_task", + "generation_task_hash", + "candidates", + "mutation_agent_id", + "mutation_probes", + "review", + "automatic_checks", + "additional_signoffs", + } +) +_DIGEST_V1_FIELDS = frozenset({"schema_version", "kind", "algorithm", "value"}) +_SOURCE_CLAIM_SPAN_FIELDS = frozenset({"span_id", "locator", "content_hash", "permitted_excerpt"}) +_SEMANTIC_OBLIGATION_FIELDS = frozenset( + { + "obligation_id", + "kind", + "description", + "source_span_ids", + "normalized_fragment", + "lean_fragment", + "authority", + } +) +_SEMANTIC_OBLIGATION_KINDS = frozenset( + { + "quantifier_order", + "assumption", + "conclusion", + "side_condition", + "definition", + "edge_case", + "non_vacuity", + } +) + @dataclass(frozen=True, slots=True) class TaskBinding: @@ -723,12 +765,16 @@ def _verify_builder_fidelity_artifact( fidelity = bundle.contract.fidelity reference = bundle.fidelity_evidence - if fidelity is None or reference is None: + if fidelity is None and reference is None: if self.allow_test_only_unreviewed_bundles: return None raise InvalidTransition( "a canonical Builder fidelity artifact is required before registration" ) + if fidelity is None or reference is None: + raise InvalidTransition( + "reviewed Builder handoffs require a canonical fidelity artifact reference" + ) artifact = ArtifactRef(digest=reference.digest.value, size=reference.size) try: self.artifacts.verify(artifact) @@ -742,7 +788,33 @@ def _verify_builder_fidelity_artifact( ) if payload.get("schema_version") != reference.artifact_schema: raise InvalidTransition("Builder fidelity artifact has an unexpected schema") + self._require_exact_object_keys( + payload, + _FIDELITY_ARTIFACT_V1_FIELDS, + label="Builder fidelity artifact", + ) task = self._required_object(payload, "task") + generation_task = self._required_object(payload, "generation_task") + source_spans = self._source_claim_span_projection(bundle, task) + expected_generation_task = self._generation_task_projection(bundle, task, source_spans) + if generation_task != expected_generation_task: + raise InvalidTransition( + "Builder fidelity artifact generation task differs from the Builder projection" + ) + generation_task_hash = digest_bytes( + HashKindV1.PROMPT, + canonical_json_bytes(generation_task), + ) + declared_generation_task_hash = self._required_fidelity_digest( + payload, + "generation_task_hash", + expected_kind=HashKindV1.PROMPT, + ) + if declared_generation_task_hash != generation_task_hash: + raise InvalidTransition( + "Builder fidelity artifact generation task does not match generation_task_hash" + ) + self._validate_candidate_generation_task_hashes(payload, generation_task_hash) source_hash = self._required_object(task, "source_hash") statement_hash = self._required_object(task, "selected_statement_hash") checks = ( @@ -784,6 +856,234 @@ def _verify_builder_fidelity_artifact( raise InvalidTransition(f"Builder fidelity artifact has a different {label}") return artifact + def _generation_task_projection( + self, + bundle: FormalizationTaskBundleV1, + task: JsonObject, + source_spans: list[JsonObject], + ) -> JsonObject: + obligations: list[JsonObject] = [] + frozen_span_ids = {span.span_id.value for span in bundle.contract.source.spans} + obligation_ids: set[str] = set() + for value in self._required_list(task, "obligations"): + if not isinstance(value, dict): + raise InvalidTransition("Builder fidelity artifact task obligations are invalid") + full_obligation = self._json_object(value) + self._require_exact_object_keys( + full_obligation, + _SEMANTIC_OBLIGATION_FIELDS, + label="Builder fidelity artifact task obligation", + ) + obligation_id = self._required_nonblank_fidelity_obligation_text( + full_obligation, + "obligation_id", + ) + if obligation_id in obligation_ids: + raise InvalidTransition( + "Builder fidelity artifact task obligation identifiers must be unique" + ) + obligation_ids.add(obligation_id) + kind = self._required_text(full_obligation, "kind") + if kind not in _SEMANTIC_OBLIGATION_KINDS: + raise InvalidTransition("Builder fidelity artifact task obligation kind is invalid") + self._required_nonblank_fidelity_obligation_text(full_obligation, "description") + source_span_ids = self._required_texts(full_obligation, "source_span_ids") + if ( + not source_span_ids + or len(set(source_span_ids)) != len(source_span_ids) + or not set(source_span_ids) <= frozen_span_ids + ): + raise InvalidTransition( + "Builder fidelity artifact task obligation source spans are invalid" + ) + normalized_fragment = self._required_nonblank_fidelity_obligation_text( + full_obligation, + "normalized_fragment", + ) + if normalized_fragment not in bundle.contract.mathematics.normalized_statement: + raise InvalidTransition( + "Builder fidelity artifact task obligation fragment is absent from the " + "normalized statement" + ) + self._required_nonblank_fidelity_obligation_text(full_obligation, "lean_fragment") + if self._required_text(full_obligation, "authority") != "expert": + raise InvalidTransition( + "Builder fidelity artifact task obligation authority is invalid" + ) + obligations.append( + self._json_object( + { + "obligation_id": obligation_id, + "kind": kind, + "source_span_ids": list(source_span_ids), + "normalized_fragment": normalized_fragment, + } + ) + ) + if not obligations: + raise InvalidTransition("Builder fidelity artifact task requires semantic obligations") + return self._json_object( + { + "source_spans": source_spans, + "mathematics": bundle.contract.mathematics.model_dump(mode="json"), + "formalization": { + "task_kind": bundle.contract.task_kind.value, + "declaration_name": bundle.contract.formal.declaration_name, + "namespace": bundle.contract.formal.namespace, + "lean_version": bundle.contract.formal.environment.lean_version, + "mathlib_revision": bundle.contract.formal.environment.mathlib_revision, + "imports_allowlist": list(bundle.contract.formal.imports_allowlist), + "axioms_allowlist": list(bundle.contract.formal.axioms_allowlist), + "rendering_profile": "autolean.full-declaration-exact.v1", + }, + "obligations": obligations, + } + ) + + def _source_claim_span_projection( + self, + bundle: FormalizationTaskBundleV1, + task: JsonObject, + ) -> list[JsonObject]: + """Rebuild private claim payloads from frozen public span bindings. + + The retained artifact supplies a private excerpt only. Its span identifier, locator, and + typed content hash must exactly reproduce the frozen contract, and the excerpt must hash + to that contract hash before either task projection can include it. + """ + + claims = self._required_list(task, "source_spans") + frozen_spans = bundle.contract.source.spans + frozen_spans_by_id = {span.span_id.value: span for span in frozen_spans} + claims_by_id: dict[str, JsonObject] = {} + validated_claims: list[JsonObject] = [] + for value in claims: + if not isinstance(value, dict): + raise InvalidTransition("Builder fidelity artifact task source spans are invalid") + claim = self._json_object(value) + self._require_exact_object_keys( + claim, + _SOURCE_CLAIM_SPAN_FIELDS, + label="Builder fidelity artifact task source span", + ) + span_id = self._required_text(claim, "span_id") + if span_id in claims_by_id: + raise InvalidTransition( + "Builder fidelity artifact task source span identifiers must be unique" + ) + claims_by_id[span_id] = claim + validated_claims.append(claim) + if not claims or set(claims_by_id) != set(frozen_spans_by_id): + raise InvalidTransition( + "Builder fidelity artifact task source spans differ from frozen spans" + ) + + projection: list[JsonObject] = [] + for frozen_claim in validated_claims: + span_id = self._required_text(frozen_claim, "span_id") + frozen_span = frozen_spans_by_id[span_id] + if self._required_text(frozen_claim, "locator") != frozen_span.locator: + raise InvalidTransition( + "Builder fidelity artifact task source span locator differs" + ) + content_hash = self._required_fidelity_digest( + frozen_claim, + "content_hash", + expected_kind=HashKindV1.SOURCE_SPAN, + ) + if content_hash != frozen_span.content_hash: + raise InvalidTransition("Builder fidelity artifact task source span hash differs") + excerpt = self._required_text(frozen_claim, "permitted_excerpt") + if not excerpt.strip() or digest_text(HashKindV1.SOURCE_SPAN, excerpt) != content_hash: + raise InvalidTransition( + "Builder fidelity artifact task source excerpt differs from hash" + ) + if ( + frozen_span.permitted_excerpt is not None + and excerpt != frozen_span.permitted_excerpt + ): + raise InvalidTransition( + "Builder fidelity artifact task source excerpt differs from frozen span" + ) + projection.append( + self._json_object( + { + "span_id": frozen_span.span_id.value, + "locator": frozen_span.locator, + "content_hash": frozen_span.content_hash.model_dump(mode="json"), + "permitted_excerpt": excerpt, + } + ) + ) + + if validated_claims != projection: + raise InvalidTransition( + "Builder fidelity artifact task source spans differ from frozen spans" + ) + return projection + + def _validate_candidate_generation_task_hashes( + self, + payload: JsonObject, + generation_task_hash: DigestV1, + ) -> None: + candidates = self._required_list(payload, "candidates") + if len(candidates) < 2: + raise InvalidTransition("Builder fidelity artifact requires at least two candidates") + for index, value in enumerate(candidates): + if not isinstance(value, dict): + raise InvalidTransition( + f"Builder fidelity artifact candidate {index} must be a JSON object" + ) + candidate = self._json_object(value) + if ( + self._required_fidelity_digest( + candidate, + "generation_task_hash", + expected_kind=HashKindV1.PROMPT, + ) + != generation_task_hash + ): + raise InvalidTransition( + "Builder fidelity artifact candidate " + f"{index} has a different generation task hash" + ) + + @staticmethod + def _require_exact_object_keys( + payload: JsonObject, + expected: frozenset[str], + *, + label: str, + ) -> None: + if set(payload) != expected: + raise InvalidTransition(f"{label} has unexpected or missing fields") + + def _required_fidelity_digest( + self, + payload: JsonObject, + key: str, + *, + expected_kind: HashKindV1, + ) -> DigestV1: + raw = self._required_object(payload, key) + self._require_exact_object_keys( + raw, + _DIGEST_V1_FIELDS, + label=f"Builder fidelity artifact {key}", + ) + try: + digest = DigestV1.model_validate(raw) + except ValueError as error: + raise InvalidTransition( + f"Builder fidelity artifact {key} has an invalid digest" + ) from error + if digest.kind is not expected_kind: + raise InvalidTransition( + f"Builder fidelity artifact {key} has an unexpected digest kind" + ) + return digest + def _verify_verification_attestation( self, binding: TaskBinding, @@ -1389,6 +1689,15 @@ def _required_text(payload: JsonObject, key: str) -> str: raise InvalidTransition(f"corrupt control-plane payload: {key}") return value + @classmethod + def _required_nonblank_fidelity_obligation_text(cls, payload: JsonObject, key: str) -> str: + value = cls._required_text(payload, key) + if not value.strip(): + raise InvalidTransition( + f"Builder fidelity artifact task obligation {key} must not be blank" + ) + return value + @staticmethod def _optional_text(payload: JsonObject, key: str) -> str | None: value = payload.get(key) diff --git a/packages/control_plane/tests/test_service.py b/packages/control_plane/tests/test_service.py index 15d3a64..2db60d6 100644 --- a/packages/control_plane/tests/test_service.py +++ b/packages/control_plane/tests/test_service.py @@ -14,6 +14,8 @@ AlignmentTargetV1, AttestationPurposeV1, ExecutionGraphV1, + FidelityEvidenceArtifactRefV1, + FidelityReportV1, FidelityRiskV1, FormalGraphV1, FormalizationTaskBundleV1, @@ -43,6 +45,8 @@ VerificationReportV1, build_proof_boundary, builder_attestation_payload, + canonical_json_bytes, + digest_bytes, digest_model, digest_text, proof_dependency_manifest_hash, @@ -125,6 +129,7 @@ def _bundle( signer: HmacAttestationSignerV1 | None = None, nonce: str | None = None, bundle_key: str = "bundle", + additional_source_spans: tuple[SourceSpanV1, ...] = (), ) -> FormalizationTaskBundleV1: source_id = _id("source") span = SourceSpanV1( @@ -140,7 +145,7 @@ def _bundle( version="1", locator="fixture://source", content_hash=digest_text(HashKindV1.SOURCE_BYTES, "fixture source"), - spans=(span,), + spans=(span, *additional_source_spans), ) rights = RightsRecordV1( rights_id=_id("rights"), @@ -264,10 +269,171 @@ def _rebundle( return unsigned.model_copy(update={"builder_attestation": attestation}) +def _fidelity_payload(draft: StatementContractV1) -> dict[str, object]: + source_span = draft.source.spans[0] + source_claims = [ + { + "span_id": span.span_id.value, + "locator": span.locator, + "content_hash": span.content_hash.model_dump(mode="json"), + "permitted_excerpt": span.permitted_excerpt, + } + for span in draft.source.spans + ] + obligation = { + "obligation_id": "conclusion", + "kind": "conclusion", + "source_span_ids": [source_span.span_id.value], + "normalized_fragment": "n equals n", + } + task = { + "contract_id": draft.contract_id.value, + "revision": draft.revision, + "draft_contract_hash": draft.semantic_hash().model_dump(mode="json"), + "source_hash": draft.source.content_hash.model_dump(mode="json"), + "source_spans": source_claims, + "informal_statement": draft.mathematics.informal_statement, + "normalized_statement": draft.mathematics.normalized_statement, + "normalized_statement_sha256": hashlib.sha256( + draft.mathematics.normalized_statement.encode("utf-8") + ).hexdigest(), + "selected_lean_statement": draft.formal.lean_statement_source, + "selected_statement_hash": draft.formal.statement_source_hash.model_dump(mode="json"), + "obligations": [ + { + **obligation, + "description": "The source states reflexive equality.", + "lean_fragment": "n = n", + "authority": "expert", + } + ], + } + generation_task = { + "source_spans": source_claims, + "mathematics": draft.mathematics.model_dump(mode="json"), + "formalization": { + "task_kind": draft.task_kind.value, + "declaration_name": draft.formal.declaration_name, + "namespace": draft.formal.namespace, + "lean_version": draft.formal.environment.lean_version, + "mathlib_revision": draft.formal.environment.mathlib_revision, + "imports_allowlist": list(draft.formal.imports_allowlist), + "axioms_allowlist": list(draft.formal.axioms_allowlist), + "rendering_profile": "autolean.full-declaration-exact.v1", + }, + "obligations": [obligation], + } + generation_task_hash = digest_bytes( + HashKindV1.PROMPT, + canonical_json_bytes(generation_task), + ).model_dump(mode="json") + candidates = [ + { + "candidate_id": f"candidate-{suffix}", + "actor_id": f"translator-{suffix}", + "independence_group": f"group-{suffix}", + "contract_id": task["contract_id"], + "revision": task["revision"], + "draft_contract_hash": task["draft_contract_hash"], + "source_hash": task["source_hash"], + "normalized_statement_sha256": task["normalized_statement_sha256"], + "generation_task_hash": generation_task_hash, + "lean_statement_source": task["selected_lean_statement"], + "reverse_rendering": task["normalized_statement"], + "covered_obligation_ids": [obligation["obligation_id"]], + } + for suffix in ("a", "b") + ] + return { + "schema_version": "autolean.builder-fidelity-evidence.v1", + "task": task, + "generation_task": generation_task, + "generation_task_hash": generation_task_hash, + "candidates": candidates, + "mutation_agent_id": "mutation-agent", + "mutation_probes": [], + "review": {"review_id": "fixture-review"}, + "automatic_checks": [], + "additional_signoffs": [], + } + + +def _reviewed_bundle( + artifact_store: ArtifactStore, + *, + fidelity_payload: dict[str, object] | None = None, + base: FormalizationTaskBundleV1 | None = None, +) -> tuple[FormalizationTaskBundleV1, dict[str, object]]: + base = _bundle() if base is None else base + assert base.contract.freeze is not None + draft = base.contract.model_copy(update={"status": StatementStatusV1.DRAFT, "freeze": None}) + payload = _fidelity_payload(draft) if fidelity_payload is None else fidelity_payload + raw = canonical_json_bytes(payload) + artifact = artifact_store.put_bytes(raw) + evidence_hash = digest_bytes(HashKindV1.FREEZE_EVIDENCE, raw) + reviewed_draft = draft.model_copy( + update={ + "fidelity": FidelityReportV1( + report_id=_id("fidelity-report"), + evidence_hash=evidence_hash, + risk_level=draft.policy.fidelity_risk, + ) + } + ) + frozen = reviewed_draft.model_copy( + update={ + "status": StatementStatusV1.FROZEN, + "freeze": FreezeRecordV1( + contract_hash=reviewed_draft.semantic_hash(), + source_hash=reviewed_draft.source.content_hash, + source_preparation_id=base.contract.freeze.source_preparation_id, + source_preparation_hash=base.contract.freeze.source_preparation_hash, + statement_source_hash=reviewed_draft.formal.statement_source_hash, + elaborated_type_hash=reviewed_draft.formal.elaborated_type_hash, + frozen_by="fixture", + ), + } + ) + unsigned = FormalizationTaskBundleV1( + bundle_id=base.bundle_id, + contract=frozen, + graphs=base.graphs, + graph_snapshot_hash=base.graph_snapshot_hash, + proof_boundary=build_proof_boundary(frozen), + fidelity_evidence=FidelityEvidenceArtifactRefV1( + digest=evidence_hash, + size=artifact.size, + ), + ) + attestation = _builder_signer().issue( + purpose=AttestationPurposeV1.BUILDER_FREEZE, + payload=builder_attestation_payload(unsigned), + evidence_identity=artifact.uri, + ttl_seconds=3600, + ) + return unsigned.model_copy(update={"builder_attestation": attestation}), payload + + +def _refresh_generation_task_hash(payload: dict[str, object]) -> None: + generation_task = payload["generation_task"] + assert isinstance(generation_task, dict) + generation_task_hash = digest_bytes( + HashKindV1.PROMPT, + canonical_json_bytes(generation_task), + ).model_dump(mode="json") + payload["generation_task_hash"] = generation_task_hash + candidates = payload["candidates"] + assert isinstance(candidates, list) + for candidate in candidates: + assert isinstance(candidate, dict) + candidate["generation_task_hash"] = generation_task_hash + + def _plane( tmp_path: Path, *, clock: Callable[[], datetime] | None = None, + allow_test_only_unreviewed_bundles: bool = True, ) -> ControlPlane: database = tmp_path / "control.db" return ControlPlane( @@ -276,7 +442,7 @@ def _plane( artifacts=ArtifactStore(tmp_path / "artifacts"), attestation_verifier=_attestation_verifier(clock), allow_test_only_direct_verifier_attestations=True, - allow_test_only_unreviewed_bundles=True, + allow_test_only_unreviewed_bundles=allow_test_only_unreviewed_bundles, ) @@ -401,6 +567,380 @@ def test_control_plane_rejects_legacy_bundle_without_source_preparation( ) +def test_registration_accepts_a_complete_v1_fidelity_artifact(tmp_path: Path) -> None: + plane = _plane(tmp_path) + bundle, payload = _reviewed_bundle(plane.artifacts) + + expected_generation_hash = digest_bytes( + HashKindV1.PROMPT, + canonical_json_bytes(payload["generation_task"]), + ) + assert not canonical_json_bytes(payload["generation_task"]).endswith(b"\n") + assert payload["generation_task_hash"] == expected_generation_hash.model_dump(mode="json") + + binding = plane.register_bundle(bundle, idempotency_key="reviewed-v1") + + assert binding.fidelity_evidence_artifact is not None + assert binding.fidelity_evidence_artifact.digest == bundle.fidelity_evidence.digest.value + + +def test_registration_rejects_legacy_reviewed_v1_without_generation_task( + tmp_path: Path, +) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + legacy_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(legacy_payload, dict) + legacy_payload.pop("generation_task") + legacy_payload.pop("generation_task_hash") + legacy, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=legacy_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="unexpected or missing fields"): + plane.register_bundle(legacy, idempotency_key="legacy-reviewed-v1") + + +def test_registration_rejects_wrong_top_level_generation_task_hash(tmp_path: Path) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + top_hash = tampered_payload["generation_task_hash"] + assert isinstance(top_hash, dict) + top_hash["value"] = "0" * 64 + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="does not match generation_task_hash"): + plane.register_bundle(tampered, idempotency_key="wrong-top-level-generation-hash") + + +def test_registration_rejects_wrong_candidate_generation_task_hash(tmp_path: Path) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + candidates = tampered_payload["candidates"] + assert isinstance(candidates, list) + first_candidate = candidates[0] + assert isinstance(first_candidate, dict) + candidate_hash = first_candidate["generation_task_hash"] + assert isinstance(candidate_hash, dict) + candidate_hash["value"] = "1" * 64 + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="candidate 0 has a different generation task hash"): + plane.register_bundle(tampered, idempotency_key="wrong-candidate-generation-hash") + + +def test_registration_rejects_generation_projection_content_drift(tmp_path: Path) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + generation_task = tampered_payload["generation_task"] + assert isinstance(generation_task, dict) + formalization = generation_task["formalization"] + assert isinstance(formalization, dict) + formalization["declaration_name"] = "different_fixture" + generation_task_hash = digest_bytes( + HashKindV1.PROMPT, + canonical_json_bytes(generation_task), + ).model_dump(mode="json") + tampered_payload["generation_task_hash"] = generation_task_hash + candidates = tampered_payload["candidates"] + assert isinstance(candidates, list) + for candidate in candidates: + assert isinstance(candidate, dict) + candidate["generation_task_hash"] = generation_task_hash + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="differs from the Builder projection"): + plane.register_bundle(tampered, idempotency_key="generation-projection-drift") + + +def test_registration_rejects_extra_generation_task_fields(tmp_path: Path) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + generation_task = tampered_payload["generation_task"] + assert isinstance(generation_task, dict) + generation_task["selected_formal_answer"] = "forbidden extra" + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="differs from the Builder projection"): + plane.register_bundle(tampered, idempotency_key="extra-generation-field") + + +@pytest.mark.parametrize("allow_test_only_unreviewed_bundles", (False, True)) +def test_registration_rejects_selected_lean_source_as_generation_fragment( + tmp_path: Path, + allow_test_only_unreviewed_bundles: bool, +) -> None: + plane = _plane( + tmp_path, + allow_test_only_unreviewed_bundles=allow_test_only_unreviewed_bundles, + ) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + generation_task = tampered_payload["generation_task"] + assert isinstance(task, dict) and isinstance(generation_task, dict) + task_obligations = task["obligations"] + generation_obligations = generation_task["obligations"] + assert isinstance(task_obligations, list) and isinstance(generation_obligations, list) + assert isinstance(task_obligations[0], dict) and isinstance(generation_obligations[0], dict) + task_obligations[0]["normalized_fragment"] = base.contract.formal.lean_statement_source + generation_obligations[0]["normalized_fragment"] = base.contract.formal.lean_statement_source + _refresh_generation_task_hash(tampered_payload) + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="fragment is absent"): + plane.register_bundle(tampered, idempotency_key="selected-lean-fragment") + + +def test_registration_rejects_task_and_generation_source_span_drift(tmp_path: Path) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + generation_task = tampered_payload["generation_task"] + assert isinstance(task, dict) and isinstance(generation_task, dict) + task_spans = task["source_spans"] + generation_spans = generation_task["source_spans"] + assert isinstance(task_spans, list) and isinstance(generation_spans, list) + assert isinstance(task_spans[0], dict) and isinstance(generation_spans[0], dict) + task_spans[0]["locator"] = "attacker://different-source" + generation_spans[0]["locator"] = "attacker://different-source" + _refresh_generation_task_hash(tampered_payload) + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="source span locator differs"): + plane.register_bundle(tampered, idempotency_key="source-span-drift") + + +def test_registration_accepts_valid_reordered_private_source_claims(tmp_path: Path) -> None: + second_span = SourceSpanV1( + span_id=_id("second-span"), + locator="source:2", + content_hash=digest_text(HashKindV1.SOURCE_SPAN, "m equals m"), + permitted_excerpt="m equals m", + ) + plane = _plane(tmp_path) + base = _bundle(additional_source_spans=(second_span,)) + base, payload = _reviewed_bundle(plane.artifacts, base=base) + reordered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(reordered_payload, dict) + task = reordered_payload["task"] + generation_task = reordered_payload["generation_task"] + assert isinstance(task, dict) and isinstance(generation_task, dict) + task_spans = task["source_spans"] + generation_spans = generation_task["source_spans"] + assert isinstance(task_spans, list) and isinstance(generation_spans, list) + task_spans.reverse() + generation_spans.reverse() + _refresh_generation_task_hash(reordered_payload) + reordered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=reordered_payload, + base=base, + ) + + binding = plane.register_bundle(reordered, idempotency_key="reordered-private-source-claims") + + assert binding.fidelity_evidence_artifact is not None + + +@pytest.mark.parametrize("mode", ("unknown", "duplicate")) +def test_registration_rejects_unknown_or_duplicate_task_source_span_ids( + tmp_path: Path, + mode: str, +) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + generation_task = tampered_payload["generation_task"] + assert isinstance(task, dict) and isinstance(generation_task, dict) + task_spans = task["source_spans"] + generation_spans = generation_task["source_spans"] + assert isinstance(task_spans, list) and isinstance(generation_spans, list) + assert isinstance(task_spans[0], dict) and isinstance(generation_spans[0], dict) + if mode == "unknown": + task_spans[0]["span_id"] = _id("unknown-span").value + generation_spans[0]["span_id"] = _id("unknown-span").value + else: + task_spans.append(dict(task_spans[0])) + generation_spans.append(dict(generation_spans[0])) + _refresh_generation_task_hash(tampered_payload) + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="source span"): + plane.register_bundle(tampered, idempotency_key=f"{mode}-source-span") + + +def test_registration_rejects_fragment_absent_from_frozen_normalized_statement( + tmp_path: Path, +) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + generation_task = tampered_payload["generation_task"] + assert isinstance(task, dict) and isinstance(generation_task, dict) + task_obligations = task["obligations"] + generation_obligations = generation_task["obligations"] + assert isinstance(task_obligations, list) and isinstance(generation_obligations, list) + assert isinstance(task_obligations[0], dict) and isinstance(generation_obligations[0], dict) + task_obligations[0]["normalized_fragment"] = "unbound normalized fragment" + generation_obligations[0]["normalized_fragment"] = "unbound normalized fragment" + _refresh_generation_task_hash(tampered_payload) + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="fragment is absent"): + plane.register_bundle(tampered, idempotency_key="unbound-normalized-fragment") + + +def test_registration_rejects_duplicate_task_obligation_source_span_ids(tmp_path: Path) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + assert isinstance(task, dict) + obligations = task["obligations"] + assert isinstance(obligations, list) and isinstance(obligations[0], dict) + source_span_ids = obligations[0]["source_span_ids"] + assert isinstance(source_span_ids, list) and isinstance(source_span_ids[0], str) + source_span_ids.append(source_span_ids[0]) + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="source spans are invalid"): + plane.register_bundle(tampered, idempotency_key="duplicate-obligation-source-span") + + +@pytest.mark.parametrize( + "field", + ("obligation_id", "description", "normalized_fragment", "lean_fragment"), +) +def test_registration_rejects_blank_task_obligation_text(tmp_path: Path, field: str) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + assert isinstance(task, dict) + obligations = task["obligations"] + assert isinstance(obligations, list) and isinstance(obligations[0], dict) + obligations[0][field] = " " + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match=f"{field} must not be blank"): + plane.register_bundle(tampered, idempotency_key=f"blank-obligation-{field}") + + +@pytest.mark.parametrize("mode", ("extra", "missing")) +def test_registration_rejects_task_obligation_extra_or_missing_fields( + tmp_path: Path, + mode: str, +) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + task = tampered_payload["task"] + assert isinstance(task, dict) + obligations = task["obligations"] + assert isinstance(obligations, list) and isinstance(obligations[0], dict) + if mode == "extra": + obligations[0]["selected_formal_answer"] = "forbidden" + else: + obligations[0].pop("description") + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="task obligation"): + plane.register_bundle(tampered, idempotency_key=f"{mode}-obligation-field") + + +@pytest.mark.parametrize("mode", ("extra", "missing")) +def test_registration_rejects_generation_obligation_extra_or_missing_fields( + tmp_path: Path, + mode: str, +) -> None: + plane = _plane(tmp_path) + base, payload = _reviewed_bundle(plane.artifacts) + tampered_payload = json.loads(canonical_json_bytes(payload)) + assert isinstance(tampered_payload, dict) + generation_task = tampered_payload["generation_task"] + assert isinstance(generation_task, dict) + obligations = generation_task["obligations"] + assert isinstance(obligations, list) and isinstance(obligations[0], dict) + if mode == "extra": + obligations[0]["lean_fragment"] = "forbidden" + else: + obligations[0].pop("normalized_fragment") + _refresh_generation_task_hash(tampered_payload) + tampered, _ = _reviewed_bundle( + plane.artifacts, + fidelity_payload=tampered_payload, + base=base, + ) + + with pytest.raises(InvalidTransition, match="differs from the Builder projection"): + plane.register_bundle(tampered, idempotency_key=f"{mode}-generation-obligation-field") + + def test_protocol_accepts_only_evidence_for_the_registered_frozen_contract(tmp_path: Path) -> None: plane = _plane(tmp_path) bundle = _bundle()