car-multi 0.53.0

//! Multiplayer farm-out: one Foreman run spread over several CAR instances.
//!
//! [`run_farm_out`](super::harness::run_farm_out) takes a single
//! [`WorktreeAgent`] and runs every subtask through it. [`FleetPool`] *is* a
//! `WorktreeAgent`, backed by several — the local coding CLI plus a worker on
//! each reachable peer — so the harness, the gate, and the integration step are
//! untouched. Distribution is a placement decision underneath an unchanged
//! interface, which is the whole reason it costs no new soundness argument:
//!
//! - The worktree, the patch capture, the AST containment check, the build/test
//!   leg, the policy consult, and the union integration all still happen on the
//!   orchestrating host.
//! - A remote worker's only output is edits in that worktree. A broken or
//!   hostile one produces a patch the local gate then rejects, exactly like a
//!   local agent having a bad day.
//!
//! ## Placement
//!
//! Least-loaded first, ties broken by declaration order, so a caller expresses
//! preference by ordering its workers. Each worker has a `capacity` — the number
//! of subtasks it will run at once — enforced by a semaphore, because a machine
//! that accepts eight parallel coding CLIs when it can serve two turns a
//! speed-up into a thrash.
//!
//! ## Failover, and why it must reset the worktree
//!
//! A worker that errors (network dropped, CLI missing, peer declined) hands the
//! subtask to the next candidate. But a half-finished attempt leaves edits
//! behind, and the next worker would then be editing someone else's partial
//! work and the gate would attribute the mess to the subtask. So every failover
//! **resets the worktree to the commit it was provisioned at** before retrying.
//! A run that cannot reset does not retry: silently continuing from a dirty tree
//! is the one outcome worse than failing the subtask.
//!
//! ## Quarantine
//!
//! Failing over per-subtask is not enough on its own. A worker that fails never
//! takes a permit, so it keeps maximum availability and `candidates` ranks it
//! FIRST again for the next subtask — a peer that dies mid-run is then tried,
//! and times out, once for every subtask left (car#1323).
//!
//! So a REMOTE worker that returns [`ForemanError::Worker`] is excluded for the
//! rest of the run. That error means the failure is a fact about the machine
//! rather than the subtask, which is exactly the condition under which retrying
//! it buys nothing; a subtask that merely failed returns
//! [`ForemanError::Agent`] and changes nothing about the worker. The local
//! worker is never quarantined — it is the one guaranteed-reachable machine,
//! and the pool must not be able to empty itself.
//!
//! The bound is the worker's CAPACITY, not one attempt. A level runs under
//! `futures::future::join_all`, so several subtasks can be inside a dying peer
//! before any of them sets the flag — and `acquire_owned` WAITS on the
//! top-ranked candidate rather than skipping to a free one, so the rest of the
//! level parks on its semaphore. The flag is therefore re-read on the far side
//! of the acquire, which is what turns "one dispatch per remaining subtask" into
//! "one per permit". Wall-clock cost is one timeout, not N.
//!
//! Run-local and one-way: a peer that comes back stays out until the next run.
//! Re-probing liveness mid-run is a different feature, and the cost this fixes
//! is already paid by then. The one member of the set with a shorter natural
//! life is a rate limit, whose window is hourly — a run longer than that loses a
//! peer that would have been served again. Still not worth a re-probe.
//!
//! Nothing produces `ForemanError::Worker` locally today: `ForemanExternalAgent`
//! never returns it. The remote-only guard is there because a local one would
//! mean this host's own coding CLI vanished, which quarantining cannot route
//! around and which could leave the pool with nothing to run.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use tokio::sync::Semaphore;

use super::harness::{AgentRunSummary, ForemanError, WorktreeAgent, WorktreeAgentRequest};

/// One place subtasks can run.
pub struct FleetWorker {
    /// Instance name, as the fleet composite names it. Appears in the placement
    /// ledger and the run report, so an operator can see which machine produced
    /// which patch.
    pub id: String,
    /// How the subtask actually runs there — a local coding CLI, or a client
    /// that dispatches to a peer and applies the patch it returns.
    pub agent: Arc<dyn WorktreeAgent>,
    /// Subtasks this worker runs at once. Clamped to at least 1: a worker with
    /// no capacity is a worker that should not have been offered.
    pub capacity: usize,
    /// Whether this worker is on another host. Reported, not routed on — the
    /// pool prefers whoever is free, and the caller expresses any other
    /// preference through ordering.
    pub remote: bool,
}

impl FleetWorker {
    /// A worker on this host.
    pub fn local(id: impl Into<String>, agent: Arc<dyn WorktreeAgent>, capacity: usize) -> Self {
        Self {
            id: id.into(),
            agent,
            capacity: capacity.max(1),
            remote: false,
        }
    }

    /// A worker on another CAR instance.
    pub fn remote(id: impl Into<String>, agent: Arc<dyn WorktreeAgent>, capacity: usize) -> Self {
        Self {
            id: id.into(),
            agent,
            capacity: capacity.max(1),
            remote: true,
        }
    }
}

