car-server-core 0.47.0

//! The native coding loop: plan → edit → verify → repair, on CAR inference.
//!
//! Shape mirrors `car-bench`'s `InferenceAgentRunner` (multi-turn tool-use
//! conversation) wrapped in `car-builder`'s repair-loop philosophy: each
//! iteration appends the previous iteration's failing check output to a
//! conversation that PERSISTS across repair rounds (F2, audit 2026-07-06 —
//! see the note above `run_native_loop`), bounded each turn by
//! [`compact_history_to_window`]. The loop only exits green when
//! [`evaluate_contract`] — not the model — says so.

use std::collections::HashMap;
use std::sync::atomic::Ordering;
use std::sync::Arc;

use async_trait::async_trait;
use car_engine::ToolExecutor;
use car_inference::tasks::generate::{Message, Provenance};
use car_inference::{GenerateParams, GenerateRequest, InferenceEngine, InferenceResult};
use serde_json::Value;

use super::budget::SessionDeadline;
use super::contract::{evaluate_contract, CheckResult, OutcomeContract};
use super::session::{CancelFlag, CoderEventKind, EventSink};
use super::shell_tool::WorktreeExecutor;
use super::skill_memory::{FailureSignature, RepairMemory};
use crate::assistant::agent_loop::compact_history_to_window;

/// The model seam: one turn of generation. Implemented by
/// [`InferenceEngine`] for production; test harnesses script it (the same
/// injected-generation philosophy as `car-builder`).
#[async_trait]
pub trait TurnGenerator: Send + Sync {
    async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String>;

    /// The context window (in tokens) of the model this generator drives, or
    /// `0` when unknown. Multi-turn drivers use it to bound their running
    /// message history before it overflows the window. Defaults to `0` so
    /// test doubles need not implement it (the loop then skips compaction).
    fn context_window(&self, _model: &str) -> usize {
        0
    }
}

#[async_trait]
impl TurnGenerator for InferenceEngine {
    async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
        self.generate_tracked(req).await.map_err(|e| e.to_string())
    }

    fn context_window(&self, model: &str) -> usize {
        self.model_context_window(model)
    }
}

/// The mid-session user-input seam: the loop hands a prompt to the host, which
/// surfaces it (emit `UserInputRequested`), blocks for the user's reply (while
/// respecting cancellation and a bound), and returns the text — or an `Err`
/// describing why no answer is coming (timeout, cancellation, no listener). An
/// `Err` is fed back to the model as a tool error so it can proceed without the
/// answer rather than the loop wedging.
///
/// Optional by design: when no asker is wired (most tests, the foreman/external
/// fallbacks), the `ask_user` tool is simply not offered to the model.
#[async_trait]
pub trait AskUser: Send + Sync {
    async fn ask(&self, prompt: &str) -> Result<String, String>;
}

/// Can the runtime reach a usable credential right now?
///
/// Exists so a session blocked on **sign-in** can wait for the human instead of
/// dying. Before this, an expired token and a dead datacenter both surfaced as
/// `LoopFailure::Infrastructure` and ended the run — discarding a worktree of
/// real edits because a token lapsed while the operator sat at the machine. The
/// two are not the same condition: one is unrecoverable in-session, the other
/// is recoverable in seconds by asking.
///
/// Optional by design, mirroring [`AskUser`]: with no gate wired the loop keeps
/// its previous behaviour exactly, so tests and the foreman/external rungs are
/// unaffected.
#[async_trait]
pub trait AuthGate: Send + Sync + std::fmt::Debug {
    /// Whether a credential the backbone will accept is available.
    async fn is_authenticated(&self) -> bool;
}

/// Does this inference error mean "the human must sign in", as opposed to "the
/// backbone is unreachable"?
///
/// Matched on the message because the typed distinction is lost by the time the
/// error reaches the loop: `car_auth::AuthOperationError` separates
/// `CoordinationDeadline` from `Terminal`, but inference collapses everything
/// into one error string. Deliberately narrow — it must never swallow a genuine
/// outage, because waiting for a human to fix a dead datacenter would hang the
/// session rather than failing it.
/// Poll until a credential appears, the caller cancels, or the window closes.
///
/// Returns `true` only when auth actually came back. Three things bound it, and
/// all three matter: the `wait` window, an explicit cancel, and the **session
/// deadline** — waiting for a human must never let a run outlive the ceiling
/// its caller set, or "wait for sign-in" becomes a way to ignore a budget.
async fn wait_for_auth(
    gate: &dyn AuthGate,
    wait: std::time::Duration,
    cancel: &CancelFlag,
    deadline: &SessionDeadline,
) -> bool {
    const POLL: std::time::Duration = std::time::Duration::from_secs(2);
    let started = std::time::Instant::now();
    loop {
        if gate.is_authenticated().await {
            return true;
        }
        if cancel.load(Ordering::SeqCst) || deadline.admit().is_some() || started.elapsed() >= wait
        {
            return false;
        }
        tokio::time::sleep(POLL).await;
    }
}

fn is_auth_failure(message: &str) -> bool {
    let m = message.to_ascii_lowercase();
    m.contains("no credential for proprietary")
        || m.contains("auth login")
        || m.contains("session has expired")
        || m.contains("cannot read parslee credentials")
        || m.contains("credential store unreadable")
}

/// The name of the model-invokable mid-session question tool. Recognized by the
/// loop (not the `WorktreeExecutor`) so the channel plumbing stays with the
/// sink + cancel flag the loop already holds.
pub const ASK_USER_TOOL: &str = "ask_user";

/// How many times the SAME read-only call (tool + args) may repeat, with no
/// intervening mutating call, before the attempt is treated as a no-progress
/// thrash — the signature of a backbone that never returns its tool results, so
/// the model re-reads the same thing forever without ever acting on it. Surfaced
/// by dogfooding: a coder on a backbone that dropped tool history issued 210
/// identical `read_file` calls and zero edits before the turn cap. On a trip the
/// loop breaks to contract evaluation (earlier edits may already have passed);
/// only a second no-progress iteration aborts the session.
const NO_PROGRESS_REPEAT_LIMIT: u32 = 6;

/// Read-only coder tools — repeating one changes nothing, so a run of identical
/// read-only calls with no mutating call between them is the no-progress signal.
/// Anything not listed here (edit_file/write_file/shell/… and any external tool)
/// is treated as *progress* and clears the guard, so a legitimate read→edit or
/// build→edit→build loop never trips — the safe-by-default direction.
///
/// Known, accepted gap: a thrash driven through `shell` (e.g. `cat foo`/`ls`
/// every turn) is treated as progress and won't trip this guard — we can't tell
/// a read-only shell from a mutating one without a heuristic that would
/// re-introduce false positives on legitimate `shell` build/poll loops. Such a
/// thrash still degrades to the bounded `max_turns` terminal, not the silent
/// budget-burn. If ever worth closing, do it with a tool-agnostic backstop (an
/// iteration with zero successful *mutating* calls is no-progress) rather than
/// classifying shell command strings.
fn is_read_only_tool(name: &str) -> bool {
    matches!(name, "read_file" | "list_dir" | "find_files" | "grep_files")
}

/// Tool definition for [`ASK_USER_TOOL`], appended to the model-visible tool
/// list only when an [`AskUser`] handler is wired.
fn ask_user_tool_def() -> Value {
    serde_json::json!({
        "name": ASK_USER_TOOL,
        "description": "Ask the human user a question and wait for their reply. \
                        Use ONLY when you genuinely cannot proceed without a \
                        decision or missing fact the user alone can supply (an \
                        ambiguous requirement, a destructive choice, a missing \
                        credential). Do not use it for things you can determine \
                        by reading the repo or running commands. The call blocks \
                        until the user answers or a timeout elapses; on timeout \
                        you receive an error and should proceed with your best \
                        judgment.",
        "parameters": {
            "type": "object",
            "properties": {
                "prompt": {
                    "type": "string",
                    "description": "The question to show the user, phrased so a short reply answers it."
                }
            },
            "required": ["prompt"]
        }
    })
}

/// Tuning for the native loop.
#[derive(Debug, Clone)]
pub struct NativeLoopConfig {
    /// Pinned model id; `None` routes adaptively (TaskHint::Code).
    pub model: Option<String>,
    /// Contract-evaluation rounds before giving up.
    pub max_iterations: u32,
    /// Model turns within one iteration before forcing evaluation.
    pub max_turns_per_iteration: u32,
    /// Generation budget per turn.
    pub max_tokens_per_turn: usize,
    /// Extra guidance appended to the system prompt, set by the harness
    /// evolution loop (car#708). `None` = the byte-identical prompt the coder
    /// used before overlays existed.
    pub prompt_overlay: Option<String>,
    /// The session's absolute deadline, SHARED with every other rung of the
    /// fallback ladder. An `Arc` rather than a value so a rung cannot restart a
    /// clock it does not own — the defect that made the first version a
    /// per-loop ceiling calling itself a session one.
    pub deadline: Arc<SessionDeadline>,
    /// Lets a session blocked on sign-in wait for the human rather than dying.
    /// `None` keeps the previous behaviour byte for byte.
    pub auth_gate: Option<Arc<dyn AuthGate>>,
    /// How long to wait for the human to re-authenticate before giving up.
    ///
    /// Generous because it bounds a *person*, not a process — they may be away
    /// from the machine. The session deadline still applies on top, so this can
    /// never extend a run past its own ceiling.
    pub auth_wait: std::time::Duration,
}

impl Default for NativeLoopConfig {
    fn default() -> Self {
        Self {
            model: None,
            max_iterations: 8,
            max_turns_per_iteration: 24,
            max_tokens_per_turn: 4096,
            prompt_overlay: None,
            deadline: SessionDeadline::shared_default(),
            auth_gate: None,
            auth_wait: std::time::Duration::from_secs(600),
        }
    }
}

impl NativeLoopConfig {
    /// Fold the general harness knobs the Evolution Agent tunes
    /// ([`car_memgine::HarnessConfig`]) onto the coder's own budgets, so a
    /// harness patch applied through `evolution.run`'s gated path actually
    /// changes coder behavior. The coder's native loop does NOT read
    /// `HarnessConfig` directly (it lives on the general `car_engine::Runtime`
    /// executor), so without this an applied patch would be inert and the A/B
    /// improvement loop could never converge. Mapping: `planning_max_replans`
    /// (how many replan/repair rounds the runtime grants) → the coder's
    /// `max_iterations` (contract-eval repair rounds); `max_retries` (per-action
    /// retry budget) → a floor on `max_turns_per_iteration`. **Only ever RAISES
    /// a budget** — a harness fix grants headroom; it never starves the coder
    /// below its base config.
    ///
    /// The session wall clock (`deadline`) is deliberately NOT raised here. It
    /// is an operator-owned safety ceiling, not a tuning knob, and the Evolution
    /// Agent granting itself more wall time would defeat the one bound that
    /// stops a runaway session. The consequence is real and worth stating: a
    /// patch raising `max_iterations` 8 -> 16 can be capped by a wall clock the
    /// harness cannot touch, so a converged-looking A/B result may in fact have
    /// been cut off. Raise `max_session_wall_secs` in `~/.car/coder.toml` when
    /// running long-budget experiments.
    pub fn merge_harness(&mut self, h: &car_memgine::HarnessConfig) {
        self.max_iterations = self
            .max_iterations
            .max(h.planning_max_replans.saturating_add(1));
        self.max_turns_per_iteration = self.max_turns_per_iteration.max(h.max_retries);
        // The prompt overlay (car#708). Unlike the budgets above this is not a
        // max() — an overlay is either in force or it is not, and a half-applied
        // one is meaningless. Rolling it back is the inverse patch clearing it.
        self.prompt_overlay = h.prompt_overlay.clone();
    }
}

/// Why a loop run ended without green checks.
///
/// One axis: "what stopped this from passing". **Descriptive, not
/// prescriptive** — each variant records what happened, never what to do next.
/// See [`LoopOutcome::failure`] for why, and do not read a retry instruction
/// out of any variant here.
///
/// Produced by four modules (`external_loop`, `native_loop`, `foreman_loop`,
/// `rpc`), so "the worker" below means whatever ran the work — an external CLI,
/// the native inference loop, or a foreman farm-out.
///
/// The first two are knowable *before* the contract can say anything: no work
/// was attempted, so there is nothing to evaluate. The rest describe a run that
/// produced work. Where a loop can evaluate the contract it does so before
/// classifying — a transport that dies mid-run must not pronounce a session
/// failed while the contract might already be green, because the worktree is
/// the state, not the process. (`rpc`'s agent-build path is the exception: it
/// has no shell contract to evaluate, only scenario results.)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LoopFailure {
    /// The worker could not be started, or never received the task — missing
    /// binary, unready CLI, pipes or prompt delivery that failed before handoff.
    /// Nothing was attempted, so the caller is free to try a different engine.
    EngineUnavailable,
    /// The user cancelled. Terminal, and explicitly NOT a fallback trigger:
    /// substituting another engine would run work the human just stopped.
    Cancelled,
    /// The machinery failed rather than the work: an invocation that died
    /// mid-run, or a backbone that stopped answering. Edits may be partially
    /// applied.
    Infrastructure,
    /// The run needs the human to sign in, and nobody did within the window.
    ///
    /// Split out of `Infrastructure` because the two call for opposite
    /// responses: an outage is not worth waiting on, whereas this resolves in
    /// seconds if someone is asked. Folding them together is what made an
    /// expired 15-minute token discard a 29-minute session — the run did not
    /// need to end, it needed to ask. Edits survive in the worktree; re-running
    /// after `car auth login` resumes from there.
    NeedsAuth,
    /// The worker ran, reported its own error, and the checks are still red.
    Execution,
    /// The worker ran clean and the checks are still red — the implementation
    /// was wrong, not the machinery.
    Verification,
    /// A loop hit its wall-clock ceiling and the next iteration was not
    /// admitted.
    ///
    /// Distinct from `Verification` because a harness-imposed cut is not a task
    /// loss, and a scorer needs to tell them apart: `ab::ArmOutcome::scorable`
    /// exists precisely to keep "the harness stopped it" out of the scored
    /// denominator, and the alternative to a typed cause is another compare
    /// against error prose — which this enum was introduced to end.
    ///
    /// It does NOT protect the recurrence/skill machinery, despite the obvious
    /// guess: `record_failure` and `record_recurrence` run at the END of a red
    /// iteration, while admission is denied at the START of the next one, so a
    /// cut-off approach has already been written to skill memory as a failure
    /// before this variant exists. Fixing that is a separate change.
    BudgetExhausted,
}

