aion_server/worker/

declared_body.rs

//! Server-side execution of declared action bodies.
//!
//! An action whose deployed contract carries an [`ActionBodyContract`] is
//! executed BY THE SERVER, with no connected worker: the dispatch is
//! intercepted at the [`ActivityDispatcher`] seam before task-queue routing,
//! the declared command runs through the worker SDK's own executor
//! ([`aion_worker::shell::ShellAction`] — argv-element substitution, no
//! shell, process-group containment), and the result flows back through the
//! engine's normal completion path. The engine still schedules, records, and
//! replays the activity exactly as if a worker had served it.
//!
//! Actions with no declared body are delegated to the wrapped production
//! dispatcher unchanged, so remote workers keep working exactly as before.
//!
//! # The command is readable while it runs
//!
//! The executing activity carries a live transcript seam
//! ([`ActivityContext::with_transcript`](aion_worker::ActivityContext::with_transcript)),
//! so every line the command writes to stdout or stderr is published onto the
//! server's transcript sequencer AS IT ARRIVES — the same stream, envelope, and
//! cursor reads an agent step's transcript uses (see
//! [`super::declared_body_transcript`]). The activity's recorded result is
//! untouched by this: it still carries the command's complete output.
//!
//! # The command can be stopped, by its bound and by its run
//!
//! Two things end a server-executed command early, and both reach the same
//! cancellation the worker path already acts on — `SIGTERM` → grace →
//! `SIGKILL` across the whole process group, with the verdict withheld until
//! the group has been proven gone.
//!
//! The first is the attempt's own deadline. A dispatch carrying an authored
//! per-attempt timeout (#223) ends its command at that bound HERE, in the
//! server, where the process is — see [`run_bounded`]. The engine's own
//! deadline stops the run WAITING and cannot reach a process, which is the
//! right division of labour for a remote worker and no division at all for a
//! body the server itself started. A dispatch that authored no bound is
//! unbounded, exactly as before: the server adds no deadline of its own.
//!
//! The second is the run being cancelled. Every executing attempt registers in
//! [`super::declared_body_cancel::DeclaredCommandAttempts`] for exactly as long
//! as its command runs, which is how the cancel path reaches an activity no
//! worker holds and no heartbeat tracks. An attempt that cannot register is
//! refused rather than run: a command a cancelled run could not stop is the
//! defect the registration exists to prevent.

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

use aion::{ActivityDispatch, ActivityDispatcher};
use aion_package::{ActionBodyContract, ContentHash};
use aion_worker::shell::ShellAction;

use super::declared_body_ambiguity::{DeclaringVersion, ambiguous_body_refusal};
use super::declared_body_cancel::DeclaredCommandAttempts;
use super::declared_body_selection::select_declared_body;
use super::declared_body_transcript::publish_declared_transcript;
use super::workspace_root::{WORKSPACE_ROOT_PLACEHOLDER, WorkspaceRoot};
use crate::activity_publisher::ActivityEventPublisher;

/// What a declared-body lookup found for one `(task_queue, action)` address.
#[derive(Clone, Debug)]
pub enum DeclaredBodyLookup {
    /// No retained contract declares a body for this action — it is a
    /// requirement on an out-of-band worker and must be delegated.
    None,
    /// Exactly one distinct body is declared across every retained package
    /// version. Safe to execute.
    Declared(ActionBodyContract),
    /// Retained package versions declare DIFFERENT bodies for this action.
    /// Executing one of them would guess which deploy the running workflow
    /// meant, so the dispatch is refused by name instead.
    Ambiguous {
        /// Every retained version that declares a body for this action, in
        /// catalog order. Carried rather than counted because the refusal has
        /// to name the versions the operator must retire — a bare count leaves
        /// them holding a terminal error with no way to act on it.
        declaring: Vec<DeclaringVersion>,
    },
    /// The catalog could not be read. The reader reports why; the dispatch
    /// is delegated so a readable worker path can still serve it.
    Unreadable(String),
}

/// Which run a declared-body lookup is being made for.
///
/// A body is a property of the run's own package version, not of the queue, so
/// the lookup cannot answer correctly without knowing whose dispatch it is —
/// see [`super::declared_body_selection`].
#[derive(Clone, Copy, Debug)]
pub struct DispatchingRun<'a> {
    /// The workflow the activity belongs to.
    pub workflow_id: &'a aion_core::WorkflowId,
    /// The concrete run within that workflow.
    pub run_id: &'a aion_core::RunId,
}

/// A reader over the deployed contracts' declared action bodies.
pub trait DeclaredBodies: Send + Sync {
    /// Look up the declared body for `action` on `task_queue`, as the run
    /// issuing the dispatch sees it.
    fn body_for(
        &self,
        task_queue: &str,
        action: &str,
        run: DispatchingRun<'_>,
    ) -> DeclaredBodyLookup;
}

/// Shared, install-once handle the dispatcher holds from construction and the
/// boot path fills in once the engine exists.
///
/// Mirrors [`super::QueueDeclarationSource`]: the dispatcher is built before
/// the engine, so the seam it consults is handed over afterwards through a
/// clone of this handle rather than by rebuilding the dispatcher.
#[derive(Clone, Default)]
pub struct DeclaredBodySource {
    inner: Arc<OnceLock<Arc<dyn DeclaredBodies>>>,
}

impl std::fmt::Debug for DeclaredBodySource {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("DeclaredBodySource")
            .field("installed", &self.inner.get().is_some())
            .finish()
    }
}

impl DeclaredBodySource {
    /// Install the reader. A second install is ignored and logged: the source
    /// is process-wide and must not silently change identity.
    pub fn install(&self, source: Arc<dyn DeclaredBodies>) {
        if self.inner.set(source).is_err() {
            tracing::warn!("declared body source already installed; ignoring duplicate set");
        }
    }

