car_server_core/assistant/

agent_loop.rs

//! The assistant agent loop: propose → validate → execute → observe.
//!
//! One multi-turn tool-use conversation, driven by CAR inference and executed
//! through a [`Runtime`] (validator + policy + permission tiers + event log)
//! whose tool executor is the [`GeneralExecutor`]. The same loop backs the
//! one-shot CLI, the REPL, and — per turn — the `agent.chat` surface; streaming
//! is decoupled through a synchronous `emit` sink so a caller can forward events
//! to stdout or to `agent.chat.event` notifications without the loop knowing.
//!
//! [`Runtime`]: car_engine::Runtime
//! [`GeneralExecutor`]: super::executor::GeneralExecutor

use car_engine::{builtin_tool_labels, format_tool_result, tool_output_is_external, Runtime};
use car_inference::tasks::generate::{ContentBlock, Message, Provenance, ToolCall};
use car_inference::{GenerateParams, GenerateRequest};
use car_ir::{ActionProposal, ActionStatus};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use super::memory::MemoryTools;
use super::tool_memory::{approach_from_call, FailureSignature, ToolMemory, RECOVERY_WINDOW_TURNS};
use crate::coder::native_loop::TurnGenerator;

/// Upper bound (bytes) on a single tool observation fed back into context, so a
/// large read/output can't blow the request size. Truncates on a char boundary.
///
/// Public because it is the threshold the whole `value_store_previews` A/B
/// turns on: below it both arms render identically, so a benchmark suite whose
/// observations all fit measures nothing. `car-bench`'s harness suite has a
/// guard test asserting its held-out split still contains a task that clears
/// this number, and that test has to read the real constant — a copy of `16 *
/// 1024` in the bench would go stale the day this one moved, and the A/B would
/// go quietly inert again.
pub const OBSERVATION_CAP: usize = 16 * 1024;

/// The shipped default for [`AssistantConfig::value_store_previews`] — the one
/// place any production call site reads it from, so "what does CAR do out of the
/// box" is a single greppable fact rather than a literal repeated at the four
/// production construction sites (`car do` twice, the MCP assistant, and coder
/// discussions), which can drift apart. The remaining hard-coded `false`s in the
/// workspace are deliberate and stay: eight test fixtures in `chat.rs`, this
/// module's own off-arm fixture, and `HarnessBenchConfig`, whose `false` names
/// the *truncating arm* of the A/B rather than a product setting.
///
/// `no_production_call_site_hard_codes_the_preview_default` (here, and its twin
/// in `car-cli`) is what keeps that true — pinning the constant's value alone
/// would let a re-hardcoded literal reinstate the old default with every test
/// still green.
///
/// **`false`, confirmed by the measured A/B on 2026-08-27**, which is what
/// Parslee-ai/car#813 asked for before this shape changed. Three paired
/// replicates of `car-bench-harness --split all` (29 tasks, model
/// `anthropic/claude-sonnet-5`, seed 0), off arm versus `--value-store-previews`:
///
/// | mean over 3 reps | off (truncating) | on (retained previews) |
/// |---|---|---|
/// | pass rate | 0.967 | **1.000** |
/// | model calls / task | 2.235 | 2.407 (**+7.7%**) |
/// | tokens / task | 14,467.7 | **10,073.1 (−30.4%)** |
///
/// **#813's stated trigger did not fire:** it asked for fewer model calls at
/// equal-or-better pass rate without a token blow-up, and calls rose 7.7% while
/// tokens fell 30.4%. The run also did not count actual `$rN` resolutions, so it
/// cannot attribute the extra calls to handle use rather than repeated tools.
/// With one model, three replicates, fixed arm order, and substantial variance
/// on byte-identical below-cap tasks, the pass-rate and token results do not
/// justify overriding the calls criterion. The conservative measured decision
/// is therefore to keep the product default off and improve/re-measure the
/// preview format before any wider rollout.
///
/// The measurement, caveats, and how to re-run it:
/// `docs/solutions/value-store-previews-ab-2026-08-27.md`.
pub const VALUE_STORE_PREVIEWS_DEFAULT: bool = false;

/// Streamed events from one loop run. `emit` is called synchronously as the loop
/// progresses; a chat caller forwards these to `agent.chat.event`, a CLI caller
/// prints them.
pub enum AssistantEvent {
    /// Attribution for one completed model turn. Emitted before that turn's
    /// text/tool events so every rendered answer can be tied to its actual
    /// serving model, including transparent on-device degradation.
    ModelServed {
        model_id: String,
        local_last_resort: bool,
    },
    /// The model's free-text for a turn (may be empty when it only calls tools).
    Text(String),
    /// A tool is about to run.
    ToolCall { name: String, params: Value },
    /// A tool finished. `ok` is false for a failed/denied call.
    ToolResult {
        name: String,
        ok: bool,
        content: String,
    },
    /// Terminal: the model answered with no further tool calls.
    Done { text: String },
    /// Terminal: the run failed (inference/transport error).
    Error(String),
    /// Goal-loop verifier result after one iteration. This surfaces CAR's
    /// grounded completion evidence to CLI/chat hosts instead of hiding it in
    /// tracing logs.
    GoalEvaluated {
        iteration: u32,
        met: bool,
        grounded: bool,
        reason: String,
    },
}

/// Static configuration for a loop run.
#[derive(Clone)]
pub struct AssistantConfig {
    /// Model id, or `None` to let the router choose (pin a tool-capable model
    /// for real tool use — the local completion path ignores tools).
    pub model: Option<String>,
    /// Fail on the selected model rather than silently substituting another.
    /// Native chat enables this only for an explicit per-turn selection.
    pub strict_model: bool,
    /// Hard cap on loop turns.
    pub max_turns: u32,
    /// The model-visible tool list (from `GeneralExecutor::all_tool_defs()`).
    pub tools: Vec<Value>,
    /// Tool names that require human approval before running (the standing tier
    /// doesn't auto-allow them, e.g. writes/shell on the local host without
    /// `--full-access`). Empty when the tier auto-allows everything.
    pub gated_tools: Vec<String>,
    /// Optional per-agent approval policy. Given a tool name + params, returns
    /// whether to allow, require approval, or deny — the runtime enforcement of
    /// the `agent_permissions.*` posture for the running agent. When set it
    /// takes precedence over `gated_tools`; when `None`, `gated_tools` (the
    /// standing-tier list) applies, so existing callers are unchanged.
    pub approval_policy: Option<ApprovalPolicyFn>,
    /// Host-side proactive memory bank for the assistant loop. When present, CAR
    /// runs a deterministic memory-maintenance + selective-intervention pass
    /// before each model turn, so long-running agents do not depend on the model
    /// remembering to call `recall` at the right time.
    pub proactive_memory: Option<Arc<MemoryTools>>,
    /// Durable learned tool repairs. When present, the loop recalls what
    /// recovered this kind of tool failure last time, and records what recovers
    /// one this time — the difference between an agent that remembers and one
    /// that gets better (see [`super::tool_memory`]).
    ///
    /// `None` disables both halves. A surface that must be reproducible — the
    /// benchmark harness above all — leaves it unset deliberately: a run whose
    /// prompt depends on what the operator's assistant learned last Tuesday is
    /// not a measurement.
    pub tool_memory: Option<Arc<ToolMemory>>,
    /// Information-flow tool labels used to classify whether a tool result came
    /// from outside the trust boundary (car#723).
    ///
    /// `None` falls back to [`car_engine::builtin_tool_labels`], so the
    /// network-reaching commodity tools are always classified even when a caller
    /// supplies nothing. Deliberately `Option<_>` rather than a plain map with a
    /// `Default`: an empty map would silently classify everything as internal,
    /// and a security marking that a forgotten field can switch off is not one.
    /// Callers that load `.car/tool-labels.json` should pass the merged map so a
    /// project's own `trust: untrusted` declarations are honoured here too.
    pub tool_labels: Option<HashMap<String, car_verify::infoflow::ToolLabels>>,
    /// The run's task list, rendered into a per-turn state block at the tail of
    /// the request (Parslee-ai/car#814 items 2-3). `None` renders no block.
    pub todos: Option<Arc<tokio::sync::Mutex<super::todo::TodoList>>>,
    /// Retain tool results for the run and put a typed bounded preview in the
    /// transcript, instead of destructively truncating (Parslee-ai/car#813).
    ///
    /// Production call sites pass [`VALUE_STORE_PREVIEWS_DEFAULT`], which is
    /// `false` because the measured A/B did not meet #813's fewer-model-calls
    /// criterion. See that constant for the results and caveats.
    ///
    /// Still a field rather than a constant read inside the loop, because the
    /// bench needs both arms in one binary and a caller may want the old shape.
    /// With this `false` the observation path is byte-for-byte what it was:
    /// `cap()`, same cap, same notice.
    ///
    /// Below [`OBSERVATION_CAP`] the two arms *render* identically — the flag
    /// can only change an observation that crosses the cap. Note the flag also
    /// gates `SessionValues::resolve_refs` on every tool call, so once a handle
    /// exists a later below-cap call whose argument is a `$rN` reference is
    /// rewritten on the on arm only. That is a no-op until something over the
    /// cap has been retained; it is not the same statement as "nothing below
    /// the cap can ever differ".
    pub value_store_previews: bool,
    /// Constrain the run's FINAL answer to JSON. Applies to the answer, not
    /// to the work: it is NOT sent on any turn that offers tools, because a
    /// JSON-constrained request suppresses tool use on real providers (GLM
    /// 5.3 Flash answered in one turn without a single tool call under
    /// `json_object`; the same goal unconstrained ran two delegations and two
    /// reads and got it right). The loop instead checks the final
    /// no-tool-call answer itself and, only if it is not the requested shape,
    /// re-asks ONCE with no tools and `response_format` set — see
    /// [`final_text_matches_format`] and the repair branch. A run that offers
    /// no tools at all has nothing to suppress and carries the format on
    /// every turn. A final answer that already parses costs no extra call.
    ///
    /// `None` is the correct default for every caller that does not consume
    /// the answer as data. A knob that tightens the output contract, never a
    /// toggle between implementations (CLAUDE.md rule 1a). Provider-dependent:
    /// the Anthropic protocol rejects it up front (`UnsupportedMode`).
    pub response_format: Option<car_inference::ResponseFormat>,
    /// Override the context window (tokens) that bounds the running history
    /// each turn. `None` uses the registry's window for `model` (`0` when
    /// unknown, which disables compaction). Resolved through
    /// [`resolve_context_window`], which clamps a value ABOVE a known registry
    /// window back down to it: compaction exists to stop provider-side
    /// truncation of the original task (see `compact_history_to_window`), and
    /// a window larger than the real one would recreate exactly that. A value
    /// below the registry window is honored as-is — it only tightens.
    pub context_window_override: Option<usize>,
    /// Refuse any tool call whose name is not among `tools` with an error
    /// result, before approval or dispatch. Set on a `delegate` child so the
    /// tool subset it was granted holds at EXECUTION, not just advertisement:
    /// every tool is registered with the runtime, so a hallucinated call to an
    /// ungranted one would otherwise run (the same reason `run_task`'s GUI
    /// sub-agent enforces its restriction on the call, not the def).
    ///
    /// `false` everywhere else, so existing callers whose advertised list is
    /// deliberately narrower than the registry keep today's behavior.
    pub refuse_unadvertised_tools: bool,
    /// Validates a parsed final answer against the caller's JSON Schema.
    /// Carried as a closure so `car-server-core` needs no schema-validation
    /// dependency: `car-cli` compiles the `--json-schema` file with the
    /// `jsonschema` crate it already has and passes `Validator::is_valid`
    /// here. Without it a `JsonSchema` format is parse-only in the loop — and
    /// since tool turns never carry the format on the wire, NOBODY would
    /// enforce the schema on a tool-bearing run. Ignored for `JsonObject`.
    pub response_format_validator: Option<ResponseFormatValidator>,
    /// Run-level ceiling on `delegate` use. `None` applies
    /// [`DelegateBudget::default`] (20 delegations, 300 child turns); a call
    /// past either limit is an error result, never a spawn. Per parent run —
    /// children cannot delegate, so nothing nests under it.
    pub delegate_budget: Option<DelegateBudget>,
}

/// A compiled JSON-Schema check for the final answer — `true` when the parsed
/// answer conforms. See [`AssistantConfig::response_format_validator`].
pub type ResponseFormatValidator = Arc<dyn Fn(&Value) -> bool + Send + Sync>;

/// How much delegating one run may do, whatever each child's own cap says.
/// Bounds the model-call amplification a delegating parent can cause: without
/// it a parent at `max_turns` 50 could issue 50 children of 60 turns each.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DelegateBudget {
    /// Delegations that may be spawned in one run.
    pub max_delegations: u32,
    /// Child turns, summed over every delegation in the run.
    pub max_child_turns: u32,
}

impl Default for DelegateBudget {
    fn default() -> Self {
        Self {
            max_delegations: 20,
            max_child_turns: 300,
        }
    }
}

/// The effective context window for a run: the caller's override, clamped to
/// the registry window when that is known and smaller. Returns the window plus
/// an advisory when the clamp fired, so the caller can surface it.
///
/// `registry_window == 0` means "unknown" (local/test generators): an override
/// is then taken at face value, because there is nothing to clamp against and
/// the alternative — ignoring it — would silently disable the compaction the
/// caller asked for.
pub fn resolve_context_window(
    override_tokens: Option<usize>,
    registry_window: usize,
) -> (usize, Option<String>) {
    match override_tokens {
        None => (registry_window, None),
        Some(requested) if registry_window > 0 && requested > registry_window => (
            registry_window,
            Some(format!(
                "context window override {requested} exceeds the model's known window \
                 {registry_window}; using {registry_window} (a larger value would let the \
                 history overflow the real window and truncate the task provider-side, \
                 which is what compaction exists to prevent)"
            )),
        ),
        Some(requested) => (requested, None),
    }
}

/// Does a final answer satisfy the requested [`car_inference::ResponseFormat`]?
///
/// `JsonObject` requires a JSON *object*; `JsonSchema` requires parseable JSON
/// that also satisfies `validator` when the caller supplied one (the CLI
/// compiles the schema file and passes its check in) — parse-only without one,
/// deliberately not a partial schema re-implementation in this crate.
/// Tolerates a ```json fence around the payload, which models emit even under
/// JSON mode; the caller strips it via [`extract_json_payload`].
pub fn final_text_matches_format(
    text: &str,
    format: &car_inference::ResponseFormat,
    validator: Option<&ResponseFormatValidator>,
) -> bool {
    let payload = extract_json_payload(text);
    match format {
        car_inference::ResponseFormat::JsonObject => {
            serde_json::from_str::<Value>(payload).is_ok_and(|v| v.is_object())
        }
        car_inference::ResponseFormat::JsonSchema { .. } => {
            match serde_json::from_str::<Value>(payload) {
                // The schema itself, when the caller compiled one. Parse-only
                // otherwise (this crate carries no validator of its own).
                Ok(v) => validator.is_none_or(|is_valid| is_valid(&v)),
                Err(_) => false,
            }
        }
    }
}

/// Strip a single surrounding ``` / ```json fence, if present. Returns the
/// trimmed input otherwise.
pub fn extract_json_payload(text: &str) -> &str {
    let t = text.trim();
    let Some(rest) = t.strip_prefix("```") else {
        return t;
    };
    let Some(rest) = rest.strip_suffix("```") else {
        return t;
    };
    // Drop an optional language tag on the opening fence line.
    let rest = match rest.split_once('\n') {
        Some((tag, body)) if tag.trim().chars().all(|c| c.is_ascii_alphanumeric()) => body,
        _ => rest,
    };
    rest.trim()
}

/// The one-shot nudge sent when the final answer did not match the requested
/// format. Delivered as a user-role message: `Message::System` appears once at
/// the head of a conversation, and a mid-transcript system message is not a
/// shape every provider accepts.
const FORMAT_REPAIR_NUDGE: &str =
    "Your previous answer was not the requested JSON. Return only the JSON object — \
     no prose, no code fence, no tool calls.";
/// Schema variant of the nudge — a schema's root need not be an object, so it
/// must not say "object".
const FORMAT_REPAIR_NUDGE_SCHEMA: &str =
    "Your previous answer did not match the required JSON Schema. Return only JSON that \
     conforms to the schema — no prose, no code fence, no tool calls.";

/// The nudge for one format.
fn format_repair_nudge(format: &car_inference::ResponseFormat) -> &'static str {
    match format {
        car_inference::ResponseFormat::JsonObject => FORMAT_REPAIR_NUDGE,
        car_inference::ResponseFormat::JsonSchema { .. } => FORMAT_REPAIR_NUDGE_SCHEMA,
    }
}

/// Emitted (as [`AssistantEvent::Text`]) when the repair fires, so the run's
/// event stream records that the final answer was re-asked rather than taken
/// as returned.
pub const FORMAT_REPAIR_NOTICE: &str =
    "[format repair: final answer was not the requested JSON; re-asked the model once with no tools]";
/// Emitted when the single repair also missed. The repaired text is still the
/// answer — the contract is "one repair", not "retry until valid".
pub const FORMAT_REPAIR_STILL_INVALID: &str =
    "[format repair: the repaired answer still does not match the requested format; returning it as-is]";
/// Prefix of the notice emitted when the repair call itself failed (for
/// example an Anthropic-protocol model rejecting `response_format`). The DRAFT
/// answer is returned — a provider that cannot enforce the format is not a
/// reason to throw away the answer the model already gave.
pub const FORMAT_REPAIR_FAILED_PREFIX: &str = "[format repair failed:";

/// The loop-intercepted sub-agent tool. Advertised like any other tool (so a
/// caller can allowlist or omit it) but never dispatched to the executor: the
/// loop recognizes the name and runs a child loop in-process — the same shape
/// as `run_task`'s GUI sub-agent, chosen over a `ToolExecutor` because an
/// executor is built before the `Runtime` and `AssistantConfig` a child needs.
pub const DELEGATE_TOOL: &str = "delegate";
/// Child turn budget when the call names none.
pub const DELEGATE_DEFAULT_MAX_TURNS: u32 = 25;
/// Hard ceiling on a child's turn budget, whatever the call asks for.
pub const DELEGATE_MAX_TURNS_CAP: u32 = 60;

/// The names advertised in `tools`, minus [`DELEGATE_TOOL`] — what a child may
/// be granted. The delegate itself is excluded on purpose: no nesting in v1.
fn delegable_tool_names(tools: &[Value]) -> Vec<String> {
    tools
        .iter()
        .filter_map(|d| d.get("name").and_then(Value::as_str))
        .filter(|n| *n != DELEGATE_TOOL)
        .map(str::to_string)
        .collect()
}

/// Build the `delegate` tool def over the parent's advertised tools. The
/// `tools` parameter is a JSON-Schema `enum` of the parent's own tool names —
/// the verifiable subset precondition `car-multi`'s `spawn_subtask` uses — so
/// the validator rejects an escalation before the loop's own check does.
///
/// `tier: read_only`: the delegation itself changes nothing; the CHILD's calls
/// are what the gates see, and it inherits every gate the parent has.
/// `mutating: true` so a delegation that FINISHES counts as progress for the
/// no-progress guard (it is a real action, not a re-read); one that stalls,
/// errors, or hits its cap returns `ok: false` and does not.
pub fn delegate_tool_def(parent_tools: &[Value]) -> Value {
    let names = delegable_tool_names(parent_tools);
    json!({
        "name": DELEGATE_TOOL,
        "tier": "read_only",
        "mutating": true,
        "description": "Hand one self-contained sub-task to a fresh sub-agent that shares \
            your model, permissions, and working directory but starts with an EMPTY \
            transcript: it sees only the goal you write, not this conversation. Use it \
            to keep a long exploration or a noisy batch of tool output out of your own \
            context. It runs to completion before this call returns and you receive \
            ONLY its final written answer, so put everything it needs in `goal` and \
            ask it to report exactly what you need back. It cannot delegate further.",
        "parameters": {
            "type": "object",
            "properties": {
                "goal": {
                    "type": "string",
                    "description": "The single, self-contained task, with all the context the sub-agent needs and what to report back."
                },
                "tools": {
                    "type": "array",
                    "items": { "type": "string", "enum": names },
                    "description": "Tools to grant the sub-agent. Must be a subset of your own; omit for all of them."
                },
                "max_turns": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": DELEGATE_MAX_TURNS_CAP,
                    "description": "Turn budget for the sub-agent (default 25). It reports an error if it runs out."
                }
            },
            "required": ["goal"]
        }
    })
}

/// A parsed `delegate` call.
#[derive(Debug, Clone, PartialEq)]
pub struct DelegateRequest {
    pub goal: String,
    /// `None` = the parent's whole (delegable) set.
    pub tools: Option<Vec<String>>,
    pub max_turns: u32,
}

/// Parse the call's arguments. Shape errors are the model's to fix, so they
/// come back as an error result rather than sinking the run.
pub fn parse_delegate_params(params: &Value) -> Result<DelegateRequest, String> {
    let goal = params
        .get("goal")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|g| !g.is_empty())
        .ok_or("delegate needs a non-empty `goal` string")?
        .to_string();
    let tools = match params.get("tools") {
        None | Some(Value::Null) => None,
        Some(Value::Array(items)) => Some(
            items
                .iter()
                .map(|v| {
                    v.as_str().map(str::to_string).ok_or_else(|| {
                        "delegate `tools` must be an array of tool names".to_string()
                    })
                })
                .collect::<Result<Vec<_>, _>>()?,
        ),
        Some(_) => return Err("delegate `tools` must be an array of tool names".into()),
    };
    let max_turns = match params.get("max_turns") {
        None | Some(Value::Null) => DELEGATE_DEFAULT_MAX_TURNS,
        Some(v) => {
            let n = v
                .as_u64()
                .filter(|n| *n >= 1)
                .ok_or("delegate `max_turns` must be a positive integer")?;
            (n.min(DELEGATE_MAX_TURNS_CAP as u64)) as u32
        }
    };
    Ok(DelegateRequest {
        goal,
        tools,
        max_turns,
    })
}

/// Derive the child's config from the parent's. Everything is the parent's
/// (`clone()`) except:
/// * `tools` — the requested subset of the parent's delegable tools (default:
///   all of them), never including `delegate` itself. A name outside the
///   parent's set is an escalation and is refused here, mirroring
///   `spawn_subtask`'s defense-in-depth check behind its schema enum.
/// * `refuse_unadvertised_tools` — on, so the subset holds at execution.
/// * `max_turns` — the call's (capped) budget.
/// * `todos` — none; the parent's task list is not the child's.
/// * `response_format` — none; children answer in prose that the parent reads.
/// `gated_tools`, `approval_policy`, `model`, `strict_model`,
/// `context_window_override`, `proactive_memory`, `tool_labels` and
/// `value_store_previews` are inherited unchanged: a child can do nothing its
/// parent could not.
pub fn delegate_child_config(
    parent: &AssistantConfig,
    req: &DelegateRequest,
) -> Result<AssistantConfig, String> {
    let delegable = delegable_tool_names(&parent.tools);
    let requested: Vec<String> = match &req.tools {
        Some(list) => list.clone(),
        None => delegable.clone(),
    };
    let escalations: Vec<&String> = requested
        .iter()
        .filter(|t| !delegable.iter().any(|d| d == *t))
        .collect();
    if !escalations.is_empty() {
        let nested = escalations.iter().any(|t| *t == DELEGATE_TOOL);
        return Err(format!(
            "privilege escalation rejected: sub-agent tools {escalations:?} are not a subset of \
             your own tools{}",
            if nested {
                " (a sub-agent cannot delegate further)"
            } else {
                ""
            }
        ));
    }
    let tools: Vec<Value> = parent
        .tools
        .iter()
        .filter(|d| {
            d.get("name")
                .and_then(Value::as_str)
                .is_some_and(|n| requested.iter().any(|r| r == n))
        })
        .cloned()
        .collect();
    Ok(AssistantConfig {
        tools,
        refuse_unadvertised_tools: true,
        response_format_validator: None,
        delegate_budget: None,
        max_turns: req.max_turns,
        todos: None,
        response_format: None,
        ..parent.clone()
    })
}

/// The child's starting transcript: the parent's leading system prompt(s) and
/// the goal — nothing else from the parent. A fresh context is the point.
fn delegate_child_history(parent_messages: &[Message], goal: &str) -> Vec<Message> {
    let mut history: Vec<Message> = parent_messages
        .iter()
        .take_while(|m| matches!(m, Message::System { .. }))
        .cloned()
        .collect();
    history.push(Message::User {
        content: goal.to_string(),
    });
    history
}

/// What a finished delegation hands back to the parent's transcript.
struct DelegateOutcome {
    ok: bool,
    /// The tool-result content: the child's final text (capped) on success, a
    /// JSON error carrying the reason otherwise.
    content: String,
    turns: u32,
    /// Whether any of the child's tool results crossed the trust boundary,
    /// so the parent's `ToolResult` is marked accordingly.
    external: bool,
    /// The child's own receipts, for the parent to merge (tagged `via`).
    receipts: Vec<AssistantToolReceipt>,
    /// Whether a child loop actually ran (a parse or escalation refusal does
    /// not count against the run's delegation budget).
    spawned: bool,
}

/// Run one `delegate` call to completion. Sequential and in-process: the
/// parent's turn does not continue until the child returns.
///
/// Returns an explicitly boxed `dyn Future + Send` rather than being an
/// `async fn`: the loop awaits this, and this awaits the loop, so an inferred
/// future type would leave `Send` as an unsolvable cycle ("cannot satisfy …:
/// Send"). Naming the type here is what lets callers `tokio::spawn` the loop.
#[allow(clippy::too_many_arguments)]
fn run_delegate<'a>(
    generator: &'a dyn TurnGenerator,
    runtime: &'a Runtime,
    parent: &'a AssistantConfig,
    parent_messages: &'a [Message],
    params: &'a Value,
    cancel: &'a std::sync::atomic::AtomicBool,
    approval: Option<&'a dyn ApprovalGate>,
    runtime_session_id: Option<&'a str>,
    redrive_ungrounded_summary: bool,
    tool_labels: &'a HashMap<String, car_verify::infoflow::ToolLabels>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = DelegateOutcome> + Send + 'a>> {
    Box::pin(async move {
        let req = match parse_delegate_params(params) {
            Ok(r) => r,
            Err(e) => {
                return DelegateOutcome {
                    ok: false,
                    content: cap(json!({ "error": e }).to_string()),
                    turns: 0,
                    external: false,
                    receipts: Vec::new(),
                    spawned: false,
                }
            }
        };
        let child_cfg = match delegate_child_config(parent, &req) {
            Ok(c) => c,
            Err(e) => {
                return DelegateOutcome {
                    ok: false,
                    content: cap(json!({ "error": e }).to_string()),
                    turns: 0,
                    external: false,
                    receipts: Vec::new(),
                    spawned: false,
                }
            }
        };
        let mut child_messages = delegate_child_history(parent_messages, &req.goal);
        // The child's events stay inside the child: the parent's stream records
        // the delegation as ONE tool call + result (+ a one-line summary), which is
        // what a `--json` consumer can attribute. `&mut dyn FnMut` on purpose —
        // a fresh closure type here would re-instantiate the generic loop for
        // every nesting depth, and `Box::pin` is what lets an async fn recurse.
        let mut child_emit: &mut (dyn FnMut(AssistantEvent) + Send) = &mut |_| {};
        let child = run_assistant_loop_cancellable_in_session_durable(
            generator,
            runtime,
            &child_cfg,
            &mut child_messages,
            cancel,
            approval,
            None,
            runtime_session_id,
            None,
            None,
            redrive_ungrounded_summary,
            &mut child_emit,
        )
        .await;
        let external = child
            .tool_receipts
            .iter()
            .any(|r| tool_output_is_external(&r.tool, tool_labels));
        if child.status == "success" {
            DelegateOutcome {
                ok: true,
                content: cap(child.summary),
                turns: child.turns,
                external,
                receipts: child.tool_receipts,
                spawned: true,
            }
        } else {
            // An unfinished delegation must not read as an answer (the GUI
            // sub-agent's `is_error` rule): the cap, a stall, a cancel or a
            // transport error all come back as an error result with the reason.
            DelegateOutcome {
                ok: false,
                content: cap(json!({
                    "error": format!(
                        "delegate did not finish (status: {}) after {} turns: {}",
                        child.status, child.turns, child.summary
                    )
                })
                .to_string()),
                turns: child.turns,
                external,
                receipts: child.tool_receipts,
                spawned: true,
            }
        }
    })
}

/// Resolves a per-agent approval decision for a tool call. Built by the caller
/// (chat.rs) from the loaded `AgentPermissionPolicy` + the session's agent id +
/// a risk classifier, so the loop stays decoupled from the policy store.
pub type ApprovalPolicyFn =
    std::sync::Arc<dyn Fn(&str, &Value) -> ToolApprovalDecision + Send + Sync>;

/// What the per-agent policy says to do with a tool call before it runs.
pub enum ToolApprovalDecision {
    /// Auto-allow: run without asking.
    Allow,
    /// Require human approval (routes through the `ApprovalGate`).
    RequireApproval,
    /// Refuse outright with a reason.
    Deny(String),
}

/// The outcome of an approval request.
pub enum ApprovalDecision {
    Approved,
    Denied(String),
}

/// The human-in-the-loop seam. Consulted by the loop before running a
/// `gated_tools` action. Implementations: a terminal stdin prompt (REPL /
/// one-shot) or the chat `approval_pending` → park → resolve flow. When no gate
/// is wired, a gated action is denied with an actionable message.
#[async_trait::async_trait]
pub trait ApprovalGate: Send + Sync {
    async fn request(&self, tool: &str, params: &Value) -> ApprovalDecision;

    async fn request_action(&self, _call_id: &str, tool: &str, params: &Value) -> ApprovalDecision {
        self.request(tool, params).await
    }

    /// Durable write-ahead dispatch marker for an approved consequential
    /// action. A failure is fail-closed: the runtime must not execute.
    async fn before_dispatch(
        &self,
        _call_id: &str,
        _tool: &str,
        _params: &Value,
    ) -> Result<(), String> {
        Ok(())
    }

    /// Durable terminal action receipt. If this append fails, the prior
    /// dispatched record remains and resume classifies it indeterminate.
    async fn after_dispatch(
        &self,
        _call_id: &str,
        _tool: &str,
        _params: &Value,
        _ok: bool,
        _receipt: &Value,
    ) -> Result<(), String> {
        Ok(())
    }
}

/// The terminal result of a loop run.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AssistantModelAttribution {
    /// Canonical immutable id of the model that served the turn.
    pub model_id: String,
    /// True only when the appended on-device last resort actually served.
    pub local_last_resort: bool,
}

pub struct AssistantOutcome {
    /// `"success"` (model finished), `"max_turns"`, or `"error"`.
    pub status: &'static str,
    /// The final assistant text (or the error message).
    pub summary: String,
    /// Turns consumed.
    pub turns: u32,
    /// Names of tools that executed successfully.
    pub tools_called: Vec<String>,
    /// Tool executions attempted during this loop run, used to ground final
    /// prose claims such as "I ran the tests" against actual receipts.
    pub tool_receipts: Vec<AssistantToolReceipt>,
    /// Ordered attribution for every completed model call in this run. This is
    /// retained separately from the compactable transcript so the terminal run
    /// receipt remains complete even after older messages age out.
    pub models_served: Vec<AssistantModelAttribution>,
    /// The model id that produced the final turn (authoritative attribution;
    /// empty if no generate completed). Threaded out so the goal loop can stamp
    /// `model_id`/`model_tier` onto `GoalEvaluated`, mirroring the provenance
    /// `record_turn_completed` already stamps on the default path.
    pub model_used: String,
}

#[derive(Clone, Debug)]
pub struct AssistantToolReceipt {
    pub tool: String,
    pub call_id: Option<String>,
    pub ok: bool,
    pub params: Value,
    /// `Some("delegate:<call id>")` for a receipt a `delegate` child produced
    /// and the parent merged into its own list, so grounding and
    /// `receipts.by_tool` see the child's real calls while a reader can still
    /// tell them from the parent's own. `None` for the parent's own calls.
    pub via: Option<String>,
}

fn transcript_tool_receipts(messages: &[Message]) -> Vec<AssistantToolReceipt> {
    let mut calls: std::collections::HashMap<String, (String, Value)> =
        std::collections::HashMap::new();
    let mut receipts = Vec::new();
    for message in messages {
        match message {
            Message::Assistant { tool_calls, .. } => {
                for call in tool_calls {
                    if let Some(id) = call.id.as_deref() {
                        calls.insert(
                            id.to_string(),
                            (
                                call.name.clone(),
                                serde_json::to_value(&call.arguments)
                                    .unwrap_or_else(|_| Value::Object(Default::default())),
                            ),
                        );
                    }
                }
            }
            Message::ToolResult {
                tool_use_id,
                content,
                ..
            } => {
                let Some((tool, params)) = calls.get(tool_use_id).cloned() else {
                    continue;
                };
                let parsed = serde_json::from_str::<Value>(content).ok();
                let ok = parsed
                    .as_ref()
                    .map(|value| {
                        value.get("error").is_none()
                            && value.get("ok").and_then(Value::as_bool) != Some(false)
                            && value.get("status").and_then(Value::as_str) != Some("Failed")
                    })
                    .unwrap_or_else(|| {
                        let lower = content.to_ascii_lowercase();
                        !lower.contains("declined by user")
                            && !lower.contains("tool call denied")
                            && !lower.starts_with("error:")
                    });
                receipts.push(AssistantToolReceipt {
                    tool,
                    call_id: Some(tool_use_id.clone()),
                    ok,
                    params,
                    via: None,
                });
            }
            _ => {}
        }
    }
    receipts
}

