use std::collections::{BTreeSet, HashMap};
use std::fmt;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
use anyhow::{Context as _, Result, bail};
use serde::Deserialize;
use crate::agent;
use crate::config::Config;
use crate::prompt;
use crate::land::PrLifecycle;
use crate::proc::Quiet as _;
use crate::queue::{Queue, Task, TaskStatus};
use crate::run::RunStatus;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Signal {
Branch,
Sha,
Pr,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Owner {
Task,
Run,
Pr,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Hit {
pub owner: Owner,
pub id: String,
pub status: String,
pub signal: Signal,
pub token: String,
pub via: String,
}
impl fmt::Display for Hit {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let kind = match self.owner {
Owner::Task => "task",
Owner::Run => "run",
Owner::Pr => "pull request",
};
let what = match self.signal {
Signal::Branch => "names branch",
Signal::Sha => "names commit",
Signal::Pr => "names pull request",
};
write!(
f,
"{kind} {} ({}): this work {what} {}, {}",
crate::queue::short(&self.id),
self.status,
self.token,
self.via
)
}
}
#[derive(Debug, Clone)]
pub struct Duplicate {
pub hits: Vec<Hit>,
pub judge: Option<String>,
}
impl Duplicate {
pub fn new(hits: Vec<Hit>) -> Self {
Self { hits, judge: None }
}
pub fn render(&self, override_hint: &str) -> String {
let mut out = String::from("this looks like work that is already in flight:");
for h in &self.hits {
out.push_str("\n - ");
out.push_str(&h.to_string());
}
if let Some(j) = &self.judge {
out.push('\n');
out.push_str(j);
}
out.push('\n');
out.push_str(override_hint);
out
}
}
impl fmt::Display for Duplicate {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.render(
"If it is not a duplicate, pass --force to file it anyway \
(an agent should report this to the operator instead).",
))
}
}
impl std::error::Error for Duplicate {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Ruling {
Owns,
Mentions,
Unsure,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Judgement {
pub ruling: Ruling,
pub reason: String,
pub agent: String,
}
pub type JudgeFuture = Pin<Box<dyn Future<Output = Result<Judgement>> + Send>>;
const JUDGE_BUDGET: Duration = Duration::from_secs(120);
const JUDGE_TURN: Duration = Duration::from_secs(90);
fn parse_ruling(text: &str) -> Result<(Ruling, String)> {
#[derive(Deserialize)]
struct Raw {
ruling: String,
reason: String,
}
let mut body = text.trim();
if let Some(rest) = body.strip_prefix("```") {
let rest = rest.strip_prefix("json").unwrap_or(rest);
body = rest.trim().strip_suffix("```").unwrap_or(rest).trim();
}
let raw: Raw =
serde_json::from_str(body).context("the judge's reply is not a single JSON object")?;
let ruling = match raw.ruling.trim().to_ascii_lowercase().as_str() {
"owns" => Ruling::Owns,
"mentions" => Ruling::Mentions,
"unsure" => Ruling::Unsure,
other => bail!("unknown ruling `{other}`"),
};
let reason = raw.reason.split_whitespace().collect::<Vec<_>>().join(" ");
if reason.is_empty() {
bail!("the judge gave no reason");
}
Ok((ruling, reason.chars().take(300).collect()))
}
pub async fn chain_judge(
cfg: &Config,
repo: &Path,
instruction: String,
hits: Vec<Hit>,
) -> Result<Judgement> {
let chain = agent::pick_chain(
&cfg.agents,
cfg.roles.chatter.as_ref(),
&agent::installed,
"dupes judge",
)?;
let claims: Vec<String> = hits.iter().map(ToString::to_string).collect();
let body = prompt::dupes_judge(&instruction, &claims);
let artifacts = std::env::temp_dir().join(format!("magi-dupes-{:016x}", crate::rng::entropy()));
let started = Instant::now();
let mut last: anyhow::Error = anyhow::anyhow!("no judge agent ran");
let mut result = None;
for spec in &chain {
let left = JUDGE_BUDGET.saturating_sub(started.elapsed());
if left.is_zero() {
break;
}
let mut seat = agent::SeatState::new("dupes", &spec.id, crate::rng::entropy());
let inv = agent::Invocation {
cwd: repo,
prompt: &body,
timeout: left.min(JUDGE_TURN),
allow_write: false,
sessions: false,
artifacts: &artifacts,
stem: &format!("judge-{}", spec.id),
run: "dupes",
node: "dupes",
cache_dir: None,
attachments: &[],
writable: &[],
};
let out = agent::invoke(spec, &mut seat, &inv).await;
if agent::chain_advances(&out) {
last = match out {
Err(e) => e.context(format!("judge `{}` failed", spec.id)),
Ok(o) => anyhow::anyhow!(
"judge `{}` gave no usable reply (exit {:?}, timed out {}, quota {})",
spec.id,
o.exit_code,
o.timed_out,
o.quota_exhausted()
),
};
continue;
}
result = Some(
out.and_then(|o| parse_ruling(&o.text))
.map(|(ruling, reason)| Judgement {
ruling,
reason,
agent: spec.id.clone(),
}),
);
break;
}
let _ = std::fs::remove_dir_all(&artifacts);
result.unwrap_or(Err(last))
}
pub async fn screen(
hits: Vec<Hit>,
text: &str,
review_branch: Option<&str>,
judge: &(dyn Fn(String, Vec<Hit>) -> JudgeFuture + Sync),
) -> Result<(), Duplicate> {
if hits.is_empty() {
return Ok(());
}
if text.chars().count() > prompt::DUPES_JUDGE_MAX_CHARS {
return Err(Duplicate {
hits,
judge: Some(format!(
"judge could not decide: the text is longer than {} characters, \
too long to judge in full",
prompt::DUPES_JUDGE_MAX_CHARS
)),
});
}
let mut subject = text.to_owned();
if let Some(b) = review_branch {
if !subject.is_empty() {
subject.push_str("\n\n");
}
subject.push_str(&format!(
"(This is a review-only request for branch `{b}`: it would do work on that branch.)"
));
}
let note = match judge(subject, hits.clone()).await {
Ok(j) if j.ruling == Ruling::Mentions => {
tracing::info!(
agent = %j.agent,
reason = %j.reason,
hits = hits.len(),
"duplicate check: the judge says the work only mentions what is in flight"
);
return Ok(());
}
Ok(j) => {
let word = if j.ruling == Ruling::Owns {
"owns"
} else {
"unsure"
};
format!("judge ({}): {word} - {}", j.agent, j.reason)
}
Err(e) => format!("judge could not decide: {e:#}"),
};
Err(Duplicate {
hits,
judge: Some(note),
})
}
pub async fn screen_with_config(
hits: Vec<Hit>,
text: &str,
review_branch: Option<&str>,
repo: &Path,
cfg: Option<&Config>,
) -> Result<(), Duplicate> {
let judge = |instruction: String, hits: Vec<Hit>| -> JudgeFuture {
let cfg = cfg.cloned();
let repo = repo.to_path_buf();
Box::pin(async move {
match cfg {
Some(cfg) => chain_judge(&cfg, &repo, instruction, hits).await,
None => bail!("no readable configuration to resolve a judge agent from"),
}
})
};
screen(hits, text, review_branch, &judge).await
}
#[derive(Debug, Default, Deserialize)]
struct RunView {
#[serde(default)]
id: String,
#[serde(default)]
repo: PathBuf,
#[serde(default)]
status: String,
#[serde(default)]
base_commit: String,
#[serde(default)]
candidates: Vec<CandView>,
#[serde(default)]
pr: Option<PrView>,
#[serde(default)]
released_to: Option<String>,
}
#[derive(Debug, Default, Deserialize)]
struct CandView {
#[serde(default)]
branch: String,
}
#[derive(Debug, Default, Deserialize)]
struct PrView {
#[serde(default)]
url: String,
#[serde(default)]
number: u64,
#[serde(default)]
state: String,
}
impl RunView {
fn read(runs_root: &Path, id: &str) -> Option<Self> {
let raw = std::fs::read_to_string(runs_root.join(id).join("run.json")).ok()?;
serde_json::from_str(&raw).ok()
}
fn terminal(&self) -> bool {
serde_json::from_value::<RunStatus>(serde_json::Value::String(self.status.clone()))
.map(RunStatus::done)
.unwrap_or(false)
}
fn pr_open(&self) -> bool {
self.pr.as_ref().is_some_and(|p| p.state == "open")
}
}
struct Staleness<'a> {
runs_root: &'a Path,
lookup: &'a dyn Fn(&Path, u64) -> Option<PrLifecycle>,
cache: HashMap<u64, bool>,
asked: usize,
failed: bool,
}
const MAX_FORGE_LOOKUPS: usize = 5;
impl<'a> Staleness<'a> {
fn new(runs_root: &'a Path, lookup: &'a dyn Fn(&Path, u64) -> Option<PrLifecycle>) -> Self {
Self {
runs_root,
lookup,
cache: HashMap::new(),
asked: 0,
failed: false,
}
}
fn pr_released(&mut self, repo: &Path, view: &RunView) -> bool {
let Some(pr) = view.pr.as_ref().filter(|p| p.state == "open") else {
return false;
};
if let Some(next) = &view.released_to
&& next != &view.id
&& RunView::read(self.runs_root, next).is_some()
{
return true;
}
if pr.number == 0 {
return false;
}
if let Some(known) = self.cache.get(&pr.number) {
return *known;
}
if self.failed || self.asked >= MAX_FORGE_LOOKUPS {
return false;
}
self.asked += 1;
let settled = match (self.lookup)(repo, pr.number) {
Some(PrLifecycle::Merged | PrLifecycle::Closed) => true,
Some(PrLifecycle::Open) => false,
None => {
self.failed = true;
false
}
};
self.cache.insert(pr.number, settled);
settled
}
}
#[derive(Debug, Clone)]
struct Claim {
owner: Owner,
id: String,
status: String,
via: String,
branch: Option<String>,
base: Option<String>,
pr: Option<(u64, String)>,
}
pub fn check(
queue: &Queue,
runs_root: &Path,
repo: &Path,
text: &str,
review_branch: Option<&str>,
ignore_task: Option<&str>,
) -> Vec<Hit> {
check_with(
queue,
runs_root,
repo,
text,
review_branch,
ignore_task,
&gh_open_pr,
&gh_pr_state,
)
}
#[allow(clippy::too_many_arguments)]
pub fn check_with(
queue: &Queue,
runs_root: &Path,
repo: &Path,
text: &str,
review_branch: Option<&str>,
ignore_task: Option<&str>,
open_pr: &dyn Fn(&Path, u64) -> Option<String>,
pr_state: &dyn Fn(&Path, u64) -> Option<PrLifecycle>,
) -> Vec<Hit> {
let mut stale = Staleness::new(runs_root, pr_state);
let mut idents = Idents::default();
let here = idents.of(repo);
let tasks = queue.list();
let own_runs: BTreeSet<String> = tasks
.iter()
.filter(|t| Some(t.id.as_str()) == ignore_task)
.flat_map(|t| t.runs.iter().cloned())
.collect();
let own_prs: BTreeSet<u64> = own_runs
.iter()
.filter_map(|id| RunView::read(runs_root, id))
.filter_map(|v| v.pr.map(|p| p.number))
.collect();
let mut claims: Vec<Claim> = Vec::new();
let mut from_task: BTreeSet<String> = BTreeSet::new();
for t in tasks
.iter()
.filter(|t| t.status != TaskStatus::Done && Some(t.id.as_str()) != ignore_task)
.filter(|t| idents.of(&t.repo) == here)
{
claims.extend(task_claims(t, runs_root, &mut from_task, repo, &mut stale));
}
for id in crate::run::list_ids_in(runs_root) {
if own_runs.contains(&id) {
continue;
}
let Some(view) = RunView::read(runs_root, &id) else {
continue;
};
if (view.terminal() && !view.pr_open()) || idents.of(&view.repo) != here {
continue;
}
let released = stale.pr_released(repo, &view);
if view.terminal() && released {
continue;
}
claims.extend(run_claims(
&view,
Owner::Run,
None,
"its own run",
!released,
));
}
let mut hits: Vec<Hit> = Vec::new();
let mut push = |c: &Claim, signal: Signal, token: String| {
let hit = Hit {
owner: c.owner.clone(),
id: c.id.clone(),
status: c.status.clone(),
signal,
token,
via: c.via.clone(),
};
if !hits.contains(&hit) {
hits.push(hit);
}
};
let prs = pr_numbers(text);
let shas = sha_candidates(repo, text);
for c in &claims {
if let Some(b) = &c.branch {
if names_branch(text, b) || review_branch == Some(b.as_str()) {
push(c, Signal::Branch, b.clone());
}
if let Some(base) = &c.base {
for sha in &shas {
if on_branch_only(repo, sha, b, base) {
push(c, Signal::Sha, short_sha(sha));
}
}
}
}
if let Some((n, url)) = &c.pr {
if prs.contains(&Mention::Number(*n))
|| prs.iter().any(|p| matches!(p, Mention::Url(u) if u == url))
{
push(c, Signal::Pr, format!("#{n}"));
}
}
}
let mut asked = BTreeSet::new();
for n in prs.iter().filter_map(|m| match m {
Mention::Number(n) => Some(*n),
Mention::Url(u) => u.rsplit('/').next().and_then(|d| d.parse().ok()),
}) {
let token = format!("#{n}");
if hits
.iter()
.any(|h| h.signal == Signal::Pr && h.token == token)
|| own_prs.contains(&n)
|| !asked.insert(n)
{
continue;
}
if let Some(url) = open_pr(repo, n) {
hits.push(Hit {
owner: Owner::Pr,
id: token.clone(),
status: "open".into(),
signal: Signal::Pr,
token,
via: format!("an open pull request with no run record here ({url})"),
});
}
}
hits
}
fn gh_open_pr(repo: &Path, n: u64) -> Option<String> {
let v = gh_pr_view(repo, n)?;
(v["state"] == "OPEN")
.then(|| v["url"].as_str().map(str::to_owned))
.flatten()
}
fn gh_pr_state(repo: &Path, n: u64) -> Option<PrLifecycle> {
match gh_pr_view(repo, n)?["state"].as_str()? {
"OPEN" => Some(PrLifecycle::Open),
"MERGED" => Some(PrLifecycle::Merged),
"CLOSED" => Some(PrLifecycle::Closed),
_ => None,
}
}
fn gh_pr_view(repo: &Path, n: u64) -> Option<serde_json::Value> {
let mut child = Command::new("gh")
.quiet()
.args(["pr", "view", &n.to_string(), "--json", "state,url"])
.current_dir(repo)
.env_remove("GH_REPO")
.env("GH_PROMPT_DISABLED", "1")
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.ok()?;
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
loop {
match child.try_wait().ok()? {
Some(status) if status.success() => break,
Some(_) => return None,
None if std::time::Instant::now() >= deadline => {
let _ = child.kill();
let _ = child.wait();
return None;
}
None => std::thread::sleep(std::time::Duration::from_millis(50)),
}
}
let mut raw = String::new();
std::io::Read::read_to_string(&mut child.stdout.take()?, &mut raw).ok()?;
serde_json::from_str(&raw).ok()
}
fn task_claims(
t: &Task,
runs_root: &Path,
seen: &mut BTreeSet<String>,
repo: &Path,
stale: &mut Staleness<'_>,
) -> Vec<Claim> {
let status = t.status.as_str().to_owned();
let mut out = Vec::new();
if let Some(b) = &t.review_branch {
out.push(Claim {
owner: Owner::Task,
id: t.id.clone(),
status: status.clone(),
via: "its review branch".into(),
branch: Some(b.clone()),
base: None,
pr: None,
});
}
for rid in &t.runs {
if let Some(view) = RunView::read(runs_root, rid) {
seen.insert(rid.clone());
let live_pr = !(view.terminal() && stale.pr_released(repo, &view));
out.extend(run_claims(
&view,
Owner::Task,
Some((&t.id, &status)),
&format!("produced by its run {}", crate::queue::short(rid)),
live_pr,
));
}
}
out
}
fn run_claims(
view: &RunView,
owner: Owner,
task: Option<(&str, &str)>,
via: &str,
live_pr: bool,
) -> Vec<Claim> {
let (id, status) = match task {
Some((id, status)) => (id.to_owned(), status.to_owned()),
None => (view.id.clone(), view.status.clone()),
};
let pr = view
.pr
.as_ref()
.filter(|p| live_pr && p.state == "open" && p.number > 0)
.map(|p| (p.number, p.url.clone()));
let via_pr = |extra: &str| match &pr {
Some((n, _)) => format!("{via} (PR #{n} open){extra}"),
None => format!("{via}{extra}"),
};
let mut out: Vec<Claim> = view
.candidates
.iter()
.filter(|c| !c.branch.is_empty())
.map(|c| Claim {
owner: owner.clone(),
id: id.clone(),
status: status.clone(),
via: via_pr(""),
branch: Some(c.branch.clone()),
base: (!view.base_commit.is_empty()).then(|| view.base_commit.clone()),
pr: None,
})
.collect();
if pr.is_some() {
out.push(Claim {
owner,
id,
status,
via: via_pr(""),
branch: None,
base: None,
pr,
});
}
out
}
#[derive(Default)]
struct Idents(HashMap<PathBuf, PathBuf>);
impl Idents {
fn of(&mut self, path: &Path) -> PathBuf {
self.0
.entry(path.to_path_buf())
.or_insert_with(|| {
git(
path,
&["rev-parse", "--path-format=absolute", "--git-common-dir"],
)
.map(PathBuf::from)
.and_then(|p| p.canonicalize().ok())
.or_else(|| path.canonicalize().ok())
.unwrap_or_else(|| path.to_path_buf())
})
.clone()
}
}
fn git(cwd: &Path, args: &[&str]) -> Option<String> {
let out = Command::new("git")
.quiet()
.args(args)
.current_dir(cwd)
.stdin(Stdio::null())
.stderr(Stdio::null())
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.ok()?;
out.status
.success()
.then(|| String::from_utf8_lossy(&out.stdout).trim().to_owned())
}
fn git_ok(cwd: &Path, args: &[&str]) -> bool {
git(cwd, args).is_some()
}
fn is_ref_char(c: char) -> bool {
c.is_alphanumeric() || matches!(c, '_' | '-')
}
fn names_branch(text: &str, branch: &str) -> bool {
if branch.len() < 3 {
return false;
}
text.match_indices(branch).any(|(i, _)| {
let before = text[..i].chars().next_back();
let after = text[i + branch.len()..].chars().next();
let after_ok = match after {
None => true,
Some('/') => false,
Some('.') => !text[i + branch.len() + 1..]
.chars()
.next()
.is_some_and(is_ref_char),
Some(c) => !is_ref_char(c),
};
before.is_none_or(|c| !is_ref_char(c) && c != '.') && after_ok
})
}
#[derive(Debug, PartialEq, Eq)]
enum Mention {
Number(u64),
Url(String),
}
fn pr_numbers(text: &str) -> Vec<Mention> {
let mut out = Vec::new();
let bytes = text.as_bytes();
let digits = |from: usize| -> Option<(u64, usize)> {
let n = text[from..].bytes().take_while(u8::is_ascii_digit).count();
(n > 0 && n < 10)
.then(|| text[from..from + n].parse().ok().map(|v| (v, from + n)))
.flatten()
};
for (i, _) in text.match_indices('#') {
if let Some((n, _)) = digits(i + 1) {
let word_before = i > 0 && is_ref_char(bytes[i - 1] as char);
if !word_before {
out.push(Mention::Number(n));
}
}
}
let lower = text.to_ascii_lowercase();
for key in ["pull request ", "pr "] {
for (i, _) in lower.match_indices(key) {
if i > 0 && is_ref_char(bytes[i - 1] as char) {
continue;
}
let from = i + key.len();
let from = if text[from..].starts_with('#') {
from + 1
} else {
from
};
if let Some((n, _)) = digits(from) {
out.push(Mention::Number(n));
}
}
}
for (i, _) in text.match_indices("/pull/") {
if let Some((_, end)) = digits(i + 6) {
let start = text[..i]
.rfind(|c: char| c.is_whitespace() || matches!(c, '(' | '<' | '"' | '\''))
.map_or(0, |p| p + 1);
out.push(Mention::Url(text[start..end].to_owned()));
}
}
out
}
fn sha_candidates(repo: &Path, text: &str) -> Vec<String> {
let mut seen = BTreeSet::new();
let mut out = Vec::new();
for word in text.split(|c: char| !c.is_ascii_alphanumeric()) {
if !(7..=40).contains(&word.len()) || !word.bytes().all(|b| b.is_ascii_hexdigit()) {
continue;
}
if seen.len() >= 16 || !seen.insert(word.to_ascii_lowercase()) {
continue;
}
if let Some(full) = git(
repo,
&[
"rev-parse",
"--verify",
"--quiet",
&format!("{word}^{{commit}}"),
],
) {
out.push(full);
}
}
out
}
fn on_branch_only(repo: &Path, sha: &str, branch: &str, base: &str) -> bool {
let tip = format!("{branch}^{{commit}}");
git_ok(repo, &["rev-parse", "--verify", "--quiet", &tip])
&& git_ok(repo, &["merge-base", "--is-ancestor", sha, branch])
&& !git_ok(repo, &["merge-base", "--is-ancestor", sha, base])
}
fn short_sha(sha: &str) -> String {
sha.chars().take(7).collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::queue::{Source, Task};
struct Fx {
_tmp: tempfile::TempDir,
repo: PathBuf,
runs: PathBuf,
q: Queue,
}
fn sh(cwd: &Path, args: &[&str]) -> String {
let out = Command::new("git")
.quiet()
.args(["-c", "user.name=t", "-c", "user.email=t@t"])
.args(args)
.current_dir(cwd)
.output()
.unwrap();
assert!(out.status.success(), "git {args:?}: {out:?}");
String::from_utf8_lossy(&out.stdout).trim().to_owned()
}
fn fx() -> (Fx, String, String) {
let tmp = tempfile::tempdir().unwrap();
let repo = tmp.path().join("repo");
std::fs::create_dir_all(&repo).unwrap();
sh(&repo, &["init", "-q", "-b", "main"]);
std::fs::write(repo.join("a"), "1").unwrap();
sh(&repo, &["add", "."]);
sh(&repo, &["commit", "-q", "-m", "base"]);
let base = sh(&repo, &["rev-parse", "HEAD"]);
sh(&repo, &["checkout", "-q", "-b", "magi/aaaa/A"]);
std::fs::write(repo.join("a"), "2").unwrap();
sh(&repo, &["commit", "-q", "-am", "work"]);
let tip = sh(&repo, &["rev-parse", "HEAD"]);
sh(&repo, &["checkout", "-q", "main"]);
let runs = tmp.path().join("runs");
std::fs::create_dir_all(&runs).unwrap();
let q = Queue::at(tmp.path().join("queue"));
(
Fx {
_tmp: tmp,
repo,
runs,
q,
},
base,
tip,
)
}
fn write_run(f: &Fx, id: &str, status: &str, base: &str, pr: Option<(u64, &str)>) {
let dir = f.runs.join(id);
std::fs::create_dir_all(&dir).unwrap();
let pr = pr.map(|(n, s)| {
serde_json::json!({"url": format!("https://github.com/o/r/pull/{n}"), "number": n, "state": s})
});
let v = serde_json::json!({
"schema": 999, "id": id, "repo": f.repo, "status": status,
"base_commit": base, "candidates": [{"branch": "magi/aaaa/A"}], "pr": pr,
});
std::fs::write(dir.join("run.json"), v.to_string()).unwrap();
}
fn file_task(f: &Fx, status: TaskStatus, runs: &[&str]) -> Task {
let mut t = Task::new("t".into(), "x".into(), f.repo.clone(), Source::Human);
t.status = status;
t.runs = runs.iter().map(|s| (*s).to_owned()).collect();
f.q.put(&mut t).unwrap();
t
}
const RID: &str = "20260901-100000-aaaa";
fn run(f: &Fx, text: &str, review: Option<&str>) -> Vec<Hit> {
check_with(
&f.q,
&f.runs,
&f.repo,
text,
review,
None,
&|_, _| None,
&|_, _| None,
)
}
#[test]
fn branch_matches_a_live_run_and_its_task() {
let (f, base, _) = fx();
write_run(&f, RID, "reviewing", &base, None);
let t = file_task(&f, TaskStatus::Running, &[RID]);
let hits = run(&f, "land magi/aaaa/A onto a fresh branch.", None);
assert!(hits.iter().any(|h| h.owner == Owner::Task
&& h.id == t.id
&& h.signal == Signal::Branch
&& h.token == "magi/aaaa/A"));
assert!(hits.iter().any(|h| h.owner == Owner::Run && h.id == RID));
let msg = Duplicate::new(hits).to_string();
assert!(
msg.contains("--force") && msg.contains("magi/aaaa/A"),
"{msg}"
);
}
#[test]
fn branch_must_match_whole_word() {
let (f, base, _) = fx();
write_run(&f, RID, "reviewing", &base, None);
assert!(run(&f, "see magi/aaaa/AB and magi/aaaa/A/x", None).is_empty());
}
#[test]
fn sha_on_the_branch_matches_but_one_in_base_does_not() {
let (f, base, tip) = fx();
write_run(&f, RID, "reviewing", &base, None);
let hits = run(&f, &format!("land commit {} please", &tip[..8]), None);
assert!(
hits.iter().any(|h| h.signal == Signal::Sha && h.id == RID),
"{hits:?}"
);
assert!(run(&f, &format!("see {}", &base[..9]), None).is_empty());
assert!(run(&f, "deadbeef and 1234567", None).is_empty());
}
#[test]
fn pr_number_matches_in_every_spelling() {
let (f, base, _) = fx();
write_run(&f, RID, "ready", &base, Some((48, "open")));
for text in [
"finish #48",
"PR 48 is stale",
"pr #48",
"pull request 48",
"https://github.com/o/r/pull/48",
] {
let hits = run(&f, text, None);
assert!(
hits.iter().any(|h| h.signal == Signal::Pr),
"{text}: {hits:?}"
);
}
assert!(run(&f, "see #480 and PR 4 and issue48", None).is_empty());
}
#[test]
fn terminal_runs_and_done_tasks_do_not_match() {
let (f, base, tip) = fx();
write_run(&f, RID, "merged", &base, Some((48, "merged")));
file_task(&f, TaskStatus::Done, &[RID]);
let text = format!("magi/aaaa/A {} #48", &tip[..8]);
assert!(run(&f, &text, None).is_empty());
}
#[test]
fn terminal_run_with_open_pr_or_open_task_still_claims() {
let (f, base, _) = fx();
write_run(&f, RID, "ready", &base, Some((48, "open")));
assert!(!run(&f, "magi/aaaa/A", None).is_empty());
let (g, base, _) = fx();
write_run(&g, RID, "ready", &base, None);
assert!(run(&g, "magi/aaaa/A", None).is_empty());
let t = file_task(&g, TaskStatus::Held, &[RID]);
let hits = run(&g, "magi/aaaa/A", None);
assert!(hits.iter().any(|h| h.id == t.id), "{hits:?}");
}
#[test]
fn review_only_matches_a_branch_a_live_task_owns() {
let (f, base, _) = fx();
write_run(&f, RID, "ready", &base, None);
assert!(run(&f, "", Some("magi/aaaa/A")).is_empty());
let mut t = file_task(&f, TaskStatus::Queued, &[]);
t.review_branch = Some("magi/aaaa/A".into());
f.q.put(&mut t).unwrap();
let hits = run(&f, "", Some("magi/aaaa/A"));
assert!(
hits.iter()
.any(|h| h.id == t.id && h.signal == Signal::Branch)
);
}
#[test]
fn other_repository_and_edited_task_do_not_match() {
let (f, base, _) = fx();
write_run(&f, RID, "reviewing", &base, None);
let t = file_task(&f, TaskStatus::Running, &[RID]);
let other = f._tmp.path().join("other");
std::fs::create_dir_all(&other).unwrap();
sh(&other, &["init", "-q"]);
assert!(
check_with(
&f.q,
&f.runs,
&other,
"magi/aaaa/A",
None,
None,
&|_, _| None,
&|_, _| None,
)
.is_empty()
);
assert!(
check_with(
&f.q,
&f.runs,
&f.repo,
"magi/aaaa/A",
None,
Some(&t.id),
&|_, _| None,
&|_, _| None,
)
.is_empty()
);
}
#[test]
fn a_worktree_is_the_same_repository() {
let (f, base, _) = fx();
write_run(&f, RID, "reviewing", &base, None);
let wt = f._tmp.path().join("wt");
sh(
&f.repo,
&["worktree", "add", "-q", wt.to_str().unwrap(), "-b", "other"],
);
assert!(
!check_with(
&f.q,
&f.runs,
&wt,
"magi/aaaa/A",
None,
None,
&|_, _| None,
&|_, _| None,
)
.is_empty()
);
}
#[test]
fn an_open_pr_without_a_run_record_matches_through_the_forge() {
let (f, _, _) = fx();
let open = |_: &Path, n: u64| (n == 48).then(|| "https://example.test/pull/48".to_owned());
let hit = |text: &str| {
check_with(&f.q, &f.runs, &f.repo, text, None, None, &open, &|_, _| {
None
})
};
let hits = hit("finish PR #48");
assert_eq!(hits.len(), 1, "{hits:?}");
assert_eq!(hits[0].owner, Owner::Pr);
assert!(hits[0].to_string().contains("#48"));
assert!(hit("finish PR #49").is_empty());
assert!(hit("finish the work").is_empty());
}
#[test]
fn a_forge_hit_does_not_repeat_a_pr_a_run_already_explains() {
let (f, base, _) = fx();
write_run(&f, RID, "ready", &base, Some((48, "open")));
let open = |_: &Path, _: u64| Some("u".to_owned());
let hits = check_with(&f.q, &f.runs, &f.repo, "#48", None, None, &open, &|_, _| {
None
});
assert!(hits.iter().all(|h| h.owner != Owner::Pr), "{hits:?}");
assert!(!hits.is_empty());
}
#[test]
fn an_edited_tasks_own_open_pr_is_not_a_forge_hit() {
let (f, base, _) = fx();
write_run(&f, RID, "ready", &base, Some((48, "open")));
let t = file_task(&f, TaskStatus::Running, &[RID]);
let open = |_: &Path, _: u64| Some("u".to_owned());
let hits = check_with(
&f.q,
&f.runs,
&f.repo,
"#48",
None,
Some(&t.id),
&open,
&|_, _| None,
);
assert!(hits.is_empty(), "{hits:?}");
}
fn with_forge(f: &Fx, text: &str, state: Option<PrLifecycle>) -> Vec<Hit> {
check_with(
&f.q,
&f.runs,
&f.repo,
text,
None,
None,
&|_, _| None,
&move |_, _| state,
)
}
fn release(f: &Fx, id: &str, to: &str) {
let path = f.runs.join(id).join("run.json");
let mut v: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
v["released_to"] = serde_json::json!(to);
std::fs::write(path, v.to_string()).unwrap();
}
#[test]
fn stale_open_pr_that_the_forge_says_is_merged_or_closed_does_not_claim() {
for state in [PrLifecycle::Merged, PrLifecycle::Closed] {
let (f, base, _) = fx();
write_run(&f, RID, "superseded", &base, Some((48, "open")));
assert!(with_forge(&f, "follow up on #48", Some(state)).is_empty());
assert!(with_forge(&f, "magi/aaaa/A", Some(state)).is_empty());
file_task(&f, TaskStatus::Held, &[RID]);
let hits = with_forge(&f, "follow up on #48", Some(state));
assert!(hits.iter().all(|h| h.signal != Signal::Pr), "{hits:?}");
}
}
#[test]
fn stale_open_pr_with_an_unreadable_forge_still_claims() {
let (f, base, _) = fx();
write_run(&f, RID, "superseded", &base, Some((48, "open")));
let hits = with_forge(&f, "follow up on #48", None);
assert!(hits.iter().any(|h| h.signal == Signal::Pr), "{hits:?}");
}
#[test]
fn a_genuinely_open_pr_still_claims() {
let (f, base, _) = fx();
write_run(&f, RID, "blocked", &base, Some((48, "open")));
let hits = with_forge(&f, "follow up on #48", Some(PrLifecycle::Open));
assert!(hits.iter().any(|h| h.signal == Signal::Pr), "{hits:?}");
}
#[test]
fn a_released_run_defers_to_its_successor_without_asking_the_forge() {
let (f, base, _) = fx();
let next = "20260901-110000-bbbb";
write_run(&f, RID, "superseded", &base, Some((48, "open")));
write_run(&f, next, "merged", &base, Some((48, "merged")));
release(&f, RID, next);
let asked = std::cell::Cell::new(0);
let hits = check_with(
&f.q,
&f.runs,
&f.repo,
"follow up on #48",
None,
None,
&|_, _| None,
&|_, _| {
asked.set(asked.get() + 1);
None
},
);
assert!(hits.is_empty(), "{hits:?}");
assert_eq!(asked.get(), 0);
std::fs::remove_dir_all(f.runs.join(next)).unwrap();
let hits = with_forge(&f, "follow up on #48", None);
assert!(hits.iter().any(|h| h.signal == Signal::Pr), "{hits:?}");
}
#[test]
fn forge_lookups_are_cached_and_stop_after_a_failure() {
let (f, base, _) = fx();
for (i, id) in ["20260901-100000-aaa1", "20260901-100000-aaa2"]
.iter()
.enumerate()
{
write_run(&f, id, "blocked", &base, Some((48 + i as u64, "open")));
}
let asked = std::cell::Cell::new(0);
check_with(
&f.q,
&f.runs,
&f.repo,
"x",
None,
None,
&|_, _| None,
&|_, _| {
asked.set(asked.get() + 1);
None
},
);
assert_eq!(
asked.get(),
1,
"an unreadable forge is asked once, not per PR"
);
}
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
fn verdict(ruling: Ruling) -> Result<Judgement> {
Ok(Judgement {
ruling,
reason: "because".into(),
agent: "j".into(),
})
}
fn judged(
f: &Fx,
text: &str,
answer: impl Fn() -> Result<Judgement> + Send + Sync + 'static,
) -> (Result<(), Duplicate>, usize) {
let calls = Arc::new(AtomicUsize::new(0));
let seen = calls.clone();
let judge = move |_: String, _: Vec<Hit>| -> JudgeFuture {
seen.fetch_add(1, Ordering::SeqCst);
let r = answer();
Box::pin(async move { r })
};
let hits = run(f, text, None);
let out = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap()
.block_on(screen(hits, text, None, &judge));
(out, calls.load(Ordering::SeqCst))
}
const NAMES: &str = "seen on magi/aaaa/A, which does not touch this file";
fn fx_with_run() -> Fx {
let (f, base, _) = fx();
write_run(&f, RID, "reviewing", &base, None);
f
}
#[test]
fn a_mentions_ruling_lets_the_work_through() {
let f = fx_with_run();
let (out, calls) = judged(&f, NAMES, || verdict(Ruling::Mentions));
assert!(out.is_ok());
assert_eq!(calls, 1);
}
#[test]
fn an_owns_ruling_refuses_and_says_why() {
let f = fx_with_run();
let (out, _) = judged(&f, NAMES, || verdict(Ruling::Owns));
let msg = out.unwrap_err().to_string();
assert!(
msg.contains("owns - because") && msg.contains("--force"),
"{msg}"
);
assert!(msg.contains("magi/aaaa/A"), "{msg}");
}
#[test]
fn an_unsure_ruling_refuses() {
let f = fx_with_run();
let (out, _) = judged(&f, NAMES, || verdict(Ruling::Unsure));
assert!(out.unwrap_err().to_string().contains("unsure - because"));
}
#[test]
fn a_failing_judge_refuses() {
let f = fx_with_run();
let (out, calls) = judged(&f, NAMES, || Err(anyhow::anyhow!("quota")));
let msg = out.unwrap_err().to_string();
assert!(msg.contains("judge could not decide: quota"), "{msg}");
assert_eq!(calls, 1);
}
#[test]
fn garbage_and_unknown_rulings_do_not_parse() {
assert!(parse_ruling("sure, go ahead").is_err());
assert!(parse_ruling(r#"{"ruling":"maybe","reason":"x"}"#).is_err());
assert!(parse_ruling(r#"{"ruling":"mentions"}"#).is_err());
assert!(parse_ruling(r#"{"ruling":"mentions","reason":" "}"#).is_err());
let two = "{\"ruling\":\"mentions\",\"reason\":\"c\"}\n{\"ruling\":\"owns\"}";
assert!(parse_ruling(two).is_err());
assert!(parse_ruling("ok {\"ruling\":\"mentions\",\"reason\":\"c\"}").is_err());
let (r, why) = parse_ruling("{\"ruling\":\"Mentions\",\"reason\":\"cites\\nit\"}").unwrap();
assert_eq!((r, why.as_str()), (Ruling::Mentions, "cites it"));
}
#[test]
fn a_text_too_long_to_judge_in_full_refuses_without_asking() {
let f = fx_with_run();
let long = format!("{NAMES} {}", "x".repeat(prompt::DUPES_JUDGE_MAX_CHARS));
let (out, calls) = judged(&f, &long, || verdict(Ruling::Mentions));
assert!(out.unwrap_err().to_string().contains("too long to judge"));
assert_eq!(calls, 0);
}
#[test]
fn no_hit_never_asks_the_judge() {
let (f, _, _) = fx();
let (out, calls) = judged(&f, "nothing named here", || verdict(Ruling::Owns));
assert!(out.is_ok());
assert_eq!(calls, 0);
}
#[test]
fn without_a_config_a_hit_still_refuses() {
let f = fx_with_run();
let hits = run(&f, NAMES, None);
let out = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap()
.block_on(screen_with_config(hits, NAMES, None, &f.repo, None));
assert!(out.unwrap_err().judge.is_some());
}
}