mlua_swarm_server/

tasks.rs

//! HTTP surface for the Task/Run persistence axis (issue #13 ID-hierarchy
//! reconciliation: Blueprint → Task → Run → Step → Attempt).
//!
//! - `GET  /v1/tasks`          — list every persisted `TaskRecord`, newest first.
//! - `GET  /v1/tasks/:id`      — a `TaskRecord` plus every `RunRecord` kicked from it.
//! - `POST /v1/tasks/:id/runs` — re-kick an existing Task: mints a fresh `RunId`,
//!   re-resolves the stored `blueprint_ref` (refreshing `Blueprint.default_init_ctx`
//!   exactly like original launch time — issue #19 ST4), 3-layer-merges it with
//!   `TaskRecord.input_ctx` and an **optional** [`RunKickRequest`] body's
//!   `init_ctx_override` (see [`merge_init_ctx_3layer`]), dispatches through
//!   `TaskApplication::handle_with_run`, and returns the new `{task_id, run_id}`
//!   pair. A body-less request (or one that omits both fields) preserves the
//!   pre-#19 rekick behavior byte-for-byte.
//! - `GET  /v1/runs/:id`       — a single `RunRecord` (`step_entries` trace included).
//! - `GET  /v1/runs/:id/bindings` — requested/effective binding explain from
//!   the immutable launch snapshot (never from the current Blueprint).
//! - `POST /v1/runs/:id/resume` — resume an `Interrupted` Run under the SAME
//!   `run_id` (replay cursor + stored launch-input snapshot).
//! - `POST /v1/runs/:id/rerun-from` — GH #71 Layer A. Rerun a terminal Run
//!   (`Done` / `Failed` / `Interrupted`) from a caller-specified step under
//!   the SAME `run_id`; physically truncates the replay log at the cut
//!   point so re-dispatch does not collide with the pre-rerun rows. See
//!   [`run_rerun_from`] for the full contract + Known Limitations.
//!
//! `POST /v1/tasks` itself (the flow-eval entry point, `tasks_start` /
//! `run_flow_form`) stays in `crate::lib` — it is the pre-existing
//! Operator-inject-aware dispatch path this module's handlers re-kick
//! through, not a new one. This module owns the read/list/re-kick surface
//! plus the [`finalize_run`] persistence helper both paths share.
//!
//! Authorization follows the same convention as the existing `POST /v1/tasks`
//! entry: no `Authorization` header is required (the route is open), and the
//! only Operator-session correlation available is the request-body-level
//! `operator_sid` (see `crate::TaskLaunchRequest` doc) — this module invents no
//! new auth mechanism.

use axum::{
    extract::{Path, Query, State},
    http::StatusCode,
    Json,
};
use mlua_swarm::application::{
    BlueprintRef, TaskApplicationError, TaskApplicationInput, TaskApplicationOutput,
};
use mlua_swarm::blueprint::{BindRequest, BindingAttestation, BoundAgent};
use mlua_swarm::core::config::CheckPolicy;
use mlua_swarm::service::merge_init_ctx_3layer;
use mlua_swarm::service::TaskLaunchError;
use mlua_swarm::store::replay::ReplayCursor;
use mlua_swarm::store::run::{
    RunContext, RunListFilter, RunRecord, RunStatus, RunStoreError, SnapshotOrigin, StepEntry,
};
use mlua_swarm::store::task::{TaskRecord, TaskRecordStatus, TaskStoreError};
use mlua_swarm::store::trace::{kind as trace_kind, TraceEvent, TraceHandle, TraceQuery};
use mlua_swarm::{
    validate_bound_agent_snapshots, OperatorKind, Role, RunId, TaskId, TaskInputSpec,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::{ApiError, AppState};

/// Current Unix time in whole seconds. `TaskRecord` / `RunRecord` timestamps
/// are `u64` seconds (not milliseconds) — see their field docs in
/// `mlua_swarm::store::task` / `mlua_swarm::store::run`.
pub(crate) fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Serializable mirror of [`TaskApplicationInput`] — the launch-input
/// snapshot persisted into `RunRecord.input_json` at Run-creation time so a
/// later `POST /v1/runs/:id/resume` can rebuild the exact input and re-run
/// the flow under the SAME `run_id`.
///
/// [`TaskApplicationInput`] itself is deliberately not `Serialize`/
/// `Deserialize` (its doc comment explains why — keeping the exhaustive
/// `TaskApplicationInput { .. }` struct literal in the MCP adapter
/// compiling), so this is a dedicated snapshot type with the exact same
/// field set. Every field type already derives serde
/// (`BlueprintRef` / `Role` / `Duration` / `OperatorKind` / `TaskInputSpec`
/// / `CheckPolicy`), so the mirror is total — no field is dropped, and an
/// operator-injected launch round-trips as faithfully as a plain one.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct RunLaunchSnapshot {
    blueprint: BlueprintRef,
    operator_id: String,
    role: Role,
    ttl: Duration,
    init_ctx: Value,
    operator_kind: Option<OperatorKind>,
    bridge_id: Option<String>,
    hook_id: Option<String>,
    operator_backend_id: Option<String>,
    #[serde(default)]
    operator_kind_overrides: HashMap<String, OperatorKind>,
    task_input: Option<TaskInputSpec>,
    check_policy: Option<CheckPolicy>,
}

impl RunLaunchSnapshot {
    /// Capture a launch input as a snapshot (clones each field — the
    /// original is still dispatched).
    fn from_input(input: &TaskApplicationInput) -> Self {
        Self {
            blueprint: input.blueprint.clone(),
            operator_id: input.operator_id.clone(),
            role: input.role,
            ttl: input.ttl,
            init_ctx: input.init_ctx.clone(),
            operator_kind: input.operator_kind,
            bridge_id: input.bridge_id.clone(),
            hook_id: input.hook_id.clone(),
            operator_backend_id: input.operator_backend_id.clone(),
            operator_kind_overrides: input.operator_kind_overrides.clone(),
            task_input: input.task_input.clone(),
            check_policy: input.check_policy,
        }
    }

    /// Rebuild the launch input from a snapshot for resume.
    fn into_input(self) -> TaskApplicationInput {
        TaskApplicationInput {
            blueprint: self.blueprint,
            operator_id: self.operator_id,
            role: self.role,
            ttl: self.ttl,
            init_ctx: self.init_ctx,
            operator_kind: self.operator_kind,
            bridge_id: self.bridge_id,
            hook_id: self.hook_id,
            operator_backend_id: self.operator_backend_id,
            operator_kind_overrides: self.operator_kind_overrides,
            task_input: self.task_input,
            check_policy: self.check_policy,
        }
    }
}

/// Serialize a launch input into the opaque `RunRecord.input_json` blob.
/// Shared by both Run-creation sites (`run_flow_form` in `crate::lib` and
/// [`task_rekick`]) so every persisted Run carries the snapshot resume
/// needs. A serialization failure is a `400` — it means the caller handed
/// in a value the snapshot cannot round-trip, which must surface before the
/// Run is dispatched, not silently.
pub(crate) fn snapshot_launch_input(input: &TaskApplicationInput) -> Result<String, ApiError> {
    serde_json::to_string(&RunLaunchSnapshot::from_input(input))
        .map_err(|e| ApiError::bad_request(format!("launch input snapshot: {e}")))
}

/// Shared finalize step for a dispatched kick: updates the Run's
/// `result_ref` + status and the owning Task's coarse status based on the
/// `TaskApplication::handle_with_run` outcome, then returns that same
/// outcome unchanged so callers keep shaping their own wire response /
/// error.
///
/// Secondary persistence failures (the store call itself erroring) are
/// logged via `tracing::warn!` and otherwise swallowed — they must not mask
/// the primary dispatch outcome the caller already has in hand.
pub(crate) async fn finalize_run(
    state: &AppState,
    task_id: &TaskId,
    run_id: &RunId,
    outcome: Result<TaskApplicationOutput, TaskApplicationError>,
) -> Result<TaskApplicationOutput, TaskApplicationError> {
    match &outcome {
        Ok(out) => {
            if let Err(e) = state
                .run_store
                .set_result(run_id, out.final_ctx.clone())
                .await
            {
                tracing::warn!(%run_id, error = %e, "finalize_run: set_result failed");
            }
            if let Err(e) = state.run_store.update_status(run_id, RunStatus::Done).await {
                tracing::warn!(%run_id, error = %e, "finalize_run: run update_status(Done) failed");
            }
            if let Err(e) = state
                .task_store
                .update_status(task_id, TaskRecordStatus::Done)
                .await
            {
                tracing::warn!(%task_id, error = %e, "finalize_run: task update_status(Done) failed");
            }
        }
        Err(e) => {
            // GH #76 error surface: persist a structured failure envelope into
            // `RunRecord.result_ref` so the async poll path (`GET
            // /v1/runs/:id`) can surface `failed_step` / `verdict_value` /
            // `partial_ctx` symmetric to the sync path's `ApiError`
            // `details` field. Envelope shape (documented for consumer
            // disambiguation from the Ok arm's raw `final_ctx`):
            //
            // ```json
            // {
            //   "error": {
            //     "message": <string>,
            //     "failed_step": <string|null>,
            //     "verdict_value": <value|null>
            //   },
            //   "partial_ctx": <value|null>
            // }
            // ```
            //
            // Consumers detect failure via the top-level `"error"` key
            // (present iff this arm fired; the Ok arm stores the raw
            // `final_ctx` verbatim, which is either a scalar or an object
            // with the user's own keys — never a top-level `"error"`
            // sibling of `"partial_ctx"`). Non-`FlowEval` errors (e.g.
            // `TaskApplicationError::Store` / `NoStore` — dispatch never
            // reached the flow eval boundary) still get an envelope, but
            // with the structural fields `null` (the underlying error
            // simply carries no `failed_step` semantic).
            let envelope = match e {
                TaskApplicationError::Launch(TaskLaunchError::FlowEval {
                    message,
                    failed_step,
                    verdict_value,
                    partial_ctx,
                }) => json!({
                    "error": {
                        "message": message,
                        "failed_step": failed_step,
                        "verdict_value": verdict_value,
                    },
                    "partial_ctx": partial_ctx,
                }),
                other => json!({
                    "error": {
                        "message": other.to_string(),
                        "failed_step": Value::Null,
                        "verdict_value": Value::Null,
                    },
                    "partial_ctx": Value::Null,
                }),
            };
            if let Err(store_err) = state.run_store.set_result(run_id, envelope).await {
                tracing::warn!(%run_id, error = %store_err, "finalize_run: set_result (failure envelope) failed");
            }
            if let Err(store_err) = state
                .run_store
                .update_status(run_id, RunStatus::Failed)
                .await
            {
                tracing::warn!(%run_id, error = %store_err, "finalize_run: run update_status(Failed) failed");
            }
            if let Err(store_err) = state
                .task_store
                .update_status(task_id, TaskRecordStatus::Failed)
                .await
            {
                tracing::warn!(%task_id, error = %store_err, "finalize_run: task update_status(Failed) failed");
            }
            tracing::warn!(%task_id, %run_id, error = %e, "finalize_run: dispatch failed");
        }
    }
    // Trace rail: mark the Run's terminal status on the stream (the
    // `core.run_started` counterpart appended at the launch sites).
    // Best-effort like every other persistence in this fn.
    let status = if outcome.is_ok() { "done" } else { "failed" };
    TraceHandle::new(run_id.clone(), state.run_trace_store.clone())
        .append(
            trace_kind::RUN_FINISHED,
            None,
            None,
            json!({ "status": status }),
        )
        .await;
    outcome
}

/// Query params for `GET /v1/tasks`.
#[derive(Debug, Deserialize, Default)]
pub struct TasksListQuery {
    /// Caps the returned list to the first N entries (already newest-first
    /// per `TaskStore::list`). Omitted = no cap.
    #[serde(default)]
    pub limit: Option<usize>,
}

/// `GET /v1/tasks?limit=N`. Lists every persisted `TaskRecord`, newest first.
pub async fn tasks_list(
    State(state): State<AppState>,
    Query(q): Query<TasksListQuery>,
) -> Result<Json<Vec<TaskRecord>>, ApiError> {
    let mut records = state.task_store.list().await.map_err(ApiError::engine)?;
    if let Some(limit) = q.limit {
        records.truncate(limit);
    }
    Ok(Json(records))
}

/// Response body for `GET /v1/tasks/:id`.
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct TaskDetailResponse {
    /// The Task's own record.
    pub task: TaskRecord,
    /// Every Run kicked from this Task, oldest first (`RunStore::list_by_task` order).
    pub runs: Vec<RunRecord>,
}

