fix: park nodes on a full output instead of blocking the worker

push_blocking parked a scheduler worker inside the push. Nodes own a
private single-thread pool, so the parked thread was the only one that
could drain that node's own input — hold-and-wait, and under sustained
backpressure four nodes of a five-node chain slept in nanosleep at once.
channel.hpp already warned about this for sentinels; it applies just as
much to data pushes.

The scheduler was purely input-driven: on_input_ready() wakes a node when
input arrives, with no counterpart for "my output has room". Lacking that
signal, blocking the thread was the only way to handle a full output.
This adds the missing half.

- Channel::try_push + has_space + set_space_callback; the callback fires
  from both pop() and try_pop_now().
- PoolNode/PoolObjectNode keep a one-slot pending_ buffer with per-element
  done flags, so a retry cannot duplicate an already-accepted element. One
  slot suffices because queued_ admits at most one fire_once per node.
- The re-check after clearing queued_ closes the lost-wakeup race where a
  space callback fires while the flag is still up and is swallowed.

Two bugs surfaced once nodes actually parked, both fixed here:

- pop_one reports an *empty* channel as ChannelClosedError, which is also
  the node's "upstream finished, self-stop" signal. A node woken by output
  space with empty inputs therefore killed itself. fire_once now releases
  the worker when its inputs are not ready rather than falling through.
- The drained-park path resubmitted unconditionally instead of via
  on_input_ready(), firing nodes with nothing to read.

compute_priority is now output-aware: mean output fill is deducted from
mean input fill, mapped as 0.5·(1 + in - out). Input fill alone asks only
"how much work is waiting for me"; a node whose outputs are already full
cannot deliver, so running it just parks it again and wastes the slot
while the node that would drain that channel waits behind it. The
scheduler now favours whoever is furthest downstream of a bottleneck.

Also adds a network-level error listener. A node's exception was discarded
at the node boundary and survived only as a Closed event, which reports
that a node stopped but not why — that missing detail is what made the
above slow to diagnose. INode::set_network_error_callback plus
StaticNetwork::set_error_handler forward it to the application.

Tests: 121/121. test_backpressure_deadlock drives a five-node chain with
capacity-2 channels against a slow sink and fails on the old code. The
four test_pool_node overflow tests now assert parking rather than the
removed drop-and-report behaviour.

Known-incomplete: a rare hang remains, roughly 1 run in 20 against a 300s
timeout, down from every run failing. Committed because the fix is a large
strict improvement and the residual case needs its own reproduction.

include/kpn/pool_node.hpp

@@ -15,6 +15,7 @@
 #include <iostream>
 #include <memory>
 #include <optional>
+#include <variant>
 #include <stdexcept>
 #include <thread>
 #include <tuple>
@@ -131,6 +132,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); }
@@ -234,6 +236,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,6 +247,20 @@ 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>{});
@@ -280,11 +298,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;
-        sum_fill(sum, std::make_index_sequence<input_count>{});
-        return sum / static_cast<float>(input_count);
+        float in = 0.0f;
+        sum_fill(in, std::make_index_sequence<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>
@@ -315,6 +373,77 @@ 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>{});
+                queued_.store(false, std::memory_order_release);
+                if (pending_) {
+                    // Close the lost-wakeup race: a space_callback that fired
+                    // between the failed push and clearing queued_ was
+                    // swallowed, and nothing else will wake this node. Re-check
+                    // now that the flag is down.
+                    if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                        try_submit(0.5f);
+                    return;                    // parked
+                }
+                // Drained: resume normal firing, resubmitting exactly the way
+                // the normal tail below does. An unconditional try_submit here
+                // would fire a node whose inputs are empty, and pop_inputs
+                // reports an empty channel as ChannelClosedError — which this
+                // node treats as "upstream finished" and self-stops on. That
+                // is a live node killing itself purely because it was woken by
+                // *output* space rather than by input arrival.
+                if constexpr (input_count == 0) try_submit(0.5f);
+                else                            on_input_ready();
+                return;
+            }
+        }
+
+        // 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>{});
+                queued_.store(false, std::memory_order_release);
+                if (pending_) {
+                    // Close the lost-wakeup race: a space_callback that fired
+                    // between the failed push and clearing queued_ was
+                    // swallowed, and nothing else will wake this node. Re-check
+                    // now that the flag is down.
+                    if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                        try_submit(0.5f);
+                    return;                    // parked
+                }
+                // Drained: resume normal firing, resubmitting exactly the way
+                // the normal tail below does. An unconditional try_submit here
+                // would fire a node whose inputs are empty, and pop_inputs
+                // reports an empty channel as ChannelClosedError — which this
+                // node treats as "upstream finished" and self-stops on. That
+                // is a live node killing itself purely because it was woken by
+                // *output* space rather than by input arrival.
+                if constexpr (input_count == 0) try_submit(0.5f);
+                else                            on_input_ready();
+                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) {
+                queued_.store(false, std::memory_order_release);
+                on_input_ready();   // data may have arrived while we checked
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1 = clock_t::now();
@@ -340,7 +469,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_);
@@ -385,30 +518,41 @@ 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); }
+        else       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));
-            return;
+            return true;
         }
