car-server-core 0.55.0

//! `fleet.*` — the composite view of everything CAR can reach, and the
//! placement layer that lets one Foreman run span several instances.
//!
//! ## Why this exists
//!
//! `agents.peers` answers *who* is out there. `discovery.resolve` ranks
//! services against a need. Neither answers the question a caller with work has:
//! **what can the machines I can reach actually do, and which of them should
//! run this?** Before this, a Foreman run's answer to "where does this subtask
//! go" was structurally "the machine you are standing on", because nothing else
//! was ever enumerated.
//!
//! `fleet.composite` enumerates it: every agent, capability, and model across
//! this daemon and every reachable CAR instance, folded so one row names every
//! instance that offers it. `foreman.run { distributed: true }` then spreads the
//! subtasks over the instances that can serve the repository.
//!
//! ## The trust story is unchanged
//!
//! - Remote reads and dispatches ride the **existing** peer-authenticated A2A
//!   listener (`build_router_with_peer_auth`). No new listener, no new door.
//! - Discovery still is not trust: a LAN advertisement is listed as a visible
//!   candidate and is never contacted until an operator promotes it, exactly as
//!   `agents.message` requires.
//! - Accepting farmed-out work is off until an operator turns it on, is limited
//!   to named repositories, and is audited per dispatch (see [`serve`]).
//! - The merge-verify gate does not move. A peer produces a patch; this host
//!   gates it. See [`remote`].

pub mod inventory;
pub mod remote;
pub mod serve;

use std::path::PathBuf;
use std::sync::Arc;

use car_fleet::{FleetComposite, InstanceInventory, InstanceRef, InventoryProvider, WorkerProfile};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState};

pub use remote::RemoteWorktreeAgent;

/// Subtasks this host runs for peers at once, when enrolled and unset.
const DEFAULT_MAX_PARALLEL: u32 = 2;

/// Subtasks this host runs *for itself* at once during a distributed run.
const DEFAULT_LOCAL_PARALLEL: u32 = 2;

/// Longest a peer's subtask may occupy this machine, when the operator has not
/// said otherwise. Thirty minutes: past that a coding CLI is wedged, not slow.
///
/// The ORCHESTRATOR needs this too — its HTTP deadline has to cover the work the
/// far side is allowed to spend, and the dispatch carries no negotiated budget
/// (`SubtaskDispatch::timeout_secs` is `None` unless a caller sets one). One
/// constant so the two sides cannot drift apart; see `remote::dispatch_client`.
pub(super) const DEFAULT_MAX_SUBTASK_SECS: u64 = 1800;

/// How long a peer gets to answer `car/fleetInventory` before it is reported
/// unreachable. See `car_fleet::provider`.
const INVENTORY_TIMEOUT: std::time::Duration = car_fleet::DEFAULT_INVENTORY_TIMEOUT;

