fix: Network::set_error_handler must actually deliver the handler

The handler was stored in a member and never read. A node's exception was
discarded at the node boundary and the only surviving evidence was a Closed
event, which reports that a node stopped but not why — the difference between
a diagnosis and a guess. StaticNetwork has always wired this; Network accepted
the handler and silently dropped it, which is worse than not offering the
setter, because the caller believes they have a listener.

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

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

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

include/kpn/branch.hpp

@@ -52,16 +52,24 @@ bool deliver_one(Channel<T>* ch, T& val, const std::atomic<bool>& stop_flag,
     }
     const auto park_from = clock_t::now();
     for (;;) {
-        if (ch->try_push(val)) {
+        switch (ch->try_push(val)) {
-            parked = duration_t(clock_t::now() - park_from);
+            case Channel<T>::PushResult::Taken:
-            return true;
+                parked = duration_t(clock_t::now() - park_from);
+                return true;
+            case Channel<T>::PushResult::Closed:
+                // Nobody is listening any more; the channel has recorded the
+                // drop. Retrying would spin until teardown noticed.
+                parked = duration_t(clock_t::now() - park_from);
+                return false;
+            case Channel<T>::PushResult::Full:
+                break;                       // fall through to the retry logic
         }
         if (stop_flag.load(std::memory_order_relaxed)) {
-            // Teardown with work in hand. One last throwing push, purely so the
+            // Teardown with work in hand and the output still full. One last
-            // channel's own stats record the loss (drop if it is disabled,
+            // throwing push, purely so the channel's own stats record the
-            // overflow if it is merely full). The point of the lossless path is
+            // overflow — the point of the lossless path is that a loss is never
-            // that a loss is never invisible, and a silent return here would
+            // invisible, and a silent return here would reintroduce exactly the
-            // reintroduce exactly the hole this function exists to close.
+            // hole this function exists to close.
             try { ch->push(std::move(val)); }
             catch (const ChannelOverflowError&) {}
             parked = duration_t(clock_t::now() - park_from);

include/kpn/channel.hpp

@@ -165,13 +165,26 @@ public:
                head_.load(std::memory_order_acquire) < capacity_;
     }
 
-    /// Non-blocking, lossless push. Returns false when the ring is full, having
+    /// Outcome of a non-blocking push.
+    ///
+    /// try_push used to return bool, and returned *true* for a closed channel —
+    /// so "delivered" and "discarded because nobody is listening" were the same
+    /// answer. Both mean "stop trying", which is why the callers were correct,
+    /// but neither they nor the producer's own accounting could tell a value
+    /// that arrived from one that was thrown away. Only the channel's drop
+    /// counter knew.
+    enum class PushResult { Taken, Full, Closed };
+
+    /// Non-blocking, lossless push. Returns Full when the ring is full, having
     /// changed nothing — the caller keeps the value and retries when woken.
-    bool try_push(T& value) {
+    PushResult try_push(T& value) {
-        if (!accepting_.load(std::memory_order_acquire)) { stats_.record_drop(); return true; }
+        if (!accepting_.load(std::memory_order_acquire)) {
+            stats_.record_drop();
+            return PushResult::Closed;
+        }
         const std::size_t t = tail_.load(std::memory_order_relaxed);
         const std::size_t h = head_.load(std::memory_order_acquire);
-        if (t - h >= capacity_) return false;
+        if (t - h >= capacity_) return PushResult::Full;
 
         const std::size_t data_bytes = ChannelDataSize<T>::bytes(value);
         const bool was_empty = (t == h);
@@ -181,7 +194,7 @@ public:
         wake_.fetch_add(1, std::memory_order_release);
         wake_.notify_one();
         if (was_empty && push_callback_) push_callback_();
-        return true;
+        return PushResult::Taken;
     }
 
     // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to
@@ -382,9 +395,15 @@ public:
 
     // Ring occupancy, derived lazily from indices — no separate counter on the
     // hot path. Excludes any out-of-band sentinel (that lives outside the ring).
+    // head_ is loaded first, deliberately. Both indices only ever increase, so
+    // reading head_ before tail_ can at worst under-report a concurrent push;
+    // the other order can read a head_ that has advanced past the tail_ already
+    // sampled, and the unsigned difference then wraps to ~2^64. A caller
+    // polling "is this channel empty yet" against that value never terminates.
     std::size_t size() const {
-        return tail_.load(std::memory_order_relaxed)
+        const std::size_t h = head_.load(std::memory_order_relaxed);
-             - head_.load(std::memory_order_relaxed);
+        const std::size_t t = tail_.load(std::memory_order_acquire);
+        return t - h;
     }
 
     // A pending out-of-band sentinel (EOF) counts as consumable work here even
@@ -401,8 +420,9 @@ public:
     const ChannelStats& stats() const { return stats_; }
 
     ChannelSnapshot snapshot(const std::string& name) const {
-        const std::size_t t = tail_.load(std::memory_order_relaxed);
+        // head_ before tail_, for the reason given on size().
         const std::size_t h = head_.load(std::memory_order_relaxed);
+        const std::size_t t = tail_.load(std::memory_order_acquire);
         return {
             name,
             capacity_,
@@ -437,6 +457,24 @@ private:
     // delivered after every value pushed before it.
     bool take_sentinel(T& out) {
         if (!has_eof_.load(std::memory_order_acquire)) return false;
+        // Re-check emptiness *after* observing has_eof_, not before.
+        //
+        // Callers check the ring is empty and then call this, but the producer
+        // can push a value and publish the sentinel in the window between those
+        // two steps — so the sentinel would be delivered with a real value still
+        // queued behind it, breaking the "sentinel is strictly last" contract
+        // that downstream teardown depends on. a0c4bf5 closed the variant where
+        // the caller's emptiness check used a stale tail_ snapshot; this is the
+        // one where the check is fresh but simply too early.
+        //
+        // Checking here is what makes it sound: the producer publishes the
+        // sentinel with a release store *after* its ring pushes, so a consumer
+        // that has observed has_eof_ has also observed every tail_ advance
+        // before it. If the ring is non-empty now, those values genuinely
+        // precede the sentinel and must be delivered first.
+        if (head_.load(std::memory_order_relaxed)
+                != tail_.load(std::memory_order_acquire))
+            return false;
         out = extract(std::move(eof_value_));
         has_eof_.store(false, std::memory_order_release);
         stats_.record_pop();

include/kpn/diagnostics.hpp

@@ -220,6 +220,11 @@ struct ResourceSnapshot {
 struct IResourceProbe {
     virtual ~IResourceProbe() = default;
     virtual ResourceSnapshot snapshot(const std::string& name) const = 0;
+
+    /// Release every thread waiting for the resource, so teardown is not held
+    /// up by one. A network calls this on the resources registered with it when
+    /// it halts; default no-op for probes with nothing to wake.
+    virtual void close() {}
 };
 
 } // namespace kpn

include/kpn/fanout.hpp

@@ -160,7 +160,10 @@ private:
         for (;;) {
             for (std::size_t i = 0; i < N; ++i) {
                 if (!pending[i]) continue;
-                if (out_channels_[i]->try_push(*pending[i])) {
+                // Taken or Closed both mean "stop trying" — delivered, or gone
+                // with the drop recorded. Only Full is worth another pass.
+                if (out_channels_[i]->try_push(*pending[i])
+                        != Channel<T>::PushResult::Full) {
                     pending[i].reset();
                     --outstanding;
                 }

include/kpn/network.hpp

@@ -8,8 +8,10 @@
 #include <memory>
 #endif
 
+#include <condition_variable>
 #include <functional>
 #include <iomanip>
+#include <mutex>
 #include <iostream>
 #include <map>
 #include <set>
@@ -39,8 +41,15 @@ public:
 
 class Network : public INode {
 public:
-    using ErrorHandler =
+    /// Application-level error listener. Receives the exception any node's
-        std::function<void(std::string_view node_name, std::exception_ptr)>;
+    /// function throws, after that node's own handler (if any) declined it.
+    /// Return true to skip the failed invocation and keep the node running,
+    /// false to let it stop.
+    ///
+    /// Same type as StaticNetwork's, deliberately: this used to be a void
+    /// signature, which could not express the keep-running decision and, more
+    /// to the point, was never delivered anywhere.
+    using ErrorHandler = NodeErrorHandler;
     using DiagnosticsHandler =
         std::function<void(const std::vector<NodeSnapshot>&,
                            const std::vector<ChannelSnapshot>&)>;
@@ -91,6 +100,7 @@ public:
                             + " → " + dst_name + ":" + std::to_string(DstIdx);
         channel_probes_.push_back(
             std::make_unique<ChannelProbe<out_t>>(in_ch, ch_name));
+        channel_src_names_.push_back(src_name);
 
         adj_[src_name].push_back(dst_name);
         return *this;
@@ -134,6 +144,14 @@ public:
 
     void start() override {
         start_time_ = clock_t::now();
+        // Deliver the listener to the nodes. Without this the handler was
+        // stored and never read: a node's exception was discarded at the node
+        // boundary and the only surviving evidence was a Closed event, which
+        // says a node stopped but not why. StaticNetwork has always done this;
+        // Network accepted the handler and silently dropped it.
+        if (error_handler_)
+            for (auto& name : topo_)
+                nodes_.at(name)->set_network_error_callback(error_handler_);
         // Callbacks first, everywhere, before anything runs — see INode::prepare.
         for (auto& name : topo_)
             nodes_.at(name)->prepare();
@@ -213,6 +231,11 @@ public:
         watchdog_interval_ = interval;
     }
 
+    /// How long shutdown() waits for one node's outputs to drain before giving
+    /// up on them and stopping the next layer anyway.
+    void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
+
+    /// Must be called before start(); the handler is delivered to nodes there.
     void set_error_handler(ErrorHandler h)             { error_handler_  = std::move(h); }
     void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_  = std::move(h); }
     void set_event_handler(EventHandler h)             { event_handler_  = std::move(h); }
@@ -370,19 +393,46 @@ private:
         return true;
     }
 
-    void drain_output_channels(const std::string& /*name*/) const {
+    /// Wait for the channels fed by `name` to empty, or give up.
-        // Poll all channel probes until none report non-zero fill.
+    ///
-        // A short sleep prevents busy-spin; 1 ms is fine for drain purposes.
+    /// This took a node name and ignored it, polling *every* channel in the
-        bool any_full = true;
+    /// graph instead — so shutdown() waited for the whole network to be idle
-        while (any_full) {
+    /// before stopping each successive layer. With no deadline either, anything
-            any_full = false;
+    /// wedged downstream turned a graceful shutdown into the hang it exists to
-            for (auto& probe : channel_probes_) {
+    /// avoid.
-                auto snap = probe->snapshot();
+    ///
-                if (snap.current_fill > 0) { any_full = true; break; }
+    /// Two bounds, because they fail differently. The deadline covers a
-            }
+    /// consumer that has stopped consuming, where fill never changes and
-            if (any_full)
+    /// waiting cannot help. The no-progress counter covers one that is merely
-                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    /// slow: it keeps waiting while the queue is shrinking, so a slow drain is
+    /// not cut short just for taking a while.
+    void drain_output_channels(const std::string& name) const {
+        const auto deadline = clock_t::now() + drain_timeout_;
+        std::size_t last_fill = static_cast<std::size_t>(-1);
+        int         stalls    = 0;
+
+        auto fill_of = [&] {
+            std::size_t fill = 0;
+            for (std::size_t i = 0; i < channel_probes_.size(); ++i)
+                if (channel_src_names_[i] == name)
+                    fill += channel_probes_[i]->snapshot().current_fill;
+            return fill;
+        };
+
+        for (;;) {
+            const std::size_t fill = fill_of();
+            if (fill == 0) return;
+            if (fill >= last_fill) { if (++stalls > 100) break; }
+            else                   { stalls = 0; }
+            last_fill = fill;
+            if (clock_t::now() >= deadline) break;
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
+
+        if (const std::size_t left = fill_of())
+            std::cerr << "[kpn] shutdown: '" << name << "' still has " << left
+                      << " queued item(s) its consumer did not take; "
+                         "they are discarded\n";
     }
 
     // ── Cycle detection / topological sort ───────────────────────────────────
@@ -401,9 +451,20 @@ private:
 
     void start_watchdog() {
         watchdog_ = std::jthread([this](std::stop_token tok) {
+            // Interruptible wait, not sleep_for. request_stop() cannot wake a
+            // sleeping thread, so stop_watchdog()'s join blocked for up to a
+            // full interval — three seconds by default, and unbounded for
+            // anyone who set a long one to keep the periodic report quiet.
+            // Every teardown paid it.
+            std::mutex                  m;
+            std::condition_variable_any cv;
             while (!tok.stop_requested()) {
-                std::this_thread::sleep_for(watchdog_interval_);
+                {
-                if (tok.stop_requested()) break;
+                    std::unique_lock lk(m);
+                    if (cv.wait_for(lk, tok, watchdog_interval_,
+                                    [&tok] { return tok.stop_requested(); }))
+                        break;
+                }
 
                 auto s = collect_snapshots();
                 check_hung_nodes();
@@ -447,6 +508,10 @@ private:
     std::map<std::string, std::string>                     exposed_outputs_;
     std::set<std::pair<std::string, std::size_t>>          connected_outputs_;
     std::vector<std::unique_ptr<IChannelProbe>>            channel_probes_;
+    /// Name of the node feeding each probe, parallel to channel_probes_.
+    /// shutdown() drains a node's own outputs, so it has to know which they are.
+    std::vector<std::string>                               channel_src_names_;
+    std::chrono::milliseconds                              drain_timeout_{5000};
     std::vector<std::pair<std::string, IPoolProbe*>>       pool_probes_;
     ErrorHandler                                           error_handler_;
     DiagnosticsHandler                                     diag_handler_;

include/kpn/pool_node.hpp

@@ -117,9 +117,7 @@ public:
     void stop() override {
         stop_flag_.store(true, std::memory_order_seq_cst);
         disable_inputs(std::make_index_sequence<input_count>{});
-        // fire_once() observes stop_flag_ and will not resubmit.
+        await_quiescence();
-        // We do not wait for an in-flight fire_once() to complete here;
-        // callers that need that guarantee should call scheduler_->drain() first.
     }
 
     bool running() const override {
@@ -364,6 +362,12 @@ private:
     /// 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) {
+        // A stopped node must not claim the gate. The scheduler now refuses
+        // submissions after its pool stops, so the submit itself is safe — but
+        // claiming and never releasing would leave the gate held, and a restart
+        // would then have to clear it. start() does, but relying on that makes
+        // the invariant depend on a distant statement.
+        if (stop_flag_.load(std::memory_order_relaxed)) return;
         if (gate_.claim())
             scheduler_->submit([this] { fire_once(); }, priority);
     }
@@ -412,7 +416,55 @@ private:
         else if (want_more)     try_submit(prio);
     }
 
+    /// Block until no firing of this node is in flight or queued.
+    ///
+    /// stop() used to set the flag and return, leaving an executing fire_once
+    /// touching input_channels_, stats_ and pending_ while the caller went on
+    /// to destroy them. For a node with a private pool that was survivable by
+    /// accident — Node::stop() calls pool->stop(), which joins — but a node
+    /// sharing a pool had nothing joining it at all, so ~PoolNode raced its own
+    /// members. The old comment said callers wanting the guarantee should call
+    /// scheduler_->drain() first; a destructor cannot, and the default should
+    /// not be a use-after-free.
+    ///
+    /// The gate is exactly the right thing to wait on: it is claimed for the
+    /// whole of a firing and released as the last act of one. A queued but
+    /// unstarted firing also holds it, and will run, observe stop_flag_ and
+    /// release — which is why the pool must still be running when this is
+    /// called. That is already the documented order (stop nodes, then the
+    /// pool), and Node/ObjectNode do it that way.
+    ///
+    /// Bounded, because a node function that never returns must not turn
+    /// teardown into a hang; and skipped entirely when called from the firing
+    /// thread itself, since an error handler that stops its own node would
+    /// otherwise wait for a firing that is waiting for it.
+    void await_quiescence() {
+        if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id())
+            return;
+        const auto deadline = clock_t::now() + std::chrono::seconds(5);
+        while (gate_.queued()) {
+            if (clock_t::now() >= deadline) {
+                std::cerr << "[kpn] stop: node '" << name_
+                          << "' still had work in flight after 5 s; "
+                             "continuing without it\n";
+                return;
+            }
+            std::this_thread::sleep_for(std::chrono::microseconds(50));
+        }
+    }
+
+    /// Marks fire_once's thread for the duration of a firing, so await_quiescence
+    /// can tell a re-entrant stop() from an external one.
+    struct FiringMark {
+        std::atomic<std::thread::id>& slot;
+        explicit FiringMark(std::atomic<std::thread::id>& s) : slot(s) {
+            slot.store(std::this_thread::get_id(), std::memory_order_release);
+        }
+        ~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); }
+    };
+
     void fire_once() {
+        FiringMark mark(firing_thread_);
         if (stop_flag_.load(std::memory_order_relaxed)) {
             gate_.force_idle();
             return;
@@ -589,7 +641,10 @@ private:
         // 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);
+        // Closed counts as "stop trying", not as delivered: the value is gone
+        // and the channel has recorded the drop. Only Full means park and retry.
+        return ch->try_push(val) != Channel<std::tuple_element_t<I, return_tuple>>
+                                        ::PushResult::Full;
     }
 
     template<std::size_t I>
@@ -618,6 +673,9 @@ private:
     /// cleared: the callbacks capture `this` and stay valid across a restart, so
     /// re-registering them would be a pointless write to a live channel.
     bool                                     prepared_{false};
+    /// Thread currently inside fire_once, or a default id when none is.
+    /// See await_quiescence.
+    std::atomic<std::thread::id>             firing_thread_{};
 
     /// The hidden one-slot output buffer (see push_outputs). Holding the value
     /// here is what lets a node stop running without dropping it or occupying a
@@ -701,6 +759,7 @@ public:
     void stop() override {
         stop_flag_.store(true, std::memory_order_seq_cst);
         disable_inputs(std::make_index_sequence<input_count>{});
+        await_quiescence();
     }
 
     bool running() const override { return !stop_flag_.load(std::memory_order_relaxed); }
@@ -902,6 +961,12 @@ private:
     /// 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) {
+        // A stopped node must not claim the gate. The scheduler now refuses
+        // submissions after its pool stops, so the submit itself is safe — but
+        // claiming and never releasing would leave the gate held, and a restart
+        // would then have to clear it. start() does, but relying on that makes
+        // the invariant depend on a distant statement.
+        if (stop_flag_.load(std::memory_order_relaxed)) return;
         if (gate_.claim())
             scheduler_->submit([this] { fire_once(); }, priority);
     }
@@ -948,7 +1013,55 @@ private:
         else if (want_more)     try_submit(prio);
     }
 
+    /// Block until no firing of this node is in flight or queued.
+    ///
+    /// stop() used to set the flag and return, leaving an executing fire_once
+    /// touching input_channels_, stats_ and pending_ while the caller went on
+    /// to destroy them. For a node with a private pool that was survivable by
+    /// accident — Node::stop() calls pool->stop(), which joins — but a node
+    /// sharing a pool had nothing joining it at all, so ~PoolNode raced its own
+    /// members. The old comment said callers wanting the guarantee should call
+    /// scheduler_->drain() first; a destructor cannot, and the default should
+    /// not be a use-after-free.
+    ///
+    /// The gate is exactly the right thing to wait on: it is claimed for the
+    /// whole of a firing and released as the last act of one. A queued but
+    /// unstarted firing also holds it, and will run, observe stop_flag_ and
+    /// release — which is why the pool must still be running when this is
+    /// called. That is already the documented order (stop nodes, then the
+    /// pool), and Node/ObjectNode do it that way.
+    ///
+    /// Bounded, because a node function that never returns must not turn
+    /// teardown into a hang; and skipped entirely when called from the firing
+    /// thread itself, since an error handler that stops its own node would
+    /// otherwise wait for a firing that is waiting for it.
+    void await_quiescence() {
+        if (firing_thread_.load(std::memory_order_acquire) == std::this_thread::get_id())
+            return;
+        const auto deadline = clock_t::now() + std::chrono::seconds(5);
+        while (gate_.queued()) {
+            if (clock_t::now() >= deadline) {
+                std::cerr << "[kpn] stop: node '" << name_
+                          << "' still had work in flight after 5 s; "
+                             "continuing without it\n";
+                return;
+            }
+            std::this_thread::sleep_for(std::chrono::microseconds(50));
+        }
+    }
+
+    /// Marks fire_once's thread for the duration of a firing, so await_quiescence
+    /// can tell a re-entrant stop() from an external one.
+    struct FiringMark {
+        std::atomic<std::thread::id>& slot;
+        explicit FiringMark(std::atomic<std::thread::id>& s) : slot(s) {
+            slot.store(std::this_thread::get_id(), std::memory_order_release);
+        }
+        ~FiringMark() { slot.store(std::thread::id{}, std::memory_order_release); }
+    };
+
     void fire_once() {
+        FiringMark mark(firing_thread_);
         if (stop_flag_.load(std::memory_order_relaxed)) {
             gate_.force_idle();
             return;
@@ -1105,7 +1218,10 @@ private:
             return true;
         }
         // See the note on the typed overload above: park rather than block.
-        return ch->try_push(val);
+        // Closed counts as "stop trying", not as delivered: the value is gone
+        // and the channel has recorded the drop. Only Full means park and retry.
+        return ch->try_push(val) != Channel<std::tuple_element_t<I, return_tuple>>
+                                        ::PushResult::Full;
     }
 
     Obj&                                     obj_;
@@ -1123,6 +1239,9 @@ private:
     /// cleared: the callbacks capture `this` and stay valid across a restart, so
     /// re-registering them would be a pointless write to a live channel.
     bool                                     prepared_{false};
+    /// Thread currently inside fire_once, or a default id when none is.
+    /// See await_quiescence.
+    std::atomic<std::thread::id>             firing_thread_{};
 
     /// The hidden one-slot output buffer (see push_outputs). Holding the value
     /// here is what lets a node stop running without dropping it or occupying a

include/kpn/scheduler.hpp

@@ -5,6 +5,7 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <shared_mutex>
 #include <optional>
 #include <queue>
 #include <thread>
@@ -62,18 +63,30 @@ public:
     }
 
     void stop() override {
-        stopped_.store(true, std::memory_order_seq_cst);
+        // Close the pool to new work before touching anything, and do it under
+        // the lifecycle lock so no submit() is midway through indexing queues_.
+        {
+            std::unique_lock lk(lifecycle_mx_);
+            stopped_.store(true, std::memory_order_seq_cst);
+        }
         for (auto& q : queues_) {
             std::lock_guard lock(q->mx);
             std::size_t discarded = q->pq.size();
             while (!q->pq.empty()) q->pq.pop();
             total_.fetch_sub(discarded, std::memory_order_relaxed);
+            queued_.fetch_sub(discarded, std::memory_order_relaxed);
         }
         // Lock cv_mx_ before notifying so the stop signal can't be lost in the
         // gap between a worker's predicate check and its wait() (see submit()).
         { std::lock_guard<std::mutex> lk(cv_mx_); }
         cv_.notify_all();
+        // Join without the lock: a worker's task may call submit(), which takes
+        // it shared, and holding it here would deadlock against that.
         for (auto& t : workers_) if (t.joinable()) t.join();
+        // Destroying the queues is what submit() must never race. By now
+        // stopped_ is published, so any submit() that acquires the lock after
+        // this point returns without touching them.
+        std::unique_lock lk(lifecycle_mx_);
         workers_.clear();
         queues_.clear();
     }
@@ -86,6 +99,22 @@ public:
     }
 
     void submit(std::function<void()> task, float priority = 0.5f) override {
+        // A submission can arrive after this pool has been stopped, and did so
+        // by an ordinary route: a node's space callback fires from whichever
+        // thread drained the channel, which belongs to the *consumer*. Stop the
+        // producer first — as a sources-first shutdown does — and the consumer
+        // keeps draining its backlog, firing the producer's space callback into
+        // a pool whose stop() has already run queues_.clear(). submit() then
+        // indexed an empty vector: a segfault, reproducible about 12 runs in 20.
+        //
+        // The shared lock is what makes the check meaningful. Reading stopped_
+        // alone leaves the window between the read and the indexing, which is
+        // precisely where stop() clears the vector.
+        std::shared_lock lk(lifecycle_mx_);
+        if (stopped_.load(std::memory_order_acquire) || queues_.empty()) {
+            rejected_.fetch_add(1, std::memory_order_relaxed);
+            return;
+        }
         std::size_t target = next_.fetch_add(1, std::memory_order_relaxed) % thread_count_;
         {
             std::lock_guard lock(queues_[target]->mx);
@@ -93,6 +122,7 @@ public:
                 {std::move(task), priority, seq_.fetch_add(1, std::memory_order_relaxed)});
         }
         total_.fetch_add(1, std::memory_order_relaxed);
+        queued_.fetch_add(1, std::memory_order_relaxed);
         submitted_.fetch_add(1, std::memory_order_relaxed);
         // Synchronize with worker_loop's predicate evaluation: taking cv_mx_
         // here guarantees a worker is either before its predicate check (and
@@ -105,15 +135,17 @@ public:
 
     std::size_t thread_count() const { return thread_count_; }
 
+    /// Submissions dropped because the pool was stopped. See rejected_.
+    uint64_t rejected() const { return rejected_.load(std::memory_order_relaxed); }
+
     // ── IPoolProbe ────────────────────────────────────────────────────────────
 
     PoolSnapshot snapshot(const std::string& name) const override {
         std::size_t a = active_.load(std::memory_order_relaxed);
-        std::size_t t = total_.load(std::memory_order_relaxed);
         return {
             name, thread_count_,
-            t > a ? t - a : 0,   // queued (approximate)
+            queued_.load(std::memory_order_relaxed),   // queued (exact)
-            a,                    // executing
+            a,                                          // executing
             submitted_.load(std::memory_order_relaxed),
             completed_.load(std::memory_order_relaxed),
         };
@@ -142,6 +174,7 @@ private:
         if (q.pq.empty()) return std::nullopt;
         auto fn = std::move(const_cast<Task&>(q.pq.top()).fn);
         q.pq.pop();
+        queued_.fetch_sub(1, std::memory_order_relaxed);
         return fn;
     }
 
@@ -182,10 +215,13 @@ private:
             std::unique_lock lock(cv_mx_);
             cv_.wait(lock, [this] {
                 return stopped_.load(std::memory_order_seq_cst)
-                    || total_.load(std::memory_order_relaxed) > 0;
+                    || queued_.load(std::memory_order_relaxed) > 0;
             });
+            // Exit on queued_, not total_: waiting for total_ to reach zero
+            // meant waiting for someone else's task to finish, which this
+            // worker cannot help with and would spin through until it did.
             if (stopped_.load(std::memory_order_seq_cst)
-                && total_.load(std::memory_order_relaxed) == 0)
+                && queued_.load(std::memory_order_relaxed) == 0)
                 return;
         }
     }
@@ -194,17 +230,36 @@ private:
     std::vector<std::unique_ptr<WorkerQueue>>  queues_;
     std::vector<std::thread>                   workers_;
 
+    /// Guards the lifetime of queues_/workers_ against a concurrent submit().
+    /// Shared by submit, exclusive by stop, so submissions still run in
+    /// parallel with each other.
+    mutable std::shared_mutex lifecycle_mx_;
+
     std::mutex              cv_mx_;
     std::condition_variable cv_;
     std::mutex              drain_mx_;
     std::condition_variable drain_cv_;
 
     std::atomic<bool>     stopped_{true};
-    std::atomic<size_t>   total_{0};      // queued + executing
+    std::atomic<size_t>   total_{0};      // queued + executing (drain() waits on this)
+    /// Queued only — never counts a task that is already executing.
+    ///
+    /// The wait predicate used total_, which includes running tasks, so while
+    /// any one task ran every *other* worker's predicate was true: wait()
+    /// returned instantly and the worker spun through try_pop / try_steal /
+    /// wait at full speed, try_lock-ing every peer queue on each pass. One slow
+    /// task therefore pinned every other core and contended the very mutexes
+    /// the working thread needed. Sleeping requires "no work is *waiting*",
+    /// which is this.
+    std::atomic<size_t>   queued_{0};     // waiting to run
     std::atomic<size_t>   active_{0};     // executing only (for snapshot)
     std::atomic<size_t>   next_{0};       // round-robin submit cursor
     std::atomic<uint64_t> seq_{0};        // tie-break for equal-priority tasks
     std::atomic<uint64_t> submitted_{0};
+    /// Submissions refused because the pool was already stopped. Not an error —
+    /// teardown races are expected — but silence here would hide a node that
+    /// keeps trying to run after its pool is gone.
+    std::atomic<uint64_t> rejected_{0};
     std::atomic<uint64_t> completed_{0};
 };
 

include/kpn/shared_resource.hpp

@@ -14,6 +14,18 @@ namespace kpn {
 
 template<typename T> class Channel; // forward declaration for acquire_balanced
 
+/// Thrown by a pending acquire() when the resource is closed underneath it.
+///
+/// acquire() blocks on a condition variable with no timeout and no stop
+/// condition, so a node parked there ignored teardown entirely: the worker
+/// never returned, the pool's join never completed, and shutdown hung on a
+/// resource nobody was going to release. Closing the resource turns that into
+/// an exception the node's normal error path already handles.
+class ResourceClosedError : public std::runtime_error {
+public:
+    ResourceClosedError() : std::runtime_error("shared resource closed") {}
+};
+
 // ── SharedResource ────────────────────────────────────────────────────────────
 //
 // Wraps an exclusive resource (e.g. an ONNX session, a CUDA stream) and
@@ -72,6 +84,7 @@ public:
     template<typename PriorityFn>
     Guard acquire(PriorityFn&& fn) {
         std::unique_lock lock(mutex_);
+        if (closed_) throw ResourceClosedError{};
         if (!held_) {
             held_ = true;
             acq_.fetch_add(1, std::memory_order_relaxed);
@@ -83,18 +96,46 @@ public:
         current_waiters_.store(waiters_.size(), std::memory_order_relaxed);
 
         auto t0 = w.wait_start;
-        w.cv.wait(lock, [&w] { return w.ready; });
+        // Woken either by release() handing over ownership, or by close()
+        // giving up on the wait entirely.
+        w.cv.wait(lock, [&w] { return w.ready || w.closed; });
 
         int64_t wait_us = std::chrono::duration_cast<std::chrono::microseconds>(
             clock_t::now() - t0).count();
         waiters_.erase(std::find(waiters_.begin(), waiters_.end(), &w));
         current_waiters_.store(waiters_.size(), std::memory_order_relaxed);
-        acq_.fetch_add(1, std::memory_order_relaxed);
         total_wait_us_.fetch_add(static_cast<uint64_t>(wait_us > 0 ? wait_us : 0),
                                  std::memory_order_relaxed);
+
+        // Closed without being handed ownership: no Guard, so nothing to
+        // release, and held_ is left exactly as close() found it.
+        if (!w.ready) throw ResourceClosedError{};
+
+        acq_.fetch_add(1, std::memory_order_relaxed);
         return Guard(this);
     }
 
+    /// Wake every waiter and refuse further acquisitions.
+    ///
+    /// Teardown is the whole point: a node parked in acquire() is not
+    /// observing stop flags, so without this the only way out is for whoever
+    /// holds the resource to release it — which, if that node is also being
+    /// stopped, may never happen. Idempotent, and safe to call from any thread.
+    void close() override {
+        std::lock_guard lock(mutex_);
+        closed_ = true;
+        for (Waiter* w : waiters_) {
+            w->closed = true;
+            w->cv.notify_one();
+        }
+    }
+
+    /// Reopen after a close(). For reuse across runs; not needed for teardown.
+    void reopen() {
+        std::lock_guard lock(mutex_);
+        closed_ = false;
+    }
+
     // Acquire with no priority (all waiters treated equally, order is fair-ish).
     Guard acquire() {
         return acquire([] { return 0.5f; });
@@ -132,7 +173,10 @@ public:
 private:
     void release() {
         std::unique_lock lock(mutex_);
-        if (waiters_.empty()) {
+        // Hand over only to a waiter that is still waiting. A closed one is on
+        // its way out and will not take ownership, so treating it as the next
+        // holder would leave held_ true with nobody holding it.
+        if (closed_ || waiters_.empty()) {
             held_ = false;
             return;
         }
@@ -162,7 +206,8 @@ private:
         std::function<float()>  priority_fn;
         clock_t::time_point     wait_start;
         std::condition_variable cv;
-        bool                    ready{false};
+        bool                    ready{false};   // handed ownership by release()
+        bool                    closed{false};  // woken by close() instead
 
         Waiter(std::function<float()> fn, clock_t::time_point t)
             : priority_fn(std::move(fn)), wait_start(t) {}
@@ -172,6 +217,7 @@ private:
 
     T                        resource_;
     bool                     held_{false};
+    bool                     closed_{false};
     mutable std::mutex       mutex_;
     std::vector<Waiter*>     waiters_;
     std::atomic<uint64_t>    acq_{0};

include/kpn/static_network.hpp

@@ -98,13 +98,15 @@ public:
                   std::vector<INode*>        fanout_ptrs,
                   std::vector<std::string>   user_node_names,
                   std::vector<std::string>   fanout_node_names,
-                  std::vector<std::unique_ptr<IChannelProbe>> channel_probes)
+                  std::vector<std::unique_ptr<IChannelProbe>> channel_probes,
+                  std::vector<std::string>   channel_src_names)
         : fanouts_(std::move(fanouts))
         , user_nodes_topo_(std::move(user_nodes_topo))
         , fanout_nodes_ptr_(std::move(fanout_ptrs))
         , user_node_names_(std::move(user_node_names))
         , fanout_node_names_(std::move(fanout_node_names))
         , channel_probes_(std::move(channel_probes))
+        , channel_src_names_(std::move(channel_src_names))
     {}
 
     ~StaticNetwork() override { stop(); }
@@ -158,6 +160,11 @@ public:
 #ifdef KPN_WEB_DEBUG
         if (web_server_) web_server_->stop();
 #endif
+        // Release anything parked on a shared resource first. A node blocked in
+        // acquire() is not watching stop flags, so stopping it would wait on a
+        // handover that may never come — its holder is being stopped too.
+        for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); }
+
         for (auto it = fanout_nodes_ptr_.rbegin(); it != fanout_nodes_ptr_.rend(); ++it)
             (*it)->stop();
         for (auto it = user_nodes_topo_.rbegin(); it != user_nodes_topo_.rend(); ++it)
@@ -171,11 +178,12 @@ public:
 #ifdef KPN_WEB_DEBUG
         if (web_server_) web_server_->stop();
 #endif
+        for (auto& [rname, probe] : resource_probes_) { (void)rname; probe->close(); }
         // user_nodes_topo_ is already in sources-first order.
         // Stop each node and drain its output channels before moving on.
-        for (auto* n : user_nodes_topo_) {
+        for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) {
-            n->stop();
+            user_nodes_topo_[i]->stop();
-            drain_all_channels();
+            drain_outputs_of(user_node_names_[i]);
         }
         for (auto* n : fanout_nodes_ptr_) n->stop();
     }
@@ -194,6 +202,10 @@ public:
 
     void set_event_handler(EventHandler h) { event_handler_ = std::move(h); }
 
+    /// How long shutdown() waits for one node's outputs to drain before giving
+    /// up on them and stopping the next layer anyway.
+    void set_drain_timeout(std::chrono::milliseconds t) { drain_timeout_ = t; }
+
     /// Application-level error listener. Receives the exception any node's
     /// function throws, after that node's own handler (if any) declined it.
     /// Return true to skip the failed invocation and keep the node running,
@@ -274,16 +286,50 @@ private:
         return {std::move(nodes), std::move(channels), std::move(resources), std::move(pools), elapsed_s};
     }
 
-    void drain_all_channels() const {
+    /// Wait for the channels fed by `src` to empty, or give up.
-        bool any_full = true;
+    ///
-        while (any_full) {
+    /// This was an unbounded `while (anything anywhere is non-empty)` poll over
-            any_full = false;
+    /// *every* channel in the graph, which made shutdown() wait for the whole
-            for (auto& probe : channel_probes_) {
+    /// network to be idle before stopping each successive layer, and wait
-                if (probe->snapshot().current_fill > 0) { any_full = true; break; }
+    /// forever if anything downstream was wedged — turning a graceful shutdown
-            }
+    /// into the hang it exists to avoid.
-            if (any_full)
+    ///
-                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    /// Two bounds, because they fail differently. The deadline covers a
+    /// consumer that has stopped consuming: fill never changes and no amount of
+    /// waiting helps. The no-progress counter covers a consumer that is merely
+    /// slow — it keeps waiting as long as the queue is shrinking, so a slow
+    /// drain is not cut short just for exceeding a fixed time.
+    ///
+    /// Giving up is reported rather than silent: undrained data at this point
+    /// means values are about to be discarded by the stop that follows.
+    void drain_outputs_of(const std::string& src) const {
+        const auto deadline = clock_t::now() + drain_timeout_;
+        std::size_t last_fill = static_cast<std::size_t>(-1);
+        int         stalls    = 0;
+
+        for (;;) {
+            std::size_t fill = 0;
+            for (std::size_t i = 0; i < channel_probes_.size(); ++i)
+                if (channel_src_names_[i] == src)
+                    fill += channel_probes_[i]->snapshot().current_fill;
+
+            if (fill == 0) return;
+            if (fill >= last_fill) { if (++stalls > 100) break; }
+            else                   { stalls = 0; }
+            last_fill = fill;
+
+            if (clock_t::now() >= deadline) break;
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
+
+        std::size_t left = 0;
+        for (std::size_t i = 0; i < channel_probes_.size(); ++i)
+            if (channel_src_names_[i] == src)
+                left += channel_probes_[i]->snapshot().current_fill;
+        if (left)
+            std::cerr << "[kpn] shutdown: '" << src << "' still has " << left
+                      << " queued item(s) its consumer did not take; "
+                         "they are discarded\n";
     }
 
     std::string                                         name_;
@@ -294,6 +340,10 @@ private:
     std::vector<std::string>                            user_node_names_;
     std::vector<std::string>                            fanout_node_names_;
     std::vector<std::unique_ptr<IChannelProbe>>         channel_probes_;
+    /// Display name of the node feeding each probe, parallel to channel_probes_.
+    /// shutdown() drains a node's own outputs, so it has to know which they are.
+    std::vector<std::string>                            channel_src_names_;
+    std::chrono::milliseconds                           drain_timeout_{5000};
     std::vector<std::pair<std::string, IResourceProbe*>> resource_probes_;
     std::vector<std::pair<std::string, IPoolProbe*>>     pool_probes_;
     EventHandler                                        event_handler_;
@@ -330,29 +380,6 @@ auto make_network(Edges&&... edges) {
     // 4. Construct owned fanout storage on the heap (FanoutNode has jthread — not moveable)
     auto fanout_storage = std::make_unique<FanoutSto>();
 
-    // 5. Collect unique user node pointers + their display names, in edge-declaration order
-    std::vector<INode*>      user_node_ptrs;
-    std::vector<std::string> user_node_names;
-    auto collect = [&](auto& e) {
-        using SrcT = std::decay_t<decltype(e.src)>;
-        using DstT = std::decay_t<decltype(e.dst)>;
-        auto* s = static_cast<INode*>(&e.src);
-        auto* d = static_cast<INode*>(&e.dst);
-        if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), s) == user_node_ptrs.end()) {
-            auto sname = node_display_name<SrcT>();
-            user_node_ptrs.push_back(s);
-            user_node_names.push_back(sname);
-            s->set_name(sname);
-        }
-        if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) {
-            auto dname = node_display_name<DstT>();
-            user_node_ptrs.push_back(d);
-            user_node_names.push_back(dname);
-            d->set_name(dname);
-        }
-    };
-    (collect(edges), ...);
-
     // 5. Wire all expanded SimpleEdges.
     //    find_node<NodeT>: searches fanout storage then user edge pack, returns NodeT*.
     //    Uses if constexpr in a fold so mismatched types never reach assignment.
@@ -377,6 +404,38 @@ auto make_network(Edges&&... edges) {
         return ptr;
     };
 
+    // 5. Collect user node pointers + display names in *topological* order.
+    //
+    // Topo is computed above for the cycle check and used to be discarded,
+    // while this vector was filled in edge-declaration order — and then named
+    // user_nodes_topo_ and relied upon as if it were sorted. halt() stops in
+    // its reverse, and shutdown() walks it forwards stopping each node and
+    // draining its outputs before the next, which is only a graceful drain if
+    // the order really is sources-first. It held for every network in the tree
+    // because edges happen to be declared in pipeline order, and would have
+    // broken silently for one that was not.
+    //
+    // Fanout nodes appear in Topo too; they are skipped here because they are
+    // owned separately, in fanout_storage.
+    std::vector<INode*>      user_node_ptrs;
+    std::vector<std::string> user_node_names;
+    [&]<typename... Ns>(tmp::TypeList<Ns...>) {
+        ([&]<typename NodeT>() {
+            if constexpr (!requires { NodeT::is_fanout_node; }) {
+                if (auto* p = find_node.template operator()<NodeT>()) {
+                    auto* n = static_cast<INode*>(p);
+                    if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), n)
+                            == user_node_ptrs.end()) {
+                        auto nm = node_display_name<NodeT>();
+                        user_node_ptrs.push_back(n);
+                        user_node_names.push_back(nm);
+                        n->set_name(nm);
+                    }
+                }
+            }
+        }.template operator()<Ns>(), ...);
+    }(typename Topo::topo{});
+
     // Pre-pass: build fanout_id → source display name map so fanout nodes
     // can be named after the node feeding them (e.g. "capture_fanout").
     std::map<std::size_t, std::string> fanout_src_name;
