car_server_core/coder/

external_loop.rs

//! Delegation to an external agentic CLI (Claude Code, Codex, Gemini).
//!
//! The CLI does the coding inside the worktree; **CAR keeps the verdict**:
//! after every invocation the runtime re-runs the outcome contract itself
//! through the policy-gated shell tool. A CLI claiming success doesn't matter
//! — the checks do.
//!
//! When the daemon's MCP listener is bound, its URL is threaded into
//! [`InvokeOptions::mcp_endpoint`] so the CLI's **CAR-namespace** tool calls
//! (`memory_*`, `verify`, `skill_*`) route back through car-server's policy +
//! memgine — gated and audited. The CLI's own **built-in** tools (Edit, Bash)
//! still run with the CLI's permissions inside the worktree: that is the
//! residual Phase 2 stage-4b upstream limitation. The pinned `cwd`, the
//! contract re-evaluation, and the merge approval gate remain the containment
//! for those built-ins until tool round-trip governance lands in
//! `car-external-agents`.

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

use async_trait::async_trait;
use car_external_agents::{InvokeError, InvokeOptions, InvokeResult, StreamEventEmitter};

use super::budget::SessionDeadline;
use super::contract::{evaluate_contract_with_baselines, CheckResult, OutcomeContract};
use super::native_loop::{
    primary_failure, record_recurrence, recurrence_notice, LoopFailure, LoopOutcome,
};
use super::session::{CancelFlag, CoderEventKind, EventSink};
use super::shell_tool::WorktreeExecutor;

/// Tuning for external delegation.
#[derive(Debug, Clone)]
pub struct ExternalLoopConfig {
    /// Per-invocation model-turn cap (maps to the CLI's `--max-turns`).
    pub max_turns: Option<u32>,
    /// Per-invocation wall-clock budget (runner clamps to 1h).
    pub timeout_secs: Option<u64>,
    /// Fresh repair invocations after a red first pass (the stream-json
    /// protocol has no session resume yet, so repairs re-state the task plus
    /// the failing-check output).
    ///
    /// This is the **hypothesis** budget: each one buys another attempt at
    /// being right. It is deliberately not spent on transport failures — see
    /// `transient_retries`.
    ///
    /// **Defaults to 2 because recurrence escalation needs it.** Round 1
    /// establishes a failure signature, round 2 is the first that can repeat it,
    /// and only round 3 can be told it did. At 1 the loop still *detects* the
    /// repeat, but the session ends before the feedback carrying that news
    /// reaches anyone — and since `rpc` is the only construction site and takes
    /// `..Default::default()`, a default of 1 made the escalation unreachable in
    /// every shipped configuration.
    ///
    /// The cost is smaller than it looks: worst-case invocations are
    /// `max_hypotheses + transient_retries`, so this moves 3 -> 4, about +33%,
    /// and only on sessions that are already failing.
    pub repair_invokes: u32,
    /// Re-invocations after the CLI process itself died mid-run (timeout or
    /// I/O) with the contract still red.
    ///
    /// Separate from `repair_invokes` because it buys a different thing: an
    /// **availability** retry, not a new hypothesis. Sharing one counter means a
    /// single flaky timeout eats a replan the coder needed for an actual
    /// hypothesis — the difference between a session that recovers and one that
    /// silently gives up on a task it was about to finish.
    pub transient_retries: u32,
    /// Pin the external CLI's backbone (`coder.start`'s `model`). `None` = the
    /// CLI's own default.
    ///
    /// The paired A/B only measures the *harness* when both arms share a
    /// backbone; this is the external half of that invariant (the native half is
    /// `NativeLoopConfig.model`). Before this existed the pin reached only the
    /// native loop, so "both arms on gpt-5.5" was an unverified assumption.
    pub model: Option<String>,
    /// The session's absolute deadline, SHARED with every other rung of the
    /// fallback ladder. See [`super::budget`] for why this is a handle and not
    /// a value.
    pub deadline: Arc<SessionDeadline>,
    /// The session-start baseline captures differential checks compare against
    /// (car#1067). Empty when the contract declares none; differential checks
    /// fail closed under an empty map.
    pub baseline_captures: super::contract::BaselineCaptures,
}

impl Default for ExternalLoopConfig {
    fn default() -> Self {
        Self {
            max_turns: Some(50),
            timeout_secs: Some(1800),
            repair_invokes: 2,
            transient_retries: 1,
            model: None,
            deadline: SessionDeadline::shared_default(),
            baseline_captures: super::contract::BaselineCaptures::new(),
        }
    }
}

/// The CLI seam: one invocation of an external agent.
///
/// Exists for the same reason `native_loop` takes a `&dyn TurnGenerator` — the
/// loop's interesting behavior (budget accounting, retry-vs-replan, round
/// counting) is decided by what comes back from here, and none of it is
/// testable while the call is hard-wired to a real subprocess.
#[async_trait]
pub trait CliInvoker: Send + Sync {
    async fn invoke(
        &self,
        agent_id: &str,
        task: &str,
        opts: InvokeOptions,
        emitter: StreamEventEmitter,
    ) -> Result<InvokeResult, InvokeError>;
}

/// The production invoker: a real external CLI subprocess.
pub struct LiveInvoker;

