aion_server/worker/

registry.rs

//! Connected-worker registry keyed by worker-pool address and activity type.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::num::NonZeroU32;
use std::pin::Pin;
use std::sync::{Arc, Mutex, MutexGuard};
use std::task::{Context, Poll};
use std::time::Instant;

use aion_core::{
    ClusterEvent, DeploymentAssociation, InterventionCapabilities, WorkerDeathReason,
    WorkerTransport,
};
use aion_proto::{ProtoActivityTask, ProtoCancelActivity, ProtoLivenessPing, ProtoRegisterWorker};
use aion_store::{NamespaceOrigin, NamespacePlacement, NamespaceStore, WorkerDeploymentStore};
use tokio::sync::{Notify, mpsc};

use crate::cluster_publisher::ClusterEventPublisher;
use crate::config::AutoCreate;
use crate::error::ServerError;
use crate::namespace::{CallerIdentity, NamespaceGuard, NamespaceMinter, NamespaceOperation};
use crate::observability::Metrics;
use crate::worker::admission_audit::AdmissionAudit;
use crate::worker::heartbeat::DispatchExclusion;

mod capacity;
mod census;
mod reservation;

use capacity::advertised_capacity;
pub use reservation::DispatchReservation;
use reservation::eligible_candidates_in_rotation;

/// The literal task queue an empty/absent selector normalizes to.
///
/// A worker-pool address has two disjoint dimensions; the second one
/// (`task_queue`) is a liveness selector, not a correctness boundary. An empty
/// `task_queue` is normalized to this one named default pool so a producer that
/// names no queue and a worker that advertises none both land on the same pool.
///
/// Re-exported from [`aion_core::DEFAULT_TASK_QUEUE`] so the server cannot drift
/// from the canonical domain default; the name is kept stable here for existing
/// call sites.
pub use aion_core::DEFAULT_TASK_QUEUE;

/// Server-side handle used to push activity tasks to a connected worker stream.
pub type WorkerTaskSender = mpsc::Sender<WorkerMessage>;

/// Transport through which the server delivers a dispatch to a registered worker.
///
/// A worker is selected the SAME way regardless of transport (`select_worker`
/// over the `(namespace, task_queue, node)` pool key); only the delivery leg
/// differs. The default gRPC path pushes a [`WorkerMessage`] onto the worker's
/// stream `mpsc` ([`WorkerDelivery::Grpc`]); a liminal-connected worker is
/// delivered to by pushing the dispatch out on its existing liminal connection
/// ([`WorkerDelivery::Liminal`], feature-gated). This enum is the minimal
/// transport-agnostic seam: the registry holds it on each [`WorkerHandle`], and
/// the dispatch path reads the variant it needs. The gRPC variant carries exactly
/// the `mpsc::Sender` it always did, so the gRPC dispatch path is unchanged.
#[derive(Clone, Debug)]
pub enum WorkerDelivery {
    /// gRPC stream delivery: the dispatch path pushes a [`WorkerMessage`] onto
    /// this `mpsc` sender, exactly as before this enum existed.
    Grpc(WorkerTaskSender),
    /// Liminal server-push delivery: the dispatch path pushes the serialized
    /// dispatch out on the worker's existing liminal connection and awaits the
    /// correlated reply. Carries the connection identity needed to address that
    /// push.
    #[cfg(feature = "liminal-transport")]
    Liminal(crate::worker::liminal_transport::LiminalWorkerDelivery),
}

impl WorkerDelivery {
    /// Which transport this delivery rides, stripped of the handle that
    /// addresses it.
    ///
    /// The liveness probe needs to know a worker's transport WITHOUT holding its
    /// connection handle, because the question it asks is not "how do I reach
    /// this worker" but "do I carry any wire on which this worker could be
    /// asked" (#25). Answering that from a cloned sender or a connection pid
    /// would tie a coverage decision to a live handle it does not need.
    ///
    /// This is the ONE mapping from a held delivery to its wire discriminant.
    /// Two byte-identical private copies of it existed — one here for the
    /// cluster-event emitter, one in
    /// [`cluster_stream`](crate::stream::cluster_stream) for the snapshot — and
    /// a third was nearly written for the liveness verdict. A discriminant table
    /// kept in three places is three chances for a transport to be added to two
    /// of them.
    #[must_use]
    pub const fn transport(&self) -> WorkerTransport {
        match self {
            Self::Grpc(_) => WorkerTransport::Grpc,
            #[cfg(feature = "liminal-transport")]
            Self::Liminal(_) => WorkerTransport::Liminal,
        }
    }
}

/// Message queued from server-side dispatch/shutdown into a worker stream writer.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum WorkerMessage {
    /// Activity invocation pushed to a worker.
    ActivityTask(Box<ProtoActivityTask>),
    /// Graceful-shutdown notification; no new work will be dispatched.
    DrainRequest,
    /// Transport liveness ping pushed by the liveness probe (#197).
    ///
    /// It travels this channel — the SAME one dispatches travel — deliberately.
    /// The fact the probe needs is whether the server can reach this worker's
    /// DISPATCH path, and a ping on a parallel channel could be answered by a
    /// process whose dispatch path is dead. Answered by the worker SDK RUNTIME,
    /// never by action code.
    LivenessPing(ProtoLivenessPing),
    /// Ask the worker to stop ONE in-flight activity (#233).
    ///
    /// A request, not a guarantee: pushing this proves only that the server
    /// asked. Whether the work stops depends on the worker still holding the
    /// activity and on the action being interruptible, and neither is
    /// observable from here.
    ///
    /// It travels this channel — the SAME one the dispatch travelled — so the
    /// cancel cannot overtake or bypass the task it interrupts, and so a worker
    /// whose dispatch path is dead cannot appear to have been told.
    CancelActivity(ProtoCancelActivity),
}

/// Address of a worker pool: the two disjoint routing dimensions that select a
/// pool, before an `activity_type` is matched within it.
///
/// `namespace` is the correctness/isolation boundary — a workflow's activities
/// only ever reach workers in the workflow's namespace, so crossing it is a bug.
/// `task_queue` is the pool/flavour selector within that namespace (norn /
/// claude / cpu / gpu) — a miss is a liveness issue, never a correctness one.
///
/// This is a named type rather than a `(String, String)` tuple so a `node`
/// dimension (Tier 3 affinity) can be added later without re-threading every
/// call site.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PoolAddress {
    namespace: String,
    task_queue: String,
}

impl PoolAddress {
    /// Build a pool address, normalizing an empty `task_queue` to the named
    /// [`DEFAULT_TASK_QUEUE`] pool. The `namespace` is the authorization
    /// boundary and is never normalized.
    #[must_use]
    pub fn new(namespace: impl Into<String>, task_queue: impl Into<String>) -> Self {
        let task_queue = task_queue.into();
        let task_queue = if task_queue.is_empty() {
            String::from(DEFAULT_TASK_QUEUE)
        } else {
            task_queue
        };
        Self {
            namespace: namespace.into(),
            task_queue,
        }
    }

    /// The correctness/isolation boundary of this pool.
    #[must_use]
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// The pool/flavour selector within the namespace.
    #[must_use]
    pub fn task_queue(&self) -> &str {
        &self.task_queue
    }
}

/// Registry match key: a worker-pool address plus the activity type matched
/// within that pool. A named type (not an anonymous tuple) so the routing
/// identity stays self-describing and extensible.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
struct ActivityKey {
    pool: PoolAddress,
    activity_type: String,
}

impl ActivityKey {
    fn new(pool: PoolAddress, activity_type: impl Into<String>) -> Self {
        Self {
            pool,
            activity_type: activity_type.into(),
        }
    }
}

type WorkerMap = HashMap<WorkerId, WorkerHandle>;
type RegistryMap = HashMap<ActivityKey, WorkerMap>;

/// The capacity advertised by a worker that has no executor of its own.
///
/// A worker registered with a bare `mpsc` sender — the in-process façades'
/// callers, the built-in workers, and every harness that drains the channel
/// itself — runs nothing on its own account: what happens to a pushed task is
/// decided by whoever holds the receiver, so there is no number of simultaneous
/// activities such a registration would refuse. This states that, once, instead
/// of a literal repeated at every call.
///
/// It is NOT a default and no production registration can reach it. A gRPC
/// worker states its capacity in its registration frame and is refused if it
/// does not; a liminal worker, whose frame cannot carry one, registers with
/// capacity unknown and announces its own number a frame later. Neither route
/// reaches this constant — it belongs to the in-process façades alone.
pub const UNBOUNDED_SENDER_WORKER_CONCURRENCY: u32 = u32::MAX;

/// Stable identifier assigned to a connected worker stream.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct WorkerId(u64);

impl WorkerId {
    /// Build a worker id from the numeric value exposed by administrative and
    /// wire surfaces.
    #[must_use]
    pub const fn from_value(value: u64) -> Self {
        Self(value)
    }

    /// Raw numeric value, as carried by the wire `RegisterAck.worker_id` so
    /// workers can correlate their logs with the server's.
    #[must_use]
    pub const fn value(self) -> u64 {
        self.0
    }
}

/// Worker-supplied identity for one instance of a durable deployment.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WorkerInstanceIdentity {
    /// Durable deployment name supplied on the wire.
    pub deployment: String,
    /// Operator/launcher-assigned instance identifier.
    pub instance_id: String,
    /// Deployment-store lookup result captured when registration was accepted.
    pub association: DeploymentAssociation,
}

/// What a registration says about the worker beyond where it is routed:
/// the identity it registered under, its deployment instance, and the
/// intervention capabilities its harness advertises.
#[derive(Debug, Clone)]
pub struct RegistrationOptions {
    identity: String,
    max_concurrency: Option<u32>,
    instance: Option<WorkerInstanceIdentity>,
    intervention_capabilities: InterventionCapabilities,
}

impl RegistrationOptions {
    /// Options for a worker registering under `identity` and advertising
    /// `max_concurrency`, with no deployment instance and the empty
    /// (observability-only) capability set.
    ///
    /// Capacity is a CONSTRUCTOR argument rather than a `with_` setter because
    /// there is no honest value to fall back to: a default would decide how
    /// much work a process is handed on the strength of nobody having said, and
    /// selection reads it on every dispatch. Every caller states it.
    ///
    /// Zero is rejected by [`ConnectedWorkerRegistry::register_delivery`], the
    /// one funnel every transport and every façade passes through, so the check
    /// has a single site and no caller can route around it.
    #[must_use]
    pub fn identified(identity: impl Into<String>, max_concurrency: u32) -> Self {
        Self {
            identity: identity.into(),
            max_concurrency: Some(max_concurrency),
            instance: None,
            intervention_capabilities: InterventionCapabilities::none(),
        }
    }

    /// Options for a worker that has registered over a transport whose
    /// registration frame cannot carry capacity, and will announce it in the
    /// next frame instead.
    ///
    /// This is NOT a default and NOT an unbounded worker. It records that the
    /// server does not yet know, which selection reads as "do not dispatch"
    /// until [`ConnectedWorkerRegistry::set_advertised_capacity`] supplies the
    /// worker's own configured number. The only caller is the liminal
    /// transport: `liminal::protocol::WorkerRegistration` is a published wire
    /// type with six fields — namespaces, task queue, node, activity types,
    /// identity, activities — and none of them is capacity, so a liminal worker
    /// physically cannot state it in the frame that registers it.
    ///
    /// Failing closed for that one round trip is deliberate. The alternative is
    /// to pick a number on the worker's behalf, and every value is wrong: too
    /// low serializes a worker that fans, too high pushes it past its own
    /// admission — which is the defect this whole landing exists to remove.
    #[must_use]
    pub fn identified_pending_capacity(identity: impl Into<String>) -> Self {
        Self {
            identity: identity.into(),
            max_concurrency: None,
            instance: None,
            intervention_capabilities: InterventionCapabilities::none(),
        }
    }

    /// The deployment instance the worker registered as, when it sent one.
    #[must_use]
    pub fn with_instance(mut self, instance: Option<WorkerInstanceIdentity>) -> Self {
        self.instance = instance;
        self
    }

    /// The neutral [`InterventionCapabilities`] the worker's harness advertises
    /// (NOI-6). Metadata the intervention router gates on, never a routing
    /// dimension.
    #[must_use]
    pub fn with_intervention_capabilities(
        mut self,
        capabilities: InterventionCapabilities,
    ) -> Self {
        self.intervention_capabilities = capabilities;
        self
    }
}

/// Cloneable handle for a registered worker stream.
///
/// A worker serves a SET of namespaces under a single `task_queue`, so it is
/// indexed under one `(namespace, task_queue, activity_type)` key per namespace
/// in its set. `node` is an OPTIONAL locality affinity (a locality, not a
/// process — many handles may share a node id) used as a within-pool filter at
/// selection time; `None` means the worker advertised no locality.
#[derive(Clone, Debug)]
pub struct WorkerHandle {
    id: WorkerId,
    /// The identity string the worker registered under — the wire's
    /// `ProtoRegisterWorker.identity`, carried so a lease can name the process
    /// that took an attempt (WA-010 R3). Never a routing dimension.
    identity: String,
    namespaces: BTreeSet<String>,
    task_queue: String,
    node: Option<String>,
    activity_types: BTreeSet<String>,
    instance: Option<WorkerInstanceIdentity>,
    /// How many activities this worker will run AT ONCE, exactly as it
    /// advertised at registration.
    ///
    /// `NonZeroU32` rather than a plain integer because zero is not a capacity
    /// this server can act on — a worker advertising it could never be
    /// selected — so it is refused at admission and the type carries that
    /// refusal forward instead of leaving every reader to re-check it.
    max_concurrency: Option<NonZeroU32>,
    delivery: WorkerDelivery,
    /// The neutral mid-run intervention primitives this worker's harness advertises
    /// support for (NOI-6). The server gates every intervention command on THIS set
    /// and NEVER routes an unadvertised primitive. Empty = observability-only (the
    /// default for every non-agent worker), so a normal activity worker advertises
    /// no controls and the intervention router refuses every command for it.
    intervention_capabilities: InterventionCapabilities,
}

impl WorkerHandle {
    /// Worker identifier assigned by this server process.
    #[must_use]
    pub const fn id(&self) -> WorkerId {
        self.id
    }

    /// Namespaces authorized for this worker stream. The worker is reachable for
    /// a dispatch only when its set includes the workflow's namespace.
    #[must_use]
    pub const fn namespaces(&self) -> &BTreeSet<String> {
        &self.namespaces
    }

    /// The identity string this worker registered under, exactly as the
    /// registration frame carried it (empty when the worker sent none).
    #[must_use]
    pub fn identity(&self) -> &str {
        &self.identity
    }

    /// Task queue (pool/flavour) this worker serves within each namespace.
    #[must_use]
    pub fn task_queue(&self) -> &str {
        &self.task_queue
    }

    /// Optional locality affinity this worker advertised. `None` means the
    /// worker carries no node and is reachable only for unpinned dispatches.
    #[must_use]
    pub fn node(&self) -> Option<&str> {
        self.node.as_deref()
    }

    /// Activity types advertised by this worker.
    #[must_use]
    pub fn activity_types(&self) -> &BTreeSet<String> {
        &self.activity_types
    }

    /// Optional durable deployment/instance association supplied at registration.
    #[must_use]
    pub const fn instance(&self) -> Option<&WorkerInstanceIdentity> {
        self.instance.as_ref()
    }

    /// How many activities this worker runs AT ONCE, as it advertised, or
    /// `None` while it has registered but not yet said.
    ///
    /// A dispatch PRECONDITION, read by [`eligible_candidates_in_rotation`]:
    /// a worker holding this many in-flight activities is passed over rather
    /// than pushed past its own admission. Pushing past it is what produced a
    /// worker that stopped reading its stream, answered no pings, and was then
    /// deregistered as lost while it was in fact working.
    ///
    /// `None` is reachable only on the liminal transport, whose published
    /// registration frame has no capacity field, and only for the one round
    /// trip between that registration and the worker's capacity announcement.
    /// It means UNKNOWN, never unlimited: selection treats an unknown capacity
    /// as full, because the alternative is to invent a number for a process
    /// that is about to state its own.
    #[must_use]
    pub const fn max_concurrency(&self) -> Option<NonZeroU32> {
        self.max_concurrency
    }

    /// Whether the server still holds an OPEN push channel to this worker.
    ///
    /// The explicit connected fact, as opposed to the absence of noise. Silence
    /// alone cannot tell a worker that is gone from one that is busy, and the
    /// expiry sweep used to read the second as the first; this is the third
    /// input that separates them.
    ///
    /// gRPC answers from the stream sender itself — a closed receiver means the
    /// tonic task that owned the stream is gone. Liminal answers from its
    /// connection: the transport holds a live push connection or it does not.
    #[must_use]
    pub fn is_connected(&self) -> bool {
        match &self.delivery {
            WorkerDelivery::Grpc(sender) => !sender.is_closed(),
            #[cfg(feature = "liminal-transport")]
            WorkerDelivery::Liminal(delivery) => delivery.is_connected(),
        }
    }

    /// The transport this worker is delivered to through.
    #[must_use]
    pub const fn delivery(&self) -> &WorkerDelivery {
        &self.delivery
    }

