rabs-cas 2.1.7

//! Prepared-result offer + coordinator-only atomic publication (bead
//! H011; plan §62/§66; invariants I8/I9/I10; risk R50).
//!
//! The worker side ends at [`OfferPreparedActionResult::build`]: a worker
//! harvests its private write set, validates outputs against declarations,
//! uploads objects under candidate pins, and OFFERS. There is deliberately
//! no worker-reachable commit API — a worker never sends a command asking
//! another component to commit on its behalf; the type system enforces it
//! (only [`process_offer`] touches the store).
//!
//! The coordinator side is an ordered admission pipeline over the H009
//! metadata store:
//!
//! 1. authority fence: the offer's full coordinator authority must digest
//!    to the ACTIVE authority, and the generation's
//!    `created_under_authority_digest` must equal that digest (the F033
//!    equality check — one full authority copy, bound by digest);
//! 2. generation fence: the generation exists and is not tombstoned;
//! 3. attempt/lease/worker fences: the exact normalized lease-to-attempt
//!    binding, renewal sequence, boot generation, process incarnation, and
//!    current non-ambiguous worker fence all agree;
//! 4. descriptor reload + byte-compare against the coordinator's own copy;
//! 5. key + epoch validation against the action entry;
//! 6. INDEPENDENT digest recompute: the coordinator recomputes
//!    `semantic_result_digest` and `observable_result_digest` from the
//!    versioned projections and refuses on any declared-digest mismatch;
//! 7. complete object closure: every referenced object must already have
//!    a recorded location;
//! 8. same-key candidates classify through the A018 divergence taxonomy:
//!    idempotent re-offers append evidence; semantic and projection-
//!    completeness divergence quarantine the ACTION, while observable-
//!    only divergence applies the narrower presentation quarantine. Every
//!    divergence preserves the committed row (no in-place patching);
//! 9. commit: publication pointer, serving state, winner evidence row,
//!    AND the durable reachability pin in ONE store transaction (H009's
//!    `commit_publication`); the receipt is emitted only after the store
//!    reports durable success.
//!
//! The crash matrix over this protocol is bead H015.

use rabs_protocol::generation::AttemptAuthority;
use rabs_protocol::raw_bytes::RawBytes;
use rabs_protocol::result_identity::{
    ActionPublicationRecord, AttemptEvidenceBundle, CanonicalActionResultManifest, DigestAlgorithm,
    DivergenceClass, ObjectId, OutputRole, ResultKind, TypedDigest,
};
use sha2::{Digest, Sha256};

use crate::metadata_store::{
    CommitOutcome, DivergenceIncidentRow, ProvisionalAncestorRow, PublicationRow, QuarantineScope,
    RabsMetadataStore, ResultKindTag, StoreError, digest_key,
};
use crate::trust_evidence::DISPOSITION_QUARANTINED;

/// Domain separator for the canonical coordinator-authority digest.
///
/// The digest itself has ONE implementation — F033's
/// [`rabs_key::authority_binding::coordinator_authority_digest`] — and this
/// constant aliases that function's domain so the store-interning label can
/// never drift from the digest domain (G019: two encodings of one fencing
/// identity made cross-layer binding comparisons meaningless).
pub const AUTHORITY_DIGEST_DOMAIN: &str = rabs_key::typed_digest::DOMAIN_COORDINATOR_AUTHORITY;
/// Domain separator for the v1 semantic result projection.
pub const SEMANTIC_PROJECTION_DOMAIN: &str = "rabs.semantic-result-projection.sha256.v1";
/// Domain separator for the v1 observable result projection.
pub const OBSERVABLE_PROJECTION_DOMAIN: &str = "rabs.observable-result-projection.sha256.v1";

/// Serving disposition for observable-only divergence (H026): ordinary
/// replay is disabled (any disposition other than `"servable"` refuses
/// the serving gate), but the narrower tag records that the semantic
/// result itself was reproduced — only presentation/observability
/// differed.
pub const DISPOSITION_PRESENTATION_QUARANTINED: &str = "presentation-quarantined";

/// Pin class preserving the LOSING candidate of a divergence incident
/// (H026; I34): both candidates outlive GC until the incident is
/// resolved.
pub const DIVERGENCE_EVIDENCE_PIN_CLASS: &str = "divergence-evidence";

/// Length-delimited canonical framing (the F034 pattern): every field is
/// `len(u64 be) || bytes`, so no concatenation ambiguity exists.
/// Length-delimited canonical framing (the F034 pattern): every field is
/// `len(u64 be) || bytes`, so no concatenation ambiguity exists — two
/// different field splits can never produce the same digest input.
///
/// ONE implementation for the whole crate, deliberately. This existed as
/// two byte-identical copies (here and in `trust_evidence`), which is the
/// shape that rots quietly: each copy feeds a DIFFERENT domain, so
/// "improving" one changes only that domain's digests and nothing
/// cross-checks the two. A canonical encoding has to be singular to be
/// canonical.
///
/// `rabs-sandbox::env_builder` still carries a third copy for its
/// presented-env digest. It cannot share this one — rabs-sandbox does not
/// depend on rabs-cas and must not — and the natural shared home,
/// `rabs-protocol`, has no dependencies at all and so cannot host a
/// `sha2` helper. Left alone rather than moved, and noted here so the
/// next person does not assume this is the only one.
pub(crate) struct Framing(Sha256);

impl Framing {
    pub(crate) fn new(domain: &str) -> Self {
        let mut hasher = Sha256::new();
        hasher.update((domain.len() as u64).to_be_bytes());
        hasher.update(domain.as_bytes());
        Self(hasher)
    }

    pub(crate) fn field(&mut self, bytes: &[u8]) -> &mut Self {
        self.0.update((bytes.len() as u64).to_be_bytes());
        self.0.update(bytes);
        self
    }

    pub(crate) fn u64(&mut self, v: u64) -> &mut Self {
        self.field(&v.to_be_bytes())
    }

    pub(crate) fn digest_field(&mut self, d: &TypedDigest) -> &mut Self {
        self.field(d.domain.as_bytes());
        self.field(&d.bytes)
    }

    pub(crate) fn finish(self, domain: &'static str) -> TypedDigest {
        TypedDigest {
            algorithm: DigestAlgorithm::Sha256V1,
            domain,
            bytes: self.0.finalize().into(),
        }
    }
}

/// Canonical digest of a FULL coordinator authority value. The generation
/// stores only this digest (risk R117: two independently mutable full
/// copies are forbidden by construction).
///
/// Delegates to F033's single implementation in `rabs-key`; every layer
/// (actor lease admission, offer admission, publication) compares digests
/// produced by this one function (G019).
#[must_use]
pub fn authority_digest(authority: &rabs_protocol::authority::CoordinatorAuthority) -> TypedDigest {
    rabs_key::authority_binding::coordinator_authority_digest(authority)
}

const fn result_kind_tag(kind: ResultKind) -> u64 {
    match kind {
        ResultKind::Success => 0,
        ResultKind::DeterministicFailure => 1,
    }
}

pub(crate) const fn output_role_tag(role: OutputRole) -> u64 {
    match role {
        OutputRole::Materializable => 0,
        OutputRole::DepInfo => 1,
        OutputRole::ProvisionalMetadata => 2,
        OutputRole::BuildScriptMetadata => 3,
        OutputRole::TestSideEffect => 4,
    }
}

/// The v1 semantic result projection: replayable output/exit facts only.
/// Excludes both declared digest fields by construction.
#[must_use]
pub fn semantic_result_digest_v1(manifest: &CanonicalActionResultManifest) -> TypedDigest {
    let mut framing = Framing::new(SEMANTIC_PROJECTION_DOMAIN);
    framing
        .digest_field(&manifest.action_key)
        .digest_field(&manifest.canonical_descriptor_digest)
        .u64(u64::from(manifest.key_epoch))
        .u64(u64::from(manifest.projection_epoch))
        .u64(result_kind_tag(manifest.result_kind));
    match &manifest.artifact_bundle_root {
        None => framing.u64(0),
        Some(root) => framing.u64(1).digest_field(&root.0),
    };
    let mut outputs: Vec<&rabs_protocol::result_identity::LogicalOutput> =
        manifest.logical_outputs.iter().collect();
    outputs.sort_by(|a, b| {
        (output_role_tag(a.role), a.virtual_path.as_bytes())
            .cmp(&(output_role_tag(b.role), b.virtual_path.as_bytes()))
    });
    framing.u64(outputs.len() as u64);
    for output in outputs {
        framing
            .u64(output_role_tag(output.role))
            .field(output.virtual_path.as_bytes())
            .digest_field(&output.object.0);
    }
    framing.finish(SEMANTIC_PROJECTION_DOMAIN)
}

/// The v1 observable result projection: the semantic projection plus the
/// canonical observation stream digest.
#[must_use]
pub fn observable_result_digest_v1(
    manifest: &CanonicalActionResultManifest,
    canonical_observations: &TypedDigest,
) -> TypedDigest {
    let mut framing = Framing::new(OBSERVABLE_PROJECTION_DOMAIN);
    framing
        .digest_field(&semantic_result_digest_v1(manifest))
        .digest_field(canonical_observations);
    framing.finish(OBSERVABLE_PROJECTION_DOMAIN)
}

/// Worker-side offer construction failures.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OfferBuildError {
    /// The manifest failed structural validation.
    ManifestInvalid(&'static str),
    /// The evidence bundle does not reference this manifest.
    EvidenceManifestMismatch,
    /// Evidence/manifest/authority disagree on the action key.
    ActionKeyMismatch,
    /// An output was harvested that the action never declared.
    UndeclaredOutput {
        /// Escaped virtual path of the offending output.
        path: String,
    },
    /// Two provisional-ancestor references name the same
    /// (producer, role, path) — the canonical set admits no duplicates.
    DuplicateAncestorRef {
        /// Escaped virtual path of the duplicated reference.
        path: String,
    },
    /// A provisional-ancestor reference names the offering action itself.
    SelfAncestorRef,
    /// The manifest's declared artifact bundle root does not match the
    /// deterministic derivation from the output map, or is present or
    /// absent when it must not be (H039; risk R125).
    BundleRoot(rabs_key::logical_output_map::BundleRootError),
}

/// One provisional-ancestor reference carried by a prepared result
/// (H028; I32): the offering attempt consumed `consumed_object` as the
/// producer action's `(role, virtual_path)` provisional output before
/// that producer had committed. Commit verifies the producer is now
/// committed and resolves that logical output to the EXACT consumed
/// object (or an explicit adoption edge).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProvisionalAncestorRef {
    /// Action key of the producer whose provisional output was consumed.
    pub producer_action_key: TypedDigest,
    /// Role of the consumed logical output.
    pub role: OutputRole,
    /// Canonical virtual path of the consumed logical output.
    pub virtual_path: RawBytes,
    /// The exact object the attempt consumed.
    pub consumed_object: ObjectId,
}

/// Stable string tag for an output role (persisted in lineage and
/// adoption rows; the numeric `output_role_tag` above is framing-only).
const fn output_role_name(role: OutputRole) -> &'static str {
    match role {
        OutputRole::Materializable => "materializable",
        OutputRole::DepInfo => "dep-info",
        OutputRole::ProvisionalMetadata => "provisional-metadata",
        OutputRole::BuildScriptMetadata => "build-script-metadata",
        OutputRole::TestSideEffect => "test-side-effect",
    }
}

/// Inverse of [`output_role_name`] for stored role tags (obligation rows
/// persist the numeric tag; adoption edges key by name).
pub(crate) const fn output_role_name_for_tag(tag: i64) -> &'static str {
    match tag {
        1 => "dep-info",
        2 => "provisional-metadata",
        3 => "build-script-metadata",
        4 => "test-side-effect",
        _ => "materializable",
    }
}

/// A prepared result offered for publication, carrying FULL authority
/// identity (never bare ids).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OfferPreparedActionResult {
    /// The full attempt authority (coordinator + generation + attempt +
    /// lease + worker identity).
    pub authority: AttemptAuthority,
    /// The canonical result manifest.
    pub manifest: CanonicalActionResultManifest,
    /// CAS identity of the uploaded manifest object.
    pub manifest_id: ObjectId,
    /// The attempt evidence bundle.
    pub evidence: AttemptEvidenceBundle,
    /// CAS identity of the uploaded evidence object.
    pub evidence_id: ObjectId,
    /// Digest of the canonical observation stream (input to the
    /// observable projection recompute).
    pub canonical_observations: TypedDigest,
    /// Canonical (sorted, duplicate-free) provisional-ancestor set
    /// (H028; I32). Empty when no provisional outputs were consumed.
    pub provisional_ancestors: Vec<ProvisionalAncestorRef>,
}

impl OfferPreparedActionResult {
    /// Worker-side build: validate the harvested manifest against the
    /// action's output declarations, recompute and STAMP both projection
    /// digests, and bind evidence to manifest. The returned offer is the
    /// only thing a worker may send — committing is structurally out of
    /// reach.
    ///
    /// # Errors
    /// A typed [`OfferBuildError`] naming the violated rule.
    #[allow(clippy::too_many_arguments)]
    pub fn build(
        authority: AttemptAuthority,
        mut manifest: CanonicalActionResultManifest,
        manifest_id: ObjectId,
        evidence: AttemptEvidenceBundle,
        evidence_id: ObjectId,
        canonical_observations: TypedDigest,
        declared_outputs: &[(OutputRole, RawBytes)],
        mut provisional_ancestors: Vec<ProvisionalAncestorRef>,
    ) -> Result<Self, OfferBuildError> {
        manifest
            .validate()
            .map_err(OfferBuildError::ManifestInvalid)?;
        // Canonical sorted ancestor set: order-independent identity,
        // duplicates refused, self-reference refused (H028).
        provisional_ancestors.sort_by(|a, b| {
            (
                digest_key(&a.producer_action_key),
                output_role_name(a.role),
                a.virtual_path.as_bytes(),
            )
                .cmp(&(
                    digest_key(&b.producer_action_key),
                    output_role_name(b.role),
                    b.virtual_path.as_bytes(),
                ))
        });
        for pair in provisional_ancestors.windows(2) {
            if pair[0].producer_action_key == pair[1].producer_action_key
                && pair[0].role == pair[1].role
                && pair[0].virtual_path == pair[1].virtual_path
            {
                return Err(OfferBuildError::DuplicateAncestorRef {
                    path: pair[1].virtual_path.escaped(),
                });
            }
        }
        if provisional_ancestors
            .iter()
            .any(|a| a.producer_action_key == manifest.action_key)
        {
            return Err(OfferBuildError::SelfAncestorRef);
        }
        if evidence.canonical_result_manifest_id != manifest_id {
            return Err(OfferBuildError::EvidenceManifestMismatch);
        }
        if evidence.action_key != manifest.action_key || authority.action_key != manifest.action_key
        {
            return Err(OfferBuildError::ActionKeyMismatch);
        }
        for output in &manifest.logical_outputs {
            let declared = declared_outputs
                .iter()
                .any(|(role, path)| *role == output.role && *path == output.virtual_path);
            if !declared {
                return Err(OfferBuildError::UndeclaredOutput {
                    path: output.virtual_path.escaped(),
                });
            }
        }
        // H039: STAMP the deterministic bundle root from the single
        // role-tagged map (F035's one implementation), THEN verify —
        // so a worker can neither omit nor contradict it, and the
        // digests below cover the derived root.
        manifest.artifact_bundle_root =
            rabs_key::logical_output_map::compute_bundle_root(&manifest.logical_outputs);
        rabs_key::logical_output_map::verify_manifest_bundle_root(&manifest)
            .map_err(OfferBuildError::BundleRoot)?;
        manifest.semantic_result_digest = semantic_result_digest_v1(&manifest);
        manifest.observable_result_digest =
            observable_result_digest_v1(&manifest, &canonical_observations);
        Ok(Self {
            authority,
            manifest,
            manifest_id,
            evidence,
            evidence_id,
            canonical_observations,
            provisional_ancestors,
        })
    }
}

