mlua_swarm/core/

engine.rs

//! `Engine` — the long-running stateful runtime plus the `with_state`
//! helper (R1-R4 discipline).
//!
//! The engine owns the Domain side of the Data / Domain split:
//! flow control (dispatch / verdict), state (`EngineState`), and the
//! `submit_output` / `output_tail` surface that feeds it. Data-plane
//! traffic (Big Response bodies) is delegated to the `output_store` module
//! plus its paired `SpawnerLayer`s and passes through here without the
//! engine core needing to grow.

use crate::core::agent_context::{RUN_ID_KEY, STEP_CTX_KEY};
use crate::core::config::EngineCfg;
use crate::core::ctx::{Ctx, OperatorInfo, OperatorKind, SeniorBridge, SpawnHook};
use crate::core::errors::EngineError;
use crate::core::state::{
    unwrap_skip_marker, wrap_skip_marker, CapTokenRecord, DispatchOutcome, EngineState, Event,
    EventStream, OperatorSession, ResumeKey, ResumePending, SubmitOutcome, TaskSpec, TaskState,
    TaskStatus,
};
use crate::store::replay::{hash_input_value, ReplayEntry};
use crate::store::run::RunContext;
use crate::types::{
    default_role_verb_table, now_unix, CapToken, Role, RoleVerbGate, RunId, SessionId, StepId,
    TokenSigner, Verb,
};
use crate::worker::adapter::SpawnerAdapter;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, Mutex};

/// Process-wide long-running runtime. Cheap to `clone()` — an `Arc`
/// lives inside.
#[derive(Clone)]
pub struct Engine {
    inner: Arc<EngineInner>,
}

struct EngineInner {
    state: Mutex<EngineState>,
    cfg: EngineCfg,
    signer: TokenSigner,
    gate: RoleVerbGate,
    event_tx: broadcast::Sender<Event>,
    /// ID-keyed bridge registry (register-by-ID design). `SeniorBridge`
    /// and `SpawnHook` are registered by ID; sessions bind to those IDs
    /// only. Persistence stores just the ID, and on reattach the caller
    /// re-registers under the same ID to restore presence.
    senior_bridges: tokio::sync::RwLock<HashMap<String, Arc<dyn SeniorBridge>>>,
    spawn_hooks: tokio::sync::RwLock<HashMap<String, Arc<dyn SpawnHook>>>,
    /// ID registry for full-spawn Operator backends (backends that take the
    /// entire spawn via `execute`). Sibling to `senior_bridges` /
    /// `spawn_hooks`. `OperatorDelegateMiddleware` looks these up via
    /// `ctx` and, when `kind = MainAi` / `Composite`, bypasses
    /// `inner.spawn` and calls `operator.execute` instead.
    operators: tokio::sync::RwLock<HashMap<String, Arc<dyn crate::operator::Operator>>>,
    /// Base and hint layer factories for the `SpawnerStack`. At
    /// `service::linker::link` time, `compiled.router` is wrapped with
    /// the base factories plus the hint factories resolved from
    /// `blueprint.spawner_hints.layers`. This is the engine-side
    /// counterpart to the discipline "Flow / Blueprint doesn't spell out
    /// middleware implementations — it declares the capabilities it needs
    /// as hint keys".
    layer_registry: crate::middleware::LayerRegistry,
    /// Optional Data-plane `OutputStore` backend (subtask-4 / ST2 rework —
    /// see `submit_output`'s doc). `None` (the default) preserves
    /// pre-subtask-4 behavior exactly: `submit_output` /
    /// `submit_worker_result_trusted` only touch the Domain-plane
    /// `EngineState.output_store` HashMap, same as before this was added.
    /// `Some` additionally dual-writes every `Final` event into this store
    /// via [`crate::store::output::OutputStore::append`], making it
    /// queryable (e.g. by `mlua-swarm-server`'s `GET /v1/tasks/:id/ctx`)
    /// even for an in-flight run. A plain `std::sync::RwLock` (not
    /// `tokio::sync::RwLock`) — set once at boot via [`Engine::set_output_store`]
    /// from a synchronous call site (`mlua-swarm-server`'s router builder),
    /// then only ever briefly read (clone the `Option<Arc<..>>`, never held
    /// across an `.await`) from the async submit path.
    data_store: std::sync::RwLock<Option<Arc<dyn crate::store::output::OutputStore>>>,
    /// GH #50 (Subtask 2 — runtime plumbing): agent name → declared
    /// [`mlua_swarm_schema::VerdictContract`], the Engine-side registry
    /// [`Self::verdict_contract_for_task`] resolves against. Populated via
    /// [`Self::register_verdict_contracts`] — same sync-`RwLock`,
    /// set-outside-the-lock idiom as `data_store` above. Empty by default
    /// (every pre-GH-#50 `Engine`), which is exactly the opt-in "no
    /// contract declared" state `verdict_contract_for_task` treats as
    /// `None`. Populated from a live `Compiler::compile`'s
    /// `CompiledAgentTable.verdict_contracts` output by
    /// `TaskLaunchService::launch`, immediately after `compiler.compile`
    /// succeeds — see [`Self::register_verdict_contracts`]'s doc for the
    /// overwrite semantics of that merge.
    verdict_contracts: std::sync::RwLock<HashMap<String, mlua_swarm_schema::VerdictContract>>,
}

