car_server_core/fleet/

remote.rs

//! The orchestrator side: a [`car_multi::WorktreeAgent`] backed by a peer.
//!
//! From the harness's point of view this is indistinguishable from the local
//! `ForemanExternalAgent` — it is handed a worktree and a subtask, and when it
//! returns the worktree contains the edits. The difference is only where the
//! editing happened: the prompt goes to a peer, the peer runs its own coding
//! CLI in its own worktree at the same base commit, and the patch it returns is
//! applied here.
//!
//! Everything after that is unchanged and local: patch capture, AST containment,
//! duplicate-declaration detection, the build/test leg, the policy consult, and
//! the union integration. A peer cannot widen what gets accepted; it can only
//! produce edits that this host then judges.

use std::sync::Arc;

use async_trait::async_trait;
use car_fleet::{DispatchOutcome, RepoFingerprint, SubtaskDispatch};
use car_multi::{AgentRunSummary, ForemanError, WorktreeAgent, WorktreeAgentRequest};

/// Runs Foreman subtasks on one peer CAR instance.
pub struct RemoteWorktreeAgent {
    /// Peer name as this host's peer listing names it. Appears in the placement
    /// ledger and in every error, so an operator can tell which machine dropped
    /// a subtask.
    pub peer_name: String,
    base_url: String,
    identity: Option<Arc<car_a2a::peer_auth::PeerIdentity>>,
    repo: RepoFingerprint,
    run_id: String,
    adapter: Option<String>,
    timeout_secs: Option<u64>,
}

impl RemoteWorktreeAgent {
    pub fn new(
        peer_name: impl Into<String>,
        base_url: impl Into<String>,
        identity: Option<Arc<car_a2a::peer_auth::PeerIdentity>>,
        repo: RepoFingerprint,
        run_id: impl Into<String>,
    ) -> Self {
        Self {
            peer_name: peer_name.into(),
            base_url: base_url.into(),
            identity,
            repo,
            run_id: run_id.into(),
            adapter: None,
            timeout_secs: None,
        }
    }

    /// Ask the peer for a specific coding CLI. A peer that does not have it
    /// declines rather than substituting — see `car_fleet::worker`.
    pub fn with_adapter(mut self, adapter: Option<String>) -> Self {
        self.adapter = adapter;
        self
    }

    /// Wall-clock bound the peer applies to its CLI invocation.
    pub fn with_timeout_secs(mut self, timeout_secs: Option<u64>) -> Self {
        self.timeout_secs = timeout_secs;
        self
    }
}

#[async_trait]
impl WorktreeAgent for RemoteWorktreeAgent {
    async fn run_in(
        &self,
        req: &WorktreeAgentRequest<'_>,
    ) -> Result<AgentRunSummary, ForemanError> {
        let identity = self.identity.clone().ok_or_else(|| {
            ForemanError::Agent(format!(
                "cannot reach `{}`: this daemon has no peer identity, so a remote CAR would \
                 refuse it. The identity is created when the A2A surface starts.",
                self.peer_name
            ))
        })?;

        let mut dispatch = SubtaskDispatch::new(
            self.run_id.clone(),
            req.subtask.id.clone(),
            req.subtask.prompt.clone(),
            self.repo.clone(),
        );
        dispatch.files = req.subtask.files.clone();
        dispatch.allowed_tools = req.allowed_tools.clone();
        dispatch.adapter = self.adapter.clone();
        dispatch.timeout_secs = self.timeout_secs;

        let params = serde_json::to_value(&dispatch)
            .map_err(|e| ForemanError::Agent(format!("encode dispatch: {e}")))?;
        let client = car_a2a::client::A2aClient::new(&self.base_url)
            .with_http_client(dispatch_client(self.timeout_secs))
            .with_peer_identity(identity);

        // A peer at capacity is momentarily full, not the wrong machine — and
        // this host's view of its capacity is always slightly stale, since other
        // orchestrators are spending it too. So a transient decline gets one
        // more chance before the pool moves on and stops considering this peer
        // for the subtask. Everything else (wrong repo, no CLI, not enrolled)
        // would fail identically on a retry, so it does not get one.
        let mut outcome = self.dispatch_once(&client, &params).await?;
        if retry_after_decline(&outcome) {
            tokio::time::sleep(RETRY_DELAY).await;
            outcome = self.dispatch_once(&client, &params).await?;
        }

        match outcome {
            DispatchOutcome::Declined { reason, detail } => {
                let text = format!(
                    "`{}` declined ({}): {detail}",
                    self.peer_name,
                    reason.as_str()
                );
                Err(if decline_is_worker_level(reason) {
                    ForemanError::Worker(text)
                } else {
                    ForemanError::Agent(text)
                })
            }
            DispatchOutcome::Completed { patch, answer, .. } => {
                if patch.trim().is_empty() {
                    // A no-op is a real answer, and the gate already knows what
                    // to do with an unchanged worktree. Applying an empty patch
                    // would be an error for no reason.
                    return Ok(AgentRunSummary { answer });
                }
                car_multi::git_apply(req.cwd, &patch).map_err(|e| {
                    ForemanError::Agent(format!(
                        "`{}` returned a patch that does not apply to the base it was given: {e}",
                        self.peer_name
                    ))
                })?;
                Ok(AgentRunSummary { answer })
            }
        }
    }
}

