car-server-core 0.55.0

//! The worker side: running a subtask a peer farmed out to this instance.
//!
//! Accepting fleet work means running another machine's prompt through a coding
//! CLI against a checkout on this one. That is a real grant, so it is off until
//! an operator turns it on ([`super::FleetWorkerConfig`]), it is bounded
//! (`max_parallel`, and only the repositories named in the config), it reaches
//! this daemon only over the peer-signature-authenticated A2A listener, and
//! every accepted or declined dispatch leaves an audit record.
//!
//! Three of those bounds are about not being spent by someone else more than
//! the operator meant, and each answers a distinct failure:
//!
//! - **`dispatches_per_hour`** — `max_parallel` bounds concurrency, which is
//!   not a bound on spend. A peer that dispatches serially never exceeds a
//!   concurrency limit and can still drain this machine's coding-CLI quota.
//!   Charged last, immediately before the CLI runs, so the declines that cost
//!   this machine nothing (wrong repo, no adapter) do not spend the caller's
//!   allowance — and charged against the **payer**, not the calling device, or
//!   the allowance would scale with how many machines the caller owns
//!   (`budget_key`).
//! - **The clamps** — `timeout_secs` and `allowed_tools` arrive on the
//!   *dispatch*, meaning the sender chose them. A machine agreeing to take work
//!   is not agreeing to whatever limits the caller wrote down, so both are
//!   narrowed against this host's own config before anything runs.
//! - **Attribution** — every record names the peer's verified key fingerprint,
//!   taken from the request signature and never from the dispatch body. Without
//!   it an operator can see that something spent their machine and not what.
//!
//! What a worker deliberately does **not** do: build, test, gate, or integrate.
//! It edits a throwaway worktree checked out at the base commit the dispatch
//! names and returns the patch. The orchestrator gates that patch with its own
//! rules on its own machine, which is what keeps a bad worker's output a
//! rejection rather than a compromise.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};

use car_fleet::{DeclineReason, DispatchOutcome, SubtaskDispatch, WorkGuard};

/// Subtasks currently running for peers, against `max_parallel`.
///
/// Process-global rather than per-connection: the bound is a property of this
/// machine's capacity, and a peer that opened three connections must not get
/// three times the concurrency.
static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0);

/// Per-payer dispatch budget. Process-global for the same reason as
/// [`IN_FLIGHT`]: the limit is a property of this machine's quota, and a caller
/// that opened three connections must not get three budgets.
fn work_guard() -> &'static std::sync::Mutex<WorkGuard> {
    static GUARD: std::sync::OnceLock<std::sync::Mutex<WorkGuard>> = std::sync::OnceLock::new();
    GUARD.get_or_init(|| std::sync::Mutex::new(WorkGuard::default()))
}

/// The single budget bucket every trusted caller shares today.
///
/// See [`budget_key`] for why one bucket is the correct granularity, and what
/// has to be true before it stops being.
const SAME_ACCOUNT_FLEET_BUDGET: &str = "same-account-fleet";

/// Which budget a caller spends from.
///
/// **Not the device fingerprint**, which is what a first cut naturally reaches
/// for and what makes the limit wrong: a person with a laptop, a desktop and a
/// CI box would get three separate allowances against this machine, so the
/// budget would scale with the caller's hardware instead of bounding it.
///
/// The right granularity is the **payer**, and today every accepted caller has
/// the same one. `refresh_peer_trust` builds this host's trusted key set purely
/// from `host_endpoints()` — the devices on its OWN login's end-to-end encrypted
/// roster — and installs an empty set when sync is unconfigured. So a caller
/// that gets past the peer-auth layer is provably a device of this account, and
/// all of them spend the one subscription seat this machine holds. One bucket.
///
/// This stops being right the moment a second principal can be trusted — an
/// org-attested member from another login. At that point the caller asserts a
/// verified member identity and this returns it, so the budget follows the payer
/// rather than the hardware. That is a one-function change by design; the guard
/// itself already keys on whatever string it is handed.
fn budget_key(_caller: &str) -> &'static str {
    SAME_ACCOUNT_FLEET_BUDGET
}