/// Renders a `TaskSpec.initial_directive` / `EngineState.prompts`
/// `Value` down to the `String` shape that string-consuming boundaries
/// require (issue #18). Strings pass through verbatim; anything else
/// (Object / Array / Number / Bool / Null) is serde-stringified. This
/// is the single canonical rendering — the coercion that used to sit
/// inside `EngineDispatcher::dispatch` moved here and is invoked only
/// at consumer boundaries: `WorkerPayload.prompt` (HTTP
/// `/v1/worker/prompt`), `WorkerInvocation.prompt` (in-process
/// spawners), the subprocess spawner's directive arg/stdin, and the
/// WS Spawn frame text render (`operator_ws::session`). Everything
/// upstream (Blueprint dispatch → engine state → `fetch_prompt` →
/// `Operator::execute`) keeps the `Value` end-to-end.
pub(crate) fn render_directive_to_string(v: &Value) -> String {
    match v {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

/// Renders a [`crate::worker::output::ContentRef`] down to the `Value` shape
/// the BP-chain / `DispatchOutcome` consume. `Inline` passes its `value`
/// through verbatim; `FileRef` is stringified into the same
/// `{"file_ref", "mime", "size_hint"}` shape `materialize_final_submission`
/// uses for its own file-materialize projection — one canonical
/// stringification, not two independently-maintained copies (GH #36 ST1:
/// shared by both the `Final`-pull and the `Artifact`-parts fold in
/// [`Engine::dispatch_attempt_with`]'s doc).
fn content_ref_to_value(content: crate::worker::output::ContentRef) -> Value {
    match content {
        crate::worker::output::ContentRef::Inline { value } => value,
        crate::worker::output::ContentRef::FileRef {
            path,
            mime,
            size_hint,
        } => serde_json::json!({
            "file_ref": path.to_string_lossy(),
            "mime": mime,
            "size_hint": size_hint,
        }),
    }
}

/// GH #51 — reduces a [`content_ref_to_value`] result down to the `String`
/// shape the completion-time verdict-contract check compares against a
/// declared `VerdictContract.values` token set. A `Value::String` unwraps
/// to its raw contents (no surrounding JSON quotes) — this mirrors the
/// pre-GH-#51 `check_verdict_contract` (`mlua-swarm-server`'s
/// `worker.rs`), which always compared the raw submitted body string
/// directly, never a JSON-stringified copy. Any OTHER `Value` shape
/// (`Number` / `Object` / `Array` / `Bool` / `Null` — i.e. a `channel:
/// "body"` contract whose completing value is not a string at all, or a
/// `FileRef` content whose `content_ref_to_value` projection is an
/// object) falls back to `Value::to_string()`'s JSON-encoded form: it can
/// never collide with a plain declared token like `"PASS"`, so it
/// naturally fails membership — consistent with the "non-string values
/// under a body contract are violations" rule (issue #51's Proposal).
fn content_ref_to_comparable_string(content: crate::worker::output::ContentRef) -> String {
    let value = content_ref_to_value(content);
    match value {
        Value::String(s) => s,
        other => other.to_string(),
    }
}

/// `AgentContextView.extra` key carrying a step's declared submit format.
/// Declared through the GH #21 meta channels (`Blueprint.metas` /
/// `AgentMeta.ctx` / step-level `$step_meta`) and folded into the view at
/// spawn time by `AgentContextMiddleware`. Read in two places: the HTTP
/// submit lane (`mlua-swarm-server`'s `resolve_submit_value`, where
/// `"json"` means strict parse-or-422) and [`Engine::fold_parse_mode_for`]
/// (where [`SUBMIT_FORMAT_TEXT`] opts the step's fold out of the default
/// lenient container parse — see [`FoldParse`]).
pub const SUBMIT_FORMAT_KEY: &str = "submit_format";

/// The [`SUBMIT_FORMAT_KEY`] value that opts a step's fold out of lenient
/// container parsing ([`FoldParse::Raw`]): every string the worker
/// submitted — final body and staged parts alike — folds into the flow
/// ctx as itself, even when its bytes would parse as a JSON object or
/// array.
pub const SUBMIT_FORMAT_TEXT: &str = "text";

/// How [`fold_final_and_parts`] treats `Value::String` content when
/// assembling the BP-chain value — the fold half of the
/// [`SUBMIT_FORMAT_KEY`] contract.
///
/// `Lenient` is the default for every step: a string whose bytes parse as
/// a JSON **object or array** folds as the parsed structure, so a
/// downstream node can address fields inside it (`$.<step>.lanes`, a
/// `fanout` `items` expression, a `branch` cond) with no declaration —
/// uniformly across the HTTP submit, artifact staging, and in-process
/// lanes, because they all meet here. Scalar JSON (`true`, `42`,
/// `"quoted"`, `null`) deliberately stays a string: a scalar has no
/// addressable interior, so parsing it buys no path capability while
/// silently changing `Eq` conds and verdict comparisons for any declared
/// token that happens to be valid JSON. A step that wants full-JSON
/// semantics (scalars included) declares `submit_format: "json"` and gets
/// the strict submit-time parse instead; a step that needs a
/// JSON-container-looking body folded as a raw string declares
/// `submit_format: "text"` (`Raw`: no parsing at all).
///
/// Parsing at the fold — not at staging — is also what keeps materialized
/// part files verbatim: `Engine::stage_worker_artifact_trusted` /
/// `materialize_part` still see the submitted `Value::String` bytes, and
/// so do the verdict-contract checks (staging-time and completion-time),
/// which all run before the fold.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FoldParse {
    /// Default: fold a JSON-container string as its parsed structure.
    Lenient,
    /// `submit_format: "text"` opt-out: fold every string as itself.
    Raw,
}

/// The `Lenient` half of [`FoldParse`]: parse a `Value::String` whose
/// bytes lead with `{` / `[` AND parse as JSON; pass every other value
/// through untouched. The leading-byte check keeps large prose bodies (a
/// `plan.md` part, an operator completion notice) from paying a parse
/// that could only fail, and is what scopes the parse to containers — a
/// scalar body never enters `from_str` at all.
fn lenient_fold_value(v: Value) -> Value {
    let Value::String(s) = v else { return v };
    let trimmed = s.trim_start();
    if !(trimmed.starts_with('{') || trimmed.starts_with('[')) {
        return Value::String(s);
    }
    match serde_json::from_str::<Value>(&s) {
        Ok(parsed) => parsed,
        Err(_) => Value::String(s),
    }
}

/// [`Engine::dispatch_attempt_with`]'s Final-pull assembly (GH #36 ST1:
/// named multi-part worker output), factored out as a pure function of the
/// output-event tail so it is unit-testable without a live `Engine` /
/// spawner.
///
/// Finds the LAST `Final` event in `tail` (mirrors the pre-GH-#36 pull:
/// "last Final wins" if more than one was ever appended) and folds every
/// `Artifact` event in the SAME tail WHOSE NAME APPEARS IN `staged_names`
/// into a `"parts"` object keyed by `Artifact.name` — walked in tail (=
/// event-append) order, so a name staged more than once within the attempt
/// is last-write-wins (`Map` insert semantics, not an accumulating list;
/// `Engine::stage_worker_artifact_trusted`'s doc). `staged_names` is the
/// WORKER's own opt-in allowlist (`EngineState.worker_artifact_names`'s
/// doc) — an `Artifact` on the tail whose name is NOT in `staged_names`
/// (e.g. `AfterRunAuditMiddleware`'s `"audit:<step_ref>"` sidecar finding)
/// is left alone, exactly as before GH #36; this is what keeps an audited
/// step's BP-chain value byte-identical when the worker itself never
/// staged a part.
///
/// At least one matching part: the returned value is `{"out": <final
/// value>, "parts": {<name>: <value>, ...}}`. Zero matching parts: the
/// returned value is the plain final value, unchanged from the pre-GH-#36
/// shape — this is the back-compat guarantee, not an incidental default.
///
/// `None` when `tail` carries no `Final` at all (the caller's pre-existing
/// "no Final in output_tail" error path).
///
/// `mode` applies [`lenient_fold_value`] to the final value AND every
/// folded part when `Lenient` (the default resolved by
/// [`Engine::fold_parse_mode_for`]); `Raw` reproduces the pre-fold-parse
/// behavior byte-for-byte. A value that is already structured (a strict
/// `submit_format: "json"` body parsed at submit time, an in-process Lua
/// table) passes through either way.
fn fold_final_and_parts(
    tail: &[crate::worker::output::OutputEvent],
    staged_names: &[String],
    mode: FoldParse,
) -> Option<(Value, bool)> {
    let fold = |v: Value| match mode {
        FoldParse::Lenient => lenient_fold_value(v),
        FoldParse::Raw => v,
    };
    let (final_content, ok) = tail.iter().rev().find_map(|ev| match ev {
        crate::worker::output::OutputEvent::Final { content, ok } => Some((content.clone(), *ok)),
        _ => None,
    })?;
    let final_value = fold(content_ref_to_value(final_content));

    let mut parts = serde_json::Map::new();
    for ev in tail {
        if let crate::worker::output::OutputEvent::Artifact { name, content } = ev {
            if staged_names.iter().any(|staged| staged == name) {
                parts.insert(name.clone(), fold(content_ref_to_value(content.clone())));
            }
        }
    }

    let value = if parts.is_empty() {
        final_value
    } else {
        serde_json::json!({ "out": final_value, "parts": Value::Object(parts) })
    };
    Some((value, ok))
}

impl Engine {
    /// Backwards-compatible constructor that starts the engine without a
    /// layer registry, preserving the signature already used by ~88
    /// existing call sites. Use this when automatic middleware wrapping
    /// at bind time is not needed. Callers such as `mlua-swarm-server` go through
    /// `new_with_layers(cfg, registry)` to enable the hint-resolution path.
    pub fn new(cfg: EngineCfg) -> Self {
        Self::new_with_layers(cfg, crate::middleware::LayerRegistry::new())
    }

    /// Construct an `Engine` with an explicit `LayerRegistry`, enabling
    /// hint-resolution: `spawner_hints.layers` declared on a `Blueprint`
    /// are resolved against this registry when the spawner stack is bound
    /// at `service::linker::link` time.
    pub fn new_with_layers(
        cfg: EngineCfg,
        layer_registry: crate::middleware::LayerRegistry,
    ) -> Self {
        let (event_tx, _) = broadcast::channel(256);
        let signer = TokenSigner::new(&cfg.token_secret);
        Self {
            inner: Arc::new(EngineInner {
                state: Mutex::new(EngineState::new()),
                cfg,
                signer,
                gate: default_role_verb_table(),
                event_tx,
                senior_bridges: tokio::sync::RwLock::new(HashMap::new()),
                spawn_hooks: tokio::sync::RwLock::new(HashMap::new()),
                operators: tokio::sync::RwLock::new(HashMap::new()),
                layer_registry,
                data_store: std::sync::RwLock::new(None),
                verdict_contracts: std::sync::RwLock::new(HashMap::new()),
            }),
        }
    }

    /// Rebuild this `Engine` with a different `RoleVerbGate`. The gate is
    /// treated as fixed-at-build-time, so this constructs a fresh
    /// `EngineInner` (fresh empty `EngineState`) rather than mutating in
    /// place — mainly a testing convenience for swapping gate rules.
    pub fn with_gate(self, gate: RoleVerbGate) -> Self {
        // The gate is fixed at build time — the intent is to build a fresh
        // instance rather than mutating in place. As a testing convenience we
        // do allow swapping the inner Arc. Simpler form: just rebuild
        // Arc<EngineInner>.
        let inner = Arc::new(EngineInner {
            state: Mutex::new(EngineState::new()),
            cfg: self.inner.cfg.clone(),
            signer: self.inner.signer.clone(),
            gate,
            event_tx: self.inner.event_tx.clone(),
            senior_bridges: tokio::sync::RwLock::new(HashMap::new()),
            spawn_hooks: tokio::sync::RwLock::new(HashMap::new()),
            operators: tokio::sync::RwLock::new(HashMap::new()),
            layer_registry: self.inner.layer_registry.clone(),
            data_store: std::sync::RwLock::new(None),
            verdict_contracts: std::sync::RwLock::new(HashMap::new()),
        });
        Self { inner }
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Accessors. Production code drives execution through compile +
    // `service::linker::link` + `dispatch_attempt_with(spawner)` inside
    // `TaskLaunchService`; `Engine` itself is a pure execution surface — it
    // does not own a BlueprintStore / EnhanceAdapter / Compiler, nor a
    // global spawner (the spawner is carried per-request, never stashed on
    // the engine).
    // ═══════════════════════════════════════════════════════════════════════

    /// Access the `EngineCfg` this engine was built with.
    pub fn cfg(&self) -> &EngineCfg {
        &self.inner.cfg
    }

    /// Expose the internal `LayerRegistry` — used when deriving a
    /// sub-engine that needs the same registry re-injected. The
    /// per-request sub-engine in `mlua-swarm-server` reads the parent engine's
    /// registry through this accessor and passes it to
    /// `Engine::new_with_layers(cfg, parent.layer_registry().clone())`.
    pub fn layer_registry(&self) -> &crate::middleware::LayerRegistry {
        &self.inner.layer_registry
    }

    /// Access the `TokenSigner` used to mint/verify `CapToken`s.
    pub fn signer(&self) -> &TokenSigner {
        &self.inner.signer
    }

    /// Clone a handle to the process-wide `Event` broadcast sender. Prefer
    /// `subscribe` for a ready-to-use receiver.
    pub fn event_tx(&self) -> broadcast::Sender<Event> {
        self.inner.event_tx.clone()
    }

    /// Subscribe to the engine's `Event` broadcast stream.
    pub fn subscribe(&self) -> EventStream {
        self.inner.event_tx.subscribe()
    }

    /// Wires the Data-plane [`crate::store::output::OutputStore`] backend
    /// used by `submit_output` / `submit_worker_result_trusted`'s
    /// submit-time projection sink (subtask-4 / ST2 rework — see
    /// `submit_output`'s doc). Synchronous (a plain `std::sync::RwLock`
    /// write) so a caller can wire it up at boot from a non-`async`
    /// context (`mlua-swarm-server`'s router builder passes the same
    /// `Arc` it hands to its `AppState.data_store`, so `POST
    /// /v1/data/emit` and every worker's ordinary `/v1/worker/submit` land
    /// in the one store). Calling this more than once replaces the
    /// previous backend; not calling it at all (the default) preserves
    /// pre-subtask-4 behavior exactly — `submit_output` only touches the
    /// Domain-plane `EngineState.output_store` HashMap.
    pub fn set_output_store(&self, store: Arc<dyn crate::store::output::OutputStore>) {
        let mut guard = self
            .inner
            .data_store
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        *guard = Some(store);
    }

    /// Clones the currently-wired Data-plane store handle, if any. Kept
    /// private and side-effect-free (no lock held past this call) —
    /// callers (`materialize_final_submission`) do their actual `.append`
    /// work outside of any lock.
    fn output_store_backend(&self) -> Option<Arc<dyn crate::store::output::OutputStore>> {
        self.inner
            .data_store
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }

    /// GH #50 (Subtask 2): merges `contracts` (agent name → declared
    /// [`mlua_swarm_schema::VerdictContract`]) into the engine's runtime
    /// verdict-contract registry, later resolved per-task by
    /// [`Self::verdict_contract_for_task`]. Same sync-write idiom as
    /// [`Self::set_output_store`] — a plain `std::sync::RwLock` write, so
    /// this can be called from a non-`async` context. Production call
    /// site: `TaskLaunchService::launch`, immediately after a successful
    /// `Compiler::compile`, passing `compiled.router.verdict_contracts.clone()`.
    ///
    /// # Overwrite semantics (explicit — read before adding a second call site)
    ///
    /// The registry is a single flat `HashMap` **keyed by agent name only**
    /// (`String`), with process-wide (not per-task, not per-Blueprint,
    /// not per-launch) scope. Registration is additive via
    /// `HashMap::extend`: an entry for an agent name NOT already present is
    /// added; an entry for an agent name ALREADY present is REPLACED
    /// (last write wins) by the incoming one. Concretely: launching a
    /// second Blueprint that also declares a `verdict` contract for an
    /// agent named `"gate"` OVERWRITES whatever contract a first, still
    /// in-flight, launch registered for an agent of that same name — even
    /// if the two Blueprints intend it as two semantically different
    /// agents that merely share a name, and even while the first launch's
    /// tasks are still running. This is a **known limitation** of the v1
    /// design; a per-task (or per-`RunId` / per-Blueprint) scoped registry
    /// is a possible follow-up if two concurrently in-flight Blueprints
    /// declaring conflicting contracts under the same agent name turns out
    /// to matter in practice. Calling this with an empty map (or not at
    /// all — the default) is a no-op, preserving pre-GH-#50 behavior
    /// exactly (opt-in).
    pub fn register_verdict_contracts(
        &self,
        contracts: HashMap<String, mlua_swarm_schema::VerdictContract>,
    ) {
        let mut guard = self
            .inner
            .verdict_contracts
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        guard.extend(contracts);
    }

    /// GH #50 (Subtask 2): the declared
    /// [`mlua_swarm_schema::VerdictContract`] for the agent currently
    /// running `task_id`, if any. Resolves `task_id` → `TaskState.spec.agent`
    /// (via `EngineState.tasks`, the same lookup [`Self::task_attempt`]
    /// performs) and looks that agent name up in the registry
    /// [`Self::register_verdict_contracts`] populates.
    ///
    /// `None` in both of these cases — deliberately collapsed to the same
    /// value, mirroring [`Self::agent_context_for`]'s `Result`-into-`Option`
    /// pattern (`.ok().flatten()`; a lookup failure here is never itself an
    /// error worth surfacing to a caller):
    /// - `task_id` is unknown (no `TaskState` for it).
    /// - `task_id` resolves to a known agent, but that agent declared no
    ///   `verdict` contract (the opt-in default).
    ///
    /// Callers (`mlua-swarm-server`'s `worker_submit` / `worker_artifact`)
    /// treat every `None` identically: skip the submit-time verdict gate
    /// entirely, preserving pre-GH-#50 behavior byte-for-byte.
    pub async fn verdict_contract_for_task(
        &self,
        task_id: &StepId,
    ) -> Option<mlua_swarm_schema::VerdictContract> {
        let tid = task_id.clone();
        let agent = self
            .with_state("verdict_contract_for_task", move |s| {
                s.tasks.get(&tid).map(|t| t.spec.agent.clone())
            })
            .await
            .ok()
            .flatten()?;
        self.inner
            .verdict_contracts
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(&agent)
            .cloned()
    }

    /// GH #51 — the value of the LAST staged `"verdict"` `Artifact` for
    /// `(task_id, attempt)`, if any. Mirrors [`fold_final_and_parts`]'s
    /// reverse-scan-of-`output_tail` pattern (last-write-wins per name,
    /// same as that fold and [`Self::stage_worker_artifact_trusted`]'s
    /// doc), narrowed to the single literal artifact name
    /// `channel: "part"` contracts address (Pattern B — see
    /// `blueprint-authoring.md`'s "Returning verdicts to drive BP flow").
    ///
    /// Infallible accessor: `None` is the normal "nothing staged yet"
    /// case, not an error — the caller
    /// ([`Self::verdict_contract_completion_check`]) is what converts
    /// `None` into `Err(EngineError::VerdictPartMissing)`.
    pub(crate) async fn staged_verdict_value_for(
        &self,
        task_id: &StepId,
        attempt: u32,
    ) -> Option<String> {
        let tail = self.output_tail(task_id, attempt).await;
        tail.iter().rev().find_map(|ev| match ev {
            crate::worker::output::OutputEvent::Artifact { name, content } if name == "verdict" => {
                Some(content_ref_to_comparable_string(content.clone()))
            }
            _ => None,
        })
    }

    /// GH #51 — the single completion-time verdict-contract choke point,
    /// embedded inside BOTH [`Self::submit_worker_result_trusted`] and
    /// [`Self::submit_output`] (the two engine-side writes every HTTP/WS
    /// completion route ultimately passes through). Not duplicated per
    /// route handler — a future 4th completion route is gated for free
    /// as long as it funnels through one of those two functions.
    ///
    /// `ok=false` is exempt on every route (this single early-return IS
    /// the exemption, reused identically by both embedding sites — see
    /// issue #51's "ok=false completions are exempt" acceptance
    /// criterion). An agent with no declared contract, or a contract for
    /// the OTHER channel, is untouched (`Ok(())`) — same opt-in,
    /// byte-for-byte-preserving posture as
    /// [`Self::verdict_contract_for_task`]'s doc.
    ///
    /// - `channel: "body"` — `value` (the completing `Final`'s content,
    ///   already reduced to a comparable string by the caller via
    ///   [`content_ref_to_comparable_string`]) must be a member of
    ///   `contract.values`.
    /// - `channel: "part"` — [`Self::staged_verdict_value_for`] must find
    ///   a staged `"verdict"` artifact for this attempt (presence,
    ///   defense in depth over the staging-time membership check) AND its
    ///   value must be a member of `contract.values`.
    async fn verdict_contract_completion_check(
        &self,
        task_id: &StepId,
        attempt: u32,
        ok: bool,
        value: &str,
    ) -> Result<(), EngineError> {
        if !ok {
            return Ok(());
        }
        let Some(contract) = self.verdict_contract_for_task(task_id).await else {
            return Ok(());
        };
        match contract.channel {
            mlua_swarm_schema::VerdictChannel::Body => {
                if contract.values.iter().any(|v| v == value) {
                    Ok(())
                } else {
                    Err(EngineError::VerdictValueRejected {
                        value: value.to_string(),
                        allowed: contract.values.clone(),
                    })
                }
            }
            mlua_swarm_schema::VerdictChannel::Part => {
                match self.staged_verdict_value_for(task_id, attempt).await {
                    None => Err(EngineError::VerdictPartMissing {
                        allowed: contract.values.clone(),
                    }),
                    Some(staged) if contract.values.iter().any(|v| v == &staged) => Ok(()),
                    Some(staged) => Err(EngineError::VerdictValueRejected {
                        value: staged,
                        allowed: contract.values.clone(),
                    }),
                }
            }
        }
    }

    // ═══════════════════════════════════════════════════════════════════════
    // §7 with_state — single Mutex + R1-R4 (try_lock + bounded retry + max-hold panic)
    // ═══════════════════════════════════════════════════════════════════════

    /// The closure is a **sync** `FnOnce` — you cannot pass an async
    /// closure, which enforces R3 at the type level. Exceeding `max_hold`
    /// emits a `tracing::warn!` and continues, so a load-dependent overrun
    /// never unwinds the caller's task; set `EngineCfg::max_hold_panic`
    /// to escalate the overrun to a panic when hunting an R3 violation.
    pub async fn with_state<F, R>(&self, op: &'static str, f: F) -> Result<R, EngineError>
    where
        F: FnOnce(&mut EngineState) -> R,
    {
        let cfg = &self.inner.cfg;

        // R2: try_lock + bounded retry
        let mut guard_opt = None;
        for attempt in 0..=cfg.max_retry {
            match self.inner.state.try_lock() {
                Ok(g) => {
                    guard_opt = Some(g);
                    break;
                }
                Err(_) if cfg.try_only => return Err(EngineError::LockBusy(op)),
                Err(_) => {
                    let backoff = cfg.backoff_ms_step * (attempt as u64 + 1);
                    tokio::time::sleep(Duration::from_millis(backoff)).await;
                }
            }
        }
        let mut guard = guard_opt.ok_or(EngineError::LockBusyAfterRetry(op))?;

        // R4: max_hold guard
        let start = Instant::now();
        let result = f(&mut guard);
        let elapsed_ms = start.elapsed().as_millis();
        drop(guard);

        if elapsed_ms > cfg.max_hold_ms {
            // R4 violation. Warn-and-continue is the default in every build:
            // elapsed is wall-clock time, so on a loaded shared runner it
            // includes scheduler preemption and a panic here is structurally
            // flaky (and kills the run driver future, stranding the
            // RunRecord in `Running`). `max_hold_panic` opts back into the
            // hard failure for local R3-violation hunts.
            tracing::warn!(
                op,
                elapsed_ms = %elapsed_ms,
                max_hold_ms = %cfg.max_hold_ms,
                "with_state exceeded max hold — suspected R3 violation (long op inside lock)"
            );
            if cfg.max_hold_panic {
                panic!(
                    "Engine.with_state('{op}') held {elapsed_ms}ms > max {}ms — suspected R3 violation (long op inside lock)",
                    cfg.max_hold_ms
                );
            }
        }
        Ok(result)
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Token verify (= sig + expire + gate + uses_left)
    // ═══════════════════════════════════════════════════════════════════════

    /// Four steps: (1) signature verify, (2) expiry check, (3) role × verb
    /// gate, (4) `uses_left` consume.
    pub async fn verify_token(&self, token: &CapToken, verb: Verb) -> Result<(), EngineError> {
        // (1) sig
        if !self.inner.signer.verify_sig(token) {
            return Err(EngineError::BadSignature);
        }
        // (2) expire
        if token.is_expired(now_unix()) {
            return Err(EngineError::TokenExpired);
        }
        // (3) role × verb gate
        if !self.inner.gate.is_allowed(token.role, verb) {
            return Err(EngineError::RoleViolation {
                role: token.role,
                verb,
            });
        }
        // (4) server-side uses_left consume
        let fp = token.fingerprint();
        self.with_state("token.consume", move |s| {
            let rec = s
                .tokens
                .get_mut(&fp)
                .ok_or_else(|| EngineError::TokenNotFound(fp.clone()))?;
            rec.consume()
                .map_err(|_: crate::core::state::CapTokenConsumeError| {
                    EngineError::TokenUsesExhausted
                })?;
            Ok::<(), EngineError>(())
        })
        .await??;
        Ok(())
    }

    /// `verify_token` plus the **task-ownership gate**.
    ///
    /// When a Worker-role token calls a state-touch verb (`fetch_prompt` /
    /// `post_result` / `read_task_state` / `cancel_task` / `poll_task`),
    /// the gate checks that `CapTokenRecord.task_id` matches the argument
    /// `task_id`; a mismatch returns `EngineError::TokenTaskMismatch`.
    /// Operator / Senior / Observer tokens are outside the ownership gate
    /// and may touch any task.
    ///
    /// **Verbs exempt from the gate.** `start_task` and `dispatch_attempt`
    /// stay outside so recursive swarming keeps working; depth is capped
    /// by `max_spawn_depth`.
    pub async fn verify_token_for_task(
        &self,
        token: &CapToken,
        verb: Verb,
        task_id: &StepId,
    ) -> Result<(), EngineError> {
        self.verify_token(token, verb).await?;
        if token.role != Role::Worker {
            return Ok(());
        }
        let fp = token.fingerprint();
        let arg_tid = task_id.clone();
        self.with_state("token.ownership_gate", move |s| {
            let bound = s.tokens.get(&fp).and_then(|r| r.task_id.as_ref()).cloned();
            match bound {
                Some(t) if t == arg_tid => Ok(()),
                Some(t) => Err(EngineError::TokenTaskMismatch {
                    bound: t.into_string(),
                    arg: arg_tid.into_string(),
                }),
                None => Err(EngineError::TokenNotFound(fp.clone())),
            }
        })
        .await??;
        Ok(())
    }

    /// Resolve the bound `task_id` from a Worker-role token. Used on the
    /// simple `/v1/worker/submit` endpoint, where the worker POSTs with a
    /// token but no `task_id`. Returns `Err` if the token role is not
    /// Worker, or if no bound task is set.
    pub async fn task_id_from_token(&self, token: &CapToken) -> Result<StepId, EngineError> {
        if token.role != Role::Worker {
            return Err(EngineError::RoleViolation {
                role: token.role,
                verb: Verb::PostResult,
            });
        }
        let fp = token.fingerprint();
        self.with_state("task_id_from_token", move |s| {
            s.tokens
                .get(&fp)
                .and_then(|r| r.task_id.as_ref())
                .cloned()
                .ok_or_else(|| EngineError::TokenNotFound(fp.clone()))
        })
        .await?
    }

    /// Resolve a short worker handle (`wh-XXXXXXXX`) to the bound
    /// `task_id`. Used on `/v1/worker/submit` when the Bearer is a short
    /// handle string rather than a full `CapToken` JSON. A missing entry
    /// returns `TokenNotFound`, i.e. "the handle is not in the store".
    pub async fn task_id_from_handle(&self, handle: &str) -> Result<StepId, EngineError> {
        let h = handle.to_string();
        self.with_state("task_id_from_handle", move |s| {
            let fp = s
                .worker_handles
                .get(&h)
                .cloned()
                .ok_or_else(|| EngineError::TokenNotFound(format!("handle={h}")))?;
            s.tokens
                .get(&fp)
                .and_then(|r| r.task_id.as_ref())
                .cloned()
                .ok_or_else(|| EngineError::TokenNotFound(format!("fp={fp}")))
        })
        .await?
    }

    /// Submit a worker result via a short handle. Skips token verification
    /// and updates `output_tail` `Final` + `task.last_result` directly in
    /// a thin path. The caller is expected to have already resolved
    /// `task_id` via `task_id_from_handle` — the handle's presence in
    /// `worker_handles` means it was minted server-side and is therefore
    /// trusted.
    ///
    /// # GH #76 Skip tier: `outcome: SubmitOutcome`
    ///
    /// The `outcome` parameter (replacing the pre-#76 `ok: bool`) is the
    /// caller's tier declaration:
    ///
    /// | outcome  | `Final.ok` | `Final.content`                | verdict-contract check |
    /// |----------|------------|--------------------------------|------------------------|
    /// | `Pass`   | `true`     | `value` verbatim               | fires                  |
    /// | `Blocked`| `false`    | `value` verbatim               | exempt (`ok=false`)    |
    /// | `Skip`   | `true`     | `wrap_skip_marker(value)`      | exempt (Skip opt-out)  |
    ///
    /// The Skip tier is opt-out from the verdict-contract completion check
    /// on the same rationale [`crate::core::state::SubmitOutcome::Skip`]'s
    /// doc records: the agent explicitly declared "not applicable", so
    /// the payload is not a real verdict value to gate.
    pub async fn submit_worker_result_trusted(
        &self,
        task_id: &StepId,
        attempt: u32,
        value: Value,
        outcome: SubmitOutcome,
    ) -> Result<(), EngineError> {
        // Resolve outcome into the wire-level (value, ok, run_contract)
        // triple exactly once, then reuse it below. Keeping the mapping
        // literal in one place makes the "Skip wraps + skips contract"
        // invariant grep-visible.
        let (wire_value, wire_ok, run_contract_check) = match outcome {
            SubmitOutcome::Pass => (value, true, true),
            SubmitOutcome::Blocked => (value, false, false),
            SubmitOutcome::Skip => (wrap_skip_marker(value), true, false),
        };

        // GH #51 — completion-time verdict-contract enforcement, embedded
        // choke point 1 of 2 (see `Self::verdict_contract_completion_check`'s
        // doc). This path always submits a `Final` by construction (there
        // is no other event kind on `/v1/worker/submit`), so the check
        // always applies — unlike `submit_output` below, no `if let
        // OutputEvent::Final { .. }` guard is needed here since there is
        // no other `OutputEvent` variant this function could be asked to
        // write. Runs BEFORE the `output_tail` write immediately below:
        // on `Err`, this returns immediately and neither `with_state` call
        // in this function executes.
        //
        // GH #76 Skip tier: gated on `run_contract_check` — Skip is opt-out
        // (see the outcome mapping table above), Blocked stays exempt via
        // `verdict_contract_completion_check`'s existing `ok=false`
        // early return (redundant flag here for grep locality).
        if run_contract_check {
            let comparable_value =
                content_ref_to_comparable_string(crate::worker::output::ContentRef::Inline {
                    value: wire_value.clone(),
                });
            self.verdict_contract_completion_check(task_id, attempt, wire_ok, &comparable_value)
                .await?;
        }
        let task_id_for_apply = task_id.clone();
        let value_for_event = wire_value.clone();
        self.with_state("submit_worker_result_trusted.output", move |s| {
            let ev = crate::worker::output::OutputEvent::Final {
                content: crate::worker::output::ContentRef::Inline {
                    value: value_for_event,
                },
                ok: wire_ok,
            };
            s.output_store
                .entry((task_id_for_apply.clone(), attempt))
                .or_default()
                .push(ev.clone());
            s.push_event(crate::core::state::Event::WorkerOutput {
                task_id: task_id_for_apply,
                attempt,
                event: ev,
            });
        })
        .await?;
        let task_id_for_result = task_id.clone();
        let value_for_result = wire_value.clone();
        self.with_state("submit_worker_result_trusted.last_result", move |s| {
            if let Some(t) = s.tasks.get_mut(&task_id_for_result) {
                t.last_result = Some(value_for_result);
                t.updated_at = now_unix();
            }
        })
        .await?;
        // subtask-4 / ST2 rework: this path always submits a `Final` (there
        // is no other event kind on `/v1/worker/submit`), so the
        // submit-time projection sink always fires — see
        // `materialize_final_submission`'s doc and `submit_output`'s
        // Invariants (fail-open, never turns a would-have-succeeded submit
        // into a failure).
        let content = crate::worker::output::ContentRef::Inline { value: wire_value };
        self.materialize_final_submission(task_id, attempt, &content, wire_ok)
            .await?;
        Ok(())
    }

    /// Stage a named `Artifact` from a worker via a short handle (GH #36
    /// ST1: named multi-part worker output). Trusted analog of
    /// [`Self::submit_worker_result_trusted`] for `OutputEvent::Artifact`:
    /// skips token verification for the same reason (the caller already
    /// resolved `task_id` via `task_id_from_handle`, so the handle's
    /// presence in `worker_handles` is itself the trust boundary).
    ///
    /// Appends to the same per-`(task_id, attempt)` `output_store` tail
    /// [`Self::dispatch_attempt_with`]'s Final-pull later folds into
    /// `{"out": <final>, "parts": {<name>: <value>, ...}}` (see that
    /// method's doc for the fold semantics — event order, last-write-wins
    /// per name), AND records `name` in `EngineState.worker_artifact_names`
    /// — the fold's allowlist of the WORKER's own staged parts, as opposed
    /// to every `Artifact` that happens to land on the shared tail (e.g. an
    /// audit sidecar finding; see that field's doc). Also dual-writes to
    /// the Data-plane `OutputStore` the same way [`Self::submit_output`]'s
    /// `Artifact` arm does, via [`Self::materialize_artifact_submission`]
    /// (the artifact's own `name` is its Data-plane key, no
    /// canonicalization — see that method's doc).
    pub async fn stage_worker_artifact_trusted(
        &self,
        task_id: &StepId,
        attempt: u32,
        name: String,
        value: Value,
    ) -> Result<(), EngineError> {
        let content = crate::worker::output::ContentRef::Inline { value };
        let task_id_for_apply = task_id.clone();
        let name_for_apply = name.clone();
        let content_for_apply = content.clone();
        self.with_state("stage_worker_artifact_trusted.output", move |s| {
            let ev = crate::worker::output::OutputEvent::Artifact {
                name: name_for_apply.clone(),
                content: content_for_apply,
            };
            s.output_store
                .entry((task_id_for_apply.clone(), attempt))
                .or_default()
                .push(ev.clone());
            s.record_worker_artifact_name(task_id_for_apply.clone(), attempt, name_for_apply);
            s.push_event(crate::core::state::Event::WorkerOutput {
                task_id: task_id_for_apply,
                attempt,
                event: ev,
            });
        })
        .await?;
        self.materialize_artifact_submission(task_id, attempt, &name, &content)
            .await?;
        Ok(())
    }

    /// The in-process lane's half of the "this part is the WORKER's own"
    /// signal: record `name` in `EngineState.worker_artifact_names` for an
    /// `Artifact` that already went through [`Self::submit_output`].
    ///
    /// The out-of-process lane gets this for free inside
    /// [`Self::stage_worker_artifact_trusted`] (one `with_state`, tail
    /// append + name record together). An in-process worker has no HTTP
    /// route to call: it stages through `WorkerInvocation.sink`, which
    /// lands on the generic `submit_output` — the same entry point OTHER
    /// `Artifact` producers use (`AfterRunAuditMiddleware`'s
    /// `"audit:<step_ref>"` sidecar), so `submit_output` itself must NOT
    /// record. The distinction lives one layer up, in
    /// [`crate::worker::output::EngineSink`]: `InProcSpawner::spawn` is
    /// its sole constructor, so an `Artifact` arriving through that sink
    /// is by construction the worker's own, and the sink calls this
    /// immediately after its `submit_output` succeeds.
    ///
    /// Without it a `channel: "part"` in-process gate passes the
    /// completion-time contract check (which reads the tail directly) yet
    /// its part never folds into `{out, parts}` — so a downstream
    /// `$.<step>.parts["verdict"]` cond reads `null`, the exact
    /// half-working state GH #86's sink bridge left behind.
    ///
    /// Two calls rather than one atomic `with_state` is deliberate here:
    /// the tail write must be allowed to fail (contract rejection, strict
    /// `CheckPolicy`) WITHOUT leaving a phantom name behind, so the record
    /// is strictly downstream of a successful submit.
    pub(crate) async fn record_worker_artifact_name(
        &self,
        task_id: &StepId,
        attempt: u32,
        name: String,
    ) -> Result<(), EngineError> {
        let task_id = task_id.clone();
        self.with_state("record_worker_artifact_name", move |s| {
            s.record_worker_artifact_name(task_id, attempt, name);
        })
        .await
    }

    /// GH #36 ST1: the set of `Artifact` names staged for `(task_id,
    /// attempt)` by the worker itself — see
    /// `EngineState.worker_artifact_names`'s doc for the two lanes that
    /// populate it. Used by [`Self::dispatch_attempt_with`]'s Final-pull
    /// to distinguish a worker's own named parts from any other `Artifact`
    /// producer on the same tail.
    async fn worker_artifact_names_for(&self, task_id: &StepId, attempt: u32) -> Vec<String> {
        let key = (task_id.clone(), attempt);
        self.with_state("worker_artifact_names_for", move |s| {
            s.worker_artifact_names
                .get(&key)
                .cloned()
                .unwrap_or_default()
        })
        .await
        .unwrap_or_default()
    }

    /// Mint a short handle and register it in the `worker_handles` map.
    /// Called immediately after the worker-token mint inside
    /// `dispatch_attempt_with`, and issues a handle bound to the same
    /// token fingerprint. Format is `wh-<8 hex chars>` (11 chars total),
    /// designed to remove the base64 copy-paste failure mode.
    async fn mint_worker_handle(&self, worker_fp: String) -> Result<String, EngineError> {
        // The handle is a sole bearer secret on the `/v1/worker/submit`
        // short-handle path (`submit_worker_result_trusted` skips token
        // verification), so it must be unguessable — OS RNG, not the
        // predictable uid counter. 8 hex chars (~4B entropy) keeps the
        // documented `wh-<8 hex>` wire shape; collision between live
        // handles is negligible at in-process handle counts.
        let short = crate::types::secure_hex(4);
        let handle = format!("wh-{short}");
        let h = handle.clone();
        self.with_state("mint_worker_handle", move |s| {
            s.worker_handles.insert(h, worker_fp);
        })
        .await?;
        Ok(handle)
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Session API
    // ═══════════════════════════════════════════════════════════════════════

    /// Attach a new session with default `OperatorInfo` (`Automate`, no
    /// bridges/hooks). Shorthand for `attach_with(.., OperatorInfo::default())`.
    pub async fn attach(
        &self,
        operator_id: impl Into<String>,
        role: Role,
        ttl: Duration,
    ) -> Result<CapToken, EngineError> {
        self.attach_with(
            operator_id,
            role,
            ttl,
            crate::core::ctx::OperatorInfo::default(),
        )
        .await
    }

    // ═══════════════════════════════════════════════════════════════════════
    // BridgeRegistry API.
    // ═══════════════════════════════════════════════════════════════════════

    /// Register a `SeniorBridge` under a name. An existing entry with the
    /// same name is overwritten. On the persisted-session reattach path,
    /// the caller re-registers under the same ID beforehand and the
    /// bridge becomes effective again.
    pub async fn register_senior_bridge(
        &self,
        id: impl Into<String>,
        bridge: Arc<dyn SeniorBridge>,
    ) {
        self.inner
            .senior_bridges
            .write()
            .await
            .insert(id.into(), bridge);
    }

    /// Register a `SpawnHook` under a name. An existing entry with the
    /// same name is overwritten.
    pub async fn register_spawn_hook(&self, id: impl Into<String>, hook: Arc<dyn SpawnHook>) {
        self.inner.spawn_hooks.write().await.insert(id.into(), hook);
    }

    /// Register an `Operator` (a spawn-body backend) under a name. An
    /// existing entry with the same name is overwritten.
    /// `OperatorDelegateMiddleware` looks this up via `ctx` and, when
    /// `kind = MainAi` / `Composite`, bypasses `inner.spawn` and calls
    /// `operator.execute` instead.
    pub async fn register_operator(
        &self,
        id: impl Into<String>,
        operator: Arc<dyn crate::operator::Operator>,
    ) {
        self.inner
            .operators
            .write()
            .await
            .insert(id.into(), operator);
    }

    /// Unregister a `SeniorBridge` by name (e.g. on WebSocket disconnect
    /// or explicit teardown). A missing ID is a no-op.
    pub async fn unregister_senior_bridge(&self, id: &str) {
        self.inner.senior_bridges.write().await.remove(id);
    }

    /// Unregister a `SpawnHook` by name. A missing ID is a no-op.
    pub async fn unregister_spawn_hook(&self, id: &str) {
        self.inner.spawn_hooks.write().await.remove(id);
    }

    /// Unregister an `Operator` backend by name. A missing ID is a no-op.
    pub async fn unregister_operator(&self, id: &str) {
        self.inner.operators.write().await.remove(id);
    }

    /// Snapshot the list of registered `SpawnHook` IDs (for test
    /// observation and debugging).
    pub async fn list_spawn_hook_ids(&self) -> Vec<String> {
        self.inner
            .spawn_hooks
            .read()
            .await
            .keys()
            .cloned()
            .collect()
    }

    /// Snapshot the list of registered `SeniorBridge` IDs.
    pub async fn list_senior_bridge_ids(&self) -> Vec<String> {
        self.inner
            .senior_bridges
            .read()
            .await
            .keys()
            .cloned()
            .collect()
    }

    /// Snapshot the list of registered `Operator` IDs.
    pub async fn list_operator_ids(&self) -> Vec<String> {
        self.inner.operators.read().await.keys().cloned().collect()
    }

    /// Attach specifying IDs directly. The caller is expected to have
    /// pre-registered them via `register_senior_bridge` /
    /// `register_spawn_hook` / `register_operator`. This is the canonical
    /// path when persistence is in play.
    ///
    /// `kind` is the "Runtime Global" tier of the `OperatorKind` cascade
    /// (stored verbatim on `OperatorSession.operator_kind`): `Some(_)` is
    /// an explicit request (including `Some(OperatorKind::Automate)`) that
    /// outranks the BP-level tiers; `None` leaves it unspecified so the
    /// BP-level tiers / final default decide. See
    /// `crate::core::ctx::collapse_operator_kind`.
    #[allow(clippy::too_many_arguments)]
    pub async fn attach_with_ids(
        &self,
        operator_id: impl Into<String>,
        role: Role,
        ttl: Duration,
        kind: Option<OperatorKind>,
        bridge_id: Option<String>,
        hook_id: Option<String>,
        operator_backend_id: Option<String>,
        operator_kind_overrides: HashMap<String, OperatorKind>,
        bp_agent_kinds: HashMap<String, OperatorKind>,
        bp_global_kind: Option<OperatorKind>,
    ) -> Result<CapToken, EngineError> {
        let operator_id = operator_id.into();
        let token = self
            .inner
            .signer
            .session(operator_id.clone(), role, vec!["*".into()], ttl);
        let session_id = SessionId::new();
        let fp = token.fingerprint();
        let now = now_unix();
        let token_for_store = token.clone();

        self.with_state("attach_with_ids", |s| {
            s.tokens
                .insert(fp.clone(), CapTokenRecord::from_token(token_for_store));
            s.sessions.insert(
                session_id.clone(),
                OperatorSession {
                    id: session_id.clone(),
                    operator_id: operator_id.clone(),
                    role,
                    attached_at: now,
                    last_seen: now,
                    attached: true,
                    owned_task_ids: Vec::new(),
                    token_fp: fp.clone(),
                    operator_kind: kind,
                    runtime_agent_kinds: operator_kind_overrides,
                    bp_agent_kinds,
                    bp_global_kind,
                    bridge_id,
                    hook_id,
                    operator_backend_id,
                },
            );
            s.push_event(Event::SessionAttached {
                session_id: session_id.clone(),
                role,
            });
        })
        .await?;

        let _ = self
            .inner
            .event_tx
            .send(Event::SessionAttached { session_id, role });
        Ok(token)
    }

    /// Build an `OperatorInfo` by looking up the session's registered IDs
    /// on the `BridgeRegistry`, plus resolving the 4-tier `OperatorKind`
    /// cascade for `agent_name` via `crate::core::ctx::collapse_operator_kind`.
    /// Used when `dispatch_attempt` injects `Ctx`. An unresolved ID
    /// (nothing registered) is silently `None` — the bridge / hook simply
    /// does not fire and the default behaviour applies.
    async fn resolve_operator_info(
        &self,
        session: &OperatorSession,
        agent_name: &str,
    ) -> OperatorInfo {
        let senior_bridge = if let Some(id) = &session.bridge_id {
            self.inner.senior_bridges.read().await.get(id).cloned()
        } else {
            None
        };
        let spawn_hook = if let Some(id) = &session.hook_id {
            self.inner.spawn_hooks.read().await.get(id).cloned()
        } else {
            None
        };
        let operator = if let Some(id) = &session.operator_backend_id {
            self.inner.operators.read().await.get(id).cloned()
        } else {
            None
        };
        let runtime_agent = session.runtime_agent_kinds.get(agent_name).copied();
        // "Runtime Global" tier: `Some(_)` is always an explicit request
        // (see the field doc on `OperatorSession.operator_kind`).
        let runtime_global = session.operator_kind;
        let bp_agent = session.bp_agent_kinds.get(agent_name).copied();
        let bp_global = session.bp_global_kind;
        let kind = crate::core::ctx::collapse_operator_kind(
            runtime_agent,
            runtime_global,
            bp_agent,
            bp_global,
        );
        OperatorInfo {
            kind,
            id: session.operator_id.clone(),
            senior_bridge,
            spawn_hook,
            operator,
        }
    }

    /// Convenience attach that takes an `OperatorInfo` (three
    /// `Arc<dyn ...>` fields plus `kind`) **inline**.
    ///
    /// # Pipeline
    ///
    /// Each `Arc<dyn ...>` is auto-registered on the engine's registry
    /// under a synthetic ID (`br-<hex>` / `hk-<hex>` / `ob-<hex>`), and
    /// the session stores that synthetic ID. Subsequent `dispatch_attempt`
    /// calls rebuild the `Arc`s from those IDs via
    /// `resolve_operator_info`, and the three middlewares fire as usual.
    ///
    /// # ⚠ Non-persisted sessions only
    ///
    /// Because this API takes inline `Arc`s, the reattach path after
    /// session persistence cannot rebuild them — the synthetic IDs are
    /// not present in a freshly started process's registry. If you need
    /// persistence, use [`Self::attach_with_ids`] with `register_*` calls
    /// beforehand to go through **named IDs** instead.
    ///
    /// Handy for tests and short-lived in-process sessions. Production
    /// WebSocket callbacks and the like should prefer `attach_with_ids`
    /// as the canonical path.
    pub async fn attach_with(
        &self,
        operator_id: impl Into<String>,
        role: Role,
        ttl: Duration,
        operator_info: crate::core::ctx::OperatorInfo,
    ) -> Result<CapToken, EngineError> {
        let operator_id = operator_id.into();
        // The caller always hands in a fully-formed `OperatorInfo`
        // (including its `kind`), so it is stored as an explicit "Runtime
        // Global" tier request (`Some(kind)`) — this path never persists
        // BP-level tiers (both stay empty below), so `Some(kind)` resolves
        // to the same `kind` at dispatch either way; see
        // `OperatorSession.operator_kind` doc.
        let kind = operator_info.kind;
        // BridgeRegistry auto-register: when the caller hands in an
        // `Arc<dyn>` directly, register it under a synthesised ID (the inline
        // path aware of persistence). Callers who want to pre-register with a
        // named ID should use `register_senior_bridge` / `register_spawn_hook`
        // + `attach_with_ids`.
        let bridge_id = if let Some(bridge) = operator_info.senior_bridge.clone() {
            let id = format!("br-{}", crate::types::uid_hex(8));
            self.inner
                .senior_bridges
                .write()
                .await
                .insert(id.clone(), bridge);
            Some(id)
        } else {
            None
        };
        let hook_id = if let Some(hook) = operator_info.spawn_hook.clone() {
            let id = format!("hk-{}", crate::types::uid_hex(8));
            self.inner
                .spawn_hooks
                .write()
                .await
                .insert(id.clone(), hook);
            Some(id)
        } else {
            None
        };
        let operator_backend_id = if let Some(operator) = operator_info.operator.clone() {
            // `ob-` = operator-backend registry id. Renamed from `op-` in the
            // issue #11 prefix reconciliation: `op-` used to collide with the
            // WS operator sid shape (now unified into `S-<hex>` anyway), and a
            // shared prefix across two unrelated registries made log filtering
            // by prefix silently ambiguous.
            let id = format!("ob-{}", crate::types::uid_hex(8));
            self.inner
                .operators
                .write()
                .await
                .insert(id.clone(), operator);
            Some(id)
        } else {
            None
        };

        let token = self
            .inner
            .signer
            .session(operator_id.clone(), role, vec!["*".into()], ttl);
        let session_id = SessionId::new();
        let fp = token.fingerprint();
        let now = now_unix();
        let token_for_store = token.clone();

        self.with_state("attach_with", |s| {
            s.tokens
                .insert(fp.clone(), CapTokenRecord::from_token(token_for_store));
            s.sessions.insert(
                session_id.clone(),
                OperatorSession {
                    id: session_id.clone(),
                    operator_id,
                    role,
                    attached_at: now,
                    last_seen: now,
                    attached: true,
                    owned_task_ids: Vec::new(),
                    token_fp: fp.clone(),
                    operator_kind: Some(kind),
                    runtime_agent_kinds: HashMap::new(),
                    bp_agent_kinds: HashMap::new(),
                    bp_global_kind: None,
                    bridge_id,
                    hook_id,
                    operator_backend_id,
                },
            );
            s.push_event(Event::SessionAttached {
                session_id: session_id.clone(),
                role,
            });
        })
        .await?;

        let _ = self
            .inner
            .event_tx
            .send(Event::SessionAttached { session_id, role });
        Ok(token)
    }

    /// Mark the session bound to `token` as detached (`attached = false`).
    /// Tasks are left in place — a later `attach`/`attach_with_ids` call
    /// carrying the same registered bridge/hook IDs can pick them back up.
    pub async fn detach(&self, token: &CapToken) -> Result<(), EngineError> {
        self.verify_token(token, Verb::DetachSession).await?;
        let fp = token.fingerprint();
        self.with_state("detach", move |s| {
            let sid = s
                .sessions
                .iter()
                .find(|(_, sess)| sess.token_fp == fp)
                .map(|(id, _)| id.clone());
            if let Some(sid) = sid {
                if let Some(sess) = s.sessions.get_mut(&sid) {
                    sess.attached = false;
                }
                s.push_event(Event::SessionDetached {
                    session_id: sid.clone(),
                });
                let _ = sid;
            }
        })
        .await?;
        Ok(())
    }

    /// Refresh the session's `last_seen` timestamp and mark it `attached`.
    /// Called periodically by an attached client to avoid being flipped to
    /// detached by `start_detach_loop`.
    pub async fn heartbeat(&self, token: &CapToken) -> Result<(), EngineError> {
        self.verify_token(token, Verb::Heartbeat).await?;
        let now = now_unix();
        let fp = token.fingerprint();
        self.with_state("heartbeat", move |s| {
            if let Some(sess) = s.sessions.values_mut().find(|sess| sess.token_fp == fp) {
                sess.last_seen = now;
                sess.attached = true;
            }
        })
        .await?;
        Ok(())
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Task lifecycle
    // ═══════════════════════════════════════════════════════════════════════

    /// Create a new `TaskState` from `spec` and register its initial
    /// prompt. When the calling token is a Worker (i.e. this is a
    /// recursive spawn), the new task inherits `parent.spawn_depth + 1`
    /// and is rejected with `SpawnDepthExceeded` once `max_spawn_depth` is
    /// hit; an Operator-issued call starts at depth 0.
    pub async fn start_task(
        &self,
        token: &CapToken,
        spec: TaskSpec,
    ) -> Result<StepId, EngineError> {
        self.verify_token(token, Verb::StartTask).await?;
        let task_id = StepId::new();
        let initial_directive = spec.initial_directive.clone();
        let task_id_clone = task_id.clone();
        let fp = token.fingerprint();
        let max_depth = self.inner.cfg.max_spawn_depth;
        self.with_state("start_task", move |s| {
            // Recursive swarm depth gate (recursion guard):
            // Worker tokens carry CapTokenRecord.parent_task_id. Give the
            // child parent's spawn_depth + 1; if it exceeds `max`, raise an
            // error. Operator tokens (parent_task_id=None) start at depth 0.
            let parent_depth_opt = s
                .tokens
                .get(&fp)
                .and_then(|rec| rec.task_id.as_ref())
                .and_then(|tid| s.tasks.get(tid))
                .map(|t| t.spawn_depth);
            let depth = match parent_depth_opt {
                Some(d) => {
                    if d + 1 >= max_depth {
                        return Err(EngineError::SpawnDepthExceeded {
                            current: d + 1,
                            max: max_depth,
                        });
                    }
                    d + 1
                }
                None => 0,
            };

            let mut task = TaskState::new(task_id_clone.clone(), spec);
            task.spawn_depth = depth;
            s.tasks.insert(task_id_clone.clone(), task);
            s.prompts
                .insert((task_id_clone.clone(), 1), initial_directive);
            // Link to the owner session (only Operator tokens match; Worker tokens have no session).
            if let Some(sess) = s.sessions.values_mut().find(|sess| sess.token_fp == fp) {
                sess.owned_task_ids.push(task_id_clone.clone());
            }
            s.push_event(Event::TaskCreated {
                task_id: task_id_clone.clone(),
            });
            Ok::<(), EngineError>(())
        })
        .await??;
        let _ = self.inner.event_tx.send(Event::TaskCreated {
            task_id: task_id.clone(),
        });
        Ok(task_id)
    }

    /// Fetch a snapshot of `TaskState` for `task_id`, subject to the
    /// task-ownership gate (see `verify_token_for_task`).
    pub async fn read_task_state(
        &self,
        token: &CapToken,
        task_id: &StepId,
    ) -> Result<TaskState, EngineError> {
        self.verify_token_for_task(token, Verb::ReadTaskState, task_id)
            .await?;
        let task_id = task_id.clone();
        self.with_state("read_task_state", move |s| {
            s.tasks
                .get(&task_id)
                .cloned()
                .ok_or_else(|| EngineError::TaskNotFound(task_id.to_string()))
        })
        .await?
    }

    /// Mark `task_id` as `Cancelled` and wake any caller blocked in
    /// `poll_task` for it.
    pub async fn cancel_task(&self, token: &CapToken, task_id: &StepId) -> Result<(), EngineError> {
        self.verify_token_for_task(token, Verb::CancelTask, task_id)
            .await?;
        let tid = task_id.clone();
        self.with_state("cancel_task", move |s| {
            let task = s
                .tasks
                .get_mut(&tid)
                .ok_or_else(|| EngineError::TaskNotFound(tid.to_string()))?;
            task.status = TaskStatus::Cancelled;
            task.updated_at = now_unix();
            s.push_event(Event::TaskCancelled {
                task_id: tid.clone(),
            });
            Ok::<(), EngineError>(())
        })
        .await??;
        self.wake_task(task_id).await?;
        Ok(())
    }

    /// Dispatch a single attempt through the given `spawner`.
    ///
    /// The lock is only held for snapshot capture; the actual spawn and
    /// completion await happen outside the lock (R3 discipline).
    ///
    /// Sits on the Domain side of the Data / Domain split. The dispatch
    /// path itself does not touch big response bodies — those flow through
    /// the Data plane (`output_store` module + sink / input_inject
    /// `SpawnerLayer`s) around this method.
    ///
    /// The caller does the compile plus `service::linker::link` and
    /// carries the same stack through each dispatch. Because the spawner
    /// is passed per-request rather than looked up from engine-global
    /// state, parallel requests against a single `Engine` instance
    /// (different Blueprints, different spawners) do not race.
    ///
    /// `run_id`, when `Some` (issue #13 run_id propagation —
    /// `EngineDispatcher` threads it in from its `RunContext`), is
    /// inserted into `Ctx.meta.runtime["run_id"]` (a plain JSON string)
    /// alongside `worker_handle`, so `Operator::execute` implementations
    /// (e.g. `WSOperatorSession`) can read it back and surface it to the
    /// worker (Spawn directive / prompt). `None` (every pre-existing
    /// caller / test) omits the key entirely — unchanged behavior.
    pub async fn dispatch_attempt_with(
        &self,
        token: &CapToken,
        task_id: &StepId,
        spawner: &Arc<dyn SpawnerAdapter>,
        run_id: Option<&RunId>,
    ) -> Result<DispatchOutcome, EngineError> {
        self.verify_token(token, Verb::DispatchAttempt).await?;
        let task_id = task_id.clone();

        // 1) Under the lock: increment the attempt number, mark Running, snapshot the
        //    prompt, and pull `operator_info` from the session so we can inject it into Ctx.
        let fp = token.fingerprint();
        let tid_for_prep = task_id.clone();
        let (attempt, agent, session_snapshot, step_ctx) = self
            .with_state("dispatch.prep", move |s| {
                let task = s
                    .tasks
                    .get_mut(&tid_for_prep)
                    .ok_or_else(|| EngineError::TaskNotFound(tid_for_prep.to_string()))?;
                task.attempt += 1;
                task.status = TaskStatus::Running;
                task.updated_at = now_unix();
                // The spawner pulls the prompt via engine.fetch_prompt. In prep,
                // if the prompts table has no entry for this attempt yet,
                // fall back and insert `initial_directive` so the subsequent
                // fetch_prompt succeeds.
                let attempt = task.attempt;
                let initial = task.spec.initial_directive.clone();
                s.prompts
                    .entry((tid_for_prep.clone(), attempt))
                    .or_insert(initial);
                let task = s
                    .tasks
                    .get(&tid_for_prep)
                    .ok_or_else(|| EngineError::TaskNotFound(tid_for_prep.to_string()))?;
                let agent = task.spec.agent.clone();
                // GH #21 Phase 2: re-read `TaskSpec.step_ctx` on EVERY
                // attempt (not cached once at start_task) so retries and
                // Run-rekicks all carry the Step tier through to Ctx —
                // see TaskSpec.step_ctx's doc.
                let step_ctx = task.spec.step_ctx.clone();
                // Session snapshot (looked up by token nonce). When no session
                // exists (worker token invoked directly / test injection), fall
                // back to None → default OperatorInfo.
                let sess_clone = s
                    .sessions
                    .values()
                    .find(|sess| sess.token_fp == fp)
                    .cloned();
                Ok::<_, EngineError>((attempt, agent, sess_clone, step_ctx))
            })
            .await??;
        // BridgeRegistry lookup + per-agent OperatorKind cascade.
        let operator_info = match session_snapshot {
            Some(sess) => self.resolve_operator_info(&sess, &agent).await,
            None => OperatorInfo::default(),
        };

        // 2) Outside the lock: worker token mint + spawn.
        //
        // Session-style mint (max_uses=None). Within one attempt the worker is
        // expected to hit `verify_token + fetch_prompt + fetch_data + post_result`
        // multiple times in order, so `one_time` would exhaust the token on the
        // very first verb. Capability is guarded by (a) the role × verb gate and
        // (b) the short TTL (1800s).
        let worker_token = self.inner.signer.session(
            format!("worker-of-{task_id}"),
            Role::Worker,
            vec!["*".into()],
            Duration::from_secs(1800),
        );
        let worker_fp = worker_token.fingerprint();
        let task_id_for_worker = task_id.clone();
        let worker_token_for_store = worker_token.clone();
        self.with_state("dispatch.mint_worker", move |s| {
            s.tokens.insert(
                worker_fp,
                CapTokenRecord::from_worker_token(worker_token_for_store, task_id_for_worker),
            );
        })
        .await?;

        // Mint a short handle (`wh-XXXXXXXX`) and register it in worker_handles.
        // Used by the simplified Bearer path for SubAgents (short-handle form
        // avoids base64 copy-paste incidents).
        let worker_handle = self.mint_worker_handle(worker_token.fingerprint()).await?;

        let mut ctx = Ctx::new(task_id.clone(), attempt, agent.clone());
        ctx.operator = operator_info; // activates MainAIMiddleware / Senior bridge
        ctx.meta
            .runtime
            .insert("worker_handle".to_string(), Value::String(worker_handle));
        if let Some(rid) = run_id {
            ctx.meta
                .runtime
                .insert(RUN_ID_KEY.to_string(), Value::String(rid.to_string()));
        }
        // GH #21 Phase 2: the Step tier's resolved context bundle (from
        // `TaskSpec.step_ctx`, re-read every attempt above) — consumed by
        // `AgentContextMiddleware`, which unpacks its keys ahead of the
        // Agent / BP-global tiers.
        if let Some(step_ctx) = step_ctx {
            ctx.meta.runtime.insert(STEP_CTX_KEY.to_string(), step_ctx);
        }

        let worker = spawner
            .spawn(self, &ctx, task_id.clone(), attempt, worker_token)
            .await
            .map_err(|e| EngineError::DispatchFailed(e.to_string()))?;

        // 3) Outside the lock: await worker.join() (signal-only). WorkerError is
        //    stringified. The value is fetched via output_tail (sink path).
        let signal_result: Result<(), String> = worker.join().await.map_err(|e| e.to_string());

        // Pull the last Final from output_tail and use it as the value. GH
        // #36 ST1 (named multi-part worker output): also fold every
        // `Artifact` the WORKER ITSELF staged on the same tail (via
        // `stage_worker_artifact_trusted` / `POST /v1/worker/artifact`)
        // into a `"parts"` object keyed by name — event order,
        // last-write-wins per name (a name staged twice overwrites,
        // mirroring `HashMap`/`Map` insert semantics, not an accumulating
        // list). `worker_artifact_names_for` is the allowlist that scopes
        // this to the worker's own opt-in parts — an `Artifact` some OTHER
        // producer appended to this same tail (e.g.
        // `AfterRunAuditMiddleware`'s `"audit:<step_ref>"` sidecar finding)
        // is left untouched (see `fold_final_and_parts`'s doc). When at
        // least one part was staged, the BP-chain value becomes `{"out":
        // <final value>, "parts": {...}}`; zero parts staged (the
        // pre-GH-#36 case, and every non-opt-in step) leaves the value
        // exactly the plain `Final` value, byte-identical to before this
        // change.
        let value_ok: Result<(Value, bool), String> = match signal_result {
            Ok(()) => {
                let tail = self.output_tail(&task_id, attempt).await;
                let staged_names = self.worker_artifact_names_for(&task_id, attempt).await;
                let mode = self.fold_parse_mode_for(&task_id, attempt).await;
                fold_final_and_parts(&tail, &staged_names, mode)
                    .ok_or_else(|| "no Final in output_tail".to_string())
            }
            Err(msg) => Err(msg),
        };

        // 4) Under the lock: apply (split the borrow scope so push_event and task mut can co-exist).
        let outcome = self
            .with_state("dispatch.apply", |s| {
                if !s.tasks.contains_key(&task_id) {
                    return Err(EngineError::TaskNotFound(task_id.to_string()));
                }
                match value_ok {
                    Ok((value, ok)) => {
                        // GH #76 Skip tier: a Final with ok=true carrying the
                        // skip-marker sentinel is a Skip tier completion,
                        // not an ordinary Pass. TaskStatus stays `Pass`
                        // (the worker itself completed successfully);
                        // the Skip signal rides on DispatchOutcome so
                        // EngineDispatcher::dispatch can route it to
                        // the flow-continuation-without-binding-write
                        // sentinel path.
                        let skip_inner = if ok { unwrap_skip_marker(&value) } else { None };
                        let pass = ok;
                        {
                            let task = s.tasks.get_mut(&task_id).unwrap();
                            task.last_result = Some(value.clone());
                            task.updated_at = now_unix();
                            task.status = if pass {
                                TaskStatus::Pass
                            } else {
                                TaskStatus::Blocked
                            };
                        }
                        s.push_event(Event::TaskAttemptCompleted {
                            task_id: task_id.clone(),
                            attempt,
                            result: value.clone(),
                        });
                        if let Some(inner) = skip_inner {
                            s.push_event(Event::TaskPass {
                                task_id: task_id.clone(),
                                result: value.clone(),
                            });
                            Ok::<_, EngineError>(DispatchOutcome::Skip(inner))
                        } else if pass {
                            s.push_event(Event::TaskPass {
                                task_id: task_id.clone(),
                                result: value.clone(),
                            });
                            Ok::<_, EngineError>(DispatchOutcome::Pass(value))
                        } else {
                            s.push_event(Event::TaskBlocked {
                                task_id: task_id.clone(),
                                result: value.clone(),
                            });
                            Ok(DispatchOutcome::Blocked(value))
                        }
                    }
                    Err(msg) => {
                        let task = s.tasks.get_mut(&task_id).unwrap();
                        task.status = TaskStatus::Blocked;
                        task.updated_at = now_unix();
                        Err(EngineError::DispatchFailed(msg))
                    }
                }
            })
            .await??;

        // event broadcast (outside the lock — push_event feeds the in-memory tail; broadcast is a separate path).
        let _ = self.inner.event_tx.send(Event::TaskAttemptCompleted {
            task_id: task_id.clone(),
            attempt,
            result: match &outcome {
                DispatchOutcome::Pass(v)
                | DispatchOutcome::Blocked(v)
                | DispatchOutcome::Skip(v) => v.clone(),
                _ => Value::Null,
            },
        });

        // Wake any callers waiting in poll_task.
        self.wake_task(&task_id).await?;

        Ok(outcome)
    }

    /// Dispatch a single attempt, opt-in to the replay-log Core primitive
    /// ([`crate::store::replay`]) via `run_ctx`.
    ///
    /// This is the [`Self::dispatch_attempt_with`] sibling used by callers
    /// that carry a `RunContext` with `replay_store` / `replay_cursor`
    /// populated. Behavior versus the plain `dispatch_attempt_with`:
    ///
    /// - **`run_ctx.replay_cursor` is `Some` AND the cursor has a matching
    ///   `(step_ref, input_hash, occurrence)` row** — the stored value is
    ///   returned verbatim as `DispatchOutcome::Pass(v)`; the `Adapter`
    ///   (spawner + worker) is never touched. The task's `attempt` is
    ///   still bumped and `TaskStatus` set to `Pass`, so downstream state
    ///   (`task.last_result`, `TaskAttemptCompleted` / `TaskPass` events,
    ///   `wake_task`) fires the same way an ordinary Pass would.
    /// - **Miss (or `replay_cursor: None`)** — the ordinary spawn path
    ///   runs. When `run_ctx.replay_store` is `Some` AND the outcome is
    ///   `Pass`, one `ReplayEntry` is appended carrying the whole `Ctx`
    ///   snapshot (with `operator` dropped by `#[serde(skip)]`) plus the
    ///   `step_output` value. `Blocked` / `Err` outcomes are never
    ///   logged — a partial-failure row would poison the replay path
    ///   after a subsequent successful retry.
    ///
    /// `run_ctx: None` collapses to the same behavior as
    /// `dispatch_attempt_with(token, task_id, spawner, None)` — no run
    /// tracing, no replay.
    pub async fn dispatch_attempt_with_run_ctx(
        &self,
        token: &CapToken,
        task_id: &StepId,
        spawner: &Arc<dyn SpawnerAdapter>,
        run_ctx: Option<&RunContext>,
    ) -> Result<DispatchOutcome, EngineError> {
        self.verify_token(token, Verb::DispatchAttempt).await?;
        let task_id = task_id.clone();

        // 1) Under the lock: prep (bump attempt, snapshot agent/directive).
        let fp = token.fingerprint();
        let tid_for_prep = task_id.clone();
        let (attempt, agent, session_snapshot, step_ctx, initial_directive) = self
            .with_state("dispatch_run_ctx.prep", move |s| {
                let task = s
                    .tasks
                    .get_mut(&tid_for_prep)
                    .ok_or_else(|| EngineError::TaskNotFound(tid_for_prep.to_string()))?;
                task.attempt += 1;
                task.status = TaskStatus::Running;
                task.updated_at = now_unix();
                let attempt = task.attempt;
                let initial = task.spec.initial_directive.clone();
                s.prompts
                    .entry((tid_for_prep.clone(), attempt))
                    .or_insert(initial.clone());
                let task = s
                    .tasks
                    .get(&tid_for_prep)
                    .ok_or_else(|| EngineError::TaskNotFound(tid_for_prep.to_string()))?;
                let agent = task.spec.agent.clone();
                let step_ctx = task.spec.step_ctx.clone();
                let sess_clone = s
                    .sessions
                    .values()
                    .find(|sess| sess.token_fp == fp)
                    .cloned();
                Ok::<_, EngineError>((attempt, agent, sess_clone, step_ctx, initial))
            })
            .await??;

        let operator_info = match session_snapshot {
            Some(sess) => self.resolve_operator_info(&sess, &agent).await,
            None => OperatorInfo::default(),
        };

        // 2) Compute the replay key from step_ref (= agent) + hashed input.
        //    Occurrence comes from the cursor's per-key counter (bumped
        //    once per dispatch, so a loop that re-visits the same step
        //    with the same input gets 0, 1, 2, … distinct rows).
        let step_ref = agent.clone();
        let input_hash = match run_ctx.and_then(|rc| rc.binding_digests.get(&step_ref)) {
            Some(binding_digest) => hash_input_value(&serde_json::json!({
                "input": initial_directive,
                "binding_digest": binding_digest,
            })),
            None => hash_input_value(&initial_directive),
        };
        let (replay_hit_value, occurrence) = if let Some(rc) = run_ctx {
            if let Some(cursor) = &rc.replay_cursor {
                let mut guard = cursor.lock().expect("replay cursor mutex poisoned");
                let occ = guard.next_occurrence(&step_ref, &input_hash);
                let hit = guard.find(&step_ref, &input_hash, occ);
                (hit, occ)
            } else {
                (None, 0)
            }
        } else {
            (None, 0)
        };

        // 3) Build the Ctx that (a) either the spawner will see on a miss,
        //    or (b) we log alongside the replay row.
        let mut ctx = Ctx::new(task_id.clone(), attempt, agent.clone());
        ctx.operator = operator_info;
        if let Some(rc) = run_ctx {
            ctx.meta
                .runtime
                .insert(RUN_ID_KEY.to_string(), Value::String(rc.run_id.to_string()));
        }
        if let Some(step_ctx) = step_ctx {
            ctx.meta.runtime.insert(STEP_CTX_KEY.to_string(), step_ctx);
        }

        // 4) Replay-hit shortcut: skip the spawn+join, return stored value.
        let was_replay_hit = replay_hit_value.is_some();
        let value_ok: Result<(Value, bool), String> = if let Some(stored) = replay_hit_value {
            tracing::info!(
                task_id = %task_id,
                step_ref = %step_ref,
                occurrence = occurrence,
                "replayed from log; worker dispatch skipped"
            );
            Ok((stored, true))
        } else {
            // 5) Ordinary spawn path — mint a worker token+handle, run the
            //    spawner, join, and pull the last Final from output_tail.
            let worker_token = self.inner.signer.session(
                format!("worker-of-{task_id}"),
                Role::Worker,
                vec!["*".into()],
                Duration::from_secs(1800),
            );
            let worker_fp = worker_token.fingerprint();
            let task_id_for_worker = task_id.clone();
            let worker_token_for_store = worker_token.clone();
            self.with_state("dispatch_run_ctx.mint_worker", move |s| {
                s.tokens.insert(
                    worker_fp,
                    CapTokenRecord::from_worker_token(worker_token_for_store, task_id_for_worker),
                );
            })
            .await?;
            let worker_handle = self.mint_worker_handle(worker_token.fingerprint()).await?;
            ctx.meta
                .runtime
                .insert("worker_handle".to_string(), Value::String(worker_handle));

            let worker = spawner
                .spawn(self, &ctx, task_id.clone(), attempt, worker_token)
                .await
                .map_err(|e| EngineError::DispatchFailed(e.to_string()))?;
            let signal_result: Result<(), String> = worker.join().await.map_err(|e| e.to_string());
            match signal_result {
                Ok(()) => {
                    let tail = self.output_tail(&task_id, attempt).await;
                    let staged_names = self.worker_artifact_names_for(&task_id, attempt).await;
                    let mode = self.fold_parse_mode_for(&task_id, attempt).await;
                    fold_final_and_parts(&tail, &staged_names, mode)
                        .ok_or_else(|| "no Final in output_tail".to_string())
                }
                Err(msg) => Err(msg),
            }
        };

        // 6) Apply — mirrors `dispatch_attempt_with`'s apply arm exactly
        //    (task.last_result / status update + TaskAttemptCompleted /
        //    TaskPass / TaskBlocked events).
        let outcome = self
            .with_state("dispatch_run_ctx.apply", |s| {
                if !s.tasks.contains_key(&task_id) {
                    return Err(EngineError::TaskNotFound(task_id.to_string()));
                }
                match value_ok {
                    Ok((value, ok)) => {
                        // GH #76 Skip tier: Skip tier detection — same shape
                        // as the sibling `dispatch_attempt_with` apply
                        // arm above. See that arm's comment for the
                        // TaskStatus / Event / DispatchOutcome contract.
                        let skip_inner = if ok { unwrap_skip_marker(&value) } else { None };
                        let pass = ok;
                        {
                            let task = s.tasks.get_mut(&task_id).unwrap();
                            task.last_result = Some(value.clone());
                            task.updated_at = now_unix();
                            task.status = if pass {
                                TaskStatus::Pass
                            } else {
                                TaskStatus::Blocked
                            };
                        }
                        s.push_event(Event::TaskAttemptCompleted {
                            task_id: task_id.clone(),
                            attempt,
                            result: value.clone(),
                        });
                        if let Some(inner) = skip_inner {
                            s.push_event(Event::TaskPass {
                                task_id: task_id.clone(),
                                result: value.clone(),
                            });
                            Ok::<_, EngineError>(DispatchOutcome::Skip(inner))
                        } else if pass {
                            s.push_event(Event::TaskPass {
                                task_id: task_id.clone(),
                                result: value.clone(),
                            });
                            Ok::<_, EngineError>(DispatchOutcome::Pass(value))
                        } else {
                            s.push_event(Event::TaskBlocked {
                                task_id: task_id.clone(),
                                result: value.clone(),
                            });
                            Ok(DispatchOutcome::Blocked(value))
                        }
                    }
                    Err(msg) => {
                        let task = s.tasks.get_mut(&task_id).unwrap();
                        task.status = TaskStatus::Blocked;
                        task.updated_at = now_unix();
                        Err(EngineError::DispatchFailed(msg))
                    }
                }
            })
            .await??;

        // 7) On MISS + Pass + replay_store present, append a replay row.
        //    Replay-HIT rows are already logged from the original run and
        //    must never be double-logged (Core primitive contract). A
        //    secondary-persistence failure here (`tracing::warn!` +
        //    swallow) matches the `run_ctx.run_store.append_step_entry`
        //    convention in `EngineDispatcher::dispatch`: it must not mask
        //    the primary dispatch outcome the caller already has in hand.
        if !was_replay_hit {
            if let (Some(rc), DispatchOutcome::Pass(v)) = (run_ctx, &outcome) {
                if let Some(store) = &rc.replay_store {
                    match ReplayEntry::from_completion(
                        rc.run_id.clone(),
                        step_ref.clone(),
                        input_hash.clone(),
                        occurrence,
                        &ctx,
                        v,
                    ) {
                        Ok(entry) => {
                            if let Err(e) = store.append(entry).await {
                                tracing::warn!(
                                    run_id = %rc.run_id,
                                    step_ref = %step_ref,
                                    occurrence = occurrence,
                                    error = %e,
                                    "dispatch_attempt_with_run_ctx: replay_store.append failed"
                                );
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                run_id = %rc.run_id,
                                step_ref = %step_ref,
                                occurrence = occurrence,
                                error = %e,
                                "dispatch_attempt_with_run_ctx: ReplayEntry encode failed"
                            );
                        }
                    }
                }
            }
        }

        let _ = self.inner.event_tx.send(Event::TaskAttemptCompleted {
            task_id: task_id.clone(),
            attempt,
            result: match &outcome {
                DispatchOutcome::Pass(v)
                | DispatchOutcome::Blocked(v)
                | DispatchOutcome::Skip(v) => v.clone(),
                _ => Value::Null,
            },
        });

        self.wake_task(&task_id).await?;

        Ok(outcome)
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Worker-side API (= prompt / data fetch + result post)
    // ═══════════════════════════════════════════════════════════════════════

    /// Fetch the directive/prompt `Value` for `task_id`'s current attempt.
    /// Falls back to `initial_directive` when no prompt has been recorded
    /// yet for that attempt. Returns the `Value` end-to-end (issue #18);
    /// the render down to `String` happens only at the two consumer
    /// boundaries — the Worker HTTP path (`fetch_worker_payload*` →
    /// `WorkerPayload.prompt: String`) and the WS Spawn frame text
    /// render (`operator_ws::session`).
    pub async fn fetch_prompt(
        &self,
        token: &CapToken,
        task_id: &StepId,
    ) -> Result<Value, EngineError> {
        self.verify_token_for_task(token, Verb::FetchPrompt, task_id)
            .await?;
        let task_id = task_id.clone();
        self.with_state("fetch_prompt", move |s| {
            let task = s
                .tasks
                .get(&task_id)
                .ok_or_else(|| EngineError::TaskNotFound(task_id.to_string()))?;
            s.prompts
                .get(&(task_id.clone(), task.attempt.max(1)))
                .cloned()
                .ok_or_else(|| {
                    EngineError::ResourceNotFound(format!(
                        "prompt({}, attempt={})",
                        task_id, task.attempt
                    ))
                })
        })
        .await?
    }

    /// Combined fetch for `HTTP /v1/worker/prompt`: returns `prompt` +
    /// (optional) `system` + `agent` + `attempt` in a single round trip.
    /// The verb gate reuses `FetchPrompt` — same semantics as "the worker
    /// pulls its task input".
    ///
    /// `system` is the value written by `OperatorSpawner::spawn` through
    /// `bake_worker_system_prompt` when it ran; otherwise `None` (no
    /// profile present, or the bake never happened).
    pub async fn fetch_worker_payload(
        &self,
        token: &CapToken,
        task_id: &StepId,
    ) -> Result<crate::types::WorkerPayload, EngineError> {
        self.verify_token_for_task(token, Verb::FetchPrompt, task_id)
            .await?;
        let task_id_clone = task_id.clone();
        let mut payload = self
            .with_state("fetch_worker_payload", move |s| {
                let task = s
                    .tasks
                    .get(&task_id_clone)
                    .ok_or_else(|| EngineError::TaskNotFound(task_id_clone.to_string()))?;
                let attempt = task.attempt.max(1);
                let prompt = s
                    .prompts
                    .get(&(task_id_clone.clone(), attempt))
                    .cloned()
                    .ok_or_else(|| {
                        EngineError::ResourceNotFound(format!(
                            "prompt({}, attempt={})",
                            task_id_clone, attempt
                        ))
                    })?;
                let system = s
                    .systems
                    .get(&(task_id_clone.clone(), attempt))
                    .cloned()
                    .unwrap_or(None);
                let agent = task.spec.agent.clone();
                let context = s
                    .agent_ctx
                    .get(&(task_id_clone.clone(), attempt))
                    .map(|e| e.view.clone());
                Ok::<_, EngineError>(crate::types::WorkerPayload {
                    task_id: task_id_clone.clone(),
                    attempt,
                    agent,
                    prompt: render_directive_to_string(&prompt),
                    system,
                    context,
                    system_ref: None,
                })
            })
            .await??;
        self.apply_system_ref_threshold(&mut payload).await?;
        Ok(payload)
    }

    /// Fetch a worker payload via a short handle. Skips token verification
    /// and returns `prompt` + `system` + `agent` + `attempt` in a thin
    /// path. The caller is expected to have already resolved `task_id`
    /// via `task_id_from_handle` — the handle's presence in
    /// `worker_handles` means it was minted server-side and is therefore
    /// trusted.
    pub async fn fetch_worker_payload_trusted(
        &self,
        task_id: &StepId,
    ) -> Result<crate::types::WorkerPayload, EngineError> {
        let task_id_clone = task_id.clone();
        let mut payload = self
            .with_state("fetch_worker_payload_trusted", move |s| {
                let task = s
                    .tasks
                    .get(&task_id_clone)
                    .ok_or_else(|| EngineError::TaskNotFound(task_id_clone.to_string()))?;
                let attempt = task.attempt.max(1);
                let prompt = s
                    .prompts
                    .get(&(task_id_clone.clone(), attempt))
                    .cloned()
                    .ok_or_else(|| {
                        EngineError::ResourceNotFound(format!(
                            "prompt({}, attempt={})",
                            task_id_clone, attempt
                        ))
                    })?;
                let system = s
                    .systems
                    .get(&(task_id_clone.clone(), attempt))
                    .cloned()
                    .unwrap_or(None);
                let agent = task.spec.agent.clone();
                let context = s
                    .agent_ctx
                    .get(&(task_id_clone.clone(), attempt))
                    .map(|e| e.view.clone());
                Ok::<_, EngineError>(crate::types::WorkerPayload {
                    task_id: task_id_clone.clone(),
                    attempt,
                    agent,
                    prompt: render_directive_to_string(&prompt),
                    system,
                    context,
                    system_ref: None,
                })
            })
            .await??;
        self.apply_system_ref_threshold(&mut payload).await?;
        Ok(payload)
    }

    /// GH #31: shared threshold-branch tail for
    /// [`Self::fetch_worker_payload`] / [`Self::fetch_worker_payload_trusted`].
    /// Both build a raw `WorkerPayload` inside `with_state` with `system`
    /// populated as before and `system_ref: None`; this runs *outside* any
    /// lock (R3 — `SystemRefMode::File`'s `tokio::fs` write is a genuine
    /// `.await`, which `with_state`'s sync-closure contract forbids inside
    /// the lock) and rewrites `payload.system` / `payload.system_ref` in
    /// place per `SystemRefConfig.threshold_bytes`: over-threshold clears
    /// `system` and populates `system_ref`; at-or-under-threshold leaves
    /// `system` as-is and `system_ref` stays `None`. A no-op when
    /// `payload.system` is already `None` (no `system_prompt` was baked).
    async fn apply_system_ref_threshold(
        &self,
        payload: &mut crate::types::WorkerPayload,
    ) -> Result<(), EngineError> {
        let Some(rendered) = payload.system.take() else {
            return Ok(());
        };
        let cfg = self.cfg().system_ref.clone();
        if rendered.len() <= cfg.threshold_bytes {
            payload.system = Some(rendered);
            return Ok(());
        }
        use sha2::Digest;
        let size_bytes = rendered.len() as u64;
        let sha256 = hex::encode(sha2::Sha256::digest(rendered.as_bytes()));
        let task_id = &payload.task_id;
        let attempt = payload.attempt;
        let system_ref = match cfg.mode {
            crate::types::SystemRefMode::Http => crate::types::SystemRef {
                // The engine has no knowledge of scheme/host here — see
                // `SystemRefMode::Http`'s doc for who fills that in.
                uri: format!("/v1/worker/prompt/system?task_id={task_id}&attempt={attempt}"),
                sha256,
                size_bytes,
                mode: crate::types::SystemRefMode::Http,
            },
            crate::types::SystemRefMode::File => {
                tokio::fs::create_dir_all(&cfg.store_dir).await?;
                let path = cfg.store_dir.join(format!("{task_id}-{attempt}.md"));
                tokio::fs::write(&path, rendered.as_bytes()).await?;
                crate::types::SystemRef {
                    uri: format!("file://{}", path.display()),
                    sha256,
                    size_bytes,
                    mode: crate::types::SystemRefMode::File,
                }
            }
        };
        payload.system = None;
        payload.system_ref = Some(system_ref);
        Ok(())
    }

    /// GH #83: unconditionally materialize the baked system prompt for
    /// `(task_id, attempt)` to a file and return its path — the value
    /// source of the `{system_file}` placeholder in a `SubprocessDef`
    /// template. Unlike [`Self::apply_system_ref_threshold`] (whose
    /// `SystemRefMode::File` write only fires over
    /// `SystemRefConfig.threshold_bytes`, a behavior this helper does NOT
    /// touch), a template that names `{system_file}` needs a real path
    /// regardless of size, so the write here is unconditional. Reuses the
    /// same store dir and `{task_id}-{attempt}.md` naming as the File
    /// mode, so both paths converge on one on-disk identity per attempt.
    ///
    /// `Ok(None)` = no system prompt was baked for this attempt (the
    /// caller decides whether that is fail-loud — the Subprocess spawn
    /// path treats a `{system_file}` reference without a baked system as
    /// a `SpawnError`).
    pub async fn materialize_system_file(
        &self,
        task_id: &StepId,
        attempt: u32,
    ) -> Result<Option<std::path::PathBuf>, EngineError> {
        let key = (task_id.clone(), attempt);
        let rendered = self
            .with_state("materialize_system_file", move |s| {
                s.systems.get(&key).cloned().unwrap_or(None)
            })
            .await?;
        let Some(rendered) = rendered else {
            return Ok(None);
        };
        let cfg = self.cfg().system_ref.clone();
        tokio::fs::create_dir_all(&cfg.store_dir).await?;
        let path = cfg.store_dir.join(format!("{task_id}-{attempt}.md"));
        tokio::fs::write(&path, rendered.as_bytes()).await?;
        Ok(Some(path))
    }

    /// Returns the effective [`mlua_swarm_schema::ContextPolicy`]
    /// `AgentContextMiddleware` resolved and snapshotted for `(task_id,
    /// attempt)` at spawn time (the same policy already applied to that
    /// key's `EngineState.agent_ctx` entry's `.view`, GH #23 fold).
    /// Pass-all (`ContextPolicy::default()`) when no entry exists — either
    /// a pre-ST5 spawn, or a spawner stack that never layered
    /// `AgentContextMiddleware` (fail-open, mirroring [`Self::output_tail`]'s
    /// "no entry = empty default" convention).
    ///
    /// `crates/mlua-swarm-server/src/worker.rs`'s `GET /v1/worker/prompt`
    /// handler reads this back to filter `WorkerPayload.context.steps` via
    /// `ContextPolicy::allows_step`, without re-deriving the policy from
    /// the Blueprint at fetch time (`projection-adapter` ST5).
    pub async fn context_policy_for(
        &self,
        task_id: &StepId,
        attempt: u32,
    ) -> mlua_swarm_schema::ContextPolicy {
        let key = (task_id.clone(), attempt);
        self.with_state("context_policy_for", move |s| {
            s.agent_ctx
                .get(&key)
                .map(|e| e.policy.clone())
                .unwrap_or_default()
        })
        .await
        .unwrap_or_default()
    }

    /// GH #23: returns the Blueprint-wide
    /// [`crate::core::step_naming::StepNaming`] table snapshotted for
    /// `task_id` (the same `Arc` `crate::blueprint::EngineDispatcher::dispatch`
    /// stashed into `EngineState.step_namings` at dispatch time —
    /// `Self::start_task`'s `StepId`, not the `TaskId` work item). `None`
    /// when no entry exists — either the dispatcher was never given a
    /// `StepNaming` (`EngineDispatcher::with_step_naming` not called) or
    /// the lock could not be acquired; callers are expected to fall back
    /// to the pre-GH-#23 runtime union rule in that case (subtask-2/3
    /// consumers).
    pub async fn step_naming_for(
        &self,
        task_id: &StepId,
    ) -> Option<Arc<crate::core::step_naming::StepNaming>> {
        let key = task_id.clone();
        self.with_state("step_naming_for", move |s| {
            s.step_namings.get(&key).cloned()
        })
        .await
        .ok()
        .flatten()
    }

    /// GH #27 (follow-up to #23): returns the Blueprint-wide
    /// [`crate::core::projection_placement::ProjectionPlacement`] resolver
    /// snapshotted for `task_id` (the same `Arc`
    /// `crate::blueprint::EngineDispatcher::dispatch` stashed into
    /// `EngineState.projection_placements` at dispatch time — mirroring
    /// [`Self::step_naming_for`]'s contract exactly). `None` when no entry
    /// exists — either the dispatcher was never given a
    /// `ProjectionPlacement` (`EngineDispatcher::with_projection_placement`
    /// not called) or the lock could not be acquired; callers are expected
    /// to fall back to `ProjectionPlacement::default()` (byte-compat with
    /// the pre-#27 hardcoded layout) in that case.
    pub async fn projection_placement_for(
        &self,
        task_id: &StepId,
    ) -> Option<Arc<crate::core::projection_placement::ProjectionPlacement>> {
        let key = task_id.clone();
        self.with_state("projection_placement_for", move |s| {
            s.projection_placements.get(&key).cloned()
        })
        .await
        .ok()
        .flatten()
    }

    /// Record normalized per-attempt worker stats reported by a worker
    /// boundary (spawner fold site / result captor / `POST
    /// /v1/worker/submit`). Last-write-wins per `(task_id, attempt)`.
    /// Best-effort: a state-lock failure is logged and swallowed —
    /// stats are observational and must never fail the attempt that
    /// produced them. Drained by [`Self::take_worker_stats`] at the
    /// dispatcher's outcome fold.
    pub async fn record_worker_stats(
        &self,
        task_id: &StepId,
        attempt: u32,
        stats: crate::store::trace::WorkerStats,
    ) {
        if stats.is_empty() {
            return;
        }
        let key = (task_id.clone(), attempt);
        if let Err(e) = self
            .with_state("record_worker_stats", move |s| {
                s.worker_stats.insert(key, stats);
            })
            .await
        {
            tracing::warn!(
                task_id = %task_id,
                attempt,
                error = %e,
                "record_worker_stats failed (swallowed — stats are observational)"
            );
        }
    }

    /// Drain every recorded worker-stats entry for `task_id`, returning
    /// the highest-attempt one (the attempt whose outcome the dispatcher
    /// is folding). Removing ALL of the task's entries — not just the
    /// returned one — keeps retries from leaking earlier attempts into
    /// `EngineState` for the process lifetime.
    pub async fn take_worker_stats(
        &self,
        task_id: &StepId,
    ) -> Option<(u32, crate::store::trace::WorkerStats)> {
        let key_task = task_id.clone();
        self.with_state("take_worker_stats", move |s| {
            let attempts: Vec<u32> = s
                .worker_stats
                .keys()
                .filter(|(tid, _)| *tid == key_task)
                .map(|(_, a)| *a)
                .collect();
            let mut best: Option<(u32, crate::store::trace::WorkerStats)> = None;
            for attempt in attempts {
                if let Some(stats) = s.worker_stats.remove(&(key_task.clone(), attempt)) {
                    if best.as_ref().map(|(a, _)| attempt >= *a).unwrap_or(true) {
                        best = Some((attempt, stats));
                    }
                }
            }
            best
        })
        .await
        .ok()
        .flatten()
    }

    /// Returns the [`crate::store::trace::TraceHandle`] the dispatcher
    /// registered for `task_id`'s in-flight step, if any — the
    /// pervasive-insertion read port middlewares (and any other writer
    /// holding an `Engine`) use to append their own trace kinds. `None`
    /// = no trace rail for this dispatch (RunContext without a trace
    /// handle, or the step already folded).
    pub async fn trace_handle(&self, task_id: &StepId) -> Option<crate::store::trace::TraceHandle> {
        let key = task_id.clone();
        self.with_state("trace_handle", move |s| s.trace_handles.get(&key).cloned())
            .await
            .ok()
            .flatten()
    }

    /// Register (or clear, with `None`) the per-dispatch trace handle
    /// for `task_id`. Called only by `EngineDispatcher::dispatch` —
    /// insert before spawn, clear after the outcome fold. Best-effort:
    /// registry failures are swallowed (trace is observational).
    pub(crate) async fn set_trace_handle(
        &self,
        task_id: &StepId,
        handle: Option<crate::store::trace::TraceHandle>,
    ) {
        let key = task_id.clone();
        let _ = self
            .with_state("set_trace_handle", move |s| match handle {
                Some(h) => {
                    s.trace_handles.insert(key, h);
                }
                None => {
                    s.trace_handles.remove(&key);
                }
            })
            .await;
    }

    /// Returns the [`crate::core::agent_context::AgentContextView`]
    /// snapshotted for `(task_id, attempt)`, if `AgentContextMiddleware`
    /// stashed one — the same lookup [`Self::fetch_worker_payload`] /
    /// [`Self::fetch_worker_payload_trusted`] perform inline, exposed
    /// standalone for callers that only need the view (not a full
    /// `WorkerPayload`) — e.g. the HTTP debug-plane `GET
    /// /v1/tasks/:id/runs/:run/steps*` handlers resolving a
    /// materialized-file root for a step *other than* the one currently
    /// fetching its own prompt (`projection-adapter` ST5).
    pub async fn agent_context_for(
        &self,
        task_id: &StepId,
        attempt: u32,
    ) -> Option<crate::core::agent_context::AgentContextView> {
        let key = (task_id.clone(), attempt);
        self.with_state("agent_context_for", move |s| {
            s.agent_ctx.get(&key).map(|e| e.view.clone())
        })
        .await
        .ok()
        .flatten()
    }

    /// Resolves the [`FoldParse`] mode for `(task_id, attempt)` from the
    /// step's `AgentContextView.extra[`[`SUBMIT_FORMAT_KEY`]`]`:
    /// [`SUBMIT_FORMAT_TEXT`] opts the step's fold out of lenient
    /// container parsing; everything else — absent (the overwhelming
    /// majority of steps), `"json"` (whose strict parse already happened
    /// at submit time, so the fold sees a structured value it passes
    /// through), or an unrecognized value — folds `Lenient`.
    async fn fold_parse_mode_for(&self, task_id: &StepId, attempt: u32) -> FoldParse {
        match self.agent_context_for(task_id, attempt).await {
            Some(view)
                if view.extra.get(SUBMIT_FORMAT_KEY).and_then(|v| v.as_str())
                    == Some(SUBMIT_FORMAT_TEXT) =>
            {
                FoldParse::Raw
            }
            _ => FoldParse::Lenient,
        }
    }

    /// Read the current attempt number for a task (server-side lookup, no
    /// token verification). Used on `HTTP /v1/worker/result` when the
    /// worker omits `attempt` and the server has to fill it in.
    pub async fn task_attempt(&self, task_id: &StepId) -> Result<u32, EngineError> {
        let task_id = task_id.clone();
        self.with_state("task_attempt", move |s| {
            s.tasks
                .get(&task_id)
                .map(|t| t.attempt)
                .ok_or_else(|| EngineError::TaskNotFound(task_id.to_string()))
        })
        .await?
    }

    /// Server-side admin API that lets `OperatorSpawner::spawn` bake the
    /// rendered `system_prompt` into engine state. There is no verb gate
    /// — the only expected caller is inside the spawner. SubAgents fetch
    /// this alongside the prompt on the `/v1/worker/prompt` path.
    pub async fn bake_worker_system_prompt(
        &self,
        task_id: &StepId,
        attempt: u32,
        system: Option<String>,
    ) -> Result<(), EngineError> {
        let task_id = task_id.clone();
        self.with_state("bake_worker_system_prompt", move |s| {
            // GH #31: record this agent's most-recently-baked render size
            // before `system` is moved into `s.systems.insert` below. Same
            // `s.tasks.get(&task_id)` → `.spec.agent` lookup pattern
            // `fetch_worker_payload` uses (see its doc for why this keying
            // is load-bearing for a later `bp_doctor` route).
            if let Some(rendered) = system.as_ref() {
                if let Some(agent) = s.tasks.get(&task_id).map(|t| t.spec.agent.clone()) {
                    s.agent_render_sizes.insert(agent, rendered.len());
                }
            }
            s.systems.insert((task_id, attempt), system);
        })
        .await?;
        Ok(())
    }

    /// GH #31: the most-recently-baked `system_prompt` render size (in
    /// bytes) observed for `agent_name`, if `bake_worker_system_prompt` has
    /// ever recorded one — last-write-wins across every `(task_id,
    /// attempt)` dispatch of that agent. `None` when no `system_prompt`
    /// has ever been baked for this agent name. Read by the `bp_doctor`
    /// route this subtask's follow-up adds.
    pub async fn agent_last_rendered_size(&self, agent_name: &str) -> Option<usize> {
        let agent_name = agent_name.to_string();
        self.with_state("agent_last_rendered_size", move |s| {
            s.agent_render_sizes.get(&agent_name).copied()
        })
        .await
        .ok()
        .flatten()
    }

    /// GH #31: plain read-through of the baked `system` string for
    /// `(task_id, attempt)` from `EngineState.systems`, with no threshold
    /// branching. Backs `GET /v1/worker/prompt/system` (the `Http`-mode
    /// fetch target `system_ref.uri` points at) — that route needs the
    /// exact raw bytes to serve as the response body for the client's
    /// sha256 verification, not a `WorkerPayload`-wrapped value.
    ///
    /// Distinct from `apply_system_ref_threshold` (private, mutates an
    /// already-built `WorkerPayload` in place after full construction):
    /// this accessor has no threshold logic and is `pub` so
    /// `mlua-swarm-server`'s `worker` module can call it directly.
    ///
    /// Returns `Ok(None)` if no baked system exists for that `(task_id,
    /// attempt)` (either the task/attempt has no entry in `s.systems`, or
    /// the entry is present but stores `None`) — the caller maps this to
    /// a 404.
    pub async fn raw_system_prompt(
        &self,
        task_id: &StepId,
        attempt: u32,
    ) -> Result<Option<String>, EngineError> {
        let task_id = task_id.clone();
        self.with_state("raw_system_prompt", move |s| {
            s.systems.get(&(task_id, attempt)).cloned().unwrap_or(None)
        })
        .await
    }

    /// Fetch an arbitrary named resource previously stored via
    /// `set_resource`. Not task-scoped — any valid token with the
    /// `FetchData` verb may read any key.
    pub async fn fetch_data(&self, token: &CapToken, key: &str) -> Result<Value, EngineError> {
        self.verify_token(token, Verb::FetchData).await?;
        let key = key.to_string();
        self.with_state("fetch_data", move |s| {
            s.resources
                .get(&key)
                .cloned()
                .ok_or(EngineError::ResourceNotFound(key))
        })
        .await?
    }

    // ───────────────────────────────────────────────────────────────────────
    // Output path.
    // ───────────────────────────────────────────────────────────────────────

    /// Send one output event from inside a `SpawnerAdapter` or worker.
    /// Structuring is assumed to be complete by the time we cross the
    /// `SpawnerAdapter` boundary; this API just appends to the
    /// `OutputStore`, pushes to the `EventLog`, and (for `Final`) emits
    /// the `TaskAttemptCompleted` event.
    ///
    /// This is Domain-side plumbing: it feeds the engine's verdict flow,
    /// not the Data-plane store in the `output_store` module. It also
    /// does not wake the dispatch path — that is done through the
    /// spawner's completion oneshot when the worker terminates.
    ///
    /// # Submit-time projection sink (subtask-4 / ST2 rework)
    ///
    /// A `Final` event additionally fans out to the submit-time projection
    /// sink ([`Self::materialize_final_submission`]): (a) when
    /// [`Self::set_output_store`] has wired a Data-plane
    /// [`crate::store::output::OutputStore`], the event is dual-written
    /// there (`producer_agent` = `TaskState.spec.agent`, resolved to its
    /// GH #23 canonical projection name — see below), and (b) when this
    /// task's spawn ran through `AgentContextMiddleware` (so
    /// `EngineState.agent_ctx` has a `.view.work_dir` / `.view.project_root`
    /// for it), the value is additionally materialized to the
    /// [`crate::core::projection_placement::ProjectionPlacement`]
    /// resolver's target (byte-compat default layout
    /// `<root>/workspace/tasks/<task_id>/ctx/<canonical_agent>.md`) — see
    /// `crate::core::projection`'s module doc.
    ///
    /// **GH #23 subtask-2 (canonical sink):** both writes above key off the
    /// canonical name — `Engine::step_naming_for(task_id)`'s
    /// `StepNaming::canonical_of_producer(producer_agent)` when a table was
    /// snapshotted for this task (`EngineDispatcher::with_step_naming`),
    /// else `producer_agent` unchanged (fail-open, byte-identical to
    /// pre-GH-#23 behavior — see [`crate::core::step_naming`]'s module
    /// doc).
    ///
    /// **Invariants** (Subtask 4): (1) this sink is fail-open — an
    /// unresolved root, an unconfigured `OutputStore`, or either one
    /// erroring, only logs a `tracing::warn!` and never turns this
    /// `Ok(())` into an `Err`; (2) the wired `OutputStore` stays the single
    /// source of truth for cross-step queries — the materialized file is a
    /// projection of it, not a second store; (3) core does not depend on
    /// `mlua-swarm-server` — everything this sink touches
    /// (`crate::store::output` / `crate::core::projection`) already lives
    /// in this crate.
    ///
    /// # `Artifact` dual-write (GH #34 subtask-3 gap fix)
    ///
    /// An `Artifact` event ALSO fans out to the Data-plane, via
    /// [`Self::materialize_artifact_submission`] — general-form: every
    /// `Artifact` submitted through this API dual-writes, no name-prefix
    /// gate. Unlike `Final`, the dual-write key is the artifact's own
    /// `name` field, verbatim — NOT resolved through the GH #23 canonical
    /// `StepNaming` table. An artifact's `name` IS its identity (mirrors
    /// [`crate::store::output::OutputStore::get_latest_by_name`]'s doc),
    /// so no canonicalization applies. Same fail-open discipline as
    /// `Final` (Invariant 1 above), but `Artifact` does NOT drive the
    /// file-materialize half (b) — artifact findings (e.g.
    /// `AfterRunAuditMiddleware`'s `"audit:<step_ref>"`) are observational
    /// sidecar data, not a step's own submission a work_dir/project_root
    /// projection needs to track. `Progress` / `Partial` events are
    /// unaffected — no behavior change.
    pub async fn submit_output(
        &self,
        token: &crate::types::CapToken,
        task_id: &StepId,
        attempt: u32,
        event: crate::worker::output::OutputEvent,
    ) -> Result<(), EngineError> {
        self.verify_token_for_task(token, crate::types::Verb::EmitOutput, task_id)
            .await?;
        // GH #51 — completion-time verdict-contract enforcement, embedded
        // choke point 2 of 2 (see `Self::verdict_contract_completion_check`'s
        // doc). Guarded to `Final` only — the ONLY `OutputEvent` variant a
        // verdict contract's completion can meaningfully address; this
        // guard is defensive (this function is empirically called with
        // `Final` only today, both from `worker.rs`'s `worker_result` and
        // from `operator.rs`'s WS fallback) but costs nothing and protects
        // against a future non-`Final` caller. Runs BEFORE the
        // `output_tail` write immediately below: on `Err`, this returns
        // immediately and the write never happens — a rejected value
        // never reaches `output_tail` / the flow ctx.
        if let crate::worker::output::OutputEvent::Final { content, ok } = &event {
            let comparable_value = content_ref_to_comparable_string(content.clone());
            self.verdict_contract_completion_check(task_id, attempt, *ok, &comparable_value)
                .await?;
        }
        let task_id_for_apply = task_id.clone();
        let event_clone = event.clone();
        self.with_state("submit_output", move |s| {
            s.output_store
                .entry((task_id_for_apply.clone(), attempt))
                .or_default()
                .push(event_clone.clone());
            s.push_event(crate::core::state::Event::WorkerOutput {
                task_id: task_id_for_apply,
                attempt,
                event: event_clone,
            });
        })
        .await?;
        match &event {
            crate::worker::output::OutputEvent::Final { content, ok } => {
                self.materialize_final_submission(task_id, attempt, content, *ok)
                    .await?;
            }
            crate::worker::output::OutputEvent::Artifact { name, content } => {
                self.materialize_artifact_submission(task_id, attempt, name, content)
                    .await?;
            }
            _ => {}
        }
        Ok(())
    }

    /// Submit-time projection sink (subtask-4 / ST2 rework) shared by
    /// [`Self::submit_output`] and [`Self::submit_worker_result_trusted`].
    /// Best-effort / fail-open throughout (see `submit_output`'s doc
    /// Invariants): every failure path only `tracing::warn!`s and returns.
    ///
    /// Reads `(producer_agent, view)` via one read-only [`Self::with_state`]
    /// call — `producer_agent` off `TaskState.spec.agent`, `view` (the
    /// full [`crate::core::agent_context::AgentContextView`]) off
    /// `EngineState.agent_ctx[(task_id, attempt)]`, the same snapshot
    /// `crate::middleware::agent_context::AgentContextMiddleware` writes at
    /// spawn time — then does its actual (dual-write / file-write) work
    /// *outside* that lock, so a slow disk write or Data-plane store call
    /// never holds up unrelated `Engine::with_state` callers. `root` itself
    /// is resolved from `view` AFTER the lock via
    /// [`crate::core::projection_placement::ProjectionPlacement::resolve_root`]
    /// (GH #27, follow-up to #23) — the SAME resolver
    /// [`Self::step_naming_for`]'s sibling accessor
    /// [`Self::projection_placement_for`] snapshotted at dispatch time, so
    /// this sink's root-preference / fallback order is identical to the
    /// server read-back and the spawn-time pointer.
    async fn materialize_final_submission(
        &self,
        task_id: &StepId,
        attempt: u32,
        content: &crate::worker::output::ContentRef,
        ok: bool,
    ) -> Result<(), EngineError> {
        let server_policy = self.cfg().check_policy;
        let task_id_for_lookup = task_id.clone();
        let lookup = self
            .with_state("materialize_final_submission.lookup", move |s| {
                let entry = s.tasks.get(&task_id_for_lookup);
                let producer_agent = entry.map(|t| t.spec.agent.clone());
                let task_policy = entry.and_then(|t| t.spec.check_policy);
                let view = s
                    .agent_ctx
                    .get(&(task_id_for_lookup.clone(), attempt))
                    .map(|e| e.view.clone());
                (producer_agent, task_policy, view)
            })
            .await;
        // Per-task `TaskSpec.check_policy` (ST1c) wins
        // over the server-wide `EngineCfg.check_policy` when set — a
        // per-run override forwarded from the launch entry point (see
        // `TaskLaunchRequest.check_policy` /
        // `TaskLaunchInput.check_policy`). `None` leaves the server
        // default in effect (backward compat).
        let policy = lookup
            .as_ref()
            .ok()
            .and_then(|(_, tp, _)| *tp)
            .unwrap_or(server_policy);
        let (producer_agent, view) = match lookup.map(|(pa, _, view)| (pa, view)) {
            Ok(pair) => pair,
            Err(err) => {
                if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                    tracing::warn!(
                        %task_id,
                        error = %err,
                        "submit-time projection sink: state lookup failed; skipping (fail-open)"
                    );
                }
                apply_check_policy(
                    policy,
                    "submit-time projection sink: state lookup",
                    "state lookup failed; skipping (fail-open)",
                )?;
                return Ok(());
            }
        };
        let Some(producer_agent) = producer_agent else {
            // Defensive only: `task_id` is always a just-looked-up task at
            // every real call site. No task, no addressable producer name
            // — nothing to project. Not gated by `CheckPolicy` — a missing
            // task is an intentional early-exit path, not a fail-open
            // condition to surface.
            return Ok(());
        };
        let placement = self
            .projection_placement_for(task_id)
            .await
            .unwrap_or_default();
        let root = view.and_then(|v| placement.resolve_root(&v));

        // GH #23 subtask-2: resolve `producer_agent` to its canonical
        // projection name via the Blueprint-wide `StepNaming` table
        // snapshotted at dispatch time (`Engine::step_naming_for`). Both
        // write paths below ((a) data-plane, (b) file stem) use the
        // *canonical* name — `StepNaming::canonical_of_producer` returns
        // `producer_agent` unchanged for undeclared steps (byte-identical
        // to pre-GH-#23 behavior), and `None` (no table for this
        // `task_id`, e.g. a spawn that never went through
        // `EngineDispatcher::with_step_naming`) is a defensive fail-open
        // to the raw `producer_agent`, same discipline as the rest of this
        // sink.
        let canonical_agent = self
            .step_naming_for(task_id)
            .await
            .and_then(|naming| {
                naming
                    .canonical_of_producer(&producer_agent)
                    .map(str::to_string)
            })
            .unwrap_or_else(|| producer_agent.clone());

        // (a) Data-plane dual-write, when an OutputStore backend is wired.
        if let Some(store) = self.output_store_backend() {
            if let Err(err) = store
                .append(
                    task_id.as_str(),
                    attempt,
                    &canonical_agent,
                    crate::worker::output::OutputEvent::Final {
                        content: content.clone(),
                        ok,
                    },
                    Vec::new(),
                )
                .await
            {
                if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                    tracing::warn!(
                        %task_id,
                        agent = %producer_agent,
                        canonical = %canonical_agent,
                        error = %err,
                        "submit-time projection sink: OutputStore dual-write failed (fail-open)"
                    );
                }
                apply_check_policy(
                    policy,
                    "submit-time projection sink: OutputStore dual-write",
                    "OutputStore dual-write failed (fail-open)",
                )?;
            }
        }

        // (b) File materialize, when a root resolved.
        let Some(root) = root else {
            if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                tracing::warn!(
                    %task_id,
                    agent = %producer_agent,
                    canonical = %canonical_agent,
                    "submit-time projection sink: no work_dir/project_root resolved; skipping file materialize (fail-open)"
                );
            }
            apply_check_policy(
                policy,
                "submit-time projection sink: file materialize",
                "no work_dir/project_root resolved; skipping file materialize (fail-open)",
            )?;
            return Ok(());
        };
        let value = match content {
            crate::worker::output::ContentRef::Inline { value } => value.clone(),
            crate::worker::output::ContentRef::FileRef {
                path,
                mime,
                size_hint,
            } => serde_json::json!({
                "file_ref": path.to_string_lossy(),
                "mime": mime,
                "size_hint": size_hint,
            }),
        };
        let key = crate::core::projection::ProjectionKey {
            task_id: task_id.to_string(),
            run_id: None,
            step: Some(canonical_agent.clone()),
            path: None,
        };
        let adapter = crate::core::projection::FileProjectionAdapter::with_placement(
            root,
            (*placement).clone(),
        );
        if let Err(err) = adapter.materialize_submission(&key, &value, attempt, ok) {
            if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                tracing::warn!(
                    %task_id,
                    agent = %producer_agent,
                    canonical = %canonical_agent,
                    error = %err,
                    "submit-time projection sink: file materialize failed (fail-open)"
                );
            }
            apply_check_policy(
                policy,
                "submit-time projection sink: file materialize",
                "file materialize failed (fail-open)",
            )?;
        }
        Ok(())
    }

    /// Submit-time projection sink for `OutputEvent::Artifact` (GH #34
    /// subtask-3, later extended to drive the file half too). Two halves, the
    /// [`Self::materialize_final_submission`] mirror for staged named parts:
    ///
    /// - **Data-plane dual-write** — when [`Self::set_output_store`] has
    ///   wired a [`crate::store::output::OutputStore`], the artifact
    ///   dual-writes there under its own `name`, verbatim (general form:
    ///   every `Artifact` staged via [`Self::submit_output`] /
    ///   [`Self::stage_worker_artifact_trusted`] materializes this way, no
    ///   name-prefix gate).
    /// - **File materialize** — when a `root` resolves off the spawn-time
    ///   [`crate::core::agent_context::AgentContextView`], the part's
    ///   content is written raw to `<ctx-dir>/<name>` via
    ///   [`crate::core::projection::FileProjectionAdapter::materialize_part`].
    ///   That file is the IN file the *next* Agent step reads: materializing
    ///   a Step's OUTPUT to disk is the
    ///   [`crate::core::projection::FileProjectionAdapter`]'s
    ///   responsibility, and a staged named part is as much an OUTPUT the
    ///   next step consumes as a `Final` is — so the sink materializes it
    ///   too, rather than leaving parts Data-plane-only.
    ///
    /// Unlike the Final sink, no `StepNaming` canonicalization is applied:
    /// an artifact's `name` already IS the key both halves address (it
    /// names the file directly, extension included — `plan.md` — so
    /// `materialize_part` writes it verbatim, not through the `<stem>.md`
    /// synthesis the Final sink's canonical-agent path uses).
    ///
    /// Fail-open throughout, the same `check_policy` cascade as
    /// [`Self::materialize_final_submission`]: a per-task lookup error falls
    /// back to the server default (and a `None` view ⇒ the file half's
    /// unresolved-root path), an unconfigured `OutputStore` skips the
    /// dual-write, an unresolved root skips the file half, and a
    /// dual-write / file-write / name-guard error only `tracing::warn!`s
    /// (`Silent` suppresses even that) before applying [`apply_check_policy`]
    /// (`Strict` surfaces an [`EngineError`], `Warn` / `Silent` return
    /// `Ok(())`) — a staged part never turns a would-have-succeeded submit
    /// into a failure under the default policy.
    async fn materialize_artifact_submission(
        &self,
        task_id: &StepId,
        attempt: u32,
        name: &str,
        content: &crate::worker::output::ContentRef,
    ) -> Result<(), EngineError> {
        // Per-task `TaskSpec.check_policy` override + the `AgentContextView`
        // snapshot, resolved in ONE read-only `with_state` (the same lock
        // the policy lookup already needed — no extra `with_state` for the
        // view). Silent per-task lookup failure (`with_state` error) falls
        // back to the server-wide default and a `None` view (⇒ the file
        // half's own unresolved-root fail-open path); this sink never
        // surfaces the lookup error itself as a step failure.
        let server_policy = self.cfg().check_policy;
        let task_id_for_lookup = task_id.clone();
        let lookup = self
            .with_state("materialize_artifact_submission.lookup", move |s| {
                let task_policy = s
                    .tasks
                    .get(&task_id_for_lookup)
                    .and_then(|t| t.spec.check_policy);
                let view = s
                    .agent_ctx
                    .get(&(task_id_for_lookup.clone(), attempt))
                    .map(|e| e.view.clone());
                (task_policy, view)
            })
            .await
            .ok();
        let policy = lookup
            .as_ref()
            .and_then(|(tp, _)| *tp)
            .unwrap_or(server_policy);
        let view = lookup.and_then(|(_, view)| view);

        // (a) Data-plane dual-write, when an OutputStore backend is wired —
        // the artifact's own `name` is its Data-plane key (no
        // canonicalization, unlike the Final sink's `StepNaming`
        // resolution).
        if let Some(store) = self.output_store_backend() {
            if let Err(err) = store
                .append(
                    task_id.as_str(),
                    attempt,
                    name,
                    crate::worker::output::OutputEvent::Artifact {
                        name: name.to_string(),
                        content: content.clone(),
                    },
                    Vec::new(),
                )
                .await
            {
                if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                    tracing::warn!(
                        %task_id,
                        artifact = %name,
                        error = %err,
                        "submit-time projection sink: OutputStore dual-write failed for Artifact (fail-open)"
                    );
                }
                apply_check_policy(
                    policy,
                    "submit-time projection sink: Artifact OutputStore dual-write",
                    "OutputStore dual-write failed for Artifact (fail-open)",
                )?;
            }
        }

        // (b) File materialize, when a root resolved — writes the staged
        // part raw to `<ctx-dir>/<name>`, the IN file the next Agent step
        // reads (see `FileProjectionAdapter::materialize_part`'s doc for
        // why raw / why the name is verbatim). A name-guard violation lands
        // on the same fail-open path as any other write error below.
        let placement = self
            .projection_placement_for(task_id)
            .await
            .unwrap_or_default();
        let Some(root) = view.and_then(|v| placement.resolve_root(&v)) else {
            if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                tracing::warn!(
                    %task_id,
                    artifact = %name,
                    "submit-time projection sink: no work_dir/project_root resolved; skipping part file materialize (fail-open)"
                );
            }
            apply_check_policy(
                policy,
                "submit-time projection sink: part file materialize",
                "no work_dir/project_root resolved; skipping part file materialize (fail-open)",
            )?;
            return Ok(());
        };
        let value = match content {
            crate::worker::output::ContentRef::Inline { value } => value.clone(),
            crate::worker::output::ContentRef::FileRef {
                path,
                mime,
                size_hint,
            } => serde_json::json!({
                "file_ref": path.to_string_lossy(),
                "mime": mime,
                "size_hint": size_hint,
            }),
        };
        let adapter = crate::core::projection::FileProjectionAdapter::with_placement(
            root,
            (*placement).clone(),
        );
        if let Err(err) = adapter.materialize_part(task_id.as_str(), name, &value) {
            if !matches!(policy, crate::core::config::CheckPolicy::Silent) {
                tracing::warn!(
                    %task_id,
                    artifact = %name,
                    error = %err,
                    "submit-time projection sink: part file materialize failed (fail-open)"
                );
            }
            apply_check_policy(
                policy,
                "submit-time projection sink: part file materialize",
                "part file materialize failed (fail-open)",
            )?;
        }
        Ok(())
    }

    /// Snapshot the entire output tail for a given `(task_id, attempt)`.
    /// Used by the dispatch path when pulling `Final`, and by observers
    /// reading the trace.
    pub async fn output_tail(
        &self,
        task_id: &StepId,
        attempt: u32,
    ) -> Vec<crate::worker::output::OutputEvent> {
        let key = (task_id.clone(), attempt);
        self.with_state("output_tail", move |s| {
            s.output_store.get(&key).cloned().unwrap_or_default()
        })
        .await
        .unwrap_or_default()
    }

    /// Record an interim `last_result` for `task_id` without changing its
    /// `status`. Distinct from the terminal `Final` output event handled
    /// through `submit_output` / `dispatch_attempt_with`.
    pub async fn post_result(
        &self,
        token: &CapToken,
        task_id: &StepId,
        result: Value,
    ) -> Result<(), EngineError> {
        self.verify_token_for_task(token, Verb::PostResult, task_id)
            .await?;
        let task_id = task_id.clone();
        let result_clone = result.clone();
        self.with_state("post_result", move |s| {
            let task = s
                .tasks
                .get_mut(&task_id)
                .ok_or_else(|| EngineError::TaskNotFound(task_id.to_string()))?;
            task.last_result = Some(result_clone);
            task.updated_at = now_unix();
            Ok::<(), EngineError>(())
        })
        .await??;
        Ok(())
    }

    /// Store a named resource value, retrievable later via `fetch_data`.
    /// No token is required — this is a server-side/admin-style setter
    /// (mirrors `bake_worker_system_prompt`).
    pub async fn set_resource(
        &self,
        key: impl Into<String>,
        value: Value,
    ) -> Result<(), EngineError> {
        let key = key.into();
        self.with_state("set_resource", move |s| {
            s.resources.insert(key, value);
        })
        .await?;
        Ok(())
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Senior suspend / resume
    // ═══════════════════════════════════════════════════════════════════════

    /// Ask a question of the Senior, mark the task `Suspended`, and
    /// return a `ResumeKey`. The suspended state persists until another
    /// task calls `resume(key, answer)`.
    ///
    /// Resume-side waiting is `Notify`-based, so a caller (typically
    /// MainAI) can detach, reattach from a different process, and still
    /// pull the answer out via `await_resume(key, timeout)` — the answer
    /// is stored inside `EngineState`.
    pub async fn query_senior(
        &self,
        token: &CapToken,
        task_id: &StepId,
        question: Value,
    ) -> Result<ResumeKey, EngineError> {
        self.verify_token(token, Verb::QuerySenior).await?;
        let task_id = task_id.clone();
        let key = ResumeKey::for_senior(&task_id);
        let task_notify = self
            .with_state("query_senior.notify_ensure", |s| {
                s.ensure_task_notify(&task_id)
            })
            .await?;

        let key_clone = key.clone();
        let task_id_inner = task_id.clone();
        let question_clone = question.clone();
        self.with_state("query_senior.suspend", move |s| {
            let task = s
                .tasks
                .get_mut(&task_id_inner)
                .ok_or_else(|| EngineError::TaskNotFound(task_id_inner.to_string()))?;
            task.status = TaskStatus::Suspended;
            task.suspended_on = Some(key_clone.clone());
            task.updated_at = now_unix();
            s.pending_resumes
                .insert(key_clone.clone(), ResumePending::new());
            s.push_event(Event::SeniorQueried {
                task_id: task_id_inner.clone(),
                question: question_clone.clone(),
            });
            s.push_event(Event::TaskSuspended {
                task_id: task_id_inner.clone(),
                key: key_clone.clone(),
            });
            Ok::<(), EngineError>(())
        })
        .await??;

        // Notify callers waiting for a task status change (Running → Suspended).
        task_notify.notify_waiters();

        let _ = self
            .inner
            .event_tx
            .send(Event::SeniorQueried { task_id, question });
        Ok(key)
    }

    /// Store the answer for a `ResumeKey` in `EngineState` and wake the
    /// waiting caller via `Notify`. Also flips the suspended task's
    /// status back to `Running` and fires the per-task notifier.
    pub async fn resume(&self, key: ResumeKey, answer: Value) -> Result<(), EngineError> {
        let answer_for_state = answer.clone();
        let answer_for_event = answer.clone();
        let key_clone = key.clone();
        let (notify, task_notify, task_id_opt) = self
            .with_state("resume.set", move |s| {
                let pending = s
                    .pending_resumes
                    .get_mut(&key_clone)
                    .ok_or(EngineError::ResumeKeyNotFound)?;
                pending.answer = Some(answer_for_state);
                let notify = pending.notify.clone();

                let task_id = s
                    .tasks
                    .iter()
                    .find(|(_, t)| t.suspended_on.as_ref() == Some(&key_clone))
                    .map(|(id, _)| id.clone());

                let task_notify = task_id.as_ref().map(|tid| s.ensure_task_notify(tid));

                if let Some(tid) = &task_id {
                    if let Some(task) = s.tasks.get_mut(tid) {
                        task.suspended_on = None;
                        task.status = TaskStatus::Running;
                        task.updated_at = now_unix();
                    }
                    s.push_event(Event::TaskResumed {
                        task_id: tid.clone(),
                        key: key_clone.clone(),
                    });
                    s.push_event(Event::SeniorAnswered {
                        task_id: tid.clone(),
                        answer: answer_for_event.clone(),
                    });
                }
                Ok::<_, EngineError>((notify, task_notify, task_id))
            })
            .await??;

        // Outside the lock: notify_waiters for both the ResumePending and task-status waits.
        notify.notify_waiters();
        if let Some(n) = task_notify {
            n.notify_waiters();
        }

        if let Some(tid) = task_id_opt {
            let _ = self
                .inner
                .event_tx
                .send(Event::TaskResumed { task_id: tid, key });
        }
        Ok(())
    }

    /// Wait for the resume answer. Even if the caller (an Operator)
    /// detached and reattached, the answer is available immediately here
    /// — if it was already stored, this returns without waiting on the
    /// notifier.
    ///
    /// `timeout = Duration::ZERO` performs an instant check without
    /// waiting.
    pub async fn await_resume(
        &self,
        key: ResumeKey,
        timeout: Duration,
    ) -> Result<Value, EngineError> {
        // (1) Under the lock: clone the notify handle and check for an existing answer.
        let key_clone = key.clone();
        let (notify, existing) = self
            .with_state("await_resume.snapshot", move |s| {
                let pending = s
                    .pending_resumes
                    .get(&key_clone)
                    .ok_or(EngineError::ResumeKeyNotFound)?;
                Ok::<_, EngineError>((pending.notify.clone(), pending.answer.clone()))
            })
            .await??;

        // (2) If an answer has already been stored, return immediately (detach / reattach pattern).
        if let Some(v) = existing {
            return Ok(v);
        }

        // (3) Outside the lock: wait on the notify with a timeout.
        if timeout.is_zero() {
            return Err(EngineError::PollTimeout);
        }
        let waited = tokio::time::timeout(timeout, notify.notified()).await;
        if waited.is_err() {
            return Err(EngineError::PollTimeout);
        }

        // (4) Under the lock: re-read the answer (should be present now that we were notified).
        let key_clone = key.clone();
        self.with_state("await_resume.read", move |s| {
            let pending = s
                .pending_resumes
                .get(&key_clone)
                .ok_or(EngineError::ResumeKeyNotFound)?;
            pending
                .answer
                .clone()
                .ok_or_else(|| EngineError::Internal("notified but answer missing".into()))
        })
        .await?
    }

    // ═══════════════════════════════════════════════════════════════════════
    // poll_task — the "wait" path that waits for task-status changes (works for long-poll and regular wait).
    // ═══════════════════════════════════════════════════════════════════════

    /// Wait until the task's status **transitions to terminal or
    /// `Suspended`**, then return the latest `TaskState`. Returns
    /// immediately if the task is already in a terminal state.
    /// Exceeding the timeout returns `EngineError::PollTimeout`.
    ///
    /// A `hold` of `Duration::from_secs(0)` returns a snapshot immediately
    /// (no wait). Larger holds — tens of minutes up to days — are fine;
    /// the wait state is kept in memory inside the engine and does not
    /// degrade.
    pub async fn poll_task(
        &self,
        token: &CapToken,
        task_id: &StepId,
        hold: Duration,
    ) -> Result<TaskState, EngineError> {
        self.verify_token_for_task(token, Verb::PollTask, task_id)
            .await?;
        let task_id_inner = task_id.clone();

        // (1) Under the lock: take a snapshot and clone task_notify.
        let (state, notify) = self
            .with_state("poll_task.snapshot", move |s| {
                let task = s
                    .tasks
                    .get(&task_id_inner)
                    .cloned()
                    .ok_or_else(|| EngineError::TaskNotFound(task_id_inner.to_string()))?;
                let notify = s.ensure_task_notify(&task_id_inner);
                Ok::<_, EngineError>((task, notify))
            })
            .await??;

        // (2) Immediate-return condition: already terminal / Suspended (nothing left to wait on).
        if matches!(
            state.status,
            TaskStatus::Pass | TaskStatus::Blocked | TaskStatus::Cancelled | TaskStatus::Suspended
        ) {
            return Ok(state);
        }
        if hold.is_zero() {
            return Ok(state);
        }

        // (3) Outside the lock: wait on Notify with a timeout.
        let waited = tokio::time::timeout(hold, notify.notified()).await;
        if waited.is_err() {
            return Err(EngineError::PollTimeout);
        }

        // (4) Under the lock: take a fresh snapshot.
        let task_id_inner = task_id.clone();
        self.with_state("poll_task.reread", move |s| {
            s.tasks
                .get(&task_id_inner)
                .cloned()
                .ok_or_else(|| EngineError::TaskNotFound(task_id_inner.to_string()))
        })
        .await?
    }

    // ═══════════════════════════════════════════════════════════════════════
    // Background: heartbeat miss → detach loop
    // ═══════════════════════════════════════════════════════════════════════

    /// Background loop that scans sessions every `heartbeat_interval` and
    /// flips `attached = false` on any session whose `last_seen` exceeds
    /// `heartbeat_miss_threshold * interval`.
    ///
    /// The tasks themselves are kept (assuming
    /// `keepalive_on_idle = true`), so another client can reattach with
    /// the same token and resume immediately. Dropping the returned
    /// `JoinHandle` does not stop the loop — the handle exists so callers
    /// who want to abort can hold onto it.
    pub fn start_detach_loop(&self) -> tokio::task::JoinHandle<()> {
        let engine = self.clone();
        let cfg = self.inner.cfg.long_hold.clone();
        let interval = cfg.heartbeat_interval;
        let miss_secs = cfg.heartbeat_interval.as_secs() * cfg.heartbeat_miss_threshold as u64;

        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(interval);
            ticker.tick().await; // first tick is immediate
            loop {
                ticker.tick().await;
                let now = now_unix();
                let detached = engine
                    .with_state("detach_loop.scan", |s| {
                        let mut detached = Vec::new();
                        for (sid, sess) in s.sessions.iter_mut() {
                            if !sess.attached {
                                continue;
                            }
                            if now.saturating_sub(sess.last_seen) >= miss_secs {
                                sess.attached = false;
                                detached.push(sid.clone());
                            }
                        }
                        for sid in &detached {
                            s.push_event(Event::SessionDetached {
                                session_id: sid.clone(),
                            });
                        }
                        detached
                    })
                    .await
                    .unwrap_or_default();
                for sid in detached {
                    let _ = engine
                        .inner
                        .event_tx
                        .send(Event::SessionDetached { session_id: sid });
                }
            }
        })
    }

    /// Helper: wake a task whose status has changed. Called from the
    /// method body outside the lock.
    async fn wake_task(&self, task_id: &StepId) -> Result<(), EngineError> {
        let task_id = task_id.clone();
        let notify_opt = self
            .with_state("wake_task.get_notify", move |s| {
                s.task_notifies.get(&task_id).cloned()
            })
            .await?;
        if let Some(n) = notify_opt {
            n.notify_waiters();
        }
        Ok(())
    }
}

