aion-server 0.31.0

//! Non-replayed background dispatcher for the durable fan-out outbox.
//!
//! # What this is
//!
//! `OutboxDispatcher` is a Tokio task that lives entirely OUTSIDE the
//! deterministic replay domain. It reads ONLY the outbox table — never workflow
//! history — claims pending rows under the single-writer model, dispatches each
//! to a connected worker through the existing server-side push
//! [`ActivityDispatcher`](crate::worker::ActivityDispatcher), and records the
//! row's terminal outbox state:
//!
//! - dispatch succeeds → [`OutboxStore::complete_outbox_row`] (`done`);
//! - dispatch fails and the attempt budget is not exhausted →
//!   [`OutboxStore::retry_outbox_row`] with an exponential-backoff
//!   `visible_after` fence and the attempt count bumped;
//! - dispatch fails on the final attempt → [`OutboxStore::fail_outbox_row`]
//!   (`failed`, a dead letter for operator inspection).
//!
//! # Dormant by default
//!
//! Nothing here runs unless the operator sets `outbox.enabled = true`. The
//! spawn in [`crate::run`] is gated on that flag, so a default server never
//! constructs or starts this task and its behaviour is identical to before the
//! outbox existed.
//!
//! # Phase boundary (Phase 2 vs Phase 3)
//!
//! # Placement (Control-Plane Phase 2, P2-P3)
//!
//! When a [`PlacementCache`](crate::worker::PlacementCache) is attached to
//! [`WorkerOutboxDispatch`], an UNPINNED row (`row.node == None`) consults its
//! namespace's placement for live worker selection:
//!
//! - a per-activity authored pin (`row.node == Some(N)`) always wins and is
//!   dispatched off the row's own node, untouched by placement;
//! - `Unplaced` is unchanged (any worker);
//! - `Prefer{L}` prefers an L-labelled worker, with a SPILL to ANY live worker
//!   when none of the preferred labels is up (the soft, high-availability tier);
//! - `Pinned{L}` (P2-I1) REQUIRES an L-labelled worker and WAITS on absence —
//!   it NEVER spills to a node=None any-worker dispatch. When none of the required
//!   labels has a live worker it holds on the wait-for-worker backstop and retries
//!   the whole set (the "isolation > availability" hard pin, CP-Phase-2 §2.5). The
//!   companion `Some(N ∉ L)` composition case is enforced by admission at
//!   `accept_registration` (only L-node workers ever serve a `Pinned{L}` namespace),
//!   so a conflicting activity simply finds no worker and hits the same stall.
//!
//! The determinism invariant is absolute: placement is consulted ONLY for live
//! worker selection in this non-replayed task; the recorded row's `node` is NEVER
//! mutated by placement, so a workflow's command stream is identical regardless of
//! which worker a `Prefer` or `Pinned` directive routed the activity to
//! (CP-Phase-2 §2.4).
//!
//! # Phase boundary (Phase 2 vs Phase 3)
//!
//! Phase 2 scope ends at the outbox row's terminal state. A *successful*
//! dispatch here means the activity task was accepted by a connected worker; it
//! does NOT route the eventual worker completion back into workflow history.
//! That completion → [`Recorder`](aion::Recorder) wiring (the cross-node
//! completion-dedup chokepoint) is Phase 3 and is deliberately NOT built in this
//! module. Until Phase 3 lands, the dispatcher is exercised only against the
//! outbox table; it must not be commissioned on a server that relies on fan-out
//! completions reaching history.

use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, Ordering};

mod capacity_park;
use capacity_park::CapacityParkedRows;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use aion_core::ActivityId;
use aion_store::{OutboxRow, OutboxStore};
use async_trait::async_trait;
use chrono::Utc;
use tokio::sync::watch;
use tracing::{error, info, warn};

use crate::error::ServerError;
use crate::worker::{ActivityDispatcher, OutboxDeliveryCallback, ScheduledActivity};

/// Shared ownership gate for claimed rows whose delivery tasks are still live.
///
/// The dispatcher guard, liminal held wait, and stale-claim reconciler all use
/// clones of the same handle. A key is present exactly while its spawned row
/// task owns a [`DeliveryGuard`].
#[derive(Clone, Debug, Default)]
pub struct DeliveryGate {
    inner: Arc<DeliveryGateInner>,
}

#[derive(Debug, Default)]
struct DeliveryGateInner {
    keys: Mutex<HashSet<String>>,
    draining: AtomicBool,
}

impl DeliveryGate {
    /// Begin delivery for `key`, returning `None` when it is already held, the
    /// dispatcher is draining, or the gate lock is poisoned.
    #[must_use]
    pub fn begin(&self, key: &str) -> Option<DeliveryGuard> {
        if self.is_draining() {
            return None;
        }
        let mut keys = match self.inner.keys.lock() {
            Ok(keys) => keys,
            Err(error) => {
                error!(%error, dispatch_key = %key, "delivery gate lock poisoned; refusing to begin delivery");
                return None;
            }
        };
        if self.is_draining() || !keys.insert(key.to_owned()) {
            return None;
        }
        Some(DeliveryGuard {
            gate: self.clone(),
            key: key.to_owned(),
        })
    }

    /// Whether `key` is still owned by a live delivery task.
    #[must_use]
    pub fn holds(&self, key: &str) -> bool {
        match self.inner.keys.lock() {
            Ok(keys) => keys.contains(key),
            Err(error) => {
                error!(%error, dispatch_key = %key, "delivery gate lock poisoned; abandoning held wait");
                false
            }
        }
    }

    /// Prevent new deliveries and tell held waits to abandon at their next poll.
    pub fn drain(&self) {
        self.inner.draining.store(true, Ordering::Release);
    }

    /// Whether dispatcher drain has begun.
    #[must_use]
    pub fn is_draining(&self) -> bool {
        self.inner.draining.load(Ordering::Acquire)
    }

    /// Snapshot keys that stale reconciliation must not re-arm.
    #[must_use]
    pub fn snapshot_keys(&self) -> HashSet<String> {
        match self.inner.keys.lock() {
            Ok(keys) => keys.clone(),
            Err(error) => {
                error!(%error, "delivery gate lock poisoned; preserving known stale-rearm exclusions");
                error.into_inner().clone()
            }
        }
    }
}

/// RAII ownership of one dispatch key in a [`DeliveryGate`].
#[derive(Debug)]
pub struct DeliveryGuard {
    gate: DeliveryGate,
    key: String,
}

impl Drop for DeliveryGuard {
    fn drop(&mut self) {
        match self.gate.inner.keys.lock() {
            Ok(mut keys) => {
                keys.remove(&self.key);
            }
            Err(error) => {
                error!(%error, dispatch_key = %self.key, "delivery gate lock poisoned; failed to release delivery key");
            }
        }
    }
}

/// Resolved, non-optional outbox dispatcher settings.
///
/// Built from the validated [`OutboxConfig`](crate::config::OutboxConfig) only
/// once the operator has commissioned the dispatcher, so every field is a
/// concrete operator decision — there are no defaults to invent here.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct OutboxDispatcherConfig {
    /// Interval between successive claim sweeps.
    pub poll_interval: Duration,
    /// Maximum rows claimed per sweep.
    pub batch_size: u32,
    /// Total dispatch attempts before a row is dead-lettered.
    pub max_attempts: u32,
    /// Base backoff applied to the first retry.
    pub backoff_base: Duration,
    /// Geometric growth factor applied per prior attempt.
    pub backoff_multiplier: u32,
    /// Upper bound on a single retry's backoff.
    pub backoff_max: Duration,
}

impl OutboxDispatcherConfig {
    /// Computes the backoff delay before the retry that follows `attempt`.
    ///
    /// `attempt` is the just-failed attempt count (zero-based for the first
    /// dispatch). The delay grows geometrically —
    /// `backoff_base * backoff_multiplier^attempt` — and is clamped to
    /// `backoff_max`. All arithmetic saturates rather than overflowing, so a
    /// large attempt count or multiplier simply pins the delay at the ceiling.
    #[must_use]
    pub fn backoff_for_attempt(&self, attempt: u32) -> Duration {
        let max_ms = u128::from(u64::MAX);
        let multiplier = u128::from(self.backoff_multiplier);
        let mut delay_ms = self.backoff_base.as_millis().min(max_ms);
        for _ in 0..attempt {
            delay_ms = delay_ms.saturating_mul(multiplier);
            if delay_ms >= max_ms {
                break;
            }
        }
        let cap_ms = self.backoff_max.as_millis().min(max_ms);
        let clamped_ms = delay_ms.min(cap_ms);
        Duration::from_millis(u64::try_from(clamped_ms).unwrap_or(u64::MAX))
    }
}

/// Abstraction over the push-dispatch of one claimed outbox row.
///
/// The production implementation forwards to the server's
/// [`ActivityDispatcher`]; tests substitute an in-test sink that records or
/// rejects dispatches deterministically without a connected worker. Modelling
/// dispatch as a trait keeps the claim/retry/terminal-state loop testable in
/// isolation from the gRPC worker registry.
#[async_trait]
pub trait OutboxRowDispatch: Send + Sync + 'static {
    /// Dispatch one claimed row to a worker.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError`] when the row cannot be placed with a worker. A
    /// returned error drives the row into retry (or dead-letter) rather than
    /// `done`.
    async fn dispatch(&self, row: &OutboxRow) -> Result<(), ServerError>;
}

/// Production [`OutboxRowDispatch`] backed by the connected-worker registry.
///
/// Maps an [`OutboxRow`] to a [`ScheduledActivity`] and pushes it through the
/// existing [`ActivityDispatcher`]. Since NSTQ-2 the row carries its own
/// `namespace` and `task_queue`, so dispatch routes via the workflow's real
/// routing identity read straight off the row — no server default is injected.
/// Legacy rows persisted before NSTQ-2 read back as the `"default"` namespace and
/// `"default"` task queue at the store-read layer, so the fallback lives there,
/// not here.
pub struct WorkerOutboxDispatch {
    dispatcher: ActivityDispatcher,
    /// Optional short-TTL per-namespace placement cache (Control-Plane Phase 2,
    /// P2-P3). When present, an UNPINNED row (`row.node == None`) whose namespace
    /// placement is `Prefer{L}` dispatches preferring an L-labelled worker and
    /// spills to any live worker when none is up. When absent (the default, every
    /// pre-Phase-2 construction and test) dispatch is byte-identical: every row
    /// goes straight through [`ActivityDispatcher::dispatch`] off the row's own
    /// `node`. Placement is NEVER stamped back onto the row — it is consulted only
    /// here, in the non-replayed dispatcher, for worker selection.
    placement_cache: Option<crate::worker::PlacementCache>,
}

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

impl WorkerOutboxDispatch {
    /// Build a worker-backed dispatch over the shared push dispatcher.
    #[must_use]
    pub fn new(dispatcher: ActivityDispatcher) -> Self {
        Self {
            dispatcher,
            placement_cache: None,
        }
    }

    /// Attach the per-namespace placement cache so an unpinned row consults its
    /// namespace's `Prefer` directive at dispatch time (Control-Plane Phase 2,
    /// P2-P3). Pure builder addition: without it, dispatch is byte-identical to
    /// the pre-Phase-2 behaviour.
    #[must_use]
    pub fn with_placement_cache(mut self, cache: crate::worker::PlacementCache) -> Self {
        self.placement_cache = Some(cache);
        self
    }

    /// Translate an outbox row into the wire-bound scheduled activity.
    ///
    /// The routing identity (`namespace`, `task_queue`, optional `node`) is read
    /// off the row, so the activity dispatches into the workflow's real namespace
    /// pool with any node affinity the row carries (NODE-2). The pinned
    /// `ordinal` is the per-workflow activity ordinal recorded in history, so it
    /// maps directly onto the activity id the worker correlates its result
    /// against; the wire `attempt` is the row's `started_attempt` — the
    /// one-based activity attempt its `ActivityStarted` recorded (NOI-0) —
    /// never the row's zero-based delivery count, which is the retry budget
    /// and nothing more.
    fn to_scheduled(row: &OutboxRow) -> ScheduledActivity {
        ScheduledActivity {
            namespace: row.namespace.clone(),
            task_queue: row.task_queue.clone(),
            activity_type: row.activity_type.clone(),
            // node affinity is sourced off the row (NODE-2): `Some(node)` pins the
            // dispatch to workers on that node; `None` = unpinned = any worker in
            // the pool. There is no SDK-level node selection yet (NODE-4), so the
            // row carries `None` today, but the dispatcher no longer hard-codes it.
            node: row.node.clone(),
            workflow_id: row.workflow_id.clone(),
            activity_id: ActivityId::from_sequence_position(row.ordinal),
            run_id: row.run_id.clone(),
            input: row.input.clone(),
            // NOI-0: the wire carries the attempt history recorded for this
            // dispatch — never the row's delivery count. A transport redelivery
            // therefore re-sends the SAME attempt, and the lease it records is a
            // second lease for that attempt, not a new attempt no start recorded.
            attempt: row.started_attempt,
            labels: std::collections::BTreeMap::new(),
            // DECLARED: this dispatch belongs to the outbox pass that holds the
            // row's claim in the delivery gate, so a delivery still waiting must
            // stop the moment that claim is lost. The dispatcher cannot infer
            // this and must not have to.
            origin: crate::worker::dispatch::DispatchOrigin::OutboxRow {
                dispatch_key: row.dispatch_key.clone(),
            },
        }
    }
}

#[async_trait]
impl OutboxRowDispatch for WorkerOutboxDispatch {
    async fn dispatch(&self, row: &OutboxRow) -> Result<(), ServerError> {
        let scheduled = Self::to_scheduled(row);
        // A per-activity authored pin (`row.node == Some(N)`) ALWAYS wins and is
        // untouched: dispatch straight off the row's own node. Only an UNPINNED row
        // consults namespace placement (CP-Phase-2 §2.2 composition rule).
        let (Some(cache), None) = (&self.placement_cache, &scheduled.node) else {
            return self.dispatcher.dispatch(&scheduled).await;
        };
        match cache.placement(&scheduled.namespace).await {
            // SOFT placement: prefer an L-labelled worker, spill to any live one.
            // The row's `node` stays `None` throughout — preference is a pure
            // dispatch-time selection input, never written back (the determinism
            // invariant, CP-Phase-2 §2.4).
            aion_store::NamespacePlacement::Prefer { nodes } => {
                self.dispatcher
                    .dispatch_preferring(&scheduled, &nodes)
                    .await
            }
            // HARD placement (P2-I1): require an L-labelled worker and WAIT on
            // absence — NEVER spill to a node=None any-worker dispatch. The row's
            // `node` stays `None` throughout; the required set is a pure
            // dispatch-time selection input (the determinism invariant,
            // CP-Phase-2 §2.4/§2.5).
            aion_store::NamespacePlacement::Pinned { nodes } => {
                self.dispatcher.dispatch_requiring(&scheduled, &nodes).await
            }
            // Unplaced: today's behaviour — the unchanged any-worker dispatch.
            aion_store::NamespacePlacement::Unplaced => self.dispatcher.dispatch(&scheduled).await,
        }
    }
}

/// Typestate marking a dispatcher whose required row sink is not installed yet.
#[derive(Debug)]
pub struct MissingDispatch;

/// Non-replayed background dispatcher for pending outbox rows.
///
/// See the module docs for the full contract. Construct with [`Self::new`] and
/// drive with [`Self::run`]; the loop exits cleanly when the shared shutdown
/// watch flips to `true`, mirroring the server's transport shutdown signal.
pub struct OutboxDispatcher<D = MissingDispatch> {
    store: Arc<dyn OutboxStore>,
    dispatch: D,
    config: OutboxDispatcherConfig,
    /// Advisory wake (LSUB-2): an in-process `Notify` the engine's stage seam
    /// pulses when a pending outbox row is committed, so the run loop sweeps
    /// ~immediately instead of waiting up to one poll interval. The wake is
    /// strictly advisory — a dropped, coalesced, or absent wake degrades cleanly
    /// to the interval poll, which remains the correctness backstop. When no wake
    /// is wired in (the default), this is a private `Notify` that is never pulsed,
    /// so the loop behaves exactly as a pure poll.
    wake: Arc<tokio::sync::Notify>,
    /// Rows this dispatcher answered BUSY, keyed by dispatch key to the attempt
    /// they were parked at.
    ///
    /// A busy row is waiting for CAPACITY, and capacity-freeing is an event this
    /// server produces — so it is re-offered on that event rather than on a
    /// clock. Its durable `visible_after` still carries the failure backoff and
    /// remains the crash-safe fallback: if this process dies, the row is
    /// re-claimed on the timer as before, and nothing here is load-bearing for
    /// correctness. What it buys is that a healthy fan onto a healthy worker no
    /// longer waits out the operator's FAILURE backoff for a condition that
    /// clears in the time one activity takes.
    capacity_parked: Arc<CapacityParkedRows>,
    /// Optional per-tenant keyed backpressure at the claim (Control-Plane Phase 2,
    /// P2-Q2). When present, [`Self::sweep_once`] replaces the single unscoped
    /// `claim_outbox_rows(batch_size)` with a per-namespace, round-robin,
    /// headroom-capped claim, so a tenant at its concurrency ceiling has its excess
    /// Pending rows held (left durable, reconsidered next sweep) and a bursty tenant
    /// cannot starve a quiet one. When absent (the default, every pre-Phase-2
    /// construction and test) the sweep is byte-identical: one unscoped claim. With
    /// the generous platform-default ceiling and no tenant override, the ceiling
    /// never engages, so an attached-but-default backpressure is also behaviourally
    /// identical to no backpressure for normal load.
    backpressure: Option<crate::worker::Backpressure>,
    /// Optional durable pause dispatch-hold (#204). When present, each claim
    /// excludes rows whose workflow is in the paused set, so a held run's rows
    /// stay `Pending` — never `Claimed` — for the whole paused window; release is
    /// purely resume plus the next sweep. When absent (the default, every
    /// pre-#204 construction and test) the claim is byte-identical to before.
    paused_runs: Option<aion::lifecycle::PausedRuns>,
    /// Shared ownership of rows currently executing in spawned delivery tasks.
    delivery_gate: DeliveryGate,
    /// Optional callback that makes a genuine terminal delivery failure visible
    /// to the owning workflow after the row is durably dead-lettered.
    failure_callback: Option<Arc<dyn OutboxDeliveryCallback>>,
}

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

