fix: 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, because the caller believes they have a listener.

start() now delivers it to each node, exactly as StaticNetwork does.

The type changes with it. It was void(name, exception_ptr), which cannot
express the keep-running decision the node side needs — so it is now
NodeErrorHandler, the same alias StaticNetwork uses. That is a breaking change
in principle; in practice nothing in the tree called this setter, which is how
it stayed dead long enough to be worth finding.

Verified by the new case: the node throws, the handler receives the name and
the exception, returns true, and the node goes on to process the next value.
148/148.

.gitignore

@@ -27,3 +27,6 @@ Thumbs.db
 
 # Claude Code local settings
 .claude/settings.local.json
+include/kpn/ort_cache/
+build-tsan/
+build-*/

include/kpn/branch.hpp

@@ -7,12 +7,78 @@
 
 #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
@@ -126,15 +192,20 @@ 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);
-                if (idx < N && out_channels_[idx]) {
+                duration_t  parked{0};
-                    try { out_channels_[idx]->push(val); }
+                bool        delivered = true;
-                    catch (const ChannelOverflowError&) {}
+                if (idx < N && out_channels_[idx])
-                }
+                    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), duration_t(t1 - t0), cpu0, cpu1);
+                stats_.record_exec(duration_t(t2 - t1) - parked,
+                                   duration_t(t1 - t0) + parked, cpu0, cpu1);
+                if (!delivered) break;
             } catch (const ChannelClosedError&) {
                 break;
             }
@@ -261,12 +332,19 @@ 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_) {
-                    try { out_ch_->push(val); }
+                    duration_t parked{0};
-                    catch (const ChannelOverflowError&) {}
+                    const bool delivered = deliver_one(out_ch_, val, stop_flag_, parked);
                     auto cpu1 = NodeStats::cpu_now();
                     auto t2   = clock_t::now();
-                    stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
+                    stats_.record_exec(duration_t(t2 - t1) - parked,
+                                       duration_t(t1 - t0) + parked, cpu0, cpu1);
+                    if (!delivered) break;
                 }
             } catch (const ChannelClosedError&) {
                 break;

include/kpn/channel.hpp

@@ -58,6 +58,18 @@ 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 +148,55 @@ public:
             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);
+        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();
+        if (was_empty && 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
@@ -178,12 +239,37 @@ 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 —
+    //
-    // teardown is in progress, so the sentinel is moot).
+    // The slot holds exactly one undelivered token. A second offered before the
-    bool push_sentinel(T value) {
+    // first is taken is refused, not queued and not overwritten: two control
+    // 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 false;
+            return SentinelResult::Closed;
+        }
+        // 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.
+        if (has_eof_.load(std::memory_order_acquire)) {
+            stats_.record_drop();
+            return SentinelResult::SlotBusy;
         }
         eof_value_ = make_storage(std::move(value));
         has_eof_.store(true, std::memory_order_release);
@@ -192,7 +278,13 @@ public:
         wake_.fetch_add(1, std::memory_order_release);
         wake_.notify_one();
         if (push_callback_) push_callback_();
-        return true;
+        return SentinelResult::Taken;
+    }
+
+    /// 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.
@@ -248,6 +340,8 @@ 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;
         }
@@ -269,11 +363,15 @@ 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);
-        if (h == tail_.load(std::memory_order_acquire))
+        const std::size_t t = 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;
     }
 
@@ -297,9 +395,15 @@ public:
 
     // 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 {
-        return tail_.load(std::memory_order_relaxed)
+        const std::size_t h = head_.load(std::memory_order_relaxed);
-             - head_.load(std::memory_order_relaxed);
+        const std::size_t t = tail_.load(std::memory_order_acquire);
+        return t - h;
     }
 
     // A pending out-of-band sentinel (EOF) counts as consumable work here even
@@ -316,8 +420,9 @@ public:
     const ChannelStats& stats() const { return stats_; }
 
     ChannelSnapshot snapshot(const std::string& name) const {
-        const std::size_t t = tail_.load(std::memory_order_relaxed);
+        // head_ before tail_, for the reason given on size().
         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_,
@@ -352,6 +457,24 @@ 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();
@@ -363,6 +486,7 @@ 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

include/kpn/diagnostics.hpp

@@ -51,6 +51,14 @@ 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.
@@ -89,6 +97,7 @@ 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);
@@ -147,6 +156,29 @@ 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 ────────────────────────────────────────────────
@@ -188,6 +220,11 @@ 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

include/kpn/fanout.hpp

@@ -7,8 +7,10 @@
 
 #include <array>
 #include <atomic>
+#include <chrono>
 #include <iostream>
 #include <memory>
+#include <optional>
 #include <thread>
 #include <tuple>
 #include <utility>
@@ -78,7 +80,9 @@ 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 ───────────────────────────────────────────────────────────
@@ -116,6 +120,79 @@ public:
     }
 
 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 {
@@ -124,16 +201,17 @@ private:
                 auto t1  = clock_t::now();
                 auto cpu0 = NodeStats::cpu_now();
 
-                for (std::size_t i = 0; i < N; ++i) {
+                duration_t parked{0};
-                    if (out_channels_[i]) {
+                const bool delivered = deliver(val, parked);
-                        try { out_channels_[i]->push(val); }
-                        catch (const ChannelOverflowError&) {}  // drop for this output independently
-                    }
-                }
 
                 auto cpu1 = NodeStats::cpu_now();
                 auto t2   = clock_t::now();
-                stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1);
+                // Time spent waiting on a full output is *blocked*, not exec: a
+                // 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;
             }

include/kpn/inode.hpp

@@ -22,6 +22,23 @@ 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;
@@ -34,6 +51,14 @@ 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(); }
 

include/kpn/interrupt_node.hpp

@@ -104,6 +104,7 @@ 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,
         };
     }
 

include/kpn/main_thread_node.hpp

@@ -90,6 +90,8 @@ 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,
         };
     }
 

include/kpn/network.hpp

@@ -8,8 +8,10 @@
 #include <memory>
 #endif
 
+#include <condition_variable>
 #include <functional>
 #include <iomanip>
+#include <mutex>
 #include <iostream>
 #include <map>
 #include <set>
@@ -39,8 +41,15 @@ public:
 
 class Network : public INode {
 public:
-    using ErrorHandler =
+    /// Application-level error listener. Receives the exception any node's
-        std::function<void(std::string_view node_name, std::exception_ptr)>;
+    /// 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.
+    ///
+    /// 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>&)>;
@@ -91,6 +100,7 @@ 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;
@@ -134,6 +144,17 @@ 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();
@@ -210,6 +231,11 @@ 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); }
@@ -367,19 +393,46 @@ private:
         return true;
     }
 
