feat(capture): stall attribution phases A.2+A.3 — micro-probes and DxgKrnl ETW name the class

Phase A.2: a refcounted micro-probe engine (fence round-trip per adapter,
DwmGetCompositionTimingInfo tick, watchdogged DwmFlush, Level-Zero
D3DKMTGetScanLine, CPU jitter sentinel) samples continuously on detached
sacrificial threads; each stall report reads the window back and the verdict
matrix folds it with the driver telemetry into a named class: ours-worker /
ours-delivery / CLASS-1 adapter freeze / CLASS-2 compositor blocked /
frame-generation / unattributed. The metronomic WARN carries the per-class
session tally.

Phase A.3: an event-id-filtered real-time ETW session on
Microsoft-Windows-DxgKrnl (QueryChildStatus 150/151, SetPowerState 154/155,
IndicateChildStatus 272, SetTimingsFromVidPn 430, DisplayDetectControl
1096/1097) rides every stall line as a DDI bracket summary — naming the
servicing call and its duration instead of 'below Windows'. Degrades to
etw=unavailable without admin; probes degrade per-leg (absence is stated,
never guessed).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/pf-capture/Cargo.toml

@@ -47,14 +47,17 @@ windows = { version = "0.62", features = [
     "Win32_Graphics_Direct3D",
     "Win32_Graphics_Direct3D11",
     "Win32_Graphics_Direct3D_Fxc",
+    "Win32_Graphics_Dwm",
     "Win32_Graphics_Dxgi",
     "Win32_Graphics_Dxgi_Common",
     "Win32_Graphics_Gdi",
+    "Win32_System_Diagnostics_Etw",
     "Win32_System_LibraryLoader",
     "Win32_System_StationsAndDesktops",
     "Win32_System_Memory",
     "Win32_System_Performance",
     "Win32_System_Threading",
+    "Win32_System_Time",
     "Win32_UI_HiDpi",
     "Win32_UI_Input_KeyboardAndMouse",
     "Win32_UI_WindowsAndMessaging",

crates/pf-capture/src/windows/idd_push.rs

@@ -322,6 +322,12 @@ mod cursor_blend;
 mod cursor_poll;
 #[path = "idd_push/descriptor.rs"]
 mod descriptor;
+// Stall attribution (vdisplay-disturbance-immunity Phase A): the DxgKrnl ETW watch (A.3), the
+// micro-probe engine (A.2), and the stall watch + verdict matrix that folds them (A.1).
+#[path = "idd_push/dxgkrnl_etw.rs"]
+mod dxgkrnl_etw;
+#[path = "idd_push/probes.rs"]
+mod probes;
 #[path = "idd_push/stall.rs"]
 mod stall;
 use channel::ChannelBroker;
@@ -489,6 +495,12 @@ pub struct IddPushCapturer {
     /// ever read (µs) while the host starved since then. Rolled at every fresh frame.
     offered_at_fresh: u64,
     max_hb_age_us: u64,
+    /// The Phase A.2 micro-probe engine (refcounted process singleton) — its window read rides
+    /// every stall report so the verdict matrix can name the disturbance class.
+    probes: Arc<probes::ProbeEngine>,
+    /// The Phase A.3 DxgKrnl ETW watch; `None` when the session can't start (non-admin dev run)
+    /// — reports then say `etw=unavailable`.
+    etw: Option<Arc<dxgkrnl_etw::EtwWatch>>,
     /// Host-owned ROTATING output ring NVENC encodes (one YUV texture per slot). Rotating it per frame
     /// is the precondition for pipelining the encode loop: while NVENC encodes frame N's texture on the
     /// ASIC, frame N+1's convert writes a DIFFERENT texture — the two overlap. Format = `out_format()`:
@@ -1597,6 +1609,15 @@ impl IddPushCapturer {
                     }
                 }),
                 max_heartbeat_age_ms: self.max_hb_age_us / 1_000,
+                // The probe + ETW reads span the same window the report's OS-event correlation
+                // uses (the gap plus a lead-in for the disturbance that CAUSED it).
+                probes: now
+                    .checked_sub(stall.gap + Duration::from_millis(300))
+                    .map(|from| self.probes.window(from, now)),
+                etw: self.etw.as_ref().and_then(|w| {
+                    now.checked_sub(stall.gap + Duration::from_millis(300))
+                        .map(|from| w.summary(from, now))
+                }),
             };
             self.stall_watch.report(&stall, now, &evidence);
         }
@@ -2123,6 +2144,8 @@ mod tests {
                 &StallEvidence {
                     offered_delta: offered,
                     max_heartbeat_age_ms: hb_age_ms,
+                    probes: None,
+                    etw: None,
                 },
             )
         };
@@ -2143,4 +2166,97 @@ mod tests {
         assert_eq!(verdict(3_000, Some(2), 900), StallVerdict::ComposeSilence);
         assert_eq!(verdict(3_000, Some(2), 1_600), StallVerdict::WorkerStalled);
     }
+
+    /// [`stall::classify`]'s verdict matrix — how the micro-probe window refines (or declines to
+    /// refine) the driver-telemetry verdict into a named disturbance class.
+    #[test]
+    fn stall_classification_matrix() {
+        use super::stall::{classify, ProbeWindow, StallClass, StallVerdict};
+        let gap = Duration::from_millis(600);
+        let probes = |fence: Option<u64>, dwm: Option<u64>, flush: Option<u64>| ProbeWindow {
+            fence_max_us: fence,
+            dwm_tick_frozen_us: dwm,
+            dwm_flush_max_us: flush,
+            ..ProbeWindow::default()
+        };
+        // The driver's own verdicts win outright — probes can't overrule "we lost the frames".
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::WorkerStalled,
+                Some(&probes(Some(500_000), None, None))
+            ),
+            StallClass::OursWorker
+        );
+        assert_eq!(
+            classify(gap, &StallVerdict::DeliveryLeg, None),
+            StallClass::OursDelivery
+        );
+        // No probes: compose-silence alone can't name a class.
+        assert_eq!(
+            classify(gap, &StallVerdict::ComposeSilence, None),
+            StallClass::Unattributed
+        );
+        // Fences stalled ≥ gap/2 → the adapter froze — Class 1 (even without driver telemetry).
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::ComposeSilence,
+                Some(&probes(Some(400_000), Some(400_000), None))
+            ),
+            StallClass::AdapterFreeze
+        );
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::NoTelemetry,
+                Some(&probes(Some(400_000), None, None))
+            ),
+            StallClass::AdapterFreeze
+        );
+        // Fences fine (16 ms round-trips) but DWM's tick froze — Class 2; DwmFlush counts too.
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::ComposeSilence,
+                Some(&probes(Some(16_000), Some(500_000), None))
+            ),
+            StallClass::CompositorBlocked
+        );
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::ComposeSilence,
+                Some(&probes(Some(16_000), Some(20_000), Some(450_000)))
+            ),
+            StallClass::CompositorBlocked
+        );
+        // Everything alive + the driver swears E_PENDING → the frame-generation path.
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::ComposeSilence,
+                Some(&probes(Some(16_000), Some(20_000), Some(30_000)))
+            ),
+            StallClass::FrameGeneration
+        );
+        // Healthy probes but a pre-telemetry driver: delivery-leg is equally possible — honest.
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::NoTelemetry,
+                Some(&probes(Some(16_000), Some(20_000), None))
+            ),
+            StallClass::Unattributed
+        );
+        // An absent probe (None) never reads as "stalled" — absence is stated, not guessed.
+        assert_eq!(
+            classify(
+                gap,
+                &StallVerdict::ComposeSilence,
+                Some(&probes(None, Some(20_000), Some(30_000)))
+            ),
+            StallClass::FrameGeneration
+        );
+    }
 }

crates/pf-capture/src/windows/idd_push/dxgkrnl_etw.rs

