car-server-core 0.47.0

//! 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 super::memory::MemoryTools;
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.
const OBSERVATION_CAP: usize = 16 * 1024;

/// 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 {
    /// 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>>,
    /// 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).
    ///
    /// **Default `false`, deliberately.** This changes what every model sees,
    /// and #813 asks for a `car-bench` A/B — pass rate, model calls, and tokens
    /// per task, before and after — before that shape changes. The claim to
    /// verify is *fewer calls at equal-or-better pass rate*; if calls do not
    /// drop, the previews are not carrying enough shape and the format needs
    /// more work rather than wider rollout. Shipping it on by default would
    /// spend that measurement's credibility before taking it.
    ///
    /// With this `false` the observation path is byte-for-byte what it was:
    /// `cap()`, same cap, same notice. Turn it on to run the comparison.
    pub value_store_previews: bool,
}

/// 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;
}

/// The terminal result of a loop run.
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>,
    /// 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,
}

/// 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;

/// 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:";

/// Render the marker. 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.
fn format_compaction_notice(turns: usize, tokens: usize) -> String {
    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.]"
    )
}

/// 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 `context_window * 3/4` (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) {
    if context_window == 0 {
        return;
    }
    let budget = context_window / 4 * 3;
    let total: usize = messages.iter().map(approx_message_tokens).sum();
    if total <= budget {
        return;
    }

    // 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 -= approx_message_tokens(&messages[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;
    let dropped_tokens: usize = messages[head_end..drop_end]
        .iter()
        .map(approx_message_tokens)
        .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,
                ),
            };
        }
        None => messages.insert(
            head_end,
            Message::System {
                content: format_compaction_notice(dropped, dropped_tokens),
            },
        ),
    }
    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),
        ),
    ])
}

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);
    }
}

/// 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>,
    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();
    let mut tool_receipts: Vec<AssistantToolReceipt> = Vec::new();
    // 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 turns = 0u32;
    // 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.
    let context_window = cfg
        .model
        .as_deref()
        .map(|m| generator.context_window(m))
        .unwrap_or(0);
    // 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);
    // 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();

    while turns < cfg.max_turns {
        if cancel.load(Ordering::Relaxed) {
            return AssistantOutcome {
                status: "cancelled",
                summary: "cancelled".to_string(),
                turns,
                tools_called,
                tool_receipts,
                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).
        compact_history_to_window(messages, context_window);

        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,
            response_format: None,
            intent: None,
            client_ref: None,
            caller: None,
        };
        maybe_apply_assistant_proactive_memory(cfg, runtime, &mut req, messages).await;

        // 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);
            }
        }

        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,
                    model_used: last_model.clone(),
                };
            }
        };
        // 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.model_used.clone();

        // No tool calls → the model's text is the final answer.
        if result.tool_calls.is_empty() {
            last_text = result.text.clone();
            // 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,
                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());

        // 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
                    }
                }
            };

            let refusal: Option<String> = match posture {
                ToolApprovalDecision::Allow => None,
                ToolApprovalDecision::Deny(reason) => Some(reason),
                ToolApprovalDecision::RequireApproval => {
                    let decision = match approval {
                        Some(gate) => gate.request(&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,
                });
                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,
                    });
                    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 ok = action
                .map(|r| matches!(r.status, ActionStatus::Succeeded))
                .unwrap_or(false);
            // 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(),
            });
            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
                },
            });
        }

        // 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,
                    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(),
            });
        }
    }

    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,
        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 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 path_hints = summary_path_hints(summary);
    let mut claims = Vec::new();
    if s.contains("test")
        && (s.contains("passed")
            || s.contains("pass")
            || s.contains("green")
            || s.contains("verified")
            || s.contains("validated")
            || s.contains("succeeded")
            || s.contains("successful")
            || s.contains("ran the test")
            || s.contains("ran tests"))
    {
        claims.push(SummaryClaimRequirement {
            label: "tests were run/passed",
            tools: &["shell"],
            require_ok: true,
            shell_terms: TEST_TERMS,
            paths: Vec::new(),
        });
    }
    if (s.contains("build") || s.contains("cargo check"))
        && (s.contains("passed")
            || s.contains("succeeded")
            || s.contains("successful")
            || s.contains("built")
            || s.contains("green")
            || s.contains("verified")
            || s.contains("validated")
            || s.contains("ran the build")
            || s.contains("ran cargo check"))
    {
        claims.push(SummaryClaimRequirement {
            label: "build succeeded",
            tools: &["shell"],
            require_ok: true,
            shell_terms: BUILD_TERMS,
            paths: Vec::new(),
        });
    }
    if (s.contains("check")
        || s.contains("checks")
        || s.contains("lint")
        || s.contains("verified")
        || s.contains("validated"))
        && (s.contains("passed")
            || s.contains("pass")
            || s.contains("green")
            || s.contains("succeeded")
            || s.contains("successful")
            || s.contains("clean"))
    {
        claims.push(SummaryClaimRequirement {
            label: "checks were run/passed",
            tools: &["shell"],
            require_ok: true,
            shell_terms: CHECK_TERMS,
            paths: Vec::new(),
        });
    }
    if (s.contains("read ") || s.contains("inspected ") || s.contains("looked at "))
        && (s.contains("file") || s.contains("files") || s.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 (s.contains("created")
        || s.contains("wrote")
        || s.contains("updated")
        || s.contains("edited"))
        && (s.contains("file") || s.contains("files"))
    {
        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(),
        });
    }
    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))
}

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).
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,
}

/// 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>,
    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 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(),
        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 });

        let mut outcome = run_assistant_loop_cancellable_in_session(
            generator,
            runtime,
            cfg,
            messages,
            cancel,
            approval,
            None,
            runtime_session_id,
            &mut emit,
        )
        .await;
        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.
        let g = gather(&outcome).await;
        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![],
        }
    }
    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": "…"}),
        }
    }

    /// #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"}),
            },
            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() {
        let rendered = format_compaction_notice(12, 34_000);
        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);
    }

    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::new(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,
            // 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,
        }
    }

    /// The toggle must be genuinely inert when off (#813).
    ///
    /// The whole reason it exists is that turning previews on changes what
    /// every model sees, and #813 asks for a `car-bench` A/B before that
    /// happens. A default that had *any* observable effect would have already
    /// spent the measurement's credibility, so this asserts the off path still
    /// produces the old destructive-truncation observation exactly.
    #[tokio::test]
    async fn previews_are_off_by_default_and_the_off_path_is_unchanged() {
        assert!(
            !cfg().value_store_previews,
            "the default must stay off until the A/B in #813 has been run"
        );

        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_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" }),
            }],
        );
        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" }),
            }],
        );
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        assert!(
            ungrounded_summary_claims("Checks passed.", &pnpm_check).is_empty(),
            "package check receipts should ground generic check claims"
        );
    }

    #[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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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" }),
        }];
        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");
        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);
    }

    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"
        );
    }
}