magi/

notices.rs

//! Durable attention-report outbox, reconciled into task/run chats by
//! [`crate::notice_delivery`]. Reports are facts to investigate, never decisions
//! or owner authority. The compatibility list remains for deferred delivery.
//!
//! Keys name kind + subject, never prose. Repeats update evidence without
//! another page; only a newer cause or severity escalation advances delivery.
//! Chat drafts persist their receipt key atomically before outbox acknowledgement.
//! Read/dismissed legacy history is never replayed, and pending unread records
//! are never evicted by the history cap.

use std::path::{Path, PathBuf};

use anyhow::{Context, Result, bail};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};

/// On-disk format. A file is refused only when its own schema is *greater*
/// than this one; every field added later must carry `#[serde(default)]`.
pub const SCHEMA: u32 = 1;

/// The history cap. Pending unread deliveries are exempt.
pub const CAP: usize = 200;

/// How serious a notice is. Ordered so a rise is a plain comparison.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Severity {
    /// Worth knowing, nothing to do.
    Info,
    /// Something needs a look soon.
    Warn,
    /// Something failed and stays failed until someone acts.
    Error,
}

/// Where a notice points, when it points anywhere.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub enum Link {
    /// A run, by id.
    Run {
        /// The run id.
        id: String,
    },
    /// A queued task, by id.
    Task {
        /// The task id.
        id: String,
    },
    /// An external page, such as a pull request.
    Url {
        /// The address.
        url: String,
    },
}

/// One thing the operator should know.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Notice {
    /// Stable file-name-safe id derived from [`Notice::key`].
    pub id: String,
    /// Deduplication key: kind and subject, never prose.
    pub key: String,
    /// How serious.
    pub severity: Severity,
    /// What happened.
    pub message: String,
    /// Where it points, when it points anywhere.
    #[serde(default)]
    pub link: Option<Link>,
    /// First raised.
    pub first_at: Timestamp,
    /// Last raised.
    pub last_at: Timestamp,
    /// How many times this key was raised.
    #[serde(default = "one")]
    pub count: u32,
    /// When it was marked read.
    #[serde(default)]
    pub read_at: Option<Timestamp>,
    /// When it was dismissed; the tombstone.
    #[serde(default)]
    pub dismissed_at: Option<Timestamp>,
    /// The task and run ids this notice is about. A question whose `run`
    /// names one of them already pages the operator for the same cause.
    #[serde(default)]
    pub subjects: Vec<String>,
    /// Run association when the display link points elsewhere (for example a PR).
    #[serde(default)]
    pub related_run: Option<String>,
    /// The question that carries this notice as context, when it was filed
    /// already read for that reason. Never a tombstone: a recurrence with a
    /// newer cause or higher severity is unread again.
    #[serde(default)]
    pub covered_by: Option<String>,
    /// When the cause this notice reports began (a task's `held_at`). A
    /// question filed before it is about an older cause and does not cover
    /// it; `None` keeps the broad coverage of notices without one.
    #[serde(default)]
    pub since: Option<Timestamp>,
    /// Producer-defined cause identity (phase and immutable target), never prose.
    #[serde(default)]
    pub cause: Option<String>,
    /// On-disk format version.
    #[serde(default = "schema")]
    pub schema: u32,
    /// Persistent incarnation, so pruning history cannot reuse a chat receipt.
    /// Legacy records keep their original receipt key.
    #[serde(default)]
    pub delivery_id: Option<String>,
    /// Delivery generation, advanced only by the shared page decision.
    #[serde(default = "one")]
    pub generation: u32,
    /// Last generation durably accepted by a chat.
    #[serde(default)]
    pub delivered: u32,
}

fn one() -> u32 {
    1
}

fn schema() -> u32 {
    SCHEMA
}

impl Notice {
    /// A notice of `severity` for `key`.
    pub fn new(severity: Severity, key: &str, message: impl Into<String>) -> Self {
        let now = Timestamp::now();
        Self {
            id: id_of(key),
            key: key.to_owned(),
            severity,
            message: message.into(),
            link: None,
            first_at: now,
            last_at: now,
            count: 1,
            read_at: None,
            dismissed_at: None,
            subjects: Vec::new(),
            related_run: None,
            covered_by: None,
            since: None,
            cause: None,
            schema: SCHEMA,
            delivery_id: Some(format!("{now}-{:016x}", crate::rng::entropy())),
            generation: 1,
            delivered: 0,
        }
    }

    /// Name the task / run ids this notice is about.
    pub fn about<I, S>(mut self, subjects: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.subjects = subjects.into_iter().map(Into::into).collect();
        self
    }

    /// Associate a run independently of the display link.
    pub fn for_run(mut self, run: Option<String>) -> Self {
        self.related_run = run;
        self
    }

    /// An [`Severity::Info`] notice.
    pub fn info(key: &str, message: impl Into<String>) -> Self {
        Self::new(Severity::Info, key, message)
    }

    /// A [`Severity::Warn`] notice.
    pub fn warn(key: &str, message: impl Into<String>) -> Self {
        Self::new(Severity::Warn, key, message)
    }

    /// An [`Severity::Error`] notice.
    pub fn error(key: &str, message: impl Into<String>) -> Self {
        Self::new(Severity::Error, key, message)
    }

    /// Record when the reported cause began.
    pub fn since(mut self, at: Option<Timestamp>) -> Self {
        self.since = at;
        self
    }

    /// Distinguish material producer state transitions without keying on wording.
    pub fn cause(mut self, cause: impl Into<String>) -> Self {
        self.cause = Some(cause.into());
        self
    }

    /// Attach a link.
    pub fn link(mut self, link: Link) -> Self {
        self.link = Some(link);
        self
    }

    /// Neither read nor dismissed.
    pub fn unread(&self) -> bool {
        self.read_at.is_none() && self.dismissed_at.is_none()
    }

    /// Fold a repeat into this notice. Wording updates evidence; a newer cause
    /// or higher severity makes it unread and advances delivery.
    ///
    /// Returns whether this raise should page the operator: it relit the
    /// notice (higher severity or newer cause) and no open question
    /// already carries it. This is the only place that decides.
    pub fn raise_again(&mut self, again: &Notice, now: Timestamp) -> bool {
        self.count = self.count.saturating_add(1);
        let previous_last_at = self.last_at;
        self.last_at = now;
        if again.link.is_some() {
            self.link = again.link.clone();
        }
        if again.related_run.is_some() {
            self.related_run = again.related_run.clone();
        }
        if !again.subjects.is_empty() {
            self.subjects = again.subjects.clone();
        }
        let escalated = again.severity > self.severity;
        // Legacy history lacks a cause timestamp. Only a cause that began
        // after the saved raise is news; replaying that old failure is not.
        // A previously covering question can also prove a new uncovered cause.
        let newer_cause = match (self.since, again.since) {
            (Some(old), Some(new)) => new > old,
            (None, Some(new)) => {
                new > previous_last_at || (self.covered_by.is_some() && again.covered_by.is_none())
            }
            _ => false,
        };
        self.since = self.since.max(again.since);
        // Wording is evidence, never a reason to spend another agent turn.
        let changed_cause = again
            .cause
            .as_ref()
            .is_some_and(|cause| self.cause.as_ref() != Some(cause));
        if again.cause.is_some() {
            self.cause = again.cause.clone();
        }
        let relit = escalated || newer_cause || changed_cause;
        self.message = again.message.clone();
        if relit {
            self.generation = self.generation.saturating_add(1);
            self.severity = self.severity.max(again.severity);
            self.read_at = None;
            self.dismissed_at = None;
            self.covered_by = None;
        }
        // A question already pages for this cause: keep the record, skip the
        // second page. A higher severity is news the question did not carry.
        if let Some(q) = &again.covered_by
            && !escalated
            && self.unread()
        {
            self.read_at = Some(now);
            self.covered_by = Some(q.clone());
        }
        relit && self.covered_by.is_none()
    }

    /// Mark read, keeping the first read time.
    pub fn mark_read(&mut self, now: Timestamp) {
        if self.read_at.is_none() {
            self.read_at = Some(now);
        }
    }

