net-mesh 0.35.0

//! OLB-2B-E3b: the bounded node routing registry — the actor's real consumer.
//!
//! Retains one node slot per AUTHORITY-SCOPED key, shared by every clone family
//! demanding that same key, and rebuilds slots from the private-discovery source
//! in bounded quanta. It implements `DirtyApply`, so the supervised actor from
//! E2 drives it and nothing else does.
//!
//! Three authoritative structures, deliberately separate:
//!
//! - `RegistryWork` answers only "is SOME work owed?" — a coalescing wake hint
//!   (Kyra OLB-2B-E3a). The Boolean must never become the work queue;
//! - `RegistryInner::pending` holds the exact slot identities that owe work;
//! - `RegistryInner::live_actor` holds the ONE incarnation currently permitted
//!   to consume that work. It is bound to the actor LIFECYCLE via
//!   `DirtyApply::activate_incarnation` / `DirtyApply::deactivate_incarnation`,
//!   never to a high-water counter: a high-water mark cannot distinguish "a newer
//!   actor took over" from "no actor is live", so a stale attempt could consume
//!   pending identities, be rejected at installation, and still report success.
//!
//! ## Frozen lock order
//!
//! `source commit pin` → `registry lock`.
//!
//! A quantum runs as: select/invalidate under the registry lock → release → brief
//! source snapshot → reconstruct entirely off-lock → acquire the commit pin →
//! registry lock beneath it → install → release both. No source method is called
//! while the registry lock is held; no source-side lock is held across decoding,
//! sorting or projection; and no registry method a source implementation could
//! re-enter takes the commit pin.

use arc_swap::ArcSwapOption;
use std::collections::{BTreeMap, BTreeSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use super::org::OrgId;
use super::org_grant::CapabilityAuthorityId;
use super::org_routing::{ApplyOutcome, ApplyRequest, DirtyApply, RegistryWork};
use super::org_scoped_ingest::CapabilityAudienceScope;
use super::org_scoped_store::{DirtyCapabilities, PrivateCapabilityProvider};
use crate::adapter::net::identity::EntityId;

/// Max demand handles ONE clone family may hold.
///
/// Handles, not capabilities. One warmed capability costs one handle per
/// AUTHORITY-SCOPED contributor it needs — Owner plus every DISCOVER audience
/// the family actually leased for it — so a family holding two leased DISCOVER
/// grants per capability warms ~21 capabilities, not 64. The plan's older
/// "64 warmed capabilities per clone family" wording described a bound this
/// registry has never enforced; OLB-2B.3b corrects it there to
/// "64 retained authority-scoped route demands".
pub(crate) const MAX_HANDLES_PER_FAMILY: usize = 64;
/// Max distinct retained node slots.
pub(crate) const MAX_NODE_SLOTS: usize = 256;
/// Max slots rebuilt in ONE synchronous application. Work beyond this stays
/// authoritative in `pending` and re-marks, so a 256-slot rebuild becomes several
/// yielding quanta rather than one unbroken burst.
const APPLY_QUANTUM: usize = 64;

/// A clone family's identity.
///
/// Deliberately UNCONSTRUCTIBLE outside this module: it is minted only by
/// [`NodeOrgRoutingRegistry::new_family`] and reaches callers wrapped in a
/// [`RoutingFamily`]. A caller cannot name another family's id, so it cannot spend
/// another family's handle budget or forge membership in one (Kyra OLB-2B-E3b).
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct FamilyId(u64);

/// An audience scope in the PRIVATE partition.
///
/// `CapabilityAudienceScope::Public` is unrepresentable here: this registry is the
/// owner-private routing consumer, and a public scope reaching a retained slot
/// would mean the private plane retained a plaintext, globally-discoverable row
/// under private-authority facts. Rejecting it at construction makes that
/// structurally impossible rather than a filter someone can forget.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) struct PrivateAudienceScope(CapabilityAudienceScope);

impl PrivateAudienceScope {
    /// `None` for `Public` — the only rejection.
    #[allow(dead_code)] // demand surface; see the note above `RoutingFamily`.
    pub(crate) fn new(scope: CapabilityAudienceScope) -> Option<Self> {
        match scope {
            CapabilityAudienceScope::Public => None,
            private => Some(Self(private)),
        }
    }

    pub(crate) fn scope(&self) -> &CapabilityAudienceScope {
        &self.0
    }
}

/// The AUTHORITY-SCOPED identity of a retained slot.
///
/// The audience scope is part of the key, never elided: two demands under
/// different acting identities/audiences are different slots even for the same
/// capability. Sharing a slot across scopes would let indexing or reuse broaden
/// authority, which no amount of downstream filtering can undo.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) struct SlotKey {
    pub scope: PrivateAudienceScope,
    pub capability: CapabilityAuthorityId,
}

/// Why demand was refused. Deterministic and state-free: a refused demand retains
/// nothing, so the caller takes the current-authority cold path.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum DemandRefused {
    /// This family already holds `MAX_HANDLES_PER_FAMILY` handles.
    #[allow(dead_code)]
    FamilyAtCapacity,
    /// The node already retains `MAX_NODE_SLOTS` distinct slots. A live slot is
    /// NEVER evicted to satisfy new demand.
    #[allow(dead_code)]
    NodeAtCapacity,
    /// The monotone identity space backing slot incarnations and family identities
    /// is exhausted. Refusing is the only safe answer: wrapping would let a stale
    /// artifact match a live slot's incarnation and resurrect it, and aborting
    /// would take the process down over a bookkeeping limit.
    IdSpaceExhausted,
}

/// The authority-scoped base facts for one retained slot.
///
/// Stamped with all three identities, because health alone is never
/// source-currentness: the actor incarnation that built it, the slot incarnation
/// it was built for, and the source generation it was built from.
#[derive(Clone, Debug)]
pub(crate) struct SlotBaseFacts {
    #[allow(dead_code)] // read by the warmed-call consumer.
    pub providers: SourceFacts,
    /// The WHOLE source epoch these facts were built and committed under.
    pub epoch: SourceEpoch,
    /// The EXACT scope authority these facts were reconstructed under
    /// (OLB-2B.3c-pre). Per-key: never the batch-wide Grant vector, or unrelated
    /// Grant movement would invalidate Owner and unrelated Grant facts.
    pub authority: ScopedDiscoveryAuthorityStamp,
    pub actor_incarnation: u64,
    pub slot_incarnation: u64,
    /// Where this artifact sits in the consumer-Grant transition order
    /// (OLB-2B.3c-pre step 3). See [`ScopedSourceFacts::grant_fence`].
    pub grant_fence: GrantArtifactFence,
    /// The earliest `expires_at` across the retained providers, or `u64::MAX`
    /// when nothing here can expire.
    ///
    /// Expiry filtering at capture is not enough: reconstruction and the commit
    /// can both cross a deadline, and the exact-expiry timer is asynchronous and
    /// may itself be waiting on the publication gate. Carrying the bound lets the
    /// READ seam reject the whole object the moment it is crossed, so the timer
    /// governs promptness rather than correctness — which is what the scoped
    /// store's uncached reads already guarantee (Kyra OLB-2B-E3c).
    pub earliest_expiry: u64,
}

/// The whole authoritative source epoch a fact was built under.
///
/// NOT merely the scoped revision. Routing material is filtered by revocation
/// authority, so two facts built under different revocation authorities are from
/// different epochs even when the scoped revision is identical — a store swap
/// that retracts nothing moves no scoped revision at all. Stamping only the
/// scoped half let a multi-quantum recapture combine facts from two authorities
/// and settle `Current` over them (Kyra OLB-2B-E3c).
///
/// Every use of "generation" in this module means THIS: what facts are stamped
/// with, what recapture coherence compares, and what settlement reports.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct SourceEpoch {
    /// The scoped query-visible revision.
    pub generation: u64,
    /// The routing AUTHORITY epoch — monotone across revocation store installs
    /// and floor movement, allocated by the node under its authority gate. Never
    /// reused, so it cannot alias a previous authority.
    pub authority: u64,
    /// The revocation store's barriered floor generation.
    ///
    /// Separate from `authority` because it moves INDEPENDENTLY: a floor
    /// publication becomes authoritative inside the revocation store before the
    /// subscriber that advances `authority` is even notified. Retaining it means
    /// facts built against floors F0 are detectably stale once F1 is live, even
    /// during that notification gap (Kyra OLB-2B-E3c).
    pub floor_generation: u64,
    /// Whether the revocation authority was DURABILITY-POISONED when the facts
    /// were built.
    ///
    /// Carried rather than re-read live, because both directions matter and only
    /// the comparison distinguishes them. Poison rising invalidates facts built
    /// without it. Poison CLEARING invalidates facts built with it — a recovery
    /// republishes the same durable view, so it raises no floor, and facts
    /// reconstructed as `Unserved` under poison would otherwise stay reconciled
    /// forever. Reading live poison as an unconditional staleness trigger would
    /// instead churn every read under STEADY poison, which is why it is an
    /// epoch field and not a predicate (Kyra OLB-2B-E3c closure).
    pub poisoned: bool,
}

/// The EXACT authority under which one scope's rows are query-visible
/// (OLB-2B.3c-pre).
///
/// Per-SLOT, never per-batch. The batch `SourceToken` protects the
/// capture/commit transaction; this protects one cached artifact. Copying a
/// batch-wide value into every `SlotBaseFacts` would make unrelated Grant
/// movement invalidate Owner and unrelated Grant facts — exactly what W-G8
/// forbids.
///
/// For the Grant plane the installed consumer Grant IS authority, not a
/// decryption convenience: the live query admits a row only while the stored
/// grant signature and audience handle equal the CURRENTLY INSTALLED ones, so a
/// cached artifact must compare the same things. All four components are bound:
///
/// - `grant_id` alone is insufficient — remove-then-reinstall, and a DIFFERENT
///   signed grant reusing the id, both leave it equal;
/// - `install_seq` distinguishes remove/reinstall of the same grant. It is the
///   checked, terminal, non-aliasing identity signed at `300e80f6c`; a wrapping
///   one would let an old artifact match a later installation;
/// - `grant_signature` distinguishes a different signed grant under the same id —
///   the signature binds the whole canonical grant;
/// - `audience_handle` mirrors the live query's defense-in-depth check, so the
///   cached path is never weaker than the uncached one.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ScopedDiscoveryAuthorityStamp {
    /// The owner plane: authority is the node's own org, already covered by the
    /// routing authority epoch and revocation view in [`SourceEpoch`].
    Owner,
    /// The grant plane: exactly one installed consumer Grant.
    Grant {
        grant_id: [u8; 32],
        install_seq: u64,
        grant_signature: [u8; 64],
        audience_handle: [u8; 32],
    },
}

/// One scope's reconstructed facts TOGETHER with the exact authority that
/// produced them (OLB-2B.3c-pre).
///
/// They travel as one value so a caller cannot stamp facts with an authority
/// that did not produce them.
#[derive(Clone, Debug)]
pub(crate) struct ScopedSourceFacts {
    pub facts: SourceFacts,
    pub authority: ScopedDiscoveryAuthorityStamp,
    /// The instant this scope's AUTHORITY stops authorizing anything, or
    /// `u64::MAX` if it carries no deadline of its own (scope item 12).
    ///
    /// Separate from the provider rows' deadlines and folded in beside them,
    /// because the two can disagree in the direction that matters: an installed
    /// Grant with ZERO visible providers yields a row deadline of `u64::MAX`,
    /// so an `earliest_expiry` derived from rows alone would say the artifact
    /// never expires while its authority expires in an hour.
    ///
    /// Boundary convention matches the read seam's `now >= earliest_expiry` and
    /// the grant family's `now >= not_after + skew ⇒ expired`, so this is
    /// exactly `not_after + MAX_TOKEN_CLOCK_SKEW_SECS` — the first instant the
    /// authority is no longer valid, not the last instant it is.
    pub authority_deadline: u64,
    /// Where this reconstruction sits in the consumer-Grant transition order.
    ///
    /// Ordering, not authority — `authority` above is the authority. This exists
    /// so a delayed movement notification can tell whether the artifact it is
    /// about to clear predates or postdates the transition it carries.
    ///
    /// Present on EVERY reconstruction, including `Unserved` ones, which is the
    /// whole point: an absent-Grant reconstruction is stamped `Owner` and so
    /// carries no installation identity, yet it can be the exact successor a
    /// later removal produced.
    pub grant_fence: GrantArtifactFence,
}

/// An opaque, comparable summary of EVERY authority input a snapshot was taken
/// under.
///
/// The registry never interprets it — it only carries it from the snapshot to
/// the commit pin and lets the SOURCE decide whether anything moved. That is
/// deliberate: a source's authority is not just its own state. The production
/// source's routing material is filtered by revocation floors, so its token
/// covers the scoped revision AND the revocation store's identity, barriered
/// generation and poison state. A scoped-revision-only check would miss a floor
/// that became authoritative before its retraction reached scoped state, and
/// install facts the very next callback invalidates (Kyra OLB-2B-E3c).
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(crate) struct SourceToken(Vec<u64>);

impl SourceToken {
    pub(crate) fn new(words: Vec<u64>) -> Self {
        Self(words)
    }
}

/// Whether the source can speak for a slot's authority scope AT ALL.
///
/// `Unserved` is NOT "zero providers". A source that cannot serve a scope has no
/// evidence about it, and an empty provider list is exact fresh negative
/// evidence — conflating them would let an unsupported scope masquerade as a
/// proven-empty one and drive a caller to a NonViable decision it never earned
/// (Kyra OLB-2B-E3c). A slot with `Unserved` facts is fully reconciled (it owes
/// no work) but reads COLD.
#[derive(Clone, Debug)]
pub(crate) enum SourceFacts {
    Served(Arc<[PrivateCapabilityProvider]>),
    Unserved,
}

/// Bounded material captured from the source for ONE quantum.
///
/// Owns everything reconstruction needs, so decode, sort and projection run with
/// NO source lock and NO registry lock held. That is the whole point of the
/// split: the source-side capture is brief, and the expensive part of a quantum
/// blocks nothing (Kyra OLB-2B-E3b).
pub(crate) trait SourceSnapshot {
    /// The full authority token this material was captured under.
    ///
    /// There is deliberately no separate generation accessor: the generation
    /// facts are STAMPED with comes from the commit pin, which is the only thing
    /// that proved it current.
    fn token(&self) -> SourceToken;

    /// Reconstruct `key`'s authority-scoped facts, with the exact scope
    /// authority that produced them.
    fn providers(&self, key: &SlotKey) -> ScopedSourceFacts;
}

/// A SHORT currentness pin, held only across final validation and installation.
///
/// Its existence proves NO authority input covered by the token has moved since
/// the snapshot, which is what makes a conditional install safe without a
/// two-sample before/after comparison — that comparison leaves a window between
/// the second read and the installation in which the source may move, and facts
/// installed in that window are stale while carrying a generation that says
/// otherwise.
pub(crate) trait SourceCommitPin {
    /// The epoch this pin proves is still current, and holds still through the
    /// conditional installation beneath it.
    fn epoch(&self) -> SourceEpoch;

    /// Re-verify EVERY authority input this pin claims to hold still.
    ///
    /// The pin excludes what the node serializes, but a source may have
    /// authority it publishes through its own synchronization — the production
    /// source's revocation floors and poison publish inside the revocation
    /// store, which no routing gate can exclude. So the pin's guarantee is
    /// completed here rather than assumed: called under the registry lock after
    /// the conditional installation and BEFORE settlement, so an authority that
    /// moved underneath cannot produce a `Current` (Kyra OLB-2B-E3c).
    ///
    /// Run `settle` IF every authority input is still current, as ONE operation
    /// relative to source publication.
    ///
    /// Checking and then settling are two independently interleavable steps: a
    /// publication landing between them makes the settlement false with nothing
    /// left to detect it. That matters here specifically because `Current` is not
    /// a historical observation — it is what causes the supervisor to publish
    /// `Healthy` (Kyra OLB-2B-E3c). The implementation holds the source's
    /// publication barrier across both, so no publication can interleave.
    ///
    /// `None` means authority moved and nothing was settled.
    fn settle_if_current(&self, settle: &mut dyn FnMut() -> ApplyOutcome) -> Option<ApplyOutcome>;
}

/// Supplies authority-scoped provider facts.
///
/// E3c binds this to the scoped discovery state, the mutation-publication gate
/// and the revocation authority. The two-method shape is deliberate: it makes
/// "the gate is a COMMIT pin, never a reconstruction lock" a property of the seam
/// rather than a rule an implementation has to remember. Holding the publication
/// gate across a quantum would block all private-discovery ingest, expiry sweeps
/// and floor mutation behind up to [`APPLY_QUANTUM`] authority-scoped store
/// queries plus their decoding.
pub(crate) trait SlotSource: Send + Sync + 'static {
    /// Briefly capture the exact authority-scoped material for `keys`, plus the
    /// token it was captured under. Every source-side lock is released before
    /// this returns. Called with NO registry lock held.
    fn snapshot(&self, keys: &[SlotKey]) -> Box<dyn SourceSnapshot>;

    /// Acquire the commit pin IF every authority input still matches `expected`.
    ///
    /// `None` means something the reconstruction depended on moved, and the
    /// caller must install nothing. ALWAYS called before the registry lock is
    /// taken, never while holding it — see the module's frozen lock order.
    /// `keys` is the SAME batch the snapshot was taken over.
    ///
    /// The token's Grant half is a vector of the exact installation identities
    /// THIS batch selected, so re-deriving it requires knowing which grants
    /// those were. Passing the keys lets the pin compare like with like instead
    /// of falling back to a global "some Grant moved" bit, which would defeat
    /// every commit pin in flight on unrelated Grant movement (OLB-2B.3c-pre).
    fn pin_if_current(
        &self,
        keys: &[SlotKey],
        expected: &SourceToken,
    ) -> Option<Box<dyn SourceCommitPin + '_>>;

    /// Whether this source can still settle a commit, and if it cannot, whether
    /// that is recoverable (E3c blockers §2, review-pass-3 §1).
    ///
    /// [`SourceLiveness::Live`] for every ORDINARY refusal: poison, floor
    /// movement and store swaps all own a later wake, so a failed pin correctly
    /// re-queues AND re-marks against them.
    fn liveness(&self) -> SourceLiveness {
        SourceLiveness::Live
    }

    /// Capture ONE coherent observation of the node's session/direct-state
    /// projection (OLB-2B.3c step 2).
    ///
    /// Called with NO registry lock held, exactly like [`Self::snapshot`], and
    /// exactly once per quantum: every annotation in every pool the quantum
    /// publishes comes from THIS observation, which is what makes "one
    /// coherent session generation" structural rather than a rule the build
    /// loop has to remember.
    ///
    /// The default observes no sessions, so every provider is annotated COLD.
    /// That is the correct answer for a source with no session plane — not a
    /// placeholder — and it is the same answer an unresolvable sample gets.
    fn session_view(&self) -> SessionObservation {
        SessionObservation::cold()
    }
}

/// Whether a [`SlotSource`] can settle, and — when it cannot — who owns the wake
/// that ends the condition (E3c blockers §2, review-pass-3 §1).
///
/// The distinction exists because the two unsettleable states differ in exactly
/// one way that matters to the work queue: whether a re-queued identity is a
/// promise that can eventually be kept.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum SourceLiveness {
    /// Ordinary operation. A pin can be taken and a settlement can succeed; a
    /// refusal is movement, and movement owns a wake.
    Live,
    /// No pin can settle until an EXTERNAL authority movement retires the
    /// condition — the exhausted publication generation of an installed
    /// revocation store, which a replacement install retires by swapping the
    /// store and advancing the routing epoch.
    ///
    /// Retained work stays OWED in `pending`, because the movement that clears
    /// the condition can and will complete it. Nothing is marked: the actor
    /// parks, and that same movement supplies the wake. Marking here instead is
    /// precisely the yield-paced livelock review-pass-3 §1 names — the folded
    /// `(poisoned, floor_generation)` view is STABLE under exhaustion, so the
    /// pin succeeds on every pass and the settlement refuses on every pass, with
    /// no source movement required at all.
    Fenced,
    /// The identity space itself is spent: no future pin can EVER settle.
    ///
    /// Queued work is a promise that cannot be kept, so it is discarded rather
    /// than re-queued, and the retained facts were already retired at the
    /// transition by [`NodeOrgRoutingRegistry::retire_terminal`].
    Terminal,
}

impl SourceLiveness {
    /// May a refusal against this source re-arm the actor?
    ///
    /// Only ordinary movement owns a wake. Both unsettleable states must park.
    fn may_self_wake(self) -> bool {
        matches!(self, Self::Live)
    }

    /// May a refusal against this source keep the selected identities owed?
    ///
    /// `Fenced` is recoverable, so the queue is preserved across it — losing it
    /// would leave the slots cold with nothing left to rebuild them once the
    /// replacement store lands.
    fn may_requeue(self) -> bool {
        !matches!(self, Self::Terminal)
    }
}

/// The node-owned, monotone, NON-ALIASING session/direct-state generation
/// (OLB-2B.3c step 2; design §3).
///
/// One generation for the whole node, not one per peer: a pool's annotations
/// must all come from ONE coherent view, and per-peer counters cannot express
/// that — a pool assembled from two peers at two counters is exactly the mixed
/// observation §3 forbids.
///
/// Held by the node (which advances it) and by the registry (which compares
/// it). Every read is a single atomic load and takes no lock whatsoever, which
/// is what lets the actor compare it BENEATH the registry lock without
/// touching the frozen `source commit pin → registry lock` order: an atomic
/// load acquires nothing and can re-enter nothing.
///
/// `u64::MAX` is RESERVED as the terminal marker and never handed out.
/// Exhaustion is irreversible and FAIL-CLOSED: no pool may be published under
/// it and nothing may be armed on it, because a wrapped or saturated
/// generation would let a pool built under one session view compare current
/// against an unrelated later one — the same aliasing the routing authority
/// epoch and the consumer-Grant installation identity are already fenced
/// against.
#[derive(Debug)]
pub(crate) struct SessionCurrentness {
    generation: AtomicU64,
}

impl Default for SessionCurrentness {
    fn default() -> Self {
        Self {
            generation: AtomicU64::new(0),
        }
    }
}

impl SessionCurrentness {
    pub(crate) fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// The published generation, or `None` once the space is terminally spent.
    ///
    /// LOCK-FREE. See the type doc for why that is load-bearing rather than
    /// incidental.
    pub(crate) fn generation(&self) -> Option<u64> {
        match self.generation.load(Ordering::Acquire) {
            u64::MAX => None,
            live => Some(live),
        }
    }

    /// Reserve the next generation, or `None` when reserving one would reach
    /// the terminal marker.
    ///
    /// Reserve-then-commit, exactly like the consumer-Grant publication
    /// identity: the caller publishes the new session state, and only then
    /// commits. A counter advanced before the content is visible would let an
    /// observer read a generation naming a projection it cannot see yet, and a
    /// counter advanced by an unchecked add would alias at the ceiling with
    /// the new projection ALREADY published — a partial publication.
    pub(crate) fn reserve(&self) -> Option<u64> {
        let next = self.generation.load(Ordering::Acquire).checked_add(1)?;
        (next != u64::MAX).then_some(next)
    }

    /// Commit a reservation. Called AFTER the new projection is visible.
    pub(crate) fn commit(&self, generation: u64) {
        self.generation.store(generation, Ordering::Release);
    }

    /// Latch the terminal marker. Irreversible.
    ///
    /// Reached only when [`Self::reserve`] refuses, and it is the one
    /// direction that must not fail closed by REFUSING to move: a node that
    /// cannot record that its session view changed must stop publishing pools
    /// altogether rather than keep publishing under a frozen generation.
    pub(crate) fn exhaust(&self) {
        self.generation.store(u64::MAX, Ordering::Release);
    }

    /// Test-only: position the counter so a witness can reach the terminal
    /// transition without performing 2^64 session transitions.
    #[cfg(test)]
    pub(crate) fn set_for_test(&self, generation: u64) {
        self.generation.store(generation, Ordering::Release);
    }
}

/// One coherent, immutable observation's answer for one provider.
///
/// Implemented by the node's published session projection. It is queried
/// entirely off-lock, during phase-3 reconstruction, and must therefore
/// acquire nothing and observe nothing beyond the immutable value it was
/// captured from — a lookup that consulted live state would reintroduce the
/// independently-sampled maps design §3 forbids.
pub(crate) trait SessionEligibility: Send + Sync {
    fn eligibility(&self, provider: &EntityId) -> DirectEligibility;
}

/// A projection that knows about no sessions at all.
///
/// Not a stub: a source with no session plane annotates every provider COLD,
/// which is the correct answer and the same one an unresolvable sample gets.
struct ColdSessionProjection;

impl SessionEligibility for ColdSessionProjection {
    fn eligibility(&self, _provider: &EntityId) -> DirectEligibility {
        DirectEligibility::Cold
    }
}

/// ONE coherent observation of the node's session/direct-state, captured
/// off-lock beside the source snapshot.
///
/// The generation travels INSIDE the observation rather than being read
/// separately, so the rows and the generation naming them cannot come from two
/// different samples. `currentness` is the lock-free handle the publication
/// phase re-reads to decide whether this observation is still the live one.
pub(crate) struct SessionObservation {
    /// The generation these rows were published under, or `None` when the
    /// space is terminally exhausted and no pool may be published at all.
    generation: Option<u64>,
    rows: Arc<dyn SessionEligibility>,
    currentness: Arc<SessionCurrentness>,
}

impl std::fmt::Debug for SessionObservation {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SessionObservation")
            .field("generation", &self.generation)
            .finish_non_exhaustive()
    }
}

impl SessionObservation {
    pub(crate) fn new(
        generation: Option<u64>,
        rows: Arc<dyn SessionEligibility>,
        currentness: Arc<SessionCurrentness>,
    ) -> Self {
        Self {
            generation,
            rows,
            currentness,
        }
    }

    /// A source with no session plane: everything cold, at its own private
    /// generation that nothing can move.
    fn cold() -> Self {
        Self {
            generation: Some(0),
            rows: Arc::new(ColdSessionProjection),
            currentness: SessionCurrentness::new(),
        }
    }

    fn eligibility(&self, provider: &EntityId) -> DirectEligibility {
        self.rows.eligibility(provider)
    }

    /// Is this observation still the live one?
    ///
    /// False when the projection moved, and false when the generation space is
    /// spent on EITHER side — an exhausted space can witness nothing, so it
    /// refuses rather than comparing two markers equal.
    fn still_current(&self) -> bool {
        match (self.generation, self.currentness.generation()) {
            (Some(observed), Some(live)) => observed == live,
            _ => false,
        }
    }
}

/// Where ONE provider row was discovered, exactly (design §1 artifact 3:
/// "discovery provenance for THAT scope").
///
/// Derived from the slot's exact [`ScopedDiscoveryAuthorityStamp`] rather than
/// recorded independently, so it cannot drift from the authority that made the
/// row query-visible. It is a property of the SCOPE, so it is carried once per
/// pool rather than once per row: every row in a scoped pool came through the
/// same authority by construction, and a per-row copy would be a second
/// statement of the same fact, free to disagree with the first.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ProviderProvenance {
    /// The owner plane of this exact audience scope.
    OwnerPlane,
    /// An exact installed consumer DISCOVER grant. `DISCOVER ≠ INVOKE`: this
    /// says how the provider was SEEN, and claims nothing about the caller's
    /// right to invoke it.
    GrantPlane {
        grant_id: [u8; 32],
        audience_handle: [u8; 32],
    },
}

impl ProviderProvenance {
    /// The provenance implied by the authority the rows were captured under.
    fn of(authority: &ScopedDiscoveryAuthorityStamp) -> Self {
        match authority {
            ScopedDiscoveryAuthorityStamp::Owner => Self::OwnerPlane,
            ScopedDiscoveryAuthorityStamp::Grant {
                grant_id,
                audience_handle,
                ..
            } => Self::GrantPlane {
                grant_id: *grant_id,
                audience_handle: *audience_handle,
            },
        }
    }
}

/// Whether a provider is reachable over a live session, computed under ONE
/// coherent session generation (design §3).
///
/// It ANNOTATES and never creates authority: a `Direct` row is not authorized
/// to be called, and a `Cold` row is not forbidden. 2B.3d's projection uses it
/// only to preselect among candidates its own INVOKE matching already
/// admitted.
///
/// `Cold` is the deterministic answer for absence AND for every sample that
/// could not be resolved coherently — no session, a `NodeId → EntityId`
/// mapping that moved while it was being read, or an entity that resolves to
/// more than one live session. A guess would attach a session to a provider on
/// the strength of a mutable mapping, which is the identity confusion design
/// §3's join rules exist to forbid.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum DirectEligibility {
    /// Exactly one live session, terminating at the peer's own address.
    Direct { node_id: u64, session_id: u64 },
    /// Exactly one live session, reached through a relay.
    Relayed { node_id: u64, session_id: u64 },
    /// No session, or no coherent answer. Never guessed.
    Cold,
}

impl DirectEligibility {
    /// Is this provider reachable without a relay?
    #[allow(dead_code)] // consumer: the family projection (OLB-2B.3d).
    pub(crate) fn is_direct(&self) -> bool {
        matches!(self, Self::Direct { .. })
    }
}

/// One provider of one authority-scoped pool: the discovery row, its proven
/// owner relation, and its session annotation.
///
/// Deliberately NOT a candidate. It carries no matched INVOKE grant, no caller
/// eligibility, no order relative to a caller's credentials, no selection and
/// no sensed health or score — those are 2B.3d's and OLB-4's, and a row that
/// looked like a candidate is exactly how a scoped fact becomes a caller
/// decision by accident.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct UnsensedRouteRow {
    /// The verified provider entity that announced the capability.
    pub provider: EntityId,
    /// The organization proved to own the provider by its membership
    /// certificate at ingest. Carried, never re-derived: re-deriving it here
    /// would be a second owner determination free to disagree with the one the
    /// ingest path verified.
    pub owner_org: OrgId,
    /// The announcement generation the row was learned from — provenance for
    /// THIS row, beside the pool-wide plane provenance.
    pub generation: u64,
    /// The row's own effective expiry.
    pub expires_at: u64,
    /// Reachability under the pool's ONE coherent session generation.
    pub direct: DirectEligibility,
}