/// Whether this instance takes farmed-out coding work, and under what limits.
///
/// Persisted at `~/.car/fleet-worker.json` (under `CAR_HOME` when set) rather
/// than read from the environment, because it is an operator decision that must
/// survive a restart and be inspectable — an env var would make "does this
/// machine run my peers' prompts?" depend on how the daemon happened to be
/// launched.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FleetWorkerConfig {
    /// Off by default. Turning it on lets a trusted peer run a coding CLI
    /// against the checkouts named in `repos`.
    #[serde(default)]
    pub accepts_work: bool,
    /// Repository checkouts this host will serve. A dispatch is matched to one
    /// of these by root commit; anything else is declined.
    #[serde(default)]
    pub repos: Vec<PathBuf>,
    /// Peers' subtasks run here at once.
    #[serde(default = "default_max_parallel")]
    pub max_parallel: u32,
    /// This host's own share when it orchestrates a distributed run. Separate
    /// from `max_parallel`, which bounds what peers may spend here.
    #[serde(default = "default_local_parallel")]
    pub local_parallel: u32,
    /// Dispatches ONE peer may start here per hour.
    ///
    /// `max_parallel` bounds concurrency, which is not a bound on spend: a peer
    /// that dispatches one subtask, waits, and dispatches the next never
    /// exceeds it and can still drain this machine's coding-CLI quota. This is
    /// the budget. `0` accepts nothing while keeping the rest of the config.
    #[serde(default = "default_dispatches_per_hour")]
    pub dispatches_per_hour: u32,
    /// Hard ceiling on a peer-supplied `timeout_secs`.
    ///
    /// The dispatch carries a timeout the *sender* chose. A machine agreeing to
    /// take work is not agreeing to be occupied for as long as the caller
    /// likes, so the sender's value is clamped to this and an omitted one
    /// defaults to it.
    #[serde(default = "default_max_subtask_secs")]
    pub max_subtask_secs: u64,
    /// Fetch a missing base commit from this checkout's own remote instead of
    /// declining — the posture that makes a machine a **runner** rather than a
    /// person's laptop.
    ///
    /// Off by default, because on a laptop it means a peer's dispatch can cause
    /// a network fetch in a repository the operator is working in. On a runner
    /// it is the whole point: a machine that continuously tracks `origin`
    /// always has the base commit, which is what turns an idle pool into a
    /// working one. Adds no trust — the fetch targets the remote this checkout
    /// is already configured with, never anything the dispatch supplied.
    #[serde(default)]
    pub fetch_missing_base: bool,
    /// Remote a runner fetches from. `origin` unless an operator says otherwise.
    #[serde(default = "default_fetch_remote")]
    pub fetch_remote: String,
    /// Tools a peer's coding CLI may use here, intersected with whatever the
    /// dispatch asks for.
    ///
    /// `None` (the default) adds no restriction beyond the CLI's own, which is
    /// the honest default: a coding subtask needs to read, edit, and usually
    /// build, so a narrow allowlist mostly turns into every dispatch failing
    /// for reasons the sender cannot see. Set it when this machine should be
    /// stricter than the work requires.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub allowed_tools: Option<Vec<String>>,
}

fn default_max_parallel() -> u32 {
    DEFAULT_MAX_PARALLEL
}

fn default_local_parallel() -> u32 {
    DEFAULT_LOCAL_PARALLEL
}

fn default_dispatches_per_hour() -> u32 {
    car_fleet::DEFAULT_DISPATCHES_PER_WINDOW
}

fn default_max_subtask_secs() -> u64 {
    DEFAULT_MAX_SUBTASK_SECS
}

fn default_fetch_remote() -> String {
    "origin".to_string()
}

impl Default for FleetWorkerConfig {
    fn default() -> Self {
        Self {
            accepts_work: false,
            repos: Vec::new(),
            max_parallel: DEFAULT_MAX_PARALLEL,
            local_parallel: DEFAULT_LOCAL_PARALLEL,
            dispatches_per_hour: car_fleet::DEFAULT_DISPATCHES_PER_WINDOW,
            max_subtask_secs: DEFAULT_MAX_SUBTASK_SECS,
            fetch_missing_base: false,
            fetch_remote: default_fetch_remote(),
            allowed_tools: None,
        }
    }
}

impl FleetWorkerConfig {
    fn path() -> Option<PathBuf> {
        car_home::root().map(|r| r.join("fleet-worker.json"))
    }

    /// Read the config, falling back to the declining default.
    ///
    /// A malformed file reads as the default rather than as an error: the
    /// failure mode of "unparseable config" must be *declining work*, never
    /// accepting it under a half-read set of limits.
    pub fn load() -> Self {
        let Some(path) = Self::path() else {
            return Self::default();
        };
        match std::fs::read_to_string(&path) {
            Ok(text) => serde_json::from_str(&text).unwrap_or_else(|e| {
                tracing::warn!(path = %path.display(), error = %e, "unreadable fleet worker config; declining work");
                Self::default()
            }),
            Err(_) => Self::default(),
        }
    }

