aion_server/worker/

liminal_transport.rs

//! Cross-node outbox dispatch over the liminal bus (LSUB push transport).
//!
//! # What this is (production push path)
//!
//! This module wires the durable outbox's fan-out dispatch over liminal to a
//! REAL remote aion worker and returns the worker's result through the existing
//! [`OutboxDeliveryCallback`](super::bridge::OutboxDeliveryCallback), behind the
//! `liminal-transport` Cargo feature and the `outbox.transport = liminal`
//! runtime flag. The aion-server HOSTS the liminal listener: a remote worker
//! connects IN and self-describes in-band, the server registers it in the SAME
//! connected-worker registry a gRPC worker joins, and a claimed row is PUSHED out
//! on the worker's existing connection (the LSUB-0 server-push primitive).
//!
//! # Routing (NSTQ-5 / NODE-5)
//!
//! A worker is selected by the row's `(namespace, task_queue, activity_type,
//! node)` pool key through the EXISTING registry `select_worker` — the same
//! selection the gRPC path uses, so routing semantics are shared. `activity_type`
//! is NOT a routing dimension at the wire: it rides inside the [`DispatchRequest`]
//! payload and is matched by the worker after delivery, exactly as the gRPC
//! registry pushes `activity_type` in the task body while selecting the worker by
//! pool key. See `docs/NAMESPACE-TASKQUEUE-SPLIT-DESIGN.md` §4.2. The
//! [`dispatch_channel_name`] derivation remains the single source of truth for the
//! pool-channel string, pinned for any future channel-subscription subscriber so
//! the two sides cannot drift.
//!
//! # The seams it implements
//!
//! - [`RegistryLiminalDispatch`] implements
//!   [`OutboxRowDispatch`](super::outbox_dispatcher::OutboxRowDispatch): for each
//!   claimed row it selects a worker from the connected-worker registry, pushes
//!   the [`DispatchRequest`] to that worker's liminal connection via its
//!   [`LiminalWorkerDelivery`], and re-enters the worker's [`DispatchResponse`]
//!   through the SAME [`LiminalCompletionSource`] / [`OutboxDeliveryCallback`] the
//!   gRPC completion path uses. A row that reaches no matching worker, or whose
//!   worker is not liminal-delivered, returns an error so the outbox's unchanged
//!   retry/backoff drives it — the same honest no-worker contract as the gRPC
//!   path.
//! - [`LiminalConnectionNotifier`] is the SERVER half of in-band registration:
//!   when a worker connects with a [`WorkerRegistration`](WireWorkerRegistration)
//!   the notifier inserts a [`WorkerDelivery::Liminal`] into the registry, and
//!   drops it on disconnect.
//! - [`LiminalCompletionSource`] maps a [`DispatchResponse`] onto the delivery
//!   callback, threading `run_id` end-to-end so the existing continue-as-new run
//!   gates apply unchanged.
//!
//! # The channel-subscription seam (documented, distinct from the push path)
//!
//! [`dispatch_channel_name`] derives the pool channel a `(namespace, task_queue)`
//! pool addresses, optionally pinned to a `node` (NODE-2/NODE-5). The production
//! path above does not publish to that channel — it pushes to a connected worker
//! the server already owns — but the derivation is retained as the pinned contract
//! any future channel-subscription transport MUST honour so the dispatcher and a
//! subscriber cannot drift:
//!
//! - An UNPINNED worker pool addressed `(namespace, task_queue)` subscribes to
//!   `dispatch_channel_name(namespace, task_queue, None)`.
//! - A NODE-PINNED dispatch (the row carries `Some(node)`) maps to
//!   `dispatch_channel_name(namespace, task_queue, Some(node))` — a DISTINCT
//!   channel that a worker on that node must ALSO subscribe to in order to serve
//!   pinned work; the unpinned channel alone never delivers a pinned dispatch.
//!
//! That is the single contract the seam must honour.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use aion_core::{ActivityId, ContentType, Payload, RunId, WorkflowId};
use aion_store::OutboxRow;
use async_trait::async_trait;
use liminal::protocol::WorkerRegistration as WireWorkerRegistration;
use liminal_sdk::{SchemaMetadata, SchemaValidate};
use liminal_server::ServerError as LiminalServerError;
use liminal_server::server::connection::{
    ConnectionNotifier, ConnectionSupervisor, PushReplyAwaiter,
};
use serde::{Deserialize, Serialize};

use super::bridge::OutboxDeliveryCallback;
use super::envelope::{CompletionFences, CompletionToken, idempotency_key};
use super::outbox_dispatcher::{DeliveryGate, OutboxRowDispatch};
use super::registry::{ConnectedWorkerRegistry, WorkerDelivery, WorkerHandle, WorkerRegistration};
use crate::error::ServerError;

/// Upper bound on how long a server-initiated intervention push waits for the
/// worker's correlated control-plane acknowledgement. Activity dispatch replies
/// use an unbounded held wait because an activity may legitimately run a while.
const PUSH_REPLY_TIMEOUT: Duration = Duration::from_secs(30);

/// Re-arm cadence for the engine-seam bridge's UNBOUNDED reply wait
/// ([`receive_bridge_reply`]). Each elapsed poll is a benign re-arm, never a
/// failure: the bridge dispatch contract imposes no activity timeout of its own
/// (agent-style activities legitimately run for over an hour), exactly like the
/// gRPC bridge's unbounded `recv`. Worker loss still terminates the wait
/// promptly — the awaiter wakes with the typed Disconnected error the moment the
/// connection closes.
const BRIDGE_REPLY_POLL: Duration = Duration::from_secs(1);

/// Wire request carrying one scheduled activity to a liminal worker.
///
/// Mirrors the dispatch half of the gRPC `ActivityTask`: the fields the worker
/// needs to execute the activity and to correlate its result back to the exact
/// execution (`workflow_id`, `ordinal`, `run_id`). `run_id` rides end-to-end so
/// the existing continue-as-new run gates hold over the liminal wire (the design
/// doc §3.3 requirement that `RunId` stays on the wire).
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct DispatchRequest {
    /// Activity type the worker must execute.
    pub activity_type: String,
    /// Workflow that scheduled this fan-out activity. Carried in its serde form
    /// so no fragile id parsing happens on the wire.
    pub workflow_id: WorkflowId,
    /// Pinned ordinal of this activity within the workflow's fan-out range.
    pub ordinal: u64,
    /// Run that dispatched this ordinal, when known (continue-as-new safety).
    pub run_id: Option<RunId>,
    /// Opaque execution-generation proof echoed verbatim by the worker.
    pub completion_token: String,
    /// Stable external-effect key for this run and action site.
    pub idempotency_key: String,
    /// Opaque activity input bytes (JSON-tagged on the aion side).
    pub input: Vec<u8>,
    /// One-based delivery attempt, mirroring the gRPC `ActivityTask.attempt`.
    /// The engine-seam bridge threads the real attempt so a retry executes with
    /// attempt-aware handler semantics identical to the gRPC transport; the
    /// outbox path stamps the row's stored zero-based attempt as one-based.
    /// Serde-defaulted to `1` so a frame from a pre-attempt server (or an old
    /// recorded frame) still decodes as a first delivery.
    #[serde(default = "first_attempt")]
    pub attempt: u32,
    /// Engine-provided routing/metadata labels, mirroring the gRPC
    /// `ActivityTask.labels`. Empty (the serde default) on the outbox path,
    /// which has no label source.
    #[serde(default)]
    pub labels: std::collections::BTreeMap<String, String>,
    /// The server's heartbeat window in milliseconds when this dispatch is
    /// tracked by the server's per-task liveness tracker (the engine-seam
    /// bridge path), or `0` when it is not (the outbox path, whose liveness
    /// backstop is its own retry loop). A non-zero window tells the worker to
    /// pump automatic liveness beats at a quarter-window cadence so the
    /// server's heartbeat sweeper never expires a healthy long-running
    /// activity — the exact liminal mirror of the gRPC worker's automatic
    /// liveness pump.
    #[serde(default)]
    pub heartbeat_window_ms: u64,
}

/// Serde default for [`DispatchRequest::attempt`]: a frame that predates the
/// attempt field is a first delivery.
const fn first_attempt() -> u32 {
    1
}

impl SchemaValidate for DispatchRequest {
    fn schema_metadata() -> SchemaMetadata {
        SchemaMetadata::new(
            "aion.outbox.dispatch.request",
            "1",
            br#"{"type":"object"}"#.as_slice(),
        )
    }
}

/// Wire response carrying one worker result back to the outbox.
///
/// Mirrors the completion half of the gRPC `ActivityResult`: the correlation ids
/// plus either a success result or a failure reason. `LiminalCompletionSource`
/// maps this onto the existing [`OutboxDeliveryCallback`].
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct DispatchResponse {
    /// Workflow the completion belongs to.
    pub workflow_id: WorkflowId,
    /// Pinned ordinal the completion correlates against.
    pub ordinal: u64,
    /// Run that issued the dispatch, echoed back for the run gate.
    pub run_id: Option<RunId>,
    /// Opaque execution-generation proof echoed from the request.
    pub completion_token: String,
    /// Worker outcome: `Ok(result)` or `Err(reason)`.
    pub outcome: Result<String, String>,
}

impl SchemaValidate for DispatchResponse {
    fn schema_metadata() -> SchemaMetadata {
        SchemaMetadata::new(
            "aion.outbox.dispatch.response",
            "1",
            br#"{"type":"object"}"#.as_slice(),
        )
    }
}

/// Wire request carrying one neutral mid-run intervention command to a liminal
/// worker (NOI-6, §6.2).
///
/// Rides the SAME liminal server-push channel as [`DispatchRequest`], distinguished
/// on the wire by its unique required `intervention` field — a plain
/// [`DispatchRequest`] has no such field, so the worker demuxes the two by which
/// one deserializes. The whole envelope is neutral: it carries an
/// [`InterventionCommand`], never a harness type. Field-for-field mirrored by the
/// worker's `liminal::InterventionRequest`.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct InterventionRequest {
    /// The neutral command to route to the worker owning the target attempt.
    pub intervention: aion_core::InterventionCommand,
}

impl SchemaValidate for InterventionRequest {
    fn schema_metadata() -> SchemaMetadata {
        SchemaMetadata::new(
            "aion.intervention.request",
            "1",
            br#"{"type":"object"}"#.as_slice(),
        )
    }
}

/// Wire response carrying the worker's neutral intervention ack back to the server
/// (NOI-6). Field-for-field mirrored by the worker's `liminal::InterventionReply`.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct InterventionReply {
    /// The neutral applied/gated/stale outcome the operator receives.
    pub outcome: aion_core::InterventionOutcome,
}

impl SchemaValidate for InterventionReply {
    fn schema_metadata() -> SchemaMetadata {
        SchemaMetadata::new(
            "aion.intervention.reply",
            "1",
            br#"{"type":"object"}"#.as_slice(),
        )
    }
}

/// Reserved liminal channel a worker publishes automatic liveness beats on.
///
/// The liminal wire has no gRPC-style heartbeat frame, so per-task liveness
/// rides a reserved publish channel exactly as the observability transcript
/// does: the worker's runtime pumps a [`WorkerLivenessBeat`] per in-flight
/// tracked dispatch at a quarter-window cadence, and the server's
/// [`LiminalConnectionNotifier`] consumes the channel and refreshes the shared
/// [`HeartbeatTracker`] — so the #176 expiry sweeper is genuinely
/// transport-agnostic: a healthy liminal worker running a long activity is
/// never falsely expired, and a wedged one (which stops pumping) still is.
/// Mirrored byte-for-byte by the worker crate's constant of the same name.
pub const WORKER_LIVENESS_CHANNEL: &str = "aion.worker.liveness";

/// Reserved liminal channel a worker announces its intervention capabilities on.
///
/// The in-band [`WireWorkerRegistration`] frame is a published liminal protocol
/// type and cannot carry aion-level capability metadata, so a worker whose
/// harness supports interventions publishes a [`WorkerCapabilitiesAnnouncement`]
/// here immediately after registering (once per connection, so a redialed
/// worker re-announces). The notifier consumes the channel and applies the
/// announcement to the registered handle — the set the intervention router
/// gates on and the ops console's live-attempts enumeration reports. Mirrored
/// byte-for-byte by the worker crate's constant of the same name.
pub const WORKER_CAPABILITIES_CHANNEL: &str = "aion.worker.capabilities";

/// Wire announcement of a worker's advertised intervention capabilities.
///
/// Field-for-field mirror of the worker crate's `WorkerCapabilitiesAnnouncement`
/// (the same cross-crate contract the dispatch/response pairs pin). The worker
/// is identified by its CONNECTION (the publish's pid resolves the registered
/// worker), never by wire-supplied identity, so an announcement can only ever
/// apply to the worker that sent it.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerCapabilitiesAnnouncement {
    /// The neutral intervention primitives the worker's harness supports.
    pub capabilities: aion_core::InterventionCapabilities,
}

/// Wire liveness beat for one in-flight dispatch (the liminal mirror of the
/// gRPC `Heartbeat` frame, liveness-only — progress payloads are not carried).
///
/// Field-for-field mirror of the worker crate's `WorkerLivenessBeat` (same
/// serde field names + `aion-core` id types), the same cross-crate contract the
/// dispatch/response pairs pin. The worker is identified by its CONNECTION (the
/// publish's pid resolves the registered worker), never by wire-supplied
/// identity, so a beat can only ever refresh tasks of the worker that sent it.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerLivenessBeat {
    /// Workflow owning the in-flight activity being kept alive.
    pub workflow_id: WorkflowId,
    /// Pinned ordinal of the in-flight activity being kept alive.
    pub ordinal: u64,
}