/// The node-shared PRECOMPUTED routing substrate for ONE authority-scoped slot
/// (OLB-2B.3c; design §1 artifact 3) — the same `(PrivateAudienceScope,
/// capability)` key as the facts cell beside it.
///
/// Immutable, node-shared and published per slot. What it carries is exactly
/// design §1's artifact-3 row, and every field is identity as well as payload:
/// a 2B.3d reader that loaded a pool and a facts artifact with two separate
/// atomics can prove they belong together, and can refuse a pool built under a
/// session view older than the one it is validating against.
///
/// **It claims none of the following**, and the type is shaped so it cannot:
/// a matched caller INVOKE grant, caller membership or dispatcher eligibility,
/// final caller-specific authority, caller order, a selected provider, or any
/// sensed health or score. Grant-scoped rows stay Unknown/Potential until
/// SENSE exists (design §2).
#[derive(Debug)]
pub(crate) struct ScopedUnsensedRoutePool {
    /// The EXACT facts artifact this pool was derived from (design §1.1).
    ///
    /// Publication identity, compared by `Arc::ptr_eq` exactly as
    /// `invalidate_if_stale` compares facts: installation always allocates a
    /// fresh `Arc`, so pointer identity is the artifact's identity. It is what
    /// lets a 2B.3d reader verify the pool it loaded was derived from the
    /// facts it loaded — the two cells are read with two separate atomics, so
    /// the pairing must be checkable, not assumed.
    derived_from: Arc<SlotBaseFacts>,
    /// The exact authority-scoped source identity this pool speaks for.
    ///
    /// The whole key, never the capability alone: two demands under different
    /// acting identities are different pools, and a pool that could not name
    /// its own scope could be paired with another scope's facts by a reader
    /// that only compared capabilities.
    key: SlotKey,
    /// The EXACT scope authority the rows were reconstructed under — the same
    /// per-slot stamp the facts carry, restated here so a pool can be rejected
    /// on its own without re-reading the facts beside it.
    authority: ScopedDiscoveryAuthorityStamp,
    /// The whole source epoch (the exact scoped source vector) the rows were
    /// captured under.
    epoch: SourceEpoch,
    /// How this scope's providers were discovered.
    provenance: ProviderProvenance,
    /// The scoped source/provider vector, in the same deterministic provider
    /// order the facts carry.
    providers: Arc<[UnsensedRouteRow]>,
    /// The ONE coherent session generation every annotation above was computed
    /// under (design §3).
    ///
    /// This is what makes "a later session view" detectable: a reader
    /// validating against generation `g` refuses a pool stamped `< g`, and the
    /// pool-side invalidator orders by exactly this.
    session_generation: u64,
    /// The earliest instant this pool stops being usable — the same bound its
    /// facts carry (`min(provider rows, installed DISCOVER authority)`),
    /// restated so a pool can be validated on its own.
    ///
    /// Identical to `derived_from.earliest_expiry` BY CONSTRUCTION, which is
    /// why the actor's deadline arm reads the facts plane only: a second
    /// deadline source would be a second statement of one fact, free to
    /// disagree with the first, and arming on it would be a comparison that
    /// can never fail. `u64::MAX` means nothing here can expire.
    earliest_deadline: u64,
}

#[allow(dead_code)] // consumer: the family projection (OLB-2B.3d).
impl ScopedUnsensedRoutePool {
    /// The exact facts artifact this pool derives from.
    pub(crate) fn derived_from(&self) -> &Arc<SlotBaseFacts> {
        &self.derived_from
    }

    /// Does this pool derive from EXACTLY `facts`?
    ///
    /// Pointer identity, matching `invalidate_if_stale`: every installation
    /// allocates a fresh `Arc`, so this is the artifact's identity and not a
    /// content comparison that two equal reconstructions could both satisfy.
    pub(crate) fn derives_from(&self, facts: &Arc<SlotBaseFacts>) -> bool {
        Arc::ptr_eq(&self.derived_from, facts)
    }

    pub(crate) fn key(&self) -> &SlotKey {
        &self.key
    }

    pub(crate) fn authority(&self) -> &ScopedDiscoveryAuthorityStamp {
        &self.authority
    }

    pub(crate) fn epoch(&self) -> SourceEpoch {
        self.epoch
    }

    pub(crate) fn provenance(&self) -> ProviderProvenance {
        self.provenance
    }

    pub(crate) fn providers(&self) -> &[UnsensedRouteRow] {
        &self.providers
    }

    /// The ONE session generation every annotation was computed under.
    pub(crate) fn session_generation(&self) -> u64 {
        self.session_generation
    }

    pub(crate) fn earliest_deadline(&self) -> u64 {
        self.earliest_deadline
    }

    /// Test-only: a pool naming `derived_from` and nothing else, for the
    /// lifecycle witnesses signed at step 1. The production constructor is
    /// [`PreparedRoutePool::seal`].
    #[cfg(test)]
    pub(crate) fn for_test(derived_from: Arc<SlotBaseFacts>) -> Self {
        Self::for_test_at_session(derived_from, 0)
    }

    /// Test-only: the same, at an EXPLICIT session generation.
    ///
    /// The generation has to be nameable, and that is a mutation-evidence
    /// requirement rather than an ergonomic one. A read-seam witness that
    /// wants only the pool/facts PAIRING check to be able to refuse its
    /// artifact must hand that artifact the LIVE generation — otherwise the
    /// session check refuses it too, and dropping the pairing check leaves the
    /// witness green. That is exactly what happened: mutation M-Z18 survived
    /// against a fixture hard-coded to generation 0, and the survivor was a
    /// true statement about the witness rather than about the production code
    /// (design §18.0c).
    #[cfg(test)]
    pub(crate) fn for_test_at_session(
        derived_from: Arc<SlotBaseFacts>,
        session_generation: u64,
    ) -> Self {
        Self {
            key: SlotKey {
                scope: PrivateAudienceScope(CapabilityAudienceScope::Owner {
                    org_id: OrgId::from_bytes([0; 32]),
                    audience_handle: [0; 32],
                }),
                capability: CapabilityAuthorityId::for_tag("test:pool"),
            },
            authority: derived_from.authority,
            epoch: derived_from.epoch,
            provenance: ProviderProvenance::of(&derived_from.authority),
            providers: Arc::from(Vec::new()),
            session_generation,
            earliest_deadline: derived_from.earliest_expiry,
            derived_from,
        }
    }
}

/// A COMPLETE pool, built entirely off-lock, waiting only to be bound to the
/// facts artifact it derives from.
///
/// The split exists because of the ordering the design demands: the pool names
/// the exact `Arc<SlotBaseFacts>` it was derived from, and that `Arc` is
/// allocated at installation. Everything that reads the source or the session
/// projection — the row join, the session annotation, the deterministic order —
/// happens here, in phase 3, holding nothing. [`Self::seal`] then binds the
/// finished payload to the artifact; it reads no source, takes no lock, and
/// copies no row.
#[derive(Debug)]
struct PreparedRoutePool {
    key: SlotKey,
    authority: ScopedDiscoveryAuthorityStamp,
    provenance: ProviderProvenance,
    providers: Arc<[UnsensedRouteRow]>,
    session_generation: u64,
}

impl PreparedRoutePool {
    /// Bind the finished payload to the exact facts artifact it derives from.
    fn seal(self, derived_from: Arc<SlotBaseFacts>) -> Arc<ScopedUnsensedRoutePool> {
        Arc::new(ScopedUnsensedRoutePool {
            key: self.key,
            authority: self.authority,
            epoch: derived_from.epoch,
            provenance: self.provenance,
            providers: self.providers,
            session_generation: self.session_generation,
            earliest_deadline: derived_from.earliest_expiry,
            derived_from,
        })
    }
}

/// Where a lock-free reader may be scheduled between the two adjacent
/// publication stores.
///
/// Both gaps are observation points because both are reachable: the invariant
/// is that NO reader can observe a pool beside facts it was not derived from,
/// and a witness that samples only one gap proves only half of it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum PublicationGap {
    /// After the superseded pool was cleared, before the facts were installed.
    PoolCleared,
    /// After the facts were installed, before their pool was published.
    FactsInstalled,
}

/// Test-only observation of the publication-order interleaving.
///
/// A production build has NO field here, so `observe` compiles to nothing.
/// This exists because the pool-first ordering inside [`SlotCells`] cannot be
/// distinguished by any final-state assertion — both stores complete before
/// any assertion runs — so the only evidence is a reader timed BETWEEN them
/// (design §18.0, the step-1 debt this slice owes).
#[derive(Default)]
struct PublicationObserver {
    #[cfg(test)]
    #[allow(clippy::type_complexity)]
    hook: parking_lot::Mutex<Option<Arc<dyn Fn(PublicationGap) + Send + Sync>>>,
}

impl PublicationObserver {
    #[inline]
    fn observe(&self, _gap: PublicationGap) {
        #[cfg(test)]
        {
            let hook = self.hook.lock().clone();
            if let Some(hook) = hook {
                hook(_gap);
            }
        }
    }
}

/// Both publication cells of one slot, moved and cloned as a UNIT.
///
/// `DemandHandle` clones both cells under the same registry acquisition that
/// takes the slot reference (design §1.1) — holding them as one value is what
/// makes that structural rather than a discipline every call site re-earns.
/// The same goes for the mutation ordering below: every site that moves the
/// facts cell goes through [`Self::take_facts`] or [`Self::install_facts`], so
/// the pool-first rule has one implementation, not eight.
#[derive(Clone, Debug)]
struct SlotCells {
    /// `None` until a recapture installs facts, and cleared the moment anything
    /// invalidates them. `None` IS the deterministic cold outcome.
    ///
    /// An `ArcSwapOption` CELL rather than a plain field (OLB-2B.3): every
    /// MUTATION still happens under the registry lock exactly as before, so the
    /// install/invalidate ordering the E3c closure signed is unchanged. What the
    /// cell adds is a lock-free READ — a `DemandHandle` clones this `Arc` at
    /// demand time, so the warmed path loads the artifact with one atomic and
    /// never contends with the actor's quantum. That is the hot-path contract
    /// (plan pins 7/8: the warmed call is an ArcSwap load, not a lock).
    ///
    /// Cell identity is per SLOT INCARNATION: a retired slot's entry is dropped
    /// with the slot, so a handle from a dead incarnation cannot be revived. It
    /// cannot be stale either — a live handle is precisely what stops the slot
    /// being retired.
    facts: Arc<ArcSwapOption<SlotBaseFacts>>,
    /// The 2B.3c routing-pool cell, beside the facts cell it derives from —
    /// a SECOND cell, not a changed one (design §1.1): the signed 2B.3a facts
    /// cell's semantics are untouched.
    ///
    /// INVARIANT: a published pool names the currently published facts. Every
    /// mutation of `facts` clears this cell FIRST (see [`Self::take_facts`] /
    /// [`Self::install_facts`]), so a lock-free reader can observe facts
    /// without a pool — the ordinary cold-pool state — but never a pool beside
    /// facts it was not derived from.
    unsensed: Arc<ArcSwapOption<ScopedUnsensedRoutePool>>,
}

impl SlotCells {
    fn empty() -> Self {
        Self {
            facts: Arc::new(ArcSwapOption::empty()),
            unsensed: Arc::new(ArcSwapOption::empty()),
        }
    }

    /// Swap the facts artifact out — POOL FIRST (design §1.1).
    ///
    /// A pool is derived from exactly one facts artifact, so whichever facts
    /// this removes, any published pool named either that artifact or an older
    /// one; both are invalid the moment the facts move. Clearing the pool
    /// before the facts means no lock-free reader ever observes a pool whose
    /// basis is already gone. Returns the removed facts, so conditional
    /// callers keep their metrics exact.
    fn take_facts(&self) -> Option<Arc<SlotBaseFacts>> {
        self.unsensed.store(None);
        self.facts.swap(None)
    }

    /// Drop ONLY the routing pool, leaving the facts exactly as they are —
    /// the POOL-ONLY half of the asymmetric contract (design §1.1).
    ///
    /// Session projection moving does not make discovery facts stale, and
    /// clearing them would re-queue a full reconstruction for a change that
    /// touched no discovery row. Returns the removed pool.
    fn take_pool(&self) -> Option<Arc<ScopedUnsensedRoutePool>> {
        self.unsensed.swap(None)
    }

    /// Install a fresh facts artifact with NO pool beside it — POOL FIRST, for
    /// the same reason: a pool present at this moment was derived from the
    /// SUPERSEDED facts, and storing the new facts first would let a reader
    /// pair them with it.
    ///
    /// Test-only. Production installs go through
    /// [`Self::install_facts_and_pool`], which is the same ordering with the
    /// pool half filled in — there is no production path that publishes facts
    /// and then decides separately about their pool.
    #[cfg(test)]
    fn install_facts(&self, facts: Arc<SlotBaseFacts>) {
        self.install_facts_and_pool(facts, None, &PublicationObserver::default());
    }

    /// Publish `facts` and, when the build stayed current, the pool derived
    /// from them — in the ONE order in which no lock-free reader can observe a
    /// mismatched pair (design §1.1, §18.0's deferred publication-order
    /// obligation).
    ///
    /// ```text
    /// unsensed <- None        a reader here sees (old facts, no pool)
    /// facts    <- new         a reader here sees (new facts, no pool)
    /// unsensed <- new pool    a reader here sees (new facts, its own pool)
    /// ```
    ///
    /// Every intermediate state is a COLD-POOL state, which is exactly the
    /// state a reader must already handle. What is unreachable is the only
    /// dangerous one: a pool beside facts it was not derived from. The
    /// ordering has ONE implementation, here, so no publication site can get
    /// it individually wrong.
    fn install_facts_and_pool(
        &self,
        facts: Arc<SlotBaseFacts>,
        pool: Option<Arc<ScopedUnsensedRoutePool>>,
        observer: &PublicationObserver,
    ) {
        debug_assert!(
            pool.as_ref()
                .is_none_or(|pool| Arc::ptr_eq(&pool.derived_from, &facts)),
            "a published pool names the facts it is published beside"
        );
        self.unsensed.store(None);
        observer.observe(PublicationGap::PoolCleared);
        self.facts.store(Some(facts));
        observer.observe(PublicationGap::FactsInstalled);
        if pool.is_some() {
            self.unsensed.store(pool);
        }
    }
}

#[derive(Debug)]
struct Slot {
    /// Allocated fresh from the node-wide monotone id space at creation, so a
    /// retired-and-re-demanded slot never reuses an identity and work in flight
    /// for a previous incarnation can never resurrect it.
    incarnation: u64,
    /// Live demand handles across ALL families.
    #[allow(dead_code)]
    refs: usize,
    /// The slot's two publication cells — see [`SlotCells`].
    cells: SlotCells,
}

impl Slot {
    /// A fresh slot with one reference and both cells empty.
    fn new(incarnation: u64) -> Self {
        Self {
            incarnation,
            refs: 1,
            cells: SlotCells::empty(),
        }
    }
}

#[derive(Default)]
struct RegistryInner {
    slots: BTreeMap<SlotKey, Slot>,
    /// Capability → retained slot identities, so `Caps(C)` touches only C's
    /// buckets. One capability can back several slots (one per audience scope).
    slots_by_capability: BTreeMap<CapabilityAuthorityId, BTreeSet<SlotKey>>,
    /// Per family, its handle count per key. Bounds HANDLES, not distinct keys, so
    /// duplicate demand cannot bypass the bound.
    #[allow(dead_code)]
    families: BTreeMap<FamilyId, BTreeMap<SlotKey, usize>>,
    /// AUTHORITATIVE pending slot identities — the work queue proper.
    pending: BTreeSet<SlotKey>,
    /// Monotone allocator shared by slot incarnations and family identities. Never
    /// reused, never wrapped.
    next_id: u64,
    /// The ONE actor incarnation permitted to consume pending work and install
    /// facts, or `None` when no actor is live.
    live_actor: Option<u64>,
    /// Whether a COMPLETE recapture is under way.
    ///
    /// While open, its remaining set is `pending`, and settlement additionally
    /// re-queues every retained slot that is not coherent with the commit — which
    /// is how a source generation that moved between quanta restarts the epoch. A
    /// later `RebuildAll` must NOT re-expand while it is open: re-expanding would
    /// re-select the first quantum forever and the recapture would never terminate.
    recapture_open: bool,
    /// The last slot identity a quantum served, so the next one resumes strictly
    /// after it and wraps (review-pass-3 §9). Fair rotation over `pending`
    /// rather than "the first `APPLY_QUANTUM` in key order, every pass".
    select_cursor: Option<SlotKey>,
}

/// What giving up one family reference did.
///
/// Three outcomes rather than a bool, because "nothing was given back" and
/// "something was given back and the slot survives" must not collapse: the first
/// is an invariant violation the caller has to refuse on, and the second is the
/// ordinary case.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ReleaseOutcome {
    /// The reference was given back and the slot still has owners.
    Released,
    /// The reference was given back and it was the LAST one.
    Retire,
    /// This family holds no such reference. NOTHING was decremented.
    NotHeld,
}

impl RegistryInner {
    #[allow(dead_code)]
    fn family_handles(&self, family: FamilyId) -> usize {
        self.families
            .get(&family)
            .map_or(0, |keys| keys.values().sum())
    }

    /// Give up ONE of `family`'s references to `key`.
    ///
    /// Bookkeeping only: the caller commits the retirement, because retiring
    /// touches the metrics and the capacity generation that live on the registry
    /// rather than on this struct. Shared by the single-key release, the
    /// whole-set release and the replacement transaction, so the three cannot
    /// drift in how a reference is given up.
    ///
    /// **The family reference is what authorizes the global decrement**, and
    /// this helper enforces that itself rather than trusting every present and
    /// future caller. A family that does not hold `key` has nothing to give
    /// back, and decrementing `slot.refs` on its behalf would retire a slot some
    /// OTHER family still owns — leaving that family's map claiming a handle it
    /// can never release, and its set unable to release it. The check is here,
    /// not at the call sites, because a call site can be added; this cannot be
    /// bypassed by adding one.
    fn release_one(&mut self, family: FamilyId, key: &SlotKey) -> ReleaseOutcome {
        let held = self
            .families
            .get(&family)
            .and_then(|keys| keys.get(key))
            .copied()
            .unwrap_or(0);
        if held == 0 {
            return ReleaseOutcome::NotHeld;
        }
        if let Some(keys) = self.families.get_mut(&family) {
            if let Some(count) = keys.get_mut(key) {
                *count -= 1;
                if *count == 0 {
                    keys.remove(key);
                }
            }
            if keys.is_empty() {
                self.families.remove(&family);
            }
        }
        match self.slots.get_mut(key) {
            Some(slot) => {
                slot.refs -= 1;
                if slot.refs == 0 {
                    ReleaseOutcome::Retire
                } else {
                    ReleaseOutcome::Released
                }
            }
            // The family held a reference the slot map does not know about.
            // Unreachable, and NOT silently absorbed: the family entry has been
            // given back above, so the two are consistent again, but the caller
            // is told nothing retired because nothing did.
            None => ReleaseOutcome::Released,
        }
    }

    /// Allocate the next identity, or `None` when the space is exhausted.
    fn allocate_id(&mut self) -> Option<u64> {
        let next = self.next_id.checked_add(1)?;
        self.next_id = next;
        Some(next)
    }

    /// Drop `key`'s facts — and, pool first, the routing pool derived from
    /// them — reporting whether facts were actually invalidated.
    fn invalidate(&mut self, key: &SlotKey) -> bool {
        self.slots
            .get_mut(key)
            .is_some_and(|slot| slot.cells.take_facts().is_some())
    }

    /// Queue every retained slot and drop every fact — the expansion step of a
    /// recapture epoch and of a fresh actor incarnation.
    fn invalidate_and_queue_all(&mut self) -> u64 {
        let keys: Vec<SlotKey> = self.slots.keys().cloned().collect();
        let mut invalidated = 0;
        for key in keys {
            if self.invalidate(&key) {
                invalidated += 1;
            }
            self.pending.insert(key);
        }
        invalidated
    }

    /// Retained slots NOT coherent with this commit.
    ///
    /// Coherent means: facts present, stamped by the live actor incarnation, for
    /// the slot's current incarnation, from this source generation. Used ONLY
    /// while a complete-recapture epoch is open — ordinary `Caps` and first-demand
    /// work deliberately leaves unrelated slots at older generations, and demanding
    /// global coherence there would report `Progress` for a pass that finished
    /// everything it was asked to do.
    fn incoherent_with(&self, incarnation: u64, epoch: SourceEpoch) -> Vec<SlotKey> {
        self.slots
            .iter()
            .filter(|(_, slot)| {
                !slot.cells.facts.load().as_ref().is_some_and(|facts| {
                    facts.actor_incarnation == incarnation
                        && facts.slot_incarnation == slot.incarnation
                        && facts.epoch == epoch
                })
            })
            .map(|(key, _)| key.clone())
            .collect()
    }
}

/// Observable registry counters.
#[derive(Default)]
pub(crate) struct RegistryMetrics {
    refused_family_at_capacity: AtomicU64,
    refused_node_at_capacity: AtomicU64,
    refused_id_space_exhausted: AtomicU64,
    slots_retired: AtomicU64,
    installs: AtomicU64,
    discarded_obsolete: AtomicU64,
    facts_invalidated: AtomicU64,
    stale_actor_rejections: AtomicU64,
    recaptures_restarted: AtomicU64,
    /// Passes whose commit pin was held by a LIVE actor and whose settlement was
    /// still refused, because the source's own publication moved underneath it
    /// (review-pass-3 §8). Distinct from `stale_actor_rejections`, which is actor
    /// LIFECYCLE churn: conflating them steers an operator at supervisor restarts
    /// during what is actually revocation-publication pressure.
    settlements_refused: AtomicU64,
    /// Routing pools PUBLISHED beside their facts (OLB-2B.3c step 2).
    pools_published: AtomicU64,
    /// Pools retired by pool-only invalidation — session projection moved
    /// while the discovery facts stayed current.
    pools_invalidated: AtomicU64,
    /// Builds whose facts installed but whose POOL was discarded because the
    /// session view moved underneath them, or because the session generation
    /// space is spent.
    ///
    /// Distinct from `settlements_refused` and from `discarded_obsolete`: the
    /// pass installed real facts and made real progress, and only the session
    /// annotation was refused. Conflating it with either would report a
    /// converging plane as a failing one.
    pools_refused_stale_session: AtomicU64,
}

impl RegistryMetrics {
    pub(crate) fn refused_family_at_capacity(&self) -> u64 {
        self.refused_family_at_capacity.load(Ordering::Acquire)
    }
    pub(crate) fn refused_node_at_capacity(&self) -> u64 {
        self.refused_node_at_capacity.load(Ordering::Acquire)
    }
    pub(crate) fn refused_id_space_exhausted(&self) -> u64 {
        self.refused_id_space_exhausted.load(Ordering::Acquire)
    }
    pub(crate) fn slots_retired(&self) -> u64 {
        self.slots_retired.load(Ordering::Acquire)
    }
    pub(crate) fn installs(&self) -> u64 {
        self.installs.load(Ordering::Acquire)
    }
    pub(crate) fn discarded_obsolete(&self) -> u64 {
        self.discarded_obsolete.load(Ordering::Acquire)
    }
    pub(crate) fn facts_invalidated(&self) -> u64 {
        self.facts_invalidated.load(Ordering::Acquire)
    }
    pub(crate) fn stale_actor_rejections(&self) -> u64 {
        self.stale_actor_rejections.load(Ordering::Acquire)
    }
    pub(crate) fn recaptures_restarted(&self) -> u64 {
        self.recaptures_restarted.load(Ordering::Acquire)
    }
    pub(crate) fn settlements_refused(&self) -> u64 {
        self.settlements_refused.load(Ordering::Acquire)
    }
    /// The three pool counters are WITNESS surfaces for as long as the second
    /// cell is dark. The counters themselves are production state — every one
    /// of them is advanced by the production apply path — but nothing outside a
    /// witness can act on them until 2B.3d gives the plane a consumer, and an
    /// operator-facing surface published before then would describe a plane
    /// that never moves. Widening the existing `pub` metrics array instead
    /// would be a source-breaking public API change, which this slice excludes.
    #[cfg(test)]
    pub(crate) fn pools_published(&self) -> u64 {
        self.pools_published.load(Ordering::Acquire)
    }
    #[cfg(test)]
    pub(crate) fn pools_invalidated(&self) -> u64 {
        self.pools_invalidated.load(Ordering::Acquire)
    }
    #[cfg(test)]
    pub(crate) fn pools_refused_stale_session(&self) -> u64 {
        self.pools_refused_stale_session.load(Ordering::Acquire)
    }
}

/// The node's bounded routing registry.
pub(crate) struct NodeOrgRoutingRegistry {
    inner: parking_lot::Mutex<RegistryInner>,
    source: Arc<dyn SlotSource>,
    work: Arc<RegistryWork>,
    metrics: Arc<RegistryMetrics>,
    /// Advances every time a retained slot is RETIRED, and only then
    /// (OLB-2B.3b §9).
    ///
    /// `NodeAtCapacity` is the one retryable refusal, and this is what a
    /// refused family gates its retry on: node capacity can only have become
    /// available by a slot going away. Without it a family that missed at the
    /// node bound would re-derive its demand set, re-sort it and re-take the
    /// registry lock on EVERY subsequent cold call for that capability — the
    /// bound is node-wide, so a node at 256 slots would have every family
    /// hammering the one lock the warmed path exists to avoid.
    ///
    /// Deliberately NOT `slots.len()`: a retire-then-demand pair leaves the
    /// length equal while capacity genuinely moved, and a family comparing
    /// lengths would conclude nothing had changed and never retry. It is also
    /// NOT advanced on demand — growth cannot free capacity, so advancing there
    /// would only manufacture pointless retries.
    ///
    /// Read lock-free, mutated under the registry lock. Publishing it AFTER the
    /// retirement is committed (the `Release` half of the `AcqRel`) is what
    /// makes an observer that sees the new generation able to see the freed
    /// slot too.
    node_capacity_generation: AtomicU64,
    /// Advances on EVERY reference given back, retirement or not.
    ///
    /// The node capacity generation is the wake condition for a FRESH
    /// acquisition, and it is exact for that: only a retirement can make room
    /// for a new slot. It is NOT exact for a REPLACEMENT, which can become
    /// satisfiable when another family merely stops SHARING a slot the
    /// replacement wants to give up — the slot survives, no retirement happens,
    /// and the capacity generation stands still while the replacement's
    /// projection has genuinely changed. Gating replacement on the capacity
    /// generation therefore deadlocks it against its own wake condition.
    ///
    /// Read lock-free, mutated under the registry lock, published after the
    /// bookkeeping it describes.
    ref_release_generation: AtomicU64,
    /// Test-only observation of the two adjacent publication stores. Empty in
    /// a production build — see [`PublicationObserver`].
    publication_observer: PublicationObserver,
}

/// A clone family: one private identity and one shared 64-handle budget.
///
/// Cloning shares the identity — that is the point, and it is why the identity is
/// not a caller-supplied integer. Independent families are minted by
/// [`NodeOrgRoutingRegistry::new_family`] and can neither name nor spend one
/// another's budget.
// ---------------------------------------------------------------------------
// The DEMAND surface.
//
// Everything below has no in-crate production caller, and that is a reviewed
// SCOPE decision rather than a missing consumer: the path that holds demand
// handles is the warmed-call consumer, which is deliberately outside the OLB-2B
// entry boundary (Kyra). The registry's REAL consumer — the supervised actor
// driving `DirtyApply` — is fully live and node-wired in E3c.
//
// OLB-2B.3b moved the boundary one layer, and no further: `org_routing_state`
// now owns this surface, so `demand_set` and `RoutingFamily` have an in-crate
// consumer. That consumer is itself dark, because what reaches it is
// `MeshNode::call` in OLB-2B.3d. The allows below therefore stay, and they stay
// for a reason that is now one hop away rather than two.
//
// These allows are therefore permanent-until-warmed-calls and item-scoped. They
// are NOT the E1/E2/E3a/E3b module-wide allowances, which are gone.
// ---------------------------------------------------------------------------
#[allow(dead_code)]
#[derive(Clone)]
pub(crate) struct RoutingFamily {
    registry: Arc<NodeOrgRoutingRegistry>,
    id: FamilyId,
}

#[allow(dead_code)]
impl RoutingFamily {
    /// Register demand for `key` under THIS family's budget.
    pub(crate) fn demand(&self, key: SlotKey) -> Result<DemandHandle, DemandRefused> {
        self.registry.demand(self.id, key)
    }

    /// Acquire a COMPLETE demand set atomically, or nothing (OLB-2B.3b §4.1).
    pub(crate) fn demand_set(&self, keys: Vec<SlotKey>) -> Result<DemandSet, DemandRefused> {
        self.registry.demand_set(self.id, keys)
    }

    /// Handles this family currently holds.
    pub(crate) fn handles(&self) -> usize {
        self.registry.inner.lock().family_handles(self.id)
    }

    /// The node capacity generation a FRESH acquisition's refusals gate on.
    pub(crate) fn node_capacity_generation(&self) -> u64 {
        self.registry.node_capacity_generation()
    }

    /// The reference-release generation a REPLACEMENT's refusals gate on.
    pub(crate) fn ref_release_generation(&self) -> u64 {
        self.registry.ref_release_generation()
    }
}

/// A live demand for one authority-scoped key, held by one family. Releases on
/// drop; the LAST reference retires the slot.
#[allow(dead_code)]
pub(crate) struct DemandHandle {
    registry: Arc<NodeOrgRoutingRegistry>,
    family: FamilyId,
    key: SlotKey,
    /// The slot's publication cells — BOTH of them — cloned once at demand
    /// time, under the same registry acquisition that took the slot reference
    /// (OLB-2B.3; design §1.1).
    ///
    /// This is what makes the warmed read lock-free: the handle already names
    /// exactly one slot, so the hot path needs no map lookup and no registry
    /// lock — one atomic load and it holds the immutable artifact. Holding the
    /// cells is sound precisely because holding the handle is what prevents
    /// the slot being retired, so neither cell can be a dead incarnation's.
    cells: SlotCells,
}