    pub fn save(&self) -> Result<(), String> {
        let path = Self::path().ok_or("cannot resolve the CAR state root")?;
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .map_err(|e| format!("create {}: {e}", parent.display()))?;
        }
        let json = serde_json::to_string_pretty(self).map_err(|e| e.to_string())?;
        let tmp = path.with_extension("json.tmp");
        std::fs::write(&tmp, json).map_err(|e| format!("write {}: {e}", tmp.display()))?;
        std::fs::rename(&tmp, &path).map_err(|e| format!("rename into {}: {e}", path.display()))
    }
}

/// Where worker worktrees are provisioned — under the state root, never inside
/// the served repository, so a crashed dispatch cannot leave untracked files in
/// an operator's checkout.
fn worktree_base() -> PathBuf {
    car_home::root_or_relative()
        .join("fleet-worker")
        .join("worktrees")
}

/// Installed coding CLIs, cached.
///
/// Detection spawns a `--version` subprocess per adapter, which is far too
/// costly to repeat on every inventory read (and an inventory is read on every
/// composite, by every peer). Installs change rarely; a minute is ample — the
/// same reasoning and TTL `discovery.resolve` uses.
async fn detected_adapters() -> Vec<car_external_agents::ExternalAgentSpec> {
    use tokio::sync::Mutex;
    static CACHE: std::sync::OnceLock<
        Mutex<
            Option<(
                std::time::Instant,
                Vec<car_external_agents::ExternalAgentSpec>,
            )>,
        >,
    > = std::sync::OnceLock::new();
    const TTL: std::time::Duration = std::time::Duration::from_secs(60);

    let cache = CACHE.get_or_init(|| Mutex::new(None));
    let mut guard = cache.lock().await;
    if let Some((at, specs)) = guard.as_ref() {
        if at.elapsed() < TTL {
            return specs.clone();
        }
    }
    let specs = car_external_agents::detect_runnable().await;
    *guard = Some((std::time::Instant::now(), specs.clone()));
    specs
}

/// This instance's worker posture, as peers see it.
pub async fn worker_profile() -> WorkerProfile {
    let config = FleetWorkerConfig::load();
    let adapters: Vec<String> = detected_adapters()
        .await
        .into_iter()
        .map(|s| s.id)
        .collect();
    // Root commits only. A peer needs to know *whether* this host can reproduce
    // a base, not where the checkout lives — a path would leak the layout of a
    // machine the peer has no business mapping.
    let repo_root_commits = config
        .repos
        .iter()
        .filter_map(|p| car_fleet::root_commit(p).ok())
        .collect();
    WorkerProfile {
        accepts_work: config.accepts_work,
        adapters,
        max_parallel: config.max_parallel,
        repo_root_commits,
    }
}

/// Trusted peers, as inventory providers, plus rows for instances that are
/// visible but deliberately not contacted.
///
/// The second half matters as much as the first. A LAN advertisement is an
/// unauthenticated claim, so CAR never dials it until an operator promotes it —
/// but leaving those hosts out of the composite entirely would leave an operator
/// wondering why the machine they can see is not in the fleet. They appear as
/// unreachable rows whose reason names the fix.
async fn remote_providers(
    state: &ServerState,
) -> (Vec<Arc<dyn InventoryProvider>>, Vec<InstanceInventory>) {
    let identity = {
        state
            .peer_identity
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    };

    let mut providers: Vec<Arc<dyn InventoryProvider>> = Vec::new();
    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();

    // The user's own devices, from the synced oplog. Authenticated by
    // construction — the roster is readable only with their credentials.
    for peer in crate::peers::snapshot_parslee(state).await {
        if let car_peers::PeerAddress::A2a { base_url } = &peer.address {
            if seen.insert(base_url.clone()) {
                providers.push(Arc::new(inventory::PeerInventoryProvider::new(
                    InstanceRef::remote(peer.name.clone(), "parslee", base_url.clone()),
                    identity.clone(),
                )));
            }
        }
    }

    // Peers an operator promoted through `a2a.peers.add`.
    if let Ok(registry) = car_a2a::peers::PeerRegistry::user_default() {
        for entry in registry.list() {
            if seen.insert(entry.url.clone()) {
                let name = entry.label.clone().unwrap_or_else(|| entry.slug.clone());
                providers.push(Arc::new(inventory::PeerInventoryProvider::new(
                    InstanceRef::remote(name, "registry", entry.url.clone()),
                    identity.clone(),
                )));
            }
        }
    }

    let visible_only = crate::peers::snapshot_lan(state)
        .into_iter()
        .filter_map(|peer| {
            let car_peers::PeerAddress::A2a { base_url } = &peer.address else {
                return None;
            };
            if seen.contains(base_url) {
                return None;
            }
            Some(InstanceInventory::unreachable(
                InstanceRef::remote(peer.name.clone(), "lan", base_url.clone()),
                "discovered on the local network but not a trusted peer — anyone can advertise \
                 any name, so promote it with `a2a.peers.add` before CAR will contact it",
                car_fleet::now_ms(),
            ))
        })
        .collect();

    (providers, visible_only)
}