/// Whether a decline is a fact about the PEER rather than about the subtask.
///
/// EXHAUSTIVE on `DeclineReason`, deliberately, rather than folding through
/// `retry_window()`. The two questions look alike and are not: that one asks
/// whether the SAME SUBTASK is worth re-offering, this one asks whether ANY
/// remaining subtask is. Answering the second with the first's table means a new
/// variant silently inherits a classification nobody chose — and `PolicyDenied`
/// is already that variant, mapped to `Never` (right, for a retry) with no
/// producer anywhere in the workspace. Whoever wires a policy consult into
/// `run_dispatch` would get whole-run eviction for free, with no compile signal.
/// Two exhaustive matches cannot drift; one table answering an unasked question
/// is exactly what does.
///
/// The `true` arms are the ones whose inputs are fixed for the whole run —
/// enrollment and parallelism from the peer's `FleetWorkerConfig`, the repo and
/// base commit from the one `RepoFingerprint` the agent is built with, the
/// adapter from `with_adapter`, the hourly budget keyed on the caller. None of
/// them read the prompt or the files, so a different subtask gets the identical
/// answer.
fn decline_is_worker_level(reason: car_fleet::DeclineReason) -> bool {
    use car_fleet::DeclineReason as R;
    match reason {
        // The peer is not enrolled as a worker at all.
        R::NotAcceptingWork => true,
        // It does not have this repository, or this base commit. Both are the
        // run's, not the subtask's.
        R::RepoUnavailable | R::CommitUnavailable => true,
        // No installed coding CLI could run it. The adapter is chosen once, when
        // the agent is built.
        R::AdapterUnavailable => true,
        // An hourly budget, charged against the caller. `RetryWindow` calls this
        // "could serve this later, but not on the timescale of a run" — which is
        // the run, for this worker.
        R::RateLimited => true,
        // Momentary capacity. The peer is healthy and `run_in` already retries
        // it once; evicting here would drop a working machine for being busy,
        // which is the normal state of a working fleet.
        R::Busy => false,
        // No producer today. When one arrives it will most plausibly consult
        // policy against the subtask's files, tools, or parameters — per-subtask
        // by nature. Unknown classifies as subtask-level: the cost of guessing
        // wrong here is dropping a healthy peer, and this arm is where the next
        // author is forced to look.
        R::PolicyDenied => false,
    }
}

/// How long to wait to REACH a peer. Not how long it may take to answer.
const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// Headroom over the peer's own work budget for returning the patch.
const PATCH_TRANSFER_SLACK: std::time::Duration = std::time::Duration::from_secs(60);

