aion-server 0.31.0

//! The liminal arm of the delivery seam: push the dispatch out on the worker's
//! existing connection and block for its correlated reply.
//!
//! # Why this arm is the one that needed a seam
//!
//! Before #52 there was no way to reach it from the worker dispatcher at all: a
//! worker with no gRPC stream sender was deregistered on the premise that its
//! presence could only be a leak. This module is the same delivery sequence the
//! outbox arm has always run, lifted so both callers reach one implementation.
//!
//! # What it does NOT own any more
//!
//! Two responsibilities deliberately stayed with the caller, and moving them
//! here would break the invariants they exist to keep:
//!
//! - **The completion token.** One per pass, minted before the candidate walk
//!   and revoked once when the pass places nothing. Minting per delivery would
//!   put a second authorization beside the one a worker may still be holding for
//!   the same attempt.
//! - **The abandonment condition.** See
//!   [`DeliveryIntent`](super::delivery_intent::DeliveryIntent): one of its terms
//!   is unanswerable from inside a transport.

use std::sync::Arc;

use async_trait::async_trait;

use aion_core::{ActivityId, RunId, WorkflowId};
use aion_proto::ProtoActivityTask;

use super::delivery_intent::SharedDeliveryIntent;
use super::intervention::{AttemptKey, AttemptOwnerIndex};
use super::liminal_transport::{AttemptOwnerGuard, DispatchRequest, LiminalCompletionSource};
use super::registry::{WorkerDelivery, WorkerHandle};
use super::task_delivery::{DeliveryAccepted, LivenessTracking, TaskDelivery, WorkerTaskDelivery};

/// Delivers by pushing the dispatch out on the worker's liminal connection and
/// blocking for the correlated reply.
///
/// The reply **is** the activity's completion, so this arm re-enters it through
/// the same completion path the gRPC transport's out-of-band completion uses.
pub struct LiminalTaskDelivery {
    completion: Arc<LiminalCompletionSource>,
    /// NOI-6 `attempt -> owning-worker` back-index. `None` (every non-agent
    /// deployment) skips the binding, exactly as the outbox arm does —
    /// intervention is then simply never offered.
    attempt_owners: Option<AttemptOwnerIndex>,
    /// The liveness tracker and registry this arm retires a completed dispatch
    /// from, when the dispatcher above it tracks its dispatches.
    ///
    /// The liminal delivery is the ONLY seam that can do this. Unlike gRPC —
    /// where the frame is queued, `Delivered` is returned, and the result comes
    /// back later on the worker's stream for the session loop to untrack — this
    /// arm waits for the reply inline and routes the completion itself. There is
    /// no later moment and no other holder: if this does not retire the entry,
    /// nothing does, and the dispatch stays counted against the worker's
    /// capacity for the life of the registration.
    ///
    /// `None` where the dispatcher above tracks nothing, which is every
    /// in-process façade and any test whose subject is not capacity.
    completion_tracking: Option<CompletionTracking>,
}

/// What the liminal arm needs to retire a completed dispatch from the liveness
/// tracker: the tracker itself and the registry whose capacity count it feeds.
#[derive(Clone)]
struct CompletionTracking {
    heartbeat_tracker: crate::worker::heartbeat::HeartbeatTracker,
    registry: crate::worker::ConnectedWorkerRegistry,
}

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

impl LiminalTaskDelivery {
    /// Build the liminal delivery arm over the completion sink both transports
    /// share.
    #[must_use]
    pub fn new(completion: Arc<LiminalCompletionSource>) -> Self {
        Self {
            completion,
            attempt_owners: None,
            completion_tracking: None,
        }
    }

    /// Retire a completed dispatch from the liveness tracker when the reply
    /// lands, so this transport's completion removes its tracked entry exactly
    /// as the gRPC session loop's result arm does.
    ///
    /// Required wherever the dispatcher above this arm tracks its dispatches
    /// (`ActivityDispatcher::with_heartbeat_tracker`). Without it the entry made
    /// before the push is never removed on this transport, and the worker's
    /// advertised capacity is consumed permanently, one slot per delivery.
    #[must_use]
    pub fn with_completion_tracking(
        mut self,
        heartbeat_tracker: crate::worker::heartbeat::HeartbeatTracker,
        registry: crate::worker::ConnectedWorkerRegistry,
    ) -> Self {
        self.completion_tracking = Some(CompletionTracking {
            heartbeat_tracker,
            registry,
        });
        self
    }