    /// Look up the declared body, or [`DeclaredBodyLookup::None`] when no
    /// reader is installed yet.
    ///
    /// An uninstalled consult is stated at ERROR before delegating to the
    /// worker path, never silently: a dispatch can only reach this seam from
    /// a live run, and a live run's deploy is durable — so "nothing installed
    /// yet" is a boot-ordering defect, not an empty catalog. This exact
    /// silence was #266 Defect A: startup recovery replay re-dispatched
    /// adopted in-flight declared-body activities before
    /// `install_engine_backed_seams` filled this source, and every one fell
    /// through here to a queue with no pollers and parked forever. The fix
    /// (deferred startup recovery) removes the caller; this arm stays loud so
    /// any future pre-install dispatch path names itself in the log instead
    /// of stranding runs silently.
    #[must_use]
    pub fn body_for(
        &self,
        task_queue: &str,
        action: &str,
        run: DispatchingRun<'_>,
    ) -> DeclaredBodyLookup {
        self.inner.get().map_or_else(
            || {
                tracing::error!(
                    operation = "declared_command_dispatch",
                    task_queue,
                    action,
                    workflow_id = %run.workflow_id,
                    run_id = %run.run_id,
                    "declared body source consulted before it was installed; the dispatch \
                     falls through to the worker path and will park if the queue's only \
                     service is its declared bodies (#266 boot-ordering defect)"
                );
                DeclaredBodyLookup::None
            },
            |source| source.body_for(task_queue, action, run),
        )
    }
}

/// Reads declared bodies out of the engine's live workflow catalog.
pub struct EngineDeclaredBodies {
    engine: Arc<aion::Engine>,
}

impl EngineDeclaredBodies {
    /// Build a reader over `engine`'s catalog.
    #[must_use]
    pub const fn new(engine: Arc<aion::Engine>) -> Self {
        Self { engine }
    }
}

impl std::fmt::Debug for EngineDeclaredBodies {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str("EngineDeclaredBodies")
    }
}

impl EngineDeclaredBodies {
    /// The package version `run` is pinned to, or `None` when the registry
    /// cannot name it.
    ///
    /// Two ways to reach `None`, and both are reported rather than swallowed:
    /// the run has no handle (it left the registry), or the registry could not
    /// be read at all. Neither is a reason to guess a body — the caller falls
    /// back to the queue-wide reading, which refuses on disagreement.
    fn version_of(&self, run: DispatchingRun<'_>) -> Option<ContentHash> {
        match self.engine.registry().get(run.workflow_id, run.run_id) {
            Ok(Some(handle)) => Some(handle.loaded_version().clone()),
            Ok(None) => {
                tracing::warn!(
                    operation = "declared_command_dispatch",
                    workflow_id = %run.workflow_id,
                    run_id = %run.run_id,
                    "no registry handle for the dispatching run; resolving its body \
                     from the whole queue instead of from its own package version"
                );
                None
            }
            Err(error) => {
                tracing::error!(
                    operation = "declared_command_dispatch",
                    workflow_id = %run.workflow_id,
                    run_id = %run.run_id,
                    %error,
                    "registry unreadable while resolving the dispatching run's version; \
                     resolving its body from the whole queue instead"
                );
                None
            }
        }
    }
}

impl DeclaredBodies for EngineDeclaredBodies {
    fn body_for(
        &self,
        task_queue: &str,
        action: &str,
        run: DispatchingRun<'_>,
    ) -> DeclaredBodyLookup {
        let contracts = match self.engine.worker_contracts_for_queue(task_queue) {
            Ok(contracts) => contracts,
            Err(error) => return DeclaredBodyLookup::Unreadable(error.to_string()),
        };
        // The RAW retained set is the right input here, unlike worker admission
        // (see `Engine::worker_contracts_for_queue`): a run pinned to a version
        // nothing else can reach still has to execute that version's body. What
        // narrows the answer is the run's own identity, not reachability.
        select_declared_body(&contracts, action, self.version_of(run).as_ref())
    }
}

/// The dispatcher decorator that executes declared bodies at the server.
///
/// Wraps the production dispatcher. Consults the declared-body source before
/// every dispatch; delegates untouched whenever the action carries no body.
pub struct DeclaredCommandDispatcher {
    inner: Arc<dyn ActivityDispatcher>,
    bodies: DeclaredBodySource,
    attempts: DeclaredCommandAttempts,
    tokio: tokio::runtime::Handle,
    workspace_root: WorkspaceRoot,
    transcript: ActivityEventPublisher,
}

impl DeclaredCommandDispatcher {
    /// Wrap `inner`, consulting `bodies` before every dispatch, registering
    /// every attempt it executes in `attempts` so the run's cancel can reach
    /// it, expanding `{workspace_root}` in declared commands with the
    /// server-resolved `workspace_root`, and streaming each executed command's
    /// output onto `transcript` — the deployment's one transcript sequencer,
    /// shared with every agent step.
    ///
    /// `attempts` is required rather than optional because a dispatcher without
    /// one would execute commands nothing could stop, which is precisely the
    /// state this argument exists to end.
    #[must_use]
    pub fn new(
        inner: Arc<dyn ActivityDispatcher>,
        bodies: DeclaredBodySource,
        attempts: DeclaredCommandAttempts,
        tokio: tokio::runtime::Handle,
        workspace_root: WorkspaceRoot,
        transcript: ActivityEventPublisher,
    ) -> Self {
        Self {
            inner,
            bodies,
            attempts,
            tokio,
            workspace_root,
            transcript,
        }
    }

