aion_server/worker/

bridge.rs

//! NIF bridge dispatcher that routes `run_activity` calls to connected workers.
//!
//! `WorkerActivityDispatcher` implements `aion::ActivityDispatcher` so the
//! engine's activity NIFs can synchronously dispatch to a remote worker and
//! block until the result comes back.
//!
//! # Threading contract
//!
//! The engine invokes [`aion::ActivityDispatcher::dispatch`] from two kinds of
//! threads: beamr scheduler threads (concurrency combinators) and spawned
//! tokio tasks (the two-phase `dispatch_activity` completion task). The task
//! send uses `try_send()` (non-blocking channel push) and the response wait
//! blocks on `std::sync::mpsc::Receiver::recv`.
//!
//! Blocking is harmless on a beamr thread, but on a tokio runtime worker it
//! must be wrapped in `tokio::task::block_in_place`: the `try_send` wakes the
//! per-worker gRPC stream forwarder task, and tokio schedules a task woken
//! from task context into the *current* worker's LIFO slot, which no other
//! runtime worker can steal. Without the `block_in_place` core handoff the
//! forwarder sits trapped in that slot while this thread blocks, so the queued
//! `ActivityTask` is never flushed to the worker even though the worker is
//! healthy. `block_in_place` moves the worker's scheduler core (LIFO slot
//! included) to another thread before the wait begins, so dispatch-to-delivery
//! stays in the millisecond range and the runtime keeps full parallelism.
//!
//! # Wait termination
//!
//! The engine imposes no activity timeout of its own: agent-style activities
//! legitimately run for over an hour, so the completion wait is unbounded.
//! The blocking `recv` terminates on exactly one of:
//!
//! - **Completion** — the worker reports a result and the stream handler
//!   delivers it through [`ActivityCompletionSink::complete_activity`].
//! - **Worker loss** — the worker's gRPC stream ends (process death,
//!   disconnect, expired token); the stream teardown sweeps the worker's
//!   in-flight tasks through the same sink as every other TRANSPORT loss
//!   ([`HeartbeatTracker::fail_disconnected_worker`]). A liminal-delivered
//!   worker's loss is observed by its reply router thread instead (the
//!   correlated-reply awaiter wakes the moment the connection closes) and
//!   resolves the dispatch with the same transport-loss class. Both are
//!   classified by [`transport_loss`](crate::worker::transport_loss): `lost:`
//!   while the transport still has budget to deliver the activity (the engine
//!   re-dispatches the SAME attempt, recording nothing), `transport-exhausted:`
//!   once that budget is spent. Neither ever wears the action's retry
//!   vocabulary — the activity never ran.
//! - **Graceful-drain park (#207)** — during a drain, a worker stream ending
//!   (or the drain-timeout backstop) PARKS the worker's in-flight tasks
//!   through [`ActivityCompletionSink::park_activity`]: the waiter resolves
//!   with the ephemeral parked sentinel
//!   ([`aion::PARKED_ACTIVITY_REASON`]), nothing is recorded or delivered,
//!   and restart recovery re-dispatches the dangling ordinal — kill -9
//!   convergence.
//! - **Drain timeout at shutdown** — the shutdown coordinator parks all
//!   remaining in-flight tasks through the sink
//!   (`HeartbeatTracker::park_all_in_flight_workers`).
//! - **Channel teardown** — every sender for the pending entry is dropped
//!   (a cleanup path removed the entry without completing it); surfaced as
//!   a channel-closed dispatch error, never a hang.
//!
//! An activity's duration is bounded only by the workflow's own
//! `timeout_seconds` and by worker liveness — never by an engine constant.

use std::collections::BTreeMap;
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};

use aion::{ActivityDispatch, ActivityDispatcher, LeaseSignal};
use aion_core::{ActivityId, RunId, WorkflowId};
use aion_proto::{ProtoActivityId, ProtoActivityTask, ProtoPayload, ProtoWorkflowId};
use dashmap::DashMap;
use dashmap::mapref::entry::Entry;

use super::dispatch::{ActivityCompletion, ActivityCompletionSink};
use super::envelope::{CompletionFences, CompletionToken, idempotency_key};
use super::heartbeat::{HeartbeatTracker, InFlightActivity};
use super::queue_service::{
    DeliveryRefusal, ExpiredClock, PARK_POLL_INTERVAL, PoolCensus, QueueDeclarationSource,
    QueueServiceConfig, QueueServiceReason, QueueServiceState, SelectionRefusal, ServiceAddress,
    ServiceWait, WorkerUnavailable, deliver_within_schedule_to_start, select_worker_or_refuse,
};
use super::registry::{
    ConnectedWorkerRegistry, DispatchReservation, WorkerArrival, WorkerDelivery, WorkerHandle,
    WorkerId, WorkerMessage,
};
use crate::error::ServerError;
use crate::shutdown::DrainState;

mod completion_outcome;
use tracing::info_span;

type SyncSender = std::sync::mpsc::SyncSender<Result<String, String>>;

/// One live dispatch's waiter slot: the channel that unblocks its
/// `await_activity_result`, and the attempt it was dispatched at.
///
/// The attempt travels with the sender so resolution paths owned by ONE
/// dispatch (`deliver_result`, `cleanup_activity`) remove the entry only while
/// it is still that dispatch's own — after a higher attempt takes the site
/// over (aion#195), the superseded dispatch's teardown must not evict the live
/// waiter that replaced it. Whole-site resolutions (a fence-accepted
/// completion, a park) stay unconditional: they retire the site, whoever
/// holds it.
struct PendingWaiter {
    sender: SyncSender,
    attempt: u32,
}

/// The reason a superseded waiter is resolved with when a higher attempt takes
/// over its execution site (aion#195).
///
/// Worn by the DROPPED dispatch future of an attempt whose per-attempt bound
/// already failed it, so it normally reaches nobody. It still wears the
/// `retryable:` prefix deliberately: if a race ever surfaces it to the retry
/// loop, a retry is the harmless reading (the fence refuses duplicates), where
/// a terminal reading would repeat the exact defect this exists to close.
fn superseded_site_reason(new_attempt: u32) -> String {
    format!(
        "retryable: superseded by attempt {new_attempt} at this execution site — the prior \
         attempt reached its bound, its completion token is fence-refused, and its worker \
         has been asked to stop"
    )
}
type SyncReceiver = std::sync::mpsc::Receiver<Result<String, String>>;

/// Execution-scoped key for an in-flight activity dispatch.
///
/// The engine seam ([`ActivityDispatch`]) carries the *real* workflow id and
/// the *real* per-workflow activity ordinal recorded in history, so this pair
/// uniquely and stably identifies one execution. Keying by bare [`ActivityId`]
/// would be unsafe across server restarts — a stale result re-reported from a
/// worker's previous session could complete a *different* post-restart
/// dispatch reusing the same ordinal — but pairing it with the real workflow
/// id closes that race: two different workflow executions never share a
/// workflow id, so a stale `(workflow_id, activity_id)` from a previous server
/// life can only ever match the exact execution it belongs to.
///
/// The wire (`ActivityResult`) carries both ids, plus an attempt discriminator
/// (`ActivityTask.attempt`). The pending key stays attempt-free for now: a
/// retry re-dispatches under the same `(workflow_id, activity_id)` and the
/// outstanding entry is the one awaiting completion. Redelivery bookkeeping
/// can widen this key with the wire attempt later — no protocol change needed.
type PendingActivityKey = (WorkflowId, ActivityId);

/// Routes an unmatched durable-outbox completion into the live workflow.
///
/// When the outbox is ON a worker completion can arrive at the sink with no
/// pending oneshot (the dispatch was non-blocking fan-out, or the original
/// waiter was lost). Rather than dropping it, [`PendingActivities::complete`]
/// hands it to this callback, which resolves the workflow to its live engine
/// process and delivers the terminal into its mailbox. The callback is only
/// installed when the outbox is enabled, so flag-off the unmatched branch
/// stays a silent drop.
pub trait OutboxDeliveryCallback: Send + Sync {
    /// Deliver a successful completion to the live workflow.
    ///
    /// Returns `Ok(true)` when delivered to a live workflow and `Ok(false)`
    /// when no run is currently live (the expected stale-completion case that
    /// recovery re-arms).
    ///
    /// # Errors
    ///
    /// Returns [`ServerError`] when the engine rejects the delivery.
    fn deliver_completion(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        run_id: Option<&RunId>,
        result: String,
    ) -> Result<bool, ServerError>;

    /// Deliver a failure to the live workflow. Same `bool`/error contract as
    /// [`Self::deliver_completion`].
    ///
    /// # Errors
    ///
    /// Returns [`ServerError`] when the engine rejects the delivery.
    fn deliver_failure(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        run_id: Option<&RunId>,
        reason: String,
    ) -> Result<bool, ServerError>;
}

/// Tracks in-flight activity dispatches waiting for worker results.
///
/// When the server's worker stream handler receives an `ActivityResult`, it
/// calls [`complete_activity`](ActivityCompletionSink::complete_activity) to
/// deliver the result to the blocked NIF thread. Entries are keyed by
/// [`PendingActivityKey`] so a stale result from a previous server life can
/// never be matched to a different execution (#59).
///
/// Clones share both the pending map and the outbox-delivery callback through
/// `Arc`, so [`set_outbox_delivery`](Self::set_outbox_delivery) called once on
/// any clone after construction is visible to the clone the dispatcher holds.
#[derive(Clone)]
pub struct PendingActivities {
    pending: Arc<DashMap<PendingActivityKey, PendingWaiter>>,
    completion_fences: CompletionFences,
    outbox_delivery: Arc<OnceLock<Arc<dyn OutboxDeliveryCallback>>>,
    /// Where an accepted delivery's lease is recorded (WA-010 R3), shared by
    /// the bridge and every outbox dispatcher built over these seams so both
    /// paths attribute through one recorder and count on one ledger.
    lease_recorder: super::lease_record::LeaseRecorderSeam,
    /// The transport's OWN re-dispatch budget for activities whose worker died
    /// before reporting (see [`transport_loss`](crate::worker::transport_loss)).
    /// Shared across clones, so every loss for one execution site lands in one
    /// budget. There is no default: the window is required at construction, so
    /// a zero budget cannot be inherited by omission.
    transport_losses: super::transport_loss::TransportLossLedger,
}

impl std::fmt::Debug for PendingActivities {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PendingActivities")
            .field("pending", &self.pending.len())
            .field("completion_fences", &self.completion_fences)
            .field(
                "outbox_delivery_installed",
                &self.outbox_delivery.get().is_some(),
            )
            .field("lease_recorder", &self.lease_recorder)
            .field("transport_losses", &self.transport_losses)
            .finish()
    }
}

impl PendingActivities {
    /// Register the waiter for one dispatch, taking over the execution site
    /// from a superseded lower attempt when one still holds it (aion#195).
    ///
    /// The third return names the attempt that was superseded, so the caller
    /// can walk the release ladder for it (ask its worker to stop, untrack its
    /// heartbeat task). The fence issue above already superseded the holder's
    /// completion tokens — a takeover changes who may still answer, never what
    /// would have been accepted.
    fn insert(
        &self,
        workflow_id: WorkflowId,
        run_id: &RunId,
        activity_id: ActivityId,
        attempt: u32,
    ) -> Result<(CompletionToken, SyncReceiver, Option<u32>), ServerError> {
        let completion_token =
            self.completion_fences
                .issue(&workflow_id, run_id, &activity_id, attempt)?;
        let (tx, rx) = std::sync::mpsc::sync_channel(1);
        let key = (workflow_id, activity_id);
        match self.pending.entry(key) {
            Entry::Vacant(entry) => {
                entry.insert(PendingWaiter {
                    sender: tx,
                    attempt,
                });
                Ok((completion_token, rx, None))
            }
            Entry::Occupied(mut entry) => {
                let held_attempt = entry.get().attempt;
                if attempt > held_attempt {
                    // The holder is a corpse: issuing this attempt's token
                    // superseded its generation, so nothing it answers can be
                    // accepted. Waking it with the superseded sentinel lets
                    // its blocked dispatcher thread finish; its attempt-aware
                    // teardown cannot evict this replacement.
                    let superseded = entry.insert(PendingWaiter {
                        sender: tx,
                        attempt,
                    });
                    let _ = superseded.sender.send(Err(superseded_site_reason(attempt)));
                    Ok((completion_token, rx, Some(held_attempt)))
                } else {
                    let (workflow_id, activity_id) = entry.key();
                    Err(ServerError::PendingActivityCollision {
                        workflow_id: workflow_id.clone(),
                        activity_id: activity_id.clone(),
                        held_attempt,
                        incoming_attempt: attempt,
                    })
                }
            }
        }
    }

    /// Share the generation registry with non-blocking outbox dispatch.
    #[must_use]
    pub fn completion_fences(&self) -> CompletionFences {
        self.completion_fences.clone()
    }

    /// Test seam: register a pending waiter exactly as a live dispatch does,
    /// so crate-internal tests outside this module (the stream-teardown
    /// park-vs-fail pins) can observe how a sweep resolves it.
    #[cfg(test)]
    pub(crate) fn insert_for_test(
        &self,
        workflow_id: WorkflowId,
        run_id: &RunId,
        activity_id: ActivityId,
        attempt: u32,
    ) -> Result<(CompletionToken, SyncReceiver, Option<u32>), ServerError> {
        self.insert(workflow_id, run_id, activity_id, attempt)
    }

    /// Install the unmatched-completion delivery callback (idempotent).
    ///
    /// Set once, after construction, when the durable outbox is enabled. A
    /// second set is ignored and logged: the callback is process-wide and must
    /// not silently change identity.
    pub fn set_outbox_delivery(&self, callback: Arc<dyn OutboxDeliveryCallback>) {
        if self.outbox_delivery.set(callback).is_err() {
            tracing::warn!("outbox delivery callback already installed; ignoring duplicate set");
        }
    }

