test(causal): controller test no longer races the sweep's site snapshot

run_experiments snapshots the site registry once at entry. stage-x is
registered lazily (first causal_site! execution in the worker), so on a
1-core box the snapshot deterministically wins whenever a sibling test
has already paid the tsc_hz calibration — stage-x missed the sweep and
the summary assert tripped (silently, pre-propagation; the earlier
cascade attribution was incomplete). The worker now signals after its
first site entry and the test waits on it before starting the sweep.
Jarred separately: the lazy-registration trap is a lib UX hazard worth a
doc note or warm-up guidance.

.gitignore

@@ -4,3 +4,4 @@ smarm_trace.json
 /bench_results/
 __pycache__/
 *.pyc
+profile.coz

Cargo.toml

@@ -19,6 +19,10 @@ expect_used = "deny"
 [features]
 default = ["rq-mutex"]
 smarm-trace = []
+# RFC 007: native causal profiling. Zero cost when off (cf. smarm-trace): the
+# hook in `maybe_preempt` and the resume-path fast-forward compile away; the
+# two Slot ledger fields exist regardless and stay 0 (budget_cycles precedent).
+smarm-causal = []
 # RFC 016 Chunk 2: cycle-accurate per-actor time-budget accounting. Off by
 # default — it costs two extra RDTSC reads per actor resume on the hot path
 # (D6). The `ActorInfo.budget_cycles` field exists regardless; it just stays 0
@@ -89,3 +93,15 @@ harness = false
 [[example]]
 name = "observer"
 required-features = ["observer"]
+
+[[example]]
+name = "causal_pipeline"
+required-features = ["smarm-causal"]
+
+[[example]]
+name = "causal_attrib_probe"
+required-features = ["smarm-causal"]
+
+[[example]]
+name = "causal_probe"
+required-features = ["smarm-causal"]

examples/causal_attrib_probe.rs

@@ -0,0 +1,198 @@
+//! Attribution-efficiency probe (RFC 007 follow-up).
+//!
+//! Original hypothesis: the ~4pt impact shortfall on the 24-core
+//! validation (+29.3/+83.5 vs theoretical +33/+100) is a constant
+//! attribution efficiency eff ≈ 0.91 from site-exit tail truncation.
+//! The guard-drop flush closed that leak, yet eff held at ~0.93 —
+//! RESOLVED (2026-07-13 sweep): the residual is runnable off-CPU time
+//! inside the site (~4.9 slice-expiry yields/entry x ~5.6µs runqueue
+//! wait), wall time the ground truth below counts but on-CPU
+//! attribution correctly skips. The offcpu audit bucket now counts it;
+//! `eff+offcpu` printed per window should sit at ~1.00 — the
+//! closed-books check.
+//!
+//! Measurement: same pipeline as `causal_pipeline`, but the `reserve`
+//! actor also measures its raw in-site time directly (rdtsc at guard
+//! enter/exit) and counts site entries. For each experiment window at
+//! pct%:
+//!
+//!     eff = (Δglobal_delay / (pct/100)) / Δin_site_cycles
+//!
+//! and the missing time per site entry localizes the leak:
+//!
+//!     tail_us/entry = (Δin_site − Δglobal_delay/(pct/100)) / Δentries
+//!
+//! A constant eff across 25/50% with tail/entry in the tens of µs
+//! localizes a per-entry mechanism; `eff+offcpu` ≈ 1.00 confirms the
+//! runnable-gap account and rules out any remaining silent loss.
+//!
+//! Run: cargo run --release --example causal_attrib_probe --features smarm-causal
+
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+fn rdtsc() -> u64 {
+    // x86_64 only — same clock the ledger uses.
+    unsafe { core::arch::x86_64::_rdtsc() }
+}
+
+/// Same fixed-work loop as causal_pipeline (dependent LCG, preemptible).
+fn work_iters(iters: u64) {
+    let mut acc = 0x2545_f491_4f6c_dd1du64;
+    let mut i = 0u64;
+    while i < iters {
+        let chunk_end = (i + 256).min(iters);
+        while i < chunk_end {
+            acc = acc.wrapping_mul(6364136223846793005).wrapping_add(i);
+            i += 1;
+        }
+        std::hint::black_box(acc);
+        smarm::check!();
+    }
+}
+
+fn calibrate_iters_per_us() -> u64 {
+    let n = 8_000_000u64;
+    let t = Instant::now();
+    work_iters(n);
+    (n / (t.elapsed().as_micros().max(1) as u64)).max(1)
+}
+
+static IN_SITE_CYCLES: AtomicU64 = AtomicU64::new(0);
+static SITE_ENTRIES: AtomicU64 = AtomicU64::new(0);
+
+fn main() {
+    let per_us = calibrate_iters_per_us();
+    println!("calibration: {per_us} work iters/µs");
+    let work_us = move |us: u64| work_iters(us * per_us);
+
+    let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
+    println!("cores: {cores}");
+    if cores < 4 {
+        println!("probe: SKIPPED (needs the stages in parallel)");
+        return;
+    }
+
+    smarm::init(smarm::Config::default()).run(move || {
+        let stop = Arc::new(AtomicBool::new(false));
+        let (tx_ab, rx_ab) = smarm::channel::<u64>();
+        let (tx_bc, rx_bc) = smarm::channel::<u64>();
+
+        let stop_p = stop.clone();
+        let producer = smarm::spawn(move || {
+            let mut i = 0u64;
+            while !stop_p.load(Ordering::Relaxed) {
+                {
+                    let _g = smarm::causal_site!("serialize");
+                    work_us(200);
+                }
+                if tx_ab.send(i).is_err() {
+                    break;
+                }
+                i += 1;
+            }
+        });
+
+        // Reserve: the target — instrumented with ground-truth in-site time.
+        let reserve = smarm::spawn(move || {
+            while let Ok(item) = rx_ab.recv() {
+                {
+                    let t0 = rdtsc();
+                    let _g = smarm::causal_site!("reserve");
+                    work_us(400);
+                    // Measured before guard drop: exactly the span the
+                    // ledger should be attributing.
+                    IN_SITE_CYCLES.fetch_add(rdtsc().saturating_sub(t0), Ordering::Relaxed);
+                    SITE_ENTRIES.fetch_add(1, Ordering::Relaxed);
+                }
+                if tx_bc.send(item).is_err() {
+                    break;
+                }
+            }
+        });
+
+        let notify = smarm::spawn(move || {
+            while rx_bc.recv().is_ok() {
+                {
+                    let _g = smarm::causal_site!("notify");
+                    work_us(50);
+                }
+                smarm::progress!("orders-processed");
+            }
+        });
+
+        let stop_bg = stop.clone();
+        let background = smarm::spawn(move || {
+            while !stop_bg.load(Ordering::Relaxed) {
+                let _g = smarm::causal_site!("background-compaction");
+                work_us(500);
+            }
+        });
+
+        smarm::sleep(Duration::from_millis(300));
+
+        let hz = smarm::causal::tsc_hz();
+        println!("tsc_hz: {:.3} GHz", hz / 1e9);
+
+        // Manual windows so ledger/ground-truth snapshots align exactly.
+        for &pct in &[25u32, 50, 50, 25] {
+            let g0 = smarm::causal::global_delay_cycles();
+            let s0 = IN_SITE_CYCLES.load(Ordering::Relaxed);
+            let e0 = SITE_ENTRIES.load(Ordering::Relaxed);
+            let a0 = smarm::causal::ledger_counters();
+            smarm::causal::begin_experiment_for_test("reserve", pct);
+            smarm::sleep(Duration::from_millis(1000));
+            smarm::causal::end_experiment_for_test();
+            let audit = smarm::causal::ledger_counters().delta_since(&a0);
+            let injected = smarm::causal::global_delay_cycles() - g0;
+            let in_site = IN_SITE_CYCLES.load(Ordering::Relaxed) - s0;
+            let entries = SITE_ENTRIES.load(Ordering::Relaxed) - e0;
+
+            let attributed = injected as f64 / (pct as f64 / 100.0);
+            let eff = attributed / in_site as f64;
+            // Books-closure check: add back the runnable off-CPU gaps the
+            // audit counted (delta terms -> raw via /pct) — should be ~1.00.
+            let eff_closed = (injected as f64 + audit.offcpu_in_site_cycles as f64)
+                / (pct as f64 / 100.0)
+                / in_site as f64;
+            let missing = in_site as f64 - attributed;
+            let tail_us = if entries > 0 {
+                missing / entries as f64 / hz * 1e6
+            } else {
+                f64::NAN
+            };
+            println!(
+                "pct {pct:>2}%  in_site {:>8.1}ms  attributed {:>8.1}ms  eff {eff:.3}  eff+offcpu {eff_closed:.3}  entries {entries}  missing/entry {tail_us:.1}µs",
+                in_site as f64 / hz * 1e3,
+                attributed / hz * 1e3,
+            );
+            // RFC 007 deficit hunt: name the losses. Drop/discard columns are
+            // in would-be delta terms — divide by pct/100 to compare with the
+            // missing attribution above.
+            let ms = |c: u64| c as f64 / hz * 1e3;
+            println!(
+                "        audit: absorbed {:>7.1}ms  forgiven {:>6.1}ms  drop park {:>5.2}ms/{:<5} yield {:>5.2}ms/{:<5} offcpu {:>6.2}ms/{:<5} discard >max {:>5.2}ms/{:<3} unarmed {}",
+                ms(audit.spin_absorbed_cycles),
+                ms(audit.park_forgiven_cycles),
+                ms(audit.drop_park_cycles),
+                audit.drop_park_n,
+                ms(audit.drop_yield_cycles),
+                audit.drop_yield_n,
+                ms(audit.offcpu_in_site_cycles),
+                audit.offcpu_in_site_n,
+                ms(audit.discard_overmax_cycles),
+                audit.discard_overmax_n,
+                audit.discard_unarmed_n
+            );
+            smarm::sleep(Duration::from_millis(150));
+        }
+
+        stop.store(true, Ordering::Relaxed);
+        producer.join().unwrap();
+        reserve.join().unwrap();
+        notify.join().unwrap();
+        background.join().unwrap();
+        println!("probe: DONE");
+    });
+}

examples/causal_pipeline.rs

@@ -0,0 +1,296 @@
+//! Causal-profiling demo (RFC 007): a pipeline where conventional profiling
+//! lies and causal profiling doesn't.
+//!
+//!   producer --(serialize ~200µs/item)--> reserve --(~400µs/item)--> notify
+//!   background: an actor burning CPU constantly, fully off the critical path
+//!
+//! `reserve` is the true bottleneck. `serialize` is hot but overlapped with
+//! `reserve`'s backlog, and `background` is the hottest code in the process
+//! while contributing nothing to throughput. A cycle profiler ranks them
+//! background > reserve ≈ 2×serialize; the causal report instead shows
+//! throughput responding to virtual speedups of `reserve` and (near-)ignoring
+//! `serialize` and `background`.
+//!
+//! Stage cost is fixed *work* (a calibrated arithmetic loop), not fixed wall
+//! time. This matters: a timed busy-wait absorbs injected causal delay into
+//! its own budget and finishes on schedule regardless, making every
+//! experiment read as a no-op (found live on a 24-core run: dead-flat
+//! deltas). Real workloads are work-shaped, so the demo must be too.
+//!
+//! Run:
+//!   cargo run --release --example causal_pipeline --features smarm-causal
+//!
+//! Modes (`SMARM_CAUSAL_MODE`), for probing what the guard placement leaves
+//! out of the measurement (a site speeds up only what it wraps; `recv`/`send`
+//! on the serialized stage sit outside the canonical guard):
+//!   work (default) — guard wraps only the 400µs of work.
+//!   wide           — guard widened over recv + work + send, the whole
+//!                    serialized per-item path.
+//!   occupancy      — no experiments; times each segment of reserve's loop
+//!                    at baseline and reports the unguarded per-item
+//!                    overhead δ plus the impact ceiling it implies.
+//!
+//! Result (24-core run, 2026-07-13, job f9305cbb): δ measured 0.3µs/item —
+//! 0.1% of the serialized path — and `wide` does not move the @50% cell
+//! (+83.5/+86.3 vs work's +81.5/+86.7). This demo's +84-vs-+100 @50%
+//! shortfall is therefore NOT unguarded stage time; it is controller-side:
+//! injected delay reaches ~327ms of the ideal 350ms over the 700ms window,
+//! plus a ~3% real-throughput dip while experiments run. Contrast urus's
+//! causal_bench, where the same arithmetic identified a real ~70µs/request
+//! unguarded remainder (recv/reply outside the store guard). Sites measure
+//! what they wrap — and the occupancy probe tells you which case you're in.
+//!
+//! Prints a summary, writes `profile.coz` (Coz plot-compatible), and — given
+//! enough cores for the pipeline to actually run in parallel — checks the
+//! expected separation and exits nonzero if it doesn't hold, so a CI box can
+//! run this as a smoke test.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+/// LCG-mix `iters` times in dependent sequence (unvectorizable, un-elidable),
+/// staying preemptible — and causal-sampleable/delayable — via `check!()`.
+fn work_iters(iters: u64) {
+    let mut acc = 0x2545_f491_4f6c_dd1du64;
+    let mut i = 0u64;
+    while i < iters {
+        let chunk_end = (i + 256).min(iters);
+        while i < chunk_end {
+            acc = acc.wrapping_mul(6364136223846793005).wrapping_add(i);
+            i += 1;
+        }
+        std::hint::black_box(acc);
+        smarm::check!();
+    }
+}
+
+/// Measure how many `work_iters` iterations fit in a microsecond on this
+/// machine, so stage costs below are meaningful in time while staying
+/// work-shaped.
+fn calibrate_iters_per_us() -> u64 {
+    let n = 8_000_000u64;
+    let t = Instant::now();
+    work_iters(n);
+    (n / (t.elapsed().as_micros().max(1) as u64)).max(1)
+}
+
+/// Guard placement for the `reserve` stage — see module doc.
+#[derive(Clone, Copy, PartialEq)]
+enum Mode {
+    Work,
+    Wide,
+    Occupancy,
+}
+
+fn main() {
+    let mode = match std::env::var("SMARM_CAUSAL_MODE").as_deref() {
+        Err(_) | Ok("") | Ok("work") => Mode::Work,
+        Ok("wide") => Mode::Wide,
+        Ok("occupancy") => Mode::Occupancy,
+        Ok(other) => {
+            eprintln!("unknown SMARM_CAUSAL_MODE {other:?} (work|wide|occupancy)");
+            std::process::exit(2);
+        }
+    };
+    println!(
+        "mode: {}",
+        match mode {
+            Mode::Work => "work",
+            Mode::Wide => "wide",
+            Mode::Occupancy => "occupancy",
+        }
+    );
+    let per_us = calibrate_iters_per_us();
+    println!("calibration: {per_us} work iters/µs");
+    let work_us = move |us: u64| work_iters(us * per_us);
+
+    let mut failures: Vec<String> = Vec::new();
+
+    smarm::init(smarm::Config::default()).run(move || {
+        let stop = Arc::new(AtomicBool::new(false));
+
+        let (tx_ab, rx_ab) = smarm::channel::<u64>();
+        let (tx_bc, rx_bc) = smarm::channel::<u64>();
+
+        // Producer: hot serialization, but upstream of the bottleneck.
+        let stop_p = stop.clone();
+        let producer = smarm::spawn(move || {
+            let mut i = 0u64;
+            while !stop_p.load(Ordering::Relaxed) {
+                {
+                    let _g = smarm::causal_site!("serialize");
+                    work_us(200);
+                }
+                if tx_ab.send(i).is_err() {
+                    break;
+                }
+                i += 1;
+            }
+            // tx_ab drops here; downstream drains and exits.
+        });
+
+        // Reserve: the true bottleneck (~400µs of work per item).
+        let reserve = smarm::spawn(move || match mode {
+            Mode::Work => {
+                while let Ok(item) = rx_ab.recv() {
+                    {
+                        let _g = smarm::causal_site!("reserve");
+                        work_us(400);
+                    }
+                    if tx_bc.send(item).is_err() {
+                        break;
+                    }
+                }
+            }
+            // Whole serialized per-item path under the guard: a virtual
+            // speedup now also compresses recv/send, so the @50% cell should
+            // recover the theoretical 2× that `work` mode's placement caps.
+            Mode::Wide => loop {
+                let _g = smarm::causal_site!("reserve");
+                let Ok(item) = rx_ab.recv() else { break };
+                work_us(400);
+                if tx_bc.send(item).is_err() {
+                    break;
+                }
+            },
+            // Time each segment at baseline; the recv+send remainder δ is
+            // the serialized time a `work`-placed guard cannot speed up.
+            Mode::Occupancy => {
+                let (mut recv_ns, mut work_ns, mut send_ns, mut n) = (0u64, 0u64, 0u64, 0u64);
+                loop {
+                    let t0 = Instant::now();
+                    let Ok(item) = rx_ab.recv() else { break };
+                    let t1 = Instant::now();
+                    {
+                        let _g = smarm::causal_site!("reserve");
+                        work_us(400);
+                    }
+                    let t2 = Instant::now();
+                    if tx_bc.send(item).is_err() {
+                        break;
+                    }
+                    recv_ns += (t1 - t0).as_nanos() as u64;
+                    work_ns += (t2 - t1).as_nanos() as u64;
+                    send_ns += t2.elapsed().as_nanos() as u64;
+                    n += 1;
+                }
+                let items = n.max(1) as f64;
+                let (r, w, s) = (
+                    recv_ns as f64 / items / 1e3,
+                    work_ns as f64 / items / 1e3,
+                    send_ns as f64 / items / 1e3,
+                );
+                let delta = r + s;
+                let total = w + delta;
+                println!("occupancy: {n} items; per item recv {r:.1}µs + work(guarded) {w:.1}µs + send {s:.1}µs");
+                println!(
+                    "occupancy: unguarded δ = {delta:.1}µs/item = {:.1}% of the serialized path",
+                    100.0 * delta / total
+                );
+                for pct in [25u32, 50] {
+                    let f = 1.0 - f64::from(pct) / 100.0;
+                    println!(
+                        "occupancy: predicted reserve impact @{pct}% -> {:+.1}%  (ceiling if δ were guarded: {:+.1}%)",
+                        100.0 * (total / (f * w + delta) - 1.0),
+                        100.0 * (1.0 / f - 1.0)
+                    );
+                }
+            }
+        });
+
+        // Notify: light tail stage; marks the unit of useful work.
+        let notify = smarm::spawn(move || {
+            while rx_bc.recv().is_ok() {
+                {
+                    let _g = smarm::causal_site!("notify");
+                    work_us(50);
+                }
+                smarm::progress!("orders-processed");
+            }
+        });
+
+        // Background: hottest code in the process, zero throughput relevance.
+        let stop_bg = stop.clone();
+        let background = smarm::spawn(move || {
+            while !stop_bg.load(Ordering::Relaxed) {
+                let _g = smarm::causal_site!("background-compaction");
+                work_us(500);
+            }
+        });
+
+        // Warm up so queues reach steady state before measuring.
+        smarm::sleep(Duration::from_millis(300));
+
+        if mode == Mode::Occupancy {
+            // No experiments: hold steady state for a window, then drain and
+            // let the reserve actor print its segment report.
+            smarm::sleep(Duration::from_millis(1500));
+            stop.store(true, Ordering::Relaxed);
+            producer.join().unwrap();
+            reserve.join().unwrap();
+            notify.join().unwrap();
+            background.join().unwrap();
+            return;
+        }
+
+        let results = smarm::causal::run_experiments(&smarm::causal::ExperimentPlan {
+            speedups_pct: vec![0, 25, 50],
+            experiment: Duration::from_millis(700),
+            cooldown: Duration::from_millis(150),
+        });
+
+        stop.store(true, Ordering::Relaxed);
+        producer.join().unwrap();
+        reserve.join().unwrap();
+        notify.join().unwrap();
+        background.join().unwrap();
+
+        print!("{}", smarm::causal::render_summary(&results));
+        // RFC 007 deficit hunt: SMARM_CAUSAL_AUDIT=1 appends the per-cell
+        // ledger audit (injected/absorbed/forgiven + drop and discard
+        // buckets) without touching the pinned summary format.
+        if std::env::var_os("SMARM_CAUSAL_AUDIT").is_some() {
+            print!("{}", smarm::causal::render_ledger_audit(&results));
+        }
+        let coz = smarm::causal::render_coz(&results);
+        match std::fs::write("profile.coz", coz) {
+            Ok(()) => println!("\nwrote profile.coz"),
+            Err(e) => eprintln!("\nfailed to write profile.coz: {e}"),
+        }
+
+        // Verdict. The separation only exists when the four pipeline actors
+        // actually run in parallel; on a small box, report and skip.
+        let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
+        if cores < 4 {
+            println!("verdict: SKIPPED ({cores} cores; separation needs the stages in parallel)");
+            return;
+        }
+        let impact = |site: &str| {
+            smarm::causal::impact_pct(&results, site, 25, "orders-processed")
+        };
+        let mut expect = |site: &str, ok: &dyn Fn(f64) -> bool, want: &str| match impact(site) {
+            Some(p) => {
+                let verdict = if ok(p) { "ok" } else { "FAIL" };
+                println!("verdict: {site} @25% -> {p:+.1}% (want {want}) {verdict}");
+                if !ok(p) {
+                    failures.push(format!("{site}: {p:+.1}% (want {want})"));
+                }
+            }
+            None => {
+                println!("verdict: {site} @25% -> missing cell FAIL");
+                failures.push(format!("{site}: missing cell"));
+            }
+        };
+        expect("reserve", &|p| p > 15.0, "> +15%");
+        expect("serialize", &|p| p < 10.0, "< +10%");
+        expect("background-compaction", &|p| p < 10.0, "< +10%");
+
+        if failures.is_empty() {
+            println!("verdict: PASS — causal separation holds");
+        } else {
+            println!("verdict: FAIL — {}", failures.join("; "));
+            std::process::exit(1);
+        }
+    });
+}

