areev_run_core/

step.rs

//! The sans-IO step function (governed-agents §4): apply the driver's
//! events, progress the BSP machine as far as pure reasoning allows, return
//! the commands the driver must perform. Same plan + same env + same events
//! in the same order ⇒ byte-identical state and commands — that is the
//! property the DST harness and the three CI gates exercise.
//!
//! Operational semantics (v1, refining §6.2's generation model into a
//! per-node re-entry form with the same observable contract):
//!
//! - **Superstep**: open (record the journaled clock; check every budget
//!   axis) → dispatch every Ready node in canonical order → collect ALL
//!   results (retries included) → merge results through the reducers in
//!   canonical order → evaluate completed nodes' out-edges in canonical
//!   order, journaling every outcome → deliver firings → checkpoint → close
//!   (record the clock again; wall charges close − open, so a three-day
//!   HITL park costs nothing).
//! - **Re-entry**: an edge firing into a node already resolved at its
//!   current generation increments the node's generation and resets it to
//!   Waiting: the firing edge is Fired at the new generation; every other
//!   in-edge is Pending if its source is still unresolved (it may yet fire)
//!   or Dead otherwise. This is how a bounded cycle iterates — the Rev-2
//!   edge machine, whose states were all terminal, could not.
//! - **Death**: a Waiting node whose in-edges all resolved with none Fired
//!   is Dead; its out-edges propagate Dead in canonical order — a join
//!   below an untaken branch (or a died-out cycle) resolves instead of
//!   pending forever.
//! - **Fail-fast** (v1's only failure policy): a node that exhausts its
//!   retries marks the run draining — no retries elsewhere are suppressed
//!   (each node's retry decision is independent, or determinism under
//!   permutation breaks), but no edges evaluate at close and no new
//!   superstep opens; the run finishes `Failed` naming the LOWEST-index
//!   failed node, chosen at close, not at arrival.
//! - **Cancel**: drain; retries stop; parked Client asks are abandoned (the
//!   journal keeps their grains); finish `Canceled`.
//! - **Pause** (#344): `EventIn::PauseRequested` in a batch stops that call
//!   from OPENING a superstep — nothing else. The open one finishes normally
//!   and closes into an ordinary `Idle` checkpoint; terminal outcomes (cancel
//!   included) still win. Nothing is stored, so the checkpoint is the one an
//!   uninterrupted run writes, and resuming is `EventIn::Resumed`.
//! - **Budgets** (§6.7): every axis is checked at superstep open, before
//!   any dispatch. Per-dispatch reservation refines this in Wave 2, where
//!   LLM effects carry a mandatory `max_tokens` reserve to check — a v1
//!   Host tool has no reserve quantity, so open-time is the honest v1
//!   enforcement point, with overshoot bounded by one superstep's
//!   dispatches (stated, not hidden).

use crate::error::{BudgetAxis, RunError};
use crate::plan::PlanGraph;
use crate::state::{
    DecideInFlight, EdgeRes, FoldInFlight, NodeState, PendingAsk, Phase, SchedulerState,
};
use crate::types::{
    Ask, Budgets, Command, DecisionRecord, EdgeOutcome, EffectKind, EffectOutcome, EventIn,
    FailCause, FoldRecord, JournalKey, LlmNodeConfig, NodeExecutor, OfferedTool, RunOutcome,
    DECIDE_TOOL,
};
use areev_core::format::tool_schema::normalize_tool_name;
use serde_json::Value;
use std::collections::BTreeMap;

/// Effects one node attempt may spend when the host pins no other number.
/// Named because three places must agree — both of the driver's `StepEnv`
/// sites and the manifest accessor that feeds them — and because changing it
/// changes the behaviour of every existing plan that relies on the bound.
pub const DEFAULT_MAX_EFFECTS_PER_ATTEMPT: u32 = 16;

/// Transcript entries a fold keeps at the END, beyond `messages[0]`.
///
/// Four is two complete rounds at one tool call per turn, which is the shortest
/// tail that still shows the model what it just did and what came back. It is
/// expressed in entries rather than "rounds" because a round has no fixed
/// length — a turn may issue three tool calls or none — and a count of entries
/// is something `fold_range` can check without parsing the transcript's shape.
/// The cut then moves forward off any `tool` entry, which is what actually
/// keeps rounds intact.
pub const KEEP_TAIL: usize = 4;

/// The instruction a summarizer turn carries.
///
/// This text RIDES THE JOURNAL: it is part of the fold intent's `input`, so a
/// stored run records the exact prompt its summary was produced from. Changing
/// it therefore changes what a journal contains — bump [`FOLD_PROMPT_V`] and
/// say so in the changelog rather than editing it quietly.
pub const FOLD_PROMPT: &str = "Summarize the working state of this task so far \
for someone who will continue it: what has been established, which tools were \
called and what they returned in substance, what remains to be done, and any \
errors or dead ends. Be concrete; keep identifiers, paths and values verbatim. \
Reply with the summary only.";

/// The version of [`FOLD_PROMPT`], journaled beside it so a reader of an old
/// run knows which wording produced its summaries.
pub const FOLD_PROMPT_V: u32 = 1;

/// Version of the decision-guided fold's questions (C2) and of the
/// narrowing question (C1). Their wording rides the journal inside each
/// decision effect's input, so a change to it bumps this — the
/// [`FOLD_PROMPT_V`] convention. The summarizer fold's own prompt did not
/// change when these arrived, so `FOLD_PROMPT_V` did not move.
pub const FOLD_DECIDE_V: u32 = 1;

/// The `keep_call_<n>` question of a decision-guided fold (C2).
pub const KEEP_CALL_QUESTION: &str = "Does the fact that this tool call happened still \
matter for finishing the task?";

/// The `keep_result_<n>` question of a decision-guided fold (C2).
pub const KEEP_RESULT_QUESTION: &str = "Are this tool result's contents still needed \
verbatim to finish the task?";

/// The narrowing question (C1).
pub const OFFER_QUESTION: &str = "Which one of these tools is this step most likely to \
need to call in order to finish its task?";

/// A decision-guided fold keeps a truncated result's first this-many
/// CHARACTERS (not tokens — the [`bound_tool_content`] honesty rule).
pub const FOLD_TRUNCATE_CHARS: usize = 300;

/// A narrowed offer (C1) keeps the top this-many tools by probability …
pub const OFFER_TOP_K: usize = 8;

/// … plus every tool at or above this probability. Narrowing is only ever
/// asked when MORE than [`OFFER_TOP_K`] tools are pinned.
pub const OFFER_KEEP_P: f64 = 0.05;

/// What the scheduler may ask a decision backend, as FROZEN in the run
/// manifest (`docs/decision-model-proposal.md` C1/C2). Pure data: the
/// scheduler never calls a backend — it emits a decision EFFECT the driver
/// answers and journals, the same shape every other effect has.
pub struct DecideEnv<'a> {
    /// The pinned backend's `calibrated()`. The scheduler's two uses of a
    /// decision both OMIT something (tools from an offer, entries from a
    /// prompt), and rule 2 of the proposal is that an uncalibrated backend may
    /// reorder but never omit — so with `false` nothing is asked at all.
    pub calibrated: bool,
    /// The plan's human label (`name`, and its goal/description when it has
    /// one) — narrowing context (C1).
    pub plan_label: Option<&'a str>,
    /// Pinned tool name → its one-line description — narrowing context (C1).
    /// A pure function of the manifest's pinned Definitions (immutable by
    /// hash), so live and verify build the same table.
    pub tool_descriptions: &'a BTreeMap<String, String>,
}

/// The injected pure behavior. Everything here is REQUIRED to be pure —
/// enforced by the journaled decision record + replay assertion, not trust.
pub struct StepEnv<'a> {
    pub plan: &'a PlanGraph,
    /// Per node: the resolved executor (the driver's V7 freeze).
    pub executors: &'a [NodeExecutor],
    pub budgets: Budgets,
    /// Merge one result key into the accumulated state:
    /// `(key, prev, new) -> merged`. The driver's registry with LWW
    /// fallback; must be pure, deterministic, batching-invariant (§6.5).
    pub reduce: &'a dyn Fn(&str, Option<&Value>, &Value) -> Value,
    /// Evaluate an edge's condition against the state — the built-in
    /// (`cond::eval` over the parsed form) or a host evaluator over the raw
    /// string. Must be pure; only consulted for edges that carry a cond.
    pub eval_cond: &'a dyn Fn(&crate::plan::PlanEdge, &Value) -> bool,
    /// TTL for Client asks (seconds); None = asks never expire.
    pub ask_ttl_sec: Option<i64>,
    /// Validate a model-issued tool call's arguments against the pinned
    /// Definition's `input_schema` (§6.11 — the driver wires
    /// `validate_instance` over strict definitions; Ok(()) for non-strict).
    /// Must be pure: it runs identically on replay.
    pub validate_args: &'a dyn Fn(&str, &Value) -> std::result::Result<(), String>,
    /// The per-call token reservation for LLM dispatches (§6.7's
    /// per-dispatch enforcement: `spent + reserve` must fit before an LLM
    /// effect is emitted). The driver sets this to the request's mandatory
    /// `max_tokens`.
    pub llm_reserve_tokens: u64,
    /// Hard bound on effects per node attempt — an abstract node's LLM loop
    /// (turns + tool calls + re-prompts) that exceeds it fails the node
    /// with `ExecutorError` rather than looping forever. Frozen in the run
    /// manifest, so a resume bounds the loop exactly as the start did;
    /// [`DEFAULT_MAX_EFFECTS_PER_ATTEMPT`] when the manifest pins none.
    pub max_effects_per_attempt: u32,
    /// Bound on ONE tool result's size in the TRANSCRIPT (characters — there
    /// is no tokenizer here, and calling it tokens would be a lie). `None` =
    /// unbounded, which is what every run did before the knob existed. The
    /// journal is never bounded: see [`bound_tool_content`].
    pub llm_tool_result_chars: Option<usize>,
    /// Ceiling on the WHOLE transcript, in the provider's own prompt tokens
    /// (`AbstractFlow.last_prompt_tokens`). `None` = no ceiling, which is what
    /// every run did before the fold existed. Reaching it emits one journaled
    /// summarizer turn and splices its result over the folded range; failing
    /// to find a foldable range is `RUN-E024`.
    pub llm_context_tokens: Option<u64>,
    /// The decision backend the manifest pinned, if any (C1/C2). `None` —
    /// every run before decisions existed, and every host without one —
    /// asks nothing and builds exactly the commands it always did.
    pub decide: Option<DecideEnv<'a>>,
}

/// One step's output.
pub struct StepOutcome {
    pub commands: Vec<Command>,
    pub state: SchedulerState,
}

/// Apply events, progress, emit commands. Driver contract: any call that
/// may open or close a superstep includes a fresh `ClockReading`.
pub fn step(env: &StepEnv<'_>, mut st: SchedulerState, events: &[EventIn]) -> StepOutcome {
    let mut out: Vec<Command> = Vec::new();
    for ev in events {
        apply_event(env, &mut st, ev, &mut out);
    }
    // A pause is a property of THIS call's batch, never of the state (#344):
    // see `EventIn::PauseRequested`.
    let hold = events.iter().any(|e| matches!(e, EventIn::PauseRequested));
    progress(env, &mut st, &mut out, hold);
    StepOutcome { commands: out, state: st }
}

fn apply_event(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    ev: &EventIn,
    out: &mut Vec<Command>,
) {
    match ev {
        EventIn::ClockReading { unix_ms } => {
            st.clock_ms = (*unix_ms).max(st.clock_ms);
        }
        EventIn::Start { input } => {
            if st.superstep == 0 && matches!(st.phase, Phase::Idle) {
                // The accumulated state is an object by construction; a
                // non-object input is carried under "input" rather than
                // refused (the driver validates and warns at its surface).
                st.context = match input {
                    Value::Object(_) => input.clone(),
                    other => serde_json::json!({ "input": other }),
                };
                // Bootstrap exception: the entry node (OMS §8.4: nodes[0])
                // activates unconditionally at generation 0, even if a
                // back-edge targets it.
                if st.node_state[0] == NodeState::Waiting {
                    st.node_state[0] = NodeState::Ready;
                }
            }
        }
        EventIn::CancelSeen { principal, reason } => {
            if st.cancel.is_none() && !st.is_terminal() {
                st.cancel = Some((principal.clone(), reason.clone()));
            }
        }
        EventIn::EffectResolved { key, outcome } => {
            resolve_effect(env, st, key, outcome);
        }
        EventIn::Resumed => {
            // Mark where the run stopped, so the coming open can measure the
            // gap. The checkpoint's own clock IS that point: it is the last
            // reading the previous close journaled. Only meaningful between
            // supersteps — a park already set `paused_at` at the park, and
            // overwriting it here would erase the parked span.
            if matches!(st.phase, Phase::Idle) && st.paused_at.is_none() {
                st.paused_at = Some(st.clock_ms);
            }
        }
        EventIn::InputSeen { message } => {
            st.inputs_seen += 1;
            st.inbox.push(message.clone());
        }
        // Read by `step` itself, over the whole batch; nothing to apply.
        EventIn::PauseRequested => {}
        EventIn::AskForwarded { tool_call_id } => {
            let Some(pending) = st.pending_asks.get(tool_call_id).cloned() else {
                return;
            };
            drop_bubble(st, &pending.key);
            st.node_state[pending.node_idx] = NodeState::Dispatched;
            let executor = env.executors[pending.node_idx].clone();
            let input = st.context.clone();
            let (superstep, clock_ms) = (st.superstep, st.clock_ms);
            out.push(Command::WriteIntent {
                key: pending.key.clone(),
                executor: executor.clone(),
                input: input.clone(),
                superstep,
                clock_ms,
            });
            out.push(Command::Dispatch { key: pending.key, executor, input });
        }
        EventIn::ResponseSettled { tool_call_id, outcome } => {
            let Some(pending) = st.pending_asks.remove(tool_call_id) else {
                // Unknown/settled asks are validated, journaled, and
                // refused by the driver (§6.6) before ever reaching here.
                return;
            };
            // Deliberately NO wall/elapsed bookkeeping here: the reading in
            // effect at response-apply is the driver's resume time, which is
            // NOT journaled — accruing from it would make checkpoint state a
            // function of when the operator typed `resume`, and verify could
            // never reproduce it. Both accruals happen at superstep close,
            // from the journaled close reading (see `close_superstep`); the
            // park→close tail is never charged as wall (it is µs of
            // bookkeeping), and elapsed = park→close, both ends journaled.
            st.announced_asks.remove(tool_call_id);
            st.node_state[pending.node_idx] = NodeState::Dispatched;
            resolve_effect(env, st, &pending.key, outcome);
        }
    }
}