@@ -401,6 +460,7 @@ auto make_network(Edges&&... edges) {
     };
 
     std::vector<std::unique_ptr<IChannelProbe>> channel_probes;
+    std::vector<std::string>                    channel_src_names;
 
     auto wire_one = [&]<typename SE>(SE) {
         using SrcNode = typename SE::src_node_t;
@@ -418,6 +478,7 @@ auto make_network(Edges&&... edges) {
                                 + " \xe2\x86\x92 "  // UTF-8 →
                                 + node_name.template operator()<DstNode>() + ":" + std::to_string(DstIdx);
             channel_probes.push_back(std::make_unique<ChannelProbe<out_t>>(ch, ch_name));
+            channel_src_names.push_back(node_name.template operator()<SrcNode>());
         }
     };
 
@@ -445,7 +506,8 @@ auto make_network(Edges&&... edges) {
                std::move(fanout_ptrs),
                std::move(user_node_names),
                std::move(fanout_node_names),
-               std::move(channel_probes));
+               std::move(channel_probes),
+               std::move(channel_src_names));
 }
 
 } // namespace kpn

tests/test_channel.cpp

@@ -298,3 +298,28 @@ TEST_CASE("try_push_sentinel leaves a refused value untouched", "[channel][senti
     CHECK(ch.try_push_sentinel(third) == Channel<std::string>::SentinelResult::Closed);
     CHECK(third == "eof-3");
 }