/// How a loop run ended.
///
/// Build these with [`LoopOutcome::green`] and [`LoopOutcome::lost`] rather than
/// a struct literal. The `failure`-is-`None`-exactly-when-`passed` invariant is
/// documented on the field and held at every site today, but nothing enforced
/// it — a struct literal lets the next site quietly state that a run both passed
/// and failed. The constructors make that unrepresentable.
#[derive(Debug, Clone)]
pub struct LoopOutcome {
    /// Every contract check passed.
    pub passed: bool,
    /// Iterations actually executed.
    pub iterations: u32,
    /// Check results from the final evaluation.
    pub last_results: Vec<CheckResult>,
    /// Human-readable terminal error. `None` on a pass, and on a loss the loop
    /// attributes to the work rather than the machinery — an exhausted budget
    /// whose last run was clean leaves this empty and lets `rpc` render
    /// "contract not satisfied after N iteration(s)".
    ///
    /// It is NOT empty merely because the budget ran out: an exhausted
    /// [`LoopFailure::Infrastructure`] still fills this in, deliberately (see
    /// the scraping note below).
    ///
    /// Human-readable, and load-bearing beyond humans: `car-cli`'s coder-A/B
    /// scrapes this text across a process boundary to split infra failures out
    /// of the scored denominator (`coder_ab::INFRA_MARKERS`). Keep the
    /// `external agent '<id>'` / `foreman adapter '<id>'` prefixes and the bare
    /// `cancelled` spelling stable, or that split breaks silently.
    pub error: Option<String>,
    /// Typed reason the run is not green; `None` exactly when `passed`.
    ///
    /// Exists so callers branch on a value instead of matching on `error`'s
    /// prose — `rpc`'s engine fallback used to be an `e != "cancelled"` string
    /// compare.
    ///
    /// **Descriptive, not prescriptive.** It records what went wrong, not what
    /// to do about it. Retry policy belongs to whichever loop produced the
    /// value and knows its own budgets: `external_loop` retries an
    /// `Infrastructure` failure it has budget for, while the same variant from
    /// `native_loop` means the loop already retried internally and gave up. Do
    /// not write `if failure == Infrastructure { retry }` at an outer level.
    pub failure: Option<LoopFailure>,
    /// Metered inference spend for this run, when anything reported it.
    ///
    /// `None` means **unknown**, not free — the native loop does not meter, and
    /// conflating the two is how `ab::ArmOutcome.cost_usd` came to publish a
    /// structural zero: `cost_per_pass` was computed as `sum / passes` over a
    /// field nothing ever filled, so every external arm reported a cost of
    /// exactly $0.00 as though it were measured.
    pub cost_usd: Option<f64>,
}

impl LoopOutcome {
    /// Attach metered spend. Separate from the constructors so adding cost
    /// accounting did not churn all 24 construction sites, and so a caller with
    /// nothing to report simply never calls it.
    pub fn with_cost(mut self, usd: Option<f64>) -> Self {
        self.cost_usd = usd;
        self
    }

    /// Every check passed. `failure` is `None` by construction.
    pub fn green(iterations: u32, last_results: Vec<CheckResult>) -> Self {
        Self {
            passed: true,
            iterations,
            last_results,
            error: None,
            failure: None,
            cost_usd: None,
        }
    }

    /// The run ended not-green, for `failure`.
    ///
    /// `error` is the human-readable text, and stays `None` when the loss needs
    /// no explanation beyond the checks themselves — `rpc` then renders
    /// "contract not satisfied after N iteration(s)". It is NOT decoration:
    /// `car-cli`'s A/B scrapes it across a process boundary, so the
    /// `external agent '<id>'` prefix and the bare `cancelled` spelling are load-
    /// bearing where they appear.
    pub fn lost(
        failure: LoopFailure,
        error: Option<String>,
        iterations: u32,
        last_results: Vec<CheckResult>,
    ) -> Self {
        Self {
            passed: false,
            iterations,
            last_results,
            error,
            failure: Some(failure),
            cost_usd: None,
        }
    }
}

fn preview(s: &str, max: usize) -> String {
    if s.len() <= max {
        return s.to_string();
    }
    let mut end = max;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}…", &s[..end])
}

/// Render the coder's system prompt, with any evolved overlay appended.
///
/// The base prompt stays in source and the overlay can only *add* a section.
/// That asymmetry is what makes a prompt safe to evolve: an overlay able to
/// replace the prompt could delete the rules the prompt is carrying — the
/// `git commit` prohibition, for one, is held by prose and nothing else — and
/// no regression gate reliably catches a rule that silently stopped being
/// stated. It is rendered last, clearly delimited and explicitly subordinate,
/// so a conflicting instruction reads as an addition to the rules above rather
/// than a replacement for them.
fn system_prompt_with_overlay(
    contract: &OutcomeContract,
    environment: &str,
    overlay: Option<&str>,
) -> String {
    let base = system_prompt(contract, environment);
    match overlay.map(str::trim).filter(|o| !o.is_empty()) {
        None => base,
        Some(overlay) => format!(
            "{base}\n\n\
             ADDITIONAL GUIDANCE (learned from prior sessions; it ADDS to the rules \
             above and never overrides them — if it appears to conflict with anything \
             above, the rules above win):\n{overlay}"
        ),
    }
}

fn system_prompt(contract: &OutcomeContract, environment: &str) -> String {
    format!(
        "You are CAR Coder, an autonomous coding agent working in an isolated git worktree \
         of the user's repository. The worktree root is your working directory; all relative \
         paths resolve against it.\n\n\
         ENVIRONMENT:\n{environment}\n\n\
         How to work:\n\
         - Inspect before you edit. Read the relevant files and search the codebase \
           (grep_files / find_files) to understand the code BEFORE changing it. Never \
           fabricate file contents, symbols, or APIs you have not actually read.\n\
         - Plan briefly, then make surgical edits: prefer edit_file for targeted changes \
           over rewriting a whole file with write_file. Change the minimum the task needs.\n\
         - Trace the checks before you declare done. Read each outcome-contract check and \
           confirm your change actually makes it pass — the exact expected values, and \
           every symbol the check exercises.\n\
         - Verify your own work by running the EXACT command(s) from the OUTCOME CONTRACT \
           below, verbatim — copy the command string character-for-character (same \
           interpreter path, same flags, same scoped test file). Do NOT substitute a \
           broader or 'equivalent' command: running `python -m pytest tests/` when the \
           contract says `/path/to/venv/bin/python -m pytest -q tests/test_x.py` is WRONG \
           — a different interpreter (e.g. a system `python` that is a different version \
           with different installed packages) can fail on environment issues that have \
           nothing to do with your task. Read that command's real output before declaring \
           done; the contract's exact command is the only thing that decides done. Never \
           claim a check passed without having run its exact command this session and seen \
           it pass.\n\
         - The environment is not yours to fix. If the contract's exact command fails on \
           something that is not your code — a version mismatch, a missing package, an \
           import error in an unrelated module, a broken runner — your code fix is already \
           done: write your summary and STOP. The runtime re-runs the contract in the \
           correct environment to decide done, so turns spent making a wrong-environment \
           command pass cannot change the verdict. (Package installs, venv creation, and \
           interpreter shims are denied by policy; you will get a denial with a reason.)\n\
         - If the shell tool is unavailable or a command is blocked this session (e.g. a \
           permission-restricted runner returns an approval error instead of output), that \
           is NOT a task failure and NOT a reason to report the work as blocked or uncertain: \
           the runtime independently runs the outcome contract to decide done. Make your edits \
           correct, note that you could not self-run the checks, and STOP — do not retry the \
           blocked command in a loop.\n\
         - On failure, read the actual error output before retrying — fix the specific \
           cause the compiler or test named; do not guess-and-retry. If the error names a \
           missing symbol, function, or attribute, IMPLEMENT it rather than editing the \
           caller. If the same check fails again after an edit, your hypothesis was wrong: \
           re-read the exact expected-vs-actual and form a different one — do not re-apply a \
           variation of an edit that did not change the failure.\n\n\
         - Do not git commit; the runtime handles version control. (Everything else \
           the policy forbids — push, sudo, destructive operations outside the worktree \
           — comes back as a denial with a reason; don't retry a denied call verbatim.)\n\n\
         When you believe the work is complete, reply with a brief plain-text summary and \
         STOP calling tools. The runtime independently re-runs the outcome contract after \
         you stop — but do not rely on it: verify the checks yourself first, because a red \
         re-invocation costs a full round-trip.\n\n\
         OUTCOME CONTRACT (the runtime runs these to decide done):\n{}",
        contract.render()
    )
}

/// Feedback injected as a new user turn on a red repair round. `recurrences` is
/// how many EARLIER rounds this round's primary [`FailureSignature`] has already
/// been seen in (0 the first time that exact failure appears). It escalates the
/// repair discipline: on a fresh failure, direct the model to read the specific
/// error and fix the named cause; on a recurring one, tell it its approach isn't
/// working so it stops re-applying a variation of a non-converging edit.
/// Surfaced by the coder A/B on hard real-codebase tasks, where the coder edited
/// 20+ times across 6 rounds but never addressed the actual cause (a missing
/// function the test named).
///
/// Keyed on the signature (check + coarse error class), not the bare check name,
/// and counted by total recurrence rather than consecutive streak. Both changes
/// remove a wrong answer:
/// - **Name-only over-fires.** A check going `compile_error` -> `test_failure`
///   is real progress — the code now builds — but the name never changed, so a
///   name-keyed streak told the model its approach was not addressing the cause
///   at the exact moment it was.
/// - **Consecutive-only under-fires.** A coder alternating between two bad
///   fixes (A -> B -> A -> B) resets a streak counter every round and never
///   escalates, which is precisely the non-convergence worth interrupting.
///
/// Known limit: [`super::skill_memory`]'s `classify` names three buckets and
/// collapses everything else to `exit_<code>`, so the gain is toolchain-
/// dependent. Rust and pytest emit the substrings it looks for; jest, `go test`,
/// eslint and mypy mostly do not, and for those the signature degrades to
/// check-name-plus-a-constant — the old key under a new name. Widening
/// `classify` is where further accuracy lives, not here.
fn failure_feedback(results: &[CheckResult], recurrences: u32) -> String {
    let mut msg = String::from(
        "The outcome contract was evaluated and some checks FAILED. Fix the code so they pass.\n\n",
    );
    for r in results.iter().filter(|r| !r.passed) {
        msg.push_str(&format!(
            "FAILED {} (exit {:?}):\n{}\n\n",
            r.name, r.exit_code, r.output_tail
        ));
    }
    if recurrences == 0 {
        msg.push_str(
            "Before editing again: read the SPECIFIC failure above — the exact assertion, error \
             type, or traceback line — and name the single cause. If the error names a missing \
             symbol/function/attribute, implement THAT symbol. Find the code responsible for the \
             named cause and fix it directly; do not guess-and-retry.\n",
        );
    } else {
        msg.push_str(&recurrence_notice(recurrences));
    }
    msg
}

/// The escalation clause, shared by both loops so they cannot drift into making
/// different claims about the same integer.
///
/// Deliberately modest about what it knows. The key is a check name plus one of
/// five coarse buckets from `skill_memory::classify`, and outside rustc/pytest
/// shapes almost everything collapses to `exit_<code>` — so two unrelated
/// failures that both exit 1 share a signature. Claiming "this EXACT failure"
/// would assert precision the key cannot back, to a model that can see the
/// output and check. "The same check has failed the same way" is what the data
/// supports.
///
/// It also must not say "in a row": the count is a session total, so in the
/// oscillation case this mechanism exists to catch (A -> B -> A) the failures
/// demonstrably were not consecutive, and the model has the transcript.
pub(super) fn recurrence_notice(recurrences: u32) -> String {
    format!(
        "The same check has now failed the same way {} times in this session (not necessarily \
         in consecutive rounds) despite your edits — your approach is NOT addressing the real \
         cause, so do NOT re-apply a variation of the same edit. STOP and read the failure \
         literally: what exact value or behavior was EXPECTED vs what was PRODUCED? Trace that \
         exact value back to the specific code that produces it, form a DIFFERENT hypothesis \
         about the named cause, and make one targeted change to it. If the error names a missing \
         symbol/function/attribute, the fix is to IMPLEMENT it, not to adjust the caller.\n",
        recurrences + 1
    )
}

