choir_bridge/

queue.rs

//! Queue-as-bot v0 (DECISIONS.md D21, queue stage): speculative merge
//! trains, checked either by the host forge's CI or by our own (D18).
//!
//! v0 is verdict-only: it builds a train commit (base tip + each open
//! PR merged in submission order, conflicts excluded), gets a verdict
//! on it, then reports a per-PR verdict as a commit status. It never
//! moves the protected branch on its own — landing the train is a
//! separate, explicitly requested step.
//!
//! The check signal has two shapes. The forge one publishes the train
//! as the `choir/train` branch and polls the forge's check API, which
//! means the speculative merge of every open PR is pushed to a public
//! remote and tested by someone else's runners. [`run_train_ci`] is the
//! other: it hands the train to a [`choir_queue::executor::CiExecutor`],
//! so the same train is checked here, on a provider we chose, and
//! nothing speculative leaves the machine.
//!
//! Everything in this module is local git; the forge API glue lives in
//! [`crate::github`] so the train mechanics stay testable offline.

use choir_queue::differential_ledger::{effective_environment, CommandSpec};
use choir_queue::executor::{CiExecutor, Job, Verdict, PROTOCOL};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

static DIFFERENTIAL_RUN_ID: AtomicU64 = AtomicU64::new(0);

/// One PR's fate in a train build.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TrainEntry {
    /// PR number (or any caller-chosen id in tests).
    pub id: u64,
    /// The PR head oid this entry was built from.
    pub head: String,
    /// Whether the merge onto the train succeeded.
    pub merged: bool,
    /// The speculative merge commit, absent when the PR conflicted.
    pub merge: Option<String>,
    /// Human-readable note (merge position or the conflict reason).
    pub note: String,
    /// The PR's every commit is already in the train by patch identity
    /// (`git patch-id --stable` equivalence via
    /// `git cherry`): the train rewrote or landed this change earlier,
    /// so it was recognized rather than re-merged — a success, not a
    /// conflict, despite `merged` being false.
    pub already_landed: bool,
}

/// Result of building one speculative train.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Train {
    /// Tip commit of the train (== base when nothing merged).
    pub tip: String,
    /// Per-PR outcomes, in build order.
    pub entries: Vec<TrainEntry>,
}

/// Runs git in `dir` with a fixed bot identity, returning stdout.
fn git(dir: &Path, args: &[&str]) -> Result<String, String> {
    let out = std::process::Command::new("git")
        .arg("-c")
        .arg("user.name=choir-queue")
        .arg("-c")
        .arg("user.email=queue@choir.invalid")
        .arg("-c")
        .arg("commit.gpgsign=false")
        .args(args)
        .current_dir(dir)
        .env("GIT_TERMINAL_PROMPT", "0")
        .output()
        .map_err(|e| format!("spawn git: {e}"))?;
    if out.status.success() {
        Ok(String::from_utf8_lossy(&out.stdout).into_owned())
    } else {
        Err(format!(
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        ))
    }
}

/// Builds a speculative train in `repo` (a non-bare clone whose
/// worktree this function owns): detaches at `base`, then merges each
/// `(id, head)` in order with `--no-ff`. A conflicting PR is excluded
/// (merge aborted) and the train continues without it.
///
/// # Errors
///
/// Git failures other than merge conflicts (missing oids, dirty repo).
pub fn build_train(repo: &Path, base: &str, prs: &[(u64, String)]) -> Result<Train, String> {
    git(repo, &["checkout", "-q", "--detach", base])?;
    let mut entries = Vec::new();
    let mut position = 0usize;
    for (id, head) in prs {
        // Stable change identity: a PR whose every commit
        // is already in the train by patch identity — the train landed a
        // rewritten form of it, or it was rebased in — is recognized,
        // not re-merged. Checked against the current tip so a duplicate
        // of an earlier train member is caught too.
        if already_in(repo, "HEAD", head)? {
            entries.push(TrainEntry {
                id: *id,
                head: head.clone(),
                merged: false,
                merge: None,
                note: "already landed (patch identity)".to_string(),
                already_landed: true,
            });
            continue;
        }
        let msg = format!("choir train: PR #{id}");
        match git(repo, &["merge", "--no-ff", "-q", "-m", &msg, head]) {
            Ok(_) => {
                position += 1;
                let merge = git(repo, &["rev-parse", "HEAD"])?.trim().to_string();
                entries.push(TrainEntry {
                    id: *id,
                    head: head.clone(),
                    merged: true,
                    merge: Some(merge),
                    note: format!("train position {position}"),
                    already_landed: false,
                });
            }
            Err(_) => {
                // Conflict (or unmergeable): drop this PR from the train.
                git(repo, &["merge", "--abort"]).ok();
                entries.push(TrainEntry {
                    id: *id,
                    head: head.clone(),
                    merged: false,
                    merge: None,
                    note: "conflicts with train".to_string(),
                    already_landed: false,
                });
            }
        }
    }
    let tip = git(repo, &["rev-parse", "HEAD"])?.trim().to_string();
    Ok(Train { tip, entries })
}

/// Whether every commit of `head` is already contained in `upstream` by
/// patch identity. `git cherry` marks a commit `-` when an equivalent
/// change (same `git patch-id --stable`) is upstream, `+` when it is
/// not; no `+` lines — including the empty output of a plain ancestor —
/// means there is nothing left to merge.
pub fn already_in(repo: &Path, upstream: &str, head: &str) -> Result<bool, String> {
    let out = git(repo, &["cherry", upstream, head])?;
    Ok(out.lines().all(|line| !line.starts_with('+')))
}

