fix: diagnostics.hpp must include <vector>, not hope for it

NetworkSnapshot declares std::vector<NodeSnapshot> and
std::vector<ChannelSnapshot>, and the header has never included <vector>.
It compiles anyway wherever something earlier in the translation unit
drags the definition in, which is every build we have done -- and does
not compile where nothing does.

Found by a consumer, in CI, on Debian 12 / GCC 12: a header that includes
<kpn/diagnostics.hpp> before <vector> gets "'vector' in namespace 'std'
does not name a template type" at the two member declarations, and then
one cascading error for every use of .nodes or .channels downstream. The
reported errors all name the consumer's file, so the bug reads as the
consumer's until you look at the first line of the compiler output.

Nothing else in the header is missing: it uses std::atomic, std::chrono,
std::size_t and std::string, and includes all four.

Checked the rest of the public headers the same way while here, by
compiling each one standalone under GCC 12. Two other results, neither
fixed by this commit:

  network.hpp    names IChannelProbe (channel.hpp:557) without including
                 channel.hpp -- the same class of bug, still latent.
  web_debug.hpp  needs httplib.h, which is an external optional
                 dependency and not a defect.

.gitignore

@@ -25,6 +25,9 @@ venv/
 .DS_Store
 Thumbs.db
 
+# Benchmark output (scripts/bench_repro_check.py --out-dir)
+bench_runs/
+
 # Claude Code local settings
 .claude/settings.local.json
 include/kpn/ort_cache/

include/kpn/diagnostics.hpp

@@ -4,6 +4,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <string>
+#include <vector>
 #include <time.h>   // clock_gettime, CLOCK_THREAD_CPUTIME_ID
 
 namespace kpn {

include/kpn/scheduler.hpp

@@ -128,7 +128,28 @@ public:
             rejected_.fetch_add(1, std::memory_order_relaxed);
             return;
         }