/// Builds the wire request for one claimed outbox row.
///
/// Kept free-standing (not a method) so both the dispatch path and tests build
/// the request the same way.
#[must_use]
pub fn request_for_row(
    row: &OutboxRow,
    run_id: &RunId,
    completion_token: &CompletionToken,
) -> DispatchRequest {
    let activity_id = ActivityId::from_sequence_position(row.ordinal);
    DispatchRequest {
        activity_type: row.activity_type.clone(),
        workflow_id: row.workflow_id.clone(),
        ordinal: row.ordinal,
        run_id: Some(run_id.clone()),
        completion_token: completion_token.as_str().to_owned(),
        idempotency_key: idempotency_key(&row.workflow_id, run_id, &activity_id),
        input: row.input.bytes().to_vec(),
        // The stored zero-based attempt is stamped one-based on the wire (zero
        // is malformed), exactly as the gRPC outbox arm's `to_scheduled` does.
        attempt: row.attempt.saturating_add(1),
        // Outbox rows carry no engine labels (the gRPC arm sends empty too).
        labels: std::collections::BTreeMap::new(),
        // Outbox dispatches are not tracked by the server's per-task liveness
        // tracker — the outbox retry loop is their liveness backstop — so no
        // window is assigned and the worker does not pump beats for them.
        heartbeat_window_ms: 0,
    }
}

/// The single reserved character that separates channel segments. Because
/// `namespace`/`task_queue` are free-form, any occurrence of this byte INSIDE a
/// segment must be escaped so it cannot be mistaken for the segment boundary.
const SEGMENT_SEPARATOR: char = '.';

/// The escape character used by [`encode_segment`]. It must itself be escaped so
/// the encoding stays injective (otherwise `%2E` as a literal field value would
/// collide with an encoded `.`).
const SEGMENT_ESCAPE: char = '%';

/// Percent-encodes the two reserved characters (`.` and `%`) inside one channel
/// segment so distinct segment values can never collide across the join.
///
/// This is a minimal, deterministic, per-segment escape: a literal `.` becomes
/// `%2E` and a literal `%` becomes `%25`; every other byte (including the empty
/// string) passes through unchanged. Because both the separator AND the escape
/// char are encoded, the mapping `value -> encoded` is injective: it is exactly
/// reversible by replacing `%2E -> .` and `%25 -> %`, so two distinct values
/// can never encode to the same string. Dot-free, percent-free inputs (the
/// normal case, e.g. `"remote"`, `"gpu"`) are returned byte-for-byte unchanged,
/// so existing channels are stable.
fn encode_segment(segment: &str) -> String {
    // Fast path: nothing reserved, return an owned copy unchanged.
    if !segment.contains([SEGMENT_SEPARATOR, SEGMENT_ESCAPE]) {
        return segment.to_owned();
    }
    let mut encoded = String::with_capacity(segment.len());
    for ch in segment.chars() {
        match ch {
            // Encode the escape char FIRST so an already-present `%` cannot be
            // confused with one we introduce for the separator.
            SEGMENT_ESCAPE => encoded.push_str("%25"),
            SEGMENT_SEPARATOR => encoded.push_str("%2E"),
            other => encoded.push(other),
        }
    }
    encoded
}

/// Derives the liminal dispatch channel for a worker pool addressed
/// `(namespace, task_queue)`, optionally pinned to a specific `node`.
///
/// This is the **single, total source of truth** for the channel string: every
/// site that needs the channel a `(namespace, task_queue[, node])` pool
/// dispatches to — both this dispatcher and any future worker-pool subscription
/// side — MUST call this function so the two sides cannot drift. The format is
/// `"aion.dispatch.{namespace}.{task_queue}"` for an unpinned dispatch and
/// `"aion.dispatch.{namespace}.{task_queue}.{node}"` when a `node` is pinned;
/// each `{segment}` is independently passed through [`encode_segment`].
///
/// # The subscriber contract (the seam this function pins, NODE-5 / 13-x)
///
/// The subscriber side remains the documented seam (it does not exist yet; 13-0
/// uses liminal's in-server echo responder). The contract both sides MUST honour:
///
/// - An **unpinned** worker pool addressed `(namespace, task_queue)` subscribes
///   to `dispatch_channel_name(namespace, task_queue, None)` and receives every
///   unpinned dispatch for that pool.
/// - A **node-pinned** dispatch (the row carries `Some(node)`) goes to
///   `dispatch_channel_name(namespace, task_queue, Some(node))`, a DISTINCT
///   channel. A worker running on that node which is meant to serve pinned work
///   for the pool MUST ALSO subscribe to that node-specific channel — the
///   `None` channel alone will never deliver a node-pinned dispatch to it.
///
/// Because the `None` channel and any `Some(node)` channel are distinct strings,
/// a node-pinned dispatch never reaches an unpinned-only subscriber and vice
/// versa; node isolation is therefore enforced by the channel string itself.
///
/// # Injectivity (why the per-segment encode matters)
///
/// `namespace`, `task_queue` and `node` are all free-form (the design forbids
/// preset categories), so a raw `format!` would be NON-injective: a `.` inside
/// any field bleeds across the separator and pools the design declares disjoint
/// collide onto one channel — e.g. `("a.b", "c", None)` and `("a", "b.c", None)`
/// would both yield `aion.dispatch.a.b.c`, a cross-pool leak on the very
/// isolation dimension this routing exists to keep separate. Encoding each
/// segment independently (the separator `.` and the escape `%` are escaped within
/// a segment) makes the map from `(namespace, task_queue, node)` to channel
/// string injective: distinct triples always yield distinct channels, ACROSS
/// segment counts too. The node segment is appended only for `Some(node)`, and
/// because no encoded segment can contain a bare separator, a 2-segment channel
/// (unpinned) can never be confused with a 3-segment channel (pinned) — e.g.
/// `("a", "b", Some("c"))` and `("a", "b.c", None)` stay distinct, as do
/// `("a.b", "c", None)` and `("a", "b", Some("c"))`.
///
/// `activity_type` is deliberately NOT part of the channel: it is *what to run*,
/// matched by the worker after delivery (it rides inside [`DispatchRequest`]),
/// not *which pool* — see `docs/NAMESPACE-TASKQUEUE-SPLIT-DESIGN.md` §4.2. The
/// function is total (defined for every input) and stable (the same
/// `(namespace, task_queue, node)` always yields the same channel).
#[must_use]
pub fn dispatch_channel_name(namespace: &str, task_queue: &str, node: Option<&str>) -> String {
    let namespace = encode_segment(namespace);
    let task_queue = encode_segment(task_queue);
    match node {
        Some(node) => {
            let node = encode_segment(node);
            format!("aion.dispatch.{namespace}.{task_queue}.{node}")
        }
        None => format!("aion.dispatch.{namespace}.{task_queue}"),
    }
}

/// Derives the liminal dispatch channel for a claimed outbox row.
///
/// Thin wrapper over [`dispatch_channel_name`] reading the row's durable
/// `(namespace, task_queue)` (NSTQ-2 columns) and its optional `node` (NODE-2):
/// when `row.node` is `Some`, the row dispatches to the node-pinned sub-channel;
/// when `None`, it derives the byte-identical unpinned channel. Kept
/// free-standing so the dispatch path and tests derive the row's channel
/// identically.
#[must_use]
pub fn channel_for_row(row: &OutboxRow) -> String {
    dispatch_channel_name(&row.namespace, &row.task_queue, row.node.as_deref())
}

/// Wraps a reason in the existing worker-dispatch error so a non-deliverable
/// dispatch drives the outbox's unchanged retry/backoff/dead-letter path. The
/// row-derived `channel` is surfaced as the `activity_type` field for operator
/// diagnostics (the field is a free-form context string on this transport's
/// error).
fn dispatch_error(channel: &str, reason: String) -> ServerError {
    ServerError::WorkerDispatch {
        namespace: "liminal".to_owned(),
        activity_type: channel.to_owned(),
        reason,
    }
}

/// Receives one worker result over liminal and re-enters it into aion.
///
/// Holds the installed [`OutboxDeliveryCallback`] (the same prod
/// `ServerOutboxDeliveryCallback` the gRPC completion path uses) and maps a
/// [`DispatchResponse`] onto it, threading `run_id` so the continue-as-new run
/// gates apply unchanged.
pub struct LiminalCompletionSource {
    callback: Arc<dyn OutboxDeliveryCallback>,
    completion_fences: CompletionFences,
}

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

impl LiminalCompletionSource {
    /// Build a completion source over the shared outbox delivery callback.
    #[must_use]
    pub fn new(callback: Arc<dyn OutboxDeliveryCallback>) -> Self {
        Self {
            callback,
            completion_fences: CompletionFences::default(),
        }
    }

    /// Share the authoritative generation registry used by dispatch.
    #[must_use]
    pub fn with_completion_fences(mut self, completion_fences: CompletionFences) -> Self {
        self.completion_fences = completion_fences;
        self
    }

    /// Re-enter one worker result into aion through the delivery callback.
    ///
    /// Returns the callback's `bool`: `true` when delivered to a live run,
    /// `false` when no run is live (the expected stale-completion drop that
    /// recovery re-arms). A success outcome routes to `deliver_completion`; a
    /// failure outcome to `deliver_failure`.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError`] when the response carries an unparseable id or
    /// the engine rejects the delivery.
    pub fn deliver(&self, response: &DispatchResponse) -> Result<bool, ServerError> {
        let run_id = response.run_id.as_ref().ok_or_else(|| {
            dispatch_error(
                "liminal completion",
                "activity result run id is missing; refusing run-gate bypass".to_owned(),
            )
        })?;
        let activity_id = ActivityId::from_sequence_position(response.ordinal);
        let completion_token = CompletionToken::from_wire(
            &response.workflow_id,
            &activity_id,
            response.completion_token.clone(),
        )?;
        if let Err(error) =
            self.completion_fences
                .accept(&response.workflow_id, &activity_id, &completion_token)
        {
            tracing::warn!(
                workflow_id = %response.workflow_id,
                %activity_id,
                %error,
                "liminal completion fence rejected a late or stale reply"
            );
            return Err(error);
        }
        match &response.outcome {
            Ok(result) => self.callback.deliver_completion(
                &response.workflow_id,
                &activity_id,
                Some(run_id),
                result.clone(),
            ),
            Err(reason) => self.callback.deliver_failure(
                &response.workflow_id,
                &activity_id,
                Some(run_id),
                reason.clone(),
            ),
        }
    }
}

/// Rebuilds the activity input payload from the wire request.
///
/// The aion side tags activity input as JSON; the wire carries the raw bytes, so
/// a worker (or the test responder standing in for one) reconstructs the typed
/// [`Payload`] with the JSON content type.
#[must_use]
pub fn payload_from_request(request: &DispatchRequest) -> Payload {
    Payload::new(ContentType::Json, request.input.clone())
}

/// Delivery handle for a liminal-connected worker held in the worker registry.
///
/// A worker that connects over liminal is a first-class registry member selected
/// the SAME way as a gRPC worker (`select_worker` on `(namespace, task_queue,
/// node)`); this is the delivery leg the registry holds for it. It pairs the
/// [`ConnectionSupervisor`] that owns the worker's connection with that
/// connection's beamr `pid`, so [`Self::dispatch_held`] can push a
/// [`DispatchRequest`] out on the worker's existing socket (the LSUB-0
/// server-push primitive) and block for the correlated [`DispatchResponse`].
#[derive(Clone)]
pub struct LiminalWorkerDelivery {
    supervisor: ConnectionSupervisor,
    pid: u64,
}

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

impl LiminalWorkerDelivery {
    /// Build a delivery handle for the worker reachable on connection `pid`
    /// through `supervisor`.
    #[must_use]
    pub const fn new(supervisor: ConnectionSupervisor, pid: u64) -> Self {
        Self { supervisor, pid }
    }

    /// The connection pid this worker is addressed on.
    #[must_use]
    pub const fn pid(&self) -> u64 {
        self.pid
    }

    /// Push one outbox dispatch and wait without an activity-duration bound,
    /// abandoning only when `keep_waiting` becomes false at a poll boundary.
    ///
    /// # Errors
    ///
    /// Returns the same typed push, receive, and decode failures as the shared
    /// bridge wait.
    pub fn dispatch_held(
        &self,
        request: &DispatchRequest,
        keep_waiting: impl Fn() -> bool,
    ) -> Result<Option<DispatchResponse>, ServerError> {
        let awaiter = self.push_dispatch(request)?;
        receive_bridge_reply(&awaiter, keep_waiting)
    }

    /// Serialize and push one dispatch out on the worker's connection, returning
    /// the awaiter for its correlated reply — the shared push half of both
    /// unbounded waits: [`Self::dispatch_held`] on the outbox path and the
    /// engine-seam bridge dispatcher both use [`receive_bridge_reply`], matching
    /// the gRPC bridge contract. One frame format, one push primitive, one held
    /// wait policy.
    ///
    /// DELIBERATELY NO REPLY DEADLINE, unlike the liveness and intervention
    /// pushes. A dispatch's reply IS the activity's completion, so its lifetime
    /// is the activity's — `remote_gates.awl` declares legs at `timeout 45m`.
    /// [`receive_bridge_reply`] re-arms on every poll timeout and waits
    /// unbounded by contract, so this push NEVER abandons its slot and cannot
    /// leak one: the slot is held for exactly as long as work is outstanding
    /// and resolves when the reply lands. Attaching a deadline here would kill
    /// long activities, and would be a regression, not a hardening. The leak
    /// fixed on the other two sites came from ABANDONING a no-deadline slot,
    /// which this site never does.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::WorkerBusy`] when liminal's typed connection cap
    /// refuses admission, [`ServerError::WorkerConnectionLost`] for every other
    /// push-enqueue failure, and [`ServerError::WorkerDispatch`] when the request
    /// cannot be serialized.
    pub(crate) fn push_dispatch(
        &self,
        request: &DispatchRequest,
    ) -> Result<PushReplyAwaiter, ServerError> {
        let payload = serde_json::to_vec(request).map_err(|error| {
            dispatch_error("liminal-push", format!("request serialize failed: {error}"))
        })?;
        self.supervisor
            .push_to_connection(self.pid, payload)
            .map_err(|error| classify_push_error(&error))
    }