/// Structured advisory result returned by the separately built D23 runner.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DifferentialOutcome {
    /// Monotonic calibration-ledger observation id.
    pub observation_id: u64,
    /// Stable classifier spelling from the runner.
    pub verdict: DifferentialVerdict,
    /// Number of interaction flags still awaiting ground truth.
    pub pending_interactions: u64,
}

/// Closed set of D23 classifications accepted from the runner.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DifferentialVerdict {
    /// Both parents and the merge passed.
    Clean,
    /// Both parents passed and the merge failed.
    InteractionFailure,
    /// At least one parent failed.
    InconclusiveParentFailure,
}

impl DifferentialVerdict {
    /// The ledger's stable spelling. CLI output prints this rather than the
    /// Rust `Debug` name so every consumer — ledger rows, receipts, stdout —
    /// parses one vocabulary; the two spellings once cost a downstream
    /// summarizer a silent zero.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Clean => "clean",
            Self::InteractionFailure => "interaction_failure",
            Self::InconclusiveParentFailure => "inconclusive_parent_failure",
        }
    }
}

fn compatible_confidence_policy(value: &serde_json::Value) -> bool {
    value["format_version"].as_u64() == Some(1)
        && value["method"].as_str() == Some("one_sided_exact_binomial_zero_spurious")
        && value["confidence"]["numerator"].as_u64() == Some(95)
        && value["confidence"]["denominator"].as_u64() == Some(100)
        && value["target"]["numerator"].as_u64() == Some(1)
        && value["target"]["denominator"].as_u64() == Some(1000)
        && value["target"]["comparison"].as_str() == Some("strictly_less_than")
        && value["minimum_evaluated_merges"].as_u64() == Some(2_995)
        && value["requires_zero_spurious_failures"].as_bool() == Some(true)
        && value["assumptions"].as_array().is_some_and(|assumptions| {
            assumptions
                == &[
                    serde_json::Value::String("independent_runs".to_string()),
                    serde_json::Value::String(
                        "representative_queue_command_and_merge_population".to_string(),
                    ),
                ]
        })
}

fn cleanup_worktrees(repo: &Path, root: &Path, paths: &[PathBuf]) -> Result<(), String> {
    let mut first_error = None;
    for path in paths.iter().rev() {
        let path_text = path.to_string_lossy().into_owned();
        if let Err(error) = git(repo, &["worktree", "remove", "--force", &path_text]) {
            first_error.get_or_insert(error);
        }
    }
    if root.exists() {
        if let Err(error) = std::fs::remove_dir_all(root) {
            first_error
                .get_or_insert_with(|| format!("remove differential checkout root: {error}"));
        }
    }
    if let Some(parent) = root.parent() {
        std::fs::remove_dir(parent).ok();
    }
    first_error.map_or(Ok(()), Err)
}

/// Three revision worktrees whose lifetime is a whole **calibration run**
/// rather than a single observation.
///
/// The observation cost the harness cannot avoid is compiling the project
/// under test three times. The cost it *can* avoid is compiling the project's
/// dependencies three times per observation, which is what a fresh worktree
/// per observation forces: the operator's `--target-dir` is tree-relative (the
/// ledger refuses any target directory that escapes its worktree), so
/// destroying the tree destroys the build directory with it and the next
/// observation starts from zero. Holding the same three trees open across
/// observations and moving them with `git checkout` keeps those build
/// directories alive. Measured on `rust-lang/log`, one run: **16.1 s** in a
/// fresh tree against **12.0-13.4 s** in a held tree switched to a new
/// revision.
///
/// Three properties are deliberately preserved, because each of them is load
/// bearing for what the calibration claims:
///
/// - **Every run is still really run.** A held tree makes a run cheaper, never
///   skipped: `cargo test` re-links and re-executes the test binaries even when
///   nothing changed (measured: four test binaries execute in a fully warm
///   tree). That is what the ledger's `independent_runs` assumption needs, and
///   it is why this is not a result cache.
/// - **The three trees stay separate.** Each keeps its own build directory, so
///   the cross-tree artifact contamination that invalidated an earlier receipt
///   cannot recur, and the three concurrent runs still cannot contend on one
///   tool lock.
/// - **Nothing survives the run.** The worktrees and their root are removed by
///   [`Self::close`], and again by `Drop` if a caller returns early, so an
///   operator's repository is left as it was found.
///
/// What it does change, stated rather than buried: an observation now starts in
/// a tree that holds an *earlier* observation's untracked build output.
/// Tracked content is exact — the checkout is forced, so the tree matches the
/// revision — but a command that writes untracked files into its tree will see
/// them again. Callers that need a pristine tree per observation still have
/// one: [`run_differential`] opens and closes a session around a single
/// observation, and `choir-bridge calibrate --fresh-worktrees` selects that
/// path for a whole run.
///
/// Which tree plays which role is decided per observation: a revision is
/// assigned to a tree that is already sitting on it, when one is. Replaying
/// consecutive first-parent merges — the shape of a calibration corpus — makes
/// this pay every observation, because a merge's first parent *is* the
/// previous observation's merge: the tree that just ran that merge becomes the
/// parent-a tree with a no-op checkout, so its command rebuilds nothing at
/// all. Measured on `rust-lang/log`, one warm tree, solo: **8.9 s** for a
/// no-op revision against **15.0 s** for a one-merge move.
pub struct DifferentialSession {
    repo: PathBuf,
    root: PathBuf,
    /// `None` until the first observation names the revisions to check out.
    /// Worktree creation needs a revision, so there is nothing useful to
    /// create at open time. Each entry pairs a worktree path with the
    /// revision the tree currently holds, which is what role assignment
    /// matches against.
    trees: Option<[(PathBuf, String); 3]>,
}