/// Decrements the in-flight count on every return path, including a panic
/// inside the CLI invocation.
struct InFlightGuard;

impl Drop for InFlightGuard {
    fn drop(&mut self) {
        IN_FLIGHT.fetch_sub(1, Ordering::SeqCst);
    }
}

/// A `git worktree` checked out at an explicit commit, removed on drop.
///
/// `car_multi::AgentWorkspace` cannot serve here: it checks out `HEAD`, and a
/// worker must reproduce the *orchestrator's* base commit, which is generally
/// not this machine's HEAD.
struct WorkerWorktree {
    repo: PathBuf,
    path: PathBuf,
}

impl WorkerWorktree {
    fn provision(repo: &Path, base: &Path, name: &str, commit: &str) -> Result<Self, String> {
        let path = base.join(name);
        // Self-heal a worktree a crashed run left registered, exactly as the
        // foreman harness does — otherwise one interrupted dispatch poisons
        // that subtask id forever.
        let _ = git(
            repo,
            &["worktree", "remove", "--force", &path.to_string_lossy()],
        );
        let _ = git(repo, &["worktree", "prune"]);
        if path.exists() {
            let _ = std::fs::remove_dir_all(&path);
        }
        std::fs::create_dir_all(base).map_err(|e| format!("create {}: {e}", base.display()))?;
        git(
            repo,
            &[
                "worktree",
                "add",
                "--detach",
                &path.to_string_lossy(),
                commit,
            ],
        )?;
        Ok(Self {
            repo: repo.to_path_buf(),
            path,
        })
    }
}

impl Drop for WorkerWorktree {
    fn drop(&mut self) {
        let _ = git(
            &self.repo,
            &[
                "worktree",
                "remove",
                "--force",
                &self.path.to_string_lossy(),
            ],
        );
        if self.path.exists() {
            let _ = std::fs::remove_dir_all(&self.path);
        }
    }
}

fn git(cwd: &Path, args: &[&str]) -> Result<String, String> {
    let out = std::process::Command::new("git")
        .arg("-C")
        .arg(cwd)
        .args(args)
        .output()
        .map_err(|e| format!("run git {}: {e}", args.join(" ")))?;
    if !out.status.success() {
        return Err(format!(
            "git {} failed: {}",
            args.join(" "),
            String::from_utf8_lossy(&out.stderr).trim()
        ));
    }
    Ok(String::from_utf8_lossy(&out.stdout).trim().to_string())
}