/// `GET /v1/tasks/:id`. Returns the `TaskRecord` plus every `RunRecord`
/// kicked from it (`RunStore::list_by_task`, oldest kick first).
pub async fn task_get(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<TaskDetailResponse>, ApiError> {
    let task_id =
        TaskId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid task id: {e}")))?;
    let task = state
        .task_store
        .get(&task_id)
        .await
        .map_err(map_task_store_err)?;
    let runs = state
        .run_store
        .list_by_task(&task_id)
        .await
        .map_err(ApiError::engine)?;
    Ok(Json(TaskDetailResponse { task, runs }))
}

/// Request body for `POST /v1/tasks/:id/runs` (issue #19 ST4) — every
/// field is optional, and the body itself is optional (see
/// [`task_rekick`]'s `Option<Json<Self>>` parameter); a caller that sends
/// no body, or `{}`, or omits a field gets exactly today's rekick
/// behavior for that layer.
#[derive(Debug, Deserialize, Default, schemars::JsonSchema)]
pub struct RunKickRequest {
    /// Per-Run override for the flow-ir initial ctx. Merged on top of
    /// `TaskRecord.input_ctx` (itself already merged on top of
    /// `Blueprint.default_init_ctx` at original launch time) via
    /// [`merge_init_ctx_3layer`] — Run wins on key collision, same
    /// shallow-merge / non-Object-fully-replaces rule as every other
    /// layer in the cascade. `None` (absent field, or no body at all) is
    /// a no-op: the BP+Task merge alone seeds this kick, identical to
    /// pre-#19 rekick.
    #[serde(default)]
    #[schemars(with = "Option<Value>")]
    pub init_ctx_override: Option<Value>,
    /// Per-Run override for the Task-level canonical fields
    /// (`project_root` / `work_dir` / `task_metadata`). `None` falls back
    /// to `TaskRecord.task_input_spec` (the spec resolved and snapshotted
    /// at original `POST /v1/tasks` time); `Some` replaces it wholesale
    /// for this kick only — the stored `TaskRecord.task_input_spec` is
    /// never mutated by a rekick.
    #[serde(default)]
    pub task_input_override: Option<TaskInputSpec>,
    /// Per-Run ceiling (seconds) for this kick's synchronous dispatch
    /// await (issue #35 ST3 — GH #33 Guard 2 parity). `Some(0)` is
    /// rejected (400). `None` falls back to `AppState.sync_timeout_secs`
    /// (the server-wide default), same cascade as
    /// `TaskLaunchRequest.timeout_secs` (`lib.rs:818-826`).
    #[serde(default)]
    pub timeout_secs: Option<u64>,
    /// GH #37: opt into the detached (asynchronous) rekick — same
    /// semantics as `TaskLaunchRequest.detach`. `false` (default) keeps
    /// the synchronous dispatch; `true` spawns the flow eval as a
    /// detached background task bounded by the run TTL alone and returns
    /// `202 Accepted` with `status: "running"` immediately. Mutually
    /// exclusive with `timeout_secs` (`400` when combined).
    #[serde(default)]
    pub detach: bool,
    /// Per-Run pin to a live Operator session (rekick parity with
    /// `POST /v1/tasks`' `operator_sid` — see
    /// `crate::TaskLaunchRequest::operator_sid` for the full
    /// disconnected-vs-unknown / last-write-wins contract). Resolved
    /// before any Task/Run store write: an unknown sid fails fast with a
    /// `400`, never silently falling back to the BP-level alias lookup.
    /// `Some(sid)` becomes this kick's `operator_backend_id` (this handler
    /// carries no other Operator-override field) and is persisted verbatim
    /// into `RunRecord.operator_sid`; `None` (absent field, or no body at
    /// all) preserves the pre-existing Operator-default rekick path
    /// byte-for-byte.
    #[serde(default)]
    pub operator_sid: Option<String>,
}

/// Response body for `POST /v1/tasks/:id/runs`.
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct RunKickResponse {
    /// The re-kicked Task's id (echoes the path param).
    #[schemars(with = "String")]
    pub task_id: TaskId,
    /// The freshly minted Run id for this kick.
    #[schemars(with = "String")]
    pub run_id: RunId,
    /// Kick outcome at response time (GH #37). The synchronous path
    /// reports the dispatched run's terminal-side status (`done`); a
    /// detached kick reports `running` — poll `GET /v1/runs/:id` for the
    /// terminal status and result.
    pub status: RunStatus,
}

/// `POST /v1/tasks/:id/runs`. Re-kicks an existing Task: reads its stored
/// `blueprint_ref`, re-resolves it through [`TaskApplication::resolve`]
/// (issue #19 ST4 — refreshes `Blueprint.default_init_ctx` exactly like
/// original launch time, rather than replaying a launch-time-only
/// snapshot), 3-layer-merges `{bp default, TaskRecord.input_ctx, an
/// optional per-Run override}` via [`merge_init_ctx_3layer`], resolves the
/// Task-level canonical fields (`RunKickRequest.task_input_override`,
/// falling back to `TaskRecord.task_input_spec`), mints a fresh `RunId`,
/// dispatches through `TaskApplication::handle_with_run` (Operator-default
/// unless the caller pins a live session via `RunKickRequest.operator_sid`
/// — the rekick parity for `POST /v1/tasks`' own `operator_sid`; the
/// stored Task carries no persisted Operator preference of its own)
/// plus a freshly-built `RunContext` (issue #13 run_id propagation, so
/// this kick's steps get their own `step_entries` trace), and persists the
/// outcome via [`finalize_run`].
///
/// The body is optional (`Option<Json<RunKickRequest>>`) — no body, or a
/// body with both fields absent, preserves the pre-#19 rekick behavior
/// byte-for-byte (`must_not_simplify #3`).
///
/// Issue #35 ST3 ports the GH #33 sync-hang guards from `run_flow_form` to
/// this handler, both checked before any Task/Run store write: Guard 1
/// (503) fails fast when the resolved Blueprint declares the
/// `operator_delegate` spawner-hint layer and no operator is attached;
/// Guard 2 (504) wraps the dispatch await in `RunKickRequest.timeout_secs`
/// (falling back to the server-wide `sync_timeout_secs`), marking the
/// Run/Task `Failed` rather than leaving them `Running` forever on expiry.
pub async fn task_rekick(
    State(state): State<AppState>,
    Path(id): Path<String>,
    body: Option<Json<RunKickRequest>>,
) -> Result<(StatusCode, Json<RunKickResponse>), ApiError> {
    let task_id =
        TaskId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid task id: {e}")))?;
    let task = state
        .task_store
        .get(&task_id)
        .await
        .map_err(map_task_store_err)?;

    let blueprint_ref: mlua_swarm::application::BlueprintRef =
        serde_json::from_value(task.blueprint_ref.clone()).map_err(|e| {
            ApiError::bad_request(format!(
                "task {task_id}: stored blueprint_ref failed to decode: {e}"
            ))
        })?;

    // issue #19 ST4 (must_not_simplify #5): re-resolve the Blueprint the
    // same way `run_flow_form`'s TTL cascade does, so a store-backed
    // `BlueprintRef::Id` gets its *current* `default_init_ctx` on every
    // rekick rather than whatever was true at original launch time. The
    // Inline path is a pure pass-through, so this is a no-op there.
    let (resolved_bp, _bound_version) = state
        .task_app
        .resolve(&blueprint_ref)
        .await
        .map_err(|e| ApiError::from_task_resolve(&e, &format!("task {task_id}: bp resolve")))?;

    let req = body.map(|Json(r)| r).unwrap_or_default();

    // S2 parity with `run_flow_form` (`lib.rs:1035-1048`): an explicit
    // `operator_sid` pins this rekick to a live Operator session,
    // resolved *before* any Task/Run store write so an unknown sid fails
    // fast with a `400` rather than minting records for a kick that
    // references a session nothing can serve. Unlike `run_flow_form` this
    // handler has no other Operator-override field, so the resolved sid
    // flows straight into `TaskApplicationInput.operator_backend_id`
    // (below) and is persisted verbatim into `RunRecord.operator_sid`. See
    // `crate::TaskLaunchRequest::operator_sid` for the disconnected-vs-
    // unknown distinction.
    let operator_backend_id = match &req.operator_sid {
        Some(sid) => {
            let known_ids = state.engine.list_operator_ids().await;
            if !known_ids.iter().any(|id| id == sid) {
                return Err(ApiError::bad_request(format!(
                    "operator_sid: no such registered operator session '{sid}'"
                )));
            }
            Some(sid.clone())
        }
        None => None,
    };

    // GH #33 Guard 2 ceiling resolution (issue #35 ST3 — mirrors
    // `run_flow_form`'s `lib.rs:813-826` cascade): request field > server
    // config > built-in default. Validated up front, before Guard 1 and
    // before any Task/Run store writes, so a caller-supplied `Some(0)`
    // fails fast with `400` rather than minting records for a rekick that
    // was never going to dispatch.
    // GH #37: `detach: true` makes the sync ceiling meaningless (the
    // detached kick is bounded by the run TTL alone) — combining the two
    // is rejected here, same fail-fast-before-side-effects ordering.
    let detach = req.detach;
    let sync_timeout_secs = match (detach, req.timeout_secs) {
        (true, Some(_)) => {
            return Err(ApiError::bad_request(
                "timeout_secs is the synchronous rekick ceiling and does not apply to a \
                 detached rekick (detach: true), whose lifetime bound is the run TTL — omit \
                 timeout_secs"
                    .into(),
            ));
        }
        (false, Some(0)) => {
            return Err(ApiError::bad_request(
                "timeout_secs: 0 is invalid; omit the field to use the server default".into(),
            ));
        }
        (false, Some(v)) => v,
        (_, None) => state.sync_timeout_secs,
    };

    // GH #33 Guard 1 (issue #35 — adapted signal): the
    // per-request `operator_sid` above already fail-fasts an *unknown*
    // sid, but a rekick with no `operator_sid` still has no per-request
    // "operator backend referenced" signal of its own (unlike
    // `run_flow_form`, whose `op_req.operator_backend_id` is sourced from
    // `TaskLaunchRequest.operator`). The adapted signal is the Blueprint's
    // own `spawner_hints.layers`: when the resolved Blueprint declares the
    // `operator_delegate` layer and zero operators are attached at all,
    // fail fast rather than dispatching into a session nothing can serve.
    // Same ordering invariant `run_flow_form` observes: this check runs
    // before any Task/Run row is touched (no side effects on the 503
    // path).
    if resolved_bp
        .spawner_hints
        .layers
        .iter()
        .any(|l| l == "operator_delegate")
    {
        let attached = state.engine.list_operator_ids().await;
        if attached.is_empty() {
            return Err(ApiError::unavailable(format!(
                "no operator attached to serve this rekick (task {task_id}'s \
                 Blueprint declares the operator_delegate layer): attach an \
                 operator via POST /v1/operators + WS, or use the poll-style \
                 flow (GET /v1/worker/prompt + POST /v1/worker/submit)"
            )));
        }
    }

    let merged_init_ctx = merge_init_ctx_3layer(
        resolved_bp.default_init_ctx.as_ref(),
        &task.input_ctx,
        req.init_ctx_override.as_ref(),
    );

    // must_not_simplify #4: `task_input_override` wins for this kick only;
    // falling back to the Task-level snapshot never mutates
    // `TaskRecord.task_input_spec` itself.
    let task_input_spec: Option<TaskInputSpec> = match req.task_input_override {
        Some(over) => Some(over),
        None => task
            .task_input_spec
            .as_ref()
            .map(|v| serde_json::from_value(v.clone()))
            .transpose()
            .map_err(|e| {
                ApiError::bad_request(format!(
                    "task {task_id}: stored task_input_spec failed to decode: {e}"
                ))
            })?,
    };

    let run_id = RunId::new();
    let now = now_secs();

    let input = TaskApplicationInput {
        blueprint: blueprint_ref,
        operator_id: "http-run".to_string(),
        role: Role::Operator,
        ttl: Duration::from_secs(crate::default_run_ttl()),
        init_ctx: merged_init_ctx,
        operator_kind: None,
        bridge_id: None,
        hook_id: None,
        operator_backend_id,
        operator_kind_overrides: HashMap::new(),
        task_input: task_input_spec,
        // This legacy `POST /v1/tasks/:id/runs`-style path does not carry a
        // per-request check_policy override; `None` preserves the
        // server-wide default (backward compat).
        check_policy: None,
    };
    // Persist a launch-input snapshot so this kick's Run can be resumed
    // under the same run_id if it is later interrupted
    // (`POST /v1/runs/:id/resume`). Built from `input` before it is moved
    // into the dispatch below.
    let input_json = Some(snapshot_launch_input(&input)?);

    state
        .task_store
        .update_status(&task_id, TaskRecordStatus::Running)
        .await
        .map_err(ApiError::engine)?;
    state
        .run_store
        .create(RunRecord {
            id: run_id.clone(),
            task_id: task_id.clone(),
            status: RunStatus::Running,
            step_entries: Vec::new(),
            degradations: Vec::new(),
            operator_sid: req.operator_sid.clone(),
            result_ref: None,
            input_json,
            created_at: now,
            updated_at: now,
        })
        .await
        .map_err(ApiError::engine)?;

    let trace = TraceHandle::new(run_id.clone(), state.run_trace_store.clone());
    trace
        .append(
            trace_kind::RUN_STARTED,
            None,
            None,
            json!({"mode": "rekick"}),
        )
        .await;
    let run_ctx = RunContext::new(run_id.clone(), state.run_store.clone())
        .with_replay_store(state.replay_store.clone())
        .with_trace(trace);

    // GH #37 detached rekick: same driver-detach semantics as
    // `run_flow_form` — the eval runs in its own spawned task bounded by
    // the run TTL alone, `finalize_run` (or the ttl-expiry `Failed`
    // marking) is owned by that task, and this handler returns `202
    // Accepted` immediately.
    if detach {
        let ttl_secs = crate::default_run_ttl();
        let bg_state = state.clone();
        let bg_task_id = task_id.clone();
        let bg_run_id = run_id.clone();
        tokio::spawn(async move {
            let outcome = match tokio::time::timeout(
                Duration::from_secs(ttl_secs),
                bg_state.task_app.handle_with_run(input, Some(run_ctx)),
            )
            .await
            {
                Ok(outcome) => outcome,
                Err(_elapsed) => {
                    let reason = serde_json::json!({
                        "error": format!("detached rekick exceeded {ttl_secs}s ttl ceiling"),
                    });
                    if let Err(e) = bg_state.run_store.set_result(&bg_run_id, reason).await {
                        tracing::warn!(%bg_run_id, error = %e, "task_rekick: detached ttl set_result failed");
                    }
                    if let Err(e) = bg_state
                        .run_store
                        .update_status(&bg_run_id, RunStatus::Failed)
                        .await
                    {
                        tracing::warn!(%bg_run_id, error = %e, "task_rekick: detached ttl run update_status failed");
                    }
                    if let Err(e) = bg_state
                        .task_store
                        .update_status(&bg_task_id, TaskRecordStatus::Failed)
                        .await
                    {
                        tracing::warn!(%bg_task_id, error = %e, "task_rekick: detached ttl task update_status failed");
                    }
                    // This arm never reaches `finalize_run`, so the trace
                    // stream gets its terminal marker here.
                    TraceHandle::new(bg_run_id.clone(), bg_state.run_trace_store.clone())
                        .append(
                            trace_kind::RUN_FINISHED,
                            None,
                            None,
                            json!({ "status": "failed", "reason": format!("ttl {ttl_secs}s exceeded") }),
                        )
                        .await;
                    return;
                }
            };
            // `finalize_run` persists both the Ok and Err outcomes itself;
            // the passthrough return value has no consumer here.
            let _ = finalize_run(&bg_state, &bg_task_id, &bg_run_id, outcome).await;
        });
        return Ok((
            StatusCode::ACCEPTED,
            Json(RunKickResponse {
                task_id,
                run_id,
                status: RunStatus::Running,
            }),
        ));
    }

    // GH #33 Guard 2 (issue #35 ST3 — mirrors `run_flow_form`'s
    // `lib.rs:935-990` exactly): the single await point this handler
    // blocks on. On expiry the timed-out future is dropped, cancelling the
    // in-process flow eval — the flow is abandoned, not resumed. Best
    // effort: mark the Run/Task so they do not stay `Running` forever.
    let outcome = match tokio::time::timeout(
        Duration::from_secs(sync_timeout_secs),
        state.task_app.handle_with_run(input, Some(run_ctx)),
    )
    .await
    {
        Ok(outcome) => outcome,
        Err(_elapsed) => {
            let reason = serde_json::json!({
                "error": format!("sync rekick exceeded {sync_timeout_secs}s timeout ceiling")
            });
            if let Err(e) = state.run_store.set_result(&run_id, reason).await {
                tracing::warn!(%run_id, error = %e, "task_rekick: timeout set_result failed");
            }
            if let Err(e) = state
                .run_store
                .update_status(&run_id, RunStatus::Failed)
                .await
            {
                tracing::warn!(%run_id, error = %e, "task_rekick: timeout run update_status failed");
            }
            if let Err(e) = state
                .task_store
                .update_status(&task_id, TaskRecordStatus::Failed)
                .await
            {
                tracing::warn!(%task_id, error = %e, "task_rekick: timeout task update_status failed");
            }
            return Err(ApiError::timeout(format!(
                "sync rekick exceeded {sync_timeout_secs}s timeout ceiling: task {task_id}, run {run_id}"
            )));
        }
    };
    finalize_run(&state, &task_id, &run_id, outcome)
        .await
        .map_err(|e| ApiError::bad_request(format!("run: {e}")))?;

    Ok((
        StatusCode::CREATED,
        Json(RunKickResponse {
            task_id,
            run_id,
            status: RunStatus::Done,
        }),
    ))
}

/// Response body for `POST /v1/runs/:id/resume`.
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct RunResumeResponse {
    /// The resumed Run's id — echoes the path param. Resume never mints a
    /// new `RunId`; the interrupted Run is re-run in place so its
    /// replay-entry Ctx snapshots (which bake this id into
    /// `meta.runtime[run_id]`) stay consistent.
    #[schemars(with = "String")]
    pub run_id: RunId,
    /// The Task this Run belongs to.
    #[schemars(with = "String")]
    pub task_id: TaskId,
    /// Count of already-completed steps handed to the replay cursor — the
    /// engine returns each of these verbatim (no re-dispatch) before
    /// resuming fresh work. `0` = the Run was interrupted before any step
    /// completed, so it re-runs from scratch under the same `run_id`.
    pub replayed_steps: usize,
}

/// `POST /v1/runs/:id/resume`. Resumes an `Interrupted` Run under the SAME
/// `run_id` (no new `RunId` is minted): the stored launch-input snapshot
/// (`RunRecord.input_json`) is rebuilt into a `TaskApplicationInput`, a
/// `ReplayCursor` is built from the Run's logged step snapshots
/// (`ReplayStore::list_by_run`), and the flow is re-dispatched with both
/// wired into a fresh `RunContext`. On dispatch the engine's replay path
/// returns each already-completed step's stored value verbatim (cursor hit,
/// no Adapter spawn) and dispatches only the steps that never finished —
/// reconstructing the same final Ctx a restart-free run would have reached.
///
/// Status codes:
/// - `404` — no Run with this id.
/// - `409` — the Run is not `Interrupted` (already `Running` / `Done` /
///   `Failed` / `Pending`), OR a concurrent resume already won the
///   `Interrupted -> Running` compare-and-set (double-resume guard).
/// - `422` — the Run has no recorded launch-input snapshot, so it cannot be
///   resumed (an older row predating resume support, or a path that does
///   not persist one).
/// - `202 Accepted` — resume accepted; the flow re-runs in a detached
///   background task (same `tokio::spawn` + run-TTL ceiling shape as a
///   detached rekick). Poll `GET /v1/runs/:id` for the terminal status.
///
/// The launch-input decode and the `422` check run BEFORE the
/// compare-and-set so a non-resumable Run is never flipped to `Running`
/// and stranded without a driver behind it.
pub async fn run_resume(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<(StatusCode, Json<RunResumeResponse>), ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;

    // 404 when the Run does not exist.
    let run = state
        .run_store
        .get(&run_id)
        .await
        .map_err(map_run_store_err)?;

    // Status gate: only an `Interrupted` Run can be resumed.
    if run.status != RunStatus::Interrupted {
        return Err(ApiError::conflict(format!(
            "run {run_id} is {:?}, not Interrupted; only an interrupted run can be resumed",
            run.status
        )));
    }

    // Decode the launch-input snapshot BEFORE the compare-and-set: a Run
    // with no recorded input can never be resumed, and returning `422`
    // here — before flipping the status — avoids stranding it in `Running`
    // with no driver behind it.
    let Some(input_json) = run.input_json.clone() else {
        return Err(ApiError::unprocessable(format!(
            "run {run_id} cannot be resumed: no launch input was recorded for it (it \
             predates resume support, or was created by a path that does not persist one)"
        )));
    };
    let snapshot_value: Value = serde_json::from_str(&input_json).map_err(|e| {
        ApiError::unprocessable(format!(
            "run {run_id}: stored launch input failed to decode: {e}"
        ))
    })?;
    validated_bound_agents_from_snapshot(&run_id, &snapshot_value)?;
    let snapshot: RunLaunchSnapshot = serde_json::from_value(snapshot_value).map_err(|e| {
        ApiError::unprocessable(format!(
            "run {run_id}: stored launch input failed to decode: {e}"
        ))
    })?;

    // Atomically flip Interrupted -> Running. A racing double resume loses
    // the compare-and-set and gets a `409` rather than dispatching a second
    // driver over the same Run.
    let won = state
        .run_store
        .try_transition(&run_id, RunStatus::Interrupted, RunStatus::Running)
        .await
        .map_err(ApiError::engine)?;
    if !won {
        return Err(ApiError::conflict(format!(
            "run {run_id} was concurrently resumed (or left the Interrupted state); it is \
             no longer resumable"
        )));
    }

    // Build the replay cursor from the Run's logged step snapshots. An
    // empty log is fine — the cursor has zero hits and every step is
    // dispatched fresh (a from-scratch re-run under the same run_id).
    let entries = state
        .replay_store
        .list_by_run(&run_id)
        .await
        .map_err(|e| ApiError::engine(format!("replay list_by_run: {e}")))?;
    let replayed_steps = entries.len();
    let cursor = ReplayCursor::from_entries(entries);

    // RunContext for the SAME run_id — run_store + replay_store +
    // replay_cursor all wired. No new RunRecord is minted. `with_resume()`
    // marks this as a resume so any binding backfill is stamped
    // `resume_backfill` (and, D2, keeps legacy replay keys).
    let trace = TraceHandle::new(run_id.clone(), state.run_trace_store.clone());
    trace
        .append(
            trace_kind::RUN_STARTED,
            None,
            None,
            json!({"mode": "resume"}),
        )
        .await;
    let run_ctx = RunContext::new(run_id.clone(), state.run_store.clone())
        .with_replay_store(state.replay_store.clone())
        .with_replay_cursor(Arc::new(Mutex::new(cursor)))
        .with_resume()
        .with_trace(trace);

    let input = snapshot.into_input();
    let task_id = run.task_id.clone();

    // A resumed Task is running again; finalize_run resets it to
    // Done/Failed at the end, same as the rekick path.
    state
        .task_store
        .update_status(&task_id, TaskRecordStatus::Running)
        .await
        .map_err(ApiError::engine)?;

    // Detached dispatch — same `tokio::spawn` + run-TTL-ceiling shape as
    // the detached rekick path; `finalize_run` (or the ttl-expiry `Failed`
    // marking) owns the terminal persistence.
    let ttl_secs = crate::default_run_ttl();
    let bg_state = state.clone();
    let bg_task_id = task_id.clone();
    let bg_run_id = run_id.clone();
    tokio::spawn(async move {
        let outcome = match tokio::time::timeout(
            Duration::from_secs(ttl_secs),
            bg_state.task_app.handle_with_run(input, Some(run_ctx)),
        )
        .await
        {
            Ok(outcome) => outcome,
            Err(_elapsed) => {
                let reason = serde_json::json!({
                    "error": format!("resumed run exceeded {ttl_secs}s ttl ceiling"),
                });
                if let Err(e) = bg_state.run_store.set_result(&bg_run_id, reason).await {
                    tracing::warn!(%bg_run_id, error = %e, "run_resume: ttl set_result failed");
                }
                if let Err(e) = bg_state
                    .run_store
                    .update_status(&bg_run_id, RunStatus::Failed)
                    .await
                {
                    tracing::warn!(%bg_run_id, error = %e, "run_resume: ttl run update_status failed");
                }
                if let Err(e) = bg_state
                    .task_store
                    .update_status(&bg_task_id, TaskRecordStatus::Failed)
                    .await
                {
                    tracing::warn!(%bg_task_id, error = %e, "run_resume: ttl task update_status failed");
                }
                return;
            }
        };
        // `finalize_run` persists both the Ok and Err outcomes itself.
        let _ = finalize_run(&bg_state, &bg_task_id, &bg_run_id, outcome).await;
    });

    Ok((
        StatusCode::ACCEPTED,
        Json(RunResumeResponse {
            run_id,
            task_id,
            replayed_steps,
        }),
    ))
}

/// Request body for `POST /v1/runs/:id/rerun-from` (GH #71 Layer A).
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct RunRerunFromRequest {
    /// The step to re-execute. This is a raw `step_ref` (the agent name the
    /// dispatcher recorded as `ReplayEntry.step_ref`), NOT a projection
    /// canonical name. See [`run_rerun_from`] doc for the Known Limitations
    /// this carries (loop bodies match the first occurrence,
    /// `AgentMeta.projection_name` is not resolved, `BlueprintRef::Inline`
    /// re-decodes the frozen inline BP).
    pub from_step: String,
}

/// Response body for `POST /v1/runs/:id/rerun-from` (GH #71 Layer A).
#[derive(Debug, Clone, Serialize, schemars::JsonSchema)]
pub struct RunRerunFromResponse {
    /// The rerun's Run id — echoes the path param. Rerun-from-step never
    /// mints a new `RunId`; it re-runs in place so the replay-entry Ctx
    /// snapshots (which bake this id into `meta.runtime[run_id]`) stay
    /// consistent.
    #[schemars(with = "String")]
    pub run_id: RunId,
    /// The Task this Run belongs to.
    #[schemars(with = "String")]
    pub task_id: TaskId,
    /// Count of pre-cut entries handed to the replay cursor — each is
    /// returned verbatim by the engine before fresh dispatch resumes at
    /// the cut point.
    pub replayed_steps: usize,
    /// Count of entries physically dropped from the replay store — the
    /// target step's row plus every downstream row.
    pub dropped_steps: usize,
}

/// `POST /v1/runs/:id/rerun-from` — GH #71 Layer A. Re-executes a specific
/// step (and every downstream step) of a terminal Run under the SAME
/// `run_id`. Mirrors [`run_resume`], with two deltas: it accepts any
/// terminal status (`Done` / `Failed` / `Interrupted`) rather than only
/// `Interrupted`, and it physically truncates the replay log at the cut
/// point (via [`crate::AppState::replay_store`]'s `delete_from`) so that
/// re-dispatch's `append` does not collide with the pre-rerun row and so
/// `list_by_run` reflects the rerun's real history rather than the
/// pre-rerun ghost.
///
/// # Status codes
///
/// - `400` — invalid `run_id`, malformed body, or launch-snapshot decode failure.
/// - `404` — no Run with this id.
/// - `409` — the Run is `Running` / `Pending` (would race the in-flight
///   driver), OR a concurrent transition won the compare-and-set.
/// - `422` — the Run has no recorded launch-input snapshot, OR `from_step`
///   is not present in this Run's replay log, OR the run's replay log is
///   empty next to a non-empty `RunRecord.step_entries` trace (a prior
///   `rerun-from` reached the truncate stage and consumed the log), OR
///   the current-head Blueprint fails to compile (unresolved
///   `operator_ref` etc.) — the deterministic pre-flight gate that keeps
///   the replay log untouched on a compile-fail.
/// - `202 Accepted` — accepted; the flow re-runs in a detached background
///   task (same `tokio::spawn` + run-TTL ceiling shape as [`run_resume`]).
///   Poll `GET /v1/runs/:id` for the terminal status.
///
/// # Order of operations
///
/// The compare-and-set runs BEFORE the `delete_from` on purpose: a losing
/// cas returns `409` without ever touching the store, so a lost race can
/// never leave the store truncated while the status stayed at its old
/// terminal value. The compile pre-check runs BEFORE the compare-and-set
/// for the same reason: a deterministic compile failure fires a `422`
/// that leaves both `status` and the replay log untouched, so the caller
/// can fix the Blueprint and retry against the same run.
///
/// 1. 404 check.
/// 2. Status gate (fast 409 for `Running` / `Pending`).
/// 3. Decode launch snapshot (fast 400 / 422).
/// 4. Compute cut index via `list_by_run` + `.position(step_ref == from_step)`
///    (fast 422 when the step is not present, with a distinct message when
///    the log is empty but `RunRecord.step_entries` shows the run did
///    trace steps — a consumed log from a prior `rerun-from`).
/// 5. Pre-flight compile check via `TaskApplication::precompile` against
///    the launch snapshot's Blueprint (fast 422 on any `CompileError`).
///    Prevents compile-fail-inside-`tokio::spawn` from consuming the
///    replay log via step 7's `delete_from`.
/// 6. Atomic transition `<current terminal> -> Running` (409 on loss).
/// 7. Physical `delete_from(cut)` on the replay store — safe now because we
///    won the cas and own the Run.
/// 8. Build `ReplayCursor` from the truncated entries.
/// 9. Detached dispatch, same `tokio::spawn` + `default_run_ttl` shape as
///    [`run_resume`].
///
/// # Known limitations (Layer A)
///
/// 1. **`from_step` is a raw `step_ref` (agent name)** — projection alias
///    resolution via `StepNaming` is Layer B territory. For undeclared
///    steps `step_ref == canonical` so this is only visible when
///    `AgentMeta.projection_name` is in use.
/// 2. **`BlueprintRef::Inline` freezes the BP in the launch snapshot** —
///    the rerun re-decodes the same inline BP, so agent-definition edits
///    landed on disk between the original dispatch and the rerun are NOT
///    honored for inline runs. Use `BlueprintRef::Id` for the
///    iterate-and-rerun workflow.
/// 3. **Loop bodies match the first occurrence** — `step_ref` is the agent
///    name, so `.position(|e| e.step_ref == from_step)` finds the FIRST
///    occurrence and truncates from there. Rerunning a specific loop
///    iteration needs Layer B semantics.
/// 4. **Structural BP change is out of scope** — if steps were added /
///    removed / reordered between the original dispatch and the rerun,
///    the flow-ir re-eval will naturally miss the step or dispatch a
///    different downstream. Start a fresh run in that case.
pub async fn run_rerun_from(
    State(state): State<AppState>,
    Path(id): Path<String>,
    Json(req): Json<RunRerunFromRequest>,
) -> Result<(StatusCode, Json<RunRerunFromResponse>), ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;

    if req.from_step.trim().is_empty() {
        return Err(ApiError::bad_request(
            "from_step must be a non-empty step ref".to_string(),
        ));
    }

    // 404 when the Run does not exist.
    let run = state
        .run_store
        .get(&run_id)
        .await
        .map_err(map_run_store_err)?;

    // Status gate — reject in-flight statuses that would race the driver
    // already dispatching against this run_id.
    let current = run.status;
    match current {
        RunStatus::Done | RunStatus::Failed | RunStatus::Interrupted | RunStatus::Cancelled => { /* ok */
        }
        RunStatus::Running | RunStatus::Pending => {
            return Err(ApiError::conflict(format!(
                "run {run_id} is {current:?}; rerun-from requires a terminal run \
                 (Done / Failed / Interrupted / Cancelled)"
            )));
        }
    }

    // Decode the launch-input snapshot BEFORE the compare-and-set: a Run
    // with no recorded input can never be rerun-from, and returning `422`
    // here — before flipping the status — avoids stranding it in `Running`
    // with no driver behind it.
    let Some(input_json) = run.input_json.clone() else {
        return Err(ApiError::unprocessable(format!(
            "run {run_id} cannot be rerun: no launch input was recorded for it (it \
             predates resume/rerun support, or was created by a path that does not \
             persist one)"
        )));
    };
    let snapshot_value: Value = serde_json::from_str(&input_json).map_err(|e| {
        ApiError::unprocessable(format!(
            "run {run_id}: stored launch input failed to decode: {e}"
        ))
    })?;
    validated_bound_agents_from_snapshot(&run_id, &snapshot_value)?;
    let snapshot: RunLaunchSnapshot = serde_json::from_value(snapshot_value).map_err(|e| {
        ApiError::unprocessable(format!(
            "run {run_id}: stored launch input failed to decode: {e}"
        ))
    })?;

    // Load the replay log and locate the cut point via first-match on
    // `step_ref`. See §Known limitations #3 (loop bodies).
    let entries = state
        .replay_store
        .list_by_run(&run_id)
        .await
        .map_err(|e| ApiError::engine(format!("replay list_by_run: {e}")))?;
    let cut = entries
        .iter()
        .position(|e| e.step_ref == req.from_step)
        .ok_or_else(|| {
            // Distinguish two shapes of miss: (a) the log carries entries
            // but none match `from_step` (typo or wrong step name); (b) the
            // log is empty while `RunRecord.step_entries` still traces
            // steps — which means a prior `rerun-from` reached the
            // `delete_from` stage and consumed the log, and no further
            // `rerun-from` against the same run is recoverable. `run.
            // step_entries` and `replay_store` are physically separate
            // tables (the dispatcher writes to both), so an empty log next
            // to a non-empty trace is the reliable tell.
            if entries.is_empty() && !run.step_entries.is_empty() {
                ApiError::unprocessable(format!(
                    "run {run_id}: replay log is empty but {} step entries are traced \
                     on the RunRecord — the log was consumed by a prior rerun-from \
                     that reached the truncate stage. This run can no longer be \
                     rerun-from; start a fresh run via POST /v1/tasks.",
                    run.step_entries.len()
                ))
            } else {
                ApiError::unprocessable(format!(
                    "run {run_id}: from_step {:?} not present in this run's replay log \
                     (nothing to rerun-from)",
                    req.from_step
                ))
            }
        })?;

    // Pre-flight compile check against the current-head Blueprint the
    // rerun will actually launch against. Compile is deterministic — an
    // `UnresolvedOperatorRef` / `UnresolvedMetaRef` / `UnresolvedAuditAgent`
    // / verdict-cond shape violation fails the same way every attempt —
    // so surfacing it here as a 422, BEFORE the compare-and-set and
    // BEFORE `delete_from`, converts an otherwise irrecoverable replay-
    // loss (compile fails INSIDE the detached `tokio::spawn` AFTER the
    // truncation has physically dropped the pre-cut rows) into a fast
    // rejection that leaves the run's status and replay log entirely
    // untouched. Runtime-only failures (spawner error, worker submit
    // failure) are still able to consume the log — inherent to any
    // path that can only be discovered mid-dispatch — but that class
    // needs a different fix (Layer B territory).
    if let Err(e) = state.task_app.precompile(&snapshot.blueprint).await {
        return Err(ApiError::unprocessable(format!(
            "run {run_id} cannot be rerun: current-head Blueprint fails to compile — {e}"
        )));
    }

    // Atomically flip the current terminal status -> Running. A racing
    // rerun (or a boot-time recovery sweep, or a concurrent resume) loses
    // the compare-and-set and gets `409` rather than dispatching a second
    // driver over the same Run.
    let won = state
        .run_store
        .try_transition(&run_id, current, RunStatus::Running)
        .await
        .map_err(ApiError::engine)?;
    if !won {
        return Err(ApiError::conflict(format!(
            "run {run_id} was concurrently transitioned (or left the {current:?} state); \
             it is no longer rerunnable"
        )));
    }

    // We own the run now — physically truncate the replay log at the cut
    // so the rerun dispatch's `append` cannot collide with the pre-rerun
    // row and `list_by_run` reflects the rerun's real history rather than
    // the pre-rerun ghost.
    let dropped_steps = state
        .replay_store
        .delete_from(&run_id, cut)
        .await
        .map_err(|e| ApiError::engine(format!("replay delete_from: {e}")))?;

    // Cursor is built from the pre-cut prefix; every retained entry hits
    // verbatim in the engine's replay path.
    let kept = entries.into_iter().take(cut).collect::<Vec<_>>();
    let replayed_steps = kept.len();
    let cursor = ReplayCursor::from_entries(kept);

    // `with_resume()` — a rerun-from re-derives its snapshot from the current
    // Blueprint exactly like resume, so a binding backfill here is stamped
    // `resume_backfill` (and keeps legacy replay keys, D2).
    let trace = TraceHandle::new(run_id.clone(), state.run_trace_store.clone());
    trace
        .append(
            trace_kind::RUN_STARTED,
            None,
            None,
            json!({"mode": "rerun_from"}),
        )
        .await;
    let run_ctx = RunContext::new(run_id.clone(), state.run_store.clone())
        .with_replay_store(state.replay_store.clone())
        .with_replay_cursor(Arc::new(Mutex::new(cursor)))
        .with_resume()
        .with_trace(trace);

    let input = snapshot.into_input();
    let task_id = run.task_id.clone();

    // A rerun-from is a running Run again; finalize_run resets it to
    // Done/Failed at the end, same as the rekick / resume paths.
    state
        .task_store
        .update_status(&task_id, TaskRecordStatus::Running)
        .await
        .map_err(ApiError::engine)?;

    let ttl_secs = crate::default_run_ttl();
    let bg_state = state.clone();
    let bg_task_id = task_id.clone();
    let bg_run_id = run_id.clone();
    tokio::spawn(async move {
        let outcome = match tokio::time::timeout(
            Duration::from_secs(ttl_secs),
            bg_state.task_app.handle_with_run(input, Some(run_ctx)),
        )
        .await
        {
            Ok(outcome) => outcome,
            Err(_elapsed) => {
                let reason = serde_json::json!({
                    "error": format!("rerun-from run exceeded {ttl_secs}s ttl ceiling"),
                });
                if let Err(e) = bg_state.run_store.set_result(&bg_run_id, reason).await {
                    tracing::warn!(%bg_run_id, error = %e, "run_rerun_from: ttl set_result failed");
                }
                if let Err(e) = bg_state
                    .run_store
                    .update_status(&bg_run_id, RunStatus::Failed)
                    .await
                {
                    tracing::warn!(%bg_run_id, error = %e, "run_rerun_from: ttl run update_status failed");
                }
                if let Err(e) = bg_state
                    .task_store
                    .update_status(&bg_task_id, TaskRecordStatus::Failed)
                    .await
                {
                    tracing::warn!(%bg_task_id, error = %e, "run_rerun_from: ttl task update_status failed");
                }
                return;
            }
        };
        let _ = finalize_run(&bg_state, &bg_task_id, &bg_run_id, outcome).await;
    });

    Ok((
        StatusCode::ACCEPTED,
        Json(RunRerunFromResponse {
            run_id,
            task_id,
            replayed_steps,
            dropped_steps,
        }),
    ))
}

