fix: apply lossless push in PoolObjectNode too

PoolNode and PoolObjectNode each have their own push_one_out(). The previous commit only reached PoolNode's copy, so ObjectNode-based pipelines — the common case — still dropped on overflow with lossless enabled. Verified end to end: a 2300-frame source now reports exactly 2300 frames at every stage, where it previously delivered 774 to the second node. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
feat: opt-in lossless (blocking) output for nodes and fanouts
2026-07-25 23:12:51 +02:00 · 2026-07-25 23:03:56 +02:00
34 changed files with 403 additions and 4542 deletions
@@ -27,6 +27,3 @@ Thumbs.db
 # Claude Code local settings
 .claude/settings.local.json
 include/kpn/ort_cache/
 build-tsan/
 build*/
@@ -1,413 +0,0 @@
 # Performance investigation plan: fanout dispatch cost and deep-chain oversubscription
 **Status:** phase 0 implemented; gate not yet cleared
 **Date:** 2026-08-06 (phase 0 landed 2026-08-06)
 **Baseline:** master @ 3b67b7e
 **Machine:** 20 cores, GCC 16.1.1, TBB 2023.1.0, AC power, `performance` governor
 **Data:** 7 full benchmark passes, medians reported below
 ---
 ## 1. What was measured
 Throughput, items/sec, median of 7 passes. `N` is the sample size the harness
 uses for that row; it is what determines whether a row can be trusted at all.
 ### work_us = 10
 | row | KPN it/s | TBB it/s | TBB faster | N | reliable? |
 |---|---|---|---|---|---|
 | chain-1 | 89381 | 91350 | +2.2% | 3000 | solid |
 | chain-2 | 83759 | 87017 | +3.9% | 1000 | solid |
 | chain-4 | 83725 | 85973 | +2.7% | 1000 | solid |
 | chain-8 | 79730 | 81090 | +1.7% | 1000 | solid |
 | **chain-16** | 53484 | 68745 | **+28.5%** | 200 | weak |
 | **chain-32** | 32780 | 45030 | **+37.4%** | 200 | weak |
 | **wide-4** | 84906 | 95137 | **+12.0%** | 3000 | solid |
 | diamond-4 | 84826 | 86772 | +2.3% | 1000 | solid |
 ### work_us = 100
 Everything except chain-16/32 falls within ±3.4%, with KPN often ahead
 (chain-1 −0.6%, chain-4 −1.7%, chain-8 −3.4%, diamond −2.6% — negative means
 KPN faster). chain-16 is +12.5% and chain-32 +18.6%, both at N=50 and
 therefore unusable.
 ### Two deficits, different causes
 1. **Fanout, +12%.** Solidly measured. `wide-4` performs ~5 node dispatches
   per item; the gap works out to a fixed ~250 ns per dispatch, consistent
   with `chain-1`'s ~290 ns over a single dispatch. This is dispatch
   efficiency.
 2. **Deep chains, +28–37%.** The gap is 1.7–3.9% through depth 8, then jumps
   to 28.5% at depth 16 and 37.4% at depth 32. That is a cliff at core count,
   not a linear per-dispatch cost. `Node<>` owns a private `ThreadPool(1)`
   (`include/kpn/node.hpp:21`), so a depth-32 chain spawns 32 OS threads on
   20 cores. TBB bounds its worker count by hardware concurrency regardless of
   graph size.
 ### Scope note
 At 100 µs+ per node KPN is at parity or ahead. The repository's own examples
 (OpenCV cellshade, frame sources, scene-actor extraction) do milliseconds of
 work per node, where a 290 ns dispatch cost is roughly one part in thirty
 thousand. Everything in this document matters only for fine-grained pipelines.
 ---
 ## 2. Phase 0 — the gate that comes first
 **Is there a target workload with sub-30 µs nodes?**
 If no such workload exists or is planned, the correct output of this document
 is section 3 (harness) plus a README correction, and nothing else. Optimising
 for a benchmark regime the project does not operate in is not worth the risk
 described in section 6.
 ---
 ## 3. Prerequisite — make the harness able to answer
 None of the questions below are decidable with the current harness.
 `benchmarks/bench_pipeline.cpp` shrinks the sample count as work per item
 grows, so the rows under investigation run 50–200 items and swing 4–8×
 run to run.
 | id | change | why |
 |---|---|---|
 | M1 | `items_for()` → fixed floor, e.g. `max(2000, …)`, independent of work_us and depth | deep rows are currently unmeasurable |
 | M2 | report items/sec as the primary metric; keep derived overhead as secondary | overhead is `elapsed − work`, a difference of large numbers; it magnifies noise roughly 10× |
 | M3 | K in-process repetitions per config; report median and IQR | one shot per config is the root of the present noise |
 | M4 | discard a warm-up repetition | first-touch page faults, thread spin-up |
 | M5 | extend `pool_sizes[]` to `{1,2,4,8,16,20}` | currently `{1,2,4}` — the configuration the README recommends is never run |
 | M6 | record nproc, governor and AC state in the CSV header | run-to-run attribution |
 **Acceptance:** the same configuration run 7× lands within ±5% on every row.
 Until that holds, no number below should be acted on.
 This touches only the benchmark, not the library.
 ### Status — implemented 2026-08-06
 All of M1–M6 are in `benchmarks/bench_pipeline.cpp`, plus a CLI so the phase-1
 experiments are invocations rather than edits (`--depths`, `--pools`,
 `--work`, `--topos`, `--modes`, `--reps`, `--target-sec`, `--min-items`).
 M1 is not a fixed floor but a time budget with a floor: sample size derives
 from `work_us × stages / units`, the steady-state throughput bound, then
 clamps to `[--min-items, --max-sec]`. A flat 2000-item floor would have made
 `chain-32` on a 1-thread pool at 1000 µs a 64-second row; the ceiling keeps
 such rows short and reports their true `N` so a short row is visible rather
 than silent. The old ladder's error was treating depth as a throughput cost —
 in a pipeline, depth beyond the core count costs throughput, below it only
 latency.
 Also added, ahead of schedule because it is free: `ru_nivcsw` / `ru_nvcsw` per
 item are captured around every timed region, so **A3 is now a matter of
 reading a column** rather than a separate experiment.
 `scripts/bench_repro_check.py` runs the acceptance criterion directly — K
 passes, per-row deviation from the median, non-zero exit if any row exceeds
 tolerance.
 **Gate not yet cleared.** A 3-pass run of `chain-{1,8}` at 10 µs on the
 development laptop (20 cores, **powersave governor, on battery** — the header
 now records this) lands every row within 0.7%, against the 4–8× swings this
 section describes. That is encouraging but is not the acceptance run: it must
 be 7 passes over the full row set on the reference machine.
 **Provisional and not to be acted on:** in that same run `chain-16` private
 was 6% behind TBB, not the 28.5% in the table above. If that survives the
 real acceptance run, the deep-chain deficit is substantially a measurement
 artefact of the N=200 rows and workstream A shrinks accordingly.
 ---
 ## 4. Workstream A — deep chains
 **Hypothesis:** the deficit is thread oversubscription from the private-pool
 model, not dispatch cost.
 ### Investigation
 | id | experiment | falsifies the hypothesis if |
 |---|---|---|
 | A1 | sweep depth 8, 12, 16, 20, 24, 32 at 10 µs, private pools | the cliff is not near nproc |
 | A2 | repeat A1 under `taskset -c 0-7` | the cliff does **not** move to ~depth 8 |
 | A3 | `getrusage(RUSAGE_SELF).ru_nivcsw` per item, depth 8 vs 32 | involuntary context switches do not scale with depth |
 | A4 | chain-16/32 on a shared pool sized 16 and 20, vs private and vs TBB | a correctly sized shared pool does not recover the gap |
 A2 is decisive and costs one run: if the cliff tracks the core count, the
 mechanism is established.
 ### Improvement, conditional on A4
 If a correctly sized shared pool closes the gap, this is not an optimisation
 problem — the mechanism already exists and is simply not the default:
 - **A5** — change `Network`'s default from per-node private pools to a single
  shared pool sized `hardware_concurrency()`. Users should not have to know.
 - **A6** — emit a diagnostic when total node threads exceed
  `hardware_concurrency()`.
 - **A7** — README: state the threshold, with the measured cliff.
 A5 is a change to the default execution model and must clear section 6 in full.
 ### A5 now has a prerequisite (from B1/B2, 2026-08-06)
 The dispatch microbenchmark measured what a shared pool costs per dispatch,
 and it is not free: **466 ns on a private `ThreadPool(1)` against ~1.7 µs on a
 shared pool of 4**, because round-robin submission wakes a sleeping worker on
 every dispatch (see §5). A5 as written would therefore make every graph that
 currently fits inside its core count roughly 3–4× *worse* per dispatch, in
 exchange for fixing graphs that exceed it.
 **A5 must not land before the wake cost does.** The order is B9/B5 first,
 then A5, and A4 must be read with this in mind: if a shared pool "recovers the
 gap" at depth 32, check what it costs at depth 4 in the same run before
 changing any default.
 This partially inverts the prediction in §7: workstream A is not purely a
 default-and-documentation change, because the default it would switch to is
 currently the slower one per dispatch.
 ---
 ## 5. Workstream B — fanout dispatch cost
 **Hypothesis:** a fixed ~250 ns per node dispatch, paid ~5× per item in
 `wide-4`. Unlike workstream A, this genuinely is dispatch efficiency.
 Estimated budget for ~290 ns, per item — **estimates, to be replaced by B3**:
 | cost | est. |
 |---|---|
 | `shared_lock(lifecycle_mx_)` in `submit()` | 20–40 ns |
 | `queues_[target]->mx` lock/unlock | 20–40 ns |
 | `priority_queue` push + pop (heap ops, `std::function` moves) | 50–100 ns |
 | `{ lock_guard lk(cv_mx_); }` + `notify_one()` | 20–40 ns, or µs if a worker actually sleeps |
 | 2–3 × `clock_t::now()` in `fire_once` | 50–75 ns |
 | gate CAS + ~6 stats atomics | 30–60 ns |
 ### Investigation — measure before touching anything
 - **B1** — microbenchmark submit→execute turnaround for a null task on
  `ThreadPool(1)` and `ThreadPool(4)`. Yields ns/dispatch directly, in seconds
  rather than minutes.
 - **B2** — **does a worker actually sleep per item?** Count `cv_.wait` returns,
  or `strace -c -f -e futex`. The entire spin-window hypothesis depends on
  this; if workers are not sleeping, B5 is worthless and drops off the list.
 - **B3** — ablation, one variant per suspected cost, each measured against B1
  rather than guessed at:
 | variant | suspected cost |
 |---|---|
 | stats and clock calls compiled out | 2–3 × `clock_t::now()` plus ~6 atomics per firing |
 | `priority_queue` → FIFO ring | heap operations, `std::function` moves |
 | `shared_lock(lifecycle_mx_)` removed (**measurement only, unsafe**) | `include/kpn/scheduler.hpp:113` |
 | bounded spin before sleeping | `include/kpn/scheduler.hpp:210-227` |
 ### B1/B2 — first results, 2026-08-06
 `benchmarks/bench_dispatch.cpp` answers both without touching the library.
 Sleeping is inferred from `ru_nvcsw`: a thread blocking on a condition
 variable books a voluntary context switch, so voluntary switches per task is
 sleeps per task. Three modes, because "the cost of a dispatch" is three
 numbers: `latency` (idle pool, one task in flight), `batch` (submit flat out,
 drain once), `steady` (the task resubmits its successor, as `fire_once` does).
 Laptop, powersave, battery, 3 reps — **the nanoseconds are provisional; the
 sleep counts are structural and will hold.** `steady`, 10 µs payload:
 | pool threads | ns/dispatch | sleeps/task |
 |---|---|---|
 | 1 | 466 | **0.00** |
 | 2 | 1494 | 0.97 |
 | 4 | 1722 | 1.00 |
 | 8 | 1996 | 1.00 |
 **B1 is answered and the abandon criterion is not met.** A `ThreadPool(1)`
 dispatch is 291 ns for a null task, 466 ns with a payload — against the ~290 ns
 the section-1 budget estimated for `chain-1`. The estimate was good. Dispatch
 cost is not already under 100 ns, so workstream B stays alive.
 **B2 is answered, and the answer is conditional — which the question did not
 anticipate.** It is not "do workers sleep?" but "which pool?":
 - On a private `ThreadPool(1)` — the `Node<>` default — the worker **never**
  sleeps. It resubmits into its own queue and finds the work already there.
 - On any pool of 2 or more, a worker sleeps **exactly once per task**.
 `submit()` round-robins (`next_.fetch_add(1) % thread_count_`,
 `scheduler.hpp:131`), so on a shared pool every task is handed to a *different*
 worker, which is asleep, and every single dispatch pays a futex wake. That is
 the entire 466 ns → 1.7 µs difference.
 Consequently **B5 (bounded spin) is worthless for the default configuration**
 and is the highest-value item for shared pools. It does not drop off the list,
 it moves onto a different one.
 ### B9 — submit-to-self affinity (new, not in the original plan)
 If a `submit()` originating on a pool worker pushed to *that worker's own*
 queue instead of round-robining, the shared pool would inherit the property
 that makes `ThreadPool(1)` fast: the work is already local when the worker
 loops, so no wake. This is roughly what TBB does, and it plausibly subsumes
 most of B5 at lower risk — it changes task placement, not the sleep/wake
 protocol that the August wedge fixes hardened. Work stealing already exists to
 correct the resulting imbalance.
 Measure before believing it: an affinity policy can starve peers, and
 `try_steal` only rebalances when a peer goes idle.
 ### Improvement — only what B3 shows pays
 1. **B4 — compile-time-optional instrumentation.** No concurrency risk; the
   only item here that cannot reintroduce a wedge. Worth doing regardless.
 2. **B5 — bounded spin before sleeping**, mirroring the channel's existing
   `spin_count_` (~4 µs). Note the tension: b9698fa deliberately moved from
   "spin whenever any task runs" to "sleep as soon as nothing is queued" in
   order to fix pathological spinning. A *bounded* window is the middle
   ground; unbounded spin would undo that fix.
 3. **B6 — cheaper queue on the common path.** A private pool holds ≤1–2 tasks;
   `priority_queue<Task>` is heavy for that.
 4. **B7 — batched firing.** `fire_once` processes one token then re-submits;
   looping while inputs stay ready, bounded, amortises the submit, gate CAS
   and wake. The largest algorithmic win, but it changes latency and
   interacts with `compute_priority()`.
 5. **B8 — `lifecycle_mx_` off the hot path.** Last, and possibly never. It is
   load-bearing: it prevents `submit()` racing `stop()`'s `queues_.clear()`,
   a documented segfault reproducible "about 12 runs in 20".
 **Abandon criteria:** if B1 shows dispatch cost already under ~100 ns, or the
 best surviving variant buys under 5%, stop and document the finding.
 ---
 ## 6. Guardrails
 Both workstreams modify the machinery responsible for roughly twenty wedge
 fixes in August 2026, plus the lost wake fixed in 6802328. Every change:
 1. **146/146** ctest, examples included.
 2. **Wedge soak before and after** — `benchmarks/repro_wedge.cpp`, ≥50k
   iterations clean. Reference point: the pre-6802328 code wedged 5/5 inside
   45 s, at iterations 149, 1249, 332, 1740 and 493.
 3. **ThreadSanitizer** on scheduler and pool_node tests for any change to
   either.
 4. **One change at a time**, measured independently. Bundling is how the
   August audit became twenty commits.
 5. **G1 — wire the reproducer in as an opt-in CTest stress target**
   (e.g. `-L soak`) so that performance work cannot silently reintroduce a
   wedge. This should land before either workstream starts.
 ### G1 — implemented 2026-08-06
 `tests/soak_wedge.cpp` (supersedes `benchmarks/repro_wedge.cpp`, which was
 never wired into any build and can be deleted). Always compiled so it cannot
 rot; its CTest cases register only under `-DKPN_ENABLE_SOAK_TESTS=ON`, so the
 default `ctest` count is unchanged.
 ```
 cmake -B build -DKPN_ENABLE_SOAK_TESTS=ON -DKPN_SOAK_ITERS=50000
 cmake --build build --target kpn_soak_wedge
 ctest --test-dir build -L soak
 ```
 Two cases: `soak.wedge.pool` (depth 4, 4 threads — the configuration the
 August wedges were reproduced on) and `soak.wedge.private` (depth 8, one pool
 per node — the model workstream A would change). Both parameterised, so
 A5-style changes can be soaked at the depth that matters.
 A wedge is a hang, and a hang under CTest is an unattributable timeout, so the
 binary carries a watchdog: if an iteration stops making progress for
 `--watchdog-sec` it aborts naming the iteration and the phase (`pushed`,
 `drained`, `nodes stopped`, `pool stopped`). Measured cost: ~13 ms per
 iteration, so the 50k-iteration guardrail is ~11 minutes.
 **Guardrail 1 needs a correction.** The stated reference is 146/146; the
 tests-only configuration used here reports **136/136 passing**, and neither
 `examples/` nor `python/` registers any `add_test`. The true reference count
 must be pinned down before it is used to certify a change.
 ---
 ## 7. Sequencing
 | phase | contents | gate to proceed | state |
 |---|---|---|---|
 | 0 | workload question; M1–M6; G1 | ±5% reproducibility achieved | **tooling done**, acceptance run outstanding |
 | 1 | A1–A4 | A2 confirms the cliff tracks core count | harness supports it; not run |
 | 2 | A5–A7, or documentation only | A4 shows a shared pool recovers the gap | **now gated on B9/B5** |
 | 3 | B1–B3 | B2 answers the sleep question | **B1/B2 answered**; B3 outstanding |
 | 4 | B4, then whichever of B5–B8 survived B3 | each ≥5% and soak-clean | B5 rescoped to shared pools |
 B1/B2 ran early because the microbenchmark cost seconds rather than minutes,
 and the result reordered phases 2 and 4 — the shared-pool default now depends
 on the wake cost being fixed first. Phase 1 is unchanged but its A4 row needs
 a shallow-depth control, per §4.
 ### Reproducing this
 ```
 cmake -B build_bench -DKPN_BUILD_BENCHMARKS=ON -DCMAKE_BUILD_TYPE=Release
 cmake --build build_bench -j
 # Phase 0 acceptance — must pass before any number below is acted on
 python3 scripts/bench_repro_check.py ./build_bench/benchmarks/bench_pipeline \
    --passes 7 --tolerance 5 -- --work=10,100 --reps=5
 # B1/B2
 ./build_bench/benchmarks/bench_dispatch --threads=1,2,4,8,20 --reps=5 \
    | tee dispatch.csv
 # A1/A2 — the depth sweep, and the same under taskset to move the cliff
 ./build_bench/benchmarks/bench_pipeline --work=10 --topos=chain \
    --depths=8,12,16,20,24,32 --modes=priv,tbb --reps=5 | tee a1.csv
 taskset -c 0-7 ./build_bench/benchmarks/bench_pipeline --work=10 \
    --topos=chain --depths=4,6,8,10,12,16,32 --modes=priv,tbb --reps=5 | tee a2.csv
 # A4 — shared pool sized to the machine, against private and TBB.
 # Include a shallow depth: A5's risk is what a shared pool costs when the
 # graph already fits in its cores.
 ./build_bench/benchmarks/bench_pipeline --work=10 --topos=chain \
    --depths=4,16,32 --pools=16,20 --reps=5 | tee a4.csv
 ```
 Check the `# governor=` line in each CSV before trusting it. A3 needs no
 separate run: `ivcsw_per_item` is a column in every row above.
 **Success criteria**
 - chain-32 @10 within 10% of TBB in the recommended configuration
 - wide-4 @10 within 5% of TBB
 - zero wedges across 100k soak iterations
 **Prediction, recorded so it can be proven wrong:** workstream A resolves into
 a default-and-documentation change rather than an optimisation, and workstream
 B yields 5–10% on fanout from B4 and B5, with the remainder not worth the risk.
 **Prediction, revised 2026-08-06 after B1/B2** — the original is already half
 wrong and is left above unedited:
 - Workstream A does *not* resolve into a documentation change, because the
  shared pool it would recommend costs 3–4× more per dispatch than the private
  default. It resolves into B9 first.
 - The largest single win is not B4, B5 or B7 but **B9, submit-to-self
  affinity**: one sleep per dispatch is being paid on every shared pool, and
  eliminating it is worth roughly 1.2 µs per dispatch — far more than the
  5–10% predicted for fanout.
 - Standing: `chain-16`'s 28.5% deficit is a measurement artefact of N=200.
 ---
 ## 8. Related correction
 Independently of the above, the README's TBB comparison overstates its case.
 The claim that KPN++ beats TBB "for every chain and diamond topology at
 100 µs/node" is not supported: at 100 µs only chain-1 and diamond lean KPN,
 while chain-16, chain-32 and wide-4 lean TBB. The tables are also quoted in
 derived overhead, which magnifies small differences — the same rows expressed
 as throughput are mostly within a few percent. Restating them in items/sec
 would be both more accurate and more favourable.
@@ -4,13 +4,6 @@ add_executable(bench_pipeline bench_pipeline.cpp)
 target_link_libraries(bench_pipeline PRIVATE kpn)
 target_compile_options(bench_pipeline PRIVATE -O3 -march=native)
 # Dispatch microbenchmark (PERF_PLAN B1/B2): ns per ThreadPool dispatch, and
 # whether a worker actually sleeps per task. No TBB comparison — it measures
 # KPN's own scheduler, not a competitor.
 add_executable(bench_dispatch bench_dispatch.cpp)
 target_link_libraries(bench_dispatch PRIVATE kpn)
 target_compile_options(bench_dispatch PRIVATE -O3 -march=native)
 find_package(TBB QUIET)
 if(TBB_FOUND)
    target_link_libraries(bench_pipeline PRIVATE TBB::tbb)