    /// The neutral intervention primitives this worker's harness advertises (NOI-6).
    ///
    /// The intervention router gates on this set and never routes an unadvertised
    /// primitive. Empty (the default for a plain activity worker) means the worker
    /// is observability-only: the router refuses every intervention command for it.
    #[must_use]
    pub const fn intervention_capabilities(&self) -> &InterventionCapabilities {
        &self.intervention_capabilities
    }

    /// gRPC stream sender used by the gRPC dispatch path to push work, or `None`
    /// when this worker is delivered to over a non-gRPC transport (liminal).
    ///
    /// The gRPC dispatch path registers every worker with a [`WorkerDelivery::Grpc`]
    /// delivery, so this is always `Some` for a gRPC-registered worker — the
    /// behaviour the path relied on before delivery became transport-agnostic.
    #[must_use]
    pub fn sender(&self) -> Option<&WorkerTaskSender> {
        match &self.delivery {
            WorkerDelivery::Grpc(sender) => Some(sender),
            #[cfg(feature = "liminal-transport")]
            WorkerDelivery::Liminal(_) => None,
        }
    }
}

#[derive(Debug)]
struct RegistryState {
    next_worker_id: u64,
    workers: BTreeMap<WorkerId, WorkerHandle>,
    by_activity: RegistryMap,
    /// Round-robin cursor per `(namespace, task_queue, activity_type)` triple, so
    /// each pool rotates independently of every other pool.
    rotation: HashMap<ActivityKey, usize>,
    /// When a worker advertising a given `(pool, activity_type)` and node last
    /// LEFT service, per advertised node. Read by [`ConnectedWorkerRegistry::pool_census`]
    /// to answer "how old is the last compatible poller" for an address that has
    /// none right now (R1). Node-keyed because a node-pinned dispatch's notion
    /// of "compatible" is node-specific: a worker still serving the activity on
    /// another node must never make a pinned address look freshly served.
    last_departure: HashMap<ActivityKey, BTreeMap<Option<String>, Instant>>,
    /// Workers the server currently cannot reach on the server-to-worker push
    /// leg, and which are therefore excluded from selection while they remain
    /// so.
    ///
    /// Held here rather than derived at selection time because selection must
    /// not take the heartbeat tracker's lock: the liveness probe owns the
    /// evidence and publishes the verdict, and selection only reads it.
    ///
    /// This is exclusion from DISPATCH, not deregistration. An unreachable
    /// worker stays registered, keeps its in-flight work, and becomes eligible
    /// again the moment a ping is answered — nothing about it is torn down on
    /// the strength of a push failure.
    ///
    /// 🔴 A MAP, not a set, and the value is load-bearing.
    /// [`DispatchExclusion`] separates two facts its own documentation calls
    /// "DIFFERENT FACTS an operator must be able to tell apart": an
    /// `OpeningProbation` clears itself within seconds and is the ordinary cost
    /// of connecting, while a `ReachabilityLost` is an incident that does not
    /// clear on its own. A flat set collapses them, and a pool parked on the
    /// second one then waits with nothing published about why — the same shape
    /// as a delivery gate that cannot tell a key never begun from one
    /// released. The prober has the value already; this is simply where it
    /// stopped being thrown away.
    dispatch_ineligible: BTreeMap<WorkerId, DispatchExclusion>,
    /// How many dispatches each worker is currently holding.
    ///
    /// The selection-side projection of the heartbeat tracker's in-flight task
    /// map, and NOT a second accounting kept in parallel with it: both are
    /// written by the SAME call — [`HeartbeatTracker::track_task`] and
    /// [`HeartbeatTracker::complete_task`](super::heartbeat::HeartbeatTracker::complete_task)
    /// take this registry precisely so the two cannot be updated apart. The
    /// projection lives here because selection must not take the tracker's
    /// lock, which is the same reason `dispatch_ineligible` lives here rather
    /// than being derived at selection time.
    ///
    /// "One call" is not the same as "one idempotency", and the difference bit
    /// once: the tracker's map is keyed by `(worker, workflow, activity)` and an
    /// insert over a live key overwrites, while this count increments. Tracking
    /// one key twice without an intervening completion therefore counted twice
    /// and could only be decremented once — a slot leaked for the life of the
    /// registration. `track_task` now increments only when its insert actually
    /// added an entry, so the two agree on how many times one dispatch counts.
    ///
    /// It is also raised by [`ConnectedWorkerRegistry::select_and_reserve`] and
    /// [`ConnectedWorkerRegistry::reserve_worker`], which claim a slot at
    /// SELECTION and hold it — as a [`DispatchReservation`] — until the tracker
    /// takes the count over. That overlap is one extra held slot for the width
    /// of the hand-over and can only park a dispatch that would otherwise have
    /// raced, never admit one.
    ///
    /// An entry is dropped when the worker leaves service, so this map is
    /// bounded by the workers actually registered.
    in_flight: BTreeMap<WorkerId, u32>,
}

impl Default for RegistryState {
    fn default() -> Self {
        Self {
            next_worker_id: 1,
            workers: BTreeMap::new(),
            by_activity: HashMap::new(),
            rotation: HashMap::new(),
            last_departure: HashMap::new(),
            dispatch_ineligible: BTreeMap::new(),
            in_flight: BTreeMap::new(),
        }
    }
}

/// Cloneable registry of currently connected worker streams.
#[derive(Clone)]
pub struct ConnectedWorkerRegistry {
    inner: Arc<Mutex<RegistryState>>,
    metrics: Option<Metrics>,
    /// WS3 cluster-event publisher: emits `WorkerConnected`/`WorkerDisconnected`
    /// topology deltas on register/deregister. `None` keeps existing
    /// constructions (and every test) silent, exactly like `metrics`.
    cluster_publisher: Option<ClusterEventPublisher>,
    /// Minted-on-use hook (Control-Plane Phase 1). `None` disables minting, so
    /// registration is byte-identical to before the registry existed; `Some`
    /// durably records (open) or gates (closed) each authorized namespace.
    minter: Option<NamespaceMinter>,
    /// Durable deployment store used only to classify instance associations.
    deployment_store: Option<Arc<dyn WorkerDeploymentStore>>,
    worker_arrived: Arc<Notify>,
    /// The refusal side of this registry's ledger, shared by every transport.
    ///
    /// The registry records who was ADMITTED; this records who was turned away,
    /// so a repeated identical refusal can be met with silence. It lives here
    /// because both callers of the admission gate already hold the registry, so
    /// one home serves both — and one shared record is what stops the two
    /// transports drifting apart the way #147 found them.
    audit: Arc<AdmissionAudit>,
    /// Pulsed whenever a worker's capacity slot is FREED, so a dispatcher that
    /// parked work for want of capacity is woken by the event itself rather than
    /// waiting out a timer.
    ///
    /// This is the outbox dispatcher's existing advisory wake (the one the stage
    /// seam pulses when a row is committed), shared rather than duplicated: a
    /// freed slot and a newly-staged row are the same kind of news to that loop
    /// — "there may be work you can place now" — and it already coalesces and
    /// degrades cleanly to its interval poll. `None` (every in-process façade
    /// and most tests) simply means nobody is listening.
    capacity_wake: Option<Arc<Notify>>,
}

impl std::fmt::Debug for ConnectedWorkerRegistry {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ConnectedWorkerRegistry")
            .field(
                "deployment_store_attached",
                &self.deployment_store.is_some(),
            )
            .finish_non_exhaustive()
    }
}

impl Default for ConnectedWorkerRegistry {
    fn default() -> Self {
        Self {
            inner: Arc::new(Mutex::new(RegistryState::default())),
            metrics: None,
            cluster_publisher: None,
            minter: None,
            deployment_store: None,
            worker_arrived: Arc::new(Notify::new()),
            audit: Arc::new(AdmissionAudit::new()),
            capacity_wake: None,
        }
    }
}

/// A live subscription to the next change in what dispatch selection can see,
/// taken from [`ConnectedWorkerRegistry::worker_arrival`] and awaited as a
/// future.
///
/// # Why this is a value and not a bare `wait` method
///
/// The registry's two wake sources — a registration
/// ([`ConnectedWorkerRegistry::register`] and every sibling, through their
/// shared tail) and a published reachability verdict
/// ([`ConnectedWorkerRegistry::set_dispatch_ineligible`]) — are both
/// `Notify::notify_waiters`, which stores **no permit**. A caller that reads the
/// registry, misses, and only then constructs its wait has already lost any
/// arrival that landed during the read: the broadcast fired into an empty waiter
/// list. That caller then sleeps on a pool it holds positive evidence is served,
/// until some unrelated later registration happens to wake it — and on the gRPC
/// transport, where no liveness probe runs and no verdict is ever published, an
/// unrelated registration is the ONLY thing that ever could.
///
/// So the subscription is a value the caller takes **before** it looks, and
/// awaits **after** it has missed. Everything that fires in between is retained.
///
/// # Why it is retained — tokio 1.52.3 at the bytes
///
/// Line numbers are in `tokio-1.52.3/src/sync/notify.rs`, the version
/// `Cargo.lock` pins.
///
/// - `Notify::notified_owned` (`:613`) snapshots the process-wide
///   `notify_waiters` call counter into the future **at construction**
///   (`:619`, `notify_waiters_calls: get_num_notify_waiters_calls(state)`).
/// - `notify_waiters` (`:743`) increments that counter unconditionally —
///   `inner_notify_waiters` bumps it at `:755` when nobody is parked and at
///   `:761` when someone is. A broadcast into an empty list is therefore not
///   lost; it moves a number.
/// - `poll_notified`'s `State::Init` arm compares the snapshot against the live
///   counter at `:1124`, and again under the waiter lock at `:1156`; a
///   difference sends the future straight to `State::Done` → `Poll::Ready`.
///
/// That comparison is the retention property, and it belongs to **construction**.
/// `OwnedNotified::enable` (`:1059`) is called here too: it runs that same `Init`
/// arm eagerly and, on the not-yet-notified path, pushes the waiter into the list
/// at `:1219` — so the waiter is registered at a defined point rather than at
/// first poll, and a `notify_one` permit would be retained as well if one were
/// ever added to this `Notify`. There is none today; this type must not rest its
/// correctness on that staying true.
///
/// # What it costs
///
/// `OwnedNotified` rather than the borrowed `Notified<'_>` because the value
/// crosses a `&mut dyn FnMut(..)` boundary (the wait path's `park` contract) and
/// sits in a `tokio::select!` arm (the bridge's park); the borrowed form forces
/// a higher-ranked bound through the first and pinning ceremony at the second.
///
/// `Pin<Box<_>>` because `enable` needs `Pin<&mut Self>` at construction, and
/// because it makes this type `Unpin` so no call site owes pinning ceremony. The
/// price is one heap allocation per selection-loop iteration, on a path that is
/// about to park. The alternative — hand back a bare `OwnedNotified` and ask
/// every call site to remember `pin!` and `enable()` — puts the obligation back
/// on the call sites, and a call site that forgot its obligation is precisely
/// the defect this type exists to end.
#[must_use = "a WorkerArrival that is constructed and dropped is a subscription thrown away"]
pub struct WorkerArrival {
    notified: Pin<Box<tokio::sync::futures::OwnedNotified>>,
}

impl std::fmt::Debug for WorkerArrival {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.debug_struct("WorkerArrival").finish()
    }
}

impl std::future::Future for WorkerArrival {
    type Output = ();

    fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
        // `Pin<Box<_>>` is `Unpin`, so the outer pin carries no obligation and
        // the inner one is what the notification state is actually pinned by.
        std::future::Future::poll(self.get_mut().notified.as_mut(), context)
    }
}

impl ConnectedWorkerRegistry {
    /// Build a registry that records connected-worker gauge updates.
    #[must_use]
    pub fn with_metrics(metrics: Metrics) -> Self {
        Self {
            inner: Arc::new(Mutex::new(RegistryState::default())),
            metrics: Some(metrics),
            cluster_publisher: None,
            minter: None,
            deployment_store: None,
            worker_arrived: Arc::new(Notify::new()),
            audit: Arc::new(AdmissionAudit::new()),
            capacity_wake: None,
        }
    }

    /// Share the dispatcher wake this registry pulses when a capacity slot is
    /// FREED.
    ///
    /// The busy answer a full pool gives is attempt-neutral, so a parked row is
    /// waiting for capacity — and capacity-freeing is an event this server
    /// produces, not a deadline it should guess at. Without this the only thing
    /// re-offering a busy-parked row was its durable `visible_after`, which
    /// rides the operator's FAILURE backoff: raising that legitimately (to slow
    /// genuine retries) parked healthy work behind it for the same duration.
    ///
    /// Pulsed by
    /// [`record_dispatch_finished`](Self::record_dispatch_finished), which every
    /// slot-free goes through — a completed dispatch's untrack and a released
    /// reservation alike.
    #[must_use]
    pub fn with_capacity_wake(mut self, wake: Arc<Notify>) -> Self {
        self.capacity_wake = Some(wake);
        self
    }

    /// The admission audit every registration transport names its refusals
    /// through. Clones of a registry share one, so a worker refused over gRPC
    /// and then over liminal is one site, not two.
    #[must_use]
    pub fn admission_audit(&self) -> &AdmissionAudit {
        &self.audit
    }

    /// Attach the WS3 cluster-event publisher so worker topology changes are
    /// pushed to the dashboard. Pure builder addition.
    #[must_use]
    pub fn with_cluster_publisher(mut self, publisher: ClusterEventPublisher) -> Self {
        self.cluster_publisher = Some(publisher);
        self
    }

    /// Attach the durable worker-deployment store used for association lookup.
    #[must_use]
    pub fn with_worker_deployment_store(mut self, store: Arc<dyn WorkerDeploymentStore>) -> Self {
        self.deployment_store = Some(store);
        self
    }

    /// Install the minted-on-use namespace hook (Control-Plane Phase 1).
    ///
    /// After a registration is authorized and its namespace set scoped, each
    /// authorized namespace is durably recorded ([`AutoCreate::Open`]) or gated
    /// ([`AutoCreate::Closed`]) through `store`. Without this builder the
    /// registry never touches the namespace registry, so registration stays
    /// byte-identical to before the registry existed. Pure builder addition,
    /// mirroring [`Self::with_cluster_publisher`].
    ///
    /// When a cluster publisher has already been attached
    /// ([`Self::with_cluster_publisher`], called first on the boot path), it is
    /// threaded into the minter so a first worker-mint emits the live
    /// `namespace created` delta to the ops console (S8). Order-independence is
    /// not assumed: callers wire the publisher before minting on the boot path.
    #[must_use]
    pub fn with_namespace_minting(
        mut self,
        store: Arc<dyn NamespaceStore>,
        policy: AutoCreate,
    ) -> Self {
        let minter = NamespaceMinter::new(store, policy);
        let minter = match &self.cluster_publisher {
            Some(publisher) => minter.with_cluster_publisher(publisher.clone()),
            None => minter,
        };
        self.minter = Some(minter);
        self
    }

    /// Thread the boot's namespace-mint routing context into the registry's
    /// minter, so a worker registering for a namespace whose registry shard this
    /// node does not own mints through the shard's owner instead of being
    /// refused `NotOwner` forever.
    ///
    /// The SECOND of the two minter construction sites (the first is
    /// [`ServerState::namespace_minter`](crate::ServerState::namespace_minter),
    /// which serves the gRPC and HTTP start seams). Called after
    /// [`Self::with_namespace_minting`] on the boot path; a no-op when no minter
    /// is installed, and never called at all off-cluster, so default/test
    /// registries stay byte-identical.
    #[must_use]
    pub fn with_namespace_routing(mut self, routing: crate::namespace::NamespaceRouting) -> Self {
        self.minter = self.minter.map(|minter| minter.with_routing(routing));
        self
    }

    /// Authorize a worker registration and insert it into the connected-worker registry.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError`] if namespace authorization fails or the registry lock is poisoned.
    pub async fn accept_registration(
        &self,
        guard: &NamespaceGuard,
        caller: &CallerIdentity,
        registration: &ProtoRegisterWorker,
        sender: WorkerTaskSender,
    ) -> Result<WorkerRegistration, ServerError> {
        self.admit_delivery(
            guard,
            caller,
            registration,
            WorkerDelivery::Grpc(sender),
            InterventionCapabilities::none(),
        )
        .await
    }

