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            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
286/// The one decision point every caller shares. `hits` is [`check`]'s output:
287/// empty passes without the judge being asked. Otherwise the judge is asked
288/// once (a long text is squeezed to its head and tail first), and only a `duplicate: true` answer refuses (with its reason).
289///
290/// **Fail mode: open.** If the judge cannot answer (agent error, timeout,
291/// quota, unparseable reply, no readable config) the work is let through as [`Screened::Unjudged`] and a
292/// `tracing::warn` is logged; callers also say so on stderr. A mechanical
293/// match cannot tell "continue from PR #28" from a real duplicate, so refusing
294/// whenever the judge is away would bring back exactly the false positives the
295/// judge exists to remove. The cost is that a real duplicate slips through
296/// while no agent is available, which is why it is never silent.
297///
298/// For a review-only request `text` is empty, so `review_branch` is put in
299/// front of the judge as the thing the work is about.
300pub 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
340/// [`screen`] with the production judge: the `[roles] chatter` chain of `cfg`.
341/// No config (`None`) means no judge, so a hit is let through unjudged.
342pub 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/// The slice of `run.json` this module needs. Every field is optional so a
373/// record from any schema still yields what it has.
374#[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    /// The run that took this one's worktree (and pull request) over.
391    #[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    /// An unknown status word counts as unfinished: claiming too much is the
418    /// cheap error here.
419    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
430/// Decides whether the pull request a run's record still calls open is in fact
431/// settled, so a stale record stops claiming it.
432///
433/// A record freezes the last state its land loop polled. A run that was handed
434/// over (`released_to`) or left blocked while another run landed the same pull
435/// request keeps saying `open` forever, so the record alone cannot be trusted
436/// to *keep* a claim alive. Two answers, cheapest first:
437///
438/// 1. a released run does not own the pull request any more: ownership follows
439///    `released_to`, and the successor's own record (read like any other run's)
440///    decides - provided that record is readable;
441/// 2. otherwise the forge is asked, once per number, at most
442///    [`MAX_FORGE_LOOKUPS`] numbers, and never again after one lookup failed
443///    (offline stays fast). Merged or closed means no claim; open, an error or
444///    a skipped lookup means the claim stands. Claiming too much is the cheap
445///    error.
446struct 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
454/// How many pull requests one invocation may put to the forge on behalf of
455/// stale records.
456const 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    /// `true` when `view`'s recorded-open pull request is known not to be this
470    /// run's to claim any more.
471    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/// Something an unfinished piece of work owns.
505#[derive(Debug, Clone)]
506struct Claim {
507    owner: Owner,
508    id: String,
509    status: String,
510    via: String,
511    /// The owner's title and instruction, for the judge.
512    about: String,
513    branch: Option<String>,
514    /// Commit the branch forked from; without it a SHA cannot be judged.
515    base: Option<String>,
516    /// `(number, url)` of an open pull request.
517    pr: Option<(u64, String)>,
518}
519
520/// Look for work already in flight that `text` (and `review_branch`, for a
521/// review-only run) points at. `repo` is the repository the new work is for;
522/// `ignore_task` is a task being edited, whose own claims never count.
523pub 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/// [`check`] with the forge lookup supplied: `open_pr(repo, n)` says whether
544/// pull request `n` of `repo` is open (`Some(url)`) or not / unknown (`None`).
545/// It is asked only about numbers the text names that no local record already
546/// explained, so a PR magi never produced (opened by hand) still collides.
547/// `pr_state(repo, n)` is the forge's word on a pull request a run record calls
548/// open (`None` = unreadable, which keeps the claim); see [`Staleness`].
549#[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    // PRs the edited task's own runs opened: never a rival, forge or not.
570    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        // A terminal run held up only by a pull request that is not its own
597        // (any more) claims nothing at all.
598        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
679/// Ask the forge, via `gh`, whether PR `n` is open. Best effort and bounded:
680/// any failure (no `gh`, no remote, offline, a slow answer) is `None`, i.e.
681/// today's behaviour. `GH_REPO` is dropped so the PR is looked up in `repo`'s
682/// own remote, not whatever the environment points at.
683fn 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
690/// [`gh_open_pr`]'s question asked of a pull request a run record names: its
691/// lifecycle, or `None` when the forge cannot say.
692fn 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
731/// Fill in what a forge-only hit (an open pull request no record here owns)
732/// is about, from its title and body. Only the judge reads it, so this runs
733/// after the mechanical check, only on a hit, and best effort: an unreadable
734/// forge leaves `about` empty, never drops the hit.
735fn 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
752/// Keep the head and the tail of `s` within `max` characters, marking the cut.
753/// A long instruction often has background first and the concrete change last,
754/// so dropping only the end would lose exactly what the judge compares.
755fn 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
769/// The owner's work for the judge: its title (when it has one) and its
770/// instruction with whitespace collapsed, keeping the head and the tail when
771/// it is long. The whole instruction is used, not its first line: a heading
772/// such as `# Task` says nothing about what the work is.
773fn 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
784/// Longest description of one owner's work given to the judge: the same limit
785/// as the new text, so an instruction the judge accepts on one side is not
786/// cut on the other.
787const 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            // A merged or closed pull request is not in flight, even when the
814            // task that owns it is.
815            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
829/// The claims of one run, attributed to `owner` (the task, when there is one).
830///
831/// `live_pr` is false when the record's open pull request is known to be
832/// settled or somebody else's (see [`Staleness`]); it then names no PR.
833fn 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/// Repository identity: the canonical git common dir, so two worktrees of one
886/// repository are the same repository and a path spelled two ways is too.
887#[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
931/// Does `text` contain `branch` as a whole word? `.` and `/` before it are
932/// fine (`origin/magi/x/A`), a name character on either side is not, so
933/// `magi/27b2/A` does not match inside `magi/27b2/AB`.
934fn 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
960/// `#48`, `PR 48`, `PR #48`, `pull request 48`, and `.../pull/48` URLs.
961fn 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
1006/// Hex words of 7..=40 chars in `text` that resolve to a commit in `repo`.
1007fn 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
1032/// Is `sha` reachable from `branch` but not from `base`? A SHA already in the
1033/// base would match every branch there is.
1034fn 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    /// A repo with `main` and a branch `magi/aaaa/A` one commit ahead.
1070    /// Returns the fixture plus the (main, branch) tip SHAs.
1071    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        // Hex that resolves to nothing is ignored.
1172        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        // Changed deliberately: a terminal run's recorded-open PR now claims
1207        // only while the forge says open or cannot be read (`run` passes an
1208        // unreadable forge). A merged / closed answer or a released run no
1209        // longer claims; see the `stale_open_*` tests below.
1210        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        // Editing the task that owns the run never collides with itself.
1258        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        // Closed / unknown / unnamed PRs never match.
1311        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            // Same through an owning, still-live task: the PR is not its claim.
1375            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        // A successor that cannot be read decides nothing: claim stands.
1421        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    /// Run `screen` over `text` with a judge answering `answer`; returns the
1468    /// outcome and how many times the judge was asked.
1469    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}