impl OutboxDispatcher<MissingDispatch> {
    /// Begin building a dispatcher and create its shared delivery gate.
    #[must_use]
    pub fn new(store: Arc<dyn OutboxStore>, config: OutboxDispatcherConfig) -> Self {
        Self {
            store,
            dispatch: MissingDispatch,
            config,
            wake: Arc::new(tokio::sync::Notify::new()),
            capacity_parked: Arc::new(CapacityParkedRows::default()),
            backpressure: None,
            paused_runs: None,
            delivery_gate: DeliveryGate::default(),
            failure_callback: None,
        }
    }

    /// Install the required row sink, transitioning to a runnable dispatcher.
    #[must_use]
    pub fn with_dispatch(
        self,
        dispatch: Arc<dyn OutboxRowDispatch>,
    ) -> OutboxDispatcher<Arc<dyn OutboxRowDispatch>> {
        OutboxDispatcher {
            store: self.store,
            dispatch,
            config: self.config,
            wake: self.wake,
            capacity_parked: self.capacity_parked,
            backpressure: self.backpressure,
            paused_runs: self.paused_runs,
            delivery_gate: self.delivery_gate,
            failure_callback: self.failure_callback,
        }
    }
}

impl<D> OutboxDispatcher<D> {
    /// Return the dispatcher-owned gate shared with transport and reconciler.
    #[must_use]
    pub fn delivery_gate(&self) -> DeliveryGate {
        self.delivery_gate.clone()
    }
}

impl OutboxDispatcher<Arc<dyn OutboxRowDispatch>> {
    /// Attach the callback used to surface a dead-letter as workflow failure.
    #[must_use]
    pub fn with_delivery_callback(mut self, callback: Arc<dyn OutboxDeliveryCallback>) -> Self {
        self.failure_callback = Some(callback);
        self
    }

    /// Attach the durable pause dispatch-hold set (#204).
    ///
    /// With it, each sweep excludes rows whose owning workflow is currently
    /// held (paused), so those rows are never claimed while paused. Pure builder
    /// addition: without it the claim is byte-identical to before.
    #[must_use]
    pub fn with_paused_runs(mut self, paused_runs: aion::lifecycle::PausedRuns) -> Self {
        self.paused_runs = Some(paused_runs);
        self
    }

    /// Install the shared advisory wake (LSUB-2).
    ///
    /// The supplied `Notify` is pulsed by the engine's append-with-outbox seam
    /// when a pending row is committed, so the run loop sweeps promptly rather
    /// than waiting for the next interval tick. For that use a lost wake simply
    /// reverts to poll latency.
    ///
    /// It is also pulsed by the worker registry whenever a capacity slot frees
    /// (`ConnectedWorkerRegistry::with_capacity_wake`), and for THAT use the
    /// interval poll is not a backstop: a capacity-parked row's `visible_after`
    /// is the failure backoff, which the poll's `visible_after <= now` claim
    /// cannot satisfy. See the wake arm in [`Self::run`].
    #[must_use]
    pub fn with_wake(mut self, wake: Arc<tokio::sync::Notify>) -> Self {
        self.wake = wake;
        self
    }

    /// Attach per-tenant keyed backpressure at the claim (Control-Plane Phase 2,
    /// P2-Q2).
    ///
    /// With it, each sweep claims per-namespace, round-robin, capped at each
    /// tenant's CLAIMED-only headroom (`per_node_ceiling − claimed`) and a fair
    /// share of the batch, instead of one unscoped `claim_outbox_rows(batch_size)`.
    /// Pure builder addition: without it (the default) the sweep is byte-identical
    /// to the pre-Phase-2 single unscoped claim, and even WITH it a default-ceiling
    /// deployment with no tenant override never engages the ceiling for normal load.
    #[must_use]
    pub fn with_backpressure(mut self, backpressure: crate::worker::Backpressure) -> Self {
        self.backpressure = Some(backpressure);
        self
    }

    /// Run the claim/dispatch loop until `shutdown` flips to `true`.
    ///
    /// Each tick claims up to `batch_size` pending rows and spawns an independent
    /// delivery task for each one. A backend error claiming rows is logged and the loop waits for the
    /// next tick rather than tearing the task down — a transient store failure
    /// must not silently stop the dispatcher. Shutdown is observed both while
    /// waiting for the next tick and is re-checked before each sweep, so a
    /// drain never blocks on an in-progress wait.
    pub async fn run(self, mut shutdown: watch::Receiver<bool>) {
        let this = Arc::new(self);
        info!(
            poll_interval_ms = this.config.poll_interval.as_millis(),
            batch_size = this.config.batch_size,
            max_attempts = this.config.max_attempts,
            "outbox dispatcher started"
        );
        let mut interval = tokio::time::interval(this.config.poll_interval);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        loop {
            tokio::select! {
                _ = interval.tick() => {
                    if *shutdown.borrow() {
                        break;
                    }
                    Self::sweep_once(&this).await;
                }
                // Two kinds of news arrive here. A STAGE-SEAM pulse says a row
                // was committed, and for that one the interval tick really is a
                // backstop: a dropped wake costs poll latency and nothing else.
                //
                // 🔴 A CAPACITY pulse — a worker slot freeing — is NOT backstopped
                // by the poll. A capacity-parked row carries a `visible_after`
                // set to the operator's FAILURE backoff, and the poll claims on
                // `visible_after <= now`, so it cannot pick that row up: the only
                // thing that re-offers it is `reoffer_capacity_parked`, called
                // from this arm and nowhere else. A capacity wake that went
                // missing would cost the full failure backoff — 50 ms on
                // defaults, minutes or hours where an operator has raised it —
                // which is the exact outcome the capacity wake exists to prevent.
                //
                // That is why the busy arm parks the row BEFORE its durable
                // re-arm write rather than after: a pulse landing anywhere in
                // that write's window then finds the row — as PENDING, which
                // this arm records on the entry and does not act on; the busy
                // arm's confirmation, once its write has returned, reads that
                // record and re-offers immediately. Nothing here re-checks:
                // the park ordering plus that pending→confirmed handshake is
                // what makes this safe, and neither half stands alone.
                //
                // Re-check shutdown first, exactly like the interval arm, so a
                // wake never races a drain.
                () = this.wake.notified() => {
                    if *shutdown.borrow() {
                        break;
                    }
                    // The wake now carries two kinds of news: a newly-staged row
                    // (the stage seam) and a FREED CAPACITY SLOT (the registry).
                    // Re-offering first costs nothing when nothing is parked and
                    // is the whole point when something is: a row that was told
                    // the pool was busy is waiting on exactly this event.
                    this.reoffer_capacity_parked().await;
                    Self::sweep_once(&this).await;
                }
                changed = shutdown.changed() => {
                    // A send error means every sender dropped; treat that as a
                    // shutdown request rather than spinning.
                    if changed.is_err() || *shutdown.borrow() {
                        break;
                    }
                }
            }
        }
        this.delivery_gate.drain();
        info!("outbox dispatcher stopped");
    }

    /// Claim one batch and spawn an independent delivery task for every row.
    async fn sweep_once(self: &Arc<Self>) {
        // LSUB-4-3 / Fork-A2 seam: ownership is NOT enforced on this claim. The
        // claim path is an UNFENCED local `put_routed` scoped by `owned_shard_scope()`
        // — it simply returns `Ok` with only the rows on shards this node owns and
        // never surfaces a `NotOwner`. Deposition is surfaced (and the owned set
        // narrowed by re-residency) on the FENCED stamped event-append a deposed
        // owner attempts when recording a terminal (aion-store-haematite store.rs
        // ~622/1110: `DatabaseError::Fenced => StoreError::NotOwner`), not here. So
        // this sweep has nothing ownership-specific to handle: a claim error is a
        // genuine backend failure, retried next tick.
        let rows = match self.claim_rows().await {
            Ok(rows) => rows,
            Err(error) => {
                error!(%error, "outbox dispatcher failed to claim rows; retrying next tick");
                return;
            }
        };
        #[cfg(test)]
        let mut tasks = Vec::new();
        for row in rows {
            let Some(guard) = self.delivery_gate.begin(&row.dispatch_key) else {
                warn!(
                    dispatch_key = %row.dispatch_key,
                    draining = self.delivery_gate.is_draining(),
                    "outbox row is already in flight or delivery gate is unavailable; leaving it claimed"
                );
                continue;
            };
            let task = Arc::clone(self);
            let handle = tokio::spawn(async move {
                task.process_row(&row).await;
                drop(guard);
            });
            #[cfg(test)]
            tasks.push(handle);
            #[cfg(not(test))]
            drop(handle);
        }
        #[cfg(test)]
        for task in tasks {
            if let Err(error) = task.await {
                error!(
                    %error,
                    "outbox dispatcher row task failed while a unit test awaited the spawned sweep"
                );
            }
        }
    }

    /// Claim this sweep's rows, applying per-tenant keyed backpressure when it is
    /// attached (Control-Plane Phase 2, P2-Q2) and otherwise the unchanged single
    /// unscoped claim.
    ///
    /// The backpressure path round-robins a scoped, headroom-capped claim per
    /// namespace-with-pending-work — rows over a tenant's CLAIMED-only ceiling are
    /// left durably `Pending`, reconsidered next sweep — but reuses the SAME atomic
    /// [`OutboxStore::claim_outbox_rows_scoped`] semantics, so exactly-once and the
    /// durable-outbox guarantees are untouched: only the `limit` and `scope` are
    /// quota-derived (a smaller claim is already first-class).
    async fn claim_rows(&self) -> Result<Vec<OutboxRow>, aion_store::StoreError> {
        // The pause dispatch-hold (#204) excludes held workflows at claim time so
        // their rows stay Pending. With no held set (or an empty one) both paths are
        // byte-identical to the plain claim. The exclusion is applied on BOTH the
        // backpressure (scoped) and the plain (unscoped) claim, because production
        // wires backpressure unconditionally and would otherwise bypass the hold.
        let held = match &self.paused_runs {
            Some(paused_runs) => paused_runs.snapshot(),
            None => std::collections::HashSet::new(),
        };
        match &self.backpressure {
            Some(backpressure) => {
                backpressure
                    .claim_round_robin(&self.store, self.config.batch_size, &held)
                    .await
            }
            None => {
                self.store
                    .claim_outbox_rows_excluding(self.config.batch_size, &held)
                    .await
            }
        }
    }

    /// Dispatch one claimed row and record its terminal outbox state.
    async fn process_row(&self, row: &OutboxRow) {
        match self.dispatch.dispatch(row).await {
            Ok(()) => self.mark_done(row).await,
            Err(error) if self.delivery_gate.is_draining() => {
                warn!(
                    dispatch_key = %row.dispatch_key,
                    %error,
                    "outbox delivery abandoned during shutdown; leaving row claimed for recovery"
                );
            }
            Err(error) => self.handle_dispatch_error(row, &error).await,
        }
    }

    async fn mark_done(&self, row: &OutboxRow) {
        if let Err(error) = self.store.complete_outbox_row(&row.dispatch_key).await {
            // The dispatch already happened; failing to persist `done` leaves
            // the row `claimed` (never re-claimed), so log loudly for operators.
            error!(
                dispatch_key = %row.dispatch_key,
                %error,
                "outbox dispatcher dispatched a row but failed to mark it done"
            );
        }
    }

    /// Retire the row as a dead letter and surface the failure to its workflow.
    ///
    /// The whole decision — including the durable judgment marker that decides whether an operator
    /// may later redrive the row — lives in
    /// [`outbox_dead_letter`](crate::worker::outbox_dead_letter).
    async fn dead_letter(&self, row: &OutboxRow, dispatch_error: &ServerError, attempted: u32) {
        crate::worker::outbox_dead_letter::dead_letter_row(
            &self.store,
            self.failure_callback.as_ref(),
            row,
            dispatch_error,
            attempted,
            self.config.max_attempts,
        )
        .await;
    }

    /// Re-offer every row that was parked for want of CAPACITY, because a slot
    /// has just been freed.
    ///
    /// Called from the wake arm of the run loop, whose `Notify` the registry
    /// pulses on every slot-free (a completed dispatch's untrack, a released
    /// reservation). Each row's `visible_after` is pulled forward to now so the
    /// sweep that follows can claim it; the attempt is passed through unchanged,
    /// so this costs the row nothing — it is the same attempt-neutral re-arm,
    /// just on the event instead of the timer.
    ///
    /// Draining rather than iterating: a row that is still busy is re-added by
    /// the busy arm on its next pass, so this set never accumulates rows that
    /// have stopped needing it.
    async fn reoffer_capacity_parked(&self) {
        // CONFIRMED rows only. A row whose durable park is still in flight is
        // left where it is and flagged; `confirm` re-offers it the instant its
        // write lands, so the pulse is deferred rather than dropped and this
        // never issues a second write against a key the busy arm is still
        // writing.
        for (dispatch_key, attempt) in self.capacity_parked.take_confirmed() {
            self.reoffer_row(&dispatch_key, attempt).await;
        }
    }

    /// Pull one parked row's `visible_after` forward to now, so the sweep that
    /// follows can claim it.
    ///
    /// The attempt is passed through unchanged: a wait for capacity is not a
    /// failed delivery and must never cost the row an attempt.
    async fn reoffer_row(&self, dispatch_key: &str, attempt: u32) {
        if let Err(error) = self
            .store
            .retry_outbox_row(dispatch_key, attempt, Utc::now())
            .await
        {
            error!(
                dispatch_key,
                %error,
                "capacity-freed re-offer failed; the row waits for its durable fence"
            );
        }
    }