    /// Install the NOI-6 attempt-owner back-index, so a dispatched attempt binds
    /// its owning worker before the push and the intervention router can resolve
    /// the current owner of a live attempt.
    #[must_use]
    pub fn with_attempt_owners(mut self, attempt_owners: AttemptOwnerIndex) -> Self {
        self.attempt_owners = Some(attempt_owners);
        self
    }
}

/// Untracks a liminal dispatch if this delivery is ABANDONED before its
/// completion is routed.
///
/// The liminal arm waits for the worker's reply inline, so its future can be
/// cancelled mid-wait — a caller that withdraws, or a dispatcher whose runtime is
/// dropped out from under it during a failover. A cancelled future does not
/// return, so neither the delivery's own completion path nor the dispatcher's
/// non-`Delivered` arm ever runs, and the entry tracked before the push would
/// simply stay: a slot the worker never gets back, on a registry that in a
/// multi-server deployment can outlive the dispatcher that made the entry.
///
/// Disarmed the moment the completion has retired the entry itself, so the
/// normal path does not untrack twice (it would be idempotent by key anyway —
/// this only avoids the pointless second call).
struct AbandonedDispatchGuard<'a> {
    tracking: &'a CompletionTracking,
    worker_id: crate::worker::WorkerId,
    workflow_id: &'a WorkflowId,
    activity_id: &'a ActivityId,
    armed: bool,
}

impl AbandonedDispatchGuard<'_> {
    /// The completion retired the entry; this guard has nothing left to do.
    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for AbandonedDispatchGuard<'_> {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }
        let retired = crate::worker::bridge::clear_completed_task_tracking(
            &self.tracking.heartbeat_tracker,
            &self.tracking.registry,
            self.worker_id,
            self.workflow_id,
            self.activity_id,
        );
        if retired {
            tracing::warn!(
                worker_id = ?self.worker_id,
                workflow_id = %self.workflow_id,
                activity_id = %self.activity_id,
                "liminal delivery was abandoned before the worker replied; its tracked slot has \
                 been returned and the row re-drives"
            );
        }
    }
}

/// The identity a delivery needs, resolved from the wire task **before** any
/// owner binding is taken.
///
/// 🔴 This type exists to make the ordering structural rather than positional.
/// The owner binding is keyed on the run, so a binding taken before the run is
/// resolved would let an intervention aimed at one continue-as-new generation
/// resolve another generation's worker.
///
/// # Exactly how strong this is, measured rather than asserted
///
/// The binding is taken by [`Resolved::bind_owner`], a method **on the resolved
/// identity**, so it cannot be called before that identity exists: moving the
/// call above the resolution does not compile. That is the mutation this guards
/// against, and it has been run.
///
/// It is **not** proof against a deliberate bypass. [`AttemptKey::new`] is a
/// public constructor over four plain components, so an edit that calls
/// [`AttemptOwnerGuard::bind`] directly with a placeholder run compiles and
/// binds. An earlier version of this comment claimed the key "can only be built
/// from a `Resolved`", which was false — the mutation that proved it survived
/// both this structure and the test below.
///
/// The bypass is also, for the same reason, invisible at runtime: the guard
/// releases on drop, so a binding taken before a refusal is gone again before
/// any caller can look. What defends the property is that the natural edit does
/// not compile and the deliberate one is a different API call, which review can
/// see.
struct Resolved {
    workflow_id: WorkflowId,
    run_id: RunId,
    activity_id: ActivityId,
    attempt: u32,
}

