car_server_core/coder/

multiplayer.rs

//! Multiplayer development: a work item moves **Build → Improve → Polish**
//! through different developers before it may merge. Design and rationale:
//! `docs/proposals/multiplayer-development.md`.
//!
//! This is a layer *on top of* coder sessions, not a change to them. Each stage
//! is an ordinary `coder.*` session, run on its owner's own machine with the
//! engine they chose. What this module adds is the hand-off:
//!
//! - A **work item** record, committed at `.car/multiplayer/<id>.json` on the
//!   branch `car/mp/<id>` in the team's shared git remote. The artifact carries
//!   the record; nothing else travels between stages. The record has no
//!   free-text field except the intent fixed at Build, and deserializes with
//!   `deny_unknown_fields`, so a hand-off cannot smuggle the previous owner's
//!   reasoning even if someone wants it to.
//! - [`publish`], which turns a finished stage session into the next commit on
//!   that branch after checking the rules the concept depends on: one stage per
//!   account, each stage starts from the previous stage's tip, the contract only
//!   grows, and a stage never edits the record itself.
//! - [`merge_check`], which re-verifies the recorded history and re-runs the
//!   final contract before producing a squash branch for the team's normal
//!   pull-request flow. CAR never pushes to `main`.
//!
//! **Eligibility is advisory.** `account_id` is the stage owner's own
//! `auth.snapshot` account, written into a git file that anyone with push
//! access could edit. The rules catch honest mistakes; making them unforgeable
//! needs server-attested stage receipts (the proposal's slice 5).

use std::collections::HashSet;
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::Arc;

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};

use super::contract::{CheckResult, OutcomeContract};
use super::session::{CoderSession, CoderState, EventSink};
use crate::handler::JsonRpcMessage;
use crate::session::{ClientSession, ServerState};

/// Where the record lives inside the tree.
pub const RECORD_DIR: &str = ".car/multiplayer";
/// Branch prefix for work items, locally and on the remote.
pub const BRANCH_PREFIX: &str = "car/mp/";
/// Wire version of [`WorkItem`]. The record rejects unknown fields (so a
/// hand-off cannot smuggle prose) and readers require an exact match, so ANY
/// field change — even an additive optional one — must bump this: a teammate on
/// an older CAR then gets a clear "schema version" error instead of an
/// unreadable item.
pub const SCHEMA_VERSION: u32 = 1;

const COMMITTER: [&str; 6] = [
    "-c",
    "user.name=car-multiplayer",
    "-c",
    "user.email=multiplayer@parslee.ai",
    // A daemon has no one to answer a pinentry prompt.
    "-c",
    "commit.gpgSign=false",
];

/// One stage of a work item.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Stage {
    Build,
    Improve,
    Polish,
    /// A further independent pass after Polish, for work whose discovery rate
    /// has not settled.
    Extra,
}

impl Stage {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Build => "build",
            Self::Improve => "improve",
            Self::Polish => "polish",
            Self::Extra => "extra",
        }
    }

    /// The stage that follows `completed` finished stages.
    pub fn after(completed: usize) -> Self {
        match completed {
            0 => Self::Build,
            1 => Self::Improve,
            2 => Self::Polish,
            _ => Self::Extra,
        }
    }
}

/// Runtime-collected measures of what a stage changed. None of it is the
/// model's self-report.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StageSignals {
    pub files_changed: u64,
    pub lines_added: u64,
    pub lines_removed: u64,
    /// Checks this stage added to the locked contract.
    pub checks_added: u64,
}

/// One completed stage.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StageRecord {
    pub stage: Stage,
    /// The stage owner's account. Advisory — see the module docs.
    pub account_id: String,
    /// The coder session that did the work, on the owner's machine. `None`
    /// for a stage submitted from outside CAR (`multiplayer.submit_stage`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// The engine that session resolved to (`native`, `external:claude-code`,
    /// …), or `external-unmanaged` for a submitted stage.
    pub engine: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    /// The commit the stage started from: the previous stage's tip, or for
    /// Build the commit its worktree was provisioned at.
    pub base_commit: String,
    /// The stage's own work commit — the approved coder branch — or
    /// `base_commit` for an accepted no-change finding.
    pub result_commit: String,
    /// The stage accepted a "no change was needed" finding instead of a diff.
    pub no_change: bool,
    /// [`contract_hash`] of the contract as this stage left it.
    pub contract_hash: String,
    pub finished_at: u64,
    pub signals: StageSignals,
    /// Metered inference spend of the stage's session, when its engine
    /// reported one. `None` is unknown, not free: the native loop does not
    /// meter, and a stage done outside CAR is never known.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cost_usd: Option<f64>,
}

/// The work item record committed at `.car/multiplayer/<id>.json`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkItem {
    pub schema_version: u32,
    pub id: String,
    /// The repository's identity across clones (same rule as `car-fleet`).
    pub repo_root_commit: String,
    /// Where Build started. The final squash is based here.
    pub origin_commit: String,
    /// Fixed at Build. The only authored text that travels between stages.
    pub intent: String,
    /// The locked contract, as grown by every stage so far.
    pub contract: OutcomeContract,
    pub stages: Vec<StageRecord>,
}

impl WorkItem {
    pub fn next_stage(&self) -> Stage {
        Stage::after(self.stages.len())
    }

    pub fn owners(&self) -> impl Iterator<Item = &str> {
        self.stages.iter().map(|s| s.account_id.as_str())
    }
}

/// The record's path inside the tree.
pub fn record_path(id: &str) -> String {
    format!("{RECORD_DIR}/{id}.json")
}

fn valid_item_id(id: &str) -> bool {
    id.len() == 19
        && id.starts_with("mp-")
        && id[3..]
            .bytes()
            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
}

fn require_item_id(id: &str) -> Result<(), String> {
    if valid_item_id(id) {
        Ok(())
    } else {
        Err(format!(
            "invalid work item id {id:?} (expected `mp-` and 16 hex digits)"
        ))
    }
}

/// A stable digest of a contract's checks and credential grant. The
/// description is prose and deliberately excluded.
pub fn contract_hash(contract: &OutcomeContract) -> String {
    let mut checks: Vec<String> = contract
        .checks
        .iter()
        .map(|c| serde_json::to_string(c).unwrap_or_default())
        .collect();
    checks.sort();
    let mut hasher = Sha256::new();
    hasher.update(if contract.allow_credentials {
        b"1"
    } else {
        b"0"
    });
    for check in checks {
        hasher.update([0u8]);
        hasher.update(check.as_bytes());
    }
    format!("{:x}", hasher.finalize())
}

/// Whether `next` only *adds* to `prior`: every prior check present and
/// unchanged, and no credential grant the prior contract did not make.
/// Returns how many checks were added.
///
/// This is what makes independent passes safe to stack: no stage can remove or
/// rewrite a check an earlier stage pinned. It pins the check COMMANDS, not the
/// files they run — a stage that weakens the test script a check invokes keeps
/// every check byte-identical, which is what the next stage's review, and the
/// merge check's run of the final contract, are for.
pub fn contract_grows(prior: &OutcomeContract, next: &OutcomeContract) -> Result<u64, String> {
    unique_names(prior)?;
    unique_names(next)?;
    if next.allow_credentials && !prior.allow_credentials {
        return Err(
            "the contract grants credential access the locked contract did not; credential \
             grants do not travel between stages"
                .into(),
        );
    }
    for check in &prior.checks {
        match next.checks.iter().find(|c| c.name == check.name) {
            None => {
                return Err(format!(
                    "the contract drops the locked check `{}`; a stage may add checks but never \
                     remove one",
                    check.name
                ));
            }
            Some(found) if found != check => {
                return Err(format!(
                    "the contract changes the locked check `{}`; a stage may add checks but \
                     never alter one",
                    check.name
                ));
            }
            Some(_) => {}
        }
    }
    Ok(next.checks.len().saturating_sub(prior.checks.len()) as u64)
}

/// The contract as the record stores it: checks and credential grant as given,
/// but the free-text `description` replaced by the item's fixed intent. The
/// description is prose a stage can rewrite; carried along, it would be a
/// channel for the previous owner's reasoning, which the record exists to shut.
fn recorded(contract: &OutcomeContract, intent: &str) -> OutcomeContract {
    OutcomeContract {
        description: intent.to_string(),
        ..contract.clone()
    }
}

fn unique_names(contract: &OutcomeContract) -> Result<(), String> {
    let mut seen = HashSet::new();
    for check in &contract.checks {
        if !seen.insert(check.name.as_str()) {
            return Err(format!("the contract names check `{}` twice", check.name));
        }
    }
    Ok(())
}