#[async_trait]
impl CliInvoker for LiveInvoker {
    async fn invoke(
        &self,
        agent_id: &str,
        task: &str,
        opts: InvokeOptions,
        emitter: StreamEventEmitter,
    ) -> Result<InvokeResult, InvokeError> {
        car_external_agents::invoke_with_emitter(agent_id, task, opts, Some(emitter)).await
    }
}

/// The task text handed to the CLI: intent + contract + ground rules.
fn build_task(intent: &str, contract: &OutcomeContract, feedback: Option<&str>) -> String {
    let mut task = format!(
        "{intent}\n\n\
         OUTCOME CONTRACT — your work is verified by running these checks at the repository \
         root; all must pass:\n{}\n\
         Ground rules:\n\
         - Work only inside the current directory (an isolated git worktree).\n\
         - Do NOT git commit, push, or touch remotes; the runtime owns version control.\n\
         - Run the checks yourself before finishing.\n",
        contract.render()
    );
    if let Some(fb) = feedback {
        task.push_str(&format!(
            "\nA previous attempt left these checks FAILING — fix the code so they pass:\n{fb}"
        ));
    }
    task
}

/// The per-invocation options handed to the runner. The `mcp_endpoint`, when
/// set, routes the CLI's CAR-namespace tool calls through the daemon's policy +
/// memgine; `allowed_tools: None` leaves the CLI's own built-in tools on their
/// default (ungoverned) policy — see the module docs.
fn build_invoke_opts(
    executor: &WorktreeExecutor,
    cfg: &ExternalLoopConfig,
    mcp_endpoint: Option<&str>,
) -> InvokeOptions {
    InvokeOptions {
        cwd: Some(executor.worktree().to_path_buf()),
        allowed_tools: None, // the CLI's default policy; see module docs
        max_turns: cfg.max_turns,
        // Clamped to what the SESSION has left, not just this invocation's own
        // budget. Admission alone grants a whole round, so a hypothesis let in
        // just under the ceiling could otherwise run its full 1800s past it —
        // a ceiling exceeded by 50% is not a ceiling. This is not interruption:
        // the round simply starts with a shorter clock, and a CLI that hits its
        // own timeout already flows through `Infrastructure` ->
        // `evaluate_contract`, so nothing goes unjudged.
        timeout_secs: match (cfg.timeout_secs, cfg.deadline.remaining_secs()) {
            (Some(own), Some(left)) => Some(own.min(left)),
            (own, None) => own,
            (None, left) => left,
        },
        // The external half of the A/B's same-backbone invariant.
        model: cfg.model.clone(),
        // Gate + audit the CLI's CAR-namespace tool calls through the daemon
        // when its MCP listener is bound; None degrades cleanly.
        mcp_endpoint: mcp_endpoint.map(String::from),
        ..Default::default()
    }
}