/// Bound an observation to [`OBSERVATION_CAP`], stating what was lost (#813).
///
/// This is still destructive truncation — the elided bytes are NOT retained,
/// and recovering them means re-running the tool with a narrower query. The
/// full fix is a session value store plus typed previews with handles, which
/// #813 rightly says needs its own design pass and a `car-bench` A/B before it
/// changes the model-facing transcript shape.
///
/// What is fixable without that: the marker used to be a bare `…[truncated]…`,
/// so a model could not tell whether it had lost 10 bytes or 10 MB, and had no
/// signal that re-running was the only recovery. A model reasoning over a
/// clipped table would silently treat it as complete. Reporting the true size
/// and the elided amount costs nothing and makes the loss legible.
fn cap(mut s: String) -> String {
    let total = s.len();
    if total <= OBSERVATION_CAP {
        return s;
    }
    let mut end = OBSERVATION_CAP;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    let elided = total - end;
    s.truncate(end);
    // Leading newline so the notice can't be mistaken for part of the payload
    // (a clipped CSV row, a half-written JSON object).
    s.push_str(&format!(
        "\n…[truncated: showing first {end} of {total} bytes; {elided} bytes elided \
         and NOT retained. To see the rest, re-run this tool with a narrower \
         query — the elided bytes cannot be recovered by asking for them.]…"
    ));
    s
}

/// Fences for the per-turn runtime state block (#814 items 2-3). Explicit
/// delimiters because the block is appended to a message that is usually a tool
/// result, and unfenced runtime text there would read as part of the tool's
/// output.
const STATE_BLOCK_OPEN: &str = "\n\n<runtime-state>\n";
const STATE_BLOCK_CLOSE: &str = "\n</runtime-state>";

/// Append the per-turn state block to the LAST message's content.
///
/// Appended to an existing message rather than added as a new one, which is the
/// only placement that actually satisfies #814 item 3 on every provider. Item 3
/// asks for the tail so the cached prefix stays byte-stable — but the Anthropic
/// and Gemini handlers FOLD every `Message::System` into the top-level system
/// field (`protocol.rs`), so a trailing System block would land in the prefix
/// and be rewritten every turn, causing precisely the cache invalidation the
/// item exists to prevent. A trailing `Message::User` would keep its position,
/// but after a tool result it produces consecutive user-role turns, which is a
/// provider-shape risk not worth taking for a status line.
///
/// Operates on the request copy, never the durable history: the block is
/// regenerated every turn, so persisting it would stack stale copies.
fn append_state_block(messages: &mut [Message], block: &str) {
    let Some(last) = messages.last_mut() else {
        return;
    };
    let fenced = format!("{STATE_BLOCK_OPEN}{block}{STATE_BLOCK_CLOSE}");
    match last {
        Message::System { content }
        | Message::User { content }
        | Message::Assistant { content, .. }
        | Message::ToolResult { content, .. } => content.push_str(&fenced),
        // No text slot to append to; skipping is better than restructuring the
        // turn, and the next turn's message will carry the block.
        _ => {}
    }
}

/// How many remembered subjects reach the state block. Bounded deliberately:
/// this is a pointer to durable state, not a copy of it.
const STATE_BLOCK_MAX_FACTS: usize = 5;

/// Subjects of the facts this run wrote via `remember`, oldest first.
///
/// Derived from the run's tool receipts rather than from a second tracker: the
/// loop already records every call with its params, so there is nothing to keep
/// in sync and no way for the two views to disagree. Only *successful* calls
/// count — a rejected `remember` wrote nothing, and listing it would tell the
/// model it knows something it does not.
fn recent_fact_subjects(receipts: &[AssistantToolReceipt]) -> Vec<String> {
    let mut subjects: Vec<String> = Vec::new();
    for receipt in receipts.iter().filter(|r| r.ok && r.tool == "remember") {
        let Some(subject) = receipt.params.get("subject").and_then(Value::as_str) else {
            continue;
        };
        let subject = subject.trim();
        if subject.is_empty() {
            continue;
        }
        // A re-remember supersedes the earlier write rather than adding a second
        // fact (`memory.rs`), so the subject moves to the most-recent position
        // instead of appearing twice and inflating the count.
        subjects.retain(|s| s != subject);
        subjects.push(subject.to_string());
    }
    subjects
}

/// Compose the per-turn state block from live run state, or `None` when there
/// is nothing worth spending tokens on.
///
/// Subjects only, never bodies. The block exists so the model knows a fact
/// *exists* without having to remember writing it — turning a speculative
/// `recall` into an informed one. Inlining the bodies would duplicate memgine's
/// job without its relevance ranking, and would let the block grow into the
/// largest thing in the context, which is the failure mode the whole per-turn
/// design is bounded against.
fn render_state_block(todo: Option<String>, facts: &[String]) -> Option<String> {
    let mut sections: Vec<String> = Vec::new();
    if let Some(todo) = todo {
        sections.push(todo);
    }
    if !facts.is_empty() {
        // Keep the most RECENT subjects when over the cap — the oldest are the
        // ones the model is least likely to still be acting on.
        let hidden = facts.len().saturating_sub(STATE_BLOCK_MAX_FACTS);
        let listed = facts
            .iter()
            .skip(hidden)
            .map(String::as_str)
            .collect::<Vec<_>>()
            .join(", ");
        let mut line = format!("remembered this run: {listed}");
        if hidden > 0 {
            line.push_str(&format!(" (+{hidden} earlier)"));
        }
        line.push_str("\n  subjects only — call `recall` for the content");
        sections.push(line);
    }
    (!sections.is_empty()).then(|| sections.join("\n"))
}

/// Keep at least this many of the most-recent messages when compacting, so a
/// window-bounded run never loses the immediate working context.
const HISTORY_MIN_TAIL: usize = 6;

/// The share of a model's context window a running history may occupy before
/// compaction fires, as an exact integer fraction: 3/4 (75%).
///
/// The remaining quarter is the headroom for the model's own output and the
/// next turn's tool results — there is no separate reserve. This was spelled
/// `context_window / 4 * 3` inside [`compact_history_measured`], which read as
/// an implementation detail of that function rather than what it is: the
/// harness-wide policy for every multi-turn driver (the assistant loop, the
/// coder native loop, the declarative-agent runner). Naming it makes moving
/// the number a one-line decision instead of a grep, and makes each driver's
/// budget nameable in its own logs.
///
/// Evaluate as `window / DENOMINATOR * NUMERATOR` — integer division first, the
/// order the original expression used, so no window changes its budget by a
/// token.
pub(crate) const HISTORY_BUDGET_NUMERATOR: usize = 3;
/// Denominator of [`HISTORY_BUDGET_NUMERATOR`]'s fraction.
pub(crate) const HISTORY_BUDGET_DENOMINATOR: usize = 4;

/// Tokens of `context_window` a history may occupy before compaction drops its
/// oldest middle turns. `0` in (unknown window) is `0` out — callers read that
/// as "no bound is known", not "no tokens allowed".
pub(crate) fn history_budget(context_window: usize) -> usize {
    context_window / HISTORY_BUDGET_DENOMINATOR * HISTORY_BUDGET_NUMERATOR
}

/// Opening of the system message left in place of compacted turns (#815).
///
/// Doubles as the marker's own identity: [`parse_compaction_notice`] recognizes
/// it so a later compaction updates the running totals in place instead of
/// stacking notices or dropping the earlier one.
const COMPACTION_NOTICE_PREFIX: &str = "[history compacted:";

/// Where a compacted run's dropped turns can still be read — which differs by
/// caller, so the notice must too.
///
/// The assistant and coder loops run against a live event log and advertise
/// `events_query`, so their notice can tell the model where to look. The
/// declarative-agent runner has neither: its tools come from the spec's
/// allowlist over a `WorktreeExecutor`, and no event log is bound to the run.
/// Pointing that model at `events_query` would be a false recovery path — a
/// tool it cannot call, naming a log that does not exist — which is worse than
/// saying plainly that the turns are gone.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(crate) enum CompactionRecovery {
    /// The run has an event log and the `events_query` tool. Default: this is
    /// what every caller before the declarative runner had.
    #[default]
    EventsQuery,
    /// Nothing to point at — the dropped turns are gone for this run.
    Unrecoverable,
}

/// Render the marker. The `EventsQuery` arm names `events_query` explicitly
/// because a notice that something is missing, without saying how to look, only
/// converts a silent failure into a visible dead end; the `Unrecoverable` arm
/// says exactly that, for a caller where there is nothing to look in.
///
/// Both arms keep the `{COMPACTION_NOTICE_PREFIX} <turns> … ~<tokens> …` shape
/// so [`parse_compaction_notice`] reads either one back and totals accumulate
/// across compactions regardless of which caller wrote them.
fn format_compaction_notice(turns: usize, tokens: usize, recovery: CompactionRecovery) -> String {
    match recovery {
        CompactionRecovery::EventsQuery => format!(
            "{COMPACTION_NOTICE_PREFIX} {turns} earlier turns removed to fit the context \
             window, ~{tokens} tokens. They are gone from this transcript but the run's \
             event log still has them — call `events_query` (e.g. {{\"kinds\": \
             [\"action_failed\"], \"limit\": 5}}) to see what was already tried, rather \
             than assuming you never tried it.]"
        ),
        CompactionRecovery::Unrecoverable => format!(
            "{COMPACTION_NOTICE_PREFIX} {turns} earlier turns dropped to fit the model's \
             context window, ~{tokens} tokens. They are not recoverable in this run — \
             work from what is still in this transcript, and do not assume something was \
             never tried just because you cannot see it.]"
        ),
    }
}

/// Recover the running totals from an existing marker, or `None` if `message`
/// is not one.
///
/// Parses the numbers back out of the text rather than threading a counter
/// through every caller: the marker lives in `messages`, which is the only
/// state both callers of `compact_history_to_window` (the assistant loop and
/// the coder loop) already share.
fn parse_compaction_notice(message: &Message) -> Option<(usize, usize)> {
    let Message::System { content } = message else {
        return None;
    };
    let rest = content.strip_prefix(COMPACTION_NOTICE_PREFIX)?;
    let turns: usize = rest.split_whitespace().next()?.parse().ok()?;
    let tokens: usize = rest
        .split('~')
        .nth(1)?
        .split_whitespace()
        .next()?
        .parse()
        .ok()?;
    Some((turns, tokens))
}

/// A message's estimated token cost, using the *same* metric the inference
/// layer's window-fit check uses (`media_tokens::messages_history_tokens`: a
/// message's serialized-JSON length / 4, with calibrated media accounting).
/// Sharing one estimator is load-bearing — if compaction under-counts relative
/// to the truncation check, it leaves a history the check still flags.
fn approx_message_tokens(m: &Message) -> usize {
    car_inference::media_tokens::messages_history_tokens(std::slice::from_ref(m))
}

/// Bound the running conversation to the model's context window so a long,
/// tool-heavy run never overfills it. An overflowed history pushes the model to
/// its context limit and can truncate the *original task* provider-side — the
/// exact failure the gpt-5.5 benchmark run hit (17× `available_tokens=0`).
///
/// Deterministic sliding window: keep the system prompt(s) + the original task
/// (first user turn) + the most-recent exchanges, dropping the oldest middle
/// messages until the estimate fits [`history_budget`] (headroom for the
/// model's output + the next tool results). Drops land on a turn boundary so a
/// `ToolResult` is never orphaned from the `Assistant` call that produced it —
/// a dangling tool result is provider-invalid. No-op when the window is unknown
/// (`0`, e.g. local/test generators) or the history already fits.
pub(crate) fn compact_history_to_window(messages: &mut Vec<Message>, context_window: usize) {
    compact_history_measured(messages, context_window, PromptMeasure::default())
}

/// What a caller knows about the size of its next request beyond the
/// per-message chars/4 estimate. `Default` is "nothing": the estimate alone.
///
/// The estimate under-counts a real request in two ways this closes. It sees
/// only `messages` — not the tool definitions advertised on every turn
/// (`fixed_overhead`) — and it counts bytes/4 where code tokenizes denser, so
/// a 150k-window run measured ~100k while the provider billed 144k and
/// compaction never fired. The provider's own `prompt_tokens` from the
/// previous call is ground truth for the messages that call carried
/// (`reported`); what was appended since is estimated, scaled by the
/// observed reported/estimated ratio when the estimate was off by more than
/// 25%, so one compaction lands under budget instead of chasing it.
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct PromptMeasure {
    /// Per-request tokens the per-message estimate cannot see: the tool
    /// definitions (and anything else fixed per call).
    pub fixed_overhead: usize,
    /// `(prompt_tokens, messages_covered)` from the most recent completed
    /// call: the provider-reported input size (all cache buckets summed) and
    /// how many leading messages of the current history that request
    /// carried. Everything past `messages_covered` was appended since.
    pub reported: Option<(usize, usize)>,
}

/// Per-message token estimates for `messages`, in the order given — the input
/// both [`measure_scale`] and the drop loop walk.
///
/// One function so two callers cannot accidentally estimate differently, and so
/// it is visible that estimating is a PASS over the messages: the path-backed
/// media estimator reads filesystem metadata, so re-deriving these is neither
/// free nor order-neutral.
pub(crate) fn message_estimates(messages: &[Message]) -> Vec<usize> {
    messages.iter().map(approx_message_tokens).collect()
}

/// The ratio that maps this run's per-message estimates onto the provider's own
/// accounting: `reported / estimate-of-the-same-messages`.
///
/// **The gate is upward-only.** A scale is learned ONLY when the provider's
/// reported usage exceeds 125% of the estimate for the same messages; anything
/// at or below that — a report that merely agrees, and a report SMALLER than
/// the estimate — leaves the scale at `1.0`. The estimator is known to
/// under-count (bytes/4 against a denser tokenizer), so the correction it
/// exists to make is upward; scaling a history DOWN toward a low report would
/// compact less than the window needs, which is the failure this path prevents.
///
/// Takes the estimates rather than the messages so the compaction pass, which
/// already has them, does not pay for a second estimation pass over the covered
/// prefix — that pass is filesystem metadata reads for path-backed media.
/// `covered` is clamped to the slice.
pub(crate) fn measure_scale(estimates: &[usize], measure: PromptMeasure) -> f64 {
    let Some((reported, covered)) = measure.reported else {
        return 1.0;
    };
    let covered = covered.min(estimates.len());
    let covered_est: usize = estimates[..covered].iter().sum::<usize>() + measure.fixed_overhead;
    if covered_est > 0 && reported * 4 > covered_est * 5 {
        reported as f64 / covered_est as f64
    } else {
        1.0
    }
}

/// What `messages` cost in the provider's accounting: the shared request-level
/// estimate plus the caller's fixed overhead, scaled by [`measure_scale`].
///
/// One number for "how big is this history", so two passes over the same
/// history cannot disagree about whether it fits. Identical to the total
/// [`compact_history_measured`] decides on wherever a scale was learned, since
/// `scale * covered_estimate` IS the reported count by construction.
pub(crate) fn scaled_prompt_tokens(
    messages: &[Message],
    fixed_overhead: usize,
    scale: f64,
) -> usize {
    let estimate =
        car_inference::media_tokens::request_prompt_tokens("", None, None, None, Some(messages))
            + fixed_overhead;
    (estimate as f64 * scale).round() as usize
}

/// Same as [`compact_history_to_window`] with a [`PromptMeasure`]. WHETHER to
/// compact is decided on the best available measure — the provider-reported
/// count when there is one — while HOW MUCH to drop still walks the
/// per-message estimates (the only per-message measure), scaled to the
/// reported total when the two disagree by more than 25%.
pub(crate) fn compact_history_measured(
    messages: &mut Vec<Message>,
    context_window: usize,
    measure: PromptMeasure,
) {
    compact_history_measured_with_recovery(
        messages,
        context_window,
        measure,
        CompactionRecovery::default(),
    )
}

/// Same as [`compact_history_measured`] with an explicit
/// [`CompactionRecovery`], for a caller whose run cannot honor the default
/// notice's `events_query` advice. Only the notice text differs — WHAT is
/// dropped, and the tail rule that protects it, are identical.
pub(crate) fn compact_history_measured_with_recovery(
    messages: &mut Vec<Message>,
    context_window: usize,
    measure: PromptMeasure,
    recovery: CompactionRecovery,
) {
    if context_window == 0 {
        return;
    }
    let budget = history_budget(context_window);
    let estimates: Vec<usize> = message_estimates(messages);
    // The fallback total is the shared request-level estimate over the
    // history (`media_tokens::request_prompt_tokens`, the same function the
    // remote guard's warning uses) plus the per-run overhead the caller
    // measured once (tool definitions). The per-message vector above is the
    // same estimator applied one message at a time, for the drop loop.
    let estimated: usize =
        car_inference::media_tokens::request_prompt_tokens("", None, None, None, Some(messages))
            + measure.fixed_overhead;
    // `scale` maps a per-message estimate onto the provider's accounting.
    let (total, scale, reported) = match measure.reported {
        Some((reported, covered)) => {
            let covered = covered.min(messages.len());
            let appended: usize = estimates[covered..].iter().sum();
            let scale = measure_scale(&estimates, measure);
            let appended_scaled = (appended as f64 * scale).round() as usize;
            (reported + appended_scaled, scale, Some(reported))
        }
        None => (estimated, 1.0, None),
    };
    let scaled = |tokens: usize| (tokens as f64 * scale).round() as usize;
    if total <= budget {
        return;
    }
    tracing::info!(
        reported_prompt_tokens = reported,
        estimated_prompt_tokens = estimated,
        measured_prompt_tokens = total,
        scale,
        budget,
        context_window,
        "history exceeds the compaction budget"
    );

    // Pinned head: leading system prompt(s) + the first user turn (the task).
    let mut head_end = 0;
    while head_end < messages.len() && matches!(messages[head_end], Message::System { .. }) {
        head_end += 1;
    }
    if head_end < messages.len()
        && matches!(
            messages[head_end],
            Message::User { .. } | Message::UserMultimodal { .. }
        )
    {
        head_end += 1;
    }
    // A notice from an earlier compaction is part of the pinned head (#815).
    // Otherwise it sits first in the drop range and the next compaction erases
    // the record that the previous one happened — restoring exactly the silent
    // deletion the marker exists to prevent.
    let existing_notice = messages
        .get(head_end)
        .and_then(parse_compaction_notice)
        .map(|totals| {
            let at = head_end;
            head_end += 1;
            (at, totals)
        });

    // Never drop into the most-recent tail.
    if messages.len().saturating_sub(head_end) <= HISTORY_MIN_TAIL {
        return;
    }
    let max_drop = messages.len() - HISTORY_MIN_TAIL;

    // Drop oldest middle messages until we fit (or run into the tail).
    let mut drop_end = head_end;
    let mut running = total;
    while running > budget && drop_end < max_drop {
        running = running.saturating_sub(scaled(estimates[drop_end]));
        drop_end += 1;
    }
    // Land the kept suffix on a valid turn boundary. A Responses continuity
    // item precedes its Assistant message, so it is a valid boundary only as
    // that pair. If the item itself was just dropped, drop its now-orphaned
    // Assistant too; then skip any dangling tool results.
    if drop_end > head_end
        && drop_end < messages.len()
        && matches!(messages[drop_end - 1], Message::ProviderOutputItems { .. })
        && matches!(messages[drop_end], Message::Assistant { .. })
    {
        drop_end += 1;
    }
    while drop_end < messages.len() && matches!(messages[drop_end], Message::ToolResult { .. }) {
        drop_end += 1;
    }
    if drop_end <= head_end {
        return;
    }
    let dropped = drop_end - head_end;
    // Accounted in the same (scaled) measure the decision used, so the
    // notice's `~N tokens` is the provider-side size of what was removed.
    let dropped_tokens: usize = estimates[head_end..drop_end]
        .iter()
        .map(|t| scaled(*t))
        .sum();
    messages.drain(head_end..drop_end);
    // Leave a marker where the turns were (#815).
    //
    // Without one, turns simply cease to exist between one request and the
    // next and the transcript reads as continuous from the model's side — so a
    // run that degrades after compaction is indistinguishable, in the trace,
    // from a model that just got worse. "The model forgot" and "the harness
    // deleted it" are different bugs with different fixes, and only one of them
    // is the model's.
    //
    // Pinned into the head below so the next compaction cannot silently drop
    // the notice that the previous one happened, and totals accumulate across
    // compactions rather than only reporting the latest.
    match existing_notice {
        Some((at, (prior_turns, prior_tokens))) => {
            messages[at] = Message::System {
                content: format_compaction_notice(
                    prior_turns + dropped,
                    prior_tokens + dropped_tokens,
                    recovery,
                ),
            };
        }
        None => messages.insert(
            head_end,
            Message::System {
                content: format_compaction_notice(dropped, dropped_tokens, recovery),
            },
        ),
    }
    tracing::debug!(
        dropped_messages = dropped,
        kept = messages.len(),
        context_window,
        budget,
        "compacted assistant history to fit the model context window"
    );
}

fn message_memory_text(message: &Message) -> Option<String> {
    match message {
        Message::System { content }
        | Message::User { content }
        | Message::Assistant { content, .. }
        | Message::ToolResult { content, .. } => {
            let trimmed = content.trim();
            (!trimmed.is_empty()).then(|| trimmed.to_string())
        }
        Message::UserMultimodal { content } => {
            let text = content
                .iter()
                .filter_map(|block| match block {
                    ContentBlock::Text { text } => Some(text.trim()),
                    _ => None,
                })
                .filter(|s| !s.is_empty())
                .collect::<Vec<_>>()
                .join("\n");
            (!text.is_empty()).then_some(text)
        }
        _ => None,
    }
}

fn proactive_query_from_messages(messages: &[Message]) -> String {
    messages
        .iter()
        .rev()
        .find_map(|m| match m {
            Message::User { content } => {
                let trimmed = content.trim();
                (!trimmed.is_empty()).then(|| trimmed.to_string())
            }
            Message::UserMultimodal { .. } => message_memory_text(m),
            _ => None,
        })
        .unwrap_or_default()
}

fn append_context_block(req: &mut GenerateRequest, title: &str, body: &str) {
    let block = format!("## {title}\n{body}");
    req.context = Some(match req.context.take() {
        Some(existing) if !existing.trim().is_empty() => format!("{existing}\n\n{block}"),
        _ => block,
    });
}

fn proactive_maintenance_event_data(
    report: &car_memgine::ProactiveMaintenanceReport,
) -> std::collections::HashMap<String, Value> {
    let mut data = proactive_trigger_event_data(&report.trigger);
    data.insert(
        "saved_count".to_string(),
        Value::from(report.saved.len() as u64),
    );
    data.insert(
        "skipped_existing".to_string(),
        Value::from(report.skipped_existing as u64),
    );
    data.insert(
        "status_updated".to_string(),
        Value::from(report.status.is_some()),
    );
    data
}

fn proactive_intervention_event_data(
    decision: &car_memgine::ProactiveMemoryDecision,
) -> std::collections::HashMap<String, Value> {
    let mut data = std::collections::HashMap::new();
    match decision {
        car_memgine::ProactiveMemoryDecision::Inject {
            selected,
            candidates,
            bank,
            ..
        } => {
            data.insert("decision".to_string(), Value::from("inject"));
            data.insert("selected_id".to_string(), Value::from(selected.id.clone()));
            data.insert(
                "selected_kind".to_string(),
                Value::from(format!("{:?}", selected.kind).to_ascii_lowercase()),
            );
            data.insert(
                "candidate_count".to_string(),
                Value::from(candidates.len() as u64),
            );
            data.insert(
                "bank_knowledge".to_string(),
                Value::from(bank.knowledge as u64),
            );
            data.insert(
                "bank_procedural".to_string(),
                Value::from(bank.procedural as u64),
            );
            data.insert(
                "bank_open_subgoals".to_string(),
                Value::from(bank.open_subgoals as u64),
            );
        }
        car_memgine::ProactiveMemoryDecision::Silent {
            reason,
            candidates,
            bank,
        } => {
            data.insert("decision".to_string(), Value::from("silent"));
            data.insert("reason".to_string(), Value::from(reason.clone()));
            data.insert(
                "candidate_count".to_string(),
                Value::from(candidates.len() as u64),
            );
            data.insert(
                "bank_knowledge".to_string(),
                Value::from(bank.knowledge as u64),
            );
            data.insert(
                "bank_procedural".to_string(),
                Value::from(bank.procedural as u64),
            );
            data.insert(
                "bank_open_subgoals".to_string(),
                Value::from(bank.open_subgoals as u64),
            );
        }
    }
    data
}

fn proactive_trigger_event_data(
    trigger: &car_memgine::ProactiveMemoryTrigger,
) -> std::collections::HashMap<String, Value> {
    std::collections::HashMap::from([
        (
            "repeated_failures".to_string(),
            Value::from(trigger.repeated_failures as u64),
        ),
        ("tool_error".to_string(), Value::from(trigger.tool_error)),
        (
            "explicit_uncertainty".to_string(),
            Value::from(trigger.explicit_uncertainty),
        ),
        (
            "high_risk_action".to_string(),
            Value::from(trigger.high_risk_action),
        ),
        (
            "context_shift".to_string(),
            Value::from(trigger.context_shift),
        ),
    ])
}

/// Record one completed model call into the run's event log as
/// [`car_eventlog::EventKind::InferenceMetered`].
///
/// Until this existed the assistant loop wrote **no** inference telemetry, so
/// `harness_metrics::compute_harness_metrics` over an assistant journal reported
/// zero tokens and zero model calls — which made the whole
/// `trajectory_efficiency` token branch of the Evolution Agent's regression gate
/// (`car_memgine::harness_evolution`) structurally inert. The data was never
/// missing: [`car_inference::InferenceResult`] has carried `usage` and
/// `latency_ms` all along and the loop simply dropped them.
///
/// Uses the log handle the loop already holds (`runtime.log`) — the same
/// mechanism `maybe_apply_assistant_proactive_memory` uses for its
/// `ProactiveMemoryMaintained` record, and the same `append_metered` contract
/// the streaming path in `handler.rs` uses. No new plumbing, no new field on any
/// signature.
///
/// **`usage: None` is recorded as a call with no token metrics, not as zeros.**
/// `car-inference` makes `usage` optional precisely so "nobody could count" is
/// distinguishable from "this really used no tokens" (Parslee-ai/car#795), and
/// `Metrics::latency` leaves the token keys absent from the event data rather
/// than writing `0`. The event is still emitted, because a call that happened
/// with an uncountable cost is still a call: skipping it would silently
/// undercount `model_calls`, the exact number the #813 A/B turns on.
///
/// `cost_usd` is left `None`: the loop has no price table, and a fabricated 0.0
/// would be indistinguishable from a free call. The gate does not read cost.
async fn record_inference_metered(runtime: &Runtime, result: &car_inference::InferenceResult) {
    let mut data: HashMap<String, Value> = HashMap::new();
    data.insert(
        "model_id".to_string(),
        Value::from(result.served_model_id().to_string()),
    );
    // Explicit provenance for the absent-vs-zero distinction above, so a
    // consumer reading the journal need not infer it from missing keys.
    data.insert(
        "usage_measured".to_string(),
        Value::from(result.usage.is_some()),
    );

    let metrics = match &result.usage {
        Some(u) => car_eventlog::Metrics::inference(u.prompt_tokens, u.completion_tokens, None)
            .with_duration(result.latency_ms as f64),
        None => car_eventlog::Metrics::latency(result.latency_ms as f64),
    };

    runtime.log.lock().await.append_metered(
        car_eventlog::EventKind::InferenceMetered,
        None,
        None,
        data,
        metrics,
    );
}

async fn maybe_apply_assistant_proactive_memory(
    cfg: &AssistantConfig,
    runtime: &Runtime,
    req: &mut GenerateRequest,
    messages: &[Message],
) {
    let Some(memory) = &cfg.proactive_memory else {
        return;
    };
    let query = proactive_query_from_messages(messages);
    if query.trim().is_empty() {
        return;
    }
    let mut recent = messages
        .iter()
        .rev()
        .filter_map(message_memory_text)
        .take(6)
        .collect::<Vec<_>>();
    recent.reverse();
    let events = {
        let log = runtime.log.lock().await;
        log.events().to_vec()
    };
    let (maintenance, decision) = match memory.proactive_intervention(&query, recent, &events).await
    {
        Ok(out) => out,
        Err(e) => {
            tracing::debug!(error = %e, "assistant proactive memory pass failed");
            return;
        }
    };
    {
        let mut log = runtime.log.lock().await;
        log.append(
            car_eventlog::EventKind::ProactiveMemoryMaintained,
            None,
            None,
            proactive_maintenance_event_data(&maintenance),
        );
        log.append(
            car_eventlog::EventKind::ProactiveMemoryIntervention,
            None,
            None,
            proactive_intervention_event_data(&decision),
        );
    }
    if let car_memgine::ProactiveMemoryDecision::Inject { reminder, .. } = decision {
        append_context_block(req, "Proactive Memory", &reminder);
    }
}

/// Per-run state for tool-repair learning: which tool failures are still
/// unrepaired, and which learned leads have already been offered.
///
/// This lives in the loop rather than in [`ToolMemory`] because it is run
/// state, not learned state — and it cannot be reconstructed from the
/// transcript, which records that a call failed but not that a later call was
/// the *recovery* for it.
#[derive(Default)]
struct OpenFailures {
    /// The most recent unrepaired failure per tool: its signature, the turn it
    /// happened on, and the arguments that failed.
    ///
    /// One entry per tool, deliberately. A run that fails two ways on the same
    /// tool (`missing_target` at turn 1, `timeout` at turn 3) keeps only the
    /// later one, so the earlier kind is not learned from that run. The
    /// alternative — a queue, and matching a success against the right member —
    /// needs a similarity notion this has no way to compute; keeping the most
    /// recent biases toward the failure the model was actually working on when
    /// it succeeded, which is the one a recovery most likely addressed.
    by_tool: HashMap<String, OpenFailure>,
    /// Signature keys a learned lead has been offered for at least once this
    /// run. Consulted ONLY to gate penalization — a lead stays offered on every
    /// turn its failure is open, and that repetition is free because the request
    /// context is rebuilt from scratch each turn.
    offered: std::collections::HashSet<String>,
    /// Signature keys already penalized this run, so one stale lead costs one
    /// failure however many times the model retries behind it.
    penalized: std::collections::HashSet<String>,
}

struct OpenFailure {
    sig: FailureSignature,
    turn: u32,
    /// The arguments of the call that failed, so a "recovery" can be required to
    /// differ from it.
    params: Value,
}

impl OpenFailures {
    fn observe_failure(&mut self, tool: &str, sig: FailureSignature, turn: u32, params: &Value) {
        self.by_tool.insert(
            tool.to_string(),
            OpenFailure {
                sig,
                turn,
                params: params.clone(),
            },
        );
    }

    /// A success on `tool` closes its open failure and is credited as the repair
    /// — but only when it is plausibly a repair at all. Two conditions, and the
    /// second is the one that matters.
    ///
    /// **Inside the recovery window.** Outside it the failure is forgotten
    /// rather than credited: an unrelated call many turns later is not a fix.
    ///
    /// **Different arguments from the call that failed.** Without this, every
    /// routine success on a read-heavy tool harvested whatever failure happened
    /// to be open — `web_search` fails, three ordinary searches follow, and the
    /// last unrelated query becomes the durable "repair" for
    /// `web_search::not_found`. Identical arguments succeeding is a transient,
    /// not a repair, and storing it teaches a lead that fixes nothing. Requiring
    /// a difference is the cheapest available proxy for "the model changed
    /// something", and it is the honest floor: it does not prove the change is
    /// what fixed it, only that a change occurred.
    ///
    /// This matters more than it looks, because the failure side of the ledger
    /// is narrow. A lead is penalized only when it was offered and its signature
    /// then failed AGAIN in the same run, so a wrong lead that the model quietly
    /// works around accrues successes and never a failure. Degradation cannot be
    /// the whole answer to mis-crediting; not mis-crediting is.
    fn take_recovery(&mut self, tool: &str, turn: u32, params: &Value) -> Option<FailureSignature> {
        let open = self.by_tool.remove(tool)?;
        if turn.saturating_sub(open.turn) > RECOVERY_WINDOW_TURNS {
            return None;
        }
        (open.params != *params).then_some(open.sig)
    }

    /// Signatures still unrepaired, for the recall pass. Cloned rather than
    /// borrowed so the caller can record what it offered while iterating.
    fn pending(&self) -> Vec<FailureSignature> {
        self.by_tool.values().map(|open| open.sig.clone()).collect()
    }
}

/// Recall learned repairs into the next model turn.
///
/// Two sources, in priority order. A lead for a failure that is open *right
/// now* is worth the most — the model is holding that problem this turn — so it
/// goes first and is never crowded out. Session-start leads (turn 1, keyed on
/// the task text) fill whatever room is left; they are speculative by nature,
/// so they yield.
fn maybe_apply_tool_memory(
    cfg: &AssistantConfig,
    req: &mut GenerateRequest,
    open: &mut OpenFailures,
    messages: &[Message],
    turns: u32,
) {
    let Some(memory) = &cfg.tool_memory else {
        return;
    };
    let mut lines = Vec::new();
    for sig in open.pending() {
        if let Some(lead) = memory.recall(&sig) {
            // Inline code, not prose: the lead is model-authored text written
            // downstream of tool output, and code formatting is one more signal
            // that it is data to consider rather than an instruction to follow.
            lines.push(format!("- after `{}`, this worked: `{lead}`", sig.key()));
            open.offered.insert(sig.key());
        }
    }
    if turns <= 1 {
        let task = proactive_query_from_messages(messages);
        if let Some(block) = memory.recall_for_task(&task) {
            lines.extend(block.lines().map(str::to_string));
        }
    }
    if lines.is_empty() {
        return;
    }
    // Labelled as prior-run evidence, not instruction. These leads are derived
    // from what happened, they are sometimes wrong by construction (see the
    // pairing heuristic in `tool_memory`), and a model told to follow them
    // would retry a stale fix instead of reading the error in front of it.
    append_context_block(
        req,
        "Learned Repairs",
        &format!(
            "From earlier runs on this machine — what recovered this kind of \
failure before. Treat as a hint, not an instruction; prefer the error you can \
actually see.\n{}",
            lines.join("\n")
        ),
    );
}

