car_server_core/coder/

native_loop.rs

//! 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`](super::contract::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::{builtin_tool_labels, tool_output_is_external, ToolExecutor};
use car_inference::tasks::generate::{Message, Provenance};
use car_inference::{
    GenerateParams, GenerateRequest, InferenceEngine, InferenceError, InferenceResult,
};
use serde_json::Value;

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

/// The model seam: one turn of generation. Implemented by
/// [`InferenceEngine`] for production; test harnesses script it (the same
/// injected-generation philosophy as `car-builder`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnGenerationError {
    /// Strict routing removed every candidate before a provider was invoked.
    NoEligibleModel { excluded_models: String },
    /// Every other inference failure retains its existing operator-facing text.
    Other(String),
}

impl std::fmt::Display for TurnGenerationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NoEligibleModel { excluded_models } => write!(
                f,
                "no eligible model remains after strict exclusions: {}",
                excluded_models
            ),
            Self::Other(message) => f.write_str(message),
        }
    }
}

#[async_trait]
pub trait TurnGenerator: Send + Sync {
    async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String>;

    /// Coder-specific generation keeps routing exhaustion typed so an
    /// unattended caller can classify configuration failures without parsing
    /// error prose. Other consumers retain the original string seam.
    async fn generate_coder(
        &self,
        req: GenerateRequest,
    ) -> Result<InferenceResult, TurnGenerationError> {
        self.generate(req).await.map_err(TurnGenerationError::Other)
    }

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

    async fn generate_coder(
        &self,
        req: GenerateRequest,
    ) -> Result<InferenceResult, TurnGenerationError> {
        self.generate_tracked(req)
            .await
            .map_err(|error| match error {
                InferenceError::NoEligibleModel { excluded_models } => {
                    TurnGenerationError::NoEligibleModel { excluded_models }
                }
                other => TurnGenerationError::Other(other.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;
    }
}

/// What a [`CoderEventKind::ModelFallback`] says when the degrade was caused by
/// a rejected credential. One constant so the coder loop and the two contract
/// gates in `rpc.rs` word it identically (Parslee-ai/car#888).
pub(crate) const MODEL_FALLBACK_REASON: &str =
    "that lane needs sign-in (`car auth login`) — this run is continuing on a fallback model";

/// The journal spelling of a [`car_inference::FallbackReason`].
///
/// Stable strings, because they are what a reader of `<id>.events.jsonl` and
/// any offline miner match on — the enum's `Debug` is not a wire contract.
pub(crate) fn fallback_reason_label(reason: car_inference::FallbackReason) -> &'static str {
    use car_inference::FallbackReason as R;
    // A second table in another crate drifts from the first silently, so these
    // are pinned against the serde spelling by
    // `the_journal_labels_match_the_serde_spelling` below rather than trusted.
    match reason {
        R::CredentialRejected => "credential_rejected",
        R::CredentialAbsent => "credential_absent",
        R::RateLimited => "rate_limited",
        R::QuotaExhausted => "quota_exhausted",
        R::TimedOut => "timed_out",
        R::Failed => "failed",
    }
}

/// Whether a failure means the run is blocked on **sign-in** rather than on the
/// machinery — the one failure a person at the keyboard can clear in seconds.
///
/// Credential summary phrases and rejected-credential shapes are both owned by
/// car-inference. Keeping this as a delegation means a new route summary cannot
/// tell the operator to repair auth while the coder loop independently calls it
/// retryable infrastructure (Parslee-ai/car#888, #1248).
pub(crate) fn is_auth_failure(message: &str) -> bool {
    car_inference::is_auth_failure_message(message)
}

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

/// The no-change nomination tool. See [`super::no_change`] for what the runtime
/// does with a call, and why almost all of that is refusal.
pub const REPORT_NO_CHANGE_TOOL: &str = "report_no_change";

/// 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.
/// The tool the model calls to nominate "no code should change".
///
/// The description is deliberately discouraging about the two shapes that
/// cannot terminate autonomously, and explicit that a session which has already
/// edited is ineligible — a model that learns that from a refusal has already
/// wasted a round.
fn report_no_change_tool_def() -> Value {
    serde_json::json!({
        "name": REPORT_NO_CHANGE_TOOL,
        "description": "Conclude that NO code should change, and end the session by \
                        reporting that instead of producing a diff. Use this only when \
                        you have investigated and established one of: the reported \
                        problem does not exist (the code already handles it); the \
                        behaviour is intentional; or the real fix is not a code change \
                        at all. This is NOT a way to stop when the task is hard — a \
                        session that has already edited any file cannot use it, and \
                        reverting does not restore eligibility. Your conclusion is a \
                        nomination: the runtime checks it against the baseline results \
                        and may route it to a human rather than accept it.",
        "parameters": {
            "type": "object",
            "properties": {
                "kind": {
                    "type": "string",
                    "enum": ["premise_wrong", "deliberate_behavior", "non_code_decision"],
                    "description": "premise_wrong: the reported problem does not exist. \
                                    deliberate_behavior: the code does this on purpose. \
                                    non_code_decision: a real problem whose fix is a \
                                    migration, an operator decision, or a rollout."
                },
                "summary": {
                    "type": "string",
                    "description": "One line: the conclusion itself."
                },
                "evidence": {
                    "type": "string",
                    "description": "What you examined to reach it — files read, commands \
                                    run, what they showed. A reviewer reads this to decide \
                                    whether to believe you."
                }
            },
            "required": ["kind", "summary", "evidence"]
        }
    })
}

/// Read a `report_no_change` call's arguments, or say precisely what is wrong.
///
/// Strict about `kind` on purpose: a mistyped kind silently falling back to
/// `premise_wrong` would route a judgement call onto the one path that can
/// terminate without a human.
fn parse_nomination(params: &Value) -> Result<NoChangeNomination, String> {
    let kind_str = params
        .get("kind")
        .and_then(Value::as_str)
        .ok_or("ERROR: report_no_change requires `kind`")?;
    let kind = NoChangeKind::parse(kind_str).ok_or_else(|| {
        format!(
            "ERROR: unknown kind '{kind_str}' — must be one of premise_wrong, \
             deliberate_behavior, non_code_decision"
        )
    })?;
    let summary = params
        .get("summary")
        .and_then(Value::as_str)
        .ok_or("ERROR: report_no_change requires `summary`")?
        .to_string();
    let evidence = params
        .get("evidence")
        .and_then(Value::as_str)
        .ok_or("ERROR: report_no_change requires `evidence`")?
        .to_string();
    Ok(NoChangeNomination {
        kind,
        summary,
        evidence,
    })
}

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>,
    /// Models the adaptive arm must not use. Ignored when [`Self::model`] is
    /// pinned: a caller's explicit model is either honored or refused by its
    /// own admission gate, never silently routed around.
    pub exclude_models: Vec<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>>,
    /// Whether this caller can adjudicate a no-change nomination — i.e. it holds
    /// the baseline results, the contract's provenance and the mutation ledger,
    /// and will run [`super::no_change::evaluate_nomination`] on whatever the
    /// loop hands back.
    ///
    /// The `report_no_change` tool is offered ONLY when this is set. A caller
    /// that cannot judge a nomination must not invite one: `LoopOutcome` would
    /// come back neither green nor failed, and a caller not looking for that
    /// third shape reads it as an ordinary red and silently discards the
    /// finding. Advertising a tool nobody can honour is worse than not having
    /// it.
    ///
    /// Defaults to **false**, which is the correct answer for every caller that
    /// has not been taught the third shape — this is a capability declaration,
    /// not a behaviour switch.
    pub can_adjudicate_no_change: bool,
    /// 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,
    /// The session-start baseline captures differential checks compare against
    /// (car#1067). Empty when the contract declares none — and under an empty
    /// map a differential check fails closed with a "never captured" message
    /// rather than passing silently.
    pub baseline_captures: super::contract::BaselineCaptures,
}

impl Default for NativeLoopConfig {
    fn default() -> Self {
        Self {
            model: None,
            exclude_models: Vec::new(),
            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),
            can_adjudicate_no_change: false,
            baseline_captures: crate::coder::contract::BaselineCaptures::new(),
        }
    }
}

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 configured routing constraints leave no model eligible. No provider
    /// was invoked and no work was attempted, so this belongs to daemon
    /// configuration rather than the item being healed.
    Configuration,
    /// 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>,
    /// A no-change finding the loop nominated and the runtime accepted or
    /// parked. `Some` means the run ended WITHOUT a diff and without that being
    /// a failure — see [`super::no_change`].
    ///
    /// This is the one case where `passed` is false and `failure` is also
    /// `None`, which relaxes the invariant stated above: `failure` is `None`
    /// exactly when `passed` **or** `nomination.is_some()`. A nomination is
    /// neither a green contract nor a loss, and forcing it into either is the
    /// whole defect #1070 describes.
    pub nomination: Option<NoChangeNomination>,
    /// 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,
            nomination: None,
            cost_usd: None,
        }
    }

    /// The run ended without a diff, and correctly so — the runtime accepted or
    /// parked a no-change nomination. Neither green nor lost; see the
    /// `nomination` field for why that third shape has to exist.
    pub fn reported(
        finding: NoChangeNomination,
        iterations: u32,
        last_results: Vec<CheckResult>,
    ) -> Self {
        Self {
            passed: false,
            iterations,
            last_results,
            error: None,
            failure: None,
            nomination: Some(finding),
            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),
            nomination: None,
            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,
    project: Option<&str>,
    overlay: Option<&str>,
) -> String {
    let base = system_prompt(contract, environment, project);
    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, project: Option<&str>) -> String {
    // Placed AFTER the environment and BEFORE the how-to-work rules, because
    // these are constraints on how the work is done, not facts about the
    // machine. Empty when the repo carries neither instructions nor a `.car/`
    // project, so a repo with nothing gets a byte-identical prompt to before.
    let project_block = project
        .map(str::trim)
        .filter(|p| !p.is_empty())
        .map(|p| format!("{p}\n\n"))
        .unwrap_or_default();
    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\
         {project_block}\
         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. Do not publish the work \
           yourself by any route — `git push`, `gh pr create`, `gh release create`, \
           `npm publish`, `cargo publish` and the like are denied; the runtime opens the \
           pull request itself once the merge gate is approved. Read-only forge commands \
           (`gh pr view`, `gh run view`, `gh api` GET) stay available. (`sudo` and \
           destructive operations outside the worktree are denied too. Every denial comes \
           back 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)
    }
}