@@ -1,307 +0,0 @@
 // Dispatch microbenchmark — PERF_PLAN B1 and B2.
 //
 // B1 asks what a single ThreadPool dispatch costs. B2 asks whether a worker
 // actually sleeps per item, because the whole spin-window hypothesis (B5)
 // depends on the answer: if workers are not sleeping, a spin window buys
 // nothing and drops off the list.
 //
 // Both are answered here without touching the library. Sleeping is inferred
 // from ru_nvcsw — a thread blocking on a condition variable books a voluntary
 // context switch — so `vcsw/task` near 1.0 means a sleep per dispatch and near
 // 0 means the worker never went to sleep at all.
 //
 // Three modes, because "the cost of a dispatch" is three different numbers:
 //
 //   latency — one task in flight, pool idle in between. The worker is asleep
 //             at every submission, so this is dispatch cost *including* a
 //             wake. Worst case, and the case a spin window would attack.
 //
 //   batch   — submit K no-op tasks flat out, then drain. The worker is never
 //             idle, so this is the amortised floor: queue and heap operations
 //             with no wake at all. Reports the producer-side submit() cost
 //             separately from end-to-end throughput.
 //
 //   steady  — the task resubmits its successor, one in flight, each doing
 //             --work-us of work. This is what a KPN node actually does:
 //             fire_once processes a token and resubmits. On ThreadPool(1) the
 //             worker resubmits to its own queue; on ThreadPool(4) round-robin
 //             hands the task to a *different* worker, which may be asleep.
 //             That difference is the fanout cost wide-4 pays ~5x per item.
 //
 // Usage: ./bench_dispatch [--threads=1,2,4] [--mode=latency,batch,steady]
 //                         [--tasks=200000] [--work-us=0] [--reps=5] [--warmup=1]
 #include <kpn/kpn.hpp>
 #include "bench_env.hpp"
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
 #include <cstdio>
 #include <cstdlib>
 #include <mutex>
 #include <string>
 #include <vector>
 using namespace kpn;
 using sclock = std::chrono::steady_clock;
 struct Opts {
    std::vector<int> threads {1, 2, 4};
    std::vector<std::string> modes {"latency", "batch", "steady"};
    long tasks   = 200000;
    int  work_us = 0;
    int  reps    = 5;
    int  warmup  = 1;
 };
 static Opts g_opts;
 static void busy_us(int us) {
    if (us <= 0) return;
    auto end = sclock::now() + std::chrono::microseconds(us);
    while (sclock::now() < end);
 }
 struct Sample {
    double ns_per_dispatch  = 0;   // end-to-end, minus the work payload
    double submit_ns        = 0;   // producer side only (batch mode)
    double vcsw_per_task    = 0;   // B2: sleeps per dispatch
    double ivcsw_per_task   = 0;
 };
 // ── latency: one task at a time, worker asleep between submissions ────────────
 static Sample run_latency(int threads, long tasks) {
    ThreadPool pool(threads);
    pool.start();
    std::mutex mx;
    std::condition_variable cv;
    bool done = false;
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < tasks; ++i) {
        { std::lock_guard lk(mx); done = false; }
        pool.submit([&] {
            busy_us(g_opts.work_us);
            { std::lock_guard lk(mx); done = true; }
            cv.notify_one();
        });
        std::unique_lock lk(mx);
        cv.wait(lk, [&] { return done; });
    }
    auto t1 = sclock::now();
    Sample s;
    long iv = 0, vc = 0;
    ru.finish(iv, vc);
    pool.stop();
    double elapsed_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
    s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0;
    // The requesting thread blocks once per task too, so it books a voluntary
    // switch of its own; halve to attribute per side rather than per process.
    s.vcsw_per_task  = static_cast<double>(vc) / tasks / 2.0;
    s.ivcsw_per_task = static_cast<double>(iv) / tasks;
    return s;
 }
 // ── batch: submit flat out, drain once. No wake in the steady state ───────────
 static Sample run_batch(int threads, long tasks) {
    ThreadPool pool(threads);
    pool.start();
    std::atomic<long> ran{0};
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < tasks; ++i)
        pool.submit([&] {
            busy_us(g_opts.work_us);
            ran.fetch_add(1, std::memory_order_relaxed);
        });
    auto t_submitted = sclock::now();
    pool.drain();
    auto t1 = sclock::now();
    Sample s;
    long iv = 0, vc = 0;
    ru.finish(iv, vc);
    pool.stop();
    if (ran.load() != tasks)
        std::fprintf(stderr, "WARNING: batch ran %ld of %ld tasks\n",
                     ran.load(), tasks);
    double elapsed_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
    s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0;
    s.submit_ns = std::chrono::duration<double, std::nano>(t_submitted - t0).count() / tasks;
    s.vcsw_per_task  = static_cast<double>(vc) / tasks;
    s.ivcsw_per_task = static_cast<double>(iv) / tasks;
    return s;
 }
 // ── steady: the task resubmits its successor, as fire_once does ──────────────
 static Sample run_steady(int threads, long tasks) {
    ThreadPool pool(threads);
    pool.start();
    std::mutex mx;
    std::condition_variable cv;
    std::atomic<long> count{0};
    bool  finished = false;
    // Recursive submission: hold the chain in a std::function so the task can
    // resubmit itself. Captured by reference; it outlives the drain below.
    //
    // The counter is atomic rather than mutex-guarded so that this loop
    // measures the pool's dispatch path and not a lock of the benchmark's own.
    std::function<void()> step = [&] {
        busy_us(g_opts.work_us);
        long n = count.fetch_add(1, std::memory_order_relaxed) + 1;
        if (n < tasks) {
            pool.submit(step);
        } else {
            { std::lock_guard lk(mx); finished = true; }
            cv.notify_one();
        }
    };
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    pool.submit(step);
    {
        std::unique_lock lk(mx);
        cv.wait(lk, [&] { return finished; });
    }
    auto t1 = sclock::now();
    Sample s;
    long iv = 0, vc = 0;
    ru.finish(iv, vc);
    pool.stop();
    double elapsed_ns = std::chrono::duration<double, std::nano>(t1 - t0).count();
    s.ns_per_dispatch = elapsed_ns / tasks - g_opts.work_us * 1000.0;
    s.vcsw_per_task  = static_cast<double>(vc) / tasks;
    s.ivcsw_per_task = static_cast<double>(iv) / tasks;
    return s;
 }
 // ── driver ────────────────────────────────────────────────────────────────────
 static void run_row(const std::string& mode, int threads, long tasks) {
    auto once = [&] {
        if (mode == "latency") return run_latency(threads, tasks);
        if (mode == "batch")   return run_batch(threads, tasks);
        return run_steady(threads, tasks);
    };
    for (int i = 0; i < g_opts.warmup; ++i) (void)once();
    std::vector<double> ns, sub, vcsw, ivcsw;
    for (int i = 0; i < g_opts.reps; ++i) {
        Sample s = once();
        ns.push_back(s.ns_per_dispatch);
        sub.push_back(s.submit_ns);
        vcsw.push_back(s.vcsw_per_task);
        ivcsw.push_back(s.ivcsw_per_task);
    }
    const double med = bench::percentile(ns, 0.5);
    const double q1  = bench::percentile(ns, 0.25);
    const double q3  = bench::percentile(ns, 0.75);
    const double iqr = med > 0 ? 100.0 * (q3 - q1) / med : 0.0;
    const double sleeps = bench::percentile(vcsw, 0.5);
    std::fprintf(stderr, "%-9s %-8d %-8d %-10ld %-12.0f %-7.1f %-11.0f %-10.2f %-10.2f\n",
                 mode.c_str(), threads, g_opts.work_us, tasks, med, iqr,
                 bench::percentile(sub, 0.5), sleeps,
                 bench::percentile(ivcsw, 0.5));
    // Column names deliberately match bench_pipeline's key columns so that
    // scripts/bench_repro_check.py can gate this benchmark too.
    std::printf("%s,%d,%d,%d,%ld,%d,%.1f,%.2f,%.1f,%.3f,%.3f\n",
                mode.c_str(), threads, g_opts.work_us, threads, tasks,
                g_opts.reps, med, iqr, bench::percentile(sub, 0.5),
                sleeps, bench::percentile(ivcsw, 0.5));
    std::fflush(stdout);
 }
 static std::vector<int> parse_int_list(const char* s) {
    std::vector<int> out;
    const char* p = s;
    while (*p) {
        char* end = nullptr;
        long v = std::strtol(p, &end, 10);
        if (end == p) break;
        out.push_back(static_cast<int>(v));
        p = end;
        while (*p == ',' || *p == ' ') ++p;
    }
    return out;
 }
 static std::vector<std::string> parse_word_list(const std::string& s) {
    std::vector<std::string> out;
    std::size_t pos = 0;
    while (pos <= s.size()) {
        std::size_t c = s.find(',', pos);
        if (c == std::string::npos) c = s.size();
        if (c > pos) out.push_back(s.substr(pos, c - pos));
        pos = c + 1;
    }
    return out;
 }
 static void usage() {
    std::fprintf(stderr,
        "usage: bench_dispatch [options]\n"
        "  --threads=1,2,4                pool sizes\n"
        "  --mode=latency,batch,steady    which measurements to run\n"
        "  --tasks=200000                 dispatches per repetition\n"
        "  --work-us=0                    payload per task\n"
        "  --reps=5 --warmup=1\n");
 }
 int main(int argc, char** argv) {
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);
        if      (key == "--help" || key == "-h") { usage(); return 0; }
        else if (key == "--threads") g_opts.threads = parse_int_list(val.c_str());
        else if (key == "--mode")    g_opts.modes   = parse_word_list(val);
        else if (key == "--tasks")   g_opts.tasks   = std::atol(val.c_str());
        else if (key == "--work-us") g_opts.work_us = std::atoi(val.c_str());
        else if (key == "--reps")    g_opts.reps    = std::atoi(val.c_str());
        else if (key == "--warmup")  g_opts.warmup  = std::atoi(val.c_str());
        else { std::fprintf(stderr, "unknown option: %s\n", a.c_str()); usage(); return 2; }
    }
    if (g_opts.reps < 1) g_opts.reps = 1;
    if (g_opts.warmup < 0) g_opts.warmup = 0;
    char cfg[160];
    std::snprintf(cfg, sizeof cfg, "tasks=%ld work_us=%d reps=%d warmup=%d",
                  g_opts.tasks, g_opts.work_us, g_opts.reps, g_opts.warmup);
    bench::print_environment(cfg);
    std::fprintf(stderr, "\n%-9s %-8s %-8s %-10s %-12s %-7s %-11s %-10s %-10s\n",
                 "mode", "threads", "work_us", "tasks", "ns/dispatch", "iqr%",
                 "submit_ns", "vcsw/task", "ivcsw/task");
    std::fprintf(stderr, "%s\n", std::string(96, '-').c_str());
    std::printf("topology,size,work_us,threads,items,reps,ns_per_dispatch,"
                "iqr_pct,submit_ns,vcsw_per_task,ivcsw_per_task\n");
    // latency is a round trip per task, so it is far slower per dispatch than
    // the other modes; scale it down rather than run for minutes.
    for (const auto& mode : g_opts.modes)
        for (int t : g_opts.threads) {
            long tasks = mode == "latency"
                       ? std::max(2000L, g_opts.tasks / 20)
                       : g_opts.tasks;
            run_row(mode, t, tasks);
        }
 }
@@ -1,108 +0,0 @@
 // Shared benchmark plumbing: machine attribution (PERF_PLAN M6), repetition
 // statistics (M3), and context-switch capture.
 //
 // The attribution is not decoration. A result taken under the powersave
 // governor or on battery is not comparable with one taken on AC under
 // performance, and a stored CSV that does not say which it was cannot be
 // argued about later.
 #pragma once
 #include <algorithm>
 #include <cmath>
 #include <cstdio>
 #include <string>
 #include <thread>
 #include <vector>
 #include <sys/resource.h>
 namespace bench {
 inline int hw_units() {
    unsigned n = std::thread::hardware_concurrency();
    return n ? static_cast<int>(n) : 1;
 }
 inline std::string read_line_of(const char* path) {
    std::FILE* f = std::fopen(path, "r");
    if (!f) return "unknown";
    char buf[128] = {0};
    if (!std::fgets(buf, sizeof buf, f)) { std::fclose(f); return "unknown"; }
    std::fclose(f);
    std::string s(buf);
    while (!s.empty() && (s.back() == '\n' || s.back() == ' ')) s.pop_back();
    return s.empty() ? "unknown" : s;
 }
 inline std::string ac_state() {
    for (const char* p : {"/sys/class/power_supply/AC/online",
                          "/sys/class/power_supply/AC0/online",
                          "/sys/class/power_supply/ACAD/online",
                          "/sys/class/power_supply/ADP1/online"}) {
        std::string v = read_line_of(p);
        if (v != "unknown") return v == "1" ? "ac" : "battery";
    }
    return "unknown";
 }
 // M6 — emitted to both streams: the CSV so a stored result can be attributed,
 // the terminal so a run under the wrong governor is noticed while it happens.
 inline void print_environment(const std::string& config_line) {
    const std::string gov = read_line_of(
        "/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor");
    const std::string ac = ac_state();
    for (std::FILE* out : {stdout, stderr}) {
        std::fprintf(out, "# nproc=%d governor=%s power=%s\n",
                     hw_units(), gov.c_str(), ac.c_str());
        if (!config_line.empty())
            std::fprintf(out, "# %s\n", config_line.c_str());
 #if defined(__GNUC__) && !defined(__clang__)
        std::fprintf(out, "# compiler=gcc-%d.%d.%d\n",
                     __GNUC__, __GNUC_MINOR__, __GNUC_PATCHLEVEL__);
 #elif defined(__clang__)
        std::fprintf(out, "# compiler=clang-%d.%d.%d\n",
                     __clang_major__, __clang_minor__, __clang_patchlevel__);
 #endif
    }
    if (gov != "performance" || ac == "battery")
        std::fprintf(stderr,
            "# WARNING: governor=%s power=%s — results are not comparable with\n"
            "#          a run on AC power under the performance governor.\n",
            gov.c_str(), ac.c_str());
 }
 inline double percentile(std::vector<double> v, double p) {
    if (v.empty()) return 0;
    std::sort(v.begin(), v.end());
    double idx = p * (v.size() - 1);
    auto   lo  = static_cast<std::size_t>(std::floor(idx));
    auto   hi  = static_cast<std::size_t>(std::ceil(idx));
    return v[lo] + (v[hi] - v[lo]) * (idx - lo);
 }
 // Process-wide context-switch counters, sampled around a timed region.
 //
 // ru_nvcsw (voluntary) is the cheap answer to PERF_PLAN B2: a thread that
 // blocks on a condition variable books a voluntary switch, so voluntary
 // switches per dispatch is, near enough, sleeps per dispatch. ru_nivcsw
 // (involuntary) is preemption, which is what oversubscription looks like (A3).
 struct RusageDelta {
    long ivcsw0 = 0, vcsw0 = 0;
    void start() {
        rusage ru{};
        getrusage(RUSAGE_SELF, &ru);
        ivcsw0 = ru.ru_nivcsw;
        vcsw0  = ru.ru_nvcsw;
    }
    void finish(long& nivcsw, long& nvcsw) const {
        rusage ru{};
        getrusage(RUSAGE_SELF, &ru);
        nivcsw = ru.ru_nivcsw - ivcsw0;
        nvcsw  = ru.ru_nvcsw  - vcsw0;
    }
 };
 } // namespace bench
@@ -9,38 +9,24 @@
 //   private — each node owns a private ThreadPool(1)  [Node<>]
 //   pool    — all nodes share one ThreadPool(T)        [PoolNode<> + shared pool]
 //
-// Each row is run --reps times (plus discarded warm-up runs); the reported
+// Usage: ./bench_pipeline | tee results.csv
 // figure is the median items/sec, with the inter-quartile spread as a
 // reliability indicator. A row whose iqr_pct is above a few percent is not
 // measuring what it claims to measure.
 //
 // Usage: ./bench_pipeline [options] | tee results.csv
 //        ./bench_pipeline --help
 #include <kpn/kpn.hpp>
 #include "bench_env.hpp"
 #ifdef KPN_BENCH_TBB
 #include <oneapi/tbb/flow_graph.h>
 namespace tbb_flow = oneapi::tbb::flow;
 #endif
 #include <algorithm>
 #include <array>
 #include <atomic>
 #include <chrono>
 #include <cmath>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <memory>
 #include <string>
 #include <thread>
 #include <vector>
 #include <sys/resource.h>
 using namespace kpn;
 using namespace std::chrono_literals;
 using sclock = std::chrono::steady_clock;
@@ -71,63 +57,31 @@ static void push_retry(Channel<int>& ch, int val) {
    }
 }