    /// Parse the declared command into the executor's action, with the
    /// server-resolved `{workspace_root}` already spliced in.
    ///
    /// Ratification condition (#139): a body that USES the placeholder is
    /// refused terminally, by name, when the root cannot resolve to an absolute
    /// directory that exists — no fallback to cwd, temp, or anything else. A
    /// body without the placeholder never reaches the resolution at all
    /// (`expand` returns `Ok(None)` untouched).
    fn declared_action(
        &self,
        request: &ActivityDispatch,
        command: &str,
    ) -> Result<ShellAction, String> {
        let expanded = self.workspace_root.expand(command).map_err(|error| {
            format!(
                "terminal:declared body for action `{name}` uses the {placeholder} \
                 placeholder and cannot dispatch: {error}",
                name = request.name,
                placeholder = WORKSPACE_ROOT_PLACEHOLDER,
            )
        })?;
        if let Some(expansion) = &expanded {
            tracing::info!(
                operation = "declared_command_dispatch",
                workflow_id = %request.workflow_id,
                activity_id = %request.activity_id,
                activity_name = %request.name,
                task_queue = %request.task_queue,
                attempt = request.attempt,
                workspace_root = %expansion.workspace_root,
                "expanded the workspace-root placeholder in the declared command"
            );
        }
        let command = expanded
            .as_ref()
            .map_or(command, |expansion| expansion.command.as_str());
        ShellAction::new(command).map_err(|error| {
            // The AWL checker refuses these at compile time, so reaching this
            // arm means a defective contract got deployed — name the defect
            // rather than hiding it behind a generic dispatch failure.
            format!("terminal:declared command failed to parse at dispatch: {error}")
        })
    }

    /// Put the attempt on this server's cancel path BEFORE its command starts.
    ///
    /// The returned guard keeps it there for exactly as long as the command
    /// runs, so an attempt that ended — completed, failed, or unwound — can
    /// never be signalled afterwards. A registration that cannot be made is a
    /// command nothing could stop, so the dispatch is refused rather than run:
    /// an uncancellable command on the operator's machine is the whole defect
    /// this registration exists to prevent, and starting one to avoid an error
    /// message would be choosing it.
    fn join_cancel_path(
        &self,
        request: &ActivityDispatch,
        cancellation: &aion_worker::ActivityCancellationHandle,
    ) -> Result<super::DeclaredAttemptRegistration, String> {
        self.attempts
            .register(
                super::AttemptKey::new(
                    request.workflow_id.clone(),
                    request.run_id.clone(),
                    request.activity_id.clone(),
                    request.attempt,
                ),
                cancellation.clone(),
            )
            .map_err(|error| {
                format!(
                    "terminal:declared body for action `{name}` cannot dispatch: the attempt \
                     could not join this server's cancel path, and a command a cancelled run \
                     could not stop must not be started: {error}",
                    name = request.name,
                )
            })
    }

    /// Execute one declared command attempt and encode the outcome onto the
    /// FFI string contract (`retryable:`/`terminal:` on the error side).
    fn run_declared_command(
        &self,
        request: &ActivityDispatch,
        command: &str,
    ) -> Result<String, String> {
        let arguments = decode_arguments(&request.input)?;
        let action = self.declared_action(request, command)?;
        // The live transcript seam for this attempt. The context owns the
        // sending end, so dropping it after the run closes the stream and ends
        // the pump — which is then awaited, so no observed line is abandoned
        // unpublished when the command finishes.
        let (events, drain) = tokio::sync::mpsc::unbounded_channel();
        let (context, cancellation) = aion_worker::ActivityContext::with_transcript(
            request.workflow_id.clone(),
            request.run_id.clone(),
            request.activity_id.clone(),
            request.attempt,
            events,
        );
        let registration = self.join_cancel_path(request, &cancellation)?;

        tracing::info!(
            operation = "declared_command_dispatch",
            workflow_id = %request.workflow_id,
            activity_id = %request.activity_id,
            activity_name = %request.name,
            task_queue = %request.task_queue,
            attempt = request.attempt,
            "executing declared action body at the server"
        );
        // All three 2026-08-16 anonymous deaths correlated with workflow
        // execution and the third died on exactly this path; the breadcrumb
        // makes the in-flight site a death-note fact, not a log inference.
        crate::death_note::breadcrumb(&format!(
            "declared-action start action={} workflow_id={} run_id={} activity_id={} attempt={}",
            request.name, request.workflow_id, request.run_id, request.activity_id, request.attempt,
        ));

        // #223: the bound the DISPATCH authored, or `None` when it authored
        // none. Read through the engine's own decoder so the server cannot
        // answer "what did this document authorise" differently from the retry
        // loop, and so an unbounded body stays unbounded — the server invents
        // no deadline of its own.
        let bound = aion::activity_timeout_from_config(&request.config);
        let transcript = self.transcript.clone();
        let ended = self.tokio.block_on(async move {
            let pump = tokio::spawn(publish_declared_transcript(transcript, drain));
            let ended = run_bounded(&action, &arguments, &context, &cancellation, bound).await;
            // Closing the seam is what ends the pump; the context holds it.
            drop(context);
            if let Err(error) = pump.await {
                tracing::warn!(
                    %error,
                    operation = "declared_command_dispatch",
                    "declared command transcript: the publishing task ended abnormally; some \
                     output lines may not have been retained"
                );
            }
            ended
        });
        // The command is over and its group is gone, so the attempt leaves the
        // cancel path. Dropped explicitly, here and not earlier: while this
        // lives, a cancel arriving mid-run still reaches the process.
        drop(registration);

        encode_end(request, ended)
    }
}

