fix(rfc-004): gate spinning on max_spinners>0, killing wake-nobody futex_wake at N=1

`spinning` was `spin_budget_cycles > 0` alone. At N=1 the pool-size cap is 0, so
spinning was enabled while no scheduler could ever enlist: n_spinning pinned at 0,
so the enqueue submit-rule guard (n_spinning==0) fired a wake-nobody futex_wake on
every push. Now gated on `budget > 0 && max_spinners > 0`; a debug_assert guards
the invariant against regression. No behavioural change (a 0-cap runtime never
parks on the futex either way) — single_thread_spinning_is_inert and the AllDone
liveness test (shutdown_releases_pinned_parked_siblings) still pass.

switch_cost N=1 (1-core sandbox): p50 ~466ns -> ~186ns. The wake-nobody syscall
was ~60% of the round-trip in wall-clock terms; the handoff's "~12%" was perf
self-time, not latency share. ROADMAP per-switch section updated to fold in the
N=1 spike correction (shim hypothesis is many-core only). Dropped the throwaway
budget_probe bench (switch_cost reproduces every state via .spin_budget_cycles/
.max_spinners and with percentiles).

.gitignore

@@ -2,3 +2,5 @@ target
 Cargo.lock
 smarm_trace.json
 /bench_results/
+__pycache__/
+*.pyc

Cargo.toml

@@ -61,3 +61,11 @@ harness = false
 [[bench]]
 name = "rq_runtime"
 harness = false
+
+[[bench]]
+name = "spin_sweep"
+harness = false
+
+[[bench]]
+name = "switch_cost"
+harness = false

ROADMAP.md

@@ -120,11 +120,34 @@ Needs an RFC.
 
 #### Per-switch cost (context shims, epoch protocol)
 The shootout's residual: per-wake latency is 0.16–0.18 µs at N=1 and
-0.8–1.2 µs at N=8+, dominated by the context-switch shims and the epoch
-protocol, not the queue. On current evidence this is the larger constant —
-"the whole game" alongside the v0.9 work — but there is no spec yet. Needs a
-profiling spike (where do the cycles actually go per park/unpark round-trip)
-and then an RFC before it can be scheduled.
+0.8–1.2 µs at N=8+. The N=1 profiling spike is **done** (`docs/perswitch-profile-n1.md`):
+it **revises the old shim-first framing**. At N=1 the context shims + TLS sp
+accessors are ~1% of self-time, not the dominant cost — the shim hypothesis was
+always a *many-core* one (its only evidence was the N=1→N=8 jump, blamed on TLS
+access mode and cross-core coherency on the sp/epoch words), and nothing at N=1
+on one core can confirm or refute it. The N=1 round-trip is instead dominated by
+the scheduler core (`schedule_loop` + run-queue, ~33%) plus a removable
+wake-nobody `futex_wake`.
+
+That `futex_wake` is now **fixed**: it was an interaction between the RFC 004
+spin defaults and the N=1 pool-size cap of 0 — `spinning` was gated on
+`budget > 0` alone, so at N=1 spinning was enabled while the cap forbade any
+spinner from enlisting, pinning `n_spinning` at 0 and firing a wake-nobody
+`futex_wake` on every enqueue. `spinning` is now gated on
+`budget > 0 && max_spinners > 0` (a `debug_assert` guards the invariant). On the
+1-core sandbox this cut N=1 p50 from ~466 ns to ~186 ns — the syscall was ~60%
+of the round-trip in wall-clock terms (the handoff's "~12%" was perf *self-time*,
+not latency share).
+
+Two separable targets remain: **(a)** the N=1 floor — `schedule_loop` + run-queue,
+the irreducible scheduler core, no obvious cheap win left; **(b)** the N→8 slope —
+shim / TLS / coherency, which needs the 5900X (HW-counter profiling; this sandbox
+has no PMU). Still "the whole game" alongside v0.9. Next code chunk is a `remote`
+mode for `benches/switch_cost.rs` (wake straddling two schedulers) — writable in
+the sandbox, only measurable on the box. Do **not** optimise the shim until the
+N=8 HW-counter data confirms it is the cost. Then an RFC; the
+`spinning`-gating fix above may also warrant a small RFC since it touches the
+RFC-004-settled futex path.
 
 #### send_after / cancel_timer
 Message-delivery timer on the existing min-heap (`timer.rs`): deliver a value to a

benches/README.md

@@ -21,6 +21,7 @@ cargo bench                          # all of them (slow; rarely what you want)
 | `tokio_favored.rs` | Workloads tokio's model is built for. Expect to lose; the value is knowing *by how much* and catching the gap widening. |
 | `rq_micro.rs` | Run-queue **structures** in isolation (no runtime, no actors): push/pop throughput sweeping thread count × producer:consumer ratio. Covers all three queue types in one binary — the types compile in every build; only the runtime's alias is feature-selected. |
 | `rq_runtime.rs` | The **whole scheduler** with the compile-time-selected queue: yield-storm (pure queue churn), ping-pong-pairs (park/unpark latency), spawn-storm (slab + free list + queue churn), sweeping scheduler count. Comparing variants requires rebuilding per `rq-*` feature. |
+| `spin_sweep.rs` | RFC 004 spinning workers: wake **latency vs idle-CPU** over `spin_budget_cycles`. Four workloads (remote-wake, fork-join, half-load, pure-idle) × budget × `max_spinners` × scheduler count; latency percentiles plus cores-busy from `getrusage`. The data that picks the budget default and on/off posture. |
 
 ## The run-queue shootout
 
@@ -55,6 +56,70 @@ cargo bench --bench rq_runtime --no-default-features --features rq-striped
 | `SMARM_BENCH_PAIRS` / `_ROUNDTRIPS` | `32` / `1000` | `rq_runtime` ping-pong |
 | `SMARM_BENCH_SPAWNS` | `5000` | `rq_runtime` spawn-storm |
 
+## The spin sweep (RFC 004)
+
+Spinning workers trade idle CPU for wake latency: an idle worker polls
+`queue_len` for up to `spin_budget_cycles` (at most `max_spinners` at once)
+before falling back to a futex park, so work that lands during the spin is
+claimed with no syscall. `spin_budget_cycles = 0` is the feature fully off —
+the historical `thread::sleep` park, exactly preserved. `spin_sweep` measures
+both sides of that trade across the budget so the curve can pick the default.
+
+```
+cargo bench --bench spin_sweep
+# on a big box (scale the work up from the sub-second sandbox defaults):
+SMARM_BENCH_THREADS="1 2 4 8 16" SMARM_BENCH_RUNS=9 \
+SMARM_SPIN_BUDGET="0 1000 5000 20000 100000 500000 2000000" \
+SMARM_SPIN_ROUNDS=2000 SMARM_SPIN_BURSTS=500 SMARM_SPIN_ITEMS=5000 \
+  cargo bench --bench spin_sweep | tee bench_results/spin.txt
+grep '^SPINCSV' bench_results/spin.txt > bench_results/spin.csv
+```
+
+Two axes per config: **latency** (p50/p90/p99/min/max, pooled across runs) and
+**cost** (cores-busy = process CPU-seconds via `getrusage(RUSAGE_SELF)` over
+wall-seconds; `1.0` = one core pegged, up to `max_spinners` when every spinner
+is hot). The four workloads:
+
+- **remote-wake** — the latency target. A consumer parks on `recv`; a producer
+  on another scheduler timestamps just before `send`. Latency is send→receipt.
+- **fork-join** — fan-out wake latency: root forks `FANOUT` actors, joins them,
+  gap, repeat.
+- **half-load** — the cost target. Work dispatched every `PERIOD_US`, so workers
+  idle a fraction of each period; cores-busy climbs with budget as spin fills
+  the gaps. Spin cost scales with **park frequency** × budget, so this (not
+  pure-idle) is where the cost curve lives.
+- **pure-idle** — a floor check: one actor sleeps, nothing else runs. Confirms
+  budget 0 stays parked and a large budget doesn't leak *continuous* CPU (a
+  spinner spends its budget once per park, so a truly idle window barely moves).
+
+The crossover the sweep reveals: spinning only wins when work arrives while a
+worker is still inside its budget. With a fixed inter-arrival gap, budgets whose
+spin-time (`budget_cycles / TSC_freq`) exceeds the gap start catching work
+before the park — so `SMARM_SPIN_GAP_US` sets the knee; tune it against your
+box's TSC frequency.
+
+### Knobs (env vars, all optional)
+
+| var | default | used by |
+|---|---|---|
+| `SMARM_BENCH_THREADS` | `"1 2"` | scheduler-count sweep |
+| `SMARM_BENCH_RUNS` | `3` | runs per config (latency pooled, cores median) |
+| `SMARM_SPIN_BUDGET` | `"0 1000 10000 100000 1000000"` | `spin_budget_cycles` sweep |
+| `SMARM_MAX_SPINNERS` | `"-"` | `max_spinners` sweep; `-` = runtime default (N/2) |
+| `SMARM_SPIN_GAP_US` | `150` | idle gap between rounds/bursts (sets the latency knee) |
+| `SMARM_SPIN_ROUNDS` | `100` | remote-wake rounds |
+| `SMARM_SPIN_FANOUT` / `_BURSTS` | `64` / `30` | fork-join |
+| `SMARM_SPIN_ITEMS` / `_PERIOD_US` | `150` / `300` | half-load |
+| `SMARM_SPIN_WINDOW_MS` | `25` | pure-idle window |
+
+Output is the house table plus a greppable `SPINCSV,...` line per config.
+
+**On 1 core spinning is pathological** — a hot spinner starves the very thread
+that would enqueue work, so the optimal budget there is 0 and you only ever see
+the cost, never the benefit (the sweep will show remote-wake latency getting
+*worse* with budget on one core). One core validates the harness; the real curve
+comes from the many-core box.
+
 ## Reading the numbers honestly
 
 - **Core count is the experiment.** On a 1-core machine (CI, sandboxes) the

benches/spin_sweep.rs

