diff --git a/crates/chainstate/src/window.rs b/crates/chainstate/src/window.rs
index 58c3792b9..44fb60242 100644
--- a/crates/chainstate/src/window.rs
+++ b/crates/chainstate/src/window.rs
@@ -327,6 +327,9 @@ fn flush_group(
     committed: &mut Vec<ConnectOutcome>,
     payloads: &[&BlockPayload],
 ) -> core::result::Result<(), WindowApplyError> {
+    if group.pending.is_empty() {
+        return Ok(());
+    }
     // The staged prefix is a contiguous suffix starting at `committed.len()`,
     // so the newest staged block is borrowed straight from the caller's apply
     // batch — no clone in chainstate. Empty group: nothing staged, no hook.
diff --git a/crates/mempool/src/submit_package.rs b/crates/mempool/src/submit_package.rs
index bfa21a548..6cd34314d 100644
--- a/crates/mempool/src/submit_package.rs
+++ b/crates/mempool/src/submit_package.rs
@@ -10,6 +10,18 @@ use crate::{
 use alloc::vec::Vec;
 use bitcoin_rs_primitives::{Tx, unix_time_secs};
 
+/// Orders an arbitrary-order child-with-parents package for submission.
+///
+/// The returned vec is topologically ordered (independent parents sorted by
+/// txid, child last) but deliberately NOT fully structure-checked: package
+/// weight, intra-package conflicts, and package order are validated by
+/// [`crate::package::check_structure`], which
+/// [`MempoolGateway::submit_package_with_times`] runs as the single
+/// authoritative structural gate (snapshot restore enters there directly, so
+/// the check cannot live here without charging local submission a duplicate
+/// pass). This function stays private with [`MempoolGateway::submit_local_package`]
+/// as its only caller, so its output always flows through that gate and can
+/// never reach admission unchecked.
 fn ordered(mut txs: Vec<Tx>) -> Result<Vec<Tx>, SubmitError> {
     if txs.is_empty() || txs.len() > crate::standardness::MAX_PACKAGE_COUNT {
         return Err(AcceptanceRejectReason::PackageCount.into());
diff --git a/crates/node/benches/evidence.rs b/crates/node/benches/evidence.rs
index 367733966..690e46e1b 100644
--- a/crates/node/benches/evidence.rs
+++ b/crates/node/benches/evidence.rs
@@ -231,6 +231,64 @@ impl Ledger {
     }
 }
 
+/// Samples a run must hold before its p99 may be read as a tail estimate
+/// (`HPA-13`: no trustworthy p99 from too few samples).
+pub const MIN_SAMPLES_FOR_P99: usize = 100;
+
+/// Order statistics of one raw sample set, in the samples' own unit.
+///
+/// The summary never replaces the raw samples; it is printed beside them.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct Quantiles {
+    /// Sample count the statistics were computed from.
+    pub n: usize,
+    /// Median, linearly interpolated between adjacent order statistics.
+    pub p50: u64,
+    /// 95th percentile, interpolated the same way.
+    pub p95: u64,
+    /// 99th percentile, interpolated the same way.
+    pub p99: u64,
+    /// Largest sample.
+    pub max: u64,
+}
+
+impl Quantiles {
+    /// Computes p50/p95/p99/max of `samples`; `None` for an empty set, which
+    /// is a missing measurement, never a zero.
+    #[must_use]
+    pub fn of(samples: &[u64]) -> Option<Self> {
+        let mut sorted = samples.to_vec();
+        sorted.sort_unstable();
+        let max = *sorted.last()?;
+        let at = |percent: u64| -> u64 {
+            // Rank scaled by 100 keeps the interpolation exact in integers.
+            let last_rank = u64::try_from(sorted.len() - 1).unwrap_or(u64::MAX);
+            let scaled = percent.saturating_mul(last_rank);
+            let lower = usize::try_from(scaled / 100).unwrap_or(usize::MAX);
+            let remainder = scaled % 100;
+            let low = sorted[lower];
+            let high = sorted.get(lower + 1).copied().unwrap_or(low);
+            let span = u128::from(high - low);
+            let step = span * u128::from(remainder);
+            // Round half up on the interpolated step.
+            low + u64::try_from((step + 50) / 100).unwrap_or(u64::MAX)
+        };
+        Some(Self {
+            n: sorted.len(),
+            p50: at(50),
+            p95: at(95),
+            p99: at(99),
+            max,
+        })
+    }
+
+    /// Whether `p99` rests on enough samples to be read as a tail estimate.
+    #[must_use]
+    pub const fn p99_supported(&self) -> bool {
+        self.n >= MIN_SAMPLES_FOR_P99
+    }
+}
+
 #[cfg(test)]
 #[expect(clippy::expect_used, reason = "benchmark evidence rejection tests")]
 mod tests {
@@ -415,4 +473,37 @@ mod tests {
         );
         assert_eq!(reparsed.cells.len(), ledger.cells.len());
     }
+
+    #[test]
+    fn quantiles_are_exact_order_statistics_with_a_support_floor() {
+        assert_eq!(Quantiles::of(&[]), None, "an empty set is not a zero");
+
+        let single = Quantiles::of(&[7]).expect("one sample");
+        assert_eq!(
+            (single.p50, single.p95, single.p99, single.max),
+            (7, 7, 7, 7)
+        );
+        assert!(!single.p99_supported());
+
+        // Ranks land on whole samples, so the statistics are the samples.
+        let ranks: Vec<u64> = (0..=100).rev().collect();
+        let exact = Quantiles::of(&ranks).expect("101 samples");
+        assert_eq!(
+            (exact.n, exact.p50, exact.p95, exact.p99, exact.max),
+            (101, 50, 95, 99, 100)
+        );
+        assert!(exact.p99_supported());
+        assert!(
+            !Quantiles::of(&ranks[..99])
+                .expect("99 samples")
+                .p99_supported()
+        );
+
+        // Between samples the order statistic is interpolated, halves up.
+        let spread = Quantiles::of(&[40, 10, 30, 20]).expect("four samples");
+        assert_eq!(
+            (spread.p50, spread.p95, spread.p99, spread.max),
+            (25, 39, 40, 40)
+        );
+    }
 }
diff --git a/crates/node/benches/sync_pipeline.rs b/crates/node/benches/sync_pipeline.rs
index ab0736638..8241a277c 100644
--- a/crates/node/benches/sync_pipeline.rs
+++ b/crates/node/benches/sync_pipeline.rs
@@ -302,16 +302,8 @@ fn record_evidence(cell: &str, blocks: &[Block], sweeps: &[(Duration, Duration)]
         id: cell.to_owned(),
         manifest_sha256: Sha256Hex::digest(&corpus_bytes),
     });