+
+// Regression: try_push must distinguish delivered from discarded.
+//
+// It returned bool, and returned *true* for a closed channel — so "the value
+// arrived" and "the value was thrown away because nobody is listening" were the
+// same answer. Every caller was nonetheless correct, because both cases mean
+// "stop trying"; but nothing above the channel could tell the two apart, and a
+// producer counting successful pushes counted discards among them. Only the
+// channel's own drop counter knew, and only if someone read the diagnostics.
+TEST_CASE("try_push distinguishes taken, full and closed", "[channel]") {
+    Channel<int> ch(2);
+    int v = 1;
+
+    CHECK(ch.try_push(v) == Channel<int>::PushResult::Taken);
+    CHECK(ch.try_push(v) == Channel<int>::PushResult::Taken);
+    // Ring is full: the value is untouched and the caller keeps it.
+    CHECK(ch.try_push(v) == Channel<int>::PushResult::Full);
+    CHECK(v == 1);
+
+    ch.disable();
+    const auto drops_before = ch.stats().drops.load();
+    CHECK(ch.try_push(v) == Channel<int>::PushResult::Closed);
+    // Discarded, and recorded as such rather than reported as a delivery.
+    CHECK(ch.stats().drops.load() == drops_before + 1);
+}

tests/test_network.cpp