/// Encode how the attempt ended onto the FFI string contract.
///
/// Three vocabularies, one per honest outcome: the encoded result, the
/// classified failure the executor produced (`retryable:`/`terminal:`), and the
/// engine's own `timeout:` reason for an attempt that outlived its authored
/// bound.
fn encode_end(request: &ActivityDispatch, ended: AttemptEnd) -> Result<String, String> {
    let outcome = match ended {
        AttemptEnd::Ran(outcome) => outcome,
        AttemptEnd::Expired { bound, ran_anyway } => {
            if let Some(exit_code) = ran_anyway {
                // The command reached its own end inside the stopping window.
                // Its result is discarded — the attempt is already recorded as
                // having outlived its bound, and answering with a late success
                // would contradict a terminal the run has already been told
                // about — but the fact is said, not swallowed.
                tracing::warn!(
                    operation = "declared_command_dispatch",
                    workflow_id = %request.workflow_id,
                    activity_id = %request.activity_id,
                    activity_name = %request.name,
                    attempt = request.attempt,
                    exit_code,
                    bound_ms = bound.as_millis(),
                    "the declared command finished while it was being stopped on its \
                     authored bound; its result is discarded in favour of the timeout"
                );
            }
            return Err(aion::activity_timeout_reason(bound));
        }
    };

    match outcome {
        Ok(result) => serde_json::to_string(&result)
            .map_err(|error| format!("terminal:declared command result failed to encode: {error}")),
        Err(failure) => {
            let prefix = match failure.classification() {
                aion_worker::Classification::Retryable => "retryable",
                aion_worker::Classification::PolicyRefused => "policy_refused",
                aion_worker::Classification::Terminal => "terminal",
            };
            Err(format!("{prefix}:{}", failure.message()))
        }
    }
}

/// How one declared-command attempt ended.
#[derive(Debug)]
enum AttemptEnd {
    /// The command ran to its own end — completed, failed, or was stopped by
    /// something other than the authored bound.
    Ran(Result<aion_worker::shell::ShellOutcome, aion_worker::ActivityFailure>),
    /// The attempt outlived the per-attempt bound its dispatch authored, and
    /// its process group has been stopped and PROVEN gone.
    Expired {
        /// The authored bound that fired, carried so the refusal can name it.
        bound: std::time::Duration,
        /// The exit code of a command that reached its own end inside the
        /// stopping window, when that happened. `None` — the ordinary case —
        /// means the command was still running when the bound was enforced.
        ran_anyway: Option<i32>,
    },
}

/// Run the declared command, ending it at the bound its dispatch authored.
///
/// # Why the server enforces a bound the engine already applies
///
/// The engine wraps every attempt in `tokio::time::timeout` at the same
/// authored bound (`nif_activity_retry_dispatch::deliver_one_attempt`), and is
/// explicit about what that achieves: "the dispatch future is DROPPED, which
/// stops this run waiting and nothing more... the worker-side call runs on to
/// its own end and its result is discarded". For a REMOTE worker that is
/// someone else's machine and the right division of labour. For a declared body
/// it is a process tree in the server's own process group hierarchy, on the
/// operator's machine, with nothing left that could ever stop it — the run has
/// already moved on.
///
/// So the bound is enforced HERE as well, where the process is. On expiry the
/// activity's cancellation is signalled and the SAME run future is awaited to
/// its end: [`aion_worker::run_cancellable_command`] does not return until
/// `SIGTERM` → [`aion_worker::PROCESS_GROUP_TERMINATION_GRACE`] → `SIGKILL` has
/// been delivered to the whole group and the group has been PROVEN gone. This
/// therefore returns only once the command is genuinely stopped, and the
/// termination ladder and its grace are the worker path's, not a second copy.
///
/// A dispatch that authored no bound is awaited exactly as before.
async fn run_bounded(
    action: &ShellAction,
    arguments: &BTreeMap<String, serde_json::Value>,
    context: &aion_worker::ActivityContext,
    cancellation: &aion_worker::ActivityCancellationHandle,
    bound: Option<std::time::Duration>,
) -> AttemptEnd {
    let run = action.run(arguments, context);
    let Some(bound) = bound else {
        return AttemptEnd::Ran(run.await);
    };
    tokio::pin!(run);
    match tokio::time::timeout(bound, &mut run).await {
        Ok(outcome) => AttemptEnd::Ran(outcome),
        Err(_elapsed) => {
            cancellation.cancel();
            AttemptEnd::Expired {
                bound,
                ran_anyway: run.await.ok().map(|outcome| outcome.exit_code),
            }
        }
    }
}

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

impl ActivityDispatcher for DeclaredCommandDispatcher {
    fn dispatch(&self, request: ActivityDispatch) -> Result<String, String> {
        let run = DispatchingRun {
            workflow_id: &request.workflow_id,
            run_id: &request.run_id,
        };
        match self
            .bodies
            .body_for(&request.task_queue, &request.name, run)
        {
            DeclaredBodyLookup::None => self.inner.dispatch(request),
            DeclaredBodyLookup::Unreadable(reason) => {
                // Delegated, not refused: a catalog read failure must not
                // strand a queue that live workers could still serve. Loud so
                // an operator sees a bodied action falling through.
                tracing::error!(
                    operation = "declared_command_dispatch",
                    workflow_id = %request.workflow_id,
                    activity_name = %request.name,
                    task_queue = %request.task_queue,
                    %reason,
                    "declared-body catalog read failed; delegating to the worker path"
                );
                self.inner.dispatch(request)
            }
            DeclaredBodyLookup::Ambiguous { declaring } => Err(ambiguous_body_refusal(
                &request.name,
                &request.task_queue,
                &declaring,
            )),
            DeclaredBodyLookup::Declared(ActionBodyContract::Run { command }) => {
                self.run_declared_command(&request, &command)
            }
        }
    }
}

/// Decode the dispatch's JSON input into the declared action's arguments.
///
/// A declared action's parameters are named in its `.awl` declaration, so the
/// input must be a JSON object; anything else cannot bind to `$name`
/// references and is refused by shape. Retrying cannot change the input, so
/// the refusal is terminal.
fn decode_arguments(input: &str) -> Result<BTreeMap<String, serde_json::Value>, String> {
    let value: serde_json::Value = serde_json::from_str(input)
        .map_err(|error| format!("terminal:declared command input is not valid JSON: {error}"))?;
    match value {
        serde_json::Value::Object(members) => Ok(members.into_iter().collect()),
        other => Err(format!(
            "terminal:declared command input must be a JSON object binding the action's \
             parameters by name; got {}",
            json_kind(&other)
        )),
    }
}

