Skip to main content

zenkey_fleet/judge/
doctor.rs

1//! The doctor checks as engine functions (#55): every finding both frontends
2//! render comes from here — `zenctl doctor` orchestrates and renders, the
3//! zengui doctor panel calls the same [`run_doctor`] and renders the same
4//! [`DoctorReport`]. A check that lives in one frontend is a check the other
5//! frontend's user never sees (RFC 08 §6.1's argument, applied to ourselves).
6//!
7//! Check ids are **stable API**: scripts key on them (`--format json`), the
8//! GUI keys deltas on them. New checks add ids; nothing renames one. The full
9//! set is pinned in [`crate::report::CheckId`].
10
11use std::time::Duration;
12
13use crate::{Error, Result};
14use zenkey::grammar::with_base;
15use zenkey::{Declared, RegistrySlice};
16
17use crate::bus::query::{Answer, GetOpts, RepeatingRegistry, fleet_get, state_snapshot};
18use crate::judge::common::{FINDING_CAP, is_synthetic_marker};
19use crate::model::examples::Examples;
20use crate::report::{CheckId, DoctorFinding, DoctorReport, DoctorSeverity, DriftVerdict};
21
22/// What a doctor run should cost.
23#[derive(Debug, Clone)]
24pub struct DoctorSpec {
25    /// Run the deep checks too (per-family state snapshots for freshness,
26    /// storage-coverage join) — real query load, opt-in.
27    pub deep: bool,
28    /// At most this many state samples drained per family in the deep
29    /// checks (`--sample N`) — bounds the sweep's cost, not just its output.
30    /// `None` = unbounded.
31    pub sample: Option<usize>,
32    /// Per-query timeout.
33    pub timeout: Duration,
34    /// Listen passively to the data planes for this long after the GET
35    /// fan-in (`--for`, #161) and judge what rides: decode/validity,
36    /// declared-vs-observed QoS, unregistered traffic, over-rate events.
37    /// `None` = the phase does not run and the report carries no
38    /// observation section.
39    pub listen: Option<Duration>,
40}
41
42fn finding(
43    severity: DoctorSeverity,
44    check: CheckId,
45    subject: impl Into<String>,
46    evidence: impl Into<String>,
47    citation: Option<&str>,
48) -> DoctorFinding {
49    DoctorFinding {
50        severity,
51        check,
52        subject: subject.into(),
53        evidence: evidence.into(),
54        citation: citation.map(str::to_string),
55    }
56}
57
58/// The introspect key for a slice — a service origin's verbatim `@` chunk is
59/// structurally unmatchable by a fleet selector's `*` (property D4), so it
60/// takes its own key. That is the grammar working, not an exception to it.
61fn rpc_key(base: &str, slice: &RegistrySlice, procedure: &str) -> Result<String> {
62    Ok(match &slice.service_origin {
63        Some(origin) => {
64            // The slice already validated it on parse — `Other` here means the
65            // chunk is not a legal verbatim origin, which is the same finding
66            // the hand-rolled `ServiceOrigin::new` used to report.
67            // A *served* slice said this, so it is the peer that is
68            // malformed — not the caller, and not the fabric.
69            let o = origin.known().ok_or_else(|| {
70                Error::malformed(
71                    format!("slice {}", slice.name),
72                    format!("carries {:?} as a service origin", origin.token()),
73                )
74            })?;
75            with_base(base, zenkey::selector::service_rpc(o, &[procedure]))
76        }
77        None => with_base(base, zenkey::selector::fleet_rpc(&slice.name, &[procedure])),
78    })
79}
80
81/// Run every check against the live fleet and report typed findings.
82///
83/// `locals` is the caller's registry (loaded from `--registry` dirs or GUI
84/// settings). `None` means none was loaded: the served-vs-declared diff is
85/// skipped and only bus-derived checks run, and the report says so rather
86/// than reading in sync (O4 — "not asked" must not render as "clean").
87///
88/// `Option<&SliceSet>` and not `&[RegistrySlice]`: this is the engine's
89/// standing shape for "a registry, or honestly none" (`facts.rs` states it as
90/// policy), an empty slice could not tell the two apart, and the set arrives
91/// already indexed — doctor used to rebuild one from a clone of every slice
92/// halfway through the run.
93pub async fn run_doctor(
94    fleet: &crate::Fleet<'_>,
95    locals: Option<&crate::model::registry::SliceSet>,
96    spec: &DoctorSpec,
97) -> Result<DoctorReport> {
98    let (session, base) = (fleet.session(), fleet.base());
99
100    // A registry that declares nothing answers no question this run asks, so
101    // it takes the same path as none at all — normalised once, here, rather
102    // than at each of the four places that branch on it below.
103    let locals = locals.filter(|set| !set.slices().is_empty());
104    let roster = crate::bus::roster::roster(fleet, spec.timeout).await?;
105
106    let mut findings: Vec<DoctorFinding> = Vec::new();
107    let mut synced: Vec<String> = Vec::new();
108    let mut answered = 0usize;
109
110    // --- served-vs-declared diff (RFC 08 §6) --------------------------
111    for local in locals.iter().flat_map(|set| set.slices()) {
112        let key = rpc_key(base, local, "introspect")?;
113        let answers = fleet_get(fleet, &key, &GetOpts::new(spec.timeout)).await?;
114        for answer in &answers {
115            let Answer::Value(bytes) = &answer.answer else {
116                continue;
117            };
118            answered += 1;
119            let served_toml = bytes.to_bytes();
120            let served_toml = String::from_utf8_lossy(&served_toml);
121            let served = match zenkey::parse_slice(&served_toml) {
122                Ok(s) => s,
123                Err(e) => {
124                    findings.push(finding(
125                        DoctorSeverity::Error,
126                        CheckId::SliceParse,
127                        format!("{}/{}", answer.origin, local.name),
128                        format!("served slice does not parse: {e}"),
129                        Some("RFC 08 §6"),
130                    ));
131                    continue;
132                }
133            };
134            let diff = zenkey::slice::diff(&served, local);
135            if diff.is_empty() {
136                synced.push(format!(
137                    "{}/{} (registry {})",
138                    answer.origin, local.name, served.version
139                ));
140            } else {
141                for f in &diff {
142                    findings.push(finding(
143                        DoctorSeverity::Error,
144                        CheckId::SliceSync,
145                        format!("{}/{}", answer.origin, local.name),
146                        f.summary(),
147                        Some("RFC 08 §6"),
148                    ));
149                }
150            }
151        }
152    }
153
154    // One declared registry sweep (#37) serves both fallbacks below —
155    // doctor used to fan the identical wildcard GETs twice per run.
156    let sweep = if locals.is_none() {
157        let repeating = RepeatingRegistry::declare(fleet, spec.timeout).await?;
158        let slices: Vec<RegistrySlice> = repeating
159            .fetch()
160            .await?
161            .into_iter()
162            .map(|(s, _)| s)
163            .collect();
164        repeating.undeclare().await?;
165        Some(slices)
166    } else {
167        None
168    };
169
170    // With no local registry the only introspect coverage we can count is
171    // the fleet-wide wildcard.
172    if let Some(slices) = &sweep {
173        answered = slices.len();
174    }
175
176    // The roster is what makes silence legible (RFC 05 §3.1): a producer
177    // that holds an `alive` token but did not answer `introspect` is a bug,
178    // because producers MUST declare their @rpc queryables *before* their
179    // token — "alive ⇒ callable" (RFC 04 §5). Coverage is judged over the
180    // producers that were actually *asked*: with `--registry` covering a
181    // subset, a live producer outside the locals was never queried, and
182    // "not asked" must not render as "did not answer" (RFC 09 §5.1 O4).
183    let live: usize = roster.values().map(Vec::len).sum();
184    findings.extend(judge_introspect_coverage(
185        &roster,
186        locals.map(crate::model::registry::SliceSet::slices),
187        answered,
188    ));
189
190    // --- admin reachability ------------------------------------------
191    let routers = crate::routers(session, spec.timeout)
192        .await
193        .unwrap_or_default();
194    let mut router_version = None;
195    if routers.is_empty() {
196        findings.push(finding(
197            DoctorSeverity::Info,
198            CheckId::AdminUnreachable,
199            "mesh",
200            "no routers answered @/*/router (peer-only mesh, or the admin space is \
201             disabled) — storage/version checks skipped",
202            None,
203        ));
204    } else {
205        let versions: std::collections::BTreeSet<&str> = routers
206            .iter()
207            .filter_map(|r| r.version.as_deref())
208            .collect();
209        if versions.len() > 1 {
210            findings.push(finding(
211                DoctorSeverity::Error,
212                CheckId::RouterVersionSkew,
213                "mesh",
214                format!("router version skew across the mesh: {versions:?}"),
215                None,
216            ));
217        } else {
218            router_version = versions.iter().next().map(|v| v.to_string());
219        }
220    }
221
222    // --- schema conformance (RFC 08 §7) ------------------------------
223    // Which slices to judge: the locals when given, else what the fleet
224    // serves (the sweep above).
225    let slice_set: std::borrow::Cow<'_, crate::model::registry::SliceSet> = match sweep {
226        Some(slices) => {
227            std::borrow::Cow::Owned(crate::model::registry::SliceSet::from_slices(slices))
228        }
229        // The caller's set is already indexed; rebuilding it here reparsed
230        // every subject pattern to arrive at the set we were handed.
231        None => match locals {
232            Some(set) => std::borrow::Cow::Borrowed(set),
233            None => std::borrow::Cow::Owned(crate::model::registry::SliceSet::default()),
234        },
235    };
236    // One per producer, for the consumers whose question *is* the producer:
237    // totality, the listen phase's store, the served count, and the field
238    // table's declared-path join. Where several hosts answered this is the
239    // first of them — arrival order, which is not a fact about the fleet, and
240    // is why the drift check below reads the attributed list instead (#398).
241    let mut described: Vec<(String, zenkey::schema::SchemaSet)> = Vec::new();
242    // Every answer, attributed. `describe` is `@rpc/*/describe` — a wildcard
243    // origin — so this fans in across every host running the producer, and
244    // keeping one of them was how a schema disagreement came to name a
245    // producer and never a host (#398).
246    let mut described_by_origin: Vec<crate::model::decode::DescribedSchema> = Vec::new();
247    let mut undescribed = 0usize;
248    // One `GetOpts` for the sweep rather than one per producer: the elision
249    // ledger is per-options, so a fresh one per slice could never accumulate
250    // the fan-out's cost.
251    let describe_opts = GetOpts::new(spec.timeout);
252    for slice in slice_set.slices() {
253        let key = rpc_key(base, slice, "describe")?;
254        let answers = fleet_get(fleet, &key, &describe_opts).await?;
255        let before = described_by_origin.len();
256        for a in answers {
257            let origin = a.origin;
258            let Answer::Value(bytes) = a.answer else {
259                continue;
260            };
261            let cow = bytes.to_bytes();
262            let Some(set) = std::str::from_utf8(&cow)
263                .ok()
264                .and_then(|t| zenkey::schema::SchemaSet::parse(t).ok())
265            else {
266                continue;
267            };
268            if described_by_origin.len() == before {
269                described.push((slice.name.clone(), set.clone()));
270            }
271            described_by_origin.push(crate::model::decode::DescribedSchema {
272                origin,
273                producer: slice.name.clone(),
274                set,
275            });
276        }
277        // Nobody parseable answered for this producer. A producer where one
278        // host answered and another did not is *described* — the SHOULD is
279        // met — and the gap between them is the drift check's business.
280        if described_by_origin.len() == before {
281            undescribed += 1;
282        }
283    }
284    // Totality through the one engine implementation (`totality_gaps`) —
285    // doctor used to carry a parallel referenced-names path.
286    for gap in crate::model::decode::totality_gaps(&described, &slice_set) {
287        findings.push(finding(
288            DoctorSeverity::Error,
289            CheckId::DescribeTotality,
290            gap.producer.clone(),
291            format!(
292                "describe is not total — missing: {}",
293                gap.missing.join(", ")
294            ),
295            Some("RFC 08 §7"),
296        ));
297    }
298    for drift in crate::model::decode::schema_drift(&described_by_origin) {
299        let servers: Vec<String> = drift
300            .servers
301            .iter()
302            // `producer@origin`, because a type with two identities and no
303            // host to go and look at is the finding you can do least with
304            // (#398).
305            .map(|s| match s.hash.as_option() {
306                Some(h) => format!("{}@{} ({h})", s.producer, s.origin),
307                None => format!("{}@{} (no identity served)", s.producer, s.origin),
308            })
309            .collect();
310        // The two verdicts are not the same finding. A disagreement is a
311        // defect; a producer that served no identity leaves the question
312        // *unanswered*, and calling that an error would be the mirror of the
313        // bug #370 fixed — reporting a verdict nobody's evidence supports.
314        let (severity, evidence) = match drift.verdict {
315            DriftVerdict::Disagree => (
316                DoctorSeverity::Error,
317                format!("served with different schemas by {}", servers.join(", ")),
318            ),
319            DriftVerdict::Unjudgeable => (
320                DoctorSeverity::Warning,
321                format!(
322                    "agreement cannot be judged — {} served no schema identity: {} \
323                     (RFC 09 §5.1 O4; the hash exists for exactly this, RFC 08 §7)",
324                    drift
325                        .servers
326                        .iter()
327                        .filter(|s| s.hash.is_not_asked())
328                        .count(),
329                    servers.join(", ")
330                ),
331            ),
332        };
333        findings.push(finding(
334            severity,
335            CheckId::SchemaDrift,
336            drift.type_name.clone(),
337            evidence,
338            Some("RFC 08 §7"),
339        ));
340    }
341    if undescribed > 0 {
342        findings.push(finding(
343            DoctorSeverity::Info,
344            CheckId::DescribeMissing,
345            "fleet",
346            format!(
347                "{undescribed} producer(s) serve no describe (a SHOULD; generic tools \
348                 render their payloads structurally)"
349            ),
350            Some("RFC 08 §7"),
351        ));
352    }
353
354    // --- deep: freshness + storage coverage --------------------------
355    if spec.deep {
356        let now = std::time::SystemTime::now();
357        let mut unstamped = 0usize;
358        for slice in slice_set.slices() {
359            for subject in &slice.subjects {
360                let (Some(ttl), true) = (subject.ttl_s, subject.class.is(&zenkey::Class::State))
361                else {
362                    continue;
363                };
364                let Ok(pattern) = zenkey::pattern::SubjectPattern::parse(&subject.path) else {
365                    continue;
366                };
367                let selector = match &slice.service_origin {
368                    Some(origin) => with_base(
369                        base,
370                        format!("v1/{origin}/state/{}", pattern.selector_tail()),
371                    ),
372                    None => with_base(
373                        base,
374                        format!("v1/*/state/{}/{}", slice.name, pattern.selector_tail()),
375                    ),
376                };
377                let samples = state_snapshot(session, &selector, spec.timeout, spec.sample).await?;
378                let (family_findings, family_unstamped) = judge_state_samples(&samples, ttl, now);
379                findings.extend(family_findings);
380                unstamped += family_unstamped;
381            }
382        }
383        if unstamped > 0 {
384            findings.push(finding(
385                DoctorSeverity::Warning,
386                CheckId::UnstampedState,
387                "fleet",
388                format!(
389                    "{unstamped} state sample(s) carry no HLC timestamp — the deployment \
390                     lacks timestamping, which LWW requires; freshness is unjudgeable \
391                     for them"
392                ),
393                Some("RFC 04 §4"),
394            ));
395        }
396        let storages = crate::storages(session, spec.timeout)
397            .await
398            .unwrap_or_default();
399        let coverage = crate::state_coverage(&slice_set, base, &storages);
400        let uncovered: Vec<&crate::CoverageRow> = coverage
401            .iter()
402            .filter(|r| r.coverage == crate::Coverage::Uncovered)
403            .collect();
404        if !uncovered.is_empty() {
405            findings.push(finding(
406                DoctorSeverity::Info,
407                CheckId::StorageCoverage,
408                "fleet",
409                format!(
410                    "{} state famil(y|ies) have no storage coverage (volatile seeding \
411                     may ride the advanced-pub/sub cache): {}",
412                    uncovered.len(),
413                    uncovered
414                        .iter()
415                        .map(|r| format!("{}/{}", r.producer, r.path))
416                        .collect::<Vec<_>>()
417                        .join(", ")
418                ),
419                Some("RFC 04 §3.5"),
420            ));
421        }
422    }
423
424    // --- listen: judge what actually rides (#161) --------------------
425    let observation = match spec.listen {
426        Some(window) => {
427            let store = crate::model::decode::SchemaStore::new(base, spec.timeout);
428            // The GET phase above already asked every producer for its
429            // `describe` document. Hand those to the window's store rather
430            // than letting it re-ask the fleet, mid-window, for what this
431            // run is holding (RFC 08 §7; the store's frugality note).
432            for (producer, set) in &described {
433                store.insert(producer, set.clone());
434            }
435            // And sealed for the window (#337): the GET phase asked every
436            // producer the registry names, so a miss inside the window is a
437            // producer that served nothing — already counted as
438            // `describe_missing`. Left unsealed, that miss is a `describe`
439            // GET awaited inside the drain loop, re-asked every time its
440            // backoff expires, with nobody attending the broadcast.
441            let _sealed = store.seal();
442            let (listen_findings, summary) =
443                observe_traffic(fleet, &slice_set, &store, &described, window).await?;
444            findings.extend(listen_findings);
445            Some(summary)
446        }
447        None => None,
448    };
449
450    Ok(DoctorReport {
451        findings,
452        // `None` when no local registry was given: the served-vs-declared
453        // diff never ran, and the report must say so rather than looking
454        // like "ran, none in sync" (RFC 09 §5.1 O4, review finding R1).
455        synced: locals.is_some().then_some(synced).into(),
456        introspect_answered: answered,
457        live_producers: live,
458        describe_served: described.len(),
459        describe_missing: undescribed,
460        routers: routers.len(),
461        router_version,
462        deep: spec.deep,
463        observation,
464    })
465}
466
467/// How many decode attempts each key gets during the listen window — the
468/// budget that keeps a hot bus from turning the doctor into a load test.
469const DECODE_BUDGET: u8 = 2;
470
471/// The remainder wording every per-key listen check shares.
472const SAME_FINDING: &str = "more key(s) with the same finding";
473
474/// Spill a capped collector into `findings`, followed by the remainder note
475/// when the cap bit.
476///
477/// Filter and judge **into** the collector, never around it (deep-review D4):
478/// the `qos-observed-mismatch` cap used to bound the judged *keys*, so
479/// violators past the first [`FINDING_CAP`] of them vanished and the
480/// remainder note under-counted. [`Examples`] counts what it is offered, so
481/// the note cannot disagree with the population it summarises.
482fn emit_capped(
483    findings: &mut Vec<DoctorFinding>,
484    ex: Examples<DoctorFinding>,
485    check: CheckId,
486    tail: &str,
487) {
488    let more = ex.more(tail);
489
490    findings.extend(ex.into_vec());
491
492    if let Some(evidence) = more {
493        findings.push(finding(
494            DoctorSeverity::Info,
495            check,
496            "fleet",
497            evidence,
498            None,
499        ));
500    }
501}
502
503/// The declared events rate class as an hourly cap (RFC 04 §1.3):
504/// `rare` ≤ 1/h, `low` ≤ 1/min, `burst(n/h)` a declared cap.
505struct RateWindow {
506    /// The declared hourly cap, from [`RateClass::cap_per_hour`]. `None` is
507    /// a rate token this build cannot read — "cannot judge", never a
508    /// guessed budget (RFC 09 §5.1 O4).
509    cap: Option<u64>,
510    /// Samples seen on the family during the window.
511    seen: u64,
512}
513
514/// The passive listening phase: watch the data planes for `window`, judge
515/// each sample through the ladders that already exist — the Registration
516/// ladder, `qos_matches`, `decode_sample` — and aggregate per key so a hot
517/// key is one finding with a count, not a finding per sample.
518async fn observe_traffic(
519    fleet: &crate::Fleet<'_>,
520    slices: &crate::model::registry::SliceSet,
521    store: &crate::model::decode::SchemaStore,
522    described: &[(String, zenkey::schema::SchemaSet)],
523    window: Duration,
524) -> Result<(Vec<DoctorFinding>, crate::report::ObservationSummary)> {
525    use std::collections::BTreeMap;
526
527    let (session, base) = (fleet.session(), fleet.base());
528
529    // Scope statement (O5): the three data classes for host origins, plus
530    // each declared service origin's three — `*` never matches an `@` chunk
531    // (D4), so the service planes must be named to be seen. Shared with the
532    // `topic list --budget` observation (#221).
533    let scopes = crate::judge::common::data_plane_scopes(base, slices);
534
535    let monitor = crate::Monitor::start(session, crate::MonitorSpec::default()).await?;
536    let mut events = monitor.events();
537    // A scope that fails to declare tears the monitor down on the way out,
538    // rather than leaving a `**` subscriber to `Drop` (#336).
539    let monitor = monitor.watching(&scopes).await?;
540    let started = tokio::time::Instant::now();
541    let deadline = started + window;
542
543    // Field intelligence (#223): per-dotted-path stats over the structural
544    // value — sync and schema-free, so it rides every sample within the
545    // decode budget's reach and beyond.
546    let mut fields =
547        crate::judge::field::FieldObservation::new(crate::judge::field::DEFAULT_MAX_PATHS);
548    let mut samples: u64 = 0;
549    let mut dropped: u64 = 0;
550    let mut synthetic: u64 = 0;
551    // Bounded (#107): one projection per distinct key, LRU past the bound,
552    // evictions counted into the observation summary (O6).
553    let mut facts_cache = crate::model::facts::FactsCache::default();
554    let mut decode_budget: BTreeMap<String, u8> = BTreeMap::new();
555    // Per-key aggregates: key → count (+ what was wrong, first occurrence).
556    let mut unregistered: BTreeMap<String, u64> = BTreeMap::new();
557    let mut qos_bad: BTreeMap<String, (String, u64, u64)> = BTreeMap::new();
558    let mut undecodable: BTreeMap<String, (String, u64)> = BTreeMap::new();
559    let mut invalid: BTreeMap<String, (String, u64)> = BTreeMap::new();
560    // Per-family event counts: (family subject, declared rate) → what was
561    // seen and what was declared.
562    let mut event_counts: BTreeMap<(String, String), RateWindow> = BTreeMap::new();
563    // Stamping nodes that are not the publisher (#213): zid → samples.
564    let mut foreign_stampers: BTreeMap<String, u64> = BTreeMap::new();
565
566    // One timer for the whole window, not one per iteration (#346).
567    // `sleep_until` builds a future and registers a timer each time it
568    // is evaluated, and a `select!` in a loop evaluates it on every
569    // pass — at 100k samples/s that is 100k registrations a second for
570    // a deadline that never moves.
571    let window_over = tokio::time::sleep_until(deadline);
572    tokio::pin!(window_over);
573    loop {
574        let item = tokio::select! {
575            item = events.recv() => item,
576            () = &mut window_over => break,
577        };
578        match item {
579            Some(crate::StreamItem::Event(crate::FleetEvent::Sample(s))) => {
580                samples += 1;
581                if let Some(att) = &s.attachment
582                    && is_synthetic_marker(&att.to_bytes())
583                {
584                    synthetic += 1;
585                }
586                // Who stamped it (#213). A router doing the timestamping is
587                // not a fault — it is a deployment choice — but it silently
588                // changes what every latency in this suite measures, so it is
589                // worth saying out loud once.
590                if let Some(crate::StampProvenance::Foreign { stamper }) = s.stamped_by {
591                    *foreign_stampers.entry(stamper.to_string()).or_default() += 1;
592                }
593                // Same bound as `run_field`'s drain (#346): the parse is
594                // per sample by design, so the payload size is what has to be
595                // bounded, and the skip is counted rather than read as an
596                // absent document.
597                let bytes = s.payload.to_bytes();
598                if bytes.len() > crate::model::decode::OBSERVE_LIMIT {
599                    fields.observe_unread(&s.key);
600                } else {
601                    let doc = crate::model::decode::structural_value(&bytes);
602                    fields.observe(&s.key, started.elapsed().as_secs_f64(), doc.as_ref());
603                }
604                facts_cache.ensure(base, &s.key, Some(slices));
605                let facts = facts_cache.get(&s.key).expect("just ensured this key");
606                match &facts.registration {
607                    crate::model::facts::Registration::Unregistered => {
608                        *unregistered.entry(s.key.clone()).or_default() += 1;
609                    }
610                    crate::model::facts::Registration::Registered(sf) => {
611                        if let (Some(profile), Some(declared)) = (sf.declared_qos(), &sf.qos) {
612                            let entry = qos_bad
613                                .entry(s.key.clone())
614                                .or_insert_with(|| (declared.token().to_string(), 0, 0));
615                            entry.2 += 1;
616                            if !s.qos_matches(profile) {
617                                entry.1 += 1;
618                            }
619                        }
620                        if let (Some(rate), crate::model::facts::KeyShape::V1(v)) =
621                            (&sf.rate, &facts.shape)
622                            && v.class == "events"
623                        {
624                            let family = match &v.producer {
625                                Some(p) => format!("{p}/{}", sf.path),
626                                None => format!("{}/{}", v.origin, sf.path),
627                            };
628                            // The cap is taken from the typed `RateClass`
629                            // here, where it is in hand — this used to
630                            // stringify the token and re-parse it below
631                            // through a second copy of RFC 04 §1.3's
632                            // mapping (#350's sweep).
633                            event_counts
634                                .entry((family, rate.token()))
635                                .or_insert_with(|| RateWindow {
636                                    cap: rate.cap_per_hour(),
637                                    seen: 0,
638                                })
639                                .seen += 1;
640                        }
641                        let budget = decode_budget.entry(s.key.clone()).or_default();
642                        if *budget < DECODE_BUDGET {
643                            *budget += 1;
644                            // `Some`: the doctor's slice set comes from its
645                            // own live introspect sweep, so the registry was
646                            // always asked here — `NoRegistry` (#246) cannot
647                            // arise, and like every not-validated reason
648                            // other than `Undecodable` it would fall through
649                            // the `_` arm below: not asked/not checkable is
650                            // never a finding (RFC 09 §5.1 O4).
651                            let d = crate::model::decode::decode_sample(
652                                fleet,
653                                store,
654                                Some(slices),
655                                &s.key,
656                                Some(&s.encoding),
657                                &s.payload.to_bytes(),
658                            )
659                            .await;
660                            match d.verdict {
661                                crate::Verdict::NotValidated(
662                                    zenkey::schema::validate::NotValidated::Undecodable,
663                                ) => {
664                                    let e = undecodable.entry(s.key.clone()).or_insert_with(|| {
665                                        (
666                                            d.decode_error
667                                                .unwrap_or_else(|| "does not decode".into()),
668                                            0,
669                                        )
670                                    });
671                                    e.1 += 1;
672                                }
673                                crate::Verdict::Invalid(errors) => {
674                                    let e = invalid
675                                        .entry(s.key.clone())
676                                        .or_insert_with(|| (errors.join("; "), 0));
677                                    e.1 += 1;
678                                }
679                                _ => {}
680                            }
681                        }
682                    }
683                    _ => {}
684                }
685            }
686            Some(crate::StreamItem::Dropped(n)) => dropped += n,
687            Some(_) => continue,
688            None => break,
689        }
690    }
691    let keys_seen = facts_cache.len();
692    monitor.shutdown().await?;
693
694    // Key-population budgets (#221): the window's distinct keys, grouped
695    // into `{var}` families per origin, judged against each family's
696    // declared `cardinality`.
697    let budgets =
698        crate::judge::budget::BudgetObservation::observe(base, slices, facts_cache.keys());
699
700    let window_s = window.as_secs_f64();
701    let mut findings = Vec::new();
702
703    let mut ex = Examples::new(FINDING_CAP);
704    for (key, (error, n)) in &undecodable {
705        ex.push_with(|| {
706            finding(
707                DoctorSeverity::Error,
708                CheckId::PayloadUndecodable,
709                key.clone(),
710                format!(
711                    "payload does not decode as its declared type: {error} ({n} sample(s) tried)"
712                ),
713                Some("RFC 08 §7"),
714            )
715        });
716    }
717    emit_capped(&mut findings, ex, CheckId::PayloadUndecodable, SAME_FINDING);
718    let mut ex = Examples::new(FINDING_CAP);
719    for (key, (violations, n)) in &invalid {
720        ex.push_with(|| {
721            finding(
722                DoctorSeverity::Error,
723                CheckId::PayloadInvalid,
724                key.clone(),
725                format!("payload violates the served schema: {violations} ({n} sample(s) tried)"),
726                Some("RFC 08 §7"),
727            )
728        });
729    }
730    emit_capped(&mut findings, ex, CheckId::PayloadInvalid, SAME_FINDING);
731    findings.extend(judge_qos_observed(&qos_bad));
732    if !foreign_stampers.is_empty() {
733        let mut named: Vec<String> = foreign_stampers
734            .iter()
735            .map(|(zid, n)| format!("{zid} ({n} sample(s))"))
736            .collect();
737        named.sort();
738        findings.push(finding(
739            DoctorSeverity::Info,
740            CheckId::TimestampStampedElsewhere,
741            "fleet".to_string(),
742            format!(
743                "HLCs on this bus are stamped by {} node(s) that are not the publishing \
744                 session — a deployment with router-side timestamping, which is legal and \
745                 common. Latency measured from these stamps is stamper→observer, not \
746                 publisher→observer: {}",
747                foreign_stampers.len(),
748                named.join(", ")
749            ),
750            Some("RFC 09 §5.1 O7"),
751        ));
752    }
753    let mut ex = Examples::new(FINDING_CAP);
754    for (key, n) in &unregistered {
755        ex.push_with(|| {
756            finding(
757                DoctorSeverity::Warning,
758                CheckId::UnregisteredTraffic,
759                key.clone(),
760                format!(
761                    "{n} sample(s) on a subject the producer's slice does not declare — \
762                     for a conforming producer, a subject that is not registered does not exist"
763                ),
764                Some("RFC 08 §2"),
765            )
766        });
767    }
768    emit_capped(
769        &mut findings,
770        ex,
771        CheckId::UnregisteredTraffic,
772        SAME_FINDING,
773    );
774    // Over-rate only, and only when provable: within any window no longer
775    // than an hour, exceeding the hourly cap is conclusive. Absence or
776    // under-rate in a bounded window is never a finding (O1/O4).
777    if window <= Duration::from_secs(3600) {
778        for ((family, rate), RateWindow { cap, seen: count }) in &event_counts {
779            let Some(cap) = cap else {
780                continue;
781            };
782            if count > cap {
783                findings.push(finding(
784                    DoctorSeverity::Warning,
785                    CheckId::RateOverDeclared,
786                    family.clone(),
787                    format!(
788                        "{count} event(s) in {window_s:.0}s exceeds the declared \
789                         `{rate}` cap ({cap}/h)"
790                    ),
791                    Some("RFC 04 §1.3"),
792                ));
793            }
794        }
795    }
796
797    findings.extend(judge_cardinality(slices, &budgets, window_s));
798
799    // Field intelligence (#223): the three field-granular checks, judged
800    // with what is known per key — declared `ttl_s`/type from the resolved
801    // facts, declared paths from the describe sets the GET phase gathered.
802    let field_ctx = field_context_from(slices, described, &facts_cache);
803    findings.extend(crate::judge::field::judge_fields(
804        &fields, window_s, &field_ctx,
805    ));
806
807    Ok((
808        findings,
809        crate::report::ObservationSummary {
810            window_s,
811            scopes,
812            samples,
813            keys_seen,
814            dropped,
815            synthetic_marked: synthetic,
816            // The per-path table is bounded like every other table here, and
817            // its cost is a wire fact (RFC 09 §5.1 O6).
818            field_paths_dropped: fields.dropped_paths(),
819            facts_evicted: facts_cache.evicted(),
820        },
821    ))
822}
823
824/// The per-key context the field judges need (#223), built from the listen
825/// phase's resolved facts and the already-gathered describe sets — pure, so
826/// the join is testable without a bus.
827fn field_context_from(
828    slices: &crate::model::registry::SliceSet,
829    described: &[(String, zenkey::schema::SchemaSet)],
830    facts: &crate::model::facts::FactsCache,
831) -> std::collections::BTreeMap<String, crate::judge::field::KeyFieldContext> {
832    use std::collections::BTreeMap;
833
834    let mut declared_cache: BTreeMap<(String, String), Option<crate::judge::field::DeclaredPaths>> =
835        BTreeMap::new();
836    let mut ctx = BTreeMap::new();
837    for (key, f) in facts.iter() {
838        let mut c = crate::judge::field::KeyFieldContext::default();
839        if let crate::model::facts::Registration::Registered(sf) = &f.registration {
840            c.ttl_s = sf.ttl_s;
841            c.type_name = Some(sf.type_name.clone());
842            if let Some(producer) = crate::judge::common::producer_of(f, Some(slices))
843                && !sf.type_name.is_empty()
844            {
845                let declared = declared_cache
846                    .entry((producer.clone(), sf.type_name.clone()))
847                    .or_insert_with(|| {
848                        described
849                            .iter()
850                            .find(|(name, _)| *name == producer)
851                            .and_then(|(_, set)| set.get(&sf.type_name))
852                            .and_then(|schema| schema.json_document())
853                            .and_then(crate::judge::field::DeclaredPaths::from_json_schema)
854                    });
855                c.declared = declared.clone();
856            }
857        }
858        ctx.insert(key.to_string(), c);
859    }
860    ctx
861}
862
863/// The `qos-observed-mismatch` findings from the listen window's per-key
864/// aggregates: `key → (declared profile, mismatched, judged)` — pure, so
865/// the cap arithmetic is testable without a bus.
866///
867/// Filter **then** cap, [`judge_cardinality`]'s pattern (deep-review D4):
868/// the map holds every judged key, most of them clean, so capping the map
869/// *entries* first silently dropped violators past the first
870/// [`FINDING_CAP`] keys and made the remainder note miscount. The cap
871/// bounds the findings; the filter decides what a finding is.
872fn judge_qos_observed(
873    qos_bad: &std::collections::BTreeMap<String, (String, u64, u64)>,
874) -> Vec<DoctorFinding> {
875    let mut findings = Vec::new();
876
877    let mut ex = Examples::new(FINDING_CAP);
878
879    for (key, (declared, bad, total)) in qos_bad.iter().filter(|(_, (_, bad, _))| *bad > 0) {
880        ex.push_with(|| {
881            finding(
882                DoctorSeverity::Warning,
883                CheckId::QosObservedMismatch,
884                key.clone(),
885                format!(
886                    "{bad} of {total} sample(s) did not ride the declared {declared} — this \
887                     is what actually rode: an interceptor MAY rewrite QoS, so it is a \
888                     deviation, not proof of the publisher"
889                ),
890                Some("RFC 04 §3"),
891            )
892        });
893    }
894    emit_capped(
895        &mut findings,
896        ex,
897        CheckId::QosObservedMismatch,
898        SAME_FINDING,
899    );
900    findings
901}
902
903/// Judge introspect coverage — "alive ⇒ callable" (RFC 04 §5) — against the
904/// producers that were actually asked. Pure, so the O4 boundary is testable
905/// without a bus.
906///
907/// `locals: Some` is the `--registry` run: only the producers the local
908/// slices name were queried, so only those count toward coverage — a live
909/// producer whose slice a *partial* registry does not carry was never asked,
910/// and counting it as "did not answer" would be a false finding (RFC 09
911/// §5.1 O4; deep-review D3). `None` is the wildcard sweep, where every
912/// roster producer was in the fan-in. Either way the evidence states the
913/// scope it checked.
914///
915/// Matching follows the roster's own conventions: an instance suffix shares
916/// its base slice (`sysinfo-2` → `sysinfo`, RFC 03 §1.5), and a service
917/// origin's token names the service as its producer (RFC 06 §5), matched by
918/// the slice's declared origin or name.
919fn judge_introspect_coverage(
920    roster: &std::collections::BTreeMap<String, Vec<String>>,
921    locals: Option<&[RegistrySlice]>,
922    answered: usize,
923) -> Option<DoctorFinding> {
924    let live: usize = roster.values().map(Vec::len).sum();
925
926    let (in_scope, scope) = match locals {
927        None => (
928            live,
929            "scope: the whole roster (fleet-wide wildcard sweep)".to_string(),
930        ),
931        Some(locals) => {
932            let named = |origin: &str, producer: &str| {
933                let base_name = zenkey::grammar::Producer::parse_chunk(producer)
934                    .map(|p| p.name().to_string())
935                    .unwrap_or_else(|_| producer.to_string());
936                locals.iter().any(|l| {
937                    l.name == base_name
938                        || l.service_origin.as_ref().map(Declared::token) == Some(origin)
939                })
940            };
941            let in_scope: usize = roster
942                .iter()
943                .map(|(origin, producers)| producers.iter().filter(|p| named(origin, p)).count())
944                .sum();
945            let mut names: Vec<&str> = locals.iter().map(|l| l.name.as_str()).collect();
946            names.sort_unstable();
947            names.dedup();
948            let not_asked = live - in_scope;
949            (
950                in_scope,
951                format!(
952                    "scope: the producer(s) the local registry names ({}); {} other \
953                     live producer(s) were not asked and are not counted (O4)",
954                    names.join(", "),
955                    not_asked
956                ),
957            )
958        }
959    };
960    (answered < in_scope).then(|| {
961        finding(
962            DoctorSeverity::Error,
963            CheckId::IntrospectCoverage,
964            "fleet",
965            format!(
966                "{} of {} live producer(s) in scope did not answer introspect — \
967                 alive ⇒ callable, so this is a finding, not a boot race; {scope}",
968                in_scope - answered,
969                in_scope
970            ),
971            Some("RFC 04 §5"),
972        )
973    })
974}
975
976/// Judge one state family's samples against its declared ttl — pure, so the
977/// freshness math is testable without a bus. Returns the stale findings and
978/// the count of unstamped samples (aggregated by the caller into the one
979/// `unstamped-state` finding).
980fn judge_state_samples(
981    samples: &[crate::StateSample],
982    ttl: i64,
983    now: std::time::SystemTime,
984) -> (Vec<DoctorFinding>, usize) {
985    let mut findings = Vec::new();
986
987    let mut unstamped = 0usize;
988
989    for sample in samples {
990        match sample.timestamp {
991            Some(ts) => {
992                let stamped = ts.get_time().to_system_time();
993                if let Ok(age) = now.duration_since(stamped)
994                    && age.as_secs() as i64 > ttl
995                {
996                    findings.push(finding(
997                        DoctorSeverity::Error,
998                        CheckId::StaleState,
999                        sample.key.clone(),
1000                        format!(
1001                            "{}s old against ttl {ttl}s (refresh <= ttl/2)",
1002                            age.as_secs()
1003                        ),
1004                        Some("RFC 04 §1.2"),
1005                    ));
1006                }
1007            }
1008            None => unstamped += 1,
1009        }
1010    }
1011    (findings, unstamped)
1012}
1013
1014/// Judge every declared `{var}` family's key population against its declared
1015/// `cardinality` (#221) — pure, so the acceptance case (declared 16, 40
1016/// observed) is testable without a bus.
1017///
1018/// The honesty rules, verbatim from the issue:
1019///
1020/// - Observed **over** declared is a finding (RFC 04 §1.2's budget is a
1021///   MUST); observed **under** declared is **not** — an idle host declares
1022///   nothing wrong, and a bounded window proves a lower bound, never the
1023///   population (RFC 09 §5.1 O4/O6). The window rides in the evidence.
1024/// - `{path...}` rest-variable families are unbounded by construction and
1025///   are **exempt and say so** — "exempt: rest-variable", never a silent
1026///   skip and never a pass (the RFC 08 §6.1 v1.20 shape for subject checks).
1027/// - Judged **per origin**: RFC 04 §1 bounds cardinality per producer, so
1028///   one origin over the bound is conclusive and two origins' healthy
1029///   populations are never summed into a fake violation.
1030fn judge_cardinality(
1031    slices: &crate::model::registry::SliceSet,
1032    observed: &crate::judge::budget::BudgetObservation,
1033    window_s: f64,
1034) -> Vec<DoctorFinding> {
1035    let mut findings = Vec::new();
1036
1037    let mut over: Examples<DoctorFinding> = Examples::new(FINDING_CAP);
1038
1039    for slice in slices.slices() {
1040        for s in &slice.subjects {
1041            if !s.path.contains('{') {
1042                continue; // a literal subject's population is 1 by construction
1043            }
1044            if s.path.contains("...") {
1045                let seen: usize = observed
1046                    .family(&slice.name, &s.path)
1047                    .map(|origins| origins.values().map(|keys| keys.len()).sum())
1048                    .unwrap_or(0);
1049                findings.push(finding(
1050                    DoctorSeverity::Info,
1051                    CheckId::CardinalityOverDeclared,
1052                    format!("{}/{}", slice.name, s.path),
1053                    format!(
1054                        "exempt: rest-variable — a `{{var...}}` family is unbounded by \
1055                         construction, so its declared cardinality is not a bound this \
1056                         check can pass or fail; {seen} distinct key(s) observed in \
1057                         {window_s:.0}s"
1058                    ),
1059                    Some("RFC 08 §6.1"),
1060                ));
1061                continue;
1062            }
1063            let Some(declared) = s.cardinality else {
1064                continue; // nothing declared, nothing to judge (the RFC 08 §5
1065                // lint that requires the field is the producer build's)
1066            };
1067            let Some(origins) = observed.family(&slice.name, &s.path) else {
1068                continue; // unobserved is not "within budget" — no verdict
1069            };
1070            for (origin, keys) in origins {
1071                if keys.len() as i64 <= declared {
1072                    continue; // under/at declared: not a finding (O4)
1073                }
1074                let examples = Examples::collect(
1075                    crate::judge::common::EXPANSION_CAP,
1076                    keys.iter().map(String::as_str),
1077                );
1078                let subject = if origin.starts_with('@') {
1079                    format!("{origin}/{}", s.path)
1080                } else {
1081                    format!("{origin}/{}/{}", slice.name, s.path)
1082                };
1083                over.push_with(|| {
1084                    finding(
1085                        DoctorSeverity::Warning,
1086                        CheckId::CardinalityOverDeclared,
1087                        subject,
1088                        format!(
1089                            "{} distinct key(s) observed in {window_s:.0}s exceed the \
1090                         declared cardinality {declared} — e.g. {}. A bounded window \
1091                         observes a lower bound: the population is at least this",
1092                            keys.len(),
1093                            examples.as_slice().join(", ")
1094                        ),
1095                        Some("RFC 04 §1.2"),
1096                    )
1097                });
1098            }
1099        }
1100    }
1101    emit_capped(
1102        &mut findings,
1103        over,
1104        CheckId::CardinalityOverDeclared,
1105        "more origin famil(y|ies) over their declared cardinality",
1106    );
1107    findings
1108}
1109
1110#[cfg(test)]
1111mod tests {
1112    use super::*;
1113
1114    const BOUNDED: &str = r#"
1115        [registry]
1116        version = "1.0"
1117        app = "t"
1118        convention = 1
1119        [producer]
1120        name = "sysinfo"
1121        [[subject]]
1122        path = "disk/{mount}/used"
1123        class = "telemetry"
1124        type = "Point"
1125        cardinality = 16
1126    "#;
1127
1128    /// The #221 acceptance case: declared 16, 40 observed expansions — the
1129    /// finding fires with the count, the declared bound, examples, and the
1130    /// window it rests on.
1131    #[test]
1132    fn cardinality_over_declared_fires_with_count_and_examples() {
1133        let slices = crate::model::registry::SliceSet::from_toml_for_tests(BOUNDED);
1134        let keys: Vec<String> = (0..40)
1135            .map(|i| format!("v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/m{i:02}/used"))
1136            .collect();
1137        let obs = crate::judge::budget::BudgetObservation::observe(
1138            "",
1139            &slices,
1140            keys.iter().map(String::as_str),
1141        );
1142        let findings = judge_cardinality(&slices, &obs, 10.0);
1143        assert_eq!(findings.len(), 1, "{findings:?}");
1144        let f = &findings[0];
1145        assert_eq!(f.check, CheckId::CardinalityOverDeclared);
1146        assert_eq!(f.severity, DoctorSeverity::Warning);
1147        assert_eq!(f.subject, "h-aaaaaaaaaaaa/sysinfo/disk/{mount}/used");
1148        assert!(f.evidence.contains("40 distinct key(s)"), "{}", f.evidence);
1149        assert!(f.evidence.contains("cardinality 16"), "{}", f.evidence);
1150        assert!(f.evidence.contains("10s"), "the window is stated");
1151        assert!(
1152            f.evidence.contains("disk/m00/used"),
1153            "examples are named: {}",
1154            f.evidence
1155        );
1156    }
1157
1158    /// Under (or at) the declared bound is **not** a finding: an idle host
1159    /// declares nothing wrong, and a bounded window proves a lower bound,
1160    /// never the population (O4/O6).
1161    #[test]
1162    fn cardinality_under_declared_is_not_a_finding() {
1163        let slices = crate::model::registry::SliceSet::from_toml_for_tests(BOUNDED);
1164        let keys = [
1165            "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/root/used",
1166            "v1/h-aaaaaaaaaaaa/telemetry/sysinfo/disk/var/used",
1167            // Two origins at 15 each must never be summed into a fake 30 > 16.
1168            "v1/h-bbbbbbbbbbbb/telemetry/sysinfo/disk/root/used",
1169        ];
1170        let obs = crate::judge::budget::BudgetObservation::observe("", &slices, keys);
1171        assert!(judge_cardinality(&slices, &obs, 5.0).is_empty());
1172    }
1173
1174    /// The other #221 acceptance case: a `{path...}` family yields the
1175    /// exemption wording — "exempt: rest-variable" — and never a pass (nor an
1176    /// over-finding, however many members it grows).
1177    #[test]
1178    fn rest_variable_families_are_exempt_and_say_so() {
1179        let toml = r#"
1180            [registry]
1181            version = "1.0"
1182            app = "t"
1183            convention = 1
1184            [producer]
1185            name = "gnmi"
1186            [[subject]]
1187            path = "{device}/{path...}"
1188            class = "telemetry"
1189            type = "Point"
1190            cardinality = 2
1191        "#;
1192        let slices = crate::model::registry::SliceSet::from_toml_for_tests(toml);
1193        let keys: Vec<String> = (0..5)
1194            .map(|i| format!("v1/h-aaaaaaaaaaaa/telemetry/gnmi/sw1/if/eth{i}/rx"))
1195            .collect();
1196        let obs = crate::judge::budget::BudgetObservation::observe(
1197            "",
1198            &slices,
1199            keys.iter().map(String::as_str),
1200        );
1201        let findings = judge_cardinality(&slices, &obs, 5.0);
1202        assert_eq!(findings.len(), 1, "{findings:?}");
1203        let f = &findings[0];
1204        assert_eq!(f.severity, DoctorSeverity::Info, "an exemption, not a pass");
1205        assert!(
1206            f.evidence.starts_with("exempt: rest-variable"),
1207            "{}",
1208            f.evidence
1209        );
1210        assert!(f.evidence.contains("5 distinct key(s)"), "{}", f.evidence);
1211        // And with nothing observed the family still says so — exempt is a
1212        // property of the declaration, not of the traffic.
1213        let quiet = judge_cardinality(&slices, &Default::default(), 5.0);
1214        assert_eq!(quiet.len(), 1);
1215        assert!(quiet[0].evidence.starts_with("exempt: rest-variable"));
1216    }
1217
1218    /// Deep-review D4: the qos-observed-mismatch cap bounds *violators*, not
1219    /// map entries. 30 judged keys where the first 5 (in map order) are
1220    /// clean and the remaining 25 violate: every violator is counted — the
1221    /// first 20 as findings, the other 5 in a remainder note that counts
1222    /// correctly. Capping before filtering used to drop the violators past
1223    /// the first 20 map entries and miscount the note.
1224    #[test]
1225    fn qos_mismatch_cap_bounds_violators_not_map_entries() {
1226        let mut qos_bad: std::collections::BTreeMap<String, (String, u64, u64)> =
1227            std::collections::BTreeMap::new();
1228        for i in 0..30u32 {
1229            // k00..k04 sort first and are clean; k05..k29 are violators.
1230            let bad = if i < 5 { 0 } else { 1 };
1231            qos_bad.insert(
1232                format!("v1/h-a/telemetry/x/k{i:02}"),
1233                ("tel".into(), bad, 10),
1234            );
1235        }
1236        let findings = judge_qos_observed(&qos_bad);
1237        let per_key: Vec<&DoctorFinding> = findings
1238            .iter()
1239            .filter(|f| f.severity == DoctorSeverity::Warning)
1240            .collect();
1241        assert_eq!(per_key.len(), FINDING_CAP, "the cap bounds the findings");
1242        assert!(
1243            per_key.iter().all(|f| f.evidence.starts_with("1 of 10")),
1244            "only violators become findings: {findings:#?}"
1245        );
1246        assert!(
1247            per_key.iter().any(|f| f.subject.ends_with("k24")),
1248            "violators past the first {FINDING_CAP} map entries (k20..k24) are \
1249             not dropped: {findings:#?}"
1250        );
1251        let note = findings
1252            .iter()
1253            .find(|f| f.severity == DoctorSeverity::Info)
1254            .expect("a remainder note");
1255        assert_eq!(
1256            note.evidence, "… and 5 more key(s) with the same finding",
1257            "the note counts violators (25 − 20), not map entries"
1258        );
1259
1260        // At or under the cap: every violator is a finding, no note.
1261        let few: std::collections::BTreeMap<String, (String, u64, u64)> = qos_bad
1262            .iter()
1263            .take(10)
1264            .map(|(k, v)| (k.clone(), v.clone()))
1265            .collect();
1266        let findings = judge_qos_observed(&few);
1267        assert_eq!(findings.len(), 5, "{findings:#?}");
1268        assert!(
1269            findings
1270                .iter()
1271                .all(|f| f.severity == DoctorSeverity::Warning)
1272        );
1273    }
1274
1275    fn roster_of(entries: &[(&str, &[&str])]) -> std::collections::BTreeMap<String, Vec<String>> {
1276        entries
1277            .iter()
1278            .map(|(origin, producers)| {
1279                (
1280                    origin.to_string(),
1281                    producers.iter().map(|p| p.to_string()).collect(),
1282                )
1283            })
1284            .collect()
1285    }
1286
1287    fn slice_named(name: &str) -> RegistrySlice {
1288        zenkey::parse_slice(&format!(
1289            "[registry]\nversion = \"1.0\"\napp = \"t\"\nconvention = 1\n\
1290             [producer]\nname = \"{name}\"\n"
1291        ))
1292        .expect("fixture slice parses")
1293    }
1294
1295    /// Deep-review D3: with `--registry` covering a subset of the fleet, a
1296    /// live producer the locals do not name was never asked — so it must not
1297    /// count as "did not answer" (O4). One local slice, answered by its one
1298    /// origin, beside an extra live producer: no finding.
1299    #[test]
1300    fn a_partial_registry_does_not_count_unasked_producers_against_coverage() {
1301        let roster = roster_of(&[("h-aaaaaaaaaaaa", &["sysinfo", "extra"])]);
1302        let locals = [slice_named("sysinfo")];
1303        assert_eq!(
1304            judge_introspect_coverage(&roster, Some(&locals), 1),
1305            None,
1306            "the un-asked producer is out of scope, not silent"
1307        );
1308    }
1309
1310    /// …and when an in-scope producer really did not answer, the finding
1311    /// fires and its evidence states the scope it checked — including that
1312    /// the out-of-scope producer was not counted. An instance suffix shares
1313    /// its base slice (RFC 03 §1.5), so `sysinfo-2` is in scope too.
1314    #[test]
1315    fn introspect_coverage_evidence_states_its_scope() {
1316        let roster = roster_of(&[
1317            ("h-aaaaaaaaaaaa", &["sysinfo", "extra"]),
1318            ("h-bbbbbbbbbbbb", &["sysinfo-2"]),
1319        ]);
1320        let locals = [slice_named("sysinfo")];
1321        let f = judge_introspect_coverage(&roster, Some(&locals), 1).expect("a finding");
1322        assert_eq!(f.check, CheckId::IntrospectCoverage);
1323        assert!(f.evidence.contains("1 of 2"), "{}", f.evidence);
1324        assert!(
1325            f.evidence.contains("the local registry names (sysinfo)"),
1326            "{}",
1327            f.evidence
1328        );
1329        assert!(
1330            f.evidence
1331                .contains("1 other live producer(s) were not asked"),
1332            "{}",
1333            f.evidence
1334        );
1335    }
1336
1337    /// A service origin's token names the service as its producer (RFC 06
1338    /// §5); a local slice matches it by declared origin.
1339    #[test]
1340    fn a_service_slice_scopes_its_origin_into_coverage() {
1341        let roster = roster_of(&[("@catalog", &["catalog"]), ("h-aaaaaaaaaaaa", &["extra"])]);
1342        let locals = [zenkey::parse_slice(
1343            "[registry]\nversion = \"1.0\"\napp = \"t\"\nconvention = 1\n\
1344             [service]\nname = \"catalog\"\norigin = \"@catalog\"\n",
1345        )
1346        .expect("service slice parses")];
1347        assert_eq!(judge_introspect_coverage(&roster, Some(&locals), 1), None);
1348        let f = judge_introspect_coverage(&roster, Some(&locals), 0).expect("a finding");
1349        assert!(f.evidence.contains("1 of 1"), "{}", f.evidence);
1350    }
1351
1352    /// The wildcard sweep keeps the whole roster in scope, and says so.
1353    #[test]
1354    fn the_wildcard_sweep_judges_the_whole_roster() {
1355        let roster = roster_of(&[("h-aaaaaaaaaaaa", &["sysinfo", "extra"])]);
1356        let f = judge_introspect_coverage(&roster, None, 1).expect("a finding");
1357        assert!(f.evidence.contains("1 of 2"), "{}", f.evidence);
1358        assert!(f.evidence.contains("whole roster"), "{}", f.evidence);
1359        assert_eq!(judge_introspect_coverage(&roster, None, 2), None);
1360    }
1361
1362    #[test]
1363    fn freshness_judgement_is_pure_and_ttl_bound() {
1364        let now = std::time::SystemTime::now();
1365        let fresh_ts = zenoh::time::Timestamp::new(
1366            zenoh::time::NTP64::from(now.duration_since(std::time::UNIX_EPOCH).unwrap()),
1367            zenoh::time::TimestampId::rand(),
1368        );
1369        let stale_ts = zenoh::time::Timestamp::new(
1370            zenoh::time::NTP64::from(
1371                now.duration_since(std::time::UNIX_EPOCH).unwrap() - Duration::from_secs(120),
1372            ),
1373            zenoh::time::TimestampId::rand(),
1374        );
1375        let samples = vec![
1376            crate::StateSample {
1377                key: "b/v1/h-aaaaaaaaaaaa/state/p/health".into(),
1378                timestamp: Some(fresh_ts),
1379                payload_len: 2,
1380            },
1381            crate::StateSample {
1382                key: "b/v1/h-bbbbbbbbbbbb/state/p/health".into(),
1383                timestamp: Some(stale_ts),
1384                payload_len: 2,
1385            },
1386            crate::StateSample {
1387                key: "b/v1/h-cccccccccccc/state/p/health".into(),
1388                timestamp: None,
1389                payload_len: 2,
1390            },
1391        ];
1392        let (findings, unstamped) = judge_state_samples(&samples, 30, now);
1393        assert_eq!(
1394            findings.len(),
1395            1,
1396            "only the stale stamped sample is a finding"
1397        );
1398        assert_eq!(findings[0].check, CheckId::StaleState);
1399        assert!(findings[0].subject.contains("h-bbbbbbbbbbbb"));
1400        assert_eq!(unstamped, 1, "the unstamped sample is counted, not judged");
1401    }
1402}