zenkey-fleet 0.14.0

//! Conditions and the watchdog (#227) — transitions, not states.
//!
//! Three shipped features each hard-coded their own predicate over the
//! observation surface: `expect` (one window), `doctor --for` (five
//! checks), `cutover` (silence). This module is the one **closed vocabulary**
//! they were each a spelling of: [`Condition`], evaluated to three states,
//! never two (RFC 09 §5.1 O4/O6) — `ok` / `firing` / **`unobservable`**. The
//! third state is the reason this exists: an alerting tool that cannot say
//! *"I could not tell"* is the one that pages at 3am for a dropped buffer. A
//! drop under a completeness claim yields `unobservable`, never `ok`.
//!
//! The vocabulary is deliberately closed — no expressions, no templating, no
//! rules engine. A new condition is a new variant, argued for the way a new
//! doctor check id is.
//!
//! The semantic core is three tiny rules — [`judge_shortfall`],
//! [`judge_excess`], [`judge_silence`] — shared with [`crate::judge::expect`], so
//! the watchdog and the CI assertion cannot drift about what a drop means.
//! Since RFC 13 (v1.24; the material was RFC 09 §5.1 pre-v1.24) the rules
//! speak the four-pole [`Judgement`] core, and [`CondState`] is this
//! module's serde-stable **wire projection** of it — see its mapping doc.
//!
//! [`watchdog`] is the continuous observer over the vocabulary:
//! **foreground, explicitly launched, single-purpose, one process per
//! invocation, no shared state** — not the hidden, auto-started,
//! discovery-caching daemon the redesign ledger rejected
//! (`docs/redesign-2026-07.md` §6.1). It emits [`Transition`]s: one per
//! genuine state change, none per unchanged tick.

use std::collections::BTreeMap;
use std::time::Duration;

use crate::{Error, Result};

use crate::bus::monitor::SampleView;
use crate::bus::query::FleetAnswer;
use crate::model::decode::SchemaStore;
use crate::model::registry::SliceSet;
use crate::report::{CheckId, DoctorReport};
use crate::report::{CondState, Judgement, Transition, WatchdogSummary};
use sipper::{Straw, sipper};

/// The closed condition vocabulary (#227), over the existing observation
/// surface. Each variant names what *firing* means; the drop rules are in
/// the judge functions this module documents.
#[derive(Debug, Clone, PartialEq)]
pub enum Condition {
    /// Samples on `selector` rode above `hz` over the evaluation window.
    /// Firing is positive evidence, conclusive even under drops (a drop only
    /// hides more); `ok` under drops is unobservable — the true rate is
    /// higher than what was counted (O6).
    RateAbove { selector: String, hz: f64 },
    /// Samples on `selector` rode below `hz`. A shortfall under drops is
    /// unobservable — the dropped samples could have filled it (O6); enough
    /// observed is conclusive `ok` regardless.
    RateBelow { selector: String, hz: f64 },
    /// No sample matched `selector` for at least `for_s` seconds. Silence is
    /// a completeness claim — it counts what did NOT happen — so it is
    /// provable only over a drop-free span at least `for_s` long (O6), and
    /// only once the observer has watched that long (O4).
    SilentFor { selector: String, for_s: f64 },
    /// An observed payload on `selector` did not reach [`crate::Verdict::Valid`]
    /// (#159) — `Invalid` and `NotValidated` both count: asking for validity
    /// and getting "unknowable" is not valid. Scoped to what was observed
    /// and checked; the `ok` state claims "nothing checked failed", never
    /// "nothing invalid rode" — the drop count rides in the evidence.
    InvalidPayload { selector: String },
    /// An observed sample on `selector` did not ride its registry-declared
    /// QoS profile (RFC 04 §3). Same per-observed-sample scope as
    /// [`Condition::InvalidPayload`]; samples with no declared profile are
    /// unjudgeable and counted in the evidence, not the state.
    QosMismatch { selector: String },
    /// A doctor run reported at least one finding with this check id
    /// (the stable [`crate::report::CheckId`] vocabulary). A failed doctor run is
    /// unobservable for every doctor condition — never `ok`.
    DoctorCheck { check: CheckId },
    /// The origin holds no `alive` token on the liveliness roster
    /// (RFC 04 §5). A roster that could not be asked is unobservable —
    /// silence is not a verdict (RFC 05 §3.1).
    OriginDown { origin: String },
    /// The observer itself dropped samples this window (RFC 09 §5.1 O6) —
    /// self-knowledge, so never unobservable.
    Dropped,
    /// At least one alert document at or above `min` is firing under
    /// `selector` on the alert plane (RFC 04 §1.2, `…/state/*/alert/*`) —
    /// what the sensors already judged, seen by the watchdog (#463). Asked
    /// by a GET once per tick, never a subscription: a firing alert is
    /// republished only on a content change, so a subscribe-only rule
    /// started mid-outage would report `ok` forever — the bug this rule
    /// exists to catch, rebuilt inside it. One state per rule (a count and
    /// the first), like [`Condition::DoctorCheck`]; a notifier that routes
    /// each alert is `zenwatch`'s `alerts` rule. Content-agnostic: the
    /// sensor made the judgement, this reports that one exists. An ask that
    /// failed is unobservable — silence is not a verdict.
    AlertFiring { selector: String, min: AlertFloor },
}

/// The severity floor of an [`Condition::AlertFiring`] rule: an ordered
/// compare over the three severities the alert plane speaks
/// (`info < warning < critical`), default `warning`. A document whose
/// `severity` is missing or outside the three is counted only under the
/// `info` floor — the lowest bar admits every firing document; a higher one
/// admits only what says it clears it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum AlertFloor {
    Info,
    Warning,
    Critical,
}

impl AlertFloor {
    pub const ALL: [AlertFloor; 3] = [AlertFloor::Info, AlertFloor::Warning, AlertFloor::Critical];

    pub fn parse(token: &str) -> Option<AlertFloor> {
        AlertFloor::ALL.into_iter().find(|f| f.as_str() == token)
    }

    pub fn as_str(self) -> &'static str {
        match self {
            AlertFloor::Info => "info",
            AlertFloor::Warning => "warning",
            AlertFloor::Critical => "critical",
        }
    }

    /// Whether a document's `severity` clears this floor.
    fn admits(self, severity: Option<&str>) -> bool {
        match severity.and_then(AlertFloor::parse) {
            Some(s) => s >= self,
            None => self == AlertFloor::Info,
        }
    }
}

impl std::fmt::Display for AlertFloor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// One tick's ask of the alert plane for one `alert-firing` selector
/// (#463): the answers, or why there are none.
#[derive(Debug, Clone)]
pub struct AlertAsk {
    pub selector: String,
    pub outcome: std::result::Result<Vec<FleetAnswer>, String>,
}

/// The rule grammar, spelled once for the parse error and the docs.
const VOCABULARY: &str = "rate-above <SEL> <HZ> | rate-below <SEL> <HZ> | \
     silent-for <SEL> <SECS> | invalid-payload <SEL> | qos-mismatch <SEL> | \
     doctor <CHECK-ID> | origin-down <ORIGIN> | dropped | \
     alert-firing <SEL> [<MIN-SEVERITY>]";