/// One failed attempt at a subtask, kept so a run that eventually succeeded
/// still shows which workers dropped it.
///
/// This type owns a JSON contract, not just an in-memory shape. `foreman.run`'s
/// report has always rendered these field names, `car-cli`'s fleet output parses
/// them, and a coder session now PERSISTS them in its snapshot — so a rename or
/// a new required field here breaks a consumer or an on-disk record with no
/// compile signal in this crate. `placements_wire_shape_is_pinned` is what makes
/// that a test failure instead (car#1322).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FailedAttempt {
    #[serde(rename = "worker")]
    pub worker_id: String,
    pub error: String,
}

/// Where a subtask ended up running.
///
/// Recorded when the worker RETURNS, which is before the per-patch gate rules on
/// what it produced — so a placement says a machine ran a subtask, not that its
/// patch was accepted or delivered. A caller making a claim about a delivered
/// artifact has to intersect this with what it actually integrated.
///
/// Carries the same JSON contract as [`FailedAttempt`]; see its note.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Placement {
    pub subtask_id: String,
    /// The worker that produced the edits, or `None` when every worker failed.
    #[serde(rename = "worker")]
    pub worker_id: Option<String>,
    /// Whether that worker was a peer rather than this host. **Meaningless when
    /// `worker_id` is `None`** — nothing ran, so there is no location, and the
    /// all-failed record stores `false` for want of an answer rather than as a
    /// claim. Readers must not render a location without a worker.
    pub remote: bool,
    /// Workers that failed first, in the order they were tried.
    #[serde(rename = "failed_attempts", default)]
    pub attempts: Vec<FailedAttempt>,
}

/// A [`WorktreeAgent`] that spreads subtasks over a set of workers.
pub struct FleetPool {
    slots: Vec<Slot>,
    ledger: Mutex<Vec<Placement>>,
}

struct Slot {
    worker: FleetWorker,
    permits: Arc<Semaphore>,
    /// Set when this worker failed with [`ForemanError::Worker`] — a fact about
    /// the machine, not the subtask. It is then excluded for the rest of the
    /// run. Never set for a local worker: this host is the one guaranteed-
    /// reachable machine, and removing it could empty the pool.
    quarantined: AtomicBool,
}

impl FleetPool {
    /// Build a pool. Workers are tried in the order given when equally loaded,
    /// so put the preferred one first.
    ///
    /// A pool with no workers is accepted and fails every subtask with a clear
    /// message — a caller that filtered its fleet down to nothing gets told so
    /// rather than getting a silent no-op run.
    pub fn new(workers: Vec<FleetWorker>) -> Self {
        let slots = workers
            .into_iter()
            .map(|worker| {
                let permits = Arc::new(Semaphore::new(worker.capacity.max(1)));
                Slot {
                    worker,
                    permits,
                    quarantined: AtomicBool::new(false),
                }
            })
            .collect();
        Self {
            slots,
            ledger: Mutex::new(Vec::new()),
        }
    }

    /// Total subtasks this pool can run at once.
    pub fn capacity(&self) -> usize {
        self.slots.iter().map(|s| s.worker.capacity).sum()
    }

    pub fn worker_ids(&self) -> Vec<&str> {
        self.slots.iter().map(|s| s.worker.id.as_str()).collect()
    }