/// Run the external engine to completion, cancellation, or exhaustion.
///
/// The `cancel` flag is checked between invocations only, but an in-flight CLI
/// is NOT left running: `rpc::cancel_session` aborts the task handle, which
/// drops this future along with the `Child`, and every adapter sets
/// `kill_on_drop(true)` (with a Windows `JobObject` for the Node grandchildren).
/// Enforcement lives one level up; the flag here is belt-and-braces, which is
/// why threading `invoke_with_emitter_and_cancel` through [`CliInvoker`] would
/// be tidier rather than more correct. The classification
/// below already handles [`LoopFailure::Cancelled`] as its own terminal so that
/// change does not need to revisit the control flow.
///
/// ## Why the contract is evaluated before a failure is classified
///
/// Every path that got as far as launching the CLI evaluates the contract,
/// including one that ended in a timeout or a broken stream. A 30-minute
/// timeout that fires after the CLI has already edited fifteen files says
/// nothing about whether those edits satisfy the contract — and a loop that
/// returns terminal without asking has let the *transport* pronounce the
/// verdict, which is precisely what this module exists to prevent. The state
/// under judgement is the worktree, not the process that was writing to it.
///
/// Only two conditions skip evaluation, both because no work can exist yet:
/// the engine never started ([`LoopFailure::EngineUnavailable`]) and the user
/// cancelled ([`LoopFailure::Cancelled`]).
pub async fn run_external_loop(
    invoker: &dyn CliInvoker,
    agent_id: &str,
    intent: &str,
    contract: &OutcomeContract,
    executor: &WorktreeExecutor,
    sink: &Arc<EventSink>,
    cancel: &CancelFlag,
    cfg: &ExternalLoopConfig,
    // Daemon MCP URL, when bound. Routes the CLI's CAR-namespace tool calls
    // through the daemon's policy + memgine. `None` degrades cleanly.
    mcp_endpoint: Option<&str>,
) -> LoopOutcome {
    let max_hypotheses = 1 + cfg.repair_invokes;
    let mut feedback: Option<String> = None;
    let mut last_results = Vec::new();
    // Hypotheses spent. Bumped only by a replan, so a transport retry buys
    // another invocation without costing an attempt at being right.
    let mut hypothesis = 1u32;
    let mut transient_budget = cfg.transient_retries;
    // Contract-evaluation rounds — what `LoopOutcome.iterations` means to its
    // real consumer, `session.iterations` as reported by `coder.get`. A retry
    // evaluates the contract, so it counts here even though it costs no
    // hypothesis. (`ab::ArmOutcome.iterations` documents the same meaning but
    // `coder_ab` hardcodes 0 today — `car code` emits no machine-readable count.)
    let mut rounds = 0u32;
    // Set when the previous pass was a transport retry: the hypothesis banner
    // must not fire twice for one hypothesis.
    let mut retrying = false;
    // Failure signatures seen across hypotheses, so a repair that lands the
    // identical failure is told so rather than handed the same text again.
    let mut seen_sigs: HashMap<String, u32> = HashMap::new();
    // This loop's clock starts here, before the first invocation.

    // Metered spend across every invocation. `None` until something reports a
    // figure, so "unmetered" stays distinguishable from "$0.00".
    let mut spent_usd: Option<f64> = None;

    loop {
        if cancel.load(Ordering::SeqCst) {
            return LoopOutcome::lost(
                LoopFailure::Cancelled,
                Some("cancelled".into()),
                rounds,
                last_results,
            )
            .with_cost(spent_usd);
        }
        // Admission, not interruption. A retry is admitted too — it is the same
        // hypothesis re-run, and denying only fresh hypotheses would let a
        // flaky CLI run past the ceiling indefinitely.
        if let Some(reason) = cfg.deadline.admit() {
            sink.emit(CoderEventKind::BudgetExhausted {
                reason: reason.clone(),
                elapsed_secs: cfg.deadline.elapsed_secs(),
                iterations: rounds,
            });
            return LoopOutcome::lost(
                LoopFailure::BudgetExhausted,
                Some(reason),
                rounds,
                last_results,
            )
            .with_cost(spent_usd);
        }
        if !retrying {
            sink.emit(CoderEventKind::IterationStarted {
                n: hypothesis,
                max: max_hypotheses,
            });
        }
        retrying = false;

        let task = build_task(intent, contract, feedback.as_deref());
        let opts = build_invoke_opts(executor, cfg, mcp_endpoint);

        let emitter_sink = sink.clone();
        let emitter: StreamEventEmitter = Arc::new(move |event| {
            if let Ok(raw) = serde_json::to_value(&event) {
                emitter_sink.emit(CoderEventKind::ExternalEvent { raw });
            }
        });

        // What the invocation itself reported, before the contract has spoken.
        // `Ok(None)` = the CLI ran clean; `Ok(Some(msg))` = it ran and reported
        // its own error; `Err(class)` = it died, and how.
        let invocation: Result<Option<String>, (LoopFailure, String)> =
            match invoker.invoke(agent_id, &task, opts, emitter).await {
                Ok(result) if result.is_error => {
                    record_spend(&mut spent_usd, result.total_cost_usd);
                    let msg = result.error.unwrap_or_else(|| "unknown".into());
                    sink.emit(CoderEventKind::Error {
                        message: format!("external agent '{agent_id}' reported an error: {msg}"),
                    });
                    Ok(Some(msg))
                }
                Ok(result) => {
                    record_spend(&mut spent_usd, result.total_cost_usd);
                    Ok(None)
                }
                Err(e) => Err((classify_invoke_error(&e), e.to_string())),
            };

        // Two failures make evaluation meaningless rather than merely
        // unnecessary: nothing ran, so no work can exist to judge. Matched
        // variant-by-variant — a wildcard here would silently hand a future
        // variant the string `"cancelled"` and the wrong terminal state.
        let terminal = match &invocation {
            // Prefix kept verbatim: `rpc` branches on the typed variant now, but
            // `car-cli`'s A/B still scrapes this text out-of-process.
            Err((LoopFailure::EngineUnavailable, msg)) => Some((
                LoopFailure::EngineUnavailable,
                format!("external agent '{agent_id}' failed: {msg}"),
            )),
            Err((LoopFailure::Cancelled, _)) => {
                Some((LoopFailure::Cancelled, "cancelled".to_string()))
            }
            // `classify_invoke_error` produces none of these, but naming them
            // keeps the match wildcard-free so a new variant fails to compile
            // rather than silently acquiring a terminal it does not mean.
            // `NeedsAuth` is native-loop-only: the external arm's credentials
            // belong to the CLI it shells, so CAR has nothing to re-authenticate
            // on its behalf and no standing to pause the run waiting for it.
            Err((LoopFailure::Infrastructure, _))
            | Err((LoopFailure::Configuration, _))
            | Err((LoopFailure::NeedsAuth, _))
            | Err((LoopFailure::Execution, _))
            | Err((LoopFailure::Verification, _))
            | Err((LoopFailure::BudgetExhausted, _))
            | Ok(_) => None,
        };
        if let Some((failure, error)) = terminal {
            return LoopOutcome::lost(failure, Some(error), rounds, last_results)
                .with_cost(spent_usd);
        }

        // CAR's verdict, not the CLI's — and not the transport's.
        last_results =
            evaluate_contract_with_baselines(contract, executor, sink, &cfg.baseline_captures)
                .await;
        rounds += 1;
        if last_results.iter().all(|r| r.passed) {
            return LoopOutcome::green(rounds, last_results).with_cost(spent_usd);
        }

        // Red. Now — and only now — the failure has a class.
        let failure = match &invocation {
            Err((class, _)) => *class,
            Ok(Some(_)) => LoopFailure::Execution,
            Ok(None) => LoopFailure::Verification,
        };
        if failure == LoopFailure::Infrastructure && transient_budget > 0 {
            // Availability, not correctness: re-invoke against the worktree as
            // it now stands, carrying the failing checks so the retry is
            // better-informed than the attempt it replaces.
            transient_budget -= 1;
            retrying = true;
            sink.emit(CoderEventKind::InvocationRetried {
                hypothesis,
                reason: match &invocation {
                    Err((_, msg)) => msg.clone(),
                    Ok(_) => String::new(),
                },
                retries_remaining: transient_budget,
            });
            continue;
        }

        if hypothesis >= max_hypotheses {
            // An exhausted Infrastructure failure must still LOOK like one.
            // `car-cli`'s A/B splits infra out of the scored denominator by
            // scraping this string (`coder_ab::INFRA_MARKERS`); leaving it
            // `None` let `rpc` substitute "contract not satisfied after N
            // iteration(s)", which scores a dead transport as a genuine task
            // loss and quietly biases the arm it belongs to.
            let error = match (failure, &invocation) {
                (LoopFailure::Infrastructure, Err((_, msg))) => {
                    Some(format!("external agent '{agent_id}' failed: {msg}"))
                }
                _ => None,
            };
            // Returns BEFORE the feedback below is built: nothing will read it,
            // and rendering it means formatting every failing check's 4KB tail
            // on the last round of every failing session. Ordering carries the
            // invariant so a reader need not hold it: feedback is only built for
            // a round that will actually happen.
            return LoopOutcome::lost(failure, error, rounds, last_results).with_cost(spent_usd);
        }

        // Another round WILL happen, so build its handoff.
        let check_feedback = render_check_failures(&last_results);
        feedback = Some(match &invocation {
            // The CLI's own error is context the checks cannot supply.
            Ok(Some(msg)) => {
                format!("A previous attempt reported this error:\n{msg}\n\n{check_feedback}")
            }
            Err((_, msg)) => format!(
                "A previous attempt was cut short ({msg}); its work may be partially applied.\n\n\
                 {check_feedback}"
            ),
            // Only a clean run earns a recurrence. NOT because the other cases
            // were "cut short" — an `is_error` CLI may well have run to
            // completion — but because `InvokeResult.is_error` is a
            // heterogeneous bucket: it covers a non-zero exit, an empty answer,
            // and "produced no agent_message" alike, so it cannot distinguish
            // "hit a real wall" from "never produced anything". Counting the
            // latter would inflate the tally against an attempt that did not
            // happen. Accepted cost: a signature first seen on an `Execution`
            // round is never recorded, so its count stays one low all session.
            //
            // The count is computed here, in the one arm that consumes it, so
            // the policy is stated once. Hoisting it into a separate `if`
            // duplicates this condition and silently zeroes any escalation a
            // future arm might add.
            Ok(None) => {
                match record_recurrence(&mut seen_sigs, primary_failure(&last_results).as_ref()) {
                    0 => check_feedback,
                    n => format!("{check_feedback}\n\n{}", recurrence_notice(n)),
                }
            }
        });

        hypothesis += 1;
    }
}