    /// Complete a pending dispatch, or route an unmatched completion to the
    /// outbox delivery callback when one is installed.
    ///
    /// A matched entry delivers to its waiting oneshot exactly as before. An
    /// unmatched completion is dropped silently when no callback is installed
    /// (outbox OFF — byte-identical to the prior behaviour); with a callback
    /// installed (outbox ON) it is routed into the live workflow's mailbox.
    fn complete(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        run_id: Option<&RunId>,
        result: Result<String, String>,
    ) -> bool {
        // Take and drop the DashMap guard before any callback runs: the engine
        // delivery the callback invokes must never execute under a shard lock.
        let matched = self
            .pending
            .remove(&(workflow_id.clone(), activity_id.clone()));
        if let Some((_, waiter)) = matched {
            return waiter.sender.send(result).is_ok();
        }
        let Some(callback) = self.outbox_delivery.get() else {
            // Outbox OFF: silent drop, byte-identical to the prior behaviour.
            return false;
        };
        let outcome = match result {
            Ok(payload) => callback.deliver_completion(workflow_id, activity_id, run_id, payload),
            Err(reason) => callback.deliver_failure(workflow_id, activity_id, run_id, reason),
        };
        match outcome {
            Ok(true) => true,
            Ok(false) => {
                // Not live: the expected stale-completion case recovery re-arms.
                tracing::debug!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    "unmatched outbox completion for a workflow that is not currently live; \
                     recovery will re-arm it"
                );
                false
            }
            Err(error) => {
                tracing::warn!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    %error,
                    "failed to deliver unmatched outbox completion to the live workflow"
                );
                false
            }
        }
    }

    fn complete_fenced_after_accept(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        run_id: Option<&RunId>,
        completion_token: &CompletionToken,
        result: Result<String, String>,
        after_accept: impl FnOnce() -> Result<bool, ServerError>,
    ) -> Result<bool, ServerError> {
        // The acceptance hands back the generation it consumed — every token
        // outstanding for this attempt, not just the one submitted — so the two
        // restore arms below put back exactly what was taken rather than
        // re-deriving it and inventing a narrower generation.
        let accepted = self
            .completion_fences
            .accept(workflow_id, activity_id, completion_token)
            .inspect_err(|error| {
                tracing::warn!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    %error,
                    "activity completion rejected by execution-generation fence"
                );
            })?;
        let should_publish = match after_accept() {
            Ok(should_publish) => should_publish,
            Err(error) => {
                if let Err(restore_error) =
                    self.completion_fences
                        .restore_if_absent(workflow_id, activity_id, &accepted)
                {
                    tracing::error!(%restore_error, "failed to restore completion generation after settlement failure");
                }
                return Err(error);
            }
        };
        if !should_publish {
            // A non-publishing settlement means ANOTHER path owns this
            // dispatch's resolution (the lost-worker sweep, typically). That
            // path will present the same token to `accept` — so the
            // generation this call consumed must be restored, exactly as on
            // the failure arm, or the true resolver is refused with
            // `NoCurrentGeneration` and the waiter is never resolved.
            if let Err(restore_error) =
                self.completion_fences
                    .restore_if_absent(workflow_id, activity_id, &accepted)
            {
                tracing::error!(%restore_error, "failed to restore completion generation after a non-publishing settlement");
            }
            return Ok(false);
        }
        // A resolution OUTSIDE the transport domain — a real result, or a real
        // action failure — retires this execution site's transport-loss budget,
        // so a run that survived one blip does not carry it forward. A
        // transport-domain resolution deliberately does not: the running budget
        // is exactly what bounds a flapping link.
        let transport_domain = result
            .as_ref()
            .err()
            .is_some_and(|reason| super::transport_loss::is_transport_domain_reason(reason));
        if !transport_domain
            && let Err(error) = self.transport_losses.clear(workflow_id, activity_id)
        {
            tracing::warn!(
                workflow_id = %workflow_id,
                activity_id = %activity_id,
                %error,
                "failed to retire the transport-loss budget for a resolved activity"
            );
        }
        Ok(self.complete(workflow_id, activity_id, run_id, result))
    }

    /// Build a sink whose transport-loss ledger derives its budget from the
    /// operator's heartbeat window (see
    /// [`transport_loss`](crate::worker::transport_loss)).
    ///
    /// The window is a CONSTRUCTOR parameter rather than a builder step, and
    /// that is the point: there is no way to obtain a `PendingActivities`
    /// without naming the window, so a wiring cannot supply everything else and
    /// silently inherit a zero budget. The previous shape —
    /// `default().with_heartbeat_window(w)` — was two calls where the second was
    /// omissible, and nothing in the test suite could tell the wired world from
    /// the unwired one.
    #[must_use]
    pub fn new(heartbeat_window: std::time::Duration) -> Self {
        Self {
            pending: Arc::default(),
            completion_fences: CompletionFences::default(),
            outbox_delivery: Arc::default(),
            lease_recorder: super::lease_record::LeaseRecorderSeam::default(),
            transport_losses: super::transport_loss::TransportLossLedger::new(heartbeat_window),
        }
    }

    /// The shared lease-record seam, for an outbox dispatcher built over the
    /// same pending set to record through.
    #[must_use]
    pub fn lease_recorder(&self) -> super::lease_record::LeaseRecorderSeam {
        self.lease_recorder.clone()
    }

    /// Install the lease recorder once the engine exists; a duplicate install
    /// keeps the first and is logged by the seam.
    pub fn set_lease_recorder(
        &self,
        recorder: Arc<dyn super::lease_record::ActivityLeaseRecorder>,
        metrics: Option<crate::observability::Metrics>,
    ) {
        self.lease_recorder.install(recorder, metrics);
    }
}

impl ActivityCompletionSink for PendingActivities {
    /// Park one in-flight dispatch for restart recovery (#207): resolve the
    /// matched waiter with the ephemeral parked sentinel, and nothing else.
    ///
    /// This resolution is MANDATORY, not an optimization: the default
    /// `ActivityDispatcher::dispatch_async` runs the dispatcher's blocking
    /// `std::sync::mpsc::recv()` on tokio's blocking pool, and tokio `Runtime`
    /// drop joins blocking threads — an unresolved waiter would wedge process
    /// exit indefinitely. An unmatched dispatch (already resolved by another
    /// path) is a no-op: a park is NEVER routed to the outbox delivery
    /// callback, because it is not a failure and must never reach a workflow.
    fn park_activity(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
    ) -> Result<(), ServerError> {
        // A park retires the execution site, so it takes the whole current
        // generation — including a redelivery's sibling token, whose worker is
        // being parked for the same restart.
        self.completion_fences
            .revoke_current(workflow_id, activity_id)?;
        let matched = self
            .pending
            .remove(&(workflow_id.clone(), activity_id.clone()));
        if let Some((_, waiter)) = matched {
            // A send failure means the waiter side already dropped its
            // receiver (the dispatch is being cleaned up concurrently) —
            // benign: there is no thread left to unblock.
            let _ = waiter
                .sender
                .send(Err(aion::PARKED_ACTIVITY_REASON.to_owned()));
        }
        Ok(())
    }

    fn complete_activity(&self, completion: ActivityCompletion) -> Result<(), ServerError> {
        self.complete_activity_after_accept(completion, || Ok(()))
    }
}

/// Dispatcher that routes `run_activity` NIF calls to connected workers.
///
/// Synchronous interface — uses `try_send` for the task channel and
/// `std::sync::mpsc::Receiver::recv` for the response. Callers on a
/// multi-thread tokio runtime are detected and moved into
/// `tokio::task::block_in_place` so the blocking wait never starves the
/// runtime tasks that flush the worker stream (see the module docs).
pub struct WorkerActivityDispatcher {
    registry: ConnectedWorkerRegistry,
    namespace: String,
    pending: PendingActivities,
    heartbeat_tracker: HeartbeatTracker,
    drain_state: DrainState,
    tokio_handle: Option<tokio::runtime::Handle>,
    /// NOI-6 attempt→owner back-index. When installed, each liminal-delivered
    /// dispatch binds its `(workflow, activity, attempt)` to the owning worker
    /// for the dispatch's lifetime, so the intervention router (and the ops
    /// console's live-attempts enumeration) can see and target it. `None`
    /// (isolated tests) binds nothing.
    attempt_owners: Option<super::intervention::AttemptOwnerIndex>,
    /// R1 service policies and the two service clocks. Default: `strict` with
    /// no clocks — refuse the structurally unservable, wait (loudly) for
    /// everything else, and invent no deadline the operator never wrote.
    queue_service: QueueServiceConfig,
    /// Deployed queue declarations, installed once the engine exists. Answers
    /// `Unknown` until then, and `Unknown` never refuses anything.
    queue_declarations: QueueDeclarationSource,
    /// Live unserved-queue state a parked dispatch publishes itself into.
    queue_state: QueueServiceState,
    /// Cluster-event publisher an unbounded park announces itself on (#266
    /// T4). `None` (isolated tests) loses only the pushed echo; the WARN and
    /// the queryable state above cannot be switched off by wiring.
    cluster_publisher: Option<crate::cluster_publisher::ClusterEventPublisher>,
}

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

impl WorkerActivityDispatcher {
    /// Build a dispatcher for the given namespace, worker registry, and
    /// liveness tracker.
    ///
    /// The tracker must be the same instance the worker stream handler and
    /// shutdown coordinator share: the unbounded completion wait relies on
    /// stream teardown sweeping this tracker's in-flight entries to fail
    /// dispatches whose worker was lost.
    #[must_use]
    pub fn new(
        registry: ConnectedWorkerRegistry,
        namespace: impl Into<String>,
        heartbeat_tracker: HeartbeatTracker,
    ) -> Self {
        Self {
            registry,
            namespace: namespace.into(),
            // Derived from the tracker this constructor is already handed,
            // which already carries the operator's window: the dispatcher
            // cannot be given a heartbeat window and a mismatched budget.
            pending: PendingActivities::new(heartbeat_tracker.heartbeat_window()),
            heartbeat_tracker,
            drain_state: DrainState::default(),
            tokio_handle: None,
            attempt_owners: None,
            queue_service: QueueServiceConfig::default(),
            queue_declarations: QueueDeclarationSource::default(),
            queue_state: QueueServiceState::default(),
            cluster_publisher: None,
        }
    }

    /// Share the deployment-global cluster-event publisher so a dispatch
    /// parked with no availability deadline is announced on the operator's
    /// real-time channel, not only in the log (#266 T4).
    #[must_use]
    pub fn with_cluster_publisher(
        mut self,
        cluster_publisher: crate::cluster_publisher::ClusterEventPublisher,
    ) -> Self {
        self.cluster_publisher = Some(cluster_publisher);
        self
    }

    /// Share the operator's R1 queue-service settings: the default policy, the
    /// written per-queue `durable_pending` opt-ins, and the two clocks.
    #[must_use]
    pub fn with_queue_service(mut self, queue_service: QueueServiceConfig) -> Self {
        self.queue_service = queue_service;
        self
    }

    /// Share the queue-declaration source the boot path fills in once the
    /// engine exists (the handle is cloneable; installing on any clone is
    /// visible here).
    #[must_use]
    pub fn with_queue_declarations(mut self, queue_declarations: QueueDeclarationSource) -> Self {
        self.queue_declarations = queue_declarations;
        self
    }

    /// Share the queue-service state so the server can read which addresses are
    /// unserved and which runs are parked on them.
    #[must_use]
    pub fn with_queue_state(mut self, queue_state: QueueServiceState) -> Self {
        self.queue_state = queue_state;
        self
    }

    /// Share the server's NOI-6 attempt→owner back-index so liminal-delivered
    /// dispatches are visible (and targetable) to the intervention router for
    /// exactly as long as they are in flight. The production boot passes
    /// `ServerState`'s index — the SAME instance `intervenable_attempts` and
    /// `intervene` read — or the console's live-attempts list stays empty for
    /// every bridge-dispatched agent step.
    #[must_use]
    pub fn with_attempt_owners(
        mut self,
        attempt_owners: super::intervention::AttemptOwnerIndex,
    ) -> Self {
        self.attempt_owners = Some(attempt_owners);
        self
    }

    /// The completion sink this dispatcher classifies worker deaths through.
    ///
    /// Exposed so the transport-loss budget a wiring actually ends up with is
    /// observable, rather than being a property only prose asserts.
    #[must_use]
    pub const fn pending(&self) -> &PendingActivities {
        &self.pending
    }

    /// Share a caller-supplied pending-activities tracker.
    #[must_use]
    pub fn with_pending(mut self, pending: PendingActivities) -> Self {
        self.pending = pending;
        self
    }

    /// Share the server drain gate.
    #[must_use]
    pub fn with_drain_state(mut self, drain_state: DrainState) -> Self {
        self.drain_state = drain_state;
        self
    }

    /// Share the server runtime handle for sync history writes from dirty NIF threads.
    #[must_use]
    pub fn with_tokio_handle(mut self, tokio_handle: tokio::runtime::Handle) -> Self {
        self.tokio_handle = Some(tokio_handle);
        self
    }
}

impl WorkerActivityDispatcher {
    /// The drain gate for one dispatch, and the single place a draining server's
    /// refusal is turned into a reason the engine acts on.
    ///
    /// The reason handed back is the #207 parked sentinel
    /// ([`aion::PARKED_ACTIVITY_REASON`]), NOT the gate's own message. A
    /// dispatch the drain refused never reached a worker and never ran, which is
    /// precisely the state that sentinel was minted for: record nothing, deliver
    /// nothing, and let post-restart replay re-dispatch the dangling ordinal —
    /// the kill -9 convergence `shutdown::ShutdownOutcome::Parked` documents.
    ///
    /// Both alternatives are wrong here, and neither is a style preference:
    ///
    /// - the *unprefixed* gate message this used to return is classified as a
    ///   terminal action failure, so a routine deploy records `ActivityFailed`
    ///   (and, with no authored retry, `WorkflowFailed`) for work nobody
    ///   attempted — the synthesized failure #207 exists to abolish;
    /// - the TRANSPORT-loss class (`lost:`) is attempt-neutral but re-dispatches
    ///   the SAME attempt live and in-process, which on a draining server means
    ///   refuse → re-dispatch → refuse, spinning against the very gate that
    ///   refused it while the process is trying to exit.
    ///
    /// The operator-facing detail is not lost: the gate's own message is what is
    /// logged here. Only the engine-facing classification is the sentinel.
    fn ensure_accepting(
        &self,
        namespace: &str,
        activity_type: &str,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        worker_id: Option<WorkerId>,
    ) -> Result<(), String> {
        self.drain_state
            .ensure_accepting(namespace, activity_type)
            .map_err(|error| {
                log_worker_error(
                    "WorkerDispatch",
                    namespace,
                    activity_type,
                    workflow_id,
                    activity_id,
                    worker_id,
                    &error.to_string(),
                );
                aion::PARKED_ACTIVITY_REASON.to_owned()
            })
    }