-    let path =
-        Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/benchmarks/sync_pipeline.toml");
-    let mut ledger = match std::fs::read_to_string(&path) {
-        Ok(text) => Ledger::parse(&text).unwrap_or_else(|error| panic!("bench ledger: {error}")),
-        Err(_) => Ledger {
-            schema: LEDGER_SCHEMA.into(),
-            cells: Vec::new(),
-            contract: std::collections::BTreeMap::new(),
-        },
-    };
+    let path = bench_ledger_path("sync_pipeline.toml");
+    let mut ledger = load_bench_ledger(&path);
     if !ledger.cells.iter().any(|candidate| candidate.id == cell) {
         ledger.cells.push(Cell {
             id: cell.to_owned(),
@@ -339,12 +331,36 @@ fn record_evidence(cell: &str, blocks: &[Block], sweeps: &[(Duration, Duration)]
             .record(cell, sample)
             .unwrap_or_else(|error| panic!("record bench sample: {error}"));
     }
+    store_bench_ledger(&path, &ledger);
+}
+
+/// Path of one bench evidence file under the workspace `target/benchmarks/`.
+fn bench_ledger_path(name: &str) -> std::path::PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("../../target/benchmarks")
+        .join(name)
+}
+
+/// Loads an append-only bench ledger; a missing file is an empty ledger.
+fn load_bench_ledger(path: &Path) -> Ledger {
+    match std::fs::read_to_string(path) {
+        Ok(text) => Ledger::parse(&text).unwrap_or_else(|error| panic!("bench ledger: {error}")),
+        Err(_) => Ledger {
+            schema: LEDGER_SCHEMA.into(),
+            cells: Vec::new(),
+            contract: std::collections::BTreeMap::new(),
+        },
+    }
+}
+
+/// Writes `ledger` back to `path`, creating the evidence directory.
+fn store_bench_ledger(path: &Path, ledger: &Ledger) {
     std::fs::create_dir_all(path.parent().unwrap_or_else(|| Path::new(".")))
         .unwrap_or_else(|error| panic!("bench evidence dir: {error}"));
     let rendered = ledger
         .render()
         .unwrap_or_else(|error| panic!("render bench ledger: {error}"));
-    std::fs::write(&path, rendered).unwrap_or_else(|error| panic!("write bench ledger: {error}"));
+    std::fs::write(path, rendered).unwrap_or_else(|error| panic!("write bench ledger: {error}"));
 }
 
 fn deterministic_initial_sync_proxy(c: &mut Criterion) {
@@ -933,6 +949,23 @@ impl SyncFixture {
     }
 }
 
+/// A boundary of the production replay that a caller may time.
+#[derive(Clone, Copy)]
+enum ReplayPhase {
+    /// The contiguous head is delivered; the next tick applies the window.
+    ApplyStart,
+    /// The applying tick returned.
+    ApplyDone,
+}
+
+/// What one production replay observed.
+struct ReplayReport {
+    /// Items in the first getdata the replay issued.
+    requested: usize,
+    /// Height of the applied tip once the replay finished.
+    applied_height: u32,
+}
+
 struct ProductionStateSyncFixture {
     _dir: TempDir,
     state: NodeState,
@@ -971,6 +1004,21 @@ impl ProductionStateSyncFixture {
         )
     }
 
+    /// Production replay of `body_blocks` coinbase-only blocks under the
+    /// same deep header chain and single synthetic peer as the 128-block arm.
+    /// Window attribution varies only the body count, so the header tree,
+    /// storage backend and request shape stay identical across scenarios.
+    fn window_replay(body_blocks: u32) -> Self {
+        let expected_getdata_count = usize::try_from(body_blocks)
+            .unwrap_or_else(|error| panic!("invalid window body count: {error}"));
+        Self::with_config_and_header_blocks(
+            1,
+            production_state_config(),
+            |tree| populate_sync_header_chain(tree, body_blocks).0,
+            expected_getdata_count,
+        )
+    }
+
     #[cfg(feature = "fjall")]
     fn new_fjall_all_indexes_spend_heavy() -> Self {
         let mut config = production_state_config();
@@ -1034,9 +1082,19 @@ impl ProductionStateSyncFixture {
     }
 
     fn run(mut self) -> u32 {
+        let replay = self.run_phased(|_| {});
+        assert_eq!(replay.requested, self.expected_getdata_count);
+        replay.applied_height
+    }
+
+    /// The production replay of `run`, reporting the boundaries of the tick
+    /// that applies the window so a caller can time that tick on its own.
+    /// The fixture stays alive afterwards: its teardown is not part of any
+    /// interval the caller times.
+    fn run_phased(&mut self, mut phase: impl FnMut(ReplayPhase)) -> ReplayReport {
         let sync = self.state.sync();
         sync.tick();
-        self.assert_getdata_batch();
+        let requested = self.first_getdata_len();
         let inbound_blocks_tx = self.state.inbound_blocks_sender();
         // prebuilt_inbound holds heights 2..=N reversed, then height 1 last.
         // Send the out-of-order tail, tick, then deliver the contiguous head and
@@ -1055,12 +1113,26 @@ impl ProductionStateSyncFixture {
         inbound_blocks_tx
             .send(contiguous)
             .unwrap_or_else(|error| panic!("send production contiguous block failed: {error}"));
+        phase(ReplayPhase::ApplyStart);
         sync.tick();
-        self.state
+        phase(ReplayPhase::ApplyDone);
+        let applied_height = self
+            .state
             .chainstate()
             .applied_tip_snapshot()
             .unwrap_or_else(|| panic!("production sync proxy did not publish applied tip"))
-            .height
+            .height;
+        ReplayReport {
+            requested,
+            applied_height,
+        }
+    }
+
+    /// One more sync tick. After the applying tick it closes the window's
+    /// measurement-only intervals (apply-idle, download-blocked-by-apply),
+    /// which record only when they close.
+    fn settle_tick(&self) {
+        self.state.sync().tick();
     }
 
     fn stage_for_contiguous_apply(self) -> Self {
@@ -1126,12 +1198,18 @@ impl ProductionStateSyncFixture {
     }
 
     fn assert_getdata_batch(&self) {
+        assert_eq!(self.first_getdata_len(), self.expected_getdata_count);
+    }
+
+    /// Drains header requests up to the first getdata and returns its item
+    /// count.
+    fn first_getdata_len(&self) -> usize {
         let rx = self
             .outbound_rxs
             .first()
             .unwrap_or_else(|| panic!("missing primary outbound receiver"));
         let mut drained_headers = 0_usize;
-        let getdata_count = loop {
+        loop {
             match rx
                 .try_recv()
                 .unwrap_or_else(|error| panic!("expected production getdata: {error}"))
@@ -1148,8 +1226,7 @@ impl ProductionStateSyncFixture {
                 }
                 other => panic!("expected production getdata, got {other:?}"),
             }
-        };
-        assert_eq!(getdata_count, self.expected_getdata_count);
+        }
     }
 }
 
@@ -4434,8 +4511,895 @@ fn require_bare_counted_flag(args: &[String], flag: &str) {
     }
 }
 
+// ---------------------------------------------------------------------------
+// Step-24 window attribution: the production replay of one windowed apply,
+// control against candidate, single-group against multi-group.
+// ---------------------------------------------------------------------------
+
+/// Blocks per durable group commit: `DURABLE_HEAD_GROUP_BLOCKS` in
+/// `bitcoin-rs-chainstate`, private there. Drift trips the per-unit group
+/// flush assertion instead of being absorbed.
+const WINDOW_GROUP_BLOCKS: u32 = 64;
+/// Units that bring the allocator, page cache and fjall to steady state.
+/// Retained raw, excluded from every summary and from the ledger.
+const WINDOW_WARMUP_UNITS: usize = 5;
+/// Measured units per run. Equals `evidence::MIN_SAMPLES_FOR_P99`, so every
+/// per-unit series supports its own p99.
+const WINDOW_MEASURED_UNITS: usize = 100;
+/// `/proc/<pid>/stat` reports CPU time in `USER_HZ` ticks, fixed at 100.
+const WINDOW_CPU_TICK_NS: u64 = 10_000_000;
+const WINDOW_ATTRIBUTION_SCHEMA: &str = "bitcoin-rs-window-attribution-v1";
+
+/// Which tree built this binary. The label names the arm; the artifact digest
+/// in the identity is what proves the two arms are different programs.
+#[derive(Clone, Copy)]
+enum WindowArm {
+    Control,
+    Candidate,
+}
+
+impl WindowArm {
+    const fn label(self) -> &'static str {
+        match self {
+            Self::Control => "control",
+            Self::Candidate => "candidate",
+        }
+    }
+
+    fn parse(label: &str) -> Option<Self> {
+        [Self::Control, Self::Candidate]
+            .into_iter()
+            .find(|arm| arm.label() == label)
+    }
+}
+
+/// One window shape: the number of coinbase-only body blocks the replay
+/// delivers as a single contiguous window.
+#[derive(Clone, Copy)]
+struct WindowScenario {
+    id: &'static str,
+    body_blocks: u32,
+}
+
+impl WindowScenario {
+    /// Group commits the window must issue: one per started group.
+    fn groups(self) -> usize {
+        usize::try_from(self.body_blocks.div_ceil(WINDOW_GROUP_BLOCKS))
+            .unwrap_or_else(|error| panic!("invalid window group count: {error}"))
+    }
+}
+
+/// Single-group: exactly one full group, the largest window that flushes
+/// once. Multi-group: two full groups, so a mid-window flush separates
+/// them. The byte bound (`8 MiB`) is not reached by coinbase-only blocks;
+/// only the block-count bound is exercised.
+const WINDOW_SCENARIOS: [WindowScenario; 2] = [
+    WindowScenario {
+        id: "single_group",
+        body_blocks: WINDOW_GROUP_BLOCKS,
+    },
+    WindowScenario {
+        id: "multi_group",
+        body_blocks: 2 * WINDOW_GROUP_BLOCKS,
+    },
+];
+
+/// What a captured interval measures.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum WindowSpanKind {
+    /// `transition_held_us`: the chain transition lock, acquire to release.
+    TransitionHeld,
+    /// `flush_wait_us`: the caller's wait on one durable group commit.
+    FlushWait,
+    /// A group commit that failed; a replay that sees one is invalid.
+    FlushFailed,
+    /// `node.sync.apply_idle_seconds`: apply starved while downloads were owed.
+    ApplyIdle,
+    /// `node.sync.download_blocked_by_apply_seconds`: front in flight while
+    /// apply held the frontier.
+    DownloadBlocked,
+}
+
+impl WindowSpanKind {
+    /// Ledger path the interval measures.
+    const fn ledger_path(self) -> &'static str {
+        match self {
+            Self::TransitionHeld => "apply.window.transition_held",
+            Self::FlushWait => "apply.window.flush_wait",
+            Self::FlushFailed => "apply.window.flush_failed",
+            Self::ApplyIdle => "p2p.window.apply_idle",
+            Self::DownloadBlocked => "p2p.window.download_blocked_by_apply",
+        }
+    }
+}
+
+/// One interval captured from the production diagnostics: it ended when the
+/// event or histogram record arrived and lasted `duration_ns`.
+struct WindowSpan {
+    kind: WindowSpanKind,
+    end: quanta::Instant,
+    duration_ns: u64,
+    /// Blocks the group commit covered; set for `FlushWait` only.
+    blocks: Option<u64>,
+}
+
+/// Spans of the unit in flight. The replay runs on the bench thread, and the
+/// harness drains this per unit, so no span crosses a unit boundary.
+static WINDOW_SPANS: ParkingMutex<Vec<WindowSpan>> = ParkingMutex::new(Vec::new());
+
+/// Bench-side tracing subscriber. It enables only the two production events
+/// that carry `transition_held_us` and `flush_wait_us`; every other callsite
+/// is disabled at registration, so the replay pays no tracing cost beyond
+/// these events.
+struct WindowSpanSubscriber;
+
+impl WindowSpanSubscriber {
+    fn watches(metadata: &tracing::Metadata<'_>) -> bool {
+        metadata.is_event()
+            && (metadata.fields().field("transition_held_us").is_some()
+                || metadata.fields().field("flush_wait_us").is_some())
+    }
+}
+
+#[derive(Default)]
+struct WindowSpanVisitor {
+    transition_held_us: Option<u128>,
+    flush_wait_us: Option<u128>,
+    blocks: Option<u64>,
+    failed: bool,
+}
+
+impl tracing::field::Visit for WindowSpanVisitor {
+    fn record_debug(&mut self, field: &tracing::field::Field, _value: &dyn std::fmt::Debug) {
+        if field.name() == "flush_error" {
+            self.failed = true;
+        }
+    }
+
+    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
+        match field.name() {
+            "blocks" => self.blocks = Some(value),
+            "transition_held_us" => self.transition_held_us = Some(u128::from(value)),
+            "flush_wait_us" => self.flush_wait_us = Some(u128::from(value)),
+            _ => {}
+        }
+    }
+
+    fn record_u128(&mut self, field: &tracing::field::Field, value: u128) {
+        match field.name() {
+            "transition_held_us" => self.transition_held_us = Some(value),
+            "flush_wait_us" => self.flush_wait_us = Some(value),
+            _ => {}
+        }
+    }
+}
+
+impl tracing::Subscriber for WindowSpanSubscriber {
+    fn register_callsite(
+        &self,
+        metadata: &'static tracing::Metadata<'static>,
+    ) -> tracing::subscriber::Interest {
+        if Self::watches(metadata) {
+            tracing::subscriber::Interest::always()
+        } else {
+            tracing::subscriber::Interest::never()
+        }
+    }
+
+    fn enabled(&self, metadata: &tracing::Metadata<'_>) -> bool {
+        Self::watches(metadata)
+    }
+
+    fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
+        tracing::span::Id::from_u64(1)
+    }
+
+    fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
+
+    fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
+
+    fn event(&self, event: &tracing::Event<'_>) {
+        let end = quanta::Instant::now();
+        let mut visitor = WindowSpanVisitor::default();
+        event.record(&mut visitor);
+        let micros_to_nanos =
+            |micros: u128| u64::try_from(micros.saturating_mul(1_000)).unwrap_or(u64::MAX);
+        let span = match (visitor.transition_held_us, visitor.flush_wait_us) {
+            (Some(micros), None) => WindowSpan {
+                kind: WindowSpanKind::TransitionHeld,
+                end,
+                duration_ns: micros_to_nanos(micros),
+                blocks: None,
+            },
+            (None, Some(micros)) => WindowSpan {
+                kind: if visitor.failed {
+                    WindowSpanKind::FlushFailed
+                } else {
+                    WindowSpanKind::FlushWait
+                },
+                end,
+                duration_ns: micros_to_nanos(micros),
+                blocks: visitor.blocks,
+            },
+            _ => return,
+        };
+        WINDOW_SPANS.lock().push(span);
+    }
+
+    fn enter(&self, _: &tracing::span::Id) {}
+
+    fn exit(&self, _: &tracing::span::Id) {}
+}
+
+/// Scoped recorder for the unit's replay: it keeps only the two existing
+/// window-gate histograms and drops every other metric.
+struct WindowSeriesRecorder;
+
+impl metrics::Recorder for WindowSeriesRecorder {
+    fn describe_counter(
+        &self,
+        _: metrics::KeyName,
+        _: Option<metrics::Unit>,
+        _: metrics::SharedString,
+    ) {
+    }
+
+    fn describe_gauge(
+        &self,
+        _: metrics::KeyName,
+        _: Option<metrics::Unit>,
+        _: metrics::SharedString,
+    ) {
+    }
+
+    fn describe_histogram(
+        &self,
+        _: metrics::KeyName,
+        _: Option<metrics::Unit>,
+        _: metrics::SharedString,
+    ) {
+    }
+
+    fn register_counter(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
+        metrics::Counter::noop()
+    }
+
+    fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
+        metrics::Gauge::noop()
+    }
+
+    fn register_histogram(
+        &self,
+        key: &metrics::Key,
+        _: &metrics::Metadata<'_>,
+    ) -> metrics::Histogram {
+        let kind = match key.name() {
+            "node.sync.apply_idle_seconds" => WindowSpanKind::ApplyIdle,
+            "node.sync.download_blocked_by_apply_seconds" => WindowSpanKind::DownloadBlocked,
+            _ => return metrics::Histogram::noop(),
+        };
+        metrics::Histogram::from_arc(Arc::new(WindowSeriesHistogram { kind }))
+    }
+}
+
+struct WindowSeriesHistogram {
+    kind: WindowSpanKind,
+}
+
+impl metrics::HistogramFn for WindowSeriesHistogram {
+    fn record(&self, seconds: f64) {
+        let end = quanta::Instant::now();
+        WINDOW_SPANS.lock().push(WindowSpan {
+            kind: self.kind,
+            end,
+            duration_ns: (seconds * 1e9).round() as u64,
+            blocks: None,
+        });
+    }
+}
+
+/// Everything one replay of one window measured.
+struct WindowUnit {
+    started: quanta::Instant,
+    apply_started: quanta::Instant,
+    /// Ticks 1..=4: request, stage, apply, and the closing tick.
+    wall: Duration,
+    /// Tick 3 alone, the one that applies the window.
+    apply_wall: Duration,
+    /// Tick 4, which closes the measurement-only intervals.
+    settle_wall: Duration,
+    /// Process CPU in `/proc` ticks across `wall`, dead threads included.
+    cpu_ticks: u64,
+    rss_start_bytes: u64,
+    /// Process peak resident set across `wall`; the peak was reset first.
+    rss_peak_bytes: u64,
+    requested: usize,
+    applied_height: u32,
+    corpus_sha256: Sha256Hex,
+    spans: Vec<WindowSpan>,
+}
+
+impl WindowUnit {
+    fn spans_of(&self, kind: WindowSpanKind) -> impl Iterator<Item = &WindowSpan> {
+        self.spans.iter().filter(move |span| span.kind == kind)
+    }
+
+    fn durations_ns(&self, kind: WindowSpanKind) -> Vec<u64> {
+        self.spans_of(kind).map(|span| span.duration_ns).collect()
+    }
+
+    fn cpu_ns(&self) -> u64 {
+        self.cpu_ticks.saturating_mul(WINDOW_CPU_TICK_NS)
+    }
+
+    /// The window's commit transition is the longest hold in the unit; any
+    /// other hold (genesis bootstrap) stays in the raw record only.
+    fn window_transition_held_us(&self) -> u64 {
+        self.spans_of(WindowSpanKind::TransitionHeld)
+            .map(|span| span.duration_ns / 1_000)
+            .max()
+            .unwrap_or(0)
+    }
+}
+
+fn duration_nanos(duration: Duration) -> u64 {
+    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
+}
+
+fn read_proc_text(path: &str) -> String {
+    std::fs::read_to_string(path).unwrap_or_else(|error| panic!("read {path}: {error}"))
+}
+
+/// The text after `field` on its line of a `/proc` file, trimmed.
+fn proc_field(path: &str, field: &str) -> String {
+    read_proc_text(path)
+        .lines()
+        .find_map(|line| line.strip_prefix(field))
+        .map_or_else(
+            || panic!("{field} missing from {path}"),
+            |rest| rest.trim().to_owned(),
+        )
+}
+
+fn proc_kib(path: &str, field: &str) -> u64 {
+    proc_field(path, field)
+        .split_ascii_whitespace()
+        .next()
+        .and_then(|value| value.parse().ok())
+        .unwrap_or_else(|| panic!("{field} in {path} is not a KiB count"))
+}
+
+/// Process CPU time, user plus system, in `/proc` clock ticks. Counts every
+/// thread including exited ones, so scoped worker threads are not lost; the
+/// price is `WINDOW_CPU_TICK_NS` resolution, which is why runs summarise it
+/// as a total beside the per-unit values.
+fn process_cpu_ticks() -> u64 {
+    let stat = read_proc_text("/proc/self/stat");
+    // Fields after the parenthesised command name start at field 3; utime
+    // and stime are fields 14 and 15.
+    let after_comm = stat
+        .rsplit_once(')')
+        .map_or_else(|| panic!("malformed /proc/self/stat"), |(_, rest)| rest);
+    let mut fields = after_comm.split_ascii_whitespace().skip(11);
+    let mut next_ticks = |name: &str| -> u64 {
+        fields
+            .next()
+            .and_then(|value| value.parse().ok())
+            .unwrap_or_else(|| panic!("{name} missing from /proc/self/stat"))
+    };
+    let user = next_ticks("utime");
+    let system = next_ticks("stime");
+    user.saturating_add(system)
+}
+
+/// Resets the process peak resident set to the current one (`proc(5)`:
+/// writing `5` to `clear_refs`, Linux 4.0 and later), so the next `VmHWM`
+/// read is the peak of the interval that follows. Returns the resident set
+/// that interval starts from.
+///
+/// Verifies the reset took effect: after it the peak must sit within 1 MiB of
+/// the resident set, otherwise a stale lifetime peak would be reported as
+/// this unit's peak.
+fn reset_peak_rss() -> u64 {
+    std::fs::OpenOptions::new()
+        .write(true)
+        .open("/proc/self/clear_refs")
+        .and_then(|mut file| file.write_all(b"5"))
+        .unwrap_or_else(|error| panic!("reset peak RSS through /proc/self/clear_refs: {error}"));
+    let resident_kib = proc_kib("/proc/self/status", "VmRSS:");
+    let peak_kib = proc_kib("/proc/self/status", "VmHWM:");
+    assert!(
+        peak_kib.saturating_sub(resident_kib) <= 1024,
+        "VmHWM is {peak_kib} KiB after the reset but VmRSS is {resident_kib} KiB: \
+         this kernel did not reset the peak resident set"
+    );
+    resident_kib.saturating_mul(1024)
+}
+
+/// Toolchain and build facts the six-identity record needs beyond
+/// `EvidenceIdentity`.
+fn window_toolchain_identity() -> sonic_rs::Value {
+    let rustc = std::env::var_os("RUSTC").unwrap_or_else(|| "rustc".into());
+    let output = std::process::Command::new(&rustc)
+        .arg("-vV")
+        .output()
+        .unwrap_or_else(|error| {
+            panic!(
+                "run {} -vV for the toolchain identity: {error}",
+                rustc.display()
+            )
+        });
+    assert!(output.status.success(), "{} -vV failed", rustc.display());
+    let features: Vec<&str> = [
+        ("fjall", cfg!(feature = "fjall")),
+        ("rocksdb", cfg!(feature = "rocksdb")),
+        ("redb", cfg!(feature = "redb")),
+        ("kernel", cfg!(feature = "kernel")),
+        ("zmq", cfg!(feature = "zmq")),
+    ]
+    .into_iter()
+    .filter_map(|(name, enabled)| enabled.then_some(name))
+    .collect();
+    json!({
+        "rustc": String::from_utf8_lossy(&output.stdout).trim().to_owned(),
+        "profile": "bench",
+        "debug_assertions": cfg!(debug_assertions),
+        "features": features,
+        "allocator": "system",
+        "validation_engine": sample_engine_label(),
+    })
+}
+
+/// Host facts beyond `EvidenceIdentity.hardware`: kernel, pinned core set,
+/// memory, and the filesystem holding the durable state.
+///
+/// Fails closed on a RAM-backed temporary directory: a group flush there is
+/// not a durable write, so `flush_wait_us` would measure nothing.
+fn window_host_identity() -> sonic_rs::Value {
+    let tmpdir = std::env::temp_dir()
+        .canonicalize()
+        .unwrap_or_else(|error| panic!("resolve the temporary directory: {error}"));
+    let mounts = read_proc_text("/proc/self/mounts");
+    let (device, mount_point, fstype) = mounts
+        .lines()
+        .filter_map(|line| {
+            let mut fields = line.split_ascii_whitespace();
+            let device = fields.next()?;
+            let mount_point = fields.next()?;
+            let fstype = fields.next()?;
+            tmpdir
+                .starts_with(mount_point)
+                .then_some((device, mount_point, fstype))
+        })
+        .max_by_key(|(_, mount_point, _)| mount_point.len())
+        .unwrap_or_else(|| panic!("no mount holds {}", tmpdir.display()));
+    assert!(
+        !matches!(fstype, "tmpfs" | "ramfs"),
+        "{} is on {fstype}: set TMPDIR to a disk-backed directory so group flushes are durable",
+        tmpdir.display()
+    );
+    json!({
+        "kernel": read_proc_text("/proc/sys/kernel/osrelease").trim(),
+        "cpus_allowed_list": proc_field("/proc/self/status", "Cpus_allowed_list:"),
+        "mem_total_kib": proc_kib("/proc/meminfo", "MemTotal:"),
+        "storage": {
+            "tmpdir": tmpdir.display().to_string(),
+            "mount_point": mount_point,
+            "fstype": fstype,
+            "device": device,
+        },
+    })
+}
+
+/// Replays one window on fresh durable state and measures it.
+///
+/// Setup (state, headers, pre-serialized payloads) is outside every timer
+/// and every counter. The measured interval is the whole replay: the request
+/// tick, the out-of-order staging tick, the applying tick, and one closing
+/// tick that lets the window close its measurement-only intervals.
+fn run_window_unit(scenario: WindowScenario) -> WindowUnit {
+    let mut fixture =
+        ProductionStateSyncFixture::window_replay(scenario.body_blocks).prebuild_run();
+    let corpus_bytes: Vec<u8> = fixture.blocks.iter().flat_map(consensus_bytes).collect();
+    let corpus_sha256 = Sha256Hex::digest(&corpus_bytes);
+    let recorder = WindowSeriesRecorder;
+    WINDOW_SPANS.lock().clear();
+    let rss_start_bytes = reset_peak_rss();
+    let cpu_before = process_cpu_ticks();
+    let started = quanta::Instant::now();
+    let mut apply_started = started;
+    let mut apply_done = started;
+    let mut settled = started;
+    let mut report = None;
+    metrics::with_local_recorder(&recorder, || {
+        report = Some(fixture.run_phased(|phase| match phase {
+            ReplayPhase::ApplyStart => apply_started = quanta::Instant::now(),
+            ReplayPhase::ApplyDone => apply_done = quanta::Instant::now(),
+        }));
+        fixture.settle_tick();
+        settled = quanta::Instant::now();
+    });
+    let cpu_after = process_cpu_ticks();
+    let rss_peak_bytes = proc_kib("/proc/self/status", "VmHWM:").saturating_mul(1024);
+    let spans = std::mem::take(&mut *WINDOW_SPANS.lock());
+    let report = report.unwrap_or_else(|| panic!("window replay produced no report"));
+    drop(fixture);
+    WindowUnit {
+        started,
+        apply_started,
+        wall: settled.duration_since(started),
+        apply_wall: apply_done.duration_since(apply_started),
+        settle_wall: settled.duration_since(apply_done),
+        cpu_ticks: cpu_after.saturating_sub(cpu_before),
+        rss_start_bytes,
+        rss_peak_bytes,
+        requested: report.requested,
+        applied_height: report.applied_height,
+        corpus_sha256,
+        spans,
+    }
+}
+
+/// Fails closed unless the replay did what the scenario names: every body
+/// block applied, one group flush per started group covering every block,
+/// no failed flush, and the transition hold present.
+fn assert_window_unit(scenario: WindowScenario, unit: &WindowUnit) {
+    let body_blocks = usize::try_from(scenario.body_blocks)
+        .unwrap_or_else(|error| panic!("invalid window body count: {error}"));
+    assert_eq!(
+        unit.applied_height, scenario.body_blocks,
+        "{}: the replay must apply every body block",
+        scenario.id
+    );
+    assert!(
+        unit.requested >= body_blocks,
+        "{}: the first getdata asked for {} blocks, fewer than the {body_blocks} replayed",
+        scenario.id,
+        unit.requested
+    );
+    assert_eq!(
+        unit.spans_of(WindowSpanKind::FlushFailed).count(),
+        0,
+        "{}: a group flush failed during the replay",
+        scenario.id
+    );
+    assert_eq!(
+        unit.spans_of(WindowSpanKind::FlushWait).count(),
+        scenario.groups(),
+        "{}: expected one group flush per started group of {WINDOW_GROUP_BLOCKS} blocks",
+        scenario.id
+    );
+    let flushed: u64 = unit
+        .spans_of(WindowSpanKind::FlushWait)
+        .filter_map(|span| span.blocks)
+        .sum();
+    assert_eq!(
+        flushed,
+        u64::from(scenario.body_blocks),
+        "{}: group flushes must cover every body block",
+        scenario.id
+    );
+    assert!(
+        unit.spans_of(WindowSpanKind::TransitionHeld)
+            .next()
+            .is_some(),
+        "{}: no transition_held_us event; this binary lacks the chainstate timing spans",
+        scenario.id
+    );
+}
+
+/// Order statistics of `values` plus whether p99 is supported; an empty set
+/// is reported as unobserved, never as zero.
+fn quantiles_json(values: &[u64]) -> sonic_rs::Value {
+    if let Some(quantiles) = evidence::Quantiles::of(values) {
+        json!({
+            "n": quantiles.n,
+            "p50": quantiles.p50,
+            "p95": quantiles.p95,
+            "p99": quantiles.p99,
+            "max": quantiles.max,
+            "p99_supported": quantiles.p99_supported(),
+        })
+    } else {
+        json!({ "n": 0, "observed": false })
+    }
+}
+
+fn window_unit_json(index: usize, origin: quanta::Instant, unit: &WindowUnit) -> sonic_rs::Value {
+    let micros = |kind: WindowSpanKind| -> Vec<u64> {
+        unit.spans_of(kind)
+            .map(|span| span.duration_ns / 1_000)
+            .collect()
+    };
+    json!({
+        "unit": index,
+        "warmup": index < WINDOW_WARMUP_UNITS,
+        "started_offset_ns": duration_nanos(unit.started.duration_since(origin)),
+        "wall_ns": duration_nanos(unit.wall),
+        "apply_wall_ns": duration_nanos(unit.apply_wall),
+        "settle_wall_ns": duration_nanos(unit.settle_wall),
+        "cpu_ticks": unit.cpu_ticks,
+        "cpu_ns": unit.cpu_ns(),
+        "rss_start_bytes": unit.rss_start_bytes,
+        "rss_peak_bytes": unit.rss_peak_bytes,
+        "requested_blocks": unit.requested,
+        "applied_height": unit.applied_height,
+        "transition_held_us": micros(WindowSpanKind::TransitionHeld),
+        "flush_wait_us": micros(WindowSpanKind::FlushWait),
+        "flush_blocks": unit
+            .spans_of(WindowSpanKind::FlushWait)
+            .filter_map(|span| span.blocks)
+            .collect::<Vec<u64>>(),
+        "apply_idle_ns": unit.durations_ns(WindowSpanKind::ApplyIdle),
+        "download_blocked_by_apply_ns": unit.durations_ns(WindowSpanKind::DownloadBlocked),
+    })
+}
+
+/// Summaries over the measured units. Each is read beside the raw units in
+/// the same file, never instead of them. Production spans are nested inside
+/// the wall and inside each other, so none is summed with another.
+fn window_summary_json(measured: &[WindowUnit]) -> sonic_rs::Value {
+    let walls: Vec<u64> = measured
+        .iter()
+        .map(|unit| duration_nanos(unit.wall))
+        .collect();
+    let apply_walls: Vec<u64> = measured
+        .iter()
+        .map(|unit| duration_nanos(unit.apply_wall))
+        .collect();
+    let settle_walls: Vec<u64> = measured
+        .iter()
+        .map(|unit| duration_nanos(unit.settle_wall))
+        .collect();
+    let cpu: Vec<u64> = measured.iter().map(WindowUnit::cpu_ns).collect();
+    let rss_peak: Vec<u64> = measured.iter().map(|unit| unit.rss_peak_bytes).collect();
+    let rss_over_start: Vec<u64> = measured
+        .iter()
+        .map(|unit| unit.rss_peak_bytes.saturating_sub(unit.rss_start_bytes))
+        .collect();
+    let transition_held_us: Vec<u64> = measured
+        .iter()
+        .map(WindowUnit::window_transition_held_us)
+        .collect();
+    let pooled = |kind: WindowSpanKind| -> Vec<u64> {
+        measured
+            .iter()
+            .flat_map(|unit| unit.durations_ns(kind))
+            .collect()
+    };
+    let flush_wait_us: Vec<u64> = pooled(WindowSpanKind::FlushWait)
+        .into_iter()
+        .map(|nanos| nanos / 1_000)
+        .collect();
+    let cpu_total_ns = cpu
+        .iter()
+        .fold(0_u64, |total, nanos| total.saturating_add(*nanos));
+    let observed_units = |kind: WindowSpanKind| {
+        measured
+            .iter()
+            .filter(|unit| unit.spans_of(kind).next().is_some())
+            .count()
+    };
+    json!({
+        "units": measured.len(),
+        "wall_ns": quantiles_json(&walls),
+        "apply_wall_ns": quantiles_json(&apply_walls),
+        "settle_wall_ns": quantiles_json(&settle_walls),
+        "cpu_ns": quantiles_json(&cpu),
+        "cpu_total_ns": cpu_total_ns,
+        "cpu_resolution_ns": WINDOW_CPU_TICK_NS,
+        "rss_peak_bytes": quantiles_json(&rss_peak),
+        "rss_peak_over_start_bytes": quantiles_json(&rss_over_start),
+        "transition_held_us": quantiles_json(&transition_held_us),
+        "flush_wait_us": quantiles_json(&flush_wait_us),
+        "apply_idle_ns": quantiles_json(&pooled(WindowSpanKind::ApplyIdle)),
+        "apply_idle_observed_units": observed_units(WindowSpanKind::ApplyIdle),
+        "download_blocked_by_apply_ns": quantiles_json(&pooled(WindowSpanKind::DownloadBlocked)),
+        "download_blocked_by_apply_observed_units": observed_units(WindowSpanKind::DownloadBlocked),
+    })
+}
+
+fn window_sample(identity: &EvidenceIdentity, path: &str, start_ns: u64, end_ns: u64) -> Sample {
+    Sample {
+        path: path.into(),
+        owner: "node".into(),
+        identity: identity.clone(),
+        interval: Interval {
+            kind: IntervalKind::Inside,
+            start_ns,
+            end_ns: end_ns.max(start_ns),
+        },
+        cpu_ns: None,
+        elapsed_ns: end_ns.saturating_sub(start_ns),
+        rss_peak_bytes: None,
+        io_bytes: None,
+        storage_bytes: None,
+    }
+}
+
+/// One wall sample, one apply-tick sample and one sample per captured span.
+/// Offsets are on the run clock, so the nested spans overlap the wall and
+/// the ledger refuses to add them.
+fn window_evidence_samples(
+    identity: &EvidenceIdentity,
+    origin: quanta::Instant,
+    unit: &WindowUnit,
+) -> Vec<Sample> {
+    let offset = |at: quanta::Instant| duration_nanos(at.duration_since(origin));
+    let mut samples = Vec::with_capacity(unit.spans.len().saturating_add(2));
+    let started = offset(unit.started);
+    samples.push(Sample {
+        cpu_ns: Some(unit.cpu_ns()),
+        rss_peak_bytes: Some(unit.rss_peak_bytes),
+        ..window_sample(
+            identity,
+            "apply.window.replay_wall",
+            started,
+            started.saturating_add(duration_nanos(unit.wall)),
+        )
+    });
+    let apply_started = offset(unit.apply_started);
+    samples.push(window_sample(
+        identity,
+        "apply.window.apply_tick",
+        apply_started,
+        apply_started.saturating_add(duration_nanos(unit.apply_wall)),
+    ));
+    for span in &unit.spans {
+        let end = offset(span.end);
+        samples.push(window_sample(
+            identity,
+            span.kind.ledger_path(),
+            end.saturating_sub(span.duration_ns),
+            end,
+        ));
+    }
+    samples
+}
+
+/// Runs one arm of one scenario in this process and writes its evidence:
+/// every measured unit as identity-bearing ledger samples
+/// (`target/benchmarks/window_attribution.toml`, append-only across runs) and
+/// one self-contained JSON record with raw units, summaries and the six
+/// identities (`target/benchmarks/window_attribution/`).
+///
+/// The deterministic regtest workload proves the mechanism and the
+/// measurement surface; its walls are not mainnet speed and fill no product
+/// cell (`HPA-09`, `HPA-12`).
+fn run_window_attribution(arm: WindowArm, scenario: WindowScenario) {
+    let host = window_host_identity();
+    let toolchain = window_toolchain_identity();
+    tracing::subscriber::set_global_default(WindowSpanSubscriber)
+        .unwrap_or_else(|error| panic!("window span subscriber install failed: {error}"));
+    let run_started_unix_ns = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_else(|error| panic!("system clock before the epoch: {error}"))
+        .as_nanos();
+    let origin = quanta::Instant::now();
+    let total_units = WINDOW_WARMUP_UNITS.saturating_add(WINDOW_MEASURED_UNITS);
+    let mut units = Vec::with_capacity(total_units);
+    for _ in 0..total_units {
+        let unit = run_window_unit(scenario);
+        assert_window_unit(scenario, &unit);
+        units.push(unit);
+    }
+    let corpus_sha256 = units[0].corpus_sha256;
+    assert!(
+        units.iter().all(|unit| unit.corpus_sha256 == corpus_sha256),
+        "{}: replay blocks differ between units",
+        scenario.id
+    );
+    let mut identity = EvidenceIdentity::of_process(&production_state_config())
+        .unwrap_or_else(|error| panic!("evidence identity: {error}"));
+    identity.corpus = Some(CorpusIdentity {
+        id: format!("sync_pipeline.window_replay.{}", scenario.id),
+        manifest_sha256: corpus_sha256,
+    });
+    let measured = &units[WINDOW_WARMUP_UNITS..];
+
+    let cell = format!("window_attribution.{}.{}", scenario.id, arm.label());
+    let ledger_path = bench_ledger_path("window_attribution.toml");
+    let mut ledger = load_bench_ledger(&ledger_path);
+    if !ledger.cells.iter().any(|candidate| candidate.id == cell) {
+        ledger.cells.push(Cell {
+            id: cell.clone(),
+            samples: Vec::new(),
+        });
+    }
+    for unit in measured {
+        for sample in window_evidence_samples(&identity, origin, unit) {
+            ledger
+                .record(&cell, sample)
+                .unwrap_or_else(|error| panic!("record window sample: {error}"));
+        }
+    }
+    store_bench_ledger(&ledger_path, &ledger);
+
+    let record = json!({
+        "schema": WINDOW_ATTRIBUTION_SCHEMA,
+        "claim_scope": "diagnostic: deterministic regtest replay; mechanism and measurement surface only, not mainnet speed, fills no product cell",
+        "arm": arm.label(),
+        "scenario": scenario.id,
+        "body_blocks": scenario.body_blocks,
+        "group_blocks": WINDOW_GROUP_BLOCKS,
+        "groups": scenario.groups(),
+        "run_started_unix_ns": run_started_unix_ns.to_string(),
+        "warmup_units": WINDOW_WARMUP_UNITS,
+        "measured_units": WINDOW_MEASURED_UNITS,
+        "identity": identity,
+        "toolchain": toolchain,
+        "host": host,
+        "ledger": { "path": ledger_path.display().to_string(), "cell": cell },
+        "summary": window_summary_json(measured),
+        "units": units
+            .iter()
+            .enumerate()
+            .map(|(index, unit)| window_unit_json(index, origin, unit))
+            .collect::<Vec<_>>(),
+    });
+    let record_dir = bench_ledger_path("window_attribution");
+    std::fs::create_dir_all(&record_dir)
+        .unwrap_or_else(|error| panic!("window attribution record dir: {error}"));
+    let record_path = record_dir.join(format!(
+        "{}.{}.{run_started_unix_ns}.json",
+        arm.label(),
+        scenario.id
+    ));
+    let rendered = sonic_rs::to_vec_pretty(&record)
+        .unwrap_or_else(|error| panic!("render window attribution record: {error}"));
+    std::fs::write(&record_path, rendered)
+        .unwrap_or_else(|error| panic!("write window attribution record: {error}"));
+
+    println!(
+        "window_attribution arm={} scenario={} measured_units={} record={}",
+        arm.label(),
+        scenario.id,
+        measured.len(),
+        record_path.display()
+    );
+    let walls: Vec<Duration> = measured.iter().map(|unit| unit.wall).collect();
+    let apply_walls: Vec<Duration> = measured.iter().map(|unit| unit.apply_wall).collect();
+    print_percentiles("window_attribution_wall", &walls);
+    print_percentiles("window_attribution_apply_tick", &apply_walls);
+}
+
+/// Validates the `--window-attribution` invocation shape before any fixture
+/// work: the flag, one arm, one scenario, and nothing else. Cargo appends
+/// `--bench`, which carries no meaning here and is ignored.
+fn window_attribution_cli(args: &[String]) -> (WindowArm, WindowScenario) {
+    let args: Vec<&str> = args
+        .iter()
+        .map(String::as_str)
+        .filter(|arg| *arg != "--bench")
+        .collect();
+    let fail = |message: &str| -> ! {
+        eprintln!("window-attribution: {message}");
+        eprintln!(
+            "usage: sync_pipeline --window-attribution <control|candidate> <{}>",
+            WINDOW_SCENARIOS.map(|scenario| scenario.id).join("|")
+        );
+        std::process::exit(2);
+    };
+    if args.len() != 3 || args[0] != "--window-attribution" {
+        fail("expected exactly --window-attribution <arm> <scenario>");
+    }
+    let Some(arm) = WindowArm::parse(args[1]) else {
+        fail("unknown arm");
+    };
+    let Some(scenario) = WINDOW_SCENARIOS
+        .iter()
+        .copied()
+        .find(|scenario| scenario.id == args[2])
+    else {
+        fail("unknown scenario");
+    };
+    (arm, scenario)
+}
+
 /// Binary entrypoint: `--mrp-verify` runs the untimed real-fixture
 /// verification path with no Criterion involvement and no timing output;
