feat(detection-core): validate differential proof attachment

.kiro/specs/detection-core/tasks.md

@@ -202,7 +202,7 @@
   - _Requirements: 2.2, 2.3, 7.1, 7.2, 7.3, 7.4, 14.4_
   - _Boundary: Verdict and Proof_
   - _Depends: 2.5, 3.1_
-- [ ] 7.4 Validate differential proof attachment without enabling rollout
+- [x] 7.4 Validate differential proof attachment without enabling rollout
   - Re-derive reviewed producer, sandbox, oracle, affected/safe side, control, witness, and verification identities.
   - Resolve all proof artifacts under the supplied producer, reviewer, store, and byte-limit trust policy.
   - Upgrade basis only for demonstrated affected-only behavior tied to the same family, packet, location, and change/control.

src/openultrasast/core/_proof_apply.py

@@ -0,0 +1,142 @@
+"""Differential-proof application against one exact defect claim."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+
+from openultrasast.core import canonical, provenance
+from openultrasast.core._proof_attestation import proof_attestation_bytes, validate_proof_artifacts
+from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, derive_packet_digest
+from openultrasast.core.proof import (
+    DifferentialProofRecord,
+    ProofAttachment,
+    ProofOutcome,
+    ProofTrustPolicy,
+    ProofValidationError,
+    ProofVerificationReceipt,
+    derive_producer_identity_digest,
+)
+from openultrasast.core.verdict_v2 import DefectClaim
+
+
+def _validate_claim_binding(claim: DefectClaim, attachment: ProofAttachment) -> None:
+    try:
+        claim.__post_init__()
+        attachment.__post_init__()
+        attachment.proof.__post_init__()
+        attachment.proof.producer.__post_init__()
+        attachment.proof.affected.__post_init__()
+        attachment.proof.safe.__post_init__()
+        attachment.proof.control.__post_init__()
+        attachment.verification.__post_init__()
+    except (TypeError, ValueError) as exc:
+        raise ProofValidationError(f"proof attachment identity validation failed: {exc}") from exc
+    record = claim.record
+    if record.decision.basis is EvidenceBasis.EXECUTION_BACKED:
+        raise ProofValidationError("an execution-backed claim cannot replace its proof attachment")
+    if attachment.family_id != record.packet.family_id or attachment.packet_digest != derive_packet_digest(record.packet):
+        raise ProofValidationError("proof family or packet does not match the defect claim")
+    if attachment.location != claim.location:
+        raise ProofValidationError("proof attachment does not match the exact claim location")
+
+
+def _validate_change_and_outcome(record: EvidenceRecord, proof: DifferentialProofRecord) -> None:
+    change = record.packet.provenance.change_input_digest
+    if proof.affected.input_digest != change or proof.control.affected_change_input_digest != change:
+        raise ProofValidationError("affected proof input does not match the evidence change input")
+    if proof.safe.input_digest != proof.control.safe_input_digest:
+        raise ProofValidationError("safe proof input does not match the reviewed control")
+    if proof.affected.input_digest == proof.safe.input_digest:
+        raise ProofValidationError("affected and safe proof inputs must be distinct")
+    if proof.outcome is not ProofOutcome.DEMONSTRATED or not proof.affected.observed or proof.safe.observed:
+        raise ProofValidationError("only demonstrated affected-only behavior can upgrade evidence basis")
+
+
+def _validate_trust(
+    proof: DifferentialProofRecord,
+    verification: ProofVerificationReceipt,
+    policy: ProofTrustPolicy,
+) -> str:
+    try:
+        policy.__post_init__()
+    except (TypeError, ValueError) as exc:
+        raise ProofValidationError(f"proof policy validation failed: {exc}") from exc
+    producer_digest = derive_producer_identity_digest(proof.producer)
+    if producer_digest not in policy.reviewed_producer_digests:
+        raise ProofValidationError("proof producer is not independently reviewed")
+    reviewer = proof.control.reviewer_identity
+    if reviewer not in policy.reviewed_control_reviewer_digests:
+        raise ProofValidationError("control reviewer is not admitted by the proof policy")
+    producer_ids = (proof.producer.producer_id, canonical.content_digest(proof.producer.producer_id), producer_digest)
+    if reviewer in producer_ids:
+        raise ProofValidationError("proof producer cannot supply a self-asserted safe control")
+    expected = (
+        producer_digest,
+        proof.record_digest,
+        proof.affected.result_digest,
+        proof.safe.result_digest,
+        proof.witness.content_digest,
+    )
+    actual = (
+        verification.producer_identity_digest,
+        verification.proof_record_digest,
+        verification.affected_result_digest,
+        verification.safe_result_digest,
+        verification.witness_content_digest,
+    )
+    if actual != expected:
+        raise ProofValidationError("proof verification identities do not match the proof record")
+    return producer_digest
+
+
+def _resolve_artifacts(
+    proof: DifferentialProofRecord,
+    verification: ProofVerificationReceipt,
+    policy: ProofTrustPolicy,
+    resolver: canonical.ContentResolver,
+) -> None:
+    try:
+        expected = proof_attestation_bytes(
+            verification.producer_identity_digest,
+            verification.proof_record_digest,
+            verification.affected_result_digest,
+            verification.safe_result_digest,
+            verification.witness_content_digest,
+        )
+        validate_proof_artifacts(
+            proof.control.control_manifest,
+            proof.witness,
+            verification.attestation,
+            expected,
+            resolver,
+            policy.trusted_store_identities,
+            policy.maximum_artifact_bytes,
+        )
+    except (canonical.ArtifactResolutionError, canonical.ArtifactValidationError, ValueError) as exc:
+        raise ProofValidationError(f"proof artifact validation failed: {exc}") from exc
+
+
+def attach_differential_proof(
+    claim: DefectClaim,
+    attachment: ProofAttachment,
+    policy: ProofTrustPolicy,
+    resolver: canonical.ContentResolver,
+) -> DefectClaim:
+    if not isinstance(claim, DefectClaim) or not isinstance(attachment, ProofAttachment):
+        raise ProofValidationError("proof application requires a DefectClaim and ProofAttachment")
+    if not isinstance(policy, ProofTrustPolicy):
+        raise ProofValidationError("proof application requires a ProofTrustPolicy")
+    _validate_claim_binding(claim, attachment)
+    _validate_change_and_outcome(claim.record, attachment.proof)
+    producer_digest = _validate_trust(attachment.proof, attachment.verification, policy)
+    _resolve_artifacts(attachment.proof, attachment.verification, policy, resolver)
+    decision = replace(
+        claim.record.decision,
+        basis=EvidenceBasis.EXECUTION_BACKED,
+        provenance=provenance.DecisionProvenance(None, None, None, attachment.proof.record_digest, producer_digest, None),
+    )
+    record = EvidenceRecord(claim.record.packet, decision, "")
+    return DefectClaim(record, EvidenceBasis.EXECUTION_BACKED, claim.location, claim.change_anchor)
+
+
+__all__ = ["attach_differential_proof"]

src/openultrasast/core/_proof_attestation.py

@@ -0,0 +1,56 @@
+"""Canonical proof-attestation bytes and bounded artifact resolution."""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+
+from openultrasast.core import canonical
+
+PROOF_ATTESTATION_SCHEMA = "openultrasast.detection-core.proof-attestation.v2"
+
+
+def proof_attestation_bytes(
+    producer_identity_digest: str,
+    proof_record_digest: str,
+    affected_result_digest: str,
+    safe_result_digest: str,
+    witness_content_digest: str,
+) -> bytes:
+    payload: dict[str, canonical.JsonValue] = {
+        "schema_version": PROOF_ATTESTATION_SCHEMA,
+        "producer_identity_digest": producer_identity_digest,
+        "proof_record_digest": proof_record_digest,
+        "affected_result_digest": affected_result_digest,
+        "safe_result_digest": safe_result_digest,
+        "witness_content_digest": witness_content_digest,
+    }
+    return canonical.canonical_record_bytes(payload, known_schema_versions=(PROOF_ATTESTATION_SCHEMA,))
+
+
+def validate_proof_artifacts(
+    control: canonical.ContentRef,
+    witness: canonical.ContentRef,
+    attestation: canonical.ContentRef,
+    expected_attestation: bytes,
+    resolver: canonical.ContentResolver,
+    trusted_stores: Collection[str],
+    maximum_bytes: int,
+) -> None:
+    for reference in (control, witness):
+        canonical.resolve_content(
+            reference,
+            resolver,
+            trusted_store_identities=trusted_stores,
+            maximum_bytes=maximum_bytes,
+        )
+    resolved = canonical.resolve_content(
+        attestation,
+        resolver,
+        trusted_store_identities=trusted_stores,
+        maximum_bytes=maximum_bytes,
+    )
+    if resolved != expected_attestation:
+        raise canonical.ArtifactValidationError("proof attestation does not bind the exact verification identities")
+
+
+__all__ = ["PROOF_ATTESTATION_SCHEMA", "proof_attestation_bytes", "validate_proof_artifacts"]

src/openultrasast/core/proof.py

@@ -0,0 +1,345 @@
+"""Canonical differential-proof validation for execution-backed evidence."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from enum import Enum
+
+from openultrasast.core import canonical
+from openultrasast.core.semantics import SourceLocation
+from openultrasast.core.verdict_v2 import DefectClaim
+
+DIFFERENTIAL_PROOF_SCHEMA = "openultrasast.detection-core.differential-proof.v2"
+_DIGEST = re.compile(r"[0-9a-f]{64}")
+
+
+class ProofValidationError(ValueError):
+    """A proof attachment cannot reproduce execution-backed evidence."""
+
+
+class ProofOutcome(Enum):
+    DEMONSTRATED = "demonstrated"
+    INCONCLUSIVE = "inconclusive"
+    FAILED = "failed"
+
+
+def _digest(value: object, field: str) -> None:
+    if not isinstance(value, str) or _DIGEST.fullmatch(value) is None:
+        raise ProofValidationError(f"{field} must be a full lowercase SHA-256 digest")
+
+
+def _text(value: object, field: str) -> None:
+    if not isinstance(value, str) or not value:
+        raise ProofValidationError(f"{field} must be nonempty text")
+
+
+def _positive(value: object, field: str) -> None:
+    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+        raise ProofValidationError(f"{field} must be a positive integer")
+
+
+def _ref_value(value: canonical.ContentRef) -> canonical.JsonValue:
+    return {
+        "artifact_id": value.artifact_id,
+        "content_digest": value.content_digest,
+        "media_type": value.media_type,
+        "byte_length": value.byte_length,
+        "retention_policy_digest": value.retention_policy_digest,
+        "storage_receipt_digest": value.storage_receipt_digest,
+    }
+
+
+def _location_value(value: SourceLocation) -> canonical.JsonValue:
+    return {"path": value.path, "line": value.line, "column": value.column, "symbol": value.symbol}
+
+
+@dataclass(frozen=True)
+class ProofProducerIdentity:
+    producer_id: str
+    verifier_revision: str
+    sandbox_profile_digest: str
+    oracle_contract_digest: str
+    review_digest: str
+
+    def __post_init__(self) -> None:
+        _text(self.producer_id, "producer_id")
+        _text(self.verifier_revision, "verifier_revision")
+        for value, field in (
+            (self.sandbox_profile_digest, "sandbox_profile_digest"),
+            (self.oracle_contract_digest, "oracle_contract_digest"),
+            (self.review_digest, "review_digest"),
+        ):
+            _digest(value, field)
+
+
+def _producer_value(value: ProofProducerIdentity) -> canonical.JsonValue:
+    return {
+        "producer_id": value.producer_id,
+        "verifier_revision": value.verifier_revision,
+        "sandbox_profile_digest": value.sandbox_profile_digest,
+        "oracle_contract_digest": value.oracle_contract_digest,
+        "review_digest": value.review_digest,
+    }
+
+
+def derive_producer_identity_digest(value: ProofProducerIdentity) -> str:
+    value.__post_init__()
+    return canonical.content_digest(_producer_value(value))
+
+
+@dataclass(frozen=True)
+class ProofSideResult:
+    input_digest: str
+    observed: bool
+    observation_digest: str
+    run_count: int
+    result_digest: str
+
+    def __post_init__(self) -> None:
+        _digest(self.input_digest, "proof side input_digest")
+        if not isinstance(self.observed, bool):
+            raise ProofValidationError("proof side observed must be boolean")
+        _digest(self.observation_digest, "proof side observation_digest")
+        _positive(self.run_count, "proof side run_count")
+        expected = derive_proof_side_result_digest(self)
+        if not self.result_digest:
+            object.__setattr__(self, "result_digest", expected)
+        elif self.result_digest != expected:
+            raise ProofValidationError("proof side result digest cannot be re-derived")
+
+
+def _side_value(value: ProofSideResult) -> dict[str, canonical.JsonValue]:
+    return {
+        "input_digest": value.input_digest,
+        "observed": value.observed,
+        "observation_digest": value.observation_digest,
+        "run_count": value.run_count,
+    }
+
+
+def derive_proof_side_result_digest(value: ProofSideResult) -> str:
+    return canonical.content_digest(_side_value(value))
+
+
+@dataclass(frozen=True)
+class ControlBinding:
+    affected_change_input_digest: str
+    safe_input_digest: str
+    control_manifest: canonical.ContentRef
+    reviewer_identity: str
+    binding_digest: str
+
+    def __post_init__(self) -> None:
+        _digest(self.affected_change_input_digest, "affected_change_input_digest")
+        _digest(self.safe_input_digest, "safe_input_digest")
+        if not isinstance(self.control_manifest, canonical.ContentRef):
+            raise ProofValidationError("control_manifest must be a ContentRef")
+        _digest(self.reviewer_identity, "control reviewer_identity")
+        expected = derive_control_binding_digest(self)
+        if not self.binding_digest:
+            object.__setattr__(self, "binding_digest", expected)
+        elif self.binding_digest != expected:
+            raise ProofValidationError("control binding digest cannot be re-derived")
+
+
+def _control_value(value: ControlBinding) -> dict[str, canonical.JsonValue]:
+    return {
+        "affected_change_input_digest": value.affected_change_input_digest,
+        "safe_input_digest": value.safe_input_digest,
+        "control_manifest": _ref_value(value.control_manifest),
+        "reviewer_identity": value.reviewer_identity,
+    }
+
+
+def derive_control_binding_digest(value: ControlBinding) -> str:
+    return canonical.content_digest(_control_value(value))
+
+
+@dataclass(frozen=True)
+class DifferentialProofRecord:
+    schema_version: str
+    producer: ProofProducerIdentity
+    affected: ProofSideResult
+    safe: ProofSideResult
+    control: ControlBinding
+    witness: canonical.ContentRef
+    outcome: ProofOutcome
+    record_digest: str
+
+    def __post_init__(self) -> None:
+        if self.schema_version != DIFFERENTIAL_PROOF_SCHEMA:
+            raise ProofValidationError("unknown differential proof schema")
+        if not isinstance(self.producer, ProofProducerIdentity):
+            raise ProofValidationError("producer must use ProofProducerIdentity")
+        if not isinstance(self.affected, ProofSideResult) or not isinstance(self.safe, ProofSideResult):
+            raise ProofValidationError("proof sides must use ProofSideResult")
+        if not isinstance(self.control, ControlBinding) or not isinstance(self.witness, canonical.ContentRef):
+            raise ProofValidationError("proof control and witness have invalid types")
+        if not isinstance(self.outcome, ProofOutcome):
+            raise ProofValidationError("proof outcome is invalid")
+        expected = derive_differential_proof_record_digest(self)
+        if not self.record_digest:
+            object.__setattr__(self, "record_digest", expected)
+        elif self.record_digest != expected:
+            raise ProofValidationError("differential proof digest cannot be re-derived")
+
+
+def _proof_value(value: DifferentialProofRecord) -> canonical.JsonValue:
+    return {
+        "schema_version": value.schema_version,
+        "producer": _producer_value(value.producer),
+        "affected": {**_side_value(value.affected), "result_digest": value.affected.result_digest},
+        "safe": {**_side_value(value.safe), "result_digest": value.safe.result_digest},
+        "control": {**_control_value(value.control), "binding_digest": value.control.binding_digest},
+        "witness": _ref_value(value.witness),
+        "outcome": value.outcome.value,
+    }
+
+
+def derive_differential_proof_record_digest(value: DifferentialProofRecord) -> str:
+    return canonical.content_digest(_proof_value(value))
+
+
+@dataclass(frozen=True)
+class ProofVerificationReceipt:
+    producer_identity_digest: str
+    proof_record_digest: str
+    affected_result_digest: str
+    safe_result_digest: str
+    witness_content_digest: str
+    attestation: canonical.ContentRef
+    receipt_digest: str
+
+    def __post_init__(self) -> None:
+        for value, field in (
+            (self.producer_identity_digest, "producer_identity_digest"),
+            (self.proof_record_digest, "proof_record_digest"),
+            (self.affected_result_digest, "affected_result_digest"),
+            (self.safe_result_digest, "safe_result_digest"),
+            (self.witness_content_digest, "witness_content_digest"),
+        ):
+            _digest(value, field)
+        if not isinstance(self.attestation, canonical.ContentRef):
+            raise ProofValidationError("attestation must be a ContentRef")
+        expected = derive_proof_verification_receipt_digest(self)
+        if not self.receipt_digest:
+            object.__setattr__(self, "receipt_digest", expected)
+        elif self.receipt_digest != expected:
+            raise ProofValidationError("proof verification receipt digest cannot be re-derived")
+
+
+def _verification_value(value: ProofVerificationReceipt) -> canonical.JsonValue:
+    return {
+        "producer_identity_digest": value.producer_identity_digest,
+        "proof_record_digest": value.proof_record_digest,
+        "affected_result_digest": value.affected_result_digest,
+        "safe_result_digest": value.safe_result_digest,
+        "witness_content_digest": value.witness_content_digest,
+        "attestation": _ref_value(value.attestation),
+    }
+
+
+def derive_proof_verification_receipt_digest(value: ProofVerificationReceipt) -> str:
+    return canonical.content_digest(_verification_value(value))
+
+
+@dataclass(frozen=True)
+class ProofTrustPolicy:
+    reviewed_producer_digests: tuple[str, ...]
+    reviewed_control_reviewer_digests: tuple[str, ...]
+    trusted_store_identities: tuple[str, ...]
+    maximum_artifact_bytes: int
+    policy_digest: str
+
+    def __post_init__(self) -> None:
+        for values, field in (
+            (self.reviewed_producer_digests, "reviewed producer"),
+            (self.reviewed_control_reviewer_digests, "reviewed control reviewer"),
+        ):
+            if not values:
+                raise ProofValidationError(f"{field} identities must be nonempty")
+            for value in values:
+                _digest(value, field)
+            canonical.require_unique_identities(values, kind=field)
+        if not self.trusted_store_identities or not all(isinstance(value, str) and value for value in self.trusted_store_identities):
+            raise ProofValidationError("trusted store identities must be nonempty text")
+        canonical.require_unique_identities(self.trusted_store_identities, kind="trusted store")
+        _positive(self.maximum_artifact_bytes, "maximum_artifact_bytes")
+        expected = derive_proof_trust_policy_digest(self)
+        if not self.policy_digest:
+            object.__setattr__(self, "policy_digest", expected)
+        elif self.policy_digest != expected:
+            raise ProofValidationError("proof trust policy digest cannot be re-derived")
+
+
+def derive_proof_trust_policy_digest(value: ProofTrustPolicy) -> str:
+    payload: canonical.JsonValue = {
+        "reviewed_producer_digests": tuple(sorted(value.reviewed_producer_digests)),
+        "reviewed_control_reviewer_digests": tuple(sorted(value.reviewed_control_reviewer_digests)),
+        "trusted_store_identities": tuple(sorted(value.trusted_store_identities)),
+        "maximum_artifact_bytes": value.maximum_artifact_bytes,
+    }
+    return canonical.content_digest(payload)
+
+
+@dataclass(frozen=True)
+class ProofAttachment:
+    proof_id: str
+    family_id: str
+    packet_digest: str
+    location: SourceLocation
+    proof: DifferentialProofRecord
+    verification: ProofVerificationReceipt
+
+    def __post_init__(self) -> None:
+        _text(self.family_id, "proof family_id")
+        _digest(self.packet_digest, "proof packet_digest")
+        if not isinstance(self.location, SourceLocation):
+            raise ProofValidationError("proof location must be a SourceLocation")
+        if not isinstance(self.proof, DifferentialProofRecord) or not isinstance(self.verification, ProofVerificationReceipt):
+            raise ProofValidationError("proof attachment has invalid proof or verification")
+        expected = derive_proof_attachment_id(self)
+        if not self.proof_id:
+            object.__setattr__(self, "proof_id", expected)
+        elif self.proof_id != expected:
+            raise ProofValidationError("proof attachment identity cannot be re-derived")
+
+
+def derive_proof_attachment_id(value: ProofAttachment) -> str:
+    payload: canonical.JsonValue = {
+        "family_id": value.family_id,
+        "packet_digest": value.packet_digest,
+        "location": _location_value(value.location),
+        "proof_record_digest": value.proof.record_digest,
+        "verification_receipt_digest": value.verification.receipt_digest,
+    }
+    return canonical.content_digest(payload)
+
+
+def attach_differential_proof(
+    claim: DefectClaim,
+    attachment: ProofAttachment,
+    policy: ProofTrustPolicy,
+    resolver: canonical.ContentResolver,
+) -> DefectClaim:
+    """Validate and attach affected-only proof without granting rollout permission."""
+    from openultrasast.core._proof_apply import attach_differential_proof as apply_proof
+
+    return apply_proof(claim, attachment, policy, resolver)
+
+
+__all__ = [
+    "ControlBinding",
+    "DIFFERENTIAL_PROOF_SCHEMA",
+    "DifferentialProofRecord",
+    "ProofAttachment",
+    "ProofOutcome",
+    "ProofProducerIdentity",
+    "ProofSideResult",
+    "ProofTrustPolicy",
+    "ProofValidationError",
+    "ProofVerificationReceipt",
+    "attach_differential_proof",
+    "derive_producer_identity_digest",
+]