    /// Select a worker for the address, or refuse with a typed R1 reason.
    ///
    /// The wait itself lives in [`super::queue_service::wait`]: every selection
    /// miss is classified against the deployed contract records and the live
    /// poller census, published to the queue-service state, and stated at WARN.
    /// This method owns only the two things the seam cannot delegate — the
    /// drain gate and the runtime plumbing of one bounded park.
    ///
    /// The [`DispatchReservation`] returned beside the worker is the chosen
    /// worker's capacity slot, already claimed. The caller must hold it until
    /// the heartbeat tracker has taken the count over; dropping it early
    /// re-opens the window in which a simultaneous dispatch re-picks a worker
    /// this one has already filled.
    fn select_worker_or_wait(
        &self,
        address: &ServiceAddress,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
    ) -> Result<(WorkerHandle, DispatchReservation), String> {
        let wait = ServiceWait {
            registry: &self.registry,
            declarations: &self.queue_declarations,
            config: &self.queue_service,
            state: &self.queue_state,
            address,
            workflow_id,
            activity_id,
            publisher: self.cluster_publisher.as_ref(),
        };
        let mut accepting = || {
            self.ensure_accepting(
                &address.namespace,
                &address.activity_type,
                workflow_id,
                activity_id,
                None,
            )
        };
        let mut park = |arrival: WorkerArrival, budget: Option<Duration>| {
            self.park_for_worker(arrival, budget);
        };
        select_worker_or_refuse(&wait, &mut accepting, &mut park).map_err(|refusal| {
            let reason = refusal.reason_string();
            // The drain gate already logged its own refusal at its own site;
            // logging it twice would double-count the incident.
            if !matches!(refusal, SelectionRefusal::NotAccepting { .. }) {
                let error_type = match refusal {
                    SelectionRefusal::Unavailable(_) => "WorkerUnavailable",
                    _ => "WorkerRegistry",
                };
                log_worker_error(
                    error_type,
                    &address.namespace,
                    &address.activity_type,
                    workflow_id,
                    activity_id,
                    None,
                    &reason,
                );
            }
            reason
        })
    }

    /// Wait once on the `arrival` subscription — bounded by `budget` when one is
    /// given, and bounded in BOTH arms by the server's own shutdown.
    ///
    /// `arrival` was taken by [`select_worker_or_refuse`] BEFORE the selection
    /// that missed, which is why this method is handed one rather than taking
    /// its own. Subscribing here would be too late: the registry's wake sources
    /// store no permit, so a worker that registered during that selection would
    /// already have fired into an empty waiter list and this park would sleep
    /// through it.
    ///
    /// The arrival wait races the drain latch, and that race is the whole point.
    /// Without it a `None` budget — the default, because no schedule-to-start
    /// deadline is configured unless an operator writes one — parked this
    /// dispatch on a `spawn_blocking` thread with nothing left that could ever
    /// wake it: the drain gate is consulted once per selection iteration, and
    /// the iteration was stuck inside this wait. Tokio's runtime `Drop` joins
    /// the blocking pool, so `aion server` could not exit even after `main`
    /// returned an exit code — it kept the store's writer lock and blocked the
    /// next deploy into the same data directory. The captured stack of a real
    /// one is `docs/evidence/samples/aion-server-hang-after-drain-15202.sample`.
    /// The `Some(budget)` arm carried the same defect at the scale of the
    /// budget: exit was delayed by up to the whole availability deadline.
    ///
    /// Racing the latch is the fix rather than capping the wait, because the
    /// wait is legitimately unbounded while the server runs: a queue no worker
    /// serves yet is a fleet condition an operator resolves by starting a
    /// worker, and a deadline invented here would refuse work nobody asked to
    /// have refused.
    ///
    /// Waking on the latch decides nothing by itself. The selection loop
    /// re-selects, misses, and consults the SAME drain gate it always
    /// consulted — [`Self::ensure_accepting`] — which is where the refusal and
    /// its classification are produced. This seam gains no second opinion about
    /// draining; it only stops being unwakeable.
    fn park_for_worker(&self, arrival: WorkerArrival, budget: Option<Duration>) {
        let handle = self
            .tokio_handle
            .clone()
            .or_else(|| tokio::runtime::Handle::try_current().ok());
        let Some(handle) = handle else {
            // No runtime in reach (a plain OS thread in an isolated test):
            // sleep the seam's park interval, clamped to the remaining budget.
            // Bounded by construction, so the loop re-consults the drain gate
            // within one interval and needs no signal to race.
            //
            // `arrival` is DROPPED unawaited here, deliberately: there is no
            // executor to await it on. That is safe only because this arm is
            // bounded by construction — the loop re-selects within one interval
            // whether or not any wake ever fires — and it is the one place the
            // `park` contract permits discarding the subscription. Its `Drop`
            // unlinks the waiter it enabled.
            std::thread::sleep(
                budget.map_or(PARK_POLL_INTERVAL, |budget| budget.min(PARK_POLL_INTERVAL)),
            );
            return;
        };
        handle.block_on(async {
            let raced = async {
                tokio::select! {
                    () = arrival => {}
                    () = self.drain_state.wait_for_drain() => {}
                }
            };
            match budget {
                None => raced.await,
                Some(budget) => {
                    // Timeout is not failure here: the caller re-classifies and
                    // decides whether the clock has run out.
                    drop(tokio::time::timeout(budget, raced).await);
                }
            }
        });
    }

    fn track_worker_task(
        &self,
        worker_id: WorkerId,
        activity_type: &str,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        attempt: u32,
        // The token this dispatch is fenced by and the capacity slot it holds,
        // together because they are handed over together: the tracker commits
        // the slot in the same call that records the token.
        lease: (CompletionToken, DispatchReservation),
    ) -> Result<(), String> {
        let (completion_token, reservation) = lease;
        self.heartbeat_tracker
            .track_task(
                worker_id,
                InFlightActivity {
                    workflow_id: workflow_id.clone(),
                    activity_id: activity_id.clone(),
                    // The engine-provided attempt this delivery carries — the
                    // same one stamped on the wire task and bound into the
                    // NOI-6 attempt→owner index, so the tracker, the wire, and
                    // the owner index all name one attempt.
                    attempt,
                    completion_token,
                },
                Instant::now(),
                &self.registry,
                // HANDED OVER in one act: the tracker commits this slot and
                // skips its own increment, so the count never reads higher than
                // the work this worker holds. Two acts — track, then release the
                // reservation — left a window in which a concurrent selection
                // was refused a slot that was free.
                Some(reservation),
            )
            .map_err(|error| {
                let reason = error.to_string();
                log_worker_error(
                    "WorkerHeartbeatTracker",
                    &self.namespace,
                    activity_type,
                    workflow_id,
                    activity_id,
                    Some(worker_id),
                    &reason,
                );
                reason
            })
    }

    fn cleanup_activity(
        &self,
        worker_id: WorkerId,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        attempt: u32,
        completion_token: &CompletionToken,
    ) {
        // Own-attempt-only, like the token revocation below: a dispatch may
        // tear down only the waiter slot it installed. After a higher attempt
        // takes the site over (aion#195) this entry is no longer this
        // dispatch's to remove.
        self.pending
            .pending
            .remove_if(&(workflow_id.clone(), activity_id.clone()), |_, waiter| {
                waiter.attempt == attempt
            });
        // Own-token-only: an undelivered task withdraws the authorization it
        // was given and never the sibling a concurrently redelivered dispatch
        // handed to a worker that is still executing.
        if let Err(error) =
            self.pending
                .completion_fences
                .revoke(workflow_id, activity_id, completion_token)
        {
            tracing::error!(
                workflow_id = %workflow_id,
                activity_id = %activity_id,
                %error,
                "failed to revoke undelivered activity generation"
            );
        }
        let _ = self.heartbeat_tracker.complete_task(
            worker_id,
            workflow_id,
            activity_id,
            &self.registry,
        );
        self.drain_state.notify_activity_drained();
    }

    /// Deliver one dispatched task to the selected worker over ITS transport.
    ///
    /// The bridge is transport-agnostic at this seam: selection
    /// ([`Self::select_worker_or_wait`]) already treats every registry member
    /// identically, and this match delivers on whichever [`WorkerDelivery`] leg
    /// the worker registered with — the gRPC stream `mpsc` push, or the liminal
    /// server-push on the worker's existing connection. Both legs resolve
    /// through the SAME pending map, so `await_activity_result` is oblivious to
    /// the transport.
    fn send_activity_task(
        &self,
        worker: &WorkerHandle,
        task: ProtoActivityTask,
        address: &ServiceAddress,
        handoff: &super::lease_record::LeaseHandoff,
    ) -> Result<(), String> {
        let workflow_id = &handoff.key().workflow_id;
        let activity_id = &handoff.key().activity_id;
        let completion_token = handoff.token();
        match worker.delivery() {
            WorkerDelivery::Grpc(sender) => {
                let worker_id = worker.id();
                let mut accepting = || {
                    self.ensure_accepting(
                        &address.namespace,
                        &address.activity_type,
                        workflow_id,
                        activity_id,
                        Some(worker_id),
                    )
                };
                let handed_over = deliver_within_schedule_to_start(
                    sender,
                    WorkerMessage::ActivityTask(Box::new(task)),
                    self.queue_service.schedule_to_start_timeout,
                    &mut accepting,
                );
                let Err(refusal) = handed_over else {
                    // The stream took the frame: the worker holds the attempt
                    // from here, and the lease is recorded before this call
                    // returns to wait on the result (WA-010 R3).
                    handoff.accepted_blocking(self.tokio_handle.as_ref());
                    return Ok(());
                };
                self.cleanup_activity(
                    worker_id,
                    workflow_id,
                    activity_id,
                    handoff.key().attempt,
                    completion_token,
                );
                let (error_type, reason) = self.hand_off_failure(&refusal, address);
                if !matches!(refusal, DeliveryRefusal::NotAccepting { .. }) {
                    log_worker_error(
                        error_type,
                        &address.namespace,
                        &address.activity_type,
                        workflow_id,
                        activity_id,
                        Some(worker_id),
                        &reason,
                    );
                }
                Err(reason)
            }
            // The liminal leg has no bounded intake queue to saturate: the push
            // either enqueues on the worker's live connection or fails
            // synchronously because that connection is already gone. There is
            // therefore no schedule-to-start window to apply here, and reporting
            // `SATURATED` for a push failure would name a condition that did not
            // happen.
            #[cfg(feature = "liminal-transport")]
            WorkerDelivery::Liminal(delivery) => self.send_liminal_activity_task(
                worker.id(),
                delivery,
                task,
                &address.activity_type,
                handoff,
            ),
        }
    }

    /// Render one hand-off refusal into its log class and failure reason.
    ///
    /// `SATURATED` is the only one that becomes a typed [`WorkerUnavailable`]:
    /// a compatible worker WAS live and would not take the task inside the
    /// schedule-to-start clock. A full intake with no clock configured, and a
    /// closed transport, keep the failure they always had.
    fn hand_off_failure(
        &self,
        refusal: &DeliveryRefusal,
        address: &ServiceAddress,
    ) -> (&'static str, String) {
        match refusal {
            DeliveryRefusal::Saturated { waited } => {
                // A census failure must not swallow the refusal: report the
                // saturation with an empty census and say why it is empty.
                let census = self
                    .registry
                    .pool_census(
                        &address.namespace,
                        &address.task_queue,
                        &address.activity_type,
                        address.node.as_deref(),
                    )
                    .unwrap_or_else(|error| {
                        tracing::error!(
                            namespace = %address.namespace,
                            task_queue = %address.task_queue,
                            activity_type = %address.activity_type,
                            %error,
                            "poller census failed while reporting a saturated queue; \
                             the refusal carries an empty census"
                        );
                        PoolCensus::default()
                    });
                let unavailable = WorkerUnavailable {
                    reason: QueueServiceReason::Saturated,
                    clock: Some(ExpiredClock::ScheduleToStart),
                    waited: *waited,
                    address: address.clone(),
                    census,
                };
                ("WorkerUnavailable", unavailable.reason_string())
            }
            DeliveryRefusal::Full => (
                "WorkerChannelClosed",
                "worker task channel full or closed: no available capacity".to_owned(),
            ),
            DeliveryRefusal::Closed => (
                "WorkerChannelClosed",
                "worker task channel full or closed: channel closed".to_owned(),
            ),
            DeliveryRefusal::NotAccepting { reason } => ("WorkerDispatch", reason.clone()),
        }
    }

