car-server-core 0.55.0

//! One pass of the self-healing loop.
//!
//! T5 of `docs/proposals/self-healing-issue-loop.md`. Composes the pieces:
//! read → select → claim → hand to the coder → gate → open a pull request, or
//! give the item back.
//!
//! ## At most one item, then exit
//!
//! A tick is not a loop. It takes one item and returns, and the *scheduler*
//! decides when to run again. That is what makes the thing interruptible: a
//! daemon restart loses at most one in-flight item, whose claim then expires,
//! rather than a long-running process holding an unbounded amount of work.
//!
//! ## Idle is the steady state
//!
//! The queue is human-authored. A tick that finds nothing eligible has not
//! failed — humans control what gets labelled, and most of the time the honest
//! answer is that there is nothing to do. [`TickOutcome::Idle`] carries the
//! skip reasons so an operator can tell "nothing eligible" from "everything was
//! skipped for a reason I did not expect", which is the difference between a
//! quiet loop and a broken one.
//!
//! ## The loop never decides an issue is fixed
//!
//! It opens a pull request referencing the item, and stops. The item stays open
//! until a human closes it through the normal merge path. So there is no "am I
//! done" judgement for a model to get wrong — termination is external, by
//! construction.

use std::sync::Arc;

use super::heal_claims::ClaimStore;
use super::heal_gate::{decide, GateOutcome, Unreachable, Verdict};
use super::heal_intake::{Checkout, HealTarget};
use super::heal_select::{select, Candidate, Selection, SkippedItem};
use super::merge::{CiSummary, PrDeliveryOutcome};
use super::provenance::{ProvenanceTier, SessionSeed};

/// What resolving an item's provenance produced.
///
/// Three outcomes, not two, because `Option` conflated states whose correct
/// handling differs. An item that vanished between the scan and now needs no
/// backoff — it is gone from the queue and a later tick will not see it. An
/// item whose author the tier gate REFUSED is still in the queue, still the
/// oldest, and still labelled, so without a recorded failure selection picks it
/// again on every tick and halts the sweep there forever — starving everything
/// behind it. That path became reachable the moment selection stopped
/// rejecting untiered candidates outright.
#[derive(Debug)]
pub enum Intent {
    /// Cleared. The text may seed a coder session.
    Seed(SessionSeed),
    /// Closed or edited between the scan and now. The normal case for a
    /// human-authored queue, and not a failure of anything.
    Gone,
    /// The provenance gate said no.
    ///
    /// `stale` separates the two refusals, because their remedies differ and
    /// reporting one as the other sends an operator to the wrong place: a tier
    /// resolved too long ago is fixed by retrying, an untrusted author is not
    /// fixed by anything the operator does to this daemon.
    Refused { tier: ProvenanceTier, stale: bool },
}

/// A coder run that failed, and the session it left behind.
///
/// The session id is the load-bearing half. `run` can fail *after* the session
/// has reached `NeedsApproval` — no worktree at approval, an unreadable diff, a
/// green contract over an unchanged worktree — and a plain `String` gave the
/// caller no way to close that session out. Each one leaked a git worktree
/// registered in the operator's own repository, with its task handle already
/// taken so `coder.cancel` could not reach it either, and the reaper that would
/// have swept them was deleted on the strength of the invariant these paths
/// break.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunFailure {
    pub detail: String,
    /// `None` only when the failure happened before a session existed.
    pub session_id: Option<String>,
    /// The daemon configuration prevented any model from attempting the item.
    /// Such a failure must not create item backoff or a public issue comment.
    pub configuration: bool,
}

impl RunFailure {
    /// A failure with no session behind it — nothing to close out.
    pub fn early(detail: impl Into<String>) -> Self {
        Self {
            detail: detail.into(),
            session_id: None,
            configuration: false,
        }
    }

    /// A failure that left a live session.
    pub fn with_session(session_id: &str, detail: impl Into<String>) -> Self {
        Self {
            detail: detail.into(),
            session_id: Some(session_id.to_string()),
            configuration: false,
        }
    }

    /// A session stopped before model dispatch because its routing constraints
    /// leave no independent coder available.
    pub fn configuration_with_session(session_id: &str, detail: impl Into<String>) -> Self {
        Self {
            detail: detail.into(),
            session_id: Some(session_id.to_string()),
            configuration: true,
        }
    }
}

/// Why a delivery stopped, and whether trying again could ever change it.
///
/// `permanent` is not a hint. A closed pull request on the item's delivery
/// branch is a person saying stop, and `merge::deliver_pr_with` refuses it at
/// preflight every time — so an ordinary failure record would spend four more
/// coder sessions and four more panel fan-outs rediscovering the same answer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeliverRefusal {
    pub detail: String,
    pub permanent: bool,
}

impl DeliverRefusal {
    pub fn retriable(detail: impl Into<String>) -> Self {
        Self {
            detail: detail.into(),
            permanent: false,
        }
    }
    pub fn permanent(detail: impl Into<String>) -> Self {
        Self {
            detail: detail.into(),
            permanent: true,
        }
    }
}

/// What a coder run produced: the facts a gate decides over.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Attempt {
    /// Whether the outcome contract passed. The deterministic half.
    pub contract_passed: bool,
    /// What the contract reported, for the operator and the reviewers.
    pub contract_detail: String,
    /// How many reviewers were ASKED. The threshold derives from this, not
    /// from how many answered.
    pub panel_size: usize,
    pub verdicts: Vec<Verdict>,
    pub unreachable: Vec<Unreachable>,
    /// The coder session that produced the work, still holding its worktree.
    ///
    /// A session id rather than a branch, because **nothing is published until
    /// the gate approves**. In the interactive path `car/coder/<id>` is created
    /// by `coder.approve_merge` — the branch exists only after a human said
    /// yes, so it means "approved". Publishing before the panel would put
    /// unapproved work in a namespace that already carries that meaning, and
    /// leave one dead branch per rejection in the operator's repository. The
    /// panel reviews the worktree diff instead, which is the same view the
    /// human approval surface shows.
    pub session_id: String,
}

