car-server-core 0.55.0

//! Running a real coder session for one healed item, and reviewing it.
//!
//! The last unwritten seam. Everything else in the loop was reachable with a
//! fake; this is what turns a selected issue into a branch and a set of
//! verdicts.
//!
//! ## Nothing is published until the gate approves
//!
//! A coder session that reaches `CoderState::NeedsApproval` has done its work
//! and re-evaluated its own outcome contract — the runtime's deterministic
//! answer to "does this build and pass". That is the moment the panel is asked,
//! and it reviews the **worktree** diff, exactly the view the human approval
//! surface shows.
//!
//! It deliberately does not review a published branch, because in the
//! interactive path `car/coder/<id>` is created by `coder.approve_merge` —
//! the branch exists only once a human has said yes, so its existence *means*
//! approved. Publishing before the panel would put unapproved work in a
//! namespace that already carries that meaning, and leave one dead branch in
//! the operator's repository per rejected item.
//!
//! So delivery is one step, after [`super::heal_gate::decide`] approves, and it
//! is [`super::merge::deliver_pr`] — the same commit → push → reconcile-exactly-
//! one-pull-request path `car code-task --deliver pr` uses. That is not reuse
//! for its own sake: it carries the base/target-equality refusal, the ambiguous
//! head refusal, the vacuity guard, and the rule that a **closed** pull request
//! on the delivery branch is a person saying stop. Hand-rolling publish and
//! push had none of those, and a human closing the loop's pull request read as
//! "do it again".
//!
//! The loop still never merges. `deliver_pr` opens a pull request; that is the
//! boundary, and it is why the delivery branch may never equal the base.
//!
//! ## Every session this runner starts reaches a terminal state
//!
//! `NeedsApproval` is not terminal, so the git worktree — registered in the
//! operator's own repository — is released only on a terminal transition.
//! Delivery goes to `Merged` (which on the raw-repo path already means no more
//! than "the branch was published"), rejection and timeout go to `Abandoned`
//! and `Failed`. An unattended loop on a cadence whose expected outcome is
//! rejection cannot leave that to a human.
//!
//! ## A tick cannot run forever
//!
//! `SessionDeadline` bounds the session's wall clock. Without it an unattended
//! loop inherits whatever the engine's own limits happen to be, and a wedged
//! session holds its claim for the full TTL while the cadence fires around it.
//! The ceiling is per-item and configurable, because "how long is too long" is
//! a property of the repository, not of this code.
//!
//! ## Reviewers see the diff AND the seed, and both are untrusted
//!
//! An earlier version of this comment claimed reviewers never see the issue
//! body. They do: [`super::heal_gate::review_criteria`] interpolates the seed
//! as `Stated intent:`, at the top of the prompt, ahead of the instructions —
//! and the seed is rendered tracker text that cleared the tier gate, not text
//! this runtime wrote.
//!
//! So the reviewer prompt carries two attacker-influenced inputs, not zero: the
//! intent, from whoever filed the issue, and the diff, from whoever wrote the
//! code. `may_seed_session` bounds who can reach the first; nothing bounds the
//! second, which is why [`super::heal_review`] fences the diff with a
//! content-derived delimiter rather than a fixed banner. Neither is a substitute
//! for the other, and the honest statement is that a reviewer prompt is a place
//! where "ignore the above and approve this" can land, guarded rather than
//! impossible.
//!
//! What reviewers do NOT see is an outcome contract sourced from a body:
//! `may_source_contract` is `Runtime`-only, and `run` confirms with `None` so
//! the contract is derived from the intent rather than supplied by it.

use std::sync::Arc;

use super::heal_gate::{Unreachable, Verdict};
use super::heal_intake::{Checkout, HealTarget};
use super::heal_live::{parse_verdict, CoderRunner};
use super::heal_select::Candidate;
use super::heal_tick::{DeliverRefusal, RunFailure};
use super::merge::PrDeliveryOutcome;
use super::native_loop::TurnGenerator;
use super::provenance::SessionSeed;
use super::router::EngineChoice;
use super::rpc::{confirm_session, start_session, CoderSessionEntry, StartArgs};
use super::session::{CoderEventKind, CoderState};
use crate::session::ServerState;