    /// Deliver one dispatched task to a liminal-connected worker: push the SAME
    /// wire frame the outbox liminal path pushes (a
    /// [`DispatchRequest`](super::liminal_transport::DispatchRequest) — the
    /// worker's serve loop cannot tell a bridge dispatch from an outbox row) and
    /// hand the correlated-reply awaiter to a dedicated router thread that
    /// resolves this dispatch's pending entry exactly like a gRPC completion.
    ///
    /// The wire carries the SAME engine-provided `attempt` and `labels` the gRPC
    /// arm's `ActivityTask` carries (a retry over liminal executes with the real
    /// attempt, not a re-stamped first delivery), plus the server's heartbeat
    /// window so the worker's automatic liveness pump keeps this TRACKED
    /// dispatch alive under the #176 expiry sweeper. It carries the engine's
    /// concrete run identity, generation proof, and stable effect key exactly
    /// like the gRPC bridge task.
    ///
    /// A successful push also binds the attempt into the NOI-6 attempt→owner
    /// back-index (when installed) with the SAME `(workflow, activity, attempt)`
    /// key the worker stamps its intervention session with, exactly as the
    /// outbox liminal arm binds each row dispatch — so the ops console can
    /// enumerate this live attempt and route interventions to its worker. The
    /// binding is released when the reply router exits (reply, abandonment, or
    /// disconnect — every path). The gRPC arm carries no bind because the agent
    /// harness seam exists only on the liminal worker transport.
    #[cfg(feature = "liminal-transport")]
    fn send_liminal_activity_task(
        &self,
        worker_id: WorkerId,
        delivery: &super::liminal_transport::LiminalWorkerDelivery,
        task: ProtoActivityTask,
        activity_type: &str,
        handoff: &super::lease_record::LeaseHandoff,
    ) -> Result<(), String> {
        let workflow_id = &handoff.key().workflow_id;
        let activity_id = &handoff.key().activity_id;
        let completion_token =
            CompletionToken::from_wire(workflow_id, activity_id, task.completion_token.clone())
                .map_err(|error| error.to_string())?;
        let heartbeat_window_ms =
            u64::try_from(self.heartbeat_tracker.heartbeat_window().as_millis())
                .unwrap_or(u64::MAX);
        let attempt = task.attempt;
        // The run is resolved BEFORE the dispatch is built, because everything
        // downstream is keyed on it: the worker refuses a run-less dispatch
        // envelope outright, the attempt-owner binding below needs it to name a
        // generation, and the attempt's transcript is written under it. Refusing
        // here reports the fault against the dispatch that has it, rather than
        // shipping a frame whose only possible outcome is the worker's rejection.
        let run_id = task
            .run_id
            .map(RunId::try_from)
            .transpose()
            .map_err(|error| error.to_string())?
            .ok_or_else(|| {
                "activity task run id is missing; refusing to dispatch an unidentified run"
                    .to_owned()
            })?;
        let request = super::liminal_transport::DispatchRequest {
            activity_type: activity_type.to_owned(),
            workflow_id: workflow_id.clone(),
            ordinal: activity_id.sequence_position(),
            run_id: Some(run_id.clone()),
            attempt,
            completion_token: task.completion_token,
            idempotency_key: task.idempotency_key,
            labels: task.labels.into_iter().collect(),
            heartbeat_window_ms,
            input: task.input.map(|payload| payload.bytes).unwrap_or_default(),
        };
        // A push-enqueue failure means the worker's connection was already gone
        // at push time — the same synchronous-failure contract as a closed gRPC
        // stream channel above.
        let awaiter = match delivery.push_dispatch(&request) {
            Ok(awaiter) => awaiter,
            Err(error) => {
                let reason = format!("worker liminal push failed: {error}");
                self.cleanup_activity(
                    worker_id,
                    workflow_id,
                    activity_id,
                    attempt,
                    &completion_token,
                );
                log_worker_error(
                    "WorkerChannelClosed",
                    &self.namespace,
                    activity_type,
                    workflow_id,
                    activity_id,
                    Some(worker_id),
                    &reason,
                );
                return Err(reason);
            }
        };
        // NOI-6: the attempt is live on `worker_id` from this push until the
        // router resolves it — bind it for exactly that window (the guard is
        // dropped when the router thread exits).
        // The push was acknowledged: the worker holds the attempt, and the
        // lease is recorded before the reply router is even spawned (WA-010
        // R3) — the reply IS the completion, so nothing may wait for it first.
        handoff.accepted_blocking(self.tokio_handle.as_ref());
        let owner_binding = self.attempt_owners.as_ref().map(|owners| {
            super::liminal_transport::AttemptOwnerGuard::bind(
                owners.clone(),
                super::intervention::AttemptKey::new(
                    workflow_id.clone(),
                    run_id.clone(),
                    activity_id.clone(),
                    attempt,
                ),
                worker_id,
            )
        });
        self.spawn_liminal_reply_router(
            worker_id,
            awaiter,
            workflow_id,
            activity_id,
            &completion_token,
            owner_binding,
        );
        Ok(())
    }

    /// Waits (on a dedicated router thread, bounded by the dispatch's own
    /// lifetime) for the worker's correlated
    /// [`DispatchResponse`](super::liminal_transport::DispatchResponse) and
    /// re-enters it through the SAME completion bookkeeping the gRPC inbound
    /// stream applies (`process_inbound` in `worker_grpc.rs`): clear the
    /// in-flight liveness entry, wake any drain waiter, then resolve the
    /// bridge's pending map — result, failure, and retryable classification
    /// identical (the worker encodes the `retryable:`/`terminal:` reason
    /// vocabulary on the wire). An unmatched (already-resolved) REAL reply
    /// routes through the outbox delivery callback exactly like a late gRPC
    /// result.
    ///
    /// The #176 heartbeat sweeper covers this dispatch exactly as it covers a
    /// gRPC one: the dispatch is tracked in the shared [`HeartbeatTracker`] and
    /// the worker's runtime pumps automatic liveness beats over the reserved
    /// liminal channel (`WORKER_LIVENESS_CHANNEL`), so a healthy worker running
    /// an over-window activity is never falsely expired while a wedged one
    /// still is. Prompt worker-DEATH detection additionally rides the
    /// connection itself — the awaiter wakes with the typed Disconnected error
    /// the moment the connection closes, resolving the SAME retryable
    /// lost-worker failure the gRPC stream-teardown sweep reports.
    ///
    /// Two structural guards mirror the gRPC arm's tracker gating:
    ///
    /// - A SYNTHESIZED failure (disconnect / receive fault) is delivered only
    ///   when this router's own `complete_task` actually retired the tracked
    ///   entry — the same "fail only still-tracked tasks" gate
    ///   `remove_worker_tasks` gives the gRPC sweeps — so a dispatch already
    ///   resolved elsewhere (expiry sweep, shutdown drain, deregistered
    ///   fast path) never has a spurious failure injected for an ordinal whose
    ///   retry may be live on another worker.
    /// - The wait itself ends one reply-poll after the tracked entry
    ///   disappears, so an abandoned dispatch never parks this thread for the
    ///   remaining life of the worker's connection. A real reply arriving
    ///   AFTER that exit is dropped (the resolving path owns the ordinal — its
    ///   retry re-executes); this is the one deliberate divergence from the
    ///   gRPC arm, whose shared stream task routes any late result to the
    ///   outbox callback, and it is the safer half of the trade because a
    ///   stale attempt's result can never resolve a newer attempt's entry.
    #[cfg(feature = "liminal-transport")]
    fn spawn_liminal_reply_router(
        &self,
        worker_id: WorkerId,
        awaiter: liminal_server::server::connection::PushReplyAwaiter,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        completion_token: &CompletionToken,
        owner_binding: Option<super::liminal_transport::AttemptOwnerGuard>,
    ) {
        let pending = self.pending.clone();
        let heartbeat_tracker = self.heartbeat_tracker.clone();
        let registry = self.registry.clone();
        let drain_state = self.drain_state.clone();
        let workflow_id = workflow_id.clone();
        let activity_id = activity_id.clone();
        let completion_token = completion_token.clone();
        std::thread::spawn(move || {
            // Owns the NOI-6 attempt binding for the dispatch's lifetime: it
            // drops (releasing the back-index entry) when this router exits,
            // on every path — reply, abandonment, disconnect, or panic.
            let _owner_binding = owner_binding;
            route_liminal_reply(
                &pending,
                &heartbeat_tracker,
                &registry,
                &drain_state,
                &awaiter,
                (worker_id, &workflow_id, &activity_id, &completion_token),
            );
        });
    }

    /// Block until the dispatch terminates (see the module docs for the
    /// exhaustive termination list). The wait is deliberately unbounded:
    /// the engine imposes no activity timeout of its own.
    fn await_activity_result(
        &self,
        context: &ActivityDispatchContext<'_>,
        rx: &SyncReceiver,
    ) -> Result<String, String> {
        // Close the dispatch/disconnect race before blocking. A worker whose
        // stream tore down *before* this dispatch tracked its task was swept
        // without this entry, so nothing would ever deliver through `rx`.
        // `fail_lost_worker` deregisters before it collects tasks, and this
        // dispatch tracked its task before sending, so: if the worker is
        // still registered here, any later sweep is guaranteed to include
        // this task and unblock the `recv` below.
        match self.registry.is_registered(context.worker_id) {
            Ok(true) => {}
            Ok(false) => {
                // A sweep that did include this task may have delivered
                // already; prefer its verdict (or a genuine result that
                // raced the disconnect) over fabricating one.
                if let Ok(result) = rx.try_recv() {
                    return self.deliver_result(context, result);
                }
                self.cleanup_activity(
                    context.worker_id,
                    context.workflow_id,
                    context.activity_id,
                    context.attempt,
                    &context.completion_token,
                );
                // TRANSPORT domain, classified through the shared ledger: the
                // activity never executed, so the failure never wears the
                // action's retry vocabulary and is re-dispatched
                // attempt-neutrally until the transport's own budget is spent.
                let reason = self.pending.classify_worker_loss(
                    context.workflow_id,
                    context.activity_id,
                    context.worker_id,
                );
                log_worker_error(
                    "WorkerLost",
                    &self.namespace,
                    context.activity_type,
                    context.workflow_id,
                    context.activity_id,
                    Some(context.worker_id),
                    &reason,
                );
                return Err(reason);
            }
            Err(error) => {
                self.cleanup_activity(
                    context.worker_id,
                    context.workflow_id,
                    context.activity_id,
                    context.attempt,
                    &context.completion_token,
                );
                let reason = format!("worker registry inspection failed: {error}");
                log_worker_error(
                    "WorkerRegistry",
                    &self.namespace,
                    context.activity_type,
                    context.workflow_id,
                    context.activity_id,
                    Some(context.worker_id),
                    &reason,
                );
                return Err(reason);
            }
        }
        if let Ok(result) = rx.recv() {
            return self.deliver_result(context, result);
        }
        // Every sender was dropped without completing: a cleanup path
        // removed the pending entry. Surface it instead of hanging.
        self.cleanup_activity(
            context.worker_id,
            context.workflow_id,
            context.activity_id,
            context.attempt,
            &context.completion_token,
        );
        let reason = "activity response channel dropped".to_owned();
        log_worker_error(
            "WorkerChannelClosed",
            &self.namespace,
            context.activity_type,
            context.workflow_id,
            context.activity_id,
            Some(context.worker_id),
            &reason,
        );
        Err(reason)
    }

    fn deliver_result(
        &self,
        context: &ActivityDispatchContext<'_>,
        result: Result<String, String>,
    ) -> Result<String, String> {
        // Own-attempt-only (aion#195): after a higher attempt takes this site
        // over, the superseded dispatch resolving here must not evict the live
        // waiter that replaced it.
        self.pending.pending.remove_if(
            &(context.workflow_id.clone(), context.activity_id.clone()),
            |_, waiter| waiter.attempt == context.attempt,
        );
        // A parked dispatch (#207) is not a failure: the server is draining and
        // restart recovery re-dispatches the ordinal. Info, never error — a
        // routine deploy must not emit an ActivityFailed log per in-flight
        // dispatch (the incident's alarm noise).
        if let Err(reason) = &result
            && aion::is_parked_reason(reason)
        {
            tracing::info!(
                operation = "activity_dispatch",
                namespace = %self.namespace,
                workflow_id = %context.workflow_id,
                activity_id = %context.activity_id,
                activity_type = context.activity_type,
                worker_id = ?context.worker_id,
                "activity parked for restart recovery"
            );
            return result;
        }
        log_activity_completion(context, result.is_ok());
        result.inspect_err(|reason| {
            log_worker_error(
                "ActivityFailed",
                &self.namespace,
                context.activity_type,
                context.workflow_id,
                context.activity_id,
                Some(context.worker_id),
                reason,
            );
        })
    }
}

impl ActivityDispatcher for WorkerActivityDispatcher {
    fn dispatch(&self, request: ActivityDispatch) -> Result<String, String> {
        match tokio::runtime::Handle::try_current() {
            Ok(handle) => match handle.runtime_flavor() {
                tokio::runtime::RuntimeFlavor::MultiThread => {
                    // We are inside a tokio runtime (the engine spawns the
                    // sync dispatch onto its handle). Hand this worker's
                    // scheduler core to another thread before blocking so the
                    // stream forwarder woken by our `try_send` can actually
                    // run — otherwise it is trapped in this worker's
                    // non-stealable LIFO slot for as long as we block.
                    tokio::task::block_in_place(|| {
                        self.dispatch_blocking(request, &LeaseSignal::none())
                    })
                }
                flavor => Err(format!(
                    "activity dispatch blocks the calling thread until the worker responds; \
                     a {flavor:?} tokio runtime cannot host that wait because the worker \
                     stream forwarder shares its only executor thread and the task could \
                     never be delivered — run the engine on a multi-thread tokio runtime"
                )),
            },
            // No tokio context: a beamr scheduler thread or other plain OS
            // thread. Blocking here is the designed contract and cannot starve
            // the server runtime.
            Err(_) => self.dispatch_blocking(request, &LeaseSignal::none()),
        }
    }

    /// The awaited entry, overridden so the LEASE reaches the engine's
    /// per-attempt timer.
    ///
    /// The default runs the synchronous entry above and fires the signal
    /// immediately, which is right for a dispatcher that has no lease step.
    /// THIS dispatcher has one, and it is separated from the call by an
    /// unbounded park: `dispatch_blocking` waits for a worker to exist before
    /// anything can be handed over. Firing at the call would put that park
    /// inside a bound written to describe execution — the defect — so the
    /// signal travels down to the handover instead and fires at the same
    /// instant the durable `ActivityLeased` record is made.
    ///
    /// `spawn_blocking` rather than the default's `dispatch` hop: the blocking
    /// pool is exactly the "no scheduler core to give up" condition the
    /// `block_in_place` branch above exists to manufacture, so going through it
    /// again would be a hop that arranges what is already true.
    fn dispatch_async(
        self: std::sync::Arc<Self>,
        request: ActivityDispatch,
        lease: LeaseSignal,
    ) -> futures::future::BoxFuture<'static, Result<String, String>> {
        Box::pin(async move {
            let blocking =
                tokio::task::spawn_blocking(move || self.dispatch_blocking(request, &lease));
            match blocking.await {
                Ok(result) => result,
                Err(join_error) => Err(format!("activity dispatch task failed: {join_error}")),
            }
        })
    }
}