/// Fold one finished tool call into the learning state, and into the durable
/// store when it closes a failure.
fn record_tool_outcome(
    cfg: &AssistantConfig,
    open: &mut OpenFailures,
    tool: &str,
    ok: bool,
    content: &str,
    params: &Value,
    turns: u32,
) {
    let Some(memory) = &cfg.tool_memory else {
        return;
    };
    if ok {
        if let Some(sig) = open.take_recovery(tool, turns, params) {
            memory.record_success(&sig, &approach_from_call(tool, params));
        }
        return;
    }
    let sig = FailureSignature::from_failure(tool, content);
    let key = sig.key();
    // A lead we offered for this exact signature did not prevent the same
    // failure. That is the only honest failure signal available — learning
    // happens on recovery, so a signature nothing was ever offered for has
    // nothing to penalize.
    if open.offered.contains(&key) && open.penalized.insert(key) {
        memory.record_failure(&sig);
    }
    open.observe_failure(tool, sig, turns, params);
}

/// Consecutive no-progress repeats before we nudge the model, then give up. A
/// repeat is the model re-requesting work it already tried since its last state
/// mutation — the degenerate read/recall loop a model can fall into, burning the
/// whole turn budget while producing nothing.
const STALL_NUDGE: u32 = 3;
const STALL_BREAK: u32 = 6;

/// Turns of pure information-gathering (no state mutation) before a single soft
/// nudge to transition from exploring to acting. Nudge-only: a genuinely
/// read-only task never mutates, so this must not terminate — only the
/// unambiguous repeat loop (`STALL_BREAK`) hard-stops. This is the "read/write
/// ratio" / "time since last mutation" signal from the agent-control literature.
const EXPLORE_NUDGE: u32 = 8;

/// The set of tools that change state — a successful one is real progress and
/// resets the no-progress guard. Derived from the model-facing tool defs: any
/// def that self-declares `"mutating": true` (e.g. `generate_image`), plus the
/// builtin file writers whose defs come from `agent_basics` without the
/// flag. Everything else (reads, searches, AND read-only shells like
/// `wc`/`node --check`) is non-progress, so probing between reads can't silently
/// reset the guard. `shell` is deliberately excluded: a shell that never
/// accompanies a file edit isn't moving the task forward, and one that does
/// mutate is paired with a write/edit that resets the guard anyway.
fn mutating_tool_names(tool_defs: &[Value]) -> std::collections::HashSet<String> {
    let mut set: std::collections::HashSet<String> = ["write_file", "edit_file"]
        .iter()
        .map(|s| s.to_string())
        .collect();
    for def in tool_defs {
        if def
            .get("mutating")
            .and_then(Value::as_bool)
            .unwrap_or(false)
        {
            if let Some(name) = def.get("name").and_then(Value::as_str) {
                set.insert(name.to_string());
            }
        }
    }
    set
}

/// A stable signature of a turn's tool calls (names + arguments; id-independent
/// and order-independent) so two turns that request the identical work compare
/// equal — the basis for detecting a no-progress repeat.
fn tool_calls_signature(calls: &[ToolCall]) -> String {
    let mut parts: Vec<String> = calls
        .iter()
        .map(|c| {
            format!(
                "{}({})",
                c.name,
                serde_json::to_string(&c.arguments).unwrap_or_default()
            )
        })
        .collect();
    parts.sort();
    parts.join("|")
}

/// What the loop should do after one turn's tool calls, per the no-progress
/// guard.
#[derive(Debug, PartialEq, Eq)]
enum GuardStep {
    /// A genuinely new state mutation — real progress; carry on fresh.
    Progress,
    /// Nothing notable; keep going.
    Continue,
    /// Spinning without progress — inject a nudge to act or finish this turn.
    Nudge,
    /// Repeated the same action too many times — stop the run as stalled.
    Break,
}

/// Tracks whether the agent loop is advancing or spinning in place.
///
/// A turn counts as progress ONLY when a mutating tool succeeds with a
/// signature not seen since the last progress. A repeated *identical* call is
/// idempotent — re-`remember`ing the same fact, re-writing identical bytes
/// changes nothing — so it is NOT progress even for a nominally `"mutating"`
/// tool. Counting such repeats as progress was the bug behind the observed
/// `remember()` loop that reset the guard every turn and ran to `max_turns`
/// instead of tripping `STALL_BREAK`. Non-mutating repeats are unchanged.
#[derive(Default)]
struct NoProgressGuard {
    seen_sigs: std::collections::HashSet<String>,
    stall_repeats: u32,
    turns_since_mutation: u32,
    nudged: bool,
}

impl NoProgressGuard {
    /// Feed one turn: `sig` is this turn's tool-call signature, `mutated_ok`
    /// whether a mutating tool succeeded this turn.
    fn observe(&mut self, sig: &str, mutated_ok: bool) -> GuardStep {
        let sig_is_new = self.seen_sigs.insert(sig.to_string());
        if mutated_ok && sig_is_new {
            // Real, new mutation: reset — but keep THIS signature so an immediate
            // identical repeat next turn still collides (and counts as a stall).
            self.seen_sigs.clear();
            self.seen_sigs.insert(sig.to_string());
            self.stall_repeats = 0;
            self.turns_since_mutation = 0;
            self.nudged = false;
            return GuardStep::Progress;
        }
        // Non-mutating, OR a repeated identical mutation → no real progress.
        self.turns_since_mutation += 1;
        if !sig_is_new {
            self.stall_repeats += 1;
            if self.stall_repeats >= STALL_BREAK {
                return GuardStep::Break;
            }
            if self.stall_repeats >= STALL_NUDGE && !self.nudged {
                self.nudged = true;
                return GuardStep::Nudge;
            }
        }
        if self.turns_since_mutation >= EXPLORE_NUDGE && !self.nudged {
            self.nudged = true;
            return GuardStep::Nudge;
        }
        GuardStep::Continue
    }
}

/// Build a single-action `tool_call` proposal, binding the action id to the
/// call id so the result correlates back.
/// Build the executable proposal for a tool call.
///
/// `parameters` is passed separately rather than read off `call.arguments`
/// because the loop may have substituted retained-value handles into it (#813).
/// Reading the raw arguments here would resolve `$r3` for the approval gate and
/// then execute the unresolved literal — the two must not disagree about what
/// is being run.
fn build_proposal(
    source: &str,
    call: &ToolCall,
    parameters: &Value,
) -> Result<ActionProposal, String> {
    serde_json::from_value(json!({
        "source": source,
        "actions": [{
            "id": call.id,
            "type": "tool_call",
            "tool": call.name,
            "parameters": parameters,
        }],
    }))
    .map_err(|e| format!("malformed proposal: {e}"))
}

/// Run the assistant loop to a terminal outcome, mutating `messages` (which must
/// already carry the system + first user turn) and streaming progress via
/// `emit`. Reusable across one-shot, REPL, and per-chat-turn.
pub async fn run_assistant_loop(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    emit: impl FnMut(AssistantEvent),
) -> AssistantOutcome {
    let never = std::sync::atomic::AtomicBool::new(false);
    run_assistant_loop_cancellable(generator, runtime, cfg, messages, &never, None, None, emit)
        .await
}

/// Same as [`run_assistant_loop`], but checks `cancel` before each turn so the
/// `agent.chat.cancel` path can interrupt a running turn between model calls,
/// and consults `approval` (if any) before running a `gated_tools` action.
pub async fn run_assistant_loop_cancellable(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    cancel: &std::sync::atomic::AtomicBool,
    approval: Option<&dyn ApprovalGate>,
    images: Option<&[ContentBlock]>,
    emit: impl FnMut(AssistantEvent),
) -> AssistantOutcome {
    run_assistant_loop_cancellable_in_session(
        generator, runtime, cfg, messages, cancel, approval, images, None, emit,
    )
    .await
}

/// Session-aware variant of [`run_assistant_loop_cancellable`]. A caller that
/// multiplexes conversations passes the Runtime session id so stateful tool
/// guards (notably read-before-edit) stay isolated between conversations.
pub async fn run_assistant_loop_cancellable_in_session(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    cancel: &std::sync::atomic::AtomicBool,
    approval: Option<&dyn ApprovalGate>,
    images: Option<&[ContentBlock]>,
    runtime_session_id: Option<&str>,
    emit: impl FnMut(AssistantEvent),
) -> AssistantOutcome {
    run_assistant_loop_cancellable_in_session_durable(
        generator,
        runtime,
        cfg,
        messages,
        cancel,
        approval,
        images,
        runtime_session_id,
        None,
        None,
        true,
        emit,
    )
    .await
}

/// Durable supervised-session variant. The checkpoint sink is invoked after
/// every mutation of the exact model-facing transcript, including compaction,
/// assistant tool calls, refusals, and tool results.
pub async fn run_assistant_loop_cancellable_in_session_durable(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    cancel: &std::sync::atomic::AtomicBool,
    approval: Option<&dyn ApprovalGate>,
    images: Option<&[ContentBlock]>,
    runtime_session_id: Option<&str>,
    durable_session_id: Option<&str>,
    durability: Option<&dyn super::governance::AssistantDurability>,
    redrive_ungrounded_summary: bool,
    mut emit: impl FnMut(AssistantEvent),
) -> AssistantOutcome {
    use std::sync::atomic::Ordering;
    let tools = if cfg.tools.is_empty() {
        None
    } else {
        Some(cfg.tools.clone())
    };
    let mut tools_called: Vec<String> = Vec::new();
    // Restored transcripts carry the authoritative tool-call/result pairs from
    // earlier process lifetimes. Seed grounding from them so a continuity turn
    // can cite prior evidence without rerunning completed actions.
    let mut tool_receipts: Vec<AssistantToolReceipt> = transcript_tool_receipts(messages);
    // Retained tool results for this run (#813). Constructed unconditionally
    // and left empty when the feature is off, so the off path allocates one
    // empty map and takes no other behavioral difference.
    let mut values = super::value_store::SessionValues::new();
    let mut last_text = String::new();
    let mut last_model = String::new();
    let mut models_served = Vec::new();
    let mut turns = 0u32;
    let mut claim_corrections = 0u8;
    // The model's window, resolved once (the model is fixed for the run). Used
    // to bound the growing message history each turn; 0 (unknown) disables it.
    // A caller's override tightens it; one above the known window is clamped
    // (see `resolve_context_window`).
    let registry_window = cfg
        .model
        .as_deref()
        .map(|m| generator.context_window(m))
        .unwrap_or(0);
    let (context_window, window_advisory) =
        resolve_context_window(cfg.context_window_override, registry_window);
    if let Some(advisory) = window_advisory {
        tracing::warn!(
            requested = cfg.context_window_override,
            registry_window,
            "{advisory}"
        );
        emit(AssistantEvent::Text(format!(
            "[context window: {advisory}]"
        )));
    }
    // What the per-message estimate cannot see, plus the provider's own count
    // of the last request once there is one (see `PromptMeasure`). The tool
    // definitions ride on every turn; their estimate is the same shared
    // chars/4 the inference layer's guard uses.
    let mut prompt_measure = PromptMeasure {
        fixed_overhead: car_inference::media_tokens::tool_defs_tokens(&cfg.tools),
        reported: None,
    };
    // Tools whose success counts as progress (resets the no-progress guard),
    // derived from the advertised defs — so a capability tool that self-declares
    // `"mutating": true` (e.g. generate_image) is recognized without editing the
    // loop.
    let mutating_tools = mutating_tool_names(&cfg.tools);
    // The advertised names, for the delegate child's execution-time subset
    // check, and whether this run offers `delegate` at all.
    let advertised_names: std::collections::HashSet<String> = cfg
        .tools
        .iter()
        .filter_map(|d| d.get("name").and_then(Value::as_str))
        .map(str::to_string)
        .collect();
    let delegate_advertised = advertised_names.contains(DELEGATE_TOOL);
    // Run-level delegation budget (see `DelegateBudget`).
    let delegate_budget = cfg.delegate_budget.unwrap_or_default();
    let mut delegations_spawned: u32 = 0;
    let mut child_turns_used: u32 = 0;
    // Tool-result provenance labels, resolved once for the run. The fallback is
    // the built-in table, not an empty one: an empty map classifies every result
    // as internal, which would leave the marking switched off by omission
    // (car#723).
    let builtin_labels;
    let tool_labels = match &cfg.tool_labels {
        Some(m) => m,
        None => {
            builtin_labels = builtin_tool_labels();
            &builtin_labels
        }
    };
    // No-progress guard: tracks signatures of work tried since the last real
    // state mutation (a signature reappearing is a stall) plus turns spent
    // without changing anything. A repeated identical call is never progress,
    // even to a mutating tool — see NoProgressGuard.
    let mut guard = NoProgressGuard::default();
    // Tool-repair learning state for this run (`tool_memory`). Inert unless a
    // surface opted in via `AssistantConfig::tool_memory`.
    let mut open_failures = OpenFailures::default();

    while turns < cfg.max_turns {
        if cancel.load(Ordering::Relaxed) {
            return AssistantOutcome {
                status: "cancelled",
                summary: "cancelled".to_string(),
                turns,
                tools_called,
                tool_receipts,
                models_served: models_served.clone(),
                model_used: last_model.clone(),
            };
        }
        turns += 1;

        // Keep the running conversation within the model's context window so a
        // long tool-heavy run never overflows it (which degrades the model and
        // can truncate the original task provider-side).
        let before_compaction = messages.clone();
        compact_history_measured(messages, context_window, prompt_measure);
        if before_compaction != *messages {
            // The reported count covered the pre-compaction history; the next
            // call reports afresh.
            prompt_measure.reported = None;
            if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                if let Err(e) = store
                    .checkpoint(session_id, messages, "history_compacted", None)
                    .await
                {
                    let msg = format!("durable checkpoint failed after compaction: {e}");
                    emit(AssistantEvent::Error(msg.clone()));
                    return AssistantOutcome {
                        status: "error",
                        summary: msg,
                        turns,
                        tools_called,
                        tool_receipts,
                        models_served: models_served.clone(),
                        model_used: last_model,
                    };
                }
            }
        }

        let mut req = GenerateRequest {
            prompt: String::new(),
            model: cfg.model.clone(),
            params: GenerateParams {
                temperature: 0.0,
                strict_model: cfg.strict_model,
                ..Default::default()
            },
            context: None,
            context_stable_prefix: None,
            tools: tools.clone(),
            // Attach any images to the first request (they belong to the user's
            // latest message); later turns are tool-result follow-ups.
            images: if turns == 1 {
                images.map(|imgs| imgs.to_vec())
            } else {
                None
            },
            messages: Some(messages.clone()),
            cache_control: false,
            // Only on tool-less turns: a JSON-constrained request suppresses
            // tool use on real providers, so a turn that offers tools stays
            // unconstrained and the final answer is checked (and repaired
            // once, without tools) below.
            response_format: if tools.is_none() {
                cfg.response_format.clone()
            } else {
                None
            },
            intent: None,
            client_ref: None,
            expected_row_digest: None,
            expected_catalog_revision: None,
            caller: None,
        };
        maybe_apply_assistant_proactive_memory(cfg, runtime, &mut req, messages).await;
        maybe_apply_tool_memory(cfg, &mut req, &mut open_failures, messages, turns);

        // Per-turn state block (#814 items 2-3). Rendered fresh every turn from
        // live state and appended to the request copy, so durable state reaches
        // the model whether or not it thinks to ask — the recall-discipline
        // dependency the issue is about. Skipped entirely when there is nothing
        // to say, so a run with no plan and no writes costs no tokens and no
        // cache churn.
        //
        // Carries the task list and the facts written this run. It deliberately
        // does NOT carry the working directory or the approval tier, which #814
        // also lists, because neither is live state: `root` is fixed at executor
        // construction and the standing tier at bind time, and both already reach
        // the model in the system prompt's `Environment:` sentence — which
        // compaction pins, so it can never be evicted. Repeating a constant in
        // the tail would re-send it every turn for no new information and create
        // a second copy free to drift from the substrate's own description.
        let todo_render = match &cfg.todos {
            Some(todos) => todos.lock().await.render(),
            None => None,
        };
        if let Some(block) = render_state_block(todo_render, &recent_fact_subjects(&tool_receipts))
        {
            if let Some(msgs) = req.messages.as_mut() {
                append_state_block(msgs, &block);
            }
        }
        // How many leading messages of the LIVE history this request carries
        // — the index the provider's reported prompt size will be attributed
        // to. Captured after the state block and memory context are attached:
        // both land on the request's own copy (`req.messages`), never on
        // `messages`, so their tokens are part of the reported count without
        // being messages the next turn can index. The ratio in
        // `compact_history_measured` absorbs them.
        let request_covers = messages.len();

        let mut result = match generator.generate(req).await {
            Ok(r) => r,
            Err(e) => {
                let msg = format!("inference failed: {e}");
                emit(AssistantEvent::Error(msg.clone()));
                return AssistantOutcome {
                    status: "error",
                    summary: msg,
                    turns,
                    tools_called,
                    tool_receipts,
                    models_served: models_served.clone(),
                    model_used: last_model.clone(),
                };
            }
        };
        // Meter the call that just returned. Placed immediately after a
        // successful `generate` and before every early return below, so each
        // COMPLETED model call is counted exactly once no matter which of those
        // paths the turn takes.
        //
        // A call that *failed* is NOT counted: the `Err` arm above returns from
        // the loop before reaching this line. So `model_calls` is a count of
        // completed calls, never of attempts — a harness that retries a failing
        // provider ten times and gives up records zero. Read it alongside the
        // run's terminal status; do not read it as "requests issued".
        record_inference_metered(runtime, &result).await;
        let attribution = AssistantModelAttribution {
            model_id: result.served_model_id().to_string(),
            local_last_resort: result.local_last_resort,
        };
        emit(AssistantEvent::ModelServed {
            model_id: attribution.model_id.clone(),
            local_last_resort: attribution.local_last_resort,
        });
        models_served.push(attribution);
        // Ground truth for the next turn's compaction decision. All three
        // input buckets: Anthropic reports the cached prefix separately from
        // `prompt_tokens`, and the window holds the sum.
        if let Some(u) = &result.usage {
            let input = u.prompt_tokens + u.cache_read_input_tokens + u.cache_creation_input_tokens;
            if input > 0 {
                prompt_measure.reported = Some((input as usize, request_covers));
            }
        }
        // Strip any leaked model reasoning-channel prefix (e.g. gemma-4's
        // `thought`) from the answer at this choke point — the streamed
        // per-turn generate can bypass the inference-layer tool-call parsers, so
        // clean it here so no chat bubble ever shows the model's reasoning label
        // as the answer (table stakes, matching Claude Code / ChatGPT).
        result.text = car_inference::tasks::generate::strip_leaked_reasoning(&result.text);
        last_model = result.served_model_id().to_string();

        // No tool calls → the model's text is the final answer.
        if result.tool_calls.is_empty() {
            last_text = result.text.clone();
            let ungrounded = ungrounded_summary_claims(&last_text, &tool_receipts);
            if redrive_ungrounded_summary && !ungrounded.is_empty() {
                result.append_assistant_history(messages, vec![]);
                if claim_corrections < 2 && turns < cfg.max_turns {
                    claim_corrections += 1;
                    messages.push(Message::User {
                        content: format!(
                            "Evidence check rejected the draft's unsupported operational claim(s): {}. \
                             Rewrite the answer using only claims supported by successful transcript \
                             tool receipts. Preserve useful source findings, explicitly mark missing \
                             live evidence, and do not rerun completed actions merely to support prose.",
                            ungrounded.join(", ")
                        ),
                    });
                    if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                        if let Err(e) = store
                            .checkpoint(session_id, messages, "ungrounded_summary_redrive", None)
                            .await
                        {
                            let msg =
                                format!("durable checkpoint failed before claim correction: {e}");
                            emit(AssistantEvent::Error(msg.clone()));
                            return AssistantOutcome {
                                status: "error",
                                summary: msg,
                                turns,
                                tools_called,
                                tool_receipts,
                                models_served: models_served.clone(),
                                model_used: last_model,
                            };
                        }
                    }
                    continue;
                }
                let summary = annotate_summary_with_claim_note(&last_text, &ungrounded);
                emit(AssistantEvent::Error(summary.clone()));
                return AssistantOutcome {
                    status: "error",
                    summary,
                    turns,
                    tools_called,
                    tool_receipts,
                    models_served: models_served.clone(),
                    model_used: last_model,
                };
            }
            // One-shot format repair. Tool turns are never JSON-constrained
            // (a constrained request suppresses tool use), so on a run with
            // tools this check is what makes the CONTRACT hold. If the final
            // text is not the requested shape, re-ask once with the draft in
            // the transcript, no tools, and the format set — a turn that
            // cannot be anything but the answer. A final text that already
            // parses costs no extra call. The repair is announced
            // as a Text event so a `--json` consumer sees it happened; it is
            // still one repair, never a loop (a second miss is reported, not
            // retried).
            // Whether the final assistant message is already in `messages`
            // (the repair path appends the draft before re-asking).
            let mut final_appended = false;
            if let Some(format) = cfg.response_format.as_ref().filter(|f| {
                !final_text_matches_format(&last_text, f, cfg.response_format_validator.as_ref())
            }) {
                emit(AssistantEvent::Text(FORMAT_REPAIR_NOTICE.to_string()));
                result.append_assistant_history(messages, vec![]);
                messages.push(Message::User {
                    content: format_repair_nudge(format).to_string(),
                });
                let repair = GenerateRequest {
                    prompt: String::new(),
                    model: cfg.model.clone(),
                    params: GenerateParams {
                        temperature: 0.0,
                        strict_model: cfg.strict_model,
                        ..Default::default()
                    },
                    context: None,
                    context_stable_prefix: None,
                    tools: None,
                    images: None,
                    messages: Some(messages.clone()),
                    cache_control: false,
                    response_format: Some(format.clone()),
                    intent: None,
                    client_ref: None,
                    expected_row_digest: None,
                    expected_catalog_revision: None,
                    caller: None,
                };
                match generator.generate(repair).await {
                    Ok(mut repaired) => {
                        record_inference_metered(runtime, &repaired).await;
                        let attribution = AssistantModelAttribution {
                            model_id: repaired.served_model_id().to_string(),
                            local_last_resort: repaired.local_last_resort,
                        };
                        emit(AssistantEvent::ModelServed {
                            model_id: attribution.model_id.clone(),
                            local_last_resort: attribution.local_last_resort,
                        });
                        models_served.push(attribution);
                        repaired.text =
                            car_inference::tasks::generate::strip_leaked_reasoning(&repaired.text);
                        if !final_text_matches_format(
                            &repaired.text,
                            format,
                            cfg.response_format_validator.as_ref(),
                        ) {
                            emit(AssistantEvent::Text(
                                FORMAT_REPAIR_STILL_INVALID.to_string(),
                            ));
                        }
                        last_model = repaired.served_model_id().to_string();
                        last_text = repaired.text.clone();
                        result = repaired;
                    }
                    Err(e) => {
                        // Keep the draft: it is the model's answer, and the
                        // provider's refusal to enforce a format does not
                        // unmake it. Drop the nudge so the durable transcript
                        // ends on the answer, not on a question nobody
                        // answered.
                        if matches!(messages.last(), Some(Message::User { content }) if content == format_repair_nudge(format))
                        {
                            messages.pop();
                        }
                        emit(AssistantEvent::Text(format!(
                            "{FORMAT_REPAIR_FAILED_PREFIX} {e}; returning the draft answer as-is]"
                        )));
                        final_appended = true;
                    }
                }
            }
            if !final_appended {
                result.append_assistant_history(messages, vec![]);
            }
            if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                if let Err(e) = store
                    .checkpoint(session_id, messages, "assistant_final", None)
                    .await
                {
                    let msg = format!("durable checkpoint failed after assistant response: {e}");
                    emit(AssistantEvent::Error(msg.clone()));
                    return AssistantOutcome {
                        status: "error",
                        summary: msg,
                        turns,
                        tools_called,
                        tool_receipts,
                        models_served: models_served.clone(),
                        model_used: last_model,
                    };
                }
            }
            // Record why the loop stopped. On this ungrounded default path an
            // empty-tool-calls turn is treated as success even when the model was
            // truncated mid-answer — capture the truncation signal so a false
            // completion is diagnosable (docs/audits/car-tracing-design-2026-07-07).
            runtime
                .record_turn_completed(
                    "empty_tool_calls",
                    result.stop_reason.as_deref(),
                    result.was_truncated(),
                    turns,
                    &last_model,
                )
                .await;
            emit(AssistantEvent::Done {
                text: last_text.clone(),
            });
            return AssistantOutcome {
                status: "success",
                summary: last_text,
                turns,
                tools_called,
                tool_receipts,
                models_served: models_served.clone(),
                model_used: last_model.clone(),
            };
        }

        if !result.text.trim().is_empty() {
            last_text = result.text.clone();
            emit(AssistantEvent::Text(result.text.clone()));
        }

        // Assign ids to any call missing one so results correlate back.
        let mut calls = result.tool_calls.clone();
        for (i, call) in calls.iter_mut().enumerate() {
            if call.id.is_none() {
                call.id = Some(format!("call_{turns}_{i}"));
            }
        }

        result.append_assistant_history(messages, calls.clone());
        if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
            if let Err(e) = store
                .checkpoint(session_id, messages, "assistant_tool_calls", None)
                .await
            {
                let msg = format!("durable checkpoint failed before tool dispatch: {e}");
                emit(AssistantEvent::Error(msg.clone()));
                return AssistantOutcome {
                    status: "error",
                    summary: msg,
                    turns,
                    tools_called,
                    tool_receipts,
                    models_served: models_served.clone(),
                    model_used: last_model,
                };
            }
        }

        // The no-progress guard runs AFTER execution (below), so it can key on
        // whether a mutation actually SUCCEEDED — a repeatedly-*failing* write
        // (bad path, denied) is not progress, and resetting on the mere request
        // would let "40 failed writes" evade the guard.
        let mut mutated_ok = false;

        // Execute each call in emitted order (avoid the DAG racing same-turn
        // filesystem effects like mkdir-then-write).
        for call in &calls {
            let id = call.id.clone().expect("ids assigned above");
            emit(AssistantEvent::ToolCall {
                name: call.name.clone(),
                params: serde_json::to_value(&call.arguments).unwrap_or_default(),
            });

            // Per-agent approval gate. The policy (when set) decides allow /
            // require-approval / deny per the running agent's posture; otherwise
            // the standing-tier `gated_tools` list applies (unchanged behavior).
            let mut params_val = serde_json::to_value(&call.arguments).unwrap_or_default();
            // Resolve `$rN` handles BEFORE anything inspects the parameters
            // (#813). Ordering is load bearing three times over: the approval
            // policy and the human gate must see the REAL arguments, or a
            // reference becomes a way to get an unreviewed value past review;
            // and `car-validator` checks against the tool's JSON Schema, where
            // `"$r3"` is a string in a slot that may demand an array — resolving
            // first keeps every schema unweakened instead of teaching all of
            // them to admit a reference form.
            if cfg.value_store_previews {
                let resolved = values.resolve_refs(&mut params_val);
                if !resolved.is_empty() {
                    tracing::debug!(
                        tool = %call.name,
                        handles = ?resolved,
                        "resolved retained-value references in tool arguments"
                    );
                }
            }
            let params_val = params_val;
            let posture = match &cfg.approval_policy {
                Some(policy) => policy(&call.name, &params_val),
                None => {
                    if cfg.gated_tools.iter().any(|t| t == &call.name) {
                        ToolApprovalDecision::RequireApproval
                    } else {
                        ToolApprovalDecision::Allow
                    }
                }
            };
            // A delegate child may only call what it was granted (see
            // `AssistantConfig::refuse_unadvertised_tools`). Checked before
            // the gate so an ungranted tool never reaches an approver either.
            let posture = if cfg.refuse_unadvertised_tools && !advertised_names.contains(&call.name)
            {
                ToolApprovalDecision::Deny(format!(
                    "tool '{}' is not granted to this delegate; use only: {}",
                    call.name,
                    advertised_names
                        .iter()
                        .cloned()
                        .collect::<Vec<_>>()
                        .join(", ")
                ))
            } else {
                posture
            };
            let needs_approval = matches!(&posture, ToolApprovalDecision::RequireApproval);

            let refusal: Option<String> = match posture {
                ToolApprovalDecision::Allow => None,
                ToolApprovalDecision::Deny(reason) => Some(reason),
                ToolApprovalDecision::RequireApproval => {
                    let decision = match approval {
                        Some(gate) => gate.request_action(&id, &call.name, &params_val).await,
                        None => ApprovalDecision::Denied(format!(
                            "'{}' needs approval: re-run with --full-access to allow it on this host, \
                             or use the default sandbox where edits are isolated",
                            call.name
                        )),
                    };
                    match decision {
                        ApprovalDecision::Approved => None,
                        ApprovalDecision::Denied(reason) => Some(reason),
                    }
                }
            };
            if let Some(reason) = refusal {
                let content = cap(json!({ "error": reason }).to_string());
                emit(AssistantEvent::ToolResult {
                    name: call.name.clone(),
                    ok: false,
                    content: content.clone(),
                });
                messages.push(Message::ToolResult {
                    tool_use_id: id,
                    content,
                    // A refusal the runtime itself wrote — never fetched.
                    provenance: Provenance::Internal,
                });
                if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                    if let Err(e) = store
                        .checkpoint(session_id, messages, "tool_refused", None)
                        .await
                    {
                        let msg = format!("durable checkpoint failed after refusal: {e}");
                        emit(AssistantEvent::Error(msg.clone()));
                        return AssistantOutcome {
                            status: "error",
                            summary: msg,
                            turns,
                            tools_called,
                            tool_receipts,
                            models_served: models_served.clone(),
                            model_used: last_model,
                        };
                    }
                }
                continue;
            }

            // Loop-intercepted sub-agent. Only when the parent advertised it
            // — a hallucinated `delegate` call on a run that never offered
            // one is an unknown tool like any other, not a free sub-agent.
            if delegate_advertised && call.name == DELEGATE_TOOL {
                let goal_brief = params_val
                    .get("goal")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .replace('\n', " ");
                let goal_brief: String = goal_brief.chars().take(80).collect();
                let over_budget = delegations_spawned >= delegate_budget.max_delegations
                    || child_turns_used >= delegate_budget.max_child_turns;
                let done = if over_budget {
                    DelegateOutcome {
                        ok: false,
                        content: cap(json!({
                            "error": format!(
                                "delegation budget exhausted ({delegations_spawned} delegations / \
                                 {child_turns_used} child turns used; limits {} / {}) — finish with \
                                 what you have",
                                delegate_budget.max_delegations, delegate_budget.max_child_turns
                            )
                        })
                        .to_string()),
                        turns: 0,
                        external: false,
                        receipts: Vec::new(),
                        spawned: false,
                    }
                } else {
                    run_delegate(
                        generator,
                        runtime,
                        cfg,
                        messages,
                        &params_val,
                        cancel,
                        approval,
                        runtime_session_id,
                        redrive_ungrounded_summary,
                        tool_labels,
                    )
                    .await
                };
                if done.spawned {
                    delegations_spawned += 1;
                    child_turns_used = child_turns_used.saturating_add(done.turns);
                }
                emit(AssistantEvent::Text(format!(
                    "[delegate: {goal_brief} — {} turns, {}]",
                    done.turns,
                    if done.ok { "ok" } else { "error" }
                )));
                if done.ok {
                    tools_called.push(call.name.clone());
                    if mutating_tools.contains(&call.name) {
                        mutated_ok = true;
                    }
                }
                tool_receipts.push(AssistantToolReceipt {
                    tool: call.name.clone(),
                    call_id: Some(id.clone()),
                    ok: done.ok,
                    params: params_val.clone(),
                    via: None,
                });
                // The child's own receipts, merged and tagged: the parent's
                // grounding check ("tests passed" needs a successful `shell`
                // receipt) and `receipts.by_tool` must see what the child
                // actually ran, or a parent that delegated the test run is
                // flagged ungrounded for reporting a result it has evidence for.
                let via = format!("{DELEGATE_TOOL}:{id}");
                tool_receipts.extend(done.receipts.into_iter().map(|mut r| {
                    r.via = Some(via.clone());
                    r
                }));
                emit(AssistantEvent::ToolResult {
                    name: call.name.clone(),
                    ok: done.ok,
                    content: done.content.clone(),
                });
                messages.push(Message::ToolResult {
                    tool_use_id: id,
                    content: done.content,
                    // The child's prose is internal unless one of ITS tool
                    // results crossed the boundary — then the parent inherits
                    // the marking rather than laundering it through a summary.
                    provenance: if done.external {
                        Provenance::External
                    } else {
                        Provenance::Internal
                    },
                });
                if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                    if let Err(e) = store
                        .checkpoint(session_id, messages, "tool_result", None)
                        .await
                    {
                        let msg = format!("durable checkpoint failed after delegate result: {e}");
                        emit(AssistantEvent::Error(msg.clone()));
                        return AssistantOutcome {
                            status: "error",
                            summary: msg,
                            turns,
                            tools_called,
                            tool_receipts,
                            models_served: models_served.clone(),
                            model_used: last_model,
                        };
                    }
                }
                continue;
            }

            let proposal = match build_proposal(&result.model_used, call, &params_val) {
                Ok(p) => p,
                Err(e) => {
                    // A malformed call shape shouldn't sink the run; feed the
                    // error back so the model can retry with a valid shape.
                    let content = cap(json!({ "error": e }).to_string());
                    emit(AssistantEvent::ToolResult {
                        name: call.name.clone(),
                        ok: false,
                        content: content.clone(),
                    });
                    messages.push(Message::ToolResult {
                        tool_use_id: id,
                        content,
                        // A shape error the runtime itself wrote.
                        provenance: Provenance::Internal,
                    });
                    if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                        if let Err(e) = store
                            .checkpoint(session_id, messages, "malformed_tool_call", None)
                            .await
                        {
                            let msg = format!("durable checkpoint failed after tool error: {e}");
                            emit(AssistantEvent::Error(msg.clone()));
                            return AssistantOutcome {
                                status: "error",
                                summary: msg,
                                turns,
                                tools_called,
                                tool_receipts,
                                models_served: models_served.clone(),
                                model_used: last_model,
                            };
                        }
                    }
                    continue;
                }
            };

            if needs_approval {
                let dispatch = match approval {
                    Some(gate) => gate.before_dispatch(&id, &call.name, &params_val).await,
                    None => Err("approval gate disappeared before dispatch".into()),
                };
                if let Err(e) = dispatch {
                    let content =
                        cap(json!({ "error": format!("dispatch refused: {e}") }).to_string());
                    emit(AssistantEvent::ToolResult {
                        name: call.name.clone(),
                        ok: false,
                        content: content.clone(),
                    });
                    messages.push(Message::ToolResult {
                        tool_use_id: id,
                        content,
                        provenance: Provenance::Internal,
                    });
                    if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                        let _ = store
                            .checkpoint(session_id, messages, "dispatch_refused", None)
                            .await;
                    }
                    continue;
                }
            }

            let exec = match runtime_session_id {
                Some(session_id) => runtime.execute_with_session(&proposal, session_id).await,
                None => runtime.execute(&proposal).await,
            };
            let action = exec.results.first();
            let runtime_succeeded = action
                .map(|r| matches!(r.status, ActionStatus::Succeeded))
                .unwrap_or(false);
            // Shell non-zero is intentionally returned as an observation so
            // the model can repair it, but it is not a successful receipt.
            // Treating Runtime::Succeeded as "tests passed" was a false-proof
            // bug because the executor successfully *ran* a command that
            // exited 1.
            let ok = runtime_succeeded
                && (call.name != "shell"
                    || action
                        .and_then(|result| result.output.as_ref())
                        .and_then(|output| output.get("exit_code"))
                        .and_then(Value::as_i64)
                        == Some(0));
            if needs_approval {
                let receipt = json!({
                    "ok": ok,
                    "action_id": action.map(|result| result.action_id.clone()),
                    "status": action.map(|result| format!("{:?}", result.status)),
                });
                if let Some(gate) = approval {
                    if let Err(e) = gate
                        .after_dispatch(&id, &call.name, &params_val, ok, &receipt)
                        .await
                    {
                        let msg = format!(
                            "action executed but its durable terminal receipt failed: {e}; action is indeterminate"
                        );
                        emit(AssistantEvent::Error(msg.clone()));
                        return AssistantOutcome {
                            status: "error",
                            summary: msg,
                            turns,
                            tools_called,
                            tool_receipts,
                            models_served: models_served.clone(),
                            model_used: last_model,
                        };
                    }
                }
            }
            // Observation shaping (#813). Only a result that WOULD have been
            // destructively truncated is replaced by a preview: one that
            // already fits is strictly more useful shown whole, and paying a
            // handle + indirection for it would trade information the model
            // had for free against nothing.
            //
            // Note this is the narrower of the two possible readings of #813.
            // Previewing EVERY result — the shape NVIDIA's numbers come from —
            // would also cut the per-turn re-serialization cost, but it removes
            // detail from results that fit today. That is exactly the kind of
            // trade the `car-bench` A/B exists to settle, so it is deliberately
            // not assumed here.
            let content = match action {
                Some(r)
                    if cfg.value_store_previews && matches!(r.status, ActionStatus::Succeeded) =>
                {
                    let rendered = format_tool_result(r);
                    match (&r.output, rendered.len() > OBSERVATION_CAP) {
                        (Some(v), true) => {
                            let handle = values.put(v.clone());
                            format!(
                                "{}{}",
                                super::value_store::render_preview(&handle, v),
                                super::value_store::reference_hint(&handle)
                            )
                        }
                        // Fits, or carries no structured output to retain —
                        // a handle to nothing helps nobody.
                        _ => cap(rendered),
                    }
                }
                Some(r) => cap(format_tool_result(r)),
                None => cap(format!("tool '{}' produced no result", call.name)),
            };
            if ok {
                tools_called.push(call.name.clone());
                if mutating_tools.contains(&call.name) {
                    mutated_ok = true;
                }
            }
            tool_receipts.push(AssistantToolReceipt {
                tool: call.name.clone(),
                call_id: action.map(|r| r.action_id.clone()),
                ok,
                params: params_val.clone(),
                via: None,
            });
            // Learn from what just happened: a failure opens a signature, a
            // success on the same tool inside the recovery window closes it and
            // captures the call that worked. `content` is the rendered
            // observation because it is where BOTH failure shapes are already
            // normalized — a runtime `[FAILED] …` and a `shell` that ran fine
            // but exited non-zero.
            record_tool_outcome(
                cfg,
                &mut open_failures,
                &call.name,
                ok,
                &content,
                &params_val,
                turns,
            );
            emit(AssistantEvent::ToolResult {
                name: call.name.clone(),
                ok,
                content: content.clone(),
            });
            messages.push(Message::ToolResult {
                tool_use_id: id,
                content,
                // The only site that can carry bytes from outside the trust
                // boundary. Classified from the tool's information-flow labels
                // rather than a name list local to this file — see
                // `tool_output_is_external`.
                provenance: if tool_output_is_external(&call.name, tool_labels) {
                    Provenance::External
                } else {
                    Provenance::Internal
                },
            });
            if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                if let Err(e) = store
                    .checkpoint(session_id, messages, "tool_result", None)
                    .await
                {
                    let msg = format!("durable checkpoint failed after tool result: {e}");
                    emit(AssistantEvent::Error(msg.clone()));
                    return AssistantOutcome {
                        status: "error",
                        summary: msg,
                        turns,
                        tools_called,
                        tool_receipts,
                        models_served: models_served.clone(),
                        model_used: last_model,
                    };
                }
            }
        }

        // No-progress guard, now that we know what actually SUCCEEDED. A real,
        // NEW state mutation resets everything (progress). A repeated signature —
        // even to a mutating tool — is a tight loop (nudge at STALL_NUDGE, stop at
        // STALL_BREAK); too many turns gathering info / failing to change anything
        // earns one soft nudge to act (EXPLORE_NUDGE) — no hard stop, since a
        // genuinely read-only task legitimately never mutates.
        let mut inject_nudge = false;
        match guard.observe(&tool_calls_signature(&calls), mutated_ok) {
            GuardStep::Break => {
                let summary = format!(
                    "Stopped: repeated the same action {} times without changing \
                     anything — no progress was being made.",
                    guard.stall_repeats
                );
                runtime
                    .record_turn_completed("stalled", None, false, turns, &last_model)
                    .await;
                emit(AssistantEvent::Done {
                    text: summary.clone(),
                });
                return AssistantOutcome {
                    status: "stalled",
                    summary,
                    turns,
                    tools_called,
                    tool_receipts,
                    models_served: models_served.clone(),
                    model_used: last_model.clone(),
                };
            }
            GuardStep::Nudge => inject_nudge = true,
            GuardStep::Progress | GuardStep::Continue => {}
        }

        // The model has been repeating itself: prod it to act or finish. Injected
        // after the tool results so it reads as guidance on the just-seen output.
        if inject_nudge {
            messages.push(Message::User {
                content: "You have repeated the same action several times without \
                          changing anything or making progress. Stop re-reading and \
                          either take a concrete action (write or edit a file, run a \
                          command) or, if the task is genuinely complete, finish now \
                          with your summary."
                    .into(),
            });
            if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
                if let Err(e) = store
                    .checkpoint(session_id, messages, "progress_nudge", None)
                    .await
                {
                    let msg = format!("durable checkpoint failed after progress nudge: {e}");
                    emit(AssistantEvent::Error(msg.clone()));
                    return AssistantOutcome {
                        status: "error",
                        summary: msg,
                        turns,
                        tools_called,
                        tool_receipts,
                        models_served: models_served.clone(),
                        model_used: last_model,
                    };
                }
            }
        }
    }

    runtime
        .record_turn_completed("max_turns", None, false, turns, &last_model)
        .await;
    AssistantOutcome {
        status: "max_turns",
        summary: if last_text.is_empty() {
            format!("stopped after {} turns without finishing", cfg.max_turns)
        } else {
            last_text
        },
        turns,
        tools_called,
        tool_receipts,
        models_served: models_served.clone(),
        model_used: last_model.clone(),
    }
}