impl DifferentialSession {
    /// Names a session root under `repo`. Creates nothing until the first
    /// observation; opening cannot fail.
    #[must_use]
    pub fn open(repo: &Path) -> Self {
        Self {
            repo: repo.to_path_buf(),
            root: repo.join(".choir-differential").join(format!(
                "run-{}-{}",
                std::process::id(),
                DIFFERENTIAL_RUN_ID.fetch_add(1, Ordering::Relaxed)
            )),
            trees: None,
        }
    }

    /// Puts the three trees on the three requested revisions, creating them on
    /// the first call and moving them on every later one.
    fn prepare(
        &mut self,
        parent_a: &str,
        parent_b: &str,
        merged: &str,
    ) -> Result<[PathBuf; 3], String> {
        let revisions = [parent_a, parent_b, merged];
        if let Some(trees) = &mut self.trees {
            // A tree already holding a requested revision keeps it, so the
            // checkout below is a no-op there and the command that follows
            // rebuilds nothing. Roles are not pinned to trees: on a corpus of
            // consecutive first-parent merges, parent a of this observation is
            // the previous observation's merge, and this hands that revision
            // its still-built tree. Unmatched roles take the leftover trees in
            // order, which keeps the assignment stable when nothing matches.
            let mut assigned = [usize::MAX; 3];
            let mut used = [false; 3];
            for (role, revision) in revisions.iter().enumerate() {
                if let Some(index) =
                    (0..trees.len()).find(|&index| !used[index] && trees[index].1 == *revision)
                {
                    assigned[role] = index;
                    used[index] = true;
                }
            }
            for slot in &mut assigned {
                if *slot == usize::MAX {
                    let index = used
                        .iter()
                        .position(|taken| !taken)
                        .expect("three roles cannot exhaust three trees");
                    *slot = index;
                    used[index] = true;
                }
            }
            let mut paths = Vec::with_capacity(revisions.len());
            for (role, revision) in revisions.iter().enumerate() {
                let (path, head) = &mut trees[assigned[role]];
                // `--force` even when the tree already holds the revision: the
                // tree is the harness's own scratch checkout and an
                // observation must start from exactly this revision's tracked
                // content, so a command that dirtied a tracked file (a
                // lockfile, say) must not be able to leak it into the next
                // observation. It leaves untracked build output alone, which
                // is the point of holding the tree at all.
                git(path, &["checkout", "-q", "--detach", "--force", revision])?;
                *head = (*revision).to_string();
                paths.push(path.clone());
            }
            return Ok(paths
                .try_into()
                .expect("three roles produce three tree paths"));
        }
        std::fs::create_dir_all(&self.root)
            .map_err(|error| format!("create differential checkout root: {error}"))?;
        let mut created: Vec<(PathBuf, String)> = Vec::with_capacity(revisions.len());
        // Neutral names: role assignment above may hand any tree to any role
        // from the second observation on, so role-named directories would lie.
        for (name, revision) in [
            ("tree-0", parent_a),
            ("tree-1", parent_b),
            ("tree-2", merged),
        ] {
            let path = self.root.join(name);
            let path_text = path.to_string_lossy().into_owned();
            if let Err(error) = git(
                &self.repo,
                &["worktree", "add", "-q", "--detach", &path_text, revision],
            ) {
                let paths: Vec<PathBuf> = created.into_iter().map(|(path, _)| path).collect();
                cleanup_worktrees(&self.repo, &self.root, &paths).ok();
                return Err(error);
            }
            created.push((path, revision.to_string()));
        }
        let trees: [(PathBuf, String); 3] = created
            .try_into()
            .map_err(|_| "differential session needs exactly three worktrees".to_string())?;
        let paths = [trees[0].0.clone(), trees[1].0.clone(), trees[2].0.clone()];
        self.trees = Some(trees);
        Ok(paths)
    }

    fn cleanup(&mut self) -> Result<(), String> {
        let paths: Vec<PathBuf> = self
            .trees
            .take()
            .into_iter()
            .flatten()
            .map(|(path, _)| path)
            .collect();
        cleanup_worktrees(&self.repo, &self.root, &paths)
    }

    /// Removes the three worktrees and the session root, reporting the first
    /// failure. `Drop` repeats this as a best-effort backstop, so a caller that
    /// returns early still leaves nothing behind — but only `close` can tell
    /// the caller that cleanup failed.
    ///
    /// # Errors
    ///
    /// A worktree or the session root could not be removed.
    pub fn close(mut self) -> Result<(), String> {
        self.cleanup()
    }
}

impl Drop for DifferentialSession {
    fn drop(&mut self) {
        // Best effort: `close` is the reporting path. This exists so an early
        // return or a panic mid-run cannot leave `.choir-differential` in an
        // operator's repository.
        self.cleanup().ok();
    }
}

/// Resolves a merge commit to its canonical oid and its exact first and second
/// parents. Resolution happens before any checkout so an abbreviated or
/// non-canonical spelling cannot reach the ledger.
fn resolve_merge_revisions(repo: &Path, merge: &str) -> Result<(String, String, String), String> {
    let commit = format!("{merge}^{{commit}}");
    let merged_revision = git(repo, &["rev-parse", "--verify", &commit])?
        .trim()
        .to_string();
    let first = format!("{merged_revision}^1");
    let second = format!("{merged_revision}^2");
    let parent_a = git(repo, &["rev-parse", &first])?.trim().to_string();
    let parent_b = git(repo, &["rev-parse", &second])?.trim().to_string();
    Ok((parent_a, parent_b, merged_revision))
}

