Skip to main content

spar/
listen.rs

1//! Listening for what happens after a review.
2//!
3//! Every other command does one pass and exits, so a review asked for again
4//! after spar posted one waited for a person to notice and run spar again.
5//! `spar listen` waits instead, and it waits for free: the only thing that
6//! calls an agent is a review, a second look, or a check-in, and none of those
7//! starts until GitHub says something happened.
8//!
9//! A second look is asked for, never assumed: a review requested again or a
10//! mention starts one, and a push or a comment alone does not. The author may
11//! still be working, and a comment may be another reviewer's review rather than
12//! an answer to this one. The look that runs reads every commit and reply since
13//! the review on record.
14//!
15//! Two sources feed it, because neither is enough alone. GitHub's notifications
16//! say why a thread moved, a review request, a mention, a comment, and they are
17//! polled with a conditional request that costs nothing against the rate limit
18//! when nothing changed. But a push raises no notification for a reviewer, so a
19//! slower sweep of the open pull requests compares each head with the commit
20//! spar last read. Notifications are never raised for the viewer's own actions,
21//! which is also why spar's own comments never wake it.
22//!
23//! Events settle before anything runs: a review request followed by three
24//! comments in the same minute is one second look, not four. What runs is
25//! decided by a table that is a pure function of the event, a few facts about
26//! the item, and the `[listen]` settings, so it can be read and tested on its
27//! own.
28
29use std::collections::{BTreeMap, BTreeSet};
30use std::fs::File;
31use std::path::{Path, PathBuf};
32use std::time::Duration;
33
34use serde::{Deserialize, Serialize};
35
36use crate::agent::Agent;
37use crate::answer;
38use crate::checkin;
39use crate::clock;
40use crate::config::{Config, ListenCfg, OnMention, OwnPr, ReviewOrIgnore};
41use crate::error::Result;
42use crate::model::{IssueRun, ItemKind, Login, PrSweepRow, ReviewRequested, Status};
43use crate::proc;
44use crate::repo::{ApiResponse, Repo};
45use crate::review_only::{self, RecheckMode, ReviewOpts};
46use crate::spend;
47use crate::{bail, log, logdim, logwarn, spar_err};
48
49/// Polling never goes below this, whatever the config says.
50const FLOOR_SECS: u64 = 15;
51/// A `seen` entry older than this is forgotten.
52const SEEN_FOR_SECS: u64 = 7 * 86_400;
53/// Most notification pages read in one poll. A poll is all or nothing, so one
54/// with more than this fails, and says how to start from later.
55const MAX_PAGES: usize = 50;
56/// Failed dispatches of one item before it is left alone until it changes.
57const MAX_FAILURES: u32 = 5;
58
59pub const STATE_VERSION: u32 = 1;
60
61/// How one `spar listen` was asked to run.
62#[derive(Debug, Clone, Default)]
63pub struct Opts {
64    /// The issues and pull requests to watch. Empty means the repository.
65    pub items: Vec<i64>,
66    /// One poll and one sweep, act on what is due, exit.
67    pub once: bool,
68    /// Log every event and what it would trigger, and trigger nothing.
69    pub dry_run: bool,
70    /// How far back the first run looks.
71    pub since: Option<Duration>,
72}
73
74// ---------------------------------------------------------------------------
75// What GitHub returns
76// ---------------------------------------------------------------------------
77
78/// One notification thread, as `GET /repos/{owner}/{repo}/notifications`
79/// returns it.
80#[derive(Debug, Clone, Default, Deserialize, PartialEq, Eq)]
81pub struct Thread {
82    #[serde(default)]
83    pub id: String,
84    #[serde(default)]
85    pub reason: String,
86    #[serde(default)]
87    pub updated_at: String,
88    #[serde(default)]
89    pub subject: Subject,
90}
91
92#[derive(Debug, Clone, Default, Deserialize, PartialEq, Eq)]
93pub struct Subject {
94    #[serde(default)]
95    pub title: String,
96    /// The API URL of the issue or pull request; its number is the last path
97    /// segment.
98    #[serde(default)]
99    pub url: Option<String>,
100    #[serde(default)]
101    pub latest_comment_url: Option<String>,
102    /// `PullRequest`, `Issue`, or something spar does not watch.
103    #[serde(default, rename = "type")]
104    pub kind: String,
105}
106
107// ---------------------------------------------------------------------------
108// Events, and what is kept between ticks
109// ---------------------------------------------------------------------------
110
111#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
112#[serde(rename_all = "snake_case")]
113pub enum Trigger {
114    /// The head moved since spar last read it.
115    Pushed,
116    /// Somebody other than the viewer said something.
117    Replied,
118    Mentioned,
119    ReviewRequested,
120    Assigned,
121    Closed,
122}
123
124impl Trigger {
125    /// The trigger as a reason in a sentence.
126    fn noun(self) -> &'static str {
127        match self {
128            Trigger::Pushed => "a push",
129            Trigger::Replied => "a reply",
130            Trigger::Mentioned => "a mention",
131            Trigger::ReviewRequested => "a review request",
132            Trigger::Assigned => "an assignment",
133            Trigger::Closed => "a close",
134        }
135    }
136}
137
138/// One thing that happened to one item.
139#[derive(Debug, Clone, PartialEq, Eq)]
140pub struct Event {
141    pub number: i64,
142    pub kind: ItemKind,
143    pub trigger: Trigger,
144    /// Who did it, when GitHub said.
145    pub who: Option<String>,
146    /// The old and new head, for a push.
147    pub head: Option<(String, String)>,
148    /// New commits since the review, when cheap to know.
149    pub commits: Option<u64>,
150    /// Raised by a notification thread, whose reason is the one the thread
151    /// began with: the trigger says how the thread started, not always what
152    /// just happened on it.
153    pub from_thread: bool,
154}
155
156/// What has happened to one item since spar last acted on it. Persisted, so a
157/// restart in the middle of settling loses nothing: the notification is
158/// already past the watermark and will not come again.
159#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
160pub struct Pending {
161    pub kind: ItemKind,
162    pub first_seen: u64,
163    pub last_seen: u64,
164    pub triggers: BTreeSet<Trigger>,
165    #[serde(default)]
166    pub who: Vec<String>,
167    #[serde(default)]
168    pub head_from: Option<String>,
169    #[serde(default)]
170    pub head_to: Option<String>,
171}
172
173impl Pending {
174    /// "a push and 2 replies", for the line that says why something ran.
175    pub fn because(&self) -> String {
176        let mut parts: Vec<String> = self.triggers.iter().map(|t| t.noun().to_string()).collect();
177        if parts.is_empty() {
178            parts.push("nothing in particular".into());
179        }
180        match parts.len() {
181            1 => parts.remove(0),
182            2 => format!("{} and {}", parts[0], parts[1]),
183            _ => {
184                let last = parts.pop().unwrap_or_default();
185                format!("{}, and {last}", parts.join(", "))
186            }
187        }
188    }
189}
190
191/// What the listener knows about one item it is watching.
192#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
193pub struct Watched {
194    pub kind: ItemKind,
195    #[serde(default)]
196    pub title: String,
197    #[serde(default)]
198    pub author: String,
199    /// The head the listener last saw. A sweep raises Pushed only when this
200    /// changes, so a commit spar itself just read is not news.
201    #[serde(default)]
202    pub head: Option<String>,
203    #[serde(default)]
204    pub requested_noticed: bool,
205    #[serde(default)]
206    pub assigned_noticed: bool,
207    #[serde(default)]
208    pub failures: u32,
209    #[serde(default)]
210    pub retry_after: u64,
211    #[serde(default)]
212    pub said_capped: bool,
213    #[serde(default)]
214    pub said_resume: bool,
215    #[serde(default)]
216    pub last_dispatched: Option<u64>,
217    /// What a pending item was waiting on when the listener gave up on it.
218    /// Giving up waits for the item to change, and on a pull request spar has
219    /// not reviewed that change is often a push alone, which leads nowhere, so
220    /// the reasons come back with it.
221    #[serde(default)]
222    pub held: BTreeSet<Trigger>,
223}
224
225impl Watched {
226    fn new(kind: ItemKind) -> Self {
227        Self {
228            kind,
229            title: String::new(),
230            author: String::new(),
231            head: None,
232            requested_noticed: false,
233            assigned_noticed: false,
234            failures: 0,
235            retry_after: 0,
236            said_capped: false,
237            said_resume: false,
238            last_dispatched: None,
239            held: BTreeSet::new(),
240        }
241    }
242}
243
244/// Everything the listener keeps between ticks and between runs, in
245/// `.spar/state/listen.json`.
246#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
247pub struct ListenState {
248    pub version: u32,
249    /// Threads updated after this are new. RFC 3339, the shape GitHub takes.
250    #[serde(default)]
251    pub watermark: String,
252    /// `Last-Modified` of the last poll that answered, sent back as
253    /// `If-Modified-Since` so a poll that finds nothing costs nothing.
254    #[serde(default)]
255    pub last_modified: Option<String>,
256    /// Thread id to the `updated_at` it was last seen with, so one thread
257    /// returned twice is one event.
258    #[serde(default)]
259    pub seen: BTreeMap<String, String>,
260    #[serde(default)]
261    pub items: BTreeMap<i64, Watched>,
262    #[serde(default)]
263    pub pending: BTreeMap<i64, Pending>,
264}
265
266impl ListenState {
267    fn watched(&mut self, number: i64, kind: ItemKind) -> &mut Watched {
268        self.items
269            .entry(number)
270            .or_insert_with(|| Watched::new(kind))
271    }
272
273    /// Fold an event into what is pending for its item, with whatever the
274    /// item was waiting on when the listener last gave up on it.
275    fn note(&mut self, event: &Event, now: u64) {
276        let held = std::mem::take(&mut self.watched(event.number, event.kind).held);
277        merge(&mut self.pending, event, now);
278        if let Some(p) = self.pending.get_mut(&event.number) {
279            p.triggers.extend(held);
280        }
281    }
282
283    /// Stop trying an item until it changes again, holding what it was
284    /// waiting on. A requested first review that GitHub's lag behind a push
285    /// kept stopping is still requested when that push shows.
286    fn give_up(&mut self, number: i64, kind: ItemKind) {
287        if let Some(p) = self.pending.remove(&number) {
288            self.watched(number, kind).held.extend(p.triggers);
289        }
290    }
291}
292
293// ---------------------------------------------------------------------------
294// Pure pieces
295// ---------------------------------------------------------------------------
296
297/// `30m`, `2h`, `1d`, `1h30m`.
298pub fn parse_duration(text: &str) -> Result<Duration> {
299    let bad = || {
300        spar_err!(
301            "`--since` takes a number and a unit, like 30m, 2h or 1d, not {:?}",
302            text.trim()
303        )
304    };
305    let mut total = 0u64;
306    let mut digits = String::new();
307    let mut any = false;
308    for c in text.trim().chars() {
309        if c.is_ascii_digit() {
310            digits.push(c);
311            continue;
312        }
313        let n: u64 = digits.parse().map_err(|_| bad())?;
314        digits.clear();
315        let unit = match c {
316            's' => 1,
317            'm' => 60,
318            'h' => 3600,
319            'd' => 86_400,
320            _ => return Err(bad()),
321        };
322        total = total.saturating_add(n.saturating_mul(unit));
323        any = true;
324    }
325    if !digits.is_empty() || !any {
326        return Err(bad());
327    }
328    Ok(Duration::from_secs(total))
329}
330
331/// How long to wait before the next poll: the config, what GitHub asked for,
332/// and the floor, whichever is largest.
333pub fn poll_interval(cfg: &ListenCfg, header: Option<&str>) -> u64 {
334    let asked = header
335        .and_then(|h| h.trim().parse::<u64>().ok())
336        .unwrap_or(0);
337    cfg.interval.max(asked).max(FLOOR_SECS)
338}
339
340/// How long a rate limited answer says to wait, clamped to a minute and an
341/// hour, or None when the answer was not about the rate limit.
342pub fn rate_limit_wait(resp: &ApiResponse, now: u64) -> Option<u64> {
343    if resp.status != 403 && resp.status != 429 {
344        return None;
345    }
346    let retry_after = resp
347        .headers
348        .get("retry-after")
349        .and_then(|v| v.trim().parse::<u64>().ok());
350    let exhausted = resp
351        .headers
352        .get("x-ratelimit-remaining")
353        .is_some_and(|v| v.trim() == "0");
354    let reset = resp
355        .headers
356        .get("x-ratelimit-reset")
357        .and_then(|v| v.trim().parse::<u64>().ok())
358        .map(|at| at.saturating_sub(now));
359    let wait = match (retry_after, exhausted, reset) {
360        (Some(secs), _, _) => secs,
361        (None, true, Some(secs)) => secs,
362        _ => return None,
363    };
364    Some(wait.clamp(60, 3600))
365}
366
367/// The body's `message`, or the status line, for a failure that is not a
368/// rate limit.
369fn status_message(resp: &ApiResponse) -> String {
370    serde_json::from_str::<serde_json::Value>(&resp.body)
371        .ok()
372        .and_then(|v| {
373            v.get("message")
374                .and_then(|m| m.as_str())
375                .map(|m| format!("HTTP {}: {m}", resp.status))
376        })
377        .unwrap_or_else(|| format!("HTTP {}", resp.status))
378}
379
380/// The URL of the next page, from a `Link` header.
381pub fn next_link(headers: &BTreeMap<String, String>) -> Option<String> {
382    let link = headers.get("link")?;
383    link.split(',').find_map(|part| {
384        let (url, rel) = part.split_once(';')?;
385        if !rel.contains("rel=\"next\"") {
386            return None;
387        }
388        let url = url.trim().trim_start_matches('<').trim_end_matches('>');
389        (!url.is_empty()).then(|| url.to_string())
390    })
391}
392
393/// The number at the end of an issue or pull request API URL.
394pub fn number_from_url(url: &str) -> Option<i64> {
395    url.trim_end_matches('/')
396        .rsplit('/')
397        .next()
398        .and_then(|last| last.parse().ok())
399}
400
401/// Turn a poll's threads into events, skipping what was already seen, and
402/// advance the watermark past them.
403pub fn events_from_threads(threads: &[Thread], state: &mut ListenState) -> Vec<Event> {
404    let mut out = Vec::new();
405    for thread in threads {
406        if state.seen.get(&thread.id) == Some(&thread.updated_at) {
407            continue;
408        }
409        if !thread.updated_at.is_empty() {
410            state
411                .seen
412                .insert(thread.id.clone(), thread.updated_at.clone());
413            if thread.updated_at > state.watermark {
414                state.watermark = thread.updated_at.clone();
415            }
416        }
417        let kind = match thread.subject.kind.as_str() {
418            "PullRequest" => ItemKind::Pr,
419            "Issue" => ItemKind::Issue,
420            _ => continue,
421        };
422        let Some(number) = thread.subject.url.as_deref().and_then(number_from_url) else {
423            continue;
424        };
425        let trigger = match thread.reason.as_str() {
426            "review_requested" => Trigger::ReviewRequested,
427            "assign" => Trigger::Assigned,
428            "mention" | "team_mention" => Trigger::Mentioned,
429            // comment, author, manual, state_change: somebody did something
430            // on a thread the viewer is part of. The facts call learns whether
431            // it closed.
432            _ => Trigger::Replied,
433        };
434        out.push(Event {
435            number,
436            kind,
437            trigger,
438            who: None,
439            head: None,
440            commits: None,
441            from_thread: true,
442        });
443    }
444    out
445}
446
447/// What a sweep of the open pull requests says happened.
448///
449/// `record_head` answers with the commit a posted review record holds, so a
450/// pull request seen for the first time whose head already moved since its
451/// review is a pending second look rather than nothing.
452pub fn events_from_sweep(
453    rows: &[PrSweepRow],
454    viewer: &str,
455    state: &mut ListenState,
456    named: &BTreeSet<i64>,
457    min_number: i64,
458    record_head: &dyn Fn(i64) -> Option<String>,
459    repo_mode: bool,
460) -> Vec<Event> {
461    let mut out = Vec::new();
462    let mut present: BTreeSet<i64> = BTreeSet::new();
463    for row in rows {
464        // Present before the floor is applied: a pull request below it is
465        // still open, and one a notification put on the watch is not closed
466        // just because the sweep does not act on it.
467        present.insert(row.number);
468        if row.number < min_number && !named.contains(&row.number) {
469            continue;
470        }
471        let watched = state.watched(row.number, ItemKind::Pr);
472        if !row.title.is_empty() {
473            watched.title = row.title.clone();
474        }
475        if let Some(author) = &row.author {
476            watched.author = author.login.clone();
477        }
478        if !row.state.eq_ignore_ascii_case("OPEN") {
479            out.push(Event {
480                number: row.number,
481                kind: ItemKind::Pr,
482                trigger: Trigger::Closed,
483                who: None,
484                head: None,
485                commits: None,
486                from_thread: false,
487            });
488            continue;
489        }
490        let from = match &watched.head {
491            Some(old) => Some(old.clone()),
492            None => record_head(row.number),
493        };
494        if let Some(old) = from {
495            if !row.head_ref_oid.is_empty() && !old.eq_ignore_ascii_case(&row.head_ref_oid) {
496                out.push(Event {
497                    number: row.number,
498                    kind: ItemKind::Pr,
499                    trigger: Trigger::Pushed,
500                    who: None,
501                    head: Some((old, row.head_ref_oid.clone())),
502                    commits: None,
503                    from_thread: false,
504                });
505            }
506        }
507        if !row.head_ref_oid.is_empty() {
508            watched.head = Some(row.head_ref_oid.clone());
509        }
510        let asked = row
511            .review_requests
512            .iter()
513            .any(|l| crate::comments::same_login(&l.login, viewer));
514        // A formal review clears a request, so a request seen again later is
515        // the author asking again and is news. A review posted as a comment
516        // leaves it open, and the dispatch tells that one apart by when it was
517        // made.
518        if !asked {
519            watched.requested_noticed = false;
520        } else if !watched.requested_noticed {
521            watched.requested_noticed = true;
522            out.push(Event {
523                number: row.number,
524                kind: ItemKind::Pr,
525                trigger: Trigger::ReviewRequested,
526                who: None,
527                head: None,
528                commits: None,
529                from_thread: false,
530            });
531        }
532        let assigned = row
533            .assignees
534            .iter()
535            .any(|l| crate::comments::same_login(&l.login, viewer));
536        if assigned && !watched.assigned_noticed {
537            watched.assigned_noticed = true;
538            out.push(Event {
539                number: row.number,
540                kind: ItemKind::Pr,
541                trigger: Trigger::Assigned,
542                who: None,
543                head: None,
544                commits: None,
545                from_thread: false,
546            });
547        }
548    }
549    if repo_mode {
550        // A pull request that was open last time and is not in the list now
551        // has closed or merged.
552        let gone: Vec<i64> = state
553            .items
554            .iter()
555            .filter(|(n, w)| w.kind == ItemKind::Pr && !present.contains(n))
556            .map(|(n, _)| *n)
557            .collect();
558        for number in gone {
559            out.push(Event {
560                number,
561                kind: ItemKind::Pr,
562                trigger: Trigger::Closed,
563                who: None,
564                head: None,
565                commits: None,
566                from_thread: false,
567            });
568        }
569    }
570    out
571}
572
573/// Fold one event into what is pending for its item.
574pub fn merge(pending: &mut BTreeMap<i64, Pending>, event: &Event, now: u64) {
575    let p = pending.entry(event.number).or_insert_with(|| Pending {
576        kind: event.kind,
577        first_seen: now,
578        last_seen: now,
579        triggers: BTreeSet::new(),
580        who: Vec::new(),
581        head_from: None,
582        head_to: None,
583    });
584    p.last_seen = now;
585    p.triggers.insert(event.trigger);
586    if let Some(who) = &event.who {
587        if !p.who.contains(who) {
588            p.who.push(who.clone());
589        }
590    }
591    if let Some((from, to)) = &event.head {
592        if p.head_from.is_none() {
593            p.head_from = Some(from.clone());
594        }
595        p.head_to = Some(to.clone());
596    }
597}
598
599/// Whether an item has settled: quiet for `settle` seconds, or pending for
600/// `max_wait` however noisy, or `--once`.
601pub fn due(p: &Pending, now: u64, cfg: &ListenCfg, once: bool) -> bool {
602    once || now.saturating_sub(p.last_seen) >= cfg.settle
603        || now.saturating_sub(p.first_seen) >= cfg.max_wait
604}
605
606/// When the earliest pending item this run acts on becomes due and is past
607/// any wait a failure put on it. Leaving the wait out woke the loop every
608/// second until it ended.
609pub fn next_due(
610    state: &ListenState,
611    cfg: &ListenCfg,
612    acts_on: &dyn Fn(i64) -> bool,
613) -> Option<u64> {
614    state
615        .pending
616        .iter()
617        .filter(|(n, _)| acts_on(**n))
618        .map(|(n, p)| {
619            let settled = (p.last_seen + cfg.settle).min(p.first_seen + cfg.max_wait);
620            state
621                .items
622                .get(n)
623                .map_or(settled, |w| settled.max(w.retry_after))
624        })
625        .min()
626}
627
628/// What the dispatch table decided, before anything runs.
629#[derive(Debug, Clone, PartialEq, Eq)]
630pub enum Action {
631    /// A second look at this pull request. An issue routes to its pull request.
632    Recheck(i64),
633    /// A first look.
634    Review,
635    /// `spar checkin`, with or without the power to push.
636    Checkin { reply_only: bool },
637    /// `spar answer` on an issue, posting what it may, or saving it as a
638    /// draft for `spar post`.
639    Answer { draft: bool },
640    /// This issue was answered as often as the config allows; a person runs
641    /// `spar answer`.
642    AnswersCapped,
643    /// This pull request was rechecked as often as the config allows; a person
644    /// runs `spar review`. An issue names its pull request.
645    Capped(i64),
646    /// Custody state and no review record: `spar resume` owns it.
647    BelongsToResume,
648    /// Closed or merged: forget it.
649    Drop,
650    /// Nothing, with the reason.
651    Ignore(&'static str),
652}
653
654impl Action {
655    /// What the log says is about to happen to item `number`.
656    pub fn phrase(&self, number: i64) -> String {
657        match self {
658            Action::Recheck(pr) if *pr == number => "take a second look".to_string(),
659            Action::Recheck(pr) => format!("take a second look at #{pr}"),
660            Action::Review => "review it".to_string(),
661            Action::Checkin { reply_only: false } => "check in on it".to_string(),
662            Action::Checkin { reply_only: true } => "answer the comments on it".to_string(),
663            Action::Answer { draft: false } => "answer what it asks of you".to_string(),
664            Action::Answer { draft: true } => "draft an answer to what it asks of you".to_string(),
665            Action::Capped(_) | Action::AnswersCapped | Action::BelongsToResume => {
666                "leave it".to_string()
667            }
668            Action::Drop => "drop it".to_string(),
669            Action::Ignore(_) => "ignore it".to_string(),
670        }
671    }
672
673    fn runs(&self) -> bool {
674        matches!(
675            self,
676            Action::Recheck(_) | Action::Review | Action::Checkin { .. } | Action::Answer { .. }
677        )
678    }
679}
680
681/// What `plan` needs to know about an item beyond what is pending. Read once
682/// per dispatch, never per tick.
683#[derive(Debug, Clone, Default, PartialEq, Eq)]
684pub struct Facts {
685    pub number: i64,
686    pub named: bool,
687    pub open: bool,
688    /// For an issue, this and `recheck_capped` are about its linked pull
689    /// request, since that is what a second look on its behalf would be at.
690    pub author_is_viewer: bool,
691    /// A review record the author has seen.
692    pub has_posted_record: bool,
693    pub recheck_capped: bool,
694    /// The custody loop's `pr-N.json`.
695    pub has_custody_state: bool,
696    pub below_floor: bool,
697    /// For an issue: its open pull request, when spar reviewed that.
698    pub linked_pr_with_record: Option<i64>,
699    /// The pull request asks the viewer for a review right now, by name or
700    /// through a team the viewer is in.
701    pub review_pending: bool,
702    /// That request was made before the posted review began, so the review
703    /// answered it.
704    pub request_answered: bool,
705    /// Somebody, not a bot, @mentioned the viewer since the review on record,
706    /// or for an issue since its pull request's. Read only when a mention is
707    /// pending.
708    pub mentioned: bool,
709    /// For an issue with no reviewed pull request: a mention of the viewer on
710    /// it that `spar answer` has not judged. Read only when a mention is
711    /// pending.
712    pub asks_open: bool,
713    /// For the same issue: `spar answer` ran on it `max_answers` times
714    /// without a person running it since.
715    pub answers_capped: bool,
716}
717
718/// Whether a pull request's `reviewRequests` ask the viewer now: by name, or
719/// through a team `in_team` says the viewer is in. A team nobody could check
720/// counts as asking, which is how every team request used to be read.
721pub fn review_pending(
722    requests: &[Login],
723    viewer: &str,
724    in_team: &dyn Fn(&str) -> Option<bool>,
725) -> bool {
726    requests
727        .iter()
728        .any(|r| crate::comments::same_login(&r.login, viewer))
729        || requests
730            .iter()
731            .filter(|r| r.login.is_empty())
732            .any(|team| in_team(&team.slug).unwrap_or(true))
733}
734
735/// Whether the review the viewer is asked for is already on record: the newest
736/// request that counts for the viewer, as `review_pending` counts one, was made
737/// no later than the posted review began at `reviewed_at`.
738///
739/// A review posted as a comment leaves its request open, and a first sweep has
740/// never seen the request before, so without this a request the review already
741/// answered reads as the author asking again. False when no request counts or
742/// a time cannot be read: answered takes the timeline saying so.
743pub fn request_answered(
744    made: &[ReviewRequested],
745    viewer: &str,
746    in_team: &dyn Fn(&str) -> Option<bool>,
747    reviewed_at: &str,
748) -> bool {
749    let Some(reviewed) = clock::parse_rfc3339(reviewed_at) else {
750        return false;
751    };
752    made.iter()
753        .filter(|r| review_pending(std::slice::from_ref(&r.who), viewer, in_team))
754        .filter_map(|r| clock::parse_rfc3339(&r.at))
755        .max()
756        .is_some_and(|newest| newest <= reviewed)
757}
758
759/// A notification's reason is the one its thread was first raised for and it
760/// never changes, so a thread that began as a review request says so on every
761/// later update, spar's own review among them. What the pull request says now
762/// decides: when nobody is asking any more, or the request is older than the
763/// review on record and so was answered by it, the update is only something
764/// happening on a thread the viewer is part of. True when it was.
765pub fn settle(p: &mut Pending, f: &Facts) -> bool {
766    let stale = p.kind == ItemKind::Pr
767        && (!f.review_pending || f.request_answered)
768        && p.triggers.remove(&Trigger::ReviewRequested);
769    if stale {
770        p.triggers.insert(Trigger::Replied);
771    }
772    stale
773}
774
775/// A mention is a reason a thread keeps too, so a thread that began with one
776/// says so on every later update. On a pull request spar reviewed, or an issue
777/// whose pull request it reviewed, only a mention made since that review asks
778/// for a second look; any other is only something happening on a thread the
779/// viewer is part of. On any other issue only a mention `spar answer` has not
780/// judged asks anything, which is what keeps spar's own answer, moving the
781/// thread it was raised on, from asking for another. True when it was.
782pub fn settle_mention(p: &mut Pending, f: &Facts) -> bool {
783    let asked = match p.kind {
784        ItemKind::Pr => !f.has_posted_record || f.mentioned,
785        ItemKind::Issue if f.linked_pr_with_record.is_some() => f.mentioned,
786        ItemKind::Issue => f.asks_open,
787    };
788    let stale = !asked && p.triggers.remove(&Trigger::Mentioned);
789    if stale {
790        p.triggers.insert(Trigger::Replied);
791    }
792    stale
793}
794
795/// The dispatch table.
796pub fn plan(p: &Pending, f: &Facts, cfg: &ListenCfg) -> Action {
797    use Trigger::*;
798    if !f.open || p.triggers.contains(&Closed) {
799        return Action::Drop;
800    }
801    if f.below_floor && !f.named {
802        return Action::Ignore("it is below min_number and was not named");
803    }
804    match p.kind {
805        ItemKind::Issue => {
806            // Only a mention asks. A reply is not a request, and an issue
807            // thread that began with an assignment says so on every later
808            // comment.
809            if !p.triggers.contains(&Mentioned) {
810                return Action::Ignore("nothing on the issue asked for spar");
811            }
812            // The pull request's own guards hold whichever way its second look
813            // was asked for, or a mention on the issue would be a way round them.
814            match f.linked_pr_with_record {
815                Some(_) if f.author_is_viewer => Action::Ignore("its pull request is your own"),
816                Some(pr) if f.recheck_capped => Action::Capped(pr),
817                Some(pr) => Action::Recheck(pr),
818                // What a mention asks on an issue is answered on the issue,
819                // and only what nobody has answered: the thread moving under
820                // spar's own reply is not a question.
821                None => match cfg.on_mention {
822                    OnMention::Ignore => Action::Ignore("on_mention is ignore"),
823                    _ if !f.asks_open => Action::Ignore("a mention spar has already judged"),
824                    _ if f.answers_capped => Action::AnswersCapped,
825                    OnMention::Answer => Action::Answer { draft: false },
826                    OnMention::Draft => Action::Answer { draft: true },
827                },
828            }
829        }
830        ItemKind::Pr => {
831            // The viewer's own pull request first: it outranks a review
832            // record, so a `spar review` somebody ran on their own branch does
833            // not make every push a second look the same account cannot
834            // approve.
835            if f.author_is_viewer {
836                let spoken_to = p
837                    .triggers
838                    .iter()
839                    .any(|t| matches!(t, Replied | Mentioned | ReviewRequested));
840                if !spoken_to {
841                    return Action::Ignore("a push to your own pull request");
842                }
843                return match cfg.on_own_pr {
844                    OwnPr::Checkin => Action::Checkin { reply_only: false },
845                    OwnPr::Reply => Action::Checkin { reply_only: true },
846                    OwnPr::Ignore => Action::Ignore("on_own_pr is ignore"),
847                };
848            }
849            if f.has_posted_record {
850                // A second look is asked for, never assumed. A push may be
851                // work in progress, and a comment may be another reviewer's
852                // review rather than an answer to this one.
853                let asked =
854                    p.triggers.contains(&ReviewRequested) || p.triggers.contains(&Mentioned);
855                if !asked {
856                    return Action::Ignore(
857                        "what is neither a review request nor a mention, the only things that \
858                         ask for a second look",
859                    );
860                }
861                return if f.recheck_capped {
862                    Action::Capped(f.number)
863                } else {
864                    Action::Recheck(f.number)
865                };
866            }
867            if f.has_custody_state {
868                return Action::BelongsToResume;
869            }
870            if f.named {
871                return Action::Review;
872            }
873            if p.triggers.contains(&ReviewRequested) {
874                return match cfg.on_review_requested {
875                    ReviewOrIgnore::Review => Action::Review,
876                    ReviewOrIgnore::Ignore => Action::Ignore("on_review_requested is ignore"),
877                };
878            }
879            if p.triggers.contains(&Assigned) {
880                return match cfg.on_assigned {
881                    ReviewOrIgnore::Review => Action::Review,
882                    ReviewOrIgnore::Ignore => Action::Ignore("on_assigned is ignore"),
883                };
884            }
885            if p.triggers.contains(&Mentioned) {
886                return Action::Review;
887            }
888            Action::Ignore("a push or a comment on a pull request spar has not reviewed")
889        }
890    }
891}
892
893/// Forget threads not seen for a week.
894pub fn prune_seen(state: &mut ListenState, now: u64) {
895    state.seen.retain(|_, at| {
896        clock::parse_rfc3339(at).is_none_or(|t| now.saturating_sub(t) < SEEN_FOR_SECS)
897    });
898}
899
900/// A minute, doubling, to a quarter of an hour.
901pub fn backoff_secs(failures: u32) -> u64 {
902    (60u64 << failures.saturating_sub(1).min(10)).min(900)
903}
904
905/// A quarter of an hour, doubling, to six hours.
906pub fn dispatch_backoff_secs(failures: u32) -> u64 {
907    (900u64 << failures.saturating_sub(1).min(10)).min(6 * 3600)
908}
909
910/// How long an item whose dispatch did not finish waits to run again. A head
911/// that moved under it is GitHub or the author rather than a failure, so the
912/// next look starts once the push settles. Anything else backs off.
913pub fn retry_wait(moved: bool, failures: u32, cfg: &ListenCfg) -> u64 {
914    if moved {
915        cfg.settle.max(FLOOR_SECS)
916    } else {
917        dispatch_backoff_secs(failures)
918    }
919}
920
921/// What a `gh` failure is retried after, said loudly once and quietly after.
922#[derive(Debug, Default)]
923struct Backoff {
924    failures: u32,
925    said: bool,
926}
927
928impl Backoff {
929    fn failed(&mut self, what: &str, why: &str) -> u64 {
930        self.failures += 1;
931        let wait = backoff_secs(self.failures);
932        let line = format!(
933            "could not {what}: {why}. Trying again in {}.",
934            spend::clock(Duration::from_secs(wait))
935        );
936        if self.said {
937            logdim!("{line}");
938        } else {
939            logwarn!("{line}");
940            self.said = true;
941        }
942        wait
943    }
944
945    fn recovered(&mut self) {
946        if self.said {
947            log!("gh is answering again");
948        }
949        *self = Self::default();
950    }
951}
952
953// ---------------------------------------------------------------------------
954// The lock
955// ---------------------------------------------------------------------------
956
957/// Who holds the lock, for the message a second listener prints. The lock
958/// itself is not in here.
959#[derive(Debug, Serialize, Deserialize)]
960struct LockFile {
961    pid: u32,
962    started: String,
963}
964
965/// One listener per checkout. Two would read the same events, run the same
966/// reviews twice, and rebuild the worktree the other's agents are reading.
967///
968/// On unix the lock is the operating system's, on the open file, rather than
969/// the file being there. It is held exactly as long as the process lives,
970/// however long one review takes, so nothing has to keep it fresh, and a
971/// listener that died holds nothing, even after a reboot hands its pid to
972/// something else.
973#[derive(Debug)]
974struct Lock {
975    path: PathBuf,
976    /// Open for the lock's lifetime: closing it is what releases the lock.
977    _file: File,
978}
979
980impl Lock {
981    fn acquire(repo: &Repo, now: u64) -> Result<Lock> {
982        Lock::at(repo.listen_lock_path(), now)
983    }
984
985    fn at(path: PathBuf, now: u64) -> Result<Lock> {
986        if let Some(parent) = path.parent() {
987            std::fs::create_dir_all(parent)?;
988        }
989        // A listener that exits removes the file while it still holds it, so
990        // the file opened here can be one that was unlinked a moment ago.
991        // Locking that would lock a file nobody else can find, hence the check
992        // that the path still names the file locked, and another try if not.
993        for _ in 0..3 {
994            let Some(mut file) = open_locked(&path)? else {
995                let pid = std::fs::read_to_string(&path)
996                    .ok()
997                    .and_then(|text| serde_json::from_str::<LockFile>(&text).ok())
998                    .map(|h| h.pid.to_string())
999                    .unwrap_or_else(|| "unknown".into());
1000                bail!(
1001                    "another `spar listen` (pid {pid}) is already watching this checkout. \
1002                     Stop it first.{}",
1003                    if cfg!(unix) {
1004                        String::new()
1005                    } else {
1006                        format!(" If that process is gone, delete {}.", path.display())
1007                    }
1008                );
1009            };
1010            if !same_file(&file, &path) {
1011                continue;
1012            }
1013            file.set_len(0)?;
1014            serde_json::to_writer_pretty(
1015                &mut file,
1016                &LockFile {
1017                    pid: std::process::id(),
1018                    started: clock::rfc3339(now),
1019                },
1020            )?;
1021            return Ok(Lock { path, _file: file });
1022        }
1023        bail!("could not take the listen lock at {}", path.display())
1024    }
1025}
1026
1027impl Drop for Lock {
1028    fn drop(&mut self) {
1029        // Removed while still held, which is what `same_file` relies on. The
1030        // file closes, and the lock goes with it, after this returns.
1031        let _ = std::fs::remove_file(&self.path);
1032    }
1033}
1034
1035/// The lock file, opened and locked, or None when another process holds it.
1036#[cfg(unix)]
1037fn open_locked(path: &Path) -> Result<Option<File>> {
1038    use std::os::unix::io::AsRawFd;
1039    let file = std::fs::OpenOptions::new()
1040        .read(true)
1041        .write(true)
1042        .create(true)
1043        .truncate(false)
1044        .open(path)?;
1045    if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0 {
1046        return Ok(Some(file));
1047    }
1048    let e = std::io::Error::last_os_error();
1049    if e.raw_os_error() == Some(libc::EWOULDBLOCK) {
1050        Ok(None)
1051    } else {
1052        Err(e.into())
1053    }
1054}
1055
1056/// Without an advisory lock to hand, the file being there is the lock, and one
1057/// left by a listener that died has to be deleted by hand.
1058#[cfg(not(unix))]
1059fn open_locked(path: &Path) -> Result<Option<File>> {
1060    match std::fs::OpenOptions::new()
1061        .write(true)
1062        .create_new(true)
1063        .open(path)
1064    {
1065        Ok(file) => Ok(Some(file)),
1066        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => Ok(None),
1067        Err(e) => Err(e.into()),
1068    }
1069}
1070
1071#[cfg(unix)]
1072fn same_file(file: &File, path: &Path) -> bool {
1073    use std::os::unix::fs::MetadataExt;
1074    match (file.metadata(), std::fs::metadata(path)) {
1075        (Ok(open), Ok(named)) => open.dev() == named.dev() && open.ino() == named.ino(),
1076        _ => false,
1077    }
1078}
1079
1080#[cfg(not(unix))]
1081fn same_file(_file: &File, _path: &Path) -> bool {
1082    true
1083}
1084
1085// ---------------------------------------------------------------------------
1086// The two sources
1087// ---------------------------------------------------------------------------
1088
1089enum Polled {
1090    /// Nothing new, poll again after this many seconds.
1091    Nothing(u64),
1092    Threads(Vec<Thread>, u64),
1093    /// Rate limited, wait this long.
1094    RateLimited(u64),
1095    Failed(String),
1096}
1097
1098fn poll(repo: &Repo, cfg: &ListenCfg, state: &mut ListenState, now: u64) -> Polled {
1099    let endpoint = format!(
1100        "repos/{{owner}}/{{repo}}/notifications?participating=true&all=true&per_page=100&since={}",
1101        state.watermark
1102    );
1103    let last_modified = state.last_modified.clone();
1104    let headers: Vec<(&str, &str)> = last_modified
1105        .as_deref()
1106        .map(|lm| vec![("If-Modified-Since", lm)])
1107        .unwrap_or_default();
1108    let resp = match repo.gh_api_raw(&endpoint, &headers) {
1109        Ok(resp) => resp,
1110        Err(e) => return Polled::Failed(e.last_line().to_string()),
1111    };
1112    let interval = poll_interval(cfg, resp.headers.get("x-poll-interval").map(String::as_str));
1113    match resp.status {
1114        304 => Polled::Nothing(interval),
1115        200 => match every_page(repo, &resp, &state.watermark, now) {
1116            Ok(threads) => {
1117                // Only once every page is in, or the next poll's 304 would
1118                // answer for pages this one never read.
1119                state.last_modified = resp
1120                    .headers
1121                    .get("last-modified")
1122                    .or_else(|| resp.headers.get("date"))
1123                    .cloned();
1124                Polled::Threads(threads, interval)
1125            }
1126            Err(failed) => failed,
1127        },
1128        401 => Polled::Failed("gh is not authenticated; run `gh auth status`".into()),
1129        _ => refused(&resp, now),
1130    }
1131}
1132
1133/// What a failed answer means: a wait for a rate limit, a failure otherwise.
1134fn refused(resp: &ApiResponse, now: u64) -> Polled {
1135    match rate_limit_wait(resp, now) {
1136        Some(wait) => Polled::RateLimited(wait),
1137        None => Polled::Failed(status_message(resp)),
1138    }
1139}
1140
1141/// Every page of one poll, or what stopped it.
1142///
1143/// All or nothing. Notifications come newest first, so a page that was not
1144/// read holds threads older than every one that was, and moving the watermark
1145/// past the ones read would lose the rest for good. A poll that cannot read
1146/// them all changes nothing and is tried again.
1147fn every_page(
1148    repo: &Repo,
1149    first: &ApiResponse,
1150    watermark: &str,
1151    now: u64,
1152) -> std::result::Result<Vec<Thread>, Polled> {
1153    let unexpected =
1154        |e: serde_json::Error| Polled::Failed(format!("unexpected notification data: {e}"));
1155    let mut threads: Vec<Thread> = serde_json::from_str(&first.body).map_err(unexpected)?;
1156    let mut next = next_link(&first.headers);
1157    let mut pages = 1;
1158    while let Some(url) = next.take() {
1159        if pages >= MAX_PAGES {
1160            return Err(Polled::Failed(format!(
1161                "more than {MAX_PAGES} pages of notifications since {watermark}. Run once with \
1162                 --since, like --since 1d, to start from later and skip the rest"
1163            )));
1164        }
1165        let more = repo
1166            .gh_api_raw(&url, &[])
1167            .map_err(|e| Polled::Failed(e.last_line().to_string()))?;
1168        if more.status != 200 {
1169            return Err(refused(&more, now));
1170        }
1171        threads.extend(serde_json::from_str::<Vec<Thread>>(&more.body).map_err(unexpected)?);
1172        next = next_link(&more.headers);
1173        pages += 1;
1174    }
1175    Ok(threads)
1176}
1177
1178/// What one sweep read.
1179#[derive(Default)]
1180struct Swept {
1181    rows: Vec<PrSweepRow>,
1182    /// Named issues that are no longer open.
1183    closed_issues: Vec<i64>,
1184}
1185
1186/// The rows a sweep looks at: every open pull request, or the named ones
1187/// still watched.
1188fn sweep_rows(
1189    repo: &Repo,
1190    cfg: &Config,
1191    repo_mode: bool,
1192    named_prs: &[i64],
1193    named_issues: &[i64],
1194    state: &ListenState,
1195) -> Result<Swept> {
1196    if repo_mode {
1197        return Ok(Swept {
1198            rows: repo.sweep_open_prs()?,
1199            closed_issues: Vec::new(),
1200        });
1201    }
1202    let watched = |n: &&i64| state.items.contains_key(*n);
1203    let mut swept = Swept::default();
1204    for number in named_prs.iter().filter(watched) {
1205        swept.rows.push(repo.sweep_pr(*number)?);
1206    }
1207    for issue in named_issues.iter().filter(watched) {
1208        // Read here, because a notification is not a promise: an issue the
1209        // viewer never took part in can close without one, and the listener
1210        // would wait for it forever.
1211        if !repo.read_issue(*issue)?.state.eq_ignore_ascii_case("OPEN") {
1212            swept.closed_issues.push(*issue);
1213            continue;
1214        }
1215        // A named issue's open pull request joins the watch once spar has
1216        // reviewed it, so a review requested again there is a second look.
1217        if let Ok(Some(pr)) = repo.try_open_pr_for_issue(*issue, cfg.base_branch()) {
1218            if repo.read_review_record(pr.number).is_some_and(|r| r.posted) {
1219                swept.rows.push(repo.sweep_pr(pr.number)?);
1220            }
1221        }
1222    }
1223    Ok(swept)
1224}
1225
1226/// `mention` says a mention is pending, which is the only time the comments are
1227/// read for one.
1228fn facts(
1229    repo: &Repo,
1230    cfg: &Config,
1231    number: i64,
1232    kind: ItemKind,
1233    named: bool,
1234    mention: bool,
1235) -> Result<Facts> {
1236    let viewer = repo.viewer_login()?.to_string();
1237    let below_floor = number < cfg.loop_cfg.min_number;
1238    let in_team = |team: &str| repo.team_has_member(team, &viewer);
1239    match kind {
1240        ItemKind::Pr => {
1241            let row = repo.sweep_pr(number)?;
1242            let record = repo.read_review_record(number);
1243            let posted = record.as_ref().is_some_and(|r| r.posted);
1244            let pending = review_pending(&row.review_requests, &viewer, &in_team);
1245            // Only an open request can have been answered, and only by a
1246            // review somebody saw, so the timeline is read for nothing else.
1247            let answered = pending
1248                && record.as_ref().filter(|r| r.posted).is_some_and(|r| {
1249                    repo.review_requests_made(number).is_some_and(|made| {
1250                        request_answered(&made, &viewer, &in_team, &r.reviewed_at)
1251                    })
1252                });
1253            // Read, not believed: the thread may have begun with a mention
1254            // long before the review on record.
1255            let mentioned = match record.as_ref().filter(|r| r.posted) {
1256                Some(r) if mention => {
1257                    let mut rows = repo.try_pr_review_comments(number)?;
1258                    rows.extend(repo.try_pr_reviews(number)?);
1259                    rows.extend(repo.try_issue_comments(number)?);
1260                    crate::comments::mentioned_in(&rows, &viewer, &r.reviewed_at, &in_team)
1261                }
1262                _ => false,
1263            };
1264            Ok(Facts {
1265                number,
1266                named,
1267                open: row.state.eq_ignore_ascii_case("OPEN"),
1268                author_is_viewer: row
1269                    .author
1270                    .as_ref()
1271                    .is_some_and(|a| crate::comments::same_login(&a.login, &viewer)),
1272                has_posted_record: posted,
1273                recheck_capped: posted
1274                    && record
1275                        .as_ref()
1276                        .is_some_and(|r| r.rechecks >= cfg.listen.max_rechecks),
1277                has_custody_state: repo.state_path(number).exists(),
1278                below_floor,
1279                linked_pr_with_record: None,
1280                review_pending: pending,
1281                request_answered: answered,
1282                mentioned,
1283                asks_open: false,
1284                answers_capped: false,
1285            })
1286        }
1287        ItemKind::Issue => {
1288            let issue = repo.read_issue(number)?;
1289            let linked = repo
1290                .try_open_pr_for_issue(number, cfg.base_branch())
1291                .ok()
1292                .flatten()
1293                .and_then(|pr| {
1294                    repo.read_review_record(pr.number)
1295                        .filter(|r| r.posted)
1296                        .map(|r| (pr.number, r))
1297                });
1298            // What the pull request's own row of the table would check.
1299            let (author_is_viewer, recheck_capped) = match &linked {
1300                Some((pr, record)) => {
1301                    let row = repo.sweep_pr(*pr)?;
1302                    (
1303                        row.author
1304                            .as_ref()
1305                            .is_some_and(|a| crate::comments::same_login(&a.login, &viewer)),
1306                        record.rechecks >= cfg.listen.max_rechecks,
1307                    )
1308                }
1309                None => (false, false),
1310            };
1311            let mentioned = match &linked {
1312                Some((_, record)) if mention => crate::comments::mentioned_in(
1313                    &repo.try_issue_comments(number)?,
1314                    &viewer,
1315                    &record.reviewed_at,
1316                    &in_team,
1317                ),
1318                _ => false,
1319            };
1320            let (asks_open, answers_capped) = match &linked {
1321                None if mention => {
1322                    let (open, automatic) = answer::outstanding(repo, number, &viewer, &in_team)?;
1323                    (open > 0, automatic >= cfg.listen.max_answers)
1324                }
1325                _ => (false, false),
1326            };
1327            Ok(Facts {
1328                number,
1329                named,
1330                open: issue.state.eq_ignore_ascii_case("OPEN"),
1331                author_is_viewer,
1332                has_posted_record: false,
1333                recheck_capped,
1334                has_custody_state: false,
1335                below_floor,
1336                linked_pr_with_record: linked.map(|(pr, _)| pr),
1337                review_pending: false,
1338                request_answered: false,
1339                mentioned,
1340                asks_open,
1341                answers_capped,
1342            })
1343        }
1344    }
1345}
1346
1347fn dispatch(
1348    agents: &[Agent],
1349    cfg: &Config,
1350    repo: &Repo,
1351    action: &Action,
1352    number: i64,
1353    p: &Pending,
1354) -> IssueRun {
1355    match action {
1356        Action::Recheck(pr) => review_only::recheck_pr(
1357            agents,
1358            cfg,
1359            repo,
1360            *pr,
1361            RecheckMode {
1362                dry_run: false,
1363                requested: p.triggers.contains(&Trigger::ReviewRequested),
1364                verdict_on_fallback: cfg.listen.verdicts,
1365            },
1366        ),
1367        Action::Review => review_only::review_pr(
1368            agents,
1369            cfg,
1370            repo,
1371            number,
1372            ReviewOpts {
1373                dry_run: false,
1374                verdict: cfg.listen.verdicts,
1375            },
1376        ),
1377        Action::Checkin { reply_only } => checkin::checkin_pr(
1378            agents,
1379            cfg,
1380            repo,
1381            number,
1382            &checkin::Mode {
1383                dry_run: false,
1384                reply_only: *reply_only,
1385                trust: cfg.loop_cfg.checkin_trust,
1386                again: false,
1387                resolve: cfg.loop_cfg.checkin_resolve,
1388                posts: checkin::posts(cfg),
1389            },
1390        ),
1391        Action::Answer { draft } => answer::answer_issue(
1392            agents,
1393            cfg,
1394            repo,
1395            number,
1396            &answer::Mode {
1397                dry_run: false,
1398                draft: *draft,
1399                again: false,
1400                auto: true,
1401                posts: checkin::posts(cfg),
1402            },
1403        ),
1404        other => {
1405            let mut run = IssueRun::new(number, format!("#{number}"));
1406            run.status = Status::Error;
1407            run.notes
1408                .push(format!("nothing to run: {}", other.phrase(number)));
1409            run
1410        }
1411    }
1412}
1413
1414/// One event, as the log says it.
1415fn describe(repo: &Repo, event: &Event) -> String {
1416    let n = event.number;
1417    match event.trigger {
1418        Trigger::Pushed => {
1419            let (from, to) = event.head.clone().unwrap_or_default();
1420            let short = |s: &str| s.chars().take(7).collect::<String>();
1421            let commits = repo
1422                .commits_between(&from, &to)
1423                .map(|c| format!(", {c} new commit{} since the review", plural(c as usize)))
1424                .unwrap_or_default();
1425            format!(
1426                "#{n}: head moved {} to {}{commits}",
1427                short(&from),
1428                short(&to)
1429            )
1430        }
1431        Trigger::Replied => match &event.who {
1432            Some(who) => format!("#{n}: {who} commented"),
1433            None => format!("#{n}: somebody commented"),
1434        },
1435        Trigger::Mentioned => match &event.who {
1436            Some(who) => format!("#{n}: mentioned by {who}"),
1437            None => format!("#{n}: mentioned"),
1438        },
1439        // The sweep read the request off the pull request. A thread says only
1440        // that it began as one, which a later update on it says too.
1441        Trigger::ReviewRequested if event.from_thread => {
1442            format!("#{n}: news on a thread that asked for your review")
1443        }
1444        Trigger::ReviewRequested => format!("#{n}: your review was requested"),
1445        Trigger::Assigned => format!("#{n}: assigned to you"),
1446        Trigger::Closed => format!("#{n}: closed, no longer watched"),
1447    }
1448}
1449
1450fn plural(n: usize) -> &'static str {
1451    if n == 1 {
1452        ""
1453    } else {
1454        "s"
1455    }
1456}
1457
1458// ---------------------------------------------------------------------------
1459// The loop
1460// ---------------------------------------------------------------------------
1461
1462enum Ending {
1463    Interrupted,
1464    Once { failed: bool },
1465    Done,
1466}
1467
1468/// Listen until stopped, or once.
1469pub fn run(agents: &[Agent], cfg: &Config, repo: &Repo, opts: Opts) -> Result<i32> {
1470    proc::stop_on_signal()?;
1471    if !cfg!(unix) {
1472        logdim!(
1473            "Ctrl-C ends spar at once on this platform; an agent call in flight is not stopped"
1474        );
1475    }
1476    let now = clock::now_secs();
1477    let lock = Lock::acquire(repo, now)?;
1478    let viewer = repo.viewer_login()?.to_string();
1479    let name = repo
1480        .repo_name()
1481        .unwrap_or_else(|| repo.root().display().to_string());
1482
1483    // Named items, sorted into kinds. An unknown number is a hard error, as
1484    // it is everywhere else: naming something that does not exist is a typo.
1485    let named: BTreeSet<i64> = opts.items.iter().copied().collect();
1486    let mut named_prs = Vec::new();
1487    let mut named_issues = Vec::new();
1488    for number in &named {
1489        match repo.item_kind(*number)? {
1490            ItemKind::Pr => named_prs.push(*number),
1491            ItemKind::Issue => named_issues.push(*number),
1492        }
1493    }
1494    let repo_mode = named.is_empty();
1495    // What this run may act on, in numbers mode: what was named, and the pull
1496    // requests of named issues as the sweep finds them. Anything else in
1497    // listen.json, pending work included, is left for a run over the
1498    // repository.
1499    let mut watching: BTreeSet<i64> = named.clone();
1500    // A dry run changes nothing, listen.json included, so what it previewed is
1501    // still there for the real run after it.
1502    let save = |state: &ListenState| {
1503        if !opts.dry_run {
1504            write_state(repo, state);
1505        }
1506    };
1507
1508    let mut state = read_state(repo);
1509    if state.watermark.is_empty() || opts.since.is_some() {
1510        let back = opts.since.map(|d| d.as_secs()).unwrap_or(0);
1511        state.watermark = clock::rfc3339(now.saturating_sub(back));
1512        state.last_modified = None;
1513        log!("watching from {}", state.watermark);
1514    }
1515    for number in &named_prs {
1516        state.watched(*number, ItemKind::Pr);
1517    }
1518    for number in &named_issues {
1519        state.watched(*number, ItemKind::Issue);
1520    }
1521    log!(
1522        "listening to {name}{}: polling every {}s, sweeping {} every {}s. Ctrl-C to stop.",
1523        if repo_mode {
1524            String::new()
1525        } else {
1526            format!(
1527                " for {}",
1528                named
1529                    .iter()
1530                    .map(|n| format!("#{n}"))
1531                    .collect::<Vec<_>>()
1532                    .join(", ")
1533            )
1534        },
1535        cfg.listen.interval.max(FLOOR_SECS),
1536        if repo_mode {
1537            "open pull requests"
1538        } else {
1539            "the named items"
1540        },
1541        cfg.listen.sweep
1542    );
1543
1544    let started = now;
1545    let mut next_poll = now;
1546    let mut next_sweep = now;
1547    let mut backoff = Backoff::default();
1548    let mut events_seen = 0usize;
1549    let mut dispatches = 0usize;
1550    let mut tick_failed = false;
1551    let mut last_prune = now;
1552
1553    let ending = loop {
1554        if proc::interrupted() {
1555            break Ending::Interrupted;
1556        }
1557        let now = clock::now_secs();
1558        let mut events: Vec<Event> = Vec::new();
1559
1560        if now >= next_poll {
1561            match poll(repo, &cfg.listen, &mut state, now) {
1562                Polled::Nothing(interval) => {
1563                    backoff.recovered();
1564                    next_poll = now + interval;
1565                }
1566                Polled::Threads(threads, interval) => {
1567                    backoff.recovered();
1568                    events.extend(events_from_threads(&threads, &mut state));
1569                    next_poll = now + interval;
1570                }
1571                Polled::RateLimited(wait) => {
1572                    logwarn!(
1573                        "rate limited by GitHub, waiting {} until it resets",
1574                        spend::clock(Duration::from_secs(wait))
1575                    );
1576                    next_poll = now + wait;
1577                    tick_failed = true;
1578                }
1579                Polled::Failed(why) => {
1580                    next_poll = now + backoff.failed("read notifications", &why);
1581                    tick_failed = true;
1582                }
1583            }
1584        }
1585
1586        if now >= next_sweep {
1587            match sweep_rows(repo, cfg, repo_mode, &named_prs, &named_issues, &state) {
1588                Ok(Swept {
1589                    rows,
1590                    closed_issues,
1591                }) => {
1592                    backoff.recovered();
1593                    if !repo_mode {
1594                        watching.extend(rows.iter().map(|r| r.number));
1595                    }
1596                    events.extend(closed_issues.into_iter().map(|number| Event {
1597                        number,
1598                        kind: ItemKind::Issue,
1599                        trigger: Trigger::Closed,
1600                        who: None,
1601                        head: None,
1602                        commits: None,
1603                        from_thread: false,
1604                    }));
1605                    let record_head = |n: i64| {
1606                        repo.read_review_record(n)
1607                            .filter(|r| r.posted)
1608                            .map(|r| r.head)
1609                    };
1610                    let swept = events_from_sweep(
1611                        &rows,
1612                        &viewer,
1613                        &mut state,
1614                        &named,
1615                        cfg.loop_cfg.min_number,
1616                        &record_head,
1617                        repo_mode,
1618                    );
1619                    events.extend(swept);
1620                    next_sweep = now + cfg.listen.sweep;
1621                }
1622                Err(e) => {
1623                    next_sweep = now + backoff.failed("sweep the pull requests", e.last_line());
1624                    tick_failed = true;
1625                }
1626            }
1627        }
1628
1629        // In numbers mode an event about anything not watched is dropped here,
1630        // after the watermark has moved past it, so a thread about something
1631        // else is read once and never again.
1632        for event in &events {
1633            if !repo_mode && !watching.contains(&event.number) {
1634                continue;
1635            }
1636            if event.trigger != Trigger::Closed && event.who.is_none() {
1637                // Nothing to look up cheaply here; the actor comes with the
1638                // thread only through its latest comment, which the sweep
1639                // does not carry.
1640            }
1641            events_seen += 1;
1642            log!("{}", describe(repo, event));
1643            if event.trigger == Trigger::Closed {
1644                state.items.remove(&event.number);
1645                state.pending.remove(&event.number);
1646                continue;
1647            }
1648            state.note(event, now);
1649        }
1650        save(&state);
1651
1652        let due_now: Vec<i64> = state
1653            .pending
1654            .iter()
1655            .filter(|(n, p)| {
1656                (repo_mode || watching.contains(n))
1657                    && due(p, now, &cfg.listen, opts.once)
1658                    && state.items.get(n).is_none_or(|w| w.retry_after <= now)
1659            })
1660            .map(|(n, _)| *n)
1661            .collect();
1662        for number in due_now {
1663            let Some(mut p) = state.pending.get(&number).cloned() else {
1664                continue;
1665            };
1666            let named_item = named.contains(&number);
1667            let mention = p.triggers.contains(&Trigger::Mentioned);
1668            let f = match facts(repo, cfg, number, p.kind, named_item, mention) {
1669                Ok(f) => f,
1670                Err(e) => {
1671                    let w = state.watched(number, p.kind);
1672                    w.failures += 1;
1673                    w.retry_after = now + dispatch_backoff_secs(w.failures);
1674                    logwarn!("#{number}: could not read it: {}", e.last_line());
1675                    tick_failed = true;
1676                    save(&state);
1677                    continue;
1678                }
1679            };
1680            if settle(&mut p, &f) {
1681                if f.request_answered {
1682                    log!(
1683                        "#{number}: the review asked of you is the one on record, so that was \
1684                         only an update"
1685                    );
1686                } else {
1687                    log!("#{number}: no review is asked of you now, so that was only an update");
1688                }
1689            }
1690            if settle_mention(&mut p, &f) {
1691                if p.kind == ItemKind::Issue && f.linked_pr_with_record.is_none() {
1692                    log!(
1693                        "#{number}: nothing on it asks you anything spar has not judged, so that \
1694                         was only an update"
1695                    );
1696                } else {
1697                    log!(
1698                        "#{number}: nobody mentioned you since the review on record, so that was \
1699                         only an update"
1700                    );
1701                }
1702            }
1703            if p.triggers.contains(&Trigger::ReviewRequested) || f.request_answered {
1704                // Answered now, or by the review on record, whichever way it
1705                // arrived. A standing note or a review posted as a comment
1706                // leaves the request open, so the next sweep would otherwise
1707                // take it for the author asking again.
1708                state.watched(number, p.kind).requested_noticed = true;
1709            }
1710            let action = plan(&p, &f, &cfg.listen);
1711            match &action {
1712                Action::Ignore(why) => {
1713                    logdim!("#{number}: ignoring {why}");
1714                    state.pending.remove(&number);
1715                }
1716                Action::Drop => {
1717                    log!("#{number}: closed, no longer watched");
1718                    state.items.remove(&number);
1719                    state.pending.remove(&number);
1720                }
1721                Action::Capped(pr) => {
1722                    let w = state.watched(number, p.kind);
1723                    if !w.said_capped {
1724                        let what = if *pr == number {
1725                            "rechecked".to_string()
1726                        } else {
1727                            format!("its pull request #{pr} was rechecked")
1728                        };
1729                        log!(
1730                            "#{number}: {what} {} time{} already; run `spar review {pr}` to look \
1731                             again",
1732                            cfg.listen.max_rechecks,
1733                            plural(cfg.listen.max_rechecks as usize)
1734                        );
1735                        w.said_capped = true;
1736                    }
1737                    state.pending.remove(&number);
1738                }
1739                Action::AnswersCapped => {
1740                    let w = state.watched(number, p.kind);
1741                    if !w.said_capped {
1742                        if cfg.listen.max_answers == 0 {
1743                            log!(
1744                                "#{number}: max_answers is 0, so spar listen answers nothing; run \
1745                                 `spar answer {number}` to answer it"
1746                            );
1747                        } else {
1748                            log!(
1749                                "#{number}: answered {} time{} already; run `spar answer {number}` \
1750                                 to answer again",
1751                                cfg.listen.max_answers,
1752                                plural(cfg.listen.max_answers as usize)
1753                            );
1754                        }
1755                        w.said_capped = true;
1756                    }
1757                    state.pending.remove(&number);
1758                }
1759                Action::BelongsToResume => {
1760                    let w = state.watched(number, p.kind);
1761                    if !w.said_resume {
1762                        log!("#{number}: in custody, `spar resume {number}` continues it");
1763                        w.said_resume = true;
1764                    }
1765                    state.pending.remove(&number);
1766                }
1767                action if action.runs() => {
1768                    if opts.dry_run {
1769                        log!(
1770                            "#{number}: would {} after {}, dry run so nothing runs",
1771                            action.phrase(number),
1772                            p.because()
1773                        );
1774                        state.pending.remove(&number);
1775                    } else {
1776                        log!(
1777                            "#{number}: about to {} after {}",
1778                            action.phrase(number),
1779                            p.because()
1780                        );
1781                        let acted_on = match action {
1782                            Action::Recheck(pr) => *pr,
1783                            _ => number,
1784                        };
1785                        let record_before = repo.read_review_record(acted_on);
1786                        let failed_before = repo.write_summary().failed;
1787                        let run = dispatch(agents, cfg, repo, action, number, &p);
1788                        dispatches += 1;
1789                        let spent = spend::drain();
1790                        if let Err(e) = repo.record_spend(&spent) {
1791                            logdim!("could not record what this run spent: {e}");
1792                        }
1793                        crate::cli::print_outcome(&run, &spent);
1794                        if proc::interrupted() {
1795                            log!("interrupted while working on #{number}; its worktree was kept");
1796                            save(&state);
1797                            break;
1798                        }
1799                        // The commit spar just read is not news to the next
1800                        // sweep. Only a record this dispatch wrote says which
1801                        // that was: one it left alone holds an older commit,
1802                        // and taking that for the head would raise the same
1803                        // push again on every sweep.
1804                        let record_after = repo.read_review_record(acted_on);
1805                        if record_after != record_before {
1806                            if let Some(record) = record_after {
1807                                state.watched(acted_on, ItemKind::Pr).head = Some(record.head);
1808                            }
1809                        }
1810                        // A review GitHub never received has not happened,
1811                        // whatever its findings say, and clearing its trigger
1812                        // would lose it for good.
1813                        let unposted = repo.write_summary().failed > failed_before;
1814                        let w = state.watched(number, p.kind);
1815                        // Nothing posted because the head moved under it: the
1816                        // pending item stays, and the next look reads the new
1817                        // head.
1818                        let moved = run.status == Status::Moved && !unposted;
1819                        if run.status == Status::Error || moved || unposted {
1820                            w.failures += 1;
1821                            let wait = retry_wait(moved, w.failures, &cfg.listen);
1822                            // From now rather than the start of the tick, which
1823                            // a long review leaves far behind.
1824                            w.retry_after = clock::now_secs() + wait;
1825                            let gave_up = w.failures >= MAX_FAILURES;
1826                            if gave_up {
1827                                logwarn!("#{number}: giving up on it until it changes again");
1828                                state.give_up(number, p.kind);
1829                            } else {
1830                                if moved {
1831                                    log!(
1832                                        "#{number}: its head moved while spar read it, so it runs \
1833                                         again on the new head in {}",
1834                                        spend::clock(Duration::from_secs(wait))
1835                                    );
1836                                } else if unposted {
1837                                    logwarn!(
1838                                        "#{number}: not everything it wrote reached GitHub, so it \
1839                                         runs again in {}",
1840                                        spend::clock(Duration::from_secs(wait))
1841                                    );
1842                                }
1843                                if let Some(p) = state.pending.get_mut(&number) {
1844                                    p.first_seen = now;
1845                                }
1846                            }
1847                            // A moved head failed nothing: it is still pending,
1848                            // and the next `--once` reads the new head.
1849                            if !moved || gave_up {
1850                                tick_failed = true;
1851                            }
1852                        } else {
1853                            w.failures = 0;
1854                            w.retry_after = 0;
1855                            w.last_dispatched = Some(now);
1856                            state.pending.remove(&number);
1857                        }
1858                    }
1859                }
1860                _ => {}
1861            }
1862            save(&state);
1863        }
1864        if proc::interrupted() {
1865            break Ending::Interrupted;
1866        }
1867
1868        if opts.once {
1869            break Ending::Once {
1870                failed: tick_failed,
1871            };
1872        }
1873        if !repo_mode && named.iter().all(|n| !state.items.contains_key(n)) {
1874            log!("every named item is closed, nothing left to watch");
1875            break Ending::Done;
1876        }
1877        if now.saturating_sub(last_prune) >= 3600 {
1878            prune_seen(&mut state, now);
1879            last_prune = now;
1880            save(&state);
1881        }
1882
1883        let acts_on = |n: i64| repo_mode || watching.contains(&n);
1884        let next = [next_poll, next_sweep]
1885            .into_iter()
1886            .chain(next_due(&state, &cfg.listen, &acts_on))
1887            .min()
1888            .unwrap_or(now + FLOOR_SECS);
1889        let wait = next.saturating_sub(clock::now_secs()).clamp(1, 3600);
1890        if proc::sleep_unless_interrupted(Duration::from_secs(wait)) {
1891            break Ending::Interrupted;
1892        }
1893    };
1894
1895    save(&state);
1896    drop(lock);
1897    log!(
1898        "listened for {}: {events_seen} event{}, {dispatches} dispatch{}",
1899        spend::clock(Duration::from_secs(
1900            clock::now_secs().saturating_sub(started)
1901        )),
1902        plural(events_seen),
1903        if dispatches == 1 { "" } else { "es" }
1904    );
1905    let writes = crate::cli::report_writes(repo);
1906    Ok(match ending {
1907        Ending::Interrupted => 130.max(writes),
1908        Ending::Once { failed: true } => 1.max(writes),
1909        Ending::Once { failed: false } | Ending::Done => writes,
1910    })
1911}
1912
1913fn read_state(repo: &Repo) -> ListenState {
1914    let path = repo.listen_state_path();
1915    match std::fs::read_to_string(&path) {
1916        Ok(text) => match serde_json::from_str::<ListenState>(&text) {
1917            Ok(state) => state,
1918            Err(e) => {
1919                logdim!("could not read {}: {e}. Starting fresh.", path.display());
1920                ListenState::default()
1921            }
1922        },
1923        Err(_) => ListenState::default(),
1924    }
1925}
1926
1927fn write_state(repo: &Repo, state: &ListenState) {
1928    let mut state = state.clone();
1929    state.version = STATE_VERSION;
1930    if let Err(e) = crate::repo::write_json_atomic(&repo.listen_state_path(), &state) {
1931        logdim!("could not write the listen state: {e}");
1932    }
1933}
1934
1935#[cfg(test)]
1936mod tests {
1937    use super::*;
1938    use crate::model::Author;
1939
1940    fn cfg() -> ListenCfg {
1941        ListenCfg::default()
1942    }
1943
1944    fn user(login: &str) -> Login {
1945        Login {
1946            login: login.into(),
1947            ..Login::default()
1948        }
1949    }
1950
1951    fn team(slug: &str) -> Login {
1952        Login {
1953            slug: slug.into(),
1954            ..Login::default()
1955        }
1956    }
1957
1958    fn thread(id: &str, reason: &str, kind: &str, number: i64, at: &str) -> Thread {
1959        Thread {
1960            id: id.into(),
1961            reason: reason.into(),
1962            updated_at: at.into(),
1963            subject: Subject {
1964                title: format!("thing {number}"),
1965                url: Some(format!(
1966                    "https://api.github.com/repos/o/r/{}/{number}",
1967                    if kind == "PullRequest" {
1968                        "pulls"
1969                    } else {
1970                        "issues"
1971                    }
1972                )),
1973                latest_comment_url: None,
1974                kind: kind.into(),
1975            },
1976        }
1977    }
1978
1979    fn row(number: i64, head: &str, author: &str, state: &str) -> PrSweepRow {
1980        PrSweepRow {
1981            number,
1982            title: format!("PR {number}"),
1983            head_ref_oid: head.into(),
1984            author: Some(Author {
1985                login: author.into(),
1986            }),
1987            state: state.into(),
1988            ..Default::default()
1989        }
1990    }
1991
1992    fn pending(kind: ItemKind, triggers: &[Trigger], now: u64) -> Pending {
1993        Pending {
1994            kind,
1995            first_seen: now,
1996            last_seen: now,
1997            triggers: triggers.iter().copied().collect(),
1998            who: Vec::new(),
1999            head_from: None,
2000            head_to: None,
2001        }
2002    }
2003
2004    fn open_pr() -> Facts {
2005        Facts {
2006            number: 7,
2007            open: true,
2008            ..Default::default()
2009        }
2010    }
2011
2012    fn response(status: u16, headers: &[(&str, &str)]) -> ApiResponse {
2013        ApiResponse {
2014            status,
2015            headers: headers
2016                .iter()
2017                .map(|(k, v)| (k.to_string(), v.to_string()))
2018                .collect(),
2019            body: String::new(),
2020        }
2021    }
2022
2023    // -- timing ----------------------------------------------------------------
2024
2025    #[test]
2026    fn the_poll_interval_is_the_largest_of_the_config_the_header_and_the_floor() {
2027        let mut c = cfg();
2028        assert_eq!(60, poll_interval(&c, None));
2029        assert_eq!(90, poll_interval(&c, Some("90")));
2030        assert_eq!(60, poll_interval(&c, Some("30")));
2031        assert_eq!(60, poll_interval(&c, Some("soon")));
2032        c.interval = 5;
2033        assert_eq!(15, poll_interval(&c, None));
2034    }
2035
2036    #[test]
2037    fn a_rate_limit_answer_says_how_long_to_wait_and_caps_it_at_an_hour() {
2038        let now = 1_000_000;
2039        assert_eq!(
2040            Some(120),
2041            rate_limit_wait(&response(429, &[("retry-after", "120")]), now)
2042        );
2043        assert_eq!(
2044            Some(600),
2045            rate_limit_wait(
2046                &response(
2047                    403,
2048                    &[
2049                        ("x-ratelimit-remaining", "0"),
2050                        ("x-ratelimit-reset", "1000600")
2051                    ]
2052                ),
2053                now
2054            )
2055        );
2056        assert_eq!(
2057            Some(3600),
2058            rate_limit_wait(&response(403, &[("retry-after", "99999")]), now)
2059        );
2060        assert_eq!(
2061            Some(60),
2062            rate_limit_wait(&response(429, &[("retry-after", "1")]), now)
2063        );
2064    }
2065
2066    #[test]
2067    fn a_forbidden_answer_without_a_rate_limit_header_is_a_failure_not_a_wait() {
2068        assert_eq!(None, rate_limit_wait(&response(403, &[]), 0));
2069        assert_eq!(
2070            None,
2071            rate_limit_wait(&response(403, &[("x-ratelimit-remaining", "40")]), 0)
2072        );
2073        assert_eq!(
2074            None,
2075            rate_limit_wait(&response(200, &[("retry-after", "5")]), 0)
2076        );
2077    }
2078
2079    #[test]
2080    fn next_link_reads_the_next_page_and_nothing_else() {
2081        let mut headers = BTreeMap::new();
2082        headers.insert(
2083            "link".to_string(),
2084            "<https://api.github.com/x?page=2>; rel=\"next\", <https://api.github.com/x?page=9>; rel=\"last\"".to_string(),
2085        );
2086        assert_eq!(
2087            Some("https://api.github.com/x?page=2".to_string()),
2088            next_link(&headers)
2089        );
2090        headers.insert(
2091            "link".to_string(),
2092            "<https://api.github.com/x?page=1>; rel=\"prev\"".to_string(),
2093        );
2094        assert_eq!(None, next_link(&headers));
2095        assert_eq!(None, next_link(&BTreeMap::new()));
2096    }
2097
2098    // -- notifications ------------------------------------------------------------
2099
2100    #[test]
2101    fn a_thread_becomes_an_event_keyed_by_its_number_and_reason() {
2102        let mut state = ListenState::default();
2103        let threads = vec![
2104            thread(
2105                "1",
2106                "review_requested",
2107                "PullRequest",
2108                7,
2109                "2026-10-20T10:00:00Z",
2110            ),
2111            thread("2", "mention", "Issue", 42, "2026-10-20T10:01:00Z"),
2112            thread("3", "comment", "PullRequest", 8, "2026-10-20T10:02:00Z"),
2113            thread("4", "assign", "PullRequest", 9, "2026-10-20T10:03:00Z"),
2114            thread("5", "team_mention", "Issue", 43, "2026-10-20T10:04:00Z"),
2115            thread(
2116                "6",
2117                "state_change",
2118                "PullRequest",
2119                10,
2120                "2026-10-20T10:05:00Z",
2121            ),
2122        ];
2123        let events = events_from_threads(&threads, &mut state);
2124        let got: Vec<(i64, ItemKind, Trigger)> = events
2125            .iter()
2126            .map(|e| (e.number, e.kind, e.trigger))
2127            .collect();
2128        assert_eq!(
2129            vec![
2130                (7, ItemKind::Pr, Trigger::ReviewRequested),
2131                (42, ItemKind::Issue, Trigger::Mentioned),
2132                (8, ItemKind::Pr, Trigger::Replied),
2133                (9, ItemKind::Pr, Trigger::Assigned),
2134                (43, ItemKind::Issue, Trigger::Mentioned),
2135                (10, ItemKind::Pr, Trigger::Replied),
2136            ],
2137            got
2138        );
2139    }
2140
2141    #[test]
2142    fn a_thread_seen_at_the_same_update_time_is_not_an_event_twice() {
2143        let mut state = ListenState::default();
2144        let same = vec![thread(
2145            "1",
2146            "comment",
2147            "PullRequest",
2148            7,
2149            "2026-10-20T10:00:00Z",
2150        )];
2151        assert_eq!(1, events_from_threads(&same, &mut state).len());
2152        assert_eq!(0, events_from_threads(&same, &mut state).len());
2153        let moved = vec![thread(
2154            "1",
2155            "comment",
2156            "PullRequest",
2157            7,
2158            "2026-10-20T10:05:00Z",
2159        )];
2160        assert_eq!(1, events_from_threads(&moved, &mut state).len());
2161    }
2162
2163    #[test]
2164    fn a_thread_about_a_release_or_discussion_is_ignored() {
2165        let mut state = ListenState::default();
2166        let threads = vec![
2167            thread("1", "subscribed", "Release", 1, "2026-10-20T10:00:00Z"),
2168            thread("2", "mention", "Discussion", 2, "2026-10-20T10:01:00Z"),
2169            Thread {
2170                subject: Subject {
2171                    url: None,
2172                    kind: "PullRequest".into(),
2173                    ..Default::default()
2174                },
2175                ..thread("3", "comment", "PullRequest", 3, "2026-10-20T10:02:00Z")
2176            },
2177        ];
2178        assert!(events_from_threads(&threads, &mut state).is_empty());
2179        // Seen and past the watermark all the same, so they are never read
2180        // again.
2181        assert_eq!(3, state.seen.len());
2182    }
2183
2184    #[test]
2185    fn the_watermark_advances_to_the_newest_update_seen() {
2186        let mut state = ListenState {
2187            watermark: "2026-10-20T09:00:00Z".into(),
2188            ..Default::default()
2189        };
2190        let threads = vec![
2191            thread("1", "comment", "PullRequest", 7, "2026-10-20T10:02:00Z"),
2192            thread("2", "comment", "PullRequest", 8, "2026-10-20T10:01:00Z"),
2193        ];
2194        events_from_threads(&threads, &mut state);
2195        assert_eq!("2026-10-20T10:02:00Z", state.watermark);
2196    }
2197
2198    // -- the sweep ----------------------------------------------------------------
2199
2200    fn no_record(_: i64) -> Option<String> {
2201        None
2202    }
2203
2204    #[test]
2205    fn a_sweep_notices_a_head_that_moved_and_says_it_once() {
2206        let mut state = ListenState::default();
2207        let named = BTreeSet::new();
2208        let first = events_from_sweep(
2209            &[row(7, "aaa", "alice", "OPEN")],
2210            "me",
2211            &mut state,
2212            &named,
2213            0,
2214            &no_record,
2215            true,
2216        );
2217        assert!(first.is_empty(), "{first:?}");
2218        let moved = events_from_sweep(
2219            &[row(7, "bbb", "alice", "OPEN")],
2220            "me",
2221            &mut state,
2222            &named,
2223            0,
2224            &no_record,
2225            true,
2226        );
2227        assert_eq!(1, moved.len());
2228        assert_eq!(Trigger::Pushed, moved[0].trigger);
2229        assert_eq!(Some(("aaa".to_string(), "bbb".to_string())), moved[0].head);
2230        let again = events_from_sweep(
2231            &[row(7, "bbb", "alice", "OPEN")],
2232            "me",
2233            &mut state,
2234            &named,
2235            0,
2236            &no_record,
2237            true,
2238        );
2239        assert!(again.is_empty(), "{again:?}");
2240        assert_eq!("alice", state.items[&7].author);
2241    }
2242
2243    #[test]
2244    fn the_first_sweep_treats_a_reviewed_pull_request_that_already_moved_as_due() {
2245        let mut state = ListenState::default();
2246        let record = |n: i64| (n == 7).then(|| "aaa".to_string());
2247        let events = events_from_sweep(
2248            &[row(7, "bbb", "alice", "OPEN"), row(8, "ccc", "bob", "OPEN")],
2249            "me",
2250            &mut state,
2251            &BTreeSet::new(),
2252            0,
2253            &record,
2254            true,
2255        );
2256        assert_eq!(1, events.len());
2257        assert_eq!(7, events[0].number);
2258        assert_eq!(Trigger::Pushed, events[0].trigger);
2259    }
2260
2261    #[test]
2262    fn a_sweep_notices_a_review_request_once_not_every_five_minutes() {
2263        let mut state = ListenState::default();
2264        let mut asked = row(7, "aaa", "alice", "OPEN");
2265        asked.review_requests = vec![user("ME")];
2266        asked.assignees = vec![user("me")];
2267        let rows = [asked];
2268        let first = events_from_sweep(
2269            &rows,
2270            "me",
2271            &mut state,
2272            &BTreeSet::new(),
2273            0,
2274            &no_record,
2275            true,
2276        );
2277        let triggers: BTreeSet<Trigger> = first.iter().map(|e| e.trigger).collect();
2278        assert_eq!(
2279            [Trigger::ReviewRequested, Trigger::Assigned]
2280                .into_iter()
2281                .collect::<BTreeSet<_>>(),
2282            triggers
2283        );
2284        let second = events_from_sweep(
2285            &rows,
2286            "me",
2287            &mut state,
2288            &BTreeSet::new(),
2289            0,
2290            &no_record,
2291            true,
2292        );
2293        assert!(second.is_empty(), "{second:?}");
2294    }
2295
2296    #[test]
2297    fn a_sweep_notices_a_review_requested_again_after_a_review() {
2298        let mut state = ListenState::default();
2299        let mut asked = row(7, "aaa", "alice", "OPEN");
2300        asked.review_requests = vec![user("me")];
2301        let asked = [asked];
2302        let reviewed = [row(7, "aaa", "alice", "OPEN")];
2303        let mut sweep = |rows: &[PrSweepRow]| -> Vec<Trigger> {
2304            events_from_sweep(
2305                rows,
2306                "me",
2307                &mut state,
2308                &BTreeSet::new(),
2309                0,
2310                &no_record,
2311                true,
2312            )
2313            .iter()
2314            .map(|e| e.trigger)
2315            .collect()
2316        };
2317        assert_eq!(vec![Trigger::ReviewRequested], sweep(&asked));
2318        // The review cleared the request.
2319        assert!(sweep(&reviewed).is_empty());
2320        assert!(sweep(&reviewed).is_empty());
2321        // The author asked again.
2322        assert_eq!(vec![Trigger::ReviewRequested], sweep(&asked));
2323        assert!(sweep(&asked).is_empty());
2324    }
2325
2326    #[test]
2327    fn a_sweep_drops_a_pull_request_that_closed_or_vanished() {
2328        let mut state = ListenState::default();
2329        let named = BTreeSet::new();
2330        events_from_sweep(
2331            &[row(7, "aaa", "alice", "OPEN"), row(8, "bbb", "bob", "OPEN")],
2332            "me",
2333            &mut state,
2334            &named,
2335            0,
2336            &no_record,
2337            true,
2338        );
2339        let events = events_from_sweep(
2340            &[row(7, "aaa", "alice", "MERGED")],
2341            "me",
2342            &mut state,
2343            &named,
2344            0,
2345            &no_record,
2346            true,
2347        );
2348        let closed: Vec<i64> = events
2349            .iter()
2350            .filter(|e| e.trigger == Trigger::Closed)
2351            .map(|e| e.number)
2352            .collect();
2353        assert_eq!(vec![7, 8], closed);
2354
2355        // In numbers mode a pull request missing from the rows means it was
2356        // not asked about, not that it closed.
2357        let mut state = ListenState::default();
2358        let named: BTreeSet<i64> = [7, 8].into_iter().collect();
2359        events_from_sweep(
2360            &[row(7, "aaa", "alice", "OPEN"), row(8, "bbb", "bob", "OPEN")],
2361            "me",
2362            &mut state,
2363            &named,
2364            0,
2365            &no_record,
2366            false,
2367        );
2368        let events = events_from_sweep(
2369            &[row(7, "aaa", "alice", "OPEN")],
2370            "me",
2371            &mut state,
2372            &named,
2373            0,
2374            &no_record,
2375            false,
2376        );
2377        assert!(events.is_empty(), "{events:?}");
2378    }
2379
2380    #[test]
2381    fn a_sweep_ignores_a_number_below_the_floor_unless_it_was_named() {
2382        let mut state = ListenState::default();
2383        let named: BTreeSet<i64> = [3].into_iter().collect();
2384        let record = |_: i64| Some("old".to_string());
2385        let events = events_from_sweep(
2386            &[
2387                row(3, "new", "alice", "OPEN"),
2388                row(4, "new", "alice", "OPEN"),
2389                row(500, "new", "alice", "OPEN"),
2390            ],
2391            "me",
2392            &mut state,
2393            &named,
2394            480,
2395            &record,
2396            true,
2397        );
2398        let numbers: Vec<i64> = events.iter().map(|e| e.number).collect();
2399        assert_eq!(vec![3, 500], numbers);
2400        assert!(!state.items.contains_key(&4));
2401    }
2402
2403    /// A notification puts a pull request below the floor on the watch, and
2404    /// the sweep, which skips it, used to call it closed.
2405    #[test]
2406    fn a_pull_request_below_the_floor_that_is_still_open_is_not_called_closed() {
2407        let mut state = ListenState::default();
2408        state.watched(4, ItemKind::Pr);
2409        let events = events_from_sweep(
2410            &[
2411                row(4, "new", "alice", "OPEN"),
2412                row(500, "new", "alice", "OPEN"),
2413            ],
2414            "me",
2415            &mut state,
2416            &BTreeSet::new(),
2417            480,
2418            &no_record,
2419            true,
2420        );
2421        assert!(
2422            events.iter().all(|e| e.trigger != Trigger::Closed),
2423            "{events:?}"
2424        );
2425        let gone = events_from_sweep(
2426            &[row(500, "new", "alice", "OPEN")],
2427            "me",
2428            &mut state,
2429            &BTreeSet::new(),
2430            480,
2431            &no_record,
2432            true,
2433        );
2434        assert_eq!(
2435            vec![(4, Trigger::Closed)],
2436            gone.iter()
2437                .map(|e| (e.number, e.trigger))
2438                .collect::<Vec<_>>()
2439        );
2440    }
2441
2442    // -- the table -------------------------------------------------------------------
2443
2444    #[test]
2445    fn a_review_request_nobody_is_still_making_is_only_an_update() {
2446        // react-native-pubky#45: the thread began as a review request, so
2447        // GitHub kept calling it one when spar's own approval updated it.
2448        let threads = [thread(
2449            "1",
2450            "review_requested",
2451            "PullRequest",
2452            7,
2453            "2026-10-21T14:51:16Z",
2454        )];
2455        let events = events_from_threads(&threads, &mut ListenState::default());
2456        let mut waiting = BTreeMap::new();
2457        for e in &events {
2458            merge(&mut waiting, e, 0);
2459        }
2460        let reviewed = Facts {
2461            has_posted_record: true,
2462            ..open_pr()
2463        };
2464        let mut p = waiting.remove(&7).expect("pending");
2465        assert!(settle(&mut p, &reviewed));
2466        assert_eq!(
2467            [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2468            p.triggers
2469        );
2470        assert!(matches!(plan(&p, &reviewed, &cfg()), Action::Ignore(_)));
2471
2472        // On a pull request spar never reviewed it is not a reason to start.
2473        let mut p = pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0);
2474        assert!(settle(&mut p, &open_pr()));
2475        assert!(matches!(plan(&p, &open_pr(), &cfg()), Action::Ignore(_)));
2476
2477        // A request still standing is left as it is.
2478        let asked = Facts {
2479            review_pending: true,
2480            ..open_pr()
2481        };
2482        let mut p = pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0);
2483        assert!(!settle(&mut p, &asked));
2484        assert!(p.triggers.contains(&Trigger::ReviewRequested));
2485        assert_eq!(Action::Review, plan(&p, &asked, &cfg()));
2486
2487        // A request still standing that the review on record answered is
2488        // only an update too, and an update asks for no second look.
2489        let answered = Facts {
2490            has_posted_record: true,
2491            review_pending: true,
2492            request_answered: true,
2493            ..open_pr()
2494        };
2495        let mut p = pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0);
2496        assert!(settle(&mut p, &answered));
2497        assert_eq!(
2498            [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2499            p.triggers
2500        );
2501        assert!(matches!(plan(&p, &answered, &cfg()), Action::Ignore(_)));
2502
2503        // Nothing to settle without a request.
2504        let mut p = pending(ItemKind::Pr, &[Trigger::Pushed], 0);
2505        assert!(!settle(&mut p, &open_pr()));
2506        assert_eq!(
2507            [Trigger::Pushed].into_iter().collect::<BTreeSet<_>>(),
2508            p.triggers
2509        );
2510    }
2511
2512    #[test]
2513    fn a_review_request_counts_for_the_viewer_or_a_team_the_viewer_is_in() {
2514        let teams = |team: &str| match team {
2515            "example/ours" => Some(true),
2516            "example/theirs" => Some(false),
2517            _ => None,
2518        };
2519        let pending = |requests: &[Login]| review_pending(requests, "me", &teams);
2520        assert!(!pending(&[]));
2521        assert!(pending(&[user("ME")]));
2522        assert!(!pending(&[user("alice")]));
2523        assert!(pending(&[team("example/ours")]));
2524        // Approving clears the viewer's own request and leaves a CODEOWNERS
2525        // team the viewer is not in, which used to keep the request alive.
2526        assert!(!pending(&[team("example/theirs"), user("alice")]));
2527        assert!(pending(&[team("example/theirs"), team("example/ours")]));
2528        // A team GitHub would not answer for keeps the request.
2529        assert!(pending(&[team("example/unknown")]));
2530        // Asked by name, no team is looked up.
2531        let never = |_: &str| -> Option<bool> { panic!("looked a team up") };
2532        assert!(review_pending(
2533            &[team("example/ours"), user("me")],
2534            "me",
2535            &never
2536        ));
2537    }
2538
2539    /// A review posted as a comment leaves its request open, and a first sweep
2540    /// has never seen that request, so it read as the author asking again and
2541    /// got a "nothing has changed" note on every pull request reviewed that way.
2542    #[test]
2543    fn a_request_made_before_the_review_on_record_was_answered_by_it() {
2544        let teams = |team: &str| match team {
2545            "example/ours" => Some(true),
2546            _ => Some(false),
2547        };
2548        let made = |at: &str, who: Login| ReviewRequested { at: at.into(), who };
2549        let answered = |requests: &[ReviewRequested]| {
2550            request_answered(requests, "me", &teams, "2026-10-20T12:00:00Z")
2551        };
2552        assert!(answered(&[made("2026-10-20T11:00:00Z", user("me"))]));
2553        assert!(answered(&[made("2026-10-20T12:00:00Z", user("ME"))]));
2554        assert!(answered(&[made(
2555            "2026-10-20T11:00:00Z",
2556            team("example/ours")
2557        )]));
2558        // Asked again after the review: the newest request decides.
2559        assert!(!answered(&[
2560            made("2026-10-20T11:00:00Z", user("me")),
2561            made("2026-10-20T13:00:00Z", user("me")),
2562        ]));
2563        assert!(!answered(&[
2564            made("2026-10-20T11:00:00Z", user("me")),
2565            made("2026-10-20T13:00:00Z", team("example/ours")),
2566        ]));
2567        // Somebody else asked later, which asks nothing of the viewer.
2568        assert!(answered(&[
2569            made("2026-10-20T11:00:00Z", user("me")),
2570            made("2026-10-20T13:00:00Z", user("alice")),
2571            made("2026-10-20T13:00:00Z", team("example/theirs")),
2572        ]));
2573        // Nothing on the timeline to go by, or no time to compare with.
2574        assert!(!answered(&[]));
2575        assert!(!answered(&[made("2026-10-20T11:00:00Z", user("alice"))]));
2576        assert!(!answered(&[made("", user("me"))]));
2577        assert!(!request_answered(
2578            &[made("2026-10-20T11:00:00Z", user("me"))],
2579            "me",
2580            &teams,
2581            ""
2582        ));
2583    }
2584
2585    #[test]
2586    fn a_review_request_or_a_mention_on_a_reviewed_pull_request_is_a_recheck() {
2587        let f = Facts {
2588            has_posted_record: true,
2589            ..open_pr()
2590        };
2591        for triggers in [
2592            &[Trigger::ReviewRequested][..],
2593            &[Trigger::Mentioned],
2594            &[Trigger::Pushed, Trigger::Replied, Trigger::ReviewRequested],
2595        ] {
2596            assert_eq!(
2597                Action::Recheck(7),
2598                plan(&pending(ItemKind::Pr, triggers, 0), &f, &cfg()),
2599                "{triggers:?}"
2600            );
2601        }
2602    }
2603
2604    /// reardencode/rbitcoin#1137: another bot's review on a pull request spar
2605    /// had approved, with no new commit and nobody asking, got a second look
2606    /// that said nothing new at length.
2607    #[test]
2608    fn a_push_a_reply_or_an_assignment_on_a_reviewed_pull_request_asks_for_nothing() {
2609        let f = Facts {
2610            has_posted_record: true,
2611            ..open_pr()
2612        };
2613        for triggers in [
2614            &[Trigger::Pushed][..],
2615            &[Trigger::Replied],
2616            &[Trigger::Assigned],
2617            &[Trigger::Pushed, Trigger::Replied],
2618        ] {
2619            assert!(
2620                matches!(
2621                    plan(&pending(ItemKind::Pr, triggers, 0), &f, &cfg()),
2622                    Action::Ignore(_)
2623                ),
2624                "{triggers:?}"
2625            );
2626        }
2627    }
2628
2629    #[test]
2630    fn a_mention_from_before_the_review_on_record_is_only_an_update() {
2631        let reviewed = Facts {
2632            has_posted_record: true,
2633            ..open_pr()
2634        };
2635        let mut p = pending(ItemKind::Pr, &[Trigger::Mentioned], 0);
2636        assert!(settle_mention(&mut p, &reviewed));
2637        assert_eq!(
2638            [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2639            p.triggers
2640        );
2641        assert!(matches!(plan(&p, &reviewed, &cfg()), Action::Ignore(_)));
2642
2643        // Made since the review, it asks.
2644        let asked = Facts {
2645            mentioned: true,
2646            ..reviewed.clone()
2647        };
2648        let mut p = pending(ItemKind::Pr, &[Trigger::Mentioned], 0);
2649        assert!(!settle_mention(&mut p, &asked));
2650        assert_eq!(Action::Recheck(7), plan(&p, &asked, &cfg()));
2651
2652        // On a pull request spar never reviewed there is no review to be
2653        // older than, and a mention starts the first one.
2654        let mut p = pending(ItemKind::Pr, &[Trigger::Mentioned], 0);
2655        assert!(!settle_mention(&mut p, &open_pr()));
2656        assert_eq!(Action::Review, plan(&p, &open_pr(), &cfg()));
2657
2658        // An issue is held to its reviewed pull request's review.
2659        let linked = Facts {
2660            number: 42,
2661            open: true,
2662            linked_pr_with_record: Some(108),
2663            ..Default::default()
2664        };
2665        let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2666        assert!(settle_mention(&mut p, &linked));
2667        assert!(matches!(plan(&p, &linked, &cfg()), Action::Ignore(_)));
2668        let asked = Facts {
2669            mentioned: true,
2670            ..linked
2671        };
2672        let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2673        assert!(!settle_mention(&mut p, &asked));
2674        assert_eq!(Action::Recheck(108), plan(&p, &asked, &cfg()));
2675    }
2676
2677    /// spar's own answer moves the thread it answered, and the thread keeps
2678    /// the reason it was raised for, so it comes back as a mention. Only a
2679    /// mention `spar answer` has not judged is one.
2680    #[test]
2681    fn an_issue_thread_moved_by_spars_own_answer_is_only_an_update() {
2682        let judged = Facts {
2683            number: 229,
2684            open: true,
2685            ..Default::default()
2686        };
2687        let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2688        assert!(settle_mention(&mut p, &judged));
2689        assert_eq!(
2690            [Trigger::Replied].into_iter().collect::<BTreeSet<_>>(),
2691            p.triggers
2692        );
2693        assert!(matches!(plan(&p, &judged, &cfg()), Action::Ignore(_)));
2694
2695        let asked = Facts {
2696            asks_open: true,
2697            ..judged
2698        };
2699        let mut p = pending(ItemKind::Issue, &[Trigger::Mentioned], 0);
2700        assert!(!settle_mention(&mut p, &asked));
2701        assert_eq!(Action::Answer { draft: false }, plan(&p, &asked, &cfg()));
2702    }
2703
2704    #[test]
2705    fn a_review_request_or_an_assignment_on_an_unreviewed_pull_request_follows_the_config() {
2706        let f = open_pr();
2707        let mut c = cfg();
2708        assert_eq!(
2709            Action::Review,
2710            plan(
2711                &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2712                &f,
2713                &c
2714            )
2715        );
2716        assert_eq!(
2717            Action::Review,
2718            plan(&pending(ItemKind::Pr, &[Trigger::Assigned], 0), &f, &c)
2719        );
2720        assert_eq!(
2721            Action::Review,
2722            plan(&pending(ItemKind::Pr, &[Trigger::Mentioned], 0), &f, &c)
2723        );
2724        c.on_review_requested = ReviewOrIgnore::Ignore;
2725        c.on_assigned = ReviewOrIgnore::Ignore;
2726        assert!(matches!(
2727            plan(
2728                &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2729                &f,
2730                &c
2731            ),
2732            Action::Ignore(_)
2733        ));
2734        assert!(matches!(
2735            plan(&pending(ItemKind::Pr, &[Trigger::Assigned], 0), &f, &c),
2736            Action::Ignore(_)
2737        ));
2738        // A mention is always answered with a review.
2739        assert_eq!(
2740            Action::Review,
2741            plan(&pending(ItemKind::Pr, &[Trigger::Mentioned], 0), &f, &c)
2742        );
2743    }
2744
2745    #[test]
2746    fn a_push_to_an_unreviewed_pull_request_is_ignored_unless_it_was_named() {
2747        let f = open_pr();
2748        assert!(matches!(
2749            plan(&pending(ItemKind::Pr, &[Trigger::Pushed], 0), &f, &cfg()),
2750            Action::Ignore(_)
2751        ));
2752        assert!(matches!(
2753            plan(&pending(ItemKind::Pr, &[Trigger::Replied], 0), &f, &cfg()),
2754            Action::Ignore(_)
2755        ));
2756        let named = Facts {
2757            named: true,
2758            ..open_pr()
2759        };
2760        assert_eq!(
2761            Action::Review,
2762            plan(
2763                &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
2764                &named,
2765                &cfg()
2766            )
2767        );
2768        assert_eq!(
2769            Action::Review,
2770            plan(
2771                &pending(ItemKind::Pr, &[Trigger::Replied], 0),
2772                &named,
2773                &cfg()
2774            )
2775        );
2776    }
2777
2778    #[test]
2779    fn your_own_pull_request_follows_on_own_pr_and_a_push_to_it_is_ignored() {
2780        let f = Facts {
2781            author_is_viewer: true,
2782            has_posted_record: true,
2783            ..open_pr()
2784        };
2785        let mut c = cfg();
2786        assert_eq!(
2787            Action::Checkin { reply_only: false },
2788            plan(&pending(ItemKind::Pr, &[Trigger::Replied], 0), &f, &c)
2789        );
2790        assert_eq!(
2791            Action::Checkin { reply_only: false },
2792            plan(
2793                &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2794                &f,
2795                &c
2796            )
2797        );
2798        assert!(matches!(
2799            plan(&pending(ItemKind::Pr, &[Trigger::Pushed], 0), &f, &c),
2800            Action::Ignore(_)
2801        ));
2802        c.on_own_pr = OwnPr::Reply;
2803        assert_eq!(
2804            Action::Checkin { reply_only: true },
2805            plan(&pending(ItemKind::Pr, &[Trigger::Mentioned], 0), &f, &c)
2806        );
2807        c.on_own_pr = OwnPr::Ignore;
2808        assert!(matches!(
2809            plan(&pending(ItemKind::Pr, &[Trigger::Replied], 0), &f, &c),
2810            Action::Ignore(_)
2811        ));
2812    }
2813
2814    #[test]
2815    fn a_pull_request_in_custody_belongs_to_resume_and_a_capped_one_waits_for_a_person() {
2816        let custody = Facts {
2817            has_custody_state: true,
2818            ..open_pr()
2819        };
2820        assert_eq!(
2821            Action::BelongsToResume,
2822            plan(
2823                &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
2824                &custody,
2825                &cfg()
2826            )
2827        );
2828        let capped = Facts {
2829            has_posted_record: true,
2830            recheck_capped: true,
2831            ..open_pr()
2832        };
2833        assert_eq!(
2834            Action::Capped(7),
2835            plan(
2836                &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
2837                &capped,
2838                &cfg()
2839            )
2840        );
2841        assert!(matches!(
2842            plan(
2843                &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
2844                &capped,
2845                &cfg()
2846            ),
2847            Action::Ignore(_)
2848        ));
2849    }
2850
2851    /// The issue route used to recheck its pull request unconditionally, so a
2852    /// mention on the issue went round the cap and the own pull request rule
2853    /// both.
2854    #[test]
2855    fn an_issue_mention_keeps_to_its_pull_requests_cap_and_owner() {
2856        let linked = Facts {
2857            number: 42,
2858            open: true,
2859            linked_pr_with_record: Some(108),
2860            ..Default::default()
2861        };
2862        let capped = Facts {
2863            recheck_capped: true,
2864            ..linked.clone()
2865        };
2866        assert_eq!(
2867            Action::Capped(108),
2868            plan(
2869                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2870                &capped,
2871                &cfg()
2872            )
2873        );
2874        let own = Facts {
2875            author_is_viewer: true,
2876            ..linked
2877        };
2878        assert!(matches!(
2879            plan(
2880                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2881                &own,
2882                &cfg()
2883            ),
2884            Action::Ignore(_)
2885        ));
2886    }
2887
2888    #[test]
2889    fn an_issue_mention_without_a_pull_request_is_answered_and_with_one_rechecks_it() {
2890        let issue = Facts {
2891            number: 42,
2892            open: true,
2893            ..Default::default()
2894        };
2895        // Every mention on it judged already: nothing asks.
2896        assert!(matches!(
2897            plan(
2898                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2899                &issue,
2900                &cfg()
2901            ),
2902            Action::Ignore(_)
2903        ));
2904        let asks = Facts {
2905            asks_open: true,
2906            ..issue.clone()
2907        };
2908        assert_eq!(
2909            Action::Answer { draft: false },
2910            plan(
2911                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2912                &asks,
2913                &cfg()
2914            )
2915        );
2916        let mut c = cfg();
2917        c.on_mention = OnMention::Draft;
2918        assert_eq!(
2919            Action::Answer { draft: true },
2920            plan(
2921                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2922                &asks,
2923                &c
2924            )
2925        );
2926        c.on_mention = OnMention::Ignore;
2927        assert_eq!(
2928            Action::Ignore("on_mention is ignore"),
2929            plan(
2930                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2931                &asks,
2932                &c
2933            )
2934        );
2935        let capped = Facts {
2936            answers_capped: true,
2937            ..asks.clone()
2938        };
2939        assert_eq!(
2940            Action::AnswersCapped,
2941            plan(
2942                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2943                &capped,
2944                &cfg()
2945            )
2946        );
2947        assert!(matches!(
2948            plan(
2949                &pending(ItemKind::Issue, &[Trigger::Replied], 0),
2950                &issue,
2951                &cfg()
2952            ),
2953            Action::Ignore(_)
2954        ));
2955        let linked = Facts {
2956            linked_pr_with_record: Some(108),
2957            ..issue.clone()
2958        };
2959        assert_eq!(
2960            Action::Recheck(108),
2961            plan(
2962                &pending(ItemKind::Issue, &[Trigger::Mentioned], 0),
2963                &linked,
2964                &cfg()
2965            )
2966        );
2967        // Only a mention asks. An issue thread that began with an assignment
2968        // says so on every later comment, and a comment asks for nothing, on a
2969        // named issue too.
2970        assert!(matches!(
2971            plan(
2972                &pending(ItemKind::Issue, &[Trigger::Assigned], 0),
2973                &linked,
2974                &cfg()
2975            ),
2976            Action::Ignore(_)
2977        ));
2978        let named = Facts {
2979            named: true,
2980            ..linked
2981        };
2982        assert!(matches!(
2983            plan(
2984                &pending(ItemKind::Issue, &[Trigger::Replied], 0),
2985                &named,
2986                &cfg()
2987            ),
2988            Action::Ignore(_)
2989        ));
2990    }
2991
2992    #[test]
2993    fn a_closed_item_is_dropped_and_one_below_the_floor_is_ignored_unless_named() {
2994        let closed = Facts {
2995            open: false,
2996            has_posted_record: true,
2997            ..open_pr()
2998        };
2999        assert_eq!(
3000            Action::Drop,
3001            plan(
3002                &pending(ItemKind::Pr, &[Trigger::Pushed], 0),
3003                &closed,
3004                &cfg()
3005            )
3006        );
3007        let f = Facts {
3008            has_posted_record: true,
3009            ..open_pr()
3010        };
3011        assert_eq!(
3012            Action::Drop,
3013            plan(
3014                &pending(ItemKind::Pr, &[Trigger::Pushed, Trigger::Closed], 0),
3015                &f,
3016                &cfg()
3017            )
3018        );
3019        let low = Facts {
3020            below_floor: true,
3021            has_posted_record: true,
3022            ..open_pr()
3023        };
3024        assert!(matches!(
3025            plan(&pending(ItemKind::Pr, &[Trigger::Pushed], 0), &low, &cfg()),
3026            Action::Ignore(_)
3027        ));
3028        let named_low = Facts { named: true, ..low };
3029        assert_eq!(
3030            Action::Recheck(7),
3031            plan(
3032                &pending(ItemKind::Pr, &[Trigger::ReviewRequested], 0),
3033                &named_low,
3034                &cfg()
3035            )
3036        );
3037    }
3038
3039    // -- settling --------------------------------------------------------------------
3040
3041    #[test]
3042    fn an_item_is_due_once_quiet_for_the_settle_time_or_at_max_wait_anyway() {
3043        let c = cfg();
3044        let mut p = pending(ItemKind::Pr, &[Trigger::Pushed], 1000);
3045        assert!(!due(&p, 1000, &c, false));
3046        assert!(!due(&p, 1000 + c.settle - 1, &c, false));
3047        assert!(due(&p, 1000 + c.settle, &c, false));
3048        // Noise every minute keeps resetting the quiet period.
3049        p.last_seen = 1000 + c.max_wait - 30;
3050        assert!(!due(&p, 1000 + c.max_wait - 1, &c, false));
3051        assert!(due(&p, 1000 + c.max_wait, &c, false));
3052        assert!(due(&p, 1000, &c, true), "--once waits for nothing");
3053    }
3054
3055    #[test]
3056    fn events_merge_into_one_pending_record_per_item_and_the_next_due_is_the_earliest() {
3057        let c = cfg();
3058        let mut pending = BTreeMap::new();
3059        let push = Event {
3060            number: 7,
3061            kind: ItemKind::Pr,
3062            trigger: Trigger::Pushed,
3063            who: None,
3064            head: Some(("aaa".into(), "bbb".into())),
3065            commits: None,
3066            from_thread: false,
3067        };
3068        merge(&mut pending, &push, 1000);
3069        let reply = Event {
3070            trigger: Trigger::Replied,
3071            who: Some("alice".into()),
3072            head: Some(("bbb".into(), "ccc".into())),
3073            ..push.clone()
3074        };
3075        merge(&mut pending, &reply, 1060);
3076        let other = Event {
3077            number: 8,
3078            ..push.clone()
3079        };
3080        merge(&mut pending, &other, 1100);
3081        assert_eq!(2, pending.len());
3082        let p = &pending[&7];
3083        assert_eq!(1000, p.first_seen);
3084        assert_eq!(1060, p.last_seen);
3085        assert_eq!(Some("aaa".to_string()), p.head_from);
3086        assert_eq!(Some("ccc".to_string()), p.head_to);
3087        assert_eq!(vec!["alice".to_string()], p.who);
3088        assert_eq!("a push and a reply", p.because());
3089        let mut state = ListenState {
3090            pending,
3091            ..Default::default()
3092        };
3093        let all = |_: i64| true;
3094        assert_eq!(Some(1060 + c.settle), next_due(&state, &c, &all));
3095        assert_eq!(None, next_due(&ListenState::default(), &c, &all));
3096
3097        // Waiting out a failure is not being due, or the loop wakes every
3098        // second until the wait ends.
3099        state.watched(7, ItemKind::Pr).retry_after = 5000;
3100        assert_eq!(Some(1100 + c.settle), next_due(&state, &c, &all));
3101        state.watched(8, ItemKind::Pr).retry_after = 6000;
3102        assert_eq!(Some(5000), next_due(&state, &c, &all));
3103        // Nor is something this run does not act on.
3104        assert_eq!(Some(6000), next_due(&state, &c, &|n| n == 8));
3105    }
3106
3107    // -- durations, state, backoff ------------------------------------------------------
3108
3109    #[test]
3110    fn since_accepts_minutes_hours_days_and_combinations() {
3111        assert_eq!(Duration::from_secs(1800), parse_duration("30m").unwrap());
3112        assert_eq!(Duration::from_secs(7200), parse_duration("2h").unwrap());
3113        assert_eq!(Duration::from_secs(86_400), parse_duration("1d").unwrap());
3114        assert_eq!(
3115            Duration::from_secs(5400),
3116            parse_duration(" 1h30m ").unwrap()
3117        );
3118        assert_eq!(Duration::from_secs(45), parse_duration("45s").unwrap());
3119    }
3120
3121    #[test]
3122    fn since_rejects_a_bare_number_and_says_what_it_takes() {
3123        for bad in ["", "30", "2 hours", "h", "1w", "1h30"] {
3124            let err = parse_duration(bad).unwrap_err().to_string();
3125            assert!(err.contains("like 30m, 2h or 1d"), "{bad:?}: {err}");
3126        }
3127    }
3128
3129    #[test]
3130    fn listen_state_round_trips_through_json_and_old_seen_entries_are_pruned() {
3131        let mut state = ListenState {
3132            version: STATE_VERSION,
3133            watermark: "2026-10-20T10:00:00Z".into(),
3134            last_modified: Some("Tue, 20 Oct 2026 10:00:00 GMT".into()),
3135            ..Default::default()
3136        };
3137        state
3138            .seen
3139            .insert("old".into(), "2026-10-01T10:00:00Z".into());
3140        state
3141            .seen
3142            .insert("new".into(), "2026-10-20T09:00:00Z".into());
3143        state.seen.insert("junk".into(), "yesterday".into());
3144        state.watched(7, ItemKind::Pr).head = Some("aaa".into());
3145        state
3146            .pending
3147            .insert(7, pending(ItemKind::Pr, &[Trigger::Pushed], 5));
3148        let text = serde_json::to_string_pretty(&state).unwrap();
3149        assert!(text.contains("\"pushed\""), "{text}");
3150        assert!(text.contains("\"kind\": \"pr\""), "{text}");
3151        let back: ListenState = serde_json::from_str(&text).unwrap();
3152        assert_eq!(state, back);
3153
3154        let now = clock::parse_rfc3339("2026-10-20T10:00:00Z").unwrap();
3155        prune_seen(&mut state, now);
3156        let left: Vec<&String> = state.seen.keys().collect();
3157        assert_eq!(vec!["junk", "new"], left);
3158    }
3159
3160    #[test]
3161    fn gh_failures_back_off_from_a_minute_to_fifteen_and_dispatches_from_fifteen_to_six_hours() {
3162        assert_eq!(60, backoff_secs(1));
3163        assert_eq!(120, backoff_secs(2));
3164        assert_eq!(480, backoff_secs(4));
3165        assert_eq!(900, backoff_secs(5));
3166        assert_eq!(900, backoff_secs(40));
3167        assert_eq!(900, dispatch_backoff_secs(1));
3168        assert_eq!(3600, dispatch_backoff_secs(3));
3169        assert_eq!(6 * 3600, dispatch_backoff_secs(9));
3170    }
3171
3172    /// A pull request can sit minutes behind its branch. A look that stops
3173    /// for that, or for a push while it ran, goes again once the push settles,
3174    /// not a quarter of an hour later and doubling.
3175    #[test]
3176    fn a_head_that_moved_runs_again_once_it_settles_and_a_failure_backs_off() {
3177        let c = cfg();
3178        for failures in [1, 2, 4] {
3179            assert_eq!(c.settle, retry_wait(true, failures, &c));
3180            assert_eq!(
3181                dispatch_backoff_secs(failures),
3182                retry_wait(false, failures, &c)
3183            );
3184        }
3185        let quick = ListenCfg { settle: 1, ..cfg() };
3186        assert_eq!(FLOOR_SECS, retry_wait(true, 1, &quick));
3187    }
3188
3189    /// A first review given up on while GitHub lagged came back as the push
3190    /// alone once GitHub caught up, and a push to a pull request spar has not
3191    /// reviewed is ignored, so the request was lost for good.
3192    #[test]
3193    fn giving_up_holds_what_the_item_was_waiting_on_until_it_changes() {
3194        let mut state = ListenState::default();
3195        let asked = Event {
3196            number: 7,
3197            kind: ItemKind::Pr,
3198            trigger: Trigger::ReviewRequested,
3199            who: None,
3200            head: None,
3201            commits: None,
3202            from_thread: true,
3203        };
3204        state.note(&asked, 0);
3205        state.give_up(7, ItemKind::Pr);
3206        assert!(state.pending.is_empty(), "{state:?}");
3207        assert_eq!(
3208            [Trigger::ReviewRequested]
3209                .into_iter()
3210                .collect::<BTreeSet<_>>(),
3211            state.items[&7].held
3212        );
3213
3214        let shown = Event {
3215            trigger: Trigger::Pushed,
3216            head: Some(("aaa".into(), "bbb".into())),
3217            from_thread: false,
3218            ..asked.clone()
3219        };
3220        state.note(&shown, 600);
3221        let p = &state.pending[&7];
3222        assert_eq!(
3223            [Trigger::Pushed, Trigger::ReviewRequested]
3224                .into_iter()
3225                .collect::<BTreeSet<_>>(),
3226            p.triggers
3227        );
3228        assert!(state.items[&7].held.is_empty(), "{state:?}");
3229        let asked_now = Facts {
3230            review_pending: true,
3231            ..open_pr()
3232        };
3233        assert_eq!(Action::Review, plan(p, &asked_now, &cfg()));
3234        // Brought back once, not on every change after.
3235        state.pending.clear();
3236        state.note(&shown, 700);
3237        assert_eq!(
3238            [Trigger::Pushed].into_iter().collect::<BTreeSet<_>>(),
3239            state.pending[&7].triggers
3240        );
3241    }
3242
3243    #[test]
3244    fn number_from_url_reads_the_last_segment() {
3245        assert_eq!(
3246            Some(108),
3247            number_from_url("https://api.github.com/repos/o/r/pulls/108")
3248        );
3249        assert_eq!(
3250            Some(42),
3251            number_from_url("https://api.github.com/repos/o/r/issues/42/")
3252        );
3253        assert_eq!(
3254            None,
3255            number_from_url("https://api.github.com/repos/o/r/pulls")
3256        );
3257    }
3258
3259    #[cfg(unix)]
3260    #[test]
3261    fn a_held_lock_refuses_a_second_listener_however_old_and_a_left_file_is_taken_over() {
3262        let dir = std::env::temp_dir().join(format!(
3263            "spar-listen-lock-{}-{}",
3264            std::process::id(),
3265            clock::now_secs()
3266        ));
3267        std::fs::create_dir_all(&dir).unwrap();
3268        let path = dir.join("listen.lock");
3269
3270        // Held is held, however long ago it was taken: nothing has to keep a
3271        // lock fresh while one review runs for an hour.
3272        let held = Lock::at(path.clone(), 1000).expect("first lock");
3273        for now in [1000, 1000 + 86_400] {
3274            let err = Lock::at(path.clone(), now).unwrap_err().to_string();
3275            assert!(err.contains("already watching"), "{err}");
3276            assert!(err.contains(&std::process::id().to_string()), "{err}");
3277        }
3278        assert!(path.exists(), "a refused listener leaves the lock alone");
3279        drop(held);
3280        assert!(!path.exists(), "dropping the lock removes the file");
3281
3282        // A file a listener left behind, even one naming a pid that is alive,
3283        // as a reboot can arrange, holds nothing.
3284        std::fs::write(
3285            &path,
3286            serde_json::to_string(&LockFile {
3287                pid: std::process::id(),
3288                started: "x".into(),
3289            })
3290            .unwrap(),
3291        )
3292        .unwrap();
3293        let taken = Lock::at(path.clone(), 1000).expect("a left file is taken over");
3294        drop(taken);
3295        let _ = std::fs::remove_dir_all(&dir);
3296    }
3297}