/// Whether a contract can carry a multiplayer work item. Beyond the coder's own
/// validation: every stage and the merge check re-run it cold, in a fresh
/// worktree with no session-start capture, so `baseline` and `differential`
/// checks — which compare against a before-value captured at a session's start
/// — would fail every such run and make the item unmergeable.
pub fn admissible(contract: &OutcomeContract) -> Result<(), String> {
    let issues = contract.validate();
    if !issues.is_empty() {
        return Err(format!("the contract is invalid: {}", issues.join("; ")));
    }
    unique_names(contract)?;
    if let Some(check) = contract
        .checks
        .iter()
        .find(|c| c.baseline || c.differential.is_some())
    {
        return Err(format!(
            "check `{}` is a baseline/differential check; a multiplayer contract is re-run \
             cold at every stage and at merge, where no before-value exists",
            check.name
        ));
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// git
// ---------------------------------------------------------------------------

fn git_with(
    repo: &Path,
    args: &[&str],
    env: &[(&str, &Path)],
    stdin: Option<&[u8]>,
) -> Result<String, String> {
    let mut cmd = Command::new("git");
    cmd.arg("-C").arg(repo).args(args);
    // A daemon started from a terminal must not block on a credential prompt
    // for the remote; fail instead, and say so.
    cmd.env("GIT_TERMINAL_PROMPT", "0");
    for (key, value) in env {
        cmd.env(key, value);
    }
    cmd.stdin(if stdin.is_some() {
        Stdio::piped()
    } else {
        Stdio::null()
    })
    .stdout(Stdio::piped())
    .stderr(Stdio::piped());
    let mut child = cmd.spawn().map_err(|e| format!("git {args:?}: {e}"))?;
    if let Some(bytes) = stdin {
        child
            .stdin
            .take()
            .ok_or("git stdin unavailable")?
            .write_all(bytes)
            .map_err(|e| format!("git {args:?}: {e}"))?;
    }
    let out = child
        .wait_with_output()
        .map_err(|e| format!("git {args:?}: {e}"))?;
    if out.status.success() {
        Ok(String::from_utf8_lossy(&out.stdout).trim_end().to_string())
    } else {
        Err(format!(
            "git {} failed: {}",
            args.join(" "),
            String::from_utf8_lossy(&out.stderr).trim()
        ))
    }
}

fn git(repo: &Path, args: &[&str]) -> Result<String, String> {
    git_with(repo, args, &[], None)
}

fn require_remote(repo: &Path, remote: &str) -> Result<(), String> {
    if remote.is_empty() || remote.starts_with('-') {
        return Err(format!("invalid remote name {remote:?}"));
    }
    git(repo, &["remote", "get-url", "--", remote])
        .map(|_| ())
        .map_err(|_| {
            format!(
                "{} has no remote named `{remote}`; a work item is shared through the team's \
                 git remote",
                repo.display()
            )
        })
}

fn remote_ref(remote: &str, id: &str) -> String {
    format!("refs/remotes/{remote}/{BRANCH_PREFIX}{id}")
}

/// Fetch one work item's branch and return its tip.
fn fetch_item(repo: &Path, remote: &str, id: &str) -> Result<String, String> {
    require_item_id(id)?;
    let spec = format!("+refs/heads/{BRANCH_PREFIX}{id}:{}", remote_ref(remote, id));
    git(repo, &["fetch", "--quiet", "--", remote, &spec]).map_err(|e| {
        format!("could not fetch work item {id} from `{remote}` — does it exist? ({e})")
    })?;
    git(
        repo,
        &[
            "rev-parse",
            "--verify",
            &format!("{}^{{commit}}", remote_ref(remote, id)),
        ],
    )
}

/// Read the record as committed at `commit`.
fn read_item(repo: &Path, commit: &str, id: &str) -> Result<WorkItem, String> {
    let raw = git(repo, &["show", &format!("{commit}:{}", record_path(id))]).map_err(|e| {
        if e.contains("does not exist") || e.contains("exists on disk, but not in") {
            format!("{commit} carries no record for work item {id}")
        } else {
            e
        }
    })?;
    let item: WorkItem = serde_json::from_str(&raw)
        .map_err(|e| format!("work item {id} record at {commit} is not valid: {e}"))?;
    if item.id != id {
        return Err(format!(
            "record at {commit} names work item {}, not {id}",
            item.id
        ));
    }
    if item.schema_version != SCHEMA_VERSION {
        return Err(format!(
            "work item {id} has schema version {}; this CAR reads {SCHEMA_VERSION}",
            item.schema_version
        ));
    }
    Ok(item)
}

/// Build a tree from `base`'s tree with one path set (`Some(blob)`) or removed
/// (`None`), using a private index so the user's index is never touched.
fn tree_with(repo: &Path, base: &str, path: &str, blob: Option<&str>) -> Result<String, String> {
    let index = tempfile::NamedTempFile::new().map_err(|e| format!("temp index: {e}"))?;
    let env = [("GIT_INDEX_FILE", index.path())];
    git_with(repo, &["read-tree", base], &env, None)?;
    match blob {
        Some(blob) => git_with(
            repo,
            &[
                "update-index",
                "--add",
                "--cacheinfo",
                &format!("100644,{blob},{path}"),
            ],
            &env,
            None,
        )?,
        None => git_with(
            repo,
            &["update-index", "--force-remove", "--", path],
            &env,
            None,
        )?,
    };
    git_with(repo, &["write-tree"], &env, None)
}

fn commit_tree(repo: &Path, tree: &str, parent: &str, message: &str) -> Result<String, String> {
    let mut args: Vec<&str> = COMMITTER.to_vec();
    args.extend(["commit-tree", tree, "-p", parent, "-F", "-"]);
    git_with(repo, &args, &[], Some(message.as_bytes()))
}

fn signals(repo: &Path, base: &str, result: &str, checks_added: u64) -> StageSignals {
    let mut out = StageSignals {
        checks_added,
        ..StageSignals::default()
    };
    if let Ok(numstat) = git(repo, &["diff", "--numstat", base, result]) {
        for line in numstat.lines() {
            let mut cols = line.split('\t');
            let added = cols.next().and_then(|v| v.parse::<u64>().ok());
            let removed = cols.next().and_then(|v| v.parse::<u64>().ok());
            out.files_changed += 1;
            out.lines_added += added.unwrap_or(0);
            out.lines_removed += removed.unwrap_or(0);
        }
    }
    out
}

fn load_session(state_dir: &Path, session_id: &str) -> Result<CoderSession, String> {
    if !session_id.starts_with("coder-")
        || !session_id
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'-')
    {
        return Err(format!("invalid coder session id {session_id:?}"));
    }
    CoderSession::load(&state_dir.join(format!("{session_id}.json")))
        .map_err(|_| format!("no coder session '{session_id}'"))
}

// ---------------------------------------------------------------------------
// publish
// ---------------------------------------------------------------------------

/// What to publish.
pub struct PublishRequest<'a> {
    /// A finished coder session: `merged` (approved, `car/coder/<id>`
    /// published) or `reported` (a no-change finding accepted).
    pub session_id: &'a str,
    /// `None` publishes a Build and creates the work item; `Some` publishes the
    /// item's next stage.
    pub item: Option<&'a str>,
    pub remote: &'a str,
    /// The stage owner. Advisory — see the module docs.
    pub account: &'a str,
}