/// Typed coordinator refusals — the offer cannot publish. A refusal does
/// not create a divergence incident; immutable uploaded candidates remain.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OfferRefusal {
    /// The offer's full authority does not digest to the active one.
    NotActiveAuthority,
    /// The generation's `created_under_authority_digest` differs from the
    /// digest of the offer's full authority (F033 equality violation).
    GenerationAuthorityMismatch,
    /// The generation does not exist in the store.
    UnknownGeneration,
    /// The generation is tombstoned (superseded/aborted).
    GenerationTombstoned,
    /// The attempt is not recorded under that generation.
    UnknownAttempt,
    /// The execution lease is missing.
    UnknownLease,
    /// The execution lease was released.
    LeaseReleased,
    /// The execution lease expired on the coordinator's own monotonic clock.
    /// Uploaded immutable objects remain candidates; this offer cannot publish.
    LeaseExpired,
    /// Reloaded canonical descriptor digest differs byte-for-byte.
    DescriptorMismatch,
    /// Manifest action key differs from the authority's action key.
    ActionKeyMismatch,
    /// No action entry exists for the key.
    UnknownActionEntry,
    /// Manifest epochs disagree with the action entry.
    EpochMismatch,
    /// Independent recompute of the artifact bundle root from the
    /// output map disagrees with the declared root, or the map itself
    /// is structurally invalid (duplicate roles, failure with outputs)
    /// — the H039/R125 admission gate.
    BundleRootMismatch {
        /// The F035 rule that failed, rendered.
        rule: String,
    },
    /// Independent recompute of `semantic_result_digest` disagrees with
    /// the declared value.
    SemanticDigestMismatch,
    /// Independent recompute of `observable_result_digest` disagrees with
    /// the declared value.
    ObservableDigestMismatch,
    /// A referenced object has no recorded location (incomplete closure).
    IncompleteObjectClosure {
        /// Digest key of the first missing object.
        missing: String,
    },
    /// A referenced object is located but no copy satisfies the FULL
    /// durability profile the commit is configured to require (H032):
    /// acknowledging would create a committed pointer to bytes that may
    /// only exist in volatile page cache.
    ObjectNotDurable {
        /// Digest key of the first non-durable object.
        missing: String,
    },
    /// The committed manifest for a same-key offer could not be loaded
    /// for divergence classification.
    CommittedManifestUnavailable,
    /// A referenced provisional-ancestor producer (direct or transitive)
    /// has no committed publication (H028; I32: no dependent commit
    /// before producer finalization).
    ProvisionalProducerNotCommitted {
        /// Digest key of the uncommitted producer action.
        producer: String,
    },
    /// A committed ancestor's manifest bytes could not be loaded for the
    /// exact-object check.
    AncestorManifestUnavailable {
        /// Digest key of the producer action.
        producer: String,
    },
    /// The producer committed, but its canonical manifest has no logical
    /// output at the referenced (role, path).
    AncestorOutputMissing {
        /// Digest key of the producer action.
        producer: String,
        /// Escaped virtual path of the missing output.
        path: String,
    },
    /// The producer committed a DIFFERENT object at the referenced
    /// logical output and no adoption edge covers the consumed object
    /// (risk R64: the consumed provisional bytes are not what the
    /// winning attempt published).
    DivergentProvisionalAncestor {
        /// Digest key of the producer action.
        producer: String,
        /// Escaped virtual path of the divergent output.
        path: String,
    },
    /// The attempt durably CONSUMED a provisional output (an open M006
    /// obligation) whose producer lineage the offer does not declare —
    /// concealed consumption is never a commit.
    UndeclaredProvisionalConsumption {
        /// Canonical key of the consumed provisional pin.
        pin_key: String,
    },
    /// A consumed producer lineage was CANCELLED (producer failed,
    /// superseded without adoption, or lost authority): the descendant
    /// can never publish (M006/M007).
    ConsumptionLineageCancelled {
        /// Canonical key of the cancelled pin.
        pin_key: String,
    },
    /// A bound native child action is unresolved (L008): the parent
    /// build-script result cannot commit until the child commits.
    NativeChildUnresolved {
        /// Digest of the unresolved child action key.
        child_action_key: String,
    },
    /// The store refused or failed.
    Store(StoreError),
}

impl From<StoreError> for OfferRefusal {
    fn from(e: StoreError) -> Self {
        Self::Store(e)
    }
}

/// One consumer escalated after a semantic divergence (H026): the
/// consumer was previously served this action's result, so it must be
/// told the result's determinism is now in question. The decision is
/// tiered by the action's latest trust/release evaluation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumerEscalation {
    /// The served consumer (as recorded in `served-to` provenance).
    pub consumer: String,
    /// The action's latest trust-evaluation state at escalation time
    /// (`"unevaluated"` when no evaluation exists).
    pub trust_state: String,
    /// The tiered escalation decision recorded in the receipt.
    pub decision: String,
}

/// The full H026 quarantine outcome for a same-key divergence: what was
/// quarantined, which incident row records it, which pin preserves the
/// losing candidate, and who was escalated.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DivergenceQuarantine {
    /// The A018 divergence class.
    pub class: DivergenceClass,
    /// Sequence of the append-only incident row.
    pub incident_seq: u64,
    /// Id of the durable candidate-preservation pin.
    pub candidate_pin_id: u128,
    /// Consumers escalated from `served-to` provenance (semantic
    /// divergence only; empty for the other classes).
    pub escalations: Vec<ConsumerEscalation>,
}

/// Admitted outcomes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PublicationOutcome {
    /// Committed durably; the receipt exists only after the store
    /// transaction succeeded.
    Committed(ActionPublicationRecord),
    /// Same manifest already committed: evidence appended, nothing else
    /// changed.
    IdempotentEvidenceAppended,
    /// Same key, different result: the A018 class-specific quarantine is
    /// applied; the committed row is preserved untouched and BOTH
    /// candidates survive (H026).
    Quarantined(DivergenceQuarantine),
}

/// The configured CAS durability profile a commit acknowledgement gates
/// on (H032): what "the objects exist" must mean BEFORE the metadata
/// transaction may commit. `ActionResultCommitted` (the
/// [`PublicationOutcome::Committed`] receipt) exists only after both the
/// CAS side satisfies this profile and the store transaction reported
/// durable success.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CommitDurabilityProfile {
    /// Authoritative default: every object in the offer's closure must
    /// have a durable (full fsync profile), non-quarantined location. A
    /// power failure after commit can then never orphan the committed
    /// pointer.
    RequireDurableClosure,
    /// Explicitly volatile deployment (throwaway scratch cache): any
    /// non-quarantined location admits. Named, never the silent default.
    AcceptVolatileLocations,
}

const fn result_kind_to_tag(kind: ResultKind) -> ResultKindTag {
    match kind {
        ResultKind::Success => ResultKindTag::Success,
        ResultKind::DeterministicFailure => ResultKindTag::DeterministicFailure,
    }
}

/// Coordinator-only admission + atomic publication. `expected_descriptor`
/// is the coordinator's OWN reloaded copy of the canonical descriptor
/// digest; `committed_manifest` resolves a manifest digest key to the
/// manifest bytes the coordinator holds in its CAS (used only for
/// same-key divergence classification); `pin_id` is the coordinator-
/// allocated durable pin id — the publication reachability pin on
/// commit, or the candidate-preservation pin when the offer diverges
/// (H026); `seq` is the causal commit sequence stamped into the receipt
/// (and into the incident row on divergence); `durability` is the
/// configured H032 commit profile — under
/// [`CommitDurabilityProfile::RequireDurableClosure`] every closure
/// object must have a durable location BEFORE the metadata transaction
/// runs, so the committed pointer can never name bytes a power failure
/// may still lose.
/// `own_monotonic_now_ms` reads the granting coordinator's monotonic clock,
/// independent of `seq` and every peer's timestamp. Restarted coordinators
/// must acquire a new authority before using their new clock origin. The
/// callback is resampled at publication, so validation cannot extend a lease.
///
/// # Errors
/// A typed [`OfferRefusal`]. Refusals do not publish a pointer, winner evidence,
/// or publication pin. Previously verified lineage obligations may have been
/// reconciled before a late refusal; those records do not grant publication.
#[allow(clippy::too_many_arguments)]
pub fn process_offer(
    store: &mut dyn RabsMetadataStore,
    offer: &OfferPreparedActionResult,
    expected_descriptor: &TypedDigest,
    manifest_resolver: impl Fn(&str) -> Option<CanonicalActionResultManifest>,
    pin_id: u128,
    seq: u64,
    durability: CommitDurabilityProfile,
    own_monotonic_now_ms: impl Fn() -> u64,
) -> Result<PublicationOutcome, OfferRefusal> {
    // 1. Authority fence: active authority + F033 digest equality.
    let offered_authority = authority_digest(&offer.authority.coordinator);
    let active = store.active_authority()?;
    match active {
        Some(row) if row.digest == offered_authority => {}
        _ => return Err(OfferRefusal::NotActiveAuthority),
    }
    if offer
        .authority
        .action_generation
        .created_under_authority_digest
        != offered_authority
    {
        return Err(OfferRefusal::GenerationAuthorityMismatch);
    }

    // 2. Generation fence.
    let generation_id = offer.authority.action_generation.generation_id.0;
    match store.generation_state(generation_id)? {
        None => return Err(OfferRefusal::UnknownGeneration),
        Some(state) if state.tombstoned => return Err(OfferRefusal::GenerationTombstoned),
        Some(_) => {}
    }

    // 3. Attempt + lease fences.
    let attempt_id = offer.authority.attempt_id.0;
    if !store.attempt_exists(attempt_id, generation_id)? {
        return Err(OfferRefusal::UnknownAttempt);
    }
    store
        .validate_attempt_lease(&offer.authority, own_monotonic_now_ms())
        .map_err(lease_refusal)?;

    // 4. Descriptor reload + byte-compare.
    if offer.manifest.canonical_descriptor_digest != *expected_descriptor {
        return Err(OfferRefusal::DescriptorMismatch);
    }

    // 5. Key + epoch validation.
    if offer.manifest.action_key != offer.authority.action_key {
        return Err(OfferRefusal::ActionKeyMismatch);
    }
    let entry = store
        .lookup_action(&offer.manifest.action_key)?
        .ok_or(OfferRefusal::UnknownActionEntry)?;
    if entry.key_epoch != offer.manifest.key_epoch
        || entry.projection_epoch != offer.manifest.projection_epoch
    {
        return Err(OfferRefusal::EpochMismatch);
    }

    // 5.5. Independent bundle-root recompute (H039; R125): the
    // coordinator re-derives the root from the single role-tagged map
    // with F035's one implementation and refuses any manifest whose
    // declared root is absent, contradictory, or whose map is
    // structurally invalid — a worker-stamped root is never trusted.
    if let Err(rule) = rabs_key::logical_output_map::verify_manifest_bundle_root(&offer.manifest) {
        return Err(OfferRefusal::BundleRootMismatch {
            rule: format!("{rule:?}"),
        });
    }

    // 6. Independent digest recompute from the versioned projections.
    if semantic_result_digest_v1(&offer.manifest) != offer.manifest.semantic_result_digest {
        return Err(OfferRefusal::SemanticDigestMismatch);
    }
    if observable_result_digest_v1(&offer.manifest, &offer.canonical_observations)
        != offer.manifest.observable_result_digest
    {
        return Err(OfferRefusal::ObservableDigestMismatch);
    }

    // 7. Complete object closure: manifest, evidence, bundle root, every
    // logical output, every evidence constituent.
    let mut closure: Vec<&TypedDigest> = vec![&offer.manifest_id.0, &offer.evidence_id.0];
    if let Some(root) = &offer.manifest.artifact_bundle_root {
        closure.push(&root.0);
    }
    for output in &offer.manifest.logical_outputs {
        closure.push(&output.object.0);
    }
    closure.push(&offer.evidence.execution_snapshot_root.0);
    closure.push(&offer.evidence.observed_input_report.0);
    closure.push(&offer.evidence.raw_process_and_event_evidence.0);
    closure.push(&offer.evidence.provenance_receipt.0);
    if let Some(snapshot) = &offer.evidence.incremental_snapshot {
        closure.push(&snapshot.0);
    }
    for object in closure {
        if !store.object_located(object)? {
            return Err(OfferRefusal::IncompleteObjectClosure {
                missing: digest_key(object),
            });
        }
        // H032: under the authoritative profile the metadata transaction
        // may not even begin until every closure object has a durable
        // copy — located-but-volatile is a typed refusal, not a commit.
        if durability == CommitDurabilityProfile::RequireDurableClosure
            && !store.object_durably_located(object)?
        {
            return Err(OfferRefusal::ObjectNotDurable {
                missing: digest_key(object),
            });
        }
    }

    // 7.5. Provisional-ancestor closure (H028; I32): EVERY referenced
    // producer — direct AND transitive through each committed producer's
    // recorded lineage — must be committed and resolve the referenced
    // logical output to the exact consumed object (or an explicit
    // adoption edge). Direct-only checking would let an A→B→C chain
    // commit C over a hole at A.
    let ancestor_rows = verify_provisional_ancestry(store, offer, &manifest_resolver)?;

    // 7.6. Consumption-obligation cross-check (M008): the offer's
    // self-declared ancestor set is verified above; this gate verifies it
    // against the coordinator-recorded M006 obligations of the offering
    // attempt — concealed consumption, cancelled lineages, and
    // not-yet-finalized producers all refuse here.
    verify_consumption_obligations(store, offer, &manifest_resolver)?;

    // 7.7. Native child resolution gate (L008): a build-script parent
    // with bound native children cannot commit while any child result
    // is unresolved — provenance edges are written on satisfaction.
    crate::native_children::enforce_native_children_resolved(store, &offer.manifest.action_key)
        .map_err(|err| match err {
            crate::native_children::NativeChildError::ChildUnresolved { child_action_key } => {
                OfferRefusal::NativeChildUnresolved { child_action_key }
            }
            crate::native_children::NativeChildError::Store(store_err) => {
                OfferRefusal::Store(store_err)
            }
        })?;

    // 8. Same-key candidates: divergence taxonomy, never overwrite.
    if let Some(committed_key) = store.published_manifest_key(&offer.manifest.action_key)? {
        store
            .validate_attempt_lease(&offer.authority, own_monotonic_now_ms())
            .map_err(lease_refusal)?;
        let candidate_key = digest_key(&offer.manifest_id.0);
        if committed_key == candidate_key {
            // Idempotent re-offer: append evidence only, bound to the
            // committed canonical manifest it supports (H029; I37).
            store
                .append_evidence_for_attempt(
                    &offer.authority,
                    &committed_key,
                    &offer.evidence_id.0,
                    &own_monotonic_now_ms,
                )
                .map_err(lease_refusal)?;
            return Ok(PublicationOutcome::IdempotentEvidenceAppended);
        }
        let committed =
            manifest_resolver(&committed_key).ok_or(OfferRefusal::CommittedManifestUnavailable)?;
        store
            .validate_attempt_lease(&offer.authority, own_monotonic_now_ms())
            .map_err(lease_refusal)?;
        // A018 taxonomy with the manifest-id inequality already
        // established above (committed_key != candidate_key), so the
        // idempotent branch is unreachable here by construction.
        let class = if committed.semantic_result_digest != offer.manifest.semantic_result_digest {
            DivergenceClass::SemanticDivergence
        } else if committed.observable_result_digest != offer.manifest.observable_result_digest {
            DivergenceClass::ObservableOnlyDivergence
        } else {
            DivergenceClass::ProjectionCompletenessIncident
        };
        let quarantine = quarantine_divergence(
            store,
            &offered_authority,
            offer,
            class,
            &committed_key,
            generation_id,
            attempt_id,
            pin_id,
            seq,
            &own_monotonic_now_ms,
        )?;
        return Ok(PublicationOutcome::Quarantined(quarantine));
    }

    // 9. Compare-and-set commit: publication pointer + serving state +
    // winner evidence + durable reachability pin, ONE transaction.
    let row = PublicationRow {
        action_key: offer.manifest.action_key.clone(),
        descriptor_digest: offer.manifest.canonical_descriptor_digest.clone(),
        manifest_digest: offer.manifest_id.0.clone(),
        evidence_digest: offer.evidence_id.0.clone(),
        winner_generation: generation_id,
        winner_attempt: attempt_id,
        result_kind: result_kind_to_tag(offer.manifest.result_kind),
        pin_id,
        pin_owner: "coordinator".to_owned(),
        provisional_ancestors: ancestor_rows,
    };
    match store
        .commit_publication(
            &offered_authority,
            Some((&offer.authority, &own_monotonic_now_ms)),
            &row,
        )
        .map_err(lease_refusal)?
    {
        CommitOutcome::Committed => {}
        // The pipeline checked for an existing row above; hitting either
        // branch here means the store's own CAS caught a same-key row —
        // surface it as the conservative refusal, never a silent success.
        CommitOutcome::IdempotentDuplicate | CommitOutcome::ConflictQuarantined => {
            return Err(OfferRefusal::Store(StoreError::Corruption(
                "publication row appeared during admission".into(),
            )));
        }
    }
    // ActionResultCommitted: emitted only after durable success.
    Ok(PublicationOutcome::Committed(ActionPublicationRecord {
        action_key: offer.manifest.action_key.clone(),
        canonical_result_manifest_id: offer.manifest_id.clone(),
        winner_evidence_bundle_id: offer.evidence_id.clone(),
        committed_causal_sequence: seq,
    }))
}

fn lease_refusal(error: StoreError) -> OfferRefusal {
    match error {
        StoreError::UnknownLease => OfferRefusal::UnknownLease,
        StoreError::LeaseReleased => OfferRefusal::LeaseReleased,
        StoreError::LeaseExpired => OfferRefusal::LeaseExpired,
        error => OfferRefusal::Store(error),
    }
}