/// Runs one observation in an already-open session's three worktrees.
///
/// This is the loop body of a calibration run: the session is opened once and
/// reused, so the build directories the operator's command creates survive
/// between observations. Everything about the result is unchanged — the same
/// runner, the same argv, the same structured receipt validation as the
/// single-observation [`run_differential`].
///
/// The result is always advisory. This function only validates and returns a
/// closed structured result; callers have no landing-gate output to consume.
/// The state directory and command file are explicit arguments, never
/// environment configuration.
///
/// # Errors
///
/// The merge does not have two parents, a worktree cannot be created or moved
/// to the requested revision, the runner fails operationally, or its structured
/// result is malformed/claims that landing gating is enabled.
pub fn run_differential_in(
    session: &mut DifferentialSession,
    merge: &str,
    runner: &Path,
    command_file: &Path,
    state_dir: &Path,
) -> Result<DifferentialOutcome, String> {
    let (parent_a, parent_b, merged_revision) = resolve_merge_revisions(&session.repo, merge)?;
    let trees = session.prepare(&parent_a, &parent_b, &merged_revision)?;
    let output = std::process::Command::new(runner)
        .arg("run")
        .arg(command_file)
        .arg(state_dir)
        .arg(&parent_a)
        .arg(&trees[0])
        .arg(&parent_b)
        .arg(&trees[1])
        .arg(&merged_revision)
        .arg(&trees[2])
        .output()
        .map_err(|error| format!("spawn differential runner: {error}"))?;
    if !output.status.success() {
        return Err("differential runner exited unsuccessfully".to_string());
    }
    let value: serde_json::Value = serde_json::from_slice(&output.stdout)
        .map_err(|_| "differential runner returned malformed JSON".to_string())?;
    if value["format_version"].as_u64() != Some(1)
        || value["merge"].as_str() != Some(merged_revision.as_str())
        || value["calibration"]["landing_gate_enabled"].as_bool() != Some(false)
        || !matches!(
            &value["calibration"]["confidence_claim"],
            serde_json::Value::Null | serde_json::Value::Bool(_)
        )
        || !compatible_confidence_policy(&value["calibration"]["confidence_policy"])
    {
        return Err("differential runner returned an incompatible receipt".to_string());
    }
    let verdict = match value["report"]["verdict"].as_str() {
        Some("clean") => DifferentialVerdict::Clean,
        Some("interaction_failure") => DifferentialVerdict::InteractionFailure,
        Some("inconclusive_parent_failure") => DifferentialVerdict::InconclusiveParentFailure,
        _ => return Err("differential runner returned an unknown verdict".to_string()),
    };
    if !matches!(
        &value["calibration"]["target"]["met"],
        serde_json::Value::Null | serde_json::Value::Bool(_)
    ) {
        return Err("differential target verdict must be boolean or null".to_string());
    }
    Ok(DifferentialOutcome {
        observation_id: value["observation_id"]
            .as_u64()
            .ok_or("differential result needs an observation id")?,
        verdict,
        pending_interactions: value["calibration"]["pending_interactions"]
            .as_u64()
            .ok_or("differential receipt needs pending_interactions")?,
    })
}

/// Builds isolated worktrees for a speculative merge's two parents and the
/// merge itself, then invokes the explicit `choir-differential` runner.
///
/// This is [`run_differential_in`] wrapped in a session of its own, so the
/// three worktrees are created and destroyed around this one observation and
/// the command sees a pristine tree. That is the right shape for a speculative
/// train, where each merge is checked once, and it is the escape hatch for an
/// operator re-checking a flagged interaction without the previous
/// observation's build output present.
///
/// # Errors
///
/// As [`run_differential_in`], plus a failure to clean the worktrees up.
pub fn run_differential(
    repo: &Path,
    merge: &str,
    runner: &Path,
    command_file: &Path,
    state_dir: &Path,
) -> Result<DifferentialOutcome, String> {
    let mut session = DifferentialSession::open(repo);
    let result = run_differential_in(&mut session, merge, runner, command_file, state_dir);
    let cleanup = session.close();
    match (result, cleanup) {
        (Ok(value), Ok(())) => Ok(value),
        (Err(error), _) | (Ok(_), Err(error)) => Err(error),
    }
}

/// What one train verdict means for the PRs riding it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TrainReport {
    /// Whether the train may land. The only input to that decision, so
    /// that neither check signal gets a landing rule of its own.
    pub green: bool,
    /// Commit-status state: `success`, `failure`, or `error`.
    pub state: &'static str,
    /// Commit-status description. Fixed strings, never anything an
    /// executor or a pull request wrote — the Rule of Two applies to
    /// what we say as well as to what we read.
    pub description: &'static str,
}

/// Reads a train's verdict as the report its PRs get (D18, D49).
///
/// The four cases collapse to three states, and which two share one is
/// the whole point. `failure` blames the change, and only
/// [`Verdict::Failed`] may: an outage or a deadline posts `error`,
/// because telling an author their work is red when our provider fell
/// over is the confusion the seam exists to end. `green` is true for
/// exactly one case, so a train never lands on a verdict that merely
/// failed to say no.
#[must_use]
pub fn train_report(verdict: &Verdict) -> TrainReport {
    match verdict {
        Verdict::Passed => TrainReport {
            green: true,
            state: "success",
            description: "speculative train green",
        },
        Verdict::Failed { .. } => TrainReport {
            green: false,
            state: "failure",
            description: "train CI failed",
        },
        Verdict::Errored { .. } | Verdict::TimedOut => TrainReport {
            green: false,
            state: "error",
            description: "train CI could not run",
        },
    }
}

/// The report for a train we could not get a verdict on at all.
///
/// Same shape as a provider fault, because it is one: an executor that
/// could not be reached and an executor that failed to boot a VM are
/// the same news to the change riding the train.
#[must_use]
pub fn train_unavailable() -> TrainReport {
    train_report(&Verdict::Errored {
        provider: String::new(),
        detail: String::new(),
    })
}