/// Publish a finished stage: check the hand-off rules, commit the updated
/// record on top of the stage's work, and push `car/mp/<id>` to the remote.
///
/// Push is non-forcing, so two developers finishing the same stage race at
/// the remote and the second is refused rather than overwriting the first.
pub fn publish(state_dir: &Path, req: PublishRequest<'_>) -> Result<Value, String> {
    let session = load_session(state_dir, req.session_id)?;
    if session.project.is_some() {
        return Err(
            "a managed-project session delivers straight to its `main` and cannot be a \
             multiplayer stage; start the stage with `repo`"
                .into(),
        );
    }
    let repo = session.repo.clone();
    require_remote(&repo, req.remote)?;
    let contract = session
        .contract
        .clone()
        .ok_or("the session has no confirmed contract")?;
    let base = session
        .base
        .clone()
        .or_else(|| session.start_commit.clone())
        .ok_or(
            "the session does not record the commit it started from, so its stage cannot be \
             placed",
        )?;
    // A session started from a DIRTY checkout starts at a private snapshot
    // commit of the user's working tree — but its delivered branch commit is
    // parented on the checkout HEAD that snapshot sits on, because shipping
    // the user's uncommitted work is exactly what branch delivery refuses to
    // do. So the commit this stage is placed on, and the parent to expect
    // beneath the approved commit, is the snapshot's own parent. Comparing
    // against the snapshot instead reported "it has moved since approval" for
    // an ordinary `car code` in a dirty checkout, which had moved nothing.
    let base = match session.inputs_snapshot.as_deref() {
        Some(snapshot) => {
            git(&repo, &["rev-parse", "--verify", &format!("{snapshot}^")]).map_err(|e| {
                format!(
                    "the session's inputs snapshot {snapshot} has no parent commit to place \
                     its stage on: {e}"
                )
            })?
        }
        None => base,
    };
    let (result_commit, no_change) = match session.state {
        CoderState::Merged => {
            let branch = session
                .result_branch
                .as_deref()
                .ok_or("the merged session names no result branch")?;
            let commit = git(
                &repo,
                &["rev-parse", "--verify", &format!("{branch}^{{commit}}")],
            )?;
            // `coder.approve_merge` publishes exactly one squash commit on the
            // session's start. The branch is a movable name, so require that
            // shape: anything committed on it by hand afterwards is not the
            // reviewed work and must not ride into the item.
            let parent =
                git(&repo, &["rev-parse", "--verify", &format!("{commit}^")]).unwrap_or_default();
            if parent != base {
                return Err(format!(
                    "{branch} is not the single approved commit on the session's start \
                     {base}; it has moved since approval, so its tip is not the reviewed work"
                ));
            }
            (commit, false)
        }
        CoderState::Reported => (base.clone(), true),
        other => {
            return Err(format!(
                "publish a stage after its coder session is approved (`merged`) or its \
                 no-change finding is accepted (`reported`); {} is `{}`",
                req.session_id,
                other.as_str()
            ));
        }
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    admissible(&contract)?;
    let touched = git(
        &repo,
        &[
            "diff",
            "--name-only",
            &base,
            &result_commit,
            "--",
            RECORD_DIR,
        ],
    )?;
    if !touched.is_empty() {
        return Err(format!(
            "the stage edited {RECORD_DIR} ({}); only the runtime writes it",
            touched.lines().collect::<Vec<_>>().join(", ")
        ));
    }

    let (mut item, stage, parent) = match req.item {
        None => {
            if no_change {
                return Err(
                    "Build has to build something; a no-change finding cannot start a \
                            work item"
                        .into(),
                );
            }
            git(&repo, &["fetch", "--quiet", "--", req.remote]).map_err(|e| {
                format!(
                    "could not fetch `{}` to place the Build's origin ({e})",
                    req.remote
                )
            })?;
            let on_remote = git(
                &repo,
                &[
                    "for-each-ref",
                    "--contains",
                    &base,
                    "--format=%(refname)",
                    &format!("refs/remotes/{}/", req.remote),
                ],
            )?;
            if on_remote.trim().is_empty() {
                return Err(format!(
                    "the Build started at {base}, which is not on `{}`; the item's final \
                     squash is based there, so unpushed commits under it would reach the pull \
                     request without any stage owning them — push them first, or start the \
                     Build from a commit that is on the remote",
                    req.remote
                ));
            }
            let root = car_fleet::worker::root_commit(&repo).map_err(|e| e.to_string())?;
            let mut hasher = Sha256::new();
            for part in [&root, &base, req.account, req.session_id] {
                hasher.update(part.as_bytes());
                hasher.update([0u8]);
            }
            let id = format!("mp-{}", &format!("{:x}", hasher.finalize())[..16]);
            let existing = git(
                &repo,
                &[
                    "ls-remote",
                    "--heads",
                    "--",
                    req.remote,
                    &format!("refs/heads/{BRANCH_PREFIX}{id}"),
                ],
            )?;
            if !existing.is_empty() {
                return Err(format!("work item {id} already exists on `{}`", req.remote));
            }
            let item = WorkItem {
                schema_version: SCHEMA_VERSION,
                id,
                repo_root_commit: root,
                origin_commit: base.clone(),
                intent: session.intent.clone(),
                contract: recorded(&contract, &session.intent),
                stages: Vec::new(),
            };
            (item, Stage::Build, result_commit.clone())
        }
        Some(id) => {
            let tip = fetch_item(&repo, req.remote, id)?;
            let item = read_item(&repo, &tip, id)?;
            if item.owners().any(|owner| owner == req.account) {
                return Err(format!(
                    "account {} already owns a stage of {id}; each stage must be a different \
                     developer (advisory until stage receipts are attested)",
                    req.account
                ));
            }
            if base != tip {
                return Err(format!(
                    "a {} stage must start from the work item's tip {tip}; session {} started \
                     at {base} — start it with `coder.start {{ base: \"{tip}\" }}` or \
                     `multiplayer.start_stage`",
                    item.next_stage().as_str(),
                    req.session_id
                ));
            }
            let stage = item.next_stage();
            let parent = if no_change {
                tip.clone()
            } else {
                result_commit.clone()
            };
            (item, stage, parent)
        }
    };

    let checks_added = contract_grows(&item.contract, &contract)?;
    item.contract = recorded(&contract, &item.intent);
    item.stages.push(StageRecord {
        stage,
        account_id: req.account.to_string(),
        session_id: Some(req.session_id.to_string()),
        engine: session.engine.label(),
        model: session.model.clone(),
        base_commit: base.clone(),
        result_commit: result_commit.clone(),
        no_change,
        contract_hash: contract_hash(&contract),
        finished_at: now,
        signals: signals(&repo, &base, &result_commit, checks_added),
        cost_usd: session.cost_usd,
    });

    commit_and_push(&repo, req.remote, &item, stage, &parent)
}

/// Commit `item` (already carrying the new stage) on top of `parent` and push
/// `car/mp/<id>` without force.
fn commit_and_push(
    repo: &Path,
    remote: &str,
    item: &WorkItem,
    stage: Stage,
    parent: &str,
) -> Result<Value, String> {
    let mut body = serde_json::to_vec_pretty(item).map_err(|e| e.to_string())?;
    body.push(b'\n');
    let blob = git_with(repo, &["hash-object", "-w", "--stdin"], &[], Some(&body))?;
    let path = record_path(&item.id);
    let tree = tree_with(repo, parent, &path, Some(&blob))?;
    let message = format!(
        "multiplayer: {} of {}\n\nMultiplayer-Item: {}\nMultiplayer-Stage: {}\n",
        stage.as_str(),
        item.id,
        item.id,
        stage.as_str()
    );
    let commit = commit_tree(repo, &tree, parent, &message)?;
    let branch = format!("{BRANCH_PREFIX}{}", item.id);
    git(
        repo,
        &[
            "push",
            "--quiet",
            "--",
            remote,
            &format!("{commit}:refs/heads/{branch}"),
        ],
    )
    .map_err(|e| {
        format!(
            "could not push {branch} to `{remote}` — if another developer published this \
             stage first, the work item has moved on ({e})"
        )
    })?;
    // The push is what counts; the local branch is a convenience. Never move it
    // under a worktree that has it checked out, and never turn a successful
    // publish into an error because the convenience failed.
    let local_branch_updated = update_local_branch(repo, &branch, &commit);

    Ok(json!({
        "item_id": item.id,
        "local_branch_updated": local_branch_updated,
        "stage": stage.as_str(),
        "commit": commit,
        "branch": branch,
        "remote": remote,
        "next_stage": item.next_stage().as_str(),
        "owners": item.owners().collect::<Vec<_>>(),
    }))
}

/// Point `refs/heads/<branch>` at `commit` unless some worktree has that
/// branch checked out. Returns whether it moved.
fn update_local_branch(repo: &Path, branch: &str, commit: &str) -> bool {
    let full = format!("refs/heads/{branch}");
    let checked_out = git(repo, &["worktree", "list", "--porcelain"])
        .map(|list| list.lines().any(|l| l == format!("branch {full}")))
        .unwrap_or(true);
    !checked_out && git(repo, &["update-ref", &full, commit]).is_ok()
}

/// A worktree label no concurrent call shares: provisioning self-heals by
/// force-removing whatever sits at its path, so two calls on one label would
/// delete each other's tree mid-run.
fn unique_label(base: &str) -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    format!(
        "{base}-{}-{}",
        std::process::id(),
        N.fetch_add(1, Ordering::Relaxed)
    )
}

/// Run `contract` in a fresh worktree of `rev`. The checks were written by
/// other developers and execute on this machine — as merging or testing their
/// branch already would — so they run the way the model's own shell does:
/// without the forge credential (no `GH_*` tokens, neutralized git/gh/ssh
/// helpers), and under the inspector chain that refuses credential-shaped
/// commands whatever the contract's `allow_credentials` says. That is
/// hardening, not a sandbox: the rest of the environment is inherited.
async fn run_contract_at(
    repo: &Path,
    rev: &str,
    worktree_base: &Path,
    label: &str,
    contract: &OutcomeContract,
) -> Result<Vec<CheckResult>, String> {
    let (repo, base, rev, label) = (
        repo.to_path_buf(),
        worktree_base.to_path_buf(),
        rev.to_string(),
        unique_label(label),
    );
    let sink_label = label.clone();
    let (workspace, executor) = tokio::task::spawn_blocking(move || {
        let config = car_multi::WorkspaceConfig::git_worktree_at(&repo, &base).with_rev(rev);
        let workspace = car_multi::AgentWorkspace::provision(&config, &label)?;
        let executor = super::shell_tool::WorktreeExecutor::for_coder_session(workspace.path())?
            .withholding_forge_credentials();
        Ok::<_, String>((workspace, executor))
    })
    .await
    .map_err(|e| e.to_string())??;
    let mut contract = contract.clone();
    contract.allow_credentials = false;
    let sink = EventSink::new(sink_label, None, None);
    let results = super::contract::evaluate_contract(&contract, &executor, &sink).await;
    drop(executor);
    // Removing the worktree is blocking git too.
    let _ = tokio::task::spawn_blocking(move || drop(workspace)).await;
    Ok(results)
}

fn all_green(results: &[CheckResult]) -> bool {
    !results.is_empty() && results.iter().all(|r| r.passed)
}

/// A stage done outside CAR — a developer's own Claude Code or Codex session.
pub struct SubmitRequest<'a> {
    /// A checkout that has, or can fetch, `commit`.
    pub repo: &'a Path,
    pub item: &'a str,
    /// The stage's work: a commit descending from the item's tip.
    pub commit: &'a str,
    pub remote: &'a str,
    pub account: &'a str,
    /// Checks to add to the locked contract (never replace one).
    pub contract_additions: Vec<super::contract::ContractCheck>,
}

/// Submit a stage done outside CAR. CAR did not watch those edits, so it
/// judges only what it can verify itself: the commit descends from the tip,
/// changes something (an empty submission is refused — "no change" cannot be
/// adjudicated for work CAR did not see), leaves the record alone, and passes
/// the locked contract plus any additions, run here without credentials.
pub async fn submit_stage(req: SubmitRequest<'_>, worktree_base: &Path) -> Result<Value, String> {
    let (repo, remote, id, commit, account) = (
        req.repo.to_path_buf(),
        req.remote.to_string(),
        req.item.to_string(),
        req.commit.to_string(),
        req.account.to_string(),
    );
    let (tip, item, commit) = {
        let (repo, remote, id, account) =
            (repo.clone(), remote.clone(), id.clone(), account.clone());
        tokio::task::spawn_blocking(move || -> Result<(String, WorkItem, String), String> {
            let (tip, item) = prepare_stage(&repo, &remote, &id, &account)?;
            if commit.starts_with('-') {
                return Err(format!("invalid commit {commit:?}"));
            }
            let commit = git(
                &repo,
                &["rev-parse", "--verify", &format!("{commit}^{{commit}}")],
            )
            .map_err(|_| format!("{commit} does not name a commit in {}", repo.display()))?;
            git(&repo, &["merge-base", "--is-ancestor", &tip, &commit])
                .map_err(|_| format!("{commit} does not descend from the work item's tip {tip}"))?;
            if commit == tip {
                return Err(
                    "the submitted commit is the tip itself: a stage done outside CAR must \
                     change something, because CAR did not watch the work and cannot judge \
                     a \"no change\" conclusion"
                        .into(),
                );
            }
            let touched = git(
                &repo,
                &["diff", "--name-only", &tip, &commit, "--", RECORD_DIR],
            )?;
            if !touched.is_empty() {
                return Err(format!(
                    "the stage edited the work item record ({}); only the runtime writes it",
                    touched.lines().collect::<Vec<_>>().join(", ")
                ));
            }
            Ok((tip, item, commit))
        })
        .await
        .map_err(|e| e.to_string())??
    };

    let mut contract = item.contract.clone();
    for check in req.contract_additions {
        if contract.checks.iter().any(|c| c.name == check.name) {
            return Err(format!(
                "`contract_additions` may only add checks; `{}` is already in the locked \
                 contract",
                check.name
            ));
        }
        contract.checks.push(check);
    }
    admissible(&contract)?;
    let results = run_contract_at(
        &repo,
        &commit,
        worktree_base,
        &format!("{id}-submit"),
        &contract,
    )
    .await?;
    if !all_green(&results) {
        let red: Vec<&str> = results
            .iter()
            .filter(|r| !r.passed)
            .map(|r| r.name.as_str())
            .collect();
        return Err(format!(
            "the contract is not green at {commit} (failing: {}); fix the work and submit \
             again",
            if red.is_empty() {
                "no checks ran".to_string()
            } else {
                red.join(", ")
            }
        ));
    }

    tokio::task::spawn_blocking(move || {
        let mut item = item;
        let checks_added = contract_grows(&item.contract, &contract)?;
        let stage = item.next_stage();
        item.contract = recorded(&contract, &item.intent);
        item.stages.push(StageRecord {
            stage,
            account_id: account,
            session_id: None,
            engine: "external-unmanaged".into(),
            model: None,
            base_commit: tip.clone(),
            result_commit: commit.clone(),
            no_change: false,
            contract_hash: contract_hash(&contract),
            finished_at: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0),
            signals: signals(&repo, &tip, &commit, checks_added),
            cost_usd: None,
        });
        let mut out = commit_and_push(&repo, &remote, &item, stage, &commit)?;
        out["checks"] = json!(results);
        Ok(out)
    })
    .await
    .map_err(|e| e.to_string())?
}