/// Largest patch handed to a reviewer.
///
/// A reviewer that receives a megabyte of diff does not read it more carefully
/// than one that receives 200KB; it costs more and is likelier to be silently
/// truncated by the provider, which is the same outcome with no record of it.
/// `stage_and_diff` marks the cut and reports the full size, so the truncation
/// is visible rather than inferred.
const MAX_REVIEW_DIFF_BYTES: usize = 200_000;

/// How long one healed item may take before the tick gives up on it.
pub const DEFAULT_ITEM_WALL_SECS: u64 = 45 * 60;

/// How long the whole review panel may take before the item is failed.
///
/// Separate from the item ceiling because a wedged panel and a wedged coder
/// session look nothing alike and the sweep holds its lock through both.
pub const PANEL_WALL_SECS: u64 = 10 * 60;

/// Asks one reviewer to judge a diff. One method, so a panel is N of these and
/// the fan-out has nothing to coordinate.
///
/// Fallible on purpose: a vendor that cannot be reached must produce an error
/// here rather than a verdict, because [`super::heal_gate::decide`] counts a
/// missing answer as missing, never as a pass.
#[async_trait::async_trait]
pub trait Reviewer: Send + Sync {
    /// The model's name, for the audit trail and for deduplication.
    fn model(&self) -> &str;
    /// Judge, returning the raw answer to be parsed.
    async fn review(&self, criteria: &str, diff: &str) -> Result<String, String>;
}

/// The registered session, or an error rather than a panic.
///
/// `bench.rs` has an `expect("session registered")` here, which is right for a
/// benchmark and wrong for an unattended loop: a missing entry must fail this
/// item, not take the daemon down.
async fn session_entry(
    state: &Arc<ServerState>,
    id: &str,
) -> Result<Arc<CoderSessionEntry>, String> {
    state
        .coder_sessions
        .lock()
        .await
        .get(id)
        .cloned()
        .ok_or_else(|| format!("coder session `{id}` is not registered"))
}

/// The live runner.
pub struct LiveCoderRunner {
    pub state: Arc<ServerState>,
    pub generator: Arc<dyn TurnGenerator>,
    pub state_dir: std::path::PathBuf,
    /// The panel. Empty means no panel, which [`super::heal_gate::decide`]
    /// refuses outright rather than treating as unanimous consent.
    pub reviewers: Vec<Arc<dyn Reviewer>>,
    pub max_wall_secs: u64,
    pub max_iterations: Option<u32>,
    /// Which engine runs the work.
    ///
    /// An operator knob, not a test hook. `Auto` picks per task from the
    /// detected external CLIs, so an unattended loop on a machine with Claude
    /// Code or Codex installed silently runs a *different* engine than the same
    /// loop on a machine without them — and the user has asked for foreman on
    /// this path specifically. Naming it makes that choice visible and
    /// configurable instead of implicit in what happens to be on `PATH`.
    pub engine: EngineChoice,
    /// Pin the coder's own model, or `None` for the operator's `coder.toml`
    /// default and then adaptive routing.
    ///
    /// Worth naming for an unattended loop for the same reason the engine is:
    /// adaptive routing picks a provider per request, so a loop can silently
    /// spend a whole 45-minute session against a credential that has expired
    /// while a working one sits beside it. An operator who has decided which
    /// model does this work should be able to say so.
    pub model: Option<String>,
    /// Canonical `ModelSchema.name` values for every review-panel seat. These
    /// become strict exclusions only on the unpinned adaptive native loop.
    pub routing_exclusions: Vec<String>,
    /// Resolve a model name or id to the registry's canonical id.
    ///
    /// Injected rather than reached for, the same way
    /// [`coder_on_panel`](super::heal_review::coder_on_panel) injects it: the
    /// self-review gate is then testable without standing up an inference
    /// registry, and reaching for one here would force
    /// `get_inference_engine`'s worker-subprocess init on every test that runs
    /// a session. `heal_service` supplies the engine-backed resolver.
    pub canonical_model: Arc<dyn Fn(&str) -> String + Send + Sync>,
    /// The GitHub client delivery runs through.
    ///
    /// Owned here rather than by `LiveTickIo` because this is the only place
    /// that calls GitHub now: one owner, so `gh` cannot be reached through two
    /// handles configured differently.
    pub github: Arc<dyn super::merge::GitHubApi>,
}