impl WorkerActivityDispatcher {
    /// Dispatch the activity and block the calling thread until the worker
    /// responds, the worker is declared lost, or the server drains (see the
    /// module docs for the exhaustive termination list).
    ///
    /// The request carries the *real* workflow and activity ids the engine
    /// recorded in history, so the worker logs, the pending-completion key,
    /// and the heartbeat tracker all correlate directly against the event
    /// store. `config` is forwarded by the engine seam but not yet consumed
    /// here (the retry executor that reads it is unbuilt).
    ///
    /// Must never run while the calling thread still owns a tokio scheduler
    /// core: the response can only arrive after the runtime's stream
    /// forwarder flushes the queued [`WorkerMessage::ActivityTask`] to the
    /// worker, so the thread blocking here must not be the one responsible
    /// for polling that forwarder. [`ActivityDispatcher::dispatch`] enforces
    /// this with `tokio::task::block_in_place`.
    fn dispatch_blocking(
        &self,
        request: ActivityDispatch,
        lease: &LeaseSignal,
    ) -> Result<String, String> {
        let ActivityDispatch {
            namespace,
            task_queue,
            // OPTIONAL within-pool node affinity (NODE-4): `Some(n)` pins this
            // dispatch to workers advertising node `n` (require semantics);
            // `None` is unpinned and reaches any worker in the pool.
            node,
            workflow_id,
            run_id,
            activity_id,
            name,
            input,
            config: _,
            attempt,
            labels,
            // R5 advisory class: engine-side only. The wire carries no such
            // field and a worker's behaviour is identical either way — the
            // class governs how the ENGINE treats the failure, never how the
            // work is done.
            advisory: _,
        } = request;
        let started_at = Instant::now();
        self.ensure_accepting(&namespace, &name, &workflow_id, &activity_id, None)?;
        let address = ServiceAddress {
            namespace: namespace.clone(),
            task_queue: task_queue.clone(),
            activity_type: name.clone(),
            node: node.clone(),
        };
        // `reservation` holds this worker's capacity slot from the moment it was
        // chosen. It is dropped only after `track_then_send` below has handed
        // the count to the heartbeat tracker; every `?` between here and there
        // releases it on the way out, which is why none of those arms mention
        // capacity.
        let (worker, reservation) =
            self.select_worker_or_wait(&address, &workflow_id, &activity_id)?;
        let worker_id = worker.id();
        let span = info_span!(
            "activity_dispatch",
            operation = "activity_dispatch",
            namespace = %namespace,
            task_queue = %task_queue,
            node = node.as_deref(),
            workflow_id = %workflow_id,
            activity_id = %activity_id,
            activity_type = %name,
            worker_id = ?worker_id,
        );
        let _span_guard = span.enter();
        self.ensure_accepting(
            &namespace,
            &name,
            &workflow_id,
            &activity_id,
            Some(worker_id),
        )?;

        let (completion_token, rx, superseded_attempt) = self
            .pending
            .insert(workflow_id.clone(), &run_id, activity_id.clone(), attempt)
            .map_err(|error| error.to_string())?;
        if let Some(held_attempt) = superseded_attempt {
            // aion#195: this retry took the site over from a timed-out
            // attempt whose responder is still live. Walk the release ladder
            // for the corpse BEFORE this attempt's work begins: ask its
            // worker to stop, and untrack its heartbeat task so the expiry
            // sweep never presents its stale token mid-sweep.
            self.release_superseded_attempt(&workflow_id, &activity_id, held_attempt, attempt);
        }
        // The lease this delivery's acceptance records (WA-010 R3), armed on
        // the fences before the push; settled by the handoff whether the push
        // lands or not, so a completion never waits on a lease that will not
        // come.
        let handoff = super::lease_record::LeaseHandoff::arm(
            self.pending.lease_recorder(),
            super::lease_record::LeaseKey {
                workflow_id: workflow_id.clone(),
                run_id: run_id.clone(),
                activity_id: activity_id.clone(),
                attempt,
            },
            super::lease_record::attribution_for(&worker),
            self.pending.completion_fences(),
            completion_token.clone(),
            // The engine's per-attempt clock starts where the durable lease
            // record is made, not where this call began. Both are this
            // handoff's job, so it holds both — one anchor by construction,
            // rather than two sites that agree only while somebody keeps them
            // agreeing.
            lease.clone(),
        )
        .map_err(|error| error.to_string())?;
        let task = activity_task(
            &name,
            &input,
            (&workflow_id, &run_id, &activity_id),
            attempt,
            labels,
            &completion_token,
        );
        // The reservation goes INTO the tracking call: one act, no window in
        // which the count reads higher than the truth.
        self.track_then_send(
            (worker_id, &worker),
            task,
            &address,
            (&name, &workflow_id, &activity_id, attempt),
            (&completion_token, &handoff),
            reservation,
        )?;
        let context = ActivityDispatchContext {
            namespace: &namespace,
            activity_type: &name,
            worker_id,
            workflow_id: &workflow_id,
            activity_id: &activity_id,
            attempt,
            completion_token,
            started_at,
        };
        self.await_activity_result(&context, &rx)
    }

    /// Track this dispatch's liveness entry, then push it to its worker.
    ///
    /// The two are one step because the ORDER between them is load-bearing and
    /// the failure path between them is too: the task is tracked BEFORE the
    /// push, so a sweep that fires while the push is in flight can always see
    /// the dispatch it has to fail; and a tracking failure must undo the
    /// arming this dispatch has already done rather than leave a generation
    /// armed for work that was never sent.
    fn track_then_send(
        &self,
        worker: (WorkerId, &WorkerHandle),
        task: ProtoActivityTask,
        address: &ServiceAddress,
        site: (&str, &WorkflowId, &ActivityId, u32),
        lease: (&CompletionToken, &super::lease_record::LeaseHandoff),
        reservation: DispatchReservation,
    ) -> Result<(), String> {
        let (worker_id, handle) = worker;
        let (name, workflow_id, activity_id, attempt) = site;
        let (completion_token, handoff) = lease;
        if let Err(error) = self.track_worker_task(
            worker_id,
            name,
            workflow_id,
            activity_id,
            attempt,
            (completion_token.clone(), reservation),
        ) {
            self.cleanup_activity(
                worker_id,
                workflow_id,
                activity_id,
                attempt,
                completion_token,
            );
            return Err(error);
        }
        self.send_activity_task(handle, task, address, handoff)
    }

    /// Release the superseded attempt a takeover displaced (aion#195).
    ///
    /// Two acts, both best-effort and both reported: ask the worker still
    /// executing the corpse to stop (a request, not a guarantee — see
    /// [`super::activity_cancel`]), and untrack its heartbeat task so the
    /// expiry sweep never aborts mid-loop on the corpse's fence-refused stale
    /// token. Bookkeeping about the corpse must never fail the live retry, so
    /// every failure here lands in the log and nowhere else.
    fn release_superseded_attempt(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        held_attempt: u32,
        new_attempt: u32,
    ) {
        let in_flight = match self.heartbeat_tracker.in_flight_for_workflow(workflow_id) {
            Ok(in_flight) => in_flight,
            Err(error) => {
                tracing::error!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    held_attempt,
                    new_attempt,
                    %error,
                    "superseded-attempt release could not read the in-flight tracker; \
                     the corpse's worker was not asked to stop"
                );
                return;
            }
        };
        let holders = in_flight
            .into_iter()
            .filter(|task| &task.activity_id == activity_id && task.attempt == held_attempt);
        let mut asked_anyone = false;
        for task in holders {
            asked_anyone = true;
            let delivery = match super::activity_cancel::ask_worker_to_stop(
                &self.registry,
                task.worker_id,
                workflow_id,
                activity_id,
            ) {
                Ok(delivery) => Some(delivery),
                Err(error) => {
                    // ASKED AND FAILED IS STILL SUPERSEDED. This used to
                    // `continue`, which skipped the `complete_task` below and
                    // leaked the corpse's capacity slot permanently — the
                    // registry count only comes down on completion or
                    // deregistration, so a worker whose stop request failed
                    // silently shrank by one slot for the life of its
                    // registration, and the expiry sweep kept a dispatch nobody
                    // would ever answer.
                    //
                    // The stop request is best-effort by construction (it is a
                    // REQUEST, not a guarantee — see `activity_cancel`). Whether
                    // it reached the worker changes nothing about the fact this
                    // attempt has been superseded, so the bookkeeping proceeds
                    // either way and the failed request is reported on its own.
                    //
                    // NOT COVERED BY A TEST, and worth saying why rather than
                    // leaving it to be discovered: `ask_worker_to_stop` returns
                    // `Ok` for every reachable outcome — worker gone, transport
                    // cannot carry, channel full — and reaches `Err` only when
                    // the registry lock is POISONED. Constructing that state
                    // means panicking inside the registry's mutex, which would
                    // poison it for the rest of the process. The leak this arm
                    // fixes was therefore real but confined to a poisoned-lock
                    // path; the correction is that the bookkeeping no longer
                    // depends on the request having succeeded.
                    tracing::warn!(
                        workflow_id = %workflow_id,
                        activity_id = %activity_id,
                        held_attempt,
                        worker_id = task.worker_id.value(),
                        %error,
                        "superseded-attempt release could not ask the holding worker to stop; \
                         untracking the corpse anyway so its capacity slot is not leaked"
                    );
                    None
                }
            };
            if let Err(error) = self.heartbeat_tracker.complete_task(
                task.worker_id,
                workflow_id,
                activity_id,
                &self.registry,
            ) {
                tracing::warn!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    held_attempt,
                    worker_id = task.worker_id.value(),
                    %error,
                    "superseded-attempt release could not untrack the corpse's heartbeat task"
                );
            }
            let Some(delivery) = delivery else {
                // The stop request itself failed and was reported above. The
                // corpse is untracked, its slot is back, and there is no
                // delivery outcome to describe.
                continue;
            };
            if delivery.was_requested() {
                tracing::info!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    held_attempt,
                    new_attempt,
                    worker_id = task.worker_id.value(),
                    "attempt takeover: asked the worker still executing the superseded \
                     attempt to stop"
                );
            } else {
                tracing::warn!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    held_attempt,
                    new_attempt,
                    worker_id = task.worker_id.value(),
                    outcome = ?delivery,
                    "attempt takeover: the superseded attempt's worker could NOT be asked \
                     to stop — its work runs on unwitnessed until it next reports, and its \
                     completion will be refused as a stale generation"
                );
            }
        }
        if !asked_anyone {
            tracing::info!(
                workflow_id = %workflow_id,
                activity_id = %activity_id,
                held_attempt,
                new_attempt,
                "attempt takeover: no tracked worker holds the superseded attempt \
                 (already untracked, or its transport records no heartbeat task); \
                 its completion token is fence-refused regardless"
            );
        }
    }
}

/// One failure the reply router FABRICATED (a disconnect, an unservable
/// frame, or a receive fault) in the shape [`route_liminal_reply`] settles:
/// no run id, the token THIS push dispatched, the reason, and the
/// `synthesized` flag set so delivery is gated on the in-flight tracker.
#[cfg(feature = "liminal-transport")]
fn synthesized_failure(
    current_token: &CompletionToken,
    reason: String,
) -> (Option<RunId>, CompletionToken, Result<String, String>, bool) {
    (None, current_token.clone(), Err(reason), true)
}

/// Body of one liminal reply-router thread (see
/// [`WorkerActivityDispatcher::spawn_liminal_reply_router`] for the contract).
///
/// Resolves by the key THIS push dispatched (the awaiter is already
/// correlation-scoped to it), never by the reply's echoed ids: a buggy echo
/// must not cross executions.
#[cfg(feature = "liminal-transport")]
fn route_liminal_reply(
    pending: &PendingActivities,
    heartbeat_tracker: &HeartbeatTracker,
    registry: &crate::worker::ConnectedWorkerRegistry,
    drain_state: &DrainState,
    awaiter: &liminal_server::server::connection::PushReplyAwaiter,
    execution: (WorkerId, &WorkflowId, &ActivityId, &CompletionToken),
) {
    let (worker_id, workflow_id, activity_id, current_token) = execution;
    // The wait re-arms only while this dispatch is still tracked in-flight, so
    // a dispatch resolved elsewhere (expiry sweep, shutdown drain, cleanup)
    // releases this thread within one reply poll instead of parking it for the
    // remaining life of the worker's connection.
    let waited = super::liminal_transport::receive_bridge_reply(awaiter, || {
        heartbeat_tracker
            .is_tracked(worker_id, workflow_id, activity_id)
            .unwrap_or(false)
    });
    // `synthesized` marks a failure this router FABRICATED (disconnect or
    // receive fault) as opposed to a real worker reply: only fabricated
    // failures are gated on the tracker below.
    let (run_id, submitted_token, outcome, synthesized) = match waited {
        Ok(Some(response)) => {
            let submitted_token = match CompletionToken::from_wire(
                workflow_id,
                activity_id,
                response.completion_token,
            ) {
                Ok(token) => token,
                Err(error) => {
                    tracing::warn!(
                        worker_id = ?worker_id,
                        workflow_id = %workflow_id,
                        activity_id = %activity_id,
                        %error,
                        "liminal activity completion omitted its generation proof"
                    );
                    return;
                }
            };
            (response.run_id, submitted_token, response.outcome, false)
        }
        Ok(None) => {
            tracing::debug!(
                worker_id = ?worker_id,
                workflow_id = %workflow_id,
                activity_id = %activity_id,
                "liminal dispatch resolved by another path; abandoning reply wait"
            );
            return;
        }
        // UNSERVABLE, before the lost arm: liminal proved the frame larger than
        // the connection's whole outbound buffer. The lost arm below would
        // classify it through the transport-loss ledger and re-dispatch it
        // attempt-neutrally, forever — the 846-lease fault. The engine reads any
        // non-`retryable:` prefix as settled (`nif_activity_retry::is_retryable_reason`),
        // so this outcome records one `ActivityFailed` and stops.
        Err(error) if error.is_worker_dispatch_unservable() => {
            synthesized_failure(current_token, format!("terminal:{error}"))
        }
        // TRANSPORT domain (the worker's connection closed before it replied):
        // classified through the shared ledger, never as an action failure.
        Err(error) if error.is_worker_connection_lost() => synthesized_failure(
            current_token,
            pending.classify_worker_loss(workflow_id, activity_id, worker_id),
        ),
        Err(error) => synthesized_failure(
            current_token,
            format!("retryable:worker liminal reply failed: {error}"),
        ),
    };
    // The gRPC sweeps fail only still-tracked tasks (`remove_worker_tasks`);
    // this is the same structural gate: `complete_task` reports whether THIS
    // call retired the tracked entry. A poisoned tracker fails open (deliver)
    // so the blocked dispatch thread is never left hanging on a broken lock.
    let after_accept = || {
        if synthesized {
            let was_tracked = clear_completed_task_tracking(
                heartbeat_tracker,
                registry,
                worker_id,
                workflow_id,
                activity_id,
            );
            if !was_tracked {
                // Another path already resolved this dispatch (and notified drain):
                // injecting the fabricated lost-worker failure now could reach a retry.
                tracing::debug!(
                    worker_id = ?worker_id,
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    "liminal dispatch already resolved; dropping synthesized lost-worker failure"
                );
                return Ok(false);
            }
        } else {
            // A real worker result: the tracking clear is bookkeeping, never
            // a gate. Fail open — withholding a received result over a
            // poisoned tracker would leave the dispatch waiter parked on an
            // unbounded receive with no recovery path, because the lost-worker
            // sweep reads the same poisoned lock.
            let _ = clear_completed_task_tracking(
                heartbeat_tracker,
                registry,
                worker_id,
                workflow_id,
                activity_id,
            );
        }
        drain_state.notify_activity_drained();
        Ok(true)
    };
    if let Err(error) = pending.complete_fenced_after_accept(
        workflow_id,
        activity_id,
        run_id.as_ref(),
        &submitted_token,
        outcome,
        after_accept,
    ) {
        tracing::warn!(
            worker_id = ?worker_id,
            workflow_id = %workflow_id,
            activity_id = %activity_id,
            %error,
            "liminal activity completion handoff rejected"
        );
    }
}

