areev-run 1.13.2

//! The driver loop (§5.3): start, resume, respond, cancel, verify. All
//! store writes happen HERE, on the driver thread, in the order the pure
//! core commands them — intents before dispatches, results before their
//! bookkeeping, checkpoints as commanded. Parallelism lives exclusively in
//! the executor pool.

use crate::clock::Clock;
use crate::executor::{DispatchDone, DispatchJob, HostToolExecutor, Pool};
use crate::journal::{self, JournalView};
use crate::manifest::{BudgetsSpec, RunManifest};
use crate::{err_run, RunSession, VerifyReport, VerifyStep};
use areev_cal::AreevFacade;
use areev_core::authz::HARNESS_NS;
use areev_core::error::Hash;
use areev_core::types::{Fact, Grain, Observation};
use areev_run_core::{
    flow_key, step, Ask, Command, DecisionRecord, EffectOutcome, EventIn, FailCause, JournalKey,
    NodeExecutor, PlanGraph, RunError, RunOutcome, SchedulerState, StepEnv, PARKED_ASKS,
};
use serde_json::{json, Value};
use sha2::Digest;
use std::collections::BTreeMap;
use std::sync::Arc;

/// What to do with a dangling intent found on resume (§5.3 step 4): the
/// pinned default re-dispatches under the SAME idempotency key and journals
/// the redelivery; hosts that cannot deduplicate set `Fail`. No third mode.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum OnDangling {
    #[default]
    Redispatch,
    Fail,
}

/// A deterministic crash-injection point — the §5.5 gate-2 mechanism
/// (deterministic injection, never sleeps): the driver aborts at a precise
/// spot in the intent→result protocol, leaving exactly the on-disk state a
/// real crash there would leave. Production callers never set this.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CrashPoint {
    /// Abort immediately after writing the nth intent (1-based), before its
    /// dispatch: the effect never fired; the intent dangles.
    AfterIntent(u32),
    /// Abort after the nth effect COMPLETED but before its result is
    /// journaled: the effect fired in the real world; the intent dangles —
    /// the exact window §5.3's redelivery contract exists for.
    BeforeResult(u32),
}

/// Start/resume options.
#[derive(Default)]
pub struct RunOptions {
    pub budgets: BudgetsSpec,
    pub ask_ttl_sec: Option<i64>,
    pub workers: usize,
    pub on_dangling: OnDangling,
    /// Per-LLM-call `max_tokens` for abstract nodes — frozen into the
    /// manifest at start (§6.7's per-dispatch reservation). None = 1024.
    pub llm_max_tokens: Option<u32>,
    /// Effects one node attempt may spend — an abstract node's turns, tool
    /// calls and re-prompts share the counter. Frozen into the manifest at
    /// start; None = [`areev_run_core::DEFAULT_MAX_EFFECTS_PER_ATTEMPT`].
    pub max_effects_per_attempt: Option<u32>,
    /// Bound on ONE tool result's size in an abstract node's transcript, in
    /// characters — the journal keeps every result in full. Frozen into the
    /// manifest at start. None = unbounded.
    pub llm_tool_result_chars: Option<usize>,
    /// Ceiling on an abstract node's whole transcript, in the provider's own
    /// reported prompt tokens — reaching it folds the middle of the transcript
    /// into one journaled summary. Frozen into the manifest at start.
    /// None = no ceiling.
    pub llm_context_tokens: Option<u64>,
    /// Crash-injection for the §5.5 gates. `None` in production.
    pub inject_crash: Option<CrashPoint>,
    /// Who or what this run is started ON BEHALF OF (#293). Frozen into the
    /// manifest; an approval ask refuses this responder as well as
    /// `principal`.
    pub initiator: Option<String>,
    /// Write content-bearing harness records to `agent:harness.<ns>` rather
    /// than the memory-wide `agent:harness` (#301). Frozen into the manifest.
    pub harness_ns: Option<String>,
    /// Whether this run may settle a `confirmation` ask from its own
    /// initiator (#294). Off by default: a Definition can arrive in a bundle
    /// or a pack, and a weakening delivered with the thing it weakens is not
    /// a permission — the same reasoning as `--allow-executor`.
    pub allow_confirmation_asks: bool,
    /// Store the run's input as its own grain in the run's namespace rather
    /// than by value in the manifest (#301).
    pub input_in_run_namespace: bool,
    /// Run-lease TTL in milliseconds (#299). `None` keeps
    /// [`lease::DEFAULT_RUN_LEASE_MS`](crate::lease::DEFAULT_RUN_LEASE_MS). A fixed ten minutes meant a crashed
    /// driver stalled its run for ten minutes; shortening it is only safe
    /// together with the mid-superstep renewal this release adds.
    pub lease_ms: Option<i64>,
    /// This driver's node identity for the run lease (#300). `None` derives
    /// it from the hostname and pid, the way the trigger evaluator already
    /// does. Two containers both running as PID 1 under one principal
    /// produced the SAME holder string and therefore did not exclude each
    /// other, which is the normal container shape.
    pub node: Option<String>,
    /// Ceiling on the effects a WHOLE run may settle (#295). `None` = no
    /// bound beyond the per-attempt one.
    pub max_run_effects: Option<u64>,
    /// Ceiling on the host tool calls a WHOLE run may dispatch (#295).
    pub max_tool_calls: Option<u64>,
    /// Cap on concurrently executing runs in this memory (#296). `None` =
    /// unbounded, today's behaviour.
    pub max_concurrent_runs: Option<u32>,
    /// Cap on concurrently executing runs for this principal (#296).
    pub max_concurrent_runs_per_principal: Option<u32>,
}

/// The per-LLM-call ceiling when a manifest pins none (§6.7 reserves it per
/// dispatch). Named because two places must agree: the prepared request and
/// the `gen_ai.request.max_tokens` the span reports for that same request.
pub(crate) const DEFAULT_LLM_MAX_TOKENS: u32 = 1024;

/// The temperature every abstract-node turn is issued at. Mirrors the literal
/// the pool builds its `ToolCallRequest` with (`executor.rs`); telemetry that
/// claimed a different one would be worse than none.
pub(crate) const LLM_TEMPERATURE: f64 = 0.0;

/// `FailCause` as OpenTelemetry's `error.type` — a low-cardinality string, so
/// the exhaustive match is the point: a new cause must be spelled here rather
/// than falling into a catch-all that hides it.
fn error_type_of(cause: &FailCause) -> &'static str {
    match cause {
        FailCause::Timeout => "timeout",
        FailCause::ExecutorError => "executor_error",
        FailCause::SchemaValidationFailed => "schema_validation_failed",
        FailCause::UserAborted => "user_aborted",
        FailCause::Unknown => "unknown",
        FailCause::ContextOverflow => "context_overflow",
    }
}

/// The `RunEvent` fields an `EffectSettled` carries beyond "it settled":
/// journaled usage, why the model stopped, and the failure class.
///
/// Keyed on the effect KIND, not on whether the numbers look model-shaped.
/// Every outcome carries `input_tokens`/`output_tokens` fields and a tool's
/// are structurally 0, so reporting them would put "0 tokens" on every tool
/// call; and a tool free to return any JSON could return a `stop_reason` of
/// its own, which is not a model's finish reason and must not be exported as
/// one.
fn settled_detail(
    kind: areev_run_core::EffectKind,
    outcome: &EffectOutcome,
) -> (Option<u64>, Option<u64>, Option<String>, Option<String>) {
    let is_llm = kind == areev_run_core::EffectKind::Llm;
    match outcome {
        EffectOutcome::Completed { result, input_tokens, output_tokens, .. } if is_llm => (
            Some(*input_tokens),
            Some(*output_tokens),
            result.get("stop_reason").and_then(|s| s.as_str()).map(str::to_string),
            None,
        ),
        EffectOutcome::Completed { .. } => (None, None, None, None),
        EffectOutcome::Failed { cause, .. } => {
            (None, None, None, Some(error_type_of(cause).to_string()))
        }
    }
}

/// `NodeExecutor` as the one-word kind an observer sees.
fn executor_kind_of(executor: &areev_run_core::NodeExecutor) -> &'static str {
    use areev_run_core::NodeExecutor::*;
    match executor {
        Host { .. } => "host",
        Client { .. } => "client",
        Abstract { .. } => "abstract",
        Subgraph { .. } => "subgraph",
        MemoryRead { .. } => "memory",
        Decide { .. } => "decide",
    }
}

/// A decision backend's failure as the effect's `FailCause` (C1–C3). The
/// §6.3 table then decides retry: transport, malformed-answer, deadline and
/// rate-limit faults are retryable under a decision node's `retries`; an
/// unaskable question is the plan's defect, and a refused egress or a
/// missing backend is not something a retry changes.
fn decide_fail_cause(e: &areev_core::decide::DecideError) -> FailCause {
    match e.code() {
        "DEC-E004" => FailCause::Timeout,
        "DEC-E006" => FailCause::SchemaValidationFailed,
        "DEC-E001" | "DEC-E008" => FailCause::Unknown,
        _ => FailCause::ExecutorError,
    }
}

/// A terminal outcome as the one-word label the run-outcome record and the
/// run index carry (`budget_exhausted:<axis>` names the axis).
fn outcome_label(outcome: &RunOutcome) -> String {
    match outcome {
        RunOutcome::Completed => "completed".to_string(),
        RunOutcome::Stalled { .. } => "stalled".to_string(),
        RunOutcome::Failed { .. } => "failed".to_string(),
        RunOutcome::Canceled { .. } => "canceled".to_string(),
        RunOutcome::BudgetExhausted { axis } => format!("budget_exhausted:{}", axis.as_str()),
    }
}

/// One `run:<id> mg:harness` link as the run index reads it:
/// `(created_at, run_id, session namespace if stamped)`.
type RunIndexRow = (i64, String, Option<String>);

/// What one census read found.
struct RunIndexPage {
    rows: Vec<RunIndexRow>,
    /// Runs matching the scope, before paging.
    total: usize,
    /// The scan reached the end of `agent:harness`, so the census is
    /// complete and a caller may stop widening its window.
    exhausted: bool,
    /// Runs excluded from a SCOPED read because their index entry predates
    /// the session-namespace stamp. Never silently dropped: a caller reports
    /// the count, and the unscoped listing still shows them.
    unattributed: usize,
}

/// Page size for walking one run's `agent:harness` records. NOT an upper
/// bound on how many there are: besides the per-run link, cancel and outcome
/// records, that namespace receives one Observation per brokered outbound
/// call, per refusal, per blob read and per redelivery — so a call-heavy run
/// has thousands, and its outcome (written last) sits behind all of them.
/// `run_outcome` pages to exhaustion; this only sizes each step.
const RUN_HARNESS_PAGE: usize = 512;

/// Upper bound on harness grains one run-index read will scan — the same
/// interim bound the loop adapter uses for its own full reads.
const RUN_INDEX_MAX_SCAN: usize = 1_000_000;

/// Whether a run's session namespace falls inside a listing scope: an exact
/// name, or an `"org.*"` prefix that also admits `org` itself and every
/// dotted descendant — the same reading every plural store read gives it.
pub fn ns_in_scope(scope: &str, ns: &str) -> bool {
    match scope.strip_suffix(".*") {
        Some(prefix) => ns == prefix || ns.strip_prefix(prefix).is_some_and(|r| r.starts_with('.')),
        None => ns == scope,
    }
}

/// A subgraph node's child run id. Derived from the parent, the node and the
/// node's ATTEMPT — attempts are monotonic across re-entry generations, so a
/// bounded cycle re-running the node gets a fresh child while a park and its
/// forwarded answer (which advance `effect_seq`, not `attempt`) address the
/// same one and resume it.
pub fn subgraph_run_id(parent_run_id: &str, node: &str, attempt: u32) -> String {
    let mut h = sha2::Sha256::new();
    h.update(parent_run_id.as_bytes());
    h.update([0u8]);
    h.update(node.as_bytes());
    h.update([0u8]);
    h.update(attempt.to_be_bytes());
    let mut s = String::with_capacity(16);
    for b in &h.finalize()[..8] {
        use std::fmt::Write;
        let _ = write!(s, "{b:02x}");
    }
    format!("{parent_run_id}~{s}")
}

/// The runtime driver: a host over one facade.
pub struct Runner {
    pub facade: Arc<AreevFacade>,
    pub clock: Arc<dyn Clock>,
    pub executor: Arc<dyn HostToolExecutor>,
    /// The tool-calling model for abstract nodes; None = abstract nodes
    /// refuse at load (`RUN-E006`).
    pub llm: Option<Arc<dyn areev_llm::ToolCallLlm>>,
    /// §6.10 streaming subscriber; None = no events. Observational only —
    /// a slow observer delays (and eventually loses) events, never the run.
    pub observer: Option<Arc<dyn crate::stream::RunObserver>>,
    pub ns: String,
    pub principal: String,
}

/// Collect every string in a JSON tree, in a stable depth-first order.
///
/// Walking rather than stringifying keeps structure out of the detector's
/// way: an object key called `subject` is not a person's name, and a
/// pseudonymized key would break the tool's schema.
fn collect_strings(v: &Value, out: &mut Vec<String>) {
    match v {
        Value::String(s) => out.push(s.clone()),
        Value::Array(a) => a.iter().for_each(|x| collect_strings(x, out)),
        Value::Object(o) => o.iter().for_each(|(_, x)| collect_strings(x, out)),
        _ => {}
    }
}

/// Rebuild the tree, substituting strings from `next` in the same order
/// [`collect_strings`] produced them.
fn substitute_strings(v: &Value, next: &mut std::vec::IntoIter<String>) -> Value {
    match v {
        Value::String(_) => next.next().map(Value::String).unwrap_or_else(|| v.clone()),
        Value::Array(a) => {
            Value::Array(a.iter().map(|x| substitute_strings(x, next)).collect())
        }
        Value::Object(o) => Value::Object(
            o.iter().map(|(k, x)| (k.clone(), substitute_strings(x, next))).collect(),
        ),
        other => other.clone(),
    }
}

/// Every placeholder-shaped token a model turn of this run produced — the
/// set [`Runner::rehydrate_from_model`] is allowed to resolve or refuse.
///
/// Read off the journaled outcomes, not accumulated from live calls, so a
/// resumed run and a fresh one agree: `results` holds every settled effect
/// the journal records, replayed or not. "Model turn" means an LLM effect,
/// and — conservatively — a subgraph's result, which may carry a child
/// model's output into the parent.
#[derive(Default)]
struct ModelTokens {
    scanned: std::collections::BTreeSet<JournalKey>,
    tokens: std::collections::BTreeSet<String>,
}

impl ModelTokens {
    fn refresh(
        &mut self,
        results: &BTreeMap<JournalKey, EffectOutcome>,
        model_origin: &dyn Fn(&JournalKey) -> bool,
        template: &str,
    ) -> Result<(), String> {
        let none = BTreeMap::new();
        for (key, outcome) in results {
            if self.scanned.contains(key) || !model_origin(key) {
                continue;
            }
            self.scanned.insert(key.clone());
            let EffectOutcome::Completed { result, .. } = outcome else { continue };
            let mut strings = Vec::new();
            collect_strings(result, &mut strings);
            for s in strings {
                // Against an empty mapping every placeholder is a leftover,
                // in the policy's own shape.
                let r = areev_core::anon::rehydrate_with_template(&s, &none, template)
                    .map_err(|e| e.to_string())?;
                self.tokens.extend(r.unmatched);
            }
        }
        Ok(())
    }
}

/// The anonymization boundary between run state and a model.
///
/// ## Why this seam and not the tool seam
///
/// The boundary is the **model**, not the tool. A host tool posting an invoice
/// to a vendor must receive real values — a pseudonymized supplier name writes
/// a corrupt invoice. A model extracting fields from that invoice works just as
/// well on `[ORG_7C1A]`.
///
/// The gate in the store is an egress boundary on *reads*, and an abstract
/// node's prompt is not a read: a trigger hands its payload straight into
/// `run start` in process, so under an `egress` policy the one place a model is
/// actually called was the one place the policy did not reach.
///
/// ## The round trip
///
/// Out, in [`Runner::pseudonymize_for_model`]: every string in the LLM effect's input is
/// pseudonymized. Back, in [`Runner::rehydrate_from_model`]: placeholders a
/// MODEL produced are rehydrated *before* dispatch, so the tool sees real
/// values again — whether they arrive as a model's tool-call arguments or as
/// its output flowing into a downstream node's input.
///
/// ## Only what a model produced (#350)
///
/// A token is rehydrated — or refused — only when it appears in the output of
/// a model turn in this run ([`ModelTokens`]). Text a host or code tool wrote
/// itself is dispatched verbatim, even when it has the placeholder's shape: a
/// deterministic pipeline that pseudonymizes a record before it leaves writes
/// `[PERSON_1]` markers no model produced, and neither failing them as
/// "unresolvable" nor splicing in an older mapping's value for them is right.
///
/// Rehydration **fails closed**. `rehydrate` leaves a token it cannot resolve
/// intact and reports it; dispatching a partially rehydrated call would post a
/// literal `[PERSON_A4F2]` to a vendor, so an unresolved model-produced token
/// fails the node instead.
impl Runner {
    /// Is a model-facing anonymization policy live for this run's namespace?
    fn anon_boundary_active(&self) -> Result<bool, RunError> {
        let mode = self
            .facade
            .with_store(|m| m.anon_active_mode(&self.ns))
            .map_err(err_run)?;
        Ok(matches!(mode.as_deref(), Some("egress") | Some("both")))
    }

    /// Pseudonymize everything on its way to the model.
    fn pseudonymize_for_model(&self, input: &Value) -> Result<Value, RunError> {
        if !self.anon_boundary_active()? {
            return Ok(input.clone());
        }
        let mut values = Vec::new();
        collect_strings(input, &mut values);
        if values.is_empty() {
            return Ok(input.clone());
        }
        self.facade
            .with_store(|m| m.anon_egress_text(&self.ns, &mut values))
            .map_err(err_run)?;
        Ok(substitute_strings(input, &mut values.into_iter()))
    }