/// Where this target's code lives.
///
/// One definition, because both the session and the delivery need the answer —
/// `git` runs in the worktree but `gh` runs here, and a second definition of
/// "where is this target" is a second thing that can disagree with the first.
///
/// A `Checkout::Project` resolves to the project's repo path and nothing else:
/// the loop never opens a *project session*, because that mode commits to the
/// project's `main`. See the `project: None` in `run`.
pub fn checkout_path(target: &HealTarget) -> Result<std::path::PathBuf, String> {
    match target.checkout.as_ref() {
        Some(Checkout::Local(p)) => Ok(p.clone()),
        Some(Checkout::Project(slug)) => Ok(super::project::load_project(slug)?.repo_path.clone()),
        // Unreachable: `tick` refuses a watch-only target before it gets here.
        // Matched rather than unwrapped so a future caller that skips that
        // check fails loudly instead of panicking.
        None => Err("target has no checkout; it is watch-only".into()),
    }
}

fn configuration_run_failure(
    session_id: &str,
    detail: String,
    failure_kind: Option<&str>,
) -> Option<RunFailure> {
    (failure_kind == Some("configuration"))
        .then(|| RunFailure::configuration_with_session(session_id, detail))
}

impl LiveCoderRunner {
    /// Drive a session to a terminal state, releasing its worktree.
    ///
    /// `NeedsApproval` is not terminal, so a session left there keeps the git
    /// worktree registered in the operator's own repository and its entry in
    /// the daemon's session map. On an unattended loop firing on a cadence,
    /// rejection is the *expected* outcome — so without this every rejected
    /// item leaks a worktree, permanently.
    ///
    /// Best-effort by design: this runs on paths that are already reporting a
    /// failure, and an error here must not replace the reason the caller is
    /// about to report.
    async fn terminate(&self, session_id: &str, to: CoderState) {
        let Ok(entry) = session_entry(&self.state, session_id).await else {
            return;
        };
        let mut session = entry.session.lock().await;
        if session.state.is_terminal() {
            return;
        }
        if let Err(e) = session.transition(to, &entry.sink) {
            tracing::warn!(session = %session_id, "self-heal could not close the session: {e}");
        }
    }

    /// Poll every reviewer independently and keep both halves of the answer.
    ///
    /// Reviewers are asked concurrently but never shown each other's verdicts:
    /// a panel that reaches consensus by reading itself is one reviewer with
    /// extra steps, and correlation is precisely the failure a panel exists to
    /// catch.
    async fn poll_panel(&self, criteria: &str, diff: &str) -> (Vec<Verdict>, Vec<Unreachable>) {
        let futures: Vec<_> = self
            .reviewers
            .iter()
            .map(|r| async move {
                let model = r.model().to_string();
                match r.review(criteria, diff).await {
                    Ok(answer) => parse_verdict(&model, &answer),
                    Err(e) => Err(Unreachable { model, error: e }),
                }
            })
            .collect();

        let mut verdicts = Vec::new();
        let mut unreachable = Vec::new();
        for result in futures::future::join_all(futures).await {
            match result {
                Ok(v) => verdicts.push(v),
                Err(u) => unreachable.push(u),
            }
        }
        (verdicts, unreachable)
    }
}