impl Condition {
    /// Parse one rule: whitespace-separated, kind first (Zenoh key
    /// expressions cannot contain whitespace, so the split is unambiguous).
    /// The vocabulary is closed; anything else is an error that spells it.
    pub fn parse(rule: &str) -> Result<Condition> {
        let hz = |s: &str, kind: &str| -> Result<f64> {
            let v: f64 = s
                .parse()
                .map_err(|_| Error::unaskable(format!("{kind} {s:?}"), "is not a number"))?;
            if !v.is_finite() || v < 0.0 {
                return Err(Error::unaskable(
                    kind.to_string(),
                    "the threshold must be a finite non-negative number",
                ));
            }
            Ok(v)
        };
        let tokens: Vec<&str> = rule.split_whitespace().collect();
        Ok(match tokens.as_slice() {
            ["rate-above", sel, n] => Condition::RateAbove {
                selector: sel.to_string(),
                hz: hz(n, "rate-above")?,
            },
            ["rate-below", sel, n] => Condition::RateBelow {
                selector: sel.to_string(),
                hz: hz(n, "rate-below")?,
            },
            ["silent-for", sel, n] => {
                let for_s = hz(n, "silent-for")?;
                if for_s <= 0.0 {
                    return Err(Error::unaskable(
                        "silent-for",
                        "the span must be a positive number of seconds",
                    ));
                }
                Condition::SilentFor {
                    selector: sel.to_string(),
                    for_s,
                }
            }
            ["invalid-payload", sel] => Condition::InvalidPayload {
                selector: sel.to_string(),
            },
            ["qos-mismatch", sel] => Condition::QosMismatch {
                selector: sel.to_string(),
            },
            ["doctor", check] => {
                let Some(check) = CheckId::parse(check) else {
                    return Err(Error::unaskable(
                        format!("doctor {check:?}"),
                        format!(
                            "is not a check id — the stable vocabulary is: {}",
                            CheckId::ALL
                                .iter()
                                .map(|c| c.as_str())
                                .collect::<Vec<_>>()
                                .join(", ")
                        ),
                    ));
                };
                Condition::DoctorCheck { check }
            }
            ["origin-down", origin] => Condition::OriginDown {
                origin: origin.to_string(),
            },
            ["dropped"] => Condition::Dropped,
            ["alert-firing", sel] => Condition::AlertFiring {
                selector: sel.to_string(),
                min: AlertFloor::Warning,
            },
            ["alert-firing", sel, floor] => {
                let Some(min) = AlertFloor::parse(floor) else {
                    return Err(Error::unaskable(
                        format!("alert-firing {floor:?}"),
                        format!(
                            "is not a severity floor — one of {}",
                            AlertFloor::ALL
                                .iter()
                                .map(|f| f.as_str())
                                .collect::<Vec<_>>()
                                .join(", ")
                        ),
                    ));
                };
                Condition::AlertFiring {
                    selector: sel.to_string(),
                    min,
                }
            }
            _ => {
                return Err(Error::unaskable(
                    format!("{rule:?}"),
                    format!(
                        "is not a rule — the vocabulary is closed (no \
                         expressions, no templating): {VOCABULARY}"
                    ),
                ));
            }
        })
    }

    /// The wire selector this condition observes, when it observes one.
    pub fn selector(&self) -> Option<&str> {
        match self {
            Condition::RateAbove { selector, .. }
            | Condition::RateBelow { selector, .. }
            | Condition::SilentFor { selector, .. }
            | Condition::InvalidPayload { selector }
            | Condition::QosMismatch { selector } => Some(selector),
            _ => None,
        }
    }

    /// Judge one observation window. `None` for the conditions that are not
    /// window-scoped ([`Condition::DoctorCheck`], [`Condition::OriginDown`]).
    /// Judge this condition against everything one tick observed.
    ///
    /// **The single entry point**, and why `run_watchdog` has no `expect`s
    /// left (#352). The three judges below each returned `None` for the
    /// variants they do not own, which forced the caller to assert a
    /// partition the compiler could not see — four times, every one
    /// discharging the same claim. This match *is* the partition, and each
    /// arm hands its judge exactly the evidence that judge needs, so none of
    /// them has a `None` to return.
    pub fn judge(&self, ev: &TickEvidence<'_>) -> Eval {
        match self {
            Condition::DoctorCheck { check } => judge_doctor_check(*check, ev.doctor),
            Condition::OriginDown { origin } => judge_origin_down(origin, ev.roster),
            Condition::AlertFiring { selector, min } => {
                judge_alert_firing(ev.base, selector, *min, ev.alerts)
            }
            _ => self.judge_window_total(ev.window),
        }
    }

    pub fn judge_window(&self, w: &CondWindow) -> Option<Eval> {
        let synth = if w.synthetic > 0 {
            format!("; {} synthetic-marked (RFC 09 §5.3)", w.synthetic)
        } else {
            String::new()
        };
        let rate = if w.window_s > 0.0 {
            w.samples as f64 / w.window_s
        } else {
            0.0
        };
        Some(match self {
            Condition::RateAbove { hz, .. } => {
                let state = CondState::from(judge_excess(rate > *hz, w.dropped));
                let evidence = match state {
                    CondState::Unobservable => format!(
                        "{rate:.2} Hz observed but {} sample(s) dropped — the true rate \
                         is at least that, not exactly that (O6){synth}",
                        w.dropped
                    ),
                    _ => format!(
                        "{} sample(s) in {:.1}s = {rate:.2} Hz against the {hz:.2} Hz \
                         bound{synth}",
                        w.samples, w.window_s
                    ),
                };
                Eval { state, evidence }
            }
            Condition::RateBelow { hz, .. } => {
                let state = CondState::from(judge_shortfall(rate < *hz, w.dropped));
                let evidence = match state {
                    CondState::Unobservable => format!(
                        "{rate:.2} Hz observed with {} sample(s) dropped — the drops \
                         could have carried the difference (O6){synth}",
                        w.dropped
                    ),
                    _ => format!(
                        "{} sample(s) in {:.1}s = {rate:.2} Hz against the {hz:.2} Hz \
                         bound{synth}",
                        w.samples, w.window_s
                    ),
                };
                Eval { state, evidence }
            }
            Condition::SilentFor { for_s, .. } => {
                let ev = SilenceEvidence {
                    sample_within: w.last_sample_ago_s.map(|ago| ago < *for_s) == Some(true),
                    span_observed: w.observed_s >= *for_s,
                    drop_free: w.last_drop_ago_s.map(|ago| ago >= *for_s) != Some(false),
                };
                let SilenceEvidence { span_observed, .. } = ev;
                let state = CondState::from(judge_silence(ev));
                let evidence = match state {
                    CondState::Ok => format!(
                        "a sample rode {:.1}s ago, inside the {for_s:.1}s span{synth}",
                        w.last_sample_ago_s.unwrap_or(0.0)
                    ),
                    CondState::Firing => {
                        format!("no sample for {for_s:.1}s, on a drop-free observer{synth}")
                    }
                    CondState::Unobservable if !span_observed => format!(
                        "watched only {:.1}s of a {for_s:.1}s silence claim — not asked \
                         is not answered (O4){synth}",
                        w.observed_s
                    ),
                    CondState::Unobservable => format!(
                        "no sample seen, but the observer dropped inside the {for_s:.1}s \
                         span — silence is unprovable (O6){synth}"
                    ),
                };
                Eval { state, evidence }
            }
            Condition::InvalidPayload { .. } => Eval {
                state: if w.invalid > 0 {
                    CondState::Firing
                } else {
                    CondState::Ok
                },
                evidence: format!(
                    "{} of {} checked sample(s) did not reach Valid ({} observed, \
                     {} dropped{synth})",
                    w.invalid, w.checked, w.samples, w.dropped
                ),
            },
            Condition::QosMismatch { .. } => Eval {
                state: if w.qos_mismatched > 0 {
                    CondState::Firing
                } else {
                    CondState::Ok
                },
                evidence: format!(
                    "{} of {} judged sample(s) did not ride their declared profile \
                     ({} observed, {} with no declared profile to judge, \
                     {} dropped{synth})",
                    w.qos_mismatched,
                    w.qos_judged,
                    w.samples,
                    w.samples.saturating_sub(w.qos_judged),
                    w.dropped
                ),
            },
            Condition::Dropped => Eval {
                state: if w.dropped > 0 {
                    CondState::Firing
                } else {
                    CondState::Ok
                },
                evidence: format!(
                    "the observer dropped {} sample(s) in {:.1}s (O6){synth}",
                    w.dropped, w.window_s
                ),
            },
            Condition::DoctorCheck { .. }
            | Condition::OriginDown { .. }
            | Condition::AlertFiring { .. } => return None,
        })
    }

    /// [`judge_window`](Self::judge_window) for the variants that *have* a
    /// window — total, because [`judge`](Self::judge) has already routed the
    /// other two elsewhere.
    fn judge_window_total(&self, w: &CondWindow) -> Eval {
        debug_assert!(
            !matches!(
                self,
                Condition::DoctorCheck { .. } | Condition::OriginDown { .. }
            ),
            "judge() routes these two to their own evidence"
        );
        self.judge_window(w).unwrap_or_else(|| Eval {
            // Unreachable through `judge`; if some future variant reaches it,
            // "I have no window for this" is the honest answer, not a panic
            // in a watchdog that is supposed to keep running.
            state: CondState::Unobservable,
            evidence: "this rule is not judged against a sample window".into(),
        })
    }

    /// Judge a roster ask. `None` unless this is [`Condition::OriginDown`].
    /// `Err` is the ask failing, which is unobservable — silence is not a
    /// verdict (RFC 05 §3.1).
    pub fn judge_roster(
        &self,
        roster: Result<&BTreeMap<String, Vec<String>>, &str>,
    ) -> Option<Eval> {
        let Condition::OriginDown { origin } = self else {
            return None;
        };
        Some(match roster {
            Err(e) => Eval {
                state: CondState::Unobservable,
                evidence: format!("the roster could not be asked: {e}"),
            },
            Ok(r) => match r.get(origin) {
                Some(producers) => Eval {
                    state: CondState::Ok,
                    evidence: format!(
                        "{origin} holds an alive token ({} producer(s))",
                        producers.len()
                    ),
                },
                None => Eval {
                    state: CondState::Firing,
                    evidence: format!("{origin} holds no alive token (RFC 04 §5)"),
                },
            },
        })
    }