/// Book a resolved effect: buffer the result, or retry / mark failure per
/// the §6.3 table. Retries are decided PER NODE, independent of other
/// nodes' failures — otherwise whether a retry happens would depend on
/// which failure arrived first (permutation-variant).
fn resolve_effect(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    key: &JournalKey,
    outcome: &EffectOutcome,
) {
    let Some(node_idx) = env.plan.nodes.iter().position(|n| *n == key.node) else {
        return;
    };
    {
        let Phase::Open { outstanding, .. } = &mut st.phase else {
            return; // Late/duplicate resolution of a closed superstep.
        };
        if !outstanding.remove(key) {
            return; // Duplicate delivery for this key: first-commit won (§6.6).
        }
    }
    // Accounting is per-effect, whatever the effect serves (§6.7: every
    // figure is a pure function of the journal).
    match outcome {
        EffectOutcome::Completed {
            journal_bytes,
            input_tokens,
            output_tokens,
            usd_micros,
            ..
        } => {
            st.spent.journal_grains += 2; // intent + result supersession
            st.spent.storage_bytes += journal_bytes;
            st.spent.input_tokens += input_tokens;
            st.spent.output_tokens += output_tokens;
            st.spent.usd_micros += usd_micros;
        }
        EffectOutcome::Failed { journal_bytes, .. } => {
            st.spent.journal_grains += 2;
            st.spent.storage_bytes += journal_bytes;
        }
    }

    // Abstract-flow effects continue the LOOP rather than resolving what
    // owns it — a node, or one Send task against an abstract target.
    let flow = flow_key(node_idx, &key.task_path);
    if st.abstract_flows.contains_key(&flow) {
        match key.kind {
            EffectKind::Llm => {
                handle_llm_outcome(env, st, node_idx, &key.task_path, outcome);
                return;
            }
            EffectKind::Tool
                if st.abstract_flows[&flow]
                    .deciding
                    .as_ref()
                    .is_some_and(|d| d.effect_seq() == key.effect_seq) =>
            {
                handle_flow_decide(env, st, node_idx, &key.task_path, outcome);
                return;
            }
            EffectKind::Tool
                if st.abstract_flows[&flow].pending_tools.contains_key(&key.effect_seq) =>
            {
                handle_flow_tool_outcome(
                    env,
                    st,
                    node_idx,
                    &key.task_path,
                    key.effect_seq,
                    outcome,
                );
                return;
            }
            _ => {}
        }
    }

    // Send-task effects resolve the TASK, never the target node directly —
    // the node completes when its batch drains.
    if !key.task_path.is_empty() {
        handle_send_task_outcome(env, st, key, outcome);
        return;
    }
    // A resolved node's straggler (a flow-tool result landing after
    // `fail_abstract` tore its flow down mid-round): accounted above, but it
    // must not flip a DoneFailed node back to DoneOk with a raw tool result.
    if matches!(
        st.node_state[node_idx],
        NodeState::DoneOk | NodeState::DoneFailed | NodeState::Dead
    ) {
        return;
    }

    if let EffectOutcome::Completed { result, .. } = outcome {
        if matches!(env.executors[node_idx], NodeExecutor::Subgraph { .. }) {
            if let Some(asks) = result
                .get(crate::types::PARKED_ASKS)
                .and_then(|v| serde_json::from_value::<Vec<Ask>>(v.clone()).ok())
                .filter(|a| !a.is_empty())
            {
                park_bubbled(st, node_idx, key, asks);
                return;
            }
        }
    }

    let attempts_used = st.attempt[node_idx] - st.attempt_base[node_idx];
    let retry_budget = env.plan.retries[node_idx];
    let canceling = st.cancel.is_some();
    match outcome {
        EffectOutcome::Completed { result, .. } => {
            st.node_state[node_idx] = NodeState::DoneOk;
            if let Phase::Open { results, .. } = &mut st.phase {
                results.insert(node_idx, result.clone());
            }
        }
        EffectOutcome::Failed { cause, detail, .. } => {
            let can_retry =
                cause.retryable(key.kind) && attempts_used <= retry_budget && !canceling;
            if can_retry {
                // Flip to Ready inside the open superstep; `progress`
                // dispatches the next attempt.
                st.node_state[node_idx] = NodeState::Ready;
            } else {
                st.node_state[node_idx] = NodeState::DoneFailed;
                let named = st.failed.as_ref().map(|(i, _)| *i).unwrap_or(usize::MAX);
                if node_idx < named {
                    st.failed = Some((node_idx, format!("{cause:?}: {detail}")));
                }
            }
        }
    }
}

/// The `abstract_flows` key. A node's own loop keys on its index alone, so
/// the serialized shape is exactly what it was before Send could target an
/// abstract node; a task's loop qualifies that with its path.
pub fn flow_key(node_idx: usize, task_path: &str) -> String {
    if task_path.is_empty() {
        node_idx.to_string()
    } else {
        format!("{node_idx}@{task_path}")
    }
}

fn flow_attempt(st: &SchedulerState, i: usize, path: &str) -> u32 {
    match st.send_tasks.get(path) {
        Some(task) => task.attempt,
        None => st.attempt[i],
    }
}

/// Open a fresh LLM loop for a node or one Send task against an abstract
/// node, and emit its first turn.
fn start_flow(env: &StepEnv<'_>, st: &mut SchedulerState, i: usize, path: &str, input: Value, out: &mut Vec<Command>) {
    // The task, as the node's `llm` block shapes it (#391). A pointer that
    // lands on nothing fails the node here, before any effect — the same
    // state would fail the same way on every retry, so it is not one.
    let content = match first_message(env, i, &input) {
        Ok(c) => c,
        Err(why) => {
            fail_abstract_coded(env, st, i, path, &format!("SchemaValidationFailed: {why}"));
            return;
        }
    };
    st.abstract_flows.insert(
        flow_key(i, path),
        crate::state::AbstractFlow {
            messages: vec![serde_json::json!({
                "role": "user",
                "content": content,
            })],
            next_effect_seq: 0,
            pending_tools: BTreeMap::new(),
            round_results: BTreeMap::new(),
            need: None,
            unknown_strikes: 0,
            last_prompt_tokens: 0,
            folding: None,
            fold_forced: false,
            folds: 0,
            deciding: None,
            decide_fold_tried: false,
            offer: None,
            response_strikes: 0,
            tool_log: Vec::new(),
        },
    );
    // C1: with a calibrated backend and more than OFFER_TOP_K pinned tools,
    // narrow the offer before the first turn. Anything that prevents asking
    // falls through to the full offer.
    if emit_offer_decide(env, st, i, path, out) {
        return;
    }
    dispatch_llm_turn(env, st, i, path, out);
}

// ---- the per-node `llm` block (#391, #392) -----------------------------------
//
// Everything here is a pure function of the frozen block and the node's input,
// and every choice it makes that reaches the provider RIDES THE JOURNAL — the
// first message in `messages[0]`, the request fields in each turn's `llm` —
// so the driver builds the request from the journaled input alone and verify
// needs nothing else. A node with no block takes none of these paths and
// builds byte-identical commands and state.

/// The node's frozen `llm` block, if it has one.
fn llm_cfg<'e>(env: &'e StepEnv<'_>, i: usize) -> Option<&'e LlmNodeConfig> {
    match &env.executors[i] {
        NodeExecutor::Abstract { llm, .. } => llm.as_deref(),
        _ => None,
    }
}

/// The per-dispatch token reservation for node `i`'s turns: its own
/// `max_tokens` when the block sets one (it is that turn's output ceiling, so
/// it is what must fit), else the run's.
fn reserve_for(env: &StepEnv<'_>, i: usize) -> u64 {
    llm_cfg(env, i)
        .and_then(|c| c.max_tokens)
        .map(u64::from)
        .unwrap_or(env.llm_reserve_tokens)
}

fn json_kind(v: &Value) -> &'static str {
    match v {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

/// `messages[0]`'s content: `{"instruction", "state"}`, shaped by the block.
fn first_message(env: &StepEnv<'_>, i: usize, input: &Value) -> Result<Value, String> {
    let cfg = llm_cfg(env, i);
    let instruction = match (cfg.and_then(|c| c.instruction.as_ref()), cfg.and_then(|c| c.instruction_from.as_ref())) {
        (Some(text), _) => text.clone(),
        (None, Some(ptr)) => match input.pointer(ptr) {
            Some(Value::String(text)) => text.clone(),
            Some(other) => {
                return Err(format!(
                    "instruction_from {ptr:?} resolves to a {} in the node's input, not a string",
                    json_kind(other)
                ))
            }
            None => return Err(format!("instruction_from {ptr:?} resolves to nothing in the node's input")),
        },
        (None, None) => env.plan.nodes[i].clone(),
    };
    let mut content = serde_json::Map::new();
    content.insert("instruction".into(), Value::String(instruction));
    // The tool list `tools_from` read is configuration, not state: under the
    // whole-state default it would put every entry — and every value the host
    // LOCKED away from the model — into the prompt. Removed from the copy the
    // first message carries; a key list naming it is the plan's explicit ask.
    let whole = cfg.and_then(|c| c.state.as_ref()).is_none_or(|s| s.as_str() == Some("all"));
    let pruned;
    let input = match cfg.and_then(|c| c.tools_from.as_deref()) {
        Some(ptr) if whole && !ptr.is_empty() => {
            let mut copy = input.clone();
            remove_pointer(&mut copy, ptr);
            pruned = copy;
            &pruned
        }
        _ => input,
    };
    if let Some(state) = select_state(cfg.and_then(|c| c.state.as_ref()), input) {
        content.insert("state".into(), state);
    }
    Ok(Value::Object(content))
}

/// Remove the value at a JSON pointer (RFC 6901, `~0`/`~1` unescaped), if any.
fn remove_pointer(v: &mut Value, ptr: &str) {
    let Some((parent, last)) = ptr.rsplit_once('/') else { return };
    let last = last.replace("~1", "/").replace("~0", "~");
    let target = if parent.is_empty() { Some(v) } else { v.pointer_mut(parent) };
    match target {
        Some(Value::Object(o)) => {
            o.remove(&last);
        }
        Some(Value::Array(a)) => {
            if let Ok(n) = last.parse::<usize>() {
                if n < a.len() {
                    a.remove(n);
                }
            }
        }
        _ => {}
    }
}

/// The merged state a node's first message carries. `"all"` (or no block):
/// the whole input. `"none"`: nothing. A key list: those keys that are
/// present, nothing else. A steering `$inbox` rides along in every case — an
/// operator's message to the run is not state the plan may filter out.
fn select_state(sel: Option<&Value>, input: &Value) -> Option<Value> {
    let keep = |keys: &[&str]| {
        let mut o = serde_json::Map::new();
        if let Value::Object(src) = input {
            for k in keys.iter().copied().chain([crate::types::INBOX]) {
                if let Some(v) = src.get(k) {
                    o.insert(k.to_string(), v.clone());
                }
            }
        }
        o
    };
    match sel {
        None => Some(input.clone()),
        Some(Value::String(s)) if s == "all" => Some(input.clone()),
        Some(Value::String(s)) if s == "none" => {
            let o = keep(&[]);
            (!o.is_empty()).then_some(Value::Object(o))
        }
        Some(Value::Array(keys)) => {
            let keys: Vec<&str> = keys.iter().filter_map(Value::as_str).collect();
            Some(Value::Object(keep(&keys)))
        }
        // The manifest refused every other shape at start; a hand-edited one
        // gets the whole state rather than a guess.
        Some(_) => Some(input.clone()),
    }
}

/// The wire name of a node's response schema: what OpenAI accepts
/// (`^[a-zA-Z0-9_-]{1,64}$`), derived from the node id.
fn response_name(node: &str) -> String {
    let safe: String = node
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() || c == '_' || c == '-' { c } else { '_' })
        .take(64)
        .collect();
    if safe.is_empty() { "response".into() } else { safe }
}

/// The request fields a node's block sets, as journaled on each of its turns
/// under `llm` — `response_format` already in the OpenAI-compatible wire
/// shape, so the journal shows what was asked for as it went out. A fold's
/// summarizer turn takes the node's system text, temperature and ceiling (a
/// GPT-5 node's fold must not 400 on a temperature either) but neither the
/// schema nor `parallel_tool_calls`: it answers in prose and is offered no
/// tools. `None` when nothing is set, so such a turn is byte-identical to one
/// from a node with no block.
fn request_block(env: &StepEnv<'_>, i: usize, fold: bool, instruction: &str) -> Option<Value> {
    let c = llm_cfg(env, i)?;
    let mut b = serde_json::Map::new();
    if let Some(s) = &c.system {
        // OpenAI's `json_object` mode refuses (HTTP 400) a request in which
        // no message says "JSON". The built-in system prompt does; a node's
        // own may not, and neither may its instruction — so say it once.
        let wants_json_object = !fold && c.response.as_ref().is_some_and(|r| r.get("json").is_some());
        let says_json = |t: &str| t.to_ascii_lowercase().contains("json");
        let s = if wants_json_object && !says_json(s) && !says_json(instruction) {
            format!("{s} Answer with a JSON object.")
        } else {
            s.clone()
        };
        b.insert("system".into(), Value::String(s));
    }
    if let Some(t) = &c.temperature {
        b.insert("temperature".into(), t.clone());
    }
    if let Some(m) = c.max_tokens {
        b.insert("max_tokens".into(), Value::from(m));
    }
    if !fold {
        if let Some(p) = c.parallel_tool_calls {
            b.insert("parallel_tool_calls".into(), Value::Bool(p));
        }
        if let Some(r) = &c.response {
            let wire = match r.get("json_schema") {
                Some(schema) => serde_json::json!({
                    "type": "json_schema",
                    "json_schema": {
                        "name": response_name(&env.plan.nodes[i]),
                        "schema": schema,
                        "strict": true,
                    },
                }),
                None => serde_json::json!({ "type": "json_object" }),
            };
            b.insert("response_format".into(), wire);
        }
    }
    (!b.is_empty()).then_some(Value::Object(b))
}

/// Check a final answer against the node's `response` declaration: a JSON
/// object, conforming to the schema when one is declared. The runtime checks
/// whatever the provider did with `response_format`, because a provider that
/// ignored the field must not hand prose through as the schema's object.
/// The answer inside a Markdown code fence, when the whole text is one —
/// ```` ```json … ``` ```` or ```` ``` … ``` ````. Anthropic models wrap a JSON
/// answer that way even when told not to. Anything else is returned trimmed.
fn unfence(text: &str) -> &str {
    let t = text.trim();
    let Some(rest) = t.strip_prefix("```") else { return t };
    let Some(body) = rest.strip_suffix("```") else { return t };
    // The info string runs to the first newline (`json`, `JSON`, or nothing).
    match body.split_once('\n') {
        Some((info, inner)) if !info.contains('{') && !info.contains('[') => inner.trim(),
        _ => body.trim(),
    }
}

fn check_response(response: &Value, text: &str) -> Result<(), String> {
    let value: Value = serde_json::from_str(unfence(text))
        .map_err(|e| format!("the answer is not JSON ({e})"))?;
    if !value.is_object() {
        return Err(format!("the answer is a JSON {}, not an object", json_kind(&value)));
    }
    if let Some(schema) = response.get("json_schema") {
        areev_core::types::validate_instance(&value, schema)
            .map_err(|k| format!("schema violation: {}", k.as_str()))?;
    }
    Ok(())
}

/// Validate one model-issued call and build the input it dispatches with
/// (#392, #393). A dynamic tool's arguments are checked against ITS schema,
/// always — that is the contract the host declared for it. Then the locks are
/// merged, and the strict-argument table judges the MERGED input under the
/// Definition the call reaches, so a key the model was never shown (stripped
/// from its schema because locked) is not reported missing.
fn validated_dispatch_input(
    env: &StepEnv<'_>,
    tool: &OfferedTool,
    arguments: &Value,
) -> Result<Value, String> {
    if let Some(d) = &tool.dynamic {
        // The entry's schema is CLOSED (#392 review): a key it does not name
        // could otherwise set a Definition parameter the host never offered
        // — `method: "DELETE"` on an http-shaped Definition that locks only
        // the URL. A key the host LOCKED is refused outright rather than
        // silently overridden, so the model learns it cannot choose it.
        let schema = d.model_schema(tool.locked.as_ref());
        if let Value::Object(args) = arguments {
            if d.dispatch.args_into.is_none() {
                if let Some(Value::Object(lock)) = &tool.locked {
                    if let Some(k) = args.keys().find(|k| lock.contains_key(k.as_str())) {
                        return Err(format!(
                            "argument '{k}' is fixed by the host and may not be set"
                        ));
                    }
                }
            }
            // Closed only where the arguments land at the Definition's TOP
            // LEVEL (no `args_into`): there an unnamed key is a Definition
            // parameter the host never offered. Wrapped under `args_into`,
            // the arguments are the remote tool's own, and the entry's schema
            // speaks for itself — an MCP tool's `{"type": "object"}` takes
            // any key, and only `additionalProperties: false` closes it.
            let closed = d.dispatch.args_into.is_none()
                || schema.get("additionalProperties") == Some(&Value::Bool(false));
            let named = schema.get("properties").and_then(Value::as_object);
            if closed {
                if let Some(k) = args.keys().find(|k| !named.is_some_and(|p| p.contains_key(k.as_str()))) {
                    return Err(format!("argument '{k}' is not one of this tool's parameters"));
                }
            }
        }
        // Against what the model was SHOWN — locked keys stripped — so a
        // locked key the entry also lists as required is not demanded of it.
        areev_core::types::validate_instance(arguments, &schema)
            .map_err(|k| format!("schema violation: {}", k.as_str()))?;
    }
    let input = tool.dispatch_input(arguments);
    if let Some(schema) = tool.dynamic.as_ref().and_then(|d| d.dispatch.definition_schema.as_ref()) {
        areev_core::types::validate_instance(&input, schema)
            .map_err(|k| format!("schema violation: {} (the call as dispatched)", k.as_str()))?;
    }
    (env.validate_args)(tool.dispatch_name(), &input)?;
    Ok(input)
}