@@ -0,0 +1,366 @@
+//! Phase A.3 DxgKrnl ETW correlation (stall attribution, `vdisplay-disturbance-immunity.md`
+//! §4.3): a tiny real-time ETW session on `Microsoft-Windows-DxgKrnl`, event-id-filtered to the
+//! five display-miniport DDI families whose servicing freezes the present path, so a stall report
+//! can NAME the DDI (and its duration bracket) instead of saying "below Windows":
+//!
+//! - 150/151 `QueryChildStatus` start/stop — connector polling (the DDC/child-I/O class),
+//! - 272 `IndicateChildStatus` — the miniport itself reporting a connector change,
+//! - 154/155 `SetPowerState` start/stop — monitor/link power transitions (Class-1 servicing),
+//! - 430 `SetTimingsFromVidPn` — modeset-class commits (Level-Two "hardware is idle" freezes),
+//! - 1096/1097 `DisplayDetectControl` start/stop — present on newer builds; filtered in
+//!   unconditionally, harmless where absent.
+//!
+//! Kernel-side event-id filtering (`EVENT_FILTER_TYPE_EVENT_ID`) keeps the per-vblank firehose
+//! off; the session costs a few events per minute. Starting a real-time session needs admin /
+//! Performance Log Users — the packaged host (service, SYSTEM) has it; a plain dev run degrades
+//! to `None` and every report says `etw=unavailable` instead of guessing. The session's
+//! `FlushTimer` is 1 s, so a bracket from the trailing second of a gap can land AFTER that
+//! stall's report line — the next report (and the metronomic tally) still carries it.
+
+// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it (unsafe-proof program).
+#![deny(clippy::undocumented_unsafe_blocks)]
+
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex, OnceLock, Weak};
+use std::time::{Duration, Instant};
+
+use windows::core::{GUID, PWSTR};
+use windows::Win32::Foundation::ERROR_SUCCESS;
+use windows::Win32::System::Diagnostics::Etw::{
+    CloseTrace, ControlTraceW, EnableTraceEx2, OpenTraceW, ProcessTrace, StartTraceW,
+    CONTROLTRACE_HANDLE, ENABLE_TRACE_PARAMETERS, ENABLE_TRACE_PARAMETERS_VERSION_2,
+    EVENT_CONTROL_CODE_ENABLE_PROVIDER, EVENT_FILTER_DESCRIPTOR, EVENT_FILTER_TYPE_EVENT_ID,
+    EVENT_RECORD, EVENT_TRACE_CONTROL_STOP, EVENT_TRACE_LOGFILEW, EVENT_TRACE_PROPERTIES,
+    EVENT_TRACE_REAL_TIME_MODE, PROCESSTRACE_HANDLE, PROCESS_TRACE_MODE_EVENT_RECORD,
+    PROCESS_TRACE_MODE_REAL_TIME, TRACE_LEVEL_INFORMATION, WNODE_FLAG_TRACED_GUID,
+};
+use windows::Win32::System::Performance::{QueryPerformanceCounter, QueryPerformanceFrequency};
+
+/// `Microsoft-Windows-DxgKrnl` (`{802EC45A-1E99-4B83-9920-87C98277BA9D}`).
+const DXGKRNL: GUID = GUID::from_u128(0x802EC45A_1E99_4B83_9920_87C98277BA9D);
+
+/// The event ids the session filters IN (see the module docs).
+const FILTER_IDS: [u16; 8] = [150, 151, 272, 154, 155, 430, 1096, 1097];
+
+/// Session name — ours, stopped-if-stale at start (a crashed host leaves the session behind;
+/// real-time sessions are machine-global named objects).
+const SESSION: &str = "punktfunk-stallwatch-dxgkrnl";
+
+/// The consumer callback's destination: `(event QPC, event id)`, capped. A few events per minute
+/// in the field; the cap only matters under a detection storm — exactly when the tail is the
+/// least interesting part.
+static RING: Mutex<VecDeque<(i64, u16)>> = Mutex::new(VecDeque::new());
+
+fn qpc_now() -> i64 {
+    let mut v = 0i64;
+    // SAFETY: plain FFI; `v` is a valid local out-param.
+    let _ = unsafe { QueryPerformanceCounter(&mut v) };
+    v
+}
+
+fn qpc_freq() -> i64 {
+    static FREQ: OnceLock<i64> = OnceLock::new();
+    *FREQ.get_or_init(|| {
+        let mut f = 0i64;
+        // SAFETY: plain FFI; `f` is a valid local out-param; fixed at boot.
+        let _ = unsafe { QueryPerformanceFrequency(&mut f) };
+        f.max(1)
+    })
+}
+
+/// The consumer's per-event callback — record id + QPC timestamp (the session's `ClientContext`
+/// is 1, so `TimeStamp` IS a QPC value) and return; runs on the consumer thread.
+unsafe extern "system" fn on_event(record: *mut EVENT_RECORD) {
+    if record.is_null() {
+        return;
+    }
+    // SAFETY: `record` is the live event the consumer is delivering for the duration of this call.
+    let (id, ts) = unsafe {
+        (
+            (*record).EventHeader.EventDescriptor.Id,
+            (*record).EventHeader.TimeStamp,
+        )
+    };
+    let mut ring = RING.lock().unwrap();
+    if ring.len() == 2048 {
+        ring.pop_front();
+    }
+    ring.push_back((ts, id));
+}
+
+/// A live DxgKrnl watch: the controller handle (stops the session on drop) + the consumer handle.
+pub(super) struct EtwWatch {
+    session: CONTROLTRACE_HANDLE,
+    consumer: PROCESSTRACE_HANDLE,
+}
+
+// SAFETY: both fields are plain kernel handle VALUES (u64 wrappers) owned by this watch; every
+// operation on them (summary reads the static ring; Drop stops/closes) is thread-safe by the ETW
+// API contract, and the singleton hands out only `Arc<EtwWatch>`.
+unsafe impl Send for EtwWatch {}
+// SAFETY: as above — `&EtwWatch` exposes only `summary` (static-ring reads).
+unsafe impl Sync for EtwWatch {}
+
+static WATCH: Mutex<Weak<EtwWatch>> = Mutex::new(Weak::new());
+
+/// The process-wide watch, started on first use; `None` when the session cannot start (no admin
+/// rights on a dev run) — callers report `etw=unavailable` rather than guessing.
+pub(super) fn acquire() -> Option<Arc<EtwWatch>> {
+    let mut g = WATCH.lock().unwrap();
+    if let Some(w) = g.upgrade() {
+        return Some(w);
+    }
+    let w = Arc::new(EtwWatch::start()?);
+    *g = Arc::downgrade(&w);
+    Some(w)
+}
+
+/// An `EVENT_TRACE_PROPERTIES` allocation with the session-name space ETW writes into appended —
+/// the canonical controller-buffer pattern.
+fn properties_buffer() -> (Vec<u8>, usize) {
+    let base = std::mem::size_of::<EVENT_TRACE_PROPERTIES>();
+    let total = base + (SESSION.len() + 1) * 2;
+    (vec![0u8; total], base)
+}
+
+impl EtwWatch {
+    fn start() -> Option<Self> {
+        let name: Vec<u16> = SESSION.encode_utf16().chain([0]).collect();
+        // A stale session from a crashed host blocks StartTrace with ERROR_ALREADY_EXISTS — stop
+        // it by name first (fails benignly when there is none).
+        let (mut stop_buf, _) = properties_buffer();
+        // SAFETY: `stop_buf` is a live, zeroed, correctly-sized properties allocation; the name is
+        // a live nul-terminated wide string; a session handle of 0 + name = control-by-name.
+        unsafe {
+            let props = stop_buf.as_mut_ptr().cast::<EVENT_TRACE_PROPERTIES>();
+            (*props).Wnode.BufferSize = stop_buf.len() as u32;
+            let _ = ControlTraceW(
+                CONTROLTRACE_HANDLE::default(),
+                PWSTR(name.as_ptr() as *mut _),
+                props,
+                EVENT_TRACE_CONTROL_STOP,
+            );
+        }
+
+        let (mut buf, base) = properties_buffer();
+        let mut session = CONTROLTRACE_HANDLE::default();
+        // SAFETY: `buf` is a live, zeroed allocation of base + name bytes; every write below is a
+        // field of the properties struct at its head; `LoggerNameOffset = base` points at the
+        // appended name space (ETW copies the name there itself). ClientContext 1 = QPC clock —
+        // what makes event timestamps comparable to our probe windows.
+        let rc = unsafe {
+            let props = buf.as_mut_ptr().cast::<EVENT_TRACE_PROPERTIES>();
+            (*props).Wnode.BufferSize = buf.len() as u32;
+            (*props).Wnode.Flags = WNODE_FLAG_TRACED_GUID;
+            (*props).Wnode.ClientContext = 1;
+            (*props).LogFileMode = EVENT_TRACE_REAL_TIME_MODE;
+            (*props).BufferSize = 64;
+            (*props).MinimumBuffers = 2;
+            (*props).MaximumBuffers = 4;
+            (*props).FlushTimer = 1;
+            (*props).LoggerNameOffset = base as u32;
+            StartTraceW(&mut session, PWSTR(name.as_ptr() as *mut _), props)
+        };
+        if rc != ERROR_SUCCESS {
+            tracing::debug!(
+                rc = rc.0,
+                "DxgKrnl ETW session unavailable (needs admin / Performance Log Users) — \
+                 stall reports will say etw=unavailable"
+            );
+            return None;
+        }
+
+        // Enable DxgKrnl with a kernel-side event-id filter — the whole point: the provider's
+        // vblank/DPC keywords never reach us.
+        let mut filter = Vec::with_capacity(4 + FILTER_IDS.len() * 2);
+        filter.extend_from_slice(&[1u8, 0u8]); // FilterIn = TRUE, Reserved
+        filter.extend_from_slice(&(FILTER_IDS.len() as u16).to_le_bytes());
+        for id in FILTER_IDS {
+            filter.extend_from_slice(&id.to_le_bytes());
+        }
+        let mut desc = EVENT_FILTER_DESCRIPTOR {
+            Ptr: filter.as_ptr() as u64,
+            Size: filter.len() as u32,
+            Type: EVENT_FILTER_TYPE_EVENT_ID,
+        };
+        let params = ENABLE_TRACE_PARAMETERS {
+            Version: ENABLE_TRACE_PARAMETERS_VERSION_2,
+            EnableProperty: 0,
+            ControlFlags: 0,
+            SourceId: GUID::zeroed(),
+            EnableFilterDesc: &mut desc,
+            FilterDescCount: 1,
+        };
+        // SAFETY: `session` is the live handle just started; `params`/`desc`/`filter` are live
+        // locals for this synchronous call (the kernel copies the filter).
+        let rc = unsafe {
+            EnableTraceEx2(
+                session,
+                &DXGKRNL,
+                EVENT_CONTROL_CODE_ENABLE_PROVIDER.0,
+                TRACE_LEVEL_INFORMATION as u8,
+                0,
+                0,
+                0,
+                Some(&params),
+            )
+        };
+        if rc != ERROR_SUCCESS {
+            tracing::debug!(
+                rc = rc.0,
+                "DxgKrnl ETW enable failed — stopping the session"
+            );
+            let (mut buf, _) = properties_buffer();
+            // SAFETY: live handle + valid properties allocation, stopped exactly once on this path.
+            unsafe {
+                let props = buf.as_mut_ptr().cast::<EVENT_TRACE_PROPERTIES>();
+                (*props).Wnode.BufferSize = buf.len() as u32;
+                let _ = ControlTraceW(session, PWSTR::null(), props, EVENT_TRACE_CONTROL_STOP);
+            }
+            return None;
+        }
+
+        let mut log = EVENT_TRACE_LOGFILEW {
+            LoggerName: PWSTR(name.as_ptr() as *mut _),
+            ..Default::default()
+        };
+        log.Anonymous1.ProcessTraceMode =
+            PROCESS_TRACE_MODE_REAL_TIME | PROCESS_TRACE_MODE_EVENT_RECORD;
+        log.Anonymous2.EventRecordCallback = Some(on_event);
+        // SAFETY: `log` is a fully-initialized local; `name` outlives the call (OpenTrace copies
+        // what it needs before returning).
+        let consumer = unsafe { OpenTraceW(&mut log) };
+        if consumer.Value == u64::MAX {
+            tracing::debug!("DxgKrnl ETW OpenTrace failed — stopping the session");
+            let (mut buf, _) = properties_buffer();
+            // SAFETY: live handle + valid properties allocation, stopped exactly once on this path.
+            unsafe {
+                let props = buf.as_mut_ptr().cast::<EVENT_TRACE_PROPERTIES>();
+                (*props).Wnode.BufferSize = buf.len() as u32;
+                let _ = ControlTraceW(session, PWSTR::null(), props, EVENT_TRACE_CONTROL_STOP);
+            }
+            return None;
+        }
+        // The consumer: ProcessTrace blocks for the session's lifetime; Drop's STOP unblocks it.
+        let consumer_value = consumer.Value;
+        if let Err(e) = std::thread::Builder::new()
+            .name("pf-etw-dxgkrnl".into())
+            .spawn(move || {
+                // SAFETY: `consumer_value` is the live consumer handle opened above; ProcessTrace
+                // pumps it until the controller stops the session (Drop), then returns.
+                let rc = unsafe {
+                    ProcessTrace(
+                        &[PROCESSTRACE_HANDLE {
+                            Value: consumer_value,
+                        }],
+                        None,
+                        None,
+                    )
+                };
+                tracing::debug!(rc = rc.0, "DxgKrnl ETW consumer exited");
+            })
+        {
+            tracing::debug!(error = %e, "DxgKrnl ETW consumer thread failed to spawn");
+            let (mut buf, _) = properties_buffer();
+            // SAFETY: live handles + valid properties allocation, released exactly once on this path.
+            unsafe {
+                let props = buf.as_mut_ptr().cast::<EVENT_TRACE_PROPERTIES>();
+                (*props).Wnode.BufferSize = buf.len() as u32;
+                let _ = ControlTraceW(session, PWSTR::null(), props, EVENT_TRACE_CONTROL_STOP);
+                let _ = CloseTrace(consumer);
+            }
+            return None;
+        }
+        tracing::debug!("DxgKrnl ETW stall-watch session live (event-id filtered)");
+        Some(Self { session, consumer })
+    }
+
+    /// Summarize the DDI activity inside `[from, to]` — the correlation line a stall report
+    /// carries. Brackets that merely SPAN the window count too (a freeze-long `SetPowerState`
+    /// has both edges outside the hole it caused). `"none"` when the window is clean.
+    pub(super) fn summary(&self, from: Instant, to: Instant) -> String {
+        // Instant → QPC: anchor both clocks now and offset backwards.
+        let (now_i, now_q, freq) = (Instant::now(), qpc_now(), qpc_freq());
+        let to_q = now_q - duration_qpc(now_i.saturating_duration_since(to), freq);
+        let from_q = now_q - duration_qpc(now_i.saturating_duration_since(from), freq);
+        let events: Vec<(i64, u16)> = {
+            let ring = RING.lock().unwrap();
+            ring.iter().filter(|(ts, _)| *ts <= to_q).copied().collect()
+        };
+        let ms = |dq: i64| dq.max(0) * 1_000 / freq;
+        let mut parts = Vec::new();
+        for (start_id, stop_id, label) in [
+            (150u16, 151u16, "QueryChildStatus"),
+            (154, 155, "SetPowerState"),
+            (1096, 1097, "DisplayDetectControl"),
+        ] {
+            let mut open: Option<i64> = None;
+            let mut count = 0u32;
+            let mut max_ms = 0i64;
+            let mut still_open = false;
+            for &(ts, id) in &events {
+                if id == start_id {
+                    open = Some(ts);
+                } else if id == stop_id {
+                    if let Some(s) = open.take() {
+                        // The bracket [s, ts] counts when it intersects the window.
+                        if s <= to_q && ts >= from_q {
+                            count += 1;
+                            max_ms = max_ms.max(ms(ts - s));
+                        }
+                    }
+                }
+            }
+            if let Some(s) = open {
+                if s <= to_q {
+                    count += 1;
+                    max_ms = max_ms.max(ms(to_q - s));
+                    still_open = true;
+                }
+            }
+            if count > 0 {
+                parts.push(format!(
+                    "{label}×{count}(max {max_ms}ms{})",
+                    if still_open { ", open" } else { "" }
+                ));
+            }
+        }
+        for (id, label) in [
+            (272u16, "IndicateChildStatus"),
+            (430, "SetTimingsFromVidPn"),
+        ] {
+            let count = events
+                .iter()
+                .filter(|(ts, i)| *i == id && *ts >= from_q && *ts <= to_q)
+                .count();
+            if count > 0 {
+                parts.push(format!("{label}×{count}"));
+            }
+        }
+        if parts.is_empty() {
+            "none".to_string()
+        } else {
+            parts.join(" ")
+        }
+    }
+}
+
+/// A `Duration` in QPC ticks (saturating; diagnostic precision).
+fn duration_qpc(d: Duration, freq: i64) -> i64 {
+    (d.as_micros() as i64).saturating_mul(freq) / 1_000_000
+}
+
+impl Drop for EtwWatch {
+    fn drop(&mut self) {
+        let (mut buf, _) = properties_buffer();
+        // SAFETY: `self.session`/`self.consumer` are the live handles this watch owns; the STOP
+        // (with a valid properties allocation) ends the session and unblocks ProcessTrace, and
+        // CloseTrace releases the consumer — each exactly once, here.
+        unsafe {
+            let props = buf.as_mut_ptr().cast::<EVENT_TRACE_PROPERTIES>();
+            (*props).Wnode.BufferSize = buf.len() as u32;
+            let _ = ControlTraceW(self.session, PWSTR::null(), props, EVENT_TRACE_CONTROL_STOP);
+            let _ = CloseTrace(self.consumer);
+        }
+    }
+}