impl DemandHandle {
    /// The slot's currently published artifact, or `None` for the cold outcome.
    ///
    /// UNVALIDATED, exactly like its registry-side twin: this returns whatever
    /// is published, and authority revalidation is the NODE seam's job
    /// (`MeshNode::org_routing_base_facts`). Named to say so — the review that
    /// renamed `base_facts` to `base_facts_unvalidated` did it because an
    /// accessor that looks authoritative and is not is a trap, and adding a
    /// lock-free twin would re-lay it under a friendlier name.
    ///
    /// The allow is scoped to this ONE method and names its consumer: the
    /// FAMILY PROJECTION, which reads the contributing artifacts and narrows
    /// them into one route set. That is OLB-2B.3d, not 2B.3b — 2B.3b's
    /// `CapabilityRouteHandle` OWNS the demand set without reading a single
    /// artifact through it, which is exactly why the allow survived that slice.
    /// Per the E3c discipline, a leftover allow here after 2B.3d lands means
    /// that consumer never arrived.
    #[allow(dead_code)]
    pub(crate) fn base_facts_unvalidated(&self) -> Option<Arc<SlotBaseFacts>> {
        self.cells.facts.load_full()
    }

    /// The slot's currently published routing pool, or `None` for the cold
    /// outcome — UNVALIDATED, exactly like [`Self::base_facts_unvalidated`],
    /// and read with a SEPARATE atomic: a consumer that needs the pool and the
    /// facts to agree must check `ScopedUnsensedRoutePool::derived_from`
    /// against the facts it loaded rather than assume two loads paired.
    #[allow(dead_code)] // consumer: the family projection (OLB-2B.3d).
    pub(crate) fn unsensed_pool_unvalidated(&self) -> Option<Arc<ScopedUnsensedRoutePool>> {
        self.cells.unsensed.load_full()
    }
}

impl Drop for DemandHandle {
    fn drop(&mut self) {
        self.registry.release(self.family, &self.key);
    }
}

/// A family's COMPLETE demand set for one capability, owned as ONE object.
///
/// The unit of ownership is the SET, not the handle, and that is what makes an
/// atomic replacement expressible: a replacement compares two key sets in one
/// registry transaction and moves the release responsibility across, so the
/// intersection is never released and re-acquired and nothing is ever charged
/// twice. A `Vec<DemandHandle>` cannot express it — each handle releases itself
/// on drop, so transferring one means defusing its `Drop`, and a per-handle
/// defuse is exactly the double-release hazard this type removes.
#[allow(dead_code)] // consumer: the family projection (OLB-2B.3d).
pub(crate) struct DemandSet {
    registry: Arc<NodeOrgRoutingRegistry>,
    family: FamilyId,
    /// What this set NAMES, in deterministic key order.
    ///
    /// Immutable, and read on the lock-free path. Distinct from `held`, which is
    /// what it still OWES a release for; after a replacement transfers the
    /// references away the two differ, and that difference is the whole point.
    keys: Vec<SlotKey>,
    /// Publication cells — BOTH per contributor (design §1.1) — parallel to
    /// `keys`, cloned under the acquisition that retained the references. One
    /// element per key rather than two parallel vectors, so the
    /// `keys[i] <-> cells[i]` pairing cannot misalign between the planes.
    ///
    /// Immutable and lock-free readable. Holding a cell is sound whatever
    /// happens to the slot: a retired slot's cells simply read empty, which is
    /// the correct answer for a superseded set.
    cells: Vec<SlotCells>,
    /// The RELEASE RESPONSIBILITY — the keys this set must still give back.
    ///
    /// Interior-mutable because the responsibility can move to a replacement set
    /// while this one is still shared behind an `Arc`. There is no flag and no
    /// `mem::forget`: the invariant is that a set releases precisely the keys it
    /// contains at drop, and a transfer MOVES them, so at every instant exactly
    /// one set owes each reference.
    held: parking_lot::Mutex<Vec<SlotKey>>,
}

#[allow(dead_code)] // consumer: the family projection (OLB-2B.3d).
impl DemandSet {
    /// The scopes this set names, in deterministic key order.
    pub(crate) fn keys(&self) -> &[SlotKey] {
        &self.keys
    }

    pub(crate) fn len(&self) -> usize {
        self.keys.len()
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.keys.is_empty()
    }

    /// The published artifact for the `index`-th contributor, UNVALIDATED —
    /// exactly like its registry-side twin. Authority revalidation is the NODE
    /// seam's job (`MeshNode::org_routing_base_facts`); the name says so.
    pub(crate) fn base_facts_unvalidated(&self, index: usize) -> Option<Arc<SlotBaseFacts>> {
        self.cells.get(index)?.facts.load_full()
    }

    /// The published routing pool for the `index`-th contributor, UNVALIDATED,
    /// read with a SEPARATE atomic from the facts — pairing is checked via
    /// `ScopedUnsensedRoutePool::derived_from`, never assumed.
    #[allow(dead_code)] // consumer: the family projection (OLB-2B.3d).
    pub(crate) fn unsensed_pool_unvalidated(
        &self,
        index: usize,
    ) -> Option<Arc<ScopedUnsensedRoutePool>> {
        self.cells.get(index)?.unsensed.load_full()
    }

    /// Test-only: what this set still owes a release for. A replacement moves
    /// this to the successor, and a witness that could not see it would have to
    /// infer the transfer from a handle count that a double-release also
    /// produces.
    #[cfg(test)]
    pub(crate) fn held_for_test(&self) -> Vec<SlotKey> {
        self.held.lock().clone()
    }

    /// Atomically REPLACE this set with the one for `new_keys`, charged on the
    /// projected final footprint (§4.3).
    ///
    /// **Rooted in this set's OWN registry and family identity.** It takes no
    /// `RoutingFamily` argument, and that is the security property, not an
    /// ergonomic one: an earlier surface took the replacement authority from the
    /// CALLER, so family B could pass family A's set and have the registry
    /// release A's references under B's id. B's family map holds no such key, so
    /// nothing was given back on B's side while A's slot reference was still
    /// decremented — retiring a slot A owned, stranding a handle in A's map that
    /// A's set could no longer release, and handing B the successor. Cross-family
    /// and cross-registry transfer are now unrepresentable rather than rejected,
    /// because there is no argument through which to express them.
    ///
    /// A set whose responsibility has ALREADY transferred cannot be replaced
    /// again: `held` no longer matches what the set names, and a second
    /// replacement would compute `old_only = ∅` and charge the successor's whole
    /// footprint gross. That is refused as [`ReplaceRefused::Superseded`] with
    /// total no effect.
    pub(crate) fn replace(&self, new_keys: Vec<SlotKey>) -> Result<DemandSet, ReplaceRefused> {
        self.registry
            .clone()
            .replace_demand_set(self.family, self, new_keys)
    }
}

/// Why an atomic replacement did not happen.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[allow(dead_code)] // consumer: `OrgRoutingState` (OLB-2B.3b).
pub(crate) enum ReplaceRefused {
    /// The registry refused the projected footprint.
    Demand(DemandRefused),
    /// This set has already given its references away, so it is not the current
    /// owner of anything and cannot be the basis of a replacement.
    Superseded,
}

impl Drop for DemandSet {
    fn drop(&mut self) {
        // Whatever is still owed, released as one transaction. After a
        // replacement transferred them this is empty and the drop is a no-op —
        // not because a flag says so, but because the keys are gone.
        let owed = std::mem::take(&mut *self.held.lock());
        if !owed.is_empty() {
            self.registry.release_keys(self.family, &owed);
        }
    }
}

/// Where a reconstruction sits in the consumer-Grant transition order.
///
/// The ARTIFACT side of the fence, distinct from the movement side below.
///
/// `TerminalAbsence` is not "publication `u64::MAX`". A single global terminal
/// generation would be wrong the moment a second Grant scope is still installed:
/// scope A withdraws terminally and sets the global marker, scope B's
/// reconstruction observes the same marker, and B's own later withdrawal can no
/// longer order B's pre-terminal `Served` artifact. Terminal absence is a
/// property of ONE scope's reconstruction, so it lives on the artifact, and
/// still-installed scopes keep an ordinary `Publication` (Kyra, review of
/// `46af3d625`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum GrantArtifactFence {
    /// Reconstructed while the publication-identity space was still live, under
    /// the generation it observed.
    Publication(u64),
    /// Reconstructed ABSENT for this exact scope, with the publication-identity
    /// space exhausted. Terminal: no installation can follow, so nothing can
    /// supersede it.
    TerminalAbsence,
}

/// How a consumer-Grant transition orders itself against retained artifacts.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum GrantMovementFence {
    /// The ordinary case: this transition's publication generation.
    Publication(u64),
    /// The publication-identity space is EXHAUSTED (OLB-2B.3c-pre step 3).
    ///
    /// Reached only by a withdrawal, because an installation at exhaustion is
    /// refused before it publishes anything. With no further installation
    /// publishable, absence is TERMINAL for this scope, so no `Publication`
    /// artifact under it has a successor left to protect — and ordering would be
    /// meaningless anyway, since the generation that would express it cannot be
    /// allocated.
    ///
    /// **This clears every `Publication` artifact in the exact scope; it is NOT
    /// "clear everything".** A terminal movement still preserves
    /// [`GrantArtifactFence::TerminalAbsence`]. Nothing can supersede a terminal
    /// absence — but the artifact THIS withdrawal's own publication produced is
    /// one, and retiring it is the same self-inflicted churn `<=` would cause on
    /// the ordinary row. "Unconditional" describes the comparison against
    /// generations, not the set of artifacts (Kyra, review of `010c718ea`).
    ///
    /// Revocation is never refused for want of an identity. Withdrawing
    /// authority must always be possible; it is the direction that fails closed.
    Terminal,
}

/// One consumer-Grant transition, carried to routing with enough identity to
/// invalidate CONDITIONALLY (OLB-2B.3c-pre items 10/11).
///
/// A bare `grant_id` is not enough, and the gap is a real race rather than a
/// tidiness point: a notification can be delayed arbitrarily between its
/// publication and its registry work, so an OBSOLETE transition can arrive after
/// a newer installation has already been published, rebuilt and warmed. Clearing
/// by id alone destroys that successor.
///
/// ```text
/// A: remove N    publish absence, release gate, [stall]
/// B: install N+1                 publish, notify, actor warms the N+1 artifact
/// A: [resumes]   clear by grant_id  ->  destroys the N+1 artifact
/// ```
///
/// It never resurrects withdrawn authority — the read seam stays fail-closed —
/// but it lets an obsolete transition retire current work, which is the defect
/// class `invalidate_if_stale` already guards one layer up: **a delayed
/// invalidator must not delete a successor.**
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct GrantScopeMovement {
    pub grant_id: [u8; 32],
    /// The EXACT audience scope this transition is about. A grant id can name two
    /// live scopes across an audience rotation, and the one that did not move
    /// must not be churned.
    pub audience_handle: [u8; 32],
    /// How this transition orders itself against retained artifacts.
    ///
    /// Ordinarily a consumer-Grant PUBLICATION generation: a reconstruction that
    /// observed generation `g` is obsolete with respect to this movement iff
    /// `g < publication`. **Strictly less than** — an artifact from THIS
    /// publication already reflects the transition and must survive it, which is
    /// exactly the "a demand arriving after publication is safe" case design
    /// §2A.2 names.
    ///
    /// **It must order absences as well as installations.** An earlier shape
    /// carried `superseded_through`, derived from `install_seq`, and treated an
    /// `Owner`-stamped artifact on a Grant slot as never-a-successor. That was
    /// the first repair, HELD at `7348529fb`: it is false, and Kyra's
    /// production-path probe demonstrated it — a delayed INSTALL notification
    /// destroyed the newer `Unserved` artifact a later REMOVAL had produced.
    /// `install_seq` totally orders installations, not transitions, and an
    /// absence has no installation identity to be ordered by.
    ///
    /// ```text
    /// install N     publish, [stall before notifying]
    /// (warm)        artifact reconstructed under N
    /// remove N      publish absence, notify -> clears it, re-queues
    /// (rebuild)     newer artifact: Unserved, `Owner`-stamped
    /// install N     [resumes] -> destroyed that newer artifact
    /// ```
    pub fence: GrantMovementFence,
}

impl GrantScopeMovement {
    /// Does this transition cover `key`'s scope?
    ///
    /// Exact on `(grant_id, audience_handle)`. Deliberately NOT narrowed by
    /// capability: the Grant source answers ANY capability under an installed
    /// `(grant_id, audience_handle)` with `Served(empty)` rather than `Unserved`,
    /// regardless of the grant's own capability scope — verified directly, a slot
    /// for a capability the grant does not cover reconstructs as
    /// `Served(0 providers)` and is stamped `Grant`. So the grant's movement
    /// genuinely affects every capability under its audience scope, and narrowing
    /// here would leave those slots holding a Grant-stamped artifact after
    /// removal and stuck `Unserved` after install — the exact two defects this
    /// edge exists to close, reintroduced for a subset. Narrowing becomes correct
    /// only once the source refuses uncovered capabilities structurally, which is
    /// not in this slice's scope.
    fn covers(&self, key: &SlotKey) -> bool {
        matches!(
            key.scope.scope(),
            CapabilityAudienceScope::Grant {
                grant_id,
                audience_handle,
            } if grant_id == &self.grant_id && audience_handle == &self.audience_handle
        )
    }
}

impl NodeOrgRoutingRegistry {
    pub(crate) fn new(
        source: Arc<dyn SlotSource>,
        work: Arc<RegistryWork>,
        metrics: Arc<RegistryMetrics>,
    ) -> Arc<Self> {
        Arc::new(Self {
            inner: parking_lot::Mutex::new(RegistryInner::default()),
            source,
            work,
            metrics,
            node_capacity_generation: AtomicU64::new(0),
            ref_release_generation: AtomicU64::new(0),
            publication_observer: PublicationObserver::default(),
        })
    }

    /// Test-only: observe the publication-order interleaving from a lock-free
    /// reader scheduled BETWEEN the two adjacent stores.
    ///
    /// This seam exists because the pool-first ordering is invisible to every
    /// final-state assertion — both stores complete before any of them runs
    /// (design §18.0). The hook fires under the registry lock, and the reader
    /// it drives takes none, which is exactly the observer the invariant is
    /// about.
    #[cfg(test)]
    pub(crate) fn observe_publication_for_test(
        &self,
        hook: Arc<dyn Fn(PublicationGap) + Send + Sync>,
    ) {
        *self.publication_observer.hook.lock() = Some(hook);
    }

    /// The node capacity generation (OLB-2B.3b §9). See the field.
    pub(crate) fn node_capacity_generation(&self) -> u64 {
        self.node_capacity_generation.load(Ordering::Acquire)
    }

    pub(crate) fn ref_release_generation(&self) -> u64 {
        self.ref_release_generation.load(Ordering::Acquire)
    }

    /// Mint a new clone family with its own private identity and handle budget.
    pub(crate) fn new_family(self: &Arc<Self>) -> Result<RoutingFamily, DemandRefused> {
        let id = {
            let mut inner = self.inner.lock();
            inner.allocate_id()
        };
        match id {
            Some(id) => Ok(RoutingFamily {
                registry: self.clone(),
                id: FamilyId(id),
            }),
            None => {
                self.metrics
                    .refused_id_space_exhausted
                    .fetch_add(1, Ordering::AcqRel);
                Err(DemandRefused::IdSpaceExhausted)
            }
        }
    }

    /// Register demand for `key` from `family`.
    ///
    /// FIRST demand for a key creates the slot and queues an exact FULL recapture
    /// of THAT slot — never a node-wide rebuild, and never contingent on whether
    /// older source deltas were already destructively drained. A later family
    /// demanding the same key SHARES the one node slot.
    ///
    /// A duplicate demand from the same family is a distinct handle: it shares the
    /// slot and counts toward the family bound, so re-demanding cannot bypass it.
    #[allow(dead_code)]
    fn demand(
        self: &Arc<Self>,
        family: FamilyId,
        key: SlotKey,
    ) -> Result<DemandHandle, DemandRefused> {
        let mut queued = false;
        let cells = {
            let mut inner = self.inner.lock();
            if inner.family_handles(family) >= MAX_HANDLES_PER_FAMILY {
                self.metrics
                    .refused_family_at_capacity
                    .fetch_add(1, Ordering::AcqRel);
                return Err(DemandRefused::FamilyAtCapacity);
            }
            let new_slot = !inner.slots.contains_key(&key);
            if new_slot && inner.slots.len() >= MAX_NODE_SLOTS {
                // Never evict a live retained slot to satisfy new demand.
                self.metrics
                    .refused_node_at_capacity
                    .fetch_add(1, Ordering::AcqRel);
                return Err(DemandRefused::NodeAtCapacity);
            }
            // Captured under the SAME lock acquisition that takes the reference,
            // so the cells can only be the ones belonging to the incarnation
            // this handle is now keeping alive — BOTH of them, as one unit
            // (design §1.1).
            //
            // The fresh path keeps the cells it INSERTED rather than looking the
            // slot back up, so there is no post-mutation lookup left to fail
            // (review 2026-07-29 §4). What remains is the pre-existing path,
            // where `new_slot` was false under this same lock hold and nothing
            // above removes a slot — and where, unlike before, the impossible
            // branch is reached having mutated NOTHING: no reserved slot with no
            // owner, no orphaned `pending` entry, no `queued` flag whose
            // `work.mark()` the early return would skip.
            let cells = if new_slot {
                // Allocate BEFORE mutating anything, so exhaustion retains nothing.
                let Some(incarnation) = inner.allocate_id() else {
                    self.metrics
                        .refused_id_space_exhausted
                        .fetch_add(1, Ordering::AcqRel);
                    return Err(DemandRefused::IdSpaceExhausted);
                };
                let slot = Slot::new(incarnation);
                let cells = slot.cells.clone();
                inner.slots.insert(key.clone(), slot);
                inner
                    .slots_by_capability
                    .entry(key.capability)
                    .or_default()
                    .insert(key.clone());
                inner.pending.insert(key.clone());
                queued = true;
                cells
            } else {
                let Some(slot) = inner.slots.get_mut(&key) else {
                    // Unreachable. Fail closed rather than fabricating detached
                    // cells that no install would ever write to — a handle
                    // reading a cell the registry does not own would be
                    // permanently, silently cold. `NodeAtCapacity` is the most
                    // conservative of the three refusals: it retains nothing and
                    // tells the caller to take the cold path.
                    return Err(DemandRefused::NodeAtCapacity);
                };
                slot.refs += 1;
                slot.cells.clone()
            };
            *inner
                .families
                .entry(family)
                .or_default()
                .entry(key.clone())
                .or_insert(0) += 1;
            cells
        };
        if queued {
            // Private discovery did not move, so this is the ONLY thing that will
            // wake the actor for it.
            self.work.mark();
        }
        Ok(DemandHandle {
            registry: self.clone(),
            family,
            key,
            cells,
        })
    }

    /// Acquire a family's COMPLETE demand set for one capability entry, or
    /// nothing at all (OLB-2B.3b §4.1).
    ///
    /// The all-or-none property is the whole point, and it is a CORRECTNESS
    /// property rather than a tidiness one. A capability's demand set is
    /// `Owner + every DISCOVER audience the family actually leased`, and each
    /// member contributes discovery provenance the final route is built from. A
    /// kept prefix would warm a capability whose Owner plane is retained and
    /// whose Grant plane is not — which does not read as a failure anywhere
    /// downstream. It reads as a route that legitimately found no granted
    /// provider, i.e. a silent authority narrowing presenting as a routing
    /// preference (§4.1).
    ///
    /// So every refusal is decided BEFORE anything is mutated, and after the
    /// last refusal the loop cannot fail:
    ///
    /// ```text
    /// sort + deduplicate
    /// → ONE registry acquisition
    ///     → family capacity for the WHOLE set
    ///     → node capacity for every NEW distinct slot
    ///     → reserve every required incarnation, contiguously, no wrap
    ///     → retain all references
    /// → mark work once
    /// ```
    ///
    /// Deduplicating first is not cosmetic either: a repeated key would reserve
    /// two handles against the family bound and count two new slots against the
    /// node bound, so a set that fits could be refused for capacity it never
    /// needed.
    ///
    /// This is NOT `demand` in a loop. That version consumes an identity per
    /// new slot before it discovers the refusal, retires those slots when the
    /// caller drops the prefix, and leaves the node's monotone id space
    /// permanently advanced by a call that retained nothing.
    #[allow(dead_code)] // consumer: `OrgRoutingState` (OLB-2B.3b), still dark.
    fn demand_set(
        self: &Arc<Self>,
        family: FamilyId,
        keys: Vec<SlotKey>,
    ) -> Result<DemandSet, DemandRefused> {
        let mut keys = keys;
        keys.sort();
        keys.dedup();

        let mut queued = false;
        let cells = {
            let mut inner = self.inner.lock();

            // Family capacity, for the whole set at once.
            if inner.family_handles(family) + keys.len() > MAX_HANDLES_PER_FAMILY {
                self.metrics
                    .refused_family_at_capacity
                    .fetch_add(1, Ordering::AcqRel);
                return Err(DemandRefused::FamilyAtCapacity);
            }

            // Node capacity, counting only slots this set would CREATE. Keys
            // already retained by some family cost no node capacity, exactly as
            // in the single-key path — a live slot is never evicted.
            let new_slots = keys
                .iter()
                .filter(|key| !inner.slots.contains_key(key))
                .count();
            if inner.slots.len() + new_slots > MAX_NODE_SLOTS {
                self.metrics
                    .refused_node_at_capacity
                    .fetch_add(1, Ordering::AcqRel);
                return Err(DemandRefused::NodeAtCapacity);
            }

            // Reserve every incarnation this set needs as ONE contiguous block,
            // checked, before the first mutation. `allocate_id` per slot inside
            // the loop would be equivalent only if the loop could not fail —
            // and the reason it cannot fail is precisely this reservation.
            let Some(reserved_through) = inner.next_id.checked_add(new_slots as u64) else {
                self.metrics
                    .refused_id_space_exhausted
                    .fetch_add(1, Ordering::AcqRel);
                return Err(DemandRefused::IdSpaceExhausted);
            };
            let mut next_incarnation = inner.next_id;
            inner.next_id = reserved_through;

            // Every refusal is above this line. Nothing below can fail, so
            // nothing below needs an unwind path.
            let mut cells = Vec::with_capacity(keys.len());
            for key in &keys {
                let slot_cells = match inner.slots.get_mut(key) {
                    Some(slot) => {
                        slot.refs += 1;
                        slot.cells.clone()
                    }
                    None => {
                        next_incarnation += 1;
                        let slot = Slot::new(next_incarnation);
                        let slot_cells = slot.cells.clone();
                        inner.slots.insert(key.clone(), slot);
                        inner
                            .slots_by_capability
                            .entry(key.capability)
                            .or_default()
                            .insert(key.clone());
                        inner.pending.insert(key.clone());
                        queued = true;
                        slot_cells
                    }
                };
                *inner
                    .families
                    .entry(family)
                    .or_default()
                    .entry(key.clone())
                    .or_insert(0) += 1;
                cells.push(slot_cells);
            }
            debug_assert_eq!(next_incarnation, reserved_through, "reservation is exact");
            cells
        };

        if queued {
            // ONE wake for the whole set: the actor's queue is the authoritative
            // `pending`, and it already holds every new identity.
            self.work.mark();
        }
        Ok(DemandSet {
            registry: self.clone(),
            family,
            keys: keys.clone(),
            cells,
            held: parking_lot::Mutex::new(keys),
        })
    }

    /// Release one handle. The LAST reference retires the slot; a re-demand then
    /// allocates a FRESH incarnation, so work in flight cannot resurrect it.
    #[allow(dead_code)]
    fn release(&self, family: FamilyId, key: &SlotKey) {
        let mut inner = self.inner.lock();
        match inner.release_one(family, key) {
            ReleaseOutcome::Retire => {
                self.retire_committed(&mut inner, key);
                self.note_ref_released();
            }
            ReleaseOutcome::Released => self.note_ref_released(),
            ReleaseOutcome::NotHeld => {}
        }
    }

    /// Release EVERY key of a set in ONE registry transaction.
    ///
    /// One acquisition, not one per key: a set is released as a unit, so no
    /// observer can see a partially-released set and the capacity generation
    /// moves once for the whole retirement rather than interleaving with another
    /// family's demand.
    fn release_keys(&self, family: FamilyId, keys: &[SlotKey]) {
        let mut inner = self.inner.lock();
        let mut released = false;
        for key in keys {
            match inner.release_one(family, key) {
                ReleaseOutcome::Retire => {
                    self.retire_committed(&mut inner, key);
                    released = true;
                }
                ReleaseOutcome::Released => released = true,
                ReleaseOutcome::NotHeld => {}
            }
        }
        if released {
            self.note_ref_released();
        }
    }

    /// Publish that a reference was given back, so a refused REPLACEMENT knows
    /// its projection may have changed. Advances on every release, including one
    /// that leaves the slot alive with other owners.
    fn note_ref_released(&self) {
        self.ref_release_generation.fetch_add(1, Ordering::AcqRel);
    }

    /// Commit the retirement of a slot whose last reference just went away.
    ///
    /// **Clears the publication cell before dropping the slot.** The cell is an
    /// `Arc` that outstanding readers may still hold — a superseded `DemandSet`,
    /// or any holder of an `Arc<CapabilityRouteHandle>` the index no longer
    /// names — and removing the slot from `slots` does NOT reach them. Leaving
    /// the artifact in place would let a reader that owns no registry reference
    /// go on reading a dead incarnation's facts indefinitely, which is retention
    /// without ownership: exactly the property `DemandSet` exists to make
    /// impossible.
    ///
    /// Clearing it here is what makes "a retired slot's cell reads empty" TRUE
    /// rather than merely documented. It is safe against a concurrent
    /// reconstruction because the apply path re-looks the slot up under this
    /// same lock and refuses twice over: a removed slot is skipped outright, and
    /// a re-demanded one fails the `slot.incarnation == slot_incarnation` fence.
    /// A re-demand also mints a FRESH cell, so the cleared one is unreachable
    /// from the registry and nothing can ever republish into it.
    fn retire_committed(&self, inner: &mut RegistryInner, key: &SlotKey) {
        if let Some(slot) = inner.slots.get(key) {
            // Before the removal, so no window exists in which the slot is gone
            // from the index while its artifacts are still readable. BOTH
            // cells, pool first (§1.1): a retired incarnation's routing pool is
            // exactly as dead as its facts.
            slot.cells.take_facts();
        }
        inner.slots.remove(key);
        inner.pending.remove(key);
        if let Some(bucket) = inner.slots_by_capability.get_mut(&key.capability) {
            bucket.remove(key);
            if bucket.is_empty() {
                inner.slots_by_capability.remove(&key.capability);
            }
        }
        self.metrics.slots_retired.fetch_add(1, Ordering::AcqRel);
        // Node capacity genuinely moved. Published while the registry lock
        // still covers the removal, so an observer that reads the new
        // generation cannot then find the slot still there.
        self.node_capacity_generation.fetch_add(1, Ordering::AcqRel);
    }