/// Assemble the composite: this instance plus every reachable peer.
pub async fn composite(
    state: &ServerState,
    session: Option<&ClientSession>,
    include_remote: bool,
    timeout: std::time::Duration,
) -> FleetComposite {
    let local = match session {
        Some(s) => inventory::local_inventory(state, Some(&s.runtime), Some(&s.memgine)).await,
        None => inventory::local_inventory(state, None, None).await,
    };
    let self_name = local.instance.name.clone();

    let mut all = vec![local];
    if include_remote {
        let (providers, visible_only) = remote_providers(state).await;
        all.extend(car_fleet::gather(&providers, timeout).await);
        all.extend(visible_only);
    }
    car_fleet::compose(self_name, all)
}

/// `fleet.inventory` — this instance only. The same report peers receive,
/// plus the calling session's own tools and skills.
pub async fn handle_fleet_inventory(
    state: &ServerState,
    session: &ClientSession,
) -> Result<Value, String> {
    let inv =
        inventory::local_inventory(state, Some(&session.runtime), Some(&session.memgine)).await;
    serde_json::to_value(inv).map_err(|e| e.to_string())
}

/// `fleet.composite` — every agent, capability, and model this daemon can reach.
///
/// Params: `{ include_remote?: bool = true, timeout_ms?: u64 }`. Remote reads
/// are concurrent and individually bounded, so one sleeping machine costs the
/// timeout, not the call.
pub async fn handle_fleet_composite(
    msg: &JsonRpcMessage,
    state: &ServerState,
    session: &ClientSession,
) -> Result<Value, String> {
    let include_remote = msg
        .params
        .get("include_remote")
        .and_then(|v| v.as_bool())
        .unwrap_or(true);
    let timeout = msg
        .params
        .get("timeout_ms")
        .and_then(|v| v.as_u64())
        .map(std::time::Duration::from_millis)
        .unwrap_or(INVENTORY_TIMEOUT);
    let composite = composite(state, Some(session), include_remote, timeout).await;
    serde_json::to_value(composite).map_err(|e| e.to_string())
}

/// `fleet.worker.get` — whether this instance takes farmed-out work.
pub async fn handle_fleet_worker_get() -> Result<Value, String> {
    let config = FleetWorkerConfig::load();
    let profile = worker_profile().await;
    Ok(serde_json::json!({
        "config": config,
        "profile": profile,
    }))
}