-    void drain_output_channels(const std::string& /*name*/) const {
+    /// Wait for the channels fed by `name` to empty, or give up.
-        // Poll all channel probes until none report non-zero fill.
+    ///
-        // A short sleep prevents busy-spin; 1 ms is fine for drain purposes.
+    /// This took a node name and ignored it, polling *every* channel in the
-        bool any_full = true;
+    /// graph instead — so shutdown() waited for the whole network to be idle
-        while (any_full) {
+    /// before stopping each successive layer. With no deadline either, anything
-            any_full = false;
+    /// wedged downstream turned a graceful shutdown into the hang it exists to
-            for (auto& probe : channel_probes_) {
+    /// avoid.
-                auto snap = probe->snapshot();
+    ///
-                if (snap.current_fill > 0) { any_full = true; break; }
+    /// Two bounds, because they fail differently. The deadline covers a
-            }
+    /// consumer that has stopped consuming, where fill never changes and
-            if (any_full)
+    /// waiting cannot help. The no-progress counter covers one that is merely
+    /// 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 ───────────────────────────────────
@@ -398,9 +451,20 @@ 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_);
+                {
-                if (tok.stop_requested()) break;
+                    std::unique_lock lk(m);
+                    if (cv.wait_for(lk, tok, watchdog_interval_,
+                                    [&tok] { return tok.stop_requested(); }))
+                        break;
+                }
 
                 auto s = collect_snapshots();
                 check_hung_nodes();
@@ -444,6 +508,10 @@ 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_;

include/kpn/pool_node.hpp

@@ -5,6 +5,7 @@
 #include "inode.hpp"
 #include "port.hpp"
 #include "scheduler.hpp"
+#include "submit_gate.hpp"
 #include "traits.hpp"
 
 #include <array>
@@ -15,6 +16,7 @@
 #include <iostream>
 #include <memory>
 #include <optional>
+#include <variant>
 #include <stdexcept>
 #include <thread>
 #include <tuple>
@@ -22,44 +24,15 @@
 
 namespace kpn {
 
-// ── Sentinel detection ────────────────────────────────────────────────────────
+// Sentinel detection (has_eof_field / is_sentinel_value) lives in traits.hpp —
-// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
+// every node type that forwards values needs it, not just pool-scheduled ones.
-// 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 the pipeline's SceneFrame/…/MatchedSceneFrame message types, which 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.
-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;
-}
 
 // ── PoolNode ──────────────────────────────────────────────────────────────────
 //
 // Reactive alternative to Node<>. Instead of owning a blocked thread, the node
 // is submitted to a shared IScheduler whenever all its input channels become
 // non-empty. A single fire_once() call pops all inputs, executes the function,
-// and pushes outputs. At most one fire_once() runs at a time (queued_ flag).
+// and pushes outputs. At most one fire_once() runs at a time (see SubmitGate).
 //
 // Source nodes (input_count == 0) submit themselves immediately on start() and
 // resubmit after each fire_once().
@@ -108,21 +81,43 @@ public:
 
     // ── INode ─────────────────────────────────────────────────────────────────
 
+    void prepare() override {
+        if (prepared_) return;          // idempotent: the network calls this,
+        prepared_ = true;               // and start() calls it again if not.
+        register_callbacks(std::make_index_sequence<input_count>{});
+    }
+
     void start() override {
+        prepare();
         enable_inputs(std::make_index_sequence<input_count>{});
         stop_flag_.store(false, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_relaxed);
+        gate_.force_idle();
-        register_callbacks(std::make_index_sequence<input_count>{});
         if constexpr (input_count == 0)
             try_submit(0.5f);
+        else
+            // Never start with a wake already outstanding — the startup case of
+            // the invariant 9c5ce5f established for the running pipeline.
+            //
+            // The callback is installed by prepare(), before any node runs, but
+            // a network still starts its nodes one at a time: an upstream node
+            // that is already firing can push into this one between the two
+            // calls. The push is accepted by the ring and does invoke the
+            // callback, but on_input_ready() sees stop_flag_ still set and
+            // returns. Every later push sees a non-empty ring and stays silent
+            // — Channel invokes push_callback_ only on the empty->non-empty
+            // transition — so without this the node is never submitted and the
+            // pipeline reads as wedged from the first frame.
+            //
+            // on_input_ready() is the level-triggered form of the same
+            // question, so asking it once here converts the missed edge into a
+            // state check.
+            on_input_ready();
     }
 
     void stop() override {
         stop_flag_.store(true, std::memory_order_seq_cst);
         disable_inputs(std::make_index_sequence<input_count>{});
-        // fire_once() observes stop_flag_ and will not resubmit.
+        await_quiescence();
-        // We do not wait for an in-flight fire_once() to complete here;
-        // callers that need that guarantee should call scheduler_->drain() first.
     }
 
     bool running() const override {
@@ -131,6 +126,7 @@ public:
 
     void set_name(std::string name) override { name_ = std::move(name); }
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
+    void set_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
     void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
@@ -154,6 +150,9 @@ public:
             stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
             total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
             qwait_ms,
+            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
+            gate_.queued(),
+            gate_.wake_pending(),
         };
     }
 
@@ -234,6 +233,8 @@ private:
 
     template<std::size_t... Is>
     void register_callbacks(std::index_sequence<Is...>) {
+        // A parked producer is re-submitted when its output drains.
+        register_space_callbacks(std::make_index_sequence<output_count>{});
         (std::get<Is>(input_channels_)->set_push_callback(
             [this] { on_input_ready(); }), ...);
     }
@@ -243,11 +244,27 @@ private:
         for (auto& cb : cbs) if (cb) cb(ts);
     }
 
+    template<std::size_t... Os>
+    bool outputs_have_space(std::index_sequence<Os...>) const {
+        return (... && (!std::get<Os>(output_channels_) ||
+                        std::get<Os>(output_channels_)->has_space()));
+    }
+
+    template<std::size_t... Os>
+    void register_space_callbacks(std::index_sequence<Os...>) {
+        ((std::get<Os>(output_channels_)
+            ? (void)std::get<Os>(output_channels_)->set_space_callback(
+                  [this] { try_submit(0.5f); })
+            : (void)0), ...);
+    }
+
     void self_stop() {
         disable_inputs(std::make_index_sequence<input_count>{});
         disable_outputs(std::make_index_sequence<output_count>{});
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_release);
+        // force_idle, not finish_firing(): this node is stopping, and
+        // honouring a pending wake here would resubmit a dead node.
+        gate_.force_idle();
         stop_flag_.store(true, std::memory_order_relaxed);
     }
 
@@ -280,11 +297,51 @@ private:
         return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
     }
 