impl Resolved {
    /// Resolve the task's identity, refusing a task with no run.
    ///
    /// The refusal is the pre-existing one — an activity without a run cannot be
    /// dispatched at all, because an unfenced external effect has no generation
    /// to belong to.
    fn from_task(task: &ProtoActivityTask) -> Result<Self, &'static str> {
        let workflow_id = task
            .workflow_id
            .clone()
            .ok_or("task carries no workflow id")?
            .try_into()
            .map_err(|_| "task carries a malformed workflow id")?;
        let run_id: RunId = task
            .run_id
            .clone()
            .ok_or("activity run id is missing; refusing unfenced external effect")?
            .try_into()
            .map_err(|_| "task carries a malformed run id")?;
        // Infallible: `ProtoActivityId` is the sequence position, and
        // `ActivityId` is a newtype over it. Only its ABSENCE can be refused.
        let activity_id: ActivityId = task
            .activity_id
            .ok_or("task carries no activity id")?
            .into();
        Ok(Self {
            workflow_id,
            run_id,
            activity_id,
            attempt: task.attempt,
        })
    }

    /// The intervention key for this attempt, carrying the **resolved** run.
    fn attempt_key(&self) -> AttemptKey {
        AttemptKey::new(
            self.workflow_id.clone(),
            self.run_id.clone(),
            self.activity_id.clone(),
            self.attempt,
        )
    }

    /// Bind this attempt's owner, returning the guard that releases it.
    ///
    /// 🔴 A method on the **resolved** identity on purpose. The binding is keyed
    /// on the run, so taking it before the run is resolved is the defect; making
    /// it a method on `Resolved` means the call cannot be moved above the
    /// resolution — there is no receiver for it there, and the lift does not
    /// compile. See this type's docs for what that does and does not prove.
    fn bind_owner(
        &self,
        owners: Option<&AttemptOwnerIndex>,
        worker: super::registry::WorkerId,
    ) -> Option<AttemptOwnerGuard> {
        owners.map(|owners| AttemptOwnerGuard::bind(owners.clone(), self.attempt_key(), worker))
    }

    /// The outbox ordinal this activity id was derived from.
    ///
    /// [`ActivityId`] is a newtype over the scheduling sequence position, so
    /// `from_sequence_position` is exactly invertible and the ordinal the wire
    /// needs round-trips without the row being present.
    fn ordinal(&self) -> u64 {
        self.activity_id.sequence_position()
    }
}

/// Build the liminal wire request from the already-built task.
///
/// Every field is carried from the task except `heartbeat_window_ms`, which the
/// task does not have — see [`LivenessTracking`] for why that assignment is a named
/// value rather than a bare zero.
/// The [`TaskDelivery`] a typed liminal push or reply failure becomes, BY
/// CLASS. An unservable frame (proved larger than the connection's whole
/// outbound buffer) is the terminal [`Undeliverable::Unservable`]; every other
/// failure stays the alive-worker [`Undeliverable::DeliveryFailed`] this arm
/// has always reported. The class must survive this boundary: the candidate
/// walk and the outbox dispatcher read the TYPE, and a refusal flattened to a
/// string here was re-pushed once per attempt in the budget, recording one
/// lease per push for a step that could never run.
fn undelivered_by(error: &crate::error::ServerError) -> TaskDelivery {
    if error.is_worker_dispatch_unservable() {
        TaskDelivery::unservable(format!("liminal dispatch is unservable: {error}"))
    } else {
        TaskDelivery::failed(format!("liminal dispatch failed: {error}"))
    }
}

fn request_for_task(task: &ProtoActivityTask, resolved: &Resolved) -> DispatchRequest {
    DispatchRequest {
        activity_type: task.activity_type.clone(),
        workflow_id: resolved.workflow_id.clone(),
        ordinal: resolved.ordinal(),
        run_id: Some(resolved.run_id.clone()),
        completion_token: task.completion_token.clone(),
        idempotency_key: task.idempotency_key.clone(),
        input: task
            .input
            .as_ref()
            .map(|payload| payload.bytes.clone())
            .unwrap_or_default(),
        attempt: resolved.attempt,
        labels: task
            .labels
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect(),
        heartbeat_window_ms: LivenessTracking::NotTrackedPerTask.heartbeat_window_ms(),
    }
}