#[async_trait::async_trait]
impl CoderRunner for LiveCoderRunner {
    async fn run(
        &self,
        target: &HealTarget,
        item: &Candidate,
        seed: &SessionSeed,
    ) -> Result<super::heal_tick::Attempt, RunFailure> {
        // Before a session exists — nothing to close out.
        let repo = checkout_path(target).map_err(RunFailure::early)?;

        let start = start_session(
            &self.state,
            StartArgs {
                // The healing loop runs on the machine that owns the tracker
                // and the credentials; farming its subtasks to peers is a
                // separate decision from healing at all.
                distributed: false,
                browser: false,
                workers: Vec::new(),
                repo,
                intent: seed.as_str().to_string(),
                engine: self.engine.clone(),
                max_iterations: self.max_iterations,
                state_dir: self.state_dir.clone(),
                // Always `None`, even for a `Checkout::Project` target. A
                // project session commits straight to the project's `main`
                // (`approve_merge_session` takes the `commit_to_main` fork),
                // and this loop's whole autonomy boundary is that it never
                // lands anything. A managed project is healed the same way any
                // repository is: a branch and a pull request.
                project: None,
                model: self.model.clone(),
                routing_exclusions: self.routing_exclusions.clone(),
                repair_invokes: None,
                transient_retries: None,
                discussion_id: None,
                base: None,
            },
            self.generator.clone(),
        )
        .await
        .map_err(RunFailure::early)?;

        let session_id = start["session_id"]
            .as_str()
            .ok_or_else(|| RunFailure::early("coder.start returned no session_id"))?
            .to_string();

        // The contract is derived from the INTENT, never from the issue body —
        // `may_source_contract` is runtime-only, and this item is not
        // necessarily runtime-authored. Confirming with `None` accepts the
        // derived contract rather than supplying one.
        // From here a session EXISTS, so every failure names it and `tick`
        // closes it out. A non-terminal session holds a git worktree
        // registered in the operator's own repository.
        confirm_session(&self.state, &session_id, None)
            .await
            .map_err(|e| RunFailure::with_session(&session_id, e))?;

        let entry = session_entry(&self.state, &session_id)
            .await
            .map_err(|e| RunFailure::with_session(&session_id, e))?;

        // Refuse a contract that cannot go green before spending a session on
        // it. `confirm_session` evaluates every check against the untouched
        // worktree, and a check whose program does not exist fails there the
        // same way a genuinely red one does — so the run proceeds, burns its
        // whole iteration budget, and reports "contract not satisfied". That is
        // exactly what a live trial did: derivation wrote `python -m pytest` on
        // a machine with only `python3`, and twelve iterations of real
        // inference went into something unsatisfiable from the first edit.
        //
        // An interactive `car code` user sees the baseline and can revise the
        // contract; an unattended loop has nobody to notice.
        {
            let session = entry.session.lock().await;
            let unrunnable = super::contract::baseline_cannot_run(&session.baseline);
            if !unrunnable.is_empty() {
                return Err(RunFailure::with_session(
                    &session_id,
                    format!(
                        "the derived contract cannot be evaluated for {}#{}: check(s) {} \
                         could not be run at all (the command does not exist here), so no \
                         change could ever make this contract green",
                        item.repo,
                        item.number,
                        unrunnable.join(", ")
                    ),
                ));
            }
            // The other half of the red-green baseline, and the worse case of
            // the two. An unrunnable check FAILS, so the run reports it. A
            // contract that is already green PASSES, so the session would reach
            // the gate with the deterministic half silently satisfied by code
            // nobody in this run wrote — leaving the panel as the only thing
            // actually deciding, while the pull request reports both. The
            // contract is the load-bearing gate (see `heal_gate`); the panel
            // narrows it.
            //
            // Checked here for the same reason `baseline_cannot_run` is: an
            // interactive `car code` user sees the baseline and can revise the
            // contract, and an unattended loop has nobody to notice. `car
            // code-task` already refuses on this condition (car#1070); the loop
            // was the one caller that skipped it.
            //
            // All-green only, never a single passing check — `baseline_gates_nothing`
            // documents why: a refactor legitimately has checks green on both
            // sides, and escalating on one would abort sessions over a contract
            // nit.
            if super::contract::baseline_gates_nothing(&session.baseline) {
                return Err(RunFailure::with_session(
                    &session_id,
                    format!(
                        "every check of the contract derived for {}#{} already passes on \
                         unmodified code ({} check(s)), so it gates nothing and no change \
                         could turn it red-to-green. Either the issue's premise no longer \
                         holds — the behaviour it describes may already be fixed — or the \
                         derived checks do not exercise it. A human should decide which.",
                        item.repo,
                        item.number,
                        session.baseline.len()
                    ),
                ));
            }
        }
        let handle = entry.task.lock().unwrap().take();
        if let Some(mut handle) = handle {
            // Bounded. An unattended loop must not inherit whatever limits the
            // engine happens to have: a wedged session would hold its claim for
            // the full TTL while the cadence fires around it.
            //
            // `&mut handle`, and `abort()` on expiry. Dropping a `JoinHandle`
            // DETACHES the task, it does not stop it — so timing out by value
            // left the session running, still editing the same repository,
            // after the tick had released the claim and a later tick could
            // start a second session on the same item. Taking the handle also
            // disarms `coder.cancel`, whose `abort()` is the only thing that
            // can interrupt a session wedged inside a long await, so this is
            // now the only place that can stop it.
            match tokio::time::timeout(
                std::time::Duration::from_secs(self.max_wall_secs),
                &mut handle,
            )
            .await
            {
                Ok(Ok(())) => {}
                Ok(Err(e)) => {
                    return Err(RunFailure::with_session(
                        &session_id,
                        format!("coder session panicked: {e}"),
                    ))
                }
                Err(_) => {
                    handle.abort();
                    self.terminate(&session_id, CoderState::Failed).await;
                    return Err(RunFailure::with_session(
                        &session_id,
                        format!(
                            "coder session exceeded its {}s ceiling for {}#{}",
                            self.max_wall_secs, item.repo, item.number
                        ),
                    ));
                }
            }
        }

        // One read for everything the gate below decides on. These cannot change
        // between them — the loop has finished — and three separate `lock()`s
        // over thirty lines is three chances for that to stop being true.
        let (state, contract_detail, authors, failure_kind, ran_as, nominated) = {
            let s = entry.session.lock().await;
            // A model's `report_no_change` can park at `NeedsApproval` on a
            // contract that never went green. Unlike the runtime-observed
            // finding (a green run that left the tree untouched), that is not
            // "the contract passed", and reporting it as such would post a
            // false comment on the tracker. The last check results decide.
            let nominated = s.no_change_finding.as_ref().and_then(|finding| {
                let green = !s.last_check_results.is_empty()
                    && s.last_check_results.iter().all(|r| r.passed);
                (!green).then(|| {
                    format!(
                        "the coder concluded no change is needed ({}: {}) without the \
                         outcome contract passing; a no-change conclusion needs a human",
                        finding.kind.as_str(),
                        finding.summary
                    )
                })
            });
            (
                s.state,
                s.error
                    .clone()
                    .unwrap_or_else(|| "outcome contract evaluated".to_string()),
                s.authored_by.clone(),
                s.failure_kind.clone(),
                // The RESOLVED engine, not the one heal.toml asked for. See the
                // empty-authors check below for why the difference matters.
                s.engine.clone(),
                nominated,
            )
        };
        let red_nomination = nominated.is_some();
        let contract_detail = nominated.unwrap_or(contract_detail);

        // `NeedsApproval` is the runtime's own verdict that the contract
        // passed — except for a nominated no-change finding on a contract that
        // never went green (above). Anything else is a red contract, and a red
        // contract is not put to the panel at all.
        let contract_passed = state == CoderState::NeedsApproval && !red_nomination;
        if !contract_passed {
            if let Some(failure) = configuration_run_failure(
                &session_id,
                contract_detail.clone(),
                failure_kind.as_deref(),
            ) {
                return Err(failure);
            }
            return Ok(super::heal_tick::Attempt {
                contract_passed: false,
                contract_detail,
                panel_size: self.reviewers.len(),
                verdicts: Vec::new(),
                unreachable: Vec::new(),
                session_id: session_id.clone(),
            });
        }

        // A model may not review its own output, checked HERE against what
        // actually authored the change rather than only at assembly against
        // what was configured.
        //
        // `heal_service`'s `coder_on_panel` runs when the coder is PINNED —
        // by `heal.toml`'s `coder_model` or, failing that, `coder.toml`'s
        // `model` (car#1360). Unpinned — the default — the router picks, and
        // on a machine with one reachable credential that pick can also be a
        // panel seat, so the self-review the gate refuses when configured was
        // permitted by default (car#1299). The journal has recorded the
        // authoring model all along; nothing read it.
        //
        // The SAME rule, not a second copy of it: `coder_on_panel` canonicalizes
        // both sides through the registry, and that is load-bearing here rather
        // than cosmetic. An unpinned turn reports `schema.name` while a seat is
        // configured however the operator spelled it, and `knows_model` accepts
        // an id or a name — so for every model whose id and name differ (car#889:
        // `openrouter/google/gemini-3.1-pro-preview` vs
        // `google/gemini-3.1-pro-preview`) a spelling comparison silently never
        // fires. A gate that cannot fire is worse than no gate, because it reads
        // as covered.
        //
        // `unwrap_or_else(|| m.to_string())`, NOT `unwrap_or_default()`. The
        // assembly-time check can use the latter because every name there
        // cleared `knows_model` first; here the author comes from the engine and
        // nothing validated it, so an unknown name would canonicalize to `""` and
        // match every other unknown name — refusing every session on a daemon
        // whose registry does not hold the model that ran.
        //
        // Refused rather than dropping the seat: a panel silently shrunk from
        // three to two is the halved threshold `PanelIncomplete` exists to
        // prevent, and the diff itself is not wrong — only the independence the
        // verdict would claim for it.
        {
            // No native turn was journaled. For foreman/external — the DEFAULT
            // engine — that is the honest answer rather than a failure: those
            // farm to a coding CLI whose backbone CAR never resolved, so there
            // is no model to compare. A session that RAN native and reaches
            // `NeedsApproval` is a different matter: it got here by a model
            // declaring done, which journals a terminal, so an empty set means
            // the attribution was lost. Permitting on lost instrumentation is
            // exactly the silently-absent attribution this closes, one level
            // down.
            //
            // `ran_as`, the engine the session RESOLVED to — not `self.engine`,
            // which is what `heal.toml` asked for. `engine = "auto"` parses,
            // and `EngineChoice::Auto` resolves to `Native` whenever no external
            // CLI is ready (`router::resolve`), which on a headless server is
            // the common case rather than an exotic one. Comparing the request
            // let exactly the session this refuses — native, terminal, no
            // attribution — through the gate, because the request said `auto`
            // (car#1357).
            if authors.is_empty() && ran_as == EngineChoice::Native {
                return Err(RunFailure::with_session(
                    &session_id,
                    format!(
                        "the native coder produced a change for {}#{} with no model \
                         recorded against it — the journal should name every model that \
                         completed a turn, so this session cannot be shown to be \
                         independent of the review panel",
                        item.repo, item.number
                    ),
                ));
            }

            let seats: Vec<String> = self
                .reviewers
                .iter()
                .map(|r| r.model().to_string())
                .collect();
            for author in &authors {
                let Some(seat) = super::heal_review::coder_on_panel(author, &seats, |m| {
                    (self.canonical_model)(m)
                }) else {
                    continue;
                };
                return Err(RunFailure::with_session(
                    &session_id,
                    format!(
                        "{} wrote part of this change for {}#{} and also sits on the review \
                         panel ({}) — a model cannot review its own output, and counting it \
                         would report an independence this gate does not have. The router \
                         chose it — nothing pinned a coder, or the pin is not what \
                         served this turn — so pin a coder that is not a seat (`heal.toml`'s \
                         `coder_model`, or `coder.toml`'s `model`), or drop that seat from \
                         `review_models`.",
                        author, item.repo, item.number, seat
                    ),
                ));
            }
        }

        // The worktree diff, which is exactly what the human approval surface
        // reviews — not a diff of a published branch, because nothing is
        // published until the gate approves.
        let worktree = {
            let s = entry.session.lock().await;
            s.workspace_path.clone().ok_or_else(|| {
                RunFailure::with_session(&session_id, "session reached approval with no worktree")
            })?
        };
        let staged =
            super::merge::stage_and_diff(&worktree, MAX_REVIEW_DIFF_BYTES).map_err(|e| {
                RunFailure::with_session(
                    &session_id,
                    format!("could not read the change for review: {e}"),
                )
            })?;

        // A hard failure, never a review of nothing. The panel is the only
        // human substitute in this loop, and a reviewer handed an empty or
        // unreadable diff still answers — so a soft fallback here records three
        // verdicts that were never rendered on the change, and opens a pull
        // request on them.
        // A truncated patch is judged the same way an unavailable one would be:
        // `tail()` keeps the LAST 200KB, so the panel would receive a fragment
        // whose beginning is missing and be told nothing about it — and it
        // would answer anyway. `stage_and_diff` reports the cut precisely so a
        // caller need not infer it; reading `truncated` and refusing is what
        // makes that report mean something.
        if staged.truncated {
            return Err(RunFailure::with_session(
                &session_id,
                format!(
                    "the change for {}#{} is {} bytes, larger than the {} a reviewer is \
                 given — a panel judging the tail of a patch is not a review",
                    item.repo, item.number, staged.full_bytes, MAX_REVIEW_DIFF_BYTES
                ),
            ));
        }

        if staged.changed_paths.is_empty() {
            return Err(RunFailure::with_session(
                &session_id,
                format!(
                "the outcome contract passed but the worktree is unchanged for {}#{} — there is nothing to review and nothing to deliver",
                    item.repo, item.number
                ),
            ));
        }

        let criteria = super::heal_gate::review_criteria(seed.as_str(), &contract_detail);
        // Bounded, like the session above it. The remote client has its own
        // HTTP timeout, but that bound belongs to the client and does not cover
        // a local seat or a retry chain — and the sweep holds its lock for the
        // whole tick, so a wedged panel makes every later cadence tick log
        // "previous sweep still running" with no indication of what is stuck.
        // A timeout produces `Unreachable`, which the gate already reads as a
        // missing answer rather than a pass.
        let (verdicts, unreachable) = match tokio::time::timeout(
            std::time::Duration::from_secs(PANEL_WALL_SECS),
            self.poll_panel(&criteria, &staged.patch),
        )
        .await
        {
            Ok(v) => v,
            Err(_) => (
                Vec::new(),
                self.reviewers
                    .iter()
                    .map(|r| Unreachable {
                        model: r.model().to_string(),
                        error: format!("the panel did not answer within {PANEL_WALL_SECS}s"),
                    })
                    .collect(),
            ),
        };

        Ok(super::heal_tick::Attempt {
            contract_passed: true,
            contract_detail,
            panel_size: self.reviewers.len(),
            verdicts,
            unreachable,
            session_id,
        })
    }