/// Clears a completed activity's in-flight tracking entry, reporting whether
/// THIS call retired it. Fails open on a poisoned tracker — logged loudly and
/// treated as "was tracked" — because the clear is bookkeeping, never a gate:
/// withholding a completion over the poisoned lock would strand the dispatch
/// waiter on an unbounded receive, and the lost-worker sweep cannot recover it
/// (it reads the same lock).
pub(crate) fn clear_completed_task_tracking(
    heartbeat_tracker: &HeartbeatTracker,
    registry: &crate::worker::ConnectedWorkerRegistry,
    worker_id: WorkerId,
    workflow_id: &WorkflowId,
    activity_id: &ActivityId,
) -> bool {
    heartbeat_tracker
        .complete_task(worker_id, workflow_id, activity_id, registry)
        .unwrap_or_else(|error| {
            tracing::error!(
                worker_id = ?worker_id,
                workflow_id = %workflow_id,
                activity_id = %activity_id,
                %error,
                "failed to clear in-flight tracking for a completed activity; delivering anyway — a poisoned tracker must not withhold a result"
            );
            true
        })
}

struct ActivityDispatchContext<'a> {
    namespace: &'a str,
    activity_type: &'a str,
    worker_id: WorkerId,
    workflow_id: &'a WorkflowId,
    activity_id: &'a ActivityId,
    /// Attempt this dispatch was delivered at — the identity its own-attempt
    /// teardown paths remove the waiter slot by (aion#195).
    attempt: u32,
    completion_token: CompletionToken,
    started_at: Instant,
}

fn activity_task(
    activity_type: &str,
    input: &str,
    execution: (&WorkflowId, &RunId, &ActivityId),
    attempt: u32,
    labels: BTreeMap<String, String>,
    completion_token: &CompletionToken,
) -> ProtoActivityTask {
    let (workflow_id, run_id, activity_id) = execution;
    ProtoActivityTask {
        workflow_id: Some(ProtoWorkflowId::from(workflow_id.clone())),
        activity_id: Some(ProtoActivityId::from(activity_id.clone())),
        activity_type: activity_type.to_owned(),
        input: Some(ProtoPayload {
            content_type: String::from("application/json"),
            bytes: input.as_bytes().to_vec(),
        }),
        attempt,
        labels: labels.into_iter().collect(),
        run_id: Some(run_id.clone().into()),
        completion_token: completion_token.as_str().to_owned(),
        idempotency_key: idempotency_key(workflow_id, run_id, activity_id),
    }
}

fn log_activity_completion(context: &ActivityDispatchContext<'_>, succeeded: bool) {
    let duration_ms = duration_ms(context.started_at.elapsed());
    tracing::info!(
        operation = "activity_complete",
        namespace = context.namespace,
        workflow_id = %context.workflow_id,
        activity_id = %context.activity_id,
        activity_type = context.activity_type,
        worker_id = ?context.worker_id,
        duration_ms,
        outcome = if succeeded { "succeeded" } else { "failed" },
        "activity completed"
    );
}