/// `GET /v1/runs/:id`. Returns a single `RunRecord` (its `step_entries`
/// trace included).
pub async fn run_get(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<RunRecord>, ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;
    let run = state
        .run_store
        .get(&run_id)
        .await
        .map_err(map_run_store_err)?;
    Ok(Json(run))
}

/// Query params for `GET /v1/runs` (the Run collection read).
#[derive(Debug, Deserialize, Default)]
pub struct RunsListQuery {
    /// Only Runs kicked from this Task.
    #[serde(default)]
    pub task_id: Option<String>,
    /// Only Runs currently in this status (`pending` / `running` / `done`
    /// / `failed` / `interrupted`).
    #[serde(default)]
    pub status: Option<String>,
    /// Page size cap. Omitted = no cap.
    #[serde(default)]
    pub limit: Option<usize>,
    /// Skip the first N matching rows (after newest-first ordering).
    #[serde(default)]
    pub offset: Option<usize>,
}

/// Response body for `GET /v1/runs`.
#[derive(Debug, Serialize)]
pub struct RunsListResponse {
    /// Matching Runs, newest-first.
    pub runs: Vec<RunRecord>,
}

/// `GET /v1/runs?task_id=&status=&limit=&offset=` — filtered Run
/// collection, newest-first. The collection read that was missing from
/// the Run CRUD surface (only `GET /v1/runs/:id` existed before the
/// per-step run stats work).
pub async fn runs_list(
    State(state): State<AppState>,
    Query(q): Query<RunsListQuery>,
) -> Result<Json<RunsListResponse>, ApiError> {
    let task_id = q
        .task_id
        .map(TaskId::parse)
        .transpose()
        .map_err(|e| ApiError::bad_request(format!("invalid task_id: {e}")))?;
    let status = q
        .status
        .as_deref()
        .map(|s| {
            serde_json::from_value::<RunStatus>(Value::String(s.to_string())).map_err(|_| {
                ApiError::bad_request(format!(
                    "invalid status {s:?} (expected pending/running/done/failed/interrupted)"
                ))
            })
        })
        .transpose()?;
    let runs = state
        .run_store
        .list(&RunListFilter {
            task_id,
            status,
            limit: q.limit,
            offset: q.offset,
        })
        .await
        .map_err(map_run_store_err)?;
    Ok(Json(RunsListResponse { runs }))
}