+    /// Priority in [0,1], higher runs sooner.
+    ///
+    /// Input fill alone answers "how much work is waiting for me". That is
+    /// only half the question: a node whose *outputs* are already full cannot
+    /// deliver anything: running it produces a value with nowhere to go, so it
+    /// immediately parks and the slot is wasted. Meanwhile the node that would
+    /// have drained that full channel waits behind it.
+    ///
+    /// So occupancy of the outputs is deducted from occupancy of the inputs.
+    /// The scheduler then naturally favours whoever is furthest downstream of
+    /// a bottleneck — the node whose inputs are backed up but whose outputs
+    /// have room is exactly the one whose execution frees the most capacity —
+    /// and defers producers that would only deepen a queue that is already
+    /// full.
+    ///
+    /// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero:
+    /// both terms are mean fills in [0,1], so the difference is in [-1,1], and
+    /// the affine map keeps the whole range distinguishable instead of
+    /// collapsing every output-saturated node onto the same value. 0.5 remains
+    /// the neutral point, matching the default used for source nodes.
     float compute_priority() {
         if constexpr (input_count == 0) return 0.5f;
-        float sum = 0.0f;
+        float in = 0.0f;
-        sum_fill(sum, std::make_index_sequence<input_count>{});
+        sum_fill(in, std::make_index_sequence<input_count>{});
-        return sum / static_cast<float>(input_count);
+        in /= static_cast<float>(input_count);
+
+        if constexpr (output_count == 0) return in;
+
+        float out = 0.0f;
+        sum_output_fill(out, std::make_index_sequence<output_count>{});
+        out /= static_cast<float>(output_count);
+
+        const float p = 0.5f * (1.0f + in - out);
+        return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p);
+    }
+
+    /// Mean fill of the output channels, same normalisation as sum_fill.
+    /// An unconnected output holds nothing back, so it contributes 0.
+    template<std::size_t... Os>
+    void sum_output_fill(float& sum, std::index_sequence<Os...>) {
+        ((sum += (std::get<Os>(output_channels_) &&
+                  std::get<Os>(output_channels_)->capacity() > 0)
+            ? float(std::get<Os>(output_channels_)->approx_size())
+              / float(std::get<Os>(output_channels_)->capacity())
+            : 0.0f), ...);
     }
 
     template<std::size_t... Is>
@@ -295,17 +352,121 @@ private:
             : 0.5f), ...);
     }
 
+    /// Submit unless a firing is already in flight. A wake that arrives while
+    /// one is is *recorded* against it, never dropped.
+    ///
+    /// Wakes are edge-triggered: a channel fires its space callback on the
+    /// transition, once. A dropped one never returns, so a node could park a
+    /// value, release its worker, and sleep forever holding output its consumer
+    /// was waiting for, with every worker idle in cond_wait and nothing left to
+    /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same
+    /// variable, so the two cannot both be true — see submit_gate.hpp.
     void try_submit(float priority) {
-        bool expected = false;
+        // A stopped node must not claim the gate. The scheduler now refuses
-        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
+        // submissions after its pool stops, so the submit itself is safe — but
+        // claiming and never releasing would leave the gate held, and a restart
+        // would then have to clear it. start() does, but relying on that makes
+        // the invariant depend on a distant statement.
+        if (stop_flag_.load(std::memory_order_relaxed)) return;
+        if (gate_.claim())
             scheduler_->submit([this] { fire_once(); }, priority);
     }
 
     // ── Execution ─────────────────────────────────────────────────────────────
 
+    /// Decide whether this node should run again, then release the gate — in
+    /// that order, always.
+    ///
+    /// Releasing first is what let two firings of the same node overlap: the
+    /// moment the gate is free another worker may enter fire_once, while this
+    /// invocation is still reading pending_ and writing pending_done_. TSan
+    /// caught it as a race on pending_done_ between a firing submitted by the
+    /// old release_and_recheck and one submitted by try_submit. It also quietly
+    /// broke the one-slot park, which is sound only because "at most one
+    /// fire_once runs per node at a time" — with two, a value can be parked by
+    /// one firing and overwritten by the other.
+    ///
+    /// Everything this reads belongs to the firing that holds the claim, so it
+    /// is all evaluated first and the release is the last thing the firing does.
+    void finish_firing() {
+        bool  want_more = false;
+        float prio      = 0.5f;
+
+        if (!stop_flag_.load(std::memory_order_relaxed)) {
+            bool parked = false;
+            if constexpr (!std::is_void_v<return_raw>)
+                parked = pending_.has_value();
+
+            if (parked) {
+                // Still holding output: only worth running again once the
+                // consumer has made room.
+                want_more = outputs_have_space(std::make_index_sequence<output_count>{});
+            } else {
+                if constexpr (input_count == 0) {
+                    want_more = true;              // sources always run again
+                } else {
+                    want_more = count_ready(std::make_index_sequence<input_count>{})
+                                == input_count;
+                    if (want_more) prio = compute_priority();
+                }
+            }
+        }
+
+        if (gate_.release())    scheduler_->submit([this] { fire_once(); }, prio);
+        else if (want_more)     try_submit(prio);
+    }
+
+    /// Block until no firing of this node is in flight or queued.
+    ///
+    /// stop() used to set the flag and return, leaving an executing fire_once
+    /// touching input_channels_, stats_ and pending_ while the caller went on
+    /// to destroy them. For a node with a private pool that was survivable by
+    /// accident — Node::stop() calls pool->stop(), which joins — but a node
+    /// sharing a pool had nothing joining it at all, so ~PoolNode raced its own
+    /// members. The old comment said callers wanting the guarantee should call
+    /// scheduler_->drain() first; a destructor cannot, and the default should
+    /// not be a use-after-free.
+    ///
+    /// The gate is exactly the right thing to wait on: it is claimed for the
+    /// whole of a firing and released as the last act of one. A queued but
+    /// unstarted firing also holds it, and will run, observe stop_flag_ and
+    /// release — which is why the pool must still be running when this is
+    /// called. That is already the documented order (stop nodes, then the
+    /// pool), and Node/ObjectNode do it that way.
+    ///
+    /// Bounded, because a node function that never returns must not turn
+    /// teardown into a hang; and skipped entirely when called from the firing
+    /// thread itself, since an error handler that stops its own node would
+    /// otherwise wait for a firing that is waiting for it.
+    void await_quiescence() {
+        if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id())
+            return;
+        const auto deadline = clock_t::now() + std::chrono::seconds(5);
+        while (gate_.queued()) {
+            if (clock_t::now() >= deadline) {
+                std::cerr << "[kpn] stop: node '" << name_
+                          << "' still had work in flight after 5 s; "
+                             "continuing without it\n";
+                return;
+            }
+            std::this_thread::sleep_for(std::chrono::microseconds(50));
+        }
+    }
+
+    /// Marks fire_once's thread for the duration of a firing, so await_quiescence
+    /// can tell a re-entrant stop() from an external one.
+    struct FiringMark {
+        std::atomic<std::thread::id>& slot;
+        explicit FiringMark(std::atomic<std::thread::id>& s) : slot(s) {
+            slot.store(std::this_thread::get_id(), std::memory_order_release);
+        }
+        ~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); }
+    };
+
     void fire_once() {
+        FiringMark mark(firing_thread_);
         if (stop_flag_.load(std::memory_order_relaxed)) {
-            queued_.store(false, std::memory_order_release);
+            gate_.force_idle();
             return;
         }
 
@@ -315,6 +476,37 @@ private:
             t0.time_since_epoch()).count();
         stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
 