-        std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
+        // B9 — submit-to-self affinity. Round-robin hands every task to a
+        // *different* worker, and on a pool of two or more that worker is
+        // asleep, so each dispatch pays a futex wake: measured 182 ns/dispatch
+        // on a 1-thread pool against 3197 ns on 20 threads, with voluntary
+        // context switches per task rising 0.00 -> 1.19 in step.
+        //
+        // A submission originating on one of *our own* workers goes to that
+        // worker's queue instead. It is about to return to worker_loop and
+        // try_pop its own queue, so the work is already there and nothing
+        // sleeps — the property that makes ThreadPool(1) fast, extended to
+        // any pool size. Imbalance is corrected by the existing try_steal.
+        //
+        // The pool identity check is load-bearing: a worker of pool A
+        // submitting into pool B must not use A's index, which may exceed B's
+        // thread_count_ or alias an unrelated queue. Nested networks do
+        // exactly this.
+        std::size_t target;
+        if (tls_pool == this && tls_worker < thread_count_) {
+            target = tls_worker;
+        } else {
+            target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
+        }
         {
             std::lock_guard lock(queues_[target]->mx);
             queues_[target]->pq.push(
@@ -142,8 +163,23 @@ public:
         // will observe total_ > 0) or already blocked in wait() (and will be
         // woken). Without this, notify_one() can slip into the gap between the
         // worker's predicate check and its wait(), and be lost — a deadlock.
-        { std::lock_guard<std::mutex> lk(cv_mx_); }
-        cv_.notify_one();
+        //
+        // Skipped entirely when no worker is parked. waiters_ is incremented
+        // *before* wait() releases cv_mx_ and decremented after it returns,
+        // both under that mutex, so a worker on its way to sleep is already
+        // counted here. Reading zero therefore means no worker can be in
+        // wait(), and there is nothing a notify could reach — as opposed to
+        // reading zero because we raced one, which the mutex prevents.
+        //
+        // This is the hot path for an already-busy pool: with B9 the work is
+        // in the local queue and the submitting worker will find it itself,
+        // so the lock round-trip and notify were pure overhead. Measured 1.00
+        // voluntary context switches per task before this, on a pool where
+        // only one task is ever in flight.
+        if (waiters_.load(std::memory_order_seq_cst) != 0) {
+            { std::lock_guard<std::mutex> lk(cv_mx_); }
+            cv_.notify_one();
+        }
     }
 
     std::size_t thread_count() const { return thread_count_; }
@@ -193,6 +229,17 @@ private:
 
     std::optional<std::function<void()>> try_steal(std::size_t thief) {
         // Find the most-loaded peer without blocking — racy peek is fine.
+        //
+        // The threshold is >0: a peer holding a single task is a valid victim.
+        //
+        // Raising it to >1 — to stop a thief winning the race for a task its
+        // owner just submitted to itself (B9) — deadlocks. `latency` mode
+        // hangs at 12 and 20 threads: an external submit() round-robins one
+        // task onto an idle worker's queue, and if that worker is parked, no
+        // peer will take it because a queue of one is no longer stealable.
+        // Nothing else is coming to wake it, so the pool sits forever.
+        // Measured before reverting: it also made steady-state *worse*,
+        // 2229 -> 4546 ns at 12 threads.
         std::size_t victim = thief, best = 0;
         for (std::size_t i = 0; i < queues_.size(); ++i) {
             if (i == thief) continue;
@@ -221,15 +268,41 @@ private:
     }
 
     void worker_loop(std::size_t id) {
+        // Identify this thread as one of our workers, for B9's affinity check
+        // in submit(). Restored on exit rather than merely cleared: a pool
+        // whose worker runs a task that itself starts and stops a nested pool
+        // would otherwise come back with its identity erased.
+        ThreadPool* const prev_pool   = tls_pool;
+        const std::size_t prev_worker = tls_worker;
+        tls_pool   = this;
+        tls_worker = id;
+        struct Restore {
+            ThreadPool* p; std::size_t w;
+            ~Restore() { tls_pool = p; tls_worker = w; }
+        } restore{prev_pool, prev_worker};
+
         while (true) {
             if (auto fn = try_pop(*queues_[id]))   { execute(*fn); continue; }
             if (auto fn = try_steal(id))            { execute(*fn); continue; }
 
+            // B5 (bounded spin before parking) was tried here and removed: it
+            // does not pay. Swept at 50/200/1000 rounds on a 12-thread pool,
+            // steady state went 2123 / 2230 / 2574 ns against 2229 ns without
+            // it, and voluntary context switches per task stayed at ~0.97
+            // throughout. The spin cannot catch what it is aimed at, because
+            // a peer is woken the moment queued_ becomes non-zero — which
+            // happens before this worker reaches the spin at all.
             std::unique_lock lock(cv_mx_);
+            // Counted under cv_mx_ and before the predicate is evaluated, so
+            // that a submit() which reads waiters_ == 0 can be certain this
+            // worker is not about to block: to get here we already hold the
+            // mutex that submit() must take to notify.
+            waiters_.fetch_add(1, std::memory_order_seq_cst);
             cv_.wait(lock, [this] {
                 return stopped_.load(std::memory_order_seq_cst)
                     || queued_.load(std::memory_order_relaxed) > 0;
             });
+            waiters_.fetch_sub(1, std::memory_order_seq_cst);
             // 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.
@@ -239,6 +312,13 @@ private:
         }
     }
 
+    /// Which pool, and which of its workers, the calling thread is — or
+    /// nullptr on any thread that is not a pool worker. Read by submit() to
+    /// decide whether a local push is safe (B9). inline so the header stays
+    /// header-only.
+    static inline thread_local ThreadPool* tls_pool   = nullptr;
+    static inline thread_local std::size_t tls_worker = 0;
+
     const std::size_t                          thread_count_;
     std::vector<std::unique_ptr<WorkerQueue>>  queues_;
     std::vector<std::thread>                   workers_;
@@ -267,6 +347,9 @@ private:
     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
+    /// Workers currently inside cv_.wait(), maintained under cv_mx_. Lets
+    /// submit() skip the lock round-trip and notify when nobody is parked.
+    std::atomic<size_t>   waiters_{0};
     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 —

scripts/bench_repro_check.py

@@ -9,12 +9,20 @@ 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.
 
+A full sweep is hours, so the run is observable and restartable rather than
+opaque: rows stream to --out-dir as each pass produces them, and a progress
+bar tracks rows within the pass. Killing the run keeps everything already
+written; --resume picks up from the completed passes on disk.
+
 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 datetime
+import os
+import pathlib
 import statistics
 import subprocess
 import sys
@@ -27,11 +35,47 @@ KEY_COLS = ("topology", "size", "work_us", "threads")
 METRIC_COLS = ("items_per_sec", "ns_per_dispatch")
 
 
-def parse_csv(text, metric=None):
+def _progress(total, desc):
+    """A tqdm bar if tqdm is installed, else a minimal stderr fallback.
+
+    The fallback exists because this script gates a benchmark run; refusing to
+    start over a missing progress dependency would be the wrong trade.
+    """
+    try:
+        from tqdm import tqdm
+    except ImportError:
+        class Fallback:
+            def __init__(self):
+                self.n = 0
+
+            def update(self, k=1):
+                self.n += k
+                end = "\n" if (total and self.n >= total) else "\r"
+                print(f"  {desc}: {self.n}/{total or '?'} rows",
+                      file=sys.stderr, end=end, flush=True)
+
+            def close(self):
+                pass
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *exc):
+                self.close()
+
+        return Fallback()
+
+    return tqdm(total=total, desc=desc, unit="row", leave=False,
+                bar_format="  {desc}: {n_fmt}/{total_fmt} rows "
+                           "|{bar}| {elapsed}<{remaining}",
+                file=sys.stderr)
+
+
+def parse_csv_lines(lines, metric=None):
     """Return ({(topology, size, work_us, threads): value}, metric_name)."""
     rows = {}
     header = None
-    for line in text.splitlines():
+    for line in lines:
         line = line.strip()
         if not line or line.startswith("#"):
             continue
@@ -59,36 +103,64 @@ def parse_csv(text, metric=None):
     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)")