@@ -1,6 +1,10 @@
 #include <catch2/catch_test_macros.hpp>
 #include <kpn/kpn.hpp>
+#include <atomic>
 #include <chrono>
+#include <mutex>
+#include <stdexcept>
+#include <string>
 #include <thread>
 
 using namespace kpn;
@@ -54,3 +58,93 @@ TEST_CASE("stop disables input channels — producer push is silently dropped",
     in_ch.push(99);
     REQUIRE(in_ch.size() == 0);
 }
+
+// Regression: Network::set_error_handler must actually deliver the handler.
+//
+// The handler was stored in a member and never read. A node's exception was
+// discarded at the node boundary and the only surviving evidence was a Closed
+// event, which reports that a node stopped but not why — the difference between
+// a diagnosis and a guess. StaticNetwork has always wired this; Network
+// accepted the handler and silently dropped it, which is worse than not
+// offering the setter at all.
+//
+// The type changed with the fix. It was void(name, exception_ptr), which cannot
+// express the keep-running decision the node side needs, so it is now
+// NodeErrorHandler like StaticNetwork's.
+namespace {
+
+static int throwing_stage(int x) {
+    if (x == 42) throw std::runtime_error("boom");
+    return x;
+}
+
+}  // namespace
+
+TEST_CASE("network error handler receives the node's exception", "[network]") {
+    auto src  = kpn::make_node<throwing_stage>(kpn::in<"v">{}, kpn::out<"w">{}, 8);
+    kpn::Channel<int> out(8);
+    src.set_output_channel<0>(&out);
+
+    kpn::Network net;
+    net.add("stage", src).build();
+
+    std::atomic<int>  calls{0};
+    std::string       seen_name;
+    std::string       seen_what;
+    std::mutex        mx;
+
+    net.set_error_handler([&](std::string_view name, std::exception_ptr ep) {
+        std::lock_guard lk(mx);
+        seen_name = std::string(name);
+        try { if (ep) std::rethrow_exception(ep); }
+        catch (const std::exception& e) { seen_what = e.what(); }
+        calls.fetch_add(1, std::memory_order_relaxed);
+        return true;                       // handled: keep the node running
+    });
+
+    net.set_watchdog_interval(std::chrono::hours(1));   // keep the report quiet
+    net.start();
+    src.input_channel<0>().push(42);                    // throws
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    src.input_channel<0>().push(7);                     // must still be running
+    const int passed = out.pop();
+    net.stop();
+
+    CHECK(calls.load(std::memory_order_relaxed) == 1);
+    CHECK(seen_name == "stage");
+    CHECK(seen_what == "boom");
+    CHECK(passed == 7);
+}
+
+// Regression: stopping a network must not wait for the watchdog's next tick.
+//
+// The watchdog looped on std::this_thread::sleep_for(watchdog_interval_), and
+// request_stop() cannot wake a sleeping thread — so stop_watchdog()'s join
+// blocked until the current sleep expired. Every teardown paid up to a full
+// interval, three seconds by default, and a caller who set a long one to keep
+// the periodic report quiet got a stop() that looked like a hang. That is how
+// this was found: the error-handler case above set an hour.
+TEST_CASE("stopping a network does not wait for the watchdog interval", "[network]") {
+    auto node = kpn::make_node<increment>(kpn::in<"v">{}, kpn::out<"w">{}, 4);
+    kpn::Channel<int> out(4);
+    node.set_output_channel<0>(&out);
+
+    kpn::Network net;
+    net.add("inc", node).build();
+    net.set_watchdog_interval(std::chrono::hours(1));
+    net.start();
+
+    // Let the watchdog actually reach its wait. Without this the test races it:
+    // stop_watchdog() runs before the thread has entered the loop, the token is
+    // already set when it does, and it exits without ever waiting — which passes
+    // against the bug as well as the fix.
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    const auto t0 = std::chrono::steady_clock::now();
+    net.stop();
+    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+        std::chrono::steady_clock::now() - t0).count();
+
+    INFO("stop took " << ms << " ms");
+    CHECK(ms < 2000);
+}