    /// Tombstone: read and hidden from the list.
    pub fn dismiss(&mut self, now: Timestamp) {
        self.mark_read(now);
        if self.dismissed_at.is_none() {
            self.dismissed_at = Some(now);
        }
    }

    /// Legacy exhausted-deputy warnings belong on the question card now.
    /// Match only this retired namespace, not `deputy-turn:` failures.
    fn obsolete(&self) -> bool {
        self.key.starts_with("deputy:")
    }

    /// Whether this notice belongs in the notification centre and bell.
    fn visible(&self) -> bool {
        !self.obsolete() && self.dismissed_at.is_none()
    }

    /// Pruning order: obsolete, dismissed, read, then unread.
    fn keep_rank(&self) -> u8 {
        if self.obsolete() {
            0
        } else if self.dismissed_at.is_some() {
            1
        } else if self.read_at.is_some() {
            2
        } else {
            3
        }
    }
}

/// A stable, file-name-safe id for a key: a readable slug plus FNV-1a of the
/// whole key. Not `DefaultHasher`, whose output may change between builds and
/// would strand every existing file's dedupe.
pub fn id_of(key: &str) -> String {
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    for b in key.bytes() {
        hash ^= u64::from(b);
        hash = hash.wrapping_mul(0x0100_0000_01b3);
    }
    let slug: String = key
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() {
                c.to_ascii_lowercase()
            } else {
                '-'
            }
        })
        .take(32)
        .collect();
    format!("{slug}-{hash:016x}")
}

/// Whether `id` could have come from [`id_of`]. Checked before any path is
/// built from a request.
fn valid_id(id: &str) -> bool {
    !id.is_empty()
        && id.len() <= 64
        && id
            .bytes()
            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
}

/// How long a caller waits for a notice's lock before proceeding without it.
const LOCK_WAIT: std::time::Duration = std::time::Duration::from_secs(5);

/// Age after which a lock file is taken to belong to a dead process.
const LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);

/// Distinguishes temp files written by threads of one process.
static TMP_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// A notice store on disk.
#[derive(Debug, Clone)]
pub struct Notices {
    root: PathBuf,
}

impl Notices {
    /// The operator's notices, `<home>/notifications`.
    pub fn open() -> Self {
        Self::at(crate::run::home().join("notifications"))
    }

    /// A store at an explicit root; tests use this.
    pub fn at(root: PathBuf) -> Self {
        Self { root }
    }

    /// Directory holding the notice files.
    pub fn root(&self) -> &Path {
        &self.root
    }

    fn path_of(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.json"))
    }

    fn put(&self, n: &Notice) -> Result<()> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        let body = serde_json::to_string_pretty(n).context("serialize notice")?;
        let path = self.path_of(&n.id);
        // Unique per write: two processes raising or marking the same notice
        // must not share a temp file, or the second rename finds it gone.
        let tmp = path.with_extension(format!(
            "json.{}.{}.tmp",
            std::process::id(),
            TMP_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        ));
        std::fs::write(&tmp, body).with_context(|| format!("write {}", tmp.display()))?;
        if let Err(e) = std::fs::rename(&tmp, &path) {
            let _ = std::fs::remove_file(&tmp);
            return Err(e).with_context(|| format!("replace {}", path.display()));
        }
        Ok(())
    }

    /// Load one notice by id.
    pub fn get(&self, id: &str) -> Result<Notice> {
        if !valid_id(id) {
            bail!("`{id}` is not a notification id");
        }
        read_path(&self.path_of(id))
    }

    /// Run a read-modify-write of one notice under its lock file.
    ///
    /// `magi serve`, `magi web` and the CLI all update the same files, and a
    /// unique temp file only makes each write atomic - it does not stop one
    /// process saving a stale read over another's newer raise. The lock is an
    /// exclusive-create file beside the notice; a holder that died is broken
    /// after [`LOCK_STALE`], and a caller that cannot get it in
    /// [`LOCK_WAIT`] proceeds anyway rather than lose the notice.
    fn locked<T>(&self, id: &str, f: impl FnOnce() -> Result<T>) -> Result<T> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        let lock = self.root.join(format!("{id}.lock"));
        let start = std::time::Instant::now();
        let retry = crate::lock_file::Retry::new(Some(start + LOCK_WAIT));
        let mut held = false;
        while start.elapsed() < LOCK_WAIT {
            match retry.create(&lock) {
                Ok(_) => {
                    held = true;
                    break;
                }
                Err(_) => {
                    let stale = std::fs::metadata(&lock)
                        .and_then(|m| m.modified())
                        .ok()
                        .and_then(|t| t.elapsed().ok())
                        .is_some_and(|age| age > LOCK_STALE);
                    if stale {
                        let _ = std::fs::remove_file(&lock);
                    } else {
                        std::thread::sleep(std::time::Duration::from_millis(5));
                    }
                }
            }
        }
        let out = f();
        if held {
            let _ = std::fs::remove_file(&lock);
        }
        out
    }

    /// File a notice, folding it into an existing one with the same key.
    pub fn raise(&self, incoming: Notice) -> Result<Notice> {
        self.raise_paged(incoming).map(|(n, _)| n)
    }

    /// File through the common page decision, including explicit-store producers.
    pub fn raise_with(
        &self,
        mut incoming: Notice,
        notify: &crate::config::Notify,
    ) -> Result<Notice> {
        if let Some(home) = self.root.parent()
            && let Some(q) = covering_question(home, &incoming)
        {
            incoming.covered_by = Some(q);
        }
        let (notice, page) = self.raise_paged(incoming)?;
        if page {
            send_page(
                Page {
                    run: notice
                        .related_run
                        .clone()
                        .or_else(|| match &notice.link {
                            Some(Link::Run { id }) => Some(id.clone()),
                            _ => None,
                        })
                        .unwrap_or_default(),
                    summary: notice.message.clone(),
                },
                Some(notify.clone()),
            );
        }
        Ok(notice)
    }

    /// [`Notices::raise`], also saying whether the raise should page.
    pub fn raise_paged(&self, incoming: Notice) -> Result<(Notice, bool)> {
        let out = self.locked(&incoming.id.clone(), || {
            let now = Timestamp::now();
            let (stored, page) = match read_path(&self.path_of(&incoming.id)) {
                Ok(mut existing) => {
                    let page = existing.raise_again(&incoming, now);
                    (existing, page)
                }
                Err(_) => {
                    let mut fresh = incoming;
                    if fresh.covered_by.is_some() {
                        fresh.read_at = Some(now);
                    }
                    let page = fresh.covered_by.is_none();
                    (fresh, page)
                }
            };
            self.put(&stored)?;
            Ok((stored, page))
        })?;
        self.prune();
        Ok(out)
    }

    /// Mark every unread notice about `subject` read, recording `question` as
    /// the card that carries it. Best-effort per notice.
    pub fn cover(&self, question: &crate::ask::Question) {
        for n in self.all() {
            if n.unread() && covers(question, &n) {
                let _ = self.update(&n.id, |n| {
                    // Re-checked under the lock: a newer cause may have
                    // landed since `all()` read it.
                    if n.unread() && covers(question, n) {
                        n.mark_read(Timestamp::now());
                        n.covered_by = Some(question.id.clone());
                    }
                });
            }
        }
    }

    /// Load, change and save one notice under its lock.
    pub(crate) fn update(&self, id: &str, change: impl FnOnce(&mut Notice)) -> Result<Notice> {
        if !valid_id(id) {
            bail!("`{id}` is not a notification id");
        }
        self.locked(id, || {
            let mut n = read_path(&self.path_of(id))?;
            change(&mut n);
            self.put(&n)?;
            Ok(n)
        })
    }

    /// Every notice on disk, unreadable files skipped, newest first.
    pub(crate) fn all(&self) -> Vec<Notice> {
        let mut all: Vec<Notice> = std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|x| x == "json"))
            .filter_map(|p| read_path(&p).ok())
            .collect();
        all.sort_by(|a, b| b.last_at.cmp(&a.last_at).then_with(|| a.id.cmp(&b.id)));
        all
    }

    /// What the bell lists: visible, current notices, newest first.
    pub fn list(&self) -> Vec<Notice> {
        self.all().into_iter().filter(Notice::visible).collect()
    }

    /// How many notices are unread.
    pub fn count_unread(&self) -> usize {
        self.list().iter().filter(|n| n.unread()).count()
    }

    /// Mark read, keeping the first read time.
    pub fn mark_read(&self, id: &str) -> Result<Notice> {
        let now = Timestamp::now();
        self.update(id, |n| n.mark_read(now))
    }

    /// Mark every unread notice read; returns how many changed.
    pub fn mark_all_read(&self) -> Result<usize> {
        let now = Timestamp::now();
        let mut changed = 0;
        for n in self.list().into_iter().filter(Notice::unread) {
            // Re-read under the lock: a raise since the scan may have made it
            // a different, still-unread notice, which stays as it is.
            let done = self.update(&n.id, |n| n.mark_read(now));
            if done.is_ok() {
                changed += 1;
            }
        }
        Ok(changed)
    }

    /// Tombstone: read and hidden from the list.
    pub fn dismiss(&self, id: &str) -> Result<Notice> {
        let now = Timestamp::now();
        self.update(id, |n| n.dismiss(now))
    }

    /// Keep at most [`CAP`] files. Best-effort.
    fn prune(&self) {
        let mut all = self.all();
        if all.len() <= CAP {
            return;
        }
        // Most worth keeping first; drop the tail.
        all.sort_by(|a, b| {
            b.keep_rank()
                .cmp(&a.keep_rank())
                .then_with(|| b.last_at.cmp(&a.last_at))
        });
        for n in all
            .split_off(CAP)
            .into_iter()
            .filter(|n| n.obsolete() || !n.unread() || n.delivered >= n.generation)
        {
            let _ = std::fs::remove_file(self.path_of(&n.id));
        }
    }

    /// Change token for the stream. Hashes every file's name and mtime, so a
    /// prune or delete moves it as surely as a write does (a max-mtime would
    /// not move when the newest file is not the one removed).
    pub fn revision(&self) -> u64 {
        use std::hash::{Hash as _, Hasher as _};
        let mut entries: Vec<(std::ffi::OsString, u128)> = std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
            .map(|e| {
                let at = e
                    .metadata()
                    .and_then(|m| m.modified())
                    .ok()
                    .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                    .map_or(0, |d| d.as_nanos());
                (e.file_name(), at)
            })
            .collect();
        if entries.is_empty() {
            return 0;
        }
        entries.sort();
        let mut h = std::collections::hash_map::DefaultHasher::new();
        entries.hash(&mut h);
        h.finish()
    }
}