/// Checks one train commit with a [`CiExecutor`] instead of asking the
/// host forge (D18).
///
/// The forge path and this one answer the same question by opposite
/// means. Asking the forge requires publishing the speculative merge of
/// every open PR to a remote and trusting whatever ran there; this
/// checks the train in the worktree that just built it, with an
/// operator-declared command, on a provider the operator picked. What
/// comes back is a [`Verdict`] rather than a bool, so the caller can
/// still tell "the train is bad" from "we could not find out" — the
/// distinction the whole seam exists for, and one the forge's
/// `status`/`conclusion` enums also make and the bool did not.
///
/// `repo` is checked out at `tip` first, forcibly: the train build owns
/// this worktree and left it detached at the train it built, but an
/// advisory differential run between the two may have moved it.
///
/// The job is content-addressed by `tip` under git's own codec, so the
/// subject a cache keys on and the subject a `RecordCheck` names are
/// the commit itself rather than a queue-local id. `may_write_cache`
/// stays false: a train holds unreviewed code from every open PR, which
/// is exactly the untrusted case the flag is for.
///
/// # Errors
///
/// Git could not resolve or check out `tip`, the executor could not be
/// reached, or it answered with something other than one verdict for
/// the one job. All three mean we did not find out, never that the
/// train is bad — a caller must not land on this and must not blame a
/// change for it.
pub fn run_train_ci(
    repo: &Path,
    tip: &str,
    spec: &CommandSpec,
    ci: &mut dyn CiExecutor,
) -> Result<Verdict, String> {
    let oid = git(
        repo,
        &["rev-parse", "--verify", &format!("{tip}^{{commit}}")],
    )?
    .trim()
    .to_string();
    let subject = choir_hash::ContentHash::from_git_oid(&oid)
        .ok_or_else(|| format!("train tip {oid} is not a git object id"))?;
    git(repo, &["checkout", "-q", "--detach", "--force", &oid])?;

    let mut command = Vec::with_capacity(spec.args.len() + 1);
    command.push(spec.program.clone());
    command.extend(spec.args.iter().cloned());
    let mut job = Job::new(subject, command);
    job.label = "train".to_string();
    job.environment = effective_environment(&spec.env);
    job.directory = Some(repo.to_path_buf());
    if let Some(seconds) = spec.timeout_seconds {
        job.deadline = Duration::from_secs(seconds);
    }

    // The handshake before the work, so a provider speaking another
    // protocol is refused rather than believed.
    let info = ci.info().map_err(|error| error.to_string())?;
    if info.protocol != PROTOCOL {
        return Err(format!(
            "executor `{}` speaks protocol {} and this build speaks {PROTOCOL}",
            info.name, info.protocol
        ));
    }
    let verdicts = ci.run(&[job]).map_err(|error| error.to_string())?;
    match verdicts.len() {
        1 => Ok(verdicts.into_iter().next().expect("length checked")),
        n => Err(format!(
            "executor `{}` answered {n} times for one job",
            info.name
        )),
    }
}

/// Runs the advisory detector for every merge commit in a train, preserving
/// train order and returning operational errors per entry instead of turning
/// them into a landing decision.
#[must_use]
pub fn run_train_differentials(
    repo: &Path,
    train: &Train,
    runner: &Path,
    command_file: &Path,
    state_dir: &Path,
) -> Vec<(u64, Result<DifferentialOutcome, String>)> {
    train
        .entries
        .iter()
        .filter(|entry| entry.merged)
        .map(|entry| {
            let result = entry
                .merge
                .as_deref()
                .ok_or("merged train entry is missing its merge commit".to_string())
                .and_then(|merge| run_differential(repo, merge, runner, command_file, state_dir));
            (entry.id, result)
        })
        .collect()
}

/// Two-parent merge commits along `repo`'s first-parent history, newest
/// first: the population `choir-bridge harvest` replays (D27).
///
/// Exactly two parents because the differential adapter seats exactly
/// three worktrees — parent a, parent b, merged — so an octopus merge has
/// no seat for its third parent and is enumerated past rather than failed
/// on. First-parent order for the same reason it is load-bearing in
/// choir-queue's corpus module: it lists the merges that landed on this
/// branch and skips commits internal to the branches they merged, which
/// is the population D23 cares about.
///
/// `limit` keeps only the most recent `limit` merges; 0 keeps them all.
/// Replay order is the caller's: oldest-first pays best with a held
/// [`DifferentialSession`], whose docs explain why.
///
/// # Errors
///
/// Git failing to spawn or exiting nonzero.
pub fn harvestable_merges(repo: &Path, limit: usize) -> Result<Vec<String>, String> {
    let log = git(
        repo,
        &["log", "--first-parent", "--merges", "--format=%H %P"],
    )?;
    let mut merges = parse_merge_list(&log);
    if limit > 0 {
        merges.truncate(limit);
    }
    Ok(merges)
}

/// Parses `git log --format="%H %P"` output into the ids of commits with
/// exactly two parents, preserving order.
///
/// Pure, so the format contract is testable without a repository — the
/// same parse/shell-out split choir-queue's corpus module uses.
#[must_use]
pub fn parse_merge_list(log: &str) -> Vec<String> {
    log.lines()
        .filter_map(|line| {
            let mut fields = line.split_whitespace();
            let id = fields.next()?;
            (fields.count() == 2).then(|| id.to_string())
        })
        .collect()
}