    /// Push already-encoded `payload` out on the worker's connection with an
    /// explicit reply `deadline`, returning the awaiter for its correlated
    /// reply.
    ///
    /// The raw push primitive behind the typed pushes above, used by the
    /// connection liveness probe
    /// ([`LivenessProbe`](super::liminal_liveness::LivenessProbe)) so the
    /// dead-man switch rides the SAME server-push leg a dispatch does — a ping
    /// that reaches the worker proves the exact path a dispatch would take, not
    /// a parallel one that could be healthy while the real one is not.
    ///
    /// The deadline is LOAD-BEARING and is why this is not
    /// `push_to_connection`. Riding the dispatch leg means sharing its bounded
    /// §5 `max_pending_pushes_per_connection` admission, and a no-deadline slot
    /// is reclaimed only by a consumed reply or a connection close — never by
    /// the caller giving up. A probe that abandons an unanswered ping every
    /// cadence therefore LEAKS one slot per round until the cap is exhausted,
    /// at which point no dispatch, intervention or ping can be pushed on that
    /// connection again. Measured live on run `dfd2117c`: 32 abandoned pings,
    /// then permanent refusal, ≈240 s from first blocked ping. The probe built
    /// to prove the dispatch path works is what destroyed it. With a deadline,
    /// expiry removes the slot and RELEASES its cap admission.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::WorkerBusy`] when liminal's typed connection cap
    /// refuses admission, and [`ServerError::WorkerConnectionLost`] for every
    /// other push-enqueue failure.
    pub fn push_payload_with_deadline(
        &self,
        payload: Vec<u8>,
        deadline: Duration,
    ) -> Result<PushReplyAwaiter, ServerError> {
        self.supervisor
            .push_to_connection_with_deadline(self.pid, payload, deadline)
            .map_err(|error| classify_push_error(&error))
    }

    /// Push one neutral intervention command out on the worker's connection and
    /// block for its correlated ack reply (NOI-6, §6.2).
    ///
    /// Uses the same connection as activity dispatch but carries an
    /// [`InterventionRequest`] and decodes an [`InterventionReply`], so it rides the SAME server-push
    /// channel as an activity dispatch. The push is a blocking, thread-based liminal
    /// call; the async router runs it off the runtime.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::WorkerConnectionLost`] when the worker connection was
    /// gone at push time or closed before replying (so the router surfaces the
    /// too-late no-op); returns [`ServerError::WorkerDispatch`] when the request
    /// cannot be serialized, the reply times out, or the reply cannot be decoded.
    pub fn push_intervention(
        &self,
        request: &InterventionRequest,
    ) -> Result<InterventionReply, ServerError> {
        let payload = serde_json::to_vec(request).map_err(|error| {
            dispatch_error(
                "liminal-push",
                format!("intervention serialize failed: {error}"),
            )
        })?;
        // Deadline-bounded, and it must be: this wait ABANDONS on timeout (the
        // `receive` error below is terminal, not a re-arm), so a no-deadline
        // slot would be held until the connection closed and would count
        // against the connection's §5 pending-push cap forever. The deadline
        // matches the wait, so an intervention that goes unanswered releases
        // its admission instead of leaking it. Contrast `push_dispatch`, whose
        // wait re-arms and is unbounded by contract, and which therefore
        // correctly takes no deadline.
        let awaiter = self
            .supervisor
            .push_to_connection_with_deadline(self.pid, payload, PUSH_REPLY_TIMEOUT)
            .map_err(|error| {
                ServerError::worker_connection_lost(
                    "liminal-push",
                    format!("push intervention to worker failed: {error}"),
                )
            })?;
        let reply = awaiter.receive(PUSH_REPLY_TIMEOUT).map_err(|error| {
            if is_connection_closed_reply_error(&error) {
                ServerError::worker_connection_lost(
                    "liminal-push",
                    format!("worker connection closed before intervention ack: {error}"),
                )
            } else {
                dispatch_error("liminal-push", format!("intervention ack failed: {error}"))
            }
        })?;
        serde_json::from_slice(&reply).map_err(|error| {
            dispatch_error(
                "liminal-push",
                format!("intervention ack decode failed: {error}"),
            )
        })
    }
}

/// Decodes one correlated reply payload as a [`DispatchResponse`].
///
/// Shared by the outbox wait ([`LiminalWorkerDelivery::dispatch_held`]) and the
/// bridge wait ([`receive_bridge_reply`]) so the two paths can never diverge on
/// the wire's reply shape.
fn decode_dispatch_response(reply: &[u8]) -> Result<DispatchResponse, ServerError> {
    serde_json::from_slice(reply).map_err(|error| {
        dispatch_error(
            "liminal-push",
            format!("worker reply decode failed: {error}"),
        )
    })
}

/// Blocks for the correlated reply to an engine-seam BRIDGE dispatch push, with
/// the bridge's UNBOUNDED wait contract: the engine imposes no activity timeout
/// of its own, so an elapsed [`BRIDGE_REPLY_POLL`] merely re-arms the wait — the
/// exact liminal mirror of the gRPC bridge's unbounded `recv`, which is released
/// only by a completion or by stream teardown. The wait terminates on exactly:
///
/// - **the reply** — decoded as the worker's [`DispatchResponse`] and returned
///   as `Ok(Some(response))`. A reply already buffered when a poll fires always
///   wins over abandonment: the awaiter is drained before `keep_waiting` runs;
/// - **abandonment** — `keep_waiting()` returned `false` at a poll boundary
///   (the bridge passes "is this dispatch still tracked in-flight?"): the
///   dispatch was resolved by another path (expiry sweep, shutdown drain, or a
///   cleanup), so the wait returns `Ok(None)` and the router thread exits
///   instead of parking on the connection indefinitely;
/// - **worker loss** — the connection closed before replying: the awaiter wakes
///   PROMPTLY with liminal's typed Disconnected error, surfaced as
///   [`ServerError::WorkerConnectionLost`] so the bridge reports the same
///   TRANSPORT-loss class the gRPC teardown sweep does;
/// - **an unrecognized receive fault or a decode failure** — surfaced as
///   [`ServerError::WorkerDispatch`].
///
/// Runs on a dedicated bridge reply thread, never on an async runtime worker.
///
/// # Errors
///
/// Returns [`ServerError::WorkerConnectionLost`] on worker loss and
/// [`ServerError::WorkerDispatch`] on a receive fault or reply decode failure.
pub(crate) fn receive_bridge_reply(
    awaiter: &PushReplyAwaiter,
    keep_waiting: impl Fn() -> bool,
) -> Result<Option<DispatchResponse>, ServerError> {
    loop {
        match awaiter.receive(BRIDGE_REPLY_POLL) {
            Ok(reply) => return decode_dispatch_response(&reply).map(Some),
            // A bare poll timeout re-arms the wait while the dispatch is still
            // outstanding (the wait is unbounded by contract; see the bridge
            // module docs), and ends it cleanly once the dispatch was resolved
            // elsewhere — the poll cadence bounds the router thread's lifetime
            // to one poll past that resolution.
            Err(LiminalServerError::PushReplyTimeout { .. }) => {
                if !keep_waiting() {
                    return Ok(None);
                }
            }
            Err(error) if is_connection_closed_reply_error(&error) => {
                return Err(ServerError::worker_connection_lost(
                    "liminal-push",
                    format!("worker connection closed before reply: {error}"),
                ));
            }
            Err(error) => {
                return Err(dispatch_error(
                    "liminal-push",
                    format!("worker reply failed: {error}"),
                ));
            }
        }
    }
}

/// Returns true when a liminal push-reply error is the *Disconnected* case (the
/// worker's connection closed before it replied), as opposed to a genuine reply
/// timeout (the worker is alive but slow).
///
/// Liminal returns these as distinct TYPED variants —
/// `ServerError::PushReplyDisconnected` vs `PushReplyTimeout` (see `liminal-server`
/// `supervisor.rs` `PushReplyAwaiter::receive`) — so this is a type match, not a
/// message-text match: a worker that DIED (connection-lost, fast failover) is told
/// apart from one that is merely SLOW (genuine timeout, normal backoff) by variant.
fn is_connection_closed_reply_error(error: &LiminalServerError) -> bool {
    matches!(error, LiminalServerError::PushReplyDisconnected { .. })
}

/// Classify a typed liminal push refusal without relying on error text.
fn classify_push_error(error: &LiminalServerError) -> ServerError {
    if is_connection_cap_error(error) {
        ServerError::worker_busy(
            "liminal-push",
            format!("worker connection push cap reached: {error}"),
        )
    } else {
        ServerError::worker_connection_lost(
            "liminal-push",
            format!("push to worker failed: {error}"),
        )
    }
}

/// Liminal's typed admission refusal when a connection has too many held pushes.
fn is_connection_cap_error(error: &LiminalServerError) -> bool {
    matches!(error, LiminalServerError::ConnectionCapReached { .. })
}

/// Cross-node [`OutboxRowDispatch`] that selects a liminal worker from the
/// connected-worker registry and pushes the row to it.
///
/// This is the LSUB-1 server-side composition: for each claimed row it selects a
/// worker by the row's `(namespace, task_queue, activity_type, node)` via the
/// EXISTING registry `select_worker` (the same selection the gRPC path uses, so
/// routing semantics are shared), pushes the [`DispatchRequest`] to that worker's
/// liminal connection via its [`LiminalWorkerDelivery`], and re-enters the
/// worker's [`DispatchResponse`] through the SAME [`LiminalCompletionSource`] /
/// [`OutboxDeliveryCallback`] the existing completion path uses. A row that
/// reaches no matching worker, or whose worker is not liminal-delivered, returns
/// an error so the outbox's unchanged retry/backoff drives it — the same honest
/// no-worker contract as the gRPC path.
pub struct RegistryLiminalDispatch {
    registry: ConnectedWorkerRegistry,
    completion: LiminalCompletionSource,
    delivery_gate: DeliveryGate,
    /// Optional short-TTL per-namespace placement cache (Control-Plane Phase 2,
    /// P2-P3), the SAME cache the gRPC
    /// [`WorkerOutboxDispatch`](crate::worker::WorkerOutboxDispatch) is given. When
    /// present, an UNPINNED row (`row.node == None`) whose namespace placement is
    /// `Prefer{L}` selects an L-labelled worker and spills to any live worker when
    /// none is up, via the SHARED
    /// [`preferred_node_order`](crate::worker::preferred_node_order). When absent
    /// (the default, every pre-Phase-2 construction and test) selection is
    /// byte-identical to before: one `select_worker` off the row's own node.
    /// Placement is NEVER stamped back onto the row — it is consulted only here,
    /// in this non-replayed dispatcher, for worker selection.
    placement_cache: Option<crate::worker::PlacementCache>,
    /// Optional NOI-6 `attempt -> owning-worker` back-index. When installed via
    /// [`Self::with_attempt_owners`], each dispatched agent attempt binds its
    /// `(workflow, activity, attempt)` to the selected worker here BEFORE the push and
    /// releases it after the reply, so the server's intervention router resolves the
    /// CURRENT owner of a live attempt. `None` (the default, and every non-agent
    /// deployment) skips the binding — intervention is simply never offered.
    attempt_owners: Option<super::intervention::AttemptOwnerIndex>,
}

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

/// One selection attempt: the worker it found, and the node filters it WALKED.
///
/// The tiers travel WITH the outcome because a miss can only be explained over
/// the filters selection actually applied. Recomputing them at the refusal site
/// is how the two answers drift: the recomputation has no access to the
/// namespace placement the selection resolved, so it silently substitutes the
/// row's own node and describes a different fleet from the one selection saw.
struct SelectionAttempt {
    /// The selected worker, or `None` when no tier held an eligible one.
    worker: Option<WorkerHandle>,
    /// The ordered node filters walked, in the order they were tried. For a
    /// `Pinned{L}` namespace these are the required labels and nothing else —
    /// there is deliberately no `None` any-node tier to fall back on.
    tiers: Vec<Option<String>>,
}

impl RegistryLiminalDispatch {
    /// Build a registry-backed liminal dispatch that re-enters worker results
    /// through `callback` (the shared `ServerOutboxDeliveryCallback`).
    #[must_use]
    pub fn new(
        registry: ConnectedWorkerRegistry,
        callback: Arc<dyn OutboxDeliveryCallback>,
        delivery_gate: DeliveryGate,
    ) -> Self {
        Self {
            registry,
            completion: LiminalCompletionSource::new(callback),
            delivery_gate,
            placement_cache: None,
            attempt_owners: None,
        }
    }

    /// Share completion-generation authority with result ingestion.
    #[must_use]
    pub fn with_completion_fences(mut self, completion_fences: CompletionFences) -> Self {
        self.completion = self.completion.with_completion_fences(completion_fences);
        self
    }

    /// Install the NOI-6 attempt-owner back-index so each dispatched attempt binds
    /// its owning worker for the intervention router to resolve (NOI-6).
    ///
    /// The SAME index the server's [`InterventionRouter`](super::intervention::InterventionRouter)
    /// resolves through (from `ServerState::attempt_owners`), so a pushed command
    /// reaches the worker this dispatcher sent the attempt to. Pure builder addition:
    /// without it, no ownership is recorded and the router finds no owner (the
    /// too-late no-op), exactly as before.
    #[must_use]
    pub fn with_attempt_owners(
        mut self,
        attempt_owners: super::intervention::AttemptOwnerIndex,
    ) -> Self {
        self.attempt_owners = Some(attempt_owners);
        self
    }

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

    /// Select the liminal worker for `row`, applying the SHARED placement decision
    /// for an UNPINNED row when a placement cache is attached — the exact gRPC
    /// semantics ([`worker_selection_for`](crate::worker::worker_selection_for)):
    /// `Prefer{L}` spills to any live worker, `Pinned{L}` requires an L-labelled
    /// worker and NEVER spills to a node=None any-worker.
    ///
    /// A per-activity authored pin (`row.node == Some(N)`) ALWAYS wins and is
    /// selected off the row's own node, untouched by placement — exactly the gRPC
    /// composition rule. Without a cache (or with a pinned row) this collapses to
    /// the single `select_worker` off the row's own node — the pre-Phase-2
    /// behaviour.
    ///
    /// For `Pinned{L}`, when no L-labelled worker is live this returns `Ok(None)` —
    /// NOT a spill to a node=None worker — so the [`OutboxRowDispatch`] surfaces the
    /// honest no-worker error and the outbox retries/stalls until an L-labelled
    /// worker returns, mirroring the gRPC wait-for-worker path exactly (both
    /// transports agree via [`WorkerSelection`](crate::worker::WorkerSelection)).
    async fn select_liminal_worker(
        &self,
        row: &OutboxRow,
    ) -> Result<SelectionAttempt, ServerError> {
        // A pinned row or an absent cache: one selection off the row's own node.
        let (Some(cache), None) = (&self.placement_cache, &row.node) else {
            let tiers = vec![row.node.clone()];
            let worker = self.registry.select_worker(
                &row.namespace,
                &row.task_queue,
                &row.activity_type,
                row.node.as_deref(),
            )?;
            return Ok(SelectionAttempt { worker, tiers });
        };
        // Unpinned + placement-aware: resolve the shared selection decision, so this
        // liminal path and the gRPC path can never diverge on Prefer-vs-Pinned. The
        // row's `node` is never mutated — selection is a pure dispatch-time input.
        let placement = cache.placement(&row.namespace).await;
        let tiers: Vec<Option<String>> = match crate::worker::worker_selection_for(&placement) {
            // Prefer/Unplaced: walk the prefer-then-spill tiers (the `None` spill is
            // always last), stopping at the first tier with a live worker.
            crate::worker::WorkerSelection::PreferTiers(tiers) => tiers,
            // Pinned{L}: try ONLY the required labels — no `None` spill. When none is
            // live, select nothing so the caller retries/stalls (never any-node).
            crate::worker::WorkerSelection::Required(required) => {
                required.into_iter().map(Some).collect()
            }
        };
        let worker = self.select_over_tiers(row, tiers.iter().map(Option::as_deref))?;
        Ok(SelectionAttempt { worker, tiers })
    }