+        // Parked from a previous firing: retry that value before touching the
+        // inputs. Returning here releases the worker — the channel's space
+        // callback re-submits this node when the consumer drains a slot.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
+                // Whether the value went out or is still parked, finish_firing
+                // reads pending_ and picks the right follow-up: output space if
+                // still holding, input readiness if drained. Resubmitting
+                // unconditionally would fire a node whose inputs are empty, and
+                // pop_one reports an empty channel as ChannelClosedError — which
+                // this node treats as "upstream finished" and self-stops on.
+                finish_firing();
+                return;
+            }
+        }
+
+        // Woken by output space rather than by input arrival, with nothing
+        // parked left to flush: there is no work to do. Falling through would
+        // read an empty channel, and pop_one reports empty as
+        // ChannelClosedError — self-stopping a live node. Release the worker;
+        // on_input_ready() resubmits when data actually lands.
+        if constexpr (input_count > 0) {
+            if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
+                // finish_firing re-checks readiness after the work above, so
+                // data that landed while we looked is not missed.
+                finish_firing();
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1 = clock_t::now();
@@ -333,6 +525,14 @@ private:
             auto t2   = clock_t::now();
             // blocked_time = 0 for pool nodes (we don't block waiting for inputs)
             stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
+        } catch (const ChannelEmptyError&) {
+            // Not an error: there was simply nothing to take. Release and wait
+            // to be woken again. fire_once checks readiness before it gets
+            // here, and this node is the sole consumer of its inputs, so this
+            // is unreachable today — it exists so that if the check is ever
+            // weakened the cost is a wasted firing rather than a dead node.
+            finish_firing();
+            return;
         } catch (const ChannelClosedError&) {
             fire_callbacks(closed_callbacks_);
             self_stop();
@@ -340,7 +540,11 @@ private:
         } catch (const ChannelOverflowError&) {
             fire_callbacks(event_callbacks_);
         } catch (...) {
-            if (error_handler_ && error_handler_(name_, std::current_exception())) {
+            auto eptr = std::current_exception();
+            const bool handled =
+                (error_handler_     && error_handler_(name_, eptr)) ||
+                (net_error_handler_ && net_error_handler_(name_, eptr));
+            if (handled) {
                 // continue — fall through to resubmit check
             } else {
                 fire_callbacks(closed_callbacks_);
@@ -350,20 +554,15 @@ private:
         }
 
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_release);
+        // If the push above parked, finish_firing waits on output space rather
-
+        // than input arrival: this firing consumed its input, so an input-level
-        if (stop_flag_.load(std::memory_order_relaxed)) return;
+        // check would not resubmit and the node would hold its value forever
-
+        // while its consumer waits for exactly that value.
-        // Source nodes always resubmit; others resubmit only if inputs are ready.
+        finish_firing();
-        if constexpr (input_count == 0) {
-            try_submit(0.5f);
-        } else {
-            on_input_ready();
-        }
     }
 
     // Pop all inputs — safe because we're the sole consumer and fire_once
-    // is guarded by queued_ (only one fire_once runs at a time).
+    // is guarded by the submit gate (only one fire_once runs at a time).
     template<std::size_t... Is>
     args_tuple pop_inputs(std::index_sequence<Is...>) {
         return {pop_one<Is>()...};
@@ -373,8 +572,16 @@ private:
     std::tuple_element_t<I, args_tuple> pop_one() {
         auto& ch = *std::get<I>(input_channels_);
         std::tuple_element_t<I, args_tuple> val;
-        if (!ch.try_pop_now(val))
+        if (!ch.try_pop_now(val)) {
+            // try_pop_now returns false for "nothing available", which covers
+            // two very different situations. A closed channel means upstream is
+            // finished and this node should stop. An open one means only that
+            // nothing is here at this instant — and treating that as closed
+            // kills a live node, which then disables its own inputs and outputs
+            // and takes the rest of the pipeline with it.
+            if (ch.is_accepting()) throw ChannelEmptyError{};
             throw ChannelClosedError{};
+        }
         return val;
     }
 
@@ -385,27 +592,59 @@ private:
     }
 
     template<std::size_t... Is>
+    /// Pushes what it can and parks the rest. `done_` marks the elements that
+    /// were taken, so a retry never re-pushes one — a duplicate would be as
+    /// wrong as a drop, just harder to notice.
     void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
-        (push_one_out<Is>(std::get<Is>(std::move(result))), ...);
+        bool all = true;
+        ((pending_done_[Is] = pending_done_[Is] ||
+            push_one_out<Is>(std::get<Is>(std::move(result))),
+          all = all && pending_done_[Is]), ...);
+        if (all) { pending_.reset(); pending_done_.fill(false); }
+        // The retry path calls this as push_outputs(std::move(*pending_), …), so
+        // on that path `result` *is* the parked tuple. Assigning it to itself is
+        // a self-move-assignment, which for std::tuple is elementwise — and
+        // libstdc++'s std::vector does not guard against it: it swaps its data
+        // into a temporary and leaves the vector empty. A value that failed to
+        // push twice would therefore be delivered with its payload silently
+        // erased, which downstream reads as a legitimately empty result rather
+        // than as a loss. Only store when it is not already stored.
+        else if (!pending_ || &result != &*pending_) pending_ = std::move(result);
     }
 
+    /// Returns false when the ring was full and the value was NOT taken; the
+    /// caller must keep it and retry after the channel signals space.
     template<std::size_t I>
-    void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
+    bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
-        if (!ch) return;
+        if (!ch) return true;
         // Sentinels (EOF) must never be dropped: a lost token wedges every
         // downstream pop() forever. Deliver them out-of-band (push_sentinel),
         // which never overflows and never blocks this node's worker thread.
         if (is_sentinel_value(val)) {
-            ch->push_sentinel(std::move(val));
+            // A refused sentinel is a protocol error, not backpressure, so it
-            return;
+            // is reported rather than parked and retried — retrying would spin
-        }
+            // forever against a slot only the consumer can free, and there is
-        try {
+            // no correct value to deliver second anyway. Closed is normal
-            ch->push(std::move(val));
+            // during teardown and stays quiet.
-        } catch (const ChannelOverflowError&) {
+            if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
-            throw ChannelOverflowError(ch->capacity(),
+                    ::SentinelResult::SlotBusy)
-                "pool node '" + name_ + "' " + output_port_label<I>());
+                fire_callbacks(event_callbacks_);
+            return true;
         }
+        // Backpressure without parking the worker. A full channel means the
+        // consumer is behind; the value is kept and this node stops running
+        // until the channel signals space (set_space_callback re-submits it).
+        //
+        // Blocking here instead would sleep inside a scheduler worker, and
+        // nodes are pinned to workers — park enough of them and nothing is left
+        // to run the consumer that would drain the channel. That is the
+        // hold-and-wait deadlock channel.hpp warns about for sentinels; it
+        // applies to data pushes just as much.
+        // Closed counts as "stop trying", not as delivered: the value is gone
+        // and the channel has recorded the drop. Only Full means park and retry.
+        return ch->try_push(val) != Channel<std::tuple_element_t<I, return_tuple>>
+                                        ::PushResult::Full;
     }
 
     template<std::size_t I>