@@ -0,0 +1,435 @@
+//! RFC 004 spinning-worker measurement tool — the latency-vs-idle-CPU curve
+//! over `spin_budget_cycles`.
+//!
+//! Spinning workers trade idle CPU for wake latency: an idle worker burns
+//! cycles polling `queue_len` (up to `spin_budget_cycles`, capped at
+//! `max_spinners` concurrent spinners) so that incoming work is claimed with
+//! no futex syscall. `spin_budget_cycles = 0` is the feature fully off — the
+//! historical `thread::sleep` park, exactly preserved. This binary measures
+//! both sides of that trade so the sweep can pick the budget default (RFC 004
+//! Q3) and the on/off posture (Q4).
+//!
+//! Workloads — each one is a fresh runtime per sample (no cross-contamination):
+//!
+//!   remote-wake   — a consumer blocks on recv (its worker idles); a producer
+//!                   on another scheduler timestamps just before send. Latency
+//!                   = send -> receipt, the quantity spinning is meant to cut.
+//!                   A fixed gap (SMARM_SPIN_GAP_US) between rounds lets the
+//!                   consumer re-idle each round. THE LATENCY TARGET.
+//!   fork-join     — root forks FANOUT trivial actors and joins them (one
+//!                   burst), gap, repeat. Latency = fork -> all-joined; stresses
+//!                   the fan-out wake. Secondary latency target.
+//!   half-load     — work dispatched every PERIOD_US for ITEMS items, so
+//!                   workers idle a fraction of each period. THE COST TARGET:
+//!                   cores-busy climbs with budget as spin fills the idle gaps.
+//!   pure-idle     — one actor sleeps WINDOW_MS, nothing else runs. The cleanest
+//!                   single picture of spin burn: cores-busy is ~0 at budget 0
+//!                   and climbs toward max_spinners as the budget grows.
+//!
+//! Two axes, reported together per config:
+//!   latency  — p50 / p90 / p99 / min / max over the POOLED rounds of every run
+//!              (pooling gives a stabler tail than a median-of-percentiles).
+//!   cost     — cores-busy = process CPU-seconds (getrusage RUSAGE_SELF, all
+//!              runtime threads) / wall-seconds over the measured window. 1.0 =
+//!              one core pegged; approaches max_spinners when every spinner is
+//!              hot. Reported as the median over runs.
+//!
+//! The crossover the sweep reveals: spinning only wins when work arrives while
+//! a worker is still inside its budget. With a fixed inter-arrival gap, that
+//! means budgets whose spin-time (budget_cycles / TSC_freq) exceeds the gap
+//! start catching work before the park; smaller budgets park and pay the full
+//! wake. So GAP_US sets the knee — tune it against your box's TSC frequency.
+//!
+//! Knobs (env):
+//!   SMARM_BENCH_THREADS   scheduler-count sweep            default "1 2"
+//!   SMARM_BENCH_RUNS      runs per config (pooled/median)  default 3
+//!   SMARM_SPIN_BUDGET     spin_budget_cycles sweep         default "0 1000 10000 100000 1000000"
+//!   SMARM_MAX_SPINNERS    max_spinners sweep; "-" = runtime default (N/2)  default "-"
+//!   SMARM_SPIN_GAP_US     idle gap between rounds/bursts    default 150
+//!   SMARM_SPIN_ROUNDS     remote-wake rounds                default 100
+//!   SMARM_SPIN_FANOUT     fork-join actors per burst        default 64
+//!   SMARM_SPIN_BURSTS     fork-join bursts                  default 30
+//!   SMARM_SPIN_ITEMS      half-load dispatch count          default 150
+//!   SMARM_SPIN_PERIOD_US  half-load inter-dispatch period   default 300
+//!   SMARM_SPIN_WINDOW_MS  pure-idle window                  default 25
+//!
+//! Output: house table + one greppable line per config:
+//!   SPINCSV,<variant>,<workload>,<threads>,<budget>,<spinners>,<gap_us>,
+//!           <n>,<p50_us>,<p90_us>,<p99_us>,<min_us>,<max_us>,<cores_busy>,<ops_s>
+//! ("spinners" is the literal "default" when SMARM_MAX_SPINNERS is "-".)
+//!
+//! NOTE: on a 1-core box (CI, sandboxes) spinning is *pathological* — a hot
+//! spinner starves the very thread that would enqueue work, so the optimal
+//! budget there is 0 and you only ever see the cost, never the benefit. This
+//! binary on one core validates the harness and catches gross pathologies; the
+//! real latency-vs-idle-CPU curve comes from the many-core box.
+
+use smarm::runtime::{init, Config};
+use smarm::{channel, sleep, spawn};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+// --------------------------------------------------------------------------
+// env helpers (house style, matching rq_runtime.rs)
+// --------------------------------------------------------------------------
+
+fn variant() -> &'static str {
+    if cfg!(feature = "rq-mpmc") {
+        "rq-mpmc"
+    } else if cfg!(feature = "rq-striped") {
+        "rq-striped"
+    } else {
+        "rq-mutex"
+    }
+}
+
+fn env_usize(key: &str, default: usize) -> usize {
+    std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
+}
+
+fn env_u64(key: &str, default: u64) -> u64 {
+    std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
+}
+
+fn env_threads() -> Vec<usize> {
+    std::env::var("SMARM_BENCH_THREADS")
+        .map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect())
+        .unwrap_or_else(|_| vec![1, 2])
+}
+
+fn env_budgets() -> Vec<u64> {
+    std::env::var("SMARM_SPIN_BUDGET")
+        .map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect())
+        .unwrap_or_else(|_| vec![0, 1_000, 10_000, 100_000, 1_000_000])
+}
+
+/// `None` means "leave the runtime default (N/2)"; encoded in env as "-".
+fn env_spinners() -> Vec<Option<u32>> {
+    std::env::var("SMARM_MAX_SPINNERS")
+        .map(|v| {
+            v.split_whitespace()
+                .map(|t| if t == "-" { None } else { t.parse().ok() })
+                .collect()
+        })
+        .unwrap_or_else(|_| vec![None])
+}
+
+fn make_config(threads: usize, budget: u64, spinners: Option<u32>) -> Config {
+    let cfg = Config::exact(threads).spin_budget_cycles(budget);
+    match spinners {
+        Some(n) => cfg.max_spinners(n),
+        None => cfg,
+    }
+}
+
+// --------------------------------------------------------------------------
+// CPU accounting — process-wide (all runtime threads) via getrusage
+// --------------------------------------------------------------------------
+
+/// Total CPU seconds (user + system) consumed by the whole process so far.
+/// RUSAGE_SELF aggregates every thread, which is exactly what "how much did
+/// the runtime burn" needs — the spinners are separate threads from the one
+/// reading this.
+fn cpu_seconds() -> f64 {
+    // SAFETY: getrusage with RUSAGE_SELF and a valid out-pointer; we check the
+    // return code and only read the (now-initialised) timeval fields.
+    let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
+    let rc = unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut ru) };
+    assert_eq!(rc, 0, "getrusage(RUSAGE_SELF) failed");
+    let tv = |t: libc::timeval| t.tv_sec as f64 + t.tv_usec as f64 * 1e-6;
+    tv(ru.ru_utime) + tv(ru.ru_stime)
+}
+
+// --------------------------------------------------------------------------
+// One sample = one fresh runtime run
+// --------------------------------------------------------------------------
+
+struct Sample {
+    /// Per-event latencies in nanoseconds (empty for pure-idle).
+    lat_ns: Vec<u64>,
+    /// CPU-seconds / wall-seconds over the measured window.
+    cores_busy: f64,
+    /// Completed work units (for ops/s); 0 where not meaningful.
+    ops: u64,
+    /// Wall-seconds of the measured window (for ops/s).
+    wall_s: f64,
+}
+
+/// Shared scratch the root closure writes and the caller reads after `run()`
+/// returns. (`run` takes a `FnOnce` and yields no value, so we hand results
+/// out through an Arc rather than a return.)
+#[derive(Default)]
+struct Scratch {
+    lat_ns: Vec<u64>,
+    cpu_delta: f64,
+    wall_s: f64,
+    ops: u64,
+}
+
+fn finish(scratch: Arc<Mutex<Scratch>>) -> Sample {
+    let mut s = scratch.lock().unwrap();
+    let cores_busy = if s.wall_s > 0.0 { s.cpu_delta / s.wall_s } else { 0.0 };
+    Sample {
+        lat_ns: std::mem::take(&mut s.lat_ns),
+        cores_busy,
+        ops: s.ops,
+        wall_s: s.wall_s,
+    }
+}
+
+fn remote_wake(threads: usize, budget: u64, spinners: Option<u32>, rounds: usize, gap: Duration) -> Sample {
+    let rt = init(make_config(threads, budget, spinners));
+    let scratch = Arc::new(Mutex::new(Scratch::default()));
+    let out = scratch.clone();
+    rt.run(move || {
+        let (tx, rx) = channel::channel::<Instant>();
+        let lat_sink = out.clone();
+        // Consumer parks on recv; its worker idles (spins or parks per budget)
+        // until each send wakes it. It records the wake latency itself.
+        let consumer = spawn(move || {
+            let mut lats = Vec::with_capacity(rounds);
+            for _ in 0..rounds {
+                let sent = rx.recv().expect("remote-wake recv");
+                lats.push(sent.elapsed().as_nanos() as u64);
+            }
+            lat_sink.lock().unwrap().lat_ns = lats;
+        });
+        // Producer: sleep the gap (so the consumer's worker re-idles), then
+        // timestamp and send. The send is the wake event under measurement.
+        let cpu0 = cpu_seconds();
+        let w0 = Instant::now();
+        for _ in 0..rounds {
+            sleep(gap);
+            tx.send(Instant::now()).expect("remote-wake send");
+        }
+        let wall = w0.elapsed().as_secs_f64();
+        let cpu = cpu_seconds() - cpu0;
+        let _ = consumer.join();
+        let mut s = out.lock().unwrap();
+        s.cpu_delta = cpu;
+        s.wall_s = wall;
+        s.ops = rounds as u64;
+    });
+    finish(scratch)
+}
+
+fn fork_join(threads: usize, budget: u64, spinners: Option<u32>, fanout: usize, bursts: usize, gap: Duration) -> Sample {
+    let rt = init(make_config(threads, budget, spinners));
+    let scratch = Arc::new(Mutex::new(Scratch::default()));
+    let out = scratch.clone();
+    rt.run(move || {
+        let mut lats = Vec::with_capacity(bursts);
+        let cpu0 = cpu_seconds();
+        let w0 = Instant::now();
+        for _ in 0..bursts {
+            // Gap lets the workers park between bursts, so the fan-out has real
+            // wakes to do rather than landing on already-hot workers.
+            sleep(gap);
+            let t = Instant::now();
+            let handles: Vec<_> = (0..fanout).map(|_| spawn(|| {})).collect();
+            for h in handles {
+                let _ = h.join();
+            }
+            lats.push(t.elapsed().as_nanos() as u64);
+        }
+        let wall = w0.elapsed().as_secs_f64();
+        let cpu = cpu_seconds() - cpu0;
+        let mut s = out.lock().unwrap();
+        s.lat_ns = lats;
+        s.cpu_delta = cpu;
+        s.wall_s = wall;
+        s.ops = (bursts * fanout) as u64;
+    });
+    finish(scratch)
+}
+
+fn half_load(threads: usize, budget: u64, spinners: Option<u32>, items: usize, period: Duration) -> Sample {
+    let rt = init(make_config(threads, budget, spinners));
+    let scratch = Arc::new(Mutex::new(Scratch::default()));
+    let out = scratch.clone();
+    rt.run(move || {
+        let cpu0 = cpu_seconds();
+        let w0 = Instant::now();
+        // Dispatch one trivial actor every `period`. Between dispatches the
+        // workers have nothing to do, so they idle for a fraction of each
+        // period — that idle fraction is what spinning fills (or doesn't).
+        let mut handles = Vec::with_capacity(items);
+        for i in 0..items {
+            let target = period * i as u32;
+            let elapsed = w0.elapsed();
+            if target > elapsed {
+                sleep(target - elapsed);
+            }
+            handles.push(spawn(|| {}));
+        }
+        for h in handles {
+            let _ = h.join();
+        }
+        let wall = w0.elapsed().as_secs_f64();
+        let cpu = cpu_seconds() - cpu0;
+        let mut s = out.lock().unwrap();
+        s.cpu_delta = cpu;
+        s.wall_s = wall;
+        s.ops = items as u64;
+    });
+    finish(scratch)
+}
+
+fn pure_idle(threads: usize, budget: u64, spinners: Option<u32>, window: Duration) -> Sample {
+    let rt = init(make_config(threads, budget, spinners));
+    let scratch = Arc::new(Mutex::new(Scratch::default()));
+    let out = scratch.clone();
+    rt.run(move || {
+        // One actor sleeps; every other worker thread has nothing to do and
+        // spins-or-parks per budget for the whole window.
+        let cpu0 = cpu_seconds();
+        let w0 = Instant::now();
+        sleep(window);
+        let wall = w0.elapsed().as_secs_f64();
+        let cpu = cpu_seconds() - cpu0;
+        let mut s = out.lock().unwrap();
+        s.cpu_delta = cpu;
+        s.wall_s = wall;
+        s.ops = 0;
+    });
+    finish(scratch)
+}
+
+// --------------------------------------------------------------------------
+// stats
+// --------------------------------------------------------------------------
+
+/// Nearest-rank percentile over an already-sorted slice. `p` in [0, 100].
+fn pct(sorted: &[u64], p: f64) -> u64 {
+    if sorted.is_empty() {
+        return 0;
+    }
+    let idx = ((p / 100.0) * (sorted.len() - 1) as f64).round() as usize;
+    sorted[idx.min(sorted.len() - 1)]
+}
+
+fn median_f64(xs: &mut [f64]) -> f64 {
+    if xs.is_empty() {
+        return 0.0;
+    }
+    xs.sort_by(|a, b| a.partial_cmp(b).unwrap());
+    xs[xs.len() / 2]
+}
+
+struct Aggregate {
+    n: usize,
+    p50_us: f64,
+    p90_us: f64,
+    p99_us: f64,
+    min_us: f64,
+    max_us: f64,
+    cores_busy: f64,
+    ops_s: f64,
+}
+
+/// Pool latencies across all runs (stabler tail); take the median cores-busy
+/// and median ops/s across runs.
+fn aggregate(samples: Vec<Sample>) -> Aggregate {
+    let mut pooled: Vec<u64> = Vec::new();
+    let mut cores: Vec<f64> = Vec::new();
+    let mut ops_s: Vec<f64> = Vec::new();
+    for s in &samples {
+        pooled.extend_from_slice(&s.lat_ns);
+        cores.push(s.cores_busy);
+        if s.wall_s > 0.0 && s.ops > 0 {
+            ops_s.push(s.ops as f64 / s.wall_s);
+        }
+    }
+    pooled.sort_unstable();
+    let ns_to_us = |ns: u64| ns as f64 / 1000.0;
+    Aggregate {
+        n: pooled.len(),
+        p50_us: ns_to_us(pct(&pooled, 50.0)),
+        p90_us: ns_to_us(pct(&pooled, 90.0)),
+        p99_us: ns_to_us(pct(&pooled, 99.0)),
+        min_us: pooled.first().map(|&v| ns_to_us(v)).unwrap_or(0.0),
+        max_us: pooled.last().map(|&v| ns_to_us(v)).unwrap_or(0.0),
+        cores_busy: median_f64(&mut cores),
+        ops_s: median_f64(&mut ops_s),
+    }
+}
+
+// --------------------------------------------------------------------------
+// driver
+// --------------------------------------------------------------------------
+
+fn main() {
+    let threads_sweep = env_threads();
+    let budgets = env_budgets();
+    let spinners_sweep = env_spinners();
+    let runs = env_usize("SMARM_BENCH_RUNS", 3);
+    let gap = Duration::from_micros(env_u64("SMARM_SPIN_GAP_US", 150));
+    let rounds = env_usize("SMARM_SPIN_ROUNDS", 100);
+    let fanout = env_usize("SMARM_SPIN_FANOUT", 64);
+    let bursts = env_usize("SMARM_SPIN_BURSTS", 30);
+    let items = env_usize("SMARM_SPIN_ITEMS", 150);
+    let period = Duration::from_micros(env_u64("SMARM_SPIN_PERIOD_US", 300));
+    let window = Duration::from_millis(env_u64("SMARM_SPIN_WINDOW_MS", 25));
+
+    // (name, closure producing one Sample for a (threads,budget,spinners) point)
+    type Workload = (&'static str, Box<dyn Fn(usize, u64, Option<u32>) -> Sample>);
+    let workloads: Vec<Workload> = vec![
+        ("remote-wake", Box::new(move |t, b, s| remote_wake(t, b, s, rounds, gap))),
+        ("fork-join", Box::new(move |t, b, s| fork_join(t, b, s, fanout, bursts, gap))),
+        ("half-load", Box::new(move |t, b, s| half_load(t, b, s, items, period))),
+        ("pure-idle", Box::new(move |t, b, s| pure_idle(t, b, s, window))),
+    ];
+
+    let width = 122;
+    println!("\n{}", "=".repeat(width));
+    println!(
+        "  spin sweep — variant={}, runs={runs} (latency pooled, cores median), gap={}µs",
+        variant(),
+        gap.as_micros()
+    );
+    println!("{}", "=".repeat(width));
+    println!(
+        "{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7} | {:>10}",
+        "workload", "thr", "budget", "spinners", "n", "p50 µs", "p90 µs", "p99 µs", "min µs", "max µs", "cores", "ops/s"
+    );
+    println!("{}", "-".repeat(width));
+
+    for (name, run_one) in &workloads {
+        for &threads in &threads_sweep {
+            for &budget in &budgets {
+                for &spinners in &spinners_sweep {
+                    let samples: Vec<Sample> =
+                        (0..runs).map(|_| run_one(threads, budget, spinners)).collect();
+                    let a = aggregate(samples);
+                    let sp_str = match spinners {
+                        Some(n) => n.to_string(),
+                        None => "default".to_string(),
+                    };
+                    // Latency columns are blank for the no-latency probe.
+                    let has_lat = a.n > 0;
+                    let lat = |v: f64| if has_lat { format!("{v:.2}") } else { "-".to_string() };
+                    println!(
+                        "{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7.3} | {:>10}",
+                        name,
+                        threads,
+                        budget,
+                        sp_str,
+                        a.n,
+                        lat(a.p50_us),
+                        lat(a.p90_us),
+                        lat(a.p99_us),
+                        lat(a.min_us),
+                        lat(a.max_us),
+                        a.cores_busy,
+                        if a.ops_s > 0.0 { format!("{:.0}", a.ops_s) } else { "-".to_string() },
+                    );
+                    println!(
+                        "SPINCSV,{},{},{},{},{},{},{},{:.3},{:.3},{:.3},{:.3},{:.3},{:.4},{:.0}",
+                        variant(), name, threads, budget, sp_str, gap.as_micros(),
+                        a.n, a.p50_us, a.p90_us, a.p99_us, a.min_us, a.max_us, a.cores_busy, a.ops_s
+                    );
+                }
+            }
+        }
+    }
+}

benches/sweep.py

@@ -243,6 +243,90 @@ def check_regressions(current: dict, baseline: dict) -> bool:
 # Pretty print
 # ---------------------------------------------------------------------------
 
+def _threads(label: str) -> int | None:
+    """Worker-thread count implied by a runtime label.
+
+    tokio's `current_thread` is a single-threaded executor (1); an explicit
+    `multi N-thread` is N; a bare `multi-thread` has no count (None) — tokio's
+    default work-stealing pool, paired against smarm's widest config.
+    """
+    if "current_thread" in label:
+        return 1
+    m = re.search(r"(\d+)-thread", label)
+    return int(m.group(1)) if m else None
+
+
+def vs_tokio(results: dict) -> list[tuple]:
+    """Per bench, like-for-like smarm-vs-tokio rows matched by thread count.
+
+    Exact thread-count matches are paired directly (smarm 1-thread vs tokio
+    current_thread, smarm 4-thread vs tokio multi 4-thread, …). tokio's bare
+    `multi-thread` (no explicit count) is paired against the widest unmatched
+    smarm multi-thread config. Lower median µs = faster; ratio = tokio_med /
+    smarm_med, so ratio > 1 means smarm is that many times faster.
+
+    Returns rows of (bench, smarm_label, smarm_med, tokio_label, tokio_med,
+    ratio, winner). Benches without a comparable pair are skipped.
+    """
+    rows: list[tuple] = []
+    for bench, runtimes in sorted(results.items()):
+        smarm: dict[int, tuple[str, int]] = {}
+        tokio: dict[int, tuple[str, int]] = {}
+        tokio_default: tuple[str, int] | None = None  # bare 'multi-thread'
+        for label, data in runtimes.items():
+            n = _threads(label)
+            if label.startswith("smarm"):
+                if n is not None:
+                    smarm[n] = (label, data["median"])
+            elif label.startswith("tokio"):
+                if n is None:
+                    tokio_default = (label, data["median"])
+                else:
+                    tokio[n] = (label, data["median"])
+
+        def row(s: tuple[str, int], t: tuple[str, int]):
+            s_label, s_med = s
+            t_label, t_med = t
+            if s_med == 0:
+                return None
+            ratio = t_med / s_med
+            return (bench, s_label, s_med, t_label, t_med, ratio,
+                    "smarm" if ratio >= 1.0 else "tokio")
+
+        matched: set[int] = set()
+        for n in sorted(set(smarm) & set(tokio)):
+            r = row(smarm[n], tokio[n])
+            if r:
+                rows.append(r)
+                matched.add(n)
+        # tokio's default multi pool vs the widest smarm config not already paired.
+        if tokio_default is not None:
+            rem = [n for n in smarm if n not in matched and n > 1]
+            if rem:
+                r = row(smarm[max(rem)], tokio_default)
+                if r:
+                    rows.append(r)
+    return rows
+
+
+def print_vs_tokio(results: dict) -> None:
+    """Human summary + greppable VSTOKIO lines (best smarm vs best tokio)."""
+    rows = vs_tokio(results)
+    if not rows:
+        return
+    print("\n  vs tokio  (like-for-like by thread count; ratio>1 = smarm faster, lower µs better)")
+    print(f"  {'-'*78}")
+    for bench, s_label, s_med, t_label, t_med, ratio, winner in rows:
+        print(
+            f"  {bench:<22} {s_label} {s_med}µs  vs  {t_label} {t_med}µs"
+            f"  →  {ratio:.2f}x ({winner})"
+        )
+    # Machine-readable, one line per bench:
+    #   VSTOKIO,<bench>,<smarm_label>,<smarm_us>,<tokio_label>,<tokio_us>,<ratio>,<winner>
+    for bench, s_label, s_med, t_label, t_med, ratio, winner in rows:
+        print(f"VSTOKIO,{bench},{s_label},{s_med},{t_label},{t_med},{ratio:.3f},{winner}")
+
+
 def print_results(results: dict, label: str = "") -> None:
     if label:
         print(f"\n{'='*70}")
@@ -257,6 +341,7 @@ def print_results(results: dict, label: str = "") -> None:
                 f"  {rt_label:>28} | {data['result']:>10} | "
                 f"{data['median']:>10} | {data['min']:>8} | {data['max']:>8}"
             )