/// One lineage reference to verify: producer/role/path/consumed-object,
/// all in digest-key/tag form so direct refs (typed) and recorded rows
/// (strings) share one code path.
struct PendingAncestorCheck {
    producer_key: String,
    role_tag: String,
    path: Vec<u8>,
    consumed_key: String,
    /// Direct refs are recorded on the consumer's own publication row;
    /// transitive ones are already recorded on their consumer.
    direct: bool,
}

/// The H028 transitive provisional-ancestor verification (I32; R64).
///
/// Walks the offer's direct references plus, for every committed
/// producer, the lineage rows recorded at THAT producer's commit —
/// breadth-first with a visited set, so diamond graphs verify once and
/// cycles terminate. Every reference (at every depth) must satisfy:
/// producer committed, its canonical manifest resolves the (role, path)
/// logical output, and that output is the EXACT consumed object or an
/// explicit adoption edge from consumed to committed exists.
///
/// Returns the verified direct rows for the consumer's own publication.
fn verify_provisional_ancestry(
    store: &mut dyn RabsMetadataStore,
    offer: &OfferPreparedActionResult,
    manifest_resolver: &impl Fn(&str) -> Option<CanonicalActionResultManifest>,
) -> Result<Vec<ProvisionalAncestorRow>, OfferRefusal> {
    let mut direct_rows = Vec::new();
    let mut queue: std::collections::VecDeque<PendingAncestorCheck> = offer
        .provisional_ancestors
        .iter()
        .map(|a| PendingAncestorCheck {
            producer_key: digest_key(&a.producer_action_key),
            role_tag: output_role_name(a.role).to_owned(),
            path: a.virtual_path.as_bytes().to_vec(),
            consumed_key: digest_key(&a.consumed_object.0),
            direct: true,
        })
        .collect();
    let mut walked_producers = std::collections::BTreeSet::new();

    while let Some(check) = queue.pop_front() {
        // Producer committed? (Direct-only shortcuts are exactly the
        // R64 hole; this fires at every depth.)
        let manifest_key = store
            .published_manifest_key_str(&check.producer_key)?
            .ok_or_else(|| OfferRefusal::ProvisionalProducerNotCommitted {
                producer: check.producer_key.clone(),
            })?;
        let manifest = manifest_resolver(&manifest_key).ok_or_else(|| {
            OfferRefusal::AncestorManifestUnavailable {
                producer: check.producer_key.clone(),
            }
        })?;
        // Exact-object resolution at the referenced logical output.
        let output = manifest
            .logical_outputs
            .iter()
            .find(|o| {
                output_role_name(o.role) == check.role_tag
                    && o.virtual_path.as_bytes() == check.path.as_slice()
            })
            .ok_or_else(|| OfferRefusal::AncestorOutputMissing {
                producer: check.producer_key.clone(),
                path: RawBytes::new(check.path.clone()).escaped(),
            })?;
        let committed_key = digest_key(&output.object.0);
        let adopted = if committed_key == check.consumed_key {
            false
        } else if store.has_adoption_edge(
            &check.producer_key,
            &check.role_tag,
            &check.path,
            &check.consumed_key,
            &committed_key,
        )? {
            true
        } else {
            return Err(OfferRefusal::DivergentProvisionalAncestor {
                producer: check.producer_key.clone(),
                path: RawBytes::new(check.path.clone()).escaped(),
            });
        };
        if check.direct {
            direct_rows.push(ProvisionalAncestorRow {
                producer_action_key: check.producer_key.clone(),
                role: check.role_tag.clone(),
                virtual_path: check.path.clone(),
                object_key: check.consumed_key.clone(),
                adopted,
            });
        }
        // Transitive expansion: the producer's own recorded lineage.
        if walked_producers.insert(check.producer_key.clone()) {
            for row in store.list_provisional_ancestors(&check.producer_key)? {
                if walked_producers.contains(&row.producer_action_key) {
                    continue;
                }
                queue.push_back(PendingAncestorCheck {
                    producer_key: row.producer_action_key,
                    role_tag: row.role,
                    path: row.virtual_path,
                    consumed_key: row.object_key,
                    direct: false,
                });
            }
        }
    }
    Ok(direct_rows)
}
/// M008: cross-check the offer against the DURABLE M006 consumption
/// obligations of the offering attempt. H028's walk verifies the offer's
/// SELF-declared ancestor set; obligations are coordinator-recorded truth
/// of what the attempt actually consumed via `resolve_for_reader`. Every
/// non-resolved obligation must be (a) honestly declared by the offer,
/// and (b) finalized — producer committed resolving that logical output
/// to the EXACT consumed object (or an explicit adoption edge) — in which
/// case it auto-resolves; a cancelled lineage refuses permanently.
///
/// # Errors
/// Typed [`OfferRefusal`]s; store failures.
fn verify_consumption_obligations(
    store: &mut dyn RabsMetadataStore,
    offer: &OfferPreparedActionResult,
    manifest_resolver: &impl Fn(&str) -> Option<CanonicalActionResultManifest>,
) -> Result<(), OfferRefusal> {
    let rows = store.list_open_provisional_obligations(
        offer.authority.worker_peer_id.0.as_str(),
        &format!("{:032x}", offer.authority.attempt_id.0),
    )?;
    for obligation in rows {
        if obligation.status == "cancelled" {
            return Err(OfferRefusal::ConsumptionLineageCancelled {
                pin_key: obligation.pin_key,
            });
        }
        // (a) The offer must declare what it durably consumed.
        let declared = offer.provisional_ancestors.iter().any(|a| {
            digest_key(&a.producer_action_key) == obligation.producer_action_key
                && output_role_tag(a.role) == u64::try_from(obligation.role_tag).unwrap_or(u64::MAX)
                && a.virtual_path.as_bytes() == obligation.virtual_path.as_slice()
        });
        if !declared {
            return Err(OfferRefusal::UndeclaredProvisionalConsumption {
                pin_key: obligation.pin_key,
            });
        }
        // (b) Producer finalized with-exact-object (or adopted). Same
        // judgment as H028's walk; on satisfaction the obligation closes.
        let manifest_key = store
            .published_manifest_key_str(&obligation.producer_action_key)?
            .ok_or_else(|| OfferRefusal::ProvisionalProducerNotCommitted {
                producer: obligation.producer_action_key.clone(),
            })?;
        let manifest = manifest_resolver(&manifest_key).ok_or_else(|| {
            OfferRefusal::AncestorManifestUnavailable {
                producer: obligation.producer_action_key.clone(),
            }
        })?;
        let output = manifest
            .logical_outputs
            .iter()
            .find(|o| {
                output_role_tag(o.role) == u64::try_from(obligation.role_tag).unwrap_or(u64::MAX)
                    && o.virtual_path.as_bytes() == obligation.virtual_path.as_slice()
            })
            .ok_or_else(|| OfferRefusal::AncestorOutputMissing {
                producer: obligation.producer_action_key.clone(),
                path: RawBytes::new(obligation.virtual_path.clone()).escaped(),
            })?;
        let committed_key = digest_key(&output.object.0);
        if committed_key != obligation.object_key
            && !store.has_adoption_edge(
                &obligation.producer_action_key,
                output_role_name_for_tag(obligation.role_tag),
                &obligation.virtual_path,
                &obligation.object_key,
                &committed_key,
            )?
        {
            return Err(OfferRefusal::DivergentProvisionalAncestor {
                producer: obligation.producer_action_key.clone(),
                path: RawBytes::new(obligation.virtual_path.clone()).escaped(),
            });
        }
        store.resolve_provisional_obligations(&obligation.pin_key, &committed_key)?;
    }
    Ok(())
}

const fn divergence_class_tag(class: DivergenceClass) -> &'static str {
    match class {
        DivergenceClass::IdempotentSameResult => "idempotent",
        DivergenceClass::SemanticDivergence => "semantic",
        DivergenceClass::ObservableOnlyDivergence => "observable-only",
        DivergenceClass::ProjectionCompletenessIncident => "projection-completeness",
    }
}

const fn divergence_quarantine_reason(class: DivergenceClass) -> &'static str {
    match class {
        DivergenceClass::IdempotentSameResult => "same-key re-offer",
        DivergenceClass::SemanticDivergence => {
            "semantic divergence: determinism/key-soundness incident"
        }
        DivergenceClass::ObservableOnlyDivergence => {
            "observable-only divergence: presentation quarantine"
        }
        DivergenceClass::ProjectionCompletenessIncident => {
            "projection completeness incident: equal digests, different manifests"
        }
    }
}

/// Tiered escalation decision for a served consumer (H026): a consumer
/// served under a RELEASE tier may have shipped artifacts built on the
/// now-suspect result, so it gets a recall; everything below release
/// tier is notified for re-verification.
fn escalation_decision(trust_state: &str) -> &'static str {
    match trust_state {
        "ci-policy-approved" | "project-release-eligible" => "recall-and-reverify",
        _ => "notify-and-reverify",
    }
}

/// Install a divergence disposition without ever relaxing an existing
/// full quarantine. The disposition write is the first fail-closed
/// mutation: unlike a standalone scoped row, it is consulted directly
/// by the ordinary serving gate.
fn set_divergence_serving_disposition(
    store: &mut dyn RabsMetadataStore,
    action_key: &str,
    class: DivergenceClass,
    attempt_lease: Option<(&AttemptAuthority, &dyn Fn() -> u64)>,
) -> Result<(), StoreError> {
    let mut disposition = match class {
        DivergenceClass::SemanticDivergence | DivergenceClass::ProjectionCompletenessIncident => {
            DISPOSITION_QUARANTINED
        }
        DivergenceClass::ObservableOnlyDivergence => DISPOSITION_PRESENTATION_QUARANTINED,
        DivergenceClass::IdempotentSameResult => {
            return Err(StoreError::Corruption(
                "idempotent result entered divergence quarantine".into(),
            ));
        }
    };
    if disposition == DISPOSITION_PRESENTATION_QUARANTINED
        && store.serving_disposition_key(action_key)?.as_deref() == Some(DISPOSITION_QUARANTINED)
    {
        disposition = DISPOSITION_QUARANTINED;
    }
    match attempt_lease {
        Some((authority, clock)) => {
            if digest_key(&authority.action_key) != action_key {
                return Err(StoreError::AttemptAuthorityMismatch);
            }
            // Reassert even an existing quarantine inside the guarded
            // transaction: the candidate still needs live acceptance.
            store.set_serving_disposition_for_attempt(authority, disposition, clock)
        }
        None => store.set_serving_disposition_key(action_key, disposition),
    }
}

/// Apply the class-specific serving quarantine, preserve BOTH candidates
/// (the committed row is untouched; the losing candidate gets a durable
/// `divergence-evidence` pin plus reachability edges and its evidence is
/// appended to the index), open an append-only incident row, and — for
/// semantic divergence — escalate every consumer previously served this
/// result, tiered by the action's latest trust/release evaluation.
///
/// Write order is stricter-state-first within each class: a crash mid-
/// sequence can leave serving denied with partial bookkeeping, never
/// bookkept while ordinary replay remains allowed.
/// The first disposition transaction is the candidate's acceptance point:
/// it validates the exact lease and its clock before and after the write.
/// Once accepted, diagnostic recording completes without a later expiry
/// refusal; those records cannot restore serving or select another winner.
#[allow(clippy::too_many_arguments)]
fn quarantine_divergence(
    store: &mut dyn RabsMetadataStore,
    authority: &TypedDigest,
    offer: &OfferPreparedActionResult,
    class: DivergenceClass,
    committed_key: &str,
    generation_id: u128,
    attempt_id: u128,
    pin_id: u128,
    seq: u64,
    own_monotonic_now_ms: &dyn Fn() -> u64,
) -> Result<DivergenceQuarantine, OfferRefusal> {
    let action_key = digest_key(&offer.manifest.action_key);
    let candidate_key = digest_key(&offer.manifest_id.0);

    // 1. Deny ordinary replay first. This is the mutation the serving
    // gate directly consults, so a later bookkeeping failure cannot
    // leave a known divergence servable. A pre-existing full quarantine
    // is never downgraded by a later observable-only mismatch.
    set_divergence_serving_disposition(
        store,
        &action_key,
        class,
        Some((&offer.authority, own_monotonic_now_ms)),
    )
    .map_err(lease_refusal)?;

    // 2. Semantic and projection-completeness divergence make the ACTION
    // suspect and therefore get an action-entry row. Observable-only
    // divergence stops at the presentation disposition above: the
    // canonical action result itself is not suspect (T025).
    if class != DivergenceClass::ObservableOnlyDivergence {
        store.add_quarantine(
            QuarantineScope::ActionEntry,
            &action_key,
            divergence_quarantine_reason(class),
        )?;
    }

    // 3. Preserve the losing candidate: reachability edges from its
    // manifest to everything it references, then a durable pin rooted at
    // the manifest so GC preserves the whole candidate closure.
    store.add_object_edge(
        &offer.manifest_id.0,
        &offer.evidence_id.0,
        "divergence-candidate",
    )?;
    if let Some(root) = &offer.manifest.artifact_bundle_root {
        store.add_object_edge(&offer.manifest_id.0, &root.0, "divergence-candidate")?;
    }
    for output in &offer.manifest.logical_outputs {
        store.add_object_edge(
            &offer.manifest_id.0,
            &output.object.0,
            "divergence-candidate",
        )?;
    }
    store.create_pin(
        pin_id,
        &offer.manifest_id.0,
        "coordinator",
        DIVERGENCE_EVIDENCE_PIN_CLASS,
        None,
        Some(&format!("divergence-incident:{action_key}:{seq}")),
        true,
        divergence_quarantine_reason(class),
    )?;

    // 4. The candidate's evidence bundle joins the append-only index —
    // attempt evidence is preserved for BOTH candidates (I34) — bound to
    // the CANDIDATE manifest, so the committed canonical result's
    // evidence view never absorbs a divergent candidate's evidence
    // (H029; I37).
    store.append_evidence(
        &offer.manifest.action_key,
        &candidate_key,
        &offer.evidence_id.0,
        generation_id,
        attempt_id,
    )?;

    // 5. Open the incident (append-only; authority-gated).
    store.record_divergence_incident(
        authority,
        &DivergenceIncidentRow {
            action_key: action_key.clone(),
            seq,
            class: divergence_class_tag(class).to_owned(),
            committed_manifest_key: committed_key.to_owned(),
            candidate_manifest_key: candidate_key,
            candidate_evidence_key: digest_key(&offer.evidence_id.0),
            candidate_pin_hex: format!("{pin_id:032x}"),
            generation_hex: format!("{generation_id:032x}"),
            attempt_hex: format!("{attempt_id:032x}"),
            detail: divergence_quarantine_reason(class).to_owned(),
        },
    )?;

    // 6. Semantic divergence escalates previously served consumers from
    // provenance, tiered by the latest trust/release evaluation.
    let mut escalations = Vec::new();
    if class == DivergenceClass::SemanticDivergence {
        let trust_state = store
            .latest_trust_evaluation(&offer.manifest.action_key)?
            .map_or_else(|| "unevaluated".to_owned(), |row| row.state);
        for consumer in store.list_served_consumers(&action_key)? {
            let decision = escalation_decision(&trust_state);
            store.record_decision_receipt(
                "divergence-escalation",
                &consumer,
                seq,
                decision,
                &format!("semantic divergence on {action_key}; trust {trust_state}"),
            )?;
            escalations.push(ConsumerEscalation {
                consumer,
                trust_state: trust_state.clone(),
                decision: decision.to_owned(),
            });
        }
    }

    Ok(DivergenceQuarantine {
        class,
        incident_seq: seq,
        candidate_pin_id: pin_id,
        escalations,
    })
}

/// Outcome of comparing a post-eviction recomputation against the
/// retained eviction tombstone (bead H034).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RecomputationCheck {
    /// No tombstone exists for the key: nothing to compare.
    NoTombstone,
    /// Both digests match the retained values: the recomputation
    /// reproduces the evicted result; the tombstone is consumed.
    Reproduced,
    /// The recomputation DIVERGES from the evicted result: a class-
    /// specific quarantine is applied, never a silent replacement.
    Divergence(DivergenceClass),
}

/// H034: retain a published result's projection digests before its blobs
/// are evicted, so a later recomputation of the same key has something
/// to answer to.
///
/// # Errors
/// Store errors from the tombstone write.
pub fn retain_eviction_tombstone(
    store: &mut dyn RabsMetadataStore,
    manifest: &CanonicalActionResultManifest,
    evicted_seq: u64,
) -> Result<(), StoreError> {
    store.record_eviction_tombstone(
        &manifest.action_key,
        &manifest.semantic_result_digest,
        &manifest.observable_result_digest,
        evicted_seq,
    )
}