    /// Judge a doctor run. `None` unless this is [`Condition::DoctorCheck`].
    /// A failed run is unobservable for every doctor condition — never `ok`.
    pub fn judge_doctor(&self, outcome: Result<&DoctorReport, &str>) -> Option<Eval> {
        let Condition::DoctorCheck { check } = self else {
            return None;
        };
        Some(match outcome {
            Err(e) => Eval {
                state: CondState::Unobservable,
                evidence: format!("the doctor run failed: {e}"),
            },
            Ok(report) => {
                let mut hits = report.findings.iter().filter(|f| f.check == *check);
                match hits.next() {
                    Some(first) => Eval {
                        state: CondState::Firing,
                        evidence: format!(
                            "{} finding(s); first: {} — {}",
                            1 + hits.count(),
                            first.subject,
                            first.evidence
                        ),
                    },
                    None => Eval {
                        state: CondState::Ok,
                        evidence: format!("no {check} findings"),
                    },
                }
            }
        })
    }
}

impl std::fmt::Display for Condition {
    /// The canonical rule spelling — [`Condition::parse`] round-trips it,
    /// and it is the `rule` field of every [`Transition`].
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Condition::RateAbove { selector, hz } => write!(f, "rate-above {selector} {hz}"),
            Condition::RateBelow { selector, hz } => write!(f, "rate-below {selector} {hz}"),
            Condition::SilentFor { selector, for_s } => {
                write!(f, "silent-for {selector} {for_s}")
            }
            Condition::InvalidPayload { selector } => write!(f, "invalid-payload {selector}"),
            Condition::QosMismatch { selector } => write!(f, "qos-mismatch {selector}"),
            Condition::DoctorCheck { check } => write!(f, "doctor {check}"),
            Condition::OriginDown { origin } => write!(f, "origin-down {origin}"),
            Condition::Dropped => write!(f, "dropped"),
            Condition::AlertFiring { selector, min } => {
                write!(f, "alert-firing {selector} {min}")
            }
        }
    }
}

// ─── the judgement rules (the vocabulary's semantic core) ───────────────────
//
// The three judges return the four-pole [`Judgement`] core (RFC 13, v1.24;
// RFC 09 §5.1 pre-v1.24). None of them ever answers `NotAsked` — a judge is
// only called when the question was put — but the pole exists in the currency
// so a caller that *skipped* a judge can say so in the same vocabulary. The
// watchdog projects each judgement onto [`CondState`] for the wire.

/// The shortfall rule ([`Condition::RateBelow`]; `expect`'s count floor and
/// rate floor): too little was seen. Enough seen is conclusively clean even
/// under drops — a drop can only hide *more*. A shortfall with drops is
/// unobservable: the dropped samples could have filled it (RFC 09 §5.1 O6).
pub fn judge_shortfall(short: bool, dropped: u64) -> Judgement {
    match (short, dropped) {
        (false, _) => Judgement::NotEstablished {
            reason: "enough was seen — a drop only hides more".into(),
        },
        (true, 0) => Judgement::Established,
        (true, _) => Judgement::Unobservable {
            reason: format!("{dropped} dropped sample(s) could have filled the shortfall (O6)"),
        },
    }
}

/// The excess rule ([`Condition::RateAbove`]; `expect`'s rate ceiling): too
/// much was seen. An excess is positive evidence, conclusive under drops.
/// "No excess" is a completeness claim — it counts what did NOT happen — so
/// under drops it is unobservable, never clean (O6).
pub fn judge_excess(over: bool, dropped: u64) -> Judgement {
    match (over, dropped) {
        (true, _) => Judgement::Established,
        (false, 0) => Judgement::NotEstablished {
            reason: "no excess was counted, on a clean observation".into(),
        },
        (false, _) => Judgement::Unobservable {
            reason: format!(
                "{dropped} sample(s) dropped — \"did not exceed\" is a completeness \
                 claim (O6)"
            ),
        },
    }
}

/// The silence rule ([`Condition::SilentFor`]; `expect --absent`): a sample
/// inside the span conclusively breaks the silence; silence is provable only
/// over a span the observer actually watched (O4) drop-free (O6) — otherwise
/// unobservable, never clean.
/// What one silence claim rests on — three facts that are all `bool` and all
/// about the same span.
///
/// A struct rather than three positional parameters, because this feeds a
/// *judgement* and a transposition of two identically-typed booleans returns
/// a plausible wrong verdict with no compile error (#349).
/// `judge_shortfall`/`judge_excess` keep their positional `(bool, u64)` —
/// not transposable, so not a hazard.
#[derive(Debug, Clone, Copy)]
pub struct SilenceEvidence {
    /// A sample rode inside the claimed span — the conclusive break.
    pub sample_within: bool,
    /// The observer actually watched the whole span (O4). A span it did not
    /// watch is not a span it can call silent.
    pub span_observed: bool,
    /// The observer dropped nothing inside the span (O6). "Nothing arrived"
    /// under drops is a completeness claim the observation cannot carry.
    pub drop_free: bool,
}

pub fn judge_silence(ev: SilenceEvidence) -> Judgement {
    let SilenceEvidence {
        sample_within,
        span_observed,
        drop_free,
    } = ev;
    if sample_within {
        Judgement::NotEstablished {
            reason: "a sample rode inside the span".into(),
        }
    } else if span_observed && drop_free {
        Judgement::Established
    } else if !span_observed {
        Judgement::Unobservable {
            reason: "the observer has not watched the whole claimed span (O4)".into(),
        }
    } else {
        Judgement::Unobservable {
            reason: "the observer dropped inside the span — silence is unprovable (O6)".into(),
        }
    }
}

/// Everything one watchdog tick observed, in the three shapes the conditions
/// are judged against.
///
/// `doctor` and `roster` are `Option` because a tick only runs those asks if
/// some rule wants them — and "not run this tick" is *unobservable*, which is
/// the honest reading and the one the caller used to assert away with
/// `.expect("a doctor rule ran the doctor")` (#352).
pub struct TickEvidence<'e> {
    pub window: &'e CondWindow,
    pub doctor: Option<Result<&'e DoctorReport, &'e str>>,
    pub roster: Option<Result<&'e BTreeMap<String, Vec<String>>, &'e str>>,
    /// This tick's alert-plane asks, one per distinct `alert-firing`
    /// selector; `None` when no rule wanted one (#463).
    pub alerts: Option<&'e [AlertAsk]>,
    /// The deployment base the alert keys are read under.
    pub base: &'e str,
}

/// Judge one doctor check against this tick's run — total, and total in the
/// "did not run" direction too.
pub fn judge_doctor_check(check: CheckId, outcome: Option<Result<&DoctorReport, &str>>) -> Eval {
    let Some(outcome) = outcome else {
        return Eval {
            state: CondState::Unobservable,
            evidence: "the doctor did not run this tick".into(),
        };
    };
    Condition::DoctorCheck { check }
        .judge_doctor(outcome)
        .expect("a DoctorCheck is judged by the doctor")
}

/// Judge one origin against this tick's roster ask — likewise total.
pub fn judge_origin_down(
    origin: &str,
    roster: Option<Result<&BTreeMap<String, Vec<String>>, &str>>,
) -> Eval {
    let Some(roster) = roster else {
        return Eval {
            state: CondState::Unobservable,
            evidence: "the roster was not asked this tick".into(),
        };
    };
    Condition::OriginDown {
        origin: origin.to_string(),
    }
    .judge_roster(roster)
    .expect("an OriginDown is judged by the roster")
}