/// How many EARLIER rounds this signature has already appeared in, recording it
/// for the next round. `None` (a green evaluation) is not a recurrence.
///
/// The `None` arm is defensive, not reachable: both callers run only after a red
/// evaluation, so at least one check failed and `primary_failure` returns `Some`.
///
/// A count rather than a bool so the escalated feedback can state the number
/// back to the model — "this has now failed N times" is a materially stronger
/// instruction than "this has failed before".
pub(super) fn record_recurrence(
    seen: &mut HashMap<String, u32>,
    sig: Option<&FailureSignature>,
) -> u32 {
    let Some(sig) = sig else { return 0 };
    let entry = seen.entry(sig.key()).or_insert(0);
    let prior = *entry;
    *entry += 1;
    prior
}

/// The signature of the most-relevant failure in a red evaluation: the first
/// failing check. One signature per repair round keeps learning attributable.
pub(super) fn primary_failure(results: &[CheckResult]) -> Option<FailureSignature> {
    results
        .iter()
        .find(|r| !r.passed)
        .map(FailureSignature::from_check)
}

/// Append a recalled repair hint to the repair prompt. Kept terse and clearly
/// labelled as a heuristic from a prior session so the model treats it as a
/// lead, not gospel.
fn append_recall_hint(prompt: &mut String, hint: &str) {
    prompt.push_str(
        "\nHINT — a prior session resolved this same failure signature with this approach; \
         use it as a lead, verify it still applies:\n",
    );
    prompt.push_str(hint);
    prompt.push('\n');
}

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 {
                    car_inference::ContentBlock::Text { text } => Some(text.trim()),
                    _ => None,
                })
                .filter(|s| !s.is_empty())
                .collect::<Vec<_>>()
                .join("\n");
            (!text.is_empty()).then_some(text)
        }
        _ => None,
    }
}

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

async fn maybe_apply_coder_proactive_memory(
    req: &mut GenerateRequest,
    intent: &str,
    messages: &[Message],
    sink: &EventSink,
    memory: &RepairMemory,
) {
    let mut recent = messages
        .iter()
        .rev()
        .filter_map(message_memory_text)
        .take(6)
        .collect::<Vec<_>>();
    recent.reverse();
    let events = sink.events();
    let Some((maintenance, decision)) = memory.proactive_for_task(intent, recent, &events).await
    else {
        return;
    };
    sink.record_proactive_memory(&maintenance, &decision);
    if let car_memgine::ProactiveMemoryDecision::Inject { reminder, .. } = decision {
        append_context_block(req, "Proactive Memory", &reminder);
    }
}

/// Distill a durable, reusable approach summary from the iteration that turned
/// the contract green. The model's closing plan text (its own summary of what
/// it did) is the best signal; fall back to a generic marker when it stayed
/// silent so the skill still records that *something* fixed this signature.
fn winning_approach(sig: &FailureSignature, plan_text: &str) -> String {
    let plan = plan_text.trim();
    if plan.is_empty() {
        format!(
            "Re-attempted the edit; the '{}' failure of check '{}' cleared after repair.",
            sig.error_class, sig.check
        )
    } else {
        preview(plan, 1024)
    }
}

