1use std::collections::{BTreeSet, HashMap};
28use std::fmt;
29use std::future::Future;
30use std::path::{Path, PathBuf};
31use std::pin::Pin;
32use std::process::{Command, Stdio};
33use std::time::{Duration, Instant};
34
35use anyhow::{Context as _, Result, bail};
36use serde::Deserialize;
37
38use crate::agent;
39use crate::config::Config;
40use crate::prompt;
41
42use crate::land::PrLifecycle;
43use crate::proc::Quiet as _;
44use crate::queue::{Queue, Task, TaskStatus};
45use crate::run::RunStatus;
46
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub enum Signal {
50 Branch,
52 Sha,
54 Pr,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
60pub enum Owner {
61 Task,
63 Run,
65 Pr,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq)]
71pub struct Hit {
72 pub owner: Owner,
74 pub id: String,
76 pub status: String,
78 pub signal: Signal,
80 pub token: String,
82 pub via: String,
84 pub about: String,
87}
88
89impl fmt::Display for Hit {
90 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
91 let kind = match self.owner {
92 Owner::Task => "task",
93 Owner::Run => "run",
94 Owner::Pr => "pull request",
95 };
96 let what = match self.signal {
97 Signal::Branch => "names branch",
98 Signal::Sha => "names commit",
99 Signal::Pr => "names pull request",
100 };
101 write!(
102 f,
103 "{kind} {} ({}): this work {what} {}, {}",
104 crate::queue::short(&self.id),
105 self.status,
106 self.token,
107 self.via
108 )
109 }
110}
111
112#[derive(Debug, Clone)]
114pub struct Duplicate {
115 pub hits: Vec<Hit>,
117 pub judge: Option<String>,
120}
121
122impl Duplicate {
123 pub fn new(hits: Vec<Hit>) -> Self {
125 Self { hits, judge: None }
126 }
127
128 pub fn render(&self, override_hint: &str) -> String {
130 let mut out = String::from("this looks like work that is already in flight:");
131 for h in &self.hits {
132 out.push_str("\n - ");
133 out.push_str(&h.to_string());
134 }
135 if let Some(j) = &self.judge {
136 out.push('\n');
137 out.push_str(j);
138 }
139 out.push('\n');
140 out.push_str(override_hint);
141 out
142 }
143}
144
145impl fmt::Display for Duplicate {
146 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
147 f.write_str(&self.render(
148 "If it is not a duplicate, pass --force to file it anyway \
149 (an agent should report this to the operator instead).",
150 ))
151 }
152}
153
154impl std::error::Error for Duplicate {}
155
156#[derive(Debug, Clone, PartialEq, Eq)]
158pub struct Judgement {
159 pub duplicate: bool,
161 pub reason: String,
163 pub agent: String,
165}
166
167#[derive(Debug, Clone, PartialEq, Eq)]
169pub enum Screened {
170 Clean,
172 Cleared(Judgement),
174 Unjudged(String),
176}
177
178pub type JudgeFuture = Pin<Box<dyn Future<Output = Result<Judgement>> + Send>>;
181
182const JUDGE_BUDGET: Duration = Duration::from_secs(45);
184const JUDGE_TURN: Duration = Duration::from_secs(30);
186
187fn parse_judgement(text: &str) -> Result<(bool, String)> {
191 #[derive(Deserialize)]
192 struct Raw {
193 duplicate: bool,
194 reason: String,
195 }
196 let mut body = text.trim();
200 if let Some(rest) = body.strip_prefix("```") {
201 let rest = rest.strip_prefix("json").unwrap_or(rest);
202 body = rest.trim().strip_suffix("```").unwrap_or(rest).trim();
203 }
204 let raw: Raw =
205 serde_json::from_str(body).context("the judge's reply is not a single JSON object")?;
206 let reason = raw.reason.split_whitespace().collect::<Vec<_>>().join(" ");
207 if reason.is_empty() {
208 bail!("the judge gave no reason");
209 }
210 Ok((raw.duplicate, reason.chars().take(300).collect()))
211}
212
213pub async fn chain_judge(
217 cfg: &Config,
218 repo: &Path,
219 instruction: String,
220 hits: Vec<Hit>,
221) -> Result<Judgement> {
222 let chain = agent::pick_chain(
223 &cfg.agents,
224 cfg.roles.chatter.as_ref(),
225 &agent::installed,
226 "dupes judge",
227 )?;
228 let claims: Vec<(String, String)> = hits
229 .iter()
230 .map(|h| (h.to_string(), h.about.clone()))
231 .collect();
232 let body = prompt::dupes_judge(&instruction, &claims);
233 let artifacts = std::env::temp_dir().join(format!("magi-dupes-{:016x}", crate::rng::entropy()));
234 let started = Instant::now();
235 let mut last: anyhow::Error = anyhow::anyhow!("no judge agent ran");
236 let mut result = None;
237 for spec in &chain {
238 let left = JUDGE_BUDGET.saturating_sub(started.elapsed());
239 if left.is_zero() {
240 break;
241 }
242 let mut seat = agent::SeatState::new("dupes", &spec.id, crate::rng::entropy());
243 let inv = agent::Invocation {
244 cwd: repo,
245 prompt: &body,
246 timeout: left.min(JUDGE_TURN),
247 allow_write: false,
248 sessions: false,
249 artifacts: &artifacts,
250 stem: &format!("judge-{}", spec.id),
251 run: "dupes",
252 node: "dupes",
253 cache_dir: None,
254 attachments: &[],
255 writable: &[],
256 };
257 let out = agent::invoke(spec, &mut seat, &inv).await;
258 if agent::chain_advances(&out) {
259 last = match out {
260 Err(e) => e.context(format!("judge `{}` failed", spec.id)),
261 Ok(o) => anyhow::anyhow!(
262 "judge `{}` gave no usable reply (exit {:?}, timed out {}, quota {})",
263 spec.id,
264 o.exit_code,
265 o.timed_out,
266 o.quota_exhausted()
267 ),
268 };
269 continue;
270 }
271 result = Some(
272 out.and_then(|o| parse_judgement(&o.text))
273 .map(|(duplicate, reason)| Judgement {
274 duplicate,
275 reason,
276 agent: spec.id.clone(),
277 }),
278 );
279 break;
280 }
281 let _ = std::fs::remove_dir_all(&artifacts);
282 result.unwrap_or(Err(last))
283}
284
285pub async fn screen(
300 hits: Vec<Hit>,
301 text: &str,
302 review_branch: Option<&str>,
303 judge: &(dyn Fn(String, Vec<Hit>) -> JudgeFuture + Sync),
304) -> Result<Screened, Duplicate> {
305 if hits.is_empty() {
306 return Ok(Screened::Clean);
307 }
308 let unjudged = |why: String| {
309 tracing::warn!(hits = hits.len(), %why, "duplicate check: judgement unavailable, letting the work through");
310 Ok(Screened::Unjudged(why))
311 };
312 let mut subject = squeeze(text, prompt::DUPES_JUDGE_MAX_CHARS);
313 if let Some(b) = review_branch {
314 if !subject.is_empty() {
315 subject.push_str("\n\n");
316 }
317 subject.push_str(&format!(
318 "(This is a review-only request for branch `{b}`: it would do work on that branch.)"
319 ));
320 }
321 match judge(subject, hits.clone()).await {
322 Ok(j) if !j.duplicate => {
323 tracing::info!(
324 agent = %j.agent,
325 reason = %j.reason,
326 hits = hits.len(),
327 "duplicate check: the judge says this is not duplicate work"
328 );
329 Ok(Screened::Cleared(j))
330 }
331 Ok(j) => Err(Duplicate {
332 hits,
333 judge: Some(format!("judge ({}): duplicate - {}", j.agent, j.reason)),
334 }),
335 Err(e) => unjudged(format!("{e:#}")),
336 }
337}
338
339pub async fn screen_with_config(
342 hits: Vec<Hit>,
343 text: &str,
344 review_branch: Option<&str>,
345 repo: &Path,
346 cfg: Option<&Config>,
347) -> Result<Screened, Duplicate> {
348 let judge = |instruction: String, hits: Vec<Hit>| -> JudgeFuture {
349 let cfg = cfg.cloned();
350 let repo = repo.to_path_buf();
351 Box::pin(async move {
352 match cfg {
353 Some(cfg) => {
354 let dir = repo.clone();
355 let hits = tokio::task::spawn_blocking(move || {
356 let mut hits = hits;
357 describe_forge_hits(&dir, &mut hits);
358 hits
359 })
360 .await
361 .context("describing the pull request")?;
362 chain_judge(&cfg, &repo, instruction, hits).await
363 }
364 None => bail!("no readable configuration to resolve a judge agent from"),
365 }
366 })
367 };
368 screen(hits, text, review_branch, &judge).await
369}
370
371#[derive(Debug, Default, Deserialize)]
374struct RunView {
375 #[serde(default)]
376 id: String,
377 #[serde(default)]
378 repo: PathBuf,
379 #[serde(default)]
380 status: String,
381 #[serde(default)]
382 base_commit: String,
383 #[serde(default)]
384 instruction: String,
385 #[serde(default)]
386 candidates: Vec<CandView>,
387 #[serde(default)]
388 pr: Option<PrView>,
389 #[serde(default)]
391 released_to: Option<String>,
392}
393
394#[derive(Debug, Default, Deserialize)]
395struct CandView {
396 #[serde(default)]
397 branch: String,
398}
399
400#[derive(Debug, Default, Deserialize)]
401struct PrView {
402 #[serde(default)]
403 url: String,
404 #[serde(default)]
405 number: u64,
406 #[serde(default)]
407 state: String,
408}
409
410impl RunView {
411 fn read(runs_root: &Path, id: &str) -> Option<Self> {
412 let raw = std::fs::read_to_string(runs_root.join(id).join("run.json")).ok()?;
413 serde_json::from_str(&raw).ok()
414 }
415
416 fn terminal(&self) -> bool {
419 serde_json::from_value::<RunStatus>(serde_json::Value::String(self.status.clone()))
420 .map(RunStatus::done)
421 .unwrap_or(false)
422 }
423
424 fn pr_open(&self) -> bool {
425 self.pr.as_ref().is_some_and(|p| p.state == "open")
426 }
427}
428
429struct Staleness<'a> {
446 runs_root: &'a Path,
447 lookup: &'a dyn Fn(&Path, u64) -> Option<PrLifecycle>,
448 cache: HashMap<u64, bool>,
449 asked: usize,
450 failed: bool,
451}
452
453const MAX_FORGE_LOOKUPS: usize = 5;
456
457impl<'a> Staleness<'a> {
458 fn new(runs_root: &'a Path, lookup: &'a dyn Fn(&Path, u64) -> Option<PrLifecycle>) -> Self {
459 Self {
460 runs_root,
461 lookup,
462 cache: HashMap::new(),
463 asked: 0,
464 failed: false,
465 }
466 }
467
468 fn pr_released(&mut self, repo: &Path, view: &RunView) -> bool {
471 let Some(pr) = view.pr.as_ref().filter(|p| p.state == "open") else {
472 return false;
473 };
474 if let Some(next) = &view.released_to
475 && next != &view.id
476 && RunView::read(self.runs_root, next).is_some()
477 {
478 return true;
479 }
480 if pr.number == 0 {
481 return false;
482 }
483 if let Some(known) = self.cache.get(&pr.number) {
484 return *known;
485 }
486 if self.failed || self.asked >= MAX_FORGE_LOOKUPS {
487 return false;
488 }
489 self.asked += 1;
490 let settled = match (self.lookup)(repo, pr.number) {
491 Some(PrLifecycle::Merged | PrLifecycle::Closed) => true,
492 Some(PrLifecycle::Open) => false,
493 None => {
494 self.failed = true;
495 false
496 }
497 };
498 self.cache.insert(pr.number, settled);
499 settled
500 }
501}
502
503#[derive(Debug, Clone)]
505struct Claim {
506 owner: Owner,
507 id: String,
508 status: String,
509 via: String,
510 about: String,
512 branch: Option<String>,
513 base: Option<String>,
515 pr: Option<(u64, String)>,
517}
518
519pub fn check(
523 queue: &Queue,
524 runs_root: &Path,
525 repo: &Path,
526 text: &str,
527 review_branch: Option<&str>,
528 ignore_task: Option<&str>,
529) -> Vec<Hit> {
530 check_with(
531 queue,
532 runs_root,
533 repo,
534 text,
535 review_branch,
536 ignore_task,
537 &gh_open_pr,
538 &gh_pr_state,
539 )
540}
541
542#[allow(clippy::too_many_arguments)]
549pub fn check_with(
550 queue: &Queue,
551 runs_root: &Path,
552 repo: &Path,
553 text: &str,
554 review_branch: Option<&str>,
555 ignore_task: Option<&str>,
556 open_pr: &dyn Fn(&Path, u64) -> Option<String>,
557 pr_state: &dyn Fn(&Path, u64) -> Option<PrLifecycle>,
558) -> Vec<Hit> {
559 let mut stale = Staleness::new(runs_root, pr_state);
560 let mut idents = Idents::default();
561 let here = idents.of(repo);
562 let tasks = queue.list();
563 let own_runs: BTreeSet<String> = tasks
564 .iter()
565 .filter(|t| Some(t.id.as_str()) == ignore_task)
566 .flat_map(|t| t.runs.iter().cloned())
567 .collect();
568 let own_prs: BTreeSet<u64> = own_runs
570 .iter()
571 .filter_map(|id| RunView::read(runs_root, id))
572 .filter_map(|v| v.pr.map(|p| p.number))
573 .collect();
574
575 let mut claims: Vec<Claim> = Vec::new();
576 let mut from_task: BTreeSet<String> = BTreeSet::new();
577 for t in tasks
578 .iter()
579 .filter(|t| t.status != TaskStatus::Done && Some(t.id.as_str()) != ignore_task)
580 .filter(|t| idents.of(&t.repo) == here)
581 {
582 claims.extend(task_claims(t, runs_root, &mut from_task, repo, &mut stale));
583 }
584 for id in crate::run::list_ids_in(runs_root) {
585 if own_runs.contains(&id) {
586 continue;
587 }
588 let Some(view) = RunView::read(runs_root, &id) else {
589 continue;
590 };
591 if (view.terminal() && !view.pr_open()) || idents.of(&view.repo) != here {
592 continue;
593 }
594 let released = stale.pr_released(repo, &view);
595 if view.terminal() && released {
598 continue;
599 }
600 claims.extend(run_claims(
601 &view,
602 Owner::Run,
603 None,
604 "its own run",
605 !released,
606 ("", &view.instruction),
607 ));
608 }
609
610 let mut hits: Vec<Hit> = Vec::new();
611 let mut push = |c: &Claim, signal: Signal, token: String| {
612 let hit = Hit {
613 owner: c.owner.clone(),
614 id: c.id.clone(),
615 status: c.status.clone(),
616 signal,
617 token,
618 via: c.via.clone(),
619 about: c.about.clone(),
620 };
621 if !hits.contains(&hit) {
622 hits.push(hit);
623 }
624 };
625
626 let prs = pr_numbers(text);
627 let shas = sha_candidates(repo, text);
628 for c in &claims {
629 if let Some(b) = &c.branch {
630 if names_branch(text, b) || review_branch == Some(b.as_str()) {
631 push(c, Signal::Branch, b.clone());
632 }
633 if let Some(base) = &c.base {
634 for sha in &shas {
635 if on_branch_only(repo, sha, b, base) {
636 push(c, Signal::Sha, short_sha(sha));
637 }
638 }
639 }
640 }
641 if let Some((n, url)) = &c.pr {
642 if prs.contains(&Mention::Number(*n))
643 || prs.iter().any(|p| matches!(p, Mention::Url(u) if u == url))
644 {
645 push(c, Signal::Pr, format!("#{n}"));
646 }
647 }
648 }
649 let mut asked = BTreeSet::new();
650 for n in prs.iter().filter_map(|m| match m {
651 Mention::Number(n) => Some(*n),
652 Mention::Url(u) => u.rsplit('/').next().and_then(|d| d.parse().ok()),
653 }) {
654 let token = format!("#{n}");
655 if hits
656 .iter()
657 .any(|h| h.signal == Signal::Pr && h.token == token)
658 || own_prs.contains(&n)
659 || !asked.insert(n)
660 {
661 continue;
662 }
663 if let Some(url) = open_pr(repo, n) {
664 hits.push(Hit {
665 owner: Owner::Pr,
666 id: token.clone(),
667 status: "open".into(),
668 signal: Signal::Pr,
669 token,
670 via: format!("an open pull request with no run record here ({url})"),
671 about: String::new(),
672 });
673 }
674 }
675 hits
676}
677
678fn gh_open_pr(repo: &Path, n: u64) -> Option<String> {
683 let v = gh_pr_view(repo, n, "state,url")?;
684 (v["state"] == "OPEN")
685 .then(|| v["url"].as_str().map(str::to_owned))
686 .flatten()
687}
688
689fn gh_pr_state(repo: &Path, n: u64) -> Option<PrLifecycle> {
692 match gh_pr_view(repo, n, "state,url")?["state"].as_str()? {
693 "OPEN" => Some(PrLifecycle::Open),
694 "MERGED" => Some(PrLifecycle::Merged),
695 "CLOSED" => Some(PrLifecycle::Closed),
696 _ => None,
697 }
698}
699
700fn gh_pr_view(repo: &Path, n: u64, fields: &str) -> Option<serde_json::Value> {
701 let mut child = Command::new("gh")
702 .quiet()
703 .args(["pr", "view", &n.to_string(), "--json", fields])
704 .current_dir(repo)
705 .env_remove("GH_REPO")
706 .env("GH_PROMPT_DISABLED", "1")
707 .stdin(Stdio::null())
708 .stdout(Stdio::piped())
709 .stderr(Stdio::null())
710 .spawn()
711 .ok()?;
712 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
713 loop {
714 match child.try_wait().ok()? {
715 Some(status) if status.success() => break,
716 Some(_) => return None,
717 None if std::time::Instant::now() >= deadline => {
718 let _ = child.kill();
719 let _ = child.wait();
720 return None;
721 }
722 None => std::thread::sleep(std::time::Duration::from_millis(50)),
723 }
724 }
725 let mut raw = String::new();
726 std::io::Read::read_to_string(&mut child.stdout.take()?, &mut raw).ok()?;
727 serde_json::from_str(&raw).ok()
728}
729
730fn describe_forge_hits(repo: &Path, hits: &mut [Hit]) {
735 for h in hits
736 .iter_mut()
737 .filter(|h| h.owner == Owner::Pr && h.about.is_empty())
738 {
739 let Some(n) = h.token.trim_start_matches('#').parse::<u64>().ok() else {
740 continue;
741 };
742 if let Some(v) = gh_pr_view(repo, n, "title,body") {
743 h.about = about_of(
744 v["title"].as_str().unwrap_or(""),
745 v["body"].as_str().unwrap_or(""),
746 );
747 }
748 }
749}
750
751fn squeeze(s: &str, max: usize) -> String {
755 let n = s.chars().count();
756 if n <= max {
757 return s.to_owned();
758 }
759 let half = max / 2;
760 let head: String = s.chars().take(half).collect();
761 let tail: String = s.chars().skip(n - half).collect();
762 format!(
763 "{head}\n[... {} characters omitted ...]\n{tail}",
764 n - 2 * half
765 )
766}
767
768fn about_of(title: &str, instruction: &str) -> String {
773 let body = instruction.split_whitespace().collect::<Vec<_>>().join(" ");
774 let title = title.trim();
775 let full = match (title.is_empty(), body.is_empty()) {
776 (true, _) => body,
777 (false, true) => title.to_owned(),
778 (false, false) => format!("{title}: {body}"),
779 };
780 squeeze(&full, ABOUT_MAX_CHARS)
781}
782
783const ABOUT_MAX_CHARS: usize = prompt::DUPES_JUDGE_MAX_CHARS;
787
788fn task_claims(
789 t: &Task,
790 runs_root: &Path,
791 seen: &mut BTreeSet<String>,
792 repo: &Path,
793 stale: &mut Staleness<'_>,
794) -> Vec<Claim> {
795 let status = t.status.as_str().to_owned();
796 let mut out = Vec::new();
797 if let Some(b) = &t.review_branch {
798 out.push(Claim {
799 owner: Owner::Task,
800 id: t.id.clone(),
801 status: status.clone(),
802 via: "its review branch".into(),
803 about: about_of(&t.title, &t.instruction),
804 branch: Some(b.clone()),
805 base: None,
806 pr: None,
807 });
808 }
809 for rid in &t.runs {
810 if let Some(view) = RunView::read(runs_root, rid) {
811 seen.insert(rid.clone());
812 let live_pr = !(view.terminal() && stale.pr_released(repo, &view));
815 out.extend(run_claims(
816 &view,
817 Owner::Task,
818 Some((&t.id, &status)),
819 &format!("produced by its run {}", crate::queue::short(rid)),
820 live_pr,
821 (&t.title, &t.instruction),
822 ));
823 }
824 }
825 out
826}
827
828fn run_claims(
833 view: &RunView,
834 owner: Owner,
835 task: Option<(&str, &str)>,
836 via: &str,
837 live_pr: bool,
838 (title, instruction): (&str, &str),
839) -> Vec<Claim> {
840 let about = about_of(title, instruction);
841 let (id, status) = match task {
842 Some((id, status)) => (id.to_owned(), status.to_owned()),
843 None => (view.id.clone(), view.status.clone()),
844 };
845 let pr = view
846 .pr
847 .as_ref()
848 .filter(|p| live_pr && p.state == "open" && p.number > 0)
849 .map(|p| (p.number, p.url.clone()));
850 let via_pr = |extra: &str| match &pr {
851 Some((n, _)) => format!("{via} (PR #{n} open){extra}"),
852 None => format!("{via}{extra}"),
853 };
854 let mut out: Vec<Claim> = view
855 .candidates
856 .iter()
857 .filter(|c| !c.branch.is_empty())
858 .map(|c| Claim {
859 owner: owner.clone(),
860 id: id.clone(),
861 status: status.clone(),
862 via: via_pr(""),
863 about: about.clone(),
864 branch: Some(c.branch.clone()),
865 base: (!view.base_commit.is_empty()).then(|| view.base_commit.clone()),
866 pr: None,
867 })
868 .collect();
869 if pr.is_some() {
870 out.push(Claim {
871 owner,
872 id,
873 status,
874 via: via_pr(""),
875 about,
876 branch: None,
877 base: None,
878 pr,
879 });
880 }
881 out
882}
883
884#[derive(Default)]
887struct Idents(HashMap<PathBuf, PathBuf>);
888
889impl Idents {
890 fn of(&mut self, path: &Path) -> PathBuf {
891 self.0
892 .entry(path.to_path_buf())
893 .or_insert_with(|| {
894 git(
895 path,
896 &["rev-parse", "--path-format=absolute", "--git-common-dir"],
897 )
898 .map(PathBuf::from)
899 .and_then(|p| p.canonicalize().ok())
900 .or_else(|| path.canonicalize().ok())
901 .unwrap_or_else(|| path.to_path_buf())
902 })
903 .clone()
904 }
905}
906
907fn git(cwd: &Path, args: &[&str]) -> Option<String> {
908 let out = Command::new("git")
909 .quiet()
910 .args(args)
911 .current_dir(cwd)
912 .stdin(Stdio::null())
913 .stderr(Stdio::null())
914 .env("GIT_TERMINAL_PROMPT", "0")
915 .output()
916 .ok()?;
917 out.status
918 .success()
919 .then(|| String::from_utf8_lossy(&out.stdout).trim().to_owned())
920}
921
922fn git_ok(cwd: &Path, args: &[&str]) -> bool {
923 git(cwd, args).is_some()
924}
925
926fn is_ref_char(c: char) -> bool {
927 c.is_alphanumeric() || matches!(c, '_' | '-')
928}
929
930fn names_branch(text: &str, branch: &str) -> bool {
934 if branch.len() < 3 {
935 return false;
936 }
937 text.match_indices(branch).any(|(i, _)| {
938 let before = text[..i].chars().next_back();
939 let after = text[i + branch.len()..].chars().next();
940 let after_ok = match after {
941 None => true,
942 Some('/') => false,
943 Some('.') => !text[i + branch.len() + 1..]
944 .chars()
945 .next()
946 .is_some_and(is_ref_char),
947 Some(c) => !is_ref_char(c),
948 };
949 before.is_none_or(|c| !is_ref_char(c) && c != '.') && after_ok
950 })
951}
952
953#[derive(Debug, PartialEq, Eq)]
954enum Mention {
955 Number(u64),
956 Url(String),
957}
958
959fn pr_numbers(text: &str) -> Vec<Mention> {
961 let mut out = Vec::new();
962 let bytes = text.as_bytes();
963 let digits = |from: usize| -> Option<(u64, usize)> {
964 let n = text[from..].bytes().take_while(u8::is_ascii_digit).count();
965 (n > 0 && n < 10)
966 .then(|| text[from..from + n].parse().ok().map(|v| (v, from + n)))
967 .flatten()
968 };
969 for (i, _) in text.match_indices('#') {
970 if let Some((n, _)) = digits(i + 1) {
971 let word_before = i > 0 && is_ref_char(bytes[i - 1] as char);
972 if !word_before {
973 out.push(Mention::Number(n));
974 }
975 }
976 }
977 let lower = text.to_ascii_lowercase();
978 for key in ["pull request ", "pr "] {
979 for (i, _) in lower.match_indices(key) {
980 if i > 0 && is_ref_char(bytes[i - 1] as char) {
981 continue;
982 }
983 let from = i + key.len();
984 let from = if text[from..].starts_with('#') {
985 from + 1
986 } else {
987 from
988 };
989 if let Some((n, _)) = digits(from) {
990 out.push(Mention::Number(n));
991 }
992 }
993 }
994 for (i, _) in text.match_indices("/pull/") {
995 if let Some((_, end)) = digits(i + 6) {
996 let start = text[..i]
997 .rfind(|c: char| c.is_whitespace() || matches!(c, '(' | '<' | '"' | '\''))
998 .map_or(0, |p| p + 1);
999 out.push(Mention::Url(text[start..end].to_owned()));
1000 }
1001 }
1002 out
1003}
1004
1005fn sha_candidates(repo: &Path, text: &str) -> Vec<String> {
1007 let mut seen = BTreeSet::new();
1008 let mut out = Vec::new();
1009 for word in text.split(|c: char| !c.is_ascii_alphanumeric()) {
1010 if !(7..=40).contains(&word.len()) || !word.bytes().all(|b| b.is_ascii_hexdigit()) {
1011 continue;
1012 }
1013 if seen.len() >= 16 || !seen.insert(word.to_ascii_lowercase()) {
1014 continue;
1015 }
1016 if let Some(full) = git(
1017 repo,
1018 &[
1019 "rev-parse",
1020 "--verify",
1021 "--quiet",
1022 &format!("{word}^{{commit}}"),
1023 ],
1024 ) {
1025 out.push(full);
1026 }
1027 }
1028 out
1029}
1030
1031fn on_branch_only(repo: &Path, sha: &str, branch: &str, base: &str) -> bool {
1034 let tip = format!("{branch}^{{commit}}");
1035 git_ok(repo, &["rev-parse", "--verify", "--quiet", &tip])
1036 && git_ok(repo, &["merge-base", "--is-ancestor", sha, branch])
1037 && !git_ok(repo, &["merge-base", "--is-ancestor", sha, base])
1038}
1039
1040fn short_sha(sha: &str) -> String {
1041 sha.chars().take(7).collect()
1042}
1043
1044#[cfg(test)]
1045mod tests {
1046 use super::*;
1047 use crate::queue::{Source, Task};
1048
1049 struct Fx {
1050 _tmp: tempfile::TempDir,
1051 repo: PathBuf,
1052 runs: PathBuf,
1053 q: Queue,
1054 }
1055
1056 fn sh(cwd: &Path, args: &[&str]) -> String {
1057 let out = Command::new("git")
1058 .quiet()
1059 .args(["-c", "user.name=t", "-c", "user.email=t@t"])
1060 .args(args)
1061 .current_dir(cwd)
1062 .output()
1063 .unwrap();
1064 assert!(out.status.success(), "git {args:?}: {out:?}");
1065 String::from_utf8_lossy(&out.stdout).trim().to_owned()
1066 }
1067
1068 fn fx() -> (Fx, String, String) {
1071 let tmp = tempfile::tempdir().unwrap();
1072 let repo = tmp.path().join("repo");
1073 std::fs::create_dir_all(&repo).unwrap();
1074 sh(&repo, &["init", "-q", "-b", "main"]);
1075 std::fs::write(repo.join("a"), "1").unwrap();
1076 sh(&repo, &["add", "."]);
1077 sh(&repo, &["commit", "-q", "-m", "base"]);
1078 let base = sh(&repo, &["rev-parse", "HEAD"]);
1079 sh(&repo, &["checkout", "-q", "-b", "magi/aaaa/A"]);
1080 std::fs::write(repo.join("a"), "2").unwrap();
1081 sh(&repo, &["commit", "-q", "-am", "work"]);
1082 let tip = sh(&repo, &["rev-parse", "HEAD"]);
1083 sh(&repo, &["checkout", "-q", "main"]);
1084 let runs = tmp.path().join("runs");
1085 std::fs::create_dir_all(&runs).unwrap();
1086 let q = Queue::at(tmp.path().join("queue"));
1087 (
1088 Fx {
1089 _tmp: tmp,
1090 repo,
1091 runs,
1092 q,
1093 },
1094 base,
1095 tip,
1096 )
1097 }
1098
1099 fn write_run(f: &Fx, id: &str, status: &str, base: &str, pr: Option<(u64, &str)>) {
1100 let dir = f.runs.join(id);
1101 std::fs::create_dir_all(&dir).unwrap();
1102 let pr = pr.map(|(n, s)| {
1103 serde_json::json!({"url": format!("https://github.com/o/r/pull/{n}"), "number": n, "state": s})
1104 });
1105 let v = serde_json::json!({
1106 "schema": 999, "id": id, "repo": f.repo, "status": status,
1107 "base_commit": base, "candidates": [{"branch": "magi/aaaa/A"}], "pr": pr,
1108 });
1109 std::fs::write(dir.join("run.json"), v.to_string()).unwrap();
1110 }
1111
1112 fn file_task(f: &Fx, status: TaskStatus, runs: &[&str]) -> Task {
1113 let mut t = Task::new("t".into(), "x".into(), f.repo.clone(), Source::Human);
1114 t.status = status;
1115 t.runs = runs.iter().map(|s| (*s).to_owned()).collect();
1116 f.q.put(&mut t).unwrap();
1117 t
1118 }
1119
1120 const RID: &str = "20260901-100000-aaaa";
1121
1122 fn run(f: &Fx, text: &str, review: Option<&str>) -> Vec<Hit> {
1123 check_with(
1124 &f.q,
1125 &f.runs,
1126 &f.repo,
1127 text,
1128 review,
1129 None,
1130 &|_, _| None,
1131 &|_, _| None,
1132 )
1133 }
1134
1135 #[test]
1136 fn branch_matches_a_live_run_and_its_task() {
1137 let (f, base, _) = fx();
1138 write_run(&f, RID, "reviewing", &base, None);
1139 let t = file_task(&f, TaskStatus::Running, &[RID]);
1140 let hits = run(&f, "land magi/aaaa/A onto a fresh branch.", None);
1141 assert!(hits.iter().any(|h| h.owner == Owner::Task
1142 && h.id == t.id
1143 && h.signal == Signal::Branch
1144 && h.token == "magi/aaaa/A"));
1145 assert!(hits.iter().any(|h| h.owner == Owner::Run && h.id == RID));
1146 let msg = Duplicate::new(hits).to_string();
1147 assert!(
1148 msg.contains("--force") && msg.contains("magi/aaaa/A"),
1149 "{msg}"
1150 );
1151 }
1152
1153 #[test]
1154 fn branch_must_match_whole_word() {
1155 let (f, base, _) = fx();
1156 write_run(&f, RID, "reviewing", &base, None);
1157 assert!(run(&f, "see magi/aaaa/AB and magi/aaaa/A/x", None).is_empty());
1158 }
1159
1160 #[test]
1161 fn sha_on_the_branch_matches_but_one_in_base_does_not() {
1162 let (f, base, tip) = fx();
1163 write_run(&f, RID, "reviewing", &base, None);
1164 let hits = run(&f, &format!("land commit {} please", &tip[..8]), None);
1165 assert!(
1166 hits.iter().any(|h| h.signal == Signal::Sha && h.id == RID),
1167 "{hits:?}"
1168 );
1169 assert!(run(&f, &format!("see {}", &base[..9]), None).is_empty());
1170 assert!(run(&f, "deadbeef and 1234567", None).is_empty());
1172 }
1173
1174 #[test]
1175 fn pr_number_matches_in_every_spelling() {
1176 let (f, base, _) = fx();
1177 write_run(&f, RID, "ready", &base, Some((48, "open")));
1178 for text in [
1179 "finish #48",
1180 "PR 48 is stale",
1181 "pr #48",
1182 "pull request 48",
1183 "https://github.com/o/r/pull/48",
1184 ] {
1185 let hits = run(&f, text, None);
1186 assert!(
1187 hits.iter().any(|h| h.signal == Signal::Pr),
1188 "{text}: {hits:?}"
1189 );
1190 }
1191 assert!(run(&f, "see #480 and PR 4 and issue48", None).is_empty());
1192 }
1193
1194 #[test]
1195 fn terminal_runs_and_done_tasks_do_not_match() {
1196 let (f, base, tip) = fx();
1197 write_run(&f, RID, "merged", &base, Some((48, "merged")));
1198 file_task(&f, TaskStatus::Done, &[RID]);
1199 let text = format!("magi/aaaa/A {} #48", &tip[..8]);
1200 assert!(run(&f, &text, None).is_empty());
1201 }
1202
1203 #[test]
1204 fn terminal_run_with_open_pr_or_open_task_still_claims() {
1205 let (f, base, _) = fx();
1210 write_run(&f, RID, "ready", &base, Some((48, "open")));
1211 assert!(!run(&f, "magi/aaaa/A", None).is_empty());
1212 let (g, base, _) = fx();
1213 write_run(&g, RID, "ready", &base, None);
1214 assert!(run(&g, "magi/aaaa/A", None).is_empty());
1215 let t = file_task(&g, TaskStatus::Held, &[RID]);
1216 let hits = run(&g, "magi/aaaa/A", None);
1217 assert!(hits.iter().any(|h| h.id == t.id), "{hits:?}");
1218 }
1219
1220 #[test]
1221 fn review_only_matches_a_branch_a_live_task_owns() {
1222 let (f, base, _) = fx();
1223 write_run(&f, RID, "ready", &base, None);
1224 assert!(run(&f, "", Some("magi/aaaa/A")).is_empty());
1225 let mut t = file_task(&f, TaskStatus::Queued, &[]);
1226 t.review_branch = Some("magi/aaaa/A".into());
1227 f.q.put(&mut t).unwrap();
1228 let hits = run(&f, "", Some("magi/aaaa/A"));
1229 assert!(
1230 hits.iter()
1231 .any(|h| h.id == t.id && h.signal == Signal::Branch)
1232 );
1233 }
1234
1235 #[test]
1236 fn other_repository_and_edited_task_do_not_match() {
1237 let (f, base, _) = fx();
1238 write_run(&f, RID, "reviewing", &base, None);
1239 let t = file_task(&f, TaskStatus::Running, &[RID]);
1240 let other = f._tmp.path().join("other");
1241 std::fs::create_dir_all(&other).unwrap();
1242 sh(&other, &["init", "-q"]);
1243 assert!(
1244 check_with(
1245 &f.q,
1246 &f.runs,
1247 &other,
1248 "magi/aaaa/A",
1249 None,
1250 None,
1251 &|_, _| None,
1252 &|_, _| None,
1253 )
1254 .is_empty()
1255 );
1256 assert!(
1258 check_with(
1259 &f.q,
1260 &f.runs,
1261 &f.repo,
1262 "magi/aaaa/A",
1263 None,
1264 Some(&t.id),
1265 &|_, _| None,
1266 &|_, _| None,
1267 )
1268 .is_empty()
1269 );
1270 }
1271
1272 #[test]
1273 fn a_worktree_is_the_same_repository() {
1274 let (f, base, _) = fx();
1275 write_run(&f, RID, "reviewing", &base, None);
1276 let wt = f._tmp.path().join("wt");
1277 sh(
1278 &f.repo,
1279 &["worktree", "add", "-q", wt.to_str().unwrap(), "-b", "other"],
1280 );
1281 assert!(
1282 !check_with(
1283 &f.q,
1284 &f.runs,
1285 &wt,
1286 "magi/aaaa/A",
1287 None,
1288 None,
1289 &|_, _| None,
1290 &|_, _| None,
1291 )
1292 .is_empty()
1293 );
1294 }
1295
1296 #[test]
1297 fn an_open_pr_without_a_run_record_matches_through_the_forge() {
1298 let (f, _, _) = fx();
1299 let open = |_: &Path, n: u64| (n == 48).then(|| "https://example.test/pull/48".to_owned());
1300 let hit = |text: &str| {
1301 check_with(&f.q, &f.runs, &f.repo, text, None, None, &open, &|_, _| {
1302 None
1303 })
1304 };
1305 let hits = hit("finish PR #48");
1306 assert_eq!(hits.len(), 1, "{hits:?}");
1307 assert_eq!(hits[0].owner, Owner::Pr);
1308 assert!(hits[0].to_string().contains("#48"));
1309 assert!(hit("finish PR #49").is_empty());
1311 assert!(hit("finish the work").is_empty());
1312 }
1313
1314 #[test]
1315 fn a_forge_hit_does_not_repeat_a_pr_a_run_already_explains() {
1316 let (f, base, _) = fx();
1317 write_run(&f, RID, "ready", &base, Some((48, "open")));
1318 let open = |_: &Path, _: u64| Some("u".to_owned());
1319 let hits = check_with(&f.q, &f.runs, &f.repo, "#48", None, None, &open, &|_, _| {
1320 None
1321 });
1322 assert!(hits.iter().all(|h| h.owner != Owner::Pr), "{hits:?}");
1323 assert!(!hits.is_empty());
1324 }
1325
1326 #[test]
1327 fn an_edited_tasks_own_open_pr_is_not_a_forge_hit() {
1328 let (f, base, _) = fx();
1329 write_run(&f, RID, "ready", &base, Some((48, "open")));
1330 let t = file_task(&f, TaskStatus::Running, &[RID]);
1331 let open = |_: &Path, _: u64| Some("u".to_owned());
1332 let hits = check_with(
1333 &f.q,
1334 &f.runs,
1335 &f.repo,
1336 "#48",
1337 None,
1338 Some(&t.id),
1339 &open,
1340 &|_, _| None,
1341 );
1342 assert!(hits.is_empty(), "{hits:?}");
1343 }
1344
1345 fn with_forge(f: &Fx, text: &str, state: Option<PrLifecycle>) -> Vec<Hit> {
1346 check_with(
1347 &f.q,
1348 &f.runs,
1349 &f.repo,
1350 text,
1351 None,
1352 None,
1353 &|_, _| None,
1354 &move |_, _| state,
1355 )
1356 }
1357
1358 fn release(f: &Fx, id: &str, to: &str) {
1359 let path = f.runs.join(id).join("run.json");
1360 let mut v: serde_json::Value =
1361 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
1362 v["released_to"] = serde_json::json!(to);
1363 std::fs::write(path, v.to_string()).unwrap();
1364 }
1365
1366 #[test]
1367 fn stale_open_pr_that_the_forge_says_is_merged_or_closed_does_not_claim() {
1368 for state in [PrLifecycle::Merged, PrLifecycle::Closed] {
1369 let (f, base, _) = fx();
1370 write_run(&f, RID, "superseded", &base, Some((48, "open")));
1371 assert!(with_forge(&f, "follow up on #48", Some(state)).is_empty());
1372 assert!(with_forge(&f, "magi/aaaa/A", Some(state)).is_empty());
1373 file_task(&f, TaskStatus::Held, &[RID]);
1375 let hits = with_forge(&f, "follow up on #48", Some(state));
1376 assert!(hits.iter().all(|h| h.signal != Signal::Pr), "{hits:?}");
1377 }
1378 }
1379
1380 #[test]
1381 fn stale_open_pr_with_an_unreadable_forge_still_claims() {
1382 let (f, base, _) = fx();
1383 write_run(&f, RID, "superseded", &base, Some((48, "open")));
1384 let hits = with_forge(&f, "follow up on #48", None);
1385 assert!(hits.iter().any(|h| h.signal == Signal::Pr), "{hits:?}");
1386 }
1387
1388 #[test]
1389 fn a_genuinely_open_pr_still_claims() {
1390 let (f, base, _) = fx();
1391 write_run(&f, RID, "blocked", &base, Some((48, "open")));
1392 let hits = with_forge(&f, "follow up on #48", Some(PrLifecycle::Open));
1393 assert!(hits.iter().any(|h| h.signal == Signal::Pr), "{hits:?}");
1394 }
1395
1396 #[test]
1397 fn a_released_run_defers_to_its_successor_without_asking_the_forge() {
1398 let (f, base, _) = fx();
1399 let next = "20260901-110000-bbbb";
1400 write_run(&f, RID, "superseded", &base, Some((48, "open")));
1401 write_run(&f, next, "merged", &base, Some((48, "merged")));
1402 release(&f, RID, next);
1403 let asked = std::cell::Cell::new(0);
1404 let hits = check_with(
1405 &f.q,
1406 &f.runs,
1407 &f.repo,
1408 "follow up on #48",
1409 None,
1410 None,
1411 &|_, _| None,
1412 &|_, _| {
1413 asked.set(asked.get() + 1);
1414 None
1415 },
1416 );
1417 assert!(hits.is_empty(), "{hits:?}");
1418 assert_eq!(asked.get(), 0);
1419 std::fs::remove_dir_all(f.runs.join(next)).unwrap();
1421 let hits = with_forge(&f, "follow up on #48", None);
1422 assert!(hits.iter().any(|h| h.signal == Signal::Pr), "{hits:?}");
1423 }
1424
1425 #[test]
1426 fn forge_lookups_are_cached_and_stop_after_a_failure() {
1427 let (f, base, _) = fx();
1428 for (i, id) in ["20260901-100000-aaa1", "20260901-100000-aaa2"]
1429 .iter()
1430 .enumerate()
1431 {
1432 write_run(&f, id, "blocked", &base, Some((48 + i as u64, "open")));
1433 }
1434 let asked = std::cell::Cell::new(0);
1435 check_with(
1436 &f.q,
1437 &f.runs,
1438 &f.repo,
1439 "x",
1440 None,
1441 None,
1442 &|_, _| None,
1443 &|_, _| {
1444 asked.set(asked.get() + 1);
1445 None
1446 },
1447 );
1448 assert_eq!(
1449 asked.get(),
1450 1,
1451 "an unreadable forge is asked once, not per PR"
1452 );
1453 }
1454
1455 use std::sync::Arc;
1456 use std::sync::atomic::{AtomicUsize, Ordering};
1457
1458 fn verdict(duplicate: bool) -> Result<Judgement> {
1459 Ok(Judgement {
1460 duplicate,
1461 reason: "because".into(),
1462 agent: "j".into(),
1463 })
1464 }
1465
1466 fn judged(
1469 f: &Fx,
1470 text: &str,
1471 answer: impl Fn() -> Result<Judgement> + Send + Sync + 'static,
1472 ) -> (Result<Screened, Duplicate>, usize) {
1473 let calls = Arc::new(AtomicUsize::new(0));
1474 let seen = calls.clone();
1475 let judge = move |_: String, _: Vec<Hit>| -> JudgeFuture {
1476 seen.fetch_add(1, Ordering::SeqCst);
1477 let r = answer();
1478 Box::pin(async move { r })
1479 };
1480 let hits = run(f, text, None);
1481 let out = tokio::runtime::Builder::new_current_thread()
1482 .build()
1483 .unwrap()
1484 .block_on(screen(hits, text, None, &judge));
1485 (out, calls.load(Ordering::SeqCst))
1486 }
1487
1488 const NAMES: &str = "seen on magi/aaaa/A, which does not touch this file";
1489
1490 fn fx_with_run() -> Fx {
1491 let (f, base, _) = fx();
1492 write_run(&f, RID, "reviewing", &base, None);
1493 f
1494 }
1495
1496 #[test]
1497 fn a_no_ruling_lets_the_work_through() {
1498 let f = fx_with_run();
1499 let (out, calls) = judged(&f, NAMES, || verdict(false));
1500 assert!(matches!(out, Ok(Screened::Cleared(_))));
1501 assert_eq!(calls, 1);
1502 }
1503
1504 #[test]
1505 fn a_yes_ruling_refuses_and_says_why() {
1506 let f = fx_with_run();
1507 let (out, _) = judged(&f, NAMES, || verdict(true));
1508 let msg = out.unwrap_err().to_string();
1509 assert!(
1510 msg.contains("duplicate - because") && msg.contains("--force"),
1511 "{msg}"
1512 );
1513 assert!(msg.contains("magi/aaaa/A"), "{msg}");
1514 }
1515
1516 #[test]
1517 fn a_failing_judge_lets_it_through_unjudged() {
1518 let f = fx_with_run();
1519 let (out, calls) = judged(&f, NAMES, || Err(anyhow::anyhow!("quota")));
1520 match out {
1521 Ok(Screened::Unjudged(why)) => assert!(why.contains("quota"), "{why}"),
1522 other => panic!("{other:?}"),
1523 }
1524 assert_eq!(calls, 1);
1525 }
1526
1527 #[test]
1528 fn garbage_replies_do_not_parse() {
1529 assert!(parse_judgement("sure, go ahead").is_err());
1530 assert!(parse_judgement(r#"{"duplicate":"maybe","reason":"x"}"#).is_err());
1531 assert!(parse_judgement(r#"{"duplicate":false}"#).is_err());
1532 assert!(parse_judgement(r#"{"duplicate":false,"reason":" "}"#).is_err());
1533 let two = "{\"duplicate\":false,\"reason\":\"c\"}\n{\"duplicate\":true}";
1534 assert!(parse_judgement(two).is_err());
1535 assert!(parse_judgement("ok {\"duplicate\":false,\"reason\":\"c\"}").is_err());
1536 let (d, why) = parse_judgement("{\"duplicate\":true,\"reason\":\"same\\nPR\"}").unwrap();
1537 assert_eq!((d, why.as_str()), (true, "same PR"));
1538 }
1539
1540 #[test]
1541 fn a_text_too_long_is_squeezed_and_still_judged() {
1542 let f = fx_with_run();
1543 let long = format!("{NAMES} {}", "x".repeat(prompt::DUPES_JUDGE_MAX_CHARS));
1544 let (out, calls) = judged(&f, &long, || verdict(false));
1545 assert!(matches!(out, Ok(Screened::Cleared(_))));
1546 assert_eq!(calls, 1);
1547 }
1548
1549 #[test]
1550 fn no_hit_never_asks_the_judge() {
1551 let (f, _, _) = fx();
1552 let (out, calls) = judged(&f, "nothing named here", || verdict(true));
1553 assert_eq!(out.unwrap(), Screened::Clean);
1554 assert_eq!(calls, 0);
1555 }
1556
1557 #[test]
1558 fn without_a_config_a_hit_passes_unjudged() {
1559 let f = fx_with_run();
1560 let hits = run(&f, NAMES, None);
1561 let out = tokio::runtime::Builder::new_current_thread()
1562 .build()
1563 .unwrap()
1564 .block_on(screen_with_config(hits, NAMES, None, &f.repo, None));
1565 assert!(matches!(out, Ok(Screened::Unjudged(_))));
1566 }
1567
1568 #[test]
1569 fn about_carries_title_and_whole_instruction() {
1570 assert_eq!(
1571 about_of("Retries", "# Task\nFix auth\nretries"),
1572 "Retries: # Task Fix auth retries"
1573 );
1574 assert_eq!(about_of("", "x"), "x");
1575 let mid = format!("{} MIDDLE {}", "a ".repeat(2000), "b ".repeat(500));
1576 assert!(about_of("", &mid).contains("MIDDLE"));
1577 let long = format!("{} TAIL", "y ".repeat(5000));
1578 let a = about_of("", &long);
1579 assert!(a.ends_with("TAIL") && a.chars().count() < 6100, "{a}");
1580 }
1581}