    /// Apply the retry budget after a failed dispatch.
    ///
    /// The just-failed attempt is `row.attempt` (zero-based). If a further
    /// attempt remains within `max_attempts`, the row is returned to `pending`
    /// with the attempt bumped and a `visible_after` fence; otherwise it is
    /// dead-lettered to `failed`.
    ///
    /// # Fast cross-node failover (LSUB-3)
    ///
    /// When the failure is [`ServerError::WorkerConnectionLost`] — the chosen
    /// worker died mid-dispatch and liminal has already deregistered it — the row
    /// is re-armed for IMMEDIATE re-claim (`visible_after = now`, no backoff) so the
    /// next sweep promptly re-dispatches it to a live worker in the pool. The
    /// attempt is STILL consumed: this is the deliberate policy choice — immediate
    /// re-claim but attempt-consuming — so pathological worker churn stays bounded
    /// by `max_attempts` and eventually dead-letters rather than forming an
    /// unbounded re-dispatch loop. A genuine reply timeout (the worker is alive but
    /// slow) and every other error keep the normal exponential backoff unchanged.
    async fn handle_dispatch_error(&self, row: &OutboxRow, dispatch_error: &ServerError) {
        // A connection cap is a typed refusal before the worker accepted this
        // dispatch. It is congestion, not a failed delivery attempt, so it must
        // never consume the genuine-failure retry budget or dead-letter work.
        if dispatch_error.is_worker_busy() {
            let backoff = self.config.backoff_base;
            let visible_after = Utc::now() + chrono_duration(backoff);
            warn!(
                dispatch_key = %row.dispatch_key,
                attempt = row.attempt,
                backoff_ms = backoff.as_millis(),
                error = %dispatch_error,
                "outbox worker is busy; re-arming without consuming an attempt"
            );
            // REMEMBERED FIRST, BEFORE THE DURABLE WRITE, and that ordering is
            // the whole of it.
            //
            // The write below is awaited and store-wide — a Haematite write, not
            // microseconds. A slot freeing inside that window pulses the wake,
            // `reoffer_capacity_parked` runs against a set that does not yet
            // contain this row, the sweep finds nothing claimable because the
            // row's fence is still far in the future, and the permit is spent.
            // The row is then inserted with nothing left to wake it, and waits
            // out the operator's FAILURE backoff — 50 ms on defaults, two
            // minutes in the cross-node fixture, an hour in this file's own D2
            // test. That is precisely the outcome D2 exists to prevent,
            // reinstated by a narrow race.
            //
            // Nothing re-checks: `reoffer_capacity_parked` has ONE call site
            // (the wake arm), and the interval poll is not a backstop for this
            // case because it claims on `visible_after <= now`, which a
            // capacity-parked row fails by construction.
            //
            // Inserting first means a pulse landing anywhere in the window finds
            // the row. It is removed again if the write fails, so a row that was
            // never actually parked is never re-offered.
            self.capacity_parked
                .park_pending(&row.dispatch_key, row.attempt);
            if let Err(error) = self
                .store
                .retry_outbox_row(&row.dispatch_key, row.attempt, visible_after)
                .await
            {
                // The park never became durable, so withdraw it: a row left
                // remembered with no fence behind it would be re-offered on
                // every freed slot, pulling forward a `visible_after` that was
                // never written.
                self.capacity_parked.abandon(&row.dispatch_key);
                error!(
                    dispatch_key = %row.dispatch_key,
                    %error,
                    "outbox dispatcher failed to re-arm worker-busy row"
                );
                return;
            }
            // CONFIRMED, and the re-check is the other half of the protocol: a
            // pulse that arrived while this park was in flight was recorded on
            // the entry rather than acted on, because acting on it would have
            // raced this write. If one did, re-offer immediately — no further
            // wake is coming for this row.
            if self.capacity_parked.confirm(&row.dispatch_key) {
                self.reoffer_row(&row.dispatch_key, row.attempt).await;
            }
            return;
        }

        let attempted = row.attempt.saturating_add(1);
        // A frame liminal proved larger than the connection's whole outbound
        // buffer can never be delivered, on this attempt or any later one, so it
        // is dead-lettered on the FIRST observation. Spending the retry budget on
        // it would write one more lease and one more copy of the oversize input
        // per attempt for a step that can never run.
        if dispatch_error.is_worker_dispatch_unservable() {
            warn!(
                dispatch_key = %row.dispatch_key,
                attempt = row.attempt,
                error = %dispatch_error,
                "outbox dispatch is unservable at the connection's outbound bound; dead-lettering \
                 without retry"
            );
            self.dead_letter(row, dispatch_error, attempted).await;
            return;
        }
        if attempted >= self.config.max_attempts {
            self.dead_letter(row, dispatch_error, attempted).await;
            return;
        }
        // LSUB-3 fast failover: a lost worker connection re-arms for immediate
        // re-claim (skip backoff); everything else keeps the backoff curve.
        if dispatch_error.is_worker_connection_lost() {
            let visible_after = Utc::now();
            warn!(
                dispatch_key = %row.dispatch_key,
                attempt = row.attempt,
                next_attempt = attempted,
                error = %dispatch_error,
                "outbox dispatch lost the worker connection; re-arming for immediate failover"
            );
            if let Err(error) = self
                .store
                .retry_outbox_row(&row.dispatch_key, attempted, visible_after)
                .await
            {
                error!(dispatch_key = %row.dispatch_key, %error, "outbox dispatcher failed to re-arm row for failover");
            }
            return;
        }
        let backoff = self.config.backoff_for_attempt(row.attempt);
        let visible_after = Utc::now() + chrono_duration(backoff);
        warn!(
            dispatch_key = %row.dispatch_key,
            attempt = row.attempt,
            next_attempt = attempted,
            backoff_ms = backoff.as_millis(),
            error = %dispatch_error,
            "outbox dispatch failed; scheduling retry with backoff"
        );
        if let Err(error) = self
            .store
            .retry_outbox_row(&row.dispatch_key, attempted, visible_after)
            .await
        {
            error!(dispatch_key = %row.dispatch_key, %error, "outbox dispatcher failed to schedule retry");
        }
    }
}

/// Convert a (non-negative) [`Duration`] into a [`chrono::Duration`], saturating
/// at the chrono maximum rather than failing — the backoff curve is already
/// clamped to `backoff_max`, so this only guards the type boundary.
fn chrono_duration(duration: Duration) -> chrono::Duration {
    chrono::Duration::from_std(duration).unwrap_or(chrono::Duration::MAX)
}

#[cfg(test)]
mod tests {
    use super::WorkerOutboxDispatch;
    use std::path::PathBuf;
    use std::sync::Arc;
    use std::sync::Mutex;
    use std::time::{Duration, SystemTime, UNIX_EPOCH};

    use aion_core::{ActivityId, ContentType, Payload, RunId, WorkflowId};
    use aion_store::{OutboxRow, OutboxStatus, OutboxStore};
    use aion_store_haematite::HaematiteStore;
    use async_trait::async_trait;
    use chrono::Utc;

    use crate::error::CompletionRejectionReason;
    use crate::worker::envelope::{CompletionFences, CompletionToken};

    use super::{OutboxDispatcher, OutboxDispatcherConfig, OutboxRowDispatch, ServerError};

    fn config() -> OutboxDispatcherConfig {
        OutboxDispatcherConfig {
            poll_interval: Duration::from_millis(10),
            batch_size: 16,
            max_attempts: 3,
            backoff_base: Duration::from_millis(100),
            backoff_multiplier: 2,
            backoff_max: Duration::from_secs(60),
        }
    }

