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/// Human-readable process status for operator-facing failure details.
639pub fn exit_status(code: Option<i32>) -> String {
640    match code {
641        Some(code) => format!("exit code {code}"),
642        None => "exit code unavailable".to_owned(),
643    }
644}
645
646/// Trims text and drops empty values or meaningless placeholder phrases like `"(no reason given)"`.
647pub fn clean_text(s: Option<&str>) -> Option<&str> {
648    s.map(str::trim)
649        .filter(|t| !t.is_empty() && *t != "(no reason given)" && *t != "(none)")
650}
651
652/// Formats a list of identifiers or items into human-readable comma-separated prose.
653/// Returns `None` if the list is empty or contains only blank entries.
654pub fn format_list<T: AsRef<str>>(items: &[T]) -> Option<String> {
655    let non_empty: Vec<&str> = items
656        .iter()
657        .map(|s| s.as_ref().trim())
658        .filter(|s| !s.is_empty())
659        .collect();
660    if non_empty.is_empty() {
661        None
662    } else {
663        Some(non_empty.join(", "))
664    }
665}
666
667/// Ensures a sentence ends with a single period and has no trailing whitespace.
668pub fn ensure_sentence(text: &str) -> String {
669    let trimmed = text.trim();
670    if trimmed.is_empty() {
671        return String::new();
672    }
673    let trimmed_dot = trimmed.trim_end_matches('.');
674    format!("{trimmed_dot}.")
675}
676
677/// The notice for a task the machine or its attempt budget has held, if it is.
678///
679/// Called from [`crate::queue::Queue::put`], which every task transition goes
680/// through, so a hold made anywhere - the loop, the conductor, triage, a
681/// dependency removed from under a blocked task - is announced. A hold the
682/// operator placed by hand is their own action and is not news. Keyed on the
683/// task, with wording free of anything that varies between retries.
684///
685/// Falls back to [`crate::queue::Task::last_error`] when `hold_reason` is
686/// empty, the same fallback `crate::triage`'s question detail uses: a record
687/// written before `Task::fail`/`Task::handed_off` started copying `why` into
688/// `hold_reason` too would otherwise still read "no reason recorded" even
689/// though the run said exactly why it stopped.
690pub fn task_held(task: &crate::queue::Task) -> Option<Notice> {
691    use crate::queue::{HoldSource, TaskStatus};
692    if task.status != TaskStatus::Held || task.hold_source != Some(HoldSource::Machine) {
693        return None;
694    }
695    let why = task
696        .hold_reason
697        .as_deref()
698        .or(task.last_error.as_deref())
699        .and_then(|w| clean_text(Some(w)))
700        .unwrap_or("no reason recorded");
701    let why_sentence = why.trim_end_matches('.');
702    Some(
703        Notice::warn(
704            &format!("task:{}", task.id),
705            format!("Task {} is held: {why_sentence}.", task.short()),
706        )
707        .link(Link::Task {
708            id: task.id.clone(),
709        })
710        .about([task.id.clone()])
711        .since(task.held_at),
712    )
713}
714
715/// The notice for a conductor concern raised after explicit resumption without
716/// verified new blocking evidence.
717///
718/// Informational: preserves the task state without re-blocking on old evidence,
719/// explains that no state changed and no reply is required, and attaches a link
720/// to the task. Any proposed question text is omitted to avoid soliciting an
721/// answer to an unregistered question.
722pub fn conductor_resumption(
723    task_id: &str,
724    task_short: &str,
725    reason: Option<&str>,
726    blocked_by: &[String],
727    evidence: &[String],
728    has_question: bool,
729) -> Notice {
730    let mut parts: Vec<String> = Vec::new();
731    parts.push(format!(
732        "Conductor raised a concern for task {task_short} after explicit resumption, but its state was not changed because no verified new blocking evidence was found."
733    ));
734
735    if let Some(r) = clean_text(reason) {
736        parts.push(format!("Concern: {}.", r.trim_end_matches('.')));
737    }
738
739    if has_question {
740        parts.push("An unregistered question was proposed; no reply is required.".to_owned());
741    }
742
743    if let Some(deps) = format_list(blocked_by) {
744        parts.push(format!("Proposed dependencies: {deps}."));
745    }
746
747    if let Some(ev) = format_list(evidence) {
748        parts.push(format!("Cited evidence: {ev}."));
749    }
750
751    parts.push(
752        "No reply to this notice is needed. To hold this task, use manual hold via the queue."
753            .to_owned(),
754    );
755
756    Notice::warn(&format!("conductor-resumption:{task_id}"), parts.join(" "))
757        .link(Link::Task {
758            id: task_id.to_owned(),
759        })
760        .about([task_id.to_owned()])
761}
762
763/// The notice for a run whose graph returned an error before it settled.
764pub fn run_stopped(id: &str, state: &crate::run::RunState) -> Notice {
765    Notice::error(
766        &format!("run:{id}"),
767        format!("Run {} stopped with an error.", state.short()),
768    )
769    .link(Link::Run { id: id.to_owned() })
770    .about([id.to_owned()])
771}
772
773/// One page for the operator's `[notify]` command.
774#[derive(Debug, Clone, PartialEq, Eq)]
775pub struct Page {
776    /// The run id when the notice links to a run, else empty.
777    pub run: String,
778    /// The notice message.
779    pub summary: String,
780}
781
782/// Set once by `main`: until then no notice fires the command, so a unit or
783/// integration test that pins a home never reaches the operator's real config.
784static PAGER: std::sync::OnceLock<()> = std::sync::OnceLock::new();
785
786/// Let notices fire the `[notify]` command in this process.
787pub fn install_pager() {
788    let _ = PAGER.set(());
789}
790
791/// The production sender. Fire and forget on its own thread with its own
792/// runtime, so it works with or without an ambient one and never blocks the
793/// producer; `ask::notify_text` bounds the command by `NOTIFY_TIMEOUT`.
794/// `notify` is the producer's own config, else the machine layer's.
795fn send_page(page: Page, notify: Option<crate::config::Notify>) {
796    if PAGER.get().is_none() {
797        return;
798    }
799    let spawned = std::thread::Builder::new()
800        .name("notice-pager".into())
801        .spawn(move || {
802            let notify = notify.unwrap_or_else(machine_notify);
803            if notify.command.is_empty() {
804                return;
805            }
806            let rt = match tokio::runtime::Builder::new_current_thread()
807                .enable_all()
808                .build()
809            {
810                Ok(rt) => rt,
811                Err(e) => return tracing::warn!("could not page about a notification: {e:#}"),
812            };
813            if let Err(e) = rt.block_on(crate::ask::notify_text(&notify, &page.run, &page.summary))
814            {
815                tracing::warn!("could not page about a notification: {e:#}");
816            }
817        });
818    if let Err(e) = spawned {
819        tracing::warn!("could not page about a notification: {e:#}");
820    }
821}
822
823/// `[notify]` from the machine layer alone; a producer with no repo config
824/// cannot name a better one. Unreadable is no command, with a warning.
825fn machine_notify() -> crate::config::Notify {
826    let layers: Vec<PathBuf> = crate::config::Config::machine_layer()
827        .into_iter()
828        .filter(|p| p.is_file())
829        .collect();
830    match crate::config::Config::load_layers(&layers) {
831        Ok(c) => c.notify,
832        Err(e) => {
833            tracing::warn!("could not read [notify] for a notification: {e:#}");
834            Default::default()
835        }
836    }
837}
838
839/// The one function producers call. Best-effort by construction: a notice that
840/// cannot be filed is a `tracing::warn`, never a reason to fail the run, the
841/// loop or the request that wanted to mention it.
842pub fn raise(notice: Notice) {
843    if let Some(home) = crate::run::try_home() {
844        raise_in(&home, notice);
845    }
846}
847
848/// [`raise`] for a producer that already holds the repo's `[notify]`.
849pub fn raise_with(notice: Notice, notify: &crate::config::Notify) {
850    if let Some(home) = crate::run::try_home() {
851        raise_in_with(&home, notice, &|page| send_page(page, Some(notify.clone())));
852    }
853}
854
855/// Does open question `q` already page the operator for what `n` reports?
856///
857/// The cause is decided by what each side is about, not by when it happened:
858/// a conductor / triage question exists *because* its task is held, so it
859/// covers that task's hold and handover notices whenever it was filed; a
860/// question raised from inside a run (any other seat node) covers that run's
861/// ended / stopped notices and never a task hold. Exact match on
862/// `Question::run` (a task id for the former, a run id for the latter). The
863/// land approval and release questions are to-dos, not duplicates, and cover
864/// nothing.
865///
866/// A notice with a `since` (a task hold's start) is covered only by a question
867/// filed at or after it: one filed earlier is about an older cause, and
868/// letting it silence a new hold would hide a fresh failure behind a stale
869/// card. Missing a duplicate costs one extra page; hiding a failure costs the
870/// failure. A notice without `since` keeps the broad rule.
871pub fn covers(q: &crate::ask::Question, n: &Notice) -> bool {
872    if !q.status.open() || !n.subjects.contains(&q.run) {
873        return false;
874    }
875    if n.since.is_some_and(|since| q.asked_at < since) {
876        return false;
877    }
878    let about_task = n.key.starts_with("task:") || n.key.starts_with("handover:");
879    let about_run = n.key.starts_with("run:");
880    match q.node.as_str() {
881        crate::bump::NOTICE_NODE | crate::land::APPROVAL_NODE => false,
882        _ if q.run_names_task() => about_task,
883        _ => about_run,
884    }
885}
886
887fn covering_question(home: &Path, n: &Notice) -> Option<String> {
888    if n.subjects.is_empty() {
889        return None;
890    }
891    crate::ask::Questions::at(home.join("questions"))
892        .list()
893        .into_iter()
894        .find(|q| covers(q, n))
895        .map(|q| q.id)
896}
897
898/// A question was just filed for `run`: quiet the unread notices about it.
899pub fn quiet_for(home: &Path, question: &crate::ask::Question) {
900    Notices::at(home.join("notifications")).cover(question);
901}
902
903/// [`raise`] into an explicit magi home, for callers that already carry one
904/// (the janitor) and so must not reach for the process-global.
905pub fn raise_in(home: &Path, notice: Notice) {
906    raise_in_with(home, notice, &|page| send_page(page, None));
907}
908
909/// [`raise_in`] with the pager injected. `send` is called only when the raise
910/// should page (see [`Notice::raise_again`]); it must not block.
911pub fn raise_in_with(home: &Path, mut notice: Notice, send: &dyn Fn(Page)) {
912    if let Some(q) = covering_question(home, &notice) {
913        notice.covered_by = Some(q);
914    }
915    match Notices::at(home.join("notifications")).raise_paged(notice) {
916        Ok((n, true)) => send(Page {
917            run: match &n.link {
918                Some(Link::Run { id }) => id.clone(),
919                _ => String::new(),
920            },
921            summary: n.message.clone(),
922        }),
923        Ok(_) => {}
924        Err(e) => tracing::warn!("could not file a notification: {e:#}"),
925    }
926}
927
928#[cfg(test)]
929mod tests {
930    use super::*;
931
932    #[test]
933    fn legacy_deputy_warnings_are_hidden_without_hiding_other_namespaces() {
934        let (_dir, s) = store();
935        // No question lookup: these may belong to deleted or settled questions.
936        for key in ["deputy:open", "deputy:answered", "deputy:deleted"] {
937            s.raise(Notice::warn(key, "old exhausted warning")).unwrap();
938        }
939        let keys = ["deputy-turn:open", "deputyish:open", "deputy", "task:open"];
940        for key in keys {
941            s.raise(Notice::warn(key, "still relevant")).unwrap();
942        }
943        assert_eq!(s.list().len(), keys.len());
944        assert_eq!(s.count_unread(), keys.len());
945        assert!(s.list().iter().all(|n| keys.contains(&n.key.as_str())));
946        assert_eq!(s.mark_all_read().unwrap(), keys.len());
947        assert_eq!(s.count_unread(), 0);
948        assert!(s.get(&id_of("deputy:open")).unwrap().unread());
949    }
950
951    #[test]
952    fn obsolete_deputy_warnings_are_pruned_before_current_notices() {
953        let (_dir, s) = store();
954        for i in 0..CAP {
955            let id = s
956                .raise(Notice::info(&format!("current:{i}"), "keep"))
957                .unwrap()
958                .id;
959            s.dismiss(&id).unwrap();
960        }
961        // Even a newer unread legacy warning cannot evict a current tombstone.
962        s.raise(Notice::warn("deputy:old", "obsolete")).unwrap();
963        assert!(s.get(&id_of("deputy:old")).is_err());
964        assert_eq!(s.all().len(), CAP);
965        assert!(s.all().iter().all(|n| n.key.starts_with("current:")));
966    }
967
968    #[test]
969    fn process_exit_status_is_readable_even_without_a_code() {
970        assert_eq!(exit_status(Some(1)), "exit code 1");
971        assert_eq!(exit_status(None), "exit code unavailable");
972    }
973
974    fn store() -> (tempfile::TempDir, Notices) {
975        let dir = tempfile::tempdir().unwrap();
976        let s = Notices::at(dir.path().join("notifications"));
977        (dir, s)
978    }
979
980    #[test]
981    fn a_red_merge_notice_is_keyed_on_the_run_and_raised_once() {
982        let (_d, s) = store();
983        let n = merged_red("run-1", "Merged a/b PR #1 with red checks: x (u)");
984        assert_eq!(n.key, "merged-red:run-1");
985        assert_eq!(
986            n.link,
987            Some(Link::Run {
988                id: "run-1".to_owned()
989            })
990        );
991        s.raise(merged_red(
992            "run-1",
993            "Merged a/b PR #1 with red checks: x (u)",
994        ))
995        .unwrap();
996        s.raise(merged_red(
997            "run-1",
998            "Merged a/b PR #1 with red checks: x (u)",
999        ))
1000        .unwrap();
1001        assert_eq!(s.list().len(), 1);
1002    }
1003
1004    #[test]
1005    fn a_repeat_counts_but_stays_read_and_a_change_relights_it() {
1006        let now = Timestamp::now();
1007        let mut n = Notice::warn("task:1", "held");
1008        n.mark_read(now);
1009        n.raise_again(&Notice::warn("task:1", "held"), now);
1010        assert_eq!(n.count, 2);
1011        assert!(n.read_at.is_some(), "identical repeat stays read");
1012        n.raise_again(&Notice::warn("task:1", "held differently"), now);
1013        assert!(n.unread());
1014        assert_eq!(n.message, "held differently");
1015    }
1016
1017    #[test]
1018    fn a_rise_in_severity_resurrects_a_dismissed_notice_but_a_repeat_does_not() {
1019        let now = Timestamp::now();
1020        let mut n = Notice::warn("task:x", "m");
1021        n.dismiss(now);
1022        n.raise_again(&Notice::warn("task:x", "m"), now);
1023        assert!(n.dismissed_at.is_some(), "tombstone holds");
1024        n.raise_again(&Notice::error("k", "m"), now);
1025        assert!(n.unread());
1026        assert_eq!(n.severity, Severity::Error);
1027        // Severity never falls on a repeat.
1028        n.raise_again(&Notice::info("k", "other"), now);
1029        assert_eq!(n.severity, Severity::Error);
1030    }
1031
1032    #[test]
1033    fn adding_link_on_repeat_updates_link_without_relighting_read_or_dismissed() {
1034        let now = Timestamp::now();
1035        // 1. Read notice
1036        let mut read_notice = Notice::warn("task:1", "held");
1037        read_notice.mark_read(now);
1038        let link_to_add = Link::Task { id: "1".to_owned() };
1039        let again = Notice::warn("task:1", "held").link(link_to_add.clone());
1040        let relit = read_notice.raise_again(&again, now);
1041        assert!(!relit, "link-only update must not page/relight");
1042        assert!(read_notice.read_at.is_some(), "read notice stays read");
1043        assert_eq!(read_notice.link, Some(link_to_add.clone()));
1044
1045        // 2. Dismissed notice
1046        let mut dismissed_notice = Notice::warn("task:2", "held");
1047        dismissed_notice.dismiss(now);
1048        let link_to_add_2 = Link::Task { id: "2".to_owned() };
1049        let again_2 = Notice::warn("task:2", "held").link(link_to_add_2.clone());
1050        let relit_2 = dismissed_notice.raise_again(&again_2, now);
1051        assert!(!relit_2, "link-only update must not page/relight");
1052        assert!(dismissed_notice.dismissed_at.is_some(), "tombstone holds");
1053        assert_eq!(dismissed_notice.link, Some(link_to_add_2));
1054    }
1055
1056    #[test]
1057    fn identical_raises_share_one_file() {
1058        let (_d, s) = store();
1059        for _ in 0..5 {
1060            s.raise(Notice::error("run:abc", "blocked")).unwrap();
1061        }
1062        let all = s.list();
1063        assert_eq!(all.len(), 1);
1064        assert_eq!(all[0].count, 5);
1065        assert_eq!(s.count_unread(), 1);
1066    }
1067
1068    #[test]
1069    fn transitions_persist_and_dismissed_leave_the_list() {
1070        let (_d, s) = store();
1071        let a = s.raise(Notice::info("a", "one")).unwrap();
1072        let b = s.raise(Notice::warn("b", "two")).unwrap();
1073        assert_eq!(s.count_unread(), 2);
1074        s.mark_read(&a.id).unwrap();
1075        assert_eq!(s.count_unread(), 1);
1076        s.dismiss(&b.id).unwrap();
1077        assert_eq!(s.count_unread(), 0);
1078        assert_eq!(s.list().len(), 1);
1079        s.raise(Notice::warn("b", "two")).unwrap();
1080        assert_eq!(
1081            s.list().len(),
1082            1,
1083            "a tombstone survives a same-message raise"
1084        );
1085        s.raise(Notice::info("c", "three")).unwrap();
1086        assert_eq!(s.mark_all_read().unwrap(), 1);
1087        assert_eq!(s.count_unread(), 0);
1088    }
1089
1090    #[test]
1091    fn the_list_is_newest_first() {
1092        let (_d, s) = store();
1093        let mut old = Notice::info("old", "old");
1094        old.last_at = "2020-01-01T00:00:00Z".parse().unwrap();
1095        s.put(&old).unwrap();
1096        s.raise(Notice::info("new", "new")).unwrap();
1097        let keys: Vec<_> = s.list().into_iter().map(|n| n.key).collect();
1098        assert_eq!(keys, ["new", "old"]);
1099    }
1100
1101    #[test]
1102    fn the_cap_prunes_dismissed_then_read_then_oldest() {
1103        let (_d, s) = store();
1104        let keep = s.raise(Notice::error("keep", "unread")).unwrap();
1105        let gone = s.raise(Notice::info("gone", "dismissed")).unwrap();
1106        s.dismiss(&gone.id).unwrap();
1107        for i in 0..CAP - 1 {
1108            s.raise(Notice::info(&format!("k{i}"), "x")).unwrap();
1109        }
1110        let files = std::fs::read_dir(s.root()).unwrap().count();
1111        assert_eq!(files, CAP);
1112        assert!(
1113            s.get(&keep.id).is_ok(),
1114            "an unread notice outlives a dismissed one"
1115        );
1116        assert!(s.get(&gone.id).is_err(), "the dismissed one went first");
1117    }
1118
1119    #[test]
1120    fn revision_moves_on_write_and_on_removal() {
1121        let (_d, s) = store();
1122        assert_eq!(s.revision(), 0);
1123        let a = s.raise(Notice::info("a", "m")).unwrap();
1124        let r1 = s.revision();
1125        assert_ne!(r1, 0);
1126        s.raise(Notice::info("b", "m")).unwrap();
1127        let r2 = s.revision();
1128        assert_ne!(r1, r2);
1129        std::fs::remove_file(s.path_of(&a.id)).unwrap();
1130        assert_ne!(s.revision(), r2);
1131    }
1132
1133    #[test]
1134    fn ids_are_stable_and_untrusted_ids_are_refused() {
1135        assert_eq!(id_of("run:1"), id_of("run:1"));
1136        assert_ne!(id_of("run:1"), id_of("run-1"));
1137        assert!(valid_id(&id_of("release-bump:20260101-abc")));
1138        let (_d, s) = store();
1139        for bad in ["", "../x", "a/b", "A", "x.json"] {
1140            assert!(s.get(bad).is_err(), "{bad}");
1141        }
1142    }
1143
1144    #[test]
1145    fn a_newer_schema_is_refused_and_an_older_reads() {
1146        let (_d, s) = store();
1147        let mut n = Notice::info("k", "m");
1148        n.schema = SCHEMA + 1;
1149        s.put(&n).unwrap();
1150        assert!(s.get(&n.id).is_err());
1151        let old = r#"{"id":"x-1","key":"x","severity":"warn","message":"m",
1152            "first_at":"2020-01-01T00:00:00Z","last_at":"2020-01-01T00:00:00Z"}"#;
1153        std::fs::write(s.path_of("x-1"), old).unwrap();
1154        assert_eq!(s.get("x-1").unwrap().count, 1);
1155    }
1156
1157    fn state(status: crate::run::RunStatus) -> crate::run::RunState {
1158        let mut st = crate::run::RunState::new(
1159            std::path::PathBuf::from("/repo"),
1160            "main".to_owned(),
1161            "abc1234def".to_owned(),
1162            "task".to_owned(),
1163            crate::config::Config::default(),
1164        );
1165        st.status = status;
1166        st
1167    }
1168
1169    #[test]
1170    fn a_run_that_ended_badly_is_news_unless_it_only_parked() {
1171        use crate::run::RunStatus;
1172        for bad in [RunStatus::Blocked, RunStatus::Stalled, RunStatus::Failed] {
1173            let st = state(bad);
1174            let n = run_ended(&st).expect("news");
1175            assert_eq!(n.key, format!("run:{}", st.id));
1176            assert_eq!(n.severity, Severity::Error);
1177        }
1178        assert!(run_ended(&state(RunStatus::Merged)).is_none());
1179        let mut parked = state(RunStatus::Stalled);
1180        parked.parked = true;
1181        assert!(run_ended(&parked).is_none());
1182    }
1183
1184    #[test]
1185    fn concurrent_writers_of_one_key_all_succeed() {
1186        let (_d, s) = store();
1187        let handles: Vec<_> = (0..8)
1188            .map(|_| {
1189                let s = s.clone();
1190                std::thread::spawn(move || {
1191                    for _ in 0..20 {
1192                        s.raise(Notice::warn("same", "m")).unwrap();
1193                    }
1194                })
1195            })
1196            .collect();
1197        for h in handles {
1198            h.join().unwrap();
1199        }
1200        assert_eq!(s.list().len(), 1);
1201        let stray = std::fs::read_dir(s.root())
1202            .unwrap()
1203            .flatten()
1204            .filter(|e| e.path().extension().is_some_and(|x| x == "tmp"))
1205            .count();
1206        assert_eq!(stray, 0);
1207    }
1208
1209    #[test]
1210    fn a_dead_holders_lock_is_broken_and_no_lock_is_left_behind() {
1211        let (_d, s) = store();
1212        let n = s.raise(Notice::info("k", "m")).unwrap();
1213        let lock = s.root().join(format!("{}.lock", n.id));
1214        std::fs::write(&lock, "").unwrap();
1215        let old = std::time::SystemTime::now() - std::time::Duration::from_secs(60);
1216        std::fs::File::options()
1217            .write(true)
1218            .open(&lock)
1219            .unwrap()
1220            .set_modified(old)
1221            .unwrap();
1222        s.mark_read(&n.id).unwrap();
1223        assert!(!lock.exists());
1224        assert!(s.get(&n.id).unwrap().read_at.is_some());
1225    }
1226
1227    #[test]
1228    fn only_a_machine_hold_is_a_task_notice() {
1229        let mut t = crate::queue::Task::new(
1230            "t".to_owned(),
1231            "do it".to_owned(),
1232            std::path::PathBuf::from("/repo"),
1233            crate::queue::Source::Human,
1234        );
1235        assert!(task_held(&t).is_none());
1236        t.hold_manual(None);
1237        assert!(task_held(&t).is_none(), "the operator's own hold");
1238        t.hold_machine(Some("missing blocker".to_owned()));
1239        let n = task_held(&t).expect("machine hold");
1240        assert_eq!(n.key, format!("task:{}", t.id));
1241        assert!(n.message.contains("missing blocker"));
1242    }
1243
1244    fn held_task() -> crate::queue::Task {
1245        let mut t = crate::queue::Task::new(
1246            "t".to_owned(),
1247            "do it".to_owned(),
1248            std::path::PathBuf::from("/repo"),
1249            crate::queue::Source::Human,
1250        );
1251        t.hold_machine(Some("branch b is checked out".to_owned()));
1252        t
1253    }
1254
1255    fn ask(home: &Path, run: &str, node: &str) -> crate::ask::Question {
1256        let mut q = crate::ask::Question::new(
1257            run.to_owned(),
1258            node.to_owned(),
1259            "conductor".to_owned(),
1260            "cannot resume".to_owned(),
1261            String::new(),
1262            Vec::new(),
1263        );
1264        crate::ask::Questions::at(home.join("questions"))
1265            .put(&mut q)
1266            .unwrap();
1267        q
1268    }
1269
1270    fn unread(home: &Path) -> usize {
1271        Notices::at(home.join("notifications")).count_unread()
1272    }
1273
1274    #[test]
1275    fn a_hold_with_an_open_question_for_the_task_pages_once() {
1276        let dir = tempfile::tempdir().unwrap();
1277        let mut t = held_task();
1278        let q = ask(dir.path(), &t.id, "conduct");
1279        crate::queue::Queue::at(dir.path().join("queue"))
1280            .put(&mut t)
1281            .unwrap();
1282        let s = Notices::at(dir.path().join("notifications"));
1283        assert_eq!(s.count_unread(), 0);
1284        let n = s.list().pop().expect("the record is kept");
1285        assert_eq!(n.covered_by.as_deref(), Some(q.id.as_str()));
1286        assert!(n.message.contains("checked out"));
1287    }
1288
1289    #[test]
1290    fn a_refused_handover_without_a_question_pages_once() {
1291        let dir = tempfile::tempdir().unwrap();
1292        let q = crate::queue::Queue::at(dir.path().join("queue"));
1293        let mut t = held_task();
1294        t.hold_for_handover(Some("magi/x/A".to_owned()), "checked out".to_owned());
1295        q.put(&mut t).unwrap();
1296        q.put(&mut t).unwrap();
1297        let s = Notices::at(dir.path().join("notifications"));
1298        assert_eq!(s.list().len(), 1);
1299        assert_eq!(s.list()[0].key, format!("task:{}", t.id));
1300        assert_eq!(s.count_unread(), 1);
1301    }
1302
1303    #[test]
1304    fn an_older_question_does_not_silence_a_new_hold() {
1305        for node in ["deps", "conduct", "triage"] {
1306            let dir = tempfile::tempdir().unwrap();
1307            let mut t = crate::queue::Task::new(
1308                "t".to_owned(),
1309                "do it".to_owned(),
1310                std::path::PathBuf::from("/repo"),
1311                crate::queue::Source::Human,
1312            );
1313            let old = ask(dir.path(), &t.id, node);
1314            std::thread::sleep(std::time::Duration::from_millis(5));
1315            t.hold_machine(Some("new failure".to_owned()));
1316            crate::queue::Queue::at(dir.path().join("queue"))
1317                .put(&mut t)
1318                .unwrap();
1319            let s = Notices::at(dir.path().join("notifications"));
1320            let n = s.list().pop().unwrap();
1321            assert_eq!(n.covered_by, None, "{node} question {}", old.id);
1322            assert_eq!(s.count_unread(), 1);
1323        }
1324    }
1325
1326    #[test]
1327    fn a_legacy_covered_notice_pages_for_a_new_uncovered_hold() {
1328        let (_d, s) = store();
1329        let mut legacy = Notice::warn("task:x", "held");
1330        legacy.covered_by = Some("q1".to_owned());
1331        let id = legacy.id.clone();
1332        s.raise(legacy).unwrap();
1333        assert!(!s.get(&id).unwrap().unread());
1334        s.raise(Notice::warn("task:x", "held").since(Some(Timestamp::now())))
1335            .unwrap();
1336        let n = s.get(&id).unwrap();
1337        assert!(n.unread());
1338        assert_eq!(n.covered_by, None);
1339    }
1340
1341    #[test]
1342    fn a_newer_cause_relights_a_dismissed_notice() {
1343        let (_d, s) = store();
1344        let t0 = Timestamp::now();
1345        let first = Notice::warn("task:x", "held").since(Some(t0));
1346        let id = first.id.clone();
1347        s.raise(first).unwrap();
1348        s.dismiss(&id).unwrap();
1349        s.raise(Notice::warn("task:x", "held").since(Some(t0)))
1350            .unwrap();
1351        assert!(!s.get(&id).unwrap().unread(), "same cause stays dismissed");
1352        let later = t0 + std::time::Duration::from_secs(1);
1353        s.raise(Notice::warn("task:x", "held").since(Some(later)))
1354            .unwrap();
1355        assert!(s.get(&id).unwrap().unread());
1356    }
1357
1358    #[test]
1359    fn a_hold_with_no_question_still_notifies() {
1360        let dir = tempfile::tempdir().unwrap();
1361        let mut t = held_task();
1362        crate::queue::Queue::at(dir.path().join("queue"))
1363            .put(&mut t)
1364            .unwrap();
1365        assert_eq!(unread(dir.path()), 1);
1366    }
1367
1368    #[test]
1369    fn a_question_for_another_task_does_not_suppress() {
1370        let dir = tempfile::tempdir().unwrap();
1371        let mut t = held_task();
1372        ask(dir.path(), "some-other-task", "conduct");
1373        crate::queue::Queue::at(dir.path().join("queue"))
1374            .put(&mut t)
1375            .unwrap();
1376        assert_eq!(unread(dir.path()), 1);
1377    }
1378
1379    #[test]
1380    fn a_question_filed_after_the_hold_quiets_it_once() {
1381        let dir = tempfile::tempdir().unwrap();
1382        let mut t = held_task();
1383        crate::queue::Queue::at(dir.path().join("queue"))
1384            .put(&mut t)
1385            .unwrap();
1386        assert_eq!(unread(dir.path()), 1);
1387        let q = ask(dir.path(), &t.id, "conduct");
1388        assert_eq!(unread(dir.path()), 0);
1389        // A later update of the same question does not run the hook again.
1390        let s = Notices::at(dir.path().join("notifications"));
1391        let id = s.list()[0].id.clone();
1392        s.update(&id, |n| n.read_at = None).unwrap();
1393        let mut again = crate::ask::Questions::at(dir.path().join("questions"))
1394            .get(&q.id)
1395            .unwrap();
1396        crate::ask::Questions::at(dir.path().join("questions"))
1397            .put(&mut again)
1398            .unwrap();
1399        assert_eq!(unread(dir.path()), 1);
1400    }
1401
1402    #[test]
1403    fn a_blocked_run_with_an_open_question_is_quiet() {
1404        let dir = tempfile::tempdir().unwrap();
1405        ask(dir.path(), "run-1", "implement");
1406        raise_in(
1407            dir.path(),
1408            Notice::error("run:run-1", "Run r ended blocked.").about(["run-1"]),
1409        );
1410        assert_eq!(unread(dir.path()), 0);
1411        raise_in(
1412            dir.path(),
1413            Notice::error("run:run-2", "Run r ended blocked.").about(["run-2"]),
1414        );
1415        assert_eq!(unread(dir.path()), 1);
1416    }
1417
1418    #[test]
1419    fn escalation_makes_a_covered_notice_unread_again() {
1420        let dir = tempfile::tempdir().unwrap();
1421        ask(dir.path(), "t1", "conduct");
1422        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1423        assert_eq!(unread(dir.path()), 0);
1424        raise_in(dir.path(), Notice::error("task:t1", "held").about(["t1"]));
1425        assert_eq!(unread(dir.path()), 1);
1426    }
1427
1428    #[test]
1429    fn covers_matches_run_exactly_and_ignores_release_questions() {
1430        let dir = tempfile::tempdir().unwrap();
1431        let n = Notice::warn("task:x", "m").about(["t1"]);
1432        let q = ask(dir.path(), "t1", "conduct");
1433        assert!(covers(&q, &n));
1434        assert!(!covers(&q, &Notice::warn("task:x", "m")));
1435        assert!(!covers(&q, &Notice::warn("task:x", "m").about(["t"])));
1436        let release = ask(dir.path(), "t1", crate::bump::NOTICE_NODE);
1437        assert!(!covers(&release, &n));
1438    }
1439
1440    #[test]
1441    fn a_run_question_does_not_swallow_a_task_hold_and_vice_versa() {
1442        let dir = tempfile::tempdir().unwrap();
1443        ask(dir.path(), "t1", "implement");
1444        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1445        assert_eq!(unread(dir.path()), 1);
1446        ask(dir.path(), "r1", "conduct");
1447        raise_in(dir.path(), Notice::error("run:r1", "ended").about(["r1"]));
1448        assert_eq!(unread(dir.path()), 2);
1449    }
1450
1451    #[test]
1452    fn a_conduct_question_covers_the_hold_however_late_it_was_filed() {
1453        let dir = tempfile::tempdir().unwrap();
1454        let mut q = crate::ask::Question::new(
1455            "t1".to_owned(),
1456            "conduct".to_owned(),
1457            "conductor".to_owned(),
1458            "s".to_owned(),
1459            String::new(),
1460            Vec::new(),
1461        );
1462        q.asked_at = Timestamp::from_second(Timestamp::now().as_second() - 3600).unwrap();
1463        crate::ask::Questions::at(dir.path().join("questions"))
1464            .put(&mut q)
1465            .unwrap();
1466        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1467        assert_eq!(unread(dir.path()), 0);
1468    }
1469
1470    fn pages(home: &Path, n: Notice) -> Vec<Page> {
1471        let sent = std::cell::RefCell::new(Vec::new());
1472        raise_in_with(home, n, &|p| sent.borrow_mut().push(p));
1473        sent.into_inner()
1474    }
1475
1476    #[test]
1477    fn a_page_fires_on_new_changed_and_escalated_only() {
1478        let dir = tempfile::tempdir().unwrap();
1479        let h = dir.path();
1480        let n = || Notice::warn("run:p1", "stopped").link(Link::Run { id: "p1".into() });
1481        let first = pages(h, n());
1482        assert_eq!(
1483            first,
1484            vec![Page {
1485                run: "p1".into(),
1486                summary: "stopped".into()
1487            }]
1488        );
1489        assert!(pages(h, n()).is_empty(), "identical re-raise is silent");
1490        assert_eq!(pages(h, Notice::warn("run:p1", "stopped again")).len(), 1);
1491        assert_eq!(pages(h, Notice::error("run:p1", "stopped again")).len(), 1);
1492        assert!(pages(h, Notice::error("run:p1", "stopped again")).is_empty());
1493    }
1494
1495    #[test]
1496    fn a_non_run_link_leaves_run_empty() {
1497        let dir = tempfile::tempdir().unwrap();
1498        let n = Notice::warn("task:t1", "held").link(Link::Task { id: "t1".into() });
1499        assert_eq!(pages(dir.path(), n)[0].run, "");
1500    }
1501
1502    #[test]
1503    fn a_dismissed_tombstone_stays_silent_until_it_changes() {
1504        let dir = tempfile::tempdir().unwrap();
1505        let h = dir.path();
1506        pages(h, Notice::warn("disk:r", "low"));
1507        Notices::at(h.join("notifications"))
1508            .dismiss(&id_of("disk:r"))
1509            .unwrap();
1510        assert!(pages(h, Notice::warn("disk:r", "low")).is_empty());
1511        assert_eq!(pages(h, Notice::warn("disk:r", "lower")).len(), 1);
1512    }
1513
1514    #[test]
1515    fn a_newer_cause_pages_again() {
1516        let dir = tempfile::tempdir().unwrap();
1517        let h = dir.path();
1518        let at = |s: i64| Some(Timestamp::from_second(s).unwrap());
1519        let n = |s| Notice::warn("task:t2", "held").since(at(s));
1520        assert_eq!(pages(h, n(100)).len(), 1);
1521        assert!(pages(h, n(100)).is_empty());
1522        assert_eq!(pages(h, n(200)).len(), 1);
1523    }
1524
1525    #[test]
1526    fn a_covered_notice_does_not_page() {
1527        let dir = tempfile::tempdir().unwrap();
1528        let h = dir.path();
1529        let q = ask(h, "r9", "implement");
1530        let n = || Notice::error("run:r9", "ended").about(["r9".to_owned()]);
1531        assert!(pages(h, n()).is_empty());
1532        let stored = Notices::at(h.join("notifications"))
1533            .get(&id_of("run:r9"))
1534            .unwrap();
1535        assert_eq!(stored.covered_by.as_deref(), Some(q.id.as_str()));
1536        assert!(pages(h, n()).is_empty());
1537    }
1538
1539    #[test]
1540    fn a_failing_sender_path_never_propagates() {
1541        // A pager that cannot run its command only warns.
1542        let notify = crate::config::Notify {
1543            command: vec!["magi-no-such-program-xyz".into()],
1544        };
1545        let rt = tokio::runtime::Builder::new_current_thread()
1546            .enable_all()
1547            .build()
1548            .unwrap();
1549        let r = rt.block_on(crate::ask::notify_text(&notify, "", "x"));
1550        assert!(r.is_err());
1551        // `send_page` without an installed pager is a no-op and cannot panic.
1552        send_page(
1553            Page {
1554                run: String::new(),
1555                summary: "x".into(),
1556            },
1557            Some(notify),
1558        );
1559    }
1560
1561    #[test]
1562    fn formatting_helpers_clean_text_and_lists() {
1563        assert_eq!(clean_text(None), None);
1564        assert_eq!(clean_text(Some("")), None);
1565        assert_eq!(clean_text(Some("   ")), None);
1566        assert_eq!(clean_text(Some("(no reason given)")), None);
1567        assert_eq!(clean_text(Some("(none)")), None);
1568        assert_eq!(clean_text(Some("  real reason  ")), Some("real reason"));
1569
1570        let empty: Vec<String> = Vec::new();
1571        assert_eq!(format_list(&empty), None);
1572        assert_eq!(format_list(&["", "  "]), None);
1573        assert_eq!(format_list(&["task-1"]), Some("task-1".into()));
1574        assert_eq!(format_list(&["a", "b", "c"]), Some("a, b, c".into()));
1575
1576        assert_eq!(ensure_sentence(""), "");
1577        assert_eq!(ensure_sentence("hello"), "hello.");
1578        assert_eq!(ensure_sentence("hello."), "hello.");
1579        assert_eq!(ensure_sentence("hello.."), "hello.");
1580    }
1581
1582    #[test]
1583    fn conductor_resumption_notice_is_readable_and_non_soliciting() {
1584        let task_id = "20261017-000000-abcd";
1585        let task_short = "abcd";
1586        let reason = "awaiting manual review";
1587        let blocked_by = vec!["dep-1".to_owned(), "dep-2".to_owned()];
1588        let evidence = vec!["check-failed".to_owned()];
1589        let has_question = true;
1590
1591        let notice = conductor_resumption(
1592            task_id,
1593            task_short,
1594            Some(reason),
1595            &blocked_by,
1596            &evidence,
1597            has_question,
1598        );
1599
1600        assert_eq!(notice.key, format!("conductor-resumption:{task_id}"));
1601        assert_eq!(notice.severity, Severity::Warn);
1602        assert_eq!(
1603            notice.link,
1604            Some(Link::Task {
1605                id: task_id.to_owned()
1606            })
1607        );
1608        assert_eq!(notice.subjects, vec![task_id.to_owned()]);
1609
1610        let msg = &notice.message;
1611        // Contains concise understandable prose explaining what happened
1612        assert!(msg.contains("Conductor raised a concern for task abcd after explicit resumption"));
1613        assert!(msg.contains("Concern: awaiting manual review."));
1614        assert!(msg.contains("Proposed dependencies: dep-1, dep-2."));
1615        assert!(msg.contains("Cited evidence: check-failed."));
1616        // Explicitly clarifies that no reply is required
1617        assert!(msg.contains("No reply to this notice is needed."));
1618        // No raw Debug formatting or dumped structures
1619        assert!(!msg.contains("Some("));
1620        assert!(!msg.contains("None"));
1621        assert!(!msg.contains("[]"));
1622        assert!(!msg.contains("{:?}"));
1623    }
1624
1625    #[test]
1626    fn conductor_resumption_notice_omits_empty_details() {
1627        let task_id = "20261017-000000-abcd";
1628        let notice = conductor_resumption(task_id, "abcd", None, &[], &[], false);
1629
1630        let msg = &notice.message;
1631        assert!(!msg.contains("Concern:"));
1632        assert!(!msg.contains("Proposed dependencies:"));
1633        assert!(!msg.contains("Cited evidence:"));
1634        assert!(!msg.contains("unregistered question"));
1635        assert!(!msg.contains("None"));
1636        assert!(!msg.contains("[]"));
1637        assert!(msg.contains("No reply to this notice is needed."));
1638    }
1639}