/// Merge commits already recorded in `state_dir`'s observation ledger,
/// keyed by the ledger row's `revisions.merged` oid.
///
/// This is what makes `choir-bridge harvest` incremental (D27): a re-run
/// against an updated mirror consults the ledger it is about to extend and
/// replays only the merges it has never seen. A missing or unreadable
/// ledger is an empty set — the first harvest into a fresh state directory
/// must not fail on its own absence — and a malformed row is skipped
/// rather than trusted, so it can never suppress a replay.
#[must_use]
pub fn observed_merges(state_dir: &Path) -> std::collections::BTreeSet<String> {
    std::fs::read_to_string(state_dir.join("observations.jsonl"))
        .map(|text| parse_observed_merges(&text))
        .unwrap_or_default()
}

/// Parses observation-ledger lines into the set of merged-revision oids.
///
/// Pure, so the ledger-row contract is testable without a state directory.
#[must_use]
pub fn parse_observed_merges(text: &str) -> std::collections::BTreeSet<String> {
    text.lines()
        .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
        .filter_map(|row| row["revisions"]["merged"].as_str().map(str::to_string))
        .collect()
}

/// Whether a path cannot change what a build-and-test command observes.
///
/// Deliberately a short, conservative allowlist rather than a guess at what
/// matters: documentation, licences, and forge bookkeeping. Everything else
/// — including `Cargo.toml`, `build.rs`, and any `.txt` that might be a test
/// fixture — counts as relevant, because the cost of wrongly skipping a
/// merge is a semantic conflict that never enters the corpus, while the cost
/// of wrongly keeping one is three builds.
///
/// # Examples
///
/// ```
/// use choir_bridge::queue::path_is_inert;
///
/// // Documentation and forge bookkeeping cannot move a test result.
/// assert!(path_is_inert("README.md"));
/// assert!(path_is_inert("docs/operating/limits.md"));
/// assert!(path_is_inert(".github/workflows/ci.yml"));
///
/// // Everything else counts, including the files that only look inert.
/// assert!(!path_is_inert("src/lib.rs"));
/// assert!(!path_is_inert("Cargo.toml"));
/// assert!(!path_is_inert("tests/fixtures/input.txt"));
/// ```
#[must_use]
pub fn path_is_inert(path: &str) -> bool {
    if path.starts_with(".github/") {
        return true;
    }
    let name = path.rsplit('/').next().unwrap_or(path);
    if name.ends_with(".md") {
        return true;
    }
    // `LICENSE-MIT` and `LICENSE-APACHE` are the usual dual-licence spelling,
    // so the licence names match by prefix; the rest match exactly, because a
    // prefix rule on them would swallow real source files.
    if name.starts_with("LICENSE") || name.starts_with("LICENCE") {
        return true;
    }
    let stem = name.split('.').next().unwrap_or(name);
    matches!(stem, "COPYING" | "NOTICE" | "AUTHORS" | "CHANGELOG")
        || matches!(name, ".gitignore" | ".gitattributes" | ".mailmap")
}

/// Whether every file this merge changed, against **either** parent, is
/// inert by [`path_is_inert`].
///
/// A semantic conflict is an interaction between the two branches' changes,
/// so the population that can possibly hold one is the union of the two
/// parent diffs. A merge whose whole union is documentation cannot fail a
/// build-and-test command in a way the parents pass, so harvesting it buys
/// three builds' worth of nothing. Skipping it is a **population
/// restriction**, not an optimization detail: callers record it rather than
/// applying it silently, because it changes which merges the denominator
/// counts.
///
/// An empty union — a merged tree identical to both parents — returns
/// `false`: it is strange enough to be worth observing rather than assuming
/// away.
///
/// # Errors
///
/// The merge cannot be resolved to two parents, or git fails.
pub fn merge_changes_only_inert_paths(repo: &Path, merge: &str) -> Result<bool, String> {
    let (parent_a, parent_b, merged) = resolve_merge_revisions(repo, merge)?;
    let mut any = false;
    for parent in [&parent_a, &parent_b] {
        let diff = git(repo, &["diff", "--name-only", parent, &merged])?;
        for path in diff.lines().map(str::trim).filter(|p| !p.is_empty()) {
            any = true;
            if !path_is_inert(path) {
                return Ok(false);
            }
        }
    }
    Ok(any)
}

/// Records, beside the ledger, every way a harvest narrowed the population
/// it walked: the merges excluded as inert, and the consecutive-inconclusive
/// run that stopped the walk early.
///
/// This exists so a corpus reader can tell a merge that was *observed and
/// found clean* from one that was *never observed*, which a ledger of
/// observations alone cannot express. Appends one versioned line per run, so
/// an incremental re-run's restrictions accumulate rather than overwrite.
/// Writes nothing when a run restricted nothing.
///
/// # Errors
///
/// The state directory cannot be created or the record cannot be appended.
pub fn record_population_restrictions(
    state_dir: &Path,
    inert: &[String],
    stopped_after_inconclusive: Option<usize>,
) -> Result<(), String> {
    if inert.is_empty() && stopped_after_inconclusive.is_none() {
        return Ok(());
    }
    let record = serde_json::json!({
        "format_version": 1,
        "inert_merges": inert,
        "inert_rule": "every path in the union of both parent diffs is documentation, licence, or forge bookkeeping",
        "stopped_after_consecutive_inconclusive": stopped_after_inconclusive,
    });
    std::fs::create_dir_all(state_dir).map_err(|error| format!("create state dir: {error}"))?;
    let mut line = serde_json::to_vec(&record)
        .map_err(|error| format!("serialize population restrictions: {error}"))?;
    line.push(b'\n');
    std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(state_dir.join("population.jsonl"))
        .and_then(|mut file| std::io::Write::write_all(&mut file, &line))
        .map_err(|error| format!("append population restrictions: {error}"))
}