fn flow_owners(st: &SchedulerState) -> Vec<(usize, String)> {
    let mut owners: Vec<(usize, String)> = st
        .abstract_flows
        .iter()
        .filter(|(_, f)| f.need.is_some())
        .filter_map(|(k, _)| match k.split_once('@') {
            Some((i, path)) => Some((i.parse().ok()?, path.to_string())),
            None => Some((k.parse().ok()?, String::new())),
        })
        .collect();
    owners.sort();
    owners
}

fn apply_inbox(st: &mut SchedulerState) {
    let Value::Object(o) = &mut st.context else {
        return;
    };
    if st.inbox.is_empty() {
        o.remove(crate::types::INBOX);
    } else {
        o.insert(crate::types::INBOX.into(), Value::Array(std::mem::take(&mut st.inbox)));
    }
}

fn drop_bubble(st: &mut SchedulerState, key: &JournalKey) {
    let ids: Vec<String> = st
        .pending_asks
        .iter()
        .filter(|(_, p)| p.key == *key)
        .map(|(id, _)| id.clone())
        .collect();
    for id in ids {
        st.pending_asks.remove(&id);
        st.announced_asks.remove(&id);
    }
}

fn park_bubbled(st: &mut SchedulerState, i: usize, resolved: &JournalKey, asks: Vec<Ask>) {
    if !matches!(st.phase, Phase::Open { .. }) {
        return;
    }
    let stale: Vec<JournalKey> = st
        .pending_asks
        .values()
        .filter(|p| p.node_idx == i)
        .map(|p| p.key.clone())
        .collect();
    for key in stale {
        drop_bubble(st, &key);
    }
    // The next ROUND of the same attempt, never the next attempt: the child
    // run id is derived from the attempt, so bumping it here would forward
    // the answered ask into a brand-new child. Holding `attempt` still also
    // means a park costs no retry budget without touching `attempt_base`.
    let key = JournalKey { effect_seq: resolved.effect_seq + 1, ..resolved.clone() };
    if let Phase::Open { outstanding, .. } = &mut st.phase {
        outstanding.insert(key.clone());
    }
    st.node_state[i] = NodeState::AwaitingClient;
    for ask in asks {
        st.pending_asks.insert(
            ask.tool_call_id.clone(),
            PendingAsk { key: key.clone(), node_idx: i, ask },
        );
    }
}

/// A Send task's effect resolved: buffer the result (or retry / fail per
/// the same §6.3 table nodes use), and complete the target node when its
/// batch has drained — the join below a fan-out.
fn handle_send_task_outcome(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    key: &JournalKey,
    outcome: &EffectOutcome,
) {
    let Some(task) = st.send_tasks.get_mut(&key.task_path) else { return };
    // A settled task's straggler — an abstract target's flow tool landing
    // after `fail_abstract` tore the loop down. Accounted above, but it must
    // not overwrite the task's contribution with a raw tool result the model
    // never saw. The node-level guard below does the same job for nodes.
    if task.state == crate::state::SendTaskState::Done {
        return;
    }
    let target = task.node_idx;
    match outcome {
        EffectOutcome::Completed { result, .. } => {
            settle_send_task(st, &key.task_path, result.clone());
        }
        EffectOutcome::Failed { cause, detail, .. } => {
            let can_retry = cause.retryable(key.kind)
                && task.attempt <= env.plan.retries[target]
                && st.cancel.is_none();
            if can_retry {
                task.state = crate::state::SendTaskState::Queued;
                return;
            }
            task.state = crate::state::SendTaskState::Done;
            st.node_state[target] = NodeState::DoneFailed;
            let named = st.failed.as_ref().map(|(i, _)| *i).unwrap_or(usize::MAX);
            if target < named {
                st.failed = Some((
                    target,
                    format!("{cause:?}: {detail} (task {})", key.task_path),
                ));
            }
        }
    }
}

/// One task settled with a result: buffer it, then complete the target node
/// if its whole batch has drained — the join below a fan-out.
fn settle_send_task(st: &mut SchedulerState, path: &str, result: Value) {
    let Some(task) = st.send_tasks.get_mut(path) else { return };
    let target = task.node_idx;
    task.state = crate::state::SendTaskState::Done;
    if let Phase::Open { send_results, .. } = &mut st.phase {
        send_results.insert(path.to_string(), result);
    }
    let all_done = st
        .send_tasks
        .values()
        .filter(|t| t.node_idx == target)
        .all(|t| t.state == crate::state::SendTaskState::Done);
    if all_done && st.node_state[target] == NodeState::Dispatched {
        st.node_state[target] = NodeState::DoneOk;
        if let Phase::Open { results, .. } = &mut st.phase {
            results.entry(target).or_insert(Value::Null);
        }
    }
}

/// Fold a model-supplied tool name back onto the **canonical** `tool_name` of
/// the Definition it was offered as.
///
/// Anthropic and OpenAI forbid `.` in a tool name, so a pinned Definition
/// called `receipt.prepare` reaches the model as `receipt_prepare`
/// ([`normalize_tool_name`]) — and the model, correctly, calls back with what
/// it was given. Without this fold every call to a dotted tool from inside an
/// abstract node read as an unknown tool and failed the node after one
/// corrective re-prompt the model could not possibly satisfy.
///
/// Exact match wins, so a plan whose offered set contains BOTH `a.b` and `a_b`
/// still resolves `a_b` to the tool that is literally named that, and an
/// ambiguous normalized match resolves to nothing rather than to a guess — the
/// caller then treats it as unknown and re-prompts, which is the honest answer.
/// A name matching neither form is returned unchanged, for the same reason: the
/// unknown-tool path owns that decision, not this function.
fn canonical_tool_name(offered: &[OfferedTool], called: &str) -> String {
    if offered.iter().any(|t| t.tool_name == called) {
        return called.to_string();
    }
    let mut hits = offered.iter().filter(|t| normalize_tool_name(&t.tool_name) == called);
    match (hits.next(), hits.next()) {
        (Some(t), None) => t.tool_name.clone(),
        _ => called.to_string(),
    }
}

/// A model turn resolved: end the loop, request tools, or re-prompt.
fn handle_llm_outcome(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    outcome: &EffectOutcome,
) {
    let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return };
    let flow_id = flow_key(i, path);
    // A summarizer turn resolves differently from a model turn: its result is
    // not the node's answer, and any tool calls in it are ignored (it was
    // offered none). Taken here so a retryable failure also clears the marker
    // and `dispatch_llm_turn` re-derives the decision from the same transcript.
    let folding = st.abstract_flows.get_mut(&flow_id).and_then(|f| f.folding.take());
    if let Some(f) = folding {
        resolve_fold(env, st, i, path, f, outcome);
        return;
    }
    // The offer the model actually saw: the narrowed set when C1 narrowed it,
    // else every pinned tool. A pinned tool the model was NOT shown is
    // unknown to it — narrowing may only remove, never re-admit.
    let narrowed = narrowed_offer(st.abstract_flows.get(&flow_id), tools);
    let tools: &[OfferedTool] = &narrowed;
    let offered: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect();
    match outcome {
        EffectOutcome::Failed { cause, detail, .. } => match cause {
            // Transient model failures retry the TURN (same transcript,
            // next effect_seq) — idempotent and journal-clean; schema
            // failures re-prompt with the error appended (§6.11). Both are
            // bounded by max_effects_per_attempt.
            FailCause::Timeout | FailCause::ExecutorError => {
                if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                    flow.need = Some(crate::state::FlowNeed::NextTurn);
                }
            }
            FailCause::SchemaValidationFailed => {
                if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                    flow.messages.push(serde_json::json!({
                        "role": "user",
                        "content": format!(
                            "Your previous output failed validation: {detail}. \
                             Correct it and answer again."
                        ),
                    }));
                    flow.need = Some(crate::state::FlowNeed::NextTurn);
                }
            }
            // The provider itself said the prompt was too long. Retrying is
            // pointless and failing is premature: fold, then re-send the same
            // turn. `dispatch_llm_turn` re-derives everything from the
            // (unchanged) transcript, and the effect cap bounds this exactly
            // as it bounds every other loop in here — a node that cannot get
            // under the limit runs out of effects rather than spinning.
            FailCause::ContextOverflow => {
                if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                    flow.fold_forced = true;
                    flow.need = Some(crate::state::FlowNeed::NextTurn);
                }
            }
            FailCause::UserAborted | FailCause::Unknown => {
                fail_abstract(env, st, i, path, detail);
            }
        },
        EffectOutcome::Completed { result, input_tokens, .. } => {
            // The transcript's size at the MODEL's own tokenizer, for the
            // context ceiling to measure on the next dispatch. The provider
            // reported it for the transcript exactly as sent, which is why the
            // runtime never estimates.
            if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                flow.last_prompt_tokens = *input_tokens;
            }
            let text = result.get("text").and_then(|t| t.as_str()).map(str::to_string);
            // The provider's own content blocks for this turn (#284), carried
            // straight through: the scheduler never inspects them, it only
            // has to keep them so the adapter can echo the turn back
            // unchanged on the next dispatch. Absent for every provider that
            // has none, so the transcript is byte-identical to 1.8.5 there.
            let provider_content = result.get("provider_content").cloned();
            let calls: Vec<crate::state::PendingToolCall> = result
                .get("tool_calls")
                .and_then(|c| c.as_array())
                .map(|arr| {
                    arr.iter()
                        .map(|c| crate::state::PendingToolCall {
                            model_call_id: c
                                .get("id")
                                .and_then(|v| v.as_str())
                                .unwrap_or_default()
                                .to_string(),
                            // The model answers with the name it was
                            // OFFERED, which for a dotted Definition is the
                            // normalized spelling. Canonicalize here, once,
                            // so the unknown-tool check, the argument
                            // validator and the dispatch below all see the
                            // Definition's own `tool_name`.
                            tool_name: canonical_tool_name(
                                tools,
                                c.get("name").and_then(|v| v.as_str()).unwrap_or_default(),
                            ),
                            arguments: c.get("arguments").cloned().unwrap_or(Value::Null),
                        })
                        .collect()
                })
                .unwrap_or_default();

            // Unknown tool names: ONE corrective re-prompt, then the node
            // fails `ExecutorError` (§6.11) — the model must choose among
            // what was offered, and a model that cannot is not looped on.
            // `canonical_tool_name` already folded the offered spelling back,
            // so what survives here matches NEITHER form.
            let unknown: Vec<String> = calls
                .iter()
                .map(|c| c.tool_name.clone())
                .filter(|n| !offered.contains(&n.as_str()))
                .collect();
            if !unknown.is_empty() {
                let strikes = st
                    .abstract_flows
                    .get(&flow_id)
                    .map(|f| f.unknown_strikes)
                    .unwrap_or(0);
                if strikes >= 1 {
                    fail_abstract(
                        env,
                        st,
                        i,
                        path,
                        &format!("model called unknown tool(s) {unknown:?} after a corrective re-prompt"),
                    );
                    return;
                }
                let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
                flow.unknown_strikes += 1;
                // The list names the tools the way the MODEL was offered
                // them: it has to emit one of these, and handing it a dotted
                // name its own provider forbids is a correction that cannot
                // be obeyed. Both spellings are accepted on the way back.
                let choices: Vec<String> =
                    offered.iter().map(|n| normalize_tool_name(n)).collect();
                flow.messages.push(serde_json::json!({
                    "role": "user",
                    "content": format!(
                        "Unknown tool(s) {:?}; choose among: {:?}",
                        unknown, choices
                    ),
                }));
                flow.need = Some(crate::state::FlowNeed::NextTurn);
                return;
            }

            // Argument validation (§6.11): unparseable argument payloads and
            // strict-schema violations re-prompt with the error appended —
            // consuming `effect_seq`, bounded by `max_effects_per_attempt`.
            // The check runs on the input the call will DISPATCH with —
            // locks merged (#393), a dynamic tool's own schema first (#392) —
            // so a strict Definition is judged on what it will receive.
            let invalid: Option<(String, String)> = calls.iter().find_map(|c| {
                if !c.arguments.is_object() {
                    return Some((
                        c.tool_name.clone(),
                        "arguments are not a JSON object".to_string(),
                    ));
                }
                let tool = tools.iter().find(|t| t.tool_name == c.tool_name)?;
                validated_dispatch_input(env, tool, &c.arguments)
                    .err()
                    .map(|e| (c.tool_name.clone(), e))
            });
            let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
            if let Some((tool, why)) = invalid {
                flow.messages.push(serde_json::json!({
                    "role": "user",
                    "content": format!(
                        "Arguments for tool '{tool}' failed validation ({why}). \
                         Correct the call and try again."
                    ),
                }));
                flow.need = Some(crate::state::FlowNeed::NextTurn);
                return;
            }
            if calls.is_empty() {
                // The loop ends: the final text is the node's result — a
                // JSON object merges as one; anything else lands under the
                // node's own key.
                let final_text = text.unwrap_or_default();
                // Structured output (#391): the answer must be what the block
                // asked for. ONE corrective re-prompt, with the model's own
                // answer kept in the transcript so it can see what it said,
                // then the node fails — never a silent merge of prose.
                let cfg = llm_cfg(env, i);
                if let Some(response) = cfg.and_then(|c| c.response.as_ref()) {
                    if let Err(why) = check_response(response, &final_text) {
                        if flow.response_strikes >= 1 {
                            fail_abstract_coded(
                                env,
                                st,
                                i,
                                path,
                                &format!(
                                    "SchemaValidationFailed: the final answer failed the node's \
                                     response declaration after a corrective re-prompt ({why})"
                                ),
                            );
                            return;
                        }
                        flow.response_strikes += 1;
                        let mut entry = serde_json::json!({
                            "role": "assistant",
                            "text": final_text,
                            "tool_calls": [],
                        });
                        if let Some(pc) = provider_content {
                            entry["provider_content"] = pc;
                        }
                        flow.messages.push(entry);
                        flow.messages.push(serde_json::json!({
                            "role": "user",
                            "content": format!(
                                "Your final answer failed validation: {why}. Answer again with \
                                 only a JSON object that satisfies the required response format."
                            ),
                        }));
                        flow.need = Some(crate::state::FlowNeed::NextTurn);
                        return;
                    }
                }
                // A declared response was checked on the unfenced text, so it
                // merges from the same text; a node with no declaration merges
                // exactly what it always did.
                let merge_text = if cfg.is_some_and(|c| c.response.is_some()) {
                    unfence(&final_text).to_string()
                } else {
                    final_text.clone()
                };
                let mut value = serde_json::from_str::<Value>(&merge_text)
                    .ok()
                    .filter(|v| v.is_object())
                    .unwrap_or_else(|| {
                        serde_json::json!({ env.plan.nodes[i].clone(): final_text })
                    });
                // What the model read, for whatever is bound below this node
                // (#391): every tool call of the attempt, off the journaled
                // outcomes. The runtime's key wins over one the model wrote.
                if let Some(key) = cfg.and_then(|c| c.tool_results_into.as_ref()) {
                    if let Value::Object(o) = &mut value {
                        o.insert(key.clone(), Value::Array(std::mem::take(&mut flow.tool_log)));
                    }
                }
                st.abstract_flows.remove(&flow_id);
                if path.is_empty() {
                    st.node_state[i] = NodeState::DoneOk;
                    if let Phase::Open { results, .. } = &mut st.phase {
                        results.insert(i, value);
                    }
                } else {
                    settle_send_task(st, path, value);
                }
                return;
            }
            // Tool round: the assistant entry is journal-derived, so the
            // transcript is replay-stable.
            let mut entry = serde_json::json!({
                "role": "assistant",
                "text": text,
                "tool_calls": calls.iter().map(|c| serde_json::json!({
                    "id": c.model_call_id,
                    "name": c.tool_name,
                    "arguments": c.arguments,
                })).collect::<Vec<_>>(),
            });
            if let Some(pc) = provider_content {
                entry["provider_content"] = pc;
            }
            flow.messages.push(entry);
            flow.round_results.clear();
            flow.need = Some(crate::state::FlowNeed::Tools(calls));
        }
    }
}