// ---------------------------------------------------------------------------
// read side
// ---------------------------------------------------------------------------

/// The tip and record of one item, refusing a caller who already owns a stage.
pub fn prepare_stage(
    repo: &Path,
    remote: &str,
    id: &str,
    account: &str,
) -> Result<(String, WorkItem), String> {
    require_remote(repo, remote)?;
    let tip = fetch_item(repo, remote, id)?;
    let item = read_item(repo, &tip, id)?;
    if item.owners().any(|owner| owner == account) {
        return Err(format!(
            "account {account} already owns a stage of {id}; each stage must be a different \
             developer (advisory until stage receipts are attested)"
        ));
    }
    Ok((tip, item))
}

fn summary_row(item: &WorkItem, tip: &str, account: Option<&str>) -> Value {
    let distinct: HashSet<&str> = item.owners().collect();
    json!({
        "item_id": item.id,
        "intent": item.intent,
        "tip": tip,
        "stages": item.stages.iter().map(|s| json!({
            "stage": s.stage.as_str(),
            "account_id": s.account_id,
            "engine": s.engine,
            "no_change": s.no_change,
            "signals": s.signals,
            "cost_usd": s.cost_usd,
        })).collect::<Vec<_>>(),
        "next_stage": item.next_stage().as_str(),
        "eligible": account.map(|a| !item.owners().any(|owner| owner == a)),
        // Stage count and distinct owners only; `merge_check` is the verdict.
        "ready_to_merge": item.stages.len() >= 3 && distinct.len() == item.stages.len(),
    })
}

/// Every work item on `remote`, with the caller's eligibility for its next
/// stage when `account` is known.
pub fn list(repo: &Path, remote: &str, account: Option<&str>) -> Result<Value, String> {
    require_remote(repo, remote)?;
    let heads = git(
        repo,
        &[
            "ls-remote",
            "--heads",
            "--",
            remote,
            &format!("refs/heads/{BRANCH_PREFIX}*"),
        ],
    )?;
    let ids: Vec<String> = heads
        .lines()
        .filter_map(|line| line.split('\t').nth(1))
        .filter_map(|r| r.strip_prefix(&format!("refs/heads/{BRANCH_PREFIX}")))
        .filter(|id| valid_item_id(id))
        .map(str::to_string)
        .collect();
    let mut rows = Vec::new();
    let mut unreadable = Vec::new();
    for id in &ids {
        match fetch_item(repo, remote, id).and_then(|tip| {
            let item = read_item(repo, &tip, id)?;
            Ok(summary_row(&item, &tip, account))
        }) {
            Ok(row) => rows.push(row),
            Err(e) => unreadable.push(json!({ "item_id": id, "error": e })),
        }
    }
    Ok(json!({ "items": rows, "unreadable": unreadable }))
}

/// One work item's full record and tip.
pub fn get(repo: &Path, remote: &str, id: &str) -> Result<Value, String> {
    require_remote(repo, remote)?;
    let tip = fetch_item(repo, remote, id)?;
    let item = read_item(repo, &tip, id)?;
    Ok(json!({ "tip": tip, "item": item }))
}

// ---------------------------------------------------------------------------
// merge check
// ---------------------------------------------------------------------------

/// The history problems that make an item unmergeable, independent of the
/// contract run. Empty = the recorded history holds.
fn history_problems(repo: &Path, tip: &str, item: &WorkItem) -> Result<Vec<String>, String> {
    let mut problems = Vec::new();
    let required = [Stage::Build, Stage::Improve, Stage::Polish];
    if item.stages.len() < required.len() {
        problems.push(format!(
            "only {} of the required stages (build, improve, polish) are recorded",
            item.stages.len()
        ));
    }
    for (i, record) in item.stages.iter().enumerate() {
        if record.stage != Stage::after(i) {
            problems.push(format!(
                "stage {} is recorded as {}, expected {}",
                i + 1,
                record.stage.as_str(),
                Stage::after(i).as_str()
            ));
        }
    }
    let mut seen = HashSet::new();
    for owner in item.owners() {
        if !seen.insert(owner) {
            problems.push(format!("account {owner} owns more than one stage"));
        }
    }

    // The record must be tied to the code, not just to itself:
    //
    // - every commit that touched the record is a record commit that changed
    //   ONLY the record (a trailer is trivially forgeable; the diff is not);
    // - each version appends exactly one stage and only grows the contract;
    // - record commit i sits directly on stage i's result, and stage i started
    //   from record commit i-1 (Build: from the origin) — so every code commit
    //   on the branch is some stage's reviewed work;
    // - the tip IS the last record commit, so nothing was pushed after Polish.
    let path = record_path(&item.id);
    let log = git(
        repo,
        &["log", "--first-parent", "--format=%H", tip, "--", &path],
    )?;
    let mut commits: Vec<&str> = log.lines().collect();
    commits.reverse();
    if commits.last().copied() != Some(tip) {
        problems.push(format!(
            "the tip {tip} is not a record commit: something was pushed to the branch after \
             the last stage was published"
        ));
    }
    let mut previous: Option<(String, WorkItem)> = None;
    for (i, commit) in commits.iter().enumerate() {
        let message = git(repo, &["log", "-1", "--format=%B", commit])?;
        if !message
            .lines()
            .any(|l| l.trim() == format!("Multiplayer-Item: {}", item.id))
        {
            problems.push(format!(
                "{commit} changed the record but is not a record commit; only the runtime \
                 writes it"
            ));
            continue;
        }
        let changed = git(
            repo,
            &["diff", "--name-only", &format!("{commit}^"), commit],
        )?;
        if changed.lines().collect::<Vec<_>>() != [path.as_str()] {
            problems.push(format!(
                "record commit {commit} changes more than the record ({})",
                changed.lines().collect::<Vec<_>>().join(", ")
            ));
        }
        let version = match read_item(repo, commit, &item.id) {
            Ok(version) => version,
            Err(e) => {
                problems.push(e);
                continue;
            }
        };
        let Some(stage) = version.stages.last() else {
            problems.push(format!("record commit {commit} records no stage"));
            continue;
        };
        if version.stages.len() != i + 1 {
            problems.push(format!(
                "record commit {commit} is the #{} record commit but lists {} stage(s)",
                i + 1,
                version.stages.len()
            ));
        }
        let parent = git(repo, &["rev-parse", &format!("{commit}^")])?;
        if parent != stage.result_commit {
            problems.push(format!(
                "record commit {commit} does not sit on its stage's result {}",
                stage.result_commit
            ));
        }
        let expected_base = match &previous {
            None => version.origin_commit.clone(),
            Some((prev_commit, _)) => prev_commit.clone(),
        };
        if stage.base_commit != expected_base {
            problems.push(format!(
                "stage {} records base {} but should start from {expected_base}",
                i + 1,
                stage.base_commit
            ));
        }
        if git(
            repo,
            &[
                "merge-base",
                "--is-ancestor",
                &stage.base_commit,
                &stage.result_commit,
            ],
        )
        .is_err()
        {
            problems.push(format!(
                "stage {}'s result does not descend from its base",
                i + 1
            ));
        }
        if let Some((_, prev)) = &previous {
            if version.stages[..prev.stages.len().min(version.stages.len())]
                != prev.stages[..prev.stages.len().min(version.stages.len())]
                || version.origin_commit != prev.origin_commit
            {
                problems.push(format!("{commit} rewrites earlier stages of the record"));
            }
            if let Err(e) = contract_grows(&prev.contract, &version.contract) {
                problems.push(format!("{commit}: {e}"));
            }
        }
        previous = Some((commit.to_string(), version));
    }
    if commits.len() != item.stages.len() {
        problems.push(format!(
            "the record was written {} time(s) for {} stage(s)",
            commits.len(),
            item.stages.len()
        ));
    }
    if let Err(e) = admissible(&item.contract) {
        problems.push(e);
    }
    Ok(problems)
}