tests/test_core_proof.py

@@ -0,0 +1,351 @@
+"""Differential proof attachment and execution-backed basis tests."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+
+import pytest
+from detection_core_fixtures import InMemoryContentResolver, stored_artifact
+
+from openultrasast.core import canonical, provenance
+from openultrasast.core._proof_attestation import proof_attestation_bytes
+from openultrasast.core.evidence import (
+    EVIDENCE_PACKET_SCHEMA,
+    AnalyzedScope,
+    Decision,
+    EvidenceBasis,
+    EvidencePacket,
+    EvidenceRecord,
+    NeutralOutcome,
+    SourceRegion,
+    derive_packet_digest,
+)
+from openultrasast.core.proof import (
+    DIFFERENTIAL_PROOF_SCHEMA,
+    ControlBinding,
+    DifferentialProofRecord,
+    ProofAttachment,
+    ProofOutcome,
+    ProofProducerIdentity,
+    ProofSideResult,
+    ProofTrustPolicy,
+    ProofValidationError,
+    ProofVerificationReceipt,
+    attach_differential_proof,
+    derive_producer_identity_digest,
+)
+from openultrasast.core.semantics import (
+    EvidenceProvenance,
+    FieldKind,
+    FieldValue,
+    ProvenanceKind,
+    RelationKind,
+    RelationStatus,
+    SemanticNode,
+    SourceLocation,
+    make_observation,
+    make_relation,
+)
+from openultrasast.core.verdict_v2 import DefectClaim
+
+
+def _digest(value: str) -> str:
+    return canonical.content_digest(value)
+
+
+def _record() -> EvidenceRecord:
+    location = SourceLocation("src/app.py", 12, 4, "call")
+    relation = make_relation(
+        kind=RelationKind.CALL,
+        role="flow",
+        source=SemanticNode("source", "source", SourceLocation("src/app.py", 10, 2, "source")),
+        target=SemanticNode("target", "target", SourceLocation("src/app.py", 11, 2, "target")),
+        location=location,
+        status=RelationStatus.COMPLETE,
+        conditions=(),
+        reasons=(),
+        provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, _digest("source"), None, None, None),
+    )
+    scope = AnalyzedScope("base", "head", _digest("change"), (SourceRegion("src/app.py", 1, 20),))
+    stamp = provenance.ProvenanceStamp(
+        _digest("analyzer"),
+        _digest("family-contract"),
+        _digest("facts"),
+        _digest("acquisition"),
+        _digest("configuration"),
+        _digest("population"),
+        _digest("change-input"),
+    )
+    packet = EvidencePacket(
+        EVIDENCE_PACKET_SCHEMA,
+        "reference-flow",
+        stamp.family_contract_digest,
+        _digest("object"),
+        "python",
+        scope,
+        (relation,),
+        (),
+        (FieldValue("name", FieldKind.TEXT, "read"),),
+        (),
+        stamp,
+    )
+    decision = Decision(
+        "bad",
+        NeutralOutcome.VIOLATION,
+        EvidenceBasis.DETERMINISTIC,
+        scope,
+        (),
+        derive_packet_digest(packet),
+        provenance.DecisionProvenance(None, None, None, None, None, None),
+    )
+    return EvidenceRecord(packet, decision, "")
+
+
+def _claim(record: EvidenceRecord) -> DefectClaim:
+    return DefectClaim(record, record.decision.basis, record.packet.transitions[0].location, record.packet.scope.regions[0])
+
+
+@dataclass(frozen=True)
+class ProofCase:
+    record: EvidenceRecord
+    attachment: ProofAttachment
+    policy: ProofTrustPolicy
+    resolver: InMemoryContentResolver
+
+
+def _case() -> ProofCase:
+    record = _record()
+    reviewer = _digest("control-reviewer")
+    control_ref, control_resolver = stored_artifact(b'{"control":"safe"}', artifact_id="control")
+    witness_ref, witness_resolver = stored_artifact(b'{"affected":true,"safe":false}', artifact_id="witness")
+    producer = ProofProducerIdentity(
+        "search-with-proof",
+        "verifier-v4",
+        _digest("sandbox-profile"),
+        _digest("oracle-contract"),
+        _digest("producer-review"),
+    )
+    affected = ProofSideResult(record.packet.provenance.change_input_digest, True, _digest("affected-observation"), 2, "")
+    safe_input = _digest("safe-control-input")
+    safe = ProofSideResult(safe_input, False, _digest("safe-observation"), 2, "")
+    control = ControlBinding(affected.input_digest, safe_input, control_ref, reviewer, "")
+    proof = DifferentialProofRecord(
+        DIFFERENTIAL_PROOF_SCHEMA,
+        producer,
+        affected,
+        safe,
+        control,
+        witness_ref,
+        ProofOutcome.DEMONSTRATED,
+        "",
+    )
+    attestation_ref, attestation_resolver = stored_artifact(
+        proof_attestation_bytes(
+            derive_producer_identity_digest(producer),
+            proof.record_digest,
+            affected.result_digest,
+            safe.result_digest,
+            witness_ref.content_digest,
+        ),
+        artifact_id="attestation",
+    )
+    verification = ProofVerificationReceipt(
+        derive_producer_identity_digest(producer),
+        proof.record_digest,
+        affected.result_digest,
+        safe.result_digest,
+        witness_ref.content_digest,
+        attestation_ref,
+        "",
+    )
+    attachment = ProofAttachment(
+        "",
+        record.packet.family_id,
+        derive_packet_digest(record.packet),
+        record.packet.transitions[0].location,
+        proof,
+        verification,
+    )
+    policy = ProofTrustPolicy(
+        (derive_producer_identity_digest(producer),),
+        (reviewer,),
+        ("trusted-store",),
+        4096,
+        "",
+    )
+    artifacts = {
+        **control_resolver.artifacts,
+        **witness_resolver.artifacts,
+        **attestation_resolver.artifacts,
+    }
+    return ProofCase(record, attachment, policy, InMemoryContentResolver(artifacts))
+
+
+def test_demonstrated_affected_only_proof_upgrades_basis_without_rollout_authority() -> None:
+    case = _case()
+    claim = _claim(case.record)
+    derived = attach_differential_proof(claim, case.attachment, case.policy, case.resolver)
+    assert derived is not claim and derived.record.packet is case.record.packet
+    assert derived.location == claim.location and derived.change_anchor == claim.change_anchor
+    assert derived.record.decision.outcome is NeutralOutcome.VIOLATION
+    assert derived.record.decision.family_code == claim.record.decision.family_code
+    assert derived.record.decision.bounded_scope is claim.record.decision.bounded_scope
+    assert derived.record.decision.reasons == claim.record.decision.reasons
+    assert derived.record.decision.basis is EvidenceBasis.EXECUTION_BACKED
+    assert derived.record.decision.provenance.proof_record_digest == case.attachment.proof.record_digest
+    assert derived.record.decision.provenance.proof_producer_digest == derive_producer_identity_digest(case.attachment.proof.producer)
+    assert not hasattr(derived, "rollout_authorized")
+
+
+@pytest.mark.parametrize(
+    ("outcome", "affected", "safe"),
+    (
+        (ProofOutcome.INCONCLUSIVE, True, False),
+        (ProofOutcome.FAILED, True, False),
+        (ProofOutcome.DEMONSTRATED, True, True),
+        (ProofOutcome.DEMONSTRATED, False, False),
+    ),
+)
+def test_only_demonstrated_affected_only_behavior_is_eligible(outcome: ProofOutcome, affected: bool, safe: bool) -> None:
+    case = _case()
+    affected_side = replace(case.attachment.proof.affected, observed=affected, result_digest="")
+    safe_side = replace(case.attachment.proof.safe, observed=safe, result_digest="")
+    proof = replace(case.attachment.proof, affected=affected_side, safe=safe_side, outcome=outcome, record_digest="")
+    verification = replace(
+        case.attachment.verification,
+        proof_record_digest=proof.record_digest,
+        affected_result_digest=affected_side.result_digest,
+        safe_result_digest=safe_side.result_digest,
+        receipt_digest="",
+    )
+    attachment = replace(case.attachment, proof=proof, verification=verification, proof_id="")
+    with pytest.raises(ProofValidationError, match="affected-only"):
+        attach_differential_proof(_claim(case.record), attachment, case.policy, case.resolver)
+
+
+@pytest.mark.parametrize("field", ("family", "packet", "location", "affected", "safe"))
+def test_attachment_must_match_family_packet_location_and_change_control(field: str) -> None:
+    case = _case()
+    attachment = case.attachment
+    if field == "family":
+        attachment = replace(attachment, family_id="other-family", proof_id="")
+    elif field == "packet":
+        attachment = replace(attachment, packet_digest=_digest("other-packet"), proof_id="")
+    elif field == "location":
+        attachment = replace(attachment, location=SourceLocation("src/other.py", 1, None, None), proof_id="")
+    elif field == "affected":
+        side = replace(attachment.proof.affected, input_digest=_digest("other-change"), result_digest="")
+        proof = replace(attachment.proof, affected=side, record_digest="")
+        receipt = replace(
+            attachment.verification,
+            proof_record_digest=proof.record_digest,
+            affected_result_digest=side.result_digest,
+            receipt_digest="",
+        )
+        attachment = replace(attachment, proof=proof, verification=receipt, proof_id="")
+    else:
+        control = replace(attachment.proof.control, safe_input_digest=_digest("other-safe"), binding_digest="")
+        proof = replace(attachment.proof, control=control, record_digest="")
+        receipt = replace(attachment.verification, proof_record_digest=proof.record_digest, receipt_digest="")
+        attachment = replace(attachment, proof=proof, verification=receipt, proof_id="")
+    with pytest.raises(ProofValidationError):
+        attach_differential_proof(_claim(case.record), attachment, case.policy, case.resolver)
+
+
+def test_unreviewed_or_self_asserted_producer_and_control_are_refused() -> None:
+    case = _case()
+    with pytest.raises(ProofValidationError, match="producer"):
+        attach_differential_proof(
+            _claim(case.record),
+            case.attachment,
+            replace(case.policy, reviewed_producer_digests=(_digest("unknown-producer"),), policy_digest=""),
+            case.resolver,
+        )
+    self_asserted = replace(
+        case.attachment.proof.control,
+        reviewer_identity=derive_producer_identity_digest(case.attachment.proof.producer),
+        binding_digest="",
+    )
+    proof = replace(case.attachment.proof, control=self_asserted, record_digest="")
+    receipt = replace(case.attachment.verification, proof_record_digest=proof.record_digest, receipt_digest="")
+    attachment = replace(case.attachment, proof=proof, verification=receipt, proof_id="")
+    policy = replace(
+        case.policy,
+        reviewed_control_reviewer_digests=(self_asserted.reviewer_identity,),
+        policy_digest="",
+    )
+    with pytest.raises(ProofValidationError, match="self-asserted"):
+        attach_differential_proof(_claim(case.record), attachment, policy, case.resolver)
+
+
+def test_all_artifacts_must_resolve_under_store_and_byte_policy() -> None:
+    case = _case()
+    untrusted = replace(case.policy, trusted_store_identities=("other-store",), policy_digest="")
+    with pytest.raises(ProofValidationError, match="artifact"):
+        attach_differential_proof(_claim(case.record), case.attachment, untrusted, case.resolver)
+    too_small = replace(case.policy, maximum_artifact_bytes=1, policy_digest="")
+    with pytest.raises(ProofValidationError, match="artifact"):
+        attach_differential_proof(_claim(case.record), case.attachment, too_small, case.resolver)
+
+
+def test_verification_identity_substitution_is_refused() -> None:
+    case = _case()
+    object.__setattr__(case.attachment.verification, "witness_content_digest", _digest("substituted-witness"))
+    with pytest.raises(ProofValidationError, match="verification"):
+        attach_differential_proof(_claim(case.record), case.attachment, case.policy, case.resolver)
+
+
+def test_rehashed_proof_cannot_reuse_a_trusted_attestation() -> None:
+    case = _case()
+    affected = replace(
+        case.attachment.proof.affected,
+        observation_digest=_digest("substituted-observation"),
+        run_count=case.attachment.proof.affected.run_count + 1,
+        result_digest="",
+    )
+    proof = replace(case.attachment.proof, affected=affected, record_digest="")
+    verification = replace(
+        case.attachment.verification,
+        proof_record_digest=proof.record_digest,
+        affected_result_digest=affected.result_digest,
+        receipt_digest="",
+    )
+    attachment = replace(case.attachment, proof=proof, verification=verification, proof_id="")
+    with pytest.raises(ProofValidationError, match="attestation"):
+        attach_differential_proof(_claim(case.record), attachment, case.policy, case.resolver)
+
+
+def test_proof_at_another_complete_packet_location_cannot_upgrade_claim() -> None:
+    case = _case()
+    location_b = SourceLocation("src/app.py", 18, 4, "other")
+    observation = make_observation(
+        role="context",
+        location=location_b,
+        fields=(),
+        status=RelationStatus.COMPLETE,
+        provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, _digest("other-source"), None, None, None),
+    )
+    packet = replace(case.record.packet, observations=(observation,))
+    decision = replace(case.record.decision, packet_digest=derive_packet_digest(packet))
+    record = EvidenceRecord(packet, decision, "")
+    claim = DefectClaim(record, record.decision.basis, record.packet.transitions[0].location, record.packet.scope.regions[0])
+    attachment = replace(
+        case.attachment,
+        packet_digest=derive_packet_digest(packet),
+        location=location_b,
+        proof_id="",
+    )
+    with pytest.raises(ProofValidationError, match="claim location"):
+        attach_differential_proof(claim, attachment, case.policy, case.resolver)
+
+
+def test_raw_non_claim_record_cannot_gain_execution_backed_basis() -> None:
+    case = _case()
+    decision = replace(
+        case.record.decision,
+        family_code="good",
+        outcome=NeutralOutcome.SATISFIED,
+    )
+    record = EvidenceRecord(case.record.packet, decision, "")
+    with pytest.raises(ProofValidationError, match="DefectClaim"):
+        attach_differential_proof(record, case.attachment, case.policy, case.resolver)  # type: ignore[arg-type]