    /// The ONE admission every transport's registration passes through, in
    /// this order: the guard's worker-registration policy, then EACH
    /// namespace in the worker's set authorized against the caller, then the
    /// auth-scoped mint-or-gate, then the placement-admission gate against
    /// the worker's advertised node, then the insert. The delivery leg is the
    /// only thing a transport contributes.
    ///
    /// It exists because the liminal transport used to insert into the
    /// registry directly — `register_delivery` from its
    /// connection callback — so a worker dialling the liminal listener was
    /// registered into any namespace it named with none of the gates above,
    /// while the identical worker over gRPC was refused. Two entry points,
    /// one set of rules: the rules live here, and a transport that bypasses
    /// them has to be written to do so on purpose.
    ///
    /// # Errors
    ///
    /// Returns the guard's namespace denial, the mint-or-gate refusal, the
    /// placement refusal, or the registry's own insert error, exactly as the
    /// gRPC path always has.
    pub async fn admit_delivery(
        &self,
        guard: &NamespaceGuard,
        caller: &CallerIdentity,
        registration: &ProtoRegisterWorker,
        delivery: WorkerDelivery,
        intervention_capabilities: InterventionCapabilities,
    ) -> Result<WorkerRegistration, ServerError> {
        // Verify the operation against the guard's worker-registration policy,
        // then authorize EACH namespace in the worker's set: a worker serves a
        // SET of correctness boundaries, so the registration is denied unless
        // the caller is granted every one. The wire's empty `node` carries no
        // locality affinity; a non-empty value is the worker's advertised node.
        guard
            .scope(caller, &NamespaceOperation::register_worker(registration))
            .await?;
        let namespaces = guard.scope_worker_namespaces(caller, &registration.namespaces)?;
        // MINT HOOK (Control-Plane Phase 1). This runs strictly AFTER the
        // per-namespace authorization above (`scope` + `scope_worker_namespaces`),
        // so it can only ever mint a namespace the caller is already authorized
        // for — the mint is auth-scoped by construction (CVE-2025-14986: open
        // minting and namespace isolation only coexist when minting is
        // auth-gated). It runs BEFORE the worker is inserted, so a `closed`
        // rejection never leaves a half-registered worker behind.
        self.mint_or_gate_namespaces(&namespaces).await?;
        let node = optional_node(&registration.node);
        // PLACEMENT-ADMISSION GATE (Control-Plane Phase 2, P2-I1). Runs strictly
        // AFTER the mint hook (so every authorized namespace has a durable record to
        // read a placement from) and with BOTH the worker's advertised `node` and
        // the full authorized namespace set in scope. It rejects the WHOLE
        // registration (Open Decision 6) when the worker's node violates any
        // `Pinned{L}` namespace it would serve, so only L-node workers ever enter a
        // pinned namespace's pool.
        self.enforce_pinned_placement(&namespaces, node.as_deref())
            .await?;
        // CAPACITY IS A DISPATCH PRECONDITION, never defaulted. Selection reads
        // this number on every dispatch, so a worker the server has no number
        // for cannot be served correctly: the server would have to invent how
        // much work the process takes, which is how a worker ends up pushed past
        // its own admission, silent, and then deregistered as lost while it is
        // working.
        //
        // How the precondition is MET differs by transport, which is why this
        // reads the delivery's own. A gRPC registration frame has the field, so
        // an absent one is a worker declining to state it and is refused here —
        // the precedent is `activities`, refused for the same reason one field
        // over. A liminal registration frame has no such field to carry, so the
        // worker registers with capacity unknown and states it on the
        // capabilities channel a frame later; unknown is not a default, and
        // selection passes such a worker over until it says.
        let max_concurrency = advertised_capacity(registration, delivery.transport())?;
        let instance = self
            .resolve_instance_identity(registration.instance.as_ref())
            .await?;
        let options = match max_concurrency {
            Some(advertised) => {
                RegistrationOptions::identified(registration.identity.clone(), advertised)
            }
            None => RegistrationOptions::identified_pending_capacity(registration.identity.clone()),
        };
        self.register_delivery(
            namespaces,
            registration.task_queue.clone(),
            node,
            registration.activity_types.iter(),
            delivery,
            options
                .with_instance(instance)
                .with_intervention_capabilities(intervention_capabilities),
        )
    }

    /// Resolve the wire's optional deployment/instance identity into the
    /// registry's own, marking whether the named deployment is known to the
    /// deployment store (or unchecked when no store is configured).
    ///
    /// # Errors
    ///
    /// Returns the deployment store's read error.
    pub async fn resolve_instance_identity(
        &self,
        instance: Option<&aion_proto::ProtoWorkerInstanceIdentity>,
    ) -> Result<Option<WorkerInstanceIdentity>, ServerError> {
        let Some(instance) = instance else {
            return Ok(None);
        };
        let association = match &self.deployment_store {
            Some(store) => {
                if store
                    .get_worker_deployment(&instance.deployment)
                    .await
                    .map_err(ServerError::from)?
                    .is_some()
                {
                    DeploymentAssociation::Known
                } else {
                    DeploymentAssociation::Absent
                }
            }
            None => DeploymentAssociation::Unchecked,
        };
        Ok(Some(WorkerInstanceIdentity {
            deployment: instance.deployment.clone(),
            instance_id: instance.instance_id.clone(),
            association,
        }))
    }

    /// Apply the minted-on-use policy to an already-authorized namespace set.
    ///
    /// A no-op when no minter is installed (every default/test registry), so
    /// registration stays byte-identical. With a minter, the work is delegated
    /// to the shared [`NamespaceMinter::mint_or_gate`] — the single
    /// transport-agnostic implementation reused by the workflow-start safety net
    /// — with [`NamespaceOrigin::WorkerMint`] so a first mint is attributed to
    /// worker registration. See that method for the open/closed policy, the
    /// idempotent "namespace created" event, and the retryable `NotOwner`
    /// surface.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::StoreBackend`] if a durable upsert/lookup fails
    /// (including a retryable `NotOwner` fence), or [`ServerError::Namespace`]
    /// when `closed` rejects an unknown namespace.
    async fn mint_or_gate_namespaces(&self, namespaces: &[String]) -> Result<(), ServerError> {
        let Some(minter) = &self.minter else {
            return Ok(());
        };
        minter
            .mint_or_gate(namespaces, NamespaceOrigin::WorkerMint)
            .await
    }

    /// Reject the whole registration when the worker's advertised `node` violates
    /// any `Pinned{L}` namespace it would serve (Control-Plane Phase 2, P2-I1).
    ///
    /// For each authorized namespace whose placement is [`NamespacePlacement::Pinned`],
    /// the worker's advertised `node` must be `Some(n)` with `n ∈ L`; a `None` node
    /// or an `n ∉ L` is a loud, whole-registration rejection naming the namespace,
    /// the node, and the required set. This guarantees only L-node workers ever
    /// serve a hard-pinned namespace's pool, which is exactly what lets the
    /// `Some(N ∉ L)` composition case (§2.2) resolve to the correct isolation stall
    /// at dispatch rather than needing a start-time enumeration of future nodes.
    ///
    /// Non-`Pinned` placements ([`NamespacePlacement::Unplaced`]/[`NamespacePlacement::Prefer`])
    /// are UNAFFECTED — byte-identical registration. A no-op when no minter is
    /// installed (every default/test registry), so those stay behaviour-identical:
    /// the gate reads placement from the SAME registry record the minter/placement
    /// endpoint writes, never a second source of truth.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::Namespace`] (placement-admission denial) when the
    /// worker's node violates a `Pinned` namespace, or [`ServerError::StoreBackend`]
    /// if a placement read fails at the backend.
    async fn enforce_pinned_placement(
        &self,
        namespaces: &[String],
        node: Option<&str>,
    ) -> Result<(), ServerError> {
        let Some(minter) = &self.minter else {
            return Ok(());
        };
        for namespace in namespaces {
            let NamespacePlacement::Pinned { nodes } = minter.placement_of(namespace).await? else {
                continue;
            };
            let admitted = node.is_some_and(|n| nodes.contains(n));
            if !admitted {
                return Err(ServerError::placement_admission_denied(
                    namespace, node, &nodes,
                ));
            }
        }
        Ok(())
    }