/// Re-verify a work item end to end and, if it holds, produce
/// `car/mp/<id>-final`: one squash commit on the item's origin, with the
/// record removed, for the team's normal pull-request flow.
///
/// Runs the item's final contract in a fresh worktree of the tip, with
/// credential access removed whatever the contract says. That executes the
/// repository's checks — written by other developers — on this machine, which
/// is what merging a branch already means.
pub async fn merge_check(
    repo: &Path,
    remote: &str,
    id: &str,
    worktree_base: &Path,
) -> Result<Value, String> {
    let (tip, item) = {
        let repo = repo.to_path_buf();
        let (remote, id) = (remote.to_string(), id.to_string());
        tokio::task::spawn_blocking(move || -> Result<(String, WorkItem), String> {
            require_remote(&repo, &remote)?;
            let tip = fetch_item(&repo, &remote, &id)?;
            let item = read_item(&repo, &tip, &id)?;
            Ok((tip, item))
        })
        .await
        .map_err(|e| e.to_string())??
    };
    let mut problems = {
        let (repo, tip, item) = (repo.to_path_buf(), tip.clone(), item.clone());
        tokio::task::spawn_blocking(move || history_problems(&repo, &tip, &item))
            .await
            .map_err(|e| e.to_string())??
    };

    // Build the squash that would merge — the tip's tree, minus the record, on
    // the item's origin — as a bare commit object first. The contract is then
    // run on THAT tree, not on the tip, so what gets verified is what merges.
    // A branch whose history is already known to be wrong gets neither: its
    // commands are not executed on this machine.
    let squash = if problems.is_empty() {
        let (repo, tip, id, item) = (
            repo.to_path_buf(),
            tip.clone(),
            id.to_string(),
            item.clone(),
        );
        Some(
            tokio::task::spawn_blocking(move || -> Result<String, String> {
                let tree = tree_with(&repo, &tip, &record_path(&id), None)?;
                let mut message = format!(
                    "{}\n\nMultiplayer-Item: {id}\n",
                    item.intent
                        .lines()
                        .next()
                        .unwrap_or("multiplayer work item")
                );
                for record in &item.stages {
                    message.push_str(&format!(
                        "Multiplayer-{}: {}\n",
                        capitalize(record.stage.as_str()),
                        record.account_id
                    ));
                }
                commit_tree(&repo, &tree, &item.origin_commit, &message)
            })
            .await
            .map_err(|e| e.to_string())??,
        )
    } else {
        None
    };
    let results = match &squash {
        Some(commit) => {
            run_contract_at(
                repo,
                commit,
                worktree_base,
                &format!("{id}-check"),
                &item.contract,
            )
            .await?
        }
        None => Vec::new(),
    };
    if squash.is_some() && !all_green(&results) {
        problems.push("the final contract is not green on the squash that would merge".to_string());
    }

    let (final_branch, final_commit) = match squash.filter(|_| problems.is_empty()) {
        Some(commit) => {
            let (repo, id) = (repo.to_path_buf(), id.to_string());
            tokio::task::spawn_blocking(
                move || -> Result<(Option<String>, Option<String>), String> {
                    let branch = format!("{BRANCH_PREFIX}{id}-final");
                    if !update_local_branch(&repo, &branch, &commit) {
                        return Err(format!(
                            "{branch} is checked out in a worktree; switch away from it and run \
                         the merge check again"
                        ));
                    }
                    Ok((Some(branch), Some(commit)))
                },
            )
            .await
            .map_err(|e| e.to_string())??
        }
        None => (None, None),
    };

    Ok(json!({
        "item_id": id,
        "tip": tip,
        "mergeable": problems.is_empty(),
        "problems": problems,
        "checks": results,
        "final_branch": final_branch,
        "final_commit": final_commit,
        "advisory": "stage ownership is self-reported until stage receipts are attested",
    }))
}

fn capitalize(s: &str) -> String {
    let mut chars = s.chars();
    match chars.next() {
        Some(first) => first.to_ascii_uppercase().to_string() + chars.as_str(),
        None => String::new(),
    }
}

// ---------------------------------------------------------------------------
// JSON-RPC handlers
// ---------------------------------------------------------------------------

/// Multiplayer calls push to a shared remote, run other developers' checks,
/// and attribute stages to the signed-in person. None of that is an agent's
/// to do on an operator's behalf.
async fn refuse_agent(session: &ClientSession, method: &str) -> Result<(), String> {
    if let Some(agent) = session.agent_id.lock().await.clone() {
        if !session.is_host.load(std::sync::atomic::Ordering::Acquire) {
            return Err(format!(
                "`{method}` is operator-only: `{agent}` cannot act as a developer in a \
                 multiplayer work item"
            ));
        }
    }
    Ok(())
}

async fn current_account() -> Result<String, String> {
    car_auth::local_auth_snapshot()
        .await?
        .active_account_id
        .ok_or_else(|| {
            "multiplayer stages are attributed to your Parslee account; sign in first \
             (`car auth login`)"
                .to_string()
        })
}

fn remote_param(params: &Value) -> String {
    params
        .get("remote")
        .and_then(Value::as_str)
        .filter(|r| !r.trim().is_empty())
        .unwrap_or("origin")
        .to_string()
}

#[derive(Deserialize)]
struct PublishParams {
    session_id: String,
    #[serde(default)]
    item: Option<String>,
}

pub async fn handle_publish(
    req: &JsonRpcMessage,
    session: &ClientSession,
) -> Result<Value, String> {
    refuse_agent(session, "multiplayer.publish").await?;
    let params: PublishParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let remote = remote_param(&req.params);
    let account = current_account().await?;
    let state_dir = super::rpc::coder_state_dir()?;
    tokio::task::spawn_blocking(move || {
        publish(
            &state_dir,
            PublishRequest {
                session_id: &params.session_id,
                item: params.item.as_deref(),
                remote: &remote,
                account: &account,
            },
        )
    })
    .await
    .map_err(|e| e.to_string())?
}

#[derive(Deserialize)]
struct SubmitParams {
    repo: PathBuf,
    item: String,
    commit: String,
    #[serde(default)]
    contract_additions: Vec<super::contract::ContractCheck>,
}

pub async fn handle_submit_stage(
    req: &JsonRpcMessage,
    session: &ClientSession,
) -> Result<Value, String> {
    refuse_agent(session, "multiplayer.submit_stage").await?;
    let params: SubmitParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let remote = remote_param(&req.params);
    let account = current_account().await?;
    let worktrees = super::rpc::coder_state_dir()?.join("multiplayer-worktrees");
    submit_stage(
        SubmitRequest {
            repo: &params.repo,
            item: &params.item,
            commit: &params.commit,
            remote: &remote,
            account: &account,
            contract_additions: params.contract_additions,
        },
        &worktrees,
    )
    .await
}

#[derive(Deserialize)]
struct StartStageParams {
    repo: PathBuf,
    item: String,
    #[serde(default)]
    engine: Option<String>,
    #[serde(default)]
    model: Option<String>,
}

/// Start the item's next stage as an ordinary coder session at the item's tip.
/// The reply is `coder.start`'s, plus `multiplayer.locked_contract`: confirm
/// with `coder.confirm_contract { contract: <that> }` (adding checks is fine;
/// `multiplayer.publish` refuses a contract that dropped or changed one).
pub async fn handle_start_stage(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    refuse_agent(session, "multiplayer.start_stage").await?;
    let params: StartStageParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let remote = remote_param(&req.params);
    let account = current_account().await?;
    let (tip, item) = {
        let (repo, id) = (params.repo.clone(), params.item.clone());
        tokio::task::spawn_blocking(move || prepare_stage(&repo, &remote, &id, &account))
            .await
            .map_err(|e| e.to_string())??
    };
    let mut start = json!({
        "repo": params.repo,
        "intent": item.intent,
        "base": tip,
    });
    if let Some(engine) = &params.engine {
        start["engine"] = json!(engine);
    }
    if let Some(model) = &params.model {
        start["model"] = json!(model);
    }
    let start_req = JsonRpcMessage {
        params: start,
        ..req.clone()
    };
    let mut response = super::rpc::handle_coder_start(&start_req, state, session).await?;
    // Credential grants do not travel: the next owner confirms the locked
    // checks WITHOUT the previous owner's `allow_credentials`, and grants it
    // again themselves only if they choose to (publish accepts either).
    let mut locked = item.contract.clone();
    locked.allow_credentials = false;
    response["multiplayer"] = json!({
        "item_id": item.id,
        "stage": item.next_stage().as_str(),
        "locked_contract": locked,
        "owners": item.owners().collect::<Vec<_>>(),
    });
    Ok(response)
}

#[derive(Deserialize)]
struct RepoParams {
    repo: PathBuf,
    #[serde(default)]
    item: Option<String>,
}

pub async fn handle_list(req: &JsonRpcMessage, session: &ClientSession) -> Result<Value, String> {
    // Reading fetches into the repository the caller names, which runs that
    // repository's own git config and hooks: not an agent's to point at.
    refuse_agent(session, "multiplayer.list").await?;
    let params: RepoParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let remote = remote_param(&req.params);
    // Signed out is fine for reading; eligibility is then unknown (`null`).
    let account = current_account().await.ok();
    tokio::task::spawn_blocking(move || list(&params.repo, &remote, account.as_deref()))
        .await
        .map_err(|e| e.to_string())?
}

pub async fn handle_get(req: &JsonRpcMessage, session: &ClientSession) -> Result<Value, String> {
    refuse_agent(session, "multiplayer.get").await?;
    let params: RepoParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let id = params.item.ok_or("`item` is required")?;
    let remote = remote_param(&req.params);
    tokio::task::spawn_blocking(move || get(&params.repo, &remote, &id))
        .await
        .map_err(|e| e.to_string())?
}