/// Drive the native loop to completion, cancellation, or exhaustion.
///
/// When `ask` is `Some`, the model is additionally offered the [`ASK_USER_TOOL`]
/// to request mid-session input; the loop routes that one tool to the handler
/// (not the worktree executor) so a question blocks on the user-input gate while
/// honoring cancellation and the gate's timeout.
#[allow(clippy::too_many_arguments)]
pub async fn run_native_loop(
    inference: &dyn TurnGenerator,
    executor: &WorktreeExecutor,
    intent: &str,
    contract: &OutcomeContract,
    sink: &EventSink,
    cancel: &CancelFlag,
    cfg: &NativeLoopConfig,
    memory: &RepairMemory,
    ask: Option<&dyn AskUser>,
) -> LoopOutcome {
    let mut tools = WorktreeExecutor::tool_defs();
    if ask.is_some() {
        tools.push(ask_user_tool_def());
    }
    // ENVIRONMENT section (F7/L1): reuse the same cheap names-only repo summary
    // the contract-derivation prompt uses, so the coding loop and contract
    // derivation describe the repo identically. Local worktree by construction.
    let environment = super::rpc::summarize_repo(executor.worktree());
    let system = system_prompt_with_overlay(contract, &environment, cfg.prompt_overlay.as_deref());
    let mut feedback: Option<String> = None;
    let mut last_results: Vec<CheckResult> = Vec::new();
    let mut consecutive_inference_failures = 0u32;
    // Consecutive repair iterations that ended in a no-progress thrash. The
    // conversation persists across rounds, so a genuinely wedged backbone
    // re-thrashes identically; abort on the second such iteration.
    let mut no_progress_iterations = 0u32;
    // The primary failing check from the previous red round + how many rounds it
    // has recurred, so the repair feedback escalates when the coder's edits keep
    // failing the SAME check (not converging on hard tasks).
    // Every failure signature seen so far this session, with how many rounds it
    // has appeared in. Session-scoped rather than consecutive so an oscillating
    // coder still trips the escalation.
    let mut seen_sigs: HashMap<String, u32> = HashMap::new();
    // This loop's clock starts here, before the first iteration.

    // The signature of the failure carried into THIS iteration's repair, if
    // any. Drives skill recall (inject a prior fix) and outcome attribution
    // (credit/penalize that signature's skill once we know if the repair held).
    let mut prior_sig: Option<FailureSignature> = None;

    // Persist ONE conversation across repair iterations (F2, audit 2026-07-06):
    // the model keeps the files it read, the edits it made, and the dead-ends it
    // already ruled out. A failed contract appends its feedback as a new user
    // turn on this thread rather than resetting to [System, User] every round —
    // so each repair builds on accumulated understanding instead of re-deriving
    // the repo from scratch. The thread is bounded to the model's context window
    // each turn (`compact_history_to_window` below) so a long repair session
    // can't overflow it.
    // Session-start recall (F8-lite/L3): a ONE-TIME, task-scoped heuristic lead
    // from prior sessions, computed once (one short lock) and injected into the
    // persistent first user turn. Distinct from the per-repair-round recall hint
    // below, which is failure-signature-scoped. `None` when learning is disabled
    // or nothing overlaps the intent → nothing is injected (no empty section).
    let mut initial_user = format!("Task:\n{intent}\n");
    if let Some(block) = memory.recall_for_task(intent).await {
        initial_user.push_str(
            "\nRecall from prior sessions (heuristic — verify against the repo \
             before acting on it):\n",
        );
        initial_user.push_str(&block);
    }
    let mut messages = vec![
        Message::System {
            content: system.clone(),
        },
        Message::User {
            content: initial_user,
        },
    ];

    // The model's context window, resolved once (the model is fixed for the
    // run). Bounds the persistent conversation each turn; 0 (unknown — e.g. a
    // local/test generator) disables compaction, so behavior is unchanged there.
    let context_window = cfg
        .model
        .as_deref()
        .map(|m| inference.context_window(m))
        .unwrap_or(0);

    for iteration in 1..=cfg.max_iterations {
        if cancel.load(Ordering::SeqCst) {
            return LoopOutcome::lost(
                LoopFailure::Cancelled,
                Some("cancelled".into()),
                iteration - 1,
                last_results,
            );
        }
        // Admission, not interruption: the previous iteration already evaluated
        // the contract, so a denial here cannot be hiding a green result — the
        // loop would have returned before asking.
        if let Some(reason) = cfg.deadline.admit() {
            sink.emit(CoderEventKind::BudgetExhausted {
                reason: reason.clone(),
                elapsed_secs: cfg.deadline.elapsed_secs(),
                iterations: iteration - 1,
            });
            return LoopOutcome::lost(
                LoopFailure::BudgetExhausted,
                Some(reason),
                iteration - 1,
                last_results,
            );
        }
        sink.emit(CoderEventKind::IterationStarted {
            n: iteration,
            max: cfg.max_iterations,
        });

        // On a repair iteration, append the failing-check feedback (plus any
        // recalled prior-session fix for this failure signature) as a NEW user
        // turn on the persistent thread. Iteration 1 has no feedback — the task
        // seed above is the only user turn.
        if let Some(fb) = &feedback {
            let mut user = fb.clone();
            // Durable recall: if a prior session learned a fix for this failure
            // signature, inject it as a lead. No-op when learning is disabled or
            // nothing matches.
            if let Some(sig) = &prior_sig {
                if let Some(hint) = memory.recall(sig).await {
                    append_recall_hint(&mut user, &hint);
                }
            }
            messages.push(Message::User { content: user });
        }

        // The model's closing summary for this iteration — captured to distill
        // a durable repair skill if this iteration turns the contract green.
        let mut closing_plan = String::new();
        // Inner tool-use conversation.
        let mut turn = 0;
        // No-progress guard (per iteration): counts identical READ-ONLY calls
        // since the last mutating call, so a thrashing model (re-reading the same
        // thing with no edits) is caught. Cleared by any mutating call (progress)
        // and reset per iteration. `no_progress_this_iteration` records a trip so
        // the turn loop breaks to contract evaluation.
        let mut identical_read_calls: std::collections::HashMap<(String, String), u32> =
            std::collections::HashMap::new();
        let mut no_progress_this_iteration = false;
        // Journal the iteration's terminal decision for the harness miners: the
        // model declaring done (empty tool calls — possibly a truncated/starved
        // turn on a local model) vs. exhausting the per-iteration turn budget
        // (the coder-path "never finishes" signal). `last_model` attributes it.
        let mut last_model = String::new();
        let mut model_declared_done = false;
        while turn < cfg.max_turns_per_iteration {
            turn += 1;
            if cancel.load(Ordering::SeqCst) {
                return LoopOutcome::lost(
                    LoopFailure::Cancelled,
                    Some("cancelled".into()),
                    iteration,
                    last_results,
                );
            }

            // Bound the persistent conversation to the model's context window
            // before each generate so a long repair session never overflows it —
            // an overflow head-truncates the System prompt (outcome contract) +
            // task provider-side on a small local model, the exact silent failure
            // this loop must avoid. Pins System + task, drops oldest middle turns
            // on a valid turn boundary; no-op when the window is unknown or fits.
            compact_history_to_window(&mut messages, context_window);

            let mut req = GenerateRequest {
                prompt: intent.to_string(), // ignored when messages are set
                model: cfg.model.clone(),
                params: GenerateParams {
                    temperature: 0.0,
                    max_tokens: cfg.max_tokens_per_turn,
                    // A pinned backbone (e.g. `--model parslee/reasoning` for an
                    // A/B) must NOT silently degrade to a local model on a remote
                    // outage — that manufactures fake results. Fail loudly so the
                    // run is marked infra, not scored on the wrong model. Adaptive
                    // routing (model = None) keeps the resilient degrade behavior.
                    strict_model: cfg.model.is_some(),
                    ..Default::default()
                },
                tools: Some(tools.clone()),
                messages: Some(messages.clone()),
                intent: Some(car_inference::IntentHint {
                    task: Some(car_inference::TaskHint::Code),
                    // Stakes-aware routing: this loop edits and runs code in a
                    // real git worktree and can drive a merge — structurally
                    // irreversible, so it is UNCONDITIONALLY high-stakes (unlike
                    // a general planner, the coder can never be benign — its
                    // tools are write_file/run_command). Generate with the best
                    // model; cost is the wrong axis when a wrong edit lands in a
                    // real repo. No-op when `cfg.model` pins an explicit model
                    // (the router consults intent only on the unpinned arm).
                    high_stakes: true,
                    ..Default::default()
                }),
                ..Default::default()
            };
            maybe_apply_coder_proactive_memory(&mut req, intent, &messages, sink, memory).await;

            let result = match inference.generate(req).await {
                Ok(r) => {
                    consecutive_inference_failures = 0;
                    r
                }
                Err(e) => {
                    // Blocked on the HUMAN, not on the machinery. Ask and wait
                    // rather than burning a strike: retrying an expired
                    // credential three times in as many seconds cannot succeed,
                    // and the third strike throws away the worktree.
                    let message = e.to_string();
                    if is_auth_failure(&message) {
                        if let Some(gate) = cfg.auth_gate.clone() {
                            sink.emit(CoderEventKind::AuthRequired {
                                message: message.clone(),
                                wait_secs: cfg.auth_wait.as_secs(),
                            });
                            if wait_for_auth(gate.as_ref(), cfg.auth_wait, cancel, &cfg.deadline)
                                .await
                            {
                                // Recovered: this was never a failure of the
                                // work, so it must not count toward the strike
                                // budget either.
                                consecutive_inference_failures = 0;
                                continue;
                            }
                            let results = evaluate_contract(contract, executor, sink).await;
                            let passed = results.iter().all(|r| r.passed);
                            return if passed {
                                LoopOutcome::green(iteration, results)
                            } else {
                                LoopOutcome::lost(
                                    LoopFailure::NeedsAuth,
                                    Some(format!(
                                        "not signed in, and no credential appeared within {}s: {message}",
                                        cfg.auth_wait.as_secs()
                                    )),
                                    iteration,
                                    results,
                                )
                            };
                        }
                    }
                    consecutive_inference_failures += 1;
                    sink.emit(CoderEventKind::Error {
                        message: format!("inference failed (turn {turn}): {e}"),
                    });
                    if consecutive_inference_failures >= 3 {
                        // Ask the worktree before declaring a loss. The model
                        // may have landed every edit the contract wants and
                        // then had its backbone die; returning red without
                        // evaluating would let the inference transport
                        // pronounce a verdict it has no standing to give —
                        // the same defect `external_loop` carried until the
                        // classification moved after evaluation.
                        let results = evaluate_contract(contract, executor, sink).await;
                        let passed = results.iter().all(|r| r.passed);
                        // Same green, same credit. This is a second path that
                        // returns `passed: true`, so it owes memgine the skill
                        // record the ordinary green path makes below —
                        // otherwise a repair that held would be forgotten
                        // purely because the backbone died on the way out.
                        if passed {
                            if let Some(sig) = &prior_sig {
                                memory
                                    .record_success(sig, &winning_approach(sig, &closing_plan))
                                    .await;
                            }
                        }
                        // Branch rather than conditional fields: a run cannot
                        // both pass and carry a failure, and the constructors
                        // are what make that unrepresentable.
                        return if passed {
                            LoopOutcome::green(iteration, results)
                        } else {
                            // The backbone, not the code under test: the native
                            // loop's analogue of a dead transport.
                            LoopOutcome::lost(
                                LoopFailure::Infrastructure,
                                Some(format!("inference failed repeatedly: {e}")),
                                iteration,
                                results,
                            )
                        };
                    }
                    continue; // retry the same turn
                }
            };
            last_model = result.model_used.clone();

            if result.tool_calls.is_empty() {
                // A turn that hit the max_tokens ceiling is CUT OFF, not a
                // completion — treating it as "done" silently accepts a
                // truncated answer (and, on a truncated tool call, would drop
                // the call entirely). Recognize the truncation and continue so
                // the model can finish where it left off. Bounded by
                // max_turns_per_iteration, so this cannot spin forever — and if
                // the model truncates every turn, the turn-budget terminal
                // below journals that exhaustion as the mineable failure.
                // (F3/F11, audit 2026-07-06.)
                if result.was_truncated() {
                    sink.emit(CoderEventKind::Error {
                        message: format!(
                            "model turn truncated (stop_reason={:?}) — continuing so it can finish",
                            result.stop_reason
                        ),
                    });
                    result.append_assistant_history(&mut messages, vec![]);
                    messages.push(Message::User {
                        content: "Your previous response was cut off at the token limit. \
                                  Continue exactly where you left off; if you were in the \
                                  middle of a tool call, re-issue that call in full."
                            .to_string(),
                    });
                    continue;
                }
                // Model says done — break to contract evaluation. Journal the
                // terminal so the finish is mineable; with the truncation guard
                // above, reaching here means a genuine (non-truncated) finish
                // (docs/audits/car-tracing-design-2026-07-07, native_loop:344).
                sink.record_turn_completed(
                    "empty_tool_calls",
                    result.stop_reason.as_deref(),
                    result.was_truncated(),
                    turn,
                    &result.model_used,
                );
                model_declared_done = true;
                if !result.text.trim().is_empty() {
                    closing_plan = result.text.clone();
                    sink.emit(CoderEventKind::PlanText {
                        text: result.text.clone(),
                    });
                }
                break;
            }

            // Assign ids so ToolResult replies correlate (local models may
            // omit them), then execute sequentially in emitted order.
            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_{iteration}_{turn}_{i}"));
                }
            }
            result.append_assistant_history(&mut messages, calls.clone());

            for call in &calls {
                let params = Value::Object(call.arguments.clone().into_iter().collect());
                sink.emit(CoderEventKind::ToolCall {
                    tool: call.name.clone(),
                    params_preview: preview(&params.to_string(), 400),
                });
                // No-progress guard. A repeated read-only call (same tool+args)
                // with no editing/mutating call in between means the model is
                // thrashing — re-reading the same thing every turn without acting
                // on it (the signature of a backbone that never returns its tool
                // results). Any mutating call (edit/write/shell/…) is *progress*
                // and clears the tracker, so a legitimate read→edit→re-read or a
                // build→edit→build loop never trips. On a trip we DON'T fail the
                // session outright — we break to contract evaluation so earlier
                // edits that already satisfied the contract still pass; only a
                // second no-progress iteration (a genuinely wedged backbone, since
                // the conversation persists across repair rounds) aborts.
                if is_read_only_tool(&call.name) {
                    let c = identical_read_calls
                        .entry((call.name.clone(), params.to_string()))
                        .or_insert(0);
                    *c += 1;
                    if *c >= NO_PROGRESS_REPEAT_LIMIT && !no_progress_this_iteration {
                        no_progress_this_iteration = true;
                        sink.emit(CoderEventKind::Error {
                            message: format!(
                                "no-progress loop: `{}` called {c} times with identical arguments \
                                 and no intervening edit — ending this attempt",
                                call.name
                            ),
                        });
                    }
                } else {
                    // A mutating/acting call — progress. Forget the read history.
                    identical_read_calls.clear();
                }
                let (ok, content) = if call.name == ASK_USER_TOOL {
                    // Route to the user-input handler, not the worktree executor.
                    // The handler emits UserInputRequested, blocks on the gate
                    // (honoring cancel + timeout), and returns the user's text or
                    // an error the model can recover from.
                    match ask {
                        Some(asker) => {
                            let prompt = params
                                .get("prompt")
                                .and_then(Value::as_str)
                                .unwrap_or("")
                                .to_string();
                            match asker.ask(&prompt).await {
                                Ok(answer) => (true, answer),
                                Err(e) => (false, format!("ERROR: {e}")),
                            }
                        }
                        None => (
                            false,
                            "ERROR: ask_user is not available in this session".to_string(),
                        ),
                    }
                } else {
                    match executor.execute(&call.name, &params).await {
                        Ok(v) => (true, v.to_string()),
                        Err(e) => (false, format!("ERROR: {e}")),
                    }
                };
                sink.emit(CoderEventKind::ToolResult {
                    tool: call.name.clone(),
                    ok,
                    preview: preview(&content, 400),
                });
                messages.push(Message::ToolResult {
                    tool_use_id: call.id.clone().expect("assigned above"),
                    content: preview(&content, 16 * 1024),
                    // The coder's tools are local: shell, file read/write, git. None
                    // reach the network, so nothing here crosses the trust boundary.
                    provenance: Provenance::Internal,
                });
            }
            // A no-progress trip ends this attempt: stop taking turns and let the
            // contract decide (earlier edits may already have satisfied it).
            if no_progress_this_iteration {
                break;
            }
        }
        // Journal the inner loop's terminal (the contract, evaluated next, still
        // owns the pass/fail decision): a no-progress bail-out, or the "never
        // finishes" turn-budget exhaustion, vs. a genuine model-declared done.
        if no_progress_this_iteration {
            sink.record_turn_completed("no_progress_loop", None, false, turn, &last_model);
        } else if !model_declared_done {
            sink.record_turn_completed("max_turns", None, false, turn, &last_model);
        }

        // Verify. The contract — not the model's self-report — decides.
        last_results = evaluate_contract(contract, executor, sink).await;
        if last_results.iter().all(|r| r.passed) {
            // Durable learning: if THIS green came after a prior failure, the
            // repair held — credit (or ingest) the skill for that signature so
            // the next occurrence can recall the winning approach.
            if let Some(sig) = &prior_sig {
                memory
                    .record_success(sig, &winning_approach(sig, &closing_plan))
                    .await;
            }
            return LoopOutcome::green(iteration, last_results);
        }
        // Still red. If this iteration made no progress (a read-only thrash) and
        // the contract didn't pass, count it — a second consecutive no-progress
        // iteration means the backbone is wedged (it re-thrashes the persistent
        // conversation identically), so abort rather than burn every iteration.
        // A productive iteration resets the count.
        if no_progress_this_iteration {
            no_progress_iterations += 1;
            if no_progress_iterations >= 2 {
                // NOT `Infrastructure`, despite the error text's guess at a
                // backbone that drops tool results. A model that re-reads
                // without editing is just as plausibly a bad model, and the
                // guard cannot tell them apart. If this value ever reaches the
                // A/B denominator, calling it infra would launder a model
                // failure into an exclusion and inflate the pass rate — bias in
                // the direction that flatters us, which is the one to refuse.
                return LoopOutcome::lost(
                    LoopFailure::Verification,
                    Some(
                        "no-progress loop: the model repeatedly re-read the same files without \
                         making edits across two attempts — the backbone is likely not returning \
                         tool results. Aborted before exhausting the iteration budget."
                            .to_string(),
                    ),
                    iteration,
                    last_results,
                );
            }
        } else {
            no_progress_iterations = 0;
        }
        // Track whether the SAME failure keeps recurring, so the repair feedback
        // escalates from "read the error" to "your approach isn't working — form
        // a different hypothesis" when the coder isn't converging. Keyed on the
        // signature, and counted across the whole session; see
        // [`failure_feedback`] for why neither the check name nor a consecutive
        // streak was the right key.
        //
        // A no-progress iteration is excluded, for the same reason the external
        // loop excludes a cut-short attempt: the model made zero edits, so the
        // signature is trivially identical to last round's and carries no
        // information about the hypothesis. Counting it would guarantee a
        // recurrence and then tell the model "your approach is NOT addressing
        // the real cause" when the true diagnosis is "you did not edit
        // anything" — a different situation with a different fix.
        let cur_sig = primary_failure(&last_results);
        let recurrences = if no_progress_this_iteration {
            0
        } else {
            record_recurrence(&mut seen_sigs, cur_sig.as_ref())
        };

        // Record the failure against its signature (penalizing any recalled
        // approach that didn't hold) and carry it into the next repair round for
        // recall + attribution.
        feedback = Some(failure_feedback(&last_results, recurrences));
        if let Some(sig) = cur_sig {
            memory.record_failure(&sig).await;
            prior_sig = Some(sig);
        } else {
            prior_sig = None;
        }
    }

    // Clean exhaustion of the iteration budget after real attempts: the
    // machinery worked, the hypotheses did not.
    LoopOutcome::lost(
        LoopFailure::Verification,
        None,
        cfg.max_iterations,
        last_results,
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::coder::contract::ContractCheck;
    use std::sync::atomic::AtomicUsize;
    use std::sync::Arc;

    fn contract_for_prompt_test() -> OutcomeContract {
        OutcomeContract {
            description: "d".into(),
            checks: vec![],
        }
    }

    /// car#708: an overlay must reach the prompt, or the evolution loop still
    /// has nowhere to put a prompt change.
    #[test]
    fn an_overlay_is_appended_to_the_prompt() {
        let contract = contract_for_prompt_test();
        let base = system_prompt_with_overlay(&contract, "env", None);
        let with = system_prompt_with_overlay(&contract, "env", Some("Prefer smaller diffs."));

        assert!(with.contains("Prefer smaller diffs."));
        assert!(
            with.starts_with(&base),
            "the overlay must be strictly additive — the base prompt has to survive verbatim"
        );
        assert!(with.len() > base.len());
    }

    /// No overlay must be byte-identical to the prompt before overlays existed.
    #[test]
    fn no_overlay_changes_nothing() {
        let contract = contract_for_prompt_test();
        let base = system_prompt(&contract, "env");
        assert_eq!(system_prompt_with_overlay(&contract, "env", None), base);
        assert_eq!(system_prompt_with_overlay(&contract, "env", Some("")), base);
        assert_eq!(
            system_prompt_with_overlay(&contract, "env", Some("   \n ")),
            base,
            "whitespace is not an overlay"
        );
    }

    /// The overlay is rendered last and explicitly subordinate, so a
    /// conflicting instruction reads as an addition rather than a replacement.
    /// This is the only thing standing between an evolved prompt and the rules
    /// the base prompt carries in prose.
    #[test]
    fn the_overlay_is_marked_subordinate_to_the_base_rules() {
        let contract = contract_for_prompt_test();
        let with = system_prompt_with_overlay(&contract, "env", Some("Commit when done."));
        let marker = with
            .find("ADDITIONAL GUIDANCE")
            .expect("the overlay must be delimited, not silently concatenated");
        assert!(
            with[marker..].contains("the rules above win"),
            "a conflicting overlay instruction must not read as authoritative"
        );
        assert!(
            with.find("Commit when done.").unwrap() > marker,
            "the overlay must come after its own header"
        );
    }

    /// The overlay follows the harness config, and unlike the budgets it is not
    /// a max() — clearing it is how a rollback takes effect.
    #[test]
    fn merge_harness_adopts_and_clears_the_overlay() {
        let mut cfg = NativeLoopConfig::default();
        cfg.merge_harness(&car_memgine::HarnessConfig {
            prompt_overlay: Some("evolved guidance".into()),
            ..Default::default()
        });
        assert_eq!(cfg.prompt_overlay.as_deref(), Some("evolved guidance"));

        cfg.merge_harness(&car_memgine::HarnessConfig {
            prompt_overlay: None,
            ..Default::default()
        });
        assert_eq!(
            cfg.prompt_overlay, None,
            "a rollback must actually remove the overlay, not leave it latched"
        );
    }

    #[test]
    fn merge_harness_raises_coder_budgets_only_upward() {
        // A harness patch that raised planning_max_replans should raise the
        // coder's repair rounds; a lower/default knob must never lower them.
        let mut cfg = NativeLoopConfig {
            max_iterations: 8,
            max_turns_per_iteration: 24,
            ..Default::default()
        };
        cfg.merge_harness(&car_memgine::HarnessConfig {
            prompt_overlay: None,
            max_retries: 30,
            retry_backoff_ms: 0,
            planning_max_replans: 12, // > base 8 → raises to 13
        });
        assert_eq!(
            cfg.max_iterations, 13,
            "planning_max_replans+1 reaches the coder"
        );
        assert_eq!(
            cfg.max_turns_per_iteration, 30,
            "max_retries raises the turn floor"
        );

        // A default (small) harness config never starves the coder below base.
        let mut base = NativeLoopConfig {
            max_iterations: 8,
            max_turns_per_iteration: 24,
            ..Default::default()
        };
        base.merge_harness(&car_memgine::HarnessConfig::default()); // {3,0,2}
        assert_eq!(base.max_iterations, 8, "never lowered below base");
        assert_eq!(base.max_turns_per_iteration, 24);
    }

    /// Scripted generator: pops pre-canned turns in order.
    struct Script {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
        /// Every request the loop sent, in order. The repair feedback is only
        /// observable here — without it, nothing verifies that the escalation
        /// text is ever actually DELIVERED to the model, and a loop that passed
        /// a constant `0` would keep every helper test green.
        seen: std::sync::Mutex<Vec<GenerateRequest>>,
    }

    impl Script {
        fn new(turns: Vec<InferenceResult>) -> Self {
            Self {
                turns,
                cursor: AtomicUsize::new(0),
                seen: std::sync::Mutex::new(Vec::new()),
            }
        }
        /// Concatenated text of every message in the n-th request.
        fn prompt(&self, n: usize) -> String {
            let reqs = self.seen.lock().expect("seen poisoned");
            serde_json::to_string(&reqs[n].messages).unwrap_or_default()
        }
        fn prompts(&self) -> usize {
            self.seen.lock().expect("seen poisoned").len()
        }
    }

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

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

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

    // --- LoopOutcome invariant --------------------------------------------

    /// The invariant the constructors exist to enforce: `failure` is `None`
    /// exactly when `passed`. It held at all 24 literal sites by discipline
    /// alone, and discipline is not a mechanism.
    #[test]
    fn the_constructors_cannot_produce_a_passed_run_that_also_failed() {
        let green = LoopOutcome::green(3, Vec::new());
        assert!(green.passed);
        assert_eq!(green.failure, None);
        assert_eq!(green.error, None);

        let lost = LoopOutcome::lost(LoopFailure::Verification, None, 3, Vec::new());
        assert!(!lost.passed);
        assert_eq!(lost.failure, Some(LoopFailure::Verification));

        // A loss may carry explanatory text, and some of that text is a
        // cross-process contract (`car-cli`'s A/B scrapes it).
        let scraped = LoopOutcome::lost(
            LoopFailure::EngineUnavailable,
            Some("external agent 'codex' failed: no binary".into()),
            0,
            Vec::new(),
        );
        assert!(scraped.error.unwrap().starts_with("external agent '"));
    }

    // --- Repair stagnation, keyed on the failure signature ----------------

    fn failed(name: &str, exit: i64, tail: &str) -> CheckResult {
        CheckResult {
            name: name.into(),
            passed: false,
            exit_code: Some(exit),
            output_tail: tail.into(),
            duration_ms: 1,
        }
    }

    /// The false positive the signature key exists to remove. The check name
    /// never changes, but the error class does — the code went from not
    /// compiling to compiling-and-failing-a-test, which is progress. A
    /// name-keyed streak called that stagnation and told the model to abandon
    /// an approach that was working.
    #[test]
    fn a_changed_error_class_under_one_check_name_is_not_a_recurrence() {
        let mut seen = HashMap::new();
        let compile = primary_failure(&[failed("tests", 101, "error[E0433]: failed to resolve")]);
        let assertion = primary_failure(&[failed("tests", 1, "assertion `left == right` failed")]);
        assert_ne!(
            compile.as_ref().map(|s| s.key()),
            assertion.as_ref().map(|s| s.key()),
            "same check, different error class must be different signatures"
        );
        assert_eq!(record_recurrence(&mut seen, compile.as_ref()), 0);
        assert_eq!(
            record_recurrence(&mut seen, assertion.as_ref()),
            0,
            "progress must not read as a recurrence"
        );
    }

    /// The identical failure twice IS a recurrence, and the count is what the
    /// escalated feedback states back to the model.
    #[test]
    fn the_identical_failure_recurs_and_counts_up() {
        let mut seen = HashMap::new();
        let sig = primary_failure(&[failed("tests", 1, "assertion `left == right` failed")]);
        assert_eq!(record_recurrence(&mut seen, sig.as_ref()), 0);
        assert_eq!(record_recurrence(&mut seen, sig.as_ref()), 1);
        assert_eq!(record_recurrence(&mut seen, sig.as_ref()), 2);
    }

    /// The under-fire a consecutive-streak counter has: a coder alternating
    /// between two bad fixes resets a streak every round and never escalates,
    /// though it is exactly the non-convergence worth interrupting.
    #[test]
    fn an_oscillating_failure_still_recurs() {
        let mut seen = HashMap::new();
        let a = primary_failure(&[failed("tests", 1, "assertion `left == right` failed")]);
        let b = primary_failure(&[failed("build", 101, "error[E0433]: failed to resolve")]);
        assert_eq!(record_recurrence(&mut seen, a.as_ref()), 0);
        assert_eq!(record_recurrence(&mut seen, b.as_ref()), 0);
        assert_eq!(
            record_recurrence(&mut seen, a.as_ref()),
            1,
            "A -> B -> A is going in circles, not progress"
        );
    }

    /// A green evaluation has no primary failure and must not be recorded.
    #[test]
    fn a_green_evaluation_is_not_a_recurrence() {
        let mut seen = HashMap::new();
        assert_eq!(record_recurrence(&mut seen, None), 0);
        assert!(seen.is_empty());
    }

    /// The native loop's admission gate. Two loops implement this and only the
    /// external one was covered — and this is the one where an iteration has no
    /// wall bound of its own, so it is the one that can overrun furthest.
    #[tokio::test]
    async fn an_exhausted_budget_denies_admission_before_any_turn() {
        let script = Script::new(vec![turn("should never run", serde_json::json!([]))]);
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = Arc::new(EventSink::test_sink());
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let cfg = NativeLoopConfig {
            // A deadline that is already spent.
            deadline: std::sync::Arc::new(SessionDeadline::new(Some(0))),
            ..Default::default()
        };
        let outcome = run_native_loop(
            &script,
            &executor,
            "x",
            &OutcomeContract {
                description: "x".into(),
                checks: vec![ContractCheck {
                    name: "gate".into(),
                    command: "exit 1".into(),
                    expect_exit_zero: true,
                    output_contains: None,
                    timeout_secs: 10,
                }],
            },
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert_eq!(
            script.prompts(),
            0,
            "the budget gates before any model turn"
        );
        assert_eq!(outcome.failure, Some(LoopFailure::BudgetExhausted));
        assert_eq!(outcome.iterations, 0);
        assert!(outcome
            .error
            .expect("the reason must surface")
            .contains("budget exhausted"));
    }

    /// **The loop-level test.** Everything above exercises `record_recurrence`
    /// and `failure_feedback` in isolation, which would all still pass if
    /// `run_native_loop` threaded a constant `0` or if `seen_sigs` were moved
    /// inside the iteration loop and reset every round. This runs a real
    /// session against a stably-failing check and asserts the escalation
    /// reaches the model — in the round it should, and not before.
    #[tokio::test]
    async fn the_escalation_is_delivered_to_the_model_only_after_a_repeat() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = Arc::new(EventSink::test_sink());
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // Three iterations that each edit (so the no-progress guard stays out
        // of it) against a check that never goes green.
        let edit = |n: u32| {
            turn(
                "editing",
                serde_json::json!([{
                    "id": format!("c{n}"),
                    "name": "write_file",
                    "arguments": {"path": format!("f{n}.txt"), "content": "x"}
                }]),
            )
        };
        let script = Script::new(vec![
            edit(1),
            turn("done", serde_json::json!([])),
            edit(2),
            turn("done", serde_json::json!([])),
            edit(3),
            turn("done", serde_json::json!([])),
        ]);
        let contract = OutcomeContract {
            description: "never green".into(),
            checks: vec![ContractCheck {
                name: "gate".into(),
                command: "exit 1".into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        let cfg = NativeLoopConfig {
            max_iterations: 3,
            ..Default::default()
        };
        let outcome = run_native_loop(
            &script,
            &executor,
            "x",
            &contract,
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(!outcome.passed);

        // Iteration 1's prompt cannot contain it — nothing has repeated.
        assert!(
            !script.prompt(0).contains("failed the same way"),
            "escalated before anything repeated"
        );
        // By the last prompt the identical signature has recurred, so the
        // escalation must have been threaded through and delivered.
        let last = script.prompt(script.prompts() - 1);
        assert!(
            last.contains("failed the same way"),
            "the escalation never reached the model: {last}"
        );
    }

    /// A generator that fails every turn, so the loop hits its
    /// three-consecutive-inference-failures bail-out.
    fn dead_backbone() -> Script {
        Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        }
    }

    async fn run_against(script: &Script, check: &str) -> LoopOutcome {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = Arc::new(EventSink::test_sink());
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let contract = OutcomeContract {
            description: "x".into(),
            checks: vec![ContractCheck {
                name: "gate".into(),
                command: check.into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        run_native_loop(
            script,
            &executor,
            "x",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await
    }

    /// A backbone that reports "not signed in" for its first `n` calls, then
    /// behaves normally — the shape of a token lapsing mid-session.
    struct AuthFlaky {
        remaining: AtomicUsize,
        inner: Script,
    }

    #[async_trait]
    impl TurnGenerator for AuthFlaky {
        async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
            if self.remaining.load(Ordering::SeqCst) > 0 {
                self.remaining.fetch_sub(1, Ordering::SeqCst);
                return Err("no credential for proprietary provider 'parslee': \
                            set $PARSLEE_ACCESS_TOKEN or run `car auth login parslee`"
                    .to_string());
            }
            self.inner.generate(req).await
        }
    }

    /// Signs in on the first poll.
    #[derive(Debug)]
    struct SignsIn;
    #[async_trait]
    impl AuthGate for SignsIn {
        async fn is_authenticated(&self) -> bool {
            true
        }
    }

    /// Nobody ever signs in.
    #[derive(Debug)]
    struct NeverSignsIn;
    #[async_trait]
    impl AuthGate for NeverSignsIn {
        async fn is_authenticated(&self) -> bool {
            false
        }
    }

    async fn run_with_auth(
        gen: &dyn TurnGenerator,
        check: &str,
        gate: Arc<dyn AuthGate>,
        auth_wait: std::time::Duration,
    ) -> LoopOutcome {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = Arc::new(EventSink::test_sink());
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let contract = OutcomeContract {
            description: "x".into(),
            checks: vec![ContractCheck {
                name: "gate".into(),
                command: check.into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        let cfg = NativeLoopConfig {
            auth_gate: Some(gate),
            auth_wait,
            ..Default::default()
        };
        run_native_loop(
            gen,
            &executor,
            "x",
            &contract,
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await
    }

    /// The point of the whole mechanism: a token that lapses mid-run must PAUSE
    /// the session, not end it. Previously three auth failures in as many
    /// seconds burned the strike budget and returned `Infrastructure`,
    /// discarding a worktree of real edits — a 29-minute session thrown away
    /// because a 15-minute token expired while the operator sat at the machine.
    #[tokio::test]
    async fn a_lapsed_credential_waits_for_sign_in_and_then_resumes() {
        let gen = AuthFlaky {
            // More than the 3-strike budget: if these counted as strikes the
            // run would be dead well before the script is reached.
            remaining: AtomicUsize::new(5),
            inner: Script::new(vec![turn("done", serde_json::json!([]))]),
        };
        let outcome = run_with_auth(
            &gen,
            "exit 0",
            Arc::new(SignsIn),
            std::time::Duration::from_secs(5),
        )
        .await;

        assert!(
            outcome.passed,
            "the session must resume after sign-in, not die: {:?}",
            outcome.error
        );
        assert_eq!(outcome.failure, None);
    }

    /// When nobody signs in, the run still ends — but as `NeedsAuth`, not
    /// `Infrastructure`. The distinction is the actionable part: one says "a
    /// person can fix this in seconds", the other says "the backbone is down".
    #[tokio::test]
    async fn nobody_signs_in_reports_needs_auth_not_infrastructure() {
        let gen = AuthFlaky {
            remaining: AtomicUsize::new(99),
            inner: Script::new(vec![turn("done", serde_json::json!([]))]),
        };
        let outcome = run_with_auth(
            &gen,
            "exit 1",
            Arc::new(NeverSignsIn),
            std::time::Duration::ZERO,
        )
        .await;

        assert!(!outcome.passed);
        assert_eq!(
            outcome.failure,
            Some(LoopFailure::NeedsAuth),
            "an unanswered sign-in must not masquerade as an outage"
        );
    }

    /// The credential error grew a detail suffix (car#797) naming WHICH failure
    /// it is — expired, unreadable, signed out. The historical prefix must
    /// survive that: this classifier keys on it as a substring, and so does the
    /// coder-ab harness's INFRA_MARKERS. Rewording the opening would silently
    /// reclassify auth failures as ordinary execution errors and take the
    /// wait-for-sign-in path down with them.
    #[test]
    fn enriched_credential_errors_still_classify_as_auth_failures() {
        for msg in [
            "no credential for proprietary provider 'parslee' (model parslee/reasoning): the \
             Parslee token expired at unix 1234 and could not be refreshed. Re-authenticate \
             with `car auth login`",
            "no credential for proprietary provider 'parslee' (model parslee/reasoning): the \
             credential store could not be read (code=152). This is not a sign-out",
            "no credential for proprietary provider 'parslee' (model parslee/reasoning): no \
             account is signed in. Run `car auth login`",
        ] {
            assert!(
                is_auth_failure(msg),
                "enriched credential error must still read as an auth failure: {msg}"
            );
        }
    }

    /// The classifier must not swallow a genuine outage: waiting for a human to
    /// repair a dead datacenter would hang a session that should fail.
    #[test]
    fn auth_failures_are_distinguished_from_outages() {
        assert!(is_auth_failure(
            "no credential for proprietary provider 'parslee': run `car auth login parslee`"
        ));
        assert!(is_auth_failure(
            "your Parslee session has expired or was rejected"
        ));
        assert!(is_auth_failure(
            "car-auth: cannot read Parslee credentials (secret store error)"
        ));

        assert!(!is_auth_failure("connection reset by peer"));
        assert!(!is_auth_failure("503 Service Unavailable"));
        assert!(!is_auth_failure("script exhausted"));
        assert!(!is_auth_failure("model failed, trying next fallback"));
    }

    /// The native-side twin of `external_loop`'s transport regression: a dead
    /// backbone must not be able to fail a session whose worktree already
    /// satisfies the contract. This path used to return red without ever
    /// evaluating, letting the inference transport pronounce a verdict it has
    /// no standing to give.
    #[tokio::test]
    async fn a_dead_backbone_over_green_checks_still_passes() {
        let outcome = run_against(&dead_backbone(), &crate::coder::test_cmds::touch("m.txt")).await;
        assert!(outcome.passed, "the contract decides: {outcome:?}");
        assert_eq!(outcome.failure, None);
        assert!(outcome.error.is_none());
    }

    /// The other direction: a dead backbone over red checks is still a loss,
    /// and still attributed to the machinery rather than the work.
    #[tokio::test]
    async fn a_dead_backbone_over_red_checks_is_infrastructure() {
        let outcome = run_against(&dead_backbone(), "exit 1").await;
        assert!(!outcome.passed);
        assert_eq!(outcome.failure, Some(LoopFailure::Infrastructure));
        assert!(outcome
            .error
            .expect("a dead backbone must surface")
            .contains("inference failed repeatedly"));
    }

    #[tokio::test]
    async fn scripted_loop_edits_verifies_and_passes() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, collected) = EventSink::collecting("coder-native");
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // Turn 1: write the file. Turn 2: declare done (no tool calls).
        let script = Script {
            turns: vec![
                turn(
                    "creating the file",
                    serde_json::json!([{
                        "id": "c1",
                        "name": "write_file",
                        "arguments": {"path": "hello.txt", "content": "hello coder"}
                    }]),
                ),
                turn("done — file created", serde_json::json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "hello.txt exists with content".into(),
            checks: vec![ContractCheck {
                name: "exists".into(),
                command: crate::coder::test_cmds::contains("coder", "hello.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "create hello.txt containing 'hello coder'",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        assert!(outcome.passed, "outcome: {outcome:?}");
        assert_eq!(outcome.iterations, 1);
        assert!(dir.path().join("hello.txt").exists());

        // Event stream shape: iteration → tool call/result → plan → check.
        let events = collected.lock().unwrap();
        let types: Vec<&str> = events
            .iter()
            .map(|e| match &e.kind {
                CoderEventKind::IterationStarted { .. } => "iteration",
                CoderEventKind::ToolCall { .. } => "tool_call",
                CoderEventKind::ToolResult { .. } => "tool_result",
                CoderEventKind::PlanText { .. } => "plan",
                CoderEventKind::CheckStarted { .. } => "check_started",
                CoderEventKind::CheckCompleted { .. } => "check_completed",
                _ => "other",
            })
            .collect();
        assert_eq!(
            types,
            vec![
                "iteration",
                "tool_call",
                "tool_result",
                "plan",
                "check_started",
                "check_completed"
            ]
        );
    }

    // Identical read_file every turn — the thrash a backbone that drops tool
    // history induces (re-read forever, never edit, never declare done).
    fn identical_read_turn() -> InferenceResult {
        turn(
            "reading again",
            serde_json::json!([{
                "id": "c",
                "name": "read_file",
                "arguments": {"path": "src.py"}
            }]),
        )
    }

    #[tokio::test]
    async fn native_loop_no_progress_bails_but_green_contract_still_passes() {
        // A no-progress trip must NOT fail a session whose earlier state already
        // satisfies the contract — the guard breaks to contract evaluation, and a
        // green contract wins. (Regression guard against reporting green as failed.)
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, _collected) = EventSink::collecting("coder-native");
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        let script = Script {
            turns: (0..NO_PROGRESS_REPEAT_LIMIT + 2)
                .map(|_| identical_read_turn())
                .collect(),
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        // Always-green contract.
        let contract = OutcomeContract {
            description: "already satisfied".into(),
            checks: vec![ContractCheck {
                name: "ok".into(),
                // Portable `exit 0` — a bare `true` is a POSIX builtin and runs
                // through the coder's shell (`cmd /C` on Windows), where it is
                // not a command, so the "always-green" contract came back red.
                command: crate::coder::test_cmds::PASS.into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "fix the bug",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        assert!(
            outcome.passed,
            "green contract must pass despite the thrash: {outcome:?}"
        );
        assert_eq!(outcome.iterations, 1);
    }

    #[tokio::test]
    async fn native_loop_aborts_after_two_no_progress_iterations() {
        // A genuinely wedged backbone re-thrashes the persistent conversation every
        // repair round; after the second no-progress iteration (contract still red)
        // the session aborts with a diagnostic instead of burning every iteration.
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, _collected) = EventSink::collecting("coder-native");
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // Enough identical turns for two iterations to each trip the guard.
        let script = Script {
            turns: (0..(NO_PROGRESS_REPEAT_LIMIT * 2 + 4))
                .map(|_| identical_read_turn())
                .collect(),
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        // Always-red contract, so each no-progress iteration stays failed.
        let contract = OutcomeContract {
            description: "never satisfied".into(),
            checks: vec![ContractCheck {
                name: "never".into(),
                command: crate::coder::test_cmds::FAIL.to_string(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "fix the bug",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        assert!(!outcome.passed, "outcome: {outcome:?}");
        let err = outcome.error.unwrap_or_default();
        assert!(err.contains("no-progress loop"), "error was: {err}");
        // Aborted on the second iteration — not left to run all 8.
        assert_eq!(outcome.iterations, 2);
    }

    #[tokio::test]
    async fn native_loop_empty_tool_calls_journals_turn_completed() {
        // The empty-tool-calls "model says done" terminal must record a durable
        // TurnCompleted so a truncated/starved local "done" is mineable as
        // false-success rather than passing silently as a clean finish.
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let journal = dir.path().join("events.jsonl");
        // A journaled sink (record_turn_completed is journal-only — a collecting
        // sink has no journal, so it would be a no-op there).
        let sink = EventSink::new("coder-native", None, Some(journal.clone()));
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        let script = Script {
            turns: vec![
                turn(
                    "creating the file",
                    serde_json::json!([{
                        "id": "c1",
                        "name": "write_file",
                        "arguments": {"path": "hello.txt", "content": "hello coder"}
                    }]),
                ),
                turn("done — file created", serde_json::json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "hello.txt exists with content".into(),
            checks: vec![ContractCheck {
                name: "exists".into(),
                command: crate::coder::test_cmds::contains("coder", "hello.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "create hello.txt containing 'hello coder'",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(outcome.passed, "outcome: {outcome:?}");

        // Drop the sink to join the journal writer thread (flush), then reload.
        drop(sink);
        let log = car_eventlog::EventLog::load(&journal).unwrap();
        let terminals: Vec<_> = log
            .events()
            .iter()
            .filter(|e| e.kind == car_eventlog::EventKind::TurnCompleted)
            .collect();
        assert_eq!(
            terminals.len(),
            1,
            "exactly one empty-tool-calls terminal recorded"
        );
        let ev = terminals[0];
        assert_eq!(
            ev.data.get("decision"),
            Some(&serde_json::json!("empty_tool_calls"))
        );
        assert_eq!(
            ev.data.get("model_id"),
            Some(&serde_json::json!("scripted"))
        );
        // "scripted" carries no provider prefix in the allow-list → unknown tier
        // (a real local run would surface e.g. "mlx/...": local).
        assert_eq!(
            ev.data.get("model_tier"),
            Some(&serde_json::json!("unknown"))
        );
    }

    #[tokio::test]
    async fn native_loop_injects_proactive_memory_from_journaled_failures() {
        use car_memgine::MemgineEngine;
        use std::sync::Mutex as StdMutex;
        use tokio::sync::Mutex as AsyncMutex;

        struct CaptureContext {
            seen: Arc<StdMutex<Vec<Option<String>>>>,
        }

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

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let journal = dir.path().join("events.jsonl");
        let sink = EventSink::new("coder-native", None, Some(journal.clone()));
        sink.emit(CoderEventKind::ToolResult {
            tool: "shell".into(),
            ok: false,
            preview: "pytest failed because fixture data is missing".into(),
        });
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let memory = RepairMemory::new(Some(Arc::new(AsyncMutex::new(MemgineEngine::new(None)))));
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let capture = CaptureContext { seen: seen.clone() };
        let contract = OutcomeContract {
            description: "noop".into(),
            checks: vec![],
        };

        let outcome = run_native_loop(
            &capture,
            &executor,
            "fix the pytest failure",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &memory,
            None,
        )
        .await;

        assert!(outcome.passed, "outcome: {outcome:?}");
        let contexts = seen.lock().unwrap();
        let context = contexts[0].as_deref().unwrap_or("");
        assert!(
            context.contains("## Proactive Memory"),
            "request context should carry proactive memory: {context}"
        );
        assert!(
            context.contains("Action shell in proposal session failed"),
            "journaled failure should be injected: {context}"
        );
        drop(sink);
        let log = car_eventlog::EventLog::load(&journal).unwrap();
        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(&serde_json::json!("inject"))
        }));
    }

    #[tokio::test]
    async fn native_loop_turn_budget_exhaustion_journals_max_turns() {
        // The model never declares done — it keeps issuing tool calls until the
        // per-iteration turn budget is exhausted. That "never finishes" terminal
        // must be journaled as a max_turns TurnCompleted (the coder path has no
        // stall guard, so this is the only signal that the budget burned out).
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let journal = dir.path().join("events.jsonl");
        let sink = EventSink::new("coder-native", None, Some(journal.clone()));
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        let tool_turn = || {
            turn(
                "still working",
                serde_json::json!([{
                    "id": "c",
                    "name": "write_file",
                    "arguments": {"path": "scratch.txt", "content": "x"}
                }]),
            )
        };
        let script = Script {
            turns: vec![tool_turn(), tool_turn()],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "never satisfied".into(),
            checks: vec![ContractCheck {
                name: "exists".into(),
                command: crate::coder::test_cmds::contains("coder", "hello.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        let cfg = NativeLoopConfig {
            prompt_overlay: None,
            model: None,
            max_iterations: 1,
            max_turns_per_iteration: 2,
            max_tokens_per_turn: 4096,
            deadline: SessionDeadline::shared_default(),
            auth_gate: None,
            auth_wait: std::time::Duration::ZERO,
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "keep writing forever",
            &contract,
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(!outcome.passed, "outcome: {outcome:?}");

        drop(sink);
        let log = car_eventlog::EventLog::load(&journal).unwrap();
        let max_turns: Vec<_> = log
            .events()
            .iter()
            .filter(|e| {
                e.kind == car_eventlog::EventKind::TurnCompleted
                    && e.data.get("decision") == Some(&serde_json::json!("max_turns"))
            })
            .collect();
        assert_eq!(
            max_turns.len(),
            1,
            "turn-budget exhaustion recorded once as max_turns"
        );
        assert_eq!(max_turns[0].data.get("turns"), Some(&serde_json::json!(2)));
    }

    #[tokio::test]
    async fn native_loop_compacts_persistent_history_to_context_window() {
        // F2 persists ONE conversation across repair turns; without bounding it to
        // the model's context window, a long run overflows and head-truncates the
        // System prompt (outcome contract) + task provider-side on a small local
        // model — the exact silent failure this loop must avoid. Assert the loop
        // calls compaction each turn: with a tiny window and large turns, the
        // history the generator sees stays bounded (does NOT accumulate one
        // exchange per turn) and always keeps the System prompt pinned at the head.
        use std::sync::Mutex;

        struct RecordingGen {
            // (message count, starts-with-System) observed on each generate.
            seen: Arc<Mutex<Vec<(usize, bool)>>>,
            // Distinct call per turn so the (legitimate) no-progress guard, which
            // aborts on identical repeats, doesn't fire before all 12 turns run.
            turn_no: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for RecordingGen {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                let msgs = req.messages.as_ref().expect("coder always sets messages");
                let starts_with_system = matches!(msgs.first(), Some(Message::System { .. }));
                self.seen
                    .lock()
                    .unwrap()
                    .push((msgs.len(), starts_with_system));
                // A large assistant turn + a DISTINCT tool call every turn so the
                // persistent thread would grow unbounded absent compaction, and the
                // model never declares done (the contract below never passes).
                let n = self.turn_no.fetch_add(1, Ordering::SeqCst);
                Ok(turn(
                    &"x".repeat(8000),
                    serde_json::json!([{
                        "id": format!("c{n}"),
                        "name": "write_file",
                        "arguments": {"path": format!("big{n}.txt"), "content": "y"}
                    }]),
                ))
            }
            fn context_window(&self, _model: &str) -> usize {
                200 // tiny: budget = 150 tokens, far below one 8KB assistant turn
            }
        }

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, _collected) = EventSink::collecting("compact-test");
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let seen: Arc<Mutex<Vec<(usize, bool)>>> = Arc::new(Mutex::new(Vec::new()));
        let gen = RecordingGen {
            seen: seen.clone(),
            turn_no: AtomicUsize::new(0),
        };

        let cfg = NativeLoopConfig {
            prompt_overlay: None,
            model: Some("scripted".into()),
            max_iterations: 1,
            max_turns_per_iteration: 12,
            max_tokens_per_turn: 4096,
            deadline: SessionDeadline::shared_default(),
            auth_gate: None,
            auth_wait: std::time::Duration::ZERO,
        };
        let contract = OutcomeContract {
            description: "never satisfied".into(),
            checks: vec![ContractCheck {
                name: "never".into(),
                command: crate::coder::test_cmds::FAIL.to_string(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let _ = run_native_loop(
            &gen,
            &executor,
            "grow the thread",
            &contract,
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await;

        let seen = seen.lock().unwrap();
        assert_eq!(seen.len(), 12, "all 12 turns generated");
        // The System prompt (outcome contract) is pinned at the head EVERY turn —
        // compaction never drops it.
        assert!(
            seen.iter().all(|(_, sys)| *sys),
            "System prompt must stay pinned every turn"
        );
        // Bounded: absent compaction the 12th turn would see ~2 + 2*11 = 24
        // messages. Compaction caps it near the pinned head + recent tail.
        let max_len = seen.iter().map(|(n, _)| *n).max().unwrap();
        assert!(
            max_len < 14,
            "persistent history not bounded — max messages/turn = {max_len}"
        );
    }

    #[tokio::test]
    async fn native_loop_routes_high_stakes() {
        // The coder loop edits and runs code in a real worktree, so EVERY
        // inference it issues must carry the high_stakes hint (quality-first) —
        // a wrong edit lands in a real repo. Capture every turn's intent (not
        // just the first) so the invariant survives a future refactor that might
        // hoist intent-setting out of the per-turn loop into a branch.
        use std::sync::Mutex;
        struct CapturingGen {
            intents: Arc<Mutex<Vec<Option<car_inference::IntentHint>>>>,
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for CapturingGen {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                self.intents.lock().unwrap().push(req.intent.clone());
                // Turn 0 issues a tool call so the loop runs a SECOND turn; turn
                // 1 declares done so it then exits. Two captured intents prove
                // the hint rides every turn, not just the first.
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                if i == 0 {
                    Ok(turn(
                        "",
                        serde_json::json!([{
                            "id": "c1", "name": "write_file",
                            "arguments": {"path": "f.txt", "content": "x"}
                        }]),
                    ))
                } else {
                    Ok(turn("done", serde_json::json!([])))
                }
            }
        }

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let captured = Arc::new(Mutex::new(Vec::new()));
        let gen = CapturingGen {
            intents: captured.clone(),
            cursor: AtomicUsize::new(0),
        };
        let contract = OutcomeContract {
            description: "noop".into(),
            checks: vec![],
        };

        let _ = run_native_loop(
            &gen,
            &executor,
            "make a change",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        let intents = captured.lock().unwrap();
        assert!(
            intents.len() >= 2,
            "expected the loop to issue multiple inferences, got {}",
            intents.len()
        );
        for (n, intent) in intents.iter().enumerate() {
            let intent = intent
                .as_ref()
                .unwrap_or_else(|| panic!("turn {n} issued an inference with no IntentHint"));
            assert!(intent.high_stakes, "turn {n} must route high_stakes");
            assert_eq!(
                intent.task,
                Some(car_inference::TaskHint::Code),
                "turn {n} must keep the Code task hint"
            );
        }
    }

    #[tokio::test]
    async fn scripted_loop_repairs_after_red_checks() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // Iter 1: writes the WRONG content, says done → check fails.
        // Iter 2: fixes it → check passes.
        let script = Script {
            turns: vec![
                turn(
                    "",
                    serde_json::json!([{
                        "id": "c1", "name": "write_file",
                        "arguments": {"path": "x.txt", "content": "wrong"}
                    }]),
                ),
                turn("done", serde_json::json!([])),
                turn(
                    "",
                    serde_json::json!([{
                        "id": "c2", "name": "write_file",
                        "arguments": {"path": "x.txt", "content": "right"}
                    }]),
                ),
                turn("fixed", serde_json::json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "x.txt says right".into(),
            checks: vec![ContractCheck {
                name: "content".into(),
                command: crate::coder::test_cmds::contains_or_report("right", "x.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "write right into x.txt",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(outcome.passed);
        assert_eq!(outcome.iterations, 2, "one repair round expected");
    }

    /// F2 (audit 2026-07-06): the coder must PERSIST its conversation across
    /// repair iterations — the files it read, the edits it made, the dead-ends
    /// it ruled out — appending the failing-check feedback as a new turn rather
    /// than resetting to `[System, User]` each round. Prove it: iteration 2's
    /// first inference must carry iteration 1's tool call.
    #[tokio::test]
    async fn f2_iteration_two_carries_iteration_one_conversation() {
        use std::sync::Mutex as StdMutex;

        struct MsgCapture {
            seen: Arc<StdMutex<Vec<String>>>,
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for MsgCapture {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                self.seen
                    .lock()
                    .unwrap()
                    .push(serde_json::to_string(&req.messages).unwrap_or_default());
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                match i {
                    0 => Ok(turn(
                        "",
                        serde_json::json!([{
                            "id": "iter1call", "name": "write_file",
                            "arguments": {"path": "x.txt", "content": "ITER1_WRONG"}
                        }]),
                    )),
                    1 => Ok(turn("done", serde_json::json!([]))),
                    2 => Ok(turn(
                        "",
                        serde_json::json!([{
                            "id": "iter2call", "name": "write_file",
                            "arguments": {"path": "x.txt", "content": "ITER2_right"}
                        }]),
                    )),
                    _ => Ok(turn("fixed", serde_json::json!([]))),
                }
            }
        }

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let seen = Arc::new(StdMutex::new(Vec::new()));
        let gen = MsgCapture {
            seen: seen.clone(),
            cursor: AtomicUsize::new(0),
        };
        let contract = OutcomeContract {
            description: "x.txt says ITER2_right".into(),
            checks: vec![ContractCheck {
                name: "content".into(),
                command: crate::coder::test_cmds::contains("ITER2_right", "x.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &gen,
            &executor,
            "write ITER2_right into x.txt",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        assert!(outcome.passed);
        assert_eq!(outcome.iterations, 2, "expected a repair round");
        let seen = seen.lock().unwrap();
        assert!(
            seen.len() >= 4,
            "expected >=4 inferences, got {}",
            seen.len()
        );
        // Iteration 2's opening inference must carry iteration 1's tool call —
        // the conversation is persisted, not reset.
        assert!(
            seen[2].contains("ITER1_WRONG") || seen[2].contains("iter1call"),
            "F2: iteration 2 lost iteration 1's conversation:\n{}",
            seen[2]
        );
    }

    /// F3 (audit 2026-07-06): a turn that hit the max_tokens ceiling
    /// (stop_reason=length) is CUT OFF, not a completion. The loop must NOT
    /// treat an empty-tool-call truncated turn as "done" — it must recognize
    /// the truncation and continue so the model can finish.
    #[tokio::test]
    async fn f3_truncated_turn_is_not_treated_as_done() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        let script = Script {
            turns: vec![
                // Turn 0: truncated mid-thought — no tool calls, stop_reason=length.
                turn_with_stop(
                    "partial output that got cut o",
                    serde_json::json!([]),
                    Some("length"),
                ),
                // Turn 1: a clean, natural completion.
                turn_with_stop("done for real", serde_json::json!([]), Some("stop")),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "noop".into(),
            checks: vec![],
        };

        let _ = run_native_loop(
            &script,
            &executor,
            "do the thing",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        // The truncated first turn must not end the iteration: the loop should
        // continue and consume the SECOND scripted turn (cursor advances to 2).
        assert_eq!(
            script.cursor.load(Ordering::SeqCst),
            2,
            "truncated turn was mistaken for completion — loop stopped early instead of continuing"
        );
    }

    /// End-to-end learning: a repair (red → green) records a durable skill
    /// keyed on the failure signature, and a *fresh* session that hits the same
    /// signature recalls that approach into its repair prompt.
    #[tokio::test]
    async fn repair_round_learns_and_recalls_across_sessions() {
        use crate::coder::skill_memory::FailureSignature;
        use car_memgine::MemgineEngine;
        use tokio::sync::Mutex as AsyncMutex;

        // A shared memgine survives both sessions (the "gets better" store).
        let memory = RepairMemory::new(Some(Arc::new(AsyncMutex::new(MemgineEngine::new(None)))));

        // A contract whose check fails LOUDLY with a recognizable error class
        // so the signature is stable across sessions.
        let contract = OutcomeContract {
            description: "x.txt says right".into(),
            checks: vec![ContractCheck {
                name: "content".into(),
                command: crate::coder::test_cmds::contains_or_report("right", "x.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        let sig = FailureSignature {
            check: "content".into(),
            error_class: "test_failure".into(),
        };

        // --- Session 1: red then green. The green-after-red ingests the skill.
        let dir1 = tempfile::tempdir().unwrap();
        let exec1 = WorktreeExecutor::new(dir1.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let script1 = Script {
            turns: vec![
                turn(
                    "",
                    serde_json::json!([{
                        "id": "c1", "name": "write_file",
                        "arguments": {"path": "x.txt", "content": "wrong"}
                    }]),
                ),
                turn("nothing useful yet", serde_json::json!([])),
                turn(
                    "",
                    serde_json::json!([{
                        "id": "c2", "name": "write_file",
                        "arguments": {"path": "x.txt", "content": "right"}
                    }]),
                ),
                turn(
                    "wrote 'right' into x.txt to satisfy the grep",
                    serde_json::json!([]),
                ),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let outcome1 = run_native_loop(
            &script1,
            &exec1,
            "write right into x.txt",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &memory,
            None,
        )
        .await;
        assert!(outcome1.passed);
        // The winning approach is now durably recallable for this signature.
        let recalled = memory
            .recall(&sig)
            .await
            .expect("session 1 should have learned");
        assert!(recalled.contains("right"), "approach captured: {recalled}");

        // --- Session 2: the SAME signature recurs. The loop must inject the
        // recalled hint into the repair prompt on the second iteration.
        let dir2 = tempfile::tempdir().unwrap();
        let exec2 = WorktreeExecutor::new(dir2.path());
        let (sink2, collected) = EventSink::collecting("coder-learn");
        let seen_hint = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // A generator that asserts on the prompt it receives: once a recall
        // hint shows up in the user message, it writes the fix.
        struct HintWatcher {
            seen: Arc<std::sync::atomic::AtomicBool>,
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for HintWatcher {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                let saw_hint = req
                    .messages
                    .as_ref()
                    .map(|ms| {
                        ms.iter().any(
                            |m| matches!(m, Message::User { content } if content.contains("HINT")),
                        )
                    })
                    .unwrap_or(false);
                if saw_hint {
                    self.seen.store(true, Ordering::SeqCst);
                }
                Ok(match i {
                    // Iter 1: do nothing → contract red → signature recorded.
                    0 => turn("did nothing", serde_json::json!([])),
                    // Iter 2 (hint present): write the fix.
                    1 => turn(
                        "",
                        serde_json::json!([{
                            "id": "c1", "name": "write_file",
                            "arguments": {"path": "x.txt", "content": "right"}
                        }]),
                    ),
                    _ => turn("applied the recalled fix", serde_json::json!([])),
                })
            }
        }

        let script2 = HintWatcher {
            seen: seen_hint.clone(),
            cursor: AtomicUsize::new(0),
        };
        let outcome2 = run_native_loop(
            &script2,
            &exec2,
            "write right into x.txt",
            &contract,
            &sink2,
            &cancel,
            &NativeLoopConfig::default(),
            &memory,
            None,
        )
        .await;
        assert!(outcome2.passed, "session 2 should pass: {outcome2:?}");
        assert!(
            seen_hint.load(Ordering::SeqCst),
            "the recalled hint must have been injected into the repair prompt"
        );
        drop(collected);
    }

    /// The `ask_user` tool routes to the [`AskUser`] handler (not the worktree
    /// executor), emits `UserInputRequested`, and the handler's answer is fed
    /// back to the model as the tool result — which the model then uses.
    #[tokio::test]
    async fn ask_user_tool_routes_to_handler_and_answer_reaches_model() {
        use std::sync::Mutex as StdMutex;

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, collected) = EventSink::collecting("coder-ask");
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // A canned asker that records the prompt it saw and returns a fixed
        // answer (stands in for the gate + coder.respond round-trip).
        struct CannedAsker {
            seen_prompt: Arc<StdMutex<Option<String>>>,
            answer: String,
        }
        #[async_trait]
        impl AskUser for CannedAsker {
            async fn ask(&self, prompt: &str) -> Result<String, String> {
                *self.seen_prompt.lock().unwrap() = Some(prompt.to_string());
                Ok(self.answer.clone())
            }
        }
        let seen_prompt = Arc::new(StdMutex::new(None));
        let asker = CannedAsker {
            seen_prompt: seen_prompt.clone(),
            answer: "use port 8080".to_string(),
        };

        // The script: ask a question, then (turn 2) write the answer it got
        // back into a file, then declare done. A generator that echoes the
        // ask_user tool result into the write proves the answer reached it.
        struct AskThenWrite {
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for AskThenWrite {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                match i {
                    0 => Ok(turn(
                        "",
                        serde_json::json!([{
                            "id": "a1", "name": "ask_user",
                            "arguments": {"prompt": "which port?"}
                        }]),
                    )),
                    1 => {
                        // Pull the answer out of the ToolResult the loop appended.
                        let answer = req
                            .messages
                            .as_ref()
                            .and_then(|ms| {
                                ms.iter().rev().find_map(|m| match m {
                                    Message::ToolResult { content, .. } => Some(content.clone()),
                                    _ => None,
                                })
                            })
                            .unwrap_or_default();
                        Ok(turn(
                            "",
                            serde_json::json!([{
                                "id": "w1", "name": "write_file",
                                "arguments": {"path": "answer.txt", "content": answer}
                            }]),
                        ))
                    }
                    _ => Ok(turn("done", serde_json::json!([]))),
                }
            }
        }

        let contract = OutcomeContract {
            description: "answer.txt records the chosen port".into(),
            checks: vec![ContractCheck {
                name: "has_port".into(),
                command: crate::coder::test_cmds::contains("8080", "answer.txt"),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };

        let outcome = run_native_loop(
            &AskThenWrite {
                cursor: AtomicUsize::new(0),
            },
            &executor,
            "pick a port and record it",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            Some(&asker),
        )
        .await;

        assert!(outcome.passed, "outcome: {outcome:?}");
        // The handler saw the model's question.
        assert_eq!(seen_prompt.lock().unwrap().as_deref(), Some("which port?"));
        // The answer reached the model and was written through.
        assert_eq!(
            std::fs::read_to_string(dir.path().join("answer.txt")).unwrap(),
            "use port 8080"
        );
        // The ask_user call surfaced in the event stream as a tool call (the
        // semantic UserInputRequested event is the GateAsker's job, covered by
        // the rpc round-trip test).
        let events = collected.lock().unwrap();
        assert!(events.iter().any(|e| matches!(
            &e.kind,
            CoderEventKind::ToolCall { tool, .. } if tool == ASK_USER_TOOL
        )));
    }

    /// Without an asker the `ask_user` tool is not offered, and if a model calls
    /// it anyway the loop returns a recoverable tool error rather than wedging.
    #[tokio::test]
    async fn ask_user_without_handler_is_a_recoverable_error() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, _collected) = EventSink::collecting("coder-noask");
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // ask_user is absent from the offered tools when ask is None.
        struct ToolPeek {
            offered: Arc<std::sync::atomic::AtomicBool>,
        }
        #[async_trait]
        impl TurnGenerator for ToolPeek {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                let has_ask = req
                    .tools
                    .as_ref()
                    .map(|ts| ts.iter().any(|t| t["name"] == ASK_USER_TOOL))
                    .unwrap_or(false);
                self.offered.store(has_ask, Ordering::SeqCst);
                Ok(turn("done", serde_json::json!([])))
            }
        }
        let offered_flag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let contract = OutcomeContract {
            description: "noop".into(),
            checks: vec![ContractCheck {
                name: "ok".into(),
                command: crate::coder::test_cmds::PASS.to_string(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        let _ = run_native_loop(
            &ToolPeek {
                offered: offered_flag.clone(),
            },
            &executor,
            "x",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(
            !offered_flag.load(Ordering::SeqCst),
            "ask_user must not be offered when no handler is wired"
        );
    }

    #[tokio::test]
    async fn cancellation_stops_the_loop() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(true));
        let script = Script {
            turns: vec![],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "d".into(),
            checks: vec![ContractCheck {
                name: "never".into(),
                command: crate::coder::test_cmds::PASS.to_string(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        let outcome = run_native_loop(
            &script,
            &executor,
            "x",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert_eq!(outcome.error.as_deref(), Some("cancelled"));
        assert_eq!(outcome.iterations, 0);
    }

    #[test]
    fn failure_feedback_lists_only_failures() {
        let results = vec![
            CheckResult {
                name: "good".into(),
                passed: true,
                exit_code: Some(0),
                output_tail: "ok".into(),
                duration_ms: 1,
            },
            CheckResult {
                name: "bad".into(),
                passed: false,
                exit_code: Some(1),
                output_tail: "assertion failed".into(),
                duration_ms: 1,
            },
        ];
        let fb = failure_feedback(&results, 0);
        assert!(fb.contains("FAILED bad"));
        assert!(fb.contains("assertion failed"));
        assert!(!fb.contains("FAILED good"));
        // A fresh failure gets the read-the-error direction, not the escalation.
        assert!(fb.contains("name the single cause"));
        assert!(!fb.contains("failed 2 times in a row"));
    }

    #[test]
    fn failure_feedback_escalates_on_a_recurring_failure() {
        let results = vec![CheckResult {
            name: "run_tests".into(),
            passed: false,
            exit_code: Some(1),
            output_tail: "AttributeError: no attribute '_remove_slot_root'".into(),
            duration_ms: 1,
        }];
        // Same signature failing a 2nd time (recurrences=1) → escalate: stop
        // repeating the approach, implement the named missing symbol.
        let fb = failure_feedback(&results, 1);
        assert!(fb.contains("failed the same way 2 times"), "{fb}");
        assert!(fb.contains("do NOT re-apply a variation"));
        assert!(fb.contains("IMPLEMENT it"));
    }

    #[test]
    fn system_prompt_carries_the_contract() {
        let contract = OutcomeContract {
            description: "make the tests pass".into(),
            checks: vec![super::super::contract::ContractCheck {
                name: "tests".into(),
                command: "cargo test -p demo".into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 300,
            }],
        };
        let p = system_prompt(&contract, "Top-level entries: Cargo.toml, src");
        assert!(p.contains("cargo test -p demo"));
        assert!(p.contains("STOP calling tools"));
        // The coder must self-verify with the contract's exact command, not a
        // broad guess (a broad run in a large repo trips on unrelated breakage).
        assert!(p.contains("EXACT command(s) from the OUTCOME CONTRACT"));
    }

    #[test]
    fn coder_prompt_contains_discipline_and_keeps_stop_contract() {
        let contract = OutcomeContract {
            description: "make the tests pass".into(),
            checks: vec![ContractCheck {
                name: "tests".into(),
                command: "cargo test -p demo".into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 300,
            }],
        };
        let env = "Top-level entries: Cargo.toml, src\nBuild systems detected: Rust (cargo)";
        let p = system_prompt(&contract, env);

        // Real coding discipline is present.
        assert!(p.contains("Inspect before you edit"), "inspect-first");
        assert!(
            p.contains("grep_files") && p.contains("find_files"),
            "search-before-read discipline"
        );
        assert!(p.contains("prefer edit_file"), "surgical-edit discipline");
        assert!(
            p.contains("Never fabricate file contents"),
            "anti-fabrication (files)"
        );
        assert!(
            p.contains("Never claim a check passed"),
            "anti-fabrication (results)"
        );
        assert!(
            p.contains("read the actual error output before retrying"),
            "read-the-error discipline"
        );
        // Dogfooding fix #1: a blocked self-verify must not be reported as failure.
        assert!(
            p.contains("is NOT a task failure") && p.contains("blocked"),
            "blocked-verification-is-not-failure guidance"
        );
        // Dogfooding fix #3 (evidence-backed A/B, 55%->100%): trace each check.
        // The task-specific pitfall it also carried ("a bare set() loses order")
        // was one eval's failure promoted to a permanent global rule; language-
        // specific correctness guidance belongs in a contract check or a `.car/`
        // rubric, not in every coder turn forever.
        assert!(
            p.contains("Trace the checks before you declare done"),
            "check-tracing guidance"
        );
        assert!(
            !p.contains("set()"),
            "no eval-specific correctness hints in the global prompt"
        );
        // Dogfooding fix (A/B 809s vs 58s root-cause): the coder fixed the bug in
        // ~6 turns then burned ~30 fighting a python3.14 self-verify env mismatch —
        // ran plain `python -m pytest` (wrong interpreter), then wrote
        // sitecustomize.py/UserDict.py shims + pip install to "repair" it. Verify
        // with the contract's EXACT command; the env-repair BAN itself now lives in
        // `coder::policy::DenyEnvironmentRepair`, where it is enforced rather than
        // merely stated, so the prompt carries only the judgment half.
        assert!(
            p.contains("copy the command string character-for-character"),
            "exact-command self-verify (no broad substitute)"
        );
        assert!(
            p.contains("The environment is not yours to fix") && p.contains("denied by policy"),
            "environment repair: judgment in the prompt, enforcement in policy"
        );
        assert!(
            !p.contains("STRICTLY FORBIDDEN"),
            "the enumerated prose blacklist moved to the inspector chain"
        );

        // The runtime-verifies framing is REFRAMED (verify yourself first), not
        // a promise the runtime does it for you.
        assert!(p.contains("do not rely on it: verify the checks yourself first"));

        // Load-bearing behavior contracts are preserved verbatim / accurately.
        assert!(
            p.contains("reply with a brief plain-text summary and STOP calling tools"),
            "the STOP-calling-tools loop-termination contract must survive verbatim"
        );
        // `git commit` is deliberately NOT denied by the inspector chain (see
        // coder::policy::tests::git_push_and_remote_mutation_denied), so this
        // prose line is the only thing holding the rule — it must survive.
        assert!(p.contains("Do not git commit"), "policy: no git commit");
        // The ENVIRONMENT section (F7/L1) carries the repo summary.
        assert!(p.contains("ENVIRONMENT:"));
        assert!(p.contains("Rust (cargo)"));
        // And the outcome contract is still rendered.
        assert!(p.contains("cargo test -p demo"));
    }

    #[test]
    fn preview_truncates_on_char_boundary() {
        assert_eq!(preview("short", 10), "short");
        let long = "é".repeat(300);
        let p = preview(&long, 5);
        assert!(p.ends_with('…') && p.chars().count() <= 4);
    }

    /// A generator that captures the FIRST user message of the request it
    /// receives, then declares done — for asserting what the model sees on the
    /// opening turn.
    struct FirstUserCapture {
        captured: Arc<std::sync::Mutex<String>>,
    }
    #[async_trait]
    impl TurnGenerator for FirstUserCapture {
        async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
            let first_user = req
                .messages
                .as_ref()
                .and_then(|ms| {
                    ms.iter().find_map(|m| match m {
                        Message::User { content } => Some(content.clone()),
                        _ => None,
                    })
                })
                .unwrap_or_default();
            *self.captured.lock().unwrap() = first_user;
            Ok(turn("done", serde_json::json!([])))
        }
    }

    fn trivial_contract() -> OutcomeContract {
        OutcomeContract {
            description: "trivial".into(),
            checks: vec![ContractCheck {
                name: "ok".into(),
                command: crate::coder::test_cmds::PASS.to_string(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        }
    }

    #[tokio::test]
    async fn coder_first_message_carries_recall_when_facts_exist() {
        use crate::coder::skill_memory::FailureSignature;
        use car_memgine::MemgineEngine;
        use tokio::sync::Mutex as AsyncMutex;

        // Seed a shared engine with a prior-session repair lead overlapping the
        // intent keywords ("tests").
        let memory = RepairMemory::new(Some(Arc::new(AsyncMutex::new(MemgineEngine::new(None)))));
        let sig = FailureSignature {
            check: "tests".into(),
            error_class: "test_failure".into(),
        };
        memory
            .record_success(&sig, "add the missing import and re-run cargo test")
            .await;

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let captured = Arc::new(std::sync::Mutex::new(String::new()));

        let outcome = run_native_loop(
            &FirstUserCapture {
                captured: captured.clone(),
            },
            &executor,
            "the tests are failing, please fix them",
            &trivial_contract(),
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &memory,
            None,
        )
        .await;
        assert!(outcome.passed, "outcome: {outcome:?}");

        let first_user = captured.lock().unwrap().clone();
        assert!(
            first_user.contains("Recall from prior sessions"),
            "the labelled session-start recall must be in the first user turn: {first_user}"
        );
        assert!(
            first_user.contains("missing import"),
            "the recalled approach content rides along: {first_user}"
        );
    }

    #[tokio::test]
    async fn coder_first_message_recall_absent_when_empty() {
        use car_memgine::MemgineEngine;
        use tokio::sync::Mutex as AsyncMutex;

        // A live engine with NOTHING learned → recall_for_task returns None →
        // no recall section is injected (no empty boilerplate).
        let memory = RepairMemory::new(Some(Arc::new(AsyncMutex::new(MemgineEngine::new(None)))));

        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let sink = EventSink::test_sink();
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let captured = Arc::new(std::sync::Mutex::new(String::new()));

        let outcome = run_native_loop(
            &FirstUserCapture {
                captured: captured.clone(),
            },
            &executor,
            "the tests are failing, please fix them",
            &trivial_contract(),
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &memory,
            None,
        )
        .await;
        assert!(outcome.passed, "outcome: {outcome:?}");

        let first_user = captured.lock().unwrap().clone();
        assert!(
            !first_user.contains("Recall from prior sessions"),
            "no recall section when the engine has nothing relevant: {first_user}"
        );
    }
}