+/// `--window-attribution ARM SCENARIO` replays one windowed apply on fresh
+/// durable state, measures it, and writes ledger and JSON evidence;
 /// `--mrp-list-counted` lists the counted row IDs without fixture
 /// construction or sampling; `--mrp-counted ROW_ID` runs one counted row
 /// through the shared measured-work sampler; `--mrp-verify-counted` runs
@@ -4457,6 +5421,11 @@ fn main() {
         run_counted_row(row);
         return;
     }
+    if args.iter().any(|arg| arg == "--window-attribution") {
+        let (arm, scenario) = window_attribution_cli(&args);
+        run_window_attribution(arm, scenario);
+        return;
+    }
     if args.iter().any(|arg| arg == "--mrp-verify-counted") {
         require_bare_counted_flag(&args, "--mrp-verify-counted");
         mrp_verify_counted();
diff --git a/crates/node/src/metrics.rs b/crates/node/src/metrics.rs
index 2b46d8689..4166ffe49 100644
--- a/crates/node/src/metrics.rs
+++ b/crates/node/src/metrics.rs
@@ -211,7 +211,23 @@ fn describe_node_metrics() {
     );
     metrics::describe_gauge!(
         "node.sync.staged_bytes_high_water",
-        "highest staged byte total observed"
+        "highest staged serialized byte total observed"
+    );
+    metrics::describe_gauge!(
+        "node.sync.staged_decoded_bytes",
+        "decoded heap capacity bytes owned by staged blocks, excluding allocator rounding"
+    );
+    metrics::describe_gauge!(
+        "node.sync.staged_retained_bytes",
+        "attributed staged bytes: decoded heap capacity plus preserved serialized buffers, excluding allocator rounding; not process RSS"
+    );
+    metrics::describe_gauge!(
+        "node.sync.staged_retained_bytes_high_water",
+        "highest attributed staged retained byte total observed"
+    );
+    metrics::describe_gauge!(
+        "node.sync.inbound_block_queue_depth",
+        "block bodies queued in the bounded inbound channel when the sync drain pass emitted its gauges, after it emptied the receiver; a residual, not a peak"
     );
     metrics::describe_counter!(
         "storage.writes_total",
diff --git a/crates/p2p/src/block_stager.rs b/crates/p2p/src/block_stager.rs
index 5259fc998..b044d853d 100644
--- a/crates/p2p/src/block_stager.rs
+++ b/crates/p2p/src/block_stager.rs
@@ -156,7 +156,6 @@ impl BlockStager {
 
     /// Decoded heap bytes owned by currently staged blocks (capacity-based,
     /// excluding allocator rounding and the preserved serialized buffers).
-    #[cfg(test)]
     #[must_use]
     pub(crate) const fn decoded_bytes(&self) -> usize {
         self.decoded_bytes
@@ -170,8 +169,8 @@ impl BlockStager {
         self.decoded_bytes.saturating_add(self.received_bytes)
     }
 
-    /// Highest attributed retained total ever observed this run.
-    #[cfg(test)]
+    /// Highest attributed retained total ever observed this run; feeds the
+    /// retained-bytes high-water gauge.
     #[must_use]
     pub(crate) const fn retained_attributed_bytes_high_water(&self) -> usize {
         self.retained_bytes_high_water
diff --git a/crates/p2p/src/download_window.rs b/crates/p2p/src/download_window.rs
index a4da44706..e0f3bb6fe 100644
--- a/crates/p2p/src/download_window.rs
+++ b/crates/p2p/src/download_window.rs
@@ -79,6 +79,7 @@ impl PeerRequest {
     pub(crate) fn len(&self) -> usize {
         self.entries.len()
     }
+
 }
 
 /// Per-peer availability view for one request build.
diff --git a/crates/p2p/src/sync/historical.rs b/crates/p2p/src/sync/historical.rs
index 465b5b7cd..e0ebc356b 100644
--- a/crates/p2p/src/sync/historical.rs
+++ b/crates/p2p/src/sync/historical.rs
@@ -185,6 +185,7 @@ impl BlockSync {
             let Some(request) = request else {
                 continue;
             };
+
             let inventory = request
                 .entries()
                 .map(|(_, hash)| {
diff --git a/crates/p2p/src/sync/requests.rs b/crates/p2p/src/sync/requests.rs
index 8762a28f4..ca7aafb91 100644
--- a/crates/p2p/src/sync/requests.rs
+++ b/crates/p2p/src/sync/requests.rs
@@ -109,7 +109,6 @@ impl UnavailableBodies {
             })
             .collect()
     }
-
     /// Forgets `source`'s answer for `hash`: it delivered the body, so the
     /// earlier `notfound` no longer describes it.
     pub(super) fn clear(&mut self, source: PeerSource, hash: Hash256) {
diff --git a/crates/p2p/src/sync/telemetry.rs b/crates/p2p/src/sync/telemetry.rs
index 38239ed22..3346db0e9 100644
--- a/crates/p2p/src/sync/telemetry.rs
+++ b/crates/p2p/src/sync/telemetry.rs
@@ -64,6 +64,20 @@ impl BlockSync {
         )
     }
 
+    /// Publishes the staged-body, pending-window, and inbound-queue gauges
+    /// from one scheduler-lock hold.
+    ///
+    /// `staged_decoded_bytes` and `staged_retained_bytes` attribute staged
+    /// decoded-heap capacity and preserved serialized buffers; they exclude
+    /// allocator rounding, are not a process-RSS measure, and never feed an
+    /// admission decision. `received_bytes` and `staged_bytes_high_water`
+    /// remain the serialized fact the byte budget reads.
+    /// `inbound_block_queue_depth` is the bounded inbound channel's length at
+    /// emission. The only caller runs after the drain pass has emptied the
+    /// receiver, so it reports the residual queue that arrived during or
+    /// after the pass; bodies the pass already pulled are in the staged
+    /// gauges, or in the apply batch the stager no longer owns, and are not
+    /// attributed in bytes.
     pub(super) fn record_sync_metrics(&self) {
         let scheduler = self.scheduler.lock();
         let window = &scheduler.window;
@@ -81,6 +95,13 @@ impl BlockSync {
             .set(metric_count(stager.received_high_water()));
         metrics::gauge!("node.sync.staged_bytes_high_water")
             .set(metric_count(stager.received_bytes_high_water()));
+        metrics::gauge!("node.sync.staged_decoded_bytes").set(metric_bytes(stager.decoded_bytes()));
+        metrics::gauge!("node.sync.staged_retained_bytes")
+            .set(metric_bytes(stager.retained_attributed_bytes()));
+        metrics::gauge!("node.sync.staged_retained_bytes_high_water")
+            .set(metric_bytes(stager.retained_attributed_bytes_high_water()));
+        metrics::gauge!("node.sync.inbound_block_queue_depth")
+            .set(metric_count(self.inbound_blocks_rx.len()));
     }
 
     pub(super) fn record_pending_sync_metrics(&self) {
@@ -94,3 +115,31 @@ impl BlockSync {
 pub(super) fn metric_count(value: usize) -> f64 {
     f64::from(u32::try_from(value).unwrap_or(u32::MAX))
 }
+
+/// A byte attribution as a gauge value without the `u32` clamp of
+/// [`metric_count`]: the retained gauges exist to show residency growing past
+/// the serialized budget, so a total above 4 GiB must not saturate. The high
+/// and low 32-bit halves each convert to `f64` exactly.
+fn metric_bytes(value: usize) -> f64 {
+    let value = u64::try_from(value).unwrap_or(u64::MAX);
+    let high = u32::try_from(value >> 32).unwrap_or(u32::MAX);
+    let low = u32::try_from(value & u64::from(u32::MAX)).unwrap_or(u32::MAX);
+    f64::from(high).mul_add(4_294_967_296.0, f64::from(low))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::metric_bytes;
+
+    #[test]
+    fn metric_bytes_is_exact_across_the_u32_boundary() {
+        assert_eq!(metric_bytes(0).to_bits(), 0.0_f64.to_bits());
+        assert_eq!(metric_bytes(4_096).to_bits(), 4_096.0_f64.to_bits());
+        if let Ok(above_u32) = usize::try_from(5_000_000_000_u64) {
+            assert_eq!(
+                metric_bytes(above_u32).to_bits(),
+                5_000_000_000.0_f64.to_bits()
+            );
+        }
+    }
+}
diff --git a/docs/benchmarks/window-pipeline-attribution.md b/docs/benchmarks/window-pipeline-attribution.md
index 05946e9c9..4875aab5d 100644
--- a/docs/benchmarks/window-pipeline-attribution.md
+++ b/docs/benchmarks/window-pipeline-attribution.md
@@ -6,16 +6,7 @@ gates for the windowed block apply and its durable group commit (ENG-91, issue
 generates, never here (`HPA-08`, `HPA-12`).
 
 Status: `planned_not_executed`. No arm has run under this protocol, so no
-cell, ratio or percentile is on this page. Implementation status is scoped to
-the branch: the diagnostic spans (section 2) and replay harness (section 3)
-are implemented in [PR #1701](https://github.com/gosuda/bitcoin-rs/pull/1701)
-(`994e82acf425ff420921238514545c99c75ea4f4`), but are absent from the
-documentation-cleanup branch (#1703). On #1701, `crates/chainstate/src/lib.rs`
-emits `transition_held_us`, `window.rs` emits `flush_wait_us`, and
-`crates/node/benches/sync_pipeline.rs::main` dispatches `--window-attribution`.
-Use that implementation branch for the control arm; the invocation described
-in section 3 cannot run on #1703 until those changes are integrated. Implementation does not prove
-that a measurement campaign ran. Rules shared by all cells are in
+cell, ratio or percentile is on this page. Rules shared by all cells are in
 [README.md](README.md); the attribution method is
 [`docs/contracts/hot-path-attribution.md`](../contracts/hot-path-attribution.md).
 
@@ -23,23 +14,21 @@ that a measurement campaign ran. Rules shared by all cells are in
 
 | Item | State |
 |---|---|
-| Baseline pipeline | `ChainTransition::connect_window` in `crates/chainstate/src/lib.rs` (delegating to `apply_window_admitted` in `crates/chainstate/src/window.rs`): `prove_window` proves the whole window, then the sequential per-block loop stages blocks into a `WindowGroup` and flushes at 64 blocks or 8 MiB, all under the one transition the caller holds. |
+| Baseline pipeline | `apply_window_admitted` in `crates/chainstate/src/window.rs`: `prove_window` proves the whole window, then the sequential per-block loop stages blocks into a `WindowGroup` and flushes at 64 blocks or 8 MiB, all under one chain transition. |
 | Candidate S-A | Not implemented. The tree has no `prove_chunk` and no scoped worker thread. Nothing here adopts it. |
-| Diagnostic spans | Implemented in #1701, absent from #1703. See section 2 for the protocol. |
+| Diagnostic spans | `transition_held_us` and `flush_wait_us` exist (section 2). |
 | Existing series | `node.sync.apply_idle_seconds` and `node.sync.download_blocked_by_apply_seconds` exist in `crates/p2p/src/download_window.rs`. |
-| Replay harness | Implemented in #1701 as `sync_pipeline --window-attribution`; absent from #1703. See section 3. |
-| Candidate arm | Defined so it can run once S-A exists. The control implementation is available in #1701; #1703 requires integration of that implementation before running it. No control-against-candidate comparison may be written without S-A and measured evidence. |
+| Replay harness | `crates/node/benches/sync_pipeline.rs`, `--window-attribution` (section 3). |
+| Candidate arm | Defined so it can run once S-A exists. Until then only the control arm is measurable, and no control-against-candidate comparison may be written. |
 
 Out-of-transition pipelines (S-B) and verify-only lock release (S-C) stay
 rejected: tip identity alone cannot fence `replace_from`, header-gate changes,
 reorg and script-cache takes.
 
-## 2. Diagnostic spans (implemented in #1701)
+## 2. Diagnostic spans
 
-In #1701, both spans use `quanta::Instant` and `tracing::debug!`. They create no metric
-family, change no lock scope and never feed a production decision. They do gate
-the #1701 harness's validity: that replay panics when a unit lacks a
-`transition_held_us` event, so a missing span fails the run instead of recording it.
+Both spans use `quanta::Instant` and `tracing::debug!`. They create no metric
+family, change no lock scope and never feed a decision.
 
 | Span | Acquire or start | Release or end | Emitted |
 |---|---|---|---|
@@ -51,23 +40,24 @@ flush` is a development diagnostic beside `durable_head: group commit`, whose
 `sync_us` and `commit_us` are sub-intervals of the same wait. The spans are
 reported side by side and never added (`HPA-12`).
 
-## 3. Replay harness (implemented in #1701)
+## 3. Replay harness
 
-This section describes the mode implemented in #1701 as a protocol reference
-only. That implementation is unavailable on the #1703 branch alone, so nothing
-in this section is executable here; run it on a #1701 checkout.
+### 3.1 Invocation
 
-### 3.1 Invocation (reference only; runs on a #1701 checkout, not on #1703)
+Run from the repository root so `rust-toolchain.toml` selects the toolchain
+that built the binary. `TMPDIR` must name a directory on a disk-backed
+filesystem, because the harness refuses `tmpfs` and `ramfs`: a group flush
+there is not a durable write.
 
-The #1701 replay is invoked as `sync_pipeline --window-attribution
-<control|candidate> <single_group|multi_group>`: one arm and one scenario per
-process in a single process run (a Cargo-appended `--bench` is ignored by the
-parser), launched from the repository root of a #1701 checkout so
-`rust-toolchain.toml` selects the toolchain that built the binary, with cores
-pinned via `taskset -c <set>` and `TMPDIR` on a disk-backed filesystem (the
-harness refuses `tmpfs` and `ramfs`: a group flush there is not a durable
-write). The exact shape above is recorded so a future integration can replay it
-verbatim; it is not an executable step for this branch.
+```bash
+cargo bench -p bitcoin-rs-node --bench sync_pipeline -- \
+  --window-attribution <control|candidate> <single_group|multi_group>
+```
+
+The flag takes exactly one arm and one scenario. Cargo appends `--bench`, which
+the parser ignores. Pin the cores with `taskset -c <set>` before `cargo bench`.
+One invocation is one run: it executes every unit of one arm and one scenario in
+one process.
 
 ### 3.2 Scenarios and units
 
@@ -149,9 +139,7 @@ names custody only; the artifact digest proves the programs differ.
 Run each scenario alternately, control then candidate, three times
 (`A-B-A-B-A-B`). With no candidate in the tree, run the control arm three times
 per scenario: the runs set the stability and the host noise floor, and no
-comparison is produced. The control arm can use the implementation in #1701;
-running it on #1703 requires integrating the spans and harness first. The
-candidate arm still requires S-A.
+comparison is produced.
 
 ## 5. Gates for the replay (`HPA-13`)
 
diff --git a/docs/observability.md b/docs/observability.md
index 808d00e4b..48776e7ca 100644
--- a/docs/observability.md
+++ b/docs/observability.md
@@ -55,6 +55,57 @@ latency. Occupancy updates follow the existing connection permit lifetime,
 including spawn failure and shutdown; no second admission counter is added.
 No paths, method names, credentials, addresses, hashes or txids are labels.
 
+For staged block bodies, `node.sync.received_bytes` and
+`node.sync.staged_bytes_high_water` stay the serialized wire fact that the
+staging byte budget and request headroom read. Four further signals attribute
+retained residency without changing any admission rule:
+
+| Signal | Meaning |
+| --- | --- |
+| `node.sync.staged_decoded_bytes` | Sum over staged blocks of `Block::retained_heap_bytes()`: the capacity of each block's transaction, input, output, script, and witness allocations. |
+| `node.sync.staged_retained_bytes` | `staged_decoded_bytes` plus the preserved serialized buffers (`received_bytes`). |
+| `node.sync.staged_retained_bytes_high_water` | Highest `staged_retained_bytes` observed this run. |
+| `node.sync.inbound_block_queue_depth` | Length of the bounded inbound channel (`inbound_block_queue_limit`) when the drain pass emitted. A count of unstaged bodies, not bytes. |
+
+The three byte signals are attribution, not measurement of the process. They
+exclude allocator rounding, the inline `Block` value, and the stager's map and
+queue bookkeeping, so they are not an RSS or `VmHWM` figure and need not equal
+one. They cover only bodies the stager owns: a body drained for apply, or
+pulled from the channel into a stage batch, is unattributed until it is
+staged or restored, so `staged_retained_bytes = 0` in the middle of an apply
+does not mean no block memory is held. A block is measured once, when the
+stager takes ownership of its allocation (arrival, or restore after a partial
+apply, which re-measures the restored allocation because the restored copy
+need not have the drained original's capacities), and released at the
+transition that releases its serialized bytes: drain, branch retirement,
+count-budget eviction, discard, expiry, or replacement. The retained high-water
+mark advances on arrival and on restore; an arrival's mark is taken before the
+count-budget eviction it causes. `node.sync.staged_blocks_high_water` and
+`node.sync.staged_bytes_high_water` keep their earlier arrival-only behavior.
+Nothing admits, evicts, or retries on these signals. They are backlog/capacity
+gauges (OBS-01) published with the other staged gauges from one scheduler-lock
+hold by the sync drain pass that saw arrivals, expiry, applies, or failures.
+That pass runs after it has emptied the receiver, so
+`inbound_block_queue_depth` is a post-drain residual, normally near zero, not
+a peak, and it is not refreshed on passes that emit nothing.
+
+Staging policy decision (retained): production keeps the fixed 256-block /
+512 MiB wire budget. An isolated count-horizon experiment (candidate
+`f24b7ded0`, committed after the matrix so source bytes match the gated binary)
+ran 48 synthetic regtest trials (4 cells x 4 policies x 3 rounds, 320 blocks
+applied each; raw samples in
+`.outline/evidence/native-gate-20261026T183745Z-1096270/gate.log`) and was
+evaluated and rejected under the fixed Step 25 gate (>=10% peak-RSS reduction,
+throughput >=98%, strictly lower retained high-water, no
+retry/starvation/stall regression): no horizon met the RSS gate (binding
+medians ~1.00-1.01x fixed vs required <=0.90), and the modern-large 8-peer
+15 s and 30 s horizons each added a stall-episode start (median 1 vs fixed 0;
+disconnects stayed 0, which does not cancel the episode regression). Five
+early-multi rows were not valid binding comparisons (all 60 s first refusals
+arrived after the corpus was fully requested; 30 s rounds 1-2 had no refusal).
+Synthetic regtest only (early/small and modern/large-OP_RETURN corpora, 3 runs
+per arm); no mainnet, p99, or whole-node claim.
+
 The call-site audit table below is a snapshot of its audited base. It has no
 row for `index.family_planned_write_bytes`. This clause describes it.
 
diff --git a/docs/validation-tooling.md b/docs/validation-tooling.md
index 31608a1b5..fba298e4f 100644
--- a/docs/validation-tooling.md
+++ b/docs/validation-tooling.md
@@ -415,3 +415,27 @@ Owners, prerequisites, commands, and scope limitations are listed in each sectio
   # Run offline hermetic synthetic smoke probe (launches no workloads):
   uv run --offline --no-project --script tools/benchmark-campaign/test_measured_campaign.py
   ```
+
+## 12a. Filesystem-aware Linux measurement boundary
+
+- **Tier**: In-tree benchmark campaign harness (diagnostic, not a product-cell qualification).
+- **Owner scripts & tools**:
+  - Scope owner: `tools/benchmark-campaign/filesystem.py` (one owner for mount/filesystem/backing-counter scope plus read-only cumulative counter samples; no project or toolchain migration, stdlib only).
+  - Contract tests: `tools/benchmark-campaign/test_filesystem.py`.
+  - Consumer: the private index-write-attribution harness delegates its former mountinfo/sysfs/diskstats lookup, counter reads, and sector deltas to this owner; the harness keeps its own strict qualification gate.
+- **Boundary**: one target filesystem per run. Apparent bytes, allocated file bytes, per-process I/O counters, and attributable host block-device writes are reported separately and never merged. Allocated-file measurement stays in the existing Rust `storage-footprint` tool and is not duplicated here.
+- **Supported scopes**:
+  - `regular_block_device`: the mount resolves to one bare block device with a readable kernel sectors-written counter. Available (host-wide for that device).
+  - `zfs_pool`: ZFS datasets report pool-wide accounting that cannot isolate one filesystem, so the scope is unavailable; no pool total is converted into a per-filesystem value.
+  - `composite_anonymous`: overlay/tmpfs/ramfs/network/fuse/9p/virtiofs mounts (and other major-0 mounts) expose no per-filesystem block counter. Unavailable.
+  - `ambiguous_multi_device`: btrfs or any stacked (mapped) device over one or more underlying devices. A mapped device is never assumed to be a single disk; underlying counters are listed, never summed. Unavailable.
+  - `unavailable`: no mount entry, missing counter file or permission, unparsable counter, counter wrap/reset/recreation, or a `--device` override that names anything other than the mount's own `MAJ:MIN` (operator assertion alone is not proof of backing). Unavailable is reported with an explicit reason, never as zero.
+- **Mechanisms checked** (read-only; the owner spawns no subprocess): `/proc/self/mountinfo` with octal-escape decoding and longest-prefix mount selection, the chosen device's `/sys/dev/block/<MAJ:MIN>/stat` sectors-written counter (kernel 512-byte block-layer unit regardless of filesystem block size), `/proc/diskstats` sectors-written recorded as a diagnostic alongside the sysfs counter (never cross-validated against it), `slaves`/`holders` topology, and `shared-mount` plus informational queue/model attributes recorded as diagnostics. Exactly one device counter is ever read per scope: a partition's own counter, never the parent disk added to it.
+- **Honest limits**: the counter is host-wide for its device (every writer included); the collector never attests isolation and never qualifies a run. Counter wrap/reset yields unavailable, not a negative or zero delta. Per-filesystem physical writes are never fabricated from allocation counters, per-process I/O counters, or pool totals.
+- **Parent live observation** (bounded, read-only):
+  ```sh
+  # Identify the scope behind a scratch path:
+  python3 tools/benchmark-campaign/filesystem.py identify --path /srv/scratch
+  # Identify and take one cumulative sample:
+  python3 tools/benchmark-campaign/filesystem.py sample --path /srv/scratch
+  ```
diff --git a/tools/benchmark-campaign/evidence.py b/tools/benchmark-campaign/evidence.py
index 218d9d838..f94f14919 100644
--- a/tools/benchmark-campaign/evidence.py
+++ b/tools/benchmark-campaign/evidence.py
@@ -120,6 +120,9 @@ def _publish_result(result: JsonObject, output: Path) -> None:
     if sys.platform != "linux":
         raise ContractError("publication requires Linux O_TMPFILE support")
     at_empty_path = getattr(os, "AT_EMPTY_PATH", 0x1000)
+    # ``canonical_bytes`` would recurse into a hostile nest before any limit
+    # applies; validate the depth first so oversized input fails closed here.
+    _validate_json_depth(result, "result")
     payload = canonical_bytes(result) + b"\n"
     try:
         directory = os.open(output.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
diff --git a/tools/benchmark-campaign/filesystem.py b/tools/benchmark-campaign/filesystem.py
new file mode 100644
index 000000000..d22788519
--- /dev/null
+++ b/tools/benchmark-campaign/filesystem.py
@@ -0,0 +1,701 @@
+# pyright: strict
+"""Filesystem-aware Linux measurement boundary (one target filesystem per run).
+
+One owner identifies the real mount/filesystem/backing-counter scope for a
+scratch path and takes read-only cumulative write-counter samples. Four
+measurement classes stay separate and are never merged here:
+
+- apparent bytes (file lengths),
+- allocated file bytes (the existing Rust ``storage-footprint`` tool owns
+  that measurement; it is not duplicated here),
+- per-process I/O counters,
+- attributable host block-device writes (this module's only counter).
+
+Only a regular single block device behind the target filesystem can yield an
+attributable counter, and even then the counter is host-wide for that device:
+every writer to the device is included. This collector never attests
+isolation and never qualifies a run; the existing harness keeps its strict
+qualification gate unless run ownership is actually proved.
+
+Honest scope labeling:
+
+- ``regular_block_device``: mount resolves to one block device with a
+  readable kernel sectors-written counter. Available.
+- ``zfs_pool``: ZFS dataset. Pool-wide accounting cannot isolate one
+  filesystem's physical writes, so the scope is unavailable. Pool aggregate
+  commands are informational only and are never converted into a
+  per-filesystem value.
+- ``composite_anonymous``: overlay/tmpfs/ramfs/network/fuse/9p/virtiofs and
+  other anonymous (major 0) mounts with no per-filesystem block counter.
+  Unavailable.
+- ``ambiguous_multi_device``: btrfs or a stacked device over more than one
+  underlying device. The mapping cannot name a single backing counter, so the
+  scope is unavailable. Underlying devices are listed, never summed.
+- ``unavailable``: mount entry, counter file, permission, or parse failure.
+  Unavailable with an explicit reason, never zero.
+
+Read-only bound: this module only reads the mount table, the one chosen
+device's sysfs counter file, and local topology/queue attribute files listed
+in ``mechanisms_checked``. The ``/proc/diskstats`` sectors-written value and
+the ``shared_mounts`` list are recorded diagnostics alongside the sysfs
+counter, never cross-validated against it and never gating availability. It
+spawns no subprocess, writes nothing, and follows exactly one counter per
+scope (a partition's own counter, never the parent disk added to it, never
+slaves added together).
+
+Sector unit: the kernel block-layer counter unit is always 512 bytes,
+independent of the filesystem logical block size or the device physical
+block size. Queue/model attributes are recorded for context only.
+"""
+
+import argparse
+import json
+import re
+import stat
+import sys
+from pathlib import Path
+from typing import TypedDict
+
+type Json = str | int | float | bool | None | list[Json] | dict[str, Json]
+type JsonObject = dict[str, Json]
+
+SECTOR_BYTES = 512
+
+# Kernel block-layer counters are 64-bit unsigned; anything outside
+# ``0 <= v <= 2**64 - 1`` is not a counter value and must be rejected.
+U64_MAX = (1 << 64) - 1
+
+MOUNTINFO_DEFAULT = "/proc/self/mountinfo"
+DISKSTATS_DEFAULT = "/proc/diskstats"
+SYS_BASE_DEFAULT = Path("/sys/dev/block")
+
+UNATTRIBUTABLE_FSTYPES = frozenset(
+    {
+        "overlay",
+        "tmpfs",
+        "ramfs",
+        "nfs",
+        "nfs4",
+        "cifs",
+        "smb3",
+        "fuse",
+        "fuseblk",
+        "9p",
+        "virtiofs",
+        "ecryptfs",
+    }
+)
+
+QUEUE_KEYS = (
+    "rotational",
+    "logical_block_size",
+    "physical_block_size",
+    "hw_sector_size",
+    "write_cache",
+    "scheduler",
+    "nr_requests",
+)
+
+_MAJMIN_RE = re.compile(r"[0-9]+:[0-9]+")
+
+
+class MountInfo(TypedDict):
+    mount_id: str
+    parent_id: str
+    majmin: str
+    root: str
+    mountpoint: str
+    mount_options: str
+    fstype: str
+    source: str
+    super_options: str
+
+
+class CounterSample(TypedDict):
+    sysfs_sectors_written: int
+    writes_completed: int
+    in_flight: int
+    diskstats_sectors_written: int | None
+
+
+class ScopeDoc(TypedDict, total=False):
+    """Precise scope document; every field carries its own narrow type."""
+
+    scratch_realpath: str
+    counter_unit: str
+    counter_scope: str
+    mount: MountInfo | None
+    mechanisms_checked: list[str]
+    status: str
+    scope_kind: str
+    reason: str
+    mapping: str
+    majmin: str
+    name: str
+    sysfs_stat: str
+    partition: bool
+    parent_disk: str
+    stacked_slaves: list[str]
+    holders: list[str]
+    shared_mounts: list[str]
+    queue: dict[str, str | None]
+    model: str | None
+    notes: list[str]
+    sample_error: str
+
+
+def unescape_mountinfo(field: str) -> str:
+    """Decode kernel mountinfo octal escapes (space, newline, backslash)."""
+    raw = re.sub(
+        rb"\\([0-7]{3})",
+        lambda match: bytes([int(match.group(1), 8)]),
+        field.encode("utf-8", errors="surrogateescape"),
+    )
+    return raw.decode("utf-8", errors="surrogateescape")
+
+
+def _read_text(
+    path: Path | str, errors: str = "replace", newline: str | None = None
+) -> str | None:
+    # Path.open (not read_text) so ``newline`` is honored on Python 3.12,
+    # whose read_text lacks the argument; universal-newline translation
+    # would rewrite raw carriage returns inside mountinfo records.
+    try:
+        with Path(path).open(encoding="utf-8", errors=errors, newline=newline) as file:
+            return file.read()
+    except OSError:
+        return None
+
+
+def read_mountinfo(path: str | Path = MOUNTINFO_DEFAULT) -> list[MountInfo]:
+    """Parse the kernel mount table; unreadable input yields an empty list.
+
+    The table is ingested with surrogateescape so non-UTF8 mountpoint bytes
+    survive decoding, consistent with the os.fsdecode-derived scratch path
+    they are matched against. Replacement decoding would map distinct byte
+    sequences onto one placeholder and misroute child mounts to ancestors.
+    Records are read with newline translation disabled and split only on
+    "\n": the kernel escapes just space, tab, newline, and backslash, so
+    str.splitlines() would wrongly split records holding raw vertical-tab,
+    form-feed, or carriage-return bytes in mountpoints.
+    """
+    text = _read_text(path, errors="surrogateescape", newline="")
+    if text is None:
+        return []
+    mounts: list[MountInfo] = []
+    for line in text.split("\n"):
+        if " - " not in line:
+            continue
+        left, right = line.split(" - ", 1)
+        left_fields = left.split(" ")
+        right_fields = right.split(" ")
+        if len(left_fields) < 6 or len(right_fields) < 2:
+            continue
+        # The kernel separates fields with single spaces and escapes literal
+        # spaces inside mountpoints, so an empty field means a malformed or
+        # hand-edited line. Skipping it beats shifting every index and
+        # attributing the scratch path to the wrong mount (often ``/``).
+        if "" in left_fields or "" in right_fields:
+            continue
+        try:
+            entry: MountInfo = {
+                "mount_id": left_fields[0],
+                "parent_id": left_fields[1],
+                "majmin": left_fields[2],
+                "root": unescape_mountinfo(left_fields[3]),
+                "mountpoint": unescape_mountinfo(left_fields[4]),
+                "mount_options": left_fields[5],
+                "fstype": right_fields[0],
+                "source": unescape_mountinfo(right_fields[1]),
+                "super_options": right_fields[2] if len(right_fields) > 2 else "",
+            }
+        except ValueError:
+            continue
+        mounts.append(entry)
+    return mounts
+
+
+def pick_mount(real: str, mounts: list[MountInfo]) -> MountInfo | None:
+    """Longest-prefix mount containing ``real``; ``None`` when no entry does."""
+    best: MountInfo | None = None
+    for mount in mounts:
+        point = mount["mountpoint"]
+        inside = (
+            real == point or point == "/" or real.startswith(point.rstrip("/") + "/")
+        )
+        if not inside:
+            continue
+        if best is None or len(point) >= len(best["mountpoint"]):
+            best = mount
+    return best
+
+
+def _child_names(directory: Path) -> list[str] | None:
+    try:
+        return sorted(child.name for child in directory.iterdir())
+    except FileNotFoundError:
+        return []
+    except OSError:
+        return None
+
+
+def _read_stripped(path: Path) -> str | None:
+    text = _read_text(path)
+    return text.strip() if text is not None else None
+
+
+def _is_composite_fstype(fstype: str) -> bool:
+    if fstype in UNATTRIBUTABLE_FSTYPES:
+        return True
+    # Kernel reports some fuse mounts as "fuse.<subtype>"; they share the
+    # same no-per-filesystem-counter property as plain "fuse".
+    return fstype.split(".")[0] in UNATTRIBUTABLE_FSTYPES
+
+
+class CounterHandle:
+    """Read-only handle for one block device's cumulative sectors counter.
+
+    Exactly one device file is ever read per sample (the scope's own
+    ``stat``); parent-disk and underlying-device counters are listed for
+    context and never added in.
+    """
+
+    def __init__(
+        self,
+        majmin: str,
+        name: str,
+        stat_path: Path | str,
+        diskstats_path: Path | str = DISKSTATS_DEFAULT,
+    ) -> None:
+        self.majmin = majmin
+        self.name = name
+        self.sysfs_stat = Path(stat_path)
+        self.diskstats_path = Path(diskstats_path)
+        major, _, minor = majmin.partition(":")
+        self.major = major
+        self.minor = minor
+
+    def read(self) -> CounterSample:
+        """Take one cumulative sample; raise on missing/unparsable primary."""
+        fields = self.sysfs_stat.read_text(encoding="utf-8", errors="replace").split()
+        # /sys/dev/block/<maj:min>/stat layout (stable prefix across kernel
+        # versions with optional discard/flush suffix fields): index 4 is
+        # writes completed, index 6 is sectors written, index 8 is I/O in
+        # flight. The 512-byte unit is the kernel block-layer unit.
+        if len(fields) < 9:
+            raise ValueError(f"counter file has {len(fields)} fields, need >= 9")
+        sample: CounterSample = {
+            "sysfs_sectors_written": int(fields[6]),
+            "writes_completed": int(fields[4]),
+            "in_flight": int(fields[8]),
+            "diskstats_sectors_written": None,
+        }
+        if (
+            not 0 <= sample["sysfs_sectors_written"] <= U64_MAX
+            or not 0 <= sample["writes_completed"] <= U64_MAX
+            or not 0 <= sample["in_flight"] <= U64_MAX
+        ):
+            raise ValueError("counter file holds a value outside 0 <= v <= 2**64 - 1")
+        # Recorded diagnostic only: the diskstats value is stored alongside the
+        # sysfs counter and is never cross-validated against it.
+        text = _read_text(self.diskstats_path)
+        if text is not None:
+            for line in text.splitlines():
+                parts = line.split()
+                # /proc/diskstats field 10 (1-based) is sectors written.
+                if (
+                    len(parts) >= 10
+                    and parts[0] == self.major
+                    and parts[1] == self.minor
+                ):
+                    try:
+                        sectors = int(parts[9])
+                    except ValueError:
+                        break
+                    if 0 <= sectors <= U64_MAX:
+                        sample["diskstats_sectors_written"] = sectors
+                    break
+        return sample
+
+
+def try_read(handle: CounterHandle) -> CounterSample | None:
+    """Best-effort single sample; ``None`` when the counter cannot be read."""
+    try:
+        return handle.read()
+    except (OSError, ValueError, IndexError):
+        return None
+
+
+def _counter_int(value: Json) -> int | None:
+    if (
+        isinstance(value, bool)
+        or not isinstance(value, int)
+        or not 0 <= value <= U64_MAX
+    ):
+        return None
+    return value
+
+
+def delta_bytes(
+    before: CounterSample | JsonObject | None,
+    after: CounterSample | JsonObject | None,
+    key: str = "sysfs_sectors_written",
+) -> int | None:
+    """Byte delta between two cumulative samples, or ``None`` when unknown.
+
+    ``None`` covers every non-attributable case: missing sample, missing key,
+    non-integer value, and counter wrap/reset/recreation (``after < before``).
+    A wrap never reports a negative value and never synthesizes zero as a
+    measurement; callers treat ``None`` as unavailable.
+    """
+    if before is None or after is None:
+        return None
+    if not isinstance(before, dict) or not isinstance(after, dict):
+        return None
+    first = _counter_int(before.get(key))
+    second = _counter_int(after.get(key))
+    if first is None or second is None:
+        return None
+    if second < first:
+        return None
+    return (second - first) * SECTOR_BYTES
+
+
+def identify_scope(
+    scratch: Path | str,
+    override: str | None = None,
+    *,
+    mountinfo_path: str | Path = MOUNTINFO_DEFAULT,
+    sys_base: Path | str = SYS_BASE_DEFAULT,
+    diskstats_path: str | Path = DISKSTATS_DEFAULT,
+) -> tuple[CounterHandle | None, ScopeDoc]:
+    """Identify the backing-counter scope for one filesystem path.
+
+    Returns ``(handle, scope)`` where ``handle`` takes read-only samples
+    when ``scope["status"] == "available"`` and is ``None`` otherwise. The
+    scope document is JSON-serializable and carries no isolation verdict; the
+    harness keeps its own strict gate.
+    """
+    sys_base_path = Path(sys_base)
+    diskstats = Path(diskstats_path)
+    scratch_path = Path(scratch).resolve()
+    real = str(scratch_path)
+    try:
+        scratch_is_dir = stat.S_ISDIR(scratch_path.stat().st_mode)
+        scratch_reason: str | None = None
+    except OSError as error:
+        scratch_is_dir = False
+        scratch_reason = f"cannot stat scratch path {real}: {error}"
+    # ``Path.resolve()`` never touches the filesystem, so without this stat a
+    # removed scratch path would still resolve against its old mount and
+    # report ``available``. The scratch stat is itself a checked mechanism.
+    mechanisms: list[str] = [str(mountinfo_path), real]
+    if not scratch_is_dir:
+        if scratch_reason is None:
+            scratch_reason = f"scratch path {real} is not a directory"
+        return None, {
+            "scratch_realpath": real,
+            "counter_unit": "sectors of 512 bytes (kernel block-layer unit)",
+            "counter_scope": (
+                "host-visible block-device writes when available; "
+                "not flash-internal write amplification"
+            ),
+            "mount": None,
+            "mechanisms_checked": mechanisms,
+            "status": "unavailable",
+            "scope_kind": "unavailable",
+            "reason": scratch_reason,
+        }
+    info: ScopeDoc = {
+        "scratch_realpath": real,
+        "counter_unit": "sectors of 512 bytes (kernel block-layer unit)",
+        "counter_scope": (
+            "host-visible block-device writes when available; "
+            "not flash-internal write amplification"
+        ),
+        "mount": None,
+        "mechanisms_checked": mechanisms,
+    }
+    if _read_text(mountinfo_path, errors="surrogateescape", newline="") is None:
+        info.update(
+            status="unavailable",
+            scope_kind="unavailable",
+            reason=f"cannot read {mountinfo_path}: mount table is unavailable",
+        )
+        return None, info
+    mounts = read_mountinfo(mountinfo_path)
+    mount = pick_mount(real, mounts)
+    info["mount"] = mount
+    if mount is None:
+        info.update(
+            status="unavailable",
+            scope_kind="unavailable",
+            reason=(f"no {mountinfo_path} entry contains the scratch path"),
+        )
+        return None, info
+    if not _MAJMIN_RE.fullmatch(mount["majmin"]):
+        info.update(
+            status="unavailable",
+            scope_kind="unavailable",
+            reason=f"malformed MAJ:MIN {mount['majmin']!r} for {mount['mountpoint']}",
+        )
+        return None, info
+    if override is not None and not _MAJMIN_RE.fullmatch(override):
+        info.update(
+            status="unavailable",
+            scope_kind="unavailable",
+            mapping="operator_asserted",
+            reason=f"invalid --device override {override!r}: want MAJ:MIN",
+        )
+        return None, info
+    if override is not None and override != mount["majmin"]:
+        info.update(
+            status="unavailable",
+            scope_kind="unavailable",
+            mapping="operator_asserted",
+            reason=(
+                f"override {override} does not match the mount's {mount['majmin']} "
+                f"for {mount['mountpoint']}: no backing relationship between the "
+                "named device and this filesystem is established, and operator "
+                "assertion alone is not proof. Only the mount's own MAJ:MIN is "
+                "accepted, and only on the ordinary-device path."
+            ),
+        )
+        return None, info
+    chosen = override if override is not None else mount["majmin"]
+    info["mapping"] = (
+        "operator_asserted" if override is not None else "mountinfo_major_minor"
+    )
+    fstype = mount["fstype"]
+    if fstype == "zfs":
+        info.update(
+            status="unavailable",
+            scope_kind="zfs_pool",
+            reason=(
+                f"ZFS dataset {mount['source']} on {mount['mountpoint']}: pool-wide "
+                "accounting cannot isolate this filesystem's physical writes, and "
+                "an aggregate or underlying-device counter cannot be attributed "
+                "per filesystem. No pool total is converted into a per-filesystem "
+                "value here."
+            ),
+        )
+        return None, info
+    if fstype == "btrfs":
+        info.update(
+            status="unavailable",
+            scope_kind="ambiguous_multi_device",
+            reason=(
+                "btrfs spans one or more devices: a single-device counter cannot "
+                "be assumed, and per-device counters cannot be summed into a "
+                "filesystem value. No multi-device total is attributed."
+            ),
+        )
+        return None, info
+    if _is_composite_fstype(fstype) or int(chosen.split(":")[0]) == 0:
+        info.update(
+            status="unavailable",
+            scope_kind="composite_anonymous",
+            reason=(
+                f"{fstype} on anonymous device {mount['majmin']}: no per-filesystem "
+                "block counter maps to this mount, and an aggregate or "
+                "underlying-device counter cannot be isolated per filesystem. "
+                "This is reported as unavailable, never zero or an estimate."
+            ),
+        )
+        return None, info
+    sysdir = sys_base_path / chosen
+    stat_path = sysdir / "stat"
+    mechanisms.append(str(stat_path))
+    mechanisms.append(str(diskstats))
+    if not stat_path.exists():
+        info.update(
+            status="unavailable",
+            scope_kind="regular_block_device",
+            majmin=chosen,
+            sysfs_stat=str(stat_path),
+            reason=f"{stat_path} does not exist",
+        )
+        return None, info
+    try:
+        real_sys = sysdir.resolve()
+    except OSError as error:
+        info.update(
+            status="unavailable",
+            scope_kind="regular_block_device",
+            majmin=chosen,
+            sysfs_stat=str(stat_path),
+            reason=f"cannot resolve {sysdir}: {error}",
+        )
+        return None, info
+    slaves_dir = real_sys / "slaves"
+    holders_dir = real_sys / "holders"
+    mechanisms.append(str(slaves_dir))
+    mechanisms.append(str(holders_dir))
+    slaves = _child_names(slaves_dir)
+    holders = _child_names(holders_dir)
+    if slaves is None or holders is None:
+        blocked = slaves_dir if slaves is None else holders_dir
+        info.update(
+            status="unavailable",
+            scope_kind="unavailable",
+            majmin=chosen,
+            sysfs_stat=str(stat_path),
+            reason=(
+                f"cannot list {blocked}: device topology is unreadable, "
+                "so stacking cannot be ruled out"
+            ),
+        )
+        return None, info
+    # Recorded diagnostic only: other mounts on this device are listed for
+    # context and never cross-validated against the counter.
+    shared = [
+        str(entry["mountpoint"])
+        for entry in mounts
+        if entry["majmin"] == chosen and entry is not mount
+    ]
+    partition = (real_sys / "partition").exists()
+    disk_dir = real_sys.parent if partition else real_sys
+    if partition:
+        # A partition of a stacked device (md/dm partition) lists no slaves of
+        # its own; the parent disk's topology decides whether it is stacked.
+        # Only the partition's own counter is still the one read.
+        parent_slaves_dir = disk_dir / "slaves"
+        mechanisms.append(str(parent_slaves_dir))
+        parent_slaves = _child_names(parent_slaves_dir)
+        if parent_slaves is None:
+            info.update(
+                status="unavailable",
+                scope_kind="unavailable",
+                majmin=chosen,
+                sysfs_stat=str(stat_path),
+                reason=(
+                    f"cannot list {parent_slaves_dir}: device topology is "
+                    "unreadable, so stacking cannot be ruled out"
+                ),
+            )
+            return None, info
+        slaves = slaves + [name for name in parent_slaves if name not in slaves]
+    queue_dir = disk_dir / "queue"
+    mechanisms.append(str(queue_dir))
+    model_path = disk_dir / "device" / "model"
+    mechanisms.append(str(model_path))
+    if slaves:
+        info.update(
+            status="unavailable",
+            scope_kind="ambiguous_multi_device",
+            majmin=chosen,
+            name=real_sys.name,
+            sysfs_stat=str(stat_path),
+            partition=partition,
+            parent_disk=disk_dir.name,
+            stacked_slaves=slaves,
+            holders=holders,
+            shared_mounts=shared,
+            reason=(
+                f"stacked device over {slaves}: a mapped device is never assumed "
+                "to be a single disk, and per-device counters are never summed "
+                "into one filesystem value. Mapping is ambiguous by design."
+            ),
+        )
+        return None, info
+    handle = CounterHandle(chosen, real_sys.name, stat_path, diskstats)
+    try:
+        handle.read()
+    except (OSError, ValueError, IndexError) as error:
+        info.update(
+            status="unavailable",
+            scope_kind="regular_block_device",
+            majmin=chosen,
+            name=real_sys.name,
+            sysfs_stat=str(stat_path),
+            partition=partition,
+            parent_disk=disk_dir.name,
+            stacked_slaves=slaves,
+            holders=holders,
+            shared_mounts=shared,
+            reason=f"cannot parse {stat_path}: {error}",
+        )
+        return None, info
+    queue: dict[str, str | None] = {}
+    for key in QUEUE_KEYS:
+        queue[key] = _read_stripped(queue_dir / key)
+    notes: list[str] = [
+        "the counter is host-wide for this block device: every writer to it is included"
+    ]
+    if holders:
+        notes.append(f"devices {holders} are stacked on this one")
+    if shared:
+        notes.append(f"other mounts share this device: {shared}")
+    if override is not None:
+        notes.append("device mapping asserted by the operator (--device)")
+    info.update(
+        status="available",
+        scope_kind="regular_block_device",
+        majmin=chosen,
+        name=real_sys.name,
+        sysfs_stat=str(stat_path),
+        partition=partition,
+        parent_disk=disk_dir.name,
+        stacked_slaves=slaves,
+        holders=holders,
+        shared_mounts=shared,
+        queue=queue,
+        model=_read_stripped(model_path),
+        notes=notes,
+    )
+    return handle, info
+
+
+def _scope_json(scope: ScopeDoc) -> str:
+    return json.dumps(scope, indent=2, sort_keys=True)
+
+
+def _parser() -> argparse.ArgumentParser:
+    top = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    sub = top.add_subparsers(dest="command", required=True)
+    for name in ("identify", "sample"):
+        cmd = sub.add_parser(
+            name,
+            help=(
+                "identify the backing-counter scope"
+                if name == "identify"
+                else "identify the scope and take one read-only sample"
+            ),
+        )
+        cmd.add_argument(
+            "--path", required=True, help="directory on the filesystem under test"
+        )
+        cmd.add_argument(
+            "--device",
+            help="MAJ:MIN override, must equal the mount's own MAJ:MIN; a different device is rejected",
+        )
+        cmd.add_argument("--mountinfo", default=MOUNTINFO_DEFAULT)
+        cmd.add_argument("--sys-base", default=str(SYS_BASE_DEFAULT))
+        cmd.add_argument("--diskstats", default=DISKSTATS_DEFAULT)
+    return top
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Bounded read-only CLI for parent live observation."""
+    args = _parser().parse_args(argv)
+    handle, scope = identify_scope(
+        args.path,
+        args.device,
+        mountinfo_path=args.mountinfo,
+        sys_base=args.sys_base,
+        diskstats_path=args.diskstats,
+    )
+    if args.command == "identify":
+        print(_scope_json(scope))
+        return 0
+    sample = try_read(handle) if handle is not None else None
+    if sample is None and handle is not None:
+        scope["sample_error"] = f"cannot parse {handle.sysfs_stat}"
+    print(json.dumps({"scope": scope, "sample": sample}, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tools/benchmark-campaign/test_evidence.py b/tools/benchmark-campaign/test_evidence.py
new file mode 100644
index 000000000..ef983f6e1
--- /dev/null
+++ b/tools/benchmark-campaign/test_evidence.py
@@ -0,0 +1,32 @@
+# pyright: strict
+"""Contract tests for evidence publication validation."""
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from evidence import ContractError, _publish_result, canonical_bytes
+
+
+class PublishDepthTests(unittest.TestCase):
+    def test_over_deep_result_is_rejected(self) -> None:
+        nested: object = {"leaf": 1}
+        for _ in range(40):
+            nested = [nested]
+        with tempfile.TemporaryDirectory() as tmp:
+            with self.assertRaises(ContractError):
+                _publish_result({"result": nested}, Path(tmp) / "out.json")
+
+    def test_shallow_result_publishes_atomically(self) -> None:
+        result: dict[str, object] = {"schema": "test-v1", "samples": 3}
+        with tempfile.TemporaryDirectory() as tmp:
+            output = Path(tmp) / "out.json"
+            _publish_result(result, output)
+            self.assertEqual(output.read_bytes(), canonical_bytes(result) + b"\n")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tools/benchmark-campaign/test_filesystem.py b/tools/benchmark-campaign/test_filesystem.py
new file mode 100644
index 000000000..c234383ac
--- /dev/null
+++ b/tools/benchmark-campaign/test_filesystem.py
@@ -0,0 +1,840 @@
+# pyright: strict
+"""Contract tests for the filesystem-aware Linux measurement boundary."""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+import filesystem
+from filesystem import (
+    CounterHandle,
+    delta_bytes,
+    identify_scope,
+    pick_mount,
+    read_mountinfo,
+    unescape_mountinfo,
+)
+
+STAT_FIELDS = ["0"] * 17
+
+
+class MountSpec:
+    def __init__(
+        self,
+        mount_id: str = "1",
+        majmin: str = "8:1",
+        point: str = "/",
+        fstype: str = "ext4",
+        source: str = "/dev/sda1",
+    ) -> None:
+        self.mount_id = mount_id
+        self.majmin = majmin
+        self.point = point
+        self.fstype = fstype
+        self.source = source
+
+
+def _stat_text(sectors: int, writes: int = 10, inflight: int = 0) -> str:
+    fields = list(STAT_FIELDS)
+    fields[4] = str(writes)
+    fields[6] = str(sectors)
+    fields[8] = str(inflight)
+    return " ".join(fields) + "\n"
+
+
+def _mountline(spec: MountSpec) -> str:
+    return (
+        f"{spec.mount_id} 0 {spec.majmin} / {spec.point} rw,relatime - "
+        f"{spec.fstype} {spec.source} rw"
+    )
+
+
+def _write_mountinfo(path: Path, lines: list[str]) -> None:
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+class DeviceFixture:
+    def __init__(self, stat_text: str = "") -> None:
+        self.stat_text = stat_text
+        self.slaves: list[str] = []
+        self.holders: list[str] = []
+        self.partition = False
+
+
+def _make_device(sys_base: Path, majmin: str, fixture: DeviceFixture) -> Path:
+    devdir = sys_base / majmin
+    devdir.mkdir(parents=True, exist_ok=True)
+    (devdir / "stat").write_text(fixture.stat_text, encoding="utf-8")
+    for dirname, entries in (("slaves", fixture.slaves), ("holders", fixture.holders)):
+        sub = devdir / dirname
+        sub.mkdir(exist_ok=True)
+        for entry in entries:
+            (sub / entry).write_text("", encoding="utf-8")
+    if fixture.partition:
+        (devdir / "partition").write_text("", encoding="utf-8")
+    queue = devdir / "queue"
+    queue.mkdir(exist_ok=True)
+    (queue / "rotational").write_text("1\n", encoding="utf-8")
+    return devdir / "stat"
+
+
+def _ext4_device(sectors: int) -> DeviceFixture:
+    return DeviceFixture(_stat_text(sectors))
+
+
+def _standard_mount(majmin: str, point: str) -> MountSpec:
+    source = "/dev/" + majmin.replace(":", "_")
+    return MountSpec(majmin=majmin, point=point, source=source)
+
+
+def _standard_table(root: Path, spec: MountSpec) -> Path:
+    table = root / "mountinfo"
+    _write_mountinfo(table, [_mountline(spec)])
+    return table
+
+
+def _real_scratch(root: Path, name: str = "scratch") -> Path:
+    """Create a real scratch directory; ``identify_scope`` stats it."""
+    scratch = root / name
+    scratch.mkdir(parents=True, exist_ok=True)
+    return scratch
+
+
+def _mount_dict(mount_id: str, majmin: str, point: str) -> filesystem.MountInfo:
+    return {
+        "mount_id": mount_id,
+        "parent_id": "0",
+        "majmin": majmin,
+        "root": "/",
+        "mountpoint": point,
+        "mount_options": "rw",
+        "fstype": "ext4",
+        "source": "/dev/sda1",
+        "super_options": "rw",
+    }
+
+
+class MountBoundaryTests(unittest.TestCase):
+    def test_longest_prefix_wins(self) -> None:
+        mounts = [
+            _mount_dict("1", "8:1", "/"),
+            _mount_dict("2", "8:2", "/data"),
+            _mount_dict("3", "8:3", "/data/sub"),
+        ]
+        picked = pick_mount("/data/sub/x", mounts)
+        assert picked is not None
+        self.assertEqual(picked["majmin"], "8:3")
+        picked = pick_mount("/data/other", mounts)
+        assert picked is not None
+        self.assertEqual(picked["majmin"], "8:2")
+
+    def test_sibling_prefix_does_not_match(self) -> None:
+        mounts = [_mount_dict("1", "8:1", "/"), _mount_dict("2", "8:2", "/data")]
+        picked = pick_mount("/data2/x", mounts)
+        assert picked is not None
+        self.assertEqual(picked["majmin"], "8:1")
+
+    def test_exact_mountpoint_matches(self) -> None:
+        picked = pick_mount("/srv", [_mount_dict("7", "8:9", "/srv")])
+        assert picked is not None
+        self.assertEqual(picked["majmin"], "8:9")
+
+    def test_no_table_entry_gives_none(self) -> None:
+        self.assertIsNone(pick_mount("/nothing/here", []))
+
+    def test_nonutf8_child_mount_maps_to_child(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            # Surrogateescaped names stay ``str``; ``pathlib`` syscalls
+            # re-encode them to the original non-UTF8 bytes.
+            raw_child = root / os.fsdecode(b"\xffraw")
+            raw_child.mkdir()
+            (raw_child / "scratch").mkdir()
+            child = str(raw_child)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:2", _ext4_device(11))
+            table = root / "mountinfo"
+            _write_mountinfo(
+                table,
+                [
+                    _mountline(MountSpec(majmin="8:1", point="/", source="/dev/sda1")),
+                    _mountline(
+                        MountSpec(
+                            majmin="8:2",
+                            point=str(root) + r"/\377raw",
+                            source="/dev/sdb1",
+                        )
+                    ),
+                ],
+            )
+            handle, scope = identify_scope(
+                child + "/scratch", mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertEqual(scope["status"], "available")
+            self.assertEqual(scope["majmin"], "8:2")
+            mount = scope.get("mount")
+            assert isinstance(mount, dict)
+            self.assertEqual(mount.get("mountpoint"), child)
+            assert handle is not None
+            self.assertEqual(handle.read()["sysfs_sectors_written"], 11)
+
+
+class EscapeTests(unittest.TestCase):
+    def test_octal_escapes_decode(self) -> None:
+        self.assertEqual(unescape_mountinfo(r"/mnt/my\040dir"), "/mnt/my dir")
+        self.assertEqual(unescape_mountinfo(r"/mnt/a\012b"), "/mnt/a\nb")
+        self.assertEqual(unescape_mountinfo(r"/mnt/a\134b"), r"/mnt/a\b")
+
+    def test_mountinfo_escaped_mountpoint(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            table = Path(tmp) / "mountinfo"
+            spec = MountSpec(point=r"/mnt/my\040dir")
+            _write_mountinfo(table, [_mountline(spec)])
+            mounts = read_mountinfo(table)
+        self.assertEqual(len(mounts), 1)
+        self.assertEqual(mounts[0]["mountpoint"], "/mnt/my dir")
+
+    def test_mountinfo_splits_only_on_newline(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            table = Path(tmp) / "mountinfo"
+            table.write_text(
+                "1 0 8:1 / /data\x0braw rw,relatime - ext4 /dev/sda1 rw\n"
+                "2 0 8:2 / /other\x0craw rw,relatime - ext4 /dev/sda2 rw\n"
+                "3 0 8:3 / /car\riageraw rw,relatime - ext4 /dev/sda3 rw\n",
+                encoding="utf-8",
+            )
+            mounts = read_mountinfo(table)
+        self.assertEqual(
+            [mount["mountpoint"] for mount in mounts],
+            ["/data\x0braw", "/other\x0craw", "/car\riageraw"],
+        )
+
+    def test_mountinfo_keeps_carriage_return_without_read_text_newline(
+        self,
+    ) -> None:
+        # Python 3.12 Path.read_text has no ``newline`` argument; the reader
+        # must still disable newline translation so a raw CR survives.
+        with tempfile.TemporaryDirectory() as tmp:
+            table = Path(tmp) / "mountinfo"
+            table.write_bytes(
+                b"1 0 8:1 / / rw - ext4 /dev/sda1 rw\n"
+                b"2 1 8:2 / /car\riage rw - ext4 /dev/sda2 rw\n"
+            )
+
+            def _no_newline(
+                self: Path, encoding: str | None = None, errors: str | None = None
+            ) -> str:
+                # 3.12 read_text always applies universal-newline translation.
+                text = self.read_bytes().decode(encoding or "utf-8", errors or "strict")
+                return text.replace("\r\n", "\n").replace("\r", "\n")
+
+            with mock.patch.object(Path, "read_text", _no_newline):
+                mounts = read_mountinfo(table)
+        self.assertEqual(
+            [mount["mountpoint"] for mount in mounts], ["/", "/car\riage"]
+        )
+
+    def test_mountinfo_malformed_lines_skipped(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            table = Path(tmp) / "mountinfo"
+            table.write_text("garbage without separator\n", encoding="utf-8")
+            self.assertEqual(read_mountinfo(table), [])
+
+    def test_mountinfo_missing_file_is_empty(self) -> None:
+        self.assertEqual(read_mountinfo("/nonexistent-mountinfo-xyz"), [])
+
+    def test_invalid_octal_escape_skips_entry(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(4))
+            table = root / "mountinfo"
+            _write_mountinfo(
+                table,
+                [
+                    _mountline(
+                        MountSpec(
+                            mount_id="9",
+                            majmin="8:9",
+                            point=r"/bad\777dir",
+                            source="/dev/sdz1",
+                        )
+                    ),
+                    _mountline(MountSpec(majmin="8:1", point="/", source="/dev/sda1")),
+                ],
+            )
+            mounts = read_mountinfo(table)
+            self.assertEqual([mount["majmin"] for mount in mounts], ["8:1"])
+            _handle, scope = identify_scope(
+                "/", mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertEqual(scope["status"], "available")
+            self.assertEqual(scope["majmin"], "8:1")
+
+
+class CounterTests(unittest.TestCase):
+    def test_available_regular_device(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "259:2", _ext4_device(1000))
+            diskstats = root / "diskstats"
+            diskstats.write_text(
+                "259   2 nvme0n1p2 0 0 0 0 10 0 1000 0 0 0 0 0 0 0\n",
+                encoding="utf-8",
+            )
+            table = _standard_table(root, _standard_mount("259:2", "/"))
+            scratch = _real_scratch(root)
+            handle, scope = identify_scope(
+                scratch,
+                mountinfo_path=table,
+                sys_base=sys_base,
+                diskstats_path=diskstats,
+            )
+            self.assertEqual(scope["status"], "available")
+            self.assertEqual(scope["scope_kind"], "regular_block_device")
+            self.assertIsNotNone(handle)
+            assert handle is not None
+            sample = handle.read()
+            self.assertEqual(sample["sysfs_sectors_written"], 1000)
+            self.assertEqual(sample["diskstats_sectors_written"], 1000)
+            self.assertIn("mechanisms_checked", scope)
+            self.assertNotIn("isolation", scope)
+
+    def test_diskstats_missing_keeps_sysfs(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(42))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root),
+                mountinfo_path=table,
+                sys_base=sys_base,
+                diskstats_path=root / "no-diskstats",
+            )
+            self.assertEqual(scope["status"], "available")
+            assert handle is not None
+            self.assertIsNone(handle.read()["diskstats_sectors_written"])
+
+    def test_stat_missing_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root), mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("reason", scope)
+
+    def test_stat_unparsable_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", DeviceFixture("not a counter file\n"))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root), mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+
+    def test_short_stat_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", DeviceFixture("1 2 3\n"))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root), mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+
+
+class ScopeLabelTests(unittest.TestCase):
+    def _scope_for(
+        self, fstype: str, majmin: str, override: str | None = None
+    ) -> tuple[CounterHandle | None, filesystem.ScopeDoc]:
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        root = Path(tmp.name)
+        sys_base = root / "sys"
+        sys_base.mkdir()
+        _make_device(sys_base, majmin, _ext4_device(5))
+        spec = MountSpec(majmin=majmin, source="pool/dataset", fstype=fstype)
+        table = _standard_table(root, spec)
+        return identify_scope("/", override, mountinfo_path=table, sys_base=sys_base)
+
+    def test_zfs_pool_scope(self) -> None:
+        handle, scope = self._scope_for("zfs", "0:42")
+        self.assertIsNone(handle)
+        self.assertEqual(scope["status"], "unavailable")
+        self.assertEqual(scope["scope_kind"], "zfs_pool")
+        self.assertIn("pool", str(scope.get("reason", "")))
+
+    def test_zfs_matching_override_stays_unavailable(self) -> None:
+        handle, scope = self._scope_for("zfs", "0:42", override="0:42")
+        self.assertIsNone(handle)
+        self.assertEqual(scope["status"], "unavailable")
+        self.assertEqual(scope["scope_kind"], "zfs_pool")
+
+    def test_composite_anonymous_scope(self) -> None:
+        for fstype in ("overlay", "tmpfs", "nfs"):
+            handle, scope = self._scope_for(fstype, "0:33")
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertEqual(scope["scope_kind"], "composite_anonymous")
+
+    def test_btrfs_is_ambiguous(self) -> None:
+        handle, scope = self._scope_for("btrfs", "8:5")
+        self.assertIsNone(handle)
+        self.assertEqual(scope["status"], "unavailable")
+        self.assertEqual(scope["scope_kind"], "ambiguous_multi_device")
+
+    def test_multi_slave_is_ambiguous(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            fixture = DeviceFixture(_stat_text(77))
+            fixture.slaves = ["8:1", "8:2"]
+            _make_device(sys_base, "253:0", fixture)
+            spec = MountSpec(majmin="253:0", source="/dev/dm-0")
+            table = _standard_table(root, spec)
+            handle, scope = identify_scope("/", mountinfo_path=table, sys_base=sys_base)
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertEqual(scope["scope_kind"], "ambiguous_multi_device")
+
+    def test_single_slave_is_ambiguous(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            fixture = DeviceFixture(_stat_text(77))
+            fixture.slaves = ["8:1"]
+            _make_device(sys_base, "253:0", fixture)
+            spec = MountSpec(majmin="253:0", source="/dev/dm-0")
+            table = _standard_table(root, spec)
+            handle, scope = identify_scope("/", mountinfo_path=table, sys_base=sys_base)
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertEqual(scope["scope_kind"], "ambiguous_multi_device")
+
+    def test_no_mount_entry_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            table = root / "mountinfo"
+            table.write_text("", encoding="utf-8")
+            handle, scope = identify_scope(
+                _real_scratch(root), mountinfo_path=table, sys_base=root / "sys"
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+
+    def test_bad_override_format_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                "/", "bogus", mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+
+    def test_matching_override_stays_available(self) -> None:
+        handle, scope = self._scope_for("ext4", "8:1", override="8:1")
+        self.assertEqual(scope["status"], "available")
+        self.assertEqual(scope["mapping"], "operator_asserted")
+        assert handle is not None
+        self.assertEqual(handle.read()["sysfs_sectors_written"], 5)
+
+    def test_mismatched_override_is_rejected(self) -> None:
+        handle, scope = self._scope_for("ext4", "8:1", override="8:2")
+        self.assertIsNone(handle)
+        self.assertEqual(scope["status"], "unavailable")
+        self.assertIn("override", str(scope.get("reason", "")))
+
+    def test_zfs_mismatched_override_is_rejected(self) -> None:
+        handle, scope = self._scope_for("zfs", "0:42", override="8:1")
+        self.assertIsNone(handle)
+        self.assertEqual(scope["status"], "unavailable")
+        self.assertIn("override", str(scope.get("reason", "")))
+
+    def test_permission_denied_stat_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(9))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            real_read = Path.read_text
+
+            def _denied(
+                self: Path,
+                encoding: str | None = None,
+                errors: str | None = None,
+                newline: str | None = None,
+            ) -> str:
+                if self.name == "stat":
+                    raise PermissionError(13, "Permission denied")
+                return real_read(
+                    self, encoding=encoding, errors=errors, newline=newline
+                )
+
+            with mock.patch.object(Path, "read_text", _denied):
+                handle, scope = identify_scope(
+                    "/", mountinfo_path=table, sys_base=sys_base
+                )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("reason", scope)
+
+    def test_unreadable_topology_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(9))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            real_iterdir = Path.iterdir
+
+            def _denied(self: Path) -> object:
+                if self.name == "slaves":
+                    raise PermissionError(13, "Permission denied")
+                return real_iterdir(self)
+
+            with mock.patch.object(Path, "iterdir", _denied):
+                handle, scope = identify_scope(
+                    "/", mountinfo_path=table, sys_base=sys_base
+                )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("slaves", str(scope.get("reason", "")))
+
+    def test_unreadable_mountinfo_names_cause(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            real_open = Path.open
+
+            def _denied(self: Path, *args: object, **kwargs: object) -> object:
+                if self == table:
+                    raise PermissionError(13, "Permission denied")
+                return real_open(self, *args, **kwargs)  # type: ignore[arg-type]
+
+            with mock.patch.object(Path, "open", _denied):
+                handle, scope = identify_scope(
+                    "/", mountinfo_path=table, sys_base=sys_base
+                )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("cannot read", str(scope.get("reason", "")))
+
+
+class NoDoubleCountTests(unittest.TestCase):
+    def test_partition_counter_used_alone(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            (sys_base / "259:0").mkdir(parents=True)
+            (sys_base / "259:0" / "stat").write_text(
+                _stat_text(99999), encoding="utf-8"
+            )
+            fixture = DeviceFixture(_stat_text(100))
+            fixture.partition = True
+            _make_device(sys_base, "259:2", fixture)
+            diskstats = root / "diskstats"
+            diskstats.write_text(
+                "259   0 nvme0n1 0 0 0 0 0 0 99999 0 0 0 0 0 0 0\n"
+                "259   2 nvme0n1p2 0 0 0 0 9 0 100 0 0 0 0 0 0 0\n",
+                encoding="utf-8",
+            )
+            table = _standard_table(root, _standard_mount("259:2", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root),
+                mountinfo_path=table,
+                sys_base=sys_base,
+                diskstats_path=diskstats,
+            )
+            assert handle is not None
+            self.assertEqual(scope["status"], "available")
+            self.assertEqual(scope["scope_kind"], "regular_block_device")
+            self.assertEqual(scope["partition"], True)
+            sample = handle.read()
+            self.assertEqual(sample["sysfs_sectors_written"], 100)
+            after: filesystem.CounterSample = {
+                "sysfs_sectors_written": 110,
+                "writes_completed": sample["writes_completed"],
+                "in_flight": sample["in_flight"],
+                "diskstats_sectors_written": sample["diskstats_sectors_written"],
+            }
+            self.assertEqual(delta_bytes(sample, after), 10 * filesystem.SECTOR_BYTES)
+
+    def test_malformed_mount_majmin_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            for majmin in ("\u00b2:1", "\u0663:1", "8", "../8:1"):
+                table = _standard_table(root, _standard_mount(majmin, "/"))
+                handle, scope = identify_scope(
+                    "/", mountinfo_path=table, sys_base=sys_base
+                )
+                self.assertIsNone(handle)
+                self.assertEqual(scope["status"], "unavailable")
+                self.assertIn("malformed", str(scope.get("reason", "")))
+
+    def test_partition_of_stacked_device_is_ambiguous(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            disk = root / "devices" / "md0"
+            (disk / "slaves" / "sda").mkdir(parents=True)
+            (disk / "slaves" / "sdb").mkdir()
+            part = disk / "md0p1"
+            (part / "slaves").mkdir(parents=True)
+            (part / "holders").mkdir()
+            (part / "partition").write_text("1\n", encoding="utf-8")
+            (part / "stat").write_text(_stat_text(100), encoding="utf-8")
+            sys_base.mkdir()
+            (sys_base / "259:1").symlink_to(part)
+            table = _standard_table(root, _standard_mount("259:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root), mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertEqual(scope["scope_kind"], "ambiguous_multi_device")
+            self.assertEqual(scope["stacked_slaves"], ["sda", "sdb"])
+
+
+class DeltaTests(unittest.TestCase):
+    def _sample(self, sectors: int) -> filesystem.CounterSample:
+        return {
+            "sysfs_sectors_written": sectors,
+            "writes_completed": 9,
+            "in_flight": 0,
+            "diskstats_sectors_written": sectors,
+        }
+
+    def test_normal_delta_scales_by_sector(self) -> None:
+        before = self._sample(100)
+        after = self._sample(110)
+        self.assertEqual(delta_bytes(before, after), 10 * 512)
+
+    def test_wrap_reset_gives_none(self) -> None:
+        self.assertIsNone(delta_bytes(self._sample(500), self._sample(12)))
+
+    def test_missing_inputs_give_none(self) -> None:
+        self.assertIsNone(delta_bytes(None, None))
+        before = self._sample(5)
+        after: filesystem.CounterSample = {
+            "sysfs_sectors_written": before["sysfs_sectors_written"],
+            "writes_completed": before["writes_completed"],
+            "in_flight": before["in_flight"],
+            "diskstats_sectors_written": None,
+        }
+        self.assertIsNone(delta_bytes(before, after, "diskstats_sectors_written"))
+        self.assertIsNone(delta_bytes(before, before, "no_such_key"))
+
+    def test_equal_counters_give_zero(self) -> None:
+        sample = self._sample(7)
+        self.assertEqual(delta_bytes(sample, self._sample(7)), 0)
+
+    def test_cli_identify_and_sample(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(3))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            base_args = ["--mountinfo", str(table), "--sys-base", str(sys_base)]
+            self.assertEqual(
+                filesystem.main(["identify", "--path", "/"] + base_args), 0
+            )
+            self.assertEqual(filesystem.main(["sample", "--path", "/"] + base_args), 0)
+
+    def test_scope_json_serializable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(3))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            _, scope = identify_scope("/", mountinfo_path=table, sys_base=sys_base)
+        json.dumps(scope)
+
+
+class ScratchValidationTests(unittest.TestCase):
+    def _available_fixture(
+        self, root: Path
+    ) -> tuple[Path, Path, Path]:
+        sys_base = root / "sys"
+        sys_base.mkdir()
+        _make_device(sys_base, "8:1", _ext4_device(5))
+        table = _standard_table(root, _standard_mount("8:1", "/"))
+        return sys_base, table, _real_scratch(root)
+
+    def test_removed_scratch_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base, table, scratch = self._available_fixture(root)
+            scratch.rmdir()
+            handle, scope = identify_scope(
+                scratch, mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("cannot stat", str(scope.get("reason", "")))
+
+    def test_file_scratch_is_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(5))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            candidate = root / "file-scratch"
+            candidate.write_text("not a directory", encoding="utf-8")
+            handle, scope = identify_scope(
+                candidate, mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("not a directory", str(scope.get("reason", "")))
+
+    def test_mechanisms_include_scratch_stat(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base, table, scratch = self._available_fixture(root)
+            handle, scope = identify_scope(
+                scratch, mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertEqual(scope["status"], "available")
+            assert handle is not None
+            mechanisms = scope.get("mechanisms_checked")
+            assert isinstance(mechanisms, list)
+            self.assertIn(str(scratch.resolve()), mechanisms)
+
+
+class MountParseStrictnessTests(unittest.TestCase):
+    def test_consecutive_spaces_line_skipped(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            table = Path(tmp) / "mountinfo"
+            table.write_text(
+                "1 0 8:1 / / rw,relatime - ext4 /dev/sda1 rw\n"
+                "2  0 8:2 / /data rw,relatime - ext4 /dev/sdb1 rw\n",
+                encoding="utf-8",
+            )
+            mounts = read_mountinfo(table)
+            self.assertEqual([mount["majmin"] for mount in mounts], ["8:1"])
+
+    def test_consecutive_spaces_do_not_misattribute_scratch(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            scratch = _real_scratch(root)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(5))
+            table = root / "mountinfo"
+            _write_mountinfo(
+                table,
+                [
+                    _mountline(MountSpec(majmin="8:1", point="/", source="/dev/sda1")),
+                    f"9  0 8:9 / {scratch} rw,relatime - ext4 /dev/sdz1 rw",
+                ],
+            )
+            handle, scope = identify_scope(
+                scratch, mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertEqual(scope["status"], "available")
+            self.assertEqual(scope["majmin"], "8:1")
+
+
+class CounterBoundTests(unittest.TestCase):
+    def test_huge_sysfs_counter_is_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", DeviceFixture(_stat_text(1 << 64)))
+            handle = CounterHandle(
+                "8:1", "sda", sys_base / "8:1" / "stat", root / "no-diskstats"
+            )
+            with self.assertRaises(ValueError):
+                handle.read()
+
+    def test_huge_counter_makes_scope_unavailable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", DeviceFixture(_stat_text(1 << 64)))
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root), mountinfo_path=table, sys_base=sys_base
+            )
+            self.assertIsNone(handle)
+            self.assertEqual(scope["status"], "unavailable")
+            self.assertIn("2**64", str(scope.get("reason", "")))
+
+    def test_huge_diskstats_value_is_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sys_base = root / "sys"
+            sys_base.mkdir()
+            _make_device(sys_base, "8:1", _ext4_device(42))
+            diskstats = root / "diskstats"
+            diskstats.write_text(
+                f"8 1 sda 0 0 0 0 0 0 {1 << 64} 0 0 0 0 0 0\n", encoding="utf-8"
+            )
+            table = _standard_table(root, _standard_mount("8:1", "/"))
+            handle, scope = identify_scope(
+                _real_scratch(root),
+                mountinfo_path=table,
+                sys_base=sys_base,
+                diskstats_path=diskstats,
+            )
+            self.assertEqual(scope["status"], "available")
+            assert handle is not None
+            self.assertIsNone(handle.read()["diskstats_sectors_written"])
+
+    def test_delta_rejects_out_of_range_counters(self) -> None:
+        before: filesystem.CounterSample = {
+            "sysfs_sectors_written": 1 << 64,
+            "writes_completed": 9,
+            "in_flight": 0,
+            "diskstats_sectors_written": None,
+        }
+        after: filesystem.CounterSample = {
+            "sysfs_sectors_written": (1 << 64) + 10,
+            "writes_completed": 9,
+            "in_flight": 0,
+            "diskstats_sectors_written": None,
+        }
+        self.assertIsNone(delta_bytes(before, after))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tools/storage-footprint/src/linux.rs b/tools/storage-footprint/src/linux.rs
index 2bd051111..089e2621a 100644
--- a/tools/storage-footprint/src/linux.rs
+++ b/tools/storage-footprint/src/linux.rs
@@ -45,6 +45,10 @@ pub(super) struct Report {
     namespaces: BTreeMap<String, Bytes>,
 }
 
+/// Datadirs are only a few levels deep; the cap keeps a mis-pointed root
+/// from recursing without bound and exhausting the stack.
+const MAX_DEPTH: u32 = 32;
+
 pub(super) fn measure(root: &Path) -> io::Result<Report> {
     let metadata = fs::symlink_metadata(root)?;
     if !metadata.is_dir() {
@@ -58,7 +62,7 @@ pub(super) fn measure(root: &Path) -> io::Result<Report> {
         namespaces: BTreeMap::new(),
     };
     let mut seen = HashSet::new();
-    visit(root, ".", metadata.dev(), mount, &mut seen, &mut report)?;
+    visit(root, ".", metadata.dev(), mount, &mut seen, &mut report, 0)?;
     Ok(report)
 }
 
@@ -85,6 +89,7 @@ fn visit(
     mount: u64,
     seen: &mut HashSet<(u64, u64)>,
     report: &mut Report,
+    depth: u32,
 ) -> io::Result<()> {
     let metadata = fs::symlink_metadata(path)?;
     if !(metadata.is_dir() || metadata.is_file())
@@ -109,10 +114,17 @@ fn visit(
         .or_default()
         .add(&metadata)?;
     if metadata.is_dir() {
-        // Stable order gives cross-namespace hard links a deterministic owner.
-        let mut entries = fs::read_dir(path)?.collect::<io::Result<Vec<_>>>()?;
-        entries.sort_by_key(fs::DirEntry::file_name);
-        for entry in entries {
+        if depth >= MAX_DEPTH {
+            return Err(io::Error::other(format!(
+                "directory nesting exceeds {MAX_DEPTH} at {}",
+                path.display()
+            )));
+        }
+        // Entries stream one at a time so a wide directory is never
+        // materialized in memory. Totals are order-independent; a
+        // cross-namespace hard link's bytes follow first-visit order.
+        for entry in fs::read_dir(path)? {
+            let entry = entry?;
             let name;
             let child_namespace = if namespace == "." {
                 // Each top-level entry is its own namespace, including loose files.
@@ -124,7 +136,15 @@ fn visit(
             } else {
                 namespace
             };
-            visit(&entry.path(), child_namespace, device, mount, seen, report)?;
+            visit(
+                &entry.path(),
+                child_namespace,
+                device,
+                mount,
+                seen,
+                report,
+                depth + 1,
+            )?;
         }
     }
     Ok(())
@@ -151,10 +171,42 @@ mod tests {
             report.totals.allocated_bytes,
             (root.blocks() + data.blocks()) * 512
         );
-        assert_eq!(report.namespaces["a"].apparent_bytes, data.len());
-        assert_eq!(report.namespaces["a"].allocated_bytes, data.blocks() * 512);
-        assert_eq!(report.namespaces["b"].apparent_bytes, 0);
-        assert_eq!(report.namespaces["b"].allocated_bytes, 0);
+        // Entries stream in readdir order, so either namespace may own the
+        // shared inode; exactly one of them must, and totals stay exact.
+        assert_eq!(
+            report.namespaces["a"].apparent_bytes + report.namespaces["b"].apparent_bytes,
+            data.len()
+        );
+        assert_eq!(
+            report.namespaces["a"].allocated_bytes + report.namespaces["b"].allocated_bytes,
+            data.blocks() * 512
+        );
+        assert!(
+            (report.namespaces["a"].apparent_bytes == data.len()
+                && report.namespaces["b"].apparent_bytes == 0)
+                || (report.namespaces["b"].apparent_bytes == data.len()
+                    && report.namespaces["a"].apparent_bytes == 0)
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn nesting_beyond_max_depth_is_refused() -> io::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let mut parent = dir.path().to_path_buf();
+        for level in 0..=MAX_DEPTH {
+            parent = parent.join(format!("level{level}"));
+            fs::create_dir(&parent)?;
+        }
+        fs::write(parent.join("data"), b"too deep")?;
+        let failure = match measure(dir.path()) {
+            Ok(_) => return Err(io::Error::other("deep nesting was accepted")),
+            Err(failure) => failure,
+        };
+        assert!(
+            failure.to_string().contains("nesting exceeds"),
+            "unexpected error: {failure}"
+        );
         Ok(())
     }