pub async fn handle_merge_check(
    req: &JsonRpcMessage,
    session: &ClientSession,
) -> Result<Value, String> {
    refuse_agent(session, "multiplayer.merge_check").await?;
    let params: RepoParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let id = params.item.ok_or("`item` is required")?;
    let remote = remote_param(&req.params);
    let worktrees = super::rpc::coder_state_dir()?.join("multiplayer-worktrees");
    merge_check(&params.repo, &remote, &id, &worktrees).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::coder::contract::ContractCheck;
    use crate::coder::router::EngineChoice;
    use crate::coder::test_cmds;

    fn run(dir: &Path, args: &[&str]) -> String {
        let out = Command::new("git")
            .env("GIT_CONFIG_NOSYSTEM", "1")
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .arg("-C")
            .arg(dir)
            .args([
                "-c",
                "user.name=t",
                "-c",
                "user.email=t@t",
                "-c",
                "commit.gpgSign=false",
            ])
            .args(args)
            .output()
            .unwrap();
        assert!(
            out.status.success(),
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        String::from_utf8(out.stdout).unwrap().trim().to_string()
    }

    fn check(name: &str, file: &str) -> ContractCheck {
        ContractCheck {
            name: name.into(),
            command: test_cmds::file_exists(file),
            expect_exit_zero: true,
            output_contains: None,
            timeout_secs: 30,
            baseline: false,
            differential: None,
        }
    }

    fn contract(checks: Vec<ContractCheck>) -> OutcomeContract {
        OutcomeContract {
            allow_credentials: false,
            description: "files exist".into(),
            checks,
        }
    }

    /// A commit of `file` on a fresh `car/coder/<tag>` branch at `at`, the way
    /// `coder.approve_merge` publishes one.
    fn coder_branch(repo: &Path, tag: &str, at: &str, file: &str) -> String {
        let branch = format!("car/coder/{tag}");
        run(repo, &["checkout", "-q", "-b", &branch, at]);
        std::fs::write(repo.join(file), tag).unwrap();
        run(repo, &["add", file]);
        run(repo, &["commit", "-q", "-m", tag]);
        run(repo, &["checkout", "-q", "--detach"]);
        branch
    }

    /// A finished stage session, persisted where `publish` reads it.
    fn stage_session(
        state_dir: &Path,
        repo: &Path,
        state: CoderState,
        branch: Option<&str>,
        base: Option<&str>,
        start: Option<&str>,
        contract: OutcomeContract,
    ) -> String {
        let mut s = CoderSession::new(
            repo,
            "make a.txt and b.txt exist",
            EngineChoice::Native,
            1,
            Some(state_dir.to_path_buf()),
        );
        s.state = state;
        s.result_branch = branch.map(str::to_string);
        s.base = base.map(str::to_string);
        s.start_commit = start.map(str::to_string);
        s.contract = Some(contract);
        s.persist().unwrap();
        s.id
    }

    struct Team {
        _root: tempfile::TempDir,
        state_dir: tempfile::TempDir,
        alice: PathBuf,
        bob: PathBuf,
        carol: PathBuf,
        origin: String,
    }

    /// A bare remote and three developers' clones of it.
    fn team() -> Team {
        let root = tempfile::tempdir().unwrap();
        let remote = root.path().join("remote.git");
        std::fs::create_dir_all(&remote).unwrap();
        run(&remote, &["init", "-q", "--bare", "-b", "main"]);
        let clone = |name: &str| {
            let dir = root.path().join(name);
            run(
                root.path(),
                &["clone", "-q", remote.to_str().unwrap(), name],
            );
            dir
        };
        let alice = clone("alice");
        std::fs::write(alice.join("README"), "hi").unwrap();
        run(&alice, &["add", "README"]);
        run(&alice, &["commit", "-q", "-m", "init"]);
        run(&alice, &["push", "-q", "origin", "HEAD:main"]);
        let origin = run(&alice, &["rev-parse", "HEAD"]);
        let bob = clone("bob");
        let carol = clone("carol");
        Team {
            _root: root,
            state_dir: tempfile::tempdir().unwrap(),
            alice,
            bob,
            carol,
            origin,
        }
    }

    fn publish_as(t: &Team, session: &str, item: Option<&str>, who: &str) -> Result<Value, String> {
        publish(
            t.state_dir.path(),
            PublishRequest {
                session_id: session,
                item,
                remote: "origin",
                account: who,
            },
        )
    }

    /// Alice builds; returns the item id.
    fn built(t: &Team) -> String {
        let branch = coder_branch(&t.alice, "b1", &t.origin, "a.txt");
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Merged,
            Some(&branch),
            None,
            Some(&t.origin),
            contract(vec![check("a", "a.txt")]),
        );
        let out = publish_as(t, &s, None, "alice").unwrap();
        assert_eq!(out["stage"], "build");
        assert_eq!(out["next_stage"], "improve");
        out["item_id"].as_str().unwrap().to_string()
    }

    /// Bob improves from the item's tip, adding a check.
    fn improved(t: &Team, id: &str) -> String {
        let (tip, item) = prepare_stage(&t.bob, "origin", id, "bob").unwrap();
        assert_eq!(item.next_stage(), Stage::Improve);
        let branch = coder_branch(&t.bob, "i1", &tip, "b.txt");
        let s = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&tip),
            None,
            contract(vec![check("a", "a.txt"), check("b", "b.txt")]),
        );
        publish_as(t, &s, Some(id), "bob").unwrap();
        tip
    }

    /// Build, Improve, and a no-change Polish; returns the item id.
    fn polished(t: &Team) -> String {
        let id = built(t);
        improved(t, &id);
        let (tip, _) = prepare_stage(&t.carol, "origin", &id, "carol").unwrap();
        let s = stage_session(
            t.state_dir.path(),
            &t.carol,
            CoderState::Reported,
            None,
            Some(&tip),
            None,
            contract(vec![check("a", "a.txt"), check("b", "b.txt")]),
        );
        publish_as(t, &s, Some(&id), "carol").unwrap();
        id
    }

    #[tokio::test]
    async fn a_work_item_moves_through_three_developers_and_merges() {
        let t = team();
        let id = built(&t);
        improved(&t, &id);

        // Carol polishes and finds nothing to change: an accepted finding is
        // a valid stage, and records no work commit of its own.
        let (tip, _) = prepare_stage(&t.carol, "origin", &id, "carol").unwrap();
        let s = stage_session(
            t.state_dir.path(),
            &t.carol,
            CoderState::Reported,
            None,
            Some(&tip),
            None,
            contract(vec![check("a", "a.txt"), check("b", "b.txt")]),
        );
        let out = publish_as(&t, &s, Some(&id), "carol").unwrap();
        assert_eq!(out["stage"], "polish");

        let listed = list(&t.carol, "origin", Some("dave")).unwrap();
        let row = &listed["items"][0];
        assert_eq!(row["item_id"], json!(id));
        assert_eq!(row["ready_to_merge"], true);
        assert_eq!(row["eligible"], true, "dave owns no stage");
        let listed = list(&t.carol, "origin", Some("bob")).unwrap();
        assert_eq!(listed["items"][0]["eligible"], false, "bob owns one");

        let worktrees = tempfile::tempdir().unwrap();
        let verdict = merge_check(&t.carol, "origin", &id, worktrees.path())
            .await
            .unwrap();
        assert_eq!(verdict["mergeable"], true, "{verdict}");
        let final_branch = verdict["final_branch"].as_str().unwrap();
        // One squash commit on the origin, carrying the work but not the record.
        assert_eq!(
            run(&t.carol, &["rev-parse", &format!("{final_branch}^")]),
            t.origin
        );
        let files = run(&t.carol, &["ls-tree", "-r", "--name-only", final_branch]);
        assert!(
            files.contains("a.txt") && files.contains("b.txt"),
            "{files}"
        );
        assert!(!files.contains(RECORD_DIR), "{files}");
    }

    #[tokio::test]
    async fn two_stages_are_not_mergeable() {
        let t = team();
        let id = built(&t);
        improved(&t, &id);
        let worktrees = tempfile::tempdir().unwrap();
        let verdict = merge_check(&t.carol, "origin", &id, worktrees.path())
            .await
            .unwrap();
        assert_eq!(verdict["mergeable"], false);
        assert!(verdict["final_branch"].is_null());
        assert!(
            verdict["problems"].to_string().contains("only 2"),
            "{verdict}"
        );
    }

    /// History is re-verified at merge: a record rewritten by a plain commit
    /// pushed straight to the remote — bypassing `publish` — is caught, not
    /// trusted. Here someone "adds" two stages by hand.
    #[tokio::test]
    async fn a_hand_edited_record_fails_the_merge_check() {
        let t = team();
        let id = built(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let mut item = read_item(&t.bob, &tip, &id).unwrap();
        for (stage, who) in [(Stage::Improve, "bob"), (Stage::Polish, "carol")] {
            let mut forged = item.stages[0].clone();
            forged.stage = stage;
            forged.account_id = who.into();
            item.stages.push(forged);
        }
        run(&t.bob, &["checkout", "-q", "--detach", &tip]);
        std::fs::write(
            t.bob.join(record_path(&id)),
            serde_json::to_string_pretty(&item).unwrap(),
        )
        .unwrap();
        run(&t.bob, &["commit", "-q", "-am", "totally a record commit"]);
        run(
            &t.bob,
            &[
                "push",
                "-q",
                "origin",
                &format!("HEAD:refs/heads/{BRANCH_PREFIX}{id}"),
            ],
        );

        let worktrees = tempfile::tempdir().unwrap();
        let verdict = merge_check(&t.carol, "origin", &id, worktrees.path())
            .await
            .unwrap();
        assert_eq!(verdict["mergeable"], false, "{verdict}");
        assert!(
            verdict["problems"]
                .to_string()
                .contains("not a record commit"),
            "{verdict}"
        );
    }

    /// Two developers finish the same stage; the push is non-forcing, so the
    /// second is refused instead of overwriting the first.
    #[test]
    fn the_second_publisher_of_a_stage_loses_the_race() {
        let t = team();
        let id = built(&t);
        let (tip_b, _) = prepare_stage(&t.bob, "origin", &id, "bob").unwrap();
        let (tip_c, _) = prepare_stage(&t.carol, "origin", &id, "carol").unwrap();
        assert_eq!(tip_b, tip_c);
        let mk = |repo: &Path, tag: &str, file: &str| {
            let branch = coder_branch(repo, tag, &tip_b, file);
            stage_session(
                t.state_dir.path(),
                repo,
                CoderState::Merged,
                Some(&branch),
                Some(&tip_b),
                None,
                contract(vec![check("a", "a.txt")]),
            )
        };
        let bob = mk(&t.bob, "race-b", "b.txt");
        let carol = mk(&t.carol, "race-c", "c.txt");
        publish_as(&t, &bob, Some(&id), "bob").unwrap();
        let err = publish_as(&t, &carol, Some(&id), "carol").unwrap_err();
        assert!(
            err.contains("must start from the work item's tip") || err.contains("could not push"),
            "{err}"
        );
    }

    /// A stage done in a developer's own terminal, outside CAR: CAR runs the
    /// locked contract itself, records it as unmanaged, and refuses what it
    /// cannot judge.
    /// A dirty checkout's session starts from a private snapshot of the user's
    /// working tree, and branch delivery re-parents the delivered commit onto
    /// the checkout HEAD so that work never ships. The publish gate has to
    /// expect that same parent: comparing against the snapshot refused an
    /// ordinary `car mp publish` after an ordinary dirty-checkout `car code`
    /// with "it has moved since approval", about a branch nobody had touched.
    #[test]
    fn a_dirty_checkout_stage_publishes_on_the_commit_its_work_was_parented_on() {
        let t = team();
        let id = built(&t);
        let (tip, _) = prepare_stage(&t.bob, "origin", &id, "bob").unwrap();
        // The snapshot CAR takes of the dirty checkout: the user's
        // uncommitted work, committed privately on top of the checkout HEAD.
        run(&t.bob, &["checkout", "-q", &tip]);
        std::fs::write(t.bob.join("wip.txt"), "the user's uncommitted work").unwrap();
        run(&t.bob, &["add", "wip.txt"]);
        run(&t.bob, &["commit", "-q", "-m", "car: inputs snapshot"]);
        let snapshot = run(&t.bob, &["rev-parse", "HEAD"]);
        // The delivered commit: parented on the checkout HEAD, not on the
        // snapshot, so `wip.txt` is not in it.
        let branch = coder_branch(&t.bob, "dirty", &tip, "b.txt");

        let session = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&snapshot),
            None,
            contract(vec![check("a", "a.txt")]),
        );
        let mut loaded = load_session(t.state_dir.path(), &session).unwrap();
        // `load_session` returns a session with no `state_dir`, and `persist`
        // is a no-op without one — so point it back at the store first.
        loaded.state_dir = Some(t.state_dir.path().to_path_buf());
        loaded.inputs_snapshot = Some(snapshot.clone());
        loaded.persist().unwrap();
        assert_eq!(
            load_session(t.state_dir.path(), &session)
                .unwrap()
                .inputs_snapshot
                .as_deref(),
            Some(snapshot.as_str()),
            "the fixture must actually record the snapshot it is testing"
        );

        let out = publish_as(&t, &session, Some(&id), "bob").unwrap();
        assert_eq!(out["stage"], "improve");
        let published = run(&t.bob, &["rev-parse", &format!("car/mp/{id}^")]);
        assert_eq!(
            published,
            run(&t.bob, &["rev-parse", &branch]),
            "the stage carries the reviewed commit"
        );
        assert!(
            run(&t.bob, &["log", "--format=%H", &format!("car/mp/{id}")])
                .lines()
                .all(|commit| commit != snapshot),
            "the user's snapshot must not ride into the item"
        );

        // Negative control: the same shapes with no snapshot recorded still
        // refuse, so this test cannot pass with the gate removed.
        let unsnapshotted = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&snapshot),
            None,
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &unsnapshotted, Some(&id), "bob").unwrap_err();
        assert!(err.contains("has moved since approval"), "{err}");
    }

    #[tokio::test]
    async fn a_stage_done_outside_car_is_verified_before_it_is_recorded() {
        let t = team();
        let id = built(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let worktrees = tempfile::tempdir().unwrap();
        fn submit<'a>(
            t: &'a Team,
            id: &'a str,
            commit: &'a str,
            additions: Vec<ContractCheck>,
        ) -> SubmitRequest<'a> {
            SubmitRequest {
                repo: &t.bob,
                item: id,
                commit,
                remote: "origin",
                account: "bob",
                contract_additions: additions,
            }
        }

        // Nothing changed: refused, not adjudicated.
        let err = submit_stage(submit(&t, &id, &tip, vec![]), worktrees.path())
            .await
            .unwrap_err();
        assert!(err.contains("must change something"), "{err}");

        // Red: the added check names a file the work does not create.
        let branch = coder_branch(&t.bob, "own-terminal", &tip, "b.txt");
        let head = run(&t.bob, &["rev-parse", &branch]);
        let err = submit_stage(
            submit(&t, &id, &head, vec![check("c", "c.txt")]),
            worktrees.path(),
        )
        .await
        .unwrap_err();
        assert!(err.contains("not green") && err.contains("c"), "{err}");

        // A duplicate name is a replacement, not an addition.
        let err = submit_stage(
            submit(&t, &id, &head, vec![check("a", "b.txt")]),
            worktrees.path(),
        )
        .await
        .unwrap_err();
        assert!(err.contains("may only add checks"), "{err}");

        // Green with a real addition: recorded, unmanaged, contract grown.
        let out = submit_stage(
            submit(&t, &id, &head, vec![check("b", "b.txt")]),
            worktrees.path(),
        )
        .await
        .unwrap();
        assert_eq!(out["stage"], "improve");
        let item = read_item(&t.bob, out["commit"].as_str().unwrap(), &id).unwrap();
        let stage = item.stages.last().unwrap();
        assert_eq!(stage.engine, "external-unmanaged");
        assert_eq!(stage.session_id, None);
        assert_eq!(stage.signals.checks_added, 1);
        assert_eq!(item.contract.checks.len(), 2);
    }

    /// Polish is published, then someone pushes a plain code commit to the
    /// item branch. Nothing about it touches the record, so only the rule that
    /// the tip IS the last record commit can catch it — and the squash would
    /// otherwise carry unreviewed code into the pull request.
    #[tokio::test]
    async fn code_pushed_after_the_last_stage_fails_the_merge_check() {
        let t = team();
        let id = polished(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let sneaky = coder_branch(&t.bob, "after-polish", &tip, "sneaky.txt");
        run(
            &t.bob,
            &[
                "push",
                "-q",
                "origin",
                &format!("{sneaky}:refs/heads/{BRANCH_PREFIX}{id}"),
            ],
        );
        let worktrees = tempfile::tempdir().unwrap();
        let verdict = merge_check(&t.carol, "origin", &id, worktrees.path())
            .await
            .unwrap();
        assert_eq!(verdict["mergeable"], false, "{verdict}");
        assert!(verdict["problems"]
            .to_string()
            .contains("pushed to the branch after"));
        assert!(
            verdict["checks"].as_array().unwrap().is_empty(),
            "no commands ran"
        );
    }

    /// A commit that carries a correct-looking trailer and a valid next record
    /// but ALSO changes code: the trailer is forgeable, the diff is not.
    #[tokio::test]
    async fn a_forged_record_commit_that_changes_code_fails_the_merge_check() {
        let t = team();
        let id = built(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let mut item = read_item(&t.bob, &tip, &id).unwrap();
        let mut forged = item.stages[0].clone();
        forged.stage = Stage::Improve;
        forged.account_id = "bob".into();
        forged.base_commit = tip.clone();
        forged.result_commit = tip.clone();
        item.stages.push(forged);
        run(&t.bob, &["checkout", "-q", "--detach", &tip]);
        std::fs::write(
            t.bob.join(record_path(&id)),
            serde_json::to_string_pretty(&item).unwrap(),
        )
        .unwrap();
        std::fs::write(t.bob.join("backdoor.txt"), "x").unwrap();
        run(&t.bob, &["add", "-A"]);
        run(
            &t.bob,
            &[
                "commit",
                "-q",
                "-m",
                &format!("forged\n\nMultiplayer-Item: {id}"),
            ],
        );
        run(
            &t.bob,
            &[
                "push",
                "-q",
                "origin",
                &format!("HEAD:refs/heads/{BRANCH_PREFIX}{id}"),
            ],
        );
        let worktrees = tempfile::tempdir().unwrap();
        let verdict = merge_check(&t.carol, "origin", &id, worktrees.path())
            .await
            .unwrap();
        assert_eq!(verdict["mergeable"], false);
        assert!(
            verdict["problems"]
                .to_string()
                .contains("changes more than the record"),
            "{verdict}"
        );
    }

    /// The approved branch is a movable name. A commit added on it after
    /// approval is not the reviewed work and must not publish as the stage.
    #[test]
    fn a_branch_moved_after_approval_is_refused() {
        let t = team();
        let branch = coder_branch(&t.alice, "moved", &t.origin, "a.txt");
        run(&t.alice, &["checkout", "-q", &branch]);
        std::fs::write(t.alice.join("extra.txt"), "later").unwrap();
        run(&t.alice, &["add", "extra.txt"]);
        run(&t.alice, &["commit", "-q", "-m", "after approval"]);
        run(&t.alice, &["checkout", "-q", "--detach"]);
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Merged,
            Some(&branch),
            None,
            Some(&t.origin),
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, None, "alice").unwrap_err();
        assert!(err.contains("moved since approval"), "{err}");
    }

    /// Build's origin is where the final squash lands; an origin that is not
    /// on the remote would carry unpushed commits into the PR unowned.
    #[test]
    fn a_build_from_an_unpushed_commit_is_refused() {
        let t = team();
        std::fs::write(t.alice.join("local.txt"), "unpushed").unwrap();
        run(&t.alice, &["add", "local.txt"]);
        run(&t.alice, &["commit", "-q", "-m", "local only"]);
        let local = run(&t.alice, &["rev-parse", "HEAD"]);
        let branch = coder_branch(&t.alice, "from-local", &local, "a.txt");
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Merged,
            Some(&branch),
            None,
            Some(&local),
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, None, "alice").unwrap_err();
        assert!(err.contains("not on `origin`"), "{err}");
    }

    #[test]
    fn a_contract_the_merge_check_could_never_run_is_refused() {
        let t = team();
        let branch = coder_branch(&t.alice, "baseline", &t.origin, "a.txt");
        let mut capture = check("before", "a.txt");
        capture.baseline = true;
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Merged,
            Some(&branch),
            None,
            Some(&t.origin),
            contract(vec![check("a", "a.txt"), capture]),
        );
        let err = publish_as(&t, &s, None, "alice").unwrap_err();
        assert!(err.contains("baseline/differential"), "{err}");
    }

    #[test]
    fn remote_names_are_validated() {
        let t = team();
        for bad in ["-x", "--upload-pack=touch /tmp/pwn", "nope"] {
            let err = list(&t.alice, bad, None).unwrap_err();
            assert!(
                err.contains("invalid remote") || err.contains("no remote named"),
                "{bad}: {err}"
            );
        }
    }

    /// Publishing uses a private index: whatever the developer has staged or
    /// changed in their checkout is exactly as it was afterwards.
    #[test]
    fn publish_leaves_the_developers_index_and_checkout_alone() {
        let t = team();
        let branch = coder_branch(&t.alice, "idx", &t.origin, "a.txt");
        run(&t.alice, &["checkout", "-q", "main"]);
        std::fs::write(t.alice.join("staged.txt"), "mine").unwrap();
        run(&t.alice, &["add", "staged.txt"]);
        std::fs::write(t.alice.join("README"), "edited, unstaged").unwrap();
        let before = (
            run(&t.alice, &["diff", "--cached", "--name-only"]),
            run(&t.alice, &["status", "--porcelain"]),
            run(&t.alice, &["rev-parse", "HEAD"]),
        );
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Merged,
            Some(&branch),
            None,
            Some(&t.origin),
            contract(vec![check("a", "a.txt")]),
        );
        publish_as(&t, &s, None, "alice").unwrap();
        let after = (
            run(&t.alice, &["diff", "--cached", "--name-only"]),
            run(&t.alice, &["status", "--porcelain"]),
            run(&t.alice, &["rev-parse", "HEAD"]),
        );
        assert_eq!(before, after);
    }

    /// The real race: two record commits built from the same tip. The push is
    /// non-forcing, so the second is refused at the remote.
    #[test]
    fn the_push_itself_refuses_a_second_record_from_the_same_tip() {
        let t = team();
        let id = built(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let item = read_item(&t.bob, &tip, &id).unwrap();
        // Two different stage records, as two real publishers would write.
        let with_owner = |who: &str| {
            let mut next = item.clone();
            let mut stage = next.stages[0].clone();
            stage.stage = Stage::Improve;
            stage.account_id = who.into();
            next.stages.push(stage);
            next
        };
        commit_and_push(&t.bob, "origin", &with_owner("bob"), Stage::Improve, &tip).unwrap();
        run(&t.carol, &["fetch", "-q", "origin"]);
        let err = commit_and_push(
            &t.carol,
            "origin",
            &with_owner("carol"),
            Stage::Improve,
            &tip,
        )
        .unwrap_err();
        assert!(err.contains("could not push"), "{err}");
    }

    /// The contract's prose is not a hand-off channel: the record stores the
    /// item's intent in its place, whatever the stage wrote.
    #[test]
    fn the_record_keeps_the_intent_not_the_stages_prose() {
        let t = team();
        let id = built(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let item = read_item(&t.bob, &tip, &id).unwrap();
        assert_eq!(item.contract.description, item.intent);
        assert_ne!(item.contract.description, "files exist");
    }

    #[tokio::test]
    async fn a_submission_must_descend_from_the_tip_and_leave_the_record_alone() {
        let t = team();
        let id = built(&t);
        let tip = fetch_item(&t.bob, "origin", &id).unwrap();
        let worktrees = tempfile::tempdir().unwrap();
        let off_tip = coder_branch(&t.bob, "off-tip", &t.origin, "b.txt");
        let off = run(&t.bob, &["rev-parse", &off_tip]);
        let err = submit_stage(
            SubmitRequest {
                repo: &t.bob,
                item: &id,
                commit: &off,
                remote: "origin",
                account: "bob",
                contract_additions: vec![],
            },
            worktrees.path(),
        )
        .await
        .unwrap_err();
        assert!(err.contains("does not descend"), "{err}");

        let touch = coder_branch(&t.bob, "touch-record", &tip, &record_path(&id));
        let touched = run(&t.bob, &["rev-parse", &touch]);
        let err = submit_stage(
            SubmitRequest {
                repo: &t.bob,
                item: &id,
                commit: &touched,
                remote: "origin",
                account: "bob",
                contract_additions: vec![],
            },
            worktrees.path(),
        )
        .await
        .unwrap_err();
        assert!(err.contains("only the runtime writes it"), "{err}");
    }

    #[test]
    fn the_builder_cannot_also_improve() {
        let t = team();
        let id = built(&t);
        let err = prepare_stage(&t.alice, "origin", &id, "alice").unwrap_err();
        assert!(err.contains("already owns a stage"), "{err}");
        // publish enforces it too, not only the start-time check.
        let tip = fetch_item(&t.alice, "origin", &id).unwrap();
        let branch = coder_branch(&t.alice, "i2", &tip, "b.txt");
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Merged,
            Some(&branch),
            Some(&tip),
            None,
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, Some(&id), "alice").unwrap_err();
        assert!(err.contains("already owns a stage"), "{err}");
    }

    #[test]
    fn a_stage_may_not_drop_or_change_a_locked_check() {
        let t = team();
        let id = built(&t);
        let (tip, _) = prepare_stage(&t.bob, "origin", &id, "bob").unwrap();
        let branch = coder_branch(&t.bob, "i3", &tip, "b.txt");
        let dropped = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&tip),
            None,
            contract(vec![check("b", "b.txt")]),
        );
        let err = publish_as(&t, &dropped, Some(&id), "bob").unwrap_err();
        assert!(err.contains("drops the locked check `a`"), "{err}");

        let changed = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&tip),
            None,
            contract(vec![check("a", "b.txt")]),
        );
        let err = publish_as(&t, &changed, Some(&id), "bob").unwrap_err();
        assert!(err.contains("changes the locked check `a`"), "{err}");
    }

    #[test]
    fn a_stage_must_start_from_the_tip_and_leave_the_record_alone() {
        let t = team();
        let id = built(&t);
        // Started from the origin, not the item's tip.
        let branch = coder_branch(&t.bob, "i4", &t.origin, "b.txt");
        let s = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&t.origin),
            None,
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, Some(&id), "bob").unwrap_err();
        assert!(err.contains("must start from the work item's tip"), "{err}");

        // Started correctly, but edited the record.
        let (tip, _) = prepare_stage(&t.bob, "origin", &id, "bob").unwrap();
        let record = record_path(&id);
        let branch = coder_branch(&t.bob, "i5", &tip, &record);
        let s = stage_session(
            t.state_dir.path(),
            &t.bob,
            CoderState::Merged,
            Some(&branch),
            Some(&tip),
            None,
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, Some(&id), "bob").unwrap_err();
        assert!(err.contains("only the runtime writes it"), "{err}");
    }

    #[test]
    fn build_needs_a_diff_and_a_session_that_finished() {
        let t = team();
        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::Reported,
            None,
            None,
            Some(&t.origin),
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, None, "alice").unwrap_err();
        assert!(err.contains("Build has to build something"), "{err}");

        let s = stage_session(
            t.state_dir.path(),
            &t.alice,
            CoderState::NeedsApproval,
            None,
            None,
            Some(&t.origin),
            contract(vec![check("a", "a.txt")]),
        );
        let err = publish_as(&t, &s, None, "alice").unwrap_err();
        assert!(err.contains("needs_approval"), "{err}");
    }

    #[test]
    fn a_record_with_an_unknown_field_is_refused() {
        let item = json!({
            "schema_version": 1, "id": "mp-0123456789abcdef", "repo_root_commit": "r",
            "origin_commit": "o", "intent": "i",
            "contract": { "description": "d", "checks": [] },
            "stages": [], "handoff_notes": "here is why I did it this way"
        });
        let err = serde_json::from_value::<WorkItem>(item).unwrap_err();
        assert!(err.to_string().contains("handoff_notes"), "{err}");
    }

    #[test]
    fn contract_growth_rules() {
        let a = contract(vec![check("a", "a.txt")]);
        let ab = contract(vec![check("a", "a.txt"), check("b", "b.txt")]);
        assert_eq!(contract_grows(&a, &ab), Ok(1));
        assert_eq!(contract_grows(&a, &a), Ok(0));
        assert!(contract_grows(&ab, &a).is_err());
        let mut creds = ab.clone();
        creds.allow_credentials = true;
        assert!(contract_grows(&ab, &creds)
            .unwrap_err()
            .contains("credential"));
        assert_eq!(
            contract_hash(&ab),
            contract_hash(&contract(vec![check("b", "b.txt"), check("a", "a.txt")])),
            "order-insensitive"
        );
        assert_ne!(contract_hash(&a), contract_hash(&ab));
    }

    #[test]
    fn item_ids_are_validated_before_reaching_git() {
        assert!(valid_item_id("mp-0123456789abcdef"));
        for bad in [
            "mp-0123",
            "mp-0123456789ABCDEF",
            "--upload-pack=x",
            "mp-0123456789abcdeg",
        ] {
            assert!(!valid_item_id(bad), "{bad}");
        }
    }
}