crates/pf-capture/src/windows/idd_push/open.rs

@@ -635,6 +635,8 @@ impl IddPushCapturer {
                 stall_watch: StallWatch::new(),
                 offered_at_fresh: 0,
                 max_hb_age_us: 0,
+                probes: super::probes::acquire(),
+                etw: super::dxgkrnl_etw::acquire(),
                 out_ring: Vec::new(),
                 out_idx: 0,
                 video_conv: None,

crates/pf-capture/src/windows/idd_push/probes.rs

@@ -0,0 +1,573 @@
+//! Phase A.2 micro-probes (stall attribution, `vdisplay-disturbance-immunity.md` §4.2): a small
+//! engine of watchdogged sacrificial threads that continuously measure the legs a capture stall
+//! could hide in, so the stall reporter can read back "what was alive during the hole":
+//!
+//! - **fence** (one thread per hardware adapter): a tiny CopyResource + D3D11 fence round-trip —
+//!   engine liveness. A Level-Two/Three adapter freeze ("graphics hardware is idle") stalls this
+//!   for the freeze's duration on EVERY engine of that adapter.
+//! - **dwm-tick**: `DwmGetCompositionTimingInfo(NULL)` `cRefresh` advance — the compositor's own
+//!   clock. Frozen tick with live fences = DWM (or something it waits on) is blocked, not the GPU.
+//! - **dwm-flush**: a watchdogged `DwmFlush` — its latency IS the composition-wait measurement.
+//! - **scanline**: `D3DKMTGetScanLine` on an active output (Level-Zero reentrant — documented safe
+//!   against every miniport lock, so a BLOCKED call here convicts the KMD itself). Prefers a
+//!   physical head; on an exclusive topology only our IDD is active and the call's latency is
+//!   still the KMD-liveness signal ([`pf_win_display::win_display::active_scanline_target`]).
+//! - **cpu**: a high-res sleeper measuring overshoot — the DPC-storm / CPU-starvation
+//!   discriminator (a machine-wide scheduling hole inflates every other probe too).
+//!
+//! The blocking of a probe is its measurement — none of these ever run on the capture/encode
+//! path. Threads are DETACHED (never joined): a probe wedged inside a kernel call for the stall's
+//! duration exits at its next loop turn after the stop flag flips. One engine per process
+//! ([`acquire`]), refcounted across parallel capturers; probes sample at 20 Hz or slower and cost
+//! microseconds each, so the engine is invisible next to a streaming session.
+
+// Every `unsafe` block in this file carries a `// SAFETY:` proof; enforce it (unsafe-proof program).
+#![deny(clippy::undocumented_unsafe_blocks)]
+
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+use std::time::{Duration, Instant};
+
+use windows::core::{s, w, Interface};
+use windows::Win32::Foundation::{CloseHandle, HANDLE, HMODULE, HWND, LUID};
+use windows::Win32::Graphics::Direct3D::D3D_DRIVER_TYPE_UNKNOWN;
+use windows::Win32::Graphics::Direct3D11::{
+    D3D11CreateDevice, ID3D11Device5, ID3D11DeviceContext4, ID3D11Fence, ID3D11Texture2D,
+    D3D11_CREATE_DEVICE_FLAG, D3D11_FENCE_FLAG_NONE, D3D11_SDK_VERSION, D3D11_TEXTURE2D_DESC,
+    D3D11_USAGE_DEFAULT,
+};
+use windows::Win32::Graphics::Dwm::{DwmFlush, DwmGetCompositionTimingInfo, DWM_TIMING_INFO};
+use windows::Win32::Graphics::Dxgi::Common::{DXGI_FORMAT_B8G8R8A8_UNORM, DXGI_SAMPLE_DESC};
+use windows::Win32::Graphics::Dxgi::{
+    CreateDXGIFactory1, IDXGIAdapter1, IDXGIFactory1, DXGI_ADAPTER_FLAG_SOFTWARE,
+};
+use windows::Win32::System::LibraryLoader::{GetModuleHandleW, GetProcAddress};
+use windows::Win32::System::Threading::{CreateEventW, WaitForSingleObject};
+
+use super::stall::ProbeWindow;
+
+/// One probe's sample ring: `(completed_at, span, value_us)` — `value` is the measurement (a call
+/// latency or a frozen-span/overshoot), `span` the wall interval it describes ending at
+/// `completed_at`. Capped; ~20 Hz per probe → several minutes of coverage.
+struct Ring {
+    samples: Mutex<VecDeque<(Instant, Duration, u64)>>,
+}
+
+impl Ring {
+    fn new() -> Self {
+        Self {
+            samples: Mutex::new(VecDeque::with_capacity(512)),
+        }
+    }
+
+    fn push(&self, completed_at: Instant, span: Duration, value_us: u64) {
+        let mut s = self.samples.lock().unwrap();
+        if s.len() == 512 {
+            s.pop_front();
+        }
+        s.push_back((completed_at, span, value_us));
+    }
+
+    /// Max `value` among samples whose described interval `[completed_at - span, completed_at]`
+    /// intersects `[from, to]`; `None` when the ring holds nothing for the window (probe absent,
+    /// or it started after the window).
+    fn window_max(&self, from: Instant, to: Instant) -> Option<u64> {
+        let s = self.samples.lock().unwrap();
+        let mut max: Option<u64> = None;
+        for (end, span, value) in s.iter() {
+            let start = end.checked_sub(*span).unwrap_or(*end);
+            if start <= to && *end >= from {
+                max = Some(max.map_or(*value, |m| m.max(*value)));
+            }
+        }
+        max
+    }
+}
+
+/// A blocking probe's "currently inside the call since" marker: a call still blocked when the
+/// reporter reads the window is exactly the interesting case, and its ring sample does not exist
+/// yet — the marker's age stands in for it.
+struct InFlight {
+    since: Mutex<Option<Instant>>,
+}
+
+impl InFlight {
+    fn new() -> Self {
+        Self {
+            since: Mutex::new(None),
+        }
+    }
+
+    fn enter(&self) -> Instant {
+        let now = Instant::now();
+        *self.since.lock().unwrap() = Some(now);
+        now
+    }
+
+    fn exit(&self) {
+        *self.since.lock().unwrap() = None;
+    }
+
+    /// Age (µs) of a call still in flight that started before `to`; 0 when idle.
+    fn blocked_us(&self, to: Instant) -> u64 {
+        match *self.since.lock().unwrap() {
+            Some(since) if since <= to => to.duration_since(since).as_micros() as u64,
+            _ => 0,
+        }
+    }
+}
+
+/// A blocking probe = ring + in-flight marker; `window_max` folds both.
+struct BlockingProbe {
+    ring: Ring,
+    inflight: InFlight,
+}
+
+impl BlockingProbe {
+    fn new() -> Self {
+        Self {
+            ring: Ring::new(),
+            inflight: InFlight::new(),
+        }
+    }
+
+    fn measure(&self, f: impl FnOnce()) {
+        let t0 = self.inflight.enter();
+        f();
+        let dur = t0.elapsed();
+        self.inflight.exit();
+        self.ring.push(Instant::now(), dur, dur.as_micros() as u64);
+    }
+
+    fn window_max(&self, from: Instant, to: Instant) -> Option<u64> {
+        let ring = self.ring.window_max(from, to);
+        let blocked = self.inflight.blocked_us(to);
+        match (ring, blocked) {
+            (None, 0) => None,
+            (r, b) => Some(r.unwrap_or(0).max(b)),
+        }
+    }
+}
+
+struct Inner {
+    stop: AtomicBool,
+    /// One per hardware adapter, labelled by LUID (hybrids run two).
+    fences: Vec<(String, BlockingProbe)>,
+    dwm_tick: Ring,
+    dwm_flush: BlockingProbe,
+    scanline: BlockingProbe,
+    /// Whether the scanline probe currently targets a PHYSICAL head (see the module docs).
+    scanline_physical: AtomicBool,
+    scanline_running: AtomicBool,
+    cpu: Ring,
+}
+
+/// The process-wide micro-probe engine. Hold the `Arc` for the session's lifetime; the last drop
+/// stops every thread at its next loop turn.
+pub(super) struct ProbeEngine {
+    inner: Arc<Inner>,
+}
+
+impl Drop for ProbeEngine {
+    fn drop(&mut self) {
+        self.inner.stop.store(true, Ordering::Relaxed);
+    }
+}
+
+static ENGINE: Mutex<Weak<ProbeEngine>> = Mutex::new(Weak::new());
+
+/// The engine, started on first use and shared across parallel capturers (probe threads are
+/// per-PROCESS facts — two capturers asking the same questions twice would only add noise).
+pub(super) fn acquire() -> Arc<ProbeEngine> {
+    let mut g = ENGINE.lock().unwrap();
+    if let Some(e) = g.upgrade() {
+        return e;
+    }
+    let e = Arc::new(ProbeEngine::start());
+    *g = Arc::downgrade(&e);
+    e
+}
+
+impl ProbeEngine {
+    /// What the probes saw across `[from, to]` — the stall reporter's evidence read. Cheap: five
+    /// short mutex-held ring scans.
+    pub(super) fn window(&self, from: Instant, to: Instant) -> ProbeWindow {
+        let i = &self.inner;
+        ProbeWindow {
+            fence_max_us: i
+                .fences
+                .iter()
+                .filter_map(|(_, p)| p.window_max(from, to))
+                .max(),
+            dwm_tick_frozen_us: i.dwm_tick.window_max(from, to),
+            dwm_flush_max_us: i.dwm_flush.window_max(from, to),
+            scanline_max_us: if i.scanline_running.load(Ordering::Relaxed) {
+                i.scanline.window_max(from, to)
+            } else {
+                None
+            },
+            scanline_physical: i.scanline_physical.load(Ordering::Relaxed),
+            cpu_max_overshoot_us: i.cpu.window_max(from, to),
+        }
+    }
+
+    fn start() -> Self {
+        let fences = enumerate_hardware_adapters()
+            .into_iter()
+            .map(|(label, _)| (label, BlockingProbe::new()))
+            .collect();
+        let inner = Arc::new(Inner {
+            stop: AtomicBool::new(false),
+            fences,
+            dwm_tick: Ring::new(),
+            dwm_flush: BlockingProbe::new(),
+            scanline: BlockingProbe::new(),
+            scanline_physical: AtomicBool::new(false),
+            scanline_running: AtomicBool::new(false),
+            cpu: Ring::new(),
+        });
+        // Fence probes re-enumerate to pair each adapter with its probe slot by index (the two
+        // enumerations run back-to-back; a hot-plugged GPU between them only skips a probe).
+        for (idx, (label, adapter)) in enumerate_hardware_adapters().into_iter().enumerate() {
+            let inner_t = inner.clone();
+            spawn_detached(&format!("pf-probe-fence-{idx}"), move || {
+                fence_probe_loop(&inner_t, idx, &label, &adapter);
+            });
+        }
+        let inner_t = inner.clone();
+        spawn_detached("pf-probe-dwm-tick", move || dwm_tick_loop(&inner_t));
+        let inner_t = inner.clone();
+        spawn_detached("pf-probe-dwm-flush", move || dwm_flush_loop(&inner_t));
+        let inner_t = inner.clone();
+        spawn_detached("pf-probe-scanline", move || scanline_loop(&inner_t));
+        let inner_t = inner.clone();
+        spawn_detached("pf-probe-cpu", move || cpu_sentinel_loop(&inner_t));
+        tracing::debug!(
+            fence_probes = inner.fences.len(),
+            "stall-attribution micro-probes started (fence/dwm-tick/dwm-flush/scanline/cpu)"
+        );
+        Self { inner }
+    }
+}
+
+/// Spawn a detached probe thread — never joined by design (see the module docs).
+fn spawn_detached(name: &str, f: impl FnOnce() + Send + 'static) {
+    if let Err(e) = std::thread::Builder::new().name(name.into()).spawn(f) {
+        tracing::debug!(name, error = %e, "micro-probe thread failed to spawn — probe absent");
+    }
+}
+
+/// Hardware (non-software) DXGI adapters, labelled by LUID. Display-only/indirect adapters are
+/// filtered later by their device-creation failure, not here.
+fn enumerate_hardware_adapters() -> Vec<(String, IDXGIAdapter1)> {
+    let mut out = Vec::new();
+    // SAFETY: plain factory creation; the result is checked.
+    let Ok(factory) = (unsafe { CreateDXGIFactory1::<IDXGIFactory1>() }) else {
+        return out;
+    };
+    for i in 0.. {
+        // SAFETY: `factory` is live; enumeration ends at DXGI_ERROR_NOT_FOUND (the Err arm).
+        let Ok(adapter) = (unsafe { factory.EnumAdapters1(i) }) else {
+            break;
+        };
+        // SAFETY: `adapter` is live; `desc` is a valid out-param.
+        let Ok(desc) = (unsafe { adapter.GetDesc1() }) else {
+            continue;
+        };
+        if desc.Flags & DXGI_ADAPTER_FLAG_SOFTWARE.0 as u32 != 0 {
+            continue;
+        }
+        out.push((
+            format!(
+                "{:08x}:{:08x}",
+                desc.AdapterLuid.HighPart, desc.AdapterLuid.LowPart
+            ),
+            adapter,
+        ));
+    }
+    out
+}
+
+/// One adapter's engine-liveness loop: submit a trivial copy, signal a fence, wait on its event.
+/// The wait latency is the sample. Device-creation failure = probe absent for this adapter
+/// (display-only/indirect adapters land here).
+fn fence_probe_loop(inner: &Inner, idx: usize, label: &str, adapter: &IDXGIAdapter1) {
+    let Some((context, ctx4, fence, event, (a, b))) = fence_probe_setup(adapter) else {
+        tracing::debug!(
+            adapter = label,
+            "fence probe: no device on this adapter — absent"
+        );
+        return;
+    };
+    let mut value = 0u64;
+    let Some((_, probe)) = inner.fences.get(idx) else {
+        return;
+    };
+    while !inner.stop.load(Ordering::Relaxed) {
+        value += 1;
+        probe.measure(|| {
+            // SAFETY: all COM objects are live for this loop's lifetime (owned above); `a`/`b`
+            // are same-device, same-desc textures; the event handle is ours. The fence signal
+            // orders after the queued copy, so the wait measures the engine actually processing
+            // work; the 10 s ceiling only bounds a truly wedged adapter (the timeout sample still
+            // records — that IS the measurement).
+            unsafe {
+                context.CopyResource(&a, &b);
+                let _ = ctx4.Signal(&fence, value);
+                if fence.SetEventOnCompletion(value, event).is_ok() {
+                    let _ = WaitForSingleObject(event, 10_000);
+                }
+            }
+        });
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    // SAFETY: the loop exited; this thread owns `event` and closes it exactly once.
+    unsafe {
+        let _ = CloseHandle(event);
+    }
+}
+
+/// The fence probe's D3D plumbing: a device on `adapter`, its fence, the wait event, and two tiny
+/// textures kept alive for the copies. `None` when any piece is unavailable (pre-FL11 or
+/// display-only adapters, fence-less runtimes).
+#[allow(clippy::type_complexity)]
+fn fence_probe_setup(
+    adapter: &IDXGIAdapter1,
+) -> Option<(
+    windows::Win32::Graphics::Direct3D11::ID3D11DeviceContext,
+    ID3D11DeviceContext4,
+    ID3D11Fence,
+    HANDLE,
+    (ID3D11Texture2D, ID3D11Texture2D),
+)> {
+    let mut device = None;
+    let mut context = None;
+    // SAFETY: adapter is live; UNKNOWN driver type is the documented mode for an explicit
+    // adapter; out-params are valid locals checked below.
+    unsafe {
+        D3D11CreateDevice(
+            adapter,
+            D3D_DRIVER_TYPE_UNKNOWN,
+            HMODULE::default(),
+            D3D11_CREATE_DEVICE_FLAG(0),
+            None,
+            D3D11_SDK_VERSION,
+            Some(&mut device),
+            None,
+            Some(&mut context),
+        )
+        .ok()?;
+    }
+    let device = device?;
+    let context = context?;
+    let device5: ID3D11Device5 = device.cast().ok()?;
+    let ctx4: ID3D11DeviceContext4 = context.cast().ok()?;
+    let desc = D3D11_TEXTURE2D_DESC {
+        Width: 16,
+        Height: 16,
+        MipLevels: 1,
+        ArraySize: 1,
+        Format: DXGI_FORMAT_B8G8R8A8_UNORM,
+        SampleDesc: DXGI_SAMPLE_DESC {
+            Count: 1,
+            Quality: 0,
+        },
+        Usage: D3D11_USAGE_DEFAULT,
+        ..Default::default()
+    };
+    let mut a = None;
+    let mut b = None;
+    // SAFETY: `device` is live, `desc` fully initialized, out-params valid locals checked below.
+    unsafe {
+        device.CreateTexture2D(&desc, None, Some(&mut a)).ok()?;
+        device.CreateTexture2D(&desc, None, Some(&mut b)).ok()?;
+    }
+    let (a, b) = (a?, b?);
+    let mut fence = None;
+    // SAFETY: `device5` is live; out-param is a valid local checked below.
+    unsafe {
+        device5
+            .CreateFence(0, D3D11_FENCE_FLAG_NONE, &mut fence)
+            .ok()?;
+    }
+    let fence: ID3D11Fence = fence?;
+    // SAFETY: plain event creation (auto-reset, unnamed); checked below, closed by the loop.
+    let event = unsafe { CreateEventW(None, false, false, None) }.ok()?;
+    Some((context, ctx4, fence, event, (a, b)))
+}
+
+/// `cRefresh` advance sampling: each tick records how long the compositor's frame counter has been
+/// unchanged. A frozen span covering a stall (with live fences) = DWM blocked, not the GPU.
+fn dwm_tick_loop(inner: &Inner) {
+    let mut last_refresh = 0u64;
+    let mut last_change = Instant::now();
+    while !inner.stop.load(Ordering::Relaxed) {
+        let mut info = DWM_TIMING_INFO {
+            cbSize: std::mem::size_of::<DWM_TIMING_INFO>() as u32,
+            ..Default::default()
+        };
+        // SAFETY: a NULL hwnd asks for composition-wide timing; `info` is a valid local with
+        // `cbSize` stamped (the API's byte-count contract).
+        if unsafe { DwmGetCompositionTimingInfo(HWND::default(), &mut info) }.is_ok() {
+            let now = Instant::now();
+            if info.cRefresh != last_refresh {
+                last_refresh = info.cRefresh;
+                last_change = now;
+            }
+            let frozen = now.duration_since(last_change);
+            inner.dwm_tick.push(now, frozen, frozen.as_micros() as u64);
+        }
+        std::thread::sleep(Duration::from_millis(50));
+    }
+}
+
+/// Watchdogged `DwmFlush`: blocks until the next composition — the wait IS the measurement.
+fn dwm_flush_loop(inner: &Inner) {
+    while !inner.stop.load(Ordering::Relaxed) {
+        inner.dwm_flush.measure(|| {
+            // SAFETY: no arguments, no state — the call blocks until DWM composes (or fails
+            // immediately when composition is unavailable; both durations are valid samples).
+            let _ = unsafe { DwmFlush() };
+        });
+        std::thread::sleep(Duration::from_millis(100));
+    }
+}
+
+/// `D3DKMT_OPENADAPTERFROMLUID` (d3dkmthk.h): LUID in, kernel adapter handle out.
+#[repr(C)]
+struct D3dkmtOpenAdapterFromLuid {
+    adapter_luid: LUID,
+    h_adapter: u32,
+}
+
+/// `D3DKMT_GETSCANLINE` (d3dkmthk.h): `InVerticalBlank` is a C `BOOLEAN` (u8) — the pad bytes
+/// before the 4-aligned `scan_line` mirror the C layout exactly (size assert below).
+#[repr(C)]
+struct D3dkmtGetScanline {
+    h_adapter: u32,
+    vidpn_source_id: u32,
+    in_vertical_blank: u8,
+    _pad: [u8; 3],
+    scan_line: u32,
+}
+
+/// `D3DKMT_CLOSEADAPTER` (d3dkmthk.h).
+#[repr(C)]
+struct D3dkmtCloseAdapter {
+    h_adapter: u32,
+}
+
+const _: () = {
+    assert!(std::mem::size_of::<D3dkmtOpenAdapterFromLuid>() == 12);
+    assert!(std::mem::size_of::<D3dkmtGetScanline>() == 16);
+    assert!(std::mem::size_of::<D3dkmtCloseAdapter>() == 4);
+};
+
+type D3dkmtFn<T> = unsafe extern "system" fn(*mut T) -> i32;
+
+/// Resolve a `D3DKMT*` entry point from gdi32 (no stable windows-rs bindings — the same
+/// GetProcAddress route as pf-frame's scheduling-priority call).
+///
+/// # Safety
+/// `T` must be the exact d3dkmthk.h argument struct for `name` — the transmute binds the
+/// signature, and the layout asserts above pin the three structs this module uses.
+unsafe fn d3dkmt<T>(name: windows::core::PCSTR) -> Option<D3dkmtFn<T>> {
+    // SAFETY: gdi32 is a permanent system module in any process that touched GDI/D3D; the name is
+    // a static nul-terminated literal from the caller.
+    let gdi32 = unsafe { GetModuleHandleW(w!("gdi32.dll")) }.ok()?;
+    // SAFETY: `gdi32` is the live module handle just obtained.
+    let p = unsafe { GetProcAddress(gdi32, name) }?;
+    // SAFETY: the caller's contract (doc above) — `p` is the named export, whose ABI is
+    // `NTSTATUS fn(struct*)` for every D3DKMT entry point this module names.
+    Some(unsafe { std::mem::transmute::<unsafe extern "system" fn() -> isize, D3dkmtFn<T>>(p) })
+}
+
+/// Level-Zero KMD-liveness loop: `D3DKMTGetScanLine` against an active output, retargeted every
+/// ~2 s (topology moves mid-session). The call's LATENCY is the signal; a blocked Level-Zero DDI
+/// convicts the KMD below every OS lever.
+fn scanline_loop(inner: &Inner) {
+    // SAFETY: `D3dkmtOpenAdapterFromLuid` is the exact d3dkmthk.h struct for this export
+    // (layout-asserted above) — the `d3dkmt` safety contract.
+    let open = unsafe { d3dkmt::<D3dkmtOpenAdapterFromLuid>(s!("D3DKMTOpenAdapterFromLuid")) };
+    // SAFETY: `D3dkmtGetScanline` is the exact d3dkmthk.h struct for this export (asserted above).
+    let get = unsafe { d3dkmt::<D3dkmtGetScanline>(s!("D3DKMTGetScanLine")) };
+    // SAFETY: `D3dkmtCloseAdapter` is the exact d3dkmthk.h struct for this export (asserted above).
+    let close = unsafe { d3dkmt::<D3dkmtCloseAdapter>(s!("D3DKMTCloseAdapter")) };
+    let (Some(open), Some(get), Some(close)) = (open, get, close) else {
+        tracing::debug!("scanline probe: D3DKMT entry points unavailable — absent");
+        return;
+    };
+    let mut target: Option<(u32, u32)> = None; // (h_adapter, vidpn_source_id)
+    let mut ticks = 0u32;
+    while !inner.stop.load(Ordering::Relaxed) {
+        if target.is_none() || ticks % 20 == 0 {
+            if let Some((h, _)) = target.take() {
+                let mut req = D3dkmtCloseAdapter { h_adapter: h };
+                // SAFETY: `req` is a valid local for the asserted layout; `h` came from a
+                // successful open below and is closed exactly once here.
+                unsafe { close(&mut req) };
+            }
+            if let Some((lo, hi, source, physical)) =
+                pf_win_display::win_display::active_scanline_target()
+            {
+                let mut req = D3dkmtOpenAdapterFromLuid {
+                    adapter_luid: LUID {
+                        LowPart: lo,
+                        HighPart: hi,
+                    },
+                    h_adapter: 0,
+                };
+                // SAFETY: `req` is a valid local for the asserted layout; the returned handle is
+                // owned by this loop and closed on retarget/exit.
+                if unsafe { open(&mut req) } >= 0 && req.h_adapter != 0 {
+                    target = Some((req.h_adapter, source));
+                    inner.scanline_physical.store(physical, Ordering::Relaxed);
+                }
+            }
+            inner
+                .scanline_running
+                .store(target.is_some(), Ordering::Relaxed);
+        }
+        ticks = ticks.wrapping_add(1);
+        if let Some((h, source)) = target {
+            inner.scanline.measure(|| {
+                let mut req = D3dkmtGetScanline {
+                    h_adapter: h,
+                    vidpn_source_id: source,
+                    in_vertical_blank: 0,
+                    _pad: [0; 3],
+                    scan_line: 0,
+                };
+                // SAFETY: `req` is a valid local for the asserted layout; `h` is the live adapter
+                // handle owned by this loop. A failed status is still a timed (valid) sample.
+                unsafe { get(&mut req) };
+            });
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+    if let Some((h, _)) = target.take() {
+        let mut req = D3dkmtCloseAdapter { h_adapter: h };
+        // SAFETY: as the retarget close above — owned handle, closed exactly once at exit.
+        unsafe { close(&mut req) };
+    }
+}
+
+/// DPC/starvation sentinel: 5 ms sleeps, overshoot aggregated to one max per ~100 ms bucket.
+fn cpu_sentinel_loop(inner: &Inner) {
+    let mut bucket_max = 0u64;
+    let mut bucket_start = Instant::now();
+    while !inner.stop.load(Ordering::Relaxed) {
+        let t0 = Instant::now();
+        std::thread::sleep(Duration::from_millis(5));
+        let overshoot = t0.elapsed().saturating_sub(Duration::from_millis(5));
+        bucket_max = bucket_max.max(overshoot.as_micros() as u64);
+        let now = Instant::now();
+        let span = now.duration_since(bucket_start);
+        if span >= Duration::from_millis(100) {
+            inner.cpu.push(now, span, bucket_max);
+            bucket_max = 0;
+            bucket_start = now;
+        }
+    }
+}

crates/pf-capture/src/windows/idd_push/stall.rs

@@ -26,6 +26,131 @@ pub(super) struct StallEvidence {
     /// The STALEST the driver's drain heartbeat ever read while the host starved (max of
     /// now − heartbeat over the window), in milliseconds.
     pub(super) max_heartbeat_age_ms: u64,
+    /// What the micro-probe engine saw across the window (Phase A.2); `None` when the engine
+    /// isn't running.
+    pub(super) probes: Option<ProbeWindow>,
+    /// The DxgKrnl DDI activity inside the window (Phase A.3 ETW summary); `None` when the
+    /// session is unavailable (non-admin dev run).
+    pub(super) etw: Option<String>,
+}
+
+/// The micro-probes' window read (Phase A.2, built by `probes::ProbeEngine::window`): per-leg
+/// maxima across one stall window. Every field is `None` when that probe is absent (no adapter
+/// device, no active output, thread failed to spawn) — absence is stated, never guessed.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(super) struct ProbeWindow {
+    /// Worst engine-liveness fence round-trip (µs) across all hardware adapters.
+    pub(super) fence_max_us: Option<u64>,
+    /// Longest span (µs) with no `DwmGetCompositionTimingInfo` `cRefresh` advance.
+    pub(super) dwm_tick_frozen_us: Option<u64>,
+    /// Worst watchdogged `DwmFlush` latency (µs).
+    pub(super) dwm_flush_max_us: Option<u64>,
+    /// Worst `D3DKMTGetScanLine` CALL latency (µs) — Level-Zero, so blocking convicts the KMD.
+    pub(super) scanline_max_us: Option<u64>,
+    /// Whether the scanline probe had a PHYSICAL head to ask (exclusive topology leaves only our
+    /// IDD active — latency still counts, scanline values don't).
+    pub(super) scanline_physical: bool,
+    /// Worst high-res sleeper overshoot (µs) — the DPC-storm / CPU-starvation discriminator.
+    pub(super) cpu_max_overshoot_us: Option<u64>,
+}
+
+impl std::fmt::Display for ProbeWindow {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let ms = |v: Option<u64>| match v {
+            Some(us) => format!("{:.0}ms", us as f64 / 1_000.0),
+            None => "absent".to_string(),
+        };
+        write!(
+            f,
+            "fence={} dwm_tick_frozen={} dwm_flush={} scanline={}({}) cpu_overshoot={}",
+            ms(self.fence_max_us),
+            ms(self.dwm_tick_frozen_us),
+            ms(self.dwm_flush_max_us),
+            ms(self.scanline_max_us),
+            if self.scanline_physical {
+                "physical"
+            } else {
+                "virtual"
+            },
+            ms(self.cpu_max_overshoot_us),
+        )
+    }
+}
+
+/// The named disturbance class a stall's combined evidence supports — the [`attribute`] verdict
+/// (driver telemetry, Phase A.1) refined by the micro-probe window (Phase A.2). This is the
+/// per-stall output of the program's verdict matrix (design doc §4.4).
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub(super) enum StallClass {
+    /// The drain worker starved — ours (CPU/MMCSS/dead WUDFHost).
+    OursWorker,
+    /// Frames were composed and offered but never became consumable — ours (ring/publish/consume).
+    OursDelivery,
+    /// Engine-liveness fences stalled with the hole: the ADAPTER froze (Level-Two/Three DDI
+    /// servicing — link train, power transition, mux). Class 1.
+    AdapterFreeze,
+    /// Engines alive but DWM's own tick froze: the compositor is blocked on something (DDC/child
+    /// I/O vendor lock, win32k display-config queue). Class 2.
+    CompositorBlocked,
+    /// Engines alive, DWM ticking, driver drained E_PENDING — composition happened for OTHER
+    /// surfaces but produced no frame for OUR display: the frame-generation path
+    /// (IddCx/dirty-tracking/divider). Ours to chase with IddCx WPP.
+    FrameGeneration,
+    /// Not enough evidence to name a class (pre-telemetry driver and/or probes absent).
+    Unattributed,
+}
+
+impl std::fmt::Display for StallClass {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::OursWorker => "OURS-worker (drain thread starved)",
+            Self::OursDelivery => "OURS-delivery (ring/publish/consume lost composed frames)",
+            Self::AdapterFreeze => {
+                "CLASS-1 adapter freeze (engines stalled below the OS — link/power/mux servicing)"
+            }
+            Self::CompositorBlocked => {
+                "CLASS-2 compositor blocked (engines alive, DWM tick frozen — vendor lock / DDC)"
+            }
+            Self::FrameGeneration => {
+                "FRAME-GENERATION (DWM ticked, engines alive, no frame for THIS display — IddCx/dirty/divider)"
+            }
+            Self::Unattributed => "UNATTRIBUTED (insufficient telemetry)",
+        })
+    }
+}
+
+/// The verdict matrix: fold the driver-telemetry verdict and the probe window into a class.
+/// Pure — unit-tested beside the [`StallWatch`] tests. A leg is "stalled for the hole" when its
+/// worst reading covers at least half the gap (the same proportional bar as [`attribute`]).
+pub(super) fn classify(
+    gap: Duration,
+    verdict: &StallVerdict,
+    probes: Option<&ProbeWindow>,
+) -> StallClass {
+    match verdict {
+        StallVerdict::WorkerStalled => return StallClass::OursWorker,
+        StallVerdict::DeliveryLeg => return StallClass::OursDelivery,
+        StallVerdict::ComposeSilence | StallVerdict::NoTelemetry => {}
+    }
+    let Some(p) = probes else {
+        return StallClass::Unattributed;
+    };
+    let half_gap_us = (gap.as_micros() as u64) / 2;
+    let covers = |v: Option<u64>| v.is_some_and(|us| us >= half_gap_us);
+    if covers(p.fence_max_us) {
+        return StallClass::AdapterFreeze;
+    }
+    if covers(p.dwm_tick_frozen_us) || covers(p.dwm_flush_max_us) {
+        return StallClass::CompositorBlocked;
+    }
+    // Engines alive and DWM ticking: only the driver's own E_PENDING testimony can pin the
+    // frame-generation path — without it (pre-telemetry driver) the delivery leg is equally
+    // possible, so stay honest.
+    if matches!(verdict, StallVerdict::ComposeSilence) {
+        StallClass::FrameGeneration
+    } else {
+        StallClass::Unattributed
+    }
 }
 
 /// The attribution a stall's evidence supports — the Branch-1/Branch-2 fork of the