/// Decide what a submit-time projection sink should do at a fail-open
/// branch given the configured [`crate::core::config::CheckPolicy`].
///
/// Returns `Ok(())` under [`CheckPolicy::Silent`] and
/// [`CheckPolicy::Warn`] — the caller continues with fail-open. Returns
/// [`EngineError::CheckPolicyStrict`] under [`CheckPolicy::Strict`],
/// carrying the caller-supplied `context` (call-site identifier) and
/// `message` (the pre-existing warn-log message literal, preserved
/// verbatim for log parse compatibility).
///
/// This helper deliberately does **not** call `tracing::warn!` itself —
/// the caller is responsible for firing the existing warn! (with its
/// full structured-field payload — `%task_id`, `agent`, `canonical`,
/// `error`, etc.) under `Warn` mode, and for skipping the warn! under
/// `Silent` mode. Keeping the warn! at the call site preserves the
/// exact structured-field shape every existing log-parse consumer sees;
/// forwarding it through the helper would either drop those fields or
/// require a macro (deferred, see subtask-1b).
///
/// Design intent: the fail-open discipline of every submit-time
/// projection sink is byte-identical to the pre-`CheckPolicy` behaviour
/// under the default [`CheckPolicy::Warn`]. `Silent` is a per-run opt-in
/// to suppress noise (e.g., a caller that has already verified upstream
/// invariants); `Strict` is a per-run opt-in to fail loudly (e.g., a
/// caller that requires all parts to materialize). See
/// [`crate::core::config::CheckPolicy`] for the "state dirty on fail"
/// semantics of `Strict`.
pub(crate) fn apply_check_policy(
    policy: crate::core::config::CheckPolicy,
    context: &str,
    message: &str,
) -> Result<(), EngineError> {
    match policy {
        crate::core::config::CheckPolicy::Silent | crate::core::config::CheckPolicy::Warn => Ok(()),
        crate::core::config::CheckPolicy::Strict => Err(EngineError::CheckPolicyStrict {
            context: context.to_string(),
            message: message.to_string(),
        }),
    }
}