/// `fleet.worker.set` — enroll (or withdraw) this instance as a fleet worker.
///
/// **Operator-only**: refused for any session bound to an agent id. Enrolling
/// means a peer's prompt runs a coding CLI against a local checkout, which is a
/// decision for the person at the machine — and an agent must not be able to
/// grant it, least of all the agent that would benefit.
///
/// Deliberately *not* `is_host`, which the peer-approval surfaces use. Those
/// gate an operator's arbitration of messages an agent sent; this gates a
/// configuration change, and the operator's own CLI (which holds the daemon's
/// auth token, and can already start a coder session on this machine) is as much
/// the operator as the host app is. Host-only here would put the feature behind
/// a surface the CLI cannot reach without buying any safety the token boundary
/// does not already provide.
pub async fn handle_fleet_worker_set(
    msg: &JsonRpcMessage,
    session: &ClientSession,
) -> Result<Value, String> {
    let bound_agent = session.agent_id.lock().await.clone();
    if let Some(agent) = bound_agent {
        if !session.is_host.load(std::sync::atomic::Ordering::Acquire) {
            return Err(format!(
                "`fleet.worker.set` is operator-only: `{agent}` cannot enroll this machine to \
                 run peers' coding subtasks against local checkouts"
            ));
        }
    }

    let mut config = FleetWorkerConfig::load();
    if let Some(v) = msg.params.get("accepts_work").and_then(|v| v.as_bool()) {
        config.accepts_work = v;
    }
    if let Some(list) = msg.params.get("repos").and_then(|v| v.as_array()) {
        let mut repos = Vec::new();
        for entry in list {
            let path = PathBuf::from(entry.as_str().ok_or("`repos` entries must be strings")?);
            // Validate here rather than at dispatch time: an operator naming a
            // directory that is not a checkout should find out now, not when a
            // peer's subtask is declined for reasons that look like the peer's
            // fault.
            car_fleet::root_commit(&path)
                .map_err(|e| format!("`{}` is not a git repository: {e}", path.display()))?;
            repos.push(path);
        }
        config.repos = repos;
    }
    if let Some(n) = msg.params.get("max_parallel").and_then(|v| v.as_u64()) {
        config.max_parallel = n.min(64) as u32;
    }
    if let Some(n) = msg.params.get("local_parallel").and_then(|v| v.as_u64()) {
        config.local_parallel = n.min(64) as u32;
    }
    if let Some(n) = msg
        .params
        .get("dispatches_per_hour")
        .and_then(|v| v.as_u64())
    {
        config.dispatches_per_hour = n.min(10_000) as u32;
    }
    if let Some(n) = msg.params.get("max_subtask_secs").and_then(|v| v.as_u64()) {
        // A zero would mean "no time at all", which reads as a mistake rather
        // than a posture; the floor keeps a typo from silently declining
        // everything with a timeout error.
        config.max_subtask_secs = n.clamp(60, 24 * 3600);
    }
    if let Some(v) = msg
        .params
        .get("fetch_missing_base")
        .and_then(|v| v.as_bool())
    {
        config.fetch_missing_base = v;
    }
    if let Some(remote) = msg.params.get("fetch_remote").and_then(|v| v.as_str()) {
        // A remote name reaches a `git fetch` argv. Keep it to the shape git
        // itself accepts so nothing that looks like a flag or a URL can be
        // smuggled in through a config write.
        if remote.is_empty()
            || remote.starts_with('-')
            || !remote
                .chars()
                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '/'))
        {
            return Err(format!("`{remote}` is not a usable git remote name"));
        }
        config.fetch_remote = remote.to_string();
    }
    if let Some(list) = msg.params.get("allowed_tools").and_then(|v| v.as_array()) {
        config.allowed_tools = Some(
            list.iter()
                .filter_map(|v| v.as_str().map(String::from))
                .collect(),
        );
    }
    config.save()?;
    Ok(serde_json::json!({
        "config": config,
        "profile": worker_profile().await,
    }))
}

/// Answers the CAR fleet extension methods on the peer-facing A2A listener.
///
/// Holds a weak reference to the daemon state so a stopped listener cannot keep
/// it alive, mirroring `WsChatResponder`.
pub struct DaemonFleetResponder {
    state: std::sync::Weak<ServerState>,
    /// The A2A listener's runtime — the tools a peer could actually reach here,
    /// which is the honest answer to "what capabilities does this host offer
    /// *you*".
    runtime: Arc<car_engine::Runtime>,
}

impl DaemonFleetResponder {
    pub fn new(state: std::sync::Weak<ServerState>, runtime: Arc<car_engine::Runtime>) -> Self {
        Self { state, runtime }
    }
}