@@ -102,6 +227,9 @@ pub(super) struct StallWatch {
     /// in [`StallVerdict`] order — the metronomic WARN prints it, so one pasted line attributes the
     /// whole session's beat, not just the stall that tripped the metronome.
     verdicts: [u32; 4],
+    /// Running per-class tally ([`StallClass`] order: ours-worker, ours-delivery, adapter-freeze,
+    /// compositor-blocked, frame-generation, unattributed) — the verdict matrix's session summary.
+    classes: [u32; 6],
 }
 
 impl StallWatch {
@@ -122,6 +250,7 @@ impl StallWatch {
             seen: 0,
             with_os_events: 0,
             verdicts: [0; 4],
+            classes: [0; 6],
         }
     }
 
@@ -183,6 +312,15 @@ impl StallWatch {
             StallVerdict::ComposeSilence => 2,
             StallVerdict::DeliveryLeg => 3,
         }] += 1;
+        let class = classify(stall.gap, &verdict, evidence.probes.as_ref());
+        self.classes[match class {
+            StallClass::OursWorker => 0,
+            StallClass::OursDelivery => 1,
+            StallClass::AdapterFreeze => 2,
+            StallClass::CompositorBlocked => 3,
+            StallClass::FrameGeneration => 4,
+            StallClass::Unattributed => 5,
+        }] += 1;
         // debug (not warn): a single hole also happens when content legitimately pauses;
         // the reportable signal is the metronomic cycle below. Mounjay-class triage runs
         // at debug level, and the web-console debug ring captures these.