/// Response body for `GET /v1/runs/:id/steps`.
#[derive(Debug, Serialize)]
pub struct RunStepsResponse {
    /// The Run the steps belong to.
    pub run_id: String,
    /// Terminal per-step stats entries, in append (dispatch) order.
    pub steps: Vec<StepEntry>,
}

/// `GET /v1/runs/:id/steps` — the Run's terminal per-step stats
/// (`StepEntry` list) as a standalone sub-resource. Same data
/// `GET /v1/runs/:id` embeds; split out so stats consumers don't drag
/// the full RunRecord (launch snapshot etc.) per poll.
pub async fn run_steps(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<RunStepsResponse>, ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;
    let run = state
        .run_store
        .get(&run_id)
        .await
        .map_err(map_run_store_err)?;
    Ok(Json(RunStepsResponse {
        run_id: run.id.to_string(),
        steps: run.step_entries,
    }))
}

/// Query params for `GET /v1/runs/:id/trace` — see
/// `mlua_swarm::store::trace::TraceQuery` for semantics (`latest` wins
/// over `after`; `kind` entries are comma-separated prefix matches).
#[derive(Debug, Deserialize, Default)]
pub struct RunTraceQuery {
    /// Forward-paging cursor: only events with `seq > after`.
    #[serde(default)]
    pub after: Option<u64>,
    /// Page size cap.
    #[serde(default)]
    pub limit: Option<usize>,
    /// Tail mode: the LAST n matching events (ascending order).
    #[serde(default)]
    pub latest: Option<usize>,
    /// Comma-separated kind filters, prefix match (e.g. `kind=mw.` or
    /// `kind=core.step_completed,worker.`).
    #[serde(default)]
    pub kind: Option<String>,
    /// Exact `step_ref` filter.
    #[serde(default)]
    pub step: Option<String>,
    /// Exact attempt filter.
    #[serde(default)]
    pub attempt: Option<u32>,
}

/// Response body for `GET /v1/runs/:id/trace`.
#[derive(Debug, Serialize)]
pub struct RunTraceResponse {
    /// The Run the events belong to.
    pub run_id: String,
    /// Matching trace events, ascending by `seq`.
    pub events: Vec<TraceEvent>,
}

/// `GET /v1/runs/:id/trace?after=&limit=&latest=&kind=&step=&attempt=` —
/// the Run's TraceEvent stream (the RunTrace rail). Note the trace rail
/// is deliberately uncoupled from `RunStore` (a trace can outlive or
/// precede its Run row), so an unknown Run id returns an empty list, not
/// 404.
pub async fn run_trace(
    State(state): State<AppState>,
    Path(id): Path<String>,
    Query(q): Query<RunTraceQuery>,
) -> Result<Json<RunTraceResponse>, ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;
    let query = TraceQuery {
        after: q.after,
        limit: q.limit,
        latest: q.latest,
        kinds: q
            .kind
            .as_deref()
            .map(|s| {
                s.split(',')
                    .map(str::trim)
                    .filter(|k| !k.is_empty())
                    .map(str::to_string)
                    .collect()
            })
            .unwrap_or_default(),
        step_ref: q.step,
        attempt: q.attempt,
    };
    let events = state
        .run_trace_store
        .list(&run_id, &query)
        .await
        .map_err(|e| ApiError::engine(format!("trace list: {e}")))?;
    Ok(Json(RunTraceResponse {
        run_id: run_id.to_string(),
        events,
    }))
}

/// `POST /v1/runs/:id/cancel` — record a cancel request on the Run's
/// trace stream (`core.cancel_requested`) and mark the Run's status
/// to `Cancelled` for still-in-flight rows. Idempotent: repeat calls
/// re-append the trace event but keep the status setter idempotent
/// on the store side. In-flight abort itself remains a v3 carry — the
/// current effect is observational + status marker, matching the
/// `swarm_cancel` MCP tool's local semantics but reflected onto the
/// server-side `RunTraceStore` so `GET /v1/runs/:id/trace` reflects
/// it too.
pub async fn run_cancel(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<axum::http::StatusCode, ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;
    // Load the row to prove it exists; a missing row is a 404 (aligned
    // with `run_delete` — cancel needs an addressable Run).
    let record = state
        .run_store
        .get(&run_id)
        .await
        .map_err(map_run_store_err)?;
    // Best-effort trace append (never gates the response) — the
    // authoritative record is the run_store status update below.
    TraceHandle::new(run_id.clone(), state.run_trace_store.clone())
        .append(trace_kind::CANCEL_REQUESTED, None, None, json!({}))
        .await;
    // Flip the Run's status to Cancelled when it's still non-terminal.
    // Terminal Runs (Done / Failed / Interrupted / already Cancelled)
    // keep their outcome — cancel arriving after finalize is an
    // observation, not a rewrite.
    if matches!(record.status, RunStatus::Pending | RunStatus::Running) {
        if let Err(e) = state
            .run_store
            .update_status(&run_id, RunStatus::Cancelled)
            .await
        {
            tracing::warn!(%run_id, error = %e, "run_cancel: update_status(Cancelled) failed");
        }
    }
    Ok(axum::http::StatusCode::NO_CONTENT)
}