/// A JSON value's kind, named for a refusal message.
const fn json_kind(value: &serde_json::Value) -> &'static str {
    match value {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "a boolean",
        serde_json::Value::Number(_) => "a number",
        serde_json::Value::String(_) => "a string",
        serde_json::Value::Array(_) => "an array",
        serde_json::Value::Object(_) => "an object",
    }
}

/// Containment of a server-executed body: the authored per-attempt bound, and
/// the run's cancellation. Its own file because its subjects are live process
/// trees rather than dispatcher shapes; it builds them out of the fixtures
/// [`tests`] shares with it.
#[cfg(test)]
#[path = "declared_body_containment_tests.rs"]
mod declared_body_containment_tests;

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::sync::{Arc, Mutex};

    use aion::{ActivityDispatch, ActivityDispatcher};
    use aion_core::{ActivityId, RunId, WorkflowId};
    use aion_package::ActionBodyContract;

    use aion_core::ActivityEventKind;
    use aion_store::ActivityStreamKey;

    use super::super::workspace_root::{WorkspaceRoot, WorkspaceRootError};
    use super::{
        ActivityEventPublisher, DeclaredBodies, DeclaredBodyLookup, DeclaredBodySource,
        DeclaredCommandAttempts, DeclaredCommandDispatcher, DeclaringVersion, DispatchingRun,
        decode_arguments,
    };

    /// What a test returns. Every fallible step is carried rather than
    /// unwrapped, because the workspace denies panicking accessors in test
    /// code as firmly as in library code.
    pub(super) type TestResult = Result<(), Box<dyn std::error::Error>>;

    /// Inner dispatcher that records whether it was reached.
    struct RecordingInner {
        reached: Arc<Mutex<Vec<String>>>,
        reply: Result<String, String>,
    }

    impl ActivityDispatcher for RecordingInner {
        fn dispatch(&self, request: ActivityDispatch) -> Result<String, String> {
            match self.reached.lock() {
                Ok(mut names) => names.push(request.name),
                Err(poisoned) => poisoned.into_inner().push(request.name),
            }
            self.reply.clone()
        }
    }

    struct FixedBodies {
        lookup: DeclaredBodyLookup,
    }

    impl DeclaredBodies for FixedBodies {
        fn body_for(
            &self,
            _task_queue: &str,
            _action: &str,
            _run: DispatchingRun<'_>,
        ) -> DeclaredBodyLookup {
            self.lookup.clone()
        }
    }

    /// A reader that records whose dispatch it was asked about.
    ///
    /// The selection rule is unit-tested on its own inputs, which proves the
    /// rule and nothing about the plumbing. This double closes that gap: it
    /// captures the [`DispatchingRun`] the dispatcher hands over, so the
    /// identity can be compared against the request it came from.
    struct RecordingBodies {
        seen: Arc<Mutex<Vec<(WorkflowId, RunId)>>>,
    }

    impl DeclaredBodies for RecordingBodies {
        fn body_for(
            &self,
            _task_queue: &str,
            _action: &str,
            run: DispatchingRun<'_>,
        ) -> DeclaredBodyLookup {
            let observed = (run.workflow_id.clone(), run.run_id.clone());
            match self.seen.lock() {
                Ok(mut seen) => seen.push(observed),
                Err(poisoned) => poisoned.into_inner().push(observed),
            }
            DeclaredBodyLookup::None
        }
    }

    pub(super) fn request(name: &str, input: &str) -> ActivityDispatch {
        ActivityDispatch {
            namespace: "default".to_owned(),
            task_queue: "shell".to_owned(),
            node: None,
            workflow_id: WorkflowId::new_v4(),
            run_id: RunId::new_v4(),
            activity_id: ActivityId::from_sequence_position(1),
            name: name.to_owned(),
            input: input.to_owned(),
            config: "{}".to_owned(),
            attempt: 1,
            labels: BTreeMap::new(),
            advisory: false,
        }
    }

    fn dispatcher(
        lookup: DeclaredBodyLookup,
        reply: Result<String, String>,
    ) -> (DeclaredCommandDispatcher, Arc<Mutex<Vec<String>>>) {
        // These tests exercise bodies without the placeholder, so the root's
        // value is never read; it is an explicit existing directory rather
        // than a default so nothing here depends on resolution.
        let (decorated, reached, _transcript) = dispatcher_with_root(
            lookup,
            reply,
            WorkspaceRoot::from_resolution(Ok(std::env::temp_dir())),
        );
        (decorated, reached)
    }

    /// The live-tail buffer these tests give their transcript sequencer. A
    /// `const` match rather than an unwrap: the workspace denies panicking
    /// accessors in test code as firmly as in library code.
    const TRANSCRIPT_CAPACITY: std::num::NonZeroUsize = match std::num::NonZeroUsize::new(64) {
        Some(capacity) => capacity,
        None => std::num::NonZeroUsize::MIN,
    };

    /// A dispatcher whose executing attempts nothing external will signal.
    ///
    /// Correct for every test whose subject runs to its own end. A test that
    /// CANCELS its subject needs the registry the cancel signals through, and
    /// uses [`dispatcher_with_attempts`] to hold the same instance.
    pub(super) fn dispatcher_with_root(
        lookup: DeclaredBodyLookup,
        reply: Result<String, String>,
        workspace_root: WorkspaceRoot,
    ) -> (
        DeclaredCommandDispatcher,
        Arc<Mutex<Vec<String>>>,
        ActivityEventPublisher,
    ) {
        dispatcher_with_attempts(
            lookup,
            reply,
            workspace_root,
            DeclaredCommandAttempts::new(),
        )
    }

    pub(super) fn dispatcher_with_attempts(
        lookup: DeclaredBodyLookup,
        reply: Result<String, String>,
        workspace_root: WorkspaceRoot,
        attempts: DeclaredCommandAttempts,
    ) -> (
        DeclaredCommandDispatcher,
        Arc<Mutex<Vec<String>>>,
        ActivityEventPublisher,
    ) {
        let reached = Arc::new(Mutex::new(Vec::new()));
        let inner = RecordingInner {
            reached: Arc::clone(&reached),
            reply,
        };
        let bodies = DeclaredBodySource::default();
        bodies.install(Arc::new(FixedBodies { lookup }));
        let store: Arc<dyn aion_store::ObservabilityStore> =
            Arc::new(aion_store::InMemoryObservabilityStore::default());
        let transcript = ActivityEventPublisher::new(
            store,
            TRANSCRIPT_CAPACITY,
            crate::activity_publisher::TranscriptBatchPolicy::UNBATCHED,
        );
        let decorated = DeclaredCommandDispatcher::new(
            Arc::new(inner),
            bodies,
            attempts,
            tokio::runtime::Handle::current(),
            workspace_root,
            transcript.clone(),
        );
        (decorated, reached, transcript)
    }

    pub(super) fn reached_names(reached: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
        match reached.lock() {
            Ok(names) => names.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_bodiless_action_is_delegated_untouched() -> TestResult {
        let (decorated, reached) =
            dispatcher(DeclaredBodyLookup::None, Ok("\"worker-served\"".to_owned()));
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("plain", "{}")));
        let result = handle.await?;
        assert_eq!(result, Ok("\"worker-served\"".to_owned()));
        assert_eq!(reached_names(&reached), vec!["plain".to_owned()]);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_declared_body_executes_without_touching_the_worker_path() -> TestResult {
        let (decorated, reached) = dispatcher(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "echo $greeting".to_owned(),
            }),
            Err("terminal:the worker path must never be reached".to_owned()),
        );
        let handle = tokio::task::spawn_blocking(move || {
            decorated.dispatch(request(
                "greet",
                "{\"greeting\":\"hello from the contract\"}",
            ))
        });
        let result = handle.await?;
        let encoded = result.map_err(|error| format!("declared command failed: {error}"))?;
        let outcome: serde_json::Value = serde_json::from_str(&encoded)?;
        assert_eq!(outcome["stdout"], "hello from the contract");
        assert_eq!(outcome["exit_code"], 0);
        assert!(
            reached_names(&reached).is_empty(),
            "the worker path must not be consulted for a bodied action"
        );
        Ok(())
    }

    /// THE MID-STEP ANSWER: a server-run declared body's output reaches the
    /// deployment's transcript sequencer as one event per line, on both streams,
    /// keyed to the dispatch's own `(workflow, activity, attempt)` — the same
    /// durable stream an agent step's transcript is read from, so every reader
    /// that already serves transcripts serves this without change.
    ///
    /// The completion contract is asserted on the same run: the recorded result
    /// still carries the command's whole stdout.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_declared_body_publishes_its_output_onto_the_transcript() -> TestResult {
        let (decorated, reached, transcript) = dispatcher_with_root(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "sh -c 'echo one; echo two; echo warned >&2'".to_owned(),
            }),
            Err("terminal:the worker path must never be reached".to_owned()),
            WorkspaceRoot::from_resolution(Ok(std::env::temp_dir())),
        );
        let dispatch = request("noisy", "{}");
        let key = ActivityStreamKey::new(
            dispatch.workflow_id.clone(),
            dispatch.run_id.clone(),
            dispatch.activity_id.clone(),
            dispatch.attempt,
        );

        let handle = tokio::task::spawn_blocking(move || decorated.dispatch(dispatch));
        let encoded = handle
            .await?
            .map_err(|error| format!("declared command failed: {error}"))?;

        // The replay-authoritative result is untouched by the streaming.
        let outcome: serde_json::Value = serde_json::from_str(&encoded)?;
        assert_eq!(outcome["stdout"], "one\ntwo");
        assert_eq!(outcome["stderr"], "warned");
        assert!(reached_names(&reached).is_empty());

        // ...and the same output is on the durable transcript, line by line.
        let retained = transcript.replay_from(&key, 0).await?;
        let lines = retained
            .iter()
            .map(|record| match &record.event.kind {
                ActivityEventKind::Message { text, .. } => {
                    (record.event.agent_role.clone(), text.clone())
                }
                other => (record.event.agent_role.clone(), format!("{other:?}")),
            })
            .collect::<Vec<_>>();
        assert!(
            lines.contains(&("command stdout".to_owned(), "one".to_owned()))
                && lines.contains(&("command stdout".to_owned(), "two".to_owned())),
            "each stdout line must be its own transcript event: {lines:?}"
        );
        assert!(
            lines.contains(&("command stderr".to_owned(), "warned".to_owned())),
            "stderr must be on the transcript, labelled by its stream: {lines:?}"
        );
        // Sequencing is the publisher's: the durable order is gap-free from 0.
        let sequences = retained
            .iter()
            .map(|record| record.store_seq)
            .collect::<Vec<_>>();
        assert_eq!(
            sequences,
            (0..u64::try_from(retained.len())?).collect::<Vec<_>>(),
            "the sequencer assigns a gap-free durable order"
        );
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_failing_declared_command_reports_retryable_with_its_stderr() -> TestResult {
        let (decorated, _reached) = dispatcher(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "sh -c 'echo boom >&2; exit 7'".to_owned(),
            }),
            Ok("unused".to_owned()),
        );
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("fails", "{}")));
        let Err(error) = handle.await? else {
            return Err("a non-zero exit must fail the dispatch".into());
        };
        assert!(
            error.starts_with("retryable:"),
            "a non-zero exit is retryable by default: {error}"
        );
        assert!(
            error.contains("boom"),
            "stderr must ride the failure: {error}"
        );
        Ok(())
    }

    /// The hash the refusal prints must be one the deploy API will accept, or
    /// the remedy is a command that cannot run — the exact failure the old
    /// "redeploy so one body remains" wording had.
    ///
    /// The oracle is the deploy API's own parser, not a length or a shape:
    /// `EngineDeclaredBodies` renders the version with `ContentHash::to_string`,
    /// so this takes a real hash through that rendering, pulls the token back
    /// out of the printed command, and parses it the way
    /// `decode_version_target` does.
    #[test]
    fn the_printed_hash_parses_back_as_a_content_hash() -> TestResult {
        let version = aion_package::ContentHash::from_bytes([0x5a; 32]);
        let routed = aion_package::ContentHash::from_bytes([0xa5; 32]);
        let refusal = super::ambiguous_body_refusal(
            "find_repositories",
            "local",
            &[
                DeclaringVersion {
                    content_hash: version.to_string(),
                    workflow_types: vec!["sweeper".to_owned()],
                    route_active: false,
                    body: 0,
                },
                DeclaringVersion {
                    content_hash: routed.to_string(),
                    workflow_types: vec!["sweeper".to_owned()],
                    route_active: true,
                    body: 1,
                },
            ],
        );
        let Some(command) = refusal.split("`aion unload sweeper ").nth(1) else {
            return Err(format!("no unload command in the refusal: {refusal}").into());
        };
        let Some(printed) = command.split('`').next() else {
            return Err(format!("the unload command is unterminated: {refusal}").into());
        };
        let parsed: aion_package::ContentHash = printed.parse()?;
        assert_eq!(
            parsed, version,
            "the printed hash must round-trip to the version it names"
        );
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn ambiguous_bodies_refuse_terminally_by_name() -> TestResult {
        let superseded = "1111111111111111111111111111111111111111111111111111111111111111";
        let routed = "2222222222222222222222222222222222222222222222222222222222222222";
        let (decorated, reached) = dispatcher(
            DeclaredBodyLookup::Ambiguous {
                declaring: vec![
                    DeclaringVersion {
                        content_hash: superseded.to_owned(),
                        workflow_types: vec!["sweeper".to_owned()],
                        route_active: false,
                        body: 0,
                    },
                    DeclaringVersion {
                        content_hash: routed.to_owned(),
                        workflow_types: vec!["sweeper".to_owned()],
                        route_active: true,
                        body: 1,
                    },
                ],
            },
            Ok(String::new()),
        );
        let handle = tokio::task::spawn_blocking(move || decorated.dispatch(request("torn", "{}")));
        let Err(error) = handle.await? else {
            return Err("ambiguous bodies must refuse".into());
        };
        assert!(error.starts_with("terminal:"), "{error}");
        assert!(error.contains("torn"), "{error}");
        // The refusal must reach the dispatcher carrying an act-on-able remedy,
        // not just a count: the operator reads this string and nothing else.
        assert!(
            error.contains(&format!("`aion unload sweeper {superseded}`")),
            "the dispatch refusal must name the version to retire: {error}"
        );
        assert!(reached_names(&reached).is_empty());
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_placeholder_bearing_body_executes_with_the_expanded_root() -> TestResult {
        let scratch = tempfile::tempdir()?;
        let root = scratch.path().join("clones");
        let root_text = root.to_string_lossy().into_owned();
        let (decorated, reached, _transcript) = dispatcher_with_root(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "echo {workspace_root}".to_owned(),
            }),
            Err("terminal:the worker path must never be reached".to_owned()),
            WorkspaceRoot::from_resolution(Ok(root.clone())),
        );
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("provision", "{}")));
        let result = handle.await?;
        let encoded = result.map_err(|error| format!("declared command failed: {error}"))?;
        let outcome: serde_json::Value = serde_json::from_str(&encoded)?;
        assert_eq!(
            outcome["stdout"], root_text,
            "the command must observe the server-resolved root as its argv word"
        );
        assert_eq!(outcome["exit_code"], 0);
        assert!(
            root.is_dir(),
            "dispatching a placeholder-bearing body must create the missing root"
        );
        assert!(reached_names(&reached).is_empty());
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_placeholder_bearing_body_refuses_terminally_when_the_root_is_unresolved()
    -> TestResult {
        let (decorated, reached, _transcript) = dispatcher_with_root(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "echo {workspace_root}".to_owned(),
            }),
            Ok("unused".to_owned()),
            WorkspaceRoot::from_resolution(Err(WorkspaceRootError::Unresolvable {
                reason: "cannot resolve Aion home: set AION_HOME or HOME".to_owned(),
            })),
        );
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("provision", "{}")));
        let Err(error) = handle.await? else {
            return Err("an unresolved root must refuse a placeholder-bearing body".into());
        };
        assert!(error.starts_with("terminal:"), "{error}");
        assert!(
            error.contains("provision"),
            "the refusal must name the action: {error}"
        );
        assert!(
            error.contains("cannot resolve Aion home"),
            "the refusal must carry the resolution failure's reason: {error}"
        );
        assert!(
            reached_names(&reached).is_empty(),
            "a refused body must not fall through to the worker path"
        );
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_shape_changing_root_refuses_terminally_naming_the_action() -> TestResult {
        // `$` in the root would open a parameter reference after splicing.
        let (decorated, reached, _transcript) = dispatcher_with_root(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "echo {workspace_root}".to_owned(),
            }),
            Ok("unused".to_owned()),
            WorkspaceRoot::from_resolution(Ok(std::path::PathBuf::from("/absolute/with$dollar"))),
        );
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("provision", "{}")));
        let Err(error) = handle.await? else {
            return Err("a shape-changing root must refuse a placeholder-bearing body".into());
        };
        assert!(error.starts_with("terminal:"), "{error}");
        assert!(
            error.contains("provision"),
            "the refusal must name the action: {error}"
        );
        assert!(
            error.contains("would change the parsed shape"),
            "the refusal must carry the shape-changing diagnosis: {error}"
        );
        assert!(
            reached_names(&reached).is_empty(),
            "a refused body must not fall through to the worker path"
        );
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn an_uncreatable_root_refuses_terminally_naming_the_action() -> TestResult {
        // A root beneath a regular file cannot be created by any retry.
        let scratch = tempfile::tempdir()?;
        let file = scratch.path().join("occupied");
        std::fs::write(&file, b"not a directory")?;
        let (decorated, reached, _transcript) = dispatcher_with_root(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "echo {workspace_root}".to_owned(),
            }),
            Ok("unused".to_owned()),
            WorkspaceRoot::from_resolution(Ok(file.join("clones"))),
        );
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("provision", "{}")));
        let Err(error) = handle.await? else {
            return Err("an uncreatable root must refuse a placeholder-bearing body".into());
        };
        assert!(error.starts_with("terminal:"), "{error}");
        assert!(
            error.contains("provision"),
            "the refusal must name the action: {error}"
        );
        assert!(
            error.contains("could not be created"),
            "the refusal must carry the creation-failure diagnosis: {error}"
        );
        assert!(
            reached_names(&reached).is_empty(),
            "a refused body must not fall through to the worker path"
        );
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_body_without_the_placeholder_is_untouched_by_resolution_failure() -> TestResult {
        let (decorated, _reached, _transcript) = dispatcher_with_root(
            DeclaredBodyLookup::Declared(ActionBodyContract::Run {
                command: "echo $greeting".to_owned(),
            }),
            Ok("unused".to_owned()),
            WorkspaceRoot::from_resolution(Err(WorkspaceRootError::Unresolvable {
                reason: "cannot resolve Aion home: set AION_HOME or HOME".to_owned(),
            })),
        );
        let handle = tokio::task::spawn_blocking(move || {
            decorated.dispatch(request("greet", "{\"greeting\":\"still served\"}"))
        });
        let result = handle.await?;
        let encoded = result.map_err(|error| format!("declared command failed: {error}"))?;
        let outcome: serde_json::Value = serde_json::from_str(&encoded)?;
        assert_eq!(outcome["stdout"], "still served");
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn an_unreadable_catalog_delegates_to_the_worker_path() -> TestResult {
        let (decorated, reached) = dispatcher(
            DeclaredBodyLookup::Unreadable("catalog offline".to_owned()),
            Ok("\"served anyway\"".to_owned()),
        );
        let handle =
            tokio::task::spawn_blocking(move || decorated.dispatch(request("resilient", "{}")));
        let result = handle.await?;
        assert_eq!(result, Ok("\"served anyway\"".to_owned()));
        assert_eq!(reached_names(&reached), vec!["resilient".to_owned()]);
        Ok(())
    }

    #[test]
    fn non_object_input_is_refused_terminally_by_shape() {
        for (input, kind) in [
            ("[1,2]", "an array"),
            ("\"text\"", "a string"),
            ("3", "a number"),
            ("null", "null"),
            ("true", "a boolean"),
        ] {
            let Err(error) = decode_arguments(input) else {
                unreachable_refusal(input);
                return;
            };
            assert!(error.starts_with("terminal:"), "{error}");
            assert!(error.contains(kind), "{error} must name {kind}");
        }
    }

    /// Fails the calling test without a panicking accessor.
    fn unreachable_refusal(input: &str) {
        assert!(
            input.is_empty(),
            "input `{input}` must have been refused by shape"
        );
    }

    /// The selection rule cannot be right if it is asked about the wrong run.
    ///
    /// `select_declared_body` is unit-tested on inputs the test itself
    /// constructs, which proves the rule and nothing about the plumbing. This
    /// asserts the other half: the identity the dispatcher hands the reader is
    /// the identity of the dispatch it is serving, not a placeholder and not
    /// another run's.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_reader_is_asked_about_the_run_that_is_dispatching() -> TestResult {
        let seen = Arc::new(Mutex::new(Vec::new()));
        let bodies = DeclaredBodySource::default();
        bodies.install(Arc::new(RecordingBodies {
            seen: Arc::clone(&seen),
        }));
        let reached = Arc::new(Mutex::new(Vec::new()));
        let decorated = DeclaredCommandDispatcher::new(
            Arc::new(RecordingInner {
                reached: Arc::clone(&reached),
                reply: Ok("\"worker-served\"".to_owned()),
            }),
            bodies,
            DeclaredCommandAttempts::new(),
            tokio::runtime::Handle::current(),
            WorkspaceRoot::from_resolution(Ok(std::env::temp_dir())),
            ActivityEventPublisher::new(
                Arc::new(aion_store::InMemoryObservabilityStore::default()),
                TRANSCRIPT_CAPACITY,
                crate::activity_publisher::TranscriptBatchPolicy::UNBATCHED,
            ),
        );

        let dispatch = request("plain", "{}");
        let expected = (dispatch.workflow_id.clone(), dispatch.run_id.clone());
        let handle = tokio::task::spawn_blocking(move || decorated.dispatch(dispatch));
        handle
            .await?
            .map_err(|error| format!("dispatch failed: {error}"))?;

        let observed = match seen.lock() {
            Ok(observed) => observed.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        };
        assert_eq!(
            observed,
            vec![expected],
            "the body reader must be asked about the dispatching run itself"
        );
        Ok(())
    }
}