    /// Where each subtask ran, in completion order.
    pub fn placements(&self) -> Vec<Placement> {
        self.ledger
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    fn record(&self, placement: Placement) {
        self.ledger
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(placement);
    }

    /// Workers removed from the pool mid-run, in declaration order.
    ///
    /// Reported next to the pool's build-time exclusions so a degraded run says
    /// so. A worker lands here only via [`ForemanError::Worker`] — see
    /// [`Self::run_in`].
    pub fn quarantined(&self) -> Vec<&str> {
        self.slots
            .iter()
            .filter(|s| s.quarantined.load(Ordering::Relaxed))
            .map(|s| s.worker.id.as_str())
            .collect()
    }

    /// Slot indices, least-loaded first, ties in declaration order.
    ///
    /// Quarantined workers are FILTERED OUT rather than sorted last. Demotion
    /// would leave a dead peer in the list, so every subtask whose healthy
    /// worker errors fails over into it and pays a full connect timeout — the
    /// complaint car#1323 is about, in miniature. (It would NOT be reached
    /// merely because a healthy worker is busy: `acquire_owned` waits on the
    /// top-ranked candidate rather than skipping to a free one.)
    ///
    /// A snapshot: another subtask may take a permit between ranking and
    /// acquiring. That is fine — the ordering is a preference, and the semaphore
    /// is what actually bounds a worker.
    fn candidates(&self) -> Vec<usize> {
        let mut order: Vec<usize> = (0..self.slots.len())
            .filter(|&i| !self.slots[i].quarantined.load(Ordering::Relaxed))
            .collect();
        order.sort_by_key(|&i| {
            (
                std::cmp::Reverse(self.slots[i].permits.available_permits()),
                i,
            )
        });
        order
    }
}

#[async_trait]
impl WorktreeAgent for FleetPool {
    async fn run_in(
        &self,
        req: &WorktreeAgentRequest<'_>,
    ) -> Result<AgentRunSummary, ForemanError> {
        if self.slots.is_empty() {
            return Err(ForemanError::Agent(
                "no workers in the fleet pool: nothing can run this subtask".into(),
            ));
        }

        let subtask_id = req.subtask.id.clone();
        let mut attempts: Vec<FailedAttempt> = Vec::new();

        // Whether any worker has actually been handed this subtask yet. NOT the
        // old `nth > 0`: a candidate can be skipped without running (quarantined
        // since the list was ranked, or a closed semaphore), and resetting for
        // one of those would run a git operation against a tree nothing touched
        // — whose failure aborts the subtask.
        let mut attempted = false;

        for idx in self.candidates() {
            let slot = &self.slots[idx];

            // Cheap skip for a peer already known dead when this subtask got
            // here — saves queueing on its semaphore at all. NOT sufficient on
            // its own; see the re-check below.
            if slot.quarantined.load(Ordering::Relaxed) {
                continue;
            }

            // Every failover after an attempt starts from the base the worktree
            // was provisioned at. Skipping this would hand the next worker a
            // tree carrying a failed attempt's half-written edits.
            if attempted {
                if let Err(e) = reset_worktree(req.cwd) {
                    // Record before returning. This subtask has already been
                    // through at least one worker — `attempted` — and bailing
                    // without a ledger row dropped every one of those attempts,
                    // so `placements()` did not even name the subtask. The run
                    // whose receipt someone wants is exactly this one.
                    //
                    // `worker_id: None` and no synthetic attempt against
                    // `slot.worker.id`: that worker never received the subtask,
                    // and listing it under `failed_attempts` would read as a
                    // worker that ran and failed. The reset failure reaches the
                    // caller in the error; the ledger reports only what ran.
                    self.record(Placement {
                        subtask_id,
                        worker_id: None,
                        remote: false,
                        attempts,
                    });
                    return Err(ForemanError::Git(format!(
                        "cannot fail over to `{}`: {e}",
                        slot.worker.id
                    )));
                }
            }

            let _permit = match slot.permits.clone().acquire_owned().await {
                Ok(p) => p,
                Err(_) => {
                    // The semaphore is owned by this pool and never closed, so
                    // this is unreachable in practice; treat it as the worker
                    // being unavailable rather than panicking a whole run.
                    attempts.push(FailedAttempt {
                        worker_id: slot.worker.id.clone(),
                        error: "worker capacity closed".into(),
                    });
                    continue;
                }
            };

            // THE load-bearing check, and it has to be on this side of the
            // acquire. `acquire_owned` waits — it does not skip a busy worker —
            // so a level of N subtasks parks N-minus-capacity tasks on this
            // semaphore, all of which passed the check above before anything had
            // failed. Each permit a timing-out holder releases would otherwise
            // hand the subtask straight back to the machine that was declared
            // dead while it waited. Checking before the acquire alone buys
            // nothing within a level, which is the regime that matters.
            if slot.quarantined.load(Ordering::Relaxed) {
                drop(_permit);
                continue;
            }

            attempted = true;
            match slot.worker.agent.run_in(req).await {
                Ok(summary) => {
                    self.record(Placement {
                        subtask_id,
                        worker_id: Some(slot.worker.id.clone()),
                        remote: slot.worker.remote,
                        attempts,
                    });
                    return Ok(summary);
                }
                Err(e) => {
                    // A worker-level failure is a fact about the machine, so it
                    // holds for every remaining subtask. Without this the worker
                    // keeps MAXIMUM availability — it never took a permit — and
                    // `candidates` therefore ranks it FIRST for each one in turn
                    // (car#1323).
                    //
                    // Remote only. A local `Worker` error means this host's own
                    // coding CLI went missing, which quarantining cannot route
                    // around and which would leave the pool with nothing.
                    // `swap` rather than `store`: a concurrent level can land
                    // several worker-level failures on the same slot, and the
                    // log line is worth exactly once.
                    if slot.worker.remote
                        && matches!(e, ForemanError::Worker(_))
                        && !slot.quarantined.swap(true, Ordering::Relaxed)
                    {
                        tracing::warn!(
                            worker = %slot.worker.id,
                            error = %e,
                            "quarantining fleet worker for the rest of the run"
                        );
                    }
                    tracing::warn!(
                        subtask = %subtask_id,
                        worker = %slot.worker.id,
                        error = %e,
                        "fleet worker failed; trying the next"
                    );
                    attempts.push(FailedAttempt {
                        worker_id: slot.worker.id.clone(),
                        error: e.to_string(),
                    });
                }
            }
        }

        if !attempted {
            // Every candidate was skipped without being offered the subtask —
            // in practice, all of them quarantined. Reachable only for a pool
            // with no local worker, since the local one never is. Falling
            // through to the tail below would report "every fleet worker
            // failed" with an empty list of attempts, for a subtask nothing was
            // asked to run. Both the pre-acquire and post-acquire skips land
            // here, so there is one guard rather than one per path.
            return Err(ForemanError::Agent(format!(
                "no worker could be offered this subtask; quarantined for this run: {}",
                self.quarantined().join(", ")
            )));
        }

        let detail = attempts
            .iter()
            .map(|a| format!("{}: {}", a.worker_id, a.error))
            .collect::<Vec<_>>()
            .join("; ");
        self.record(Placement {
            subtask_id,
            worker_id: None,
            remote: false,
            attempts,
        });
        Err(ForemanError::Agent(format!(
            "every fleet worker failed — {detail}"
        )))
    }
}

/// Return a worktree to the commit it was checked out at.
///
/// `reset --hard` drops tracked edits; `clean -fd` drops the untracked files a
/// half-finished agent left behind. Both are needed: a subtask that created new
/// files and then errored would otherwise leave them for the next worker, and
/// the gate would read them as that worker's output.
fn reset_worktree(cwd: &std::path::Path) -> Result<(), String> {
    for args in [vec!["reset", "--hard", "--quiet"], vec!["clean", "-fdq"]] {
        let out = std::process::Command::new("git")
            .arg("-C")
            .arg(cwd)
            .args(&args)
            .output()
            .map_err(|e| format!("git {}: {e}", args.join(" ")))?;
        if !out.status.success() {
            return Err(format!(
                "git {} failed: {}",
                args.join(" "),
                String::from_utf8_lossy(&out.stderr).trim()
            ));
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::patterns::foreman::harness::Subtask;
    use std::path::{Path, PathBuf};

    /// These field names are a CONTRACT, not an implementation detail.
    ///
    /// `foreman.run`'s report renders them, `car fleet`'s output parses them,
    /// and a coder session persists them in an on-disk snapshot — none of which
    /// this crate can see. Before car#1322 they were literals in one
    /// hand-written renderer, which at least documented the shape at the point
    /// of use; they are serde attributes now, so dropping a rename would break
    /// three consumers and read a stale snapshot wrong, all green. This is the
    /// only thing standing there.
    ///
    /// Adding a REQUIRED field here has the same reach: every existing session
    /// snapshot then fails to load.
    #[test]
    fn placement_wire_shape_is_pinned() {
        let json = serde_json::to_value(vec![Placement {
            subtask_id: "s1".into(),
            worker_id: Some("studio".into()),
            remote: true,
            attempts: vec![FailedAttempt {
                worker_id: "laptop".into(),
                error: "boom".into(),
            }],
        }])
        .unwrap();

        assert_eq!(
            json,
            serde_json::json!([{
                "subtask_id": "s1",
                "worker": "studio",
                "remote": true,
                "failed_attempts": [{ "worker": "laptop", "error": "boom" }],
            }])
        );

        // And back, so a persisted snapshot written by this version still loads.
        let round: Vec<Placement> = serde_json::from_value(json).unwrap();
        assert_eq!(round[0].worker_id.as_deref(), Some("studio"));
        assert_eq!(round[0].attempts[0].worker_id, "laptop");
    }

    /// Which keys a snapshot may omit, pinned because the answer is not obvious
    /// and the wrong belief about it produces a compatibility break.
    ///
    /// serde_derive DOES implicitly default an `Option` field to `None` when the
    /// key is absent — no `#[serde(default)]` needed — while a `Vec` does not,
    /// which is why `failed_attempts` carries one and `worker` does not. That
    /// asymmetry looks like an oversight and is not. `subtask_id` and `remote`
    /// are genuinely required, and a record missing either is malformed rather
    /// than partial.
    #[test]
    fn optional_keys_are_optional_and_required_ones_are_required() {
        let p: Placement = serde_json::from_value(serde_json::json!({
            "subtask_id": "s1",
            "remote": false,
        }))
        .expect("worker and failed_attempts may both be absent");
        assert_eq!(p.worker_id, None);
        assert!(p.attempts.is_empty());

        for missing in [
            serde_json::json!({ "remote": false }),
            serde_json::json!({ "subtask_id": "s1" }),
        ] {
            assert!(
                serde_json::from_value::<Placement>(missing.clone()).is_err(),
                "{missing} must not deserialize"
            );
        }
    }
    use std::process::Command;
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn git(cwd: &Path, args: &[&str]) {
        let out = Command::new("git")
            .args(args)
            .current_dir(cwd)
            .output()
            .expect("git runs");
        assert!(
            out.status.success(),
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    }

    /// A git worktree standing in for one the harness provisions.
    fn worktree() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        git(root, &["init", "-q", "-b", "main"]);
        git(root, &["config", "user.email", "t@t.t"]);
        git(root, &["config", "user.name", "t"]);
        git(root, &["config", "core.autocrlf", "false"]);
        std::fs::write(root.join("seed.txt"), "seed\n").unwrap();
        git(root, &["add", "-A"]);
        git(root, &["commit", "-qm", "base"]);
        dir
    }

    /// Records which worker ran, and how many were in flight at the peak.
    struct Counting {
        id: &'static str,
        inflight: Arc<AtomicUsize>,
        peak: Arc<AtomicUsize>,
        ran: Arc<Mutex<Vec<String>>>,
    }

    #[async_trait]
    impl WorktreeAgent for Counting {
        async fn run_in(
            &self,
            req: &WorktreeAgentRequest<'_>,
        ) -> Result<AgentRunSummary, ForemanError> {
            let now = self.inflight.fetch_add(1, Ordering::SeqCst) + 1;
            self.peak.fetch_max(now, Ordering::SeqCst);
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            self.ran
                .lock()
                .unwrap()
                .push(format!("{}:{}", self.id, req.subtask.id));
            self.inflight.fetch_sub(1, Ordering::SeqCst);
            Ok(AgentRunSummary {
                answer: self.id.to_string(),
            })
        }
    }

    /// Writes a file, then fails — the partial-work case failover must clean up.
    struct MessyFailure;
    #[async_trait]
    impl WorktreeAgent for MessyFailure {
        async fn run_in(
            &self,
            req: &WorktreeAgentRequest<'_>,
        ) -> Result<AgentRunSummary, ForemanError> {
            std::fs::write(req.cwd.join("half-done.txt"), "partial\n").unwrap();
            std::fs::write(req.cwd.join("seed.txt"), "clobbered\n").unwrap();
            Err(ForemanError::Agent("peer dropped mid-subtask".into()))
        }
    }

    /// Asserts the worktree it receives is clean at base.
    struct ExpectsCleanTree;
    #[async_trait]
    impl WorktreeAgent for ExpectsCleanTree {
        async fn run_in(
            &self,
            req: &WorktreeAgentRequest<'_>,
        ) -> Result<AgentRunSummary, ForemanError> {
            if req.cwd.join("half-done.txt").exists() {
                return Err(ForemanError::Agent(
                    "inherited a failed worker's untracked file".into(),
                ));
            }
            let seed = std::fs::read_to_string(req.cwd.join("seed.txt")).unwrap();
            if seed != "seed\n" {
                return Err(ForemanError::Agent(
                    "inherited a failed worker's tracked edit".into(),
                ));
            }
            Ok(AgentRunSummary {
                answer: "clean".into(),
            })
        }
    }

    /// A failover that cannot reset the worktree still owes a ledger row.
    ///
    /// The subtask has been through a worker by then, and returning without
    /// recording dropped that attempt AND the subtask itself — `placements()`
    /// did not name it at all. Someone reading the ledger to find out what
    /// happened to a failed run got silence about the one that failed.
    #[tokio::test]
    async fn a_failover_that_cannot_reset_the_tree_still_records_what_ran() {
        let pool = FleetPool::new(vec![
            FleetWorker::local("first", Arc::new(MessyFailure), 1),
            FleetWorker::local("second", Arc::new(ExpectsCleanTree), 1),
        ]);
        // NOT a git repo, so `reset_worktree` fails on the failover — the same
        // shape as a repo whose `.git` went away mid-run.
        let dir = tempfile::tempdir().unwrap();
        let cwd = dir.path().to_path_buf();
        let a = Subtask::files_only("a", "a", vec![]);
        let err = pool.run_in(&request(&a, &cwd)).await.unwrap_err();
        assert!(
            matches!(err, ForemanError::Git(_)),
            "expected the reset failure to surface: {err:?}"
        );

        let placements = pool.placements();
        assert_eq!(placements.len(), 1, "the subtask must appear in the ledger");
        assert_eq!(placements[0].subtask_id, "a");
        // Nobody completed it.
        assert_eq!(placements[0].worker_id, None);
        // And the worker that DID run and fail is named, once.
        assert_eq!(placements[0].attempts.len(), 1);
        assert_eq!(placements[0].attempts[0].worker_id, "first");
        // `second` never received the subtask, so it must not appear as an
        // attempt — that is the false attribution the ledger split avoids.
        assert!(
            !placements[0]
                .attempts
                .iter()
                .any(|a| a.worker_id == "second"),
            "a worker the failover never reached is not a failed attempt"
        );
    }

    /// Counts every dispatch, and fails each with a caller-chosen error.
    struct Failing {
        calls: Arc<AtomicUsize>,
        worker_level: bool,
    }
    #[async_trait]
    impl WorktreeAgent for Failing {
        async fn run_in(
            &self,
            _req: &WorktreeAgentRequest<'_>,
        ) -> Result<AgentRunSummary, ForemanError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            Err(if self.worker_level {
                ForemanError::Worker("peer is not there".into())
            } else {
                ForemanError::Agent("the subtask failed".into())
            })
        }
    }

    fn always_ok(id: &'static str) -> Arc<Counting> {
        Arc::new(Counting {
            id,
            inflight: Arc::new(AtomicUsize::new(0)),
            peak: Arc::new(AtomicUsize::new(0)),
            ran: Arc::new(Mutex::new(Vec::new())),
        })
    }

    /// The bug: a worker that fails never takes a permit, so it keeps maximum
    /// availability and `candidates` ranks it first for EVERY remaining subtask
    /// — a peer that died mid-run is re-dispatched to, and times out, once per
    /// subtask left (car#1323).
    #[tokio::test]
    async fn a_worker_level_failure_takes_the_peer_out_for_the_rest_of_the_run() {
        let calls = Arc::new(AtomicUsize::new(0));
        let pool = FleetPool::new(vec![
            FleetWorker::remote(
                "dead-peer",
                Arc::new(Failing {
                    calls: Arc::clone(&calls),
                    worker_level: true,
                }),
                1,
            ),
            FleetWorker::local("here", always_ok("here"), 1),
        ]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();

        for id in ["a", "b", "c"] {
            let sub = Subtask::files_only(id, id, vec![]);
            pool.run_in(&request(&sub, &cwd))
                .await
                .expect("the local worker picks it up");
        }

        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "the dead peer must be offered exactly one subtask, not one per subtask"
        );
        assert_eq!(pool.quarantined(), vec!["dead-peer"]);
        // Only the first subtask records the failed attempt; the rest never
        // reach that worker at all.
        let with_attempts = pool
            .placements()
            .iter()
            .filter(|p| !p.attempts.is_empty())
            .count();
        assert_eq!(with_attempts, 1);
    }

    /// Fails worker-level, but only after giving every other subtask in the
    /// level time to queue behind it.
    struct SlowFailing {
        calls: Arc<AtomicUsize>,
    }
    #[async_trait]
    impl WorktreeAgent for SlowFailing {
        async fn run_in(
            &self,
            _req: &WorktreeAgentRequest<'_>,
        ) -> Result<AgentRunSummary, ForemanError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            tokio::time::sleep(std::time::Duration::from_millis(60)).await;
            Err(ForemanError::Worker("peer is not there".into()))
        }
    }

    /// A Foreman level runs under `futures::future::join_all`, and that is the
    /// regime the sequential tests above never enter.
    ///
    /// `acquire_owned` WAITS on the top-ranked candidate rather than skipping to
    /// a free one, so every subtask in the level parks on the dead peer's
    /// semaphore having already passed the pre-acquire check — back when nothing
    /// had failed yet. Each permit a timing-out holder releases then hands the
    /// subtask straight back to the machine that was declared dead while it
    /// waited, unless the flag is re-read on the far side of the acquire.
    ///
    /// So the bound is the peer's CAPACITY, not the size of the level.
    #[tokio::test]
    async fn a_whole_level_queued_behind_a_dying_peer_stops_dispatching_to_it() {
        const CAPACITY: usize = 2;
        const LEVEL: usize = 8;

        // The peer is declared FIRST and out-capacities the local worker, so it
        // is top-ranked at the start of the level and stays top-ranked once both
        // are saturated (`candidates` breaks a tie by declaration order). That
        // is what parks the rest of the level on its semaphore rather than
        // spreading them — the shape a real run has when a peer is the fast
        // machine and this host is the fallback.
        let calls = Arc::new(AtomicUsize::new(0));
        let pool = FleetPool::new(vec![
            FleetWorker::remote(
                "dying-peer",
                Arc::new(SlowFailing {
                    calls: Arc::clone(&calls),
                }),
                CAPACITY,
            ),
            FleetWorker::local("here", always_ok("here"), 1),
        ]);

        // A worktree per subtask, as the harness provisions them — so the
        // failover resets do not contend on one repo.
        let dirs: Vec<tempfile::TempDir> = (0..LEVEL).map(|_| worktree()).collect();
        let cwds: Vec<PathBuf> = dirs.iter().map(|d| d.path().to_path_buf()).collect();
        let subtasks: Vec<Subtask> = (0..LEVEL)
            .map(|i| Subtask::files_only(format!("s{i}"), "go", vec![]))
            .collect();

        let reqs: Vec<WorktreeAgentRequest<'_>> = subtasks
            .iter()
            .zip(cwds.iter())
            .map(|(sub, cwd)| request(sub, cwd))
            .collect();
        let results = futures::future::join_all(reqs.iter().map(|r| pool.run_in(r))).await;

        for r in &results {
            r.as_ref()
                .expect("the local worker picks every one of them up");
        }
        assert_eq!(pool.quarantined(), vec!["dying-peer"]);
        assert!(
            calls.load(Ordering::SeqCst) <= CAPACITY,
            "a level of {LEVEL} must not dispatch past the peer's capacity of \
             {CAPACITY} once it is quarantined; got {}",
            calls.load(Ordering::SeqCst)
        );
    }

    /// The boundary the whole design rests on. A subtask that merely failed says
    /// nothing about the machine, so quarantining on it would remove a healthy
    /// worker for having had one bad task — worse than the bug being fixed.
    #[tokio::test]
    async fn a_subtask_that_failed_does_not_quarantine_the_worker_that_ran_it() {
        let calls = Arc::new(AtomicUsize::new(0));
        let pool = FleetPool::new(vec![
            FleetWorker::remote(
                "healthy-peer",
                Arc::new(Failing {
                    calls: Arc::clone(&calls),
                    worker_level: false,
                }),
                1,
            ),
            FleetWorker::local("here", always_ok("here"), 1),
        ]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();

        for id in ["a", "b", "c"] {
            let sub = Subtask::files_only(id, id, vec![]);
            pool.run_in(&request(&sub, &cwd)).await.expect("local runs");
        }

        assert_eq!(
            calls.load(Ordering::SeqCst),
            3,
            "an `Agent` failure must leave the worker in the pool"
        );
        assert!(pool.quarantined().is_empty());
    }

    /// The local worker is the one guaranteed-reachable machine. A `Worker`
    /// error from it means this host's own CLI went missing, which quarantining
    /// cannot route around — and doing it anyway can empty the pool.
    #[tokio::test]
    async fn the_local_worker_is_never_quarantined() {
        let calls = Arc::new(AtomicUsize::new(0));
        let pool = FleetPool::new(vec![FleetWorker::local(
            "here",
            Arc::new(Failing {
                calls: Arc::clone(&calls),
                worker_level: true,
            }),
            1,
        )]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();

        for id in ["a", "b"] {
            let sub = Subtask::files_only(id, id, vec![]);
            pool.run_in(&request(&sub, &cwd))
                .await
                .expect_err("it fails, but it stays in the pool");
        }

        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert!(pool.quarantined().is_empty());
    }

    /// An all-remote pool can quarantine itself down to nothing. Without an
    /// explicit check the loop body never runs and the tail reports
    /// "every fleet worker failed — " with no attempts behind it, for a subtask
    /// that was never offered to anyone.
    #[tokio::test]
    async fn an_all_remote_pool_that_quarantines_everyone_says_so() {
        let calls = Arc::new(AtomicUsize::new(0));
        let pool = FleetPool::new(vec![FleetWorker::remote(
            "only-peer",
            Arc::new(Failing {
                calls: Arc::clone(&calls),
                worker_level: true,
            }),
            1,
        )]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();

        let a = Subtask::files_only("a", "a", vec![]);
        pool.run_in(&request(&a, &cwd)).await.unwrap_err();
        let b = Subtask::files_only("b", "b", vec![]);
        let err = pool.run_in(&request(&b, &cwd)).await.unwrap_err();

        assert_eq!(calls.load(Ordering::SeqCst), 1);
        let text = err.to_string();
        assert!(
            text.contains("quarantined") && text.contains("only-peer"),
            "the error must name the quarantine and the peer: {text}"
        );
    }

    fn request<'a>(subtask: &'a Subtask, cwd: &'a PathBuf) -> WorktreeAgentRequest<'a> {
        WorktreeAgentRequest {
            subtask,
            cwd,
            allowed_tools: None,
            mcp_endpoint: None,
        }
    }

    fn counting(
        id: &'static str,
        ran: &Arc<Mutex<Vec<String>>>,
    ) -> (Arc<Counting>, Arc<AtomicUsize>) {
        let peak = Arc::new(AtomicUsize::new(0));
        (
            Arc::new(Counting {
                id,
                inflight: Arc::new(AtomicUsize::new(0)),
                peak: Arc::clone(&peak),
                ran: Arc::clone(ran),
            }),
            peak,
        )
    }

    #[tokio::test]
    async fn work_spreads_across_workers_instead_of_queueing_on_one() {
        let ran = Arc::new(Mutex::new(Vec::new()));
        let (here, _) = counting("here", &ran);
        let (studio, _) = counting("studio", &ran);
        let pool = FleetPool::new(vec![
            FleetWorker::local("here", here, 1),
            FleetWorker::remote("studio", studio, 1),
        ]);
        assert_eq!(pool.capacity(), 2);

        let dir = worktree();
        let cwd = dir.path().to_path_buf();
        let a = Subtask::files_only("a", "a", vec![]);
        let b = Subtask::files_only("b", "b", vec![]);
        let (req_a, req_b) = (request(&a, &cwd), request(&b, &cwd));
        let (ra, rb) = tokio::join!(pool.run_in(&req_a), pool.run_in(&req_b));
        ra.unwrap();
        rb.unwrap();

        let placements = pool.placements();
        assert_eq!(placements.len(), 2);
        let used: std::collections::HashSet<String> = placements
            .iter()
            .filter_map(|p| p.worker_id.clone())
            .collect();
        assert_eq!(used.len(), 2, "both workers took a subtask: {placements:?}");
    }

    #[tokio::test]
    async fn a_worker_never_exceeds_its_capacity() {
        let ran = Arc::new(Mutex::new(Vec::new()));
        let (only, peak) = counting("only", &ran);
        let pool = FleetPool::new(vec![FleetWorker::local("only", only, 1)]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();
        let a = Subtask::files_only("a", "a", vec![]);
        let b = Subtask::files_only("b", "b", vec![]);
        let c = Subtask::files_only("c", "c", vec![]);
        let (req_a, req_b, req_c) = (request(&a, &cwd), request(&b, &cwd), request(&c, &cwd));
        let _ = tokio::join!(
            pool.run_in(&req_a),
            pool.run_in(&req_b),
            pool.run_in(&req_c)
        );
        assert_eq!(
            peak.load(Ordering::SeqCst),
            1,
            "capacity 1 means one at a time, however many subtasks arrive"
        );
        assert_eq!(pool.placements().len(), 3);
    }

    #[tokio::test]
    async fn failover_hands_the_next_worker_a_clean_tree() {
        let pool = FleetPool::new(vec![
            FleetWorker::remote("flaky", Arc::new(MessyFailure), 1),
            FleetWorker::local("here", Arc::new(ExpectsCleanTree), 1),
        ]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();
        let s = Subtask::files_only("s", "s", vec![]);
        let summary = pool
            .run_in(&request(&s, &cwd))
            .await
            .expect("second worker takes it");
        assert_eq!(summary.answer, "clean");

        let placements = pool.placements();
        assert_eq!(placements[0].worker_id.as_deref(), Some("here"));
        assert_eq!(placements[0].attempts.len(), 1);
        assert_eq!(placements[0].attempts[0].worker_id, "flaky");
        assert!(!cwd.join("half-done.txt").exists(), "reset cleaned up");
    }

    #[tokio::test]
    async fn every_worker_failing_reports_all_of_them() {
        let pool = FleetPool::new(vec![
            FleetWorker::remote("a", Arc::new(MessyFailure), 1),
            FleetWorker::remote("b", Arc::new(MessyFailure), 1),
        ]);
        let dir = worktree();
        let cwd = dir.path().to_path_buf();
        let s = Subtask::files_only("s", "s", vec![]);
        let err = pool.run_in(&request(&s, &cwd)).await.unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains('a') && msg.contains('b'), "{msg}");
        let placements = pool.placements();
        assert!(placements[0].worker_id.is_none());
        assert_eq!(placements[0].attempts.len(), 2);
    }

    #[tokio::test]
    async fn an_empty_pool_says_so_instead_of_silently_doing_nothing() {
        let pool = FleetPool::new(Vec::new());
        let dir = worktree();
        let cwd = dir.path().to_path_buf();
        let s = Subtask::files_only("s", "s", vec![]);
        let err = pool.run_in(&request(&s, &cwd)).await.unwrap_err();
        assert!(err.to_string().contains("no workers"), "{err}");
    }

    /// Stands in for a peer: it edits its OWN worktree (never the caller's),
    /// captures a patch, and the "orchestrator" applies that patch locally —
    /// the exact round trip `car_fleet::worker` specifies.
    struct PatchReturningWorker {
        base: PathBuf,
    }

    #[async_trait]
    impl WorktreeAgent for PatchReturningWorker {
        async fn run_in(
            &self,
            req: &WorktreeAgentRequest<'_>,
        ) -> Result<AgentRunSummary, ForemanError> {
            // Remote side: a second checkout at the same base commit.
            let remote = tempfile::tempdir().unwrap();
            let clone = remote.path().join("checkout");
            let out = Command::new("git")
                .args(["clone", "-q"])
                .arg(&self.base)
                .arg(&clone)
                .output()
                .unwrap();
            assert!(out.status.success(), "{out:?}");
            std::fs::write(clone.join("new.rs"), "pub fn added() {}\n").unwrap();
            std::fs::write(clone.join("seed.txt"), "edited\n").unwrap();
            let patch = super::super::harness::capture_patch(&clone)?;

            // Orchestrator side: apply into the worktree that will be gated.
            super::super::harness::git_apply(req.cwd, &patch)?;
            Ok(AgentRunSummary {
                answer: "remote".into(),
            })
        }
    }

    #[tokio::test]
    async fn a_patch_made_on_another_checkout_lands_in_the_gated_worktree() {
        // The claim the whole remote protocol rests on: a worker never touches
        // the orchestrator's tree, and its patch still arrives there intact —
        // new files and edits to tracked files alike.
        let dir = worktree();
        let cwd = dir.path().to_path_buf();
        let pool = FleetPool::new(vec![FleetWorker::remote(
            "studio",
            Arc::new(PatchReturningWorker { base: cwd.clone() }),
            1,
        )]);
        let s = Subtask::files_only("s", "s", vec![]);
        pool.run_in(&request(&s, &cwd)).await.expect("applied");

        assert_eq!(
            std::fs::read_to_string(cwd.join("seed.txt")).unwrap(),
            "edited\n",
            "the peer's edit to a tracked file arrived"
        );
        assert!(cwd.join("new.rs").exists(), "and so did its new file");
        let placement = &pool.placements()[0];
        assert_eq!(placement.worker_id.as_deref(), Some("studio"));
        assert!(placement.remote);
    }

    #[test]
    fn a_zero_capacity_worker_is_clamped_rather_than_deadlocking() {
        let pool = FleetPool::new(vec![FleetWorker::local(
            "here",
            Arc::new(ExpectsCleanTree),
            0,
        )]);
        assert_eq!(pool.capacity(), 1);
        assert_eq!(pool.worker_ids(), vec!["here"]);
    }
}