tests/test_pool_node.cpp

@@ -668,3 +668,55 @@ TEST_CASE("a node woken with empty inputs does not stop itself", "[pool_node]")
     node.stop();
     pool->stop();
 }
+
+// Regression: stop() must not return while a firing is still running.
+//
+// stop() set the flag, disabled the inputs and returned, leaving an executing
+// fire_once touching input_channels_, stats_ and pending_ while the caller went
+// on to destroy them. The old comment was explicit that callers wanting the
+// guarantee should call scheduler_->drain() first — but ~PoolNode calls stop(),
+// and a destructor cannot ask its caller to have done that.
+//
+// A node with a private pool survived by accident: Node::stop() calls
+// pool->stop(), which joins the worker. A node sharing a pool, which
+// make_pool_node exists to create, had nothing joining it at all, so its own
+// destructor raced the firing.
+//
+// Asserted through an observable side effect rather than by trying to catch the
+// use-after-free: if stop() returns before the node function has finished, the
+// flag it sets on the way out is still false.
+namespace {
+
+struct SlowFiring {
+    static constexpr std::string_view label() { return "slow_firing"; }
+    std::atomic<bool>* entered;
+    std::atomic<bool>* finished;
+    void operator()(int) {
+        entered->store(true, std::memory_order_release);
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+        finished->store(true, std::memory_order_release);
+    }
+};
+
+}  // namespace
+
+TEST_CASE("stop waits for a firing already in flight", "[pool_node]") {
+    std::atomic<bool> entered{false}, finished{false};
+
+    auto pool = std::make_shared<ThreadPool>(2);
+    pool->start();
+
+    SlowFiring fn{&entered, &finished};
+    auto node = make_pool_node(fn, pool, 4);
+    node.start();
+    node.input_channel<0>().push(1);
+
+    // Stop only once the node is demonstrably inside its function.
+    while (!entered.load(std::memory_order_acquire))
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+
+    node.stop();
+    CHECK(finished.load(std::memory_order_acquire));
+
+    pool->stop();
+}