#[async_trait]
impl WorkerTaskDelivery for LiminalTaskDelivery {
    async fn deliver(
        &self,
        worker: &WorkerHandle,
        task: &ProtoActivityTask,
        intent: &SharedDeliveryIntent,
        accepted: &dyn DeliveryAccepted,
    ) -> TaskDelivery {
        // Identity is resolved FIRST — before the transport is even consulted —
        // and the binding below is a method on what this produces, so it cannot
        // be moved above this line.
        //
        // Ahead of the transport check deliberately. A task with no run is
        // malformed **regardless of which worker it was aimed at**, so refusing
        // it with "not delivered over liminal" would name the wrong defect and
        // send an operator after the wrong remedy.
        let resolved = match Resolved::from_task(task) {
            Ok(resolved) => resolved,
            Err(reason) => return TaskDelivery::failed(reason),
        };

        let delivery = match worker.delivery() {
            WorkerDelivery::Liminal(delivery) => delivery.clone(),
            WorkerDelivery::Grpc(_) => {
                // A caller chose the wrong transport for the worker it selected.
                // The worker is alive and correctly registered, so its
                // registration stands — this is #52's defect refused at its
                // mirror site.
                return TaskDelivery::failed(
                    "selected worker is not delivered over liminal; the liminal transport cannot \
                     reach it",
                );
            }
        };

        // NOI-6: bind this attempt's owner BEFORE the push, so an intervention
        // that races the dispatch resolves the worker. The guard releases on
        // every exit path (reply, error, panic) so the index never keeps a
        // finished attempt.
        let _owner_guard = resolved.bind_owner(self.attempt_owners.as_ref(), worker.id());

        let request = request_for_task(task, &resolved);
        // The push and the wait are two steps on purpose: the moment the push
        // is acknowledged the worker HOLDS the attempt, and the lease is
        // recorded THERE — the reply this arm then waits for IS the completion,
        // so a lease recorded after the wait would land after the terminal it
        // exists to precede (WA-010 R3). Both steps are blocking, thread-based
        // liminal calls; each runs off the async runtime so a long-running
        // activity cannot starve a runtime worker. The caller's intent is
        // re-asked at every poll boundary of the wait, which is why it arrives
        // behind an `Arc`.
        // ARMED BEFORE THE PUSH. The tracker entry already exists (the caller
        // tracked before handing the task here), so its cleanup obligation
        // exists from this line — and the push itself is the FIRST await this
        // future can be cancelled at. A cancelled future runs no arm of any
        // match below it, so a guard armed after the push left exactly one
        // uncovered window: a cancellation while the push was in flight
        // leaked the tracked entry. The push-failure returns disarm explicitly:
        // nothing was delivered, and the caller's non-`Delivered` arm untracks.
        // The reply-wait `None`/`Err`, the join error, and the completion-record
        // failure do NOT disarm — on those the guard fires on drop AND the
        // caller untracks, and that double untrack is safe only because
        // `TaskTracker::complete_task` frees a slot solely for the removal
        // that found the entry (`was_tracked`); the second finds nothing and
        // decrements nothing. That idempotence is load-bearing here: an
        // unconditional decrement in `complete_task` would re-open an
        // over-push through this path.
        let mut abandoned =
            self.completion_tracking
                .as_ref()
                .map(|tracking| AbandonedDispatchGuard {
                    tracking,
                    worker_id: worker.id(),
                    workflow_id: &resolved.workflow_id,
                    activity_id: &resolved.activity_id,
                    armed: true,
                });
        let push_delivery = delivery.clone();
        let pushed =
            tokio::task::spawn_blocking(move || push_delivery.push_dispatch(&request)).await;
        let awaiter = match pushed {
            Ok(Ok(awaiter)) => awaiter,
            Ok(Err(error)) => {
                if let Some(guard) = abandoned.as_mut() {
                    guard.disarm();
                }
                return undelivered_by(&error);
            }
            Err(error) => {
                if let Some(guard) = abandoned.as_mut() {
                    guard.disarm();
                }
                return TaskDelivery::failed(format!("dispatch task join failed: {error}"));
            }
        };
        accepted.accepted().await;
        let waiting_intent = Arc::clone(intent);
        let dispatched = tokio::task::spawn_blocking(move || {
            super::liminal_transport::receive_bridge_reply(&awaiter, || {
                waiting_intent.still_wanted()
            })
        })
        .await;

        let response = match dispatched {
            Ok(Ok(Some(response))) => response,
            Ok(Ok(None)) => {
                // The caller withdrew while the delivery waited. The worker is
                // alive and correctly registered; only this pass gave up, and a
                // late reply is discarded by the fences.
                return TaskDelivery::failed("delivery wait abandoned before worker reply");
            }
            Ok(Err(error)) => return undelivered_by(&error),
            Err(error) => {
                return TaskDelivery::failed(format!("dispatch task join failed: {error}"));
            }
        };

        // Re-enter the worker's result through the SAME completion path the gRPC
        // transport uses; terminal dedup applies unchanged.
        //
        // 🔴 A failure HERE is the one outcome whose name reads narrower than its
        // meaning: the worker took the task, executed it, and replied — and the
        // reply did not record. `Delivered` would be wrong (the caller would
        // settle the work without its result) and `WorkerUnreachable` would be
        // wrong twice over (the worker is demonstrably alive, and deregistering
        // it would destroy a healthy registration). `DeliveryFailed` carries the
        // right obligations — registration stands, the caller withdraws its
        // token — which is what the type encodes and what the outbox arm has
        // always done here.
        if let Err(error) = self.completion.deliver(&response) {
            return TaskDelivery::failed(format!(
                "worker replied but the completion could not be recorded: {error}"
            ));
        }
        // THE COMPLETION RETIRES THE TRACKED ENTRY, here and nowhere else.
        //
        // This transport routes the completion itself, inline, before it returns
        // — so unlike gRPC there is no later stream frame for a session loop to
        // untrack on. The dispatcher above tracks before the push (the bridge's
        // `track_then_send` ordering); this is the matching half. Skipping it
        // leaves a phantom entry that nothing nominates, nothing logs, and
        // nothing removes: the worker's capacity shrinks by one per delivery and
        // `in_flight_count` never returns to zero.
        //
        // Idempotent by key — `clear_completed_task_tracking` reports whether it
        // actually retired anything — so if some other holder ever sees the same
        // completion this is a no-op rather than a second decrement.
        if let Some(tracking) = &self.completion_tracking {
            let _ = crate::worker::bridge::clear_completed_task_tracking(
                &tracking.heartbeat_tracker,
                &tracking.registry,
                worker.id(),
                &resolved.workflow_id,
                &resolved.activity_id,
            );
        }
        if let Some(guard) = abandoned.as_mut() {
            guard.disarm();
        }
        TaskDelivery::Delivered
    }
}