-// ── configuration (M1, M3, M4, M5) ────────────────────────────────────────────
+// ── result ────────────────────────────────────────────────────────────────────
-struct Config {
+struct Result {
-    std::vector<int> work_amts {10, 100, 1000};
+    const char* topology;
-    std::vector<int> pool_sizes{1, 2, 4, 8, 16, 20};   // M5
+    int         size;
-    std::vector<int> depths    {1, 2, 4, 8, 16, 32};
+    int         work_us;
-    std::vector<int> widths    {1, 2, 3, 4};
+    int         threads;     // 0 = private (1 thread per node), N = shared pool size
-    int    reps        = 5;      // M3: measured repetitions per row
+    double      items_per_sec;
-    int    warmup      = 1;      // M4: discarded repetitions per row
+    double      overhead_us;
    double target_sec  = 0.30;   // aimed-for duration of one repetition
    long   min_items   = 2000;   // M1: floor, independent of work_us and depth
    double max_sec     = 3.0;    // ceiling; only bites where min_items cannot fit
    bool   do_chain = true, do_wide = true, do_diamond = true;
    bool   do_priv  = true, do_pool = true, do_tbb = true;
 };
 static Config g_cfg;
 // M1 — sample size from a time budget with a hard floor, rather than a
 // hand-tuned ladder that collapsed to 50–200 items on exactly the rows under
 // investigation.
 //
 // `stages` is the number of node firings per item; `units` the number of
 // threads able to run them concurrently. Steady-state throughput of the
 // pipeline is bounded by work_us * stages / units, so that is the per-item
 // cost the sample size is derived from. Depth beyond `units` costs throughput;
 // depth below it costs only latency, which does not scale the run.
 static long pick_items(int work_us, int stages, int units) {
    units = std::max(1, std::min(units, bench::hw_units()));
    const double per_item_us =
        std::max(1.0, static_cast<double>(work_us)) *
        std::max(1.0, static_cast<double>(stages) / units);
    long want = static_cast<long>(g_cfg.target_sec * 1e6 / per_item_us);
    long cap  = static_cast<long>(g_cfg.max_sec    * 1e6 / per_item_us);
    want = std::max(want, g_cfg.min_items);
    // The floor wins unless honouring it would blow the time ceiling by more
    // than the ceiling allows; such rows are reported with their true N so the
    // reader can see they are short.
    if (want > cap) want = std::max(cap, 200L);
    return want;
 }
 // ── one measured repetition ───────────────────────────────────────────────────
 struct Sample {
    double items_per_sec = 0;
    double overhead_us   = 0;
    long   nivcsw        = 0;   // involuntary context switches during the run
    long   nvcsw         = 0;   // voluntary   context switches during the run
 };
 // ── chain ─────────────────────────────────────────────────────────────────────
-static Sample bench_chain(int depth, int work_us, long N) {
+static int items_for(int work_us, int depth = 1) {
-    const std::size_t CAP = static_cast<std::size_t>(N);
+    int effective = std::max(1, work_us) * std::max(1, depth);
    if (effective <= 1)     return 5000;
    if (effective <= 10)    return 3000;
    if (effective <= 100)   return 1000;
    if (effective <= 1000)  return 200;
    return 50;
 }
 static Result bench_chain(int depth, int work_us) {
    const int N   = items_for(work_us, depth);
    const int CAP = N;
    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= depth; ++i)
@@ -144,20 +98,17 @@ static Sample bench_chain(int depth, int work_us, long N) {
    std::atomic<sclock::time_point> t1;
    std::thread reader([&] {
-        for (long i = 0; i < N; ++i) chs.back()->pop();
+        for (int i = 0; i < N; ++i) chs.back()->pop();
        t1.store(sclock::now(), std::memory_order_release);
    });
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (long i = 0; i < N; ++i) push_retry(*chs[0], static_cast<int>(i));
+        for (int i = 0; i < N; ++i) push_retry(*chs[0], i);
    });
    pusher.join();
    reader.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    for (auto& n : nodes) n->stop();
    double elapsed = std::chrono::duration<double>(
@@ -165,13 +116,13 @@ static Sample bench_chain(int depth, int work_us, long N) {
    // Subtract theoretical pipeline fill cost (depth-1)*W so that overhead
    // reflects only framework latency, not the expected pipeline startup time.
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
-    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
+    double wus = (elapsed * 1e6 - pipeline_us) / N;
-    s.items_per_sec = N / elapsed;
+    return {"chain", depth, work_us, 0, N / elapsed, wus};
    return s;
 }
-static Sample bench_chain_pool(int depth, int work_us, int pool_threads, long N) {
+static Result bench_chain_pool(int depth, int work_us, int pool_threads) {
-    const std::size_t CAP = static_cast<std::size_t>(N);
+    const int N   = items_for(work_us, depth);
    const int CAP = N;
    auto pool = std::make_shared<ThreadPool>(pool_threads);
@@ -191,36 +142,33 @@ static Sample bench_chain_pool(int depth, int work_us, int pool_threads, long N)
    std::atomic<sclock::time_point> t1;
    std::thread reader([&] {
-        for (long i = 0; i < N; ++i) chs.back()->pop();
+        for (int i = 0; i < N; ++i) chs.back()->pop();
        t1.store(sclock::now(), std::memory_order_release);
    });
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (long i = 0; i < N; ++i) push_retry(*chs[0], static_cast<int>(i));
+        for (int i = 0; i < N; ++i) push_retry(*chs[0], i);
    });
    pusher.join();
    reader.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    for (auto& n : nodes) n->stop();
    pool->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
-    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
+    double wus = (elapsed * 1e6 - pipeline_us) / N;
-    s.items_per_sec = N / elapsed;
+    return {"chain", depth, work_us, pool_threads, N / elapsed, wus};
    return s;
 }
 // ── wide (fanout<W>) ──────────────────────────────────────────────────────────
 template<std::size_t W>
-static Sample bench_wide(int work_us, long N) {
+static Result bench_wide(int work_us) {
-    const std::size_t CAP = static_cast<std::size_t>(N);
+    const int N   = items_for(work_us);
    const int CAP = N;
    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, W>>(CAP);
@@ -249,36 +197,33 @@ static Sample bench_wide(int work_us, long N) {
    for (std::size_t w = 0; w < W; ++w) {
        readers[w] = std::thread([&, w] {
-            for (long i = 0; i < N; ++i) sink_chs[w]->pop();
+            for (int i = 0; i < N; ++i) sink_chs[w]->pop();
            if (readers_done.fetch_add(1, std::memory_order_acq_rel) + 1
                    == static_cast<int>(W))
                t1.store(sclock::now(), std::memory_order_release);
        });
    }
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
+        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
    });
    pusher.join();
    for (auto& r : readers) r.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    for (auto& n : nodes) n->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    s.items_per_sec = N / elapsed;
+    return {"wide", static_cast<int>(W), work_us, 0, N / elapsed, wus};
    return s;
 }
 template<std::size_t W>
-static Sample bench_wide_pool(int work_us, int pool_threads, long N) {
+static Result bench_wide_pool(int work_us, int pool_threads) {
-    const std::size_t CAP = static_cast<std::size_t>(N);
+    const int N   = items_for(work_us);
    const int CAP = N;
    auto pool   = std::make_shared<ThreadPool>(pool_threads);
    auto src_ch = std::make_shared<Channel<int>>(CAP);
@@ -309,38 +254,35 @@ static Sample bench_wide_pool(int work_us, int pool_threads, long N) {
    for (std::size_t w = 0; w < W; ++w) {
        readers[w] = std::thread([&, w] {
-            for (long i = 0; i < N; ++i) sink_chs[w]->pop();
+            for (int i = 0; i < N; ++i) sink_chs[w]->pop();
            if (readers_done.fetch_add(1, std::memory_order_acq_rel) + 1
                    == static_cast<int>(W))
                t1.store(sclock::now(), std::memory_order_release);
        });
    }
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
+        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
    });
    pusher.join();
    for (auto& r : readers) r.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    for (auto& n : nodes) n->stop();
    pool->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    s.items_per_sec = N / elapsed;
+    return {"wide", static_cast<int>(W), work_us, pool_threads, N / elapsed, wus};
    return s;
 }
 // ── diamond ───────────────────────────────────────────────────────────────────
-static Sample bench_diamond(int work_us, long N) {
+static Result bench_diamond(int work_us) {
-    const std::size_t CAP = static_cast<std::size_t>(N);
+    const int N   = items_for(work_us, 2);
    const int CAP = N;
    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, 2>>(CAP);
@@ -370,7 +312,7 @@ static Sample bench_diamond(int work_us, long N) {
    std::atomic<int> done{0};
    auto make_reader = [&](Channel<int>& ch) {
        return std::thread([&] {
-            for (long i = 0; i < N; ++i) ch.pop();
+            for (int i = 0; i < N; ++i) ch.pop();
            if (done.fetch_add(1, std::memory_order_acq_rel) + 1 == 2)
                t1.store(sclock::now(), std::memory_order_release);
        });
@@ -378,26 +320,23 @@ static Sample bench_diamond(int work_us, long N) {
    auto rL = make_reader(*snkL);
    auto rR = make_reader(*snkR);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
+        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
    });
    pusher.join(); rL.join(); rR.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop(); nL->stop(); nR->stop(); nL2->stop(); nR2->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    s.items_per_sec = N / elapsed;
+    return {"diamond", 4, work_us, 0, N / elapsed, wus};
    return s;
 }
-static Sample bench_diamond_pool(int work_us, int pool_threads, long N) {
+static Result bench_diamond_pool(int work_us, int pool_threads) {
-    const std::size_t CAP = static_cast<std::size_t>(N);
+    const int N   = items_for(work_us, 2);
    const int CAP = N;
    auto pool   = std::make_shared<ThreadPool>(pool_threads);
    auto src_ch = std::make_shared<Channel<int>>(CAP);
@@ -430,7 +369,7 @@ static Sample bench_diamond_pool(int work_us, int pool_threads, long N) {
    std::atomic<int> done{0};
    auto make_reader = [&](Channel<int>& ch) {
        return std::thread([&] {
-            for (long i = 0; i < N; ++i) ch.pop();
+            for (int i = 0; i < N; ++i) ch.pop();
            if (done.fetch_add(1, std::memory_order_acq_rel) + 1 == 2)
                t1.store(sclock::now(), std::memory_order_release);
        });
@@ -438,30 +377,28 @@ static Sample bench_diamond_pool(int work_us, int pool_threads, long N) {
    auto rL = make_reader(*snkL);
    auto rR = make_reader(*snkR);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
-        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
+        for (int i = 0; i < N; ++i) push_retry(*src_ch, i);
    });
    pusher.join(); rL.join(); rR.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    nL->stop(); nR->stop(); nL2->stop(); nR2->stop();
    pool->stop();
    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
-    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    double wus = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    s.items_per_sec = N / elapsed;
+    return {"diamond", 4, work_us, pool_threads, N / elapsed, wus};
    return s;
 }
 // ── TBB flow graph ────────────────────────────────────────────────────────────
 #ifdef KPN_BENCH_TBB
-static Sample bench_chain_tbb(int depth, int work_us, long N) {
+static Result bench_chain_tbb(int depth, int work_us) {
    const int N = items_for(work_us, depth);
    tbb_flow::graph g;
    using FN = tbb_flow::function_node<int, int>;
    std::vector<std::unique_ptr<FN>> nodes;
@@ -472,23 +409,21 @@ static Sample bench_chain_tbb(int depth, int work_us, long N) {
    for (int i = 0; i + 1 < depth; ++i)
        tbb_flow::make_edge(*nodes[i], *nodes[i + 1]);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
-    for (long i = 0; i < N; ++i) nodes[0]->try_put(static_cast<int>(i));
+    for (int i = 0; i < N; ++i) nodes[0]->try_put(i);
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    double elapsed    = std::chrono::duration<double>(t1 - t0).count();
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
-    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
+    double wus        = (elapsed * 1e6 - pipeline_us) / N;
-    s.items_per_sec = N / elapsed;
+    return {"chain_tbb", depth, work_us, -1, N / elapsed, wus};
    return s;
 }
 template<std::size_t W>
-static Sample bench_wide_tbb(int work_us, long N) {
+static Result bench_wide_tbb(int work_us) {
    const int N = items_for(work_us);
    tbb_flow::graph g;
    tbb_flow::broadcast_node<int> fan(g);
    using FN = tbb_flow::function_node<int, int>;
@@ -499,21 +434,19 @@ static Sample bench_wide_tbb(int work_us, long N) {
        tbb_flow::make_edge(fan, *n);
    }
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
-    for (long i = 0; i < N; ++i) fan.try_put(static_cast<int>(i));
+    for (int i = 0; i < N; ++i) fan.try_put(i);
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    double elapsed = std::chrono::duration<double>(t1 - t0).count();
-    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    double wus     = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    s.items_per_sec = N / elapsed;
+    return {"wide_tbb", static_cast<int>(W), work_us, -1, N / elapsed, wus};
    return s;
 }
-static Sample bench_diamond_tbb(int work_us, long N) {
+static Result bench_diamond_tbb(int work_us) {
    const int N = items_for(work_us, 2);
    tbb_flow::graph g;
    tbb_flow::broadcast_node<int> fan(g);
    using FN = tbb_flow::function_node<int, int>;
@@ -523,242 +456,71 @@ static Sample bench_diamond_tbb(int work_us, long N) {
    tbb_flow::make_edge(fan, nL);  tbb_flow::make_edge(fan, nR);
    tbb_flow::make_edge(nL, nL2);  tbb_flow::make_edge(nR, nR2);
    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
-    for (long i = 0; i < N; ++i) fan.try_put(static_cast<int>(i));
+    for (int i = 0; i < N; ++i) fan.try_put(i);
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    double elapsed = std::chrono::duration<double>(t1 - t0).count();
-    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
+    double wus     = (elapsed * 1e6) / N - static_cast<double>(work_us);
-    s.items_per_sec = N / elapsed;
+    return {"diamond_tbb", 4, work_us, -1, N / elapsed, wus};
    return s;
 }
 #endif // KPN_BENCH_TBB
 // ── repetition driver (M2, M3, M4) ────────────────────────────────────────────
 using bench::percentile;
 // A row: median of `reps` repetitions, after `warmup` discarded ones.
 // M2 — items/sec is the primary figure; derived overhead is secondary,
 // because it is a difference of large numbers and magnifies noise ~10×.
 template<class Fn>
 static void run_row(const char* topology, int size, int work_us, int sched,
                    long N, Fn&& one_rep) {
    for (int i = 0; i < g_cfg.warmup; ++i) (void)one_rep();   // M4
    std::vector<double> ips, ovh;
    long ivcsw = 0, vcsw = 0;
    for (int i = 0; i < g_cfg.reps; ++i) {
        Sample s = one_rep();
        ips.push_back(s.items_per_sec);
        ovh.push_back(s.overhead_us);
        ivcsw += s.nivcsw;
        vcsw  += s.nvcsw;
    }
    const double med  = percentile(ips, 0.5);
    const double q1   = percentile(ips, 0.25);
    const double q3   = percentile(ips, 0.75);
    const double iqr  = med > 0 ? 100.0 * (q3 - q1) / med : 0.0;
    const double lo   = *std::min_element(ips.begin(), ips.end());
    const double hi   = *std::max_element(ips.begin(), ips.end());
    const double spread = med > 0 ? 100.0 * (hi - lo) / med : 0.0;
    const double ivcsw_per_item = static_cast<double>(ivcsw) / (double(N) * g_cfg.reps);
    const double vcsw_per_item  = static_cast<double>(vcsw)  / (double(N) * g_cfg.reps);
    const std::string s = sched  < 0 ? "tbb"
                        : sched == 0 ? "priv"
                                     : std::to_string(sched);
    std::fprintf(stderr, "%-10s %-5d %-8d %-6s %-8ld %-12.0f %-7.1f %-7.1f %-9.1f %-8.2f %-8.2f\n",
                 topology, size, work_us, s.c_str(), N,
                 med, iqr, spread, percentile(ovh, 0.5), ivcsw_per_item, vcsw_per_item);
    std::printf("%s,%d,%d,%s,%ld,%d,%.0f,%.0f,%.0f,%.2f,%.2f,%.2f,%.3f,%.3f\n",
                topology, size, work_us, s.c_str(), N, g_cfg.reps,
                med, lo, hi, iqr, spread, percentile(ovh, 0.5),
                ivcsw_per_item, vcsw_per_item);
    std::fflush(stdout);
 }
 // ── argument parsing ──────────────────────────────────────────────────────────
 static std::vector<int> parse_int_list(const char* s) {
    std::vector<int> out;
    const char* p = s;
    while (*p) {
        char* end = nullptr;
        long v = std::strtol(p, &end, 10);
        if (end == p) break;
        out.push_back(static_cast<int>(v));
        p = end;
        while (*p == ',' || *p == ' ') ++p;
    }
    return out;
 }
 static bool has_word(const std::string& csv, const char* word) {
    return csv.find(word) != std::string::npos;
 }
 static void usage() {
    std::fprintf(stderr,
        "usage: bench_pipeline [options]\n"
        "  --work=10,100,1000     per-node busy-work, microseconds\n"
        "  --depths=1,2,4,8,16,32 chain depths\n"
        "  --widths=1,2,3,4       fanout widths\n"
        "  --pools=1,2,4,8,16,20  shared-pool thread counts\n"
        "  --topos=chain,wide,diamond\n"
        "  --modes=priv,pool,tbb\n"
        "  --reps=5               measured repetitions per row\n"
        "  --warmup=1             discarded repetitions per row\n"
        "  --target-sec=0.30      aimed-for duration of one repetition\n"
        "  --min-items=2000       sample-size floor\n"
        "  --max-sec=3.0          per-repetition ceiling (overrides the floor)\n");
 }
 static bool parse_args(int argc, char** argv) {
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);
        if (key == "--help" || key == "-h")   { usage(); std::exit(0); }
        else if (key == "--work")       g_cfg.work_amts  = parse_int_list(val.c_str());
        else if (key == "--depths")     g_cfg.depths     = parse_int_list(val.c_str());
        else if (key == "--widths")     g_cfg.widths     = parse_int_list(val.c_str());
        else if (key == "--pools")      g_cfg.pool_sizes = parse_int_list(val.c_str());
        else if (key == "--reps")       g_cfg.reps       = std::atoi(val.c_str());
        else if (key == "--warmup")     g_cfg.warmup     = std::atoi(val.c_str());
        else if (key == "--target-sec") g_cfg.target_sec = std::atof(val.c_str());
        else if (key == "--min-items")  g_cfg.min_items  = std::atol(val.c_str());
        else if (key == "--max-sec")    g_cfg.max_sec    = std::atof(val.c_str());
        else if (key == "--topos") {
            g_cfg.do_chain   = has_word(val, "chain");
            g_cfg.do_wide    = has_word(val, "wide");
            g_cfg.do_diamond = has_word(val, "diamond");
        }
        else if (key == "--modes") {
            g_cfg.do_priv = has_word(val, "priv");
            g_cfg.do_pool = has_word(val, "pool");
            g_cfg.do_tbb  = has_word(val, "tbb");
        }
        else { std::fprintf(stderr, "unknown option: %s\n", a.c_str()); usage(); return false; }
    }
    if (g_cfg.reps < 1) g_cfg.reps = 1;
    if (g_cfg.warmup < 0) g_cfg.warmup = 0;
    return true;
 }
 // `wide` is templated on W, so dispatch the runtime width through a switch.
 template<class F>
 static void with_width(int w, F&& f) {
    switch (w) {
        case 1: f(std::integral_constant<std::size_t, 1>{}); break;
        case 2: f(std::integral_constant<std::size_t, 2>{}); break;
        case 3: f(std::integral_constant<std::size_t, 3>{}); break;
        case 4: f(std::integral_constant<std::size_t, 4>{}); break;
        default:
            std::fprintf(stderr, "width %d not instantiated (1..4 only)\n", w);
    }
 }
 // ── main ──────────────────────────────────────────────────────────────────────
-int main(int argc, char** argv) {
+int main() {
-    // A rejected option must fail loudly: a harness driver that silently got
+    const int work_amts[]  = {10, 100, 1000};
-    // no CSV back is worse than one that stops.
+    const int pool_sizes[] = {1, 2, 4};
    if (!parse_args(argc, argv)) return 2;
-    char cfg[192];
+    std::fprintf(stderr, "%-12s %-8s %-10s %-8s %-18s %-20s\n",
-    std::snprintf(cfg, sizeof cfg,
+                 "topology", "size", "work_us", "threads", "items/sec", "overhead_us/item");
-                  "reps=%d warmup=%d target_sec=%.2f min_items=%ld max_sec=%.1f",
+    std::fprintf(stderr, "%s\n", std::string(78, '-').c_str());
-                  g_cfg.reps, g_cfg.warmup, g_cfg.target_sec,
+    std::printf("topology,size,work_us,threads,items_per_sec,overhead_us_per_item\n");
                  g_cfg.min_items, g_cfg.max_sec);
    bench::print_environment(cfg);
-    std::fprintf(stderr, "\n%-10s %-5s %-8s %-6s %-8s %-12s %-7s %-7s %-9s %-8s %-8s\n",
+    auto emit = [](const Result& r) {
-                 "topology", "size", "work_us", "sched", "items", "items/sec",
+        std::string sched = r.threads  < 0 ? "tbb"
-                 "iqr%", "range%", "ovh_us", "ivcsw/it", "vcsw/it");
+                          : r.threads == 0 ? "priv"
-    std::fprintf(stderr, "%s\n", std::string(104, '-').c_str());
+                          : std::to_string(r.threads);
-    std::printf("topology,size,work_us,threads,items,reps,items_per_sec,"
+        std::fprintf(stderr, "%-12s %-8d %-10d %-8s %-18.0f %-20.1f\n",
-                "items_per_sec_min,items_per_sec_max,iqr_pct,range_pct,"
+                     r.topology, r.size, r.work_us, sched.c_str(),
-                "overhead_us_per_item,ivcsw_per_item,vcsw_per_item\n");
+                     r.items_per_sec, r.overhead_us);
        std::printf("%s,%d,%d,%s,%.0f,%.2f\n",
                    r.topology, r.size, r.work_us, sched.c_str(),
                    r.items_per_sec, r.overhead_us);
        std::fflush(stdout);
    };
-    for (int w : g_cfg.work_amts) {
+    for (int w : work_amts) {
        g_work_us.store(w, std::memory_order_relaxed);
        std::fprintf(stderr, "\n── work_us=%-4d private pools ───────────────────────────────────────\n", w);
-        if (g_cfg.do_priv) {
+        for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain(d, w));
-            std::fprintf(stderr, "\n── work_us=%-4d private pools ──────────────────────\n", w);
+        emit(bench_wide<1>(w));
-            if (g_cfg.do_chain)
+        emit(bench_wide<2>(w));
-                for (int d : g_cfg.depths) {
+        emit(bench_wide<3>(w));
-                    long N = pick_items(w, d, d);
+        emit(bench_wide<4>(w));
-                    run_row("chain", d, w, 0, N, [&] { return bench_chain(d, w, N); });
+        emit(bench_diamond(w));
                }
            if (g_cfg.do_wide)
                for (int wd : g_cfg.widths)
                    with_width(wd, [&](auto W) {
                        long N = pick_items(w, W.value, W.value);
                        run_row("wide", static_cast<int>(W.value), w, 0, N,
                                [&] { return bench_wide<W.value>(w, N); });
                    });
            if (g_cfg.do_diamond) {
                long N = pick_items(w, 4, 4);
                run_row("diamond", 4, w, 0, N, [&] { return bench_diamond(w, N); });
            }
        }
-        if (g_cfg.do_pool) {
+        for (int pt : pool_sizes) {
-            for (int pt : g_cfg.pool_sizes) {
+            std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ─────────────────────────────\n",
                std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ───────────\n",
                         w, pt, pt == 1 ? "" : "s");
-                if (g_cfg.do_chain)
+            for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain_pool(d, w, pt));
-                    for (int d : g_cfg.depths) {
+            emit(bench_wide_pool<1>(w, pt));
-                        long N = pick_items(w, d, pt);
+            emit(bench_wide_pool<2>(w, pt));
-                        run_row("chain", d, w, pt, N,
+            emit(bench_wide_pool<3>(w, pt));
-                                [&] { return bench_chain_pool(d, w, pt, N); });
+            emit(bench_wide_pool<4>(w, pt));
-                    }
+            emit(bench_diamond_pool(w, pt));
                if (g_cfg.do_wide)
                    for (int wd : g_cfg.widths)
                        with_width(wd, [&](auto W) {
                            long N = pick_items(w, W.value, pt);
                            run_row("wide", static_cast<int>(W.value), w, pt, N,
                                    [&] { return bench_wide_pool<W.value>(w, pt, N); });
                        });
                if (g_cfg.do_diamond) {
                    long N = pick_items(w, 4, pt);
                    run_row("diamond", 4, w, pt, N,
                            [&] { return bench_diamond_pool(w, pt, N); });
                }
            }
        }
 #ifdef KPN_BENCH_TBB
-        if (g_cfg.do_tbb) {
+        std::fprintf(stderr, "\n── work_us=%-4d TBB flow graph ──────────────────────────────────────\n", w);
-            std::fprintf(stderr, "\n── work_us=%-4d TBB flow graph ─────────────────────\n", w);
+        for (int d : {1, 2, 4, 8, 16, 32}) emit(bench_chain_tbb(d, w));
-            if (g_cfg.do_chain)
+        emit(bench_wide_tbb<1>(w));
-                for (int d : g_cfg.depths) {
+        emit(bench_wide_tbb<2>(w));
-                    long N = pick_items(w, d, d);
+        emit(bench_wide_tbb<3>(w));
-                    run_row("chain_tbb", d, w, -1, N,
+        emit(bench_wide_tbb<4>(w));
-                            [&] { return bench_chain_tbb(d, w, N); });
+        emit(bench_diamond_tbb(w));
                }
            if (g_cfg.do_wide)
                for (int wd : g_cfg.widths)
                    with_width(wd, [&](auto W) {
                        long N = pick_items(w, W.value, W.value);
                        run_row("wide_tbb", static_cast<int>(W.value), w, -1, N,
                                [&] { return bench_wide_tbb<W.value>(w, N); });
                    });
            if (g_cfg.do_diamond) {
                long N = pick_items(w, 4, 4);
                run_row("diamond_tbb", 4, w, -1, N,
                        [&] { return bench_diamond_tbb(w, N); });
            }
        }
 #endif
    }
 }
@@ -7,78 +7,12 @@
 #include <array>
 #include <atomic>
 #include <chrono>
 #include <functional>
 #include <memory>
 #include <thread>
 namespace kpn {
 // ── Lossless single-output delivery ───────────────────────────────────────────
 //
 // Shared by RouterNode and FilterNode, which each deliver a value to exactly one
 // channel. Both previously did
 //
 //     try { ch->push(val); } catch (const ChannelOverflowError&) {}
 //
 // which discards the value whenever the consumer is behind. 6595e6e made node
 // outputs lossless, 28e0667 stopped them parking a worker, and a8cfe73 did the
 // same for FanoutNode — these two were in none of them, and were the last
 // remaining users of the throwing push() on a data path.
 //
 // A dropped item does not degrade a downstream result, it silently changes one.
 // Worse, a dropped *sentinel* wedges the pipeline outright: EOF is what tells
 // every downstream node to shut down, and there is nothing after it to retry.
 // A filter that passes EOF by predicate but drops it by backpressure is a
 // pipeline that never terminates.
 //
 // So sentinels go out-of-band via push_sentinel (a dedicated slot that consumes
 // no ring capacity and cannot overflow), and everything else is retried until
 // taken. Like FanoutNode and unlike a pool node, these own a private thread, so
 // waiting here costs no scheduler worker and needs no space-callback park.
 // stop_flag_ is rechecked every pass so teardown cannot hang on a full output.
 //
 // `parked` receives the time spent waiting, which the caller charges to blocked
 // rather than exec — a parked node is idle, and charging it to exec reports the
 // node as busy exactly when it is the one being held up.
 //
 // Returns false if stopped with the value undelivered.
 template<typename T>
 bool deliver_one(Channel<T>* ch, T& val, const std::atomic<bool>& stop_flag,
                 duration_t& parked) {
    if (is_sentinel_value(val)) {
        ch->push_sentinel(std::move(val));
        return true;
    }
    const auto park_from = clock_t::now();
    for (;;) {
        switch (ch->try_push(val)) {
            case Channel<T>::PushResult::Taken:
                parked = duration_t(clock_t::now() - park_from);
                return true;
            case Channel<T>::PushResult::Closed:
                // Nobody is listening any more; the channel has recorded the
                // drop. Retrying would spin until teardown noticed.
                parked = duration_t(clock_t::now() - park_from);
                return false;
            case Channel<T>::PushResult::Full:
                break;                       // fall through to the retry logic
        }
        if (stop_flag.load(std::memory_order_relaxed)) {
            // Teardown with work in hand and the output still full. One last
            // throwing push, purely so the channel's own stats record the
            // overflow — the point of the lossless path is that a loss is never
            // invisible, and a silent return here would reintroduce exactly the
            // hole this function exists to close.
            try { ch->push(std::move(val)); }
            catch (const ChannelOverflowError&) {}
            parked = duration_t(clock_t::now() - park_from);
            return false;
        }
        std::this_thread::sleep_for(std::chrono::microseconds(50));
    }
 }
 // ── RouterNode ────────────────────────────────────────────────────────────────
 //
 // Reads one item and pushes it to exactly one of N output channels, chosen by
@@ -192,20 +126,15 @@ private:
                auto        t1  = clock_t::now();
                auto        cpu0 = NodeStats::cpu_now();
                // An out-of-range selector still drops by design (documented on
                // the class): the item was routed nowhere, not lost to a full
                // channel. Only the latter is what deliver_one exists to stop.
                std::size_t idx = selector_(val);
-                duration_t  parked{0};
+                if (idx < N && out_channels_[idx]) {
-                bool        delivered = true;
+                    try { out_channels_[idx]->push(val); }
-                if (idx < N && out_channels_[idx])
+                    catch (const ChannelOverflowError&) {}
-                    delivered = deliver_one(out_channels_[idx], val, stop_flag_, parked);
+                }
                auto cpu1 = NodeStats::cpu_now();
                auto t2   = clock_t::now();
-                stats_.record_exec(duration_t(t2 - t1) - parked,
+                stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
                                   duration_t(t1 - t0) + parked, cpu0, cpu1);
                if (!delivered) break;
            } catch (const ChannelClosedError&) {
                break;
            }
@@ -332,19 +261,12 @@ private:
                auto t1  = clock_t::now();
                auto cpu0 = NodeStats::cpu_now();
                // A value the predicate rejects is dropped by design and is not
                // counted as a processed frame. One it accepts is now delivered
                // losslessly — including a sentinel, which a filter typically
                // passes unconditionally so downstream can shut down, and which
                // the old throwing push discarded whenever the output was full.
                if (pred_(val) && out_ch_) {
-                    duration_t parked{0};
+                    try { out_ch_->push(val); }
-                    const bool delivered = deliver_one(out_ch_, val, stop_flag_, parked);
+                    catch (const ChannelOverflowError&) {}
                    auto cpu1 = NodeStats::cpu_now();
                    auto t2   = clock_t::now();
-                    stats_.record_exec(duration_t(t2 - t1) - parked,
+                    stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
                                       duration_t(t1 - t0) + parked, cpu0, cpu1);
                    if (!delivered) break;
                }
            } catch (const ChannelClosedError&) {
                break;
@@ -58,18 +58,6 @@ public:
    ChannelClosedError() : std::runtime_error("channel closed") {}
 };
 // Nothing available *right now* on a channel that is still open. Distinct from
 // ChannelClosedError, which means upstream is finished and never coming back.
 //
 // Conflating the two is expensive in one direction only: a consumer that reads
 // "empty" as "closed" stops a live node permanently, and because a stopping
 // node disables its own inputs and outputs, one benign empty read takes the
 // rest of the pipeline with it. The reverse costs nothing.
 class ChannelEmptyError : public std::runtime_error {
 public:
    ChannelEmptyError() : std::runtime_error("channel empty") {}
 };
 // ── CPU pause hint ────────────────────────────────────────────────────────────
 // Signals the CPU that this is a spin-wait loop, improving HT sibling throughput
 // and preventing branch-predictor thrash on x86. Falls back to a compiler barrier.
@@ -136,6 +124,7 @@ public:
            throw ChannelOverflowError(capacity_);
        }
        const bool was_empty = (t == h);
        buf_[t & ring_mask_] = make_storage(std::move(value));
        tail_.store(t + 1, std::memory_order_release);
        stats_.record_push(t - h + 1, data_bytes);
@@ -143,60 +132,10 @@ public:
        wake_.fetch_add(1, std::memory_order_release);
        wake_.notify_one();
-        // Level-triggered, not edge-triggered — see set_push_callback.
+        if (was_empty && push_callback_)
        if (push_callback_)
            push_callback_();
    }
    /// Called when a pop frees a slot in a previously-full ring.
    ///
    /// The mirror of `set_push_callback`, and it exists for the same reason:
    /// a producer must be able to *park* rather than spin. Without it the only
    /// lossless option is `push_blocking`, which sleeps inside the caller's
    /// thread — and when that thread is a scheduler worker, parking it starves
    /// every node pinned to it (see the hold-and-wait note on push_sentinel).
    void set_space_callback(std::function<void()> cb) { space_callback_ = std::move(cb); }
    /// True when a push would currently succeed. Used to close the lost-wakeup
    /// race: a producer that parks must re-check after clearing its queued flag,
    /// because a space_callback fired in between would otherwise be swallowed.
    bool has_space() const {
        return tail_.load(std::memory_order_relaxed) -
               head_.load(std::memory_order_acquire) < capacity_;
    }
    /// Outcome of a non-blocking push.
    ///
    /// try_push used to return bool, and returned *true* for a closed channel —
    /// so "delivered" and "discarded because nobody is listening" were the same
    /// answer. Both mean "stop trying", which is why the callers were correct,
    /// but neither they nor the producer's own accounting could tell a value
    /// that arrived from one that was thrown away. Only the channel's drop
    /// counter knew.
    enum class PushResult { Taken, Full, Closed };
    /// Non-blocking, lossless push. Returns Full when the ring is full, having
    /// changed nothing — the caller keeps the value and retries when woken.
    PushResult try_push(T& value) {
        if (!accepting_.load(std::memory_order_acquire)) {
            stats_.record_drop();
            return PushResult::Closed;
        }
        const std::size_t t = tail_.load(std::memory_order_relaxed);
        const std::size_t h = head_.load(std::memory_order_acquire);
        if (t - h >= capacity_) return PushResult::Full;
        const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
        buf_[t & ring_mask_] = make_storage(std::move(value));
        tail_.store(t + 1, std::memory_order_release);
        stats_.record_push(t - h + 1, data_bytes);
        wake_.fetch_add(1, std::memory_order_release);
        wake_.notify_one();
        // Level-triggered, not edge-triggered — see set_push_callback.
        if (push_callback_) push_callback_();
        return PushResult::Taken;
    }
    // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to
    // drain instead of dropping (the throwing push()) — the producer just runs slower.
    // Use when every value must be delivered (e.g. replaying a dump for scoring, where
@@ -212,13 +151,13 @@ public:
            const std::size_t h = head_.load(std::memory_order_acquire);
            if (t - h < capacity_) {          // space available → normal push
                const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
                const bool was_empty = (t == h);
                buf_[t & ring_mask_] = make_storage(std::move(value));
                tail_.store(t + 1, std::memory_order_release);
                stats_.record_push(t - h + 1, data_bytes);
                wake_.fetch_add(1, std::memory_order_release);
                wake_.notify_one();
-                // Level-triggered, not edge-triggered — see set_push_callback.
+                if (was_empty && push_callback_) push_callback_();
                if (push_callback_) push_callback_();
                return true;
            }
            // full: yield briefly and retry (consumer will drain)
@@ -239,50 +178,13 @@ public:
    // preserving ordering (EOF arrives after all data pushed before it).
    //
    // Only the sole producer may call it (SPSC contract, same as push()).
-    //
+    // Returns false if the channel is already disabled (token discarded —
-    // The slot holds exactly one undelivered token. A second offered before the
+    // teardown is in progress, so the sentinel is moot).
-    // first is taken is refused, not queued and not overwritten: two control
+    bool push_sentinel(T value) {
    // tokens on one channel means the stream ended twice, which is a caller
    // protocol error rather than backpressure, and silently coalescing them
    // would hide it.
    /// Outcome of offering a sentinel. SlotBusy is a protocol error, not
    /// backpressure: it means a second control token was offered while the
    /// first was still undelivered, and a channel carries at most one.
    enum class SentinelResult { Taken, Closed, SlotBusy };
    /// Non-consuming form. `value` is left untouched unless the result is
    /// Taken, so a refused token is still the caller's to report.
    SentinelResult try_push_sentinel(T& value) {
        if (!accepting_.load(std::memory_order_acquire)) {
            stats_.record_drop();
-            return SentinelResult::Closed;
+            return false;
        }
        // Refuse rather than overwrite. Overwriting lost the first token
        // silently, and worse, wrote eof_value_ while the consumer could be
        // moving the previous one out of it — a data race on the storage, which
        // for a shared_ptr payload is a torn refcount rather than a stale read.
        //
        // Checking here is what makes the slot a correct SPSC handshake: the
        // producer is the only writer of eof_value_ and the only one that sets
        // has_eof_, the consumer is the only one that clears it, so observing
        // false here means the consumer has finished with the storage and will
        // not touch it again until this store publishes the next token.
        //
        // Not counted as a drop, and this is the important part. A source that
        // has reached the end of its input keeps being polled and keeps
        // returning EOF — that is the normal steady state, not an error — so a
        // token arriving while one is already pending is a *re-offer*, and
        // refusing it loses nothing: the pending token carries the same
        // meaning and is already on its way. Counting it as a drop made a
        // clean run report data loss and exit non-zero.
        //
        // The cost of that choice, stated plainly: a genuinely distinct second
        // token would also be refused silently, and the channel cannot tell the
        // two apart. Re-offering is the case that actually occurs here, and the
        // delivery guarantee that matters — the first token arrives — holds
        // either way.
        if (has_eof_.load(std::memory_order_acquire))
            return SentinelResult::SlotBusy;
        eof_value_ = make_storage(std::move(value));
        has_eof_.store(true, std::memory_order_release);
        // Wake a consumer blocked in pop(): the sentinel is now deliverable even
@@ -290,13 +192,7 @@ public:
        wake_.fetch_add(1, std::memory_order_release);
        wake_.notify_one();
        if (push_callback_) push_callback_();
-        return SentinelResult::Taken;
+        return true;
    }
    /// Consuming convenience form. Returns false when the token was not stored,
    /// whether because the channel is closed or because one is already pending.
    bool push_sentinel(T value) {
        return try_push_sentinel(value) == SentinelResult::Taken;
    }
    // Blocking pop. Returns when an item is available.
@@ -352,8 +248,6 @@ public:
                throw ChannelClosedError{};
            T value = extract(std::move(buf_[h & ring_mask_]));
            head_.store(h + 1, std::memory_order_release);
            // A slot just freed: wake any producer parked on this channel.
            if (t - h >= capacity_ && space_callback_) space_callback_();
            stats_.record_pop();
            return value;
        }
@@ -375,15 +269,11 @@ public:
    // so pool nodes — which pop only via this path — still receive the token.
    bool try_pop_now(T& out) {
        const std::size_t h = head_.load(std::memory_order_relaxed);
-        const std::size_t t = tail_.load(std::memory_order_acquire);
+        if (h == tail_.load(std::memory_order_acquire))
        if (h == t)
            return take_sentinel(out);
        out = extract(std::move(buf_[h & ring_mask_]));
        head_.store(h + 1, std::memory_order_release);
        stats_.record_pop();
        // Pool nodes pop only through here, so this is where a parked producer
        // gets woken: the ring was full, and it no longer is.
        if (t - h >= capacity_ && space_callback_) space_callback_();
        return true;
    }
@@ -400,56 +290,16 @@ public:
        wake_.notify_all();
    }
-    // Register a callback fired after every successful push.
+    // Register a callback fired when the queue transitions empty→non-empty.
    //
    // It fires on every push, not on the empty→non-empty transition, and that
    // is a correctness requirement rather than a simplification.
    //
    // The edge version tested `was_empty = (t == h)` using an `h` sampled
    // *before* the item was published. A PoolNode consumer decides whether to
    // run again from the level (count_ready → approx_size), so the two sides
    // could each read the other as stale and both stand down:
    //
    //   producer (push)                    consumer (PoolNode firing)
    //   ------------------------           ----------------------------
    //   samples t=782, h=781
    //     -> was_empty = false, no wake
    //                                      pops idx 781, head_ = 782
    //                                      count_ready(): head_==tail_==782
    //                                      -> not ready, gate released to Idle
    //   tail_.store(783)
    //
    // The item is in the ring, the node is idle, and no wake is outstanding.
    // Worse, the failure is absorbing: every later push now sees a non-empty
    // ring, so `was_empty` is false forever and the callback never fires again.
    // The node sleeps while its backlog grows and its consumer waits on it.
    //
    // Re-reading head_ after the tail_ store does not fix it. That is the
    // store-buffer pattern, and under acquire/release both sides may legally
    // read stale; forbidding it needs seq_cst on the producer's tail_ store and
    // head_ load *and* on the consumer's head_ store and tail_ load — a fence
    // on both hot paths. Firing unconditionally is correct by construction:
    // the callback runs after the publishing store, so a consumer that observes
    // the level at all observes the item.
    //
    // The redundant wakes are cheap. on_input_ready re-checks the level, and
    // SubmitGate::claim() collapses a wake arriving during a firing into the
    // firing already in flight, so the cost is one CAS, not one extra run.
    void set_push_callback(std::function<void()> cb) {
        push_callback_ = std::move(cb);
    }
    // Ring occupancy, derived lazily from indices — no separate counter on the
    // hot path. Excludes any out-of-band sentinel (that lives outside the ring).
    // head_ is loaded first, deliberately. Both indices only ever increase, so
    // reading head_ before tail_ can at worst under-report a concurrent push;
    // the other order can read a head_ that has advanced past the tail_ already
    // sampled, and the unsigned difference then wraps to ~2^64. A caller
    // polling "is this channel empty yet" against that value never terminates.
    std::size_t size() const {
-        const std::size_t h = head_.load(std::memory_order_relaxed);
+        return tail_.load(std::memory_order_relaxed)
-        const std::size_t t = tail_.load(std::memory_order_acquire);
+             - head_.load(std::memory_order_relaxed);
        return t - h;
    }
    // A pending out-of-band sentinel (EOF) counts as consumable work here even
@@ -466,9 +316,8 @@ public:
    const ChannelStats& stats() const { return stats_; }
    ChannelSnapshot snapshot(const std::string& name) const {
-        // head_ before tail_, for the reason given on size().
+        const std::size_t t = tail_.load(std::memory_order_relaxed);
        const std::size_t h = head_.load(std::memory_order_relaxed);
        const std::size_t t = tail_.load(std::memory_order_acquire);
        return {
            name,
            capacity_,
@@ -503,24 +352,6 @@ private:
    // delivered after every value pushed before it.
    bool take_sentinel(T& out) {
        if (!has_eof_.load(std::memory_order_acquire)) return false;
        // Re-check emptiness *after* observing has_eof_, not before.
        //
        // Callers check the ring is empty and then call this, but the producer
        // can push a value and publish the sentinel in the window between those
        // two steps — so the sentinel would be delivered with a real value still
        // queued behind it, breaking the "sentinel is strictly last" contract
        // that downstream teardown depends on. a0c4bf5 closed the variant where
        // the caller's emptiness check used a stale tail_ snapshot; this is the
        // one where the check is fresh but simply too early.
        //
        // Checking here is what makes it sound: the producer publishes the
        // sentinel with a release store *after* its ring pushes, so a consumer
        // that has observed has_eof_ has also observed every tail_ advance
        // before it. If the ring is non-empty now, those values genuinely
        // precede the sentinel and must be delivered first.
        if (head_.load(std::memory_order_relaxed)
                != tail_.load(std::memory_order_acquire))
            return false;
        out = extract(std::move(eof_value_));
        has_eof_.store(false, std::memory_order_release);
        stats_.record_pop();
@@ -532,7 +363,6 @@ private:
    std::size_t                     ring_mask_;
    std::unique_ptr<storage_type[]> buf_;
    std::function<void()>           push_callback_;
    std::function<void()> space_callback_;
    ChannelStats                    stats_;
    // Out-of-band sentinel (EOF): stored outside the ring so its delivery never
@@ -51,14 +51,6 @@ struct NodeStats {
    std::atomic<int64_t> max_exec_us{0};
    std::atomic<int64_t> total_blocked_us{0};
    // Cumulative wall time inside fire_once, summed over every invocation.
    // The EMA above cannot be turned into a total: it is exponentially
    // weighted, so frames * ema_exec_us tracks the tail of the run rather than
    // the whole of it, and on a workload whose per-frame cost varies (a face
    // detector on a film: crowd scenes then empty landscapes) the two differ by
    // a lot. Answering "how much time went into this node" needs a real sum.
    std::atomic<int64_t> total_exec_us{0};
    // Thread CPU time — actual CPU consumed by this node's thread,
    // measured via CLOCK_THREAD_CPUTIME_ID. Excludes time sleeping or
    // blocked on mutexes/channels. Sampled once per frame.
@@ -97,7 +89,6 @@ struct NodeStats {
        frames_processed.fetch_add(1, std::memory_order_relaxed);
        int64_t us = static_cast<int64_t>(exec_time.count() * 1000.0);
        total_exec_us.fetch_add(us, std::memory_order_relaxed);
        uint64_t n = frames_processed.load(std::memory_order_relaxed);
        int64_t prev = ema_exec_us.load(std::memory_order_relaxed);
@@ -156,29 +147,6 @@ struct NodeSnapshot {
    double      total_cpu_ms;       // cumulative CPU time consumed by this node's thread
    double      cpu_util_pct;       // exec_ms / (exec_ms + blocked_ms) * 100
    double      queue_wait_ms{0};   // PoolNode: cumulative time spent in pool queue
    // Cumulative wall time inside fire_once. Unlike ema_exec_ms this is a true
    // sum, so it is the field to use for "share of the run spent in this node".
    // Note it still includes time parked pushing into a full output channel;
    // total_cpu_ms is the part that backpressure cannot inflate.
    //
    // Declared before the two bools below because every node type initialises
    // this aggregate positionally, and all of them supply total_exec_ms as the
    // element after queue_wait_ms.
    double      total_exec_ms{0};
    // Live scheduling state, for observing the AR-004 invariant "a node never
    // sleeps with a wake outstanding". The invariant was previously asserted in
    // comments but invisible at runtime, so a lost wake could only be found in a
    // debugger — and this bug does not reproduce under one (it needs full speed).
    // Two atomic loads at snapshot time, nothing on the hot path.
    //
    // Read them together with the node's channel fill:
    //   queued=0, wake=1                -> wake recorded and never consumed
    //   queued=0, wake=0, input full    -> wake never generated at all
    //   queued=1 while nothing running  -> submitted but never scheduled
    bool        queued{false};
    bool        wake_pending{false};
 };
 // ── Pool statistics + snapshot ────────────────────────────────────────────────
@@ -220,11 +188,6 @@ struct ResourceSnapshot {
 struct IResourceProbe {
    virtual ~IResourceProbe() = default;
    virtual ResourceSnapshot snapshot(const std::string& name) const = 0;
    /// Release every thread waiting for the resource, so teardown is not held
    /// up by one. A network calls this on the resources registered with it when
    /// it halts; default no-op for probes with nothing to wake.
    virtual void close() {}
 };
 } // namespace kpn
@@ -7,10 +7,8 @@
 #include <array>
 #include <atomic>
 #include <chrono>
 #include <iostream>
 #include <memory>
 #include <optional>
 #include <thread>
 #include <tuple>
 #include <utility>
@@ -80,9 +78,7 @@ public:
                blocked_ms,
                elapsed_s > 0 ? frames / elapsed_s : 0.0,
                stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
-                total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
+                total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0};
                0.0,  // queue_wait_ms — fanout is not pool-scheduled
                stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0};
    }
    // ── Port access ───────────────────────────────────────────────────────────
@@ -119,80 +115,17 @@ public:
        out_channels_[I] = ch;
    }
    // Lossless fanout: block until each consumer drains rather than dropping.
    void set_lossless_output(bool on) override { lossless_ = on; }
    // Opt a single output back out of blocking.  Needed when one branch may
    // stall indefinitely — a display tap nobody is servicing, say — since
    // blocking on it would apply backpressure to every other branch too.
    void set_lossy_output(std::size_t i, bool lossy = true) {
        if (i < N) lossy_out_[i] = lossy;
    }
 private:
    // Deliver `val` to every connected output, losslessly.
    //
    // Previously a full output cost the value: push() threw and the exception was
    // swallowed per output. A dropped item does not degrade a downstream result,
    // it silently changes one, and the consumer cannot tell it happened — so the
    // fanout waits instead, and the producer upstream runs slower.
    //
    // Unlike a pool node, a fanout owns a private thread, so waiting here costs
    // no scheduler worker and needs no space-callback park; a bounded retry is
    // enough. `stop_flag_` is re-checked every pass so teardown cannot hang on a
    // full output regardless of the order the network stops its nodes in.
    //
    // Outputs are retried independently, so a full output never delays delivery
    // to one with room. Note what that does *not* buy: the next input is not
    // popped until every output has accepted the current item, so one branch can
    // never run ahead of another by more than the slower branch's buffering.
    //
    // **That bound is a precondition on any topology where the branches rejoin.**
    // If a consumer on branch B blocks waiting for something branch A computes,
    // B's buffering must exceed the lead A needs, or the two wedge — B waiting on
    // A, A starved because the fanout is holding an item B will not take. Making
    // the fanout lossless is what puts that precondition on the topology; while
    // it dropped, the question could not arise.
    //
    // `parked` receives the time spent waiting on a full output, which the caller
    // charges to blocked rather than exec.
    //
    // Returns false if stopped with the value undelivered.
    bool deliver(const T& val, duration_t& parked) {
        std::array<std::optional<T>, N> pending;
        std::size_t outstanding = 0;
        for (std::size_t i = 0; i < N; ++i)
            if (out_channels_[i]) { pending[i].emplace(val); ++outstanding; }
        bool first_pass = true;
        auto park_from  = clock_t::now();
        for (;;) {
            for (std::size_t i = 0; i < N; ++i) {
                if (!pending[i]) continue;
                // Taken or Closed both mean "stop trying" — delivered, or gone
                // with the drop recorded. Only Full is worth another pass.
                if (out_channels_[i]->try_push(*pending[i])
                        != Channel<T>::PushResult::Full) {
                    pending[i].reset();
                    --outstanding;
                }
            }
            if (first_pass) { park_from = clock_t::now(); first_pass = false; }
            if (outstanding == 0) {
                parked = duration_t(clock_t::now() - park_from);
                return true;
            }
            if (stop_flag_.load(std::memory_order_relaxed)) {
                // Teardown with work in hand. One last throwing push per
                // outstanding output, purely so the channel's own stats record
                // the loss (drop if it is disabled, overflow if it is merely
                // full). The whole point of the lossless path is that a loss is
                // never invisible, and a silent `return` here would reintroduce
                // exactly the hole this function exists to close.
                for (std::size_t i = 0; i < N; ++i) {
                    if (!pending[i]) continue;
                    try { out_channels_[i]->push(std::move(*pending[i])); }
                    catch (const ChannelOverflowError&) {}
                }
                parked = duration_t(clock_t::now() - park_from);
                return false;
            }
            std::this_thread::sleep_for(std::chrono::microseconds(50));
        }
    }
    void run_loop() {
        while (!stop_flag_.load(std::memory_order_relaxed)) {
            try {
@@ -201,17 +134,27 @@ private:
                auto t1  = clock_t::now();
                auto cpu0 = NodeStats::cpu_now();
-                duration_t parked{0};
+                for (std::size_t i = 0; i < N; ++i) {
-                const bool delivered = deliver(val, parked);
+                    if (out_channels_[i]) {
                        // Lossless: block until this consumer drains.  Note the
                        // branches differ in more than blocking — the dropping
                        // path discards per-output independently and silently,
                        // so a slow consumer on one branch costs frames on that
                        // branch only, with no diagnostic.  That is the right
                        // default for display taps but hides frame loss from
                        // analysis branches.
                        if (lossless_ && !lossy_out_[i])
                            out_channels_[i]->push_blocking(val);
                        else {
                            try { out_channels_[i]->push(val); }
                            catch (const ChannelOverflowError&) {}  // drop independently
                        }
                    }
                }
                auto cpu1 = NodeStats::cpu_now();
                auto t2   = clock_t::now();
-                // Time spent waiting on a full output is *blocked*, not exec: a
+                stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
                // parked fanout is idle, and charging it to exec would report the
                // node as busy exactly when it is the one being held up.
                stats_.record_exec(duration_t(t2 - t1) - parked,
                                   duration_t(t1 - t0) + parked, cpu0, cpu1);
                if (!delivered) break;
            } catch (const ChannelClosedError&) {
                break;
            }
@@ -220,6 +163,8 @@ private:
    std::string                 name_;
    std::size_t                 fifo_capacity_;
    bool                        lossless_{false};
    std::array<bool, N>         lossy_out_{};   // per-output opt-out of blocking
    std::shared_ptr<Channel<T>> input_ch_;
    std::array<Channel<T>*, N>  out_channels_{};
    std::atomic<bool>           stop_flag_{false};
@@ -22,23 +22,6 @@ enum class NodeEvent { Overflow, Closed };
 struct INode {
    virtual ~INode()       = default;
    // Install channel callbacks, without starting anything.
    //
    // A node's push/space callbacks live in std::function members on channels
    // it shares with its neighbours, and a neighbour that is already running
    // reads them on its own thread. Writing one while the pipeline runs is a
    // data race on the std::function — ThreadSanitizer reports it, and the
    // consequence in the field was the missed startup wake a8cfe73 had to
    // patch around.
    //
    // So a network calls prepare() on every node before it calls start() on
    // any of them: all the writes happen while nothing is running, and once a
    // node is live the callbacks are read-only. start() calls prepare() itself
    // if it has not been called, so standalone nodes still work; it is
    // idempotent, and the network relies on that.
    virtual void prepare() {}
    virtual void start()   = 0;
    virtual void stop()    = 0;
    virtual bool running() const = 0;
@@ -51,17 +34,13 @@ struct INode {
    virtual void set_network_overflow_callback(NodeEventCallback) {}
    virtual void set_network_closed_callback(NodeEventCallback)   {}
    // Network-level error listener. Consulted when a node's function throws
    // and no per-node handler resolved it. Without this the exception is
    // discarded and the failure is only visible as a Closed event, which says
    // a node stopped but not why — the difference between a diagnosis and a
    // guess. Same contract as NodeErrorHandler: true to continue, false to
    // stop the node.
    virtual void set_network_error_callback(NodeErrorHandler) {}
    // halt(): alias for stop() — immediate, discards in-flight work.
    virtual void halt() { stop(); }
    // Opt into lossless (blocking) output for nodes that support it.  Default
    // is a no-op so node types with no output channels ignore it.
    virtual void set_lossless_output(bool) {}
    // shutdown(): graceful drain before stopping. Base implementation falls
    // back to stop(). Network and StaticNetwork override with topo-ordered drain.
    virtual void shutdown() { stop(); }
@@ -104,7 +104,6 @@ public:
            stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
            total_ms > 0 ? 100.0 : 0.0,
            qwait_ms,
            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
        };
    }
@@ -90,8 +90,6 @@ public:
            elapsed_s > 0 ? frames / elapsed_s : 0.0,
            stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
            total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
            0.0,  // queue_wait_ms — main-thread node is not pool-scheduled
            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
        };
    }
@@ -8,10 +8,8 @@
 #include <memory>
 #endif
 #include <condition_variable>
 #include <functional>
 #include <iomanip>
 #include <mutex>
 #include <iostream>
 #include <map>
 #include <set>
@@ -41,15 +39,8 @@ public:
 class Network : public INode {
 public:
-    /// Application-level error listener. Receives the exception any node's
+    using ErrorHandler =
-    /// function throws, after that node's own handler (if any) declined it.
+        std::function<void(std::string_view node_name, std::exception_ptr)>;
    /// Return true to skip the failed invocation and keep the node running,
    /// false to let it stop.
    ///
    /// Same type as StaticNetwork's, deliberately: this used to be a void
    /// signature, which could not express the keep-running decision and, more
    /// to the point, was never delivered anywhere.
    using ErrorHandler = NodeErrorHandler;
    using DiagnosticsHandler =
        std::function<void(const std::vector<NodeSnapshot>&,
                           const std::vector<ChannelSnapshot>&)>;
@@ -100,7 +91,6 @@ public:
                            + " → " + dst_name + ":" + std::to_string(DstIdx);
        channel_probes_.push_back(
            std::make_unique<ChannelProbe<out_t>>(in_ch, ch_name));
        channel_src_names_.push_back(src_name);
        adj_[src_name].push_back(dst_name);
        return *this;
@@ -144,17 +134,6 @@ public:
    void start() override {
        start_time_ = clock_t::now();
        // Deliver the listener to the nodes. Without this the handler was
        // stored and never read: a node's exception was discarded at the node
        // boundary and the only surviving evidence was a Closed event, which
        // says a node stopped but not why. StaticNetwork has always done this;
        // Network accepted the handler and silently dropped it.
        if (error_handler_)
            for (auto& name : topo_)
                nodes_.at(name)->set_network_error_callback(error_handler_);
        // Callbacks first, everywhere, before anything runs — see INode::prepare.
        for (auto& name : topo_)
            nodes_.at(name)->prepare();
        for (auto& name : topo_)
            nodes_.at(name)->start();
        start_watchdog();
@@ -231,11 +210,6 @@ public:
        watchdog_interval_ = interval;
    }
    /// How long shutdown() waits for one node's outputs to drain before giving
    /// up on them and stopping the next layer anyway.
    void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
    /// Must be called before start(); the handler is delivered to nodes there.
    void set_error_handler(ErrorHandler h)             { error_handler_  = std::move(h); }
    void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_  = std::move(h); }
    void set_event_handler(EventHandler h)             { event_handler_  = std::move(h); }
@@ -393,46 +367,19 @@ private:
        return true;
    }
-    /// Wait for the channels fed by `name` to empty, or give up.
+    void drain_output_channels(const std::string& /*name*/) const {
-    ///
+        // Poll all channel probes until none report non-zero fill.
-    /// This took a node name and ignored it, polling *every* channel in the
+        // A short sleep prevents busy-spin; 1 ms is fine for drain purposes.
-    /// graph instead — so shutdown() waited for the whole network to be idle
+        bool any_full = true;
-    /// before stopping each successive layer. With no deadline either, anything
+        while (any_full) {
-    /// wedged downstream turned a graceful shutdown into the hang it exists to
+            any_full = false;
-    /// avoid.
+            for (auto& probe : channel_probes_) {
-    ///
+                auto snap = probe->snapshot();
-    /// Two bounds, because they fail differently. The deadline covers a
+                if (snap.current_fill > 0) { any_full = true; break; }
-    /// consumer that has stopped consuming, where fill never changes and
+            }
-    /// waiting cannot help. The no-progress counter covers one that is merely
+            if (any_full)
    /// slow: it keeps waiting while the queue is shrinking, so a slow drain is
    /// not cut short just for taking a while.
    void drain_output_channels(const std::string& name) const {
        const auto deadline = clock_t::now() + drain_timeout_;
        std::size_t last_fill = static_cast<std::size_t>(-1);
        int         stalls    = 0;
        auto fill_of = [&] {
            std::size_t fill = 0;
            for (std::size_t i = 0; i < channel_probes_.size(); ++i)
                if (channel_src_names_[i] == name)
                    fill += channel_probes_[i]->snapshot().current_fill;
            return fill;
        };
        for (;;) {
            const std::size_t fill = fill_of();
            if (fill == 0) return;
            if (fill >= last_fill) { if (++stalls > 100) break; }
            else                   { stalls = 0; }
            last_fill = fill;
            if (clock_t::now() >= deadline) break;
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
        if (const std::size_t left = fill_of())
            std::cerr << "[kpn] shutdown: '" << name << "' still has " << left
                      << " queued item(s) its consumer did not take; "
                         "they are discarded\n";
    }
    // ── Cycle detection / topological sort ───────────────────────────────────
@@ -451,20 +398,9 @@ private:
    void start_watchdog() {
        watchdog_ = std::jthread([this](std::stop_token tok) {
            // Interruptible wait, not sleep_for. request_stop() cannot wake a
            // sleeping thread, so stop_watchdog()'s join blocked for up to a
            // full interval — three seconds by default, and unbounded for
            // anyone who set a long one to keep the periodic report quiet.
            // Every teardown paid it.
            std::mutex                  m;
            std::condition_variable_any cv;
            while (!tok.stop_requested()) {
-                {
+                std::this_thread::sleep_for(watchdog_interval_);
-                    std::unique_lock lk(m);
+                if (tok.stop_requested()) break;
                    if (cv.wait_for(lk, tok, watchdog_interval_,
                                    [&tok] { return tok.stop_requested(); }))
                        break;
                }
                auto s = collect_snapshots();
                check_hung_nodes();
@@ -508,10 +444,6 @@ private:
    std::map<std::string, std::string>                     exposed_outputs_;
    std::set<std::pair<std::string, std::size_t>>          connected_outputs_;
    std::vector<std::unique_ptr<IChannelProbe>>            channel_probes_;
    /// Name of the node feeding each probe, parallel to channel_probes_.
    /// shutdown() drains a node's own outputs, so it has to know which they are.
    std::vector<std::string>                               channel_src_names_;
    std::chrono::milliseconds                              drain_timeout_{5000};
    std::vector<std::pair<std::string, IPoolProbe*>>       pool_probes_;
    ErrorHandler                                           error_handler_;
    DiagnosticsHandler                                     diag_handler_;
@@ -5,7 +5,6 @@
 #include <functional>
 #include <memory>
 #include <mutex>
 #include <shared_mutex>
 #include <optional>
 #include <queue>
 #include <thread>
@@ -53,19 +52,6 @@ public:
    }
    void start() override {
        // Under the lifecycle lock for the same reason stop() is: submit()
        // reads queues_ and this rebuilds it. A network starts its nodes one at
        // a time, and a node already started fires into the next one's channel,
        // whose push callback submits — so a submission can genuinely land
        // while another pool is still inside start(). ThreadSanitizer reports
        // it as a read at submit() against this write, and the consequence is
        // worse than a torn read: push_back can reallocate the vector under a
        // reader that has already indexed it.
        //
        // Queues are all constructed before any worker is spawned, which is
        // what keeps worker_loop's own queues_[id] out of this — it never takes
        // the lock, so holding it across the spawn cannot deadlock.
        std::unique_lock lk(lifecycle_mx_);
        stopped_.store(false, std::memory_order_relaxed);
        queues_.clear();
        for (std::size_t i = 0; i < thread_count_; ++i)
@@ -76,30 +62,18 @@ public:
    }
    void stop() override {
        // Close the pool to new work before touching anything, and do it under
        // the lifecycle lock so no submit() is midway through indexing queues_.
        {
            std::unique_lock lk(lifecycle_mx_);
        stopped_.store(true, std::memory_order_seq_cst);
        }
        for (auto& q : queues_) {
            std::lock_guard lock(q->mx);
            std::size_t discarded = q->pq.size();
            while (!q->pq.empty()) q->pq.pop();
            total_.fetch_sub(discarded, std::memory_order_relaxed);
            queued_.fetch_sub(discarded, std::memory_order_relaxed);
        }
        // Lock cv_mx_ before notifying so the stop signal can't be lost in the
        // gap between a worker's predicate check and its wait() (see submit()).
        { std::lock_guard<std::mutex> lk(cv_mx_); }
        cv_.notify_all();
        // Join without the lock: a worker's task may call submit(), which takes
        // it shared, and holding it here would deadlock against that.
        for (auto& t : workers_) if (t.joinable()) t.join();
        // Destroying the queues is what submit() must never race. By now
        // stopped_ is published, so any submit() that acquires the lock after
        // this point returns without touching them.
        std::unique_lock lk(lifecycle_mx_);
        workers_.clear();
        queues_.clear();
    }
@@ -112,22 +86,6 @@ public:
    }
    void submit(std::function<void()> task, float priority = 0.5f) override {
        // A submission can arrive after this pool has been stopped, and did so
        // by an ordinary route: a node's space callback fires from whichever
        // thread drained the channel, which belongs to the *consumer*. Stop the
        // producer first — as a sources-first shutdown does — and the consumer
        // keeps draining its backlog, firing the producer's space callback into
        // a pool whose stop() has already run queues_.clear(). submit() then
        // indexed an empty vector: a segfault, reproducible about 12 runs in 20.
        //
        // The shared lock is what makes the check meaningful. Reading stopped_
        // alone leaves the window between the read and the indexing, which is
        // precisely where stop() clears the vector.
        std::shared_lock lk(lifecycle_mx_);
        if (stopped_.load(std::memory_order_acquire) || queues_.empty()) {
            rejected_.fetch_add(1, std::memory_order_relaxed);
            return;
        }
        std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
        {
            std::lock_guard lock(queues_[target]->mx);
@@ -135,7 +93,6 @@ public:
                {std::move(task), priority, seq_.fetch_add(1, std::memory_order_relaxed)});
        }
        total_.fetch_add(1, std::memory_order_relaxed);
        queued_.fetch_add(1, std::memory_order_relaxed);
        submitted_.fetch_add(1, std::memory_order_relaxed);
        // Synchronize with worker_loop's predicate evaluation: taking cv_mx_
        // here guarantees a worker is either before its predicate check (and
@@ -148,16 +105,14 @@ public:
    std::size_t thread_count() const { return thread_count_; }
    /// Submissions dropped because the pool was stopped. See rejected_.
    uint64_t rejected() const { return rejected_.load(std::memory_order_relaxed); }
    // ── IPoolProbe ────────────────────────────────────────────────────────────
    PoolSnapshot snapshot(const std::string& name) const override {
        std::size_t a = active_.load(std::memory_order_relaxed);
        std::size_t t = total_.load(std::memory_order_relaxed);
        return {
            name, thread_count_,
-            queued_.load(std::memory_order_relaxed),   // queued (exact)
+            t > a ? t - a : 0,   // queued (approximate)
            a,                    // executing
            submitted_.load(std::memory_order_relaxed),
            completed_.load(std::memory_order_relaxed),
@@ -187,7 +142,6 @@ private:
        if (q.pq.empty()) return std::nullopt;
        auto fn = std::move(const_cast<Task&>(q.pq.top()).fn);
        q.pq.pop();
        queued_.fetch_sub(1, std::memory_order_relaxed);
        return fn;
    }
@@ -228,13 +182,10 @@ private:
            std::unique_lock lock(cv_mx_);
            cv_.wait(lock, [this] {
                return stopped_.load(std::memory_order_seq_cst)
-                    || queued_.load(std::memory_order_relaxed) > 0;
+                    || total_.load(std::memory_order_relaxed) > 0;
            });
            // Exit on queued_, not total_: waiting for total_ to reach zero
            // meant waiting for someone else's task to finish, which this
            // worker cannot help with and would spin through until it did.
            if (stopped_.load(std::memory_order_seq_cst)
-                && queued_.load(std::memory_order_relaxed) == 0)
+                && total_.load(std::memory_order_relaxed) == 0)
                return;
        }
    }
@@ -243,36 +194,17 @@ private:
    std::vector<std::unique_ptr<WorkerQueue>>  queues_;
    std::vector<std::thread>                   workers_;
    /// Guards the lifetime of queues_/workers_ against a concurrent submit().
    /// Shared by submit, exclusive by stop, so submissions still run in
    /// parallel with each other.
    mutable std::shared_mutex lifecycle_mx_;
    std::mutex              cv_mx_;
    std::condition_variable cv_;
    std::mutex              drain_mx_;
    std::condition_variable drain_cv_;
    std::atomic<bool>     stopped_{true};
-    std::atomic<size_t>   total_{0};      // queued + executing (drain() waits on this)
+    std::atomic<size_t>   total_{0};      // queued + executing
    /// Queued only — never counts a task that is already executing.
    ///
    /// The wait predicate used total_, which includes running tasks, so while
    /// any one task ran every *other* worker's predicate was true: wait()
    /// returned instantly and the worker spun through try_pop / try_steal /
    /// wait at full speed, try_lock-ing every peer queue on each pass. One slow
    /// task therefore pinned every other core and contended the very mutexes
    /// the working thread needed. Sleeping requires "no work is *waiting*",
    /// which is this.
    std::atomic<size_t>   queued_{0};     // waiting to run
    std::atomic<size_t>   active_{0};     // executing only (for snapshot)
    std::atomic<size_t>   next_{0};       // round-robin submit cursor
    std::atomic<uint64_t> seq_{0};        // tie-break for equal-priority tasks
    std::atomic<uint64_t> submitted_{0};
    /// Submissions refused because the pool was already stopped. Not an error —
    /// teardown races are expected — but silence here would hide a node that
    /// keeps trying to run after its pool is gone.
    std::atomic<uint64_t> rejected_{0};
    std::atomic<uint64_t> completed_{0};
 };
@@ -14,18 +14,6 @@ namespace kpn {
 template<typename T> class Channel; // forward declaration for acquire_balanced
 /// Thrown by a pending acquire() when the resource is closed underneath it.
 ///
 /// acquire() blocks on a condition variable with no timeout and no stop
 /// condition, so a node parked there ignored teardown entirely: the worker
 /// never returned, the pool's join never completed, and shutdown hung on a
 /// resource nobody was going to release. Closing the resource turns that into
 /// an exception the node's normal error path already handles.
 class ResourceClosedError : public std::runtime_error {
 public:
    ResourceClosedError() : std::runtime_error("shared resource closed") {}
 };
 // ── SharedResource ────────────────────────────────────────────────────────────
 //
 // Wraps an exclusive resource (e.g. an ONNX session, a CUDA stream) and
@@ -84,7 +72,6 @@ public:
    template<typename PriorityFn>
    Guard acquire(PriorityFn&& fn) {
        std::unique_lock lock(mutex_);
        if (closed_) throw ResourceClosedError{};
        if (!held_) {
            held_ = true;
            acq_.fetch_add(1, std::memory_order_relaxed);
@@ -96,46 +83,18 @@ public:
        current_waiters_.store(waiters_.size(), std::memory_order_relaxed);
        auto t0 = w.wait_start;
-        // Woken either by release() handing over ownership, or by close()
+        w.cv.wait(lock, [&w] { return w.ready; });
        // giving up on the wait entirely.
        w.cv.wait(lock, [&w] { return w.ready || w.closed; });
        int64_t wait_us = std::chrono::duration_cast<std::chrono::microseconds>(
            clock_t::now() - t0).count();
        waiters_.erase(std::find(waiters_.begin(), waiters_.end(), &w));
        current_waiters_.store(waiters_.size(), std::memory_order_relaxed);
        acq_.fetch_add(1, std::memory_order_relaxed);
        total_wait_us_.fetch_add(static_cast<uint64_t>(wait_us > 0 ? wait_us : 0),
                                 std::memory_order_relaxed);
        // Closed without being handed ownership: no Guard, so nothing to
        // release, and held_ is left exactly as close() found it.
        if (!w.ready) throw ResourceClosedError{};
        acq_.fetch_add(1, std::memory_order_relaxed);
        return Guard(this);
    }
    /// Wake every waiter and refuse further acquisitions.
    ///
    /// Teardown is the whole point: a node parked in acquire() is not
    /// observing stop flags, so without this the only way out is for whoever
    /// holds the resource to release it — which, if that node is also being
    /// stopped, may never happen. Idempotent, and safe to call from any thread.
    void close() override {
        std::lock_guard lock(mutex_);
        closed_ = true;
        for (Waiter* w : waiters_) {
            w->closed = true;
            w->cv.notify_one();
        }
    }
    /// Reopen after a close(). For reuse across runs; not needed for teardown.
    void reopen() {
        std::lock_guard lock(mutex_);
        closed_ = false;
    }
    // Acquire with no priority (all waiters treated equally, order is fair-ish).
    Guard acquire() {
        return acquire([] { return 0.5f; });
@@ -173,10 +132,7 @@ public:
 private:
    void release() {
        std::unique_lock lock(mutex_);
-        // Hand over only to a waiter that is still waiting. A closed one is on
+        if (waiters_.empty()) {
        // its way out and will not take ownership, so treating it as the next
        // holder would leave held_ true with nobody holding it.
        if (closed_ || waiters_.empty()) {
            held_ = false;
            return;
        }
@@ -206,8 +162,7 @@ private:
        std::function<float()>  priority_fn;
        clock_t::time_point     wait_start;
        std::condition_variable cv;
-        bool                    ready{false};   // handed ownership by release()
+        bool                    ready{false};
        bool                    closed{false};  // woken by close() instead
        Waiter(std::function<float()> fn, clock_t::time_point t)
            : priority_fn(std::move(fn)), wait_start(t) {}
@@ -217,7 +172,6 @@ private:
    T                        resource_;
    bool                     held_{false};
    bool                     closed_{false};
    mutable std::mutex       mutex_;
    std::vector<Waiter*>     waiters_;
    std::atomic<uint64_t>    acq_{0};
@@ -98,15 +98,13 @@ public:
                  std::vector<INode*>        fanout_ptrs,
                  std::vector<std::string>   user_node_names,
                  std::vector<std::string>   fanout_node_names,
-                  std::vector<std::unique_ptr<IChannelProbe>> channel_probes,
+                  std::vector<std::unique_ptr<IChannelProbe>> channel_probes)
                  std::vector<std::string>   channel_src_names)
        : fanouts_(std::move(fanouts))
        , user_nodes_topo_(std::move(user_nodes_topo))
        , fanout_nodes_ptr_(std::move(fanout_ptrs))
        , user_node_names_(std::move(user_node_names))
        , fanout_node_names_(std::move(fanout_node_names))
        , channel_probes_(std::move(channel_probes))
        , channel_src_names_(std::move(channel_src_names))
    {}
    ~StaticNetwork() override { stop(); }
@@ -124,19 +122,6 @@ public:
                    [this, n](auto ts) { event_handler_(n, NodeEvent::Closed, ts); });
            }
        }
        if (error_handler_) {
            for (auto* node : user_nodes_topo_)
                node->set_network_error_callback(error_handler_);
        }
        // Install every node's channel callbacks before starting any of them.
        // Those callbacks are std::function members on channels shared with
        // neighbours; a neighbour that is already running reads them from its
        // own thread, so writing one after the pipeline is live is a data race
        // (ThreadSanitizer reports it on any multi-node network). Doing all the
        // writes here, while nothing runs, makes them read-only thereafter.
        for (auto* n : user_nodes_topo_)  n->prepare();
        for (auto* n : fanout_nodes_ptr_) n->prepare();
        for (auto* n : user_nodes_topo_)  n->start();
        for (auto* n : fanout_nodes_ptr_) n->start();
 #ifdef KPN_WEB_DEBUG
@@ -160,11 +145,6 @@ public:
 #ifdef KPN_WEB_DEBUG
        if (web_server_) web_server_->stop();
 #endif
        // Release anything parked on a shared resource first. A node blocked in
        // acquire() is not watching stop flags, so stopping it would wait on a
        // handover that may never come — its holder is being stopped too.
        for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); }
        for (auto it = fanout_nodes_ptr_.rbegin(); it != fanout_nodes_ptr_.rend(); ++it)
            (*it)->stop();
        for (auto it = user_nodes_topo_.rbegin(); it != user_nodes_topo_.rend(); ++it)
@@ -178,12 +158,11 @@ public:
 #ifdef KPN_WEB_DEBUG
        if (web_server_) web_server_->stop();
 #endif
        for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); }
        // user_nodes_topo_ is already in sources-first order.
        // Stop each node and drain its output channels before moving on.
-        for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
+        for (auto* n : user_nodes_topo_) {
-            user_nodes_topo_[i]->stop();
+            n->stop();
-            drain_outputs_of(user_node_names_[i]);
+            drain_all_channels();
        }
        for (auto* n : fanout_nodes_ptr_) n->stop();
    }
@@ -202,18 +181,6 @@ public:
    void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
    /// How long shutdown() waits for one node's outputs to drain before giving
    /// up on them and stopping the next layer anyway.
    void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
    /// Application-level error listener. Receives the exception any node's
    /// function throws, after that node's own handler (if any) declined it.
    /// Return true to skip the failed invocation and keep the node running,
    /// false to let it stop. Without a listener the exception is discarded
    /// and only a Closed event survives, which reports that a node stopped
    /// but not why.
    void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
 #ifdef KPN_WEB_DEBUG
    void set_web_debug_port(uint16_t port) { web_debug_port_ = port; }
    // Called by DebugHub::register_network() so the hub owns the debug server.
@@ -252,6 +219,24 @@ public:
    FanoutStorage& fanouts_storage() { return *fanouts_; }
    // Make every node in this network push losslessly (block until the consumer
    // drains) instead of dropping on a full channel.  Includes the fanout nodes
    // make_network() inserts automatically, which is the part user code cannot
    // reach: they are unnamed, and they drop silently per-output, so a network
    // whose own nodes are all lossless can still lose items at a fanout.
    //
    // Only safe when every consumer eventually drains.  A branch that can stall
    // indefinitely — a display node nobody is servicing, say — will block the
    // whole pipeline through backpressure.  Call before start().
    void set_lossless(bool on = true) {
        for (auto* n : user_nodes_topo_)  if (n) n->set_lossless_output(on);
        for (auto* n : fanout_nodes_ptr_) if (n) n->set_lossless_output(on);
    }
    // Block until every channel is empty.  Useful before stop() so work already
    // in flight completes rather than being discarded at teardown.
    void drain() const { drain_all_channels(); }
 private:
    struct Snapshots {
        std::vector<NodeSnapshot>     nodes;
@@ -286,50 +271,16 @@ private:
        return {std::move(nodes), std::move(channels), std::move(resources), std::move(pools), elapsed_s};
    }
-    /// Wait for the channels fed by `src` to empty, or give up.
+    void drain_all_channels() const {
-    ///
+        bool any_full = true;
-    /// This was an unbounded `while (anything anywhere is non-empty)` poll over
+        while (any_full) {
-    /// *every* channel in the graph, which made shutdown() wait for the whole
+            any_full = false;
-    /// network to be idle before stopping each successive layer, and wait
+            for (auto& probe : channel_probes_) {
-    /// forever if anything downstream was wedged — turning a graceful shutdown
+                if (probe->snapshot().current_fill > 0) { any_full = true; break; }
-    /// into the hang it exists to avoid.
+            }
-    ///
+            if (any_full)
    /// Two bounds, because they fail differently. The deadline covers a
    /// consumer that has stopped consuming: fill never changes and no amount of
    /// waiting helps. The no-progress counter covers a consumer that is merely
    /// slow — it keeps waiting as long as the queue is shrinking, so a slow
    /// drain is not cut short just for exceeding a fixed time.
    ///
    /// Giving up is reported rather than silent: undrained data at this point
    /// means values are about to be discarded by the stop that follows.
    void drain_outputs_of(const std::string& src) const {
        const auto deadline = clock_t::now() + drain_timeout_;
        std::size_t last_fill = static_cast<std::size_t>(-1);
        int         stalls    = 0;
        for (;;) {
            std::size_t fill = 0;
            for (std::size_t i = 0; i < channel_probes_.size(); ++i)
                if (channel_src_names_[i] == src)
                    fill += channel_probes_[i]->snapshot().current_fill;
            if (fill == 0) return;
            if (fill >= last_fill) { if (++stalls > 100) break; }
            else                   { stalls = 0; }
            last_fill = fill;
            if (clock_t::now() >= deadline) break;
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
        std::size_t left = 0;
        for (std::size_t i = 0; i < channel_probes_.size(); ++i)
            if (channel_src_names_[i] == src)
                left += channel_probes_[i]->snapshot().current_fill;
        if (left)
            std::cerr << "[kpn] shutdown: '" << src << "' still has " << left
                      << " queued item(s) its consumer did not take; "
                         "they are discarded\n";
    }
    std::string                                         name_;
@@ -340,14 +291,9 @@ private:
    std::vector<std::string>                            user_node_names_;
    std::vector<std::string>                            fanout_node_names_;
    std::vector<std::unique_ptr<IChannelProbe>>         channel_probes_;
    /// Display name of the node feeding each probe, parallel to channel_probes_.
    /// shutdown() drains a node's own outputs, so it has to know which they are.
    std::vector<std::string>                            channel_src_names_;
    std::chrono::milliseconds                           drain_timeout_{5000};
    std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
    std::vector<std::pair<std::string, IPoolProbe*>>     pool_probes_;
    EventHandler                                        event_handler_;
    NodeErrorHandler                                    error_handler_;
    clock_t::time_point                                 start_time_;
 #ifdef KPN_WEB_DEBUG
    uint16_t                                            web_debug_port_{9090};
@@ -380,6 +326,29 @@ auto make_network(Edges&&... edges) {
    // 4. Construct owned fanout storage on the heap (FanoutNode has jthread — not moveable)
    auto fanout_storage = std::make_unique<FanoutSto>();
    // 5. Collect unique user node pointers + their display names, in edge-declaration order
    std::vector<INode*>      user_node_ptrs;
    std::vector<std::string> user_node_names;
    auto collect = [&](auto& e) {
        using SrcT = std::decay_t<decltype(e.src)>;
        using DstT = std::decay_t<decltype(e.dst)>;
        auto* s = static_cast<INode*>(&e.src);
        auto* d = static_cast<INode*>(&e.dst);
        if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), s) == user_node_ptrs.end()) {
            auto sname = node_display_name<SrcT>();
            user_node_ptrs.push_back(s);
            user_node_names.push_back(sname);
            s->set_name(sname);
        }
        if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) {
            auto dname = node_display_name<DstT>();
            user_node_ptrs.push_back(d);
            user_node_names.push_back(dname);
            d->set_name(dname);
        }
    };
    (collect(edges), ...);
    // 5. Wire all expanded SimpleEdges.
    //    find_node<NodeT>: searches fanout storage then user edge pack, returns NodeT*.
    //    Uses if constexpr in a fold so mismatched types never reach assignment.
@@ -404,38 +373,6 @@ auto make_network(Edges&&... edges) {
        return ptr;
    };
    // 5. Collect user node pointers + display names in *topological* order.
    //
    // Topo is computed above for the cycle check and used to be discarded,
    // while this vector was filled in edge-declaration order — and then named
    // user_nodes_topo_ and relied upon as if it were sorted. halt() stops in
    // its reverse, and shutdown() walks it forwards stopping each node and
    // draining its outputs before the next, which is only a graceful drain if
    // the order really is sources-first. It held for every network in the tree
    // because edges happen to be declared in pipeline order, and would have
    // broken silently for one that was not.
    //
    // Fanout nodes appear in Topo too; they are skipped here because they are
    // owned separately, in fanout_storage.
    std::vector<INode*>      user_node_ptrs;
    std::vector<std::string> user_node_names;
    [&]<typename... Ns>(tmp::TypeList<Ns...>) {
        ([&]<typename NodeT>() {
            if constexpr (!requires { NodeT::is_fanout_node; }) {
                if (auto* p = find_node.template operator()<NodeT>()) {
                    auto* n = static_cast<INode*>(p);
                    if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), n)
                            == user_node_ptrs.end()) {
                        auto nm = node_display_name<NodeT>();
                        user_node_ptrs.push_back(n);
                        user_node_names.push_back(nm);
                        n->set_name(nm);
                    }
                }
            }
        }.template operator()<Ns>(), ...);
    }(typename Topo::topo{});
    // Pre-pass: build fanout_id → source display name map so fanout nodes
    // can be named after the node feeding them (e.g. "capture_fanout").
    std::map<std::size_t, std::string> fanout_src_name;