    /// Rehydrate the placeholders a model produced, or say exactly which one
    /// could not be.
    ///
    /// Scoped to [`ModelTokens`]: a placeholder-shaped string no model turn of
    /// this run emitted is host-written text and passes through verbatim —
    /// never failed as unresolvable, never rewritten to whatever an older
    /// mapping of the namespace holds under that key (#350).
    ///
    /// `Err` here is the fail-closed case: the caller turns it into a failed
    /// effect rather than dispatching a call with a placeholder still in it.
    fn rehydrate_from_model(
        &self,
        input: &Value,
        model: &mut ModelTokens,
        results: &BTreeMap<JournalKey, EffectOutcome>,
        model_origin: &dyn Fn(&JournalKey) -> bool,
    ) -> Result<Value, String> {
        let active = self.anon_boundary_active().map_err(|e| e.to_string())?;
        if !active {
            return Ok(input.clone());
        }
        // The policy's OWN template, not the default. Scanning for the default
        // silhouette while the memory mints another shape would report no
        // leftovers and fail open — the exact inversion of what this check is
        // for.
        let template = self
            .facade
            .with_store(|m| m.anon_placeholder(&self.ns))
            .map_err(|e| e.to_string())?
            .unwrap_or_else(|| "[{CATEGORY}_{ID}]".to_string());
        let mut values = Vec::new();
        collect_strings(input, &mut values);
        if values.is_empty() {
            return Ok(input.clone());
        }
        model.refresh(results, model_origin, &template)?;
        if model.tokens.is_empty() {
            // No model turn produced a placeholder: nothing here is ours to
            // resolve, whatever its shape.
            return Ok(input.clone());
        }
        let mapping: BTreeMap<String, String> = self
            .facade
            .with_store(|m| m.anon_mapping_for(&self.ns))
            .map_err(|e| e.to_string())?
            .into_iter()
            .filter(|(k, _)| model.tokens.contains(k))
            .collect();
        let mut out = Vec::with_capacity(values.len());
        let mut unmatched: Vec<String> = Vec::new();
        for v in values {
            let r = areev_core::anon::rehydrate_with_template(&v, &mapping, &template)
                .map_err(|e| e.to_string())?;
            // A leftover no model produced is host-written text, left as is.
            for u in r.unmatched.into_iter().filter(|u| model.tokens.contains(u)) {
                if !unmatched.contains(&u) {
                    unmatched.push(u);
                }
            }
            out.push(r.text);
        }
        if !unmatched.is_empty() {
            return Err(format!(
                "the model produced placeholder{} this run cannot resolve: {}. \
                 Dispatching would send the placeholder itself to the tool, so \
                 the effect is refused instead",
                if unmatched.len() == 1 { "" } else { "s" },
                unmatched.join(", ")
            ));
        }
        Ok(substitute_strings(input, &mut out.into_iter()))
    }
}

/// The request fields of one LLM turn, read off its JOURNALED input (#391):
/// the node's `llm` block as the scheduler wrote it under `llm`, else the
/// run's defaults. Read from the journal rather than the manifest so the
/// request a turn went out with is a function of the record alone.
pub(crate) struct TurnRequest {
    pub system: Option<String>,
    /// `None` = send no temperature (`"omit"`).
    pub temperature: Option<f64>,
    pub max_tokens: u32,
    pub response_format: Option<areev_llm::ResponseFormat>,
    pub parallel_tool_calls: Option<bool>,
}

impl TurnRequest {
    pub(crate) fn from_input(input: &Value, manifest: &RunManifest) -> Self {
        let block = input.get("llm");
        let field = |k: &str| block.and_then(|b| b.get(k));
        TurnRequest {
            system: field("system").and_then(Value::as_str).map(str::to_string),
            temperature: match field("temperature") {
                Some(Value::String(s)) if s == "omit" => None,
                Some(t) => t.as_f64().or(Some(LLM_TEMPERATURE)),
                None => Some(LLM_TEMPERATURE),
            },
            max_tokens: field("max_tokens")
                .and_then(Value::as_u64)
                .and_then(|n| u32::try_from(n).ok())
                .unwrap_or_else(|| manifest.llm_max_tokens.unwrap_or(DEFAULT_LLM_MAX_TOKENS)),
            response_format: field("response_format").and_then(areev_llm::ResponseFormat::from_openai),
            parallel_tool_calls: field("parallel_tool_calls").and_then(Value::as_bool),
        }
    }
}

/// Journal-shaped transcript messages → the tool-calling seam's shape.
/// Pure translation, on the driver thread; the pool never parses state.
fn seam_messages(input: &Value) -> Vec<areev_llm::ChatMessage> {
    let Some(msgs) = input.get("messages").and_then(|m| m.as_array()) else {
        return Vec::new();
    };
    msgs.iter()
        .filter_map(|m| {
            let role = m.get("role")?.as_str()?;
            match role {
                "user" => Some(areev_llm::ChatMessage::User(
                    m.get("content")
                        .map(|c| {
                            c.as_str().map(str::to_string).unwrap_or_else(|| c.to_string())
                        })
                        .unwrap_or_default(),
                )),
                "assistant" => Some(areev_llm::ChatMessage::Assistant {
                    text: m.get("text").and_then(|t| t.as_str()).map(str::to_string),
                    tool_calls: m
                        .get("tool_calls")
                        .and_then(|c| c.as_array())
                        .map(|arr| {
                            arr.iter()
                                .map(|c| areev_llm::ToolCallOut {
                                    id: c
                                        .get("id")
                                        .and_then(|v| v.as_str())
                                        .unwrap_or_default()
                                        .to_string(),
                                    name: c
                                        .get("name")
                                        .and_then(|v| v.as_str())
                                        .unwrap_or_default()
                                        .to_string(),
                                    arguments: c
                                        .get("arguments")
                                        .cloned()
                                        .unwrap_or(Value::Null),
                                    arguments_raw: None,
                                })
                                .collect()
                        })
                        .unwrap_or_default(),
                    // #284: the provider's own content blocks, carried back
                    // into the transcript so the adapter can replay the turn
                    // byte-identically. Absent on every transcript written
                    // before this existed, and on every provider that has
                    // none, so nothing moves for them.
                    provider_content: m.get("provider_content").cloned(),
                }),
                "tool" => Some(areev_llm::ChatMessage::ToolResult {
                    tool_call_id: m
                        .get("tool_call_id")
                        .and_then(|v| v.as_str())
                        .unwrap_or_default()
                        .to_string(),
                    content: m
                        .get("content")
                        .map(|c| {
                            c.as_str().map(str::to_string).unwrap_or_else(|| c.to_string())
                        })
                        .unwrap_or_default(),
                    is_error: m.get("is_error").and_then(|v| v.as_bool()).unwrap_or(false),
                }),
                _ => None,
            }
        })
        .collect()
}

/// The pure §6.11 argument validator over the manifest's strict schemas:
/// non-strict (or schema-less) tools pass; strict ones run
/// `validate_instance`, and the violation kind becomes the re-prompt text.
pub(crate) fn make_validate_args(
    schemas: &BTreeMap<String, Value>,
) -> impl Fn(&str, &Value) -> std::result::Result<(), String> + '_ {
    move |tool: &str, args: &Value| match schemas.get(tool) {
        Some(schema) => areev_core::types::validate_instance(args, schema)
            .map_err(|k| format!("schema violation: {}", k.as_str())),
        None => Ok(()),
    }
}

pub(crate) fn builtin_eval(edge: &areev_run_core::PlanEdge, state: &Value) -> bool {
    edge.cond
        .as_ref()
        .map(|c| areev_run_core::cond::eval(c, state))
        .unwrap_or(true)
}

/// Name the top-level fields where two scheduler-state JSONs disagree — the
/// RUN-E009 diagnostic. "Diverges" without saying WHERE would send whoever
/// reads the report straight back to a debugger.
fn diff_fields(stored: &Value, replayed: &Value) -> String {
    let (Some(a), Some(b)) = (stored.as_object(), replayed.as_object()) else {
        return "state shape differs".into();
    };
    let mut diffs: Vec<String> = Vec::new();
    for key in a.keys().chain(b.keys()) {
        if a.get(key) != b.get(key) && !diffs.iter().any(|d| d.starts_with(key.as_str())) {
            diffs.push(format!(
                "{key}: stored={} replayed={}",
                a.get(key).map(|v| v.to_string()).unwrap_or_else(|| "∅".into()),
                b.get(key).map(|v| v.to_string()).unwrap_or_else(|| "∅".into()),
            ));
        }
    }
    diffs.join("; ")
}

pub(crate) fn idempotency_key(key: &JournalKey, input: &Value) -> String {
    let mut h = sha2::Sha256::new();
    h.update(key.idempotency_prefix().as_bytes());
    h.update([0x1f]);
    h.update(input.to_string().as_bytes());
    let out = h.finalize();
    let mut s = String::with_capacity(64);
    for b in out {
        use std::fmt::Write;
        let _ = write!(s, "{b:02x}");
    }
    s
}

impl Runner {
    /// D5 enforcement at the driver boundary. The facade session's rights
    /// decide (owner unless the host bound a principal — the single-user
    /// path never meets authorization, same as every other surface); a
    /// bound session needs the run verb granted in the file.
    fn check_run_verb(&self, verb: areev_core::authz::Verb) -> Result<(), RunError> {
        self.facade
            .authz()
            .check(verb, &self.ns)
            .map_err(|e| RunError::Unauthorized { what: e.to_string() })
    }

    /// `respond` authorizes the RESPONDER, not the session: on a governed
    /// session (non-owner), the responder principal's own grants must cover
    /// `run.respond` — an approval boundary where "whoever holds the
    /// process" can approve is not an approval boundary.
    fn check_responder(&self, responder: &str) -> Result<(), RunError> {
        if self.facade.authz().is_owner() {
            return Ok(());
        }
        let session = self
            .facade
            .principal_session(responder)
            .map_err(err_run)?;
        session
            .authz()
            .check(areev_core::authz::Verb::RunRespond, &self.ns)
            .map_err(|e| RunError::Unauthorized { what: e.to_string() })
    }

    /// A declared memory read (#255) needs `read` on its namespace. Checked
    /// before the run exists, so a plan asking for a namespace the session
    /// cannot see refuses at start instead of failing a node mid-run — and
    /// checked again at every read, since a resume need not share the
    /// starting session's grants.
    fn check_read_grants(&self, manifest: &RunManifest) -> Result<(), RunError> {
        let authz = self.facade.authz();
        for p in &manifest.pinned {
            let Some(ns) = p.read.as_ref().and_then(|r| r.get("ns")).and_then(|n| n.as_str())
            else {
                continue;
            };
            authz.check(areev_core::authz::Verb::Read, ns).map_err(|e| RunError::Unauthorized {
                what: format!("node '{}' declares a read of '{ns}': {e}", p.node),
            })?;
        }
        Ok(())
    }

    fn load_plan(&self, plan_hash: &Hash) -> Result<PlanGraph, RunError> {
        let wf = self
            .facade
            .with_store(|m| m.get(plan_hash))
            .map_err(|e| RunError::UnresolvedRef { what: format!("workflow: {e}") })?
            .to_workflow()
            .map_err(|e| RunError::InvalidPlan { why: e.to_string() })?;
        PlanGraph::build(&wf)
    }

    /// The Definition one offered tool is shown to the model as. A pinned
    /// Definition is read by its frozen address, with its locked keys taken
    /// out of the schema (#393) — the model is not invited to set what it
    /// cannot choose. A dynamic tool (#392) is the entry the manifest pinned:
    /// its own name, description and schema, never the dispatch Definition's.
    fn offered_definition(
        &self,
        t: &areev_run_core::OfferedTool,
    ) -> Result<areev_core::types::Tool, RunError> {
        if let Some(d) = &t.dynamic {
            let mut def = areev_core::types::Tool::new(&d.name)
                .kind(areev_core::types::ToolKind::Definition);
            if !d.description.is_empty() {
                def = def.tool_description(&d.description);
            }
            // The same stripped form the scheduler validates the model's
            // arguments against (run-core `DynamicTool::model_schema`).
            def.input_schema = Some(d.model_schema(t.locked.as_ref()));
            return Ok(def);
        }
        let h = Hash::from_hex(&t.tool_hash).map_err(err_run)?;
        let mut def = self
            .facade
            .with_store(|m| m.get(&h))
            .map_err(err_run)?
            .to_tool()
            .map_err(err_run)?;
        if let (Some(lock), Some(schema)) = (&t.locked, def.input_schema.as_mut()) {
            crate::llmcfg::strip_locked(schema, lock);
        }
        Ok(def)
    }

    /// Pseudonymize a dynamic tool's prose for the model: its description,
    /// and every `description`, `title`, `default` and `examples` value in its
    /// schema. A no-op without an `egress` boundary.
    fn pseudonymize_tool_text(&self, def: &mut areev_core::types::Tool) -> Result<(), RunError> {
        if !self.anon_boundary_active()? {
            return Ok(());
        }
        const PROSE: [&str; 4] = ["description", "title", "default", "examples"];
        // One walk order for both passes, so the values come back to the
        // slots they left.
        fn walk(schema: &mut Value, f: &mut dyn FnMut(&mut Value)) {
            let Value::Object(o) = schema else { return };
            for (k, v) in o.iter_mut() {
                match k.as_str() {
                    k if PROSE.contains(&k) => f(v),
                    "properties" => {
                        if let Value::Object(props) = v {
                            for p in props.values_mut() {
                                walk(p, f);
                            }
                        }
                    }
                    "items" | "additionalProperties" => walk(v, f),
                    _ => {}
                }
            }
        }
        let mut schema = def.input_schema.clone().unwrap_or(Value::Null);
        let mut slots = vec![Value::String(def.tool_description.clone().unwrap_or_default())];
        walk(&mut schema, &mut |v| slots.push(v.clone()));
        let masked = self.pseudonymize_for_model(&Value::Array(slots))?;
        let mut masked = masked.as_array().cloned().unwrap_or_default().into_iter();
        if let Some(Value::String(d)) = masked.next() {
            if def.tool_description.is_some() {
                def.tool_description = Some(d);
            }
        }
        walk(&mut schema, &mut |v| {
            if let Some(m) = masked.next() {
                *v = m;
            }
        });
        if def.input_schema.is_some() {
            def.input_schema = Some(schema);
        }
        Ok(())
    }

    fn load_manifest(&self, run_id: &str) -> Result<RunManifest, RunError> {
        self.facade.with_store(|m| RunManifest::load(m, run_id))
    }

    /// The §6.11 argument-validation table: tool_name → input_schema for
    /// every STRICT pinned Definition. Loaded once per drive/verify (the
    /// scheduler's `validate_args` closure must never touch the store), and
    /// a pure function of the manifest — so live and replay agree.
    fn load_arg_schemas(
        &self,
        manifest: &RunManifest,
    ) -> Result<BTreeMap<String, Value>, RunError> {
        crate::shadow::arg_schemas_for(&self.facade, manifest)
    }

    /// Start a new run of the Workflow at `plan_hash` and drive it until it
    /// finishes or parks on a Client ask.
    pub fn start(
        &self,
        plan_hash: &Hash,
        run_id: &str,
        input: Value,
        opts: &RunOptions,
    ) -> Result<RunSession, RunError> {
        self.check_run_verb(areev_core::authz::Verb::RunExecute)?;
        areev_core::types::validate_run_id(run_id).map_err(err_run)?;
        // A run id is an identity: starting twice under one id would
        // interleave two journals (§5.4 forks are the sanctioned reuse).
        if self.load_manifest(run_id).is_ok() {
            return Err(RunError::Tainted {
                why: format!("run '{run_id}' already exists — resume it or fork"),
            });
        }
        let plan = self.load_plan(plan_hash)?;
        let llm_window = self.llm.as_ref().and_then(|l| l.context_window());
        let manifest = self.facade.with_store(|m| {
            RunManifest::resolve(
                m,
                &self.ns,
                run_id,
                plan_hash,
                &plan,
                &self.principal,
                opts.budgets,
                opts.ask_ttl_sec,
                input.clone(),
                self.llm.is_some(),
            )
            .map(|m| m.with_limits(opts, llm_window))
        })?;
        // Freeze the model, the engine and the attribution alongside every
        // other thing a run is pinned to (#287, #288, #293, #294, #301).
        let manifest = manifest
            .with_llm_pin(self.llm_pin())
            .with_engine_pin()
            .with_attribution(opts)
            // C3's V7 half: a decision node on a host with no backend is
            // RUN-E030 here, before anything is written.
            .with_decider_pin(self.decider_pin())?;
        // #294: a plan that pins a CONFIRMATION Definition refuses at start
        // unless the host opted in, naming the node and the flag.
        //
        // The `--allow-executor` reasoning: a Definition can arrive in a
        // bundle or a pack, and a weakening delivered together with the thing
        // it weakens is not a permission. Refusing here rather than at the
        // ask means a run never starts down a path whose approval boundary
        // this host has not sanctioned.
        if !manifest.allow_confirmation_asks {
            if let Some(p) = manifest
                .pinned
                .iter()
                .find(|p| p.ask_kind.as_deref() == Some(crate::manifest::ASK_KIND_CONFIRMATION))
            {
                return Err(RunError::CodeExecRefused {
                    condition: format!(
                        "node '{}' binds a tool declaring ask_kind=\"confirmation\", which \
                         lets the run's own initiator settle it — this host has not \
                         allowed that. Pass --allow-confirmation-asks (or \
                         $AREEV_RUN_ALLOW_CONFIRMATION_ASKS=1) if that is intended",
                        p.node
                    ),
                });
            }
        }
        // Refuse an unpinned code executor before the run exists: it would
        // otherwise take a lease, write a manifest, and fail on first
        // dispatch, leaving a terminal run to explain. The address is named
        // so pinning it is a copy-paste.
        for p in manifest.code_pins() {
            if let Some(uri) = &p.executor_uri {
                if !self.executor.code_allowed(&p.tool_hash, uri) {
                    return Err(RunError::CodeExecRefused {
                        condition: format!(
                            "node '{}' names executor {uri}, which this host has not \
                             pinned — pin it: --allow-executor {} (`areev run` and \
                             `areev trigger`), allow_executor (Python/Node), or \
                             $AREEV_RUN_ALLOW_EXECUTOR (any surface, including a \
                             heartbeat)",
                            p.node,
                            crate::executor::strip_cas(uri)
                        ),
                    });
                }
                // Same fail-closed shape for the declared runtime (#86): a
                // plan asking for the sandbox on a host with none refuses
                // BEFORE the run exists, naming the missing configuration.
                if let Some(rt) = &p.runtime {
                    if !self.executor.runtime_supported(rt) {
                        return Err(RunError::CodeExecRefused {
                            condition: format!(
                                "node '{}' declares runtime {rt:?}, which this host \
                                 cannot dispatch — configure the sandbox: \
                                 --sandbox-cmd areev-sandbox (`areev run` and \
                                 `areev trigger`), sandbox_cmd (Python/Node), or \
                                 $AREEV_RUN_SANDBOX_CMD (any surface, including a \
                                 heartbeat)",
                                p.node
                            ),
                        });
                    }
                }
            }
        }
        // An abstract node crosses the model boundary, and a boundary whose
        // placeholders are not value-derived cannot be replayed. Refuse here
        // rather than let `verify` diverge later, which would look like a
        // journal integrity failure instead of a configuration one.
        if manifest.pinned.iter().any(|p| p.executor == "abstract") && self.anon_boundary_active()?
        {
            let policies = self.facade.with_store(|m| m.anon_policies()).map_err(err_run)?;
            if let Some((_, policy)) = policies.iter().find(|(ns, _)| ns == &self.ns) {
                if policy.scope != "memory" {
                    return Err(RunError::AnonReplayUnsafe {
                        ns: self.ns.clone(),
                        scope: policy.scope.clone(),
                    });
                }
            }
        }
        self.check_read_grants(&manifest)?;
        // Claim the concurrency slot BEFORE the manifest is written (#296):
        // a start refused at the cap must leave nothing behind, so the run id
        // stays genuinely free rather than half-existing as a manifest with
        // no journal.
        let slots = crate::lease::RunSlots::claim(
            &self.facade,
            run_id,
            &self.principal,
            self.clock.now_ms() as i64,
            opts.lease_ms.unwrap_or(crate::lease::DEFAULT_RUN_LEASE_MS),
            opts.max_concurrent_runs,
            opts.max_concurrent_runs_per_principal,
        )?;
        // #301: under `--input-placement run-ns` the input is its own grain
        // in the run's namespace, and the manifest keeps only its address.
        let input_ns = opts.input_in_run_namespace.then(|| self.ns.clone());
        if let Err(e) = self
            .facade
            .with_store(|m| manifest.persist_with_input(m, &self.ns, input_ns.as_deref()))
            .map_err(err_run)
        {
            let mut slots = slots;
            slots.release(&self.facade);
            return Err(e);
        }

        let st = SchedulerState::new(run_id, &plan);
        let events = vec![
            EventIn::ClockReading { unix_ms: self.clock.now_ms() },
            EventIn::Start { input },
        ];
        self.drive(
            &plan,
            plan_hash,
            &manifest,
            st,
            JournalView::default(),
            events,
            opts,
            Some(slots),
        )
    }

