Skip to main content

magi/
dupes.rs

1//! Duplicate-work detection at the moment work is filed.
2//!
3//! Two tasks once chased the same commit without knowing about each other: an
4//! open task and its open pull request already covered a change when a chat
5//! filed a second task that named the same branch and commit. Nothing at
6//! filing time said so. This module is that check, and it is deliberately
7//! small: **concrete identifiers only** - a branch name, a commit SHA, a pull
8//! request number - never text similarity. A false positive costs one
9//! `--force`; a false negative is the behaviour before this module existed.
10//!
11//! A *claim* is a thing an unfinished piece of work owns: the branches of its
12//! candidates and the pull request it opened. The claims come from
13//!
14//! - every task that is not `done`: its `review_branch`, and each run in
15//!   `Task::runs`;
16//! - every run that has not reached a terminal status;
17//! - every run that is terminal but whose pull request is still open (the
18//!   shape of the original accident: the run ended, the PR did not).
19//!
20//! Run records are read through a tolerant view rather than `RunState::load`:
21//! a schema bump must not turn into a silent false negative (the same lesson
22//! as `clean::fold_due`).
23//!
24//! Everything here is synchronous, read-only and takes its roots as
25//! arguments, so tests need no process-global home.
26
27use 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/// What kind of identifier matched.
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub enum Signal {
50    /// A branch name.
51    Branch,
52    /// A commit SHA.
53    Sha,
54    /// A pull request number or URL.
55    Pr,
56}
57
58/// Who owns the thing that matched.
59#[derive(Debug, Clone, PartialEq, Eq)]
60pub enum Owner {
61    /// A queued task.
62    Task,
63    /// A recorded run.
64    Run,
65    /// A pull request the forge reports open that no record here owns.
66    Pr,
67}
68
69/// One reason a new piece of work looks like one already under way.
70#[derive(Debug, Clone, PartialEq, Eq)]
71pub struct Hit {
72    /// Task or run.
73    pub owner: Owner,
74    /// Full id of the task or run.
75    pub id: String,
76    /// The owner's status word (`queued`, `reviewing`, ...).
77    pub status: String,
78    /// What kind of identifier matched.
79    pub signal: Signal,
80    /// The identifier as it matched: a branch, a SHA, `#48`.
81    pub token: String,
82    /// How the owner is tied to it, e.g. `produced by its run c9eb`.
83    pub via: String,
84    /// The owner's title and instruction (collapsed, cut), for the
85    /// judge only: never shown in [`Display`]. Empty when no record has it.
86    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/// The refusal: one or more [`Hit`]s and nothing filed.
113#[derive(Debug, Clone)]
114pub struct Duplicate {
115    /// What matched.
116    pub hits: Vec<Hit>,
117    /// The judge's word, when one was asked: a ready-made line for the
118    /// refusal text. `None` when no judge ran.
119    pub judge: Option<String>,
120}
121
122impl Duplicate {
123    /// A refusal on the identifier match alone.
124    pub fn new(hits: Vec<Hit>) -> Self {
125        Self { hits, judge: None }
126    }
127
128    /// The refusal text, ending in how to override it.
129    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/// The judge's answer, with the agent that gave it.
157#[derive(Debug, Clone, PartialEq, Eq)]
158pub struct Judgement {
159    /// True when the new work is genuinely a duplicate of the matched claims.
160    pub duplicate: bool,
161    /// One line, as the agent wrote it.
162    pub reason: String,
163    /// Id of the agent that answered.
164    pub agent: String,
165}
166
167/// How [`screen`] let work through.
168#[derive(Debug, Clone, PartialEq, Eq)]
169pub enum Screened {
170    /// No identifier matched; no judge was asked.
171    Clean,
172    /// A match was found and the judge said it is not a duplicate.
173    Cleared(Judgement),
174    /// A match was found and nobody could judge it (the reason): let through.
175    Unjudged(String),
176}
177
178/// A judge: given the instruction and the matched claims, rules on them.
179/// Owned arguments, so a boxed future needs no borrow.
180pub type JudgeFuture = Pin<Box<dyn Future<Output = Result<Judgement>> + Send>>;
181
182/// Whole-chain wall-clock budget for the judge call.
183const JUDGE_BUDGET: Duration = Duration::from_secs(45);
184/// Per-agent cap inside that budget.
185const JUDGE_TURN: Duration = Duration::from_secs(30);
186
187/// Read the judge's reply. Strict: anything but one JSON object with a bool
188/// `duplicate` and a non-empty reason is an error (the caller lets the work
189/// through unjudged, see [`screen`]).
190fn parse_judgement(text: &str) -> Result<(bool, String)> {
191    #[derive(Deserialize)]
192    struct Raw {
193        duplicate: bool,
194        reason: String,
195    }
196    // The whole reply must be one object (optionally in a code fence): hunting
197    // for an object inside prose or a second answer could recover a verdict
198    // from a reply that is not itself a single valid one.
199    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
213/// Ask the `[roles] chatter` chain (or the default agent) once. Each agent is
214/// tried at most once, with a fresh seat; only an error, quota or unusable
215/// answer moves on. A usable answer that does not parse ends the chain.
216pub 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
285/// The one decision point every caller shares. `hits` is [`check`]'s output:
286/// empty passes without the judge being asked. Otherwise the judge is asked
287/// once (a long text is squeezed to its head and tail first), and only a `duplicate: true` answer refuses (with its reason).
288///
289/// **Fail mode: open.** If the judge cannot answer (agent error, timeout,
290/// quota, unparseable reply, no readable config) the work is let through as [`Screened::Unjudged`] and a
291/// `tracing::warn` is logged; callers also say so on stderr. A mechanical
292/// match cannot tell "continue from PR #28" from a real duplicate, so refusing
293/// whenever the judge is away would bring back exactly the false positives the
294/// judge exists to remove. The cost is that a real duplicate slips through
295/// while no agent is available, which is why it is never silent.
296///
297/// For a review-only request `text` is empty, so `review_branch` is put in
298/// front of the judge as the thing the work is about.
299pub 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
339/// [`screen`] with the production judge: the `[roles] chatter` chain of `cfg`.
340/// No config (`None`) means no judge, so a hit is let through unjudged.
341pub 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/// The slice of `run.json` this module needs. Every field is optional so a
372/// record from any schema still yields what it has.
373#[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    /// The run that took this one's worktree (and pull request) over.
390    #[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    /// An unknown status word counts as unfinished: claiming too much is the
417    /// cheap error here.
418    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
429/// Decides whether the pull request a run's record still calls open is in fact
430/// settled, so a stale record stops claiming it.
431///
432/// A record freezes the last state its land loop polled. A run that was handed
433/// over (`released_to`) or left blocked while another run landed the same pull
434/// request keeps saying `open` forever, so the record alone cannot be trusted
435/// to *keep* a claim alive. Two answers, cheapest first:
436///
437/// 1. a released run does not own the pull request any more: ownership follows
438///    `released_to`, and the successor's own record (read like any other run's)
439///    decides - provided that record is readable;
440/// 2. otherwise the forge is asked, once per number, at most
441///    [`MAX_FORGE_LOOKUPS`] numbers, and never again after one lookup failed
442///    (offline stays fast). Merged or closed means no claim; open, an error or
443///    a skipped lookup means the claim stands. Claiming too much is the cheap
444///    error.
445struct 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
453/// How many pull requests one invocation may put to the forge on behalf of
454/// stale records.
455const 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    /// `true` when `view`'s recorded-open pull request is known not to be this
469    /// run's to claim any more.
470    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/// Something an unfinished piece of work owns.
504#[derive(Debug, Clone)]
505struct Claim {
506    owner: Owner,
507    id: String,
508    status: String,
509    via: String,
510    /// The owner's title and instruction, for the judge.
511    about: String,
512    branch: Option<String>,
513    /// Commit the branch forked from; without it a SHA cannot be judged.
514    base: Option<String>,
515    /// `(number, url)` of an open pull request.
516    pr: Option<(u64, String)>,
517}
518
519/// Look for work already in flight that `text` (and `review_branch`, for a
520/// review-only run) points at. `repo` is the repository the new work is for;
521/// `ignore_task` is a task being edited, whose own claims never count.
522pub 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/// [`check`] with the forge lookup supplied: `open_pr(repo, n)` says whether
543/// pull request `n` of `repo` is open (`Some(url)`) or not / unknown (`None`).
544/// It is asked only about numbers the text names that no local record already
545/// explained, so a PR magi never produced (opened by hand) still collides.
546/// `pr_state(repo, n)` is the forge's word on a pull request a run record calls
547/// open (`None` = unreadable, which keeps the claim); see [`Staleness`].
548#[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    // PRs the edited task's own runs opened: never a rival, forge or not.
569    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        // A terminal run held up only by a pull request that is not its own
596        // (any more) claims nothing at all.
597        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
678/// Ask the forge, via `gh`, whether PR `n` is open. Best effort and bounded:
679/// any failure (no `gh`, no remote, offline, a slow answer) is `None`, i.e.
680/// today's behaviour. `GH_REPO` is dropped so the PR is looked up in `repo`'s
681/// own remote, not whatever the environment points at.
682fn 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
689/// [`gh_open_pr`]'s question asked of a pull request a run record names: its
690/// lifecycle, or `None` when the forge cannot say.
691fn 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
730/// Fill in what a forge-only hit (an open pull request no record here owns)
731/// is about, from its title and body. Only the judge reads it, so this runs
732/// after the mechanical check, only on a hit, and best effort: an unreadable
733/// forge leaves `about` empty, never drops the hit.
734fn 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
751/// Keep the head and the tail of `s` within `max` characters, marking the cut.
752/// A long instruction often has background first and the concrete change last,
753/// so dropping only the end would lose exactly what the judge compares.
754fn 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
768/// The owner's work for the judge: its title (when it has one) and its
769/// instruction with whitespace collapsed, keeping the head and the tail when
770/// it is long. The whole instruction is used, not its first line: a heading
771/// such as `# Task` says nothing about what the work is.
772fn 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
783/// Longest description of one owner's work given to the judge: the same limit
784/// as the new text, so an instruction the judge accepts on one side is not
785/// cut on the other.
786const 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            // A merged or closed pull request is not in flight, even when the
813            // task that owns it is.
814            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
828/// The claims of one run, attributed to `owner` (the task, when there is one).
829///
830/// `live_pr` is false when the record's open pull request is known to be
831/// settled or somebody else's (see [`Staleness`]); it then names no PR.
832fn 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/// Repository identity: the canonical git common dir, so two worktrees of one
885/// repository are the same repository and a path spelled two ways is too.
886#[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
930/// Does `text` contain `branch` as a whole word? `.` and `/` before it are
931/// fine (`origin/magi/x/A`), a name character on either side is not, so
932/// `magi/27b2/A` does not match inside `magi/27b2/AB`.
933fn 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
959/// `#48`, `PR 48`, `PR #48`, `pull request 48`, and `.../pull/48` URLs.
960fn 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
1005/// Hex words of 7..=40 chars in `text` that resolve to a commit in `repo`.
1006fn 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
1031/// Is `sha` reachable from `branch` but not from `base`? A SHA already in the
1032/// base would match every branch there is.
1033fn 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    /// A repo with `main` and a branch `magi/aaaa/A` one commit ahead.
1069    /// Returns the fixture plus the (main, branch) tip SHAs.
1070    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        // Hex that resolves to nothing is ignored.
1171        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        // Changed deliberately: a terminal run's recorded-open PR now claims
1206        // only while the forge says open or cannot be read (`run` passes an
1207        // unreadable forge). A merged / closed answer or a released run no
1208        // longer claims; see the `stale_open_*` tests below.
1209        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        // Editing the task that owns the run never collides with itself.
1257        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        // Closed / unknown / unnamed PRs never match.
1310        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            // Same through an owning, still-live task: the PR is not its claim.
1374            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        // A successor that cannot be read decides nothing: claim stands.
1420        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    /// Run `screen` over `text` with a judge answering `answer`; returns the
1467    /// outcome and how many times the judge was asked.
1468    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}