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