/// Judge one `alert-firing` rule against this tick's alert-plane asks
/// (#463) — likewise total. Each answer is one live alert document (a GET
/// returns no tombstones, so a resolved alert simply stops answering);
/// its `severity`, `rule` and `summary` are lifted the way `zenwatch`'s
/// `alerts` rule lifts them ([`crate::alert_transition`]), and nothing else
/// in it is read. The state is a count against the floor; the evidence
/// names the count and the first, so the line says what is wrong without
/// the watchdog knowing what a unit or a port is.
pub fn judge_alert_firing(
    base: &str,
    selector: &str,
    min: AlertFloor,
    asks: Option<&[AlertAsk]>,
) -> Eval {
    let Some(asks) = asks else {
        return Eval {
            state: CondState::Unobservable,
            evidence: "the alert plane was not asked this tick".into(),
        };
    };
    let Some(ask) = asks.iter().find(|a| a.selector == selector) else {
        return Eval {
            state: CondState::Unobservable,
            evidence: format!("no ask ran for {selector} this tick"),
        };
    };
    let answers = match &ask.outcome {
        Err(e) => {
            return Eval {
                state: CondState::Unobservable,
                evidence: format!("the alert plane could not be asked: {e}"),
            };
        }
        Ok(a) => a,
    };
    let mut documents = 0usize;
    let mut unreadable = 0usize;
    let mut firing = 0usize;
    let mut first: Option<String> = None;
    for answer in answers {
        let crate::bus::query::Answer::Value(bytes) = &answer.answer else {
            unreadable += 1;
            continue;
        };
        let doc = crate::model::decode::structural_value(&bytes.to_bytes());
        let transition = crate::model::alert::alert_transition(
            base,
            &answer.key,
            zenoh::sample::SampleKind::Put,
            doc.as_ref()
                .map(|v| (crate::report::RenderSource::Structural, v)),
            None,
            "",
        );
        let Some(t) = transition else {
            // Not an alert key at all — the selector was wider than the
            // plane. Counted, not judged.
            unreadable += 1;
            continue;
        };
        documents += 1;
        if !min.admits(t.severity.as_deref()) {
            continue;
        }
        firing += 1;
        if first.is_none() {
            let mut line = format!("{}/{}", t.origin, t.producer);
            if let Some(rule) = &t.rule {
                line.push(' ');
                line.push_str(rule);
            }
            if let Some(summary) = &t.summary {
                line.push_str(" — ");
                line.push_str(summary);
            }
            first = Some(line);
        }
    }
    let skipped = if unreadable > 0 {
        format!("; {unreadable} answer(s) not alert documents")
    } else {
        String::new()
    };
    if firing == 0 {
        Eval {
            state: CondState::Ok,
            evidence: format!(
                "no alert firing at >= {min} ({documents} document(s) read{skipped})"
            ),
        }
    } else {
        Eval {
            state: CondState::Firing,
            evidence: format!(
                "{firing} alert(s) firing at >= {min}; first: {}{skipped}",
                first.unwrap_or_default()
            ),
        }
    }
}

// ─── observations and evaluations ───────────────────────────────────────────

/// What one evaluation window observed on one condition's selector — the
/// facts, separated from the judgement so the judgement is pure.
///
/// `CondWindow` and not `Window`: this type is re-exported at the crate root
/// beside `BudgetWindow` and `RecordBounds`, and a bare `Window` there reads
/// as *the* window of an engine that has several. Nothing serializes the
/// name (the type carries no `Serialize`), so the rename is Rust-side only.
#[derive(Debug, Clone, Copy, Default)]
pub struct CondWindow {
    /// The span this window judges, seconds.
    pub window_s: f64,
    /// How long the observer has been watching in total — a claim about a
    /// span longer than this is unobservable (O4).
    pub observed_s: f64,
    /// Samples matching the selector within the window.
    pub samples: u64,
    /// Stream drops within the window — unattributable to any one selector,
    /// so they taint every completeness claim (O6).
    pub dropped: u64,
    /// Seconds since the last matching sample; `None` = none seen since the
    /// watch began.
    pub last_sample_ago_s: Option<f64>,
    /// Seconds since the last stream drop; `None` = the stream never dropped.
    pub last_drop_ago_s: Option<f64>,
    /// Samples whose payload did not reach `Valid`, among those checked.
    pub invalid: u64,
    /// Samples actually decode-checked (a budget bounds the cost).
    pub checked: u64,
    /// Samples that did not ride their declared QoS, among those judged.
    pub qos_mismatched: u64,
    /// Samples with a declared profile to judge against.
    pub qos_judged: u64,
    /// Samples carrying the RFC 09 §5.3 synthetic-traffic marker — generated
    /// traffic judged as real would be a self-inflicted page, so every
    /// evidence line carries the count.
    pub synthetic: u64,
}

/// One evaluation: the three-valued state, and the evidence for it.
#[derive(Debug, Clone, PartialEq)]
pub struct Eval {
    pub state: CondState,
    pub evidence: String,
}

/// One rule's transition detector: feed evaluations in, get a [`Transition`]
/// back **only** when the state genuinely changed. An unchanged tick returns
/// `None` — transitions, not states.
#[derive(Debug, Clone)]
pub struct RuleState {
    /// The condition itself, not its `Display`.
    ///
    /// It used to hold the rendered string and clone it into every
    /// transition, with the two representations kept equal only by a
    /// round-trip test — a second representation of a value that was
    /// `Clone` and in scope (#352). The rendering happens where the
    /// `Transition` is built, once, from the one source.
    rule: Condition,
    state: Option<CondState>,
}

impl RuleState {
    pub fn new(rule: Condition) -> RuleState {
        RuleState { rule, state: None }
    }

    /// The condition this state tracks.
    pub fn rule(&self) -> &Condition {
        &self.rule
    }

    /// The last observed state; `None` until the first evaluation.
    pub fn state(&self) -> Option<CondState> {
        self.state
    }

    /// Feed one evaluation. The first ever emits (from `null` — the baseline
    /// is said once); after that only a genuine change does.
    pub fn observe(&mut self, eval: Eval, at: impl Into<String>) -> Option<Transition> {
        if self.state == Some(eval.state) {
            return None;
        }
        let from = self.state;
        self.state = Some(eval.state);
        Some(Transition {
            rule: self.rule.to_string(),
            from,
            to: eval.state,
            at: at.into(),
            evidence: eval.evidence,
        })
    }
}

/// Run-over-run delta over a doctor report: one [`RuleState`] per stable
/// check id ([`CheckId`]), fed by `doctor --transitions`. The
/// first run states the baseline (one transition per check id); every later run yields
/// only genuine changes. A failed run flips every check to `unobservable` —
/// a doctor that could not run has not said the fleet is healthy.
#[derive(Debug, Clone)]
pub struct DoctorWatch {
    /// One state per check. A `Vec<(Condition, RuleState)>` until #352 — the
    /// condition was in both halves of the pair.
    checks: Vec<RuleState>,
}

impl DoctorWatch {
    pub fn new() -> DoctorWatch {
        DoctorWatch {
            checks: CheckId::ALL
                .iter()
                .map(|id| RuleState::new(Condition::DoctorCheck { check: *id }))
                .collect(),
        }
    }

    /// Feed one doctor run (or its failure) and collect the transitions.
    pub fn observe(&mut self, outcome: Result<&DoctorReport, &str>, at: &str) -> Vec<Transition> {
        self.checks
            .iter_mut()
            .filter_map(|state| {
                let Condition::DoctorCheck { check } = *state.rule() else {
                    // Unconstructible: `new` builds only `DoctorCheck`s.
                    return None;
                };
                let eval = judge_doctor_check(check, Some(outcome));
                state.observe(eval, at)
            })
            .collect()
    }
}

impl Default for DoctorWatch {
    fn default() -> Self {
        DoctorWatch::new()
    }
}

// ─── the watchdog runner ────────────────────────────────────────────────────

/// What a watchdog run watches, and for how long.
#[derive(Debug, Clone)]
pub struct WatchdogSpec {
    /// The rules, evaluated every tick.
    pub rules: Vec<Condition>,
    /// Evaluation cadence. A tick that runs long (a doctor rule's fan-in)
    /// slides rather than backlogs; windows are measured, not nominal.
    pub tick: Duration,
    /// Stop after this many ticks; `None` = run until the caller stops it.
    pub ticks: Option<u64>,
    /// Per-ask timeout for the roster and doctor conditions.
    pub timeout: Duration,
}

/// How many decode attempts each key gets per tick under an
/// `invalid-payload` rule — the same budget the doctor listen phase runs,
/// for the same reason: a watchdog must not become a load test.
const DECODE_BUDGET: u8 = 2;

/// What one tick counted on one rule's selector.
#[derive(Default, Clone, Copy)]
struct TickCounters {
    samples: u64,
    invalid: u64,
    checked: u64,
    qos_mismatched: u64,
    qos_judged: u64,
    synthetic: u64,
}

/// One rule's whole per-run state, together.
///
/// This was four `Vec`s held in lockstep by index — `states`,
/// `keyexprs`, `counters`, `last_sample` — across a hundred and thirty
/// lines, with nothing structurally preventing them from disagreeing in
/// length, and a `counters.fill(default())` reset that could silently
/// miss one of them (#352).
struct RuleRuntime {
    rule: Condition,
    /// The rule's selector, compiled once for sample attribution.
    keyexpr: Option<zenoh::key_expr::KeyExpr<'static>>,
    counters: TickCounters,
    last_sample: Option<tokio::time::Instant>,
    state: RuleState,
}