/// How long the far side may take to answer one dispatch.
///
/// `A2aClient`'s default is a 30-second TOTAL request deadline — right for the
/// small RPCs the rest of the A2A surface makes, and catastrophically wrong
/// here: the peer runs the coding CLI INSIDE this request and is allowed
/// `max_subtask_secs`, half an hour by default, to finish. Thirty seconds
/// against a thirty-minute budget means the orchestrator hangs up on a peer that
/// is still editing, for every subtask worth farming out.
///
/// The dispatch carries no negotiated budget — `SubtaskDispatch::timeout_secs`
/// is `None` unless a caller sets one — so this falls back to the same constant
/// the peer clamps to, and adds slack for the patch coming back over the wire.
fn dispatch_deadline(timeout_secs: Option<u64>) -> std::time::Duration {
    let budget = timeout_secs.unwrap_or(super::DEFAULT_MAX_SUBTASK_SECS);
    std::time::Duration::from_secs(budget) + PATCH_TRANSFER_SLACK
}

/// The HTTP client one farmed-out subtask is dispatched on.
///
/// Two deadlines, because they answer different questions. `connect_timeout`
/// bounds REACHING the machine and is short — a CAR peer is on a LAN or a known
/// address, five seconds is generous for a handshake, and it is what makes a
/// powered-off peer fail fast instead of consuming the whole request deadline.
/// `timeout` bounds the WORK; see [`dispatch_deadline`].
///
/// That split is also what lets [`call_error_is_worker_level`] tell a dead
/// machine from a slow one at all — collapsed into one deadline, both arrive as
/// a timeout and are indistinguishable.
fn dispatch_client(timeout_secs: Option<u64>) -> reqwest::Client {
    reqwest::Client::builder()
        .connect_timeout(CONNECT_TIMEOUT)
        .timeout(dispatch_deadline(timeout_secs))
        .build()
        // Same posture as `A2aClient::new`: this configuration is static and
        // valid, and a caller has nothing useful to do with the failure.
        .unwrap_or_else(|_| reqwest::Client::new())
}

/// Whether a failed `car/foremanSubtask` call means the peer is unreachable.
///
/// Narrow twice over.
///
/// `ClientError::Transport` wraps ANY `reqwest::Error`, including the body-decode
/// failure from `resp.json()`, so matching the variant would quarantine a live
/// peer that answered non-JSON.
///
/// And within that, `is_connect()` ALONE — not `is_timeout()`. Measured, not
/// assumed (`reqwest_tells_a_dead_peer_from_a_slow_one`): a refused connection
/// and a connect-phase timeout both set `is_connect()`, while a peer that
/// accepted the connection and is still working sets only `is_timeout()`. Since
/// the peer runs the coding CLI inside this request, that last case is the
/// COMMON one, and treating it as death would quarantine the whole fleet on the
/// first subtask that took more than a moment.
///
/// Everything else — an HTTP status, a JSON-RPC error envelope, a shape this
/// host cannot parse — is the peer ANSWERING. It may well be misconfigured or
/// version-skewed, but that is a claim this function cannot make from one reply,
/// and the cost of being wrong is dropping a healthy worker.
fn call_error_is_worker_level(err: &car_a2a::client::ClientError) -> bool {
    match err {
        car_a2a::client::ClientError::Transport(e) => e.is_connect(),
        car_a2a::client::ClientError::Status { .. }
        | car_a2a::client::ClientError::Serialize(_)
        | car_a2a::client::ClientError::Rpc { .. }
        | car_a2a::client::ClientError::BadResultShape(_)
        | car_a2a::client::ClientError::Malformed(_) => false,
    }
}

/// Wait before re-offering a subtask to a peer that was full.
///
/// Short: the pool has other workers, and the point is to catch a peer that was
/// finishing something, not to wait out a queue.
const RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(2);

/// Whether an outcome is worth one immediate re-dispatch to the same peer.
///
/// Pure so the policy is testable without a peer: the interesting property is
/// that exactly one decline reason (congestion) is retried and every other —
/// each of which is a fact about that machine's configuration or checkout —
/// is not.
fn retry_after_decline(outcome: &DispatchOutcome) -> bool {
    match outcome {
        DispatchOutcome::Declined { reason, .. } => {
            // Only congestion. A rate-limited peer *could* serve this later, but
            // not on the timescale of a run — re-offering it two seconds later
            // buys the identical answer a round trip after the first one.
            matches!(reason.retry_window(), car_fleet::RetryWindow::Immediately)
        }
        DispatchOutcome::Completed { .. } => false,
    }
}