    fn unique_temp_path(name: &str) -> PathBuf {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0, |duration| duration.as_nanos());
        std::env::temp_dir().join(format!(
            "aion-server-outbox-dispatcher-{name}-{}-{nanos}.db",
            std::process::id()
        ))
    }

    /// One shared store handle drives both the dispatcher and the test
    /// assertions, so they observe the same rows without a second connection.
    async fn open_store(name: &str) -> Result<Arc<HaematiteStore>, ServerError> {
        // test-ruled patience: 250ms covers the measured 93-150ms fork window; not a default.
        HaematiteStore::open_or_create(
            unique_temp_path(name),
            haematite::NodeCacheBudget::Unlimited,
        )
        .await
        .map(Arc::new)
        .map_err(ServerError::from)
    }

    /// 🔴 NOI-0 at the wire: the dispatched attempt is the row's recorded
    /// attempt, whatever the delivery count says. A row redelivered twice
    /// (budget 2) still dispatches attempt 1 — the same attempt its
    /// `ActivityStarted` recorded — so the lease and completion it produces
    /// name an attempt history holds; a re-staged row at recorded attempt 4
    /// dispatches 4.
    #[test]
    fn the_wire_attempt_is_the_recorded_attempt_not_the_delivery_count() {
        let workflow_id = WorkflowId::new(uuid::Uuid::from_u128(0x77));
        let mut redelivered = pending_row(&workflow_id, 0);
        redelivered.attempt = 2;
        assert_eq!(WorkerOutboxDispatch::to_scheduled(&redelivered).attempt, 1);

        let restaged = pending_row(&workflow_id, 1).with_started_attempt(4);
        assert_eq!(WorkerOutboxDispatch::to_scheduled(&restaged).attempt, 4);
    }

    fn pending_row(workflow_id: &WorkflowId, ordinal: u64) -> OutboxRow {
        OutboxRow::pending(
            workflow_id.clone(),
            ordinal,
            String::from("charge"),
            Payload::new(ContentType::Json, b"{}".to_vec()),
            Utc::now(),
        )
        .with_run_id(Some(RunId::new_v4()))
    }

    /// Records every dispatched row; configurable to always succeed or always fail.
    struct RecordingDispatch {
        succeed: bool,
        dispatched: Mutex<Vec<OutboxRow>>,
    }

    impl RecordingDispatch {
        fn new(succeed: bool) -> Self {
            Self {
                succeed,
                dispatched: Mutex::new(Vec::new()),
            }
        }

        fn count(&self) -> Result<usize, ServerError> {
            Ok(self
                .dispatched
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording dispatch"))?
                .len())
        }
    }

    #[async_trait]
    impl OutboxRowDispatch for RecordingDispatch {
        async fn dispatch(&self, row: &OutboxRow) -> Result<(), ServerError> {
            self.dispatched
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording dispatch"))?
                .push(row.clone());
            if self.succeed {
                Ok(())
            } else {
                Err(ServerError::worker_dispatch(
                    "default",
                    "charge",
                    "no worker in test",
                ))
            }
        }
    }

    /// #204 GATE-1 (dispatch-hold mechanism): while a workflow is in the paused
    /// set, its pending outbox row is NEVER claimed — it stays `Pending` across
    /// sweeps and no dispatch reaches the worker. After the workflow is removed
    /// from the paused set (resume), the ordinary sweep claims the held row with
    /// no bespoke re-arm and it dispatches to completion.
    #[tokio::test]
    async fn paused_run_row_is_held_pending_then_dispatches_after_release()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("pause-hold").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let paused_runs = aion::lifecycle::PausedRuns::default();
        paused_runs.insert(workflow_id.clone());

        let dispatch = Arc::new(RecordingDispatch::new(true));
        let dispatcher = OutboxDispatcher::new(store.clone(), config())
            .with_dispatch(dispatch.clone())
            .with_paused_runs(paused_runs.clone());
        let dispatcher = Arc::new(dispatcher);

        // Sweep repeatedly while paused: the row is never claimed, never
        // dispatched, and remains Pending the whole window.
        for _ in 0..3 {
            dispatcher.sweep_once().await;
        }
        assert_eq!(
            dispatch.count()?,
            0,
            "no dispatch reaches any worker while the run is paused"
        );
        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|s| s.status),
            Some(OutboxStatus::Pending),
            "the held row stays Pending (never Claimed) for the whole paused window"
        );

        // Release (resume): remove from the hold. The next ordinary sweep claims
        // the held row with no bespoke re-arm and dispatches it to Done.
        paused_runs.remove(&workflow_id);
        dispatcher.sweep_once().await;
        assert_eq!(dispatch.count()?, 1, "the released row dispatches");
        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|s| s.status),
            Some(OutboxStatus::Done),
            "the released row completes after resume"
        );
        Ok(())
    }

    #[tokio::test]
    async fn sweep_dispatches_claimed_rows_and_marks_them_done()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("done").await?;
        let workflow_id = WorkflowId::new_v4();
        let row_a = pending_row(&workflow_id, 0);
        let row_b = pending_row(&workflow_id, 1);
        store
            .append_outbox_batch(&[row_a.clone(), row_b.clone()])
            .await?;

        let dispatch = Arc::new(RecordingDispatch::new(true));
        let dispatcher =
            OutboxDispatcher::new(store.clone(), config()).with_dispatch(dispatch.clone());
        let dispatcher = Arc::new(dispatcher);
        // Drive a single sweep directly (no timer) for deterministic assertions.
        dispatcher.sweep_once().await;

        assert_eq!(dispatch.count()?, 2, "both pending rows are dispatched");
        assert_eq!(
            store
                .outbox_row_state(&row_a.dispatch_key)
                .await?
                .map(|s| s.status),
            Some(OutboxStatus::Done)
        );
        assert_eq!(
            store
                .outbox_row_state(&row_b.dispatch_key)
                .await?
                .map(|s| s.status),
            Some(OutboxStatus::Done)
        );
        // Nothing is claimable after a successful sweep.
        assert!(store.claim_outbox_rows(10).await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn failed_dispatch_retries_with_backoff_and_bumps_attempt()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("retry").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        // A backoff wide enough that the fence assertion below measures the
        // FENCE and not the machine: with the shared 100ms base, a loaded box can
        // spend longer than the window between the retry write and the claim, and
        // the row legitimately becomes claimable again. The subject is "a fence
        // exists and holds", not the size of the interval.
        let mut config = config();
        config.backoff_base = Duration::from_secs(30);
        let before = Utc::now();
        let dispatch = Arc::new(RecordingDispatch::new(false));
        let seam: Arc<dyn OutboxRowDispatch> = Arc::clone(&dispatch) as Arc<dyn OutboxRowDispatch>;
        let dispatcher = OutboxDispatcher::new(store.clone(), config).with_dispatch(seam);
        let dispatcher = Arc::new(dispatcher);
        dispatcher.sweep_once().await;

        // The sweep CLAIMED the staged row and took it to the dispatch seam, even
        // though that seam could not deliver it. This is what "the dispatcher
        // works the table while no worker can receive" means: a connected worker
        // is not a precondition for claiming. `OutboxRowDispatch::dispatch`
        // reports no-worker as an ordinary dispatch error, which is exactly what
        // this double returns, so the production no-worker path lands here too.
        assert_eq!(
            dispatch.count()?,
            1,
            "the sweep must claim the staged row and offer it to the dispatch seam"
        );

        let state = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("retried row must still exist")?;
        // Returned to pending, attempt bumped from 0 to 1.
        assert_eq!(state.status, OutboxStatus::Pending);
        assert_eq!(state.attempt, 1);
        // visible_after advanced into the future by at least the base backoff.
        assert!(
            state.visible_after >= before + chrono::Duration::seconds(30),
            "visible_after must advance by at least the base backoff"
        );
        // The fence HOLDS, not merely stored: a backend that persists
        // visible_after and then ignores it when claiming would satisfy the
        // assertion above and fail here.
        assert!(
            store.claim_outbox_rows(10).await?.is_empty(),
            "the backoff fence must hold the row out of the claimable set"
        );
        Ok(())
    }

    #[tokio::test]
    async fn shutdown_abandonment_leaves_row_claimed_for_recovery()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("shutdown-claimed").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;
        let claimed = store
            .claim_outbox_rows(1)
            .await?
            .into_iter()
            .next()
            .ok_or("row must be claimed before dispatch")?;

        let dispatcher = OutboxDispatcher::new(store.clone(), config())
            .with_dispatch(Arc::new(RecordingDispatch::new(false)));
        dispatcher.delivery_gate().drain();
        dispatcher.process_row(&claimed).await;

        let state = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("abandoned row must still exist")?;
        assert_eq!(
            state.status,
            OutboxStatus::Claimed,
            "shutdown must leave an abandoned delivery claimed for reconciler recovery"
        );
        assert_eq!(
            state.attempt, row.attempt,
            "shutdown abandonment must not consume an attempt"
        );
        Ok(())
    }

    /// Always fails with a [`ServerError::WorkerDispatchUnservable`], standing in
    /// for liminal proving the frame larger than the connection's outbound buffer.
    struct UnservableDispatch;

    #[async_trait]
    impl OutboxRowDispatch for UnservableDispatch {
        async fn dispatch(&self, _row: &OutboxRow) -> Result<(), ServerError> {
            Err(ServerError::worker_dispatch_unservable(
                "liminal-push",
                "the dispatch frame is 6214149 bytes and the buffer is 4194304 bytes",
            ))
        }
    }

    /// An unservable dispatch is dead-lettered on the FIRST observation: the row
    /// goes to `failed` with exactly one attempt consumed and is never re-armed,
    /// although the budget (`max_attempts: 3`) would have allowed two more. This
    /// is the outbox half of ending the 846-lease fault.
    #[tokio::test]
    async fn unservable_dispatch_dead_letters_on_first_observation()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("unservable").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let dispatcher = OutboxDispatcher::new(store.clone(), config())
            .with_dispatch(Arc::new(UnservableDispatch));
        let dispatcher = Arc::new(dispatcher);
        dispatcher.sweep_once().await;

        let state = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("dead-lettered row must still exist")?;
        assert_eq!(
            state.status,
            OutboxStatus::Failed,
            "an unservable dispatch is dead-lettered on its first observation"
        );
        // The retry path is the only writer that bumps a row's attempt
        // (`retry_outbox_row`); the dead-letter write changes status alone. An
        // attempt still at its seeded zero is therefore the proof that no retry
        // write happened between the first observation and the dead-letter.
        assert_eq!(
            state.attempt, 0,
            "the row was dead-lettered on its first observation, never re-armed for a retry"
        );
        assert!(
            store.claim_outbox_rows(10).await?.is_empty(),
            "a dead-lettered row is never claimable again"
        );
        Ok(())
    }

    /// Always fails with a [`ServerError::WorkerConnectionLost`], standing in for
    /// the chosen worker dying mid-dispatch.
    struct ConnectionLostDispatch;

    #[async_trait]
    impl OutboxRowDispatch for ConnectionLostDispatch {
        async fn dispatch(&self, _row: &OutboxRow) -> Result<(), ServerError> {
            Err(ServerError::worker_connection_lost(
                "liminal-push",
                "worker connection closed before reply",
            ))
        }
    }

    /// LSUB-3: a lost worker connection re-arms the row for IMMEDIATE re-claim
    /// (no backoff) so the next sweep fails it over to a live worker — while STILL
    /// consuming one attempt so churn stays bounded. Contrast with
    /// `failed_dispatch_retries_with_backoff_and_bumps_attempt`, where a generic
    /// failure pushes `visible_after` into the future.
    #[tokio::test]
    async fn connection_lost_rearms_immediately_and_consumes_attempt()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("conn-lost").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let before = Utc::now();
        let dispatcher = OutboxDispatcher::new(store.clone(), config())
            .with_dispatch(Arc::new(ConnectionLostDispatch));
        let dispatcher = Arc::new(dispatcher);
        dispatcher.sweep_once().await;
        let after = Utc::now();

        let state = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("re-armed row must still exist")?;
        // Returned to pending with the attempt consumed (0 -> 1): churn stays
        // bounded by max_attempts and eventually dead-letters.
        assert_eq!(state.status, OutboxStatus::Pending);
        assert_eq!(state.attempt, 1, "the failover still consumes one attempt");
        // visible_after is "now", NOT pushed out by the base backoff: it sits in
        // the [before, after] window of this sweep. The claim below is the direct
        // proof that the generic failure path's future fence was not applied.
        assert!(
            state.visible_after >= before && state.visible_after <= after,
            "visible_after must be re-armed to now (immediate re-claim), not backed off"
        );
        // The row is IMMEDIATELY claimable again — the next sweep re-dispatches it.
        assert_eq!(
            store.claim_outbox_rows(10).await?.len(),
            1,
            "the re-armed row is immediately claimable for failover"
        );
        Ok(())
    }

    /// LSUB-3: a lost worker connection STILL dead-letters once the attempt budget
    /// is exhausted — immediate re-claim never forms an unbounded re-dispatch loop.
    #[tokio::test]
    async fn connection_lost_dead_letters_after_max_attempts()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("conn-lost-dead").await?;
        let workflow_id = WorkflowId::new_v4();
        // Seed at the final attempt so the next connection-lost failure exhausts
        // the budget rather than re-arming forever.
        let mut row = pending_row(&workflow_id, 0);
        row.attempt = config().max_attempts - 1;
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let dispatcher = OutboxDispatcher::new(store.clone(), config())
            .with_dispatch(Arc::new(ConnectionLostDispatch));
        let dispatcher = Arc::new(dispatcher);
        dispatcher.sweep_once().await;

        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|s| s.status),
            Some(OutboxStatus::Failed),
            "connection-lost churn is bounded by max_attempts and dead-letters"
        );
        assert!(store.claim_outbox_rows(10).await?.is_empty());
        Ok(())
    }

    /// A dispatcher that reports the pool BUSY for the first `busy_rounds`
    /// dispatches, then delivers — the shape a fan onto a saturated worker has.
    use std::sync::atomic::{AtomicUsize, Ordering};

    struct BusyThenDeliver {
        busy_rounds: AtomicUsize,
        dispatched: AtomicUsize,
    }

    #[async_trait]
    impl OutboxRowDispatch for BusyThenDeliver {
        async fn dispatch(&self, row: &OutboxRow) -> Result<(), ServerError> {
            self.dispatched.fetch_add(1, Ordering::SeqCst);
            if self.busy_rounds.load(Ordering::SeqCst) == 0 {
                return Ok(());
            }
            self.busy_rounds.fetch_sub(1, Ordering::SeqCst);
            Err(ServerError::worker_busy(
                row.task_queue.clone(),
                "every worker is at its advertised concurrency",
            ))
        }
    }

    use super::capacity_park::CapacityParkedRows;
    use aion_store::{ClaimScope, StoreError};
    use chrono::DateTime;
    use std::sync::atomic::AtomicBool;

    /// Hands out one row, then reports whether it was in the dispatcher's
    /// capacity-parked set at the moment the re-arm write ran.
    struct ParkOrderStore {
        row: OutboxRow,
        claimed: AtomicBool,
        saw_write: AtomicBool,
        parked_at_write: AtomicBool,
        parked: std::sync::OnceLock<Arc<CapacityParkedRows>>,
    }

    #[async_trait]
    impl OutboxStore for ParkOrderStore {
        async fn append_outbox_batch(&self, _rows: &[OutboxRow]) -> Result<(), StoreError> {
            Ok(())
        }
        async fn claim_outbox_rows(&self, _limit: u32) -> Result<Vec<OutboxRow>, StoreError> {
            if self
                .claimed
                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                .is_ok()
            {
                Ok(vec![self.row.clone()])
            } else {
                Ok(Vec::new())
            }
        }
        async fn claim_outbox_rows_scoped(
            &self,
            _scope: &ClaimScope,
            _limit: u32,
        ) -> Result<Vec<OutboxRow>, StoreError> {
            Ok(Vec::new())
        }
        async fn rearm_stale_claimed_outbox_rows(
            &self,
            _older_than: DateTime<Utc>,
            _visible_after: DateTime<Utc>,
            _limit: u32,
            _excluded: &std::collections::HashSet<String>,
        ) -> Result<Vec<OutboxRow>, StoreError> {
            Ok(Vec::new())
        }
        async fn complete_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
            Ok(())
        }
        async fn retry_outbox_row(
            &self,
            dispatch_key: &str,
            _next_attempt: u32,
            _visible_after: DateTime<Utc>,
        ) -> Result<(), StoreError> {
            // THE OBSERVATION. Whatever the busy arm has done so far is what
            // a pulse landing right now would see.
            if !self.saw_write.swap(true, Ordering::SeqCst)
                && let Some(parked) = self.parked.get()
            {
                self.parked_at_write
                    .store(parked.contains(dispatch_key), Ordering::SeqCst);
            }
            Ok(())
        }
        async fn fail_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
            Ok(())
        }
        async fn count_inflight_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
            Ok(1)
        }
        async fn count_claimed_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
            Ok(0)
        }
        async fn pending_outbox_routes(&self) -> Result<Vec<ClaimScope>, StoreError> {
            Ok(Vec::new())
        }
    }

    /// Hands out one row and FAILS every `retry_outbox_row`, so the busy arm's
    /// durable park never lands.
    struct FailingRearmStore {
        row: OutboxRow,
        claimed: AtomicBool,
        attempted: AtomicBool,
    }

    #[async_trait]
    impl OutboxStore for FailingRearmStore {
        async fn append_outbox_batch(&self, _rows: &[OutboxRow]) -> Result<(), StoreError> {
            Ok(())
        }
        async fn claim_outbox_rows(&self, _limit: u32) -> Result<Vec<OutboxRow>, StoreError> {
            if self
                .claimed
                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                .is_ok()
            {
                Ok(vec![self.row.clone()])
            } else {
                Ok(Vec::new())
            }
        }
        async fn claim_outbox_rows_scoped(
            &self,
            _scope: &ClaimScope,
            _limit: u32,
        ) -> Result<Vec<OutboxRow>, StoreError> {
            Ok(Vec::new())
        }
        async fn rearm_stale_claimed_outbox_rows(
            &self,
            _older_than: DateTime<Utc>,
            _visible_after: DateTime<Utc>,
            _limit: u32,
            _excluded: &std::collections::HashSet<String>,
        ) -> Result<Vec<OutboxRow>, StoreError> {
            Ok(Vec::new())
        }
        async fn complete_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
            Ok(())
        }
        async fn retry_outbox_row(
            &self,
            _dispatch_key: &str,
            _next_attempt: u32,
            _visible_after: DateTime<Utc>,
        ) -> Result<(), StoreError> {
            self.attempted.store(true, Ordering::SeqCst);
            Err(StoreError::Backend("re-arm write refused".to_owned()))
        }
        async fn fail_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
            Ok(())
        }
        async fn count_inflight_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
            Ok(1)
        }
        async fn count_claimed_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
            Ok(0)
        }
        async fn pending_outbox_routes(&self) -> Result<Vec<ClaimScope>, StoreError> {
            Ok(Vec::new())
        }
    }

    /// Fires a capacity pulse from INSIDE the busy arm's park write, so the
    /// pulse provably lands while the entry is still pending.
    struct PendingPulseStore {
        row: OutboxRow,
        claimed: AtomicBool,
        pulsed: AtomicBool,
        pending_at_pulse: AtomicBool,
        /// Every write, in order: its `visible_after`, and whether it was
        /// issued while an earlier write was still in flight. The second half is
        /// the whole point — a re-offer nested inside the park write is one that
        /// acted on a row whose park was not yet durable, which is the race this
        /// protocol exists to prevent.
        writes: std::sync::Mutex<Vec<(DateTime<Utc>, bool)>>,
        /// How many writes are in flight, so nesting is observable.
        depth: AtomicUsize,
        dispatcher:
            std::sync::OnceLock<std::sync::Weak<OutboxDispatcher<Arc<dyn OutboxRowDispatch>>>>,
    }

    #[async_trait]
    impl OutboxStore for PendingPulseStore {
        async fn append_outbox_batch(&self, _rows: &[OutboxRow]) -> Result<(), StoreError> {
            Ok(())
        }
        async fn claim_outbox_rows(&self, _limit: u32) -> Result<Vec<OutboxRow>, StoreError> {
            if self
                .claimed
                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                .is_ok()
            {
                Ok(vec![self.row.clone()])
            } else {
                Ok(Vec::new())
            }
        }
        async fn claim_outbox_rows_scoped(
            &self,
            _scope: &ClaimScope,
            _limit: u32,
        ) -> Result<Vec<OutboxRow>, StoreError> {
            Ok(Vec::new())
        }
        async fn rearm_stale_claimed_outbox_rows(
            &self,
            _older_than: DateTime<Utc>,
            _visible_after: DateTime<Utc>,
            _limit: u32,
            _excluded: &std::collections::HashSet<String>,
        ) -> Result<Vec<OutboxRow>, StoreError> {
            Ok(Vec::new())
        }
        async fn complete_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
            Ok(())
        }
        async fn retry_outbox_row(
            &self,
            dispatch_key: &str,
            _next_attempt: u32,
            visible_after: DateTime<Utc>,
        ) -> Result<(), StoreError> {
            let nested = self.depth.fetch_add(1, Ordering::SeqCst) > 0;
            if let Ok(mut writes) = self.writes.lock() {
                writes.push((visible_after, nested));
            }
            // THE PULSE, fired once, from inside the park write — the exact
            // interleaving the pending→confirmed protocol exists for.
            if self
                .pulsed
                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                .is_ok()
                && let Some(dispatcher) = self.dispatcher.get().and_then(std::sync::Weak::upgrade)
            {
                self.pending_at_pulse.store(
                    dispatcher.capacity_parked.is_pending(dispatch_key),
                    Ordering::SeqCst,
                );
                dispatcher.reoffer_capacity_parked().await;
            }
            self.depth.fetch_sub(1, Ordering::SeqCst);
            Ok(())
        }
        async fn fail_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
            Ok(())
        }
        async fn count_inflight_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
            Ok(1)
        }
        async fn count_claimed_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
            Ok(0)
        }
        async fn pending_outbox_routes(&self) -> Result<Vec<ClaimScope>, StoreError> {
            Ok(Vec::new())
        }
    }

    /// aion#204 round 5: a pulse arriving while a park is still in flight is
    /// DEFERRED, not dropped — and not acted on either.
    ///
    /// The row's delivery task is spawned, so the run loop can be in its wake
    /// arm while the busy arm's `now + backoff` write is unserialised. A
    /// re-offer acting there would issue a second write against the same key,
    /// and if the park's write commits second the row ends up fenced AND
    /// forgotten — the original lost-wake outcome at lower odds.
    ///
    /// So the re-offer skips a pending row and records the pulse on it, and the
    /// confirmation re-checks that record and re-offers immediately. This test
    /// fires the pulse from inside the park write, which is that interleaving
    /// exactly, and asserts all three properties: the pulse really did land
    /// while pending, it wrote nothing at the time, and the row was re-offered
    /// once the park was confirmed.
    ///
    /// `backoff_base` is an hour, so the two writes are unmistakable: the park's
    /// fence is an hour out and the re-offer's is now.
    #[tokio::test]
    async fn a_pulse_during_a_pending_park_is_deferred_then_honoured()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        let store = Arc::new(PendingPulseStore {
            row,
            claimed: AtomicBool::new(false),
            pulsed: AtomicBool::new(false),
            pending_at_pulse: AtomicBool::new(false),
            writes: std::sync::Mutex::new(Vec::new()),
            depth: AtomicUsize::new(0),
            dispatcher: std::sync::OnceLock::new(),
        });
        let mut config = config();
        config.backoff_base = Duration::from_secs(3600);
        let dispatch: Arc<dyn OutboxRowDispatch> = Arc::new(BusyThenDeliver {
            busy_rounds: AtomicUsize::new(1),
            dispatched: AtomicUsize::new(0),
        });
        let dispatcher = Arc::new(
            OutboxDispatcher::new(store.clone() as Arc<dyn OutboxStore>, config)
                .with_dispatch(dispatch),
        );
        store
            .dispatcher
            .set(Arc::downgrade(&dispatcher))
            .map_err(|_| "the dispatcher handle must be settable once")?;

        let before = Utc::now();
        dispatcher.sweep_once().await;

        assert!(
            store.pending_at_pulse.load(Ordering::SeqCst),
            "the pulse must have landed while the park was still pending, or this test is not \
             exercising the race it claims to"
        );
        let writes = store
            .writes
            .lock()
            .map_err(|_| "writes lock poisoned")?
            .clone();
        assert_eq!(
            writes.len(),
            2,
            "exactly two writes: the park's own fence, then the deferred re-offer. One write \
             means the pulse was dropped; three means the re-offer raced the park"
        );
        assert!(
            writes[0].0 > before + chrono::Duration::seconds(1800),
            "the first write is the park's fence, an hour out: {:?}",
            writes[0].0
        );
        assert!(
            !writes[1].1,
            "the re-offer must NOT have been issued while the park write was still in flight. \
             Nested, it is a second writer on the same key: if the park's write serialises second \
             the row ends up fenced AND forgotten, which is the outcome this protocol exists to \
             prevent"
        );
        assert!(
            writes[1].0 < before + chrono::Duration::seconds(60),
            "the second write is the deferred re-offer, pulling the row forward to now — without \
             it the pulse that arrived during the park is lost and the row sleeps out the hour: \
             {:?}",
            writes[1].0
        );
        Ok(())
    }

    /// aion#204 round 5: a park whose durable write FAILS leaves no phantom
    /// behind.
    ///
    /// The row is remembered before the write, so the failure path has to take
    /// it back. Left remembered, it would be re-offered on every freed slot
    /// forever — each re-offer pulling forward a `visible_after` that was never
    /// written, so the sweep could never claim it and the entry could never
    /// clear. A phantom that survives its own park is worse than a row that was
    /// never parked: it burns a wake every time capacity frees.
    #[tokio::test]
    async fn a_park_whose_durable_write_fails_is_not_left_remembered()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        let key = row.dispatch_key.clone();
        let store = Arc::new(FailingRearmStore {
            row,
            claimed: AtomicBool::new(false),
            attempted: AtomicBool::new(false),
        });
        let dispatch: Arc<dyn OutboxRowDispatch> = Arc::new(BusyThenDeliver {
            busy_rounds: AtomicUsize::new(1),
            dispatched: AtomicUsize::new(0),
        });
        let dispatcher = Arc::new(
            OutboxDispatcher::new(store.clone() as Arc<dyn OutboxStore>, config())
                .with_dispatch(dispatch),
        );

        dispatcher.sweep_once().await;

        assert!(
            store.attempted.load(Ordering::SeqCst),
            "the busy arm must have attempted its durable park, or this test observed nothing"
        );
        assert!(
            !dispatcher.capacity_parked.contains(&key),
            "a row whose durable park failed must not stay remembered: it would be re-offered on \
             every freed slot, pulling forward a fence that was never written"
        );
        Ok(())
    }

    /// aion#204 round 4: a busy row is REMEMBERED BEFORE its durable re-arm is
    /// written, so a slot freeing during that write cannot spend the wake on an
    /// empty set.
    ///
    /// The re-arm is an awaited, store-wide write. Remembering the row after it
    /// leaves a window in which the pulse arrives, `reoffer_capacity_parked`
    /// finds nothing, the sweep finds nothing claimable (the row's fence is
    /// still far in the future), and the permit is spent — after which the row
    /// is inserted with nothing left to wake it and waits out the operator's
    /// FAILURE backoff. Nothing re-checks: the re-offer has one call site, and
    /// the interval poll claims on `visible_after <= now`, which a
    /// capacity-parked row fails by construction.
    ///
    /// The window is invisible to the sibling test above, where the insert
    /// happens before any slot frees. This one pins the ORDERING directly: the
    /// store records whether the row was already remembered at the instant the
    /// durable write ran. That is the whole property, observed at the only
    /// moment it is decidable.
    #[tokio::test]
    async fn a_busy_row_is_remembered_before_its_durable_re_arm_is_written()
    -> Result<(), Box<dyn std::error::Error>> {
        let workflow_id = WorkflowId::new_v4();
        let store = Arc::new(ParkOrderStore {
            row: pending_row(&workflow_id, 0),
            claimed: AtomicBool::new(false),
            saw_write: AtomicBool::new(false),
            parked_at_write: AtomicBool::new(false),
            parked: std::sync::OnceLock::new(),
        });
        let dispatch: Arc<dyn OutboxRowDispatch> = Arc::new(BusyThenDeliver {
            busy_rounds: AtomicUsize::new(1),
            dispatched: AtomicUsize::new(0),
        });
        let dispatcher = Arc::new(
            OutboxDispatcher::new(store.clone() as Arc<dyn OutboxStore>, config())
                .with_dispatch(dispatch),
        );
        // The set the busy arm writes into, shared with the observing store.
        store
            .parked
            .set(Arc::clone(&dispatcher.capacity_parked))
            .map_err(|_| "the parked handle must be settable once")?;

        dispatcher.sweep_once().await;

        assert!(
            store.saw_write.load(Ordering::SeqCst),
            "the busy arm must have written its durable re-arm, or this test observed nothing"
        );
        assert!(
            store.parked_at_write.load(Ordering::SeqCst),
            "the row must already be remembered when its durable re-arm is written. Remembered \
             after, a slot freeing during that write spends the wake on an empty set and the row \
             sleeps out the failure backoff — the outcome the capacity wake exists to prevent"
        );
        Ok(())
    }

    /// aion#204 D2: a busy row is re-offered when a SLOT FREES, not when the
    /// failure backoff expires.
    ///
    /// A busy answer is attempt-neutral, so the row waits — and what it waits
    /// for is capacity, which is an event this server produces. It used to wait
    /// on `backoff_base`, the FAILURE curve: an operator raising that
    /// legitimately (to slow genuine retries) parked healthy work behind it for
    /// the same duration. Observed in the failover fixture at 120 s against a
    /// 40 s deadline.
    ///
    /// `backoff_base` here is an hour, so nothing can pass by waiting it out —
    /// the only way this test completes is the capacity wake. The durable fence
    /// still carries that hour, deliberately: it stays the crash-safe fallback,
    /// and the point of the fix is that it is no longer the only path.
    #[tokio::test]
    async fn a_busy_row_is_re_offered_when_a_slot_frees_not_when_backoff_expires()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("capacity-wake").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let mut config = config();
        config.backoff_base = Duration::from_secs(3600);
        let dispatch: Arc<dyn OutboxRowDispatch> = Arc::new(BusyThenDeliver {
            busy_rounds: AtomicUsize::new(1),
            dispatched: AtomicUsize::new(0),
        });
        let wake = Arc::new(tokio::sync::Notify::new());
        let dispatcher = Arc::new(
            OutboxDispatcher::new(store.clone(), config)
                .with_dispatch(dispatch)
                .with_wake(Arc::clone(&wake)),
        );

        // One sweep: the pool is busy, so the row is parked behind an hour.
        dispatcher.sweep_once().await;
        let parked = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("the row must still exist")?;
        assert_eq!(parked.status, OutboxStatus::Pending);
        assert_eq!(parked.attempt, row.attempt, "busy costs no attempt");
        assert!(
            store.claim_outbox_rows(10).await?.is_empty(),
            "the durable fence must really be an hour out, or this test proves nothing"
        );

        // A SLOT FREES. This is what the registry pulses on every untrack and
        // every released reservation; the run loop re-offers on it.
        dispatcher.reoffer_capacity_parked().await;
        dispatcher.sweep_once().await;

        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|state| state.status),
            Some(OutboxStatus::Done),
            "the freed slot must deliver the row; without the re-offer it waits out the failure \
             backoff, which is the wrong clock for a pool that is merely working"
        );
        Ok(())
    }

    async fn wait_for_confirmed_capacity_park(
        capacity_parked: &CapacityParkedRows,
        dispatch_key: &str,
    ) -> Result<(), tokio::time::error::Elapsed> {
        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if capacity_parked.contains(dispatch_key)
                    && !capacity_parked.is_pending(dispatch_key)
                {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
    }

    async fn wait_for_outbox_done(
        store: &HaematiteStore,
        dispatch_key: &str,
    ) -> Result<(), Box<dyn std::error::Error>> {
        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if store
                    .outbox_row_state(dispatch_key)
                    .await?
                    .is_some_and(|state| state.status == OutboxStatus::Done)
                {
                    return Ok::<(), aion_store::StoreError>(());
                }
                tokio::task::yield_now().await;
            }
        })
        .await??;
        Ok(())
    }

    /// A tracked activity retiring through `HeartbeatTracker` frees the worker's
    /// only advertised slot, pulses the registry's shared capacity wake, and
    /// drives the live `OutboxDispatcher` run loop to re-offer and deliver a row
    /// parked behind an otherwise hour-long failure fence.
    #[tokio::test]
    async fn heartbeat_retirement_wakes_the_run_loop_and_delivers_a_capacity_parked_row()
    -> Result<(), Box<dyn std::error::Error>> {
        use std::time::Instant;

        use crate::worker::dispatch::ActivityDispatcher;
        use crate::worker::heartbeat::{HeartbeatTracker, InFlightActivity};
        use crate::worker::registry::{ConnectedWorkerRegistry, WorkerMessage};

        let store = open_store("heartbeat-capacity-wake").await?;
        let row = pending_row(&WorkflowId::new_v4(), 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        // The capacity ledger and dispatcher run loop share the same wake,
        // exactly as production wiring does.
        let wake = Arc::new(tokio::sync::Notify::new());
        let registry = ConnectedWorkerRegistry::default().with_capacity_wake(Arc::clone(&wake));

        let (tx, mut rx) = tokio::sync::mpsc::channel(2);
        let activity_types = [String::from("charge")];
        let registration = registry.register("default", activity_types.iter(), tx, 1)?;
        let worker_id = registration
            .worker_id()
            .ok_or("the local worker registration must have an id")?;

        let tracker = HeartbeatTracker::new(Duration::from_secs(30));
        let blocking_workflow_id = WorkflowId::new_v4();
        let blocking_activity_id = ActivityId::from_sequence_position(99);
        tracker.track_task(
            worker_id,
            InFlightActivity {
                workflow_id: blocking_workflow_id.clone(),
                activity_id: blocking_activity_id.clone(),
                attempt: 1,
                completion_token: CompletionToken::for_test(),
            },
            Instant::now(),
            &registry,
            None,
        )?;
        assert_eq!(tracker.in_flight_for_worker(worker_id)?, 1);
        assert_eq!(registry.in_flight_for_worker(worker_id)?, 1);

        let mut dispatcher_config = config();
        dispatcher_config.poll_interval = Duration::from_secs(3_600);
        dispatcher_config.backoff_base = Duration::from_secs(3_600);

        let dispatcher_builder = OutboxDispatcher::new(store.clone(), dispatcher_config);
        let delivery_gate = dispatcher_builder.delivery_gate();
        let activity_dispatcher = ActivityDispatcher::new(registry.clone())
            .with_delivery_gate(delivery_gate)
            .with_heartbeat_tracker(tracker.clone());
        let dispatch: Arc<dyn OutboxRowDispatch> =
            Arc::new(WorkerOutboxDispatch::new(activity_dispatcher));
        let dispatcher = dispatcher_builder
            .with_dispatch(dispatch)
            .with_wake(Arc::clone(&wake));

        let capacity_parked = Arc::clone(&dispatcher.capacity_parked);
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let dispatcher_task = tokio::spawn(dispatcher.run(shutdown_rx));

        wait_for_confirmed_capacity_park(&capacity_parked, &row.dispatch_key)
            .await
            .map_err(|_| "the outbox row was not durably capacity-parked")?;

        let parked = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("the parked outbox row must still exist")?;
        assert_eq!(parked.status, OutboxStatus::Pending);
        assert_eq!(
            parked.attempt, row.attempt,
            "capacity parking must not consume an outbox attempt"
        );
        assert_eq!(
            registry.in_flight_for_worker(worker_id)?,
            1,
            "the sole worker slot must still be occupied before retirement"
        );
        assert!(
            rx.try_recv().is_err(),
            "a full worker must receive no outbox task before its slot is freed"
        );

        assert!(
            tracker.complete_task(
                worker_id,
                &blocking_workflow_id,
                &blocking_activity_id,
                &registry,
            )?,
            "the blocker must still be tracked when it is retired"
        );
        assert_eq!(tracker.in_flight_for_worker(worker_id)?, 0);
        assert_eq!(registry.in_flight_for_worker(worker_id)?, 0);

        wait_for_outbox_done(&store, &row.dispatch_key).await?;

        let delivered = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await?
            .ok_or("the registered worker channel closed before delivery")?;
        assert!(
            matches!(delivered, WorkerMessage::ActivityTask(_)),
            "the capacity wake must deliver the parked row through the real worker path"
        );

        shutdown_tx
            .send(true)
            .map_err(|_| "the outbox dispatcher exited before test shutdown")?;
        tokio::time::timeout(Duration::from_secs(5), dispatcher_task).await??;

        Ok(())
    }

    /// aion#204 HOLD-2 at the ROW: a busy pool costs the row nothing and never
    /// dead-letters it, and the row is delivered once a slot frees.
    ///
    /// The dispatch-seam half of this is asserted in `worker::dispatch`; this is
    /// the half that decides whether work SURVIVES. `max_attempts` here is 3 and
    /// the pool is busy for 5 rounds — more busy returns than the whole retry
    /// budget — so a row that spent an attempt per busy answer would be
    /// dead-lettered before the pool ever freed. It must instead come back at
    /// the SAME attempt every time, stay `Pending`, and then deliver.
    ///
    /// The attempt assertion is the load-bearing one: "still pending" alone
    /// would also hold for a row that was burning its budget on the way to
    /// `Failed`.
    #[tokio::test]
    async fn a_busy_pool_costs_an_outbox_row_no_attempts_and_never_dead_letters()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("busy").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let busy_rounds: u32 = 5;
        assert!(
            busy_rounds > config().max_attempts,
            "the pool must stay busy for longer than the whole retry budget, or this test \
             cannot tell an attempt-neutral re-arm from a lucky one"
        );
        let dispatch: Arc<dyn OutboxRowDispatch> = Arc::new(BusyThenDeliver {
            busy_rounds: AtomicUsize::new(busy_rounds as usize),
            dispatched: AtomicUsize::new(0),
        });
        let dispatcher =
            Arc::new(OutboxDispatcher::new(store.clone(), config()).with_dispatch(dispatch));

        for round in 0..busy_rounds {
            dispatcher.sweep_once().await;
            let state = store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .ok_or("the row must still exist")?;
            assert_eq!(
                state.status,
                OutboxStatus::Pending,
                "round {round}: a busy pool must leave the row pending, never failed"
            );
            assert_eq!(
                state.attempt,
                row.attempt,
                "round {round}: a busy pool must not spend an attempt — after {} busy rounds \
                 against a budget of {}, a row that spent one would already be dead-lettered",
                round + 1,
                config().max_attempts
            );
            // The re-arm is fenced by `backoff_base`; step past it so the next
            // sweep can claim the row again.
            tokio::time::sleep(config().backoff_base + Duration::from_millis(20)).await;
        }

        // THE RELEASE CONTROL. A slot frees and the same row is delivered — so
        // the rounds above were capacity, not a row nothing could ever serve.
        dispatcher.sweep_once().await;
        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|state| state.status),
            Some(OutboxStatus::Done),
            "once the pool frees a slot the row must be delivered"
        );
        Ok(())
    }

    #[tokio::test]
    async fn dispatch_fails_row_after_max_attempts() -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("fail").await?;
        let workflow_id = WorkflowId::new_v4();
        // Seed the row already at attempt == max_attempts - 1, so the next
        // failed dispatch is the final attempt and dead-letters it.
        let mut row = pending_row(&workflow_id, 0);
        row.attempt = config().max_attempts - 1;
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let dispatch = Arc::new(RecordingDispatch::new(false));
        let dispatcher = OutboxDispatcher::new(store.clone(), config()).with_dispatch(dispatch);
        let dispatcher = Arc::new(dispatcher);
        dispatcher.sweep_once().await;

        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|s| s.status),
            Some(OutboxStatus::Failed)
        );
        // A dead-lettered row is never claimable again.
        assert!(store.claim_outbox_rows(10).await?.is_empty());
        Ok(())
    }

    /// LSUB-4-6 (`mark_done` write failure): dispatch SUCCEEDS but `complete_outbox_row`
    /// fails — the row must stay `Claimed` (never silently dropped, never retried or
    /// dead-lettered), so a later rearm/reconcile can re-dispatch it (deduped to one
    /// terminal in history). Driven through a mock so the post-condition is asserted
    /// deterministically; `retry`/`fail` record a flag (never reached) instead of
    /// panicking, keeping the test free of the restriction lints.
    #[tokio::test]
    async fn mark_done_failure_leaves_row_claimed_for_later_rearm()
    -> Result<(), Box<dyn std::error::Error>> {
        use aion_store::{ClaimScope, StoreError};
        use chrono::{DateTime, Utc};
        use std::sync::atomic::{AtomicBool, Ordering};

        /// Hands out exactly one claimable row, then fails `complete_outbox_row`.
        /// Records whether `complete` was attempted and whether any other terminal
        /// transition was reached (it must not be) so the test proves the row is
        /// left `Claimed`.
        struct CompleteFailsStore {
            row: OutboxRow,
            claimed: AtomicBool,
            completed: AtomicBool,
            other_terminal: AtomicBool,
        }

        #[async_trait]
        impl OutboxStore for CompleteFailsStore {
            async fn append_outbox_batch(&self, _rows: &[OutboxRow]) -> Result<(), StoreError> {
                Ok(())
            }
            async fn claim_outbox_rows(&self, _limit: u32) -> Result<Vec<OutboxRow>, StoreError> {
                // Hand out the row exactly once (compare-and-swap false -> true).
                if self
                    .claimed
                    .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
                    .is_ok()
                {
                    Ok(vec![self.row.clone()])
                } else {
                    Ok(Vec::new())
                }
            }
            async fn claim_outbox_rows_scoped(
                &self,
                _scope: &ClaimScope,
                _limit: u32,
            ) -> Result<Vec<OutboxRow>, StoreError> {
                Ok(Vec::new())
            }
            async fn rearm_stale_claimed_outbox_rows(
                &self,
                _older_than: DateTime<Utc>,
                _visible_after: DateTime<Utc>,
                _limit: u32,
                _excluded: &std::collections::HashSet<String>,
            ) -> Result<Vec<OutboxRow>, StoreError> {
                Ok(Vec::new())
            }
            async fn complete_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
                // The write fails AFTER dispatch already happened; the row stays Claimed.
                self.completed.store(true, Ordering::SeqCst);
                Err(StoreError::Backend("mark-done write failed".to_owned()))
            }
            async fn retry_outbox_row(
                &self,
                _dispatch_key: &str,
                _next_attempt: u32,
                _visible_after: DateTime<Utc>,
            ) -> Result<(), StoreError> {
                self.other_terminal.store(true, Ordering::SeqCst);
                Ok(())
            }
            async fn fail_outbox_row(&self, _dispatch_key: &str) -> Result<(), StoreError> {
                self.other_terminal.store(true, Ordering::SeqCst);
                Ok(())
            }
            async fn count_inflight_outbox_rows(
                &self,
                _namespace: &str,
            ) -> Result<u64, StoreError> {
                // The single staged row is in-flight (Pending or stuck-Claimed) until completed.
                Ok(u64::from(!self.completed.load(Ordering::SeqCst)))
            }
            async fn count_claimed_outbox_rows(&self, _namespace: &str) -> Result<u64, StoreError> {
                // No backpressure path exercises this store, so a zero claimed count is sufficient.
                Ok(0)
            }
            async fn pending_outbox_routes(&self) -> Result<Vec<ClaimScope>, StoreError> {
                Ok(Vec::new())
            }
        }

        let workflow_id = WorkflowId::new_v4();
        let store = Arc::new(CompleteFailsStore {
            row: pending_row(&workflow_id, 0),
            claimed: AtomicBool::new(false),
            completed: AtomicBool::new(false),
            other_terminal: AtomicBool::new(false),
        });
        let dispatch = Arc::new(RecordingDispatch::new(true));
        let dispatcher =
            OutboxDispatcher::new(store.clone(), config()).with_dispatch(dispatch.clone());
        let dispatcher = Arc::new(dispatcher);
        // Sweep: claims the row, dispatches it (succeeds), then complete fails.
        dispatcher.sweep_once().await;

        assert_eq!(dispatch.count()?, 1, "the row was dispatched exactly once");
        assert!(
            store.completed.load(Ordering::SeqCst),
            "mark_done was attempted (and failed) after the successful dispatch"
        );
        // The row is left Claimed: NOT retried, NOT dead-lettered. A later rearm /
        // reconcile re-dispatches it, deduped to one terminal in history.
        assert!(
            !store.other_terminal.load(Ordering::SeqCst),
            "a mark_done failure must not retry or dead-letter the row (it stays Claimed)"
        );
        Ok(())
    }

    /// Drive a dispatcher's `run` loop until a row reaches `Done` or the deadline
    /// elapses; returns whether it reached `Done` in time.
    async fn wait_for_done(
        store: &HaematiteStore,
        dispatch_key: &str,
        deadline: std::time::Instant,
    ) -> Result<bool, ServerError> {
        loop {
            let done = store
                .outbox_row_state(dispatch_key)
                .await?
                .map(|s| s.status)
                == Some(OutboxStatus::Done);
            if done {
                return Ok(true);
            }
            if std::time::Instant::now() > deadline {
                return Ok(false);
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    }

    /// LSUB-2 (fast path): a wake drives a staged row to `Done` FAST — well under
    /// the poll interval — proving the wake, not the poll, ran the sweep. The poll
    /// is set to 10s so it cannot explain a sub-second dispatch.
    #[tokio::test]
    async fn wake_dispatches_staged_row_well_under_poll_interval()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("wake-fast").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        // A 10s poll interval: any dispatch inside the 1s assert deadline must be
        // the wake's doing, not the poll (a >=100x margin keeps it non-flaky).
        let mut slow_poll = config();
        slow_poll.poll_interval = Duration::from_secs(10);
        let wake = Arc::new(tokio::sync::Notify::new());
        let dispatch = Arc::new(RecordingDispatch::new(true));
        let dispatcher = OutboxDispatcher::new(store.clone(), slow_poll)
            .with_dispatch(dispatch.clone())
            .with_wake(Arc::clone(&wake));
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let handle = tokio::spawn(dispatcher.run(shutdown_rx));

        // The row is already staged, so the stored permit from `notify_one` is
        // consumed by the first `notified()` and the sweep finds the pending row.
        wake.notify_one();

        let reached = wait_for_done(
            store.as_ref(),
            &row.dispatch_key,
            std::time::Instant::now() + Duration::from_secs(1),
        )
        .await?;
        assert!(
            reached,
            "the wake must dispatch the staged row within 1s, far under the 10s poll"
        );
        assert_eq!(dispatch.count()?, 1);

        shutdown_tx.send(true)?;
        tokio::time::timeout(Duration::from_secs(5), handle)
            .await
            .map_err(|_| "outbox dispatcher did not stop after shutdown")??;
        Ok(())
    }

    /// LSUB-2 (correctness backstop): with the wake NEVER pulsed, a staged row
    /// STILL reaches `Done` via the interval poll. Together with the fast-path test
    /// this proves the wake is advisory-only — a dropped/absent wake degrades
    /// cleanly to the existing poll and never loses a dispatch.
    #[tokio::test]
    async fn poll_dispatches_staged_row_when_wake_never_fires()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("wake-absent").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        // Short (10ms) poll, and a wake handle that is installed but never pulsed.
        let wake = Arc::new(tokio::sync::Notify::new());
        let dispatch = Arc::new(RecordingDispatch::new(true));
        let dispatcher = OutboxDispatcher::new(store.clone(), config())
            .with_dispatch(dispatch.clone())
            .with_wake(Arc::clone(&wake));
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let handle = tokio::spawn(dispatcher.run(shutdown_rx));

        // Deliberately never call `wake.notify_one()`: only the poll can drive this.
        let reached = wait_for_done(
            store.as_ref(),
            &row.dispatch_key,
            std::time::Instant::now() + Duration::from_secs(5),
        )
        .await?;
        assert!(
            reached,
            "the poll must dispatch the staged row even though the wake never fired"
        );
        assert_eq!(dispatch.count()?, 1);

        shutdown_tx.send(true)?;
        tokio::time::timeout(Duration::from_secs(5), handle)
            .await
            .map_err(|_| "outbox dispatcher did not stop after shutdown")??;
        Ok(())
    }

    #[tokio::test]
    async fn run_loop_drains_pending_then_stops_on_shutdown()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("run-loop").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let dispatch = Arc::new(RecordingDispatch::new(true));
        let dispatcher =
            OutboxDispatcher::new(store.clone(), config()).with_dispatch(dispatch.clone());
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let handle = tokio::spawn(dispatcher.run(shutdown_rx));

        // Wait for the row to be dispatched and marked done by the loop.
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        loop {
            let done = store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|s| s.status)
                == Some(OutboxStatus::Done);
            if done {
                break;
            }
            if std::time::Instant::now() > deadline {
                return Err("outbox dispatcher loop did not mark the row done".into());
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(dispatch.count()?, 1);

        // Signal shutdown; the task must observe it and stop cleanly.
        shutdown_tx.send(true)?;
        tokio::time::timeout(Duration::from_secs(5), handle)
            .await
            .map_err(|_| "outbox dispatcher did not stop after shutdown")??;
        Ok(())
    }

    #[test]
    fn backoff_grows_geometrically_and_clamps_to_max() {
        let config = config();
        // attempt 0: base (100ms); attempt 1: 200ms; attempt 2: 400ms.
        assert_eq!(config.backoff_for_attempt(0), Duration::from_millis(100));
        assert_eq!(config.backoff_for_attempt(1), Duration::from_millis(200));
        assert_eq!(config.backoff_for_attempt(2), Duration::from_millis(400));
        // A very large attempt clamps at backoff_max, never overflows.
        assert_eq!(config.backoff_for_attempt(1000), config.backoff_max);
    }

    /// NSTQ-2: the production [`WorkerOutboxDispatch`] routes a claimed row by the
    /// `namespace` + `task_queue` carried ON THE ROW, not by any server default.
    /// A row stamped `namespace = "remote"` reaches a worker registered in the
    /// `remote` namespace. A row stamped `namespace = "default"` is NOT served by
    /// that `remote` worker, proving the routing identity comes off the row and
    /// the server default is no longer injected.
    ///
    /// 🔴 THE OBSERVABLE FOR THE NEGATIVE HALF CHANGED WITH #190, AND THE
    /// INVARIANT DID NOT. This pin used to read isolation off a 200ms TIMEOUT:
    /// the `default` row's dispatch parked forever waiting for a worker that
    /// never registers, so a timeout stood in for "not served". Since #190 an
    /// outbox row is REFUSED rather than parked, so the dispatch now returns
    /// promptly — and a test asserting the timeout would have gone red for the
    /// fix while namespace isolation was never in question.
    ///
    /// So assert the invariant rather than the mechanism that used to express
    /// it, in the two places it is actually visible: the remote worker's
    /// channel receives NOTHING for the `default` row, and the refusal NAMES
    /// the queue it could not serve. That is a strictly better witness than the
    /// timeout it replaces — a timeout cannot tell namespace isolation from any
    /// other stall, whereas this refusal could not be produced by a delivery to
    /// the wrong namespace.
    #[tokio::test]
    async fn worker_dispatch_routes_by_row_namespace_not_server_default()
    -> Result<(), Box<dyn std::error::Error>> {
        use crate::worker::dispatch::ActivityDispatcher;
        use crate::worker::registry::{ConnectedWorkerRegistry, WorkerMessage};
        use aion_store::OutboxRow;

        let registry = ConnectedWorkerRegistry::default();
        let (tx, mut rx) = tokio::sync::mpsc::channel(1);
        let activity_types = [String::from("charge")];
        // Only a `remote`-namespace worker is connected.
        let _registration = registry.register(
            "remote",
            activity_types.iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;

        let gate = super::DeliveryGate::default();
        let dispatch = super::WorkerOutboxDispatch::new(
            ActivityDispatcher::new(registry.clone()).with_delivery_gate(gate.clone()),
        );

        // A row whose routing identity is `remote` reaches the remote worker.
        let workflow_id = WorkflowId::new_v4();
        let remote_row = OutboxRow::pending(
            workflow_id.clone(),
            0,
            String::from("charge"),
            Payload::new(ContentType::Json, b"{}".to_vec()),
            Utc::now(),
        )
        .with_run_id(Some(RunId::new_v4()))
        .with_namespace("remote")
        .with_task_queue("default");

        dispatch_claimed(&dispatch, &gate, &remote_row).await?;
        let message = rx.recv().await.ok_or("expected pushed activity task")?;
        assert!(
            matches!(message, WorkerMessage::ActivityTask(_)),
            "the remote-namespace worker must receive the activity task for a remote row"
        );

        // A row whose namespace is `default` is NOT served by the remote worker.
        let default_row = remote_row.clone().with_namespace("default");
        let _default_claim = gate
            .begin(&default_row.dispatch_key)
            .ok_or("the gate must admit the default-namespace row")?;
        let refused = OutboxRowDispatch::dispatch(&dispatch, &default_row).await;
        let error = refused.err().ok_or(
            "a default-namespace row must not be served by a remote-namespace worker, so its \
             dispatch must not succeed",
        )?;
        let reported = error.to_string();
        assert!(
            reported.contains("task queue default"),
            "the refusal must name the queue nothing could serve, so the outbox's retry log and \
             the eventual dead letter say WHY; got: {reported}"
        );

        // The half a returned error cannot show: nothing was pushed to the
        // remote-namespace worker. Without this, a dispatch that delivered to
        // the wrong namespace AND then failed for some other reason would read
        // identically to correct isolation.
        assert!(
            rx.try_recv().is_err(),
            "the remote-namespace worker must receive NOTHING for a default-namespace row"
        );
        Ok(())
    }

    /// NODE-2: `to_scheduled` sources node affinity off the row. A row stamped
    /// `Some(node)` produces a `ScheduledActivity` pinned to that node; a row with
    /// no affinity (`None`) produces an unpinned dispatch.
    #[test]
    fn to_scheduled_sources_node_affinity_from_row() {
        let workflow_id = WorkflowId::new_v4();
        let pinned = OutboxRow::pending(
            workflow_id.clone(),
            0,
            String::from("charge"),
            Payload::new(ContentType::Json, b"{}".to_vec()),
            Utc::now(),
        )
        .with_node(Some(String::from("box-7")));
        let scheduled = super::WorkerOutboxDispatch::to_scheduled(&pinned);
        assert_eq!(scheduled.node.as_deref(), Some("box-7"));

        let unpinned = pending_row(&workflow_id, 1);
        let scheduled = super::WorkerOutboxDispatch::to_scheduled(&unpinned);
        assert_eq!(scheduled.node, None);
    }

    /// A row's stored ZERO-based attempt is stamped ONE-based for delivery.
    ///
    /// Zero is malformed on the wire, and the attempt is part of what the
    /// completion fences key on: a redelivery of the same attempt must add a
    /// SIBLING authorization beside the one a worker may still be holding,
    /// rather than displace it. An off-by-one here silently changes which
    /// authorization a reply matches.
    ///
    /// 🔴 This assertion was carried here from `request_for_row` in
    /// `liminal_transport`, which the #52 R4 rewiring left with no caller and
    /// which was deleted with the rest of `RegistryLiminalDispatch`, so that
    /// the live property kept a witness. Its second half once pinned the
    /// DEFECT WA-010 R5 removed — "the stored zero-based attempt goes
    /// one-based on delivery", i.e. a retried row delivered as attempt 3 —
    /// and is inverted here: the delivery count never reaches the wire. The
    /// surviving property is the one that mattered: zero is malformed on the
    /// wire, and a never-retried row delivers as attempt 1.
    #[test]
    fn to_scheduled_never_stamps_zero_and_never_stamps_the_delivery_count() {
        let workflow_id = WorkflowId::new_v4();
        let fresh = pending_row(&workflow_id, 0);
        assert_eq!(
            fresh.attempt, 0,
            "precondition: a pending row starts at the stored zero"
        );
        assert_eq!(
            super::WorkerOutboxDispatch::to_scheduled(&fresh).attempt,
            1,
            "a never-retried row must deliver as attempt 1, never as 0"
        );

        let mut retried = pending_row(&workflow_id, 1);
        retried.attempt = 2;
        assert_eq!(
            super::WorkerOutboxDispatch::to_scheduled(&retried).attempt,
            1,
            "a retried row re-delivers the SAME recorded attempt; the delivery count is a budget"
        );
    }

    // --- P2-P3: Prefer two-tier spill + the determinism invariant -----------

    /// Register a worker advertising `node` for `activity_type` in `namespace`,
    /// returning the registration token (held to keep it connected) and its
    /// receiver so the test can observe a delivered task.
    fn register_node_worker(
        registry: &crate::worker::registry::ConnectedWorkerRegistry,
        namespace: &str,
        node: &str,
        activity_type: &str,
    ) -> Result<
        (
            crate::worker::registry::WorkerRegistration,
            tokio::sync::mpsc::Receiver<crate::worker::registry::WorkerMessage>,
        ),
        Box<dyn std::error::Error>,
    > {
        let (tx, rx) = tokio::sync::mpsc::channel(1);
        let types = [activity_type.to_owned()];
        let registration = registry.register_namespaces(
            [namespace.to_owned()],
            String::from("default"),
            Some(node.to_owned()),
            types.iter(),
            tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        Ok((registration, rx))
    }

    /// Build an UNPINNED outbox row (`node == None`) in `namespace` for `charge`.
    fn unpinned_row(namespace: &str) -> OutboxRow {
        OutboxRow::pending(
            WorkflowId::new_v4(),
            0,
            String::from("charge"),
            Payload::new(ContentType::Json, b"{}".to_vec()),
            Utc::now(),
        )
        .with_run_id(Some(RunId::new_v4()))
        .with_namespace(namespace)
        .with_task_queue("default")
    }

    /// Build a `WorkerOutboxDispatch` over `registry` whose placement cache reads
    /// `namespace_store` (zero TTL so each dispatch sees the latest placement).
    fn placement_dispatch(
        registry: &crate::worker::registry::ConnectedWorkerRegistry,
        namespace_store: Arc<dyn aion_store::NamespaceStore>,
    ) -> (super::WorkerOutboxDispatch, super::DeliveryGate) {
        use crate::worker::dispatch::ActivityDispatcher;
        let cache = crate::worker::PlacementCache::new(namespace_store, Duration::ZERO);
        // The SAME gate the test claims its rows in. A dispatcher holding a
        // private default would answer "this row's claim does not stand" for
        // every key, because a key never begun is indistinguishable from one
        // released — which is production's wiring requirement, not a test detail.
        let gate = super::DeliveryGate::default();
        let dispatch = super::WorkerOutboxDispatch::new(
            ActivityDispatcher::new(registry.clone()).with_delivery_gate(gate.clone()),
        )
        .with_placement_cache(cache);
        (dispatch, gate)
    }

    /// Dispatch `row` the way the outbox loop does: claim it in the gate FIRST
    /// and hold that claim across the dispatch.
    ///
    /// A test that dispatched an outbox row WITHOUT claiming it would be
    /// exercising a state production never reaches — the loop claims every row
    /// before it dispatches (see `OutboxDispatcher::run`) — and the delivery
    /// intent asks whether that claim still stands.
    async fn dispatch_claimed(
        dispatch: &super::WorkerOutboxDispatch,
        gate: &super::DeliveryGate,
        row: &aion_store::OutboxRow,
    ) -> Result<(), Box<dyn std::error::Error>> {
        let _claim = gate
            .begin(&row.dispatch_key)
            .ok_or("the gate must admit a row no other pass holds")?;
        OutboxRowDispatch::dispatch(dispatch, row).await?;
        Ok(())
    }

    /// P2-P3 (prefer): an unpinned row in a `Prefer{n1}` namespace selects the
    /// n1 worker when one is live, even with an n2 worker also connected.
    #[tokio::test]
    async fn prefer_selects_preferred_node_worker_when_present()
    -> Result<(), Box<dyn std::error::Error>> {
        use aion_store::{InMemoryStore, NamespaceOrigin, NamespacePlacement, NamespaceStore};

        let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        ns_store
            .register_namespace("t", NamespaceOrigin::Explicit)
            .await?;
        let n1: std::collections::BTreeSet<String> = ["n1".to_owned()].into_iter().collect();
        ns_store
            .set_namespace_placement("t", NamespacePlacement::Prefer { nodes: n1 })
            .await?;

        let registry = crate::worker::registry::ConnectedWorkerRegistry::default();
        let (_n1_reg, mut n1_rx) = register_node_worker(&registry, "t", "n1", "charge")?;
        let (_n2_reg, mut n2_rx) = register_node_worker(&registry, "t", "n2", "charge")?;
        let (dispatch, gate) = placement_dispatch(&registry, Arc::clone(&ns_store));

        let row = unpinned_row("t");
        dispatch_claimed(&dispatch, &gate, &row).await?;

        assert!(
            n1_rx.recv().await.is_some(),
            "the n1 worker receives the task"
        );
        assert!(
            n2_rx.try_recv().is_err(),
            "the n2 worker must NOT receive the task while n1 is live"
        );
        // The recorded row's node is UNTOUCHED by preference (determinism gate).
        assert_eq!(row.node, None, "placement must never mutate the row's node");
        Ok(())
    }

    /// P2-P3 (spill): an unpinned row in a `Prefer{n1}` namespace SPILLS to the
    /// only live worker (n2) when no n1 worker is connected — the demoable
    /// node-loss failover behaviour.
    #[tokio::test]
    async fn prefer_spills_to_any_worker_when_preferred_node_absent()
    -> Result<(), Box<dyn std::error::Error>> {
        use aion_store::{InMemoryStore, NamespaceOrigin, NamespacePlacement, NamespaceStore};

        let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        ns_store
            .register_namespace("t", NamespaceOrigin::Explicit)
            .await?;
        let n1: std::collections::BTreeSet<String> = ["n1".to_owned()].into_iter().collect();
        ns_store
            .set_namespace_placement("t", NamespacePlacement::Prefer { nodes: n1 })
            .await?;

        // Only an n2 worker is live: no n1-labelled worker exists at all.
        let registry = crate::worker::registry::ConnectedWorkerRegistry::default();
        let (_n2_reg, mut n2_rx) = register_node_worker(&registry, "t", "n2", "charge")?;
        let (dispatch, gate) = placement_dispatch(&registry, Arc::clone(&ns_store));

        let row = unpinned_row("t");
        dispatch_claimed(&dispatch, &gate, &row).await?;

        assert!(
            n2_rx.recv().await.is_some(),
            "with no n1 worker live, the dispatch spills to the live n2 worker"
        );
        assert_eq!(row.node, None, "spill must never mutate the row's node");
        Ok(())
    }

    /// P2-P3 (unplaced unchanged): an `Unplaced` namespace dispatches to any live
    /// worker exactly as before, regardless of node label.
    #[tokio::test]
    async fn unplaced_namespace_dispatches_to_any_worker() -> Result<(), Box<dyn std::error::Error>>
    {
        use aion_store::{InMemoryStore, NamespaceOrigin, NamespaceStore};

        let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        // Registered but left Unplaced (the default).
        ns_store
            .register_namespace("t", NamespaceOrigin::Explicit)
            .await?;

        let registry = crate::worker::registry::ConnectedWorkerRegistry::default();
        let (_n2_reg, mut n2_rx) = register_node_worker(&registry, "t", "n2", "charge")?;
        let (dispatch, gate) = placement_dispatch(&registry, Arc::clone(&ns_store));

        dispatch_claimed(&dispatch, &gate, &unpinned_row("t")).await?;
        assert!(
            n2_rx.recv().await.is_some(),
            "an Unplaced namespace reaches any live worker"
        );
        Ok(())
    }

    /// P2-P3 (authored pin wins): a row with an authored node `Some(N)` STILL
    /// requires N regardless of the namespace's `Prefer{other}` placement — the
    /// per-activity pin is authoritative and the placement never overrides it.
    #[tokio::test]
    async fn authored_node_pin_wins_over_namespace_prefer() -> Result<(), Box<dyn std::error::Error>>
    {
        use aion_store::{InMemoryStore, NamespaceOrigin, NamespacePlacement, NamespaceStore};

        let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        ns_store
            .register_namespace("t", NamespaceOrigin::Explicit)
            .await?;
        // Namespace prefers n1, but the row is authored-pinned to n2.
        let n1: std::collections::BTreeSet<String> = ["n1".to_owned()].into_iter().collect();
        ns_store
            .set_namespace_placement("t", NamespacePlacement::Prefer { nodes: n1 })
            .await?;

        let registry = crate::worker::registry::ConnectedWorkerRegistry::default();
        let (_n1_reg, mut n1_rx) = register_node_worker(&registry, "t", "n1", "charge")?;
        let (_n2_reg, mut n2_rx) = register_node_worker(&registry, "t", "n2", "charge")?;
        let (dispatch, gate) = placement_dispatch(&registry, Arc::clone(&ns_store));

        // Authored pin: node = Some("n2").
        let row = unpinned_row("t").with_node(Some(String::from("n2")));
        dispatch_claimed(&dispatch, &gate, &row).await?;

        assert!(
            n2_rx.recv().await.is_some(),
            "the authored Some(n2) pin is honoured regardless of the namespace Prefer{{n1}}"
        );
        assert!(
            n1_rx.try_recv().is_err(),
            "the preferred-n1 worker must NOT receive a task authored-pinned to n2"
        );
        // The authored node is preserved exactly (determinism gate).
        assert_eq!(row.node.as_deref(), Some("n2"));
        Ok(())
    }

    /// DETERMINISM GATE (non-negotiable): the recorded row's `node` is
    /// byte-identical regardless of WHICH worker placement routed the activity to.
    /// Under `Prefer{n1}`, the same unpinned row dispatched once to the n1 worker
    /// and once (after n1 leaves) spilled to n2 keeps `node == None` BOTH times —
    /// `to_scheduled` reads the row's node, never the placement, so replay sees an
    /// identical command stream irrespective of the live dispatch target.
    #[tokio::test]
    async fn placement_never_mutates_recorded_row_node_across_routings()
    -> Result<(), Box<dyn std::error::Error>> {
        use aion_store::{InMemoryStore, NamespaceOrigin, NamespacePlacement, NamespaceStore};

        let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        ns_store
            .register_namespace("t", NamespaceOrigin::Explicit)
            .await?;
        let n1: std::collections::BTreeSet<String> = ["n1".to_owned()].into_iter().collect();
        ns_store
            .set_namespace_placement("t", NamespacePlacement::Prefer { nodes: n1 })
            .await?;
        let registry = crate::worker::registry::ConnectedWorkerRegistry::default();
        let (dispatch, gate) = placement_dispatch(&registry, Arc::clone(&ns_store));

        // Routing A: n1 worker present → preferred selection.
        let (n1_reg, mut n1_rx) = register_node_worker(&registry, "t", "n1", "charge")?;
        let row_a = unpinned_row("t");
        dispatch_claimed(&dispatch, &gate, &row_a).await?;
        assert!(n1_rx.recv().await.is_some());
        let scheduled_a = super::WorkerOutboxDispatch::to_scheduled(&row_a);

        // n1 leaves; only n2 remains.
        n1_reg.deregister()?;
        let (_n2_reg, mut n2_rx) = register_node_worker(&registry, "t", "n2", "charge")?;

        // Routing B: same shape of unpinned row → spills to n2.
        let row_b = unpinned_row("t");
        dispatch_claimed(&dispatch, &gate, &row_b).await?;
        assert!(n2_rx.recv().await.is_some());
        let scheduled_b = super::WorkerOutboxDispatch::to_scheduled(&row_b);

        // The recorded row node — and thus the scheduled task's node — is None in
        // BOTH routings: the dispatch target (n1 vs n2) did not perturb it.
        assert_eq!(row_a.node, None);
        assert_eq!(row_b.node, None);
        assert_eq!(
            scheduled_a.node, scheduled_b.node,
            "the scheduled task node is identical regardless of which worker served it"
        );
        assert_eq!(
            scheduled_a.node, None,
            "an unpinned row stays unpinned at dispatch"
        );
        Ok(())
    }

    #[tokio::test]
    async fn claim_marks_row_claimed_then_sweep_advances_to_done()
    -> Result<(), Box<dyn std::error::Error>> {
        // Pins that the dispatcher reads only the outbox (claim → terminal),
        // never workflow history: a bare claim flips status to claimed, and a
        // successful sweep then advances it to done.
        let store = open_store("claimed").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let claimed = store.claim_outbox_rows(10).await?;
        assert_eq!(claimed.len(), 1);
        assert_eq!(claimed[0].status, OutboxStatus::Claimed);
        Ok(())
    }

    // --- P2-Q2: per-tenant keyed backpressure at the claim ------------------

    use crate::worker::{Backpressure, OwnedShardFraction, QuotaCache};

    /// Build a namespace store carrying an explicit `max_in_flight_activities`
    /// override for each `(namespace, quota)` pair, so the quota cache resolves a
    /// concrete per-tenant ceiling.
    async fn namespace_store_with_quotas(
        quotas: &[(&str, u32)],
    ) -> Result<Arc<dyn aion_store::NamespaceStore>, ServerError> {
        use aion_store::{InMemoryStore, NamespaceOrigin, NamespaceRecord, NamespaceStore};
        let store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        for (namespace, quota) in quotas {
            let mut record =
                NamespaceRecord::new_minted(namespace, NamespaceOrigin::Explicit, Utc::now());
            record.config.max_in_flight_activities = Some(*quota);
            store
                .put_namespace(record)
                .await
                .map_err(ServerError::from)?;
        }
        Ok(store)
    }

    /// Build own-all keyed backpressure (fraction 1) over `ns_store` with the given
    /// generous platform default; a zero TTL so each sweep reads the latest quota.
    fn own_all_backpressure(
        ns_store: Arc<dyn aion_store::NamespaceStore>,
        platform_default: u32,
    ) -> Backpressure {
        let quota = QuotaCache::new(ns_store, platform_default, Duration::ZERO);
        Backpressure::new(quota, OwnedShardFraction::own_all())
    }

    /// Append `count` fresh pending rows in `namespace` and immediately claim them,
    /// leaving them durably `Claimed` (concurrently executing, never completed) so
    /// they occupy `count` of the namespace's concurrency slots. Returns their
    /// dispatch keys so a test can later `complete_outbox_row` specific slots to
    /// free headroom.
    async fn seed_claimed(
        store: &Arc<dyn OutboxStore>,
        namespace: &str,
        count: usize,
    ) -> Result<Vec<String>, ServerError> {
        let rows: Vec<OutboxRow> = (0..count)
            .map(|_| {
                pending_row(&WorkflowId::new_v4(), 0)
                    .with_namespace(namespace)
                    .with_task_queue("default")
            })
            .collect();
        store.append_outbox_batch(&rows).await?;
        let claimed = store
            .claim_outbox_rows(u32::try_from(count).unwrap_or(u32::MAX))
            .await?;
        assert_eq!(
            claimed.len(),
            count,
            "seed must claim exactly the seeded rows"
        );
        Ok(claimed.into_iter().map(|row| row.dispatch_key).collect())
    }

    /// Append `count` fresh pending rows in `namespace`, left `Pending` (backlog).
    async fn seed_pending(
        store: &Arc<dyn OutboxStore>,
        namespace: &str,
        count: usize,
    ) -> Result<Vec<OutboxRow>, ServerError> {
        let rows: Vec<OutboxRow> = (0..count)
            .map(|_| {
                pending_row(&WorkflowId::new_v4(), 0)
                    .with_namespace(namespace)
                    .with_task_queue("default")
            })
            .collect();
        store.append_outbox_batch(&rows).await?;
        Ok(rows)
    }

    /// Append `count` fresh pending rows in `namespace` on `task_queue`, left
    /// `Pending`. Used to spread ONE namespace's backlog across several routes.
    async fn seed_pending_on_queue(
        store: &Arc<dyn OutboxStore>,
        namespace: &str,
        task_queue: &str,
        count: usize,
    ) -> Result<(), ServerError> {
        let rows: Vec<OutboxRow> = (0..count)
            .map(|_| {
                pending_row(&WorkflowId::new_v4(), 0)
                    .with_namespace(namespace)
                    .with_task_queue(task_queue)
            })
            .collect();
        store.append_outbox_batch(&rows).await?;
        Ok(())
    }

    async fn count_pending(store: &HaematiteStore, namespace: &str) -> Result<u64, ServerError> {
        // Pending = in-flight − claimed (both durable, namespace-scoped).
        let inflight = store.count_inflight_outbox_rows(namespace).await?;
        let claimed = store.count_claimed_outbox_rows(namespace).await?;
        Ok(inflight - claimed)
    }

    /// A tenant AT its ceiling holds its excess Pending rows (they stay Pending,
    /// never Failed/dropped), and dispatch resumes once Claimed rows complete and
    /// headroom returns. This is the keyed-backpressure core.
    #[tokio::test]
    async fn tenant_at_ceiling_holds_pending_then_resumes_when_headroom_returns()
    -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-ceiling").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        // Ceiling 3 for "t". Seed 3 Claimed (fills the ceiling) + 4 Pending backlog.
        let ns_store = namespace_store_with_quotas(&[("t", 3)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        let executing = seed_claimed(&store, "t", 3).await?;
        let backlog = seed_pending(&store, "t", 4).await?;

        // headroom = ceiling(3) − claimed(3) = 0: NOTHING new is claimed this sweep.
        let claimed = backpressure
            .claim_round_robin(&store, 16, &std::collections::HashSet::new())
            .await?;
        assert!(claimed.is_empty(), "at the ceiling, no new row is claimed");
        // Every backlog row is STILL Pending — held, not Failed, not dropped.
        assert_eq!(count_pending(&raw, "t").await?, 4);
        for row in &backlog {
            assert_eq!(
                raw.outbox_row_state(&row.dispatch_key)
                    .await?
                    .map(|state| state.status),
                Some(OutboxStatus::Pending),
                "a held backlog row stays Pending — never Failed or dropped"
            );
        }

        // Two executing (Claimed) rows complete → claimed drops 3 → 1 → headroom 2.
        for dispatch_key in executing.iter().take(2) {
            raw.complete_outbox_row(dispatch_key).await?;
        }
        assert_eq!(raw.count_claimed_outbox_rows("t").await?, 1);

        // Next sweep: headroom 2 → exactly 2 backlog rows are claimed, 2 stay Pending.
        let resumed = backpressure
            .claim_round_robin(&store, 16, &std::collections::HashSet::new())
            .await?;
        assert_eq!(resumed.len(), 2, "headroom returned, 2 held rows dispatch");
        assert_eq!(
            count_pending(&raw, "t").await?,
            2,
            "2 backlog rows still held"
        );
        Ok(())
    }

    /// #204 GATE-1 (production path): the pause dispatch-hold is honoured on the
    /// BACKPRESSURE (scoped) claim, which is what the production dispatcher runs. A
    /// held workflow's pending row is never claimed under `claim_round_robin` — it
    /// stays Pending the whole paused window — and once the hold is released the very
    /// next backpressure sweep claims it. Without the scoped exclusion the hold would
    /// silently evaporate under backpressure.
    #[tokio::test]
    async fn backpressure_claim_honours_the_pause_hold() -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-pause-hold").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        // Ample ceiling so nothing is held for quota reasons — only the pause hold
        // can keep the row Pending here.
        let ns_store = namespace_store_with_quotas(&[("t", 100)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        let seeded = seed_pending(&store, "t", 1).await?;
        let workflow_id = seeded[0].workflow_id.clone();

        // Held: the backpressure sweep must NOT claim the row.
        let mut held = std::collections::HashSet::new();
        held.insert(workflow_id.clone());
        let claimed = backpressure.claim_round_robin(&store, 16, &held).await?;
        assert!(
            claimed.is_empty(),
            "a paused run's row is never claimed under backpressure"
        );
        assert_eq!(
            raw.outbox_row_state(&seeded[0].dispatch_key)
                .await?
                .map(|state| state.status),
            Some(OutboxStatus::Pending),
            "the held row stays Pending (never Claimed) under the backpressure sweep"
        );

        // Released (resume): the next backpressure sweep claims it with no re-arm.
        let released = backpressure
            .claim_round_robin(&store, 16, &std::collections::HashSet::new())
            .await?;
        assert_eq!(
            released.len(),
            1,
            "the released row dispatches after resume"
        );
        Ok(())
    }

    /// A tenant with a BIG Pending backlog and ZERO Claimed is NOT wedged: it claims
    /// up to its ceiling. This is the whole point of CLAIMED-only headroom — a
    /// Pending+Claimed input would count the backlog against the ceiling and let the
    /// tenant claim nothing, wedging it against its own work.
    #[tokio::test]
    async fn big_pending_backlog_with_zero_claimed_is_not_wedged()
    -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-no-wedge").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        // Ceiling 5, zero Claimed, a 20-row Pending backlog.
        let ns_store = namespace_store_with_quotas(&[("t", 5)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        seed_pending(&store, "t", 20).await?;
        assert_eq!(raw.count_claimed_outbox_rows("t").await?, 0);
        assert_eq!(raw.count_inflight_outbox_rows("t").await?, 20);

        // headroom = ceiling(5) − claimed(0) = 5: it claims exactly 5, NOT zero.
        // (A Pending+Claimed headroom would be 5 − 20 = 0 and wedge the tenant.)
        let claimed = backpressure
            .claim_round_robin(&store, 100, &std::collections::HashSet::new())
            .await?;
        assert_eq!(
            claimed.len(),
            5,
            "claimed-only headroom lets a 0-claimed tenant claim up to its ceiling"
        );
        assert_eq!(
            count_pending(&raw, "t").await?,
            15,
            "the rest stay durably Pending"
        );
        Ok(())
    }

    /// FAIRNESS: two namespaces, one bursty (huge backlog) and one quiet (a single
    /// row). The quiet tenant gets a claim EVERY sweep — round-robin, never FIFO
    /// drain of the bursty tenant first. Neither exceeds its ceiling.
    #[tokio::test]
    async fn round_robin_gives_quiet_tenant_a_slot_every_sweep()
    -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-fairness").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        // Generous ceilings so the ceiling is not the limiter — fairness is.
        let ns_store = namespace_store_with_quotas(&[("bursty", 1000), ("quiet", 1000)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        seed_pending(&store, "bursty", 500).await?;
        seed_pending(&store, "quiet", 1).await?;

        // A single small-batch sweep: with FIFO the batch would be all-bursty and
        // never reach the one quiet row. Round-robin must give quiet a slot.
        let claimed = backpressure
            .claim_round_robin(&store, 8, &std::collections::HashSet::new())
            .await?;
        assert!(
            claimed.iter().any(|row| row.namespace == "quiet"),
            "the quiet tenant's single row is claimed this very sweep (no FIFO starvation)"
        );
        assert!(
            claimed.iter().any(|row| row.namespace == "bursty"),
            "the bursty tenant is also served — round-robin shares, it does not block"
        );
        assert_eq!(
            count_pending(&raw, "quiet").await?,
            0,
            "quiet is fully drained"
        );
        Ok(())
    }

    /// FAIRNESS is per-NAMESPACE, not per-ROUTE — the exact case the reviewer proved
    /// failing under the old per-route budget. A bursty tenant SPREAD ACROSS MANY
    /// task queues (8 routes, 500 pending) and a quiet single-route tenant (1
    /// pending) share a small batch (8). Under a per-route budget the bursty tenant's
    /// 8 routes exhaust the whole batch before the quiet tenant's single route is
    /// reached, starving it for the sweep. Per-NAMESPACE fairness must reserve the
    /// quiet tenant a guaranteed slice up front, so its row is claimed THIS sweep
    /// while the bursty tenant is still served up to its own fair slice.
    #[tokio::test]
    async fn quiet_tenant_not_starved_by_bursty_tenant_spread_across_routes()
    -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-fairness-routes").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        // Generous ceilings so the ceiling is not the limiter — fairness is.
        let ns_store = namespace_store_with_quotas(&[("bursty", 1000), ("quiet", 1000)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        // Bursty tenant: 500 pending rows spread evenly across 8 distinct task_queues
        // (8 routes). Quiet tenant: a single row on one route.
        for q in 0..8u32 {
            seed_pending_on_queue(&store, "bursty", &format!("tq-{q}"), 500 / 8).await?;
        }
        seed_pending(&store, "quiet", 1).await?;

        // A single small-batch (8) sweep. With a per-route budget, the bursty
        // tenant's 8 routes consume all 8 before the quiet route is reached.
        let claimed = backpressure
            .claim_round_robin(&store, 8, &std::collections::HashSet::new())
            .await?;
        let bursty = claimed.iter().filter(|r| r.namespace == "bursty").count();
        let quiet = claimed.iter().filter(|r| r.namespace == "quiet").count();
        assert_eq!(
            quiet, 1,
            "the quiet tenant's single row IS claimed this sweep — per-namespace \
             fairness reserves it a slice even though the bursty tenant has 8 routes"
        );
        assert!(
            bursty >= 1,
            "the bursty tenant is also served up to its fair slice, not blocked"
        );
        assert!(
            bursty <= 7,
            "the bursty tenant cannot consume the whole batch and starve the quiet one"
        );
        assert_eq!(
            count_pending(&raw, "quiet").await?,
            0,
            "quiet is fully drained this sweep"
        );
        Ok(())
    }

    /// EXACTLY-ONCE under throttling: no row is dispatched twice, and the dedup
    /// guard is intact. Two full sweeps under a tight ceiling claim a total set with
    /// NO duplicate dispatch keys, and re-appending an already-staged batch is
    /// ignored (INSERT OR IGNORE), never re-claimed.
    #[tokio::test]
    async fn throttled_claim_dispatches_each_row_exactly_once()
    -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-exactly-once").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        let ns_store = namespace_store_with_quotas(&[("t", 3)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        let staged = seed_pending(&store, "t", 6).await?;

        // Re-appending the SAME batch is a dedup no-op (INSERT OR IGNORE): the row
        // set is unchanged, so throttling can never resurrect a duplicate.
        store.append_outbox_batch(&staged).await?;
        assert_eq!(
            raw.count_inflight_outbox_rows("t").await?,
            6,
            "no duplicate rows staged"
        );

        // Sweep repeatedly, completing each claimed row so headroom frees, until the
        // backlog drains. Collect every claimed dispatch key across all sweeps.
        let mut all_claimed: Vec<String> = Vec::new();
        for _ in 0..10 {
            let claimed = backpressure
                .claim_round_robin(&store, 16, &std::collections::HashSet::new())
                .await?;
            assert!(
                claimed.len() <= 3,
                "the ceiling caps concurrent claims at 3 per sweep"
            );
            for row in &claimed {
                store.complete_outbox_row(&row.dispatch_key).await?;
                all_claimed.push(row.dispatch_key.clone());
            }
            if all_claimed.len() == 6 {
                break;
            }
        }
        // Every one of the 6 rows dispatched exactly once — no double-dispatch.
        all_claimed.sort();
        all_claimed.dedup();
        assert_eq!(
            all_claimed.len(),
            6,
            "all 6 rows dispatched, each exactly once"
        );
        // Nothing remains claimable: the backlog is fully drained (nothing dropped).
        assert!(
            backpressure
                .claim_round_robin(&store, 16, &std::collections::HashSet::new())
                .await?
                .is_empty()
        );
        Ok(())
    }

    /// REPLAY-STABILITY: a heavily-throttled fan-out claim shapes only WHICH rows
    /// dispatch this sweep and WHEN — it never mutates a claimed row's recorded
    /// identity (`workflow_id`, `ordinal`, `node`, `input`). The scheduled task derived from
    /// a throttled claim is byte-identical to the un-throttled claim of the same row,
    /// so replay sees an identical command stream regardless of throttling.
    #[tokio::test]
    async fn throttled_claim_preserves_recorded_row_identity()
    -> Result<(), Box<dyn std::error::Error>> {
        // Un-throttled baseline: claim a fixed set of rows straight off the store.
        let baseline_store = open_store("bp-replay-baseline").await?;
        let base: Arc<dyn OutboxStore> = baseline_store.clone();
        let base_rows = seed_pending(&base, "t", 4).await?;
        let baseline = base.claim_outbox_rows(16).await?;

        // Throttled: the SAME logical rows (same workflow_ids/ordinals) under a tight
        // ceiling, claimed across several headroom-limited sweeps.
        let throttled_store = open_store("bp-replay-throttled").await?;
        let store: Arc<dyn OutboxStore> = throttled_store.clone();
        let ns_store = namespace_store_with_quotas(&[("t", 1)]).await?;
        let backpressure = own_all_backpressure(ns_store, 1024);
        // Re-stage rows with identical (workflow_id, ordinal) so dispatch keys match.
        let replayed: Vec<OutboxRow> = base_rows
            .iter()
            .map(|row| {
                pending_row(&row.workflow_id, row.ordinal)
                    .with_namespace("t")
                    .with_task_queue("default")
            })
            .collect();
        store.append_outbox_batch(&replayed).await?;

        let mut throttled = Vec::new();
        for _ in 0..10 {
            let claimed = backpressure
                .claim_round_robin(&store, 16, &std::collections::HashSet::new())
                .await?;
            for row in &claimed {
                store.complete_outbox_row(&row.dispatch_key).await?;
            }
            throttled.extend(claimed);
            if throttled.len() == 4 {
                break;
            }
        }

        // The recorded identity of each row (the replay-visible content) is identical
        // between the un-throttled and throttled claims — only claim timing differed.
        let key = |rows: &[OutboxRow]| -> Vec<(String, u64, Option<String>)> {
            let mut identity: Vec<(String, u64, Option<String>)> = rows
                .iter()
                .map(|row| (row.dispatch_key.clone(), row.ordinal, row.node.clone()))
                .collect();
            identity.sort();
            identity
        };
        assert_eq!(
            key(&baseline),
            key(&throttled),
            "throttling changed neither the dispatch key set nor any row's recorded identity"
        );
        Ok(())
    }

    /// BYTE-IDENTICAL DEFAULT: with the generous platform default and NO tenant
    /// override, the round-robin claim admits the SAME ROW SET the plain unscoped
    /// claim would — the ceiling never engages for normal load. Proven by seeding the
    /// SAME logical rows (identical `(workflow_id, ordinal)` → identical dispatch
    /// keys) into both stores, claiming each way, and asserting the claimed
    /// dispatch-key SETS are equal (not merely the counts).
    #[tokio::test]
    async fn default_ceiling_claim_matches_unscoped_claim() -> Result<(), Box<dyn std::error::Error>>
    {
        use aion_store::{InMemoryStore, NamespaceStore};

        // One shared set of 10 logical rows drives BOTH stores, so the derived
        // dispatch keys are identical and the claimed SETS are directly comparable.
        let shared: Vec<OutboxRow> = (0..10)
            .map(|ordinal| {
                pending_row(&WorkflowId::new_v4(), ordinal)
                    .with_namespace("t")
                    .with_task_queue("default")
            })
            .collect();

        // Backpressure store: no tenant override anywhere, generous default 1024.
        let bp_store_raw = open_store("bp-default-bp").await?;
        let bp_store: Arc<dyn OutboxStore> = bp_store_raw.clone();
        let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
        let backpressure = own_all_backpressure(ns_store, 1024);
        bp_store.append_outbox_batch(&shared).await?;

        // Plain store: the SAME 10 rows, claimed with the unscoped path.
        let plain_raw = open_store("bp-default-plain").await?;
        let plain: Arc<dyn OutboxStore> = plain_raw.clone();
        plain.append_outbox_batch(&shared).await?;

        let via_bp = backpressure
            .claim_round_robin(&bp_store, 16, &std::collections::HashSet::new())
            .await?;
        let via_plain = plain.claim_outbox_rows(16).await?;

        // Compare the actual claimed dispatch-key SETS, not just the lengths: this
        // genuinely proves byte-identical default admission, row for row.
        let key_set = |rows: &[OutboxRow]| -> std::collections::BTreeSet<String> {
            rows.iter().map(|row| row.dispatch_key.clone()).collect()
        };
        let bp_keys = key_set(&via_bp);
        let plain_keys = key_set(&via_plain);
        assert_eq!(
            bp_keys, plain_keys,
            "under the generous default the ceiling never engages: the SAME row set \
             is claimed, dispatch key for dispatch key"
        );
        assert_eq!(
            bp_keys,
            key_set(&shared),
            "all 10 seeded rows claim in one sweep (headroom ≫ backlog)"
        );
        // Nothing held on the backpressure path — no Pending remains, exactly as the
        // plain unscoped claim leaves nothing Pending: byte-identical for normal load.
        assert_eq!(count_pending(&bp_store_raw, "t").await?, 0);
        assert_eq!(count_pending(&plain_raw, "t").await?, 0);
        Ok(())
    }

    /// PROPORTIONAL per-node enforcement: a node owning a fraction f of shards caps
    /// at ceil(quota × f), not the full cluster-wide quota — so per-node ceilings sum
    /// to ≈quota with no central counter. A node owning 2 of 8 shards under quota 8
    /// claims at most ceil(8 × 2/8) = 2 per sweep.
    #[tokio::test]
    async fn proportional_ceiling_caps_a_partial_shard_node()
    -> Result<(), Box<dyn std::error::Error>> {
        let raw = open_store("bp-proportional").await?;
        let store: Arc<dyn OutboxStore> = raw.clone();
        let ns_store = namespace_store_with_quotas(&[("t", 8)]).await?;
        // This node owns 2 of 8 shards → fraction 1/4 → per-node ceiling ceil(8/4)=2.
        let quota = QuotaCache::new(ns_store, 1024, Duration::ZERO);
        let backpressure = Backpressure::new(quota, OwnedShardFraction::new(2, 8));
        seed_pending(&store, "t", 10).await?;

        let claimed = backpressure
            .claim_round_robin(&store, 16, &std::collections::HashSet::new())
            .await?;
        assert_eq!(
            claimed.len(),
            2,
            "the per-node ceiling ceil(quota × owned/total) = ceil(8 × 2/8) = 2 caps the claim"
        );
        assert_eq!(
            count_pending(&raw, "t").await?,
            8,
            "the remaining 8 stay durably Pending"
        );
        Ok(())
    }

    /// FENCE-1 T3's seam: a dispatch that RE-DELIVERS the row's own attempt
    /// between the delivery and the worker's completion.
    ///
    /// It mints the authorization the first worker receives, then mints a second
    /// one for the SAME run and the SAME attempt — the re-dispatch an
    /// attempt-neutral transport loss produces — and reports the dispatch as the
    /// success it genuinely was, so `process_row` settles the row exactly as
    /// production does. No clock, no sleep, and no production test seam: the
    /// redelivery is forced by the stub standing where a worker transport
    /// stands, using only the public fence operations the real dispatch paths
    /// call.
    struct RedeliveringDispatch {
        fences: CompletionFences,
        issued: Mutex<Vec<CompletionToken>>,
    }

    impl RedeliveringDispatch {
        fn new(fences: CompletionFences) -> Self {
            Self {
                fences,
                issued: Mutex::new(Vec::new()),
            }
        }

        fn issued(&self) -> Result<Vec<CompletionToken>, ServerError> {
            Ok(self
                .issued
                .lock()
                .map_err(|_| ServerError::lock_poisoned("redelivering dispatch"))?
                .clone())
        }
    }

    #[async_trait]
    impl OutboxRowDispatch for RedeliveringDispatch {
        async fn dispatch(&self, row: &OutboxRow) -> Result<(), ServerError> {
            // The same three coordinates `WorkerOutboxDispatch::to_scheduled`
            // derives: the ordinal is the activity id, the row's run is the
            // execution generation, and the stored zero-based attempt is stamped
            // one-based for the wire.
            let activity_id = ActivityId::from_sequence_position(row.ordinal);
            let run_id = row.run_id.clone().ok_or_else(|| {
                ServerError::worker_dispatch(
                    "default",
                    "charge",
                    "outbox row has no run id; refusing unfenced external effect",
                )
            })?;
            let attempt = row.started_attempt;
            let delivered = self
                .fences
                .issue(&row.workflow_id, &run_id, &activity_id, attempt)?;
            let redelivered =
                self.fences
                    .issue(&row.workflow_id, &run_id, &activity_id, attempt)?;
            let mut issued = self
                .issued
                .lock()
                .map_err(|_| ServerError::lock_poisoned("redelivering dispatch"))?;
            issued.push(delivered);
            issued.push(redelivered);
            Ok(())
        }
    }

    /// FENCE-1 T3: end to end through the outbox dispatcher — a redelivery
    /// between dispatch and completion does not orphan the finished result, and
    /// nothing retries either way.
    ///
    /// Step 4 is the half that makes the base transcript a picture of the defect
    /// rather than of a refusal: it reads the same on the base and after the
    /// change. The row settled `Done` at DISPATCH because the dispatch genuinely
    /// succeeded, so no repair path can see it — `outbox_reconciler` re-arms only
    /// `claimed` rows, `outbox_settle` only cancels rows of hard-terminal
    /// workflows, and `outbox_redrive` only reaches dead letters. Only step 2
    /// flips: on the base the worker that genuinely held the first delivery is
    /// refused, and its result is simply gone.
    #[tokio::test]
    async fn a_redelivery_between_dispatch_and_completion_does_not_orphan_the_result()
    -> Result<(), Box<dyn std::error::Error>> {
        let store = open_store("redelivery-fence").await?;
        let workflow_id = WorkflowId::new_v4();
        let row = pending_row(&workflow_id, 0);
        store
            .append_outbox_batch(std::slice::from_ref(&row))
            .await?;

        let fences = CompletionFences::default();
        let dispatch = Arc::new(RedeliveringDispatch::new(fences.clone()));
        let dispatcher = Arc::new(
            OutboxDispatcher::new(store.clone(), config()).with_dispatch(dispatch.clone()),
        );
        dispatcher.sweep_once().await;

        let activity_id = ActivityId::from_sequence_position(row.ordinal);
        let issued = dispatch.issued()?;
        let [delivered, redelivered] = issued.as_slice() else {
            return Err("the dispatch must mint the delivery and its redelivery".into());
        };

        // 1. Done-on-dispatch is untouched and still pinned.
        assert_eq!(
            store
                .outbox_row_state(&row.dispatch_key)
                .await?
                .map(|state| state.status),
            Some(OutboxStatus::Done),
            "the row settles Done at DISPATCH; this fix does not move the settle point"
        );

        // 2. The worker that genuinely held the FIRST delivery completes, and its
        //    result is accepted.
        let accepted = fences.accept(&workflow_id, &activity_id, delivered);
        assert!(
            accepted.is_ok(),
            "the finished result of the worker that held the first delivery must be accepted: \
             {accepted:?}"
        );

        // 3. The redelivered worker's completion is then the duplicate.
        let duplicate = fences.accept(&workflow_id, &activity_id, redelivered);
        assert!(
            matches!(
                duplicate,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::NoCurrentGeneration,
                    ..
                })
            ),
            "the first accepted completion consumes every outstanding token for the site: \
             {duplicate:?}"
        );

        // 4. And nothing retried — the absence half of the red proof, which reads
        //    identically on the base.
        let state = store
            .outbox_row_state(&row.dispatch_key)
            .await?
            .ok_or("the outbox row vanished")?;
        assert_eq!(
            state.status,
            OutboxStatus::Done,
            "the row is still Done: not re-armed to Pending, and not dead-lettered to Failed"
        );
        assert_eq!(
            state.attempt, row.attempt,
            "no attempt was consumed, so no retry budget moved"
        );
        assert!(
            store.claim_outbox_rows(10).await?.is_empty(),
            "no row is claimable: nothing in the durable outbox will ever re-offer this work"
        );
        Ok(())
    }
}