/// Bound ONE tool result for the TRANSCRIPT. The journal is untouched: the
/// driver wrote the full result grain before this event ever reached `step`,
/// so what is dropped here is dropped from scheduler state only and stays
/// addressable at `(run, task_path, node, attempt, effect_seq)` forever.
///
/// `cap` is in **characters**, and is named that because that is what it is.
/// This crate holds no tokenizer, and a "token" bound implemented as
/// `chars / 4` would be a guess wearing a precise name. It is also why this is
/// a separate bound from the context ceiling, which uses the provider's own
/// reported prompt tokens: one oversized entry is a different problem from a
/// transcript that grew, and no fold can shrink a single entry.
///
/// Under the cap the value passes through **byte-identically**, so a run with
/// no cap configured builds exactly the transcript it always did. Over it, the
/// model sees the head, the tail, the true length, and where the whole thing
/// lives. Pure and total: a function of the journaled outcome and the manifest,
/// so replay rebuilds the same entry.
pub fn bound_tool_content(
    raw: &Value,
    cap: Option<usize>,
    attempt: u32,
    effect_seq: u32,
) -> Value {
    let Some(cap) = cap else { return raw.clone() };
    // A string result is measured as its text; anything else as the JSON the
    // model would have been shown.
    let text = match raw {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    };
    let total = text.chars().count();
    if total <= cap {
        return raw.clone();
    }
    let half = cap / 2;
    serde_json::json!({
        "truncated": true,
        "chars": total,
        "head": text.chars().take(half).collect::<String>(),
        "tail": text.chars().skip(total - half).collect::<String>(),
        "journal": { "attempt": attempt, "effect_seq": effect_seq },
    })
}

/// A summarizer turn resolved: splice its result over the range it stood in
/// for, and hand the loop back to an ordinary next turn.
///
/// The splice is an edit to **scheduler state only**. Every entry it removes
/// remains a journaled intent + result grain — nothing is deleted, nothing is
/// rewritten, and the fold message says where the record is. That is the root
/// `CLAUDE.md`'s immutability invariant applied to the runtime.
fn resolve_fold(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    f: FoldInFlight,
    outcome: &EffectOutcome,
) {
    let flow_id = flow_key(i, path);
    match outcome {
        EffectOutcome::Completed { result, .. } => {
            let summary =
                result.get("text").and_then(Value::as_str).unwrap_or_default().to_string();
            let attempt = flow_attempt(st, i, path);
            let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
            // The range was computed against this very transcript and nothing
            // has touched it since — but a scheduler must not panic on a range
            // it cannot apply. Dropping the fold and taking an ordinary turn
            // re-derives the decision on the next dispatch.
            if f.from >= f.to || f.to > flow.messages.len() {
                flow.need = Some(crate::state::FlowNeed::NextTurn);
                return;
            }
            let n = flow.folds + 1;
            let fold_msg = serde_json::json!({
                "role": "user",
                "content": format!(
                    "[Areev fold {n}: transcript entries {}..{} of attempt {attempt} were \
                     replaced by this summary, journaled at effect_seq {}. The full record \
                     — every turn and every tool result — is in this run's journal.]\n{summary}",
                    f.from, f.to, f.effect_seq,
                ),
            });
            flow.messages.splice(f.from..f.to, [fold_msg]);
            flow.folds = n;
            // The summarizer answered the trigger; a later one may try a
            // decision first again.
            flow.decide_fold_tried = false;
            // The spliced transcript has not been measured yet. Zero means "ask
            // the provider again", not "it is small" — the next real turn's
            // reported prompt tokens decide whether another fold is due.
            flow.last_prompt_tokens = 0;
            flow.need = Some(crate::state::FlowNeed::NextTurn);
            // Run-level too, once the flow borrow is done: this flow is gone
            // by the terminal checkpoint, and that is where the run-outcome
            // record is written from.
            st.folds = st.folds.saturating_add(1);
        }
        EffectOutcome::Failed { cause, detail, .. } => match cause {
            // Transient: retry the TURN, which re-derives the fold decision
            // from the unchanged transcript. Bounded by the effect cap like
            // every other retry in the loop, so it cannot spin.
            FailCause::Timeout | FailCause::ExecutorError => {
                if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                    flow.need = Some(crate::state::FlowNeed::NextTurn);
                }
            }
            // Even the SUMMARIZER's own request can overflow — it carries the
            // range it is summarizing. Re-deriving the fold from the same
            // transcript would ask the identical question again, so this is
            // terminal: the range that has to shrink is the one thing a fold
            // cannot shrink for itself.
            FailCause::ContextOverflow => {
                let detail = RunError::ContextExceeded {
                    node: env.plan.nodes[i].clone(),
                    tokens: st
                        .abstract_flows
                        .get(&flow_id)
                        .map(|f| f.last_prompt_tokens)
                        .unwrap_or(0),
                    ceiling: env.llm_context_tokens,
                }
                .to_string();
                fail_abstract_coded(env, st, i, path, &detail);
            }
            // Terminal, including a schema failure: there is no schema on a
            // summarizer turn, so a corrective re-prompt would be theatre.
            FailCause::Unknown
            | FailCause::UserAborted
            | FailCause::SchemaValidationFailed => {
                fail_abstract(env, st, i, path, detail);
            }
        },
    }
}

/// A model-issued tool call resolved. Tool failures inside an abstract loop
/// are MODEL-VISIBLE error results, never scheduler retries — the model
/// decides what to do about its own tool's failure.
fn handle_flow_tool_outcome(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    effect_seq: u32,
    outcome: &EffectOutcome,
) {
    // Read before the flow is borrowed mutably — the attempt is what makes the
    // truncation's journal pointer resolvable.
    let attempt = flow_attempt(st, i, path);
    let cap = env.llm_tool_result_chars;
    let logging = llm_cfg(env, i).is_some_and(|c| c.tool_results_into.is_some());
    let journal_id = JournalKey {
        run_id: st.run_id.clone(),
        task_path: path.to_string(),
        node: env.plan.nodes[i].clone(),
        attempt,
        effect_seq,
        kind: EffectKind::Tool,
    }
    .tool_call_id();
    let Some(flow) = st.abstract_flows.get_mut(&flow_key(i, path)) else { return };
    let Some(call) = flow.pending_tools.remove(&effect_seq) else { return };
    if logging {
        // `tool_results_into` (#391): the record a downstream gate reads. The
        // input is what was DISPATCHED (locks merged), the result is the WHOLE
        // journaled result — never the transcript's bounded excerpt — and the
        // entry is placed by `effect_seq`, so arrival order cannot reorder it.
        let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return };
        let input = tools
            .iter()
            .find(|t| t.tool_name == call.tool_name)
            .map(|t| t.dispatch_input(&call.arguments))
            .unwrap_or_else(|| call.arguments.clone());
        let (result, is_error) = match outcome {
            EffectOutcome::Completed { result, .. } => (result.clone(), false),
            EffectOutcome::Failed { cause, detail, .. } => {
                (Value::String(format!("{cause:?}: {detail}")), true)
            }
        };
        let entry = serde_json::json!({
            "name": call.tool_name,
            "tool_call_id": journal_id,
            "model_call_id": call.model_call_id,
            "input": input,
            "result": result,
            "is_error": is_error,
            "effect_seq": effect_seq,
        });
        let at = flow
            .tool_log
            .iter()
            .position(|e| e.get("effect_seq").and_then(Value::as_u64).unwrap_or(0) > u64::from(effect_seq))
            .unwrap_or(flow.tool_log.len());
        flow.tool_log.insert(at, entry);
    }
    let entry = match outcome {
        EffectOutcome::Completed { result, .. } => serde_json::json!({
            "role": "tool",
            "tool_call_id": call.model_call_id,
            "content": bound_tool_content(result, cap, attempt, effect_seq),
            "is_error": false,
        }),
        // Failure details are short by construction, but they carry a tool's
        // stderr — so the same rule applies rather than one honest path and one
        // hopeful one.
        EffectOutcome::Failed { cause, detail, .. } => serde_json::json!({
            "role": "tool",
            "tool_call_id": call.model_call_id,
            "content": bound_tool_content(
                &Value::String(format!("{cause:?}: {detail}")),
                cap,
                attempt,
                effect_seq,
            ),
            "is_error": true,
        }),
    };
    flow.round_results.insert(effect_seq, entry);
    if flow.pending_tools.is_empty() {
        // The round is settled: append results in effect_seq order —
        // canonical transcript order regardless of completion order.
        let entries: Vec<Value> = flow.round_results.values().cloned().collect();
        flow.messages.extend(entries);
        flow.round_results.clear();
        flow.need = Some(crate::state::FlowNeed::NextTurn);
    }
}

fn progress(env: &StepEnv<'_>, st: &mut SchedulerState, out: &mut Vec<Command>, hold: bool) {
    loop {
        if st.is_terminal() {
            return;
        }
        if matches!(st.phase, Phase::Open { .. }) {
            if progress_open(env, st, out) {
                continue;
            }
            return;
        }
        if !progress_idle(env, st, out, hold) {
            return;
        }
    }
}

/// One pass over an open superstep. Returns true when the caller should
/// loop (state changed), false to yield for external events.
fn progress_open(env: &StepEnv<'_>, st: &mut SchedulerState, out: &mut Vec<Command>) -> bool {
    let draining = st.cancel.is_some() || st.exhausted.is_some();

    // Abstract flows with settled needs: emit their next effects in
    // canonical node order (never emission-time order).
    if !draining {
        let needy = flow_owners(st);
        if !needy.is_empty() {
            for (i, path) in needy {
                let need =
                    st.abstract_flows.get_mut(&flow_key(i, &path)).and_then(|f| f.need.take());
                match need {
                    Some(crate::state::FlowNeed::NextTurn) => {
                        dispatch_llm_turn(env, st, i, &path, out);
                    }
                    Some(crate::state::FlowNeed::Tools(calls)) => {
                        for call in calls {
                            dispatch_flow_tool(env, st, i, &path, call, out);
                        }
                    }
                    None => {}
                }
            }
            return true;
        }
    }

    // Retries (nodes flipped back to Ready inside the open step).
    let retry_nodes: Vec<usize> = (0..env.plan.nodes.len())
        .filter(|&i| st.node_state[i] == NodeState::Ready)
        .collect();
    if !retry_nodes.is_empty() && !draining {
        for i in retry_nodes {
            dispatch_node(env, st, i, out);
        }
        return true;
    }

    // Send-task retries (tasks flipped back to Queued inside the open step).
    let retry_tasks: Vec<String> = st
        .send_tasks
        .iter()
        .filter(|(_, t)| t.state == crate::state::SendTaskState::Queued)
        .map(|(p, _)| p.clone())
        .collect();
    if !retry_tasks.is_empty() && !draining {
        for p in retry_tasks {
            dispatch_send_task(env, st, &p, out);
        }
        return true;
    }

    let (outstanding_empty, only_asks_left) = match &st.phase {
        Phase::Open { outstanding, .. } => {
            let only_asks = !outstanding.is_empty()
                && outstanding
                    .iter()
                    .all(|k| st.pending_asks.values().any(|p| p.key == *k));
            (outstanding.is_empty(), only_asks)
        }
        _ => unreachable!("progress_open requires Phase::Open"),
    };

    if outstanding_empty || (st.cancel.is_some() && only_asks_left) {
        if st.cancel.is_some() && only_asks_left {
            // A canceled run does not wait for humans: abandon the asks.
            if let Phase::Open { outstanding, .. } = &mut st.phase {
                outstanding.clear();
            }
            st.pending_asks.clear();
            st.announced_asks.clear();
        }
        close_superstep(env, st, out);
        return true;
    }
    if only_asks_left {
        // Pause the wall segment: parked time is elapsed, never wall.
        if let Some(open) = st.wall_open.take() {
            st.spent.wall_ms += st.clock_ms.saturating_sub(open);
            st.paused_at = Some(st.clock_ms);
        }
        park(st, out);
    }
    false
}

/// One pass over the idle phase. Returns true to keep looping.
fn progress_idle(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    out: &mut Vec<Command>,
    hold: bool,
) -> bool {
    if let Some((by, reason)) = st.cancel.clone() {
        finish(st, out, RunOutcome::Canceled { by, reason });
        return false;
    }
    if let Some(axis) = st.exhausted {
        finish(st, out, RunOutcome::BudgetExhausted { axis });
        return false;
    }
    if let Some((idx, detail)) = st.failed.clone() {
        finish(
            st,
            out,
            RunOutcome::Failed { node: env.plan.nodes[idx].clone(), detail },
        );
        return false;
    }
    let ready: Vec<usize> = (0..env.plan.nodes.len())
        .filter(|&i| st.node_state[i] == NodeState::Ready)
        .collect();
    let queued_tasks: Vec<String> = st
        .send_tasks
        .iter()
        .filter(|(_, t)| t.state == crate::state::SendTaskState::Queued)
        .map(|(p, _)| p.clone())
        .collect();
    // Abstract flows whose next turn survived a reservation refusal: a
    // fork with raised budgets resumes exactly here (the pinned promise).
    let needy_flows = flow_owners(st);
    if ready.is_empty() && queued_tasks.is_empty() && needy_flows.is_empty() {
        // A step call before any Start is a driver protocol slip, not a
        // stall — yield; terminalizing would brick the run id.
        if st.superstep == 0 && st.node_state[0] == NodeState::Waiting {
            return false;
        }
        // Defensive: an in-flight node at Idle means a driver protocol bug,
        // not a stall — yield rather than misreport.
        if st
            .node_state
            .iter()
            .any(|s| matches!(s, NodeState::Dispatched | NodeState::AwaitingClient))
        {
            return false;
        }
        let completed = env
            .plan
            .terminals()
            .iter()
            .any(|&t| st.node_state[t] == NodeState::DoneOk);
        if completed {
            finish(st, out, RunOutcome::Completed);
        } else {
            let node = stalled_culprit(env, st);
            finish(st, out, RunOutcome::Stalled { node });
        }
        return false;
    }

    // Budget pre-flight, every axis, before any dispatch (§6.7).
    let axes = [
        (BudgetAxis::Supersteps, st.spent.supersteps, Some(env.budgets.max_supersteps)),
        (BudgetAxis::WallMs, st.spent.wall_ms, env.budgets.max_wall_ms),
        (
            BudgetAxis::Tokens,
            st.spent.input_tokens + st.spent.output_tokens,
            env.budgets.max_tokens,
        ),
        (BudgetAxis::Usd, st.spent.usd_micros, env.budgets.max_usd_micros),
        (BudgetAxis::Storage, st.spent.storage_bytes, env.budgets.max_storage_bytes),
        // #295. The effects axis is derived, not counted: `journal_grains`
        // is an intent plus a result per settled effect, so halving it is
        // the count with no new state to diverge an existing checkpoint.
        (BudgetAxis::Effects, st.spent.journal_grains / 2, env.budgets.max_effects),
        (BudgetAxis::ToolCalls, st.spent.tool_calls, env.budgets.max_tool_calls),
    ];
    for (axis, spent, max) in axes {
        if let Some(max) = max {
            if spent >= max {
                finish(st, out, RunOutcome::BudgetExhausted { axis });
                return false;
            }
        }
    }

    // A host pause (#344) holds HERE — after every terminal decision above,
    // before anything is opened. The state is left exactly as the last close
    // checkpointed it, so the parked checkpoint is the uninterrupted run's.
    if hold {
        return false;
    }

    // Open the superstep.
    st.superstep += 1;
    st.spent.supersteps += 1;
    // A resume boundary: the span from the previous close to this open is
    // time the run did not exist. It accrues as `elapsed` and is NEVER
    // charged as wall — the same rule a park follows, applied to the gap a
    // crash leaves. Journaling the reading is what lets `verify` reproduce
    // it: the driver's own clock is not derivable from the journal.
    let resumed_at = st.paused_at.take().map(|paused| {
        st.elapsed_ms += st.clock_ms.saturating_sub(paused);
        st.clock_ms
    });
    st.wall_open = Some(st.clock_ms);
    st.phase = Phase::Open {
        outstanding: Default::default(),
        results: BTreeMap::new(),
        send_results: BTreeMap::new(),
        record: DecisionRecord {
            superstep: st.superstep,
            clock_open_ms: st.clock_ms,
            resumed_at,
            ..DecisionRecord::default()
        },
    };
    apply_inbox(st);
    for i in ready {
        dispatch_node(env, st, i, out);
    }
    for p in queued_tasks {
        dispatch_send_task(env, st, &p, out);
    }
    for (i, path) in needy_flows {
        let need = st.abstract_flows.get_mut(&flow_key(i, &path)).and_then(|fl| fl.need.take());
        match need {
            Some(crate::state::FlowNeed::NextTurn) => dispatch_llm_turn(env, st, i, &path, out),
            Some(crate::state::FlowNeed::Tools(calls)) => {
                for call in calls {
                    dispatch_flow_tool(env, st, i, &path, call, out);
                }
            }
            None => {}
        }
    }
    true
}