@@ -426,9 +665,29 @@ private:
     input_channels_t                         input_channels_;
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
-    std::atomic<bool>                        queued_{false};
+    /// Serialises firings and records wakes that arrive during one. See
+    /// submit_gate.hpp for why this cannot be two separate flags.
+    SubmitGate                               gate_;
+    /// Whether prepare() has installed the channel callbacks. Only ever touched
+    /// from the thread driving start()/stop(), never from a worker, and never
+    /// cleared: the callbacks capture `this` and stay valid across a restart, so
+    /// re-registering them would be a pointless write to a live channel.
+    bool                                     prepared_{false};
+    /// Thread currently inside fire_once, or a default id when none is.
+    /// See await_quiescence.
+    std::atomic<std::thread::id>             firing_thread_{};
+
+    /// The hidden one-slot output buffer (see push_outputs). Holding the value
+    /// here is what lets a node stop running without dropping it or occupying a
+    /// scheduler worker. One slot suffices because at most one fire_once() runs
+    /// per node at a time — with concurrent firing it would have to hold a whole
+    /// FIFO's worth.
+    std::conditional_t<std::is_void_v<return_raw>, std::monostate,
+                       std::optional<return_tuple>> pending_{};
+    std::array<bool, (output_count ? output_count : 1)> pending_done_{};
     NodeStats                                stats_;
     NodeErrorHandler                         error_handler_;
+    NodeErrorHandler                         net_error_handler_;
     std::chrono::milliseconds                max_exec_time_{0};
     std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
     std::array<NodeEventCallback, 2>         closed_callbacks_{};
@@ -479,23 +738,34 @@ public:
 
     ~PoolObjectNode() override { stop(); }
 
+    void prepare() override {
+        if (prepared_) return;
+        prepared_ = true;
+        register_callbacks(std::make_index_sequence<input_count>{});
+    }
+
     void start() override {
+        prepare();
         enable_inputs(std::make_index_sequence<input_count>{});
         stop_flag_.store(false, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_relaxed);
+        gate_.force_idle();
-        register_callbacks(std::make_index_sequence<input_count>{});
         if constexpr (input_count == 0)
             try_submit(0.5f);
+        else
+            // Never start with a wake already outstanding — see PoolNode::start().
+            on_input_ready();
     }
 
     void stop() override {
         stop_flag_.store(true, std::memory_order_seq_cst);
         disable_inputs(std::make_index_sequence<input_count>{});
+        await_quiescence();
     }
 
     bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); }
     void set_name(std::string name) override { name_ = std::move(name); }
     void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); }
+    void set_network_error_callback(NodeErrorHandler h) override { net_error_handler_ = std::move(h); }
     void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; }
 
     void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); }
@@ -519,6 +789,9 @@ public:
             stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
             total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
             qwait_ms,
+            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
+            gate_.queued(),
+            gate_.wake_pending(),
         };
     }
 
@@ -567,6 +840,8 @@ private:
     }
     template<std::size_t... Is>
     void register_callbacks(std::index_sequence<Is...>) {
+         // A parked producer is re-submitted when its output drains.
+         register_space_callbacks(std::make_index_sequence<output_count>{});
         (std::get<Is>(input_channels_)->set_push_callback([this] { on_input_ready(); }), ...);
     }
 
@@ -575,11 +850,27 @@ private:
         for (auto& cb : cbs) if (cb) cb(ts);
     }
 
+     template<std::size_t... Os>
+     bool outputs_have_space(std::index_sequence<Os...>) const {
+         return (... && (!std::get<Os>(output_channels_) ||
+                         std::get<Os>(output_channels_)->has_space()));
+     }
+
+     template<std::size_t... Os>
+     void register_space_callbacks(std::index_sequence<Os...>) {
+         ((std::get<Os>(output_channels_)
+             ? (void)std::get<Os>(output_channels_)->set_space_callback(
+                   [this] { try_submit(0.5f); })
+             : (void)0), ...);
+     }
+
     void self_stop() {
         disable_inputs(std::make_index_sequence<input_count>{});
         disable_outputs(std::make_index_sequence<output_count>{});
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_release);
+        // force_idle, not finish_firing(): this node is stopping, and
+        // honouring a pending wake here would resubmit a dead node.
+        gate_.force_idle();
         stop_flag_.store(true, std::memory_order_relaxed);
     }
 
@@ -606,11 +897,51 @@ private:
         return ((std::get<Is>(input_channels_)->approx_size() > 0 ? 1u : 0u) + ...);
     }
 
+    /// Priority in [0,1], higher runs sooner.
+    ///
+    /// Input fill alone answers "how much work is waiting for me". That is
+    /// only half the question: a node whose *outputs* are already full cannot
+    /// deliver anything: running it produces a value with nowhere to go, so it
+    /// immediately parks and the slot is wasted. Meanwhile the node that would
+    /// have drained that full channel waits behind it.
+    ///
+    /// So occupancy of the outputs is deducted from occupancy of the inputs.
+    /// The scheduler then naturally favours whoever is furthest downstream of
+    /// a bottleneck — the node whose inputs are backed up but whose outputs
+    /// have room is exactly the one whose execution frees the most capacity —
+    /// and defers producers that would only deepen a queue that is already
+    /// full.
+    ///
+    /// Mapped as 0.5·(1 + in - out) rather than clamping (in - out) at zero:
+    /// both terms are mean fills in [0,1], so the difference is in [-1,1], and
+    /// the affine map keeps the whole range distinguishable instead of
+    /// collapsing every output-saturated node onto the same value. 0.5 remains
+    /// the neutral point, matching the default used for source nodes.
     float compute_priority() {
         if constexpr (input_count == 0) return 0.5f;
-        float sum = 0.0f;
+        float in = 0.0f;
-        sum_fill(sum, std::make_index_sequence<input_count>{});
+        sum_fill(in, std::make_index_sequence<input_count>{});
-        return sum / static_cast<float>(input_count);
+        in /= static_cast<float>(input_count);
+
+        if constexpr (output_count == 0) return in;
+
+        float out = 0.0f;
+        sum_output_fill(out, std::make_index_sequence<output_count>{});
+        out /= static_cast<float>(output_count);
+
+        const float p = 0.5f * (1.0f + in - out);
+        return p < 0.0f ? 0.0f : (p > 1.0f ? 1.0f : p);
+    }
+
+    /// Mean fill of the output channels, same normalisation as sum_fill.
+    /// An unconnected output holds nothing back, so it contributes 0.
+    template<std::size_t... Os>
+    void sum_output_fill(float& sum, std::index_sequence<Os...>) {
+        ((sum += (std::get<Os>(output_channels_) &&
+                  std::get<Os>(output_channels_)->capacity() > 0)
+            ? float(std::get<Os>(output_channels_)->approx_size())
+              / float(std::get<Os>(output_channels_)->capacity())
+            : 0.0f), ...);
     }
     template<std::size_t... Is>
     void sum_fill(float& sum, std::index_sequence<Is...>) {
@@ -620,15 +951,119 @@ private:
             : 0.5f), ...);
     }
 
