gpuviewer-core 0.1.1

Telemetry core for gpuviewer: vendor backends, data model, event derivation
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
//! Event derivation — the "story" layer.
//!
//! Two-tier honesty contract (non-negotiable, see docs/research/04-synthesis.md §5 risk 2):
//! - `Confidence::Fact` events assert observed state transitions plainly (throttle bit set,
//!   process exited). They carry the raw evidence that produced them.
//! - `Confidence::Likely` events are inferences (extrapolated OOM ETA, suspected dataloader
//!   stall) and must always read as hedged.

use std::collections::{HashMap, VecDeque};

use crate::model::{fmt_bytes, DeviceId, DynamicSample, ProcessSample, ThrottleReasons};
use serde::{Deserialize, Serialize};

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Severity {
    Info,
    Warning,
    Critical,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Confidence {
    Fact,
    Likely,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventKind {
    ThrottleStart,
    ThrottleEnd,
    ProcessAttached,
    ProcessExited,
    VramPressure,
    IdleGap,
    /// The collector itself fell behind its tick cadence — the recording has a hole, and
    /// the recorder must say so rather than let the gap masquerade as device idleness.
    /// Emitted by the tui collector (the engine owns tick timing, not this module).
    CollectorStall,
    /// History was truncated or restarted (ring wrap on resize, store re-init); consumers
    /// must not treat the discontinuity as device behavior.
    HistoryReset,
    /// Device stopped answering queries while a workload was attached — possibly a hung
    /// kernel or driver. An inference by nature: always `Confidence::Likely`.
    HangSuspected,
    /// A registered device stopped answering its dynamic probe entirely (consecutive
    /// whole-probe failures, not per-metric absence — `NOT_SUPPORTED` stays `None` in the
    /// sample and is never this). Driver reset, NVML dying, eGPU unplug, and an xe rebind
    /// all look identical from this side of the probe, so the kind asserts only the
    /// observed silence — the CAUSE is never claimed. Emitted by the tui collector (it
    /// owns the probe loop and its tick counting). Always `Confidence::Fact`.
    DeviceLost,
    /// A device previously declared lost answered again. The samples between loss and
    /// return were never collected; that gap stays blank in history — a hole, never
    /// zeros. Always `Confidence::Fact`.
    DeviceReturned,
    /// A GPU-attached process is burning CPU while the GPU sits idle — the classic
    /// CPU-bound dataloader. An inference: always `Confidence::Likely`.
    CpuSpillover,
    /// A recording session began folding history into the database — the flight
    /// recorder's own power-on mark. Without it (and its stop twin) the timeline cannot
    /// distinguish "the GPU sat idle" from "gpuviewer wasn't running": unrecorded time
    /// renders blank, and the boundary events are what make that blank legible. Emitted
    /// by the tui collector (it owns the recorder lifecycle). Always `Confidence::Fact`.
    RecordingStarted,
    /// The recording session ended cleanly (the stop mark and the partial rollup tail
    /// reached the store). A session that dies without this mark — SIGKILL, OOM kill,
    /// power loss — writes nothing, which is itself information: the NEXT session's
    /// start mark narrates the missing stop. Always `Confidence::Fact`.
    RecordingStopped,
    /// Writes to the history database started failing (disk full, permissions revoked,
    /// the file removed under a running session) — or, on the recovery edge, started
    /// working again. The recorder swallows the error itself so a bad disk never takes
    /// the live view down with it, but swallowing it SILENTLY is the one thing it must
    /// not do: this product's entire promise is that you can scroll back, and a recorder
    /// that quietly stopped recording breaks that promise exactly when it matters. The
    /// same rule the collector applies to its own stalls, applied to its own storage.
    /// Emitted by the tui collector. Always `Confidence::Fact`.
    RecordingDegraded,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Event {
    pub ts_ms: u64,
    pub device: DeviceId,
    pub kind: EventKind,
    pub severity: Severity,
    pub confidence: Confidence,
    /// One-line human narration ("GPU0 thermal throttling began — clocks 2520→1815 MHz").
    pub title: String,
    /// The raw evidence behind the narration, always auditable.
    pub evidence: String,
}

/// VRAM-trend window length and event cooldown.
const VRAM_WINDOW_MS: u64 = 180_000;
const VRAM_MIN_SPAN_MS: u64 = 60_000;
const VRAM_PRESSURE_FRAC: f64 = 0.85;
const VRAM_MIN_SLOPE_BYTES_PER_MIN: f64 = 16.0 * 1024.0 * 1024.0;
const VRAM_COOLDOWN_MS: u64 = 90_000;

/// Idle-gap (training stall) thresholds: a gap only narrates after sustained activity,
/// only once it lasted long enough to matter, and only if a real allocation stayed
/// attached throughout — otherwise it is just an idle GPU, not a stall.
const IDLE_ACTIVE_UTIL_PCT: f32 = 50.0;
const IDLE_ACTIVE_MIN_MS: u64 = 30_000;
const IDLE_GAP_UTIL_PCT: f32 = 10.0;
const IDLE_GAP_MIN_MS: u64 = 10_000;
const IDLE_HOLDER_MIN_BYTES: u64 = 256 * 1024 * 1024;

/// Hang-suspicion thresholds. The most confidently-wrong-prone inference in the product, so
/// the bar is deliberately steep: VRAM held but engines flat-dead, the holder's own util
/// also flat, and the whole pattern sustained for ten unbroken minutes before we dare say
/// "likely hung". A live trough that recovers, or any GPU activity, must not trip it.
const HANG_DEVICE_UTIL_PCT: f32 = 2.0;
const HANG_PROC_UTIL_PCT: f32 = 2.0;
const HANG_HOLDER_MIN_BYTES: u64 = 1024 * 1024 * 1024;
const HANG_RESET_UTIL_PCT: f32 = 10.0;
const HANG_MIN_MS: u64 = 600_000;

/// CPU-spillover thresholds. A freshly-loaded model that sits on a near-idle GPU while its
/// own process pegs multiple cores is the signature of a partial CPU offload (the model did
/// not fit in VRAM). We assess over a fixed window so a model still warming up is not judged
/// prematurely, and demand a high CPU bar plus a near-dead GPU before claiming it.
const SPILLOVER_HOLDER_MIN_BYTES: u64 = 2 * 1024 * 1024 * 1024;
const SPILLOVER_WINDOW_MS: u64 = 90_000;
const SPILLOVER_MAX_MEAN_UTIL_PCT: f64 = 15.0;
const SPILLOVER_BUSY_UTIL_PCT: f32 = 30.0;
const SPILLOVER_MIN_MEAN_CPU_PCT: f64 = 150.0;
const SPILLOVER_MIN_CPU_SAMPLES: u32 = 3;
/// Minimum observed-util coverage of the assessment window, as a percentage of its ticks.
/// The "GPU stayed ~idle" half of the claim needs the same grounding as the CPU half: when
/// util is `None` throughout (source cannot observe it), the mean would read as 0% and pass
/// the idle gate with zero actual GPU evidence — exactly the confidently-wrong narration the
/// honesty contract bans. Below half-window coverage the judgment is refused, silently.
const SPILLOVER_MIN_UTIL_COVERAGE_PCT: u32 = 50;

#[derive(Default)]
struct DevState {
    prev: Option<DynamicSample>,
    procs: HashMap<u32, ProcessSample>,
    seen_first_procs: bool,
    throttle_since: Option<u64>,
    /// Clock just before throttling began, for the "2520→1815 MHz" narration.
    pre_throttle_clock: Option<u32>,
    vram_window: VecDeque<(u64, u64)>,
    last_pressure_evt: Option<u64>,
    /// ts when util first crossed `IDLE_ACTIVE_UTIL_PCT`; None while below it.
    active_since: Option<u64>,
    /// Latched after `IDLE_ACTIVE_MIN_MS` of sustained activity: a trough only reads
    /// as a training stall if real work preceded it (a desktop idling is no story).
    idle_eligible: bool,
    idle_gap: Option<IdleGap>,
    hang: Option<HangEpisode>,
    /// Open CPU-spillover assessments keyed by the holder's pid; one per new big-memory
    /// process, closed (and judged) when its window elapses or it is cancelled.
    spillovers: HashMap<u32, Spillover>,
}

/// An idle gap in flight: narrated (or discarded) only once util recovers and the
/// gap's duration is actually known.
struct IdleGap {
    start_ms: u64,
    /// Util of the sample just before the drop, for the "92% → 2%" evidence.
    pre_util_pct: f32,
    /// Running mean of util inside the gap (sum and sample count).
    util_sum: f64,
    util_n: u32,
    /// pid → (name, mem at gap start) for processes holding ≥ `IDLE_HOLDER_MIN_BYTES`
    /// when the gap opened; pruned as they exit. Empty at gap end means nobody stayed
    /// attached, so the process_exited event already tells the story.
    holders: HashMap<u32, (String, u64)>,
    /// Set when a `HangSuspected` event fired while this gap was open. A hang is just an
    /// idle gap that lasted long enough to look dead; narrating both for one trough would
    /// double-count the same incident, so the gap stays silent on recovery.
    hang_narrated: bool,
}

/// A hang suspicion in flight: VRAM held with both device and holder engines flat-dead,
/// anchored to the largest qualifying holder. Emits once the pattern survives `HANG_MIN_MS`
/// unbroken; reset (without emitting) the moment activity returns, the holder exits, or
/// util goes unobservable.
struct HangEpisode {
    start_ms: u64,
    /// The largest qualifying holder when the episode opened — the anchor of the narration.
    /// A different (or larger) holder appearing later does not move the anchor; the claim is
    /// about *this* allocation having gone quiet.
    holder_pid: u32,
    holder_name: String,
    holder_mem: u64,
    /// Running mean of device util across the episode, for the evidence line.
    util_sum: f64,
    util_n: u32,
    /// Latched once the event has been emitted, so a sustained hang narrates exactly once.
    fired: bool,
}

/// A CPU-spillover assessment in flight for one freshly-attached big-memory process.
struct Spillover {
    name: String,
    mem_bytes: u64,
    start_ms: u64,
    util_sum: f64,
    util_n: u32,
    cpu_sum: f64,
    cpu_n: u32,
    /// Total ticks the assessment has lived through, observed-util or not — the denominator
    /// for the `SPILLOVER_MIN_UTIL_COVERAGE_PCT` floor.
    tick_n: u32,
}

#[derive(Default)]
pub struct EventEngine {
    state: HashMap<DeviceId, DevState>,
    short_names: HashMap<DeviceId, String>,
}

impl EventEngine {
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a friendly short name ("GPU0") used in narration.
    pub fn register_device(&mut self, id: DeviceId, short_name: String) {
        self.short_names.insert(id, short_name);
    }

    fn short(&self, id: &DeviceId) -> String {
        self.short_names
            .get(id)
            .cloned()
            .unwrap_or_else(|| id.0.clone())
    }

    /// Derive this tick's events.
    ///
    /// `processes` is `None` when the process list was **unobservable** — the backend's
    /// process probe failed outright. That is emphatically not the same as `Some(&[])`,
    /// which asserts the device really had no processes: an empty list makes every
    /// previously-seen pid look like it exited, and `process_events` would narrate a
    /// fact-grade "python (pid 4521) left GPU0, freeing 21.3 GiB" off a driver hiccup.
    /// Fact-grade narration of an event that never happened is the failure mode this
    /// project treats as fatal, so the unobservable case gets its own arm: the four
    /// process-dependent derivations are skipped, and any inference resting on
    /// *continuous* process observation is reset, because this tick broke that premise.
    /// Device-level derivations (throttle, VRAM pressure) are unaffected and keep running.
    pub fn observe(
        &mut self,
        device: &DeviceId,
        sample: &DynamicSample,
        processes: Option<&[ProcessSample]>,
        mem_total: Option<u64>,
        temp_slowdown_c: Option<f32>,
    ) -> Vec<Event> {
        let name = self.short(device);
        let st = self.state.entry(device.clone()).or_default();
        let mut out = Vec::new();

        throttle_events(st, device, &name, sample, temp_slowdown_c, &mut out);
        match processes {
            Some(processes) => {
                // Before process_events flips `seen_first_procs` / overwrites `st.procs`,
                // so the newness diff that opens a spillover window sees this tick's
                // arrivals.
                spillover_events(st, device, &name, sample, processes, &mut out);
                process_events(st, device, &name, sample.ts_ms, processes, &mut out);
                // After process_events, so holder tracking sees this tick's process list.
                hang_events(st, device, &name, sample, &mut out);
                // After hang_events, so a hang that just fired can suppress the gap it
                // lived in.
                idle_gap_events(st, device, &name, sample, &mut out);
            }
            None => {
                // Same discipline as `util_pct: None` below: drop the in-flight
                // inferences rather than carry them across a blind tick. A hang claims a
                // holder was resident for 10 unbroken minutes and an idle gap claims one
                // stayed attached throughout — neither can be said across a tick where
                // nobody could see the process list, and an open spillover window is
                // judging a pid it can no longer observe.
                st.hang = None;
                st.idle_gap = None;
                st.spillovers.clear();
                // `st.procs` is deliberately LEFT INTACT: it is the last observed truth,
                // not a claim about now. Clearing it would make every held process look
                // like it exited on the next good tick — the same false narration by a
                // slower route.
            }
        }
        vram_pressure_events(st, device, &name, sample, mem_total, &mut out);

        st.prev = Some(sample.clone());
        out
    }
}

fn throttle_events(
    st: &mut DevState,
    device: &DeviceId,
    name: &str,
    sample: &DynamicSample,
    temp_slowdown_c: Option<f32>,
    out: &mut Vec<Event>,
) {
    // Throttle unobservable on this source (`None` ≠ "not throttling" — design §5.4):
    // neither a start nor an end can be asserted, so drop the open episode silently —
    // the same blind-spot rule util uses for idle gaps and hangs. Narrating an "end"
    // off a blind spot would be a fabricated fact.
    let Some(throttle) = sample.throttle else {
        st.throttle_since = None;
        st.pre_throttle_clock = None;
        return;
    };
    let prev_any = st
        .prev
        .as_ref()
        .and_then(|p| p.throttle)
        .map(|t| t.any())
        .unwrap_or(false);
    let now_any = throttle.any();

    if !prev_any && now_any {
        let labels = throttle.labels().join(", ");
        let pre_clock = st.prev.as_ref().and_then(|p| p.sm_clock_mhz);
        st.pre_throttle_clock = pre_clock;
        st.throttle_since = Some(sample.ts_ms);

        let clocks = match (pre_clock, sample.sm_clock_mhz) {
            (Some(a), Some(b)) if b < a => format!(" — clocks {a}{b} MHz"),
            _ => String::new(),
        };
        let temp_part = match (sample.temp_c, temp_slowdown_c) {
            (Some(t), Some(thr)) => format!("; {t:.0}°C vs {thr:.0}°C slowdown threshold"),
            (Some(t), None) => format!("; {t:.0}°C"),
            _ => String::new(),
        };
        out.push(Event {
            ts_ms: sample.ts_ms,
            device: device.clone(),
            kind: EventKind::ThrottleStart,
            severity: severity_for(&throttle),
            confidence: Confidence::Fact,
            title: format!("{name} began throttling ({labels}){clocks}"),
            evidence: format!("throttle bits: [{labels}]{temp_part}"),
        });
    } else if prev_any && !now_any {
        let dur = st
            .throttle_since
            .take()
            .map(|t0| format!(" after {}", fmt_dur_ms(sample.ts_ms.saturating_sub(t0))))
            .unwrap_or_default();
        // Only claim "recovered" when clocks are actually back near pre-throttle levels;
        // a throttle that ends because the GPU went idle is not a recovery.
        let clocks = match (st.pre_throttle_clock.take(), sample.sm_clock_mhz) {
            (Some(a), Some(b)) if b as f64 >= a as f64 * 0.9 => {
                format!("; clocks recovered to {b} MHz")
            }
            (Some(a), Some(b)) => format!("; clocks now {b} MHz ({a} MHz pre-throttle)"),
            _ => String::new(),
        };
        out.push(Event {
            ts_ms: sample.ts_ms,
            device: device.clone(),
            kind: EventKind::ThrottleEnd,
            severity: Severity::Info,
            confidence: Confidence::Fact,
            title: format!("{name} stopped throttling{dur}"),
            evidence: format!("throttle bits cleared{clocks}"),
        });
    }
}

fn process_events(
    st: &mut DevState,
    device: &DeviceId,
    name: &str,
    ts_ms: u64,
    processes: &[ProcessSample],
    out: &mut Vec<Event>,
) {
    let now: HashMap<u32, &ProcessSample> = processes.iter().map(|p| (p.pid, p)).collect();

    // Suppress the attach-flood on the very first observation: those processes were already
    // there; narrating them as new would be a lie.
    if st.seen_first_procs {
        for (pid, p) in &now {
            if !st.procs.contains_key(pid) {
                let mem = p
                    .mem_bytes
                    .map(|b| format!(", using {}", fmt_bytes(b)))
                    .unwrap_or_default();
                out.push(Event {
                    ts_ms,
                    device: device.clone(),
                    kind: EventKind::ProcessAttached,
                    severity: Severity::Info,
                    confidence: Confidence::Fact,
                    title: format!("{} (pid {}) attached to {name}{mem}", p.name, pid),
                    evidence: format!("new {} client in process list", p.kind.prose()),
                });
            }
        }
        let gone: Vec<ProcessSample> = st
            .procs
            .values()
            .filter(|p| !now.contains_key(&p.pid))
            .cloned()
            .collect();
        for p in gone {
            let freed = p
                .mem_bytes
                .map(|b| format!(", freeing {}", fmt_bytes(b)))
                .unwrap_or_default();
            out.push(Event {
                ts_ms,
                device: device.clone(),
                kind: EventKind::ProcessExited,
                severity: Severity::Info,
                confidence: Confidence::Fact,
                title: format!("{} (pid {}) left {name}{freed}", p.name, p.pid),
                evidence: format!(
                    "pid {} no longer in process list; last seen holding {}",
                    p.pid,
                    p.mem_bytes
                        .map(fmt_bytes)
                        .unwrap_or_else(|| "unknown memory".into())
                ),
            });
        }
    }
    st.seen_first_procs = true;
    st.procs = now.into_iter().map(|(k, v)| (k, v.clone())).collect();
}

fn idle_gap_events(
    st: &mut DevState,
    device: &DeviceId,
    name: &str,
    sample: &DynamicSample,
    out: &mut Vec<Event>,
) {
    let Some(util) = sample.util_pct else {
        // Utilization went unavailable: we can no longer see activity or idleness, so
        // any gap claim from here on would be guesswork. Drop all tracking instead.
        st.active_since = None;
        st.idle_eligible = false;
        st.idle_gap = None;
        return;
    };

    if let Some(mut gap) = st.idle_gap.take() {
        // Holders must stay attached for the WHOLE gap; one that exits mid-gap is
        // already narrated by process_exited — an idle_gap on top would double-count.
        gap.holders.retain(|pid, _| st.procs.contains_key(pid));

        if util < IDLE_ACTIVE_UTIL_PCT {
            gap.util_sum += util as f64;
            gap.util_n += 1;
            st.idle_gap = Some(gap);
            return;
        }

        // Gap over — its duration is finally known, so decide whether it narrates.
        let dur_ms = sample.ts_ms.saturating_sub(gap.start_ms);
        let holder = gap
            .holders
            .iter()
            .max_by_key(|(_, (_, mem))| *mem)
            .map(|(pid, (pname, mem))| (*pid, pname.clone(), *mem));
        if dur_ms >= IDLE_GAP_MIN_MS && !gap.hang_narrated {
            if let Some((pid, pname, mem)) = holder {
                let dur = fmt_dur_ms(dur_ms);
                let mean_util = gap.util_sum / gap.util_n.max(1) as f64;
                out.push(Event {
                    ts_ms: sample.ts_ms,
                    device: device.clone(),
                    kind: EventKind::IdleGap,
                    severity: Severity::Info,
                    confidence: Confidence::Likely,
                    title: format!(
                        "{name} sat idle {dur} while {pname} (pid {pid}) stayed attached \
                         — likely a dataloader or checkpoint stall"
                    ),
                    evidence: format!(
                        "util {:.0}% → mean {mean_util:.1}% over {dur} ({}..{} ms); \
                         {pname} (pid {pid}) held {} for the whole gap",
                        gap.pre_util_pct,
                        gap.start_ms,
                        sample.ts_ms,
                        fmt_bytes(mem),
                    ),
                });
            }
        }
        // Recovery starts a fresh activity clock: the next gap only narrates after the
        // device has re-earned IDLE_ACTIVE_MIN_MS of sustained work.
        st.active_since = Some(sample.ts_ms);
        st.idle_eligible = false;
        return;
    }

    if util >= IDLE_ACTIVE_UTIL_PCT {
        let since = *st.active_since.get_or_insert(sample.ts_ms);
        if sample.ts_ms.saturating_sub(since) >= IDLE_ACTIVE_MIN_MS {
            st.idle_eligible = true;
        }
        return;
    }

    st.active_since = None;
    if util >= IDLE_GAP_UTIL_PCT || !st.idle_eligible {
        return;
    }
    // Gap opens. Capture who is attached with a real allocation right now; only they
    // can anchor the "stayed attached" claim when the gap ends.
    let holders: HashMap<u32, (String, u64)> = st
        .procs
        .values()
        .filter_map(|p| {
            let mem = p.mem_bytes?;
            (mem >= IDLE_HOLDER_MIN_BYTES).then(|| (p.pid, (p.name.clone(), mem)))
        })
        .collect();
    st.idle_gap = Some(IdleGap {
        start_ms: sample.ts_ms,
        pre_util_pct: st.prev.as_ref().and_then(|p| p.util_pct).unwrap_or(util),
        util_sum: util as f64,
        util_n: 1,
        holders,
        hang_narrated: false,
    });
}

/// `HangSuspected` — VRAM held, engines flat-dead, holder alive: the job has likely hung.
///
/// An inference of the riskiest kind, so the gate is steep: the device must be effectively
/// idle (`≤ HANG_DEVICE_UTIL_PCT`), a holder must be sitting on ≥ 1 GiB while its *own*
/// engine activity is also flat (or unreported), and that exact pattern must survive a full
/// `HANG_MIN_MS` without a break before we narrate. We anchor to the largest qualifying
/// holder at episode start and never re-anchor: the claim is that *this* allocation went
/// quiet. The episode is dropped (never narrated) the instant any premise stops holding —
/// the device wakes up, the holder exits, util goes unobservable, or even a sub-throttle
/// flicker of activity — because a hang we cannot stand fully behind is worse than silence.
fn hang_events(
    st: &mut DevState,
    device: &DeviceId,
    name: &str,
    sample: &DynamicSample,
    out: &mut Vec<Event>,
) {
    let Some(util) = sample.util_pct else {
        // Util unobservable: we cannot see "zero engine activity", so we cannot claim a
        // hang. Drop the episode rather than freeze a stale window across the blind spot.
        st.hang = None;
        return;
    };

    // The largest holder that is itself quiet: ≥ 1 GiB resident with its own util flat or
    // simply not reported (a hung kernel reports no per-process util — absence is expected).
    let candidate = st
        .procs
        .values()
        .filter(|p| p.mem_bytes.unwrap_or(0) >= HANG_HOLDER_MIN_BYTES)
        .filter(|p| p.util_pct.map(|u| u <= HANG_PROC_UTIL_PCT).unwrap_or(true))
        .max_by_key(|p| p.mem_bytes.unwrap_or(0));
    let condition = util <= HANG_DEVICE_UTIL_PCT && candidate.is_some();

    if let Some(mut ep) = st.hang.take() {
        // The anchored holder must still be alive; if it exited, `process_exited` already
        // told the story and the premise ("process still alive") is gone.
        let holder_alive = st.procs.contains_key(&ep.holder_pid);
        if util > HANG_RESET_UTIL_PCT || !holder_alive || !condition {
            // Any break ends the episode silently — continuity is the whole claim.
            return;
        }
        ep.util_sum += util as f64;
        ep.util_n += 1;
        let elapsed = sample.ts_ms.saturating_sub(ep.start_ms);
        if elapsed >= HANG_MIN_MS && !ep.fired {
            ep.fired = true;
            let mean_util = ep.util_sum / ep.util_n.max(1) as f64;
            let dur = fmt_dur_ms(elapsed);
            out.push(Event {
                ts_ms: sample.ts_ms,
                device: device.clone(),
                kind: EventKind::HangSuspected,
                severity: Severity::Warning,
                confidence: Confidence::Likely,
                title: format!(
                    "{name}: {} (pid {}) likely hung — held {} for {dur} with zero GPU \
                     activity, process still alive",
                    ep.holder_name,
                    ep.holder_pid,
                    fmt_bytes(ep.holder_mem),
                ),
                evidence: format!(
                    "device util mean {mean_util:.1}% over {dur} ({}..{} ms); \
                     {} (pid {}) held {} throughout while its own engine activity stayed flat",
                    ep.start_ms,
                    sample.ts_ms,
                    ep.holder_name,
                    ep.holder_pid,
                    fmt_bytes(ep.holder_mem),
                ),
            });
            // A hang is an idle gap that lasted too long to look alive; if a gap is still
            // open over this same trough, mute it so one incident is narrated once.
            if let Some(gap) = st.idle_gap.as_mut() {
                gap.hang_narrated = true;
            }
        }
        st.hang = Some(ep);
        return;
    }

    if condition {
        let holder = candidate.expect("condition implies a candidate");
        st.hang = Some(HangEpisode {
            start_ms: sample.ts_ms,
            holder_pid: holder.pid,
            holder_name: holder.name.clone(),
            holder_mem: holder.mem_bytes.unwrap_or(0),
            util_sum: util as f64,
            util_n: 1,
            fired: false,
        });
    }
}

/// `CpuSpillover` — a freshly-loaded model whose GPU stays idle while its process burns
/// CPU: the signature of a partial CPU offload (the model did not fit in VRAM).
///
/// We open a fixed `SPILLOVER_WINDOW_MS` assessment when a *new* process attaches holding
/// ≥ 2 GiB, then judge at window close: narrate only if the GPU averaged near-idle while the
/// process averaged multiple busy cores, with enough CPU samples to mean it. The assessment
/// is cancelled silently — never narrated — if the process exits mid-window, the GPU shows
/// real use at any point, or (honesty rule) we never once saw its CPU: with no CPU
/// visibility we cannot claim it is "burning CPU", so we say nothing rather than guess.
/// The same honesty rule covers the GPU side: util must have been observed on at least
/// half the window's ticks, else "the GPU is ~idle" would rest on no observation at all.
fn spillover_events(
    st: &mut DevState,
    device: &DeviceId,
    name: &str,
    sample: &DynamicSample,
    processes: &[ProcessSample],
    out: &mut Vec<Event>,
) {
    let now: HashMap<u32, &ProcessSample> = processes.iter().map(|p| (p.pid, p)).collect();

    // Open a window for each newly-attached big-memory holder. Skip the first observation:
    // those processes were already resident, not freshly loaded, so they are no story.
    if st.seen_first_procs {
        for (pid, p) in &now {
            if st.procs.contains_key(pid) || st.spillovers.contains_key(pid) {
                continue;
            }
            if p.mem_bytes.unwrap_or(0) >= SPILLOVER_HOLDER_MIN_BYTES {
                st.spillovers.insert(
                    *pid,
                    Spillover {
                        name: p.name.clone(),
                        mem_bytes: p.mem_bytes.unwrap_or(0),
                        start_ms: sample.ts_ms,
                        util_sum: 0.0,
                        util_n: 0,
                        cpu_sum: 0.0,
                        cpu_n: 0,
                        tick_n: 0,
                    },
                );
            }
        }
    }

    if st.spillovers.is_empty() {
        return;
    }

    // A device showing real use cancels every open assessment at once: the premise of the
    // whole inference is that the GPU is idle, and one busy reading refutes it.
    let gpu_busy = sample
        .util_pct
        .map(|u| u >= SPILLOVER_BUSY_UTIL_PCT)
        .unwrap_or(false);

    let mut to_emit: Vec<Event> = Vec::new();
    st.spillovers.retain(|pid, sp| {
        if gpu_busy {
            return false;
        }
        let Some(p) = now.get(pid) else {
            // Exited mid-window: cancelled silently (its `process_exited` fact stands).
            return false;
        };
        sp.tick_n += 1;
        if let Some(u) = sample.util_pct {
            sp.util_sum += u as f64;
            sp.util_n += 1;
        }
        if let Some(c) = p.cpu_pct {
            sp.cpu_sum += c as f64;
            sp.cpu_n += 1;
        }

        if sample.ts_ms.saturating_sub(sp.start_ms) < SPILLOVER_WINDOW_MS {
            return true; // window still open
        }

        // Window closed — judge. Means require samples; no CPU sample at all means no CPU
        // visibility, and we refuse to claim a CPU burn we never observed. Symmetrically,
        // util must have been *observed* on at least half the window's ticks: a blind
        // window (util None throughout) would otherwise mean 0% and fabricate "the GPU
        // is ~idle" with zero actual GPU evidence.
        let util_grounded =
            sp.util_n > 0 && sp.util_n * 100 >= sp.tick_n * SPILLOVER_MIN_UTIL_COVERAGE_PCT;
        let mean_util = sp.util_sum / sp.util_n.max(1) as f64;
        let mean_cpu = sp.cpu_sum / sp.cpu_n.max(1) as f64;
        if util_grounded
            && sp.cpu_n >= SPILLOVER_MIN_CPU_SAMPLES
            && mean_util < SPILLOVER_MAX_MEAN_UTIL_PCT
            && mean_cpu >= SPILLOVER_MIN_MEAN_CPU_PCT
        {
            let span = fmt_dur_ms(sample.ts_ms.saturating_sub(sp.start_ms));
            to_emit.push(Event {
                ts_ms: sample.ts_ms,
                device: device.clone(),
                kind: EventKind::CpuSpillover,
                severity: Severity::Warning,
                confidence: Confidence::Likely,
                title: format!(
                    "{} (pid {pid}) loaded {} but {name} is ~idle while its CPU runs hot \
                     — likely partial CPU offload (model may not fit in VRAM)",
                    sp.name,
                    fmt_bytes(sp.mem_bytes),
                ),
                evidence: format!(
                    "over {span} ({}..{} ms): {name} util mean {mean_util:.1}%, \
                     {} (pid {pid}) CPU mean {mean_cpu:.0}% of one core ({} samples)",
                    sp.start_ms, sample.ts_ms, sp.name, sp.cpu_n,
                ),
            });
        }
        false // window done either way
    });
    out.extend(to_emit);
}

fn vram_pressure_events(
    st: &mut DevState,
    device: &DeviceId,
    name: &str,
    sample: &DynamicSample,
    mem_total: Option<u64>,
    out: &mut Vec<Event>,
) {
    let (Some(used), Some(total)) = (sample.mem_used_bytes, mem_total) else {
        return;
    };
    if total == 0 {
        return;
    }

    // A sharp drop (process exit, allocator reset) invalidates the trend: an endpoint
    // slope over a window straddling the old peak would understate the *current* climb
    // rate — wrong in the dangerous direction. Restart the window instead.
    if let Some(&(_, last_used)) = st.vram_window.back() {
        if last_used.saturating_sub(used) > total / 20 {
            st.vram_window.clear();
        }
    }

    st.vram_window.push_back((sample.ts_ms, used));
    while let Some(&(t0, _)) = st.vram_window.front() {
        if sample.ts_ms.saturating_sub(t0) > VRAM_WINDOW_MS {
            st.vram_window.pop_front();
        } else {
            break;
        }
    }

    let frac = used as f64 / total as f64;
    if frac < VRAM_PRESSURE_FRAC {
        return;
    }
    let (&(t0, b0), &(t1, b1)) = match (st.vram_window.front(), st.vram_window.back()) {
        (Some(a), Some(b)) if t_span(a, b) >= VRAM_MIN_SPAN_MS => (a, b),
        _ => return,
    };
    let span_min = (t1 - t0) as f64 / 60_000.0;
    let slope_per_min = (b1 as f64 - b0 as f64) / span_min;
    if slope_per_min < VRAM_MIN_SLOPE_BYTES_PER_MIN {
        return;
    }
    if let Some(last) = st.last_pressure_evt {
        if sample.ts_ms.saturating_sub(last) < VRAM_COOLDOWN_MS {
            return;
        }
    }
    st.last_pressure_evt = Some(sample.ts_ms);

    let headroom = total.saturating_sub(used) as f64;
    let eta_min = headroom / slope_per_min;
    // Only name a "largest holder" when at least one process has a *known* size —
    // with mem_bytes all-None (WSL2, unprivileged fdinfo) max_by_key would crown an
    // arbitrary process on zero evidence.
    let grower = st
        .procs
        .values()
        .filter(|p| p.mem_bytes.is_some())
        .max_by_key(|p| p.mem_bytes)
        .map(|p| format!(" (largest holder: {} pid {})", p.name, p.pid))
        .unwrap_or_default();

    out.push(Event {
        ts_ms: sample.ts_ms,
        device: device.clone(),
        kind: EventKind::VramPressure,
        severity: Severity::Warning,
        confidence: Confidence::Likely,
        title: format!(
            "{name} VRAM {:.0}% and climbing ~{}/min — likely full in ~{:.0} min{grower}",
            frac * 100.0,
            fmt_bytes(slope_per_min as u64),
            eta_min
        ),
        evidence: format!(
            "used {}/{} ({:.1}%); slope +{}/min over last {:.1} min (linear extrapolation)",
            fmt_bytes(used),
            fmt_bytes(total),
            frac * 100.0,
            fmt_bytes(slope_per_min as u64),
            span_min
        ),
    });
}

fn severity_for(t: &ThrottleReasons) -> Severity {
    if t.hw_slowdown {
        Severity::Critical
    } else {
        Severity::Warning
    }
}

fn t_span(a: &(u64, u64), b: &(u64, u64)) -> u64 {
    b.0.saturating_sub(a.0)
}

fn fmt_dur_ms(ms: u64) -> String {
    // Round to nearest second — a 3.9s episode is "4s", not "3s".
    let s = (ms + 500) / 1000;
    if s >= 60 {
        format!("{}m {}s", s / 60, s % 60)
    } else {
        format!("{s}s")
    }
}

#[cfg(test)]
mod tests {
    //! CPU-spillover observed-util coverage floor — regression tests for the honesty rule
    //! that "the GPU is ~idle" must rest on real util observations, never on a mean over
    //! zero (or too few) samples. The broader event-derivation suite lives in `lib.rs`;
    //! these mirror its synthetic 1 Hz trace style.

    use super::*;
    use crate::model::ProcessKind;

    /// A 1 Hz sample whose util may be absent — exercises the coverage-floor paths.
    fn opt_sample(ts_ms: u64, util_pct: Option<f32>) -> DynamicSample {
        DynamicSample {
            ts_ms,
            util_pct,
            util_engine: None,
            mem_used_bytes: Some(8 << 30),
            power_mw: None,
            temp_c: None,
            fan_pct: None,
            sm_clock_mhz: None,
            mem_clock_mhz: None,
            encoder_pct: None,
            decoder_pct: None,
            throttle: Some(ThrottleReasons::default()),
        }
    }

    /// Build a process holding `mem` bytes, with optional self-util and CPU%.
    fn proc_with(
        pid: u32,
        name: &str,
        mem: u64,
        util_pct: Option<f32>,
        cpu_pct: Option<f32>,
    ) -> ProcessSample {
        ProcessSample {
            pid,
            name: name.into(),
            kind: ProcessKind::Compute,
            mem_bytes: Some(mem),
            util_pct,
            cpu_pct,
            container: None,
        }
    }

    /// Drive a 1 Hz trace with an optional util value, holding `procs` constant across it.
    fn drive_opt(
        engine: &mut EventEngine,
        dev: &DeviceId,
        ts_range: std::ops::RangeInclusive<u64>,
        util_pct: Option<f32>,
        procs: &[ProcessSample],
    ) -> Vec<Event> {
        let mut out = Vec::new();
        for ts in ts_range.step_by(1000) {
            out.extend(engine.observe(
                dev,
                &opt_sample(ts, util_pct),
                Some(procs),
                Some(16 << 30),
                None,
            ));
        }
        out
    }

    /// Util `None` on every tick of the window: the source cannot observe the GPU at all.
    /// Before the coverage floor, the mean read as 0/max(1) = 0% and passed the idle gate —
    /// narrating "the GPU is ~idle" with zero actual GPU evidence. Must stay silent.
    #[test]
    fn spillover_silent_when_util_unobserved_all_window() {
        let mut engine = EventEngine::new();
        let dev = DeviceId("test".into());

        // Baseline tick with no procs so the model reads as freshly attached.
        drive_opt(&mut engine, &dev, 0..=0, None, &[]);
        // Hot CPU the whole window, but device util is None throughout.
        let ollama = vec![proc_with(7777, "ollama", 12 << 30, Some(0.0), Some(310.0))];
        let out = drive_opt(&mut engine, &dev, 1_000..=91_000, None, &ollama);
        assert!(
            out.iter().all(|e| e.kind != EventKind::CpuSpillover),
            "util never observed — the GPU-idle claim has no evidence and must stay silent"
        );
    }

    /// Util observed on fewer than half the window's ticks (31 of 91), all of them low:
    /// the observed mean would pass the idle gate, but the coverage floor refuses the
    /// judgment — too much of the window is a blind spot to mean it.
    #[test]
    fn spillover_silent_when_util_coverage_below_half_window() {
        let mut engine = EventEngine::new();
        let dev = DeviceId("test".into());

        drive_opt(&mut engine, &dev, 0..=0, None, &[]);
        let ollama = vec![proc_with(7777, "ollama", 12 << 30, Some(0.0), Some(310.0))];
        // Blind for the first 60 ticks, observed-low for the last 31.
        let mut out = drive_opt(&mut engine, &dev, 1_000..=60_000, None, &ollama);
        out.extend(drive_opt(
            &mut engine,
            &dev,
            61_000..=91_000,
            Some(5.0),
            &ollama,
        ));
        assert!(
            out.iter().all(|e| e.kind != EventKind::CpuSpillover),
            "31 of 91 ticks observed is below half-window coverage — must stay silent"
        );
    }

    /// Coverage just over the floor (46 of 91 ticks observed, all low) restores the claim:
    /// the floor blocks blindness, not legitimate partial visibility.
    #[test]
    fn spillover_fires_once_coverage_reaches_half_window() {
        let mut engine = EventEngine::new();
        let dev = DeviceId("test".into());
        engine.register_device(dev.clone(), "GPU0".into());

        drive_opt(&mut engine, &dev, 0..=0, None, &[]);
        let ollama = vec![proc_with(7777, "ollama", 12 << 30, Some(0.0), Some(310.0))];
        // Blind for 45 ticks, observed-low for 46: 46/91 clears the half-window floor.
        let mut out = drive_opt(&mut engine, &dev, 1_000..=45_000, None, &ollama);
        out.extend(drive_opt(
            &mut engine,
            &dev,
            46_000..=91_000,
            Some(5.0),
            &ollama,
        ));
        let n = out
            .iter()
            .filter(|e| e.kind == EventKind::CpuSpillover)
            .count();
        assert_eq!(
            n, 1,
            "46 of 91 ticks observed clears the floor — the grounded claim must narrate once"
        );
    }

    /// The fully-observed near-idle window still narrates: the floor must not silence the
    /// textbook case it exists to protect.
    #[test]
    fn spillover_still_fires_with_observed_low_util() {
        let mut engine = EventEngine::new();
        let dev = DeviceId("test".into());
        engine.register_device(dev.clone(), "GPU0".into());

        drive_opt(&mut engine, &dev, 0..=0, Some(3.0), &[]);
        let ollama = vec![proc_with(7777, "ollama", 12 << 30, Some(0.0), Some(310.0))];
        let out = drive_opt(&mut engine, &dev, 1_000..=91_000, Some(5.0), &ollama);
        let n = out
            .iter()
            .filter(|e| e.kind == EventKind::CpuSpillover)
            .count();
        assert_eq!(
            n, 1,
            "fully-observed near-idle GPU plus hot CPU must still narrate exactly once"
        );
    }
}