car-server-core 0.34.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::{format_tool_result, Runtime};
use car_inference::tasks::generate::{ContentBlock, Message, ToolCall};
use car_inference::{GenerateParams, GenerateRequest};
use car_ir::{ActionProposal, ActionStatus};
use serde_json::{json, Value};

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

/// Static configuration for a loop run.
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>,
    /// 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>,
}

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

fn cap(mut s: String) -> String {
    if s.len() <= OBSERVATION_CAP {
        return s;
    }
    let mut end = OBSERVATION_CAP;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    s.truncate(end);
    s.push_str("…[truncated]…");
    s
}

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

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

    // 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: skip forward past any
    // dangling ToolResult / ProviderOutputItems whose Assistant call was dropped.
    while drop_end < messages.len()
        && matches!(
            messages[drop_end],
            Message::ToolResult { .. } | Message::ProviderOutputItems { .. }
        )
    {
        drop_end += 1;
    }
    if drop_end <= head_end {
        return;
    }
    let dropped = drop_end - head_end;
    messages.drain(head_end..drop_end);
    tracing::debug!(
        dropped_messages = dropped,
        kept = messages.len(),
        context_window,
        budget,
        "compacted assistant history to fit the model context window"
    );
}

/// 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/memory 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", "remember"]
        .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("|")
}

/// Build a single-action `tool_call` proposal, binding the action id to the
/// call id so the result correlates back.
fn build_proposal(source: &str, call: &ToolCall) -> Result<ActionProposal, String> {
    serde_json::from_value(json!({
        "source": source,
        "actions": [{
            "id": call.id,
            "type": "tool_call",
            "tool": call.name,
            "parameters": call.arguments,
        }],
    }))
    .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]>,
    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 last_text = 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);
    // No-progress guard: signatures of non-mutating work tried since the last
    // state mutation (a signature reappearing is a stall), plus a count of turns
    // spent gathering info without changing anything.
    let mut seen_sigs: std::collections::HashSet<String> = std::collections::HashSet::new();
    let mut stall_repeats = 0u32;
    let mut turns_since_mutation = 0u32;
    let mut stall_nudged = false;

    while turns < cfg.max_turns {
        if cancel.load(Ordering::Relaxed) {
            return AssistantOutcome {
                status: "cancelled",
                summary: "cancelled".to_string(),
                turns,
                tools_called,
            };
        }
        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 req = GenerateRequest {
            prompt: String::new(),
            model: cfg.model.clone(),
            params: GenerateParams {
                temperature: 0.0,
                ..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,
        };

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

        // No tool calls → the model's text is the final answer.
        if result.tool_calls.is_empty() {
            last_text = result.text.clone();
            emit(AssistantEvent::Done {
                text: last_text.clone(),
            });
            return AssistantOutcome {
                status: "success",
                summary: last_text,
                turns,
                tools_called,
            };
        }

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

        messages.push(Message::Assistant {
            content: result.text.clone(),
            tool_calls: 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 params_val = serde_json::to_value(&call.arguments).unwrap_or_default();
            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,
                });
                continue;
            }

            let proposal = match build_proposal(&result.model_used, call) {
                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,
                    });
                    continue;
                }
            };

            let exec = runtime.execute(&proposal).await;
            let action = exec.results.first();
            let ok = action
                .map(|r| matches!(r.status, ActionStatus::Succeeded))
                .unwrap_or(false);
            let content = cap(action
                .map(format_tool_result)
                .unwrap_or_else(|| format!("tool '{}' produced no result", call.name)));
            if ok {
                tools_called.push(call.name.clone());
                if mutating_tools.contains(&call.name) {
                    mutated_ok = true;
                }
            }
            emit(AssistantEvent::ToolResult {
                name: call.name.clone(),
                ok,
                content: content.clone(),
            });
            messages.push(Message::ToolResult {
                tool_use_id: id,
                content,
            });
        }

        // No-progress guard, now that we know what actually SUCCEEDED. A real
        // state mutation resets everything (real progress). Otherwise: a repeated
        // signature is a tight loop (nudge at STALL_NUDGE, stop at STALL_BREAK),
        // and 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;
        if mutated_ok {
            seen_sigs.clear();
            stall_repeats = 0;
            turns_since_mutation = 0;
            stall_nudged = false; // a fresh stall episode later can be nudged again
        } else {
            turns_since_mutation += 1;
            if !seen_sigs.insert(tool_calls_signature(&calls)) {
                stall_repeats += 1;
                if stall_repeats >= STALL_BREAK {
                    let summary = format!(
                        "Stopped: repeated the same action {stall_repeats} times without \
                         changing anything — no progress was being made."
                    );
                    emit(AssistantEvent::Done {
                        text: summary.clone(),
                    });
                    return AssistantOutcome {
                        status: "stalled",
                        summary,
                        turns,
                        tools_called,
                    };
                }
                if stall_repeats >= STALL_NUDGE && !stall_nudged {
                    stall_nudged = true;
                    inject_nudge = true;
                }
            }
            if turns_since_mutation >= EXPLORE_NUDGE && !stall_nudged {
                stall_nudged = true;
                inject_nudge = true;
            }
        }

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

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

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

    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 outcome = run_assistant_loop_cancellable(
            generator, runtime, cfg, messages, cancel, approval, None, &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;
        let verdict = evaluate_goal(&spec.condition, &inputs);
        evidence.push(verdict.clone());
        // Audit the "why continue?" decision (tracing target so a host can
        // surface it without a new event-log kind — the WS goal_evaluated
        // notification is a later slice).
        tracing::info!(
            target: "car::goal",
            iteration = run_state.turns,
            met = verdict.met,
            grounded = verdict.grounded,
            reason = %verdict.reason,
            "goal evaluated"
        );

        if verdict.met {
            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;

    // ---- 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()],
        }
    }
    fn tool_res(id: &str, body: &str) -> Message {
        Message::ToolResult {
            tool_use_id: id.into(),
            content: body.into(),
        }
    }

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

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

    /// 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()),
            max_turns: 6,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: Vec::new(),
            approval_policy: None,
        }
    }

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

    /// 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 `touch donefile`, then finish. File now exists.
        let script = Script {
            turns: vec![
                turn("Let me start.", json!([])),
                turn(
                    "creating it",
                    json!([{ "id": "s1", "name": "shell", "arguments": { "command": "touch donefile" } }]),
                ),
                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 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| {},
        )
        .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");
    }

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