/// Map a transport-level error onto the outcome it implies.
///
/// The dividing question is whether the agent ever received the task, because
/// that is what decides if any work can exist to judge:
/// - `Spawn` / `Setup` — the process never started, or started and never got
///   the prompt (pipes, stdin, MCP config). No work exists; another engine may
///   be tried.
/// - `Timeout` / `Io` — the agent had the task and the run died underneath it.
///   Edits may be on disk, so the contract gets consulted and the same
///   hypothesis may be retried.
/// - `Cancelled` — the human stopped it. Never substitute another engine.
///
/// `Setup` exists because most of what used to be `Io` was this case: writing
/// the MCP tempfile, acquiring stdout, delivering the prompt. Calling those
/// retryable meant re-running a full contract evaluation against an untouched
/// worktree and then declining the fallback that used to fire.
fn classify_invoke_error(e: &car_external_agents::InvokeError) -> LoopFailure {
    use car_external_agents::InvokeError as E;
    match e {
        E::Spawn(_) | E::Setup(_) => LoopFailure::EngineUnavailable,
        E::Timeout(_) | E::Io(_) => LoopFailure::Infrastructure,
        E::Cancelled => LoopFailure::Cancelled,
    }
}

/// Fold one invocation's reported spend into the session total.
///
/// Stays `None` until a provider actually reports a figure, so a native run (or
/// a CLI that reports nothing) is recorded as *unknown* rather than as $0.00 —
/// the conflation that made `ab::ArmOutcome.cost_usd` a published zero.
/// Non-finite or negative figures are ignored rather than allowed to poison the
/// total.
fn record_spend(total: &mut Option<f64>, reported: Option<f64>) {
    let Some(usd) = reported else { return };
    if !usd.is_finite() || usd < 0.0 {
        return;
    }
    *total = Some(total.unwrap_or(0.0) + usd);
}