    /// This driver's LLM pin, or `None` when no transport is configured.
    ///
    /// Built from the transport's own answers — `provider()`, `model()`, and
    /// the two defaulted seam methods a host can override — so a host
    /// wrapping its own transport can have Areev enforce the host's own
    /// notion of "the same configuration" without Areev modelling it.
    pub(crate) fn llm_pin(&self) -> Option<crate::manifest::LlmPin> {
        self.llm.as_ref().map(|l| crate::manifest::LlmPin {
            provider: l.provider().to_string(),
            model: l.model().to_string(),
            region: l.region().map(str::to_string),
            tag: l.pin_tag().map(str::to_string),
            profile: l.request_profile_digest(),
        })
    }

    /// Install a decision backend (`docs/decision-model-proposal.md` C1–C3)
    /// — the one call a host makes.
    ///
    /// Wraps [`Runner::executor`] in a [`crate::DecidingExecutor`], so it
    /// composes with whatever executor stack the host already built. Pass the
    /// host's pseudonymizing chain (`areev_llm::PseudonymizingDecider` over
    /// `areev_llm::decide::resolve_chain`): the driver additionally routes a
    /// decision's `state` through the SAME namespace egress boundary an
    /// abstract node's prompt takes, but a remote backend under no namespace
    /// policy still sees what the host's wrapper lets out.
    pub fn with_decider(mut self, decider: Arc<dyn areev_core::decide::DecisionBackend>) -> Self {
        self.executor = Arc::new(crate::executor::DecidingExecutor::new(
            Arc::clone(&self.executor),
            decider,
        ));
        self
    }

    /// This host's decision-backend pin, or `None` when none is installed.
    pub(crate) fn decider_pin(&self) -> Option<crate::manifest::DeciderPin> {
        self.executor.decider().map(|d| crate::manifest::DeciderPin {
            describe: d.describe(),
            calibrated: d.calibrated(),
        })
    }

    /// Refuse a resume of a run with a decision node on a host that cannot
    /// answer it (`RUN-E030`), before the lease — the resume twin of the
    /// start-time refusal. The scheduler's optional asks need no such check:
    /// without a backend they fail open at dispatch.
    fn check_decider(&self, manifest: &RunManifest) -> Result<(), RunError> {
        if self.executor.decider().is_some() {
            return Ok(());
        }
        match manifest.pinned.iter().find(|p| p.executor == crate::manifest::DECIDE_EXECUTOR) {
            Some(p) => Err(RunError::NoDecider { node: p.node.clone() }),
            None => Ok(()),
        }
    }

    /// Answer one decision effect (C1–C3) through the host's backend.
    ///
    /// - A **decision node** (C3) builds `{state, questions}` from the node's
    ///   input and its frozen `decide` pin: the pinned questions win over an
    ///   input `questions` key; `state` is the input's `state` key, else the
    ///   whole input. A strict Definition's `input_schema` is checked against
    ///   that `{state, questions}` exactly as a model's call to a strict host
    ///   tool is. The result lands under the pin's `into` key.
    /// - A **scheduler ask** (C1/C2, `mg:decide`) carries its request in
    ///   `input.decide`; the answer is returned bare.
    ///
    /// `state` crosses the namespace egress boundary first
    /// (`pseudonymize_for_model`, the one an abstract node's prompt takes);
    /// a refusal there sends nothing. Every failure is a FAILED effect — for
    /// the scheduler's asks that is the fail-open path, for a decision node
    /// the node's ordinary retry table.
    fn run_decide(
        &self,
        key: &JournalKey,
        executor: &NodeExecutor,
        input: &Value,
        manifest: &RunManifest,
        schemas: &BTreeMap<String, Value>,
    ) -> EffectOutcome {
        use areev_core::decide::{questions_from_wire, DecideRequest};
        let fail = |cause: FailCause, detail: String| EffectOutcome::Failed {
            journal_bytes: detail.len() as u64,
            cause,
            detail,
        };
        let NodeExecutor::Decide { tool_name, .. } = executor else {
            return fail(FailCause::Unknown, "not a decision effect".into());
        };
        let ask = input.get("decide").filter(|_| tool_name == areev_run_core::DECIDE_TOOL);
        let (state, questions, into) = match ask {
            Some(ask) => (
                ask.get("state").cloned().unwrap_or(Value::Null),
                ask.get("questions").cloned().unwrap_or(Value::Null),
                None,
            ),
            None => {
                let decl = manifest
                    .pinned
                    .iter()
                    .find(|p| p.node == key.node && p.executor == crate::manifest::DECIDE_EXECUTOR)
                    .and_then(|p| p.decide.clone())
                    .unwrap_or_else(|| json!({}));
                let questions = decl
                    .get("questions")
                    .cloned()
                    .or_else(|| input.get("questions").cloned())
                    .unwrap_or(Value::Null);
                let state = input.get("state").cloned().unwrap_or_else(|| input.clone());
                if let Some(schema) = schemas.get(tool_name) {
                    let args = json!({ "state": state, "questions": questions });
                    if let Err(k) = areev_core::types::validate_instance(&args, schema) {
                        return fail(
                            FailCause::SchemaValidationFailed,
                            format!("decision input fails the tool's schema: {}", k.as_str()),
                        );
                    }
                }
                let into = decl
                    .get("into")
                    .and_then(Value::as_str)
                    .map(str::to_string)
                    .unwrap_or_else(|| key.node.clone());
                (state, questions, Some(into))
            }
        };
        let questions = match questions_from_wire(&questions) {
            Ok(q) => q,
            Err(e) => return fail(FailCause::SchemaValidationFailed, e.to_string()),
        };
        let Some(decider) = self.executor.decider() else {
            return fail(
                FailCause::Unknown,
                RunError::NoDecider { node: key.node.clone() }.to_string(),
            );
        };
        let state = match self.pseudonymize_for_model(&state) {
            Ok(v) => v,
            Err(e) => {
                return fail(
                    FailCause::Unknown,
                    format!("egress pseudonymization of the decision state failed, nothing was sent: {e}"),
                )
            }
        };
        match decider.decide(&DecideRequest::new(state, questions)) {
            Ok(d) => {
                let body = d.to_json();
                let result = match into {
                    Some(k) => json!({ k: body }),
                    None => body,
                };
                let journal_bytes = result.to_string().len() as u64;
                EffectOutcome::Completed {
                    result,
                    journal_bytes,
                    input_tokens: d.input_tokens.unwrap_or(0),
                    output_tokens: d.output_tokens.unwrap_or(0),
                    // The provider-reported price, charged to the run's
                    // USD budget like a model call's; 0 when unreported.
                    usd_micros: d.usd_micros.unwrap_or(0),
                }
            }
            Err(e) => fail(decide_fail_cause(&e), e.to_string()),
        }
    }

    /// Refuse a resume whose model configuration differs from the one the run
    /// started under (#287).
    ///
    /// A manifest with no pin — every run written before 1.9.0 — resumes
    /// under anything, as it always did.
    fn check_llm_pin(&self, manifest: &RunManifest) -> Result<(), RunError> {
        let Some(pinned) = &manifest.llm else {
            return Ok(());
        };
        match self.llm_pin() {
            Some(offered) if pinned.matches(&offered) => Ok(()),
            // A pinned run resumed with NO transport is a mismatch, not a
            // missing-LLM error: the run has an answer for what it needs and
            // this driver is not it. RUN-E006 would send an operator looking
            // for a configuration problem instead of a pin.
            other => Err(RunError::ModelMismatch {
                pinned: pinned.describe(),
                offered: other
                    .map(|p| p.describe())
                    .unwrap_or_else(|| "no LLM configured".into()),
            }),
        }
    }

    /// Refuse a resume across a scheduler generation (#288).
    ///
    /// The VERSION string is deliberately not compared: a patch upgrade must
    /// not strand every parked approval run. Only the epoch, which moves
    /// exactly when a change makes an existing journal replay differently.
    fn check_engine_pin(&self, manifest: &RunManifest) -> Result<(), RunError> {
        let Some(pinned) = &manifest.engine else {
            return Ok(());
        };
        let this = areev_run_core::SCHEDULER_EPOCH;
        if pinned.scheduler_epoch == this {
            return Ok(());
        }
        Err(RunError::EngineMismatch {
            written_by: pinned.version.clone(),
            epoch: pinned.scheduler_epoch,
            this_epoch: this,
        })
    }

    /// Resume a run from its latest checkpoint: settle expired asks, adopt
    /// journaled responses, re-deliver dangling intents (§5.3).
    pub fn resume(&self, run_id: &str, opts: &RunOptions) -> Result<RunSession, RunError> {
        self.check_run_verb(areev_core::authz::Verb::RunExecute)?;
        self.resume_run(run_id, opts, true)
    }