tests/test_scheduler.cpp

@@ -227,3 +227,116 @@ TEST_CASE("work stealing: tasks complete with more threads than initial queue ta
     REQUIRE(counter.load() == 4);
     pool.stop();
 }
+
+// Regression: submitting to a stopped pool must be a no-op, not a segfault.
+//
+// stop() ends with queues_.clear(), and submit() went straight to
+// queues_[target] with no check — so a submission arriving after stop indexed
+// an empty vector.
+//
+// This is not a contrived teardown ordering; it happens on a normal path. A
+// node's space callback fires from whichever thread drained the channel, and
+// that thread belongs to the *consumer*. Stop the producer first — which a
+// sources-first shutdown does by design — and the consumer keeps draining its
+// backlog, firing the producer's space callback into a pool that has already
+// been torn down. Before this fix the static-network shutdown case crashed
+// about 12 runs in 20.
+//
+// Checking stopped_ without the lock would not be enough: the window between
+// reading the flag and indexing the vector is exactly where clear() runs.
+TEST_CASE("submitting to a stopped pool is refused, not fatal", "[scheduler]") {
+    ThreadPool pool(2);
+    pool.start();
+    pool.stop();
+
+    std::atomic<int> ran{0};
+    for (int i = 0; i < 10; ++i)
+        pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); });
+
+    CHECK(ran.load(std::memory_order_relaxed) == 0);
+    CHECK(pool.rejected() == 10);
+}
+
+TEST_CASE("submitting while the pool stops does not crash", "[scheduler]") {
+    // The racing form of the case above: a producer thread submitting
+    // continuously while stop() runs underneath it. Nothing is asserted about
+    // how many tasks run — the point is that every submission either enqueues
+    // or is refused, and none touches a destroyed queue.
+    for (int rep = 0; rep < 20; ++rep) {
+        ThreadPool pool(4);
+        pool.start();
+
+        std::atomic<bool> go{false};
+        std::atomic<int>  ran{0};
+        std::thread submitter([&] {
+            while (!go.load(std::memory_order_acquire)) {}
+            for (int i = 0; i < 2000; ++i)
+                pool.submit([&] { ran.fetch_add(1, std::memory_order_relaxed); });
+        });
+
+        go.store(true, std::memory_order_release);
+        std::this_thread::sleep_for(std::chrono::microseconds(200));
+        pool.stop();
+        submitter.join();
+
+        // Everything submitted was either executed or refused; nothing vanished
+        // into a queue that no longer existed.
+        CHECK(pool.rejected() + pool.snapshot("p").tasks_completed <= 2000);
+    }
+}
+
+// Regression: idle workers must sleep while another worker is busy.
+//
+// The wait predicate was `stopped_ || total_ > 0`, and total_ counts queued
+// *plus executing*. So while any one task ran, every other worker's predicate
+// was true: wait() returned immediately and the worker spun through try_pop,
+// try_steal and back to wait at full speed — try_lock-ing every peer queue on
+// each pass. One slow task pinned every other core and contended the very
+// mutexes the working thread needed to make progress.
+//
+// That is the shape of this pipeline's load exactly: a handful of nodes whose
+// work is tens of milliseconds of ONNX inference. It was latent only because
+// each node currently owns a private single-thread pool, where there is no
+// idle peer to spin. Any use of a shared pool — which make_pool_node exists
+// for — hits it immediately.
+//
+// Measured as CPU time rather than wall time, because the bug does not make
+// anything slower to finish; it makes seven cores burn while one works. A
+// sleeping task consumes no CPU, so with workers correctly asleep the whole
+// pool should account for almost none.
+TEST_CASE("idle workers do not spin while one task runs", "[scheduler]") {
+    auto cpu_ms = [] {
+        struct timespec ts{};
+        clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &ts);
+        return ts.tv_sec * 1000.0 + ts.tv_nsec / 1e6;
+    };
+
+    constexpr int kThreads = 8;
+    constexpr int kWorkMs  = 300;
+
+    ThreadPool pool(kThreads);
+    pool.start();
+    std::this_thread::sleep_for(20ms);      // let workers reach the wait
+
+    const double before = cpu_ms();
+
+    // One long task plus a trivial one per remaining worker. The trivial ones
+    // matter: a worker that has never been woken stays blocked in wait() and
+    // never re-evaluates the predicate, so the spin only appears once a worker
+    // *finishes* something and re-enters the loop while a peer is still busy.
+    // Submitting only the long task does not reproduce it.
+    pool.submit([&] { std::this_thread::sleep_for(std::chrono::milliseconds(kWorkMs)); });
+    for (int i = 0; i < kThreads - 1; ++i) pool.submit([] {});
+
+    pool.drain();
+    const double used = cpu_ms() - before;
+    pool.stop();
+
+    // Measured on this tree: 1991 ms of CPU with the total_ predicate against
+    // 0.4 ms with queued_, and 19205 voluntary context switches against 10 —
+    // roughly (kThreads - 1) cores burned for the duration of one sleeping
+    // task. The threshold sits far from both so the case is not sensitive to
+    // how loaded the machine is.
+    INFO("cpu " << used << " ms over " << kWorkMs << " ms of sleeping work");
+    CHECK(used < kWorkMs);
+}