examples/causal_probe.rs

@@ -0,0 +1,145 @@
+//! Diagnostic probe for RFC 007 on a target box. Measures, in order:
+//!  1. TSC frequency against `Instant` (the crate assumes 3 GHz).
+//!  2. TSC sanity under actor migration: distribution of wall time actually
+//!     spent in `burn_us(400)` across many runs — a bimodal/short tail means
+//!     cross-core TSC offsets are cutting burns short.
+//!  3. Pipeline stage rates with no experiment running (who is the real
+//!     bottleneck?).
+//!  4. The same rates during a 50% experiment on `background-compaction`
+//!     (a correct implementation must slow every stage; an off-critical-path
+//!     target must reduce end-to-end throughput proportionally).
+//!
+//! Run: cargo run --release --example causal_probe --features smarm-causal
+
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+fn burn_us(us: u64) {
+    let cycles = us * 3_000;
+    let start = smarm::preempt::rdtsc();
+    while smarm::preempt::rdtsc().saturating_sub(start) < cycles {
+        smarm::check!();
+    }
+}
+
+fn main() {
+    // 1. TSC calibration (plain OS thread, before the runtime starts).
+    let c0 = smarm::preempt::rdtsc();
+    let t0 = Instant::now();
+    std::thread::sleep(Duration::from_millis(200));
+    let hz = (smarm::preempt::rdtsc() - c0) as f64 / t0.elapsed().as_secs_f64();
+    println!("tsc_hz: {:.3e} (crate assumes 3.0e9)", hz);
+
+    smarm::init(smarm::Config::default()).run(move || {
+        // 2. burn_us(400) wall-time distribution inside a migrating actor.
+        let h = smarm::spawn(|| {
+            let mut samples: Vec<u64> = (0..500)
+                .map(|_| {
+                    let t = Instant::now();
+                    burn_us(400);
+                    t.elapsed().as_micros() as u64
+                })
+                .collect();
+            samples.sort_unstable();
+            println!(
+                "burn_us(400) wall us: min {} p10 {} p50 {} p90 {} max {}",
+                samples[0], samples[50], samples[250], samples[450], samples[499]
+            );
+        });
+        h.join().unwrap();
+
+        // 3+4. Pipeline with per-stage counters.
+        let stop = Arc::new(AtomicBool::new(false));
+        let produced = Arc::new(AtomicU64::new(0));
+        let reserved = Arc::new(AtomicU64::new(0));
+        let notified = Arc::new(AtomicU64::new(0));
+
+        let (tx_ab, rx_ab) = smarm::channel::<u64>();
+        let (tx_bc, rx_bc) = smarm::channel::<u64>();
+
+        let stop_p = stop.clone();
+        let produced2 = produced.clone();
+        let producer = smarm::spawn(move || {
+            let mut i = 0u64;
+            while !stop_p.load(Ordering::Relaxed) {
+                {
+                    let _g = smarm::causal_site!("serialize");
+                    burn_us(200);
+                }
+                if tx_ab.send(i).is_err() {
+                    break;
+                }
+                produced2.fetch_add(1, Ordering::Relaxed);
+                i += 1;
+            }
+        });
+
+        let reserved2 = reserved.clone();
+        let reserve = smarm::spawn(move || {
+            while let Ok(item) = rx_ab.recv() {
+                {
+                    let _g = smarm::causal_site!("reserve");
+                    burn_us(400);
+                }
+                reserved2.fetch_add(1, Ordering::Relaxed);
+                if tx_bc.send(item).is_err() {
+                    break;
+                }
+            }
+        });
+
+        let notified2 = notified.clone();
+        let notify = smarm::spawn(move || {
+            while rx_bc.recv().is_ok() {
+                {
+                    let _g = smarm::causal_site!("notify");
+                    burn_us(50);
+                }
+                notified2.fetch_add(1, Ordering::Relaxed);
+                smarm::progress!("orders-processed");
+            }
+        });
+
+        let stop_bg = stop.clone();
+        let background = smarm::spawn(move || {
+            while !stop_bg.load(Ordering::Relaxed) {
+                let _g = smarm::causal_site!("background-compaction");
+                burn_us(500);
+            }
+        });
+
+        smarm::sleep(Duration::from_millis(300));
+
+        let window = |label: &str| {
+            let (p0, r0, n0) = (
+                produced.load(Ordering::Relaxed),
+                reserved.load(Ordering::Relaxed),
+                notified.load(Ordering::Relaxed),
+            );
+            let d0 = smarm::causal::global_delay_cycles();
+            let t = Instant::now();
+            smarm::sleep(Duration::from_millis(700));
+            let secs = t.elapsed().as_secs_f64();
+            println!(
+                "{label}: produced {:.0}/s reserved {:.0}/s notified {:.0}/s injected {:.0}ms(assumed-3GHz)",
+                (produced.load(Ordering::Relaxed) - p0) as f64 / secs,
+                (reserved.load(Ordering::Relaxed) - r0) as f64 / secs,
+                (notified.load(Ordering::Relaxed) - n0) as f64 / secs,
+                (smarm::causal::global_delay_cycles() - d0) as f64 / 3.0e9 * 1e3,
+            );
+        };
+
+        window("no-experiment ");
+        smarm::causal::begin_experiment_for_test("background-compaction", 50);
+        window("bg-comp @ 50% ");
+        smarm::causal::end_experiment_for_test();
+        window("post-experiment");
+
+        stop.store(true, Ordering::Relaxed);
+        producer.join().unwrap();
+        reserve.join().unwrap();
+        notify.join().unwrap();
+        background.join().unwrap();
+    });
+}

src/causal.rs

