Skip to main content

magi/
release_watch.rs

1//! Watches every open release-bump pull request until it lands.
2//!
3//! [`crate::bump`] opens a `chore/release-vX.Y.Z` pull request with auto-merge
4//! armed and then stops looking. A required check that goes red (a flaky
5//! windows or browser job) leaves auto-merge waiting forever, and nothing
6//! pages anybody because the notices there only fire when the pull request
7//! cannot be *opened*. This module is the part that keeps looking.
8//!
9//! It runs inside `magi serve` as its own task, like [`crate::waiter`], and is
10//! tied to no run: a pending bump is coalesced across runs and outlives the
11//! one that opened it. The forge is the source of truth (every open pull
12//! request whose head branch starts with `chore/release-v`, in every checkout
13//! magi knows); the pending marker is not consulted, so a missing one changes
14//! nothing.
15//!
16//! - **The decision is pure.** [`decide`] maps a checks snapshot and the
17//!   recorded [`WatchState`] to one [`Step`]. A forge answer that cannot be
18//!   read is `Unknown`: nothing moves, not even the stall clock.
19//! - **A failed run is rerun once, durably.** The run id is written to the
20//!   state file *before* `gh run rerun <id> --failed` is called, so a restart
21//!   never reruns twice; if the call then fails, the cost is one escalation to
22//!   a human, never a rerun loop.
23//! - **Escalation is a notice plus a question** ([`crate::bump::NOTICE_NODE`],
24//!   no `cwd`, no run). Nothing is ever merged, closed or pushed: the watcher
25//!   only reruns jobs. Silence is a hold. The owner's choice is applied by the
26//!   watcher itself on its next lap, recorded in `WatchState::applied` so an
27//!   answer is applied once.
28//! - **`[release] mode = "local"` replaces all of the above for that
29//!   repository's release pull requests.** No check is awaited, reran or
30//!   stalled on. The watcher asks the owner (`merge` / `hold`, bound to the
31//!   head it observed) unless `graph.land_approval` is off, in which case it
32//!   merges that head without asking. It merges with `--match-head-commit`, and after the merge
33//!   drives [`crate::release_local`] - tag, then the configured commands - with
34//!   its progress in the watch record. A failure is a notice and a question
35//!   (`retry` / `leave it`), never a blind retry. See [`decide_local`].
36//! - **Notice wording is fixed per stage** and carries only the pull request,
37//!   so a repeated poll never relights it; everything that varies (job names,
38//!   links) rides in the question.
39
40use std::collections::BTreeMap;
41use std::future::Future;
42use std::path::{Path, PathBuf};
43use std::pin::Pin;
44use std::time::Duration;
45
46use anyhow::{Result, bail};
47use serde::{Deserialize, Serialize};
48
49use crate::ask::{Answer, Question, Questions};
50use crate::land::{self, PrLifecycle, RollupView, Verdict};
51use crate::notices::{self, Link, Notice, Notices};
52use crate::release_local::{self, Job};
53use crate::run::RunState;
54
55/// Pause between laps.
56pub(crate) const LAP: Duration = Duration::from_secs(300);
57
58/// How long after a rerun the forge may still report the old failure before
59/// that is read as "still red". `gh run rerun` returns before the checks flip
60/// to pending.
61pub(crate) const RERUN_GRACE: i64 = 180;
62
63/// Choice: rerun the failed jobs once more. The three choices are matched
64/// verbatim when applied.
65pub const RERUN_AGAIN: &str = "rerun again";
66/// Choice: stay quiet until a check or the head changes.
67pub const HOLD: &str = "hold";
68/// Choice: stop watching this pull request.
69pub const LEAVE_IT: &str = "leave it";
70/// Choice (local mode): run the failed release again from where it stopped.
71pub const RETRY: &str = "retry";
72
73/// What one look at one pull request concludes.
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub(crate) enum Step {
76    /// Merged or closed: stop watching.
77    Done,
78    /// Nothing to do yet.
79    Wait,
80    /// The forge answer could not be read; change nothing.
81    Unknown,
82    /// Rerun the failed jobs of these workflow runs, once each.
83    Rerun(Vec<String>),
84    /// A human is needed.
85    Escalate(Why),
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub(crate) enum Why {
90    /// Still red after the one rerun, or red with nothing to rerun.
91    StillRed,
92    /// No progress for longer than `[daemon] release_stall_minutes`.
93    Stalled,
94}
95
96/// Everything remembered about one watched pull request.
97#[derive(Debug, Clone, Default, Serialize, Deserialize)]
98#[serde(default)]
99pub(crate) struct WatchState {
100    /// Checkout the pull request belongs to.
101    pub repo: String,
102    pub url: String,
103    /// Head commit the rest of this record is about.
104    pub head: String,
105    /// Workflow runs already rerun for this head, with when (unix seconds).
106    pub reruns: BTreeMap<String, i64>,
107    /// Digest of the head and every check's verdict, and when it last changed.
108    pub fingerprint: String,
109    pub progress_at: i64,
110    /// The open question asked about this pull request.
111    pub question: Option<String>,
112    /// Fingerprint the owner chose to hold: no new question until it changes.
113    pub held: Option<String>,
114    /// The owner said to leave it.
115    pub ignored: bool,
116    /// Question ids whose answer was already applied.
117    pub applied: Vec<String>,
118    /// Local mode: the head the open (or last) approval question was about.
119    pub asked_head: Option<String>,
120    /// Local mode: the head the owner held or whose merge was refused; no new
121    /// approval question until the head moves.
122    pub held_head: Option<String>,
123    /// Local mode: the run that opened this release PR, so a failed release can
124    /// hold the task that run finished (the daemon marks it Done at the merge).
125    pub run: Option<String>,
126    /// Local mode: the task this watcher put on hold for a failed release, so a
127    /// successful retry can give it back (and a restart does not hold it twice).
128    pub held_task: Option<String>,
129    /// Local mode: the release after the merge.
130    pub job: Option<Job>,
131    /// Set when this pull request was last seen as a local-mode release, so a
132    /// lap that cannot read the config does not fall back to Actions rules.
133    pub local: bool,
134}
135
136impl WatchState {
137    /// Whether this record belongs to a local-mode release. Records written
138    /// before `local` existed lack the flag, but only local mode ever sets
139    /// `run`, `asked_head`, `held_head` or `job`.
140    fn is_local(&self) -> bool {
141        self.local
142            || self.run.is_some()
143            || self.asked_head.is_some()
144            || self.held_head.is_some()
145            || self.job.is_some()
146    }
147}
148
149pub(crate) fn is_failed(v: Verdict) -> bool {
150    v == Verdict::Fail
151}
152
153/// Digest of what "progress" means: the head and every check's verdict.
154pub(crate) fn fingerprint(snap: &RollupView) -> String {
155    let mut parts: Vec<String> = snap
156        .checks
157        .iter()
158        .map(|c| format!("{}={:?}", c.name, c.verdict))
159        .collect();
160    parts.sort();
161    format!("{}|{}", snap.head, parts.join(","))
162}
163
164/// Fold a fresh snapshot into the record: a new head forgets the old head's
165/// reruns, and a changed fingerprint restarts the stall clock.
166pub(crate) fn observe(st: &mut WatchState, snap: &RollupView, now: i64) {
167    if st.head != snap.head {
168        st.reruns.clear();
169        st.head = snap.head.clone();
170    }
171    let fp = fingerprint(snap);
172    if st.fingerprint != fp {
173        st.fingerprint = fp;
174        st.progress_at = now;
175    }
176}
177
178/// Decide what to do about a pull request. `snap` is `None` when the forge
179/// could not be read. `stall_secs == 0` disables the stall rule.
180pub(crate) fn decide(
181    snap: Option<&RollupView>,
182    st: &WatchState,
183    now: i64,
184    stall_secs: i64,
185) -> Step {
186    let Some(snap) = snap else {
187        return Step::Unknown;
188    };
189    if snap.state != PrLifecycle::Open {
190        return Step::Done;
191    }
192    if st.ignored || st.held.as_deref() == Some(fingerprint(snap).as_str()) {
193        return Step::Wait;
194    }
195
196    let any_pending = snap.checks.iter().any(|c| c.verdict == Verdict::Pending);
197    // A run is complete once none of its own checks is still pending; another
198    // run's pending job says nothing about it.
199    let complete = |run: &str| {
200        !snap
201            .checks
202            .iter()
203            .any(|c| c.verdict == Verdict::Pending && c.run.as_deref() == Some(run))
204    };
205    let mut fresh = Vec::new();
206    let mut spent_recently = false;
207    let mut spent = false;
208    let mut external = false;
209    for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
210        match c.run.as_deref() {
211            None => external = true,
212            Some(run) if !complete(run) => {}
213            Some(run) => match st.reruns.get(run) {
214                None => {
215                    if !fresh.iter().any(|r: &String| r == run) {
216                        fresh.push(run.to_owned());
217                    }
218                }
219                Some(&at) if now - at < RERUN_GRACE => spent_recently = true,
220                Some(_) => spent = true,
221            },
222        }
223    }
224    if !fresh.is_empty() {
225        return Step::Rerun(fresh);
226    }
227    if spent_recently {
228        return Step::Wait;
229    }
230    if spent || (external && !any_pending) {
231        return Step::Escalate(Why::StillRed);
232    }
233    if stall_secs > 0 && now - st.progress_at >= stall_secs {
234        return Step::Escalate(Why::Stalled);
235    }
236    Step::Wait
237}
238
239/// `owner/repo#N` out of a pull request url, the stable name for notices.
240pub(crate) fn pr_key(url: &str) -> Option<String> {
241    let rest = url.split("://").nth(1)?;
242    let mut it = rest.split('/');
243    let _host = it.next()?;
244    let owner = it.next()?;
245    let repo = it.next()?;
246    if it.next()? != "pull" {
247        return None;
248    }
249    let n: String = it
250        .next()?
251        .chars()
252        .take_while(char::is_ascii_digit)
253        .collect();
254    (!n.is_empty()).then(|| format!("{owner}/{repo}#{n}"))
255}
256
257fn notice_key(pr: &str) -> String {
258    format!("release-pr:{pr}")
259}
260
261/// Fixed per stage: no counts, times or job names.
262fn rerun_message(pr: &str) -> String {
263    format!("Release PR {pr}: failed checks were rerun once")
264}
265
266fn human_message(pr: &str) -> String {
267    format!("Release PR {pr} is stuck and needs a human")
268}
269
270fn failed_workflows(snap: &RollupView) -> Vec<String> {
271    snap.checks
272        .iter()
273        .filter(|c| is_failed(c.verdict))
274        .filter_map(|c| c.run.clone())
275        .collect::<std::collections::BTreeSet<_>>()
276        .into_iter()
277        .collect()
278}
279
280fn maintenance_choices(snap: &RollupView) -> Vec<String> {
281    let mut choices = vec![HOLD.to_owned(), LEAVE_IT.to_owned()];
282    if !failed_workflows(snap).is_empty() {
283        choices.insert(0, RERUN_AGAIN.to_owned());
284    }
285    choices
286}
287
288fn question_detail(snap: &RollupView, why: Why, st: &WatchState) -> String {
289    let mut s = String::new();
290    s.push_str(&format!("Pull request: {}\n\n", snap.url));
291    s.push_str(match why {
292        Why::StillRed => "Checks are still red after the failed jobs were rerun once.\n\n",
293        Why::Stalled => "The pull request has made no progress for too long.\n\n",
294    });
295    let failed: Vec<_> = snap
296        .checks
297        .iter()
298        .filter(|c| is_failed(c.verdict))
299        .collect();
300    if failed.is_empty() {
301        s.push_str("No check is failing.\n");
302    } else {
303        s.push_str("Failed jobs:\n");
304        for c in failed {
305            match &c.url {
306                Some(u) => s.push_str(&format!("- {} ({u})\n", c.name)),
307                None => s.push_str(&format!("- {}\n", c.name)),
308            }
309        }
310    }
311    s.push_str(&format!(
312        "\nWorkflow runs rerun so far: {}.\n\n\
313         - `{RERUN_AGAIN}`: rerun the failed jobs once more.\n\
314         - `{HOLD}`: stay quiet until a check or the head changes.\n\
315         - `{LEAVE_IT}`: stop watching this pull request.\n\n\
316         Nothing is merged or pushed by magi; silence is a hold.\n",
317        st.reruns.len()
318    ));
319    if failed_workflows(snap).is_empty() {
320        s = s.replace(
321            &format!("- `{RERUN_AGAIN}`: rerun the failed jobs once more.\n"),
322            "",
323        );
324    }
325    s
326}
327
328/// Where the owner's approval stands, as far as [`decide_local`] cares.
329#[derive(Debug, Clone, Copy, PartialEq, Eq)]
330pub(crate) enum Approved {
331    /// No question was filed (or it is gone).
332    NoQuestion,
333    /// Filed, not answered.
334    Open,
335    /// The owner said `merge`.
336    Merge,
337    /// The owner said anything else, or it was abandoned: silence is a hold.
338    Hold,
339}
340
341/// What one look at an open local-mode release pull request concludes.
342#[derive(Debug, Clone, PartialEq, Eq)]
343pub(crate) enum LocalStep {
344    /// Nothing to do now.
345    Wait,
346    /// Ask the owner whether to merge this head.
347    Ask,
348    /// Merge, pinned to this head.
349    Merge(String),
350    /// Remember the owner's no for this head and stay quiet until it moves.
351    Hold(String),
352}
353
354/// Decide an *open* local-mode pull request. No I/O, and no CI: nothing is
355/// awaited, rerun or stalled on, because nothing will ever report. The head is
356/// what the owner approved: an answer about an older head asks again.
357///
358/// `ask_first` is `graph.land_approval`. When it is off, a head nobody has been
359/// asked about is merged straight away, pinned to that head; an approval
360/// question that is already open is still respected.
361pub(crate) fn decide_local(
362    head: &str,
363    st: &WatchState,
364    approved: Approved,
365    ask_first: bool,
366) -> LocalStep {
367    if st.ignored || st.held_head.as_deref() == Some(head) {
368        return LocalStep::Wait;
369    }
370    let about_this_head = st.asked_head.as_deref() == Some(head);
371    match approved {
372        Approved::Open => LocalStep::Wait,
373        Approved::NoQuestion if !ask_first => LocalStep::Merge(head.to_owned()),
374        Approved::NoQuestion => LocalStep::Ask,
375        Approved::Merge if about_this_head => LocalStep::Merge(head.to_owned()),
376        Approved::Hold if about_this_head => LocalStep::Hold(head.to_owned()),
377        // The head moved after the question was filed.
378        Approved::Merge | Approved::Hold if !ask_first => LocalStep::Merge(head.to_owned()),
379        Approved::Merge | Approved::Hold => LocalStep::Ask,
380    }
381}
382
383/// Start watching a release pull request before the first lap sees it, so one
384/// merged within a lap of being opened still gets released. Never overwrites.
385pub(crate) fn register(home: &Path, repo: &Path, url: &str, run: &str) {
386    let Some(pr) = pr_key(url) else {
387        return;
388    };
389    let w = Watcher::new(Box::new(GhForge), home.to_path_buf());
390    if w.state_path(&pr).exists() {
391        return;
392    }
393    let st = WatchState {
394        repo: repo.to_string_lossy().into_owned(),
395        url: url.to_owned(),
396        run: Some(run.to_owned()),
397        local: true,
398        ..WatchState::default()
399    };
400    w.save(&pr, &st);
401}
402
403/// Saved explicit ownership links, loaded only for batches with watch questions.
404pub(crate) fn states_for_questions(home: &Path, questions: &[Question]) -> Vec<WatchState> {
405    if !questions
406        .iter()
407        .any(|q| q.node == crate::bump::NOTICE_NODE && q.seat == "release-watch")
408    {
409        return Vec::new();
410    }
411    Watcher::new(Box::new(GhForge), home.to_path_buf()).stored()
412}
413
414/// The watch record whose open question is `question_id`, read from the magi
415/// `home`. `None` when no record names it (never guessed from the text).
416pub(crate) fn state_for_question(home: &Path, question_id: &str) -> Option<WatchState> {
417    Watcher::new(Box::new(GhForge), home.to_path_buf())
418        .stored()
419        .into_iter()
420        .find(|s| s.question.as_deref() == Some(question_id))
421}
422
423/// Resolve a release notification through its saved watch record, never its prose.
424pub(crate) fn question_for_notice(home: &Path, key: &str) -> Option<String> {
425    let pr = key.strip_prefix("release-pr:")?;
426    let state = Watcher::new(Box::new(GhForge), home.to_path_buf()).load(pr);
427    (pr_key(&state.url).as_deref() == Some(pr))
428        .then_some(state.question)
429        .flatten()
430}
431
432/// The task a release-watch question's answer could touch: the one that
433/// finished the run which opened the release pull request (`WatchState::run`).
434/// `None` when the record or the run is not known - then there is nothing to
435/// check, and the settle goes on (an answer here never releases a task).
436pub fn task_of_question(
437    home: &Path,
438    q: &Question,
439    queue: &crate::queue::Queue,
440) -> Option<crate::queue::Task> {
441    let run = state_for_question(home, &q.id)?.run?;
442    queue.list().into_iter().find(|t| t.runs.contains(&run))
443}
444
445/// Recover only a matching watch record's displayed local approval target.
446/// Never parse the question's prose or use the PR's current head as consent.
447pub(crate) fn approval_target(home: &Path, q: &Question) -> Option<crate::merge_decision::Target> {
448    if q.node != crate::bump::NOTICE_NODE
449        || q.seat != "release-watch"
450        || !q.choices.iter().any(|c| c == land::APPROVE)
451    {
452        return None;
453    }
454    let state = state_for_question(home, &q.id)?;
455    Some(crate::merge_decision::Target {
456        pr: state.url,
457        head: state.asked_head?,
458        language: "en".into(),
459    })
460}
461
462/// Explain the watcher-owned effects and immutable approval target to a deputy.
463pub(crate) fn deputy_brief(q: &Question, home: &Path) -> String {
464    let mut s = String::from(
465        "This question was filed by magi's release watcher about a release pull \
466         request. Whatever the owner picks, only the watcher acts on it, on its \
467         next lap; you apply nothing and never close, merge, rerun or push \
468         anything. Silence is a hold.",
469    );
470    match state_for_question(home, &q.id) {
471        Some(st) => {
472            s.push_str(&format!(
473                "\n\nPull request: {}\nCheckout: {}",
474                st.url, st.repo
475            ));
476        }
477        None => s.push_str(
478            "\n\nThe watcher's record for this question could not be found, so the \
479             pull request is known to you only through the question's own text. \
480             Say so to the owner rather than guessing.",
481        ),
482    }
483    if let Some(target) = &q.merge_target {
484        s.push_str(&format!("\n\nThis approval is bound to PR {} at head `{}`. A deputy merge requires the owner's whole latest message and quote to be exactly `merge {}`. A standing conversation can bind it as `{}: merge {}`. A generic batch instruction never authorizes merge.", target.pr, target.head, target.head, q.id, target.head));
485    }
486    s.push_str("\n\nWhat each option does:");
487    for c in &q.choices {
488        let what = match c.as_str() {
489            RERUN_AGAIN => {
490                "the daemon reruns the recorded failed workflows once if the PR head still matches"
491            }
492            HOLD if q.choices.iter().any(|c| c == land::APPROVE) => {
493                "leave the pull request open and unmerged; ask again only when the head moves"
494            }
495            HOLD => "stay quiet until a check or the head changes",
496            LEAVE_IT => {
497                "stop watching this pull request and dismiss its notice. It does NOT \
498                 close the pull request"
499            }
500            RETRY => "run the failed release again from where it stopped",
501            land::APPROVE => {
502                "squash-merge exactly the head the question names - irreversible - \
503                 then tag and release"
504            }
505            _ => "recorded as the answer",
506        };
507        s.push_str(&format!("\n- `{c}`: {what}"));
508    }
509    s.push_str(
510        "\n\nIf the owner's words ask for the pull request itself to be closed, that \
511         is something only they can do by hand; do not read it as a choice unless \
512         it plainly means stop watching.",
513    );
514    s
515}
516
517/// Starts the hold reason of a task held for a failed release, so only that
518/// hold is undone when the release later succeeds.
519const HOLD_PREFIX: &str = "[release] ";
520
521fn approval_message(pr: &str) -> String {
522    format!("Release PR {pr} is waiting for your approval to merge")
523}
524
525fn failed_message(pr: &str) -> String {
526    format!("Release PR {pr} merged but the release is on hold")
527}
528
529fn released_key(pr: &str) -> String {
530    format!("release-done:{pr}")
531}
532
533type Fut<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
534
535/// Every forge access, so tests inject a fake instead of `gh`.
536pub(crate) trait ReleaseForge: Send + Sync {
537    /// `(branch, url)` of every open release-bump pull request in `repo`.
538    fn list<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>>;
539    fn snapshot<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<RollupView>>;
540    fn rerun<'a>(&'a self, repo: &'a Path, run: &'a str) -> Fut<'a, Result<()>>;
541    /// The repository's config, to read `[release]` and `[merge] remote`.
542    fn config<'a>(&'a self, _repo: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
543        Box::pin(async { Ok(crate::config::Config::default()) })
544    }
545    /// `graph.land_approval` of the repository. An error means "unknown", and
546    /// the caller then asks.
547    fn land_approval<'a>(&'a self, _repo: &'a Path) -> Fut<'a, Result<bool>> {
548        Box::pin(async { Ok(true) })
549    }
550    /// Head branch and merge commit of a pull request (local mode).
551    fn info<'a>(&'a self, _repo: &'a Path, url: &'a str) -> Fut<'a, Result<PrInfo>> {
552        Box::pin(async move { bail!("no pull request info for {url}") })
553    }
554    /// Merge `url` (squash, delete branch) pinned to `head`. `Ok(true)` once
555    /// the forge says it merged, `Ok(false)` when it did not.
556    fn merge<'a>(&'a self, _repo: &'a Path, url: &'a str, _head: &'a str) -> Fut<'a, Result<bool>> {
557        Box::pin(async move { bail!("cannot merge {url}") })
558    }
559}
560
561/// What the forge says about a pull request's branch and merge.
562#[derive(Debug, Clone, Default, PartialEq, Eq)]
563pub(crate) struct PrInfo {
564    /// Head branch name, e.g. `chore/release-v1.2.3`.
565    pub branch: String,
566    /// The commit the pull request merged as, once merged.
567    pub merge_commit: Option<String>,
568    /// Pull request title.
569    pub title: String,
570}
571
572/// Parse `gh pr view --json headRefName,mergeCommit,title`. No I/O.
573pub(crate) fn parse_info(json: &str) -> Result<PrInfo> {
574    let v: serde_json::Value = serde_json::from_str(json)?;
575    let text = |k: &str| {
576        v.get(k)
577            .and_then(|x| x.as_str())
578            .unwrap_or_default()
579            .to_owned()
580    };
581    Ok(PrInfo {
582        branch: text("headRefName"),
583        title: text("title"),
584        merge_commit: v
585            .get("mergeCommit")
586            .and_then(|m| m.get("oid"))
587            .and_then(|o| o.as_str())
588            .filter(|o| !o.is_empty())
589            .map(str::to_owned),
590    })
591}
592
593/// The real forge: `gh`.
594pub(crate) struct GhForge;
595
596impl ReleaseForge for GhForge {
597    fn list<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
598        Box::pin(crate::bump::list_open_release_prs(repo))
599    }
600
601    fn snapshot<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<RollupView>> {
602        Box::pin(async move {
603            let args = [
604                "pr".to_owned(),
605                "view".to_owned(),
606                url.to_owned(),
607                "--json".to_owned(),
608                "url,number,state,headRefOid,statusCheckRollup".to_owned(),
609            ];
610            let (ok, out) = land::gh(repo, &args).await?;
611            if !ok {
612                bail!("gh pr view {url}: {out}");
613            }
614            land::parse_rollup(&out)
615        })
616    }
617
618    fn rerun<'a>(&'a self, repo: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
619        Box::pin(async move {
620            let args = [
621                "run".to_owned(),
622                "rerun".to_owned(),
623                run.to_owned(),
624                "--failed".to_owned(),
625            ];
626            let (ok, out) = land::gh(repo, &args).await?;
627            if !ok {
628                bail!("gh run rerun {run}: {out}");
629            }
630            Ok(())
631        })
632    }
633
634    fn config<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
635        Box::pin(async move { Ok(crate::config::Config::discover(repo, None)?.0) })
636    }
637
638    fn land_approval<'a>(&'a self, repo: &'a Path) -> Fut<'a, Result<bool>> {
639        // A second `discover` may read a newer generation of the config than
640        // the one `config` returned; only this one value rides on it, and an
641        // unreadable config makes the caller ask.
642        Box::pin(async move {
643            Ok(crate::config::Config::discover(repo, None)?
644                .0
645                .graph
646                .land_approval)
647        })
648    }
649
650    fn info<'a>(&'a self, repo: &'a Path, url: &'a str) -> Fut<'a, Result<PrInfo>> {
651        Box::pin(async move {
652            let args = [
653                "pr".to_owned(),
654                "view".to_owned(),
655                url.to_owned(),
656                "--json".to_owned(),
657                "headRefName,mergeCommit,title".to_owned(),
658            ];
659            let (ok, out) = land::gh(repo, &args).await?;
660            if !ok {
661                bail!("gh pr view {url}: {out}");
662            }
663            parse_info(&out)
664        })
665    }
666
667    fn merge<'a>(&'a self, repo: &'a Path, url: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
668        Box::pin(async move {
669            let title = self
670                .info(repo, url)
671                .await
672                .map(|i| i.title)
673                .unwrap_or_default();
674            let title = if title.trim().is_empty() {
675                "chore: release".to_owned()
676            } else {
677                title
678            };
679            // Same shape as the direct merge in `bump`, pinned to the head the
680            // owner approved: a push in between is refused by the forge.
681            let mut argv = crate::bump::bump_merge_argv(url, &title);
682            argv.push("--match-head-commit".to_owned());
683            argv.push(head.to_owned());
684            let (ok, out) = land::gh(repo, &argv).await?;
685            if ok {
686                return Ok(true);
687            }
688            // jj keeps HEAD detached, so `--delete-branch` exits non-zero after
689            // the merge happened: the forge decides.
690            let after = land::lifecycle(repo, url).await.ok();
691            if land::merged_after_all(&argv, &out, after).is_some() {
692                return Ok(true);
693            }
694            bail!("gh pr merge {url}: {out}")
695        })
696    }
697}
698
699/// The watcher: a forge, a magi home, and nothing else.
700pub(crate) struct Watcher {
701    forge: Box<dyn ReleaseForge>,
702    home: PathBuf,
703}
704
705impl Watcher {
706    pub(crate) fn new(forge: Box<dyn ReleaseForge>, home: PathBuf) -> Self {
707        Self { forge, home }
708    }
709
710    fn dir(&self) -> PathBuf {
711        self.home.join("release-watch")
712    }
713
714    fn state_path(&self, pr: &str) -> PathBuf {
715        self.dir().join(format!("{}.json", notices::id_of(pr)))
716    }
717
718    fn load(&self, pr: &str) -> WatchState {
719        std::fs::read_to_string(self.state_path(pr))
720            .ok()
721            .and_then(|s| serde_json::from_str(&s).ok())
722            .unwrap_or_default()
723    }
724
725    fn save(&self, pr: &str, st: &WatchState) -> bool {
726        let write = || -> Result<()> {
727            std::fs::create_dir_all(self.dir())?;
728            let path = self.state_path(pr);
729            let tmp = path.with_extension("json.tmp");
730            std::fs::write(&tmp, serde_json::to_string_pretty(st)?)?;
731            std::fs::rename(&tmp, &path)?;
732            Ok(())
733        };
734        match write() {
735            Ok(()) => true,
736            Err(e) => {
737                tracing::warn!("could not save the release watch for {pr}: {e:#}");
738                false
739            }
740        }
741    }
742
743    /// Stored records, for pull requests that left the open list.
744    fn stored(&self) -> Vec<WatchState> {
745        let Ok(rd) = std::fs::read_dir(self.dir()) else {
746            return Vec::new();
747        };
748        rd.flatten()
749            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
750            .filter_map(|e| std::fs::read_to_string(e.path()).ok())
751            .filter_map(|s| serde_json::from_str(&s).ok())
752            .collect()
753    }
754
755    fn questions(&self) -> Questions {
756        Questions::at(self.home.join("questions"))
757    }
758
759    fn notice_subjects(&self, pr: &str) -> Vec<String> {
760        let state = self.load(pr);
761        state.held_task.into_iter().chain(state.run).collect()
762    }
763
764    fn raise(&self, pr: &str, url: &str, cause: String, message: String) {
765        notices::raise_in(
766            &self.home,
767            Notice::warn(&notice_key(pr), message)
768                .cause(cause)
769                .link(Link::Url {
770                    url: url.to_owned(),
771                })
772                .about(self.notice_subjects(pr))
773                .for_run(self.load(pr).run),
774        );
775    }
776
777    /// One pass over every repo. `stall_secs` is `[daemon] release_stall_minutes`
778    /// in seconds. Best-effort throughout.
779    pub(crate) async fn lap(
780        &self,
781        repos: &[PathBuf],
782        stall_secs: i64,
783        now: i64,
784        halt: &(dyn Fn() -> bool + Sync),
785    ) {
786        let stored = self.stored();
787        // A repo with a watch record is covered even when nothing else names it.
788        let mut repos = repos.to_vec();
789        for s in &stored {
790            let p = PathBuf::from(&s.repo);
791            if !s.repo.is_empty() && !repos.contains(&p) {
792                repos.push(p);
793            }
794        }
795        for repo in &repos {
796            if halt() {
797                return;
798            }
799            let mut urls: Vec<String> = match self.forge.list(repo).await {
800                Ok(v) => v.into_iter().map(|(_, u)| u).collect(),
801                Err(e) => {
802                    tracing::warn!(
803                        "could not list release pull requests in {}: {e:#}",
804                        repo.display()
805                    );
806                    Vec::new()
807                }
808            };
809            // A pull request that left the open list is kept until its end is
810            // confirmed, never assumed.
811            let here = repo.to_string_lossy();
812            for s in stored.iter().filter(|s| s.repo == here.as_ref()) {
813                if !urls.contains(&s.url) {
814                    urls.push(s.url.clone());
815                }
816            }
817            for url in urls {
818                if halt() {
819                    return;
820                }
821                self.watch(repo, &url, stall_secs, now, halt).await;
822            }
823        }
824    }
825
826    /// Look at one pull request and act once.
827    pub(crate) async fn watch(
828        &self,
829        repo: &Path,
830        url: &str,
831        stall_secs: i64,
832        now: i64,
833        halt: &(dyn Fn() -> bool + Sync),
834    ) {
835        let Some(pr) = pr_key(url) else {
836            tracing::warn!("not a pull request url: {url}");
837            return;
838        };
839        let mut st = self.load(&pr);
840        st.repo = repo.to_string_lossy().into_owned();
841        st.url = url.to_owned();
842
843        let snap = match self.forge.snapshot(repo, url).await {
844            Ok(s) => Some(s),
845            Err(e) => {
846                tracing::warn!("could not read {url}: {e:#}");
847                None
848            }
849        };
850        match self.forge.config(repo).await {
851            Ok(cfg) if cfg.release.is_local() => {
852                st.local = true;
853                self.watch_local(repo, &pr, st, snap, Some(&cfg)).await;
854                return;
855            }
856            Ok(_) => st.local = false,
857            Err(e) if st.is_local() => {
858                tracing::warn!(
859                    "could not read the config of {}: {e:#}; watching {url} as a local release without it",
860                    repo.display()
861                );
862                self.watch_local(repo, &pr, st, snap, None).await;
863                return;
864            }
865            Err(e) => tracing::warn!(
866                "could not read the config of {}: {e:#}; watching {url} as an Actions release",
867                repo.display()
868            ),
869        }
870        if decide(snap.as_ref(), &st, now, stall_secs) == Step::Done {
871            self.finish(&pr, &st);
872            return;
873        }
874        let Some(snap) = snap else {
875            return;
876        };
877
878        // The owner's word comes first.
879        if let Some(id) = st.question.clone() {
880            match self.questions().get(&id) {
881                Ok(q) if q.status.open() => return,
882                Ok(q) => self.apply_answer(&pr, &mut st, &snap, &q, now),
883                Err(_) => st.question = None,
884            }
885            // Whatever was decided, the record is saved before anything else.
886            observe(&mut st, &snap, now);
887            self.save(&pr, &st);
888        } else {
889            observe(&mut st, &snap, now);
890        }
891
892        match decide(Some(&snap), &st, now, stall_secs) {
893            Step::Done | Step::Unknown | Step::Wait => {}
894            Step::Rerun(runs) => {
895                // Durable before the call: at most once per failing run.
896                for r in &runs {
897                    st.reruns.insert(r.clone(), now);
898                }
899                // Without a durable record the call must not happen: a restart
900                // would rerun the same run again.
901                if !self.save(&pr, &st) || halt() {
902                    return;
903                }
904                let mut ok = false;
905                for r in &runs {
906                    match self.forge.rerun(repo, r).await {
907                        Ok(()) => ok = true,
908                        Err(e) => tracing::warn!("could not rerun run {r} of {url}: {e:#}"),
909                    }
910                }
911                if ok {
912                    self.raise(
913                        &pr,
914                        url,
915                        format!("rerun:{}:{}", snap.head, st.fingerprint),
916                        rerun_message(&pr),
917                    );
918                }
919            }
920            Step::Escalate(why) => {
921                self.raise(
922                    &pr,
923                    url,
924                    format!("checks-held:{}:{}", snap.head, st.fingerprint),
925                    human_message(&pr),
926                );
927                let mut q = Question::new(
928                    String::new(),
929                    crate::bump::NOTICE_NODE.to_owned(),
930                    "release-watch".to_owned(),
931                    format!("Release PR {pr} is stuck: what now?"),
932                    question_detail(&snap, why, &st),
933                    maintenance_choices(&snap),
934                );
935                if q.choices.iter().any(|c| c == RERUN_AGAIN) {
936                    q.actions.insert(
937                        RERUN_AGAIN.into(),
938                        crate::ask::ChoiceAction::RerunChecks {
939                            repo: repo.to_path_buf(),
940                            pr: snap.url.clone(),
941                            head: snap.head.clone(),
942                            runs: failed_workflows(&snap).into_iter().take(20).collect(),
943                        },
944                    );
945                }
946                match self.questions().put_executable(&mut q) {
947                    Ok(()) => st.question = Some(q.id.clone()),
948                    Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
949                }
950            }
951        }
952        self.save(&pr, &st);
953    }
954
955    /// The local-mode path of [`Watcher::watch`]: approval, merge, release.
956    async fn watch_local(
957        &self,
958        repo: &Path,
959        pr: &str,
960        mut st: WatchState,
961        snap: Option<RollupView>,
962        cfg: Option<&crate::config::Config>,
963    ) {
964        let Some(snap) = snap else {
965            return;
966        };
967        match snap.state {
968            PrLifecycle::Closed => self.finish(pr, &st),
969            PrLifecycle::Merged => match cfg {
970                Some(cfg) => self.release_merged(repo, pr, st, cfg).await,
971                // The release commands live in the config: wait, keep the record.
972                None => {
973                    tracing::warn!("{pr} is merged but its config is unreadable; not released yet")
974                }
975            },
976            PrLifecycle::Open => {
977                let approved = match st.question.clone() {
978                    None => Approved::NoQuestion,
979                    Some(id) => match self.questions().get(&id) {
980                        Ok(q) if q.status.open() => Approved::Open,
981                        Ok(q) => match &q.answer {
982                            Some(Answer::Choice(c))
983                                if land::approval(Some(c)) == land::Approval::Merge =>
984                            {
985                                Approved::Merge
986                            }
987                            _ => Approved::Hold,
988                        },
989                        Err(_) => Approved::NoQuestion,
990                    },
991                };
992                let ask_first = if cfg.is_none() {
993                    true
994                } else {
995                    match self.forge.land_approval(repo).await {
996                        Ok(v) => v,
997                        Err(e) => {
998                            tracing::warn!("could not read graph.land_approval for {pr}: {e:#}");
999                            true
1000                        }
1001                    }
1002                };
1003                let step = decide_local(&snap.head, &st, approved, ask_first);
1004                if cfg.is_none() && matches!(step, LocalStep::Merge(_)) {
1005                    // Nothing to release with: keep the answer for a lap that can read the config.
1006                    tracing::warn!("not merging {pr} while its config is unreadable");
1007                    return;
1008                }
1009                if matches!(
1010                    step,
1011                    LocalStep::Merge(_) | LocalStep::Hold(_) | LocalStep::Ask
1012                ) {
1013                    // A settled (or stale) question is consumed exactly once.
1014                    st.question = None;
1015                }
1016                match step {
1017                    LocalStep::Wait => {}
1018                    LocalStep::Hold(head) => st.held_head = Some(head),
1019                    LocalStep::Ask => {
1020                        st.held_head = None;
1021                        st.asked_head = Some(snap.head.clone());
1022                        self.raise(
1023                            pr,
1024                            &snap.url,
1025                            format!("approval:{}", snap.head),
1026                            approval_message(pr),
1027                        );
1028                        let mut q = Question::new(
1029                            String::new(),
1030                            crate::bump::NOTICE_NODE.to_owned(),
1031                            "release-watch".to_owned(),
1032                            format!("Release PR {pr}: merge it and release?"),
1033                            format!(
1034                                "Pull request: {}\nHead: {}\n\n\
1035                                 `[release] mode = \"local\"`: no CI is awaited. `{}` merges \
1036                                 exactly this head, then magi tags the merge commit and runs the \
1037                                 configured release commands. `{}` leaves the pull request \
1038                                 open. Silence is a hold.\n",
1039                                snap.url,
1040                                snap.head,
1041                                land::APPROVE,
1042                                land::HOLD
1043                            ),
1044                            vec![land::APPROVE.to_owned(), land::HOLD.to_owned()],
1045                        );
1046                        q.merge_target = Some(crate::merge_decision::Target {
1047                            pr: snap.url.clone(),
1048                            head: snap.head.clone(),
1049                            language: "en".into(),
1050                        });
1051                        match self.questions().put_executable(&mut q) {
1052                            Ok(()) => st.question = Some(q.id.clone()),
1053                            Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1054                        }
1055                    }
1056                    LocalStep::Merge(head) => {
1057                        match self.forge.merge(repo, &snap.url, &head).await {
1058                            Ok(true) => {
1059                                // Release at once rather than a lap later.
1060                                self.save(pr, &st);
1061                                if let Some(cfg) = cfg {
1062                                    self.release_merged(repo, pr, st, cfg).await;
1063                                }
1064                                return;
1065                            }
1066                            res => {
1067                                match res {
1068                                    Err(e) => {
1069                                        tracing::warn!("could not merge {}: {e:#}", snap.url)
1070                                    }
1071                                    _ => tracing::warn!("{} did not merge", snap.url),
1072                                }
1073                                st.held_head = Some(head);
1074                                self.raise(
1075                                    pr,
1076                                    &snap.url,
1077                                    format!("merge-failed:{}", snap.head),
1078                                    format!(
1079                                        "Release PR {pr} could not be merged; merge it by hand"
1080                                    ),
1081                                );
1082                            }
1083                        }
1084                    }
1085                }
1086                self.save(pr, &st);
1087            }
1088        }
1089    }
1090
1091    /// The pull request is merged: create the job once, then drive it.
1092    async fn release_merged(
1093        &self,
1094        repo: &Path,
1095        pr: &str,
1096        mut st: WatchState,
1097        cfg: &crate::config::Config,
1098    ) {
1099        if st.ignored {
1100            return self.finish(pr, &st);
1101        }
1102        if st.job.is_none() {
1103            let info = match self.forge.info(repo, &st.url).await {
1104                Ok(i) => i,
1105                Err(e) => {
1106                    tracing::warn!("could not read the merged {}: {e:#}", st.url);
1107                    return;
1108                }
1109            };
1110            let (Some(version), Some(commit)) = (
1111                release_local::version_from_branch(&info.branch),
1112                info.merge_commit,
1113            ) else {
1114                // Not a release branch, or the forge has not recorded the merge
1115                // commit yet: nothing to release (yet).
1116                if release_local::version_from_branch(&info.branch).is_none() {
1117                    self.finish(pr, &st);
1118                }
1119                return;
1120            };
1121            let mut job = Job::new(&version, &st.url, &commit);
1122            // An escalated pull request keeps its first branch name but gets a
1123            // new title; the manifest at the merge commit picks between them.
1124            job.accepted = release_local::version_from_title(&info.title)
1125                .into_iter()
1126                .filter(|t| *t != version)
1127                .collect();
1128            st.job = Some(job);
1129            if !self.save(pr, &st) {
1130                return;
1131            }
1132        }
1133        let Some(mut job) = st.job.take() else {
1134            return;
1135        };
1136        if job.finished {
1137            // Saved as finished, but a stop may have come before the task was
1138            // given back: do that now, and keep the record until it is.
1139            st.job = Some(job);
1140            if !self.record_on_run(&st) {
1141                self.save(pr, &st);
1142                return;
1143            }
1144            return self.complete(pr, &mut st);
1145        }
1146        if job.interrupted() {
1147            job.failed = Some(format!(
1148                "magi stopped while `{}` was running; it may have partly run, so it is not repeated on its own",
1149                job.running.clone().unwrap_or_default()
1150            ));
1151        }
1152        if let Some(why) = job.failed.clone() {
1153            // Reconcile first: a stop between saving the failure and holding
1154            // the task (or an interrupted step) must not leave the task Done.
1155            self.hold_task(&mut st, &format!("release {} is on hold: {why}", job.tag()));
1156            // Held: only the owner's `retry` runs it again.
1157            let retry = self.settle_job_question(pr, &mut st, &mut job);
1158            if !retry {
1159                st.job = Some(job);
1160                self.save(pr, &st);
1161                return;
1162            }
1163        }
1164        let shell = cfg.shell();
1165        let env = release_local::Env {
1166            repo,
1167            home: &self.home,
1168            key: pr,
1169            remote: &cfg.merge.remote,
1170            shell: &shell,
1171            release: &cfg.release,
1172        };
1173        let result = {
1174            let mut save = |j: &Job| {
1175                let mut copy = st.clone();
1176                copy.job = Some(j.clone());
1177                self.save(pr, &copy)
1178            };
1179            release_local::run_job(&env, &mut job, &mut save).await
1180        };
1181        match result {
1182            Ok(()) => {
1183                self.raise_released(pr, &st.url, &job);
1184                st.job = Some(job);
1185                if self.record_on_run(&st) {
1186                    self.complete(pr, &mut st);
1187                } else {
1188                    // Finished and saved as such: the next lap retries only
1189                    // the run record, never the commands.
1190                    self.save(pr, &st);
1191                }
1192            }
1193            Err(e) => {
1194                job.failed = Some(format!("{e:#}"));
1195                self.hold_task(&mut st, &format!("release {} failed: {e:#}", job.tag()));
1196                self.raise(
1197                    pr,
1198                    &st.url,
1199                    format!(
1200                        "release-failed:{}:{}:{}",
1201                        job.commit,
1202                        job.done,
1203                        job.log.len()
1204                    ),
1205                    failed_message(pr),
1206                );
1207                self.file_job_question(pr, &mut st, &job);
1208                st.job = Some(job);
1209                self.save(pr, &st);
1210                self.record_on_run(&st);
1211            }
1212        }
1213    }
1214
1215    /// Copy the job onto the run that opened the release PR, so what the
1216    /// release did outlives the watcher's record. Returns whether the record
1217    /// may go: `false` only when a readable run could not be saved. No run, or
1218    /// a run that is gone, is warned about and counts as done.
1219    fn record_on_run(&self, st: &WatchState) -> bool {
1220        let (Some(run), Some(job)) = (st.run.as_deref(), st.job.as_ref()) else {
1221            return true;
1222        };
1223        let mut state = match RunState::load_under(run, &self.home) {
1224            Ok(s) => s,
1225            Err(e) => {
1226                tracing::warn!("could not record the release on run {run}: {e:#}");
1227                return true;
1228            }
1229        };
1230        let bump = state.release_bump.get_or_insert_with(Default::default);
1231        if bump.release.as_ref() == Some(job) {
1232            return true;
1233        }
1234        bump.release = Some(job.clone());
1235        state.event(
1236            "release",
1237            format!(
1238                "release {} {}",
1239                job.tag(),
1240                if job.finished {
1241                    "finished"
1242                } else if let Some(why) = &job.failed {
1243                    why
1244                } else {
1245                    "stopped"
1246                }
1247            ),
1248        );
1249        match state.save_under(&self.home) {
1250            Ok(()) => true,
1251            Err(e) => {
1252                tracing::warn!("could not save the release on run {run}: {e:#}");
1253                false
1254            }
1255        }
1256    }
1257
1258    /// Hold the task whose run opened this release PR: the daemon marked it
1259    /// Done when the change merged, but the release is not done. Best-effort;
1260    /// a task the owner already holds or that cannot be found is left alone.
1261    fn hold_task(&self, st: &mut WatchState, reason: &str) {
1262        if st.held_task.is_some() {
1263            return;
1264        }
1265        let Some(run) = st.run.clone() else {
1266            return;
1267        };
1268        let q = crate::queue::Queue::at(self.home.join("queue"));
1269        for mut t in q.list() {
1270            if t.runs.contains(&run) {
1271                if t.status != crate::queue::TaskStatus::Held {
1272                    t.notice_cause("release-failed", Some(&run), None);
1273                    t.hold_machine(Some(format!("{HOLD_PREFIX}{reason}")));
1274                    match q.put(&mut t) {
1275                        Ok(()) => st.held_task = Some(t.id.clone()),
1276                        Err(e) => tracing::warn!("could not hold task {} for {run}: {e:#}", t.id),
1277                    }
1278                }
1279                return;
1280            }
1281        }
1282    }
1283
1284    /// Give back the hold [`Watcher::hold_task`] put on a task, once the
1285    /// release succeeded. Only a machine hold carrying our marker is undone: a
1286    /// task the owner has since re-held or changed is left alone.
1287    ///
1288    /// Returns whether nothing is left to give back: `false` only when the
1289    /// write failed, in which case `held_task` is kept so a later lap retries.
1290    #[must_use]
1291    fn unhold_task(&self, st: &mut WatchState) -> bool {
1292        let Some(id) = st.held_task.take() else {
1293            return true;
1294        };
1295        let q = crate::queue::Queue::at(self.home.join("queue"));
1296        let mut t = match q.get(&id) {
1297            Ok(t) => t,
1298            // Only a task file confirmed absent is "nothing to restore";
1299            // any other failure may be transient, so keep the record.
1300            Err(_) if matches!(q.path_of(&id).try_exists(), Ok(false)) => return true,
1301            Err(e) => {
1302                tracing::warn!("could not read task {id} to restore it after the release: {e:#}");
1303                st.held_task = Some(id);
1304                return false;
1305            }
1306        };
1307        let ours = t.status == crate::queue::TaskStatus::Held
1308            && t.hold_source == Some(crate::queue::HoldSource::Machine)
1309            && t.hold_reason
1310                .as_deref()
1311                .is_some_and(|r| r.starts_with(HOLD_PREFIX));
1312        if ours {
1313            t.succeed();
1314            if let Err(e) = q.put(&mut t) {
1315                tracing::warn!("could not restore task {id} after the release: {e:#}");
1316                st.held_task = Some(id);
1317                return false;
1318            }
1319        }
1320        true
1321    }
1322
1323    /// The release is done (`st.job` is finished): give the task back, then
1324    /// drop the record. If the task cannot be restored the record stays, so
1325    /// the next lap tries again instead of leaving the task Held for good.
1326    fn complete(&self, pr: &str, st: &mut WatchState) {
1327        if self.unhold_task(st) {
1328            self.finish(pr, st);
1329        } else {
1330            self.save(pr, st);
1331        }
1332    }
1333
1334    fn raise_released(&self, pr: &str, url: &str, job: &Job) {
1335        notices::raise_in(
1336            &self.home,
1337            Notice::info(&released_key(pr), format!("Released {} ({pr})", job.tag()))
1338                .link(Link::Url {
1339                    url: url.to_owned(),
1340                })
1341                .about(self.notice_subjects(pr))
1342                .for_run(self.load(pr).run),
1343        );
1344    }
1345
1346    /// Ask what to do about a held job. One question per failure.
1347    fn file_job_question(&self, pr: &str, st: &mut WatchState, job: &Job) {
1348        let last = job
1349            .log
1350            .last()
1351            .map(|l| format!("\nLast step: {}\n\n{}\n", l.name, l.tail))
1352            .unwrap_or_default();
1353        let mut q = Question::new(
1354            String::new(),
1355            crate::bump::NOTICE_NODE.to_owned(),
1356            "release-watch".to_owned(),
1357            format!("Release {} of {pr} is on hold: retry?", job.tag()),
1358            format!(
1359                "Pull request: {}\nWhy it stopped: {}\n{last}\n\
1360                 Full output is under the magi home's release-local directory.\n\n\
1361                 - `{RETRY}`: run it again from where it stopped (the tag is not \
1362                 recreated and finished commands are skipped).\n\
1363                 - `{LEAVE_IT}`: stop watching; release by hand.\n\n\
1364                 Nothing is retried on its own; silence is a hold.\n",
1365                job.pr_url,
1366                job.failed.clone().unwrap_or_default()
1367            ),
1368            vec![RETRY.to_owned(), LEAVE_IT.to_owned()],
1369        );
1370        match self.questions().put_executable(&mut q) {
1371            Ok(()) => st.question = Some(q.id.clone()),
1372            Err(e) => tracing::warn!("could not file the question for {pr}: {e:#}"),
1373        }
1374    }
1375
1376    /// Read the owner's word on a held job. `true` means `retry`: the job was
1377    /// resumed. Files the question when there is none yet and the owner has not
1378    /// already held this exact failure.
1379    fn settle_job_question(&self, pr: &str, st: &mut WatchState, job: &mut Job) -> bool {
1380        let why = job.failed.clone().unwrap_or_default();
1381        if let Some(id) = st.question.clone() {
1382            match self.questions().get(&id) {
1383                Ok(q) if q.status.open() => return false,
1384                Ok(q) => {
1385                    st.question = None;
1386                    match &q.answer {
1387                        Some(Answer::Choice(c)) if c == RETRY => {
1388                            job.resume();
1389                            st.held = None;
1390                            return true;
1391                        }
1392                        Some(Answer::Choice(c)) if c == LEAVE_IT => {
1393                            st.ignored = true;
1394                            let _ = Notices::at(self.home.join("notifications"))
1395                                .dismiss(&notices::id_of(&notice_key(pr)));
1396                        }
1397                        // Abandoned or anything else: this failure stays held.
1398                        _ => st.held = Some(why),
1399                    }
1400                    return false;
1401                }
1402                Err(_) => st.question = None,
1403            }
1404        }
1405        if !st.ignored && st.held.as_deref() != Some(why.as_str()) {
1406            self.raise(
1407                pr,
1408                &st.url,
1409                format!(
1410                    "release-failed:{}:{}:{}",
1411                    job.commit,
1412                    job.done,
1413                    job.log.len()
1414                ),
1415                failed_message(pr),
1416            );
1417            self.file_job_question(pr, st, job);
1418        }
1419        false
1420    }
1421
1422    /// Apply a settled question's outcome to the record, once.
1423    fn apply_answer(
1424        &self,
1425        pr: &str,
1426        st: &mut WatchState,
1427        snap: &RollupView,
1428        q: &Question,
1429        now: i64,
1430    ) {
1431        st.question = None;
1432        if st.applied.contains(&q.id) {
1433            return;
1434        }
1435        st.applied.push(q.id.clone());
1436        if st.applied.len() > 20 {
1437            st.applied.remove(0);
1438        }
1439        match &q.answer {
1440            Some(Answer::Choice(c)) if c == RERUN_AGAIN => {
1441                st.held = None;
1442                if let Some(crate::ask::ChoiceAction::RerunChecks { runs, .. }) = q.chosen_action()
1443                {
1444                    // The daemon owns the external call. Suppress the watcher's
1445                    // automatic rerun, including across a watcher restart.
1446                    for run in runs {
1447                        st.reruns.insert(run.clone(), now);
1448                    }
1449                    return;
1450                }
1451                for c in snap.checks.iter().filter(|c| is_failed(c.verdict)) {
1452                    if let Some(r) = &c.run {
1453                        st.reruns.remove(r);
1454                    }
1455                }
1456            }
1457            Some(Answer::Choice(c)) if c == LEAVE_IT => {
1458                st.ignored = true;
1459                if let Err(e) = Notices::at(self.home.join("notifications"))
1460                    .dismiss(&notices::id_of(&notice_key(pr)))
1461                {
1462                    tracing::warn!("could not dismiss the notice for {pr}: {e:#}");
1463                }
1464            }
1465            // `hold`, no answer, abandoned: quiet until something changes.
1466            _ => st.held = Some(fingerprint(snap)),
1467        }
1468    }
1469
1470    /// The pull request is confirmed merged or closed: clear everything.
1471    fn finish(&self, pr: &str, st: &WatchState) {
1472        let _ =
1473            Notices::at(self.home.join("notifications")).dismiss(&notices::id_of(&notice_key(pr)));
1474        if let Some(id) = &st.question {
1475            let _ = self.questions().update(id, |q| {
1476                q.abandon("the release pull request is no longer open");
1477                Ok(())
1478            });
1479        }
1480        let _ = std::fs::remove_file(self.state_path(pr));
1481    }
1482}
1483
1484#[cfg(test)]
1485mod tests {
1486    use super::*;
1487    use crate::land::CheckView;
1488    use anyhow::Context;
1489    use std::sync::Mutex;
1490
1491    const URL: &str = "https://github.com/o/r/pull/7";
1492
1493    fn check(name: &str, v: Verdict, run: Option<&str>) -> CheckView {
1494        CheckView {
1495            name: name.to_owned(),
1496            verdict: v,
1497            run: run.map(str::to_owned),
1498            url: run.map(|r| format!("https://github.com/o/r/actions/runs/{r}/job/1")),
1499        }
1500    }
1501
1502    fn snap(state: PrLifecycle, head: &str, checks: Vec<CheckView>) -> RollupView {
1503        RollupView {
1504            url: URL.to_owned(),
1505            number: 7,
1506            state,
1507            head: head.to_owned(),
1508            checks,
1509        }
1510    }
1511
1512    fn open(checks: Vec<CheckView>) -> RollupView {
1513        snap(PrLifecycle::Open, "h1", checks)
1514    }
1515
1516    fn st_at(progress: i64) -> WatchState {
1517        WatchState {
1518            progress_at: progress,
1519            ..WatchState::default()
1520        }
1521    }
1522
1523    #[test]
1524    fn release_notice_uses_saved_question_and_checks_its_current_status() {
1525        let dir = tempfile::tempdir().unwrap();
1526        let watcher = Watcher::new(Box::new(GhForge), dir.path().to_path_buf());
1527        let questions = Questions::at(dir.path().join("questions"));
1528        let mut q = Question::new(
1529            String::new(),
1530            crate::bump::NOTICE_NODE.into(),
1531            "release-watch".into(),
1532            "Release?".into(),
1533            String::new(),
1534            Vec::new(),
1535        );
1536        questions.put(&mut q).unwrap();
1537        let st = WatchState {
1538            url: "https://example.invalid/o/r/pull/7".into(),
1539            question: Some(q.id.clone()),
1540            ..WatchState::default()
1541        };
1542        assert!(watcher.save("o/r#7", &st));
1543        let n = Notice::warn("release-pr:o/r#7", "Release PR is waiting for approval.");
1544        assert_eq!(
1545            notices::answerable_question(&n, &questions),
1546            Some(q.id.clone())
1547        );
1548        questions
1549            .update(&q.id, |q| {
1550                q.abandon("obsolete approval");
1551                Ok(())
1552            })
1553            .unwrap();
1554        assert_eq!(notices::answerable_question(&n, &questions), None);
1555        assert_eq!(question_for_notice(dir.path(), "release-pr:o/r#8"), None);
1556    }
1557
1558    #[test]
1559    fn unreadable_is_unknown_and_merged_or_closed_is_done() {
1560        assert_eq!(decide(None, &st_at(0), 10, 60), Step::Unknown);
1561        for s in [PrLifecycle::Merged, PrLifecycle::Closed] {
1562            assert_eq!(
1563                decide(Some(&snap(s, "h", vec![])), &st_at(0), 10, 60),
1564                Step::Done
1565            );
1566        }
1567    }
1568
1569    #[test]
1570    fn a_failed_run_is_rerun_even_while_another_run_is_pending() {
1571        let s = open(vec![
1572            check("win", Verdict::Fail, Some("11")),
1573            check("lint", Verdict::Pending, Some("12")),
1574        ]);
1575        assert_eq!(
1576            decide(Some(&s), &st_at(0), 1, 3600),
1577            Step::Rerun(vec!["11".into()])
1578        );
1579    }
1580
1581    #[test]
1582    fn a_run_with_a_job_still_pending_is_not_complete() {
1583        let s = open(vec![
1584            check("win", Verdict::Fail, Some("11")),
1585            check("mac", Verdict::Pending, Some("11")),
1586        ]);
1587        assert_eq!(decide(Some(&s), &st_at(0), 1, 3600), Step::Wait);
1588    }
1589
1590    #[test]
1591    fn one_rerun_per_run_then_grace_then_escalation() {
1592        let s = open(vec![check("win", Verdict::Fail, Some("11"))]);
1593        let mut st = st_at(0);
1594        st.reruns.insert("11".into(), 100);
1595        assert_eq!(decide(Some(&s), &st, 100 + RERUN_GRACE - 1, 0), Step::Wait);
1596        assert_eq!(
1597            decide(Some(&s), &st, 100 + RERUN_GRACE, 0),
1598            Step::Escalate(Why::StillRed)
1599        );
1600    }
1601
1602    #[test]
1603    fn a_failure_with_no_run_id_goes_straight_to_a_human_once_settled() {
1604        let s = open(vec![check("ci/ext", Verdict::Fail, None)]);
1605        assert_eq!(
1606            decide(Some(&s), &st_at(0), 1, 0),
1607            Step::Escalate(Why::StillRed)
1608        );
1609        let s = open(vec![
1610            check("ci/ext", Verdict::Fail, None),
1611            check("x", Verdict::Pending, Some("5")),
1612        ]);
1613        assert_eq!(decide(Some(&s), &st_at(0), 1, 0), Step::Wait);
1614    }
1615
1616    #[test]
1617    fn no_progress_for_the_bounded_time_escalates_and_zero_disables_it() {
1618        let s = open(vec![check("a", Verdict::Pass, Some("1"))]);
1619        assert_eq!(decide(Some(&s), &st_at(0), 3599, 3600), Step::Wait);
1620        assert_eq!(
1621            decide(Some(&s), &st_at(0), 3600, 3600),
1622            Step::Escalate(Why::Stalled)
1623        );
1624        assert_eq!(decide(Some(&s), &st_at(0), 99_999, 0), Step::Wait);
1625    }
1626
1627    #[test]
1628    fn held_and_ignored_stay_quiet_until_the_fingerprint_moves() {
1629        let s = open(vec![check("a", Verdict::Fail, None)]);
1630        let mut st = st_at(0);
1631        st.held = Some(fingerprint(&s));
1632        assert_eq!(decide(Some(&s), &st, 99_999, 60), Step::Wait);
1633        let moved = open(vec![
1634            check("a", Verdict::Pass, None),
1635            check("b", Verdict::Fail, None),
1636        ]);
1637        assert_eq!(
1638            decide(Some(&moved), &st, 99_999, 60),
1639            Step::Escalate(Why::StillRed)
1640        );
1641        st.ignored = true;
1642        assert_eq!(decide(Some(&moved), &st, 99_999, 60), Step::Wait);
1643    }
1644
1645    #[test]
1646    fn observe_restarts_the_clock_only_on_change_and_forgets_reruns_on_a_new_head() {
1647        let mut st = st_at(0);
1648        let a = open(vec![check("a", Verdict::Pending, Some("1"))]);
1649        observe(&mut st, &a, 10);
1650        assert_eq!(st.progress_at, 10);
1651        observe(&mut st, &a, 50);
1652        assert_eq!(st.progress_at, 10);
1653        st.reruns.insert("1".into(), 5);
1654        observe(
1655            &mut st,
1656            &snap(PrLifecycle::Open, "h2", a.checks.clone()),
1657            60,
1658        );
1659        assert!(st.reruns.is_empty());
1660        assert_eq!(st.progress_at, 60);
1661    }
1662
1663    #[test]
1664    fn pr_key_names_owner_repo_and_number() {
1665        assert_eq!(pr_key(URL).as_deref(), Some("o/r#7"));
1666        assert_eq!(pr_key("https://github.com/o/r/issues/7"), None);
1667        assert_eq!(pr_key("nonsense"), None);
1668    }
1669
1670    #[test]
1671    fn notice_wording_does_not_vary_between_polls() {
1672        assert_eq!(rerun_message("o/r#7"), rerun_message("o/r#7"));
1673        assert!(
1674            !human_message("o/r#7")
1675                .chars()
1676                .any(|c| c.is_ascii_digit() && c != '7')
1677        );
1678    }
1679
1680    #[test]
1681    fn the_release_question_is_not_claimed_by_other_machinery() {
1682        let q = Question::new(
1683            String::new(),
1684            crate::bump::NOTICE_NODE.into(),
1685            "release-watch".into(),
1686            "s".into(),
1687            String::new(),
1688            vec![HOLD.into()],
1689        );
1690        // The watcher's question is served by a deputy, but still covers nothing.
1691        assert_eq!(
1692            crate::deputy::kind_of(&q),
1693            Some(crate::deputy::Kind::Release)
1694        );
1695        let n = Notice::warn("release-pr:o/r#7", "m").about([String::new()]);
1696        assert!(!notices::covers(&q, &n));
1697        // A choice-less notice on the same node (seat `bump`) gets no deputy.
1698        let mut plain = q.clone();
1699        plain.seat = "bump".into();
1700        assert_eq!(crate::deputy::kind_of(&plain), None);
1701    }
1702
1703    #[derive(Default)]
1704    struct Fake {
1705        snap: Mutex<Option<RollupView>>,
1706        reruns: Mutex<Vec<String>>,
1707        local: Mutex<bool>,
1708        info: Mutex<Option<PrInfo>>,
1709        merges: Mutex<Vec<String>>,
1710        /// `graph.land_approval` as read; `None` is an unreadable config.
1711        approval: Mutex<Option<bool>>,
1712        merge_fails: Mutex<bool>,
1713        /// `Config::discover` itself fails.
1714        config_fails: Mutex<bool>,
1715    }
1716
1717    impl ReleaseForge for std::sync::Arc<Fake> {
1718        fn list<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<Vec<(String, String)>>> {
1719            Box::pin(async { Ok(vec![("chore/release-v1.0.0".to_owned(), URL.to_owned())]) })
1720        }
1721        fn snapshot<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<RollupView>> {
1722            let s = self.snap.lock().unwrap().clone();
1723            Box::pin(async move { s.context("unreadable") })
1724        }
1725        fn rerun<'a>(&'a self, _: &'a Path, run: &'a str) -> Fut<'a, Result<()>> {
1726            self.reruns.lock().unwrap().push(run.to_owned());
1727            Box::pin(async { Ok(()) })
1728        }
1729        fn config<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<crate::config::Config>> {
1730            if *self.config_fails.lock().unwrap() {
1731                return Box::pin(async { bail!("unreadable config") });
1732            }
1733            let mut c = crate::config::Config::default();
1734            if *self.local.lock().unwrap() {
1735                c.release.mode = crate::config::ReleaseMode::Local;
1736                c.release.commands = vec!["true".to_owned()];
1737            }
1738            Box::pin(async move { Ok(c) })
1739        }
1740        fn info<'a>(&'a self, _: &'a Path, _: &'a str) -> Fut<'a, Result<PrInfo>> {
1741            let i = self.info.lock().unwrap().clone();
1742            Box::pin(async move { i.context("no info") })
1743        }
1744        fn merge<'a>(&'a self, _: &'a Path, _: &'a str, head: &'a str) -> Fut<'a, Result<bool>> {
1745            self.merges.lock().unwrap().push(head.to_owned());
1746            let fails = *self.merge_fails.lock().unwrap();
1747            Box::pin(async move { if fails { bail!("refused") } else { Ok(true) } })
1748        }
1749        fn land_approval<'a>(&'a self, _: &'a Path) -> Fut<'a, Result<bool>> {
1750            let a = *self.approval.lock().unwrap();
1751            Box::pin(async move { a.context("unreadable config") })
1752        }
1753    }
1754
1755    fn rig() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher) {
1756        let dir = tempfile::tempdir().unwrap();
1757        let fake = std::sync::Arc::new(Fake::default());
1758        // The repository's default: `graph.land_approval` on.
1759        *fake.approval.lock().unwrap() = Some(true);
1760        let w = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1761        (dir, fake, w)
1762    }
1763
1764    fn red() -> RollupView {
1765        open(vec![check("win", Verdict::Fail, Some("11"))])
1766    }
1767
1768    #[tokio::test]
1769    async fn reruns_once_survives_a_restart_then_escalates_with_one_question() {
1770        let (dir, fake, w) = rig();
1771        *fake.snap.lock().unwrap() = Some(red());
1772        let repo = PathBuf::from("/nowhere");
1773        let no = || false;
1774        w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1775        assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
1776
1777        // A restart is a new Watcher over the same home: no second rerun.
1778        let w2 = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
1779        w2.lap(
1780            std::slice::from_ref(&repo),
1781            3600,
1782            1000 + RERUN_GRACE - 1,
1783            &no,
1784        )
1785        .await;
1786        assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1787        assert!(w2.questions().list().is_empty());
1788
1789        w2.lap(
1790            std::slice::from_ref(&repo),
1791            3600,
1792            1000 + RERUN_GRACE + 1,
1793            &no,
1794        )
1795        .await;
1796        w2.lap(
1797            std::slice::from_ref(&repo),
1798            3600,
1799            1000 + RERUN_GRACE + 400,
1800            &no,
1801        )
1802        .await;
1803        assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1804        let qs = w2.questions().list();
1805        assert_eq!(qs.len(), 1, "asked once, not every lap");
1806        assert_eq!(qs[0].node, crate::bump::NOTICE_NODE);
1807        assert!(qs[0].detail.contains("win") && qs[0].detail.contains(URL));
1808        assert_eq!(qs[0].choices, vec![RERUN_AGAIN, HOLD, LEAVE_IT]);
1809        let ns = Notices::at(dir.path().join("notifications")).list();
1810        assert_eq!(ns.len(), 1);
1811        assert_eq!(ns[0].key, "release-pr:o/r#7");
1812        assert_eq!(
1813            ns[0].link,
1814            Some(Link::Url {
1815                url: URL.to_owned(),
1816            })
1817        );
1818    }
1819
1820    async fn escalated() -> (tempfile::TempDir, std::sync::Arc<Fake>, Watcher, String) {
1821        let (dir, fake, w) = rig();
1822        *fake.snap.lock().unwrap() = Some(red());
1823        let repo = PathBuf::from("/nowhere");
1824        let no = || false;
1825        w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
1826        w.lap(&[repo], 3600, 2000, &no).await;
1827        let id = w.questions().list()[0].id.clone();
1828        (dir, fake, w, id)
1829    }
1830
1831    fn answer(w: &Watcher, id: &str, choice: &str) {
1832        w.questions()
1833            .update(id, |q| q.answer(Answer::Choice(choice.to_owned())))
1834            .unwrap();
1835    }
1836
1837    #[tokio::test]
1838    async fn legacy_rerun_again_reruns_exactly_once_more() {
1839        let (_d, fake, w, id) = escalated().await;
1840        w.questions()
1841            .update(&id, |q| {
1842                q.actions.clear();
1843                Ok(())
1844            })
1845            .unwrap();
1846        answer(&w, &id, RERUN_AGAIN);
1847        let repo = PathBuf::from("/nowhere");
1848        let no = || false;
1849        w.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1850        w.lap(&[repo], 3600, 3001, &no).await;
1851        assert_eq!(fake.reruns.lock().unwrap().len(), 2);
1852    }
1853
1854    #[tokio::test]
1855    async fn typed_rerun_is_left_to_the_daemon_instead_of_the_watcher() {
1856        let (_d, fake, w, id) = escalated().await;
1857        let q = w.questions().get(&id).unwrap();
1858        assert!(
1859            matches!(q.actions.get(RERUN_AGAIN), Some(crate::ask::ChoiceAction::RerunChecks { head, runs, .. })
1860            if head == "h1" && runs == &["11"])
1861        );
1862        answer(&w, &id, RERUN_AGAIN);
1863        w.lap(&[PathBuf::from("/nowhere")], 3600, 3000, &|| false)
1864            .await;
1865        assert_eq!(
1866            fake.reruns.lock().unwrap().len(),
1867            1,
1868            "only the earlier automatic rerun"
1869        );
1870    }
1871
1872    #[test]
1873    fn no_failed_workflow_means_no_rerun_choice() {
1874        let snap = open(vec![check("external", Verdict::Fail, None)]);
1875        assert_eq!(maintenance_choices(&snap), vec![HOLD, LEAVE_IT]);
1876    }
1877
1878    #[tokio::test]
1879    async fn hold_and_silence_do_not_reask_and_leave_it_dismisses() {
1880        let (d, fake, w, id) = escalated().await;
1881        answer(&w, &id, HOLD);
1882        let repo = PathBuf::from("/nowhere");
1883        let no = || false;
1884        for t in [3000, 90_000, 200_000] {
1885            w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1886        }
1887        assert_eq!(w.questions().list().len(), 1, "held: no second question");
1888        assert_eq!(fake.reruns.lock().unwrap().len(), 1);
1889
1890        let (d2, _f2, w2, id2) = escalated().await;
1891        answer(&w2, &id2, LEAVE_IT);
1892        w2.lap(std::slice::from_ref(&repo), 3600, 3000, &no).await;
1893        let ns = Notices::at(d2.path().join("notifications")).list();
1894        assert!(ns.is_empty(), "dismissed notices are hidden");
1895        drop(d);
1896    }
1897
1898    /// Give the question a deputy seat the way `Deputies::attach` + a turn do.
1899    fn with_deputy(w: &Watcher, id: &str, home: &Path) -> String {
1900        let seat = crate::agent::SeatState::new(&crate::ask::deputy_seat_key(id), "a", 1);
1901        let key = seat.key.clone();
1902        let brief = deputy_brief(&w.questions().get(id).unwrap(), home);
1903        w.questions()
1904            .update(id, |q| {
1905                let mut d = crate::ask::Deputy::new(brief);
1906                d.seat = Some(seat);
1907                q.deputy = Some(d);
1908                Ok(())
1909            })
1910            .unwrap();
1911        key
1912    }
1913
1914    #[tokio::test]
1915    async fn the_brief_names_the_pull_request_and_what_leave_it_really_does() {
1916        let (d, _f, w, id) = escalated().await;
1917        let q = w.questions().get(&id).unwrap();
1918        let b = deputy_brief(&q, d.path());
1919        assert!(b.contains(URL), "{b}");
1920        assert!(
1921            b.contains("`leave it`") && b.contains("does NOT close"),
1922            "{b}"
1923        );
1924        assert!(b.contains("`rerun again`") && b.contains("`hold`"), "{b}");
1925        assert!(!b.contains("`merge`"), "no merge on an escalation: {b}");
1926
1927        // No record naming the question: said, not invented.
1928        let empty = tempfile::tempdir().unwrap();
1929        let b = deputy_brief(&q, empty.path());
1930        assert!(b.contains("could not be found") && !b.contains(URL), "{b}");
1931    }
1932
1933    #[tokio::test]
1934    async fn a_settled_leave_it_is_applied_once_by_the_watcher_and_closes_nothing() {
1935        let (d, fake, w, id) = escalated().await;
1936        let seat = with_deputy(&w, &id, d.path());
1937        let say = "クローズしていいよ。private repo だから、何回やっても失敗しちゃうから";
1938        w.questions().update(&id, |q| q.say(say)).unwrap();
1939        w.questions()
1940            .update(&id, |q| {
1941                q.settle_by_deputy(&seat, LEAVE_IT, "クローズしていいよ", None)
1942            })
1943            .unwrap();
1944        let q = w.questions().get(&id).unwrap();
1945        assert_eq!(q.answer, Some(Answer::Choice(LEAVE_IT.to_owned())));
1946
1947        let repo = PathBuf::from("/nowhere");
1948        let no = || false;
1949        for t in [3000, 3100] {
1950            w.lap(std::slice::from_ref(&repo), 3600, t, &no).await;
1951        }
1952        let st = w.stored().pop().unwrap();
1953        assert!(st.ignored, "watching stopped");
1954        assert_eq!(st.applied, vec![id.clone()], "applied exactly once");
1955        assert!(
1956            Notices::at(d.path().join("notifications"))
1957                .list()
1958                .is_empty()
1959        );
1960        assert_eq!(fake.reruns.lock().unwrap().len(), 1, "no extra rerun");
1961        assert_eq!(w.questions().list().len(), 1, "no second question");
1962    }
1963
1964    #[tokio::test]
1965    async fn the_task_of_a_release_question_comes_from_the_watch_record() {
1966        let (d, _f, w, id) = escalated().await;
1967        let q = w.questions().get(&id).unwrap();
1968        let queue = crate::queue::Queue::at(d.path().join("queue"));
1969        assert!(
1970            task_of_question(d.path(), &q, &queue).is_none(),
1971            "no run known"
1972        );
1973        assert!(task_of_question(d.path(), &q, &queue).is_none());
1974        let mut st = w.stored().pop().unwrap();
1975        st.run = Some("run-1".into());
1976        w.save("o/r#7", &st);
1977        // A known run with no task is still nothing to refuse on.
1978        assert!(task_of_question(d.path(), &q, &queue).is_none());
1979        let mut t = crate::queue::Task::new(
1980            "t".into(),
1981            "i".into(),
1982            PathBuf::from("/nowhere"),
1983            crate::queue::Source::Human,
1984        );
1985        t.runs.push("run-1".into());
1986        queue.put(&mut t).unwrap();
1987        assert_eq!(task_of_question(d.path(), &q, &queue).unwrap().id, t.id);
1988    }
1989
1990    #[tokio::test]
1991    async fn merged_clears_the_notice_the_question_and_the_record() {
1992        let (d, fake, w, _id) = escalated().await;
1993        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
1994        let repo = PathBuf::from("/nowhere");
1995        w.lap(&[repo], 3600, 5000, &(|| false)).await;
1996        assert!(
1997            Notices::at(d.path().join("notifications"))
1998                .list()
1999                .is_empty()
2000        );
2001        assert!(w.questions().list().iter().all(|q| !q.status.open()));
2002        assert!(w.stored().is_empty());
2003    }
2004
2005    #[tokio::test]
2006    async fn an_unreadable_forge_changes_nothing() {
2007        let (d, fake, w) = rig();
2008        *fake.snap.lock().unwrap() = None;
2009        w.lap(&[PathBuf::from("/nowhere")], 60, 99_999, &(|| false))
2010            .await;
2011        assert!(fake.reruns.lock().unwrap().is_empty());
2012        assert!(w.questions().list().is_empty());
2013        assert!(!d.path().join("release-watch").exists());
2014    }
2015
2016    #[tokio::test]
2017    async fn no_rerun_when_the_record_cannot_be_saved() {
2018        let (d, fake, w) = rig();
2019        *fake.snap.lock().unwrap() = Some(red());
2020        // A file where the directory should be makes every save fail.
2021        std::fs::write(d.path().join("release-watch"), "x").unwrap();
2022        w.lap(&[PathBuf::from("/nowhere")], 3600, 1000, &(|| false))
2023            .await;
2024        assert!(fake.reruns.lock().unwrap().is_empty());
2025    }
2026
2027    #[tokio::test]
2028    async fn a_park_stops_the_lap_before_any_call() {
2029        let (_d, fake, w) = rig();
2030        *fake.snap.lock().unwrap() = Some(red());
2031        w.lap(&[PathBuf::from("/nowhere")], 60, 1000, &(|| true))
2032            .await;
2033        assert!(fake.reruns.lock().unwrap().is_empty());
2034    }
2035
2036    // ---- [release] mode = "local" ----
2037
2038    fn local_st(asked: Option<&str>, held: Option<&str>) -> WatchState {
2039        WatchState {
2040            asked_head: asked.map(str::to_owned),
2041            held_head: held.map(str::to_owned),
2042            ..WatchState::default()
2043        }
2044    }
2045
2046    #[test]
2047    fn local_decisions_never_wait_on_ci_and_bind_approval_to_the_head() {
2048        use Approved::*;
2049        let none = local_st(None, None);
2050        assert_eq!(decide_local("h1", &none, NoQuestion, true), LocalStep::Ask);
2051        assert_eq!(decide_local("h1", &none, Open, true), LocalStep::Wait);
2052        let asked = local_st(Some("h1"), None);
2053        assert_eq!(
2054            decide_local("h1", &asked, Merge, true),
2055            LocalStep::Merge("h1".to_owned())
2056        );
2057        assert_eq!(
2058            decide_local("h1", &asked, Hold, true),
2059            LocalStep::Hold("h1".to_owned())
2060        );
2061        // The head moved after the owner answered: ask again, never merge.
2062        assert_eq!(decide_local("h2", &asked, Merge, true), LocalStep::Ask);
2063        assert_eq!(decide_local("h2", &asked, Hold, true), LocalStep::Ask);
2064        // A held head stays quiet until it moves; ignored stays quiet.
2065        let held = local_st(Some("h1"), Some("h1"));
2066        assert_eq!(decide_local("h1", &held, NoQuestion, true), LocalStep::Wait);
2067        assert_eq!(decide_local("h2", &held, NoQuestion, true), LocalStep::Ask);
2068        let mut ign = local_st(None, None);
2069        ign.ignored = true;
2070        assert_eq!(decide_local("h1", &ign, NoQuestion, true), LocalStep::Wait);
2071    }
2072
2073    #[test]
2074    fn without_land_approval_an_unasked_head_merges_but_stops_stay_stopped() {
2075        use Approved::*;
2076        let none = local_st(None, None);
2077        assert_eq!(
2078            decide_local("h1", &none, NoQuestion, false),
2079            LocalStep::Merge("h1".to_owned())
2080        );
2081        // An open question is respected, a held head and ignored stay quiet.
2082        assert_eq!(decide_local("h1", &none, Open, false), LocalStep::Wait);
2083        let held = local_st(Some("h1"), Some("h1"));
2084        assert_eq!(
2085            decide_local("h1", &held, NoQuestion, false),
2086            LocalStep::Wait
2087        );
2088        assert_eq!(
2089            decide_local("h2", &held, NoQuestion, false),
2090            LocalStep::Merge("h2".to_owned())
2091        );
2092        let mut ign = local_st(None, None);
2093        ign.ignored = true;
2094        assert_eq!(decide_local("h1", &ign, NoQuestion, false), LocalStep::Wait);
2095        // The owner's recorded no for this head still holds.
2096        let asked = local_st(Some("h1"), None);
2097        assert_eq!(
2098            decide_local("h1", &asked, Hold, false),
2099            LocalStep::Hold("h1".to_owned())
2100        );
2101    }
2102
2103    #[tokio::test]
2104    async fn a_registered_local_pr_still_asks_but_never_merges_while_the_config_is_unreadable() {
2105        let (dir, fake, w) = rig();
2106        *fake.config_fails.lock().unwrap() = true;
2107        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2108        let repo = PathBuf::from("/nowhere");
2109        let no = || false;
2110        let pr = pr_key(URL).unwrap();
2111        register(dir.path(), &repo, URL, "run1");
2112        // `register` goes through the real forge; make sure the record exists.
2113        assert!(w.load(&pr).local);
2114        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2115        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2116        assert!(fake.merges.lock().unwrap().is_empty());
2117        let qs = w.questions().list();
2118        assert_eq!(qs.len(), 1);
2119        assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2120        // The owner says merge: nothing merges and the answer is not consumed.
2121        answer(&w, &qs[0].id, land::APPROVE);
2122        w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2123        assert!(fake.merges.lock().unwrap().is_empty());
2124        assert_eq!(w.load(&pr).question.as_deref(), Some(qs[0].id.as_str()));
2125        // The config is readable again: the kept answer merges.
2126        *fake.config_fails.lock().unwrap() = false;
2127        *fake.local.lock().unwrap() = true;
2128        w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2129        assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
2130    }
2131
2132    #[tokio::test]
2133    async fn a_local_record_written_before_the_flag_existed_is_still_local() {
2134        let (_d, fake, w) = rig();
2135        *fake.config_fails.lock().unwrap() = true;
2136        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2137        let repo = PathBuf::from("/nowhere");
2138        let no = || false;
2139        let pr = pr_key(URL).unwrap();
2140        let old = WatchState {
2141            repo: repo.to_string_lossy().into_owned(),
2142            url: URL.to_owned(),
2143            run: Some("run1".to_owned()),
2144            ..WatchState::default()
2145        };
2146        assert!(!old.local);
2147        w.save(&pr, &old);
2148        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2149        assert!(fake.merges.lock().unwrap().is_empty());
2150        let qs = w.questions().list();
2151        assert_eq!(qs.len(), 1);
2152        assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2153    }
2154
2155    #[tokio::test]
2156    async fn an_unregistered_pr_is_an_actions_release_when_the_config_is_unreadable() {
2157        let (_d, fake, w) = rig();
2158        *fake.config_fails.lock().unwrap() = true;
2159        *fake.snap.lock().unwrap() = Some(red());
2160        let repo = PathBuf::from("/nowhere");
2161        let no = || false;
2162        w.lap(std::slice::from_ref(&repo), 3600, 1000, &no).await;
2163        assert_eq!(*fake.reruns.lock().unwrap(), vec!["11".to_owned()]);
2164    }
2165
2166    #[tokio::test]
2167    async fn a_merged_local_pr_waits_with_its_record_while_the_config_is_unreadable() {
2168        let (dir, fake, w) = rig();
2169        *fake.config_fails.lock().unwrap() = true;
2170        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2171        let repo = PathBuf::from("/nowhere");
2172        let no = || false;
2173        let pr = pr_key(URL).unwrap();
2174        register(dir.path(), &repo, URL, "run1");
2175        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2176        assert!(w.state_path(&pr).exists(), "the record is kept");
2177    }
2178
2179    #[tokio::test]
2180    async fn land_approval_off_merges_the_observed_head_without_a_question() {
2181        let (_d, fake, w) = rig();
2182        *fake.local.lock().unwrap() = true;
2183        *fake.approval.lock().unwrap() = Some(false);
2184        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2185        let repo = PathBuf::from("/nowhere");
2186        let no = || false;
2187        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2188        assert_eq!(*fake.merges.lock().unwrap(), vec!["h1".to_owned()]);
2189        assert!(w.questions().list().is_empty(), "no approval question");
2190    }
2191
2192    #[tokio::test]
2193    async fn land_approval_on_or_unreadable_still_asks_and_never_merges() {
2194        for approval in [Some(true), None] {
2195            let (_d, fake, w) = rig();
2196            *fake.local.lock().unwrap() = true;
2197            *fake.approval.lock().unwrap() = approval;
2198            *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2199            let repo = PathBuf::from("/nowhere");
2200            let no = || false;
2201            w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2202            w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2203            assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2204            let qs = w.questions().list();
2205            assert_eq!(qs.len(), 1);
2206            assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2207        }
2208    }
2209
2210    #[tokio::test]
2211    async fn land_approval_off_does_not_merge_on_an_unreadable_forge_or_retry_a_refusal() {
2212        let (_d, fake, w) = rig();
2213        *fake.local.lock().unwrap() = true;
2214        *fake.approval.lock().unwrap() = Some(false);
2215        let repo = PathBuf::from("/nowhere");
2216        let no = || false;
2217        // Unreadable forge answer: nothing moves.
2218        *fake.snap.lock().unwrap() = None;
2219        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2220        assert!(fake.merges.lock().unwrap().is_empty());
2221        // A refused merge holds that head: one attempt, no retry loop.
2222        *fake.merge_fails.lock().unwrap() = true;
2223        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h1", vec![]));
2224        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2225        w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2226        assert_eq!(fake.merges.lock().unwrap().len(), 1);
2227        assert!(w.questions().list().is_empty());
2228        // A moved head is judged afresh.
2229        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2230        w.lap(std::slice::from_ref(&repo), 60, 4, &no).await;
2231        assert_eq!(
2232            *fake.merges.lock().unwrap(),
2233            vec!["h1".to_owned(), "h2".to_owned()]
2234        );
2235    }
2236
2237    #[test]
2238    fn pr_info_is_read_from_gh_json() {
2239        let i = parse_info(
2240            r#"{"headRefName":"chore/release-v1.2.3","title":"chore: release v1.2.3","mergeCommit":{"oid":"abc"}}"#,
2241        )
2242        .unwrap();
2243        assert_eq!(i.branch, "chore/release-v1.2.3");
2244        assert_eq!(i.merge_commit.as_deref(), Some("abc"));
2245        let open = parse_info(r#"{"headRefName":"b","title":"t","mergeCommit":null}"#).unwrap();
2246        assert_eq!(open.merge_commit, None);
2247    }
2248
2249    #[tokio::test]
2250    async fn a_local_release_asks_once_without_ci_reruns_or_stall_escalation() {
2251        let (_d, fake, w) = rig();
2252        *fake.local.lock().unwrap() = true;
2253        // Red checks that would rerun in Actions mode, and a stall clock far
2254        // past the limit: neither matters here.
2255        *fake.snap.lock().unwrap() = Some(red());
2256        let repo = PathBuf::from("/nowhere");
2257        let no = || false;
2258        w.lap(std::slice::from_ref(&repo), 60, 1_000_000, &no).await;
2259        w.lap(std::slice::from_ref(&repo), 60, 2_000_000, &no).await;
2260        assert!(fake.reruns.lock().unwrap().is_empty());
2261        let qs = w.questions().list();
2262        assert_eq!(qs.len(), 1, "one approval question, not one per lap");
2263        assert_eq!(qs[0].choices, vec![land::APPROVE, land::HOLD]);
2264        assert!(fake.merges.lock().unwrap().is_empty(), "silence is a hold");
2265    }
2266
2267    #[tokio::test]
2268    async fn a_merge_failure_pages_after_the_approval_notice_was_delivered() {
2269        let (dir, fake, w) = rig();
2270        *fake.local.lock().unwrap() = true;
2271        *fake.snap.lock().unwrap() = Some(open(vec![]));
2272        let repo = PathBuf::from("/nowhere");
2273        w.lap(std::slice::from_ref(&repo), 60, 1, &|| false).await;
2274        let store = Notices::at(dir.path().join("notifications"));
2275        let approval = store.list().pop().unwrap();
2276        assert_eq!(approval.cause.as_deref(), Some("approval:h1"));
2277        // A chat receipt leaves the outbox read; the same key must relight
2278        // when the approved merge itself fails, despite identical severity.
2279        store.mark_read(&approval.id).unwrap();
2280        let question = w.questions().list().pop().unwrap();
2281        answer(&w, &question.id, land::APPROVE);
2282        *fake.merge_fails.lock().unwrap() = true;
2283        w.lap(std::slice::from_ref(&repo), 60, 2, &|| false).await;
2284        let failure = store.get(&approval.id).unwrap();
2285        assert!(failure.unread());
2286        assert_eq!(failure.generation, approval.generation + 1);
2287        assert_eq!(failure.cause.as_deref(), Some("merge-failed:h1"));
2288        assert!(failure.message.contains("could not be merged"));
2289        store.mark_read(&failure.id).unwrap();
2290        let restarted = Watcher::new(Box::new(fake.clone()), dir.path().to_path_buf());
2291        restarted
2292            .lap(std::slice::from_ref(&repo), 60, 3, &|| false)
2293            .await;
2294        assert_eq!(fake.merges.lock().unwrap().len(), 1);
2295        assert!(!store.get(&approval.id).unwrap().unread());
2296        assert_eq!(
2297            store.get(&approval.id).unwrap().generation,
2298            failure.generation
2299        );
2300    }
2301
2302    #[tokio::test]
2303    async fn merge_is_pinned_to_the_approved_head_and_a_moved_head_asks_again() {
2304        let (_d, fake, w) = rig();
2305        *fake.local.lock().unwrap() = true;
2306        *fake.snap.lock().unwrap() = Some(open(vec![]));
2307        let repo = PathBuf::from("/nowhere");
2308        let no = || false;
2309        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2310        let original = &w.questions().list()[0];
2311        assert_eq!(original.merge_target.as_ref().unwrap().head, "h1");
2312        let id = original.id.clone();
2313        w.questions()
2314            .update(&id, |q| {
2315                q.merge_target = None;
2316                Ok(())
2317            })
2318            .unwrap();
2319        assert_eq!(
2320            w.questions()
2321                .get(&id)
2322                .unwrap()
2323                .merge_target
2324                .as_ref()
2325                .unwrap()
2326                .head,
2327            "h1",
2328            "legacy binding comes from the matching persisted watch record"
2329        );
2330        // The head moves before the owner answers.
2331        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Open, "h2", vec![]));
2332        answer(&w, &id, land::APPROVE);
2333        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2334        assert!(fake.merges.lock().unwrap().is_empty());
2335        assert_eq!(w.questions().list().len(), 2, "asked again about h2");
2336        // Answering the new question merges exactly h2.
2337        let id2 = w
2338            .questions()
2339            .list()
2340            .into_iter()
2341            .find(|q| q.status.open())
2342            .unwrap()
2343            .id;
2344        answer(&w, &id2, land::APPROVE);
2345        w.lap(std::slice::from_ref(&repo), 60, 3, &no).await;
2346        assert_eq!(*fake.merges.lock().unwrap(), vec!["h2".to_owned()]);
2347    }
2348
2349    #[tokio::test]
2350    async fn a_failed_release_holds_with_one_notice_and_one_question_and_never_retries() {
2351        let (d, fake, w) = rig();
2352        *fake.local.lock().unwrap() = true;
2353        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2354        *fake.info.lock().unwrap() = Some(PrInfo {
2355            branch: "chore/release-v1.0.0".to_owned(),
2356            merge_commit: Some("deadbeef".to_owned()),
2357            title: "chore: release v1.0.0".to_owned(),
2358        });
2359        // The checkout is not a repository, so the release cannot start.
2360        let repo = d.path().join("nowhere");
2361        let no = || false;
2362        w.lap(std::slice::from_ref(&repo), 60, 1, &no).await;
2363        let st = w.load("o/r#7");
2364        let job = st.job.expect("the job is recorded");
2365        assert!(job.failed.is_some() && !job.tag_done && !job.finished);
2366        let qs = w.questions().list();
2367        assert_eq!(qs.len(), 1);
2368        assert_eq!(qs[0].choices, vec![RETRY, LEAVE_IT]);
2369        // Later laps change nothing: no retry, no second question.
2370        w.lap(std::slice::from_ref(&repo), 60, 2, &no).await;
2371        assert_eq!(w.questions().list().len(), 1);
2372        let ns = Notices::at(d.path().join("notifications")).list();
2373        assert_eq!(ns.len(), 1);
2374        assert_eq!(
2375            ns[0].link,
2376            Some(Link::Url {
2377                url: URL.to_owned(),
2378            })
2379        );
2380    }
2381
2382    #[tokio::test]
2383    async fn an_escalated_pull_request_records_the_title_version_as_acceptable() {
2384        let (d, fake, w) = rig();
2385        *fake.local.lock().unwrap() = true;
2386        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2387        *fake.info.lock().unwrap() = Some(PrInfo {
2388            branch: "chore/release-v0.1.4".to_owned(),
2389            merge_commit: Some("deadbeef".to_owned()),
2390            title: "chore: release v0.2.0 (minor bump)".to_owned(),
2391        });
2392        let repo = d.path().join("nowhere");
2393        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2394        let job = w.load("o/r#7").job.expect("the job is recorded");
2395        assert_eq!(job.version, "0.1.4");
2396        assert_eq!(job.accepted, vec!["0.2.0".to_owned()]);
2397    }
2398
2399    #[tokio::test]
2400    async fn a_failed_release_holds_the_task_whose_run_opened_the_pr() {
2401        let (d, fake, w) = rig();
2402        *fake.local.lock().unwrap() = true;
2403        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2404        *fake.info.lock().unwrap() = Some(PrInfo {
2405            branch: "chore/release-v1.0.0".to_owned(),
2406            merge_commit: Some("deadbeef".to_owned()),
2407            title: "t".to_owned(),
2408        });
2409        let repo = d.path().join("nowhere");
2410        let q = crate::queue::Queue::at(d.path().join("queue"));
2411        let mut t = crate::queue::Task::new(
2412            "t".to_owned(),
2413            "i".to_owned(),
2414            repo.clone(),
2415            crate::queue::Source::Human,
2416        );
2417        t.runs.push("run1".to_owned());
2418        t.status = crate::queue::TaskStatus::Done;
2419        q.put(&mut t).unwrap();
2420        register(d.path(), &repo, URL, "run1");
2421        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2422        let held = q.get(&t.id).unwrap();
2423        assert_eq!(held.status, crate::queue::TaskStatus::Held);
2424        assert!(
2425            held.hold_reason
2426                .unwrap_or_default()
2427                .contains("release v1.0.0")
2428        );
2429    }
2430
2431    #[tokio::test]
2432    async fn unhold_gives_the_task_back_only_when_the_hold_is_ours() {
2433        let (d, _fake, w) = rig();
2434        let q = crate::queue::Queue::at(d.path().join("queue"));
2435        let mut t = crate::queue::Task::new(
2436            "t".to_owned(),
2437            "i".to_owned(),
2438            PathBuf::from("/r"),
2439            crate::queue::Source::Human,
2440        );
2441        t.runs.push("run1".to_owned());
2442        t.status = crate::queue::TaskStatus::Done;
2443        q.put(&mut t).unwrap();
2444        let mut st = WatchState {
2445            run: Some("run1".to_owned()),
2446            ..WatchState::default()
2447        };
2448        w.hold_task(&mut st, "failed");
2449        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2450        // A restart reconciling again does not hold twice.
2451        w.hold_task(&mut st, "failed again");
2452        assert!(w.unhold_task(&mut st));
2453        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2454        assert!(st.held_task.is_none());
2455
2456        // The owner re-held it by hand with their own reason: left alone.
2457        let mut h = q.get(&t.id).unwrap();
2458        st.held_task = Some(h.id.clone());
2459        h.hold_manual(Some("mine".to_owned()));
2460        q.put(&mut h).unwrap();
2461        assert!(w.unhold_task(&mut st));
2462        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2463    }
2464
2465    #[tokio::test]
2466    async fn a_restart_after_the_failure_was_saved_still_holds_the_task() {
2467        let (d, fake, w) = rig();
2468        *fake.local.lock().unwrap() = true;
2469        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2470        let repo = d.path().join("nowhere");
2471        let q = crate::queue::Queue::at(d.path().join("queue"));
2472        let mut t = crate::queue::Task::new(
2473            "t".to_owned(),
2474            "i".to_owned(),
2475            repo.clone(),
2476            crate::queue::Source::Human,
2477        );
2478        t.runs.push("run1".to_owned());
2479        t.status = crate::queue::TaskStatus::Done;
2480        q.put(&mut t).unwrap();
2481        // The state a crash leaves: failure persisted, task never held.
2482        let mut job = Job::new("1.0.0", URL, "deadbeef");
2483        job.failed = Some("command 1 exited 3".to_owned());
2484        let st = WatchState {
2485            repo: repo.to_string_lossy().into_owned(),
2486            url: URL.to_owned(),
2487            run: Some("run1".to_owned()),
2488            job: Some(job),
2489            ..WatchState::default()
2490        };
2491        w.save("o/r#7", &st);
2492        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2493        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Held);
2494    }
2495
2496    #[tokio::test]
2497    async fn a_restart_after_the_job_finished_still_gives_the_task_back() {
2498        let (d, fake, w) = rig();
2499        *fake.local.lock().unwrap() = true;
2500        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2501        let repo = d.path().join("nowhere");
2502        let q = crate::queue::Queue::at(d.path().join("queue"));
2503        let mut t = crate::queue::Task::new(
2504            "t".to_owned(),
2505            "i".to_owned(),
2506            repo.clone(),
2507            crate::queue::Source::Human,
2508        );
2509        t.runs.push("run1".to_owned());
2510        t.hold_machine(Some(format!("{HOLD_PREFIX}release v1.0.0 failed")));
2511        q.put(&mut t).unwrap();
2512        // The state a crash leaves: finished saved, task never given back.
2513        let mut job = Job::new("1.0.0", URL, "deadbeef");
2514        job.finished = true;
2515        let st = WatchState {
2516            repo: repo.to_string_lossy().into_owned(),
2517            url: URL.to_owned(),
2518            run: Some("run1".to_owned()),
2519            job: Some(job),
2520            held_task: Some(t.id.clone()),
2521            ..WatchState::default()
2522        };
2523        w.save("o/r#7", &st);
2524        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2525        assert_eq!(q.get(&t.id).unwrap().status, crate::queue::TaskStatus::Done);
2526        assert!(!w.state_path("o/r#7").exists());
2527    }
2528
2529    #[tokio::test]
2530    async fn a_pull_request_registered_at_open_is_picked_up_even_if_unlisted() {
2531        let (d, _fake, w) = rig();
2532        register(d.path(), Path::new("/r"), URL, "run1");
2533        let st = w.load("o/r#7");
2534        assert_eq!(st.url, URL);
2535        assert_eq!(st.repo, "/r");
2536        // Never overwrites a live record.
2537        let mut live = st.clone();
2538        live.head = "keep".to_owned();
2539        w.save("o/r#7", &live);
2540        register(d.path(), Path::new("/r"), URL, "run1");
2541        assert_eq!(w.load("o/r#7").head, "keep");
2542    }
2543
2544    #[tokio::test]
2545    async fn a_finished_release_is_kept_on_the_run_and_a_missing_run_still_completes() {
2546        let (d, fake, w) = rig();
2547        *fake.local.lock().unwrap() = true;
2548        *fake.snap.lock().unwrap() = Some(snap(PrLifecycle::Merged, "h1", vec![]));
2549        let repo = d.path().join("nowhere");
2550        let mut run = RunState::new(
2551            repo.clone(),
2552            "main".to_owned(),
2553            "0123456789abcdef".to_owned(),
2554            "t".to_owned(),
2555            crate::config::Config::default(),
2556        );
2557        run.id = "run1".to_owned();
2558        run.save_under(d.path()).unwrap();
2559        let mut job = Job::new("1.0.0", URL, "deadbeef");
2560        job.finished = true;
2561        job.log.push(release_local::StepLog {
2562            name: "cmd".to_owned(),
2563            code: Some(0),
2564            tail: "ok".to_owned(),
2565            output: None,
2566        });
2567        let mk = |run: &str| WatchState {
2568            repo: repo.to_string_lossy().into_owned(),
2569            url: URL.to_owned(),
2570            run: Some(run.to_owned()),
2571            job: Some(job.clone()),
2572            ..WatchState::default()
2573        };
2574        w.save("o/r#7", &mk("run1"));
2575        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2576        assert!(!w.state_path("o/r#7").exists());
2577        let kept = RunState::load_under("run1", d.path()).unwrap();
2578        assert_eq!(kept.release_bump.unwrap().release, Some(job.clone()));
2579
2580        // A run that is gone must not keep the watcher record forever.
2581        w.save("o/r#7", &mk("gone"));
2582        w.lap(std::slice::from_ref(&repo), 60, 1, &(|| false)).await;
2583        assert!(!w.state_path("o/r#7").exists());
2584    }
2585
2586    #[tokio::test]
2587    async fn raise_released_attaches_url_link() {
2588        let (d, _fake, w) = rig();
2589        let job = Job::new("1.0.0", URL, "deadbeef");
2590        w.raise_released("o/r#7", URL, &job);
2591        let ns = Notices::at(d.path().join("notifications")).list();
2592        assert_eq!(ns.len(), 1);
2593        assert_eq!(ns[0].key, "release-done:o/r#7");
2594        assert_eq!(
2595            ns[0].link,
2596            Some(Link::Url {
2597                url: URL.to_owned(),
2598            })
2599        );
2600    }
2601}