/// Reproduction runs a harvest adds after a first `interaction_failure`
/// verdict, so a specimen records six runs in total — the bar specimen #1
/// (rust-lang/log `11eda98d`, reproduced 6/6) set for calling a flag a
/// semantic conflict rather than a flake.
pub const SPECIMEN_REPRODUCTION_RUNS: usize = 5;

/// Packages a flagged merge and its reproduction runs as a corpus
/// specimen under `state_dir/specimens/<merge>.json` (D27).
///
/// The specimen is metadata, not evidence: the observation rows it points
/// at (by id) stay in the ledger, and the JSON records how often the
/// interaction failure reproduced so a reader can tell a 6/6 conflict
/// from a 1/6 flake without replaying anything. Nothing is published;
/// the file lives in the operator's state directory.
///
/// # Errors
///
/// The merge cannot be resolved to two parents, or the specimen file
/// cannot be written.
pub fn write_specimen(
    repo: &Path,
    merge: &str,
    outcomes: &[Result<DifferentialOutcome, String>],
    state_dir: &Path,
) -> Result<PathBuf, String> {
    let (parent_a, parent_b, merged_revision) = resolve_merge_revisions(repo, merge)?;
    let mut interaction_failures = 0u64;
    let mut clean = 0u64;
    let mut inconclusive = 0u64;
    let mut errors = 0u64;
    let mut observation_ids = Vec::new();
    for outcome in outcomes {
        match outcome {
            Ok(outcome) => {
                observation_ids.push(outcome.observation_id);
                match outcome.verdict {
                    DifferentialVerdict::Clean => clean += 1,
                    DifferentialVerdict::InteractionFailure => interaction_failures += 1,
                    DifferentialVerdict::InconclusiveParentFailure => inconclusive += 1,
                }
            }
            Err(_) => errors += 1,
        }
    }
    let specimen = serde_json::json!({
        "format_version": 1,
        "merge": merged_revision,
        "parent_a": parent_a,
        "parent_b": parent_b,
        "runs": outcomes.len(),
        "interaction_failures": interaction_failures,
        "clean": clean,
        "inconclusive_parent_failures": inconclusive,
        "run_errors": errors,
        "observation_ids": observation_ids,
    });
    let dir = state_dir.join("specimens");
    std::fs::create_dir_all(&dir).map_err(|error| format!("create specimens dir: {error}"))?;
    let path = dir.join(format!("{merged_revision}.json"));
    let body = serde_json::to_vec_pretty(&specimen)
        .map_err(|error| format!("serialize specimen: {error}"))?;
    std::fs::write(&path, body).map_err(|error| format!("write specimen: {error}"))?;
    Ok(path)
}

/// Lands a green train: pushes `tip` to `branch` on the remote at
/// `url` WITHOUT force, so git's fast-forward rule is the race guard —
/// if the branch moved since the train was built, the push is rejected
/// and the caller should rebuild on the next round.
///
/// # Errors
///
/// Push failures, including the non-fast-forward rejection.
pub fn land(repo: &Path, url: &str, tip: &str, branch: &str) -> Result<(), String> {
    git(
        repo,
        &["push", "-q", url, &format!("{tip}:refs/heads/{branch}")],
    )
    .map(|_| ())
}

/// Reverts a landed train (D23 auto-revert arm): reverts each of the
/// train's merge commits (`base..tip`, first-parent, newest first) on
/// top of `tip`, then pushes the result to `branch` WITHOUT force — if
/// the branch moved past `tip` since landing, the push is rejected and
/// a human decides. Returns the new branch tip. The reverted tree is
/// byte-identical to `base`'s tree; history keeps the full record.
///
/// # Errors
///
/// Git failures, including a revert that itself conflicts (possible
/// when later commits touched the same lines) and the non-fast-forward
/// rejection — both leave the remote branch untouched.
pub fn revert_train(
    repo: &Path,
    url: &str,
    base: &str,
    tip: &str,
    branch: &str,
) -> Result<String, String> {
    let merges = git(
        repo,
        &[
            "rev-list",
            "--first-parent",
            "--merges",
            &format!("{base}..{tip}"),
        ],
    )?;
    let merges: Vec<&str> = merges.split_whitespace().collect();
    if merges.is_empty() {
        return Err("no train merges between base and tip".to_string());
    }
    git(repo, &["checkout", "-q", "--detach", tip])?;
    for merge in &merges {
        // -m 1 = revert to the first parent (the train spine).
        if let Err(e) = git(repo, &["revert", "-m", "1", "--no-edit", merge]) {
            git(repo, &["revert", "--abort"]).ok();
            return Err(format!(
                "revert of {merge} conflicts, leaving branch alone: {e}"
            ));
        }
    }
    let new_tip = git(repo, &["rev-parse", "HEAD"])?.trim().to_string();
    git(
        repo,
        &["push", "-q", url, &format!("{new_tip}:refs/heads/{branch}")],
    )?;
    Ok(new_tip)
}

/// One pull request's fate in a queued round.
///
/// The queue answers per change rather than per train, which is the
/// difference [`run_queue`] buys over [`build_train`]: a red build
/// blames the change that was red, and the changes ahead of it still
/// land in the same round.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PrOutcome {
    /// Merged, tested green on its own speculative state, and part of
    /// [`QueuedRound::tip`].
    Landed,
    /// Merge conflict: evicted for its author to resolve, first-class,
    /// without blocking anything behind it (D6).
    Conflicted,
    /// CI failed on this change's own speculative state.
    Failed,
    /// Ejected because it declared a dependency on a change that
    /// failed. Not a verdict on this change.
    Ejected {
        /// The failing change it depends on.
        on: u64,
    },
    /// Already in by patch identity: the train rewrote or landed this
    /// edit earlier, so it was recognized rather than re-merged.
    AlreadyLanded,
    /// A merge strategy resolved beyond what the change proposed. Git's
    /// merge cannot produce this; it is here because the map from
    /// [`choir_queue::Rejection`] must be total, and a case dropped on
    /// the floor is a PR that gets no status at all.
    Unsafe,
    /// The round stopped before reaching this change. Nothing was
    /// decided about it and it is still waiting.
    Waiting,
}