tests/test_shared_resource.cpp

@@ -237,3 +237,51 @@ TEST_CASE("make_shared_resource constructs with forwarded args", "[shared_resour
     auto g = res.acquire();
     REQUIRE(*g == "hello");
 }
+
+// Regression: a waiter must be releasable, or teardown waits on it forever.
+//
+// acquire() blocks on a condition variable whose predicate only becomes true
+// when release() hands over ownership. There was no timeout and no stop
+// condition, so a node parked there ignored teardown entirely: its worker never
+// returned, the pool's join never completed, and shutdown hung waiting for a
+// resource nobody was going to release — which is exactly the case when the
+// holder is being stopped too.
+//
+// close() turns that into an exception the node's existing error path already
+// handles, and networks now call it on registered resources before stopping any
+// node, for the same reason.
+TEST_CASE("closing a shared resource releases its waiters", "[shared_resource]") {
+    SharedResource<int> res(42);
+
+    auto holder = res.acquire();          // resource is now held
+
+    std::atomic<bool> threw{false}, returned{false};
+    std::thread waiter([&] {
+        try {
+            auto g = res.acquire();       // blocks: someone else holds it
+            (void)g;
+        } catch (const ResourceClosedError&) {
+            threw.store(true, std::memory_order_release);
+        }
+        returned.store(true, std::memory_order_release);
+    });
+
+    // Let it park, then tear down without ever releasing the holder.
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    REQUIRE_FALSE(returned.load(std::memory_order_acquire));
+
+    res.close();
+    waiter.join();
+
+    CHECK(threw.load(std::memory_order_acquire));
+}
+
+TEST_CASE("acquiring a closed resource fails immediately", "[shared_resource]") {
+    SharedResource<int> res(7);
+    res.close();
+    CHECK_THROWS_AS(res.acquire(), ResourceClosedError);
+
+    // Reusable across runs once reopened.
+    res.reopen();
+    CHECK_NOTHROW(res.acquire());
+}