@@ -460,7 +397,6 @@ auto make_network(Edges&&... edges) {
    };
    std::vector<std::unique_ptr<IChannelProbe>> channel_probes;
    std::vector<std::string>                    channel_src_names;
    auto wire_one = [&]<typename SE>(SE) {
        using SrcNode = typename SE::src_node_t;
@@ -478,7 +414,6 @@ auto make_network(Edges&&... edges) {
                                + " \xe2\x86\x92 "  // UTF-8 →
                                + node_name.template operator()<DstNode>() + ":" + std::to_string(DstIdx);
            channel_probes.push_back(std::make_unique<ChannelProbe<out_t>>(ch, ch_name));
            channel_src_names.push_back(node_name.template operator()<SrcNode>());
        }
    };
@@ -506,8 +441,7 @@ auto make_network(Edges&&... edges) {
               std::move(fanout_ptrs),
               std::move(user_node_names),
               std::move(fanout_node_names),
-               std::move(channel_probes),
+               std::move(channel_probes));
               std::move(channel_src_names));
 }
 } // namespace kpn
@@ -1,101 +0,0 @@
 #pragma once
 #include <atomic>
 namespace kpn {
 // ── SubmitGate ────────────────────────────────────────────────────────────────
 //
 // Decides, for one node, whether a wake must turn into a scheduler submission.
 // Exactly one firing of a node may be in flight at a time, and a wake that
 // arrives while one is already in flight must not be lost — it has to be
 // honoured when that firing finishes, or the node sleeps holding work.
 //
 // 9c5ce5f wrote this as two independent atomics: queued_ said a firing was in
 // flight, wake_pending_ recorded a wake that arrived during one. That cannot be
 // made correct, because the release side has to read and write both, and a wake
 // can land between the two operations:
 //
 //   producer (try_submit)              worker (release_and_recheck)
 //   ------------------------           ----------------------------
 //   CAS reads queued_ == true, fails
 //                                      queued_.store(false)
 //                                      wake_pending_.exchange(false) -> false
 //   wake_pending_.store(true)
 //
 // End state: queued_ false, wake_pending_ true, nothing running and nothing
 // scheduled. The node sleeps with a wake outstanding, which is precisely the
 // invariant that commit set out to establish. It is not a memory-ordering
 // subtlety — the interleaving above holds under seq_cst.
 //
 // It survived because every caller happened to follow release_and_recheck()
 // with a level re-check (on_input_ready(), or outputs_have_space() on the
 // parked path), which rediscovers the state a lost wake would have signalled.
 // That is a property of the call sites, not of the mechanism, and any new early
 // return that forgets the re-check turns it back into a hang.
 //
 // One atomic with three states makes the race unrepresentable: "idle" and "wake
 // outstanding" are the same variable, so no interleaving can produce both.
 //
 //   Idle       nothing in flight
 //   Queued     a firing is in flight or queued; no wake since it was claimed
 //   QueuedWake a firing is in flight or queued, and a wake arrived meanwhile
 //
 class SubmitGate {
 public:
    /// Register a wake. Returns true when the caller must submit the node;
    /// false when a firing is already in flight and the wake has been recorded
    /// against it instead.
    bool claim() noexcept {
        int cur = state_.load(std::memory_order_acquire);
        for (;;) {
            if (cur == kIdle) {
                if (state_.compare_exchange_weak(cur, kQueued,
                        std::memory_order_acq_rel, std::memory_order_acquire))
                    return true;
            } else if (cur == kQueued) {
                if (state_.compare_exchange_weak(cur, kQueuedWake,
                        std::memory_order_acq_rel, std::memory_order_acquire))
                    return false;
            } else {
                return false;            // a wake is already recorded
            }
        }
    }
    /// End the in-flight firing. Returns true when a wake arrived during it and
    /// the caller must submit again — in which case the gate stays claimed, so
    /// the node is handed straight from one firing to the next and is never
    /// momentarily idle with work outstanding. Returns false when the node is
    /// now idle.
    bool release() noexcept {
        int cur = state_.load(std::memory_order_acquire);
        for (;;) {
            if (cur == kQueuedWake) {
                if (state_.compare_exchange_weak(cur, kQueued,
                        std::memory_order_acq_rel, std::memory_order_acquire))
                    return true;
            } else {
                // kQueued, or kIdle if a stop already forced the gate down.
                if (state_.compare_exchange_weak(cur, kIdle,
                        std::memory_order_acq_rel, std::memory_order_acquire))
                    return false;
            }
        }
    }
    /// Drop the claim and any recorded wake. For stop paths only: honouring a
    /// wake there would resubmit a dead node.
    void force_idle() noexcept { state_.store(kIdle, std::memory_order_release); }
    bool queued()       const noexcept { return state_.load(std::memory_order_relaxed) != kIdle; }
    bool wake_pending() const noexcept { return state_.load(std::memory_order_relaxed) == kQueuedWake; }
 private:
    static constexpr int kIdle       = 0;
    static constexpr int kQueued     = 1;
    static constexpr int kQueuedWake = 2;
    std::atomic<int> state_{kIdle};
 };
 } // namespace kpn