/// The sweep a tick ran beside its drain, as the rules see it: the doctor
/// run and the roster ask, each `None` when no rule wanted it — which is
/// *unobservable* for the rules that would have needed it, the honest
/// reading (#352).
#[derive(Debug, Clone, Copy, Default)]
pub struct SweepOutcome<'e> {
    pub doctor: Option<Result<&'e DoctorReport, &'e str>>,
    pub roster: Option<Result<&'e BTreeMap<String, Vec<String>>, &'e str>>,
    /// The alert-plane asks, one per distinct `alert-firing` selector
    /// (#463); `None` when no rule wanted one.
    pub alerts: Option<&'e [AlertAsk]>,
}

/// A set of rules judged tick by tick over **one** event stream — the
/// watchdog's per-tick body, lifted out of [`watchdog`] so a second driver
/// can run it (#218).
///
/// The driver owns the stream, the drain loop and the sweep; this owns
/// everything the rules know: per-rule counters, the last sample and drop
/// instants, the per-tick decode budget, and the transition detectors. Feed
/// it every sample ([`observe_sample`](Self::observe_sample)) and every drop
/// ([`observe_drop`](Self::observe_drop)) the stream yields, then
/// [`evaluate`](Self::evaluate) once per tick and get back only what changed.
///
/// **Why the seam exists.** A trigger capture (`zenctl record --on`) must
/// judge *the same event stream it records*: one subscription, one drop
/// ledger. Had the capture run a watchdog of its own beside its recorder,
/// the drops the judge saw and the drops in the file would have been two
/// different facts about two different observers — and a `{"dropped": n}`
/// in the file would say nothing about whether the rule that fired was
/// judged over a clean window. With the body a value, the recorder drains
/// one stream and hands every item to both the ring and the rules.
///
/// Sample attribution is by key-expression intersection against each rule's
/// selector; a sample whose key does not parse as one counts for no rule.
/// Time is `tokio::time::Instant`, so a driver under paused time judges
/// exact windows.
pub struct RuleSet<'a> {
    rules: Vec<RuleRuntime>,
    /// The distinct selectors the rules observe, in first-seen order.
    watched: Vec<String>,
    base: &'a str,
    slices: Option<&'a SliceSet>,
    started: tokio::time::Instant,
    last_eval: tokio::time::Instant,
    last_drop: Option<tokio::time::Instant>,
    dropped_tick: u64,
    decode_budget: BTreeMap<String, u8>,
    ticks: u64,
    transitions: u64,
}

impl<'a> RuleSet<'a> {
    /// Compile the rules. Fails on a selector that is not a key expression —
    /// before anything is declared, so the `?` has nothing to tear down
    /// (#336). The watch clock starts here: [`CondWindow::observed_s`] is
    /// measured from construction, so build the set right before the
    /// subscriptions are declared.
    pub fn new(rules: &[Condition], base: &'a str, slices: Option<&'a SliceSet>) -> Result<Self> {
        let compiled = rules
            .iter()
            .map(|rule| {
                Ok(RuleRuntime {
                    rule: rule.clone(),
                    keyexpr: rule
                        .selector()
                        .map(|sel| {
                            zenoh::key_expr::KeyExpr::try_from(sel.to_string())
                                .map_err(|e| Error::unaskable_from(format!("{sel:?}"), e))
                        })
                        .transpose()?,
                    counters: TickCounters::default(),
                    last_sample: None,
                    state: RuleState::new(rule.clone()),
                })
            })
            .collect::<Result<Vec<_>>>()?;
        let mut watched: Vec<String> = Vec::new();
        for rule in rules {
            if let Some(sel) = rule.selector()
                && !watched.iter().any(|s| s == sel)
            {
                watched.push(sel.to_string());
            }
        }
        let now = tokio::time::Instant::now();
        Ok(RuleSet {
            rules: compiled,
            watched,
            base,
            slices,
            started: now,
            last_eval: now,
            last_drop: None,
            dropped_tick: 0,
            decode_budget: BTreeMap::new(),
            ticks: 0,
            transitions: 0,
        })
    }

    /// The distinct selectors the rules observe — what the driver must
    /// subscribe to before the first window opens (O4).
    pub fn watched(&self) -> &[String] {
        &self.watched
    }

    /// Some rule judges a doctor run, so the driver owes one per tick.
    pub fn wants_doctor(&self) -> bool {
        self.rules
            .iter()
            .any(|r| matches!(r.rule, Condition::DoctorCheck { .. }))
    }

    /// Some rule judges the liveliness roster, so the driver owes one ask
    /// per tick.
    pub fn wants_roster(&self) -> bool {
        self.rules
            .iter()
            .any(|r| matches!(r.rule, Condition::OriginDown { .. }))
    }

    /// Some rule judges the alert plane, so the driver owes one GET per
    /// distinct selector per tick (#463).
    pub fn wants_alerts(&self) -> bool {
        self.rules
            .iter()
            .any(|r| matches!(r.rule, Condition::AlertFiring { .. }))
    }

    /// The distinct `alert-firing` selectors, in rule order — what the
    /// driver asks each tick.
    pub fn alert_selectors(&self) -> Vec<String> {
        let mut out: Vec<String> = Vec::new();
        for r in &self.rules {
            if let Condition::AlertFiring { selector, .. } = &r.rule
                && !out.iter().any(|s| s == selector)
            {
                out.push(selector.clone());
            }
        }
        out
    }

    /// Some rule judges payload validity, so the driver owes a warmed,
    /// sealed schema store (#337) and a decode per
    /// [`wants_verdict`](Self::wants_verdict).
    pub fn wants_decode(&self) -> bool {
        self.rules
            .iter()
            .any(|r| matches!(r.rule, Condition::InvalidPayload { .. }))
    }

    /// Whether this sample should be decoded before it is observed: an
    /// `invalid-payload` rule matches its key and the key's per-tick decode
    /// budget has room. Spends the budget — ask once per sample, then hand
    /// the verdict to [`observe_sample`](Self::observe_sample). The decode
    /// stays the driver's, because it is async and this is not.
    pub fn wants_verdict(&mut self, s: &SampleView) -> bool {
        let Ok(key) = zenoh::key_expr::KeyExpr::try_from(s.key.as_str()) else {
            return false;
        };
        let matched = self.rules.iter().any(|rt| {
            matches!(rt.rule, Condition::InvalidPayload { .. })
                && rt.keyexpr.as_ref().is_some_and(|sel| sel.intersects(&key))
        });
        if !matched {
            return false;
        }
        let budget = self.decode_budget.entry(s.key.clone()).or_default();
        if *budget < DECODE_BUDGET {
            *budget += 1;
            true
        } else {
            false
        }
    }

    /// Count one observed sample against every rule its key matches.
    /// `verdict` is the decode the driver ran when
    /// [`wants_verdict`](Self::wants_verdict) said so; `None` means the
    /// sample was not checked, which is counted as exactly that.
    pub fn observe_sample(
        &mut self,
        s: &SampleView,
        facts_cache: &mut crate::model::facts::FactsCache,
        verdict: Option<&crate::Verdict>,
    ) {
        let Ok(key) = zenoh::key_expr::KeyExpr::try_from(s.key.as_str()) else {
            return;
        };
        let synthetic = s
            .attachment
            .as_ref()
            .is_some_and(|a| crate::judge::common::is_synthetic_marker(&a.to_bytes()));
        let now = tokio::time::Instant::now();
        for rt in self.rules.iter_mut() {
            let Some(sel) = &rt.keyexpr else { continue };
            if !sel.intersects(&key) {
                continue;
            }
            rt.counters.samples += 1;
            if synthetic {
                rt.counters.synthetic += 1;
            }
            rt.last_sample = Some(now);
            match &rt.rule {
                Condition::InvalidPayload { .. } => {
                    // An `invalid-payload` rule counts every not-`Valid`
                    // verdict the same way, so with no registry loaded
                    // `NoRegistry` (#246) changes no transition — only the
                    // reason the sample was not validated.
                    if let Some(v) = verdict {
                        rt.counters.checked += 1;
                        if !matches!(v, crate::Verdict::Valid) {
                            rt.counters.invalid += 1;
                        }
                    }
                }
                Condition::QosMismatch { .. } => {
                    facts_cache.ensure(self.base, &s.key, self.slices);
                    let facts = facts_cache.get(&s.key).expect("just ensured this key");
                    if let crate::model::facts::Registration::Registered(sf) = &facts.registration
                        && let Some(profile) = sf.declared_qos()
                    {
                        rt.counters.qos_judged += 1;
                        if !s.qos_matches(profile) {
                            rt.counters.qos_mismatched += 1;
                        }
                    }
                }
                _ => {}
            }
        }
    }

    /// The stream dropped `n` samples here (RFC 09 §5.1 O6): unattributable
    /// to any one selector, so it taints every completeness claim this tick.
    pub fn observe_drop(&mut self, n: u64) {
        self.dropped_tick += n;
        self.last_drop = Some(tokio::time::Instant::now());
    }