#[async_trait::async_trait]
impl car_a2a::FleetResponder for DaemonFleetResponder {
    async fn inventory(&self) -> Result<Value, String> {
        let state = self
            .state
            .upgrade()
            .ok_or_else(|| "daemon is shutting down".to_string())?;
        // No session: a peer sees this host's agents, models, and the tools its
        // own A2A surface exposes — never another session's private skill graph.
        let inv = inventory::local_inventory(&state, Some(&self.runtime), None).await;
        serde_json::to_value(inv).map_err(|e| e.to_string())
    }

    async fn run_subtask(&self, dispatch: Value, caller: Option<&str>) -> Result<Value, String> {
        // No verified caller, no work. The transport that reaches this method
        // authenticates every request by peer signature, so `None` means
        // something changed underneath it — and an unattributable dispatch is
        // exactly the one that cannot be rate-limited or audited.
        let Some(caller) = caller else {
            return Err(
                "refusing an unattributed fleet dispatch: this surface accepts work only from a CAR peer whose signature identifies it"
                    .to_string(),
            );
        };
        let dispatch: car_fleet::SubtaskDispatch =
            serde_json::from_value(dispatch).map_err(|e| format!("invalid dispatch: {e}"))?;
        let outcome = serve::run_dispatch(dispatch, caller).await?;
        serde_json::to_value(outcome).map_err(|e| e.to_string())
    }
}

/// Why a reachable instance is not in the pool.
///
/// Exhaustive and reported, because the alternative is the failure mode that
/// kills a distributed run quietly: every peer turns out to be ineligible, the
/// pool collapses to this host, the run completes correctly but at local speed,
/// and nothing ever says so. The caller asked for distribution and got a slower
/// local run that looks identical to a fast one that happened to be slow.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PoolExclusion {
    /// Reachable, but its operator has not enrolled it as a fleet worker.
    NotEnrolled,
    /// Enrolled, but has no checkout of this repository.
    RepositoryNotServed,
    /// Eligible, but the caller's `workers` list did not name it.
    NotRequested,
    /// Could not be read at all — see the instance's inventory row.
    Unreachable,
    /// Answered, but published no address to dispatch to.
    NoAddress,
}

impl PoolExclusion {
    pub fn as_str(self) -> &'static str {
        match self {
            PoolExclusion::NotEnrolled => "not_enrolled",
            PoolExclusion::RepositoryNotServed => "repository_not_served",
            PoolExclusion::NotRequested => "not_requested",
            PoolExclusion::Unreachable => "unreachable",
            PoolExclusion::NoAddress => "no_address",
        }
    }
}

/// What the pool ended up containing, and what it left out.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct PoolPlan {
    /// Peer instances actually in the pool. Empty means the run is distributed
    /// in name only.
    pub remote_workers: Vec<String>,
    /// Every other reachable instance, with the reason it is not a worker.
    pub excluded: Vec<(String, String)>,
}

impl PoolPlan {
    /// Whether a run asked to be distributed will in fact run only here.
    pub fn local_only(&self) -> bool {
        self.remote_workers.is_empty()
    }

    /// One line a person can act on, or `None` when peers are in the pool.
    pub fn degraded_reason(&self) -> Option<String> {
        if !self.local_only() {
            return None;
        }
        if self.excluded.is_empty() {
            return Some(
                "no other CAR instance is reachable, so this ran on one machine".to_string(),
            );
        }
        let mut counts: std::collections::BTreeMap<&str, usize> = std::collections::BTreeMap::new();
        for (_, reason) in &self.excluded {
            *counts.entry(reason.as_str()).or_default() += 1;
        }
        let detail = counts
            .into_iter()
            .map(|(reason, n)| format!("{n} {reason}"))
            .collect::<Vec<_>>()
            .join(", ");
        Some(format!(
            "no peer could take a subtask, so this ran on one machine ({detail})"
        ))
    }
}