/// The tool list a coding session advertises to the model: the worktree
/// executor's static built-ins, plus [`ASK_USER_TOOL`] when the caller supplied
/// an [`AskUser`] handler.
///
/// Deliberately [`WorktreeExecutor::tool_defs`] and **not**
/// `executor.all_tool_defs()`. Taking the whole delegate surface would sweep in
/// the Parslee platform tools, which need a sign-in a coding run does not have.
/// The delegate tools a coder *should* see are added back by name in
/// [`run_native_loop`] — graph-memory `recall` (car#1071) and the governed
/// network pair (car#1073) — so what the model was offered stays a list a reader
/// can enumerate rather than whatever happens to be attached. Because this part
/// is derived from the type rather than the instance, the daemon's `coder.*`
/// loop and headless `car code-task` show the model the same built-ins
/// (Parslee-ai/car#1063). Changing that is a design decision, not a cleanup.
pub(crate) fn native_loop_tool_defs(with_ask: bool) -> Vec<Value> {
    let mut tools = WorktreeExecutor::tool_defs();
    if with_ask {
        tools.push(ask_user_tool_def());
    }
    tools
}

/// Every tool schema the native coder can advertise across its optional run
/// modes. Individual runs still filter `ask_user`, `report_no_change`, and
/// graph-memory recall according to the handlers and policy they carry.
pub fn model_tool_catalog() -> Vec<Value> {
    let mut tools = native_loop_tool_defs(true);
    tools.push(report_no_change_tool_def());
    tools.extend(
        crate::assistant::memory::MemoryTools::tool_defs()
            .into_iter()
            .filter(|tool| tool["name"] == "recall"),
    );
    // The inventory is the union across optional run modes. A real session
    // still receives these only after its explicit browser opt-in.
    tools.extend(
        crate::assistant::browser_tools::BrowserTools::new(std::env::temp_dir()).tool_defs(),
    );
    tools.sort_by(|left, right| left["name"].as_str().cmp(&right["name"].as_str()));
    tools.dedup_by(|left, right| left["name"] == right["name"]);
    tools
}

/// The network tools a coding session offers: `http_request` and `web_search`,
/// attached to every coder executor by
/// [`WorktreeExecutor::for_coder_session`] (car#1073).
///
/// `granted` is the operator's answer — [`WorktreeExecutor::permits_full_access`]
/// at the call site — and it is passed in rather than read here on purpose. It
/// resolves from the per-agent approval policy, a process-global file, so a test
/// that wanted to drive both sides of this could only do it by mutating the
/// environment out from under every test running beside it. Splitting the
/// decision from the lookup keeps the offered list testable either way.
pub(crate) fn network_tool_defs(executor: &WorktreeExecutor, granted: bool) -> Vec<Value> {
    if !granted {
        return Vec::new();
    }
    ["http_request", "web_search"]
        .iter()
        .flat_map(|name| executor.delegate_defs_named(name))
        .collect()
}

/// Everything a coding session offers the model, assembled in one place.
///
/// [`run_native_loop`] does not build this list itself — it calls this. That is
/// the point: a test that hand-rolls "what the loop offers" proves only that the
/// hand-rolled copy is right, and goes green through exactly the change it was
/// written to catch. #1063 extracted the static built-ins for that reason; this
/// finishes the job for the delegate half.
///
/// Three additions on top of [`native_loop_tool_defs`]:
///
/// - The no-change nomination tool, only for a caller that can adjudicate one.
/// - Graph-memory `recall` (car#1071), when the executor carries it. Only
///   `recall` is ever attached to a coder session — see `recall_only_memory_defs`
///   for why the write half is withheld.
/// - The governed network pair (car#1073), only once the operator has granted
///   `car-coder` the `full_access` tier. A tool in the prompt that the gate will
///   hard-block costs the model a turn and teaches it the wrong thing about the
///   task, so the grant is read here rather than discovered at dispatch.
/// - The assistant's browser surface (car#1069), only when the executor carries
///   it because this session explicitly opted in. The opt-in is the per-session
///   approval for those exact full-access tools; a configured Deny still wins.
///
/// `advertise_delegates` opens delegate dispatch, and is called beside each
/// addition so a name is never offered without being callable. Note what it does
/// NOT do: it is a single flag for every attached delegate, not a per-tool one,
/// and `recall` is attached unconditionally — so by the time the network branch
/// runs, dispatch is already open. Containment for the network pair is the
/// `full_access` tier those defs declare and the per-agent gate that reads it,
/// not this flag. Do not add a side-effecting delegate here on the belief that
/// leaving it unadvertised keeps it unreachable.
pub(crate) fn coder_session_tool_defs(
    executor: &WorktreeExecutor,
    with_ask: bool,
    can_adjudicate_no_change: bool,
) -> Vec<Value> {
    let mut tools = native_loop_tool_defs(with_ask);
    if can_adjudicate_no_change {
        tools.push(report_no_change_tool_def());
    }
    let memory_defs = executor.delegate_defs_named("recall");
    if !memory_defs.is_empty() {
        executor.advertise_delegates();
        tools.extend(memory_defs);
    }
    let net_defs = network_tool_defs(executor, executor.permits_full_access());
    if !net_defs.is_empty() {
        executor.advertise_delegates();
        tools.extend(net_defs);
    }
    let mut browser_defs = executor.delegate_defs_with_prefix("browse_");
    browser_defs.extend(executor.delegate_defs_with_prefix("browser_"));
    if !browser_defs.is_empty() {
        executor.advertise_delegates();
        tools.extend(browser_defs);
    }
    // Withhold what operator policy forbids outright.
    //
    // The inspector chain already refuses these at dispatch, so this is not a
    // new control — it stops the loop paying for one. A tool no call can
    // satisfy costs a schema on every request and a whole turn when the model
    // tries it, and a coding loop has a turn budget. Narrowing only: nothing
    // here can make a denied tool callable.
    //
    // Empty unless the executor came from `for_coder_session`, which is the
    // only constructor that loads operator rules.
    let denied = executor.denied_tools();
    if !denied.is_empty() {
        tools.retain(|def| !denied.contains(def["name"].as_str().unwrap_or_default()));
        // Same reasoning as the assistant's: withholding a tool removes the
        // operator's only incidental proof that the rule loaded. Every blocked
        // attempt used to leave a denial behind; a model that is never offered
        // the tool never attempts it, so a policy that silently failed to load
        // now looks exactly like one working perfectly. Said once per run, not
        // per turn — this function is called once, before the turn loop.
        //
        // A log line, not an event-log record, for the reason given at the
        // assistant's copy: no `EventKind` means "withheld at assembly", and
        // adding one changes a serialized shape that crosses every binding.
        tracing::info!(
            withdrawn = ?denied,
            "operator policy denies these tools outright; withheld from the coding loop's \
             advertised list (still refused at dispatch by the inspector chain)"
        );
    }
    tools
}