    /// Select the first live worker over an ordered sequence of node filters,
    /// returning `Ok(None)` when no filter matches a live worker. Shared by the
    /// `Prefer` (tiers end in a `None` spill) and `Pinned` (required labels only,
    /// no spill) selection arms so both walk the registry identically.
    fn select_over_tiers<'a>(
        &self,
        row: &OutboxRow,
        tiers: impl Iterator<Item = Option<&'a str>>,
    ) -> Result<Option<WorkerHandle>, ServerError> {
        for tier in tiers {
            let selected = self.registry.select_worker(
                &row.namespace,
                &row.task_queue,
                &row.activity_type,
                tier,
            )?;
            if selected.is_some() {
                return Ok(selected);
            }
        }
        Ok(None)
    }

    /// Why this row found no worker — stated as the fact it IS (#197 R3).
    ///
    /// A selection miss has two causes an operator must act on differently, and
    /// the single catch-all this replaces conflated them into a misdiagnosis:
    ///
    /// - **nobody is there.** No worker registered for the row's address. The
    ///   remedy is to start one.
    /// - **workers are there and none is currently eligible.** They registered
    ///   and the liveness verdict excludes them — each is either still serving
    ///   its opening probation (ordinary, self-clearing within a couple of
    ///   probe cadences) or has had its eligibility withdrawn (an incident, and
    ///   the WARN naming it is already in the log). The remedy is to read that
    ///   verdict, and NEVER to start another worker — a second one would serve
    ///   the same probation and change nothing.
    ///
    /// The distinction became newly visible when the probe learned to reach
    /// gRPC-delivered workers: before, no gRPC worker was ever pinged, so its
    /// exclusion was permanent and invisible behind "nobody is registered".
    ///
    /// `tiers` MUST be the sequence [`Self::select_liminal_worker`] actually
    /// walked, which is why it travels here from the selection rather than
    /// being re-derived. A `Pinned{L}` namespace walks the required labels and
    /// never spills to a `None` any-node tier; counting over the row's own node
    /// instead would match every worker in the pool and blame the miss on an
    /// unlabelled worker that was never a candidate — telling the operator not
    /// to start the labelled worker that is the only remedy.
    ///
    /// A census read failure falls back to the bare miss, LOUDLY. Losing the
    /// REASON must never suppress the refusal itself, and inventing an
    /// exclusion count this call could not read would be worse than saying
    /// less — but a poisoned registry lock is a corruption signal that must not
    /// leave only a quieter refusal behind as its trace.
    fn selection_miss_reason(&self, row: &OutboxRow, tiers: &[Option<String>]) -> String {
        let excluded = match self.registry.ineligible_workers_over_tiers(
            &row.namespace,
            &row.task_queue,
            &row.activity_type,
            tiers,
        ) {
            Ok(excluded) => excluded,
            Err(error) => {
                tracing::warn!(
                    %error,
                    namespace = %row.namespace,
                    task_queue = %row.task_queue,
                    activity_type = %row.activity_type,
                    "could not read the dispatch-eligibility census for a selection miss; this \
                     refusal falls back to the bare no-worker message and may therefore describe \
                     a pool whose workers are merely EXCLUDED as though it were empty"
                );
                0
            }
        };
        if excluded == 0 {
            return "no liminal worker registered for the row's pool".to_owned();
        }
        format!(
            "{excluded} worker(s) are registered for the row's pool but NONE is currently \
             dispatch-eligible: each is either serving its opening liveness probation or has had \
             its eligibility withdrawn. This is not an empty pool — starting another worker will \
             not help. Read the liveness verdict for these workers (the probation/withdrawal lines \
             carry the worker ids and the reason)"
        )
    }

    fn revoke_completion(
        &self,
        row: &OutboxRow,
        activity_id: &ActivityId,
        completion_token: &CompletionToken,
    ) -> Result<(), ServerError> {
        self.completion
            .completion_fences
            .revoke(&row.workflow_id, activity_id, completion_token)
    }
}

#[async_trait]
impl OutboxRowDispatch for RegistryLiminalDispatch {
    async fn dispatch(&self, row: &OutboxRow) -> Result<(), ServerError> {
        // Select the worker the SAME way the gRPC path does: by the row's
        // (namespace, task_queue, activity_type) pool key with the row's optional
        // node affinity, applying the SHARED `Prefer` two-tier spill for an
        // unpinned row when a placement cache is attached. No worker for the pool
        // => honest no-worker error => the outbox retries (never a false `done`).
        let attempt = self.select_liminal_worker(row).await?;
        let Some(worker) = attempt.worker else {
            return Err(dispatch_error(
                &channel_for_row(row),
                self.selection_miss_reason(row, &attempt.tiers),
            ));
        };

        let delivery = match worker.delivery() {
            WorkerDelivery::Liminal(delivery) => delivery.clone(),
            WorkerDelivery::Grpc(_) => {
                return Err(dispatch_error(
                    &channel_for_row(row),
                    "selected worker is not delivered over liminal".to_owned(),
                ));
            }
        };

        // The run is resolved BEFORE the owner binding, because the binding is
        // keyed on it: a run-blind key would let an intervention aimed at one
        // continue-as-new generation resolve another's worker. The refusal is
        // the pre-existing one — a row without a run cannot be dispatched at all
        // — simply moved ahead of the bind it now feeds.
        let activity_id = ActivityId::from_sequence_position(row.ordinal);
        let run_id = row.run_id.as_ref().ok_or_else(|| {
            dispatch_error(
                &channel_for_row(row),
                "activity run id is missing; refusing unfenced external effect".to_owned(),
            )
        })?;

        // NOI-6: bind this attempt's owner BEFORE the push, so an intervention that
        // races the dispatch resolves the worker. The guard releases on every exit
        // path (reply, error, panic) so the index never keeps a finished attempt.
        // The key mirrors the worker's execute-path stamp exactly: activity_id from
        // the ordinal, run_id from the row, attempt = the wire's one-based delivery
        // attempt (the same `request_for_row` stamp the worker echoes into its
        // session key). See `LiminalActivityWorker::execute` / `run_agent_dispatch`.
        let _owner_guard = self.attempt_owners.as_ref().map(|owners| {
            AttemptOwnerGuard::bind(
                owners.clone(),
                super::intervention::AttemptKey::new(
                    row.workflow_id.clone(),
                    run_id.clone(),
                    activity_id.clone(),
                    row.attempt.saturating_add(1),
                ),
                worker.id(),
            )
        });

        // Push the dispatch to the worker and block for its correlated reply. The
        // push is a blocking, thread-based liminal call; run it off the async
        // runtime so a long-running activity cannot starve a runtime worker.
        let completion_token = self
            .completion
            .completion_fences
            .issue(&row.workflow_id, &activity_id)?;
        let request = request_for_row(row, run_id, &completion_token);
        let gate = self.delivery_gate.clone();
        let dispatch_key = row.dispatch_key.clone();
        let registry = self.registry.clone();
        let worker_id = worker.id();
        let dispatched = tokio::task::spawn_blocking(move || {
            delivery.dispatch_held(&request, || {
                if gate.is_draining() || !gate.holds(&dispatch_key) {
                    return false;
                }
                match registry.worker_by_id(worker_id) {
                    Ok(Some(_)) => true,
                    Ok(None) => false,
                    Err(error) => {
                        tracing::warn!(
                            %error,
                            %dispatch_key,
                            "delivery wait could not verify worker registration; abandoning"
                        );
                        false
                    }
                }
            })
        })
        .await
        .map_err(|error| {
            dispatch_error(
                &channel_for_row(row),
                format!("dispatch task join failed: {error}"),
            )
        });
        let response = match dispatched {
            Ok(Ok(Some(response))) => response,
            Ok(Ok(None)) => {
                tracing::warn!(
                    dispatch_key = %row.dispatch_key,
                    workflow_id = %row.workflow_id,
                    %activity_id,
                    "delivery wait abandoned; late reply will be discarded"
                );
                self.revoke_completion(row, &activity_id, &completion_token)?;
                return Err(dispatch_error(
                    &channel_for_row(row),
                    "delivery wait abandoned before worker reply".to_owned(),
                ));
            }
            Ok(Err(error)) | Err(error) => {
                self.revoke_completion(row, &activity_id, &completion_token)?;
                return Err(error);
            }
        };

        // Re-enter the worker's result through the SAME completion path the gRPC
        // transport uses (terminal dedup in `record_fan_out_completion` applies
        // unchanged). The dispatch itself succeeded — the row's terminal state is
        // recorded by the completion callback, exactly as in the gRPC path.
        if let Err(error) = self.completion.deliver(&response) {
            self.revoke_completion(row, &activity_id, &completion_token)?;
            return Err(error);
        }
        Ok(())
    }
}

/// RAII guard that releases an [`AttemptOwnerIndex`](super::intervention::AttemptOwnerIndex)
/// binding when the dispatch resolves — on the reply, an error, or a panic — so
/// the back-index tracks exactly the attempts currently in flight (NOI-6).
///
/// Shared by both liminal dispatch arms: the outbox row wait holds it across
/// its blocking `dispatch` call, and the engine-seam bridge hands it to the
/// dispatch's reply-router thread, which drops it on every exit path.
pub(crate) struct AttemptOwnerGuard {
    owners: super::intervention::AttemptOwnerIndex,
    key: super::intervention::AttemptKey,
}

impl AttemptOwnerGuard {
    /// Bind `key` to `worker` in `owners` and return the guard that releases
    /// the binding on drop.
    pub(crate) fn bind(
        owners: super::intervention::AttemptOwnerIndex,
        key: super::intervention::AttemptKey,
        worker: super::registry::WorkerId,
    ) -> Self {
        owners.bind(key.clone(), worker);
        Self { owners, key }
    }
}

impl Drop for AttemptOwnerGuard {
    fn drop(&mut self) {
        self.owners.release(&self.key);
    }
}

/// Normalize a wire `node` (`Option<String>`) onto the registry's optional
/// locality affinity, applying the SAME none-convention the gRPC registration
/// path uses (`registry::optional_node`): an empty string carries no node, so it
/// collapses to `None`; any non-empty value is the worker's advertised node.
///
/// The wire already models `node` as `Option<String>`, but a worker that joins
/// `Some("")` (the empty-string node) must not register a distinct empty-node
/// affinity that no pinned dispatch could ever match — it is semantically
/// unpinned, exactly as the gRPC proto3 empty default is. Folding it to `None`
/// here keeps the two registration paths byte-for-byte equivalent.
fn normalize_wire_node(node: Option<&str>) -> Option<String> {
    node.filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
}

/// Connection-keyed [`ConnectionNotifier`] that turns liminal's in-band worker
/// registration into a first-class [`ConnectedWorkerRegistry`] membership.
///
/// This is the SERVER half of LSUB-L2: when a worker connects with a
/// [`WireWorkerRegistration`] (the SDK's `connect_with_registration`), liminal's
/// connection process invokes [`on_worker_registered`](Self::on_worker_registered)
/// with the connection's beamr `pid` and the worker's declared
/// `(namespaces, task_queue, node, activity_types)`. The notifier builds a
/// [`WorkerDelivery::Liminal`] over the connection and inserts it into the
/// registry — the SAME registry entry, selected the SAME way, as a gRPC worker —
/// retiring the LSUB-1 out-of-band `active_connection_pids()` + hard-coded
/// registration hack.
///
/// # Lifetime of the registration guard
///
/// [`ConnectedWorkerRegistry::register_delivery`] returns a
/// [`WorkerRegistration`] guard whose drop deregisters the worker. The notifier
/// OWNS that guard keyed by `pid` (`Mutex<HashMap<u64, WorkerRegistration>>`), so
/// the registration lives exactly as long as the connection: it is inserted on
/// register and removed (dropped) on
/// [`on_worker_unregistered`](Self::on_worker_unregistered), which liminal fires
/// on connection close.
///
/// # Construction-order cycle (notifier <-> supervisor)
///
/// [`LiminalWorkerDelivery`] needs a [`ConnectionSupervisor`] handle to push to
/// the worker's connection, but the supervisor is itself constructed WITH this
/// notifier ([`ConnectionSupervisor::with_services_and_notifier`]) — a cycle. The
/// notifier therefore holds the supervisor behind a [`OnceLock`], populated
/// IMMEDIATELY after the supervisor is built via [`Self::bind_supervisor`]. The
/// `OnceLock` is never read before it is set in correct wiring (a worker can only
/// register after the listener — built after the supervisor and after
/// `bind_supervisor` — accepts its connection); if it somehow were, registration
/// is REJECTED with a typed error rather than panicking, so there is no
/// production `unwrap`/`expect` and no second always-`None` code path.
pub struct LiminalConnectionNotifier {
    registry: ConnectedWorkerRegistry,
    contract_catalog: Option<Arc<aion::Engine>>,
    supervisor: OnceLock<ConnectionSupervisor>,
    guards: Mutex<HashMap<u64, WorkerRegistration>>,
    /// The neutral intervention primitives a liminal-connected agent worker
    /// advertises (NOI-6, item 4). The liminal `WorkerRegistration` wire has a fixed
    /// shape that cannot carry this, so it is configured on the notifier at the
    /// composition root from the harness's advertised `AgentSession::capabilities()`
    /// and recorded on every registered worker's handle, where the intervention
    /// router gates on it. Default empty = observability-only (a plain activity
    /// worker), so the router offers no controls for it.
    intervention_capabilities: aion_core::InterventionCapabilities,
    /// The transcript sequencer a worker's observability publishes drain into
    /// (NOI-5b), plus the Tokio [`Handle`](tokio::runtime::Handle) to bridge the
    /// synchronous connection-process callback onto the async publish. `None` (the
    /// default, and every non-agent boot) makes the observability tap a no-op, so a
    /// worker publish to the reserved channel is simply ignored by the notifier.
    transcript: Option<TranscriptTap>,
    /// The SAME per-task liveness tracker the bridge dispatcher tracks into and
    /// the #176 expiry sweeper expires from. A worker's automatic
    /// [`WorkerLivenessBeat`] publishes on [`WORKER_LIVENESS_CHANNEL`] refresh
    /// their task stamps here, keeping the sweeper honest for liminal-delivered
    /// dispatches. `None` (a wiring that never bridges dispatches, e.g. isolated
    /// tests) consumes and drops the beats.
    heartbeat_tracker: Option<super::heartbeat::HeartbeatTracker>,
}