/// Build the worker pool for a distributed Foreman run.
///
/// Always includes this host, so a run whose peers all decline still completes
/// locally rather than failing — distribution is an optimization, and an
/// optimization that can fail the job is a liability. Returns the pool and the
/// worker ids in it, for the run report.
///
/// `only` restricts placement to the named instances (the caller's
/// `workers: [...]` parameter); `None` uses every instance that reports it can
/// serve this repository.
pub async fn build_pool(
    state: &ServerState,
    repo_root: &std::path::Path,
    run_id: &str,
    adapter: &str,
    only: Option<&[String]>,
) -> Result<(car_multi::FleetPool, PoolPlan), String> {
    let fingerprint = car_fleet::read_fingerprint(repo_root).map_err(|e| {
        format!(
            "cannot identify the repository at {}: {e}",
            repo_root.display()
        )
    })?;
    let config = FleetWorkerConfig::load();

    let local: Arc<dyn car_multi::WorktreeAgent> = Arc::new(
        car_external_agents::ForemanExternalAgent::new(adapter.to_string()),
    );
    let mut workers = vec![car_multi::FleetWorker::local(
        car_a2a::lan::host_label(),
        local,
        config.local_parallel.max(1) as usize,
    )];

    let composite = composite(state, None, true, INVENTORY_TIMEOUT).await;
    let identity = {
        state
            .peer_identity
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    };

    // Classify every reachable peer, in or out, so the caller is told when a
    // "distributed" run has quietly become a local one.
    let mut plan = PoolPlan::default();
    let eligible: std::collections::HashSet<&str> = composite
        .workers_for(&fingerprint.root_commit)
        .into_iter()
        .map(|c| c.instance.as_str())
        .collect();
    for inv in &composite.instances {
        if inv.instance.kind == car_fleet::InstanceKind::Local {
            continue;
        }
        let name = inv.instance.name.clone();
        let reason = if !inv.reachable() {
            PoolExclusion::Unreachable
        } else if !eligible.contains(name.as_str()) {
            // Enrolled-but-wrong-repo and not-enrolled-at-all are different
            // problems with different fixes, so they are not collapsed.
            match &inv.worker {
                Some(w) if w.accepts_work => PoolExclusion::RepositoryNotServed,
                _ => PoolExclusion::NotEnrolled,
            }
        } else if only.is_some_and(|only| !only.iter().any(|n| n == &name)) {
            PoolExclusion::NotRequested
        } else if inv.instance.base_url.is_none() {
            PoolExclusion::NoAddress
        } else {
            continue;
        };
        plan.excluded.push((name, reason.as_str().to_string()));
    }

    for candidate in composite.workers_for(&fingerprint.root_commit) {
        if candidate.kind == car_fleet::InstanceKind::Local {
            continue;
        }
        if let Some(only) = only {
            if !only.iter().any(|n| n == &candidate.instance) {
                continue;
            }
        }
        // The peer's own inventory row carries the URL it answered on.
        let Some(base_url) = composite
            .instances
            .iter()
            .find(|i| i.instance.name == candidate.instance)
            .and_then(|i| i.instance.base_url.clone())
        else {
            continue;
        };
        // Ask for the same adapter the local run uses when the peer has it;
        // otherwise let the peer choose, rather than declining a machine that
        // could still help.
        let wanted = candidate
            .adapters
            .iter()
            .any(|a| a == adapter)
            .then(|| adapter.to_string());
        let agent = RemoteWorktreeAgent::new(
            candidate.instance.clone(),
            base_url,
            identity.clone(),
            fingerprint.clone(),
            run_id,
        )
        .with_adapter(wanted);
        plan.remote_workers.push(candidate.instance.clone());
        workers.push(car_multi::FleetWorker::remote(
            candidate.instance.clone(),
            Arc::new(agent),
            candidate.max_parallel.max(1) as usize,
        ));
    }

    if let Some(reason) = plan.degraded_reason() {
        tracing::warn!(%reason, "distributed foreman run has no peer workers");
    }
    Ok((car_multi::FleetPool::new(workers), plan))
}

/// Render a placement ledger for a run report.
///
/// Takes the ledger rather than the pool: `placements()` clones the whole vec
/// under a mutex, so a caller that already holds one would otherwise pay for a
/// second, non-atomic read of the same thing.
///
/// The field names live on `Placement`/`FailedAttempt` as serde renames, which
/// this crate does not own — see their doc comments and the round-trip test that
/// pins them (car#1322).
pub fn placements_value(placements: &[car_multi::Placement]) -> Value {
    // Infallible: plain structs of `String`/`bool`/`Vec`, no map with non-string
    // keys and no custom `Serialize`. A silent `null` would be the wrong answer
    // for a receipt anyway.
    serde_json::to_value(placements).expect("placement ledger is plain data")
}