+    print_vs_tokio(results)
 
 
 def print_sweep_table(sweep_results: list[tuple[int, int, dict]]) -> None:

benches/switch_cost.rs

@@ -0,0 +1,256 @@
+//! Per-switch (context-switch) cost microbench — the profiling-spike harness
+//! for the ROADMAP "Per-switch cost (context shims, epoch protocol)" item.
+//!
+//! The spin work (RFC 004) is closed; the next perf target is the per-switch
+//! cost itself. Shootout evidence: per-wake latency is ~0.16–0.18µs at N=1 but
+//! ~0.8–1.2µs at N=8+, and the residual is attributed to the context-switch
+//! shims (`src/context.rs`) and the epoch protocol — NOT the queue. This binary
+//! isolates that round-trip so the cycles can be attributed under `perf` and an
+//! rdtsc bracket, feeding the RFC.
+//!
+//! WHAT THE ROUND-TRIP IS
+//!
+//! `yield_now()` from inside an actor does exactly one park/unpark round-trip
+//! with nothing else attached:
+//!
+//!     actor: switch_to_scheduler  ──►  scheduler re-queues the actor (slot-word
+//!            (context.rs shim)          epoch/state transition, run_queue push),
+//!                                       pops it straight back, switch_to_actor
+//!     actor resumes  ◄──────────────────────────────────────────────────────
+//!
+//! No IO thread traffic, no channel, no timer, no cross-thread wake. On a
+//! single-scheduler runtime the re-queue+repop never leaves this core, so the
+//! sample is the *pure* shim + epoch + queue-op cost with zero coherency
+//! traffic. That is the `local` baseline; a `remote` mode (wake straddling two
+//! schedulers, to expose the N=1→N=8 coherency/TLS-mode jump) is a deliberate
+//! follow-up and is NOT in this file yet — local first, per the spike plan.
+//!
+//! TWO LENSES ON THE SAME LOOP
+//!
+//!   wall    — `Instant` bracket per round-trip. Source of truth for µs, directly
+//!             comparable to the shootout's per-wake latency numbers.
+//!   cycles  — `rdtsc` bracket per round-trip. Source of truth for the cycle
+//!             budget the RFC will reason in (the spin budget is in cycles too).
+//!
+//! Reporting both lets us *derive* the effective TSC frequency (cycles/ns) from
+//! the same samples instead of hardcoding a nominal 3.7GHz — the spin_sweep
+//! lesson was that nominal-vs-actual TSC drift is exactly what produces
+//! red-herring numbers. If the derived freq matches the box's known base clock,
+//! the two lenses corroborate; if not, that mismatch is itself a finding.
+//!
+//! Knobs (env):
+//!   SMARM_SWITCH_ROUNDS   round-trips timed per run     default 200000
+//!   SMARM_SWITCH_WARMUP   untimed warmup round-trips     default 10000
+//!   SMARM_SWITCH_RUNS     runs (pooled latency, median)  default 5
+//!
+//! Output: house table + one greppable line per run-set:
+//!   SWITCHCSV,<variant>,<mode>,<rounds>,<runs>,<n>,<p50_ns>,<p90_ns>,<p99_ns>,
+//!            <min_ns>,<max_ns>,<mean_ns>,<mean_cyc>,<derived_ghz>
+//!
+//! NOTE: a single yielding actor is the cooperative-scheduling tightest loop —
+//! it never parks on a futex (the work is always immediately re-queued), so
+//! this measures the switch+epoch+queue path, NOT the futex park. That is
+//! intentional: the futex park is the spin work's territory (RFC 004), already
+//! characterised. The unattributed constant the shootout flagged lives in the
+//! switch itself, which is what this loop hammers.
+
+use smarm::runtime::{init, Config};
+use smarm::{run, spawn, yield_now};
+use std::sync::{Arc, Mutex};
+
+// --------------------------------------------------------------------------
+// env helpers (house style, matching spin_sweep.rs / rq_runtime.rs)
+// --------------------------------------------------------------------------
+
+fn variant() -> &'static str {
+    if cfg!(feature = "rq-mpmc") {
+        "rq-mpmc"
+    } else if cfg!(feature = "rq-striped") {
+        "rq-striped"
+    } else {
+        "rq-mutex"
+    }
+}
+
+fn env_usize(key: &str, default: usize) -> usize {
+    std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
+}
+
+// --------------------------------------------------------------------------
+// rdtsc — serialised so the bracket actually fences the round-trip.
+//
+// Plain `rdtsc` can be reordered around the work by an out-of-order core, which
+// would smear the bracket. `rdtscp` retires prior instructions before reading
+// the counter, and the trailing `lfence` blocks later instructions from
+// climbing above the second read. Pair = (rdtscp; lfence) … work … (rdtscp;
+// lfence): a standard cycle-accurate bracket. We read TSC_AUX too but ignore
+// it; the point is the ordering guarantee, not the core id.
+// --------------------------------------------------------------------------
+
+#[inline(always)]
+fn rdtsc_serialised() -> u64 {
+    #[cfg(target_arch = "x86_64")]
+    unsafe {
+        let mut aux = 0u32;
+        let t = core::arch::x86_64::__rdtscp(&mut aux);
+        core::arch::x86_64::_mm_lfence();
+        t
+    }
+    #[cfg(not(target_arch = "x86_64"))]
+    {
+        // Non-x86 fallback: nanosecond clock standing in for cycles. The derived
+        // "GHz" column then reads ~1.0 and is meaningless, but the wall lens and
+        // the harness still work. The spike target box is x86-64.
+        use std::time::Instant;
+        thread_local! { static T0: Instant = Instant::now(); }
+        T0.with(|t0| t0.elapsed().as_nanos() as u64)
+    }
+}
+
+// --------------------------------------------------------------------------
+// percentile / median helpers (verbatim house idiom from spin_sweep.rs)
+// --------------------------------------------------------------------------
+
+/// Nearest-rank percentile over an already-sorted slice. `p` in [0, 100].
+fn pct(sorted: &[u64], p: f64) -> u64 {
+    if sorted.is_empty() {
+        return 0;
+    }
+    let idx = ((p / 100.0) * (sorted.len() - 1) as f64).round() as usize;
+    sorted[idx.min(sorted.len() - 1)]
+}
+
+// --------------------------------------------------------------------------
+// one run: a single actor yields ROUNDS times; we bracket each yield from
+// inside the actor (the only vantage point — the actor is suspended during the
+// scheduler half, so an external timer can't see a single round-trip).
+//
+// Per iteration we capture BOTH a wall-ns delta and a TSC-cycle delta around
+// the same `yield_now()`. The loop overhead (two clock reads + a Vec push +
+// the branch) rides along in every sample equally; we subtract an empty-loop
+// self-calibration below so the reported number is the round-trip, not the
+// instrumentation.
+// --------------------------------------------------------------------------
+
+struct RunSample {
+    lat_ns: Vec<u64>,
+    cyc: Vec<u64>,
+}
+
+fn one_run(threads: usize, rounds: usize, warmup: usize) -> RunSample {
+    let out: Arc<Mutex<Option<RunSample>>> = Arc::new(Mutex::new(None));
+    let out2 = out.clone();
+
+    let cfg = Config::exact(threads);
+    init(cfg);
+
+    run(move || {
+        let h = spawn(move || {
+            // Warmup: let the actor's stack/queue slot go hot, JIT-free but
+            // cache-warm, before any sample is kept.
+            for _ in 0..warmup {
+                yield_now();
+            }
+
+            let mut lat_ns = Vec::with_capacity(rounds);
+            let mut cyc = Vec::with_capacity(rounds);
+
+            for _ in 0..rounds {
+                let w0 = std::time::Instant::now();
+                let c0 = rdtsc_serialised();
+                yield_now();
+                let c1 = rdtsc_serialised();
+                let w1 = w0.elapsed();
+                cyc.push(c1.saturating_sub(c0));
+                lat_ns.push(w1.as_nanos() as u64);
+            }
+
+            *out2.lock().unwrap() = Some(RunSample { lat_ns, cyc });
+        });
+        let _ = h.join();
+    });
+
+    let sample = out.lock().unwrap().take().expect("actor stored a sample");
+    sample
+}
+
+/// Empty-loop self-calibration: the same bracket with the `yield_now()` removed,
+/// run inline (no runtime). Gives the floor cost of two serialised clock reads +
+/// the push, in both lenses, to subtract from the round-trip samples.
+fn calibrate(rounds: usize) -> (u64, u64) {
+    let mut lat_ns = Vec::with_capacity(rounds);
+    let mut cyc = Vec::with_capacity(rounds);
+    let mut sink = 0u64;
+    for _ in 0..rounds {
+        let w0 = std::time::Instant::now();
+        let c0 = rdtsc_serialised();
+        // no yield — measure the bracket itself
+        let c1 = rdtsc_serialised();
+        let w1 = w0.elapsed();
+        sink ^= c1;
+        cyc.push(c1.saturating_sub(c0));
+        lat_ns.push(w1.as_nanos() as u64);
+    }
+    std::hint::black_box(sink);
+    cyc.sort_unstable();
+    lat_ns.sort_unstable();
+    // Use the medians as the floor — robust to the occasional interrupt.
+    (pct(&lat_ns, 50.0), pct(&cyc, 50.0))
+}
+
+fn main() {
+    let rounds = env_usize("SMARM_SWITCH_ROUNDS", 200_000);
+    let warmup = env_usize("SMARM_SWITCH_WARMUP", 10_000);
+    let runs = env_usize("SMARM_SWITCH_RUNS", 5);
+    let mode = "local";
+
+    // Calibrate the instrumentation floor once, with a healthy sample.
+    let (floor_ns, floor_cyc) = calibrate(rounds.min(50_000).max(10_000));
+
+    let mut pooled_ns: Vec<u64> = Vec::new();
+    let mut pooled_cyc: Vec<u64> = Vec::new();
+
+    for _ in 0..runs {
+        let s = one_run(1, rounds, warmup);
+        // Subtract the instrumentation floor; saturating so a sub-floor outlier
+        // (clock granularity) clamps to 0 rather than wrapping.
+        pooled_ns.extend(s.lat_ns.iter().map(|&v| v.saturating_sub(floor_ns)));
+        pooled_cyc.extend(s.cyc.iter().map(|&v| v.saturating_sub(floor_cyc)));
+    }
+
+    pooled_ns.sort_unstable();
+    pooled_cyc.sort_unstable();
+
+    let n = pooled_ns.len();
+    let mean_ns = pooled_ns.iter().map(|&v| v as f64).sum::<f64>() / n.max(1) as f64;
+    let mean_cyc = pooled_cyc.iter().map(|&v| v as f64).sum::<f64>() / n.max(1) as f64;
+    // Derived effective frequency: cycles per ns = GHz. Cross-checks the two
+    // lenses against the box's known base clock.
+    let derived_ghz = if mean_ns > 0.0 { mean_cyc / mean_ns } else { 0.0 };
+
+    let p50 = pct(&pooled_ns, 50.0);
+    let p90 = pct(&pooled_ns, 90.0);
+    let p99 = pct(&pooled_ns, 99.0);
+    let lo = *pooled_ns.first().unwrap_or(&0);
+    let hi = *pooled_ns.last().unwrap_or(&0);
+
+    // House table.
+    println!();
+    println!("per-switch cost — {} mode, variant={}", mode, variant());
+    println!(
+        "  rounds={} warmup={} runs={}  (instrumentation floor: {} ns / {} cyc, subtracted)",
+        rounds, warmup, runs, floor_ns, floor_cyc
+    );
+    println!("  {:<10} {:<10} {:<10} {:<10} {:<10}", "p50 ns", "p90 ns", "p99 ns", "min ns", "max ns");
+    println!("  {:<10} {:<10} {:<10} {:<10} {:<10}", p50, p90, p99, lo, hi);
+    println!(
+        "  mean {:.1} ns | mean {:.0} cyc | derived {:.3} GHz",
+        mean_ns, mean_cyc, derived_ghz
+    );
+
+    // Greppable line — same spirit as SPINCSV.
+    println!(
+        "SWITCHCSV,{},{},{},{},{},{},{},{},{},{},{:.1},{:.0},{:.3}",
+        variant(), mode, rounds, runs, n, p50, p90, p99, lo, hi, mean_ns, mean_cyc, derived_ghz
+    );
+}