#[cfg(test)]
mod check_policy_helper_tests {
    use super::apply_check_policy;
    use crate::core::config::CheckPolicy;
    use crate::core::errors::EngineError;

    /// `Silent` returns `Ok(())` without producing an error. Log
    /// suppression (the "no `tracing::warn!`" half of the semantics) is
    /// enforced at the call site, not inside the helper — see the
    /// helper's doc comment for why.
    #[test]
    fn silent_returns_ok() {
        let result = apply_check_policy(CheckPolicy::Silent, "call/site", "sink message");
        assert!(matches!(result, Ok(())));
    }

    /// `Warn` (the default) returns `Ok(())` — the caller continues
    /// with fail-open, having already fired its own `tracing::warn!`
    /// with the full structured-field payload.
    #[test]
    fn warn_returns_ok() {
        let result = apply_check_policy(CheckPolicy::Warn, "call/site", "sink message");
        assert!(matches!(result, Ok(())));
    }

    /// `Strict` returns
    /// [`EngineError::CheckPolicyStrict`] with `context` and `message`
    /// copied verbatim from the caller — the completion route surfaces
    /// this as a step / launch error so a caller that has opted in can
    /// fail fast instead of proceeding with a partially-realized
    /// submission.
    #[test]
    fn strict_returns_error_with_context_and_message() {
        let result = apply_check_policy(
            CheckPolicy::Strict,
            "submit-time projection sink: file materialize",
            "no work_dir/project_root resolved; skipping file materialize (fail-open)",
        );
        match result {
            Err(EngineError::CheckPolicyStrict { context, message }) => {
                assert_eq!(context, "submit-time projection sink: file materialize");
                assert_eq!(
                    message,
                    "no work_dir/project_root resolved; skipping file materialize (fail-open)"
                );
            }
            other => panic!("expected CheckPolicyStrict, got {:?}", other),
        }
    }
}