+    /// Submit unless a firing is already in flight. A wake that arrives while
+    /// one is is *recorded* against it, never dropped.
+    ///
+    /// Wakes are edge-triggered: a channel fires its space callback on the
+    /// transition, once. A dropped one never returns, so a node could park a
+    /// value, release its worker, and sleep forever holding output its consumer
+    /// was waiting for, with every worker idle in cond_wait and nothing left to
+    /// re-trigger it. SubmitGate makes "idle" and "wake outstanding" the same
+    /// variable, so the two cannot both be true — see submit_gate.hpp.
     void try_submit(float priority) {
-        bool expected = false;
+        // A stopped node must not claim the gate. The scheduler now refuses
-        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
+        // submissions after its pool stops, so the submit itself is safe — but
+        // claiming and never releasing would leave the gate held, and a restart
+        // would then have to clear it. start() does, but relying on that makes
+        // the invariant depend on a distant statement.
+        if (stop_flag_.load(std::memory_order_relaxed)) return;
+        if (gate_.claim())
             scheduler_->submit([this] { fire_once(); }, priority);
     }
 
+    /// Decide whether this node should run again, then release the gate — in
+    /// that order, always.
+    ///
+    /// Releasing first is what let two firings of the same node overlap: the
+    /// moment the gate is free another worker may enter fire_once, while this
+    /// invocation is still reading pending_ and writing pending_done_. TSan
+    /// caught it as a race on pending_done_ between a firing submitted by the
+    /// old release_and_recheck and one submitted by try_submit. It also quietly
+    /// broke the one-slot park, which is sound only because "at most one
+    /// fire_once runs per node at a time" — with two, a value can be parked by
+    /// one firing and overwritten by the other.
+    ///
+    /// Everything this reads belongs to the firing that holds the claim, so it
+    /// is all evaluated first and the release is the last thing the firing does.
+    void finish_firing() {
+        bool  want_more = false;
+        float prio      = 0.5f;
+
+        if (!stop_flag_.load(std::memory_order_relaxed)) {
+            bool parked = false;
+            if constexpr (!std::is_void_v<return_raw>)
+                parked = pending_.has_value();
+
+            if (parked) {
+                // Still holding output: only worth running again once the
+                // consumer has made room.
+                want_more = outputs_have_space(std::make_index_sequence<output_count>{});
+            } else {
+                if constexpr (input_count == 0) {
+                    want_more = true;              // sources always run again
+                } else {
+                    want_more = count_ready(std::make_index_sequence<input_count>{})
+                                == input_count;
+                    if (want_more) prio = compute_priority();
+                }
+            }
+        }
+
+        if (gate_.release())    scheduler_->submit([this] { fire_once(); }, prio);
+        else if (want_more)     try_submit(prio);
+    }
+
+    /// Block until no firing of this node is in flight or queued.
+    ///
+    /// stop() used to set the flag and return, leaving an executing fire_once
+    /// touching input_channels_, stats_ and pending_ while the caller went on
+    /// to destroy them. For a node with a private pool that was survivable by
+    /// accident — Node::stop() calls pool->stop(), which joins — but a node
+    /// sharing a pool had nothing joining it at all, so ~PoolNode raced its own
+    /// members. The old comment said callers wanting the guarantee should call
+    /// scheduler_->drain() first; a destructor cannot, and the default should
+    /// not be a use-after-free.
+    ///
+    /// The gate is exactly the right thing to wait on: it is claimed for the
+    /// whole of a firing and released as the last act of one. A queued but
+    /// unstarted firing also holds it, and will run, observe stop_flag_ and
+    /// release — which is why the pool must still be running when this is
+    /// called. That is already the documented order (stop nodes, then the
+    /// pool), and Node/ObjectNode do it that way.
+    ///
+    /// Bounded, because a node function that never returns must not turn
+    /// teardown into a hang; and skipped entirely when called from the firing
+    /// thread itself, since an error handler that stops its own node would
+    /// otherwise wait for a firing that is waiting for it.
+    void await_quiescence() {
+        if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id())
+            return;
+        const auto deadline = clock_t::now() + std::chrono::seconds(5);
+        while (gate_.queued()) {
+            if (clock_t::now() >= deadline) {
+                std::cerr << "[kpn] stop: node '" << name_
+                          << "' still had work in flight after 5 s; "
+                             "continuing without it\n";
+                return;
+            }
+            std::this_thread::sleep_for(std::chrono::microseconds(50));
+        }
+    }
+
+    /// Marks fire_once's thread for the duration of a firing, so await_quiescence
+    /// can tell a re-entrant stop() from an external one.
+    struct FiringMark {
+        std::atomic<std::thread::id>& slot;
+        explicit FiringMark(std::atomic<std::thread::id>& s) : slot(s) {
+            slot.store(std::this_thread::get_id(), std::memory_order_release);
+        }
+        ~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); }
+    };
+
     void fire_once() {
+        FiringMark mark(firing_thread_);
         if (stop_flag_.load(std::memory_order_relaxed)) {
-            queued_.store(false, std::memory_order_release);
+            gate_.force_idle();
             return;
         }
         auto t0 = clock_t::now();
@@ -636,6 +1071,35 @@ private:
             t0.time_since_epoch()).count();
         stats_.exec_start_us.store(now_us, std::memory_order_relaxed);
 