#[derive(Debug, Clone)]
struct SummaryClaimRequirement {
    label: &'static str,
    tools: &'static [&'static str],
    require_ok: bool,
    shell_terms: &'static [&'static str],
    paths: Vec<String>,
}

const TEST_TERMS: &[&str] = &[
    "test",
    "pytest",
    "cargo test",
    "cargo nextest",
    "npm test",
    "npm run test",
    "pnpm test",
    "pnpm run test",
    "yarn test",
    "bun test",
    "go test",
    "swift test",
    "dotnet test",
    "ctest",
    "cmake --build",
    "make test",
];
const BUILD_TERMS: &[&str] = &[
    "build",
    "cargo check",
    "cargo build",
    "npm run build",
    "pnpm build",
    "yarn build",
    "bun run build",
    "cmake --build",
    "go build",
    "swift build",
    "dotnet build",
    "mvn package",
    "gradle build",
    "./gradlew build",
];
const CHECK_TERMS: &[&str] = &[
    "cargo check",
    "git diff --check",
    "npm run lint",
    "npm run check",
    "pnpm check",
    "pnpm lint",
    "yarn check",
    "yarn lint",
    "bun run check",
    "eslint",
    "clippy",
    "swiftlint",
    "ruff",
    "mypy",
    "biome check",
];
// Shell-command substrings that evidence a read / write, used to decide whether a
// final-summary claim ("I read the files", "I created the file") is backed by a
// real tool receipt. The assistant's shell is `cmd /C` on Windows, so these must
// carry the cmd spellings too — otherwise a Windows run that genuinely wrote a
// file produces no recognized receipt and the claim check false-negatives against
// the model (annotating a truthful summary as unverified, or failing a
// judge-dependent verdict closed).
// Matching is a plain `cmd.contains(term)` (see `receipt_supports_claim`), so a
// term must not be a substring of an unrelated command: `"dir "` is deliberately
// absent because it also matches `mkdir `, which would let a *write* stand in as
// a *read* receipt on every platform.
const READ_TERMS: &[&str] = &[
    "cat ", "sed ", "rg ", "grep ", "ls ", "find ", // POSIX
    "type ", "findstr ", // cmd
];
const WRITE_TERMS: &[&str] = &[
    "touch ",
    "cat >",
    "tee ",
    "python ",
    "node ",
    "perl ", // POSIX
    "type nul >",
    "echo >", // cmd
];
const GIT_STATUS_TERMS: &[&str] = &["git status"];
const GIT_REVISION_TERMS: &[&str] = &["git rev-parse", "git log", "git show"];
const APP_INSIGHTS_TERMS: &[&str] = &["az monitor app-insights query"];
const DEPLOYMENT_EVIDENCE_TERMS: &[&str] = &[
    "az pipelines show",
    "az pipelines runs show",
    "az devops invoke",
];
const SUMMARY_PATH_EXTENSIONS: &[&str] = &[
    ".rs", ".py", ".js", ".ts", ".tsx", ".jsx", ".go", ".swift", ".java", ".kt", ".kts", ".c",
    ".h", ".cc", ".hh", ".cpp", ".hpp", ".cxx", ".hxx", ".cs", ".fs", ".vb", ".php", ".rb", ".ex",
    ".exs", ".md", ".txt", ".json", ".yaml", ".yml", ".toml", ".html", ".css", ".xml", ".sh",
    ".sql",
];

fn normalize_summary_path_token(raw: &str) -> Option<String> {
    let token = raw.trim_matches(|c: char| {
        matches!(
            c,
            '"' | '\'' | '`' | ',' | ';' | ':' | ')' | '(' | '[' | ']' | '{' | '}' | '.'
        )
    });
    if token.is_empty() || token.starts_with('-') || token.contains("://") || token.contains("..") {
        return None;
    }
    let looks_like_path = token.contains('/')
        || SUMMARY_PATH_EXTENSIONS
            .iter()
            .any(|ext| token.to_ascii_lowercase().ends_with(ext));
    if !looks_like_path {
        return None;
    }
    Some(
        token
            .trim_start_matches("./")
            .replace('\\', "/")
            .to_ascii_lowercase(),
    )
}

fn summary_path_hints(summary: &str) -> Vec<String> {
    let mut paths = Vec::new();
    for raw in summary.split_whitespace() {
        if let Some(path) = normalize_summary_path_token(raw) {
            if !paths.contains(&path) {
                paths.push(path);
            }
        }
    }
    paths
}

fn summary_claim_requirements(summary: &str) -> Vec<SummaryClaimRequirement> {
    let s = summary.to_ascii_lowercase();
    let units: Vec<&str> = s
        .split(['\n', '.'])
        .map(str::trim)
        .filter(|unit| !unit.is_empty())
        .collect();
    let path_hints = summary_path_hints(summary);
    let mut claims = Vec::new();
    if units.iter().any(|unit| {
        unit.contains("ran the test")
            || unit.contains("ran tests")
            || (unit.contains("verified with") && unit.contains("test"))
            || (unit.contains("test")
                && (unit.contains("passed")
                    || unit.contains("green")
                    || unit.contains("succeeded")
                    || unit.contains("successful")))
    }) {
        claims.push(SummaryClaimRequirement {
            label: "tests were run/passed",
            tools: &["shell"],
            require_ok: true,
            shell_terms: TEST_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        (unit.contains("build") || unit.contains("cargo check"))
            && (unit.contains("passed")
                || unit.contains("succeeded")
                || unit.contains("successful")
                || unit.contains("built")
                || unit.contains("green")
                || unit.contains("ran the build")
                || unit.contains("ran cargo check"))
    }) {
        claims.push(SummaryClaimRequirement {
            label: "build succeeded",
            tools: &["shell"],
            require_ok: true,
            shell_terms: BUILD_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        (unit.contains("check") || unit.contains("lint"))
            && (unit.contains("passed")
                || unit.contains("green")
                || unit.contains("succeeded")
                || unit.contains("successful"))
    }) {
        claims.push(SummaryClaimRequirement {
            label: "checks were run/passed",
            tools: &["shell"],
            require_ok: true,
            shell_terms: CHECK_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        (unit.contains("read ") || unit.contains("inspected ") || unit.contains("looked at "))
            && (unit.contains("file") || unit.contains("source"))
    }) {
        claims.push(SummaryClaimRequirement {
            label: "files were read/inspected",
            tools: &["read_file", "list_dir", "find_files", "grep_files", "shell"],
            require_ok: true,
            shell_terms: READ_TERMS,
            paths: path_hints.clone(),
        });
    }
    if units.iter().any(|unit| {
        (unit.contains("created")
            || unit.contains("wrote")
            || unit.contains("updated")
            || unit.contains("edited"))
            && unit.contains("file")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "files were created/updated",
            tools: &["write_file", "edit_file", "shell"],
            require_ok: true,
            shell_terms: WRITE_TERMS,
            paths: path_hints.clone(),
        });
    }
    if units.iter().any(|unit| {
        unit.contains("repository is clean")
            || unit.contains("repo is clean")
            || unit.contains("working tree is clean")
            || unit.contains("status: clean")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "repository cleanliness was verified",
            tools: &["shell"],
            require_ok: true,
            shell_terms: GIT_STATUS_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        unit.contains("head matches origin")
            || unit.contains("head is aligned with origin")
            || unit.contains("head and origin are identical")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "repository revision/remote relationship was verified",
            tools: &["shell"],
            require_ok: true,
            shell_terms: GIT_REVISION_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        (unit.contains("app insights")
            || unit.contains("application insights")
            || unit.contains("telemetry"))
            && (unit.contains("query showed")
                || unit.contains("query confirmed")
                || unit.contains("we observed")
                || unit.contains("live telemetry showed")
                || unit.contains("no recurrence")
                || unit.contains("recurred after"))
            && !unit.contains("not obtained")
            && !unit.contains("unable")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "live Application Insights evidence was observed",
            tools: &["shell", "browse_observe"],
            require_ok: true,
            shell_terms: APP_INSIGHTS_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        (unit.contains("production") || unit.contains("live"))
            && (unit.contains("browser") || unit.contains("portal") || unit.contains("page"))
            && (unit.contains("inspected")
                || unit.contains("observed")
                || unit.contains("verified"))
            && !unit.contains("not obtained")
            && !unit.contains("unable")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "production browser state was observed",
            tools: &["browse_observe"],
            require_ok: true,
            shell_terms: &[],
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        unit.contains("deployment")
            && (unit.contains("successfully fixed")
                || unit.contains("was deployed")
                || unit.contains("after fix")
                || unit.contains("post-deployment"))
            && !unit.contains("cannot")
            && !unit.contains("not obtained")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "deployment state/change was verified",
            tools: &["shell"],
            require_ok: true,
            shell_terms: DEPLOYMENT_EVIDENCE_TERMS,
            paths: Vec::new(),
        });
    }
    if units.iter().any(|unit| {
        unit.contains("subscription")
            && (unit.contains("outside") || unit.contains("not in"))
            && unit
                .as_bytes()
                .windows(2)
                .any(|window| window[0] == b'n' && window[1].is_ascii_digit())
            && !unit.contains("cannot")
            && !unit.contains("not verified")
            && !unit.contains("not obtained")
            && !unit.contains("insufficient evidence")
    }) {
        claims.push(SummaryClaimRequirement {
            label: "named aircraft subscription status was observed live",
            tools: &["shell"],
            require_ok: true,
            shell_terms: APP_INSIGHTS_TERMS,
            paths: Vec::new(),
        });
    }
    claims
}

fn shell_command(params: &Value) -> Option<String> {
    params
        .get("command")
        .and_then(Value::as_str)
        .map(|s| s.to_ascii_lowercase())
}

fn normalized_receipt_path(params: &Value) -> Option<String> {
    params.get("path").and_then(Value::as_str).map(|path| {
        path.trim_start_matches("./")
            .replace('\\', "/")
            .to_ascii_lowercase()
    })
}

fn text_mentions_summary_path(text: &str, path: &str) -> bool {
    let text = text.replace('\\', "/").to_ascii_lowercase();
    text.contains(path) || text.contains(&format!("./{path}"))
}

fn receipt_mentions_summary_path(receipt: &AssistantToolReceipt, path: &str) -> bool {
    if receipt.tool == "shell" {
        return shell_command(&receipt.params)
            .map(|cmd| text_mentions_summary_path(&cmd, path))
            .unwrap_or(false);
    }
    normalized_receipt_path(&receipt.params)
        .map(|receipt_path| text_mentions_summary_path(&receipt_path, path))
        .unwrap_or(false)
}

fn receipt_satisfies_claim(
    receipt: &AssistantToolReceipt,
    claim: &SummaryClaimRequirement,
) -> bool {
    if claim.require_ok && !receipt.ok {
        return false;
    }
    if !claim.tools.iter().any(|t| *t == receipt.tool) {
        return false;
    }
    if !claim.paths.is_empty()
        && !claim
            .paths
            .iter()
            .any(|path| receipt_mentions_summary_path(receipt, path))
    {
        return false;
    }
    if receipt.tool != "shell" || claim.shell_terms.is_empty() {
        return true;
    }
    let Some(cmd) = shell_command(&receipt.params) else {
        return false;
    };
    claim.shell_terms.iter().any(|term| cmd.contains(term))
}

/// Operational claims the final prose makes that no same-run tool receipt
/// supports — "I ran the tests" with no matching shell call, "I created X"
/// with no matching write.
///
/// Public because a machine-readable caller (`car do --json`, and any host
/// embedding the assistant) needs this as a **field**, not as prose appended
/// to the summary. Folded into the text it is a note a relaying model can
/// silently drop; as data it is a caution the caller must decide what to do
/// with. This is the mechanical half of "receipts decide completion".
///
/// Detection is lexical and deliberately conservative: it flags claims whose
/// wording names an operation, and stays silent otherwise. An empty result
/// means "nothing detected", NOT "the summary is verified".
pub fn ungrounded_summary_claims(
    summary: &str,
    receipts: &[AssistantToolReceipt],
) -> Vec<&'static str> {
    summary_claim_requirements(summary)
        .into_iter()
        .filter(|claim| {
            !receipts
                .iter()
                .any(|receipt| receipt_satisfies_claim(receipt, claim))
        })
        .map(|claim| claim.label)
        .collect()
}

fn apply_summary_claim_grounding(
    mut verdict: car_verify::goal::GoalVerdict,
    outcome: &AssistantOutcome,
) -> car_verify::goal::GoalVerdict {
    if !verdict.met {
        return verdict;
    }
    let ungrounded = ungrounded_summary_claims(&outcome.summary, &outcome.tool_receipts);
    if ungrounded.is_empty() {
        return verdict;
    }
    verdict.grounded = false;
    verdict.reason = format!(
        "{}; ungrounded assistant summary claim(s): {}",
        verdict.reason,
        ungrounded.join(", ")
    );
    verdict
}