@@ -0,0 +1,959 @@
+//! Native causal profiling (RFC 007). Enabled by `--features smarm-causal`;
+//! zero cost without it (same discipline as `smarm-trace`).
+//!
+//! The Coz algorithm, transposed onto actors: to estimate what speeding up
+//! code site S by p% would do to throughput, we instead *slow everything
+//! else down* by p% of the time spent in S, and watch the progress-point
+//! rates respond. Where Coz must inject real `usleep`s into OS threads from
+//! the outside, smarm owns every clock that matters:
+//!
+//! - Sampling and delay injection happen at `maybe_preempt`'s amortised
+//!   cadence — an existing, safe hook (never inside a prep-to-park region).
+//! - Injected delay is subtracted from the actor's timeslice
+//!   (`preempt::extend_timeslice`), so experiments don't perturb scheduling.
+//! - Delay bookkeeping is *actor*-granular: each `Slot` carries an absorbed-
+//!   delay ledger, compared against a global ledger. Parked actors absorb
+//!   accrued delay for free on resume (Coz's blocked-thread rule) — waiting
+//!   is never penalised.
+//!
+//! v1 scope (per RFC discussion): explicit scoped sites (`causal_site!`)
+//! rather than PC sampling (jar Q1 stays open); throughput progress points
+//! only; timer-heap deadlines are *not* shifted (documented gap — long
+//! experiments can make real-time timeouts fire early in virtual terms);
+//! multi-scheduler coherence is best-effort via global atomics.
+//!
+//! Usage:
+//! ```ignore
+//! let _g = smarm::causal_site!("inventory-reserve");   // in suspect code
+//! smarm::progress!("orders-processed");                // per unit of work
+//! let results = smarm::causal::run_experiments(&Default::default());
+//! print!("{}", smarm::causal::render_summary(&results));
+//! std::fs::write("profile.coz", smarm::causal::render_coz(&results))?;
+//! ```
+
+#[cfg(feature = "smarm-causal")]
+mod inner {
+    use crate::preempt;
+    use std::cell::Cell;
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::sync::{Mutex, OnceLock};
+    use std::time::{Duration, Instant};
+
+    // -----------------------------------------------------------------------
+    // Global state
+    // -----------------------------------------------------------------------
+
+    /// Active experiment, packed `(site_id << 32) | speedup_pct`. 0 = idle.
+    /// A single word so the hot path reads one atomic; experiments are global
+    /// across scheduler threads (jar Q7, v1: plain Relaxed atomics).
+    static EXPERIMENT: AtomicU64 = AtomicU64::new(0);
+
+    /// Monotone experiment-window counter, bumped by every `begin()`. Lets
+    /// the offcpu-gap stash (RFC 007) tell apart two windows with an
+    /// identical site+pct word — live in the attrib probe's 50,50 schedule
+    /// — so a gap straddling `end()`/`begin()` never counts a cooldown.
+    static EXPERIMENT_EPOCH: AtomicU64 = AtomicU64::new(0);
+
+    /// Global virtual-delay ledger, in TSC cycles: the total delay every
+    /// actor *should* have experienced since startup. Grows while a sample
+    /// lands in the experiment's target site; each actor's `Slot` ledger
+    /// chases it by spin-absorbing at preemption checks.
+    static GLOBAL_DELAY: AtomicU64 = AtomicU64::new(0);
+
+    /// Registered site names; site id = index + 1 (0 = "no site").
+    static SITES: OnceLock<Mutex<Vec<&'static str>>> = OnceLock::new();
+
+    /// Registered progress points (leaked for `'static`, like trace's drain
+    /// state — the set is small and lives for the process).
+    static PROGRESS: OnceLock<Mutex<Vec<&'static ProgressPoint>>> = OnceLock::new();
+
+    thread_local! {
+        /// TSC at this thread's previous causal check, the sample "period"
+        /// denominator. Re-armed on every actor resume so scheduler time and
+        /// a previous actor's tail never count toward a sample. 0 = unarmed.
+        static LAST_SAMPLE_TSC: Cell<u64> = const { Cell::new(0) };
+    }
+
+    /// Guard against TSC weirdness (migration between unsynced sockets,
+    /// virtualisation steps): a single sample interval larger than this is
+    /// discarded rather than believed. ~33ms at 3 GHz — far beyond any real
+    /// gap between preemption checks inside a slice.
+    const MAX_SAMPLE_CYCLES: u64 = 100_000_000;
+
+    /// Cap on delay spun in one visit, so one check can never wedge an actor
+    /// for a human-visible pause; the remainder is absorbed on later visits.
+    /// ~3ms at 3 GHz.
+    const MAX_SPIN_PER_VISIT: u64 = 10_000_000;
+
+    // -----------------------------------------------------------------------
+    // Ledger audit (RFC 007 deficit hunt): where injected delay is born,
+    // paid, and forgiven — and where would-be attribution is silently lost
+    // (deschedule tails, clamp discards). Monotone Relaxed totals, read via
+    // `ledger_counters()`; `run_experiments` windows them into
+    // `ExperimentResult`. Measure-only: nothing here changes injection or
+    // absorption behaviour.
+    // -----------------------------------------------------------------------
+
+    /// Cycles bystanders actually spun to pay down the global ledger.
+    static SPIN_ABSORBED_CYCLES: AtomicU64 = AtomicU64::new(0);
+    /// Cycles waived at wake after a real park (the blocked-thread rule).
+    static PARK_FORGIVEN_CYCLES: AtomicU64 = AtomicU64::new(0);
+    /// Would-be attribution lost when the target-site actor parks mid-site.
+    static DROP_PARK_CYCLES: AtomicU64 = AtomicU64::new(0);
+    static DROP_PARK_N: AtomicU64 = AtomicU64::new(0);
+    /// Same loss at yields (explicit, slice-expiry, or a park that requeued).
+    static DROP_YIELD_CYCLES: AtomicU64 = AtomicU64::new(0);
+    static DROP_YIELD_N: AtomicU64 = AtomicU64::new(0);
+    /// Samples discarded by the TSC-weirdness clamp, in would-be delta terms.
+    static DISCARD_OVERMAX_CYCLES: AtomicU64 = AtomicU64::new(0);
+    static DISCARD_OVERMAX_N: AtomicU64 = AtomicU64::new(0);
+    /// In-site samples dropped because the thread's clock was unarmed.
+    static DISCARD_UNARMED_N: AtomicU64 = AtomicU64::new(0);
+    /// Would-be attribution over runnable off-CPU gaps inside the target
+    /// site (yield-descheduled -> resumed within the same window). Not a
+    /// loss: on-CPU-only attribution is the Coz model — queue-wait is not
+    /// shrunk by speeding the site's code — but counted so the audit books
+    /// close against wall in-site time (the located @50 "deficit").
+    static OFFCPU_IN_SITE_CYCLES: AtomicU64 = AtomicU64::new(0);
+    static OFFCPU_IN_SITE_N: AtomicU64 = AtomicU64::new(0);
+
+    fn sites() -> &'static Mutex<Vec<&'static str>> {
+        SITES.get_or_init(|| Mutex::new(Vec::new()))
+    }
+
+    fn progress_points() -> &'static Mutex<Vec<&'static ProgressPoint>> {
+        PROGRESS.get_or_init(|| Mutex::new(Vec::new()))
+    }
+
+    /// Recover from lock poisoning: all these registries hold plain data that
+    /// is valid at every instruction boundary, so a panicked registrant can't
+    /// leave them torn.
+    fn lock_unpoisoned<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
+        match m.lock() {
+            Ok(g) => g,
+            Err(poisoned) => poisoned.into_inner(),
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // Sites
+    // -----------------------------------------------------------------------
+
+    /// Register (or look up) a causal site by name; returns its nonzero id.
+    /// Called once per `causal_site!` expansion via a `OnceLock`, so the
+    /// mutex is off every hot path.
+    pub fn site_id(name: &'static str) -> u32 {
+        let mut v = lock_unpoisoned(sites());
+        if let Some(pos) = v.iter().position(|n| *n == name) {
+            return (pos + 1) as u32;
+        }
+        v.push(name);
+        v.len() as u32
+    }
+
+    fn site_name(id: u32) -> Option<String> {
+        if id == 0 {
+            return None;
+        }
+        let v = lock_unpoisoned(sites());
+        v.get((id - 1) as usize).map(|s| (*s).to_string())
+    }
+
+    /// RAII marker: while alive, the current *actor* (not thread — the id
+    /// lives in its `Slot` and survives preemption/migration) is "inside"
+    /// the site. Nesting restores the outer site on drop. Inert outside an
+    /// actor (scheduler/OS-thread stacks).
+    pub struct SiteGuard {
+        /// Slot of the actor that entered, null if entered outside an actor.
+        /// Valid for the guard's whole life: the guard lives on the actor's
+        /// stack, and a slot is never reclaimed while its actor is alive —
+        /// the same argument as `preempt::check_cancelled`.
+        slot: *const crate::runtime::Slot,
+        prev: u32,
+    }
+
+    impl SiteGuard {
+        /// Enter `site` for the on-CPU actor.
+        pub fn enter(site: u32) -> Self {
+            let slot = preempt::current_slot_ptr();
+            if slot.is_null() {
+                return SiteGuard { slot, prev: 0 };
+            }
+            // SAFETY: non-null ⇒ points at the on-CPU actor's slot; see the
+            // field docs for the lifetime argument.
+            let prev = unsafe { (*slot).causal_site() };
+            unsafe { (*slot).set_causal_site(site) };
+            site_transition(slot, prev, site);
+            SiteGuard { slot, prev }
+        }
+    }
+
+    impl Drop for SiteGuard {
+        fn drop(&mut self) {
+            if !self.slot.is_null() {
+                // SAFETY (both): as in `enter` — the actor (and thus its
+                // slot) is alive for as long as this guard is on its stack.
+                let site = unsafe { (*self.slot).causal_site() };
+                unsafe { (*self.slot).set_causal_site(self.prev) };
+                site_transition(self.slot, site, self.prev);
+            }
+        }
+    }
+
+    /// Name of the site the on-CPU actor is currently inside, if any.
+    /// (Introspection/testing; not a hot path.)
+    pub fn current_site_name() -> Option<String> {
+        let slot = preempt::current_slot_ptr();
+        if slot.is_null() {
+            return None;
+        }
+        // SAFETY: on-CPU actor's slot, valid for the whole resume.
+        site_name(unsafe { (*slot).causal_site() })
+    }
+
+    // -----------------------------------------------------------------------
+    // Progress points
+    // -----------------------------------------------------------------------
+
+    /// A named throughput counter. One per distinct name; `progress!` call
+    /// sites sharing a name share the counter.
+    pub struct ProgressPoint {
+        name: &'static str,
+        count: AtomicU64,
+    }
+
+    impl ProgressPoint {
+        /// The hot path: one Relaxed RMW. (Contended across actors by design
+        /// — a progress point is a global rate meter.)
+        #[inline]
+        pub fn bump(&self) {
+            self.count.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    /// Register (or look up) a progress point. Called once per `progress!`
+    /// expansion via a `OnceLock`; the mutex is off the hot path.
+    pub fn register_progress(name: &'static str) -> &'static ProgressPoint {
+        let mut v = lock_unpoisoned(progress_points());
+        if let Some(p) = v.iter().find(|p| p.name == name) {
+            return p;
+        }
+        let p: &'static ProgressPoint = Box::leak(Box::new(ProgressPoint {
+            name,
+            count: AtomicU64::new(0),
+        }));
+        v.push(p);
+        p
+    }
+
+    /// Snapshot of all progress points as `(name, count)`.
+    pub fn progress_snapshot() -> Vec<(String, u64)> {
+        lock_unpoisoned(progress_points())
+            .iter()
+            .map(|p| (p.name.to_string(), p.count.load(Ordering::Relaxed)))
+            .collect()
+    }
+
+    // -----------------------------------------------------------------------
+    // The hot hook: sample + absorb
+    // -----------------------------------------------------------------------
+
+    /// Called from `maybe_preempt` at the amortised timeslice-check cadence,
+    /// under the `PREEMPTION_ENABLED` gate (so never in a prep-to-park or
+    /// no-preempt region — spinning here is as safe as yielding is).
+    ///
+    /// One Relaxed load and out when no experiment is running.
+    #[inline]
+    pub(crate) fn check() {
+        let exp = EXPERIMENT.load(Ordering::Relaxed);
+        if exp == 0 {
+            return;
+        }
+        cold_check(exp);
+    }
+
+    /// The experiment-active path, kept out of the inlined fast path.
+    #[cold]
+    fn cold_check(exp: u64) {
+        let slot = preempt::current_slot_ptr();
+        if slot.is_null() {
+            return;
+        }
+        let now = preempt::rdtsc();
+        let last = LAST_SAMPLE_TSC.with(|c| c.replace(now));
+        let target_site = (exp >> 32) as u32;
+        let pct = exp & 0xffff_ffff;
+
+        // SAFETY (both derefs below): non-null ⇒ the on-CPU actor's slot,
+        // never reclaimed while the actor runs — see `check_cancelled`.
+        let my_site = unsafe { (*slot).causal_site() };
+
+        if my_site == target_site && pct > 0 {
+            // A sample landed in the target site: everyone else must fall
+            // behind by pct% of the sampled interval. Grow the global ledger
+            // and credit ourselves the same amount — the credited gap *is*
+            // the virtual speedup.
+            if last == 0 {
+                // Unarmed clock: no interval to attribute — count the loss.
+                DISCARD_UNARMED_N.fetch_add(1, Ordering::Relaxed);
+                return;
+            }
+            // SAFETY: `slot` is the on-CPU actor's slot (checked non-null
+            // above); see `check_cancelled` for the lifetime argument.
+            unsafe { attribute(slot, now.saturating_sub(last), pct) };
+        } else {
+            // Not the winner: chase the global ledger by spinning off the
+            // difference, then push the slice start forward so injected
+            // delay never counts as compute (the clock correction that Coz
+            // cannot do from outside).
+            let global = GLOBAL_DELAY.load(Ordering::Relaxed);
+            let mine = unsafe { (*slot).causal_delay() };
+            if mine >= global {
+                return;
+            }
+            let spin = (global - mine).min(MAX_SPIN_PER_VISIT);
+            let start = preempt::rdtsc();
+            while preempt::rdtsc().saturating_sub(start) < spin {
+                core::hint::spin_loop();
+            }
+            SPIN_ABSORBED_CYCLES.fetch_add(spin, Ordering::Relaxed);
+            unsafe { (*slot).set_causal_delay(mine.wrapping_add(spin)) };
+            preempt::extend_timeslice(spin);
+            // The spin is not part of the next sample interval either.
+            LAST_SAMPLE_TSC.with(|c| c.set(preempt::rdtsc()));
+        }
+    }
+
+    /// Attribute one target-site sample of `interval` cycles at `pct`%:
+    /// grow the global ledger and credit the sampling actor's own ledger by
+    /// the same amount — the credited gap *is* the virtual speedup. Shared
+    /// by the cold check and the guard-boundary flush. Applies the same
+    /// clamps as sampling always has: zero intervals and clock hiccups are
+    /// discarded, not the run.
+    ///
+    /// SAFETY: `slot` must point at the on-CPU actor's slot (the
+    /// `check_cancelled` lifetime argument).
+    unsafe fn attribute(slot: *const crate::runtime::Slot, interval: u64, pct: u64) {
+        if interval == 0 {
+            return; // now == last: nothing to attribute, nothing lost
+        }
+        if interval > MAX_SAMPLE_CYCLES {
+            // TSC-weirdness clamp: the sample is discarded, not the run.
+            // Count the loss in would-be delta terms so the audit's columns
+            // compare directly against `injected_cycles`.
+            DISCARD_OVERMAX_N.fetch_add(1, Ordering::Relaxed);
+            DISCARD_OVERMAX_CYCLES
+                .fetch_add(interval.saturating_mul(pct) / 100, Ordering::Relaxed);
+            return;
+        }
+        let delta = interval.saturating_mul(pct) / 100;
+        GLOBAL_DELAY.fetch_add(delta, Ordering::Relaxed);
+        let mine = (*slot).causal_delay();
+        (*slot).set_causal_delay(mine.wrapping_add(delta));
+    }
+
+    /// Site-boundary hook, called by `SiteGuard` enter/drop when the
+    /// actor's current site changes from `old` to `new`. Sample-only —
+    /// never spins — so it is safe anywhere, including no-preempt regions
+    /// where `check()` cannot run.
+    ///
+    /// - Leaving the experiment's target site: flush the pending interval.
+    ///   Cold checks only sample when they happen to fire in-site, so the
+    ///   tail between the last check and the guard drop was otherwise
+    ///   discarded on every site entry — measured live at ~22-29µs/entry,
+    ///   ~6-7% of all target time (eff 0.93), which under-reported every
+    ///   impact (+83.5% where theory says +100%).
+    /// - Entering the target site: re-arm the sample clock, so time spent
+    ///   *before* the site can never be attributed to it by the first
+    ///   in-site check (the symmetric over-attribution).
+    #[inline]
+    fn site_transition(slot: *const crate::runtime::Slot, old: u32, new: u32) {
+        let exp = EXPERIMENT.load(Ordering::Relaxed);
+        if exp == 0 || old == new {
+            return;
+        }
+        let target = (exp >> 32) as u32;
+        let pct = exp & 0xffff_ffff;
+        if old == target && new != target {
+            let now = preempt::rdtsc();
+            let last = LAST_SAMPLE_TSC.with(|c| c.replace(now));
+            if pct > 0 {
+                if last != 0 {
+                    // SAFETY: forwarded from the guard, which holds the on-CPU
+                    // actor's slot for its whole life (see `SiteGuard::slot`).
+                    unsafe { attribute(slot, now.saturating_sub(last), pct) };
+                } else {
+                    DISCARD_UNARMED_N.fetch_add(1, Ordering::Relaxed);
+                }
+            }
+        } else if new == target && old != target {
+            LAST_SAMPLE_TSC.with(|c| c.set(preempt::rdtsc()));
+        }
+    }
+
+    /// Resume-path hook (scheduler thread, actor off-CPU). Two duties:
+    ///
+    /// - If the last deschedule was a *real park*, time blocked absorbs any
+    ///   delay accrued meanwhile for free — Coz's blocked-thread rule, which
+    ///   keeps experiments from punishing actors for waiting. An actor that
+    ///   merely yielded (slice expiry) was runnable the whole time and keeps
+    ///   its debt: it must pay by spinning at its next check. Forgiving on
+    ///   every resume would make any yield-cadence actor delay-immune and
+    ///   experiments inert (found live on a 24-core run: nothing slowed).
+    /// - If the deschedule was a *yield* in the live experiment's target
+    ///   site, count the off-CPU gap it opened into the offcpu audit bucket
+    ///   (RFC 007: the located @50 deficit — runnable queue-wait is wall
+    ///   time in-site that on-CPU attribution correctly skips). Same-window
+    ///   only, enforced by the experiment epoch; measure-only.
+    /// - Arm this thread's sample clock so the first interval of the resume
+    ///   excludes scheduler time.
+    #[inline]
+    pub(crate) fn on_resume(slot: &crate::runtime::Slot) {
+        let (desched_tsc, desched_epoch) = slot.take_causal_desched();
+        if desched_tsc != 0 && desched_epoch == EXPERIMENT_EPOCH.load(Ordering::Relaxed) {
+            // Same epoch ⇒ no `begin()` since the stash; a nonzero word ⇒
+            // no `end()` either — the gap closed inside its own window.
+            let exp = EXPERIMENT.load(Ordering::Relaxed);
+            if exp != 0 {
+                let pct = exp & 0xffff_ffff;
+                let gap = preempt::rdtsc()
+                    .saturating_sub(desched_tsc)
+                    .min(MAX_SAMPLE_CYCLES);
+                OFFCPU_IN_SITE_CYCLES
+                    .fetch_add(gap.saturating_mul(pct) / 100, Ordering::Relaxed);
+                OFFCPU_IN_SITE_N.fetch_add(1, Ordering::Relaxed);
+            }
+        }
+        if slot.take_causal_parked() {
+            let global = GLOBAL_DELAY.load(Ordering::Relaxed);
+            let mine = slot.causal_delay();
+            if mine < global {
+                PARK_FORGIVEN_CYCLES.fetch_add(global - mine, Ordering::Relaxed);
+                slot.set_causal_delay(global);
+            }
+        }
+        LAST_SAMPLE_TSC.with(|c| c.set(preempt::rdtsc()));
+    }
+
+    /// Deschedule-path hook (scheduler side, same OS thread the actor just
+    /// ran on). If an experiment is live and the departing actor sits in the
+    /// target site, the sample tail `[last sample -> now]` is about to be
+    /// lost: nothing flushes it here, and `on_resume` re-arms the clock
+    /// before the actor runs again. Measure-only (RFC 007 deficit hunt) —
+    /// tally the would-be attribution into the park/yield drop buckets and
+    /// leave behaviour untouched. The interval is capped at
+    /// MAX_SAMPLE_CYCLES: past that the flush would have discarded it anyway
+    /// (counted separately). `now` includes the few hundred ns of scheduler
+    /// bookkeeping since the actor actually stopped — an acceptable
+    /// overcount for a diagnostic.
+    ///
+    /// Slice-expiry yields sample at the same checkpoint that deschedules
+    /// them, so their tails are ~zero by construction; a fat yield bucket
+    /// therefore points at explicit `yield_now` calls or requeued parks.
+    ///
+    /// Yields additionally stash the deschedule instant on the slot so
+    /// `on_resume` can count the runnable off-CPU gap (offcpu bucket).
+    pub(crate) fn on_deschedule(slot: &crate::runtime::Slot, real_park: bool) {
+        let exp = EXPERIMENT.load(Ordering::Relaxed);
+        if exp == 0 {
+            return;
+        }
+        let target = (exp >> 32) as u32;
+        let pct = exp & 0xffff_ffff;
+        if pct == 0 || slot.causal_site() != target {
+            return;
+        }
+        let now = preempt::rdtsc();
+        if !real_park {
+            // Runnable gap opens here; `on_resume` closes and counts it
+            // (offcpu bucket). Parks are excluded: blocked time is already
+            // represented by forgiveness, and blocked wall time is not
+            // queue-wait.
+            slot.set_causal_desched(now, EXPERIMENT_EPOCH.load(Ordering::Relaxed));
+        }
+        let last = LAST_SAMPLE_TSC.with(|c| c.get());
+        if last == 0 {
+            return;
+        }
+        let interval = now.saturating_sub(last).min(MAX_SAMPLE_CYCLES);
+        let would_be = interval.saturating_mul(pct) / 100;
+        if real_park {
+            DROP_PARK_N.fetch_add(1, Ordering::Relaxed);
+            DROP_PARK_CYCLES.fetch_add(would_be, Ordering::Relaxed);
+        } else {
+            DROP_YIELD_N.fetch_add(1, Ordering::Relaxed);
+            DROP_YIELD_CYCLES.fetch_add(would_be, Ordering::Relaxed);
+        }
+    }
+
+    // -----------------------------------------------------------------------
+    // Experiments
+    // -----------------------------------------------------------------------
+
+    fn begin(site: u32, pct: u32) {
+        EXPERIMENT_EPOCH.fetch_add(1, Ordering::Relaxed);
+        EXPERIMENT.store(((site as u64) << 32) | pct as u64, Ordering::Relaxed);
+    }
+
+    fn end() {
+        EXPERIMENT.store(0, Ordering::Relaxed);
+    }
+
+    /// Total virtual delay injected so far, in TSC cycles.
+    pub fn global_delay_cycles() -> u64 {
+        GLOBAL_DELAY.load(Ordering::Relaxed)
+    }
+
+    /// Cumulative ledger-audit totals since startup (RFC 007 deficit hunt).
+    /// All monotone; window a span by snapshotting before/after and taking
+    /// `delta_since`. Cycle fields are in would-be-injected delta terms so
+    /// they compare directly against `injected_cycles`.
+    #[derive(Clone, Copy, Debug, Default)]
+    pub struct LedgerCounters {
+        pub spin_absorbed_cycles: u64,
+        pub park_forgiven_cycles: u64,
+        pub drop_park_cycles: u64,
+        pub drop_park_n: u64,
+        pub drop_yield_cycles: u64,
+        pub drop_yield_n: u64,
+        pub discard_overmax_cycles: u64,
+        pub discard_overmax_n: u64,
+        pub discard_unarmed_n: u64,
+        pub offcpu_in_site_cycles: u64,
+        pub offcpu_in_site_n: u64,
+    }
+
+    impl LedgerCounters {
+        /// Field-wise difference against an earlier snapshot.
+        pub fn delta_since(&self, before: &LedgerCounters) -> LedgerCounters {
+            LedgerCounters {
+                spin_absorbed_cycles: self
+                    .spin_absorbed_cycles
+                    .saturating_sub(before.spin_absorbed_cycles),
+                park_forgiven_cycles: self
+                    .park_forgiven_cycles
+                    .saturating_sub(before.park_forgiven_cycles),
+                drop_park_cycles: self.drop_park_cycles.saturating_sub(before.drop_park_cycles),
+                drop_park_n: self.drop_park_n.saturating_sub(before.drop_park_n),
+                drop_yield_cycles: self
+                    .drop_yield_cycles
+                    .saturating_sub(before.drop_yield_cycles),
+                drop_yield_n: self.drop_yield_n.saturating_sub(before.drop_yield_n),
+                discard_overmax_cycles: self
+                    .discard_overmax_cycles
+                    .saturating_sub(before.discard_overmax_cycles),
+                discard_overmax_n: self.discard_overmax_n.saturating_sub(before.discard_overmax_n),
+                discard_unarmed_n: self.discard_unarmed_n.saturating_sub(before.discard_unarmed_n),
+                offcpu_in_site_cycles: self
+                    .offcpu_in_site_cycles
+                    .saturating_sub(before.offcpu_in_site_cycles),
+                offcpu_in_site_n: self.offcpu_in_site_n.saturating_sub(before.offcpu_in_site_n),
+            }
+        }
+    }
+
+    /// Snapshot the cumulative audit counters.
+    pub fn ledger_counters() -> LedgerCounters {
+        LedgerCounters {
+            spin_absorbed_cycles: SPIN_ABSORBED_CYCLES.load(Ordering::Relaxed),
+            park_forgiven_cycles: PARK_FORGIVEN_CYCLES.load(Ordering::Relaxed),
+            drop_park_cycles: DROP_PARK_CYCLES.load(Ordering::Relaxed),
+            drop_park_n: DROP_PARK_N.load(Ordering::Relaxed),
+            drop_yield_cycles: DROP_YIELD_CYCLES.load(Ordering::Relaxed),
+            drop_yield_n: DROP_YIELD_N.load(Ordering::Relaxed),
+            discard_overmax_cycles: DISCARD_OVERMAX_CYCLES.load(Ordering::Relaxed),
+            discard_overmax_n: DISCARD_OVERMAX_N.load(Ordering::Relaxed),
+            discard_unarmed_n: DISCARD_UNARMED_N.load(Ordering::Relaxed),
+            offcpu_in_site_cycles: OFFCPU_IN_SITE_CYCLES.load(Ordering::Relaxed),
+            offcpu_in_site_n: OFFCPU_IN_SITE_N.load(Ordering::Relaxed),
+        }
+    }
+
+    /// Absorbed-delay ledger of the on-CPU actor (testing/introspection).
+    pub fn my_absorbed_delay_cycles() -> u64 {
+        let slot = preempt::current_slot_ptr();
+        if slot.is_null() {
+            return 0;
+        }
+        // SAFETY: on-CPU actor's slot; see `check_cancelled`.
+        unsafe { (*slot).causal_delay() }
+    }
+
+    /// Test support: start an experiment targeting `site_name` at `pct`%
+    /// virtual speedup. Registers the site if needed.
+    pub fn begin_experiment_for_test(name: &'static str, pct: u32) {
+        begin(site_id(name), pct);
+    }
+
+    /// Test support: stop the running experiment.
+    pub fn end_experiment_for_test() {
+        end();
+    }
+
+    /// Test support: grow the global delay ledger directly, as if target-site
+    /// samples had attributed `cycles` — deterministic driver for the timer
+    /// virtual-time tests. Calibrates the TSC eagerly so conversion later
+    /// never stalls a scheduler loop.
+    pub fn inject_delay_cycles_for_test(cycles: u64) {
+        let _ = tsc_hz();
+        GLOBAL_DELAY.fetch_add(cycles, Ordering::Relaxed);
+    }
+
+    /// Convert ledger cycles to wall time at the measured TSC rate.
+    pub fn cycles_to_duration(cycles: u64) -> Duration {
+        Duration::from_secs_f64(cycles as f64 / tsc_hz())
+    }
+
+    /// Controller parameters: which speedups to try per site, and the
+    /// experiment/cooldown windows.
+    pub struct ExperimentPlan {
+        pub speedups_pct: Vec<u32>,
+        pub experiment: Duration,
+        pub cooldown: Duration,
+    }
+
+    impl Default for ExperimentPlan {
+        fn default() -> Self {
+            ExperimentPlan {
+                speedups_pct: vec![0, 25, 50],
+                experiment: Duration::from_millis(500),
+                cooldown: Duration::from_millis(100),
+            }
+        }
+    }
+
+    /// One completed experiment cell.
+    #[derive(Default)]
+    pub struct ExperimentResult {
+        pub site: String,
+        pub speedup_pct: u32,
+        pub duration: Duration,
+        /// Progress-point deltas over the window, `(name, count)`.
+        pub deltas: Vec<(String, u64)>,
+        /// Virtual delay injected during the window (cycles).
+        pub injected_cycles: u64,
+        // Ledger-audit deltas over the window (RFC 007 deficit hunt); see
+        // `LedgerCounters` for field semantics. `spin_absorbed_cycles > 0`
+        // in a 0% cell means the window paid debt left over from an earlier
+        // one — the baseline-contamination signature.
+        pub spin_absorbed_cycles: u64,
+        pub park_forgiven_cycles: u64,
+        pub drop_park_cycles: u64,
+        pub drop_park_n: u64,
+        pub drop_yield_cycles: u64,
+        pub drop_yield_n: u64,
+        pub discard_overmax_cycles: u64,
+        pub discard_overmax_n: u64,
+        pub discard_unarmed_n: u64,
+        pub offcpu_in_site_cycles: u64,
+        pub offcpu_in_site_n: u64,
+    }
+
+    /// Run the plan synchronously on the calling (OS) thread: for every
+    /// registered site × speedup, run one experiment window and record
+    /// progress-point deltas, with a cooldown between cells. Sites and
+    /// progress points must already be registered (the workload has to be
+    /// running); the caller owns workload start/stop.
+    ///
+    /// v1 controller: exhaustive sweep, fixed windows, no adaptive site
+    /// selection or confidence stopping (jar Q5).
+    ///
+    /// Callable from a plain OS thread *or* from inside an actor: sleeping
+    /// parks the green thread when we're on one (so no scheduler thread is
+    /// blocked), and falls back to `thread::sleep` otherwise.
+    pub fn run_experiments(plan: &ExperimentPlan) -> Vec<ExperimentResult> {
+        fn controller_sleep(d: Duration) {
+            if preempt::current_slot_ptr().is_null() {
+                std::thread::sleep(d);
+            } else {
+                // Wall-anchored: the controller's window/cooldown sleeps
+                // *define* the experiment's wall length; letting them chase
+                // the delay it is itself injecting would stretch every window
+                // (observed ~2x at 50% speedup). Deltas are rate-normalized
+                // either way — this fixes cost, not bias.
+                crate::scheduler::sleep_wall(d);
+            }
+        }
+        // Calibrate before any window so report rendering never has to sleep.
+        let _ = tsc_hz();
+        let site_list: Vec<(u32, String)> = {
+            let v = lock_unpoisoned(sites());
+            v.iter()
+                .enumerate()
+                .map(|(i, n)| ((i + 1) as u32, (*n).to_string()))
+                .collect()
+        };
+        let mut out = Vec::new();
+        for (sid, sname) in &site_list {
+            for &pct in &plan.speedups_pct {
+                let before = progress_snapshot();
+                let injected_before = global_delay_cycles();
+                let audit_before = ledger_counters();
+                let t0 = Instant::now();
+                begin(*sid, pct);
+                controller_sleep(plan.experiment);
+                end();
+                // Snapshot immediately: injection and spin freeze at `end()`
+                // (checks gate on the experiment word), but forgiveness does
+                // not — a later snapshot would leak cooldown wakes into the
+                // window.
+                let audit = ledger_counters().delta_since(&audit_before);
+                let elapsed = t0.elapsed();
+                let after = progress_snapshot();
+                let deltas = after
+                    .iter()
+                    .map(|(n, c)| {
+                        let b = before
+                            .iter()
+                            .find(|(bn, _)| bn == n)
+                            .map(|(_, bc)| *bc)
+                            .unwrap_or(0);
+                        (n.clone(), c.saturating_sub(b))
+                    })
+                    .collect();
+                out.push(ExperimentResult {
+                    site: sname.clone(),
+                    speedup_pct: pct,
+                    duration: elapsed,
+                    deltas,
+                    injected_cycles: global_delay_cycles() - injected_before,
+                    spin_absorbed_cycles: audit.spin_absorbed_cycles,
+                    park_forgiven_cycles: audit.park_forgiven_cycles,
+                    drop_park_cycles: audit.drop_park_cycles,
+                    drop_park_n: audit.drop_park_n,
+                    drop_yield_cycles: audit.drop_yield_cycles,
+                    drop_yield_n: audit.drop_yield_n,
+                    discard_overmax_cycles: audit.discard_overmax_cycles,
+                    discard_overmax_n: audit.discard_overmax_n,
+                    discard_unarmed_n: audit.discard_unarmed_n,
+                    offcpu_in_site_cycles: audit.offcpu_in_site_cycles,
+                    offcpu_in_site_n: audit.offcpu_in_site_n,
+                });
+                controller_sleep(plan.cooldown);
+            }
+        }
+        out
+    }
+
+    // -----------------------------------------------------------------------
+    // Reports
+    // -----------------------------------------------------------------------
+
+    /// Measured TSC frequency (Hz), calibrated once. The crate-wide 3 GHz
+    /// constant is fine for the *relative* timeslice check, but report
+    /// normalisation divides wall time by injected time, so a 20% Hz error
+    /// skews every impact number — measured live: a 3.7 GHz box inflated all
+    /// baselines uniformly. Calibrated against `Instant` over ~50ms on first
+    /// use; `run_experiments` triggers it before its first window (using the
+    /// park-aware sleep, so no scheduler thread is blocked when called from
+    /// an actor).
+    static TSC_HZ_MEASURED: OnceLock<f64> = OnceLock::new();
+
+    /// Measured TSC frequency in Hz. Calibrates on first call (~50ms).
+    pub fn tsc_hz() -> f64 {
+        *TSC_HZ_MEASURED.get_or_init(|| {
+            let c0 = preempt::rdtsc();
+            let t0 = Instant::now();
+            let d = Duration::from_millis(50);
+            if preempt::current_slot_ptr().is_null() {
+                std::thread::sleep(d);
+            } else {
+                // Wall-anchored: calibration divides TSC delta by *wall*
+                // elapsed; a virtual sleep dilated by concurrent injection
+                // would still measure correctly (elapsed() is wall) but
+                // waste window time — and must never depend on the ledger
+                // it exists to convert.
+                crate::scheduler::sleep_wall(d);
+            }
+            (preempt::rdtsc().wrapping_sub(c0)) as f64 / t0.elapsed().as_secs_f64()
+        })
+    }
+
+    /// Normalized rate for one cell: count over the *virtual* window
+    /// (wall − injected) — Coz's normalization: injected delay does not
+    /// exist in the virtual timeline. A bottleneck site keeps its raw count
+    /// while shrinking the divisor → positive impact; a fully overlapped
+    /// site loses count proportionally → ~zero.
+    fn normalized_rate(r: &ExperimentResult, point: &str) -> Option<f64> {
+        let count = r.deltas.iter().find(|(n, _)| n == point).map(|(_, c)| *c)?;
+        let injected_secs = r.injected_cycles as f64 / tsc_hz();
+        let virtual_secs = (r.duration.as_secs_f64() - injected_secs).max(1e-9);
+        Some(count as f64 / virtual_secs)
+    }
+
+    /// Impact of virtually speeding up `site` by `speedup_pct` on progress
+    /// point `point`, in percent relative to that site's own 0% baseline
+    /// cell. `None` if either cell or the point is missing, or the baseline
+    /// rate is zero. This is the machine-readable form of the summary's
+    /// "vs baseline" column, for programmatic checks (CI, examples).
+    pub fn impact_pct(
+        results: &[ExperimentResult],
+        site: &str,
+        speedup_pct: u32,
+        point: &str,
+    ) -> Option<f64> {
+        let cell = results
+            .iter()
+            .find(|r| r.site == site && r.speedup_pct == speedup_pct)?;
+        let base = results.iter().find(|r| r.site == site && r.speedup_pct == 0)?;
+        let rate = normalized_rate(cell, point)?;
+        let b = normalized_rate(base, point)?;
+        if b <= 0.0 {
+            return None;
+        }
+        Some((rate / b - 1.0) * 100.0)
+    }
+
+    /// Human-readable summary: per (site, progress point), the throughput at
+    /// each virtual speedup and the change relative to that site's own 0%
+    /// baseline. A near-zero column across speedups means: optimising this
+    /// site buys you nothing — the RFC's headline answer.
+    ///
+    /// Ends with a one-line fidelity note (RFC 007 Validation): reported
+    /// impacts are conservative — attribution counts on-CPU site time only,
+    /// so runnable queue-wait inside the site (the located @50 "deficit",
+    /// eff ≈ 0.93 live) is never injected and gains are lower bounds; site
+    /// *rankings* are unaffected.
+    pub fn render_summary(results: &[ExperimentResult]) -> String {
+        use std::fmt::Write;
+        let mut s = String::new();
+        let _ = writeln!(s, "== smarm causal profile ==");
+        let mut sites_seen: Vec<&str> = Vec::new();
+        for r in results {
+            if !sites_seen.contains(&r.site.as_str()) {
+                sites_seen.push(&r.site);
+            }
+        }
+        for site in sites_seen {
+            let _ = writeln!(s, "site {site}");
+            for r in results.iter().filter(|r| r.site == site) {
+                for (name, _) in &r.deltas {
+                    let rate = match normalized_rate(r, name) {
+                        Some(x) => x,
+                        None => continue,
+                    };
+                    let rel = impact_pct(results, site, r.speedup_pct, name)
+                        .map(|p| format!("{p:+.1}%"))
+                        .unwrap_or_else(|| "n/a".to_string());
+                    let _ = writeln!(
+                        s,
+                        "  speedup {:>3}%  {name:<24} {rate:>12.1}/s  vs baseline {rel}  (injected {:.1}ms)",
+                        r.speedup_pct,
+                        r.injected_cycles as f64 / tsc_hz() * 1e3
+                    );
+                }
+            }
+        }
+        if !results.is_empty() {
+            let _ = writeln!(
+                s,
+                "note: impacts are lower bounds — site time counts on-CPU only (runnable queue-wait is not attributed); rankings unaffected"
+            );
+        }
+        s
+    }
+
+    /// Ledger-audit companion to `render_summary` (RFC 007 deficit hunt):
+    /// per cell, where the window's virtual delay went — born (injected),
+    /// paid (absorbed), waived (forgiven at wake) — and the attribution the
+    /// sampler lost: tails dropped at parks/yields inside the target site,
+    /// plus clamp discards. Cycle columns in ms at the calibrated TSC rate.
+    /// `absorbed` above `injected` in a cell (0% especially) means it paid
+    /// debt left over from earlier windows.
+    pub fn render_ledger_audit(results: &[ExperimentResult]) -> String {
+        use std::fmt::Write;
+        let hz = tsc_hz();
+        let ms = |c: u64| c as f64 / hz * 1e3;
+        let mut s = String::new();
+        let _ = writeln!(s, "== smarm causal ledger audit ==");
+        for r in results {
+            let _ = writeln!(
+                s,
+                "site {:<22} @{:>2}%  injected {:>7.1}ms  absorbed {:>7.1}ms  forgiven {:>7.1}ms  \
+                 drop park {:>6.2}ms/{:<4} yield {:>6.2}ms/{:<4} offcpu {:>6.2}ms/{:<5} \
+                 discard >max {:>6.2}ms/{:<3} unarmed {}",
+                r.site,
+                r.speedup_pct,
+                ms(r.injected_cycles),
+                ms(r.spin_absorbed_cycles),
+                ms(r.park_forgiven_cycles),
+                ms(r.drop_park_cycles),
+                r.drop_park_n,
+                ms(r.drop_yield_cycles),
+                r.drop_yield_n,
+                ms(r.offcpu_in_site_cycles),
+                r.offcpu_in_site_n,
+                ms(r.discard_overmax_cycles),
+                r.discard_overmax_n,
+                r.discard_unarmed_n
+            );
+        }
+        s
+    }
+
+    /// Coz-compatible profile text (`profile.coz`), so Coz's existing plot
+    /// tooling renders our experiments — the RFC's "don't build a UI" call.
+    pub fn render_coz(results: &[ExperimentResult]) -> String {
+        use std::fmt::Write;
+        let mut s = String::new();
+        let _ = writeln!(s, "startup\ttime=0");
+        for r in results {
+            let _ = writeln!(
+                s,
+                "experiment\tselected={}\tspeedup={:.2}\tduration={}\tselected-samples=1",
+                r.site,
+                r.speedup_pct as f64 / 100.0,
+                r.duration.as_nanos()
+            );
+            for (name, count) in &r.deltas {
+                let _ = writeln!(s, "throughput-point\tname={name}\tdelta={count}");
+            }
+        }
+        s
+    }
+}
+
+#[cfg(feature = "smarm-causal")]
+pub use inner::*;
+
+/// Mark one unit of useful work complete at a named throughput progress
+/// point (RFC 007). One Relaxed increment when `smarm-causal` is on; nothing
+/// at all when it's off.
+#[cfg(feature = "smarm-causal")]
+#[macro_export]
+macro_rules! progress {
+    ($name:literal) => {{
+        static __SMARM_PP: ::std::sync::OnceLock<&'static $crate::causal::ProgressPoint> =
+            ::std::sync::OnceLock::new();
+        __SMARM_PP
+            .get_or_init(|| $crate::causal::register_progress($name))
+            .bump();
+    }};
+}
+
+#[cfg(not(feature = "smarm-causal"))]
+#[macro_export]
+macro_rules! progress {
+    ($name:literal) => {{}};
+}
+
+/// Enter a named causal-profiling site for the current actor; the returned
+/// guard exits it (restoring any enclosing site) on drop. Site identity is
+/// stored in the actor's slot, so it survives preemption and migration.
+/// Expands to a unit no-op without `smarm-causal`.
+#[cfg(feature = "smarm-causal")]
+#[macro_export]
+macro_rules! causal_site {
+    ($name:literal) => {{
+        static __SMARM_SITE: ::std::sync::OnceLock<u32> = ::std::sync::OnceLock::new();
+        $crate::causal::SiteGuard::enter(*__SMARM_SITE.get_or_init(|| $crate::causal::site_id($name)))
+    }};
+}
+
+#[cfg(not(feature = "smarm-causal"))]
+#[macro_export]
+macro_rules! causal_site {
+    ($name:literal) => {
+        ()
+    };
+}

src/channel.rs

@@ -132,7 +132,21 @@ impl<T> Drop for Sender<T> {
 
 impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
-        self.inner.lock().receiver_alive = false;
+        // The only consumer is gone: queued messages can never be delivered.
+        // Drop them now instead of stranding them until the last Sender goes
+        // away (a registry entry under lazy prune can keep a Sender — and thus
+        // the Arc<Inner> — alive long after the server exits). Dropping a queued
+        // Envelope::Call drops its reply_tx, waking any caller parked in `call`
+        // with ServerDown, so the documented guarantee holds on *every* teardown
+        // path, not only the all-senders-drop one. Drain under the lock, then
+        // run item destructors after releasing it (a reply_tx drop reaches into
+        // a *different* channel's lock + the scheduler, so it must not nest).
+        let drained = {
+            let mut g = self.inner.lock();
+            g.receiver_alive = false;
+            std::mem::take(&mut g.queue)
+        };
+        drop(drained);
     }
 }
 