// ─── UT: R4 max-hold guard — warn + continue by default, panic on opt-in ────
#[cfg(test)]
mod max_hold_guard_tests {
    use super::*;

    /// `max_hold_panic = true` keeps the hard failure available: an
    /// over-budget closure unwinds with the historical message so an R3
    /// violation is impossible to miss during a local hunt.
    #[tokio::test]
    #[should_panic(expected = "suspected R3 violation")]
    async fn with_state_over_max_hold_panics_when_opted_in() {
        let engine = Engine::new(EngineCfg {
            max_hold_ms: 0,
            max_hold_panic: true,
            ..EngineCfg::default()
        });
        // `with_state` takes a sync `FnOnce`, so a blocking sleep is the
        // only way to overrun the budget from inside the lock.
        let _ = engine
            .with_state("test.over_max_hold", |_s| {
                std::thread::sleep(Duration::from_millis(5));
            })
            .await;
    }

    /// Default config only warns in every build: the call returns `Ok` and
    /// the caller's task survives. This keeps a run driver future from being
    /// unwound (RunRecord stranded in `Running`) and keeps CI deterministic —
    /// wall-clock hold time on a loaded shared runner includes scheduler
    /// preemption, which is not an R3 violation.
    #[tokio::test]
    async fn with_state_over_max_hold_warns_and_returns_by_default() {
        let engine = Engine::new(EngineCfg {
            max_hold_ms: 0,
            ..EngineCfg::default()
        });
        let result = engine
            .with_state("test.over_max_hold", |_s| {
                std::thread::sleep(Duration::from_millis(5));
                42u32
            })
            .await;
        assert_eq!(result.expect("default config must not panic"), 42);
    }
}

// ─── UT: issue #14 — token store keyed by fingerprint, not nonce ────────────
#[cfg(test)]
mod token_fingerprint_store_tests {
    use super::*;

    /// A token that was never attached fails verify with a `TokenNotFound`
    /// that carries the fingerprint — never the nonce. The error string can
    /// surface in HTTP error bodies, so this is the secret-hygiene contract.
    #[tokio::test]
    async fn verify_unknown_token_reports_fingerprint_not_nonce() {
        let engine = Engine::new(EngineCfg::default());
        // Signed by the engine's own signer (sig passes) but never inserted
        // into the store — verify must fail at step (4), the store lookup.
        let token = engine.signer().session(
            "ghost",
            Role::Operator,
            vec!["*".into()],
            Duration::from_secs(60),
        );
        let err = engine
            .verify_token(&token, Verb::ReadTaskState)
            .await
            .expect_err("token is not in the store");
        let msg = err.to_string();
        assert!(
            msg.contains(&token.fingerprint()),
            "error must carry the fingerprint: {msg}"
        );
        assert!(
            !msg.contains(&token.nonce),
            "error must not leak the nonce: {msg}"
        );
    }

    /// attach → verify → heartbeat → detach all resolve the session /
    /// token record through fingerprint keys (mint/verify lifecycle
    /// regression guard for the issue #14 key migration).
    #[tokio::test]
    async fn attach_verify_heartbeat_detach_cycle_with_fp_keying() {
        let engine = Engine::new(EngineCfg::default());
        let token = engine
            .attach("op-1", Role::Operator, Duration::from_secs(60))
            .await
            .expect("attach");
        engine
            .verify_token(&token, Verb::ReadTaskState)
            .await
            .expect("verify consumes via fp key");
        engine
            .heartbeat(&token)
            .await
            .expect("heartbeat finds the session by fp");
        engine
            .detach(&token)
            .await
            .expect("detach finds the session by fp");
    }
}

// ─── UT: `OperatorKind` "Runtime Global" tier — `Option` semantics ─────────
//
// Regression coverage for the "explicit Automate is indistinguishable from
// unspecified" defect: `OperatorSession.operator_kind` (and the
// `attach_with_ids` `kind` parameter it stores) is `Option<OperatorKind>`,
// so `Some(Automate)` is an explicit Runtime Global request that must
// outrank `bp_global`, while `None` must let `bp_global` decide. Exercises
// the real `resolve_operator_info` cascade path (not just
// `collapse_operator_kind` in isolation), attaching via `attach_with_ids`
// exactly as `TaskLaunchService::launch` does.
#[cfg(test)]
mod resolve_operator_info_runtime_global_tests {
    use super::*;

    async fn attach_and_resolve(
        runtime_global: Option<OperatorKind>,
        bp_global: Option<OperatorKind>,
    ) -> OperatorInfo {
        let engine = Engine::new(EngineCfg::default());
        let token = engine
            .attach_with_ids(
                "ut-op",
                Role::Operator,
                Duration::from_secs(30),
                runtime_global,
                None,
                None,
                None,
                HashMap::new(),
                HashMap::new(),
                bp_global,
            )
            .await
            .expect("attach_with_ids ok");
        let session = engine
            .with_state("test.find_session", |s| {
                s.sessions
                    .values()
                    .find(|sess| sess.token_fp == token.fingerprint())
                    .cloned()
            })
            .await
            .expect("with_state ok")
            .expect("session present after attach_with_ids");
        engine.resolve_operator_info(&session, "agent-x").await
    }