/// The observability-drain leg of the notifier: a bounded, ORDERED queue into
/// the one drain task that publishes transcript events sequentially.
///
/// One consumer is load-bearing, not an implementation detail: a spawned task
/// per event (the pre-2026-07-23 shape) made every in-flight frame a
/// concurrent writer racing the same stream head, turning the sequencer's
/// optimistic-append loop into an O(N²) conflict stampede under load — and
/// destroyed `worker_seq` order on the durable transcript. Sequential
/// draining preserves arrival order and leaves the conflict-retry loop to
/// handle only genuine cross-process races (failover adoption).
#[derive(Clone)]
struct TranscriptTap {
    queue: tokio::sync::mpsc::Sender<aion_core::ActivityEvent>,
}

/// Bound on the transcript drain queue: events a slow durable append cannot
/// keep up with are dropped (with a warning) rather than buffered without
/// limit. Live streaming is best-effort; the bound protects server memory.
const TRANSCRIPT_QUEUE_CAPACITY: usize = 4096;

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

impl LiminalConnectionNotifier {
    /// Build a notifier that registers connecting workers into `registry`.
    ///
    /// The supervisor handle is bound separately via [`Self::bind_supervisor`]
    /// immediately after the supervisor is constructed, resolving the
    /// notifier <-> supervisor construction cycle (see the type docs).
    #[must_use]
    pub fn new(registry: ConnectedWorkerRegistry) -> Self {
        Self {
            registry,
            contract_catalog: None,
            supervisor: OnceLock::new(),
            guards: Mutex::new(HashMap::new()),
            intervention_capabilities: aion_core::InterventionCapabilities::none(),
            transcript: None,
            heartbeat_tracker: None,
        }
    }

    /// Install the live package catalog used to refuse incompatible workers
    /// before liminal acknowledges or publishes their registration.
    #[must_use]
    pub fn with_contract_catalog(mut self, engine: Arc<aion::Engine>) -> Self {
        self.contract_catalog = Some(engine);
        self
    }

    /// Install the shared per-task liveness tracker so a worker's automatic
    /// [`WorkerLivenessBeat`] publishes refresh the SAME in-flight entries the
    /// bridge dispatcher tracks and the #176 sweeper expires.
    ///
    /// MUST be wired on every boot that hosts the engine-seam bridge (the
    /// production composition does), or the sweeper would expire healthy
    /// liminal workers running activities longer than the heartbeat window.
    /// Without it (isolated tests that never bridge-dispatch) beats are
    /// consumed and dropped.
    #[must_use]
    pub fn with_heartbeat_tracker(mut self, tracker: super::heartbeat::HeartbeatTracker) -> Self {
        self.heartbeat_tracker = Some(tracker);
        self
    }

    /// Install the transcript sequencer a worker's observability publishes drain into
    /// (NOI-5b), capturing the CURRENT Tokio runtime handle to bridge the synchronous
    /// connection-process callback onto the async publish.
    ///
    /// MUST be called from within a Tokio runtime (the server boot path is), so the
    /// captured [`Handle`](tokio::runtime::Handle) can spawn the append+fan-out when a
    /// worker publishes a transcript event over the reserved channel. Without this
    /// builder the observability tap is a no-op (a plain, non-agent deployment).
    ///
    /// # Panics
    ///
    /// Panics if called outside a Tokio runtime — a construction-time wiring error in
    /// the server boot, never a runtime condition (the boot path always builds the
    /// notifier inside the server runtime).
    #[must_use]
    pub fn with_transcript_publisher(
        mut self,
        publisher: crate::activity_publisher::ActivityEventPublisher,
    ) -> Self {
        let (queue, mut events) =
            tokio::sync::mpsc::channel::<aion_core::ActivityEvent>(TRANSCRIPT_QUEUE_CAPACITY);
        // The ONE drain task: transcript events publish sequentially in
        // arrival order (see the `TranscriptTap` docs for why one consumer is
        // load-bearing). It lives as long as any queue sender does.
        //
        // The drain COALESCES under the operator's flush policy: the events
        // waiting on this queue are committed as batches, so a chatty agent
        // costs one storage-tree commit per batch instead of one per event —
        // and one commit is one whole-leaf rewrite (forensics 2026-08-17).
        // Arrival order is untouched; only the number of commits changes.
        tokio::runtime::Handle::current().spawn(async move {
            let dropped = publisher.drain(&mut events, "observability_tap").await;
            if dropped > 0 {
                tracing::warn!(
                    dropped,
                    operation = "observability_tap",
                    "observability tap: transcript events were not retained"
                );
            }
        });
        self.transcript = Some(TranscriptTap { queue });
        self
    }

    /// Set the neutral intervention capability set every worker registering through
    /// this notifier advertises (NOI-6, item 4).
    ///
    /// The composition root wires this from the harness's advertised
    /// `AgentSession::capabilities()` so a liminal-connected agent worker's handle
    /// carries the primitives its harness supports, which the intervention router
    /// gates on. Without this builder the set is empty (observability-only), so a
    /// plain activity worker advertises no controls. Pure builder addition, mirroring
    /// the registry's capability-carrying registration façade.
    #[must_use]
    pub fn with_intervention_capabilities(
        mut self,
        capabilities: aion_core::InterventionCapabilities,
    ) -> Self {
        self.intervention_capabilities = capabilities;
        self
    }

    /// Bind the connection supervisor the notifier pushes through, immediately
    /// after it is constructed with this notifier.
    ///
    /// Returns `true` when the supervisor was stored, `false` when it was already
    /// bound (a second bind is a wiring bug and is ignored, never overwriting the
    /// live handle). Call this exactly once, right after
    /// [`ConnectionSupervisor::with_services_and_notifier`].
    pub fn bind_supervisor(&self, supervisor: ConnectionSupervisor) -> bool {
        self.supervisor.set(supervisor).is_ok()
    }

    /// Snapshot of every live liminal connection the dead-man switch must ping:
    /// its pid, the worker it registered, and the push leg to reach it.
    ///
    /// Built from the SAME guard map that owns each connection's registration,
    /// so a connection that has closed (its guard removed) is structurally
    /// absent from the snapshot and is never pinged. A poisoned guard map is
    /// recovered rather than skipped: silently returning no targets would turn
    /// the dead-man switch off exactly when the server is already unhealthy.
    #[must_use]
    pub fn liveness_targets(&self) -> Vec<super::liminal_liveness::LivenessTarget> {
        let Some(supervisor) = self.supervisor.get() else {
            return Vec::new();
        };
        let guards = match self.guards.lock() {
            Ok(guards) => guards,
            Err(poisoned) => poisoned.into_inner(),
        };
        guards
            .iter()
            .filter_map(|(pid, guard)| {
                guard
                    .worker_id()
                    .map(|worker_id| super::liminal_liveness::LivenessTarget {
                        pid: *pid,
                        worker_id,
                        delivery: LiminalWorkerDelivery::new(supervisor.clone(), *pid),
                    })
            })
            .collect()
    }

    /// Refresh one worker's in-flight task stamp from a [`WorkerLivenessBeat`]
    /// published on [`WORKER_LIVENESS_CHANNEL`].
    ///
    /// The worker is resolved by the publishing CONNECTION (`pid` -> the
    /// registration guard this notifier owns), never by wire identity, so a
    /// beat can only refresh tasks assigned to the worker that sent it. An
    /// untracked task is benign (an outbox dispatch the tracker never held, or
    /// a beat racing its completion) and is dropped silently; a malformed
    /// payload or unregistered connection is logged and dropped — a bad beat
    /// must never tear down the connection callback.
    fn record_liveness_beat(&self, pid: u64, payload: &[u8]) {
        let Some(tracker) = &self.heartbeat_tracker else {
            return;
        };
        let beat: WorkerLivenessBeat = match serde_json::from_slice(payload) {
            Ok(beat) => beat,
            Err(error) => {
                tracing::warn!(%error, "liveness tap: malformed WorkerLivenessBeat payload");
                return;
            }
        };
        let worker_id = match self.guards.lock() {
            Ok(guards) => guards.get(&pid).and_then(WorkerRegistration::worker_id),
            Err(poisoned) => poisoned
                .into_inner()
                .get(&pid)
                .and_then(WorkerRegistration::worker_id),
        };
        let Some(worker_id) = worker_id else {
            tracing::warn!(
                connection_pid = pid,
                "liveness tap: beat from a connection with no registered worker"
            );
            return;
        };
        let activity_id = ActivityId::from_sequence_position(beat.ordinal);
        if let Err(error) = tracker.record_liveness(
            worker_id,
            &beat.workflow_id,
            &activity_id,
            std::time::Instant::now(),
        ) {
            tracing::error!(
                %error,
                connection_pid = pid,
                "liveness tap: heartbeat tracker refresh failed"
            );
        }
    }

    /// Apply one worker's [`WorkerCapabilitiesAnnouncement`] published on
    /// [`WORKER_CAPABILITIES_CHANNEL`] to its registered handle.
    ///
    /// The worker is resolved by the publishing CONNECTION (`pid` -> the
    /// registration guard this notifier owns), never by wire identity, so an
    /// announcement can only ever update the worker that sent it. A malformed
    /// payload or unregistered connection is logged and dropped — a bad
    /// announcement must never tear down the connection callback.
    fn record_capabilities_announcement(&self, pid: u64, payload: &[u8]) {
        let announcement: WorkerCapabilitiesAnnouncement = match serde_json::from_slice(payload) {
            Ok(announcement) => announcement,
            Err(error) => {
                tracing::warn!(
                    %error,
                    "capabilities tap: malformed WorkerCapabilitiesAnnouncement payload"
                );
                return;
            }
        };
        let worker_id = match self.guards.lock() {
            Ok(guards) => guards.get(&pid).and_then(WorkerRegistration::worker_id),
            Err(poisoned) => poisoned
                .into_inner()
                .get(&pid)
                .and_then(WorkerRegistration::worker_id),
        };
        let Some(worker_id) = worker_id else {
            tracing::warn!(
                connection_pid = pid,
                "capabilities tap: announcement from a connection with no registered worker"
            );
            return;
        };
        match self
            .registry
            .set_intervention_capabilities(worker_id, &announcement.capabilities)
        {
            Ok(true) => {}
            Ok(false) => tracing::warn!(
                connection_pid = pid,
                worker_id = ?worker_id,
                "capabilities tap: announcement raced the worker's deregistration"
            ),
            Err(error) => tracing::error!(
                %error,
                connection_pid = pid,
                "capabilities tap: registry capability update failed"
            ),
        }
    }

    fn validate_registration_contract(
        &self,
        registration: &WireWorkerRegistration,
    ) -> Result<(), LiminalServerError> {
        let Some(engine) = &self.contract_catalog else {
            return Ok(());
        };
        let advertised =
            registration
                .activities
                .iter()
                .map(|activity| {
                    let input_schema =
                        serde_json::from_str(&activity.input_schema_json).map_err(|error| {
                            LiminalServerError::ListenerAccept {
                                message: format!(
                                    "liminal worker activity `{}` input schema is invalid: {error}",
                                    activity.name
                                ),
                            }
                        })?;
                    let output_schema = serde_json::from_str(&activity.output_schema_json)
                        .map_err(|error| LiminalServerError::ListenerAccept {
                            message: format!(
                                "liminal worker activity `{}` output schema is invalid: {error}",
                                activity.name
                            ),
                        })?;
                    Ok(aion_package::ActivityDescriptor {
                        name: activity.name.clone(),
                        input_schema,
                        output_schema,
                    })
                })
                .collect::<Result<Vec<_>, LiminalServerError>>()?;
        // Both advertised forms travel into the gate together: the NAME set the
        // dispatcher selects on and the CONTRACT set admission compares. A
        // refusal computed from one while the record printed the other is what
        // produced the 2026-07-30 self-contradicting log.
        let activity_types = registration
            .activity_types
            .iter()
            .cloned()
            .collect::<std::collections::BTreeSet<_>>();
        super::contracts::validate_worker_contracts(
            engine,
            self.registry.admission_audit(),
            &registration.task_queue,
            registration.node.as_deref(),
            &registration.identity,
            super::contracts::WorkerAdvertisement {
                activity_types: &activity_types,
                contracts: &advertised,
            },
        )
        .map_err(|error| LiminalServerError::ListenerAccept {
            message: error.to_string(),
        })
    }