/// Result of one queued round.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueuedRound {
    /// The state the queue ended on: the commit to land, equal to the
    /// base when nothing landed.
    pub tip: String,
    /// Every pull request the round was given, in submission order.
    pub outcomes: Vec<(u64, PrOutcome)>,
    /// Why the round stopped early without blaming anybody: an
    /// executor that could not answer, or a merge we could not run.
    /// Distinct from every outcome above, because it is about us.
    pub stalled: Option<String>,
}

/// Reads one change's outcome as the status its pull request gets.
///
/// Total by construction, and asserted as a correspondence rather than
/// as a table of literals: `green` exactly where the change is in, and
/// `failure` -- the state that asks an author to do something --
/// exactly where the change itself is what went wrong.
#[must_use]
pub fn pr_report(outcome: &PrOutcome) -> TrainReport {
    match outcome {
        PrOutcome::Landed => TrainReport {
            green: true,
            state: "success",
            description: "merged and green on the speculative train",
        },
        PrOutcome::AlreadyLanded => TrainReport {
            green: true,
            state: "success",
            description: "already landed (patch identity)",
        },
        PrOutcome::Conflicted => TrainReport {
            green: false,
            state: "failure",
            description: "merge conflict; resolve and resubmit",
        },
        PrOutcome::Failed => TrainReport {
            green: false,
            state: "failure",
            description: "train CI failed on this change",
        },
        PrOutcome::Unsafe => TrainReport {
            green: false,
            state: "failure",
            description: "merge resolution edited beyond the change",
        },
        PrOutcome::Ejected { .. } => TrainReport {
            green: false,
            state: "error",
            description: "ejected: a change it depends on failed",
        },
        PrOutcome::Waiting => TrainReport {
            green: false,
            state: "error",
            description: "queue stalled before reaching this change",
        },
    }
}

/// Runs one speculative round through the D5 merge queue.
///
/// This is [`build_train`]'s successor and differs from it in what a
/// verdict is about. `build_train` merges everything and asks CI once,
/// so one red change makes every PR in the train red. The queue tests
/// each change against the state produced by everything ahead of it,
/// which is the "exactly as if they had been tested one at a time"
/// property D5 exists for: the green prefix lands, the failure is
/// blamed on the change that failed, and the window narrows.
///
/// Nothing speculative leaves the machine and no ref is moved here. The
/// caller decides whether to land [`QueuedRound::tip`].
///
/// # Errors
///
/// The base or a pull request head does not resolve, or the executor
/// speaks a protocol this build does not.
pub fn run_queue(
    repo: &Path,
    base: &str,
    prs: &[(u64, String)],
    spec: &CommandSpec,
    ci: &mut dyn CiExecutor,
) -> Result<QueuedRound, String> {
    let info = ci.info().map_err(|error| error.to_string())?;
    if info.protocol != PROTOCOL {
        return Err(format!(
            "executor `{}` speaks protocol {} and this build speaks {PROTOCOL}",
            info.name, info.protocol
        ));
    }

    let speculator = choir_queue::git::GitSpeculator::new(repo.to_path_buf());
    let base_oid = speculator.verify(base)?;
    let mut queue = choir_queue::MergeQueue::with_speculator(
        &base_oid,
        Box::new(choir_queue::git::GitSpeculator::new(repo.to_path_buf())),
    );

    let mut command = Vec::with_capacity(spec.args.len() + 1);
    command.push(spec.program.clone());
    command.extend(spec.args.iter().cloned());
    queue.set_job_template(choir_queue::JobTemplate {
        command,
        environment: effective_environment(&spec.env),
        deadline: spec.timeout_seconds.map(Duration::from_secs),
        // The train is speculative by definition: a shared cache
        // written from a state nobody approved is the CREEP shape, and
        // the queue is exactly where an attacker would reach it.
        may_write_cache: false,
    });

    for (id, head) in prs {
        let oid = speculator.verify(head)?;
        queue.submit(choir_queue::Change {
            id: *id,
            workspace: format!("pr/{id}"),
            base: base_oid.clone(),
            proposed: oid,
            depends: Vec::new(),
        });
    }

    // In memory, because the forge is canonical here and choir mirrors
    // it (D21): the landings this records are the queue's own ordering
    // of a round, not a claim about the repository, and nothing outside
    // this call reads them.
    let report = queue.drain_in_memory(ci);

    let mut outcomes: Vec<(u64, PrOutcome)> = Vec::with_capacity(prs.len());
    for (id, _) in prs {
        let outcome = if report.merged.contains(id) {
            PrOutcome::Landed
        } else if let Some((_, why)) = report.rejected.iter().find(|(r, _)| r == id) {
            match why {
                choir_queue::Rejection::Conflict => PrOutcome::Conflicted,
                choir_queue::Rejection::CiFailure => PrOutcome::Failed,
                choir_queue::Rejection::SafetyViolation { .. } => PrOutcome::Unsafe,
                choir_queue::Rejection::DependencyEjection { on } => PrOutcome::Ejected { on: *on },
                choir_queue::Rejection::AlreadyLanded => PrOutcome::AlreadyLanded,
            }
        } else {
            PrOutcome::Waiting
        };
        outcomes.push((*id, outcome));
    }

    Ok(QueuedRound {
        tip: report.final_state,
        outcomes,
        stalled: report.provider_error,
    })
}