    /// Close the tick: judge every rule over the window measured since the
    /// last evaluation, reset the per-tick counts, and hand back only the
    /// genuine changes — none for an unchanged rule. `at` is the wall-clock
    /// stamp the transitions carry.
    pub fn evaluate(
        &mut self,
        now: tokio::time::Instant,
        at: &str,
        sweep: SweepOutcome<'_>,
    ) -> Vec<Transition> {
        let mut out = Vec::new();
        for rt in self.rules.iter_mut() {
            let window = CondWindow {
                window_s: (now - self.last_eval).as_secs_f64(),
                observed_s: (now - self.started).as_secs_f64(),
                samples: rt.counters.samples,
                dropped: self.dropped_tick,
                last_sample_ago_s: rt.last_sample.map(|t| (now - t).as_secs_f64()),
                last_drop_ago_s: self.last_drop.map(|t| (now - t).as_secs_f64()),
                invalid: rt.counters.invalid,
                checked: rt.counters.checked,
                qos_mismatched: rt.counters.qos_mismatched,
                qos_judged: rt.counters.qos_judged,
                synthetic: rt.counters.synthetic,
            };
            let eval = rt.rule.judge(&TickEvidence {
                window: &window,
                doctor: sweep.doctor,
                roster: sweep.roster,
                alerts: sweep.alerts,
                base: self.base,
            });
            if let Some(transition) = rt.state.observe(eval, at) {
                out.push(transition);
            }
        }
        // One reset, over one collection — the four-`Vec` version had a
        // `counters.fill(..)` that could miss a sibling (#352).
        for rt in self.rules.iter_mut() {
            rt.counters = TickCounters::default();
        }
        self.dropped_tick = 0;
        self.decode_budget.clear();
        self.ticks += 1;
        self.transitions += out.len() as u64;
        self.last_eval = now;
        out
    }

    /// Ticks evaluated so far.
    pub fn ticks(&self) -> u64 {
        self.ticks
    }

    /// When the last tick closed (construction, before the first): the
    /// driver's next deadline is measured from here, so a slow consumer of
    /// the transitions widens the next window rather than skipping one.
    pub fn last_eval(&self) -> tokio::time::Instant {
        self.last_eval
    }

    /// Transitions emitted so far.
    pub fn transitions(&self) -> u64 {
        self.transitions
    }
}