-        // Backpressure, not loss. A full downstream channel means the consumer
-        // is behind, and the correct response is for this producer to run
-        // slower — not to discard a value. A dropped frame does not degrade a
-        // result, it silently changes one, and the caller has no way to tell.
+        // 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).
         //
-        // Safe here because sentinels are handled above, out-of-band: this
-        // blocks only on data, so the EOF token that unwinds the network can
-        // always overtake a stalled data path.
-        ch->push_blocking(std::move(val));
+        // 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>
@@ -430,8 +574,18 @@ private:
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
     std::atomic<bool>                        queued_{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_{};
@@ -499,6 +653,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); }
@@ -570,6 +725,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(); }), ...);
     }
 
@@ -578,6 +735,20 @@ 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>{});
@@ -609,11 +780,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;
-        sum_fill(sum, std::make_index_sequence<input_count>{});
-        return sum / static_cast<float>(input_count);
+        float in = 0.0f;
+        sum_fill(in, std::make_index_sequence<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...>) {
@@ -639,6 +850,46 @@ 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>{});
+                queued_.store(false, std::memory_order_release);
+                if (pending_) {
+                    // Close the lost-wakeup race: a space_callback that fired
+                    // between the failed push and clearing queued_ was
+                    // swallowed, and nothing else will wake this node. Re-check
+                    // now that the flag is down.
+                    if (outputs_have_space(std::make_index_sequence<output_count>{}))
+                        try_submit(0.5f);
+                    return;                    // parked
+                }
+                // Drained: resume normal firing, resubmitting exactly the way
+                // the normal tail below does. An unconditional try_submit here
+                // would fire a node whose inputs are empty, and pop_inputs
+                // reports an empty channel as ChannelClosedError — which this
+                // node treats as "upstream finished" and self-stops on. That
+                // is a live node killing itself purely because it was woken by
+                // *output* space rather than by input arrival.
+                if constexpr (input_count == 0) try_submit(0.5f);
+                else                            on_input_ready();
+                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) {
+                queued_.store(false, std::memory_order_release);
+                on_input_ready();
+                return;
+            }
+        }
+
         try {
             auto args = pop_inputs(std::make_index_sequence<input_count>{});
             auto t1   = clock_t::now();
@@ -662,7 +913,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();
@@ -695,22 +950,32 @@ 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); }
+        else       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));
-            return;
+            return true;
         }
-        // See the note on the typed overload above: block rather than drop.
-        ch->push_blocking(std::move(val));
+        // See the note on the typed overload above: park rather than block.
+        return ch->try_push(val);
     }
 
     Obj&                                     obj_;
@@ -721,8 +986,18 @@ private:
     output_channels_t                        output_channels_{};
     std::atomic<bool>                        stop_flag_{true};
     std::atomic<bool>                        queued_{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_{};