#[cfg(test)]
mod tests {
    use crate::worker::registry::RegistrationOptions;

    /// A delivery that records whether its accept point was reached, so a
    /// refusal can be pinned as never having handed the attempt over.
    #[derive(Default)]
    struct NeverAccepted {
        reached: std::sync::atomic::AtomicBool,
    }

    #[async_trait::async_trait]
    impl super::DeliveryAccepted for NeverAccepted {
        async fn accepted(&self) {
            self.reached
                .store(true, std::sync::atomic::Ordering::SeqCst);
        }
    }
    use std::sync::Arc;

    use aion_core::{ActivityId, InterventionCapabilities, RunId, WorkflowId};
    use aion_proto::{ProtoActivityId, ProtoActivityTask, ProtoWorkflowId};
    use uuid::Uuid;

    use crate::error::ServerError;
    use crate::worker::bridge::OutboxDeliveryCallback;
    use crate::worker::delivery_intent::{AlwaysWanted, SharedDeliveryIntent};
    use crate::worker::intervention::AttemptOwnerIndex;
    use crate::worker::liminal_transport::LiminalCompletionSource;
    use crate::worker::registry::{ConnectedWorkerRegistry, WorkerDelivery};
    use crate::worker::task_delivery::{TaskDelivery, WorkerTaskDelivery};

    use super::{LiminalTaskDelivery, Resolved};