    /// Atomically REPLACE one family's demand set with another (OLB-2B.3b §4.3).
    ///
    /// Not "acquire the new set, then drop the old one". That charges the family
    /// and the node at the TRANSIENT GROSS peak — every replacement handle while
    /// every superseded handle is still charged — so a replacement that plainly
    /// fits its final footprint is refused at the bound. A family holding 64 of
    /// 64 whose width-2 capability loses an audience projects to
    /// `64 - 2 + 1 = 63` and MUST succeed; charged gross it asks for 66.
    ///
    /// Nor "release the old set, then acquire the new one". That answers the
    /// bound correctly and breaks the no-effect property instead: a refusal
    /// would have already destroyed retained authority, and the re-acquisition
    /// that was supposed to restore it can fail on its own.
    ///
    /// So: ONE transaction, charged on the PROJECTED FINAL footprint.
    ///
    /// ```text
    /// common     = old ∩ new    reference kept EXACTLY as it is — no churn
    /// old_only   = old \ new    released; credits node capacity only if LAST
    /// new_only   = new \ old    charged; costs a node slot only if absent
    ///
    /// family:  held - |old_only| + |new_only|   <= MAX_HANDLES_PER_FAMILY
    /// node:    slots - credited  + created      <= MAX_NODE_SLOTS
    /// ```
    ///
    /// The node credit is deliberately conditional. An `old_only` slot another
    /// family still demands does NOT retire, so it frees nothing and crediting
    /// it would admit a 257th slot. The shared-old and last-reference cases are
    /// separately witnessed for exactly that reason.
    ///
    /// Every refusal is decided before the first mutation, so a refused
    /// replacement leaves handles, slots, refs, pending work, identities and
    /// generations exactly as they were — and, because the caller keeps its
    /// `DemandSet`, leaves the superseded entry fully live and owned.
    fn replace_demand_set(
        self: &Arc<Self>,
        family: FamilyId,
        old: &DemandSet,
        new_keys: Vec<SlotKey>,
    ) -> Result<DemandSet, ReplaceRefused> {
        let mut new_keys = new_keys;
        new_keys.sort();
        new_keys.dedup();

        // Lock order: the set's release responsibility, then the registry. The
        // set's own `Drop` takes them in the same order, so the two cannot
        // deadlock against each other.
        let mut held = old.held.lock();

        // A set that has already transferred owns nothing. Replacing it would
        // compute `old_only = ∅` and charge the successor's whole footprint
        // gross — the HOLD-2 defect, arriving through a stale basis instead of a
        // stale accounting rule. Refused before anything is read or written.
        if *held != old.keys {
            return Err(ReplaceRefused::Superseded);
        }

        let mut queued = false;
        let mut released = false;
        let cells = {
            let mut inner = self.inner.lock();

            let old_only: Vec<SlotKey> = held
                .iter()
                .filter(|key| new_keys.binary_search(key).is_err())
                .cloned()
                .collect();
            let new_only: Vec<SlotKey> = new_keys
                .iter()
                .filter(|key| held.binary_search(key).is_err())
                .cloned()
                .collect();

            // FAMILY, on the projected final count.
            let projected =
                inner.family_handles(family).saturating_sub(old_only.len()) + new_only.len();
            if projected > MAX_HANDLES_PER_FAMILY {
                self.metrics
                    .refused_family_at_capacity
                    .fetch_add(1, Ordering::AcqRel);
                return Err(ReplaceRefused::Demand(DemandRefused::FamilyAtCapacity));
            }

            // NODE, on the projected final retained set. A slot is credited only
            // when THIS reference is its last.
            let credited = old_only
                .iter()
                .filter(|key| inner.slots.get(key).is_some_and(|slot| slot.refs == 1))
                .count();
            let created = new_only
                .iter()
                .filter(|key| !inner.slots.contains_key(key))
                .count();
            let projected_slots = inner.slots.len().saturating_sub(credited) + created;
            if projected_slots > MAX_NODE_SLOTS {
                self.metrics
                    .refused_node_at_capacity
                    .fetch_add(1, Ordering::AcqRel);
                return Err(ReplaceRefused::Demand(DemandRefused::NodeAtCapacity));
            }

            // IDENTITIES for genuinely new slots, reserved as one checked block.
            let Some(reserved_through) = inner.next_id.checked_add(created as u64) else {
                self.metrics
                    .refused_id_space_exhausted
                    .fetch_add(1, Ordering::AcqRel);
                return Err(ReplaceRefused::Demand(DemandRefused::IdSpaceExhausted));
            };
            let mut next_incarnation = inner.next_id;
            inner.next_id = reserved_through;

            // Every refusal is above this line. Nothing below can fail.

            // Released FIRST, so the retained-slot count never rises above the
            // projected final figure even inside the transaction.
            for key in &old_only {
                // `NotHeld` is unreachable: `held` was verified to equal this
                // set's named keys above, and a set names only keys its own
                // family holds. Retiring on it anyway would be the cross-family
                // corruption in a different shape, so it is simply not a retire.
                match inner.release_one(family, key) {
                    ReleaseOutcome::Retire => {
                        self.retire_committed(&mut inner, key);
                        released = true;
                    }
                    ReleaseOutcome::Released => released = true,
                    ReleaseOutcome::NotHeld => {}
                }
            }
            for key in &new_only {
                match inner.slots.get_mut(key) {
                    Some(slot) => slot.refs += 1,
                    None => {
                        next_incarnation += 1;
                        inner.slots.insert(key.clone(), Slot::new(next_incarnation));
                        inner
                            .slots_by_capability
                            .entry(key.capability)
                            .or_default()
                            .insert(key.clone());
                        inner.pending.insert(key.clone());
                        queued = true;
                    }
                }
                *inner
                    .families
                    .entry(family)
                    .or_default()
                    .entry(key.clone())
                    .or_insert(0) += 1;
            }
            debug_assert_eq!(next_incarnation, reserved_through, "reservation is exact");

            // Cells for the WHOLE new set — both planes per key. A `common` key
            // yields the very cells it already had, because its reference was
            // never touched.
            new_keys
                .iter()
                .map(|key| {
                    // `expect_used` lint guard: a `common` key kept its
                    // reference untouched and a `new_only` key was inserted or
                    // bumped just above, so the lookup cannot miss; suppress
                    // locally rather than `filter_map`, which would silently
                    // misalign the `keys[i]` <-> `cells[i]` pairing.
                    #[allow(clippy::expect_used)]
                    let slot = inner
                        .slots
                        .get(key)
                        .expect("every key of the new set is retained above");
                    slot.cells.clone()
                })
                .collect::<Vec<_>>()
        };

        // The TRANSFER, and the only place it happens: the superseded set now
        // owes nothing, because everything it owed is either still held (common)
        // or has just been released (old_only). No flag decides this — the keys
        // themselves moved, so exactly one set owes a release at every instant.
        held.clear();
        drop(held);

        if released {
            self.note_ref_released();
        }
        if queued {
            self.work.mark();
        }
        Ok(DemandSet {
            registry: self.clone(),
            family,
            keys: new_keys.clone(),
            cells,
            held: parking_lot::Mutex::new(new_keys),
        })
    }

    /// The RAW retained base facts for `key` — retention only, with NO
    /// revalidation whatsoever. `None` is the deterministic UNRETAINED outcome.
    ///
    /// **Not a read seam. Production callers want `MeshNode::org_routing_base_facts`**
    /// (`mesh.rs`), which is the validated twin: it re-checks the live authority
    /// epoch, the revocation floor generation, the poison bit, terminal
    /// exhaustion, `Unserved`, wall-clock expiry and the incarnation fence, and
    /// retires the artifact when any of them says no. NONE of that lives here.
    ///
    /// Named for what it is rather than left as the more discoverable of two
    /// similarly-named accessors (review-pass-3 §7): the code that holds a
    /// registry handle would otherwise reach for this one by default, and every
    /// check above is a security property the warmed-call consumer must not be
    /// able to skip by accident. Crate-private, and its remaining callers are
    /// witnesses asserting on RETENTION specifically — several of them precisely
    /// to prove that the seam fences something the registry still holds.
    pub(crate) fn base_facts_unvalidated(&self, key: &SlotKey) -> Option<Arc<SlotBaseFacts>> {
        self.inner.lock().slots.get(key)?.cells.facts.load_full()
    }

    /// Authority moved to `live`: drop every fact stamped with an OLDER
    /// authority, re-queue those slots and wake.
    ///
    /// Called by the node when revocation authority changes. Authority movement
    /// need not touch scoped state at all — a store swap that retracts nothing
    /// advances no scoped revision and publishes no scoped wake — so without this
    /// the actor would never learn that everything it holds was built under an
    /// authority that no longer applies (Kyra OLB-2B-E3c).
    ///
    /// CONDITIONAL on the stamped epoch rather than unconditional. The publisher
    /// releases the authority gate before calling this, so a reconciliation under
    /// the NEW authority can install valid facts in between; wiping everything
    /// would delete them and re-queue work that was already done, turning a
    /// returned `Current` into immediately-owed work. Epochs are monotone, so
    /// `< live` is exact and cannot match a successor.
    pub(crate) fn invalidate_authority_older_than(&self, live: u64) {
        let owed = {
            let mut inner = self.inner.lock();
            let stale: Vec<SlotKey> = inner
                .slots
                .iter()
                .filter(|(_, slot)| {
                    slot.cells
                        .facts
                        .load()
                        .as_ref()
                        .is_some_and(|facts| facts.epoch.authority < live)
                })
                .map(|(key, _)| key.clone())
                .collect();
            let mut invalidated = 0;
            for key in stale {
                if inner.invalidate(&key) {
                    invalidated += 1;
                }
                inner.pending.insert(key);
            }
            self.metrics
                .facts_invalidated
                .fetch_add(invalidated, Ordering::AcqRel);
            !inner.pending.is_empty()
        };
        if owed {
            self.work.mark();
        }
    }

    /// One slot proved unusable at READ time: drop THOSE facts, re-queue, wake.
    ///
    /// Conditional on the exact artifact the reader observed. A reader can be
    /// arbitrarily delayed between loading facts and deciding they are stale, and
    /// reconciliation may have installed a current replacement in the meantime —
    /// an unconditional removal would delete that replacement and re-queue work
    /// that was already done (Kyra OLB-2B-E3c). `Arc::ptr_eq` is exact here
    /// because installation always allocates a fresh `Arc`.
    pub(crate) fn invalidate_if_stale(&self, key: &SlotKey, observed: &Arc<SlotBaseFacts>) {
        let owed = {
            let mut inner = self.inner.lock();
            let Some(slot) = inner.slots.get_mut(key) else {
                return;
            };
            let still_observed = slot
                .cells
                .facts
                .load()
                .as_ref()
                .is_some_and(|live| Arc::ptr_eq(live, observed));
            if !still_observed {
                // Someone installed a newer artifact. Leave it alone — and its
                // pool with it (§1.1): a delayed invalidator deletes nothing
                // derived from an artifact it did not observe.
                return;
            }
            slot.cells.take_facts();
            self.metrics
                .facts_invalidated
                .fetch_add(1, Ordering::AcqRel);
            inner.pending.insert(key.clone());
            true
        };
        if owed {
            self.work.mark();
        }
    }

    /// The node's session/direct-state projection moved to `live`: drop every
    /// pool computed under an OLDER session view, re-queue exactly those
    /// slots, wake the actor, and PRESERVE every facts artifact (OLB-2B.3c
    /// step 2; design §1.1's pool-only half).
    ///
    /// Returns the number of pools retired.
    ///
    /// **The asymmetry is the whole point.** Session projection moving says
    /// nothing about discovery: the providers, their proven owners, their
    /// expiries and the authority that made them query-visible are all exactly
    /// as current as they were. Clearing the facts too would re-queue a full
    /// source reconstruction for a peer connecting — and would do it on every
    /// peer transition, which is the loudest churn source a node has.
    ///
    /// **Conditional on the generation the pool was COMPUTED under, and
    /// strictly less than.** A notification can be delayed arbitrarily between
    /// its publication and this call, so an obsolete transition can arrive
    /// after a newer projection has already been published and a newer pool
    /// built from it. Two survivors, and both are reachable:
    ///
    /// - a pool from a LATER session view — the ordinary delayed-invalidator
    ///   case;
    /// - a pool from THIS view — an actor pass that captured the projection
    ///   after it was published and before this notification ran already
    ///   reflects the transition, exactly as `<` versus `<=` decides for the
    ///   consumer-Grant fence.
    ///
    /// It cannot delete a newer pool derived from NEWER facts either, and not
    /// by luck: generations are monotone, so a build that observed the
    /// projection after this transition necessarily stamped `>= live`.
    ///
    /// A retained slot whose pool is ABSENT is left entirely alone — no clear
    /// and no re-queue. The build that will publish its pool is either already
    /// owed (its identity is in `pending`) or already in flight, and an
    /// in-flight one revalidates the generation before it publishes. Waking on
    /// absence instead would make every peer transition re-queue every retained
    /// slot on the node, which is the "invalidate everything on any movement"
    /// breadth failure W-W3 exists to refuse.
    ///
    /// The caller must NOT hold the session publication gate: this takes the
    /// registry lock, and the documented order is
    /// `peer-transition/session publication → registry lock`. Publishing the
    /// new projection, advancing the generation and RELEASING the gate all
    /// happen before this runs, on the same discipline as the consumer-Grant
    /// edge (design §2A.2, item 10).
    pub(crate) fn invalidate_session_older_than(&self, live: u64) -> u64 {
        let (retired, owed) = {
            let mut inner = self.inner.lock();
            let superseded: Vec<SlotKey> = inner
                .slots
                .iter()
                .filter(|(_, slot)| {
                    slot.cells
                        .unsensed
                        .load()
                        .as_ref()
                        .is_some_and(|pool| pool.session_generation < live)
                })
                .map(|(key, _)| key.clone())
                .collect();
            let mut retired = 0u64;
            for key in superseded {
                if let Some(slot) = inner.slots.get(&key) {
                    if slot.cells.take_pool().is_some() {
                        retired += 1;
                    }
                }
                inner.pending.insert(key);
            }
            self.metrics
                .pools_invalidated
                .fetch_add(retired, Ordering::AcqRel);
            (retired, !inner.pending.is_empty())
        };
        if owed {
            self.work.mark();
        }
        retired
    }

    /// A consumer Grant moved: retire every retained artifact of that exact
    /// audience scope which the transition SUPERSEDES, re-queue exactly those
    /// slots, and wake the actor (items 10/11).
    ///
    /// Returns the number of artifacts retired.
    ///
    /// **Conditional, and decided under the registry lock.** Each slot is judged
    /// against the artifact it actually holds, at the mutation boundary — not
    /// against a snapshot read before the lock. A pre-lock "is this still
    /// current?" check would be insufficient on its own: a newer publication can
    /// land between the check and the clear, and the decision to preserve a
    /// successor has to still be true when the clear happens.
    ///
    /// A slot that is retained but holds NO artifact is still re-queued — that is
    /// the whole install direction, and it is harmless coalescing besides.
    /// Everything else is judged by the publication generation the artifact was
    /// reconstructed under.
    ///
    /// Three kinds of survivor, and the wording matters because two earlier
    /// versions of this comment erased one of them each:
    ///
    /// - **this publication's own artifact** — equal generation. A demand that
    ///   arrived after the publication and before its notification produces one,
    ///   and it already reflects the transition;
    /// - a LATER installation's artifact;
    /// - an equal-or-later ABSENCE artifact. Treating absence as
    ///   never-a-successor is the defect Kyra's probe found at `7348529fb`.
    ///
    /// **Not keyed on `ScopedDiscoveryState::revision`** (item 11, design §2A.2).
    /// A consumer-Grant transition mutates the grant registry, not the scoped
    /// store, so the scoped revision does not move and cannot be the trigger.
    ///
    /// The caller must NOT hold the consumer-Grant gate: the commit pin takes
    /// that gate and then this lock, so acquiring this beneath it would invert
    /// the frozen order. Item 10 states the same thing as an ordering rule.
    pub(crate) fn invalidate_grant_scope(&self, movement: &GrantScopeMovement) -> u64 {
        let (retired, owed) = {
            let mut inner = self.inner.lock();
            let affected: Vec<SlotKey> = inner
                .slots
                .keys()
                .filter(|key| movement.covers(key))
                .cloned()
                .collect();
            let mut retired = 0u64;
            for key in affected {
                let Some(cells) = inner.slots.get(&key).map(|slot| slot.cells.clone()) else {
                    continue;
                };
                // Ordered by the PUBLICATION generation the reconstruction
                // observed, not by the authority it stamped — that is what makes
                // the fence total over installations AND absences. A slot with no
                // artifact has nothing to preserve and still owes work, so it is
                // re-queued: harmless coalescing, since `pending` is a set and a
                // cold slot needs rebuilding regardless.
                //
                // STRICTLY less than. An artifact from THIS publication already
                // reflects the transition and must survive it — that is the
                // "a demand arriving after publication is safe" case of §2A.2,
                // and `<=` would have the notification destroy its own
                // publication's current artifact.
                //             artifact:  Publication(a)      TerminalAbsence
                //  movement:
                //  Publication(p)         clear iff a < p     preserve
                //  Terminal               clear               preserve
                //
                // The `Terminal` row is NOT "clear everything". Terminal
                // withdrawal must still preserve the absence artifact its OWN
                // publication produced — the same equality property W-W9/W-W10
                // pin for ordinary publications, and it does not come for free
                // just because nothing can supersede a terminal state.
                //
                // `Publication(_)` under a `Terminal` movement always clears: at
                // exhaustion the last live identity may equal the artifact's, so
                // an identity comparison could not distinguish "before the
                // withdrawal" from "after" — which is exactly the alias this
                // arm exists to avoid.
                //
                // FOUR arms for four cells, and the two `TerminalAbsence` ones
                // are written out separately even though they agree. Collapsed
                // to `(TerminalAbsence, _)` they could not be mutated apart, and
                // that is not hypothetical: the whole gate stayed green with
                // ordinary movement clearing terminal absence, because the only
                // witness in that column tested the `Terminal` row (Kyra, review
                // of `010c718ea`). A cell that cannot be mutated alone cannot be
                // witnessed alone.
                let superseded = match cells.facts.load().as_ref() {
                    None => true,
                    Some(facts) => match (facts.grant_fence, movement.fence) {
                        // Terminal absence survives its own ORDINARY publication.
                        // The removal that SPENDS the space is an ordinary
                        // transition — it reserves the last live identity — yet
                        // the absence it causes reconstructs terminal. W-W15.
                        (
                            GrantArtifactFence::TerminalAbsence,
                            GrantMovementFence::Publication(_),
                        ) => false,
                        // And survives a terminal one. W-W14.
                        (GrantArtifactFence::TerminalAbsence, GrantMovementFence::Terminal) => {
                            false
                        }
                        (GrantArtifactFence::Publication(_), GrantMovementFence::Terminal) => true,
                        (
                            GrantArtifactFence::Publication(artifact),
                            GrantMovementFence::Publication(publication),
                        ) => artifact < publication,
                    },
                };
                if !superseded {
                    // A LATER installation already published here. Leave it
                    // entirely alone — no clear, and no re-queue either.
                    continue;
                }
                if cells.take_facts().is_some() {
                    retired += 1;
                }
                inner.pending.insert(key);
            }
            self.metrics
                .facts_invalidated
                .fetch_add(retired, Ordering::AcqRel);
            (retired, !inner.pending.is_empty())
        };
        if owed {
            self.work.mark();
        }
        retired
    }

    /// The authority epoch space is terminally exhausted: retire every retained
    /// fact and every queued identity, and do NOT wake the actor (E3c blockers
    /// §2). Called exactly once, at the `NewlyExhausted` transition.
    ///
    /// Two ways this deliberately differs from
    /// [`Self::invalidate_authority_older_than`]:
    ///
    /// - it is UNCONDITIONAL on the stamped epoch. Facts stamped
    ///   `authority == u64::MAX` are exactly the ones a strictly-older
    ///   comparison structurally spares — yet their stamp is the identity that
    ///   just became unable to prove currentness, so leaving them retained
    ///   keeps them readable for as long as no reader happens by;
    /// - NOTHING is re-queued and no work is marked. The source refuses every
    ///   commit pin from now on, so a rebuilt fact could never be installed —
    ///   a re-queue is a promise of work that cannot complete, and the wake
    ///   that re-arms the actor would have it spin on `Superseded` until
    ///   shutdown.
    pub(crate) fn retire_terminal(&self) {
        let mut inner = self.inner.lock();
        let mut invalidated = 0u64;
        for slot in inner.slots.values_mut() {
            if slot.cells.take_facts().is_some() {
                invalidated += 1;
            }
        }
        inner.pending.clear();
        inner.recapture_open = false;
        self.metrics
            .facts_invalidated
            .fetch_add(invalidated, Ordering::AcqRel);
    }

    /// The EARLIEST deadline any retained artifact carries, or `None` if none
    /// carries one (scope item 12).
    ///
    /// This is what the actor arms its expiry wake to. `u64::MAX` means "no
    /// deadline" and is deliberately excluded rather than returned — an
    /// artifact with no deadline must not produce a wake at the end of time,
    /// and `Option` says that where a sentinel would invite arithmetic on it.
    pub(crate) fn next_artifact_deadline(&self) -> Option<u64> {
        self.inner
            .lock()
            .slots
            .values()
            .filter_map(|slot| slot.cells.facts.load().as_ref().map(|f| f.earliest_expiry))
            .filter(|deadline| *deadline != u64::MAX)
            .min()
    }

    /// Retire every retained artifact whose deadline `now_secs` has reached, and
    /// RE-QUEUE its slot so the actor rebuilds it (scope item 12).
    ///
    /// The read seam already refuses an expired artifact, so this is not what
    /// makes expiry safe — it is what makes it PROMPT, and what makes the
    /// registry's retained set agree with what a reader would be told. Without
    /// it, an artifact whose authority expired sits retained until some reader
    /// happens by; with zero provider rows that reader is the only thing that
    /// could ever retire it, because nothing else in the artifact carries a
    /// deadline at all.
    ///
    /// Re-queuing is what closes the loop: the rebuilt reconstruction finds the
    /// authority gone and installs `Unserved`, whose deadline is `u64::MAX`, so
    /// the next arm finds nothing and the actor parks. That is also why this
    /// cannot spin — a retired deadline is not reinstalled.
    pub(crate) fn retire_expired(&self, now_secs: u64) -> u64 {
        let mut inner = self.inner.lock();
        let expired: Vec<SlotKey> = inner
            .slots
            .iter()
            .filter(|(_, slot)| {
                slot.cells
                    .facts
                    .load()
                    .as_ref()
                    .is_some_and(|f| now_secs >= f.earliest_expiry)
            })
            .map(|(key, _)| key.clone())
            .collect();
        let mut retired = 0u64;
        for key in expired {
            if let Some(slot) = inner.slots.get_mut(&key) {
                if slot.cells.take_facts().is_some() {
                    retired += 1;
                }
            }
            inner.pending.insert(key);
        }
        self.metrics
            .facts_invalidated
            .fetch_add(retired, Ordering::AcqRel);
        retired
    }

    /// Test-only: re-queue `key` WITHOUT invalidating either plane, so a
    /// witness can drive a fresh quantum over a slot that still holds its
    /// published artifacts. That is the state the publication-order witness
    /// needs: a SUPERSEDED pool must be present when the next install starts,
    /// or every observation gap is trivially pool-free.
    #[cfg(test)]
    pub(crate) fn requeue_for_test(&self, key: &SlotKey) {
        self.inner.lock().pending.insert(key.clone());
    }

    /// Test-only: drop ONLY `key`'s pool, leaving its facts — the state a slot
    /// is in between a refused pool publication and its rebuild.
    #[cfg(test)]
    pub(crate) fn take_pool_for_test(&self, key: &SlotKey) {
        let inner = self.inner.lock();
        if let Some(slot) = inner.slots.get(key) {
            slot.cells.take_pool();
        }
    }

    /// Test-only: drop `key`'s facts and re-queue it, so a witness can drive
    /// another quantum over the same slot.
    #[cfg(test)]
    pub(crate) fn invalidate_for_test(&self, key: &SlotKey) {
        let mut inner = self.inner.lock();
        inner.invalidate(key);
        inner.pending.insert(key.clone());
    }

    /// Test-only: give back ONE of `family`'s references to `key`, exactly as an
    /// internal release would.
    ///
    /// Exists so the ownership invariant can be witnessed DIRECTLY — that a
    /// family which holds no such reference cannot decrement the global one —
    /// rather than inferred from a cross-family transfer the crate-private
    /// surface no longer lets anyone express.
    #[cfg(test)]
    pub(crate) fn release_one_for_test(&self, family: &RoutingFamily, key: &SlotKey) {
        let mut inner = self.inner.lock();
        match inner.release_one(family.id, key) {
            ReleaseOutcome::Retire => {
                self.retire_committed(&mut inner, key);
                drop(inner);
                self.note_ref_released();
            }
            ReleaseOutcome::Released => {
                drop(inner);
                self.note_ref_released();
            }
            ReleaseOutcome::NotHeld => {}
        }
    }

    /// Test-only: install `facts` for `key`, creating the slot if needed.
    ///
    /// Lets a witness reproduce a quantum that raced a deadline — facts valid at
    /// capture, expired by the time they are read — without having to win that
    /// race against the wall clock.
    #[cfg(test)]
    pub(crate) fn install_facts_for_test(&self, key: SlotKey, facts: Arc<SlotBaseFacts>) {
        let mut inner = self.inner.lock();
        let slot = inner.slots.entry(key).or_insert_with(|| Slot::new(1));
        // The production ordering, pool first — so a witness that wants a pool
        // beside these facts injects it AFTER this call, exactly as the actor
        // build cycle publishes after installing (§18 step 2).
        slot.cells.install_facts(facts);
    }

    /// Test-only: publish a routing pool for `key`, for two-cell lifecycle
    /// witnesses. The production publisher is the actor build cycle (§18 step
    /// 2), which is why this seam exists — the lifecycle lands, and must be
    /// witnessed, before the builder does.
    #[cfg(test)]
    pub(crate) fn install_unsensed_pool_for_test(
        &self,
        key: &SlotKey,
        pool: Arc<ScopedUnsensedRoutePool>,
    ) {
        let inner = self.inner.lock();
        if let Some(slot) = inner.slots.get(key) {
            slot.cells.unsensed.store(Some(pool));
        }
    }

    /// The RAW retained routing pool for `key` — the second cell's twin of
    /// [`Self::base_facts_unvalidated`], with the same warning.
    ///
    /// **Not a read seam.** Production callers want
    /// `MeshNode::org_routing_unsensed_pool` (`mesh.rs`), which validates the
    /// facts beneath it, proves the pool derives from exactly those facts, and
    /// refuses a pool computed under a superseded session view. None of that
    /// lives here, and the name says so for the same reason its twin's does.
    pub(crate) fn unsensed_pool_unvalidated(
        &self,
        key: &SlotKey,
    ) -> Option<Arc<ScopedUnsensedRoutePool>> {
        self.inner.lock().slots.get(key)?.cells.unsensed.load_full()
    }

    pub(crate) fn retained_slots(&self) -> usize {
        self.inner.lock().slots.len()
    }

    pub(crate) fn pending_slots(&self) -> usize {
        self.inner.lock().pending.len()
    }

    /// Test-only: the high-water mark of the monotone identity allocator.
    ///
    /// The no-effect assertion for a refused demand set has to be TOTAL —
    /// {slots, pending, handles, identities} — and the identity component is
    /// the one a naive `demand`-in-a-loop implementation silently fails while
    /// every other component looks clean, because the prefix it retained is
    /// released again on the error path (OLB-2B.3b, W-A5; the same lesson as
    /// step 1's `300e80f6c`).
    #[cfg(test)]
    pub(crate) fn allocated_ids_for_test(&self) -> u64 {
        self.inner.lock().next_id
    }

    /// Test-only: drive the identity allocator to exhaustion, so a witness can
    /// reach the TERMINAL refusal without minting 2^64 slots.
    #[cfg(test)]
    pub(crate) fn exhaust_ids_for_test(&self) {
        self.inner.lock().next_id = u64::MAX;
    }
}

impl NodeOrgRoutingRegistry {
    /// Re-arm the actor for a refusal that was genuine source MOVEMENT, and only
    /// then (E3c blockers §3, review-pass-3 §2).
    ///
    /// The suppression is not an optimisation. A `Fenced` or `Terminal` source
    /// refuses without anything having moved and will refuse the next pass
    /// identically, so marking against it is a self-sustaining wake — the exact
    /// livelock the exhaustion fences exist to prevent. Movement, by contrast,
    /// is finite and the pass that follows it can succeed.
    fn mark_if_movement(&self) {
        if self.source.liveness().may_self_wake() {
            self.work.mark();
        }
    }
}

impl DirtyApply for NodeOrgRoutingRegistry {
    /// A fresh incarnation claims work authority and invalidates EVERYTHING.
    ///
    /// Nothing a dead incarnation built is trustworthy, so rather than leaving
    /// stale facts readable until the first recapture lands, this drops them and
    /// queues every retained slot. `SlotBaseFacts::actor_incarnation` therefore
    /// records which live actor built a fact rather than merely enabling a check
    /// for one that no longer is.
    fn activate_incarnation(&self, incarnation: u64) {
        let owed = {
            let mut inner = self.inner.lock();
            inner.live_actor = Some(incarnation);
            inner.recapture_open = false;
            let invalidated = inner.invalidate_and_queue_all();
            self.metrics
                .facts_invalidated
                .fetch_add(invalidated, Ordering::AcqRel);
            !inner.pending.is_empty()
        };
        if owed {
            self.work.mark();
        }
    }

    /// This incarnation is over. Revoking authority here — from the actor's own
    /// fence — is what makes "stale actor" a state rather than a comparison
    /// against a high-water mark.
    fn deactivate_incarnation(&self, incarnation: u64) {
        let mut inner = self.inner.lock();
        if inner.live_actor == Some(incarnation) {
            inner.live_actor = None;
        }
    }

    fn next_deadline(&self) -> Option<u64> {
        self.next_artifact_deadline()
    }

    fn retire_expired(&self, now_secs: u64) -> u64 {
        NodeOrgRoutingRegistry::retire_expired(self, now_secs)
    }