tests/test_static_network.cpp

@@ -271,3 +271,164 @@ TEST_CASE("static_network: fanout with labelled same-function consumers", "[stat
     REQUIRE(outB.pop() == 7);
     net.stop();
 }
+
+// Regression: shutdown() must return even when a consumer stopped consuming.
+//
+// The drain step was an unbounded `while (anything anywhere is non-empty)` poll
+// over *every* channel in the graph. Two defects in one loop: it waited for the
+// whole network to be idle before stopping each successive layer rather than
+// just the node it had stopped — the dynamic Network's version even took a node
+// name and ignored it — and it had no deadline, so anything wedged downstream
+// turned a graceful shutdown into the hang it exists to avoid.
+//
+// It could also fail to terminate with nothing wedged at all. current_fill came
+// from a snapshot that loaded tail_ before head_; a concurrent pop between the
+// two reads yields a head_ past the sampled tail_, and the unsigned difference
+// wraps to ~2^64. Any poll for "is it empty yet" against that value runs
+// forever. Both indices only ever increase, so loading head_ first can at worst
+// under-report a push, which this loop tolerates and a wrap does not.
+//
+// Here the sink never takes anything, so its input cannot drain and the only
+// correct outcome is to give up and say so. The bound asserted is deliberately
+// loose: the point is that it terminates, not how fast.
+namespace {
+
+struct DrainSource {
+    static constexpr std::string_view label() { return "drain_source"; }
+    int n{0};
+    int operator()() {
+        std::this_thread::sleep_for(std::chrono::microseconds(100));
+        return n++;
+    }
+};
+
+struct NeverConsumes {
+    static constexpr std::string_view label() { return "never_consumes"; }
+    std::atomic<bool>* wedged;
+    void operator()(int) {
+        // Blocks for the duration of the test: the input channel behind it
+        // fills and stays full.
+        while (!wedged->load(std::memory_order_acquire))
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+};
+
+}  // namespace
+
+TEST_CASE("shutdown returns when a consumer has wedged", "[static_network][shutdown]") {
+    std::atomic<bool> release{false};
+
+    DrainSource   src_fn;
+    NeverConsumes sink_fn{&release};
+
+    kpn::ObjectNode<DrainSource,   kpn::in<>,    kpn::out<"v">, "drain_source",   0> s(src_fn,  4);
+    kpn::ObjectNode<NeverConsumes, kpn::in<"v">, kpn::out<>,    "never_consumes", 0> k(sink_fn, 4);
+
+    auto net = kpn::make_network(kpn::edge(s.output<"v">(), k.input<"v">()));
+    net.set_drain_timeout(std::chrono::milliseconds(100));
+    net.start();
+
+    // Let the channel fill and the sink jam.
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    // Unjam the sink well after the drain timeout should have expired. Stopping
+    // a node joins its worker, so a sink blocked forever would hang the test in
+    // stop() rather than in the drain loop this case is about.
+    std::thread unjam([&] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(800));
+        release.store(true, std::memory_order_release);
+    });
+
+    const auto t0 = std::chrono::steady_clock::now();
+    net.shutdown();
+    const auto elapsed = std::chrono::steady_clock::now() - t0;
+
+    unjam.join();
+
+    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
+    INFO("shutdown took " << ms << " ms");
+    CHECK(ms < 3000);   // unbounded before; one 100 ms drain timeout after
+}
+
+// Regression: node order must come from the topological sort, not from the
+// order the edges happened to be written in.
+//
+// make_network computes Topo for the cycle check and then dropped it, filling
+// the node vector in edge-declaration order — and named it user_nodes_topo_.
+// halt() stops in its reverse, and shutdown() walks it forwards stopping each
+// node and draining its outputs before moving to the next, which is a graceful
+// drain only if the order really is sources-first.
+//
+// Every network in this tree declares edges in pipeline order, so the two
+// coincided and nothing failed. This case declares them backwards, which is
+// legal and which make_network otherwise accepts silently.
+//
+// Asserted through shutdown() rather than by reading the order back, because
+// the order is private and the ordering is not the point — what it buys is.
+// A sources-first shutdown lets the values already in flight reach the sink;
+// stopping the sink first strands them, and the drain step then has nobody
+// left to take them.
+namespace {
+
+struct OrderSource {
+    static constexpr std::string_view label() { return "order_source"; }
+    std::atomic<int>* made;
+    int operator()() {
+        std::this_thread::sleep_for(std::chrono::microseconds(20));
+        return made->fetch_add(1, std::memory_order_relaxed);
+    }
+};
+
+// Deliberately slower than the source, so a deep backlog builds up in its input
+// channel. That backlog is what a sources-first shutdown preserves and a
+// sink-first one throws away, and it needs to be big enough that the difference
+// cannot be mistaken for one value in flight.
+struct OrderRelay {
+    static constexpr std::string_view label() { return "order_relay"; }
+    int operator()(int v) {
+        std::this_thread::sleep_for(std::chrono::microseconds(300));
+        return v;
+    }
+};
+
+struct OrderSink {
+    static constexpr std::string_view label() { return "order_sink"; }
+    std::atomic<int>* seen;
+    void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); }
+};
+
+}  // namespace
+
+TEST_CASE("edges declared out of order still start and stop sources-first",
+          "[static_network][shutdown]") {
+    std::atomic<int> seen{0}, made{0};
+
+    OrderSource src_fn{&made};
+    OrderRelay  relay_fn;
+    OrderSink   sink_fn{&seen};
+
+    kpn::ObjectNode<OrderSource, kpn::in<>,    kpn::out<"v">, "order_source", 0> s(src_fn,   8);
+    kpn::ObjectNode<OrderRelay,  kpn::in<"v">, kpn::out<"w">, "order_relay",  0> r(relay_fn, 64);
+    kpn::ObjectNode<OrderSink,   kpn::in<"w">, kpn::out<>,    "order_sink",   0> k(sink_fn,  64);
+
+    // Sink edge first, source edge last — the reverse of pipeline order.
+    auto net = kpn::make_network(
+        kpn::edge(r.output<"w">(), k.input<"w">()),
+        kpn::edge(s.output<"v">(), r.input<"v">())
+    );
+    net.start();
+    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+
+    const int before = seen.load(std::memory_order_relaxed);
+    REQUIRE(before > 0);              // the pipeline ran at all
+
+    net.shutdown();
+
+    // Sources stop first and each layer drains before the next stops, so the
+    // backlog queued in front of the relay still reaches the sink. Stopping in
+    // declaration order stops the relay first and discards all of it.
+    const int after = seen.load(std::memory_order_relaxed);
+    INFO("made " << made.load() << ", delivered " << before
+         << " before shutdown, " << after << " after");
+    CHECK(after - before >= 20);
+}