/// Dispatch one queued Send task: its own attempt counter, its own
/// task-scoped input, the target node's Host executor.
fn dispatch_send_task(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    path: &str,
    out: &mut Vec<Command>,
) {
    let Some(task) = st.send_tasks.get_mut(path) else { return };
    task.attempt += 1;
    task.state = crate::state::SendTaskState::Running;
    let (target, attempt, input) = (task.node_idx, task.attempt, task.input.clone());
    let executor = env.executors[target].clone();
    if let NodeExecutor::Abstract { .. } = &executor {
        start_flow(env, st, target, path, input, out);
        return;
    }
    let key = JournalKey {
        run_id: st.run_id.clone(),
        task_path: path.to_string(),
        node: env.plan.nodes[target].clone(),
        attempt,
        effect_seq: 0,
        kind: EffectKind::Tool,
    };
    let clock = st.clock_ms;
    let superstep = st.superstep;
    if let Phase::Open { outstanding, record, .. } = &mut st.phase {
        outstanding.insert(key.clone());
        record.task_dispatched.push((path.to_string(), attempt));
    }
    out.push(Command::WriteIntent {
        key: key.clone(),
        executor: executor.clone(),
        input: input.clone(),
        superstep,
        clock_ms: clock,
    });
    out.push(Command::Dispatch { key, executor, input });
}

/// Allocate the next attempt for a Ready node and emit its first effect —
/// a Host/Subgraph/MemoryRead dispatch, a Client ask, or an abstract node's
/// opening LLM turn.
fn dispatch_node(env: &StepEnv<'_>, st: &mut SchedulerState, i: usize, out: &mut Vec<Command>) {
    st.attempt[i] += 1;
    let executor = env.executors[i].clone();

    // Abstract nodes run as an LLM loop: fresh flow, then the first turn.
    if let NodeExecutor::Abstract { .. } = &executor {
        st.node_state[i] = NodeState::Dispatched;
        let input = st.context.clone();
        start_flow(env, st, i, "", input, out);
        return;
    }

    let key = JournalKey {
        run_id: st.run_id.clone(),
        task_path: String::new(),
        node: env.plan.nodes[i].clone(),
        attempt: st.attempt[i],
        effect_seq: 0,
        kind: EffectKind::Tool,
    };
    let input = st.context.clone();
    let clock = st.clock_ms;
    let superstep = st.superstep;

    {
        let Phase::Open { outstanding, record, .. } = &mut st.phase else {
            return;
        };
        outstanding.insert(key.clone());
        record.dispatched.push((i, st.attempt[i]));
    }

    out.push(Command::WriteIntent {
        key: key.clone(),
        executor: executor.clone(),
        input: input.clone(),
        superstep,
        clock_ms: clock,
    });
    match &executor {
        NodeExecutor::Host { .. }
        | NodeExecutor::Subgraph { .. }
        | NodeExecutor::MemoryRead { .. }
        | NodeExecutor::Decide { .. } => {
            st.node_state[i] = NodeState::Dispatched;
            out.push(Command::Dispatch { key, executor, input });
        }
        NodeExecutor::Client { tool_name, approval, .. } => {
            st.node_state[i] = NodeState::AwaitingClient;
            let ask = Ask {
                tool_call_id: key.tool_call_id(),
                node: env.plan.nodes[i].clone(),
                tool_name: tool_name.clone(),
                input,
                expires_at_sec: env
                    .ask_ttl_sec
                    .map(|ttl| (st.clock_ms / 1000) as i64 + ttl),
                // Read from the pinned executor (#294) rather than hard-coded
                // true. An approval boundary is still the default and still
                // what an unannotated Definition means; a `confirmation` is
                // an explicit, host-opted-in declaration that the run's own
                // initiator may answer this one — so a reversible write can
                // be self-confirmed in ONE run, with one journal and one
                // `verify`, instead of a prepare run plus a commit run and a
                // product-side ledger joining them.
                approval: *approval,
            };
            st.pending_asks
                .insert(ask.tool_call_id.clone(), PendingAsk { key, node_idx: i, ask });
        }
        NodeExecutor::Abstract { .. } => unreachable!("handled above"),
    }
}

/// Which transcript entries a fold may replace: everything between the node's
/// input and the last [`KEEP_TAIL`] entries. `None` when that middle is empty —
/// there is nothing a summary could stand in for.
///
/// Two rules make the result safe to send:
///
/// - **`messages[0]` is never folded.** It is the node's instruction and its
///   input state — the only entry that says what the node is for. A summary of
///   the task cannot replace the task.
/// - **The cut never separates a `tool` result from the `assistant` entry that
///   issued it.** Providers reject a tool result with no preceding tool call,
///   so the cut moves FORWARD past any leading `tool` entries, shrinking the
///   kept tail rather than splitting a round. Moving it backward would instead
///   fold the assistant entry and orphan the results.
///
/// Pure, and only ever called at a `NextTurn` boundary (`dispatch_llm_turn`
/// runs with `pending_tools` empty by construction), so a fold cannot land
/// mid-round.
fn fold_range(messages: &[Value]) -> Option<(usize, usize)> {
    let from = 1;
    let mut to = messages.len().saturating_sub(KEEP_TAIL);
    if to <= from {
        return None;
    }
    // Keep the round intact: a `tool` entry at the cut belongs to an assistant
    // entry inside the folded range, so advance until the boundary is not one.
    while to < messages.len() && messages[to].get("role").and_then(Value::as_str) == Some("tool") {
        to += 1;
    }
    if to <= from || to > messages.len() {
        return None;
    }
    Some((from, to))
}

/// Emit the summarizer turn for `from..to` of node `i`'s transcript.
///
/// An ordinary [`EffectKind::Llm`] effect in every respect that matters: it
/// consumes an `effect_seq`, spends the per-call reservation, and is journaled
/// as intent + result. That is the whole design — a fold is one more journaled
/// turn, so `verify` answers it from the journal like any other and never calls
/// the model. The `fold` key in its input is what tells the driver to offer the
/// summarizer NO tools, and what makes the decision visible in `run inspect`.
fn emit_fold_turn(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    from: usize,
    to: usize,
    out: &mut Vec<Command>,
) {
    let flow_id = flow_key(i, path);
    let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
    if flow.next_effect_seq >= env.max_effects_per_attempt {
        fail_abstract(env, st, i, path, "llm loop exceeded max_effects_per_attempt");
        return;
    }
    let effect_seq = flow.next_effect_seq;
    flow.next_effect_seq += 1;
    flow.need = None;
    let mut messages: Vec<Value> = flow.messages[from..to].to_vec();
    messages.push(serde_json::json!({ "role": "user", "content": FOLD_PROMPT }));
    flow.folding = Some(crate::state::FoldInFlight { from, to, effect_seq });
    // Consumed: the refusal has been answered with a fold. A further refusal
    // sets it again, and the effect cap is what bounds that.
    flow.fold_forced = false;
    let key = JournalKey {
        run_id: st.run_id.clone(),
        task_path: path.to_string(),
        node: env.plan.nodes[i].clone(),
        attempt: flow_attempt(st, i, path),
        effect_seq,
        kind: EffectKind::Llm,
    };
    let mut input = serde_json::json!({
        "messages": messages,
        "fold": { "from": from, "to": to, "seq": effect_seq, "prompt_v": FOLD_PROMPT_V },
    });
    if let Some(block) = request_block(env, i, true, "") {
        input["llm"] = block;
    }
    emit_effect(env, st, i, path, key, input, out);
}

/// Emit the next LLM turn of node `i`'s abstract flow, with the §6.7
/// per-dispatch token reservation: `spent + reserve` must fit BEFORE the
/// effect is emitted — the refinement pre-flight-only checking could not
/// give, now that LLM effects carry a reservable `max_tokens`.
fn dispatch_llm_turn(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    out: &mut Vec<Command>,
) {
    let flow_id = flow_key(i, path);
    let reserve = reserve_for(env, i);
    if let Some(max) = env.budgets.max_tokens {
        let spent = st.spent.input_tokens + st.spent.output_tokens;
        if spent + reserve > max {
            st.exhausted = Some(crate::error::BudgetAxis::Tokens);
            // The un-dispatched turn survives as a need, so a resume with a
            // raised ceiling picks the loop up exactly here.
            if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                flow.need = Some(crate::state::FlowNeed::NextTurn);
            }
            return;
        }
    }
    // The context ceiling, checked AFTER the budget reservation and BEFORE the
    // effect cap: a fold spends an effect like any turn, so it must not be the
    // thing that pushes the node over its count without the count having been
    // checked. Measured on the PROVIDER's reported prompt tokens for the last
    // completed turn — never on an estimate — plus the output we are about to
    // reserve, because a prompt that fits and a reply that does not is the
    // same rejection.
    {
        let flow_now = st.abstract_flows.get(&flow_id);
        let projected = flow_now
            .map(|f| f.last_prompt_tokens)
            .unwrap_or(0)
            .saturating_add(reserve);
        let idle = flow_now.is_some_and(|f| f.folding.is_none());
        // Two independent reasons to fold. The ceiling is a prediction and can
        // be unset; the provider's refusal is a fact and always wins.
        let forced = flow_now.is_some_and(|f| f.fold_forced);
        let over_ceiling = env.llm_context_tokens.is_some_and(|c| projected > c);
        if idle && (forced || over_ceiling) {
            match st.abstract_flows.get(&flow_id).and_then(|f| fold_range(&f.messages)) {
                Some((from, to)) => {
                    // C2: a calibrated decision backend gets the first look at
                    // the window — once per trigger. Anything that stops it
                    // (none pinned, uncalibrated, already tried, nothing to
                    // judge, the effect cap) is the summarizer fold,
                    // unchanged.
                    let tried = st.abstract_flows.get(&flow_id).is_some_and(|f| f.decide_fold_tried);
                    if !tried && emit_fold_decide(env, st, i, path, from, to, out) {
                        return;
                    }
                    emit_fold_turn(env, st, i, path, from, to, out);
                    return;
                }
                // Nothing foldable: the node's input plus the kept tail alone
                // exceed the ceiling. Another fold cannot help, and looping
                // would spend the effect budget summarizing summaries.
                None => {
                    let tokens = st
                        .abstract_flows
                        .get(&flow_id)
                        .map(|f| f.last_prompt_tokens)
                        .unwrap_or(0);
                    let detail = RunError::ContextExceeded {
                        node: env.plan.nodes[i].clone(),
                        tokens,
                        ceiling: env.llm_context_tokens,
                    }
                    .to_string();
                    fail_abstract_coded(env, st, i, path, &detail);
                    return;
                }
            }
        }
    }
    let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
    if flow.next_effect_seq >= env.max_effects_per_attempt {
        fail_abstract(env, st, i, path, "llm loop exceeded max_effects_per_attempt");
        return;
    }
    // A decision-guided fold was tried and the transcript has since been
    // MEASURED under its ceiling (the check above did not fire on a real
    // count): a later trigger may ask a decision again.
    if flow.decide_fold_tried && flow.last_prompt_tokens > 0 {
        flow.decide_fold_tried = false;
    }
    let effect_seq = flow.next_effect_seq;
    flow.next_effect_seq += 1;
    flow.need = None;
    let messages = flow.messages.clone();
    let offer = flow.offer.clone();
    let key = JournalKey {
        run_id: st.run_id.clone(),
        task_path: path.to_string(),
        node: env.plan.nodes[i].clone(),
        attempt: flow_attempt(st, i, path),
        effect_seq,
        kind: EffectKind::Llm,
    };
    let mut input = serde_json::json!({ "messages": messages });
    // C1: the narrowed offer rides the turn's journaled input. The driver
    // offers exactly these Definitions, keyed off the journal (as the fold's
    // no-tools rule is), so verify makes the identical choice. Absent when
    // nothing was narrowed, so such a turn is byte-identical to before.
    if let Some(offer) = offer {
        input["offer"] = offer;
    }
    // #391: the node's request fields ride the journaled input too, which is
    // how the driver builds the request from the journal alone — and how a
    // reader of the run sees the system text and response format as sent.
    let instruction = messages
        .first()
        .and_then(|m| m.pointer("/content/instruction"))
        .and_then(Value::as_str)
        .unwrap_or_default();
    if let Some(block) = request_block(env, i, false, instruction) {
        input["llm"] = block;
    }
    emit_effect(env, st, i, path, key, input, out);
}

/// Emit one model-issued tool call from node `i`'s flow.
fn dispatch_flow_tool(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    call: crate::state::PendingToolCall,
    out: &mut Vec<Command>,
) {
    let flow_id = flow_key(i, path);
    // Draining (exhausted mid-pass): preserve the call as a need instead of
    // emitting new work — state.rs pins "no new work is emitted".
    if st.exhausted.is_some() {
        if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
            match &mut flow.need {
                Some(crate::state::FlowNeed::Tools(calls)) => calls.push(call),
                other => *other = Some(crate::state::FlowNeed::Tools(vec![call])),
            }
        }
        return;
    }
    // Run-level tool-call ceiling (#295), checked per dispatch the way the
    // token reservation is. Exhaustion is a resumable BUDGET stop, never a
    // node failure: the undispatched call survives as the flow's `need`, so
    // `run fork` under a raised cap continues exactly here.
    if let Some(max) = env.budgets.max_tool_calls {
        if st.spent.tool_calls >= max {
            st.exhausted = Some(crate::error::BudgetAxis::ToolCalls);
            if let Some(flow) = st.abstract_flows.get_mut(&flow_id) {
                match &mut flow.need {
                    Some(crate::state::FlowNeed::Tools(calls)) => calls.push(call),
                    other => *other = Some(crate::state::FlowNeed::Tools(vec![call])),
                }
            }
            return;
        }
    }
    let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
    if flow.next_effect_seq >= env.max_effects_per_attempt {
        fail_abstract(env, st, i, path, "llm loop exceeded max_effects_per_attempt");
        return;
    }
    // Resolve the offered tool BEFORE booking anything: a resolution set
    // that changed across a fork must fail the node loudly, not panic —
    // and never leave a booked pending tool with no outstanding key.
    let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return };
    let Some(tool) = tools.iter().find(|t| t.tool_name == call.tool_name).cloned() else {
        fail_abstract(
            env,
            st,
            i,
            path,
            &format!("offered tool '{}' missing from the manifest's executors", call.tool_name),
        );
        return;
    };
    // The input it DISPATCHES with (#392, #393): locks merged over the
    // model's arguments, a dynamic tool's wrapped first. Validated when the
    // turn resolved; re-derived here because only the model's own arguments
    // are kept in state (the transcript must show what the model said). A
    // failure now means the offer changed under the call — a fork that
    // migrated the plan — and fails the node loudly, as the missing-tool case
    // above does.
    let arguments = match validated_dispatch_input(env, &tool, &call.arguments) {
        Ok(input) => input,
        Err(why) => {
            fail_abstract(
                env,
                st,
                i,
                path,
                &format!("tool '{}' arguments no longer validate: {why}", call.tool_name),
            );
            return;
        }
    };
    let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
    let effect_seq = flow.next_effect_seq;
    flow.next_effect_seq += 1;
    flow.pending_tools.insert(effect_seq, call.clone());
    let key = JournalKey {
        run_id: st.run_id.clone(),
        task_path: path.to_string(),
        node: env.plan.nodes[i].clone(),
        attempt: flow_attempt(st, i, path),
        effect_seq,
        kind: EffectKind::Tool,
    };
    let host = NodeExecutor::Host {
        tool_hash: tool.tool_hash.clone(),
        tool_name: tool.tool_name.clone(),
    };
    // Counted only when a cap is set (#295), so a run with no cap keeps a
    // byte-identical `Spent` and `verify` of an existing run is untouched.
    if env.budgets.max_tool_calls.is_some() {
        st.spent.tool_calls += 1;
    }
    let clock = st.clock_ms;
    let superstep = st.superstep;
    if let Phase::Open { outstanding, .. } = &mut st.phase {
        outstanding.insert(key.clone());
    }
    out.push(Command::WriteIntent {
        key: key.clone(),
        executor: host.clone(),
        input: arguments.clone(),
        superstep,
        clock_ms: clock,
    });
    out.push(Command::Dispatch { key, executor: host, input: arguments });
}