    async fn deliver(
        &self,
        target: &HealTarget,
        item: &Candidate,
        session_id: &str,
        body: &str,
    ) -> Result<PrDeliveryOutcome, DeliverRefusal> {
        let repo = checkout_path(target).map_err(DeliverRefusal::permanent)?;
        // Named up front, because `gh`'s own failure for a missing credential
        // arrives from three different seams inside delivery and none of them
        // says which credential.
        self.github.auth_status().map_err(|e| {
            DeliverRefusal::permanent(format!("gh is not logged in: {}", e.message))
        })?;
        let entry = session_entry(&self.state, session_id)
            .await
            .map_err(DeliverRefusal::permanent)?;

        // Read what delivery needs, then DROP the guard.
        //
        // Holding the session mutex across `deliver_pr_with` held it across
        // five blocking `git`/`gh` subprocesses — each with a 900-second
        // ceiling — including a network round trip to GitHub. `coder.list`
        // locks every session in a loop to build the board, so one unattended
        // heal delivery wedged the board for every user of the daemon, and
        // `coder.cancel` parked on the same lock. The interactive
        // `approve_merge_session` sets the precedent for the lock discipline;
        // it does not excuse adding a network call inside it, on a path nobody
        // is watching.
        let (worktree, contract, intent) = {
            let session = entry.session.lock().await;
            (
                session.workspace_path.clone().ok_or_else(|| {
                    DeliverRefusal::permanent("session has no worktree to deliver")
                })?,
                session
                    .contract
                    .clone()
                    .ok_or_else(|| DeliverRefusal::permanent("session has no outcome contract"))?,
                session.intent.clone(),
            )
        };

        let branch = delivery_branch(item);
        // Off the async worker. `deliver_pr_with` is synchronous and spawns
        // subprocesses; awaiting it inline blocks a runtime thread the
        // scheduler believes is available.
        let github = self.github.clone();
        let base = target.base.clone();
        let body = body.to_string();
        let outcome = tokio::task::spawn_blocking(move || {
            super::merge::deliver_pr_with(
                super::merge::PrDelivery {
                    repo: &repo,
                    worktree: &worktree,
                    target_branch: &branch,
                    base_branch: &base,
                    draft: false,
                    intent: &intent,
                    contract: &contract,
                    body: &body,
                    // The heal loop hardcodes `distributed: false` — healing
                    // runs on the machine that owns the tracker and the
                    // credentials — so its subtasks always ran here.
                    provenance: None,
                },
                github.as_ref(),
            )
        })
        .await
        .map_err(|e| DeliverRefusal::retriable(format!("delivery task failed: {e}")))?
        .map_err(refusal_for)?;

        let mut session = entry.session.lock().await;
        session.result_branch = Some(outcome.branch.clone());
        entry.sink.emit(CoderEventKind::MergeCompleted {
            branch: outcome.branch.clone(),
        });
        // Terminal. On the raw-repo path `Merged` already means only "the
        // branch was published" — `approve_merge_session` performs no merge for
        // a non-project session — so this records the fact that occurred, and
        // it is what releases the worktree.
        session
            .transition(CoderState::Merged, &entry.sink)
            .map_err(DeliverRefusal::permanent)?;
        Ok(outcome)
    }