/// Run one farmed-out subtask, or decline it.
///
/// Never returns `Err` for a refusal: a decline is a routing answer the
/// orchestrator acts on (place it elsewhere), and only an internal failure —
/// something this host got wrong — is an error.
pub async fn run_dispatch(
    dispatch: SubtaskDispatch,
    caller: &str,
) -> Result<DispatchOutcome, String> {
    let started = std::time::Instant::now();
    let config = super::FleetWorkerConfig::load();

    if !config.accepts_work {
        return Ok(declined(
            caller,
            &dispatch,
            DeclineReason::NotAcceptingWork,
            "this instance is not enrolled as a fleet worker (`fleet.worker.set`)",
        ));
    }

    // Admission before anything expensive. `fetch_update` rather than a
    // fetch_add-then-compare so two concurrent dispatches cannot both observe
    // the last free slot.
    let admitted = IN_FLIGHT
        .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
            (n < config.max_parallel as usize).then_some(n + 1)
        })
        .is_ok();
    if !admitted {
        return Ok(declined(
            caller,
            &dispatch,
            DeclineReason::Busy,
            &format!("already running {} subtasks", config.max_parallel),
        ));
    }
    let _guard = InFlightGuard;

    let repo = match car_fleet::resolve_repo(&dispatch.repo, &config.repos) {
        Ok(path) => path.to_path_buf(),
        // Right repository, missing commit — the case a **runner** treats as
        // work rather than a refusal. Fetching happens in a blocking task: it
        // is a network round trip, and holding the dispatch executor on it
        // would stall every other subtask this machine is running.
        Err(DeclineReason::CommitUnavailable) if config.fetch_missing_base => {
            let clones: Vec<std::path::PathBuf> =
                car_fleet::clones_of(&dispatch.repo, &config.repos)
                    .into_iter()
                    .map(|p| p.to_path_buf())
                    .collect();
            let remote = config.fetch_remote.clone();
            let wanted = dispatch.repo.head_commit.clone();
            let fetched = tokio::task::spawn_blocking(move || {
                clones
                    .into_iter()
                    .find(|path| car_fleet::fetch_base(path, &remote, &wanted))
            })
            .await
            .map_err(|e| format!("fetch task panicked: {e}"))?;
            match fetched {
                Some(path) => path,
                None => {
                    return Ok(declined(
                        caller,
                        &dispatch,
                        DeclineReason::CommitUnavailable,
                        "the base commit is not on this worker's remote either — push the \
                         branch before distributing",
                    ))
                }
            }
        }
        Err(reason) => {
            return Ok(declined(
                caller,
                &dispatch,
                reason,
                match reason {
                    DeclineReason::CommitUnavailable => {
                        "the repository is here but the base commit is not — push it somewhere \
                         this worker can fetch from, or run the subtask locally"
                    }
                    _ => "no configured checkout of that repository",
                },
            ))
        }
    };

    let adapters = super::detected_adapters().await;
    let adapter = match dispatch.adapter.as_deref() {
        // An explicit request is honoured only if that adapter is installed;
        // silently substituting a different CLI would make an orchestrator's
        // model choice a fiction.
        Some(wanted) => adapters
            .iter()
            .find(|a| a.id == wanted)
            .map(|a| a.id.clone()),
        None => adapters.first().map(|a| a.id.clone()),
    };
    let Some(adapter) = adapter else {
        return Ok(declined(
            caller,
            &dispatch,
            DeclineReason::AdapterUnavailable,
            "no installed coding CLI could run this subtask",
        ));
    };

    // The budget is charged LAST among the checks, immediately before work
    // starts, and against the PAYER rather than the calling device (see
    // `budget_key`). Every decline above this line cost this machine nothing — a
    // dispatch for a repository it does not have spends no quota — and charging
    // for those would let a misconfigured orchestrator lock itself out for an
    // hour without ever having been served. The guard counts on admission, so
    // reaching here is the commitment to run.
    let verdict = {
        let now = car_fleet::now_ms();
        let mut guard = work_guard().lock().unwrap_or_else(|e| e.into_inner());
        guard.set_limit(config.dispatches_per_hour);
        guard.evict_idle(now);
        guard.admit(budget_key(caller), now)
    };
    if !verdict.is_accept() {
        return Ok(declined(
            caller,
            &dispatch,
            DeclineReason::RateLimited,
            &verdict.reason(),
        ));
    }

    let base = super::worktree_base();
    let name = sanitize(&format!("{}-{}", dispatch.run_id, dispatch.subtask_id));
    let worktree = match WorkerWorktree::provision(&repo, &base, &name, &dispatch.repo.head_commit)
    {
        Ok(w) => w,
        Err(e) => return Err(format!("provision worktree: {e}")),
    };

    let subtask = car_multi::Subtask::files_only(
        dispatch.subtask_id.clone(),
        dispatch.prompt.clone(),
        dispatch.files.clone(),
    );
    let request = car_multi::WorktreeAgentRequest {
        subtask: &subtask,
        cwd: &worktree.path,
        allowed_tools: narrow_tools(
            dispatch.allowed_tools.clone(),
            config.allowed_tools.as_deref(),
        ),
        // No MCP endpoint: the orchestrator's daemon URL is not reachable from
        // here, and pointing a worker at this daemon's own MCP surface would
        // route a peer's tool calls through local policy under the local
        // principal. Worker CLIs run with their own built-in tools only.
        mcp_endpoint: None,
        // And so no MCP config directory: with no endpoint the adapter writes
        // no config file at all, so there is nothing for `$TMPDIR` to hold
        // (car#1534). This is not a session, so there is no coder state dir to
        // point at either.
        mcp_config_dir: None,
    };
    let mut agent = car_external_agents::ForemanExternalAgent::new(adapter.clone());
    // The sender proposes; this machine disposes. An omitted timeout is not
    // "unbounded", it is this host's ceiling.
    agent.timeout_secs = Some(
        dispatch
            .timeout_secs
            .unwrap_or(config.max_subtask_secs)
            .min(config.max_subtask_secs),
    );

    let answer = match car_multi::WorktreeAgent::run_in(&agent, &request).await {
        Ok(summary) => summary.answer,
        Err(e) => {
            audit(caller, &dispatch, "error", &e.to_string());
            return Err(format!("subtask failed: {e}"));
        }
    };

    let patch = {
        let cwd = worktree.path.clone();
        match tokio::task::spawn_blocking(move || car_multi::capture_patch(&cwd)).await {
            Ok(Ok(p)) => p,
            Ok(Err(e)) => {
                audit(caller, &dispatch, "error", &e.to_string());
                return Err(format!("capture patch: {e}"));
            }
            Err(e) => return Err(format!("capture task panicked: {e}")),
        }
    };

    audit(
        caller,
        &dispatch,
        "completed",
        &format!("{} bytes via {adapter}", patch.len()),
    );
    Ok(DispatchOutcome::Completed {
        patch,
        answer,
        adapter,
        duration_ms: started.elapsed().as_millis() as u64,
    })
}

