diff --git a/issues/build/timing-verdicts-ruled-off-qemu-have-no-metal-arm.md b/issues/build/timing-verdicts-ruled-off-qemu-have-no-metal-arm.md
index 41e7c2d7def..3d912143db2 100644
--- a/issues/build/timing-verdicts-ruled-off-qemu-have-no-metal-arm.md
+++ b/issues/build/timing-verdicts-ruled-off-qemu-have-no-metal-arm.md
@@ -73,7 +73,7 @@ the event it stands for has no word a test can read:
   `INTO_WINDOW`, and `redirty_mid_flush`'s swept delay.
 - Drains of work handed to another thread: the 200 ms `iod` drains in
   `writeback_durability` and `home_backing_revoked`, `fat_backing_revoked`'s
-  drain, and `log_gate`'s `STORM_SETTLE` and `QUIET_READS`.
+  drain, and `log_gate`'s `QUIET_READS`.
 - Product clocks a QEMU boot races: `boot_deadline_ends_a_wedge`'s 15 s
   deadline against the job list reaching its shutdown,
   `hard_lockup_ends_a_deaf_cpu`'s 30 s, `loader_watchdog_arms` reading
diff --git a/issues/diagnostics/a-shards-timestamps-run-backwards-at-seq-517.md b/issues/diagnostics/a-shards-timestamps-run-backwards-at-seq-517.md
deleted file mode 100644
index 8e560986f01..00000000000
--- a/issues/diagnostics/a-shards-timestamps-run-backwards-at-seq-517.md
+++ /dev/null
@@ -1,49 +0,0 @@
----
-status: open
-kind: defect
-opened: 2026-09-06
----
-
-# A shard's timestamps run backwards at seq 517, and sometimes reds `sched_stress`
-
-The log gate reports one record per full-tier run whose timestamp is behind the
-record before it *in its own shard*, which the gate states is impossible:
-
-```
-[log] unbracketed: log-gate: FAILED: cpu6 seq 517 is stamped 736451308 ns,
-behind the 739564366 ns of the record before it — within a shard the sequence
-order is the timestamp order, and `emit` stamps inside the same bracket it
-reserves in
-```
-
-Four full-tier runs on this host, one of them on `origin/metal` (`b3c314cf`)
-with no working-tree diff at all:
-
-| run | tree | shard | seq | inversion | outcome |
-|---|---|---|---|---|---|
-| 1 | `t14-run4` | cpu5 | 517 | 650224011 behind 651750439 ns (1.5 ms) | `FAIL sched_stress`, 324/325 |
-| 2 | `b3c314cf` (base) | cpu6 | 517 | 742053639 behind 743716169 ns (1.7 ms) | 325/325 green |
-| 3 | `t14-run4` | cpu6 | 517 | 736451308 behind 739564366 ns (3.1 ms) | 325/325 green |
-| 4 | `t14-run4`, one more boot record | cpu6 | 518 | 736406061 behind 737807237 ns (1.4 ms) | 325/325 green |
-
-So it is **not** the `t14-run4` diff — it fires on the untouched base. The only
-reason it is not a permanent red is that the boot it lands in is usually not one
-a test is judging.
-
-It is always an AP's shard, never cpu0's, and always **one fixed position** in
-that shard. Run 4 is what shows that: this branch added one boot record ahead of
-it and the inversion moved 517 -> 518 with it. So what selects the record is its
-index in the shard, not its sequence number, not the wall time, and not the CPU
-— which is a much narrower thing to look for than a race that happens to recur.
-
-The gate's own sentence names the invariant that is broken: `emit` stamps
-`record.at_ns` and reserves the shard slot inside one `LogCommitGuard` bracket,
-so a later `seq` in a shard cannot carry an earlier stamp unless either the
-stamp or the reservation escapes that bracket, or `clock::nanos_since_boot`
-reads backwards on that CPU. The last of those is the cheapest to check first
-and overlaps `issues/kernel/ap-tsc-trail-is-assumed-and-never-checked.md`.
-
-**Exit condition**: the inversion explained and gone — either a `seq 517` that
-holds its bracket, or a demonstration that the AP's TSC is what moved, priced
-against that issue. Until then a `sched_stress` red carrying this line is this
-defect and not the author's diff.
diff --git a/issues/isolation/test-runner-holds-logread-where-no-log-builtin-runs.md b/issues/isolation/test-runner-holds-logread-where-no-log-builtin-runs.md
new file mode 100644
index 00000000000..7d551f411f6
--- /dev/null
+++ b/issues/isolation/test-runner-holds-logread-where-no-log-builtin-runs.md
@@ -0,0 +1,25 @@
+---
+status: open
+kind: defect
+opened: 2026-09-28
+---
+
+# test-runner holds `logread` where no log builtin runs
+
+`logread` in a `[programs.test-runner]` row is authority for the builtins that
+read the log inside test-runner (`log-gate`, `log-storm`, `log-close`,
+`userland/test-runner/src/main.rs`'s `BUILTINS`). Every boot that runs one is a
+`tests/testcases` boot (`tests/common/logread.rs` boots the machine tests' config;
+`tests/toyos.rs`'s one job list naming `log-close` is `tests/testcases`). Seven
+other manifests grant it anyway: `tests/partclaimcase`, `tests/blockdcase`,
+`tests/doommusiccase`, `tests/logrotatecase`, `tests/metalcase`, `tests/netcase`
+and `tests/sshdcase`.
+
+`partclaimcase` and `blockdcase` also grant `dup`, so there every child
+test-runner spawns receives a `SysCap` duplicate carrying `LOG` as well.
+
+**Evidence:** `rg -n '"log-gate"|"log-storm"|"log-close"' tests userland`, and
+`rg -n logread -g system.toml tests`.
+
+**Exit:** every `logread` in a test manifest is one a program on that boot
+uses, or the row says which use it is for.
diff --git a/issues/kernel/the-kernel-still-creates-threads.md b/issues/kernel/the-kernel-still-creates-threads.md
index f9e353060f4..a263a8df7e8 100644
--- a/issues/kernel/the-kernel-still-creates-threads.md
+++ b/issues/kernel/the-kernel-still-creates-threads.md
@@ -25,10 +25,8 @@ code creates a schedulable task other than the per-CPU idle loop.
   last thread tears down its own process on its way out of the kernel, and the
   scheduler frees that thread's kernel stack after switching away. Blocked on
   #549 landing.
-- **K3:** the test-only `logstorm`/`lognest` producers are deleted if the log
-  gate does not need kernel-context producers. Blocked on nothing.
 - **K4:** `klogd` goes: the owner-approved driver-model design moves the
   console to logd, and this track owns that move.
 - **K5:** `iod` goes with the kernel's write-back queue; met only when #536
   lands with no new `kthread::spawn`.
-- **K6:** delete the machinery named above. Blocked on K2–K5.
+- **K6:** delete the machinery named above. Blocked on K2, K4 and K5.
diff --git a/kernel/src/actuator.rs b/kernel/src/actuator.rs
index cd2f2cd8bae..a091ef9acc8 100644
--- a/kernel/src/actuator.rs
+++ b/kernel/src/actuator.rs
@@ -423,9 +423,6 @@ actuators! {
     /// Log the monotonic time and which CPUs are alive every 250ms.
     heartbeat = "heartbeat";
 
-    /// Have every CPU emit patterned log records at once from spawned kernel threads.
-    log_storm = "log-storm";
-
     /// Remove the IF/TF bracket around shard selection through publication — the negative control on the log's interrupt-atomicity claim.
     log_unbracketed_reserve = "log-unbracketed-reserve";
 
@@ -435,9 +432,6 @@ actuators! {
     /// The same IPI, sent between the shard-pointer read and the unlocked `xadd` — stages order damage the log gate detects, unlike the row above's invisible corruption.
     log_nested_reserve = "log-nested-reserve";
 
-    /// Turn the reservation's `xadd` into a load, an open interrupt window, and a store.
-    log_shared_reservation = "log-shared-reservation";
-
     /// Let a handle close cancel every poll on the log's watch in the machine.
     log_close_cancels_any_syscap = "log-close-cancels-any-syscap";
 
diff --git a/kernel/src/arch/aarch64/mod.rs b/kernel/src/arch/aarch64/mod.rs
index 1ac0e061e99..5c9e00e144a 100644
--- a/kernel/src/arch/aarch64/mod.rs
+++ b/kernel/src/arch/aarch64/mod.rs
@@ -109,18 +109,6 @@ pub unsafe fn percpu_fetch_add(
     _guard: &IrqGuard,
 ) -> u64 {
     let previous = counter.load(core::sync::atomic::Ordering::Relaxed);
-    // Under `log-shared-reservation`, open the window the guard closes, so a
-    // nested record can land between the load and the store.
-    if crate::actuator::log_shared_reservation() && crate::log::nested::inject() {
-        // SAFETY: each writes `DAIF.I` and touches no memory.
-        unsafe {
-            core::arch::asm!("msr daifclr, #2");
-            for _ in 0..256 {
-                core::hint::spin_loop();
-            }
-            core::arch::asm!("msr daifset, #2");
-        }
-    }
     // A load and a store, not an atomic add: the guard masks the only other
     // writer this CPU has, and no other CPU writes the counter.
     counter.store(previous + 1, core::sync::atomic::Ordering::Relaxed);
diff --git a/kernel/src/arch/x86_64/mod.rs b/kernel/src/arch/x86_64/mod.rs
index e44661ba2fb..41ec8a0f805 100644
--- a/kernel/src/arch/x86_64/mod.rs
+++ b/kernel/src/arch/x86_64/mod.rs
@@ -107,23 +107,6 @@ pub unsafe fn percpu_fetch_add(
     counter: &core::sync::atomic::AtomicU64,
     _guard: &IrqGuard,
 ) -> u64 {
-    // Under `log-shared-reservation`, stage a load/store race instead of the `xadd` below.
-    if crate::actuator::log_shared_reservation() {
-        let previous = counter.load(core::sync::atomic::Ordering::Relaxed);
-        if crate::log::nested::inject() {
-            // SAFETY: `sti`/`cli` each write one `RFLAGS` bit and touch no memory.
-            unsafe {
-                core::arch::asm!("sti");
-                for _ in 0..256 {
-                    core::hint::spin_loop();
-                }
-                core::arch::asm!("cli");
-            }
-        }
-        counter.store(previous + 1, core::sync::atomic::Ordering::Relaxed);
-        return previous;
-    }
-
     let previous: u64;
     // Not `AtomicU64::fetch_add`: its locked xadd is costly under QEMU TCG emulation.
     // SAFETY: `counter.as_ptr()` is live; unlocked `xadd` retires whole, atomic against an interrupt here.
diff --git a/kernel/src/log/mod.rs b/kernel/src/log/mod.rs
index 5d4208f7258..07dceac9db2 100644
--- a/kernel/src/log/mod.rs
+++ b/kernel/src/log/mod.rs
@@ -11,7 +11,7 @@ pub mod read;
 pub mod recovery;
 pub mod registry;
 pub mod shard;
-#[cfg(feature = "boot-actuators")]
+#[cfg(any(feature = "boot-actuators", feature = "test-actuators"))]
 pub mod storm;
 pub mod user;
 
diff --git a/kernel/src/log/nested.rs b/kernel/src/log/nested.rs
index fb5b423bb55..d3a61271c44 100644
--- a/kernel/src/log/nested.rs
+++ b/kernel/src/log/nested.rs
@@ -7,7 +7,7 @@
 //! produces a shard whose `at_ns` descends, which `Descent::advance` and the
 //! log gate assume cannot happen.
 
-/// Producer id the burst's records use — outside the range any real storm thread can have.
+/// Producer id the burst's records use — one the log gate's own producer never takes.
 #[cfg(feature = "boot-actuators")]
 pub const NEST_PRODUCER: u64 = u64::MAX;
 
@@ -16,9 +16,8 @@ mod armed {
     use core::sync::atomic::{AtomicBool, Ordering};
 
     use crate::log::shard::SHARD_RECORDS;
-    use crate::sched::kthread;
 
-    /// One-shot for the body-copy injection point, consumed by `mid_body` or, under `log-shared-reservation`, by the outer `inject`.
+    /// One-shot for the body-copy injection point, consumed by `mid_body`.
     static ARMED: AtomicBool = AtomicBool::new(false);
 
     /// One-shot for the reservation-window injection point, consumed by [`reserve_window`]; kept separate from `ARMED` so it can't starve the body window.
@@ -36,17 +35,19 @@ mod armed {
         // Both actuators name the same injection; arming both would inject into one record twice.
         assert!(
             !(crate::actuator::log_nested_emit() && crate::actuator::log_nested_reserve()),
-            "log-nested-emit and log-nested-reserve both name the one injection this thread arms"
+            "log-nested-emit and log-nested-reserve both name the one injection this read arms"
         );
         if STARTED.swap(true, Ordering::Relaxed) {
             return;
         }
         crate::log!("lognest start records={SHARD_RECORDS}");
-        // A kernel thread, not the syscall that arms it: `IF` is clear for a whole syscall, so injecting there would never test the guard.
-        kthread::spawn("lognest", body, 0);
+        // `IF` is clear for a whole syscall, so injecting with it clear would never test the guard; Ring 0 is not preempted, so the body runs whole on this CPU.
+        crate::arch::cpu::enable_interrupts();
+        body();
+        crate::arch::cpu::disable_interrupts();
     }
 
-    extern "C" fn body(_arg: u64) -> ! {
+    fn body() {
         if crate::actuator::log_nested_reserve() {
             ARMED_RESERVE.store(true, Ordering::Relaxed);
             crate::log!(
@@ -61,12 +62,10 @@ mod armed {
             ARMED.store(false, Ordering::Relaxed);
         }
         crate::log!("lognest done emitted={SHARD_RECORDS}");
-
-        crate::watch::park_forever();
     }
 
     /// Consumes the one-shot and sends this CPU its own IPI; `true` if this call sent it.
-    pub fn inject() -> bool {
+    fn inject() -> bool {
         if !ARMED.swap(false, Ordering::Relaxed) {
             return false;
         }
@@ -108,21 +107,12 @@ mod armed {
     }
 }
 
-/// Arms the injection on a dedicated kernel thread, once; compiled only under `boot-actuators`.
+/// Arms the injection and emits the record it lands in, inline in the calling `SYS_LOG_READ`, once; compiled only under `boot-actuators`.
 #[cfg(feature = "boot-actuators")]
 pub fn start_once() {
-    #[cfg(feature = "boot-actuators")]
     armed::start_once();
 }
 
-/// Consumes the one-shot at the reservation, for `log-shared-reservation`; `true` if an IPI went out.
-pub fn inject() -> bool {
-    #[cfg(feature = "boot-actuators")]
-    return armed::inject();
-    #[cfg(not(feature = "boot-actuators"))]
-    false
-}
-
 /// Injection point halfway through a record's body copy; always compiled so `kernel-loom`'s separate copy of `log::shard` names one path.
 pub fn mid_body() {
     #[cfg(feature = "boot-actuators")]
diff --git a/kernel/src/log/storm.rs b/kernel/src/log/storm.rs
index a237a19de9e..2de2b6f8090 100644
--- a/kernel/src/log/storm.rs
+++ b/kernel/src/log/storm.rs
@@ -1,12 +1,5 @@
 //! Generates patterned records so the log gate's reader can check a conservation law over them.
 
-use core::sync::atomic::{AtomicBool, Ordering};
-
-use crate::sched::kthread;
-
-// Exceeds a shard's capacity, so the drop path under test is reached at every `--smp` count.
-const STORM_RECORDS: u64 = 1024;
-
 // Must exceed one machine word: a single-store payload couldn't reveal a torn write.
 const PAYLOAD: usize = 96;
 
@@ -21,7 +14,7 @@ pub fn payload_byte(checksum: u64, offset: usize) -> u8 {
     b'a' + (checksum.wrapping_add(offset as u64) % 26) as u8
 }
 
-/// One patterned record for `thread`/`index`; also called by `log-nested-reserve` from an interrupt handler.
+/// One patterned record for `thread`/`index`; called by `SYS_DEBUG`'s `LOG_PATTERNED` and by `log-nested-reserve` from an interrupt handler.
 /// The reader regenerates this text independently from `t=`/`i=`, so the format here must stay in sync with it.
 pub fn emit_patterned(thread: u64, index: u64) {
     let checksum = checksum(thread, index);
@@ -33,30 +26,3 @@ pub fn emit_patterned(thread: u64, index: u64) {
     let payload = core::str::from_utf8(&payload).unwrap_or("");
     crate::log!("logstorm t={thread} i={index} k={checksum:016x} {payload}");
 }
-
-static STARTED: AtomicBool = AtomicBool::new(false);
-
-/// Spawns one storm thread per shard, once for the life of the machine.
-/// Called from `SYS_LOG_READ`, which is what makes the storm concurrent with a reader by construction.
-pub fn start_once() {
-    if STARTED.swap(true, Ordering::Relaxed) {
-        return;
-    }
-    let threads = super::shard_count();
-    // The reader parses this line to learn the storm's shape.
-    crate::log!("logstorm start threads={threads} records={STORM_RECORDS}");
-    for thread in 0..threads {
-        kthread::spawn("logstorm", body, thread as u64);
-    }
-}
-
-extern "C" fn body(thread: u64) -> ! {
-    for index in 0..STORM_RECORDS {
-        emit_patterned(thread, index);
-    }
-    // The reader decides from its own cursor rather than waiting on this record: a barrier here was tried and hung at scale.
-    crate::log!("logstorm done t={thread} emitted={STORM_RECORDS}");
-
-    // Parks rather than exits: kthread rows are never removed, and spinning here would compete with the reader for the rest of the boot.
-    crate::watch::park_forever();
-}
diff --git a/kernel/src/log/user.rs b/kernel/src/log/user.rs
index 67f6cb6e8db..eb76c13c2a9 100644
--- a/kernel/src/log/user.rs
+++ b/kernel/src/log/user.rs
@@ -46,12 +46,7 @@ pub fn read(
     out: &mut UserBytesMut,
     capacity: usize,
 ) -> Result<usize, SyscallError> {
-    // Started on first read, not at boot: an unread storm has already spent itself before a cursor exists to notice it.
-    #[cfg(feature = "boot-actuators")]
-    if crate::actuator::log_storm() {
-        super::storm::start_once();
-    }
-    // Armed here too, once: one thread serves both injection windows; `log::nested` picks the target from whichever actuators are armed.
+    // Run once, inside the first read's own syscall; `log::nested` picks the window from whichever actuator is armed.
     #[cfg(feature = "boot-actuators")]
     if crate::actuator::log_nested_emit() || crate::actuator::log_nested_reserve() {
         super::nested::start_once();
diff --git a/kernel/src/sched/kthread.rs b/kernel/src/sched/kthread.rs
index decb64376a3..baa4c07fd61 100644
--- a/kernel/src/sched/kthread.rs
+++ b/kernel/src/sched/kthread.rs
@@ -18,11 +18,8 @@ use crate::sync::Lock;
 
 use super::payload::ThreadSched;
 
-/// `klogd` and `iod`, plus one `log-storm` thread per shard in the actuator build.
-#[cfg(not(feature = "boot-actuators"))]
+/// `klogd` and `iod`.
 const MAX_KERNEL_TASKS: usize = 2;
-#[cfg(feature = "boot-actuators")]
-const MAX_KERNEL_TASKS: usize = 2 + toyos_abi::log::MAX_LOG_SHARDS;
 
 /// Collides with no packed id: neither id map issues `u32::MAX`.
 const NO_TASK: u64 = u64::MAX;
diff --git a/kernel/src/syscall/dispatch.rs b/kernel/src/syscall/dispatch.rs
index e0f8a33e842..313f23aed2c 100644
--- a/kernel/src/syscall/dispatch.rs
+++ b/kernel/src/syscall/dispatch.rs
@@ -602,6 +602,10 @@ pub(crate) fn syscall_dispatch(num: u64, a1: u64, a2: u64, a3: u64, a4: u64) ->
                     None => SyscallError::InvalidArgument.to_u64(),
                 }
             }
+            DA::LOG_PATTERNED => {
+                crate::log::storm::emit_patterned(0, a2);
+                0
+            }
             _ => SyscallError::InvalidArgument.to_u64(),
         },
         SYS_SCHED_INFO => match ctx.copy_out(UserAddr::new(a1), &sys_sched_info()) {
diff --git a/kernel/src/watch.rs b/kernel/src/watch.rs
index 718a8c6a5ce..f8bf0aee262 100644
--- a/kernel/src/watch.rs
+++ b/kernel/src/watch.rs
@@ -266,18 +266,6 @@ pub fn wait_uncancellable_until(p: &Parkable, watch: &Watch, token: u64, ready:
     }
 }
 
-/// Parks forever rather than exiting: exiting frees a stack a producer may still write to.
-#[cfg(feature = "boot-actuators")]
-#[track_caller]
-pub fn park_forever() -> ! {
-    let parkable = crate::scheduler::Parkable::at_entry();
-    let handle = crate::sched::driver::current_handle().expect("a kernel thread is a task");
-    let armed = arm(handle.watch(), 0, WaitClass::Other).expect("a task can arm");
-    loop {
-        let _ = wait(&parkable, &armed, Deadline::never());
-    }
-}
-
 #[track_caller]
 fn wait_inner(
     _p: &Parkable,
diff --git a/src/build.rs b/src/build.rs
index d775d1d105f..b52cf500943 100644
--- a/src/build.rs
+++ b/src/build.rs
@@ -2533,7 +2533,7 @@ mod tests {
     fn the_pre_flash_gate_clears_a_valued_parameter_and_refuses_an_actuator() {
         let root = Path::new(env!("CARGO_MANIFEST_DIR"));
         assert_eq!(flashable_params(root, &[format!("{}0x1000", toyos_blackbox::PARAM)]), Ok(()));
-        assert!(flashable_params(root, &["log-storm".to_string()]).is_err());
+        assert!(flashable_params(root, &["wedge-before-reset".to_string()]).is_err());
     }
 
     #[test]
diff --git a/tests/blockdcase/system.toml b/tests/blockdcase/system.toml
index 623c5130aa2..fd9a95d4916 100644
--- a/tests/blockdcase/system.toml
+++ b/tests/blockdcase/system.toml
@@ -22,9 +22,6 @@ syscap = ["logread"]
 # `device` because five of the guest binaries claim the keyboard or the mouse
 # and no manifest row can name them — they are not `[programs]` keys — and `dup`
 # because a claim moves and one boot runs several of them.
-# `logread` because the log gate reads the kernel's own records and cannot be
-# a spawned binary: a `SysCap` dup is not part of the namespace test-runner
-# hands down.
 # `power` because `run shutdown` is how a dozen host-side gates end their guest
 # and read what reached the volume, and test-runner spawns `/system/bin/shutdown`
 # directly — it holds no `launcher` connector, so the applet's authority is the
diff --git a/tests/common/logread.rs b/tests/common/logread.rs
index 8bba44eb139..ba10d353240 100644
--- a/tests/common/logread.rs
+++ b/tests/common/logread.rs
@@ -22,6 +22,9 @@ use super::qemu::{BootOptions, QemuInstance};
 /// either way.
 const GATE: &str = "log-gate";
 
+/// The same gate with its own producer thread storming the log beside it.
+const STORM_GATE: &str = "log-storm";
+
 /// The whole run's ceiling: a gate that never finishes is what it reds.
 const CEILING: Duration = Duration::from_secs(60);
 
@@ -62,7 +65,11 @@ fn conservation(
     rust_bins: &[(String, Vec<u8>)],
     smp: u32,
 ) -> Result<(), String> {
-    let report = storm(test_config, c_bins, rust_bins, smp, &["log-storm"])?;
+    // The test kernel by build rather than by actuator: the producer's
+    // `SYS_DEBUG` is what needs it, and nothing is armed.
+    let options =
+        BootOptions { smp, kernel_features: toyos_build::build::TEST_KERNEL, ..Default::default() };
+    let report = storm(test_config, c_bins, rust_bins, STORM_GATE, options)?;
     let shards = report.get("shards")?;
     if shards != smp as u64 {
         return Err(format!(
@@ -73,40 +80,43 @@ fn conservation(
     }
     // Non-vacuity, and it is the half a green law cannot supply: a reader that
     // took every record after the storm had ended has proved nothing about
-    // concurrent producers.
+    // concurrent producers, and one the ring never lapped has proved nothing
+    // about `lost`.
     let concurrent = report.get("concurrent")?;
     let dropped = report.get("dropped")?;
     let read = report.get("read")?;
-    if concurrent == 0 || read == 0 {
+    let lost = report.get("lost")?;
+    if concurrent == 0 || read == 0 || lost == 0 {
         return Err(format!(
-            "--smp {smp} read {read} record(s), {concurrent} of them while the storm ran\n{}",
+            "--smp {smp} read {read} record(s), {concurrent} of them while the storm ran, and \
+             lost {lost}\n{}",
             report.stdout
         ));
     }
     eprintln!(
         "  [log] smp={smp}: emitted={} read={read} dropped={dropped} concurrent={concurrent} \
-         lost={} wakes={}",
+         lost={lost} wakes={}",
         report.get("emitted")?,
-        report.get("lost")?,
         report.get("wakes")?,
     );
     Ok(())
 }
 
-pub fn log_conservation_smp1(
+/// **`--smp 2`**, so the producer thread has a CPU the reader is not on.
+pub fn log_conservation_smp2(
     test_config: &Path,
     c_bins: &[(String, Vec<u8>)],
     rust_bins: &[(String, Vec<u8>)],
 ) -> Result<(), String> {
-    conservation(test_config, c_bins, rust_bins, 1)
+    conservation(test_config, c_bins, rust_bins, 2)
 }
 
 /// The nested-`emit` gate: an interrupt that logs, inside another `emit`, on one CPU.
 ///
 /// **The one case loom cannot express and the host cannot stage.** The
-/// stimulus is a self-IPI sent from inside a record's own body copy, on a
-/// kernel thread — where `IF` is set and `emit`'s IF-off bracket is the only thing
-/// holding the interrupt off. The handler emits exactly one shard generation of
+/// stimulus is a self-IPI sent from inside a record's own body copy, inside
+/// `SYS_LOG_READ` with `IF` opened for it — where `emit`'s IF-off bracket is the
+/// only thing holding the interrupt off. The handler emits exactly one shard generation of
 /// patterned records; the outer record is then dropped by the ring's own
 /// drop-oldest policy, which is what makes "the burst laps the shard" a
 /// statement with an arithmetic behind it.
@@ -125,7 +135,8 @@ pub fn log_nested_emit(
     c_bins: &[(String, Vec<u8>)],
     rust_bins: &[(String, Vec<u8>)],
 ) -> Result<(), String> {
-    let report = storm(test_config, c_bins, rust_bins, 1, &["log-nested-emit"])?;
+    let options = BootOptions { smp: 1, kernel_params: &["log-nested-emit"], ..Default::default() };
+    let report = storm(test_config, c_bins, rust_bins, GATE, options)?;
     let declared = report.get("declared")?;
     let read = report.get("read")?;
     if read == 0 {
@@ -160,7 +171,8 @@ pub fn log_reserve_window(
     c_bins: &[(String, Vec<u8>)],
     rust_bins: &[(String, Vec<u8>)],
 ) -> Result<(), String> {
-    let report = storm(test_config, c_bins, rust_bins, 8, &["log-nested-reserve"])?;
+    let options = BootOptions { smp: 8, kernel_params: &["log-nested-reserve"], ..Default::default() };
+    let report = storm(test_config, c_bins, rust_bins, GATE, options)?;
     let declared = report.get("declared")?;
     let read = report.get("read")?;
     let dropped = report.get("dropped")?;
@@ -299,21 +311,8 @@ pub fn log_poll_outlives_a_close(
     c_bins: &[(String, Vec<u8>)],
     rust_bins: &[(String, Vec<u8>)],
 ) -> Result<(), String> {
-    close_probe(test_config, c_bins, rust_bins, &[])
-}
-
-fn close_probe(
-    test_config: &Path,
-    c_bins: &[(String, Vec<u8>)],
-    rust_bins: &[(String, Vec<u8>)],
-    params: &'static [&'static str],
-) -> Result<(), String> {
-    let mut qemu = QemuInstance::boot_with_options(
-        test_config,
-        c_bins,
-        rust_bins,
-        BootOptions { kernel_params: params, ..Default::default() },
-    );
+    let mut qemu =
+        QemuInstance::boot_with_options(test_config, c_bins, rust_bins, BootOptions::default());
     let result = qemu.run_test("log-close", CEILING);
     if let Some(err) = &result.error {
         return Err(format!("{err}\nstdout:\n{}", result.stdout));
@@ -333,21 +332,17 @@ fn close_probe(
     Ok(())
 }
 
-/// Boot one machine with the storm armed and read the gate's verdict off it.
+/// Boot one machine as `options` says, run `gate` on it and read its verdict off it.
 fn storm(
     test_config: &Path,
     c_bins: &[(String, Vec<u8>)],
     rust_bins: &[(String, Vec<u8>)],
-    smp: u32,
-    params: &'static [&'static str],
+    gate: &str,
+    options: BootOptions,
 ) -> Result<Report, String> {
-    let mut qemu = QemuInstance::boot_with_options(
-        test_config,
-        c_bins,
-        rust_bins,
-        BootOptions { smp, kernel_params: params, ..Default::default() },
-    );
-    let result = qemu.run_test(GATE, CEILING);
+    let (smp, params) = (options.smp, options.kernel_params);
+    let mut qemu = QemuInstance::boot_with_options(test_config, c_bins, rust_bins, options);
+    let result = qemu.run_test(gate, CEILING);
     if let Some(err) = &result.error {
         return Err(format!(
             "--smp {smp} {params:?}: {err}\nstdout:\n{}\nserial tail:\n{}",
@@ -408,18 +403,9 @@ fn fields(stdout: &str) -> Result<BTreeMap<String, u64>, Contaminated> {
         let Some(rest) = line.split_once("log-gate: ").map(|(_, r)| r) else { continue };
         for word in rest.split_whitespace() {
             let Some((key, value)) = word.split_once('=') else { continue };
-            // `migrated=3/8` is two numbers: the second is the producer count,
-            // which the migration gate reports beside it.
-            let (value, producers) = match value.split_once('/') {
-                Some((a, b)) => (a, b.trim_end_matches(&[',', ';'][..]).parse::<u64>().ok()),
-                None => (value, None),
-            };
             if let Ok(n) = value.trim_end_matches(&[',', ';'][..]).parse::<u64>() {
                 put(&mut out, key, n)?;
             }
-            if let Some(n) = producers {
-                put(&mut out, "producers", n)?;
-            }
         }
         // "N record(s) over M read(s) from S shard(s)" — the shape of the line
         // rather than a key, because those three are what the sentence is.
diff --git a/tests/doommusiccase/system.toml b/tests/doommusiccase/system.toml
index f75d630351f..a8953f41ad7 100644
--- a/tests/doommusiccase/system.toml
+++ b/tests/doommusiccase/system.toml
@@ -26,9 +26,6 @@ devices = ["hda-audio", "virtio-sound"]
 syscap = ["rt"]
 
 # test-runner passes its namespace to doom.
-# `logread` is the log gate's own: a spawned test binary does not inherit a
-# `SysCap` dup, so the gate that reads the kernel's records runs inside
-# `test-runner` itself.
 [programs.test-runner]
 receives = ["soundd"]
 syscap = ["logread"]
diff --git a/tests/logrotatecase/system.toml b/tests/logrotatecase/system.toml
index 9938b47c3d7..48a8e4b2a6a 100644
--- a/tests/logrotatecase/system.toml
+++ b/tests/logrotatecase/system.toml
@@ -61,9 +61,6 @@ service = true
 receives = ["netd", "launcher"]
 
 # The union its guest binaries need on this machine shape.
-# `logread` is the log gate's own: a spawned test binary does not inherit a
-# `SysCap` dup, so the gate that reads the kernel's records runs inside
-# `test-runner` itself.
 # `power` is the connector `run shutdown` below asks init through.
 [programs.test-runner]
 receives = ["compositor", "soundd", "netd", "power"]
diff --git a/tests/metalcase/system.toml b/tests/metalcase/system.toml
index eeca4fe25ed..606871bf0dc 100644
--- a/tests/metalcase/system.toml
+++ b/tests/metalcase/system.toml
@@ -54,9 +54,6 @@ service = true
 receives = ["netd", "launcher"]
 
 # The union its guest binaries need on this machine shape.
-# `logread` is the log gate's own: a spawned test binary does not inherit a
-# `SysCap` dup, so the gate that reads the kernel's records runs inside
-# `test-runner` itself.
 [programs.test-runner]
 receives = ["compositor", "soundd", "netd"]
 syscap = ["logread"]
diff --git a/tests/netcase/system.toml b/tests/netcase/system.toml
index 1ae7dc9ddde..211a3e55df2 100644
--- a/tests/netcase/system.toml
+++ b/tests/netcase/system.toml
@@ -34,9 +34,6 @@ devices = ["pci:1af4:1041"]
 # This is also the only config whose test binaries take the launcher path to
 # `Command::spawn` at all: everywhere else test-runner holds no connector and
 # every spawn is direct.
-# `logread` is the log gate's own: a spawned test binary does not inherit a
-# `SysCap` dup, so the gate that reads the kernel's records runs inside
-# `test-runner` itself.
 [programs.test-runner]
 receives = ["netd", "launcher"]
 syscap = ["logread"]
diff --git a/tests/partclaimcase/system.toml b/tests/partclaimcase/system.toml
index 329040673ef..aec72307e05 100644
--- a/tests/partclaimcase/system.toml
+++ b/tests/partclaimcase/system.toml
@@ -29,9 +29,6 @@ syscap = ["rt"]
 # `device` because five of the guest binaries claim the keyboard or the mouse
 # and no manifest row can name them — they are not `[programs]` keys — and `dup`
 # because a claim moves and one boot runs several of them.
-# `logread` because the log gate reads the kernel's own records and cannot be
-# a spawned binary: a `SysCap` dup is not part of the namespace test-runner
-# hands down.
 # `power` because `endowment_denied` narrows it *away* to prove the two power
 # syscalls refuse a capability without it, which a capability that never
 # carried it would make vacuous. `run shutdown` does not use it: the applet asks
diff --git a/tests/sshdcase/system.toml b/tests/sshdcase/system.toml
index 09320cddb9b..81b55ede0bc 100644
--- a/tests/sshdcase/system.toml
+++ b/tests/sshdcase/system.toml
@@ -35,9 +35,6 @@ devices = ["pci:1af4:1041"]
 service = true
 receives = ["netd", "launcher"]
 
-# `logread` is the log gate's own: a spawned test binary does not inherit a
-# `SysCap` dup, so the gate that reads the kernel's records runs inside
-# `test-runner` itself.
 [programs.test-runner]
 receives = ["netd"]
 syscap = ["logread"]
diff --git a/tests/test-durations b/tests/test-durations
index 895ee3c3f96..152b686bac3 100644
--- a/tests/test-durations
+++ b/tests/test-durations
@@ -244,7 +244,6 @@ loader_watchdog_arms 10064
 locale_detect 9959
 locale_detect_unrecognized 160
 log_backing_read_error 4887
-log_conservation_smp1 4686
 log_flush_retry 20526
 log_nested_emit 5008
 log_partition_identity 9516
diff --git a/tests/toyos.rs b/tests/toyos.rs
index c22a68cdc8d..13f000c55e0 100644
--- a/tests/toyos.rs
+++ b/tests/toyos.rs
@@ -1081,7 +1081,7 @@ const MACHINE_TESTS: &[(&str, Sched, Tier)] = &[
     // them reads a clock. A loaded host makes the producers outrun the reader
     // further, which moves records from `read` into `lost` and leaves the law
     // exactly where it was.
-    ("log_conservation_smp1", Sched::Parallel, Tier::Weekly),
+    ("log_conservation_smp2", Sched::Parallel, Tier::Weekly),
     ("log_nested_emit", Sched::Parallel, Tier::Weekly),
     // The same interrupt one window earlier — between a record's shard-pointer
     // read and its `xadd` — and its negative control, which is the only reader
@@ -10132,8 +10132,8 @@ fn run_machine_test(
         // Body in `tests/common/iommu.rs`, same reason.
         "iommu_discovery" => common::iommu::iommu_discovery(test_config, c_bins, rust_bins),
         // Body in `tests/common/logread.rs`, so the hunk here stays one line.
-        "log_conservation_smp1" => {
-            common::logread::log_conservation_smp1(test_config, c_bins, rust_bins)
+        "log_conservation_smp2" => {
+            common::logread::log_conservation_smp2(test_config, c_bins, rust_bins)
         }
         "log_nested_emit" => common::logread::log_nested_emit(test_config, c_bins, rust_bins),
         "log_reserve_window" => {
diff --git a/toyos-abi/src/syscall.rs b/toyos-abi/src/syscall.rs
index f471ce741fe..42c89cc2edf 100644
--- a/toyos-abi/src/syscall.rs
+++ b/toyos-abi/src/syscall.rs
@@ -855,6 +855,9 @@ pub mod debug_action {
     /// shipped field, and the install and the close that follow are the shipped
     /// paths making the shipped decision (`kernel::object::handle`).
     pub const SLOT_TO_LAST_GENERATION: u64 = 20;
+    /// Emit one patterned kernel log record, `logstorm t=0 i=<arg> …`, whose
+    /// text the reader regenerates from its two numbers.
+    pub const LOG_PATTERNED: u64 = 21;
 }
 
 /// Every kind of kernel object, in the order the kernel's own `kobject!`
diff --git a/userland/test-runner/src/log_gate.rs b/userland/test-runner/src/log_gate.rs
index e1cbcaba56f..b8723d90d1e 100644
--- a/userland/test-runner/src/log_gate.rs
+++ b/userland/test-runner/src/log_gate.rs
@@ -14,26 +14,32 @@
 //! text, a lost record that is not counted fails the ledger, and a duplicated
 //! one fails it the other way.
 //!
-//! **Nothing this reader waits for is a record the ring may drop.** It used to
-//! read until every producer had said `logstorm done`, and that record is the
-//! last thing one producer writes rather than the last thing written to its
-//! shard: two producers placed on one CPU means the second's records lap the
-//! first's `done`, and the loop then waited for something that was never
-//! coming — twice in seven suites on the dev host, each time the whole 30 s
-//! ceiling in the fast tier. So the termination condition is the *cursor*: the
-//! log has been drained and nothing new has arrived for [`QUIET_READS`] reads
-//! and [`STORM_SETTLE`] of guest time. A `done` is a cross-check where it
-//! survived and is never waited on, and the same holds of `logstorm start` and
-//! of the nesting burst's own `done`. **The rule this shape exists to keep is
+//! **The storm is a thread of this process**, calling `SYS_DEBUG`'s
+//! `LOG_PATTERNED` once per record and counting each call after it returns.
+//! **Every interleave the verdict rests on is an event, not a schedule**: the
+//! producer stops after [`HANDOVER`] records until this reader has taken one;
+//! this reader then reads nothing until the producer has emitted enough more to
+//! lap its cursor on some shard, so `lost` is never zero; and the producer then
+//! emits until a read has taken storm records while its counter moved.
+//!
+//! **Nothing this reader waits for is a record the ring may drop.** The
+//! termination condition is the *cursor*: the log has been drained and nothing
+//! new has arrived for [`QUIET_READS`] reads, once the producer has returned
+//! from its last call. The nesting burst's own `done` is a cross-check where it
+//! survived and is never waited on. **The rule this shape exists to keep is
 //! general**: a workload whose liveness depends on a record the ring is allowed
 //! to drop is the same mistake wherever it appears.
 
 use std::collections::BTreeMap;
-use std::time::{Duration, Instant};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::mpsc::{self, Receiver, SyncSender};
+use std::sync::Arc;
+use std::thread::JoinHandle;
 
 use toyos::log::{LogTail, Record, MAX_LOG_SHARDS};
 use toyos::poller::{Poller, READABLE};
 use toyos::syscap::SysCap;
+use toyos_abi::syscall::debug_action::LOG_PATTERNED;
 
 /// The first sequence number any shard issues — one, so a slot nothing has ever
 /// written cannot read as record 0 of every shard on every boot.
@@ -41,8 +47,7 @@ use toyos::syscap::SysCap;
 const FIRST_SEQ: u64 = 1;
 
 /// Records per `SYS_LOG_READ`. Above the shard count, which the call refuses
-/// below, and far under a storm's rate — so the reader really is outrun and the
-/// loss path is reached rather than assumed.
+/// below.
 const BATCH: usize = 64;
 
 /// Empty reads in a row before the log is called quiet.
@@ -51,30 +56,8 @@ const BATCH: usize = 64;
 /// single empty read can land while a producer is inside its publication
 /// bracket: `drain_ordered` stops that shard and says nothing about it, so a
 /// ledger closed on the first empty read can be short by what was in flight.
-///
-/// **It is the whole termination condition now**, so what it costs when it is
-/// wrong is worth stating: a quiet run that lands mid-storm ends the read early
-/// and the verdict is computed over less of the workload. It cannot make the
-/// verdict *wrong* — the conservation law is over the sequence numbers this
-/// reader took and the loss the kernel counted for the same cursor, and both
-/// are a consistent snapshot at any point — and the non-vacuity clauses in
-/// [`verdict`] are what refuse a run that raced nothing. [`STORM_SETTLE`] is
-/// what makes an early end implausible rather than merely unlikely.
 const QUIET_READS: u32 = 8;
 
-/// How long after the last producer record the log must stay quiet before a
-/// storm counts as finished.
-///
-/// Eight empty reads are sixteen milliseconds of parks, and a producer stalled
-/// inside its publication bracket for that long — a vCPU that the host has not
-/// scheduled, which is the twelve-wide suite's ordinary state — takes its shard
-/// out of the merge and can leave every other shard drained. A hundred
-/// milliseconds of *guest* time on top costs one tenth of a second on three
-/// boots and buys an order of magnitude on that window. It is armed only once a
-/// producer's record has been seen, so an ordinary boot's gate ends on the
-/// quiet reads alone.
-const STORM_SETTLE: Duration = Duration::from_millis(100);
-
 /// How long a park on the log's readiness source waits before giving up on it.
 ///
 /// It is the gate's pacing as much as its wait: with nothing left to say the
@@ -84,9 +67,8 @@ const IDLE_NANOS: u64 = 2_000_000;
 /// How long the deterministic readiness round waits for its own record.
 ///
 /// Generous, because what it bounds is a scheduler getting round to a child's
-/// exit on a machine that has just run a storm on every CPU — not the post,
-/// which is one function call after the drain. A gate that timed out here would
-/// be reporting the host's load and not the kernel's.
+/// exit — not the post, which is one function call after the drain. A gate
+/// that timed out here would be reporting the host's load and not the kernel's.
 const READINESS_WAIT_NANOS: u64 = 2_000_000_000;
 
 /// The poll's token. One handle is watched, so it identifies the round rather
@@ -96,41 +78,30 @@ const LOG_TOKEN: u64 = 1;
 /// `kernel/src/log/storm.rs`'s `PAYLOAD`.
 const PAYLOAD: usize = 96;
 
+/// The producer id `LOG_PATTERNED`'s records declare.
+const STORM_PRODUCER: u64 = 0;
+
+/// Storm records emitted before the producer waits for this reader to take
+/// one: far under a shard's 512, so the ring still holds them when it arrives.
+const HANDOVER: u64 = 64;
+
+/// `kernel/src/log/shard.rs`'s `SHARD_RECORDS`: one more than `shards` times
+/// this, emitted between two reads, puts more than a shard's worth into one
+/// shard, whichever CPUs the producer ran on.
+const SHARD_RECORDS: u64 = 512;
+
 /// `kernel/src/log/nested.rs`'s `NEST_PRODUCER`: the burst an interrupt handler
 /// emits declares itself as this, so it goes through the same per-producer
 /// ledger and the same byte-for-byte regeneration as a storm's records.
 const NEST_PRODUCER: u64 = u64::MAX;
 
-
-/// One storm producer's ledger.
+/// One producer's ledger.
 #[derive(Default)]
 struct Producer {
-    /// The next index expected from this thread, and `None` before its first
+    /// The next index expected from this producer, and `None` before its first
     /// record.
     next: Option<u64>,
     read: u64,
-    /// What its own `done` record declared, once seen.
-    emitted: Option<u64>,
-    /// Shards this producer's records were found on. **More than one is a
-    /// producer that migrated mid-storm**, and on this kernel that is zero of
-    /// them and always will be: nothing switches a Ring 0 context out between
-    /// two instructions, so a producer cannot be moved off its CPU inside the
-    /// reservation window
-    /// (`kernel/src/log/storm.rs`'s header carries the measurement). It is
-    /// reported and asserted on by nothing, which is the honest shape for a
-    /// count whose only interesting value is unreachable.
-    shards: u32,
-    shard_mask: u32,
-}
-
-impl Producer {
-    fn mark_shard(&mut self, cpu: u16) {
-        let bit = 1u32 << (cpu as u32 % 32);
-        if self.shard_mask & bit == 0 {
-            self.shard_mask |= bit;
-            self.shards += 1;
-        }
-    }
 }
 
 /// One shard's ledger: the sequence numbers the kernel issued on that CPU.
@@ -145,12 +116,22 @@ struct ShardLedger {
     last_at_ns: u64,
 }
 
+/// The gate over whatever the boot's actuators write.
 pub fn run(cap: Option<&SysCap>) -> i32 {
+    report(cap, false)
+}
+
+/// The gate with a storm beside it.
+pub fn run_storm(cap: Option<&SysCap>) -> i32 {
+    report(cap, true)
+}
+
+fn report(cap: Option<&SysCap>, storm: bool) -> i32 {
     let Some(cap) = cap else {
         println!("log-gate: this program holds no system capability, so it holds no `logread`");
         return 1;
     };
-    match gate(cap) {
+    match gate(cap, storm) {
         Ok(()) => 0,
         Err(e) => {
             println!("log-gate: FAILED: {e}");
@@ -162,66 +143,37 @@ pub fn run(cap: Option<&SysCap>) -> i32 {
 struct Run {
     shards: [ShardLedger; MAX_LOG_SHARDS],
     producers: BTreeMap<u64, Producer>,
-    /// Producers this machine's storm has, once one of its records has been
-    /// seen. **Derived from the shard count rather than from an announcement**:
-    /// the storm starts inside the reader's own first `SYS_LOG_READ` and can
-    /// lap a shard before that call returns, so its opening line is a record
-    /// like any other and may be dropped. One thread per shard is what
-    /// `log::storm::start_once` spawns, and the cursor is what says how many
-    /// shards there are.
-    storm: Option<u32>,
-    /// What `logstorm start` or a producer's `done` declared, where one of
-    /// those records survived. They must agree. **A cross-check and never a
-    /// requirement**: both kinds are records like any other and the ring is
-    /// allowed to drop either, so [`verdict`] derives the count from the
-    /// highest index any producer reached when neither arrives.
-    declared: Option<u64>,
-    /// The nesting gate's declared burst, once its `done` has been read. Read
-    /// the same way, for the same reason.
+    /// The nesting gate's declared burst, once its `done` has been read. A
+    /// cross-check and never a requirement: the burst laps its shard, so the
+    /// ring is allowed to drop it.
     nest: Option<u64>,
     records: u64,
     reads: u64,
-    /// Producer records — a storm's or the nesting burst's — this reader took.
-    producer_records: u64,
-    /// Producer records read **strictly before the last batch that carried
-    /// one**, which is exactly "records this reader took while the producers
-    /// were still emitting": a later batch carrying a producer record proves
-    /// the workload had not finished when this one was read. **Zero would mean
-    /// this reader raced nothing**, which is the one way a green conservation
-    /// law says nothing at all.
-    ///
-    /// It needs no `done` and no clock, only the order of the batches.
+    /// Storm records taken, after the lap, by a read across which the
+    /// producer's counter moved.
     concurrent: u64,
-    /// When the last batch carrying a producer record was read. `None` until
-    /// one is, which is what leaves an ordinary boot's gate on the quiet reads
-    /// alone.
-    last_producer_at: Option<Instant>,
     /// Times the log's readiness source completed a poll.
     completions: u64,
 }
 
-fn gate(cap: &SysCap) -> Result<(), String> {
+fn gate(cap: &SysCap, storm: bool) -> Result<(), String> {
     let mut tail = LogTail::new();
     let mut buf = [Record::EMPTY; BATCH];
     let mut run = Run {
         shards: [ShardLedger::default(); MAX_LOG_SHARDS],
         producers: BTreeMap::new(),
-        storm: None,
-        declared: None,
         nest: None,
         records: 0,
         reads: 0,
-        producer_records: 0,
         concurrent: 0,
-        last_producer_at: None,
         completions: 0,
     };
 
-    // **Armed before the first read and kept armed**, which is what makes a
-    // completion deterministic rather than lucky: the first read is what starts
-    // the storm, so the records that answer this poll are committed after it was
-    // registered, and re-arming after every harvest means a post landing *during*
-    // the storm finds a pending poll rather than a gap.
+    // **Armed before the storm starts and kept armed**, which is what makes a
+    // completion deterministic rather than lucky: the storm's records are
+    // committed after this poll was registered, and re-arming after every
+    // harvest means a post landing *during* the storm finds a pending poll
+    // rather than a gap.
     //
     // **It used to arm only on an empty read, and that made the assertion
     // depend on the shape of the boot.** During a storm no read is empty, so the
@@ -232,7 +184,17 @@ fn gate(cap: &SysCap) -> Result<(), String> {
     // than about the readiness source. `min_complete` 0 with no timeout submits
     // and harvests without blocking, so this costs one syscall a round.
     let poller = Poller::new(1);
-    let mut armed = false;
+    poller.watch(cap, READABLE, LOG_TOKEN);
+    let mut armed = true;
+
+    let produced = Arc::new(AtomicU64::new(0));
+    let stop = Arc::new(AtomicBool::new(false));
+    let (handover, taken) = mpsc::sync_channel(1);
+    let (lap, lapped) = mpsc::sync_channel(1);
+    let mut handover = storm.then_some(handover);
+    let mut producer = storm
+        .then(|| spawn_producer(Arc::clone(&produced), Arc::clone(&stop), taken, lap));
+    let mut after_lap = false;
 
     let mut quiet = 0u32;
     loop {
@@ -246,9 +208,11 @@ fn gate(cap: &SysCap) -> Result<(), String> {
             armed = false;
         });
 
+        let before = produced.load(Ordering::Acquire);
         let batch = tail
             .read(cap, &mut buf)
             .map_err(|e| format!("SYS_LOG_READ refused a {BATCH}-record buffer: {e:?}"))?;
+        let moved = produced.load(Ordering::Acquire) != before;
         run.reads += 1;
         if batch.is_empty() {
             quiet += 1;
@@ -257,29 +221,26 @@ fn gate(cap: &SysCap) -> Result<(), String> {
             run.records += batch.len() as u64;
         }
 
-        // **The concurrency evidence, from the order of the batches alone.**
-        // Taken across the whole batch rather than per record: if this batch
-        // carried a producer record, then everything this reader had taken from
-        // a producer *before* it was taken while that producer was still
-        // emitting. The last such batch is what fixes the number, so it is
-        // assigned and not accumulated.
-        let producer_records_before = run.producer_records;
-        let shards = tail.shards();
+        let storm_before = storm_read(&run);
         for record in batch {
-            account(record, &mut run, shards)?;
+            account(record, &mut run)?;
+        }
+        let took = storm_read(&run) - storm_before;
+        if after_lap && moved && took > 0 {
+            run.concurrent += took;
+            stop.store(true, Ordering::Release);
         }
-        if run.producer_records > producer_records_before {
-            run.concurrent = producer_records_before;
-            run.last_producer_at = Some(Instant::now());
+        if let Some(handover) = handover.take_if(|_| storm_read(&run) > 0) {
+            let records = u64::from(tail.shards()) * SHARD_RECORDS + 1;
+            handover.send(records).map_err(|_| ended(producer.take()))?;
+            lapped.recv().map_err(|_| ended(producer.take()))?;
+            after_lap = true;
         }
 
-        // **The cursor decides, not a record.** Caught up, quiet for
-        // `QUIET_READS` reads, and — once a producer has been seen — quiet for
-        // `STORM_SETTLE` of guest time as well.
-        let settled = run
-            .last_producer_at
-            .is_none_or(|at| at.elapsed() >= STORM_SETTLE);
-        if quiet >= QUIET_READS && settled {
+        if producer.as_ref().is_some_and(JoinHandle::is_finished) {
+            join(producer.take())?;
+        }
+        if quiet >= QUIET_READS && producer.is_none() {
             break;
         }
         if batch.is_empty() {
@@ -295,16 +256,12 @@ fn gate(cap: &SysCap) -> Result<(), String> {
             });
         }
     }
+    let emitted = produced.load(Ordering::Acquire);
 
     // **The readiness source, observed deterministically rather than raced.**
-    // Every completion above is a `klogd` post landing while this poll happened
-    // to be pending, and during a storm that is a race against eight producers:
-    // it measured `wakes=1` at `--smp 4` and **zero** at `--smp 8` once
-    // `/system/bin/logd` was reading the cursor too, which is a red about
-    // scheduling. So if the storm produced none, make one — the shape
-    // `log_poll_outlives_a_close` already proves on this tree: a child that
-    // runs and exits commits `process.rs`'s `exit:` line, which is one kernel
-    // record from userland with no actuator and no privilege behind it.
+    // If the reads above completed no poll, make one: a child that runs and
+    // exits commits `process.rs`'s `exit:` line, which is one kernel record
+    // from userland with no actuator and no privilege behind it.
     if run.completions == 0 {
         let mut child = std::process::Command::new("/system/bin/echo")
             .arg("log-gate")
@@ -320,11 +277,67 @@ fn gate(cap: &SysCap) -> Result<(), String> {
         });
     }
 
-    verdict(&tail, &run)
+    verdict(&tail, &run, storm, emitted)
+}
+
+/// The storm: one kernel record per call, counted after each call returns.
+/// [`HANDOVER`] records, then the lap the reader names once it has taken one,
+/// then records until the reader sets `stop`.
+fn spawn_producer(
+    produced: Arc<AtomicU64>,
+    stop: Arc<AtomicBool>,
+    taken: Receiver<u64>,
+    lapped: SyncSender<()>,
+) -> JoinHandle<Result<(), String>> {
+    std::thread::spawn(move || {
+        let emit = || {
+            let index = produced.load(Ordering::Relaxed);
+            let answer = toyos_abi::syscall::debug_with(LOG_PATTERNED, index);
+            if answer != 0 {
+                return Err(format!(
+                    "SYS_DEBUG LOG_PATTERNED answered {answer:#x} at index {index}"
+                ));
+            }
+            produced.store(index + 1, Ordering::Release);
+            Ok(())
+        };
+        for _ in 0..HANDOVER {
+            emit()?;
+        }
+        let lap = taken.recv().map_err(|_| "the reader ended before it took a storm record")?;
+        for _ in 0..lap {
+            emit()?;
+        }
+        lapped.send(()).map_err(|_| "the reader ended before the storm lapped it")?;
+        while !stop.load(Ordering::Acquire) {
+            emit()?;
+        }
+        Ok(())
+    })
+}
+
+/// Why the producer's end of a channel closed: it returned, and its join says why.
+fn ended(producer: Option<JoinHandle<Result<(), String>>>) -> String {
+    match join(producer) {
+        Err(e) => e,
+        Ok(()) => "the producer returned before the storm lapped this reader".into(),
+    }
+}
+
+fn join(producer: Option<JoinHandle<Result<(), String>>>) -> Result<(), String> {
+    match producer.map(JoinHandle::join) {
+        None | Some(Ok(Ok(()))) => Ok(()),
+        Some(Ok(Err(e))) => Err(e),
+        Some(Err(_)) => Err("the producer thread panicked".into()),
+    }
+}
+
+fn storm_read(run: &Run) -> u64 {
+    run.producers.get(&STORM_PRODUCER).map_or(0, |p| p.read)
 }
 
 /// Put one record through both ledgers.
-fn account(record: &Record, run: &mut Run, shards: u32) -> Result<(), String> {
+fn account(record: &Record, run: &mut Run) -> Result<(), String> {
     let cpu = record.cpu as usize;
     let ledger = run.shards.get_mut(cpu).ok_or_else(|| {
         format!("a record claims cpu{cpu}, past the ABI's {MAX_LOG_SHARDS} shards")
@@ -366,15 +379,6 @@ fn account(record: &Record, run: &mut Run, shards: u32) -> Result<(), String> {
         ));
     }
 
-    // **What the batch-boundary concurrency evidence counts.** Every record
-    // either of this machine's two workloads wrote, `start` and `done` records
-    // included: the question it answers is "had the producers finished when
-    // this batch was read", and a `done` is a producer still working as much as
-    // a patterned record is.
-    if message.starts_with("logstorm ") || message.starts_with("lognest ") {
-        run.producer_records += 1;
-    }
-
     if let Some(rest) = message.strip_prefix("lognest done ") {
         let emitted = rest
             .split_whitespace()
@@ -392,38 +396,6 @@ fn account(record: &Record, run: &mut Run, shards: u32) -> Result<(), String> {
         // which is the ring's declared policy and not a loss of evidence.
         return Ok(());
     }
-    if let Some(rest) = message.strip_prefix("logstorm start ") {
-        // Informative and cross-checked where it survives; never depended on.
-        let (threads, records) = parse_start(rest)?;
-        if threads != shards {
-            return Err(format!(
-                "the storm declared {threads} producer(s) on a machine of {shards} shard(s)"
-            ));
-        }
-        run.storm = Some(shards);
-        run.declared.get_or_insert(records);
-        return Ok(());
-    }
-    if let Some(rest) = message.strip_prefix("logstorm done ") {
-        let (thread, emitted) = parse_done(rest)?;
-        run.storm = Some(shards);
-        match run.declared {
-            None => run.declared = Some(emitted),
-            Some(declared) if declared != emitted => {
-                return Err(format!(
-                    "producer t={thread} emitted {emitted} records where another \
-                     declared {declared}"
-                ))
-            }
-            Some(_) => {}
-        }
-        let producer = run.producers.entry(thread).or_default();
-        producer.mark_shard(record.cpu);
-        if producer.emitted.replace(emitted).is_some() {
-            return Err(format!("producer t={thread} said `done` twice"));
-        }
-        return Ok(());
-    }
     let Some(rest) = message.strip_prefix("logstorm t=") else {
         // An ordinary kernel record. It is in the shard ledger above, which is
         // where the conservation law is computed; it declares nothing this gate
@@ -432,6 +404,12 @@ fn account(record: &Record, run: &mut Run, shards: u32) -> Result<(), String> {
     };
 
     let (thread, index) = parse_record(rest)?;
+    if thread != STORM_PRODUCER && thread != NEST_PRODUCER {
+        return Err(format!(
+            "cpu{cpu} seq {} names producer t={thread}, which no gate runs",
+            record.seq
+        ));
+    }
     let expected = storm_message(thread, index);
     if message != expected {
         return Err(format!(
@@ -439,13 +417,7 @@ fn account(record: &Record, run: &mut Run, shards: u32) -> Result<(), String> {
             record.seq
         ));
     }
-    // The nesting burst declares itself past every shard, so it is a producer
-    // for the ledger's purposes and never one the storm is waiting on.
-    if thread != NEST_PRODUCER {
-        run.storm = Some(shards);
-    }
     let producer = run.producers.entry(thread).or_default();
-    producer.mark_shard(record.cpu);
     if let Some(next) = producer.next {
         if index < next {
             return Err(format!(
@@ -477,40 +449,6 @@ fn storm_message(thread: u64, index: u64) -> String {
     format!("logstorm t={thread} i={index} k={checksum:016x} {payload}")
 }
 
-fn parse_start(rest: &str) -> Result<(u32, u64), String> {
-    let mut threads = None;
-    let mut records = None;
-    for word in rest.split_whitespace() {
-        if let Some(v) = word.strip_prefix("threads=") {
-            threads = v.parse::<u32>().ok();
-        }
-        if let Some(v) = word.strip_prefix("records=") {
-            records = v.parse::<u64>().ok();
-        }
-    }
-    match (threads, records) {
-        (Some(t), Some(r)) => Ok((t, r)),
-        _ => Err(format!("`logstorm start` is unreadable: {rest}")),
-    }
-}
-
-fn parse_done(rest: &str) -> Result<(u64, u64), String> {
-    let mut thread = None;
-    let mut emitted = None;
-    for word in rest.split_whitespace() {
-        if let Some(v) = word.strip_prefix("t=") {
-            thread = v.parse::<u64>().ok();
-        }
-        if let Some(v) = word.strip_prefix("emitted=") {
-            emitted = v.parse::<u64>().ok();
-        }
-    }
-    match (thread, emitted) {
-        (Some(t), Some(e)) => Ok((t, e)),
-        _ => Err(format!("`logstorm done` is unreadable: {rest}")),
-    }
-}
-
 fn parse_record(rest: &str) -> Result<(u64, u64), String> {
     let mut words = rest.split_whitespace();
     let thread = words
@@ -527,7 +465,7 @@ fn parse_record(rest: &str) -> Result<(u64, u64), String> {
 
 /// The conservation law, and everything the gate prints for a reader of its
 /// output.
-fn verdict(tail: &LogTail, run: &Run) -> Result<(), String> {
+fn verdict(tail: &LogTail, run: &Run, storm: bool, emitted: u64) -> Result<(), String> {
     let seen: Vec<usize> =
         (0..MAX_LOG_SHARDS).filter(|&i| run.shards[i].first.is_some()).collect();
     if seen.is_empty() {
@@ -573,110 +511,21 @@ fn verdict(tail: &LogTail, run: &Run) -> Result<(), String> {
         ));
     }
 
-    let mut emitted_total = 0u64;
-    let mut read_total = 0u64;
-    let mut migrated = 0u64;
-    let mut said_done = 0u64;
-    let mut unseen = 0u64;
-    if let Some(threads) = run.storm {
-        // **What every producer emitted, from a record where one survived and
-        // from the ledger where none did.** `logstorm start` is written before
-        // the first producer runs and each `done` after that producer's last
-        // record; the storm laps every shard twice, so the ring is allowed to
-        // drop any of them and this gate may not wait for one. The floor is the
-        // highest index any producer reached — a producer emits `0..count`, so
-        // the highest index seen plus one is a count no producer exceeded, and
-        // the producer that finished last on a shard has its final records at
-        // the newest end of it.
-        let derived = run
-            .producers
-            .iter()
-            .filter(|(&t, _)| t != NEST_PRODUCER)
-            .filter_map(|(_, p)| p.next)
-            .max();
-        let declared = match (run.declared, derived) {
-            (Some(declared), _) => declared,
-            (None, Some(derived)) => derived,
-            (None, None) => {
-                return Err("storm records were read and none of them named an index".into())
-            }
-        };
-        for thread in 0..threads as u64 {
-            let Some(producer) = run.producers.get(&thread) else {
-                // **A producer this reader never saw at all is the ring's
-                // declared policy and not a failure**, and this used to be a
-                // hard error. Two producers placed on one CPU write one shard,
-                // and 1,024 records from the second lap all 1,024 of the first:
-                // measured 2 of 7 full suites on the dev host, 2026-08-15, with
-                // 2,582 records overwritten in a shard on the run that produced
-                // it. Refusing it would be refusing the behaviour under test.
-                //
-                // It is not free either — see the ledger check below, which is
-                // what stops "the reader saw nothing of it" from covering a
-                // producer that never ran.
-                unseen += 1;
-                emitted_total += declared;
-                continue;
-            };
-            // **A cross-check where the record survived, never a requirement.**
-            // A producer whose `done` was lapped is a producer the ring
-            // dropped a record of, which is the behaviour under test.
-            if let Some(emitted) = producer.emitted {
-                said_done += 1;
-                if emitted != declared {
-                    return Err(format!(
-                        "producer t={thread} emitted {emitted} records against a declared \
-                         {declared}"
-                    ));
-                }
-            }
-            if producer.read > declared {
-                return Err(format!(
-                    "producer t={thread} emitted {declared} records and this reader took {}",
-                    producer.read
-                ));
-            }
-            if producer.next.is_some_and(|next| next > declared) {
-                return Err(format!(
-                    "producer t={thread} answered index {} of a declared {declared}",
-                    producer.next.unwrap_or(0) - 1
-                ));
-            }
-            emitted_total += declared;
-            read_total += producer.read;
-            if producer.shards > 1 {
-                migrated += 1;
-            }
-        }
-        // **A producer nobody saw has to be one the ring dropped, and the
-        // ledger is what says so.** `unseen` producers emitted `declared`
-        // records each and none of them was read, so at least that many
-        // sequence numbers must be among the ones the kernel counted lost. It
-        // is a necessary condition rather than an attribution — the cursor's
-        // `lost` is per shard and does not name producers — and it is what
-        // separates "the ring lapped its whole run", which is the behaviour
-        // under test, from "that thread never ran", which is a kernel that did
-        // not spawn what it said it did.
-        if unseen > 0 {
-            let owed = unseen * declared;
-            if reported < owed {
-                return Err(format!(
-                    "{unseen} producer(s) emitted {declared} record(s) each and this reader took                      none of them, while the kernel counted {reported} lost in all — a producer                      can only be invisible because its records were dropped, and the ledger does                      not account for the {owed} that would take"
-                ));
-            }
-        }
+    let read_total = storm_read(run);
+    if storm {
         if read_total == 0 {
             return Err("the storm ran and this reader read none of it".into());
         }
-        if run.concurrent == 0 {
-            return Err(
-                "every record was read after the storm had finished, so this reader raced nothing"
-                    .into(),
-            );
+        let next = run.producers.get(&STORM_PRODUCER).and_then(|p| p.next).unwrap_or(0);
+        if next > emitted {
+            return Err(format!(
+                "the storm answered index {} of {emitted} emitted",
+                next - 1
+            ));
         }
         // The readiness source, asserted where it is reachable: the poll was
-        // armed before the read that starts the storm, so the records that
-        // answer it were committed after it was registered.
+        // armed before the storm started, so the records that answer it were
+        // committed after it was registered.
         if run.completions == 0 {
             return Err(
                 "the log's readiness source completed no poll — not across the storm, and not on \
@@ -687,11 +536,10 @@ fn verdict(tail: &LogTail, run: &Run) -> Result<(), String> {
     }
 
     if let Some(burst) = run.producers.get(&NEST_PRODUCER) {
-        // The burst's own `done` is read the same way a storm's is: a
-        // cross-check where it survived, and the ledger's own floor where it
-        // did not. The burst laps its shard by construction, so a reader
-        // that required that record would be requiring one the design says may
-        // go.
+        // The burst's own `done` is a cross-check where it survived, and the
+        // ledger's own floor where it did not. The burst laps its shard by
+        // construction, so a reader that required that record would be
+        // requiring one the design says may go.
         let declared = match (run.nest, burst.next) {
             (Some(declared), _) => declared,
             (None, Some(next)) => next,
@@ -709,14 +557,9 @@ fn verdict(tail: &LogTail, run: &Run) -> Result<(), String> {
             ));
         }
         println!(
-            // `nest_shards` and not `shards`: the line below reports the
-            // machine's shard count under that name, and two lines defining one
-            // name is a host-side reader that silently takes whichever came
-            // last (`tests/common/logread.rs`).
-            "log-gate: nest declared={declared} read={} dropped={} nest_shards={}",
+            "log-gate: nest declared={declared} read={} dropped={}",
             burst.read,
             declared - burst.read,
-            burst.shards,
         );
     }
 
@@ -727,19 +570,12 @@ fn verdict(tail: &LogTail, run: &Run) -> Result<(), String> {
         run.reads,
         seen.len()
     );
-    if run.storm.is_some() {
-        // `done=` is the count of producers whose own `done` record survived
-        // the ring, and it is evidence rather than an assertion — the gate no
-        // longer waits for one and the number is what says how often the ring
-        // ate one. Bare, not `k/n`: the host's reader takes the denominator of
-        // an `a/b` field as the producer count and two of those would collide
-        // (`tests/common/logread.rs`).
+    if storm {
         println!(
-            "log-gate: storm emitted={emitted_total} read={read_total} dropped={} \
-             concurrent={} migrated={migrated}/{} done={said_done} unseen={unseen} wakes={}",
-            emitted_total - read_total,
+            "log-gate: storm emitted={emitted} read={read_total} dropped={} \
+             concurrent={} wakes={}",
+            emitted - read_total,
             run.concurrent,
-            run.producers.len(),
             run.completions,
         );
     }
diff --git a/userland/test-runner/src/main.rs b/userland/test-runner/src/main.rs
index 85407cdcdf0..69c328db6cb 100644
--- a/userland/test-runner/src/main.rs
+++ b/userland/test-runner/src/main.rs
@@ -29,6 +29,7 @@ use toyos::syscap::SysCap;
 /// collision is about does not exist in one.
 const BUILTINS: &[(&str, fn(Option<&SysCap>) -> i32)] = &[
     ("log-gate", log_gate::run),
+    ("log-storm", log_gate::run_storm),
     ("log-close", log_close::run),
     ("kbd-close", kbd_close::run),
 ];