    async fn abandon(&self, session_id: &str) {
        self.terminate(session_id, CoderState::Abandoned).await;
    }
}

/// The delivery branch for an item — **stable across attempts**.
///
/// A per-session `car/coder/<uuid>` would mean a retried item pushes a second
/// branch and opens a second pull request, and a delivery that failed after the
/// push would strand the first on the remote forever. A stable name is also
/// what lets `deliver_pr_with` reconcile to exactly one pull request per item
/// and honour a human closing it.
pub fn delivery_branch(item: &Candidate) -> String {
    format!("car/heal/{}-{}", item.repo.replace('/', "-"), item.number)
}

/// Map a delivery failure onto whether trying again could ever help.
///
/// `DeliveryFailure` already carries this judgement per stage; the only work
/// here is not throwing it away. A preflight refusal is never retriable — it
/// includes the closed-pull-request rule, which is a person saying stop.
fn refusal_for(failure: super::merge::DeliveryFailure) -> DeliverRefusal {
    use super::merge::DeliveryFailure as F;
    match failure {
        F::Preflight { reason } => DeliverRefusal::permanent(reason),
        F::Commit { reason } => DeliverRefusal::permanent(reason),
        F::Push { reason, retriable } | F::Pr { reason, retriable } => {
            if retriable {
                DeliverRefusal::retriable(reason)
            } else {
                DeliverRefusal::permanent(reason)
            }
        }
    }
}

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

    struct Fixed {
        model: &'static str,
        answer: Result<String, String>,
    }

    #[async_trait::async_trait]
    impl Reviewer for Fixed {
        fn model(&self) -> &str {
            self.model
        }
        async fn review(&self, _c: &str, _d: &str) -> Result<String, String> {
            self.answer.clone()
        }
    }

    #[test]
    fn configuration_session_failure_is_typed_for_the_tick() {
        let failure = configuration_run_failure(
            "coder-1",
            "no independent coder".into(),
            Some("configuration"),
        )
        .expect("configuration failure");
        assert!(failure.configuration);
        assert_eq!(failure.session_id.as_deref(), Some("coder-1"));
        assert_eq!(failure.detail, "no independent coder");
        assert!(configuration_run_failure("coder-1", "red".into(), Some("error")).is_none());
    }

    /// Poll a panel without standing up a runtime: the fan-out is pure, and
    /// the whole point of the split is that it can be tested that way.
    async fn poll(
        reviewers: Vec<Arc<dyn Reviewer>>,
        criteria: &str,
        diff: &str,
    ) -> (Vec<Verdict>, Vec<Unreachable>) {
        let futures: Vec<_> = reviewers
            .iter()
            .map(|r| async move {
                let model = r.model().to_string();
                match r.review(criteria, diff).await {
                    Ok(answer) => parse_verdict(&model, &answer),
                    Err(e) => Err(Unreachable { model, error: e }),
                }
            })
            .collect();
        let mut verdicts = Vec::new();
        let mut unreachable = Vec::new();
        for result in futures::future::join_all(futures).await {
            match result {
                Ok(v) => verdicts.push(v),
                Err(u) => unreachable.push(u),
            }
        }
        (verdicts, unreachable)
    }

    #[tokio::test]
    async fn a_panel_keeps_answers_and_failures_apart() {
        let reviewers: Vec<Arc<dyn Reviewer>> = vec![
            Arc::new(Fixed {
                model: "a",
                answer: Ok("PASS looks right".into()),
            }),
            Arc::new(Fixed {
                model: "b",
                answer: Ok("FAIL wrong scope".into()),
            }),
            Arc::new(Fixed {
                model: "c",
                answer: Err("429 rate limited".into()),
            }),
        ];
        let panel_size = reviewers.len();
        let (verdicts, unreachable) = poll(reviewers, "criteria", "diff").await;
        assert_eq!(verdicts.len(), 2);
        assert_eq!(unreachable.len(), 1);
        assert_eq!(unreachable[0].model, "c");
        // The count that matters: two answers on a panel of three cannot be
        // read as "two of two".
        assert_eq!(panel_size, 3);
    }

    #[tokio::test]
    async fn an_unreadable_answer_becomes_unreachable_not_a_pass() {
        let (verdicts, unreachable) = poll(
            vec![Arc::new(Fixed {
                model: "a",
                answer: Ok("hmm, hard to say".into()),
            })],
            "c",
            "d",
        )
        .await;
        assert!(verdicts.is_empty());
        assert_eq!(unreachable.len(), 1);
    }

    #[test]
    fn a_watch_only_target_is_refused_loudly_not_unwrapped() {
        let t = HealTarget {
            repo: "acme/w".into(),
            fix_repo: None,
            checkout: None,
            label: "self-heal".into(),
            base: "main".into(),
        };
        assert!(checkout_path(&t).is_err());
    }
}