/// [`placements_value`] for a caller holding the pool.
pub fn placements_json(pool: &car_multi::FleetPool) -> Value {
    placements_value(&pool.placements())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_distributed_run_with_no_peers_says_why_rather_than_going_quiet() {
        let plan = PoolPlan {
            remote_workers: Vec::new(),
            excluded: vec![
                ("studio".into(), "not_enrolled".into()),
                ("laptop".into(), "not_enrolled".into()),
                ("ci-box".into(), "repository_not_served".into()),
            ],
        };
        assert!(plan.local_only());
        let reason = plan.degraded_reason().expect("degraded");
        assert!(reason.contains("2 not_enrolled"), "{reason}");
        assert!(reason.contains("1 repository_not_served"), "{reason}");
    }

    #[test]
    fn a_pool_with_peers_reports_no_degradation() {
        let plan = PoolPlan {
            remote_workers: vec!["studio".into()],
            excluded: vec![("laptop".into(), "not_enrolled".into())],
        };
        assert!(!plan.local_only());
        assert!(plan.degraded_reason().is_none());
    }

    #[test]
    fn no_reachable_peers_at_all_is_its_own_message() {
        let plan = PoolPlan::default();
        let reason = plan.degraded_reason().expect("degraded");
        assert!(
            reason.contains("no other CAR instance is reachable"),
            "{reason}"
        );
    }

    #[test]
    fn the_shipped_default_declines_work() {
        // The one default that must never drift: an instance nobody configured
        // does not run other machines' prompts.
        let c = FleetWorkerConfig::default();
        assert!(!c.accepts_work);
        assert!(c.repos.is_empty());
    }

    #[test]
    fn a_laptop_does_not_fetch_unless_told_to() {
        // Fetching is the runner posture. On a machine someone is working at,
        // a peer's dispatch should not reach into a repository and pull.
        let c = FleetWorkerConfig::default();
        assert!(!c.fetch_missing_base);
        assert_eq!(c.fetch_remote, "origin");
    }

    #[test]
    fn a_config_written_before_runner_mode_existed_still_declines() {
        let parsed: FleetWorkerConfig =
            serde_json::from_str("{\"accepts_work\": true}").expect("parses");
        assert!(
            !parsed.fetch_missing_base,
            "absent must not read as enabled"
        );
        assert_eq!(parsed.fetch_remote, "origin");
    }

    #[test]
    fn the_sender_does_not_choose_this_machines_limits() {
        // Every field here bounds what a peer may spend, so each one must have
        // a default — a config written before they existed must not read as
        // "unlimited".
        let parsed: FleetWorkerConfig =
            serde_json::from_str("{\"accepts_work\": true, \"repos\": []}").expect("parses");
        assert_eq!(
            parsed.dispatches_per_hour,
            car_fleet::DEFAULT_DISPATCHES_PER_WINDOW
        );
        assert_eq!(parsed.max_subtask_secs, DEFAULT_MAX_SUBTASK_SECS);
        assert!(parsed.allowed_tools.is_none());
    }

    #[test]
    fn an_unparseable_config_declines_rather_than_half_accepting() {
        let parsed: FleetWorkerConfig =
            serde_json::from_str("{\"accepts_work\": true}").expect("partial config parses");
        assert!(parsed.accepts_work);
        assert_eq!(parsed.max_parallel, DEFAULT_MAX_PARALLEL, "limits default");
        assert!(
            parsed.repos.is_empty(),
            "and with no repos it can still serve nothing"
        );
    }

    #[test]
    fn worker_worktrees_live_outside_every_served_repository() {
        let base = worktree_base();
        assert!(
            base.ends_with("fleet-worker/worktrees") || base.ends_with("fleet-worker\\worktrees")
        );
    }
}