/// Watch the rules and yield one [`Transition`] per genuine change, none per
/// unchanged tick. The subscriber set is declared before the first window
/// opens (O4); every selector rule is judged per tick over the measured
/// window, doctor and roster rules by one ask per tick each.
///
/// A driver over [`RuleSet`] (#218): this function owns the monitor, the
/// drain loop and the per-tick sweep; the rules' state is the set's.
///
/// A [`Straw`] rather than a [`Stream`](futures_core::Stream) (#397), because
/// a watchdog run is a sequence **and** a final value: transitions while it
/// runs, a [`WatchdogSummary`] when it stops, and the acknowledged monitor
/// teardown (#207/#336) in between. A bare `Stream` has room for the first
/// only — which is why this stayed a callback through #343, and why the
/// callback could not fail: `emit` was infallible by construction, so a
/// caller whose emission *could* fail had to stash the error and answer for
/// it after the run. Dropping it instead let `zenctl watchdog` finish clean
/// having emitted nothing (#360).
///
/// Drive it with `sip` for the transitions and `await` for the summary:
///
/// ```ignore
/// let mut run = watchdog(&fleet, slices, &store, &spec).pin();
/// while let Some(transition) = run.sip().await {
///     writeln!(out, "{}", serde_json::to_string(&transition)?)?;
/// }
/// let summary = run.await?;
/// ```
///
/// The summary is the *output*, not an item, so a consumer that stops sipping
/// early and awaits still gets the teardown — there is no `finish` to forget.
pub fn watchdog<'a>(
    fleet: &'a crate::Fleet<'a>,
    slices: Option<&'a SliceSet>,
    store: &'a SchemaStore,
    spec: &'a WatchdogSpec,
) -> impl Straw<WatchdogSummary, Transition, Error> + 'a {
    sipper(async move |mut sender: sipper::Sender<Transition>| {
        use crate::{FleetEvent, StreamItem};

        let (session, base) = (fleet.session(), fleet.base());

        // Compiled *before* the monitor exists, so the `?` has nothing to tear
        // down (#336).
        let mut rules = RuleSet::new(&spec.rules, base, slices)?;
        let (wants_doctor, wants_roster, wants_decode, wants_alerts) = (
            rules.wants_doctor(),
            rules.wants_roster(),
            rules.wants_decode(),
            rules.wants_alerts(),
        );
        let alert_selectors = rules.alert_selectors();

        // Warmed before the first tick and sealed for the run (#337): a decode
        // inside the drain loop must never become a `describe` GET, because
        // nothing attends the broadcast while one is in flight and the tick's
        // verdict is about the window that lost the samples. zenctl hands this
        // store over cold. Each tick's sweep re-warms whatever is still
        // unserved — from beside the drain, where waiting costs nothing.
        if wants_decode {
            crate::model::decode::prewarm(fleet, store, slices).await;
        }
        let _sealed = store.seal();

        // Declared before the window opens — not-asked must never read as "no".
        let monitor = crate::Monitor::start(session, crate::MonitorSpec::default()).await?;
        let mut events = monitor.events();
        let monitor = monitor.watching(rules.watched()).await?;

        // Bounded (#107): the watchdog runs until stopped, so an unbounded
        // per-key map here is a leak on any bus with churning keys. Evictions
        // ride the summary (O6).
        let mut facts_cache = crate::model::facts::FactsCache::default();

        let mut closed = false;
        loop {
            let deadline = rules.last_eval() + spec.tick;
            // The tick's bus work runs **beside** the drain, not after it (#338).
            //
            // A roster GET, a registry sweep, per-producer describes and state
            // snapshots take seconds, and every one of them used to happen with
            // the drain loop stopped — so the broadcast overflowed, and because
            // `dropped_tick` was reset immediately afterwards, the loss was
            // billed to the *following* window. In the one tool whose entire
            // product is a per-window verdict.
            //
            // Now the sweep is a future the drain selects on: sampling never
            // stops, and a sweep that outlives the tick period simply widens this
            // window — `window_s` is measured from the last evaluation, never
            // assumed — so the drops land in the tick that incurred them.
            let sweep = async {
                let doctor = if wants_doctor {
                    Some(
                        crate::judge::doctor::run_doctor(
                            fleet,
                            slices,
                            &crate::judge::doctor::DoctorSpec {
                                deep: false,
                                sample: None,
                                timeout: spec.timeout,
                                listen: None,
                            },
                        )
                        .await
                        .map_err(|e| e.to_string()),
                    )
                } else {
                    None
                };
                let roster = if wants_roster {
                    Some(
                        crate::bus::roster::roster(fleet, spec.timeout)
                            .await
                            .map_err(|e| e.to_string()),
                    )
                } else {
                    None
                };
                // The alert plane (#463): one bounded GET per distinct
                // selector, beside the roster ask. A GET, not a subscription
                // — a firing alert is republished only on a content change.
                let alerts = if wants_alerts {
                    let mut asks = Vec::with_capacity(alert_selectors.len());
                    for selector in &alert_selectors {
                        let outcome = crate::bus::query::fleet_get(
                            fleet,
                            selector,
                            &crate::bus::query::GetOpts::new(spec.timeout),
                        )
                        .await
                        .map_err(|e| e.to_string());
                        asks.push(AlertAsk {
                            selector: selector.clone(),
                            outcome,
                        });
                    }
                    Some(asks)
                } else {
                    None
                };
                // The schema warming rides here too (#337): still-unserved
                // producers are re-asked at the store's own backoff, off the
                // drain loop.
                if wants_decode {
                    crate::model::decode::prewarm(fleet, store, slices).await;
                }
                (doctor, roster, alerts)
            };
            let mut sweep = std::pin::pin!(sweep);
            let mut swept = None;
            // One timer per tick, not one per drained sample (#346).
            let tick_over = tokio::time::sleep_until(deadline);
            tokio::pin!(tick_over);
            while !closed {
                let item = tokio::select! {
                    item = events.recv() => item,
                    // The tick cannot close before its own sweep has landed, and
                    // the drain keeps running until it does.
                    outcome = &mut sweep, if swept.is_none() => {
                        swept = Some(outcome);
                        continue;
                    }
                    () = &mut tick_over, if swept.is_some() => break,
                };
                match item {
                    Some(StreamItem::Event(FleetEvent::Sample(s))) => {
                        // Decode once per sample (budgeted per key per tick),
                        // shared by every invalid-payload rule the key matches.
                        let verdict = if rules.wants_verdict(&s) {
                            Some(
                                crate::model::decode::decode_sample(
                                    fleet,
                                    store,
                                    slices,
                                    &s.key,
                                    Some(&s.encoding),
                                    &s.payload.to_bytes(),
                                )
                                .await
                                .verdict,
                            )
                        } else {
                            None
                        };
                        rules.observe_sample(&s, &mut facts_cache, verdict.as_ref());
                    }
                    Some(StreamItem::Dropped(n)) => rules.observe_drop(n),
                    Some(_) => {}
                    None => closed = true,
                }
            }

            // Evaluate the tick over the measured window, then say only what
            // changed. The sweep has already landed unless the stream closed
            // under it — in which case there is nothing left to drain, and
            // awaiting it here costs the tick nothing.
            let (doctor_outcome, roster_outcome, alert_asks) = match swept {
                Some(outcome) => outcome,
                None => sweep.await,
            };
            let now = tokio::time::Instant::now();
            let at = crate::tape::record::rfc3339_now();
            let transitions = rules.evaluate(
                now,
                &at,
                SweepOutcome {
                    doctor: doctor_outcome
                        .as_ref()
                        .map(|o| o.as_ref().map_err(String::as_str)),
                    roster: roster_outcome
                        .as_ref()
                        .map(|o| o.as_ref().map_err(String::as_str)),
                    alerts: alert_asks.as_deref(),
                },
            );
            for transition in transitions {
                // Awaits, where the callback returned: the consumer's write
                // now happens *here*, so its error returns from where it
                // happened instead of being stashed for after the run (#360).
                // The emission point is the tick evaluation — the drain loop
                // above has already ended for this tick — so a slow consumer
                // widens the next window rather than stalling a drain (#338).
                sender.send(transition).await;
            }
            if closed || spec.ticks.is_some_and(|n| rules.ticks() >= n) {
                break;
            }
        }
        monitor.shutdown().await?;
        Ok(WatchdogSummary {
            ticks: rules.ticks(),
            transitions: rules.transitions(),
            facts_evicted: facts_cache.evicted(),
        })
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::report::{DoctorFinding, DoctorSeverity};

    fn report_with(checks: &[CheckId]) -> DoctorReport {
        DoctorReport {
            findings: checks
                .iter()
                .map(|c| DoctorFinding {
                    severity: DoctorSeverity::Error,
                    check: *c,
                    subject: "s".into(),
                    evidence: "e".into(),
                    citation: None,
                })
                .collect(),
            synced: crate::report::Asked::NotAsked,
            introspect_answered: 0,
            live_producers: 0,
            describe_served: 0,
            describe_missing: 0,
            routers: 0,
            router_version: None,
            deep: false,
            observation: None,
        }
    }

    /// Every variant's canonical spelling parses back to itself, and a rule
    /// outside the vocabulary is an error that names the vocabulary — closed
    /// means closed.
    #[test]
    fn the_vocabulary_round_trips_and_is_closed() {
        let rules = [
            "rate-above v1/*/telemetry/** 5",
            "rate-below v1/h-aaaaaaaaaaaa/state/p/health 0.5",
            "silent-for v1/*/events/** 30",
            "invalid-payload v1/*/state/**",
            "qos-mismatch v1/*/telemetry/**",
            "doctor slice-sync",
            "origin-down h-aaaaaaaaaaaa",
            "dropped",
            "alert-firing v1/*/state/*/alert/* critical",
        ];
        for rule in rules {
            let parsed = Condition::parse(rule).expect(rule);
            assert_eq!(parsed.to_string(), rule, "canonical spelling round-trips");
        }
        // The floor defaults to `warning`, and the canonical spelling says so.
        let bare = Condition::parse("alert-firing v1/*/state/*/alert/*").expect("bare");
        assert_eq!(
            bare,
            Condition::AlertFiring {
                selector: "v1/*/state/*/alert/*".into(),
                min: AlertFloor::Warning
            }
        );
        assert_eq!(
            bare.to_string(),
            "alert-firing v1/*/state/*/alert/* warning"
        );
        let err = Condition::parse("alert-firing v1/*/state/*/alert/* urgent").unwrap_err();
        assert!(err.to_string().contains("info, warning, critical"), "{err}");
        let err = Condition::parse("if rate > 5 then page").unwrap_err();
        assert!(err.to_string().contains("closed"), "{err}");
        assert!(err.to_string().contains("rate-above"), "{err}");
        // A doctor rule outside the stable check-id vocabulary is refused at
        // parse, naming the vocabulary.
        let err = Condition::parse("doctor no-such-check").unwrap_err();
        assert!(err.to_string().contains("slice-sync"), "{err}");
    }

    fn alert_answer(key: &str, doc: &str) -> FleetAnswer {
        FleetAnswer {
            origin: key.split('/').nth(1).unwrap_or("").to_string(),
            key: key.to_string(),
            encoding: Some("application/json".into()),
            attachment: None,
            timestamp: None,
            answer: crate::bus::query::Answer::Value(zenoh::bytes::ZBytes::from(doc.as_bytes())),
        }
    }

    /// #463: one state per rule — a count against the floor and the first
    /// document named; a tombstone never answers a GET, so "resolved" is
    /// simply an answer that stopped coming.
    #[test]
    fn alert_firing_counts_against_the_floor_and_names_the_first() {
        let sel = "v1/*/state/*/alert/*";
        let asks = [AlertAsk {
            selector: sel.into(),
            outcome: Ok(vec![
                alert_answer(
                    "v1/h-3fa9c2d41b7e/state/systemd/alert/aaaaaaaaaaaaaaaa",
                    r#"{"severity":"critical","rule":"expect-service-active","summary":"expected service forgejo.service active"}"#,
                ),
                alert_answer(
                    "v1/h-3fa9c2d41b7e/state/netlink/alert/bbbbbbbbbbbbbbbb",
                    r#"{"severity":"info","rule":"link_flap","message":"eth0 flapped"}"#,
                ),
                // Not an alert document at all: the selector was wider than the plane.
                alert_answer("v1/h-3fa9c2d41b7e/state/netlink/health", r#"{"ok":true}"#),
            ]),
        }];
        let e = judge_alert_firing("", sel, AlertFloor::Critical, Some(&asks));
        assert_eq!(e.state, CondState::Firing);
        assert!(
            e.evidence.starts_with(
                "1 alert(s) firing at >= critical; first: h-3fa9c2d41b7e/systemd \
                 expect-service-active — expected service forgejo.service active"
            ),
            "{}",
            e.evidence
        );
        assert!(
            e.evidence.contains("1 answer(s) not alert documents"),
            "{}",
            e.evidence
        );
        // The default floor admits the critical one only; `info` admits both.
        assert_eq!(
            judge_alert_firing("", sel, AlertFloor::Warning, Some(&asks)).state,
            CondState::Firing
        );
        let all = judge_alert_firing("", sel, AlertFloor::Info, Some(&asks));
        assert!(
            all.evidence.starts_with("2 alert(s) firing at >= info"),
            "{}",
            all.evidence
        );
        // Nothing firing above the floor is `ok`, with the documents counted.
        let quiet = [AlertAsk {
            selector: sel.into(),
            outcome: Ok(vec![alert_answer(
                "v1/h-3fa9c2d41b7e/state/netlink/alert/bbbbbbbbbbbbbbbb",
                r#"{"severity":"info","rule":"link_flap"}"#,
            )]),
        }];
        let e = judge_alert_firing("", sel, AlertFloor::Warning, Some(&quiet));
        assert_eq!(e.state, CondState::Ok);
        assert_eq!(
            e.evidence,
            "no alert firing at >= warning (1 document(s) read)"
        );
        // Not asked, or asked and failed: unobservable, never ok.
        assert_eq!(
            judge_alert_firing("", sel, AlertFloor::Warning, None).state,
            CondState::Unobservable
        );
        let failed = [AlertAsk {
            selector: sel.into(),
            outcome: Err("timed out".into()),
        }];
        let e = judge_alert_firing("", sel, AlertFloor::Warning, Some(&failed));
        assert_eq!(e.state, CondState::Unobservable);
        assert!(e.evidence.contains("timed out"), "{}", e.evidence);
    }

    /// A document with no readable severity clears only the `info` floor.
    #[test]
    fn a_severity_the_plane_does_not_speak_clears_only_the_lowest_floor() {
        assert!(AlertFloor::Info.admits(None));
        assert!(AlertFloor::Info.admits(Some("weird")));
        assert!(!AlertFloor::Warning.admits(None));
        assert!(AlertFloor::Warning.admits(Some("warning")));
        assert!(AlertFloor::Warning.admits(Some("critical")));
        assert!(!AlertFloor::Critical.admits(Some("warning")));
    }

    /// The acceptance rule of #227: a drop under a completeness claim yields
    /// `unobservable`, **never** `ok` — across all three core judges, now
    /// spoken in the [`Judgement`] core and projected onto [`CondState`]
    /// (RFC 13, v1.24).
    #[test]
    fn a_drop_under_a_completeness_claim_is_unobservable_never_ok() {
        let wire = CondState::from;
        // Excess: the "did not exceed" side counts what did not happen.
        assert!(judge_excess(false, 1).is_unobservable());
        assert_eq!(wire(judge_excess(false, 0)), CondState::Ok);
        // …while firing is positive evidence, conclusive under drops.
        assert_eq!(judge_excess(true, 7), Judgement::Established);
        // Shortfall: the drops could have carried the difference.
        assert!(judge_shortfall(true, 1).is_unobservable());
        assert_eq!(judge_shortfall(true, 0), Judgement::Established);
        // …while "enough seen" is conclusive: a drop only hides more.
        assert_eq!(wire(judge_shortfall(false, 9)), CondState::Ok);
        // Silence: unprovable over a dropped or unwatched span. Named fields
        // rather than three bare `bool`s, which is the whole of #349 — read
        // the old spelling `judge_silence(false, true, false)` and say which
        // one was the drop.
        let silence = |sample_within, span_observed, drop_free| {
            judge_silence(SilenceEvidence {
                sample_within,
                span_observed,
                drop_free,
            })
        };
        assert!(silence(false, true, false).is_unobservable());
        assert!(silence(false, false, true).is_unobservable());
        assert_eq!(silence(false, true, true), Judgement::Established);
        assert_eq!(wire(silence(true, true, false)), CondState::Ok);
    }

    /// The wire projection's documented mapping, polarity note included:
    /// `NotEstablished` (established-clean) is `ok`, `Established` (the
    /// condition holds) is `firing`, and **both** unestablished poles land
    /// on `unobservable` — the wire cannot say more (RFC 13, v1.24).
    #[test]
    fn cond_state_is_the_documented_projection_of_the_judgement_core() {
        assert_eq!(CondState::from(Judgement::Established), CondState::Firing);
        assert_eq!(
            CondState::from(Judgement::NotEstablished {
                reason: "clean".into()
            }),
            CondState::Ok
        );
        assert_eq!(
            CondState::from(Judgement::NotAsked),
            CondState::Unobservable
        );
        assert_eq!(
            CondState::from(Judgement::Unobservable {
                reason: "drops".into()
            }),
            CondState::Unobservable
        );
    }

    /// The window judges apply those rules: `rate-above` firing survives
    /// drops, its ok does not; a young watch cannot claim silence.
    #[test]
    fn window_judgement_applies_the_drop_rules() {
        let rule = Condition::parse("rate-above k/** 1").unwrap();
        let base = CondWindow {
            window_s: 10.0,
            observed_s: 10.0,
            ..CondWindow::default()
        };
        let over = CondWindow {
            samples: 20,
            dropped: 5,
            ..base
        };
        assert_eq!(rule.judge_window(&over).unwrap().state, CondState::Firing);
        let under_dropped = CondWindow {
            samples: 2,
            dropped: 5,
            ..base
        };
        assert_eq!(
            rule.judge_window(&under_dropped).unwrap().state,
            CondState::Unobservable
        );

        let rule = Condition::parse("silent-for k/** 30").unwrap();
        let young = CondWindow {
            window_s: 5.0,
            observed_s: 5.0,
            ..CondWindow::default()
        };
        let eval = rule.judge_window(&young).unwrap();
        assert_eq!(eval.state, CondState::Unobservable);
        assert!(eval.evidence.contains("watched only"), "{}", eval.evidence);
        let silent = CondWindow {
            window_s: 5.0,
            observed_s: 60.0,
            ..CondWindow::default()
        };
        assert_eq!(rule.judge_window(&silent).unwrap().state, CondState::Firing);
        let recently_dropped = CondWindow {
            last_drop_ago_s: Some(10.0),
            ..silent
        };
        assert_eq!(
            rule.judge_window(&recently_dropped).unwrap().state,
            CondState::Unobservable
        );
        let spoken = CondWindow {
            samples: 1,
            last_sample_ago_s: Some(3.0),
            ..silent
        };
        assert_eq!(rule.judge_window(&spoken).unwrap().state, CondState::Ok);
    }

    /// The synthetic-traffic marker count (RFC 09 §5.3, the #162 rider)
    /// rides every window evidence line when present.
    #[test]
    fn synthetic_marked_samples_are_said_out_loud() {
        let rule = Condition::parse("rate-above k/** 0.1").unwrap();
        let w = CondWindow {
            window_s: 10.0,
            observed_s: 10.0,
            samples: 20,
            synthetic: 3,
            ..CondWindow::default()
        };
        let eval = rule.judge_window(&w).unwrap();
        assert!(
            eval.evidence.contains("3 synthetic-marked"),
            "{}",
            eval.evidence
        );
    }

    /// The transition machine: the first evaluation states the baseline
    /// (from `null`), an unchanged tick emits nothing, a genuine change
    /// emits exactly one line.
    #[test]
    fn transitions_fire_once_per_genuine_change_and_never_per_tick() {
        let eval = |state| Eval {
            state,
            evidence: "e".into(),
        };
        // The condition itself, not its rendering — which is the point of
        // #352: the two can no longer disagree.
        let mut rs = RuleState::new(Condition::Dropped);
        let first = rs.observe(eval(CondState::Ok), "t0").expect("baseline");
        assert_eq!(first.rule, "dropped", "the transition renders its rule");
        assert_eq!(first.from, None, "the baseline comes from null (O4)");
        assert_eq!(first.to, CondState::Ok);
        assert!(rs.observe(eval(CondState::Ok), "t1").is_none());
        assert!(rs.observe(eval(CondState::Ok), "t2").is_none());
        let change = rs.observe(eval(CondState::Firing), "t3").expect("a change");
        assert_eq!(change.from, Some(CondState::Ok));
        assert_eq!(change.to, CondState::Firing);
        assert!(rs.observe(eval(CondState::Firing), "t4").is_none());
    }

    /// The ndjson shape of a transition is a wire contract for scripts:
    /// `{"rule","from","to","at","evidence"}`, states snake_case, `from`
    /// null on the baseline.
    #[test]
    fn transition_json_shape_is_pinned() {
        let t = Transition {
            rule: "silent-for k/** 30".into(),
            from: None,
            to: CondState::Unobservable,
            at: "2026-08-22T00:00:00Z".into(),
            evidence: "watched only 5.0s of a 30.0s silence claim".into(),
        };
        assert_eq!(
            serde_json::to_value(&t).unwrap(),
            serde_json::json!({
                "rule": "silent-for k/** 30",
                "from": null,
                "to": "unobservable",
                "at": "2026-08-22T00:00:00Z",
                "evidence": "watched only 5.0s of a 30.0s silence claim",
            })
        );
        let t = Transition {
            from: Some(CondState::Ok),
            to: CondState::Firing,
            ..t
        };
        let json = serde_json::to_value(&t).unwrap();
        assert_eq!(json["from"], "ok");
        assert_eq!(json["to"], "firing");
    }

    /// `doctor --transitions`'s delta: the first run is a full baseline (every
    /// stable check id, once), an identical second run says nothing, a new
    /// finding transitions exactly its check — and a failed run flips every
    /// check to unobservable, never ok.
    #[test]
    fn doctor_watch_reports_deltas_not_states() {
        let mut watch = DoctorWatch::new();
        let clean = report_with(&[]);
        let baseline = watch.observe(Ok(&clean), "t0");
        assert_eq!(baseline.len(), CheckId::ALL.len());
        assert!(baseline.iter().all(|t| t.from.is_none()));
        assert!(baseline.iter().all(|t| t.to == CondState::Ok));

        assert!(
            watch.observe(Ok(&clean), "t1").is_empty(),
            "an unchanged run emits nothing"
        );

        let drifted = report_with(&[CheckId::SchemaDrift, CheckId::SchemaDrift]);
        let changes = watch.observe(Ok(&drifted), "t2");
        assert_eq!(changes.len(), 1, "only the changed check transitions");
        assert_eq!(changes[0].rule, "doctor schema-drift");
        assert_eq!(changes[0].to, CondState::Firing);
        assert!(changes[0].evidence.contains("2 finding(s)"));

        let failed = watch.observe(Err("session lost"), "t3");
        assert_eq!(
            failed.len(),
            CheckId::ALL.len(),
            "a failed run is unobservable for every check — never ok"
        );
        assert!(failed.iter().all(|t| t.to == CondState::Unobservable));
    }
}