/// Shared effect emission for abstract turns.
fn emit_effect(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    key: JournalKey,
    input: Value,
    out: &mut Vec<Command>,
) {
    let executor = env.executors[i].clone();
    let clock = st.clock_ms;
    let superstep = st.superstep;
    if let Phase::Open { outstanding, record, .. } = &mut st.phase {
        outstanding.insert(key.clone());
        if key.effect_seq == 0 {
            if path.is_empty() {
                record.dispatched.push((i, key.attempt));
            } else {
                record.task_dispatched.push((path.to_string(), key.attempt));
            }
        }
    }
    out.push(Command::WriteIntent {
        key: key.clone(),
        executor: executor.clone(),
        input: input.clone(),
        superstep,
        clock_ms: clock,
    });
    out.push(Command::Dispatch { key, executor, input });
}

// ---- decisions inside an abstract flow (C1, C2) ------------------------------
//
// Both are OPTIONAL looks a decision backend gets before the deterministic
// rule runs. The scheduler never calls a backend: it emits a decision EFFECT
// (`NodeExecutor::Decide`, `EffectKind::Tool`, the flow's next `effect_seq`)
// whose input carries the request, the driver answers it through the host's
// backend and journals it like any other effect, and the answer comes back as
// an ordinary `EffectResolved`. So `verify` answers a decision from the
// journal and never re-asks, a crash re-delivers it under the same key, and
// the only new thing replay has to reproduce is what the scheduler did with
// the answer — a pure function of it.
//
// The fail-open rule (proposal §2, rules 2 and 3) is structural here: every
// path that does not produce a CALIBRATED, well-formed answer resolves to
// exactly the deterministic behaviour the flow would have had with no backend
// at all — the full offer, the summarizer fold.

/// The offer the model sees: every pinned tool, or — once C1 narrowed it —
/// only the names the flow's journaled `offer` lists. Manifest order either
/// way; narrowing only ever removes.
fn narrowed_offer(flow: Option<&crate::state::AbstractFlow>, tools: &[OfferedTool]) -> Vec<OfferedTool> {
    let names: Option<Vec<&str>> = flow
        .and_then(|f| f.offer.as_ref())
        .and_then(|o| o.get("tools"))
        .and_then(Value::as_array)
        .map(|a| a.iter().filter_map(Value::as_str).collect());
    match names {
        Some(names) => tools.iter().filter(|t| names.contains(&t.tool_name.as_str())).cloned().collect(),
        None => tools.to_vec(),
    }
}

/// `{provider, model, calibrated, latency_ms}` off a decision effect's result
/// (`Decision::to_json()`, as the driver journals it) — proposal rule 4.
fn decide_provenance(result: &Value) -> Value {
    serde_json::json!({
        "provider": result.get("provider").cloned().unwrap_or(Value::Null),
        "model": result.get("model").cloned().unwrap_or(Value::Null),
        "calibrated": result.get("calibrated").cloned().unwrap_or(Value::Null),
        "latency_ms": result.get("latency_ms").cloned().unwrap_or(Value::Null),
    })
}

/// Emit one decision effect for node `i`'s flow and mark it in flight.
///
/// Returns `false`, emitting nothing, when the flow is gone, the run is
/// draining, or the next `effect_seq` would pass the effect cap. A decision is
/// optional: running out of effects for one must not fail the node — the
/// deterministic path that runs instead meets the cap on its own terms.
fn emit_decide_effect(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    request: Value,
    in_flight: impl FnOnce(u32) -> DecideInFlight,
    out: &mut Vec<Command>,
) -> bool {
    if st.exhausted.is_some() || st.cancel.is_some() {
        return false;
    }
    let attempt = flow_attempt(st, i, path);
    let Some(flow) = st.abstract_flows.get_mut(&flow_key(i, path)) else { return false };
    if flow.next_effect_seq >= env.max_effects_per_attempt {
        return false;
    }
    let effect_seq = flow.next_effect_seq;
    flow.next_effect_seq += 1;
    flow.need = None;
    flow.deciding = Some(in_flight(effect_seq));
    let key = JournalKey {
        run_id: st.run_id.clone(),
        task_path: path.to_string(),
        node: env.plan.nodes[i].clone(),
        attempt,
        effect_seq,
        kind: EffectKind::Tool,
    };
    let executor = NodeExecutor::Decide { tool_hash: String::new(), tool_name: DECIDE_TOOL.into() };
    let input = serde_json::json!({ "decide": request });
    let (clock, superstep) = (st.clock_ms, st.superstep);
    if let Phase::Open { outstanding, record, .. } = &mut st.phase {
        outstanding.insert(key.clone());
        if effect_seq == 0 {
            if path.is_empty() {
                record.dispatched.push((i, attempt));
            } else {
                record.task_dispatched.push((path.to_string(), attempt));
            }
        }
    }
    out.push(Command::WriteIntent {
        key: key.clone(),
        executor: executor.clone(),
        input: input.clone(),
        superstep,
        clock_ms: clock,
    });
    out.push(Command::Dispatch { key, executor, input });
    true
}

/// C1: before an abstract flow's first turn, ask a calibrated backend which
/// of the pinned tools this step needs — only when more than
/// [`OFFER_TOP_K`] are pinned. `false` = not asked; the full offer stands.
fn emit_offer_decide(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    out: &mut Vec<Command>,
) -> bool {
    let Some(d) = env.decide.as_ref().filter(|d| d.calibrated) else { return false };
    let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return false };
    // The node's own rule (#396): `narrow: false` keeps its whole offer,
    // `narrow: {top_k}` moves the K.
    let Some(top_k) = narrow_top_k(env, i) else { return false };
    // A choice carries 2..=255 options; past that the question cannot be
    // asked, and the full offer is the honest answer.
    if tools.len() <= top_k || tools.len() > 255 {
        return false;
    }
    let Some(flow) = st.abstract_flows.get(&flow_key(i, path)) else { return false };
    let first = flow.messages.first().and_then(|m| m.get("content")).cloned().unwrap_or(Value::Null);
    // A dynamic tool (#392) is described by its OWN pinned description, its
    // first line — exactly what a Definition's `tool_description` gives
    // (#396). Read off the offer rather than the run-wide table, because two
    // nodes may each declare a dynamic tool of one name.
    let describe = |t: &OfferedTool| {
        t.dynamic
            .as_ref()
            .map(|dt| dt.description.lines().next().unwrap_or_default().trim().to_string())
            .filter(|s| !s.is_empty())
            .or_else(|| d.tool_descriptions.get(&t.tool_name).filter(|s| !s.trim().is_empty()).cloned())
            .unwrap_or_else(|| t.tool_name.clone())
    };
    let criteria: serde_json::Map<String, Value> =
        tools.iter().map(|t| (t.tool_name.clone(), Value::String(describe(t)))).collect();
    let listed: Vec<Value> = tools
        .iter()
        .map(|t| serde_json::json!({ "name": t.tool_name, "description": describe(t) }))
        .collect();
    let request = serde_json::json!({
        "purpose": "offer",
        "v": FOLD_DECIDE_V,
        "state": {
            "plan": d.plan_label,
            "step": first.get("instruction").cloned().unwrap_or(Value::Null),
            "input": first.get("state").cloned().unwrap_or(Value::Null),
            "tools": listed,
        },
        "questions": {
            "tool": { "type": "choice", "instructions": OFFER_QUESTION, "criteria": criteria },
        },
    });
    emit_decide_effect(env, st, i, path, request, |effect_seq| DecideInFlight::Offer { effect_seq }, out)
}

/// The tool a `tool` transcript entry answers, by the model's call id, read
/// off the nearest preceding assistant entry.
fn tool_entry_name(messages: &[Value], n: usize) -> Value {
    let id = messages[n].get("tool_call_id").and_then(Value::as_str).unwrap_or_default();
    messages[..n]
        .iter()
        .rev()
        .find(|m| m.get("role").and_then(Value::as_str) == Some("assistant"))
        .and_then(|a| a.get("tool_calls").and_then(Value::as_array))
        .and_then(|calls| {
            calls.iter().find(|c| c.get("id").and_then(Value::as_str) == Some(id))
        })
        .and_then(|c| c.get("name").cloned())
        .unwrap_or(Value::Null)
}

/// A transcript entry's content as the text a model would have been shown.
fn content_text(m: &Value) -> String {
    match m.get("content") {
        Some(Value::String(s)) => s.clone(),
        Some(other) => other.to_string(),
        None => String::new(),
    }
}

/// C2: before the summarizer fold, show a calibrated backend the foldable
/// window — every tool result replaced by a short note — and ask two `noul`
/// questions per result. `false` = not asked; the summarizer runs.
fn emit_fold_decide(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    from: usize,
    to: usize,
    out: &mut Vec<Command>,
) -> bool {
    if !env.decide.as_ref().is_some_and(|d| d.calibrated) {
        return false;
    }
    let Some(flow) = st.abstract_flows.get(&flow_key(i, path)) else { return false };
    let messages = &flow.messages;
    let mut transcript = Vec::with_capacity(to.saturating_sub(from));
    let mut questions = serde_json::Map::new();
    for (n, m) in messages.iter().enumerate().take(to).skip(from) {
        match m.get("role").and_then(Value::as_str).unwrap_or_default() {
            "tool" => {
                let failed = m.get("is_error").and_then(Value::as_bool).unwrap_or(false);
                transcript.push(serde_json::json!({
                    "n": n,
                    "role": "tool",
                    "tool": tool_entry_name(messages, n),
                    "result": format!(
                        "{}, {} chars (omitted)",
                        if failed { "error" } else { "ok" },
                        content_text(m).chars().count()
                    ),
                }));
                questions.insert(
                    format!("keep_call_{n}"),
                    serde_json::json!({ "type": "noul", "instructions": KEEP_CALL_QUESTION }),
                );
                questions.insert(
                    format!("keep_result_{n}"),
                    serde_json::json!({ "type": "noul", "instructions": KEEP_RESULT_QUESTION }),
                );
            }
            "assistant" => transcript.push(serde_json::json!({
                "n": n,
                "role": "assistant",
                "text": m.get("text").cloned().unwrap_or(Value::Null),
                "tool_calls": m.get("tool_calls").cloned().unwrap_or(Value::Null),
            })),
            role => transcript.push(serde_json::json!({
                "n": n,
                "role": role,
                "content": m.get("content").cloned().unwrap_or(Value::Null),
            })),
        }
    }
    // Nothing a decision could keep or drop: the summarizer's job.
    if questions.is_empty() {
        return false;
    }
    let task = messages
        .first()
        .and_then(|m| m.get("content"))
        .map(|c| c.get("instruction").cloned().unwrap_or_else(|| c.clone()))
        .unwrap_or(Value::Null);
    let request = serde_json::json!({
        "purpose": "fold",
        "v": FOLD_DECIDE_V,
        "from": from,
        "to": to,
        "state": { "task": task, "transcript": transcript },
        "questions": questions,
    });
    emit_decide_effect(
        env,
        st,
        i,
        path,
        request,
        |effect_seq| DecideInFlight::Fold { from, to, effect_seq },
        out,
    )
}

/// A decision effect inside a flow resolved: route it to what it was asked
/// for. Either way the loop continues with an ordinary next turn.
fn handle_flow_decide(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    outcome: &EffectOutcome,
) {
    let Some(asked) = st.abstract_flows.get_mut(&flow_key(i, path)).and_then(|f| f.deciding.take())
    else {
        return;
    };
    match asked {
        DecideInFlight::Offer { effect_seq } => {
            resolve_offer_decide(env, st, i, path, effect_seq, outcome)
        }
        DecideInFlight::Fold { from, to, effect_seq } => {
            resolve_fold_decide(env, st, i, path, from, to, effect_seq, outcome)
        }
    }
}

/// C1's keep rule over a validated probability table: the top
/// [`OFFER_TOP_K`] by probability (ties by manifest order), plus every tool at
/// or above [`OFFER_KEEP_P`], returned in MANIFEST order. `None` when the
/// table does not cover every pinned tool with a finite probability — a
/// malformed answer narrows nothing.
/// The K node `i` narrows to, or `None` when its block says `narrow: false`.
fn narrow_top_k(env: &StepEnv<'_>, i: usize) -> Option<usize> {
    match llm_cfg(env, i) {
        Some(c) => c.narrow_top_k(OFFER_TOP_K),
        None => Some(OFFER_TOP_K),
    }
}

fn narrow_offer(
    tools: &[OfferedTool],
    probs: &serde_json::Map<String, Value>,
    top_k: usize,
) -> Option<Vec<String>> {
    let mut ranked: Vec<(usize, f64)> = Vec::with_capacity(tools.len());
    for (idx, t) in tools.iter().enumerate() {
        let p = probs.get(&t.tool_name).and_then(Value::as_f64).filter(|p| p.is_finite())?;
        ranked.push((idx, p));
    }
    ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal).then(a.0.cmp(&b.0)));
    let keep: std::collections::BTreeSet<usize> = ranked
        .iter()
        .enumerate()
        .filter(|(rank, (_, p))| *rank < top_k || *p >= OFFER_KEEP_P)
        .map(|(_, (idx, _))| *idx)
        .collect();
    Some(keep.into_iter().map(|idx| tools[idx].tool_name.clone()).collect())
}

fn resolve_offer_decide(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    effect_seq: u32,
    outcome: &EffectOutcome,
) {
    let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return };
    // Rule 2: only a CALIBRATED answer may omit a tool. Anything else — an
    // uncalibrated answer, a failure, a malformed table — keeps the full
    // offer, which is what the flow would have had with no backend.
    let narrowed = match outcome {
        EffectOutcome::Completed { result, .. }
            if result.get("calibrated") == Some(&Value::Bool(true)) =>
        {
            result
                .pointer("/answers/tool/probabilities")
                .and_then(Value::as_object)
                .and_then(|probs| narrow_offer(tools, probs, narrow_top_k(env, i).unwrap_or(OFFER_TOP_K)))
                .filter(|keep| !keep.is_empty() && keep.len() < tools.len())
                .map(|keep| {
                    let mut record = decide_provenance(result);
                    record["tools"] = serde_json::json!(keep);
                    record["seq"] = serde_json::json!(effect_seq);
                    record
                })
        }
        _ => None,
    };
    let Some(flow) = st.abstract_flows.get_mut(&flow_key(i, path)) else { return };
    flow.offer = narrowed;
    flow.need = Some(crate::state::FlowNeed::NextTurn);
}