/// Narrow a sender's tool allowlist against this machine's own.
///
/// Intersection, never union: whichever side is stricter wins. An unset worker
/// allowlist adds no restriction (see `FleetWorkerConfig::allowed_tools`), and
/// an unset *sender* list with a worker list set means the worker's list — a
/// caller that named nothing does not thereby get everything.
fn narrow_tools(requested: Option<Vec<String>>, worker: Option<&[String]>) -> Option<Vec<String>> {
    match (requested, worker) {
        (_, None) => None,
        (None, Some(worker)) => Some(worker.to_vec()),
        (Some(requested), Some(worker)) => Some(
            requested
                .into_iter()
                .filter(|t| worker.iter().any(|w| w == t))
                .collect(),
        ),
    }
}

fn declined(
    caller: &str,
    dispatch: &SubtaskDispatch,
    reason: DeclineReason,
    detail: &str,
) -> DispatchOutcome {
    audit(caller, dispatch, reason.as_str(), detail);
    DispatchOutcome::Declined {
        reason,
        detail: detail.to_string(),
    }
}

/// One line per dispatch, accepted or refused.
///
/// Running a peer's prompt against a local checkout is the most consequential
/// thing the fleet surface does, so it leaves a record whatever the outcome —
/// including the refusals, which are how an operator notices a peer repeatedly
/// asking for a repository this machine will not serve.
fn audit(caller: &str, dispatch: &SubtaskDispatch, outcome: &str, detail: &str) {
    use std::io::Write;
    let Some(dir) = car_home::root() else {
        return;
    };
    if std::fs::create_dir_all(&dir).is_err() {
        return;
    }
    let record = serde_json::json!({
        "ts": chrono::Utc::now().to_rfc3339(),
        // The peer's verified key fingerprint, from the request signature —
        // never anything the dispatch says about itself. An audit that cannot
        // name who spent the machine answers the wrong question.
        "peer": caller,
        "run_id": dispatch.run_id,
        "subtask_id": dispatch.subtask_id,
        "repo_root_commit": dispatch.repo.root_commit,
        "base_commit": dispatch.repo.head_commit,
        "outcome": outcome,
        "detail": detail,
    });
    let Ok(line) = serde_json::to_string(&record) else {
        return;
    };
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(dir.join("fleet-work.jsonl"))
    {
        let _ = writeln!(f, "{line}");
    }
}