    #[tokio::test]
    async fn explicit_some_automate_outranks_bp_global_main_ai() {
        // Runtime Global explicitly requests Automate; bp_global is MainAi.
        // The explicit `Some(Automate)` must win — this is exactly the case
        // the old `== OperatorKind::default()` convention got wrong (it
        // could not tell "explicitly Automate" from "unspecified" and would
        // have let `bp_global` (MainAi) take over instead).
        let info =
            attach_and_resolve(Some(OperatorKind::Automate), Some(OperatorKind::MainAi)).await;
        assert_eq!(
            info.kind,
            OperatorKind::Automate,
            "explicit Some(Automate) runtime_global must outrank bp_global MainAi"
        );
    }

    #[tokio::test]
    async fn none_lets_bp_global_main_ai_win() {
        // Runtime Global left unspecified (`None`); bp_global is MainAi.
        // With nothing more specific set, `bp_global` must decide.
        let info = attach_and_resolve(None, Some(OperatorKind::MainAi)).await;
        assert_eq!(
            info.kind,
            OperatorKind::MainAi,
            "None runtime_global must let bp_global MainAi win"
        );
    }
}

/// issue #13 run_id propagation: `dispatch_attempt_with`'s `run_id` param
/// must land in `Ctx.meta.runtime["run_id"]` (the same slot pattern as the
/// pre-existing `worker_handle`), or be omitted entirely when `None`. Same
/// `CtxProbe` shape as `middleware::worker_binding`'s test module — an
/// inner `SpawnerAdapter` that snapshots the `Ctx` it was called with and
/// fails the spawn (only the ctx snapshot matters here).
#[cfg(test)]
mod dispatch_attempt_with_run_id_tests {
    use super::*;
    use crate::worker::adapter::{SpawnError, SpawnerAdapter};
    use crate::worker::Worker;
    use std::sync::Mutex as StdMutex;

    struct CtxProbe {
        seen: Arc<StdMutex<Option<Ctx>>>,
    }

    #[async_trait::async_trait]
    impl SpawnerAdapter for CtxProbe {
        async fn spawn(
            &self,
            _engine: &Engine,
            ctx: &Ctx,
            _task_id: StepId,
            _attempt: u32,
            _token: CapToken,
        ) -> Result<Box<dyn Worker>, SpawnError> {
            *self.seen.lock().unwrap() = Some(ctx.clone());
            Err(SpawnError::Internal("probe stop".into()))
        }
    }

    async fn dispatch_with_probe(run_id: Option<&RunId>) -> Ctx {
        let engine = Engine::new(EngineCfg::default());
        let token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let tid = engine
            .start_task(
                &token,
                TaskSpec {
                    agent: "probe".into(),
                    initial_directive: "hi".into(),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        let seen: Arc<StdMutex<Option<Ctx>>> = Arc::new(StdMutex::new(None));
        let spawner: Arc<dyn SpawnerAdapter> = Arc::new(CtxProbe { seen: seen.clone() });
        // The probe always errors the spawn (`SpawnError::Internal`); we
        // only care about the `Ctx` snapshot it captured, so the dispatch
        // outcome itself (`Err`) is discarded.
        let _ = engine
            .dispatch_attempt_with(&token, &tid, &spawner, run_id)
            .await;
        let captured = seen.lock().unwrap().clone();
        captured.expect("inner ctx captured")
    }

    #[tokio::test]
    async fn run_id_lands_in_ctx_meta_runtime_when_some() {
        let run_id = RunId::new();
        let observed = dispatch_with_probe(Some(&run_id)).await;
        assert_eq!(
            observed.meta.runtime.get("run_id").and_then(|v| v.as_str()),
            Some(run_id.as_str()),
            "ctx.meta.runtime[\"run_id\"] must carry the run_id passed to dispatch_attempt_with"
        );
    }

    #[tokio::test]
    async fn run_id_key_absent_when_none() {
        let observed = dispatch_with_probe(None).await;
        assert!(
            !observed.meta.runtime.contains_key("run_id"),
            "no run_id key must be injected when dispatch_attempt_with is called with None"
        );
    }
}

/// GH #21 Phase 2: `TaskSpec.step_ctx` must land in
/// `Ctx.meta.runtime[STEP_CTX_KEY]` — re-read from the spec on EVERY
/// attempt (the prep closure re-reads `task.spec.step_ctx` every call, not
/// caching it once at `start_task`), so a retry (attempt 2) carries it
/// too. Same `CtxProbe` shape as `dispatch_attempt_with_run_id_tests`.
#[cfg(test)]
mod dispatch_attempt_with_step_ctx_tests {
    use super::*;
    use crate::worker::adapter::{SpawnError, SpawnerAdapter};
    use crate::worker::Worker;
    use std::sync::Mutex as StdMutex;

    struct CtxProbe {
        seen: Arc<StdMutex<Option<Ctx>>>,
    }

    #[async_trait::async_trait]
    impl SpawnerAdapter for CtxProbe {
        async fn spawn(
            &self,
            _engine: &Engine,
            ctx: &Ctx,
            _task_id: StepId,
            _attempt: u32,
            _token: CapToken,
        ) -> Result<Box<dyn Worker>, SpawnError> {
            *self.seen.lock().unwrap() = Some(ctx.clone());
            Err(SpawnError::Internal("probe stop".into()))
        }
    }

    #[tokio::test]
    async fn step_ctx_lands_in_ctx_meta_runtime_on_attempt_1_and_2() {
        let engine = Engine::new(EngineCfg::default());
        let token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let tid = engine
            .start_task(
                &token,
                TaskSpec {
                    agent: "probe".into(),
                    initial_directive: "hi".into(),
                    step_ctx: Some(serde_json::json!({ "work_dir": "/step" })),
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        let seen: Arc<StdMutex<Option<Ctx>>> = Arc::new(StdMutex::new(None));
        let spawner: Arc<dyn SpawnerAdapter> = Arc::new(CtxProbe { seen: seen.clone() });

        // The probe always errors the spawn; only the ctx snapshot matters.
        let _ = engine
            .dispatch_attempt_with(&token, &tid, &spawner, None)
            .await;
        let first = seen
            .lock()
            .unwrap()
            .clone()
            .expect("attempt 1 ctx captured");
        assert_eq!(
            first.meta.runtime.get(STEP_CTX_KEY),
            Some(&serde_json::json!({ "work_dir": "/step" })),
            "attempt 1 must carry TaskSpec.step_ctx in ctx.meta.runtime[STEP_CTX_KEY]"
        );

        let _ = engine
            .dispatch_attempt_with(&token, &tid, &spawner, None)
            .await;
        let second = seen
            .lock()
            .unwrap()
            .clone()
            .expect("attempt 2 ctx captured");
        assert_eq!(
            second.meta.runtime.get(STEP_CTX_KEY),
            Some(&serde_json::json!({ "work_dir": "/step" })),
            "attempt 2 (retry) must ALSO carry TaskSpec.step_ctx — prep re-reads the spec every attempt"
        );
    }

    #[tokio::test]
    async fn step_ctx_key_absent_when_none() {
        let engine = Engine::new(EngineCfg::default());
        let token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let tid = engine
            .start_task(
                &token,
                TaskSpec {
                    agent: "probe".into(),
                    initial_directive: "hi".into(),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        let seen: Arc<StdMutex<Option<Ctx>>> = Arc::new(StdMutex::new(None));
        let spawner: Arc<dyn SpawnerAdapter> = Arc::new(CtxProbe { seen: seen.clone() });
        let _ = engine
            .dispatch_attempt_with(&token, &tid, &spawner, None)
            .await;
        let observed = seen.lock().unwrap().clone().expect("ctx captured");
        assert!(
            !observed.meta.runtime.contains_key(STEP_CTX_KEY),
            "no step_ctx key must be injected when TaskSpec.step_ctx is None"
        );
    }
}

// ─── issue #18: `TaskSpec.initial_directive` `Value` pass-through ──────────
#[cfg(test)]
mod initial_directive_value_passthrough_tests {
    use super::*;

    async fn seeded_engine(initial_directive: Value) -> (Engine, CapToken, StepId) {
        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: "planner".to_string(),
                    initial_directive,
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        (engine, op_token, task_id)
    }

    /// Mint + register a `Role::Worker` token the same way
    /// `dispatch_attempt_with` does — `fetch_prompt` is worker-verb-gated.
    async fn mint_worker_token(engine: &Engine, task_id: &StepId) -> CapToken {
        let worker_token = engine.signer().session(
            format!("worker-of-{task_id}"),
            Role::Worker,
            vec!["*".into()],
            Duration::from_secs(600),
        );
        let fp = worker_token.fingerprint();
        let record = CapTokenRecord::from_worker_token(worker_token.clone(), task_id.clone());
        engine
            .with_state("test.mint_worker", move |s| {
                s.tokens.insert(fp, record);
            })
            .await
            .expect("mint worker token");
        worker_token
    }

    /// `EngineDispatcher::dispatch` no longer stringifies the evaluated
    /// `Step.in` value before seeding `TaskSpec.initial_directive` — an
    /// Object seed must round-trip through `start_task` /
    /// `read_task_state` byte-for-byte as the same `Value::Object`, not a
    /// JSON-stringified `Value::String`.
    #[tokio::test]
    async fn object_seed_passes_through_task_spec_unchanged() {
        let seed = serde_json::json!({"key": "value"});
        let (engine, token, task_id) = seeded_engine(seed.clone()).await;
        let state = engine
            .read_task_state(&token, &task_id)
            .await
            .expect("read_task_state");
        assert_eq!(
            state.spec.initial_directive, seed,
            "TaskSpec.initial_directive must equal the raw Object seed, not a stringified copy"
        );
    }

    /// `Engine::fetch_prompt` returns the `Value` end-to-end (issue #18):
    /// an Object seed stays a `Value::Object` and is not stringified in
    /// the engine layer. The Worker HTTP boundary
    /// (`fetch_worker_payload*`) is what performs the render down to a
    /// JSON literal `String` for `WorkerPayload.prompt`.
    #[tokio::test]
    async fn object_seed_passes_through_fetch_prompt_as_value() {
        let seed = serde_json::json!({"key": "value"});
        let (engine, _token, task_id) = seeded_engine(seed.clone()).await;
        let worker_token = mint_worker_token(&engine, &task_id).await;
        let prompt = engine
            .fetch_prompt(&worker_token, &task_id)
            .await
            .expect("fetch_prompt");
        assert_eq!(
            prompt, seed,
            "fetch_prompt must return the raw Object Value, not a stringified copy"
        );
    }

    /// The Worker HTTP boundary is the render point: `fetch_worker_payload*`
    /// coerces the stored `Value` down to `WorkerPayload.prompt: String`
    /// (JSON-literal shape for non-strings). Verifies the boundary render
    /// stays intact for an Object seed.
    #[tokio::test]
    async fn object_seed_renders_as_json_literal_at_worker_payload_boundary() {
        let seed = serde_json::json!({"key": "value"});
        let (engine, _token, task_id) = seeded_engine(seed).await;
        let worker_token = mint_worker_token(&engine, &task_id).await;
        let payload = engine
            .fetch_worker_payload(&worker_token, &task_id)
            .await
            .expect("fetch_worker_payload");
        assert_eq!(
            payload.prompt, r#"{"key":"value"}"#,
            "WorkerPayload.prompt must be the JSON literal String render of the Value seed"
        );
    }

    /// A `String` seed is unaffected — still passes through verbatim, both
    /// as the `TaskSpec.initial_directive` `Value` and as the Worker
    /// `fetch_prompt` return (issue #18 Invariant 2).
    #[tokio::test]
    async fn string_seed_passes_through_unchanged() {
        let (engine, token, task_id) = seeded_engine(serde_json::json!("do the thing")).await;
        let state = engine
            .read_task_state(&token, &task_id)
            .await
            .expect("read_task_state");
        assert_eq!(
            state.spec.initial_directive,
            serde_json::json!("do the thing")
        );
        let worker_token = mint_worker_token(&engine, &task_id).await;
        let prompt = engine
            .fetch_prompt(&worker_token, &task_id)
            .await
            .expect("fetch_prompt");
        assert_eq!(prompt, serde_json::json!("do the thing"));
    }
}

/// GH #31: `fetch_worker_payload{,_trusted}`'s size-threshold branch
/// between inline (`WorkerPayload.system`) and by-reference
/// (`WorkerPayload.system_ref`) delivery, plus the `bake_worker_system_prompt`
/// `agent_render_sizes` bookkeeping that feeds `agent_last_rendered_size`.
#[cfg(test)]
mod system_ref_threshold_tests {
    use super::*;

    async fn seeded_engine_with_cfg(cfg: EngineCfg) -> (Engine, CapToken, StepId) {
        let engine = Engine::new(cfg);
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: "planner".to_string(),
                    initial_directive: serde_json::json!("do the thing"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        (engine, op_token, task_id)
    }

    /// Same worker-token-minting fixture as
    /// `initial_directive_value_passthrough_tests::mint_worker_token`
    /// (kept local to this module — the two `mod`s do not share private
    /// helpers across `cfg(test)` boundaries).
    async fn mint_worker_token(engine: &Engine, task_id: &StepId) -> CapToken {
        let worker_token = engine.signer().session(
            format!("worker-of-{task_id}"),
            Role::Worker,
            vec!["*".into()],
            Duration::from_secs(600),
        );
        let fp = worker_token.fingerprint();
        let record = CapTokenRecord::from_worker_token(worker_token.clone(), task_id.clone());
        engine
            .with_state("test.mint_worker", move |s| {
                s.tokens.insert(fp, record);
            })
            .await
            .expect("mint worker token");
        worker_token
    }

    /// Under-threshold: `system` stays inline, `system_ref` stays `None`.
    #[tokio::test]
    async fn under_threshold_stays_inline() {
        let (engine, _op_token, task_id) = seeded_engine_with_cfg(EngineCfg::default()).await;
        let worker_token = mint_worker_token(&engine, &task_id).await;
        let rendered = "a short system prompt".to_string();
        engine
            .bake_worker_system_prompt(&task_id, 1, Some(rendered.clone()))
            .await
            .expect("bake");
        let payload = engine
            .fetch_worker_payload(&worker_token, &task_id)
            .await
            .expect("fetch_worker_payload");
        assert_eq!(payload.system, Some(rendered));
        assert!(payload.system_ref.is_none());
    }

    /// Over-threshold: `system` is cleared and `system_ref` is populated
    /// with a `sha256` matching the known input string. Exercises
    /// `fetch_worker_payload_trusted` (the `_trusted` sibling must be
    /// behaviorally identical to `fetch_worker_payload`).
    #[tokio::test]
    async fn over_threshold_switches_to_system_ref_with_matching_sha256() {
        let mut cfg = EngineCfg::default();
        cfg.system_ref.threshold_bytes = 16;
        cfg.system_ref.mode = crate::types::SystemRefMode::File;
        cfg.system_ref.store_dir =
            std::env::temp_dir().join(format!("mse-system-ref-test-{}", crate::types::now_unix()));
        let (engine, _op_token, task_id) = seeded_engine_with_cfg(cfg).await;
        let rendered =
            "this system prompt is deliberately longer than the 16 byte threshold".to_string();
        engine
            .bake_worker_system_prompt(&task_id, 1, Some(rendered.clone()))
            .await
            .expect("bake");
        let payload = engine
            .fetch_worker_payload_trusted(&task_id)
            .await
            .expect("fetch_worker_payload_trusted");
        assert!(
            payload.system.is_none(),
            "over-threshold response must not also inline `system`"
        );
        let system_ref = payload
            .system_ref
            .expect("over-threshold response must populate system_ref");
        assert_eq!(system_ref.size_bytes, rendered.len() as u64);
        assert_eq!(system_ref.mode, crate::types::SystemRefMode::File);
        use sha2::Digest;
        let expected_sha256 = hex::encode(sha2::Sha256::digest(rendered.as_bytes()));
        assert_eq!(system_ref.sha256, expected_sha256);
        assert!(system_ref.uri.starts_with("file://"));
        let written = tokio::fs::read_to_string(system_ref.uri.trim_start_matches("file://"))
            .await
            .expect("File mode must have written the referenced path");
        assert_eq!(written, rendered);
    }

    /// `Http` mode never writes a file — `system_ref.uri` is the bare path
    /// the engine can construct on its own, scheme/host-free.
    #[tokio::test]
    async fn over_threshold_http_mode_constructs_path_only_uri() {
        let mut cfg = EngineCfg::default();
        cfg.system_ref.threshold_bytes = 16;
        cfg.system_ref.mode = crate::types::SystemRefMode::Http;
        let (engine, _op_token, task_id) = seeded_engine_with_cfg(cfg).await;
        let worker_token = mint_worker_token(&engine, &task_id).await;
        let rendered =
            "this system prompt is deliberately longer than the 16 byte threshold".to_string();
        engine
            .bake_worker_system_prompt(&task_id, 1, Some(rendered))
            .await
            .expect("bake");
        let payload = engine
            .fetch_worker_payload(&worker_token, &task_id)
            .await
            .expect("fetch_worker_payload");
        let system_ref = payload.system_ref.expect("system_ref must be populated");
        assert_eq!(system_ref.mode, crate::types::SystemRefMode::Http);
        assert_eq!(
            system_ref.uri,
            format!("/v1/worker/prompt/system?task_id={task_id}&attempt=1")
        );
    }

    /// `bake_worker_system_prompt` records the render size keyed by agent
    /// name (last-write-wins), readable via `agent_last_rendered_size`.
    #[tokio::test]
    async fn bake_records_agent_render_size_last_write_wins() {
        let (engine, _op_token, task_id) = seeded_engine_with_cfg(EngineCfg::default()).await;
        assert_eq!(engine.agent_last_rendered_size("planner").await, None);
        engine
            .bake_worker_system_prompt(&task_id, 1, Some("a".repeat(10)))
            .await
            .expect("bake 1");
        assert_eq!(engine.agent_last_rendered_size("planner").await, Some(10));
        engine
            .bake_worker_system_prompt(&task_id, 2, Some("b".repeat(20)))
            .await
            .expect("bake 2");
        assert_eq!(
            engine.agent_last_rendered_size("planner").await,
            Some(20),
            "most-recently-observed size wins, not the largest"
        );
    }

    /// GH #83: `materialize_system_file` writes the baked system prompt
    /// unconditionally — a system well UNDER `threshold_bytes` still
    /// lands on disk, because a `{system_file}` template reference needs
    /// a real path regardless of size.
    #[tokio::test]
    async fn materialize_system_file_writes_under_threshold_system() {
        let mut cfg = EngineCfg::default();
        cfg.system_ref.store_dir =
            std::env::temp_dir().join(format!("mse-system-file-test-{}", crate::types::now_unix()));
        assert!(cfg.system_ref.threshold_bytes > 64, "fixture premise");
        let (engine, _op_token, task_id) = seeded_engine_with_cfg(cfg).await;
        let rendered = "a short system prompt".to_string();
        engine
            .bake_worker_system_prompt(&task_id, 1, Some(rendered.clone()))
            .await
            .expect("bake");
        let path = engine
            .materialize_system_file(&task_id, 1)
            .await
            .expect("materialize_system_file")
            .expect("baked system must yield a path");
        let written = tokio::fs::read_to_string(&path)
            .await
            .expect("materialized path must exist");
        assert_eq!(written, rendered);
    }

    /// GH #83: no baked system → `Ok(None)` (the Subprocess spawn path
    /// turns this into a fail-loud `SpawnError` when `{system_file}` is
    /// actually referenced).
    #[tokio::test]
    async fn materialize_system_file_none_when_nothing_baked() {
        let (engine, _op_token, task_id) = seeded_engine_with_cfg(EngineCfg::default()).await;
        let path = engine
            .materialize_system_file(&task_id, 1)
            .await
            .expect("materialize_system_file");
        assert!(path.is_none());
    }
}

/// subtask-4 / ST2 rework: `submit_output` / `submit_worker_result_trusted`'s
/// submit-time projection sink (`Engine::materialize_final_submission`) —
/// the Data-plane `OutputStore` dual-write plus the
/// `FileProjectionAdapter`-backed file materialize, both fail-open. See
/// the subtask-4 Tests this module covers inline on each test.
#[cfg(test)]
mod submit_time_projection_sink_tests {
    use super::*;
    use crate::core::agent_context::AgentContextView;
    use crate::store::output::{ContentRef, InMemoryOutputStore, OutputEvent};

    /// Starts a task under `agent`, returning `(engine, op_token, task_id,
    /// worker_token)` — same helper shape as the sibling test modules
    /// above (`initial_directive_value_passthrough_tests::seeded_engine` /
    /// `mint_worker_token`), duplicated locally per this file's
    /// established per-module convention.
    async fn seeded_task(agent: &str) -> (Engine, CapToken, StepId, CapToken) {
        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: agent.to_string(),
                    initial_directive: Value::String("go".into()),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        let worker_token = engine.signer().session(
            format!("worker-of-{task_id}"),
            Role::Worker,
            vec!["*".into()],
            Duration::from_secs(600),
        );
        let fp = worker_token.fingerprint();
        let record = CapTokenRecord::from_worker_token(worker_token.clone(), task_id.clone());
        engine
            .with_state("test.mint_worker", move |s| {
                s.tokens.insert(fp, record);
            })
            .await
            .expect("mint worker token");
        (engine, op_token, task_id, worker_token)
    }

    /// Sibling of [`seeded_task`] that lets a caller pin the engine's
    /// `EngineCfg.check_policy` before the engine is constructed — used
    /// by the `check_policy_*` regression tests below to exercise the
    /// three [`crate::core::config::CheckPolicy`] modes without touching
    /// the shared `seeded_task` helper (which every unrelated sink test
    /// depends on).
    async fn seeded_task_with_policy(
        agent: &str,
        policy: crate::core::config::CheckPolicy,
    ) -> (Engine, CapToken, StepId, CapToken) {
        let cfg = EngineCfg {
            check_policy: policy,
            ..EngineCfg::default()
        };
        let engine = Engine::new(cfg);
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: agent.to_string(),
                    initial_directive: Value::String("go".into()),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        let worker_token = engine.signer().session(
            format!("worker-of-{task_id}"),
            Role::Worker,
            vec!["*".into()],
            Duration::from_secs(600),
        );
        let fp = worker_token.fingerprint();
        let record = CapTokenRecord::from_worker_token(worker_token.clone(), task_id.clone());
        engine
            .with_state("test.mint_worker", move |s| {
                s.tokens.insert(fp, record);
            })
            .await
            .expect("mint worker token");
        (engine, op_token, task_id, worker_token)
    }

    /// Seeds `EngineState.agent_ctx[(task_id, attempt)].view` directly —
    /// the same snapshot `AgentContextMiddleware` writes at spawn time
    /// (see its module doc), stood up here without the full spawner
    /// stack so these tests can exercise `submit_output` in isolation.
    async fn seed_agent_context(engine: &Engine, task_id: &StepId, attempt: u32, work_dir: &str) {
        let task_id = task_id.clone();
        let work_dir = work_dir.to_string();
        engine
            .with_state("test.seed_agent_context", move |s| {
                s.agent_ctx.insert(
                    (task_id, attempt),
                    crate::core::state::AgentCtxEntry {
                        view: AgentContextView {
                            work_dir: Some(work_dir),
                            ..Default::default()
                        },
                        policy: Default::default(),
                    },
                );
            })
            .await
            .expect("seed agent_ctx");
    }

    /// GH #27 (follow-up to #23): seeds `EngineState.agent_ctx` with an
    /// arbitrary `work_dir` / `project_root` pair (either may be `None`),
    /// unlike [`seed_agent_context`] (which only ever sets `work_dir`) —
    /// lets these tests exercise `ProjectionPlacement::resolve_root`'s
    /// fallback in both directions.
    async fn seed_agent_context_roots(
        engine: &Engine,
        task_id: &StepId,
        attempt: u32,
        work_dir: Option<&str>,
        project_root: Option<&str>,
    ) {
        let task_id = task_id.clone();
        let work_dir = work_dir.map(str::to_string);
        let project_root = project_root.map(str::to_string);
        engine
            .with_state("test.seed_agent_context_roots", move |s| {
                s.agent_ctx.insert(
                    (task_id, attempt),
                    crate::core::state::AgentCtxEntry {
                        view: AgentContextView {
                            work_dir,
                            project_root,
                            ..Default::default()
                        },
                        policy: Default::default(),
                    },
                );
            })
            .await
            .expect("seed agent_ctx");
    }

    /// GH #27 (follow-up to #23): seeds `EngineState.projection_placements`
    /// directly — the same snapshot `EngineDispatcher::dispatch` stashes
    /// at dispatch time (mirroring [`seed_step_naming`]'s contract) — so
    /// these tests can exercise a declared `ProjectionPlacement` without
    /// driving a real `Compiler::compile`.
    async fn seed_projection_placement(
        engine: &Engine,
        task_id: &StepId,
        placement: crate::core::projection_placement::ProjectionPlacement,
    ) {
        let task_id = task_id.clone();
        let placement = Arc::new(placement);
        engine
            .with_state("test.seed_projection_placement", move |s| {
                s.projection_placements.insert(task_id, placement);
            })
            .await
            .expect("seed projection_placements");
    }

    /// GH #23 subtask-2: builds a fixture
    /// [`crate::core::step_naming::StepNaming`] table declaring `producer`
    /// → `canonical` (`AgentMeta.projection_name`), then seeds it into
    /// `EngineState.step_namings` for `task_id` — the same snapshot
    /// `EngineDispatcher::dispatch` stashes at dispatch time
    /// (`blueprint.rs`'s "construct once, read many" contract), stood up
    /// here without the full Blueprint-compile path so these tests can
    /// exercise the canonical-sink resolution in isolation.
    async fn seed_step_naming(engine: &Engine, task_id: &StepId, producer: &str, canonical: &str) {
        use crate::blueprint::{
            current_schema_version, AgentDef, AgentKind, AgentMeta, Blueprint, BlueprintMetadata,
            CompilerHints, CompilerStrategy,
        };
        use crate::core::step_naming::StepNaming;
        use mlua_flow_ir::{Expr, Node};

        let flow = Node::Step {
            ref_: producer.to_string(),
            in_: Expr::Path {
                at: "$.in".parse().expect("literal test path: $.in"),
            },
            out: Expr::Path {
                at: format!("$.{producer}_out")
                    .parse()
                    .expect("literal test path"),
            },
        };
        let bp = Blueprint {
            schema_version: current_schema_version(),
            id: "sink-canonical-ut".into(),
            flow,
            agents: vec![AgentDef {
                name: producer.to_string(),
                kind: AgentKind::RustFn,
                spec: serde_json::json!({ "fn_id": producer }),
                profile: None,
                meta: Some(AgentMeta {
                    projection_name: Some(canonical.to_string()),
                    ..Default::default()
                }),
                runner: None,
                runner_ref: None,
                verdict: None,
                lints: None,
            }],
            operators: vec![],
            metas: vec![],
            hints: CompilerHints::default(),
            strategy: CompilerStrategy::default(),
            metadata: BlueprintMetadata::default(),
            spawner_hints: Default::default(),
            default_agent_kind: AgentKind::Operator,
            default_operator_kind: None,
            default_init_ctx: None,
            default_agent_ctx: None,
            default_context_policy: None,
            projection_placement: None,
            audits: vec![],
            degradation_policy: None,
            runners: vec![],
            default_runner: None,
            subprocesses: vec![],
            check_policy: None,
            blueprint_ref_includes: Vec::new(),
        };
        let (naming, warnings) = StepNaming::from_blueprint(&bp).expect("no collision");
        assert!(warnings.is_empty(), "single-step fixture has no collisions");
        let naming = Arc::new(naming);
        let task_id = task_id.clone();
        engine
            .with_state("test.seed_step_naming", move |s| {
                s.step_namings.insert(task_id, naming);
            })
            .await
            .expect("seed step_namings");
    }

    fn final_event(value: Value, ok: bool) -> crate::worker::output::OutputEvent {
        crate::worker::output::OutputEvent::Final {
            content: crate::worker::output::ContentRef::Inline { value },
            ok,
        }
    }

    /// Subtask 4 Test #2: `submit_output`'s `Final` writes
    /// `<root>/workspace/tasks/<task_id>/ctx/<agent>.md`, content matching
    /// the submitted value.
    #[tokio::test]
    async fn submit_output_final_materializes_file_when_work_dir_resolved() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, worker_token) = seeded_task("planner").await;
        seed_agent_context(&engine, &task_id, 1, &dir.path().to_string_lossy()).await;

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"plan": "do it"}), true),
            )
            .await
            .expect("submit_output");

        let expected_file = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/planner.md");
        assert!(
            expected_file.exists(),
            "materialized submission file missing at {expected_file:?}"
        );
        let body = std::fs::read_to_string(expected_file).unwrap();
        assert!(body.contains(r#""plan": "do it""#), "body: {body}");
    }

    /// Subtask 4 Test #3: `work_dir` unresolved (no `agent_ctx`
    /// snapshot for this `(task_id, attempt)`) — submit still succeeds,
    /// fail-open, no file.
    #[tokio::test]
    async fn submit_output_final_skips_file_when_root_unresolved() {
        let (engine, _op, task_id, worker_token) = seeded_task("planner").await;
        // No seed_agent_context call — root is unresolved.

        let result = engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!("hi"), true),
            )
            .await;
        assert!(
            result.is_ok(),
            "submit must succeed even with no resolvable root (fail-open, Invariant 1)"
        );
    }

    /// Regression for the check_policy cascade: the default
    /// [`crate::core::config::CheckPolicy::Warn`] preserves the
    /// pre-`CheckPolicy` fail-open semantics — a submit whose root is
    /// unresolved still succeeds. Byte-compat with
    /// `submit_output_final_skips_file_when_root_unresolved`; this test
    /// pins the mode explicitly so a future default change to
    /// `Strict` (silent breakage) is caught here.
    #[tokio::test]
    async fn submit_output_final_check_policy_warn_preserves_fail_open() {
        let (engine, _op, task_id, worker_token) =
            seeded_task_with_policy("planner", crate::core::config::CheckPolicy::Warn).await;

        let result = engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!("hi"), true),
            )
            .await;
        assert!(
            result.is_ok(),
            "Warn mode preserves fail-open: submit must succeed when root unresolved"
        );
    }

    /// Regression for the check_policy cascade:
    /// [`crate::core::config::CheckPolicy::Strict`] surfaces the "no
    /// work_dir/project_root resolved" fail-open condition as an
    /// [`EngineError::CheckPolicyStrict`], letting a caller who has
    /// opted in fail fast instead of proceeding with a partially-
    /// realized submission. The error's `context` identifies the call
    /// site (`"file materialize"`), and `message` preserves the
    /// pre-`CheckPolicy` warn literal verbatim (log-parse compat).
    #[tokio::test]
    async fn submit_output_final_check_policy_strict_surfaces_error_when_root_unresolved() {
        let (engine, _op, task_id, worker_token) =
            seeded_task_with_policy("planner", crate::core::config::CheckPolicy::Strict).await;

        let err = engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!("hi"), true),
            )
            .await
            .expect_err("Strict mode must return an error when root unresolved");
        match err {
            EngineError::CheckPolicyStrict { context, message } => {
                assert!(
                    context.contains("file materialize"),
                    "context must identify the call site: {context}"
                );
                assert!(
                    message.contains("no work_dir/project_root resolved"),
                    "message must preserve the warn-log literal for log-parse compat: {message}"
                );
            }
            other => panic!(
                "expected EngineError::CheckPolicyStrict, got a different variant: {other:?}"
            ),
        }
    }

    /// Regression for the check_policy cascade:
    /// [`crate::core::config::CheckPolicy::Silent`] returns `Ok(())` (
    /// like `Warn`) without surfacing an error. The log-suppression side
    /// of `Silent` (no `tracing::warn!`) is enforced at the call site
    /// via the `if !matches!(policy, Silent) { warn!(...) }` guard —
    /// verifying tracing output shape here would couple the test to a
    /// subscriber setup, so the assertion is limited to the error-
    /// return semantics (matches the helper unit tests in
    /// `check_policy_helper_tests`).
    #[tokio::test]
    async fn submit_output_final_check_policy_silent_returns_ok_when_root_unresolved() {
        let (engine, _op, task_id, worker_token) =
            seeded_task_with_policy("planner", crate::core::config::CheckPolicy::Silent).await;

        let result = engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!("hi"), true),
            )
            .await;
        assert!(
            result.is_ok(),
            "Silent mode returns Ok(()) at the error surface: submit must succeed"
        );
    }

    /// Subtask 4 Test #4 (file half): re-submitting under the same
    /// `(task_id, agent)` overwrites the materialized file with the
    /// latest value.
    #[tokio::test]
    async fn resubmit_overwrites_materialized_file_with_latest() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, worker_token) = seeded_task("planner").await;
        seed_agent_context(&engine, &task_id, 1, &dir.path().to_string_lossy()).await;

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!("first"), true),
            )
            .await
            .expect("first submit");
        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!("second"), true),
            )
            .await
            .expect("second submit");

        let expected_file = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/planner.md");
        let body = std::fs::read_to_string(expected_file).unwrap();
        assert!(body.contains("second"), "body must reflect latest: {body}");
        assert!(
            !body.contains("first"),
            "body must not carry the stale value: {body}"
        );
    }

    /// GH #27 (follow-up to #23): the byte-compat default
    /// `ProjectionPlacement` (`root_preference = WorkDir`) falls back to
    /// `project_root` when `work_dir` is absent — the same fallback
    /// [`crate::core::projection_placement::ProjectionPlacement::resolve_root`]
    /// now performs for every one of the "3 path" call sites, this one
    /// exercised at the submit-sink layer.
    #[tokio::test]
    async fn submit_output_final_falls_back_to_project_root_when_work_dir_absent() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, worker_token) = seeded_task("planner").await;
        seed_agent_context_roots(
            &engine,
            &task_id,
            1,
            None,
            Some(&dir.path().to_string_lossy()),
        )
        .await;

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"plan": "via project_root"}), true),
            )
            .await
            .expect("submit_output");

        let expected_file = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/planner.md");
        assert!(
            expected_file.exists(),
            "materialized submission file missing at {expected_file:?} \
             (work_dir absent must fall back to project_root)"
        );
    }

    /// GH #27 (follow-up to #23): a declared `ProjectionPlacement`
    /// (`root_preference = ProjectRoot`, custom `dir_template`) changes
    /// BOTH which root is preferred (project_root wins even though
    /// work_dir is also present) AND the target directory layout — proof
    /// the submit sink consults the snapshotted resolver rather than a
    /// hardcoded layout.
    #[tokio::test]
    async fn submit_output_final_uses_declared_projection_placement() {
        let work_dir = tempfile::TempDir::new().unwrap();
        let project_root = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, worker_token) = seeded_task("planner").await;
        seed_agent_context_roots(
            &engine,
            &task_id,
            1,
            Some(&work_dir.path().to_string_lossy()),
            Some(&project_root.path().to_string_lossy()),
        )
        .await;
        seed_projection_placement(
            &engine,
            &task_id,
            crate::core::projection_placement::ProjectionPlacement {
                root_preference: crate::core::projection_placement::RootPreference::ProjectRoot,
                dir_template: "custom/{task_id}/out".to_string(),
            },
        )
        .await;

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"plan": "via custom placement"}), true),
            )
            .await
            .expect("submit_output");

        let expected_file = project_root
            .path()
            .join("custom")
            .join(task_id.as_str())
            .join("out/planner.md");
        assert!(
            expected_file.exists(),
            "materialized submission file missing at custom placement target {expected_file:?}"
        );
        let unexpected_file = work_dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/planner.md");
        assert!(
            !unexpected_file.exists(),
            "declared root_preference=ProjectRoot must not fall back to work_dir: {unexpected_file:?}"
        );
    }

    /// Subtask 4 Invariant 3 / crux requirement #3: when
    /// [`Engine::set_output_store`] wires a Data-plane [`crate::store::output::OutputStore`],
    /// `submit_output`'s `Final` dual-writes into it under
    /// `producer_agent = TaskState.spec.agent` — the store becomes
    /// queryable via `get_latest_by_name`, independent of whether a root
    /// resolved for the file half.
    #[tokio::test]
    async fn submit_output_final_dual_writes_into_configured_output_store() {
        let (engine, _op, task_id, worker_token) = seeded_task("reviewer").await;
        let data_store: Arc<dyn crate::store::output::OutputStore> =
            Arc::new(InMemoryOutputStore::new());
        engine.set_output_store(data_store.clone());

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"verdict": "pass"}), true),
            )
            .await
            .expect("submit_output");

        let record = data_store
            .get_latest_by_name("reviewer")
            .await
            .expect("dual-written record");
        match record.event {
            OutputEvent::Final { content, ok } => {
                assert!(ok);
                match content {
                    ContentRef::Inline { value } => {
                        assert_eq!(value, serde_json::json!({"verdict": "pass"}));
                    }
                    other => panic!("expected Inline content, got {other:?}"),
                }
            }
            other => panic!("expected Final event, got {other:?}"),
        }
    }

    /// GH #34 subtask-3 gap fix: an `Artifact` event submitted via
    /// `submit_output` dual-writes into a wired Data-plane `OutputStore`
    /// under its OWN `name`, verbatim — mirrors
    /// `submit_output_final_dual_writes_into_configured_output_store`
    /// above, but for the `Artifact` variant.
    #[tokio::test]
    async fn submit_output_artifact_dual_writes_into_configured_output_store() {
        let (engine, _op, task_id, worker_token) = seeded_task("echo").await;
        let data_store: Arc<dyn crate::store::output::OutputStore> =
            Arc::new(InMemoryOutputStore::new());
        engine.set_output_store(data_store.clone());

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                OutputEvent::Artifact {
                    name: "audit:echo".to_string(),
                    content: ContentRef::Inline {
                        value: serde_json::json!({"finding": "clean"}),
                    },
                },
            )
            .await
            .expect("submit_output");

        let record = data_store
            .get_latest_by_name("audit:echo")
            .await
            .expect("dual-written artifact record");
        match record.event {
            OutputEvent::Artifact { name, content } => {
                assert_eq!(name, "audit:echo");
                match content {
                    ContentRef::Inline { value } => {
                        assert_eq!(value, serde_json::json!({"finding": "clean"}));
                    }
                    other => panic!("expected Inline content, got {other:?}"),
                }
            }
            other => panic!("expected Artifact event, got {other:?}"),
        }
        // The `Artifact` dual-write must never collide with / overwrite
        // the producing step's own `Final` name — `submit_output` never
        // materialized a `Final` here, so `"echo"` must stay unresolved.
        assert!(
            data_store.get_latest_by_name("echo").await.is_err(),
            "artifact write must not fabricate a record under the raw producer_agent name"
        );
    }

    /// Invariant 1 (fail-open) for `Artifact`, mirroring
    /// `submit_output_final_skips_file_when_root_unresolved`'s Final-side
    /// coverage: no `OutputStore` wired at all — submit still succeeds.
    #[tokio::test]
    async fn submit_output_artifact_is_fail_open_when_no_output_store_configured() {
        let (engine, _op, task_id, worker_token) = seeded_task("echo").await;

        let result = engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                OutputEvent::Artifact {
                    name: "audit:echo".to_string(),
                    content: ContentRef::Inline {
                        value: serde_json::json!("finding"),
                    },
                },
            )
            .await;
        assert!(
            result.is_ok(),
            "submit must succeed even with no OutputStore wired (fail-open, Invariant 1)"
        );
    }

    /// `submit_worker_result_trusted` (the `/v1/worker/submit` short-handle
    /// path) triggers the exact same sink as `submit_output` — parity
    /// across both worker-submit entry points.
    #[tokio::test]
    async fn submit_worker_result_trusted_also_triggers_projection_sink() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, _worker_token) = seeded_task("planner").await;
        seed_agent_context(&engine, &task_id, 1, &dir.path().to_string_lossy()).await;
        let data_store: Arc<dyn crate::store::output::OutputStore> =
            Arc::new(InMemoryOutputStore::new());
        engine.set_output_store(data_store.clone());

        engine
            .submit_worker_result_trusted(
                &task_id,
                1,
                serde_json::json!("trusted-value"),
                SubmitOutcome::Pass,
            )
            .await
            .expect("submit_worker_result_trusted");

        let expected_file = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/planner.md");
        assert!(expected_file.exists());
        let record = data_store
            .get_latest_by_name("planner")
            .await
            .expect("dual-written record");
        assert!(matches!(record.event, OutputEvent::Final { ok: true, .. }));
    }

    /// GH #23 subtask-2 (canonical sink): a declared `projection_name`
    /// (`AgentMeta.projection_name`, surfaced via `StepNaming`) redirects
    /// `submit_output`'s Final canonical sink — both the Data-plane
    /// dual-write name and the materialized file stem resolve to the
    /// canonical name, not the raw `producer_agent`.
    #[tokio::test]
    async fn submit_output_final_uses_canonical_name_when_step_naming_declares_one() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, worker_token) = seeded_task("reviewer").await;
        seed_agent_context(&engine, &task_id, 1, &dir.path().to_string_lossy()).await;
        seed_step_naming(&engine, &task_id, "reviewer", "verdict-final").await;
        let data_store: Arc<dyn crate::store::output::OutputStore> =
            Arc::new(InMemoryOutputStore::new());
        engine.set_output_store(data_store.clone());

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"verdict": "pass"}), true),
            )
            .await
            .expect("submit_output");

        let record = data_store
            .get_latest_by_name("verdict-final")
            .await
            .expect("dual-written record under canonical name");
        assert!(matches!(record.event, OutputEvent::Final { ok: true, .. }));
        assert!(
            data_store.get_latest_by_name("reviewer").await.is_err(),
            "raw producer_agent name must not be written once canonical resolves"
        );

        let expected_file = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/verdict-final.md");
        assert!(
            expected_file.exists(),
            "materialized file stem must be canonical at {expected_file:?}"
        );
    }

    /// GH #23 subtask-2: no `StepNaming` table snapshotted for this
    /// `task_id` (the pre-GH-#23 / no-`with_step_naming` path) is a
    /// defensive fail-open — the canonical sink falls back to the raw
    /// `producer_agent`, byte-identical to
    /// `submit_output_final_dual_writes_into_configured_output_store`
    /// above (which never calls `seed_step_naming`).
    #[tokio::test]
    async fn submit_output_final_falls_back_to_producer_agent_when_no_step_naming_table() {
        let (engine, _op, task_id, worker_token) = seeded_task("reviewer").await;
        let data_store: Arc<dyn crate::store::output::OutputStore> =
            Arc::new(InMemoryOutputStore::new());
        engine.set_output_store(data_store.clone());

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"verdict": "pass"}), true),
            )
            .await
            .expect("submit_output");

        let record = data_store
            .get_latest_by_name("reviewer")
            .await
            .expect("fail-open dual-write under raw producer_agent name");
        assert!(matches!(record.event, OutputEvent::Final { ok: true, .. }));
    }

    /// GH #23 subtask-2 (Layer 2): `OutputStore::get_latest_by_name_in_run`
    /// resolves the value `submit_output` dual-wrote for this exact
    /// `(task_id, attempt)` run, independent of `get_latest_by_name`'s
    /// cross-Run race (two Runs sharing a producer name never bleed into
    /// each other through the Run-scoped accessor).
    #[tokio::test]
    async fn submit_output_final_is_resolvable_via_run_scoped_lookup() {
        let (engine, _op, task_id, worker_token) = seeded_task("reviewer").await;
        let data_store: Arc<dyn crate::store::output::OutputStore> =
            Arc::new(InMemoryOutputStore::new());
        engine.set_output_store(data_store.clone());

        engine
            .submit_output(
                &worker_token,
                &task_id,
                1,
                final_event(serde_json::json!({"verdict": "pass"}), true),
            )
            .await
            .expect("submit_output");

        let record = data_store
            .get_latest_by_name_in_run(task_id.as_str(), 1, "reviewer")
            .await
            .expect("run-scoped lookup resolves the dual-written record");
        assert!(matches!(record.event, OutputEvent::Final { ok: true, .. }));

        // A different attempt of the same task must not resolve — the
        // Run-scoped lookup does not fall back across attempts.
        assert!(
            data_store
                .get_latest_by_name_in_run(task_id.as_str(), 2, "reviewer")
                .await
                .is_err(),
            "a different attempt must not resolve the same-named record"
        );
    }

    // ─── staged part file materialize ───

    /// Staging a part with a resolved `work_dir` writes
    /// `<work_dir>/workspace/tasks/<task_id>/ctx/<name>` with the part's
    /// content RAW (no front matter / fenced wrapper).
    #[tokio::test]
    async fn stage_artifact_materializes_part_file_when_work_dir_resolved() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, _worker_token) = seeded_task("planner").await;
        seed_agent_context(&engine, &task_id, 1, &dir.path().to_string_lossy()).await;

        engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "plan.md".to_string(),
                serde_json::json!("# Plan\n\nstep one\n"),
            )
            .await
            .expect("stage artifact");

        let expected_file = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("ctx/plan.md");
        assert!(
            expected_file.exists(),
            "materialized part file missing at {expected_file:?}"
        );
        let body = std::fs::read_to_string(expected_file).unwrap();
        // Raw — no YAML front matter / fenced-json wrapper.
        assert_eq!(body, "# Plan\n\nstep one\n");
    }

    /// No resolvable root + `Warn` — staging still
    /// succeeds (fail-open), and no part file is written.
    #[tokio::test]
    async fn stage_artifact_check_policy_warn_skips_part_file_when_root_unresolved() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, _worker_token) =
            seeded_task_with_policy("planner", crate::core::config::CheckPolicy::Warn).await;
        // No seed_agent_context — root unresolved.

        let result = engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "plan.md".to_string(),
                serde_json::json!("x"),
            )
            .await;
        assert!(
            result.is_ok(),
            "Warn mode preserves fail-open: stage must succeed when root unresolved"
        );
        assert!(
            !dir.path().join("workspace").exists(),
            "no part file may be materialized when root is unresolved"
        );
    }

    /// No resolvable root + `Strict` — staging surfaces
    /// the fail-open condition as an [`EngineError::CheckPolicyStrict`],
    /// its message identifying the "part file materialize" call site.
    #[tokio::test]
    async fn stage_artifact_check_policy_strict_surfaces_error_when_root_unresolved() {
        let (engine, _op, task_id, _worker_token) =
            seeded_task_with_policy("planner", crate::core::config::CheckPolicy::Strict).await;

        let err = engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "plan.md".to_string(),
                serde_json::json!("x"),
            )
            .await
            .expect_err("Strict mode must return an error when root unresolved");
        match err {
            EngineError::CheckPolicyStrict { context, message } => {
                assert!(
                    context.contains("part file materialize"),
                    "context must identify the call site: {context}"
                );
                assert!(
                    message.contains("part file materialize"),
                    "message must identify the part-file sink: {message}"
                );
                assert!(
                    message.contains("no work_dir/project_root resolved"),
                    "message must preserve the warn-log literal: {message}"
                );
            }
            other => panic!(
                "expected EngineError::CheckPolicyStrict, got a different variant: {other:?}"
            ),
        }
    }

    /// A path-traversal `name` (`../evil.md`) with a
    /// resolved root — the name guard fails the write, but fail-open keeps
    /// the stage succeeding, and nothing is written outside the ctx dir.
    #[tokio::test]
    async fn stage_artifact_traversal_name_is_fail_open_and_writes_nothing() {
        let dir = tempfile::TempDir::new().unwrap();
        let (engine, _op, task_id, _worker_token) = seeded_task("planner").await;
        seed_agent_context(&engine, &task_id, 1, &dir.path().to_string_lossy()).await;

        let result = engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "../evil.md".to_string(),
                serde_json::json!("pwned"),
            )
            .await;
        assert!(
            result.is_ok(),
            "default (Warn) policy is fail-open even on a rejected part name"
        );
        // The escaped target (ctx dir's parent) must not have been written.
        let escaped = dir
            .path()
            .join("workspace/tasks")
            .join(task_id.as_str())
            .join("evil.md");
        assert!(
            !escaped.exists(),
            "a traversal name must never write outside the ctx dir: {escaped:?}"
        );
    }
}