@@ -97,41 +97,4 @@ struct repeat_tuple<T, N, std::index_sequence<Is...>> {
 template<typename T, std::size_t N>
 using repeat_tuple_t = typename repeat_tuple<T, N>::type;
 // ── Sentinel detection ────────────────────────────────────────────────────────
 // A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
 // carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
 // source Frame) or nested one level under a `.source` member (`v.source.eof`,
 // as on message types that wrap the originating Frame). Sentinels are delivered
 // losslessly and non-blockingly via Channel::push_sentinel() instead of the
 // throwing push(), so backpressure can never drop the token that unblocks
 // downstream teardown.
 //
 // Types with neither shape are never treated as sentinels — both traits are
 // SFINAE-safe and the runtime check compiles away to `false` for them, so this
 // stays a no-op for pipelines that don't use an eof convention.
 //
 // Lives here rather than in pool_node.hpp because every node type that forwards
 // values needs it, not just the pool-scheduled ones. FilterNode and RouterNode
 // not having it is what let an EOF token be dropped on a full output.
 template<typename T, typename = void>
 struct has_eof_field : std::false_type {};
 template<typename T>
 struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
    : std::true_type {};
 template<typename T, typename = void>
 struct has_source_eof_field : std::false_type {};
 template<typename T>
 struct has_source_eof_field<T,
    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
    : std::true_type {};
 template<typename T>
 constexpr bool is_sentinel_value(const T& v) {
    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
    else                                               return false;
 }
 } // namespace kpn