docs/perswitch-profile-n1.md

@@ -0,0 +1,77 @@
+# Per-switch cost — N=1 local profile (spike findings)
+
+Measured with `benches/switch_cost.rs` (local mode: one actor, one scheduler,
+tight `yield_now()` loop = one park/unpark round-trip with no IO/channel/timer
+and no cross-core traffic). Sandbox: 1 core, kernel 6.18, **no PMU** (hardware
+counters unavailable), so attribution is from `perf record -e task-clock`
+(software timer sampling) plus the bench's own rdtsc/wall brackets.
+
+## Numbers (stable across runs)
+
+- Round-trip p50 ≈ **303 ns ≈ 828 cyc** (instrumentation floor subtracted).
+- Derived effective clock ≈ **2.73 GHz** (rdtsc cyc / wall ns — the two lenses
+  corroborate, so the cycle counts are trustworthy).
+- p90 316 ns, p99 472 ns; max is a multi-ms OS-deschedule outlier (1 shared
+  core) — ignore the max, trust the percentiles (the harness pools them).
+
+## Attribution (perf task-clock, self-time, 28.6k samples / 12M round-trips)
+
+| share | symbol | bucket |
+|------:|--------|--------|
+| 24.8% | `runtime::schedule_loop` | scheduler logic (slot-word/epoch + dispatch) |
+|  8.6% | `MutexQueue::push`/`pop`/`len` | run-queue ops |
+| ~12%  | `do_syscall_64`+`syscall`+`futex_*` | **futex_wake on the hot path** |
+|  3.5% | `IoThread::drain_completions` | the always-on IO thread (`run()` starts one) |
+| ~30%  | `main` + `clock_gettime`/Timespec + `quicksort` | **instrumentation** (timing + percentile sort) |
+|  ~1%  | `switch_to_scheduler`+`switch_to_actor_asm`+ sp accessors | **the context shims + TLS** |
+
+## Headline finding — revises the handoff hypothesis
+
+The handoff named the **context shims** (`context.rs`: two `call`s into the
+TLS sp accessors per switch) as the prime suspect for the per-switch cost.
+**At N=1 that is not where the time goes — the shims + TLS are ~1% of
+self-time.** The N=1 cost is dominated by:
+
+1. **`schedule_loop` + run-queue ops (~33%)** — the epoch/slot-word transition
+   and the mutex run-queue push/pop on every re-queue.
+2. **A `futex_wake` syscall (~12%)** firing on the hot path even though nothing
+   is parked. This is the **submit-rule wake** the handoff itself flagged (RFC
+   004 finding #1: "a parallelism/latency optimisation, NOT a liveness guard").
+   In a single-scheduler always-runnable loop it is pure cost — no one is ever
+   parked to wake. **First cheap-win candidate: suppress the submit wake when
+   there is no parked waiter** (a parked-count / n_spinning check before the
+   syscall). Needs care — finding #2's AllDone broadcast is the liveness-critical
+   one and must NOT be touched.
+
+## What this does and does NOT show
+
+- The handoff's shim hypothesis was a **many-core** hypothesis: its evidence was
+  the N=1→N=8 jump (0.18→1.2µs), attributed to TLS access mode (`__tls_get_addr`
+  vs `#[thread_local]`) and cross-core coherency on the sp/epoch words. **None of
+  that is observable at N=1 on one core.** This profile does NOT refute it; it
+  establishes that the shim is cheap *until cores contend*.
+- So the spike question sharpens into two separable costs:
+  - **N=1 floor:** scheduler logic + a likely-removable futex_wake. Actionable now.
+  - **N→8 slope:** the shim/TLS/coherency cost. Needs the many-core box + a
+    `remote` bench mode (wake straddling two schedulers) + hardware PMU counters
+    (cache-misses, `MEM_LOAD…HITM` for coherency) — none available in this sandbox.
+
+## Reproduce
+
+```sh
+. "$HOME/.cargo/env"
+cargo build --release --bench switch_cost
+BIN=$(ls -t target/release/deps/switch_cost-* | grep -v '\.d$' | head -1)
+PERF=/usr/lib/linux-tools-6.8.0-124/perf   # 6.8 perf on 6.18 kernel; sw events only here
+
+# bench alone (numbers):
+SMARM_SWITCH_ROUNDS=3000000 SMARM_SWITCH_WARMUP=50000 SMARM_SWITCH_RUNS=4 "$BIN"
+
+# attribution (sw task-clock; HW counters need a real PMU / the 5900X):
+SMARM_SWITCH_ROUNDS=3000000 "$PERF" record -F 4000 -g --call-graph fp -o /tmp/switch.data -- "$BIN"
+"$PERF" report -i /tmp/switch.data --stdio --no-children
+```
+
+On the 5900X with a real PMU, drop `-e task-clock` for `-e cycles,instructions,
+cache-misses,mem_load_retired.l3_miss` to get the coherency picture the N=8 case
+needs.

src/raw_mutex.rs

@@ -242,7 +242,7 @@ impl<T> Drop for RawMutexGuard<'_, T> {
 // futex (x86-64 Linux; master is x86-only, see arm-port branch)
 // ---------------------------------------------------------------------------
 
-fn futex_wait(state: &AtomicU32, expected: u32) {
+pub(crate) fn futex_wait(state: &AtomicU32, expected: u32) {
     // SAFETY: `state` is a valid, aligned u32 for the duration of the call.
     // Spurious wakeups and EAGAIN (value already changed) are both handled by
     // the caller's retry loop.
@@ -257,7 +257,7 @@ fn futex_wait(state: &AtomicU32, expected: u32) {
     }
 }
 
-fn futex_wake(state: &AtomicU32, n: i32) {
+pub(crate) fn futex_wake(state: &AtomicU32, n: i32) {
     // SAFETY: as above.
     unsafe {
         libc::syscall(

src/runtime.rs

@@ -118,8 +118,9 @@ use crate::timer::Timers;
 use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor};
 
 use std::sync::atomic::{
-    AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering,
+    fence, AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering,
 };
+use crate::raw_mutex::{futex_wait, futex_wake};
 use std::sync::{Arc, Mutex};
 use std::thread;
 
@@ -130,6 +131,13 @@ use std::thread;
 /// Default capacity of the actor slot table. Slots are ~256 bytes, so the
 /// default costs ~4 MiB, allocated once at `init`. See [`Config::max_actors`].
 pub const DEFAULT_MAX_ACTORS: usize = 16_384;
+/// RFC 004: default scheduler spin budget before parking, in TSC cycles.
+/// ≈0.5µs at a 3.7 GHz nominal TSC. Sized from the spin_sweep fork-join knee:
+/// the fork→join handoff is sub-µs, so ~0.5µs of spin keeps a worker hot across
+/// it for a ~4× p50 latency win, and budgets past ~5k cycles buy nothing but
+/// idle CPU. Only effective where the default spinner cap permits spinning
+/// (small pools, n≤4); see the cap resolution in `init`.
+pub const DEFAULT_SPIN_BUDGET_CYCLES: u64 = 2_000;
 
 /// Runtime configuration.
 ///
@@ -155,6 +163,11 @@ pub struct Config {
     stack_pool_cap: usize,
     max_actors: usize,
     wake_slot: bool,
+    /// RFC 004: TSC cycles a scheduler spins before parking. `0` (default)
+    /// disables spinning entirely — the historical park-immediately behaviour.
+    spin_budget_cycles: u64,
+    /// RFC 004: cap on concurrent spinners. `None` resolves to N/2 at init.
+    max_spinners: Option<u32>,
 }
 
 impl Config {
@@ -168,6 +181,8 @@ impl Config {
             stack_pool_cap: n * 4,
             max_actors: DEFAULT_MAX_ACTORS,
             wake_slot: false,
+            spin_budget_cycles: DEFAULT_SPIN_BUDGET_CYCLES,
+            max_spinners: None,
         }
     }
 
@@ -185,6 +200,8 @@ impl Config {
             stack_pool_cap: max * 4,
             max_actors: DEFAULT_MAX_ACTORS,
             wake_slot: false,
+            spin_budget_cycles: DEFAULT_SPIN_BUDGET_CYCLES,
+            max_spinners: None,
         }
     }
 
@@ -241,6 +258,25 @@ impl Config {
         self
     }
 
+    /// RFC 004: TSC cycles an idle scheduler spins searching for work before
+    /// parking on the futex. The idle-CPU-vs-wake-latency dial, in the same
+    /// unit as [`timeslice_cycles`](Self::timeslice_cycles) (uses `rdtsc()`).
+    /// `0` (the default) disables spinning and the futex park entirely,
+    /// recovering the historical `thread::sleep` idle behaviour exactly.
+    pub fn spin_budget_cycles(mut self, n: u64) -> Self {
+        self.spin_budget_cycles = n;
+        self
+    }
+
+    /// RFC 004: cap on the number of schedulers that may spin concurrently.
+    /// Default (`None`) resolves to N/2 at init. Capping at half keeps half
+    /// the pool hot and parks the rest — the half-load case is then the easy
+    /// case. Ignored when `spin_budget_cycles == 0`.
+    pub fn max_spinners(mut self, n: u32) -> Self {
+        self.max_spinners = Some(n);
+        self
+    }
+
     /// The number of scheduler threads this config resolves to.
     pub fn resolved_thread_count(&self) -> usize {
         if let Some(e) = self.exact {
@@ -265,6 +301,8 @@ impl Default for Config {
             stack_pool_cap: avail * 4,
             max_actors: DEFAULT_MAX_ACTORS,
             wake_slot: false,
+            spin_budget_cycles: DEFAULT_SPIN_BUDGET_CYCLES,
+            max_spinners: None,
         }
     }
 }
@@ -484,6 +522,36 @@ pub(crate) struct RuntimeInner {
     /// Incremented in `spawn` before the enqueue; decremented at the very end
     /// of `finalize_actor`, after every wakeup finalize produces.
     pub(crate) live_actors: AtomicU32,
+    /// RFC 004: lock-free shared-run-queue depth. `+1` in `enqueue` (the sole
+    /// `run_queue.push` site — covers spawn-publish, unpark, and slot
+    /// displacement), `-1` on a successful pop. This is the poll source the
+    /// idle spinners read; it never takes the queue mutex. Slot-parked work is
+    /// deliberately NOT counted: it is thread-local and only its owning thread
+    /// can resume it, so it is not work a spinning sibling could grab.
+    pub(crate) queue_len: AtomicU64,
+    /// RFC 004: schedulers currently *searching* for work (spinning) — the
+    /// middle state between running and parked. The submit rule loads this to
+    /// decide whether a wake is needed (a live spinner will catch the work
+    /// with no syscall). Capped at `max_spinners`.
+    pub(crate) n_spinning: AtomicU32,
+    /// RFC 004: the scheduler park futex word. A monotonic sequence: a waker
+    /// does `fetch_add(1) + futex_wake`, a parker captures the value and
+    /// `futex_wait`s on it, so a bump landing in the arm→wait gap returns the
+    /// wait immediately (EAGAIN) instead of sleeping — the lost-wakeup guard.
+    /// One shared word; `futex_wake(1)` wakes exactly one parked scheduler.
+    pub(crate) park_seq: AtomicU32,
+    /// RFC 004: `spin_budget_cycles > 0 && max_spinners > 0`. When false the
+    /// feature is fully off — the idle path is the historical `thread::sleep`
+    /// park, `n_spinning` and `park_seq` are never touched, and the submit rule
+    /// is a no-op. Note the cap term: a budget with a 0 cap (e.g. the N=1
+    /// pool-size gate) can enlist no spinner, so it is treated as off — this is
+    /// what makes `spin_budget_cycles = 0` AND the 0-cap case recover today's
+    /// behaviour exactly.
+    pub(crate) spinning: bool,
+    /// RFC 004: TSC cycles a scheduler searches before parking (`Config`).
+    pub(crate) spin_budget_cycles: u64,
+    /// RFC 004: cap on concurrent spinners (`Config`, default N/2).
+    pub(crate) max_spinners: u32,
     /// Timer heap. Independent lock: never nested with any other.
     pub(crate) timers: Mutex<Timers>,
     /// IO subsystem. `None` between runs. Lock order: io before everything.
@@ -521,16 +589,42 @@ impl RuntimeInner {
         stack_pool_cap: usize,
         max_actors: usize,
         wake_slot: bool,
+        spin_budget_cycles: u64,
+        max_spinners: u32,
     ) -> Arc<Self> {
         let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
         let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
         // Low indices on top of the stack so early spawns get low pids.
         let free: Vec<u32> = (0..max_actors as u32).rev().collect();
+        // RFC 004: the spin/futex machinery is only live when BOTH a budget is
+        // set AND the cap permits at least one spinner. A budget with a 0 cap
+        // (e.g. the N=1 pool-size gate) can never enlist a spinner, so
+        // `n_spinning` is pinned at 0 forever — which would make the enqueue
+        // submit-rule guard (`n_spinning == 0`) fire a wake-nobody futex_wake on
+        // every push. Gating on the cap too keeps that path inert exactly when
+        // it can do no work. Equivalent to budget=0 here, with no behavioural
+        // change (a 0-cap runtime never parks on the futex either way).
+        let spinning = spin_budget_cycles > 0 && max_spinners > 0;
+        // Guardrail: catch any future config path that re-enables spinning while
+        // leaving the cap at 0 — the pathology this gate exists to prevent. If
+        // we are spinning, at least one spinner must be enlistable.
+        debug_assert!(
+            !spinning || max_spinners > 0,
+            "spinning enabled (budget={spin_budget_cycles}) with max_spinners=0: \
+             no scheduler can ever enlist, so every enqueue would fire a \
+             wake-nobody futex_wake"
+        );
         Arc::new(Self {
             run_queue: crate::run_queue::RunQueue::new(thread_count, max_actors),
             slots,
             free: RawMutex::new(free),
             live_actors: AtomicU32::new(0),
+            queue_len: AtomicU64::new(0),
+            n_spinning: AtomicU32::new(0),
+            park_seq: AtomicU32::new(0),
+            spinning,
+            spin_budget_cycles,
+            max_spinners,
             timers: Mutex::new(Timers::new()),
             io: Mutex::new(None),
             next_monitor_id: AtomicU64::new(0),
@@ -572,9 +666,127 @@ impl RuntimeInner {
             "enqueue of a pid not in (gen, Queued)"
         );
         self.run_queue.push(pid);
+        // RFC 004: maintain the lock-free depth the idle spinners poll. Release
+        // so a spinner's Acquire load observes the pushed entry.
+        self.queue_len.fetch_add(1, Ordering::Release);
+        // RFC 004 submit rule: ensure a searcher will see this work. A live
+        // spinner (n_spinning > 0) claims it with no syscall — the common,
+        // cheap path; only when nobody is searching do we pay a wake. The
+        // SeqCst fence forbids reordering the n_spinning load ahead of the
+        // queue_len store above. Paired with the parker's symmetric
+        // (decrement n_spinning; SeqCst fence; re-check queue_len), it
+        // guarantees at least one of {we observe the spinner, the parker
+        // observes this work} — the Dekker property that keeps work from
+        // stranding with every scheduler asleep. Loom does not model the
+        // futex path (see sync_shim.rs), so this is correct by construction.
+        if self.spinning {
+            fence(Ordering::SeqCst);
+            if self.n_spinning.load(Ordering::Acquire) == 0 {
+                self.unpark_one_scheduler();
+            }
+        }
         crate::te!(crate::trace::Event::Enqueue(pid));
     }
 
+    /// RFC 004: wake exactly one parked scheduler. Bumping the futex sequence
+    /// before the wake closes the arm→wait race: a parker that already
+    /// captured the old sequence fails its `futex_wait` value-compare and
+    /// returns (EAGAIN) instead of sleeping through the wake. Only called when
+    /// no scheduler is searching, so it is off the common path.
+    pub(crate) fn unpark_one_scheduler(&self) {
+        self.park_seq.fetch_add(1, Ordering::Release);
+        futex_wake(&self.park_seq, 1);
+    }
+
+    /// RFC 004: release ALL futex-parked schedulers (termination). Required in
+    /// a no-io runtime, where there is no wake pipe to fall back on: a sibling
+    /// parked indefinitely on the futex would otherwise sleep past program
+    /// end. Each woken thread re-runs the idle verdict, reaches AllDone, and
+    /// returns — idempotent, mirroring the io wake-pipe terminal broadcast.
+    pub(crate) fn unpark_all_schedulers(&self) {
+        self.park_seq.fetch_add(1, Ordering::Release);
+        futex_wake(&self.park_seq, i32::MAX);
+    }
+
+    /// RFC 004: the no-timer / no-io idle policy. Spin on `queue_len` for the
+    /// configured budget (if under the spinner cap), then park on the futex.
+    /// Returns when the caller should loop back and re-attempt a pop. Only
+    /// invoked when `self.spinning` (budget > 0); with budget 0 the caller
+    /// keeps the historical `thread::sleep` park.
+    fn idle_spin_then_park(&self) {
+        // 1. Try to enlist as a spinner, strictly honouring the cap (CAS, so
+        //    `n_spinning` never exceeds `max_spinners` even transiently).
+        let mut spinning = false;
+        loop {
+            let c = self.n_spinning.load(Ordering::Acquire);
+            if c >= self.max_spinners {
+                break; // cap met (or 0): skip the spin, park directly
+            }
+            if self
+                .n_spinning
+                .compare_exchange_weak(c, c + 1, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
+                spinning = true;
+                break;
+            }
+        }
+
+        if spinning {
+            let start = crate::preempt::rdtsc();
+            loop {
+                if self.queue_len.load(Ordering::Acquire) > 0 {
+                    // Found work. Leave the spinner set; if we were the LAST
+                    // spinner (1 → 0), wake one parked sibling before going
+                    // heads-down (rule 4) so a burst still arriving is still
+                    // observed by a searcher.
+                    let prev = self.n_spinning.fetch_sub(1, Ordering::AcqRel);
+                    if prev == 1 {
+                        fence(Ordering::SeqCst);
+                        if self.queue_len.load(Ordering::Acquire) > 0 {
+                            self.unpark_one_scheduler();
+                        }
+                    }
+                    return; // back to the pop
+                }
+                if crate::preempt::rdtsc().wrapping_sub(start) >= self.spin_budget_cycles {
+                    break; // budget exhausted: park
+                }
+                core::hint::spin_loop();
+            }
+            // Budget exhausted: leave the spinner set, then park (below). The
+            // decrement must precede the park's queue re-check — that is the
+            // parker half of the submit-rule Dekker pairing.
+            self.n_spinning.fetch_sub(1, Ordering::AcqRel);
+        }
+
+        // 2. Park on the futex. Three guards close every lost-wakeup / hang:
+        //    - SeqCst fence + queue_len re-check: the Dekker pairing with the
+        //      submit rule (which loads n_spinning AFTER publishing the work),
+        //      so a submit that declined to wake us because it saw us spinning
+        //      is observed here instead.
+        //    - capture park_seq BEFORE the final re-check: a submit-wake
+        //      landing in the gap bumps the sequence, so futex_wait's
+        //      value-compare returns immediately rather than sleeping.
+        //    - live_actors re-check: the shutdown guard. In this arm io_out is
+        //      0 by construction (the io>0 case is handled by a sibling arm),
+        //      so live_actors == 0 IS the AllDone verdict; if termination is
+        //      racing our park we loop back and reach AllDone instead of
+        //      sleeping past program end. (unpark_all_schedulers covers the
+        //      already-parked case from the other side.)
+        fence(Ordering::SeqCst);
+        let observed = self.park_seq.load(Ordering::Acquire);
+        if self.queue_len.load(Ordering::Acquire) > 0 {
+            return;
+        }
+        if self.live_actors.load(Ordering::Acquire) == 0 {
+            return;
+        }
+        futex_wait(&self.park_seq, observed);
+        // Woken by a wake, a stale-sequence EAGAIN, or spuriously: loop back
+        // and re-pop. The verdict re-runs from scratch, so any wake is benign.
+    }
+
     /// Make `pid` runnable if it is parked; coalesce or defer otherwise.
     /// The runtime-internal core of `scheduler::unpark`. WILDCARD wake:
     /// consumes the epoch but does not check it — reserved for terminal
@@ -704,6 +916,22 @@ pub struct Runtime {
 /// Initialise the runtime with the given config. Returns a reusable handle.
 pub fn init(config: Config) -> Runtime {
     let n = config.resolved_thread_count();
+    // RFC 004: default spinner cap is gated on pool size, from the spin_sweep
+    // fork-join data. Spinning is a ~4× p50 latency win at n≤4 workers but pure
+    // cost at n≥8 (large pools stay hot on their own, so a spin budget only
+    // steals CPU from workers doing real drain). So:
+    //   n == 1  → 0  : a lone thread has no cross-thread handoff to hide.
+    //   2..=4   → n  : the win regime; let every worker spin its budget.
+    //   n >= 5  → 0  : park immediately, exactly the historical behaviour —
+    //                  the non-zero DEFAULT_SPIN_BUDGET_CYCLES is inert here
+    //                  because no worker is permitted to enter the spin path.
+    // An explicit Config::max_spinners(_) overrides this (e.g. a future autotune
+    // utility that picks budget+cap from measured load).
+    let max_spinners = config.max_spinners.unwrap_or(match n {
+        0 | 1 => 0,
+        2..=4 => n as u32,
+        _ => 0,
+    });
     Runtime {
         inner: RuntimeInner::new(
             n,
@@ -712,6 +940,8 @@ pub fn init(config: Config) -> Runtime {
             config.stack_pool_cap,
             config.max_actors,
             config.wake_slot,
+            config.spin_budget_cycles,
+            max_spinners,
         ),
         thread_count: n,
     }
@@ -1207,7 +1437,12 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
 
             stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed);
             let pop = match inner.run_queue.pop() {
-                Some(pid) => Pop::Got(pid),
+                Some(pid) => {
+                    // RFC 004: mirror the enqueue increment. Release pairs with
+                    // the spinners' Acquire load of queue_len.
+                    inner.queue_len.fetch_sub(1, Ordering::Release);
+                    Pop::Got(pid)
+                }
                 None => {
                     // Termination does not lean on pop-None being a fence (with
                     // the ring queues it is only a snapshot). The argument is
@@ -1249,6 +1484,13 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                     if let Some(io) = inner.io.lock().unwrap().as_ref() {
                         io.wake();
                     }
+                    // RFC 004: a no-io runtime has no wake pipe; futex-parked
+                    // siblings must be released or shutdown hangs. Broadcast;
+                    // each re-runs the verdict and reaches AllDone itself. The
+                    // sequence bump also EAGAINs any sibling mid-park.
+                    if inner.spinning {
+                        inner.unpark_all_schedulers();
+                    }
                     return;
                 }
                 Pop::Idle { io_outstanding, wake_fd } => {
@@ -1274,9 +1516,18 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
                             crate::io::drain_wake_pipe(fd);
                         }
                         _ => {
+                            // RFC 004: the no-timer / no-io idle arm — the
+                            // ~100µs thread::sleep worst case this RFC targets.
+                            // With spinning enabled, spin-before-park on a
+                            // wakeable futex; with budget 0 the historical
+                            // sleep is preserved exactly.
+                            if inner.spinning {
+                                inner.idle_spin_then_park();
+                            } else {
                                 thread::sleep(std::time::Duration::from_micros(100));
                             }
                         }
+                    }
                     continue;
                 }
             }

tests/spinning_workers.rs

@@ -0,0 +1,260 @@
+//! RFC 004 spinning-worker tests.
+//!
+//! The futex scheduler-park path is NOT loom-modelled (sync_shim.rs: the full
+//! runtime — context switches, futexes, TLS — is never run under cfg(loom)),
+//! so the Dekker submit/park ordering is guarded here instead, by driving the
+//! one true deadlock state and catching it as a timeout rather than a hung
+//! binary: every scheduler parked on the futex while runnable work exists.
+//!
+//! All tests use a NO-IO runtime (pure message passing) so the idle path is
+//! the bare `_` arm RFC 004 replaces — the futex park — not the timer/io arms,
+//! which keep their own wake sources. A lost wakeup therefore manifests as a
+//! permanent stall: a `recv` that never returns, an `rt.run` that never
+//! reaches AllDone. The watchdog turns that into a failed assertion.
+
+use smarm::channel::channel;
+use smarm::runtime::{init, Config};
+use smarm::spawn;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Run `body` on a side thread; fail (rather than hang the test binary) if it
+/// does not finish within `secs`. A hang here means a scheduler parked on the
+/// futex with work outstanding and nothing woke it — exactly the RFC 004
+/// failure mode. The leaked thread is acceptable: the process exits non-zero.
+fn run_with_timeout(secs: u64, body: impl FnOnce() + Send + 'static) {
+    let (tx, rx) = std::sync::mpsc::channel();
+    std::thread::spawn(move || {
+        body();
+        let _ = tx.send(());
+    });
+    if rx.recv_timeout(Duration::from_secs(secs)).is_err() {
+        panic!(
+            "RFC 004 deadlock: workload did not terminate within {secs}s \
+             — a scheduler wakeup was lost (work stranded with all schedulers \
+             parked on the futex)"
+        );
+    }
+}
+
+/// A single token ping-ponging between two actors for `rounds` rounds, on a
+/// `threads`-wide no-io runtime with the given spin budget. With more threads
+/// than active actors, the surplus schedulers go idle and park; every send to
+/// the (parked) peer is a cross-thread wakeup. A lost wake stalls the bounce.
+/// The post-run count confirms no message was silently dropped.
+fn bounce(threads: usize, rounds: u64, budget: u64) {
+    run_with_timeout(30, move || {
+        let rt = init(Config::exact(threads).spin_budget_cycles(budget));
+        let echoed = Arc::new(AtomicU64::new(0));
+        let echoed_actor = echoed.clone();
+        rt.run(move || {
+            let (tx_ab, rx_ab) = channel::<u64>();
+            let (tx_ba, rx_ba) = channel::<u64>();
+            let echo = spawn(move || {
+                for _ in 0..rounds {
+                    let v = rx_ab.recv().expect("echo recv");
+                    echoed_actor.fetch_add(1, Ordering::Relaxed);
+                    tx_ba.send(v + 1).expect("echo send");
+                }
+            });
+            for i in 0..rounds {
+                tx_ab.send(i).expect("ping send");
+                let r = rx_ba.recv().expect("ping recv");
+                assert_eq!(r, i + 1, "token corrupted in flight");
+            }
+            echo.join().expect("echo join");
+        });
+        assert_eq!(
+            echoed.load(Ordering::Relaxed),
+            rounds,
+            "messages were lost, not just delayed"
+        );
+    });
+}
+
+/// Tiny budget: schedulers park almost immediately, so essentially every
+/// cross-thread handoff exercises the futex wake. Sharpest stress on the
+/// submit-rule / lost-wakeup guards.
+#[test]
+fn bounce_forces_park_and_wake() {
+    bounce(4, 20_000, 2_000);
+}
+
+/// Large budget: idle schedulers stay hot spinners, so the submit rule mostly
+/// takes the `n_spinning > 0` skip and the last-spinner handoff (rule 4) does
+/// the waking. Confirms that path also makes progress and stays correct.
+#[test]
+fn bounce_with_hot_spinners() {
+    bounce(4, 20_000, 2_000_000);
+}
+
+/// Oversubscribed: more active actors than schedulers, so the run queue is
+/// rarely empty and the park/spin/submit machinery is hammered from every
+/// thread at once.
+#[test]
+fn bounce_oversubscribed() {
+    bounce(2, 30_000, 5_000);
+}
+
+/// Bursty fork: a parent makes many children runnable at once (the motivating
+/// fan-out), each sends one reply, the parent collects all of them. Exercises
+/// the submit rule under a burst — the first job eats one wake, the woken
+/// worker becomes a spinner and absorbs the rest — and confirms none are lost.
+fn fanout(threads: usize, children: u64, budget: u64) {
+    run_with_timeout(30, move || {
+        let rt = init(Config::exact(threads).spin_budget_cycles(budget));
+        rt.run(move || {
+            let (tx, rx) = channel::<u64>();
+            for c in 0..children {
+                let tx = tx.clone();
+                spawn(move || {
+                    tx.send(c).expect("child send");
+                });
+            }
+            drop(tx);
+            let mut sum = 0u64;
+            for _ in 0..children {
+                sum += rx.recv().expect("collector recv");
+            }
+            assert_eq!(sum, (0..children).sum(), "a child reply was lost");
+        });
+    });
+}
+
+#[test]
+fn fanout_burst_parks() {
+    fanout(4, 256, 2_000);
+}
+
+#[test]
+fn fanout_burst_hot() {
+    fanout(4, 256, 2_000_000);
+}
+
+/// Shutdown guard: a mostly-idle spinning runtime must still terminate. Most
+/// schedulers park on the futex with no work ever coming; only the AllDone
+/// broadcast (`unpark_all_schedulers`) can release them. If that broadcast is
+/// missing, `rt.run` never returns and the watchdog fires.
+#[test]
+fn idle_runtime_shuts_down() {
+    run_with_timeout(15, || {
+        let rt = init(Config::exact(8).spin_budget_cycles(2_000));
+        rt.run(|| {
+            // One trivial cross-thread handoff, then everyone goes idle and the
+            // runtime must drain to AllDone through parked schedulers.
+            let (tx, rx) = channel::<u64>();
+            let h = spawn(move || {
+                tx.send(42).expect("send");
+            });
+            assert_eq!(rx.recv().expect("recv"), 42);
+            h.join().expect("join");
+        });
+    });
+}
+
+/// Repeated lifecycles: init→run→AllDone several times in one process, to
+/// catch any park/broadcast state that fails to reset between runtimes.
+#[test]
+fn repeated_spinning_lifecycles() {
+    run_with_timeout(30, || {
+        for _ in 0..20 {
+            let rt = init(Config::exact(4).spin_budget_cycles(2_000));
+            let echoed = Arc::new(AtomicU64::new(0));
+            let e = echoed.clone();
+            rt.run(move || {
+                let (tx, rx) = channel::<u64>();
+                let worker = spawn(move || {
+                    for _ in 0..500 {
+                        let v = rx.recv().expect("recv");
+                        e.fetch_add(v, Ordering::Relaxed);
+                    }
+                });
+                for i in 0..500u64 {
+                    tx.send(i).expect("send");
+                }
+                worker.join().expect("worker join");
+            });
+            assert_eq!(echoed.load(Ordering::Relaxed), (0..500u64).sum());
+        }
+    });
+}
+
+/// Shutdown with siblings *pinned* in the futex park — the test that actually
+/// exercises the AllDone broadcast. A single long-running actor holds `live`
+/// at 1 (CPU-bound, never sending or spawning) long enough for every other
+/// scheduler to idle, spin out its budget, and commit to `futex_wait`. Because
+/// nothing is ever enqueued during that window, the submit-rule wakes never
+/// fire and those siblings stay parked; the final finalize enqueues no one, so
+/// the live-check self-termination cannot save them either. Only
+/// `unpark_all_schedulers` at AllDone can. Gut that broadcast and this hangs.
+#[test]
+fn shutdown_releases_pinned_parked_siblings() {
+    run_with_timeout(15, || {
+        let rt = init(Config::exact(8).spin_budget_cycles(2_000));
+        rt.run(|| {
+            // ~50ms of pure CPU: far longer than the µs it takes the other
+            // seven schedulers to park, so they are provably in futex_wait
+            // before this returns and live drops to 0.
+            let start = std::time::Instant::now();
+            let mut x = 0u64;
+            while start.elapsed() < Duration::from_millis(50) {
+                x = x.wrapping_mul(6364136223846793005).wrapping_add(1);
+                std::hint::black_box(x);
+            }
+            std::hint::black_box(x);
+        });
+    });
+}
+
+/// cross-thread handoff — but the futex park path must still be inert and
+/// correct (the lone thread is never parked while it has work). Pins the
+/// degenerate cap=0 case.
+#[test]
+fn single_thread_spinning_is_inert() {
+    run_with_timeout(15, || {
+        let rt = init(Config::exact(1).spin_budget_cycles(2_000));
+        let echoed = Arc::new(AtomicU64::new(0));
+        let e = echoed.clone();
+        rt.run(move || {
+            let (tx, rx) = channel::<u64>();
+            let worker = spawn(move || {
+                for _ in 0..1_000 {
+                    let v = rx.recv().expect("recv");
+                    e.fetch_add(v, Ordering::Relaxed);
+                }
+            });
+            for i in 0..1_000u64 {
+                tx.send(i).expect("send");
+            }
+            worker.join().expect("join");
+        });
+        assert_eq!(echoed.load(Ordering::Relaxed), (0..1_000u64).sum());
+    });
+}
+
+/// `spin_budget_cycles == 0` is the opt-out: the feature is fully off and the
+/// runtime uses the historical `thread::sleep` idle path. Must still run a
+/// cross-thread workload to completion (it always did) — this just guards that
+/// the gating left the off-path intact.
+#[test]
+fn budget_zero_keeps_old_behaviour() {
+    run_with_timeout(20, || {
+        let rt = init(Config::exact(4).spin_budget_cycles(0));
+        let sum = Arc::new(AtomicU64::new(0));
+        let s = sum.clone();
+        rt.run(move || {
+            let (tx, rx) = channel::<u64>();
+            let worker = spawn(move || {
+                for _ in 0..1_000 {
+                    s.fetch_add(rx.recv().expect("recv"), Ordering::Relaxed);
+                }
+            });
+            for i in 0..1_000u64 {
+                tx.send(i).expect("send");
+            }
+            worker.join().expect("join");
+        });
+        assert_eq!(sum.load(Ordering::Relaxed), (0..1_000u64).sum());
+    });
+}