/// GH #36 ST1: named multi-part worker output. Covers (a) the pure
/// `fold_final_and_parts` assembly `dispatch_attempt_with`'s Final-pull
/// delegates to, (b) `stage_worker_artifact_trusted`'s per-attempt
/// isolation on `EngineState.output_store` / `.worker_artifact_names` (the
/// same `HashMap<(StepId, u32), _>` key shape `submit_worker_result_trusted`
/// uses — a fresh attempt is a fresh key, so nothing to explicitly "clean
/// up"), and (c) the allowlist behavior that keeps a non-opt-in `Artifact`
/// producer (e.g. `AfterRunAuditMiddleware`) from being folded in.
#[cfg(test)]
mod named_multi_part_worker_output_tests {
    use super::*;
    use crate::worker::output::{ContentRef, OutputEvent};

    fn artifact(name: &str, value: Value) -> OutputEvent {
        OutputEvent::Artifact {
            name: name.to_string(),
            content: ContentRef::Inline { value },
        }
    }

    fn final_ev(value: Value, ok: bool) -> OutputEvent {
        OutputEvent::Final {
            content: ContentRef::Inline { value },
            ok,
        }
    }

    fn names(list: &[&str]) -> Vec<String> {
        list.iter().map(|s| s.to_string()).collect()
    }

    /// Two staged parts (both in `staged_names`) + a `Final` fold into
    /// `{"out", "parts"}`, each value carried through verbatim.
    #[test]
    fn fold_final_and_parts_assembles_out_and_parts_shape() {
        let tail = vec![
            artifact("summary", serde_json::json!("the summary")),
            artifact("diff", serde_json::json!({"lines": 3})),
            final_ev(serde_json::json!("final text"), true),
        ];
        let staged = names(&["summary", "diff"]);
        let (value, ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Lenient).expect("Final present");
        assert!(ok);
        assert_eq!(
            value,
            serde_json::json!({
                "out": "final text",
                "parts": {
                    "summary": "the summary",
                    "diff": {"lines": 3},
                }
            })
        );
    }

    /// Zero staged parts: the value is exactly the plain `Final` value — no
    /// `{"out", "parts"}` wrapping. This is the back-compat guarantee (GH
    /// #36 must not change the shape for a worker that never POSTs to
    /// `/v1/worker/artifact`).
    #[test]
    fn fold_final_and_parts_with_no_parts_returns_plain_final_value() {
        let tail = vec![final_ev(serde_json::json!("plain value"), true)];
        let (value, ok) =
            fold_final_and_parts(&tail, &[], FoldParse::Lenient).expect("Final present");
        assert!(ok);
        assert_eq!(value, serde_json::json!("plain value"));
    }

    /// The same staged part `name` appearing twice in one attempt: the
    /// LATER (tail-order) value wins — `parts` is a `Map`, not an
    /// accumulating list.
    #[test]
    fn fold_final_and_parts_same_name_twice_last_write_wins() {
        let tail = vec![
            artifact("a", serde_json::json!("first")),
            artifact("a", serde_json::json!("second")),
            final_ev(serde_json::json!("f"), true),
        ];
        let staged = names(&["a"]);
        let (value, _ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Lenient).expect("Final present");
        assert_eq!(
            value,
            serde_json::json!({"out": "f", "parts": {"a": "second"}})
        );
    }

    /// No `Final` anywhere in the tail (only staged parts, e.g. the worker
    /// crashed before submitting) — `None`, the caller's pre-existing "no
    /// Final in output_tail" error path.
    #[test]
    fn fold_final_and_parts_returns_none_when_no_final_present() {
        let tail = vec![artifact("a", serde_json::json!("v"))];
        let staged = names(&["a"]);
        assert!(fold_final_and_parts(&tail, &staged, FoldParse::Lenient).is_none());
    }

    /// An `Artifact` on the tail whose name is NOT in `staged_names` (e.g.
    /// `AfterRunAuditMiddleware`'s `"audit:<step_ref>"` sidecar finding on
    /// an audited step's own tail) must NOT be folded into `"parts"` — the
    /// value stays the plain `Final` value, exactly the pre-GH-#36
    /// behavior for every producer that isn't the worker's own
    /// `/v1/worker/artifact` staging. This is the regression this fold was
    /// almost shipped without (see `dispatch_attempt_with`'s doc).
    #[test]
    fn fold_final_and_parts_ignores_artifacts_outside_the_staged_allowlist() {
        let tail = vec![
            final_ev(serde_json::json!({"echoed": "hi"}), true),
            artifact("audit:echo", serde_json::json!({"finding": "clean"})),
        ];
        // `staged_names` empty: the worker itself never staged anything —
        // the audit sidecar Artifact must be ignored.
        let (value, ok) =
            fold_final_and_parts(&tail, &[], FoldParse::Lenient).expect("Final present");
        assert!(ok);
        assert_eq!(value, serde_json::json!({"echoed": "hi"}));
    }

    /// Mixed tail: one staged (allowlisted) part and one non-staged
    /// (audit-style) `Artifact` — only the staged one is folded in.
    #[test]
    fn fold_final_and_parts_folds_only_the_staged_subset_of_a_mixed_tail() {
        let tail = vec![
            artifact("summary", serde_json::json!("s")),
            artifact("audit:echo", serde_json::json!({"finding": "clean"})),
            final_ev(serde_json::json!("f"), true),
        ];
        let staged = names(&["summary"]);
        let (value, _ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Lenient).expect("Final present");
        assert_eq!(
            value,
            serde_json::json!({"out": "f", "parts": {"summary": "s"}})
        );
    }