    /// Admit one connecting worker past the TRANSPORT's own refusals, or refuse
    /// it with a structured reason.
    ///
    /// Split out of [`ConnectionNotifier::on_worker_registered`] so every
    /// refusal this transport owns — unbound supervisor, registry error, lease
    /// failure — funnels through exactly one place that logs it. Before that
    /// split the rejection reason was carried in the `Rejected` ack and logged
    /// NOWHERE: on 2026-07-29 a worker whose `.v4` contract field-mismatched was
    /// refused on every single dial with the server silent on all of them, and
    /// the refused process looked merely asleep.
    ///
    /// **The contract check is deliberately NOT here.** That 2026-07-29 fix was
    /// made in this caller rather than at the shared gate, and within a week the
    /// gRPC transport re-manifested the identical silent refusal with nothing in
    /// the tree able to notice (#147). The gate now names its own refusal for
    /// every transport, so the contract check runs BEFORE this function and its
    /// error propagates from here already spoken for.
    fn admit_registration(
        &self,
        pid: u64,
        registration: &WireWorkerRegistration,
    ) -> Result<(), LiminalServerError> {
        // The delivery leg needs the supervisor to push to this connection. In
        // correct wiring it is bound before any connection is accepted; a missing
        // binding is a rejected registration, never a panic.
        let supervisor =
            self.supervisor
                .get()
                .ok_or_else(|| LiminalServerError::ListenerAccept {
                    message: format!(
                        "liminal worker registration for connection {pid} rejected: \
                     notifier supervisor handle not yet bound"
                    ),
                })?;

        let delivery = WorkerDelivery::Liminal(LiminalWorkerDelivery::new(supervisor.clone(), pid));
        let node = normalize_wire_node(registration.node.as_deref());
        // Insert into the SAME registry, selected the SAME way, as a gRPC worker.
        // A registry error (poisoned lock) becomes a Rejected ack so the worker
        // never believes it is registered when it is not.
        let guard = self
            .registry
            .register_delivery_with_capabilities(
                registration.namespaces.iter().cloned(),
                registration.task_queue.clone(),
                node,
                registration.activity_types.iter(),
                delivery,
                self.intervention_capabilities.clone(),
            )
            .map_err(|error| LiminalServerError::ListenerAccept {
                message: format!(
                    "liminal worker registration for connection {pid} rejected: {error}"
                ),
            })?;
        let worker_id = guard
            .worker_id()
            .ok_or_else(|| LiminalServerError::ListenerAccept {
                message: format!(
                    "liminal worker registration for connection {pid} has no worker id"
                ),
            })?;

        // OWN the guard for the connection's lifetime, keyed by pid. Dropping it
        // (on unregister) deregisters the worker, so the registration lives
        // exactly as long as the connection.
        let mut guards = self.guards.lock().map_err(|_| {
            // The accepted registry entry cannot be tracked for deregistration, so
            // reject (and drop the just-created guard, deregistering it) rather
            // than leak a never-deregistered association.
            LiminalServerError::ListenerAccept {
                message: format!(
                    "liminal worker registration for connection {pid} rejected: \
                     notifier guard map poisoned"
                ),
            }
        })?;
        guards.insert(pid, guard);
        drop(guards);
        if let Some(tracker) = &self.heartbeat_tracker {
            if let Err(error) = tracker.register_connection(worker_id, std::time::Instant::now()) {
                let removed = match self.guards.lock() {
                    Ok(mut guards) => guards.remove(&pid),
                    Err(poisoned) => poisoned.into_inner().remove(&pid),
                };
                drop(removed);
                return Err(LiminalServerError::ListenerAccept {
                    message: format!(
                        "liminal worker registration for connection {pid} rejected: {error}"
                    ),
                });
            }
        }
        tracing::info!(
            connection_pid = pid,
            identity = %registration.identity,
            task_queue = %registration.task_queue,
            "registered liminal worker in-band"
        );
        Ok(())
    }
}

impl ConnectionNotifier for LiminalConnectionNotifier {
    fn on_worker_registered(
        &self,
        pid: u64,
        registration: &WireWorkerRegistration,
    ) -> Result<(), LiminalServerError> {
        // The contract gate NAMES its own refusal, for every transport, with the
        // queue, the node, the worker build identity and the structured reason.
        // So this path propagates that refusal SILENTLY rather than restating
        // it: a second copy of the same log is how #147 happened, and a log
        // written twice per refusal is #94's 12 MB/hour with a second author.
        self.validate_registration_contract(registration)?;
        // Every refusal this TRANSPORT owns is named here instead, at WARN, with
        // the same reason that rides back in the `Rejected` ack, so both sides'
        // logs carry identical text and a refused worker can be diagnosed from
        // either end.
        self.admit_registration(pid, registration)
            .inspect_err(|error| {
                // The two advertised sets are logged SEPARATELY and named for
                // what each one is. Logging `activity_types` alone beside a
                // contract-mismatch reason produced a record that listed an
                // action as advertised in the same line that reported it
                // `<missing>` — true of two different sets, and unresolvable by
                // the operator it exists to help.
                let advertised_contracts = registration
                    .activities
                    .iter()
                    .map(|activity| activity.name.as_str())
                    .collect::<Vec<_>>();
                tracing::warn!(
                    connection_pid = pid,
                    identity = %registration.identity,
                    task_queue = %registration.task_queue,
                    namespaces = ?registration.namespaces,
                    advertised_activity_types = ?registration.activity_types,
                    advertised_contracts = ?advertised_contracts,
                    reason = %error,
                    "REFUSED liminal worker registration"
                );
            })
    }

    fn on_worker_unregistered(&self, pid: u64) {
        // Remove + drop the guard for pid, deregistering the worker. A poisoned
        // lock on the close path has no peer to report to; recover the guard map
        // and still drop the guard so the registry does not keep routing to a
        // gone connection.
        let removed = match self.guards.lock() {
            Ok(mut guards) => guards.remove(&pid),
            Err(poisoned) => poisoned.into_inner().remove(&pid),
        };
        let worker_id = removed.as_ref().and_then(WorkerRegistration::worker_id);
        if let (Some(tracker), Some(worker_id)) = (&self.heartbeat_tracker, worker_id) {
            if let Err(error) = tracker.unregister_connection(worker_id) {
                tracing::error!(
                    %error,
                    connection_pid = pid,
                    worker_id = worker_id.value(),
                    "failed to clear liminal worker connection lease"
                );
            }
        }
        if removed.is_some() {
            tracing::info!(
                connection_pid = pid,
                "deregistered liminal worker on disconnect"
            );
        }
    }

    fn on_channel_publish(&self, pid: u64, channel: &str, payload: &[u8]) -> bool {
        if let Some(tracker) = &self.heartbeat_tracker {
            let worker_id = match self.guards.lock() {
                Ok(guards) => guards.get(&pid).and_then(WorkerRegistration::worker_id),
                Err(poisoned) => poisoned
                    .into_inner()
                    .get(&pid)
                    .and_then(WorkerRegistration::worker_id),
            };
            if let Some(worker_id) = worker_id {
                if let Err(error) =
                    tracker.record_connection_activity(worker_id, std::time::Instant::now())
                {
                    tracing::error!(
                        %error,
                        connection_pid = pid,
                        worker_id = worker_id.value(),
                        "failed to advance liminal worker connection lease"
                    );
                }
            }
        }
        // Reserved liveness channel: refresh the beat's in-flight task stamp in
        // the shared heartbeat tracker (always consumed, never fanned out).
        if channel == WORKER_LIVENESS_CHANNEL {
            self.record_liveness_beat(pid, payload);
            return true;
        }
        // Reserved capabilities channel: apply the worker's advertised
        // intervention capabilities to its registered handle (always consumed).
        if channel == WORKER_CAPABILITIES_CHANNEL {
            self.record_capabilities_announcement(pid, payload);
            return true;
        }
        // Only consume the reserved observability channel; any other channel falls
        // through to liminal's normal fan-out (this returns false).
        if channel != liminal_sdk::OBSERVABILITY_CHANNEL {
            return false;
        }
        let Some(tap) = &self.transcript else {
            // No transcript sequencer installed (a non-agent deployment): still
            // CONSUME the reserved channel so it never leaks into the fan-out, but
            // drop the event — there is nothing to persist it into.
            return true;
        };
        let event: aion_core::ActivityEvent = match serde_json::from_slice(payload) {
            Ok(event) => event,
            Err(error) => {
                tracing::warn!(%error, "observability tap: malformed ActivityEvent payload");
                return true;
            }
        };
        // Hand the event to the ordered drain queue (the synchronous
        // connection-process callback never blocks). A full queue means the
        // durable append cannot keep up: the event is dropped with a warning,
        // never buffered without bound.
        if let Err(error) = tap.queue.try_send(event) {
            tracing::warn!(%error, "observability tap: transcript queue rejected event");
        }
        true
    }
}

/// The production [`InterventionTransport`](super::intervention::InterventionTransport):
/// pushes a routed command to the owning worker over its liminal server-push
/// connection (NOI-6, §6.2).
///
/// It reads the worker handle's [`WorkerDelivery::Liminal`] leg and pushes the
/// neutral [`InterventionRequest`] via [`LiminalWorkerDelivery::push_intervention`],
/// running the blocking push off the async runtime. A worker delivered over gRPC
/// (no liminal leg) surfaces the stale-target no-op via a connection-lost error, so
/// the router NACKs the operator rather than routing to a leg it cannot reach.
#[derive(Clone, Debug, Default)]
pub struct LiminalInterventionTransport;

#[async_trait]
impl super::intervention::InterventionTransport for LiminalInterventionTransport {
    async fn push(
        &self,
        worker: &super::registry::WorkerHandle,
        command: aion_core::InterventionCommand,
    ) -> Result<aion_core::InterventionOutcome, ServerError> {
        let delivery = match worker.delivery() {
            WorkerDelivery::Liminal(delivery) => delivery.clone(),
            WorkerDelivery::Grpc(_) => {
                // No liminal leg to push to: the intervention transport rides the
                // liminal push channel only, so this is unreachable for the target.
                return Err(ServerError::worker_connection_lost(
                    "liminal-push",
                    "owning worker is not delivered over liminal".to_owned(),
                ));
            }
        };
        let request = InterventionRequest {
            intervention: command,
        };
        let reply = tokio::task::spawn_blocking(move || delivery.push_intervention(&request))
            .await
            .map_err(|error| {
                dispatch_error(
                    "liminal-push",
                    format!("intervention task join failed: {error}"),
                )
            })??;
        Ok(reply.outcome)
    }
}

#[cfg(test)]
#[path = "liminal_contract_tests.rs"]
mod contract_tests;

#[cfg(test)]
mod tests {
    use super::{channel_for_row, dispatch_channel_name, normalize_wire_node};
    use aion_core::{ActivityId, ContentType, Payload, WorkflowId};
    use aion_store::{OutboxRow, OutboxStatus};
    use chrono::Utc;
    use uuid::Uuid;

    /// The NOI-6 dispatch owner guard RELEASES its binding on drop, on EVERY exit path
    /// (reply, error, panic) — so the attempt-owner back-index tracks exactly the
    /// attempts currently in flight. This is the invariant the dispatch path relies on
    /// to never leak a finished attempt's owner.
    #[tokio::test]
    async fn attempt_owner_guard_releases_on_drop() -> Result<(), Box<dyn std::error::Error>> {
        use super::super::intervention::{AttemptKey, AttemptOwnerIndex};
        use super::super::registry::{ConnectedWorkerRegistry, WorkerDelivery};
        use super::AttemptOwnerGuard;

        // A real registration yields a real WorkerId (there is no fabricated id).
        let registry = ConnectedWorkerRegistry::default();
        let (tx, _rx) = tokio::sync::mpsc::channel(1);
        let types = [String::from("agent")];
        let registration = registry.register_delivery_with_capabilities(
            [String::from("default")],
            String::from("default"),
            None,
            types.iter(),
            WorkerDelivery::Grpc(tx),
            aion_core::InterventionCapabilities::none(),
        )?;
        let worker = registration
            .worker_id()
            .ok_or("registration must assign a worker id")?;

        let owners = AttemptOwnerIndex::new();
        let key = AttemptKey::new(
            WorkflowId::new(Uuid::nil()),
            aion_core::RunId::new(Uuid::from_u128(0x11)),
            ActivityId::from_sequence_position(3),
            1,
        );
        owners.bind(key.clone(), worker);
        assert_eq!(
            owners.owner(&key),
            Some(worker),
            "owner bound before the guard"
        );
        {
            let _guard = AttemptOwnerGuard {
                owners: owners.clone(),
                key: key.clone(),
            };
            assert_eq!(
                owners.owner(&key),
                Some(worker),
                "still bound while in flight"
            );
        }
        // The guard dropped at the end of the block: the binding is released, so a
        // later intervention resolves no owner (the too-late no-op).
        assert_eq!(
            owners.owner(&key),
            None,
            "owner released when the dispatch returns"
        );
        Ok(())
    }

    /// The channel format is pinned EXACTLY: any change is a wire-compatibility
    /// break (the dispatcher and any worker subscription must agree byte-for-byte).
    /// The UNPINNED (`None`) channel MUST stay byte-identical to the pre-NODE-5
    /// format so existing pool subscriptions are stable.
    #[test]
    fn channel_format_is_pinned() {
        assert_eq!(
            dispatch_channel_name("remote", "gpu", None),
            "aion.dispatch.remote.gpu"
        );
        assert_eq!(
            dispatch_channel_name("local", "norn", None),
            "aion.dispatch.local.norn"
        );
    }

    /// A node-pinned dispatch appends the node as an injectively-encoded
    /// sub-segment: `f(ns, tq, Some(node))` == `aion.dispatch.{ns}.{tq}.{node}`.
    #[test]
    fn node_pinned_channel_appends_node_subsegment() {
        assert_eq!(
            dispatch_channel_name("remote", "gpu", Some("box-7")),
            "aion.dispatch.remote.gpu.box-7"
        );
    }

    /// Same input always yields the same channel (the function is stable/total),
    /// for both the unpinned and node-pinned cases.
    #[test]
    fn channel_derivation_is_stable() {
        assert_eq!(
            dispatch_channel_name("default", "default", None),
            dispatch_channel_name("default", "default", None)
        );
        assert_eq!(
            dispatch_channel_name("default", "default", Some("box-1")),
            dispatch_channel_name("default", "default", Some("box-1"))
        );
    }

    /// Distinct `(namespace, task_queue)` pools derive distinct channels — the
    /// whole point of NSTQ-5: `(remote, gpu)` and `(local, norn)` never collide.
    #[test]
    fn distinct_pools_get_distinct_channels() {
        assert_ne!(
            dispatch_channel_name("remote", "gpu", None),
            dispatch_channel_name("local", "norn", None)
        );
    }

