Struct Engine 

pub struct Engine { /* private fields */ }

Process-wide long-running runtime. Cheap to clone() — an Arc lives inside.

Implementations

impl Engine

pub fn new(cfg: EngineCfg) -> Self

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_with_layers(cfg: EngineCfg, layer_registry: LayerRegistry) -> Self

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 with_gate(self, gate: RoleVerbGate) -> Self

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 cfg(&self) -> &EngineCfg

Access the EngineCfg this engine was built with.

pub fn layer_registry(&self) -> &LayerRegistry

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 signer(&self) -> &TokenSigner

Access the TokenSigner used to mint/verify CapTokens.

pub fn event_tx(&self) -> Sender<Event>

Clone a handle to the process-wide Event broadcast sender. Prefer subscribe for a ready-to-use receiver.

pub fn subscribe(&self) -> EventStream

Subscribe to the engine’s Event broadcast stream.

pub fn set_output_store(&self, store: Arc<dyn OutputStore>)

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 register_verdict_contracts( &self, contracts: HashMap<String, VerdictContract>, )

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 async fn verdict_contract_for_task( &self, task_id: &StepId, ) -> Option<VerdictContract>

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 with_state<F, R>( &self, op: &'static str, f: F, ) -> Result<R, EngineError>

where F: FnOnce(&mut EngineState) -> R,

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 verify_token( &self, token: &CapToken, verb: Verb, ) -> Result<(), EngineError>

Four steps: (1) signature verify, (2) expiry check — skipped for Role::Operator, (3) role × verb gate, (4) uses_left consume.

Why step (2) is role-conditional

An Operator session token stays inside the process. Self::attach / Self::attach_with_ids mint it and the server holds it for exactly as long as the attach lives; unlike a Worker token it is never serialized out to a spawned SubAgent, and it is never rendered as an Authorization: Bearer <CapToken::encode()> header. (The HTTP /v1/sessions route hands the caller only the opaque session id, which the server resolves back to the token it kept.) A TTL on it therefore bounds no capability that a spawned worker could be holding; its only observable effect is to reject the next legitimate start_task / dispatch_attempt as soon as one step outlives the attach TTL. That is a misfire, not a defence, so Role::Operator skips the expiry check entirely.

Every other role keeps it. A Worker token goes out over the wire to a subprocess or a remote SubAgent, where the bearer can outlive the step it was minted for, and the TTL is the only bound on a leaked one; the same reasoning is applied conservatively to Senior / Observer, which are not proven to stay in-process. Those roles still fail with EngineError::TokenExpired.

CapToken::expire_at and the signed payload are unchanged — an Operator token still carries an expire_at, it just no longer gates verification.

pub async fn verify_token_for_task( &self, token: &CapToken, verb: Verb, task_id: &StepId, ) -> Result<(), EngineError>

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 task_id_from_token( &self, token: &CapToken, ) -> Result<StepId, EngineError>

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_handle( &self, handle: &str, ) -> Result<StepId, EngineError>

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 remint_worker_token( &self, expiring: &CapToken, ) -> Result<CapToken, EngineError>

Reissue a Role::Worker capability whose delivery is running late, against the record this engine already holds for it.

The failure this exists for

A Operator::execute implementation builds its whole spawn frame — capability token included — and only then tries to write it. The WS implementation parks that write for the length of a client disconnect with no deadline (bounding the wait is infra’s call; see mse_server::operator_ws::session’s module doc), while the token inside has been counting down EngineCfg::worker_token_ttl_secs since Self::dispatch_attempt_with minted it. Past that TTL Self::verify_token rejects it — the expiry check is skipped only for Role::Operator — so the frame arrived carrying a capability that was already dead, and the SubAgent found out at submit, after doing the entire job. Re-minting at the moment of delivery is what makes the TTL bound the token’s time in the wild rather than its time waiting to leave the server.

This cannot widen what the bearer may do

Nothing here is taken from the caller’s intent; every field is copied from what the engine already granted:

• the presented token must verify against this signer, so a caller cannot hand in a token it composed itself;
• it must be Role::Worker, and the reissue is Role::Worker — the role is never re-chosen;
• agent_id and scopes are copied from the presented token, so the subject and the scope set are the ones already in force;
• the new record binds the same task_id the stored record binds, which is what verify_token_for_task’s ownership gate reads — a reissue can therefore never reach a different task;
• max_uses is the stored record’s remaining budget, not the original allowance, so a reissue of a spent token is still spent.

The only thing that moves is expire_at.

The old record is left in place

Deliberately, on two counts. The short worker handle (worker_handles, minted next to the original in Self::dispatch_attempt_with) resolves through the original fingerprint, and OperatorSpawner’s completion path still holds the original token to push a fallback Final with (mse::operator, the submit_output call after operator.execute returns). Dropping the record would turn both into TokenNotFound. Two records for one attempt is the cost, and they are equivalent: same subject, same role, same scopes, same bound task.

Errors

EngineError::BadSignature for a token this signer did not mint, EngineError::RoleViolation for a non-Worker role, EngineError::TokenNotFound when no record backs the presented token or the record binds no task, and EngineError::TokenUsesExhausted for a revoked record — the same mapping Self::verify_token applies to a revoked one.

pub async fn submit_worker_result_trusted( &self, task_id: &StepId, attempt: u32, value: Value, outcome: SubmitOutcome, ) -> Result<(), EngineError>

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 stage_worker_artifact_trusted( &self, task_id: &StepId, attempt: u32, name: String, value: Value, ) -> Result<(), EngineError>

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 attach( &self, operator_id: impl Into<String>, role: Role, ttl: Duration, ) -> Result<CapToken, EngineError>

Attach a new session with default OperatorInfo (Automate, no bridges/hooks). Shorthand for attach_with(.., OperatorInfo::default()).

ttl is still stamped onto the minted token’s CapToken::expire_at, but for Role::Operator it no longer gates verification — Self::verify_token skips the expiry check for that role, so an Operator session keeps working past ttl. Pass a non-Operator role and the TTL is enforced as before.

pub async fn register_senior_bridge( &self, id: impl Into<String>, bridge: Arc<dyn SeniorBridge>, )

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_spawn_hook( &self, id: impl Into<String>, hook: Arc<dyn SpawnHook>, )

Register a SpawnHook under a name. An existing entry with the same name is overwritten.

pub async fn register_operator( &self, id: impl Into<String>, operator: Arc<dyn Operator>, )

Register an Operator (a spawn-body backend) under a name. An existing entry with the same name is overwritten.

Two things read this map, neither of them a dispatch-time ctx lookup: Self::list_operator_ids, which a host uses to reject a launch naming an unregistered operator_sid, and the host’s seat resolver, which turns the Run’s current holder into a destination on each dispatch. The ctx-mediated reader this doc used to name, OperatorDelegateMiddleware, was removed.

pub async fn unregister_senior_bridge(&self, id: &str)

Unregister a SeniorBridge by name (e.g. on WebSocket disconnect or explicit teardown). A missing ID is a no-op.

pub async fn unregister_spawn_hook(&self, id: &str)

Unregister a SpawnHook by name. A missing ID is a no-op.

pub async fn unregister_operator(&self, id: &str)

Unregister an Operator backend by name. A missing ID is a no-op.

pub async fn list_spawn_hook_ids(&self) -> Vec<String>

Snapshot the list of registered SpawnHook IDs (for test observation and debugging).

pub async fn list_senior_bridge_ids(&self) -> Vec<String>

Snapshot the list of registered SeniorBridge IDs.

pub async fn list_operator_ids(&self) -> Vec<String>

Snapshot the list of registered Operator IDs.

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>

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 LaunchEnvelope.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.

ttl is still stamped onto the minted token’s CapToken::expire_at, but for Role::Operator it no longer gates verification — Self::verify_token skips the expiry check for that role, so a long step can no longer make the next start_task / dispatch_attempt fail with EngineError::TokenExpired. Pass a non-Operator role and the TTL is enforced as before.

pub async fn attach_with( &self, operator_id: impl Into<String>, role: Role, ttl: Duration, operator_info: OperatorInfo, ) -> Result<CapToken, EngineError>

Convenience attach that takes an OperatorInfo (two 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 Arcs from those IDs via resolve_operator_info, and the middlewares that read them fire as usual — SeniorEscalationMiddleware off senior_bridge, MainAIMiddleware off spawn_hook. There were three; the third was OperatorDelegateMiddleware, and the ob-<hex> id it consumed now resolves to nothing here (see crate::core::ctx::OperatorInfo, “Persistence boundary”).

⚠ Non-persisted sessions only

Because this API takes inline Arcs, 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.

ttl is still stamped onto the minted token’s CapToken::expire_at, but for Role::Operator it no longer gates verification — see Self::verify_token for why the expiry check is role-conditional.

pub async fn detach(&self, token: &CapToken) -> Result<(), EngineError>

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 heartbeat(&self, token: &CapToken) -> Result<(), EngineError>

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 start_task( &self, token: &CapToken, spec: TaskSpec, ) -> Result<StepId, EngineError>

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 read_task_state( &self, token: &CapToken, task_id: &StepId, ) -> Result<TaskState, EngineError>

Fetch a snapshot of TaskState for task_id, subject to the task-ownership gate (see verify_token_for_task).

pub async fn cancel_task( &self, token: &CapToken, task_id: &StepId, ) -> Result<(), EngineError>

Mark task_id as Cancelled and wake any caller blocked in poll_task for it.

pub async fn dispatch_attempt_with( &self, token: &CapToken, task_id: &StepId, spawner: &Arc<dyn SpawnerAdapter>, run_id: Option<&RunId>, ) -> Result<DispatchOutcome, EngineError>

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 SpawnerLayers) 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_run_ctx( &self, token: &CapToken, task_id: &StepId, spawner: &Arc<dyn SpawnerAdapter>, run_ctx: Option<&RunContext>, ) -> Result<DispatchOutcome, EngineError>

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 fetch_prompt( &self, token: &CapToken, task_id: &StepId, ) -> Result<Value, EngineError>

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_worker_payload( &self, token: &CapToken, task_id: &StepId, ) -> Result<WorkerPayload, EngineError>

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_trusted( &self, task_id: &StepId, ) -> Result<WorkerPayload, EngineError>

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 materialize_system_file( &self, task_id: &StepId, attempt: u32, ) -> Result<Option<PathBuf>, EngineError>

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 context_policy_for( &self, task_id: &StepId, attempt: u32, ) -> ContextPolicy

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 step_naming_for(&self, task_id: &StepId) -> Option<Arc<StepNaming>>

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 projection_placement_for( &self, task_id: &StepId, ) -> Option<Arc<ProjectionPlacement>>

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 record_worker_stats( &self, task_id: &StepId, attempt: u32, stats: WorkerStats, )

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 take_worker_stats( &self, task_id: &StepId, ) -> Option<(u32, WorkerStats)>

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 trace_handle(&self, task_id: &StepId) -> Option<TraceHandle>

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 agent_context_for( &self, task_id: &StepId, attempt: u32, ) -> Option<AgentContextView>

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 task_attempt(&self, task_id: &StepId) -> Result<u32, EngineError>

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 bake_worker_system_prompt( &self, task_id: &StepId, attempt: u32, system: Option<String>, ) -> Result<(), EngineError>

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 agent_last_rendered_size(&self, agent_name: &str) -> Option<usize>

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 raw_system_prompt( &self, task_id: &StepId, attempt: u32, ) -> Result<Option<String>, EngineError>

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 fetch_data( &self, token: &CapToken, key: &str, ) -> Result<Value, EngineError>

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 submit_output( &self, token: &CapToken, task_id: &StepId, attempt: u32, event: OutputEvent, ) -> Result<(), EngineError>

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 output_tail( &self, task_id: &StepId, attempt: u32, ) -> Vec<OutputEvent>

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 post_result( &self, token: &CapToken, task_id: &StepId, result: Value, ) -> Result<(), EngineError>

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 set_resource( &self, key: impl Into<String>, value: Value, ) -> Result<(), EngineError>

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 query_senior( &self, token: &CapToken, task_id: &StepId, question: Value, ) -> Result<ResumeKey, EngineError>

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 resume( &self, key: ResumeKey, answer: Value, ) -> Result<(), EngineError>

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 await_resume( &self, key: ResumeKey, timeout: Duration, ) -> Result<Value, EngineError>

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 poll_task( &self, token: &CapToken, task_id: &StepId, hold: Duration, ) -> Result<TaskState, EngineError>

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 fn start_detach_loop(&self) -> JoinHandle<()>

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.

Trait Implementations

impl Clone for Engine

fn clone(&self) -> Engine

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.