Skip to main content

magi/
notices.rs

1//! Notices: what needs the operator's attention but is not a question.
2//!
3//! A merged run whose release bump failed, a task held after three attempts, a
4//! disk gate that refused to start work - each used to end in a log line or in
5//! a question-shaped record nobody could answer. A [`Notice`] is the missing
6//! shape: something to *know*, with a severity, a timestamp and an optional
7//! link, that the operator marks read or dismisses. It is deliberately not a
8//! [`crate::ask::Question`]: nothing waits on it and there is nothing to answer.
9//!
10//! # Shape
11//!
12//! The same split as [`crate::queue`] and [`crate::ask`]. [`Notice`] is data
13//! plus *pure* transitions; [`Notices`] owns every filesystem call and is
14//! constructed with its root, so a test drives a real store in a temp
15//! directory and nothing here is process-global.
16//!
17//! One notice is one JSON file, written atomically, because `magi serve`,
18//! `magi web` and the CLI all write here from different processes and a rename
19//! is the only cross-process write that needs no coordination. The file name is
20//! derived from the notice's *key* by a stable function, which is what makes
21//! deduplication lock-free: two processes raising the same problem write the
22//! same file. A raise racing another may lose one increment of `count`; that is
23//! accepted.
24//!
25//! # Deduplication, and what dismissing means
26//!
27//! A key names a kind and a subject (`release-bump:<run>`, `task:<id>`), never
28//! the prose. Raising a key that exists bumps `count` and `last_at`. It returns
29//! the notice to *unread* only when the message changed or the severity rose:
30//! a retry loop re-raising the identical problem must not relight the bell.
31//! Dismissal is a tombstone (`dismissed_at`) rather than a delete, so the same
32//! message re-raised does not resurrect what the operator already waved away.
33//! The cost is that a genuine recurrence with identical wording stays hidden
34//! until the cap prunes the tombstone; producers should therefore keep
35//! messages stable and put anything that varies elsewhere.
36
37use std::path::{Path, PathBuf};
38
39use anyhow::{Context, Result, bail};
40use jiff::Timestamp;
41use serde::{Deserialize, Serialize};
42
43/// On-disk format. A file is refused only when its own schema is *greater*
44/// than this one; every field added later must carry `#[serde(default)]`.
45pub const SCHEMA: u32 = 1;
46
47/// The most notices kept. Pruned on every write: obsolete first, then
48/// dismissed, read and unread, oldest first within each.
49pub const CAP: usize = 200;
50
51/// How serious a notice is. Ordered so a rise is a plain comparison.
52#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
53#[serde(rename_all = "lowercase")]
54pub enum Severity {
55    /// Worth knowing, nothing to do.
56    Info,
57    /// Something needs a look soon.
58    Warn,
59    /// Something failed and stays failed until someone acts.
60    Error,
61}
62
63/// Where a notice points, when it points anywhere.
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(tag = "kind", rename_all = "lowercase")]
66pub enum Link {
67    /// A run, by id.
68    Run {
69        /// The run id.
70        id: String,
71    },
72    /// A queued task, by id.
73    Task {
74        /// The task id.
75        id: String,
76    },
77    /// An external page, such as a pull request.
78    Url {
79        /// The address.
80        url: String,
81    },
82}
83
84/// One thing the operator should know.
85#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct Notice {
87    /// Stable file-name-safe id derived from [`Notice::key`].
88    pub id: String,
89    /// Deduplication key: kind and subject, never prose.
90    pub key: String,
91    /// How serious.
92    pub severity: Severity,
93    /// What happened.
94    pub message: String,
95    /// Where it points, when it points anywhere.
96    #[serde(default)]
97    pub link: Option<Link>,
98    /// First raised.
99    pub first_at: Timestamp,
100    /// Last raised.
101    pub last_at: Timestamp,
102    /// How many times this key was raised.
103    #[serde(default = "one")]
104    pub count: u32,
105    /// When it was marked read.
106    #[serde(default)]
107    pub read_at: Option<Timestamp>,
108    /// When it was dismissed; the tombstone.
109    #[serde(default)]
110    pub dismissed_at: Option<Timestamp>,
111    /// The task and run ids this notice is about. A question whose `run`
112    /// names one of them already pages the operator for the same cause.
113    #[serde(default)]
114    pub subjects: Vec<String>,
115    /// The question that carries this notice as context, when it was filed
116    /// already read for that reason. Never a tombstone: a recurrence with a
117    /// changed message or higher severity is unread again.
118    #[serde(default)]
119    pub covered_by: Option<String>,
120    /// When the cause this notice reports began (a task's `held_at`). A
121    /// question filed before it is about an older cause and does not cover
122    /// it; `None` keeps the broad coverage of notices without one.
123    #[serde(default)]
124    pub since: Option<Timestamp>,
125    /// On-disk format version.
126    #[serde(default = "schema")]
127    pub schema: u32,
128}
129
130fn one() -> u32 {
131    1
132}
133
134fn schema() -> u32 {
135    SCHEMA
136}
137
138impl Notice {
139    /// A notice of `severity` for `key`.
140    pub fn new(severity: Severity, key: &str, message: impl Into<String>) -> Self {
141        let now = Timestamp::now();
142        Self {
143            id: id_of(key),
144            key: key.to_owned(),
145            severity,
146            message: message.into(),
147            link: None,
148            first_at: now,
149            last_at: now,
150            count: 1,
151            read_at: None,
152            dismissed_at: None,
153            subjects: Vec::new(),
154            covered_by: None,
155            since: None,
156            schema: SCHEMA,
157        }
158    }
159
160    /// Name the task / run ids this notice is about.
161    pub fn about<I, S>(mut self, subjects: I) -> Self
162    where
163        I: IntoIterator<Item = S>,
164        S: Into<String>,
165    {
166        self.subjects = subjects.into_iter().map(Into::into).collect();
167        self
168    }
169
170    /// An [`Severity::Info`] notice.
171    pub fn info(key: &str, message: impl Into<String>) -> Self {
172        Self::new(Severity::Info, key, message)
173    }
174
175    /// A [`Severity::Warn`] notice.
176    pub fn warn(key: &str, message: impl Into<String>) -> Self {
177        Self::new(Severity::Warn, key, message)
178    }
179
180    /// An [`Severity::Error`] notice.
181    pub fn error(key: &str, message: impl Into<String>) -> Self {
182        Self::new(Severity::Error, key, message)
183    }
184
185    /// Record when the reported cause began.
186    pub fn since(mut self, at: Option<Timestamp>) -> Self {
187        self.since = at;
188        self
189    }
190
191    /// Attach a link.
192    pub fn link(mut self, link: Link) -> Self {
193        self.link = Some(link);
194        self
195    }
196
197    /// Neither read nor dismissed.
198    pub fn unread(&self) -> bool {
199        self.read_at.is_none() && self.dismissed_at.is_none()
200    }
201
202    /// Fold a repeat of this notice into it. Always counts; only a changed
203    /// message or a higher severity makes it unread and visible again.
204    ///
205    /// Returns whether this raise should page the operator: it relit the
206    /// notice (new message, higher severity, newer cause) and no open question
207    /// already carries it. This is the only place that decides.
208    pub fn raise_again(&mut self, again: &Notice, now: Timestamp) -> bool {
209        self.count = self.count.saturating_add(1);
210        self.last_at = now;
211        if again.link.is_some() {
212            self.link = again.link.clone();
213        }
214        if !again.subjects.is_empty() {
215            self.subjects = again.subjects.clone();
216        }
217        let escalated = again.severity > self.severity;
218        // A newer cause is news even when it reads the same; an old record
219        // with no `since` just adopts one without lighting up again.
220        // A legacy record that a question had silenced has no `since` to
221        // compare, but a stamped raise that no question covers any more is not
222        // the cause that question was filed for: it pages.
223        let newer_cause = match (self.since, again.since) {
224            (Some(old), Some(new)) => new > old,
225            (None, Some(_)) => self.covered_by.is_some() && again.covered_by.is_none(),
226            _ => false,
227        };
228        if again.since.is_some() {
229            self.since = again.since;
230        }
231        let relit = again.message != self.message || escalated || newer_cause;
232        if relit {
233            self.message = again.message.clone();
234            self.severity = self.severity.max(again.severity);
235            self.read_at = None;
236            self.dismissed_at = None;
237            self.covered_by = None;
238        }
239        // A question already pages for this cause: keep the record, skip the
240        // second page. A higher severity is news the question did not carry.
241        if let Some(q) = &again.covered_by
242            && !escalated
243            && self.unread()
244        {
245            self.read_at = Some(now);
246            self.covered_by = Some(q.clone());
247        }
248        relit && self.covered_by.is_none()
249    }
250
251    /// Mark read, keeping the first read time.
252    pub fn mark_read(&mut self, now: Timestamp) {
253        if self.read_at.is_none() {
254            self.read_at = Some(now);
255        }
256    }
257
258    /// Tombstone: read and hidden from the list.
259    pub fn dismiss(&mut self, now: Timestamp) {
260        self.mark_read(now);
261        if self.dismissed_at.is_none() {
262            self.dismissed_at = Some(now);
263        }
264    }
265
266    /// Legacy exhausted-deputy warnings belong on the question card now.
267    /// Match only this retired namespace, not `deputy-turn:` failures.
268    fn obsolete(&self) -> bool {
269        self.key.starts_with("deputy:")
270    }
271
272    /// Whether this notice belongs in the notification centre and bell.
273    fn visible(&self) -> bool {
274        !self.obsolete() && self.dismissed_at.is_none()
275    }
276
277    /// Pruning order: obsolete, dismissed, read, then unread.
278    fn keep_rank(&self) -> u8 {
279        if self.obsolete() {
280            0
281        } else if self.dismissed_at.is_some() {
282            1
283        } else if self.read_at.is_some() {
284            2
285        } else {
286            3
287        }
288    }
289}
290
291/// A stable, file-name-safe id for a key: a readable slug plus FNV-1a of the
292/// whole key. Not `DefaultHasher`, whose output may change between builds and
293/// would strand every existing file's dedupe.
294pub fn id_of(key: &str) -> String {
295    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
296    for b in key.bytes() {
297        hash ^= u64::from(b);
298        hash = hash.wrapping_mul(0x0100_0000_01b3);
299    }
300    let slug: String = key
301        .chars()
302        .map(|c| {
303            if c.is_ascii_alphanumeric() {
304                c.to_ascii_lowercase()
305            } else {
306                '-'
307            }
308        })
309        .take(32)
310        .collect();
311    format!("{slug}-{hash:016x}")
312}
313
314/// Whether `id` could have come from [`id_of`]. Checked before any path is
315/// built from a request.
316fn valid_id(id: &str) -> bool {
317    !id.is_empty()
318        && id.len() <= 64
319        && id
320            .bytes()
321            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
322}
323
324/// How long a caller waits for a notice's lock before proceeding without it.
325const LOCK_WAIT: std::time::Duration = std::time::Duration::from_secs(5);
326
327/// Age after which a lock file is taken to belong to a dead process.
328const LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
329
330/// Distinguishes temp files written by threads of one process.
331static TMP_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
332
333/// A notice store on disk.
334#[derive(Debug, Clone)]
335pub struct Notices {
336    root: PathBuf,
337}
338
339impl Notices {
340    /// The operator's notices, `<home>/notifications`.
341    pub fn open() -> Self {
342        Self::at(crate::run::home().join("notifications"))
343    }
344
345    /// A store at an explicit root; tests use this.
346    pub fn at(root: PathBuf) -> Self {
347        Self { root }
348    }
349
350    /// Directory holding the notice files.
351    pub fn root(&self) -> &Path {
352        &self.root
353    }
354
355    fn path_of(&self, id: &str) -> PathBuf {
356        self.root.join(format!("{id}.json"))
357    }
358
359    fn put(&self, n: &Notice) -> Result<()> {
360        std::fs::create_dir_all(&self.root)
361            .with_context(|| format!("create {}", self.root.display()))?;
362        let body = serde_json::to_string_pretty(n).context("serialize notice")?;
363        let path = self.path_of(&n.id);
364        // Unique per write: two processes raising or marking the same notice
365        // must not share a temp file, or the second rename finds it gone.
366        let tmp = path.with_extension(format!(
367            "json.{}.{}.tmp",
368            std::process::id(),
369            TMP_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
370        ));
371        std::fs::write(&tmp, body).with_context(|| format!("write {}", tmp.display()))?;
372        if let Err(e) = std::fs::rename(&tmp, &path) {
373            let _ = std::fs::remove_file(&tmp);
374            return Err(e).with_context(|| format!("replace {}", path.display()));
375        }
376        Ok(())
377    }
378
379    /// Load one notice by id.
380    pub fn get(&self, id: &str) -> Result<Notice> {
381        if !valid_id(id) {
382            bail!("`{id}` is not a notification id");
383        }
384        read_path(&self.path_of(id))
385    }
386
387    /// Run a read-modify-write of one notice under its lock file.
388    ///
389    /// `magi serve`, `magi web` and the CLI all update the same files, and a
390    /// unique temp file only makes each write atomic - it does not stop one
391    /// process saving a stale read over another's newer raise. The lock is an
392    /// exclusive-create file beside the notice; a holder that died is broken
393    /// after [`LOCK_STALE`], and a caller that cannot get it in
394    /// [`LOCK_WAIT`] proceeds anyway rather than lose the notice.
395    fn locked<T>(&self, id: &str, f: impl FnOnce() -> Result<T>) -> Result<T> {
396        std::fs::create_dir_all(&self.root)
397            .with_context(|| format!("create {}", self.root.display()))?;
398        let lock = self.root.join(format!("{id}.lock"));
399        let start = std::time::Instant::now();
400        let mut held = false;
401        while start.elapsed() < LOCK_WAIT {
402            match std::fs::OpenOptions::new()
403                .write(true)
404                .create_new(true)
405                .open(&lock)
406            {
407                Ok(_) => {
408                    held = true;
409                    break;
410                }
411                Err(_) => {
412                    let stale = std::fs::metadata(&lock)
413                        .and_then(|m| m.modified())
414                        .ok()
415                        .and_then(|t| t.elapsed().ok())
416                        .is_some_and(|age| age > LOCK_STALE);
417                    if stale {
418                        let _ = std::fs::remove_file(&lock);
419                    } else {
420                        std::thread::sleep(std::time::Duration::from_millis(5));
421                    }
422                }
423            }
424        }
425        let out = f();
426        if held {
427            let _ = std::fs::remove_file(&lock);
428        }
429        out
430    }
431
432    /// File a notice, folding it into an existing one with the same key.
433    pub fn raise(&self, incoming: Notice) -> Result<Notice> {
434        self.raise_paged(incoming).map(|(n, _)| n)
435    }
436
437    /// [`Notices::raise`], also saying whether the raise should page.
438    pub fn raise_paged(&self, incoming: Notice) -> Result<(Notice, bool)> {
439        let out = self.locked(&incoming.id.clone(), || {
440            let now = Timestamp::now();
441            let (stored, page) = match read_path(&self.path_of(&incoming.id)) {
442                Ok(mut existing) => {
443                    let page = existing.raise_again(&incoming, now);
444                    (existing, page)
445                }
446                Err(_) => {
447                    let mut fresh = incoming;
448                    if fresh.covered_by.is_some() {
449                        fresh.read_at = Some(now);
450                    }
451                    let page = fresh.covered_by.is_none();
452                    (fresh, page)
453                }
454            };
455            self.put(&stored)?;
456            Ok((stored, page))
457        })?;
458        self.prune();
459        Ok(out)
460    }
461
462    /// Mark every unread notice about `subject` read, recording `question` as
463    /// the card that carries it. Best-effort per notice.
464    pub fn cover(&self, question: &crate::ask::Question) {
465        for n in self.all() {
466            if n.unread() && covers(question, &n) {
467                let _ = self.update(&n.id, |n| {
468                    // Re-checked under the lock: a newer cause may have
469                    // landed since `all()` read it.
470                    if n.unread() && covers(question, n) {
471                        n.mark_read(Timestamp::now());
472                        n.covered_by = Some(question.id.clone());
473                    }
474                });
475            }
476        }
477    }
478
479    /// Load, change and save one notice under its lock.
480    fn update(&self, id: &str, change: impl FnOnce(&mut Notice)) -> Result<Notice> {
481        if !valid_id(id) {
482            bail!("`{id}` is not a notification id");
483        }
484        self.locked(id, || {
485            let mut n = read_path(&self.path_of(id))?;
486            change(&mut n);
487            self.put(&n)?;
488            Ok(n)
489        })
490    }
491
492    /// Every notice on disk, unreadable files skipped, newest first.
493    fn all(&self) -> Vec<Notice> {
494        let mut all: Vec<Notice> = std::fs::read_dir(&self.root)
495            .into_iter()
496            .flatten()
497            .flatten()
498            .map(|e| e.path())
499            .filter(|p| p.extension().is_some_and(|x| x == "json"))
500            .filter_map(|p| read_path(&p).ok())
501            .collect();
502        all.sort_by(|a, b| b.last_at.cmp(&a.last_at).then_with(|| a.id.cmp(&b.id)));
503        all
504    }
505
506    /// What the bell lists: visible, current notices, newest first.
507    pub fn list(&self) -> Vec<Notice> {
508        self.all().into_iter().filter(Notice::visible).collect()
509    }
510
511    /// How many notices are unread.
512    pub fn count_unread(&self) -> usize {
513        self.list().iter().filter(|n| n.unread()).count()
514    }
515
516    /// Mark read, keeping the first read time.
517    pub fn mark_read(&self, id: &str) -> Result<Notice> {
518        let now = Timestamp::now();
519        self.update(id, |n| n.mark_read(now))
520    }
521
522    /// Mark every unread notice read; returns how many changed.
523    pub fn mark_all_read(&self) -> Result<usize> {
524        let now = Timestamp::now();
525        let mut changed = 0;
526        for n in self.list().into_iter().filter(Notice::unread) {
527            // Re-read under the lock: a raise since the scan may have made it
528            // a different, still-unread notice, which stays as it is.
529            let done = self.update(&n.id, |n| n.mark_read(now));
530            if done.is_ok() {
531                changed += 1;
532            }
533        }
534        Ok(changed)
535    }
536
537    /// Tombstone: read and hidden from the list.
538    pub fn dismiss(&self, id: &str) -> Result<Notice> {
539        let now = Timestamp::now();
540        self.update(id, |n| n.dismiss(now))
541    }
542
543    /// Keep at most [`CAP`] files. Best-effort.
544    fn prune(&self) {
545        let mut all = self.all();
546        if all.len() <= CAP {
547            return;
548        }
549        // Most worth keeping first; drop the tail.
550        all.sort_by(|a, b| {
551            b.keep_rank()
552                .cmp(&a.keep_rank())
553                .then_with(|| b.last_at.cmp(&a.last_at))
554        });
555        for n in all.split_off(CAP) {
556            let _ = std::fs::remove_file(self.path_of(&n.id));
557        }
558    }
559
560    /// Change token for the stream. Hashes every file's name and mtime, so a
561    /// prune or delete moves it as surely as a write does (a max-mtime would
562    /// not move when the newest file is not the one removed).
563    pub fn revision(&self) -> u64 {
564        use std::hash::{Hash as _, Hasher as _};
565        let mut entries: Vec<(std::ffi::OsString, u128)> = std::fs::read_dir(&self.root)
566            .into_iter()
567            .flatten()
568            .flatten()
569            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
570            .map(|e| {
571                let at = e
572                    .metadata()
573                    .and_then(|m| m.modified())
574                    .ok()
575                    .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
576                    .map_or(0, |d| d.as_nanos());
577                (e.file_name(), at)
578            })
579            .collect();
580        if entries.is_empty() {
581            return 0;
582        }
583        entries.sort();
584        let mut h = std::collections::hash_map::DefaultHasher::new();
585        entries.hash(&mut h);
586        h.finish()
587    }
588}
589
590fn read_path(path: &Path) -> Result<Notice> {
591    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
592    let n: Notice =
593        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
594    if n.schema > SCHEMA {
595        bail!(
596            "notice {} was written by a newer magi (schema {}, this build speaks up to {SCHEMA})",
597            n.id,
598            n.schema
599        );
600    }
601    Ok(n)
602}
603
604/// The notice for a run that ended Blocked, Stalled or Failed, if it did.
605///
606/// A parked run was only asked to stop, so it is not news. A run that left a
607/// pull request behind still is: the PR is waiting on someone. Keyed on the
608/// run with a message free of anything that varies between attempts.
609pub fn run_ended(state: &crate::run::RunState) -> Option<Notice> {
610    use crate::run::RunStatus;
611    let failed = matches!(
612        state.status,
613        RunStatus::Blocked | RunStatus::Stalled | RunStatus::Failed
614    );
615    (failed && !state.parked).then(|| {
616        Notice::error(
617            &format!("run:{}", state.id),
618            format!("Run {} ended {}.", state.short(), state.status.as_str()),
619        )
620        .link(Link::Run {
621            id: state.id.clone(),
622        })
623        .about([state.id.clone()])
624    })
625}
626
627/// The notice for a pull request that merged while checks were red.
628///
629/// Keyed on the run. `summary` names the repository, pull request and failing
630/// checks: with no notify command configured this is the only place the
631/// operator learns them. A run merges once, so the wording cannot churn.
632pub fn merged_red(run_id: &str, summary: &str) -> Notice {
633    Notice::warn(&format!("merged-red:{run_id}"), summary).link(Link::Run {
634        id: run_id.to_owned(),
635    })
636}
637
638/// The notice for a task the machine or its attempt budget has held, if it is.
639///
640/// Called from [`crate::queue::Queue::put`], which every task transition goes
641/// through, so a hold made anywhere - the loop, the conductor, triage, a
642/// dependency removed from under a blocked task - is announced. A hold the
643/// operator placed by hand is their own action and is not news. Keyed on the
644/// task, with wording free of anything that varies between retries.
645///
646/// Falls back to [`crate::queue::Task::last_error`] when `hold_reason` is
647/// empty, the same fallback `crate::triage`'s question detail uses: a record
648/// written before `Task::fail`/`Task::handed_off` started copying `why` into
649/// `hold_reason` too would otherwise still read "no reason recorded" even
650/// though the run said exactly why it stopped.
651pub fn task_held(task: &crate::queue::Task) -> Option<Notice> {
652    use crate::queue::{HoldSource, TaskStatus};
653    if task.status != TaskStatus::Held || task.hold_source != Some(HoldSource::Machine) {
654        return None;
655    }
656    let why = task
657        .hold_reason
658        .as_deref()
659        .or(task.last_error.as_deref())
660        .unwrap_or("no reason recorded");
661    Some(
662        Notice::warn(
663            &format!("task:{}", task.id),
664            format!("Task {} is held: {why}.", task.short()),
665        )
666        .link(Link::Task {
667            id: task.id.clone(),
668        })
669        .about([task.id.clone()])
670        .since(task.held_at),
671    )
672}
673
674/// The notice for a run whose graph returned an error before it settled.
675pub fn run_stopped(id: &str, state: &crate::run::RunState) -> Notice {
676    Notice::error(
677        &format!("run:{id}"),
678        format!("Run {} stopped with an error.", state.short()),
679    )
680    .link(Link::Run { id: id.to_owned() })
681    .about([id.to_owned()])
682}
683
684/// One page for the operator's `[notify]` command.
685#[derive(Debug, Clone, PartialEq, Eq)]
686pub struct Page {
687    /// The run id when the notice links to a run, else empty.
688    pub run: String,
689    /// The notice message.
690    pub summary: String,
691}
692
693/// Set once by `main`: until then no notice fires the command, so a unit or
694/// integration test that pins a home never reaches the operator's real config.
695static PAGER: std::sync::OnceLock<()> = std::sync::OnceLock::new();
696
697/// Let notices fire the `[notify]` command in this process.
698pub fn install_pager() {
699    let _ = PAGER.set(());
700}
701
702/// The production sender. Fire and forget on its own thread with its own
703/// runtime, so it works with or without an ambient one and never blocks the
704/// producer; `ask::notify_text` bounds the command by `NOTIFY_TIMEOUT`.
705/// `notify` is the producer's own config, else the machine layer's.
706fn send_page(page: Page, notify: Option<crate::config::Notify>) {
707    if PAGER.get().is_none() {
708        return;
709    }
710    let spawned = std::thread::Builder::new()
711        .name("notice-pager".into())
712        .spawn(move || {
713            let notify = notify.unwrap_or_else(machine_notify);
714            if notify.command.is_empty() {
715                return;
716            }
717            let rt = match tokio::runtime::Builder::new_current_thread()
718                .enable_all()
719                .build()
720            {
721                Ok(rt) => rt,
722                Err(e) => return tracing::warn!("could not page about a notification: {e:#}"),
723            };
724            if let Err(e) = rt.block_on(crate::ask::notify_text(&notify, &page.run, &page.summary))
725            {
726                tracing::warn!("could not page about a notification: {e:#}");
727            }
728        });
729    if let Err(e) = spawned {
730        tracing::warn!("could not page about a notification: {e:#}");
731    }
732}
733
734/// `[notify]` from the machine layer alone; a producer with no repo config
735/// cannot name a better one. Unreadable is no command, with a warning.
736fn machine_notify() -> crate::config::Notify {
737    let layers: Vec<PathBuf> = crate::config::Config::machine_layer()
738        .into_iter()
739        .filter(|p| p.is_file())
740        .collect();
741    match crate::config::Config::load_layers(&layers) {
742        Ok(c) => c.notify,
743        Err(e) => {
744            tracing::warn!("could not read [notify] for a notification: {e:#}");
745            Default::default()
746        }
747    }
748}
749
750/// The one function producers call. Best-effort by construction: a notice that
751/// cannot be filed is a `tracing::warn`, never a reason to fail the run, the
752/// loop or the request that wanted to mention it.
753pub fn raise(notice: Notice) {
754    if let Some(home) = crate::run::try_home() {
755        raise_in(&home, notice);
756    }
757}
758
759/// [`raise`] for a producer that already holds the repo's `[notify]`.
760pub fn raise_with(notice: Notice, notify: &crate::config::Notify) {
761    if let Some(home) = crate::run::try_home() {
762        raise_in_with(&home, notice, &|page| send_page(page, Some(notify.clone())));
763    }
764}
765
766/// Does open question `q` already page the operator for what `n` reports?
767///
768/// The cause is decided by what each side is about, not by when it happened:
769/// a conductor / triage question exists *because* its task is held, so it
770/// covers that task's hold and handover notices whenever it was filed; a
771/// question raised from inside a run (any other seat node) covers that run's
772/// ended / stopped notices and never a task hold. Exact match on
773/// `Question::run` (a task id for the former, a run id for the latter). The
774/// land approval and release questions are to-dos, not duplicates, and cover
775/// nothing.
776///
777/// A notice with a `since` (a task hold's start) is covered only by a question
778/// filed at or after it: one filed earlier is about an older cause, and
779/// letting it silence a new hold would hide a fresh failure behind a stale
780/// card. Missing a duplicate costs one extra page; hiding a failure costs the
781/// failure. A notice without `since` keeps the broad rule.
782pub fn covers(q: &crate::ask::Question, n: &Notice) -> bool {
783    if !q.status.open() || !n.subjects.contains(&q.run) {
784        return false;
785    }
786    if n.since.is_some_and(|since| q.asked_at < since) {
787        return false;
788    }
789    let about_task = n.key.starts_with("task:") || n.key.starts_with("handover:");
790    let about_run = n.key.starts_with("run:");
791    match q.node.as_str() {
792        crate::bump::NOTICE_NODE | crate::land::APPROVAL_NODE => false,
793        _ if q.run_names_task() => about_task,
794        _ => about_run,
795    }
796}
797
798fn covering_question(home: &Path, n: &Notice) -> Option<String> {
799    if n.subjects.is_empty() {
800        return None;
801    }
802    crate::ask::Questions::at(home.join("questions"))
803        .list()
804        .into_iter()
805        .find(|q| covers(q, n))
806        .map(|q| q.id)
807}
808
809/// A question was just filed for `run`: quiet the unread notices about it.
810pub fn quiet_for(home: &Path, question: &crate::ask::Question) {
811    Notices::at(home.join("notifications")).cover(question);
812}
813
814/// [`raise`] into an explicit magi home, for callers that already carry one
815/// (the janitor) and so must not reach for the process-global.
816pub fn raise_in(home: &Path, notice: Notice) {
817    raise_in_with(home, notice, &|page| send_page(page, None));
818}
819
820/// [`raise_in`] with the pager injected. `send` is called only when the raise
821/// should page (see [`Notice::raise_again`]); it must not block.
822pub fn raise_in_with(home: &Path, mut notice: Notice, send: &dyn Fn(Page)) {
823    if let Some(q) = covering_question(home, &notice) {
824        notice.covered_by = Some(q);
825    }
826    match Notices::at(home.join("notifications")).raise_paged(notice) {
827        Ok((n, true)) => send(Page {
828            run: match &n.link {
829                Some(Link::Run { id }) => id.clone(),
830                _ => String::new(),
831            },
832            summary: n.message.clone(),
833        }),
834        Ok(_) => {}
835        Err(e) => tracing::warn!("could not file a notification: {e:#}"),
836    }
837}
838
839#[cfg(test)]
840mod tests {
841    use super::*;
842
843    #[test]
844    fn legacy_deputy_warnings_are_hidden_without_hiding_other_namespaces() {
845        let (_dir, s) = store();
846        // No question lookup: these may belong to deleted or settled questions.
847        for key in ["deputy:open", "deputy:answered", "deputy:deleted"] {
848            s.raise(Notice::warn(key, "old exhausted warning")).unwrap();
849        }
850        let keys = ["deputy-turn:open", "deputyish:open", "deputy", "task:open"];
851        for key in keys {
852            s.raise(Notice::warn(key, "still relevant")).unwrap();
853        }
854        assert_eq!(s.list().len(), keys.len());
855        assert_eq!(s.count_unread(), keys.len());
856        assert!(s.list().iter().all(|n| keys.contains(&n.key.as_str())));
857        assert_eq!(s.mark_all_read().unwrap(), keys.len());
858        assert_eq!(s.count_unread(), 0);
859        assert!(s.get(&id_of("deputy:open")).unwrap().unread());
860    }
861
862    #[test]
863    fn obsolete_deputy_warnings_are_pruned_before_current_notices() {
864        let (_dir, s) = store();
865        for i in 0..CAP {
866            let id = s
867                .raise(Notice::info(&format!("current:{i}"), "keep"))
868                .unwrap()
869                .id;
870            s.dismiss(&id).unwrap();
871        }
872        // Even a newer unread legacy warning cannot evict a current tombstone.
873        s.raise(Notice::warn("deputy:old", "obsolete")).unwrap();
874        assert!(s.get(&id_of("deputy:old")).is_err());
875        assert_eq!(s.all().len(), CAP);
876        assert!(s.all().iter().all(|n| n.key.starts_with("current:")));
877    }
878
879    fn store() -> (tempfile::TempDir, Notices) {
880        let dir = tempfile::tempdir().unwrap();
881        let s = Notices::at(dir.path().join("notifications"));
882        (dir, s)
883    }
884
885    #[test]
886    fn a_red_merge_notice_is_keyed_on_the_run_and_raised_once() {
887        let (_d, s) = store();
888        let n = merged_red("run-1", "Merged a/b PR #1 with red checks: x (u)");
889        assert_eq!(n.key, "merged-red:run-1");
890        assert_eq!(
891            n.link,
892            Some(Link::Run {
893                id: "run-1".to_owned()
894            })
895        );
896        s.raise(merged_red(
897            "run-1",
898            "Merged a/b PR #1 with red checks: x (u)",
899        ))
900        .unwrap();
901        s.raise(merged_red(
902            "run-1",
903            "Merged a/b PR #1 with red checks: x (u)",
904        ))
905        .unwrap();
906        assert_eq!(s.list().len(), 1);
907    }
908
909    #[test]
910    fn a_repeat_counts_but_stays_read_and_a_change_relights_it() {
911        let now = Timestamp::now();
912        let mut n = Notice::warn("task:1", "held");
913        n.mark_read(now);
914        n.raise_again(&Notice::warn("task:1", "held"), now);
915        assert_eq!(n.count, 2);
916        assert!(n.read_at.is_some(), "identical repeat stays read");
917        n.raise_again(&Notice::warn("task:1", "held differently"), now);
918        assert!(n.unread());
919        assert_eq!(n.message, "held differently");
920    }
921
922    #[test]
923    fn a_rise_in_severity_resurrects_a_dismissed_notice_but_a_repeat_does_not() {
924        let now = Timestamp::now();
925        let mut n = Notice::warn("task:x", "m");
926        n.dismiss(now);
927        n.raise_again(&Notice::warn("task:x", "m"), now);
928        assert!(n.dismissed_at.is_some(), "tombstone holds");
929        n.raise_again(&Notice::error("k", "m"), now);
930        assert!(n.unread());
931        assert_eq!(n.severity, Severity::Error);
932        // Severity never falls on a repeat.
933        n.raise_again(&Notice::info("k", "other"), now);
934        assert_eq!(n.severity, Severity::Error);
935    }
936
937    #[test]
938    fn adding_link_on_repeat_updates_link_without_relighting_read_or_dismissed() {
939        let now = Timestamp::now();
940        // 1. Read notice
941        let mut read_notice = Notice::warn("task:1", "held");
942        read_notice.mark_read(now);
943        let link_to_add = Link::Task { id: "1".to_owned() };
944        let again = Notice::warn("task:1", "held").link(link_to_add.clone());
945        let relit = read_notice.raise_again(&again, now);
946        assert!(!relit, "link-only update must not page/relight");
947        assert!(read_notice.read_at.is_some(), "read notice stays read");
948        assert_eq!(read_notice.link, Some(link_to_add.clone()));
949
950        // 2. Dismissed notice
951        let mut dismissed_notice = Notice::warn("task:2", "held");
952        dismissed_notice.dismiss(now);
953        let link_to_add_2 = Link::Task { id: "2".to_owned() };
954        let again_2 = Notice::warn("task:2", "held").link(link_to_add_2.clone());
955        let relit_2 = dismissed_notice.raise_again(&again_2, now);
956        assert!(!relit_2, "link-only update must not page/relight");
957        assert!(dismissed_notice.dismissed_at.is_some(), "tombstone holds");
958        assert_eq!(dismissed_notice.link, Some(link_to_add_2));
959    }
960
961    #[test]
962    fn identical_raises_share_one_file() {
963        let (_d, s) = store();
964        for _ in 0..5 {
965            s.raise(Notice::error("run:abc", "blocked")).unwrap();
966        }
967        let all = s.list();
968        assert_eq!(all.len(), 1);
969        assert_eq!(all[0].count, 5);
970        assert_eq!(s.count_unread(), 1);
971    }
972
973    #[test]
974    fn transitions_persist_and_dismissed_leave_the_list() {
975        let (_d, s) = store();
976        let a = s.raise(Notice::info("a", "one")).unwrap();
977        let b = s.raise(Notice::warn("b", "two")).unwrap();
978        assert_eq!(s.count_unread(), 2);
979        s.mark_read(&a.id).unwrap();
980        assert_eq!(s.count_unread(), 1);
981        s.dismiss(&b.id).unwrap();
982        assert_eq!(s.count_unread(), 0);
983        assert_eq!(s.list().len(), 1);
984        s.raise(Notice::warn("b", "two")).unwrap();
985        assert_eq!(
986            s.list().len(),
987            1,
988            "a tombstone survives a same-message raise"
989        );
990        s.raise(Notice::info("c", "three")).unwrap();
991        assert_eq!(s.mark_all_read().unwrap(), 1);
992        assert_eq!(s.count_unread(), 0);
993    }
994
995    #[test]
996    fn the_list_is_newest_first() {
997        let (_d, s) = store();
998        let mut old = Notice::info("old", "old");
999        old.last_at = "2020-01-01T00:00:00Z".parse().unwrap();
1000        s.put(&old).unwrap();
1001        s.raise(Notice::info("new", "new")).unwrap();
1002        let keys: Vec<_> = s.list().into_iter().map(|n| n.key).collect();
1003        assert_eq!(keys, ["new", "old"]);
1004    }
1005
1006    #[test]
1007    fn the_cap_prunes_dismissed_then_read_then_oldest() {
1008        let (_d, s) = store();
1009        let keep = s.raise(Notice::error("keep", "unread")).unwrap();
1010        let gone = s.raise(Notice::info("gone", "dismissed")).unwrap();
1011        s.dismiss(&gone.id).unwrap();
1012        for i in 0..CAP - 1 {
1013            s.raise(Notice::info(&format!("k{i}"), "x")).unwrap();
1014        }
1015        let files = std::fs::read_dir(s.root()).unwrap().count();
1016        assert_eq!(files, CAP);
1017        assert!(
1018            s.get(&keep.id).is_ok(),
1019            "an unread notice outlives a dismissed one"
1020        );
1021        assert!(s.get(&gone.id).is_err(), "the dismissed one went first");
1022    }
1023
1024    #[test]
1025    fn revision_moves_on_write_and_on_removal() {
1026        let (_d, s) = store();
1027        assert_eq!(s.revision(), 0);
1028        let a = s.raise(Notice::info("a", "m")).unwrap();
1029        let r1 = s.revision();
1030        assert_ne!(r1, 0);
1031        s.raise(Notice::info("b", "m")).unwrap();
1032        let r2 = s.revision();
1033        assert_ne!(r1, r2);
1034        std::fs::remove_file(s.path_of(&a.id)).unwrap();
1035        assert_ne!(s.revision(), r2);
1036    }
1037
1038    #[test]
1039    fn ids_are_stable_and_untrusted_ids_are_refused() {
1040        assert_eq!(id_of("run:1"), id_of("run:1"));
1041        assert_ne!(id_of("run:1"), id_of("run-1"));
1042        assert!(valid_id(&id_of("release-bump:20260101-abc")));
1043        let (_d, s) = store();
1044        for bad in ["", "../x", "a/b", "A", "x.json"] {
1045            assert!(s.get(bad).is_err(), "{bad}");
1046        }
1047    }
1048
1049    #[test]
1050    fn a_newer_schema_is_refused_and_an_older_reads() {
1051        let (_d, s) = store();
1052        let mut n = Notice::info("k", "m");
1053        n.schema = SCHEMA + 1;
1054        s.put(&n).unwrap();
1055        assert!(s.get(&n.id).is_err());
1056        let old = r#"{"id":"x-1","key":"x","severity":"warn","message":"m",
1057            "first_at":"2020-01-01T00:00:00Z","last_at":"2020-01-01T00:00:00Z"}"#;
1058        std::fs::write(s.path_of("x-1"), old).unwrap();
1059        assert_eq!(s.get("x-1").unwrap().count, 1);
1060    }
1061
1062    fn state(status: crate::run::RunStatus) -> crate::run::RunState {
1063        let mut st = crate::run::RunState::new(
1064            std::path::PathBuf::from("/repo"),
1065            "main".to_owned(),
1066            "abc1234def".to_owned(),
1067            "task".to_owned(),
1068            crate::config::Config::default(),
1069        );
1070        st.status = status;
1071        st
1072    }
1073
1074    #[test]
1075    fn a_run_that_ended_badly_is_news_unless_it_only_parked() {
1076        use crate::run::RunStatus;
1077        for bad in [RunStatus::Blocked, RunStatus::Stalled, RunStatus::Failed] {
1078            let st = state(bad);
1079            let n = run_ended(&st).expect("news");
1080            assert_eq!(n.key, format!("run:{}", st.id));
1081            assert_eq!(n.severity, Severity::Error);
1082        }
1083        assert!(run_ended(&state(RunStatus::Merged)).is_none());
1084        let mut parked = state(RunStatus::Stalled);
1085        parked.parked = true;
1086        assert!(run_ended(&parked).is_none());
1087    }
1088
1089    #[test]
1090    fn concurrent_writers_of_one_key_all_succeed() {
1091        let (_d, s) = store();
1092        let handles: Vec<_> = (0..8)
1093            .map(|_| {
1094                let s = s.clone();
1095                std::thread::spawn(move || {
1096                    for _ in 0..20 {
1097                        s.raise(Notice::warn("same", "m")).unwrap();
1098                    }
1099                })
1100            })
1101            .collect();
1102        for h in handles {
1103            h.join().unwrap();
1104        }
1105        assert_eq!(s.list().len(), 1);
1106        let stray = std::fs::read_dir(s.root())
1107            .unwrap()
1108            .flatten()
1109            .filter(|e| e.path().extension().is_some_and(|x| x == "tmp"))
1110            .count();
1111        assert_eq!(stray, 0);
1112    }
1113
1114    #[test]
1115    fn a_dead_holders_lock_is_broken_and_no_lock_is_left_behind() {
1116        let (_d, s) = store();
1117        let n = s.raise(Notice::info("k", "m")).unwrap();
1118        let lock = s.root().join(format!("{}.lock", n.id));
1119        std::fs::write(&lock, "").unwrap();
1120        let old = std::time::SystemTime::now() - std::time::Duration::from_secs(60);
1121        std::fs::File::options()
1122            .write(true)
1123            .open(&lock)
1124            .unwrap()
1125            .set_modified(old)
1126            .unwrap();
1127        s.mark_read(&n.id).unwrap();
1128        assert!(!lock.exists());
1129        assert!(s.get(&n.id).unwrap().read_at.is_some());
1130    }
1131
1132    #[test]
1133    fn only_a_machine_hold_is_a_task_notice() {
1134        let mut t = crate::queue::Task::new(
1135            "t".to_owned(),
1136            "do it".to_owned(),
1137            std::path::PathBuf::from("/repo"),
1138            crate::queue::Source::Human,
1139        );
1140        assert!(task_held(&t).is_none());
1141        t.hold_manual(None);
1142        assert!(task_held(&t).is_none(), "the operator's own hold");
1143        t.hold_machine(Some("missing blocker".to_owned()));
1144        let n = task_held(&t).expect("machine hold");
1145        assert_eq!(n.key, format!("task:{}", t.id));
1146        assert!(n.message.contains("missing blocker"));
1147    }
1148
1149    fn held_task() -> crate::queue::Task {
1150        let mut t = crate::queue::Task::new(
1151            "t".to_owned(),
1152            "do it".to_owned(),
1153            std::path::PathBuf::from("/repo"),
1154            crate::queue::Source::Human,
1155        );
1156        t.hold_machine(Some("branch b is checked out".to_owned()));
1157        t
1158    }
1159
1160    fn ask(home: &Path, run: &str, node: &str) -> crate::ask::Question {
1161        let mut q = crate::ask::Question::new(
1162            run.to_owned(),
1163            node.to_owned(),
1164            "conductor".to_owned(),
1165            "cannot resume".to_owned(),
1166            String::new(),
1167            Vec::new(),
1168        );
1169        crate::ask::Questions::at(home.join("questions"))
1170            .put(&mut q)
1171            .unwrap();
1172        q
1173    }
1174
1175    fn unread(home: &Path) -> usize {
1176        Notices::at(home.join("notifications")).count_unread()
1177    }
1178
1179    #[test]
1180    fn a_hold_with_an_open_question_for_the_task_pages_once() {
1181        let dir = tempfile::tempdir().unwrap();
1182        let mut t = held_task();
1183        let q = ask(dir.path(), &t.id, "conduct");
1184        crate::queue::Queue::at(dir.path().join("queue"))
1185            .put(&mut t)
1186            .unwrap();
1187        let s = Notices::at(dir.path().join("notifications"));
1188        assert_eq!(s.count_unread(), 0);
1189        let n = s.list().pop().expect("the record is kept");
1190        assert_eq!(n.covered_by.as_deref(), Some(q.id.as_str()));
1191        assert!(n.message.contains("checked out"));
1192    }
1193
1194    #[test]
1195    fn a_refused_handover_without_a_question_pages_once() {
1196        let dir = tempfile::tempdir().unwrap();
1197        let q = crate::queue::Queue::at(dir.path().join("queue"));
1198        let mut t = held_task();
1199        t.hold_for_handover(Some("magi/x/A".to_owned()), "checked out".to_owned());
1200        q.put(&mut t).unwrap();
1201        q.put(&mut t).unwrap();
1202        let s = Notices::at(dir.path().join("notifications"));
1203        assert_eq!(s.list().len(), 1);
1204        assert_eq!(s.list()[0].key, format!("task:{}", t.id));
1205        assert_eq!(s.count_unread(), 1);
1206    }
1207
1208    #[test]
1209    fn an_older_question_does_not_silence_a_new_hold() {
1210        for node in ["deps", "conduct", "triage"] {
1211            let dir = tempfile::tempdir().unwrap();
1212            let mut t = crate::queue::Task::new(
1213                "t".to_owned(),
1214                "do it".to_owned(),
1215                std::path::PathBuf::from("/repo"),
1216                crate::queue::Source::Human,
1217            );
1218            let old = ask(dir.path(), &t.id, node);
1219            std::thread::sleep(std::time::Duration::from_millis(5));
1220            t.hold_machine(Some("new failure".to_owned()));
1221            crate::queue::Queue::at(dir.path().join("queue"))
1222                .put(&mut t)
1223                .unwrap();
1224            let s = Notices::at(dir.path().join("notifications"));
1225            let n = s.list().pop().unwrap();
1226            assert_eq!(n.covered_by, None, "{node} question {}", old.id);
1227            assert_eq!(s.count_unread(), 1);
1228        }
1229    }
1230
1231    #[test]
1232    fn a_legacy_covered_notice_pages_for_a_new_uncovered_hold() {
1233        let (_d, s) = store();
1234        let mut legacy = Notice::warn("task:x", "held");
1235        legacy.covered_by = Some("q1".to_owned());
1236        let id = legacy.id.clone();
1237        s.raise(legacy).unwrap();
1238        assert!(!s.get(&id).unwrap().unread());
1239        s.raise(Notice::warn("task:x", "held").since(Some(Timestamp::now())))
1240            .unwrap();
1241        let n = s.get(&id).unwrap();
1242        assert!(n.unread());
1243        assert_eq!(n.covered_by, None);
1244    }
1245
1246    #[test]
1247    fn a_newer_cause_relights_a_dismissed_notice() {
1248        let (_d, s) = store();
1249        let t0 = Timestamp::now();
1250        let first = Notice::warn("task:x", "held").since(Some(t0));
1251        let id = first.id.clone();
1252        s.raise(first).unwrap();
1253        s.dismiss(&id).unwrap();
1254        s.raise(Notice::warn("task:x", "held").since(Some(t0)))
1255            .unwrap();
1256        assert!(!s.get(&id).unwrap().unread(), "same cause stays dismissed");
1257        let later = t0 + std::time::Duration::from_secs(1);
1258        s.raise(Notice::warn("task:x", "held").since(Some(later)))
1259            .unwrap();
1260        assert!(s.get(&id).unwrap().unread());
1261    }
1262
1263    #[test]
1264    fn a_hold_with_no_question_still_notifies() {
1265        let dir = tempfile::tempdir().unwrap();
1266        let mut t = held_task();
1267        crate::queue::Queue::at(dir.path().join("queue"))
1268            .put(&mut t)
1269            .unwrap();
1270        assert_eq!(unread(dir.path()), 1);
1271    }
1272
1273    #[test]
1274    fn a_question_for_another_task_does_not_suppress() {
1275        let dir = tempfile::tempdir().unwrap();
1276        let mut t = held_task();
1277        ask(dir.path(), "some-other-task", "conduct");
1278        crate::queue::Queue::at(dir.path().join("queue"))
1279            .put(&mut t)
1280            .unwrap();
1281        assert_eq!(unread(dir.path()), 1);
1282    }
1283
1284    #[test]
1285    fn a_question_filed_after_the_hold_quiets_it_once() {
1286        let dir = tempfile::tempdir().unwrap();
1287        let mut t = held_task();
1288        crate::queue::Queue::at(dir.path().join("queue"))
1289            .put(&mut t)
1290            .unwrap();
1291        assert_eq!(unread(dir.path()), 1);
1292        let q = ask(dir.path(), &t.id, "conduct");
1293        assert_eq!(unread(dir.path()), 0);
1294        // A later update of the same question does not run the hook again.
1295        let s = Notices::at(dir.path().join("notifications"));
1296        let id = s.list()[0].id.clone();
1297        s.update(&id, |n| n.read_at = None).unwrap();
1298        let mut again = crate::ask::Questions::at(dir.path().join("questions"))
1299            .get(&q.id)
1300            .unwrap();
1301        crate::ask::Questions::at(dir.path().join("questions"))
1302            .put(&mut again)
1303            .unwrap();
1304        assert_eq!(unread(dir.path()), 1);
1305    }
1306
1307    #[test]
1308    fn a_blocked_run_with_an_open_question_is_quiet() {
1309        let dir = tempfile::tempdir().unwrap();
1310        ask(dir.path(), "run-1", "implement");
1311        raise_in(
1312            dir.path(),
1313            Notice::error("run:run-1", "Run r ended blocked.").about(["run-1"]),
1314        );
1315        assert_eq!(unread(dir.path()), 0);
1316        raise_in(
1317            dir.path(),
1318            Notice::error("run:run-2", "Run r ended blocked.").about(["run-2"]),
1319        );
1320        assert_eq!(unread(dir.path()), 1);
1321    }
1322
1323    #[test]
1324    fn escalation_makes_a_covered_notice_unread_again() {
1325        let dir = tempfile::tempdir().unwrap();
1326        ask(dir.path(), "t1", "conduct");
1327        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1328        assert_eq!(unread(dir.path()), 0);
1329        raise_in(dir.path(), Notice::error("task:t1", "held").about(["t1"]));
1330        assert_eq!(unread(dir.path()), 1);
1331    }
1332
1333    #[test]
1334    fn covers_matches_run_exactly_and_ignores_release_questions() {
1335        let dir = tempfile::tempdir().unwrap();
1336        let n = Notice::warn("task:x", "m").about(["t1"]);
1337        let q = ask(dir.path(), "t1", "conduct");
1338        assert!(covers(&q, &n));
1339        assert!(!covers(&q, &Notice::warn("task:x", "m")));
1340        assert!(!covers(&q, &Notice::warn("task:x", "m").about(["t"])));
1341        let release = ask(dir.path(), "t1", crate::bump::NOTICE_NODE);
1342        assert!(!covers(&release, &n));
1343    }
1344
1345    #[test]
1346    fn a_run_question_does_not_swallow_a_task_hold_and_vice_versa() {
1347        let dir = tempfile::tempdir().unwrap();
1348        ask(dir.path(), "t1", "implement");
1349        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1350        assert_eq!(unread(dir.path()), 1);
1351        ask(dir.path(), "r1", "conduct");
1352        raise_in(dir.path(), Notice::error("run:r1", "ended").about(["r1"]));
1353        assert_eq!(unread(dir.path()), 2);
1354    }
1355
1356    #[test]
1357    fn a_conduct_question_covers_the_hold_however_late_it_was_filed() {
1358        let dir = tempfile::tempdir().unwrap();
1359        let mut q = crate::ask::Question::new(
1360            "t1".to_owned(),
1361            "conduct".to_owned(),
1362            "conductor".to_owned(),
1363            "s".to_owned(),
1364            String::new(),
1365            Vec::new(),
1366        );
1367        q.asked_at = Timestamp::from_second(Timestamp::now().as_second() - 3600).unwrap();
1368        crate::ask::Questions::at(dir.path().join("questions"))
1369            .put(&mut q)
1370            .unwrap();
1371        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1372        assert_eq!(unread(dir.path()), 0);
1373    }
1374
1375    fn pages(home: &Path, n: Notice) -> Vec<Page> {
1376        let sent = std::cell::RefCell::new(Vec::new());
1377        raise_in_with(home, n, &|p| sent.borrow_mut().push(p));
1378        sent.into_inner()
1379    }
1380
1381    #[test]
1382    fn a_page_fires_on_new_changed_and_escalated_only() {
1383        let dir = tempfile::tempdir().unwrap();
1384        let h = dir.path();
1385        let n = || Notice::warn("run:p1", "stopped").link(Link::Run { id: "p1".into() });
1386        let first = pages(h, n());
1387        assert_eq!(
1388            first,
1389            vec![Page {
1390                run: "p1".into(),
1391                summary: "stopped".into()
1392            }]
1393        );
1394        assert!(pages(h, n()).is_empty(), "identical re-raise is silent");
1395        assert_eq!(pages(h, Notice::warn("run:p1", "stopped again")).len(), 1);
1396        assert_eq!(pages(h, Notice::error("run:p1", "stopped again")).len(), 1);
1397        assert!(pages(h, Notice::error("run:p1", "stopped again")).is_empty());
1398    }
1399
1400    #[test]
1401    fn a_non_run_link_leaves_run_empty() {
1402        let dir = tempfile::tempdir().unwrap();
1403        let n = Notice::warn("task:t1", "held").link(Link::Task { id: "t1".into() });
1404        assert_eq!(pages(dir.path(), n)[0].run, "");
1405    }
1406
1407    #[test]
1408    fn a_dismissed_tombstone_stays_silent_until_it_changes() {
1409        let dir = tempfile::tempdir().unwrap();
1410        let h = dir.path();
1411        pages(h, Notice::warn("disk:r", "low"));
1412        Notices::at(h.join("notifications"))
1413            .dismiss(&id_of("disk:r"))
1414            .unwrap();
1415        assert!(pages(h, Notice::warn("disk:r", "low")).is_empty());
1416        assert_eq!(pages(h, Notice::warn("disk:r", "lower")).len(), 1);
1417    }
1418
1419    #[test]
1420    fn a_newer_cause_pages_again() {
1421        let dir = tempfile::tempdir().unwrap();
1422        let h = dir.path();
1423        let at = |s: i64| Some(Timestamp::from_second(s).unwrap());
1424        let n = |s| Notice::warn("task:t2", "held").since(at(s));
1425        assert_eq!(pages(h, n(100)).len(), 1);
1426        assert!(pages(h, n(100)).is_empty());
1427        assert_eq!(pages(h, n(200)).len(), 1);
1428    }
1429
1430    #[test]
1431    fn a_covered_notice_does_not_page() {
1432        let dir = tempfile::tempdir().unwrap();
1433        let h = dir.path();
1434        let q = ask(h, "r9", "implement");
1435        let n = || Notice::error("run:r9", "ended").about(["r9".to_owned()]);
1436        assert!(pages(h, n()).is_empty());
1437        let stored = Notices::at(h.join("notifications"))
1438            .get(&id_of("run:r9"))
1439            .unwrap();
1440        assert_eq!(stored.covered_by.as_deref(), Some(q.id.as_str()));
1441        assert!(pages(h, n()).is_empty());
1442    }
1443
1444    #[test]
1445    fn a_failing_sender_path_never_propagates() {
1446        // A pager that cannot run its command only warns.
1447        let notify = crate::config::Notify {
1448            command: vec!["magi-no-such-program-xyz".into()],
1449        };
1450        let rt = tokio::runtime::Builder::new_current_thread()
1451            .enable_all()
1452            .build()
1453            .unwrap();
1454        let r = rt.block_on(crate::ask::notify_text(&notify, "", "x"));
1455        assert!(r.is_err());
1456        // `send_page` without an installed pager is a no-op and cannot panic.
1457        send_page(
1458            Page {
1459                run: String::new(),
1460                summary: "x".into(),
1461            },
1462            Some(notify),
1463        );
1464    }
1465}