/// What a decision-guided fold will do to the window, per entry.
struct Prune {
    kept: Vec<usize>,
    truncated: Vec<usize>,
    dropped: Vec<usize>,
}

/// The decision → edit plan (C2). Pure; `Err` names why nothing is applied.
///
/// Per tool-result entry `n`: `keep_result ≥ 0.5` stays verbatim;
/// `keep_call ≥ 0.5 > keep_result` keeps the call and truncates the result;
/// both below drops it. Then the never-separate rule, at ROUND granularity: a
/// round (an assistant entry and the results that answer it) leaves the prompt
/// only WHOLE — every result dropped, and the assistant entry with them. A
/// round with anything kept keeps every call, and its would-be-dropped
/// results are truncated instead. Editing an assistant entry's `tool_calls`
/// would desynchronize it from its `provider_content` (#284), which the
/// adapter replays verbatim — and a call with no result, or a result with no
/// call, is a request every provider refuses.
fn plan_prune(messages: &[Value], from: usize, to: usize, result: &Value) -> Result<Prune, String> {
    if result.get("calibrated") != Some(&Value::Bool(true)) {
        return Err("the backend answered uncalibrated — an uncalibrated backend may reorder, \
                    never omit"
            .into());
    }
    let answers = result.get("answers").ok_or("the decision carries no answers")?;
    let p = |id: &str| -> Result<f64, String> {
        answers
            .get(id)
            .and_then(|a| a.get("noul"))
            .and_then(Value::as_f64)
            .filter(|x| x.is_finite() && (0.0..=1.0).contains(x))
            .ok_or_else(|| format!("no valid answer for {id}"))
    };
    #[derive(Clone, Copy, PartialEq)]
    enum Verdict {
        Keep,
        Truncate,
        Drop,
    }
    // (owning assistant index, [(tool index, verdict)]) in transcript order.
    type Round = (Option<usize>, Vec<(usize, Verdict)>);
    let mut rounds: Vec<Round> = Vec::new();
    let mut owner: Option<usize> = None;
    for (n, m) in messages.iter().enumerate().take(to).skip(from) {
        match m.get("role").and_then(Value::as_str).unwrap_or_default() {
            "assistant" => {
                owner = Some(n);
                rounds.push((owner, Vec::new()));
            }
            "tool" => {
                let (keep_call, keep_result) =
                    (p(&format!("keep_call_{n}"))?, p(&format!("keep_result_{n}"))?);
                let v = if keep_result >= 0.5 {
                    Verdict::Keep
                } else if keep_call >= 0.5 {
                    Verdict::Truncate
                } else {
                    Verdict::Drop
                };
                match rounds.last_mut() {
                    Some((o, members)) if *o == owner && owner.is_some() => members.push((n, v)),
                    // A result with no assistant entry in the window: never
                    // dropped, since its call could not go with it.
                    _ => rounds.push((None, vec![(n, v)])),
                }
            }
            // Anything else (a user note, a re-prompt) closes the round.
            _ => owner = None,
        }
    }
    let (mut kept, mut truncated, mut dropped) = (Vec::new(), Vec::new(), Vec::new());
    for (own, members) in rounds {
        if members.is_empty() {
            continue;
        }
        let whole = own.is_some() && members.iter().all(|(_, v)| *v == Verdict::Drop);
        if whole {
            dropped.extend(own);
            dropped.extend(members.iter().map(|(n, _)| *n));
            continue;
        }
        for (n, v) in members {
            // Truncating what is already short changes nothing but the note;
            // it is reported as kept, because it was.
            let short = content_text(&messages[n]).chars().count() <= FOLD_TRUNCATE_CHARS;
            if v == Verdict::Keep || short {
                kept.push(n);
            } else {
                truncated.push(n);
            }
        }
    }
    if truncated.is_empty() && dropped.is_empty() {
        return Err("the decision kept every entry verbatim".into());
    }
    kept.sort_unstable();
    truncated.sort_unstable();
    dropped.sort_unstable();
    Ok(Prune { kept, truncated, dropped })
}

#[allow(clippy::too_many_arguments)]
fn resolve_fold_decide(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    from: usize,
    to: usize,
    effect_seq: u32,
    outcome: &EffectOutcome,
) {
    let flow_id = flow_key(i, path);
    let attempt = flow_attempt(st, i, path);
    let Some(flow) = st.abstract_flows.get(&flow_id) else { return };
    let (plan, provenance) = match outcome {
        EffectOutcome::Completed { result, .. } => {
            let plan = if from < to && to <= flow.messages.len() {
                plan_prune(&flow.messages, from, to, result)
            } else {
                Err("the window no longer fits the transcript".into())
            };
            (plan, Some(decide_provenance(result)))
        }
        EffectOutcome::Failed { cause, detail, .. } => {
            (Err(format!("the decision failed ({cause:?}: {detail})")), None)
        }
    };
    let mut record = FoldRecord {
        kind: "decide".into(),
        v: FOLD_DECIDE_V,
        node: env.plan.nodes[i].clone(),
        task_path: path.to_string(),
        attempt,
        seq: effect_seq,
        from,
        to,
        provenance: provenance.clone(),
        ..FoldRecord::default()
    };
    let Some(flow) = st.abstract_flows.get_mut(&flow_id) else { return };
    // Either way this trigger has had its decision: the next one goes to the
    // summarizer until the transcript is measured under the ceiling again.
    flow.decide_fold_tried = true;
    flow.need = Some(crate::state::FlowNeed::NextTurn);
    match plan {
        Err(reason) => {
            // Fail open: nothing is edited, and the NextTurn above re-enters
            // `dispatch_llm_turn`, whose trigger still holds — the summarizer
            // fold runs exactly as it would have with no backend.
            record.reason = Some(reason);
        }
        Ok(prune) => {
            let n = flow.folds + 1;
            let who = provenance
                .as_ref()
                .map(|p| {
                    format!(
                        "{}/{}",
                        p.get("provider").and_then(Value::as_str).unwrap_or("?"),
                        p.get("model").and_then(Value::as_str).unwrap_or("?")
                    )
                })
                .unwrap_or_default();
            let note = serde_json::json!({
                "role": "user",
                "content": format!(
                    "[Areev fold {n} (decision): of transcript entries {from}..{to} of attempt \
                     {attempt}, entries {:?} were dropped and {:?} truncated to \
                     {FOLD_TRUNCATE_CHARS} characters by a decision model ({who}), journaled at \
                     effect_seq {effect_seq}. The full record — every turn and every tool result \
                     — is in this run's journal.]",
                    prune.dropped, prune.truncated,
                ),
            });
            let mut next = Vec::with_capacity(flow.messages.len() + 1);
            for (idx, m) in flow.messages.iter().enumerate() {
                if idx == from {
                    next.push(note.clone());
                }
                if prune.dropped.contains(&idx) {
                    continue;
                }
                if prune.truncated.contains(&idx) {
                    let text = content_text(m);
                    let total = text.chars().count();
                    let head: String = text.chars().take(FOLD_TRUNCATE_CHARS).collect();
                    let mut cut = m.clone();
                    cut["content"] = Value::String(format!(
                        "{head}… [truncated to {FOLD_TRUNCATE_CHARS} of {total} characters by a \
                         decision model; the full result is in this run's journal]"
                    ));
                    next.push(cut);
                } else {
                    next.push(m.clone());
                }
            }
            flow.messages = next;
            flow.folds = n;
            // The pruned transcript is unmeasured: zero means "ask the
            // provider", and a refusal that provoked this has been answered.
            flow.last_prompt_tokens = 0;
            flow.fold_forced = false;
            st.folds = st.folds.saturating_add(1);
            record.applied = true;
            record.kept = prune.kept;
            record.truncated = prune.truncated;
            record.dropped = prune.dropped;
        }
    }
    if let Phase::Open { record: rec, .. } = &mut st.phase {
        rec.folds.push(record);
    }
}

/// Fail an abstract node (loop bound, terminal model failure): fail-fast
/// semantics, the same path a Host node's retry exhaustion takes. The detail is
/// classified `ExecutorError:` — the loop's own failures are the executor's as
/// far as a reader of the outcome is concerned.
fn fail_abstract(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    detail: &str,
) {
    fail_abstract_coded(env, st, i, path, &format!("ExecutorError: {detail}"));
}

/// Fail an abstract node with a detail that already leads with its own
/// `RUN-Ennn`. Separate from [`fail_abstract`] because the workspace rule is
/// that the code is the LEADING token of what a user reads, and an
/// `ExecutorError:` prefix in front of one buries it — besides being wrong:
/// a transcript the scheduler refused to send is not an executor's failure.
fn fail_abstract_coded(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    i: usize,
    path: &str,
    detail: &str,
) {
    st.abstract_flows.remove(&flow_key(i, path));
    if let Some(task) = st.send_tasks.get_mut(path) {
        task.state = crate::state::SendTaskState::Done;
    }
    st.node_state[i] = NodeState::DoneFailed;
    let named = st.failed.as_ref().map(|(n, _)| *n).unwrap_or(usize::MAX);
    if i < named {
        let at = if path.is_empty() { String::new() } else { format!(" (task {path})") };
        st.failed = Some((i, format!("{detail}{at}")));
    }
    let _ = env;
}

/// Park: checkpoint + envelope, once per ask (a parked run polled again
/// must not re-announce).
fn park(st: &mut SchedulerState, out: &mut Vec<Command>) {
    let fresh: Vec<Ask> = st
        .pending_asks
        .values()
        .filter(|p| !st.announced_asks.contains(&p.ask.tool_call_id))
        .map(|p| p.ask.clone())
        .collect();
    if fresh.is_empty() {
        return;
    }
    for a in &fresh {
        st.announced_asks.insert(a.tool_call_id.clone());
    }
    if let Phase::Open { record, .. } = &st.phase {
        let mut record = record.clone();
        // The park reading MUST be journaled: the wall segment accrued at
        // park uses st.clock_ms, and verify can only reproduce it from the
        // checkpoint's own record (feeding an unjournaled reading would
        // false-diverge any superstep that resolved an effect before
        // parking).
        record.clock_close_ms = st.clock_ms;
        let state_json = serde_json::to_value(&*st).unwrap_or(Value::Null);
        out.push(Command::WriteCheckpoint {
            superstep: st.superstep,
            decision_record: record,
            state_json,
        });
    }
    out.push(Command::EmitEnvelope { asks: fresh });
}

/// Close the open superstep: extract Send spawns, reduce in canonical
/// order, evaluate edges in canonical order (unless draining), checkpoint.
fn close_superstep(env: &StepEnv<'_>, st: &mut SchedulerState, out: &mut Vec<Command>) {
    let Phase::Open { mut results, mut send_results, mut record, .. } =
        std::mem::replace(&mut st.phase, Phase::Idle)
    else {
        return;
    };

    // Send extraction (§5.1): a result's reserved `$send` key is a spawn
    // decision, made HERE at close in canonical order — static spawners by
    // node index, then task spawners by path — journaled in the decision
    // record, and stripped before reduction. A malformed decision is
    // fail-fast (the spawner is named; no tasks spawn; the run drains).
    if st.failed.is_none() && st.cancel.is_none() {
        let mut decisions: Vec<(String, usize, Vec<Value>)> = Vec::new();
        for (i, r) in results.iter_mut() {
            if let Some(list) = take_send(r) {
                decisions.push((String::new(), *i, list));
            }
        }
        let task_nodes: BTreeMap<String, usize> = st
            .send_tasks
            .iter()
            .map(|(p, t)| (p.clone(), t.node_idx))
            .collect();
        for (path, r) in send_results.iter_mut() {
            if let Some(list) = take_send(r) {
                decisions.push((path.clone(), task_nodes[path], list));
            }
        }
        apply_spawns(env, st, &mut record, decisions);
    }

    // Reducers, canonical order (§5.2: state-merge order is canonical —
    // node index, never completion order; Send results after the static
    // set, in spawn order — task paths zero-pad their ordinals so
    // lexicographic IS numeric). An object result merges per top-level
    // key; anything else merges under the node's own id.
    let send_merges: Vec<(usize, Value)> = send_results
        .iter()
        .map(|(p, v)| (st.send_tasks[p].node_idx, v.clone()))
        .collect();
    for (node_idx, result) in results
        .iter()
        .map(|(i, v)| (*i, v.clone()))
        .chain(send_merges)
    {
        match result {
            Value::Object(map) => {
                for (k, v) in map {
                    let prev = st.context.get(&k).cloned();
                    let merged = (env.reduce)(&k, prev.as_ref(), &v);
                    st.context[k.as_str()] = merged;
                }
            }
            Value::Null => {}
            other => {
                let key = env.plan.nodes[node_idx].clone();
                let prev = st.context.get(&key).cloned();
                let merged = (env.reduce)(&key, prev.as_ref(), &other);
                st.context[key.as_str()] = merged;
            }
        }
    }

    // Edge evaluation — skipped when the run is dying (fail-fast /
    // cancel). An EXHAUSTED run still evaluates: it is designed to resume
    // under raised budgets, so its last superstep's firings must land
    // (dispatch stays drained either way — readiness without dispatch).
    if st.failed.is_none() && st.cancel.is_none() {
        for i in results.keys().copied().collect::<Vec<_>>() {
            for ei in env.plan.out_edges[i].clone() {
                let edge = &env.plan.edges[ei];
                let outcome = if edge.max_cycles.is_some_and(|m| st.edge_fired[ei] >= m) {
                    EdgeOutcome::CycleExhausted
                } else {
                    // Unconditional edges always fire; the evaluator is only
                    // consulted for edges that carry a condition.
                    let fires = (edge.cond.is_none() && edge.cond_raw.is_none())
                        || (env.eval_cond)(edge, &st.context);
                    if fires { EdgeOutcome::Fired } else { EdgeOutcome::CondFalse }
                };
                if outcome == EdgeOutcome::Fired {
                    st.edge_fired[ei] += 1;
                }
                let gen = deliver(env, st, ei, outcome == EdgeOutcome::Fired);
                record.edges.push((ei, gen, outcome));
            }
        }
        propagate_death(env, st, &mut record);
    }

    record.clock_close_ms = st.clock_ms;
    // Close the active wall segment (open→park segments were already
    // charged at park time). A superstep that parked charges NOTHING after
    // the park; the park→close span accrues as `elapsed` instead — both
    // endpoints are journaled readings, so replay reproduces both figures
    // byte-exactly (§6.7: a three-day approval pause is reported, never
    // billed — and never a replay divergence either).
    if let Some(open) = st.wall_open.take() {
        st.spent.wall_ms += st.clock_ms.saturating_sub(open);
    }
    if let Some(paused) = st.paused_at.take() {
        st.elapsed_ms += st.clock_ms.saturating_sub(paused);
    }
    let state_json = serde_json::to_value(&*st).unwrap_or(Value::Null);
    out.push(Command::WriteCheckpoint {
        superstep: st.superstep,
        decision_record: record,
        state_json,
    });
}

/// Pull a result's `$send` list out (stripping the key so reducers never
/// see it). Returns None when the result carries no spawn decision.
fn take_send(result: &mut Value) -> Option<Vec<Value>> {
    let obj = result.as_object_mut()?;
    let taken = obj.remove("$send")?;
    match taken {
        Value::Array(list) => Some(list),
        other => Some(vec![other]), // validated (and refused) in apply_spawns
    }
}

/// Per-decision spawn cap (also the per-parent lifetime cap via the
/// 4-digit ordinal space). Far above any sane fan-out; a backstop, not a
/// tuning knob.
const MAX_SPAWNS: usize = 4096;