+        // Parked from a previous firing: retry that value before touching the
+        // inputs. Returning here releases the worker — the channel's space
+        // callback re-submits this node once the consumer drains a slot.
+        if constexpr (!std::is_void_v<return_raw>) {
+            if (pending_) {
+                push_outputs(std::move(*pending_), std::make_index_sequence<output_count>{});
+                // Whether the value went out or is still parked, finish_firing
+                // reads pending_ and picks the right follow-up: output space if
+                // still holding, input readiness if drained. Resubmitting
+                // unconditionally would fire a node whose inputs are empty, and
+                // pop_one reports an empty channel as ChannelClosedError — which
+                // this node treats as "upstream finished" and self-stops on.
+                finish_firing();
+                return;
+            }
+        }
+
+        // See the equivalent guard in PoolNode::fire_once: a space-callback
+        // wake with nothing parked must release the worker, not fall through
+        // into pop_inputs on an empty channel.
+        if constexpr (input_count > 0) {
+            if (count_ready(std::make_index_sequence<input_count>{}) != input_count) {
+                // finish_firing re-checks readiness after the work above, so
+                // data that landed while we looked is not missed.
+                finish_firing();
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1   = clock_t::now();
@@ -652,6 +1116,14 @@ private:
             auto cpu1 = NodeStats::cpu_now();
             auto t2   = clock_t::now();
             stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
+        } catch (const ChannelEmptyError&) {
+            // Not an error: there was simply nothing to take. Release and wait
+            // to be woken again. fire_once checks readiness before it gets
+            // here, and this node is the sole consumer of its inputs, so this
+            // is unreachable today — it exists so that if the check is ever
+            // weakened the cost is a wasted firing rather than a dead node.
+            finish_firing();
+            return;
         } catch (const ChannelClosedError&) {
             fire_callbacks(closed_callbacks_);
             self_stop();
@@ -659,7 +1131,11 @@ private:
         } catch (const ChannelOverflowError&) {
             fire_callbacks(event_callbacks_);
         } catch (...) {
-            if (error_handler_ && error_handler_(name_, std::current_exception())) {
+            auto eptr = std::current_exception();
+            const bool handled =
+                (error_handler_     && error_handler_(name_, eptr)) ||
+                (net_error_handler_ && net_error_handler_(name_, eptr));
+            if (handled) {
             } else {
                 fire_callbacks(closed_callbacks_);
                 self_stop();
@@ -668,10 +1144,11 @@ private:
         }
 
         stats_.exec_start_us.store(0, std::memory_order_relaxed);
-        queued_.store(false, std::memory_order_release);
+        // If the push above parked, finish_firing waits on output space rather
-        if (stop_flag_.load(std::memory_order_relaxed)) return;
+        // than input arrival: this firing consumed its input, so an input-level
-        if constexpr (input_count == 0) try_submit(0.5f);
+        // check would not resubmit and the node would hold its value forever
-        else                            on_input_ready();
+        // while its consumer waits for exactly that value.
+        finish_firing();
     }
 
     template<std::size_t... Is>
@@ -681,7 +1158,16 @@ private:
     std::tuple_element_t<I, args_tuple> pop_one() {
         auto& ch = *std::get<I>(input_channels_);
         std::tuple_element_t<I, args_tuple> val;
-        if (!ch.try_pop_now(val)) throw ChannelClosedError{};
+        if (!ch.try_pop_now(val)) {
+            // try_pop_now returns false for "nothing available", which covers
+            // two very different situations. A closed channel means upstream is
+            // finished and this node should stop. An open one means only that
+            // nothing is here at this instant — and treating that as closed
+            // kills a live node, which then disables its own inputs and outputs
+            // and takes the rest of the pipeline with it.
+            if (ch.is_accepting()) throw ChannelEmptyError{};
+            throw ChannelClosedError{};
+        }
         return val;
     }
 
@@ -692,26 +1178,50 @@ private:
     }
 
     template<std::size_t... Is>
+    /// Pushes what it can and parks the rest. `pending_done_` marks the elements
+    /// already taken, so a retry never re-pushes one — a duplicate is as wrong as
+    /// a drop and harder to notice.
     void push_outputs(return_tuple&& result, std::index_sequence<Is...>) {
-        (push_one_out<Is>(std::get<Is>(std::move(result))), ...);
+        bool all = true;
+        ((pending_done_[Is] = pending_done_[Is] ||
+            push_one_out<Is>(std::get<Is>(std::move(result))),
+          all = all && pending_done_[Is]), ...);
+        if (all) { pending_.reset(); pending_done_.fill(false); }
+        // The retry path calls this as push_outputs(std::move(*pending_), …), so
+        // on that path `result` *is* the parked tuple. Assigning it to itself is
+        // a self-move-assignment, which for std::tuple is elementwise — and
+        // libstdc++'s std::vector does not guard against it: it swaps its data
+        // into a temporary and leaves the vector empty. A value that failed to
+        // push twice would therefore be delivered with its payload silently
+        // erased, which downstream reads as a legitimately empty result rather
+        // than as a loss. Only store when it is not already stored.
+        else if (!pending_ || &result != &*pending_) pending_ = std::move(result);
     }
+    /// Returns false when the ring was full and the value was NOT taken; the
+    /// caller must keep it and retry after the channel signals space.
     template<std::size_t I>
-    void push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
+    bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
         auto* ch = std::get<I>(output_channels_);
-        if (!ch) return;
+        if (!ch) return true;
         // Sentinels (EOF) must never be dropped: a lost token wedges every
         // downstream pop() forever. Deliver them out-of-band (push_sentinel),
         // which never overflows and never blocks this node's worker thread.
         if (is_sentinel_value(val)) {
-            ch->push_sentinel(std::move(val));
+            // A refused sentinel is a protocol error, not backpressure, so it
-            return;
+            // is reported rather than parked and retried — retrying would spin
-        }
+            // forever against a slot only the consumer can free, and there is
-        try {
+            // no correct value to deliver second anyway. Closed is normal
-            ch->push(std::move(val));
+            // during teardown and stays quiet.
-        } catch (const ChannelOverflowError&) {
+            if (ch->try_push_sentinel(val) == Channel<std::tuple_element_t<I, return_tuple>>
-            throw ChannelOverflowError(ch->capacity(),
+                    ::SentinelResult::SlotBusy)
-                "pool node '" + name_ + "'");
+                fire_callbacks(event_callbacks_);
+            return true;
         }
+        // See the note on the typed overload above: park rather than block.
+        // Closed counts as "stop trying", not as delivered: the value is gone
+        // and the channel has recorded the drop. Only Full means park and retry.
+        return ch->try_push(val) != Channel<std::tuple_element_t<I, return_tuple>>
+                                        ::PushResult::Full;
     }
 
     Obj&                                     obj_;
@@ -721,9 +1231,29 @@ private:
     input_channels_t                         input_channels_;
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
-    std::atomic<bool>                        queued_{false};
+    /// Serialises firings and records wakes that arrive during one. See
+    /// submit_gate.hpp for why this cannot be two separate flags.
+    SubmitGate                               gate_;
+    /// Whether prepare() has installed the channel callbacks. Only ever touched
+    /// from the thread driving start()/stop(), never from a worker, and never
+    /// cleared: the callbacks capture `this` and stay valid across a restart, so
+    /// re-registering them would be a pointless write to a live channel.
+    bool                                     prepared_{false};
+    /// Thread currently inside fire_once, or a default id when none is.
+    /// See await_quiescence.
+    std::atomic<std::thread::id>             firing_thread_{};
+
+    /// The hidden one-slot output buffer (see push_outputs). Holding the value
+    /// here is what lets a node stop running without dropping it or occupying a
+    /// scheduler worker. One slot suffices because at most one fire_once() runs
+    /// per node at a time — with concurrent firing it would have to hold a whole
+    /// FIFO's worth.
+    std::conditional_t<std::is_void_v<return_raw>, std::monostate,
+                       std::optional<return_tuple>> pending_{};
+    std::array<bool, (output_count ? output_count : 1)> pending_done_{};
     NodeStats                                stats_;
     NodeErrorHandler                         error_handler_;
+    NodeErrorHandler                         net_error_handler_;
     std::chrono::milliseconds                max_exec_time_{0};
     std::array<NodeEventCallback, 2>         event_callbacks_{};  // [0]=user [1]=network
     std::array<NodeEventCallback, 2>         closed_callbacks_{};

include/kpn/scheduler.hpp

@@ -5,6 +5,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <shared_mutex>
 #include <optional>
 #include <queue>
 #include <thread>
@@ -62,18 +63,30 @@ 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();
     }
@@ -86,6 +99,22 @@ 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);
@@ -93,6 +122,7 @@ 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
@@ -105,14 +135,16 @@ 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_,
-            t > a ? t - a : 0,   // queued (approximate)
+            queued_.load(std::memory_order_relaxed),   // queued (exact)
             a,                                          // executing
             submitted_.load(std::memory_order_relaxed),
             completed_.load(std::memory_order_relaxed),
@@ -142,6 +174,7 @@ 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;
     }
 
@@ -182,10 +215,13 @@ private:
             std::unique_lock lock(cv_mx_);
             cv_.wait(lock, [this] {
                 return stopped_.load(std::memory_order_seq_cst)
-                    || total_.load(std::memory_order_relaxed) > 0;
+                    || queued_.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)
-                && total_.load(std::memory_order_relaxed) == 0)
+                && queued_.load(std::memory_order_relaxed) == 0)
                 return;
         }
     }