/// The failing half of a contract evaluation, rendered for a model to act on.
fn render_check_failures(results: &[CheckResult]) -> String {
    results
        .iter()
        .filter(|r| !r.passed)
        .map(|r| {
            format!(
                "FAILED {} (exit {:?}):\n{}",
                r.name, r.exit_code, r.output_tail
            )
        })
        .collect::<Vec<_>>()
        .join("\n\n")
}

#[cfg(test)]
mod tests {
    use std::collections::VecDeque;
    use std::sync::atomic::AtomicU32;
    use std::sync::Mutex;

    use super::*;
    use crate::coder::contract::ContractCheck;
    use crate::coder::session::CoderEvent;
    // Checks run through the coder's shell — `sh -lc` on Unix, `cmd /C` on
    // Windows — so fixtures use the portable builders rather than POSIX
    // literals. `true` is not a program on Windows (car#760).
    use crate::coder::test_cmds::PASS;

    fn contract() -> OutcomeContract {
        OutcomeContract {
            description: "x".into(),
            checks: vec![ContractCheck {
                name: "tests".into(),
                command: "cargo test".into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 300,
                baseline: false,
                differential: None,
            }],
        }
    }

    #[test]
    fn task_carries_intent_contract_and_ground_rules() {
        let t = build_task("add a CLI flag", &contract(), None);
        assert!(t.contains("add a CLI flag"));
        assert!(t.contains("cargo test"));
        assert!(t.contains("Do NOT git commit"));
        assert!(!t.contains("FAILING"));
    }

    #[test]
    fn repair_task_carries_failure_feedback() {
        let t = build_task("x", &contract(), Some("FAILED tests (exit Some(1)):\nboom"));
        assert!(t.contains("previous attempt"));
        assert!(t.contains("boom"));
    }