/// 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 tools = coder_session_tool_defs(executor, ask.is_some(), cfg.can_adjudicate_no_change);
    // 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());
    // Information-flow labels for provenance classification of tool results.
    // Built once per run rather than per call: the map is static for the run,
    // and the classification happens on every tool result.
    let tool_labels = builtin_tool_labels();
    // What the repository says about how to work in it — its CLAUDE.md/AGENTS.md
    // and any `.car/` knowledge the team recorded. Until this landed the loop
    // saw only the two lines above, so review-time rules that no outcome
    // contract can express (car#1071) never reached the model at all.
    let project = super::project_context::project_context(executor.worktree());
    let system = system_prompt_with_overlay(
        contract,
        &environment,
        project.as_deref(),
        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;
    // Whether this session already told the operator that its preferred
    // inference lane was skipped for want of a sign-in. Session-scoped and
    // announced ONCE: the point is to make a silent degrade visible, not to
    // repeat it on every turn of a long run (Parslee-ai/car#888).
    let mut announced_model_fallback = false;
    // The hop list written on the PREVIOUS turn, so an unchanged one is not
    // restated. Consecutive-only on purpose — see the loop below.
    let mut last_journaled_hops: Vec<(String, String, &'static str)> = Vec::new();
    // Whether the UNGATED auth path has already asked for a sign-in this run.
    // See the guard below for why only the ungated path is deduplicated.
    let mut announced_ungated_auth = false;
    // 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);
    let adaptive_exclusions = if cfg.model.is_none() {
        cfg.exclude_models.clone()
    } else {
        Vec::new()
    };
    let strict_exclusions = !adaptive_exclusions.is_empty();
    // The budget this run's history is bounded to, named once at the top rather
    // than only implied by a compaction that may never fire. Same shared
    // fraction the assistant loop and the declarative-agent runner use; log
    // only — the loop's behavior is unchanged.
    tracing::debug!(
        context_window,
        budget = history_budget(context_window),
        "coder history compaction budget resolved for this run"
    );

    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();
        // Every distinct model that served a turn in THIS iteration, in
        // first-seen order. `TurnCompleted` is a per-iteration terminal, so it
        // names only whichever model happened to finish — and an unpinned
        // session routes per request (`strict_model` is `cfg.model.is_some()`),
        // so a model can write turns 1-3 and be replaced before the terminal is
        // written. That left no record of it anywhere, which is how a model
        // could author most of a change and still pass the self-review gate
        // that refuses a reviewer who wrote what it is judging (car#1333).
        // EVERY exit from the turn loop owes a `record_turn_completed`. This is
        // carried on the stack and written out only by that call, so a `return`
        // added below without one silently drops the iteration's attribution —
        // which is how the gap this closes came to exist, the terminal already
        // being the only record.
        let mut models_this_iteration: Vec<String> = Vec::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,
                    exclude_models: adaptive_exclusions.clone(),
                    strict_exclusions,
                    ..Default::default()
                }),
                ..Default::default()
            };
            maybe_apply_coder_proactive_memory(&mut req, intent, &messages, sink, memory).await;

            let result = match inference.generate_coder(req).await {
                Ok(r) => {
                    consecutive_inference_failures = 0;
                    // The call SUCCEEDED, but not on the lane the operator
                    // configured — that one's credential was rejected and the
                    // chain degraded to another model. Say so once, rather than
                    // let the run look healthy on a backbone nobody chose
                    // (Parslee-ai/car#888).
                    //
                    // The JOURNAL takes every transition; the stream is
                    // latched. A durable record of a run that degraded twice
                    // needs both facts, and the latch exists only so the live
                    // stream does not narrate every routing decision inside a
                    // phase (car#1351).
                    // One row per HOP, with an honest `to`: the next candidate
                    // the chain tried, or the model that finally served. A
                    // chain of 1 -> 2 -> 3 -> 4 is three transitions, not
                    // "1 -> 4".
                    //
                    // Deduped against the PREVIOUS turn only, not against
                    // everything seen. A persistent condition — a dead
                    // credential, a permanently-absent local model — sets the
                    // same hops on every turn, and 50 byte-identical rows for
                    // one transition is a tally, not a transition log. But a
                    // lane that rate-limits at turn 3, recovers, and
                    // rate-limits again at turn 20 is a NEW episode, and a
                    // session-wide set would suppress it — leaving a journal
                    // that cannot tell "changed once, early" from "flapped six
                    // times", which is the question this record exists to
                    // answer (linus review, car#1351).
                    let hops: Vec<(String, String, &'static str)> = r
                        .fallback_from
                        .iter()
                        .enumerate()
                        .map(|(i, fb)| {
                            let to = r
                                .fallback_from
                                .get(i + 1)
                                .map(|next| next.candidate.clone())
                                .unwrap_or_else(|| r.model_used.clone());
                            (fb.candidate.clone(), to, fallback_reason_label(fb.reason))
                        })
                        .collect();
                    if hops != last_journaled_hops {
                        for (from, to, reason) in &hops {
                            sink.record_model_fallback(from, to, reason);
                        }
                        last_journaled_hops = hops;
                    }
                    if let Some(lane) = r.auth_fallback_from.clone() {
                        if !announced_model_fallback {
                            announced_model_fallback = true;
                            sink.emit(CoderEventKind::ModelFallback {
                                from: lane,
                                to: r.model_used.clone(),
                                reason: MODEL_FALLBACK_REASON.into(),
                            });
                        }
                    }
                    r
                }
                Err(e) => {
                    if let TurnGenerationError::NoEligibleModel { excluded_models } = &e {
                        // Strict exclusions are set by self-heal so a review
                        // seat cannot author the change it judges. Exhausting
                        // that set is configuration, not a failure of this
                        // backlog item, and retrying it three times would still
                        // invoke no provider. Return on the first turn with a
                        // typed cause; heal_tick uses that cause to avoid
                        // writing item backoff or a public failure comment.
                        let message = format!(
                            "no independent coder model is available: review models {} are excluded; \
                             configure an independent coder_model or review_models in heal.toml",
                            excluded_models
                        );
                        sink.emit(CoderEventKind::Error {
                            message: message.clone(),
                        });
                        sink.record_turn_completed(
                            "no_eligible_model",
                            None,
                            false,
                            turn,
                            &last_model,
                            &models_this_iteration,
                        );
                        let results = evaluate_contract_with_baselines(
                            contract,
                            executor,
                            sink,
                            &cfg.baseline_captures,
                        )
                        .await;
                        let passed = results.iter().all(|result| result.passed);
                        return if passed {
                            LoopOutcome::green(iteration, results)
                        } else {
                            LoopOutcome::lost(
                                LoopFailure::Configuration,
                                Some(message),
                                iteration,
                                results,
                            )
                        };
                    }
                    // 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) {
                        let gate = cfg.auth_gate.clone();
                        // The prompt is emitted whether or not this session can
                        // WAIT for the sign-in. Waiting needs a gate to poll,
                        // but "you need to sign in" is true either way — and the
                        // gate was wired only for a PINNED remote model
                        // (`rpc.rs` builds it from `session.model`), so on the
                        // default adaptive-routing path the operator got a
                        // generic inference failure and three strikes that
                        // discard the worktree, with nothing anywhere saying to
                        // sign in (Parslee-ai/car#888).
                        // ...but only ONCE on the ungated path, and EVERY time
                        // on the gated one. The two differ because only the
                        // gated path can resume: it waits, so a second lapse
                        // after a successful sign-in is a genuinely new event
                        // the operator has to act on again, and suppressing it
                        // would leave a resumed-then-lapsed-again run silent.
                        // The ungated path cannot resume — it falls straight
                        // through to strike counting — so three consecutive
                        // auth failures would write the same "sign in to
                        // continue" line into the transcript three times,
                        // saying nothing new after the first.
                        let announce = if gate.is_some() {
                            true
                        } else if announced_ungated_auth {
                            false
                        } else {
                            announced_ungated_auth = true;
                            true
                        };
                        if announce {
                            sink.emit(CoderEventKind::AuthRequired {
                                message: message.clone(),
                                // 0 = "not waiting" — an honest wire value, not
                                // a shortened wait.
                                wait_secs: gate.as_ref().map_or(0, |_| cfg.auth_wait.as_secs()),
                            });
                        }
                        // Deliberately no wait on the ungated path: with no gate
                        // to poll there is nothing that could observe a sign-in
                        // arriving, so waiting would just hang the run for the
                        // whole `auth_wait` window — worst of all for the user
                        // who never signed in and has no local model, who would
                        // trade a fast failure for a silent stall. Fall through
                        // to the ordinary strike counting instead.
                        if let Some(gate) = gate {
                            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;
                            }
                            // Journal the terminal BEFORE returning. This exit
                            // and the three-strike one below leave the turn loop
                            // without passing through either ordinary terminal,
                            // so every model that served this iteration was
                            // dropped with the stack frame — and a green here
                            // reaches `NeedsApproval` like any other. In
                            // iteration 1 that left the attribution empty and
                            // the self-review gate refused a change whose
                            // contract had passed; in iteration 2+ prior
                            // terminals made the set look complete while the
                            // model that wrote this iteration's edit was missing
                            // from it, which is car#1333 reached by another door.
                            sink.record_turn_completed(
                                "auth_lapsed",
                                None,
                                false,
                                turn,
                                &last_model,
                                &models_this_iteration,
                            );
                            let results = evaluate_contract_with_baselines(
                                contract,
                                executor,
                                sink,
                                &cfg.baseline_captures,
                            )
                            .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 {
                        // Same reasoning as the auth exit above: this returns
                        // without reaching either ordinary terminal, so the
                        // iteration's attribution has to be written here or it
                        // is lost. The comment below already says this path owes
                        // the ordinary one its bookkeeping; it paid memgine and
                        // skipped the journal.
                        sink.record_turn_completed(
                            "inference_failed",
                            None,
                            false,
                            turn,
                            &last_model,
                            &models_this_iteration,
                        );
                        // 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_with_baselines(
                            contract,
                            executor,
                            sink,
                            &cfg.baseline_captures,
                        )
                        .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 {
                            // A credential failure asks for a person, not an
                            // infrastructure retry. This matters most when the
                            // inference error carries both causes: the remote
                            // login failed first and the local fallback then
                            // hit its memory ceiling. The auth cause wins so
                            // `car code-task` emits `config_error` instead of
                            // the retriable `infra_inference` class
                            // (Parslee-ai/car#1248).
                            let failure = if is_auth_failure(&message) {
                                LoopFailure::NeedsAuth
                            } else {
                                LoopFailure::Infrastructure
                            };
                            LoopOutcome::lost(
                                failure,
                                Some(format!("inference failed repeatedly: {e}")),
                                iteration,
                                results,
                            )
                        };
                    }
                    continue; // retry the same turn
                }
            };
            last_model = result.model_used.clone();
            let served = result.model_used.trim();
            if !served.is_empty() && !models_this_iteration.iter().any(|m| m == served) {
                models_this_iteration.push(served.to_string());
            }

            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,
                    &models_this_iteration,
                );
                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),
                });
                // A nomination leaves the loop immediately and unjudged. The
                // loop has neither the baseline results nor the contract's
                // provenance, and handing it those so it could decide for
                // itself is exactly the shape this design refuses — see
                // `super::no_change`. A malformed call is a plain tool error
                // the model can correct, not a terminal.
                if call.name == REPORT_NO_CHANGE_TOOL && cfg.can_adjudicate_no_change {
                    match parse_nomination(&params) {
                        Ok(nomination) => {
                            return LoopOutcome::reported(
                                nomination,
                                iteration,
                                last_results.clone(),
                            );
                        }
                        Err(message) => {
                            sink.emit(CoderEventKind::ToolResult {
                                tool: call.name.clone(),
                                ok: false,
                                preview: message.clone(),
                            });
                            messages.push(Message::ToolResult {
                                tool_use_id: call.id.clone().expect("assigned above"),
                                content: message,
                                provenance: Provenance::Internal,
                            });
                            continue;
                        }
                    }
                }
                // 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),
                    // Classified per call, from the same information-flow labels
                    // the exfiltration gate reads — never from a constant.
                    //
                    // This was `Provenance::Internal` unconditionally, justified
                    // by a comment reading "the coder's tools are local … none
                    // reach the network". That was true when it was written and
                    // is exactly the kind of premise that rots: car#1073 wants a
                    // network delegate, car#1069 a browser, and the moment either
                    // lands the constant is a lie that marks fetched pages as
                    // trusted. Deriving from labels means attaching such a tool
                    // classifies it correctly with no edit here — `http_request`,
                    // `web_search` and `browser` already carry `net_send` in
                    // `builtin_tool_labels`.
                    //
                    // An unlabeled tool reads Internal, which is the same blind
                    // spot the flow gate already has (`tool_output_is_external`
                    // documents why a name-matching guess would be worse). It is
                    // shared deliberately rather than papered over here.
                    provenance: if tool_output_is_external(&call.name, &tool_labels) {
                        Provenance::External
                    } else {
                        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,
                &models_this_iteration,
            );
        } else if !model_declared_done {
            sink.record_turn_completed(
                "max_turns",
                None,
                false,
                turn,
                &last_model,
                &models_this_iteration,
            );
        }

        // Verify. The contract — not the model's self-report — decides.
        last_results =
            evaluate_contract_with_baselines(contract, executor, sink, &cfg.baseline_captures)
                .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 {

    /// The journal label table lives in this crate; the wire spelling is
    /// serde's, in another. Two tables that must agree and cannot see each
    /// other drift silently — this is what makes that a compile-time-adjacent
    /// failure instead of a subtly wrong audit record.
    #[test]
    fn the_journal_labels_match_the_serde_spelling() {
        use car_inference::FallbackReason as R;
        for r in [
            R::CredentialRejected,
            R::CredentialAbsent,
            R::RateLimited,
            R::TimedOut,
            R::Failed,
        ] {
            let serde_spelling = serde_json::to_value(r).unwrap();
            assert_eq!(
                serde_spelling.as_str(),
                Some(fallback_reason_label(r)),
                "{r:?}"
            );
        }
    }

    /// car#1071: `recall` is both advertised to the model AND dispatchable.
    ///
    /// These are separate facts since car#1078 made delegate dispatch
    /// default-closed. A tool in the prompt that the executor refuses is a
    /// worse failure than one that was never offered — the model spends a turn
    /// on it and gets `unknown tool`. The loop advertises and opens dispatch in
    /// the same place so the two sets cannot drift.
    #[test]
    fn recall_is_advertised_and_reachable_together() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();

        // Attached, but closed until a run advertises it.
        assert!(
            !exec.delegates_reachable(),
            "delegates start closed — attachment is not reachability"
        );
        let memory_defs = exec.delegate_defs_named("recall");
        assert_eq!(memory_defs.len(), 1, "recall must be attached");

        // The loop's own sequence.
        exec.advertise_delegates();
        assert!(exec.delegates_reachable(), "advertising must open dispatch");

        let mut tools = native_loop_tool_defs(false);
        tools.extend(memory_defs);
        let names: Vec<String> = tools
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert!(names.iter().any(|n| n == "recall"));
        assert!(
            names.iter().any(|n| n == "shell"),
            "built-ins still present"
        );
        // The Parslee document tools are attached but NOT advertised to a
        // coding run: advertising `recall` must not open the whole surface.
        assert!(
            !names.iter().any(|n| n.starts_with("parslee_")),
            "advertising recall must not offer the rest of the delegate surface"
        );
    }

    /// Provenance is derived, not asserted — so a network tool attached later
    /// is classified correctly with no edit to the loop.
    ///
    /// This is the property, not the current behaviour. The loop hardcoded
    /// `Provenance::Internal` on the premise that "the coder's tools are local
    /// … none reach the network". car#1073 (network), car#1069 (browser) and
    /// car#1071 (memory) each falsify that premise, and a fetched page marked
    /// Internal is attacker-controlled text wearing a trusted label — worse
    /// given car#1081, where a triaged issue may come from a public tracker.
    #[test]
    fn tool_results_are_classified_from_labels_not_from_a_constant() {
        let labels = car_engine::builtin_tool_labels();

        // Today's coder surface: local, and correctly Internal.
        for local in [
            "shell",
            "read_file",
            "write_file",
            "edit_file",
            "grep_files",
        ] {
            assert!(
                !car_engine::tool_output_is_external(local, &labels),
                "{local} is local and must not be marked External"
            );
        }

        // The tools car#1073 and car#1069 attach. They are External because
        // they carry `net_send` in the same labels the exfiltration gate reads.
        // That is what makes attaching them safe without a second list to
        // remember.
        for networked in [
            "http_request",
            "web_search",
            "browse_navigate",
            "browse_observe",
            "browser_await_answer",
        ] {
            assert!(
                car_engine::tool_output_is_external(networked, &labels),
                "{networked} reaches the network and must be marked External"
            );
        }

        // An unlabeled tool reads Internal. Stated so the blind spot is a
        // recorded decision rather than a surprise: the flow gate shares it,
        // and a name-matching guess would be wrong in both directions.
        assert!(!car_engine::tool_output_is_external(
            "some_unlabeled_tool",
            &labels
        ));
    }

    /// A caller that cannot judge a nomination must not be offered the tool.
    /// Otherwise `LoopOutcome` comes back neither green nor failed and the
    /// caller silently discards the finding as an ordinary red.
    #[test]
    fn report_no_change_is_offered_only_to_a_caller_that_can_adjudicate() {
        // Built through the same seam the loop uses (#1063), so this cannot
        // pass against a hand-rolled list that has drifted from the real one.
        let names = |cfg: &NativeLoopConfig| -> Vec<String> {
            let mut tools = native_loop_tool_defs(false);
            if cfg.can_adjudicate_no_change {
                tools.push(report_no_change_tool_def());
            }
            tools
                .iter()
                .filter_map(|t| t["name"].as_str().map(String::from))
                .collect()
        };

        let cannot = NativeLoopConfig::default();
        assert!(
            !cannot.can_adjudicate_no_change,
            "false is the only safe default"
        );
        assert!(!names(&cannot).contains(&REPORT_NO_CHANGE_TOOL.to_string()));

        let can = NativeLoopConfig {
            can_adjudicate_no_change: true,
            ..Default::default()
        };
        assert!(names(&can).contains(&REPORT_NO_CHANGE_TOOL.to_string()));
    }

    /// A mistyped `kind` must not fall back to `premise_wrong` — that is the one
    /// value which can terminate a session without a human.
    #[test]
    fn an_unknown_nomination_kind_is_rejected_not_defaulted() {
        let bad = serde_json::json!({
            "kind": "premise_wrongg",
            "summary": "s",
            "evidence": "e"
        });
        let err = parse_nomination(&bad).unwrap_err();
        assert!(err.contains("unknown kind"), "{err}");

        for missing in ["kind", "summary", "evidence"] {
            let mut obj = serde_json::json!({
                "kind": "premise_wrong", "summary": "s", "evidence": "e"
            });
            obj.as_object_mut().unwrap().remove(missing);
            assert!(
                parse_nomination(&obj).is_err(),
                "`{missing}` must be required"
            );
        }

        let ok = parse_nomination(&serde_json::json!({
            "kind": "non_code_decision", "summary": "s", "evidence": "e"
        }))
        .unwrap();
        assert_eq!(ok.kind, NoChangeKind::NonCodeDecision);
    }

    use super::*;
    use crate::coder::contract::ContractCheck;
    use crate::coder::CoderEvent;
    use std::sync::atomic::AtomicUsize;
    use std::sync::Arc;

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

    /// A coding run shows the model the static built-ins, whatever delegate the
    /// executor happens to carry — which is what makes the daemon's `coder.*`
    /// loop and headless `car code-task` equivalent now that both build their
    /// executor with `for_coder_session` (Parslee-ai/car#1063). Routing this
    /// through `executor.all_tool_defs()` would add sign-in-gated
    /// A tool operator policy denies outright is not offered to the coding
    /// model. The inspector chain refuses it either way; this stops the loop
    /// spending a schema and a turn discovering that.
    ///
    /// Asserted as a set difference against the same executor with no denies,
    /// so it cannot pass by the tool simply being absent.
    #[test]
    fn the_coding_loop_withholds_tools_operator_policy_denies() {
        fn names(defs: &[Value]) -> std::collections::BTreeSet<String> {
            defs.iter()
                .filter_map(|d| d["name"].as_str().map(String::from))
                .collect()
        }
        let dir = tempfile::tempdir().unwrap();

        let open = WorktreeExecutor::new(dir.path());
        let before = names(&coder_session_tool_defs(&open, false, false));
        assert!(
            before.contains("shell"),
            "control must offer the tool the next one denies: {before:?}"
        );

        let denied = WorktreeExecutor::new(dir.path())
            .with_denied_tools(["shell".to_string()].into_iter().collect());
        let after = names(&coder_session_tool_defs(&denied, false, false));

        assert_eq!(
            before.difference(&after).cloned().collect::<Vec<_>>(),
            vec!["shell".to_string()],
            "exactly the denied tool is withheld"
        );
        assert!(
            after.difference(&before).next().is_none(),
            "withholding must not ADD anything"
        );
    }

    /// document-generation tools to every coding prompt, and fails here.
    #[test]
    fn the_coding_loop_advertises_the_built_ins_not_the_delegate() {
        fn names(defs: &[Value]) -> Vec<String> {
            defs.iter()
                .filter_map(|d| d["name"].as_str().map(String::from))
                .collect()
        }

        let advertised = names(&native_loop_tool_defs(false));
        assert_eq!(advertised, names(&WorktreeExecutor::tool_defs()));
        assert!(
            !advertised.iter().any(|n| n.starts_with("parslee_")),
            "delegate tools leaked into the coding loop: {advertised:?}"
        );
        assert!(advertised.iter().any(|n| n == "shell"));

        // The executor a real session gets does carry the delegate — the point
        // is that it does not change what the coding model is shown.
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        assert!(names(&exec.all_tool_defs())
            .iter()
            .any(|n| n.starts_with("parslee_")));

        // `ask` is the one addition, and only when a handler is supplied.
        let with_ask = names(&native_loop_tool_defs(true));
        assert_eq!(with_ask.len(), advertised.len() + 1);
        assert_eq!(with_ask.last().unwrap(), ASK_USER_TOOL);
    }

    /// The network pair is offered only when the operator granted the tier
    /// (car#1073). Both sides of the grant, without touching the process-global
    /// permission file — which is the reason `granted` is a parameter.
    #[test]
    fn the_network_pair_is_offered_only_once_the_operator_grants_it() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();

        assert!(
            network_tool_defs(&exec, false).is_empty(),
            "an ungranted session must not be shown a tool the gate will refuse"
        );

        let granted: Vec<String> = network_tool_defs(&exec, true)
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert_eq!(granted, vec!["http_request", "web_search"]);
    }

    /// Browser availability is a real per-session switch: omission leaves the
    /// surface absent, while opt-in advertises the same lazy assistant tools
    /// and opens delegate dispatch.
    #[test]
    fn final_policy_filter_withholds_browser_tools_after_opt_in_assembly() {
        let dir = tempfile::tempdir().unwrap();
        let policy_dir = dir.path().join(".car/policies");
        std::fs::create_dir_all(&policy_dir).unwrap();
        std::fs::write(
            policy_dir.join("browser.toml"),
            "deny_tool = [\"browse_navigate\"]\n",
        )
        .unwrap();
        let executor = WorktreeExecutor::for_coder_session(dir.path())
            .unwrap()
            .with_browser_tools();
        let tools = coder_session_tool_defs(&executor, false, false);
        assert!(!tools.iter().any(|tool| tool["name"] == "browse_navigate"));
        assert!(tools.iter().any(|tool| tool["name"] == "browse_observe"));
    }

    #[test]
    fn the_coding_loop_offers_browser_tools_only_after_opt_in() {
        let dir = tempfile::tempdir().unwrap();
        let plain = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        let plain_names: Vec<String> = coder_session_tool_defs(&plain, true, true)
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert!(!plain_names.iter().any(|n| n.starts_with("browse_")));
        assert!(!plain_names.iter().any(|n| n.starts_with("browser_")));

        let enabled = WorktreeExecutor::for_coder_session(dir.path())
            .unwrap()
            .with_browser_tools();
        assert!(!enabled.delegates_reachable());
        let enabled_names: Vec<String> = coder_session_tool_defs(&enabled, true, true)
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        for required in [
            "browse_navigate",
            "browse_click",
            "browse_type",
            "browse_scroll",
            "browse_keypress",
            "browse_wait",
            "browse_observe",
            "browser_await_answer",
            "browser_await_signin",
            "browser_record_start",
            "browser_record_stop",
        ] {
            assert!(
                enabled_names.contains(&required.to_string()),
                "missing {required}"
            );
        }
        assert!(enabled.delegates_reachable());
    }

    /// 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, None);
        let with =
            system_prompt_with_overlay(&contract, "env", None, 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());
    }

    /// A repo carrying no instructions and no `.car/` must get exactly the
    /// prompt it got before this existed. Asserted rather than assumed: this is
    /// a change to every coder session's system prompt, and the no-op case is
    /// the one that must not regress.
    #[test]
    fn a_repo_with_no_project_context_gets_an_unchanged_prompt() {
        let contract = contract_for_prompt_test();
        let bare = system_prompt(&contract, "env", None);
        assert_eq!(system_prompt(&contract, "env", Some("")), bare);
        assert_eq!(system_prompt(&contract, "env", Some("  \n ")), bare);
    }

    /// The instructions reach the model, and the framing that makes them usable
    /// travels with them.
    #[test]
    fn project_instructions_reach_the_system_prompt() {
        let contract = contract_for_prompt_test();
        // Held in scope rather than leaked with `keep()` — the directory only
        // needs to outlive the call.
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(
            dir.path().join("CLAUDE.md"),
            "No cargo feature flags. Ever.",
        )
        .unwrap();
        let block = crate::coder::project_context::project_context(dir.path())
            .expect("a repo with CLAUDE.md yields a block");

        let p = system_prompt(&contract, "env", Some(&block));
        assert!(p.contains("No cargo feature flags. Ever."));
        // The rule alone is not enough — a model that reads a convention and
        // then edits a check to satisfy it has made things worse.
        assert!(p.contains("does NOT check them"));
        assert!(p.contains("never weaken or edit a contract check"));
        // Ordering: constraints land after the environment, before how-to-work.
        let env_at = p.find("ENVIRONMENT:").expect("environment section");
        let rules_at = p.find("No cargo feature flags").expect("instructions");
        let how_at = p.find("How to work:").expect("how-to-work section");
        assert!(env_at < rules_at && rules_at < how_at);
    }

    /// 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", None);
        assert_eq!(
            system_prompt_with_overlay(&contract, "env", None, None),
            base
        );
        assert_eq!(
            system_prompt_with_overlay(&contract, "env", None, Some("")),
            base
        );
        assert_eq!(
            system_prompt_with_overlay(&contract, "env", None, 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", None, 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,
            timed_out: false,
            deadline_clamped: false,
        }
    }

    /// 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,
                    baseline: false,
                    differential: None,
                }],
            },
            &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,
                baseline: false,
                differential: None,
            }],
        };
        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,
                baseline: false,
                differential: None,
            }],
        };
        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,
                baseline: false,
                differential: None,
            }],
        };
        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(
            "Parslee credential store unreadable for `parslee/reasoning`"
        ));
        assert!(is_auth_failure(
            "openai credential environment variable missing: `OPENAI_API_KEY` for explicitly requested `openai/gpt-5.6`"
        ));
        // The shape an EXPIRED (not missing) Parslee token actually takes.
        // Matched none of the hand-rolled substrings above, so it burned
        // inference strikes instead of pausing for sign-in — Parslee-ai/car#888.
        assert!(is_auth_failure(
            "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: \
             Authentication required"
        ));

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

    /// Fails every call with one fixed message. Unlike `dead_backbone`, the
    /// message is the test's — the classification under test keys on it.
    struct AlwaysFails {
        message: String,
    }

    #[async_trait]
    impl TurnGenerator for AlwaysFails {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            Err(self.message.clone())
        }
    }

    /// Runs a loop with the DEFAULT config (so `auth_gate` is `None`) against a
    /// red check, collecting every event.
    async fn run_collecting(
        gen: &dyn TurnGenerator,
        check: &str,
        cfg: NativeLoopConfig,
    ) -> (LoopOutcome, Vec<CoderEvent>) {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, collected) = EventSink::collecting("coder-auth");
        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,
                baseline: false,
                differential: None,
            }],
        };
        let outcome = run_native_loop(
            gen,
            &executor,
            "x",
            &contract,
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await;
        let events = collected.lock().expect("collector poisoned").clone();
        (outcome, events)
    }

    /// **The regression this whole change exists for.** `NativeLoopConfig`
    /// defaults `auth_gate` to `None`, and the daemon only builds a gate for an
    /// explicitly PINNED remote model — so on the default adaptive-routing path
    /// an expired token produced no `auth_required` at all: just three generic
    /// inference strikes and a discarded worktree, with nothing anywhere saying
    /// to sign in (Parslee-ai/car#888).
    ///
    /// The prompt must be emitted with no gate present, carrying `wait_secs: 0`
    /// because this path deliberately does NOT wait — with nothing to poll a
    /// sign-in against, waiting would only stall the run for the whole window —
    /// and exactly once, because a path that cannot resume has nothing new to
    /// say on the second and third strike.
    #[tokio::test]
    async fn an_ungated_auth_failure_still_asks_for_sign_in_without_waiting() {
        let gen = AlwaysFails {
            message: "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: \
                      Authentication required"
                .to_string(),
        };
        let (outcome, events) = run_collecting(&gen, "exit 1", NativeLoopConfig::default()).await;

        let prompts: Vec<(u64, String)> = events
            .iter()
            .filter_map(|e| match &e.kind {
                CoderEventKind::AuthRequired { wait_secs, message } => {
                    Some((*wait_secs, message.clone()))
                }
                _ => None,
            })
            .collect();
        // Every one of the three strikes takes the auth branch, but the ungated
        // path cannot resume — so the second and third prompts would say
        // nothing the first did not. Exactly one, not merely "at least one".
        let strikes = events
            .iter()
            .filter(|e| {
                matches!(&e.kind, CoderEventKind::Error { message }
                    if message.contains("inference failed (turn"))
            })
            .count();
        assert_eq!(strikes, 3, "the run must have burned all three strikes");
        assert_eq!(
            prompts.len(),
            1,
            "an expired credential must ask for a sign-in exactly ONCE with no auth \
             gate — one prompt across all {strikes} strikes, not one per strike: {:?}",
            events.iter().map(|e| &e.kind).collect::<Vec<_>>()
        );
        for (wait_secs, message) in &prompts {
            assert_eq!(
                *wait_secs, 0,
                "an ungated loop is not waiting; saying it is would be a lie on the wire"
            );
            assert!(message.contains("401"), "the cause must survive: {message}");
        }
        // ...and the ungated path still falls through to ordinary strike
        // counting. A wait here would hang a user who never signed in and has
        // no local model for the entire `auth_wait` window. Once those strikes
        // run out, the typed terminal must still name auth rather than generic
        // inference infrastructure.
        assert_eq!(
            outcome.failure,
            Some(LoopFailure::NeedsAuth),
            "no gate means no wait, not that an expired login becomes infrastructure"
        );
    }

    #[tokio::test]
    async fn mixed_auth_and_local_oom_is_classified_as_auth() {
        let gen = AlwaysFails {
            message: "inference failed: Parslee login expired for `parslee/reasoning` — \
                      run `car auth login`; fallback then failed: This model needs about \
                      9059 MB, beyond the configured 6553 MB local-model allocation"
                .to_string(),
        };
        let (outcome, events) = run_collecting(&gen, "exit 1", NativeLoopConfig::default()).await;

        assert_eq!(outcome.failure, Some(LoopFailure::NeedsAuth));
        let auth_message = events
            .iter()
            .find_map(|event| match &event.kind {
                CoderEventKind::AuthRequired { message, .. } => Some(message.as_str()),
                _ => None,
            })
            .expect("mixed failure must emit auth_required");
        let auth_pos = auth_message
            .find("Parslee login expired")
            .expect("credential cause must be named");
        let oom_pos = auth_message
            .find("9059 MB")
            .expect("fallback OOM must remain as secondary detail");
        assert!(auth_pos < oom_pos, "credential cause must be named first");
    }

    #[tokio::test]
    async fn genuine_local_oom_remains_inference_infrastructure() {
        let gen = AlwaysFails {
            message: "inference failed: This model needs about 9059 MB, beyond the configured \
                      6553 MB local-model allocation"
                .to_string(),
        };
        let (outcome, events) = run_collecting(&gen, "exit 1", NativeLoopConfig::default()).await;

        assert_eq!(outcome.failure, Some(LoopFailure::Infrastructure));
        assert!(events
            .iter()
            .all(|event| !matches!(&event.kind, CoderEventKind::AuthRequired { .. })));
    }

    /// Fails its first `remaining` calls with `message`, then serves `inner`.
    /// The sibling of `AuthFlaky` for the OTHER credential shape: a token that
    /// EXISTS but was REJECTED, which reaches the loop as a free-form inference
    /// error rather than car-auth's "no credential" wording, so the message has
    /// to be the test's (Parslee-ai/car#888).
    struct RejectedThenServes {
        remaining: AtomicUsize,
        message: String,
        inner: Script,
    }

    #[async_trait]
    impl TurnGenerator for RejectedThenServes {
        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(self.message.clone());
            }
            self.inner.generate(req).await
        }
    }

    /// Signs in as soon as it is asked, and counts the asks — so a test can
    /// prove the loop actually took the WAIT path rather than falling through
    /// to strike counting. Mirrors `SignsIn`'s timing: `wait_for_auth` polls
    /// before it sleeps, so a gate that is already true costs no wall clock.
    #[derive(Debug, Default)]
    struct SignsInWhenAsked {
        polls: AtomicUsize,
    }
    #[async_trait]
    impl AuthGate for SignsInWhenAsked {
        async fn is_authenticated(&self) -> bool {
            self.polls.fetch_add(1, Ordering::SeqCst);
            true
        }
    }

    /// End to end on the GATED path with the LITERAL error from
    /// Parslee-ai/car#888. `a_lapsed_credential_waits_for_sign_in_and_then_
    /// resumes` proves the mechanism, but drives it with one of the five
    /// original "no credential" substrings — so the composition this change is
    /// actually about (a real `HTTP 401 Unauthorized` + a wired gate → prompt
    /// with a real wait → resume, strikes reset) was only ever proven by
    /// reading the code.
    #[tokio::test]
    async fn a_rejected_credential_with_a_gate_waits_and_then_resumes() {
        let gate = Arc::new(SignsInWhenAsked::default());
        let cfg = NativeLoopConfig {
            auth_gate: Some(gate.clone()),
            // Non-zero, so `wait_secs` on the wire is a real window and this
            // test would catch the ungated `0` leaking onto the gated path.
            auth_wait: std::time::Duration::from_secs(5),
            ..Default::default()
        };
        let gen = RejectedThenServes {
            // One lapse, then the credential is good again.
            remaining: AtomicUsize::new(1),
            message: "inference failed: Parslee org lookup failed: HTTP 401 Unauthorized: \
                      Authentication required"
                .to_string(),
            inner: Script::new(vec![turn("done", serde_json::json!([]))]),
        };
        let (outcome, events) = run_collecting(&gen, "exit 0", cfg.clone()).await;

        let prompts: Vec<u64> = events
            .iter()
            .filter_map(|e| match &e.kind {
                CoderEventKind::AuthRequired { wait_secs, .. } => Some(*wait_secs),
                _ => None,
            })
            .collect();
        assert_eq!(
            prompts,
            vec![cfg.auth_wait.as_secs()],
            "a gated lapse must advertise the REAL wait window, not 0: {:?}",
            events.iter().map(|e| &e.kind).collect::<Vec<_>>()
        );
        assert!(
            cfg.auth_wait.as_secs() > 0,
            "the window must be non-zero or the assertion above proves nothing"
        );
        assert!(
            gate.polls.load(Ordering::SeqCst) >= 1,
            "the loop must have actually waited on the gate"
        );

        // Resumed rather than lost: the strike the lapse would otherwise have
        // burned was reset, so neither terminal auth verdict fires.
        assert!(
            outcome.passed,
            "the session must resume after sign-in, not die: {:?}",
            outcome.error
        );
        assert_eq!(
            outcome.failure, None,
            "a recovered lapse is neither NeedsAuth nor Infrastructure"
        );
    }

    /// A run that keeps working on a fallback model because the preferred lane's
    /// credential was rejected must SAY so — once. Silence is the degrade this
    /// event exists to end; repeating it every turn is the noise the guard
    /// exists to prevent (Parslee-ai/car#888).
    #[tokio::test]
    async fn a_degraded_lane_is_announced_once_per_run() {
        let degraded = |text: &str, tool_calls: serde_json::Value| {
            let mut t = turn(text, tool_calls);
            t.auth_fallback_from = Some("parslee/reasoning".to_string());
            t.model_used = "local/qwen3".to_string();
            t
        };
        // Two SUCCESSFUL calls, both carrying the degrade: turn 1 edits, turn 2
        // declares done. One announcement must cover both.
        let script = Script::new(vec![
            degraded(
                "editing",
                serde_json::json!([{
                    "id": "c1",
                    "name": "write_file",
                    "arguments": {"path": "hello.txt", "content": "hello coder"}
                }]),
            ),
            degraded("done", serde_json::json!([])),
        ]);
        let (outcome, events) =
            run_collecting(&script, "exit 0", NativeLoopConfig::default()).await;
        assert!(outcome.passed, "outcome: {outcome:?}");
        assert_eq!(
            script.prompts(),
            2,
            "both degraded turns must actually have run"
        );

        let announcements: Vec<(&str, &str)> = events
            .iter()
            .filter_map(|e| match &e.kind {
                CoderEventKind::ModelFallback { from, to, reason } => {
                    assert!(
                        reason.contains("car auth login"),
                        "the reason must name the remedy: {reason}"
                    );
                    Some((from.as_str(), to.as_str()))
                }
                _ => None,
            })
            .collect();
        assert_eq!(
            announcements,
            vec![("parslee/reasoning", "local/qwen3")],
            "exactly one announcement, naming the dead lane and the model that answered"
        );
    }

    /// 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,
                baseline: false,
                differential: None,
            }],
        };

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

    /// Even a denied browser attempt leaves the ordinary coder call/result
    /// receipt pair. Policy runs before BrowserTools, so this also proves a
    /// denial does not launch Chromium as a side effect.
    #[tokio::test]
    async fn browser_policy_denial_is_recorded_as_tool_receipts() {
        let dir = tempfile::tempdir().unwrap();
        let policies = dir.path().join(".car").join("policies");
        std::fs::create_dir_all(&policies).unwrap();
        std::fs::write(
            policies.join("browser.toml"),
            "deny_tool = [\"browse_navigate\"]\n",
        )
        .unwrap();
        let executor = WorktreeExecutor::for_coder_session(dir.path())
            .unwrap()
            .with_browser_tools();
        let collected: Arc<std::sync::Mutex<Vec<crate::coder::CoderEvent>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));
        let collector = Arc::clone(&collected);
        let emitter: crate::coder::EventEmitter = Arc::new(move |event| {
            collector.lock().unwrap().push(event);
        });
        let journal = dir.path().join("browser-receipts.events.jsonl");
        let sink = EventSink::new("coder-browser", Some(emitter), Some(journal.clone()));
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let script = Script {
            turns: vec![
                turn(
                    "checking the page",
                    serde_json::json!([{
                        "id": "browser-call",
                        "name": "browse_navigate",
                        "arguments": {"url": "https://example.com"}
                    }]),
                ),
                turn("done", serde_json::json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "receipt probe".into(),
            checks: vec![],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "inspect a page",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(outcome.passed, "outcome: {outcome:?}");

        let events = collected.lock().unwrap();
        assert!(events.iter().any(|event| matches!(
            &event.kind,
            CoderEventKind::ToolCall { tool, .. } if tool == "browse_navigate"
        )));
        assert!(events.iter().any(|event| matches!(
            &event.kind,
            CoderEventKind::ToolResult { tool, ok: false, preview }
                if tool == "browse_navigate" && preview.contains("operator policy")
        )));
        drop(events);
        drop(sink);

        let durable = car_eventlog::EventLog::load_read_only(&journal).unwrap();
        assert!(durable.events().iter().any(|event| {
            event.kind == car_eventlog::EventKind::ActionExecuting
                && event.action_id.as_deref() == Some("browse_navigate")
        }));
        assert!(durable.events().iter().any(|event| {
            event.kind == car_eventlog::EventKind::ActionFailed
                && event.action_id.as_deref() == Some("browse_navigate")
        }));
    }

    // 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,
                baseline: false,
                differential: None,
            }],
        };

        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,
                baseline: false,
                differential: None,
            }],
        };

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

    /// A backbone that dies after the edit lands still names who wrote it.
    ///
    /// Three consecutive inference failures return `green` from INSIDE the turn
    /// loop when the contract passes — the model landed every edit and then lost
    /// its backbone. That exit reached neither ordinary terminal, so the whole
    /// iteration's attribution went with the stack frame. In iteration 1 the set
    /// came back empty and the self-review gate refused a change whose contract
    /// had passed; in iteration 2+ prior terminals made the set look complete
    /// while this iteration's author was missing from it, which is the very
    /// hole car#1333 is about.
    #[tokio::test]
    async fn a_dead_backbone_after_the_edit_still_records_the_author() {
        let dir = tempfile::tempdir().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir(&workspace).unwrap();
        let executor = WorktreeExecutor::new(&workspace);
        let journal = dir.path().join("journal").join("events.jsonl");
        let sink = EventSink::new("coder-dead-backbone", None, Some(journal.clone()));
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // One turn writes the file the contract checks, then the script is
        // exhausted and every further `generate` errors — three strikes.
        let mut editing = turn(
            "creating the file",
            serde_json::json!([{
                "id": "c1",
                "name": "write_file",
                "arguments": {"path": "hello.txt", "content": "hello coder"}
            }]),
        );
        editing.model_used = "writer".to_string();
        let script = Script::new(vec![editing]);

        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,
                baseline: false,
                differential: None,
            }],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "create hello.txt containing 'hello coder'",
            &contract,
            &sink,
            &cancel,
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;
        // The worktree is asked before a loss is declared, so this is green.
        assert!(outcome.passed, "outcome: {outcome:?}");

        drop(sink);
        let log = car_eventlog::EventLog::load(&journal).unwrap();
        let ev = log
            .events()
            .iter()
            .find(|e| e.kind == car_eventlog::EventKind::TurnCompleted)
            .expect("the backbone-death exit must journal a terminal too");
        assert_eq!(
            ev.data.get("decision"),
            Some(&serde_json::json!("inference_failed"))
        );
        assert_eq!(
            ev.data.get("models_served"),
            Some(&serde_json::json!(["writer"])),
            "the model that landed the edit must be named: {:?}",
            ev.data
        );
    }

    /// A model that wrote turns inside an iteration another model finished is
    /// journaled too (car#1333).
    ///
    /// A `TurnCompleted` names only whichever model reached the terminal, so a
    /// chain that routed the editing turns to one model and the closing turn to
    /// another left the first with no record anywhere — and it could then sit on
    /// the self-review panel judging a change it had written.
    #[tokio::test]
    async fn native_loop_journals_every_model_that_served_an_iteration() {
        let dir = tempfile::tempdir().unwrap();
        let workspace = dir.path().join("workspace");
        std::fs::create_dir(&workspace).unwrap();
        let executor = WorktreeExecutor::new(&workspace);
        let journal = dir.path().join("journal").join("events.jsonl");
        let sink = EventSink::new("coder-models", None, Some(journal.clone()));
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // `writer` does the edit; the chain then serves the closing turn from
        // `finisher`, which is the only one the terminal would name.
        let mut editing = turn(
            "creating the file",
            serde_json::json!([{
                "id": "c1",
                "name": "write_file",
                "arguments": {"path": "hello.txt", "content": "hello coder"}
            }]),
        );
        editing.model_used = "writer".to_string();
        let mut closing = turn("done — file created", serde_json::json!([]));
        closing.model_used = "finisher".to_string();

        let script = Script::new(vec![editing, closing]);
        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,
                baseline: false,
                differential: None,
            }],
        };

        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(sink);
        let log = car_eventlog::EventLog::load(&journal).unwrap();
        let ev = log
            .events()
            .iter()
            .find(|e| e.kind == car_eventlog::EventKind::TurnCompleted)
            .expect("a terminal was journaled");

        // The terminal still names who reached it…
        assert_eq!(
            ev.data.get("model_id"),
            Some(&serde_json::json!("finisher"))
        );
        // …and `models_served` names everyone, in first-seen order.
        assert_eq!(
            ev.data.get("models_served"),
            Some(&serde_json::json!(["writer", "finisher"])),
            "the model that wrote the edit must be recorded: {:?}",
            ev.data
        );
    }

    #[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 workspace = dir.path().join("workspace");
        std::fs::create_dir(&workspace).unwrap();
        let executor = WorktreeExecutor::new(&workspace);
        let journal = dir.path().join("journal").join("events.jsonl");
        // Keep the private journal root separate from the executor root. On
        // elevated Windows, hardening an ancestor after workspace files exist
        // can make those inherited entries unreadable to child processes.
        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,
                baseline: false,
                differential: None,
            }],
        };

        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,
                baseline: false,
                differential: None,
            }],
        };
        let cfg = NativeLoopConfig {
            prompt_overlay: None,
            model: None,
            exclude_models: Vec::new(),
            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,
            can_adjudicate_no_change: false,
            baseline_captures: crate::coder::contract::BaselineCaptures::new(),
        };

        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()),
            exclude_models: Vec::new(),
            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,
            can_adjudicate_no_change: false,
            baseline_captures: crate::coder::contract::BaselineCaptures::new(),
        };
        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,
                baseline: false,
                differential: None,
            }],
        };

        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_and_excludes_review_models() {
        // 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 {
                exclude_models: vec!["reviewer-a".into(), "reviewer-b".into()],
                ..Default::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"
                );
                assert_eq!(
                    intent.exclude_models,
                    ["reviewer-a", "reviewer-b"],
                    "turn {n} must exclude every review-panel seat"
                );
                assert!(
                    intent.strict_exclusions,
                    "turn {n} must refuse instead of falling back to a review-panel seat"
                );
            }
        }

        // An explicit coder model is an operator decision, not an adaptive
        // candidate. Keep the configured panel list out of its IntentHint so
        // the router never silently substitutes around that pin.
        let pinned_captured = Arc::new(Mutex::new(Vec::new()));
        let pinned = CapturingGen {
            intents: pinned_captured.clone(),
            cursor: AtomicUsize::new(0),
        };
        let _ = run_native_loop(
            &pinned,
            &executor,
            "make another change",
            &contract,
            &EventSink::test_sink(),
            &Arc::new(std::sync::atomic::AtomicBool::new(false)),
            &NativeLoopConfig {
                model: Some("operator-pinned".into()),
                exclude_models: vec!["reviewer-a".into()],
                ..Default::default()
            },
            &RepairMemory::disabled(),
            None,
        )
        .await;
        assert!(pinned_captured.lock().unwrap().iter().all(|intent| {
            intent
                .as_ref()
                .is_some_and(|hint| hint.exclude_models.is_empty() && !hint.strict_exclusions)
        }));
    }

    #[tokio::test]
    async fn no_independent_coder_stops_after_one_route_attempt_as_configuration() {
        struct NoEligible {
            calls: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for NoEligible {
            async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
                panic!("native coder must use the typed generation seam")
            }

            async fn generate_coder(
                &self,
                _req: GenerateRequest,
            ) -> Result<InferenceResult, TurnGenerationError> {
                self.calls.fetch_add(1, Ordering::SeqCst);
                Err(TurnGenerationError::NoEligibleModel {
                    excluded_models: "reviewer-a, reviewer-b".into(),
                })
            }
        }

        let dir = tempfile::tempdir().unwrap();
        let generator = NoEligible {
            calls: AtomicUsize::new(0),
        };
        let outcome = run_native_loop(
            &generator,
            &WorktreeExecutor::new(dir.path()),
            "make a change",
            &OutcomeContract {
                description: "must change".into(),
                checks: vec![ContractCheck {
                    baseline: false,
                    differential: None,
                    name: "red baseline".into(),
                    command: crate::coder::test_cmds::FAIL.into(),
                    expect_exit_zero: true,
                    output_contains: None,
                    timeout_secs: 10,
                }],
            },
            &EventSink::test_sink(),
            &Arc::new(std::sync::atomic::AtomicBool::new(false)),
            &NativeLoopConfig {
                exclude_models: vec!["reviewer-a".into(), "reviewer-b".into()],
                ..Default::default()
            },
            &RepairMemory::disabled(),
            None,
        )
        .await;

        assert_eq!(generator.calls.load(Ordering::SeqCst), 1);
        assert_eq!(outcome.failure, Some(LoopFailure::Configuration));
        let error = outcome.error.expect("configuration detail");
        assert!(error.contains("reviewer-a"), "{error}");
        assert!(error.contains("reviewer-b"), "{error}");
        assert!(error.contains("heal.toml"), "{error}");
    }

    #[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,
                baseline: false,
                differential: None,
            }],
        };

        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,
                baseline: false,
                differential: None,
            }],
        };

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

    /// End-to-end: a `report_no_change` call leaves the loop immediately,
    /// carrying the nomination out UNJUDGED.
    ///
    /// The three assertions that matter are the shape of the result. It is not
    /// green, it is not a failure, and it carries the finding — the third shape
    /// #1070 exists to create. A loop that returned this as an ordinary red
    /// would put the caller straight back on the old behaviour.
    #[tokio::test]
    async fn a_nomination_exits_the_loop_unjudged() {
        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(
                    "this is already handled",
                    serde_json::json!([{
                        "id": "n1",
                        "name": REPORT_NO_CHANGE_TOOL,
                        "arguments": {
                            "kind": "premise_wrong",
                            "summary": "the handler already covers the empty case",
                            "evidence": "read handler.rs:88 and ran the suite green"
                        }
                    }]),
                ),
                // Must never be consumed: the nomination ends the run.
                turn("should not be reached", serde_json::json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "noop".into(),
            checks: vec![],
        };
        let cfg = NativeLoopConfig {
            can_adjudicate_no_change: true,
            ..Default::default()
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "fix the empty case",
            &contract,
            &sink,
            &cancel,
            &cfg,
            &RepairMemory::disabled(),
            None,
        )
        .await;

        let nomination = outcome
            .nomination
            .expect("the finding must survive out of the loop");
        assert_eq!(nomination.kind, NoChangeKind::PremiseWrong);
        assert!(nomination.summary.contains("already covers"));
        assert!(nomination.evidence.contains("handler.rs:88"));
        assert!(!outcome.passed, "no diff, so not green");
        assert!(
            outcome.failure.is_none(),
            "a nomination is not a loss — booking it as one is the whole defect"
        );
        assert_eq!(
            script.cursor.load(Ordering::SeqCst),
            1,
            "the loop kept going after the nomination"
        );
    }

    /// The same call from a caller that did NOT declare it can adjudicate must
    /// not become a nomination. It falls through to the executor as an unknown
    /// tool, and the run ends by the ordinary rules.
    #[tokio::test]
    async fn a_nomination_from_an_unprepared_caller_is_not_honoured() {
        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(
                    "declaring no change",
                    serde_json::json!([{
                        "id": "n1",
                        "name": REPORT_NO_CHANGE_TOOL,
                        "arguments": {
                            "kind": "premise_wrong",
                            "summary": "s",
                            "evidence": "e"
                        }
                    }]),
                ),
                turn("giving up", serde_json::json!([])),
            ],
            cursor: AtomicUsize::new(0),
            seen: std::sync::Mutex::new(Vec::new()),
        };
        let contract = OutcomeContract {
            description: "noop".into(),
            checks: vec![],
        };

        let outcome = run_native_loop(
            &script,
            &executor,
            "fix it",
            &contract,
            &sink,
            &cancel,
            // Default: cannot adjudicate.
            &NativeLoopConfig::default(),
            &RepairMemory::disabled(),
            None,
        )
        .await;

        assert!(
            outcome.nomination.is_none(),
            "a caller that cannot judge a nomination must never receive one"
        );
    }

    /// 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,
                baseline: false,
                differential: None,
            }],
        };
        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,
                baseline: false,
                differential: None,
            }],
        };

        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,
                baseline: false,
                differential: None,
            }],
        };
        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,
                baseline: false,
                differential: None,
            }],
        };
        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,
                timed_out: false,
                deadline_clamped: false,
            },
            CheckResult {
                name: "bad".into(),
                passed: false,
                exit_code: Some(1),
                output_tail: "assertion failed".into(),
                duration_ms: 1,
                timed_out: false,
                deadline_clamped: false,
            },
        ];
        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,
            timed_out: false,
            deadline_clamped: false,
        }];
        // 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,
                baseline: false,
                differential: None,
            }],
        };
        let p = system_prompt(&contract, "Top-level entries: Cargo.toml, src", None);
        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,
                baseline: false,
                differential: None,
            }],
        };
        let env = "Top-level entries: Cargo.toml, src\nBuild systems detected: Rust (cargo)";
        let p = system_prompt(&contract, env, None);

        // 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");
        // car#1074: the prompt used to claim a class ("push, sudo, destructive
        // operations") that the chain did not implement — `gh pr create` was
        // allowed. `coder::policy::DenyForgePublication` closes that, and the
        // prompt must name the route it actually blocks.
        assert!(
            p.contains("gh pr create") && p.contains("the runtime opens the"),
            "publication is denied by any route, and the runtime does the publishing"
        );
        assert!(
            p.contains("Read-only forge commands"),
            "the allowed half of the forge guard must be stated, not just the denied half"
        );
        // 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,
                baseline: false,
                differential: None,
            }],
        }
    }

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