fn read_path(path: &Path) -> Result<Notice> {
    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
    let n: Notice =
        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
    if n.schema > SCHEMA {
        bail!(
            "notice {} was written by a newer magi (schema {}, this build speaks up to {SCHEMA})",
            n.id,
            n.schema
        );
    }
    Ok(n)
}

/// A non-empty optional detail, trimmed for display. Never rewrites saved prose.
pub(crate) fn detail(value: Option<&str>) -> Option<&str> {
    clean_text(value)
}

/// Short explanation of a CLI turn that produced no usable response.
pub(crate) fn turn_status(code: Option<i32>, timed_out: bool, quota_exhausted: bool) -> String {
    let mut parts = vec![exit_status(code)];
    if timed_out {
        parts.push("time limit reached".to_owned());
    }
    if quota_exhausted {
        parts.push("quota exhausted".to_owned());
    }
    parts.join("; ")
}

/// A currently open registered question associated with this notice, if any.
/// `covered_by` records historical deduplication, not current answerability.
pub(crate) fn answerable_question(n: &Notice, questions: &crate::ask::Questions) -> Option<String> {
    let direct = n.covered_by.as_deref().or_else(|| {
        ["question:", "deputy-turn:", "repair-owner:", "action:"]
            .iter()
            .find_map(|prefix| n.key.strip_prefix(prefix))
    });
    let release = if direct.is_none() && n.key.starts_with("release-pr:") {
        questions
            .root()
            .parent()
            .and_then(|home| crate::release_watch::question_for_notice(home, &n.key))
    } else {
        None
    };
    let id = direct.or(release.as_deref())?;
    questions
        .get(id)
        .ok()
        .filter(|q| q.status.open())
        .map(|q| q.id)
}

/// The notice for a run that ended Blocked, Stalled or Failed, if it did.
///
/// A parked run was only asked to stop, so it is not news. A run that left a
/// pull request behind still is: the PR is waiting on someone. Keyed on the
/// run with a message free of anything that varies between attempts.
pub fn run_ended(state: &crate::run::RunState) -> Option<Notice> {
    use crate::run::RunStatus;
    let failed = matches!(
        state.status,
        RunStatus::Blocked | RunStatus::Stalled | RunStatus::Failed
    );
    (failed && !state.parked).then(|| {
        Notice::error(
            &format!("run:{}", state.id),
            format!(
                "Run {} ended {}. {}",
                state.id,
                state.status.as_str(),
                run_cause(state)
            ),
        )
        .link(Link::Run {
            id: state.id.clone(),
        })
        .about([state.id.clone()])
        .since(Some(
            state
                .base_sync_recovery
                .stop
                .as_ref()
                .filter(|_| state.base_sync_blocked())
                .map(|s| s.at)
                .or_else(|| state.passes.last().map(|p| p.started_at))
                .unwrap_or(state.created_at),
        ))
    })
}

/// The notice for a pull request that merged while checks were red.
///
/// Keyed on the run. `summary` names the repository, pull request and failing
/// checks: with no notify command configured this is the only place the
/// operator learns them. A run merges once, so the wording cannot churn.
pub fn merged_red(run_id: &str, summary: &str) -> Notice {
    Notice::warn(&format!("merged-red:{run_id}"), summary).link(Link::Run {
        id: run_id.to_owned(),
    })
}

/// Human-readable process status for operator-facing failure details.
pub fn exit_status(code: Option<i32>) -> String {
    match code {
        Some(code) => format!("exit code {code}"),
        None => "exit code unavailable".to_owned(),
    }
}

/// Trims text and drops empty values or meaningless placeholder phrases like `"(no reason given)"`.
pub fn clean_text(s: Option<&str>) -> Option<&str> {
    s.map(str::trim)
        .filter(|t| !t.is_empty() && *t != "(no reason given)" && *t != "(none)")
}

/// Formats a list of identifiers or items into human-readable comma-separated prose.
/// Returns `None` if the list is empty or contains only blank entries.
pub fn format_list<T: AsRef<str>>(items: &[T]) -> Option<String> {
    let non_empty: Vec<&str> = items
        .iter()
        .map(|s| s.as_ref().trim())
        .filter(|s| !s.is_empty())
        .collect();
    if non_empty.is_empty() {
        None
    } else {
        Some(non_empty.join(", "))
    }
}

/// Ensures a sentence ends with a single period and has no trailing whitespace.
pub fn ensure_sentence(text: &str) -> String {
    let trimmed = text.trim();
    if trimmed.is_empty() {
        return String::new();
    }
    let trimmed_dot = trimmed.trim_end_matches('.');
    format!("{trimmed_dot}.")
}

/// The notice for a task the machine or its attempt budget has held, if it is.
///
/// Called from [`crate::queue::Queue::put`], which every task transition goes
/// through, so a hold made anywhere - the loop, the conductor, triage, a
/// dependency removed from under a blocked task - is announced. A hold the
/// operator placed by hand is their own action and is not news. Keyed on the
/// task, with wording free of anything that varies between retries.
///
/// Falls back to [`crate::queue::Task::last_error`] when `hold_reason` is
/// empty, the same fallback `crate::triage`'s question detail uses: a record
/// written before `Task::fail`/`Task::handed_off` started copying `why` into
/// `hold_reason` too would otherwise still read "no reason recorded" even
/// though the run said exactly why it stopped.
pub fn task_held(task: &crate::queue::Task) -> Option<Notice> {
    use crate::queue::{HoldSource, TaskStatus};
    if task.status != TaskStatus::Held || task.hold_source != Some(HoldSource::Machine) {
        return None;
    }
    let why = clean_text(task.hold_reason.as_deref())
        .or_else(|| clean_text(task.last_error.as_deref()))
        .unwrap_or("no reason recorded");
    let why_sentence = why.trim_end_matches('.');
    Some(
        Notice::warn(
            &format!("task:{}", task.id),
            format!("Task {} is held: {why_sentence}.", task.short()),
        )
        .link(Link::Task {
            id: task.id.clone(),
        })
        .about([task.id.clone()])
        .since(task.held_at),
    )
}