@@ -162,7 +162,6 @@ public:
    // ── INode ─────────────────────────────────────────────────────────────────
    void prepare() override { node_.prepare(); }
    void start() override { node_.start(); }
    void stop()  override { node_.stop(); }
    bool running() const override { return node_.running(); }
@@ -65,11 +65,6 @@ static std::string to_json(const std::vector<NodeSnapshot>& nodes,
          << ",\"fps\":"          << n.throughput_fps
          << ",\"total_cpu_ms\":" << n.total_cpu_ms
          << ",\"cpu_util_pct\":" << n.cpu_util_pct
          // Scheduling state — lets a WEDGED pipeline be interrogated over HTTP
          // without a debugger, which matters because the lost-wake bug does not
          // reproduce under one. See NodeSnapshot for how to read the pair.
          << ",\"queued\":"       << (n.queued ? "true" : "false")
          << ",\"wake_pending\":" << (n.wake_pending ? "true" : "false")
          << "}";
    }
    o << "],\"edges\":[";
@@ -1,122 +0,0 @@
 #!/usr/bin/env python3
 """Check the PERF_PLAN Phase-0 acceptance criterion.
 Runs bench_pipeline several times and reports, per row, how far the passes
 spread around their median. The plan's gate is: the same configuration run 7x
 lands within +/-5% on every row. Until that holds, no measured difference
 between KPN and TBB is worth acting on.
 Exits non-zero if any row exceeds the tolerance, so it can gate a session of
 performance work rather than merely inform one.
 Usage:
    scripts/bench_repro_check.py ./build_bench/benchmarks/bench_pipeline \\
        --passes 7 --tolerance 5 -- --work=10 --topos=chain,wide --reps=5
 """
 import argparse
 import statistics
 import subprocess
 import sys
 KEY_COLS = ("topology", "size", "work_us", "threads")
 # bench_pipeline reports throughput, bench_dispatch reports per-dispatch cost.
 # Either is a valid thing to demand reproducibility of; deviation from the
 # median is symmetric, so it does not matter which direction is "better".
 METRIC_COLS = ("items_per_sec", "ns_per_dispatch")
 def parse_csv(text, metric=None):
    """Return ({(topology, size, work_us, threads): value}, metric_name)."""
    rows = {}
    header = None
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        fields = line.split(",")
        if header is None:
            if fields[0] != "topology":
                continue
            header = fields
            if metric is None:
                for cand in METRIC_COLS:
                    if cand in header:
                        metric = cand
                        break
                else:
                    sys.exit(f"no metric column found in header: {header}")
            elif metric not in header:
                sys.exit(f"metric {metric!r} not in header: {header}")
            continue
        rec = dict(zip(header, fields))
        try:
            key = tuple(rec[c] for c in KEY_COLS)
            rows[key] = float(rec[metric])
        except (KeyError, ValueError):
            continue
    return rows, metric
 def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("binary", help="path to bench_pipeline")
    ap.add_argument("--passes", type=int, default=7)
    ap.add_argument("--tolerance", type=float, default=5.0,
                    help="max allowed deviation from the median, percent")
    ap.add_argument("--metric", default=None, choices=METRIC_COLS,
                    help="column to check (default: whichever the CSV carries)")
    # Everything after a standalone `--` goes to bench_pipeline verbatim.
    # argparse.REMAINDER would swallow this script's own flags instead.
    argv = sys.argv[1:]
    extra = []
    if "--" in argv:
        cut = argv.index("--")
        argv, extra = argv[:cut], argv[cut + 1:]
    args = ap.parse_args(argv)
    passes = []
    metric = args.metric
    for i in range(args.passes):
        print(f"pass {i + 1}/{args.passes} ...", file=sys.stderr, flush=True)
        proc = subprocess.run([args.binary] + extra,
                              capture_output=True, text=True)
        if proc.returncode != 0:
            print(proc.stderr, file=sys.stderr)
            sys.exit(f"{args.binary} failed with {proc.returncode}")
        rows, metric = parse_csv(proc.stdout, metric)
        passes.append(rows)
    keys = set(passes[0])
    for p in passes[1:]:
        keys &= set(p)
    if not keys:
        sys.exit("no rows common to every pass")
    print(f"\n{'row':<34} {'median ' + metric:>20} {'worst dev':>10}  verdict")
    print("-" * 72)
    failures = 0
    for key in sorted(keys):
        values = [p[key] for p in passes]
        med = statistics.median(values)
        worst = max(abs(v - med) / med * 100 for v in values) if med else 0.0
        ok = worst <= args.tolerance
        failures += not ok
        label = "{}-{} w={} s={}".format(*key)
        print(f"{label:<34} {med:>20.1f} {worst:>9.1f}%  {'ok' if ok else 'NOISY'}")
    print("-" * 72)
    if failures:
        print(f"{failures}/{len(keys)} rows exceed +/-{args.tolerance:g}% — "
              f"the Phase-0 gate is not met.")
        return 1
    print(f"all {len(keys)} rows within +/-{args.tolerance:g}% "
          f"over {args.passes} passes — Phase-0 gate met.")
    return 0
 if __name__ == "__main__":
    sys.exit(main())
@@ -34,9 +34,7 @@ add_executable(kpn_tests
    test_static_network.cpp
    test_shared_resource.cpp
    test_pool_node.cpp
    test_backpressure_deadlock.cpp
    test_scheduler.cpp
    test_submit_gate.cpp
 )
 target_link_libraries(kpn_tests PRIVATE
@@ -52,31 +50,12 @@ target_link_libraries(kpn_tests PRIVATE
 add_executable(kpn_tests_stress test_channel_stress.cpp)
 target_link_libraries(kpn_tests_stress PRIVATE kpn Catch2::Catch2WithMain)
 # ── Wedge soak (PERF_PLAN G1) ─────────────────────────────────────────────────
 # Long-running end-to-end loop over the configurations that historically wedged.
 # Always built, so it cannot rot, but its CTest cases are registered only under
 # -DKPN_ENABLE_SOAK_TESTS=ON: they run for minutes and would otherwise dominate
 # every `ctest` invocation. Performance work runs it before and after a change:
 #
 #   cmake -B build -DKPN_ENABLE_SOAK_TESTS=ON -DKPN_SOAK_ITERS=50000
 #   cmake --build build --target kpn_soak_wedge
 #   ctest --test-dir build -L soak
 #
 # The binary self-diagnoses: an iteration that stops making progress trips a
 # watchdog that aborts naming the iteration and phase, rather than hanging.
 add_executable(kpn_soak_wedge soak_wedge.cpp)
 target_link_libraries(kpn_soak_wedge PRIVATE kpn)
 target_compile_options(kpn_soak_wedge PRIVATE -O2)
 option(KPN_ENABLE_SOAK_TESTS "Register the wedge soak cases with CTest" OFF)
 set(KPN_SOAK_ITERS 5000 CACHE STRING "Iterations per wedge soak case")
 # ── Sanitizer flags ───────────────────────────────────────────────────────────
 # kpn_sanitizer_flags() is defined in the top-level CMakeLists and is a no-op
 # unless -DKPN_SANITIZER=... is set. Sanitizer must be on both compile and link.
 kpn_sanitizer_flags(_kpn_san)
 if(_kpn_san)
-    foreach(_t kpn_tests kpn_tests_stress kpn_soak_wedge)
+    foreach(_t kpn_tests kpn_tests_stress)
        target_compile_options(${_t} PRIVATE ${_kpn_san})
        target_link_options(${_t}    PRIVATE ${_kpn_san})
    endforeach()
@@ -96,16 +75,3 @@ catch_discover_tests(kpn_tests DISCOVERY_MODE PRE_TEST)
 # Register the stress suite under its own label so CI can run / time it
 # separately from the fast unit tests.
 catch_discover_tests(kpn_tests_stress DISCOVERY_MODE PRE_TEST PROPERTIES LABELS "stress")
 if(KPN_ENABLE_SOAK_TESTS)
    # pool: the configuration the August wedges were reproduced on.
    add_test(NAME soak.wedge.pool
             COMMAND kpn_soak_wedge --mode=pool --depth=4 --threads=4
                     --items=1000 --work-us=10 --iters=${KPN_SOAK_ITERS})
    # private: one pool per node — the model workstream A would change.
    add_test(NAME soak.wedge.private
             COMMAND kpn_soak_wedge --mode=priv --depth=8
                     --items=1000 --work-us=10 --iters=${KPN_SOAK_ITERS})
    set_tests_properties(soak.wedge.pool soak.wedge.private PROPERTIES
                         LABELS "soak" TIMEOUT 3600)
 endif()
@@ -1,236 +0,0 @@
 // Wedge soak test (PERF_PLAN G1).
 //
 // Runs a pipeline configuration end-to-end in a loop and fails if any single
 // iteration stops making progress. Its purpose is to keep performance work
 // from silently reintroducing one of the wedges fixed in August 2026 — the
 // scheduler and channel wake paths are where both perf workstreams operate.
 //
 // Originally the minimal reproducer for the shared-pool chain wedge at
 // (chain, depth=4, work_us=10, pool_threads=4); the pre-6802328 code wedged
 // 5/5 within 45 s, at iterations 149, 1249, 332, 1740 and 493.
 //
 // A wedge is a hang, so a plain loop would hang CTest until its timeout with
 // no indication of where. The watchdog turns that into a failure naming the
 // iteration and the phase it stalled in.
 //
 // Usage: ./kpn_soak_wedge [options]
 //   --iters=5000        iterations to run
 //   --mode=pool|priv    shared ThreadPool(--threads), or one private pool/node
 //   --depth=4           chain depth
 //   --threads=4         shared pool size (--mode=pool only)
 //   --items=1000        items pushed per iteration
 //   --work-us=10        busy-work per node
 //   --watchdog-sec=30   per-iteration progress deadline
 #include <kpn/kpn.hpp>
 #include <atomic>
 #include <chrono>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <memory>
 #include <string>
 #include <thread>
 #include <vector>
 #if defined(__linux__)
 #include <sys/prctl.h>
 #endif
 using namespace kpn;
 using sclock = std::chrono::steady_clock;
 static std::atomic<int> g_work_us{10};
 static int chain_fn(int x) {
    int us = g_work_us.load(std::memory_order_relaxed);
    if (us > 0) {
        auto end = sclock::now() + std::chrono::microseconds(us);
        while (sclock::now() < end);
    }
    return x;
 }
 using ChainNode     = Node<chain_fn, in<>, out<>>;
 using PoolChainNode = PoolNode<chain_fn, in<>, out<>>;
 static void push_retry(Channel<int>& ch, int val) {
    while (true) {
        try   { ch.push(val); return; }
        catch (const ChannelOverflowError&) { std::this_thread::yield(); }
        catch (const ChannelClosedError&)   { return; }
    }
 }
 // ── watchdog ──────────────────────────────────────────────────────────────────
 //
 // The worker bumps g_progress at every phase boundary. The watchdog aborts if
 // it stops moving, so a wedge is reported as a failure at a known iteration
 // rather than as an unattributable CTest timeout.
 static std::atomic<unsigned long> g_progress{0};
 static std::atomic<int>           g_iter{0};
 static std::atomic<const char*>   g_phase{"init"};
 static std::atomic<bool>          g_done{false};
 static void mark(const char* phase) {
    g_phase.store(phase, std::memory_order_relaxed);
    g_progress.fetch_add(1, std::memory_order_release);
 }
 static void watchdog(double deadline_sec) {
    unsigned long last = g_progress.load(std::memory_order_acquire);
    auto last_move = sclock::now();
    while (!g_done.load(std::memory_order_acquire)) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        unsigned long now = g_progress.load(std::memory_order_acquire);
        if (now != last) { last = now; last_move = sclock::now(); continue; }
        double stalled = std::chrono::duration<double>(sclock::now() - last_move).count();
        if (stalled > deadline_sec) {
            std::fprintf(stderr,
                "\nWEDGE: no progress for %.0fs at iteration %d, phase '%s'\n",
                stalled, g_iter.load(std::memory_order_relaxed),
                g_phase.load(std::memory_order_relaxed));
            std::fflush(stderr);
            std::abort();   // core dump / stack trace at the point of the wedge
        }
    }
 }
 // ── one iteration ─────────────────────────────────────────────────────────────
 struct Opts {
    int    iters        = 5000;
    int    depth        = 4;
    int    threads      = 4;
    int    items        = 1000;
    int    work_us      = 10;
    bool   shared_pool  = true;
    double watchdog_sec = 30.0;
 };
 static void one_round_pool(const Opts& o) {
    const std::size_t CAP = static_cast<std::size_t>(o.items);
    auto pool = std::make_shared<ThreadPool>(o.threads);
    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= o.depth; ++i)
        chs.push_back(std::make_shared<Channel<int>>(CAP));
    std::vector<std::unique_ptr<PoolChainNode>> nodes;
    for (int i = 0; i < o.depth; ++i) {
        nodes.push_back(std::make_unique<PoolChainNode>(pool, CAP));
        nodes.back()->set_input_channel<0>(chs[i]);
        nodes.back()->set_output_channel<0>(chs[i + 1].get());
    }
    pool->start();
    for (auto& n : nodes) n->start();
    mark("started");
    std::thread reader([&] {
        for (int i = 0; i < o.items; ++i) chs.back()->pop();
    });
    std::thread pusher([&] {
        for (int i = 0; i < o.items; ++i) push_retry(*chs[0], i);
    });
    pusher.join();  mark("pushed");
    reader.join();  mark("drained");
    for (auto& n : nodes) n->stop();
    mark("nodes stopped");
    pool->stop();
    mark("pool stopped");
 }
 static void one_round_private(const Opts& o) {
    const std::size_t CAP = static_cast<std::size_t>(o.items);
    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= o.depth; ++i)
        chs.push_back(std::make_shared<Channel<int>>(CAP));
    std::vector<std::unique_ptr<ChainNode>> nodes;
    for (int i = 0; i < o.depth; ++i) {
        nodes.push_back(std::make_unique<ChainNode>(CAP));
        nodes.back()->set_input_channel<0>(chs[i]);
        nodes.back()->set_output_channel<0>(chs[i + 1].get());
    }
    for (auto& n : nodes) n->start();
    mark("started");
    std::thread reader([&] {
        for (int i = 0; i < o.items; ++i) chs.back()->pop();
    });
    std::thread pusher([&] {
        for (int i = 0; i < o.items; ++i) push_retry(*chs[0], i);
    });
    pusher.join();  mark("pushed");
    reader.join();  mark("drained");
    for (auto& n : nodes) n->stop();
    mark("nodes stopped");
 }
 // ── main ──────────────────────────────────────────────────────────────────────
 static void usage() {
    std::fprintf(stderr,
        "usage: kpn_soak_wedge [--iters=N] [--mode=pool|priv] [--depth=D]\n"
        "                      [--threads=T] [--items=N] [--work-us=U]\n"
        "                      [--watchdog-sec=S]\n");
 }
 int main(int argc, char** argv) {
 #if defined(__linux__)
    // Allow gdb to attach under ptrace_scope=1 when a wedge is caught.
    prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY, 0, 0, 0);
 #endif
    Opts o;
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);
        if      (key == "--iters")        o.iters        = std::atoi(val.c_str());
        else if (key == "--depth")        o.depth        = std::atoi(val.c_str());
        else if (key == "--threads")      o.threads      = std::atoi(val.c_str());
        else if (key == "--items")        o.items        = std::atoi(val.c_str());
        else if (key == "--work-us")      o.work_us      = std::atoi(val.c_str());
        else if (key == "--watchdog-sec") o.watchdog_sec = std::atof(val.c_str());
        else if (key == "--mode")         o.shared_pool  = (val != "priv");
        else { usage(); return 2; }
    }
    g_work_us.store(o.work_us, std::memory_order_relaxed);
    std::fprintf(stderr,
        "soak: mode=%s depth=%d threads=%d items=%d work_us=%d iters=%d watchdog=%.0fs\n",
        o.shared_pool ? "pool" : "priv", o.depth,
        o.shared_pool ? o.threads : o.depth, o.items, o.work_us,
        o.iters, o.watchdog_sec);
    std::thread wd(watchdog, o.watchdog_sec);
    const auto t0 = sclock::now();
    for (int i = 0; i < o.iters; ++i) {
        g_iter.store(i, std::memory_order_relaxed);
        if (o.shared_pool) one_round_pool(o);
        else               one_round_private(o);
        if ((i + 1) % 100 == 0) {
            std::fprintf(stderr, "\r  %d/%d", i + 1, o.iters);
            std::fflush(stderr);
        }
    }
    g_done.store(true, std::memory_order_release);
    wd.join();
    double secs = std::chrono::duration<double>(sclock::now() - t0).count();
    std::fprintf(stderr, "\ncompleted %d iterations in %.1fs with no wedge\n",
                 o.iters, secs);
    return 0;
 }
@@ -1,421 +0,0 @@
 // Regression: a blocking push must not park a pool worker.
 //
 // Node outputs use push_blocking so a full channel costs time rather than data
 // (a dropped frame does not degrade a downstream result, it silently changes
 // one). But push_blocking sleeps *inside* fire_once, which runs on a pool
 // worker — and nodes are pinned to workers by index. Park enough workers in
 // that retry loop and there is nobody left to run the consumer that would drain
 // the channel, so the whole chain wedges.
 //
 // This is the failure channel.hpp:174 already warns about for sentinels
 // ("a blocking push would park that thread and stop it draining its own input,
 // cascading into a hold-and-wait deadlock under backpressure"). The warning
 // applies to data pushes too.
 //
 // Observed in the field as an intermittent hang: frame_source, camera_pos,
 // face_detector and face_aligner all asleep in push_blocking at once.
 #include <catch2/catch_test_macros.hpp>
 #include <kpn/kpn.hpp>
 #include <atomic>
 #include <chrono>
 #include <thread>
 namespace {
 struct Produce {
    static constexpr std::string_view label() { return "produce"; }
    int n{0};
    int operator()() { return n++; }
 };
 struct Relay {
    static constexpr std::string_view label() { return "relay"; }
    int operator()(int v) { return v; }
 };
 // Deliberately slower than the producer, so the channels between them fill.
 struct SlowSink {
    static constexpr std::string_view label() { return "slow_sink"; }
    std::atomic<int>* seen;
    void operator()(int) {
        std::this_thread::sleep_for(std::chrono::milliseconds(2));
        seen->fetch_add(1, std::memory_order_relaxed);
    }
 };
 }  // namespace
 TEST_CASE("a saturated chain keeps draining", "[backpressure][deadlock]") {
    std::atomic<int> seen{0};
    Produce  p_fn;
    Relay    r1_fn, r2_fn, r3_fn;
    SlowSink s_fn{&seen};
    // Small channels so they saturate immediately, and a chain longer than a
    // modest pool — the shape that starves workers.
    kpn::ObjectNode<Produce,  kpn::in<>,        kpn::out<"a">, "produce",   0> p (p_fn,  2);
    kpn::ObjectNode<Relay,    kpn::in<"a">,     kpn::out<"b">, "relay1",    0> r1(r1_fn, 2);
    kpn::ObjectNode<Relay,    kpn::in<"b">,     kpn::out<"c">, "relay2",    0> r2(r2_fn, 2);
    kpn::ObjectNode<Relay,    kpn::in<"c">,     kpn::out<"d">, "relay3",    0> r3(r3_fn, 2);
    kpn::ObjectNode<SlowSink, kpn::in<"d">,     kpn::out<>,    "slow_sink", 0> s (s_fn,  2);
    auto net = kpn::make_network(
        kpn::edge(p.output<"a">(),  r1.input<"a">()),
        kpn::edge(r1.output<"b">(), r2.input<"b">()),
        kpn::edge(r2.output<"c">(), r3.input<"c">()),
        kpn::edge(r3.output<"d">(), s.input<"d">())
    );
    net.start();
    // The sink is the slowest stage at 2 ms/item, so 40 items is ~80 ms of real
    // work. Anything approaching the timeout means the chain stopped draining
    // rather than merely running slowly.
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(20);
    while (seen.load(std::memory_order_relaxed) < 40 &&
           std::chrono::steady_clock::now() < deadline)
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    const int got = seen.load(std::memory_order_relaxed);
    net.stop();
    INFO("items drained: " << got << " of 40");
    CHECK(got >= 40);
 }
 // Regression: a saturated chain must never stall.
 //
 // push_outputs parks from two places: the retry at the top of fire_once, and
 // the ordinary push after the node function returns. Both release the worker,
 // so both face the same lost wakeup — a space callback firing while queued_ is
 // still up is swallowed by try_submit's CAS. Only the retry path re-checked for
 // space afterwards. The normal path fell through to on_input_ready(), which
 // resubmits only if inputs are ready — and the firing that just parked had
 // consumed its input, so they are not.
 //
 // The strand is permanent under saturation: the node holds its value, its
 // consumer waits for exactly that value, and its producer fills the node's
 // input channel and parks too. Nothing moves again.
 //
 // The test above cannot catch it — 40 items drain before any strand occurs.
 // This one runs the chain saturated and watches for progress to *freeze*, which
 // is the signature of the deadlock. It deliberately does not assert a total:
 // capacity-1 channels are slow, and "slow" must never be reported as "wedged".
 namespace {
 struct FreeRun {
    static constexpr std::string_view label() { return "free_run"; }
    int n{0};
    int operator()() { return n++; }
 };
 struct CountingSink {
    static constexpr std::string_view label() { return "counting_sink"; }
    std::atomic<int>* seen;
    void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); }
 };
 }  // namespace
 TEST_CASE("a saturated chain never stalls", "[backpressure][deadlock]") {
    std::atomic<int> seen{0};
    FreeRun      p_fn;
    Relay        r1_fn, r2_fn;
    CountingSink s_fn{&seen};
    // Capacity 1 everywhere: every push contends, so the park path is taken
    // constantly and the race window is sampled millions of times.
    kpn::ObjectNode<FreeRun,      kpn::in<>,     kpn::out<"a">, "free_run", 0> p (p_fn,  1);
    kpn::ObjectNode<Relay,        kpn::in<"a">,  kpn::out<"b">, "relay1",   0> r1(r1_fn, 1);
    kpn::ObjectNode<Relay,        kpn::in<"b">,  kpn::out<"c">, "relay2",   0> r2(r2_fn, 1);
    kpn::ObjectNode<CountingSink, kpn::in<"c">,  kpn::out<>,    "sink",     0> s (s_fn,  1);
    auto net = kpn::make_network(
        kpn::edge(p.output<"a">(),  r1.input<"a">()),
        kpn::edge(r1.output<"b">(), r2.input<"b">()),
        kpn::edge(r2.output<"c">(), s.input<"c">())
    );
    net.start();
    // A live chain moves thousands of items a second, so 3 s with no movement
    // at all is a wedge, not a slow patch. Sampling for 25 s gives the race
    // ample opportunity: the pipeline hit it roughly twice in 30 runs.
    const auto  giveup    = std::chrono::steady_clock::now() + std::chrono::seconds(25);
    int         last      = 0;
    auto        last_move = std::chrono::steady_clock::now();
    bool        stalled   = false;
    int         stall_at  = 0;
    while (std::chrono::steady_clock::now() < giveup) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        const int now_seen = seen.load(std::memory_order_relaxed);
        if (now_seen != last) {
            last      = now_seen;
            last_move = std::chrono::steady_clock::now();
        } else if (std::chrono::steady_clock::now() - last_move >
                   std::chrono::seconds(3)) {
            stalled  = true;
            stall_at = now_seen;
            break;
        }
    }
    net.stop();
    INFO("chain stalled after " << stall_at << " items");
    CHECK_FALSE(stalled);
    // Guard against the test passing because nothing ever ran.
    CHECK(last > 1000);
 }
 // Regression: a node must not start with a wake already outstanding.
 //
 // 9c5ce5f established the invariant for the running pipeline — a node never
 // sleeps with a wake it dropped. start() broke the same invariant before the
 // pipeline was even running:
 //
 //     enable_inputs(...);          // channel goes live here
 //     stop_flag_.store(false);
 //     queued_.store(false);
 //     register_callbacks(...);     // push callback installed here
 //
 // StaticNetwork starts nodes sources-first, so an upstream node is already
 // firing into this one during that gap. A push landing there is accepted by the
 // ring but wakes nobody: Channel::push invokes push_callback_ only on the
 // empty→non-empty transition, and at that instant the callback is null. Every
 // later push sees a non-empty ring and stays silent. The node is never
 // submitted, and since a sink has no outputs there is no space callback to
 // rescue it either.
 //
 // The signature is distinctive: **zero** items delivered, not a stall partway.
 // The chain reads as wedged from the first frame. Under `ctest -j4` on a loaded
 // machine it reproduced 7 times in 24, and never once in 10 unloaded runs —
 // contention widens the window between those two statements. That is almost
 // certainly the "rare hang, ~1 run in 20 at a 300 s timeout" 28e0667 recorded as
 // known-incomplete.
 //
 // This test needs no contention: it constructs the state the race leaves behind
 // directly, by enabling the input and pushing before start() is ever called.
 namespace {
 int passthrough(int x) { return x; }
 }  // namespace
 TEST_CASE("a node started with data already queued still fires",
          "[backpressure][startup]") {
    auto pool = std::make_shared<kpn::ThreadPool>(2);
    pool->start();
    auto node = kpn::make_pool_node<passthrough>(pool, 8);
    kpn::Channel<int> out_ch(8);
    node.set_output_channel<0>(&out_ch);
    // The missed edge: the channel is live and already holds a value, but no
    // callback was installed when it arrived, so the wake has been and gone.
    node.input_channel<0>().enable();
    node.input_channel<0>().push(21);
    node.start();
    // Bounded wait — a plain pop() would hang rather than fail on a regression.
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    while (out_ch.size() == 0 && std::chrono::steady_clock::now() < deadline)
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
    const bool delivered = out_ch.size() > 0;
    const int  got       = delivered ? out_ch.pop() : -1;
    node.stop();
    pool->stop();
    INFO("value delivered: " << got);
    REQUIRE(delivered);
    CHECK(got == 21);
 }
 // Regression: a fanout absorbs an unequal pair of consumers by slowing, not by
 // dropping.
 //
 // 6595e6e made node outputs lossless and 28e0667 stopped them parking a worker,
 // but FanoutNode was in neither: it kept `catch (ChannelOverflowError&) {}` per
 // output, so whichever branch fell behind lost items — silently, and by an
 // amount that depended on timing. Two runs of the same input could therefore
 // disagree, which is fatal for a fixture the rest of the suite is scored
 // against.
 //
 // The two assertions are the two halves of the requirement:
 //   - no gaps: the slow branch receives *every* item, not most of them;
 //   - bounded lead: the fast branch is throttled to the slow one rather than
 //     racing ahead over a drain that is quietly discarding the difference.
 //
 // Either alone would pass on a broken implementation. A fanout that pushed only
 // to the slow branch has no gaps; one that dropped everything for the slow
 // branch keeps a bounded lead by never letting it fall behind.
 namespace {
 // Records the sequence it sees, so a dropped item shows up as a gap rather than
 // merely as a smaller total.
 struct SeqCheck {
    std::atomic<int>*  next_expected;
    std::atomic<bool>* saw_gap;
    int                delay_us{0};
    void record(int v) const {
        if (delay_us)
            std::this_thread::sleep_for(std::chrono::microseconds(delay_us));
        const int want = next_expected->load(std::memory_order_relaxed);
        if (v != want) saw_gap->store(true, std::memory_order_relaxed);
        else           next_expected->store(want + 1, std::memory_order_relaxed);
    }
 };
 struct FastBranch : SeqCheck {
    static constexpr std::string_view label() { return "fast_branch"; }
    void operator()(int v) { record(v); }
 };
 struct SlowBranch : SeqCheck {
    static constexpr std::string_view label() { return "slow_branch"; }
    void operator()(int v) { record(v); }
 };
 }  // namespace
 TEST_CASE("a fanout absorbs an unequal pair by slowing, not dropping",
          "[backpressure][fanout]") {
    std::atomic<int>  fast_next{0}, slow_next{0};
    std::atomic<bool> fast_gap{false}, slow_gap{false};
    FreeRun    p_fn;
    FastBranch fast_fn{{&fast_next, &fast_gap, 0}};
    SlowBranch slow_fn{{&slow_next, &slow_gap, 500}};   // 0.5 ms/item
    // Small channels so the slow branch saturates in the first few milliseconds
    // and stays saturated for the whole run.
    kpn::ObjectNode<FreeRun,    kpn::in<>,       kpn::out<"v">, "free_run", 0> p (p_fn,    8);
    kpn::ObjectNode<FastBranch, kpn::in<"fast">, kpn::out<>,    "fast",     0> fa(fast_fn, 8);
    kpn::ObjectNode<SlowBranch, kpn::in<"slow">, kpn::out<>,    "slow",     0> sl(slow_fn, 8);
    // Two edges from one output port: make_network auto-inserts FanoutNode<int,2>.
    auto net = kpn::make_network(
        kpn::edge(p.output<"v">(), fa.input<"fast">()),
        kpn::edge(p.output<"v">(), sl.input<"slow">())
    );
    net.start();
    std::this_thread::sleep_for(std::chrono::seconds(1));
    net.stop();
    const int fast_seen = fast_next.load(std::memory_order_relaxed);
    const int slow_seen = slow_next.load(std::memory_order_relaxed);
    INFO("fast branch " << fast_seen << " items, slow branch " << slow_seen);
    CHECK_FALSE(fast_gap.load(std::memory_order_relaxed));
    CHECK_FALSE(slow_gap.load(std::memory_order_relaxed));
    // Guard against passing because nothing ran: 1 s at 0.5 ms/item is ~2000.
    CHECK(slow_seen > 200);
    // The lead is bounded by the buffering between the two — the fanout's own
    // input, the two output channels, and one item in each node's hand. A
    // dropping fanout has no such bound: the fast branch runs at full speed and
    // the difference is the loss.
    CHECK(fast_seen - slow_seen < 200);
 }
 // Regression: a filter must not drop an EOF sentinel into a full output.
 //
 // RouterNode and FilterNode were the last nodes on a data path still using the
 // throwing push() and swallowing the result:
 //
 //     try { out_ch_->push(val); } catch (const ChannelOverflowError&) {}
 //
 // 6595e6e made node outputs lossless, 28e0667 stopped them parking a worker,
 // a8cfe73 did the same for FanoutNode. These two were in none of them.
 //
 // For ordinary values that is the familiar silent-loss problem. For a sentinel
 // it is a hang. EOF is what tells every downstream node to shut down, and
 // nothing comes after it to retry — so a filter that passes EOF by predicate
 // but drops it by backpressure produces a pipeline that never terminates. The
 // scene-actor-extraction decimator is exactly this shape: `if (f.eof) return
 // true;` in the predicate, feeding a chain whose slowest node is an ONNX
 // embedder, so the output is reliably full at the moment EOF arrives.
 //
 // The test forces that state rather than racing for it: the sink is slow enough
 // that the filter's output channel is saturated for the whole run, so EOF meets
 // a full ring with certainty.
 //
 // Both assertions are needed. `saw_eof` alone would pass on an implementation
 // that dropped every ordinary value and delivered only the sentinel; `count`
 // alone would pass on the broken one, which delivers plenty of values and loses
 // only the token that matters.
 namespace {
 struct EofFrame {
    int  seq{0};
    bool eof{false};
 };
 // EOF is emitted exactly once, as a real source does. Everything after it is a
 // filler frame the predicate rejects, which keeps the node alive without
 // re-offering the sentinel — a source that retried EOF would mask the bug,
 // since a later attempt could find the channel drained.
 struct EofSource {
    static constexpr std::string_view label() { return "eof_source"; }
    int  n{0};
    int  total{0};
    EofFrame operator()() {
        if (n > total) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            return {-1, false};          // filler: dropped by the predicate
        }
        EofFrame f{n, n == total};
        ++n;
        return f;
    }
 };
 struct EofSink {
    static constexpr std::string_view label() { return "eof_sink"; }
    std::atomic<int>*  count;
    std::atomic<bool>* saw_eof;
    void operator()(EofFrame f) {
        std::this_thread::sleep_for(std::chrono::microseconds(200));
        if (f.eof) saw_eof->store(true, std::memory_order_release);
        else       count->fetch_add(1, std::memory_order_relaxed);
    }
 };
 }  // namespace
 TEST_CASE("a filter delivers EOF into a saturated output", "[backpressure][filter]") {
    std::atomic<int>  count{0};
    std::atomic<bool> saw_eof{false};
    EofSource src_fn{0, 40};
    EofSink   sink_fn{&count, &saw_eof};
    // Every real frame passes the predicate, so the only thing between source
    // and sink is backpressure. Small channels keep the output saturated.
    auto filt = kpn::make_filter<EofFrame>(
        [](const EofFrame& f) { return f.seq >= 0; }, 4);
    kpn::ObjectNode<EofSource, kpn::in<>,      kpn::out<"f">, "eof_source", 0> s(src_fn,  4);
    kpn::ObjectNode<EofSink,   kpn::in<"f">,   kpn::out<>,    "eof_sink",   0> k(sink_fn, 4);
    auto net = kpn::make_network(
        kpn::edge(s.output<"f">(),      filt.input<0>()),
        kpn::edge(filt.output<0>(),     k.input<"f">())
    );
    net.start();
    // Generous relative to 41 frames at 200 us, and this is a liveness test:
    // the broken implementation never sets saw_eof no matter how long it runs.
    for (int i = 0; i < 200 && !saw_eof.load(std::memory_order_acquire); ++i)
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    net.stop();
    INFO("values delivered: " << count.load() << " of 40");
    CHECK(saw_eof.load(std::memory_order_acquire));
    CHECK(count.load(std::memory_order_relaxed) == 40);
 }