/// What one tick did.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "outcome", rename_all = "snake_case")]
pub enum TickOutcome {
    /// Nothing was eligible. The reasons are carried so a quiet loop can be
    /// told apart from a stuck one.
    Idle { skipped: Vec<SkippedItem> },
    /// A pull request was opened for the item.
    Opened {
        repo: String,
        number: u64,
        pr_url: String,
        gate: String,
        /// Typed check-run/combined-status state for the delivered head SHA.
        ci: CiSummary,
        /// Human-readable form of `ci`, suitable for a delivery log.
        delivery: String,
    },
    /// The item was attempted and did not clear the gate. The claim is
    /// released; the item returns to the queue for a human or a later tick.
    Rejected {
        repo: String,
        number: u64,
        gate: String,
    },
    /// The tick could not run at all — the target is unusable, or a read
    /// failed. Distinct from `Idle` because the remedy is different.
    Failed { detail: String },
}

/// Everything a tick needs from the outside world, injected.
///
/// A trait rather than concrete calls so the composition — the order of claim,
/// work, gate, release — is testable without GitHub, inference, or a worktree.
/// That order is the part most likely to be wrong, and the part a live test
/// would exercise least reliably.
#[async_trait::async_trait]
pub trait TickIo: Send + Sync {
    /// Candidates on this target, already tiered and label-checked.
    async fn candidates(&self, target: &HealTarget) -> Result<Vec<Candidate>, String>;
    /// Open pull requests on this target, for the coverage check.
    async fn open_prs(
        &self,
        target: &HealTarget,
    ) -> Result<Vec<super::heal_intake::RawPullRequest>, String>;
    /// The trust-cleared seed for this item, or `None` if it cannot be cleared.
    ///
    /// [`SessionSeed`] and not `String`: the newtype exists precisely so cleared
    /// text cannot be moved around as an ordinary string, and returning a
    /// `String` here handed that guarantee back. `None` means the item must be
    /// skipped — never "use the raw body instead".
    async fn intent_for(&self, item: &Candidate) -> Result<Intent, String>;
    /// Run a coder session and return the EVIDENCE, never the verdict.
    ///
    /// An earlier shape had this return a finished `GateOutcome`, which meant
    /// the injected implementation could hand back `Approved` having run no
    /// contract and asked no reviewer — and `decide` had no callers at all. A
    /// gate a caller can mint is not a gate. `GateOutcome` is now sealed so
    /// only [`super::heal_gate::decide`] can build one, and this returns the
    /// inputs to it.
    async fn run_coder(
        &self,
        target: &HealTarget,
        item: &Candidate,
        intent: &SessionSeed,
    ) -> Result<Attempt, RunFailure>;
    /// Redact text bound for a public tracker.
    ///
    /// Gate summaries carry contract output and vendor error strings, which is
    /// where hostnames and credentials surface. The target may be a public
    /// repository, and a comment cannot be unposted.
    fn redact(&self, text: &str) -> String;
    /// Publish the work and open the pull request, as one step.
    ///
    /// Called only after [`decide`] approves — this is the loop's single write
    /// to the world, and everything before it is reversible by doing nothing.
    /// The outcome includes check-run/combined-status state read for the exact
    /// delivered head SHA.
    async fn deliver(
        &self,
        target: &HealTarget,
        item: &Candidate,
        session_id: &str,
        gate: &GateOutcome,
    ) -> Result<PrDeliveryOutcome, DeliverRefusal>;

    /// Close out a session whose work will not be delivered.
    ///
    /// Required rather than defaulted: a coder session holds a git worktree
    /// registered in the operator's own repository until it reaches a terminal
    /// state, and rejection is the *expected* common outcome of a strict
    /// majority. An implementation that forgets this leaks a worktree per
    /// rejected item, forever, and a default no-op would let it.
    async fn abandon(&self, session_id: &str);
    /// Say on the item why the loop stopped, so a human is not left guessing.
    async fn comment(&self, item: &Candidate, text: &str) -> Result<(), String>;
    fn now_ms(&self) -> u64;
}

/// Run one tick against one target.
/// Where the claim ledger is written, and when.
///
/// Injected rather than done by the caller after `tick` returns, because
/// *after* is too late: a claim that reaches disk only once the whole tick has
/// finished is lost if the daemon dies during the 45-minute coder session it
/// was taken for, and the next start re-picks an item whose work is already in
/// flight — possibly still in flight, in a detached session. `heal_service`'s
/// own module docs state this contract; nothing implemented it.
pub trait ClaimSink: Send + Sync {
    fn persist(&self, claims: &ClaimStore, now_ms: u64);
}

/// A sink that writes nothing — for tests, and for a caller with no state dir.
pub struct NoClaimSink;
impl ClaimSink for NoClaimSink {
    fn persist(&self, _claims: &ClaimStore, _now_ms: u64) {}
}