/// H034: compare a re-executed result's digests against the eviction
/// tombstone. A match consumes the tombstone (the result is reproduced
/// and may republish); a mismatch applies the A018 class-specific
/// quarantine and LEAVES the tombstone in place as incident evidence.
///
/// # Errors
/// Store errors from the lookup/quarantine writes.
pub fn check_recomputation_against_tombstone(
    store: &mut dyn RabsMetadataStore,
    action: &TypedDigest,
    new_semantic: &TypedDigest,
    new_observable: &TypedDigest,
) -> Result<RecomputationCheck, StoreError> {
    let Some((retained_semantic, retained_observable)) = store.eviction_tombstone(action)? else {
        return Ok(RecomputationCheck::NoTombstone);
    };
    if retained_semantic == *new_semantic && retained_observable == *new_observable {
        store.consume_eviction_tombstone(action)?;
        return Ok(RecomputationCheck::Reproduced);
    }
    let class = if retained_semantic == *new_semantic {
        DivergenceClass::ObservableOnlyDivergence
    } else {
        DivergenceClass::SemanticDivergence
    };
    let action_key = digest_key(action);
    set_divergence_serving_disposition(store, &action_key, class, None)?;
    if class == DivergenceClass::SemanticDivergence {
        store.add_quarantine(
            QuarantineScope::ActionEntry,
            &action_key,
            "post-eviction recomputation divergence",
        )?;
    }
    Ok(RecomputationCheck::Divergence(class))
}

#[cfg(test)]
mod tests {
    use super::*;
    use rabs_protocol::authority::{ClusterId, CoordinatorAuthority, CoordinatorIncarnationId};
    use rabs_protocol::generation::{
        ActionGeneration, ActionGenerationId, AttemptId, ExecutionLeaseId, LeaseRenewal,
        LeaseRenewalSeq, WorkerBootGeneration, WorkerIncarnationId,
    };
    use rabs_protocol::result_identity::LogicalOutput;
    use rabs_protocol::wire_time::PeerId;
    use rabs_protocol::worker_fence::WorkerSessionOffer;

    use crate::metadata_store::{
        ActionEntryRow, AuthorityRow, FsqliteEngine, ProvisionalObligationInsert, RusqliteEngine,
        SqlMetadataStore,
    };
    use std::sync::atomic::{AtomicU64, Ordering};

    static DB_COUNTER: AtomicU64 = AtomicU64::new(0);

    fn fresh_path(tag: &str) -> std::path::PathBuf {
        let n = DB_COUNTER.fetch_add(1, Ordering::SeqCst);
        std::env::temp_dir().join(format!("rabs-h011-{}-{}-{}.db", std::process::id(), tag, n))
    }

    fn digest(domain: &'static str, tag: u8) -> TypedDigest {
        TypedDigest {
            algorithm: DigestAlgorithm::Sha256V1,
            domain,
            bytes: [tag; 32],
        }
    }

    fn object(tag: u8) -> ObjectId {
        ObjectId(digest("rabs.object.sha256.v1", tag))
    }

    fn content_object(bytes: &[u8]) -> ObjectId {
        ObjectId(
            crate::digest_set::digest_set(bytes, crate::digest_set::DigestRequest::default(), None)
                .unwrap()
                .atp_content_id,
        )
    }

    fn coordinator_authority() -> CoordinatorAuthority {
        CoordinatorAuthority {
            cluster_id: ClusterId("cluster-a".to_owned()),
            credential_generation: 1,
            term: 3,
            incarnation_id: CoordinatorIncarnationId(77),
        }
    }

    fn attempt_authority() -> AttemptAuthority {
        let coordinator = coordinator_authority();
        let created_under = authority_digest(&coordinator);
        AttemptAuthority {
            coordinator,
            action_key: digest("rabs.action-key.sha256.v1", 7),
            action_generation: ActionGeneration {
                generation_id: ActionGenerationId(11),
                per_key_ordinal: 1,
                created_under_authority_digest: created_under,
            },
            attempt_id: AttemptId(20),
            execution_lease_id: ExecutionLeaseId(30),
            lease_renewal_seq: LeaseRenewalSeq(1),
            worker_peer_id: PeerId("worker-a".to_owned()),
            worker_boot_generation: WorkerBootGeneration(1),
            worker_incarnation_id: WorkerIncarnationId(5),
        }
    }

    fn distinct_attempt_authority() -> AttemptAuthority {
        let mut authority = attempt_authority();
        authority.attempt_id = AttemptId(21);
        authority.execution_lease_id = ExecutionLeaseId(31);
        authority.worker_peer_id = PeerId("worker-b".to_owned());
        authority.worker_boot_generation = WorkerBootGeneration(2);
        authority.worker_incarnation_id = WorkerIncarnationId(6);
        authority
    }

    fn manifest() -> CanonicalActionResultManifest {
        CanonicalActionResultManifest {
            action_key: digest("rabs.action-key.sha256.v1", 7),
            canonical_descriptor_digest: digest("rabs.descriptor.sha256.v1", 8),
            key_epoch: 1,
            projection_epoch: 1,
            result_kind: ResultKind::Success,
            artifact_bundle_root: Some(object(40)),
            logical_outputs: vec![LogicalOutput {
                role: OutputRole::Materializable,
                virtual_path: RawBytes::new(b"out/lib.rlib".to_vec()),
                object: object(41),
            }],
            // Stamped by build().
            semantic_result_digest: digest(SEMANTIC_PROJECTION_DOMAIN, 0),
            observable_result_digest: digest(OBSERVABLE_PROJECTION_DOMAIN, 0),
        }
    }

    fn evidence(manifest_id: &ObjectId) -> AttemptEvidenceBundle {
        AttemptEvidenceBundle {
            action_key: digest("rabs.action-key.sha256.v1", 7),
            canonical_result_manifest_id: manifest_id.clone(),
            execution_snapshot_root: object(60),
            observed_input_report: object(61),
            raw_process_and_event_evidence: object(62),
            provenance_receipt: object(63),
            incremental_snapshot: None,
        }
    }

    fn declared() -> Vec<(OutputRole, RawBytes)> {
        vec![(
            OutputRole::Materializable,
            RawBytes::new(b"out/lib.rlib".to_vec()),
        )]
    }

    fn offer() -> OfferPreparedActionResult {
        let manifest_id = object(50);
        OfferPreparedActionResult::build(
            attempt_authority(),
            manifest(),
            manifest_id.clone(),
            evidence(&manifest_id),
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap()
    }

    /// A store with the full admission world installed: active authority,
    /// action entry, generation, attempt, lease, and every closure object
    /// located.
    fn ready_store(store: &mut dyn RabsMetadataStore) {
        let attempt_authority = attempt_authority();
        let auth = authority_digest(&attempt_authority.coordinator);
        store
            .acquire_authority(&AuthorityRow {
                digest: auth.clone(),
                cluster_id: "cluster-a".to_owned(),
                incarnation: 77,
                term: 3,
                acquired_seq: 1,
            })
            .unwrap();
        store
            .upsert_action_entry(&ActionEntryRow {
                action_key: digest("rabs.action-key.sha256.v1", 7),
                key_epoch: 1,
                projection_epoch: 1,
            })
            .unwrap();
        store
            .create_bound_generation(
                &auth,
                &attempt_authority.action_generation,
                &digest("rabs.action-key.sha256.v1", 7),
            )
            .unwrap();
        store
            .admit_worker_session(
                &auth,
                &WorkerSessionOffer {
                    worker_peer_id: attempt_authority.worker_peer_id.clone(),
                    boot_generation: attempt_authority.worker_boot_generation,
                    incarnation: attempt_authority.worker_incarnation_id,
                    reenrollment_proof: None,
                },
                1,
            )
            .unwrap();
        store
            .admit_attempt_lease(&attempt_authority, 1, 100)
            .unwrap();
        for tag in [40, 41, 50, 51, 60, 61, 62, 63] {
            let id = object(tag);
            store.record_object(&id.0, 64).unwrap();
            store
                .add_location(&id.0, &format!("/cas/{tag}"), Some(1), "raw", true)
                .unwrap();
        }
        // H039: build() stamps the F035-derived bundle root, and the
        // closure check requires it located like any other object.
        locate_bundle_root(store, &offer());
    }

    /// Record + locate an offer's derived bundle root (idempotent).
    fn locate_bundle_root(store: &mut dyn RabsMetadataStore, offer: &OfferPreparedActionResult) {
        if let Some(root) = &offer.manifest.artifact_bundle_root {
            store.record_object(&root.0, 0).unwrap();
            store
                .add_location(&root.0, "/cas/bundle-root", Some(1), "raw", true)
                .unwrap();
        }
    }

    fn locate_test_object(
        store: &mut dyn RabsMetadataStore,
        object: &ObjectId,
        path: &str,
        logical_size: usize,
    ) {
        store
            .record_object(&object.0, u64::try_from(logical_size).unwrap())
            .unwrap();
        store
            .add_location(&object.0, path, Some(1), "raw", true)
            .unwrap();
    }

    fn expected_descriptor() -> TypedDigest {
        digest("rabs.descriptor.sha256.v1", 8)
    }

    fn no_committed(_: &str) -> Option<CanonicalActionResultManifest> {
        None
    }

    /// A committed producer world for the H028 scenarios: `key_tag`
    /// names the producer action, its canonical manifest (id
    /// `manifest_tag`) resolves one ProvisionalMetadata output at `path`
    /// to `output_tag`, and its publication row carries `ancestors`.
    #[allow(clippy::too_many_arguments)]
    fn plant_producer(
        store: &mut dyn RabsMetadataStore,
        resolver_map: &mut std::collections::BTreeMap<String, CanonicalActionResultManifest>,
        key_tag: u8,
        manifest_tag: u8,
        path: &str,
        output_tag: u8,
        pin: u128,
        ancestors: Vec<ProvisionalAncestorRow>,
    ) {
        let action = digest("rabs.action-key.sha256.v1", key_tag);
        let mut producer_manifest = manifest();
        producer_manifest.action_key = action.clone();
        producer_manifest.logical_outputs = vec![LogicalOutput {
            role: OutputRole::ProvisionalMetadata,
            virtual_path: RawBytes::from(path),
            object: object(output_tag),
        }];
        resolver_map.insert(digest_key(&object(manifest_tag).0), producer_manifest);
        let auth = authority_digest(&coordinator_authority());
        assert_eq!(
            store
                .commit_publication(
                    &auth,
                    None,
                    &PublicationRow {
                        action_key: action,
                        descriptor_digest: expected_descriptor(),
                        manifest_digest: object(manifest_tag).0,
                        evidence_digest: object(manifest_tag).0.clone(),
                        winner_generation: 11,
                        winner_attempt: 20,
                        result_kind: ResultKindTag::Success,
                        pin_id: pin,
                        pin_owner: "coordinator".to_owned(),
                        provisional_ancestors: ancestors,
                    },
                )
                .unwrap(),
            CommitOutcome::Committed
        );
    }

    fn ancestor_ref(producer_tag: u8, path: &str, consumed_tag: u8) -> ProvisionalAncestorRef {
        ProvisionalAncestorRef {
            producer_action_key: digest("rabs.action-key.sha256.v1", producer_tag),
            role: OutputRole::ProvisionalMetadata,
            virtual_path: RawBytes::from(path),
            consumed_object: object(consumed_tag),
        }
    }

    fn offer_with_ancestors(ancestors: Vec<ProvisionalAncestorRef>) -> OfferPreparedActionResult {
        let manifest_id = object(50);
        OfferPreparedActionResult::build(
            attempt_authority(),
            manifest(),
            manifest_id.clone(),
            evidence(&manifest_id),
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            ancestors,
        )
        .unwrap()
    }

    #[test]
    fn h028_transitive_ancestor_closure_adoption_and_divergence() {
        // World: A (key 100) committed ProvisionalMetadata "out/a.rmeta"
        // -> object 110; B (key 101) committed "out/b.rmeta" -> object
        // 111 and RECORDS having consumed A's object 110; C offers,
        // naming only B (consumed object 111). T020.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut manifests = std::collections::BTreeMap::new();
        plant_producer(
            &mut store,
            &mut manifests,
            100,
            120,
            "out/a.rmeta",
            110,
            800,
            vec![],
        );
        plant_producer(
            &mut store,
            &mut manifests,
            101,
            121,
            "out/b.rmeta",
            111,
            801,
            vec![ProvisionalAncestorRow {
                producer_action_key: digest_key(&digest("rabs.action-key.sha256.v1", 100)),
                role: "provisional-metadata".to_owned(),
                virtual_path: b"out/a.rmeta".to_vec(),
                object_key: digest_key(&object(110).0),
                adopted: false,
            }],
        );
        let resolver = {
            let manifests = manifests.clone();
            move |key: &str| manifests.get(key).cloned()
        };

        // Happy path: full A->B->C chain verified; C commits and its
        // publication records the direct lineage row.
        let c_offer = offer_with_ancestors(vec![ancestor_ref(101, "out/b.rmeta", 111)]);
        assert!(matches!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver.clone(),
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));
        let c_key = digest_key(&digest("rabs.action-key.sha256.v1", 7));
        let recorded = store.list_provisional_ancestors(&c_key).unwrap();
        assert_eq!(recorded.len(), 1);
        assert_eq!(
            recorded[0].producer_action_key,
            digest_key(&digest("rabs.action-key.sha256.v1", 101))
        );
        assert_eq!(recorded[0].object_key, digest_key(&object(111).0));
        assert!(!recorded[0].adopted);
    }