@@ -1,4 +1,3 @@
 #include <string>
 #include <catch2/catch_test_macros.hpp>
 #include <catch2/catch_approx.hpp>
 #include <kpn/channel.hpp>
@@ -239,98 +238,3 @@ TEST_CASE("try_pop_now delivers a pending sentinel once the ring is empty",
    REQUIRE(out == 99);
    REQUIRE_FALSE(ch.try_pop_now(out));   // nothing left
 }
 // Regression: the sentinel slot holds one token and refuses a second.
 //
 // push_sentinel used to write eof_value_ unconditionally. Offering a second
 // token before the first was taken therefore did two wrong things at once: it
 // lost the first silently — and a lost EOF wedges every downstream pop forever
 // — and it wrote the storage while the consumer could be moving the previous
 // value out of it. For the shared_ptr storage that non-trivial types use, that
 // is a torn refcount, not merely a stale read.
 //
 // Refusing is correct rather than queueing: two control tokens on one channel
 // means the stream ended twice, which is a caller protocol error. Coalescing
 // them would hide it, and there is no second value that could sensibly follow
 // the end of a stream.
 TEST_CASE("a second sentinel is refused, not swallowed", "[channel][sentinel]") {
    Channel<int> ch(4);
    REQUIRE(ch.push_sentinel(1));
    // Slot occupied: the first token is still undelivered.
    REQUIRE_FALSE(ch.push_sentinel(2));
    // The first survives intact — the overwrite is what used to lose it.
    int out = 0;
    REQUIRE(ch.try_pop_now(out));
    CHECK(out == 1);
    // And the slot is reusable once drained.
    REQUIRE(ch.push_sentinel(3));
    REQUIRE(ch.try_pop_now(out));
    CHECK(out == 3);
 }
 TEST_CASE("a refused sentinel is not counted as a drop", "[channel][sentinel]") {
    // A refusal means a token arrived while an equivalent one was already
    // pending — not that anything was lost. Counting it as a drop was wrong in
    // a way that showed up immediately on real content: a source at the end of
    // its input keeps being polled and keeps returning EOF, so the token is
    // re-offered on every firing, and the pipeline reported hundreds of dropped
    // frames on a clean run and exited non-zero.
    //
    // The delivery guarantee is unaffected: the first token is pending and will
    // arrive. Only the accounting changed.
    Channel<int> ch(4);
    REQUIRE(ch.push_sentinel(1));
    const auto before = ch.stats().drops.load();
    REQUIRE_FALSE(ch.push_sentinel(2));
    CHECK(ch.stats().drops.load() == before);
    // And the one that was accepted is still the one delivered.
    int out = 0;
    REQUIRE(ch.try_pop_now(out));
    CHECK(out == 1);
 }
 TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][sentinel]") {
    // The non-consuming form exists so a refused token is still the caller's to
    // report. The consuming push_sentinel cannot offer that, since the value is
    // already moved into its parameter.
    Channel<std::string> ch(4);
    std::string first = "eof-1", second = "eof-2";
    REQUIRE(ch.try_push_sentinel(first) == Channel<std::string>::SentinelResult::Taken);
    REQUIRE(ch.try_push_sentinel(second) == Channel<std::string>::SentinelResult::SlotBusy);
    CHECK(second == "eof-2");          // not moved from
    ch.disable();
    std::string third = "eof-3";
    CHECK(ch.try_push_sentinel(third) == Channel<std::string>::SentinelResult::Closed);
    CHECK(third == "eof-3");
 }
 // Regression: try_push must distinguish delivered from discarded.
 //
 // It returned bool, and returned *true* for a closed channel — so "the value
 // arrived" and "the value was thrown away because nobody is listening" were the
 // same answer. Every caller was nonetheless correct, because both cases mean
 // "stop trying"; but nothing above the channel could tell the two apart, and a
 // producer counting successful pushes counted discards among them. Only the
 // channel's own drop counter knew, and only if someone read the diagnostics.
 TEST_CASE("try_push distinguishes taken, full and closed", "[channel]") {
    Channel<int> ch(2);
    int v = 1;
    CHECK(ch.try_push(v) == Channel<int>::PushResult::Taken);
    CHECK(ch.try_push(v) == Channel<int>::PushResult::Taken);
    // Ring is full: the value is untouched and the caller keeps it.
    CHECK(ch.try_push(v) == Channel<int>::PushResult::Full);
    CHECK(v == 1);
    ch.disable();
    const auto drops_before = ch.stats().drops.load();
    CHECK(ch.try_push(v) == Channel<int>::PushResult::Closed);
    // Discarded, and recorded as such rather than reported as a delivery.
    CHECK(ch.stats().drops.load() == drops_before + 1);
 }
@@ -19,7 +19,6 @@
 // Channel<T> is SPSC: exactly one producer thread and one consumer thread per
 // channel. Every scenario below honours that contract.
 #include <string>
 #include <catch2/catch_test_macros.hpp>
 #include <atomic>
 #include <chrono>
@@ -153,18 +152,11 @@ TEST_CASE("SPSC: producer racing a disable() never throws and never hangs",
    }
 }
-TEST_CASE("SPSC: push_callback fires for every push, never missed",
+TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition",
          "[channel][stress]") {
-    // Regression: this callback is the *only* thing that wakes a PoolNode, and
+    // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by
-    // it used to fire only on the empty->non-empty edge, computed from a head_
+    // the consumer-side notification path. Run it under contention to make sure
-    // sampled before the item was published. A concurrent pop() could drain the
+    // the was_empty detection isn't torn by a concurrent pop().
    // ring to empty in that window, so neither side saw the other: the item sat
    // in the ring with the consumer idle, and because the trigger was an edge it
    // never recovered. See set_push_callback in channel.hpp.
    //
    // The old version of this test asserted only `1 <= callbacks <= N`, which a
    // *missed* callback satisfies — it named the hazard and could not detect it.
    // One callback per successful push is the contract, so assert exactly that.
    Channel<int> ch(/*capacity=*/4, /*spin_count=*/4);
    std::atomic<int> callbacks{0};
    ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); });
@@ -182,9 +174,10 @@ TEST_CASE("SPSC: push_callback fires for every push, never missed",
    for (int i = 0; i < N; ++i) (void)ch.pop();
    producer.join();
-    // Exactly one callback per successful push. Fewer means a wake was dropped,
+    // At least one transition, at most one per item; mainly we assert the run
-    // which is the bug; more would mean a spurious wake was manufactured.
+    // completed without TSan flagging a race on push_callback_/was_empty.
-    REQUIRE(callbacks.load() == N);
+    REQUIRE(callbacks.load() >= 1);
    REQUIRE(callbacks.load() <= N);
 }
 // Ordering contract of the out-of-band sentinel under contention.
@@ -286,51 +279,3 @@ TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)",
        REQUIRE(ch.approx_size() == 0);
    }
 }
 // Contended: a producer offering sentinels while the consumer takes them.
 //
 // The old push_sentinel wrote eof_value_ with no regard for whether the
 // consumer was reading it, so a second offer racing a take was a data race on
 // the storage — for the shared_ptr form used by non-trivial types, on the
 // refcount. Under TSan the old code reports it; the handshake added alongside
 // this test makes the producer's write conditional on observing the slot free,
 // which is what serialises the two.
 //
 // Payload is a std::string so the storage is the shared_ptr path rather than
 // the trivially-copyable one, and each token carries its own identity so a torn
 // value shows up as a mismatch rather than as a plausible-looking result.
 TEST_CASE("SPSC: offering sentinels concurrently with takes is race-free",
          "[channel][stress][sentinel]") {
    constexpr int kRounds = 20000;
    Channel<std::string> ch(4);
    std::atomic<int> taken{0};
    std::atomic<bool> torn{false};
    std::atomic<bool> done{false};
    std::thread consumer([&] {
        std::string out;
        while (!done.load(std::memory_order_acquire) || ch.approx_size() > 0) {
            if (ch.try_pop_now(out)) {
                if (out.rfind("eof-", 0) != 0) torn.store(true, std::memory_order_relaxed);
                taken.fetch_add(1, std::memory_order_relaxed);
            }
        }
    });
    int accepted = 0;
    for (int i = 0; i < kRounds; ++i) {
        std::string tok = "eof-" + std::to_string(i);
        if (ch.try_push_sentinel(tok) == Channel<std::string>::SentinelResult::Taken)
            ++accepted;
    }
    done.store(true, std::memory_order_release);
    consumer.join();
    INFO("accepted " << accepted << " taken " << taken.load());
    CHECK_FALSE(torn.load(std::memory_order_relaxed));
    // Every accepted token must be delivered: the slot is refused while full,
    // so acceptance and delivery are one-to-one.
    CHECK(taken.load(std::memory_order_relaxed) == accepted);
    CHECK(accepted > 0);
 }
@@ -1,10 +1,6 @@
 #include <catch2/catch_test_macros.hpp>
 #include <kpn/kpn.hpp>
 #include <atomic>
 #include <chrono>
 #include <mutex>
 #include <stdexcept>
 #include <string>
 #include <thread>
 using namespace kpn;
@@ -58,93 +54,3 @@ TEST_CASE("stop disables input channels — producer push is silently dropped",
    in_ch.push(99);
    REQUIRE(in_ch.size() == 0);
 }
 // Regression: Network::set_error_handler must actually deliver the handler.
 //
 // The handler was stored in a member and never read. A node's exception was
 // discarded at the node boundary and the only surviving evidence was a Closed
 // event, which reports that a node stopped but not why — the difference between
 // a diagnosis and a guess. StaticNetwork has always wired this; Network
 // accepted the handler and silently dropped it, which is worse than not
 // offering the setter at all.
 //
 // The type changed with the fix. It was void(name, exception_ptr), which cannot
 // express the keep-running decision the node side needs, so it is now
 // NodeErrorHandler like StaticNetwork's.
 namespace {
 static int throwing_stage(int x) {
    if (x == 42) throw std::runtime_error("boom");
    return x;
 }
 }  // namespace
 TEST_CASE("network error handler receives the node's exception", "[network]") {
    auto src  = kpn::make_node<throwing_stage>(kpn::in<"v">{}, kpn::out<"w">{}, 8);
    kpn::Channel<int> out(8);
    src.set_output_channel<0>(&out);
    kpn::Network net;
    net.add("stage", src).build();
    std::atomic<int>  calls{0};
    std::string       seen_name;
    std::string       seen_what;
    std::mutex        mx;
    net.set_error_handler([&](std::string_view name, std::exception_ptr ep) {
        std::lock_guard lk(mx);
        seen_name = std::string(name);
        try { if (ep) std::rethrow_exception(ep); }
        catch (const std::exception& e) { seen_what = e.what(); }
        calls.fetch_add(1, std::memory_order_relaxed);
        return true;                       // handled: keep the node running
    });
    net.set_watchdog_interval(std::chrono::hours(1));   // keep the report quiet
    net.start();
    src.input_channel<0>().push(42);                    // throws
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    src.input_channel<0>().push(7);                     // must still be running
    const int passed = out.pop();
    net.stop();
    CHECK(calls.load(std::memory_order_relaxed) == 1);
    CHECK(seen_name == "stage");
    CHECK(seen_what == "boom");
    CHECK(passed == 7);
 }
 // Regression: stopping a network must not wait for the watchdog's next tick.
 //
 // The watchdog looped on std::this_thread::sleep_for(watchdog_interval_), and
 // request_stop() cannot wake a sleeping thread — so stop_watchdog()'s join
 // blocked until the current sleep expired. Every teardown paid up to a full
 // interval, three seconds by default, and a caller who set a long one to keep
 // the periodic report quiet got a stop() that looked like a hang. That is how
 // this was found: the error-handler case above set an hour.
 TEST_CASE("stopping a network does not wait for the watchdog interval", "[network]") {
    auto node = kpn::make_node<increment>(kpn::in<"v">{}, kpn::out<"w">{}, 4);
    kpn::Channel<int> out(4);
    node.set_output_channel<0>(&out);
    kpn::Network net;
    net.add("inc", node).build();
    net.set_watchdog_interval(std::chrono::hours(1));
    net.start();
    // Let the watchdog actually reach its wait. Without this the test races it:
    // stop_watchdog() runs before the thread has entered the loop, the token is
    // already set when it does, and it exits without ever waiting — which passes
    // against the bug as well as the fix.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    const auto t0 = std::chrono::steady_clock::now();
    net.stop();
    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - t0).count();
    INFO("stop took " << ms << " ms");
    CHECK(ms < 2000);
 }
@@ -243,13 +243,7 @@ TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") {
 // ── Overflow callback ─────────────────────────────────────────────────────────
