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
//!
//! - [`LiminalWorkerDelivery`] is the per-worker push handle this module hands
//!   to the registry at registration. Delivering over it is
//!   [`LiminalTaskDelivery`](super::liminal_task_delivery::LiminalTaskDelivery),
//!   the liminal arm of the shared delivery seam, which pushes the
//!   [`DispatchRequest`] out on the worker's existing connection and re-enters
//!   the worker's [`DispatchResponse`] through the SAME
//!   [`LiminalCompletionSource`] / [`OutboxDeliveryCallback`] the gRPC
//!   completion path uses.
//!
//!   🔴 This module does NOT select a worker and no longer implements
//!   [`OutboxRowDispatch`](super::outbox_dispatcher::OutboxRowDispatch). It used
//!   to, through a `RegistryLiminalDispatch` that ran its own copy of the
//!   selection — and a server with `outbox.transport = liminal` therefore
//!   reached EVERY selected worker over liminal, refusing the gRPC-registered
//!   ones it had just selected. That was #52. Selection now happens once, in
//!   [`ActivityDispatcher`](super::dispatch::ActivityDispatcher), and each
//!   chosen worker is served over the transport IT registered on.
//! - [`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::future::Future;
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};
use super::registry::{ConnectedWorkerRegistry, WorkerDelivery, WorkerRegistration};
use crate::error::ServerError;
use crate::namespace::{CallerIdentity, NamespaceGuard};

/// 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,
}

/// 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(),
        )?;
        match self
            .completion_fences
            .accept(&response.workflow_id, &activity_id, &completion_token)
        {
            Ok(accepted) => {
                // The consumed generation is bound and deliberately NOT
                // restored: unlike the bridge, this path has no settlement that
                // can fail after acceptance — the callback below IS the
                // settlement and its failure is returned straight to the
                // caller — so there is no `restore_if_absent` site on this
                // transport for the slip to feed.
                tracing::debug!(
                    workflow_id = %response.workflow_id,
                    %activity_id,
                    attempt = accepted.attempt(),
                    "liminal completion consumed the execution generation"
                );
            }
            Err(error) => {
                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::push_dispatch`] can push a
/// [`DispatchRequest`] out on the worker's existing socket (the LSUB-0
/// server-push primitive) and [`receive_bridge_reply`] can 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 dispatch and wait for its reply without an activity-duration
    /// bound, abandoning only when `keep_waiting` becomes false at a poll
    /// boundary — [`Self::push_dispatch`] then [`receive_bridge_reply`] in one
    /// call.
    ///
    /// The production arms (the outbox delivery and the engine-seam bridge) do
    /// NOT call this: they call the two halves themselves because the lease
    /// accept hook ([`super::task_delivery::DeliveryAccepted`]) must run between
    /// the push being acknowledged and the reply wait — on liminal the reply IS
    /// the completion, and a lease recorded after it would be a lease recorded
    /// after the fact. This composition is for callers outside the crate that
    /// stand in for a dispatcher (the e2e harnesses), where the receive half is
    /// not reachable on its own.
    ///
    /// # 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: the outbox arm (`LiminalTaskDelivery`) and the
    /// engine-seam bridge dispatcher both push here and wait with
    /// [`receive_bridge_reply`], matching the gRPC bridge contract. One frame
    /// format, one push primitive, one held wait policy. The two halves are
    /// separate calls so the attempt's lease can be recorded between them,
    /// at the instant the push is acknowledged (WA-010 R3).
    ///
    /// 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 arm and the bridge — both wait through
/// [`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 { .. })
}

/// 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 namespace admission every registration passes through, bound by
    /// the commissioning path. `None` FAILS CLOSED: a notifier nobody bound an
    /// admission to refuses every registration by name rather than admitting
    /// unguarded, which is exactly the defect this field closes.
    admission: Option<RegistrationAdmission>,
}

