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