src/io.rs

@@ -498,8 +498,12 @@ fn make_pipe() -> io::Result<(RawFd, RawFd)> {
     Ok((fds[0], fds[1]))
 }
 
-/// Drain pending bytes from the wake pipe. The scheduler calls this after
-/// a `poll` wakeup so the next idle call sees an empty pipe.
+/// Drain pending bytes from the wake pipe. Nonblocking (pipe is O_NONBLOCK).
+///
+/// DISCIPLINE: called only by the phase-1 drain-lock winner, immediately
+/// before `drain_completions`. Bytes are the notification channel for
+/// completions; consuming one anywhere else can strand the completion it
+/// announces (see the lost-wakeup note at the call site in `schedule_loop`).
 pub fn drain_wake_pipe(fd: RawFd) {
     let mut buf = [0u8; 64];
     loop {

src/lib.rs

@@ -38,6 +38,7 @@ pub(crate) mod sync_shim;
 #[doc(hidden)] // pub only so benches/rq_micro.rs can drive the raw structures
 pub mod run_queue;
 pub mod trace;
+pub mod causal;
 
 // ---------------------------------------------------------------------------
 // Global allocator
@@ -79,7 +80,8 @@ pub use registry::{
 };
 pub use runtime::{init, Config, Runtime};
 pub use scheduler::{
-    block_on_io, cancel_timer, request_stop, run, self_pid, send_after, send_after_named, sleep,
+    block_on_io, cancel_timer, request_stop, run, self_pid, send_after, send_after_named,
+    send_after_named_wall, send_after_wall, sleep, sleep_wall,
     spawn, spawn_addr, spawn_under, wait_readable, wait_readable_timeout, wait_writable,
     wait_writable_timeout, yield_now, FdArm, JoinError, JoinHandle,
 };

src/preempt.rs

@@ -98,6 +98,25 @@ pub(crate) fn clear_current_slot() {
     CURRENT_SLOT.with(|c| c.set(std::ptr::null()));
 }
 
+/// RFC 007 (`smarm-causal`) — raw pointer to the on-CPU actor's slot, null on
+/// the scheduler's own stack. Same lifetime argument as `note_overrun`: the
+/// slot is never reclaimed while its actor is on-CPU.
+#[cfg(feature = "smarm-causal")]
+#[inline]
+pub(crate) fn current_slot_ptr() -> *const crate::runtime::Slot {
+    CURRENT_SLOT.with(|c| c.get())
+}
+
+/// RFC 007 (`smarm-causal`) — push the slice start forward by `cycles`, so
+/// virtually-injected delay spun inside `maybe_preempt` does not count against
+/// the actor's timeslice (the clock-correction half of the RFC: the runtime
+/// owns this clock, so it can subtract its own perturbation).
+#[cfg(feature = "smarm-causal")]
+#[inline]
+pub(crate) fn extend_timeslice(cycles: u64) {
+    TIMESLICE_START.with(|c| c.set(c.get().wrapping_add(cycles)));
+}
+
 /// Tally a timeslice overrun against the on-CPU actor (RFC 016 Chunk 2). A
 /// no-op if no actor is bound (the scheduler's own stack). Reached only from
 /// the slice-expiry branch, which is already the yield path, so its cost is
@@ -247,6 +266,11 @@ pub fn maybe_preempt() {
                 // Observe a pending stop first: if we are being cancelled
                 // there is no point yielding, we unwind instead.
                 check_cancelled();
+                // RFC 007: causal-profiling sample/absorb point. Shares the
+                // amortised cadence, and the PREEMPTION_ENABLED gate — so it
+                // can never spin inside a prep-to-park or no-preempt region.
+                #[cfg(feature = "smarm-causal")]
+                crate::causal::check();
                 let start = TIMESLICE_START.with(|s| s.get());
                 if rdtsc().saturating_sub(start) > CONFIGURED_TIMESLICE_CYCLES.with(|t| t.get()) {
                     // Tally the overrun (RFC 016 Chunk 2) before handing back —

src/registry.rs

@@ -217,14 +217,20 @@ pub(crate) struct MailboxInfo {
 }
 
 /// The directory. Invariant (held under the registry lock): every value in
-/// `by_name` is the index of a [`Mailbox`] present in `by_index`, and that
-/// mailbox's `pid.index()` equals the key. Stale entries (dead actors) violate
-/// nothing — they are simply pruned on contact.
+/// `by_name` is the full [`Pid`] (index *and* generation) of an actor that
+/// published a [`Mailbox`] into `by_index` at registration time. Stale entries
+/// (dead holders — including holders whose slot has since been re-tenanted by
+/// a different actor) violate nothing — they are pruned on contact, and the
+/// generation makes "dead" decidable even after slot reuse.
 pub(crate) struct Registry {
     /// `pid.index() -> the actor's mailbox`. The handle store.
     by_index: HashMap<u32, Mailbox>,
-    /// `name -> pid.index()`. Several names may map to one actor.
-    by_name: HashMap<&'static str, u32>,
+    /// `name -> holder pid`. Several names may map to one actor. The full pid
+    /// (not just the index) is load-bearing: an index alone cannot tell a dead
+    /// holder from the live actor now tenanting its recycled slot, which made
+    /// such a name read as live-held — unresolvable *and* unregisterable — and
+    /// would misdeliver to a same-typed tenant (soak20 signature 2).
+    by_name: HashMap<&'static str, Pid>,
 }
 
 impl Registry {
@@ -232,10 +238,15 @@ impl Registry {
         Self { by_index: HashMap::new(), by_name: HashMap::new() }
     }
 
-    /// Drop a dead actor's mailbox and every name that pointed at it.
-    fn prune(&mut self, index: u32) {
-        self.by_index.remove(&index);
-        self.by_name.retain(|_, idx| *idx != index);
+    /// Drop a dead holder's artifacts: every name bound to it, and its mailbox
+    /// — but only while the mailbox is still *its own*. A recycled slot's
+    /// mailbox belongs to the live tenant (publish replaces it wholesale on
+    /// pid mismatch) and is left untouched.
+    fn prune_holder(&mut self, holder: Pid) {
+        self.by_name.retain(|_, p| *p != holder);
+        if self.by_index.get(&holder.index()).is_some_and(|mb| mb.pid == holder) {
+            self.by_index.remove(&holder.index());
+        }
     }
 
     /// RFC 016 snapshot input: per-slot-index registry view — the actor's
@@ -244,12 +255,15 @@ impl Registry {
     /// under the registry Leaf; the per-channel `queued_len` takes a Channel
     /// lock, legal under the Leaf (Leaf → Channel). Carries each mailbox's full
     /// `pid` so the caller can discard a stale incarnation's entry against the
-    /// slab's live generation. Names that dangle (point at no mailbox) are
-    /// dropped — they violate no invariant and get pruned on next contact.
+    /// slab's live generation. Names are matched to mailboxes by *full pid*, so
+    /// a stale name (dead holder) still annotates the corpse's own mailbox if
+    /// that survives, but never a recycled slot's new tenant; names that attach
+    /// to no mailbox are dropped — they violate no invariant and get pruned on
+    /// next contact.
     pub(crate) fn introspect_map(&self) -> HashMap<u32, MailboxInfo> {
-        let mut names: HashMap<u32, Vec<&'static str>> = HashMap::new();
-        for (&name, &idx) in &self.by_name {
-            names.entry(idx).or_default().push(name);
+        let mut names: HashMap<Pid, Vec<&'static str>> = HashMap::new();
+        for (&name, &pid) in &self.by_name {
+            names.entry(pid).or_default().push(name);
         }
         let mut out: HashMap<u32, MailboxInfo> = HashMap::with_capacity(self.by_index.len());
         for (&idx, mb) in &self.by_index {
@@ -258,7 +272,7 @@ impl Registry {
                 idx,
                 MailboxInfo {
                     pid: mb.pid,
-                    names: names.remove(&idx).unwrap_or_default(),
+                    names: names.remove(&mb.pid).unwrap_or_default(),
                     depth: depth.min(u32::MAX as usize) as u32,
                 },
             );
@@ -276,7 +290,7 @@ impl Registry {
         let names = self
             .by_name
             .iter()
-            .filter_map(|(&n, &i)| (i == idx).then_some(n))
+            .filter_map(|(&n, &p)| (p == mb.pid).then_some(n))
             .collect();
         Some(MailboxInfo { pid: mb.pid, names, depth: depth.min(u32::MAX as usize) as u32 })
     }
@@ -315,21 +329,22 @@ pub(crate) fn register_with<M: Send + 'static>(
         if !live(inner, me) {
             return Err(RegisterError::NoProc);
         }
-        if let Some(&holder_idx) = reg.by_name.get(key) {
-            match reg.by_index.get(&holder_idx).map(|m| m.pid) {
-                Some(holder) if holder == me => {} // same actor: just add the channel below
-                Some(holder) if live(inner, holder) => {
+        if let Some(&holder) = reg.by_name.get(key) {
+            if holder == me {
+                // Same actor: just add the channel below.
+            } else if live(inner, holder) {
                 return Err(RegisterError::NameTaken { holder });
-                }
-                Some(_) => reg.prune(holder_idx), // dead holder: free the name
-                None => {
-                    reg.by_name.remove(key); // dangling name: free it
-                }
+            } else {
+                // Dead holder: free the name (and its other stale artifacts).
+                // Liveness is judged against the *stored* pid, generation
+                // included — a recycled slot's live tenant no longer makes a
+                // dead name read as taken (soak20 signature 2).
+                reg.prune_holder(holder);
             }
         }
         // Publish (or extend) the mailbox with this channel, then bind the name.
         publish_channel::<M>(&mut reg, me, tx);
-        reg.by_name.insert(key, me.index());
+        reg.by_name.insert(key, me);
         Ok(())
     })
 }
@@ -394,18 +409,15 @@ pub(crate) fn install_for<M: Send + 'static>(pid: Pid, tx: Sender<M>) {
 pub fn whereis(name: &str) -> Option<Pid> {
     with_runtime(|inner| {
         let mut reg = inner.registry.lock();
-        let idx = *reg.by_name.get(name)?;
-        match reg.by_index.get(&idx).map(|m| m.pid) {
-            Some(pid) if live(inner, pid) => Some(pid),
-            Some(_) => {
-                reg.prune(idx);
+        let pid = *reg.by_name.get(name)?;
+        if live(inner, pid) {
+            Some(pid)
+        } else {
+            // Generation-checked against the stored holder: a recycled slot's
+            // live tenant reads dead here, and the stale name heals.
+            reg.prune_holder(pid);
             None
         }
-            None => {
-                reg.by_name.remove(name);
-                None
-            }
-        }
     })
 }
 
@@ -431,19 +443,18 @@ pub fn lookup_as<A: Addressable>(name: &str) -> Option<Pid<A>> {
 pub(crate) fn resolve_named_sender<M: Send + 'static>(name: &str) -> Option<(Pid, Sender<M>)> {
     with_runtime(|inner| {
         let mut reg = inner.registry.lock();
-        let idx = *reg.by_name.get(name)?;
-        let pid = match reg.by_index.get(&idx).map(|m| m.pid) {
-            Some(pid) if live(inner, pid) => pid,
-            Some(_) => {
-                reg.prune(idx);
+        let pid = *reg.by_name.get(name)?;
+        if !live(inner, pid) {
+            // Stored-pid liveness, generation included: a name whose holder
+            // died is pruned (heals) even if the slot has a new tenant —
+            // previously the tenant's mailbox made the name unresolvable
+            // *without* pruning, wedging it for the tenant's lifetime.
+            reg.prune_holder(pid);
             return None;
         }
-            None => {
-                reg.by_name.remove(name);
-                return None;
-            }
-        };
-        let tx = reg.by_index.get(&idx).and_then(Mailbox::clone_sender::<M>)?;
+        // A live holder's mailbox is its own (publish replaces wholesale on
+        // pid mismatch, and one live actor per slot), so index lookup is safe.
+        let tx = reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>)?;
         Some((pid, tx))
     })
 }
@@ -454,11 +465,8 @@ pub(crate) fn resolve_named_sender<M: Send + 'static>(name: &str) -> Option<(Pid
 pub fn unregister(name: &str) -> Option<Pid> {
     with_runtime(|inner| {
         let mut reg = inner.registry.lock();
-        let idx = reg.by_name.remove(name)?;
-        match reg.by_index.get(&idx).map(|m| m.pid) {
-            Some(pid) if live(inner, pid) => Some(pid),
-            _ => None,
-        }
+        let pid = reg.by_name.remove(name)?;
+        if live(inner, pid) { Some(pid) } else { None }
     })
 }
 
@@ -476,22 +484,20 @@ pub fn send<M: Send + 'static>(name: Name<M>, msg: M) -> Result<(), SendError<M>
         // before sending (a send can unpark a receiver).
         let tx = {
             let mut reg = inner.registry.lock();
-            let idx = match reg.by_name.get(key) {
-                Some(&i) => i,
+            let pid = match reg.by_name.get(key) {
+                Some(&p) => p,
                 None => return Err(SendError::Unresolved(msg)),
             };
-            let pid = match reg.by_index.get(&idx).map(|m| m.pid) {
-                Some(pid) => pid,
-                None => {
-                    reg.by_name.remove(key);
-                    return Err(SendError::Unresolved(msg));
-                }
-            };
             if !live(inner, pid) {
-                reg.prune(idx);
+                // Stored-pid liveness (generation included). Previously this
+                // checked the slot's *current* mailbox pid, so a recycled
+                // slot's live tenant passed — and a same-typed tenant would
+                // have received the message (misdelivery), a differently
+                // typed one a misleading NoChannel.
+                reg.prune_holder(pid);
                 return Err(SendError::Unresolved(msg));
             }
-            match reg.by_index.get(&idx).and_then(Mailbox::clone_sender::<M>) {
+            match reg.by_index.get(&pid.index()).and_then(Mailbox::clone_sender::<M>) {
                 Some(tx) => tx,
                 None => return Err(SendError::NoChannel(msg)),
             }
@@ -526,7 +532,7 @@ fn send_to_pid<M: Send + 'static>(
             }
             // Our incarnation's mailbox, but the actor has died: prune + Dead.
             Some(stored) if stored == pid => {
-                reg.prune(pid.index());
+                reg.prune_holder(pid);
                 return Err(SendError::Dead(msg));
             }
             // A different incarnation (or nothing) occupies the slot: the actor

src/runtime.rs

@@ -445,6 +445,42 @@ pub(crate) struct Slot {
     /// and only when the `budget-accounting` feature is on (it costs two RDTSC
     /// per resume); stays 0 otherwise. Same single-writer Relaxed discipline.
     budget_cycles: AtomicU64,
+    /// RFC 007 (`smarm-causal`) — id of the causal-profiling site this actor
+    /// is currently inside (0 = none). Lives in the slot, not a thread-local,
+    /// so it survives preemption and cross-scheduler migration. Written only
+    /// by the actor itself (guard enter/exit on its own thread), read by that
+    /// same thread in `maybe_preempt` — single-writer Relaxed, like `overruns`.
+    /// Exists regardless of the feature; stays 0 without it.
+    causal_site: AtomicU32,
+    /// RFC 007 (`smarm-causal`) — virtual-speedup delay cycles this actor has
+    /// absorbed (or been credited). Compared against the global ledger in
+    /// `causal::check`; fast-forwarded on resume-from-park so blocked time
+    /// absorbs delay for free (Coz's blocked-thread rule). Same single-writer
+    /// discipline.
+    causal_delay: AtomicU64,
+    /// RFC 007 (`smarm-causal`) — true iff this actor's last deschedule was a
+    /// real park (a successful `park_return`, not a slice yield and not the
+    /// instant-wake re-queue). Consumed by `causal::on_resume`: only a wake
+    /// from genuine blocking forgives outstanding virtual delay; a merely
+    /// preempted (runnable) actor stays in debt and must pay by spinning.
+    /// Starts false: a fresh spawn is born current (`reset_counters` sets
+    /// `causal_delay` to the global ledger), and anything injected while it
+    /// sits spawn-queued is owed, not waived. Written by the owning
+    /// scheduler thread at the deschedule /
+    /// resume boundary only — same single-writer discipline as `causal_delay`.
+    causal_parked: AtomicBool,
+    /// RFC 007 — TSC at this actor's last *runnable* (yield) deschedule
+    /// while it sat in the live experiment's target site; 0 = no gap
+    /// pending. Paired with `causal_desched_epoch`; written on the
+    /// deschedule path, consumed at the next resume — same single-writer
+    /// discipline as `causal_delay`.
+    causal_desched_tsc: AtomicU64,
+    /// RFC 007 — experiment epoch live at that deschedule (see
+    /// `causal::EXPERIMENT_EPOCH`): the resume counts the gap only into
+    /// the same window, so one straddling `end()` — or a later `begin()`
+    /// with an identical site+pct word — is dropped instead of leaking a
+    /// cooldown across windows.
+    causal_desched_epoch: AtomicU64,
     /// Cold lifecycle data. See [`SlotCold`].
     pub(crate) cold: RawMutex<SlotCold>,
 }
@@ -459,6 +495,14 @@ impl Slot {
             overruns: AtomicU64::new(0),
             messages_received: AtomicU64::new(0),
             budget_cycles: AtomicU64::new(0),
+            causal_site: AtomicU32::new(0),
+            causal_delay: AtomicU64::new(0),
+            // Overwritten at every occupancy by `reset_counters` (born
+            // current); false so a hypothetical reset-skipping path cannot
+            // waive the whole global backlog at first resume.
+            causal_parked: AtomicBool::new(false),
+            causal_desched_tsc: AtomicU64::new(0),
+            causal_desched_epoch: AtomicU64::new(0),
             cold: RawMutex::new(SlotCold {
                 actor: None,
                 waiters: Vec::new(),
@@ -547,6 +591,92 @@ impl Slot {
         self.overruns.store(0, Ordering::Relaxed);
         self.messages_received.store(0, Ordering::Relaxed);
         self.budget_cycles.store(0, Ordering::Relaxed);
+        self.causal_site.store(0, Ordering::Relaxed);
+        // Born current (Coz's new-thread rule): a fresh incarnation neither
+        // owes the process's accumulated delay history nor books it as park
+        // forgiveness. The old init (delay 0, parked true) waived the whole
+        // monotone backlog once per spawn via the first resume — found live
+        // under close-mode conn churn (~95k spawns/s): millions of phantom
+        // forgiven ms per 700ms window, even in 0% cells. Parked starts
+        // false: delay injected while spawn-queued is *owed* (the newborn is
+        // runnable, not blocked) and paid at its first check — the semantics
+        // `audit_zero_pct_window_absorbs_leftover_debt` pins.
+        #[cfg(feature = "smarm-causal")]
+        self.causal_delay
+            .store(crate::causal::global_delay_cycles(), Ordering::Relaxed);
+        #[cfg(not(feature = "smarm-causal"))]
+        self.causal_delay.store(0, Ordering::Relaxed);
+        self.causal_parked.store(false, Ordering::Relaxed);
+        self.causal_desched_tsc.store(0, Ordering::Relaxed);
+        self.causal_desched_epoch.store(0, Ordering::Relaxed);
+    }
+
+    /// RFC 007 — mark that this actor's deschedule was a genuine park.
+    /// Called from the scheduler's `YieldIntent::Park` branch on a successful
+    /// `park_return` only.
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn set_causal_parked(&self) {
+        self.causal_parked.store(true, Ordering::Relaxed);
+    }
+
+    /// RFC 007 — consume the parked marker at resume: returns whether the
+    /// last deschedule was a real park, and clears it so the next resume
+    /// defaults to "was runnable" unless the park branch says otherwise.
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn take_causal_parked(&self) -> bool {
+        self.causal_parked.swap(false, Ordering::Relaxed)
+    }
+
+    /// RFC 007 — stash the runnable-deschedule instant and the experiment
+    /// epoch it happened under (offcpu-gap audit; see `causal::on_resume`).
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn set_causal_desched(&self, tsc: u64, epoch: u64) {
+        self.causal_desched_tsc.store(tsc, Ordering::Relaxed);
+        self.causal_desched_epoch.store(epoch, Ordering::Relaxed);
+    }
+
+    /// RFC 007 — consume the stash: `(tsc, epoch)`; `(0, _)` = none pending.
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn take_causal_desched(&self) -> (u64, u64) {
+        let tsc = self.causal_desched_tsc.swap(0, Ordering::Relaxed);
+        let epoch = self.causal_desched_epoch.load(Ordering::Relaxed);
+        (tsc, epoch)
+    }
+
+    /// RFC 007 — current causal site id (0 = none). Single-writer: only the
+    /// on-CPU actor's thread writes, via the site-guard enter/exit.
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn causal_site(&self) -> u32 {
+        self.causal_site.load(Ordering::Relaxed)
+    }
+
+    /// RFC 007 — write the current causal site id (guard enter/exit).
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn set_causal_site(&self, id: u32) {
+        self.causal_site.store(id, Ordering::Relaxed);
+    }
+
+    /// RFC 007 — absorbed/credited virtual-delay cycles.
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn causal_delay(&self) -> u64 {
+        self.causal_delay.load(Ordering::Relaxed)
+    }
+
+    /// RFC 007 — set the absorbed-delay ledger (spin-absorb, credit, or the
+    /// resume-path fast-forward). Single-writer per the resume protocol: the
+    /// actor's own thread while on-CPU, the resuming scheduler thread at the
+    /// resume boundary — never both at once.
+    #[cfg_attr(not(feature = "smarm-causal"), allow(dead_code))]
+    #[inline]
+    pub(crate) fn set_causal_delay(&self, v: u64) {
+        self.causal_delay.store(v, Ordering::Relaxed);
     }
 
     /// A pid's-eye snapshot of the slot. Cold paths re-read this under the
@@ -899,6 +1029,11 @@ pub fn init(config: Config) -> Runtime {
 impl Runtime {
     /// Run `f` as the initial actor, block until all actors finish.
     /// Can be called multiple times sequentially on the same `Runtime`.
+    ///
+    /// A panic in the initial actor propagates out of `run` (re-raised
+    /// after teardown, so the `Runtime` stays reusable even under a
+    /// caller's `catch_unwind`). Panics in other actors surface through
+    /// joins, links, monitors, and supervisors as usual.
     pub fn run(&self, f: impl FnOnce() + Send + 'static) {
         // Install smarm's panic hook on first call. The default Rust hook is
         // not reentrant — concurrent actor panics can trigger a double-panic
@@ -967,16 +1102,27 @@ impl Runtime {
         self.inner.root_swept.store(false, Ordering::Relaxed);
         self.inner.set_root(initial_handle.pid());
 
-        // Launch N-1 extra scheduler threads. The calling thread is thread 0.
+        // Launch N-1 extra scheduler threads, named `smarm-sched-{slot}` so
+        // they are identifiable in `/proc/<pid>/task/*/comm`, stack dumps and
+        // debuggers. The calling thread is thread 0 and keeps its caller-given
+        // name (an embedder typically names it when spawning `run`).
         let mut os_threads = Vec::new();
         for slot in 1..self.thread_count {
             let inner = self.inner.clone();
-            let t = thread::spawn(move || {
+            let t = thread::Builder::new()
+                .name(format!("smarm-sched-{slot}"))
+                .spawn(move || {
                     RUNTIME.with(|r| *r.borrow_mut() = Some(inner.clone()));
                     SCHED_SLOT.with(|s| s.set(slot));
                     schedule_loop(&inner, slot);
                     RUNTIME.with(|r| *r.borrow_mut() = None);
                 });
+            // `thread::spawn` (the previous form) also panics when the OS
+            // refuses a thread, so this keeps the failure semantics.
+            let t = match t {
+                Ok(t) => t,
+                Err(e) => panic!("failed to spawn smarm scheduler thread: {e}"),
+            };
             os_threads.push(t);
         }
 
@@ -989,6 +1135,23 @@ impl Runtime {
             let _ = t.join();
         }
 
+        // The root's outcome, read before its handle drops (the outstanding
+        // handle pins the slot: no reclaim, no generation change under us).
+        // A root panic must escape `run()` — propagated at the tail below,
+        // after teardown. Dropping it unread here made every assert inside
+        // `run` silently vacuous (found live: a failing-first test passed).
+        let root_outcome = {
+            let slot = match self.inner.slot_at(initial_handle.pid()) {
+                Some(slot) => slot,
+                None => panic!("run(): root pid out of range"),
+            };
+            debug_assert!(
+                slot.status_for(initial_handle.pid()) == Status::Done,
+                "root not Done at run() teardown"
+            );
+            slot.cold.lock().outcome.take()
+        };
+
         // Drop initial handle (decrements outstanding_handles count).
         drop(initial_handle);
 
@@ -1022,6 +1185,26 @@ impl Runtime {
         // Flush trace to disk (no-op without smarm-trace).
         #[cfg(feature = "smarm-trace")]
         crate::trace::flush();
+
+        // Propagate a root panic — last, after the full teardown above, so
+        // a `catch_unwind` around `run()` leaves the `Runtime` reusable.
+        // `Exit` and `Stopped` return normally: a cooperatively stopped
+        // root is not an error. `resume_unwind` re-raises the original
+        // payload; the throw-site hook output was suppressed in-actor, so
+        // the caller sees the payload message without the origin file:line.
+        if let Some(Outcome::Panic(payload)) = root_outcome {
+            // Surface the message: the throw-site hook output was suppressed
+            // in-actor, so without this a harness shows a bare FAILED.
+            let msg: Option<&str> = payload
+                .downcast_ref::<&'static str>()
+                .copied()
+                .or_else(|| payload.downcast_ref::<String>().map(String::as_str));
+            eprintln!(
+                "smarm: root actor panicked: {}",
+                msg.unwrap_or("<non-string panic payload>")
+            );
+            std::panic::resume_unwind(payload);
+        }
     }
 
     /// Snapshot of runtime statistics for introspection / tests.
@@ -1342,7 +1525,20 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
             let completions = match inner.io.lock() {
                 Ok(mut io) => io
                     .as_mut()
-                    .map(|io| io.drain_completions())
+                    .map(|io| {
+                        // Consume wake-pipe bytes ONLY here, under the drain
+                        // lock and strictly before draining completions.
+                        // Producers push their completion before writing the
+                        // byte, so every byte consumed here has its completion
+                        // visible to the drain below. Consuming bytes anywhere
+                        // else — in particular after an idle poll, outside the
+                        // lock — loses wakeups: a try_lock loser can eat the
+                        // byte for a completion the winner never saw, leaving
+                        // it stranded (and its EPOLLONESHOT fd disarmed) until
+                        // an unrelated timer forces another drain pass.
+                        crate::io::drain_wake_pipe(io.wake_fd());
+                        io.drain_completions()
+                    })
                     .unwrap_or_default(),
                 Err(e) => panic!("smarm: io lock poisoned (core corrupt): {e}"),
             };
@@ -1552,16 +1748,22 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                                 let timeout = deadline - now;
                                 match fd_opt {
                                     Some(fd) => {
+                                        // Wake only; the byte (if any) is
+                                        // consumed by the next drain-lock
+                                        // winner in phase 1. Level-triggered
+                                        // poll means an unconsumed byte makes
+                                        // this return immediately, so a loser
+                                        // spins briefly until the winner
+                                        // releases — never sleeps through it.
                                         crate::io::poll_wake(fd, Some(timeout));
-                                        crate::io::drain_wake_pipe(fd);
                                     }
                                     None => thread::sleep(timeout),
                                 }
                             }
                         }
                         (None, Some(fd)) if io_outstanding > 0 => {
+                            // See above: no byte consumption outside phase 1.
                             crate::io::poll_wake(fd, None);
-                            crate::io::drain_wake_pipe(fd);
                         }
                         _ => {
                             thread::sleep(std::time::Duration::from_micros(100));
@@ -1616,6 +1818,11 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
             crate::preempt::reset_timeslice();
         }
         PREEMPTION_ENABLED.with(|c| c.set(true));
+        // RFC 007: delay accrued while this actor was off-CPU is absorbed for
+        // free (Coz's blocked-thread rule) — fast-forward its ledger and arm
+        // the per-thread sample clock before it runs.
+        #[cfg(feature = "smarm-causal")]
+        crate::causal::on_resume(slot);
 
         crate::te!(crate::trace::Event::Resume(pid));
         unsafe { switch_to_actor() };
@@ -1642,6 +1849,11 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
             let gen = pid.generation();
             match intent {
                 YieldIntent::Yield => {
+                    // RFC 007 audit: measure the sample tail this yield drops
+                    // (near-zero for slice-expiry yields, which sample at the
+                    // same checkpoint; fat for explicit yield_now in-site).
+                    #[cfg(feature = "smarm-causal")]
+                    crate::causal::on_deschedule(slot, false);
                     // Running OR RunningNotified → Queued; a notification
                     // arriving mid-run coalesces into the re-queue.
                     crate::te!(crate::trace::Event::Yield(pid));
@@ -1650,11 +1862,24 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                 }
                 YieldIntent::Park => {
                     if slot.word.park_return(gen) {
+                        // RFC 007 audit: an in-site park drops its sample
+                        // tail (nothing flushes it; on_resume re-arms).
+                        #[cfg(feature = "smarm-causal")]
+                        crate::causal::on_deschedule(slot, true);
+                        // RFC 007: a real park — the eventual wake forgives
+                        // delay accrued while blocked (the instant-wake
+                        // re-queue below does NOT: that actor never blocked).
+                        #[cfg(feature = "smarm-causal")]
+                        slot.set_causal_parked();
                         crate::te!(crate::trace::Event::Park(pid));
                     } else {
                         // An unpark landed in the prep-to-park window; the
                         // word is back to Queued — re-queue instead of
-                        // parking. The lost-wakeup window, closed.
+                        // parking. The lost-wakeup window, closed: this actor
+                        // never blocked, so its dropped tail counts as a
+                        // yield in the RFC 007 audit.
+                        #[cfg(feature = "smarm-causal")]
+                        crate::causal::on_deschedule(slot, false);
                         crate::te!(crate::trace::Event::UnparkFlagConsumed(pid));
                         inner.enqueue(pid);
                     }

src/scheduler.rs

@@ -403,6 +403,28 @@ pub fn sleep(duration: std::time::Duration) {
     park_current();
 }
 
+/// Like [`sleep`], but wall-anchored: under causal profiling (feature
+/// `smarm-causal`) the deadline is honoured in wall time instead of chasing
+/// injected virtual delay. Identical to [`sleep`] without the feature. For
+/// measurement machinery whose durations define wall time (the causal
+/// controller's windows, TSC calibration) — workload code wants [`sleep`].
+pub fn sleep_wall(duration: std::time::Duration) {
+    let me = match current_pid() {
+        Some(pid) => pid,
+        None => panic!("sleep_wall() called outside an actor"),
+    };
+    let _np = NoPreempt::enter();
+    let epoch = begin_wait();
+    let deadline = crate::timer::deadline_from_now(duration);
+    with_runtime(|inner| {
+        match inner.timers.lock() {
+            Ok(mut timers) => timers.insert_sleep_wall(deadline, me, epoch),
+            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
+        }
+    });
+    park_current();
+}
+
 pub fn insert_wait_timer(
     deadline: std::time::Instant,
     pid: Pid,
@@ -475,6 +497,51 @@ pub fn send_after_named<M: Send + 'static>(
     })
 }
 
+/// Wall-anchored [`send_after`] (RFC 007): under causal profiling the timer
+/// opts out of the virtual-time shift and fires at its raw deadline instead
+/// of dilating with injected delay — the user-facing opt-out whose substrate
+/// [`sleep_wall`] landed. Use it for deadlines that reflect the outside
+/// world (protocol timeouts, wall-clock schedules) rather than workload
+/// pacing. Without the `smarm-causal` feature it is identical to
+/// [`send_after`]. Cancellation via [`cancel_timer`] is unchanged.
+pub fn send_after_wall<A: crate::pid::Addressable>(
+    after: std::time::Duration,
+    dest: Pid<A>,
+    msg: A::Msg,
+) -> crate::timer::TimerId {
+    let deadline = crate::timer::deadline_from_now(after);
+    let fire = Box::new(move || {
+        let _ = crate::registry::send_to(dest, msg);
+    });
+    with_runtime(|inner| {
+        match inner.timers.lock() {
+            Ok(mut timers) => timers.insert_send_wall(deadline, dest.erase(), fire),
+            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
+        }
+    })
+}
+
+/// Wall-anchored [`send_after_named`] (RFC 007): re-resolving name delivery,
+/// raw-deadline anchor — see [`send_after_wall`] for the semantics.
+pub fn send_after_named_wall<M: Send + 'static>(
+    after: std::time::Duration,
+    dest: Name<M>,
+    msg: M,
+) -> crate::timer::TimerId {
+    let deadline = crate::timer::deadline_from_now(after);
+    // Informational only (who armed it); not used for delivery.
+    let armed_by = current_pid().unwrap_or(Pid::new(0, 0));
+    let fire = Box::new(move || {
+        let _ = crate::registry::send(dest, msg);
+    });
+    with_runtime(|inner| {
+        match inner.timers.lock() {
+            Ok(mut timers) => timers.insert_send_wall(deadline, armed_by, fire),
+            Err(e) => panic!("smarm: timers lock poisoned (core corrupt): {e}"),
+        }
+    })
+}
+
 /// Deliver `msg` onto a channel the caller owns after `after`, rather than to a
 /// registry address. The sibling of [`send_after`] used by the gen_server timer
 /// layer (RFC 015 §5): arming a server timer must land the fire on the loop's

src/timer.rs

@@ -102,6 +102,19 @@ pub struct Entry {
     seq: u64,
     pub pid: Pid,
     pub reason: Reason,
+    /// RFC 007 virtual time: the global delay ledger reading when this entry
+    /// was (re-)queued. `pop_due` shifts the effective deadline by any delay
+    /// injected since, so timers dilate together with the causally-delayed
+    /// workload instead of firing early in virtual terms.
+    #[cfg(feature = "smarm-causal")]
+    delay_stamp: u64,
+    /// RFC 007: a wall-anchored entry opts out of the virtual-time shift —
+    /// its deadline is honoured in wall time regardless of injected delay.
+    /// Used by the causal controller's own measurement/cooldown sleeps so
+    /// experiment windows keep a fixed wall length; ordinary workload timers
+    /// stay virtual (`false`).
+    #[cfg(feature = "smarm-causal")]
+    wall: bool,
 }
 
 impl PartialEq for Entry {
@@ -152,6 +165,19 @@ impl Timers {
         self.insert(deadline, pid, Reason::Sleep { epoch });
     }
 
+    /// Insert a *wall-anchored* `Sleep` timer: fires at `deadline` in wall
+    /// time even while causal profiling (feature `smarm-causal`) is injecting
+    /// virtual delay — it never chases the delay ledger. Without the feature
+    /// this is identical to [`insert_sleep`](Self::insert_sleep).
+    ///
+    /// Intended for measurement machinery (the causal controller's window and
+    /// cooldown sleeps, TSC calibration) whose durations *define* wall time
+    /// rather than participate in the workload. Workload code should use the
+    /// ordinary virtual-anchored timers.
+    pub fn insert_sleep_wall(&mut self, deadline: Instant, pid: Pid, epoch: u32) {
+        self.push(deadline, pid, Reason::Sleep { epoch }, true);
+    }
+
     /// Arm a cancellable `send_after` timer: run `fire` at `deadline` unless
     /// [`cancel`](Self::cancel)led first. `pid` is informational only (the
     /// destination, or who armed it — useful for introspection); it is *not*
@@ -163,11 +189,24 @@ impl Timers {
         pid: Pid,
         fire: Box<dyn FnOnce() + Send>,
     ) -> TimerId {
-        let seq = self.next_seq;
-        self.next_seq = self.next_seq.wrapping_add(1);
-        self.armed.insert(seq);
-        self.heap.push(Reverse(Entry { deadline, seq, pid, reason: Reason::Send { fire } }));
-        TimerId(seq)
+        self.armed.insert(self.next_seq);
+        TimerId(self.push(deadline, pid, Reason::Send { fire }, false))
+    }
+
+    /// Arm a *wall-anchored* cancellable `send_after` timer (RFC 007): the
+    /// same contract as [`insert_send`](Self::insert_send), but the entry
+    /// opts out of the virtual-time shift and fires at its raw deadline
+    /// regardless of injected delay — the `Send`-reason sibling of
+    /// [`insert_sleep_wall`](Self::insert_sleep_wall). Without the
+    /// `smarm-causal` feature this is identical to `insert_send`.
+    pub fn insert_send_wall(
+        &mut self,
+        deadline: Instant,
+        pid: Pid,
+        fire: Box<dyn FnOnce() + Send>,
+    ) -> TimerId {
+        self.armed.insert(self.next_seq);
+        TimerId(self.push(deadline, pid, Reason::Send { fire }, true))
     }
 
     /// Cancel an armed `send_after` timer. Returns `true` if the timer was
@@ -179,11 +218,31 @@ impl Timers {
         self.armed.remove(&id.0)
     }
 
-    /// Insert an arbitrary timer entry.
+    /// Insert an arbitrary (virtual-anchored) timer entry.
     pub fn insert(&mut self, deadline: Instant, pid: Pid, reason: Reason) {
+        self.push(deadline, pid, reason, false);
+    }
+
+    /// Common insertion path. `wall` selects the RFC 007 anchor (see
+    /// [`insert_sleep_wall`](Self::insert_sleep_wall)); it is accepted — and
+    /// ignored — without the `smarm-causal` feature so callers don't fork.
+    /// Returns the entry's `seq`.
+    fn push(&mut self, deadline: Instant, pid: Pid, reason: Reason, wall: bool) -> u64 {
+        #[cfg(not(feature = "smarm-causal"))]
+        let _ = wall;
         let seq = self.next_seq;
         self.next_seq = self.next_seq.wrapping_add(1);
-        self.heap.push(Reverse(Entry { deadline, seq, pid, reason }));
+        self.heap.push(Reverse(Entry {
+            deadline,
+            seq,
+            pid,
+            reason,
+            #[cfg(feature = "smarm-causal")]
+            delay_stamp: crate::causal::global_delay_cycles(),
+            #[cfg(feature = "smarm-causal")]
+            wall,
+        }));
+        seq
     }
 
     pub fn is_empty(&self) -> bool {
@@ -210,23 +269,61 @@ impl Timers {
     /// one is silently dropped here (its `seq` was already removed from
     /// `armed` by [`cancel`](Self::cancel)). Returning it removes it from
     /// `armed`, so a later `cancel` of a fired timer reports `false`.
+    ///
+    /// RFC 007 virtual time (feature `smarm-causal`): before an entry fires,
+    /// any global delay injected since it was (re-)queued is added to its
+    /// deadline; an entry whose *effective* deadline hasn't passed is pushed
+    /// back with the shifted deadline and a fresh stamp, so it keeps chasing
+    /// delay injected while it waits. Consequences, both benign:
+    /// [`peek_deadline`](Self::peek_deadline) may under-report (raw deadline
+    /// earlier than effective), costing at most one spurious scheduler wake
+    /// per injected chunk; and a shift never converts wall time — with zero
+    /// debt the path is byte-identical to the featureless one. Wall-anchored
+    /// entries ([`insert_sleep_wall`](Self::insert_sleep_wall)) are exempt
+    /// from the shift and always fire at their raw deadline.
     pub fn pop_due(&mut self, now: Instant) -> Vec<Entry> {
         let mut out = Vec::new();
+        #[cfg(feature = "smarm-causal")]
+        let global = crate::causal::global_delay_cycles();
         while let Some(r) = self.heap.peek() {
-            if r.0.deadline <= now {
-                let entry = match self.heap.pop() {
+            if r.0.deadline > now {
+                break;
+            }
+            #[allow(unused_mut)]
+            let mut entry = match self.heap.pop() {
                 Some(e) => e.0,
                 None => panic!("smarm: timer heap pop after peek returned None (core corrupt)"),
             };
-                if matches!(entry.reason, Reason::Send { .. }) && !self.armed.remove(&entry.seq) {
+            if matches!(entry.reason, Reason::Send { .. }) && !self.armed.contains(&entry.seq) {
                 // Cancelled before it came due: discard, do not deliver.
+                // (Checked before any shift so a cancelled entry is never
+                // re-queued just to be discarded later.)
                 continue;
             }
-                out.push(entry);
-            } else {
-                break;
+            #[cfg(feature = "smarm-causal")]
+            if !entry.wall {
+                let debt = global.saturating_sub(entry.delay_stamp);
+                if debt > 0 {
+                    let shifted = entry
+                        .deadline
+                        .checked_add(crate::causal::cycles_to_duration(debt))
+                        .unwrap_or(entry.deadline);
+                    if shifted > now {
+                        // Not due in virtual time: re-queue at the shifted
+                        // deadline, stamped, keeping `seq` (and thus `Send`
+                        // cancellation identity) intact.
+                        entry.deadline = shifted;
+                        entry.delay_stamp = global;
+                        self.heap.push(Reverse(entry));
+                        continue;
                     }
                 }
+            }
+            if matches!(entry.reason, Reason::Send { .. }) {
+                self.armed.remove(&entry.seq);
+            }
+            out.push(entry);
+        }
         out
     }
 }

tests/runtime.rs

@@ -485,3 +485,35 @@ fn multi_thread_timer_only_no_pipe_contention() {
         SLEEP_MS * 2,
     );
 }
+
+// ---------------------------------------------------------------------------
+// Root panic propagation
+
+/// A panic in the root actor escapes `run()` to the caller. Anything else
+/// makes every assert inside `run` silently vacuous — found live when a
+/// failing-first test passed: the tripped assert was caught by the
+/// trampoline, recorded as `Outcome::Panic` on the root slot, and dropped
+/// unread with the initial handle.
+#[test]
+#[should_panic(expected = "root actor panic escapes")]
+fn root_panic_escapes_run() {
+    rt1().run(|| {
+        panic!("root actor panic escapes");
+    });
+}
+
+/// Teardown completes before the root panic propagates: a caller that
+/// catches it can immediately `run()` again on the same `Runtime` (the
+/// documented sequential-reuse contract).
+#[test]
+fn runtime_reusable_after_root_panic() {
+    let r = rt1();
+    let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+        r.run(|| panic!("boom"));
+    }));
+    assert!(caught.is_err(), "root panic must escape run()");
+    let ran = Arc::new(AtomicBool::new(false));
+    let ran_t = ran.clone();
+    r.run(move || ran_t.store(true, Ordering::Relaxed));
+    assert!(ran.load(Ordering::Relaxed), "runtime unusable after root panic");
+}

tests/select.rs

@@ -108,8 +108,14 @@ fn loser_arm_wake_after_parked_select_stays_precise() {
             t0.elapsed() >= Duration::from_millis(40),
             "one-shot park returned early: a stale loser-arm wake landed"
         );
+        // Arm 0 is now closed (the sender actor exited after its sends) and
+        // a closed arm reports ready forever under priority order — observe
+        // the disconnect and drop it from the set, per the documented
+        // closed-arm rule.
+        assert_eq!(select(&[&rxa, &rxb]), 0);
+        assert!(rxa.try_recv().is_err(), "arm 0 must report disconnect");
         // The loser's message was never lost.
-        assert_eq!(select(&[&rxa, &rxb]), 1);
+        assert_eq!(select(&[&rxb]), 0);
         assert_eq!(rxb.try_recv().unwrap(), Some(2));
         h.join().unwrap();
     });

tests/stale_name_slot_reuse.rs

@@ -0,0 +1,169 @@
+//! Reproducer (soak20 signature 2, refcount_test.exs "watcher crash"):
+//! `by_name` stores only the slot *index*, so a name whose holder died — never
+//! unregistered, since no smarm stop path unregisters (prune is lazy) — and
+//! whose slot was then re-tenanted by an unrelated actor reads as *live-held*:
+//!
+//!   - `register` of the name fails `NameTaken { holder: <unrelated tenant> }`,
+//!     so the bridge's generated `start()` (a `let _ =`) silently no-ops and
+//!     `start_server/1` reports `:ok` for a server that never came up;
+//!   - a by-name `call` resolves the tenant's mailbox, misses on the message
+//!     `TypeId`, and fails `ServerDown` fast — and does NOT prune (only the
+//!     dead-holder and dangling-name arms prune), so the name never heals
+//!     while the tenant lives. The wedge is self-sustaining.
+//!
+//! Wild signature: 110235x fast `{:error, :server_down}` probes over the full
+//! 5 s await window after a swallowed restart (200-run width-20 soak, run 59).
+//!
+//! The test asserts the *contract*: after its holder dies, a name must be
+//! re-registrable regardless of what happened to the slot. Red pre-fix.
+
+use smarm::{
+    call, init, request_stop, whereis, CallError, Config, GenServer, GenServerBuilder,
+    GenServerName, RegisterError,
+};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+const TARGET: GenServerName<Target> = GenServerName::new("stale_reuse_target");
+
+/// The named server whose death opens the window. Trivial on purpose.
+struct Target;
+
+impl GenServer for Target {
+    type Call = ();
+    type Reply = ();
+    type Cast = ();
+    type Info = ();
+    type Timer = ();
+
+    fn handle_call(&mut self, _req: ()) {}
+    fn handle_cast(&mut self, _op: ()) {}
+}
+
+/// The unrelated tenant. A *different* server type, so its mailbox holds a
+/// different `Envelope` `TypeId` — a same-typed tenant would make the by-name
+/// `call` *deliver to the wrong server* instead of failing, which is the same
+/// root hole wearing a worse hat.
+struct Filler;
+
+impl GenServer for Filler {
+    type Call = ();
+    type Reply = ();
+    type Cast = ();
+    type Info = ();
+    type Timer = ();
+
+    fn handle_call(&mut self, _req: ()) {}
+    fn handle_cast(&mut self, _op: ()) {}
+}
+
+#[derive(Debug)]
+struct Observed {
+    old_slot: (u32, u32),
+    tenant_slot: (u32, u32),
+    /// `whereis` of the dead name after re-tenanting — `Some` is the misread.
+    whereis_after_reuse: Option<(u32, u32)>,
+    /// By-name call after re-tenanting — the wild `server_down` fast-fail.
+    call_after_reuse: Result<(), CallError>,
+    /// The contract under test: re-registering the dead name.
+    restart: Result<(), RegisterError>,
+}
+
+#[test]
+fn dead_name_with_reused_slot_must_be_re_registrable() {
+    let out: Arc<Mutex<Option<Observed>>> = Arc::new(Mutex::new(None));
+    let out_w = out.clone();
+
+    // A deliberately tiny slab forces prompt slot recycling: with every filler
+    // held alive, the freed slot is the only *recycled* one, so a filler lands
+    // on it deterministically well before the slab (a loud panic) runs out.
+    init(Config::exact(2).max_actors(32)).run(move || {
+        // 1. Named server up; record its slot.
+        let target = GenServerBuilder::new(Target)
+            .named(TARGET)
+            .start()
+            .expect("name should be free at test start");
+        let old_pid = target.pid();
+
+        // 2. Kill it WITHOUT unregistering (no stop path does). Death is
+        //    confirmed via the *ref*, never the name — a by-name resolve of a
+        //    dead-but-not-yet-reused holder takes the prune arm and heals the
+        //    name, destroying the precondition.
+        request_stop(old_pid);
+        loop {
+            match target.call(()) {
+                Err(CallError::ServerDown) => break,
+                Ok(()) => smarm::sleep(Duration::from_millis(5)),
+            }
+        }
+        drop(target);
+
+        // 3. Re-tenant the slot: spawn fillers (all kept alive) until one
+        //    lands on the old index.
+        let mut fillers = Vec::new();
+        let mut tenant = None;
+        for i in 0..24 {
+            let name: &'static str = Box::leak(format!("stale_filler_{i}").into_boxed_str());
+            let f = GenServerBuilder::new(Filler)
+                .named(GenServerName::<Filler>::new(name))
+                .start()
+                .expect("filler names are fresh");
+            let fp = f.pid();
+            fillers.push(f);
+            if fp.index() == old_pid.index() {
+                tenant = Some(fp);
+                break;
+            }
+        }
+        let tenant = tenant.expect(
+            "precondition: the freed slot must be re-tenanted within the tiny slab \
+             (slots are recycled; every filler is held alive)",
+        );
+
+        // 4. Observe the poisoned state through the same paths the bridge uses.
+        let whereis_after_reuse = whereis(TARGET.as_str()).map(|p| (p.index(), p.generation()));
+        let call_after_reuse = call(TARGET, ());
+        let restart = GenServerBuilder::new(Target)
+            .named(TARGET)
+            .start()
+            .map(|_fresh_ref| ());
+
+        *out_w.lock().unwrap() = Some(Observed {
+            old_slot: (old_pid.index(), old_pid.generation()),
+            tenant_slot: (tenant.index(), tenant.generation()),
+            whereis_after_reuse,
+            call_after_reuse,
+            restart,
+        });
+
+        drop(fillers);
+    });
+
+    let o = out.lock().unwrap().take().expect("run body completed");
+    eprintln!("observed: {o:?}");
+
+    assert!(
+        o.restart.is_ok(),
+        "re-registering '{}' after its holder died failed with {:?}: the dead name \
+         reads as held by the live, unrelated tenant {:?} because by_name kept only \
+         the slot index (old slot {:?}). This is the silent-no-op start_server path \
+         of soak20 signature 2.",
+        TARGET.as_str(),
+        o.restart,
+        o.tenant_slot,
+        o.old_slot,
+    );
+
+    // The healed semantics around the re-register: the stale name reads
+    // *unbound* (never the tenant), and a by-name call fails ServerDown rather
+    // than resolving anything of the tenant's.
+    assert_eq!(
+        o.whereis_after_reuse, None,
+        "whereis of a dead name must prune and report unbound, not the slot's new tenant",
+    );
+    assert_eq!(
+        o.call_after_reuse,
+        Err(CallError::ServerDown),
+        "a by-name call to a dead name must fail ServerDown",
+    );
+}

tests/stop_with_queued_call.rs

@@ -0,0 +1,120 @@
+//! Reproducer: a *named* gen_server stopped with `request_stop` while a `call`
+//! sits **un-dequeued** in its inbox does NOT release the parked caller with
+//! `CallError::ServerDown`. The caller parks forever, contradicting the
+//! documented gen_server guarantee ("Any caller currently waiting in `call`
+//! sees `Err(ServerDown)`").
+//!
+//! Root cause (channel.rs): `Receiver::Drop` only flips `receiver_alive = false`
+//! and never drains `queue`. The queued `Envelope::Call(_, reply_tx)` therefore
+//! survives as long as the channel `Arc<Inner>` does — and for a *named* server
+//! the registry holds a `Sender` clone (lazy prune) that keeps the `Arc` alive
+//! after the server is gone. So the queued `reply_tx` is never dropped, the
+//! caller's `reply_rx` never closes, and `reply_rx.recv()` parks forever.
+//!
+//! Anonymous servers happen to dodge this: when their last `GenServerRef`
+//! drops, every `Sender` drops, the `Arc` refcount hits zero, `Inner` (and its
+//! queue) is dropped, and the queued `reply_tx` goes with it — waking the
+//! caller. The bug is specific to "a `Sender` outlives the `Receiver`", which a
+//! registry entry guarantees for every named server.
+
+use smarm::{
+    call, channel, init, request_stop, spawn, Config, GenServer, GenServerBuilder, GenServerName,
+    CallError, Receiver, RecvTimeoutError,
+};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+const BLOCKER: GenServerName<Blocker> = GenServerName::new("repro_blocker");
+
+/// A server that, on its single cast, parks forever on a gate channel the test
+/// never feeds. This deterministically holds the server *inside a handler* (not
+/// at the inbox recv), so any subsequent `call` queues behind it and stays
+/// un-dequeued — exactly the state `request_stop` then has to clean up.
+struct Blocker {
+    gate: Option<Receiver<()>>,
+}
+
+impl GenServer for Blocker {
+    type Call = ();
+    type Reply = ();
+    type Cast = ();
+    type Info = ();
+    type Timer = ();
+
+    // Trivial + instant: if this ever ran for the queued call, the caller would
+    // get Ok(()) immediately. It must NOT run — the server is parked on the gate
+    // when the stop arrives.
+    fn handle_call(&mut self, _req: ()) {}
+
+    // Park forever (until cancelled). recv() on an open channel with no message
+    // parks the actor; the gate sender is held by the test and never fires.
+    fn handle_cast(&mut self, _op: ()) {
+        if let Some(gate) = self.gate.take() {
+            let _ = gate.recv();
+        }
+    }
+}
+
+#[test]
+fn named_server_request_stop_releases_queued_caller_with_server_down() {
+    // Final observation, asserted after the run.
+    //   Some(Err(ServerDown)) -> contract honored (fixed)
+    //   None                  -> caller never released; parked past the 3s
+    //                            bound (bug reproduced)
+    let outcome: Arc<Mutex<Option<Result<(), CallError>>>> = Arc::new(Mutex::new(None));
+    let outcome_w = outcome.clone();
+
+    init(Config::exact(2)).run(move || {
+        // Gate the server will park on. Held for the whole run so the server's
+        // gate.recv() parks (rather than seeing Disconnected and returning).
+        let (gate_tx, gate_rx) = channel::<()>();
+
+        // Channel the queued caller reports its result back on.
+        let (res_tx, res_rx) = channel::<Result<(), CallError>>();
+
+        // 1. Start the named server and keep its ref alive.
+        let server = GenServerBuilder::new(Blocker { gate: Some(gate_rx) })
+            .named(BLOCKER)
+            .start()
+            .expect("name should be free");
+        let spid = server.pid();
+
+        // 2. Send the cast and let the server dequeue it and park on the gate.
+        server.cast(()).expect("server is live");
+        smarm::sleep(Duration::from_millis(100));
+
+        // 3. A separate caller issues a by-name `call`. The server is parked on
+        //    the gate, so this Call envelope queues un-dequeued; the caller then
+        //    parks on its reply channel.
+        spawn(move || {
+            let r = call(BLOCKER, ());
+            let _ = res_tx.send(r);
+        });
+        smarm::sleep(Duration::from_millis(100));
+
+        // 4. Stop the server. Its loop unwinds out of the gate.recv() and drops
+        //    the inbox Receiver — at which point the queued caller is *supposed*
+        //    to be released with ServerDown.
+        request_stop(spid);
+
+        // 5. Bounded wait. A correct runtime releases the caller in well under
+        //    3s; the bug leaves it parked, so we time out.
+        let observed = match res_rx.recv_timeout(Duration::from_secs(3)) {
+            Ok(r) => Some(r),
+            Err(RecvTimeoutError::Timeout) => None,
+            Err(RecvTimeoutError::Disconnected) => None,
+        };
+        *outcome_w.lock().unwrap() = observed;
+
+        // Keep the gate sender alive until the very end.
+        drop(gate_tx);
+    });
+
+    let observed = outcome.lock().unwrap().take();
+    assert_eq!(
+        observed,
+        Some(Err(CallError::ServerDown)),
+        "queued caller was not released with ServerDown after the named server \
+         was request_stop'd (None = parked forever => bug reproduced)"
+    );
+}

tests/timer.rs

@@ -401,7 +401,66 @@ fn send_after_to_dead_typed_pid_is_silent() {
         assert_eq!(report_rx.recv().unwrap(), 1); // sink has now exited
         let _id = send_after(Duration::from_millis(15), sink, 2);
         sleep(Duration::from_millis(45)); // let it fire against the dead pid
-        // No panic, and nothing further delivered.
-        assert_eq!(report_rx.try_recv(), Ok(None));
+        // No panic; the sink is gone, so its report sender dropped with it —
+        // closed+empty is Err (documented), which also proves nothing
+        // further was delivered.
+        assert!(report_rx.try_recv().is_err(), "nothing further delivered");
+    });
+}
+
+// ---------------------------------------------------------------------------
+// Wall-anchored send_after (RFC 007 user-facing opt-out). The API exists in
+// both feature configs; featureless it is behaviourally identical to
+// `send_after` — these tests pin exactly that.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn armed_wall_send_timer_is_returned_and_fires() {
+    let mut t = Timers::new();
+    let now = Instant::now();
+    let fired = Arc::new(AtomicBool::new(false));
+    let f = fired.clone();
+    let _id = t.insert_send_wall(
+        now + Duration::from_millis(10),
+        Pid::new(0, 0),
+        Box::new(move || f.store(true, Ordering::SeqCst)),
+    );
+
+    let mut due = t.pop_due(now + Duration::from_millis(20));
+    assert_eq!(due.len(), 1, "an armed wall send timer should pop when due");
+    run_fire(due.pop().unwrap());
+    assert!(fired.load(Ordering::SeqCst), "running the thunk delivers");
+    assert!(t.is_empty());
+}
+
+use smarm::send_after_named_wall;
+
+#[test]
+fn send_after_named_wall_delivers_after_the_delay() {
+    const WPING: Name<u64> = Name::new("send_after_wall_ping");
+    run(|| {
+        let (tx, rx) = channel::<u64>();
+        register(WPING, tx).unwrap();
+        let t0 = Instant::now();
+        let _id = send_after_named_wall(Duration::from_millis(30), WPING, 99);
+        assert_eq!(rx.recv().unwrap(), 99);
+        assert!(
+            t0.elapsed() >= Duration::from_millis(25),
+            "delivered too early: {:?}",
+            t0.elapsed()
+        );
+    });
+}
+
+#[test]
+fn send_after_named_wall_cancels() {
+    const WC: Name<u64> = Name::new("send_after_wall_cancel");
+    run(|| {
+        let (tx, rx) = channel::<u64>();
+        register(WC, tx).unwrap();
+        let id = send_after_named_wall(Duration::from_millis(50), WC, 7);
+        assert!(cancel_timer(id), "cancel before fire returns true");
+        sleep(Duration::from_millis(90));
+        assert_eq!(rx.try_recv(), Ok(None), "cancelled wall timer delivered");
     });
 }