    #[test]
    fn m008_transitive_debt_blocks_commit_until_producers_finalize() {
        // T020 A→B→C with DURABLE consumption records: C's attempt owes an
        // M006 obligation on B's provisional output, recorded before B
        // commits. The commit transaction refuses until the whole chain
        // finalizes with-exact-object, then auto-resolves the debt.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut manifests = std::collections::BTreeMap::new();
        plant_producer(
            &mut store,
            &mut manifests,
            100,
            120,
            "out/a.rmeta",
            110,
            800,
            vec![],
        );

        store
            .record_provisional_consumption(&ProvisionalObligationInsert {
                consumer_worker: "worker-a".to_owned(),
                consumer_attempt: 20,
                pin_key: "test-pin-b".to_owned(),
                producer_action_key: digest_key(&digest("rabs.action-key.sha256.v1", 101)),
                producer_generation: 11,
                producer_attempt: 21,
                role_tag: 2,
                virtual_path: b"out/b.rmeta".to_vec(),
                object_key: digest_key(&object(111).0),
                created_seq: 1,
            })
            .unwrap();

        let resolver = {
            let manifests = manifests.clone();
            move |key: &str| manifests.get(key).cloned()
        };
        let c_offer = offer_with_ancestors(vec![ancestor_ref(101, "out/b.rmeta", 111)]);

        // B uncommitted: refused; and the obligation stays open (nothing
        // resolved by a refusal).
        assert!(matches!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::ProvisionalProducerNotCommitted { .. })
        ));
        assert_eq!(
            store
                .count_open_provisional_obligations("test-pin-b")
                .unwrap(),
            1
        );

        // B commits (recording its own A-lineage); the SAME offer now
        // passes and the debt resolves as a side effect of committing.
        plant_producer(
            &mut store,
            &mut manifests,
            101,
            121,
            "out/b.rmeta",
            111,
            801,
            vec![ProvisionalAncestorRow {
                producer_action_key: digest_key(&digest("rabs.action-key.sha256.v1", 100)),
                role: "provisional-metadata".to_owned(),
                virtual_path: b"out/a.rmeta".to_vec(),
                object_key: digest_key(&object(110).0),
                adopted: false,
            }],
        );
        let resolver = {
            let manifests = manifests.clone();
            move |key: &str| manifests.get(key).cloned()
        };
        assert!(matches!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver,
                901,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Ok(PublicationOutcome::Committed(_))
        ));
        assert_eq!(
            store
                .count_open_provisional_obligations("test-pin-b")
                .unwrap(),
            0
        );
    }

    #[test]
    fn m008_concealed_consumption_refuses_commit() {
        // The attempt durably consumed a provisional output but offers
        // WITHOUT declaring that ancestor: H028 would see an empty set
        // and pass — the obligation cross-check is what catches it.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut manifests = std::collections::BTreeMap::new();
        plant_producer(
            &mut store,
            &mut manifests,
            100,
            120,
            "out/a.rmeta",
            110,
            800,
            vec![],
        );
        let pin_key = "test-pin-concealed".to_owned();
        store
            .record_provisional_consumption(&ProvisionalObligationInsert {
                consumer_worker: "worker-a".to_owned(),
                consumer_attempt: 20,
                pin_key: pin_key.clone(),
                producer_action_key: digest_key(&digest("rabs.action-key.sha256.v1", 100)),
                producer_generation: 11,
                producer_attempt: 20,
                role_tag: 2,
                virtual_path: b"out/a.rmeta".to_vec(),
                object_key: digest_key(&object(110).0),
                created_seq: 1,
            })
            .unwrap();
        let resolver = {
            let manifests = manifests.clone();
            move |key: &str| manifests.get(key).cloned()
        };
        let c_offer = offer_with_ancestors(vec![]);
        assert_eq!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap_err(),
            OfferRefusal::UndeclaredProvisionalConsumption { pin_key }
        );
    }

    #[test]
    fn m008_cancelled_lineage_permanently_refuses_commit() {
        // Producer committed AND honestly declared — but its lineage was
        // CANCELLED (failed/superseded): the descendant can never publish.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut manifests = std::collections::BTreeMap::new();
        plant_producer(
            &mut store,
            &mut manifests,
            100,
            120,
            "out/a.rmeta",
            110,
            800,
            vec![],
        );
        let pin_key = "test-pin-cancelled".to_owned();
        store
            .record_provisional_consumption(&ProvisionalObligationInsert {
                consumer_worker: "worker-a".to_owned(),
                consumer_attempt: 20,
                pin_key: pin_key.clone(),
                producer_action_key: digest_key(&digest("rabs.action-key.sha256.v1", 100)),
                producer_generation: 11,
                producer_attempt: 20,
                role_tag: 2,
                virtual_path: b"out/a.rmeta".to_vec(),
                object_key: digest_key(&object(110).0),
                created_seq: 1,
            })
            .unwrap();
        store.cancel_provisional_obligations(&pin_key).unwrap();
        let resolver = {
            let manifests = manifests.clone();
            move |key: &str| manifests.get(key).cloned()
        };
        let c_offer = offer_with_ancestors(vec![ancestor_ref(100, "out/a.rmeta", 110)]);
        assert_eq!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap_err(),
            OfferRefusal::ConsumptionLineageCancelled { pin_key }
        );
    }

    #[test]
    fn h028_transitive_hole_refuses_where_direct_only_would_commit() {
        // B is committed, but B's recorded lineage names producer 102
        // which never committed: C's direct check on B alone would pass
        // — the transitive walk must refuse (I32).
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut manifests = std::collections::BTreeMap::new();
        plant_producer(
            &mut store,
            &mut manifests,
            101,
            121,
            "out/b.rmeta",
            111,
            801,
            vec![ProvisionalAncestorRow {
                producer_action_key: digest_key(&digest("rabs.action-key.sha256.v1", 102)),
                role: "provisional-metadata".to_owned(),
                virtual_path: b"out/x.rmeta".to_vec(),
                object_key: digest_key(&object(112).0),
                adopted: false,
            }],
        );
        let resolver = move |key: &str| manifests.get(key).cloned();
        let c_offer = offer_with_ancestors(vec![ancestor_ref(101, "out/b.rmeta", 111)]);
        assert_eq!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::ProvisionalProducerNotCommitted {
                producer: digest_key(&digest("rabs.action-key.sha256.v1", 102)),
            })
        );
        // Nothing committed for C.
        assert!(
            !store
                .has_publication(&digest("rabs.action-key.sha256.v1", 7))
                .unwrap()
        );
    }

    #[test]
    fn h028_divergent_consumed_object_refuses_until_adoption_edge() {
        // C consumed object 119 from B's losing attempt, but B's winning
        // attempt committed object 111 at the same logical output: typed
        // refusal (R64) — until the coordinator records the explicit
        // adoption edge 119 -> 111, after which the re-offer commits
        // with adopted=true lineage.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut manifests = std::collections::BTreeMap::new();
        plant_producer(
            &mut store,
            &mut manifests,
            101,
            121,
            "out/b.rmeta",
            111,
            801,
            vec![],
        );
        let resolver = {
            let manifests = manifests.clone();
            move |key: &str| manifests.get(key).cloned()
        };
        let b_key = digest_key(&digest("rabs.action-key.sha256.v1", 101));
        let c_offer = offer_with_ancestors(vec![ancestor_ref(101, "out/b.rmeta", 119)]);
        assert_eq!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver.clone(),
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::DivergentProvisionalAncestor {
                producer: b_key.clone(),
                path: "out/b.rmeta".to_owned(),
            })
        );

        // Authority-gated adoption edge; conflicting rewrite refused.
        let auth = authority_digest(&coordinator_authority());
        store
            .record_adoption_edge(
                &auth,
                &b_key,
                "provisional-metadata",
                b"out/b.rmeta",
                &digest_key(&object(119).0),
                &digest_key(&object(111).0),
            )
            .unwrap();
        assert_eq!(
            store.record_adoption_edge(
                &auth,
                &b_key,
                "provisional-metadata",
                b"out/b.rmeta",
                &digest_key(&object(119).0),
                &digest_key(&object(112).0),
            ),
            Err(StoreError::AdoptionEdgeConflict)
        );

        assert!(matches!(
            process_offer(
                &mut store,
                &c_offer,
                &expected_descriptor(),
                resolver,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));
        let c_key = digest_key(&digest("rabs.action-key.sha256.v1", 7));
        let recorded = store.list_provisional_ancestors(&c_key).unwrap();
        assert_eq!(recorded.len(), 1);
        assert!(
            recorded[0].adopted,
            "adoption edge must be recorded as such"
        );
    }

    #[test]
    fn h028_canonical_ancestor_set_refuses_duplicates_and_self() {
        let manifest_id = object(50);
        assert_eq!(
            OfferPreparedActionResult::build(
                attempt_authority(),
                manifest(),
                manifest_id.clone(),
                evidence(&manifest_id),
                object(51),
                digest("rabs.observation-stream.sha256.v1", 9),
                &declared(),
                vec![
                    ancestor_ref(101, "out/b.rmeta", 111),
                    ancestor_ref(101, "out/b.rmeta", 119),
                ],
            ),
            Err(OfferBuildError::DuplicateAncestorRef {
                path: "out/b.rmeta".to_owned()
            })
        );
        assert_eq!(
            OfferPreparedActionResult::build(
                attempt_authority(),
                manifest(),
                manifest_id.clone(),
                evidence(&manifest_id),
                object(51),
                digest("rabs.observation-stream.sha256.v1", 9),
                &declared(),
                vec![ancestor_ref(7, "out/self.rmeta", 111)],
            ),
            Err(OfferBuildError::SelfAncestorRef)
        );
        // The canonical set is SORTED regardless of input order.
        let shuffled = OfferPreparedActionResult::build(
            attempt_authority(),
            manifest(),
            manifest_id.clone(),
            evidence(&manifest_id),
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            vec![
                ancestor_ref(102, "out/x.rmeta", 112),
                ancestor_ref(101, "out/b.rmeta", 111),
            ],
        )
        .unwrap();
        let ordered = OfferPreparedActionResult::build(
            attempt_authority(),
            manifest(),
            manifest_id.clone(),
            evidence(&manifest_id),
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            vec![
                ancestor_ref(101, "out/b.rmeta", 111),
                ancestor_ref(102, "out/x.rmeta", 112),
            ],
        )
        .unwrap();
        assert_eq!(shuffled, ordered, "ancestor order never changes identity");
    }

    #[test]
    fn h036_publication_pin_is_coordinator_owned_and_worker_release_refuses() {
        // T039/T049 seed at this layer (R111/R127). The H015 crash
        // matrix already proves publication ⟺ pin atomicity at every
        // kill point; this fixture pins the pin's POSTURE and the
        // release-attempt surface.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        assert!(matches!(
            process_offer(
                &mut store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));
        // The pin exists, unreleased, coordinator-owned, rooted at the
        // committed manifest, in the action-publication class, with NO
        // expiry (durable — never lease-decayed).
        let pin = store.pin_row(900).unwrap().unwrap();
        assert_eq!(pin.owner, "coordinator");
        assert_eq!(pin.class, "action-publication");
        assert_eq!(pin.root_key, digest_key(&object(50).0));
        assert!(!pin.released);
        assert_eq!(pin.expires_at_seq, None, "publication pins never expire");

        // R127: WORKERS can never release it — any non-coordinator
        // identity refuses with a typed owner mismatch, and repeated
        // attempts change nothing.
        for intruder in ["worker-a", "worker-b", "rabs-wkr:22", ""] {
            assert_eq!(
                store.release_pin(900, intruder),
                Err(StoreError::PinOwnerMismatch),
                "worker identity {intruder:?} must be refused"
            );
        }
        let after = store.pin_row(900).unwrap().unwrap();
        assert!(!after.released, "refused releases must not release");
        assert_eq!(after, pin, "refused releases must change nothing");

        // Releasing a pin that does not exist is UnknownPin, not a
        // silent no-op — a worker cannot even probe usefully.
        assert_eq!(
            store.release_pin(999, "coordinator"),
            Err(StoreError::UnknownPin)
        );

        // The coordinator CAN release (decommission path exists and is
        // owner-gated, proving the refusals above are about ownership,
        // not a frozen table).
        store.release_pin(900, "coordinator").unwrap();
        assert!(store.pin_row(900).unwrap().unwrap().released);
    }

    #[test]
    fn h035_equal_digests_different_manifest_opens_projection_completeness_incident() {
        // THE T035 fixture (I48; R103): two candidates whose semantic
        // AND observable digests are IDENTICAL — the manifest content
        // is byte-for-byte the same logical projection — but whose
        // canonical manifest OBJECTS differ (a serializer emitted
        // different bytes for the same content, or the projection
        // omitted a field that differs). That is never "evidence-only
        // variation": it is a serializer/projection-completeness
        // incident and the action quarantines.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let first = offer();
        assert!(matches!(
            process_offer(
                &mut store,
                &first,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));

        // Same manifest CONTENT (identical projections → identical
        // recomputed digests), different manifest OBJECT id.
        let doppel_id = object(52);
        store.record_object(&doppel_id.0, 64).unwrap();
        store
            .add_location(&doppel_id.0, "/cas/52", Some(1), "raw", true)
            .unwrap();
        let doppel = OfferPreparedActionResult::build(
            attempt_authority(),
            manifest(),
            doppel_id.clone(),
            evidence(&doppel_id),
            object(54),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        store.record_object(&object(54).0, 64).unwrap();
        store
            .add_location(&object(54).0, "/cas/54", Some(1), "raw", true)
            .unwrap();
        assert_eq!(
            doppel.manifest.semantic_result_digest, first.manifest.semantic_result_digest,
            "fixture precondition: equal semantic digests"
        );
        assert_eq!(
            doppel.manifest.observable_result_digest, first.manifest.observable_result_digest,
            "fixture precondition: equal observable digests"
        );
        assert_ne!(doppel.manifest_id, first.manifest_id);

        let committed = first.manifest.clone();
        let outcome = process_offer(
            &mut store,
            &doppel,
            &expected_descriptor(),
            move |_| Some(committed.clone()),
            905,
            6,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        // The RIGHT incident class — and specifically NOT the
        // idempotent/evidence-append path (R103's mislabel) and NOT
        // the narrower presentation quarantine.
        let PublicationOutcome::Quarantined(quarantine) = outcome else {
            panic!("equal-digests/different-manifest must quarantine, got {outcome:?}");
        };
        assert_eq!(
            quarantine.class,
            DivergenceClass::ProjectionCompletenessIncident
        );
        assert!(
            quarantine.escalations.is_empty(),
            "consumer escalation is the semantic-divergence arm only"
        );
        let action = digest("rabs.action-key.sha256.v1", 7);
        let action_key = digest_key(&action);
        // FULL serving disable (not presentation-quarantined).
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_QUARANTINED
        );
        // The incident row records the class and BOTH manifests.
        let incidents = store.list_divergence_incidents(&action_key).unwrap();
        assert_eq!(incidents.len(), 1);
        assert_eq!(incidents[0].class, "projection-completeness");
        assert_eq!(
            incidents[0].committed_manifest_key,
            digest_key(&first.manifest_id.0)
        );
        assert_eq!(
            incidents[0].candidate_manifest_key,
            digest_key(&doppel_id.0)
        );
        // Committed row preserved untouched; candidate preserved under
        // the divergence-evidence pin.
        assert_eq!(
            store.published_manifest_key(&action).unwrap().unwrap(),
            digest_key(&first.manifest_id.0)
        );
        let pin = store.pin_row(905).unwrap().unwrap();
        assert_eq!(pin.class, DIVERGENCE_EVIDENCE_PIN_CLASS);
        assert_eq!(pin.root_key, digest_key(&doppel_id.0));
        // The candidate's evidence is preserved, bound to the
        // CANDIDATE manifest (H029) — recorded, but never presented
        // as ordinary evidence-only variation for the committed one.
        let candidate_view = store
            .list_evidence_keys_for_manifest(&digest_key(&doppel_id.0))
            .unwrap();
        assert!(candidate_view.contains(&digest_key(&object(54).0)));
        let committed_view = store
            .list_evidence_keys_for_manifest(&digest_key(&first.manifest_id.0))
            .unwrap();
        assert!(!committed_view.contains(&digest_key(&object(54).0)));
    }

    #[test]
    fn h039_bundle_root_and_role_uniqueness_are_enforced_at_both_gates() {
        // T047 rejection fixtures (R125).
        // Gate 1 — worker build(): a duplicate (role, path) row is a
        // typed structural refusal (the A031 validate runs before the
        // F035 root check; the BundleRoot(Structural) arm guards
        // against the two validators drifting); the offer never
        // exists either way.
        let manifest_id = object(50);
        let mut duplicated = manifest();
        duplicated
            .logical_outputs
            .push(duplicated.logical_outputs[0].clone());
        assert!(matches!(
            OfferPreparedActionResult::build(
                attempt_authority(),
                duplicated,
                manifest_id.clone(),
                evidence(&manifest_id),
                object(51),
                digest("rabs.observation-stream.sha256.v1", 9),
                &[declared()[0].clone(), declared()[0].clone()],
                Vec::new(),
            ),
            Err(OfferBuildError::ManifestInvalid(_))
        ));

        // build() STAMPS the derived root: whatever garbage the
        // fixture manifest carried, the built offer's root is the
        // F035 derivation.
        let built = offer();
        assert_eq!(
            built.manifest.artifact_bundle_root,
            rabs_key::logical_output_map::compute_bundle_root(&built.manifest.logical_outputs),
            "worker cannot ship a contradictory root"
        );

        // Gate 2 — coordinator admission: a tampered root (re-stamped
        // digests, so digest recompute alone would PASS) is refused by
        // the independent bundle-root recompute, before any write.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let mut tampered = offer();
        tampered.manifest.artifact_bundle_root = Some(object(99));
        tampered.manifest.semantic_result_digest = semantic_result_digest_v1(&tampered.manifest);
        tampered.manifest.observable_result_digest = observable_result_digest_v1(
            &tampered.manifest,
            &digest("rabs.observation-stream.sha256.v1", 9),
        );
        assert!(matches!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::BundleRootMismatch { .. })
        ));
        let action = digest("rabs.action-key.sha256.v1", 7);
        assert!(
            !store.has_publication(&action).unwrap(),
            "refusals write nothing"
        );

        // A root MISSING where outputs exist refuses the same way.
        let mut rootless = offer();
        rootless.manifest.artifact_bundle_root = None;
        rootless.manifest.semantic_result_digest = semantic_result_digest_v1(&rootless.manifest);
        rootless.manifest.observable_result_digest = observable_result_digest_v1(
            &rootless.manifest,
            &digest("rabs.observation-stream.sha256.v1", 9),
        );
        assert!(matches!(
            process_offer(
                &mut store,
                &rootless,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::BundleRootMismatch { .. })
        ));

        // The untampered offer commits: the gates reject exactly the
        // contradictions, not the happy path.
        assert!(matches!(
            process_offer(
                &mut store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));
    }

    #[test]
    fn h032_commit_ack_gates_on_the_configured_durability_profile() {
        // A located-but-volatile closure object refuses the commit under
        // the authoritative profile with a typed refusal naming it.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        store
            .add_location(&object(61).0, "/cas/61", Some(1), "raw", false)
            .unwrap();
        assert_eq!(
            process_offer(
                &mut store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::ObjectNotDurable {
                missing: digest_key(&object(61).0),
            })
        );
        // The refusal wrote nothing: no publication, no ack.
        let action = digest("rabs.action-key.sha256.v1", 7);
        assert!(!store.has_publication(&action).unwrap());

        // POWER-FAIL SIMULATION over the refused state: the volatile
        // copy is lost (quarantined as suspect on reopen). There is no
        // committed pointer to the lost object — the gate held.
        store
            .set_location_quarantined(&object(61).0, "/cas/61", true)
            .unwrap();
        assert!(!store.has_publication(&action).unwrap());

        // An explicitly volatile deployment admits the same offer — the
        // relaxation is NAMED, never the silent default.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut volatile_store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut volatile_store);
        volatile_store
            .add_location(&object(61).0, "/cas/61", Some(1), "raw", false)
            .unwrap();
        assert!(matches!(
            process_offer(
                &mut volatile_store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::AcceptVolatileLocations,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));

        // Durable closure commits under the authoritative profile, and a
        // power failure afterward cannot orphan the pointer: every
        // closure object still has a durable location.
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut durable_store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut durable_store);
        assert!(matches!(
            process_offer(
                &mut durable_store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));
        assert!(durable_store.has_publication(&action).unwrap());
        for tag in [40, 41, 50, 51, 60, 61, 62, 63] {
            assert!(
                durable_store
                    .object_durably_located(&object(tag).0)
                    .unwrap(),
                "committed closure object {tag} must be durable"
            );
        }
    }

    #[test]
    fn h011_worker_build_validates_and_stamps_projection_digests() {
        let built = offer();
        // The stamped digests ARE the recomputable projections.
        assert_eq!(
            built.manifest.semantic_result_digest,
            semantic_result_digest_v1(&built.manifest)
        );
        assert_eq!(
            built.manifest.observable_result_digest,
            observable_result_digest_v1(&built.manifest, &built.canonical_observations)
        );

        // Undeclared output refused.
        let manifest_id = object(50);
        assert_eq!(
            OfferPreparedActionResult::build(
                attempt_authority(),
                manifest(),
                manifest_id.clone(),
                evidence(&manifest_id),
                object(51),
                digest("rabs.observation-stream.sha256.v1", 9),
                &[],
                Vec::new(),
            ),
            Err(OfferBuildError::UndeclaredOutput {
                path: "out/lib.rlib".to_owned()
            })
        );

        // Evidence bound to a different manifest refused.
        assert_eq!(
            OfferPreparedActionResult::build(
                attempt_authority(),
                manifest(),
                manifest_id,
                evidence(&object(99)),
                object(51),
                digest("rabs.observation-stream.sha256.v1", 9),
                &declared(),
                Vec::new(),
            ),
            Err(OfferBuildError::EvidenceManifestMismatch)
        );

        // Structurally invalid manifest refused (deterministic failure
        // with outputs).
        let mut bad = manifest();
        bad.result_kind = ResultKind::DeterministicFailure;
        let manifest_id = object(50);
        assert!(matches!(
            OfferPreparedActionResult::build(
                attempt_authority(),
                bad,
                manifest_id.clone(),
                evidence(&manifest_id),
                object(51),
                digest("rabs.observation-stream.sha256.v1", 9),
                &declared(),
                Vec::new(),
            ),
            Err(OfferBuildError::ManifestInvalid(_))
        ));
    }

    #[test]
    fn h011_commit_writes_publication_evidence_and_pin_atomically() {
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let outcome = process_offer(
            &mut store,
            &offer(),
            &expected_descriptor(),
            no_committed,
            900,
            42,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        let PublicationOutcome::Committed(receipt) = outcome else {
            panic!("expected commit, got {outcome:?}");
        };
        assert_eq!(receipt.committed_causal_sequence, 42);
        assert_eq!(receipt.canonical_result_manifest_id, object(50));
        assert_eq!(receipt.winner_evidence_bundle_id, object(51));
        assert!(
            store
                .has_publication(&digest("rabs.action-key.sha256.v1", 7))
                .unwrap()
        );
        let snapshot = store.differential_snapshot().unwrap();
        // Publication pin + evidence row landed with the commit.
        assert!(snapshot.iter().any(|l| l.starts_with("pins|")
            && l.contains("action-publication")
            && l.contains("coordinator")));
        assert!(
            snapshot
                .iter()
                .any(|l| l.starts_with("action_evidence_index|"))
        );
    }

    #[test]
    fn h011_fence_refusals_write_nothing() {
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();

        // No active authority at all.
        assert_eq!(
            process_offer(
                &mut store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::NotActiveAuthority)
        );

        ready_store(&mut store);

        // F033: generation bound to a DIFFERENT authority digest.
        let mut tampered = offer();
        tampered
            .authority
            .action_generation
            .created_under_authority_digest = digest(AUTHORITY_DIGEST_DOMAIN, 99);
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::GenerationAuthorityMismatch)
        );

        // Unknown generation.
        let mut tampered = offer();
        tampered.authority.action_generation.generation_id = ActionGenerationId(999);
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::UnknownGeneration)
        );

        // Unknown attempt.
        let mut tampered = offer();
        tampered.authority.attempt_id = AttemptId(999);
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::UnknownAttempt)
        );

        // Released lease.
        store.release_lease(30).unwrap();
        assert_eq!(
            process_offer(
                &mut store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::LeaseReleased)
        );
        // Nothing was published by any refusal.
        assert!(
            !store
                .has_publication(&digest("rabs.action-key.sha256.v1", 7))
                .unwrap()
        );
    }

    #[test]
    fn h011_expired_lease_refuses_publication_and_late_renewal_without_mutation() {
        fn scenario(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let prepared = offer();
            let before = store.differential_snapshot().unwrap();
            for now in [100, 101, u64::MAX] {
                for durability in [
                    CommitDurabilityProfile::RequireDurableClosure,
                    CommitDurabilityProfile::AcceptVolatileLocations,
                ] {
                    assert_eq!(
                        process_offer(
                            store,
                            &prepared,
                            &expected_descriptor(),
                            no_committed,
                            900,
                            1,
                            durability,
                            || now,
                        ),
                        Err(OfferRefusal::LeaseExpired)
                    );
                }
                assert_eq!(
                    store.validate_attempt_lease(&prepared.authority, now),
                    Err(StoreError::LeaseExpired)
                );
            }

            // Advancing the renewal sequence at the exact deadline must
            // never revive the expired attempt or acquire publication rights.
            assert_eq!(
                store.renew_attempt_lease(
                    &prepared.authority,
                    LeaseRenewal {
                        lease: prepared.authority.execution_lease_id,
                        seq: LeaseRenewalSeq(2),
                    },
                    200,
                    &|| 100,
                ),
                Err(StoreError::LeaseExpired)
            );
            assert!(
                !store
                    .has_publication(&prepared.authority.action_key)
                    .unwrap()
            );
            assert!(
                store
                    .list_evidence_keys(&prepared.authority.action_key)
                    .unwrap()
                    .is_empty()
            );
            for tag in [40, 41, 50, 51, 60, 61, 62, 63] {
                assert!(store.object_located(&object(tag).0).unwrap());
            }
            assert!(
                store
                    .object_located(&prepared.manifest.artifact_bundle_root.as_ref().unwrap().0)
                    .unwrap()
            );
            let after = store.differential_snapshot().unwrap();
            assert_eq!(after, before, "expiry must preserve uploaded candidates");
            after
        }

        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&fresh_path("expiry-ref")).unwrap())
                .unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&fresh_path("expiry-fsq")).unwrap())
                .unwrap();
        assert_eq!(scenario(&mut reference), scenario(&mut candidate));
    }

    #[test]
    fn h011_lease_expiring_during_admission_cannot_commit() {
        fn scenario(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let prepared = offer();
            let before = store.differential_snapshot().unwrap();
            let readings = std::cell::Cell::new(0);
            let clock = || {
                let reading = readings.get();
                readings.set(reading + 1);
                if reading == 0 { 99 } else { 100 }
            };
            assert_eq!(
                process_offer(
                    store,
                    &prepared,
                    &expected_descriptor(),
                    no_committed,
                    900,
                    1,
                    CommitDurabilityProfile::RequireDurableClosure,
                    clock,
                ),
                Err(OfferRefusal::LeaseExpired)
            );
            assert!(
                !store
                    .has_publication(&prepared.authority.action_key)
                    .unwrap()
            );
            assert!(
                store
                    .list_evidence_keys(&prepared.authority.action_key)
                    .unwrap()
                    .is_empty()
            );
            assert!(store.pin_row(900).unwrap().is_none());
            assert!(store.object_located(&prepared.manifest_id.0).unwrap());
            assert!(store.object_located(&prepared.evidence_id.0).unwrap());
            let after = store.differential_snapshot().unwrap();
            assert_eq!(after, before);
            after
        }

        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&fresh_path("late-expiry-ref")).unwrap())
                .unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&fresh_path("late-expiry-fsq")).unwrap())
                .unwrap();
        assert_eq!(scenario(&mut reference), scenario(&mut candidate));
    }

    #[test]
    fn h011_publication_resamples_clock_and_rolls_back_transaction_time_expiry() {
        use crate::metadata_store::{SqlEngine, SqlValue};
        use std::cell::Cell;

        // Advance the clock at a real backend boundary: opening the
        // transaction, or writing the publication row. The latter must roll
        // back the entire pointer/evidence/pin transaction on expiry.
        struct ExpireDuringTransaction<'a, E> {
            inner: E,
            clock: &'a Cell<u64>,
            armed: bool,
            trigger: &'static str,
        }

        impl<E: SqlEngine> SqlEngine for ExpireDuringTransaction<'_, E> {
            fn execute(&mut self, sql: &str, params: &[SqlValue]) -> Result<usize, StoreError> {
                let affected = self.inner.execute(sql, params)?;
                if self.armed && sql.starts_with(self.trigger) {
                    self.clock.set(100);
                    self.armed = false;
                }
                Ok(affected)
            }

            fn query(
                &mut self,
                sql: &str,
                params: &[SqlValue],
            ) -> Result<Vec<Vec<SqlValue>>, StoreError> {
                self.inner.query(sql, params)
            }
        }

        fn scenario(engine: impl SqlEngine, trigger: &'static str) -> Vec<String> {
            let clock = Cell::new(99);
            let engine = ExpireDuringTransaction {
                inner: engine,
                clock: &clock,
                armed: false,
                trigger,
            };
            let mut store = SqlMetadataStore::open(engine).unwrap();
            ready_store(&mut store);
            let prepared = offer();
            let row = PublicationRow {
                action_key: prepared.authority.action_key.clone(),
                descriptor_digest: expected_descriptor(),
                manifest_digest: prepared.manifest_id.0.clone(),
                evidence_digest: prepared.evidence_id.0.clone(),
                winner_generation: prepared.authority.action_generation.generation_id.0,
                winner_attempt: prepared.authority.attempt_id.0,
                result_kind: ResultKindTag::Success,
                pin_id: 900,
                pin_owner: "coordinator".to_owned(),
                provisional_ancestors: Vec::new(),
            };
            let before = store.differential_snapshot().unwrap();
            store
                .validate_attempt_lease(&prepared.authority, clock.get())
                .unwrap();
            store.engine_mut().armed = true;
            assert_eq!(
                store.commit_publication(
                    &authority_digest(&prepared.authority.coordinator),
                    Some((&prepared.authority, &|| clock.get())),
                    &row,
                ),
                Err(StoreError::LeaseExpired)
            );
            assert_eq!(clock.get(), 100);
            assert!(!store.has_publication(&row.action_key).unwrap());
            assert!(store.pin_row(row.pin_id).unwrap().is_none());
            assert!(
                store
                    .list_evidence_keys(&row.action_key)
                    .unwrap()
                    .is_empty()
            );
            let after = store.differential_snapshot().unwrap();
            assert_eq!(after, before);
            after
        }

        for trigger in ["BEGIN", "INSERT INTO action_publications"] {
            assert_eq!(
                scenario(
                    RusqliteEngine::open(&fresh_path("txn-expiry-ref")).unwrap(),
                    trigger,
                ),
                scenario(
                    FsqliteEngine::open(&fresh_path("txn-expiry-fsq")).unwrap(),
                    trigger,
                ),
            );
        }
    }

    #[test]
    fn h011_same_key_updates_expiring_inside_transactions_preserve_existing_state() {
        use crate::metadata_store::{SqlEngine, SqlValue};
        use std::cell::Cell;

        #[derive(Clone, Copy)]
        enum SameKeyUpdate {
            AppendEvidence,
            SemanticQuarantine,
            ObservableAlreadyQuarantined,
        }

        struct ExpireDuringTransaction<'a, E> {
            inner: E,
            clock: &'a Cell<u64>,
            armed: bool,
            trigger: &'static str,
        }

        impl<E: SqlEngine> SqlEngine for ExpireDuringTransaction<'_, E> {
            fn execute(&mut self, sql: &str, params: &[SqlValue]) -> Result<usize, StoreError> {
                let affected = self.inner.execute(sql, params)?;
                if self.armed && sql.starts_with(self.trigger) {
                    self.clock.set(100);
                    self.armed = false;
                }
                Ok(affected)
            }

            fn query(
                &mut self,
                sql: &str,
                params: &[SqlValue],
            ) -> Result<Vec<Vec<SqlValue>>, StoreError> {
                self.inner.query(sql, params)
            }
        }

        fn scenario(
            engine: impl SqlEngine,
            update: SameKeyUpdate,
            trigger: &'static str,
        ) -> Vec<String> {
            let clock = Cell::new(99);
            let engine = ExpireDuringTransaction {
                inner: engine,
                clock: &clock,
                armed: false,
                trigger,
            };
            let mut store = SqlMetadataStore::open(engine).unwrap();
            ready_store(&mut store);
            let winner = offer();
            assert!(matches!(
                process_offer(
                    &mut store,
                    &winner,
                    &expected_descriptor(),
                    no_committed,
                    900,
                    1,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || clock.get(),
                )
                .unwrap(),
                PublicationOutcome::Committed(_)
            ));
            let action = winner.authority.action_key.clone();
            let action_key = digest_key(&action);
            if matches!(update, SameKeyUpdate::ObservableAlreadyQuarantined) {
                store
                    .set_serving_disposition_key(&action_key, DISPOSITION_QUARANTINED)
                    .unwrap();
            }
            let candidate = match update {
                SameKeyUpdate::AppendEvidence => divergent_offer(
                    &mut store,
                    None,
                    50,
                    55,
                    digest("rabs.observation-stream.sha256.v1", 9),
                ),
                SameKeyUpdate::SemanticQuarantine => divergent_offer(
                    &mut store,
                    Some(42),
                    52,
                    55,
                    digest("rabs.observation-stream.sha256.v1", 9),
                ),
                SameKeyUpdate::ObservableAlreadyQuarantined => divergent_offer(
                    &mut store,
                    None,
                    52,
                    55,
                    digest("rabs.observation-stream.sha256.v1", 10),
                ),
            };
            let evidence_before = store.list_evidence_keys(&action).unwrap();
            assert!(!evidence_before.contains(&digest_key(&candidate.evidence_id.0)));
            let serving_before = store.serving_disposition_key(&action_key).unwrap();
            let expected_disposition =
                if matches!(update, SameKeyUpdate::ObservableAlreadyQuarantined) {
                    DISPOSITION_QUARANTINED
                } else {
                    "servable"
                };
            assert_eq!(serving_before.as_deref(), Some(expected_disposition));
            let before = store.differential_snapshot().unwrap();

            // Candidate admission remains live at 99. The real backend then
            // crosses the deadline while entering or writing the mutation
            // transaction. Neither new trust evidence nor quarantine may land.
            store.engine_mut().armed = true;
            let committed = winner.manifest.clone();
            assert_eq!(
                process_offer(
                    &mut store,
                    &candidate,
                    &expected_descriptor(),
                    move |_| Some(committed.clone()),
                    901,
                    2,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || clock.get(),
                ),
                Err(OfferRefusal::LeaseExpired)
            );
            assert_eq!(clock.get(), 100, "expiry boundary must be exercised");
            assert_eq!(
                store.published_manifest_key(&action).unwrap(),
                Some(digest_key(&winner.manifest_id.0))
            );
            assert_eq!(
                store.serving_disposition_key(&action_key).unwrap(),
                serving_before
            );
            assert_eq!(store.list_evidence_keys(&action).unwrap(), evidence_before);
            assert!(store.pin_row(901).unwrap().is_none());
            assert!(
                store
                    .list_divergence_incidents(&action_key)
                    .unwrap()
                    .is_empty()
            );
            assert!(store.object_located(&candidate.manifest_id.0).unwrap());
            assert!(store.object_located(&candidate.evidence_id.0).unwrap());
            for output in &candidate.manifest.logical_outputs {
                assert!(store.object_located(&output.object.0).unwrap());
            }
            assert!(
                store
                    .object_located(&candidate.manifest.artifact_bundle_root.as_ref().unwrap().0)
                    .unwrap()
            );
            let after = store.differential_snapshot().unwrap();
            assert_eq!(
                after, before,
                "expired same-key offers must leave no writes"
            );
            after
        }

        for (update, trigger) in [
            (SameKeyUpdate::AppendEvidence, "BEGIN"),
            (
                SameKeyUpdate::AppendEvidence,
                "INSERT OR IGNORE INTO action_evidence_index",
            ),
            (SameKeyUpdate::SemanticQuarantine, "BEGIN"),
            (
                SameKeyUpdate::SemanticQuarantine,
                "UPDATE action_serving_states",
            ),
            (SameKeyUpdate::ObservableAlreadyQuarantined, "BEGIN"),
            (
                SameKeyUpdate::ObservableAlreadyQuarantined,
                "UPDATE action_serving_states",
            ),
        ] {
            assert_eq!(
                scenario(
                    RusqliteEngine::open(&fresh_path("same-key-expiry-ref")).unwrap(),
                    update,
                    trigger,
                ),
                scenario(
                    FsqliteEngine::open(&fresh_path("same-key-expiry-fsq")).unwrap(),
                    update,
                    trigger,
                ),
            );
        }
    }

    #[test]
    fn h011_live_lease_commits_just_before_its_monotonic_deadline() {
        fn scenario(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let prepared = offer();
            let outcome = process_offer(
                store,
                &prepared,
                &expected_descriptor(),
                no_committed,
                900,
                10_000,
                CommitDurabilityProfile::RequireDurableClosure,
                || 99,
            )
            .unwrap();
            let PublicationOutcome::Committed(receipt) = outcome else {
                panic!("expected commit before the lease deadline, got {outcome:?}");
            };
            // Causal sequence numbers must not be used as clock readings.
            assert_eq!(receipt.committed_causal_sequence, 10_000);
            assert!(
                store
                    .has_publication(&prepared.authority.action_key)
                    .unwrap()
            );
            store.differential_snapshot().unwrap()
        }

        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&fresh_path("live-lease-ref")).unwrap())
                .unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&fresh_path("live-lease-fsq")).unwrap())
                .unwrap();
        assert_eq!(scenario(&mut reference), scenario(&mut candidate));
    }

    #[test]
    fn h011_renewed_lease_survives_reopen_and_allows_publication_until_new_deadline() {
        fn prepare(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let authority = attempt_authority();
            store
                .renew_attempt_lease(
                    &authority,
                    LeaseRenewal {
                        lease: authority.execution_lease_id,
                        seq: LeaseRenewalSeq(2),
                    },
                    200,
                    &|| 99,
                )
                .unwrap();
            store.differential_snapshot().unwrap()
        }

        fn publish(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            let mut renewed = offer();
            renewed.authority.lease_renewal_seq = LeaseRenewalSeq(2);
            let state = store
                .validate_attempt_lease(&renewed.authority, 150)
                .unwrap();
            assert_eq!(state.renewal_seq, 2);
            assert_eq!(state.expires_at_own_monotonic_ms, 200);
            assert!(matches!(
                process_offer(
                    store,
                    &renewed,
                    &expected_descriptor(),
                    no_committed,
                    900,
                    1,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 150,
                )
                .unwrap(),
                PublicationOutcome::Committed(_)
            ));
            assert!(
                store
                    .has_publication(&renewed.authority.action_key)
                    .unwrap()
            );
            let committed = store.differential_snapshot().unwrap();
            assert_eq!(
                process_offer(
                    store,
                    &renewed,
                    &expected_descriptor(),
                    no_committed,
                    901,
                    2,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 200,
                ),
                Err(OfferRefusal::LeaseExpired)
            );
            assert_eq!(store.differential_snapshot().unwrap(), committed);
            committed
        }

        let reference_path = fresh_path("renewed-lease-ref");
        let candidate_path = fresh_path("renewed-lease-fsq");
        let prepared = {
            let mut reference =
                SqlMetadataStore::open(RusqliteEngine::open(&reference_path).unwrap()).unwrap();
            let mut candidate =
                SqlMetadataStore::open(FsqliteEngine::open(&candidate_path).unwrap()).unwrap();
            let prepared = prepare(&mut reference);
            assert_eq!(prepared, prepare(&mut candidate));
            prepared
        };
        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&reference_path).unwrap()).unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&candidate_path).unwrap()).unwrap();
        assert_eq!(reference.differential_snapshot().unwrap(), prepared);
        assert_eq!(candidate.differential_snapshot().unwrap(), prepared);
        assert_eq!(publish(&mut reference), publish(&mut candidate));
    }

    #[test]
    fn h011_tombstoned_generation_refused() {
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        store.tombstone_generation(11).unwrap();
        assert_eq!(
            process_offer(
                &mut store,
                &offer(),
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::GenerationTombstoned)
        );
    }

    #[test]
    fn h011_content_validation_refusals() {
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);

        // Descriptor byte-compare.
        assert_eq!(
            process_offer(
                &mut store,
                &offer(),
                &digest("rabs.descriptor.sha256.v1", 99),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::DescriptorMismatch)
        );

        // Epoch mismatch against the action entry.
        let mut tampered = offer();
        tampered.manifest.key_epoch = 2;
        tampered.manifest.semantic_result_digest = semantic_result_digest_v1(&tampered.manifest);
        tampered.manifest.observable_result_digest =
            observable_result_digest_v1(&tampered.manifest, &tampered.canonical_observations);
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::EpochMismatch)
        );

        // Declared semantic digest disagrees with independent recompute.
        let mut tampered = offer();
        tampered.manifest.semantic_result_digest = digest(SEMANTIC_PROJECTION_DOMAIN, 99);
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::SemanticDigestMismatch)
        );

        // Declared observable digest disagrees.
        let mut tampered = offer();
        tampered.manifest.observable_result_digest = digest(OBSERVABLE_PROJECTION_DOMAIN, 99);
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::ObservableDigestMismatch)
        );
    }

    #[test]
    fn h011_incomplete_object_closure_refused() {
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        // Take one evidence constituent's location away by using an offer
        // referencing an object that was never located.
        let manifest_id = object(50);
        let mut tampered_evidence = evidence(&manifest_id);
        tampered_evidence.provenance_receipt = object(200);
        let tampered = OfferPreparedActionResult::build(
            attempt_authority(),
            manifest(),
            manifest_id,
            tampered_evidence,
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        assert_eq!(
            process_offer(
                &mut store,
                &tampered,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            ),
            Err(OfferRefusal::IncompleteObjectClosure {
                missing: digest_key(&object(200).0)
            })
        );
    }

    #[test]
    fn h011_repeat_offer_is_idempotent_and_divergence_quarantines() {
        let engine = RusqliteEngine::open_in_memory().unwrap();
        let mut store = SqlMetadataStore::open(engine).unwrap();
        ready_store(&mut store);
        let first = offer();
        assert!(matches!(
            process_offer(
                &mut store,
                &first,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));

        // Identical re-offer: evidence appended, still exactly one
        // publication, no quarantine.
        assert_eq!(
            process_offer(
                &mut store,
                &first,
                &expected_descriptor(),
                no_committed,
                901,
                2,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::IdempotentEvidenceAppended
        );

        // A different manifest object for the same key with a different
        // semantic digest: determinism incident, action quarantined, the
        // committed row preserved.
        let mut divergent_manifest = manifest();
        divergent_manifest.logical_outputs[0].object = object(42);
        let divergent_id = object(52);
        store.record_object(&object(42).0, 64).unwrap();
        store
            .add_location(&object(42).0, "/cas/42", Some(1), "raw", true)
            .unwrap();
        store.record_object(&divergent_id.0, 64).unwrap();
        store
            .add_location(&divergent_id.0, "/cas/52", Some(1), "raw", true)
            .unwrap();
        let divergent = OfferPreparedActionResult::build(
            attempt_authority(),
            divergent_manifest,
            divergent_id.clone(),
            evidence(&divergent_id),
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        locate_bundle_root(&mut store, &divergent);
        let committed = first.manifest.clone();
        let outcome = process_offer(
            &mut store,
            &divergent,
            &expected_descriptor(),
            move |_| Some(committed.clone()),
            902,
            3,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        let PublicationOutcome::Quarantined(quarantine) = outcome else {
            panic!("expected quarantine, got {outcome:?}");
        };
        assert_eq!(quarantine.class, DivergenceClass::SemanticDivergence);
        assert_eq!(quarantine.incident_seq, 3);
        assert_eq!(quarantine.candidate_pin_id, 902);
        // Committed manifest pointer unchanged.
        assert_eq!(
            store
                .published_manifest_key(&digest("rabs.action-key.sha256.v1", 7))
                .unwrap()
                .unwrap(),
            digest_key(&object(50).0)
        );
    }

    /// A second offer for the same key whose manifest object differs.
    /// `output_tag`/`manifest_tag`/`evidence_tag` pick the divergent
    /// output object, manifest id, and evidence id; `observations` picks
    /// the observation stream.
    fn divergent_offer(
        store: &mut dyn RabsMetadataStore,
        output_tag: Option<u8>,
        manifest_tag: u8,
        evidence_tag: u8,
        observations: TypedDigest,
    ) -> OfferPreparedActionResult {
        let mut m = manifest();
        if let Some(tag) = output_tag {
            m.logical_outputs[0].object = object(tag);
        }
        let id = object(manifest_tag);
        for tag in output_tag
            .iter()
            .copied()
            .chain([manifest_tag, evidence_tag])
        {
            store.record_object(&object(tag).0, 64).unwrap();
            store
                .add_location(&object(tag).0, &format!("/cas/{tag}"), Some(1), "raw", true)
                .unwrap();
        }
        let built = OfferPreparedActionResult::build(
            attempt_authority(),
            m,
            id.clone(),
            evidence(&id),
            object(evidence_tag),
            observations,
            &declared(),
            Vec::new(),
        )
        .unwrap();
        locate_bundle_root(store, &built);
        built
    }

    #[test]
    fn h026_semantic_divergence_quarantines_preserves_and_escalates() {
        let mut store = SqlMetadataStore::open(RusqliteEngine::open_in_memory().unwrap()).unwrap();
        ready_store(&mut store);
        let action = digest("rabs.action-key.sha256.v1", 7);
        let action_key = digest_key(&action);
        let auth = authority_digest(&coordinator_authority());

        let first = offer();
        assert!(matches!(
            process_offer(
                &mut store,
                &first,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));

        // Two consumers were served this result under a RELEASE tier.
        store
            .record_served_consumer(&action_key, "consumer-b")
            .unwrap();
        store
            .record_served_consumer(&action_key, "consumer-a")
            .unwrap();
        store
            .append_trust_evaluation(
                &auth,
                &action,
                &crate::metadata_store::TrustEvaluationRow {
                    version: 1,
                    state: "project-release-eligible".to_owned(),
                    reason: "release gate".to_owned(),
                    evaluated_seq: 2,
                },
            )
            .unwrap();

        // Semantic divergence: different output object under the same key.
        let divergent = divergent_offer(
            &mut store,
            Some(42),
            52,
            55,
            digest("rabs.observation-stream.sha256.v1", 9),
        );
        let committed = first.manifest.clone();
        let outcome = process_offer(
            &mut store,
            &divergent,
            &expected_descriptor(),
            move |_| Some(committed.clone()),
            902,
            3,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        let PublicationOutcome::Quarantined(quarantine) = outcome else {
            panic!("expected quarantine, got {outcome:?}");
        };
        assert_eq!(quarantine.class, DivergenceClass::SemanticDivergence);

        // Serving is DISABLED with the full quarantine disposition.
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_QUARANTINED
        );

        // BOTH candidates preserved: the committed row is untouched...
        assert_eq!(
            store.published_manifest_key(&action).unwrap().unwrap(),
            digest_key(&object(50).0)
        );
        // ...and the losing candidate sits under a durable
        // divergence-evidence pin with reachability edges.
        let pin = store.pin_row(902).unwrap().unwrap();
        assert_eq!(pin.class, DIVERGENCE_EVIDENCE_PIN_CLASS);
        assert_eq!(pin.root_key, digest_key(&object(52).0));
        assert!(!pin.released);
        let snapshot = store.differential_snapshot().unwrap();
        assert!(
            snapshot
                .iter()
                .any(|l| l.starts_with("object_edges|") && l.contains("divergence-candidate"))
        );
        // The candidate's evidence bundle joined the append-only index.
        assert!(
            store
                .list_evidence_keys(&action)
                .unwrap()
                .contains(&digest_key(&object(55).0))
        );
        // H029/I37: the candidate's evidence is bound to the CANDIDATE
        // manifest — the committed canonical result's evidence view never
        // absorbs a divergent candidate's evidence.
        let candidate_view = store
            .list_evidence_keys_for_manifest(&digest_key(&object(52).0))
            .unwrap();
        assert!(candidate_view.contains(&digest_key(&object(55).0)));
        let committed_view = store
            .list_evidence_keys_for_manifest(&digest_key(&object(50).0))
            .unwrap();
        assert!(!committed_view.contains(&digest_key(&object(55).0)));

        // The incident row names class, both manifests, and the pin.
        let incidents = store.list_divergence_incidents(&action_key).unwrap();
        assert_eq!(incidents.len(), 1);
        assert_eq!(incidents[0].seq, 3);
        assert_eq!(incidents[0].class, "semantic");
        assert_eq!(
            incidents[0].committed_manifest_key,
            digest_key(&object(50).0)
        );
        assert_eq!(
            incidents[0].candidate_manifest_key,
            digest_key(&object(52).0)
        );
        assert_eq!(incidents[0].candidate_pin_hex, format!("{:032x}", 902));

        // Previously served consumers escalated by RELEASE tier: recall.
        assert_eq!(
            quarantine.escalations,
            vec![
                ConsumerEscalation {
                    consumer: "consumer-a".to_owned(),
                    trust_state: "project-release-eligible".to_owned(),
                    decision: "recall-and-reverify".to_owned(),
                },
                ConsumerEscalation {
                    consumer: "consumer-b".to_owned(),
                    trust_state: "project-release-eligible".to_owned(),
                    decision: "recall-and-reverify".to_owned(),
                },
            ]
        );
        assert!(snapshot.iter().any(|l| {
            l.starts_with(
                "decision_receipts|divergence-escalation|consumer-a|3|recall-and-reverify",
            )
        }));
    }

    #[test]
    fn t025_observable_only_divergence_has_presentation_not_action_quarantine() {
        let mut store = SqlMetadataStore::open(RusqliteEngine::open_in_memory().unwrap()).unwrap();
        ready_store(&mut store);
        let action = digest("rabs.action-key.sha256.v1", 7);
        let action_key = digest_key(&action);

        let first = offer();
        assert!(matches!(
            process_offer(
                &mut store,
                &first,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));
        store
            .record_served_consumer(&action_key, "consumer-a")
            .unwrap();

        // Same semantic result, different observation stream, offered as
        // a different manifest object: observable-only divergence.
        let divergent = divergent_offer(
            &mut store,
            None,
            53,
            56,
            digest("rabs.observation-stream.sha256.v1", 10),
        );
        assert_eq!(
            divergent.manifest.semantic_result_digest,
            first.manifest.semantic_result_digest
        );
        let committed = first.manifest.clone();
        let outcome = process_offer(
            &mut store,
            &divergent,
            &expected_descriptor(),
            move |_| Some(committed.clone()),
            903,
            4,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        let PublicationOutcome::Quarantined(quarantine) = outcome else {
            panic!("expected quarantine, got {outcome:?}");
        };
        assert_eq!(quarantine.class, DivergenceClass::ObservableOnlyDivergence);

        // The NARROWER disposition: ordinary replay disabled, tagged as
        // presentation-only.
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_PRESENTATION_QUARANTINED
        );
        // Candidate still preserved; incident recorded with the narrower
        // class; NO consumer escalation for observable-only.
        assert!(store.pin_row(903).unwrap().is_some());
        let incidents = store.list_divergence_incidents(&action_key).unwrap();
        assert_eq!(incidents.len(), 1);
        assert_eq!(incidents[0].class, "observable-only");
        assert!(quarantine.escalations.is_empty());
        let snapshot = store.differential_snapshot().unwrap();
        assert!(
            !snapshot
                .iter()
                .any(|line| { line.starts_with("decision_receipts|divergence-escalation|") })
        );
        assert!(
            !snapshot.iter().any(|line| {
                line.starts_with("quarantines|action-entry|") && line.contains(&action_key)
            }),
            "observable-only divergence must not quarantine the canonical action entry"
        );

        // A later observable-only candidate must never RELAX an existing
        // full quarantine installed by a stronger incident.
        store
            .set_serving_disposition_key(&action_key, DISPOSITION_QUARANTINED)
            .unwrap();
        store
            .add_quarantine(
                QuarantineScope::ActionEntry,
                &action_key,
                "preexisting-full-quarantine",
            )
            .unwrap();
        let later_observable = divergent_offer(
            &mut store,
            None,
            57,
            58,
            digest("rabs.observation-stream.sha256.v1", 11),
        );
        let committed = first.manifest.clone();
        let later_outcome = process_offer(
            &mut store,
            &later_observable,
            &expected_descriptor(),
            move |_| Some(committed.clone()),
            905,
            6,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        assert!(matches!(
            later_outcome,
            PublicationOutcome::Quarantined(DivergenceQuarantine {
                class: DivergenceClass::ObservableOnlyDivergence,
                ..
            })
        ));
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_QUARANTINED
        );
        assert!(store.differential_snapshot().unwrap().iter().any(|line| {
            line.starts_with("quarantines|action-entry|")
                && line.contains("preexisting-full-quarantine")
        }));
    }

    #[test]
    fn t025_distinct_worker_attempts_with_different_evidence_share_canonical_result() {
        let mut store = SqlMetadataStore::open(RusqliteEngine::open_in_memory().unwrap()).unwrap();
        ready_store(&mut store);
        let action = digest("rabs.action-key.sha256.v1", 7);
        let action_key = digest_key(&action);

        // Build worker A's result in two passes: stamp the manifest,
        // encode its canonical bytes, derive the REAL CAS object id, and
        // rebuild with evidence bound to that content-derived id.
        let first_authority = attempt_authority();
        let first_timing_payload = b"worker=worker-a;started=1;elapsed_us=3100;cpu_us=2200";
        let first_resource_payload = b"worker=worker-a;cores=2;peak_rss_bytes=1048576";
        let first_timing = content_object(first_timing_payload);
        let first_resources = content_object(first_resource_payload);
        let first_placeholder_id = object(50);
        let mut first_placeholder_evidence = evidence(&first_placeholder_id);
        first_placeholder_evidence.execution_snapshot_root = first_resources.clone();
        first_placeholder_evidence.raw_process_and_event_evidence = first_timing.clone();
        let first_stamped = OfferPreparedActionResult::build(
            first_authority.clone(),
            manifest(),
            first_placeholder_id,
            first_placeholder_evidence,
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        let first_manifest_bytes =
            crate::manifest_codec::encode_manifest_v1(&first_stamped.manifest);
        let first_manifest_id = content_object(&first_manifest_bytes);
        let mut first_evidence = evidence(&first_manifest_id);
        first_evidence.execution_snapshot_root = first_resources.clone();
        first_evidence.raw_process_and_event_evidence = first_timing.clone();
        let first = OfferPreparedActionResult::build(
            first_authority,
            manifest(),
            first_manifest_id.clone(),
            first_evidence,
            object(51),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        assert_eq!(
            crate::manifest_codec::encode_manifest_v1(&first.manifest),
            first_manifest_bytes
        );
        locate_test_object(
            &mut store,
            &first_manifest_id,
            "/cas/manifest-worker-a",
            first_manifest_bytes.len(),
        );
        locate_test_object(
            &mut store,
            &first_timing,
            "/cas/timing-worker-a",
            first_timing_payload.len(),
        );
        locate_test_object(
            &mut store,
            &first_resources,
            "/cas/resources-worker-a",
            first_resource_payload.len(),
        );
        let first_outcome = process_offer(
            &mut store,
            &first,
            &expected_descriptor(),
            no_committed,
            900,
            1,
            CommitDurabilityProfile::RequireDurableClosure,
            || 10,
        )
        .unwrap();
        let PublicationOutcome::Committed(first_record) = first_outcome else {
            panic!("expected first commit, got {first_outcome:?}");
        };
        assert_eq!(first_record.canonical_result_manifest_id, first_manifest_id);

        // A genuinely distinct attempt runs on another worker at a
        // later recorded time with its own lease and resource/timing
        // evidence. None of those attempt facts belong to the canonical
        // result manifest (R80/R115).
        let second_authority = distinct_attempt_authority();
        let auth = authority_digest(&second_authority.coordinator);
        store
            .admit_worker_session(
                &auth,
                &WorkerSessionOffer {
                    worker_peer_id: second_authority.worker_peer_id.clone(),
                    boot_generation: second_authority.worker_boot_generation,
                    incarnation: second_authority.worker_incarnation_id,
                    reenrollment_proof: None,
                },
                10,
            )
            .unwrap();
        store
            .admit_attempt_lease(&second_authority, 17, 200)
            .unwrap();

        let second_timing_payload = b"worker=worker-b;started=10;elapsed_us=8700;cpu_us=6900";
        let second_resource_payload = b"worker=worker-b;cores=6;peak_rss_bytes=3145728";
        let second_timing = content_object(second_timing_payload);
        let second_resources = content_object(second_resource_payload);
        let second_placeholder_id = object(52);
        let mut second_placeholder_evidence = evidence(&second_placeholder_id);
        second_placeholder_evidence.execution_snapshot_root = second_resources.clone();
        second_placeholder_evidence.raw_process_and_event_evidence = second_timing.clone();
        let second_stamped = OfferPreparedActionResult::build(
            second_authority.clone(),
            manifest(),
            second_placeholder_id,
            second_placeholder_evidence,
            object(54),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        let second_manifest_bytes =
            crate::manifest_codec::encode_manifest_v1(&second_stamped.manifest);
        let second_manifest_id = content_object(&second_manifest_bytes);
        let mut second_evidence = evidence(&second_manifest_id);
        second_evidence.execution_snapshot_root = second_resources.clone();
        second_evidence.raw_process_and_event_evidence = second_timing.clone();
        let reoffer = OfferPreparedActionResult::build(
            second_authority.clone(),
            manifest(),
            second_manifest_id.clone(),
            second_evidence,
            object(54),
            digest("rabs.observation-stream.sha256.v1", 9),
            &declared(),
            Vec::new(),
        )
        .unwrap();
        assert_ne!(
            first.authority.worker_peer_id,
            reoffer.authority.worker_peer_id
        );
        assert_ne!(first.authority.attempt_id, reoffer.authority.attempt_id);
        assert_ne!(
            first.authority.execution_lease_id,
            reoffer.authority.execution_lease_id
        );
        assert_ne!(
            first.evidence.execution_snapshot_root,
            reoffer.evidence.execution_snapshot_root
        );
        assert_ne!(
            first.evidence.raw_process_and_event_evidence,
            reoffer.evidence.raw_process_and_event_evidence
        );
        assert_eq!(first_manifest_bytes, second_manifest_bytes);
        assert_eq!(first_manifest_id, second_manifest_id);
        assert_eq!(first.manifest_id, reoffer.manifest_id);
        assert_eq!(first.manifest, reoffer.manifest);
        locate_test_object(
            &mut store,
            &second_manifest_id,
            "/cas/manifest-worker-b",
            second_manifest_bytes.len(),
        );
        locate_test_object(
            &mut store,
            &second_timing,
            "/cas/timing-worker-b",
            second_timing_payload.len(),
        );
        locate_test_object(
            &mut store,
            &second_resources,
            "/cas/resources-worker-b",
            second_resource_payload.len(),
        );
        locate_test_object(&mut store, &object(54), "/cas/54", 64);
        assert_eq!(
            process_offer(
                &mut store,
                &reoffer,
                &expected_descriptor(),
                no_committed,
                904,
                18,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::IdempotentEvidenceAppended
        );
        let evidence_keys = store.list_evidence_keys(&action).unwrap();
        assert!(evidence_keys.contains(&digest_key(&object(51).0)));
        assert!(evidence_keys.contains(&digest_key(&object(54).0)));
        // H029: both equivalent attempts' evidence bundles bind to the ONE
        // committed canonical manifest, and the per-manifest view shows
        // the appended set.
        let manifest_view = store
            .list_evidence_keys_for_manifest(&digest_key(&first_manifest_id.0))
            .unwrap();
        assert!(manifest_view.contains(&digest_key(&object(51).0)));
        assert!(manifest_view.contains(&digest_key(&object(54).0)));
        let snapshot = store.differential_snapshot().unwrap();
        let second_attempt_hex = format!("{:032x}", second_authority.attempt_id.0);
        let second_attempt_line = snapshot
            .iter()
            .find(|line| line.starts_with(&format!("action_attempts|{second_attempt_hex}|")))
            .expect("second attempt row");
        let attempt_fields: Vec<_> = second_attempt_line.split('|').collect();
        assert_eq!(attempt_fields[4], "worker-b");
        assert_eq!(attempt_fields[5], "17");
        assert_eq!(
            attempt_fields[8],
            format!("{:032x}", second_authority.execution_lease_id.0)
        );
        let second_evidence_line = snapshot
            .iter()
            .find(|line| {
                let fields: Vec<_> = line.split('|').collect();
                fields.len() == 8
                    && fields[0] == "action_evidence_index"
                    && fields[6] == second_attempt_hex
            })
            .expect("second attempt evidence row");
        let evidence_fields: Vec<_> = second_evidence_line.split('|').collect();
        assert_eq!(evidence_fields[7], digest_key(&first_manifest_id.0));
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            "servable"
        );
        assert!(
            store
                .list_divergence_incidents(&action_key)
                .unwrap()
                .is_empty()
        );
        assert!(!snapshot.iter().any(|line| line.starts_with("quarantines|")));
    }

    #[test]
    fn h026_incident_rows_are_append_only_and_authority_gated() {
        let mut store = SqlMetadataStore::open(RusqliteEngine::open_in_memory().unwrap()).unwrap();
        let auth = authority_digest(&coordinator_authority());
        let row = DivergenceIncidentRow {
            action_key: "k".to_owned(),
            seq: 1,
            class: "semantic".to_owned(),
            committed_manifest_key: "m1".to_owned(),
            candidate_manifest_key: "m2".to_owned(),
            candidate_evidence_key: "e2".to_owned(),
            candidate_pin_hex: format!("{:032x}", 7),
            generation_hex: format!("{:032x}", 11),
            attempt_hex: format!("{:032x}", 20),
            detail: "detail".to_owned(),
        };
        // No active authority: refused, nothing written.
        assert_eq!(
            store.record_divergence_incident(&auth, &row),
            Err(StoreError::NotActiveAuthority)
        );
        ready_store(&mut store);
        store.record_divergence_incident(&auth, &row).unwrap();
        // Identical re-record: idempotent.
        store.record_divergence_incident(&auth, &row).unwrap();
        // Conflicting rewrite: typed refusal; the stored row survives.
        let mut tampered = row.clone();
        tampered.detail = "rewritten".to_owned();
        assert_eq!(
            store.record_divergence_incident(&auth, &tampered),
            Err(StoreError::AppendConflict("divergence_incidents".into()))
        );
        let incidents = store.list_divergence_incidents("k").unwrap();
        assert_eq!(incidents.len(), 1);
        assert_eq!(incidents[0], row);
    }

    #[test]
    fn h026_differential_reference_vs_frankensqlite() {
        // Semantic divergence with escalation AND observable-only
        // divergence must leave both engines byte-identical.
        fn scenario(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let action = digest("rabs.action-key.sha256.v1", 7);
            let action_key = digest_key(&action);
            let first = offer();
            assert!(matches!(
                process_offer(
                    store,
                    &first,
                    &expected_descriptor(),
                    no_committed,
                    900,
                    1,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 10,
                )
                .unwrap(),
                PublicationOutcome::Committed(_)
            ));
            store
                .record_served_consumer(&action_key, "consumer-a")
                .unwrap();
            let divergent = divergent_offer(
                store,
                Some(42),
                52,
                55,
                digest("rabs.observation-stream.sha256.v1", 9),
            );
            let committed = first.manifest.clone();
            let outcome = process_offer(
                store,
                &divergent,
                &expected_descriptor(),
                move |_| Some(committed.clone()),
                902,
                3,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap();
            assert!(matches!(outcome, PublicationOutcome::Quarantined(_)));
            store.differential_snapshot().unwrap()
        }
        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&fresh_path("ref26")).unwrap()).unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&fresh_path("fsq26")).unwrap()).unwrap();
        assert_eq!(scenario(&mut reference), scenario(&mut candidate));
    }

    #[test]
    fn h011_differential_full_pipeline_reference_vs_frankensqlite() {
        // The entire H011 scenario (commit, idempotent re-offer,
        // divergence quarantine, plus every refusal fence) must leave the
        // reference and FrankenSQLite stores byte-identical.
        fn scenario(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let first = offer();
            assert!(matches!(
                process_offer(
                    store,
                    &first,
                    &expected_descriptor(),
                    no_committed,
                    900,
                    1,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 10,
                )
                .unwrap(),
                PublicationOutcome::Committed(_)
            ));
            assert_eq!(
                process_offer(
                    store,
                    &first,
                    &expected_descriptor(),
                    no_committed,
                    901,
                    2,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 10,
                )
                .unwrap(),
                PublicationOutcome::IdempotentEvidenceAppended
            );
            assert_eq!(
                process_offer(
                    store,
                    &first,
                    &digest("rabs.descriptor.sha256.v1", 99),
                    no_committed,
                    902,
                    3,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 10,
                ),
                Err(OfferRefusal::DescriptorMismatch)
            );
            store.differential_snapshot().unwrap()
        }
        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&fresh_path("ref")).unwrap()).unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&fresh_path("fsq")).unwrap()).unwrap();
        assert_eq!(scenario(&mut reference), scenario(&mut candidate));
    }

    #[test]
    fn h034_evict_then_recompute_divergence_is_detected() {
        let mut store = SqlMetadataStore::open(RusqliteEngine::open_in_memory().unwrap()).unwrap();
        ready_store(&mut store);
        let first = offer();
        assert!(matches!(
            process_offer(
                &mut store,
                &first,
                &expected_descriptor(),
                no_committed,
                900,
                1,
                CommitDurabilityProfile::RequireDurableClosure,
                || 10,
            )
            .unwrap(),
            PublicationOutcome::Committed(_)
        ));

        // Blobs get evicted; the digests are RETAINED first.
        retain_eviction_tombstone(&mut store, &first.manifest, 50).unwrap();
        for tag in [40u8, 41, 50] {
            store
                .remove_location_by_key(&digest_key(&object(tag).0), &format!("/cas/{tag}"))
                .unwrap();
        }

        let action = digest("rabs.action-key.sha256.v1", 7);
        let action_key = digest_key(&action);

        // Same semantic result, different observable projection: the
        // narrower presentation quarantine, never an action quarantine.
        assert_eq!(
            check_recomputation_against_tombstone(
                &mut store,
                &action,
                &first.manifest.semantic_result_digest,
                &digest(OBSERVABLE_PROJECTION_DOMAIN, 99),
            )
            .unwrap(),
            RecomputationCheck::Divergence(DivergenceClass::ObservableOnlyDivergence)
        );
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_PRESENTATION_QUARANTINED
        );
        assert!(!store.differential_snapshot().unwrap().iter().any(|line| {
            line.starts_with("quarantines|action-entry|") && line.contains(&action_key)
        }));

        // A DIFFERENT semantic result does make the action suspect. The
        // full quarantine replaces the narrower disposition, while the
        // tombstone remains incident evidence.
        let outcome = check_recomputation_against_tombstone(
            &mut store,
            &action,
            &digest(SEMANTIC_PROJECTION_DOMAIN, 99),
            &first.manifest.observable_result_digest,
        )
        .unwrap();
        assert_eq!(
            outcome,
            RecomputationCheck::Divergence(DivergenceClass::SemanticDivergence)
        );
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_QUARANTINED
        );
        assert!(store.differential_snapshot().unwrap().iter().any(|line| {
            line.starts_with("quarantines|action-entry|")
                && line.contains("post-eviction recomputation divergence")
        }));
        assert!(
            store.eviction_tombstone(&action).unwrap().is_some(),
            "divergence keeps the tombstone as incident evidence"
        );

        // Reversing the arrival order must not downgrade the full
        // quarantine back to presentation-only.
        assert_eq!(
            check_recomputation_against_tombstone(
                &mut store,
                &action,
                &first.manifest.semantic_result_digest,
                &digest(OBSERVABLE_PROJECTION_DOMAIN, 98),
            )
            .unwrap(),
            RecomputationCheck::Divergence(DivergenceClass::ObservableOnlyDivergence)
        );
        assert_eq!(
            store.serving_disposition_key(&action_key).unwrap().unwrap(),
            DISPOSITION_QUARANTINED
        );

        // A faithful reproduction matches and CONSUMES the tombstone.
        assert_eq!(
            check_recomputation_against_tombstone(
                &mut store,
                &action,
                &first.manifest.semantic_result_digest,
                &first.manifest.observable_result_digest,
            )
            .unwrap(),
            RecomputationCheck::Reproduced
        );
        assert!(store.eviction_tombstone(&action).unwrap().is_none());
        assert_eq!(
            check_recomputation_against_tombstone(
                &mut store,
                &action,
                &first.manifest.semantic_result_digest,
                &first.manifest.observable_result_digest,
            )
            .unwrap(),
            RecomputationCheck::NoTombstone
        );
    }

    #[test]
    fn h034_differential_reference_vs_frankensqlite() {
        fn scenario(store: &mut dyn RabsMetadataStore) -> Vec<String> {
            ready_store(store);
            let first = offer();
            assert!(matches!(
                process_offer(
                    store,
                    &first,
                    &expected_descriptor(),
                    no_committed,
                    900,
                    1,
                    CommitDurabilityProfile::RequireDurableClosure,
                    || 10,
                )
                .unwrap(),
                PublicationOutcome::Committed(_)
            ));
            retain_eviction_tombstone(store, &first.manifest, 50).unwrap();
            let action = digest("rabs.action-key.sha256.v1", 7);
            assert_eq!(
                check_recomputation_against_tombstone(
                    store,
                    &action,
                    &digest(SEMANTIC_PROJECTION_DOMAIN, 99),
                    &first.manifest.observable_result_digest,
                )
                .unwrap(),
                RecomputationCheck::Divergence(DivergenceClass::SemanticDivergence)
            );
            store.differential_snapshot().unwrap()
        }
        let mut reference =
            SqlMetadataStore::open(RusqliteEngine::open(&fresh_path("ref34")).unwrap()).unwrap();
        let mut candidate =
            SqlMetadataStore::open(FsqliteEngine::open(&fresh_path("fsq34")).unwrap()).unwrap();
        assert_eq!(scenario(&mut reference), scenario(&mut candidate));
    }
}