/// 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,
            admission: None,
        }
    }

    /// Bind the namespace admission a liminal registration is judged by —
    /// the same guard, mint-or-gate and placement admission the gRPC path
    /// applies — and the Tokio runtime `handle` the synchronous
    /// connection-process callback bridges onto to run it.
    ///
    /// The handle is named, never captured from ambient context: the e2e
    /// harnesses build this notifier in a sync `start()` BEFORE they build
    /// their runtime, and a capture there was `None`, refusing every
    /// registration on the very tests meant to exercise it while the
    /// production path (built inside `run_server`) happened to work. A
    /// listener commissioned outside any runtime has no handle to give and
    /// refuses to be commissioned, at the call site, by name.
    ///
    /// `auth_enabled` mirrors `auth.enabled`: the liminal protocol's
    /// registration frame carries an `identity` string and no credential, so
    /// on a server that authenticates callers a liminal registration cannot
    /// be scoped and is refused by name; on an auth-off server it is admitted
    /// as the operator identity, exactly as a gRPC registration with no
    /// metadata is.
    #[must_use]
    pub fn with_admission(
        mut self,
        guard: NamespaceGuard,
        auth_enabled: bool,
        handle: tokio::runtime::Handle,
    ) -> Self {
        self.admission = Some(RegistrationAdmission {
            guard,
            auth_enabled,
            handle,
        });
        self
    }

    /// 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> {
        // Admit through the SAME gate a gRPC worker passes: guard policy, each
        // namespace authorized, mint-or-gate, placement admission, then the
        // insert. A notifier with no admission bound refuses — it never
        // inserts unguarded.
        let Some(admission) = &self.admission else {
            return Err(LiminalServerError::ListenerAccept {
                message: format!(
                    "liminal worker registration for connection {pid} rejected: this listener \
                     has no namespace admission bound, so it cannot judge which namespaces the \
                     worker may serve; a listener is commissioned with one before it accepts"
                ),
            });
        };
        if admission.auth_enabled {
            return Err(LiminalServerError::ListenerAccept {
                message: format!(
                    "liminal worker registration for connection {pid} (identity `{}`) rejected: \
                     this server authenticates callers (auth.enabled = true) and the liminal \
                     registration frame carries no credential to authenticate, so the worker's \
                     namespaces cannot be scoped to a caller; register this worker over gRPC \
                     with a bearer token, or run the server with auth disabled",
                    registration.identity
                ),
            });
        }
        // 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 caller = liminal_caller_identity(&registration.identity);
        let proto = proto_registration_from_wire(registration);
        let guard = bridge_admission(
            &admission.handle,
            self.registry.admit_delivery(
                &admission.guard,
                &caller,
                &proto,
                delivery,
                self.intervention_capabilities.clone(),
            ),
        )
        .map_err(|error| LiminalServerError::ListenerAccept {
            message: format!("liminal worker registration for connection {pid} rejected: {error}"),
        })?
        .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
            && 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(())
    }
}

/// The admission a liminal registration is judged by, bound at commissioning.
struct RegistrationAdmission {
    guard: NamespaceGuard,
    auth_enabled: bool,
    /// The runtime the async admission is bridged onto from the connection
    /// thread — named at commissioning, never captured from ambient context.
    handle: tokio::runtime::Handle,
}

/// Run the async admission to completion from the synchronous connection
/// callback, on whichever thread liminal invokes it from.
///
/// Three callers, three answers, no panic path:
/// - liminal's own connection thread (production) is on no runtime, so the
///   future is driven by `handle.block_on` directly;
/// - a multi-thread runtime worker (a test that drives the notifier from an
///   async body) cannot be blocked without stalling its scheduler, so the
///   worker is first handed back via `block_in_place`;
/// - a current-thread runtime worker has nothing to hand back to — blocking
///   it would deadlock the very runtime the admission runs on — so the call
///   is refused by name rather than hung.
///
/// # Errors
///
/// Returns the refusal for the current-thread case; the admission's own
/// verdict rides in the `Ok`.
fn bridge_admission<F: Future>(
    handle: &tokio::runtime::Handle,
    admission: F,
) -> Result<F::Output, ServerError> {
    match tokio::runtime::Handle::try_current() {
        Err(_) => Ok(handle.block_on(admission)),
        Ok(current) => match current.runtime_flavor() {
            tokio::runtime::RuntimeFlavor::MultiThread => {
                Ok(tokio::task::block_in_place(|| handle.block_on(admission)))
            }
            _ => Err(ServerError::Config {
                message: "the liminal registration callback ran on a current-thread Tokio \
                          runtime worker, which cannot be blocked while the namespace \
                          admission runs; drive the listener from a multi-thread runtime \
                          or from a plain thread, as liminal's connection process does"
                    .to_owned(),
            }),
        },
    }
}

/// The caller a liminal registration is admitted as on an auth-off server:
/// the operator identity, subject = the frame's `identity` (or `operator`
/// when the frame left it empty) — the liminal twin of the gRPC development
/// identity built from `x-aion-subject`.
fn liminal_caller_identity(identity: &str) -> CallerIdentity {
    let subject = identity.trim();
    CallerIdentity::operator(if subject.is_empty() {
        "operator"
    } else {
        subject
    })
}

fn wire_activity_to_proto(
    activity: &liminal::protocol::WorkerActivityDescriptor,
) -> aion_proto::ProtoActivityDescriptor {
    aion_proto::ProtoActivityDescriptor {
        name: activity.name.clone(),
        input_schema_json: activity.input_schema_json.clone(),
        output_schema_json: activity.output_schema_json.clone(),
    }
}

/// The liminal registration frame in the shape the namespace guard judges:
/// the same fields, the wire's `None` node as the proto's empty string.
fn proto_registration_from_wire(
    registration: &WireWorkerRegistration,
) -> aion_proto::ProtoRegisterWorker {
    aion_proto::ProtoRegisterWorker {
        namespaces: registration.namespaces.clone(),
        activity_types: registration.activity_types.clone(),
        task_queue: registration.task_queue.clone(),
        node: normalize_wire_node(registration.node.as_deref()).unwrap_or_default(),
        activities: registration
            .activities
            .iter()
            .map(wire_activity_to_proto)
            .collect(),
        identity: registration.identity.clone(),
        instance: None,
    }
}

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)
            && 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
                && 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 crate::worker::registry::RegistrationOptions;
    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(
            [String::from("default")],
            String::from("default"),
            None,
            types.iter(),
            WorkerDelivery::Grpc(tx),
            RegistrationOptions::identified(String::from("liminal-transport-worker"))
                .with_intervention_capabilities(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,
            started_attempt: 1,
            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 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()));
    }
}