fn duration_ms(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

fn log_worker_error(
    error_type: &'static str,
    namespace: &str,
    activity_type: &str,
    workflow_id: &WorkflowId,
    activity_id: &ActivityId,
    worker_id: Option<super::registry::WorkerId>,
    reason: &str,
) {
    tracing::error!(
        operation = "activity_dispatch",
        namespace,
        workflow_id = %workflow_id,
        activity_id = %activity_id,
        activity_type,
        worker_id = ?worker_id,
        error_type,
        reason,
        "worker interaction failed"
    );
}

#[cfg(test)]
mod tests {
    use std::sync::{
        Mutex,
        atomic::{AtomicBool, Ordering},
    };

    use aion_core::{ActivityError, ActivityErrorKind, ContentType, Payload};

    use crate::worker::dispatch::ActivityCompletionOutcome;

    use super::*;

    fn activity_id(pos: u64) -> ActivityId {
        ActivityId::from_sequence_position(pos)
    }

    /// One delivery of attempt 1 for `(workflow_id, id)` in its own run — what
    /// every single-dispatch unit test below is. The run and the attempt are the
    /// two discriminators the fence separates a REDELIVERY from a retry and from
    /// a new execution generation with, so they are stated here rather than
    /// implied.
    fn insert_attempt_one(
        pending: &PendingActivities,
        workflow_id: &WorkflowId,
        id: &ActivityId,
    ) -> Result<(CompletionToken, SyncReceiver), ServerError> {
        let (token, rx, superseded) =
            pending.insert(workflow_id.clone(), &RunId::new_v4(), id.clone(), 1)?;
        assert!(
            superseded.is_none(),
            "a first insert must never report a superseded holder"
        );
        Ok((token, rx))
    }

    #[test]
    fn pending_insert_and_complete_delivers_result() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(1);
        let rx = insert_attempt_one(&pending, &workflow_id, &id)?.1;

        assert!(pending.complete(&workflow_id, &id, None, Ok("done".to_owned())));
        assert_eq!(
            rx.recv_timeout(Duration::from_millis(50)),
            Ok(Ok("done".to_owned()))
        );
        Ok(())
    }

    /// #246 / #254 red: retrying an activity while the same worker still holds
    /// the prior attempt must never replace its responder silently. Today's
    /// attempt-free `DashMap::insert` drops the first sender, producing the
    /// terminal "activity response channel dropped" specimen.
    #[test]
    fn same_execution_collision_is_a_named_typed_refusal() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(2);
        // The SAME execution redelivered: same run, same attempt. The fence
        // treats that as a sibling authorization; the pending map still refuses
        // to replace a live responder.
        let run_id = RunId::new_v4();
        let _first = pending.insert(workflow_id.clone(), &run_id, id.clone(), 1)?;

        let error = pending
            .insert(workflow_id, &run_id, id, 1)
            .err()
            .ok_or_else(|| {
                ServerError::worker_dispatch(
                    "test",
                    "same-worker-collision",
                    "a live responder collision was silently replaced",
                )
            })?;

        assert!(
            error.to_string().contains("pending activity collision"),
            "{error}"
        );
        Ok(())
    }

    /// aion#195 THE PIN: a HIGHER attempt arriving at a held execution site is
    /// a retry whose predecessor timed out — it takes the site over instead of
    /// dying on it. Before the fix, this insert returned the collision error
    /// and the workflow FAILED while the healthy responder ran on (workflow
    /// b38ecf80: attempt 2 collided with attempt 1's live responder one second
    /// after its backoff and the whole run was lost).
    #[test]
    fn a_higher_attempt_takes_over_a_stale_execution_site() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(3);
        let run_id = RunId::new_v4();
        let (_stale_token, stale_rx, none) =
            pending.insert(workflow_id.clone(), &run_id, id.clone(), 1)?;
        assert!(none.is_none());

        let (fresh_token, fresh_rx, superseded) =
            pending.insert(workflow_id.clone(), &run_id, id.clone(), 2)?;
        assert_eq!(
            superseded,
            Some(1),
            "the takeover must name the attempt it displaced"
        );

        // The superseded waiter is woken with the retryable sentinel, so its
        // blocked dispatcher thread finishes instead of hanging forever.
        let stale_resolution = stale_rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|_| {
                ServerError::worker_dispatch(
                    "test",
                    "takeover",
                    "the superseded waiter was never woken",
                )
            })?;
        let reason = match stale_resolution {
            Ok(delivered) => {
                return Err(ServerError::worker_dispatch(
                    "test",
                    "takeover",
                    format!("the superseded waiter resolved Ok({delivered}) instead of an error"),
                ));
            }
            Err(reason) => reason,
        };
        assert!(
            reason.starts_with("retryable:") && reason.contains("superseded by attempt 2"),
            "{reason}"
        );

        // The site now belongs to attempt 2: a completion presenting attempt
        // 2's token delivers to attempt 2's waiter.
        pending.complete_activity(ActivityCompletion {
            workflow_id: workflow_id.clone(),
            activity_id: id.clone(),
            run_id: None,
            completion_token: fresh_token,
            outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                ContentType::Json,
                br#""second attempt wins""#.to_vec(),
            )),
        })?;
        assert_eq!(
            fresh_rx.recv_timeout(Duration::from_millis(50)),
            Ok(Ok(r#""second attempt wins""#.to_owned()))
        );
        Ok(())
    }

    /// aion#195: a LOWER attempt arriving at a held site is a stale
    /// re-dispatch, not a retry — it must still be refused, and the refusal
    /// must name both attempts.
    #[test]
    fn a_lower_attempt_is_still_refused_by_name() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(4);
        let run_id = RunId::new_v4();
        let _held = pending.insert(workflow_id.clone(), &run_id, id.clone(), 3)?;
        let error = pending
            .insert(workflow_id, &run_id, id, 2)
            .err()
            .ok_or_else(|| {
                ServerError::worker_dispatch(
                    "test",
                    "stale-redispatch",
                    "a lower attempt silently replaced a live responder",
                )
            })?;
        let text = error.to_string();
        assert!(
            text.contains("attempt 2 arrived while attempt 3 still holds"),
            "{text}"
        );
        Ok(())
    }

    #[test]
    fn pending_complete_unknown_returns_false() {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        assert!(!pending.complete(
            &WorkflowId::new_v4(),
            &activity_id(99),
            None,
            Ok("orphan".to_owned())
        ));
    }

    #[derive(Default)]
    struct RecordingOutboxCallback {
        completions: Mutex<Vec<(WorkflowId, ActivityId, String)>>,
        failures: Mutex<Vec<(WorkflowId, ActivityId, String)>>,
        live: bool,
    }

    impl OutboxDeliveryCallback for RecordingOutboxCallback {
        fn deliver_completion(
            &self,
            workflow_id: &WorkflowId,
            activity_id: &ActivityId,
            run_id: Option<&RunId>,
            result: String,
        ) -> Result<bool, ServerError> {
            let _ = run_id;
            self.completions
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?
                .push((workflow_id.clone(), activity_id.clone(), result));
            Ok(self.live)
        }

        fn deliver_failure(
            &self,
            workflow_id: &WorkflowId,
            activity_id: &ActivityId,
            run_id: Option<&RunId>,
            reason: String,
        ) -> Result<bool, ServerError> {
            let _ = run_id;
            self.failures
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?
                .push((workflow_id.clone(), activity_id.clone(), reason));
            Ok(self.live)
        }
    }

    #[test]
    fn unmatched_completion_routes_to_outbox_callback_when_installed() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let callback = Arc::new(RecordingOutboxCallback {
            live: true,
            ..RecordingOutboxCallback::default()
        });
        // Install on one clone; the wiring must be visible to every clone.
        pending.clone().set_outbox_delivery(callback.clone());

        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(7);

        // No pending entry: the completion is unmatched and must route to the
        // callback rather than being dropped. A live workflow reports true.
        assert!(pending.complete(&workflow_id, &id, None, Ok("done".to_owned())));
        let completions = callback
            .completions
            .lock()
            .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?;
        assert_eq!(completions.len(), 1);
        assert_eq!(completions[0].0, workflow_id);
        assert_eq!(completions[0].1, id);
        assert_eq!(completions[0].2, "done");
        Ok(())
    }

    #[test]
    fn unmatched_failure_routes_to_outbox_callback_and_not_live_reports_false()
    -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        // live = false models the expected stale-completion case.
        let callback = Arc::new(RecordingOutboxCallback::default());
        pending.set_outbox_delivery(callback.clone());

        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(8);

        assert!(!pending.complete(&workflow_id, &id, None, Err("retryable:boom".to_owned())));
        let failures = callback
            .failures
            .lock()
            .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?;
        assert_eq!(failures.len(), 1);
        assert_eq!(failures[0].2, "retryable:boom");
        Ok(())
    }

    #[test]
    fn unmatched_completion_is_silent_drop_when_no_callback_installed() {
        // Flag-off byte-identical behaviour: no callback, unmatched returns
        // false (silent drop) exactly as before.
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        assert!(!pending.complete(
            &WorkflowId::new_v4(),
            &activity_id(9),
            None,
            Ok("x".to_owned())
        ));
    }

    #[test]
    fn matched_completion_never_reaches_outbox_callback() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let callback = Arc::new(RecordingOutboxCallback {
            live: true,
            ..RecordingOutboxCallback::default()
        });
        pending.set_outbox_delivery(callback.clone());

        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(10);
        let rx = insert_attempt_one(&pending, &workflow_id, &id)?.1;

        assert!(pending.complete(&workflow_id, &id, None, Ok("matched".to_owned())));
        assert_eq!(
            rx.recv_timeout(Duration::from_millis(50)),
            Ok(Ok("matched".to_owned()))
        );
        assert!(
            callback
                .completions
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?
                .is_empty(),
            "a matched completion must deliver to its waiter, not the outbox callback"
        );
        Ok(())
    }

    /// #207: parking resolves the matched waiter with the ephemeral parked
    /// sentinel — the exact string the engine's retry loop classifies as
    /// `Parked` — and nothing else.
    #[test]
    fn park_activity_resolves_matched_waiter_with_the_parked_sentinel() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(11);
        let rx = insert_attempt_one(&pending, &workflow_id, &id)?.1;

        pending.park_activity(&workflow_id, &id)?;
        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|e| ServerError::worker_dispatch("", "", format!("channel: {e}")))?;
        assert_eq!(result, Err(aion::PARKED_ACTIVITY_REASON.to_owned()));
        Ok(())
    }

    /// #207: an unmatched park is a no-op and is NEVER routed to the outbox
    /// delivery callback — a park is not a failure and must never reach a
    /// workflow.
    #[test]
    fn unmatched_park_is_a_noop_and_never_reaches_the_outbox_callback() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let callback = Arc::new(RecordingOutboxCallback {
            live: true,
            ..RecordingOutboxCallback::default()
        });
        pending.set_outbox_delivery(callback.clone());

        pending.park_activity(&WorkflowId::new_v4(), &activity_id(12))?;

        assert!(
            callback
                .failures
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?
                .is_empty(),
            "a park must never be delivered as an outbox failure"
        );
        assert!(
            callback
                .completions
                .lock()
                .map_err(|_| ServerError::lock_poisoned("recording outbox callback"))?
                .is_empty(),
            "a park must never be delivered as an outbox completion"
        );
        Ok(())
    }

    #[test]
    fn completion_sink_routes_success() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(2);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;
        let payload = Payload::new(ContentType::Json, br#"{"greeting":"hi"}"#.to_vec());

        pending.complete_activity(ActivityCompletion {
            workflow_id,
            activity_id: id,
            run_id: None,
            completion_token,
            outcome: ActivityCompletionOutcome::Succeeded(payload),
        })?;

        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|e| ServerError::worker_dispatch("", "", format!("channel: {e}")))?;
        assert_eq!(result, Ok(r#"{"greeting":"hi"}"#.to_owned()));
        Ok(())
    }

    #[test]
    fn accepted_hook_runs_before_the_result_is_observable() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(13);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;
        let accepted = AtomicBool::new(false);

        pending.complete_activity_after_accept(
            ActivityCompletion {
                workflow_id,
                activity_id: id,
                run_id: None,
                completion_token,
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#""ordered""#.to_vec(),
                )),
            },
            || {
                accepted.store(true, Ordering::SeqCst);
                Ok(())
            },
        )?;

        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|error| ServerError::worker_dispatch("", "", format!("channel: {error}")))?;
        assert_eq!(result, Ok(r#""ordered""#.to_owned()));
        assert!(
            accepted.load(Ordering::SeqCst),
            "the accepted-path cleanup hook must finish before the waiter can observe the result"
        );
        Ok(())
    }

    #[test]
    fn a_non_publishing_settlement_restores_the_generation_for_the_true_resolver()
    -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(16);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;

        // The synthesized-failure path finding its dispatch already resolved
        // by another path: settlement reports "do not publish". The generation
        // it consumed on accept must come back, because the OTHER path (the
        // lost-worker sweep) presents the same token next.
        let settled = pending.complete_fenced_after_accept(
            &workflow_id,
            &id,
            None,
            &completion_token,
            Err("retryable:worker lost".to_owned()),
            || Ok(false),
        )?;
        assert!(!settled, "a non-publishing settlement must report false");
        assert!(
            rx.try_recv().is_err(),
            "a non-publishing settlement must not publish"
        );

        // The true resolver arrives with the same token: it must be accepted,
        // not refused with NoCurrentGeneration.
        let published = pending.complete_fenced_after_accept(
            &workflow_id,
            &id,
            None,
            &completion_token,
            Err("retryable:worker lost".to_owned()),
            || Ok(true),
        )?;
        assert!(
            published,
            "the true resolver must be accepted after a non-publishing settlement"
        );
        assert_eq!(
            rx.recv_timeout(Duration::from_millis(50))
                .map_err(|error| ServerError::worker_dispatch(
                    "",
                    "",
                    format!("channel: {error}")
                ))?,
            Err("retryable:worker lost".to_owned())
        );
        Ok(())
    }

    #[test]
    fn a_poisoned_tracker_fails_open_and_never_withholds_a_completion() {
        let tracker = HeartbeatTracker::new(TEST_HEARTBEAT_WINDOW);
        tracker.poison_for_tests();
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(17);
        // An empty registry: the worker was never registered here, so the
        // selection-side count has nothing to correct. This test is about the
        // TRACKER's poison policy, and the registry is present only because
        // the two counts are written by one call.
        let registry = ConnectedWorkerRegistry::default();
        let was_tracked = clear_completed_task_tracking(
            &tracker,
            &registry,
            WorkerId::from_value(1),
            &workflow_id,
            &id,
        );
        assert!(
            was_tracked,
            "a poisoned tracker must fail open (deliver), never withhold"
        );
    }

    #[test]
    fn accepted_hook_failure_blocks_publication_and_restores_generation() -> Result<(), ServerError>
    {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(15);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;

        let rejected = pending.complete_activity_after_accept(
            ActivityCompletion {
                workflow_id: workflow_id.clone(),
                activity_id: id.clone(),
                run_id: None,
                completion_token: completion_token.clone(),
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#""ordered""#.to_vec(),
                )),
            },
            || Err(ServerError::lock_poisoned("accepted-path settlement")),
        );

        assert!(matches!(rejected, Err(ServerError::LockPoisoned { .. })));
        assert!(
            rx.try_recv().is_err(),
            "failed accepted-path settlement must not publish the completion"
        );

        pending.complete_activity_after_accept(
            ActivityCompletion {
                workflow_id,
                activity_id: id,
                run_id: None,
                completion_token,
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#""ordered""#.to_vec(),
                )),
            },
            || Ok(()),
        )?;
        assert_eq!(
            rx.recv_timeout(Duration::from_millis(50))
                .map_err(|error| ServerError::worker_dispatch(
                    "",
                    "",
                    format!("channel: {error}")
                ))?,
            Ok(r#""ordered""#.to_owned())
        );
        Ok(())
    }

    #[test]
    fn refused_generation_never_runs_the_accepted_hook() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(14);
        let (_completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;
        let accepted = AtomicBool::new(false);

        let rejected = pending.complete_activity_after_accept(
            ActivityCompletion {
                workflow_id,
                activity_id: id,
                run_id: None,
                completion_token: CompletionToken::for_test(),
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#""stale""#.to_vec(),
                )),
            },
            || {
                accepted.store(true, Ordering::SeqCst);
                Ok(())
            },
        );

        assert!(matches!(
            rejected,
            Err(ServerError::ActivityCompletionRejected { .. })
        ));
        assert!(
            !accepted.load(Ordering::SeqCst),
            "a refused generation must preserve liveness by skipping accepted-path cleanup"
        );
        assert!(
            rx.try_recv().is_err(),
            "a refused generation must leave the current waiter unresolved"
        );
        Ok(())
    }

    #[test]
    fn malformed_payload_does_not_consume_the_current_generation() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(12);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;

        let malformed = pending.complete_activity(ActivityCompletion {
            workflow_id: workflow_id.clone(),
            activity_id: id.clone(),
            run_id: None,
            completion_token: completion_token.clone(),
            outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                ContentType::Json,
                vec![0xff],
            )),
        });
        assert!(matches!(malformed, Err(ServerError::WorkerDispatch { .. })));
        assert!(
            rx.try_recv().is_err(),
            "an invalid result must leave the waiter unresolved"
        );

        pending.complete_activity(ActivityCompletion {
            workflow_id,
            activity_id: id,
            run_id: None,
            completion_token,
            outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                ContentType::Json,
                br#""valid""#.to_vec(),
            )),
        })?;
        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|error| ServerError::worker_dispatch("", "", format!("channel: {error}")))?;
        assert_eq!(result, Ok(r#""valid""#.to_owned()));
        Ok(())
    }

    #[test]
    fn completion_sink_routes_retryable_error() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(3);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;

        pending.complete_activity(ActivityCompletion {
            workflow_id,
            activity_id: id,
            run_id: None,
            completion_token,
            outcome: ActivityCompletionOutcome::Failed(ActivityError {
                kind: ActivityErrorKind::Retryable,
                message: "temporary".to_owned(),
                details: None,
            }),
        })?;

        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|e| ServerError::worker_dispatch("", "", format!("channel: {e}")))?;
        assert_eq!(result, Err("retryable:temporary".to_owned()));
        Ok(())
    }

    #[test]
    fn completion_sink_routes_policy_refusal_with_its_own_prefix() -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let workflow_id = WorkflowId::new_v4();
        let id = activity_id(4);
        let (completion_token, rx) = insert_attempt_one(&pending, &workflow_id, &id)?;

        pending.complete_activity(ActivityCompletion {
            workflow_id,
            activity_id: id,
            run_id: None,
            completion_token,
            outcome: ActivityCompletionOutcome::Failed(ActivityError {
                kind: ActivityErrorKind::PolicyRefused,
                message: "provider safety policy".to_owned(),
                details: None,
            }),
        })?;

        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|error| ServerError::worker_dispatch("", "", format!("channel: {error}")))?;
        assert_eq!(
            result,
            Err("policy_refused:provider safety policy".to_owned())
        );
        Ok(())
    }

    /// Regression test (#59, brief D12): pending tracking must be keyed by
    /// the full `(WorkflowId, ActivityId)` pair. The dispatcher fabricates
    /// activity ids from a process-local counter that resets on server
    /// restart, so a stale result re-reported from a worker's previous
    /// session carries the same bare `ActivityId` as a fresh post-restart
    /// dispatch. Under bare-`ActivityId` keying the stale result completed
    /// the wrong execution; with pair keying it is dropped and the genuine
    /// result still completes.
    #[test]
    fn stale_result_for_other_workflow_does_not_complete_pending_dispatch()
    -> Result<(), ServerError> {
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let post_restart_workflow = WorkflowId::new_v4();
        let pre_restart_workflow = WorkflowId::new_v4();
        // Counter resets to the same sequence position after restart.
        let id = activity_id(1);
        let (completion_token, rx) = insert_attempt_one(&pending, &post_restart_workflow, &id)?;

        // Stale pre-restart result: same activity id, different workflow.
        let rejected = pending.complete_activity(ActivityCompletion {
            workflow_id: pre_restart_workflow,
            activity_id: id.clone(),
            run_id: None,
            completion_token: CompletionToken::for_test(),
            outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                ContentType::Json,
                br#""stale""#.to_vec(),
            )),
        });
        assert!(matches!(
            rejected,
            Err(ServerError::ActivityCompletionRejected { .. })
        ));
        assert!(
            rx.try_recv().is_err(),
            "stale result for a different workflow must not complete this dispatch"
        );

        // The genuine result for the pending execution still completes.
        pending.complete_activity(ActivityCompletion {
            workflow_id: post_restart_workflow,
            activity_id: id,
            run_id: None,
            completion_token,
            outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                ContentType::Json,
                br#""fresh""#.to_vec(),
            )),
        })?;
        let result = rx
            .recv_timeout(Duration::from_millis(50))
            .map_err(|e| ServerError::worker_dispatch("", "", format!("channel: {e}")))?;
        assert_eq!(result, Ok(r#""fresh""#.to_owned()));
        Ok(())
    }

    /// Liveness tracker for dispatcher unit tests; the window only matters
    /// to expiry checks, which nothing in these tests drives.
    /// The heartbeat window these unit tests wire, named once so the
    /// transport-loss budget they run under is explicit rather than inherited.
    /// It is the same window [`test_tracker`] uses, so a sink and a tracker
    /// built side by side in a test agree exactly as they do in production.
    const TEST_HEARTBEAT_WINDOW: Duration = Duration::from_secs(5);

    /// The budget a sink ends up with is DERIVED from the operator's window,
    /// and this is the arithmetic — not prose about it.
    ///
    /// Before #100 the wiring was `default().with_heartbeat_window(w)`: two
    /// calls where the second was omissible, and deleting it from all five
    /// production sites reddened nothing, because every test ran the zero-budget
    /// default. This pin and its sibling below are what make that deletion
    /// impossible to perform silently.
    #[test]
    fn a_sink_derives_its_transport_budget_from_the_operator_window() {
        let window = Duration::from_secs(7);
        let sink = PendingActivities::new(window);

        assert_eq!(
            sink.transport_losses().budget(),
            window * super::super::transport_loss::TRANSPORT_LOSS_BUDGET_WINDOWS,
            "the transport-loss budget must be the operator's heartbeat window times \
             TRANSPORT_LOSS_BUDGET_WINDOWS; a sink carrying any other budget is one no \
             operator declared"
        );
    }

    /// The wiring itself, which is where the defect actually lived: a dispatcher
    /// is handed a `HeartbeatTracker` and must not be able to carry a transport
    /// budget that disagrees with it.
    ///
    /// This asserts on the dispatcher a caller really builds, so it sits on the
    /// path a forgetful wiring would take rather than on the constructor in
    /// isolation.
    #[test]
    fn a_dispatcher_carries_the_transport_budget_of_the_tracker_it_was_given() {
        let window = Duration::from_secs(11);
        let registry = ConnectedWorkerRegistry::default();
        let dispatcher =
            WorkerActivityDispatcher::new(registry, "default", HeartbeatTracker::new(window));

        assert_eq!(
            dispatcher.pending().transport_losses().budget(),
            window * super::super::transport_loss::TRANSPORT_LOSS_BUDGET_WINDOWS,
            "the dispatcher's transport budget must be derived from the heartbeat window it \
             was constructed with; a zero or mismatched budget here means a worker loss is \
             declared transport-exhausted after ONE re-dispatch instead of after the \
             operator's window"
        );
        assert_ne!(
            dispatcher.pending().transport_losses().budget(),
            Duration::ZERO,
            "a ZERO budget is the specific regression this pin exists for: it grants exactly \
             one re-dispatchable loss and then declares the infrastructure flapping"
        );
    }

    fn test_tracker() -> HeartbeatTracker {
        HeartbeatTracker::new(TEST_HEARTBEAT_WINDOW)
    }

    /// A `greet` dispatch request carrying real (test-synthesized) ids, the
    /// engine-seam shape `WorkerActivityDispatcher::dispatch` now consumes.
    fn greet_request() -> ActivityDispatch {
        ActivityDispatch {
            namespace: "default".to_owned(),
            task_queue: "default".to_owned(),
            node: None,
            workflow_id: WorkflowId::new_v4(),
            run_id: RunId::new_v4(),
            activity_id: ActivityId::from_sequence_position(0),
            name: "greet".to_owned(),
            input: "{}".to_owned(),
            config: "{}".to_owned(),
            attempt: 1,
            labels: std::collections::BTreeMap::new(),
            advisory: false,
        }
    }

    #[test]
    fn dispatcher_fails_immediately_when_draining_without_workers() {
        let registry = ConnectedWorkerRegistry::default();
        let drain = DrainState::default();
        let dispatcher = WorkerActivityDispatcher::new(registry, "default", test_tracker())
            .with_drain_state(drain.clone());

        let _ = drain.begin();

        let result = dispatcher.dispatch(greet_request());

        assert!(result.is_err());
        let err = result.err().unwrap_or_default();
        assert!(
            err.contains("drain"),
            "expected drain rejection, got: {err}"
        );
    }

    /// Regression test for the production stall where every remote activity
    /// timed out: the engine invoked the sync `dispatch` from inside a
    /// spawned tokio task (`futures::future::lazy` polled on a runtime
    /// worker), and the woken stream-consumer task landed in that blocked
    /// worker's non-stealable LIFO slot, so the queued `ActivityTask` was
    /// only delivered when the then-extant 30s dispatch timeout fired (the
    /// dispatch wait is unbounded today; the stall would now be a hang).
    ///
    /// Mirrors the real wiring minus tonic: the real registry channel that
    /// the gRPC stream forwarder drains, a worker task awaiting that channel
    /// on the same runtime, completion through the production
    /// `ActivityCompletionSink`, and the sync dispatch invoked from a
    /// runtime worker task — the worst case the `block_in_place` guard in
    /// `dispatch` defends against (the engine itself now routes through
    /// `dispatch_async`, off the async workers).
    /// aion#195 END TO END: attempt 1 is delivered to a worker that never
    /// answers (the engine's per-attempt bound has already failed it and
    /// dropped its future); attempt 2 of the same execution then arrives. The
    /// site is taken over, the worker is asked to STOP the corpse on its own
    /// dispatch stream — the cancel frame precedes attempt 2's task — the
    /// corpse's heartbeat task is untracked, attempt 2 runs and completes, and
    /// attempt 1's blocked dispatcher thread resolves with the retryable
    /// superseded sentinel instead of hanging or failing the workflow.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn a_retry_takes_over_a_timed_out_attempt_and_asks_its_worker_to_stop()
    -> Result<(), Box<dyn std::error::Error>> {
        let registry = ConnectedWorkerRegistry::default();
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let (worker_tx, mut worker_rx) = tokio::sync::mpsc::channel(32);
        let activity_types = [String::from("greet")];
        let registration = registry.register(
            "default",
            activity_types.iter(),
            worker_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let worker_id = registration
            .worker_id()
            .ok_or("test worker registration missing id")?;
        let tracker = test_tracker();
        let dispatcher = Arc::new(
            WorkerActivityDispatcher::new(registry, "default", tracker.clone())
                .with_pending(pending.clone()),
        );

        let request = greet_request();
        let workflow_id = request.workflow_id.clone();
        let activity_id = request.activity_id.clone();

        let first_dispatcher = Arc::clone(&dispatcher);
        let first_request = request.clone();
        let first_attempt = tokio::spawn(futures::future::lazy(move |_| {
            first_dispatcher.dispatch(first_request)
        }));
        let Some(WorkerMessage::ActivityTask(first_task)) = worker_rx.recv().await else {
            return Err("expected attempt 1's task on the worker channel".into());
        };
        assert_eq!(first_task.attempt, 1);

        let mut retry_request = request.clone();
        retry_request.attempt = 2;
        let retry_dispatcher = Arc::clone(&dispatcher);
        let retry_attempt = tokio::spawn(futures::future::lazy(move |_| {
            retry_dispatcher.dispatch(retry_request)
        }));

        // The release ladder runs before attempt 2's task is pushed, so the
        // worker sees the STOP for the corpse first, then the new work.
        let Some(WorkerMessage::CancelActivity(cancel)) = worker_rx.recv().await else {
            return Err(
                "expected the superseded attempt's cancel frame before attempt 2's task".into(),
            );
        };
        assert_eq!(
            cancel
                .activity_id
                .map(|id| id.sequence_position)
                .unwrap_or_default(),
            activity_id.sequence_position()
        );
        let Some(WorkerMessage::ActivityTask(retry_task)) = worker_rx.recv().await else {
            return Err("expected attempt 2's task on the worker channel".into());
        };
        assert_eq!(retry_task.attempt, 2);

        // The corpse's heartbeat task was untracked, so the expiry sweep can
        // never present its stale token; only attempt 2 remains in flight.
        let in_flight = tracker.in_flight_for_workflow(&workflow_id)?;
        assert_eq!(
            in_flight
                .iter()
                .map(|task| task.attempt)
                .collect::<Vec<_>>(),
            vec![2],
            "exactly attempt 2 must remain tracked for worker {worker_id:?}"
        );

        let retry_token = CompletionToken::from_wire(
            &workflow_id,
            &activity_id,
            retry_task.completion_token.clone(),
        )
        .map_err(|error| error.to_string())?;
        pending.complete_activity(ActivityCompletion {
            workflow_id: workflow_id.clone(),
            activity_id: activity_id.clone(),
            run_id: None,
            completion_token: retry_token,
            outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                ContentType::Json,
                br#""attempt 2 delivered""#.to_vec(),
            )),
        })?;

        let retry_result = retry_attempt.await.map_err(|error| error.to_string())?;
        assert_eq!(retry_result, Ok(r#""attempt 2 delivered""#.to_owned()));
        let first_result = first_attempt.await.map_err(|error| error.to_string())?;
        let first_reason = match first_result {
            Ok(delivered) => {
                return Err(
                    format!("attempt 1 resolved Ok({delivered}) instead of superseded").into(),
                );
            }
            Err(reason) => reason,
        };
        assert!(
            first_reason.starts_with("retryable:")
                && first_reason.contains("superseded by attempt 2"),
            "{first_reason}"
        );
        registration.deregister()?;
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dispatch_inside_runtime_task_delivers_promptly_and_round_trips()
    -> Result<(), Box<dyn std::error::Error>> {
        let registry = ConnectedWorkerRegistry::default();
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let (worker_tx, mut worker_rx) = tokio::sync::mpsc::channel(32);
        let activity_types = [String::from("greet")];
        let registration = registry.register(
            "default",
            activity_types.iter(),
            worker_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;

        let sink = pending.clone();
        let echo_worker = tokio::spawn(async move {
            let Some(WorkerMessage::ActivityTask(task)) = worker_rx.recv().await else {
                return Err("expected an activity task on the worker channel".to_owned());
            };
            let workflow_id = task
                .workflow_id
                .ok_or("task missing workflow id")
                .and_then(|id| WorkflowId::try_from(id).map_err(|_| "bad workflow id"))?;
            let activity_id = task
                .activity_id
                .map(ActivityId::from)
                .ok_or("task missing activity id")?;
            let completion_token =
                CompletionToken::from_wire(&workflow_id, &activity_id, task.completion_token)
                    .map_err(|error| error.to_string())?;
            sink.complete_activity(ActivityCompletion {
                workflow_id,
                activity_id,
                run_id: None,
                completion_token,
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#"{"greeting":"hello"}"#.to_vec(),
                )),
            })
            .map_err(|error| error.to_string())
        });

        let dispatcher = Arc::new(
            WorkerActivityDispatcher::new(registry, "default", test_tracker())
                .with_pending(pending),
        );
        let started = Instant::now();
        // Invoke the sync dispatch inside the first poll of a spawned task:
        // the worst-case calling context for the `block_in_place` guard.
        let dispatch_task = tokio::spawn(futures::future::lazy(move |_| {
            dispatcher.dispatch(greet_request())
        }));
        let result = dispatch_task.await.map_err(|error| error.to_string())?;
        let elapsed = started.elapsed();

        assert_eq!(result, Ok(r#"{"greeting":"hello"}"#.to_owned()));
        assert!(
            elapsed < Duration::from_secs(5),
            "dispatch round trip took {elapsed:?}; task delivery must not \
             depend on the blocked dispatch thread"
        );
        echo_worker.await.map_err(|error| error.to_string())??;
        registration.deregister()?;
        Ok(())
    }

    /// A current-thread runtime cannot host the blocking wait (the stream
    /// forwarder would share its only executor thread), so dispatch must
    /// fail fast with a precise error instead of blocking forever.
    #[tokio::test]
    async fn dispatch_on_current_thread_runtime_fails_fast()
    -> Result<(), Box<dyn std::error::Error>> {
        let registry = ConnectedWorkerRegistry::default();
        let (worker_tx, _worker_rx) = tokio::sync::mpsc::channel(32);
        let activity_types = [String::from("greet")];
        let registration = registry.register(
            "default",
            activity_types.iter(),
            worker_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let dispatcher = WorkerActivityDispatcher::new(registry, "default", test_tracker());

        let started = Instant::now();
        let result = dispatcher.dispatch(greet_request());
        let elapsed = started.elapsed();

        let err = result.err().ok_or("expected dispatch to fail")?;
        assert!(
            err.contains("multi-thread tokio runtime"),
            "unexpected error: {err}"
        );
        assert!(
            elapsed < Duration::from_secs(5),
            "fail-fast path took {elapsed:?}"
        );
        registration.deregister()?;
        Ok(())
    }

    /// Bridge-level mirror of the e2e node-pin proof: two workers share the
    /// `(namespace, task_queue)` pool but advertise different nodes; an
    /// `ActivityDispatch` pinned to one node must reach ONLY the worker on that
    /// node through the live engine-seam `WorkerActivityDispatcher`. This is the
    /// regression guard for the bridge discarding the dispatch's `node`.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dispatch_pinned_to_node_reaches_only_that_node()
    -> Result<(), Box<dyn std::error::Error>> {
        let registry = ConnectedWorkerRegistry::default();
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let activity_types = [String::from("greet")];
        let (n1_tx, mut n1_rx) = tokio::sync::mpsc::channel(32);
        let (n2_tx, mut n2_rx) = tokio::sync::mpsc::channel(32);
        // Register the DECOY (n2) FIRST so it owns the lowest worker id. The
        // bridge's `select_worker` orders candidates by worker id and starts at
        // the pool's rotation cursor, which is 0 for a pool nothing has selected
        // from yet — so the FIRST unfiltered selection here would name n2 (the
        // decoy), and a bridge that DISCARDED the node would route to it: the n1
        // echo would never fire and the round trip would time out. With the node
        // threaded through, selection is filtered to n1 before the cursor is
        // consulted, so the rotation cannot rescue a bridge that dropped it.
        let on_n2 = registry.register_namespaces(
            [String::from("default")],
            "default",
            Some(String::from("n2")),
            activity_types.iter(),
            n2_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;
        let on_n1 = registry.register_namespaces(
            [String::from("default")],
            "default",
            Some(String::from("n1")),
            activity_types.iter(),
            n1_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;

        // Echo only on the n1 channel: the dispatch can only complete if the
        // task was routed to n1. If it leaked to n2, the n1 wait would stall and
        // the round trip below would time out instead.
        let sink = pending.clone();
        let echo_n1 = tokio::spawn(async move {
            let Some(WorkerMessage::ActivityTask(task)) = n1_rx.recv().await else {
                return Err("expected an activity task on the n1 worker channel".to_owned());
            };
            let workflow_id = task
                .workflow_id
                .ok_or("task missing workflow id")
                .and_then(|id| WorkflowId::try_from(id).map_err(|_| "bad workflow id"))?;
            let activity_id = task
                .activity_id
                .map(ActivityId::from)
                .ok_or("task missing activity id")?;
            let completion_token =
                CompletionToken::from_wire(&workflow_id, &activity_id, task.completion_token)
                    .map_err(|error| error.to_string())?;
            sink.complete_activity(ActivityCompletion {
                workflow_id,
                activity_id,
                run_id: None,
                completion_token,
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#"{"greeting":"hello"}"#.to_vec(),
                )),
            })
            .map_err(|error| error.to_string())
        });

        let dispatcher = Arc::new(
            WorkerActivityDispatcher::new(registry.clone(), "default", test_tracker())
                .with_pending(pending),
        );

        let pinned = ActivityDispatch {
            node: Some(String::from("n1")),
            ..greet_request()
        };
        let started = Instant::now();
        let result = tokio::spawn(futures::future::lazy(move |_| dispatcher.dispatch(pinned)))
            .await
            .map_err(|error| error.to_string())?;
        let elapsed = started.elapsed();

        assert_eq!(result, Ok(r#"{"greeting":"hello"}"#.to_owned()));
        assert!(
            elapsed < Duration::from_secs(5),
            "pinned dispatch round trip took {elapsed:?}; the task must route to n1"
        );
        echo_n1.await.map_err(|error| error.to_string())??;

        // The n2 worker (wrong node) must never have been handed the task.
        assert!(
            n2_rx.try_recv().is_err(),
            "node=Some(\"n1\") dispatch must not reach the n2 worker"
        );

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

    /// An unpinned (`node = None`) dispatch is byte-identical to today: it
    /// reaches a worker in the pool regardless of the worker's advertised node.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn unpinned_dispatch_reaches_a_pooled_worker_regardless_of_node()
    -> Result<(), Box<dyn std::error::Error>> {
        let registry = ConnectedWorkerRegistry::default();
        let pending = PendingActivities::new(TEST_HEARTBEAT_WINDOW);
        let activity_types = [String::from("greet")];
        let (n1_tx, mut n1_rx) = tokio::sync::mpsc::channel(32);
        let on_n1 = registry.register_namespaces(
            [String::from("default")],
            "default",
            Some(String::from("n1")),
            activity_types.iter(),
            n1_tx,
            crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
        )?;

        let sink = pending.clone();
        let echo = tokio::spawn(async move {
            let Some(WorkerMessage::ActivityTask(task)) = n1_rx.recv().await else {
                return Err("expected an activity task on the worker channel".to_owned());
            };
            let workflow_id = task
                .workflow_id
                .ok_or("task missing workflow id")
                .and_then(|id| WorkflowId::try_from(id).map_err(|_| "bad workflow id"))?;
            let activity_id = task
                .activity_id
                .map(ActivityId::from)
                .ok_or("task missing activity id")?;
            let completion_token =
                CompletionToken::from_wire(&workflow_id, &activity_id, task.completion_token)
                    .map_err(|error| error.to_string())?;
            sink.complete_activity(ActivityCompletion {
                workflow_id,
                activity_id,
                run_id: None,
                completion_token,
                outcome: ActivityCompletionOutcome::Succeeded(Payload::new(
                    ContentType::Json,
                    br#"{"greeting":"hello"}"#.to_vec(),
                )),
            })
            .map_err(|error| error.to_string())
        });

        let dispatcher = Arc::new(
            WorkerActivityDispatcher::new(registry.clone(), "default", test_tracker())
                .with_pending(pending),
        );

        // greet_request() carries node: None — the unpinned path.
        let result = tokio::spawn(futures::future::lazy(move |_| {
            dispatcher.dispatch(greet_request())
        }))
        .await
        .map_err(|error| error.to_string())?;

        assert_eq!(result, Ok(r#"{"greeting":"hello"}"#.to_owned()));
        echo.await.map_err(|error| error.to_string())??;
        on_n1.deregister()?;
        Ok(())
    }
}