    /// The completion sink is never reached by these tests — every one of them
    /// refuses before a worker is pushed to — so both methods report "no live
    /// run" if they are ever called, rather than pretending to succeed.
    struct NoopCallback;

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

    const WORKFLOW: u128 = 0x51;

    /// A task carrying everything a delivery needs EXCEPT a run id.
    fn task_without_a_run() -> ProtoActivityTask {
        ProtoActivityTask {
            workflow_id: Some(ProtoWorkflowId::from(WorkflowId::new(Uuid::from_u128(
                WORKFLOW,
            )))),
            activity_id: Some(ProtoActivityId::from(ActivityId::from_sequence_position(3))),
            activity_type: String::from("agent"),
            input: None,
            attempt: 1,
            labels: std::collections::HashMap::new(),
            // The whole point of this fixture.
            run_id: None,
            completion_token: String::from("token"),
            idempotency_key: String::from("key"),
        }
    }

    fn liminal_delivery(owners: &AttemptOwnerIndex) -> LiminalTaskDelivery {
        LiminalTaskDelivery::new(Arc::new(LiminalCompletionSource::new(Arc::new(
            NoopCallback,
        ))))
        .with_attempt_owners(owners.clone())
    }

    /// 🔴 NAMED PROPERTY: the run is resolved BEFORE any attempt owner is bound.
    ///
    /// # What breaks without it
    ///
    /// The owner binding is keyed on the run. A binding taken before the run is
    /// resolved would let an intervention aimed at one continue-as-new
    /// generation resolve a DIFFERENT generation's worker — an operator
    /// stopping run A reaching the worker executing run B.
    ///
    /// # 🔴 What this test does and does NOT witness — measured by mutation
    ///
    /// It witnesses the **refusal**: a run-less task is rejected, is diagnosed
    /// by its missing run rather than by the transport, does not deregister the
    /// worker, and leaves no owner binding behind.
    ///
    /// It does **not** witness the ordering, and an earlier version of this
    /// comment claimed it did. The mutation — lifting the bind above the
    /// resolution with a placeholder run — was run, and **this test stayed
    /// green**, because [`AttemptOwnerGuard`] releases on drop: a binding taken
    /// before a refusal is already gone by the time any caller can look. There
    /// is no observation window on this path, and there cannot be one.
    ///
    /// The ordering is defended structurally instead: the bind is a method on
    /// [`Resolved`], so the natural lift does not compile. That claim was also
    /// checked by mutation rather than asserted — see [`Resolved`] for exactly
    /// how far it goes, including the bypass it does not stop.
    #[tokio::test]
    async fn run_resolution_precedes_attempt_owner_binding()
    -> Result<(), Box<dyn std::error::Error>> {
        let owners = AttemptOwnerIndex::new();
        let delivery = liminal_delivery(&owners);

        // A real registration yields a real WorkerId; no id here is fabricated.
        let registry = ConnectedWorkerRegistry::default();
        let (sender, _receiver) = 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(sender),
            RegistrationOptions::identified(
                "agent-1",
                crate::worker::UNBOUNDED_SENDER_WORKER_CONCURRENCY,
            )
            .with_intervention_capabilities(InterventionCapabilities::none()),
        )?;
        let worker_id = registration
            .worker_id()
            .ok_or("a registration must assign a worker id")?;
        let worker = registry
            .worker_by_id(worker_id)?
            .ok_or("the worker just registered must be readable")?;

        let intent: SharedDeliveryIntent = Arc::new(AlwaysWanted);
        let accepted = NeverAccepted::default();
        let outcome = delivery
            .deliver(&worker, &task_without_a_run(), &intent, &accepted)
            .await;
        assert!(
            !accepted.reached.load(std::sync::atomic::Ordering::SeqCst),
            "a task refused by its missing run must never reach the accept point"
        );