@@ -190,10 +328,13 @@ impl StallWatch {
             gap_ms = stall.gap.as_millis() as u64,
             os_display_events = %pf_win_display::display_events::summarize(&events),
             verdict = %verdict,
+            class = %class,
+            probes = evidence.probes.as_ref().map(tracing::field::display),
+            etw = evidence.etw.as_deref().unwrap_or("unavailable"),
             offered_during_gap = evidence.offered_delta,
             max_heartbeat_age_ms = evidence.max_heartbeat_age_ms,
             "IDD-push capture stall — the desktop was composing at speed, then the ring \
-             delivered no frame for the gap; the verdict names the leg that lost them"
+             delivered no frame for the gap; the class names the leg that lost them"
         );
         if let Some(period) = stall.metronomic {
             let suspects = pf_win_display::display_events::connected_inactive_physicals();
@@ -208,6 +349,16 @@ impl StallWatch {
                 "worker-stalled {}, compose-silence {}, delivery-leg {}, no-telemetry {}",
                 self.verdicts[1], self.verdicts[2], self.verdicts[3], self.verdicts[0]
             );
+            let class_tally = format!(
+                "ours-worker {}, ours-delivery {}, adapter-freeze {}, compositor-blocked {}, \
+                 frame-generation {}, unattributed {}",
+                self.classes[0],
+                self.classes[1],
+                self.classes[2],
+                self.classes[3],
+                self.classes[4],
+                self.classes[5]
+            );
             // Half-or-more of the stalls carrying a coinciding OS event = the reaction
             // cascade is OS-visible; otherwise the disturbance never surfaces above the
             // driver. Different classes, different cures — say which one this box has.
@@ -217,6 +368,7 @@ impl StallWatch {
                     os_correlated = correlated,
                     connected_inactive = %suspects,
                     verdicts = %verdict_tally,
+                    classes = %class_tally,
                     "capture stalls are METRONOMIC and coincide with Windows monitor \
                      hot-plug/re-enumeration events — a connected display (or its \
                      cable/switch/AVR) re-probes the link on a timer and Windows re-reacts \
@@ -233,6 +385,7 @@ impl StallWatch {
                     os_correlated = correlated,
                     connected_inactive = %suspects,
                     verdicts = %verdict_tally,
+                    classes = %class_tally,
                     "capture stalls are METRONOMIC with NO coinciding OS display event — \
                      the disturbance is BELOW Windows: the GPU driver servicing a \
                      connected-but-asleep sink (standby HPD/DDC/link probing), \

crates/pf-win-display/src/win_display.rs

@@ -1524,6 +1524,49 @@ pub fn source_desktop_rect(target_id: u32) -> Option<(i32, i32, i32, i32)> {
     None
 }
 
+/// Scanline-probe target (stall-attribution Phase A.2): the adapter LUID + VidPn SOURCE id of an
+/// ACTIVE path, preferring a real display over one of OUR virtual monitors. `D3DKMTGetScanLine`
+/// wants a source that actually scans out, so a physical head (`physical == true`) gives the
+/// honest Level-Zero KMD-liveness probe; on an exclusive topology only our IDD is active, and the
+/// caller still gets it (`physical == false`) — the CALL's latency is the measurement either way,
+/// the flag just keeps the report from over-reading the returned scanline values. Returns
+/// `(adapter_luid_low, adapter_luid_high, vidpn_source_id, physical)`; `None` when nothing is
+/// active or the query fails.
+pub fn active_scanline_target() -> Option<(u32, i32, u32, bool)> {
+    // SAFETY: `query_active_config` is this module's own CCD helper: it takes nothing and returns owned
+    // `Vec`s built from a fresh `QueryDisplayConfig`, so it has no caller obligation at all.
+    let (paths, _modes) = query_active_config()?;
+    let mut fallback: Option<(u32, i32, u32, bool)> = None;
+    for p in &paths {
+        if p.flags & DISPLAYCONFIG_PATH_ACTIVE == 0 {
+            continue;
+        }
+        let candidate = (
+            p.sourceInfo.adapterId.LowPart,
+            p.sourceInfo.adapterId.HighPart,
+            p.sourceInfo.id,
+            false,
+        );
+        let mut req = DISPLAYCONFIG_TARGET_DEVICE_NAME::default();
+        req.header.r#type = DISPLAYCONFIG_DEVICE_INFO_GET_TARGET_NAME;
+        req.header.size = size_of::<DISPLAYCONFIG_TARGET_DEVICE_NAME>() as u32;
+        req.header.adapterId = p.targetInfo.adapterId;
+        req.header.id = p.targetInfo.id;
+        // SAFETY: `req.header` is a live local whose `size` field was just set to the enclosing
+        // struct's own `size_of` (the byte-count contract); the struct outlives this synchronous call.
+        if unsafe { DisplayConfigGetDeviceInfo(&mut req.header) } != 0 {
+            fallback.get_or_insert(candidate);
+            continue;
+        }
+        if is_our_virtual_display(&utf16z_str(&req.monitorDevicePath)) {
+            fallback.get_or_insert(candidate);
+            continue;
+        }
+        return Some((candidate.0, candidate.1, candidate.2, true));
+    }
+    fallback
+}
+
 /// The union of every ACTIVE path's source rect — the virtual-desktop bounds `(x, y, w, h)` in
 /// desktop coordinates. Read from CCD rather than `GetSystemMetrics(SM_*VIRTUALSCREEN)` so the
 /// answer is the CONSOLE's real layout even when the calling process sits in another session (the