    #[test]
    fn mcp_endpoint_is_threaded_into_invoke_opts() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let cfg = ExternalLoopConfig::default();
        let opts = build_invoke_opts(&executor, &cfg, Some("http://127.0.0.1:9102/mcp"));
        assert_eq!(
            opts.mcp_endpoint.as_deref(),
            Some("http://127.0.0.1:9102/mcp")
        );
        // The CLI's own built-in tools stay on the default policy.
        assert!(opts.allowed_tools.is_none());
    }

    #[test]
    fn absent_mcp_endpoint_degrades_to_none() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let cfg = ExternalLoopConfig::default();
        let opts = build_invoke_opts(&executor, &cfg, None);
        assert!(opts.mcp_endpoint.is_none());
    }

    // --- Failure classification ------------------------------------------

    /// Pins the mapping. Note this is an array of literals, NOT the exhaustive
    /// guard — a new upstream variant would compile straight past it. What
    /// actually forces the decision is the wildcard-free match in
    /// [`classify_invoke_error`], which fails to compile instead.
    #[test]
    fn every_invoke_error_maps_to_its_outcome() {
        use car_external_agents::InvokeError as E;
        let cases = [
            (E::Spawn("no binary".into()), LoopFailure::EngineUnavailable),
            // Pre-handoff I/O: the agent never received the task, so this is
            // `Spawn`'s neighbour, not a retryable mid-run fault.
            (
                E::Setup("stdin closed".into()),
                LoopFailure::EngineUnavailable,
            ),
            (E::Timeout(1800), LoopFailure::Infrastructure),
            (E::Io("stdout read".into()), LoopFailure::Infrastructure),
            (E::Cancelled, LoopFailure::Cancelled),
        ];
        for (err, want) in cases {
            assert_eq!(classify_invoke_error(&err), want, "{err}");
        }
    }

    /// The distinction the `Setup` split exists for: both are I/O, but one left
    /// a worktree worth evaluating and the other could not have.
    #[test]
    fn setup_and_midrun_io_are_not_the_same_outcome() {
        use car_external_agents::InvokeError as E;
        assert_ne!(
            classify_invoke_error(&E::Setup("stdin closed".into())),
            classify_invoke_error(&E::Io("stdout read".into())),
        );
    }

    // --- Loop behavior, against a scripted CLI ----------------------------

    /// A CLI whose every invocation is scripted, so the loop's budgets, round
    /// counting and retry policy are observable without a subprocess.
    struct ScriptedInvoker {
        script: Mutex<VecDeque<Result<InvokeResult, InvokeError>>>,
        calls: AtomicU32,
        /// Every task text handed over, in order — the only place the repair
        /// feedback is observable from outside the loop.
        tasks: Mutex<Vec<String>>,
    }

    impl ScriptedInvoker {
        fn new(script: Vec<Result<InvokeResult, InvokeError>>) -> Self {
            Self {
                script: Mutex::new(script.into()),
                calls: AtomicU32::new(0),
                tasks: Mutex::new(Vec::new()),
            }
        }
        fn calls(&self) -> u32 {
            self.calls.load(Ordering::SeqCst)
        }
        fn task(&self, n: usize) -> String {
            self.tasks.lock().expect("tasks poisoned")[n].clone()
        }
    }

    #[async_trait]
    impl CliInvoker for ScriptedInvoker {
        async fn invoke(
            &self,
            _agent_id: &str,
            task: &str,
            _opts: InvokeOptions,
            _emitter: StreamEventEmitter,
        ) -> Result<InvokeResult, InvokeError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            self.tasks
                .lock()
                .expect("tasks poisoned")
                .push(task.to_string());
            self.script
                .lock()
                .expect("script poisoned")
                .pop_front()
                .unwrap_or_else(|| Err(InvokeError::Spawn("script exhausted".into())))
        }
    }

    /// A contract whose single check always fails / always passes, cheaply.
    fn contract_with(command: &str) -> OutcomeContract {
        OutcomeContract {
            description: "x".into(),
            checks: vec![ContractCheck {
                name: "gate".into(),
                command: command.into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 30,
                baseline: false,
                differential: None,
            }],
        }
    }

    fn clean_run() -> Result<InvokeResult, InvokeError> {
        Ok(InvokeResult::default())
    }

    fn errored_run(msg: &str) -> Result<InvokeResult, InvokeError> {
        Ok(InvokeResult {
            is_error: true,
            error: Some(msg.into()),
            ..Default::default()
        })
    }

    async fn run(
        invoker: &dyn CliInvoker,
        contract: &OutcomeContract,
        cfg: &ExternalLoopConfig,
    ) -> (LoopOutcome, Vec<CoderEvent>) {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());
        let (sink, collected) = EventSink::collecting("t");
        let sink = Arc::new(sink);
        let cancel: CancelFlag = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let outcome = run_external_loop(
            invoker, "codex", "x", contract, &executor, &sink, &cancel, cfg, None,
        )
        .await;
        let events = collected.lock().unwrap().clone();
        (outcome, events)
    }

    /// **The regression test for the core defect.** A dead transport must not
    /// be able to fail a session whose worktree already satisfies the contract.
    /// Before classification moved after evaluation, this returned an error.
    #[tokio::test]
    async fn a_timeout_over_green_checks_still_passes() {
        let invoker = ScriptedInvoker::new(vec![Err(InvokeError::Timeout(1800))]);
        let (outcome, _) = run(
            &invoker,
            &contract_with(PASS),
            &ExternalLoopConfig::default(),
        )
        .await;
        assert!(outcome.passed, "the contract, not the transport, decides");
        assert_eq!(outcome.failure, None);
        assert_eq!(outcome.iterations, 1);
    }

    /// A transient retry buys an invocation without spending a hypothesis.
    /// Budgets: 1 + repair_invokes(1) hypotheses, transient_retries(1) retries
    /// => exactly 3 invocations, and 3 contract-evaluation rounds.
    #[tokio::test]
    async fn a_transient_retry_does_not_spend_a_hypothesis() {
        let invoker = ScriptedInvoker::new(vec![
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
        ]);
        let (outcome, events) = run(
            &invoker,
            &contract_with("exit 1"),
            &ExternalLoopConfig::default(),
        )
        .await;
        assert_eq!(invoker.calls(), 4, "3 hypotheses + 1 transient retry");
        assert_eq!(outcome.iterations, 4, "every invocation evaluated");
        // The retry gets its own event, and does NOT re-fire the hypothesis
        // banner — otherwise `iteration 1/2` would print twice for one attempt.
        let started = events
            .iter()
            .filter(|e| matches!(e.kind, CoderEventKind::IterationStarted { .. }))
            .count();
        let retried = events
            .iter()
            .filter(|e| matches!(e.kind, CoderEventKind::InvocationRetried { .. }))
            .count();
        assert_eq!(started, 3, "one banner per hypothesis");
        assert_eq!(retried, 1);
    }

    /// Exhausting the transient budget must still LOOK infrastructural, or
    /// `car-cli`'s A/B scores a dead transport as a genuine task loss and
    /// biases the arm's pass rate.
    #[tokio::test]
    async fn exhausted_infrastructure_keeps_the_scraped_error_prefix() {
        let invoker = ScriptedInvoker::new(vec![
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
        ]);
        let (outcome, _) = run(
            &invoker,
            &contract_with("exit 1"),
            &ExternalLoopConfig::default(),
        )
        .await;
        assert_eq!(outcome.failure, Some(LoopFailure::Infrastructure));
        let err = outcome
            .error
            .expect("an exhausted infra failure must still surface as infra");
        assert!(
            err.starts_with("external agent '"),
            "car-cli INFRA_MARKERS depends on this prefix: {err}"
        );
    }

    /// Setup failures never reach the contract: nothing ran, so there is
    /// nothing to evaluate, and the caller gets its fallback immediately.
    #[tokio::test]
    async fn a_setup_failure_does_not_retry_or_evaluate() {
        let invoker = ScriptedInvoker::new(vec![Err(InvokeError::Setup("stdout missing".into()))]);
        let (outcome, _) = run(
            &invoker,
            &contract_with("exit 1"),
            &ExternalLoopConfig::default(),
        )
        .await;
        assert_eq!(invoker.calls(), 1, "no retry: nothing ran");
        assert_eq!(outcome.iterations, 0, "the contract was never consulted");
        assert_eq!(outcome.failure, Some(LoopFailure::EngineUnavailable));
        assert!(outcome.error.unwrap().starts_with("external agent '"));
    }

    /// A CLI that ran and reported its own error is `Execution`; one that ran
    /// clean and simply got it wrong is `Verification`. Same action today, but
    /// the repair feedback differs — only `Execution` has the CLI's own error
    /// to pass back.
    #[tokio::test]
    async fn execution_and_verification_are_distinguished() {
        let cfg = ExternalLoopConfig {
            repair_invokes: 0,
            ..Default::default()
        };
        let (errored, _) = run(
            &ScriptedInvoker::new(vec![errored_run("tool denied")]),
            &contract_with("exit 1"),
            &cfg,
        )
        .await;
        assert_eq!(errored.failure, Some(LoopFailure::Execution));
        assert!(errored.error.is_none(), "the CLI ran; this is a task loss");

        let (clean, _) = run(
            &ScriptedInvoker::new(vec![clean_run()]),
            &contract_with("exit 1"),
            &cfg,
        )
        .await;
        assert_eq!(clean.failure, Some(LoopFailure::Verification));
    }

    /// A repair that lands the IDENTICAL failure is told so, rather than handed
    /// the same feedback text a second time. Before this, every repair round
    /// re-sent the failing checks verbatim with no signal that the previous
    /// attempt had changed nothing.
    #[tokio::test]
    async fn a_repeated_failure_escalates_the_repair_feedback() {
        // Needs three hypotheses, not the default two: round 1 establishes the
        // signature, round 2 is the first that can REPEAT it, and only round 3
        // can be told. See `repair_invokes` on why the default cannot escalate.
        let cfg = ExternalLoopConfig {
            repair_invokes: 2,
            ..Default::default()
        };
        let invoker = ScriptedInvoker::new(vec![clean_run(), clean_run(), clean_run()]);
        let (outcome, _) = run(&invoker, &contract_with("exit 1"), &cfg).await;
        assert_eq!(invoker.calls(), 3);
        assert_eq!(outcome.failure, Some(LoopFailure::Verification));

        // Rounds 1 and 2: nothing has repeated yet from the model's side.
        assert!(!invoker.task(0).contains("failed the same way"));
        assert!(!invoker.task(1).contains("failed the same way"));
        // Round 3: round 2 reproduced round 1's signature exactly. Say so.
        let repair = invoker.task(2);
        assert!(repair.contains("failed the same way 2 times"), "{repair}");
        assert!(repair.contains("DIFFERENT hypothesis"));
    }

    /// A transport failure must NOT escalate: the attempt was cut short before
    /// it could have changed the outcome, so a repeated check result says
    /// nothing about the hypothesis.
    #[tokio::test]
    async fn a_cut_short_attempt_does_not_escalate() {
        let invoker = ScriptedInvoker::new(vec![
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
            Err(InvokeError::Timeout(1)),
        ]);
        let (_, _) = run(
            &invoker,
            &contract_with("exit 1"),
            &ExternalLoopConfig::default(),
        )
        .await;
        for n in 0..invoker.calls() as usize {
            assert!(
                !invoker.task(n).contains("failed the same way"),
                "a timeout is not evidence about the hypothesis (task {n})"
            );
        }
    }

    /// An exhausted session budget denies the FIRST admission — before any CLI
    /// is invoked — and is its own terminal, not a task loss. Conflating it with
    /// `Verification` would teach the recurrence machinery that an approach
    /// failed when it was merely cut off.
    #[tokio::test]
    async fn an_exhausted_budget_denies_admission_before_invoking() {
        let cfg = ExternalLoopConfig {
            // A deadline that is already spent.
            deadline: std::sync::Arc::new(SessionDeadline::new(Some(0))),
            ..Default::default()
        };
        let invoker = ScriptedInvoker::new(vec![clean_run()]);
        let (outcome, events) = run(&invoker, &contract_with("exit 1"), &cfg).await;
        assert_eq!(invoker.calls(), 0, "the budget gates before any work");
        assert_eq!(outcome.failure, Some(LoopFailure::BudgetExhausted));
        assert_ne!(outcome.failure, Some(LoopFailure::Verification));
        assert!(outcome
            .error
            .expect("the reason must surface")
            .contains("session budget exhausted"));
        assert!(events
            .iter()
            .any(|e| matches!(e.kind, CoderEventKind::BudgetExhausted { .. })));
    }

    /// **The clamp.** Admission grants a whole round, so without this a
    /// hypothesis admitted just under the ceiling could run its full 1800s past
    /// it — a ceiling exceeded by 50% is not a ceiling. The invocation's own
    /// timeout is reduced to what the session has left.
    #[test]
    fn an_invocation_timeout_is_clamped_to_the_session_remainder() {
        let dir = tempfile::tempdir().unwrap();
        let executor = WorktreeExecutor::new(dir.path());

        // 10s left on the session, 1800s asked for by the invocation.
        let tight = ExternalLoopConfig {
            timeout_secs: Some(1800),
            deadline: std::sync::Arc::new(SessionDeadline::new(Some(10))),
            ..Default::default()
        };
        let opts = build_invoke_opts(&executor, &tight, None);
        assert_eq!(
            opts.timeout_secs,
            Some(10),
            "the round must not outlive the session"
        );

        // Plenty of session left: the invocation keeps its own, smaller bound.
        let roomy = ExternalLoopConfig {
            timeout_secs: Some(60),
            ..Default::default()
        };
        assert_eq!(
            build_invoke_opts(&executor, &roomy, None).timeout_secs,
            Some(60)
        );

        // No session ceiling: the invocation's own bound stands unchanged.
        let unbounded = ExternalLoopConfig {
            timeout_secs: Some(60),
            deadline: SessionDeadline::unlimited(),
            ..Default::default()
        };
        assert_eq!(
            build_invoke_opts(&executor, &unbounded, None).timeout_secs,
            Some(60)
        );
    }

    /// The whole point of the `Arc`: a second rung of the fallback ladder gets
    /// the SAME clock, not a fresh one. Before this, `external -> native` and
    /// `foreman -> native` each restarted the ceiling.
    #[test]
    fn a_second_rung_shares_the_first_rungs_clock() {
        let first = ExternalLoopConfig {
            deadline: std::sync::Arc::new(SessionDeadline::new(Some(0))),
            ..Default::default()
        };
        // How `rpc` builds the fallback rung: clone the handle, not the value.
        let second = ExternalLoopConfig {
            deadline: std::sync::Arc::clone(&first.deadline),
            ..Default::default()
        };
        assert!(
            std::sync::Arc::ptr_eq(&first.deadline, &second.deadline),
            "the fallback must not buy the session another full ceiling"
        );
        assert!(
            second.deadline.admit().is_some(),
            "an already-spent session must stay spent across the ladder"
        );
    }

    /// The default budget must not interfere with an ordinary session.
    #[tokio::test]
    async fn the_default_budget_does_not_gate_a_normal_run() {
        let invoker = ScriptedInvoker::new(vec![clean_run()]);
        let (outcome, _) = run(
            &invoker,
            &contract_with(PASS),
            &ExternalLoopConfig::default(),
        )
        .await;
        assert!(outcome.passed);
        assert_eq!(invoker.calls(), 1);
    }

    /// A clean run that loses on the checks is a task loss, not an infra one —
    /// it must NOT carry the scraped infra prefix.
    #[tokio::test]
    async fn a_verification_loss_carries_no_infra_marker() {
        let cfg = ExternalLoopConfig {
            repair_invokes: 0,
            ..Default::default()
        };
        let (outcome, _) = run(
            &ScriptedInvoker::new(vec![clean_run()]),
            &contract_with("exit 1"),
            &cfg,
        )
        .await;
        assert_eq!(outcome.failure, Some(LoopFailure::Verification));
        assert!(
            outcome.error.is_none(),
            "a genuine task loss must stay in the scored denominator"
        );
    }

    /// The one test that exercises the REAL classification path end-to-end.
    /// Every other loop test scripts the invoker, so without this nothing
    /// verifies that a genuinely absent CLI still produces `Spawn` ->
    /// `EngineUnavailable` -> the scraped prefix. That is the standing cost of
    /// introducing a seam, and it is worth paying once.
    #[tokio::test]
    async fn a_missing_cli_is_engine_unavailable_through_the_live_invoker() {
        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 outcome = run_external_loop(
            &LiveInvoker,
            "no-such-cli",
            "x",
            &contract(),
            &executor,
            &sink,
            &cancel,
            &ExternalLoopConfig::default(),
            None,
        )
        .await;
        assert!(!outcome.passed);
        assert_eq!(outcome.failure, Some(LoopFailure::EngineUnavailable));
        let err = outcome.error.expect("spawn failure must surface");
        assert!(err.starts_with("external agent '"), "{err}");
        assert!(err.contains("no-such-cli"), "{err}");
    }

    /// Cancellation must stay distinguishable from "the engine could not run",
    /// because `rpc` starts a native loop on the latter and moves to a
    /// different terminal state. Conflating them starts work the user stopped.
    #[tokio::test]
    async fn cancellation_is_not_engine_unavailable() {
        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(true));
        let invoker = ScriptedInvoker::new(vec![]);
        let outcome = run_external_loop(
            &invoker,
            "claude-code",
            "x",
            &contract(),
            &executor,
            &sink,
            &cancel,
            &ExternalLoopConfig::default(),
            None,
        )
        .await;
        assert_eq!(invoker.calls(), 0, "pre-cancelled must not invoke");
        assert_eq!(outcome.failure, Some(LoopFailure::Cancelled));
        assert_ne!(outcome.failure, Some(LoopFailure::EngineUnavailable));
        // The bare spelling `rpc` and the A/B both still read.
        assert_eq!(outcome.error.as_deref(), Some("cancelled"));
    }

    #[test]
    fn rendered_feedback_carries_only_failing_checks() {
        let results = vec![
            CheckResult {
                name: "build".into(),
                passed: true,
                exit_code: Some(0),
                output_tail: "ok".into(),
                duration_ms: 1,
                timed_out: false,
                deadline_clamped: false,
            },
            CheckResult {
                name: "tests".into(),
                passed: false,
                exit_code: Some(1),
                output_tail: "assertion failed".into(),
                duration_ms: 2,
                timed_out: false,
                deadline_clamped: false,
            },
        ];
        let rendered = render_check_failures(&results);
        assert!(rendered.contains("FAILED tests"));
        assert!(rendered.contains("assertion failed"));
        assert!(!rendered.contains("build"), "passing checks are noise");
    }
}