        match outcome {
            TaskDelivery::Delivered => {
                return Err("a task with no run must never be delivered".into());
            }
            TaskDelivery::Undeliverable(undeliverable) => {
                // The refusal names the RUN, not the transport: identity is
                // resolved before the transport is consulted, so a run-less task
                // is diagnosed as run-less whichever worker it was aimed at.
                assert!(
                    undeliverable.reason().contains("run id is missing"),
                    "a run-less task must be refused BY ITS MISSING RUN, not by the transport; \
                     got: {}",
                    undeliverable.reason()
                );
                assert!(
                    !undeliverable.deregisters_worker(),
                    "a malformed task is not evidence that the worker is gone"
                );
            }
        }

        // 🔴 THE PROPERTY, as far as this test can witness it: nothing is bound
        // when the run refuses. Note what this does NOT say — an earlier
        // version of this comment claimed nothing COULD be bound because an
        // `AttemptKey` is only constructible from a resolved identity, and that
        // was false. `AttemptKey::new` is public over four plain components.
        // See this test's docs and [`Resolved`] for what actually holds.
        let bound = owners.attempts_for_workflow(&WorkflowId::new(Uuid::from_u128(WORKFLOW)));
        assert!(
            bound.is_empty(),
            "the attempt-owner index must be untouched when the run refused; a binding here \
             means the bind was lifted above the run resolution, and an intervention could \
             resolve the wrong generation's worker. Found: {bound:?}"
        );
        Ok(())
    }

    /// A task carrying everything a delivery needs, run included.
    fn task_with_a_run() -> ProtoActivityTask {
        ProtoActivityTask {
            run_id: Some(aion_proto::ProtoRunId::from(RunId::new(Uuid::from_u128(
                0x52,
            )))),
            ..task_without_a_run()
        }
    }

    /// 🔴 NAMED PROPERTY: the abandonment guard is armed BEFORE the push, so a
    /// delivery cancelled while its push is still in flight returns the slot
    /// the caller tracked for it.
    ///
    /// # What breaks without it
    ///
    /// A guard constructed after the push leaves one uncovered window — the
    /// push's own `await`, the FIRST point this future can be cancelled at. A
    /// dispatcher that withdraws there (a shutdown, a caller that stops
    /// wanting the work) runs no arm below the push, so nothing untracks the
    /// entry made before it: the worker's advertised capacity is down one slot
    /// permanently, on a registry that in a multi-server deployment outlives
    /// the dispatcher that made the entry.
    ///
    /// # How the window is held open, deterministically
    ///
    /// The push runs on `spawn_blocking`. The runtime here has a blocking pool
    /// of exactly ONE thread, and that thread is occupied by a closure parked
    /// on a gate the test holds shut — so the push's blocking task is provably
    /// still queued, not merely likely to be, when the delivery future is
    /// polled once and dropped. The worker is a liminal handle over a
    /// supervisor with no connection behind it: had the push ever run it
    /// would have FAILED, and the failure arm disarms the guard and leaves the
    /// slot to the caller — which is exactly why the push must not be allowed
    /// to run for this test to say anything. The gate is opened afterwards so
    /// the runtime shuts down cleanly.
    ///
    /// Measured by mutation: moving the guard's construction below the push
    /// (a full revert of the fix) leaves the slot held and this test red.
    #[test]
    fn a_delivery_abandoned_during_its_push_returns_the_tracked_slot()
    -> Result<(), Box<dyn std::error::Error>> {
        use std::time::{Duration, Instant};

        use crate::worker::envelope::CompletionToken;
        use crate::worker::heartbeat::{HeartbeatTracker, InFlightActivity};
        use crate::worker::liminal_transport::LiminalWorkerDelivery;

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .max_blocking_threads(1)
            .build()?;
        runtime.block_on(async {
            // Occupy the only blocking thread until the gate opens.
            let (open_gate, gate) = std::sync::mpsc::channel::<()>();
            let occupant = tokio::task::spawn_blocking(move || gate.recv());

            let registry = ConnectedWorkerRegistry::default();
            let tracker = HeartbeatTracker::new(Duration::from_secs(30));
            let supervisor = liminal_server::server::connection::ConnectionSupervisor::new()?;
            let types = [String::from("agent")];
            let registration = registry.register_delivery(
                [String::from("default")],
                String::from("default"),
                None,
                types.iter(),
                WorkerDelivery::Liminal(LiminalWorkerDelivery::new(supervisor, 7)),
                RegistrationOptions::identified("agent-1", 1)
                    .with_intervention_capabilities(InterventionCapabilities::none()),
            )?;
            let worker_id = registration
                .worker_id()
                .ok_or("a registration must assign a worker id")?;
            let worker = registry
                .worker_by_id(worker_id)?
                .ok_or("the worker just registered must be readable")?;

            // The caller's half of the contract: tracked BEFORE the delivery is
            // handed the task, holding the worker's one slot.
            let workflow_id = WorkflowId::new(Uuid::from_u128(WORKFLOW));
            let activity_id = ActivityId::from_sequence_position(3);
            tracker.track_task(
                worker_id,
                InFlightActivity {
                    workflow_id: workflow_id.clone(),
                    activity_id: activity_id.clone(),
                    attempt: 1,
                    completion_token: CompletionToken::for_test(),
                },
                Instant::now(),
                &registry,
                None,
            )?;
            assert_eq!(
                registry.in_flight_for_worker(worker_id)?,
                1,
                "precondition: the tracked dispatch holds the worker's slot"
            );

            let owners = AttemptOwnerIndex::new();
            let delivery = liminal_delivery(&owners)
                .with_completion_tracking(tracker.clone(), registry.clone());
            let intent: SharedDeliveryIntent = Arc::new(AlwaysWanted);
            let accepted = NeverAccepted::default();
            let task = task_with_a_run();
            {
                // Owned on the heap so the `drop` below is the real one: a
                // stack-pinned future is only released at the end of its scope.
                let mut future = Box::pin(delivery.deliver(&worker, &task, &intent, &accepted));
                let first_poll = futures::poll!(future.as_mut());
                assert!(
                    first_poll.is_pending(),
                    "the push is queued behind the occupied blocking thread, so the first poll \
                     must suspend AT the push; a ready outcome means the window this test \
                     holds open was not held"
                );
                // The dispatcher withdraws while the push is in flight.
                drop(future);
            }
            assert!(
                !accepted.reached.load(std::sync::atomic::Ordering::SeqCst),
                "a delivery abandoned during its push never reached the accept point"
            );

            // 🔴 THE PROPERTY: the slot is back, and the entry is gone.
            assert_eq!(
                registry.in_flight_for_worker(worker_id)?,
                0,
                "a delivery abandoned during its push must return the tracked slot; a slot \
                 still held means the guard was armed AFTER the push and the cancellation \
                 window is open again"
            );
            assert!(
                !tracker.complete_task(worker_id, &workflow_id, &activity_id, &registry)?,
                "the guard retired the entry itself; the caller's later untrack must find \
                 nothing left to retire"
            );

            // Let the queued push run to its (failing) end and the occupant exit,
            // so runtime shutdown does not wait on a parked blocking thread.
            open_gate.send(())?;
            occupant.await??;
            Ok::<(), Box<dyn std::error::Error>>(())
        })
    }

    /// The property's SECOND witness, and the one that outlives a refactor of
    /// the call site: identity resolution refuses a run-less task, so no
    /// `Resolved` is produced — and [`Resolved::bind_owner`], the only binding
    /// path the delivery takes, is a method on it.
    ///
    /// 🔴 That is a claim about the CALL SITE, not about the key type. An
    /// earlier version said an `AttemptKey` "cannot be built without" a
    /// `Resolved`; it can, and the mutation that proved it survived. What this
    /// witnesses is that resolution refuses first.
    ///
    /// Kept beside the behavioural test deliberately. The refusal and the
    /// structure are different claims, and a change that removes one should
    /// still meet the other.
    #[test]
    fn identity_resolution_refuses_a_task_with_no_run() -> Result<(), Box<dyn std::error::Error>> {
        let Err(refusal) = Resolved::from_task(&task_without_a_run()) else {
            return Err("a task with no run must not resolve".into());
        };
        assert!(
            refusal.contains("run id is missing"),
            "the refusal must name the run so an operator is sent to the right remedy: {refusal}"
        );
        Ok(())
    }
}