pub async fn tick(
    io: &Arc<dyn TickIo>,
    target: &HealTarget,
    claims: &mut ClaimStore,
    run_id: &str,
    sink: &dyn ClaimSink,
) -> TickOutcome {
    // A target that cannot be written to is watch-only. Establish that before
    // spending any API call: reading a queue we can do nothing about is pure
    // cost.
    if !target.can_write() {
        // Reported, not silent: `heal_config` promises a watch-only target is a
        // legal configuration, so an operator must be able to see that is why
        // nothing happened.
        return TickOutcome::Idle {
            skipped: vec![SkippedItem {
                repo: target.repo.clone(),
                number: 0,
                reason: super::heal_select::Skip::WatchOnly,
            }],
        };
    }
    if !super::heal_intake::is_valid_repo_spec(&target.repo) {
        return TickOutcome::Failed {
            detail: format!("`{}` is not a valid owner/name spec", target.repo),
        };
    }

    let now = io.now_ms();
    let candidates = match io.candidates(target).await {
        Ok(c) => c,
        Err(e) => return TickOutcome::Failed { detail: e },
    };
    let prs = match io.open_prs(target).await {
        Ok(p) => p,
        Err(e) => return TickOutcome::Failed { detail: e },
    };

    let Selection { chosen, skipped } = select(
        target,
        &candidates,
        &prs,
        claims.as_map(),
        claims.attempts(),
        now,
    );
    let Some(item) = chosen else {
        return TickOutcome::Idle { skipped };
    };

    // Claim BEFORE any work. The reverse order — work, then claim — leaves a
    // window where a second tick starts the same item, which is the whole thing
    // claiming exists to prevent.
    // Persisted immediately below, before the first thing that can take
    // minutes. See `ClaimSink`.
    if let Err(refused) = claims.claim(&item.repo, item.number, run_id, now) {
        return TickOutcome::Idle {
            skipped: vec![SkippedItem {
                repo: item.repo.clone(),
                number: item.number,
                reason: super::heal_select::Skip::Claimed {
                    run_id: refused.held_by,
                },
            }],
        };
    }
    sink.persist(claims, now);

    // From here every exit must release the claim, or a failed tick parks the
    // item until its TTL expires.
    let intent = match io.intent_for(&item).await {
        Ok(Intent::Seed(i)) => i,
        Ok(Intent::Gone) => {
            // The queue moved. No failure recorded: there is nothing to back
            // off from, and a later tick will not see this item at all.
            claims.release(&item.repo, item.number, run_id);
            return TickOutcome::Idle {
                skipped: vec![SkippedItem {
                    repo: item.repo.clone(),
                    number: item.number,
                    reason: super::heal_select::Skip::Gone,
                }],
            };
        }
        Ok(Intent::Refused { tier, stale }) => {
            // Recorded, or the sweep halts here forever. This item is still in
            // the queue, still the oldest, and still labelled, so selection
            // picks it again on the next tick and every tick after — starving
            // everything behind it while the operator sees only "not cleared".
            // The backoff is what moves the queue past it.
            // A stale tier is fixed by retrying; an untrusted author is not
            // fixed by anything done to this daemon, so it exhausts at once
            // rather than costing 1+1+2+4+8 hours of permission calls to reach
            // the same answer five times.
            if stale {
                claims.record_failure(
                    &item.repo,
                    item.number,
                    "provenance was resolved too long ago to rely on",
                    now,
                );
            } else {
                claims.record_permanent_failure(
                    &item.repo,
                    item.number,
                    &format!("author is not cleared to seed a session (tier: {tier:?})"),
                    now,
                );
            }
            claims.release(&item.repo, item.number, run_id);
            return TickOutcome::Idle {
                skipped: vec![SkippedItem {
                    repo: item.repo.clone(),
                    number: item.number,
                    // A stale tier is NOT an authorisation failure. Reporting
                    // it as one sends an operator to check permissions when
                    // the remedy is to retry.
                    reason: if stale {
                        super::heal_select::Skip::IntentNotCleared
                    } else {
                        super::heal_select::Skip::UntrustedAuthor { tier }
                    },
                }],
            };
        }
        Err(e) => {
            claims.release(&item.repo, item.number, run_id);
            return TickOutcome::Failed { detail: e };
        }
    };

    let attempt = match io.run_coder(target, &item, &intent).await {
        Ok(v) => v,
        Err(failure) => {
            // Close the session out if one was left. `run` can fail after the
            // session reached `NeedsApproval`, and a non-terminal session holds
            // a git worktree in the operator's repository until something
            // transitions it.
            if let Some(id) = &failure.session_id {
                io.abandon(id).await;
            }
            let detail = failure.detail;
            if failure.configuration {
                // Nothing about this item failed: the daemon excluded every
                // review seat and had no independent coder left. Recording
                // item backoff or posting on the issue would blame the backlog
                // for a heal.toml condition only the operator can change.
                claims.release(&item.repo, item.number, run_id);
                return TickOutcome::Failed { detail };
            }
            claims.record_failure(&item.repo, item.number, &detail, now);
            claims.release(&item.repo, item.number, run_id);
            let _ = io
                .comment(
                    &item,
                    &io.redact(&format!("self-heal could not run: {detail}")),
                )
                .await;
            return TickOutcome::Failed { detail };
        }
    };

    // The verdict is computed HERE, from evidence, by the one function that
    // can build a `GateOutcome`.
    let gate = decide(
        attempt.contract_passed,
        &attempt.contract_detail,
        attempt.panel_size,
        &attempt.verdicts,
        &attempt.unreachable,
    );

    if !gate.approved() {
        let summary = gate.summary();
        // Terminal, so the RAII handle drops and the git worktree registered in
        // the operator's repository goes with it. Before the comment, because
        // the comment is best-effort and the worktree is not.
        io.abandon(&attempt.session_id).await;
        // Record the failure BEFORE releasing, or the next tick picks the same
        // item immediately and the loop becomes a metronome.
        claims.record_failure(&item.repo, item.number, &summary, now);
        claims.release(&item.repo, item.number, run_id);
        // Say why on the item itself, redacted: an unattended loop that fails
        // silently teaches people to ignore it, and one that leaks a vendor
        // error string into a public tracker cannot be unposted.
        let _ = io
            .comment(&item, &io.redact(&format!("self-heal stopped: {summary}")))
            .await;
        return TickOutcome::Rejected {
            repo: item.repo.clone(),
            number: item.number,
            gate: summary,
        };
    }

    match io.deliver(target, &item, &attempt.session_id, &gate).await {
        Ok(delivered) => {
            // The work landed. Forget prior failures so a later, unrelated
            // failure starts from a clean backoff rather than an old count.
            claims.clear_failures(&item.repo, item.number);
            let delivery = delivered.delivery_report();
            TickOutcome::Opened {
                repo: item.repo.clone(),
                number: item.number,
                pr_url: delivered.pr_url,
                gate: gate.summary(),
                ci: delivered.ci,
                delivery,
            }
        }
        Err(refusal) => {
            io.abandon(&attempt.session_id).await;
            let detail = format!(
                "gate passed but the pull request could not be opened: {}",
                refusal.detail
            );
            // A refusal nothing can change exhausts the item now rather than
            // four sessions from now. See `DeliverRefusal`.
            if refusal.permanent {
                claims.record_permanent_failure(&item.repo, item.number, &detail, now);
            } else {
                claims.record_failure(&item.repo, item.number, &detail, now);
            }
            claims.release(&item.repo, item.number, run_id);
            // The case a human most needs told, so it comments like every other
            // stop rather than failing silently.
            let _ = io.comment(&item, &io.redact(&detail)).await;
            TickOutcome::Failed { detail }
        }
    }
}