    /// Insert an already-authorized worker stream into the default task queue of
    /// a single `namespace`, with no node affinity.
    ///
    /// Convenience over [`Self::register_namespaces`] for callers that serve one
    /// namespace and do not select a task queue (notably tests of the default
    /// pool).
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn register<'a>(
        &self,
        namespace: impl Into<String>,
        activity_types: impl IntoIterator<Item = &'a String>,
        sender: WorkerTaskSender,
        max_concurrency: u32,
    ) -> Result<WorkerRegistration, ServerError> {
        self.register_namespaces(
            [namespace.into()],
            String::from(DEFAULT_TASK_QUEUE),
            None,
            activity_types,
            sender,
            max_concurrency,
        )
    }

    /// Insert an already-authorized worker stream into one explicit worker pool
    /// (single namespace + task queue), with no node affinity.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn register_pool<'a>(
        &self,
        pool: PoolAddress,
        activity_types: impl IntoIterator<Item = &'a String>,
        sender: WorkerTaskSender,
        max_concurrency: u32,
    ) -> Result<WorkerRegistration, ServerError> {
        let PoolAddress {
            namespace,
            task_queue,
        } = pool;
        self.register_namespaces(
            [namespace],
            task_queue,
            None,
            activity_types,
            sender,
            max_concurrency,
        )
    }

    /// Insert an already-authorized worker stream serving a SET of namespaces
    /// under one `task_queue`, with an optional `node` locality affinity.
    ///
    /// The worker is indexed under one `(namespace, task_queue, activity_type)`
    /// key per namespace in its set, so a dispatch in any of those namespaces
    /// can reach it. `node` is recorded on the handle and used only as a
    /// within-pool filter at selection time — it is NOT part of [`PoolAddress`].
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn register_namespaces<'a>(
        &self,
        namespaces: impl IntoIterator<Item = String>,
        task_queue: impl Into<String>,
        node: Option<String>,
        activity_types: impl IntoIterator<Item = &'a String>,
        sender: WorkerTaskSender,
        max_concurrency: u32,
    ) -> Result<WorkerRegistration, ServerError> {
        // A sender-only registration carries no identity: the callers of this
        // façade (built-in and test workers) never sent one, and a lease that
        // names such a worker says so with the empty string rather than a
        // name nobody registered. Capacity is NOT treated that way — it is
        // required here too, because selection acts on it and a stand-in value
        // would decide how much work these workers are handed.
        self.register_delivery(
            namespaces,
            task_queue,
            node,
            activity_types,
            WorkerDelivery::Grpc(sender),
            RegistrationOptions::identified(String::new(), max_concurrency),
        )
    }

    /// Insert an already-authorized worker serving a SET of namespaces under one
    /// `task_queue` and optional `node`, delivered to through an explicit
    /// [`WorkerDelivery`] transport, carrying the [`RegistrationOptions`] it
    /// registered with.
    ///
    /// This is the one transport-agnostic registration: [`Self::register_namespaces`]
    /// is the gRPC façade over it (it wraps the stream sender in
    /// [`WorkerDelivery::Grpc`]). Selection (`select_worker`/`workers_for`) is
    /// identical across transports; only the held delivery differs.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn register_delivery<'a>(
        &self,
        namespaces: impl IntoIterator<Item = String>,
        task_queue: impl Into<String>,
        node: Option<String>,
        activity_types: impl IntoIterator<Item = &'a String>,
        delivery: WorkerDelivery,
        options: RegistrationOptions,
    ) -> Result<WorkerRegistration, ServerError> {
        // THE ONE ZERO-CAPACITY CHECK. Every transport's admission and every
        // in-process façade funnels here, so a worker advertising no capacity
        // is refused once, by name, instead of being registered into a pool
        // selection would then always skip.
        let max_concurrency = options
            .max_concurrency
            .map(|advertised| {
                NonZeroU32::new(advertised).ok_or_else(|| ServerError::Wire {
                    wire: aion_proto::WireError::backend(
                        "worker registration advertises max_concurrency 0; a worker that runs no \
                         activities could never be selected for a dispatch",
                    ),
                })
            })
            .transpose()?;
        let namespaces = namespaces.into_iter().collect::<BTreeSet<_>>();
        let task_queue = task_queue.into();
        let activity_types = activity_types.into_iter().cloned().collect::<BTreeSet<_>>();
        let mut state = self.state()?;
        let worker_id = WorkerId(state.next_worker_id);
        state.next_worker_id = state.next_worker_id.saturating_add(1);

        // Capture the node affinity for the WS3 WorkerConnected delta before the
        // handle moves it.
        let node_for_event = node.clone();
        let instance_for_event = options.instance.clone();
        let handle = WorkerHandle {
            id: worker_id,
            identity: options.identity,
            namespaces: namespaces.clone(),
            task_queue: task_queue.clone(),
            node,
            activity_types: activity_types.clone(),
            instance: options.instance,
            max_concurrency,
            delivery,
            intervention_capabilities: options.intervention_capabilities,
        };

        for namespace in &namespaces {
            let pool = PoolAddress::new(namespace.clone(), task_queue.clone());
            for activity_type in &activity_types {
                let key = ActivityKey::new(pool.clone(), activity_type.clone());
                // This address is served again, so its departure record has no
                // remaining meaning — drop it in lockstep with the insert. The
                // departure map is therefore bounded by the addresses real
                // workers have served and left, never by caller-supplied
                // dispatch strings (which never create an entry at all).
                if let Some(by_node) = state.last_departure.get_mut(&key) {
                    by_node.remove(&handle.node);
                    if by_node.is_empty() {
                        state.last_departure.remove(&key);
                    }
                }
                state
                    .by_activity
                    .entry(key)
                    .or_default()
                    .insert(worker_id, handle.clone());
            }
        }
        let transport = handle.delivery.transport();
        state.workers.insert(worker_id, handle);
        drop(state);

        if let Some(metrics) = &self.metrics {
            for namespace in &namespaces {
                metrics.worker_connected(namespace);
            }
        }

        // WS3: one WorkerConnected delta carrying the full namespace set (the
        // event is namespace-list-valued; the deploy-scoped cluster channel sees
        // it whole). Edge-triggered by the real insert, never a poll.
        if let Some(publisher) = &self.cluster_publisher {
            let namespaces_vec: Vec<String> = namespaces.iter().cloned().collect();
            let task_queue_owned = task_queue.clone();
            drop(publisher.emit(|meta| {
                ClusterEvent::WorkerConnected {
                    meta,
                    worker_id: worker_id.value().to_string(),
                    namespaces: namespaces_vec,
                    task_queue: task_queue_owned,
                    transport,
                    node: node_for_event,
                    deployment: instance_for_event
                        .as_ref()
                        .map(|identity| identity.deployment.clone()),
                    deployment_association: instance_for_event.map(|identity| identity.association),
                }
            }));
        }

        self.worker_arrived.notify_waiters();

        Ok(WorkerRegistration {
            registry: self.clone(),
            parts: Some(WorkerRegistrationParts {
                worker_id,
                namespaces,
                task_queue,
                activity_types,
            }),
        })
    }

    /// Subscribe to the next change in what dispatch selection can see: a new
    /// worker registers, or a reachability verdict is published
    /// ([`Self::set_dispatch_ineligible`]). Both are ways a pool that had no
    /// selectable worker gains one, and a wait that only woke on the first would
    /// sleep through a pool whose workers are all excluded — those workers are
    /// already registered, so no registration is coming for them.
    ///
    /// **Take the subscription BEFORE you read the registry, and await it only
    /// after the read has missed.** That ordering is the whole contract; see
    /// [`WorkerArrival`] for why a subscription taken after the read loses the
    /// arrival that landed during it.
    ///
    /// Callers must re-check the registry after waking: the newly arrived worker
    /// may not serve the namespace or activity type the caller needs, and a
    /// republished verdict may have restored nobody.
    ///
    /// The returned value carries its own `#[must_use]` message: a subscription
    /// constructed and dropped is a subscription thrown away.
    pub fn worker_arrival(&self) -> WorkerArrival {
        let mut notified = Box::pin(Arc::clone(&self.worker_arrived).notified_owned());
        // `enable` reports whether the wake had ALREADY landed. There is nothing
        // to do with that answer here either way: the future is fused, so an
        // already-notified subscription simply returns `Ready` on its first
        // poll, which IS the retention this type exists to provide. What the
        // call is for is its other half — putting the waiter in the list now,
        // at a defined point, rather than at whenever the caller first polls.
        notified.as_mut().enable();
        WorkerArrival { notified }
    }

    /// Return a snapshot of the DISPATCH-ELIGIBLE workers registered for the
    /// `(namespace, task_queue, activity_type)` pool, ordered by worker id and
    /// then rotated so each call starts from the next worker in the pool. The
    /// rotation cursor is per triple, so each pool round-robins independently.
    ///
    /// When `node` is `Some`, the result is filtered to workers whose advertised
    /// node equals it — a dispatch pinned to a node reaches only workers on that
    /// node (NODE affinity = require). When `node` is `None`, the behaviour is
    /// exactly the unpinned pool: every worker in the `(namespace, task_queue)`
    /// pool is a candidate regardless of locality. node is a within-pool filter,
    /// NOT part of the pool key, so the per-triple rotation cursor is shared
    /// across pinned and unpinned lookups of the same pool.
    ///
    /// Candidates, ordering, eligibility and the cursor all come from
    /// [`eligible_candidates_in_rotation`], which [`Self::select_and_reserve`]
    /// reads too: ONE derivation and ONE cursor, so the gRPC push dispatcher and every
    /// other dispatch path rotate over the same workers in the same order and
    /// cannot drift about who is dispatchable or whose turn it is.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn workers_for(
        &self,
        namespace: &str,
        task_queue: &str,
        activity_type: &str,
        node: Option<&str>,
    ) -> Result<Vec<WorkerHandle>, ServerError> {
        let mut state = self.state()?;
        let key = ActivityKey::new(PoolAddress::new(namespace, task_queue), activity_type);
        Ok(eligible_candidates_in_rotation(&mut state, key, node))
    }

    /// Return a snapshot of every connected worker stream.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn all_workers(&self) -> Result<Vec<WorkerHandle>, ServerError> {
        let state = self.state()?;
        Ok(state.workers.values().cloned().collect())
    }

    /// Return the handle for a worker by id, or `None` when it is not registered.
    ///
    /// The intervention router resolves the owning worker of a target attempt by id
    /// (NOI-6): the attempt-owner back-index stores a [`WorkerId`], and the router
    /// reads back the live handle to gate on its advertised capabilities and select
    /// its delivery. A `None` result means the owner disconnected — the router
    /// treats that as the attempt-scoped no-op.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn worker_by_id(&self, worker_id: WorkerId) -> Result<Option<WorkerHandle>, ServerError> {
        Ok(self.state()?.workers.get(&worker_id).cloned())
    }

    /// Replace the advertised intervention capabilities of a registered worker
    /// (NOI-6). The liminal registration frame cannot carry capabilities, so a
    /// liminal agent worker announces them on the reserved capabilities channel
    /// right after registering, and this applies the announcement to the live
    /// handle the intervention router gates on. Returns `false` when the worker
    /// is no longer registered (a disconnect racing the announcement — benign).
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn set_intervention_capabilities(
        &self,
        worker_id: WorkerId,
        capabilities: &InterventionCapabilities,
    ) -> Result<bool, ServerError> {
        let mut state = self.state()?;
        if !state.workers.contains_key(&worker_id) {
            return Ok(false);
        }
        if let Some(handle) = state.workers.get_mut(&worker_id) {
            handle.intervention_capabilities = capabilities.clone();
        }
        // The selection index holds handle clones; keep them capability-consistent
        // even though capabilities are never a routing dimension.
        for workers in state.by_activity.values_mut() {
            if let Some(handle) = workers.get_mut(&worker_id) {
                handle.intervention_capabilities = capabilities.clone();
            }
        }
        Ok(true)
    }

    /// Broadcast a graceful drain request to every connected worker stream.
    /// Workers are removed from routing before any transport signal is attempted.
    /// A liminal worker has no drain control frame, so it is force-fenced and
    /// deregistered with an error-level identity-bearing log instead of being
    /// silently skipped.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn broadcast_drain(&self) -> Result<usize, ServerError> {
        let workers = self.all_workers()?;
        let mut delivered = 0usize;
        for worker in workers {
            if self.drain_worker(worker.id())? {
                delivered = delivered.saturating_add(1);
            }
        }
        Ok(delivered)
    }

    /// Stop assigning work to one worker and request a graceful transport drain.
    ///
    /// Returns `false` if the worker is not registered or if its transport has no
    /// drain channel. In the latter case the worker is deregistered immediately,
    /// after an error-level log names the worker and its transport limitation.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn drain_worker(&self, worker_id: WorkerId) -> Result<bool, ServerError> {
        let worker = {
            let mut state = self.state()?;
            let Some(worker) = state.workers.get(&worker_id).cloned() else {
                return Ok(false);
            };
            Self::remove_worker_from_service(&mut state, &worker);
            worker
        };
        match worker.delivery() {
            WorkerDelivery::Grpc(sender) => {
                if sender.try_send(WorkerMessage::DrainRequest).is_ok() {
                    tracing::info!(worker_id = worker_id.value(), "worker drain requested");
                    Ok(true)
                } else {
                    tracing::error!(
                        worker_id = worker_id.value(),
                        "worker drain signal failed; force-deregistering closed transport"
                    );
                    self.deregister(worker_id)?;
                    Ok(false)
                }
            }
            #[cfg(feature = "liminal-transport")]
            WorkerDelivery::Liminal(delivery) => {
                tracing::error!(
                    worker_id = worker_id.value(),
                    connection_pid = delivery.pid(),
                    "liminal transport has no drain control channel; worker fenced and \
                     deregistered at drain start"
                );
                self.deregister(worker_id)?;
                Ok(false)
            }
        }
    }

    /// Publish the liveness probe's reachability verdict: the set of workers the
    /// server cannot currently reach on the push leg, which
    /// [`eligible_candidates_in_rotation`] then skips for BOTH selectors.
    ///
    /// Replaces the whole set rather than toggling one worker, so the published
    /// verdict is always exactly one round's evidence and a worker can never be
    /// left excluded by a stale entry nobody cleared.
    ///
    /// Publishing WAKES the selection wait ([`WorkerArrival`]). A dispatch that
    /// found no eligible worker is blocked on this verdict every bit as squarely
    /// as on a registration, and the workers it needs are already registered —
    /// so a park that only woke on registrations would sleep through their
    /// recovery. A publication landing between a dispatch's census and its park
    /// is retained, because the dispatch holds a [`WorkerArrival`] taken before
    /// the census; this method owes nothing to that ordering beyond firing.
    ///
    /// The wake is unconditional rather than gated on the set having shrunk,
    /// because a change-gated wake would make correctness depend on this method
    /// judging what "changed" means for a caller it cannot see — a set that
    /// shrank for a worker in some other pool is no restoration for THIS
    /// dispatch, and a set republished identically may still coincide with the
    /// registration that serves it. Every round publishes, so waking on each one
    /// is self-healing at the probe's own cadence — it invents no clock of its
    /// own — and a publication with nobody parked costs one waiterless
    /// `notify_waiters`.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn set_dispatch_ineligible(
        &self,
        unreachable: BTreeMap<WorkerId, DispatchExclusion>,
    ) -> Result<(), ServerError> {
        self.state()?.dispatch_ineligible = unreachable;
        self.worker_arrived.notify_waiters();
        Ok(())
    }

    /// The transport each named worker is delivered over, for the workers still
    /// registered (#25).
    ///
    /// A worker absent from the result has LEFT the registry — the caller must
    /// treat that as "no transport", never as a default one. That distinction is
    /// the whole reason this returns a map rather than a vector in the caller's
    /// order: the liveness probe scopes its verdict by transport coverage, and a
    /// departed worker whose transport was guessed would be judged by a probe
    /// that never had a wire to it.
    ///
    /// Read under ONE lock acquisition rather than one per worker, so the
    /// answer describes a single registry state instead of a smear across a
    /// round of registrations and disconnects.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn transports_of(
        &self,
        workers: impl IntoIterator<Item = WorkerId>,
    ) -> Result<BTreeMap<WorkerId, WorkerTransport>, ServerError> {
        let state = self.state()?;
        Ok(workers
            .into_iter()
            .filter_map(|worker_id| {
                state
                    .workers
                    .get(&worker_id)
                    .map(|worker| (worker_id, worker.delivery.transport()))
            })
            .collect())
    }

    /// Whether a worker is currently excluded from dispatch selection for
    /// unreachability.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn is_dispatch_ineligible(&self, worker_id: WorkerId) -> Result<bool, ServerError> {
        Ok(self.state()?.dispatch_ineligible.contains_key(&worker_id))
    }

    /// Every gRPC-delivered worker the liveness probe must ping this round,
    /// paired with the stream sender the ping rides (#197).
    ///
    /// The liminal half of the same census is
    /// [`LiminalConnectionNotifier::liveness_targets`](crate::worker::LiminalConnectionNotifier::liveness_targets),
    /// which enumerates CONNECTIONS. This one enumerates REGISTRATIONS, because
    /// a gRPC worker's only server-side identity is its registry handle: the
    /// stream is owned by a tonic task and reachable solely through the sender
    /// the registration carries.
    ///
    /// Deliberately not filtered by current eligibility. A worker excluded from
    /// dispatch is precisely the worker whose next answered ping restores it,
    /// so skipping the excluded set would make exclusion permanent — the exact
    /// defect this lane exists to remove.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn grpc_liveness_targets(
        &self,
    ) -> Result<Vec<super::grpc_liveness::GrpcLivenessTarget>, ServerError> {
        Ok(self
            .state()?
            .workers
            .values()
            .filter_map(|worker| match &worker.delivery {
                WorkerDelivery::Grpc(sender) => Some(super::grpc_liveness::GrpcLivenessTarget {
                    worker_id: worker.id,
                    sender: sender.clone(),
                }),
                #[cfg(feature = "liminal-transport")]
                WorkerDelivery::Liminal(_) => None,
            })
            .collect())
    }

    /// The currently published exclusion set, read as a whole.
    ///
    /// The liveness probe reads this BEFORE publishing a round's verdict, so it
    /// can announce the workers that just LEFT the set. Without the whole
    /// previous set there is no way to name a recovery: per-worker queries can
    /// only be asked about workers the new verdict already mentions, and a
    /// recovered worker is precisely the one it does not.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn dispatch_ineligible(
        &self,
    ) -> Result<BTreeMap<WorkerId, DispatchExclusion>, ServerError> {
        Ok(self.state()?.dispatch_ineligible.clone())
    }

    /// Return whether a worker stream is currently registered.
    ///
    /// The activity dispatch path uses this after queuing a task to detect a
    /// worker whose stream tore down concurrently: a sweep that ran before
    /// the dispatch tracked its task can never complete it, so the dispatch
    /// must fail the activity itself instead of waiting forever.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn is_registered(&self, worker_id: WorkerId) -> Result<bool, ServerError> {
        Ok(self.state()?.workers.contains_key(&worker_id))
    }

    /// Remove a worker by id from every namespace/activity index it advertised.
    ///
    /// Emits a WS3 [`WorkerDeathReason::Disconnect`] delta — the truthful default
    /// for a removed worker whose stream/registration went away. Callers that can
    /// PROVE a finer reason (a liveness-timeout sweep) call
    /// [`Self::deregister_with_reason`] instead, so the dashboard never sees a
    /// fabricated distinction.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn deregister(&self, worker_id: WorkerId) -> Result<(), ServerError> {
        self.deregister_with_reason(worker_id, WorkerDeathReason::Disconnect)
    }

    /// Remove a worker by id, attributing the departure to an explicit
    /// [`WorkerDeathReason`] the caller can prove at its call site (for example a
    /// heartbeat sweep passes [`WorkerDeathReason::Timeout`]).
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn deregister_with_reason(
        &self,
        worker_id: WorkerId,
        reason: WorkerDeathReason,
    ) -> Result<(), ServerError> {
        let mut state = self.state()?;
        let removed_namespaces = Self::remove_worker(&mut state, worker_id);
        drop(state);

        let Some(namespaces) = removed_namespaces else {
            // Already gone: no metrics double-count, no duplicate delta.
            return Ok(());
        };

        if let Some(metrics) = &self.metrics {
            for namespace in &namespaces {
                metrics.worker_disconnected(namespace);
            }
        }
        self.emit_worker_disconnected(worker_id, &namespaces, reason);

        Ok(())
    }

    /// Emit a WS3 `WorkerDisconnected` delta if a publisher is attached.
    fn emit_worker_disconnected(
        &self,
        worker_id: WorkerId,
        namespaces: &BTreeSet<String>,
        reason: WorkerDeathReason,
    ) {
        if let Some(publisher) = &self.cluster_publisher {
            let namespaces_vec: Vec<String> = namespaces.iter().cloned().collect();
            drop(publisher.emit(|meta| ClusterEvent::WorkerDisconnected {
                meta,
                worker_id: worker_id.value().to_string(),
                namespaces: namespaces_vec,
                reason,
            }));
        }
    }

    /// Remove a worker from every `(namespace, task_queue, activity_type)` index
    /// it advertised. Returns the namespace set it served (for metrics), or
    /// `None` if the worker was already gone.
    fn remove_worker(state: &mut RegistryState, worker_id: WorkerId) -> Option<BTreeSet<String>> {
        let handle = state.workers.remove(&worker_id)?;
        Self::remove_worker_from_service(state, &handle);
        Some(handle.namespaces)
    }

    /// Fence a worker from every routing index while retaining its connection
    /// handle so an in-progress graceful drain can finish and tear down normally.
    fn remove_worker_from_service(state: &mut RegistryState, handle: &WorkerHandle) {
        // The in-flight projection goes with the registration. A departing
        // worker's tasks are swept by the tracker on the same paths that call
        // this, and a count left behind would keep a worker id that no longer
        // exists occupying capacity nothing can free.
        state.in_flight.remove(&handle.id);
        let departed_at = Instant::now();
        for namespace in &handle.namespaces {
            let pool = PoolAddress::new(namespace.clone(), handle.task_queue.clone());
            for activity_type in &handle.activity_types {
                let key = ActivityKey::new(pool.clone(), activity_type.clone());
                // Departure time is recorded for EVERY address this worker
                // served, so a later census can age the last compatible poller
                // instead of reporting a bare "nobody" (R1).
                state
                    .last_departure
                    .entry(key.clone())
                    .or_default()
                    .insert(handle.node.clone(), departed_at);
                if let Some(workers) = state.by_activity.get_mut(&key) {
                    workers.remove(&handle.id);
                    if workers.is_empty() {
                        state.by_activity.remove(&key);
                        // Prune the round-robin cursor in lockstep: the cursor
                        // map is keyed on arbitrary caller-supplied strings and
                        // is lazily created by `workers_for`, so leaving stale
                        // entries behind leaks memory unboundedly on a
                        // never-dying server. When the last worker for a triple
                        // leaves, its cursor has no remaining meaning.
                        state.rotation.remove(&key);
                    }
                }
            }
        }
    }

    fn state(&self) -> Result<MutexGuard<'_, RegistryState>, ServerError> {
        self.inner
            .lock()
            .map_err(|_| ServerError::lock_poisoned("connected worker registry"))
    }
}

/// Normalize a wire `node` string into an optional locality affinity: an empty
/// value (the proto3 default) carries no node, anything else is the worker's
/// advertised node id.
///
/// Shared with contract admission ([`super::contracts::validate_worker_contracts`])
/// rather than restated there: admission decides which of a package's actions a
/// connection owes from the same locality this registry then routes by, and two
/// independent normalizations of the wire default would be a place for that
/// agreement to drift silently.
pub(crate) fn optional_node(node: &str) -> Option<String> {
    if node.is_empty() {
        None
    } else {
        Some(node.to_owned())
    }
}

#[derive(Clone, Debug)]
struct WorkerRegistrationParts {
    worker_id: WorkerId,
    namespaces: BTreeSet<String>,
    task_queue: String,
    activity_types: BTreeSet<String>,
}

/// Registration token owned by the worker stream task.
///
/// Dropping the token performs best-effort cleanup for disconnect paths. Call
/// [`WorkerRegistration::deregister`] when the caller needs a typed poison error.
#[derive(Debug)]
pub struct WorkerRegistration {
    registry: ConnectedWorkerRegistry,
    parts: Option<WorkerRegistrationParts>,
}

impl WorkerRegistration {
    /// Worker id assigned to this registration.
    #[must_use]
    pub fn worker_id(&self) -> Option<WorkerId> {
        self.parts.as_ref().map(|parts| parts.worker_id)
    }

    /// Authorized namespace set for this registration.
    #[must_use]
    pub fn namespaces(&self) -> Option<&BTreeSet<String>> {
        self.parts.as_ref().map(|parts| &parts.namespaces)
    }

    /// Task queue (pool/flavour) this registration serves within each namespace.
    #[must_use]
    pub fn task_queue(&self) -> Option<&str> {
        self.parts.as_ref().map(|parts| parts.task_queue.as_str())
    }

    /// Activity types advertised by this registration.
    #[must_use]
    pub fn activity_types(&self) -> Option<&BTreeSet<String>> {
        self.parts.as_ref().map(|parts| &parts.activity_types)
    }

    /// Explicitly remove this worker from the registry.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the registry lock is poisoned.
    pub fn deregister(mut self) -> Result<(), ServerError> {
        let Some(parts) = self.parts.take() else {
            return Ok(());
        };
        self.registry.deregister(parts.worker_id)
    }
}