/// The notice for a conductor concern raised after explicit resumption without
/// verified new blocking evidence.
///
/// Informational: preserves the task state without re-blocking on old evidence,
/// explains that no state changed and no reply is required, and attaches a link
/// to the task. Any proposed question text is omitted to avoid soliciting an
/// answer to an unregistered question.
pub fn conductor_resumption(
    task_id: &str,
    task_short: &str,
    reason: Option<&str>,
    blocked_by: &[String],
    evidence: &[String],
    has_question: bool,
) -> Notice {
    let mut parts: Vec<String> = Vec::new();
    parts.push(format!(
        "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."
    ));

    if let Some(r) = clean_text(reason) {
        parts.push(format!("Concern: {}.", r.trim_end_matches('.')));
    }

    if has_question {
        parts.push("An unregistered question was proposed; no reply is required.".to_owned());
    }

    if let Some(deps) = format_list(blocked_by) {
        parts.push(format!("Proposed dependencies: {deps}."));
    }

    if let Some(ev) = format_list(evidence) {
        parts.push(format!("Cited evidence: {ev}."));
    }

    parts.push("Existing holds and questions still apply.".to_owned());
    parts.push(
        "No reply to this notice is needed. To hold this task, use manual hold via the queue."
            .to_owned(),
    );

    Notice::warn(&format!("conductor-resumption:{task_id}"), parts.join(" "))
        .link(Link::Task {
            id: task_id.to_owned(),
        })
        .about([task_id.to_owned()])
}

/// The notice for a run whose graph returned an error before it settled.
pub fn run_stopped(id: &str, state: &crate::run::RunState) -> Notice {
    Notice::error(
        &format!("run:{id}"),
        format!("Run {} stopped with an error. {}", id, run_cause(state)),
    )
    .link(Link::Run { id: id.to_owned() })
    .about([id.to_owned()])
    .since(Some(
        state
            .base_sync_recovery
            .stop
            .as_ref()
            .filter(|_| state.base_sync_blocked())
            .map(|s| s.at)
            .or_else(|| state.passes.last().map(|p| p.started_at))
            .unwrap_or(state.created_at),
    ))
}

/// Last recorded graph evidence; immediate errors are attached by the caller.
fn run_cause(state: &crate::run::RunState) -> String {
    let verification_cause = state.verification_blocking_cause();
    let cause = state
        .prep_failure
        .as_deref()
        .or_else(|| state.resource_block.as_ref().map(|b| b.detail.as_str()))
        .or_else(|| {
            state
                .candidate_preservation
                .values()
                .next()
                .map(String::as_str)
        })
        .or(verification_cause.as_deref())
        .or_else(|| state.events.last().map(|e| e.message.as_str()));
    if state.base_sync_blocked() || state.resource_block.is_some() {
        return format!(
            "{} Next: {}",
            cause.unwrap_or("Synchronization stopped."),
            state.base_sync_next_action()
        );
    }
    cause
        .and_then(|s| s.lines().find(|line| !line.trim().is_empty()))
        .unwrap_or("No cause recorded; inspect the run.")
        .to_owned()
}

/// One page for the operator's `[notify]` command.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Page {
    /// The run id when the notice links to a run, else empty.
    pub run: String,
    /// The notice message.
    pub summary: String,
}

/// Set once by `main`: until then no notice fires the command, so a unit or
/// integration test that pins a home never reaches the operator's real config.
static PAGER: std::sync::OnceLock<()> = std::sync::OnceLock::new();

/// Let notices fire the `[notify]` command in this process.
pub fn install_pager() {
    let _ = PAGER.set(());
}

/// The production sender. Fire and forget on its own thread with its own
/// runtime, so it works with or without an ambient one and never blocks the
/// producer; `ask::notify_text` bounds the command by `NOTIFY_TIMEOUT`.
/// `notify` is the producer's own config, else the machine layer's.
fn send_page(page: Page, notify: Option<crate::config::Notify>) {
    if PAGER.get().is_none() {
        return;
    }
    let spawned = std::thread::Builder::new()
        .name("notice-pager".into())
        .spawn(move || {
            let notify = notify.unwrap_or_else(machine_notify);
            if notify.command.is_empty() {
                return;
            }
            let rt = match tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
            {
                Ok(rt) => rt,
                Err(e) => return tracing::warn!("could not page about a notification: {e:#}"),
            };
            if let Err(e) = rt.block_on(crate::ask::notify_text(&notify, &page.run, &page.summary))
            {
                tracing::warn!("could not page about a notification: {e:#}");
            }
        });
    if let Err(e) = spawned {
        tracing::warn!("could not page about a notification: {e:#}");
    }
}

/// `[notify]` from the machine layer alone; a producer with no repo config
/// cannot name a better one. Unreadable is no command, with a warning.
fn machine_notify() -> crate::config::Notify {
    let layers: Vec<PathBuf> = crate::config::Config::machine_layer()
        .into_iter()
        .filter(|p| p.is_file())
        .collect();
    match crate::config::Config::load_layers(&layers) {
        Ok(c) => c.notify,
        Err(e) => {
            tracing::warn!("could not read [notify] for a notification: {e:#}");
            Default::default()
        }
    }
}

/// The one function producers call. Best-effort by construction: a notice that
/// cannot be filed is a `tracing::warn`, never a reason to fail the run, the
/// loop or the request that wanted to mention it.
pub fn raise(notice: Notice) {
    if let Some(home) = crate::run::try_home() {
        raise_in(&home, notice);
    }
}

/// [`raise`] for a producer that already holds the repo's `[notify]`.
pub fn raise_with(notice: Notice, notify: &crate::config::Notify) {
    if let Some(home) = crate::run::try_home() {
        raise_in_with(&home, notice, &|page| send_page(page, Some(notify.clone())));
    }
}

/// Does open question `q` already page the operator for what `n` reports?
///
/// The cause is decided by what each side is about, not by when it happened:
/// a conductor / triage question exists *because* its task is held, so it
/// covers that task's hold and handover notices whenever it was filed; a
/// question raised from inside a run (any other seat node) covers that run's
/// ended / stopped notices and never a task hold. Exact match on
/// `Question::run` (a task id for the former, a run id for the latter). The
/// land approval and release questions are to-dos, not duplicates, and cover
/// nothing.
///
/// A notice with a `since` (a task hold's start) is covered only by a question
/// filed at or after it: one filed earlier is about an older cause, and
/// letting it silence a new hold would hide a fresh failure behind a stale
/// card. Missing a duplicate costs one extra page; hiding a failure costs the
/// failure. A notice without `since` keeps the broad rule.
pub fn covers(q: &crate::ask::Question, n: &Notice) -> bool {
    // Delivery alone proves no relationship to a later failure in the same
    // run pass. Retired questions only retain coverage already recorded while
    // they were open; never infer a cause match from the run ID or pass start.
    let recorded_coverage =
        q.routed_chat.is_some() && n.covered_by.as_deref() == Some(q.id.as_str());
    if !(q.status.open() || recorded_coverage) || !n.subjects.contains(&q.run) {
        return false;
    }
    if n.since.is_some_and(|since| q.asked_at < since) {
        return false;
    }
    let about_task = n.key.starts_with("task:") || n.key.starts_with("handover:");
    let about_run = n.key.starts_with("run:");
    match q.node.as_str() {
        crate::bump::NOTICE_NODE | crate::land::APPROVAL_NODE => false,
        _ if q.run_names_task() => about_task,
        _ => about_run,
    }
}

fn covering_question(home: &Path, n: &Notice) -> Option<String> {
    if n.subjects.is_empty() {
        return None;
    }
    crate::ask::Questions::at(home.join("questions"))
        .list()
        .into_iter()
        .find(|q| covers(q, n))
        .map(|q| q.id)
}