/// The checkout argument a coder session takes for this target.
pub fn coder_target(target: &HealTarget) -> Option<(Option<std::path::PathBuf>, Option<String>)> {
    match target.checkout.as_ref()? {
        Checkout::Local(p) => Some((Some(p.clone()), None)),
        Checkout::Project(slug) => Some((None, Some(slug.clone()))),
    }
}

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

    use crate::coder::heal_intake::{ChecksState, RawPullRequest};
    use crate::coder::heal_select::{CandidateKind, CLAIM_TTL_MS};
    use crate::coder::provenance::ProvenanceTier;
    use std::sync::Mutex;

    #[derive(Default)]
    struct Fake {
        candidates: Vec<Candidate>,
        prs: Vec<RawPullRequest>,
        intent: Option<String>,
        intent_err: Option<String>,
        attempt: Option<Attempt>,
        coder_err: Option<String>,
        coder_configuration_error: bool,
        pr_err: Option<String>,
        ci_unavailable: bool,
        calls: Mutex<Vec<String>>,
        /// The fake's clock. Larger than `CLAIM_TTL_MS` so a test can express
        /// "claimed before the window" without underflowing.
        now: u64,
        /// Which refusal `intent: None` stands for: a tier resolved too long
        /// ago (retry works) rather than an untrusted author (it does not).
        intent_stale: bool,
    }

    #[async_trait::async_trait]
    impl TickIo for Fake {
        async fn candidates(&self, _t: &HealTarget) -> Result<Vec<Candidate>, String> {
            self.calls.lock().unwrap().push("candidates".into());
            Ok(self.candidates.clone())
        }
        async fn open_prs(&self, _t: &HealTarget) -> Result<Vec<RawPullRequest>, String> {
            Ok(self.prs.clone())
        }
        async fn intent_for(&self, _i: &Candidate) -> Result<Intent, String> {
            self.calls.lock().unwrap().push("intent".into());
            if let Some(e) = &self.intent_err {
                return Err(e.clone());
            }
            Ok(match self.intent.clone() {
                Some(text) => Intent::Seed(SessionSeed::from_trusted(text)),
                // `None` is a gate refusal — `Gone` is the other, distinct
                // case, and `intent_stale` picks which refusal.
                None => Intent::Refused {
                    tier: if self.intent_stale {
                        ProvenanceTier::Maintainer
                    } else {
                        ProvenanceTier::Public
                    },
                    stale: self.intent_stale,
                },
            })
        }

        fn redact(&self, text: &str) -> String {
            text.replace("sk-secret", "[redacted]")
        }
        async fn run_coder(
            &self,
            _t: &HealTarget,
            _i: &Candidate,
            _intent: &SessionSeed,
        ) -> Result<Attempt, RunFailure> {
            self.calls.lock().unwrap().push("coder".into());
            if let Some(e) = &self.coder_err {
                // The fake's coder failure names a session, so the tick's
                // abandon path is exercised rather than skipped.
                return Err(if self.coder_configuration_error {
                    RunFailure::configuration_with_session("coder-fake", e.clone())
                } else {
                    RunFailure::with_session("coder-fake", e.clone())
                });
            }
            Ok(self.attempt.clone().unwrap_or(Attempt {
                contract_passed: true,
                contract_detail: "green".into(),
                panel_size: 3,
                verdicts: vec![
                    Verdict {
                        model: "a".into(),
                        pass: true,
                        reason: "ok".into(),
                    },
                    Verdict {
                        model: "b".into(),
                        pass: true,
                        reason: "ok".into(),
                    },
                ],
                unreachable: vec![],
                session_id: "coder-e2e-1".into(),
            }))
        }
        async fn deliver(
            &self,
            _t: &HealTarget,
            _i: &Candidate,
            _s: &str,
            _g: &GateOutcome,
        ) -> Result<PrDeliveryOutcome, DeliverRefusal> {
            self.calls.lock().unwrap().push("deliver".into());
            if let Some(e) = &self.pr_err {
                return Err(DeliverRefusal::retriable(e.clone()));
            }
            let mut out = fake_delivery();
            if self.ci_unavailable {
                out.ci.state = super::super::merge::CiState::Pending;
                out.ci.checks.clear();
                out.ci.observation_error = Some("HTTP 503".into());
            }
            Ok(out)
        }
        async fn abandon(&self, _s: &str) {
            self.calls.lock().unwrap().push("abandon".into());
        }
        async fn comment(&self, _i: &Candidate, _t: &str) -> Result<(), String> {
            self.calls.lock().unwrap().push("comment".into());
            Ok(())
        }
        fn now_ms(&self) -> u64 {
            if self.now == 0 {
                10 * CLAIM_TTL_MS
            } else {
                self.now
            }
        }
    }

    fn target() -> HealTarget {
        HealTarget {
            repo: "acme/widgets".into(),
            fix_repo: None,
            checkout: Some(Checkout::Project("widgets".into())),
            label: "self-heal".into(),
            base: "main".into(),
        }
    }

    fn item() -> Candidate {
        Candidate {
            repo: "acme/widgets".into(),
            number: 5,
            tier: Some(ProvenanceTier::Maintainer),
            labelled: true,
            created_ms: 0,
            kind: CandidateKind::Issue,
        }
    }

    /// Keep a typed handle AND a trait object over the same fake, so a test can
    /// assert on recorded calls without casting a `dyn` pointer back.
    fn io(f: Fake) -> (Arc<Fake>, Arc<dyn TickIo>) {
        let typed = Arc::new(f);
        let dynamic: Arc<dyn TickIo> = typed.clone();
        (typed, dynamic)
    }

    fn calls_of(f: &Arc<Fake>) -> Vec<String> {
        f.calls.lock().unwrap().clone()
    }

    fn fake_delivery() -> PrDeliveryOutcome {
        PrDeliveryOutcome {
            branch: "self-heal/5".into(),
            commit: "head123".into(),
            pushed: true,
            pr_number: 1,
            pr_url: "https://example/pr/1".into(),
            pr_action: super::super::merge::PrAction::Opened,
            draft: false,
            ci: CiSummary {
                observation_error: None,
                head_sha: "head123".into(),
                state: super::super::merge::CiState::Green,
                checks: vec![
                    super::super::merge::CiCheck {
                        name: "lint".into(),
                        state: super::super::merge::CiState::Green,
                    },
                    super::super::merge::CiCheck {
                        name: "test".into(),
                        state: super::super::merge::CiState::Green,
                    },
                ],
            },
        }
    }

    /// THE starvation regression, and it was opened by the tier fix itself.
    ///
    /// Before untiered candidates were allowed past selection, nothing reached
    /// `intent_for` and this path was dead. Now every labelled issue reaches
    /// it, and a refusal used to release the claim and record nothing — so the
    /// oldest labelled item with an untrusted author (the normal case in a
    /// public repo: a maintainer labels a contributor's bug report) was chosen
    /// on every tick, refused on every tick, and halted the sweep there. Every
    /// item behind it starved forever.
    #[tokio::test]
    async fn a_refused_author_is_recorded_so_the_queue_can_move_past_it() {
        let mut older = item();
        older.number = 1;
        older.created_ms = 1;
        let (_, f) = io(Fake {
            candidates: vec![older],
            // The fake's `None` intent is a gate refusal.
            intent: None,
            now: 1_000_000,
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&f, &target(), &mut claims, "run-1", &NoClaimSink).await;

        match out {
            TickOutcome::Idle { skipped } => assert!(matches!(
                skipped[0].reason,
                super::super::heal_select::Skip::UntrustedAuthor { .. }
            )),
            other => panic!("expected idle, got {other:?}"),
        }
        assert_eq!(claims.held_by("acme/widgets", 1), None);
        assert!(!claims.attempts().is_empty(), "no failure was recorded");

        // THE assertion that matters, and the one the first version of this
        // test was missing: it checked only that the WRITE happened. The
        // attempt ledger was written by `tick` and read by nobody —
        // `select` never received it and `Skip::RecentlyFailed` was
        // constructed nowhere in production — so the very next tick chose the
        // same item again and the sweep never advanced. Proving a backoff
        // requires ticking twice.
        let out2 = tick(&f, &target(), &mut claims, "run-2", &NoClaimSink).await;
        match out2 {
            TickOutcome::Idle { skipped } => assert!(
                matches!(
                    skipped[0].reason,
                    super::super::heal_select::Skip::RecentlyFailed { .. }
                ),
                "the second tick re-selected the item: {:?}",
                skipped[0].reason
            ),
            other => panic!("expected the item to be held back, got {other:?}"),
        }
    }

    /// The backoff must also EXPIRE, or a transient failure parks an item for
    /// good and the ledger becomes a denylist.
    #[tokio::test]
    async fn a_stale_refusal_becomes_eligible_again_once_its_backoff_passes() {
        let mut claims = ClaimStore::new();
        claims.record_failure("acme/widgets", 5, "transient", 0);
        let (_, f) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            // Past the first backoff step.
            now: super::super::heal_claims::BACKOFF_BASE_MS + 1,
            ..Default::default()
        });
        let out = tick(&f, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(
            !matches!(out, TickOutcome::Idle { .. }),
            "an expired backoff must release the item, got {out:?}"
        );
    }

    /// The other half: an item that VANISHED needs no backoff. It is gone from
    /// the queue, so a later tick will not see it, and recording a failure
    /// would leave a phantom entry against an issue nobody can look at.
    #[tokio::test]
    async fn an_item_that_vanished_is_not_recorded_as_a_failure() {
        struct Vanished;
        #[async_trait::async_trait]
        impl TickIo for Vanished {
            async fn candidates(&self, _t: &HealTarget) -> Result<Vec<Candidate>, String> {
                Ok(vec![item()])
            }
            async fn open_prs(&self, _t: &HealTarget) -> Result<Vec<RawPullRequest>, String> {
                Ok(vec![])
            }
            async fn intent_for(&self, _i: &Candidate) -> Result<Intent, String> {
                Ok(Intent::Gone)
            }
            fn redact(&self, t: &str) -> String {
                t.to_string()
            }
            async fn run_coder(
                &self,
                _t: &HealTarget,
                _i: &Candidate,
                _s: &SessionSeed,
            ) -> Result<Attempt, RunFailure> {
                panic!("a vanished item must not reach the coder")
            }
            async fn deliver(
                &self,
                _t: &HealTarget,
                _i: &Candidate,
                _s: &str,
                _g: &GateOutcome,
            ) -> Result<PrDeliveryOutcome, DeliverRefusal> {
                panic!("a vanished item must not be delivered")
            }
            async fn abandon(&self, _s: &str) {}
            async fn comment(&self, _i: &Candidate, _t: &str) -> Result<(), String> {
                Ok(())
            }
            fn now_ms(&self) -> u64 {
                1_000_000
            }
        }
        let io: Arc<dyn TickIo> = Arc::new(Vanished);
        let mut claims = ClaimStore::new();
        let out = tick(&io, &target(), &mut claims, "run-1", &NoClaimSink).await;
        match out {
            TickOutcome::Idle { skipped } => {
                assert_eq!(skipped[0].reason, super::super::heal_select::Skip::Gone)
            }
            other => panic!("expected idle, got {other:?}"),
        }
        assert!(
            claims.attempts().is_empty(),
            "a vanished item leaves no backoff against an issue nobody can see"
        );
    }

    /// A stale tier is not an authorisation failure, and saying so would send
    /// an operator to check permissions when the remedy is to retry.
    #[tokio::test]
    async fn a_stale_tier_reports_as_not_cleared_not_as_untrusted() {
        struct Stale;
        #[async_trait::async_trait]
        impl TickIo for Stale {
            async fn candidates(&self, _t: &HealTarget) -> Result<Vec<Candidate>, String> {
                Ok(vec![item()])
            }
            async fn open_prs(&self, _t: &HealTarget) -> Result<Vec<RawPullRequest>, String> {
                Ok(vec![])
            }
            async fn intent_for(&self, _i: &Candidate) -> Result<Intent, String> {
                Ok(Intent::Refused {
                    tier: ProvenanceTier::Maintainer,
                    stale: true,
                })
            }
            fn redact(&self, t: &str) -> String {
                t.to_string()
            }
            async fn run_coder(
                &self,
                _t: &HealTarget,
                _i: &Candidate,
                _s: &SessionSeed,
            ) -> Result<Attempt, RunFailure> {
                panic!("unreached")
            }
            async fn deliver(
                &self,
                _t: &HealTarget,
                _i: &Candidate,
                _s: &str,
                _g: &GateOutcome,
            ) -> Result<PrDeliveryOutcome, DeliverRefusal> {
                panic!("unreached")
            }
            async fn abandon(&self, _s: &str) {}
            async fn comment(&self, _i: &Candidate, _t: &str) -> Result<(), String> {
                Ok(())
            }
            fn now_ms(&self) -> u64 {
                1_000_000
            }
        }
        let io: Arc<dyn TickIo> = Arc::new(Stale);
        let mut claims = ClaimStore::new();
        match tick(&io, &target(), &mut claims, "run-1", &NoClaimSink).await {
            TickOutcome::Idle { skipped } => assert_eq!(
                skipped[0].reason,
                super::super::heal_select::Skip::IntentNotCleared
            ),
            other => panic!("expected idle, got {other:?}"),
        }
        // Still recorded: a stale tier that keeps being stale would otherwise
        // halt the sweep on this item exactly as an untrusted author did.
        assert!(!claims.attempts().is_empty());
    }

    /// A coder failure that left a live session must close it out.
    ///
    /// `run` can fail AFTER the session reached `NeedsApproval` — no worktree,
    /// an unreadable diff, a green contract over an unchanged worktree — and a
    /// plain `String` error gave the caller no way to name it. Each one leaked
    /// a git worktree registered in the operator's own repository, with the
    /// task handle already taken so `coder.cancel` could not reach it, and the
    /// reaper that would have swept them was deleted on the strength of the
    /// invariant these paths break.
    #[tokio::test]
    async fn a_failed_coder_run_still_closes_its_session() {
        let (typed, f) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            coder_err: Some("the worktree is unchanged".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&f, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Failed { .. }));
        assert!(
            calls_of(&typed).contains(&"abandon".to_string()),
            "the session was left non-terminal, holding a worktree: {:?}",
            calls_of(&typed)
        );
    }

    /// Exhausting reviewer exclusions is a daemon configuration failure, not
    /// evidence that the selected backlog item is bad. The tick still closes
    /// the session and releases its claim, but leaves neither item backoff nor
    /// a public failure comment.
    #[tokio::test]
    async fn no_independent_coder_does_not_penalize_or_comment_on_the_item() {
        let (typed, f) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            coder_err: Some("no independent coder model is available; configure heal.toml".into()),
            coder_configuration_error: true,
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&f, &target(), &mut claims, "run-1", &NoClaimSink).await;

        assert!(matches!(out, TickOutcome::Failed { .. }));
        assert!(claims.attempts().is_empty(), "item gained failure backoff");
        assert_eq!(claims.held_by("acme/widgets", 5), None);
        assert_eq!(
            calls_of(&typed),
            vec!["candidates", "intent", "coder", "abandon"],
            "configuration failure must not post on the backlog item"
        );
    }

    #[tokio::test]
    async fn an_empty_queue_is_idle_not_failure() {
        let (_, f) = io(Fake {
            intent: Some("fix it".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&f, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Idle { .. }));
    }

    #[tokio::test]
    async fn a_watch_only_target_spends_no_api_call() {
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            ..Default::default()
        });
        let mut t = target();
        t.checkout = None;
        let out = tick(&arc, &t, &mut claims_new(), "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Idle { .. }));
        // Reading a queue we can do nothing about is pure cost.
        assert!(calls_of(&typed).is_empty());
    }

    fn claims_new() -> ClaimStore {
        ClaimStore::new()
    }

    #[tokio::test]
    async fn an_invalid_repo_spec_fails_rather_than_idling() {
        let (_, f) = io(Fake::default());
        let mut t = target();
        t.repo = "not-a-spec".into();
        let out = tick(&f, &t, &mut claims_new(), "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Failed { .. }));
    }

    #[tokio::test]
    async fn unavailable_ci_keeps_delivery_claim_without_failure_backoff() {
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix parser".into()),
            ci_unavailable: true,
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(
            matches!(out, TickOutcome::Opened { ref delivery, .. } if delivery.contains("CI unavailable"))
        );
        assert_eq!(claims.held_by("acme/widgets", 5), Some("run-1"));
        assert!(claims.attempts().is_empty());
        assert!(!calls_of(&typed).contains(&"abandon".to_string()));
        let again = tick(&arc, &target(), &mut claims, "run-2", &NoClaimSink).await;
        assert!(matches!(again, TickOutcome::Idle { .. }));
        assert_eq!(
            calls_of(&typed)
                .iter()
                .filter(|c| c.as_str() == "deliver")
                .count(),
            1
        );
    }

    #[tokio::test]
    async fn a_happy_tick_claims_then_works_then_opens() {
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix the parser".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        match out {
            TickOutcome::Opened {
                number,
                pr_url,
                ci,
                delivery,
                ..
            } => {
                assert_eq!(number, 5);
                assert!(pr_url.contains("pr/1"));
                assert_eq!(ci.head_sha, "head123");
                assert_eq!(ci.state, super::super::merge::CiState::Green);
                assert_eq!(
                    delivery,
                    "delivered with green checks and ready for review at head123"
                );
            }
            other => panic!("expected Opened, got {other:?}"),
        }
        // The claim is HELD after success: the item is not free until a human
        // closes it, and a later tick must not immediately redo the work.
        assert_eq!(claims.held_by("acme/widgets", 5), Some("run-1"));

        let calls = calls_of(&typed);
        let claim_before_work = calls.iter().position(|c| c == "coder").unwrap();
        assert!(
            calls[..claim_before_work].contains(&"intent".to_string()),
            "intent is cleared before the coder runs: {calls:?}"
        );
    }

    #[tokio::test]
    async fn an_uncleared_intent_releases_the_claim() {
        // `None` means the item could not be trust-cleared. It must never fall
        // back to the raw body, and it must not leave the item claimed.
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: None,
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Idle { .. }));
        assert_eq!(claims.held_by("acme/widgets", 5), None, "claim released");

        assert!(
            !calls_of(&typed).contains(&"coder".to_string()),
            "the coder must not run on uncleared intent"
        );
    }

    #[tokio::test]
    async fn a_rejected_gate_comments_and_releases() {
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            attempt: Some(Attempt {
                contract_passed: true,
                contract_detail: "green".into(),
                panel_size: 3,
                verdicts: vec![
                    Verdict {
                        model: "a".into(),
                        pass: true,
                        reason: "ok".into(),
                    },
                    Verdict {
                        model: "b".into(),
                        pass: false,
                        reason: "changes unrelated behaviour".into(),
                    },
                    Verdict {
                        model: "c".into(),
                        pass: false,
                        reason: "scope".into(),
                    },
                ],
                unreachable: vec![],
                session_id: "coder-e2e-1".into(),
            }),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Rejected { .. }));
        assert_eq!(claims.held_by("acme/widgets", 5), None);

        let calls = calls_of(&typed);
        assert!(
            calls.contains(&"comment".to_string()),
            "must explain itself"
        );
        assert!(
            !calls.contains(&"deliver".to_string()),
            "a rejected change must not be opened"
        );
    }

    #[tokio::test]
    async fn a_failed_coder_releases_the_claim() {
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            coder_err: Some("worktree gone".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Failed { .. }));
        assert_eq!(claims.held_by("acme/widgets", 5), None);
    }

    #[tokio::test]
    async fn a_failed_pr_open_releases_so_a_later_tick_can_retry() {
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            pr_err: Some("gh auth expired".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Failed { .. }));
        assert_eq!(
            claims.held_by("acme/widgets", 5),
            None,
            "otherwise the item sits claimed and invisible"
        );
    }

    #[tokio::test]
    async fn an_item_claimed_by_another_run_is_left_alone() {
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        claims
            .claim("acme/widgets", 5, "other-run", 10 * CLAIM_TTL_MS)
            .unwrap();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Idle { .. }));
        assert_eq!(claims.held_by("acme/widgets", 5), Some("other-run"));

        assert!(!calls_of(&typed).contains(&"coder".to_string()));
    }

    #[tokio::test]
    async fn an_expired_claim_lets_a_later_tick_pick_it_up() {
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        // The fake's clock is 10x the TTL, so a claim at time 0 is long dead.
        claims.claim("acme/widgets", 5, "dead-run", 0).unwrap();
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Opened { .. }));
    }

    #[tokio::test]
    async fn an_issue_already_covered_is_not_worked_twice() {
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            prs: vec![RawPullRequest::new(
                "acme/widgets",
                9,
                "someone",
                "fix",
                "closes #5",
                vec![],
                ChecksState::Passing,
                "",
                false,
            )],
            intent: Some("fix it".into()),
            ..Default::default()
        });
        let out = tick(&arc, &target(), &mut claims_new(), "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Idle { .. }));
    }

    #[test]
    fn a_project_target_and_a_local_target_map_to_different_coder_arguments() {
        let (repo, project) = coder_target(&target()).unwrap();
        assert!(repo.is_none() && project.as_deref() == Some("widgets"));

        let mut t = target();
        t.checkout = Some(Checkout::Local("/tmp/x".into()));
        let (repo, project) = coder_target(&t).unwrap();
        assert!(repo.is_some() && project.is_none());
    }

    #[tokio::test]
    async fn a_rejected_item_is_not_retried_on_the_very_next_tick() {
        // THE bug both reviews ranked first. Oldest-first selection plus
        // release-on-failure plus no memory = a metronome pointed at one issue,
        // burning a coder session and a model panel every tick while everything
        // behind it starves.
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            attempt: Some(Attempt {
                contract_passed: true,
                contract_detail: "green".into(),
                panel_size: 3,
                verdicts: vec![
                    Verdict {
                        model: "a".into(),
                        pass: false,
                        reason: "no".into(),
                    },
                    Verdict {
                        model: "b".into(),
                        pass: false,
                        reason: "no".into(),
                    },
                ],
                unreachable: vec![],
                session_id: "coder-b".into(),
            }),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        let first = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(first, TickOutcome::Rejected { .. }));
        assert_eq!(claims.attempts().len(), 1, "the failure is remembered");

        // The claim was released, so only the backoff can stop a re-pick.
        let a = &claims.attempts()[&crate::coder::heal_select::claim_key("acme/widgets", 5)];
        assert!(!a.ready(arc.now_ms()), "not eligible again immediately");
    }

    #[tokio::test]
    async fn a_caller_cannot_mint_approval_it_did_not_earn() {
        // `run_coder` returns evidence; only `decide` builds a verdict. A
        // failing contract cannot be reported as approved no matter what the
        // panel says.
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            attempt: Some(Attempt {
                contract_passed: false,
                contract_detail: "cargo test failed".into(),
                panel_size: 3,
                verdicts: vec![
                    Verdict {
                        model: "a".into(),
                        pass: true,
                        reason: "lgtm".into(),
                    },
                    Verdict {
                        model: "b".into(),
                        pass: true,
                        reason: "lgtm".into(),
                    },
                    Verdict {
                        model: "c".into(),
                        pass: true,
                        reason: "lgtm".into(),
                    },
                ],
                unreachable: vec![],
                session_id: "coder-b".into(),
            }),
            ..Default::default()
        });
        let out = tick(
            &arc,
            &target(),
            &mut ClaimStore::new(),
            "run-1",
            &NoClaimSink,
        )
        .await;
        match out {
            TickOutcome::Rejected { gate, .. } => {
                assert!(gate.contains("contract"), "{gate}");
            }
            other => panic!("a red contract must not open a PR: {other:?}"),
        }
    }

    #[tokio::test]
    async fn a_success_clears_the_failure_history() {
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            ..Default::default()
        });
        let mut claims = ClaimStore::new();
        claims.record_failure("acme/widgets", 5, "earlier", 0);
        let out = tick(&arc, &target(), &mut claims, "run-1", &NoClaimSink).await;
        assert!(matches!(out, TickOutcome::Opened { .. }));
        assert!(claims.attempts().is_empty(), "a later failure starts clean");
    }

    #[tokio::test]
    async fn every_stop_says_why_on_the_item() {
        // A gate that passed but whose PR could not be opened is the case a
        // human most needs told, and it used to return silently.
        let (typed, arc) = io(Fake {
            candidates: vec![item()],
            intent: Some("fix it".into()),
            pr_err: Some("gh auth expired".into()),
            ..Default::default()
        });
        let out = tick(
            &arc,
            &target(),
            &mut ClaimStore::new(),
            "run-1",
            &NoClaimSink,
        )
        .await;
        assert!(matches!(out, TickOutcome::Failed { .. }));
        assert!(
            calls_of(&typed).contains(&"comment".to_string()),
            "a silent failure teaches people to ignore the loop"
        );
    }

    #[tokio::test]
    async fn a_watch_only_target_reports_why_it_did_nothing() {
        let (_, arc) = io(Fake::default());
        let mut t = target();
        t.checkout = None;
        match tick(&arc, &t, &mut ClaimStore::new(), "run-1", &NoClaimSink).await {
            TickOutcome::Idle { skipped } => {
                assert_eq!(skipped.len(), 1);
                assert_eq!(
                    skipped[0].reason,
                    crate::coder::heal_select::Skip::WatchOnly
                );
            }
            other => panic!("expected an explained idle, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn an_uncleared_intent_reports_clearance_not_authorisation() {
        // Reporting "untrusted author: maintainer" sends an operator to check
        // permissions when the remedy is to retry.
        let (_, arc) = io(Fake {
            candidates: vec![item()],
            intent: None,
            intent_stale: true,
            ..Default::default()
        });
        match tick(
            &arc,
            &target(),
            &mut ClaimStore::new(),
            "run-1",
            &NoClaimSink,
        )
        .await
        {
            TickOutcome::Idle { skipped } => {
                assert_eq!(
                    skipped[0].reason,
                    crate::coder::heal_select::Skip::IntentNotCleared
                );
            }
            other => panic!("expected an explained idle, got {other:?}"),
        }
    }
}