-// NOTE: output overflow is no longer reachable on the data path. A node whose
+TEST_CASE("pool node overflow callback fires on full output channel", "[pool_node][overflow]") {
 // output channel is full now PARKS — it keeps the value in a hidden one-slot
 // buffer, releases its scheduler worker, and is re-submitted when the consumer
 // frees a slot. The overflow callback survives for other producers (a direct
 // Channel::push by non-node code still throws), but a pool node cannot trigger
 // it, so these cases assert the stronger property instead: nothing is dropped.
 TEST_CASE("pool node parks instead of overflowing a full output", "[pool_node][overflow]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();
@@ -270,13 +264,10 @@ TEST_CASE("pool node parks instead of overflowing a full output", "[pool_node][o
    node.stop();
    pool->stop();
-    // Parked, not overflowed: the value is still owned by the node.
+    REQUIRE(overflow_count.load() > 0);
    REQUIRE(overflow_count.load() == 0);
    // And it was never handed downstream, so nothing was lost or duplicated.
    REQUIRE(full_ch.size() == 1);
 }
-TEST_CASE("parking is per node, not shared", "[pool_node][overflow]") {
+TEST_CASE("pool node overflow callback is independent per instance", "[pool_node][overflow]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();
@@ -305,10 +296,7 @@ TEST_CASE("parking is per node, not shared", "[pool_node][overflow]") {
    nodeB.stop();
    pool->stop();
-    // Neither overflows now: A parks on its full output, B runs normally. The
+    REQUIRE(a_overflows.load() > 0);
    // point of the case is unchanged — one node's backpressure must not leak
    // into another's callbacks.
    REQUIRE(a_overflows.load() == 0);
    REQUIRE(b_overflows.load() == 0);
 }
@@ -424,9 +412,7 @@ TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]")
    node.stop();
    pool->stop();
-    // Parking replaced overflow on the data path, so the network callback no
+    REQUIRE(net_overflows.load() > 0);
    // longer fires for a pool node's own output. See the note above.
    REQUIRE(net_overflows.load() == 0);
 }
 TEST_CASE("network_closed_callback fires on crash", "[pool_node][network]") {
@@ -473,250 +459,6 @@ TEST_CASE("per-node and network overflow callbacks both fire independently", "[p
    node.stop();
    pool->stop();
-    // Both zero now: the node parks rather than overflowing. The case still
+    REQUIRE(per_node.load() > 0);
-    // guards that the two callbacks are wired independently.
+    REQUIRE(network.load() > 0);
    REQUIRE(per_node.load() == 0);
    REQUIRE(network.load() == 0);
 }
 // Regression: NodeSnapshot's fields must line up with what nodes initialise.
 //
 // The snapshot is an aggregate that every node type fills positionally, and
 // a8cfe73 appended queued/wake_pending/total_exec_ms to it in an order no call
 // site used: each node supplies total_exec_ms as the element straight after
 // queue_wait_ms, but the struct declared the two bools there. So the exec total
 // landed in `queued`, `queued` landed in `wake_pending`, and `wake_pending`
 // landed in total_exec_ms. The compiler said so (-Wnarrowing, bool to double,
 // once per node instantiation) and the build carried on.
 //
 // It matters more than a cosmetic mix-up: these three fields exist to diagnose a
 // wedge, and a wedged pipeline reported total_exec_ms as 0 or 1 and `queued` as
 // "did this node ever run". Reading them would have pointed at the wrong node.
 //
 // Asserted against ema_exec_ms because that field is independently computed and
 // was already correct: a true sum over several frames cannot be below the
 // exponentially-weighted average of the same samples.
 TEST_CASE("node snapshot fields line up with the values nodes supply",
          "[pool_node][diagnostics]") {
    auto pool = std::make_shared<ThreadPool>(1);
    pool->start();
    auto node = make_pool_node<double_it>(pool, 64);
    Channel<int> out(64);
    node.set_output_channel<0>(&out);
    node.start();
    for (int i = 0; i < 8; ++i) node.input_channel<0>().push(i);
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    auto snap = node.node_snapshot("n", 1.0);
    node.stop();
    pool->stop();
    INFO("frames=" << snap.frames_processed
         << " ema=" << snap.ema_exec_ms
         << " total=" << snap.total_exec_ms);
    REQUIRE(snap.frames_processed == 8);
    // The mis-ordered aggregate put wake_pending here, so this was 0.0 or 1.0.
    CHECK(snap.total_exec_ms >= snap.ema_exec_ms);
    // ...and the exec total here, which is non-zero, so `queued` read true for
    // any node that had ever run — including one asleep with nothing to do.
    CHECK_FALSE(snap.queued);
    CHECK_FALSE(snap.wake_pending);
 }
 // Regression: a value parked twice must keep its payload.
 //
 // push_outputs ends with
 //
 //     else pending_ = std::move(result);
 //
 // and the retry path calls it as push_outputs(std::move(*pending_), …), so on
 // that path `result` is the parked tuple itself. The assignment was therefore a
 // self-move-assignment. std::tuple's is elementwise, and libstdc++'s
 // std::vector does not guard against self-move: it swaps its data into a
 // temporary and leaves the vector empty. So the first park was clean (the
 // argument is a local temporary) and the second erased the payload.
 //
 // The value was still delivered, still in order, still counted — just empty.
 // Downstream cannot distinguish that from a frame on which the node genuinely
 // found nothing, which is why it never surfaced as an error: in
 // scene-actor-extraction it reads as "no faces in this frame" and the run
 // completes with a quietly wrong answer.
 //
 // Reaching it needs *two* outputs. With one, the only thing that resubmits a
 // parked node is that output's own space callback, which by definition fires
 // when there is room — so the retry always succeeds and never reassigns. With
 // two, output A draining resubmits the node while output B is still full: the
 // retry skips A (already delivered, tracked in pending_done_) and fails on B,
 // and that is the reassignment that eats B's payload.
 //
 // Driven through raw channels rather than consumer nodes so each step is
 // forced rather than raced: B is pre-filled and stays full for exactly as long
 // as the test wants it to.
 namespace {
 struct TwoPayloads {
    static constexpr std::string_view label() { return "two_payloads"; }
    std::tuple<std::vector<int>, std::vector<int>> operator()() {
        return {std::vector<int>(4, 1), std::vector<int>(4, 2)};
    }
 };
 }  // namespace
 TEST_CASE("a twice-parked value keeps its payload", "[pool_node][backpressure]") {
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();
    TwoPayloads fn;
    auto node = make_pool_node(fn, pool);
    // Both capacity 1. A must be *full* for its pop to signal space at all —
    // Channel fires the space callback only on the full->not-full edge, so a
    // roomy A would never resubmit the node and the retry would never happen.
    Channel<std::vector<int>> out_a(1), out_b(1);
    node.set_output_channel<0>(&out_a);
    node.set_output_channel<1>(&out_b);
    // B is full before the node ever runs, so the very first firing parks.
    out_b.push(std::vector<int>(4, 99));
    node.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    // Draining A resubmits the node while B is still full: this is the retry
    // that reassigned the tuple to itself.
    (void)out_a.pop();
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    // Now let B through and collect what the node had been holding for it.
    (void)out_b.pop();                       // the pre-fill
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    std::vector<int> parked = out_b.pop();   // the value parked across two tries
    node.stop();
    pool->stop();
    INFO("parked payload size " << parked.size());
    CHECK(parked.size() == 4);
    if (parked.size() == 4) CHECK(parked[0] == 2);
 }
 // Regression: a node woken with nothing to read must not stop itself.
 //
 // pop_one reported an empty channel the same way it reported a closed one, by
 // throwing ChannelClosedError, and fire_once treats that as "upstream is
 // finished" and calls self_stop(). self_stop disables the node's own inputs
 // *and* outputs, so one benign empty read does not merely skip a frame — it
 // kills the node and, through the disabled channels, the rest of the pipeline.
 //
 // A node genuinely does get woken with empty inputs: a space callback fires
 // when its output drains, which has nothing to do with input arrival. fire_once
 // guards against it by checking readiness before popping, and that guard is
 // what this test pins. pop_one now also distinguishes the two cases, so if the
 // guard is ever weakened the cost is a wasted firing rather than a dead node.
 //
 // The sequence below reaches the guard deliberately. The output is capacity 1
 // so that draining it signals space at all — Channel fires the space callback
 // only on the full->not-full edge — and by the final pop the input is long
 // since consumed, so the resulting firing has nothing to read.
 namespace {
 struct CountingRelay {
    static constexpr std::string_view label() { return "counting_relay"; }
    std::atomic<int>* calls;
    int operator()(int v) { calls->fetch_add(1, std::memory_order_relaxed); return v; }
 };
 }  // namespace
 TEST_CASE("a node woken with empty inputs does not stop itself", "[pool_node]") {
    std::atomic<int>  calls{0};
    std::atomic<int>  closed{0};
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();
    CountingRelay fn{&calls};
    auto node = make_pool_node(fn, pool, 8);
    Channel<int> out(1);
    node.set_output_channel<0>(&out);
    node.set_closed_callback([&](auto) { closed.fetch_add(1, std::memory_order_relaxed); });
    out.push(99);                     // output full before the node runs
    node.start();
    node.input_channel<0>().push(1);  // fires, cannot deliver, parks
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    REQUIRE(out.pop() == 99);         // space -> retry delivers the parked value
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    REQUIRE(out.pop() == 1);          // space again -> fires with empty inputs
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    // That firing had nothing to read. The node must still be alive.
    CHECK(closed.load(std::memory_order_relaxed) == 0);
    CHECK(node.running());
    // And must still do its job when real input arrives.
    node.input_channel<0>().push(2);
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    CHECK(out.pop() == 2);
    CHECK(calls.load(std::memory_order_relaxed) == 2);
    node.stop();
    pool->stop();
 }
 // Regression: stop() must not return while a firing is still running.
 //
 // stop() set the flag, disabled the inputs and returned, leaving an executing
 // fire_once touching input_channels_, stats_ and pending_ while the caller went
 // on to destroy them. The old comment was explicit that callers wanting the
 // guarantee should call scheduler_->drain() first — but ~PoolNode calls stop(),
 // and a destructor cannot ask its caller to have done that.
 //
 // A node with a private pool survived by accident: Node::stop() calls
 // pool->stop(), which joins the worker. A node sharing a pool, which
 // make_pool_node exists to create, had nothing joining it at all, so its own
 // destructor raced the firing.
 //
 // Asserted through an observable side effect rather than by trying to catch the
 // use-after-free: if stop() returns before the node function has finished, the
 // flag it sets on the way out is still false.
 namespace {
 struct SlowFiring {
    static constexpr std::string_view label() { return "slow_firing"; }
    std::atomic<bool>* entered;
    std::atomic<bool>* finished;
    void operator()(int) {
        entered->store(true, std::memory_order_release);
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        finished->store(true, std::memory_order_release);
    }
 };
 }  // namespace
 TEST_CASE("stop waits for a firing already in flight", "[pool_node]") {
    std::atomic<bool> entered{false}, finished{false};
    auto pool = std::make_shared<ThreadPool>(2);
    pool->start();
    SlowFiring fn{&entered, &finished};
    auto node = make_pool_node(fn, pool, 4);
    node.start();
    node.input_channel<0>().push(1);
    // Stop only once the node is demonstrably inside its function.
    while (!entered.load(std::memory_order_acquire))
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    node.stop();
    CHECK(finished.load(std::memory_order_acquire));
    pool->stop();
 }
@@ -227,116 +227,3 @@ TEST_CASE("work stealing: tasks complete with more threads than initial queue ta
    REQUIRE(counter.load() == 4);
    pool.stop();
 }
 // Regression: submitting to a stopped pool must be a no-op, not a segfault.
 //
 // stop() ends with queues_.clear(), and submit() went straight to
 // queues_[target] with no check — so a submission arriving after stop indexed
 // an empty vector.
 //
 // This is not a contrived teardown ordering; it happens on a normal path. A
 // node's space callback fires from whichever thread drained the channel, and
 // that thread belongs to the *consumer*. Stop the producer first — which a
 // sources-first shutdown does by design — and the consumer keeps draining its
 // backlog, firing the producer's space callback into a pool that has already
 // been torn down. Before this fix the static-network shutdown case crashed
 // about 12 runs in 20.
 //
 // Checking stopped_ without the lock would not be enough: the window between
 // reading the flag and indexing the vector is exactly where clear() runs.
 TEST_CASE("submitting to a stopped pool is refused, not fatal", "[scheduler]") {
    ThreadPool pool(2);
    pool.start();
    pool.stop();
    std::atomic<int> ran{0};
    for (int i = 0; i < 10; ++i)
        pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); });
    CHECK(ran.load(std::memory_order_relaxed) == 0);
    CHECK(pool.rejected() == 10);
 }
 TEST_CASE("submitting while the pool stops does not crash", "[scheduler]") {
    // The racing form of the case above: a producer thread submitting
    // continuously while stop() runs underneath it. Nothing is asserted about
    // how many tasks run — the point is that every submission either enqueues
    // or is refused, and none touches a destroyed queue.
    for (int rep = 0; rep < 20; ++rep) {
        ThreadPool pool(4);
        pool.start();
        std::atomic<bool> go{false};
        std::atomic<int>  ran{0};
        std::thread submitter([&] {
            while (!go.load(std::memory_order_acquire)) {}
            for (int i = 0; i < 2000; ++i)
                pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); });
        });
        go.store(true, std::memory_order_release);
        std::this_thread::sleep_for(std::chrono::microseconds(200));
        pool.stop();
        submitter.join();
        // Everything submitted was either executed or refused; nothing vanished
        // into a queue that no longer existed.
        CHECK(pool.rejected() + pool.snapshot("p").tasks_completed <= 2000);
    }
 }
 // Regression: idle workers must sleep while another worker is busy.
 //
 // The wait predicate was `stopped_ || total_ > 0`, and total_ counts queued
 // *plus executing*. So while any one task ran, every other worker's predicate
 // was true: wait() returned immediately and the worker spun through try_pop,
 // try_steal and back to wait at full speed — try_lock-ing every peer queue on
 // each pass. One slow task pinned every other core and contended the very
 // mutexes the working thread needed to make progress.
 //
 // That is the shape of this pipeline's load exactly: a handful of nodes whose
 // work is tens of milliseconds of ONNX inference. It was latent only because
 // each node currently owns a private single-thread pool, where there is no
 // idle peer to spin. Any use of a shared pool — which make_pool_node exists
 // for — hits it immediately.
 //
 // Measured as CPU time rather than wall time, because the bug does not make
 // anything slower to finish; it makes seven cores burn while one works. A
 // sleeping task consumes no CPU, so with workers correctly asleep the whole
 // pool should account for almost none.
 TEST_CASE("idle workers do not spin while one task runs", "[scheduler]") {
    auto cpu_ms = [] {
        struct timespec ts{};
        clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &ts);
        return ts.tv_sec * 1000.0 + ts.tv_nsec / 1e6;
    };
    constexpr int kThreads = 8;
    constexpr int kWorkMs  = 300;
    ThreadPool pool(kThreads);
    pool.start();
    std::this_thread::sleep_for(20ms);      // let workers reach the wait
    const double before = cpu_ms();
    // One long task plus a trivial one per remaining worker. The trivial ones
    // matter: a worker that has never been woken stays blocked in wait() and
    // never re-evaluates the predicate, so the spin only appears once a worker
    // *finishes* something and re-enters the loop while a peer is still busy.
    // Submitting only the long task does not reproduce it.
    pool.submit([&] { std::this_thread::sleep_for(std::chrono::milliseconds(kWorkMs)); });
    for (int i = 0; i < kThreads - 1; ++i) pool.submit([] {});
    pool.drain();
    const double used = cpu_ms() - before;
    pool.stop();
    // Measured on this tree: 1991 ms of CPU with the total_ predicate against
    // 0.4 ms with queued_, and 19205 voluntary context switches against 10 —
    // roughly (kThreads - 1) cores burned for the duration of one sleeping
    // task. The threshold sits far from both so the case is not sensitive to
    // how loaded the machine is.
    INFO("cpu " << used << " ms over " << kWorkMs << " ms of sleeping work");
    CHECK(used < kWorkMs);
 }
@@ -237,51 +237,3 @@ TEST_CASE("make_shared_resource constructs with forwarded args", "[shared_resour
    auto g = res.acquire();
    REQUIRE(*g == "hello");
 }
 // Regression: a waiter must be releasable, or teardown waits on it forever.
 //
 // acquire() blocks on a condition variable whose predicate only becomes true
 // when release() hands over ownership. There was no timeout and no stop
 // condition, so a node parked there ignored teardown entirely: its worker never
 // returned, the pool's join never completed, and shutdown hung waiting for a
 // resource nobody was going to release — which is exactly the case when the
 // holder is being stopped too.
 //
 // close() turns that into an exception the node's existing error path already
 // handles, and networks now call it on registered resources before stopping any
 // node, for the same reason.
 TEST_CASE("closing a shared resource releases its waiters", "[shared_resource]") {
    SharedResource<int> res(42);
    auto holder = res.acquire();          // resource is now held
    std::atomic<bool> threw{false}, returned{false};
    std::thread waiter([&] {
        try {
            auto g = res.acquire();       // blocks: someone else holds it
            (void)g;
        } catch (const ResourceClosedError&) {
            threw.store(true, std::memory_order_release);
        }
        returned.store(true, std::memory_order_release);
    });
    // Let it park, then tear down without ever releasing the holder.
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    REQUIRE_FALSE(returned.load(std::memory_order_acquire));
    res.close();
    waiter.join();
    CHECK(threw.load(std::memory_order_acquire));
 }
 TEST_CASE("acquiring a closed resource fails immediately", "[shared_resource]") {
    SharedResource<int> res(7);
    res.close();
    CHECK_THROWS_AS(res.acquire(), ResourceClosedError);
    // Reusable across runs once reopened.
    res.reopen();
    CHECK_NOTHROW(res.acquire());
 }
@@ -271,164 +271,3 @@ TEST_CASE("static_network: fanout with labelled same-function consumers", "[stat
    REQUIRE(outB.pop() == 7);
    net.stop();
 }
 // Regression: shutdown() must return even when a consumer stopped consuming.
 //
 // The drain step was an unbounded `while (anything anywhere is non-empty)` poll
 // over *every* channel in the graph. Two defects in one loop: it waited for the
 // whole network to be idle before stopping each successive layer rather than
 // just the node it had stopped — the dynamic Network's version even took a node
 // name and ignored it — and it had no deadline, so anything wedged downstream
 // turned a graceful shutdown into the hang it exists to avoid.
 //
 // It could also fail to terminate with nothing wedged at all. current_fill came
 // from a snapshot that loaded tail_ before head_; a concurrent pop between the
 // two reads yields a head_ past the sampled tail_, and the unsigned difference
 // wraps to ~2^64. Any poll for "is it empty yet" against that value runs
 // forever. Both indices only ever increase, so loading head_ first can at worst
 // under-report a push, which this loop tolerates and a wrap does not.
 //
 // Here the sink never takes anything, so its input cannot drain and the only
 // correct outcome is to give up and say so. The bound asserted is deliberately
 // loose: the point is that it terminates, not how fast.
 namespace {
 struct DrainSource {
    static constexpr std::string_view label() { return "drain_source"; }
    int n{0};
    int operator()() {
        std::this_thread::sleep_for(std::chrono::microseconds(100));
        return n++;
    }
 };
 struct NeverConsumes {
    static constexpr std::string_view label() { return "never_consumes"; }
    std::atomic<bool>* wedged;
    void operator()(int) {
        // Blocks for the duration of the test: the input channel behind it
        // fills and stays full.
        while (!wedged->load(std::memory_order_acquire))
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
 };
 }  // namespace
 TEST_CASE("shutdown returns when a consumer has wedged", "[static_network][shutdown]") {
    std::atomic<bool> release{false};
    DrainSource   src_fn;
    NeverConsumes sink_fn{&release};
    kpn::ObjectNode<DrainSource,   kpn::in<>,    kpn::out<"v">, "drain_source",   0> s(src_fn,  4);
    kpn::ObjectNode<NeverConsumes, kpn::in<"v">, kpn::out<>,    "never_consumes", 0> k(sink_fn, 4);
    auto net = kpn::make_network(kpn::edge(s.output<"v">(), k.input<"v">()));
    net.set_drain_timeout(std::chrono::milliseconds(100));
    net.start();
    // Let the channel fill and the sink jam.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    // Unjam the sink well after the drain timeout should have expired. Stopping
    // a node joins its worker, so a sink blocked forever would hang the test in
    // stop() rather than in the drain loop this case is about.
    std::thread unjam([&] {
        std::this_thread::sleep_for(std::chrono::milliseconds(800));
        release.store(true, std::memory_order_release);
    });
    const auto t0 = std::chrono::steady_clock::now();
    net.shutdown();
    const auto elapsed = std::chrono::steady_clock::now() - t0;
    unjam.join();
    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
    INFO("shutdown took " << ms << " ms");
    CHECK(ms < 3000);   // unbounded before; one 100 ms drain timeout after
 }
 // Regression: node order must come from the topological sort, not from the
 // order the edges happened to be written in.
 //
 // make_network computes Topo for the cycle check and then dropped it, filling
 // the node vector in edge-declaration order — and named it user_nodes_topo_.
 // halt() stops in its reverse, and shutdown() walks it forwards stopping each
 // node and draining its outputs before moving to the next, which is a graceful
 // drain only if the order really is sources-first.
 //
 // Every network in this tree declares edges in pipeline order, so the two
 // coincided and nothing failed. This case declares them backwards, which is
 // legal and which make_network otherwise accepts silently.
 //
 // Asserted through shutdown() rather than by reading the order back, because
 // the order is private and the ordering is not the point — what it buys is.
 // A sources-first shutdown lets the values already in flight reach the sink;
 // stopping the sink first strands them, and the drain step then has nobody
 // left to take them.
 namespace {
 struct OrderSource {
    static constexpr std::string_view label() { return "order_source"; }
    std::atomic<int>* made;
    int operator()() {
        std::this_thread::sleep_for(std::chrono::microseconds(20));
        return made->fetch_add(1, std::memory_order_relaxed);
    }
 };
 // Deliberately slower than the source, so a deep backlog builds up in its input
 // channel. That backlog is what a sources-first shutdown preserves and a
 // sink-first one throws away, and it needs to be big enough that the difference
 // cannot be mistaken for one value in flight.
 struct OrderRelay {
    static constexpr std::string_view label() { return "order_relay"; }
    int operator()(int v) {
        std::this_thread::sleep_for(std::chrono::microseconds(300));
        return v;
    }
 };
 struct OrderSink {
    static constexpr std::string_view label() { return "order_sink"; }
    std::atomic<int>* seen;
    void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); }
 };
 }  // namespace
 TEST_CASE("edges declared out of order still start and stop sources-first",
          "[static_network][shutdown]") {
    std::atomic<int> seen{0}, made{0};
    OrderSource src_fn{&made};
    OrderRelay  relay_fn;
    OrderSink   sink_fn{&seen};
    kpn::ObjectNode<OrderSource, kpn::in<>,    kpn::out<"v">, "order_source", 0> s(src_fn,   8);
    kpn::ObjectNode<OrderRelay,  kpn::in<"v">, kpn::out<"w">, "order_relay",  0> r(relay_fn, 64);
    kpn::ObjectNode<OrderSink,   kpn::in<"w">, kpn::out<>,    "order_sink",   0> k(sink_fn,  64);
    // Sink edge first, source edge last — the reverse of pipeline order.
    auto net = kpn::make_network(
        kpn::edge(r.output<"w">(), k.input<"w">()),
        kpn::edge(s.output<"v">(), r.input<"v">())
    );
    net.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    const int before = seen.load(std::memory_order_relaxed);
    REQUIRE(before > 0);              // the pipeline ran at all
    net.shutdown();
    // Sources stop first and each layer drains before the next stops, so the
    // backlog queued in front of the relay still reaches the sink. Stopping in
    // declaration order stops the relay first and discards all of it.
    const int after = seen.load(std::memory_order_relaxed);
    INFO("made " << made.load() << ", delivered " << before
         << " before shutdown, " << after << " after");
    CHECK(after - before >= 20);
 }
@@ -1,173 +0,0 @@
 // Regression: a node must never end up idle with a wake outstanding.
 //
 // 9c5ce5f established that invariant and implemented it as two independent
 // atomics — queued_ for "a firing is in flight", wake_pending_ for "a wake
 // arrived during one". Two variables cannot express it, because the release
 // side has to read and write both and a wake can land in between:
 //
 //   producer (try_submit)              worker (release_and_recheck)
 //   ------------------------           ----------------------------
 //   CAS reads queued_ == true, fails
 //                                      queued_.store(false)
 //                                      wake_pending_.exchange(false) -> false
 //   wake_pending_.store(true)
 //
 // queued_ false, wake_pending_ true, nothing running and nothing scheduled.
 // Not a memory-ordering subtlety: the interleaving holds under seq_cst.
 //
 // LegacyGate below is that protocol verbatim, with a hook between the failed
 // CAS and the wake_pending_ store so the interleaving can be forced rather than
 // waited for. That makes the loss deterministic and the test non-flaky, and it
 // keeps the defect on record now that the code implementing it is gone.
 //
 // A note on what is NOT tested here, because it would be misleading to imply
 // otherwise: there is no black-box, node-level test that fails before this fix
 // and passes after. Every call site of release_and_recheck() happens to follow
 // it with a level re-check — on_input_ready(), or outputs_have_space() on the
 // parked path — which rediscovers the state a lost wake would have signalled.
 // That masking is a property of the call sites, not of the mechanism, and the
 // point of the fix is that a future early return that forgets the re-check no
 // longer reintroduces a hang. The value is structural, so the tests are
 // structural: the state machine is pinned by contract, and the defect it
 // replaces is pinned by demonstration.
 #include <catch2/catch_test_macros.hpp>
 #include <kpn/submit_gate.hpp>
 #include <atomic>
 #include <functional>
 #include <thread>
 using namespace kpn;
 namespace {
 // The pre-fix protocol, with a seam at the point where the race lives.
 class LegacyGate {
 public:
    std::function<void()> before_recording_wake;
    bool claim() noexcept {
        bool expected = false;
        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
            return true;
        if (before_recording_wake) before_recording_wake();
        wake_pending_.store(true, std::memory_order_release);
        return false;
    }
    bool release() noexcept {
        queued_.store(false, std::memory_order_release);
        if (wake_pending_.exchange(false, std::memory_order_acq_rel)) {
            bool expected = false;
            if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
                return true;
        }
        return false;
    }
    bool queued()       const noexcept { return queued_.load(std::memory_order_relaxed); }
    bool wake_pending() const noexcept { return wake_pending_.load(std::memory_order_relaxed); }
 private:
    std::atomic<bool> queued_{false};
    std::atomic<bool> wake_pending_{false};
 };
 }  // namespace
 TEST_CASE("the two-atomic gate loses a wake, deterministically", "[submit_gate]") {
    LegacyGate gate;
    bool resubmitted = true;
    REQUIRE(gate.claim());              // a firing is now in flight
    // Force the interleaving: the firing completes in the window between the
    // second wake's failed CAS and its record of that wake.
    gate.before_recording_wake = [&] { resubmitted = gate.release(); };
    const bool submitted = gate.claim();
    // The wake was neither submitted by the producer nor honoured by the
    // release. Nothing is scheduled, and nothing else will re-trigger it.
    CHECK_FALSE(submitted);
    CHECK_FALSE(resubmitted);
    CHECK_FALSE(gate.queued());
    CHECK(gate.wake_pending());         // recorded, and never to be consumed
 }
 TEST_CASE("submit gate: a wake during a firing is honoured", "[submit_gate]") {
    SubmitGate gate;
    REQUIRE(gate.claim());              // idle -> queued, caller submits
    REQUIRE(gate.queued());
    REQUIRE_FALSE(gate.wake_pending());
    REQUIRE_FALSE(gate.claim());        // second wake is recorded, not submitted
    REQUIRE(gate.wake_pending());
    REQUIRE(gate.release());            // and honoured when the firing ends
    // The gate stays claimed across the handover, so the node is never
    // momentarily idle while a submission for it is in flight. This is the
    // state the legacy gate could not represent.
    REQUIRE(gate.queued());
    REQUIRE_FALSE(gate.wake_pending());
    REQUIRE_FALSE(gate.release());      // no further wake: now idle
    REQUIRE_FALSE(gate.queued());
 }
 TEST_CASE("submit gate: repeated wakes collapse to one resubmission", "[submit_gate]") {
    // Collapsing is deliberate. A firing consumes one item and its caller then
    // re-checks the input level, so the gate only has to guarantee that at
    // least one more firing follows a wake, not one per wake.
    SubmitGate gate;
    REQUIRE(gate.claim());
    for (int i = 0; i < 10; ++i) REQUIRE_FALSE(gate.claim());
    REQUIRE(gate.release());
    REQUIRE_FALSE(gate.release());
 }
 TEST_CASE("submit gate: force_idle drops a recorded wake", "[submit_gate]") {
    // Stop paths use this deliberately — honouring a wake there would resubmit
    // a node that has already been told to stop.
    SubmitGate gate;
    REQUIRE(gate.claim());
    REQUIRE_FALSE(gate.claim());
    REQUIRE(gate.wake_pending());
    gate.force_idle();
    REQUIRE_FALSE(gate.queued());
    REQUIRE_FALSE(gate.wake_pending());
    REQUIRE(gate.claim());              // and the gate is reusable afterwards
 }
 TEST_CASE("submit gate: concurrent claim and release stay consistent", "[submit_gate]") {
    // Not a lost-wake test — see the header note. This is a TSan target and a
    // check that the CAS loops always terminate and always leave the gate in a
    // reachable state: exactly one party may hold the claim at a time, so the
    // count of claims granted must equal the count of releases that ended idle.
    SubmitGate gate;
    std::atomic<long> granted{0}, ended_idle{0};
    std::atomic<bool> stop{false};
    std::thread waker([&] {
        while (!stop.load(std::memory_order_relaxed))
            if (gate.claim()) granted.fetch_add(1, std::memory_order_relaxed);
    });
    std::thread worker([&] {
        while (!stop.load(std::memory_order_relaxed))
            if (gate.queued() && !gate.release())
                ended_idle.fetch_add(1, std::memory_order_relaxed);
    });
    std::this_thread::sleep_for(std::chrono::milliseconds(200));
    stop.store(true, std::memory_order_relaxed);
    waker.join();
    worker.join();
    // Drain whatever claim is outstanding so the two counts can be compared.
    while (gate.queued())
        if (!gate.release()) ended_idle.fetch_add(1, std::memory_order_relaxed);
    INFO("granted " << granted.load() << " ended idle " << ended_idle.load());
    REQUIRE(granted.load() > 0);
    CHECK(granted.load() == ended_idle.load());
 }