/// A question was just filed for `run`: quiet the unread notices about it.
pub fn quiet_for(home: &Path, question: &crate::ask::Question) {
    Notices::at(home.join("notifications")).cover(question);
}

/// [`raise`] into an explicit magi home, for callers that already carry one
/// (the janitor) and so must not reach for the process-global.
pub fn raise_in(home: &Path, notice: Notice) {
    raise_in_with(home, notice, &|page| send_page(page, None));
}

/// [`raise_in`] with the pager injected. `send` is called only when the raise
/// should page (see [`Notice::raise_again`]); it must not block.
pub fn raise_in_with(home: &Path, mut notice: Notice, send: &dyn Fn(Page)) {
    if let Some(q) = covering_question(home, &notice) {
        notice.covered_by = Some(q);
    }
    match Notices::at(home.join("notifications")).raise_paged(notice) {
        Ok((n, true)) => send(Page {
            run: n
                .related_run
                .clone()
                .or_else(|| match &n.link {
                    Some(Link::Run { id }) => Some(id.clone()),
                    _ => None,
                })
                .unwrap_or_default(),
            summary: n.message.clone(),
        }),
        Ok(_) => {}
        Err(e) => tracing::warn!("could not file a notification: {e:#}"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn notification_details_omit_blanks_and_explain_process_results() {
        assert_eq!(detail(None), None);
        assert_eq!(detail(Some(" \n\t")), None);
        assert_eq!(detail(Some(" concern ")), Some("concern"));
        assert_eq!(exit_status(Some(7)), "exit code 7");
        assert_eq!(exit_status(None), "exit code unavailable");
        assert_eq!(
            turn_status(None, true, true),
            "exit code unavailable; time limit reached; quota exhausted"
        );
        let mut task = held_task();
        task.hold_reason = Some("   ".into());
        task.last_error = Some(" failure recorded ".into());
        assert!(
            task_held(&task)
                .unwrap()
                .message
                .ends_with("failure recorded.")
        );
    }

    #[test]
    fn resumption_notice_and_external_page_are_informational() {
        let dir = tempfile::tempdir().unwrap();
        let n = conductor_resumption(
            "t1",
            "t1",
            Some("Inspection remains a concern."),
            &["dep1".into(), " ".into()],
            &[],
            false,
        );
        assert!(n.message.contains("Concern: Inspection remains a concern."));
        assert!(n.message.contains("Proposed dependencies: dep1."));
        assert!(n.message.contains("its state was not changed"));
        assert!(
            n.message
                .contains("Existing holds and questions still apply")
        );
        assert!(n.message.contains("No reply to this notice is needed"));
        assert_eq!(n.link, Some(Link::Task { id: "t1".into() }));
        let sent = pages(dir.path(), n.clone());
        assert_eq!(sent[0].summary, n.message);
        assert!(pages(dir.path(), n).is_empty(), "repeat remains silent");
        let empty = conductor_resumption("t2", "t2", Some(" "), &[], &[], false);
        for absent in [
            "Concern:",
            "Proposed dependencies:",
            "Some(",
            "None",
            "[]",
            "question:",
            "evidence:",
            "?",
        ] {
            assert!(!empty.message.contains(absent), "{absent}");
        }
    }

    #[test]
    fn only_currently_open_registered_questions_have_response_links() {
        let dir = tempfile::tempdir().unwrap();
        let q = ask(dir.path(), "t1", "conduct");
        let questions = crate::ask::Questions::at(dir.path().join("questions"));
        let mut n = Notice::warn("task:t1", "held");
        n.covered_by = Some(q.id.clone());
        assert_eq!(answerable_question(&n, &questions), Some(q.id.clone()));
        questions
            .update(&q.id, |q| {
                q.answer(crate::ask::Answer::Text("continue".into()))
            })
            .unwrap();
        assert_eq!(answerable_question(&n, &questions), None);
        let abandoned = ask(dir.path(), "t2", "conduct");
        let failure = Notice::warn(&format!("question:{}", abandoned.id), "delivery failed");
        assert_eq!(
            answerable_question(&failure, &questions),
            Some(abandoned.id.clone())
        );
        questions
            .update(&abandoned.id, |q| {
                q.abandon("run gone");
                Ok(())
            })
            .unwrap();
        assert_eq!(answerable_question(&failure, &questions), None);
        n.covered_by = Some("missing".into());
        assert_eq!(answerable_question(&n, &questions), None);
        assert_eq!(
            answerable_question(
                &conductor_resumption("t1", "t1", None, &[], &[], false),
                &questions
            ),
            None
        );
    }

    #[test]
    fn legacy_deputy_warnings_are_hidden_without_hiding_other_namespaces() {
        let (_dir, s) = store();
        // No question lookup: these may belong to deleted or settled questions.
        for key in ["deputy:open", "deputy:answered", "deputy:deleted"] {
            s.raise(Notice::warn(key, "old exhausted warning")).unwrap();
        }
        let keys = ["deputy-turn:open", "deputyish:open", "deputy", "task:open"];
        for key in keys {
            s.raise(Notice::warn(key, "still relevant")).unwrap();
        }
        assert_eq!(s.list().len(), keys.len());
        assert_eq!(s.count_unread(), keys.len());
        assert!(s.list().iter().all(|n| keys.contains(&n.key.as_str())));
        assert_eq!(s.mark_all_read().unwrap(), keys.len());
        assert_eq!(s.count_unread(), 0);
        assert!(s.get(&id_of("deputy:open")).unwrap().unread());
    }

    #[test]
    fn obsolete_deputy_warnings_are_pruned_before_current_notices() {
        let (_dir, s) = store();
        for i in 0..CAP {
            let id = s
                .raise(Notice::info(&format!("current:{i}"), "keep"))
                .unwrap()
                .id;
            s.dismiss(&id).unwrap();
        }
        // Even a newer unread legacy warning cannot evict a current tombstone.
        s.raise(Notice::warn("deputy:old", "obsolete")).unwrap();
        assert!(s.get(&id_of("deputy:old")).is_err());
        assert_eq!(s.all().len(), CAP);
        assert!(s.all().iter().all(|n| n.key.starts_with("current:")));
    }

    #[test]
    fn process_exit_status_is_readable_even_without_a_code() {
        assert_eq!(exit_status(Some(1)), "exit code 1");
        assert_eq!(exit_status(None), "exit code unavailable");
    }

    fn store() -> (tempfile::TempDir, Notices) {
        let dir = tempfile::tempdir().unwrap();
        let s = Notices::at(dir.path().join("notifications"));
        (dir, s)
    }

    #[test]
    fn a_red_merge_notice_is_keyed_on_the_run_and_raised_once() {
        let (_d, s) = store();
        let n = merged_red("run-1", "Merged a/b PR #1 with red checks: x (u)");
        assert_eq!(n.key, "merged-red:run-1");
        assert_eq!(
            n.link,
            Some(Link::Run {
                id: "run-1".to_owned()
            })
        );
        s.raise(merged_red(
            "run-1",
            "Merged a/b PR #1 with red checks: x (u)",
        ))
        .unwrap();
        s.raise(merged_red(
            "run-1",
            "Merged a/b PR #1 with red checks: x (u)",
        ))
        .unwrap();
        assert_eq!(s.list().len(), 1);
    }

    #[test]
    fn producer_cause_transitions_page_once_without_using_message_prose() {
        let (dir, store) = store();
        let sent = std::cell::RefCell::new(Vec::new());
        let raise = |cause: &str, message: &str| {
            raise_in_with(
                dir.path(),
                Notice::warn("release-pr:pr", message).cause(cause),
                &|p| sent.borrow_mut().push(p),
            )
        };
        raise("approval:h1", "approval is needed");
        let id = id_of("release-pr:pr");
        store.mark_read(&id).unwrap();
        raise("approval:h1", "approval wording updated");
        assert!(!store.get(&id).unwrap().unread());
        raise("merge-failed:h1", "merge refused");
        assert!(store.get(&id).unwrap().unread());
        assert_eq!(sent.borrow().len(), 2);
        assert_eq!(store.get(&id).unwrap().generation, 2);
        store.mark_read(&id).unwrap();
        raise("merge-failed:h1", "merge refusal wording updated");
        assert_eq!(sent.borrow().len(), 2);
        assert!(!store.get(&id).unwrap().unread());
    }

    #[test]
    fn a_repeat_updates_evidence_without_relighting_it() {
        let now = Timestamp::now();
        let mut n = Notice::warn("task:1", "held");
        n.mark_read(now);
        n.raise_again(&Notice::warn("task:1", "held"), now);
        assert_eq!(n.count, 2);
        assert!(n.read_at.is_some(), "identical repeat stays read");
        n.raise_again(&Notice::warn("task:1", "held differently"), now);
        assert!(!n.unread());
        assert_eq!(n.message, "held differently");
    }

    #[test]
    fn a_rise_in_severity_resurrects_a_dismissed_notice_but_a_repeat_does_not() {
        let now = Timestamp::now();
        let mut n = Notice::warn("task:x", "m");
        n.dismiss(now);
        n.raise_again(&Notice::warn("task:x", "m"), now);
        assert!(n.dismissed_at.is_some(), "tombstone holds");
        n.raise_again(&Notice::error("k", "m"), now);
        assert!(n.unread());
        assert_eq!(n.severity, Severity::Error);
        // Severity never falls on a repeat.
        n.raise_again(&Notice::info("k", "other"), now);
        assert_eq!(n.severity, Severity::Error);
    }

    #[test]
    fn adding_link_on_repeat_updates_link_without_relighting_read_or_dismissed() {
        let now = Timestamp::now();
        // 1. Read notice
        let mut read_notice = Notice::warn("task:1", "held");
        read_notice.mark_read(now);
        let link_to_add = Link::Task { id: "1".to_owned() };
        let again = Notice::warn("task:1", "held").link(link_to_add.clone());
        let relit = read_notice.raise_again(&again, now);
        assert!(!relit, "link-only update must not page/relight");
        assert!(read_notice.read_at.is_some(), "read notice stays read");
        assert_eq!(read_notice.link, Some(link_to_add.clone()));

        // 2. Dismissed notice
        let mut dismissed_notice = Notice::warn("task:2", "held");
        dismissed_notice.dismiss(now);
        let link_to_add_2 = Link::Task { id: "2".to_owned() };
        let again_2 = Notice::warn("task:2", "held").link(link_to_add_2.clone());
        let relit_2 = dismissed_notice.raise_again(&again_2, now);
        assert!(!relit_2, "link-only update must not page/relight");
        assert!(dismissed_notice.dismissed_at.is_some(), "tombstone holds");
        assert_eq!(dismissed_notice.link, Some(link_to_add_2));
    }

    #[test]
    fn identical_raises_share_one_file() {
        let (_d, s) = store();
        for _ in 0..5 {
            s.raise(Notice::error("run:abc", "blocked")).unwrap();
        }
        let all = s.list();
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].count, 5);
        assert_eq!(s.count_unread(), 1);
    }

    #[test]
    fn transitions_persist_and_dismissed_leave_the_list() {
        let (_d, s) = store();
        let a = s.raise(Notice::info("a", "one")).unwrap();
        let b = s.raise(Notice::warn("b", "two")).unwrap();
        assert_eq!(s.count_unread(), 2);
        s.mark_read(&a.id).unwrap();
        assert_eq!(s.count_unread(), 1);
        s.dismiss(&b.id).unwrap();
        assert_eq!(s.count_unread(), 0);
        assert_eq!(s.list().len(), 1);
        s.raise(Notice::warn("b", "two")).unwrap();
        assert_eq!(
            s.list().len(),
            1,
            "a tombstone survives a same-message raise"
        );
        s.raise(Notice::info("c", "three")).unwrap();
        assert_eq!(s.mark_all_read().unwrap(), 1);
        assert_eq!(s.count_unread(), 0);
    }

    #[test]
    fn the_list_is_newest_first() {
        let (_d, s) = store();
        let mut old = Notice::info("old", "old");
        old.last_at = "2020-01-01T00:00:00Z".parse().unwrap();
        s.put(&old).unwrap();
        s.raise(Notice::info("new", "new")).unwrap();
        let keys: Vec<_> = s.list().into_iter().map(|n| n.key).collect();
        assert_eq!(keys, ["new", "old"]);
    }

    #[test]
    fn the_cap_prunes_dismissed_then_read_then_oldest() {
        let (_d, s) = store();
        let keep = s.raise(Notice::error("keep", "unread")).unwrap();
        let gone = s.raise(Notice::info("gone", "dismissed")).unwrap();
        s.dismiss(&gone.id).unwrap();
        for i in 0..CAP - 1 {
            s.raise(Notice::info(&format!("k{i}"), "x")).unwrap();
        }
        let files = std::fs::read_dir(s.root()).unwrap().count();
        assert_eq!(files, CAP);
        assert!(
            s.get(&keep.id).is_ok(),
            "an unread notice outlives a dismissed one"
        );
        assert!(s.get(&gone.id).is_err(), "the dismissed one went first");
    }

    #[test]
    fn revision_moves_on_write_and_on_removal() {
        let (_d, s) = store();
        assert_eq!(s.revision(), 0);
        let a = s.raise(Notice::info("a", "m")).unwrap();
        let r1 = s.revision();
        assert_ne!(r1, 0);
        s.raise(Notice::info("b", "m")).unwrap();
        let r2 = s.revision();
        assert_ne!(r1, r2);
        std::fs::remove_file(s.path_of(&a.id)).unwrap();
        assert_ne!(s.revision(), r2);
    }

    #[test]
    fn ids_are_stable_and_untrusted_ids_are_refused() {
        assert_eq!(id_of("run:1"), id_of("run:1"));
        assert_ne!(id_of("run:1"), id_of("run-1"));
        assert!(valid_id(&id_of("release-bump:20260101-abc")));
        let (_d, s) = store();
        for bad in ["", "../x", "a/b", "A", "x.json"] {
            assert!(s.get(bad).is_err(), "{bad}");
        }
    }

    #[test]
    fn a_newer_schema_is_refused_and_an_older_reads() {
        let (_d, s) = store();
        let mut n = Notice::info("k", "m");
        n.schema = SCHEMA + 1;
        s.put(&n).unwrap();
        assert!(s.get(&n.id).is_err());
        let old = r#"{"id":"x-1","key":"x","severity":"warn","message":"m",
            "first_at":"2020-01-01T00:00:00Z","last_at":"2020-01-01T00:00:00Z"}"#;
        std::fs::write(s.path_of("x-1"), old).unwrap();
        assert_eq!(s.get("x-1").unwrap().count, 1);
    }

    fn state(status: crate::run::RunStatus) -> crate::run::RunState {
        let mut st = crate::run::RunState::new(
            std::path::PathBuf::from("/repo"),
            "main".to_owned(),
            "abc1234def".to_owned(),
            "task".to_owned(),
            crate::config::Config::default(),
        );
        st.status = status;
        st
    }

    #[test]
    fn a_run_that_ended_badly_is_news_unless_it_only_parked() {
        use crate::run::RunStatus;
        for bad in [RunStatus::Blocked, RunStatus::Stalled, RunStatus::Failed] {
            let st = state(bad);
            let n = run_ended(&st).expect("news");
            assert_eq!(n.key, format!("run:{}", st.id));
            assert_eq!(n.severity, Severity::Error);
        }
        assert!(run_ended(&state(RunStatus::Merged)).is_none());
        let mut parked = state(RunStatus::Stalled);
        parked.parked = true;
        assert!(run_ended(&parked).is_none());
    }

    #[test]
    fn concurrent_writers_of_one_key_all_succeed() {
        let (_d, s) = store();
        let handles: Vec<_> = (0..8)
            .map(|_| {
                let s = s.clone();
                std::thread::spawn(move || {
                    for _ in 0..20 {
                        s.raise(Notice::warn("same", "m")).unwrap();
                    }
                })
            })
            .collect();
        for h in handles {
            h.join().unwrap();
        }
        assert_eq!(s.list().len(), 1);
        let stray = std::fs::read_dir(s.root())
            .unwrap()
            .flatten()
            .filter(|e| e.path().extension().is_some_and(|x| x == "tmp"))
            .count();
        assert_eq!(stray, 0);
    }

    #[test]
    fn a_dead_holders_lock_is_broken_and_no_lock_is_left_behind() {
        let (_d, s) = store();
        let n = s.raise(Notice::info("k", "m")).unwrap();
        let lock = s.root().join(format!("{}.lock", n.id));
        std::fs::write(&lock, "").unwrap();
        let old = std::time::SystemTime::now() - std::time::Duration::from_secs(60);
        std::fs::File::options()
            .write(true)
            .open(&lock)
            .unwrap()
            .set_modified(old)
            .unwrap();
        s.mark_read(&n.id).unwrap();
        assert!(!lock.exists());
        assert!(s.get(&n.id).unwrap().read_at.is_some());
    }

    #[test]
    fn only_a_machine_hold_is_a_task_notice() {
        let mut t = crate::queue::Task::new(
            "t".to_owned(),
            "do it".to_owned(),
            std::path::PathBuf::from("/repo"),
            crate::queue::Source::Human,
        );
        assert!(task_held(&t).is_none());
        t.hold_manual(None);
        assert!(task_held(&t).is_none(), "the operator's own hold");
        t.hold_machine(Some("missing blocker".to_owned()));
        let n = task_held(&t).expect("machine hold");
        assert_eq!(n.key, format!("task:{}", t.id));
        assert!(n.message.contains("missing blocker"));
    }

    fn held_task() -> crate::queue::Task {
        let mut t = crate::queue::Task::new(
            "t".to_owned(),
            "do it".to_owned(),
            std::path::PathBuf::from("/repo"),
            crate::queue::Source::Human,
        );
        t.hold_machine(Some("branch b is checked out".to_owned()));
        t
    }

    fn ask(home: &Path, run: &str, node: &str) -> crate::ask::Question {
        let mut q = crate::ask::Question::new(
            run.to_owned(),
            node.to_owned(),
            "conductor".to_owned(),
            "cannot resume".to_owned(),
            String::new(),
            Vec::new(),
        );
        crate::ask::Questions::at(home.join("questions"))
            .put(&mut q)
            .unwrap();
        q
    }

    fn unread(home: &Path) -> usize {
        Notices::at(home.join("notifications")).count_unread()
    }

    #[test]
    fn a_hold_with_an_open_question_for_the_task_pages_once() {
        let dir = tempfile::tempdir().unwrap();
        let mut t = held_task();
        let q = ask(dir.path(), &t.id, "conduct");
        crate::queue::Queue::at(dir.path().join("queue"))
            .put(&mut t)
            .unwrap();
        let s = Notices::at(dir.path().join("notifications"));
        assert_eq!(s.count_unread(), 0);
        let n = s.list().pop().expect("the record is kept");
        assert_eq!(n.covered_by.as_deref(), Some(q.id.as_str()));
        assert!(n.message.contains("checked out"));
    }

    #[test]
    fn a_refused_handover_without_a_question_pages_once() {
        let dir = tempfile::tempdir().unwrap();
        let q = crate::queue::Queue::at(dir.path().join("queue"));
        let mut t = held_task();
        t.hold_for_handover(Some("magi/x/A".to_owned()), "checked out".to_owned());
        q.put(&mut t).unwrap();
        q.put(&mut t).unwrap();
        let s = Notices::at(dir.path().join("notifications"));
        assert_eq!(s.list().len(), 1);
        assert_eq!(s.list()[0].key, format!("task:{}", t.id));
        assert_eq!(s.count_unread(), 1);
    }

    #[test]
    fn an_older_question_does_not_silence_a_new_hold() {
        for node in ["deps", "conduct", "triage"] {
            let dir = tempfile::tempdir().unwrap();
            let mut t = crate::queue::Task::new(
                "t".to_owned(),
                "do it".to_owned(),
                std::path::PathBuf::from("/repo"),
                crate::queue::Source::Human,
            );
            let old = ask(dir.path(), &t.id, node);
            std::thread::sleep(std::time::Duration::from_millis(5));
            t.hold_machine(Some("new failure".to_owned()));
            crate::queue::Queue::at(dir.path().join("queue"))
                .put(&mut t)
                .unwrap();
            let s = Notices::at(dir.path().join("notifications"));
            let n = s.list().pop().unwrap();
            assert_eq!(n.covered_by, None, "{node} question {}", old.id);
            assert_eq!(s.count_unread(), 1);
        }
    }

    #[test]
    fn a_legacy_covered_notice_pages_for_a_new_uncovered_hold() {
        let (_d, s) = store();
        let mut legacy = Notice::warn("task:x", "held");
        legacy.covered_by = Some("q1".to_owned());
        let id = legacy.id.clone();
        s.raise(legacy).unwrap();
        assert!(!s.get(&id).unwrap().unread());
        s.raise(Notice::warn("task:x", "held").since(Some(Timestamp::now())))
            .unwrap();
        let n = s.get(&id).unwrap();
        assert!(n.unread());
        assert_eq!(n.covered_by, None);
    }

    #[test]
    fn a_newer_cause_relights_a_dismissed_notice() {
        let (_d, s) = store();
        let t0 = Timestamp::now();
        let first = Notice::warn("task:x", "held").since(Some(t0));
        let id = first.id.clone();
        s.raise(first).unwrap();
        s.dismiss(&id).unwrap();
        s.raise(Notice::warn("task:x", "held").since(Some(t0)))
            .unwrap();
        assert!(!s.get(&id).unwrap().unread(), "same cause stays dismissed");
        let later = t0 + std::time::Duration::from_secs(1);
        s.raise(Notice::warn("task:x", "held").since(Some(later)))
            .unwrap();
        assert!(s.get(&id).unwrap().unread());
    }

    #[test]
    fn a_hold_with_no_question_still_notifies() {
        let dir = tempfile::tempdir().unwrap();
        let mut t = held_task();
        crate::queue::Queue::at(dir.path().join("queue"))
            .put(&mut t)
            .unwrap();
        assert_eq!(unread(dir.path()), 1);
    }

    #[test]
    fn a_question_for_another_task_does_not_suppress() {
        let dir = tempfile::tempdir().unwrap();
        let mut t = held_task();
        ask(dir.path(), "some-other-task", "conduct");
        crate::queue::Queue::at(dir.path().join("queue"))
            .put(&mut t)
            .unwrap();
        assert_eq!(unread(dir.path()), 1);
    }

    #[test]
    fn a_question_filed_after_the_hold_quiets_it_once() {
        let dir = tempfile::tempdir().unwrap();
        let mut t = held_task();
        crate::queue::Queue::at(dir.path().join("queue"))
            .put(&mut t)
            .unwrap();
        assert_eq!(unread(dir.path()), 1);
        let q = ask(dir.path(), &t.id, "conduct");
        assert_eq!(unread(dir.path()), 0);
        // A later update of the same question does not run the hook again.
        let s = Notices::at(dir.path().join("notifications"));
        let id = s.list()[0].id.clone();
        s.update(&id, |n| n.read_at = None).unwrap();
        let mut again = crate::ask::Questions::at(dir.path().join("questions"))
            .get(&q.id)
            .unwrap();
        crate::ask::Questions::at(dir.path().join("questions"))
            .put(&mut again)
            .unwrap();
        assert_eq!(unread(dir.path()), 1);
    }

    #[test]
    fn a_blocked_run_with_an_open_question_is_quiet() {
        let dir = tempfile::tempdir().unwrap();
        ask(dir.path(), "run-1", "implement");
        raise_in(
            dir.path(),
            Notice::error("run:run-1", "Run r ended blocked.").about(["run-1"]),
        );
        assert_eq!(unread(dir.path()), 0);
        raise_in(
            dir.path(),
            Notice::error("run:run-2", "Run r ended blocked.").about(["run-2"]),
        );
        assert_eq!(unread(dir.path()), 1);
    }

    #[test]
    fn escalation_makes_a_covered_notice_unread_again() {
        let dir = tempfile::tempdir().unwrap();
        ask(dir.path(), "t1", "conduct");
        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
        assert_eq!(unread(dir.path()), 0);
        raise_in(dir.path(), Notice::error("task:t1", "held").about(["t1"]));
        assert_eq!(unread(dir.path()), 1);
    }

    #[test]
    fn covers_matches_run_exactly_and_ignores_release_questions() {
        let dir = tempfile::tempdir().unwrap();
        let n = Notice::warn("task:x", "m").about(["t1"]);
        let q = ask(dir.path(), "t1", "conduct");
        assert!(covers(&q, &n));
        assert!(!covers(&q, &Notice::warn("task:x", "m")));
        assert!(!covers(&q, &Notice::warn("task:x", "m").about(["t"])));
        let release = ask(dir.path(), "t1", crate::bump::NOTICE_NODE);
        assert!(!covers(&release, &n));
    }

    #[test]
    fn a_run_question_does_not_swallow_a_task_hold_and_vice_versa() {
        let dir = tempfile::tempdir().unwrap();
        ask(dir.path(), "t1", "implement");
        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
        assert_eq!(unread(dir.path()), 1);
        ask(dir.path(), "r1", "conduct");
        raise_in(dir.path(), Notice::error("run:r1", "ended").about(["r1"]));
        assert_eq!(unread(dir.path()), 2);
    }

    #[test]
    fn a_conduct_question_covers_the_hold_however_late_it_was_filed() {
        let dir = tempfile::tempdir().unwrap();
        let mut q = crate::ask::Question::new(
            "t1".to_owned(),
            "conduct".to_owned(),
            "conductor".to_owned(),
            "s".to_owned(),
            String::new(),
            Vec::new(),
        );
        q.asked_at = Timestamp::from_second(Timestamp::now().as_second() - 3600).unwrap();
        crate::ask::Questions::at(dir.path().join("questions"))
            .put(&mut q)
            .unwrap();
        raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
        assert_eq!(unread(dir.path()), 0);
    }

    fn pages(home: &Path, n: Notice) -> Vec<Page> {
        let sent = std::cell::RefCell::new(Vec::new());
        raise_in_with(home, n, &|p| sent.borrow_mut().push(p));
        sent.into_inner()
    }

    #[test]
    fn a_page_fires_on_new_causes_and_escalation_only() {
        let dir = tempfile::tempdir().unwrap();
        let h = dir.path();
        let n = || Notice::warn("run:p1", "stopped").link(Link::Run { id: "p1".into() });
        let first = pages(h, n());
        assert_eq!(
            first,
            vec![Page {
                run: "p1".into(),
                summary: "stopped".into()
            }]
        );
        assert!(pages(h, n()).is_empty(), "identical re-raise is silent");
        assert!(pages(h, Notice::warn("run:p1", "stopped again")).is_empty());
        assert_eq!(pages(h, Notice::error("run:p1", "stopped again")).len(), 1);
        assert!(pages(h, Notice::error("run:p1", "stopped again")).is_empty());
    }

    #[test]
    fn a_non_run_link_leaves_run_empty() {
        let dir = tempfile::tempdir().unwrap();
        let n = Notice::warn("task:t1", "held").link(Link::Task { id: "t1".into() });
        assert_eq!(pages(dir.path(), n)[0].run, "");
    }

    #[test]
    fn a_dismissed_tombstone_ignores_wording_changes() {
        let dir = tempfile::tempdir().unwrap();
        let h = dir.path();
        pages(h, Notice::warn("disk:r", "low"));
        Notices::at(h.join("notifications"))
            .dismiss(&id_of("disk:r"))
            .unwrap();
        assert!(pages(h, Notice::warn("disk:r", "low")).is_empty());
        assert!(pages(h, Notice::warn("disk:r", "lower")).is_empty());
    }

    #[test]
    fn a_newer_cause_pages_again() {
        let dir = tempfile::tempdir().unwrap();
        let h = dir.path();
        let at = |s: i64| Some(Timestamp::from_second(s).unwrap());
        let n = |s| Notice::warn("task:t2", "held").since(at(s));
        assert_eq!(pages(h, n(100)).len(), 1);
        assert!(pages(h, n(100)).is_empty());
        assert_eq!(pages(h, n(200)).len(), 1);
    }

    #[test]
    fn a_covered_notice_does_not_page() {
        let dir = tempfile::tempdir().unwrap();
        let h = dir.path();
        let q = ask(h, "r9", "implement");
        let n = || Notice::error("run:r9", "ended").about(["r9".to_owned()]);
        assert!(pages(h, n()).is_empty());
        let stored = Notices::at(h.join("notifications"))
            .get(&id_of("run:r9"))
            .unwrap();
        assert_eq!(stored.covered_by.as_deref(), Some(q.id.as_str()));
        assert!(pages(h, n()).is_empty());
    }

    #[test]
    fn a_failing_sender_path_never_propagates() {
        // A pager that cannot run its command only warns.
        let notify = crate::config::Notify {
            command: vec!["magi-no-such-program-xyz".into()],
        };
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let r = rt.block_on(crate::ask::notify_text(&notify, "", "x"));
        assert!(r.is_err());
        // `send_page` without an installed pager is a no-op and cannot panic.
        send_page(
            Page {
                run: String::new(),
                summary: "x".into(),
            },
            Some(notify),
        );
    }

    #[test]
    fn formatting_helpers_clean_text_and_lists() {
        assert_eq!(clean_text(None), None);
        assert_eq!(clean_text(Some("")), None);
        assert_eq!(clean_text(Some("   ")), None);
        assert_eq!(clean_text(Some("(no reason given)")), None);
        assert_eq!(clean_text(Some("(none)")), None);
        assert_eq!(clean_text(Some("  real reason  ")), Some("real reason"));

        let empty: Vec<String> = Vec::new();
        assert_eq!(format_list(&empty), None);
        assert_eq!(format_list(&["", "  "]), None);
        assert_eq!(format_list(&["task-1"]), Some("task-1".into()));
        assert_eq!(format_list(&["a", "b", "c"]), Some("a, b, c".into()));

        assert_eq!(ensure_sentence(""), "");
        assert_eq!(ensure_sentence("hello"), "hello.");
        assert_eq!(ensure_sentence("hello."), "hello.");
        assert_eq!(ensure_sentence("hello.."), "hello.");
    }

    #[test]
    fn conductor_resumption_notice_is_readable_and_non_soliciting() {
        let task_id = "20261017-000000-abcd";
        let task_short = "abcd";
        let reason = "awaiting manual review";
        let blocked_by = vec!["dep-1".to_owned(), "dep-2".to_owned()];
        let evidence = vec!["check-failed".to_owned()];
        let has_question = true;

        let notice = conductor_resumption(
            task_id,
            task_short,
            Some(reason),
            &blocked_by,
            &evidence,
            has_question,
        );

        assert_eq!(notice.key, format!("conductor-resumption:{task_id}"));
        assert_eq!(notice.severity, Severity::Warn);
        assert_eq!(
            notice.link,
            Some(Link::Task {
                id: task_id.to_owned()
            })
        );
        assert_eq!(notice.subjects, vec![task_id.to_owned()]);

        let msg = &notice.message;
        // Contains concise understandable prose explaining what happened
        assert!(msg.contains("Conductor raised a concern for task abcd after explicit resumption"));
        assert!(msg.contains("Concern: awaiting manual review."));
        assert!(msg.contains("Proposed dependencies: dep-1, dep-2."));
        assert!(msg.contains("Cited evidence: check-failed."));
        // Explicitly clarifies that no reply is required
        assert!(msg.contains("No reply to this notice is needed."));
        // No raw Debug formatting or dumped structures
        assert!(!msg.contains("Some("));
        assert!(!msg.contains("None"));
        assert!(!msg.contains("[]"));
        assert!(!msg.contains("{:?}"));
    }

    #[test]
    fn conductor_resumption_notice_omits_empty_details() {
        let task_id = "20261017-000000-abcd";
        let notice = conductor_resumption(task_id, "abcd", None, &[], &[], false);

        let msg = &notice.message;
        assert!(!msg.contains("Concern:"));
        assert!(!msg.contains("Proposed dependencies:"));
        assert!(!msg.contains("Cited evidence:"));
        assert!(!msg.contains("unregistered question"));
        assert!(!msg.contains("None"));
        assert!(!msg.contains("[]"));
        assert!(msg.contains("No reply to this notice is needed."));
    }
}