    /// A node-pinned dispatch and the unpinned dispatch for the SAME pool derive
    /// DISTINCT channels, and two distinct nodes for the same pool also differ —
    /// the property node isolation rests on (the subscriber contract).
    #[test]
    fn node_pin_separates_channels() {
        let unpinned = dispatch_channel_name("remote", "gpu", None);
        let box7 = dispatch_channel_name("remote", "gpu", Some("box-7"));
        let box8 = dispatch_channel_name("remote", "gpu", Some("box-8"));
        assert_ne!(
            unpinned, box7,
            "pinned dispatch must not reach unpinned pool"
        );
        assert_ne!(box7, box8, "distinct nodes must not collide");
    }

    /// The core injectivity property: free-form fields containing the segment
    /// separator `.` must NOT bleed across the join. With the raw `format!` the
    /// disjoint pools `("a.b", "c")` and `("a", "b.c")` both collapsed onto
    /// `aion.dispatch.a.b.c` — a cross-pool/cross-namespace leak. The per-segment
    /// encode keeps them distinct.
    #[test]
    fn dotted_fields_do_not_collide_across_segments() {
        assert_ne!(
            dispatch_channel_name("a.b", "c", None),
            dispatch_channel_name("a", "b.c", None),
            "a '.' in a field must not bleed across the segment separator"
        );
    }

    /// Injectivity holds ACROSS segment counts: a 2-segment (unpinned) channel
    /// can never be confused with a 3-segment (node-pinned) channel even when a
    /// `.` in a field would otherwise make the raw strings line up. Both
    /// directions of the brief's collision cases must stay distinct.
    #[test]
    fn node_subsegment_does_not_collide_with_dotted_fields() {
        // A node sub-segment vs the same dot living inside task_queue.
        assert_ne!(
            dispatch_channel_name("a", "b", Some("c")),
            dispatch_channel_name("a", "b.c", None),
            "a node sub-segment must not collide with a dotted task_queue"
        );
        // The dot living inside namespace vs a node sub-segment.
        assert_ne!(
            dispatch_channel_name("a.b", "c", None),
            dispatch_channel_name("a", "b", Some("c")),
            "a dotted namespace must not collide with a node-pinned channel"
        );
    }

    /// More reserved-char shifts that the raw `format!` collapsed but the encode
    /// must keep distinct — the dot can sit on either side of the boundary.
    #[test]
    fn reserved_char_shifts_stay_distinct() {
        // Dot at the end of namespace vs start of task_queue.
        assert_ne!(
            dispatch_channel_name("ns.", "tq", None),
            dispatch_channel_name("ns", ".tq", None)
        );
        // Empty field vs the dot living in the other field.
        assert_ne!(
            dispatch_channel_name("", "a.b", None),
            dispatch_channel_name(".a", "b", None)
        );
        // The escape char itself must not let a literal `%2E` impersonate an
        // encoded `.`: `("%2E", "x")` (literal percent-two-E) must differ from
        // `(".", "x")` (an actual dot, which encodes to `%2E`).
        assert_ne!(
            dispatch_channel_name("%2E", "x", None),
            dispatch_channel_name(".", "x", None)
        );
    }

    /// Encoding is injective in ALL THREE segments independently and is exactly
    /// reversible (the property the channel relies on), so a small exhaustive
    /// sweep of reserved-char arrangements — INCLUDING the optional node taking
    /// `None` and every reserved-char value — yields all-distinct channels. This
    /// covers cross-segment-count collisions (the `None` vs `Some` boundary) too.
    #[test]
    fn encoding_is_injective_over_reserved_char_triples() {
        let fields = ["a", "a.b", "a.", ".a", ".", "", "%", "%2E", "a%b", "%2."];
        let nodes = [
            None,
            Some("a"),
            Some("a.b"),
            Some("."),
            Some(""),
            Some("%2E"),
        ];
        let mut channels = std::collections::HashSet::new();
        for ns in fields {
            for tq in fields {
                for node in nodes {
                    let channel = dispatch_channel_name(ns, tq, node);
                    assert!(
                        channels.insert(channel.clone()),
                        "collision on ({ns:?}, {tq:?}, {node:?}) -> {channel}"
                    );
                }
            }
        }
    }

    fn row(namespace: &str, task_queue: &str) -> OutboxRow {
        let workflow_id = WorkflowId::new(Uuid::new_v4());
        OutboxRow {
            dispatch_key: format!("{workflow_id}:0"),
            workflow_id,
            ordinal: 0,
            run_id: Some(aion_core::RunId::new_v4()),
            namespace: namespace.to_owned(),
            task_queue: task_queue.to_owned(),
            node: None,
            activity_type: "charge-card".to_owned(),
            input: Payload::new(ContentType::Json, Vec::new()),
            status: OutboxStatus::Pending,
            attempt: 0,
            visible_after: Utc::now(),
            claimed_at: None,
            failure_delivered: false,
        }
    }

    /// A row's channel is derived from its durable `(namespace, task_queue)`
    /// columns (NSTQ-2), through the same single derivation function — and
    /// `activity_type` does NOT enter the channel. With `node = None` the channel
    /// is byte-identical to the pre-NODE-5 2-segment form.
    #[test]
    fn channel_for_row_uses_namespace_and_task_queue_only() {
        let remote_gpu = row("remote", "gpu");
        let local_norn = row("local", "norn");
        assert_eq!(channel_for_row(&remote_gpu), "aion.dispatch.remote.gpu");
        assert_eq!(channel_for_row(&local_norn), "aion.dispatch.local.norn");
        assert_ne!(channel_for_row(&remote_gpu), channel_for_row(&local_norn));

        // Two rows that differ ONLY in activity_type derive the SAME channel:
        // activity_type is matched after delivery, not used to select the pool.
        let mut other_activity = row("remote", "gpu");
        other_activity.activity_type = "refund".to_owned();
        assert_eq!(
            channel_for_row(&remote_gpu),
            channel_for_row(&other_activity),
            "activity_type must not affect the channel"
        );
    }

    /// A row carrying `Some(node)` (NODE-2) derives the node-pinned sub-channel,
    /// distinct from the same pool's unpinned channel; a row with `None` derives
    /// the 2-segment channel. `channel_for_row` threads `row.node` through the
    /// single derivation function.
    #[test]
    fn channel_for_row_derives_node_subchannel_when_pinned() {
        let mut pinned = row("remote", "gpu");
        pinned.node = Some("box-7".to_owned());
        assert_eq!(channel_for_row(&pinned), "aion.dispatch.remote.gpu.box-7");

        let unpinned = row("remote", "gpu");
        assert_eq!(channel_for_row(&unpinned), "aion.dispatch.remote.gpu");
        assert_ne!(channel_for_row(&pinned), channel_for_row(&unpinned));
    }

    /// The outbox wire request stamps the row's stored ZERO-based attempt as a
    /// ONE-based delivery attempt (zero is malformed on the wire) — the exact
    /// stamp the gRPC outbox arm's `to_scheduled` applies — carries no labels,
    /// and assigns no liveness window (outbox rows are not tracker-tracked; the
    /// outbox retry loop is their liveness backstop).
    #[test]
    fn request_for_row_stamps_one_based_attempt_and_no_window() -> Result<(), crate::ServerError> {
        let mut retried = row("remote", "gpu");
        retried.attempt = 2;
        let run_id = retried
            .run_id
            .as_ref()
            .ok_or_else(|| crate::ServerError::worker_dispatch("", "", "test row missing run"))?;
        let token = super::CompletionToken::for_test();
        let request = super::request_for_row(&retried, run_id, &token);
        assert_eq!(
            request.attempt, 3,
            "zero-based row attempt goes one-based on the wire"
        );
        assert!(request.labels.is_empty());
        assert_eq!(request.heartbeat_window_ms, 0);

        let fresh = row("remote", "gpu");
        let fresh_run = fresh
            .run_id
            .as_ref()
            .ok_or_else(|| crate::ServerError::worker_dispatch("", "", "test row missing run"))?;
        assert_eq!(super::request_for_row(&fresh, fresh_run, &token).attempt, 1);
        Ok(())
    }

    /// The wire `node` is normalized onto the registry's optional affinity with
    /// the SAME none-convention the gRPC registration path uses: `None` and the
    /// empty-string node both collapse to unpinned (`None`), a non-empty value is
    /// the advertised node. An empty-string node must NOT register a distinct
    /// empty affinity no pinned dispatch could match.
    #[test]
    fn wire_node_normalizes_empty_to_none() {
        assert_eq!(normalize_wire_node(None), None);
        assert_eq!(normalize_wire_node(Some("")), None);
        assert_eq!(normalize_wire_node(Some("box-7")), Some("box-7".to_owned()));
    }

    // --- #163: the Prefer two-tier spill on the LIMINAL selection path ---------
    //
    // These exercise `RegistryLiminalDispatch::select_liminal_worker` — the
    // liminal transport's worker selection — proving it consults the SAME shared
    // `preferred_node_order` two-tier spill the gRPC path uses (the cross-node
    // demo behaviour), and that placement NEVER mutates the recorded row's node.
    // Selection is delivery-agnostic (`select_worker` filters by node regardless
    // of transport), so a worker registered with any delivery drives the same
    // selection the production liminal-delivered worker would; the tests assert on
    // the SELECTED handle's node, which is exactly what #163 changed.
    mod placement_selection {
        use std::collections::BTreeSet;
        use std::sync::Arc;
        use std::time::Duration;

        use aion_core::{ActivityId, Payload, RunId, WorkflowId};
        use aion_store::{
            InMemoryStore, NamespaceOrigin, NamespacePlacement, NamespaceStore, OutboxRow,
        };

        use crate::error::ServerError;
        use crate::worker::bridge::OutboxDeliveryCallback;
        use crate::worker::registry::{ConnectedWorkerRegistry, WorkerMessage, WorkerRegistration};
        use crate::worker::{DeliveryGate, PlacementCache};

        use super::super::RegistryLiminalDispatch;

        /// No-op delivery callback: the selection tests never deliver a result, so
        /// the completion sink is never invoked. Both methods are unreachable in
        /// these tests and simply report "no live run" if ever called.
        struct NoopCallback;

        impl OutboxDeliveryCallback for NoopCallback {
            fn deliver_completion(
                &self,
                _workflow_id: &WorkflowId,
                _activity_id: &ActivityId,
                _run_id: Option<&RunId>,
                _result: String,
            ) -> Result<bool, ServerError> {
                Ok(false)
            }
            fn deliver_failure(
                &self,
                _workflow_id: &WorkflowId,
                _activity_id: &ActivityId,
                _run_id: Option<&RunId>,
                _reason: String,
            ) -> Result<bool, ServerError> {
                Ok(false)
            }
        }

        fn labels(values: &[&str]) -> BTreeSet<String> {
            values.iter().map(|v| (*v).to_owned()).collect()
        }

        /// Register a worker advertising `node` for `charge` in `namespace`,
        /// returning the registration guard (held to keep it connected).
        fn register_node_worker(
            registry: &ConnectedWorkerRegistry,
            namespace: &str,
            node: &str,
        ) -> Result<WorkerRegistration, ServerError> {
            let (tx, _rx) = tokio::sync::mpsc::channel::<WorkerMessage>(1);
            let types = [String::from("charge")];
            registry.register_namespaces(
                [namespace.to_owned()],
                String::from("default"),
                Some(node.to_owned()),
                types.iter(),
                tx,
            )
        }

        /// Register a worker carrying NO node label (`node == None`) — the
        /// any-node worker a `Prefer` tier spills to and a `Pinned` tier never
        /// admits. The distinction is the whole subject of the refusal tests
        /// below.
        fn register_unlabelled_worker(
            registry: &ConnectedWorkerRegistry,
            namespace: &str,
        ) -> Result<WorkerRegistration, ServerError> {
            let (tx, _rx) = tokio::sync::mpsc::channel::<WorkerMessage>(1);
            let types = [String::from("charge")];
            registry.register_namespaces(
                [namespace.to_owned()],
                String::from("default"),
                None,
                types.iter(),
                tx,
            )
        }

        /// Publish `worker` as dispatch-ineligible, exactly as a liveness round
        /// would.
        fn exclude(
            registry: &ConnectedWorkerRegistry,
            worker: &WorkerRegistration,
        ) -> Result<(), Box<dyn std::error::Error>> {
            let worker_id = worker
                .worker_id()
                .ok_or("registration assigned no worker id")?;
            registry.set_dispatch_ineligible([worker_id].into_iter().collect())?;
            Ok(())
        }

        /// Drive the REAL dispatch path and hand back the refusal's reason.
        ///
        /// Through `OutboxRowDispatch::dispatch`, not through the reason
        /// function directly: the reason is a STRING an operator reads, and a
        /// test that called the private helper would leave the wiring between
        /// selection and refusal — the exact seam that got the tiers wrong —
        /// uncovered.
        async fn refusal_reason(
            dispatch: &RegistryLiminalDispatch,
            row: &OutboxRow,
        ) -> Result<String, Box<dyn std::error::Error>> {
            match super::super::OutboxRowDispatch::dispatch(dispatch, row).await {
                Ok(()) => Err("the dispatch must be refused: no worker is selectable".into()),
                Err(ServerError::WorkerDispatch { reason, .. }) => Ok(reason),
                Err(other) => {
                    Err(format!("expected a worker-dispatch refusal, got: {other}").into())
                }
            }
        }

        /// Build an UNPINNED outbox row (`node == None`) in `namespace` for `charge`.
        fn unpinned_row(namespace: &str) -> OutboxRow {
            OutboxRow::pending(
                WorkflowId::new_v4(),
                0,
                String::from("charge"),
                Payload::from_json(&serde_json::json!({}))
                    .unwrap_or_else(|_| Payload::new(aion_core::ContentType::Json, Vec::new())),
                chrono::Utc::now(),
            )
            .with_namespace(namespace)
            .with_task_queue("default")
        }

        /// A namespace store with `namespace` set to `Prefer{nodes}`.
        async fn prefer_store(
            namespace: &str,
            nodes: &[&str],
        ) -> Result<Arc<dyn NamespaceStore>, ServerError> {
            let store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
            store
                .register_namespace(namespace, NamespaceOrigin::Explicit)
                .await?;
            store
                .set_namespace_placement(
                    namespace,
                    NamespacePlacement::Prefer {
                        nodes: labels(nodes),
                    },
                )
                .await?;
            Ok(store)
        }