/// Append a **non-authoritative** claim-check note to a goal-loop reply.
///
/// Used only on the deterministic-pass path (F9): when the goal check already
/// passed against ground truth but the final prose named an operational claim
/// ("tests passed", "created file X") with no matching same-run tool receipt,
/// the completion still stands — this only flags the unverified wording to the
/// reader. Written ONLY to the returned `GoalLoopResult.outcome.summary`; the
/// caller must never fold it into the `messages` Vec, which chat.rs persists
/// into the session thread (the note would then leak into later turns' context).
pub fn annotate_summary_with_claim_note(summary: &str, ungrounded: &[&'static str]) -> String {
    if ungrounded.is_empty() {
        return summary.to_string();
    }
    format!(
        "{summary}\n\n[claim check] unverified summary claim(s) this run \
         (no matching tool receipt): {}",
        ungrounded.join(", ")
    )
}

/// The result of a goal-driven assistant run: the last iteration's
/// [`AssistantOutcome`] plus the `GoalRun` audit (per-iteration verdicts,
/// grounded flag, halt reason).
pub struct GoalLoopResult {
    pub outcome: AssistantOutcome,
    pub run: car_verify::goal::GoalRun,
}

/// Upper bound on one `gather` pass — the "is this iteration's condition met?"
/// evaluation run after every iteration (car#1112). `gather` is caller-supplied
/// and may itself await a human approval, a subprocess, or (for a future
/// [`car_verify::goal::GoalCondition::ModelJudge`]) an inference call; none of
/// those callees is guaranteed to resolve on their own. Reuses the crate's
/// existing single-check ceiling (`car do`'s shell tool) rather than inventing
/// a second magic number for what is, structurally, the same kind of wait.
const GOAL_EVALUATION_TIMEOUT: Duration =
    Duration::from_secs(crate::coder::shell_tool::DEFAULT_SHELL_TIMEOUT_SECS);

/// Drive the assistant as a **goal loop**: keep running iterations (each a full
/// [`run_assistant_loop_cancellable`] pass — the model works until it stops
/// emitting tool calls) until a **deterministic** [`car_verify::goal::GoalCondition`]
/// holds or a [`car_verify::goal::GoalGovernor`] bound is hit. This is CAR's
/// answer to `/goal`: the "am I done?" decision is made by
/// [`car_verify::goal::evaluate_goal`] over ground truth gathered from the
/// runtime (`gather`), never by a model reading its own transcript.
///
/// `messages` should carry only the system turn; the loop drives every user turn
/// from the pinned goal (the drift anchor, re-derived each iteration via
/// [`car_verify::goal::anchor_directive`], so no turn can repoint the objective).
/// `gather(&outcome)` projects a `GoalGather` after each iteration — the caller
/// owns which command/model checks to run; the runtime folds in receipts/state.
pub async fn run_assistant_goal_loop<G, GF>(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    cancel: &std::sync::atomic::AtomicBool,
    approval: Option<&dyn ApprovalGate>,
    spec: &car_verify::goal::GoalSpec,
    gather: G,
    emit: impl FnMut(AssistantEvent),
) -> GoalLoopResult
where
    G: FnMut(&AssistantOutcome) -> GF,
    GF: std::future::Future<Output = car_engine::GoalGather>,
{
    run_assistant_goal_loop_in_session(
        generator, runtime, cfg, messages, cancel, approval, spec, None, gather, emit,
    )
    .await
}

/// Session-aware variant of [`run_assistant_goal_loop`].
pub async fn run_assistant_goal_loop_in_session<G, GF>(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    cancel: &std::sync::atomic::AtomicBool,
    approval: Option<&dyn ApprovalGate>,
    spec: &car_verify::goal::GoalSpec,
    runtime_session_id: Option<&str>,
    gather: G,
    emit: impl FnMut(AssistantEvent),
) -> GoalLoopResult
where
    G: FnMut(&AssistantOutcome) -> GF,
    GF: std::future::Future<Output = car_engine::GoalGather>,
{
    run_assistant_goal_loop_in_session_durable(
        generator,
        runtime,
        cfg,
        messages,
        cancel,
        approval,
        spec,
        runtime_session_id,
        None,
        None,
        gather,
        emit,
    )
    .await
}

pub async fn run_assistant_goal_loop_in_session_durable<G, GF>(
    generator: &dyn TurnGenerator,
    runtime: &Runtime,
    cfg: &AssistantConfig,
    messages: &mut Vec<Message>,
    cancel: &std::sync::atomic::AtomicBool,
    approval: Option<&dyn ApprovalGate>,
    spec: &car_verify::goal::GoalSpec,
    runtime_session_id: Option<&str>,
    durable_session_id: Option<&str>,
    durability: Option<&dyn super::governance::AssistantDurability>,
    mut gather: G,
    mut emit: impl FnMut(AssistantEvent),
) -> GoalLoopResult
where
    G: FnMut(&AssistantOutcome) -> GF,
    GF: std::future::Future<Output = car_engine::GoalGather>,
{
    use car_verify::goal::{
        anchor_directive, evaluate_goal, governor_check, GoalHalt, GoalRun, GoalRunState,
        GoalStatus, GoalVerdict,
    };
    use std::sync::atomic::Ordering;

    let start = std::time::Instant::now();
    let mut run_state = GoalRunState::default();
    let mut evidence: Vec<GoalVerdict> = Vec::new();
    let mut all_models_served = Vec::new();
    let mut last_reason = String::new();
    let mut last_outcome = AssistantOutcome {
        status: "goal_pending",
        summary: String::new(),
        turns: 0,
        tools_called: Vec::new(),
        tool_receipts: Vec::new(),
        models_served: Vec::new(),
        model_used: String::new(),
    };

    let finish = |status: GoalStatus,
                  grounded: bool,
                  reason: String,
                  iterations: u32,
                  evidence: Vec<GoalVerdict>,
                  outcome: AssistantOutcome|
     -> GoalLoopResult {
        GoalLoopResult {
            run: GoalRun {
                status,
                iterations,
                grounded,
                cost_usd: 0.0,
                last_reason: reason,
                evidence,
            },
            outcome,
        }
    };

    loop {
        run_state.elapsed_secs = start.elapsed().as_secs();
        if cancel.load(Ordering::Relaxed) {
            return finish(
                GoalStatus::Halted {
                    halt: GoalHalt::Cancelled,
                },
                evidence.last().map(|v| v.grounded).unwrap_or(true),
                "cancelled".into(),
                run_state.turns,
                evidence,
                last_outcome,
            );
        }
        if let Some(halt) = governor_check(&spec.governor, &run_state) {
            return finish(
                GoalStatus::Halted { halt },
                evidence.last().map(|v| v.grounded).unwrap_or(true),
                if last_reason.is_empty() {
                    halt.as_str().to_string()
                } else {
                    format!("{} ({})", halt.as_str(), last_reason)
                },
                run_state.turns,
                evidence,
                last_outcome,
            );
        }

        // Anchor the directive from the pinned goal and push it as the next
        // user turn (the loop owns all user turns).
        let directive = anchor_directive(&spec.goal, &last_reason);
        messages.push(Message::User { content: directive });
        if let (Some(store), Some(session_id)) = (durability, durable_session_id) {
            if let Err(e) = store
                .checkpoint(
                    session_id,
                    messages,
                    "goal_directive",
                    serde_json::to_value(spec).ok(),
                )
                .await
            {
                last_outcome.status = "error";
                last_outcome.summary = format!("durable goal checkpoint failed: {e}");
                return finish(
                    GoalStatus::Halted {
                        halt: GoalHalt::Cancelled,
                    },
                    false,
                    last_outcome.summary.clone(),
                    run_state.turns,
                    evidence,
                    last_outcome,
                );
            }
        }

        let mut outcome = run_assistant_loop_cancellable_in_session_durable(
            generator,
            runtime,
            cfg,
            messages,
            cancel,
            approval,
            None,
            runtime_session_id,
            durable_session_id,
            durability,
            false,
            &mut emit,
        )
        .await;
        all_models_served.append(&mut outcome.models_served);
        outcome.models_served = all_models_served.clone();
        run_state.turns += 1;
        // Progress = the iteration actually executed a tool. A prose-only turn
        // that didn't move the world is thrash (drives the no-progress guard).
        if outcome.tools_called.is_empty() {
            run_state.turns_since_progress += 1;
        } else {
            run_state.turns_since_progress = 0;
        }

        if outcome.status == "cancelled" {
            return finish(
                GoalStatus::Halted {
                    halt: GoalHalt::Cancelled,
                },
                evidence.last().map(|v| v.grounded).unwrap_or(true),
                "cancelled".into(),
                run_state.turns,
                evidence,
                outcome,
            );
        }

        // Gather ground truth and evaluate the deterministic condition. Bounded
        // (car#1112): this iteration already produced a real reply in `outcome`
        // — that reply is owed to the caller regardless of whether the
        // condition can be graded, so a `gather` that never resolves must halt
        // immediately (fail open) rather than hang the turn or burn the rest
        // of the iteration budget re-running the model against a condition
        // that structurally can never be evaluated.
        let g = match tokio::time::timeout(GOAL_EVALUATION_TIMEOUT, gather(&outcome)).await {
            Ok(g) => g,
            Err(_) => {
                let reason = format!(
                    "goal check did not complete within {}s — treating this turn's reply as \
                     unevaluated rather than blocking on it",
                    GOAL_EVALUATION_TIMEOUT.as_secs()
                );
                // `grounded: false` (car#1113 review), not `true`: `grounded`
                // means the verdict rests on deterministic ground truth, but
                // there IS no verdict here — the check never ran. `grounded:
                // true` next to `met: false` reads on the wire as
                // "deterministically evaluated, definitively not met", which
                // is the opposite of what this halt means. Confirmed harmless
                // to flip: `met: false` alone already keeps
                // `ChatGoalState.status` at `"running"` in
                // `handler::update_chat_goal_from_event`'s `goal_evaluated`
                // arm (it only reads `grounded` when `met == true`), and this
                // arm returns immediately after, so no loop logic downstream
                // consumes the flipped value either.
                let verdict = GoalVerdict {
                    met: false,
                    grounded: false,
                    reason: reason.clone(),
                };
                evidence.push(verdict.clone());
                runtime
                    .record_goal_evaluated(
                        &spec.goal,
                        &spec.condition,
                        run_state.turns,
                        verdict.met,
                        verdict.grounded,
                        &verdict.reason,
                        &outcome.model_used,
                    )
                    .await;
                tracing::warn!(
                    target: "car::goal",
                    iteration = run_state.turns,
                    timeout_secs = GOAL_EVALUATION_TIMEOUT.as_secs(),
                    "goal evaluation timed out — halting with the primary reply intact"
                );
                emit(AssistantEvent::GoalEvaluated {
                    iteration: run_state.turns,
                    met: false,
                    grounded: false,
                    reason: reason.clone(),
                });
                return finish(
                    GoalStatus::Halted {
                        halt: GoalHalt::EvaluationTimeout,
                    },
                    false,
                    reason,
                    run_state.turns,
                    evidence,
                    outcome,
                );
            }
        };
        let inputs = runtime.gather_goal_inputs(&g).await;

        // Pre-grounding verdict. `base.met && base.grounded` is exactly "the
        // deterministic check passed": a met verdict is grounded iff it rested
        // only on deterministic leaves (Command / StatePredicate / receipts / …);
        // a met verdict that leaned on a `ModelJudge` is grounded=false.
        let base = evaluate_goal(&spec.condition, &inputs);
        let verdict = if base.met && base.grounded {
            // Deterministic pass: ground truth already verified completion, so
            // final-summary claim grounding is DEMOTED from an authority (it used
            // to flip grounded=false and re-drive the loop on word choice — F9) to
            // a reply annotation. Keep grounded=true; if the prose named an
            // operational claim with no matching same-run receipt, log it and note
            // it on the returned reply text ONLY — never touch `messages`
            // (persisted into the session thread) or the verdict's grounded flag /
            // durable event. A deterministically-verified completion is not the
            // false-completion pattern the Phase-0 miners look for, so the prose
            // mismatch is logged at info, not recorded as a failure signal.
            let ungrounded = ungrounded_summary_claims(&outcome.summary, &outcome.tool_receipts);
            if !ungrounded.is_empty() {
                tracing::info!(
                    target: "car::goal",
                    iteration = run_state.turns,
                    claims = %ungrounded.join(", "),
                    "deterministic goal check passed; final-summary claim(s) unmatched \
                     to a tool receipt — annotating reply, keeping grounded=true"
                );
                outcome.summary = annotate_summary_with_claim_note(&outcome.summary, &ungrounded);
            }
            base
        } else {
            // Not a deterministic pass (unmet, or met only via a `ModelJudge`):
            // UNCHANGED behavior — claim grounding retains its authority to flip
            // grounded=false and concat the reason, the loop keeps iterating, and
            // the Phase-0 miners keep receiving the ungrounded `GoalEvaluated`
            // signal (harness_adapt ungrounded-completion, evolution failure fold).
            apply_summary_claim_grounding(base, &outcome)
        };
        evidence.push(verdict.clone());
        runtime
            .record_goal_evaluated(
                &spec.goal,
                &spec.condition,
                run_state.turns,
                verdict.met,
                verdict.grounded,
                &verdict.reason,
                &outcome.model_used,
            )
            .await;
        // Audit and stream the "why continue?" decision. The typed event-log
        // entry above is durable; this tracing/UI event is for live operators.
        tracing::info!(
            target: "car::goal",
            iteration = run_state.turns,
            met = verdict.met,
            grounded = verdict.grounded,
            reason = %verdict.reason,
            "goal evaluated"
        );
        emit(AssistantEvent::GoalEvaluated {
            iteration: run_state.turns,
            met: verdict.met,
            grounded: verdict.grounded,
            reason: verdict.reason.clone(),
        });

        if verdict.met && verdict.grounded {
            return finish(
                GoalStatus::Achieved,
                verdict.grounded,
                verdict.reason,
                run_state.turns,
                evidence,
                outcome,
            );
        }
        last_reason = verdict.reason;
        last_outcome = outcome;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::assistant::executor::GeneralExecutor;
    use async_trait::async_trait;
    use car_engine::{LocalSubstrate, Runtime, Substrate, ToolExecutor};
    use car_inference::{InferenceEngine, InferenceResult};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex as StdMutex};

    // ---- observation truncation (#813) ----

    /// Truncation must state its magnitude. The old marker was a bare
    /// `…[truncated]…`, so a model could not tell whether it had lost 10 bytes
    /// or 10 MB, and a clipped table read as a complete one.
    ///
    /// This does NOT make truncation non-destructive — the elided bytes are
    /// still gone. That is the session-value-store work in #813, which needs
    /// its own design pass. This only makes the loss legible.
    #[test]
    fn truncation_reports_true_size_and_elided_amount() {
        let total = OBSERVATION_CAP + 5_000;
        let out = cap("x".repeat(total));

        assert!(
            out.contains(&format!("of {total} bytes")),
            "the TRUE size must be reported, not just the fact of truncation: {}",
            &out[out.len().saturating_sub(200)..]
        );
        assert!(
            out.contains("5000 bytes elided"),
            "the elided amount must be reported so the model can judge the loss: {}",
            &out[out.len().saturating_sub(200)..]
        );
        assert!(
            out.contains("NOT retained"),
            "the model must be told re-running is the only recovery"
        );
        // The payload itself is still bounded — the notice is additive.
        assert!(out.starts_with(&"x".repeat(1_000)));
    }

    /// An observation at or under the cap must pass through untouched — no
    /// notice, no allocation of a truncation message.
    #[test]
    fn observations_within_the_cap_are_unmodified() {
        let small = "y".repeat(OBSERVATION_CAP);
        assert_eq!(cap(small.clone()), small);
        let tiny = "hello".to_string();
        assert_eq!(cap(tiny.clone()), tiny);
    }

    /// Truncation must land on a char boundary — a multi-byte payload clipped
    /// mid-codepoint would panic on `truncate`.
    #[test]
    fn truncation_respects_char_boundaries() {
        // 3-byte chars, so OBSERVATION_CAP (16384) is not a boundary multiple.
        let s = "€".repeat(OBSERVATION_CAP);
        let out = cap(s);
        assert!(out.contains("bytes elided"));
        assert!(out.is_char_boundary(0));
    }

    // ---- no-progress guard ----

    #[test]
    fn guard_breaks_on_repeated_mutation_not_just_reads() {
        // The bug: a "mutating" tool (e.g. remember) called with identical args
        // every turn reset the guard forever and ran to max_turns. A repeated
        // identical call is idempotent — no new progress — so it must trip
        // STALL_BREAK like any other stall.
        let mut g = NoProgressGuard::default();
        // First remember of this fact IS progress (new signature).
        assert_eq!(
            g.observe("remember({\"body\":\"x\"})", true),
            GuardStep::Progress
        );
        // Re-remembering the same fact makes no progress; it accumulates to a
        // hard stop rather than resetting.
        let sig = "remember({\"body\":\"x\"})";
        let mut steps = vec![];
        for _ in 0..STALL_BREAK {
            steps.push(g.observe(sig, true));
        }
        assert!(
            steps.contains(&GuardStep::Break),
            "repeated identical mutation must eventually Break, got {steps:?}"
        );
        assert!(
            steps.contains(&GuardStep::Nudge),
            "should nudge before breaking"
        );
    }

    #[test]
    fn guard_treats_distinct_mutations_as_progress() {
        // Remembering several DIFFERENT facts is real work — never a stall.
        let mut g = NoProgressGuard::default();
        for i in 0..20 {
            let sig = format!("remember({{\"body\":\"fact-{i}\"}})");
            assert_eq!(g.observe(&sig, true), GuardStep::Progress);
        }
    }

    #[test]
    fn guard_read_only_repeat_still_breaks() {
        // Non-mutating behavior is unchanged: a repeated read loop stalls out.
        let mut g = NoProgressGuard::default();
        let mut steps = vec![];
        for _ in 0..(STALL_BREAK + 1) {
            steps.push(g.observe("recall({\"q\":\"x\"})", false));
        }
        assert!(steps.contains(&GuardStep::Break));
    }

    #[test]
    fn guard_read_only_task_never_hard_stops_without_repeat() {
        // Distinct reads never repeat a signature, so they only ever earn the
        // soft EXPLORE_NUDGE — never a Break (a genuinely read-only task is legit).
        let mut g = NoProgressGuard::default();
        let mut steps = vec![];
        for i in 0..(EXPLORE_NUDGE + 5) {
            steps.push(g.observe(&format!("read_file({{\"p\":\"f{i}\"}})"), false));
        }
        assert!(
            !steps.contains(&GuardStep::Break),
            "distinct reads must not Break"
        );
        assert!(
            steps.contains(&GuardStep::Nudge),
            "should soft-nudge after EXPLORE_NUDGE"
        );
    }

    // ---- history compaction (context-window bound) ----

    fn sys(t: &str) -> Message {
        Message::System { content: t.into() }
    }
    fn usr(t: &str) -> Message {
        Message::User { content: t.into() }
    }
    fn asst_call(id: &str) -> Message {
        Message::Assistant {
            content: String::new(),
            tool_calls: vec![serde_json::from_value(json!({
                "name": "write_file",
                "arguments": {"path": "a.js"},
                "id": id
            }))
            .unwrap()],
            thinking: vec![],
            model_id: None,
            local_last_resort: false,
        }
    }
    fn tool_res(id: &str, body: &str) -> Message {
        Message::ToolResult {
            tool_use_id: id.into(),
            content: body.into(),
            provenance: Default::default(),
        }
    }
    fn provider_item(id: &str, body: &str) -> Message {
        Message::ProviderOutputItems {
            protocol: car_inference::protocol::OPENAI_RESPONSES_PROTOCOL.into(),
            items: vec![json!({
                "type": "reasoning",
                "id": id,
                "status": "completed",
                "encrypted_content": body,
            })],
        }
    }

    /// A kept history must never begin a segment with an orphaned ToolResult
    /// (one whose Assistant call was dropped) — that is provider-invalid.
    fn no_orphan_tool_results(msgs: &[Message]) -> bool {
        let mut seen_call_ids: std::collections::HashSet<String> = Default::default();
        for m in msgs {
            match m {
                Message::Assistant { tool_calls, .. } => {
                    for c in tool_calls {
                        if let Some(id) = &c.id {
                            seen_call_ids.insert(id.clone());
                        }
                    }
                }
                Message::ToolResult { tool_use_id, .. } if !seen_call_ids.contains(tool_use_id) => {
                    return false;
                }
                _ => {}
            }
        }
        true
    }

    #[test]
    fn mutating_tools_are_derived_from_metadata_plus_builtin_file_writers() {
        let tools = vec![
            json!({"name": "remember", "mutating": true}),
            json!({"name": "recall"}),
            json!({"name": "generate_image", "mutating": true}),
        ];
        let names = mutating_tool_names(&tools);

        assert!(names.contains("write_file"));
        assert!(names.contains("edit_file"));
        assert!(names.contains("remember"));
        assert!(names.contains("generate_image"));
        assert!(!names.contains("recall"));
    }

    #[test]
    fn compaction_is_noop_under_budget_and_when_window_unknown() {
        let mut m = vec![
            sys("s"),
            usr("task"),
            asst_call("c1"),
            tool_res("c1", "small"),
        ];
        let before = m.clone();
        compact_history_to_window(&mut m, 128_000); // tiny history, huge window
        assert_eq!(m, before, "under-budget history must be untouched");
        compact_history_to_window(&mut m, 0); // unknown window
        assert_eq!(m, before, "unknown window must be a no-op");
    }

    #[test]
    fn compaction_pins_system_and_task_keeps_tail_no_orphans() {
        let big = "x".repeat(20_000); // ~5k tokens each
        let mut m = vec![sys("system"), usr("THE ORIGINAL TASK")];
        for i in 0..12 {
            m.push(asst_call(&format!("c{i}")));
            m.push(tool_res(&format!("c{i}"), &big));
        }
        let window = 20_000; // budget = 15k tokens — forces heavy trimming
        compact_history_to_window(&mut m, window);

        // Pinned head survives.
        assert!(matches!(&m[0], Message::System { .. }), "system pinned");
        assert!(
            matches!(&m[1], Message::User { content } if content == "THE ORIGINAL TASK"),
            "original task pinned"
        );
        // Recent tail survives (last exchange present).
        assert!(
            matches!(m.last(), Some(Message::ToolResult { tool_use_id, .. }) if tool_use_id == "c11"),
            "most-recent tool result kept"
        );
        // Structurally valid: no dangling tool results.
        assert!(
            no_orphan_tool_results(&m),
            "no orphaned tool results after trim"
        );
        // It actually shrank.
        assert!(m.len() < 26, "history was compacted (was 26 msgs)");
    }

    /// #814 items 2-3 — durable state must reach the model without it having to
    /// ask, and must land at the TAIL so the cached prefix stays byte-stable.
    ///
    /// Appending to the last message rather than adding a trailing
    /// `Message::System` is what makes the tail placement real: the Anthropic
    /// and Gemini handlers fold every System message into the top-level system
    /// field, so a trailing System block would land in the cached PREFIX on two
    /// of three providers — the exact invalidation item 3 exists to prevent.
    #[test]
    fn state_block_lands_at_the_tail_inside_the_last_message() {
        let mut messages = vec![
            sys("system prompt"),
            usr("do the thing"),
            tool_res("c1", "tool output here"),
        ];
        let before_prefix = format!("{:?}{:?}", messages[0], messages[1]);

        append_state_block(&mut messages, "todo: 1/3 done\n  [ ] 2 wire the CLI");

        // The block is inside the LAST message…
        let Message::ToolResult { content, .. } = &messages[2] else {
            panic!("last message should still be the tool result");
        };
        assert!(content.starts_with("tool output here"), "{content}");
        assert!(
            content.contains("wire the CLI"),
            "state must be present: {content}"
        );
        // …fenced, so it cannot read as part of the tool's own output.
        assert!(content.contains("<runtime-state>"), "{content}");
        assert!(content.contains("</runtime-state>"), "{content}");
        // …and no message was added or reordered.
        assert_eq!(messages.len(), 3);
        // The prefix is untouched — this is the property item 3 is about.
        assert_eq!(
            before_prefix,
            format!("{:?}{:?}", messages[0], messages[1]),
            "appending state must not perturb the cached prefix"
        );
    }

    /// The block is regenerated every turn, so it must never be persisted —
    /// otherwise stale copies stack up in the history, one per turn.
    #[tokio::test]
    async fn state_block_never_enters_the_durable_history() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let todos = Arc::new(tokio::sync::Mutex::new(super::super::todo::TodoList::new()));
        todos
            .lock()
            .await
            .write(&[json!({"text": "wire the CLI"})])
            .unwrap();

        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![turn("done", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("do it")];
        let mut cfg = cfg();
        cfg.todos = Some(Arc::clone(&todos));
        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;

        // The model saw it…
        let sent = seen.lock().unwrap();
        let sent_msgs = sent[0].messages.as_ref().expect("messages sent");
        let tail = format!("{:?}", sent_msgs.last().unwrap());
        assert!(
            tail.contains("wire the CLI"),
            "the model must see live state: {tail}"
        );

        // …and the stored history did not keep it.
        assert!(
            !messages
                .iter()
                .any(|m| format!("{m:?}").contains("<runtime-state>")),
            "the block must not persist into history, or it stacks one copy per turn"
        );
    }

    /// An empty plan renders nothing at all — no fence, no tokens, no cache
    /// churn for a block with no content.
    #[tokio::test]
    async fn no_state_block_when_there_is_nothing_to_say() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![turn("done", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("do it")];
        let mut cfg = cfg();
        cfg.todos = Some(Arc::new(tokio::sync::Mutex::new(
            super::super::todo::TodoList::new(),
        )));
        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;

        let sent = seen.lock().unwrap();
        let all = format!("{:?}", sent[0].messages);
        assert!(
            !all.contains("<runtime-state>"),
            "empty plan must render nothing: {all}"
        );
    }

    fn remember_receipt(subject: &str, ok: bool) -> AssistantToolReceipt {
        AssistantToolReceipt {
            tool: "remember".to_string(),
            call_id: None,
            ok,
            params: json!({"subject": subject, "body": "…"}),
            via: None,
        }
    }

    /// #814 — the recall-discipline gap the issue is actually about.
    ///
    /// A fact written at turn 3 was invisible at turn 7 unless the model
    /// independently decided to `recall`. Surfacing the subject does not hand it
    /// the content, but it does mean the model no longer has to *remember that
    /// it remembered* — the recall becomes informed rather than speculative.
    #[test]
    fn written_facts_reach_the_state_block_without_being_asked_for() {
        let receipts = [
            remember_receipt("deploy target", true),
            AssistantToolReceipt {
                tool: "read_file".to_string(),
                call_id: None,
                ok: true,
                params: json!({"path": "x"}),
                via: None,
            },
            remember_receipt("user timezone", true),
        ];

        let subjects = recent_fact_subjects(&receipts);
        assert_eq!(subjects, vec!["deploy target", "user timezone"]);

        let block = render_state_block(None, &subjects).expect("facts alone must render a block");
        assert!(block.contains("deploy target"), "{block}");
        assert!(block.contains("user timezone"), "{block}");
        // Subjects only: the bodies stay in memgine, which ranks them.
        assert!(
            block.contains("recall"),
            "must point at the content: {block}"
        );
        assert!(!block.contains('…'), "bodies must not be inlined: {block}");
    }

    /// A `remember` the runtime REJECTED wrote nothing. Listing it would tell
    /// the model it knows something it does not — worse than saying nothing,
    /// because it suppresses the retry.
    #[test]
    fn a_failed_remember_is_not_reported_as_known() {
        let receipts = [
            remember_receipt("landed fact", true),
            remember_receipt("rejected fact", false),
        ];
        assert_eq!(recent_fact_subjects(&receipts), vec!["landed fact"]);
    }

    /// A re-remember supersedes the earlier write rather than adding a second
    /// fact, so the subject must move — not duplicate, which would both inflate
    /// the count and spend the cap on one subject.
    #[test]
    fn re_remembering_a_subject_moves_it_instead_of_duplicating() {
        let receipts = [
            remember_receipt("api base url", true),
            remember_receipt("deploy target", true),
            remember_receipt("api base url", true),
        ];
        assert_eq!(
            recent_fact_subjects(&receipts),
            vec!["deploy target", "api base url"]
        );
    }

    /// The block is bounded: a run that remembers 40 things must not turn the
    /// tail into the largest part of the request. The most RECENT survive.
    #[test]
    fn the_fact_list_is_bounded_and_says_what_it_dropped() {
        let subjects: Vec<String> = (0..12).map(|i| format!("fact {i}")).collect();
        let block = render_state_block(None, &subjects).expect("must render");

        assert!(block.contains("fact 11"), "newest must survive: {block}");
        assert!(!block.contains("fact 6"), "oldest must be cut: {block}");
        assert!(
            block.contains("+7 earlier"),
            "a silent cut reads as 'that's all there is': {block}"
        );
    }

    /// Both sections are independent: either one alone renders, and neither
    /// renders an empty fence.
    #[test]
    fn sections_render_independently_and_nothing_renders_nothing() {
        assert!(render_state_block(None, &[]).is_none());
        assert!(render_state_block(Some("todo: 0/1 done".into()), &[]).is_some());
        assert!(render_state_block(None, &["a fact".to_string()]).is_some());

        let both = render_state_block(Some("todo: 0/1 done".into()), &["a fact".to_string()])
            .expect("must render");
        assert!(both.contains("todo:"), "{both}");
        assert!(both.contains("a fact"), "{both}");
    }

    /// Parslee-ai/car#815 — compaction must not be invisible.
    ///
    /// Turns used to simply cease to exist between one request and the next,
    /// so a run that degraded afterwards looked, in the trace, exactly like a
    /// model that got worse. "The model forgot" and "the harness deleted it"
    /// are different bugs with different fixes.
    #[test]
    fn compaction_leaves_a_marker_the_model_can_see() {
        let big = "x".repeat(20_000);
        let mut m = vec![sys("system"), usr("THE ORIGINAL TASK")];
        for i in 0..12 {
            m.push(asst_call(&format!("c{i}")));
            m.push(tool_res(&format!("c{i}"), &big));
        }
        compact_history_to_window(&mut m, 20_000);

        let notice = m
            .iter()
            .find_map(|msg| match msg {
                Message::System { content } if content.starts_with(COMPACTION_NOTICE_PREFIX) => {
                    Some(content.clone())
                }
                _ => None,
            })
            .expect("a compaction notice must be left in place of the removed turns");

        assert!(
            notice.contains("earlier turns removed"),
            "the notice must say turns were removed: {notice}"
        );
        assert!(
            notice.contains("events_query"),
            "a notice that says something is missing without saying how to look \
             only turns a silent failure into a visible dead end: {notice}"
        );
        // It sits at the head, where the removal happened — not appended at the
        // end, where it would read as a fact about the latest turn.
        assert!(
            matches!(&m[2], Message::System { content } if content.starts_with(COMPACTION_NOTICE_PREFIX)),
            "notice belongs where the turns were, after system + task"
        );
    }

    /// A second compaction must UPDATE the notice, not erase it or stack a
    /// second one. Erasing it would restore the exact silent-deletion property
    /// the marker exists to prevent — and the erasure would happen precisely in
    /// the long runs that need the signal most.
    #[test]
    fn repeated_compaction_accumulates_into_one_notice() {
        let big = "x".repeat(20_000);
        let mut m = vec![sys("system"), usr("THE ORIGINAL TASK")];
        for i in 0..12 {
            m.push(asst_call(&format!("c{i}")));
            m.push(tool_res(&format!("c{i}"), &big));
        }
        compact_history_to_window(&mut m, 20_000);
        let (first_turns, first_tokens) =
            parse_compaction_notice(&m[2]).expect("first notice parses");

        // Grow the history again and re-compact.
        for i in 12..24 {
            m.push(asst_call(&format!("c{i}")));
            m.push(tool_res(&format!("c{i}"), &big));
        }
        compact_history_to_window(&mut m, 20_000);

        let notices: Vec<&String> = m
            .iter()
            .filter_map(|msg| match msg {
                Message::System { content } if content.starts_with(COMPACTION_NOTICE_PREFIX) => {
                    Some(content)
                }
                _ => None,
            })
            .collect();
        assert_eq!(
            notices.len(),
            1,
            "exactly one notice, not a stack: {notices:?}"
        );

        let (turns, tokens) = parse_compaction_notice(&m[2]).expect("notice still parses");
        assert!(
            turns > first_turns && tokens > first_tokens,
            "totals must accumulate across compactions ({first_turns}/{first_tokens} \
             -> {turns}/{tokens})"
        );
    }

    /// The marker round-trips through its own text. If this drifts, repeated
    /// compaction silently resets the running totals to the latest pass.
    #[test]
    fn compaction_notice_round_trips() {
        // Arity only: the helper gained a recovery arm for the declarative
        // runner. `default()` IS this loop's existing text — see
        // `the_assistant_loops_compaction_notice_text_is_unchanged`.
        let rendered = format_compaction_notice(12, 34_000, CompactionRecovery::default());
        let parsed = parse_compaction_notice(&Message::System { content: rendered });
        assert_eq!(parsed, Some((12, 34_000)));
        // Anything else is not a notice.
        assert_eq!(
            parse_compaction_notice(&sys("ordinary system prompt")),
            None
        );
        assert_eq!(parse_compaction_notice(&usr("a user turn")), None);
    }

    #[test]
    fn compaction_keeps_responses_item_with_its_assistant_turn() {
        let big = "x".repeat(20_000);
        let mut messages = vec![sys("system"), usr("THE ORIGINAL TASK")];
        for i in 0..12 {
            messages.push(provider_item(&format!("rs_{i}"), &big));
            messages.push(asst_call(&format!("c{i}")));
            messages.push(tool_res(&format!("c{i}"), "ok"));
        }

        compact_history_to_window(&mut messages, 20_000);

        for (index, message) in messages.iter().enumerate() {
            if matches!(message, Message::ProviderOutputItems { .. }) {
                assert!(
                    matches!(messages.get(index + 1), Some(Message::Assistant { .. })),
                    "provider continuity item was orphaned from its assistant"
                );
            }
        }
        assert!(
            no_orphan_tool_results(&messages),
            "compacted history contains an orphan tool result"
        );
    }

    /// End-to-end wiring: the real assistant loop, driven by a generator with a
    /// small window that emits a large assistant message each turn, must bound
    /// the running history — proving the loop calls the compactor with the
    /// model's window every turn (the fix that eliminates the `available_tokens=0`
    /// overflow). Deterministic — no live model.
    #[tokio::test]
    async fn loop_compacts_history_to_window() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        // Window 4000 → compaction budget 3000 tokens. Each turn emits ~2000
        // tokens of assistant text + a tiny tool call, so the raw history would
        // blow past the window within a few turns.
        struct WindowedBig {
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for WindowedBig {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                if i < 6 {
                    // Distinct args each turn so this exercises compaction only,
                    // not the (separate) no-progress repeat guard.
                    Ok(turn(
                        &"x".repeat(8000),
                        json!([{ "id": format!("c{i}"), "name": "calculate",
                                 "arguments": { "expression": format!("1+{i}") } }]),
                    ))
                } else {
                    Ok(turn("done", json!([])))
                }
            }
            fn context_window(&self, _model: &str) -> usize {
                4000
            }
        }

        let generator = WindowedBig {
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "system".into(),
            },
            Message::User {
                content: "THE TASK".into(),
            },
        ];
        let mut c = cfg();
        c.max_turns = 8;

        let out = run_assistant_loop(&generator, &rt, &c, &mut messages, |_e| {}).await;

        assert_eq!(out.status, "success");
        // Uncompacted this run would leave ~14 messages; compaction keeps the
        // pinned head + a recent tail, so it is materially bounded.
        assert!(
            messages.len() <= 11,
            "history bounded by compaction, got {} messages",
            messages.len()
        );
        assert!(
            matches!(&messages[0], Message::System { .. }),
            "system stays pinned"
        );
        assert!(
            matches!(&messages[1], Message::User { content } if content == "THE TASK"),
            "original task stays pinned"
        );
        assert!(
            no_orphan_tool_results(&messages),
            "no orphaned tool results in the live loop"
        );
    }

    /// The no-progress guard: a model stuck re-reading the same file (the
    /// observed gpt-5.x pathology — 49 reads, 0 writes) must be halted as
    /// `stalled`, well before the turn cap, instead of burning the whole budget.
    #[tokio::test]
    async fn loop_halts_a_no_progress_repeat_loop() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        struct Stuck;
        #[async_trait]
        impl TurnGenerator for Stuck {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                // The identical read-only action, forever.
                Ok(turn(
                    "re-reading",
                    json!([{ "name": "read_file", "arguments": { "path": "app.js" } }]),
                ))
            }
        }

        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "task".into(),
            },
        ];
        let mut c = cfg();
        c.max_turns = 40; // high on purpose: the guard, not the cap, must stop it

        let out = run_assistant_loop(&Stuck, &rt, &c, &mut messages, |_e| {}).await;

        assert_eq!(
            out.status, "stalled",
            "a no-progress loop must halt as `stalled`, not run to max_turns"
        );
        assert!(
            out.turns < 40,
            "must stop well before the turn cap, got {} turns",
            out.turns
        );
    }

    /// A read + read-only-shell cycle (re-read a file, `wc` it, re-read, `wc`…)
    /// makes no state change. Because `shell` is not a state-mutating tool, it no
    /// longer resets the guard, so this cycle is caught — the exact hole that let
    /// the observed run interleave `shell(wc)` between reads and loop forever.
    #[tokio::test]
    async fn loop_halts_a_read_plus_readonly_shell_cycle() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        struct Cycle {
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for Cycle {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                if i.is_multiple_of(2) {
                    Ok(turn(
                        "read",
                        json!([{ "name": "read_file", "arguments": { "path": "app.js" } }]),
                    ))
                } else {
                    Ok(turn(
                        "probe",
                        json!([{ "name": "shell", "arguments": { "command": "wc -l app.js" } }]),
                    ))
                }
            }
        }

        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "task".into(),
            },
        ];
        let mut c = cfg();
        c.max_turns = 40;

        let out = run_assistant_loop(
            &Cycle {
                cursor: AtomicUsize::new(0),
            },
            &rt,
            &c,
            &mut messages,
            |_e| {},
        )
        .await;

        assert_eq!(
            out.status, "stalled",
            "a read/read-only-shell cycle with no file change must halt"
        );
        assert!(out.turns < 40, "stopped before the cap, got {}", out.turns);
    }

    /// A repeatedly-*failing* mutation is not progress. The model asks for the
    /// identical `write_file` every turn but it's rejected (escapes the root),
    /// so nothing ever changes. The guard must key on mutation SUCCESS, not the
    /// mere request, and halt — the "40 failed writes" twin of the read loop.
    #[tokio::test]
    async fn loop_halts_a_repeatedly_failing_mutation() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        struct FailWrite;
        #[async_trait]
        impl TurnGenerator for FailWrite {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                // Escapes the clamped root every time → the executor rejects it,
                // so it is a mutating *request* that never *succeeds*.
                Ok(turn(
                    "writing",
                    json!([{ "name": "write_file",
                             "arguments": { "path": "../../etc/evil", "content": "x" } }]),
                ))
            }
        }

        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "task".into(),
            },
        ];
        let mut c = cfg();
        c.max_turns = 40;

        let out = run_assistant_loop(&FailWrite, &rt, &c, &mut messages, |_e| {}).await;

        assert_eq!(
            out.status, "stalled",
            "a repeatedly-failing mutation makes no progress and must halt (not reset the guard)"
        );
        assert!(out.turns < 40, "stopped before the cap, got {}", out.turns);
    }

    /// A scripted turn that reports token usage, the way every real provider
    /// does. `turn()` deliberately reports none, so the two together cover the
    /// measured and unmeasured halves of the `usage: Option` contract.
    fn turn_with_usage(
        text: &str,
        tool_calls: Value,
        prompt_tokens: u64,
        completion_tokens: u64,
    ) -> InferenceResult {
        serde_json::from_value(json!({
            "text": text,
            "tool_calls": tool_calls,
            "trace_id": "t",
            "model_used": "scripted",
            "latency_ms": 25,
            "usage": {
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "total_tokens": prompt_tokens + completion_tokens,
                "context_window": 8192,
            },
        }))
        .expect("scripted InferenceResult shape with usage")
    }

    /// GAP 1, the load-bearing assertion: a completed assistant-loop run must
    /// write `InferenceMetered` events carrying real token counts, and
    /// `compute_harness_metrics` over that trajectory must report
    /// `model_calls > 0` and `total_tokens > 0`.
    ///
    /// Before this, the loop dropped `InferenceResult::usage` entirely, so a
    /// `HarnessMetrics` computed from an assistant journal was structurally
    /// blank — zero tokens, zero calls — and the Evolution Agent's regression
    /// gate could never fire its token-improvement branch. Offline: the
    /// generator is scripted, so this runs in CI with no API key.
    #[tokio::test]
    async fn assistant_loop_meters_every_model_call_with_real_tokens() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        // Turn 1: a real tool call. Turn 2: finish with prose after a
        // scripted on-device last-resort fallback.
        let mut fallback_turn = turn_with_usage("The answer is 42.", json!([]), 200, 15);
        fallback_turn.model_identity.resolved_model_id = "mlx/qwen3-4b:4bit".into();
        fallback_turn.local_last_resort = true;
        let script = Script {
            turns: vec![
                turn_with_usage(
                    "computing",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                    120,
                    30,
                ),
                fallback_turn,
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "what is 6*7?".into(),
            },
        ];
        let mut assistant_events = Vec::new();
        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |event| {
            assistant_events.push(event)
        })
        .await;
        assert_eq!(outcome.status, "success");
        assert_eq!(
            outcome.models_served,
            vec![
                AssistantModelAttribution {
                    model_id: "scripted".into(),
                    local_last_resort: false,
                },
                AssistantModelAttribution {
                    model_id: "mlx/qwen3-4b:4bit".into(),
                    local_last_resort: true,
                },
            ],
            "the terminal run receipt must retain every turn, including the local fallback"
        );
        assert_eq!(
            outcome.model_used, "mlx/qwen3-4b:4bit",
            "final attribution must use the canonical resolved model id"
        );

        let transcript_attributions: Vec<_> = messages
            .iter()
            .filter_map(|message| match message {
                Message::Assistant {
                    model_id,
                    local_last_resort,
                    ..
                } => Some((model_id.as_deref(), *local_last_resort)),
                _ => None,
            })
            .collect();
        assert_eq!(
            transcript_attributions,
            vec![(Some("scripted"), false), (Some("mlx/qwen3-4b:4bit"), true)],
            "the exact replayable transcript must carry each serving attribution"
        );

        let attributions: Vec<_> = assistant_events
            .iter()
            .filter_map(|event| match event {
                AssistantEvent::ModelServed {
                    model_id,
                    local_last_resort,
                } => Some((model_id.as_str(), *local_last_resort)),
                _ => None,
            })
            .collect();
        assert_eq!(
            attributions,
            vec![("scripted", false), ("mlx/qwen3-4b:4bit", true)],
            "every completed assistant turn must emit its canonical serving model and fallback marker"
        );

        let events = rt.log.lock().await.events().to_vec();
        let metered: Vec<_> = events
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::InferenceMetered)
            .collect();
        assert_eq!(
            metered.len(),
            2,
            "one InferenceMetered per model call; the loop made 2 generate() calls"
        );
        let metered_models: Vec<_> = metered
            .iter()
            .map(|event| event.data.get("model_id").and_then(Value::as_str))
            .collect();
        assert_eq!(
            metered_models,
            vec![Some("scripted"), Some("mlx/qwen3-4b:4bit")],
            "metered events must use the same canonical serving ids"
        );
        for ev in &metered {
            assert_eq!(
                ev.data.get("usage_measured").and_then(|v| v.as_bool()),
                Some(true)
            );
        }

        let m = car_eventlog::harness_metrics::compute_harness_metrics(&events);
        assert_eq!(
            m.trajectory_efficiency.model_calls, 2,
            "harness metrics must see the model calls"
        );
        assert_eq!(
            m.trajectory_efficiency.total_tokens,
            120 + 30 + 200 + 15,
            "tokens must be the sum of the scripted usage, not an estimate"
        );
        assert!(m.trajectory_efficiency.wall_clock_ms > 0.0);

        // The gate at car_memgine::harness_evolution also needs the ACTION legs
        // (`actions_succeeded > 0` gates `candidate_did_work`, `success_rate` is
        // the only regression guard). The assistant loop routes tool calls
        // through `runtime.execute`, which meters them — assert that here so a
        // regression in either leg surfaces as a failure of THIS test rather
        // than as silently empty candidate metrics at promotion time.
        assert!(
            m.trajectory_efficiency.actions_succeeded > 0,
            "the executed `calculate` call must be recorded as a succeeded action; \
             got {m:?}"
        );
        assert!(
            m.trajectory_efficiency.success_rate.is_some(),
            "success_rate is the evolution gate's only regression guard and must be measured"
        );
    }

    /// The `usage: Option` contract must survive into the journal: a provider
    /// that reports no usage still yields a counted model call, but must NOT
    /// fabricate zero tokens. `model_calls` and `total_tokens` therefore move
    /// independently — which is why the A/B reads both.
    #[tokio::test]
    async fn unmeasured_usage_still_counts_the_call_but_fabricates_no_tokens() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        // `turn()` reports no usage at all.
        let script = Script {
            turns: vec![turn("done, no usage reported", json!([]))],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "hi".into(),
            },
        ];
        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");

        let events = rt.log.lock().await.events().to_vec();
        let metered: Vec<_> = events
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::InferenceMetered)
            .collect();
        assert_eq!(metered.len(), 1, "the call happened, so it is counted");
        assert_eq!(
            metered[0]
                .data
                .get("usage_measured")
                .and_then(|v| v.as_bool()),
            Some(false),
            "the journal must say the count was unavailable, not imply a zero"
        );

        let m = car_eventlog::harness_metrics::compute_harness_metrics(&events);
        assert_eq!(m.trajectory_efficiency.model_calls, 1);
        assert_eq!(
            m.trajectory_efficiency.total_tokens, 0,
            "no usage reported means no tokens attributed — absent, not invented"
        );
    }

    fn turn(text: &str, tool_calls: Value) -> InferenceResult {
        serde_json::from_value(json!({
            "text": text,
            "tool_calls": tool_calls,
            "trace_id": "t",
            "model_used": "scripted",
            "latency_ms": 0,
        }))
        .expect("scripted InferenceResult shape")
    }

    struct Script {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
    }

    #[async_trait]
    impl TurnGenerator for Script {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            let i = self.cursor.fetch_add(1, Ordering::SeqCst);
            self.turns.get(i).cloned().ok_or("script exhausted".into())
        }
    }

    struct CapturingGenerator {
        seen: Arc<StdMutex<Vec<GenerateRequest>>>,
    }

    #[async_trait]
    impl TurnGenerator for CapturingGenerator {
        async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
            self.seen.lock().unwrap().push(req);
            Ok(turn("done", json!([])))
        }
    }

    struct CapturingScript {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
        seen: Arc<StdMutex<Vec<GenerateRequest>>>,
    }

    #[async_trait]
    impl TurnGenerator for CapturingScript {
        async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
            self.seen.lock().unwrap().push(req);
            let i = self.cursor.fetch_add(1, Ordering::SeqCst);
            self.turns.get(i).cloned().ok_or("script exhausted".into())
        }
    }

    /// Build a real Runtime whose executor is a GeneralExecutor over a local
    /// substrate rooted at `dir` — the same wiring `build_assistant_runtime`
    /// produces, minus the network delegate.
    async fn runtime_for(dir: &std::path::Path) -> Runtime {
        let substrate: Arc<dyn Substrate> = Arc::new(LocalSubstrate::new());
        let exec: Arc<dyn ToolExecutor> =
            Arc::new(GeneralExecutor::new(substrate.clone(), dir, true));
        let engine = Arc::new(InferenceEngine::new(Default::default()));
        let rt = Runtime::new()
            .with_inference(engine)
            .with_executor(exec)
            .with_substrate(substrate);
        rt.register_agent_basics().await;
        rt.register_tool_entry(
            car_engine::ToolEntry::builtin(car_ir::builtins::shell()).with_side_effects(true),
        )
        .await;
        rt
    }

    fn cfg() -> AssistantConfig {
        AssistantConfig {
            model: Some("scripted".into()),
            strict_model: false,
            max_turns: 6,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: Vec::new(),
            approval_policy: None,
            proactive_memory: None,
            tool_memory: None,
            // None => built-in labels, which cover the network-reaching
            // commodity tools. A caller that loads .car/tool-labels.json
            // should pass the merged map (car#723).
            tool_labels: None,
            todos: None,
            value_store_previews: false,
            response_format: None,
            context_window_override: None,
            refuse_unadvertised_tools: false,
            response_format_validator: None,
            delegate_budget: None,
        }
    }

    /// What CAR does out of the box is a measured decision, not a literal that
    /// drifts (#813).
    ///
    /// The default used to be a bare `false` written at every production
    /// construction site, so "is it on?" could only be answered by grepping and
    /// hoping the sites agreed. It is now one constant, and this pins it to the
    /// value the A/B in [`VALUE_STORE_PREVIEWS_DEFAULT`]'s docs chose. Flipping
    /// it on without a new measurement fails here, which is the point: the
    /// numbers, not a preference, decide it.
    #[test]
    fn the_shipped_default_is_the_measured_one() {
        assert!(
            !VALUE_STORE_PREVIEWS_DEFAULT,
            "retained previews stay OFF because the 3-replicate car-bench-harness \
             A/B did not meet #813's fewer-calls criterion. Changing this needs a \
             new measurement, not an edit."
        );
    }

    /// Pinning the constant's *value* is not enough — production has to read it
    /// (#813).
    ///
    /// [`the_shipped_default_is_the_measured_one`] fails if the constant flips,
    /// but it says nothing about who consults it. A change that hard-coded
    /// either arm at a construction site would fork the shared default with
    /// every other test still green. So this scans the crate's own production
    /// source for both literals.
    ///
    /// Everything from the first `#[cfg(test)]` onward is excluded: test
    /// scaffolding is entitled to pin either arm explicitly, and one fixture in
    /// this very module does.
    #[test]
    fn no_production_call_site_hard_codes_the_preview_default() {
        let crate_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        for rel in [
            "src/assistant/agent_loop.rs",
            "src/assistant/chat.rs",
            "src/coder/discuss.rs",
            "src/mcp_assistant.rs",
        ] {
            let src = std::fs::read_to_string(crate_dir.join(rel))
                .unwrap_or_else(|e| panic!("reading {rel}: {e}"));
            let production = match src.find("\n#[cfg(test)]") {
                Some(cut) => &src[..cut],
                None => src.as_str(),
            };
            for literal in ["value_store_previews: false", "value_store_previews: true"] {
                assert!(
                    !production.contains(literal),
                    "{rel} hard-codes the preview arm in production code. The \
                     shipped default is VALUE_STORE_PREVIEWS_DEFAULT, chosen from a \
                     measured A/B; a literal here forks it silently."
                );
            }
        }
    }

    /// The toggle must be genuinely inert when off (#813).
    ///
    /// This mattered before the A/B because a default with *any* observable
    /// effect would have spent the measurement's credibility before it was
    /// taken. It matters just as much after: the off arm is what the measured
    /// baseline in [`VALUE_STORE_PREVIEWS_DEFAULT`] was taken against, and a
    /// caller that opts back out is entitled to the old behavior exactly. So
    /// this still asserts the off path produces the destructive-truncation
    /// observation byte for byte.
    #[tokio::test]
    async fn the_off_arm_still_truncates_exactly_as_before() {
        assert!(
            !cfg().value_store_previews,
            "this fixture is the OFF arm — it pins the pre-#813 observation path, \
             not the shipped default (see VALUE_STORE_PREVIEWS_DEFAULT)"
        );

        let dir = tempfile::tempdir().unwrap();
        let big = "x".repeat(OBSERVATION_CAP + 40_000);
        std::fs::write(dir.path().join("big.txt"), &big).unwrap();
        let rt = runtime_for(dir.path()).await;

        let script = Script {
            turns: vec![
                turn(
                    "reading",
                    json!([{ "id": "c1", "name": "read_file", "arguments": { "path": "./big.txt" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "read it".into(),
            },
        ];
        run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;

        let observation = messages
            .iter()
            .find_map(|m| match m {
                Message::ToolResult { content, .. } => Some(content.clone()),
                _ => None,
            })
            .expect("a tool observation");
        assert!(
            observation.contains("…[truncated:"),
            "off path must still truncate destructively: {}",
            &observation[observation.len().saturating_sub(200)..]
        );
        assert!(
            !observation.contains("[full value retained"),
            "no handle may leak into the default transcript"
        );
    }

    /// The property #813 is named for: with previews on, the data that used to
    /// be destroyed is still reachable *mid-run*.
    ///
    /// Proven end-to-end rather than by inspecting the store: turn 1 reads a
    /// file far larger than the cap, turn 2 passes the handle to `write_file`,
    /// and the bytes that never appeared in the transcript come back out on
    /// disk byte-identical. Under the old `cap()` this is impossible — rows
    /// 3-100, so to speak, were gone.
    #[tokio::test]
    async fn a_retained_value_survives_the_transcript_and_can_be_used_by_a_later_tool() {
        let dir = tempfile::tempdir().unwrap();
        // Distinct head and tail so a truncated copy could not pass.
        let big = format!(
            "HEAD-MARKER\n{}\nTAIL-MARKER",
            "z".repeat(OBSERVATION_CAP + 40_000)
        );
        std::fs::write(dir.path().join("big.txt"), &big).unwrap();
        let rt = runtime_for(dir.path()).await;

        let script = Script {
            turns: vec![
                turn(
                    "reading",
                    json!([{ "id": "c1", "name": "read_file", "arguments": { "path": "./big.txt" } }]),
                ),
                turn(
                    "copying",
                    json!([{ "id": "c2", "name": "write_file",
                             "arguments": { "path": "./copy.txt", "content": "$r1.content" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "copy it".into(),
            },
        ];
        let mut cfg = cfg();
        cfg.value_store_previews = true;
        cfg.max_turns = 8;
        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;

        let observation = messages
            .iter()
            .find_map(|m| match m {
                Message::ToolResult { content, .. } => Some(content.clone()),
                _ => None,
            })
            .expect("a tool observation");

        // The transcript carries shape, not the payload.
        assert!(
            observation.contains("content: text(len="),
            "the large field must announce its size and shape: {observation}"
        );
        assert!(
            observation.contains("[full value retained"),
            "the model must be told the value is reachable: {observation}"
        );
        assert!(
            observation.len() < 2_000,
            "preview must be bounded, got {} bytes",
            observation.len()
        );
        assert!(
            !observation.contains(&"z".repeat(1_000)),
            "the payload itself must not be in the transcript"
        );

        // …and the elided bytes came back through the handle.
        //
        // Compared against read_file's OWN output rather than the file on disk:
        // that tool returns line-numbered content (`     1\tHEAD-MARKER`), so a
        // byte-identical round-trip against the source was never the property.
        // What matters is that everything past the truncation point survived.
        let copied = std::fs::read_to_string(dir.path().join("copy.txt"))
            .expect("the second tool must have run with the resolved value");
        assert!(
            copied.len() > OBSERVATION_CAP,
            "only {} bytes came back; the value was not retained in full",
            copied.len()
        );
        assert!(
            copied.contains("HEAD-MARKER"),
            "the head — the only part destructive truncation ever kept — is missing"
        );
        assert!(
            copied.contains("TAIL-MARKER"),
            "the TAIL is the part cap() always destroyed; recovering it is the \
             whole point of #813"
        );
        // And it never travelled through the transcript to get there.
        assert!(
            !observation.contains("TAIL-MARKER"),
            "the tail must have come from the store, not the context: {observation}"
        );
    }

    #[tokio::test]
    async fn loop_runs_a_tool_then_finishes() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        // Turn 1: call calculate. Turn 2: finish with prose (no tool calls).
        let script = Script {
            turns: vec![
                turn(
                    "computing",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("The answer is 42.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "what is 6*7?".into(),
            },
        ];
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg(), &mut messages, |e| events.push(e)).await;

        assert_eq!(outcome.status, "success");
        assert_eq!(outcome.summary, "The answer is 42.");
        assert!(outcome.tools_called.contains(&"calculate".to_string()));
        assert!(events
            .iter()
            .any(|e| matches!(e, AssistantEvent::ToolResult { name, ok: true, .. } if name == "calculate")));
    }

    #[tokio::test]
    async fn loop_replays_managed_responses_continuity_on_second_turn() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let reasoning = json!({
            "type": "reasoning",
            "id": "rs_agent",
            "status": "completed",
            "summary": [{"type": "summary_text", "text": "safe"}],
            "encrypted_content": "opaque-agent",
        });
        let mut first = turn(
            "checking",
            json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6*7" } }]),
        );
        first.provider_output_items = vec![reasoning.clone()];
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![first, turn("done", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: seen.clone(),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "calculate".into(),
            },
        ];

        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_e| {}).await;

        assert_eq!(outcome.status, "success");
        assert!(
            !outcome.summary.contains("opaque-agent"),
            "opaque continuity must never become user-visible text"
        );
        let seen = seen.lock().unwrap();
        let second = seen[1].messages.as_ref().expect("second-turn history");
        assert!(matches!(
            &second[2],
            Message::ProviderOutputItems { protocol, items }
                if protocol == car_inference::protocol::OPENAI_RESPONSES_PROTOCOL
                    && items == &vec![reasoning]
        ));
        assert!(matches!(
            &second[3],
            Message::Assistant { content, .. } if content == "checking"
        ));
        assert!(matches!(&second[4], Message::ToolResult { .. }));
    }

    #[tokio::test]
    async fn loop_injects_proactive_memory_before_generation() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::memory::MemoryTools::open(
            dir.path().join("assistant-memory.json"),
        ));
        memory
            .execute(
                "remember",
                &json!({
                    "subject": "phoenix task requirement",
                    "body": "Requirement: for phoenix task work, run pytest before finishing."
                }),
            )
            .await
            .unwrap();
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let generator = CapturingGenerator { seen: seen.clone() };
        let mut cfg = cfg();
        cfg.proactive_memory = Some(memory);
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "finish the phoenix task".into(),
            },
        ];

        let outcome = run_assistant_loop(&generator, &rt, &cfg, &mut messages, |_| {}).await;

        assert_eq!(outcome.status, "success");
        // Scope the std MutexGuard so it drops before the `.await` below
        // (clippy::await_holding_lock).
        {
            let captured = seen.lock().unwrap();
            let context = captured[0].context.as_deref().unwrap_or("");
            assert!(
                context.contains("## Proactive Memory"),
                "request context should carry proactive memory: {context}"
            );
            assert!(
                context.contains("run pytest before finishing"),
                "selected memory should be injected: {context}"
            );
        }
        let log = rt.log.lock().await;
        assert!(log
            .events()
            .iter()
            .any(|e| e.kind == car_eventlog::EventKind::ProactiveMemoryMaintained));
        assert!(log.events().iter().any(|e| {
            e.kind == car_eventlog::EventKind::ProactiveMemoryIntervention
                && e.data.get("decision") == Some(&json!("inject"))
        }));
    }

    #[tokio::test]
    async fn loop_learns_the_call_that_recovered_a_failed_tool() {
        // The whole claim of `tool_memory`, driven through the real loop: a
        // tool fails, the next call to the SAME tool succeeds, and what
        // succeeded is what a later run gets to see.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let script = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "retrying",
                    json!([{ "id": "c2", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.tool_memory = Some(memory.clone());
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute six times seven".into(),
            },
        ];

        let outcome = run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;

        assert_eq!(outcome.status, "success");
        assert_eq!(
            memory.learned_count(),
            1,
            "the recovering call should have been learned"
        );
    }

    #[tokio::test]
    async fn a_learned_repair_reaches_the_next_run_that_hits_the_same_failure() {
        // The end-to-end claim: run one, learn; run two, the lead is in the
        // request context BEFORE the model's next turn. Two independent loops
        // over one store, which is exactly the cross-session shape.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let mut cfg = cfg();
        cfg.tool_memory = Some(memory.clone());

        let learning = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "retrying",
                    json!([{ "id": "c2", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute six times seven".into(),
            },
        ];
        run_assistant_loop(&learning, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(memory.learned_count(), 1, "run one must learn something");

        // Run two: same failure, fresh transcript, fresh loop.
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let second = CapturingScript {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "d1", "name": "calculate", "arguments": { "expression": "9 ** ** 9" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: seen.clone(),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute nine times nine".into(),
            },
        ];
        run_assistant_loop(&second, &rt, &cfg, &mut messages, |_| {}).await;

        let captured = seen.lock().unwrap();
        let first_context = captured[0].context.as_deref().unwrap_or("");
        assert!(
            !first_context.contains("## Learned Repairs"),
            "nothing has failed yet on this run: {first_context}"
        );
        let after_failure = captured[1].context.as_deref().unwrap_or("");
        assert!(
            after_failure.contains("## Learned Repairs"),
            "the turn after the failure should carry the lead: {after_failure}"
        );
        assert!(
            after_failure.contains("6*7"),
            "the lead should be the call that actually recovered: {after_failure}"
        );
    }

    #[tokio::test]
    async fn an_unrelated_later_success_is_not_credited_as_a_repair() {
        // The pairing heuristic's guard rail: a success on a DIFFERENT tool
        // never closes an open failure, so the loop cannot learn a lead that
        // had nothing to do with the failure.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let script = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "moving on",
                    json!([{ "id": "c2", "name": "write_file", "arguments": { "path": "note.txt", "content": "hi" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.tool_memory = Some(memory.clone());
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "do two things".into(),
            },
        ];

        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;

        assert_eq!(
            memory.learned_count(),
            0,
            "a different tool succeeding is not a repair for the one that failed"
        );
    }

    /// Drive the loop over a scripted fail→(gap)→succeed sequence and report
    /// what was learned. `gap` extra turns sit between the failure and the
    /// success so the recovery window can be probed at its boundary.
    async fn learn_over_gap(gap: usize, recover_with: &str) -> usize {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let mut turns = vec![turn(
            "trying",
            json!([{ "id": "c0", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
        )];
        // Filler turns on a DIFFERENT tool, so only the clock advances.
        for i in 0..gap {
            turns.push(turn(
                "thinking",
                json!([{ "id": format!("g{i}"), "name": "todo_write",
                         "arguments": { "items": [{"task": format!("step {i}"), "status": "pending"}] } }]),
            ));
        }
        turns.push(turn(
            "retrying",
            json!([{ "id": "cN", "name": "calculate", "arguments": { "expression": recover_with } }]),
        ));
        turns.push(turn("done", json!([])));
        let script = Script {
            turns,
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.max_turns = 12;
        cfg.tool_memory = Some(memory.clone());
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute six times seven".into(),
            },
        ];
        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        memory.learned_count()
    }

    #[tokio::test]
    async fn a_recovery_inside_the_window_is_learned_and_one_outside_it_is_not() {
        // The window boundary itself. A `>` / `>=` slip here would silently
        // credit unrelated later successes, which is the whole failure mode
        // RECOVERY_WINDOW_TURNS exists to bound.
        assert_eq!(learn_over_gap(0, "6*7").await, 1, "next turn is a recovery");
        assert_eq!(
            learn_over_gap(RECOVERY_WINDOW_TURNS as usize - 1, "6*7").await,
            1,
            "the last turn inside the window still counts"
        );
        assert_eq!(
            learn_over_gap(RECOVERY_WINDOW_TURNS as usize + 2, "6*7").await,
            0,
            "well past the window is not a repair"
        );
    }

    #[tokio::test]
    async fn an_identical_retry_that_happens_to_work_is_not_a_repair() {
        // The fix for the review's central finding: a success whose arguments
        // match the ones that failed changed nothing, so it teaches nothing.
        // Without this guard every routine success on a tool with an open
        // failure harvested a bogus lead.
        assert_eq!(
            learn_over_gap(0, "6 ** ** 7").await,
            0,
            "same arguments succeeding is a transient, not a repair"
        );
    }

    #[tokio::test]
    async fn a_success_on_a_different_tool_never_closes_another_tools_failure() {
        // `take_recovery` is keyed by tool name before the differs-args check.
        // Drop that key and a routine `todo_write` success becomes the durable
        // "repair" for an open `calculate` failure — a lead filed under the
        // wrong tool entirely.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let script = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "different tool",
                    json!([{ "id": "c2", "name": "todo_write",
                             "arguments": { "items": [{"task": "unrelated", "status": "pending"}] } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.tool_memory = Some(memory.clone());
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "do things".into(),
            },
        ];
        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(
            memory.learned_count(),
            0,
            "a different tool's success is not a repair for this one"
        );
    }

    #[tokio::test]
    async fn one_stale_lead_costs_exactly_one_failure_however_many_retries() {
        // The `penalized` guard. Without it, four retries behind a single
        // offered lead would push it from healthy to degraded in one run
        // (fail > success + 2), silently retiring a lead that may be fine.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let store = dir.path().join("repairs.json");
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            store.clone(),
        ));
        let mut cfg = cfg();
        cfg.max_turns = 12;
        cfg.tool_memory = Some(memory.clone());

        // Run one: learn a lead for calculate's failure signature.
        let learn = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "a1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "retrying",
                    json!([{ "id": "a2", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute".into(),
            },
        ];
        run_assistant_loop(&learn, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(memory.learned_count(), 1);

        // Run two: the lead is offered, then the same signature fails four more
        // times. Exactly one failure should be recorded, so the lead survives.
        let mut turns = Vec::new();
        for i in 0..5 {
            turns.push(turn(
                "failing",
                json!([{ "id": format!("b{i}"), "name": "calculate",
                         "arguments": { "expression": format!("{i} ** ** {i}") } }]),
            ));
        }
        turns.push(turn("giving up", json!([])));
        let retry_storm = Script {
            turns,
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute".into(),
            },
        ];
        run_assistant_loop(&retry_storm, &rt, &cfg, &mut messages, |_| {}).await;

        let sig = crate::assistant::tool_memory::FailureSignature::from_failure(
            "calculate",
            "[FAILED] bad expression",
        );
        assert!(
            memory.recall(&sig).is_some(),
            "one offered lead must cost one failure, not one per retry — \
             five penalties would have degraded it"
        );
    }

    #[tokio::test]
    async fn penalty_markers_do_not_leak_between_runs() {
        // `offered` / `penalized` live on the per-run OpenFailures. If they were
        // hoisted to the store, a fresh run's FIRST failure would count as
        // "offered and failed again" and penalize a lead that was never served.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let mut cfg = cfg();
        cfg.tool_memory = Some(memory.clone());

        let learn = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "a1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "retrying",
                    json!([{ "id": "a2", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute".into(),
            },
        ];
        run_assistant_loop(&learn, &rt, &cfg, &mut messages, |_| {}).await;

        // Three more runs, each with a single unrepaired failure. A leaked
        // marker would penalize on every one and degrade the lead by run four.
        for round in 0..3 {
            let single = Script {
                turns: vec![
                    turn(
                        "trying",
                        json!([{ "id": format!("r{round}"), "name": "calculate",
                                 "arguments": { "expression": format!("{round} ** ** {round}") } }]),
                    ),
                    turn("done", json!([])),
                ],
                cursor: AtomicUsize::new(0),
            };
            let mut messages = vec![
                Message::System {
                    content: "sys".into(),
                },
                Message::User {
                    content: "compute".into(),
                },
            ];
            run_assistant_loop(&single, &rt, &cfg, &mut messages, |_| {}).await;
        }

        let sig = crate::assistant::tool_memory::FailureSignature::from_failure(
            "calculate",
            "[FAILED] bad expression",
        );
        assert!(
            memory.recall(&sig).is_some(),
            "a first failure in a fresh run is not evidence against a lead"
        );
    }

    #[tokio::test]
    async fn the_none_path_writes_nothing_to_disk() {
        // `learning_is_off_unless_the_surface_opted_in` asserts no recall block.
        // This asserts the other half of "off": no file appears either.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let store = dir.path().join("repairs.json");
        let script = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn(
                    "retrying",
                    json!([{ "id": "c2", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute".into(),
            },
        ];
        run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert!(
            !store.exists(),
            "a surface that did not opt in must leave no store behind"
        );
    }

    #[test]
    fn a_delegate_child_inherits_the_learning_store() {
        // `delegate_child_config` builds the child with `..parent.clone()`.
        // Replacing that with an explicit literal would compile clean and
        // silently stop every sub-agent from learning, so pin it.
        let dir = tempfile::tempdir().unwrap();
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            dir.path().join("repairs.json"),
        ));
        let mut parent = cfg();
        parent.tools = vec![json!({
            "name": "calculate",
            "description": "d",
            "input_schema": {"type": "object"}
        })];
        parent.tool_memory = Some(memory.clone());
        let child = delegate_child_config(
            &parent,
            &DelegateRequest {
                goal: "sub".into(),
                tools: None,
                max_turns: 2,
            },
        )
        .expect("child config");
        let inherited = child.tool_memory.expect("child inherits the store");
        assert!(
            Arc::ptr_eq(&inherited, &memory),
            "the child must learn into the SAME store, not a fresh one"
        );
    }

    #[tokio::test]
    async fn a_secret_in_a_recovering_call_never_reaches_the_store_through_the_loop() {
        // The store-level redaction test never sees a loop-produced approach.
        // This drives a credential-shaped token through the real loop.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let store = dir.path().join("repairs.json");
        let memory = Arc::new(crate::assistant::tool_memory::ToolMemory::open(
            store.clone(),
        ));
        let script = Script {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "write_file",
                             "arguments": { "path": "x.txt", "content": "nope", "bogus": true } }]),
                ),
                turn(
                    "retrying",
                    json!([{ "id": "c2", "name": "write_file",
                             "arguments": { "path": "x.txt",
                                            "content": "token ghp_ABCDEFGHIJKLMNOPQRST" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.tool_memory = Some(memory.clone());
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "write the file".into(),
            },
        ];
        run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        if store.exists() {
            let on_disk = std::fs::read_to_string(&store).unwrap();
            assert!(
                !on_disk.contains("ghp_ABCDEFGHIJKLMNOPQRST"),
                "a credential must not survive into the durable store: {on_disk}"
            );
        }
    }

    #[tokio::test]
    async fn learning_is_off_unless_the_surface_opted_in() {
        // `tool_memory: None` must leave the loop byte-identical: no context
        // block, and nothing written anywhere.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                turn(
                    "trying",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6 ** ** 7" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: seen.clone(),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "compute".into(),
            },
        ];

        run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;

        let captured = seen.lock().unwrap();
        assert!(
            captured.iter().all(|req| !req
                .context
                .as_deref()
                .unwrap_or("")
                .contains("Learned Repairs")),
            "no surface opted in, so nothing should be recalled"
        );
    }

    #[tokio::test]
    async fn loop_journals_turn_completed_at_empty_tool_calls_terminal() {
        // The default (ungrounded) path's completion decision must be a durable,
        // queryable event. Drives the loop to the empty-tool-calls terminal and
        // asserts the journaled TurnCompleted — this would FAIL if the emit at
        // agent_loop.rs were removed (the flagship path previously had no such
        // driven-loop assertion, unlike the coder path).
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let script = Script {
            turns: vec![
                turn(
                    "computing",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "6*7" } }]),
                ),
                turn("The answer is 42.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "what is 6*7?".into(),
            },
        ];
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg(), &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "success");
        // Model provenance is threaded out to the outcome (leftover A plumbing).
        assert_eq!(outcome.model_used, "scripted");

        let log = rt.log.lock().await;
        let tc = log
            .events()
            .iter()
            .find(|e| e.kind == car_eventlog::EventKind::TurnCompleted)
            .expect("empty-tool-calls terminal must journal a TurnCompleted");
        assert_eq!(
            tc.data.get("decision"),
            Some(&serde_json::json!("empty_tool_calls"))
        );
        assert_eq!(tc.data.get("turns"), Some(&serde_json::json!(2)));
        assert_eq!(
            tc.data.get("model_id"),
            Some(&serde_json::json!("scripted"))
        );
        // "scripted" has no provider prefix in the allow-list → unknown tier.
        assert_eq!(
            tc.data.get("model_tier"),
            Some(&serde_json::json!("unknown"))
        );
    }

    #[tokio::test]
    async fn loop_journals_turn_completed_at_max_turns_terminal() {
        // The model never finishes — it calls a tool every turn until the cap is
        // hit. The max_turns terminal must journal a TurnCompleted so a run that
        // "stopped after N turns without finishing" is distinguishable from a
        // clean finish in the audit trail.
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let tool_turn = || {
            turn(
                "still going",
                json!([{ "id": "c", "name": "calculate", "arguments": { "expression": "1+1" } }]),
            )
        };
        let script = Script {
            turns: (0..10).map(|_| tool_turn()).collect(),
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "loop".into(),
            },
        ];
        // Cap below the stall-break threshold so max_turns is the terminal.
        let cfg = AssistantConfig {
            max_turns: 3,
            ..cfg()
        };
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg, &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "max_turns");

        let log = rt.log.lock().await;
        let tc = log
            .events()
            .iter()
            .find(|e| {
                e.kind == car_eventlog::EventKind::TurnCompleted
                    && e.data.get("decision") == Some(&serde_json::json!("max_turns"))
            })
            .expect("max_turns terminal must journal a TurnCompleted");
        assert_eq!(tc.data.get("turns"), Some(&serde_json::json!(3)));
    }

    #[test]
    fn summary_claim_grounding_requires_matching_receipts() {
        let ungrounded = ungrounded_summary_claims("I ran the tests and they passed.", &[]);
        assert_eq!(ungrounded, vec!["tests were run/passed"]);

        let grounded = ungrounded_summary_claims(
            "I ran the tests and they passed.",
            &[AssistantToolReceipt {
                tool: "shell".into(),
                call_id: Some("s1".into()),
                ok: true,
                params: json!({ "command": "cargo test -q" }),
                via: None,
            }],
        );
        assert!(grounded.is_empty(), "{grounded:?}");

        let failed = ungrounded_summary_claims(
            "I ran the tests and they passed.",
            &[AssistantToolReceipt {
                tool: "shell".into(),
                call_id: Some("s1".into()),
                ok: false,
                params: json!({ "command": "cargo test -q" }),
                via: None,
            }],
        );
        assert_eq!(failed, vec!["tests were run/passed"]);
    }

    #[test]
    fn summary_claim_grounding_catches_verification_and_check_claims() {
        assert_eq!(
            ungrounded_summary_claims("Verified with cargo test.", &[]),
            vec!["tests were run/passed"]
        );
        assert_eq!(
            ungrounded_summary_claims("cargo check passed.", &[]),
            vec!["build succeeded", "checks were run/passed"]
        );
        assert_eq!(
            ungrounded_summary_claims("All checks are green.", &[]),
            vec!["checks were run/passed"]
        );

        let cargo_check = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s1".into()),
            ok: true,
            params: json!({ "command": "cargo check -p car-server-core" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("cargo check passed.", &cargo_check).is_empty(),
            "cargo check receipt should ground both build and check claims"
        );

        let diff_check = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s2".into()),
            ok: true,
            params: json!({ "command": "git diff --check" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("All checks are green.", &diff_check).is_empty(),
            "diff-check receipt should ground generic check claims"
        );

        let tests = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s3".into()),
            ok: true,
            params: json!({ "command": "npm run test -- --watch=false" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("Verified with npm run test.", &tests).is_empty(),
            "npm run test receipt should ground verification test claims"
        );

        assert_eq!(
            ungrounded_summary_claims("ctest passed.", &[]),
            vec!["tests were run/passed"]
        );

        let ctest = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s4".into()),
            ok: true,
            params: json!({ "command": "ctest --test-dir build --output-on-failure" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("ctest passed.", &ctest).is_empty(),
            "ctest receipt should ground CMake test claims"
        );

        let cmake_build = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s5".into()),
            ok: true,
            params: json!({ "command": "cmake -S . -B build && cmake --build build" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("CMake build succeeded.", &cmake_build).is_empty(),
            "cmake --build receipt should ground CMake build claims"
        );

        let pnpm_check = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s6".into()),
            ok: true,
            params: json!({ "command": "pnpm check" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("Checks passed.", &pnpm_check).is_empty(),
            "package check receipts should ground generic check claims"
        );
    }

    #[test]
    fn production_investigation_claims_require_matching_live_receipts() {
        let summary = "Repository is clean and HEAD matches origin. Application Insights telemetry showed no recurrence. The production portal page was inspected.";
        assert_eq!(
            ungrounded_summary_claims(summary, &[]),
            vec![
                "repository cleanliness was verified",
                "repository revision/remote relationship was verified",
                "live Application Insights evidence was observed",
                "production browser state was observed",
            ]
        );

        let receipts = vec![
            AssistantToolReceipt {
                tool: "shell".into(),
                call_id: Some("git".into()),
                ok: true,
                params: json!({"command": "git status && git rev-parse HEAD && git rev-parse origin/main"}),
                via: None,
            },
            AssistantToolReceipt {
                tool: "shell".into(),
                call_id: Some("ai".into()),
                ok: true,
                params: json!({"command": "az monitor app-insights query --analytics-query 'exceptions | summarize count()'"}),
                via: None,
            },
            AssistantToolReceipt {
                tool: "browse_observe".into(),
                call_id: Some("browser".into()),
                ok: true,
                params: json!({}),
                via: None,
            },
        ];
        assert!(ungrounded_summary_claims(summary, &receipts).is_empty());

        let cautious = "Source tests assert Information-level logging without an exception object.\nApplication Insights query: not obtained.\nProduction browser state: not obtained.\nCannot determine whether N744JS is outside the subscription or whether deployment 23708 fixed the issue.";
        assert!(
            ungrounded_summary_claims(cautious, &[]).is_empty(),
            "explicitly source-scoped and negated claims must not be rejected"
        );
    }

    #[test]
    fn summary_file_claim_grounding_requires_matching_named_path() {
        let other_edit = [AssistantToolReceipt {
            tool: "edit_file".into(),
            call_id: Some("e1".into()),
            ok: true,
            params: json!({ "path": "src/other.rs" }),
            via: None,
        }];
        assert_eq!(
            ungrounded_summary_claims("Updated file src/lib.rs.", &other_edit),
            vec!["files were created/updated"]
        );

        let matching_edit = [AssistantToolReceipt {
            tool: "edit_file".into(),
            call_id: Some("e2".into()),
            ok: true,
            params: json!({ "path": "./src/lib.rs" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("Updated file src/lib.rs.", &matching_edit).is_empty(),
            "matching edit_file path should ground the specific update claim"
        );

        let shell_touch = [AssistantToolReceipt {
            tool: "shell".into(),
            call_id: Some("s1".into()),
            ok: true,
            params: json!({ "command": "touch src/lib.rs" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("Created file src/lib.rs.", &shell_touch).is_empty(),
            "matching shell command path should ground the specific creation claim"
        );
    }

    #[test]
    fn summary_read_claim_grounding_requires_matching_named_path() {
        let other_read = [AssistantToolReceipt {
            tool: "read_file".into(),
            call_id: Some("r1".into()),
            ok: true,
            params: json!({ "path": "src/other.rs" }),
            via: None,
        }];
        assert_eq!(
            ungrounded_summary_claims("Inspected file src/lib.rs.", &other_read),
            vec!["files were read/inspected"]
        );

        let matching_read = [AssistantToolReceipt {
            tool: "read_file".into(),
            call_id: Some("r2".into()),
            ok: true,
            params: json!({ "path": "src/lib.rs" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("Inspected file src/lib.rs.", &matching_read).is_empty(),
            "matching read_file path should ground the specific inspection claim"
        );

        let generic_update = [AssistantToolReceipt {
            tool: "edit_file".into(),
            call_id: Some("e1".into()),
            ok: true,
            params: json!({ "path": "src/lib.rs" }),
            via: None,
        }];
        assert!(
            ungrounded_summary_claims("Updated files.", &generic_update).is_empty(),
            "generic file claims should keep the existing tool-class grounding"
        );
    }

    /// End-to-end goal loop over REAL ground truth: the model uses the real
    /// `shell` tool to create a file; the deterministic `Command` condition
    /// reads the real filesystem; the loop re-drives until it converges. This
    /// is the behavior `/goal` cannot guarantee — completion is decided by the
    /// runtime, not a transcript read.
    #[tokio::test]
    async fn goal_loop_converges_when_the_command_check_passes() {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalSpec, GoalStatus};

        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        // Iteration 1: prose only (no tool) — no progress, goal not met.
        // Iteration 2: shell-create the file, then finish. File now exists.
        let create = crate::coder::test_cmds::touch("donefile");
        let script = Script {
            turns: vec![
                turn("Let me start.", json!([])),
                turn(
                    "creating it",
                    json!([{ "id": "s1", "name": "shell", "arguments": { "command": create } }]),
                ),
                turn("Done — created donefile.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };

        let spec = GoalSpec {
            goal: "create a file named donefile".into(),
            condition: GoalCondition::Command {
                id: "donefile".into(),
                expect_exit: 0,
            },
            governor: GoalGovernor {
                max_turns: Some(5),
                ..Default::default()
            },
        };

        let mut messages = vec![Message::System {
            content: "sys".into(),
        }];
        let never = std::sync::atomic::AtomicBool::new(false);
        let donefile = dir.path().join("donefile");
        let mut events = Vec::new();

        let result = run_assistant_goal_loop(
            &script,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            &spec,
            |_outcome| {
                // The deterministic check: does the file exist on disk?
                let exists = donefile.exists();
                async move {
                    let mut g = car_engine::GoalGather::default();
                    g.command_exits
                        .insert("donefile".into(), if exists { 0 } else { 1 });
                    g
                }
            },
            |e| events.push(e),
        )
        .await;

        assert_eq!(
            result.run.status,
            GoalStatus::Achieved,
            "{:?}",
            result.run.last_reason
        );
        assert_eq!(
            result.run.iterations, 2,
            "should converge on the 2nd iteration"
        );
        assert!(
            result.run.grounded,
            "a Command-check completion is grounded"
        );
        assert!(donefile.exists(), "the real file must have been created");
        assert_eq!(
            result.outcome.models_served.len(),
            3,
            "the terminal goal-run receipt must retain model calls from every iteration"
        );
        assert!(result
            .outcome
            .models_served
            .iter()
            .all(|attribution| attribution.model_id == "scripted"));
        let checks: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                AssistantEvent::GoalEvaluated {
                    iteration,
                    met,
                    grounded,
                    reason,
                } => Some((*iteration, *met, *grounded, reason.as_str())),
                _ => None,
            })
            .collect();
        assert_eq!(checks.len(), 2, "one verifier event per goal iteration");
        assert_eq!(checks[0].0, 1);
        assert!(
            !checks[0].1,
            "first iteration should not meet the command condition"
        );
        assert_eq!(checks[1].0, 2);
        assert!(
            checks[1].1,
            "second iteration should meet the command condition"
        );
        assert!(checks[1].2, "command-backed completion is grounded");

        let log = rt.log.lock().await;
        let goal_events: Vec<_> = log
            .events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::GoalEvaluated)
            .collect();
        assert_eq!(
            goal_events.len(),
            2,
            "event log should audit each verifier pass"
        );
        assert_eq!(goal_events[0].data.get("iteration"), Some(&json!(1)));
        assert_eq!(goal_events[0].data.get("met"), Some(&json!(false)));
        assert_eq!(
            goal_events[1].data.get("goal"),
            Some(&json!("create a file named donefile"))
        );
        assert_eq!(
            goal_events[1].data.get("condition"),
            Some(&json!({"kind": "command", "id": "donefile", "expect_exit": 0}))
        );
        assert_eq!(goal_events[1].data.get("iteration"), Some(&json!(2)));
        assert_eq!(goal_events[1].data.get("met"), Some(&json!(true)));
        assert_eq!(goal_events[1].data.get("grounded"), Some(&json!(true)));
    }

    /// F9 regression: a deterministic goal check that PASSED must not be
    /// re-opened just because the final prose named an operational claim with no
    /// matching tool receipt. The loop achieves on the first pass, records the
    /// completion as grounded (ground truth verified it), and the prose mismatch
    /// travels only as a non-authoritative note on the reply text.
    #[tokio::test]
    async fn deterministic_pass_not_reopened_by_ungrounded_prose() {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalSpec, GoalStatus};

        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        // The deterministic command check exits 0, but the model claims "tests
        // passed" with no matching shell receipt this run.
        let script = Script {
            turns: vec![turn("I ran the tests and they passed.", json!([]))],
            cursor: AtomicUsize::new(0),
        };
        let spec = GoalSpec {
            goal: "make tests pass".into(),
            condition: GoalCondition::Command {
                id: "tests".into(),
                expect_exit: 0,
            },
            governor: GoalGovernor {
                max_turns: Some(3),
                ..Default::default()
            },
        };
        let mut messages = vec![Message::System {
            content: "sys".into(),
        }];
        let never = std::sync::atomic::AtomicBool::new(false);
        let mut events = Vec::new();

        let result = run_assistant_goal_loop(
            &script,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            &spec,
            |_outcome| async move {
                let mut g = car_engine::GoalGather::default();
                g.command_exits.insert("tests".into(), 0);
                g
            },
            |e| events.push(e),
        )
        .await;

        // Achieved on the FIRST pass — the deterministic command check decided
        // completion; the ungrounded prose did not re-drive the loop.
        assert_eq!(
            result.run.status,
            GoalStatus::Achieved,
            "{:?}",
            result.run.last_reason
        );
        assert_eq!(result.run.iterations, 1);
        assert!(result.run.grounded, "command-backed completion is grounded");
        assert_eq!(result.run.evidence.len(), 1);
        assert!(result.run.evidence[0].met && result.run.evidence[0].grounded);
        // The unverified claim is annotated onto the returned reply text.
        assert!(
            result.outcome.summary.contains("[claim check]")
                && result.outcome.summary.contains("tests were run/passed"),
            "summary should carry the claim-check note: {}",
            result.outcome.summary
        );
        // ...but NEVER into the persisted `messages` thread (would leak into
        // later turns' context via chat.rs's thread persistence).
        assert!(
            !serde_json::to_string(&messages)
                .unwrap_or_default()
                .contains("[claim check]"),
            "the claim-check note must not leak into the thread messages"
        );
        // The streamed GoalEvaluated verdict stays grounded=true.
        let streamed: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                AssistantEvent::GoalEvaluated { grounded, .. } => Some(*grounded),
                _ => None,
            })
            .collect();
        assert_eq!(streamed, vec![true], "streamed verdict stays grounded=true");
        // The durable GoalEvaluated records grounded=true and a CLEAN reason —
        // the prose mismatch is not folded as a false-completion failure signal.
        let log = rt.log.lock().await;
        let goal_events: Vec<_> = log
            .events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::GoalEvaluated)
            .collect();
        assert_eq!(goal_events.len(), 1);
        assert_eq!(goal_events[0].data.get("met"), Some(&json!(true)));
        assert_eq!(goal_events[0].data.get("grounded"), Some(&json!(true)));
        assert!(
            !goal_events[0]
                .data
                .get("reason")
                .and_then(|r| r.as_str())
                .unwrap_or("")
                .contains("ungrounded assistant summary claim"),
            "durable reason must not record the prose mismatch as a failure"
        );
    }

    /// The fail-closed path is UNCHANGED when the met verdict is NOT a
    /// deterministic pass: a `ModelJudge`-satisfied goal is `grounded=false`, so
    /// an ungrounded summary claim keeps `grounded=false`, `met && grounded`
    /// never holds, and the loop halts on the governor. The Phase-0 miners keep
    /// receiving the ungrounded `GoalEvaluated` signal.
    #[tokio::test]
    async fn ungrounded_claim_without_deterministic_pass_still_fails_closed() {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalHalt, GoalSpec, GoalStatus};

        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let script = Script {
            turns: vec![turn("I ran the tests and they passed.", json!([]))],
            cursor: AtomicUsize::new(0),
        };
        // Model-judge completion: met is decided by a transcript-only verdict, so
        // the verdict is grounded=false — NOT a deterministic pass.
        let spec = GoalSpec {
            goal: "make tests pass".into(),
            condition: GoalCondition::ModelJudge { id: "judge".into() },
            governor: GoalGovernor {
                max_turns: Some(1),
                ..Default::default()
            },
        };
        let mut messages = vec![Message::System {
            content: "sys".into(),
        }];
        let never = std::sync::atomic::AtomicBool::new(false);

        let result = run_assistant_goal_loop(
            &script,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            &spec,
            |_outcome| async move {
                let mut g = car_engine::GoalGather::default();
                g.model_verdicts.insert("judge".into(), true);
                g
            },
            |_| {},
        )
        .await;

        assert_eq!(
            result.run.status,
            GoalStatus::Halted {
                halt: GoalHalt::TurnBudget
            }
        );
        assert_eq!(result.run.evidence.len(), 1);
        assert!(result.run.evidence[0].met);
        assert!(
            !result.run.evidence[0].grounded,
            "a model-judge completion with an ungrounded claim stays ungrounded"
        );
        assert!(result
            .run
            .last_reason
            .contains("ungrounded assistant summary claim"));
        // The claim travels in the verdict reason as before — NOT as a reply note
        // (annotation is exclusive to the deterministic-pass path).
        assert!(!result.outcome.summary.contains("[claim check]"));
    }

    /// On a deterministic pass, prose whose operational claim IS backed by a
    /// same-run receipt is left unannotated — the claim-check note only appears
    /// for genuinely unmatched claims.
    #[tokio::test]
    async fn grounded_prose_on_deterministic_pass_unannotated() {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalSpec, GoalStatus};

        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        // The model actually creates the file via shell, then claims it — the
        // "files were created/updated" claim is grounded by the create receipt
        // (which is why WRITE_TERMS must know the cmd spelling too, not just
        // POSIX `touch`).
        let create = crate::coder::test_cmds::touch("donefile");
        let script = Script {
            turns: vec![
                turn(
                    "creating it",
                    json!([{ "id": "s1", "name": "shell", "arguments": { "command": create } }]),
                ),
                turn("Done — created donefile.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let spec = GoalSpec {
            goal: "create a file named donefile".into(),
            condition: GoalCondition::Command {
                id: "donefile".into(),
                expect_exit: 0,
            },
            governor: GoalGovernor {
                max_turns: Some(3),
                ..Default::default()
            },
        };
        let mut messages = vec![Message::System {
            content: "sys".into(),
        }];
        let never = std::sync::atomic::AtomicBool::new(false);
        let donefile = dir.path().join("donefile");

        let result = run_assistant_goal_loop(
            &script,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            &spec,
            |_outcome| {
                let exists = donefile.exists();
                async move {
                    let mut g = car_engine::GoalGather::default();
                    g.command_exits
                        .insert("donefile".into(), if exists { 0 } else { 1 });
                    g
                }
            },
            |_| {},
        )
        .await;

        assert_eq!(
            result.run.status,
            GoalStatus::Achieved,
            "{:?}",
            result.run.last_reason
        );
        assert_eq!(result.run.iterations, 1);
        assert!(result.run.grounded);
        // No claim-check note: the file-write claim matched the shell receipt.
        assert_eq!(result.outcome.summary, "Done — created donefile.");
        assert!(!result.outcome.summary.contains("[claim check]"));
    }

    /// A goal that can never be met halts on the governor's turn budget — a
    /// hard bound, not `/goal`'s soft "or stop after N turns" prose.
    #[tokio::test]
    async fn goal_loop_halts_on_turn_budget() {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalHalt, GoalSpec, GoalStatus};

        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        // The model always just finishes with prose; the file is never created.
        struct Idle;
        #[async_trait]
        impl TurnGenerator for Idle {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                Ok(turn("thinking...", json!([])))
            }
        }

        let spec = GoalSpec {
            goal: "impossible".into(),
            condition: GoalCondition::Command {
                id: "never".into(),
                expect_exit: 0,
            },
            governor: GoalGovernor {
                max_turns: Some(3),
                ..Default::default()
            },
        };
        let mut messages = vec![Message::System {
            content: "sys".into(),
        }];
        let never = std::sync::atomic::AtomicBool::new(false);

        let result = run_assistant_goal_loop(
            &Idle,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            &spec,
            |_o| async {
                let mut g = car_engine::GoalGather::default();
                g.command_exits.insert("never".into(), 1);
                g
            },
            |_e| {},
        )
        .await;

        assert_eq!(
            result.run.status,
            GoalStatus::Halted {
                halt: GoalHalt::TurnBudget
            }
        );
        assert_eq!(result.run.iterations, 3);
    }

    /// car#1112: a `gather` that never resolves — the stand-in for a stuck
    /// approval wait, a wedged subprocess, or (were `ModelJudge` ever wired
    /// into a caller) an inference call to a dead route — must not hang the
    /// turn forever, and must not discard the primary reply the model already
    /// produced. Before this fix there was no bound on `gather().await` at
    /// all; `#[tokio::test(start_paused = true)]` proves the loop's own
    /// timeout is what ends this run, not a wall-clock coincidence — a real
    /// build would hang here without it.
    #[tokio::test(start_paused = true)]
    async fn goal_loop_fails_open_when_the_check_never_resolves() {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalHalt, GoalSpec, GoalStatus};

        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;

        // The model answers cleanly on the very first iteration. This reply
        // is what must reach the caller regardless of what the (stuck) check
        // does next.
        struct Answers;
        #[async_trait]
        impl TurnGenerator for Answers {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                Ok(turn("Here is your answer.", json!([])))
            }
        }

        let spec = GoalSpec {
            goal: "answer the question".into(),
            condition: GoalCondition::Command {
                id: "verify".into(),
                expect_exit: 0,
            },
            governor: GoalGovernor {
                // Deliberately generous — a fix that only works because the
                // turn budget also happens to be tiny isn't the fix under
                // test. If the evaluation timeout weren't wired in, this run
                // would hang forever well before ever spending a 2nd turn.
                max_turns: Some(8),
                ..Default::default()
            },
        };
        let mut messages = vec![Message::System {
            content: "sys".into(),
        }];
        let never = std::sync::atomic::AtomicBool::new(false);
        let mut events = Vec::new();

        let result = run_assistant_goal_loop(
            &Answers,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            &spec,
            |_outcome| std::future::pending::<car_engine::GoalGather>(),
            |e| events.push(e),
        )
        .await;

        assert_eq!(
            result.run.status,
            GoalStatus::Halted {
                halt: GoalHalt::EvaluationTimeout
            },
            "{:?}",
            result.run.last_reason
        );
        assert_eq!(
            result.run.iterations, 1,
            "must halt on the FIRST stuck evaluation, not burn the rest of the turn budget \
             re-running the model against a check that can never be graded"
        );
        assert!(
            result.run.last_reason.contains("did not complete within"),
            "{}",
            result.run.last_reason
        );
        assert_eq!(
            result.outcome.summary, "Here is your answer.",
            "the primary reply must survive an evaluation pass that never resolves"
        );
        assert!(
            events.iter().any(|e| matches!(
                e,
                AssistantEvent::GoalEvaluated {
                    met: false,
                    grounded: false,
                    ..
                }
            )),
            "the unevaluated outcome must still be streamed as a goal_evaluated event — \
             grounded: false, not true: there is no verdict to be grounded, the check \
             never ran (car#1113 review)"
        );
        assert!(
            !result.run.grounded,
            "GoalRun.grounded must not claim a deterministic verdict exists when the \
             check never got the chance to run"
        );
    }

    struct FixedGate(bool);
    #[async_trait]
    impl ApprovalGate for FixedGate {
        async fn request(&self, _tool: &str, _params: &Value) -> ApprovalDecision {
            if self.0 {
                ApprovalDecision::Approved
            } else {
                ApprovalDecision::Denied("user declined".into())
            }
        }
    }

    struct CapturingGen {
        images_seen: std::sync::Arc<std::sync::Mutex<Option<usize>>>,
    }
    #[async_trait]
    impl TurnGenerator for CapturingGen {
        async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
            *self.images_seen.lock().unwrap() = req.images.as_ref().map(|v| v.len());
            Ok(turn("done", json!([]))) // no tool calls → finish on turn 1
        }
    }

    #[tokio::test]
    async fn images_are_attached_to_the_first_request() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = std::sync::Arc::new(std::sync::Mutex::new(None));
        let generator = CapturingGen {
            images_seen: seen.clone(),
        };
        let img = ContentBlock::ImageUrl {
            url: "https://example.com/x.png".into(),
            detail: "auto".into(),
        };
        let mut messages = vec![
            Message::System {
                content: "s".into(),
            },
            Message::User {
                content: "describe".into(),
            },
        ];
        let never = std::sync::atomic::AtomicBool::new(false);
        let imgs = [img];
        run_assistant_loop_cancellable(
            &generator,
            &rt,
            &cfg(),
            &mut messages,
            &never,
            None,
            Some(&imgs),
            |_| {},
        )
        .await;
        assert_eq!(
            *seen.lock().unwrap(),
            Some(1),
            "the image should reach the first request"
        );
    }

    #[tokio::test]
    async fn gated_tool_is_denied_without_a_gate() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let script = Script {
            turns: vec![
                turn(
                    "",
                    json!([{ "id": "w1", "name": "write_file", "arguments": { "path": "x.txt", "content": "no" } }]),
                ),
                turn("could not write", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.gated_tools = vec!["write_file".into()];
        let mut messages = vec![
            Message::System {
                content: "s".into(),
            },
            Message::User {
                content: "write x".into(),
            },
        ];
        let never = std::sync::atomic::AtomicBool::new(false);
        let outcome = run_assistant_loop_cancellable(
            &script,
            &rt,
            &cfg,
            &mut messages,
            &never,
            None,
            None,
            |_| {},
        )
        .await;
        assert_eq!(outcome.status, "success");
        assert!(
            !dir.path().join("x.txt").exists(),
            "gated write must not run"
        );
        assert!(!outcome.tools_called.contains(&"write_file".to_string()));
    }

    #[tokio::test]
    async fn gated_tool_runs_when_approved() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let script = Script {
            turns: vec![
                turn(
                    "",
                    json!([{ "id": "w1", "name": "write_file", "arguments": { "path": "ok.txt", "content": "yes" } }]),
                ),
                turn("wrote it", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut cfg = cfg();
        cfg.gated_tools = vec!["write_file".into()];
        let gate = FixedGate(true);
        let mut messages = vec![
            Message::System {
                content: "s".into(),
            },
            Message::User {
                content: "write ok".into(),
            },
        ];
        let never = std::sync::atomic::AtomicBool::new(false);
        let outcome = run_assistant_loop_cancellable(
            &script,
            &rt,
            &cfg,
            &mut messages,
            &never,
            Some(&gate),
            None,
            |_| {},
        )
        .await;
        assert_eq!(outcome.status, "success");
        assert_eq!(
            std::fs::read_to_string(dir.path().join("ok.txt")).unwrap(),
            "yes"
        );
    }

    #[tokio::test]
    async fn loop_writes_a_file_through_the_runtime() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let script = Script {
            turns: vec![
                turn(
                    "",
                    json!([{ "id": "w1", "name": "write_file", "arguments": { "path": "hi.txt", "content": "hello" } }]),
                ),
                turn("Wrote hi.txt.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        };
        let mut messages = vec![
            Message::System {
                content: "sys".into(),
            },
            Message::User {
                content: "write hi.txt".into(),
            },
        ];
        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(
            std::fs::read_to_string(dir.path().join("hi.txt")).unwrap(),
            "hello"
        );
    }

    // ---- response_format passthrough + one-shot repair ----

    fn json_object_cfg() -> AssistantConfig {
        AssistantConfig {
            response_format: Some(car_inference::ResponseFormat::JsonObject),
            ..cfg()
        }
    }

    fn repair_notices(events: &[AssistantEvent]) -> (usize, usize) {
        let mut fired = 0;
        let mut still_invalid = 0;
        for e in events {
            if let AssistantEvent::Text(t) = e {
                if t == FORMAT_REPAIR_NOTICE {
                    fired += 1;
                }
                if t == FORMAT_REPAIR_STILL_INVALID {
                    still_invalid += 1;
                }
            }
        }
        (fired, still_invalid)
    }

    /// (a) The format is NEVER on a turn that offers tools — it suppresses
    /// tool use on real providers — and appears only on the tool-less repair
    /// turn. A run that offers no tools carries it on every turn.
    #[tokio::test]
    async fn response_format_is_never_on_tool_turns_only_on_the_repair_turn() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let tool_turn = || {
            turn(
                "computing",
                json!([{ "id": "c", "name": "calculate", "arguments": { "expression": "1+1" } }]),
            )
        };

        // With tools: tool turn, non-JSON final, repair.
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                tool_turn(),
                turn("The sum is 2.", json!([])),
                turn(r#"{"sum": 2}"#, json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("add 1 and 1, answer as JSON")];
        let outcome =
            run_assistant_loop(&script, &rt, &json_object_cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(outcome.summary, r#"{"sum": 2}"#);
        {
            let reqs = seen.lock().unwrap();
            assert_eq!(reqs.len(), 3);
            for (i, r) in reqs[..2].iter().enumerate() {
                assert!(r.tools.is_some(), "request {i} offers tools");
                assert!(
                    r.response_format.is_none(),
                    "request {i} offers tools, so it must not be JSON-constrained"
                );
            }
            assert!(reqs[2].tools.is_none(), "the repair turn offers no tools");
            assert_eq!(
                reqs[2].response_format,
                Some(car_inference::ResponseFormat::JsonObject),
                "and is the one request that carries the format"
            );
        }

        // Without tools there is nothing to suppress: every turn carries it.
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![turn(r#"{"sum": 2}"#, json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("add 1 and 1, answer as JSON")];
        let no_tools_cfg = AssistantConfig {
            tools: Vec::new(),
            ..json_object_cfg()
        };
        let outcome = run_assistant_loop(&script, &rt, &no_tools_cfg, &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        {
            let reqs = seen.lock().unwrap();
            assert_eq!(reqs.len(), 1, "a valid answer costs no extra call");
            assert!(reqs[0].tools.is_none());
            assert_eq!(
                reqs[0].response_format,
                Some(car_inference::ResponseFormat::JsonObject)
            );
        }

        // No format configured: none anywhere.
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![tool_turn(), turn("two", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("add 1 and 1")];
        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        let reqs = seen.lock().unwrap();
        assert_eq!(reqs.len(), 2);
        assert!(reqs.iter().all(|r| r.response_format.is_none()));
    }

    /// (b) A final answer that is not the requested shape triggers EXACTLY
    /// one repair call: no tools, the format set, the draft + nudge in the
    /// transcript. The repaired text is the answer and the event stream says
    /// the repair happened.
    #[tokio::test]
    async fn invalid_final_answer_triggers_exactly_one_toolless_repair() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                turn("Sure! The answer is: sum = 2.", json!([])),
                turn(r#"{"sum": 2}"#, json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("add 1 and 1, answer as JSON")];
        let mut events = Vec::new();
        let repair_cfg = AssistantConfig {
            model: Some("newsroom-editor".into()),
            strict_model: true,
            ..json_object_cfg()
        };
        let outcome =
            run_assistant_loop(&script, &rt, &repair_cfg, &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(
            outcome.summary, r#"{"sum": 2}"#,
            "the repaired text is the answer"
        );
        assert_eq!(
            outcome.turns, 1,
            "a repair is a model call, not a loop turn"
        );

        let reqs = seen.lock().unwrap();
        assert_eq!(reqs.len(), 2, "draft + exactly one repair");
        for (index, request) in reqs.iter().enumerate() {
            assert_eq!(
                request.model.as_deref(),
                Some("newsroom-editor"),
                "request {index} must retain the configured editor model"
            );
            assert!(
                request.params.strict_model,
                "request {index} must retain strict model selection"
            );
            assert_eq!(
                request.expected_row_digest, None,
                "assistant config exposes no immutable row precondition"
            );
            assert_eq!(
                request.expected_catalog_revision, None,
                "assistant config exposes no catalog revision precondition"
            );
        }
        let repair = &reqs[1];
        assert!(
            repair.tools.is_none(),
            "the repair turn advertises no tools"
        );
        assert_eq!(
            repair.response_format,
            Some(car_inference::ResponseFormat::JsonObject)
        );
        let history = repair.messages.as_ref().unwrap();
        assert!(
            matches!(history.last(), Some(Message::User { content }) if content == FORMAT_REPAIR_NUDGE),
            "the nudge is the last message the repair sees"
        );
        assert!(
            matches!(&history[history.len() - 2], Message::Assistant { content, .. } if content.contains("sum = 2")),
            "the draft is in the transcript so the model can see what it got wrong"
        );

        let (fired, still_invalid) = repair_notices(&events);
        assert_eq!(fired, 1, "the repair must be visible in the event stream");
        assert_eq!(still_invalid, 0);
        assert!(
            matches!(events.last(), Some(AssistantEvent::Done { text }) if text == r#"{"sum": 2}"#)
        );
        // The durable transcript records the whole exchange: draft, nudge, repair.
        assert!(
            matches!(messages.last(), Some(Message::Assistant { content, .. }) if content == r#"{"sum": 2}"#)
        );
        assert!(
            matches!(&messages[messages.len() - 2], Message::User { content } if content == FORMAT_REPAIR_NUDGE)
        );
    }

    /// (c) A valid final answer triggers no repair — one request, no notice.
    #[tokio::test]
    async fn valid_final_answer_triggers_no_repair() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            // A fenced object is accepted: models emit the fence even under
            // JSON mode, and the payload inside is what the caller parses.
            turns: vec![turn("```json\n{\"sum\": 2}\n```", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("add 1 and 1, answer as JSON")];
        let mut events = Vec::new();
        let outcome = run_assistant_loop(&script, &rt, &json_object_cfg(), &mut messages, |e| {
            events.push(e)
        })
        .await;
        assert_eq!(outcome.status, "success");
        assert_eq!(seen.lock().unwrap().len(), 1);
        assert_eq!(repair_notices(&events), (0, 0));
    }

    /// The contract is ONE repair, not retry-until-valid: a second miss is
    /// reported and the repaired text returned as-is.
    #[tokio::test]
    async fn a_repair_that_still_misses_is_reported_not_retried() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                turn("not json", json!([])),
                turn("still not json", json!([])),
                turn(r#"{"never": "reached"}"#, json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("answer as JSON")];
        let mut events = Vec::new();
        let outcome = run_assistant_loop(&script, &rt, &json_object_cfg(), &mut messages, |e| {
            events.push(e)
        })
        .await;
        assert_eq!(outcome.status, "success");
        assert_eq!(outcome.summary, "still not json");
        assert_eq!(seen.lock().unwrap().len(), 2, "one repair, never a second");
        assert_eq!(repair_notices(&events), (1, 1));
    }

    /// `JsonObject` needs an object; `JsonSchema` needs parseable JSON
    /// (parse-only: this crate carries no schema validator). Fences are
    /// tolerated either way.
    #[test]
    fn final_text_format_check_semantics() {
        use car_inference::ResponseFormat::{JsonObject, JsonSchema};
        let schema = JsonSchema {
            schema: json!({"type": "array"}),
            strict: false,
            name: None,
        };
        assert!(final_text_matches_format(r#"{"a": 1}"#, &JsonObject, None));
        assert!(final_text_matches_format(
            "```json\n{\"a\": 1}\n```",
            &JsonObject,
            None
        ));
        assert!(
            !final_text_matches_format("[1, 2]", &JsonObject, None),
            "an array is not an object"
        );
        assert!(!final_text_matches_format(
            "Here: {\"a\": 1}",
            &JsonObject,
            None
        ));
        assert!(
            final_text_matches_format("[1, 2]", &schema, None),
            "schema mode is parse-only"
        );
        assert!(!final_text_matches_format("nope", &schema, None));
        let requires_legs: ResponseFormatValidator = Arc::new(|v| v.get("legs").is_some());
        assert!(
            final_text_matches_format(r#"{"legs": []}"#, &schema, Some(&requires_legs)),
            "conforming JSON passes the caller's schema check"
        );
        assert!(
            !final_text_matches_format(r#"{"nope": 1}"#, &schema, Some(&requires_legs)),
            "valid JSON of the wrong shape must fail once a validator exists"
        );
        assert_eq!(extract_json_payload("```\n[1]\n```"), "[1]");
        assert_eq!(extract_json_payload("  [1] "), "[1]");
        assert_eq!(
            extract_json_payload("```json\n{}"),
            "```json\n{}",
            "an unclosed fence is left alone"
        );
    }

    // ---- context_window_override ----

    /// A generator that knows its model's window, so the override has
    /// something to be clamped against.
    struct WindowedScript {
        inner: CapturingScript,
        window: usize,
    }

    #[async_trait]
    impl TurnGenerator for WindowedScript {
        async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
            self.inner.generate(req).await
        }
        fn context_window(&self, _model: &str) -> usize {
            self.window
        }
    }

    fn windowed(window: usize) -> (WindowedScript, Arc<StdMutex<Vec<GenerateRequest>>>) {
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = WindowedScript {
            inner: CapturingScript {
                turns: vec![turn("done", json!([]))],
                cursor: AtomicUsize::new(0),
                seen: Arc::clone(&seen),
            },
            window,
        };
        (script, seen)
    }

    /// ~60k estimated tokens: fits a 200k window (budget 150k), overflows a
    /// 20k one (budget 15k).
    fn long_history() -> Vec<Message> {
        let big = "x".repeat(20_000);
        let mut m = vec![sys("system"), usr("THE ORIGINAL TASK")];
        for i in 0..12 {
            m.push(asst_call(&format!("c{i}")));
            m.push(tool_res(&format!("c{i}"), &big));
        }
        m
    }

    fn has_compaction_notice(messages: &[Message]) -> bool {
        messages.iter().any(|m| {
            matches!(m, Message::System { content } if content.starts_with(COMPACTION_NOTICE_PREFIX))
        })
    }

    fn window_advisories(events: &[AssistantEvent]) -> usize {
        events
            .iter()
            .filter(|e| matches!(e, AssistantEvent::Text(t) if t.starts_with("[context window:")))
            .count()
    }

    /// Override 20k on a 200k model: compaction fires on a history that the
    /// real window would have carried whole.
    #[tokio::test]
    async fn context_window_override_below_the_registry_window_tightens_compaction() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let (script, seen) = windowed(200_000);
        let mut messages = long_history();
        let cfg = AssistantConfig {
            context_window_override: Some(20_000),
            refuse_unadvertised_tools: false,
            response_format_validator: None,
            delegate_budget: None,
            ..cfg()
        };
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg, &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "success");
        assert!(
            has_compaction_notice(&messages),
            "the 20k override must compact"
        );
        // The compacted history is what the model saw, not just what was stored.
        let reqs = seen.lock().unwrap();
        assert!(has_compaction_notice(reqs[0].messages.as_ref().unwrap()));
        assert_eq!(window_advisories(&events), 0, "tightening is not clamped");
    }

    /// No override: the registry window governs and this history fits.
    #[tokio::test]
    async fn no_context_window_override_leaves_the_registry_window_in_charge() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let (script, _seen) = windowed(200_000);
        let mut messages = long_history();
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg(), &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "success");
        assert!(!has_compaction_notice(&messages), "60k fits a 200k window");
        assert_eq!(window_advisories(&events), 0);
    }

    /// Override 400k on a 200k model: clamped back to 200k (no compaction of
    /// a history that fits 200k, but the clamp is announced), because a window
    /// larger than the real one would reintroduce the provider-side truncation
    /// compaction exists to prevent.
    #[tokio::test]
    async fn context_window_override_above_the_registry_window_is_clamped_and_announced() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let (script, _seen) = windowed(200_000);
        let mut messages = long_history();
        let cfg = AssistantConfig {
            context_window_override: Some(400_000),
            refuse_unadvertised_tools: false,
            response_format_validator: None,
            delegate_budget: None,
            ..cfg()
        };
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg, &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "success");
        assert!(!has_compaction_notice(&messages));
        assert_eq!(window_advisories(&events), 1, "the clamp must be visible");
    }

    #[test]
    fn resolve_context_window_clamps_only_upward_against_a_known_window() {
        assert_eq!(resolve_context_window(None, 200_000), (200_000, None));
        assert_eq!(resolve_context_window(None, 0), (0, None));
        assert_eq!(
            resolve_context_window(Some(20_000), 200_000),
            (20_000, None)
        );
        let (w, advisory) = resolve_context_window(Some(400_000), 200_000);
        assert_eq!(w, 200_000);
        assert!(advisory
            .unwrap()
            .contains("exceeds the model's known window"));
        // Unknown registry window: nothing to clamp against, the override
        // stands (ignoring it would silently disable the compaction asked for).
        assert_eq!(resolve_context_window(Some(400_000), 0), (400_000, None));
    }

    #[test]
    fn the_assistant_loops_compaction_notice_text_is_unchanged() {
        // The notice became caller-parameterized so the declarative runner can
        // stop promising an `events_query` it cannot call. Every existing
        // caller must still emit the SAME bytes — asserted as a literal, so a
        // future edit to the other arm cannot quietly reword this one.
        assert_eq!(
            format_compaction_notice(3, 1234, CompactionRecovery::default()),
            "[history compacted: 3 earlier turns removed to fit the context window, \
             ~1234 tokens. They are gone from this transcript but the run's event log \
             still has them — call `events_query` (e.g. {\"kinds\": [\"action_failed\"], \
             \"limit\": 5}) to see what was already tried, rather than assuming you never \
             tried it.]"
        );
        assert_eq!(
            format_compaction_notice(3, 1234, CompactionRecovery::EventsQuery),
            format_compaction_notice(3, 1234, CompactionRecovery::default()),
            "EventsQuery is the default; no caller changes behavior by omitting it"
        );

        // The honest arm: same parseable shape, no recovery path it cannot offer.
        let unrecoverable = format_compaction_notice(3, 1234, CompactionRecovery::Unrecoverable);
        assert!(unrecoverable.starts_with(COMPACTION_NOTICE_PREFIX));
        assert!(!unrecoverable.contains("events_query"));
        assert!(!unrecoverable.contains("event log"));
        assert_eq!(
            parse_compaction_notice(&Message::System {
                content: unrecoverable,
            }),
            Some((3, 1234)),
            "both arms must round-trip through parse_compaction_notice"
        );
    }

    #[test]
    fn history_budget_is_the_same_number_the_inline_expression_produced() {
        // The fraction moved into a named constant; the arithmetic must not.
        // Integer division FIRST (`w / 4 * 3`), which is not the same as
        // `w * 3 / 4` for every window, and a window of 0 stays 0 so the
        // unknown-window early return keeps its meaning.
        for window in [0usize, 1, 3, 5, 4_096, 8_192, 131_072, 200_000, 1_048_576] {
            assert_eq!(
                history_budget(window),
                window / 4 * 3,
                "budget changed for window {window}"
            );
        }
        assert_eq!(history_budget(200_000), 150_000);
        assert_eq!(history_budget(0), 0);
    }

    // ---- compaction decides on the provider-reported prompt size ----

    /// A history whose chars/4 estimate is small (~6-7k tokens): the reported
    /// count, not the estimate, must be what trips compaction.
    fn modest_history() -> Vec<Message> {
        let body = "x".repeat(2_000);
        let mut m = vec![sys("system"), usr("THE ORIGINAL TASK")];
        for i in 0..12 {
            m.push(asst_call(&format!("c{i}")));
            m.push(tool_res(&format!("c{i}"), &body));
        }
        m
    }

    /// The live failure: a 150k-window run whose estimate sat under budget
    /// while the provider billed 144k, and compaction never fired. A reported
    /// prompt size far above the estimate must trigger compaction on the
    /// NEXT turn.
    #[tokio::test]
    async fn reported_prompt_tokens_far_above_the_estimate_trigger_compaction_next_turn() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = WindowedScript {
            inner: CapturingScript {
                turns: vec![
                    turn_with_usage(
                        "computing",
                        json!([{ "id": "k", "name": "calculate", "arguments": { "expression": "1+1" } }]),
                        190_000,
                        10,
                    ),
                    turn("done", json!([])),
                ],
                cursor: AtomicUsize::new(0),
                seen: Arc::clone(&seen),
            },
            window: 200_000,
        };
        let mut messages = modest_history();
        let estimate = messages.iter().map(approx_message_tokens).sum::<usize>();
        assert!(
            estimate < 20_000,
            "fixture estimate must sit far under the 150k budget: {estimate}"
        );

        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");

        let reqs = seen.lock().unwrap();
        assert_eq!(reqs.len(), 2);
        assert!(
            !has_compaction_notice(reqs[0].messages.as_ref().unwrap()),
            "turn 1 has no report yet and the estimate fits"
        );
        assert!(
            has_compaction_notice(reqs[1].messages.as_ref().unwrap()),
            "turn 2 must compact on the 190k the provider reported for turn 1"
        );
        assert!(has_compaction_notice(&messages));
    }

    /// No usage report → the estimate alone decides, exactly as before.
    #[tokio::test]
    async fn no_usage_report_falls_back_to_the_estimate() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = WindowedScript {
            inner: CapturingScript {
                turns: vec![
                    turn(
                        "computing",
                        json!([{ "id": "k", "name": "calculate", "arguments": { "expression": "1+1" } }]),
                    ),
                    turn("done", json!([])),
                ],
                cursor: AtomicUsize::new(0),
                seen: Arc::clone(&seen),
            },
            window: 200_000,
        };
        let mut messages = modest_history();
        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(seen.lock().unwrap().len(), 2);
        assert!(!has_compaction_notice(&messages));
    }

    /// The reported count is scaled onto the per-message estimates for the
    /// drop loop, so one compaction lands under budget; the notice accounts
    /// in that same measure and still round-trips.
    #[test]
    fn measured_compaction_scales_the_drop_to_the_reported_size_and_round_trips() {
        let mut m = modest_history();
        let len = m.len();
        let estimate: usize = m.iter().map(approx_message_tokens).sum();
        // ~7k estimated; the provider says 40× that, over a 150k budget.
        let measure = PromptMeasure {
            fixed_overhead: 0,
            reported: Some((estimate * 40, len)),
        };
        compact_history_measured(&mut m, 200_000, measure);
        let notice = m
            .iter()
            .find(|msg| parse_compaction_notice(msg).is_some())
            .expect("must compact on the reported size");
        let (turns, tokens) = parse_compaction_notice(notice).unwrap();
        assert!(turns > 0);
        assert!(
            tokens > estimate,
            "dropped tokens are accounted in the scaled (provider) measure: {tokens} vs raw estimate {estimate}"
        );
        // Unscaled, dropping the same turns would have removed at most the
        // raw estimate — the scale is what makes one pass sufficient.
        let remaining: usize = m.iter().map(approx_message_tokens).sum();
        assert!(remaining < estimate);
    }

    /// The tool definitions are part of every request; a history that fits
    /// the budget on its own but not with the tools must compact.
    #[test]
    fn fixed_overhead_counts_toward_the_budget() {
        let mut m = modest_history();
        let estimate: usize = m.iter().map(approx_message_tokens).sum();
        // Budget is 3/4 of the window: sit just under it on the history alone.
        let window = estimate * 4 / 3 + 40;
        compact_history_measured(&mut m, window, PromptMeasure::default());
        assert!(!has_compaction_notice(&m), "history alone fits");
        compact_history_measured(
            &mut m,
            window,
            PromptMeasure {
                fixed_overhead: 5_000,
                reported: None,
            },
        );
        assert!(has_compaction_notice(&m), "history + tool defs does not");
    }

    /// A reported count within 25% of the estimate does not rescale the
    /// per-message numbers; one further off does.
    #[test]
    fn reported_count_only_rescales_beyond_a_quarter_off() {
        let m = modest_history();
        let len = m.len();
        let estimate: usize = m.iter().map(approx_message_tokens).sum();
        // Within 25%: decide on the reported total, drop by raw estimates.
        let mut close = m.clone();
        compact_history_measured(
            &mut close,
            estimate * 4 / 3,
            PromptMeasure {
                fixed_overhead: 0,
                reported: Some((estimate * 11 / 10, len)),
            },
        );
        let (_, close_tokens) = close
            .iter()
            .find_map(parse_compaction_notice)
            .expect("110% of a budget-sized estimate must compact");
        // Far off: the same drop is accounted ~20x larger.
        let mut far = m.clone();
        compact_history_measured(
            &mut far,
            estimate * 4 / 3,
            PromptMeasure {
                fixed_overhead: 0,
                reported: Some((estimate * 20, len)),
            },
        );
        let (_, far_tokens) = far.iter().find_map(parse_compaction_notice).unwrap();
        assert!(
            far_tokens > close_tokens * 5,
            "{far_tokens} vs {close_tokens}"
        );
    }

    // ---- delegate: loop-intercepted sub-agent ----

    /// The parent's config with `delegate` advertised over its own tools.
    fn delegate_cfg() -> AssistantConfig {
        let mut tools = GeneralExecutor::tool_defs();
        tools.push(delegate_tool_def(&tools));
        AssistantConfig { tools, ..cfg() }
    }

    fn delegate_call(params: Value) -> InferenceResult {
        turn(
            "delegating",
            json!([{ "id": "d1", "name": DELEGATE_TOOL, "arguments": params }]),
        )
    }

    fn calc_call() -> InferenceResult {
        turn(
            "computing",
            json!([{ "id": "k", "name": "calculate", "arguments": { "expression": "1+1" } }]),
        )
    }

    fn tool_names(req: &GenerateRequest) -> Vec<String> {
        req.tools
            .as_deref()
            .unwrap_or_default()
            .iter()
            .filter_map(|d| d.get("name").and_then(Value::as_str))
            .map(str::to_string)
            .collect()
    }

    /// The parent's `ToolResult` for the delegate call.
    fn delegate_result(messages: &[Message]) -> (String, bool) {
        messages
            .iter()
            .find_map(|m| match m {
                Message::ToolResult {
                    tool_use_id,
                    content,
                    provenance,
                } if tool_use_id == "d1" => {
                    Some((content.clone(), *provenance == Provenance::External))
                }
                _ => None,
            })
            .expect("the delegate call must have a tool result")
    }

    #[test]
    fn delegate_tool_def_enumerates_parent_tools_and_excludes_itself() {
        let mut tools = GeneralExecutor::tool_defs();
        let def = delegate_tool_def(&tools);
        assert_eq!(def["name"], DELEGATE_TOOL);
        assert_eq!(def["tier"], "read_only");
        assert_eq!(def["mutating"], true, "a finished delegation is progress");
        assert_eq!(def["parameters"]["required"], json!(["goal"]));
        let en = def["parameters"]["properties"]["tools"]["items"]["enum"]
            .as_array()
            .unwrap()
            .clone();
        assert!(en.iter().any(|v| v == "calculate"));
        assert!(!en.iter().any(|v| v == DELEGATE_TOOL));
        // Built over a set that already carries `delegate`: still excluded.
        tools.push(def);
        let again = delegate_tool_def(&tools);
        assert!(!again["parameters"]["properties"]["tools"]["items"]["enum"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == DELEGATE_TOOL));
        assert!(mutating_tool_names(&tools).contains(DELEGATE_TOOL));
    }

    #[test]
    fn delegate_params_parse_with_defaults_and_cap() {
        let r = parse_delegate_params(&json!({"goal": " count "})).unwrap();
        assert_eq!(r.goal, "count");
        assert_eq!(r.tools, None);
        assert_eq!(r.max_turns, DELEGATE_DEFAULT_MAX_TURNS);
        let r =
            parse_delegate_params(&json!({"goal": "x", "tools": ["calculate"], "max_turns": 500}))
                .unwrap();
        assert_eq!(r.tools.as_deref(), Some(&["calculate".to_string()][..]));
        assert_eq!(r.max_turns, DELEGATE_MAX_TURNS_CAP);
        assert!(parse_delegate_params(&json!({"goal": ""})).is_err());
        assert!(parse_delegate_params(&json!({"goal": "x", "max_turns": 0})).is_err());
        assert!(parse_delegate_params(&json!({"goal": "x", "tools": "calculate"})).is_err());
    }

    /// The child config is the parent's, minus what a child must not have.
    #[test]
    fn delegate_child_config_derives_from_the_parent() {
        let mut parent = delegate_cfg();
        parent.gated_tools = vec!["shell".into()];
        parent.context_window_override = Some(20_000);
        parent.response_format = Some(car_inference::ResponseFormat::JsonObject);
        parent.todos = Some(Arc::new(tokio::sync::Mutex::new(
            super::super::todo::TodoList::new(),
        )));
        let req = parse_delegate_params(&json!({"goal": "g", "tools": ["calculate"]})).unwrap();
        let child = delegate_child_config(&parent, &req).unwrap();
        assert_eq!(
            child
                .tools
                .iter()
                .map(|d| d["name"].as_str().unwrap())
                .collect::<Vec<_>>(),
            vec!["calculate"]
        );
        assert!(child.refuse_unadvertised_tools);
        assert_eq!(child.max_turns, DELEGATE_DEFAULT_MAX_TURNS);
        assert!(child.todos.is_none());
        assert!(child.response_format.is_none(), "children answer in prose");
        assert_eq!(
            child.gated_tools, parent.gated_tools,
            "gates inherited whole"
        );
        assert_eq!(child.context_window_override, Some(20_000));
        assert_eq!(child.model, parent.model);
        // Default subset = everything delegable, still without `delegate`.
        let all = delegate_child_config(
            &parent,
            &parse_delegate_params(&json!({"goal": "g"})).unwrap(),
        )
        .unwrap();
        let names: Vec<&str> = all
            .tools
            .iter()
            .map(|d| d["name"].as_str().unwrap())
            .collect();
        assert!(names.contains(&"calculate"));
        assert!(!names.contains(&DELEGATE_TOOL));
        assert_eq!(names.len(), parent.tools.len() - 1);
    }

    /// Parent issues `delegate` → the child runs with its OWN history (system
    /// prompt + goal, none of the parent's messages), on the parent's tools
    /// minus `delegate`, and the parent receives only the child's final text.
    #[tokio::test]
    async fn delegate_child_runs_with_a_fresh_history_and_returns_only_its_final_text() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "what is 1+1? reply with the number only"})),
                calc_call(),                         // child turn 1
                turn("2", json!([])),                // child turn 2: final
                turn("The answer is 2.", json!([])), // parent turn 2
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![
            sys("PARENT SYSTEM PROMPT"),
            usr("PARENT TASK: add one and one"),
        ];
        let mut events = Vec::new();
        let outcome = run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |e| {
            events.push(e)
        })
        .await;
        assert_eq!(outcome.status, "success");
        assert_eq!(outcome.summary, "The answer is 2.");
        assert_eq!(outcome.turns, 2, "child turns are not the parent's");

        let reqs = seen.lock().unwrap();
        assert_eq!(reqs.len(), 4);
        let child_first = reqs[1].messages.as_ref().unwrap();
        assert_eq!(
            child_first.len(),
            2,
            "exactly system + goal: {child_first:?}"
        );
        assert!(
            matches!(&child_first[0], Message::System { content } if content == "PARENT SYSTEM PROMPT")
        );
        assert!(
            matches!(&child_first[1], Message::User { content } if content.starts_with("what is 1+1?"))
        );
        assert!(
            !serde_json::to_string(child_first)
                .unwrap()
                .contains("PARENT TASK"),
            "nothing from the parent's transcript reaches the child"
        );
        let child_tools = tool_names(&reqs[1]);
        assert!(child_tools.contains(&"calculate".to_string()));
        assert!(
            !child_tools.contains(&DELEGATE_TOOL.to_string()),
            "no nesting"
        );
        assert!(tool_names(&reqs[0]).contains(&DELEGATE_TOOL.to_string()));
        // The parent's second request carries the delegate result and none of
        // the child's transcript.
        let parent_second = reqs[3].messages.as_ref().unwrap();
        assert!(!serde_json::to_string(parent_second)
            .unwrap()
            .contains("computing"));
        drop(reqs);

        let (content, external) = delegate_result(&messages);
        assert_eq!(content, "2", "only the child's final text comes back");
        assert!(!external, "calculate is internal");

        // Receipt + events: the delegation is one tool call and one result.
        assert!(events
            .iter()
            .any(|e| matches!(e, AssistantEvent::ToolCall { name, .. } if name == DELEGATE_TOOL)));
        assert!(events.iter().any(|e| matches!(e, AssistantEvent::ToolResult { name, ok: true, content } if name == DELEGATE_TOOL && content == "2")));
        assert!(events.iter().any(|e| matches!(e, AssistantEvent::Text(t) if t.starts_with("[delegate: what is 1+1?") && t.ends_with("2 turns, ok]"))));
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, AssistantEvent::ToolCall { name, .. } if name == "calculate")),
            "the child's own tool calls are not forwarded"
        );
        let receipt = outcome
            .tool_receipts
            .iter()
            .find(|r| r.tool == DELEGATE_TOOL)
            .unwrap();
        assert!(receipt.ok);
        assert_eq!(receipt.call_id.as_deref(), Some("d1"));
        assert_eq!(outcome.tools_called, vec![DELEGATE_TOOL.to_string()]);
    }

    /// The granted subset holds at EXECUTION: a child call to an ungranted
    /// tool is refused with an error result, never dispatched.
    #[tokio::test]
    async fn delegate_child_tool_subset_is_enforced_at_execution() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "write hi.txt", "tools": ["calculate"]})),
                turn(
                    "writing",
                    json!([{ "id": "w", "name": "write_file", "arguments": { "path": "hi.txt", "content": "hello" } }]),
                ),
                turn("could not write", json!([])),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome =
            run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert!(
            !dir.path().join("hi.txt").exists(),
            "the ungranted write must not run"
        );
        let reqs = seen.lock().unwrap();
        assert_eq!(tool_names(&reqs[1]), vec!["calculate".to_string()]);
        let child_second = reqs[2].messages.as_ref().unwrap();
        let refusal = child_second
            .iter()
            .find_map(|m| match m {
                Message::ToolResult {
                    tool_use_id,
                    content,
                    ..
                } if tool_use_id == "w" => Some(content.clone()),
                _ => None,
            })
            .unwrap();
        assert!(
            refusal.contains("not granted to this delegate"),
            "{refusal}"
        );
        assert!(
            refusal.contains("calculate"),
            "says what IS allowed: {refusal}"
        );
    }

    /// No nesting: a child that calls `delegate` is refused at execution, and
    /// a parent that tries to GRANT `delegate` is refused as an escalation.
    #[tokio::test]
    async fn delegate_cannot_nest() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "go deeper"})),
                turn(
                    "nesting",
                    json!([{ "id": "n", "name": DELEGATE_TOOL, "arguments": { "goal": "deeper still" } }]),
                ),
                turn("could not nest", json!([])),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome =
            run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        {
            let reqs = seen.lock().unwrap();
            assert_eq!(reqs.len(), 4, "the nested call must not spawn a grandchild");
            let child_second = reqs[2].messages.as_ref().unwrap();
            let refusal = serde_json::to_string(child_second).unwrap();
            assert!(
                refusal.contains("not granted to this delegate"),
                "{refusal}"
            );
        }

        // Granting `delegate` explicitly is an escalation.
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "go deeper", "tools": [DELEGATE_TOOL]})),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        let mut events = Vec::new();
        run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |e| {
            events.push(e)
        })
        .await;
        assert_eq!(seen.lock().unwrap().len(), 2, "no child ran");
        let (content, _) = delegate_result(&messages);
        assert!(
            content.contains("privilege escalation rejected"),
            "{content}"
        );
        assert!(content.contains("cannot delegate further"), "{content}");
        assert!(events.iter().any(|e| matches!(e, AssistantEvent::ToolResult { name, ok: false, .. } if name == DELEGATE_TOOL)));
    }

    /// A name outside the parent's set is refused before any child runs.
    #[tokio::test]
    async fn delegate_escalation_is_refused_without_spawning() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "x", "tools": ["calculate", "launch_missiles"]})),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome =
            run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(seen.lock().unwrap().len(), 2);
        let (content, _) = delegate_result(&messages);
        assert!(content.contains("launch_missiles"), "{content}");
        let receipt = outcome
            .tool_receipts
            .iter()
            .find(|r| r.tool == DELEGATE_TOOL)
            .unwrap();
        assert!(!receipt.ok);
        assert!(
            outcome.tools_called.is_empty(),
            "a refused delegation is not progress"
        );
    }

    /// A child that runs out of turns comes back as an ERROR result carrying
    /// the reason — the parent must not read an unfinished delegation as an
    /// answer.
    #[tokio::test]
    async fn delegate_turn_cap_is_an_error_result() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "keep computing", "max_turns": 1})),
                calc_call(), // child turn 1 = its whole budget
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        let mut events = Vec::new();
        let outcome = run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |e| {
            events.push(e)
        })
        .await;
        assert_eq!(outcome.status, "success");
        assert_eq!(seen.lock().unwrap().len(), 3);
        let (content, _) = delegate_result(&messages);
        assert!(content.contains("did not finish"), "{content}");
        assert!(content.contains("status: max_turns"), "{content}");
        assert!(events
            .iter()
            .any(|e| matches!(e, AssistantEvent::Text(t) if t.ends_with("1 turns, error]"))));
        assert!(events.iter().any(|e| matches!(e, AssistantEvent::ToolResult { name, ok: false, .. } if name == DELEGATE_TOOL)));
        assert!(outcome.tools_called.is_empty());
    }

    /// The child's final text is capped like any other observation.
    #[tokio::test]
    async fn delegate_summary_is_capped() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let long = "y".repeat(OBSERVATION_CAP * 3);
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "dump"})),
                turn(&long, json!([])),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |_| {}).await;
        let (content, _) = delegate_result(&messages);
        assert!(content.len() < long.len());
        assert!(
            content.contains("bytes elided"),
            "the cap must say what it dropped"
        );
    }

    /// Approval inheritance by construction: a read-only parent (gated
    /// write/shell, no approver) cannot get a write done through a child —
    /// the child's write is denied by the GATE, not by the subset.
    #[tokio::test]
    async fn delegate_child_inherits_the_parents_approval_gate() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(
                    json!({"goal": "write hi.txt", "tools": ["write_file", "calculate"]}),
                ),
                turn(
                    "writing",
                    json!([{ "id": "w", "name": "write_file", "arguments": { "path": "hi.txt", "content": "hello" } }]),
                ),
                turn("denied", json!([])),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut cfg = delegate_cfg();
        cfg.gated_tools = vec!["write_file".into(), "edit_file".into(), "shell".into()];
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome = run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert!(!dir.path().join("hi.txt").exists());
        let reqs = seen.lock().unwrap();
        // The subset DID grant write_file — it is the gate that says no.
        assert!(tool_names(&reqs[1]).contains(&"write_file".to_string()));
        let child_second = serde_json::to_string(reqs[2].messages.as_ref().unwrap()).unwrap();
        assert!(child_second.contains("needs approval"), "{child_second}");
        assert!(!child_second.contains("not granted"), "{child_second}");
    }

    /// A run that never advertised `delegate` treats a `delegate` call as an
    /// unknown tool: no child, no free sub-agent.
    #[tokio::test]
    async fn delegate_is_not_intercepted_unless_advertised() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![delegate_call(json!({"goal": "x"})), turn("done", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(seen.lock().unwrap().len(), 2, "no child ran");
        let (content, _) = delegate_result(&messages);
        assert!(!content.is_empty());
        assert!(
            !content.contains("did not finish"),
            "not a delegation at all: {content}"
        );
    }

    // ---- round 4: grounding, budget, schema enforcement, repair failure ----

    /// A parent that delegates "run the tests" and reports the result is NOT
    /// flagged ungrounded: the child's own receipts are merged (tagged) into
    /// the parent's list, so the claim check sees the child's shell call.
    #[tokio::test]
    async fn delegate_child_receipts_ground_the_parents_claims() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "run the test suite and report", "tools": ["shell"]})),
                turn(
                    "running",
                    json!([{ "id": "s", "name": "shell", "arguments": { "command": "echo cargo test ok" } }]),
                ),
                turn("The suite ran.", json!([])),
                turn("I ran the tests and they passed.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("run the tests")];
        let outcome =
            run_assistant_loop(&script, &rt, &delegate_cfg(), &mut messages, |_| {}).await;
        // Without the merge this is status "error" after two redrives (the
        // script would be exhausted); with it the claim is grounded.
        assert_eq!(outcome.status, "success", "{}", outcome.summary);
        assert_eq!(outcome.summary, "I ran the tests and they passed.");
        let child_shell = outcome
            .tool_receipts
            .iter()
            .find(|r| r.tool == "shell")
            .expect("the child's shell receipt must be in the parent's list");
        assert!(child_shell.ok);
        assert_eq!(child_shell.via.as_deref(), Some("delegate:d1"));
        // The delegate's own receipt is still there, untagged.
        let del = outcome
            .tool_receipts
            .iter()
            .find(|r| r.tool == DELEGATE_TOOL)
            .unwrap();
        assert!(del.via.is_none());
        assert!(
            ungrounded_summary_claims(&outcome.summary, &outcome.tool_receipts).is_empty(),
            "the merged shell receipt grounds the tests-passed claim"
        );
    }

    /// The run-level budget: a delegate call past `max_delegations` is an
    /// error result and no child runs.
    #[tokio::test]
    async fn delegate_budget_caps_delegations() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "first"})),
                turn("one", json!([])), // child 1
                delegate_call(json!({"goal": "second"})),
                // No second child: the budget refusal is synchronous.
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut cfg = delegate_cfg();
        cfg.delegate_budget = Some(DelegateBudget {
            max_delegations: 1,
            max_child_turns: 300,
        });
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome = run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(seen.lock().unwrap().len(), 4);
        let refusals: Vec<&AssistantToolReceipt> = outcome
            .tool_receipts
            .iter()
            .filter(|r| r.tool == DELEGATE_TOOL && !r.ok)
            .collect();
        assert_eq!(refusals.len(), 1);
        let refusal_text = messages
            .iter()
            .find_map(|m| match m {
                Message::ToolResult { content, .. } if content.contains("budget exhausted") => {
                    Some(content.clone())
                }
                _ => None,
            })
            .expect("the refusal must reach the model");
        assert!(refusal_text.contains("1 delegations"), "{refusal_text}");
    }

    /// The other half of the budget: cumulative child turns.
    #[tokio::test]
    async fn delegate_budget_caps_cumulative_child_turns() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "first"})),
                calc_call(),            // child 1 turn 1
                turn("one", json!([])), // child 1 turn 2
                delegate_call(json!({"goal": "second"})),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut cfg = delegate_cfg();
        cfg.delegate_budget = Some(DelegateBudget {
            max_delegations: 20,
            max_child_turns: 2,
        });
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome = run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(seen.lock().unwrap().len(), 5, "no second child spawned");
        assert!(
            messages.iter().any(|m| matches!(m, Message::ToolResult { content, .. } if content.contains("2 child turns"))),
            "the refusal names the spent turn budget"
        );
    }

    /// Item 7: with a caller-supplied schema validator, valid JSON of the
    /// WRONG shape triggers the repair (the schema-worded nudge), and the
    /// conforming repaired answer passes.
    #[tokio::test]
    async fn json_schema_shape_mismatch_triggers_the_repair() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                turn(r#"{"nope": 1}"#, json!([])),
                turn(r#"{"legs": []}"#, json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut cfg = cfg();
        cfg.response_format = Some(car_inference::ResponseFormat::JsonSchema {
            schema: json!({"type": "object", "required": ["legs"]}),
            strict: false,
            name: None,
        });
        cfg.response_format_validator = Some(Arc::new(|v| v.get("legs").is_some()));
        let mut messages = vec![sys("sys"), usr("plan the flight, answer as JSON")];
        let mut events = Vec::new();
        let outcome =
            run_assistant_loop(&script, &rt, &cfg, &mut messages, |e| events.push(e)).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(outcome.summary, r#"{"legs": []}"#);
        let reqs = seen.lock().unwrap();
        assert_eq!(reqs.len(), 2, "valid-but-wrong-shape JSON must be repaired");
        assert!(reqs[1].tools.is_none());
        let nudge = reqs[1]
            .messages
            .as_ref()
            .unwrap()
            .last()
            .and_then(|m| match m {
                Message::User { content } => Some(content.clone()),
                _ => None,
            })
            .unwrap();
        assert!(
            nudge.contains("JSON Schema"),
            "schema-worded, not 'object': {nudge}"
        );
        assert_eq!(repair_notices(&events), (1, 0));
    }

    /// Item 4: a repair call that ERRORS (an Anthropic-protocol model
    /// rejecting response_format) keeps the DRAFT as the answer — success,
    /// with a visible failure notice and no dangling nudge in the transcript.
    #[tokio::test]
    async fn a_failed_repair_call_keeps_the_draft_answer() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            // One scripted turn: the repair call hits "script exhausted".
            turns: vec![turn("The answer is 2, not JSON.", json!([]))],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("answer as JSON")];
        let mut events = Vec::new();
        let outcome = run_assistant_loop(&script, &rt, &json_object_cfg(), &mut messages, |e| {
            events.push(e)
        })
        .await;
        assert_eq!(outcome.status, "success", "the draft is still an answer");
        assert_eq!(outcome.summary, "The answer is 2, not JSON.");
        assert!(
            events.iter().any(|e| matches!(e, AssistantEvent::Text(t) if t.starts_with(FORMAT_REPAIR_FAILED_PREFIX))),
            "the failure must be visible"
        );
        assert!(
            matches!(messages.last(), Some(Message::Assistant { content, .. }) if content == "The answer is 2, not JSON."),
            "the transcript ends on the draft answer, not a dangling nudge: {:?}",
            messages.last()
        );
    }

    /// Item 8: a delegate child cannot touch the parent's task list —
    /// `todo_write` outside the child's granted set is refused at execution
    /// (before dispatch), so the shared executor never runs it.
    #[tokio::test]
    async fn delegate_child_cannot_touch_the_parents_todo_list() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let todos = Arc::new(tokio::sync::Mutex::new(super::super::todo::TodoList::new()));
        todos
            .lock()
            .await
            .write(&[json!({"text": "the parent's plan"})])
            .unwrap();
        let before = todos.lock().await.render();

        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "reorganize", "tools": ["calculate"]})),
                turn(
                    "writing todos",
                    json!([{ "id": "t", "name": "todo_write", "arguments": { "todos": [{"text": "hijacked"}] } }]),
                ),
                turn("could not", json!([])),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut cfg = delegate_cfg();
        cfg.todos = Some(Arc::clone(&todos));
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome = run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        assert_eq!(
            todos.lock().await.render(),
            before,
            "the parent's list is untouched"
        );
        let child_second =
            serde_json::to_string(seen.lock().unwrap()[2].messages.as_ref().unwrap()).unwrap();
        assert!(
            child_second.contains("not granted to this delegate"),
            "{child_second}"
        );
    }

    /// Item 9a: the LIVE state of the provenance marking — a child receipt
    /// from an external-labelled tool marks the parent's ToolResult External.
    #[tokio::test]
    async fn delegate_marks_the_parent_result_external_when_a_child_receipt_is() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let mut tools = GeneralExecutor::tool_defs();
        // A network tool the built-in labels mark as crossing the boundary.
        tools.push(json!({
            "name": "http_request",
            "description": "Fetch a URL.",
            "parameters": {"type": "object", "properties": {"url": {"type": "string"}}}
        }));
        tools.push(delegate_tool_def(&tools));
        let cfg = AssistantConfig { tools, ..cfg() };

        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = CapturingScript {
            turns: vec![
                delegate_call(json!({"goal": "fetch the page", "tools": ["http_request"]})),
                turn(
                    "fetching",
                    json!([{ "id": "h", "name": "http_request", "arguments": { "url": "https://example.invalid/" } }]),
                ),
                turn("could not fetch", json!([])),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: Arc::clone(&seen),
        };
        let mut messages = vec![sys("sys"), usr("task")];
        let outcome = run_assistant_loop(&script, &rt, &cfg, &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        let (_, external) = delegate_result(&messages);
        assert!(
            external,
            "a child receipt from an external-labelled tool must mark the parent's result External"
        );
    }

    /// Item 9b: after a compaction the previous call's reported count is
    /// STALE (it described the pre-compaction history) and must be dropped —
    /// the next decision runs on the fresh estimate, so a short compacted
    /// history is not immediately compacted again on the old 190k number.
    #[tokio::test]
    async fn reported_count_is_reset_after_compaction_not_reused_stale() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime_for(dir.path()).await;
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let script = WindowedScript {
            inner: CapturingScript {
                turns: vec![
                    turn_with_usage(
                        "computing",
                        json!([{ "id": "k", "name": "calculate", "arguments": { "expression": "1+1" } }]),
                        190_000,
                        10,
                    ),
                    // Reports NO usage: a stale 190k, if kept, would compact again.
                    turn(
                        "still computing",
                        json!([{ "id": "k2", "name": "calculate", "arguments": { "expression": "2+2" } }]),
                    ),
                    turn("done", json!([])),
                ],
                cursor: AtomicUsize::new(0),
                seen: Arc::clone(&seen),
            },
            window: 200_000,
        };
        let mut messages = modest_history();
        let outcome = run_assistant_loop(&script, &rt, &cfg(), &mut messages, |_| {}).await;
        assert_eq!(outcome.status, "success");
        let reqs = seen.lock().unwrap();
        assert_eq!(reqs.len(), 3);
        let notice_of = |req: &GenerateRequest| {
            req.messages
                .as_ref()
                .unwrap()
                .iter()
                .find_map(parse_compaction_notice)
        };
        let after_first = notice_of(&reqs[1]).expect("turn 2 compacts on the 190k report");
        let after_second = notice_of(&reqs[2]).expect("the notice persists");
        assert_eq!(
            after_first, after_second,
            "no second compaction: the stale 190k report must not survive the first one"
        );
    }
}