@@ -194,17 +230,36 @@ 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
+    std::atomic<size_t>   total_{0};      // queued + executing (drain() waits on this)
+    /// 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};
 };
 

include/kpn/shared_resource.hpp

@@ -14,6 +14,18 @@ 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
@@ -72,6 +84,7 @@ 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);
@@ -83,18 +96,46 @@ public:
         current_waiters_.store(waiters_.size(), std::memory_order_relaxed);
 
         auto t0 = w.wait_start;
-        w.cv.wait(lock, [&w] { return w.ready; });
+        // Woken either by release() handing over ownership, or by close()
+        // 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; });
@@ -132,7 +173,10 @@ public:
 private:
     void release() {
         std::unique_lock lock(mutex_);
-        if (waiters_.empty()) {
+        // Hand over only to a waiter that is still waiting. A closed one is on
+        // 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;
         }
@@ -162,7 +206,8 @@ private:
         std::function<float()>  priority_fn;
         clock_t::time_point     wait_start;
         std::condition_variable cv;
-        bool                    ready{false};
+        bool                    ready{false};   // handed ownership by release()
+        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) {}
@@ -172,6 +217,7 @@ private:
 
     T                        resource_;
     bool                     held_{false};
+    bool                     closed_{false};
     mutable std::mutex       mutex_;
     std::vector<Waiter*>     waiters_;
     std::atomic<uint64_t>    acq_{0};

include/kpn/static_network.hpp

@@ -98,13 +98,15 @@ 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(); }
@@ -122,6 +124,19 @@ 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
@@ -145,6 +160,11 @@ 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)
@@ -158,11 +178,12 @@ 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 (auto* n : user_nodes_topo_) {
+        for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
-            n->stop();
+            user_nodes_topo_[i]->stop();
-            drain_all_channels();
+            drain_outputs_of(user_node_names_[i]);
         }
         for (auto* n : fanout_nodes_ptr_) n->stop();
     }
@@ -181,6 +202,18 @@ 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.
@@ -253,16 +286,50 @@ private:
         return {std::move(nodes), std::move(channels), std::move(resources), std::move(pools), elapsed_s};
     }
 
-    void drain_all_channels() const {
+    /// Wait for the channels fed by `src` to empty, or give up.
-        bool any_full = true;
+    ///
-        while (any_full) {
+    /// This was an unbounded `while (anything anywhere is non-empty)` poll over
-            any_full = false;
+    /// *every* channel in the graph, which made shutdown() wait for the whole
-            for (auto& probe : channel_probes_) {
+    /// network to be idle before stopping each successive layer, and wait
-                if (probe->snapshot().current_fill > 0) { any_full = true; break; }
+    /// forever if anything downstream was wedged — turning a graceful shutdown
-            }
+    /// 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_;
@@ -273,9 +340,14 @@ 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};
@@ -308,29 +380,6 @@ 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.
@@ -355,6 +404,38 @@ 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;
@@ -379,6 +460,7 @@ 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;
@@ -396,6 +478,7 @@ 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>());
         }
     };
 
@@ -423,7 +506,8 @@ 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

include/kpn/submit_gate.hpp

@@ -0,0 +1,101 @@
+#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

include/kpn/traits.hpp

@@ -97,4 +97,41 @@ 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

include/kpn/variant_node.hpp

@@ -162,6 +162,7 @@ public:
 
     // ── INode ─────────────────────────────────────────────────────────────────
 
+    void prepare() override { node_.prepare(); }
     void start() override { node_.start(); }
     void stop()  override { node_.stop(); }
     bool running() const override { return node_.running(); }

include/kpn/web_debug.hpp

@@ -65,6 +65,11 @@ 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\":[";

tests/CMakeLists.txt

@@ -34,7 +34,9 @@ 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

tests/test_backpressure_deadlock.cpp

@@ -0,0 +1,421 @@
+// 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);
+}

tests/test_channel.cpp

@@ -1,3 +1,4 @@
+#include <string>
 #include <catch2/catch_test_macros.hpp>
 #include <catch2/catch_approx.hpp>
 #include <kpn/channel.hpp>
@@ -238,3 +239,87 @@ 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 counted as a drop", "[channel][sentinel]") {
+    // Visibility matters more here than for a dropped value: the refusal means
+    // a control token went nowhere, and the only alternative to a counter is
+    // for it to vanish.
+    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 + 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);
+}

tests/test_channel_stress.cpp

@@ -19,6 +19,7 @@
 // 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>
@@ -279,3 +280,51 @@ 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);
+}

tests/test_network.cpp

@@ -1,6 +1,10 @@
 #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;
@@ -54,3 +58,93 @@ 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);
+}

tests/test_pool_node.cpp

@@ -243,7 +243,13 @@ TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") {
 
 // ── Overflow callback ─────────────────────────────────────────────────────────
 
-TEST_CASE("pool node overflow callback fires on full output channel", "[pool_node][overflow]") {
+// NOTE: output overflow is no longer reachable on the data path. A node whose
+// 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();
 
@@ -264,10 +270,13 @@ TEST_CASE("pool node overflow callback fires on full output channel", "[pool_nod
     node.stop();
     pool->stop();
 
-    REQUIRE(overflow_count.load() > 0);
+    // Parked, not overflowed: the value is still owned by the node.
+    REQUIRE(overflow_count.load() == 0);
+    // And it was never handed downstream, so nothing was lost or duplicated.
+    REQUIRE(full_ch.size() == 1);
 }
 
-TEST_CASE("pool node overflow callback is independent per instance", "[pool_node][overflow]") {
+TEST_CASE("parking is per node, not shared", "[pool_node][overflow]") {
     auto pool = std::make_shared<ThreadPool>(2);
     pool->start();
 
@@ -296,7 +305,10 @@ TEST_CASE("pool node overflow callback is independent per instance", "[pool_node
     nodeB.stop();
     pool->stop();
 
-    REQUIRE(a_overflows.load() > 0);
+    // Neither overflows now: A parks on its full output, B runs normally. The
+    // 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);
 }
 
@@ -412,7 +424,9 @@ TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]")
     node.stop();
     pool->stop();
 
-    REQUIRE(net_overflows.load() > 0);
+    // Parking replaced overflow on the data path, so the network callback no
+    // 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]") {
@@ -459,6 +473,250 @@ TEST_CASE("per-node and network overflow callbacks both fire independently", "[p
     node.stop();
     pool->stop();
 
-    REQUIRE(per_node.load() > 0);
+    // Both zero now: the node parks rather than overflowing. The case still
-    REQUIRE(network.load() > 0);
+    // guards that the two callbacks are wired independently.
+    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();
 }

tests/test_scheduler.cpp

@@ -227,3 +227,116 @@ 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);
+}

tests/test_shared_resource.cpp

@@ -237,3 +237,51 @@ 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());
+}

tests/test_static_network.cpp

@@ -271,3 +271,164 @@ 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);
+}

tests/test_submit_gate.cpp

@@ -0,0 +1,173 @@
+// 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());
+}