        /// A namespace store with `namespace` set to `Pinned{nodes}` (P2-I1).
        async fn pinned_store(
            namespace: &str,
            nodes: &[&str],
        ) -> Result<Arc<dyn NamespaceStore>, ServerError> {
            let store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
            store
                .register_namespace(namespace, NamespaceOrigin::Explicit)
                .await?;
            store
                .set_namespace_placement(
                    namespace,
                    NamespacePlacement::Pinned {
                        nodes: labels(nodes),
                    },
                )
                .await?;
            Ok(store)
        }

        /// Build a `RegistryLiminalDispatch` over `registry` whose placement cache
        /// reads `ns_store` (zero TTL so each selection sees the latest placement).
        fn liminal_dispatch(
            registry: &ConnectedWorkerRegistry,
            ns_store: Arc<dyn NamespaceStore>,
        ) -> RegistryLiminalDispatch {
            let cache = PlacementCache::new(ns_store, Duration::ZERO);
            RegistryLiminalDispatch::new(
                registry.clone(),
                Arc::new(NoopCallback),
                DeliveryGate::default(),
            )
            .with_placement_cache(cache)
        }

        /// R3 (#197): a pool whose only worker is eligibility-excluded says SO,
        /// and does not report itself empty.
        ///
        /// The remedies differ and are close to opposite: an empty pool needs a
        /// worker started, an excluded pool needs its liveness verdict read —
        /// and starting a second worker there changes nothing, because the new
        /// one serves the same probation.
        #[tokio::test]
        async fn an_excluded_pool_refusal_names_the_exclusion()
        -> Result<(), Box<dyn std::error::Error>> {
            let registry = ConnectedWorkerRegistry::default();
            let worker = register_unlabelled_worker(&registry, "t")?;
            exclude(&registry, &worker)?;
            let dispatch = RegistryLiminalDispatch::new(
                registry.clone(),
                Arc::new(NoopCallback),
                DeliveryGate::default(),
            );

            let reason = refusal_reason(&dispatch, &unpinned_row("t")).await?;
            assert!(
                reason.contains("NONE is currently dispatch-eligible"),
                "the refusal must name the exclusion it found; said: {reason}"
            );
            assert!(
                !reason.contains("no liminal worker registered"),
                "a pool holding a registered worker is not an empty pool; said: {reason}"
            );
            Ok(())
        }

        /// THE CONTROL for the test above: a genuinely empty pool still reports
        /// itself empty.
        ///
        /// Without it a refusal hard-coded to the exclusion sentence would
        /// satisfy that test and misdiagnose every truly unserved queue,
        /// telling an operator not to start the worker they need.
        #[tokio::test]
        async fn an_empty_pool_refusal_says_nobody_is_registered()
        -> Result<(), Box<dyn std::error::Error>> {
            let registry = ConnectedWorkerRegistry::default();
            let dispatch = RegistryLiminalDispatch::new(
                registry.clone(),
                Arc::new(NoopCallback),
                DeliveryGate::default(),
            );

            let reason = refusal_reason(&dispatch, &unpinned_row("t")).await?;
            assert!(
                reason.contains("no liminal worker registered for the row\'s pool"),
                "an empty pool must say nobody is registered; said: {reason}"
            );
            assert!(
                !reason.contains("starting another worker will not help"),
                "starting a worker is EXACTLY the remedy for an empty pool; said: {reason}"
            );
            Ok(())
        }

        /// 🔴 F3: a `Pinned{{n1}}` namespace with no n1-labelled worker must not
        /// be blamed on an ineligible UNLABELLED worker.
        ///
        /// Selection walked `Required{{n1}}` — the unlabelled worker was never a
        /// candidate, because a hard pin has no `None` spill. Counting the
        /// exclusion over the row\'s own node instead (`None`, which matches
        /// EVERY worker in the pool) makes the refusal say "starting another
        /// worker will not help" when starting an n1-labelled worker is the one
        /// thing that would.
        ///
        /// A refusal that sends an operator away from the only remedy is worse
        /// than the catch-all it replaced, so the count must be taken over the
        /// SAME tier sequence selection actually walked.
        #[tokio::test]
        async fn a_pinned_namespace_miss_is_not_blamed_on_an_ineligible_unlabelled_worker()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = pinned_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            // The only worker in the pool: unlabelled, and excluded. No
            // n1-labelled worker exists at all.
            let worker = register_unlabelled_worker(&registry, "t")?;
            exclude(&registry, &worker)?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let reason = refusal_reason(&dispatch, &unpinned_row("t")).await?;
            assert!(
                !reason.contains("starting another worker will not help"),
                "the excluded worker was never admissible for a hard pin, so it cannot be the \
                 reason this row found nobody — and an n1-labelled worker IS the remedy this \
                 refusal just told the operator not to try. Said: {reason}"
            );
            assert!(
                reason.contains("no liminal worker registered for the row\'s pool"),
                "no worker admissible to the pin is registered, and that is what the refusal \
                 must say; said: {reason}"
            );
            Ok(())
        }

        /// THE CONTROL for the pin test: with an n1-labelled worker present and
        /// excluded, the SAME namespace does report the exclusion.
        ///
        /// Without it, a count that always returned zero under a pin would
        /// satisfy the test above and silently retire the whole R3 distinction
        /// for every pinned namespace.
        #[tokio::test]
        async fn a_pinned_namespace_does_report_an_excluded_admissible_worker()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = pinned_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            let worker = register_node_worker(&registry, "t", "n1")?;
            exclude(&registry, &worker)?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let reason = refusal_reason(&dispatch, &unpinned_row("t")).await?;
            assert!(
                reason.contains("NONE is currently dispatch-eligible"),
                "an n1-labelled worker IS admissible to Pinned{{n1}}, so its exclusion is exactly \
                 why this row found nobody; said: {reason}"
            );
            Ok(())
        }

        /// #163 (prefer): an unpinned row in a `Prefer{n1}` namespace selects the
        /// n1 worker on the liminal path when one is live, even with an n2 worker
        /// also connected.
        #[tokio::test]
        async fn prefer_selects_preferred_node_worker_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = prefer_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            let _n1 = register_node_worker(&registry, "t", "n1")?;
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let row = unpinned_row("t");
            let selected = dispatch
                .select_liminal_worker(&row)
                .await?
                .worker
                .ok_or("a worker must be selected")?;
            assert_eq!(
                selected.node(),
                Some("n1"),
                "the liminal path prefers the n1 worker while it is live"
            );
            // Determinism gate: preference never mutates the recorded row's node.
            assert_eq!(row.node, None, "placement must never mutate the row's node");
            Ok(())
        }

        /// #163 (spill): an unpinned row in a `Prefer{n1}` namespace SPILLS to the
        /// only live worker (n2) on the liminal path when no n1 worker is
        /// connected — the cross-node node-loss failover behaviour.
        #[tokio::test]
        async fn prefer_spills_to_any_live_worker_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = prefer_store("t", &["n1"]).await?;
            // Only an n2 worker is live: no n1-labelled worker exists at all.
            let registry = ConnectedWorkerRegistry::default();
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let row = unpinned_row("t");
            let selected = dispatch
                .select_liminal_worker(&row)
                .await?
                .worker
                .ok_or("the spill must select the live n2 worker")?;
            assert_eq!(
                selected.node(),
                Some("n2"),
                "with no n1 worker live, the liminal selection spills to the live n2 worker"
            );
            assert_eq!(row.node, None, "spill must never mutate the row's node");
            Ok(())
        }

        /// #163 (determinism, mirrors the gRPC `placement_never_mutates_recorded_row_node`
        /// test): under `Prefer{n1}` the SAME unpinned row selected once to the n1
        /// worker and once (after n1 leaves) spilled to n2 keeps `node == None`
        /// BOTH times — selection reads the row's node, never the placement, so
        /// replay sees an identical command stream irrespective of the target.
        #[tokio::test]
        async fn placement_never_mutates_recorded_row_node_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = prefer_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            // Routing A: n1 present -> preferred selection.
            let n1 = register_node_worker(&registry, "t", "n1")?;
            let row_a = unpinned_row("t");
            let selected_a = dispatch
                .select_liminal_worker(&row_a)
                .await?
                .worker
                .ok_or("routing A must select a worker")?;
            assert_eq!(selected_a.node(), Some("n1"));

            // n1 leaves; only n2 remains.
            n1.deregister()?;
            let _n2 = register_node_worker(&registry, "t", "n2")?;

            // Routing B: same shape of unpinned row -> spills to n2.
            let row_b = unpinned_row("t");
            let selected_b = dispatch
                .select_liminal_worker(&row_b)
                .await?
                .worker
                .ok_or("routing B must spill to a worker")?;
            assert_eq!(selected_b.node(), Some("n2"));

            // The recorded row node is None in BOTH routings: the dispatch target
            // (n1 vs n2) did not perturb it.
            assert_eq!(row_a.node, None);
            assert_eq!(row_b.node, None);
            assert_eq!(
                row_a.node, row_b.node,
                "the recorded row node is identical regardless of which worker was selected"
            );
            Ok(())
        }

        /// #164 (P2-I1 hard pin): an unpinned row in a `Pinned{n1}` namespace
        /// selects the n1 worker on the liminal path when live — exactly like
        /// Prefer's happy path.
        #[tokio::test]
        async fn pinned_selects_required_node_worker_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = pinned_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            let _n1 = register_node_worker(&registry, "t", "n1")?;
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let row = unpinned_row("t");
            let selected = dispatch
                .select_liminal_worker(&row)
                .await?
                .worker
                .ok_or("the required n1 worker must be selected")?;
            assert_eq!(selected.node(), Some("n1"));
            assert_eq!(row.node, None, "placement must never mutate the row's node");
            Ok(())
        }

        /// #164 (P2-I1 no spill — the load-bearing test): an unpinned row in a
        /// `Pinned{n1}` namespace with ONLY a live n2 worker selects NOTHING — it
        /// must NEVER spill to the wrong-node worker. This is the exact opposite of
        /// the `prefer_spills_to_any_live_worker_on_liminal_path` behaviour and would
        /// FAIL under the old fall-through (which selected any worker for Pinned).
        /// The `Ok(None)` drives the outbox no-worker retry/stall, mirroring the
        /// gRPC wait.
        #[tokio::test]
        async fn pinned_never_spills_to_a_wrong_node_worker_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = pinned_store("t", &["n1"]).await?;
            // Only an n2 worker is live: no n1-labelled worker exists at all.
            let registry = ConnectedWorkerRegistry::default();
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let row = unpinned_row("t");
            let selected = dispatch.select_liminal_worker(&row).await?.worker;
            assert!(
                selected.is_none(),
                "Pinned{{n1}} must NOT spill to the live n2 worker — it selects nothing \
                 so the outbox retries/stalls until an n1 worker returns"
            );
            assert_eq!(row.node, None, "placement must never mutate the row's node");
            Ok(())
        }

        /// #163 (authored pin wins): a row authored-pinned to `Some(n2)` STILL
        /// selects an n2 worker on the liminal path regardless of the namespace's
        /// `Prefer{n1}` — the per-activity pin is authoritative and placement never
        /// overrides it.
        #[tokio::test]
        async fn authored_node_pin_wins_over_namespace_prefer_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store = prefer_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            let _n1 = register_node_worker(&registry, "t", "n1")?;
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            // Authored pin: node = Some("n2").
            let row = unpinned_row("t").with_node(Some(String::from("n2")));
            let selected = dispatch
                .select_liminal_worker(&row)
                .await?
                .worker
                .ok_or("the authored pin must select the n2 worker")?;
            assert_eq!(
                selected.node(),
                Some("n2"),
                "the authored Some(n2) pin is honoured regardless of the namespace Prefer{{n1}}"
            );
            // The authored node is preserved exactly (determinism gate).
            assert_eq!(row.node.as_deref(), Some("n2"));
            Ok(())
        }

        /// #163 (byte-identical default): an `Unplaced` namespace selects any live
        /// worker on the liminal path exactly as the pre-Phase-2 single
        /// `select_worker` would — the ceiling/placement never engages.
        #[tokio::test]
        async fn unplaced_namespace_selects_any_worker_on_liminal_path()
        -> Result<(), Box<dyn std::error::Error>> {
            let ns_store: Arc<dyn NamespaceStore> = Arc::new(InMemoryStore::default());
            // Registered but left Unplaced (the default placement).
            ns_store
                .register_namespace("t", NamespaceOrigin::Explicit)
                .await?;
            let registry = ConnectedWorkerRegistry::default();
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            let dispatch = liminal_dispatch(&registry, Arc::clone(&ns_store));

            let selected = dispatch
                .select_liminal_worker(&unpinned_row("t"))
                .await?
                .worker
                .ok_or("an Unplaced namespace still selects a live worker")?;
            assert_eq!(
                selected.node(),
                Some("n2"),
                "an Unplaced namespace reaches any live worker, exactly as before"
            );
            Ok(())
        }

        /// #163 (byte-identical, no cache): with NO placement cache attached, the
        /// liminal selection is the single `select_worker` off the row's own node —
        /// byte-identical to the pre-#163 construction. An unpinned row reaches any
        /// live worker; the namespace's `Prefer` is not even consulted.
        #[tokio::test]
        async fn no_cache_selection_is_byte_identical_to_pre_163()
        -> Result<(), Box<dyn std::error::Error>> {
            // The namespace prefers n1, but with no cache the preference is ignored.
            let _ns_store = prefer_store("t", &["n1"]).await?;
            let registry = ConnectedWorkerRegistry::default();
            let _n2 = register_node_worker(&registry, "t", "n2")?;
            // No `.with_placement_cache(...)`: the pre-#163 construction.
            let dispatch = RegistryLiminalDispatch::new(
                registry.clone(),
                Arc::new(NoopCallback),
                DeliveryGate::default(),
            );

            let selected = dispatch
                .select_liminal_worker(&unpinned_row("t"))
                .await?
                .worker
                .ok_or("without a cache the unpinned row still selects any worker")?;
            assert_eq!(
                selected.node(),
                Some("n2"),
                "with no placement cache the selection is the unchanged any-worker path"
            );
            Ok(())
        }
    }
}