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};
pub const RECORD_DIR: &str = ".car/multiplayer";
pub const BRANCH_PREFIX: &str = "car/mp/";
pub const SCHEMA_VERSION: u32 = 1;
const COMMITTER: [&str; 6] = [
"-c",
"user.name=car-multiplayer",
"-c",
"user.email=multiplayer@parslee.ai",
"-c",
"commit.gpgSign=false",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Stage {
Build,
Improve,
Polish,
Extra,
}
impl Stage {
pub fn as_str(self) -> &'static str {
match self {
Self::Build => "build",
Self::Improve => "improve",
Self::Polish => "polish",
Self::Extra => "extra",
}
}
pub fn after(completed: usize) -> Self {
match completed {
0 => Self::Build,
1 => Self::Improve,
2 => Self::Polish,
_ => Self::Extra,
}
}
}
#[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,
pub checks_added: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StageRecord {
pub stage: Stage,
pub account_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_id: Option<String>,
pub engine: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub model: Option<String>,
pub base_commit: String,
pub result_commit: String,
pub no_change: bool,
pub contract_hash: String,
pub finished_at: u64,
pub signals: StageSignals,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cost_usd: Option<f64>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkItem {
pub schema_version: u32,
pub id: String,
pub repo_root_commit: String,
pub origin_commit: String,
pub intent: String,
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())
}
}
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)"
))
}
}
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())
}
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)
}
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(())
}
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(())
}
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);
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}")
}
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)),
],
)
}
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)
}
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}'"))
}
pub struct PublishRequest<'a> {
pub session_id: &'a str,
pub item: Option<&'a str>,
pub remote: &'a str,
pub account: &'a str,
}
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",
)?;
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}}")],
)?;
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)
}
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})"
)
})?;
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<_>>(),
}))
}
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()
}
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)
)
}
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);
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)
}
pub struct SubmitRequest<'a> {
pub repo: &'a Path,
pub item: &'a str,
pub commit: &'a str,
pub remote: &'a str,
pub account: &'a str,
pub contract_additions: Vec<super::contract::ContractCheck>,
}
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())?
}
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)),
"ready_to_merge": item.stages.len() >= 3 && distinct.len() == item.stages.len(),
})
}
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 }))
}
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 }))
}
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"));
}
}
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)
}
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())??
};
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(),
}
}
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: ¶ms.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: ¶ms.repo,
item: ¶ms.item,
commit: ¶ms.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>,
}
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) = ¶ms.engine {
start["engine"] = json!(engine);
}
if let Some(model) = ¶ms.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?;
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> {
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);
let account = current_account().await.ok();
tokio::task::spawn_blocking(move || list(¶ms.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(¶ms.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(¶ms.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,
}
}
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
}
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,
}
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,
},
)
}
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()
}
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
}
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);
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();
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}"
);
}
#[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}"
);
}
#[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}"
);
}
#[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();
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"]);
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();
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"
);
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,
}
}
let err = submit_stage(submit(&t, &id, &tip, vec![]), worktrees.path())
.await
.unwrap_err();
assert!(err.contains("must change something"), "{err}");
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}");
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}");
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);
}
#[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"
);
}
#[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}"
);
}
#[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}");
}
#[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}"
);
}
}
#[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);
}
#[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();
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}");
}
#[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}");
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);
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}");
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}");
}
}
}