+def parse_csv(text, metric=None):
+    return parse_csv_lines(text.splitlines(), metric)
 
-    # 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)
+def count_rows(binary, extra):
+    """Enumerate the sweep cheaply, so the progress bar has a real total.
 
+    Asks the binary itself rather than reimplementing the sweep in Python,
+    which would silently drift from the C++ defaults. Returns None if the
+    probe fails -- an unknown total degrades the bar, it does not stop the run.
+    """
+    probe = [binary] + extra + ["--reps=0", "--warmup=0"]
+    try:
+        proc = subprocess.run(probe, capture_output=True, text=True,
+                              timeout=600)
+    except (subprocess.SubprocessError, OSError):
+        return None
+    if proc.returncode != 0:
+        return None
+    rows, _ = parse_csv(proc.stdout)
+    return len(rows) or None
+
+
+def run_pass(binary, extra, total, desc, sink, metric):
+    """Run one pass, streaming rows to `sink` and the bar as they arrive.
+
+    capture_output would withhold every row until the pass ended, which for a
+    multi-hour sweep means no way to tell a slow run from a wedged one.
+    """
+    lines = []
+    bar = _progress(total, desc)
+    proc = subprocess.Popen([binary] + extra, stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE, text=True, bufsize=1)
+    try:
+        for line in proc.stdout:
+            lines.append(line)
+            if sink:
+                sink.write(line)
+                sink.flush()          # a killed run keeps its rows
+            stripped = line.strip()
+            if (stripped and not stripped.startswith("#")
+                    and "," in stripped
+                    and not stripped.startswith("topology,")):
+                bar.update(1)
+    finally:
+        bar.close()
+        proc.stdout.close()
+        stderr = proc.stderr.read()
+        proc.stderr.close()
+        rc = proc.wait()
+
+    if rc != 0:
+        print(stderr, file=sys.stderr)
+        sys.exit(f"{binary} failed with {rc}")
+    return parse_csv_lines(lines, metric)
+
+
+def report(passes, metric, tolerance, npasses):
     keys = set(passes[0])
     for p in passes[1:]:
         keys &= set(p)
@@ -103,20 +175,84 @@ def main():
         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
+        ok = worst <= 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}% — "
+        print(f"{failures}/{len(keys)} rows exceed +/-{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.")
+    print(f"all {len(keys)} rows within +/-{tolerance:g}% "
+          f"over {npasses} passes — Phase-0 gate met.")
     return 0
 
 
+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)")
+    ap.add_argument("--out-dir", default=None,
+                    help="write pass-NN.csv as rows arrive "
+                         "(default: bench_runs/<timestamp>)")
+    ap.add_argument("--resume", action="store_true",
+                    help="reuse complete pass-NN.csv files in --out-dir")
+
+    # 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)
+
+    out_dir = args.out_dir
+    if out_dir is None:
+        if args.resume:
+            sys.exit("--resume needs an explicit --out-dir")
+        stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
+        out_dir = os.path.join("bench_runs", stamp)
+    out = pathlib.Path(out_dir)
+    out.mkdir(parents=True, exist_ok=True)
+
+    total = count_rows(args.binary, extra)
+    print(f"writing to {out}/", file=sys.stderr)
+    if total:
+        print(f"{total} rows per pass, {args.passes} passes", file=sys.stderr)
+
+    passes = []
+    metric = args.metric
+    for i in range(args.passes):
+        path = out / f"pass-{i + 1:02d}.csv"
+
+        if args.resume and path.exists():
+            rows, metric = parse_csv(path.read_text(), metric)
+            # A partial file from a killed run must not be silently averaged
+            # in as if it were a whole pass.
+            if total and len(rows) < total:
+                print(f"pass {i + 1}/{args.passes}: {path.name} has "
+                      f"{len(rows)}/{total} rows — rerunning", file=sys.stderr)
+            else:
+                print(f"pass {i + 1}/{args.passes}: reusing {path.name} "
+                      f"({len(rows)} rows)", file=sys.stderr)
+                passes.append(rows)
+                continue
+
+        print(f"pass {i + 1}/{args.passes} ...", file=sys.stderr, flush=True)
+        with open(path, "w") as sink:
+            rows, metric = run_pass(args.binary, extra, total,
+                                    f"pass {i + 1}/{args.passes}", sink, metric)
+        passes.append(rows)
+
+    return report(passes, metric, args.tolerance, args.passes)
+
+
 if __name__ == "__main__":
     sys.exit(main())