    /// One bounded reconciliation quantum.
    ///
    /// The phases exist for the lock discipline. The registry lock is held ONLY to
    /// select/invalidate and later to install. The source is touched twice, both
    /// times briefly: once to snapshot the material for this quantum, and once to
    /// pin currentness across the installation. Decoding, sorting and projection
    /// happen between those, holding NOTHING.
    fn apply(&self, incarnation: u64, request: ApplyRequest) -> ApplyOutcome {
        // --- phase 1: invalidate + select under the lock, bounded to one quantum ---
        let selected: Vec<(SlotKey, u64)> = {
            let mut inner = self.inner.lock();

            if inner.live_actor != Some(incarnation) {
                // Stale or absent authority. Consume NOTHING: taking pending
                // identities here and then failing at installation is exactly how
                // authoritative work gets lost while the caller is told it
                // succeeded (Kyra OLB-2B-E3b).
                self.metrics
                    .stale_actor_rejections
                    .fetch_add(1, Ordering::AcqRel);
                drop(inner);
                // Re-arm so the LIVE incarnation still sees the work that woke us.
                self.work.mark();
                return ApplyOutcome::Superseded;
            }

            // Slots this request names, invalidated under THIS lock — before any
            // rebuild starts and including slots beyond the quantum. Leaving stale
            // facts readable while their capability is known to have moved is a
            // silent-staleness hazard that no later install can undo (Kyra
            // OLB-2B-E3b): `Caps` deliberately leaves global health alone, so
            // per-slot invalidation is the ONLY thing fencing those routes.
            let mut invalidated = 0u64;
            let mut named: BTreeSet<SlotKey> = BTreeSet::new();
            match &request.batch.dirty {
                DirtyCapabilities::RebuildAll => {
                    if !inner.recapture_open {
                        inner.recapture_open = true;
                        invalidated += inner.invalidate_and_queue_all();
                    }
                    // An already-open epoch keeps its remaining set in `pending`;
                    // re-expanding would re-select the first quantum forever. A
                    // source generation that moves mid-epoch restarts it from
                    // `settle`, where the COMMITTED generation is known.
                    named = inner.pending.clone();
                }
                DirtyCapabilities::Caps(caps) => {
                    let affected: Vec<SlotKey> = caps
                        .iter()
                        .filter_map(|capability| inner.slots_by_capability.get(capability))
                        .flat_map(|bucket| bucket.iter().cloned())
                        .collect();
                    for key in affected {
                        if inner.invalidate(&key) {
                            invalidated += 1;
                        }
                        inner.pending.insert(key.clone());
                        named.insert(key);
                    }
                }
                DirtyCapabilities::Clean => {}
            }
            if request.registry_work {
                named.extend(inner.pending.iter().cloned());
            }
            self.metrics
                .facts_invalidated
                .fetch_add(invalidated, Ordering::AcqRel);

            // ROTATED, not key-ordered (review-pass-3 §9). Taking the first
            // `APPLY_QUANTUM` in `(scope, capability)` order every pass starves a
            // subset under sustained churn: if each pass re-dirties at least a
            // quantum's worth of slots whose keys sort BELOW a victim's, the
            // victim is re-outsorted every pass and never rebuilds — cold reads
            // and permanent `pending` occupancy for as long as the churn lasts,
            // even though every pass settles. Distinct from the whole-recapture
            // starvation in review-pass-2 §2, which is why the epoch fix does not
            // cover it.
            //
            // The cursor resumes strictly AFTER the last key served, wrapping
            // once. That is fair by construction — a slot the cursor has passed
            // cannot be re-served until it comes round again — and it costs
            // nothing on the common path, where `named` fits in one quantum.
            let rotated: Vec<SlotKey> = match &inner.select_cursor {
                Some(cursor) => named
                    .range((
                        std::ops::Bound::Excluded(cursor.clone()),
                        std::ops::Bound::Unbounded,
                    ))
                    .chain(named.range((
                        std::ops::Bound::Unbounded,
                        std::ops::Bound::Included(cursor.clone()),
                    )))
                    .take(APPLY_QUANTUM)
                    .cloned()
                    .collect(),
                None => named.iter().take(APPLY_QUANTUM).cloned().collect(),
            };
            inner.select_cursor = rotated.last().cloned().or(inner.select_cursor.take());
            let mut selected = Vec::with_capacity(rotated.len());
            for key in rotated {
                // Consuming the identity is safe now: authority was verified under
                // THIS lock, and the install phase re-queues anything it cannot
                // install.
                inner.pending.remove(&key);
                if let Some(slot) = inner.slots.get(&key) {
                    selected.push((key, slot.incarnation));
                }
            }
            // Everything beyond the quantum stays AUTHORITATIVE in `pending`, not
            // in the wake flag.
            selected
        };

        // A pass with nothing to build still has to settle — an epoch may have
        // become complete, or may owe a coherence re-queue. It takes the same
        // commit pin, so `Current` is never reported over an unverified generation.
        if selected.is_empty() {
            let probe = self.source.snapshot(&[]);
            let token = probe.token();
            drop(probe);
            let Some(commit) = self.source.pin_if_current(&[], &token) else {
                // The refusal is REGISTRY-visible movement, exactly as it is on
                // the non-empty path, and it must be marked for exactly the same
                // reason (E3c blockers §3, review-pass-3 §2). What makes the
                // EMPTY path the dangerous one is that its usual compensator is
                // structurally absent: `invalidate_authority_older_than` marks
                // only when `pending` ends non-empty, which is a guaranteed
                // no-op with zero retained slots — and every production node
                // retains zero slots until the warmed-call consumer lands. An
                // authority-only movement (a store install that retracts
                // nothing, a floor publication raising no scoped floor, a poison
                // mark) advances no scoped watch either, so without this mark the
                // actor parks with `owed_recapture` set and health strands at
                // `Rebuilding` indefinitely on an otherwise idle node.
                self.mark_if_movement();
                return ApplyOutcome::Superseded;
            };
            let epoch = commit.epoch();
            let mut inner = self.inner.lock();

            // Authority is revalidated HERE too, not only on the installing path
            // (Kyra OLB-2B-E3b). The phase-1 check is stale by now: the snapshot
            // above ran off-lock, and the actor can be fenced in that window.
            // Installing nothing is not sufficient grounds to settle — `Current`
            // on a full request is what lets the supervisor publish `Healthy`, so
            // a revoked actor reporting it advertises routes on the authority of
            // an incarnation that no longer exists.
            if inner.live_actor != Some(incarnation) {
                self.metrics
                    .stale_actor_rejections
                    .fetch_add(1, Ordering::AcqRel);
                let owed = !inner.pending.is_empty();
                drop(inner);
                if owed {
                    self.work.mark();
                }
                return ApplyOutcome::Superseded;
            }

            let Some(outcome) = commit
                .settle_if_current(&mut || settle(&mut inner, incarnation, epoch, &self.metrics))
            else {
                drop(inner);
                // Counted, and counted as what it IS (review-pass-3 §8): a
                // settlement refused because the source's own publication moved
                // under the pin. This path used to count nothing at all.
                self.metrics
                    .settlements_refused
                    .fetch_add(1, Ordering::AcqRel);
                // Same obligation as the pin refusal above: authority moved
                // inside the probe→settle window, and with no retained slot the
                // movement's own invalidation cannot wake anyone.
                self.mark_if_movement();
                return ApplyOutcome::Superseded;
            };
            let owed = !inner.pending.is_empty();
            drop(inner);
            if owed {
                self.work.mark();
            }
            return outcome;
        }

        // --- phase 2: BRIEF source capture, no registry lock held ---
        let keys: Vec<SlotKey> = selected.iter().map(|(key, _)| key.clone()).collect();
        let snapshot = self.source.snapshot(&keys);
        let snapshot_token = snapshot.token();
        // ONE coherent session observation for the WHOLE quantum (design §3).
        //
        // Captured once, beside the source snapshot and with no registry lock
        // held. Every annotation in every pool this pass publishes comes from
        // this value, so "one coherent session generation" is structural: there
        // is no second sample for a row to be annotated from, and no window in
        // which two rows of one pool could observe different projections.
        let sessions = self.source.session_view();

        // --- phase 3: reconstruct holding NOTHING ---
        // Not the registry lock, and not the source's publication gate: this is the
        // expensive part of a quantum, and blocking private-discovery ingest,
        // expiry and floor mutation behind it is exactly what the snapshot/commit
        // split exists to prevent (Kyra OLB-2B-E3b).
        //
        // The routing pool is built COMPLETELY here too — the row join, the
        // session annotation and the deterministic order are all source/session
        // work, and none of it may happen beneath a lock. What phase 5 does is
        // bind the finished payload to the facts artifact it derives from,
        // which reads nothing and copies no row.
        let built: Vec<BuiltSlot> = selected
            .iter()
            .map(|(key, slot_incarnation)| {
                let facts = snapshot.providers(key);
                let prepared = prepare_route_pool(key, &facts, &sessions);
                BuiltSlot {
                    key: key.clone(),
                    slot_incarnation: *slot_incarnation,
                    facts,
                    prepared,
                }
            })
            .collect();
        drop(snapshot);

        // --- phase 4: the COMMIT pin, before the registry lock ---
        // The SAME `keys` the snapshot was taken over, not an independently
        // rebuilt copy: `pin_if_current`'s contract is that the two batches are
        // identical, and two separately-constructed vectors is the shape in
        // which that quietly stops being true (review 2026-07-29 §3).
        let Some(commit) = self.source.pin_if_current(&keys, &snapshot_token) else {
            // An UNSETTLEABLE refusal is not movement (E3c blockers §2,
            // review-pass-3 §1): the mark below would spin the actor on
            // `Superseded` until shutdown, because nothing has to move for the
            // next pass to refuse identically. `Terminal` additionally discards
            // the queue — a rebuild could never install again, and
            // `retire_terminal` already cleared everything retained at the
            // transition.
            let liveness = self.source.liveness();
            let mut inner = self.inner.lock();
            if liveness.may_requeue() {
                // Install nothing, and put every still-live selected slot back.
                for built in &built {
                    if inner
                        .slots
                        .get(&built.key)
                        .is_some_and(|slot| slot.incarnation == built.slot_incarnation)
                    {
                        inner.pending.insert(built.key.clone());
                    }
                }
            }
            self.metrics
                .discarded_obsolete
                .fetch_add(built.len() as u64, Ordering::AcqRel);
            drop(inner);
            if liveness.may_self_wake() {
                // The movement itself owns a source-watch wake, but that wake
                // alone carries no `registry_work` flag — and without it the
                // requeued identities would not be unioned into the next pass's
                // targets. Mark.
                self.work.mark();
            }
            return ApplyOutcome::Superseded;
        };
        let epoch = commit.epoch();

        // --- phase 5: registry lock BENEATH the commit pin; install ---
        // The source cannot move while `commit` is held. What can still have
        // changed is the actor's authority and the slots themselves.
        let mut slot_moved = false;
        let outcome = {
            let mut inner = self.inner.lock();

            if inner.live_actor != Some(incarnation) {
                // Authority was revoked while we built. Every still-live slot we
                // selected still owes work — requeue it rather than dropping it,
                // and never report a current installation.
                for built in &built {
                    if inner
                        .slots
                        .get(&built.key)
                        .is_some_and(|slot| slot.incarnation == built.slot_incarnation)
                    {
                        inner.pending.insert(built.key.clone());
                    }
                }
                self.metrics
                    .stale_actor_rejections
                    .fetch_add(1, Ordering::AcqRel);
                self.metrics
                    .discarded_obsolete
                    .fetch_add(built.len() as u64, Ordering::AcqRel);
                drop(inner);
                self.work.mark();
                return ApplyOutcome::Superseded;
            }

            // The COMPLETE currentness stamp's session half, revalidated here
            // and nowhere earlier (design §18 step 2's publish-if-current).
            //
            // Read BENEATH the registry lock deliberately, and it is sound to
            // do so because it is a single atomic load: it acquires nothing, so
            // the frozen `source commit pin → registry lock` order is untouched
            // and no source method is called under the lock.
            //
            // Reading it before the lock would leave a hole rather than merely
            // being weaker. A transition publishes its projection, advances the
            // generation, and only THEN takes this lock to retire superseded
            // pools; a pre-lock read could pass, the transition could then
            // advance and complete its whole invalidation, and this pass would
            // still store a pool the transition had already decided to remove.
            // Reading under the lock orders the two: either we see the newer
            // generation and publish nothing, or the transition's invalidation
            // is strictly after our store and clears it.
            let session_current = sessions.still_current();
            let RegistryInner { slots, pending, .. } = &mut *inner;
            for BuiltSlot {
                key,
                slot_incarnation,
                facts,
                prepared,
            } in built
            {
                let Some(slot) = slots.get_mut(&key) else {
                    // Retired while we built: do NOT resurrect it.
                    self.metrics
                        .discarded_obsolete
                        .fetch_add(1, Ordering::AcqRel);
                    continue;
                };
                if slot.incarnation != slot_incarnation {
                    // Retired and re-demanded: this artifact belongs to a dead
                    // incarnation. The live one still owes work.
                    self.metrics
                        .discarded_obsolete
                        .fetch_add(1, Ordering::AcqRel);
                    pending.insert(key);
                    slot_moved = true;
                    continue;
                }
                let ScopedSourceFacts {
                    facts,
                    authority,
                    authority_deadline,
                    grant_fence,
                } = facts;
                let row_expiry = match &facts {
                    SourceFacts::Served(providers) => providers
                        .iter()
                        .map(|p| p.expires_at)
                        .min()
                        .unwrap_or(u64::MAX),
                    // Cold at every read anyway; nothing to bound.
                    SourceFacts::Unserved => u64::MAX,
                };
                // The artifact expires at the EARLIER of what its rows say and
                // what its authority says (scope item 12). Rows alone are not
                // enough, and the zero-provider case is why: an installed Grant
                // with no visible providers gives `row_expiry == u64::MAX`, so a
                // rows-only deadline would claim the artifact never expires
                // while the authority behind it expires in an hour — and nothing
                // else in the artifact could ever retire it.
                let earliest_expiry = row_expiry.min(authority_deadline);
                let facts = Arc::new(SlotBaseFacts {
                    providers: facts,
                    epoch,
                    authority,
                    grant_fence,
                    actor_incarnation: incarnation,
                    slot_incarnation,
                    earliest_expiry,
                });
                // PUBLISH-IF-CURRENT, the pool half. The facts are current —
                // the commit pin proved it, and nothing about session movement
                // makes a discovery row stale — so they install either way. The
                // POOL installs only if the session view it was computed under
                // is still the live one; otherwise this pass publishes facts
                // with no pool beside them, which is the ordinary cold-pool
                // state, and re-queues the slot so the next pass rebuilds the
                // annotation. The stale pool is never observable, because it is
                // never stored.
                let pool = match prepared {
                    Some(prepared) if session_current => Some(prepared.seal(facts.clone())),
                    Some(_) => {
                        self.metrics
                            .pools_refused_stale_session
                            .fetch_add(1, Ordering::AcqRel);
                        // The transition that moved the projection cannot have
                        // re-queued this slot on our behalf: it retires pools
                        // it can SEE, and this one was never published. The
                        // re-queue is ours, and it is the one legitimate wake —
                        // `owed` below marks exactly once for the whole pass.
                        pending.insert(key.clone());
                        None
                    }
                    // `Unserved`, or a spent session-generation space: no
                    // annotation can be made, and a pool is not published over
                    // absent evidence.
                    None => None,
                };
                let published = pool.is_some();
                // Pool first, inside `install_facts_and_pool` (§1.1): a pool
                // present at this moment was derived from the facts being
                // SUPERSEDED, and storing the new facts first would let a
                // lock-free reader pair them with it.
                slot.cells
                    .install_facts_and_pool(facts, pool, &self.publication_observer);
                self.metrics.installs.fetch_add(1, Ordering::AcqRel);
                if published {
                    self.metrics.pools_published.fetch_add(1, Ordering::AcqRel);
                }
            }

            // FINAL complete-vector validation, still under the pin and the
            // registry lock. Authority the source publishes through its own
            // synchronization (revocation floors, poison) can move underneath a
            // held pin, and the facts just written would then be stale. They are
            // already unreadable — the read seam compares the whole stamped
            // epoch — but `Current` is itself load-bearing: it is what lets the
            // supervisor publish `Healthy`. So settle only if nothing moved
            // (Kyra OLB-2B-E3c).
            let settled = commit
                .settle_if_current(&mut || settle(&mut inner, incarnation, epoch, &self.metrics));
            let Some(settled) = settled else {
                // The settlement — not the pin — is where an exhausted store
                // publication generation refuses (review-pass-3 §1). The folded
                // `(poisoned = true, floor_generation = 0)` view is SELF-CONSISTENT
                // across passes, so `pin_if_current` accepts every time and only
                // `matches()` can say no. Re-queueing plus a mark here is therefore
                // a spin that requires no source movement at all: park instead, and
                // let the replacement store install supply the wake.
                let liveness = self.source.liveness();
                if liveness.may_requeue() {
                    for (key, slot_incarnation) in &selected {
                        if inner
                            .slots
                            .get(key)
                            .is_some_and(|slot| slot.incarnation == *slot_incarnation)
                        {
                            inner.pending.insert(key.clone());
                        }
                    }
                }
                // NOT `stale_actor_rejections` (review-pass-3 §8). This branch is
                // reachable only with a LIVE actor — revalidated a few lines above
                // under this same lock — so attributing it to actor lifecycle
                // churn steered operators at supervisor restarts during what is
                // really revocation-publication pressure.
                self.metrics
                    .settlements_refused
                    .fetch_add(1, Ordering::AcqRel);
                drop(inner);
                if liveness.may_self_wake() {
                    self.work.mark();
                }
                return ApplyOutcome::Superseded;
            };

            let owed = !inner.pending.is_empty();
            drop(inner);
            if owed {
                // More quanta owed, or slot movement re-queued work: re-arm rather
                // than looping inside this synchronous call.
                self.work.mark();
            }
            if slot_moved {
                // Slot/demand-origin rejection: its wake is OURS to provide, and
                // the re-queue above made `owed` true, so it was.
                ApplyOutcome::Superseded
            } else {
                settled
            }
        };

        // `commit` drops HERE — after the installation, never before it.
        drop(commit);
        outcome
    }
}

/// One slot's complete off-lock reconstruction: the facts, and the routing
/// pool derived from them, waiting for the artifact identity phase 5 mints.
struct BuiltSlot {
    key: SlotKey,
    slot_incarnation: u64,
    facts: ScopedSourceFacts,
    /// `None` when no pool may be published for this slot at all — `Unserved`
    /// facts, or a terminally exhausted session-generation space.
    prepared: Option<PreparedRoutePool>,
}

/// Project ONE slot's routing pool, entirely off-lock (design §18 step 2).
///
/// Runs in phase 3 with no registry lock, no source lock and no session lock:
/// `facts` is owned material the snapshot already returned, and `sessions` is
/// an immutable observation captured once for the whole quantum.
///
/// Returns `None` — publish no pool — in exactly two cases, and both are
/// refusals rather than empty answers:
///
/// - **`SourceFacts::Unserved`.** The source has no evidence about this scope,
///   so there is nothing to project. A pool with zero rows would be exact
///   negative evidence, which is precisely what `Unserved` is not (see
///   [`SourceFacts`]). A `Served` bucket with zero providers DOES get a pool:
///   that is the exact fresh negative evidence, and W-G13's zero-provider
///   Grant case depends on it existing;
/// - **a spent session-generation space.** Nothing can witness that an
///   annotation is current, so nothing is annotated. Fail closed, and note
///   that this publishes no pool rather than publishing an un-annotated one:
///   a pool whose rows all read `Cold` because the generation space ran out is
///   indistinguishable from one whose providers genuinely have no sessions.
fn prepare_route_pool(
    key: &SlotKey,
    facts: &ScopedSourceFacts,
    sessions: &SessionObservation,
) -> Option<PreparedRoutePool> {
    let SourceFacts::Served(providers) = &facts.facts else {
        return None;
    };
    let session_generation = sessions.generation?;
    // Provider order is the snapshot's own deterministic order, preserved
    // exactly. Re-sorting here would be a second ordering rule, free to
    // disagree with the one the source already imposed off-lock.
    let rows: Vec<UnsensedRouteRow> = providers
        .iter()
        .map(|provider| UnsensedRouteRow {
            provider: provider.provider.clone(),
            // The owner relation the ingest path PROVED, carried through.
            owner_org: provider.owner_org,
            generation: provider.generation,
            expires_at: provider.expires_at,
            // Annotated from the ONE observation this quantum captured. A
            // provider the projection cannot resolve unambiguously is `Cold`,
            // never a guess (design §3's join rules).
            direct: sessions.eligibility(&provider.provider),
        })
        .collect();
    Some(PreparedRoutePool {
        key: key.clone(),
        authority: facts.authority,
        provenance: ProviderProvenance::of(&facts.authority),
        providers: rows.into(),
        session_generation,
    })
}

