fix: the sentinel slot holds one token and refuses a second

push_sentinel wrote eof_value_ unconditionally. Offering a second token
before the first was taken did two wrong things at once.

It lost the first silently, and a lost EOF is not a lost frame — it is the
token every downstream node is waiting for in order to shut down, so losing
it wedges the pipeline.

And it wrote the storage while the consumer could be moving the previous
value out of it. I expected that to be a stale read; ThreadSanitizer shows
it is worse. On the shared_ptr storage that non-trivial types use, the
racing write tears the refcount, and the stress case added here reports
heap-use-after-free in extract() alongside the data race.

try_push_sentinel now refuses when the slot is occupied, which turns the
slot into 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 it false is what licenses the write. Refusal is
recorded as a drop, and PoolNode reports it through the overflow event
callback, because a refused control token going unnoticed is the failure
this commit exists to stop.

Refusing rather than queueing is deliberate. Two control tokens on one
channel means the stream ended twice, which is a caller protocol error and
not backpressure; parking and retrying would spin against a slot only the
consumer can free, and there is no sensible second value to deliver after
the end of a stream. The non-consuming try_push_sentinel exists so a refused
token is still the caller's to report — the consuming push_sentinel cannot
offer that, since the value has already been moved into its parameter.

Single-shot EOF is what every current caller does, so this is latent for
them today. It stops being latent the moment a pipeline is reused for a
second input, which is what the persistent-pipeline work in 4b6e498 sets up.

Verified in both directions under -DKPN_SANITIZER=thread: the new contended
case reports three data races and a heap-use-after-free against the old
overwrite, and is clean with the handshake. Full suite 137/137, TSan clean
across unit and stress suites.

.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,70 @@
 
 #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 (;;) {
+        if (ch->try_push(val)) {
+            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, purely so the
+            // channel's own stats record the loss (drop if it is disabled,
+            // overflow if it is merely full). 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 +184,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 +324,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,42 @@ 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_;
+    }
+
+    /// Non-blocking, lossless push. Returns false when the ring is full, having
+    /// changed nothing — the caller keeps the value and retries when woken.
+    bool try_push(T& value) {
+        if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); return true; }
+        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 false;
+
+        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 true;
+    }
+
     // 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 +226,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 +265,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 +327,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 +350,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;
     }
 
@@ -363,6 +448,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 ────────────────────────────────────────────────

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,76 @@ 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;
+                if (out_channels_[i]->try_push(*pending[i])) {
+                    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 +198,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

@@ -134,6 +134,9 @@ public:
 
     void start() override {
         start_time_ = clock_t::now();
+        // 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();

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,13 +81,37 @@ 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 {
@@ -131,6 +128,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 +152,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 +235,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 +246,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 +299,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 +354,67 @@ 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;
+        if (gate_.claim())
-        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
             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);
+    }
+
     void fire_once() {
         if (stop_flag_.load(std::memory_order_relaxed)) {
-            queued_.store(false, std::memory_order_release);
+            gate_.force_idle();
             return;
         }
 
@@ -315,6 +424,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 +473,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 +488,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 +502,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 +520,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 +540,56 @@ 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.
+        return ch->try_push(val);
     }
 
     template<std::size_t I>
@@ -426,9 +610,26 @@ 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};
+
+    /// 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,13 +680,22 @@ 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 {
@@ -496,6 +706,7 @@ public:
     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 +730,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 +781,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 +791,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 +838,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 +892,65 @@ 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;
+        if (gate_.claim())
-        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
             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);
+    }
+
     void fire_once() {
         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 +958,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 +1003,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 +1018,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 +1031,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 +1045,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 +1065,47 @@ 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.
+        return ch->try_push(val);
     }
 
     Obj&                                     obj_;
@@ -721,9 +1115,26 @@ 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};
+
+    /// 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/static_network.hpp

@@ -122,6 +122,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
@@ -181,6 +194,14 @@ public:
 
     void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
 
+    /// 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.
@@ -276,6 +297,7 @@ private:
     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};

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,62 @@ 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");
+}

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_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,198 @@ 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();
 }

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());
+}