    /// `resume` past its grant check. `check_llm` is false only on cancel's
    /// finalization of a paused run (#344), which dispatches nothing and so
    /// has no model to be pinned to — and whose caller holds `run.cancel`,
    /// deliberately the lower bar.
    fn resume_run(
        &self,
        run_id: &str,
        opts: &RunOptions,
        check_llm: bool,
    ) -> Result<RunSession, RunError> {
        let manifest = self.load_manifest(run_id)?;
        // Before the lease and before any grain is written (#287, #288): a
        // run that must not continue here must not LOOK like it started to.
        self.check_engine_pin(&manifest)?;
        if check_llm {
            self.check_llm_pin(&manifest)?;
            self.check_decider(&manifest)?;
        }
        let plan_hash = Hash::from_hex(&manifest.plan_hash)
            .map_err(|_| RunError::ManifestMismatch { why: "bad plan hash".into() })?;
        let plan = self.load_plan(&plan_hash)?;
        let mut view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, run_id))
            .map_err(err_run)?;
        let Some(last) = view.checkpoints.last() else {
            // A crash inside the FIRST superstep: the manifest exists but no
            // checkpoint does. The manifest carries the input for exactly
            // this — rebuild from it (dangling intents are adopted by the
            // normal §5.3 path below) rather than bricking the run id.
            let st = SchedulerState::new(run_id, &plan);
            // No `Resumed` here: with no checkpoint there is no previous
            // close to measure a gap from, and the first superstep opens at
            // this very reading exactly as a fresh `start` would.
            let events = vec![
                EventIn::ClockReading { unix_ms: self.clock.now_ms() },
                EventIn::Start { input: manifest.input.clone() },
            ];
            return self.drive(&plan, &plan_hash, &manifest, st, view, events, opts, None);
        };
        let st: SchedulerState = serde_json::from_value(last.scheduler.clone())
            .map_err(|e| RunError::ManifestMismatch { why: format!("checkpoint state: {e}") })?;
        if st.run_id != run_id {
            return Err(RunError::ManifestMismatch {
                why: format!("checkpoint belongs to run '{}'", st.run_id),
            });
        }
        if st.is_terminal() {
            return Ok(RunSession::Finished {
                outcome: st.outcome().cloned().expect("terminal"),
                run_id: run_id.to_string(),
            });
        }

        let now = self.clock.now_ms();
        // Continuing a PAUSED run consumes its request (#344), and says so in
        // the journal: who resumed it and which request that answered. A
        // request that was never honoured (asked while the run sat parked on
        // an ask, or crashed) is NOT consumed — the drive below honours it at
        // the next boundary, which is what the asker was promised.
        if view.pause.is_paused() && matches!(st.phase, areev_run_core::Phase::Idle) {
            if let Some(req) = view.pause.active.clone() {
                self.facade
                    .with_store(|m| {
                        journal::write_pause_released(
                            m,
                            &self.ns,
                            run_id,
                            &req.request,
                            now,
                            &self.principal,
                        )
                    })
                    .map_err(err_run)?;
                view.pause.apply(&journal::PauseEvent::Released {
                    request: req.request,
                    by: self.principal.clone(),
                    at: now as i64,
                });
            }
        }
        // A run picked back up between supersteps stopped existing for the
        // span between its last close and this reading. `Resumed` marks that
        // so the coming open accrues the gap as ELAPSED rather than charging
        // it as active wall — and journals the reading, which is what lets
        // `verify` reproduce the boundary it cannot otherwise see.
        //
        // Idle only: a parked checkpoint is `Phase::Open` and already set its
        // own pause point at the park, so marking again would double-count.
        //
        // A fork's SEED is the other exclusion, and for a different reason: it
        // is a synthetic checkpoint `fork()` wrote, not one a superstep closed,
        // and the fork has never executed. Nothing stopped, so nothing resumed
        // — and on a time-travel fork the span from the base checkpoint to now
        // can be days, which is lineage, not this run's downtime. Once a fork
        // HAS executed, a later resume of it marks the boundary like any other.
        let fork_seed = manifest.fork_of.is_some() && view.checkpoints.len() == 1;
        let mut events = if matches!(st.phase, areev_run_core::Phase::Idle) && !fork_seed {
            vec![EventIn::Resumed, EventIn::ClockReading { unix_ms: now }]
        } else {
            vec![EventIn::ClockReading { unix_ms: now }]
        };

        // Settled-while-parked responses (a respond() wrote the result) and
        // expired asks (settled here as Timeout — never re-dispatched,
        // never silently eternal).
        let executors = manifest.executors();
        for (id, pending) in st.pending_asks.clone() {
            if matches!(executors.get(pending.node_idx), Some(NodeExecutor::Subgraph { .. })) {
                // A bubbled ask settles — or expires — in the CHILD. Either
                // way the parent's move is the same: forward, and let the
                // child's own resume decide what the answer was.
                let child =
                    subgraph_run_id(run_id, &pending.key.node, pending.key.attempt);
                let open = self.child_asks(&child)?.iter().any(|a| a.tool_call_id == id);
                let expired = pending
                    .ask
                    .expires_at_sec
                    .is_some_and(|exp| (now / 1000) as i64 > exp);
                if !open || expired {
                    events.push(EventIn::AskForwarded { tool_call_id: id });
                }
                continue;
            }
            if let Some(entry) = view.entries.get(&pending.key) {
                if let Some((_, outcome)) = &entry.result {
                    events.push(EventIn::ResponseSettled {
                        tool_call_id: id.clone(),
                        outcome: outcome.clone(),
                    });
                    continue;
                }
            }
            if pending
                .ask
                .expires_at_sec
                .is_some_and(|exp| (now / 1000) as i64 > exp)
            {
                let detail = "client ask expired".to_string();
                let outcome = EffectOutcome::Failed {
                    cause: FailCause::Timeout,
                    journal_bytes: detail.len() as u64,
                    detail,
                };
                let executor = executors
                    .get(pending.node_idx)
                    .cloned()
                    .ok_or_else(|| RunError::ManifestMismatch {
                        why: "ask node outside manifest".into(),
                    })?;
                let intent = view
                    .entries
                    .get(&pending.key)
                    .map(|e| e.intent)
                    .ok_or_else(|| RunError::Tainted {
                        why: "pending ask with no journaled intent".into(),
                    })?;
                self.facade
                    .with_store(|m| {
                        journal::write_result(
                            m,
                            &self.ns,
                            run_id,
                            &plan_hash,
                            &intent,
                            &pending.key,
                            &executor,
                            &outcome,
                            st.superstep,
                            now,
                            &self.principal,
                        )
                    })
                    .map_err(err_run)?;
                events.push(EventIn::ResponseSettled { tool_call_id: id, outcome });
            }
        }

        self.drive(&plan, &plan_hash, &manifest, st, view, events, opts, None)
    }

    /// §5.4 time-travel fork: a NEW run descending from one of `base`'s
    /// checkpoints. The fork's manifest records `(base_run,
    /// base_checkpoint)`, an `mg:fork_of` Fact indexes the relationship,
    /// and the SEED checkpoint `derived_from`s the base checkpoint — the
    /// verifiability chain. The base run is never touched.
    ///
    /// - **Same plan** (`new_plan = None`): the base checkpoint's scheduler
    ///   state carries over under the new run id; the fork re-executes
    ///   everything past the checkpoint under its own journal keys.
    ///   A `BudgetExhausted` terminal re-opens (that is how a run continues
    ///   under raised budgets — budgets are the fork's own knobs).
    /// - **New plan** (in-flight migration): full §6.1 validation re-runs;
    ///   the graph restarts at its entry with the base checkpoint's CONTEXT
    ///   as the input.
    ///
    /// v1 refuses forking a checkpoint that parked mid-superstep (pending
    /// asks address journal keys of the parent) — pick a closed superstep.
    ///
    /// Returns the seed checkpoint hash; `resume` continues the fork.
    pub fn fork(
        &self,
        base_run_id: &str,
        at_superstep: Option<u64>,
        new_run_id: &str,
        new_plan: Option<&Hash>,
        opts: &RunOptions,
    ) -> Result<Hash, RunError> {
        self.check_run_verb(areev_core::authz::Verb::RunExecute)?;
        areev_core::types::validate_run_id(new_run_id).map_err(err_run)?;
        if self.load_manifest(new_run_id).is_ok() {
            return Err(RunError::Tainted {
                why: format!("run '{new_run_id}' already exists — fork ids must be fresh"),
            });
        }
        let base_manifest = self.load_manifest(base_run_id)?;
        let view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, base_run_id))
            .map_err(err_run)?;
        let base_ckpt = match at_superstep {
            Some(ss) => {
                // A terminal checkpoint shares its superstep with that
                // superstep's close — `--at N` means the CLOSED state, so
                // prefer the last Idle checkpoint there (the terminal one
                // still wins when it is all there is, e.g. BudgetExhausted).
                let matching: Vec<_> =
                    view.checkpoints.iter().filter(|c| c.superstep == ss).collect();
                matching
                    .iter()
                    .rev()
                    .find(|c| c.scheduler.get("phase") == Some(&json!("Idle")))
                    .or(matching.last())
                    .copied()
                    .ok_or_else(|| RunError::ManifestMismatch {
                        why: format!(
                            "run '{base_run_id}' has no checkpoint at superstep {ss}"
                        ),
                    })?
            }
            None => view.checkpoints.last().ok_or_else(|| RunError::ManifestMismatch {
                why: format!("run '{base_run_id}' has no checkpoint to fork from"),
            })?,
        };
        let mut st: SchedulerState = serde_json::from_value(base_ckpt.scheduler.clone())
            .map_err(|e| RunError::ManifestMismatch { why: format!("checkpoint state: {e}") })?;
        if !st.pending_asks.is_empty()
            || matches!(st.phase, areev_run_core::Phase::Open { .. })
        {
            return Err(RunError::Tainted {
                why: format!(
                    "checkpoint at superstep {} parked mid-superstep — v1 forks \
                     only from closed supersteps (respond or pick another)",
                    base_ckpt.superstep
                ),
            });
        }
        match st.outcome() {
            Some(RunOutcome::BudgetExhausted { .. }) => {
                // The sanctioned continue-under-raised-budgets path.
                st.phase = areev_run_core::Phase::Idle;
                st.exhausted = None;
            }
            Some(other) => {
                return Err(RunError::Tainted {
                    why: format!(
                        "base checkpoint is terminal ({other:?}) — time-travel \
                         forks pick an earlier superstep (--at N)"
                    ),
                });
            }
            None => {}
        }

        let fork_base = crate::manifest::ForkBase {
            base_run: base_run_id.to_string(),
            base_checkpoint: base_ckpt.hash.to_hex(),
            base_superstep: base_ckpt.superstep,
        };
        let mut manifest = match new_plan {
            Some(h) => {
                // In-flight migration: validate the new plan in full, then
                // restart its graph with the inherited context as input.
                let plan = self.load_plan(h)?;
                let context = st.context.clone();
                let llm_window = self.llm.as_ref().and_then(|l| l.context_window());
                let manifest = self.facade.with_store(|m| {
                    RunManifest::resolve(
                        m,
                        &self.ns,
                        new_run_id,
                        h,
                        &plan,
                        &self.principal,
                        opts.budgets,
                        opts.ask_ttl_sec,
                        context.clone(),
                        self.llm.is_some(),
                    )
                    .map(|m| m.with_limits(opts, llm_window))
                })?;
                let manifest = manifest
                    .with_llm_pin(self.llm_pin())
                    .with_engine_pin()
                    .with_attribution(opts)
                    .with_decider_pin(self.decider_pin())?;
                self.check_read_grants(&manifest)?;
                let mut fresh = SchedulerState::new(new_run_id, &plan);
                // The Start bootstrap, applied here so the seed checkpoint
                // is immediately resumable (§ Start: entry activates).
                fresh.context = context;
                fresh.clock_ms = st.clock_ms;
                fresh.node_state[0] = areev_run_core::NodeState::Ready;
                st = fresh;
                manifest
            }
            None => {
                // Resolution freeze: the fork inherits the base's pins
                // verbatim — never re-resolves. Budgets/TTL are the fork's
                // own knobs.
                st.run_id = new_run_id.to_string();
                st.inputs_seen = 0;
                RunManifest {
                    run_id: new_run_id.to_string(),
                    // The FORKER triggers this run — separation of duties
                    // must compare against them, not the base run's starter.
                    principal: self.principal.clone(),
                    budgets: opts.budgets,
                    ask_ttl_sec: opts.ask_ttl_sec,
                    // Same reasoning for the INITIATOR (#293): the `..base`
                    // spread would otherwise inherit whoever the base run was
                    // started on behalf of, and a fork is a new decision by a
                    // new person. It is the fork's own knob, and empty when
                    // the forker did not set one.
                    initiator: opts.initiator.clone(),
                    // A fork is the SANCTIONED way to move a run to a new
                    // model or a new engine (#287, #288) — so it pins what it
                    // is actually running under, not what the base ran under.
                    llm: self.llm_pin(),
                    engine: Some(crate::manifest::EnginePin::current()),
                    // Likewise the decision backend (C1–C3): the fork decides
                    // under what it is running with.
                    decider: self.decider_pin(),
                    ..base_manifest
                }
            }
        };
        // RUN-E030 for a fork of a plan with a decision node, on a host that
        // cannot answer it — before the fork is written.
        let pin = manifest.decider.clone();
        manifest = manifest.with_decider_pin(pin)?;
        manifest.fork_of = Some(fork_base.clone());
        self.facade
            .with_store(|m| manifest.persist_in_namespace(m, &self.ns))
            .map_err(err_run)?;

        // The fork index Fact: `runs_touching`-style joins report forks for
        // inherited-prefix grains through this.
        let now = self.clock.now_ms();
        let mut fact = Fact::new(
            &format!("run:{new_run_id}"),
            "mg:fork_of",
            &format!("{}@{}", fork_base.base_run, fork_base.base_checkpoint),
        )
        .namespace(HARNESS_NS)
        .created_at(now as i64);
        fact.common.author_did = Some(self.principal.clone());
        fact.common.extra_fields.insert("run_id".into(), json!(new_run_id));
        fact.common
            .extra_fields
            .insert("base_run_id".into(), json!(fork_base.base_run));
        self.facade.with_store(|m| m.add(&fact)).map_err(err_run)?;

        // The seed checkpoint, derived_from the BASE checkpoint (§5.4's
        // hash chain) — what `resume` continues from.
        let state_json = serde_json::to_value(&st).unwrap_or(Value::Null);
        let record = DecisionRecord {
            superstep: st.superstep,
            clock_open_ms: st.clock_ms,
            clock_close_ms: st.clock_ms,
            ..Default::default()
        };
        let base_hash = base_ckpt.hash;
        self.facade
            .with_store(|m| {
                journal::write_checkpoint(
                    m,
                    &self.ns,
                    new_run_id,
                    &state_json,
                    &record,
                    st.clock_ms,
                    Some(&base_hash),
                    &self.principal,
                )
            })
            .map_err(err_run)
    }

    /// Answer a Client ask (§6.6's validation order). Responding and
    /// resuming are separate acts: this writes the journaled result and
    /// returns; `resume` spends the compute.
    pub fn respond(
        &self,
        run_id: &str,
        tool_call_id: &str,
        result: Value,
        is_error: bool,
        responder: &str,
    ) -> Result<(), RunError> {
        let manifest = self.load_manifest(run_id)?;
        let plan_hash = Hash::from_hex(&manifest.plan_hash)
            .map_err(|_| RunError::ManifestMismatch { why: "bad plan hash".into() })?;
        let view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, run_id))
            .map_err(err_run)?;
        let Some(last) = view.checkpoints.last() else {
            return Err(RunError::ManifestMismatch { why: "run has no checkpoint".into() });
        };
        let st: SchedulerState = serde_json::from_value(last.scheduler.clone())
            .map_err(|e| RunError::ManifestMismatch { why: format!("checkpoint state: {e}") })?;

        // Fresh clock reading — respond is its own effect boundary (§6.6):
        // there is no scheduler clock while parked. The reading lands on
        // the result grain, so replay sees it.
        let now = self.clock.now_ms();

        let journal_rejection = |why: &str| {
            // The losing/invalid response is journaled BEFORE the caller is
            // told no — "two officers approved conflicting outcomes seconds
            // apart" is audit evidence, not noise (§6.6).
            let mut obs = Observation::new(responder, "human")
                .subject(&format!("run:{run_id}"))
                .object(&format!("rejected response to ask {tool_call_id}: {why}"))
                .namespace(HARNESS_NS)
                .created_at(now as i64);
            obs.common
                .extra_fields
                .insert("run_id".into(), json!(run_id));
            let _ = self.facade.with_store(|m| m.add(&obs));
        };

        // The `run.respond` grant check runs HERE rather than first (#292):
        // the docs promise every rejection is journaled before the error
        // returns, and a refusal recorded without a `run_id` is not evidence
        // anyone can join to a run. Loading the manifest first costs one read
        // on a path that is already human-paced.
        if let Err(e) = self.check_responder(responder) {
            journal_rejection(&format!(
                "responder lacks run.respond on {}: {e}",
                self.ns
            ));
            return Err(e);
        }
        let Some(pending) = st.pending_asks.get(tool_call_id) else {
            journal_rejection("no such pending ask (settled, expired, or never asked)");
            return Err(RunError::UnknownAsk { tool_call_id: tool_call_id.to_string() });
        };
        let bubbled = matches!(
            manifest.executors().get(pending.node_idx),
            Some(NodeExecutor::Subgraph { .. })
        );
        if !bubbled
            && view
                .entries
                .get(&pending.key)
                .is_some_and(|e| e.result.is_some())
        {
            journal_rejection("already settled — first commit won");
            return Err(RunError::UnknownAsk { tool_call_id: tool_call_id.to_string() });
        }
        if pending
            .ask
            .expires_at_sec
            .is_some_and(|exp| (now / 1000) as i64 > exp)
        {
            journal_rejection("expired");
            return Err(RunError::UnknownAsk { tool_call_id: tool_call_id.to_string() });
        }
        // Approval separation of duties (§6.6, D5): structurally refused,
        // not conventioned — mirrors the loop's self-approval block.
        //
        // The refusal is JOURNALED first (#292). This is the most
        // audit-relevant rejection the runtime makes — "the person who asked
        // for this tried to approve it" — and it was the one the docs
        // promised and the code did not write. For a bubbled ask the record
        // lands on the parent run id, where the judgement is made.
        //
        // `initiator` (#293) closes the other half: a run a service starts on
        // a person's behalf executes under the SERVICE's principal, so
        // without it the person behind the request was invisible to this
        // check and could approve their own work. The field is free-form
        // attribution — a value naming no principal (a trigger occurrence id)
        // simply never matches a responder.
        if pending.ask.approval {
            let self_approval = responder == manifest.principal;
            let initiator_approval = manifest.initiator.as_deref() == Some(responder);
            if self_approval || initiator_approval {
                let who = if self_approval { "principal" } else { "initiator" };
                journal_rejection(&format!(
                    "approval by the triggering {who} '{responder}'"
                ));
                return Err(RunError::Unauthorized {
                    what: format!(
                        "approval ask {tool_call_id} requires a responder other than \
                         the triggering {who} '{responder}'"
                    ),
                });
            }
        }
        // A bubbled ask settles in the child, but it is judged HERE first:
        // the child's manifest names whichever principal happened to drive
        // the subgraph's dispatch, and separation of duties has to hold
        // against the principal who triggered the run the operator answered.
        if bubbled {
            let child = subgraph_run_id(run_id, &pending.key.node, pending.key.attempt);
            return self.respond(&child, tool_call_id, result, is_error, responder);
        }

        let outcome = if is_error {
            let detail = result.to_string();
            EffectOutcome::Failed {
                cause: FailCause::UserAborted,
                journal_bytes: detail.len() as u64,
                detail,
            }
        } else {
            EffectOutcome::Completed {
                journal_bytes: result.to_string().len() as u64,
                result,
                input_tokens: 0,
                output_tokens: 0,
                usd_micros: 0,
            }
        };
        let executor = manifest
            .executors()
            .get(pending.node_idx)
            .cloned()
            .ok_or_else(|| RunError::ManifestMismatch { why: "ask node outside manifest".into() })?;
        let intent = view
            .entries
            .get(&pending.key)
            .map(|e| e.intent)
            .ok_or_else(|| RunError::UnknownAsk { tool_call_id: tool_call_id.to_string() })?;
        self.facade
            .with_store(|m| {
                journal::write_result(
                    m,
                    &self.ns,
                    run_id,
                    &plan_hash,
                    &intent,
                    &pending.key,
                    &executor,
                    &outcome,
                    st.superstep,
                    now,
                    responder,
                )
            })
            .map_err(err_run)?;
        Ok(())
    }

    /// Write the cancel marker (the kill-switch primitive). The drive loop
    /// polls it every iteration; `resume` after cancel drains.
    pub fn cancel(&self, run_id: &str, principal: &str, reason: &str) -> Result<(), RunError> {
        // Deliberately the LOW bar (D5): the brake must never be blocked by
        // missing privilege — RunCancel, not RunExecute.
        self.check_run_verb(areev_core::authz::Verb::RunCancel)?;
        let now = self.clock.now_ms();
        let mut f = Fact::new(&format!("run:{run_id}"), "mg:cancel", reason)
            .namespace(HARNESS_NS)
            .created_at(now as i64);
        f.common.author_did = Some(principal.to_string());
        f.common.extra_fields.insert("run_id".into(), json!(run_id));
        self.facade.with_store(|m| m.add(&f)).map_err(err_run)?;
        // A PAUSED run (#344) has no driver to notice the marker, and it is
        // parked precisely because its host has not decided what happens
        // next — so cancel finalizes it here, rather than leaving a canceled
        // run looking paused until someone happens to resume it. Nothing
        // dispatches: from an Idle checkpoint with the marker set the
        // scheduler finishes `Canceled` before it could open a superstep.
        //
        // Best effort on purpose. The marker is already durable, and every
        // path that can fail here (another driver holding the lease, a store
        // error) is one where a later drive drains the run on that marker
        // anyway; the brake itself must never fail because the tidy-up did.
        if self.is_paused(run_id) {
            let _ = self.resume_run(run_id, &RunOptions::default(), false);
        }
        Ok(())
    }

    /// Ask a live run to PAUSE at its next superstep boundary (#344).
    ///
    /// Writes the pause request — a Fact journaled with the principal and the
    /// reason — and returns. A driver advancing the run honours it where it
    /// would otherwise open the next superstep: the open one finishes and
    /// checkpoints normally, and the run parks there holding no lease and no
    /// concurrency slot. `resume` continues it under the same run id,
    /// manifest and pins, consuming the request; no node re-executes, because
    /// nothing past the checkpoint was ever dispatched.
    ///
    /// The same grant as `resume` (`run.execute`) — stopping a run resumably
    /// is a scheduling decision about the run, unlike cancel's deliberately
    /// low bar, which exists so the brake is never blocked.
    ///
    /// Idempotent: asking again while a request stands (honoured or not)
    /// writes nothing and answers with the standing request. Refused with
    /// `RUN-E029` when the run already finished or a cancel is pending
    /// against it — cancel wins over pause.
    pub fn pause(
        &self,
        run_id: &str,
        principal: &str,
        reason: &str,
    ) -> Result<crate::PauseReceipt, RunError> {
        self.check_run_verb(areev_core::authz::Verb::RunExecute)?;
        // An unknown run id is an error, not a request filed for later: a
        // pause can only mean a run that exists.
        self.load_manifest(run_id)?;
        let view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, run_id))
            .map_err(err_run)?;
        if let Some(last) = view.checkpoints.last() {
            let st: SchedulerState = serde_json::from_value(last.scheduler.clone())
                .map_err(|e| RunError::ManifestMismatch { why: format!("checkpoint state: {e}") })?;
            if let Some(outcome) = st.outcome() {
                return Err(RunError::NotPausable {
                    run_id: run_id.to_string(),
                    why: format!("it already finished ({})", outcome_label(outcome)),
                });
            }
        }
        if let Some((by, why)) = self.cancel_marker(run_id) {
            return Err(RunError::NotPausable {
                run_id: run_id.to_string(),
                why: format!(
                    "a cancel by {by} is pending against it ({why}) — cancel wins over \
                     pause, and the run drains to canceled"
                ),
            });
        }
        if let Some(standing) = view.pause.active.clone() {
            let status = if view.pause.is_paused() { "paused" } else { "requested" };
            return Ok(crate::PauseReceipt {
                run_id: run_id.to_string(),
                status: status.into(),
                already: true,
                request: standing,
            });
        }
        let now = self.clock.now_ms();
        let ordinal = view.pause.requests;
        let h = self
            .facade
            .with_store(|m| {
                journal::write_pause_request(m, &self.ns, run_id, reason, ordinal, now, principal)
            })
            .map_err(err_run)?;
        Ok(crate::PauseReceipt {
            run_id: run_id.to_string(),
            status: "requested".into(),
            already: false,
            request: journal::PauseRequest {
                request: h.to_hex(),
                paused_by: principal.to_string(),
                because: reason.to_string(),
                requested_at: now as i64,
            },
        })
    }

    /// Parked on a honoured, unconsumed pause request at a between-supersteps
    /// checkpoint — nobody is driving it.
    fn is_paused(&self, run_id: &str) -> bool {
        let Ok(view) = self.facade.with_store(|m| journal::load(m, &self.ns, run_id)) else {
            return false;
        };
        view.pause.is_paused()
            && view
                .checkpoints
                .last()
                .is_some_and(|c| c.scheduler.get("phase") == Some(&json!("Idle")))
    }

    /// Queue a steering message for a running run. It is consumed by the
    /// nodes the NEXT superstep dispatches, arriving in their input under
    /// `$inbox` — an in-band channel, so a chat-style plan does not have to
    /// misuse a human-gate ask to receive one.
    pub fn input(&self, run_id: &str, message: &str, principal: &str) -> Result<(), RunError> {
        self.check_run_verb(areev_core::authz::Verb::RunExecute)?;
        let now = self.clock.now_ms();
        self.facade
            .with_store(|m| journal::write_input(m, &self.ns, run_id, message, now, principal))
            .map_err(err_run)?;
        Ok(())
    }

    fn cancel_marker(&self, run_id: &str) -> Option<(String, String)> {
        // A cancel on ANY ancestor stops the whole subtree: child run ids
        // are `{parent}~{16-hex}` by construction, so walking suffix-strips
        // covers arbitrarily nested subgraphs (the <5-minute kill-switch
        // clause must not be unbounded under nesting).
        let mut candidate = run_id.to_string();
        loop {
            let hit = self
                .facade
                .with_store(|m| m.latest(HARNESS_NS, &format!("run:{candidate}"), "mg:cancel"))
                .ok()
                .flatten()
                .map(|g| {
                    (
                        g.get_str("author_did").unwrap_or("unknown").to_string(),
                        g.get_str("object").unwrap_or("").to_string(),
                    )
                });
            if hit.is_some() {
                return hit;
            }
            match candidate.rfind('~') {
                Some(pos) if candidate.len() - pos == 17 => candidate.truncate(pos),
                _ => return None,
            }
        }
    }

    fn child_asks(&self, child_run_id: &str) -> Result<Vec<Ask>, RunError> {
        let view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, child_run_id))
            .map_err(err_run)?;
        let Some(last) = view.checkpoints.last() else {
            return Ok(Vec::new());
        };
        let st: SchedulerState = serde_json::from_value(last.scheduler.clone())
            .map_err(|e| RunError::ManifestMismatch { why: format!("child state: {e}") })?;
        Ok(st
            .pending_asks
            .into_values()
            .filter(|p| view.entries.get(&p.key).is_none_or(|e| e.result.is_none()))
            .map(|p| p.ask)
            .collect())
    }

    /// Execute a subgraph node as a CHILD RUN: own run id (deterministic —
    /// derived from the parent and the node, so replays and permutations
    /// agree and a parked child resumes rather than restarting), own
    /// journal, linked by content on the parent's effect grains. The child's
    /// final context becomes the node's result, merging into the parent
    /// state through the reducers like any other node result; a child parked
    /// on a Client ask completes with a `$parked_asks` result instead, which
    /// parks the parent on the same asks.
    fn run_subgraph_effect(
        &self,
        parent_run_id: &str,
        key: &JournalKey,
        workflow_hash: &str,
        input: &Value,
        opts: &RunOptions,
    ) -> EffectOutcome {
        let child_run_id = subgraph_run_id(parent_run_id, &key.node, key.attempt);
        let fail = |cause: FailCause, detail: String| EffectOutcome::Failed {
            journal_bytes: detail.len() as u64,
            cause,
            detail,
        };
        let child_session = if self
            .facade
            .with_store(|m| RunManifest::load(m, &child_run_id))
            .is_ok()
        {
            self.resume(&child_run_id, opts)
        } else {
            let Ok(h) = Hash::from_hex(workflow_hash) else {
                return fail(FailCause::Unknown, "bad subgraph plan hash".into());
            };
            self.start(&h, &child_run_id, input.clone(), opts)
        };
        match child_session {
            Ok(RunSession::Finished { outcome: RunOutcome::Completed, .. }) => {
                // The child's final context is the node's contribution —
                // minus the bubble marker, which only THIS function may
                // mint. A tool inside the child could otherwise write
                // `$parked_asks` into the child's state and park the parent
                // on asks it invented, putting a forged approval in front of
                // a human.
                let mut context = self
                    .facade
                    .with_store(|m| journal::load(m, &self.ns, &child_run_id))
                    .ok()
                    .and_then(|v| v.checkpoints.last().cloned())
                    .and_then(|c| c.scheduler.get("context").cloned())
                    .unwrap_or(Value::Null);
                if let Value::Object(o) = &mut context {
                    o.remove(PARKED_ASKS);
                }
                EffectOutcome::Completed {
                    journal_bytes: context.to_string().len() as u64,
                    result: context,
                    input_tokens: 0,
                    output_tokens: 0,
                    usd_micros: 0,
                }
            }
            Ok(RunSession::Finished { outcome, .. }) => fail(
                FailCause::Unknown,
                format!("subgraph run '{child_run_id}' finished {outcome:?}"),
            ),
            Ok(RunSession::Parked { .. }) => match self.child_asks(&child_run_id) {
                Ok(asks) if !asks.is_empty() => {
                    let result = json!({ PARKED_ASKS: asks });
                    EffectOutcome::Completed {
                        journal_bytes: result.to_string().len() as u64,
                        result,
                        input_tokens: 0,
                        output_tokens: 0,
                        usd_micros: 0,
                    }
                }
                Ok(_) => fail(
                    FailCause::Unknown,
                    format!("subgraph run '{child_run_id}' parked with no ask to answer"),
                ),
                Err(e) => fail(FailCause::Unknown, format!("subgraph asks: {e}")),
            },
            Err(e) => fail(FailCause::Unknown, format!("subgraph: {e}")),
        }
    }

    /// The shared drive loop. One clock reading per iteration; journal hits
    /// replay without executing; everything else goes to the pool.
    #[allow(clippy::too_many_arguments)]
    #[allow(clippy::too_many_arguments)]
    fn drive(
        &self,
        plan: &PlanGraph,
        plan_hash: &Hash,
        manifest: &RunManifest,
        st: SchedulerState,
        view: JournalView,
        initial_events: Vec<EventIn>,
        opts: &RunOptions,
        // Slots the caller already claimed (#296). `start` claims BEFORE it
        // persists the manifest, so a run refused at the cap leaves nothing
        // behind at all; `resume` and `fork` let this claim them.
        pre_claimed: Option<crate::lease::RunSlots>,
    ) -> Result<RunSession, RunError> {
        let executors = manifest.executors();
        let arg_schemas = self.load_arg_schemas(manifest)?;
        let validate_args = make_validate_args(&arg_schemas);
        let reduce = crate::reducers::make_reduce(&manifest.reducers);
        let decide_ctx = crate::shadow::decide_context_for(&self.facade, manifest)?;
        let env = StepEnv {
            plan,
            executors: &executors,
            budgets: manifest.budgets.to_budgets(),
            reduce: &reduce,
            eval_cond: &builtin_eval,
            ask_ttl_sec: manifest.ask_ttl_sec,
            validate_args: &validate_args,
            llm_reserve_tokens: manifest.llm_reserve_tokens(),
            max_effects_per_attempt: manifest.max_effects_per_attempt(),
            llm_tool_result_chars: manifest.llm_tool_result_chars,
            llm_context_tokens: manifest.llm_context_tokens,
            decide: decide_ctx.env(manifest),
        };
        let run_id = st.run_id.clone();

        // Take the run lease before advancing anything. Two drivers on one run
        // used to last-write-wins in the journal, silently — the `Tainted` doc
        // comment claimed forked tips were detected, but nothing detected them.
        // This prevents the case instead of noticing it afterwards. On the
        // embedded backend one memory is one writer, so this always succeeds
        // and costs a single meta row.
        // Bind the run's principal to the broker before any effect fires, so
        // an owned credential (`--credential name=VAR@principal`) is refused
        // for a run executing as anyone else (#101). Every entry — start,
        // resume, fork — funnels through here, which is why it lands here and
        // not at each caller.
        self.executor.bind_run_principal(&self.principal);
        self.executor.bind_artifact_store(Arc::clone(&self.facade));

        // #300: the holder carries the HOST, not just the pid. Two
        // containers both running as PID 1 under one service principal used
        // to produce the same holder string, and `acquire` re-enters an equal
        // holder by design — so the second pod took a LIVE lease and both
        // drivers advanced one run.
        let holder = crate::lease::holder_for(&self.principal, opts.node.as_deref());
        let lease_ms = match opts.lease_ms {
            None => crate::lease::DEFAULT_RUN_LEASE_MS,
            Some(ms) if ms < crate::lease::MIN_RUN_LEASE_MS => {
                return Err(RunError::InvalidPlan {
                    why: format!(
                        "--lease is {}s, below the {}s floor — under it, ordinary \
                         scheduling jitter between renewals looks like a dead driver, \
                         and a takeover mid-flight is the failure the lease exists to \
                         prevent",
                        ms / 1000,
                        crate::lease::MIN_RUN_LEASE_MS / 1000
                    ),
                })
            }
            Some(ms) => ms,
        };
        // #296: a concurrency slot, claimed BEFORE anything is written, so a
        // run refused at the cap leaves nothing behind and its id stays free.
        let mut slots = match pre_claimed {
            Some(s) => s,
            None => crate::lease::RunSlots::claim(
                &self.facade,
                &run_id,
                &self.principal,
                self.clock.now_ms() as i64,
                lease_ms,
                opts.max_concurrent_runs,
                opts.max_concurrent_runs_per_principal,
            )?,
        };
        let mut lease = match crate::lease::RunLease::acquire(
            &self.facade,
            &run_id,
            &holder,
            self.clock.now_ms() as i64,
            lease_ms,
        ) {
            Ok(l) => l,
            Err(e) => {
                slots.release(&self.facade);
                return Err(e);
            }
        };

        // §6.10: the event bus wraps the observer; `emit` never blocks.
        // Arc'd because LLM jobs carry a TokenChunk sink into the pool.
        let bus = self
            .observer
            .as_ref()
            .map(|o| {
                Arc::new(crate::stream::EventBus::new(
                    Arc::clone(o),
                    crate::stream::EVENT_BUFFER,
                ))
            });
        let emit = |ev: crate::stream::RunEvent| {
            if let Some(b) = &bus {
                b.emit(ev);
            }
        };
        emit(if view.checkpoints.is_empty() {
            crate::stream::RunEvent::RunStarted { run_id: run_id.clone() }
        } else {
            crate::stream::RunEvent::RunResumed { run_id: run_id.clone() }
        });
        let pool = Pool::new(
            Arc::clone(&self.executor),
            self.llm.clone(),
            opts.workers.max(1),
        );
        let mut intents: BTreeMap<JournalKey, Hash> =
            view.entries.iter().map(|(k, e)| (k.clone(), e.intent)).collect();
        // Which dispatched effects are SUMMARIZER turns. Read off the
        // journaled `input.fold` at prepare time (the same marker that decides
        // the turn is offered no tools) and remembered so the settle path can
        // record the summary it produced.
        let mut fold_keys: std::collections::BTreeSet<JournalKey> = Default::default();
        let mut results: BTreeMap<JournalKey, EffectOutcome> = view
            .entries
            .iter()
            .filter_map(|(k, e)| e.result.as_ref().map(|(_, o)| (k.clone(), o.clone())))
            .collect();
        // What a model produced, for the egress round trip (#350).
        let mut model_tokens = ModelTokens::default();
        let model_origin = |k: &JournalKey| {
            k.kind == areev_run_core::EffectKind::Llm
                || plan
                    .nodes
                    .iter()
                    .position(|n| *n == k.node)
                    .and_then(|i| executors.get(i))
                    .is_some_and(|e| {
                        matches!(e, areev_run_core::NodeExecutor::Subgraph { .. })
                    })
        };
        let mut prev_ckpt: Option<Hash> = view.checkpoints.last().map(|c| c.hash);
        let mut input_cursor = view.cursor;
        // The host-pause state (#344): the standing request, if any, polled
        // forward on the same cursor as steering messages.
        let mut pause = view.pause.clone();
        // Steering messages the journal already holds but this state has not
        // applied. They wait in `steer` with everything polled later: an
        // input is fed ONLY while a superstep is open, which is what makes it
        // inert until the next one and lets replay place it from the
        // checkpoint's `inputs_seen` alone.
        let mut steer: Vec<Value> =
            view.inputs.iter().skip(st.inputs_seen as usize).cloned().collect();
        let mut events = initial_events;
        // Distinct refusals already journaled for this drive, so a retry loop
        // against one blocked host records one fact rather than many.
        let mut refusals_seen: std::collections::BTreeSet<(String, String, String)> =
            Default::default();
        // How many brokered CALLS have already been journaled. A cursor, not a
        // set: a refusal is a policy fact and forty attempts are one of them,
        // but a successful call is an effect and forty of those are forty
        // things that happened, so they are journaled in order and never
        // deduplicated (#101).
        //
        // Seeded with what the broker ALREADY holds, not 0: the trigger
        // multi-fire path reuses one broker across sequential runs and
        // `calls()` accumulates for the broker's whole life, never draining, so
        // starting the cursor at 0 would re-journal an earlier run's calls
        // under THIS run's id, principal and clock — duplicate, misattributed
        // audit grains. Runs drive sequentially, so whatever is already present
        // belongs to an earlier one and is not ours to journal.
        let mut calls_journaled: usize = self.executor.calls().len();
        let mut blob_reads_journaled: usize = self.executor.blob_reads().len();
        let mut in_flight: usize = 0;
        let mut st = st;
        let mut intents_written: u32 = 0;
        let mut results_seen: u32 = 0;

        // Crash-window danglings (§5.3 step 4): fail fast when the host
        // cannot honor idempotency keys; otherwise the re-emitted intents
        // below adopt these and re-dispatch under the same key.
        if opts.on_dangling == OnDangling::Fail {
            if let Some(last) = view.checkpoints.last() {
                if let Some(e) = view
                    .dangling(last.superstep)
                    .into_iter()
                    .find(|e| !st.pending_asks.values().any(|p| p.key == e.key))
                {
                    return Err(RunError::DanglingIntent {
                        key: e.key.tool_call_id(),
                    });
                }
            }
        }

        loop {
            // The kill switch: polled every iteration (a µs point read).
            if st.cancel.is_none() {
                if let Some((by, reason)) = self.cancel_marker(&run_id) {
                    // `step()`'s contract: any call that may CLOSE a
                    // superstep carries a fresh ClockReading (doc comment on
                    // `step()`). A freshly-seen cancel closes the run's
                    // FINAL superstep, but without a reading taken HERE
                    // `st.clock_ms` stays whatever an earlier, already-
                    // completed wave last set it to — which can predate the
                    // cancel Fact's own `created_at`, journaling a
                    // clock_close_ms that appears to close before the
                    // operator asked to cancel. `cancel_marker` only
                    // returns once its read has observed the write `cancel`
                    // performed, so a reading taken now is monotone at or
                    // after it. `verify()`'s replay already assumes this
                    // shape at both its own cancel_peek sites (a
                    // ClockReading immediately preceding CancelSeen in the
                    // same batch) — this is the live counterpart.
                    events.push(EventIn::ClockReading { unix_ms: self.clock.now_ms() });
                    events.push(EventIn::CancelSeen { principal: by, reason });
                }
            }
            // Steering, polled at the same wave boundary as the cancel
            // marker. A failed read is not fatal: the cursor stays put, so
            // the next wave picks the same messages up rather than losing
            // them to a transient store error.
            if !st.is_terminal() {
                if let Ok((queued, pauses, at)) = self
                    .facade
                    .with_store(|m| journal::poll_run(m, &self.ns, &run_id, input_cursor))
                {
                    input_cursor = at;
                    steer.extend(queued);
                    for ev in &pauses {
                        pause.apply(ev);
                    }
                }
                if !steer.is_empty() && matches!(st.phase, areev_run_core::Phase::Open { .. }) {
                    events.extend(
                        steer.drain(..).map(|message| EventIn::InputSeen { message }),
                    );
                }
                // A standing pause rides EVERY batch until the run parks on
                // it: the scheduler holds nothing between calls, so the call
                // that finally closes into Idle has to carry it too.
                if pause.active.is_some() {
                    events.push(EventIn::PauseRequested);
                }
            }

            let out = step(&env, st, &events);
            st = out.state;
            events = Vec::new();

            let mut parked_envelope: Option<Value> = None;
            let mut finished: Option<RunOutcome> = None;
            // This iteration's dispatch wave, in COMMAND ORDER: every
            // Dispatch lands here — journal-answered replays as
            // Some(outcome), fresh pool submissions as None — and is fed
            // back to step() as one batch at the wave boundary below.
            let mut wave: Vec<(JournalKey, Option<EffectOutcome>)> = Vec::new();

            for cmd in out.commands {
                match cmd {
                    Command::WriteIntent { key, executor, input, superstep, clock_ms } => {
                        if let Some(existing) = intents.get(&key).copied() {
                            // Adopted: a crash re-opened this superstep. A
                            // REDELIVERY Observation is recorded only when
                            // the effect will actually redeliver (no result
                            // journaled) — a journal-answered replay is not
                            // a redelivery and must not fabricate one.
                            if results.contains_key(&key) {
                                continue;
                            }
                            let mut obs = Observation::new(&self.principal, "system")
                                .subject(&format!("run:{run_id}"))
                                .object(&format!(
                                    "redelivery of intent {} for {}",
                                    existing.to_hex(),
                                    key.tool_call_id()
                                ))
                                .namespace(HARNESS_NS)
                                .created_at(clock_ms as i64);
                            obs.common
                                .extra_fields
                                .insert("run_id".into(), json!(run_id));
                            self.facade
                                .with_store(|m| m.add_if_novel(&obs))
                                .map_err(err_run)?;
                            continue;
                        }
                        let h = self
                            .facade
                            .with_store(|m| {
                                journal::write_intent(
                                    m,
                                    &self.ns,
                                    &run_id,
                                    plan_hash,
                                    &key,
                                    &executor,
                                    &input,
                                    superstep,
                                    clock_ms,
                                    &self.principal,
                                )
                            })
                            .map_err(err_run)?;
                        // The GenAI enrichment is assembled HERE and nowhere
                        // else: the §6.10 observer thread holds no store
                        // handle (one memory, one writer, and the driver has
                        // it), so a span attribute that does not ride the
                        // event can never be recovered.
                        let node_idx = plan.nodes.iter().position(|n| *n == key.node);
                        // A tool the MODEL called: its pending entry is booked
                        // by the same step() pass that emitted this WriteIntent
                        // (step.rs inserts before pushing the command), so the
                        // model's own call id is readable now.
                        let flow = node_idx.map(|i| flow_key(i, &key.task_path));
                        let pending = flow
                            .as_ref()
                            .and_then(|f| st.abstract_flows.get(f))
                            .and_then(|f| f.pending_tools.get(&key.effect_seq));
                        let in_agent =
                            flow.as_ref().is_some_and(|f| st.abstract_flows.contains_key(f));
                        let is_llm = key.kind == areev_run_core::EffectKind::Llm;
                        emit(crate::stream::RunEvent::NodeDispatched {
                            superstep,
                            node: key.node.clone(),
                            task_path: key.task_path.clone(),
                            attempt: key.attempt,
                            effect_seq: key.effect_seq,
                            effect_kind: Some(
                                match key.kind {
                                    areev_run_core::EffectKind::Llm => "llm",
                                    areev_run_core::EffectKind::Tool => "tool",
                                }
                                .into(),
                            ),
                            executor_kind: Some(executor_kind_of(&executor).into()),
                            agent_name: in_agent.then(|| key.node.clone()),
                            tool_name: match &executor {
                                areev_run_core::NodeExecutor::Host { tool_name, .. }
                                | areev_run_core::NodeExecutor::Client { tool_name, .. }
                                | areev_run_core::NodeExecutor::Decide { tool_name, .. } => {
                                    Some(tool_name.clone())
                                }
                                _ => None,
                            },
                            tool_call_id: pending.map(|p| p.model_call_id.clone()),
                            // The model and its ceiling describe the REQUEST,
                            // so they ride the model turn only — a flow tool
                            // carries the provider (it is that agent's tool
                            // call) but no request model of its own.
                            model: is_llm
                                .then(|| self.llm.as_ref().map(|l| l.model().to_string()))
                                .flatten(),
                            provider: in_agent
                                .then(|| self.llm.as_ref().map(|l| l.provider().to_string()))
                                .flatten(),
                            // What THIS request carries (#391): the node's
                            // own ceiling and temperature when its block sets
                            // them, nothing for an omitted temperature — a span
                            // must not claim a parameter that never went out.
                            max_tokens: is_llm
                                .then(|| TurnRequest::from_input(&input, manifest).max_tokens),
                            temperature: if is_llm {
                                TurnRequest::from_input(&input, manifest).temperature.and_then(
                                    |t| self.llm.as_ref().map_or(Some(t), |l| l.effective_temperature(t)),
                                )
                            } else {
                                None
                            },
                        });
                        intents.insert(key, h);
                        intents_written += 1;
                        if opts.inject_crash == Some(CrashPoint::AfterIntent(intents_written)) {
                            return Err(RunError::Storage {
                                detail: "injected crash after intent write".into(),
                            });
                        }
                    }
                    Command::Dispatch { key, executor, input } => {
                        if let Some(outcome) = results.get(&key) {
                            // Replay: the journal answers; nothing executes.
                            // Resolved WITH the wave (not immediately), so a
                            // mixed replay+fresh superstep feeds one batch in
                            // dispatch order — the cadence verify replays.
                            wave.push((key, Some(outcome.clone())));
                            continue;
                        }
                        // Subgraphs and declared memory reads run INLINE on
                        // the driver thread — both need the store, which pool
                        // workers never touch. Parallel siblings therefore
                        // serialize (documented v1 bound). A read is the
                        // runtime's own act (#255): no tool, no subprocess,
                        // no handle leaves this thread.
                        let inline = match &executor {
                            areev_run_core::NodeExecutor::Subgraph { workflow_hash } => Some((
                                self.run_subgraph_effect(
                                    &run_id,
                                    &key,
                                    workflow_hash,
                                    &input,
                                    opts,
                                ),
                                None,
                            )),
                            areev_run_core::NodeExecutor::MemoryRead { spec, .. } => {
                                let read =
                                    crate::memread::execute(&self.facade, &self.ns, spec, &input);
                                Some((read.outcome, read.record))
                            }
                            // A decision (C1–C3) is the host's backend
                            // answering, on this thread — never the pool,
                            // never `--tool-cmd`. Its result grain is the
                            // replay answer, so verify never re-asks.
                            areev_run_core::NodeExecutor::Decide { .. } => Some((
                                self.run_decide(&key, &executor, &input, manifest, &arg_schemas),
                                None,
                            )),
                            _ => None,
                        };
                        if let Some((outcome, record)) = inline {
                            let intent = intents.get(&key).copied().ok_or_else(|| {
                                RunError::Storage {
                                    detail: format!(
                                        "{} dispatch without intent",
                                        executor_kind_of(&executor)
                                    ),
                                }
                            })?;
                            let now = self.clock.now_ms();
                            self.facade
                                .with_store(|m| {
                                    journal::write_result_with_read(
                                        m,
                                        &self.ns,
                                        &run_id,
                                        plan_hash,
                                        &intent,
                                        &key,
                                        &executor,
                                        &outcome,
                                        record.as_ref(),
                                        st.superstep,
                                        now,
                                        &self.principal,
                                    )
                                })
                                .map_err(err_run)?;
                            let (in_tok, out_tok, finish, err_type) =
                                settled_detail(key.kind, &outcome);
                            emit(crate::stream::RunEvent::EffectSettled {
                                superstep: st.superstep,
                                node: key.node.clone(),
                                task_path: key.task_path.clone(),
                                attempt: key.attempt,
                                effect_seq: key.effect_seq,
                                ok: matches!(outcome, EffectOutcome::Completed { .. }),
                                input_tokens: in_tok,
                                output_tokens: out_tok,
                                finish_reason: finish,
                                error_type: err_type,
                            });
                            results.insert(key.clone(), outcome.clone());
                            wave.push((key, Some(outcome)));
                            continue;
                        }
                        // LLM effects: prepare the request ON THIS THREAD
                        // (transcript translation + pinned Definitions from
                        // the store), then hand the pure job to the pool.
                        let llm_prepared = if key.kind == areev_run_core::EffectKind::Llm {
                            let areev_run_core::NodeExecutor::Abstract { tools, .. } = &executor
                            else {
                                return Err(RunError::Storage {
                                    detail: "llm effect on a non-abstract node".into(),
                                });
                            };
                            // A SUMMARIZER turn is offered no tools. Keyed on
                            // the journaled `fold` marker in the effect's own
                            // input, so verify makes the identical choice from
                            // the journal alone — reading it off scheduler
                            // state would not replay.
                            let folding = input.get("fold").is_some();
                            if folding {
                                fold_keys.insert(key.clone());
                            }
                            // C1: a narrowed offer rides the turn's own
                            // journaled input as `offer.tools` — the same
                            // journal-keyed rule, so verify offers the same
                            // set. Narrowing only removes from the pinned
                            // universe.
                            let narrowed: Option<Vec<areev_run_core::OfferedTool>> = input
                                .get("offer")
                                .and_then(|o| o.get("tools"))
                                .and_then(Value::as_array)
                                .map(|names| {
                                    tools
                                        .iter()
                                        .filter(|t| {
                                            names.iter().any(|n| n.as_str() == Some(&t.tool_name))
                                        })
                                        .cloned()
                                        .collect()
                                });
                            let offered: &[areev_run_core::OfferedTool] = if folding {
                                &[]
                            } else {
                                narrowed.as_deref().unwrap_or(tools)
                            };
                            let mut defs = Vec::with_capacity(offered.len());
                            for t in offered {
                                let mut def = self.offered_definition(t)?;
                                // A dynamic tool's text came from the run
                                // INPUT — host data, possibly naming a
                                // person — so its description and the prose
                                // annotations of its schema cross the
                                // boundary pseudonymized, as the input does.
                                // Structural strings (types, enums, patterns)
                                // are left alone: the arguments are validated
                                // against them as stored.
                                if t.dynamic.is_some() {
                                    self.pseudonymize_tool_text(&mut def)?;
                                }
                                defs.push(def);
                            }
                            // §6.10 TokenChunk sink: only when observed —
                            // otherwise the plain call (no stream overhead).
                            let on_token = bus.as_ref().map(|b| {
                                let b = Arc::clone(b);
                                let node = key.node.clone();
                                let task_path = key.task_path.clone();
                                let attempt = key.attempt;
                                Box::new(move |t: &str| {
                                    b.emit(crate::stream::RunEvent::TokenChunk {
                                        node: node.clone(),
                                        task_path: task_path.clone(),
                                        attempt,
                                        text: t.to_string(),
                                    });
                                }) as crate::executor::TokenSink
                            });
                            let to_model = self.pseudonymize_for_model(&input)?;
                            // The node's request fields (#391), read off the
                            // turn's own journaled `llm` — what the scheduler
                            // wrote is what is sent, and what verify sees.
                            // Read off `to_model`, never the raw input: the
                            // node's system text crosses the model boundary
                            // exactly as its messages do, so an `egress`
                            // policy pseudonymizes it too (#391 review).
                            // Only the PROSE is taken from the masked copy:
                            // the response schema goes out as declared, since
                            // the answer is checked against it as declared.
                            let mut req = TurnRequest::from_input(&input, manifest);
                            req.system = TurnRequest::from_input(&to_model, manifest).system;
                            Some(crate::executor::PreparedLlm {
                                messages: seam_messages(&to_model),
                                tools: defs,
                                max_tokens: req.max_tokens,
                                on_token,
                                system: req.system,
                                temperature: req.temperature,
                                response_format: req.response_format,
                                parallel_tool_calls: req.parallel_tool_calls,
                                record_sent: input.get("llm").is_some()
                                    || self
                                        .llm
                                        .as_ref()
                                        .is_some_and(|l| l.request_profile_digest().is_some()),
                            })
                        } else {
                            None
                        };
                        // A code-carrying tool: read and hash-verify the blob
                        // HERE, on the driver thread, for the same reason the
                        // LLM branch resolves its Definitions here — the pool
                        // touches no store handle. `get_blob` verifies the
                        // digest on every read, so the bytes handed to the
                        // pool always are the address the manifest pinned.
                        // A model-issued call (inside an abstract node)
                        // reaches the Definition it was offered as, found by
                        // its frozen address — a lookup by NODE finds the
                        // abstract node's own pin, which names no code, and
                        // so used to send a call to a code-carrying
                        // Definition to `--tool-cmd` instead of its blob. A
                        // dynamic tool (#392) is journaled under its own name
                        // but EXECUTES as that Definition: its name is what
                        // the host executor and the broker's grants know.
                        let flow_target = match &executor {
                            areev_run_core::NodeExecutor::Host { tool_hash, tool_name }
                                if key.kind == areev_run_core::EffectKind::Tool
                                    && manifest
                                        .pinned
                                        .iter()
                                        .any(|p| p.node == key.node && p.executor == "abstract") =>
                            {
                                manifest
                                    .host_pin_by_hash(tool_hash)
                                    .map(|p| (p, p.tool_name != *tool_name))
                            }
                            _ => None,
                        };
                        let exec_name = flow_target
                            .filter(|(_, renamed)| *renamed)
                            .map(|(p, _)| p.tool_name.clone());
                        let code_pin = match flow_target {
                            Some((p, _)) => Some(p),
                            None => manifest.pinned.iter().find(|p| p.node == key.node),
                        };
                        let code_prepared = match code_pin.filter(|p| p.executor_uri.is_some()) {
                            Some(pin) => {
                                let uri = pin.executor_uri.as_deref().unwrap();
                                let bytes = self
                                    .facade
                                    .with_store(|m| m.get_blob(uri))
                                    .map_err(err_run)?;
                                Some(crate::executor::PreparedCode {
                                    uri: uri.to_string(),
                                    bytes,
                                    // The manifest-frozen runtime rides with
                                    // the bytes, so a mid-run supersession
                                    // cannot re-route sandbox to native.
                                    runtime: pin.runtime.clone(),
                                    limits: pin.runtime_limits.clone(),
                                    // Frozen at start with the runtime, for
                                    // the same reason: a supersession mid-run
                                    // must not widen what a module may reach.
                                    capabilities: pin.capabilities.clone(),
                                })
                            }
                            None => None,
                        };
                        let idem = idempotency_key(&key, &input);
                        // Rehydrate for the tool, never for the record: the
                        // intent above journaled the pseudonymized input and
                        // `idem` derives from it, so verify replays identically.
                        // An LLM turn is exempt: what it sends is `to_model`,
                        // prepared above from the unrehydrated input.
                        let rehydrated = if key.kind == areev_run_core::EffectKind::Llm {
                            Ok(input)
                        } else {
                            self.rehydrate_from_model(
                                &input,
                                &mut model_tokens,
                                &results,
                                &model_origin,
                            )
                        };
                        let input = match rehydrated {
                            Ok(v) => v,
                            Err(why) => {
                                // Fail the node, not the run: a plan's retries
                                // and edges still decide what happens next.
                                let outcome = EffectOutcome::Failed {
                                    cause: FailCause::ExecutorError,
                                    detail: why,
                                    journal_bytes: 0,
                                };
                                wave.push((key, Some(outcome)));
                                continue;
                            }
                        };
                        wave.push((key.clone(), None));
                        pool.submit(DispatchJob {
                            key,
                            executor,
                            input,
                            idempotency_key: idem,
                            llm: llm_prepared,
                            code: code_prepared,
                            exec_name,
                        });
                        in_flight += 1;
                    }
                    Command::EmitEnvelope { asks } => {
                        for ask in &asks {
                            emit(crate::stream::RunEvent::AskRaised {
                                superstep: st.superstep,
                                node: ask.node.clone(),
                                tool_call_id: ask.tool_call_id.clone(),
                            });
                        }
                        parked_envelope = Some(json!({
                            "kind": "requires_action",
                            "run_id": run_id,
                            "checkpoint": prev_ckpt.map(|h| h.to_hex()),
                            "asks": asks,
                        }));
                    }
                    Command::WriteCheckpoint { decision_record, state_json, .. } => {
                        let clock = decision_record
                            .clock_close_ms
                            .max(decision_record.clock_open_ms);
                        let superstep = decision_record.superstep;
                        // Journaled at each superstep boundary rather than at the
                        // terminal, so a crash loses at most one superstep of
                        // evidence. One Observation per DISTINCT refusal. A tool retrying against the same
                        // blocked host is one audit fact, not forty; the count
                        // of attempts stays in the log line, which is the fast
                        // path a human debugging actually reads.
                        for r in self.executor.refusals() {
                            let key = (r.caller.clone(), r.destination.clone(), r.reason.clone());
                            if !refusals_seen.insert(key) {
                                continue;
                            }
                            self.facade
                                .with_store(|m| {
                                    journal::write_egress_refusal(
                                        m,
                                        &run_id,
                                        &r,
                                        clock,
                                        &self.principal,
                                        manifest.harness_ns.as_deref(),
                                    )
                                })
                                .map_err(err_run)?;
                        }
                        // The mirror: every mediated call a capability tool
                        // actually made, on the same superstep boundary, so a
                        // crash loses at most one superstep of evidence.
                        let calls = self.executor.calls();
                        for c in calls.iter().skip(calls_journaled) {
                            self.facade
                                .with_store(|m| {
                                    journal::write_egress_call(
                                        m,
                                        &run_id,
                                        c,
                                        clock,
                                        &self.principal,
                                        manifest.harness_ns.as_deref(),
                                    )
                                })
                                .map_err(err_run)?;
                        }
                        calls_journaled = calls.len();
                        // And the other mediated door (#106): every stored
                        // byte a capability tool opened. Same boundary, same
                        // skip-cursor, same not-a-journal-entry status — the
                        // tool this exists for parses untrusted attachments,
                        // so "which bytes did it actually read" is evidence an
                        // incident asks for first.
                        let reads = self.executor.blob_reads();
                        for r in reads.iter().skip(blob_reads_journaled) {
                            self.facade
                                .with_store(|m| {
                                    journal::write_blob_read(
                                        m,
                                        &run_id,
                                        r,
                                        clock,
                                        &self.principal,
                                        manifest.harness_ns.as_deref(),
                                    )
                                })
                                .map_err(err_run)?;
                        }
                        blob_reads_journaled = reads.len();
                        let h = self
                            .facade
                            .with_store(|m| {
                                journal::write_checkpoint(
                                    m,
                                    &self.ns,
                                    &run_id,
                                    &state_json,
                                    &decision_record,
                                    clock,
                                    prev_ckpt.as_ref(),
                                    &self.principal,
                                )
                            })
                            .map_err(err_run)?;
                        emit(crate::stream::RunEvent::CheckpointWritten {
                            superstep,
                            hash: h.to_hex(),
                        });
                        prev_ckpt = Some(h);
                        // Renew on the superstep boundary — the point at which
                        // progress became durable, and therefore the point at
                        // which a driver that has lost the run must stop.
                        lease.renew(&self.facade, &run_id, self.clock.now_ms() as i64)?;
                        slots.renew(&self.facade, &run_id, self.clock.now_ms() as i64);
                    }
                    Command::Finish { outcome } => {
                        finished = Some(outcome);
                    }
                }
            }

            if let Some(outcome) = finished {
                // Terminal checkpoint: the Finished state itself.
                let state_json = serde_json::to_value(&st).unwrap_or(Value::Null);
                let record = DecisionRecord {
                    superstep: st.superstep,
                    clock_open_ms: st.clock_ms,
                    clock_close_ms: st.clock_ms,
                    ..Default::default()
                };
                self.facade
                    .with_store(|m| {
                        journal::write_checkpoint(
                            m,
                            &self.ns,
                            &run_id,
                            &state_json,
                            &record,
                            st.clock_ms,
                            prev_ckpt.as_ref(),
                            &self.principal,
                        )
                    })
                    .map_err(err_run)?;
                // The run-outcome record (§8 Wave 4): one compact
                // Observation per terminal run — outcome + the spent
                // figures — the loop's run_outcome analyzer's food and the
                // cost-attribution source. Journaled AFTER the terminal
                // checkpoint; replay never sees it (it is not a journal
                // entry), so verify is unaffected.
                let outcome_label = outcome_label(&outcome);
                let mut obs = Observation::new(&self.principal, "system")
                    .subject(&format!("run:{run_id}"))
                    .object(&outcome_label)
                    .namespace(HARNESS_NS)
                    .created_at(st.clock_ms as i64);
                let ex = &mut obs.common.extra_fields;
                ex.insert("run_id".into(), json!(run_id));
                ex.insert("observation_kind".into(), json!("run_outcome"));
                ex.insert("plan_hash".into(), json!(plan_hash.to_hex()));
                ex.insert("spent_input_tokens".into(), json!(st.spent.input_tokens));
                ex.insert("spent_output_tokens".into(), json!(st.spent.output_tokens));
                ex.insert("spent_usd_micros".into(), json!(st.spent.usd_micros));
                ex.insert("spent_wall_ms".into(), json!(st.spent.wall_ms));
                ex.insert("spent_supersteps".into(), json!(st.spent.supersteps));
                // Context pressure: how often this run had to summarize itself
                // to keep going. Zero on the overwhelming majority of runs, and
                // omitted there so the record stays compact — a workflow that
                // folds every time is telling you its nodes have outgrown the
                // window, which is the loop's to notice.
                if st.folds > 0 {
                    ex.insert("folds".into(), json!(st.folds));
                }
                if let Some(detail) = match &outcome {
                    RunOutcome::Failed { node, detail } => {
                        Some(format!("{node}: {detail}"))
                    }
                    RunOutcome::Stalled { node } => Some(node.clone()),
                    _ => None,
                } {
                    ex.insert("outcome_detail".into(), json!(detail));
                }
                self.facade.with_store(|m| m.add(&obs)).map_err(err_run)?;
                emit(crate::stream::RunEvent::RunFinished {
                    run_id: run_id.clone(),
                    outcome: format!("{outcome:?}"),
                    dropped_events: bus.as_ref().map(|b| b.dropped()).unwrap_or(0),
                });
                // A terminal run needs no lease: releasing lets `verify`,
                // `fork` or an inspect start at once rather than waiting one
                // lease TTL for a run that is already over.
                slots.release(&self.facade);
                lease.release(&self.facade);
                return Ok(RunSession::Finished { outcome, run_id });
            }

            if !wave.is_empty() {
                // The wave boundary: verify replays every superstep as ONE
                // close reading followed by its resolutions in dispatch
                // order, so the live driver must feed exactly that.
                // Trickling completions in arrival batches — each with its
                // own clock reading — makes scheduler state depend on
                // thread timing: an unjournaled decision that diverges
                // replay (and shifts every later scripted-clock value).
                let mut completed: BTreeMap<JournalKey, DispatchDone> = BTreeMap::new();
                while in_flight > 0 {
                    let done = pool
                        .done_rx
                        .recv()
                        .map_err(|_| RunError::Storage { detail: "executor pool died".into() })?;
                    in_flight -= 1;
                    results_seen += 1;
                    if opts.inject_crash == Some(CrashPoint::BeforeResult(results_seen)) {
                        // The effect FIRED (the executor ran); its result
                        // dies with this "process". The §5.3 redelivery
                        // contract owns what happens next.
                        return Err(RunError::Storage {
                            detail: "injected crash before result write".into(),
                        });
                    }
                    completed.insert(done.key.clone(), done);
                    // Renew INSIDE the superstep (#299), after each result.
                    //
                    // This is what makes a short TTL safe, and what the old
                    // fixed ten minutes was hiding: an abstract node's turns
                    // and tool calls all happen inside ONE superstep, so with
                    // a 300 s tool timeout and sixteen effects a perfectly
                    // healthy driver could outlive its own lease and be taken
                    // over mid-flight. The lease is a host-local `meta` row,
                    // so renewing more often leaves `verify` untouched.
                    let now_ms = self.clock.now_ms() as i64;
                    lease.renew(&self.facade, &run_id, now_ms)?;
                    slots.renew(&self.facade, &run_id, now_ms);
                }
                let now = self.clock.now_ms();
                events.push(EventIn::ClockReading { unix_ms: now });
                for (key, replayed) in wave {
                    let outcome = match replayed {
                        Some(outcome) => outcome,
                        None => {
                            let done = completed.remove(&key).ok_or_else(|| {
                                RunError::Storage {
                                    detail: "completion missing for a dispatched effect".into(),
                                }
                            })?;
                            let intent = intents.get(&key).copied().ok_or_else(|| {
                                RunError::Storage { detail: "completion without intent".into() }
                            })?;
                            // Result durability precedes bookkeeping. The
                            // executor is the one the job DISPATCHED under
                            // (a flow tool runs as Host inside an Abstract
                            // node) — the result grain re-states what ran,
                            // matching its intent.
                            self.facade
                                .with_store(|m| {
                                    journal::write_result(
                                        m,
                                        &self.ns,
                                        &run_id,
                                        plan_hash,
                                        &intent,
                                        &key,
                                        &done.executor,
                                        &done.outcome,
                                        st.superstep,
                                        now,
                                        &self.principal,
                                    )
                                })
                                .map_err(err_run)?;
                            // A summarizer turn's answer is the only prose in
                            // the whole run that says what the agent had
                            // worked out. Surfaced as harness EVIDENCE the
                            // loop can read (the journal Tool grain that also
                            // holds it is not reachable by recall), never as
                            // the agent's own memory — see
                            // `journal::write_fold_summary`.
                            if fold_keys.contains(&key) {
                                if let EffectOutcome::Completed { result, .. } = &done.outcome {
                                    let summary = result
                                        .get("text")
                                        .and_then(|t| t.as_str())
                                        .unwrap_or_default()
                                        .trim()
                                        .to_string();
                                    if !summary.is_empty() {
                                        self.facade
                                            .with_store(|m| {
                                                journal::write_fold_summary(
                                                    m,
                                                    &self.ns,
                                                    &run_id,
                                                    &key,
                                                    plan_hash,
                                                    &summary,
                                                    now,
                                                    &self.principal,
                                        manifest.harness_ns.as_deref(),
                                    )
                                            })
                                            .map_err(err_run)?;
                                    }
                                }
                            }
                            let (in_tok, out_tok, finish, err_type) =
                                settled_detail(key.kind, &done.outcome);
                            emit(crate::stream::RunEvent::EffectSettled {
                                superstep: st.superstep,
                                node: key.node.clone(),
                                task_path: key.task_path.clone(),
                                attempt: key.attempt,
                                effect_seq: key.effect_seq,
                                ok: matches!(done.outcome, EffectOutcome::Completed { .. }),
                                input_tokens: in_tok,
                                output_tokens: out_tok,
                                finish_reason: finish,
                                error_type: err_type,
                            });
                            results.insert(key.clone(), done.outcome.clone());
                            done.outcome
                        }
                    };
                    events.push(EventIn::EffectResolved { key, outcome });
                }
                continue;
            }

            if let Some(envelope) = parked_envelope {
                // in_flight is 0 here: every dispatch drains with its wave.
                //
                // Release: a parked run is not a process. It is a journaled
                // checkpoint waiting on a human, so there is no in-memory state
                // to protect and any driver may pick it up — that portability is
                // the property the runtime is built on. Holding the lease across
                // a park would make the next `areev run resume`, in a different
                // process with a different holder id, wait out a full lease TTL
                // for a run nobody is advancing.
                slots.release(&self.facade);
                lease.release(&self.facade);
                return Ok(RunSession::Parked { envelope, run_id });
            }

            // Parked on a host pause (#344): the step that carried the request
            // ended between supersteps. Everything up to here is durable —
            // the close checkpointed, every dispatched effect settled with its
            // wave — so there is nothing in flight to lose.
            if let (Some(req), areev_run_core::Phase::Idle) = (pause.active.clone(), &st.phase) {
                let now = self.clock.now_ms();
                self.facade
                    .with_store(|m| {
                        journal::write_pause_applied(
                            m,
                            &self.ns,
                            &run_id,
                            &req.request,
                            st.superstep,
                            now,
                            &self.principal,
                        )
                    })
                    .map_err(err_run)?;
                emit(crate::stream::RunEvent::RunPaused {
                    run_id: run_id.clone(),
                    superstep: st.superstep,
                    paused_by: req.paused_by.clone(),
                    because: req.because.clone(),
                });
                // Same release as a park on an ask: a paused run is a
                // journaled checkpoint, not a process — it holds no lease and
                // no concurrency slot, so any driver may continue it.
                slots.release(&self.facade);
                lease.release(&self.facade);
                return Ok(RunSession::Parked {
                    envelope: json!({
                        "kind": "paused",
                        "reason": "paused",
                        "run_id": run_id,
                        "checkpoint": prev_ckpt.map(|h| h.to_hex()),
                        "superstep": st.superstep,
                        "paused_by": req.paused_by,
                        "because": req.because,
                        "requested_at": req.requested_at,
                        "paused_at": now,
                        "asks": [],
                    }),
                    run_id,
                });
            }

            // Parked with the envelope announced in an earlier session.
            if !st.pending_asks.is_empty() {
                slots.release(&self.facade);
                lease.release(&self.facade);
                return Ok(RunSession::Parked {
                    envelope: json!({
                        "kind": "requires_action",
                        "run_id": run_id,
                        "checkpoint": prev_ckpt.map(|h| h.to_hex()),
                        "asks": st
                            .pending_asks
                            .values()
                            .map(|p| p.ask.clone())
                            .collect::<Vec<_>>(),
                    }),
                    run_id,
                });
            }
            return Err(RunError::Storage {
                detail: "driver made no progress with no work in flight".into(),
            });
        }
    }

    /// Shadow evaluation (§8 Wave 4): journal-consistent replay of the given
    /// runs, writing nothing and dispatching nothing — the verify path holds
    /// no executor, no pool, and no model, so "zero effect dispatches" is a
    /// property of the code shape, and the report states it explicitly for
    /// the §7.4 pre-apply gate. A run that fails to load is reported
    /// inconsistent, never skipped silently.
    pub fn shadow_eval(&self, run_ids: &[String]) -> crate::ShadowReport {
        let mut report = crate::ShadowReport::default();
        for run_id in run_ids {
            let row = match self.verify(run_id) {
                Ok(v) => {
                    let effects = self
                        .facade
                        .with_store(|m| journal::load(m, &self.ns, run_id))
                        .map(|view| {
                            view.entries.values().filter(|e| e.result.is_some()).count() as u64
                        })
                        .unwrap_or(0);
                    crate::ShadowRun {
                        run_id: run_id.clone(),
                        consistent: v.verified,
                        effects_replayed: effects,
                        steps: v.steps,
                    }
                }
                Err(e) => crate::ShadowRun {
                    run_id: run_id.clone(),
                    consistent: false,
                    effects_replayed: 0,
                    steps: vec![crate::VerifyStep {
                        superstep: 0,
                        verdict: format!("shadow load failed: {e}"),
                        ok: false,
                    }],
                },
            };
            report.runs.push(row);
        }
        report.all_consistent =
            !report.runs.is_empty() && report.runs.iter().all(|r| r.consistent);
        report.effect_dispatches = 0;
        report
    }

    /// The `run_id`s of the most recent runs in this namespace, newest
    /// first — manifest link Facts in `agent:harness` are the census.
    ///
    /// Ids only and no total, so this reads a BOUNDED window and widens it
    /// only when the window held fewer than `limit` runs. `agent:harness`
    /// also holds cancel, fork and audit Facts, which dilute a fixed window
    /// until it carries fewer runs than it has room for — that dilution is
    /// the bug (#165), and escalating on it is what fixes it without making
    /// every "give me the newest run" call read the whole namespace.
    pub fn recent_runs(&self, limit: usize) -> Result<Vec<String>, RunError> {
        let mut scan = limit.saturating_mul(8).max(64);
        loop {
            let page = self.run_index(None, 0, limit, scan)?;
            if page.rows.len() >= limit || page.exhausted || scan >= RUN_INDEX_MAX_SCAN {
                return Ok(page.rows.into_iter().map(|(_, id, _)| id).collect());
            }
            scan = scan.saturating_mul(4).min(RUN_INDEX_MAX_SCAN);
        }
    }

    /// The run index, paged and optionally scoped to one session namespace
    /// (#165): rows newest first with outcome and spend, the total that
    /// matched, and whether the page cut the set short.
    ///
    /// Every `run:<id> mg:harness` link in `agent:harness` is read, not a
    /// bounded window of recent Facts: that namespace also holds cancel,
    /// fork and audit records, so a window of "the newest N Facts" held
    /// fewer than N runs and silently dropped the older ones — which, on a
    /// memory with two tenants, read as "this namespace has no runs" while
    /// it had open approvals. A `total` is inherently a full census, which
    /// is what this call promises and `recent_runs` deliberately does not.
    ///
    /// `ns` accepts an exact namespace or an `"org.*"` prefix scope, and a
    /// scoped read returns the runs KNOWN to match it. A link written before
    /// the namespace was stamped on it (`run_ns` absent) is not known to
    /// match anything, so it is excluded and counted in `unattributed`
    /// instead — never silently, and never by guessing: a run's plan
    /// namespace is not its session namespace, so inferring one would be a
    /// guess, and admitting it to every scope would make scoping meaningless
    /// on a memory upgraded across this change. The unscoped listing (the
    /// default) still shows every such run.
    pub fn list_runs(
        &self,
        ns: Option<&str>,
        offset: usize,
        limit: usize,
    ) -> Result<crate::RunListPage, RunError> {
        let page = self.run_index(ns, offset, limit, RUN_INDEX_MAX_SCAN)?;
        let (total, unattributed) = (page.total, page.unattributed);
        let truncated = offset + page.rows.len() < total;
        let runs = page
            .rows
            .into_iter()
            .map(|(created_at, run_id, run_ns)| {
                let (outcome, usd, tokens) = self.run_outcome(&run_id);
                crate::RunRow { run_id, ns: run_ns, created_at, outcome, spent_usd_micros: usd, spent_tokens: tokens }
            })
            .collect();
        Ok(crate::RunListPage { runs, total, truncated, offset, limit, unattributed })
    }

    /// One run's terminal outcome and spend, from its run-outcome
    /// Observation. An INDEXED read (`run_idx`, via `run_grains`), not a
    /// scan: the Observation carries a top-level `run_id`, so it is joined
    /// to the run directly. Both alternatives were worse — a scan per row is
    /// the N+1 this replaced, and one shared scan over the outcome census
    /// cannot terminate early, because a run that is still OPEN has no
    /// outcome to find and open runs are the normal case on the very page
    /// (the approval queue) this serves.
    ///
    /// An unfinished run answers `("open", None, None)`, which is also what
    /// a run whose census record is missing answers — the list must not fail
    /// because one record is absent.
    fn run_outcome(&self, run_id: &str) -> (String, Option<u64>, Option<u64>) {
        // Page to EXHAUSTION, keeping the last match. `run_grains` is
        // ascending by seq and the outcome is the run's last harness record,
        // so a single page would report a call-heavy run — one that logged
        // more egress calls than the page holds — as open forever.
        let mut found = None;
        let mut after = 0i64;
        loop {
            let page = self
                .facade
                .with_store(|m| m.run_grains(HARNESS_NS, run_id, after, RUN_HARNESS_PAGE))
                .unwrap_or_default();
            let exhausted = page.len() < RUN_HARNESS_PAGE;
            for (seq, g) in page {
                after = seq;
                if g.get_str("observation_kind") == Some("run_outcome")
                    && g.get_str("run_id") == Some(run_id)
                {
                    found = Some(g);
                }
            }
            if exhausted {
                break;
            }
        }
        match found {
            Some(g) => (
                g.get_str("object").unwrap_or("open").to_string(),
                g.get_u64("spent_usd_micros"),
                Some(
                    g.get_u64("spent_input_tokens").unwrap_or(0)
                        + g.get_u64("spent_output_tokens").unwrap_or(0),
                ),
            ),
            None => ("open".to_string(), None, None),
        }
    }

    /// The `mg:harness` links, newest first, deduplicated per run, scoped
    /// and paged: `(page, total)`.
    fn run_index(
        &self,
        ns: Option<&str>,
        offset: usize,
        limit: usize,
        scan: usize,
    ) -> Result<RunIndexPage, RunError> {
        let scanned = self
            .facade
            .with_store(|m| m.recent(HARNESS_NS, Some(areev_core::types::GrainType::Fact), scan))
            .map_err(err_run)?;
        // Fewer grains than asked for means the namespace ran out, so the
        // census is complete and a caller may stop widening.
        let exhausted = scanned.len() < scan;
        let mut rows: Vec<RunIndexRow> = scanned
            .into_iter()
            .filter(|g| g.get_str("relation") == Some("mg:harness"))
            .filter_map(|g| {
                let id = g.get_str("run_id")?.to_string();
                let run_ns = g.get_str("run_ns").map(String::from);
                Some((g.get_i64("created_at").unwrap_or(0), id, run_ns))
            })
            .collect();
        rows.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(&b.1)));
        let mut seen = std::collections::HashSet::new();
        rows.retain(|(_, id, _)| seen.insert(id.clone()));
        let mut unattributed = 0usize;
        if let Some(scope) = ns.filter(|s| !s.is_empty() && *s != "*") {
            rows.retain(|(_, _, run_ns)| match run_ns.as_deref() {
                Some(n) => ns_in_scope(scope, n),
                None => {
                    unattributed += 1;
                    false
                }
            });
        }
        let total = rows.len();
        let rows = rows.into_iter().skip(offset).take(limit).collect();
        Ok(RunIndexPage { rows, total, exhausted, unattributed })
    }

    /// One run's full picture for an operator UI (issue #34): the frozen
    /// manifest, terminal state, spend, journal size, pending asks, fork
    /// lineage. Read-only — the same report `areev run inspect` prints.
    pub fn inspect(&self, run_id: &str) -> Result<crate::InspectReport, RunError> {
        let manifest = self.load_manifest(run_id)?;
        let view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, run_id))
            .map_err(err_run)?;
        let last = view.checkpoints.last();
        let scheduler = last.map(|c| c.scheduler.clone()).unwrap_or(Value::Null);
        let asks = scheduler.get("pending_asks").cloned().unwrap_or(json!({}));
        let mut phase = scheduler.get("phase").and_then(|p| p.as_str().map(String::from)).or_else(
            || {
                scheduler.get("phase").map(|p| {
                    if p.get("Finished").is_some() { "finished".into() } else { "open".into() }
                })
            },
        );
        // A host pause (#344): who asked, when, why — and, once a driver
        // honoured it, where it parked. A paused run's checkpoint is an
        // ordinary Idle one, so the phase is the pause record's to name. A
        // terminal run keeps `finished` whatever request is left standing.
        let finished = phase.as_deref() == Some("finished");
        let pause = view.pause.active.as_ref().filter(|_| !finished).map(|req| {
            let mut p = json!({
                "status": if view.pause.is_paused() { "paused" } else { "requested" },
                "paused_by": req.paused_by,
                "because": req.because,
                "requested_at": req.requested_at,
                "request": req.request,
            });
            if let (Some((superstep, at)), Some(o)) = (view.pause.applied, p.as_object_mut()) {
                o.insert("paused_at".into(), json!(at));
                o.insert("superstep".into(), json!(superstep));
            }
            p
        });
        if view.pause.is_paused() && !finished {
            phase = Some("paused".into());
        }
        Ok(crate::InspectReport {
            run_id: run_id.to_string(),
            plan_hash: manifest.plan_hash.clone(),
            principal: manifest.principal.clone(),
            pinned: manifest
                .pinned
                .iter()
                // `llm.dispatch` pins (#392) follow the nodes' rows, marked.
                .chain(&manifest.dispatch_pins)
                .map(|p| {
                    // The resolution, not a summary of it (#230). A node
                    // bound to a Definition that names a code blob used to
                    // print as a bare `"executor": "host"` — identical to a
                    // `--tool-cmd` node — so a reader could not tell a
                    // capability tool that resolved correctly from one whose
                    // Definition had not been found at all. That cost an
                    // afternoon and produced the wrong conclusion. What a run
                    // froze at start is what `inspect` must say.
                    let mut row =
                        json!({"node": p.node, "tool": p.tool_name, "executor": p.executor});
                    let row_obj = row.as_object_mut().expect("json! built an object");
                    if let Some(uri) = &p.executor_uri {
                        row_obj.insert("executor_uri".into(), json!(uri));
                    }
                    if let Some(rt) = &p.runtime {
                        row_obj.insert("runtime".into(), json!(rt));
                    }
                    if let Some(caps) = &p.capabilities {
                        row_obj.insert("capabilities".into(), caps.clone());
                    }
                    if let Some(read) = &p.read {
                        row_obj.insert("read".into(), read.clone());
                    }
                    if let Some(lock) = &p.locked_params {
                        row_obj.insert("locked_params".into(), lock.clone());
                    }
                    if manifest.dispatch_pins.iter().any(|d| std::ptr::eq(d, p)) {
                        row_obj.insert("dispatch_only".into(), json!(true));
                    }
                    // The node's frozen `llm` block (#391), dynamic tools
                    // as pinned (#392) — what a resume will send, not what
                    // the plan says today.
                    if let Some(llm) = &p.llm {
                        row_obj.insert(
                            "llm".into(),
                            serde_json::to_value(llm).unwrap_or(Value::Null),
                        );
                    }
                    row
                })
                .collect(),
            budgets: serde_json::to_value(manifest.budgets).unwrap_or(Value::Null),
            fork_of: manifest
                .fork_of
                .as_ref()
                .map(|f| json!({"base_run": f.base_run, "base_superstep": f.base_superstep})),
            checkpoints: view.checkpoints.len(),
            journal_entries: view.entries.len(),
            superstep: scheduler.get("superstep").cloned(),
            phase,
            spent: scheduler.get("spent").cloned(),
            pending_asks: asks,
            warnings: manifest.warnings.clone(),
            limits: json!({
                "llm_max_tokens": manifest.llm_reserve_tokens(),
                "max_effects_per_attempt": manifest.max_effects_per_attempt(),
                "llm_context_tokens": manifest.llm_context_tokens,
                "llm_tool_result_chars": manifest.llm_tool_result_chars,
            }),
            pause,
        })
    }

    /// The EU AI Act Article 14 report for `run_id` (or the newest run of
    /// `plan`, or the newest run overall when neither is given), measured
    /// from the journal rather than asserted (issue #34): where a human can
    /// intervene, who is authorized to, what expires when, and how fast the
    /// kill switch actually drained. Read-only — the same report
    /// `areev run oversight-report` prints.
    pub fn oversight_report(
        &self,
        run_id: Option<&str>,
        plan: Option<&Hash>,
    ) -> Result<crate::OversightReport, RunError> {
        let target_run = match (run_id, plan) {
            (Some(id), _) => Some(id.to_string()),
            (None, Some(plan)) => {
                let plan_hex = plan.to_hex();
                self.recent_runs(64)?.into_iter().find(|id| {
                    self.load_manifest(id).map(|mf| mf.plan_hash == plan_hex).unwrap_or(false)
                })
            }
            (None, None) => self.recent_runs(1)?.into_iter().next(),
        };
        let Some(run_id) = target_run else {
            return Err(RunError::UnresolvedRef {
                what: "no runs recorded yet (or none for that plan) — nothing to report on"
                    .into(),
            });
        };
        let manifest = self.load_manifest(&run_id)?;
        let gated: Vec<Value> = manifest
            .pinned
            .iter()
            .filter(|p| p.executor == "client")
            .map(|p| json!({"node": p.node, "tool": p.tool_name}))
            .collect();
        // Authorized responders: principals granted run.respond in the file.
        let responders: Vec<String> = self
            .facade
            .with_store(|m| {
                m.recent("agent:authz", Some(areev_core::types::GrainType::Fact), 4096)
            })
            .map(|rows| {
                rows.into_iter()
                    .filter(|g| {
                        g.get_str("relation") == Some("mg:permits")
                            && g.get_str("object").is_some_and(|o| o.contains("run.respond"))
                    })
                    .filter_map(|g| g.get_str("subject").map(String::from))
                    .collect()
            })
            .unwrap_or_default();
        // Kill-switch time, measured: cancel Fact created_at → terminal
        // checkpoint created_at, over every canceled run we can see.
        let mut cancel_ms: Vec<i64> = Vec::new();
        for id in self.recent_runs(64).unwrap_or_default() {
            let cancel_at = self
                .facade
                .with_store(|m| {
                    m.latest("agent:harness", &format!("run:{id}"), "mg:cancel").ok().flatten()
                })
                .and_then(|g| g.get_i64("created_at"));
            let Some(cancel_at) = cancel_at else { continue };
            if let Ok(view) = self.facade.with_store(|m| journal::load(m, &self.ns, &id)) {
                if let Some(last) = view.checkpoints.last() {
                    let drained_at = last.decisions.clock_close_ms as i64;
                    if drained_at >= cancel_at {
                        cancel_ms.push(drained_at - cancel_at);
                    }
                }
            }
        }
        Ok(crate::OversightReport {
            run_id,
            plan_hash: manifest.plan_hash.clone(),
            human_gates: json!({
                "client_gated_nodes": gated,
                "every_client_ask_is_an_approval": true,
                "separation_of_duties": "responder != triggering principal, refused structurally",
                "ask_ttl_sec": manifest.ask_ttl_sec,
            }),
            authorized_responders: json!({
                "principals_granted_run_respond": responders,
                "note": "empty = owner-only sessions (single-user mode); grants live IN the file as mg:permits Facts",
            }),
            budgets: serde_json::to_value(manifest.budgets).unwrap_or(Value::Null),
            kill_switch: json!({
                "verb": "run.cancel (deliberately the lowest-privilege run verb)",
                "measured_cancel_to_drain_ms": cancel_ms,
                "note": "measured from journaled cancel Facts to the terminal checkpoint's journaled close",
            }),
        })
    }

    /// §5.5 gate 1 / verify tier 1: **journal-consistent replay**. Re-drives
    /// the run from the manifest's input with the clock scripted from
    /// journaled readings and every effect answered from the journal —
    /// writing NOTHING — and byte-compares every checkpoint the replay
    /// commands against the stored chain. A missing journaled result stops
    /// verification at that step (reported, not asserted); a divergent
    /// checkpoint is the `RUN-E009` condition, reported with both hashes.
    pub fn verify(&self, run_id: &str) -> Result<VerifyReport, RunError> {
        let manifest = self.load_manifest(run_id)?;
        let plan_hash = Hash::from_hex(&manifest.plan_hash)
            .map_err(|_| RunError::ManifestMismatch { why: "bad plan hash".into() })?;
        let plan = self.load_plan(&plan_hash)?;
        let view = self
            .facade
            .with_store(|m| journal::load(m, &self.ns, run_id))
            .map_err(err_run)?;
        let executors = manifest.executors();
        let arg_schemas = self.load_arg_schemas(&manifest)?;
        let validate_args = make_validate_args(&arg_schemas);
        let reduce = crate::reducers::make_reduce(&manifest.reducers);
        // From the MANIFEST's pin, never from this host: verify must ask what
        // the run asked, whoever is verifying — and it answers every ask from
        // the journal, so no backend is ever called here.
        let decide_ctx = crate::shadow::decide_context_for(&self.facade, &manifest)?;
        let env = StepEnv {
            plan: &plan,
            executors: &executors,
            budgets: manifest.budgets.to_budgets(),
            reduce: &reduce,
            eval_cond: &builtin_eval,
            ask_ttl_sec: manifest.ask_ttl_sec,
            validate_args: &validate_args,
            llm_reserve_tokens: manifest.llm_reserve_tokens(),
            max_effects_per_attempt: manifest.max_effects_per_attempt(),
            llm_tool_result_chars: manifest.llm_tool_result_chars,
            llm_context_tokens: manifest.llm_context_tokens,
            decide: decide_ctx.env(&manifest),
        };

        let mut report = VerifyReport::default();
        if view.checkpoints.is_empty() {
            report.steps.push(VerifyStep {
                superstep: 0,
                verdict: "no checkpoints — nothing to verify".into(),
                ok: false,
            });
            return Ok(report);
        }

        // A fork replays its TAIL: the seed checkpoint is trusted through
        // the §5.4 base-checkpoint hash chain (it `derived_from`s the base),
        // and replay verifies everything the fork itself executed.
        let (mut st, mut ckpt_idx, mut events);
        if manifest.fork_of.is_some() {
            st = serde_json::from_value::<SchedulerState>(view.checkpoints[0].scheduler.clone())
                .map_err(|e| RunError::ManifestMismatch {
                    why: format!("fork seed checkpoint: {e}"),
                })?;
            ckpt_idx = 1;
            if view.checkpoints.len() == 1 {
                report.steps.push(VerifyStep {
                    superstep: st.superstep,
                    verdict: format!(
                        "fork seed checkpoint {} chains to base {} — the fork has \
                         not executed yet",
                        view.checkpoints[0].hash.to_hex(),
                        manifest.fork_of.as_ref().map(|f| f.base_checkpoint.as_str()).unwrap_or("?"),
                    ),
                    ok: true,
                });
                report.verified = true;
                return Ok(report);
            }
            events = vec![EventIn::ClockReading {
                unix_ms: view.checkpoints[1].decisions.clock_open_ms,
            }];
        } else {
            st = SchedulerState::new(run_id, &plan);
            ckpt_idx = 0;
            // The first reading is the first superstep's journaled open.
            events = vec![
                EventIn::ClockReading {
                    unix_ms: view.checkpoints[0].decisions.clock_open_ms,
                },
                EventIn::Start { input: manifest.input.clone() },
            ];
        }
        // Cancel is an OUT-OF-BAND marker (drive polls the Fact; the
        // journal has no CancelSeen entry) — replay it from where it IS
        // journaled: the checkpoint state. WHERE inside the superstep the
        // live driver saw it is read off the journal's own evidence: the
        // driver polls at wave boundaries, so a checkpoint that carries
        // both this superstep's resolutions and the cancel means the
        // marker rode the SAME event batch as the resolutions; a step
        // whose dispatches the journal never answered, while the coming
        // checkpoint carries the cancel, means the live driver canceled
        // BEFORE dispatching — that boundary is rewound and re-fed with
        // CancelSeen up front. Feeding it blindly at the superstep's open
        // (the old rule) replayed a phase the live driver can only
        // produce when the marker predates the run.
        let cancel_peek = |ckpt_idx: usize, st: &SchedulerState| -> Option<EventIn> {
            if st.cancel.is_some() {
                return None;
            }
            let c = view.checkpoints.get(ckpt_idx)?.scheduler.get("cancel")?.as_array()?;
            Some(EventIn::CancelSeen {
                principal: c.first()?.as_str()?.to_string(),
                reason: c.get(1)?.as_str()?.to_string(),
            })
        };
        // The live driver feeds a steering message only while a superstep is
        // open, and an open is the only thing that drains `inbox` — so a
        // message is inert for the whole superstep that observed it. Replay
        // therefore never needs to know WHICH wave saw one: feeding them with
        // the batch that closes a checkpoint, bounded by that checkpoint's
        // own `inputs_seen`, reproduces the state exactly.
        let input_peek = |ckpt_idx: usize, st: &SchedulerState| -> Vec<EventIn> {
            let want = view
                .checkpoints
                .get(ckpt_idx)
                .and_then(|c| c.scheduler.get("inputs_seen"))
                .and_then(|v| v.as_u64())
                .unwrap_or(0);
            view.inputs
                .iter()
                .take(want as usize)
                .skip(st.inputs_seen as usize)
                .map(|m| EventIn::InputSeen { message: m.clone() })
                .collect()
        };
        let mut guard = 0;
        'replay: loop {
            guard += 1;
            if guard > 100_000 {
                report.steps.push(VerifyStep {
                    superstep: st.superstep,
                    verdict: "replay did not terminate".into(),
                    ok: false,
                });
                return Ok(report);
            }
            let pre_state = st.clone();
            let pre_events = events.clone();
            let pre_ckpt_idx = ckpt_idx;
            let pre_report_len = report.steps.len();
            let out = step(&env, st, &events);
            st = out.state;
            events = Vec::new();
            let mut resolved: Vec<EventIn> = Vec::new();
            let mut parked = false;
            let mut unanswered: Option<JournalKey> = None;
            // Set when the checkpoint just verified is followed by one the
            // live driver opened after a RESUME: (the verified checkpoint,
            // the reading the driver took on picking the run back up).
            let mut resume_boundary: Option<(usize, u64)> = None;
            // Set when the checkpoint just verified is followed DIRECTLY by a
            // canceled terminal at the same superstep: the live driver stopped
            // there (a pause, #344, or a crash) and a later drive finished the
            // run on its cancel marker without opening anything. (The verified
            // checkpoint, the terminal's reading, the cancel.)
            let mut cancel_boundary: Option<(usize, u64, EventIn)> = None;
            for cmd in out.commands {
                match cmd {
                    Command::WriteIntent { .. } => {}
                    Command::Dispatch { key, .. } => {
                        match view.entries.get(&key).and_then(|e| {
                            e.result.as_ref().map(|(_, o)| (o.clone(), e.result_clock_ms))
                        }) {
                            Some((outcome, _clock)) => {
                                resolved.push(EventIn::EffectResolved { key, outcome });
                            }
                            None => {
                                unanswered = Some(key);
                                break;
                            }
                        }
                    }
                    Command::EmitEnvelope { .. } => {
                        parked = true;
                    }
                    Command::WriteCheckpoint { superstep, state_json, .. } => {
                        let Some(stored) = view.checkpoints.get(ckpt_idx) else {
                            report.steps.push(VerifyStep {
                                superstep,
                                verdict: "replay produced a checkpoint the journal lacks"
                                    .into(),
                                ok: false,
                            });
                            break 'replay;
                        };
                        ckpt_idx += 1;
                        let ok = stored.superstep == superstep
                            && stored.scheduler == state_json;
                        report.steps.push(VerifyStep {
                            superstep,
                            verdict: if ok {
                                format!(
                                    "checkpoint {} journal-consistent",
                                    stored.hash.to_hex()
                                )
                            } else {
                                format!(
                                    "checkpoint {} DIVERGES from replay (RUN-E009): {}",
                                    stored.hash.to_hex(),
                                    diff_fields(&stored.scheduler, &state_json)
                                )
                            },
                            ok,
                        });
                        if !ok {
                            break 'replay;
                        }
                        // Did the live driver STOP here? `step` closes a
                        // superstep and opens the next one in the same call,
                        // at the same reading — so a replay always runs
                        // straight through a boundary the live driver may
                        // have crashed at and come back to minutes later.
                        // The next checkpoint's journaled `resumed_at` is the
                        // only evidence of that gap, and rewinding to feed it
                        // is what stops the crash span being charged as
                        // active wall (the RUN-E009-on-every-recovered-run
                        // defect).
                        if let Some(at) =
                            view.checkpoints.get(ckpt_idx).and_then(|c| c.decisions.resumed_at)
                        {
                            resume_boundary = Some((ckpt_idx - 1, at));
                        } else if let Some(next) = view.checkpoints.get(ckpt_idx).filter(|n| {
                            // An ordinary finish ALSO follows its close at
                            // the same superstep; what sets this one apart is
                            // a close that carried no cancel followed by a
                            // terminal that does — a cancel seen in a LATER
                            // drive, which only a stop at Idle allows.
                            n.superstep == superstep
                                && state_json.get("phase") == Some(&json!("Idle"))
                                && state_json.get("cancel").is_none_or(|c| c.is_null())
                                && n.scheduler
                                    .get("phase")
                                    .and_then(|p| p.get("Finished"))
                                    .and_then(|f| f.get("Canceled"))
                                    .is_some()
                        }) {
                            // The stop-then-cancel boundary. The replay ran
                            // straight on into the next superstep; the live
                            // driver never did — it resumed at `Idle` (so fed
                            // `Resumed`) and finished on the cancel marker.
                            let cancel = next
                                .scheduler
                                .get("cancel")
                                .and_then(|v| v.as_array())
                                .and_then(|a| {
                                    Some(EventIn::CancelSeen {
                                        principal: a.first()?.as_str()?.to_string(),
                                        reason: a.get(1)?.as_str()?.to_string(),
                                    })
                                });
                            let at = next.scheduler.get("clock_ms").and_then(|v| v.as_u64());
                            if let (Some(cancel), Some(at)) = (cancel, at) {
                                cancel_boundary = Some((ckpt_idx - 1, at, cancel));
                            }
                        }
                    }
                    Command::Finish { .. } => {}
                }
            }
            if let Some((verified_idx, at)) = resume_boundary {
                // Rewind to the checkpoint just verified and re-enter the
                // boundary the way the driver did. Reading the stored state
                // back trusts nothing new: it was byte-compared one command
                // ago. Whatever this pass dispatched after the close is
                // discarded and re-derived at the resumed clock, under the
                // same journal keys.
                st = serde_json::from_value::<SchedulerState>(
                    view.checkpoints[verified_idx].scheduler.clone(),
                )
                .map_err(|e| RunError::ManifestMismatch {
                    why: format!("checkpoint {verified_idx} state: {e}"),
                })?;
                events = vec![EventIn::Resumed, EventIn::ClockReading { unix_ms: at }];
                continue;
            }
            if let Some((verified_idx, at, cancel)) = cancel_boundary {
                // Same rewind, fed the batch the finalizing drive saw: the
                // resume marker, its reading, and the cancel — which finishes
                // the run from Idle before anything opens.
                st = serde_json::from_value::<SchedulerState>(
                    view.checkpoints[verified_idx].scheduler.clone(),
                )
                .map_err(|e| RunError::ManifestMismatch {
                    why: format!("checkpoint {verified_idx} state: {e}"),
                })?;
                events = vec![EventIn::Resumed, EventIn::ClockReading { unix_ms: at }, cancel];
                continue;
            }
            if let Some(key) = unanswered {
                if let Some(cancel_ev) = cancel_peek(pre_ckpt_idx, &pre_state) {
                    // The live driver canceled at this boundary BEFORE
                    // dispatching (the journal never answered these
                    // effects, and the coming checkpoint carries the
                    // cancel): rewind and re-feed the same boundary with
                    // CancelSeen — the batch the live driver actually saw.
                    st = pre_state;
                    ckpt_idx = pre_ckpt_idx;
                    report.steps.truncate(pre_report_len);
                    events = pre_events;
                    events.push(cancel_ev);
                    continue;
                }
                // A PAUSED run (#344) stopped at the checkpoint just verified
                // and has dispatched nothing since: the replay's next open is
                // work the run has simply not done yet. That is the whole run
                // verified, not a gap — the same verdict a park on an ask gets.
                if ckpt_idx >= view.checkpoints.len() && view.pause.is_paused() {
                    report.steps.push(VerifyStep {
                        superstep: pre_state.superstep.max(st.superstep.saturating_sub(1)),
                        verdict: "run is paused by its host — verified up to the pause".into(),
                        ok: true,
                    });
                    break 'replay;
                }
                report.steps.push(VerifyStep {
                    superstep: st.superstep,
                    verdict: format!(
                        "effect {} has no journaled result — not \
                         verifiable past this point",
                        key.tool_call_id()
                    ),
                    ok: false,
                });
                break 'replay;
            }
            if st.is_terminal() {
                // The terminal checkpoint (written by drive on Finish).
                if let Some(stored) = view.checkpoints.get(ckpt_idx) {
                    let replay_json = serde_json::to_value(&st).unwrap_or(Value::Null);
                    let ok = stored.scheduler == replay_json;
                    report.steps.push(VerifyStep {
                        superstep: st.superstep,
                        verdict: if ok {
                            format!(
                                "terminal checkpoint {} journal-consistent",
                                stored.hash.to_hex()
                            )
                        } else {
                            format!(
                                "terminal checkpoint {} DIVERGES from replay (RUN-E009)",
                                stored.hash.to_hex()
                            )
                        },
                        ok,
                    });
                }
                break 'replay;
            }
            if !resolved.is_empty() {
                // The close-time reading: this superstep's journaled close.
                let close = view
                    .checkpoints
                    .get(ckpt_idx)
                    .map(|c| c.decisions.clock_close_ms)
                    .unwrap_or(st.clock_ms);
                events.push(EventIn::ClockReading { unix_ms: close });
                events.extend(resolved);
                // The checkpoint these resolutions close may also carry
                // the cancel: the live driver polls the marker at wave
                // boundaries, so it saw it WITH this batch — CancelSeen
                // rides the same feed, never a boundary of its own.
                if let Some(cancel_ev) = cancel_peek(ckpt_idx, &st) {
                    events.push(cancel_ev);
                }
                events.extend(input_peek(ckpt_idx, &st));
                continue;
            }
            if parked {
                // Settle parked asks from the journal. The reading fed here
                // is the NEXT stored checkpoint's journaled close — the
                // scheduler accrues park-elapsed at close from that reading
                // (never from response-apply time, which is not journaled),
                // so this is exactly what makes the replayed state byte-equal.
                let mut settled_any = false;
                let close = view
                    .checkpoints
                    .get(ckpt_idx)
                    .map(|c| c.decisions.clock_close_ms)
                    .unwrap_or(st.clock_ms);
                for (id, pending) in st.pending_asks.clone() {
                    if matches!(
                        executors.get(pending.node_idx),
                        Some(NodeExecutor::Subgraph { .. })
                    ) {
                        if view.entries.contains_key(&pending.key) {
                            events.push(EventIn::AskForwarded { tool_call_id: id });
                            settled_any = true;
                        }
                        continue;
                    }
                    if let Some(entry) = view.entries.get(&pending.key) {
                        if let Some((_, outcome)) = &entry.result {
                            if !settled_any {
                                events.push(EventIn::ClockReading { unix_ms: close });
                            }
                            events.push(EventIn::ResponseSettled {
                                tool_call_id: id,
                                outcome: outcome.clone(),
                            });
                            settled_any = true;
                        }
                    }
                }
                // Queued steering rides the SAME batch as whatever settles
                // this park — never a continue of its own, which would leave
                // a cancel-while-parked unfed and spin the replay loop.
                events.extend(input_peek(ckpt_idx, &st));
                if settled_any {
                    continue;
                }
                // A cancel that arrived while parked: the next checkpoint
                // carries it — feed it (with its journaled close) so the
                // drain and terminal checkpoints verify too, instead of
                // breaking here and silently skipping them.
                let cancel_next = view.checkpoints.get(ckpt_idx).and_then(|c| {
                    c.scheduler.get("cancel").and_then(|v| v.as_array()).and_then(|a| {
                        Some((
                            a.first()?.as_str()?.to_string(),
                            a.get(1)?.as_str()?.to_string(),
                        ))
                    })
                });
                if st.cancel.is_none() {
                    if let Some((by, reason)) = cancel_next {
                        events.push(EventIn::ClockReading { unix_ms: close });
                        events.push(EventIn::CancelSeen { principal: by, reason });
                        continue;
                    }
                }
                report.steps.push(VerifyStep {
                    superstep: st.superstep,
                    verdict: if ckpt_idx < view.checkpoints.len() {
                        format!(
                            "run is parked awaiting a response — verified up to                              the park; {} trailing checkpoint(s) NOT verified",
                            view.checkpoints.len() - ckpt_idx
                        )
                    } else {
                        "run is parked awaiting a response — verified up to the park".into()
                    },
                    ok: ckpt_idx >= view.checkpoints.len(),
                });
                break 'replay;
            }
            // No dispatches, not parked, not terminal: feed the next open
            // reading (a between-supersteps boundary).
            if let Some(next) = view.checkpoints.get(ckpt_idx) {
                events.push(EventIn::ClockReading {
                    unix_ms: next.decisions.clock_open_ms,
                });
                continue;
            }
            break 'replay;
        }
        report.verified = !report.steps.is_empty() && report.steps.iter().all(|s| s.ok);
        Ok(report)
    }
}