/// Settle the pass: `Current` only when no work is owed.
///
/// `Current` is the only outcome that may advance routing health, so a pass that
/// left work owed must not claim it.
///
/// Global coherence across every retained slot is required ONLY while a complete
/// recapture epoch is open, and it is enforced by RE-QUEUEING the slots that are
/// not coherent rather than by refusing to settle. Two consequences, both load
/// bearing:
///
/// - ordinary `Caps` and first-demand work no longer reports `Progress` merely
///   because unrelated slots legitimately sit at older generations — a pass that
///   rebuilt everything it was asked to rebuild is complete (Kyra OLB-2B-E3b);
/// - `Progress` now IMPLIES `pending` is non-empty, so the caller always marks.
///   `Progress` with no owed work and no wake — which strands the actor — is
///   unrepresentable rather than merely avoided.
///
/// Re-queueing is also what restarts a recapture whose source generation moved
/// between quanta: the slots built against the older generation stop being
/// coherent, so they go back on the queue and are rebuilt against this one.
fn settle(
    inner: &mut RegistryInner,
    incarnation: u64,
    epoch: SourceEpoch,
    metrics: &RegistryMetrics,
) -> ApplyOutcome {
    if inner.recapture_open {
        let mut displaced = 0;
        for key in inner.incoherent_with(incarnation, epoch) {
            if inner.invalidate(&key) {
                displaced += 1;
            }
            inner.pending.insert(key);
        }
        if displaced > 0 {
            // Slots that WERE installed are no longer coherent with this commit:
            // the source moved mid-epoch and the recapture restarts over them.
            metrics
                .facts_invalidated
                .fetch_add(displaced, Ordering::AcqRel);
            metrics.recaptures_restarted.fetch_add(1, Ordering::AcqRel);
        }
    }

    if inner.pending.is_empty() {
        inner.recapture_open = false;
        ApplyOutcome::Current {
            source_generation: epoch.generation,
        }
    } else {
        ApplyOutcome::Progress {
            source_generation: epoch.generation,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::adapter::net::behavior::org::OrgId;
    use std::sync::atomic::AtomicBool;

    fn scope(seed: u8) -> PrivateAudienceScope {
        PrivateAudienceScope::new(CapabilityAudienceScope::Owner {
            org_id: OrgId::from_bytes([seed; 32]),
            audience_handle: [seed; 32],
        })
        .expect("owner scopes are private")
    }

    fn key(seed: u8, tag: &str) -> SlotKey {
        SlotKey {
            scope: scope(seed),
            capability: CapabilityAuthorityId::for_tag(tag),
        }
    }

    type Hook = Box<dyn Fn() + Send + Sync>;
    type SharedHook = Arc<dyn Fn() + Send + Sync>;

    fn entity(seed: u8) -> EntityId {
        EntityId::from_bytes([seed; 32])
    }

    fn provider_row(seed: u8, generation: u64, expires_at: u64) -> PrivateCapabilityProvider {
        PrivateCapabilityProvider {
            provider: entity(seed),
            owner_org: OrgId::from_bytes([seed.wrapping_add(0x40); 32]),
            expires_at,
            generation,
        }
    }

    /// The test double for the node's published session/direct-state
    /// projection.
    ///
    /// Shaped exactly like the production one: an immutable value carrying its
    /// own generation, republished as a whole, ordered by a shared
    /// [`SessionCurrentness`]. A witness that could move rows without moving
    /// the generation would be modelling a node this code cannot produce.
    #[derive(Default)]
    struct TestProjection {
        generation: u64,
        rows: BTreeMap<EntityId, DirectEligibility>,
        /// The registry, so the ANNOTATION JOIN can assert directly that it
        /// runs with no registry lock held. `providers` already asserts it for
        /// the row capture; this is the pool-build half, and it is the one the
        /// design's lock-discipline obligation is about.
        registry: Option<Arc<NodeOrgRoutingRegistry>>,
    }

    impl SessionEligibility for TestProjection {
        fn eligibility(&self, provider: &EntityId) -> DirectEligibility {
            if let Some(registry) = self.registry.as_ref() {
                assert!(
                    registry.inner.try_lock().is_some(),
                    "no registry lock may be held while the pool is built"
                );
            }
            self.rows
                .get(provider)
                .copied()
                .unwrap_or(DirectEligibility::Cold)
        }
    }

    struct TestSessions {
        currentness: Arc<SessionCurrentness>,
        published: parking_lot::Mutex<Arc<TestProjection>>,
        registry: parking_lot::Mutex<Option<Arc<NodeOrgRoutingRegistry>>>,
        /// `session_view` calls, so a witness can assert the quantum captured
        /// EXACTLY ONE observation — the structural half of "one coherent
        /// session generation".
        observations: AtomicU64,
    }

    impl TestSessions {
        fn new() -> Self {
            Self {
                currentness: SessionCurrentness::new(),
                published: parking_lot::Mutex::new(Arc::new(TestProjection::default())),
                registry: parking_lot::Mutex::new(None),
                observations: AtomicU64::new(0),
            }
        }

        /// Republish, in the production order: reserve, store, commit.
        fn publish(&self, rows: BTreeMap<EntityId, DirectEligibility>) -> u64 {
            let generation = self.currentness.reserve().expect("generation space");
            let registry = self.registry.lock().clone();
            *self.published.lock() = Arc::new(TestProjection {
                generation,
                rows,
                registry,
            });
            self.currentness.commit(generation);
            generation
        }

        fn observe(&self) -> SessionObservation {
            self.observations.fetch_add(1, Ordering::AcqRel);
            let published = self.published.lock().clone();
            let generation = self.currentness.generation().map(|_| published.generation);
            SessionObservation::new(generation, published, self.currentness.clone())
        }
    }

    fn direct(node_id: u64) -> DirectEligibility {
        DirectEligibility::Direct {
            node_id,
            session_id: node_id ^ 0xfeed,
        }
    }

    /// The observable state of the source the witnesses drive.
    ///
    /// `gate` stands in for BOTH the query-visible generation and the scoped
    /// mutation-publication gate E3c will bind here: taking it is what a mutation
    /// must do to advance. That makes "the gate is not held across reconstruction"
    /// directly assertable — a rival mutation on another thread must be able to
    /// take it mid-build.
    struct SourceState {
        gate: parking_lot::Mutex<u64>,
        /// Set to make a HELD commit pin report that authority moved underneath
        /// it — the transition no gate can exclude.
        authority_moved: AtomicBool,
        /// Runs during reconstruction, so a witness can move the world mid-build.
        during_build: parking_lot::Mutex<Option<Hook>>,
        /// Runs inside the off-lock snapshot — the window in which the phase-1
        /// authority check goes stale.
        on_snapshot: parking_lot::Mutex<Option<Hook>>,
        /// Runs when the COMMIT pin is released — i.e. strictly after installation.
        on_commit_release: parking_lot::Mutex<Option<SharedHook>>,
        queried: parking_lot::Mutex<Vec<SlotKey>>,
        snapshots: AtomicU64,
        registry: parking_lot::Mutex<Option<Arc<NodeOrgRoutingRegistry>>>,
        /// The provider rows every captured scope reconstructs with. Empty by
        /// default, which is `Served(0 providers)` — exact negative evidence,
        /// and the shape most of the pre-2B.3c witnesses assert against.
        rows: parking_lot::Mutex<Vec<PrivateCapabilityProvider>>,
        /// The captured scope's own authority deadline, so the deadline
        /// witnesses can place the ZERO-provider case exactly.
        authority_deadline: AtomicU64,
        /// Reconstruct every scope as `Unserved` — the source cannot speak for
        /// it at all, which is NOT the same as a served empty bucket.
        unserved: AtomicBool,
        /// The node's session/direct-state projection (OLB-2B.3c step 2).
        sessions: TestSessions,
    }

    impl SourceState {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                gate: parking_lot::Mutex::new(1),
                authority_moved: AtomicBool::new(false),
                during_build: parking_lot::Mutex::new(None),
                on_snapshot: parking_lot::Mutex::new(None),
                on_commit_release: parking_lot::Mutex::new(None),
                queried: parking_lot::Mutex::new(Vec::new()),
                snapshots: AtomicU64::new(0),
                registry: parking_lot::Mutex::new(None),
                rows: parking_lot::Mutex::new(Vec::new()),
                authority_deadline: AtomicU64::new(u64::MAX),
                unserved: AtomicBool::new(false),
                sessions: TestSessions::new(),
            })
        }
        fn queries(&self) -> Vec<SlotKey> {
            self.queried.lock().clone()
        }
        fn reset(&self) {
            self.queried.lock().clear();
        }
        fn generation(&self) -> u64 {
            *self.gate.lock()
        }
        /// Advance the query-visible generation, as a mutation+publication would.
        fn advance(&self) {
            *self.gate.lock() += 1;
        }
        fn assert_no_registry_lock(&self, context: &str) {
            if let Some(registry) = self.registry.lock().clone() {
                assert!(registry.inner.try_lock().is_some(), "{context}");
            }
        }
    }

    struct TestSource(Arc<SourceState>);

    /// Owns its bounded material: reconstruction touches no source lock.
    struct TestSnapshot {
        state: Arc<SourceState>,
        generation: u64,
        _captured: Vec<SlotKey>,
    }

    impl SourceSnapshot for TestSnapshot {
        fn token(&self) -> SourceToken {
            SourceToken::new(vec![self.generation])
        }
        fn providers(&self, key: &SlotKey) -> ScopedSourceFacts {
            self.state.queried.lock().push(key.clone());
            self.state
                .assert_no_registry_lock("no registry lock may be held across reconstruction");
            assert!(
                self.state.gate.try_lock().is_some(),
                "no source/publication lock may be held across decode, sort or projection"
            );
            let hook = self.state.during_build.lock().take();
            if let Some(hook) = hook {
                hook();
            }
            let unserved = self.state.unserved.load(Ordering::Acquire);
            let facts = if unserved {
                SourceFacts::Unserved
            } else {
                SourceFacts::Served(self.state.rows.lock().clone().into())
            };
            ScopedSourceFacts {
                facts,
                authority: ScopedDiscoveryAuthorityStamp::Owner,
                grant_fence: GrantArtifactFence::Publication(0),
                // An `Unserved` reconstruction carries NO deadline, exactly as
                // the production source's does: there is no authority here to
                // expire, and the artifact reads cold at every read anyway.
                authority_deadline: if unserved {
                    u64::MAX
                } else {
                    self.state.authority_deadline.load(Ordering::Acquire)
                },
            }
        }
    }

    struct TestCommitPin<'a> {
        state: Arc<SourceState>,
        generation: parking_lot::MutexGuard<'a, u64>,
    }

    impl SourceCommitPin for TestCommitPin<'_> {
        fn settle_if_current(
            &self,
            settle: &mut dyn FnMut() -> ApplyOutcome,
        ) -> Option<ApplyOutcome> {
            (!self.state.authority_moved.load(Ordering::Acquire)).then(settle)
        }
        fn epoch(&self) -> SourceEpoch {
            SourceEpoch {
                generation: *self.generation,
                authority: 0,
                floor_generation: 0,
                poisoned: false,
            }
        }
    }

    impl Drop for TestCommitPin<'_> {
        fn drop(&mut self) {
            let hook = self.state.on_commit_release.lock().clone();
            if let Some(hook) = hook {
                hook();
            }
        }
    }

    impl SlotSource for TestSource {
        fn snapshot(&self, keys: &[SlotKey]) -> Box<dyn SourceSnapshot> {
            self.0
                .assert_no_registry_lock("no registry lock may be held across the source snapshot");
            // Brief: the gate guard is released before this returns.
            let generation = *self.0.gate.lock();
            self.0.snapshots.fetch_add(1, Ordering::AcqRel);
            let hook = self.0.on_snapshot.lock().take();
            if let Some(hook) = hook {
                hook();
            }
            Box::new(TestSnapshot {
                state: self.0.clone(),
                generation,
                _captured: keys.to_vec(),
            })
        }

        fn session_view(&self) -> SessionObservation {
            self.0
                .assert_no_registry_lock("no registry lock may be held across the session capture");
            self.0.sessions.observe()
        }

        fn pin_if_current(
            &self,
            _keys: &[SlotKey],
            expected: &SourceToken,
        ) -> Option<Box<dyn SourceCommitPin + '_>> {
            self.0.assert_no_registry_lock(
                "the commit pin must be acquired BEFORE the registry lock",
            );
            let generation = self.0.gate.lock();
            if SourceToken::new(vec![*generation]) != *expected {
                return None;
            }
            Some(Box::new(TestCommitPin {
                state: self.0.clone(),
                generation,
            }))
        }
    }

    struct Fixture {
        registry: Arc<NodeOrgRoutingRegistry>,
        source: Arc<SourceState>,
        metrics: Arc<RegistryMetrics>,
    }

    impl Fixture {
        fn family(&self) -> RoutingFamily {
            self.registry.new_family().expect("family")
        }
    }

    /// Builds a registry with actor incarnation 1 already live — the state
    /// `RoutingSupervisor` establishes before it applies anything.
    fn fixture() -> Fixture {
        let source = SourceState::new();
        let metrics: Arc<RegistryMetrics> = Arc::default();
        let registry = NodeOrgRoutingRegistry::new(
            Arc::new(TestSource(source.clone())),
            Arc::default(),
            metrics.clone(),
        );
        *source.registry.lock() = Some(registry.clone());
        *source.sessions.registry.lock() = Some(registry.clone());
        registry.activate_incarnation(1);
        Fixture {
            registry,
            source,
            metrics,
        }
    }

    /// A minimal published artifact, for witnesses about cell LIFETIME rather
    /// than cell content.
    fn facts() -> Arc<SlotBaseFacts> {
        Arc::new(SlotBaseFacts {
            providers: SourceFacts::Unserved,
            epoch: SourceEpoch::default(),
            authority: ScopedDiscoveryAuthorityStamp::Owner,
            actor_incarnation: 1,
            slot_incarnation: 1,
            grant_fence: GrantArtifactFence::Publication(0),
            earliest_expiry: u64::MAX,
        })
    }

    fn request(registry_work: bool, dirty: DirtyCapabilities) -> ApplyRequest {
        ApplyRequest {
            batch: crate::adapter::net::behavior::org_scoped_store::PrivateDiscoveryChangeBatch {
                generation: 1,
                dirty,
            },
            registry_work,
        }
    }

    fn caps(tags: &[&str]) -> DirtyCapabilities {
        DirtyCapabilities::Caps(
            tags.iter()
                .map(|t| CapabilityAuthorityId::for_tag(t))
                .collect(),
        )
    }

    // ---------------------------------------------------------------- identity

    /// The public partition cannot form a routing slot key at all.
    #[test]
    fn the_public_scope_cannot_form_a_slot_key() {
        assert!(
            PrivateAudienceScope::new(CapabilityAudienceScope::Public).is_none(),
            "this is the PRIVATE routing consumer"
        );
        assert!(PrivateAudienceScope::new(CapabilityAudienceScope::Owner {
            org_id: OrgId::from_bytes([7; 32]),
            audience_handle: [7; 32],
        })
        .is_some());
    }

    /// Clones of ONE family share its identity and its 64-handle budget; an
    /// independently minted family gets its own.
    #[test]
    fn a_clone_family_shares_one_identity_and_one_budget() {
        let f = fixture();
        let family = f.family();
        let clone = family.clone();
        let independent = f.family();

        let _a = family.demand(key(1, "nrpc:a")).expect("a");
        let _b = clone.demand(key(1, "nrpc:b")).expect("b");
        assert_eq!(family.handles(), 2, "the clone spends the SAME budget");
        assert_eq!(clone.handles(), 2);
        assert_eq!(independent.handles(), 0, "a distinct family is unaffected");

        // Exhaust the shared budget through the clone; the original is exhausted too.
        let mut held = Vec::new();
        for index in 0..(MAX_HANDLES_PER_FAMILY - 2) {
            held.push(
                clone
                    .demand(key(1, &format!("nrpc:s{index}")))
                    .expect("within the shared bound"),
            );
        }
        assert_eq!(
            family.demand(key(1, "nrpc:over")).err(),
            Some(DemandRefused::FamilyAtCapacity),
            "the original clone is bounded by what its clone spent"
        );
        assert!(
            independent.demand(key(1, "nrpc:own")).is_ok(),
            "an independent family still has its own budget"
        );
    }

    // ------------------------------------------------------------- recapture

    /// First demand gets a full CURRENT recapture even when earlier source deltas
    /// were already destructively drained — the slot's own pending identity is what
    /// drives it, not any surviving source delta.
    #[test]
    fn first_demand_after_drained_deltas_still_gets_a_full_recapture() {
        let f = fixture();
        // Historical deltas: applied and consumed before this slot ever existed.
        f.registry
            .apply(1, request(false, DirtyCapabilities::RebuildAll));
        f.source.reset();

        let family = f.family();
        let _held = family.demand(key(1, "nrpc:a")).expect("demand");
        assert_eq!(f.registry.pending_slots(), 1);

        // A CLEAN source pass still rebuilds it, from registry work alone.
        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(f.source.queries(), vec![key(1, "nrpc:a")]);
        assert!(matches!(outcome, ApplyOutcome::Current { .. }));
        let facts = f
            .registry
            .base_facts_unvalidated(&key(1, "nrpc:a"))
            .expect("built");
        assert_eq!(facts.slot_incarnation, 2, "family 1, then this slot");
        assert_eq!(facts.actor_incarnation, 1);
        assert_eq!(f.registry.pending_slots(), 0);
    }

    /// Two families converge on ONE node slot and ONE reconstruction.
    #[test]
    fn two_families_share_one_slot_and_one_reconstruction() {
        let f = fixture();
        let a = f.family().demand(key(1, "nrpc:a")).expect("a");
        let b = f.family().demand(key(1, "nrpc:a")).expect("b");
        assert_eq!(f.registry.retained_slots(), 1, "one shared node slot");

        f.source.reset();
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(
            f.source.queries(),
            vec![key(1, "nrpc:a")],
            "reconstructed once per SLOT, not once per family"
        );
        drop((a, b));
    }

    /// A DIFFERENT audience scope is a different slot: indexing never broadens
    /// authority by sharing across scopes.
    #[test]
    fn a_different_audience_scope_is_a_different_slot() {
        let f = fixture();
        let family = f.family();
        let _a = family.demand(key(1, "nrpc:a")).expect("a");
        let _b = family.demand(key(2, "nrpc:a")).expect("b");
        assert_eq!(
            f.registry.retained_slots(),
            2,
            "same capability, different scope => distinct slots"
        );

        f.source.reset();
        f.registry.apply(1, request(false, caps(&["nrpc:a"])));
        let mut queried = f.source.queries();
        queried.sort();
        let mut expected = vec![key(1, "nrpc:a"), key(2, "nrpc:a")];
        expected.sort();
        assert_eq!(
            queried, expected,
            "the capability bucket holds both scoped slots, each rebuilt separately"
        );
    }

    // ---------------------------------------------------------------- bounds

    /// The 65th family handle is refused without corrupting the first 64, and a
    /// DUPLICATE demand counts toward the bound rather than bypassing it.
    #[test]
    fn the_sixty_fifth_family_handle_is_refused_without_corrupting_the_first_64() {
        let f = fixture();
        let family = f.family();
        let mut held = Vec::new();
        for index in 0..MAX_HANDLES_PER_FAMILY {
            held.push(
                family
                    .demand(key(1, &format!("nrpc:f{index}")))
                    .expect("within the bound"),
            );
        }
        assert_eq!(
            family.demand(key(1, "nrpc:over")).err(),
            Some(DemandRefused::FamilyAtCapacity)
        );
        // A duplicate of an EXISTING key is still a handle, so it is refused too.
        assert_eq!(
            family.demand(key(1, "nrpc:f0")).err(),
            Some(DemandRefused::FamilyAtCapacity),
            "duplicate demand cannot bypass the handle bound"
        );
        assert_eq!(f.metrics.refused_family_at_capacity(), 2);
        assert_eq!(
            f.registry.retained_slots(),
            MAX_HANDLES_PER_FAMILY,
            "the first 64 are intact and no slot was created for the refusals"
        );
        assert_eq!(held.len(), MAX_HANDLES_PER_FAMILY);
    }

    /// The 257th unique node slot is deterministically unretained, and no live slot
    /// is evicted for it.
    #[test]
    fn the_two_hundred_fifty_seventh_slot_is_deterministically_unretained() {
        let f = fixture();
        let mut held = Vec::new();
        let mut family = f.family();
        for index in 0..MAX_NODE_SLOTS {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            held.push(
                family
                    .demand(key(1, &format!("nrpc:n{index}")))
                    .expect("within the node bound"),
            );
        }
        assert_eq!(f.registry.retained_slots(), MAX_NODE_SLOTS);

        let beyond = key(1, "nrpc:beyond");
        assert_eq!(
            f.family().demand(beyond.clone()).err(),
            Some(DemandRefused::NodeAtCapacity)
        );
        assert_eq!(f.metrics.refused_node_at_capacity(), 1);
        assert_eq!(
            f.registry.retained_slots(),
            MAX_NODE_SLOTS,
            "no live slot was evicted to make room"
        );
        assert!(
            f.registry.base_facts_unvalidated(&beyond).is_none(),
            "the refused key is cold/unretained"
        );

        // The STRUCTURAL pool bound (design §4, §18.3): pool publications
        // live on slot cells and nowhere else, so 256 retained slots can each
        // own one pool — and the refused 257th key has no cell for a pool to
        // exist in. The install seam is a per-slot store, so against a refused
        // key it is a no-op rather than an allocation.
        let first = key(1, "nrpc:n0");
        f.registry.install_facts_for_test(first.clone(), facts());
        let base = f.registry.base_facts_unvalidated(&first).expect("facts");
        f.registry.install_unsensed_pool_for_test(
            &first,
            Arc::new(ScopedUnsensedRoutePool::for_test(base)),
        );
        assert!(
            f.registry.unsensed_pool_unvalidated(&first).is_some(),
            "a retained slot owns its pool publication"
        );
        f.registry.install_unsensed_pool_for_test(
            &beyond,
            Arc::new(ScopedUnsensedRoutePool::for_test(facts())),
        );
        assert!(
            f.registry.unsensed_pool_unvalidated(&beyond).is_none(),
            "no slot, no cell, no 257th pool — the bound is structural, not \
             counted"
        );
    }

    /// An exhausted identity space refuses deterministically: no slot retained, no
    /// wrap, no panic — and no family minted either.
    #[test]
    fn an_exhausted_identity_space_refuses_deterministically() {
        let f = fixture();
        let family = f.family();
        let _live = family.demand(key(1, "nrpc:a")).expect("a");
        let retained = f.registry.retained_slots();

        f.registry.inner.lock().next_id = u64::MAX;

        assert_eq!(
            family.demand(key(1, "nrpc:fresh")).err(),
            Some(DemandRefused::IdSpaceExhausted),
            "a NEW slot needs a fresh incarnation and cannot have one"
        );
        assert_eq!(
            f.registry.retained_slots(),
            retained,
            "the refusal retained nothing"
        );
        assert_eq!(
            f.registry.inner.lock().next_id,
            u64::MAX,
            "and mutated no counter"
        );
        assert_eq!(
            f.registry.new_family().err(),
            Some(DemandRefused::IdSpaceExhausted)
        );
        assert_eq!(f.metrics.refused_id_space_exhausted(), 2);

        // A duplicate demand for an EXISTING slot needs no new identity.
        assert!(
            family.demand(key(1, "nrpc:a")).is_ok(),
            "sharing a retained slot allocates nothing"
        );
    }

    // ------------------------------- OLB-2B.3b: atomic demand-set acquisition

    /// The TOTAL state a refused acquisition must leave untouched.
    ///
    /// Total on purpose (step 1's `300e80f6c` lesson): a witness that asserts a
    /// SUBSET of {slots, pending, handles, identities, retirements} passes
    /// against the naive `demand`-in-a-loop implementation, whose prefix IS
    /// released on the error path — so slots and handles come back to their
    /// starting values and only the identity space and the retirement counter
    /// still show what happened.
    #[derive(Debug, PartialEq, Eq)]
    struct NoEffect {
        slots: usize,
        pending: usize,
        handles: usize,
        ids: u64,
        retired: u64,
    }

    fn no_effect(f: &Fixture, family: &RoutingFamily) -> NoEffect {
        NoEffect {
            slots: f.registry.retained_slots(),
            pending: f.registry.pending_slots(),
            handles: family.handles(),
            ids: f.registry.allocated_ids_for_test(),
            retired: f.metrics.slots_retired(),
        }
    }

    fn set(seeds: &[u8], tag: &str) -> Vec<SlotKey> {
        seeds.iter().map(|s| key(*s, tag)).collect()
    }

    /// W-A1 — the family bound is checked for the WHOLE set, and a refusal
    /// retains nothing.
    ///
    /// The set is deliberately larger than the remaining budget by ONE, with
    /// several NEW keys ahead of the one that does not fit. That is the schedule
    /// a per-key loop gets wrong: it admits the prefix, mints an incarnation for
    /// each, and only then discovers the bound.
    #[test]
    fn a_demand_set_past_the_family_bound_retains_nothing() {
        let f = fixture();
        let family = f.family();
        let mut held = Vec::new();
        for i in 0..62u32 {
            held.push(
                family
                    .demand(key(1, &format!("nrpc:fill-{i}")))
                    .expect("fill"),
            );
        }
        assert_eq!(family.handles(), 62);
        let before = no_effect(&f, &family);

        assert_eq!(
            family.demand_set(set(&[10, 11, 12], "nrpc:wide")).err(),
            Some(DemandRefused::FamilyAtCapacity),
            "62 + 3 exceeds the 64-handle family budget"
        );
        assert_eq!(
            no_effect(&f, &family),
            before,
            "a refused set retains no slot, queues no work, spends no handle, \
             consumes no identity and retires nothing"
        );
        assert_eq!(f.metrics.refused_family_at_capacity(), 1);

        // The bound is exact, not conservative: a set that FITS still goes
        // through, so the refusal above was the budget and not the batching.
        let fits = family
            .demand_set(set(&[10, 11], "nrpc:wide"))
            .expect("62 + 2 fits exactly");
        assert_eq!(fits.len(), 2);
        assert_eq!(family.handles(), 64);
    }

    /// W-A2 — the node bound is checked for every NEW distinct slot, and a
    /// refusal retains nothing.
    #[test]
    fn a_demand_set_past_the_node_bound_retains_nothing() {
        let f = fixture();
        // Fill the node to 255 with four families, none of which may exceed 64.
        let mut fillers = Vec::new();
        let mut held = Vec::new();
        for chunk in 0..4u32 {
            let filler = f.family();
            for i in 0..64u32 {
                let n = chunk * 64 + i;
                if n == 255 {
                    break;
                }
                held.push(
                    filler
                        .demand(key(2, &format!("nrpc:node-{n}")))
                        .expect("fill"),
                );
            }
            fillers.push(filler);
        }
        assert_eq!(f.registry.retained_slots(), 255);

        let family = f.family();
        let before = no_effect(&f, &family);
        assert_eq!(
            family.demand_set(set(&[20, 21], "nrpc:pair")).err(),
            Some(DemandRefused::NodeAtCapacity),
            "255 + 2 new slots exceeds the 256-slot node bound"
        );
        assert_eq!(
            no_effect(&f, &family),
            before,
            "the FIRST of the two must not be retained"
        );
        assert_eq!(f.metrics.refused_node_at_capacity(), 1);

        // One new slot still fits — so the refusal counted the set, not the call.
        let one = family
            .demand_set(set(&[20], "nrpc:pair"))
            .expect("255 + 1 fits");
        assert_eq!(one.len(), 1);
        assert_eq!(f.registry.retained_slots(), 256);
    }

    /// W-A2b — an ALREADY-RETAINED key costs no node capacity, so a set that
    /// shares slots with another family is admitted at the bound.
    ///
    /// The control for W-A2: without it, "count every key" and "count every NEW
    /// key" both refuse the case above, and the witness would not distinguish
    /// the implementation from a strictly-worse one that never shares.
    #[test]
    fn a_demand_set_over_retained_slots_costs_no_node_capacity() {
        let f = fixture();
        let mut fillers = Vec::new();
        let mut held = Vec::new();
        for chunk in 0..4u32 {
            let filler = f.family();
            for i in 0..64u32 {
                held.push(
                    filler
                        .demand(key(2, &format!("nrpc:node-{}", chunk * 64 + i)))
                        .expect("fill"),
                );
            }
            fillers.push(filler);
        }
        assert_eq!(f.registry.retained_slots(), 256, "the node is FULL");

        let family = f.family();
        let shared = family
            .demand_set(vec![key(2, "nrpc:node-0"), key(2, "nrpc:node-1")])
            .expect("sharing retained slots needs no node capacity");
        assert_eq!(shared.len(), 2);
        assert_eq!(f.registry.retained_slots(), 256, "and created nothing");
        assert_eq!(f.metrics.refused_node_at_capacity(), 0);
    }

    /// W-A3 — exhaustion refuses the whole set and consumes no identity.
    #[test]
    fn an_exhausted_identity_space_refuses_a_whole_demand_set() {
        let f = fixture();
        let family = f.family();
        let _live = family.demand(key(1, "nrpc:a")).expect("a");
        f.registry.exhaust_ids_for_test();
        let before = no_effect(&f, &family);

        assert_eq!(
            family.demand_set(set(&[30, 31], "nrpc:fresh")).err(),
            Some(DemandRefused::IdSpaceExhausted),
            "two new slots need two incarnations and there are none"
        );
        assert_eq!(no_effect(&f, &family), before, "and nothing moved");
        assert_eq!(f.metrics.refused_id_space_exhausted(), 1);

        // A set over ONLY retained slots needs no incarnation, so exhaustion
        // does not refuse it. The control that stops "refuse everything once
        // exhausted" from passing this witness.
        let shared = family
            .demand_set(vec![key(1, "nrpc:a")])
            .expect("a retained slot needs no fresh identity");
        assert_eq!(shared.len(), 1);
    }

    /// A superseded reader must not keep reading a RETIRED incarnation's facts.
    ///
    /// Dies to: retiring a slot without clearing its publication cell. The cell is
    /// an `Arc` outstanding readers still hold, so removing the slot from the
    /// registry map does not reach them — the artifact stays readable forever
    /// through a set that owns no reference at all. That is retention without
    /// ownership, which is the one thing `DemandSet` exists to make impossible.
    #[test]
    fn a_superseded_reader_cannot_read_a_retired_incarnations_facts() {
        let f = fixture();
        let family = f.family();
        let old_key = key(1, "nrpc:retire");
        let old = family.demand_set(vec![old_key.clone()]).expect("acquired");

        f.registry.install_facts_for_test(old_key.clone(), facts());
        assert!(
            old.base_facts_unvalidated(0).is_some(),
            "precondition: the reader can see its own incarnation's facts"
        );

        // Replace with a DISJOINT key: the old slot's last reference goes away.
        let new = old.replace(vec![key(2, "nrpc:retire")]).expect("replaced");
        assert_eq!(f.registry.retained_slots(), 1, "the old slot retired");

        assert!(
            old.base_facts_unvalidated(0).is_none(),
            "an old reader must not retain facts from the retired incarnation"
        );
        drop(new);
    }

    /// A COMMON key stays readable while the successor owns it, and goes empty when
    /// the successor gives up the last reference.
    ///
    /// The control for the witness above: clearing on retirement must not clear on
    /// TRANSFER. The old reader shares the successor's cell, and the scope is still
    /// genuinely retained — by the successor — so it must still read.
    #[test]
    fn a_common_key_reader_follows_the_successors_ownership() {
        let f = fixture();
        let family = f.family();
        let shared = key(1, "nrpc:common");
        let old = family
            .demand_set(vec![shared.clone(), key(2, "nrpc:common")])
            .expect("acquired");
        f.registry.install_facts_for_test(shared.clone(), facts());
        assert!(old.base_facts_unvalidated(0).is_some());

        // The successor keeps the common key and sheds the other.
        let new = old.replace(vec![shared.clone()]).expect("replaced");
        assert!(
            old.base_facts_unvalidated(0).is_some(),
            "the common scope is still retained — by the successor — so it still reads"
        );

        drop(new);
        assert!(
            old.base_facts_unvalidated(0).is_none(),
            "and goes empty the moment the successor releases the last reference"
        );
        assert_eq!(f.registry.retained_slots(), 0);
    }

    /// A fresh demand for a retired `SlotKey` gets a FRESH cell and incarnation, and
    /// cannot repopulate the stale one.
    #[test]
    fn a_fresh_demand_after_retirement_gets_a_fresh_cell() {
        let f = fixture();
        let family = f.family();
        let reused = key(1, "nrpc:reused");
        let old = family.demand_set(vec![reused.clone()]).expect("acquired");
        f.registry.install_facts_for_test(reused.clone(), facts());
        assert!(old.base_facts_unvalidated(0).is_some());

        let successor = old.replace(vec![key(2, "nrpc:reused")]).expect("replaced");
        assert!(old.base_facts_unvalidated(0).is_none());

        // Demand the SAME key again. It is a new slot with a new incarnation.
        let fresh = family
            .demand_set(vec![reused.clone()])
            .expect("re-demanded");
        f.registry.install_facts_for_test(reused.clone(), facts());
        assert!(
            fresh.base_facts_unvalidated(0).is_some(),
            "the live incarnation publishes into its own cell"
        );
        assert!(
            old.base_facts_unvalidated(0).is_none(),
            "and the stale cell stays empty — a fresh publication cannot reach it"
        );
        drop((successor, fresh));
    }

    /// An in-flight apply for a DEAD incarnation cannot republish into the cleared
    /// cell, even when the same key has been re-demanded.
    #[test]
    fn a_stale_incarnation_apply_cannot_republish_a_cleared_cell() {
        let f = fixture();
        let family = f.family();
        let contested = key(1, "nrpc:race");
        let old = family
            .demand_set(vec![contested.clone()])
            .expect("acquired");
        f.registry
            .install_facts_for_test(contested.clone(), facts());

        // Retire it, then re-demand the same key: a new incarnation, a new cell.
        let successor = old.replace(vec![key(2, "nrpc:race")]).expect("replaced");
        let fresh = family
            .demand_set(vec![contested.clone()])
            .expect("re-demanded");

        // Drive a real quantum. The reconstruction reaches the LIVE slot only.
        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            matches!(outcome, ApplyOutcome::Current { .. }),
            "{outcome:?}"
        );
        assert!(
            old.base_facts_unvalidated(0).is_none(),
            "the dead incarnation's cell is unreachable from the registry"
        );
        drop((successor, fresh));
    }

    // ---------------------- OLB-2B.3c §18 step 1: the second publication cell

    /// **Both cells, one acquisition** (design §1.1): the handle's pool cell IS
    /// the slot's, cloned under the same registry acquisition that took the
    /// reference — so a pool published later is visible through a handle taken
    /// earlier.
    ///
    /// Dies to: fabricating a detached pool cell at demand time, which would
    /// leave the handle permanently, silently cold on the pool plane while its
    /// facts plane works.
    #[test]
    fn a_demand_handle_couples_both_publication_cells() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:pool");
        let handle = family.demand(k.clone()).expect("retained");

        f.registry.install_facts_for_test(k.clone(), facts());
        let base = f.registry.base_facts_unvalidated(&k).expect("facts");
        let pool = Arc::new(ScopedUnsensedRoutePool::for_test(base));
        f.registry.install_unsensed_pool_for_test(&k, pool.clone());

        assert!(
            handle.base_facts_unvalidated().is_some(),
            "the facts plane reads through the handle"
        );
        assert!(
            handle
                .unsensed_pool_unvalidated()
                .is_some_and(|read| Arc::ptr_eq(&read, &pool)),
            "and the pool plane reads the EXACT published pool — the same \
             cell, not a copy taken at demand time"
        );
    }

    /// The same coupling for a SET: each contributor reads its OWN slot's
    /// pool, so the `keys[i] <-> cells[i]` pairing holds across both planes.
    #[test]
    fn a_demand_set_couples_both_cells_per_contributor() {
        let f = fixture();
        let family = f.family();
        let set = family
            .demand_set(vec![key(1, "nrpc:pool"), key(2, "nrpc:pool")])
            .expect("acquired");

        for k in set.keys().to_vec() {
            f.registry.install_facts_for_test(k.clone(), facts());
            let base = f.registry.base_facts_unvalidated(&k).expect("facts");
            f.registry.install_unsensed_pool_for_test(
                &k,
                Arc::new(ScopedUnsensedRoutePool::for_test(base)),
            );
        }
        for (index, k) in set.keys().to_vec().iter().enumerate() {
            let through_set = set.unsensed_pool_unvalidated(index).expect("pool");
            let through_registry = f.registry.unsensed_pool_unvalidated(k).expect("pool");
            assert!(
                Arc::ptr_eq(&through_set, &through_registry),
                "contributor {index} reads its own slot's pool — the planes \
                 cannot misalign"
            );
        }
    }

    /// Retiring the last reference clears BOTH cells: a retired incarnation's
    /// routing pool is exactly as dead as its facts (W-S1, extended to the
    /// second cell).
    ///
    /// Dies to: retiring the slot without clearing the pool cell — a
    /// superseded reader would keep reading a retired incarnation's ROUTING
    /// POOL after its facts were correctly cleared.
    #[test]
    fn retiring_the_last_reference_clears_both_cells() {
        let f = fixture();
        let family = f.family();
        let old_key = key(1, "nrpc:retire");
        let old = family.demand_set(vec![old_key.clone()]).expect("acquired");
        f.registry.install_facts_for_test(old_key.clone(), facts());
        let base = f.registry.base_facts_unvalidated(&old_key).expect("facts");
        f.registry.install_unsensed_pool_for_test(
            &old_key,
            Arc::new(ScopedUnsensedRoutePool::for_test(base)),
        );
        assert!(old.unsensed_pool_unvalidated(0).is_some(), "precondition");

        // Replace with a DISJOINT key: the old slot's last reference goes away.
        let new = old.replace(vec![key(2, "nrpc:retire")]).expect("replaced");
        assert!(
            old.base_facts_unvalidated(0).is_none() && old.unsensed_pool_unvalidated(0).is_none(),
            "a retired incarnation's cells BOTH read empty"
        );
        drop(new);
    }

    /// A TRANSFER clears neither cell: the common key's pool follows the
    /// successor's ownership exactly as its facts do (W-S2, extended).
    #[test]
    fn a_transfer_leaves_the_common_keys_pool_published() {
        let f = fixture();
        let family = f.family();
        let shared = key(1, "nrpc:common");
        let old = family
            .demand_set(vec![shared.clone(), key(2, "nrpc:common")])
            .expect("acquired");
        f.registry.install_facts_for_test(shared.clone(), facts());
        let base = f.registry.base_facts_unvalidated(&shared).expect("facts");
        let pool = Arc::new(ScopedUnsensedRoutePool::for_test(base));
        f.registry
            .install_unsensed_pool_for_test(&shared, pool.clone());
        let index = old
            .keys()
            .iter()
            .position(|k| *k == shared)
            .expect("the set names the shared key");

        // The successor keeps the common key and sheds the other.
        let new = old.replace(vec![shared.clone()]).expect("replaced");
        assert!(
            old.unsensed_pool_unvalidated(index)
                .is_some_and(|read| Arc::ptr_eq(&read, &pool)),
            "the transfer cleared neither plane — the scope is still retained, \
             by the successor, and the pool still reads"
        );

        drop(new);
        assert!(
            old.unsensed_pool_unvalidated(index).is_none(),
            "and both planes go dark when the successor releases the last \
             reference"
        );
    }

    /// FACTS invalidation clears the derived pool with the facts (§1.1): the
    /// pool's basis is gone, so the pool is gone. Every facts-moving site
    /// routes through the same two-cell move, pool first, so this pins them
    /// all through the shared helper.
    #[test]
    fn facts_invalidation_clears_the_derived_pool() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:pool");
        let _live = family.demand(k.clone()).expect("retained");
        f.registry.install_facts_for_test(k.clone(), facts());
        let base = f.registry.base_facts_unvalidated(&k).expect("facts");
        f.registry
            .install_unsensed_pool_for_test(&k, Arc::new(ScopedUnsensedRoutePool::for_test(base)));

        f.registry.invalidate_for_test(&k);
        assert!(f.registry.base_facts_unvalidated(&k).is_none());
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_none(),
            "the pool's basis is gone, so the pool is gone"
        );
    }

    /// A DELAYED reader invalidation that observed a SUPERSEDED artifact
    /// deletes nothing on either plane; observing the CURRENT artifact deletes
    /// both. The §1.1 delayed-invalidator rule, with the control adjacent —
    /// without it, "clear both unconditionally" passes the first half.
    #[test]
    fn a_stale_observation_invalidates_neither_plane() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:pool");
        let _live = family.demand(k.clone()).expect("retained");

        // The artifact an arbitrarily-delayed reader once observed…
        let superseded = facts();
        f.registry
            .install_facts_for_test(k.clone(), superseded.clone());
        // …and the newer artifact, with ITS pool, current now.
        f.registry.install_facts_for_test(k.clone(), facts());
        let current = f.registry.base_facts_unvalidated(&k).expect("facts");
        let pool = Arc::new(ScopedUnsensedRoutePool::for_test(current.clone()));
        f.registry.install_unsensed_pool_for_test(&k, pool.clone());

        f.registry.invalidate_if_stale(&k, &superseded);
        assert!(
            f.registry
                .base_facts_unvalidated(&k)
                .is_some_and(|live| Arc::ptr_eq(&live, &current)),
            "facts the delayed reader did not observe survive"
        );
        assert!(
            f.registry
                .unsensed_pool_unvalidated(&k)
                .is_some_and(|read| Arc::ptr_eq(&read, &pool)),
            "and so does the pool derived from them"
        );

        // The control: observing the CURRENT artifact clears both planes.
        f.registry.invalidate_if_stale(&k, &current);
        assert!(f.registry.base_facts_unvalidated(&k).is_none());
        assert!(f.registry.unsensed_pool_unvalidated(&k).is_none());
    }

    /// Installing NEWER facts leaves no pool published: the pool present at
    /// install time was derived from the SUPERSEDED facts, and it is cleared
    /// before the new facts land so no lock-free reader can pair the new
    /// facts with the old pool. Repopulating it is the actor build cycle's
    /// job (§18 step 2).
    #[test]
    fn installing_newer_facts_clears_the_superseded_pool() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:pool");
        let _live = family.demand(k.clone()).expect("retained");
        f.registry.install_facts_for_test(k.clone(), facts());
        let base = f.registry.base_facts_unvalidated(&k).expect("facts");
        f.registry
            .install_unsensed_pool_for_test(&k, Arc::new(ScopedUnsensedRoutePool::for_test(base)));

        f.registry.install_facts_for_test(k.clone(), facts());
        assert!(
            f.registry.base_facts_unvalidated(&k).is_some(),
            "the newer facts are published"
        );
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_none(),
            "and the superseded facts' pool is not beside them"
        );
    }

    /// The SUCCESSOR of a replacement stays index-aligned across both planes:
    /// for every index, the successor's cells are the registry's cells for
    /// `successor.keys()[index]`, never another contributor's.
    ///
    /// Dies to: reversing the successor's collected cell vector
    /// (`cells.reverse()` in `replace_demand_set`) — a mutation the seven
    /// witnesses above all survive, because the acquisition-alignment witness
    /// never exercises replacement and the transfer witness reads through the
    /// OLD set. Misaligned, a future projection would read one contributor's
    /// facts and pool under another contributor's key. Found by the
    /// independent evidence audit of `3dbee8514`, where exactly that mutation
    /// ran 49/49 green.
    #[test]
    fn a_replacement_successor_stays_index_aligned_across_both_planes() {
        let f = fixture();
        let family = f.family();
        let a = key(1, "nrpc:align");
        let b = key(2, "nrpc:align");
        let c = key(3, "nrpc:align");
        let old = family
            .demand_set(vec![a.clone(), b.clone()])
            .expect("acquired");

        // B is COMMON and C is FRESH: the successor mixes a kept cell pair
        // with a newly created one, which is exactly where a collection-order
        // defect can hide.
        let successor = old.replace(vec![b.clone(), c.clone()]).expect("replaced");

        // Distinct artifacts per contributor, on BOTH planes — `facts()`
        // allocates a fresh `Arc` each call, so pointer identity
        // distinguishes them even with identical content.
        for k in successor.keys().to_vec() {
            f.registry.install_facts_for_test(k.clone(), facts());
            let base = f.registry.base_facts_unvalidated(&k).expect("facts");
            f.registry.install_unsensed_pool_for_test(
                &k,
                Arc::new(ScopedUnsensedRoutePool::for_test(base)),
            );
        }
        for (index, k) in successor.keys().to_vec().iter().enumerate() {
            let facts_via_set = successor.base_facts_unvalidated(index).expect("facts");
            let facts_via_registry = f.registry.base_facts_unvalidated(k).expect("facts");
            assert!(
                Arc::ptr_eq(&facts_via_set, &facts_via_registry),
                "successor contributor {index} reads its OWN key's facts"
            );
            let pool_via_set = successor.unsensed_pool_unvalidated(index).expect("pool");
            let pool_via_registry = f.registry.unsensed_pool_unvalidated(k).expect("pool");
            assert!(
                Arc::ptr_eq(&pool_via_set, &pool_via_registry),
                "and its OWN key's pool — the pairing survives a replacement"
            );
        }
        drop(old);
    }

    // ------------------- OLB-2B.3c §18 step 2: the actor build cycle

    /// **Production pool construction.** A real actor pass installs facts AND
    /// the pool derived from them, and the pool carries every component §1
    /// requires: the exact provider, its PROVEN owner org, the discovery
    /// provenance of the scope, the exact scoped source vector, the session
    /// annotation, the coherent generation, and the scoped deadline.
    ///
    /// The facts identity is checked by POINTER, not by content: that is what
    /// a 2B.3d reader has to be able to do, and two equal reconstructions
    /// would satisfy a content comparison.
    #[test]
    fn an_actor_pass_publishes_the_pool_it_derived() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:build");
        *f.source.rows.lock() = vec![provider_row(7, 42, 9_000), provider_row(8, 43, 4_000)];
        let generation = f.source.sessions.publish(
            [(entity(7), direct(0xabc))]
                .into_iter()
                .collect::<BTreeMap<_, _>>(),
        );
        let handle = family.demand(k.clone()).expect("retained");

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(matches!(outcome, ApplyOutcome::Current { .. }));

        let facts = f.registry.base_facts_unvalidated(&k).expect("facts");
        let pool = f.registry.unsensed_pool_unvalidated(&k).expect("pool");
        assert!(
            pool.derives_from(&facts),
            "the pool names the EXACT facts artifact published beside it"
        );
        assert_eq!(pool.key(), &k, "and its own authority-scoped identity");
        assert_eq!(pool.authority(), &facts.authority);
        assert_eq!(
            pool.epoch(),
            facts.epoch,
            "the exact scoped source vector, not a re-derived one"
        );
        assert_eq!(pool.provenance(), ProviderProvenance::OwnerPlane);
        assert_eq!(pool.session_generation(), generation);
        assert_eq!(
            pool.earliest_deadline(),
            4_000,
            "the earliest provider expiry bounds the pool exactly as it bounds \
             the facts"
        );
        assert_eq!(
            f.registry.next_artifact_deadline(),
            Some(4_000),
            "and PRIVATE-DISCOVERY expiry ARMS the actor on it: the pool's \
             bound is the artifact's, so one arm covers both planes"
        );

        let rows = pool.providers();
        assert_eq!(rows.len(), 2, "the whole scoped provider vector");
        assert_eq!(rows[0].provider, entity(7));
        assert_eq!(
            rows[0].owner_org,
            OrgId::from_bytes([7u8.wrapping_add(0x40); 32]),
            "the owner relation the ingest path PROVED, carried through"
        );
        assert_eq!(rows[0].generation, 42, "per-row announcement provenance");
        assert_eq!(rows[0].expires_at, 9_000);
        assert_eq!(
            rows[0].direct,
            direct(0xabc),
            "annotated under the ONE captured session view"
        );
        assert_eq!(
            rows[1].direct,
            DirectEligibility::Cold,
            "a provider the projection does not resolve is COLD, never guessed"
        );

        assert!(
            handle
                .unsensed_pool_unvalidated()
                .is_some_and(|read| Arc::ptr_eq(&read, &pool)),
            "and the lock-free handle reads the very pool the actor published"
        );
        assert_eq!(f.metrics.pools_published(), 1);
        assert_eq!(f.metrics.pools_refused_stale_session(), 0);
    }

    /// `Unserved` publishes NO pool; `Served(0 providers)` publishes one.
    ///
    /// The control is the whole witness: a pool over `Unserved` would be an
    /// artifact claiming an empty provider vector for a scope the source has
    /// no evidence about, which is exactly the conflation [`SourceFacts`]
    /// exists to refuse. The zero-provider SERVED case must still publish, or
    /// W-G13's empty-bucket Grant scope would have no pool to retire.
    #[test]
    fn an_unserved_reconstruction_publishes_no_pool_but_a_served_empty_one_does() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:unserved");
        f.source.sessions.publish(BTreeMap::new());
        let _held = family.demand(k.clone()).expect("retained");

        f.source.unserved.store(true, Ordering::Release);
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            f.registry.base_facts_unvalidated(&k).is_some(),
            "the Unserved artifact IS published — it is exact structural \
             evidence that the source cannot speak for this scope"
        );
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_none(),
            "but no pool is published over it"
        );

        // The control: exact fresh negative evidence DOES get a pool.
        f.source.unserved.store(false, Ordering::Release);
        f.registry.invalidate_for_test(&k);
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        let pool = f
            .registry
            .unsensed_pool_unvalidated(&k)
            .expect("a served empty bucket is exact evidence and gets a pool");
        assert!(pool.providers().is_empty());
    }

    /// **Publish-if-current, session half.** The session view moves DURING the
    /// build: the facts install (nothing about session movement makes a
    /// discovery row stale), the pool does not, the live slot is re-queued,
    /// and the stale pool is never observable.
    ///
    /// Dies to: publishing the pool unconditionally, or gating it on a
    /// generation read BEFORE the registry lock.
    #[test]
    fn a_session_view_that_moved_during_the_build_publishes_no_pool() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:race");
        *f.source.rows.lock() = vec![provider_row(7, 1, u64::MAX)];
        let captured = f
            .source
            .sessions
            .publish([(entity(7), direct(1))].into_iter().collect());
        let _held = family.demand(k.clone()).expect("retained");

        // The projection moves after the observation is captured and before
        // the publication phase — the exact window publish-if-current exists
        // for.
        let sessions = f.source.clone();
        *f.source.during_build.lock() = Some(Box::new(move || {
            sessions
                .sessions
                .publish([(entity(7), direct(2))].into_iter().collect());
        }));

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            matches!(outcome, ApplyOutcome::Progress { .. }),
            "the pass made progress and still owes the pool"
        );
        assert!(
            f.registry.base_facts_unvalidated(&k).is_some(),
            "the facts are CURRENT — the commit pin proved it, and sessions \
             say nothing about discovery"
        );
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_none(),
            "the pool built under the superseded view is never observable"
        );
        assert_eq!(f.metrics.pools_published(), 0);
        assert_eq!(f.metrics.pools_refused_stale_session(), 1);
        assert_eq!(
            f.registry.pending_slots(),
            1,
            "the live slot is re-queued, so the next pass rebuilds the pool"
        );

        // …and it converges: one more pass publishes the pool at the CURRENT
        // generation. Without this the witness would accept a plane that
        // refuses forever.
        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(matches!(outcome, ApplyOutcome::Current { .. }));
        let pool = f.registry.unsensed_pool_unvalidated(&k).expect("pool");
        assert!(pool.session_generation() > captured);
        assert_eq!(pool.providers()[0].direct, direct(2));
    }

    /// **Publish-if-current, source half.** Source movement during the build
    /// discards BOTH planes — there is nothing current to publish either half
    /// of — and re-queues the live slot.
    #[test]
    fn a_source_that_moved_during_the_build_publishes_neither_plane() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:srcrace");
        f.source
            .sessions
            .publish([(entity(7), direct(1))].into_iter().collect());
        *f.source.rows.lock() = vec![provider_row(7, 1, u64::MAX)];
        let _held = family.demand(k.clone()).expect("retained");

        let source = f.source.clone();
        *f.source.during_build.lock() = Some(Box::new(move || source.advance()));

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(matches!(outcome, ApplyOutcome::Superseded));
        assert!(f.registry.base_facts_unvalidated(&k).is_none());
        assert!(f.registry.unsensed_pool_unvalidated(&k).is_none());
        assert_eq!(f.metrics.pools_published(), 0);
        assert_eq!(f.registry.pending_slots(), 1);
    }

    /// **Mixed-generation refusal.** A quantum captures EXACTLY ONE session
    /// observation, so no pool can be assembled from two views — and a pass
    /// whose observation is superseded publishes nothing rather than accepting
    /// the mixed one.
    ///
    /// The projection moves between the FIRST and SECOND provider's
    /// annotation. Every row still carries the captured view (a per-row
    /// re-sample would show `direct(2)` on the second row), and the whole pool
    /// is then refused.
    ///
    /// Dies to: re-observing the projection per row or per slot instead of
    /// once per quantum.
    #[test]
    fn a_pool_cannot_compose_two_session_generations() {
        let f = fixture();
        let family = f.family();
        let a = key(1, "nrpc:mixed");
        let b = key(2, "nrpc:mixed");
        *f.source.rows.lock() = vec![provider_row(7, 1, u64::MAX)];
        f.source
            .sessions
            .publish([(entity(7), direct(1))].into_iter().collect());
        let _held = family
            .demand_set(vec![a.clone(), b.clone()])
            .expect("retained");

        let moved = f.source.clone();
        *f.source.during_build.lock() = Some(Box::new(move || {
            moved
                .sessions
                .publish([(entity(7), direct(2))].into_iter().collect());
        }));

        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(
            f.source.sessions.observations.load(Ordering::Acquire),
            1,
            "ONE observation for the whole quantum — there is no second sample \
             for a row to be annotated from"
        );
        assert_eq!(
            f.metrics.pools_refused_stale_session(),
            2,
            "and both slots refuse rather than publishing a mixed observation"
        );
        assert!(f.registry.unsensed_pool_unvalidated(&a).is_none());
        assert!(f.registry.unsensed_pool_unvalidated(&b).is_none());
    }

    /// **Pool-only invalidation.** Session movement clears the pool, PRESERVES
    /// the facts, and re-queues the slot.
    ///
    /// Dies to: clearing both planes (which would re-queue a full source
    /// reconstruction on every peer transition), or clearing neither.
    #[test]
    fn session_movement_clears_the_pool_and_preserves_the_facts() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:sessmove");
        f.source.sessions.publish(BTreeMap::new());
        let _held = family.demand(k.clone()).expect("retained");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        let facts = f.registry.base_facts_unvalidated(&k).expect("facts");
        assert!(f.registry.unsensed_pool_unvalidated(&k).is_some());

        let live = f.source.sessions.publish(BTreeMap::new());
        assert_eq!(f.registry.invalidate_session_older_than(live), 1);
        assert!(
            f.registry
                .base_facts_unvalidated(&k)
                .is_some_and(|live| Arc::ptr_eq(&live, &facts)),
            "the discovery facts are exactly as current as they were"
        );
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_none(),
            "and only the annotation is gone"
        );
        assert_eq!(f.registry.pending_slots(), 1);
        assert_eq!(f.metrics.pools_invalidated(), 1);
    }

    /// **The delayed pool invalidator.** An obsolete session transition can
    /// delete neither a pool from a LATER view, nor a pool from its OWN view,
    /// nor a pool derived from newer facts.
    ///
    /// The equality arm is separate on purpose: `<=` would have every
    /// transition destroy the pool its own publication produced, which is the
    /// self-inflicted churn `invalidate_grant_scope`'s fence already had to be
    /// corrected for once.
    #[test]
    fn an_obsolete_session_invalidator_cannot_delete_a_newer_pool() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:delayed");
        let obsolete = f.source.sessions.publish(BTreeMap::new());
        let _held = family.demand(k.clone()).expect("retained");

        // A pool built under a LATER view than the parked transition names.
        let newer = f.source.sessions.publish(BTreeMap::new());
        assert!(newer > obsolete);
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        let pool = f.registry.unsensed_pool_unvalidated(&k).expect("pool");
        assert_eq!(pool.session_generation(), newer);

        assert_eq!(
            f.registry.invalidate_session_older_than(obsolete),
            0,
            "the obsolete transition retires nothing"
        );
        assert!(
            f.registry
                .unsensed_pool_unvalidated(&k)
                .is_some_and(|read| Arc::ptr_eq(&read, &pool)),
            "the newer pool survives it"
        );

        // The EQUALITY arm: a transition at the pool's OWN generation is the
        // "a demand arriving after publication is safe" case.
        assert_eq!(f.registry.invalidate_session_older_than(newer), 0);
        assert!(f
            .registry
            .unsensed_pool_unvalidated(&k)
            .is_some_and(|read| Arc::ptr_eq(&read, &pool)));

        // The control, adjacent: a genuinely newer transition DOES retire it.
        let live = f.source.sessions.publish(BTreeMap::new());
        assert_eq!(f.registry.invalidate_session_older_than(live), 1);
        assert!(f.registry.unsensed_pool_unvalidated(&k).is_none());
    }

    /// A retained slot with NO published pool is left entirely alone by
    /// session movement — no clear, and no re-queue.
    ///
    /// Dies to: re-queueing on absence. That mutation satisfies every other
    /// pool witness perfectly and makes each peer transition re-queue every
    /// retained slot on the node — the breadth failure W-W3 exists to refuse,
    /// arriving on the session plane.
    #[test]
    fn session_movement_does_not_requeue_a_slot_with_no_pool() {
        let f = fixture();
        let family = f.family();
        let cold = key(1, "nrpc:nopool");
        let warm = key(2, "nrpc:nopool");
        f.source.sessions.publish(BTreeMap::new());
        let _held = family
            .demand_set(vec![cold.clone(), warm.clone()])
            .expect("retained");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        // Strip ONE slot's pool, leaving its facts — the state a slot is in
        // between a refused publication and its rebuild.
        f.registry.take_pool_for_test(&cold);
        assert_eq!(f.registry.pending_slots(), 0, "precondition");

        let live = f.source.sessions.publish(BTreeMap::new());
        assert_eq!(
            f.registry.invalidate_session_older_than(live),
            1,
            "only the slot that HAD a superseded pool is retired"
        );
        assert_eq!(
            f.registry.pending_slots(),
            1,
            "and only that slot is re-queued — a pool-less slot's rebuild is \
             already owed or already in flight"
        );
    }

    /// A terminally exhausted session-generation space publishes NO pool, and
    /// does not spin doing it.
    ///
    /// Fail-closed in the direction that matters: a wrapped or saturated
    /// generation would let a pool built under one view compare current
    /// against an unrelated later one. Facts still install — discovery is
    /// unaffected — so the pass settles and the actor parks.
    #[test]
    fn an_exhausted_session_generation_publishes_no_pool_without_spinning() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:spent");
        f.source.sessions.publish(BTreeMap::new());
        let _held = family.demand(k.clone()).expect("retained");
        f.source.sessions.currentness.exhaust();

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            matches!(outcome, ApplyOutcome::Current { .. }),
            "the pass is COMPLETE: it owes no pool it could ever build"
        );
        assert!(f.registry.base_facts_unvalidated(&k).is_some());
        assert!(f.registry.unsensed_pool_unvalidated(&k).is_none());
        assert_eq!(
            f.registry.pending_slots(),
            0,
            "nothing is re-queued, so nothing spins"
        );
        assert_eq!(
            f.metrics.pools_refused_stale_session(),
            0,
            "and it is not counted as a refused RACE — nothing raced"
        );
    }

    /// **The step-1 publication-order debt, paid.** A lock-free reader
    /// scheduled BETWEEN the two adjacent publication stores can never observe
    /// a pool beside facts it was not derived from.
    ///
    /// Final-state assertions cannot see this — both stores complete before
    /// any of them runs (design §18.0) — so the reader is placed inside the
    /// gaps by the production publication path itself.
    ///
    /// Dies to: swapping the two stores, which makes the first gap expose the
    /// SUPERSEDED pool beside the new facts.
    #[test]
    fn no_reader_can_observe_a_pool_beside_foreign_facts() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:interleave");
        f.source.sessions.publish(BTreeMap::new());
        let handle = family.demand(k.clone()).expect("retained");

        // A first publication, so a SUPERSEDED pool is present when the second
        // one starts. Without it every gap is trivially pool-free.
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_some(),
            "precondition: a superseded pool must be published when the second \
             installation begins, or every gap is trivially pool-free"
        );

        let observed: Arc<parking_lot::Mutex<Vec<(PublicationGap, bool, bool)>>> =
            Arc::new(parking_lot::Mutex::new(Vec::new()));
        {
            let observed = observed.clone();
            let handle_facts = handle.cells.facts.clone();
            let handle_pool = handle.cells.unsensed.clone();
            f.registry
                .observe_publication_for_test(Arc::new(move |gap| {
                    // The lock-free reader: two separate atomics, exactly as a
                    // 2B.3d consumer would load them.
                    let facts = handle_facts.load_full();
                    let pool = handle_pool.load_full();
                    let paired = match (&facts, &pool) {
                        (Some(facts), Some(pool)) => pool.derives_from(facts),
                        // No pool is the ordinary cold-pool state; facts absent
                        // with a pool present is a mismatch by construction.
                        (_, None) => true,
                        (None, Some(_)) => false,
                    };
                    observed.lock().push((gap, pool.is_some(), paired));
                }));
        }

        // Re-queued WITHOUT invalidating, so the superseded pool is still
        // published when the next installation begins.
        f.registry.requeue_for_test(&k);
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));

        let observed = observed.lock().clone();
        assert_eq!(
            observed.len(),
            2,
            "both gaps must be observed — a witness that samples one proves \
             half the invariant"
        );
        assert!(
            observed.iter().all(|(_, _, paired)| *paired),
            "no reader observed a pool beside facts it was not derived from: \
             {observed:?}"
        );
        assert!(
            observed.iter().all(|(_, pool_present, _)| !*pool_present),
            "and every intermediate state is a COLD-POOL state, which is the \
             state a reader already has to handle: {observed:?}"
        );
        let final_pool = f.registry.unsensed_pool_unvalidated(&k).expect("pool");
        let final_facts = f.registry.base_facts_unvalidated(&k).expect("facts");
        assert!(final_pool.derives_from(&final_facts));
    }

    /// A pool's deadline retires BOTH planes and re-arms the actor, and the
    /// rebuild converges rather than spinning.
    ///
    /// The ZERO-provider case is the one that matters: with no rows, the
    /// artifact's only deadline is its installed authority's, so a rows-only
    /// bound would leave the pool published under dead authority forever
    /// (W-G13, extended to the second cell).
    #[test]
    fn an_authority_deadline_with_zero_providers_retires_both_planes_and_rearms() {
        let f = fixture();
        let family = f.family();
        let k = key(1, "nrpc:deadline");
        f.source.sessions.publish(BTreeMap::new());
        f.source.authority_deadline.store(1_000, Ordering::Release);
        let _held = family.demand(k.clone()).expect("retained");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));

        let pool = f.registry.unsensed_pool_unvalidated(&k).expect("pool");
        assert!(
            pool.providers().is_empty(),
            "zero rows: the authority deadline is the only one there is"
        );
        assert_eq!(pool.earliest_deadline(), 1_000);
        assert_eq!(
            f.registry.next_artifact_deadline(),
            Some(1_000),
            "and the actor ARMS on it"
        );

        assert_eq!(f.registry.retire_expired(1_000), 1);
        assert!(f.registry.base_facts_unvalidated(&k).is_none());
        assert!(
            f.registry.unsensed_pool_unvalidated(&k).is_none(),
            "the pool goes with the facts it derived from"
        );
        assert_eq!(f.registry.pending_slots(), 1, "the rebuild is armed");

        // No spin: the rebuild against dead authority installs `Unserved`,
        // which carries no deadline and no pool, so the next arm finds nothing.
        f.source.unserved.store(true, Ordering::Release);
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(f.registry.next_artifact_deadline(), None);
        assert_eq!(f.registry.pending_slots(), 0);
    }

    /// The 256 bound is STRUCTURAL on the pool plane too: pool ownership is
    /// physically attached to the source slot, so a refused 257th slot owns no
    /// cell and can publish no pool.
    #[test]
    fn the_two_hundred_fifty_seventh_slot_can_publish_no_pool() {
        let f = fixture();
        f.source.sessions.publish(BTreeMap::new());
        let mut held = Vec::new();
        for index in 0..MAX_NODE_SLOTS {
            let family = f.family();
            let key = key(1, &format!("nrpc:cap{index}"));
            held.push(family.demand(key).expect("within the node bound"));
        }
        assert_eq!(f.registry.retained_slots(), MAX_NODE_SLOTS);

        let refused = key(1, "nrpc:cap256");
        let overflow = f.family();
        assert_eq!(
            overflow.demand(refused.clone()).err(),
            Some(DemandRefused::NodeAtCapacity)
        );

        // Rebuild everything the node DOES retain, in as many quanta as it
        // takes, then count the pools.
        for _ in 0..16 {
            if f.registry.pending_slots() == 0 {
                break;
            }
            f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        }
        assert_eq!(f.metrics.pools_published(), MAX_NODE_SLOTS as u64);
        assert!(
            f.registry.unsensed_pool_unvalidated(&refused).is_none(),
            "the refused slot owns no cell, so no pool can exist for it"
        );
        assert_eq!(f.registry.retained_slots(), MAX_NODE_SLOTS);
    }

    /// `release_one` cannot decrement a global slot reference for a family that
    /// holds none — the bookkeeping invariant, asserted directly.
    ///
    /// Dies to: dropping the `NotHeld` guard. Without it, a caller that names the
    /// wrong family retires a slot some OTHER family still owns, stranding a handle
    /// in that family's map that its set can never give back.
    #[test]
    fn release_one_cannot_decrement_a_reference_a_family_does_not_hold() {
        let f = fixture();
        let owner = f.family();
        let stranger = f.family();
        let owned = key(1, "nrpc:owned");
        let held = owner.demand_set(vec![owned.clone()]).expect("acquired");

        let before = f.registry.retained_slots();
        let generation = f.registry.node_capacity_generation();
        f.registry.release_one_for_test(&stranger, &owned);

        assert_eq!(
            f.registry.retained_slots(),
            before,
            "a family that holds nothing gives nothing back"
        );
        assert_eq!(f.registry.node_capacity_generation(), generation);
        assert_eq!(owner.handles(), 1, "the owner still holds its handle");
        assert!(
            held.base_facts_unvalidated(0).is_some() || f.registry.retained_slots() == 1,
            "and its slot survives"
        );
        drop(held);
        assert_eq!(
            f.registry.retained_slots(),
            0,
            "the owner can still release"
        );
    }

    /// A set whose responsibility has already transferred cannot be replaced again.
    ///
    /// Dies to: dropping the `held != keys` guard. A second replacement through the
    /// spent set computes `old_only = ∅` and charges the successor's whole footprint
    /// gross — the HOLD-2 defect, reached through a stale basis instead of a stale
    /// accounting rule.
    #[test]
    fn an_already_transferred_set_cannot_be_replaced_again() {
        let f = fixture();
        let family = f.family();
        let old = family
            .demand_set(vec![key(1, "nrpc:once")])
            .expect("acquired");
        let successor = old
            .replace(vec![key(2, "nrpc:once")])
            .expect("first replacement");

        let handles = family.handles();
        let slots = f.registry.retained_slots();
        assert_eq!(
            old.replace(vec![key(3, "nrpc:once")]).err(),
            Some(ReplaceRefused::Superseded),
            "a spent set is not the basis of anything"
        );
        assert_eq!(family.handles(), handles, "and nothing moved");
        assert_eq!(f.registry.retained_slots(), slots);
        drop(successor);
    }

    /// W-A4 — duplicate keys collapse to one slot and one handle each.
    ///
    /// The single-key `demand` deliberately counts a repeat as a second handle
    /// (re-demanding must not bypass the bound). A demand SET is a set: the same
    /// scope named twice is one contributor, and charging it twice would refuse
    /// families for capacity they never needed.
    #[test]
    fn duplicate_keys_in_a_demand_set_collapse() {
        let f = fixture();
        let family = f.family();
        let handles = family
            .demand_set(vec![
                key(1, "nrpc:dup"),
                key(1, "nrpc:dup"),
                key(2, "nrpc:dup"),
                key(1, "nrpc:dup"),
            ])
            .expect("acquired");

        assert_eq!(handles.len(), 2, "four keys, two distinct scopes");
        assert_eq!(family.handles(), 2, "and two handles against the budget");
        assert_eq!(f.registry.retained_slots(), 2);
        assert_eq!(
            f.registry.allocated_ids_for_test(),
            3,
            "one identity for the family and one per distinct slot — not per key"
        );
    }

    /// W-A5 — a refusal reached AFTER the identity space was consulted still
    /// consumes no identity, and the set is ordered so a per-key loop provably
    /// would.
    ///
    /// The family holds 63 of 64 handles and asks for three NEW keys. A loop
    /// admits the first (63 < 64), minting an incarnation, and refuses the
    /// second. The prefix handle is then dropped, the slot retires, and every
    /// count returns to its starting value EXCEPT the identity high-water mark
    /// and the retirement counter — which is exactly why the assertion is total.
    #[test]
    fn a_refusal_after_identities_are_considered_consumes_none() {
        let f = fixture();
        let family = f.family();
        let mut held = Vec::new();
        for i in 0..63u32 {
            held.push(
                family
                    .demand(key(1, &format!("nrpc:fill-{i}")))
                    .expect("fill"),
            );
        }
        let before = no_effect(&f, &family);
        assert_eq!(before.handles, 63);

        assert_eq!(
            family.demand_set(set(&[40, 41, 42], "nrpc:late")).err(),
            Some(DemandRefused::FamilyAtCapacity)
        );
        let after = no_effect(&f, &family);
        assert_eq!(
            after.ids, before.ids,
            "the identity space is untouched — the component a loop leaks"
        );
        assert_eq!(
            after.retired, before.retired,
            "and nothing was created only to be retired again"
        );
        assert_eq!(after, before, "totally, not just in those two components");
    }

    /// W-A6 — a successful acquisition marks the actor ONCE and queues every new
    /// identity, so the set is warmed by one pass rather than one pass per key.
    #[test]
    fn a_successful_demand_set_queues_every_key_and_marks_once() {
        let f = fixture();
        let family = f.family();
        let handles = family
            .demand_set(set(&[50, 51, 52], "nrpc:batch"))
            .expect("acquired");
        assert_eq!(handles.len(), 3);
        assert_eq!(f.registry.pending_slots(), 3, "every key owes work");

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            matches!(outcome, ApplyOutcome::Current { .. }),
            "{outcome:?}"
        );
        for seed in [50u8, 51, 52] {
            assert!(
                f.registry
                    .base_facts_unvalidated(&key(seed, "nrpc:batch"))
                    .is_some(),
                "the whole set is warmed by ONE pass"
            );
        }
    }

    /// The demand set arrives in deterministic key order, Owner scopes first.
    ///
    /// §3.1 projects Owner pool first and then Grant pools; a handle vector in
    /// call order would make that ordering the caller's problem.
    #[test]
    fn a_demand_set_is_ordered_owner_scopes_first() {
        let f = fixture();
        let family = f.family();
        let grant = PrivateAudienceScope::new(CapabilityAudienceScope::Grant {
            grant_id: [9u8; 32],
            audience_handle: [9u8; 32],
        })
        .expect("grant scopes are private");
        let cap = CapabilityAuthorityId::for_tag("nrpc:order");
        let handles = family
            .demand_set(vec![
                SlotKey {
                    scope: grant.clone(),
                    capability: cap,
                },
                key(1, "nrpc:order"),
            ])
            .expect("acquired");

        assert_eq!(handles.keys()[0], key(1, "nrpc:order"), "Owner first");
        assert_eq!(handles.keys()[1].scope, grant, "Grant after");
    }

    // -------------------------------------------------------------- lifecycle

    /// Dropping one of several references preserves the slot; dropping the LAST
    /// retires it.
    #[test]
    fn only_the_last_reference_retires_the_slot() {
        let f = fixture();
        let a = f.family().demand(key(1, "nrpc:a")).expect("a");
        let b = f.family().demand(key(1, "nrpc:a")).expect("b");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(f
            .registry
            .base_facts_unvalidated(&key(1, "nrpc:a"))
            .is_some());

        drop(a);
        assert_eq!(f.registry.retained_slots(), 1, "one reference remains");
        assert!(f
            .registry
            .base_facts_unvalidated(&key(1, "nrpc:a"))
            .is_some());

        drop(b);
        assert_eq!(
            f.registry.retained_slots(),
            0,
            "the last reference retired it"
        );
        assert_eq!(f.metrics.slots_retired(), 1);
        assert!(f
            .registry
            .base_facts_unvalidated(&key(1, "nrpc:a"))
            .is_none());
    }

    /// A build in flight cannot resurrect a slot retired and re-demanded beneath
    /// it — driven through the REAL lifecycle: `DemandHandle::drop` retires the
    /// slot and a real family owner re-demands it, both from inside the source
    /// query. The stale artifact is discarded, the recreated slot stays indexed and
    /// pending, and the pass reports `Superseded` so its wake is owed.
    #[test]
    fn a_late_build_cannot_resurrect_a_replaced_slot_incarnation() {
        let f = fixture();
        let target = key(1, "nrpc:a");
        let original = f.family();
        let successor = f.family();

        let held: Arc<parking_lot::Mutex<Option<DemandHandle>>> = Arc::new(
            parking_lot::Mutex::new(Some(original.demand(target.clone()).expect("first demand"))),
        );
        let first_incarnation = f
            .registry
            .inner
            .lock()
            .slots
            .get(&target)
            .expect("retained")
            .incarnation;

        let recreated: Arc<parking_lot::Mutex<Option<DemandHandle>>> =
            Arc::new(parking_lot::Mutex::new(None));
        {
            let held = held.clone();
            let recreated = recreated.clone();
            let successor = successor.clone();
            let target = target.clone();
            *f.source.during_build.lock() = Some(Box::new(move || {
                // The real RAII release: last reference retires the slot.
                drop(held.lock().take());
                // A real family owner re-demands it — a FRESH incarnation.
                *recreated.lock() = Some(successor.demand(target.clone()).expect("re-demand"));
            }));
        }

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(
            outcome,
            ApplyOutcome::Superseded,
            "slot movement supersedes the attempt"
        );
        assert!(
            f.registry.base_facts_unvalidated(&target).is_none(),
            "the artifact for the dead incarnation was discarded, not installed"
        );
        assert_eq!(f.metrics.discarded_obsolete(), 1);
        assert_eq!(f.metrics.installs(), 0);
        assert_eq!(f.metrics.slots_retired(), 1);

        let inner = f.registry.inner.lock();
        let slot = inner.slots.get(&target).expect("recreated slot retained");
        assert_ne!(
            slot.incarnation, first_incarnation,
            "the recreated slot has a FRESH identity"
        );
        assert!(
            inner
                .slots_by_capability
                .get(&target.capability)
                .is_some_and(|bucket| bucket.contains(&target)),
            "and is still indexed by capability"
        );
        assert!(
            inner.pending.contains(&target),
            "the live incarnation still owes work, queued authoritatively"
        );
        drop(inner);
        assert!(recreated.lock().is_some());
    }

    /// A brand-new actor incarnation invalidates everything the previous one built
    /// and queues every retained slot.
    #[test]
    fn a_new_incarnation_invalidates_every_retained_slot() {
        let f = fixture();
        let family = f.family();
        let _a = family.demand(key(1, "nrpc:a")).expect("a");
        let _b = family.demand(key(1, "nrpc:b")).expect("b");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(f
            .registry
            .base_facts_unvalidated(&key(1, "nrpc:a"))
            .is_some());
        assert_eq!(f.registry.pending_slots(), 0);

        f.registry.deactivate_incarnation(1);
        f.registry.activate_incarnation(2);

        assert!(
            f.registry
                .base_facts_unvalidated(&key(1, "nrpc:a"))
                .is_none(),
            "nothing a dead incarnation built stays readable"
        );
        assert_eq!(
            f.registry.pending_slots(),
            2,
            "every retained slot re-queued"
        );
    }

    // ---------------------------------------------------------- authority

    /// An application from an actor that is NOT the live one consumes no pending
    /// work, installs nothing, and never reports `Current`.
    #[test]
    fn a_stale_actor_neither_consumes_pending_work_nor_reports_current() {
        let f = fixture();
        let _held = f.family().demand(key(1, "nrpc:a")).expect("a");
        assert_eq!(f.registry.pending_slots(), 1);

        f.source.reset();
        let outcome = f
            .registry
            .apply(2, request(true, DirtyCapabilities::RebuildAll));
        assert_eq!(outcome, ApplyOutcome::Superseded);
        assert!(
            f.source.queries().is_empty(),
            "a stale actor never even queries the source"
        );
        assert_eq!(
            f.registry.pending_slots(),
            1,
            "the authoritative work is still owed"
        );
        assert_eq!(f.metrics.stale_actor_rejections(), 1);

        // With NO actor live, the same holds.
        f.registry.deactivate_incarnation(1);
        assert_eq!(
            f.registry.apply(1, request(true, DirtyCapabilities::Clean)),
            ApplyOutcome::Superseded
        );
        assert_eq!(f.registry.pending_slots(), 1);
        assert_eq!(f.metrics.stale_actor_rejections(), 2);

        // The live successor finds the work intact.
        f.registry.activate_incarnation(3);
        let outcome = f.registry.apply(3, request(true, DirtyCapabilities::Clean));
        assert!(matches!(outcome, ApplyOutcome::Current { .. }));
        assert_eq!(f.source.queries(), vec![key(1, "nrpc:a")]);
    }

    /// The EMPTY-SELECTION settlement path revalidates authority too.
    ///
    /// Nothing is installed on that path, which is exactly why it is easy to miss —
    /// but `Current` on a full request is what lets the supervisor publish
    /// `Healthy`, so an actor revoked after phase 1 must not reach it. The
    /// revocation lands during the off-lock snapshot, the window in which the
    /// phase-1 check goes stale (Kyra OLB-2B-E3b).
    #[test]
    fn an_actor_revoked_after_selection_cannot_settle_an_empty_pass_as_current() {
        // Baseline: with the actor live, this exact pass settles Current.
        let f = fixture();
        assert_eq!(
            f.registry
                .apply(1, request(false, DirtyCapabilities::RebuildAll)),
            ApplyOutcome::Current {
                source_generation: f.source.generation()
            },
            "a full request over no retained slots completes"
        );

        // The same pass, with the actor fenced inside the off-lock snapshot.
        let f = fixture();
        {
            let registry = f.registry.clone();
            *f.source.on_snapshot.lock() = Some(Box::new(move || {
                registry.deactivate_incarnation(1);
            }));
        }
        let outcome = f
            .registry
            .apply(1, request(false, DirtyCapabilities::RebuildAll));
        assert_eq!(
            outcome,
            ApplyOutcome::Superseded,
            "an actor revoked after phase 1 cannot settle Current"
        );
        assert_eq!(f.metrics.stale_actor_rejections(), 1);

        // And the same holds for an ordinary quiet registry-work pass.
        let f = fixture();
        let target = key(1, "nrpc:a");
        let _held = f.family().demand(target.clone()).expect("a");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(f.registry.pending_slots(), 0, "nothing left to select");
        {
            let registry = f.registry.clone();
            *f.source.on_snapshot.lock() = Some(Box::new(move || {
                registry.deactivate_incarnation(1);
            }));
        }
        assert_eq!(
            f.registry.apply(1, request(true, DirtyCapabilities::Clean)),
            ApplyOutcome::Superseded
        );
        assert_eq!(f.metrics.stale_actor_rejections(), 1);
    }

    /// Authority revoked DURING the build: every still-live selected slot is
    /// requeued, nothing is installed, and the outcome is never `Current`.
    #[test]
    fn authority_revoked_during_a_build_requeues_every_live_slot() {
        let f = fixture();
        let family = f.family();
        let a = key(1, "nrpc:a");
        let b = key(1, "nrpc:b");
        let _ha = family.demand(a.clone()).expect("a");
        let hb = family.demand(b.clone()).expect("b");

        let registry = f.registry.clone();
        let hb_cell: Arc<parking_lot::Mutex<Option<DemandHandle>>> =
            Arc::new(parking_lot::Mutex::new(Some(hb)));
        {
            let hb_cell = hb_cell.clone();
            *f.source.during_build.lock() = Some(Box::new(move || {
                // The actor's fence fires mid-build, and one slot also retires —
                // that one must NOT be requeued, because it no longer exists.
                registry.deactivate_incarnation(1);
                drop(hb_cell.lock().take());
            }));
        }

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(outcome, ApplyOutcome::Superseded);
        assert_eq!(
            f.metrics.installs(),
            0,
            "nothing from a revoked actor lands"
        );
        assert_eq!(f.metrics.stale_actor_rejections(), 1);

        let inner = f.registry.inner.lock();
        assert!(inner.pending.contains(&a), "the live slot still owes work");
        assert!(
            !inner.pending.contains(&b),
            "a retired slot is not resurrected into the queue"
        );
    }

    // ------------------------------------------------------------ invalidation

    /// A `Caps` delta invalidates the affected slot's facts UNDER THE PHASE-1 LOCK,
    /// before any rebuild begins — asserted from inside the source query, while the
    /// rebuild is still in flight. An unrelated slot keeps its facts.
    #[test]
    fn a_caps_delta_invalidates_affected_facts_before_the_rebuild_begins() {
        let f = fixture();
        let family = f.family();
        let c = key(1, "nrpc:c");
        let d = key(1, "nrpc:d");
        let _hc = family.demand(c.clone()).expect("c");
        let _hd = family.demand(d.clone()).expect("d");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(f.registry.base_facts_unvalidated(&c).is_some());
        assert!(f.registry.base_facts_unvalidated(&d).is_some());

        let observed = Arc::new(AtomicBool::new(false));
        {
            let registry = f.registry.clone();
            let observed = observed.clone();
            let c = c.clone();
            let d = d.clone();
            *f.source.during_build.lock() = Some(Box::new(move || {
                assert!(
                    registry.base_facts_unvalidated(&c).is_none(),
                    "C's stale facts must already be gone while C rebuilds"
                );
                assert!(
                    registry.base_facts_unvalidated(&d).is_some(),
                    "D was not named by the delta and keeps its facts"
                );
                observed.store(true, Ordering::Release);
            }));
        }

        f.source.reset();
        f.registry.apply(1, request(false, caps(&["nrpc:c"])));
        assert!(observed.load(Ordering::Acquire), "the hook must have run");
        assert_eq!(f.source.queries(), vec![c.clone()]);
        assert!(
            f.registry.base_facts_unvalidated(&c).is_some(),
            "and is rebuilt after"
        );
    }

    /// Caps invalidation covers slots BEYOND the quantum: a slot that will not be
    /// rebuilt for several passes must not keep serving stale facts in the meantime.
    #[test]
    fn caps_invalidates_affected_slots_beyond_the_quantum() {
        let f = fixture();
        let mut family = f.family();
        let mut keys = Vec::new();
        let mut held = Vec::new();
        for index in 0..(APPLY_QUANTUM + 8) {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            let k = key(1, &format!("nrpc:q{index}"));
            held.push(family.demand(k.clone()).expect("demanded"));
            keys.push(k);
        }
        // Two passes to build them all.
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(keys
            .iter()
            .all(|k| f.registry.base_facts_unvalidated(k).is_some()));

        let tags: Vec<String> = (0..(APPLY_QUANTUM + 8))
            .map(|index| format!("nrpc:q{index}"))
            .collect();
        let all: Vec<&str> = tags.iter().map(String::as_str).collect();

        f.source.reset();
        let outcome = f.registry.apply(1, request(false, caps(&all)));
        assert_eq!(
            f.source.queries().len(),
            APPLY_QUANTUM,
            "only one quantum is rebuilt"
        );
        assert!(
            matches!(outcome, ApplyOutcome::Progress { .. }),
            "work remains, so this is not a complete installation"
        );
        assert_eq!(
            keys.iter()
                .filter(|k| f.registry.base_facts_unvalidated(k).is_some())
                .count(),
            APPLY_QUANTUM,
            "EVERY affected slot was invalidated; only the quantum was rebuilt"
        );
        assert_eq!(f.registry.pending_slots(), 8);
    }

    /// review-pass-2 §2 (the >`APPLY_QUANTUM` witness its disposition required) —
    /// a multi-quantum recapture under SUSTAINED source movement loses no
    /// identity, and converges the moment the source settles.
    ///
    /// With more than one quantum's worth of retained slots, a complete recapture
    /// needs ceil(N/64) CONSECUTIVE quanta at one epoch, and a single ingest,
    /// expiry sweep or floor raise between them re-queues what was already built.
    /// The failure this must exclude is not the retrying — that is the design —
    /// but SILENT LOSS: an identity dropped by a refused pass would leave its slot
    /// cold with nothing owed to rebuild it, and the recapture would report
    /// completion over a hole. The actor-side rate control for the same scenario
    /// is `a_sustained_superseded_streak_backs_off_and_reports_degraded`.
    #[test]
    fn sustained_source_movement_across_a_multi_quantum_recapture_loses_no_identity() {
        let f = fixture();
        let mut family = f.family();
        let mut held = Vec::new();
        let mut keys = Vec::new();
        let total = APPLY_QUANTUM + 1;
        for index in 0..total {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            let slot = key(1, &format!("nrpc:m{index:03}"));
            held.push(family.demand(slot.clone()).expect("demanded"));
            keys.push(slot);
        }
        assert_eq!(f.registry.pending_slots(), total);

        // Sustained movement: every pass has the source advance mid-build, so
        // every commit pin refuses.
        for pass in 0..6 {
            let state = f.source.clone();
            *f.source.during_build.lock() = Some(Box::new(move || state.advance()));
            assert_eq!(
                f.registry
                    .apply(1, request(true, DirtyCapabilities::RebuildAll)),
                ApplyOutcome::Superseded,
                "pass {pass}: a pin that refused installs nothing"
            );
            assert_eq!(
                f.registry.pending_slots(),
                total,
                "pass {pass}: EVERY identity stays owed — a dropped one would leave \
                 its slot cold with nothing to rebuild it"
            );
            assert!(
                keys.iter()
                    .all(|k| f.registry.base_facts_unvalidated(k).is_none()),
                "pass {pass}: and nothing obsolete was installed"
            );
        }

        // The source settles: the preserved queue is what lets the recapture
        // finish, across as many quanta as it takes.
        let mut outcome = ApplyOutcome::Superseded;
        for _ in 0..8 {
            outcome = f
                .registry
                .apply(1, request(true, DirtyCapabilities::RebuildAll));
            if matches!(outcome, ApplyOutcome::Current { .. }) {
                break;
            }
        }
        assert!(
            matches!(outcome, ApplyOutcome::Current { .. }),
            "a settled source converges (last outcome {outcome:?})"
        );
        assert_eq!(f.registry.pending_slots(), 0);
        assert!(
            keys.iter()
                .all(|k| f.registry.base_facts_unvalidated(k).is_some()),
            "and every slot the recapture covered is warm — no hole under a Current"
        );
    }

    /// review-pass-3 §9 — sustained low-sorting churn must not starve a
    /// high-sorting pending slot.
    ///
    /// Selection used to take the first `APPLY_QUANTUM` in `(scope, capability)`
    /// order on EVERY pass, with no cursor, FIFO or aging. So if each pass
    /// re-dirties a full quantum of slots whose keys sort below a victim's, the
    /// victim is re-outsorted every pass and never rebuilds — cold reads and
    /// permanent `pending` occupancy for as long as the churn lasts. This is not
    /// review-pass-2 §2's whole-recapture starvation: every pass here settles,
    /// and the loss is confined to a subset.
    #[test]
    fn sustained_low_sorting_churn_cannot_starve_a_high_sorting_slot() {
        let f = fixture();
        let mut family = f.family();
        let mut held = Vec::new();
        let mut entries: Vec<(SlotKey, String)> = Vec::new();
        for index in 0..=APPLY_QUANTUM {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            let tag = format!("nrpc:q{index:03}");
            let slot = key(1, &tag);
            held.push(family.demand(slot.clone()).expect("demanded"));
            entries.push((slot, tag));
        }
        // Capability ids are hashes, so SELECTION order is not tag order — pick
        // the victim by the order the registry actually sorts in.
        entries.sort_by(|a, b| a.0.cmp(&b.0));
        let victim = entries.last().expect("victim").0.clone();
        let churning: Vec<&str> = entries[..APPLY_QUANTUM]
            .iter()
            .map(|(_, tag)| tag.as_str())
            .collect();

        // Warm everything, then cold the victim so it is the one identity owed
        // besides the churn.
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(f.registry.base_facts_unvalidated(&victim).is_some());
        f.registry.invalidate_for_test(&victim);

        let mut served_on = None;
        for pass in 0..4 {
            f.source.reset();
            f.registry.apply(1, request(true, caps(&churning)));
            assert_eq!(
                f.source.queries().len(),
                APPLY_QUANTUM,
                "pass {pass}: the quantum is saturated, so this is genuine contention \
                 rather than a pass with room to spare"
            );
            if f.registry.base_facts_unvalidated(&victim).is_some() {
                served_on = Some(pass);
                break;
            }
        }
        assert!(
            served_on.is_some(),
            "the highest-sorting slot must come round: without rotation the churn \
             below it wins every pass, forever"
        );
    }

    /// `Caps(C)` touches only C's indexed slots; `RebuildAll` touches every
    /// retained slot; an undemanded capability is never projected.
    #[test]
    fn caps_touches_only_its_bucket_and_rebuild_all_touches_every_slot() {
        let f = fixture();
        let family = f.family();
        let _a = family.demand(key(1, "nrpc:a")).expect("a");
        let _b = family.demand(key(1, "nrpc:b")).expect("b");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));

        f.source.reset();
        f.registry.apply(1, request(false, caps(&["nrpc:b"])));
        assert_eq!(f.source.queries(), vec![key(1, "nrpc:b")]);

        f.source.reset();
        f.registry
            .apply(1, request(false, DirtyCapabilities::RebuildAll));
        assert_eq!(f.source.queries().len(), 2, "every retained slot");

        f.source.reset();
        f.registry.apply(1, request(false, caps(&["nrpc:absent"])));
        assert!(
            f.source.queries().is_empty(),
            "an undemanded capability is never projected"
        );
    }

    /// A slot named by BOTH the source delta and the registry's pending set is
    /// reconstructed ONCE.
    #[test]
    fn combined_source_and_registry_work_deduplicates_a_slot() {
        let f = fixture();
        let target = key(1, "nrpc:a");
        let _held = f.family().demand(target.clone()).expect("a");
        assert_eq!(
            f.registry.pending_slots(),
            1,
            "registry work owes this slot"
        );

        f.source.reset();
        f.registry.apply(1, request(true, caps(&["nrpc:a"])));
        assert_eq!(
            f.source.queries(),
            vec![target],
            "named by both domains, built once"
        );
    }

    // ------------------------------------------------------------ completion

    /// A multi-quantum recapture reports `Progress` until the LAST quantum, keeps
    /// the epoch's remaining set authoritative, and only then reports `Current`
    /// with every retained slot coherently stamped.
    #[test]
    fn a_multi_quantum_recapture_reports_progress_until_it_completes() {
        let f = fixture();
        let mut family = f.family();
        let mut keys = Vec::new();
        let mut held = Vec::new();
        for index in 0..(APPLY_QUANTUM + 10) {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            let k = key(1, &format!("nrpc:q{index}"));
            held.push(family.demand(k.clone()).expect("demanded"));
            keys.push(k);
        }

        f.source.reset();
        let first = f
            .registry
            .apply(1, request(false, DirtyCapabilities::RebuildAll));
        assert_eq!(
            f.source.queries().len(),
            APPLY_QUANTUM,
            "at most one quantum per synchronous application"
        );
        assert!(
            matches!(first, ApplyOutcome::Progress { .. }),
            "an incomplete recapture must NOT claim a current installation"
        );
        assert_eq!(
            f.registry.pending_slots(),
            10,
            "the remainder lives in the registry, not in the wake flag"
        );

        // The actor promotes the next wake to RebuildAll while a recapture is
        // owed. That must CONTINUE the epoch, not re-expand it.
        f.source.reset();
        let second = f
            .registry
            .apply(1, request(true, DirtyCapabilities::RebuildAll));
        assert_eq!(
            f.source.queries().len(),
            10,
            "the epoch resumed where it left off"
        );
        assert!(
            matches!(second, ApplyOutcome::Current { .. }),
            "the final quantum completes the recapture"
        );
        assert_eq!(f.registry.pending_slots(), 0);

        let generation = f.source.generation();
        for k in &keys {
            let facts = f.registry.base_facts_unvalidated(k).expect("built");
            assert_eq!(facts.actor_incarnation, 1);
            assert_eq!(facts.epoch.generation, generation);
        }
        assert!(!f.registry.inner.lock().recapture_open, "the epoch closed");
    }

    /// A recapture whose source generation moved between quanta RESTARTS, so
    /// `Current` is never reported over a set built against two generations.
    #[test]
    fn a_recapture_restarts_when_the_source_generation_moves_between_quanta() {
        let f = fixture();
        let mut family = f.family();
        let mut held = Vec::new();
        for index in 0..(APPLY_QUANTUM + 4) {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            held.push(
                family
                    .demand(key(1, &format!("nrpc:q{index}")))
                    .expect("demanded"),
            );
        }

        let first = f
            .registry
            .apply(1, request(false, DirtyCapabilities::RebuildAll));
        assert!(matches!(first, ApplyOutcome::Progress { .. }));
        assert_eq!(f.registry.pending_slots(), 4);

        // The source moves between quanta — nothing is pinned right now.
        f.source.advance();

        let second = f
            .registry
            .apply(1, request(true, DirtyCapabilities::RebuildAll));
        assert!(
            matches!(second, ApplyOutcome::Progress { .. }),
            "the restart cannot complete in one quantum"
        );
        assert_eq!(f.metrics.recaptures_restarted(), 1);
        assert_eq!(
            f.registry.pending_slots(),
            APPLY_QUANTUM,
            "the 64 slots built against the OLD generation went back on the queue"
        );

        let third = f
            .registry
            .apply(1, request(true, DirtyCapabilities::RebuildAll));
        assert!(matches!(third, ApplyOutcome::Current { .. }));
        let generation = f.source.generation();
        assert!(
            f.registry.inner.lock().slots.values().all(|slot| slot
                .cells
                .facts
                .load()
                .as_ref()
                .is_some_and(|facts| facts.epoch.generation == generation)),
            "ONE coherent source generation across the whole retained set"
        );
    }

    /// Ordinary `Caps` work at a NEWER generation is COMPLETE.
    ///
    /// Rebuilding C at G+1 leaves unrelated D legitimately stamped at G. Demanding
    /// global coherence outside a recapture epoch would report `Progress` for a
    /// pass that finished everything it was asked to do — and with `pending` empty
    /// there is nothing to mark, so the actor would park owing a recapture that
    /// nothing will ever wake (Kyra OLB-2B-E3b).
    #[test]
    fn ordinary_caps_at_a_newer_generation_reports_current() {
        let f = fixture();
        let family = f.family();
        let c = key(1, "nrpc:c");
        let d = key(1, "nrpc:d");
        let _hc = family.demand(c.clone()).expect("c");
        let _hd = family.demand(d.clone()).expect("d");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        let first = f.source.generation();

        // The source moves, dirtying ONLY c.
        f.source.advance();
        let outcome = f.registry.apply(1, request(false, caps(&["nrpc:c"])));

        let second = f.source.generation();
        assert_ne!(first, second);
        assert_eq!(
            f.registry
                .base_facts_unvalidated(&c)
                .expect("c rebuilt")
                .epoch
                .generation,
            second
        );
        assert_eq!(
            f.registry
                .base_facts_unvalidated(&d)
                .expect("d untouched")
                .epoch
                .generation,
            first,
            "an unrelated slot legitimately keeps its older stamp"
        );
        assert_eq!(f.registry.pending_slots(), 0, "no work is owed");
        assert_eq!(
            outcome,
            ApplyOutcome::Current {
                source_generation: second
            },
            "the pass rebuilt everything it was asked to; that IS complete"
        );
    }

    /// First-demand work at a NEWER generation is likewise complete.
    #[test]
    fn first_demand_at_a_newer_generation_reports_current() {
        let f = fixture();
        let family = f.family();
        let established = key(1, "nrpc:a");
        let _ha = family.demand(established.clone()).expect("a");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        let first = f.source.generation();

        f.source.advance();
        let fresh = key(1, "nrpc:b");
        let _hb = family.demand(fresh.clone()).expect("b");

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        let second = f.source.generation();
        assert_eq!(
            f.registry
                .base_facts_unvalidated(&fresh)
                .expect("built")
                .epoch
                .generation,
            second
        );
        assert_eq!(
            f.registry
                .base_facts_unvalidated(&established)
                .expect("retained")
                .epoch
                .generation,
            first,
            "the unrelated retained slot was not rebuilt and did not need to be"
        );
        assert_eq!(f.registry.pending_slots(), 0);
        assert_eq!(
            outcome,
            ApplyOutcome::Current {
                source_generation: second
            }
        );
    }

    /// Structural: `Progress` is never returned with an empty queue, so it always
    /// carries the wake its contract promises.
    #[test]
    fn progress_always_implies_owed_work() {
        let f = fixture();
        let mut family = f.family();
        let mut held = Vec::new();
        for index in 0..(APPLY_QUANTUM + 3) {
            if index > 0 && index % MAX_HANDLES_PER_FAMILY == 0 {
                family = f.family();
            }
            held.push(
                family
                    .demand(key(1, &format!("nrpc:q{index}")))
                    .expect("demanded"),
            );
        }

        for pass in 0..4 {
            let outcome = f
                .registry
                .apply(1, request(true, DirtyCapabilities::RebuildAll));
            let owed = f.registry.pending_slots();
            match outcome {
                ApplyOutcome::Progress { .. } => assert!(
                    owed > 0,
                    "pass {pass}: Progress with nothing owed strands the actor"
                ),
                ApplyOutcome::Current { .. } => assert_eq!(owed, 0, "pass {pass}"),
                other => panic!("pass {pass}: unexpected {other:?}"),
            }
        }
    }

    // ------------------------------------------------------------- currentness

    /// A concurrent mutation lands DURING reconstruction — without waiting for it —
    /// and the commit pin for the snapshot's generation then fails. Nothing stale
    /// installs, and every still-live selected slot goes back on the queue.
    ///
    /// This is the transition the snapshot/commit split exists for: the rival
    /// mutation is performed from another thread with `try_lock`, so it FAILS
    /// outright if reconstruction were holding the publication gate.
    #[test]
    fn a_mutation_during_reconstruction_defeats_the_commit_pin_without_waiting() {
        let f = fixture();
        let target = key(1, "nrpc:a");
        let _held = f.family().demand(target.clone()).expect("a");

        let moved = Arc::new(AtomicBool::new(false));
        {
            let source = f.source.clone();
            let moved = moved.clone();
            *f.source.during_build.lock() = Some(Box::new(move || {
                let source = source.clone();
                let advanced = std::thread::spawn(move || match source.gate.try_lock() {
                    Some(mut generation) => {
                        *generation += 1;
                        true
                    }
                    None => false,
                })
                .join()
                .expect("mutation thread");
                assert!(
                    advanced,
                    "a rival mutation must not have to wait on reconstruction"
                );
                moved.store(true, Ordering::Release);
            }));
        }

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(moved.load(Ordering::Acquire), "the mutation must have run");
        assert_eq!(outcome, ApplyOutcome::Superseded);
        assert_eq!(f.metrics.installs(), 0, "nothing stale installed");
        assert!(f.registry.base_facts_unvalidated(&target).is_none());
        assert_eq!(
            f.registry.pending_slots(),
            1,
            "the live slot re-entered the authoritative queue"
        );

        // And the next pass, against the settled generation, completes.
        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(
            outcome,
            ApplyOutcome::Current {
                source_generation: f.source.generation()
            }
        );
    }

    /// The commit pin is acquired BEFORE the registry lock and is still held when
    /// the facts land: its `Drop` observes the installation already done.
    #[test]
    fn the_commit_pin_outlives_the_installation() {
        let f = fixture();
        let target = key(1, "nrpc:a");
        let _held = f.family().demand(target.clone()).expect("a");

        let observed = Arc::new(AtomicBool::new(false));
        {
            let registry = f.registry.clone();
            let metrics = f.metrics.clone();
            let observed = observed.clone();
            let target = target.clone();
            *f.source.on_commit_release.lock() = Some(Arc::new(move || {
                assert_eq!(
                    metrics.installs(),
                    1,
                    "the commit pin must still be held when the facts are installed"
                );
                assert!(registry.base_facts_unvalidated(&target).is_some());
                observed.store(true, Ordering::Release);
            }));
        }

        let outcome = f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert!(
            observed.load(Ordering::Acquire),
            "the commit pin must have dropped"
        );
        let generation = f.source.generation();
        assert_eq!(
            outcome,
            ApplyOutcome::Current {
                source_generation: generation
            }
        );
        assert_eq!(
            f.registry
                .base_facts_unvalidated(&target)
                .expect("built")
                .epoch
                .generation,
            generation,
            "facts carry exactly the generation the commit pin proved current"
        );
    }

    /// No registry lock is held across the source snapshot or reconstruction, and
    /// no source/publication lock is held across reconstruction — all asserted from
    /// inside the seams themselves, which fail if either is held.
    #[test]
    fn neither_lock_is_held_across_the_snapshot_or_reconstruction() {
        let f = fixture();
        let _held = f.family().demand(key(1, "nrpc:a")).expect("a");
        f.registry.apply(1, request(true, DirtyCapabilities::Clean));
        assert_eq!(f.source.queries(), vec![key(1, "nrpc:a")]);
        assert_eq!(
            f.source.snapshots.load(Ordering::Acquire),
            1,
            "one brief capture per quantum"
        );
    }
}