/// Collapse a dispatch's ids into one safe path segment. Ids come from another
/// host, so nothing here may reach outside the worktree base.
fn sanitize(name: &str) -> String {
    let s: String = name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '-'
            }
        })
        .collect();
    let trimmed = s.trim_matches('-').to_string();
    if trimmed.is_empty() {
        "subtask".to_string()
    } else {
        trimmed.chars().take(96).collect()
    }
}

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

    fn dispatch() -> SubtaskDispatch {
        SubtaskDispatch::new(
            "run",
            "sub",
            "do the thing",
            RepoFingerprint {
                root_commit: "r".into(),
                head_commit: "h".into(),
                name: None,
            },
        )
    }

    #[test]
    fn a_peers_ids_cannot_escape_the_worktree_base() {
        assert_eq!(sanitize("../../etc/passwd"), "etc-passwd");
        // Every separator collapses to a dash rather than being dropped: the
        // property is that nothing traversable survives, not that the result is
        // pretty.
        assert_eq!(sanitize("run/../..\\x"), "run-------x");
        for hostile in ["../../etc/passwd", "run/../..\\x", "a/b/c", "..\\..\\win"] {
            let safe = sanitize(hostile);
            assert!(!safe.contains('/') && !safe.contains('\\'), "{safe}");
            assert!(!safe.contains(".."), "{safe}");
        }
        assert_eq!(sanitize(""), "subtask");
        assert_eq!(sanitize("---"), "subtask");
        assert!(!sanitize(&"x".repeat(500)).contains('/'));
        assert_eq!(sanitize(&"x".repeat(500)).len(), 96);
    }

    #[test]
    fn every_trusted_caller_spends_one_budget_not_one_per_machine() {
        // The bug this encodes: keying on the device fingerprint gives a person
        // with three machines three allowances against this worker, so the
        // limit scales with the caller's hardware instead of bounding it. Every
        // caller that gets past peer auth is a device of the same account
        // (`refresh_peer_trust` trusts only this login's roster), so they share
        // one bucket.
        assert_eq!(budget_key("aa:bb:cc:dd"), budget_key("11:22:33:44"));
    }

    #[test]
    fn the_stricter_tool_list_wins_in_both_directions() {
        // Sender asks for more than the worker allows → intersection.
        assert_eq!(
            narrow_tools(
                Some(vec!["Read".into(), "Bash".into()]),
                Some(&["Read".to_string()])
            ),
            Some(vec!["Read".into()])
        );
        // Sender names nothing while the worker restricts → the worker's list,
        // not "everything".
        assert_eq!(
            narrow_tools(None, Some(&["Read".to_string()])),
            Some(vec!["Read".into()])
        );
        // Worker unrestricted → the sender's list passes through unchanged.
        assert_eq!(narrow_tools(Some(vec!["Bash".into()]), None), None);
        // Disjoint → empty, which the CLI reads as "no tools", not "all".
        assert_eq!(
            narrow_tools(Some(vec!["Bash".into()]), Some(&["Read".to_string()])),
            Some(Vec::new())
        );
    }

    #[test]
    fn a_senders_timeout_cannot_exceed_this_machines_ceiling() {
        let ceiling = 1800u64;
        let clamp = |requested: Option<u64>| requested.unwrap_or(ceiling).min(ceiling);
        assert_eq!(clamp(Some(60)), 60, "a shorter request is honoured");
        assert_eq!(clamp(Some(99_999)), ceiling, "a longer one is clamped");
        assert_eq!(
            clamp(None),
            ceiling,
            "omitted is the ceiling, not unbounded"
        );
    }

    #[tokio::test]
    async fn a_daemon_that_did_not_opt_in_declines_before_touching_anything() {
        // The config file is process-global state under CAR_HOME; the shipped
        // default is what matters here and it declines.
        if super::super::FleetWorkerConfig::load().accepts_work {
            // An operator's real config is present in this environment — the
            // assertion below would be testing their machine, not the default.
            return;
        }
        let outcome = run_dispatch(dispatch(), "aa:bb:cc:dd")
            .await
            .expect("no internal error");
        assert!(matches!(
            outcome,
            DispatchOutcome::Declined {
                reason: DeclineReason::NotAcceptingWork,
                ..
            }
        ));
    }
}