/// `DELETE /v1/runs/:id` — retention prune: deletes the Run row and its
/// trace stream together (`404` when the Run row is absent; the trace
/// stream is pruned best-effort either way). Replay rows are untouched —
/// `ReplayStore` has its own truncation semantics owned by the
/// rerun-from path.
///
/// Trust tier: same auth-free open-router posture as every other route
/// in this module (`POST /v1/tasks` included) — the server is a
/// local-first, loopback-bound single-operator daemon. Flagged in
/// holistic review as the surface's first unauthenticated destructive
/// verb; acceptable under the loopback bind, revisit if the bind ever
/// goes non-local.
pub async fn run_delete(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<axum::http::StatusCode, ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;
    state
        .run_store
        .delete(&run_id)
        .await
        .map_err(map_run_store_err)?;
    if let Err(e) = state.run_trace_store.delete_run(&run_id).await {
        tracing::warn!(%run_id, error = %e, "run_delete: trace delete_run failed (run row already deleted)");
    }
    Ok(axum::http::StatusCode::NO_CONTENT)
}

/// Whether a Run-scoped binding has only a declaration or also carries a
/// provider attestation accepted by Core.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum RunBindingStatus {
    /// No provider attestation was recorded; `requested` is still the exact
    /// declaration pinned at launch time.
    DeclarationOnly,
    /// Core accepted and pinned the provider's effective capability report.
    Attested,
}

/// Mechanical requested/effective comparison for one immutable binding.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, schemars::JsonSchema)]
pub struct RunBindingDifference {
    /// Whether the requested model string and resolved model string differ.
    pub model_changed: bool,
    /// Requested tools absent from the effective grant. Accepted attestations
    /// normally leave this empty because launch validation is fail-closed.
    pub missing_requested_tools: Vec<String>,
    /// Effective tools not present in the minimum requested grant.
    pub additional_effective_tools: Vec<String>,
    /// Whether the requested and effective launch variants differ.
    pub launch_variant_changed: bool,
}

/// Explain view for one agent, derived exclusively from the persisted Run
/// snapshot rather than from the current Blueprint registry.
#[derive(Debug, Clone, PartialEq, Serialize, schemars::JsonSchema)]
pub struct RunBindingExplainEntry {
    /// Logical agent name.
    pub agent: String,
    /// Declaration tier that selected the Runner.
    pub runner_source: mlua_swarm::blueprint::RunnerResolutionSource,
    /// Provider-attestation state.
    pub status: RunBindingStatus,
    /// Exact platform-neutral request reconstructed from the pinned snapshot.
    pub requested: Option<BindRequest>,
    /// Core-validated provider report, when one was accepted at launch.
    pub effective: Option<BindingAttestation>,
    /// Mechanical difference between `requested` and `effective`; absent for
    /// declaration-only bindings.
    pub difference: Option<RunBindingDifference>,
    /// Final immutable replay identity, including the attestation when present.
    pub binding_digest: mlua_swarm::blueprint::BindingDigest,
}

/// Response body for `GET /v1/runs/:id/bindings`.
#[derive(Debug, Clone, PartialEq, Serialize, schemars::JsonSchema)]
pub struct RunBindingsExplainResponse {
    /// Run whose launch snapshot was inspected.
    #[schemars(with = "String")]
    pub run_id: RunId,
    /// Owning Task recorded on that Run.
    #[schemars(with = "String")]
    pub task_id: TaskId,
    /// Provenance of the inspected `bound_agents` snapshot. `launch` means the
    /// bindings were pinned at the Run's initial launch; `resume_backfill`
    /// means they were re-derived from the current Blueprint when a
    /// pre-binding-snapshot Run was resumed/reran — so they carry no
    /// launch-time pin guarantee. A snapshot that carries `bound_agents` but
    /// no origin marker reports `resume_backfill` (the safe side — see
    /// [`SnapshotOrigin::from_snapshot`]).
    pub snapshot_origin: SnapshotOrigin,
    /// Every agent snapshot in Blueprint declaration order.
    pub bindings: Vec<RunBindingExplainEntry>,
}

fn requested_binding(bound: &BoundAgent) -> Option<BindRequest> {
    mlua_swarm::binding_request_for_snapshot(bound)
}

fn binding_difference(
    requested: &BindRequest,
    effective: &BindingAttestation,
) -> RunBindingDifference {
    let missing_requested_tools = requested
        .requested_tools
        .iter()
        .filter(|tool| !effective.effective_tools.contains(tool))
        .cloned()
        .collect();
    let additional_effective_tools = effective
        .effective_tools
        .iter()
        .filter(|tool| !requested.requested_tools.contains(tool))
        .cloned()
        .collect();
    RunBindingDifference {
        model_changed: requested.requested_model != effective.resolved_model,
        missing_requested_tools,
        additional_effective_tools,
        launch_variant_changed: requested.launch_variant != effective.launch_variant,
    }
}

fn validated_bound_agents_from_snapshot(
    run_id: &RunId,
    snapshot: &Value,
) -> Result<Option<Vec<BoundAgent>>, ApiError> {
    let Some(bound_value) = snapshot.get("bound_agents") else {
        return Ok(None);
    };
    let bound_agents: Vec<BoundAgent> =
        serde_json::from_value(bound_value.clone()).map_err(|e| {
            ApiError::unprocessable(format!(
                "run {run_id} contains an invalid binding snapshot: {e}"
            ))
        })?;
    validate_bound_agent_snapshots(&bound_agents).map_err(|error| {
        ApiError::unprocessable(format!(
            "run {run_id} contains an inconsistent binding snapshot: {error}"
        ))
    })?;
    Ok(Some(bound_agents))
}