impl RemoteWorktreeAgent {
    /// One `car/foremanSubtask` round trip, decoded.
    async fn dispatch_once(
        &self,
        client: &car_a2a::client::A2aClient,
        params: &serde_json::Value,
    ) -> Result<DispatchOutcome, ForemanError> {
        let raw: serde_json::Value =
            client
                .call("car/foremanSubtask", params)
                .await
                .map_err(|e| {
                    let text = format!("`{}` did not run the subtask: {e}", self.peer_name);
                    if call_error_is_worker_level(&e) {
                        ForemanError::Worker(text)
                    } else {
                        ForemanError::Agent(text)
                    }
                })?;
        serde_json::from_value(raw).map_err(|e| {
            ForemanError::Agent(format!("`{}` answered unusably: {e}", self.peer_name))
        })
    }
}

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

    fn fingerprint() -> RepoFingerprint {
        RepoFingerprint {
            root_commit: "root".into(),
            head_commit: "head".into(),
            name: Some("car".into()),
        }
    }

    #[tokio::test]
    async fn without_a_peer_identity_it_says_so_locally() {
        // The far side would answer 401 and the operator would go looking at
        // the wrong machine. The fix is local, so the message is too.
        let agent =
            RemoteWorktreeAgent::new("studio", "http://studio:8731", None, fingerprint(), "run-1");
        let subtask = Subtask::files_only("s", "do it", vec![]);
        let cwd = std::path::PathBuf::from(".");
        let req = WorktreeAgentRequest {
            subtask: &subtask,
            cwd: &cwd,
            allowed_tools: None,
            mcp_endpoint: None,
        };
        let err = agent.run_in(&req).await.unwrap_err();
        assert!(err.to_string().contains("peer identity"), "{err}");
    }

    #[test]
    fn only_a_full_peer_is_offered_the_subtask_twice() {
        let declined = |reason| DispatchOutcome::Declined {
            reason,
            detail: String::new(),
        };
        assert!(retry_after_decline(&declined(DeclineReason::Busy)));
        for reason in [
            DeclineReason::RateLimited,
            DeclineReason::NotAcceptingWork,
            DeclineReason::RepoUnavailable,
            DeclineReason::CommitUnavailable,
            DeclineReason::AdapterUnavailable,
            DeclineReason::PolicyDenied,
        ] {
            assert!(
                !retry_after_decline(&declined(reason)),
                "{reason:?} would decline identically on a retry"
            );
        }
        assert!(!retry_after_decline(&DispatchOutcome::Completed {
            patch: String::new(),
            answer: String::new(),
            adapter: "claude-code".into(),
            duration_ms: 1,
        }));
    }

    /// The boundary the pool's quarantine rests on: which declines are a fact
    /// about the MACHINE (so it will answer every remaining subtask the same
    /// way) rather than about the subtask that was offered.
    ///
    /// `Busy` is the one that must stay subtask-level among the reasons that
    /// exist. Getting it wrong removes a healthy peer for being momentarily
    /// full — the normal state of a working fleet.
    #[test]
    fn only_a_momentarily_full_peer_stays_in_the_pool() {
        for reason in [
            DeclineReason::NotAcceptingWork,
            DeclineReason::RepoUnavailable,
            DeclineReason::CommitUnavailable,
            DeclineReason::AdapterUnavailable,
            DeclineReason::RateLimited,
        ] {
            assert!(
                decline_is_worker_level(reason),
                "{reason:?} answers every subtask of this run identically"
            );
        }
        assert!(!decline_is_worker_level(DeclineReason::Busy));
    }

    /// `PolicyDenied` has no producer in the workspace. It must NOT evict, and
    /// this test exists to say that is a decision rather than an oversight: a
    /// policy consult will most plausibly read the subtask's files or tools, so
    /// treating it as a fact about the machine would drop a healthy peer the
    /// first time one subtask touched a denied path.
    #[test]
    fn an_unwired_decline_reason_does_not_evict_the_peer() {
        assert!(!decline_is_worker_level(DeclineReason::PolicyDenied));
    }

    /// What `reqwest` actually reports, because the whole quarantine rests on
    /// it and the two cases are one variant apart.
    ///
    /// A peer runs the coding CLI INSIDE the dispatch request, so "nothing came
    /// back yet" is the healthy case and must never be read as death. Measured
    /// against real sockets rather than asserted from the docs.
    #[tokio::test]
    async fn reqwest_tells_a_dead_peer_from_a_slow_one() {
        // Nothing listening: bind, read the port, drop the listener.
        let dead_port = {
            let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            l.local_addr().unwrap().port()
        };
        let refused = reqwest::Client::new()
            .post(format!("http://127.0.0.1:{dead_port}/"))
            .send()
            .await
            .expect_err("nothing is listening");
        assert!(refused.is_connect(), "a refused connection is a dead peer");

        // Accepts the connection, then never answers — a peer mid-subtask.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let busy_port = listener.local_addr().unwrap().port();
        let held = tokio::spawn(async move {
            let _accepted = listener.accept().await;
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
        });
        let slow = reqwest::Client::builder()
            .timeout(std::time::Duration::from_millis(300))
            .build()
            .unwrap()
            .post(format!("http://127.0.0.1:{busy_port}/"))
            .send()
            .await
            .expect_err("the deadline passes before an answer");
        assert!(slow.is_timeout(), "it is a timeout");
        assert!(
            !slow.is_connect(),
            "but NOT a connect failure — the peer is there and working"
        );
        held.abort();
    }

    /// The deadline the dispatch runs under has to cover the work the far side
    /// is allowed to spend. `A2aClient`'s 30-second default — what this used to
    /// inherit — is less than a fiftieth of it, so the orchestrator hung up on
    /// every peer that was still editing.
    #[test]
    fn the_dispatch_deadline_covers_the_peer_s_whole_budget() {
        // The unnegotiated case, which is every case today: the peer clamps to
        // its own ceiling, so the orchestrator must wait at least that long.
        let default = dispatch_deadline(None);
        assert!(
            default >= std::time::Duration::from_secs(super::super::DEFAULT_MAX_SUBTASK_SECS),
            "hanging up before the peer's own ceiling aborts work that is still running"
        );
        assert!(
            default > std::time::Duration::from_secs(30),
            "the A2aClient default is what this exists to replace"
        );
        // And a negotiated budget is honoured, plus slack for the patch.
        assert_eq!(
            dispatch_deadline(Some(60)),
            std::time::Duration::from_secs(60) + PATCH_TRANSFER_SLACK
        );
        // Reaching the machine is a separate, much shorter question.
        assert!(CONNECT_TIMEOUT < default);
    }

    /// A peer that is not there vs. a peer that answered badly.
    ///
    /// Real sockets, because the distinction lives inside `reqwest::Error` —
    /// `ClientError::Transport` wraps the body-decode failure too, so matching
    /// the variant would quarantine a live peer that returned non-JSON. A
    /// hand-built error could not tell these apart.
    #[tokio::test]
    async fn a_peer_that_is_gone_is_worker_level_and_one_that_answers_badly_is_not() {
        use car_a2a::client::A2aClient;

        // Nothing listening: bind, read the port, drop the listener.
        let dead_port = {
            let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            l.local_addr().unwrap().port()
        };
        let err = A2aClient::new(format!("http://127.0.0.1:{dead_port}"))
            .call::<_, serde_json::Value>("car/foremanSubtask", &serde_json::json!({}))
            .await
            .expect_err("nothing is listening");
        assert!(
            call_error_is_worker_level(&err),
            "connect refused means the machine is not there: {err}"
        );

        // Listening, answers 200 with something that is not JSON.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let live_port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            if let Ok((mut sock, _)) = listener.accept().await {
                use tokio::io::AsyncWriteExt;
                let _ = sock
                    .write_all(
                        b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
                          content-length: 8\r\n\r\nnot json",
                    )
                    .await;
                let _ = sock.flush().await;
            }
        });
        let err = A2aClient::new(format!("http://127.0.0.1:{live_port}"))
            .call::<_, serde_json::Value>("car/foremanSubtask", &serde_json::json!({}))
            .await
            .expect_err("the body does not parse");
        assert!(
            !call_error_is_worker_level(&err),
            "a peer that ANSWERED is alive, however unusably: {err}"
        );
    }
}