impl Drop for WorkerRegistration {
    fn drop(&mut self) {
        let Some(parts) = self.parts.take() else {
            return;
        };
        let removed_namespaces = self.registry.inner.lock().ok().and_then(|mut state| {
            ConnectedWorkerRegistry::remove_worker(&mut state, parts.worker_id)
        });
        if let Some(namespaces) = removed_namespaces {
            if let Some(metrics) = &self.registry.metrics {
                for namespace in &namespaces {
                    metrics.worker_disconnected(namespace);
                }
            }
            // A dropped registration token means the worker's stream/connection
            // went away — the truthful reason is Disconnect, not a fabricated
            // timeout/deregister distinction this path cannot prove.
            self.registry.emit_worker_disconnected(
                parts.worker_id,
                &namespaces,
                WorkerDeathReason::Disconnect,
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use crate::config::NamespaceMode;
    use crate::namespace::{NamespaceResolver, StaticScheduleNamespaces, StaticWorkflowNamespaces};
    use crate::worker::heartbeat::{DISPATCH_PROBATION_PINGS, HeartbeatTracker};

    use super::*;

    fn guard() -> NamespaceGuard {
        NamespaceGuard::new(NamespaceResolver::authorization_only(
            NamespaceMode::SharedEngine,
            StaticWorkflowNamespaces::default(),
            StaticScheduleNamespaces::default(),
        ))
    }

    fn caller(namespace: &str) -> CallerIdentity {
        CallerIdentity::new("worker", [namespace.to_owned()])
    }

    /// A test-expectation failure as a `ServerError`, so `Result`-returning
    /// tests can fail on an unexpected `None` without `panic!`/`expect`.
    fn test_failure(message: &str) -> ServerError {
        ServerError::worker_dispatch("default".to_owned(), "test".to_owned(), message.to_owned())
    }

    /// A release gate whose party size is fixed when the gate OPENS, not when
    /// it is built.
    ///
    /// A `Barrier` has to be sized before the first racer is spawned. If the
    /// operating system then refuses one of those spawns, the racers that did
    /// start wait forever on a party that can never be complete, and the join
    /// that would have reported the refusal never runs — the test hangs
    /// burning a core instead of failing by name. Counting arrivals instead
    /// keeps the release simultaneous (nobody proceeds until every racer that
    /// started has arrived) while letting the releasing thread decide the
    /// party from the spawns it actually got.
    #[derive(Default)]
    struct StartGate {
        /// Arrival count and the open flag, under one lock so a waiter's
        /// predicate and the release are read and written atomically.
        state: Mutex<StartGateState>,
        /// Signalled on every arrival and once on release.
        change: std::sync::Condvar,
    }

    /// The state [`StartGate`] guards.
    #[derive(Default)]
    struct StartGateState {
        /// How many racers have announced themselves.
        arrived: usize,
        /// Whether the releasing thread has opened the gate.
        open: bool,
    }

    impl StartGate {
        /// Announce arrival and block until the gate opens.
        ///
        /// The arrival is counted BEFORE any early return, the poisoned one
        /// included. A racer that panics poisons the mutex; a peer that then
        /// returned without counting itself would leave the releasing thread's
        /// party permanently short, and `thread::scope` joins every racer
        /// before that thread's error can reach the caller — so the test would
        /// hang exactly where it meant to fail by name. Recovering the guard
        /// and counting first means every racer that reached this function is
        /// accounted for, whatever it does next.
        fn arrive_and_wait(&self) -> Result<(), ServerError> {
            let poisoned = self.state.is_poisoned();
            let mut state = self
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            state.arrived += 1;
            self.change.notify_all();
            if poisoned {
                return Err(poisoned_start_gate());
            }
            while !state.open {
                match self.change.wait(state) {
                    Ok(next) => state = next,
                    Err(poison) => {
                        // Release the lock before leaving, so the thread that
                        // opens the gate is not blocked behind this racer.
                        drop(poison.into_inner());
                        return Err(poisoned_start_gate());
                    }
                }
            }
            Ok(())
        }

        /// Wait until `party` racers have arrived, then release them together.
        ///
        /// Every exit opens the gate FIRST. Reporting a poisoned lock without
        /// releasing would leave the racers that did arrive parked in `wait`
        /// with `open` still false, and the scope's join would never return —
        /// the same hang, wearing the other hat. A poisoned peer also ends the
        /// wait: a racer that panicked before announcing itself is never going
        /// to arrive, so a party that counts it can never complete.
        fn open_for(&self, party: usize) -> Result<(), ServerError> {
            let mut poisoned = self.state.is_poisoned();
            let mut state = self
                .state
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            while !poisoned && state.arrived < party {
                match self.change.wait(state) {
                    Ok(next) => state = next,
                    Err(poison) => {
                        state = poison.into_inner();
                        poisoned = true;
                    }
                }
            }
            state.open = true;
            self.change.notify_all();
            drop(state);
            if poisoned {
                return Err(poisoned_start_gate());
            }
            Ok(())
        }
    }

    /// The failure a poisoned start gate is reported as: a racer panicked
    /// while holding the lock, which is a test failure in its own right and
    /// must not be collapsed into "no reservation won".
    fn poisoned_start_gate() -> ServerError {
        test_failure("the racing selection start gate was poisoned by a panicking racer")
    }

    fn registration(namespace: &str, activity_types: &[&str]) -> ProtoRegisterWorker {
        registration_with_queue(namespace, "", activity_types)
    }

    fn registration_with_queue(
        namespace: &str,
        task_queue: &str,
        activity_types: &[&str],
    ) -> ProtoRegisterWorker {
        registration_full(&[namespace], task_queue, "", activity_types)
    }

    fn registration_full(
        namespaces: &[&str],
        task_queue: &str,
        node: &str,
        activity_types: &[&str],
    ) -> ProtoRegisterWorker {
        ProtoRegisterWorker {
            namespaces: namespaces.iter().map(|value| (*value).to_owned()).collect(),
            activity_types: activity_types
                .iter()
                .map(|value| (*value).to_owned())
                .collect(),
            task_queue: task_queue.to_owned(),
            node: node.to_owned(),
            activities: Vec::new(),
            identity: WIRE_IDENTITY.to_owned(),
            instance: None,
            max_concurrency: Some(4),
        }
    }

    /// A registration that says nothing about its capacity is REFUSED, and the
    /// refusal names the field.
    ///
    /// There is no lenient path and no default here on purpose. The server
    /// counts a worker's in-flight dispatches against this number on every
    /// selection, so a registration without it could only be served by the
    /// server inventing how much work the process takes — which is how a worker
    /// ends up pushed past its own admission, silent, and then deregistered as
    /// lost while it works. The in-tree precedent is one field over:
    /// `activities` empty "predates contract commitment and is refused".
    #[tokio::test]
    async fn a_registration_without_capacity_is_refused_by_name() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(1);
        let mut frame = registration("tenant-a", &["charge"]);
        frame.max_concurrency = None;

        let refusal = registry
            .accept_registration(&guard(), &caller("tenant-a"), &frame, sender)
            .await
            .err()
            .ok_or_else(|| test_failure("a capacity-less registration must be refused"))?;
        assert!(
            refusal.to_string().contains("max_concurrency"),
            "the refusal must NAME the field a worker has to fix: {refusal}"
        );
        assert!(
            registry
                .workers_for("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .is_empty(),
            "a refused registration must leave nothing behind in the pool"
        );
        Ok(())
    }

    /// A registration advertising ZERO capacity is refused too, and with its own
    /// words: a worker that runs nothing could never be selected, so accepting
    /// it would register a worker that exists only to be skipped.
    #[tokio::test]
    async fn a_registration_advertising_zero_capacity_is_refused() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(1);
        let mut frame = registration("tenant-a", &["charge"]);
        frame.max_concurrency = Some(0);

        let refusal = registry
            .accept_registration(&guard(), &caller("tenant-a"), &frame, sender)
            .await
            .err()
            .ok_or_else(|| test_failure("a zero-capacity registration must be refused"))?;
        assert!(
            refusal.to_string().contains("max_concurrency 0"),
            "the refusal must say what was WRONG with the value, not merely that one was \
             required: {refusal}"
        );
        Ok(())
    }

    /// SELECTION NEVER EXCEEDS ADVERTISED CAPACITY, and the worker comes back
    /// the moment a slot frees.
    ///
    /// The pool here is one worker advertising two slots. With both held it is
    /// not a candidate — not deregistered, not ineligible, simply not selected —
    /// which is the resting state a busy worker should have had all along. The
    /// restore half is the vacuity control: without it, a filter that excluded
    /// the worker permanently (or a registry that had lost it) would pass.
    #[tokio::test]
    async fn selection_skips_a_worker_that_has_reached_its_advertised_capacity()
    -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(4);
        let types = ["charge".to_owned()];
        let _registration = registry.register_namespaces(
            ["tenant-a".to_owned()],
            DEFAULT_TASK_QUEUE,
            None,
            types.iter(),
            sender,
            2,
        )?;
        let Some(worker) = registry
            .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
            .map(|(worker, _reservation)| worker)
        else {
            return Err(test_failure("an idle worker must be selectable"));
        };
        let worker_id = worker.id();

        registry.record_dispatch_started(worker_id)?;
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "one dispatch against a capacity of two leaves a slot"
        );

        registry.record_dispatch_started(worker_id)?;
        assert_eq!(registry.in_flight_for_worker(worker_id)?, 2);
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_none(),
            "a worker holding every slot it advertised must not be selected again: pushing past \
             its own admission is what made a working worker look silent"
        );
        // It is FULL, not gone: the census still counts it as a compatible
        // worker, so the taxonomy can say the pool is busy rather than empty.
        let census = registry.pool_census("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?;
        assert_eq!(census.compatible_workers, 1);
        assert_eq!(census.compatible_workers_at_capacity, 1);
        assert_eq!(census.eligible_compatible_workers, 0);

        registry.record_dispatch_finished(worker_id)?;
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "the freed slot must make the worker selectable again — without this the test above \
             would pass for a worker that had simply been lost"
        );
        Ok(())
    }

    /// A worker registered with capacity UNKNOWN is not dispatched to, and
    /// becomes dispatchable the moment it announces.
    ///
    /// This is the liminal transport's whole shape at the registry level. Its
    /// published registration frame has no capacity field, so it registers
    /// pending and states its configured number one frame later on the
    /// capabilities channel. Both halves are asserted because each alone is a
    /// trap: without the first, the server invents a number for a process that
    /// is about to state its own; without the second, a liminal worker
    /// registers, looks healthy in every listing, and is never sent any work.
    ///
    /// Driven through `register_delivery` rather than a liminal connection
    /// because the invariant is the REGISTRY's, not the transport's — a handle
    /// with no advertised capacity is unselectable whoever produced it.
    #[tokio::test]
    async fn a_worker_with_no_advertised_capacity_is_unselectable_until_it_announces()
    -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(4);
        let types = ["charge".to_owned()];
        let registration = registry.register_delivery(
            ["tenant-a".to_owned()],
            DEFAULT_TASK_QUEUE,
            None,
            types.iter(),
            WorkerDelivery::Grpc(sender),
            RegistrationOptions::identified_pending_capacity("pending-1"),
        )?;

        // REGISTERED — it is in the pool and the census can see it...
        let census = registry.pool_census("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?;
        assert_eq!(
            census.compatible_workers, 1,
            "a worker awaiting its capacity announcement is registered, not absent"
        );
        // ...but NOT dispatchable, because the server does not know what it runs.
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_none(),
            "a worker whose capacity is unknown must not be selected: the only way to dispatch \
             to it is to invent a number on its behalf, which is the defect this replaces"
        );

        // From the REGISTRATION guard, not from `workers_for`: that list is the
        // dispatch-eligible one, and the whole point of this test is that this
        // worker is registered and not yet dispatchable. Asking the eligibility
        // list for a worker it is designed to omit is how the first draft of
        // this test failed.
        let worker_id = registration
            .worker_id()
            .ok_or_else(|| test_failure("the pending worker must have a registration id"))?;
        assert!(
            registry
                .worker_by_id(worker_id)?
                .is_some_and(|handle| handle.max_concurrency().is_none()),
            "the handle must report capacity UNKNOWN rather than a stand-in number"
        );
        // The census must call it what it is. Selection excludes an unannounced
        // worker by the same test it excludes a full one, so this is exactly
        // where the two would get confused — and the operator sentence attached
        // to "at capacity" is wrong for a worker that is idle and mute.
        assert_eq!(
            census.compatible_workers_capacity_unannounced, 1,
            "an unannounced worker must be counted as unannounced"
        );
        assert_eq!(
            census.compatible_workers_at_capacity, 0,
            "an unannounced worker must NOT be counted as busy: the two conditions have \
             different remedies and the census is what the operator sentence is derived from"
        );
        assert_eq!(census.eligible_compatible_workers, 0);

        // THE ANNOUNCEMENT. Two, not one: a value the registry could also have
        // reached by defaulting would not prove the announcement was read.
        assert!(
            registry.set_advertised_capacity(worker_id, 2)?,
            "announcing capacity for a registered worker must apply"
        );
        let announced = registry.pool_census("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?;
        assert_eq!(
            announced.compatible_workers_capacity_unannounced, 0,
            "the announcement must clear the unannounced count"
        );
        assert_eq!(announced.eligible_compatible_workers, 1);
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "the announced capacity must make the worker dispatchable"
        );
        registry.record_dispatch_started(worker_id)?;
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "one dispatch against an announced capacity of two leaves a slot: the announced \
             number is what selection now counts against"
        );
        registry.record_dispatch_started(worker_id)?;
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_none(),
            "the announced capacity must bound selection exactly as a registered one does"
        );

        // A zero announcement is refused by name, so the later channel cannot
        // admit what the registration funnel rejects.
        let refusal = registry
            .set_advertised_capacity(worker_id, 0)
            .err()
            .ok_or_else(|| test_failure("a zero-capacity announcement must be refused"))?;
        assert!(
            refusal.to_string().contains("max_concurrency 0"),
            "the refusal must say what was wrong with the value: {refusal}"
        );
        Ok(())
    }

    /// A SIMULTANEOUS fan never over-subscribes the worker it lands on.
    ///
    /// This is the arm the sequential test above cannot cover. Selection has
    /// always filtered on the in-flight count; what it did not do was CLAIM the
    /// slot it selected, so the claim landed later, in the caller, after the
    /// pending entry and the lease record had been written. Every dispatch that
    /// started in that window read the same pre-dispatch count. A sequential
    /// test passes against that defect — one caller, one window, nobody to race
    /// — and the real fan of a fan-out does not.
    ///
    /// So: far more racers than slots, released together on a barrier, each
    /// HOLDING what it wins. The count that matters is how many got a
    /// reservation, and it must be exactly the advertised concurrency. A
    /// late-claiming selector gives every racer a worker and fails here loudly.
    ///
    /// The release half at the end is the vacuity control: a registry that had
    /// simply lost the worker, or a filter that excluded it permanently, would
    /// also hand out no surplus.
    ///
    /// What this test does and does not discriminate, measured rather than
    /// assumed: moving the claim a few instructions outside the selection lock
    /// SURVIVES here — 32 racers on this host never landed in a window that
    /// narrow. Moving it behind any real work does not: with the claim deferred
    /// the width of the pending insert and the lease record, this fails 3 runs
    /// out of 3, reporting all 32 racers claiming 4 slots. It is a pin on the
    /// defect at its real width, not a proof that no narrower one exists.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn simultaneous_selections_claim_no_more_slots_than_advertised() -> Result<(), ServerError>
    {
        /// Advertised concurrency of the single worker every racer targets.
        const CAPACITY: u32 = 4;
        /// Racers released at once — deliberately far more than `CAPACITY`, so
        /// a selector that claims late has many chances to over-subscribe.
        const RACERS: usize = 32;

        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(RACERS);
        let types = ["charge".to_owned()];
        let _registration = registry.register_namespaces(
            ["tenant-a".to_owned()],
            DEFAULT_TASK_QUEUE,
            None,
            types.iter(),
            sender,
            CAPACITY,
        )?;

        let gate = StartGate::default();
        let (won, refused): (
            Vec<Option<(WorkerHandle, DispatchReservation)>>,
            Option<String>,
        ) = std::thread::scope(|scope| {
            // Spawn through the fallible builder, not `Scope::spawn`: the
            // latter panics when the operating system refuses a thread,
            // and that panic is raised on THIS thread while the racers
            // that did start are already parked on a party that can never
            // complete — the scope's join then never runs and the test
            // hangs instead of naming the refusal.
            let mut handles = Vec::with_capacity(RACERS);
            let mut refused = None;
            for index in 0..RACERS {
                let registry = registry.clone();
                let gate = &gate;
                let spawned = std::thread::Builder::new()
                    .name(format!("registry-racer-{index}"))
                    .spawn_scoped(scope, move || {
                        gate.arrive_and_wait()?;
                        registry.select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)
                    });
                match spawned {
                    Ok(handle) => handles.push(handle),
                    Err(error) => {
                        refused = Some(format!(
                            "the operating system refused racing selection thread {index}: \
                                 {error}"
                        ));
                        break;
                    }
                }
            }
            // Release exactly the racers that started. Sizing the release
            // here rather than at construction is what makes a refused
            // spawn reportable: the survivors are freed, joined, and the
            // refusal is returned by name.
            gate.open_for(handles.len())?;
            let won = handles
                .into_iter()
                .map(|handle| {
                    // A racer thread that panicked would otherwise be
                    // reported as "no reservation won", which reads exactly
                    // like the correct answer. Surface it as its own
                    // failure instead.
                    handle
                        .join()
                        .map_err(|_| test_failure("a racing selection thread panicked"))?
                })
                .collect::<Result<Vec<_>, ServerError>>()?;
            Ok::<_, ServerError>((won, refused))
        })?;
        if let Some(refusal) = refused {
            return Err(test_failure(&refusal));
        }

        let granted = won.iter().filter(|slot| slot.is_some()).count();
        assert_eq!(
            granted, CAPACITY as usize,
            "{RACERS} simultaneous selections claimed {granted} slots on a worker advertising              {CAPACITY}: the choice and the claim are not one critical section, so every racer              read the same pre-dispatch count"
        );
        assert_eq!(
            registry.in_flight_for_worker(
                won.iter()
                    .flatten()
                    .next()
                    .ok_or_else(|| test_failure("no racer won a slot"))?
                    .0
                    .id()
            )?,
            CAPACITY,
            "the projection must agree with the number of reservations actually held"
        );

        drop(won);
        assert!(
            registry
                .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "released reservations must make the worker selectable again — without this the              count above would also pass for a worker that had been lost"
        );
        assert_eq!(
            registry.in_flight_for_worker(
                registry
                    .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
                    .map(|(worker, _reservation)| worker)
                    .ok_or_else(|| test_failure("the worker must still be registered"))?
                    .id()
            )?,
            0,
            "every reservation must have given its slot back on drop; a leak here shrinks the              pool silently for the life of the registration"
        );
        Ok(())
    }

    /// A freed slot WAKES a parked selection.
    ///
    /// A dispatch parked because every compatible worker is full is blocked on
    /// exactly one event and on nothing else: no worker is going to register,
    /// and no reachability verdict is going to change. A completion that
    /// corrected the count without waking would leave that dispatch asleep
    /// until something unrelated happened to fire.
    #[tokio::test]
    async fn a_freed_dispatch_slot_wakes_a_parked_selection() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(1);
        let types = ["charge".to_owned()];
        let _registration = registry.register_namespaces(
            ["tenant-a".to_owned()],
            DEFAULT_TASK_QUEUE,
            None,
            types.iter(),
            sender,
            1,
        )?;
        let Some(worker) = registry
            .select_and_reserve("tenant-a", DEFAULT_TASK_QUEUE, "charge", None)?
            .map(|(worker, _reservation)| worker)
        else {
            return Err(test_failure("an idle worker must be selectable"));
        };
        registry.record_dispatch_started(worker.id())?;

        // Subscribe BEFORE the release, exactly as a parking dispatch does.
        let arrival = registry.worker_arrival();
        registry.record_dispatch_finished(worker.id())?;
        tokio::time::timeout(std::time::Duration::from_secs(1), arrival)
            .await
            .map_err(|_| {
                test_failure(
                    "a freed dispatch slot did not wake the selection wait; a dispatch parked on \
                     a full pool would sleep through the very completion that serves it",
                )
            })?;
        Ok(())
    }

    /// The identity the wire fixture registers under, so a test can assert it
    /// reached the handle rather than being dropped at admission (WA-010 R3).
    const WIRE_IDENTITY: &str = "wire-worker-7";

    fn multi_caller(namespaces: &[&str]) -> CallerIdentity {
        CallerIdentity::new("worker", namespaces.iter().map(|value| (*value).to_owned()))
    }

    /// `set_intervention_capabilities` replaces a live worker's advertised set —
    /// the announcement path a liminal agent worker takes after its in-band
    /// registration (the registration frame cannot carry capabilities) — and
    /// reports an unknown worker as `false` (an announcement racing a
    /// disconnect is benign, never an error).
    #[tokio::test]
    async fn set_intervention_capabilities_updates_live_worker() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(1);
        let types = ["scout".to_owned()];
        let guard = registry.register_delivery(
            ["default".to_owned()],
            "default",
            None,
            types.iter(),
            WorkerDelivery::Grpc(sender),
            RegistrationOptions::identified(
                "scout-1",
                crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
            )
            .with_intervention_capabilities(InterventionCapabilities::none()),
        )?;
        let Some(worker_id) = guard.worker_id() else {
            return Err(test_failure("registration carries an id"));
        };

        let announced = InterventionCapabilities {
            supported: vec![aion_core::InterventionPrimitive::InjectMessage],
        };
        assert!(
            registry.set_intervention_capabilities(worker_id, &announced)?,
            "a live worker's capabilities must be updatable"
        );
        let Some(handle) = registry.worker_by_id(worker_id)? else {
            return Err(test_failure("worker stays registered"));
        };
        assert_eq!(handle.intervention_capabilities(), &announced);

        assert!(
            !registry.set_intervention_capabilities(WorkerId(u64::MAX), &announced)?,
            "an unknown worker reports false, never an error"
        );
        Ok(())
    }

    /// WA-010 R3: the wire's `identity` reaches the handle. Before this pin it
    /// was dropped at the registry insert, so a lease could never name
    /// the process that took the attempt.
    #[tokio::test]
    async fn admission_carries_the_wire_identity_onto_the_handle() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = mpsc::channel(1);
        let admitted = registry
            .accept_registration(
                &guard(),
                &caller("tenant-a"),
                &registration("tenant-a", &["charge"]),
                sender,
            )
            .await?;
        let Some(worker_id) = admitted.worker_id() else {
            return Err(test_failure("registration carries an id"));
        };
        let Some(handle) = registry.worker_by_id(worker_id)? else {
            return Err(test_failure("the admitted worker is readable"));
        };
        assert_eq!(handle.identity(), WIRE_IDENTITY);

        let (plain_sender, _plain_receiver) = mpsc::channel(1);
        let types = ["charge".to_owned()];
        let plain = registry.register_namespaces(
            ["tenant-a".to_owned()],
            DEFAULT_TASK_QUEUE,
            None,
            types.iter(),
            plain_sender,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let Some(plain_id) = plain.worker_id() else {
            return Err(test_failure("registration carries an id"));
        };
        let Some(plain_handle) = registry.worker_by_id(plain_id)? else {
            return Err(test_failure("the plain worker is readable"));
        };
        assert_eq!(
            plain_handle.identity(),
            "",
            "a sender-only registration never sent an identity and must not invent one"
        );
        Ok(())
    }

    #[tokio::test]
    async fn register_and_deregister_are_namespace_isolated() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (tenant_a_tx, _tenant_a_rx) = mpsc::channel(1);
        let (tenant_b_tx, _tenant_b_rx) = mpsc::channel(1);

        let tenant_a = registry
            .accept_registration(
                &guard(),
                &caller("tenant-a"),
                &registration("tenant-a", &["charge", "charge"]),
                tenant_a_tx,
            )
            .await?;
        let tenant_b = registry
            .accept_registration(
                &guard(),
                &caller("tenant-b"),
                &registration("tenant-b", &["charge"]),
                tenant_b_tx,
            )
            .await?;

        let tq = DEFAULT_TASK_QUEUE;
        assert_eq!(
            registry.workers_for("tenant-a", tq, "charge", None)?.len(),
            1
        );
        assert_eq!(
            registry.workers_for("tenant-b", tq, "charge", None)?.len(),
            1
        );
        assert!(
            registry
                .workers_for("tenant-a", tq, "missing", None)?
                .is_empty()
        );

        let tenant_a_id = tenant_a.worker_id();
        tenant_a.deregister()?;

        assert!(
            registry
                .workers_for("tenant-a", tq, "charge", None)?
                .is_empty()
        );
        assert_eq!(
            registry.workers_for("tenant-b", tq, "charge", None)?.len(),
            1
        );
        assert_ne!(tenant_a_id, tenant_b.worker_id());

        tenant_b.deregister()?;
        assert!(
            registry
                .workers_for("tenant-b", tq, "charge", None)?
                .is_empty()
        );
        Ok(())
    }

    #[tokio::test]
    async fn denied_namespace_is_not_registered() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = mpsc::channel(1);
        let denied = registry
            .accept_registration(
                &guard(),
                &caller("tenant-a"),
                &registration("tenant-b", &["charge"]),
                tx,
            )
            .await;

        assert!(denied.is_err());
        assert!(
            registry
                .workers_for("tenant-b", DEFAULT_TASK_QUEUE, "charge", None)?
                .is_empty()
        );
        Ok(())
    }

    #[tokio::test]
    async fn task_queues_partition_disjoint_pools_within_one_namespace() -> Result<(), ServerError>
    {
        // Same namespace + same activity_type, two DIFFERENT task queues: the
        // pools are disjoint, a lookup for one queue never returns the other's
        // worker, and round-robin holds independently per (ns, tq, type) triple.
        let registry = ConnectedWorkerRegistry::default();
        let (norn_tx, _norn_rx) = mpsc::channel(1);
        let (claude_a_tx, _claude_a_rx) = mpsc::channel(1);
        let (claude_b_tx, _claude_b_rx) = mpsc::channel(1);

        let norn = registry
            .accept_registration(
                &guard(),
                &caller("local"),
                &registration_with_queue("local", "norn", &["dev"]),
                norn_tx,
            )
            .await?;
        // Two workers on the SAME (local, claude) pool to exercise round-robin.
        let claude_a = registry
            .accept_registration(
                &guard(),
                &caller("local"),
                &registration_with_queue("local", "claude", &["dev"]),
                claude_a_tx,
            )
            .await?;
        let claude_b = registry
            .accept_registration(
                &guard(),
                &caller("local"),
                &registration_with_queue("local", "claude", &["dev"]),
                claude_b_tx,
            )
            .await?;

        let norn_pool = registry.workers_for("local", "norn", "dev", None)?;
        assert_eq!(norn_pool.len(), 1, "norn pool has exactly its one worker");
        let norn_id = norn.worker_id().ok_or_else(missing_id)?;
        assert_eq!(norn_pool[0].id(), norn_id);

        let claude_pool = registry.workers_for("local", "claude", "dev", None)?;
        assert_eq!(
            claude_pool.len(),
            2,
            "claude pool sees only its two workers"
        );
        let claude_ids: BTreeSet<WorkerId> = claude_pool.iter().map(WorkerHandle::id).collect();
        assert!(
            !claude_ids.contains(&norn_id),
            "the norn worker must never appear in the claude pool"
        );

        // A dispatch targeting `norn` never reaches a `claude` worker, and vice
        // versa: the disjoint key is the boundary.
        assert!(
            !registry
                .workers_for("local", "norn", "dev", None)?
                .iter()
                .any(|worker| claude_ids.contains(&worker.id()))
        );

        // Round-robin per triple: the (local, claude, dev) cursor advances
        // independently and cycles through both claude workers, while the
        // (local, norn, dev) cursor keeps returning its single worker.
        let first = registry.workers_for("local", "claude", "dev", None)?[0].id();
        let second = registry.workers_for("local", "claude", "dev", None)?[0].id();
        assert_ne!(
            first, second,
            "claude pool round-robins across both workers"
        );
        assert_eq!(
            registry.workers_for("local", "norn", "dev", None)?[0].id(),
            norn_id,
            "the norn pool rotation is unaffected by claude traffic"
        );

        norn.deregister()?;
        claude_a.deregister()?;
        claude_b.deregister()?;
        Ok(())
    }

    #[tokio::test]
    async fn same_task_queue_in_different_namespaces_is_isolated() -> Result<(), ServerError> {
        // Same task_queue string, two DIFFERENT namespaces: namespace is the
        // correctness boundary, so the pools are isolated.
        let registry = ConnectedWorkerRegistry::default();
        let (local_tx, _local_rx) = mpsc::channel(1);
        let (remote_tx, _remote_rx) = mpsc::channel(1);

        let local = registry
            .accept_registration(
                &guard(),
                &caller("local"),
                &registration_with_queue("local", "gpu", &["render"]),
                local_tx,
            )
            .await?;
        let remote = registry
            .accept_registration(
                &guard(),
                &caller("remote"),
                &registration_with_queue("remote", "gpu", &["render"]),
                remote_tx,
            )
            .await?;

        let local_pool = registry.workers_for("local", "gpu", "render", None)?;
        let remote_pool = registry.workers_for("remote", "gpu", "render", None)?;
        assert_eq!(local_pool.len(), 1);
        assert_eq!(remote_pool.len(), 1);
        assert_ne!(
            local_pool[0].id(),
            remote_pool[0].id(),
            "a shared task_queue string does not merge two namespaces"
        );

        local.deregister()?;
        assert!(
            registry
                .workers_for("local", "gpu", "render", None)?
                .is_empty(),
            "deregistering the local worker leaves the remote namespace untouched"
        );
        assert_eq!(
            registry.workers_for("remote", "gpu", "render", None)?.len(),
            1
        );

        remote.deregister()?;
        Ok(())
    }

    #[tokio::test]
    async fn worker_serving_a_namespace_set_is_reachable_in_each() -> Result<(), ServerError> {
        // A worker advertising {a, b} is reachable for dispatch in BOTH a and b;
        // a worker in {a} is NOT reachable in b.
        let registry = ConnectedWorkerRegistry::default();
        let (ab_tx, _ab_rx) = mpsc::channel(1);
        let (a_tx, _a_rx) = mpsc::channel(1);

        let worker_ab = registry
            .accept_registration(
                &guard(),
                &multi_caller(&["a", "b"]),
                &registration_full(&["a", "b"], "default", "", &["dev"]),
                ab_tx,
            )
            .await?;
        let worker_a = registry
            .accept_registration(
                &guard(),
                &caller("a"),
                &registration_full(&["a"], "default", "", &["dev"]),
                a_tx,
            )
            .await?;

        let in_a = registry.workers_for("a", "default", "dev", None)?;
        let in_b = registry.workers_for("b", "default", "dev", None)?;
        let both_id = worker_ab.worker_id().ok_or_else(missing_id)?;
        let only_a_id = worker_a.worker_id().ok_or_else(missing_id)?;

        // Namespace a sees BOTH workers; namespace b sees ONLY the {a, b} worker.
        let a_ids: BTreeSet<WorkerId> = in_a.iter().map(WorkerHandle::id).collect();
        assert_eq!(a_ids, BTreeSet::from([both_id, only_a_id]));
        assert_eq!(in_b.len(), 1, "only the {{a, b}} worker is reachable in b");
        assert_eq!(in_b[0].id(), both_id);
        assert!(
            !in_b.iter().any(|worker| worker.id() == only_a_id),
            "the {{a}}-only worker must not be reachable in b"
        );

        // Deregistering the {a, b} worker removes it from BOTH buckets.
        worker_ab.deregister()?;
        assert!(
            registry
                .workers_for("b", "default", "dev", None)?
                .is_empty()
        );
        assert_eq!(registry.workers_for("a", "default", "dev", None)?.len(), 1);

        worker_a.deregister()?;
        Ok(())
    }

    #[tokio::test]
    async fn node_pin_filters_within_pool() -> Result<(), ServerError> {
        // Two workers in the same (namespace, task_queue) pool on different
        // nodes: unpinned round-robins across both; pinned to node N reaches
        // ONLY the worker(s) on N; pinned to a node with no worker finds none.
        let registry = ConnectedWorkerRegistry::default();
        let (n1_tx, _n1_rx) = mpsc::channel(1);
        let (n2_tx, _n2_rx) = mpsc::channel(1);

        let on_n1 = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "n1", &["dev"]),
                n1_tx,
            )
            .await?;
        let on_n2 = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "n2", &["dev"]),
                n2_tx,
            )
            .await?;
        let n1_id = on_n1.worker_id().ok_or_else(missing_id)?;
        let n2_id = on_n2.worker_id().ok_or_else(missing_id)?;

        // Unpinned: both workers are candidates and round-robin advances.
        let unpinned = registry.workers_for("ns", "tq", "dev", None)?;
        assert_eq!(unpinned.len(), 2, "unpinned reaches the whole pool");
        let first = registry.workers_for("ns", "tq", "dev", None)?[0].id();
        let second = registry.workers_for("ns", "tq", "dev", None)?[0].id();
        assert_ne!(first, second, "unpinned round-robins across both nodes");

        // Pinned to n1: only the n1 worker; pinned to n2: only the n2 worker.
        let pinned_n1 = registry.workers_for("ns", "tq", "dev", Some("n1"))?;
        assert_eq!(pinned_n1.len(), 1);
        assert_eq!(pinned_n1[0].id(), n1_id);
        let pinned_n2 = registry.workers_for("ns", "tq", "dev", Some("n2"))?;
        assert_eq!(pinned_n2.len(), 1);
        assert_eq!(pinned_n2[0].id(), n2_id);

        // select_worker honours the same filter.
        assert_eq!(
            registry
                .select_and_reserve("ns", "tq", "dev", Some("n1"))?
                .map(|(worker, _reservation)| worker)
                .map(|worker| worker.id()),
            Some(n1_id)
        );

        // Pinned to a node with no worker finds no candidate (the dispatcher
        // then waits via the same no-worker path the existing test exercises).
        assert!(
            registry
                .workers_for("ns", "tq", "dev", Some("absent"))?
                .is_empty(),
            "a pin to a node with no worker yields no candidate"
        );
        assert!(
            registry
                .select_and_reserve("ns", "tq", "dev", Some("absent"))?
                .map(|(worker, _reservation)| worker)
                .is_none()
        );

        on_n1.deregister()?;
        on_n2.deregister()?;
        Ok(())
    }

    #[tokio::test]
    async fn shared_node_id_round_robins_across_workers() -> Result<(), ServerError> {
        // Two workers SHARING a node id in the same pool: a dispatch pinned to
        // that node round-robins across BOTH (node is locality, not process).
        let registry = ConnectedWorkerRegistry::default();
        let (a_tx, _a_rx) = mpsc::channel(1);
        let (b_tx, _b_rx) = mpsc::channel(1);

        let worker_a = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "shared", &["dev"]),
                a_tx,
            )
            .await?;
        let worker_b = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "shared", &["dev"]),
                b_tx,
            )
            .await?;
        let a_id = worker_a.worker_id().ok_or_else(missing_id)?;
        let b_id = worker_b.worker_id().ok_or_else(missing_id)?;

        let pinned = registry.workers_for("ns", "tq", "dev", Some("shared"))?;
        assert_eq!(
            pinned.len(),
            2,
            "both workers on the shared node are candidates"
        );
        let pinned_ids: BTreeSet<WorkerId> = pinned.iter().map(WorkerHandle::id).collect();
        assert_eq!(pinned_ids, BTreeSet::from([a_id, b_id]));

        let first = registry.workers_for("ns", "tq", "dev", Some("shared"))?[0].id();
        let second = registry.workers_for("ns", "tq", "dev", Some("shared"))?[0].id();
        assert_ne!(
            first, second,
            "a pin to a shared node round-robins across both workers on it"
        );

        worker_a.deregister()?;
        worker_b.deregister()?;
        Ok(())
    }

    /// T1 — the finding, as a test. `select_worker` took the LOWEST matching
    /// worker id, so a workflow whose activities are dispatched over anything
    /// but the gRPC push leg sent every one of them to the first-registered
    /// worker and left every other node idle. Selection is the candidate at the
    /// pool's rotation cursor, so four consecutive unpinned selections walk the
    /// two nodes twice.
    ///
    /// The sequence is asserted, not merely "they differ": a selector that
    /// alternated by luck of hash order would satisfy a difference assertion and
    /// still not be a round-robin.
    #[tokio::test]
    async fn unpinned_selection_rotates_across_nodes() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (n1_tx, _n1_rx) = mpsc::channel(1);
        let (n2_tx, _n2_rx) = mpsc::channel(1);

        let on_n1 = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "n1", &["dev"]),
                n1_tx,
            )
            .await?;
        let on_n2 = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "n2", &["dev"]),
                n2_tx,
            )
            .await?;

        let mut visited = Vec::new();
        for _ in 0..4 {
            let selected = registry
                .select_and_reserve("ns", "tq", "dev", None)?
                .map(|(worker, _reservation)| worker)
                .ok_or_else(|| test_failure("an unpinned selection must find a worker"))?;
            visited.push(selected.node().map(str::to_owned));
        }
        assert_eq!(
            visited,
            vec![
                Some(String::from("n1")),
                Some(String::from("n2")),
                Some(String::from("n1")),
                Some(String::from("n2")),
            ],
            "unpinned selection must rotate across the nodes rather than pin itself to the \
             lowest worker id"
        );

        on_n1.deregister()?;
        on_n2.deregister()?;
        Ok(())
    }

    /// T2 — one eligible-candidate derivation means the exclusion a liveness
    /// verdict publishes binds BOTH selectors.
    ///
    /// `workers_for` ignored it entirely and handed the push dispatcher a worker
    /// the server had already found unreachable; `select_worker` honoured it. The
    /// excluded worker here is the LOWEST id on purpose — the one the old
    /// `min_by_key` selector would have taken every time — so a rotation that
    /// merely skipped position zero could not pass this.
    #[test]
    fn an_ineligible_worker_is_never_selected_by_either_selector() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let types = [String::from("dev")];
        let mut receivers = Vec::new();
        let mut registrations = Vec::new();
        for _ in 0..3 {
            let (tx, rx) = mpsc::channel(1);
            receivers.push(rx);
            registrations.push(registry.register_namespaces(
                [String::from("ns")],
                "tq",
                None,
                types.iter(),
                tx,
                crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
            )?);
        }
        let mut ids = Vec::new();
        for registration in &registrations {
            ids.push(registration.worker_id().ok_or_else(missing_id)?);
        }
        ids.sort_unstable();
        let excluded = *ids.first().ok_or_else(missing_id)?;
        registry.set_dispatch_ineligible(
            [(excluded, DispatchExclusion::ReachabilityLost)]
                .into_iter()
                .collect(),
        )?;

        // Four calls: two full rotations of the two-worker eligible set, through
        // each selector in turn, so an exclusion honoured only at one cursor
        // position could not hide.
        for _ in 0..4 {
            let selected = registry
                .select_and_reserve("ns", "tq", "dev", None)?
                .map(|(worker, _reservation)| worker)
                .ok_or_else(|| test_failure("two eligible workers remain in the pool"))?;
            assert_ne!(
                selected.id(),
                excluded,
                "select_worker must never return the worker the liveness verdict excluded"
            );
            let candidates = registry.workers_for("ns", "tq", "dev", None)?;
            assert_eq!(
                candidates.len(),
                2,
                "workers_for must offer the two ELIGIBLE workers, not all three registered ones"
            );
            assert!(
                candidates.iter().all(|worker| worker.id() != excluded),
                "workers_for must not list the excluded worker in ANY position: the push \
                 dispatcher walks the whole list"
            );
        }

        for registration in registrations {
            registration.deregister()?;
        }
        Ok(())
    }

    /// The hazard the eligible-set derivation introduces on the push dispatcher,
    /// demonstrated at the bytes rather than argued.
    ///
    /// `pool_census` deliberately counts REGISTERED node-matched workers with no
    /// eligibility filter (#197 R3: `classify` must be able to tell an empty pool
    /// from an excluded one), so an all-ineligible pool reports itself SERVED
    /// while selection finds nobody. `dispatch_to_node` reads exactly this pair,
    /// and treating the disagreement as the registration race it used to be —
    /// re-selecting at once — would spin that loop hot with no park and no log
    /// for as long as the exclusion lasts. It parks instead; the proof is
    /// `a_dispatch_to_an_all_ineligible_pool_parks_until_eligibility_returns` in
    /// `dispatch.rs`, and the park's wake-up is proven directly below.
    #[test]
    fn an_all_ineligible_pool_selects_nobody_while_the_census_still_reads_served()
    -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let types = [String::from("dev")];
        let (tx, _rx) = mpsc::channel(1);
        let worker = registry.register_namespaces(
            [String::from("ns")],
            "tq",
            None,
            types.iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let worker_id = worker.worker_id().ok_or_else(missing_id)?;
        // 🔴 An OPENING PROBATION specifically, and the choice is load-bearing.
        // This test's subject is the census/selection disagreement that
        // `dispatch_to_node` must park on rather than spin through, and that
        // disagreement survives exactly for the exclusion that clears itself.
        // A `ReachabilityLost` pool no longer reaches this state: it now
        // classifies as `POLLERS_UNREACHABLE` and publishes a reason instead of
        // parking silently, which is the whole of the withdrawn-pool fix. Using
        // it here would make this test assert the behaviour that was replaced.
        registry.set_dispatch_ineligible(
            [(
                worker_id,
                DispatchExclusion::OpeningProbation { answers: 1 },
            )]
            .into_iter()
            .collect(),
        )?;

        assert!(
            registry.workers_for("ns", "tq", "dev", None)?.is_empty(),
            "the push dispatcher's candidate list must exclude an unreachable worker"
        );
        assert!(
            registry
                .select_and_reserve("ns", "tq", "dev", None)?
                .map(|(worker, _reservation)| worker)
                .is_none(),
            "and so must the single selector"
        );
        let census = registry.pool_census("ns", "tq", "dev", None)?;
        assert_eq!(
            census.compatible_workers, 1,
            "the census counts the REGISTERED worker: #197 R3 needs the count to separate an \
             empty pool from an excluded one"
        );
        assert!(
            census.is_served(),
            "so `classify` reads this address as served and returns None while selection has \
             nobody — the disagreement dispatch_to_node must park on rather than spin through"
        );
        assert!(
            census.will_be_served(),
            "and the park is the RIGHT outcome here: a probation clears itself within seconds, \
             so nobody should be warned and nothing should be published. The pool that has LOST \
             reachability is the one that must not reach this state — see \
             `the_exclusion_cause_is_what_decides`"
        );
        assert_eq!(
            census.compatible_workers_reachability_lost, 0,
            "precondition for the assertion above: this fixture's exclusion is a probation"
        );

        worker.deregister()?;
        Ok(())
    }

    /// T8 — the arrival subscription RETAINS a wake that lands before it is
    /// awaited, and a subscription taken after that wake does not.
    ///
    /// This is the flight-1 judge's finding reproduced at the bytes and then
    /// closed, both arms against the same registry so the pair is read together
    /// and neither can rot alone.
    ///
    /// Both registry wake sources are `Notify::notify_waiters`, which stores NO
    /// permit. A dispatcher that read the registry, missed, and only then
    /// constructed its wait therefore parked past a registration that had
    /// already happened — holding positive census evidence of a live worker,
    /// with no second event on the way. On `OutboxTransport::Grpc` no liveness
    /// probe runs and no reachability verdict is ever published, so "the next
    /// unrelated registration anywhere in the registry" was the only wake left.
    ///
    /// Polled by hand with a no-op waker rather than raced against a runtime, so
    /// retention is PROVEN rather than timed: no clock, no timeout, no runtime.
    #[test]
    fn an_arrival_subscription_retains_a_wake_taken_before_it() -> Result<(), ServerError> {
        use std::future::Future;
        use std::task::Waker;

        let registry = ConnectedWorkerRegistry::default();
        let mut context = Context::from_waker(Waker::noop());

        // CONTROL ARM — the base's sequence, subscribe AFTER the look.
        // The registration fires `notify_waiters` into an empty waiter list.
        let (first_tx, _first_rx) = mpsc::channel(1);
        let first = registry.register(
            "ns",
            [String::from("dev")].iter(),
            first_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let mut too_late = std::pin::pin!(registry.worker_arrived.notified());
        assert!(
            matches!(too_late.as_mut().poll(&mut context), Poll::Pending),
            "a wait constructed AFTER the registration cannot have retained it: \
             `notify_waiters` stored no permit and there was no waiter in the list to \
             broadcast to. This is the finding, reproduced."
        );
        assert!(
            matches!(too_late.as_mut().poll(&mut context), Poll::Pending),
            "and it stays parked — without this control the arm above would pass on a wait \
             that merely reports Pending on its first poll for registration reasons"
        );

        // SUBJECT ARM — the fix's sequence, subscribe BEFORE the look.
        let arrival = registry.worker_arrival();
        let (second_tx, _second_rx) = mpsc::channel(1);
        let second = registry.register(
            "ns",
            [String::from("dev")].iter(),
            second_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let mut arrival = std::pin::pin!(arrival);
        assert!(
            matches!(arrival.as_mut().poll(&mut context), Poll::Ready(())),
            "a subscription taken BEFORE the registry read must retain the registration that \
             landed during it: the dispatch that missed by a microsecond must not park"
        );
        assert!(
            matches!(too_late.as_mut().poll(&mut context), Poll::Ready(())),
            "the control's own wait was in the list by now, so the SECOND registration wakes \
             it — proving the control arm above was parked on the lost wake and not on a \
             registry that never notified at all"
        );

        first.deregister()?;
        second.deregister()?;
        Ok(())
    }

    /// The other half of that resolution: publishing a reachability verdict WAKES
    /// the selection wait.
    ///
    /// A dispatch parked because every compatible worker is ineligible is waiting
    /// on the verdict, not on a registration — those workers are already
    /// registered, so waiting for a registration sleeps through their recovery.
    ///
    /// Polled by hand with a no-op waker rather than raced against a runtime, so
    /// the wake is proven rather than timed. The middle poll is THE CONTROL: a
    /// wait that completed on its second poll for any reason at all would satisfy
    /// the final assertion and prove nothing about the publication.
    #[test]
    fn publishing_a_reachability_verdict_wakes_the_selection_wait() -> Result<(), ServerError> {
        use std::future::Future;
        use std::task::Waker;

        let registry = ConnectedWorkerRegistry::default();
        let mut waiting = std::pin::pin!(registry.worker_arrival());
        let mut context = Context::from_waker(Waker::noop());

        assert!(
            matches!(waiting.as_mut().poll(&mut context), Poll::Pending),
            "the wait parks until something changes what selection can see"
        );
        assert!(
            matches!(waiting.as_mut().poll(&mut context), Poll::Pending),
            "and stays parked while nothing has been published: without this control the \
             assertion below would pass on a wait that simply completes on a second poll"
        );

        // Exactly what a probe round does once a worker has served its probation.
        registry.set_dispatch_ineligible(BTreeMap::new())?;
        assert!(
            matches!(waiting.as_mut().poll(&mut context), Poll::Ready(())),
            "a published verdict must wake a dispatch parked on eligibility; no registration is \
             coming for a worker that never left the registry"
        );
        Ok(())
    }

    #[tokio::test]
    async fn rotation_cursor_is_pruned_when_last_worker_leaves() -> Result<(), ServerError> {
        // The round-robin cursor is keyed on arbitrary caller-supplied strings;
        // it must not outlive the pool it rotates, or a never-dying server leaks
        // memory. After the last worker for a triple deregisters, no cursor for
        // that triple may remain.
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = mpsc::channel(1);
        let worker = registry
            .accept_registration(
                &guard(),
                &caller("ns"),
                &registration_full(&["ns"], "tq", "", &["dev"]),
                tx,
            )
            .await?;

        // Drive the lazy cursor insert.
        let _ = registry.workers_for("ns", "tq", "dev", None)?;
        let key = ActivityKey::new(PoolAddress::new("ns", "tq"), "dev");
        assert!(
            registry.state()?.rotation.contains_key(&key),
            "a lookup must have created the rotation cursor"
        );

        worker.deregister()?;
        let state = registry.state()?;
        assert!(
            !state.rotation.contains_key(&key),
            "the rotation cursor must be pruned once the last worker leaves"
        );
        assert!(
            !state.by_activity.contains_key(&key),
            "the activity bucket must also be gone"
        );
        Ok(())
    }

    fn missing_id() -> ServerError {
        ServerError::lock_poisoned("registration unexpectedly missing a worker id")
    }

    // ---- Minted-on-use (Control-Plane Phase 1) -----------------------------

    fn namespace_store() -> Arc<dyn NamespaceStore> {
        Arc::new(aion_store::InMemoryStore::default())
    }

    fn minting_registry(
        store: &Arc<dyn NamespaceStore>,
        policy: AutoCreate,
    ) -> ConnectedWorkerRegistry {
        ConnectedWorkerRegistry::default().with_namespace_minting(Arc::clone(store), policy)
    }

    #[tokio::test]
    async fn open_register_mints_durable_record_and_is_idempotent() -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = minting_registry(&store, AutoCreate::Open);

        // First registration mints the namespace.
        let (tx_one, _rx_one) = mpsc::channel(1);
        let first = registry
            .accept_registration(
                &guard(),
                &caller("orders"),
                &registration("orders", &["charge"]),
                tx_one,
            )
            .await?;
        let record = store
            .get_namespace("orders")
            .await?
            .ok_or_else(|| ServerError::namespace_denied("expected a minted record"))?;
        assert_eq!(record.name, "orders");
        assert_eq!(record.origin, NamespaceOrigin::WorkerMint);

        // Re-registering the same namespace is idempotent: no duplicate row,
        // and the prior worker is unaffected.
        let (tx_two, _rx_two) = mpsc::channel(1);
        registry
            .accept_registration(
                &guard(),
                &caller("orders"),
                &registration("orders", &["refund"]),
                tx_two,
            )
            .await?;
        let all = store.list_namespaces().await?;
        assert_eq!(
            all.iter().filter(|r| r.name == "orders").count(),
            1,
            "re-register must not create a duplicate namespace row"
        );
        drop(first);
        Ok(())
    }

    #[tokio::test]
    async fn open_register_mints_each_namespace_in_a_multi_namespace_worker()
    -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = minting_registry(&store, AutoCreate::Open);
        let (tx, _rx) = mpsc::channel(1);

        registry
            .accept_registration(
                &guard(),
                &multi_caller(&["alpha", "beta"]),
                &registration_full(&["alpha", "beta"], "", "", &["charge"]),
                tx,
            )
            .await?;

        assert!(store.get_namespace("alpha").await?.is_some());
        assert!(store.get_namespace("beta").await?.is_some());
        Ok(())
    }

    // ---- Placement admission (Control-Plane Phase 2, P2-I1) -----------------

    /// Pre-mint `namespace` and set its placement to `Pinned{nodes}`, returning a
    /// minting registry over the same store so `accept_registration` reads the
    /// placement from the SAME durable record.
    async fn pinned_registry(
        store: &Arc<dyn NamespaceStore>,
        namespace: &str,
        nodes: &[&str],
    ) -> Result<ConnectedWorkerRegistry, ServerError> {
        store
            .register_namespace(namespace, NamespaceOrigin::Explicit)
            .await?;
        store
            .set_namespace_placement(
                namespace,
                NamespacePlacement::Pinned {
                    nodes: nodes.iter().map(|n| (*n).to_owned()).collect(),
                },
            )
            .await?;
        Ok(minting_registry(store, AutoCreate::Open))
    }

    /// A worker on a node IN the required set registers successfully into a
    /// `Pinned{n1}` namespace, and is reachable in the pool.
    #[tokio::test]
    async fn pinned_admits_a_worker_on_a_required_node() -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = pinned_registry(&store, "iso", &["n1"]).await?;
        let (tx, _rx) = mpsc::channel(1);

        let _registration = registry
            .accept_registration(
                &guard(),
                &caller("iso"),
                &registration_full(&["iso"], "", "n1", &["charge"]),
                tx,
            )
            .await?;

        assert_eq!(
            registry
                .workers_for("iso", DEFAULT_TASK_QUEUE, "charge", Some("n1"))?
                .len(),
            1,
            "an n1 worker must be admitted into the Pinned{{n1}} namespace's pool"
        );
        Ok(())
    }

    /// A worker on a node NOT in the required set is rejected — the WHOLE
    /// registration fails (loud) and no worker is inserted. This would FAIL under
    /// no admission gate (the worker would join and steal Pinned dispatches).
    #[tokio::test]
    async fn pinned_rejects_a_wrong_node_worker() -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = pinned_registry(&store, "iso", &["n1"]).await?;
        let (tx, _rx) = mpsc::channel(1);

        let denied = registry
            .accept_registration(
                &guard(),
                &caller("iso"),
                &registration_full(&["iso"], "", "n2", &["charge"]),
                tx,
            )
            .await;
        assert!(
            matches!(denied, Err(ServerError::Namespace { .. })),
            "a wrong-node (n2) worker must be rejected from a Pinned{{n1}} namespace"
        );
        assert!(
            registry
                .workers_for("iso", DEFAULT_TASK_QUEUE, "charge", None)?
                .is_empty(),
            "a rejected registration must not insert a worker on any node"
        );
        Ok(())
    }

    /// A worker advertising NO node (`node == ""` → `None`) is rejected from a
    /// `Pinned{n1}` namespace: an unlabelled worker can never satisfy a hard pin.
    #[tokio::test]
    async fn pinned_rejects_a_node_less_worker() -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = pinned_registry(&store, "iso", &["n1"]).await?;
        let (tx, _rx) = mpsc::channel(1);

        let denied = registry
            .accept_registration(
                &guard(),
                &caller("iso"),
                &registration_full(&["iso"], "", "", &["charge"]),
                tx,
            )
            .await;
        assert!(
            matches!(denied, Err(ServerError::Namespace { .. })),
            "a node-less worker must be rejected from a Pinned{{n1}} namespace"
        );
        assert!(
            registry
                .workers_for("iso", DEFAULT_TASK_QUEUE, "charge", None)?
                .is_empty(),
            "a rejected node-less registration must not insert a worker"
        );
        Ok(())
    }

    /// Reject-WHOLE-registration (Open Decision 6): a worker serving BOTH a
    /// non-isolated namespace and a `Pinned{n1}` namespace on a wrong node is
    /// rejected entirely — the compliant namespace does NOT get a partial admit.
    #[tokio::test]
    async fn pinned_violation_rejects_the_whole_multi_namespace_registration()
    -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = pinned_registry(&store, "iso", &["n1"]).await?;
        let (tx, _rx) = mpsc::channel(1);

        let denied = registry
            .accept_registration(
                &guard(),
                &multi_caller(&["free", "iso"]),
                &registration_full(&["free", "iso"], "", "n2", &["charge"]),
                tx,
            )
            .await;
        assert!(
            matches!(denied, Err(ServerError::Namespace { .. })),
            "a wrong-node worker serving a Pinned namespace fails the WHOLE registration"
        );
        assert!(
            registry
                .workers_for("free", DEFAULT_TASK_QUEUE, "charge", None)?
                .is_empty(),
            "the compliant namespace must NOT be partially admitted"
        );
        Ok(())
    }

    /// Unplaced and Prefer namespaces are UNAFFECTED: a node-less worker registers
    /// normally (byte-identical to the pre-P2-I1 behaviour). Only Pinned gates.
    #[tokio::test]
    async fn unplaced_and_prefer_admission_is_unaffected_by_the_pinned_gate()
    -> Result<(), ServerError> {
        let store = namespace_store();
        // `unpl` is left Unplaced (default); `pref` is Prefer{n1}. A node-less
        // worker must be admitted into BOTH.
        store
            .register_namespace("pref", NamespaceOrigin::Explicit)
            .await?;
        store
            .set_namespace_placement(
                "pref",
                NamespacePlacement::Prefer {
                    nodes: ["n1".to_owned()].into_iter().collect(),
                },
            )
            .await?;
        let registry = minting_registry(&store, AutoCreate::Open);

        let (tx_a, _rx_a) = mpsc::channel(1);
        let _reg_a = registry
            .accept_registration(
                &guard(),
                &caller("unpl"),
                &registration_full(&["unpl"], "", "", &["charge"]),
                tx_a,
            )
            .await?;
        let (tx_b, _rx_b) = mpsc::channel(1);
        let _reg_b = registry
            .accept_registration(
                &guard(),
                &caller("pref"),
                &registration_full(&["pref"], "", "", &["charge"]),
                tx_b,
            )
            .await?;

        assert_eq!(
            registry
                .workers_for("unpl", DEFAULT_TASK_QUEUE, "charge", None)?
                .len(),
            1,
            "an Unplaced namespace admits a node-less worker unchanged"
        );
        assert_eq!(
            registry
                .workers_for("pref", DEFAULT_TASK_QUEUE, "charge", None)?
                .len(),
            1,
            "a Prefer namespace admits a node-less worker unchanged (only Pinned gates)"
        );
        Ok(())
    }

    /// A default (no-minter) registry is byte-identical: the placement gate is a
    /// no-op with no minter installed, so a node-less worker registers freely even
    /// though there is no way to have set a placement in the first place.
    #[tokio::test]
    async fn no_minter_registry_skips_the_placement_gate() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = mpsc::channel(1);
        let _registration = registry
            .accept_registration(
                &guard(),
                &caller("plain"),
                &registration_full(&["plain"], "", "", &["charge"]),
                tx,
            )
            .await?;
        assert_eq!(
            registry
                .workers_for("plain", DEFAULT_TASK_QUEUE, "charge", None)?
                .len(),
            1,
            "with no minter the placement gate is a no-op — registration is unchanged"
        );
        Ok(())
    }

    #[tokio::test]
    async fn concurrent_registrations_for_a_new_namespace_create_exactly_one_record()
    -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = minting_registry(&store, AutoCreate::Open);

        let mut handles = Vec::new();
        for _ in 0..8 {
            let registry = registry.clone();
            handles.push(tokio::spawn(async move {
                let (tx, rx) = mpsc::channel(1);
                let outcome = registry
                    .accept_registration(
                        &guard(),
                        &caller("rush"),
                        &registration("rush", &["charge"]),
                        tx,
                    )
                    .await;
                // Keep the receiver alive for the duration of the registration.
                drop(rx);
                outcome.map(|registration| registration.worker_id())
            }));
        }
        for handle in handles {
            handle
                .await
                .map_err(|_| ServerError::lock_poisoned("registration task panicked"))??;
        }

        let all = store.list_namespaces().await?;
        assert_eq!(
            all.iter().filter(|r| r.name == "rush").count(),
            1,
            "racing registrations must converge on exactly one durable record"
        );
        Ok(())
    }

    #[tokio::test]
    async fn closed_rejects_unknown_namespace_and_does_not_create_it() -> Result<(), ServerError> {
        let store = namespace_store();
        let registry = minting_registry(&store, AutoCreate::Closed);
        let (tx, _rx) = mpsc::channel(1);

        let denied = registry
            .accept_registration(
                &guard(),
                &caller("ghost"),
                &registration("ghost", &["charge"]),
                tx,
            )
            .await;
        assert!(
            matches!(denied, Err(ServerError::Namespace { .. })),
            "closed policy must reject an unknown namespace"
        );
        assert!(
            store.get_namespace("ghost").await?.is_none(),
            "closed policy must NOT create the namespace it rejected"
        );
        let tq = DEFAULT_TASK_QUEUE;
        assert!(
            registry
                .workers_for("ghost", tq, "charge", None)?
                .is_empty(),
            "a rejected registration must not insert a worker"
        );
        Ok(())
    }

    #[tokio::test]
    async fn closed_admits_a_known_namespace() -> Result<(), ServerError> {
        let store = namespace_store();
        // Pre-mint the namespace (the POST /namespaces escape hatch's effect).
        store
            .register_namespace("known", NamespaceOrigin::Explicit)
            .await?;
        let registry = minting_registry(&store, AutoCreate::Closed);
        let (tx, _rx) = mpsc::channel(1);

        // Bind the registration token: dropping it deregisters the worker.
        let _registration = registry
            .accept_registration(
                &guard(),
                &caller("known"),
                &registration("known", &["charge"]),
                tx,
            )
            .await?;
        let tq = DEFAULT_TASK_QUEUE;
        assert_eq!(
            registry.workers_for("known", tq, "charge", None)?.len(),
            1,
            "a known namespace must register under closed policy"
        );
        Ok(())
    }

    #[tokio::test]
    async fn no_minter_leaves_registration_untouched() -> Result<(), ServerError> {
        // The default registry installs no minter: registration succeeds and
        // never touches any namespace registry (byte-identical legacy path).
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = mpsc::channel(1);
        let _registration = registry
            .accept_registration(
                &guard(),
                &caller("orders"),
                &registration("orders", &["charge"]),
                tx,
            )
            .await?;
        let tq = DEFAULT_TASK_QUEUE;
        assert_eq!(registry.workers_for("orders", tq, "charge", None)?.len(), 1);
        Ok(())
    }

    /// R1 census: an address no worker has ever served reports an empty fleet
    /// and no poller age at all — "never seen" is not "seen long ago".
    #[test]
    fn a_never_served_address_censuses_empty_with_no_poller_age() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let census = registry.pool_census("default", "general", "greet", None)?;
        assert_eq!(census.workers_in_pool, 0);
        assert_eq!(census.workers_serving_activity, 0);
        assert_eq!(census.compatible_workers, 0);
        assert_eq!(census.last_compatible_poller_age, None);
        assert!(!census.is_served());
        Ok(())
    }

    /// R1 census: pool membership, activity coverage, and node coverage are
    /// three separate counts — that separation is what tells `NO_LIVE_POLLERS`
    /// apart from `POLLERS_INCOMPATIBLE`.
    #[test]
    fn the_census_separates_pool_activity_and_node_coverage() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = mpsc::channel(1);
        let _worker = registry.register_namespaces(
            [String::from("default")],
            "general",
            Some(String::from("n1")),
            [String::from("greet")].iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;

        let unpinned = registry.pool_census("default", "general", "greet", None)?;
        assert_eq!(unpinned.workers_in_pool, 1);
        assert_eq!(unpinned.workers_serving_activity, 1);
        assert_eq!(unpinned.compatible_workers, 1);
        assert_eq!(unpinned.last_compatible_poller_age, Some(Duration::ZERO));

        // Same pool, activity nobody advertises: pollers are live but
        // incompatible.
        let other_activity = registry.pool_census("default", "general", "settle", None)?;
        assert_eq!(other_activity.workers_in_pool, 1);
        assert_eq!(other_activity.workers_serving_activity, 0);
        assert_eq!(other_activity.compatible_workers, 0);

        // Same activity, wrong node: coverage exists in the pool but not for
        // this dispatch.
        let wrong_node = registry.pool_census("default", "general", "greet", Some("n2"))?;
        assert_eq!(wrong_node.workers_serving_activity, 1);
        assert_eq!(wrong_node.compatible_workers, 0);
        assert_eq!(wrong_node.last_compatible_poller_age, None);
        Ok(())
    }

    /// ACCEPTANCE (b) FOR #58, wiring half: a worker whose PROCESS looks alive
    /// but which the server cannot REACH on the push leg must lose dispatch
    /// eligibility — and losing it must actually remove it from selection.
    ///
    /// This is the composition the lease split exists for. The tracker half is
    /// pinned in `heartbeat::reachability_tests`; this pins the half that turns
    /// that verdict into a dispatch decision, using the SAME two types the
    /// liveness probe wires together — a real [`HeartbeatTracker`] census
    /// feeding a real [`ConnectedWorkerRegistry`].
    ///
    /// The staging is Finding B exactly: the worker's own liveness pump keeps
    /// beating (`record_connection_activity`) well past the window, so every
    /// signal that says "this process is alive" says so — and NO ping is ever
    /// answered, so the one signal that says "the server can reach it" is
    /// absent. Before the split those were one lease and the pump's weaker
    /// positive evidence buried the ping's negative evidence.
    ///
    /// The control is the second half of the test: the same worker, reachable,
    /// must still be selected. Without it a registry that selected NOBODY would
    /// satisfy the first assertion and prove nothing.
    #[test]
    fn a_pump_alive_worker_the_server_cannot_reach_is_not_selected() -> Result<(), ServerError> {
        const WINDOW: Duration = Duration::from_secs(30);

        let registry = ConnectedWorkerRegistry::default();
        let tracker = HeartbeatTracker::new(WINDOW);
        let (tx, _rx) = mpsc::channel(1);
        let worker = registry.register_namespaces(
            [String::from("default")],
            "general",
            None,
            [String::from("greet")].iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let Some(worker_id) = worker.worker_id() else {
            return Err(test_failure("registration carries an id"));
        };
        let start = Instant::now();
        tracker.register_connection(worker_id, start)?;

        // Publishing the tracker's verdict is one motion the liveness probe
        // performs at the end of every round, and this test performs it four
        // times. Written once so the four publications cannot drift apart and
        // quietly stop testing the same thing.
        let publish = |now: Instant| -> Result<(), ServerError> {
            registry.set_dispatch_ineligible(
                tracker
                    .unreachable_workers(now)?
                    .into_iter()
                    .map(|excluded| (excluded.worker_id, excluded.exclusion))
                    .collect(),
            )
        };

        // Baseline: the worker SERVES ITS PROBATION and is therefore selectable.
        // Registration alone grants nothing — the handshake ack is sent, never
        // confirmed received — so eligibility here is earned by answered pings.
        // If this did not hold, no worker could ever be dispatched to.
        for _ in 0..DISPATCH_PROBATION_PINGS {
            assert!(
                tracker.record_dispatch_reachability(worker_id, start)?,
                "the worker is tracked while it serves its probation"
            );
        }
        publish(start)?;
        assert!(
            registry
                .select_and_reserve("default", "general", "greet", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "a worker that has answered a full run of pings must be selectable: this is the \
             control, and without it a registry selecting NOBODY would satisfy every assertion \
             below"
        );

        // Now the poisoned-connection shape: the pump beats past the window and
        // not one ping is answered. Each unanswered probe is recorded exactly as
        // the liveness probe records it, because that is the trajectory a real
        // poisoned connection takes — staleness alone would be a shape only a
        // stopped probe could produce.
        let much_later = start + WINDOW * 4;
        assert!(
            tracker.record_connection_activity(worker_id, much_later)?,
            "the worker is still tracked; its process is plainly alive"
        );
        assert!(
            tracker.record_dispatch_unreachable(worker_id)?,
            "the probe fired and went unanswered"
        );
        publish(much_later)?;

        assert!(
            registry.is_dispatch_ineligible(worker_id)?,
            "a worker the server cannot reach must be marked ineligible however alive its \
             process looks"
        );
        assert!(
            registry
                .select_and_reserve("default", "general", "greet", None)?
                .map(|(worker, _reservation)| worker)
                .is_none(),
            "an ineligible worker must not be SELECTED: selecting one produces a dispatch that \
             can only fail, and on the liminal transport it fails by consuming connection \
             capacity — making the unreachability worse"
        );

        // The recovery path: exclusion must be WITHDRAWABLE, or a single bad
        // round would strand a healthy worker forever. It is withdrawn by a
        // served probation, not by one answer — and the intermediate assertion
        // below pins that distinction rather than assuming it.
        assert!(
            tracker.record_dispatch_reachability(worker_id, much_later)?,
            "the worker is still tracked"
        );
        publish(much_later)?;
        assert!(
            registry.is_dispatch_ineligible(worker_id)?,
            "one answer part-way through a fresh probation must NOT restore eligibility: a link \
             answering one probe in three would otherwise flap in and out of selection"
        );

        for _ in 1..DISPATCH_PROBATION_PINGS {
            assert!(
                tracker.record_dispatch_reachability(worker_id, much_later)?,
                "the worker is still tracked"
            );
        }
        publish(much_later)?;
        assert!(
            !registry.is_dispatch_ineligible(worker_id)?,
            "an answered ping must clear the exclusion"
        );
        assert!(
            registry
                .select_and_reserve("default", "general", "greet", None)?
                .map(|(worker, _reservation)| worker)
                .is_some(),
            "and the worker must be selectable again"
        );
        Ok(())
    }

    /// R1 census: after the last compatible worker leaves, the address reports
    /// how long ago it was served — the age an operator sees on the parked
    /// dispatch's WARN.
    #[test]
    fn a_departed_worker_leaves_a_last_compatible_poller_age() -> Result<(), ServerError> {
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = mpsc::channel(1);
        let worker = registry.register_namespaces(
            [String::from("default")],
            "general",
            None,
            [String::from("greet")].iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        worker.deregister()?;

        let census = registry.pool_census("default", "general", "greet", None)?;
        assert_eq!(census.workers_in_pool, 0);
        assert_eq!(census.compatible_workers, 0);
        let age = census
            .last_compatible_poller_age
            .ok_or_else(|| test_failure("a departed worker must leave an age behind"))?;
        assert!(
            age < Duration::from_secs(60),
            "the recorded departure is implausibly old: {age:?}"
        );

        // A worker returning to the address clears the departure record: the
        // map that answers "how long ago" never accumulates live addresses.
        let (tx, _rx) = mpsc::channel(1);
        let _back = registry.register_namespaces(
            [String::from("default")],
            "general",
            None,
            [String::from("greet")].iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let served = registry.pool_census("default", "general", "greet", None)?;
        assert_eq!(served.last_compatible_poller_age, Some(Duration::ZERO));
        assert!(served.is_served());
        Ok(())
    }
}