/// `GET /v1/runs/:id/bindings`. Explains the exact immutable agent bindings
/// used by this Run. The handler never reads or resolves the current Blueprint;
/// old Runs without a binding snapshot return `422` instead of guessed state.
pub async fn run_bindings_explain(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Json<RunBindingsExplainResponse>, ApiError> {
    let run_id =
        RunId::parse(id).map_err(|e| ApiError::bad_request(format!("invalid run id: {e}")))?;
    let run = state
        .run_store
        .get(&run_id)
        .await
        .map_err(map_run_store_err)?;
    let input_json = run.input_json.as_deref().ok_or_else(|| {
        ApiError::unprocessable(format!(
            "run {run_id} has no launch snapshot; binding explain is unavailable"
        ))
    })?;
    let snapshot: Value = serde_json::from_str(input_json).map_err(|e| {
        ApiError::unprocessable(format!(
            "run {run_id} launch snapshot is invalid JSON; binding explain is unavailable: {e}"
        ))
    })?;
    let bound_agents = validated_bound_agents_from_snapshot(&run_id, &snapshot)?.ok_or_else(|| {
        ApiError::unprocessable(format!(
            "run {run_id} predates immutable binding snapshots; current Blueprint state was not consulted"
        ))
    })?;

    let bindings = bound_agents
        .into_iter()
        .map(|bound| {
            let requested = requested_binding(&bound);
            let effective = bound.attestation.clone();
            let difference = requested
                .as_ref()
                .zip(effective.as_ref())
                .map(|(request, attestation)| binding_difference(request, attestation));
            RunBindingExplainEntry {
                agent: bound.agent.name,
                runner_source: bound.runner_source,
                status: if effective.is_some() {
                    RunBindingStatus::Attested
                } else {
                    RunBindingStatus::DeclarationOnly
                },
                requested,
                effective,
                difference,
                binding_digest: bound.binding_digest,
            }
        })
        .collect();

    Ok(Json(RunBindingsExplainResponse {
        run_id: run.id,
        task_id: run.task_id,
        snapshot_origin: SnapshotOrigin::from_snapshot(&snapshot),
        bindings,
    }))
}

/// `pub(crate)` so `crate::projection`'s `GET /v1/tasks/:id/ctx` handler can
/// reuse this module's existing-Task-existence-check error mapping (same
/// 404-vs-500 split `task_get` already applies).
pub(crate) fn map_task_store_err(e: TaskStoreError) -> ApiError {
    match e {
        TaskStoreError::NotFound(id) => ApiError::not_found(format!("task not found: {id}")),
        other => ApiError::engine(other),
    }
}

fn map_run_store_err(e: RunStoreError) -> ApiError {
    match e {
        RunStoreError::NotFound(id) => ApiError::not_found(format!("run not found: {id}")),
        other => ApiError::engine(other),
    }
}

// ──────────────────────────────────────────────────────────────────────────
// UT
// ──────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use mlua_swarm::application::BlueprintRef;
    use mlua_swarm::blueprint::{
        current_schema_version, AgentDef, AgentKind, Blueprint, BlueprintMetadata, CompilerHints,
        CompilerStrategy, Runner,
    };
    use mlua_swarm::core::config::EngineCfg;
    use mlua_swarm::core::engine::Engine;
    use mlua_swarm::store::output::InMemoryOutputStore;
    use mlua_swarm::store::run::InMemoryRunStore;
    use mlua_swarm::store::task::InMemoryTaskStore;
    use std::collections::HashMap;
    use std::sync::Arc;
    use tokio::sync::Mutex;

    /// A single-step flow.ir Blueprint that always succeeds: `Step { ref:
    /// "identity", in: lit("hello"), out: $.out }` against the baseline
    /// `RustFn` identity worker (same shape as `seed_blueprint` in
    /// `mlua-swarm-cli`'s `serve.rs`, self-contained here rather than
    /// importing a binary crate).
    fn identity_blueprint() -> Blueprint {
        Blueprint {
            schema_version: current_schema_version(),
            id: "tasks-test-bp".into(),
            flow: serde_json::from_value(serde_json::json!({
                "kind": "step",
                "ref": mlua_swarm::worker::baseline::AG_IDENTITY,
                "in": {"op": "lit", "value": "hello"},
                "out": {"op": "path", "at": "$.out"},
            }))
            .expect("flow parse"),
            agents: vec![AgentDef {
                name: mlua_swarm::worker::baseline::AG_IDENTITY.into(),
                kind: AgentKind::RustFn,
                spec: serde_json::json!({"fn_id": mlua_swarm::worker::baseline::AG_IDENTITY}),
                profile: None,
                meta: None,
                runner: None,
                runner_ref: None,
                verdict: 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(),
        }
    }

    /// Minimal `AppState` for handler-level tests — mirrors the construction
    /// `build_router_full` does internally, but skips the `Router` wrapper so
    /// tests can call handler functions directly (this crate's established
    /// unit-test convention; see e.g. `operator_ws::login`'s tests).
    fn test_state() -> AppState {
        let engine = Engine::new_with_layers(EngineCfg::default(), crate::default_layer_registry());
        let compiler = mlua_swarm::Compiler::new(crate::default_registry());
        let launch = Arc::new(mlua_swarm::TaskLaunchService::new(engine.clone(), compiler));
        AppState {
            engine,
            sessions: Arc::new(Mutex::new(crate::SessionStore::default())),
            task_app: Arc::new(mlua_swarm::TaskApplication::new_inline_only(launch)),
            ws_operator_factory: None,
            data_store: Arc::new(InMemoryOutputStore::new()),
            operator_sessions: Arc::new(Mutex::new(HashMap::new())),
            roles_to_sid: Arc::new(Mutex::new(HashMap::new())),
            task_store: Arc::new(InMemoryTaskStore::new()),
            run_store: Arc::new(InMemoryRunStore::new()),
            replay_store: Arc::new(mlua_swarm::store::replay::InMemoryReplayStore::new()),
            run_trace_store: Arc::new(mlua_swarm::store::trace::InMemoryRunTraceStore::new()),
            base_url: None,
            sync_timeout_secs: 300,
        }
    }

    fn post_tasks_req(goal: &str) -> crate::TaskLaunchRequest {
        crate::TaskLaunchRequest {
            blueprint: BlueprintRef::Inline {
                value: Box::new(identity_blueprint()),
            },
            init_ctx: serde_json::json!({"in": "hello"}),
            project_root: None,
            work_dir: None,
            task_metadata: None,
            ttl_secs: None,
            operator: None,
            operator_sid: None,
            timeout_secs: None,
            goal: Some(goal.to_string()),
            detach: false,
            check_policy: None,
        }
    }

    #[test]
    fn task_id_serializes_as_bare_string() {
        // Sanity check for the newtype-struct transparency relied on
        // throughout this module's response shapes (`TaskId` / `RunId`
        // serialize as plain JSON strings, not `{"0": "..."}`).
        let v = serde_json::to_value(TaskId::parse("T-abc").unwrap()).expect("serialize");
        assert_eq!(v, serde_json::json!("T-abc"));
    }

    #[tokio::test]
    async fn post_then_get_drill_down() {
        let state = test_state();

        let posted = crate::tasks_start(State(state.clone()), Json(post_tasks_req("smoke goal")))
            .await
            .expect("tasks_start")
            .0;
        let task_id = posted.task_id.clone();
        let run_id = posted.run_id.clone();

        // GET /v1/tasks lists it.
        let list = tasks_list(State(state.clone()), Query(TasksListQuery { limit: None }))
            .await
            .expect("tasks_list")
            .0;
        assert!(
            list.iter().any(|t| t.id == task_id),
            "task {task_id} missing from list of {} tasks",
            list.len()
        );

        // GET /v1/tasks/:id drills down to the Task + its Run.
        let detail = task_get(State(state.clone()), Path(task_id.to_string()))
            .await
            .expect("task_get")
            .0;
        assert_eq!(detail.task.id, task_id);
        assert_eq!(detail.task.goal, "smoke goal");
        assert_eq!(detail.task.status, TaskRecordStatus::Done);
        assert_eq!(detail.runs.len(), 1);
        assert_eq!(detail.runs[0].id, run_id);
        assert_eq!(detail.runs[0].status, RunStatus::Done);

        // GET /v1/runs/:id returns the same Run directly.
        let run = run_get(State(state.clone()), Path(run_id.to_string()))
            .await
            .expect("run_get")
            .0;
        assert_eq!(run.id, run_id);
        assert_eq!(run.task_id, task_id);
        assert_eq!(run.result_ref, Some(posted.final_ctx));

        // issue #13 run_id propagation: `POST /v1/tasks` (`run_flow_form`)
        // wires a `RunContext` into `TaskApplication::handle_with_run`, so
        // the single dispatched step must be traced into `step_entries`.
        assert_eq!(
            run.step_entries.len(),
            1,
            "expected one step_entry for the 1-step identity Blueprint, got {:?}",
            run.step_entries
        );
        assert_eq!(
            run.step_entries[0].step_ref,
            Some(mlua_swarm::worker::baseline::AG_IDENTITY.to_string())
        );
        assert_eq!(run.step_entries[0].status, Some("passed".to_string()));
    }

    // ──────────────────────────────────────────────────────────────────
    // GH #33 — sync-hang guards (readiness precheck / timeout ceiling)
    // ──────────────────────────────────────────────────────────────────

    /// Same 1-step identity flow as [`identity_blueprint`], but opts into
    /// the Blueprint-global Operator delegate axis
    /// (`spawner_hints.layers = ["operator_delegate"]`) so a registered
    /// `Operator` backend can be exercised end-to-end through the real
    /// `tasks_start` dispatch path (`OperatorDelegateMiddleware` bypasses
    /// `inner.spawn` and calls `operator.execute` instead — see
    /// `mlua_swarm::middleware::OperatorDelegateMiddleware` doc).
    fn identity_blueprint_with_operator_delegate() -> Blueprint {
        Blueprint {
            spawner_hints: mlua_swarm::SpawnerHints {
                layers: vec!["operator_delegate".to_string()],
            },
            ..identity_blueprint()
        }
    }

    /// `Operator` stub whose `execute` never resolves — the GH #33 Guard 2
    /// fixture ("a registered-but-never-acking operator").
    struct StallingOperator;

    #[async_trait::async_trait]
    impl mlua_swarm::Operator for StallingOperator {
        async fn execute(
            &self,
            _ctx: &mlua_swarm::Ctx,
            _system: Option<String>,
            _prompt: Value,
            _worker: Option<mlua_swarm::WorkerBinding>,
            _worker_token: mlua_swarm::CapToken,
        ) -> Result<mlua_swarm::WorkerResult, mlua_swarm::WorkerError> {
            std::future::pending::<()>().await;
            unreachable!("StallingOperator.execute must never resolve")
        }
    }

    /// A launch request that references an operator backend by id (via
    /// `operator.operator_backend_id`, the coarse Guard 1 signal) against
    /// [`identity_blueprint_with_operator_delegate`].
    fn operator_launch_req(
        backend_id: &str,
        timeout_secs: Option<u64>,
    ) -> crate::TaskLaunchRequest {
        crate::TaskLaunchRequest {
            blueprint: BlueprintRef::Inline {
                value: Box::new(identity_blueprint_with_operator_delegate()),
            },
            init_ctx: serde_json::json!({"in": "hello"}),
            project_root: None,
            work_dir: None,
            task_metadata: None,
            ttl_secs: None,
            operator: Some(crate::OperatorReq {
                operator_backend_id: Some(backend_id.to_string()),
                ..Default::default()
            }),
            operator_sid: None,
            timeout_secs,
            goal: Some("operator delegate test goal".to_string()),
            detach: false,
            check_policy: None,
        }
    }

    /// Guard 1: an operator-requiring launch with zero attached operators
    /// must fail immediately with a structured `503`, not hang waiting on
    /// a session nothing can serve.
    #[tokio::test]
    async fn sync_launch_zero_operators_fails_fast() {
        let state = test_state();
        // No `state.engine.register_operator(...)` call — zero operators
        // attached, matching `list_operator_ids()` being empty.
        let req = operator_launch_req("nonexistent-op", None);

        let started = std::time::Instant::now();
        let result = crate::tasks_start(State(state), Json(req)).await;
        let elapsed = started.elapsed();

        let err = match result {
            Err(e) => e,
            Ok(_) => panic!("zero attached operators must fail the operator-delegate launch"),
        };
        assert_eq!(err.status, StatusCode::SERVICE_UNAVAILABLE);
        assert!(
            err.message.contains("no operator attached"),
            "error message must mention the missing operator: {}",
            err.message
        );
        assert!(
            elapsed < Duration::from_secs(1),
            "guard 1 must fail fast (no dispatch, no timeout wait): took {elapsed:?}"
        );
    }

    /// Guard 2: a launch that resolves to a registered-but-stalled
    /// operator session must return a structured `504` within the
    /// requested `timeout_secs` ceiling, not hang the request forever.
    #[tokio::test]
    async fn sync_launch_stalled_times_out() {
        let state = test_state();
        state
            .engine
            .register_operator("stall-op", Arc::new(StallingOperator))
            .await;
        let req = operator_launch_req("stall-op", Some(1));

        let started = std::time::Instant::now();
        // Outer safety-net timeout: if guard 2 itself regressed into an
        // infinite hang, fail this test loudly instead of stalling `cargo
        // test` indefinitely.
        let result = tokio::time::timeout(
            Duration::from_secs(5),
            crate::tasks_start(State(state), Json(req)),
        )
        .await
        .expect("tasks_start must resolve well within 5s when guard 2's ceiling is 1s");
        let elapsed = started.elapsed();

        let err = match result {
            Err(e) => e,
            Ok(_) => panic!("a stalled operator session must time out, not succeed"),
        };
        assert_eq!(err.status, StatusCode::GATEWAY_TIMEOUT);
        assert!(
            err.message.contains('1'),
            "error message must mention the configured 1s ceiling: {}",
            err.message
        );
        assert!(
            elapsed < Duration::from_secs(3),
            "guard 2 must fire close to the requested 1s ceiling: took {elapsed:?}"
        );
    }

    /// Invariant 2: a launch that never references an operator backend
    /// must never be rejected by guard 1 — the simplest existing passing
    /// fixture (`post_tasks_req`) still succeeds unaffected.
    #[tokio::test]
    async fn sync_launch_without_operator_path_unaffected() {
        let state = test_state();
        let result = crate::tasks_start(
            State(state),
            Json(post_tasks_req("non-operator launch goal")),
        )
        .await;
        if let Err(e) = &result {
            panic!(
                "non-operator launch must succeed unaffected by guard 1: {}",
                e.message
            );
        }
    }

    /// Guard 2 ceiling resolution: `timeout_secs: Some(0)` is invalid
    /// (design doc: "0 = reject with 400 or treat as invalid — pick one
    /// and test it") — rejected fast, before any Task/Run side effects.
    #[tokio::test]
    async fn sync_launch_zero_timeout_secs_rejected() {
        let state = test_state();
        let mut req = post_tasks_req("zero timeout goal");
        req.timeout_secs = Some(0);

        let result = crate::tasks_start(State(state), Json(req)).await;
        let err = match result {
            Err(e) => e,
            Ok(_) => panic!("timeout_secs: Some(0) must be rejected, not treated as a no-op"),
        };
        assert_eq!(err.status, StatusCode::BAD_REQUEST);
        assert!(
            err.message.contains("timeout_secs"),
            "error message must reference timeout_secs: {}",
            err.message
        );
    }

    // ──────────────────────────────────────────────────────────────────
    // GH #37 — detached launch / rekick (driver decoupled from request)
    // ──────────────────────────────────────────────────────────────────

    /// Polls the run store until the given Run reaches a terminal status,
    /// panicking after ~5s — the detached paths complete in the
    /// background, so tests must wait on the store rather than the
    /// response.
    async fn wait_for_terminal_run(state: &AppState, run_id: &RunId) -> RunRecord {
        for _ in 0..50 {
            let rec = state.run_store.get(run_id).await.expect("run get");
            if !matches!(rec.status, RunStatus::Pending | RunStatus::Running) {
                return rec;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        panic!("run {run_id} did not reach a terminal status within ~5s");
    }

    /// GH #37: `detach: true` returns `202 Accepted` immediately with
    /// `status: "running"` and a null `final_ctx`; the eval completes in
    /// the background and the Run/Task reach `Done` with the result and
    /// step trace persisted — the same terminal state the sync path
    /// produces.
    #[tokio::test]
    async fn detached_launch_returns_202_and_completes_in_background() {
        let state = test_state();
        let mut req = post_tasks_req("detached goal");
        req.detach = true;

        let reply = crate::tasks_start(State(state.clone()), Json(req))
            .await
            .expect("tasks_start (detached)");
        assert_eq!(reply.1, StatusCode::ACCEPTED);
        let posted = reply.0;
        assert_eq!(posted.status, RunStatus::Running);
        assert_eq!(
            posted.final_ctx,
            serde_json::Value::Null,
            "a detached launch has no final_ctx at response time"
        );

        let rec = wait_for_terminal_run(&state, &posted.run_id).await;
        assert_eq!(rec.status, RunStatus::Done);
        assert!(
            rec.result_ref.is_some(),
            "finalize_run must persist the background eval's final_ctx"
        );
        assert_eq!(
            rec.step_entries.len(),
            1,
            "the background eval must trace its step_entries like the sync path: {:?}",
            rec.step_entries
        );
        let task = state
            .task_store
            .get(&posted.task_id)
            .await
            .expect("task get");
        assert_eq!(task.status, TaskRecordStatus::Done);
    }

    /// GH #37: `detach: true` + `timeout_secs` is contradictory (the sync
    /// ceiling has no meaning for a detached run) — rejected with `400`
    /// before any Task/Run side effects.
    #[tokio::test]
    async fn detached_launch_with_timeout_secs_rejected() {
        let state = test_state();
        let mut req = post_tasks_req("detached + ceiling goal");
        req.detach = true;
        req.timeout_secs = Some(60);

        let err = match crate::tasks_start(State(state.clone()), Json(req)).await {
            Err(e) => e,
            Ok(_) => panic!("detach + timeout_secs must be rejected"),
        };
        assert_eq!(err.status, StatusCode::BAD_REQUEST);
        assert!(
            err.message.contains("detach"),
            "error message must explain the detach/timeout_secs conflict: {}",
            err.message
        );
        let tasks = state.task_store.list().await.expect("task list");
        assert!(
            tasks.is_empty(),
            "the 400 must fire before any TaskRecord is minted"
        );
    }

    /// GH #37: a detached rekick returns `202 Accepted` with `status:
    /// "running"` immediately and completes in the background, adding a
    /// second `Done` Run to the same Task.
    #[tokio::test]
    async fn rekick_detached_returns_202_and_completes_in_background() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("detached rekick goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let (status, rekicked) = task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: None,
                task_input_override: None,
                timeout_secs: None,
                detach: true,
                operator_sid: None,
            })),
        )
        .await
        .expect("task_rekick (detached)");
        assert_eq!(status, StatusCode::ACCEPTED);
        assert_eq!(rekicked.0.status, RunStatus::Running);
        assert_ne!(rekicked.0.run_id, posted.run_id);

        let rec = wait_for_terminal_run(&state, &rekicked.0.run_id).await;
        assert_eq!(rec.status, RunStatus::Done);
        assert!(
            rec.result_ref.is_some(),
            "finalize_run must persist the background rekick's final_ctx"
        );
    }

    /// GH #37: `detach: true` + `timeout_secs` on the rekick path is the
    /// same contradiction as on the launch path — `400`, no new Run
    /// minted.
    #[tokio::test]
    async fn rekick_detached_with_timeout_secs_rejected() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("detached rekick ceiling goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let err = match task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: None,
                task_input_override: None,
                timeout_secs: Some(60),
                detach: true,
                operator_sid: None,
            })),
        )
        .await
        {
            Err(e) => e,
            Ok(_) => panic!("detach + timeout_secs must be rejected on rekick"),
        };
        assert_eq!(err.status, StatusCode::BAD_REQUEST);
        assert!(
            err.message.contains("detach"),
            "error message must explain the detach/timeout_secs conflict: {}",
            err.message
        );
        let runs = state
            .run_store
            .list_by_task(&posted.task_id)
            .await
            .expect("runs list");
        assert_eq!(
            runs.len(),
            1,
            "the 400 must fire before a second Run is minted"
        );
    }

    #[tokio::test]
    async fn rekick_adds_a_second_run_to_the_same_task() {
        let state = test_state();
        let posted = crate::tasks_start(State(state.clone()), Json(post_tasks_req("rekick goal")))
            .await
            .expect("tasks_start")
            .0;
        let task_id = posted.task_id.clone();
        let first_run_id = posted.run_id.clone();

        let (status, rekicked) = task_rekick(State(state.clone()), Path(task_id.to_string()), None)
            .await
            .expect("task_rekick");
        assert_eq!(status, StatusCode::CREATED);
        let second_run_id = rekicked.0.run_id.clone();
        assert_ne!(first_run_id, second_run_id);

        let detail = task_get(State(state.clone()), Path(task_id.to_string()))
            .await
            .expect("task_get")
            .0;
        assert_eq!(
            detail.runs.len(),
            2,
            "expected 2 runs, got {:?}",
            detail.runs
        );
        let ids: Vec<&RunId> = detail.runs.iter().map(|r| &r.id).collect();
        assert!(ids.contains(&&first_run_id));
        assert!(ids.contains(&&second_run_id));

        // issue #13 run_id propagation: each kick's own `EngineDispatcher`
        // (built fresh per `TaskApplication::handle_with_run` call) must
        // trace its own dispatched step into its own `RunRecord` —
        // independent `step_entries`, not shared/accumulated across kicks.
        let first_run = detail
            .runs
            .iter()
            .find(|r| r.id == first_run_id)
            .expect("first run present in detail.runs");
        let second_run = detail
            .runs
            .iter()
            .find(|r| r.id == second_run_id)
            .expect("second run present in detail.runs");
        assert_eq!(
            first_run.step_entries.len(),
            1,
            "first run step_entries: {:?}",
            first_run.step_entries
        );
        assert_eq!(
            second_run.step_entries.len(),
            1,
            "second run step_entries: {:?}",
            second_run.step_entries
        );
        assert_eq!(
            first_run.step_entries[0].step_ref,
            Some(mlua_swarm::worker::baseline::AG_IDENTITY.to_string())
        );
        assert_eq!(
            second_run.step_entries[0].step_ref,
            Some(mlua_swarm::worker::baseline::AG_IDENTITY.to_string())
        );
        assert_eq!(first_run.step_entries[0].status, Some("passed".to_string()));
        assert_eq!(
            second_run.step_entries[0].status,
            Some("passed".to_string())
        );
        assert_ne!(
            first_run.step_entries[0].step_id, second_run.step_entries[0].step_id,
            "each kick dispatches its own StepId — runs must not share step_entries"
        );
    }

    #[tokio::test]
    async fn rekick_unknown_task_returns_404() {
        let state = test_state();
        // `.expect_err()` needs the Ok variant to be `Debug`; `Json<T>`'s
        // `Debug` impl is not guaranteed for every `T` across axum versions,
        // so a plain match sidesteps that bound entirely.
        match task_rekick(State(state), Path("T-does-not-exist".to_string()), None).await {
            Ok(_) => panic!("expected 404 for an unknown task"),
            Err(e) => assert_eq!(e.status, StatusCode::NOT_FOUND),
        }
    }

    // ──────────────────────────────────────────────────────────────────
    // issue #19 ST4: `RunKickRequest` (optional body / 3-layer merge)
    // ──────────────────────────────────────────────────────────────────

    /// A single-step flow.ir Blueprint that echoes `$.greeting` into
    /// `$.out` — unlike [`identity_blueprint`] (a fixed `lit("hello")`
    /// input), this one reads its `Step.in` from `ctx`, so it observes
    /// whichever `init_ctx` layer actually won the merge.
    fn greeting_blueprint() -> Blueprint {
        Blueprint {
            schema_version: current_schema_version(),
            id: "tasks-test-greeting-bp".into(),
            flow: serde_json::from_value(serde_json::json!({
                "kind": "step",
                "ref": mlua_swarm::worker::baseline::AG_IDENTITY,
                "in": {"op": "path", "at": "$.greeting"},
                "out": {"op": "path", "at": "$.out"},
            }))
            .expect("flow parse"),
            agents: vec![AgentDef {
                name: mlua_swarm::worker::baseline::AG_IDENTITY.into(),
                kind: AgentKind::RustFn,
                spec: serde_json::json!({"fn_id": mlua_swarm::worker::baseline::AG_IDENTITY}),
                profile: None,
                meta: None,
                runner: None,
                runner_ref: None,
                verdict: 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(),
        }
    }

    fn post_greeting_task_req(
        greeting: &str,
        project_root: Option<&str>,
    ) -> crate::TaskLaunchRequest {
        crate::TaskLaunchRequest {
            blueprint: BlueprintRef::Inline {
                value: Box::new(greeting_blueprint()),
            },
            init_ctx: serde_json::json!({ "greeting": greeting }),
            project_root: project_root.map(str::to_string),
            work_dir: None,
            task_metadata: None,
            ttl_secs: None,
            operator: None,
            operator_sid: None,
            timeout_secs: None,
            goal: Some("st4 rekick goal".to_string()),
            detach: false,
            check_policy: None,
        }
    }

    #[tokio::test]
    async fn rekick_no_body_preserves_stored_task_input_ctx_byte_for_byte() {
        // must_not_simplify #3: a body-less rekick must behave exactly
        // like pre-#19 — the Task's own `input_ctx` alone seeds the kick.
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_greeting_task_req("from-task", None)),
        )
        .await
        .expect("tasks_start")
        .0;
        assert_eq!(posted.final_ctx["out"]["echoed"], "from-task");

        let (status, rekicked) =
            task_rekick(State(state.clone()), Path(posted.task_id.to_string()), None)
                .await
                .expect("task_rekick");
        assert_eq!(status, StatusCode::CREATED);

        let run = run_get(State(state.clone()), Path(rekicked.0.run_id.to_string()))
            .await
            .expect("run_get")
            .0;
        assert_eq!(
            run.result_ref.expect("result_ref present")["out"]["echoed"],
            "from-task"
        );
    }

    #[tokio::test]
    async fn rekick_with_init_ctx_override_wins_over_stored_task_input_ctx() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_greeting_task_req("from-task", None)),
        )
        .await
        .expect("tasks_start")
        .0;
        assert_eq!(posted.final_ctx["out"]["echoed"], "from-task");

        let (status, rekicked) = task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: Some(serde_json::json!({ "greeting": "from-run" })),
                task_input_override: None,
                timeout_secs: None,
                detach: false,
                operator_sid: None,
            })),
        )
        .await
        .expect("task_rekick");
        assert_eq!(status, StatusCode::CREATED);

        let run = run_get(State(state.clone()), Path(rekicked.0.run_id.to_string()))
            .await
            .expect("run_get")
            .0;
        assert_eq!(
            run.result_ref.expect("result_ref present")["out"]["echoed"],
            "from-run",
            "Run's init_ctx_override must win over the stored Task input_ctx"
        );
    }

    #[tokio::test]
    async fn rekick_with_stored_task_input_spec_dispatches_and_leaves_it_unmutated() {
        // Done Criteria: "Task record が task-level canonical fields を
        // 保持している時の rekick test". A Task created with
        // `project_root` set gets a `task_input_spec` snapshot; a
        // body-less rekick must both dispatch successfully (the stored
        // spec decodes and resolves without erroring) and leave
        // `TaskRecord.task_input_spec` untouched (must_not_simplify #4 —
        // a rekick never mutates the stored Task-level snapshot).
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_greeting_task_req("from-task", Some("/repo"))),
        )
        .await
        .expect("tasks_start")
        .0;

        let before = state
            .task_store
            .get(&posted.task_id)
            .await
            .expect("task fetch");
        let before_spec: Option<TaskInputSpec> = before
            .task_input_spec
            .as_ref()
            .map(|v| serde_json::from_value(v.clone()).expect("decode task_input_spec"));
        assert_eq!(
            before_spec,
            Some(TaskInputSpec {
                project_root: Some("/repo".to_string()),
                work_dir: None,
                task_metadata: None,
            })
        );

        let (status, _rekicked) =
            task_rekick(State(state.clone()), Path(posted.task_id.to_string()), None)
                .await
                .expect("task_rekick");
        assert_eq!(status, StatusCode::CREATED);

        let after = state
            .task_store
            .get(&posted.task_id)
            .await
            .expect("task fetch");
        assert_eq!(
            after.task_input_spec, before.task_input_spec,
            "rekick must not mutate the stored Task-level task_input_spec snapshot"
        );
    }

    #[tokio::test]
    async fn rekick_with_task_input_override_does_not_mutate_stored_task_record() {
        // must_not_simplify #4: `task_input_override` wins for this kick
        // only — the stored `TaskRecord.task_input_spec` is untouched.
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_greeting_task_req("from-task", Some("/repo"))),
        )
        .await
        .expect("tasks_start")
        .0;

        let (status, _rekicked) = task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: None,
                task_input_override: Some(TaskInputSpec {
                    project_root: Some("/override".to_string()),
                    work_dir: None,
                    task_metadata: None,
                }),
                timeout_secs: None,
                detach: false,
                operator_sid: None,
            })),
        )
        .await
        .expect("task_rekick");
        assert_eq!(status, StatusCode::CREATED);

        let after = state
            .task_store
            .get(&posted.task_id)
            .await
            .expect("task fetch");
        let after_spec: Option<TaskInputSpec> = after
            .task_input_spec
            .as_ref()
            .map(|v| serde_json::from_value(v.clone()).expect("decode task_input_spec"));
        assert_eq!(
            after_spec,
            Some(TaskInputSpec {
                project_root: Some("/repo".to_string()),
                work_dir: None,
                task_metadata: None,
            }),
            "a per-Run task_input_override must not leak into the stored TaskRecord"
        );
    }

    // ──────────────────────────────────────────────────────────────────
    // GH #33 → task_rekick — sync-hang guards (issue #35 ST3 parity)
    // ──────────────────────────────────────────────────────────────────

    /// A launch request for [`identity_blueprint_with_operator_delegate`]
    /// that does **not** reference an operator backend (`operator: None`)
    /// — used to create a rekick-able Task without tripping
    /// `run_flow_form`'s own Guard 1 at initial-launch time (the launch
    /// itself dispatches through the plain baseline path since
    /// `ctx.operator.operator` stays unset either way; the BP's
    /// `operator_delegate` layer only matters to `task_rekick`'s Guard 1,
    /// which reads `resolved_bp.spawner_hints.layers` directly rather than
    /// a per-request field).
    fn delegate_launch_req(goal: &str) -> crate::TaskLaunchRequest {
        crate::TaskLaunchRequest {
            blueprint: BlueprintRef::Inline {
                value: Box::new(identity_blueprint_with_operator_delegate()),
            },
            init_ctx: serde_json::json!({"in": "hello"}),
            project_root: None,
            work_dir: None,
            task_metadata: None,
            ttl_secs: None,
            operator: None,
            operator_sid: None,
            timeout_secs: None,
            goal: Some(goal.to_string()),
            detach: false,
            check_policy: None,
        }
    }

    /// Guard 1 (adapted signal): a Task whose stored Blueprint declares
    /// the `operator_delegate` layer, rekicked with zero attached
    /// operators, must fail immediately with a structured `503` — not
    /// dispatch and not hang waiting on a session nothing can serve.
    #[tokio::test]
    async fn rekick_zero_operators_with_operator_delegate_blueprint_fails_fast() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(delegate_launch_req("operator delegate rekick goal")),
        )
        .await
        .expect("tasks_start (no operator referenced, dispatches through baseline)")
        .0;
        // No `state.engine.register_operator(...)` call — zero operators
        // attached, matching `list_operator_ids()` being empty.

        let started = std::time::Instant::now();
        let result = task_rekick(State(state), Path(posted.task_id.to_string()), None).await;
        let elapsed = started.elapsed();

        let err = match result {
            Err(e) => e,
            Ok(_) => panic!(
                "rekicking a Task whose Blueprint declares operator_delegate with zero \
                 attached operators must fail, not dispatch"
            ),
        };
        assert_eq!(err.status, StatusCode::SERVICE_UNAVAILABLE);
        assert!(
            err.message.contains("no operator attached"),
            "error message must mention the missing operator: {}",
            err.message
        );
        assert!(
            elapsed < Duration::from_secs(1),
            "guard 1 must fail fast (no dispatch, no timeout wait): took {elapsed:?}"
        );
    }

    /// Guard 2: a rekick with a `timeout_secs` ceiling shorter than the
    /// dispatch takes must return a structured `504` within the outer
    /// safety-net timeout, not hang the request forever.
    #[tokio::test]
    async fn rekick_stalled_operator_times_out() {
        let state = test_state();
        state
            .engine
            .register_operator("stall-op", Arc::new(StallingOperator))
            .await;
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(delegate_launch_req("stalled rekick goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let started = std::time::Instant::now();
        // Outer safety-net timeout: if guard 2 itself regressed into an
        // infinite hang, fail this test loudly instead of stalling `cargo
        // test` indefinitely.
        let result = tokio::time::timeout(
            Duration::from_secs(5),
            task_rekick(
                State(state),
                Path(posted.task_id.to_string()),
                Some(Json(RunKickRequest {
                    init_ctx_override: None,
                    task_input_override: None,
                    timeout_secs: Some(1),
                    detach: false,
                    operator_sid: None,
                })),
            ),
        )
        .await
        .expect("task_rekick must resolve well within 5s when guard 2's ceiling is 1s");
        let elapsed = started.elapsed();

        match &result {
            Err(e) => {
                assert_eq!(e.status, StatusCode::GATEWAY_TIMEOUT);
                assert!(
                    e.message.contains('1'),
                    "error message must mention the configured 1s ceiling: {}",
                    e.message
                );
                assert!(
                    elapsed < Duration::from_secs(3),
                    "guard 2 must fire close to the requested 1s ceiling: took {elapsed:?}"
                );
            }
            Ok(_) => {
                // This rekick passes no `operator_sid`, so its
                // `operator_backend_id` stays `None` and the
                // registered-but-unattached `StallingOperator` is never
                // actually engaged by the dispatch; the flow resolves
                // through the plain baseline path instead. Guard 2's
                // `tokio::time::timeout`
                // wrap is exercised (and does not falsely fire) rather
                // than tripped — assert the fast-success shape so a
                // regression that makes rekick dispatch slow (or that
                // makes Guard 2 falsely trip on a fast dispatch) is still
                // caught by the elapsed-time assertion below.
                assert!(
                    elapsed < Duration::from_secs(1),
                    "a rekick that never engages an Operator (task_rekick has no \
                     per-request operator override) must resolve fast, not stall: took {elapsed:?}"
                );
            }
        }
    }

    /// Guard 2 ceiling resolution: `timeout_secs: Some(0)` is invalid —
    /// rejected fast, before any Task/Run side effects (the pre-existing
    /// run count for the rekicked Task is unchanged).
    #[tokio::test]
    async fn rekick_timeout_secs_zero_rejected() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("zero timeout rekick goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let before = task_get(State(state.clone()), Path(posted.task_id.to_string()))
            .await
            .expect("task_get")
            .0;
        let runs_before = before.runs.len();

        let result = task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: None,
                task_input_override: None,
                timeout_secs: Some(0),
                detach: false,
                operator_sid: None,
            })),
        )
        .await;
        let err = match result {
            Err(e) => e,
            Ok(_) => panic!("timeout_secs: Some(0) must be rejected, not treated as a no-op"),
        };
        assert_eq!(err.status, StatusCode::BAD_REQUEST);
        assert!(
            err.message.contains("timeout_secs"),
            "error message must reference timeout_secs: {}",
            err.message
        );

        let after = task_get(State(state), Path(posted.task_id.to_string()))
            .await
            .expect("task_get")
            .0;
        assert_eq!(
            after.runs.len(),
            runs_before,
            "a rejected timeout_secs: Some(0) rekick must not create a new Run"
        );
    }

    /// Invariant: a plain (non-`operator_delegate`) Task rekick must
    /// never be rejected by Guard 1 — the simplest existing passing
    /// rekick fixture still succeeds unaffected.
    #[tokio::test]
    async fn rekick_non_operator_path_unaffected_by_guard_1() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("non-operator rekick goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let result = task_rekick(State(state), Path(posted.task_id.to_string()), None).await;
        if let Err(e) = &result {
            panic!(
                "a plain (non-operator_delegate) Task rekick must succeed unaffected by \
                 guard 1: {}",
                e.message
            );
        }
    }

    // ──────────────────────────────────────────────────────────────────
    // B-1: rekick operator_sid pin (parity with POST /v1/tasks)
    // ──────────────────────────────────────────────────────────────────

    /// A rekick pinning an unknown `operator_sid` fails fast with a `400`
    /// before any Task/Run store write — no new Run is minted (S2 parity
    /// with `run_flow_form`'s `operator_sid` fail-fast).
    #[tokio::test]
    async fn rekick_unknown_operator_sid_rejected_before_side_effects() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("unknown operator_sid rekick goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let before = task_get(State(state.clone()), Path(posted.task_id.to_string()))
            .await
            .expect("task_get")
            .0;
        let runs_before = before.runs.len();

        let result = task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: None,
                task_input_override: None,
                timeout_secs: None,
                detach: false,
                operator_sid: Some("S-not-registered".to_string()),
            })),
        )
        .await;
        let err = match result {
            Err(e) => e,
            Ok(_) => panic!("an unknown operator_sid must be rejected, not dispatched"),
        };
        assert_eq!(err.status, StatusCode::BAD_REQUEST);
        assert!(
            err.message.contains("operator_sid"),
            "error message must reference operator_sid: {}",
            err.message
        );

        let after = task_get(State(state), Path(posted.task_id.to_string()))
            .await
            .expect("task_get")
            .0;
        assert_eq!(
            after.runs.len(),
            runs_before,
            "a rejected unknown-operator_sid rekick must not create a new Run"
        );
    }

    /// A rekick pinning a *registered* `operator_sid` dispatches
    /// successfully and persists the sid verbatim onto the new
    /// `RunRecord.operator_sid`. The Task's stored Blueprint is the plain
    /// baseline (no `operator_delegate` layer), so the registered Operator
    /// is never actually engaged — the kick resolves through the baseline
    /// path and the assertion is purely on the persisted correlation
    /// field.
    #[tokio::test]
    async fn rekick_with_registered_operator_sid_persists_it_on_the_run() {
        let state = test_state();
        // Register an Operator whose sid the rekick can pin. It is never
        // engaged (plain Blueprint does not delegate), so a `StallingOperator`
        // is a fine stand-in for "a live, registered session".
        state
            .engine
            .register_operator("S-live-op", Arc::new(StallingOperator))
            .await;
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("registered operator_sid rekick goal")),
        )
        .await
        .expect("tasks_start")
        .0;

        let (status, rekicked) = task_rekick(
            State(state.clone()),
            Path(posted.task_id.to_string()),
            Some(Json(RunKickRequest {
                init_ctx_override: None,
                task_input_override: None,
                timeout_secs: None,
                detach: false,
                operator_sid: Some("S-live-op".to_string()),
            })),
        )
        .await
        .expect("task_rekick with a registered operator_sid");
        assert_eq!(status, StatusCode::CREATED);

        let run = state
            .run_store
            .get(&rekicked.0.run_id)
            .await
            .expect("run get");
        assert_eq!(
            run.operator_sid,
            Some("S-live-op".to_string()),
            "the pinned operator_sid must be persisted verbatim on the RunRecord"
        );
    }

    #[tokio::test]
    async fn run_get_unknown_id_returns_404() {
        let state = test_state();
        match run_get(State(state), Path("R-does-not-exist".to_string())).await {
            Ok(_) => panic!("expected 404 for an unknown run"),
            Err(e) => assert_eq!(e.status, StatusCode::NOT_FOUND),
        }
    }

    #[tokio::test]
    async fn run_bindings_explain_reports_pinned_requested_effective_diff() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("binding explain")),
        )
        .await
        .expect("tasks_start")
        .0;
        let run = state
            .run_store
            .get(&posted.run_id)
            .await
            .expect("stored run");
        let mut snapshot: Value = serde_json::from_str(run.input_json.as_deref().unwrap()).unwrap();
        let mut bound_agents: Vec<BoundAgent> =
            serde_json::from_value(snapshot["bound_agents"].clone()).unwrap();
        let bound = &mut bound_agents[0];
        bound.runner = Some(Runner::WsClaudeCode {
            variant: "coder".to_string(),
            tools: vec!["Read".to_string()],
        });
        bound.recompute_binding_digest().unwrap();
        let request_digest = bound.binding_digest.clone();
        bound
            .set_attestation(BindingAttestation {
                request_digest: request_digest.clone(),
                provider_id: "operator-manifest".to_string(),
                provider_revision: Some("claude-code-1.2".to_string()),
                resolved_model: Some("claude-sonnet-4".to_string()),
                effective_tools: vec!["Bash".to_string(), "Read".to_string()],
                launch_variant: Some("coder".to_string()),
                capability_snapshot_digest: Some(mlua_swarm::blueprint::BindingDigest::sha256(
                    b"manifest-v1",
                )),
            })
            .unwrap();
        snapshot["bound_agents"] = serde_json::to_value(&bound_agents).unwrap();
        state
            .run_store
            .set_input_json(&posted.run_id, serde_json::to_string(&snapshot).unwrap())
            .await
            .unwrap();

        let explained = run_bindings_explain(State(state), Path(posted.run_id.to_string()))
            .await
            .expect("binding explain")
            .0;
        let entry = &explained.bindings[0];
        assert_eq!(entry.status, RunBindingStatus::Attested);
        assert_eq!(
            entry.requested.as_ref().unwrap().request_digest,
            request_digest
        );
        assert_eq!(
            entry
                .effective
                .as_ref()
                .unwrap()
                .provider_revision
                .as_deref(),
            Some("claude-code-1.2")
        );
        assert_eq!(
            entry
                .difference
                .as_ref()
                .unwrap()
                .additional_effective_tools,
            vec!["Bash"]
        );
        assert!(entry
            .difference
            .as_ref()
            .unwrap()
            .missing_requested_tools
            .is_empty());
        assert_ne!(entry.binding_digest, request_digest);
    }

    #[tokio::test]
    async fn run_bindings_explain_reports_snapshot_origin() {
        let state = test_state();
        let posted =
            crate::tasks_start(State(state.clone()), Json(post_tasks_req("origin explain")))
                .await
                .expect("tasks_start")
                .0;

        // An initial launch pins `origin = launch`.
        let explained = run_bindings_explain(State(state.clone()), Path(posted.run_id.to_string()))
            .await
            .expect("binding explain")
            .0;
        assert_eq!(explained.snapshot_origin, SnapshotOrigin::Launch);

        // Flip the persisted marker to `resume_backfill` → explain reflects it.
        let run = state.run_store.get(&posted.run_id).await.unwrap();
        let mut snapshot: Value = serde_json::from_str(run.input_json.as_deref().unwrap()).unwrap();
        snapshot["bound_agents_origin"] = serde_json::json!("resume_backfill");
        state
            .run_store
            .set_input_json(&posted.run_id, serde_json::to_string(&snapshot).unwrap())
            .await
            .unwrap();
        let explained = run_bindings_explain(State(state.clone()), Path(posted.run_id.to_string()))
            .await
            .expect("binding explain")
            .0;
        assert_eq!(explained.snapshot_origin, SnapshotOrigin::ResumeBackfill);

        // A snapshot with `bound_agents` but NO origin marker maps to the
        // safe side (`resume_backfill`) and still returns 200 — the 422 is
        // reserved for snapshots lacking `bound_agents` entirely.
        snapshot
            .as_object_mut()
            .unwrap()
            .remove("bound_agents_origin");
        state
            .run_store
            .set_input_json(&posted.run_id, serde_json::to_string(&snapshot).unwrap())
            .await
            .unwrap();
        let explained = run_bindings_explain(State(state), Path(posted.run_id.to_string()))
            .await
            .expect("explain still 200 without an origin marker")
            .0;
        assert_eq!(explained.snapshot_origin, SnapshotOrigin::ResumeBackfill);
    }

    #[tokio::test]
    async fn run_bindings_explain_never_guesses_for_legacy_snapshot() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("legacy binding explain")),
        )
        .await
        .expect("tasks_start")
        .0;
        state
            .run_store
            .set_input_json(&posted.run_id, "{}".to_string())
            .await
            .unwrap();

        let error = run_bindings_explain(State(state), Path(posted.run_id.to_string()))
            .await
            .expect_err("legacy run must not be re-resolved");
        assert_eq!(error.status, StatusCode::UNPROCESSABLE_ENTITY);
        assert!(error
            .message
            .contains("current Blueprint state was not consulted"));
    }

    #[tokio::test]
    async fn run_bindings_explain_rejects_a_tampered_snapshot() {
        let state = test_state();
        let posted = crate::tasks_start(
            State(state.clone()),
            Json(post_tasks_req("tampered binding explain")),
        )
        .await
        .expect("tasks_start")
        .0;
        let run = state.run_store.get(&posted.run_id).await.unwrap();
        let mut snapshot: Value = serde_json::from_str(run.input_json.as_deref().unwrap()).unwrap();
        snapshot["bound_agents"][0]["agent"]["name"] = Value::String("tampered".into());
        state
            .run_store
            .set_input_json(&posted.run_id, serde_json::to_string(&snapshot).unwrap())
            .await
            .unwrap();

        let error = run_bindings_explain(State(state), Path(posted.run_id.to_string()))
            .await
            .expect_err("digest drift must fail closed");
        assert_eq!(error.status, StatusCode::UNPROCESSABLE_ENTITY);
        assert!(error.message.contains("inconsistent binding snapshot"));
    }

    #[tokio::test]
    async fn task_get_unknown_id_returns_404() {
        let state = test_state();
        match task_get(State(state), Path("T-does-not-exist".to_string())).await {
            Ok(_) => panic!("expected 404 for an unknown task"),
            Err(e) => assert_eq!(e.status, StatusCode::NOT_FOUND),
        }
    }

    // ──────────────────────────────────────────────────────────────────
    // GH #76 error surface: finalize_run Err arm populates result_ref with the
    // structured failure envelope; run_get surfaces it.
    // ──────────────────────────────────────────────────────────────────

    /// Seed a Task + Run row so `finalize_run` can update them.
    async fn seed_task_and_run(state: &AppState) -> (TaskId, RunId) {
        let task_id = TaskId::new();
        let run_id = RunId::new();
        state
            .task_store
            .create(TaskRecord {
                id: task_id.clone(),
                goal: "finalize-run-err-envelope".to_string(),
                blueprint_ref: json!("inline"),
                input_ctx: Value::Null,
                task_input_spec: None,
                status: TaskRecordStatus::Running,
                created_at: 0,
                updated_at: 0,
            })
            .await
            .expect("seed TaskRecord");
        state
            .run_store
            .create(RunRecord {
                id: run_id.clone(),
                task_id: task_id.clone(),
                status: RunStatus::Running,
                step_entries: Vec::new(),
                degradations: Vec::new(),
                operator_sid: None,
                result_ref: None,
                input_json: Some("{}".to_string()),
                created_at: 0,
                updated_at: 0,
            })
            .await
            .expect("seed RunRecord");
        (task_id, run_id)
    }

    #[tokio::test]
    async fn finalize_run_err_arm_populates_result_ref_with_structured_envelope() {
        let state = test_state();
        let (task_id, run_id) = seed_task_and_run(&state).await;

        let err: Result<TaskApplicationOutput, TaskApplicationError> =
            Err(TaskApplicationError::Launch(TaskLaunchError::FlowEval {
                message: "blocked: {\"verdict\":\"BLOCKED\"}".to_string(),
                failed_step: Some("gate".to_string()),
                verdict_value: Some(json!({"verdict": "BLOCKED", "reason": "not-applicable"})),
                partial_ctx: Some(
                    json!({"steps": {"ST-abc": {"step_ref": "gate", "status": "blocked"}}}),
                ),
            }));

        let _ = finalize_run(&state, &task_id, &run_id, err).await;

        let run = state.run_store.get(&run_id).await.expect("run present");
        assert_eq!(run.status, RunStatus::Failed);
        let envelope = run
            .result_ref
            .as_ref()
            .expect("result_ref must be Some on Err arm");
        assert_eq!(
            envelope["error"]["message"],
            "blocked: {\"verdict\":\"BLOCKED\"}"
        );
        assert_eq!(envelope["error"]["failed_step"], "gate");
        assert_eq!(envelope["error"]["verdict_value"]["verdict"], "BLOCKED");
        assert_eq!(
            envelope["partial_ctx"]["steps"]["ST-abc"]["status"],
            "blocked"
        );

        // Owning Task status also flipped.
        let task = state.task_store.get(&task_id).await.expect("task present");
        assert_eq!(task.status, TaskRecordStatus::Failed);
    }

    #[tokio::test]
    async fn finalize_run_err_arm_non_flow_eval_populates_envelope_with_null_structural_fields() {
        let state = test_state();
        let (task_id, run_id) = seed_task_and_run(&state).await;

        // A non-FlowEval error (e.g. NoStore) still lands the envelope
        // shape with `error.message` populated; the structural fields go
        // to JSON `null` (no breadcrumb source available).
        let err: Result<TaskApplicationOutput, TaskApplicationError> =
            Err(TaskApplicationError::NoStore);

        let _ = finalize_run(&state, &task_id, &run_id, err).await;
        let run = state.run_store.get(&run_id).await.expect("run present");
        let envelope = run
            .result_ref
            .as_ref()
            .expect("result_ref must be Some on Err arm");
        assert!(envelope["error"]["message"]
            .as_str()
            .expect("message string")
            .contains("store"));
        assert_eq!(envelope["error"]["failed_step"], Value::Null);
        assert_eq!(envelope["error"]["verdict_value"], Value::Null);
        assert_eq!(envelope["partial_ctx"], Value::Null);
    }

    /// Regression: the Ok arm still stores the raw `final_ctx` verbatim
    /// (NOT an envelope) — consumers that never saw a failure keep their
    /// pre-#76 shape. The disambiguation is the top-level `"error"` key:
    /// present iff the Err arm fired.
    #[tokio::test]
    async fn finalize_run_ok_arm_still_stores_raw_final_ctx_verbatim() {
        let state = test_state();
        let (task_id, run_id) = seed_task_and_run(&state).await;

        let ok: Result<TaskApplicationOutput, TaskApplicationError> = Ok(TaskApplicationOutput {
            token: mlua_swarm::CapToken {
                agent_id: "ut".to_string(),
                role: mlua_swarm::Role::Operator,
                scopes: vec!["*".to_string()],
                issued_at: 0,
                expire_at: u64::MAX,
                max_uses: None,
                nonce: "ut-nonce".to_string(),
                sig_hex: String::new(),
            },
            final_ctx: json!({"out": {"echoed": "hi"}}),
            bound_version: None,
        });

        let _ = finalize_run(&state, &task_id, &run_id, ok).await;
        let run = state.run_store.get(&run_id).await.expect("run present");
        assert_eq!(run.status, RunStatus::Done);
        let stored = run.result_ref.as_ref().expect("result_ref Some");
        // Raw final_ctx verbatim — NOT an envelope; no top-level "error" key.
        assert_eq!(stored, &json!({"out": {"echoed": "hi"}}));
        assert!(
            stored.get("error").is_none(),
            "Ok arm must never write an `error` key at the top of result_ref (envelope disambiguation)"
        );
    }

    /// `GET /v1/runs/:id` returns the `RunRecord` verbatim, so after a
    /// finalize_run Err arm the structured envelope surfaces through the
    /// existing handler — no new response type needed. Failure detection
    /// via the top-level `"error"` key inside `result_ref`.
    #[tokio::test]
    async fn run_get_surfaces_structured_failure_envelope_from_result_ref() {
        let state = test_state();
        let (_task_id, run_id) = seed_task_and_run(&state).await;
        let err: Result<TaskApplicationOutput, TaskApplicationError> =
            Err(TaskApplicationError::Launch(TaskLaunchError::FlowEval {
                message: "blocked: bad verdict".to_string(),
                failed_step: Some("scout".to_string()),
                verdict_value: Some(json!("BLOCKED")),
                partial_ctx: Some(json!({"steps": {}})),
            }));
        let _ = finalize_run(&state, &_task_id, &run_id, err).await;

        let Json(run) = run_get(State(state), Path(run_id.to_string()))
            .await
            .expect("run_get");
        assert_eq!(run.status, RunStatus::Failed);
        let envelope = run.result_ref.expect("result_ref Some");
        assert_eq!(envelope["error"]["failed_step"], "scout");
        assert_eq!(envelope["error"]["verdict_value"], "BLOCKED");
    }
}