    /// Lenient fold: a `Value::String` final body / staged part whose
    /// bytes parse as a JSON **container** folds structured with NO
    /// declaration — the default that makes `$.<step>.lanes` /
    /// `$.<step>.parts["plan-meta.json"].lanes` addressable across all
    /// three lanes (they all meet at this fold).
    #[test]
    fn lenient_fold_parses_container_strings_in_final_and_parts() {
        let tail = vec![
            artifact(
                "plan-meta.json",
                Value::String(r#"{"lanes":[{"id":1},{"id":2}]}"#.to_string()),
            ),
            final_ev(Value::String(r#"{"lanes":["a","b"]}"#.to_string()), true),
        ];
        let staged = names(&["plan-meta.json"]);
        let (value, ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Lenient).expect("Final present");
        assert!(ok);
        assert_eq!(
            value,
            serde_json::json!({
                "out": {"lanes": ["a", "b"]},
                "parts": {"plan-meta.json": {"lanes": [{"id": 1}, {"id": 2}]}},
            })
        );
    }

    /// Containers-only lock: scalar JSON (`true` / `42` / a quoted
    /// string / `null`), bare verdict tokens, and container-lookalikes
    /// that do not parse ALL keep folding as strings under `Lenient` — a
    /// scalar has no addressable interior, and parsing it would silently
    /// change `Eq` conds / verdict comparisons for tokens that happen to
    /// be valid JSON.
    #[test]
    fn lenient_fold_keeps_scalar_json_and_non_json_strings() {
        let tail = vec![
            artifact("verdict", Value::String("PASS".to_string())),
            artifact("bool", Value::String("true".to_string())),
            artifact("num", Value::String("42".to_string())),
            artifact("quoted", Value::String("\"quoted\"".to_string())),
            artifact("null", Value::String("null".to_string())),
            artifact("broken", Value::String("{not json".to_string())),
            final_ev(Value::String("PASS".to_string()), true),
        ];
        let staged = names(&["verdict", "bool", "num", "quoted", "null", "broken"]);
        let (value, _ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Lenient).expect("Final present");
        assert_eq!(
            value,
            serde_json::json!({
                "out": "PASS",
                "parts": {
                    "verdict": "PASS",
                    "bool": "true",
                    "num": "42",
                    "quoted": "\"quoted\"",
                    "null": "null",
                    "broken": "{not json",
                },
            })
        );
    }

    /// `submit_format: "text"` opt-out (`FoldParse::Raw`): a
    /// JSON-container string folds as itself — the escape hatch for a
    /// step that needs the raw text of a JSON-looking body.
    #[test]
    fn raw_mode_keeps_container_strings_unparsed() {
        let tail = vec![
            artifact("data.json", Value::String(r#"{"k":1}"#.to_string())),
            final_ev(Value::String(r#"["a","b"]"#.to_string()), true),
        ];
        let staged = names(&["data.json"]);
        let (value, _ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Raw).expect("Final present");
        assert_eq!(
            value,
            serde_json::json!({
                "out": r#"["a","b"]"#,
                "parts": {"data.json": r#"{"k":1}"#},
            })
        );
    }

    /// Leading-whitespace container strings still parse under `Lenient`
    /// (`trim_start` before the leading-byte check), and already
    /// structured values (a strict `"json"` body parsed at submit time,
    /// an in-process Lua table) pass through both modes untouched.
    #[test]
    fn lenient_fold_trims_leading_whitespace_and_passes_structured_through() {
        let tail = vec![
            artifact("structured", serde_json::json!({"already": true})),
            final_ev(Value::String("  \n {\"k\": 1}".to_string()), true),
        ];
        let staged = names(&["structured"]);
        let (value, _ok) =
            fold_final_and_parts(&tail, &staged, FoldParse::Lenient).expect("Final present");
        assert_eq!(
            value,
            serde_json::json!({
                "out": {"k": 1},
                "parts": {"structured": {"already": true}},
            })
        );
    }

    /// `fold_parse_mode_for`: `submit_format: "text"` in the step's
    /// `AgentContextView.extra` resolves `Raw`; absent view, absent key,
    /// `"json"`, and unrecognized values all resolve `Lenient` (the
    /// default).
    #[tokio::test]
    async fn fold_parse_mode_for_resolves_text_to_raw_and_everything_else_to_lenient() {
        let engine = Engine::new(EngineCfg::default());
        let task_id = StepId::new();

        // No agent_ctx entry at all → Lenient.
        assert_eq!(
            engine.fold_parse_mode_for(&task_id, 1).await,
            FoldParse::Lenient
        );

        let seed = |declared: Option<Value>| {
            let engine = engine.clone();
            let task_id = task_id.clone();
            async move {
                engine
                    .with_state("test.seed_submit_format", move |s| {
                        let mut entry = crate::core::state::AgentCtxEntry::default();
                        if let Some(v) = declared {
                            entry.view.extra.insert(SUBMIT_FORMAT_KEY.to_string(), v);
                        }
                        s.agent_ctx.insert((task_id, 1), entry);
                    })
                    .await
                    .expect("seed agent_ctx");
            }
        };

        seed(None).await;
        assert_eq!(
            engine.fold_parse_mode_for(&task_id, 1).await,
            FoldParse::Lenient
        );
        seed(Some(Value::String("json".to_string()))).await;
        assert_eq!(
            engine.fold_parse_mode_for(&task_id, 1).await,
            FoldParse::Lenient
        );
        seed(Some(Value::String("yaml".to_string()))).await;
        assert_eq!(
            engine.fold_parse_mode_for(&task_id, 1).await,
            FoldParse::Lenient
        );
        seed(Some(Value::String(SUBMIT_FORMAT_TEXT.to_string()))).await;
        assert_eq!(
            engine.fold_parse_mode_for(&task_id, 1).await,
            FoldParse::Raw
        );
    }

    /// `stage_worker_artifact_trusted` writes onto the `(task_id, attempt)`
    /// key exactly like `submit_worker_result_trusted` does — a part staged
    /// under attempt N is invisible to an `output_tail` / allowlist read of
    /// attempt N+1 (a fresh attempt starts empty; nothing carries over).
    #[tokio::test]
    async fn stage_worker_artifact_trusted_is_isolated_per_attempt() {
        let engine = Engine::new(EngineCfg::default());
        let task_id = StepId::new();

        engine
            .stage_worker_artifact_trusted(&task_id, 1, "a".to_string(), serde_json::json!("v1"))
            .await
            .expect("stage attempt 1");

        let attempt_1_tail = engine.output_tail(&task_id, 1).await;
        assert_eq!(attempt_1_tail.len(), 1);
        assert!(matches!(
            &attempt_1_tail[0],
            OutputEvent::Artifact { name, .. } if name == "a"
        ));
        assert_eq!(
            engine.worker_artifact_names_for(&task_id, 1).await,
            vec!["a".to_string()]
        );

        let attempt_2_tail = engine.output_tail(&task_id, 2).await;
        assert!(
            attempt_2_tail.is_empty(),
            "attempt 2 must not see attempt 1's staged part"
        );
        assert!(
            engine
                .worker_artifact_names_for(&task_id, 2)
                .await
                .is_empty(),
            "attempt 2's allowlist must not see attempt 1's staged name"
        );
    }
}

// ─── GH #50 (Subtask 2): `Engine::register_verdict_contracts` /
// `Engine::verdict_contract_for_task` ────────────────────────────────────
#[cfg(test)]
mod verdict_contract_registry_tests {
    use super::*;

    async fn seeded_engine(agent: &str) -> (Engine, StepId) {
        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: agent.to_string(),
                    initial_directive: serde_json::json!("x"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        (engine, task_id)
    }

    /// An agent with no registered contract at all → `None` (the opt-in
    /// default; every pre-GH-#50 `Engine`).
    #[tokio::test]
    async fn returns_none_when_no_contract_registered_for_the_agent() {
        let (engine, task_id) = seeded_engine("gate").await;
        assert_eq!(engine.verdict_contract_for_task(&task_id).await, None);
    }

    /// A registered contract for the running task's agent is returned
    /// verbatim.
    #[tokio::test]
    async fn returns_the_registered_contract_for_the_running_agent() {
        let (engine, task_id) = seeded_engine("gate").await;
        let contract = mlua_swarm_schema::VerdictContract {
            channel: mlua_swarm_schema::VerdictChannel::Body,
            values: vec!["PASS".to_string(), "BLOCKED".to_string()],
        };
        engine.register_verdict_contracts(HashMap::from([("gate".to_string(), contract.clone())]));
        assert_eq!(
            engine.verdict_contract_for_task(&task_id).await,
            Some(contract)
        );
    }

    /// A registered contract for a DIFFERENT agent name never leaks onto
    /// an unrelated task.
    #[tokio::test]
    async fn does_not_leak_a_contract_registered_for_a_different_agent() {
        let (engine, task_id) = seeded_engine("gate").await;
        engine.register_verdict_contracts(HashMap::from([(
            "other-agent".to_string(),
            mlua_swarm_schema::VerdictContract {
                channel: mlua_swarm_schema::VerdictChannel::Body,
                values: vec!["PASS".to_string()],
            },
        )]));
        assert_eq!(engine.verdict_contract_for_task(&task_id).await, None);
    }

    /// An unknown `task_id` → `None`, not a panic / error.
    #[tokio::test]
    async fn returns_none_for_an_unknown_task_id() {
        let engine = Engine::new(EngineCfg::default());
        let unknown = StepId::new();
        assert_eq!(engine.verdict_contract_for_task(&unknown).await, None);
    }

    /// `register_verdict_contracts` is additive (`HashMap::extend`): a
    /// second call registering a DIFFERENT agent does not clobber the
    /// first call's entry.
    #[tokio::test]
    async fn register_verdict_contracts_is_additive_across_calls() {
        let (engine, task_id) = seeded_engine("gate").await;
        let contract = mlua_swarm_schema::VerdictContract {
            channel: mlua_swarm_schema::VerdictChannel::Part,
            values: vec!["ALLOW".to_string()],
        };
        engine.register_verdict_contracts(HashMap::from([("gate".to_string(), contract.clone())]));
        engine.register_verdict_contracts(HashMap::from([(
            "unrelated-agent".to_string(),
            mlua_swarm_schema::VerdictContract {
                channel: mlua_swarm_schema::VerdictChannel::Body,
                values: vec!["X".to_string()],
            },
        )]));
        assert_eq!(
            engine.verdict_contract_for_task(&task_id).await,
            Some(contract)
        );
    }
}

// ─── GH #51: completion-time verdict-contract enforcement — the shared
// `Engine::verdict_contract_completion_check` choke point embedded inside
// `submit_worker_result_trusted` / `submit_output`, exercised here at the
// `submit_output` level (the WS Operator fallback route's own unit-test
// coverage — see `crates/mlua-swarm-server/tests/verdict_contract.rs` for
// the HTTP-round-trip coverage of the other 2 routes) ───────────────────
#[cfg(test)]
mod verdict_contract_completion_tests {
    use super::*;

    /// Seeds a `Pending` task bound to `agent` and mints a bound
    /// `Role::Worker` token for it — the same mint-and-register pattern
    /// `initial_directive_value_passthrough_tests::mint_worker_token`
    /// uses (duplicated here: that helper is private to its own sibling
    /// `#[cfg(test)]` module, not reachable via `super::*` from this one).
    async fn seeded_task_with_worker_token(agent: &str) -> (Engine, CapToken, StepId) {
        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: agent.to_string(),
                    initial_directive: serde_json::json!("x"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");
        let worker_token = engine.signer().session(
            format!("worker-of-{task_id}"),
            Role::Worker,
            vec!["*".into()],
            Duration::from_secs(600),
        );
        let fp = worker_token.fingerprint();
        let record = CapTokenRecord::from_worker_token(worker_token.clone(), task_id.clone());
        engine
            .with_state("test.mint_worker", move |s| {
                s.tokens.insert(fp, record);
            })
            .await
            .expect("mint worker token");
        (engine, worker_token, task_id)
    }

    fn body_contract(values: &[&str]) -> mlua_swarm_schema::VerdictContract {
        mlua_swarm_schema::VerdictContract {
            channel: mlua_swarm_schema::VerdictChannel::Body,
            values: values.iter().map(|v| v.to_string()).collect(),
        }
    }

    fn part_contract(values: &[&str]) -> mlua_swarm_schema::VerdictContract {
        mlua_swarm_schema::VerdictContract {
            channel: mlua_swarm_schema::VerdictChannel::Part,
            values: values.iter().map(|v| v.to_string()).collect(),
        }
    }

    fn final_event(value: Value, ok: bool) -> crate::worker::output::OutputEvent {
        crate::worker::output::OutputEvent::Final {
            content: crate::worker::output::ContentRef::Inline { value },
            ok,
        }
    }

    /// Route 3 (WS Operator fallback, `submit_output` level) — a
    /// `channel: "part"` contract's attempt completes via a plain
    /// `Final` without ever staging a `"verdict"` artifact: rejected
    /// with `EngineError::VerdictPartMissing`, and nothing lands on
    /// `output_tail` — the rejected value never reaches the flow ctx.
    #[tokio::test]
    async fn submit_output_rejects_missing_verdict_part() {
        let (engine, token, task_id) = seeded_task_with_worker_token("gate").await;
        engine.register_verdict_contracts(HashMap::from([(
            "gate".to_string(),
            part_contract(&["PASS", "BLOCKED"]),
        )]));

        let err = engine
            .submit_output(
                &token,
                &task_id,
                1,
                final_event(serde_json::json!("anything"), true),
            )
            .await
            .expect_err("missing staged verdict part must be rejected");
        assert!(
            matches!(err, EngineError::VerdictPartMissing { .. }),
            "unexpected error variant: {err:?}"
        );

        let tail = engine.output_tail(&task_id, 1).await;
        assert!(
            !tail
                .iter()
                .any(|ev| matches!(ev, crate::worker::output::OutputEvent::Final { .. })),
            "a rejected completion must not write a Final onto output_tail"
        );
    }

    /// Route 3 — a `channel: "part"` contract completes normally when the
    /// worker DID stage a matching `"verdict"` artifact first (defense in
    /// depth: presence AND membership both hold).
    #[tokio::test]
    async fn submit_output_accepts_when_verdict_part_is_staged_and_a_member() {
        let (engine, token, task_id) = seeded_task_with_worker_token("gate").await;
        engine.register_verdict_contracts(HashMap::from([(
            "gate".to_string(),
            part_contract(&["PASS", "BLOCKED"]),
        )]));
        engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "verdict".to_string(),
                serde_json::json!("PASS"),
            )
            .await
            .expect("stage verdict part");

        engine
            .submit_output(
                &token,
                &task_id,
                1,
                final_event(serde_json::json!("full report"), true),
            )
            .await
            .expect("staged + member verdict part must be accepted");

        let tail = engine.output_tail(&task_id, 1).await;
        assert!(
            tail.iter()
                .any(|ev| matches!(ev, crate::worker::output::OutputEvent::Final { .. })),
            "an accepted completion must write its Final onto output_tail"
        );
    }

    /// Route 3 — a `channel: "body"` contract's completing value is NOT a
    /// member of `values`: rejected with
    /// `EngineError::VerdictValueRejected`, no `Final` written.
    #[tokio::test]
    async fn submit_output_rejects_body_value_outside_contract() {
        let (engine, token, task_id) = seeded_task_with_worker_token("gate").await;
        engine.register_verdict_contracts(HashMap::from([(
            "gate".to_string(),
            body_contract(&["PASS", "BLOCKED"]),
        )]));

        let err = engine
            .submit_output(
                &token,
                &task_id,
                1,
                final_event(serde_json::json!("UNKNOWN"), true),
            )
            .await
            .expect_err("out-of-contract body value must be rejected");
        match err {
            EngineError::VerdictValueRejected { value, allowed } => {
                assert_eq!(value, "UNKNOWN");
                assert_eq!(allowed, vec!["PASS".to_string(), "BLOCKED".to_string()]);
            }
            other => panic!("unexpected error variant: {other:?}"),
        }

        let tail = engine.output_tail(&task_id, 1).await;
        assert!(
            !tail
                .iter()
                .any(|ev| matches!(ev, crate::worker::output::OutputEvent::Final { .. })),
            "a rejected completion must not write a Final onto output_tail"
        );
    }

    /// `ok=false` bypasses the completion-time check entirely, regardless
    /// of channel or membership — the exemption acceptance criterion,
    /// exercised at the `submit_output` choke point.
    #[tokio::test]
    async fn submit_output_ok_false_bypasses_the_check() {
        let (engine, token, task_id) = seeded_task_with_worker_token("gate").await;
        engine.register_verdict_contracts(HashMap::from([(
            "gate".to_string(),
            body_contract(&["PASS", "BLOCKED"]),
        )]));

        engine
            .submit_output(
                &token,
                &task_id,
                1,
                final_event(serde_json::json!("UNKNOWN"), false),
            )
            .await
            .expect("ok=false must bypass the verdict contract check entirely");

        let tail = engine.output_tail(&task_id, 1).await;
        assert!(
            tail.iter()
                .any(|ev| matches!(ev, crate::worker::output::OutputEvent::Final { .. })),
            "an ok=false completion is exempt, not rejected — its Final must still land"
        );
    }

    /// `staged_verdict_value_for` mirrors `fold_final_and_parts`'s
    /// last-write-wins semantics: staging `"verdict"` twice within the
    /// same attempt returns the LAST value, not the first.
    #[tokio::test]
    async fn staged_verdict_value_for_is_last_write_wins() {
        let (engine, _token, task_id) = seeded_task_with_worker_token("gate").await;
        engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "verdict".to_string(),
                serde_json::json!("PASS"),
            )
            .await
            .expect("stage first verdict part");
        engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "verdict".to_string(),
                serde_json::json!("BLOCKED"),
            )
            .await
            .expect("stage second verdict part");

        assert_eq!(
            engine.staged_verdict_value_for(&task_id, 1).await,
            Some("BLOCKED".to_string())
        );
    }

    /// `staged_verdict_value_for` ignores artifacts staged under any name
    /// OTHER than the literal `"verdict"` — mirrors `channel: "part"`
    /// contracts only ever addressing that one part.
    #[tokio::test]
    async fn staged_verdict_value_for_ignores_other_artifact_names() {
        let (engine, _token, task_id) = seeded_task_with_worker_token("gate").await;
        engine
            .stage_worker_artifact_trusted(
                &task_id,
                1,
                "notes".to_string(),
                serde_json::json!("irrelevant"),
            )
            .await
            .expect("stage unrelated part");

        assert_eq!(engine.staged_verdict_value_for(&task_id, 1).await, None);
    }

    /// `staged_verdict_value_for` → `None` when nothing was ever staged —
    /// the normal case the completion check turns into
    /// `EngineError::VerdictPartMissing`.
    #[tokio::test]
    async fn staged_verdict_value_for_returns_none_when_nothing_staged() {
        let (engine, _token, task_id) = seeded_task_with_worker_token("gate").await;
        assert_eq!(engine.staged_verdict_value_for(&task_id, 1).await, None);
    }
}

// ─── GH #76 Skip tier: DispatchOutcome::Skip tier + SubmitOutcome API ────────────
#[cfg(test)]
mod skip_tier_tests {
    use super::*;
    use crate::blueprint::compiler::{RustFnInProcessSpawnerFactory, SpawnerFactory};
    use crate::blueprint::EngineDispatcher;
    use crate::core::state::{
        is_skip_marker, unwrap_skip_marker, wrap_skip_marker, SubmitOutcome, SKIP_MARKER_KEY,
    };
    use crate::store::run::{InMemoryRunStore, RunContext, RunRecord, RunStatus, RunStore};
    use crate::types::{RunId, TaskId};
    use crate::worker::adapter::WorkerResult;
    use mlua_flow_ir::AsyncDispatcher;
    use mlua_swarm_schema::{AgentDef, AgentKind};
    use serde_json::json;

    /// `DispatchOutcome::Skip(v)` roundtrips through serde JSON without
    /// loss — the enum is serialized with the default externally-tagged
    /// form (same as `Pass`/`Blocked`), so no `#[serde(...)]` tuning is
    /// needed for the new variant.
    #[test]
    fn dispatch_outcome_skip_variant_serializes_roundtrip() {
        let outcome = DispatchOutcome::Skip(json!({ "verdict": "SKIP", "reason": "n/a" }));
        let serialized = serde_json::to_string(&outcome).expect("serialize");
        let round: DispatchOutcome = serde_json::from_str(&serialized).expect("deserialize");
        match round {
            DispatchOutcome::Skip(v) => {
                assert_eq!(v, json!({ "verdict": "SKIP", "reason": "n/a" }));
            }
            other => panic!("expected Skip after roundtrip, got {other:?}"),
        }
    }

    /// The `is_skip_marker` / `unwrap_skip_marker` / `wrap_skip_marker`
    /// helper triangle round-trips consistently and rejects plain
    /// payloads. Pinning the reserved-key contract in a unit test guards
    /// against a future edit accidentally renaming the sentinel key
    /// (which would silently break every downstream reader).
    #[test]
    fn skip_marker_helpers_wrap_detect_and_unwrap() {
        assert!(!is_skip_marker(&json!("plain string")));
        assert!(!is_skip_marker(&json!({ "verdict": "PASS" })));
        assert!(!is_skip_marker(&json!(null)));

        let inner = json!({ "reason": "not applicable" });
        let wrapped = wrap_skip_marker(inner.clone());
        assert!(is_skip_marker(&wrapped));
        assert_eq!(wrapped[SKIP_MARKER_KEY], json!(true));
        assert_eq!(unwrap_skip_marker(&wrapped), Some(inner));

        // A malformed sentinel (marker key present but `value` absent) is
        // still a Skip signal, defaulting the carried payload to Null so
        // downstream match arms never observe `None` on a marker match.
        let malformed = json!({ SKIP_MARKER_KEY: true });
        assert!(is_skip_marker(&malformed));
        assert_eq!(unwrap_skip_marker(&malformed), Some(Value::Null));

        // Plain payloads → `unwrap_skip_marker` returns `None` (the
        // caller falls back to the ordinary Pass/Blocked path).
        assert_eq!(unwrap_skip_marker(&json!("plain")), None);
    }

    /// The `SubmitOutcome::Skip` mapping wraps the payload in the
    /// skip-marker sentinel AND records `Final.ok = true` — matching the
    /// invariant in the outcome mapping table in
    /// `submit_worker_result_trusted`'s doc. This is the wire shape
    /// `dispatch_attempt_with*` reads back to route into
    /// `DispatchOutcome::Skip`.
    #[tokio::test]
    async fn submit_worker_result_trusted_skip_outcome_records_final_ok_true_with_sentinel() {
        use crate::worker::output::OutputEvent;
        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: "analyst".into(),
                    initial_directive: json!("go"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");

        let inner_verdict = json!({ "verdict": "SKIP", "reason": "migration=no" });
        engine
            .submit_worker_result_trusted(&task_id, 1, inner_verdict.clone(), SubmitOutcome::Skip)
            .await
            .expect("submit with Skip outcome");

        let tail = engine.output_tail(&task_id, 1).await;
        let final_ev = tail
            .iter()
            .rev()
            .find_map(|ev| match ev {
                OutputEvent::Final { content, ok } => Some((content.clone(), *ok)),
                _ => None,
            })
            .expect("Final present after Skip submit");
        assert!(
            final_ev.1,
            "Skip records Final.ok = true (flow-continuation)"
        );
        let stored_value = super::content_ref_to_value(final_ev.0);
        assert!(
            is_skip_marker(&stored_value),
            "Skip wraps the payload in the sentinel: got {stored_value}"
        );
        assert_eq!(unwrap_skip_marker(&stored_value), Some(inner_verdict));
    }

    /// The new `SubmitOutcome::Pass` / `SubmitOutcome::Blocked` arms
    /// preserve byte-for-byte the pre-#76 wire shape (Final.ok mirrors
    /// the tier; the value is not wrapped). Regression against a future
    /// edit that accidentally routes Pass/Blocked through the Skip
    /// wrapper.
    #[tokio::test]
    async fn submit_worker_result_trusted_pass_and_blocked_wire_unchanged() {
        use crate::worker::output::OutputEvent;
        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");

        // Pass path.
        let pass_task = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: "worker".into(),
                    initial_directive: json!("go"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task pass");
        engine
            .submit_worker_result_trusted(&pass_task, 1, json!("pass-value"), SubmitOutcome::Pass)
            .await
            .expect("submit Pass");
        let pass_tail = engine.output_tail(&pass_task, 1).await;
        let (pass_content, pass_ok) = pass_tail
            .iter()
            .rev()
            .find_map(|ev| match ev {
                OutputEvent::Final { content, ok } => Some((content.clone(), *ok)),
                _ => None,
            })
            .expect("Final present");
        assert!(pass_ok);
        assert_eq!(
            super::content_ref_to_value(pass_content),
            json!("pass-value"),
            "Pass value must not be wrapped"
        );

        // Blocked path.
        let blocked_task = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: "worker".into(),
                    initial_directive: json!("go"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task blocked");
        engine
            .submit_worker_result_trusted(
                &blocked_task,
                1,
                json!("blocked-value"),
                SubmitOutcome::Blocked,
            )
            .await
            .expect("submit Blocked");
        let blocked_tail = engine.output_tail(&blocked_task, 1).await;
        let (blocked_content, blocked_ok) = blocked_tail
            .iter()
            .rev()
            .find_map(|ev| match ev {
                OutputEvent::Final { content, ok } => Some((content.clone(), *ok)),
                _ => None,
            })
            .expect("Final present");
        assert!(!blocked_ok);
        assert_eq!(
            super::content_ref_to_value(blocked_content),
            json!("blocked-value"),
            "Blocked value must not be wrapped"
        );
    }

    /// End-to-end (engine layer): a worker that returns a skip-marker
    /// sentinel value via `WorkerResult { value: wrap_skip_marker(inner),
    /// ok: true }` — which is what a Skip-aware caller of
    /// `submit_worker_result_trusted(..., SubmitOutcome::Skip)` places on
    /// the wire — is folded by `dispatch_attempt_with_run_ctx` into
    /// `DispatchOutcome::Skip(inner)`. Proves the sentinel → outcome
    /// routing that the flow-ir binding boundary depends on.
    #[tokio::test]
    async fn dispatcher_folds_skip_sentinel_into_skip_outcome() {
        let inner_verdict = json!({ "verdict": "SKIP", "reason": "not applicable" });
        let inner_for_worker = inner_verdict.clone();
        let factory = RustFnInProcessSpawnerFactory::new().register_fn("analyst", move |_inv| {
            let value = wrap_skip_marker(inner_for_worker.clone());
            async move {
                Ok(WorkerResult {
                    value,
                    ok: true,
                    stats: None,
                })
            }
        });
        let def = AgentDef {
            name: "analyst".into(),
            kind: AgentKind::RustFn,
            spec: json!({ "fn_id": "analyst" }),
            profile: None,
            meta: None,
            runner: None,
            runner_ref: None,
            verdict: None,
            lints: None,
        };
        let spawner = factory.build(&def, None).expect("build");

        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let task_id = engine
            .start_task(
                &op_token,
                TaskSpec {
                    agent: "analyst".into(),
                    initial_directive: json!("go"),
                    step_ctx: None,
                    check_policy: None,
                },
            )
            .await
            .expect("start_task");

        let outcome = engine
            .dispatch_attempt_with_run_ctx(&op_token, &task_id, &spawner, None)
            .await
            .expect("dispatch ok");

        match outcome {
            DispatchOutcome::Skip(v) => {
                assert_eq!(v, inner_verdict, "Skip carries the unwrapped inner verdict");
            }
            other => panic!("expected DispatchOutcome::Skip, got {other:?}"),
        }
    }

    /// `EngineDispatcher::dispatch` (the `AsyncDispatcher` impl flow-ir
    /// invokes) maps `DispatchOutcome::Skip(v)` to `Ok(wrap_skip_marker(v))`
    /// — a successful return whose Value carries the sentinel across the
    /// flow-ir boundary. Pinning this mapping in a test guards the arm
    /// order (a wildcard `Ok(other) =>` arm accidentally placed BEFORE the
    /// Skip arm would route Skip to `EvalError::DispatcherError` and
    /// abort the flow — the exact failure mode this tier prevents).
    #[tokio::test]
    async fn engine_dispatcher_maps_skip_outcome_to_ok_sentinel_value() {
        let inner_verdict = json!({ "verdict": "SKIP", "reason": "not applicable" });
        let inner_for_worker = inner_verdict.clone();
        let factory = RustFnInProcessSpawnerFactory::new().register_fn("analyst", move |_inv| {
            let value = wrap_skip_marker(inner_for_worker.clone());
            async move {
                Ok(WorkerResult {
                    value,
                    ok: true,
                    stats: None,
                })
            }
        });
        let def = AgentDef {
            name: "analyst".into(),
            kind: AgentKind::RustFn,
            spec: json!({ "fn_id": "analyst" }),
            profile: None,
            meta: None,
            runner: None,
            runner_ref: None,
            verdict: None,
            lints: None,
        };
        let spawner = factory.build(&def, None).expect("build");

        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");
        let dispatcher = EngineDispatcher::with_spawner(engine.clone(), op_token, spawner);

        let out = dispatcher
            .dispatch("analyst", json!("go"))
            .await
            .expect("dispatch returns Ok for Skip tier (not EvalError::DispatcherError)");

        assert!(
            is_skip_marker(&out),
            "returned value must carry the skip-marker sentinel across the flow-ir boundary: got {out}"
        );
        assert_eq!(unwrap_skip_marker(&out), Some(inner_verdict));
    }

    /// `EngineDispatcher::dispatch`'s `RunContext` step-entry log records
    /// `status = "skipped"` for a Skip completion (distinct from
    /// `"passed"` / `"blocked"`), so post-run inspection of
    /// `RunRecord.step_entries` can distinguish flow-continuation-with-
    /// binding-write from flow-continuation-without-binding-write.
    #[tokio::test]
    async fn engine_dispatcher_step_entry_status_is_skipped_for_skip_outcome() {
        let inner_verdict = json!({ "verdict": "SKIP" });
        let inner_for_worker = inner_verdict.clone();
        let factory = RustFnInProcessSpawnerFactory::new().register_fn("analyst", move |_inv| {
            let value = wrap_skip_marker(inner_for_worker.clone());
            async move {
                Ok(WorkerResult {
                    value,
                    ok: true,
                    stats: None,
                })
            }
        });
        let def = AgentDef {
            name: "analyst".into(),
            kind: AgentKind::RustFn,
            spec: json!({ "fn_id": "analyst" }),
            profile: None,
            meta: None,
            runner: None,
            runner_ref: None,
            verdict: None,
            lints: None,
        };
        let spawner = factory.build(&def, None).expect("build");

        let engine = Engine::new(EngineCfg::default());
        let op_token = engine
            .attach("ut-op", Role::Operator, Duration::from_secs(30))
            .await
            .expect("attach");

        // Seed a RunContext with an InMemoryRunStore so the dispatcher
        // appends a step_entry we can then read back.
        let run_id = RunId::new();
        let run_store: Arc<dyn RunStore> = Arc::new(InMemoryRunStore::new());
        run_store
            .create(RunRecord {
                id: run_id.clone(),
                task_id: TaskId::new(),
                status: RunStatus::Running,
                step_entries: Vec::new(),
                degradations: Vec::new(),
                operator_sid: None,
                result_ref: None,
                input_json: None,
                created_at: 0,
                updated_at: 0,
            })
            .await
            .expect("create run record");
        let run_ctx = RunContext::new(run_id.clone(), run_store.clone());

        let dispatcher =
            EngineDispatcher::with_spawner(engine.clone(), op_token, spawner).with_run(run_ctx);

        let out = dispatcher
            .dispatch("analyst", json!("go"))
            .await
            .expect("dispatch ok");
        assert!(is_skip_marker(&out));

        let record = run_store.get(&run_id).await.expect("run record present");
        let step = record
            .step_entries
            .first()
            .expect("at least one step_entry appended for the dispatched step");
        assert_eq!(
            step.status.as_deref(),
            Some("skipped"),
            "Skip outcome must record StepEntry.status = \"skipped\""
        );
        assert_eq!(step.step_ref.as_deref(), Some("analyst"));
    }
}