/// Validate and apply the superstep's spawn decisions: fresh task paths
/// (`parent/NNNN`, monotonic per parent), targets pinned Host-executed,
/// the target node preempted into `Dispatched`. Violations are fail-fast
/// naming the spawner.
fn apply_spawns(
    env: &StepEnv<'_>,
    st: &mut SchedulerState,
    record: &mut DecisionRecord,
    decisions: Vec<(String, usize, Vec<Value>)>,
) {
    let fail = |st: &mut SchedulerState, spawner: usize, why: String| {
        st.node_state[spawner] = NodeState::DoneFailed;
        let named = st.failed.as_ref().map(|(i, _)| *i).unwrap_or(usize::MAX);
        if spawner < named {
            st.failed = Some((spawner, format!("ExecutorError: {why}")));
        }
    };
    // Validate everything before spawning anything: a half-applied spawn
    // decision would leave tasks the journaled record never mentions.
    // (parent path, spawner idx, [(target idx, task input)]).
    type ValidatedSpawn = (String, usize, Vec<(usize, Value)>);
    let mut validated: Vec<ValidatedSpawn> = Vec::new();
    for (parent, spawner, list) in decisions {
        if list.len() > MAX_SPAWNS {
            fail(st, spawner, format!("$send spawns {} tasks (cap {MAX_SPAWNS})", list.len()));
            return;
        }
        let mut entries = Vec::with_capacity(list.len());
        for e in &list {
            let Some(target_name) = e.get("node").and_then(|n| n.as_str()) else {
                fail(
                    st,
                    spawner,
                    "$send entries must be {\"node\": \"<name>\", \"input\": …}".into(),
                );
                return;
            };
            let Some(target) = env.plan.nodes.iter().position(|n| n == target_name) else {
                fail(st, spawner, format!("$send targets unknown node '{target_name}'"));
                return;
            };
            // A declared memory read is a legal target: a fan-out of as-of
            // reads (one per policy in an aggregate) resolves each task's
            // pointers against that task's own input.
            if !matches!(
                env.executors[target],
                NodeExecutor::Host { .. }
                    | NodeExecutor::Abstract { .. }
                    | NodeExecutor::MemoryRead { .. }
                    | NodeExecutor::Decide { .. }
            ) {
                fail(
                    st,
                    spawner,
                    format!(
                        "$send target '{target_name}' is not a Host tool, abstract node, memory \
                         read or decision node"
                    ),
                );
                return;
            }
            if target == spawner {
                fail(st, spawner, format!("$send target '{target_name}' is the spawner itself"));
                return;
            }
            let input = e.get("input").cloned().unwrap_or_else(|| serde_json::json!({}));
            entries.push((target, input));
        }
        validated.push((parent, spawner, entries));
    }
    for (parent, spawner, entries) in validated {
        for (target, input) in entries {
            let counter = st.spawn_counter.entry(parent.clone()).or_insert(0);
            let ordinal = *counter;
            if ordinal > 9999 {
                // The 4-digit pad is what makes lexicographic order spawn
                // order; past it the merge-order invariant breaks — fail
                // the spawner loudly instead.
                fail(st, spawner, format!(
                    "spawn ordinal space exhausted under '{parent}' (9999 max per parent)"
                ));
                return;
            }
            *counter += 1;
            let path = format!("{parent}/{ordinal:04}");
            st.send_tasks.insert(
                path.clone(),
                crate::state::SendTask {
                    node_idx: target,
                    input,
                    attempt: 0,
                    state: crate::state::SendTaskState::Queued,
                },
            );
            st.node_state[target] = NodeState::Dispatched;
            record.spawns.push((path, target));
        }
    }
}

/// Deliver an edge resolution to its destination, with re-entry. Returns
/// the destination generation the delivery landed in.
fn deliver(env: &StepEnv<'_>, st: &mut SchedulerState, ei: usize, fired: bool) -> u32 {
    let dst = env.plan.edges[ei].dst;
    let resolved = matches!(
        st.node_state[dst],
        NodeState::DoneOk | NodeState::DoneFailed | NodeState::Dead
    );
    if resolved {
        if !fired {
            return st.node_gen[dst]; // a dead edge never re-enters
        }
        st.node_gen[dst] += 1;
        st.node_state[dst] = NodeState::Waiting;
        // Attempt stays MONOTONIC (journal keys must be generation-unique);
        // the new generation's retry budget measures from here.
        st.attempt_base[dst] = st.attempt[dst];
        st.in_res[dst].clear();
        for &other in &env.plan.in_edges[dst] {
            if other == ei {
                continue;
            }
            let src = env.plan.edges[other].src;
            let src_unresolved = matches!(
                st.node_state[src],
                NodeState::Waiting
                    | NodeState::Ready
                    | NodeState::Dispatched
                    | NodeState::AwaitingClient
            );
            if !src_unresolved {
                st.in_res[dst].insert(other, EdgeRes::Dead);
            }
        }
        st.in_res[dst].insert(ei, EdgeRes::Fired);
    } else {
        st.in_res[dst]
            .insert(ei, if fired { EdgeRes::Fired } else { EdgeRes::Dead });
    }
    refresh_readiness(env, st, dst);
    st.node_gen[dst]
}

/// Waiting → Ready when no in-edge is Pending and ≥1 Fired; → Dead when all
/// resolved and none Fired (§6.2: edges must RESOLVE, not fire).
///
/// A node's FIRST generation (`node_gen[i] == 0`) only gates on in-edges
/// that could possibly have resolved by then: a back-edge that closes a
/// cycle through this very node (`env.plan.cycle_edge`) cannot fire before
/// the node's own first run, so requiring it here would deadlock any cycle
/// whose re-entry point isn't the plan's entry node (RUN issue #33) — every
/// OTHER in-edge into a reachable non-root node is guaranteed non-back (the
/// DFS tree edge that first discovered it, at minimum), so this never
/// starves a node of every gating edge. The entry node sidesteps this rule
/// entirely via an unconditional bootstrap (`apply_event`'s `Start`
/// handler), since it may have zero in-edges at all. Once a node is
/// re-entered (generation ≥ 1), `deliver`'s re-entry branch already
/// resolves every other in-edge from concrete current node state, so the
/// full AND-join applies again from generation 1 onward.
fn refresh_readiness(env: &StepEnv<'_>, st: &mut SchedulerState, i: usize) {
    if st.node_state[i] != NodeState::Waiting {
        return;
    }
    let ins = &env.plan.in_edges[i];
    let gen0 = st.node_gen[i] == 0;
    let all_gated_resolved = ins
        .iter()
        .all(|&e| (gen0 && env.plan.cycle_edge[e]) || st.in_res[i].contains_key(&e));
    if !all_gated_resolved {
        return;
    }
    let any_fired = ins.iter().any(|e| st.in_res[i].get(e) == Some(&EdgeRes::Fired));
    st.node_state[i] = if any_fired { NodeState::Ready } else { NodeState::Dead };
}

/// Dead-path propagation to fixpoint, canonical order.
fn propagate_death(env: &StepEnv<'_>, st: &mut SchedulerState, record: &mut DecisionRecord) {
    loop {
        let mut changed = false;
        for i in 0..env.plan.nodes.len() {
            if st.node_state[i] != NodeState::Dead {
                continue;
            }
            for &ei in &env.plan.out_edges[i] {
                let dst = env.plan.edges[ei].dst;
                if st.node_state[dst] == NodeState::Waiting
                    && !st.in_res[dst].contains_key(&ei)
                {
                    st.in_res[dst].insert(ei, EdgeRes::Dead);
                    record.edges.push((ei, st.node_gen[dst], EdgeOutcome::DeadPath));
                    refresh_readiness(env, st, dst);
                    changed = true;
                }
            }
        }
        if !changed {
            return;
        }
    }
}

/// The `Stalled` diagnosis: the lowest-index completed node whose out-edges
/// all resolved without ever firing.
fn stalled_culprit(env: &StepEnv<'_>, st: &SchedulerState) -> String {
    for i in 0..env.plan.nodes.len() {
        if st.node_state[i] == NodeState::DoneOk
            && !env.plan.out_edges[i].is_empty()
            && env.plan.out_edges[i].iter().all(|&ei| st.edge_fired[ei] == 0)
        {
            return env.plan.nodes[i].clone();
        }
    }
    // No completed node's own out-edges explain it (every completed node
    // fired something downstream) — name the lowest-index node that never
    // got a chance to run at all, the actual culprit, rather than blaming
    // the entry unconditionally when its own edge fired correctly.
    for i in 0..env.plan.nodes.len() {
        if st.node_state[i] == NodeState::Waiting {
            return env.plan.nodes[i].clone();
        }
    }
    env.plan.nodes[0].clone()
}

fn finish(st: &mut SchedulerState, out: &mut Vec<Command>, outcome: RunOutcome) {
    st.phase = Phase::Finished(outcome.clone());
    out.push(Command::Finish { outcome });
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    /// No cap = the transcript entry every run before this knob built. Not
    /// "close enough": byte-identical, so an unbounded run's checkpoints are
    /// unchanged and its `verify` still passes.
    #[test]
    fn no_cap_passes_the_result_through_verbatim() {
        let big = json!({"rows": vec!["x"; 5_000]});
        assert_eq!(bound_tool_content(&big, None, 3, 7), big);
    }

    /// The boundary, from both sides. At exactly the cap nothing happens; one
    /// character more and the entry becomes the bounded shape.
    #[test]
    fn the_cap_is_inclusive() {
        let at = Value::String("a".repeat(64));
        assert_eq!(bound_tool_content(&at, Some(64), 0, 0), at);

        let over = Value::String("a".repeat(65));
        let bounded = bound_tool_content(&over, Some(64), 3, 7);
        assert_eq!(bounded["truncated"], true);
        assert_eq!(bounded["chars"], 65, "the TRUE length, not the kept length");
        assert_eq!(bounded["head"].as_str().unwrap().chars().count(), 32);
        assert_eq!(bounded["tail"].as_str().unwrap().chars().count(), 32);
        // Where the whole thing still is. A model told "truncated" with no
        // pointer has learned nothing it can act on, and neither has a human
        // reading the transcript back.
        assert_eq!(bounded["journal"], json!({"attempt": 3, "effect_seq": 7}));
    }

    /// The cap counts CHARACTERS, which is what it is called. Slicing by bytes
    /// here would both mis-measure a multibyte result and be able to cut a
    /// UTF-8 sequence in half.
    #[test]
    fn a_multibyte_result_is_cut_on_character_boundaries() {
        // 40 characters, 120 bytes: a byte cap of 20 would land mid-sequence.
        let text: String = "日本語です".repeat(8);
        assert_eq!(text.chars().count(), 40);
        assert!(text.len() > 40, "the point of the case");

        let bounded = bound_tool_content(&Value::String(text.clone()), Some(20), 1, 2);
        assert_eq!(bounded["chars"], 40);
        let head = bounded["head"].as_str().unwrap();
        let tail = bounded["tail"].as_str().unwrap();
        assert_eq!(head.chars().count(), 10);
        assert_eq!(tail.chars().count(), 10);
        assert!(text.starts_with(head), "the head is a real prefix");
        assert!(text.ends_with(tail), "the tail is a real suffix");
    }

    /// A structured result is measured and shown as the JSON the model would
    /// have been handed, not as Rust's `Debug` of it.
    #[test]
    fn a_json_result_is_bounded_as_its_serialized_form() {
        let big = json!({"k": "v".repeat(200)});
        let bounded = bound_tool_content(&big, Some(40), 0, 1);
        assert_eq!(bounded["chars"], big.to_string().chars().count());
        assert!(bounded["head"].as_str().unwrap().starts_with(r#"{"k":"#), "{bounded}");
        assert!(bounded["tail"].as_str().unwrap().ends_with(r#""}"#), "{bounded}");
    }

    /// A degenerate cap must not panic or produce nonsense — it keeps nothing
    /// and still says how much there was and where it is.
    #[test]
    fn a_cap_of_zero_keeps_nothing_and_still_points_at_the_journal() {
        let bounded = bound_tool_content(&Value::String("abc".into()), Some(0), 9, 4);
        assert_eq!(bounded["chars"], 3);
        assert_eq!(bounded["head"], "");
        assert_eq!(bounded["tail"], "");
        assert_eq!(bounded["journal"]["effect_seq"], 4);
    }

    // ---- the fold's range decision -------------------------------------

    #[test]
    fn a_state_selection_keeps_the_steering_inbox_whatever_it_lists() {
        let input = serde_json::json!({"a": 1, "b": 2, "$inbox": [{"text": "stop"}]});
        assert_eq!(select_state(None, &input), Some(input.clone()));
        assert_eq!(select_state(Some(&serde_json::json!("all")), &input), Some(input.clone()));
        assert_eq!(
            select_state(Some(&serde_json::json!(["b", "zz"])), &input),
            Some(serde_json::json!({"b": 2, "$inbox": [{"text": "stop"}]}))
        );
        assert_eq!(
            select_state(Some(&serde_json::json!("none")), &input),
            Some(serde_json::json!({"$inbox": [{"text": "stop"}]}))
        );
        assert_eq!(select_state(Some(&serde_json::json!("none")), &serde_json::json!({"a": 1})), None);
    }

    #[test]
    fn a_response_name_is_what_every_provider_accepts() {
        assert_eq!(response_name("reply"), "reply");
        assert_eq!(response_name("ap.reply now"), "ap_reply_now");
        assert_eq!(response_name(&"n".repeat(80)).len(), 64);
        assert_eq!(response_name(""), "response");
    }

    fn msg(role: &str) -> Value {
        json!({"role": role, "content": "x"})
    }

    /// A short transcript has no middle: `messages[0]` is the node's input and
    /// the rest is the kept tail, so there is nothing a summary could replace.
    /// Returning `None` here is what produces `RUN-E024` rather than a fold
    /// that would delete the task description.
    #[test]
    fn nothing_to_fold_below_the_kept_tail() {
        for n in 0..=KEEP_TAIL + 1 {
            let messages: Vec<Value> = (0..n).map(|_| msg("user")).collect();
            assert_eq!(fold_range(&messages), None, "len {n} must not be foldable");
        }
    }

    /// The first entry is the node's instruction and input state. A summary of
    /// the task cannot stand in for the task, so the range always starts at 1.
    #[test]
    fn the_nodes_input_is_never_folded() {
        let messages: Vec<Value> = (0..12).map(|_| msg("user")).collect();
        let (from, to) = fold_range(&messages).expect("a middle exists");
        assert_eq!(from, 1, "messages[0] is the node's own input");
        assert_eq!(to, 12 - KEEP_TAIL);
    }

    /// Providers reject a `tool` result whose `assistant` tool call is not in
    /// the request. So the cut moves FORWARD off a tool entry — shrinking the
    /// kept tail — rather than backward, which would fold the assistant entry
    /// and orphan the results that follow it.
    #[test]
    fn the_cut_never_separates_a_tool_result_from_its_assistant_turn() {
        // Tail of 4 would start at index 8, which is a `tool`: the cut must
        // advance to 10, the next non-tool entry.
        let messages = vec![
            msg("user"),      // 0 node input
            msg("assistant"), // 1
            msg("tool"),      // 2
            msg("user"),      // 3
            msg("assistant"), // 4
            msg("tool"),      // 5
            msg("user"),      // 6
            msg("assistant"), // 7
            msg("tool"),      // 8  <- naive cut lands here
            msg("tool"),      // 9
            msg("user"),      // 10
            msg("assistant"), // 11
        ];
        let (from, to) = fold_range(&messages).expect("a middle exists");
        assert_eq!(from, 1);
        assert_eq!(to, 10, "advanced past both tool entries");
        assert_ne!(
            messages[to].get("role").and_then(Value::as_str),
            Some("tool"),
            "the entry AFTER the fold must never be an orphaned tool result"
        );
    }

    /// A round wide enough to fill the whole tail — one assistant turn issuing
    /// eight tool calls — folds ENTIRELY, keeping no tail at all. That is the
    /// right answer, not a missing guard: every candidate tail entry is a `tool`
    /// result whose `assistant` entry sits inside the folded range, so keeping
    /// any of them would orphan it. Fold all of it and make progress, rather
    /// than refuse and kill the node with `RUN-E024`.
    #[test]
    fn a_round_that_fills_the_tail_folds_entirely_rather_than_orphaning_it() {
        let mut messages = vec![msg("user"), msg("assistant")];
        messages.extend((0..8).map(|_| msg("tool")));
        assert_eq!(fold_range(&messages), Some((1, 10)));
    }
}