KPN/include/kpn/pool_node.hpp

#pragma once
#include "channel.hpp"
#include "diagnostics.hpp"
#include "fixed_string.hpp"
#include "inode.hpp"
#include "port.hpp"
#include "scheduler.hpp"
#include "traits.hpp"

#include <array>
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <iostream>
#include <memory>
#include <optional>
#include <variant>
#include <stdexcept>
#include <thread>
#include <tuple>
#include <type_traits>

namespace kpn {

// ── Sentinel detection ────────────────────────────────────────────────────────
// A value is a "sentinel" (must-deliver control token, e.g. EOF) if its type
// carries a bool-convertible eof flag — either directly (`v.eof`, as on a raw
// source Frame) or nested one level under a `.source` member (`v.source.eof`,
// as on the pipeline's SceneFrame/…/MatchedSceneFrame message types, which wrap
// the originating Frame). Sentinels are delivered losslessly and non-blockingly
// via Channel::push_sentinel() instead of the throwing push(), so backpressure
// can never drop the token that unblocks downstream teardown.
//
// Types with neither shape are never treated as sentinels — both traits are
// SFINAE-safe and the runtime check compiles away to `false` for them, so this
// stays a no-op for pipelines that don't use an eof convention.
template<typename T, typename = void>
struct has_eof_field : std::false_type {};
template<typename T>
struct has_eof_field<T, std::void_t<decltype(static_cast<bool>(std::declval<const T&>().eof))>>
    : std::true_type {};

template<typename T, typename = void>
struct has_source_eof_field : std::false_type {};
template<typename T>
struct has_source_eof_field<T,
    std::void_t<decltype(static_cast<bool>(std::declval<const T&>().source.eof))>>
    : std::true_type {};

template<typename T>
constexpr bool is_sentinel_value(const T& v) {
    if constexpr (has_eof_field<T>::value)             return static_cast<bool>(v.eof);
    else if constexpr (has_source_eof_field<T>::value) return static_cast<bool>(v.source.eof);
    else                                               return false;
}

// ── PoolNode ──────────────────────────────────────────────────────────────────
//
// Reactive alternative to Node<>. Instead of owning a blocked thread, the node
// is submitted to a shared IScheduler whenever all its input channels become
// non-empty. A single fire_once() call pops all inputs, executes the function,
// and pushes outputs. At most one fire_once() runs at a time (queued_ flag).
//
// Source nodes (input_count == 0) submit themselves immediately on start() and
// resubmit after each fire_once().
//
// Multiple PoolNodes can share one ThreadPool for resource-bounded execution,
// or each can have a dedicated single-thread pool for serialisation.

template<auto Func,
         typename InputTag  = in<>,
         typename OutputTag = out<>,
         fixed_string Label    = "",
         std::size_t UniqueTag = 0>
class PoolNode;

template<auto Func, fixed_string... InNames, fixed_string... OutNames,
         fixed_string Label, std::size_t UniqueTag>
class PoolNode<Func, in<InNames...>, out<OutNames...>, Label, UniqueTag> : public INode {
public:
    using F            = decltype(Func);
    using args_tuple   = args_t<F>;
    using return_raw   = return_t<F>;
    using return_tuple = normalised_return_t<return_raw>;

    static constexpr std::string_view label()     { return Label.view(); }
    static constexpr std::size_t      unique_tag  = UniqueTag;

    static constexpr std::size_t input_count  = arity_v<F>;
    static constexpr std::size_t output_count = std::tuple_size_v<return_tuple>;

    static_assert(
        sizeof...(InNames) == 0 || sizeof...(InNames) == input_count,
        "make_pool_node: number of input names must match function arity, or provide none"
    );
    static_assert(
        sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
        "make_pool_node: number of output names must match return tuple size, or provide none"
    );

    explicit PoolNode(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5)
        : scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
    {
        init_input_channels(std::make_index_sequence<input_count>{});
    }

    ~PoolNode() override { stop(); }

    // ── INode ─────────────────────────────────────────────────────────────────

    void start() override {
        enable_inputs(std::make_index_sequence<input_count>{});
        stop_flag_.store(false, std::memory_order_relaxed);
        queued_.store(false, std::memory_order_relaxed);
        register_callbacks(std::make_index_sequence<input_count>{});
        if constexpr (input_count == 0)
            try_submit(0.5f);
        else
            // Never start with a wake already outstanding — the startup case of
            // the invariant 9c5ce5f established for the running pipeline.
            //
            // enable_inputs() opens the channel several statements before
            // register_callbacks() installs the push callback, and the network
            // starts nodes sources-first, so an upstream node is already firing
            // into this one during that gap. A push landing there is accepted by
            // the ring but wakes nobody: Channel::push only invokes the callback
            // on the empty→non-empty transition, and at that instant the
            // callback is still null. Every later push sees a non-empty ring and
            // stays silent, so the node is never submitted — the pipeline reads
            // as wedged from the first frame, with no item ever delivered.
            //
            // on_input_ready() is the level-triggered form of the same question,
            // so asking it once here converts the missed edge into a state check.
            on_input_ready();
    }

    void stop() override {
        stop_flag_.store(true, std::memory_order_seq_cst);
        disable_inputs(std::make_index_sequence<input_count>{});
        // fire_once() observes stop_flag_ and will not resubmit.
        // 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 {
        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); }
    void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
    void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
    void set_network_closed_callback(NodeEventCallback cb)   override { closed_callbacks_[1] = std::move(cb); }

    const NodeStats& stats() const override { return stats_; }

    NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
        uint64_t frames    = stats_.frames_processed.load(std::memory_order_relaxed);
        double exec_ms     = stats_.ema_exec_us.load(std::memory_order_relaxed) / 1000.0;
        double blocked_ms  = stats_.total_blocked_us.load(std::memory_order_relaxed) / 1000.0;
        double qwait_ms    = stats_.queue_wait_us.load(std::memory_order_relaxed) / 1000.0;
        double total_ms    = exec_ms + blocked_ms;
        return {
            name, frames, exec_ms,
            stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0,
            blocked_ms,
            elapsed_s > 0 ? frames / elapsed_s : 0.0,
            stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
            total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
            qwait_ms,
            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
            queued_.load(std::memory_order_relaxed),
            wake_pending_.load(std::memory_order_relaxed),
        };
    }

    // ── Port access — by index ────────────────────────────────────────────────

    template<std::size_t I>
    InputPort<PoolNode, I> input() {
        static_assert(I < input_count, "input index out of range");
        return {*this};
    }

    template<std::size_t I>
    OutputPort<PoolNode, I> output() {
        static_assert(I < output_count, "output index out of range");
        return {*this};
    }

    // ── Port access — by name ─────────────────────────────────────────────────

    template<fixed_string Name>
    auto input() {
        constexpr std::size_t idx = index_of<Name, InNames...>();
        static_assert(idx != npos, "unknown input port name");
        return input<idx>();
    }

    template<fixed_string Name>
    auto output() {
        constexpr std::size_t idx = index_of<Name, OutNames...>();
        static_assert(idx != npos, "unknown output port name");
        return output<idx>();
    }

    // ── Internal channel accessors ────────────────────────────────────────────

    template<std::size_t I>
    Channel<std::tuple_element_t<I, args_tuple>>& input_channel() {
        return *std::get<I>(input_channels_);
    }

    template<std::size_t I>
    void set_input_channel(
        std::shared_ptr<Channel<std::tuple_element_t<I, args_tuple>>> ch) {
        std::get<I>(input_channels_) = std::move(ch);
    }

    template<std::size_t I>
    void set_output_channel(
        Channel<std::tuple_element_t<I, return_tuple>>* ch) {
        std::get<I>(output_channels_) = ch;
    }

private:
    // ── Channel storage ───────────────────────────────────────────────────────

    template<std::size_t... Is>
    void init_input_channels(std::index_sequence<Is...>) {
        ((std::get<Is>(input_channels_) =
              std::make_shared<Channel<std::tuple_element_t<Is, args_tuple>>>(fifo_capacity_)),
         ...);
    }

    template<std::size_t... Is>
    void enable_inputs(std::index_sequence<Is...>) {
        (std::get<Is>(input_channels_)->enable(), ...);
    }

    template<std::size_t... Is>
    void disable_inputs(std::index_sequence<Is...>) {
        (std::get<Is>(input_channels_)->disable(), ...);
    }

    template<std::size_t... Is>
    void disable_outputs(std::index_sequence<Is...>) {
        auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
        (disable_one(std::get<Is>(output_channels_)), ...);
    }

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

    static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
        const auto ts = std::chrono::steady_clock::now();
        for (auto& cb : cbs) if (cb) cb(ts);
    }

    template<std::size_t... Os>
    bool outputs_have_space(std::index_sequence<Os...>) const {
        return (... && (!std::get<Os>(output_channels_) ||
                        std::get<Os>(output_channels_)->has_space()));
    }

    template<std::size_t... Os>
    void register_space_callbacks(std::index_sequence<Os...>) {
        ((std::get<Os>(output_channels_)
            ? (void)std::get<Os>(output_channels_)->set_space_callback(
                  [this] { try_submit(0.5f); })
            : (void)0), ...);
    }

    void self_stop() {
        disable_inputs(std::make_index_sequence<input_count>{});
        disable_outputs(std::make_index_sequence<output_count>{});
        stats_.exec_start_us.store(0, std::memory_order_relaxed);
        // Plain store, not release_and_recheck(): this node is stopping, and
        // honouring a pending wake here would resubmit a dead node.
        queued_.store(false, std::memory_order_release);
        stop_flag_.store(true, std::memory_order_relaxed);
    }

    template<typename Tup, std::size_t... Is>
    static auto make_input_channel_tuple(std::index_sequence<Is...>)
        -> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;

    using input_channels_t = decltype(make_input_channel_tuple<args_tuple>(
        std::make_index_sequence<input_count>{}));

    template<typename Tup, std::size_t... Is>
    static auto make_output_channel_tuple(std::index_sequence<Is...>)
        -> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;

    using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
        std::make_index_sequence<output_count>{}));

    // ── Scheduling ────────────────────────────────────────────────────────────

    // Called by channel push_callbacks (on the producer's thread).
    void on_input_ready() {
        if (stop_flag_.load(std::memory_order_relaxed)) return;
        std::size_t ready = count_ready(std::make_index_sequence<input_count>{});
        if (ready == input_count)
            try_submit(compute_priority());
    }

    template<std::size_t... Is>
    std::size_t count_ready(std::index_sequence<Is...>) {
        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 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...>) {
        ((sum += std::get<Is>(input_channels_)->capacity() > 0
            ? float(std::get<Is>(input_channels_)->approx_size())
              / float(std::get<Is>(input_channels_)->capacity())
            : 0.5f), ...);
    }

    /// Submit unless already queued. A wake that arrives while this node is
    /// queued or running is *recorded*, never dropped.
    ///
    /// Wakes are edge-triggered: a channel fires its space callback on the
    /// transition, once. If that lands while queued_ is up, the CAS below fails
    /// and — before wake_pending_ — the wake was gone. A node could then park a
    /// value, release its worker, and sleep forever holding output its consumer
    /// was waiting for, with every worker idle in cond_wait and nothing left to
    /// re-trigger it. Recording the drop turns the signal level-triggered: the
    /// invariant is that a node never sleeps with a wake outstanding, enforced
    /// by release_and_recheck() at every point that releases the node.
    void try_submit(float priority) {
        bool expected = false;
        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
            scheduler_->submit([this] { fire_once(); }, priority);
        else
            wake_pending_.store(true, std::memory_order_release);
    }

    /// Clear queued_, then honour any wake that was dropped while it was up.
    /// Every path that finishes or parks a firing must release the node through
    /// here rather than storing queued_ directly.
    void release_and_recheck(float priority = 0.5f) {
        queued_.store(false, std::memory_order_release);
        if (wake_pending_.exchange(false, std::memory_order_acq_rel))
            try_submit(priority);
    }

    // ── Execution ─────────────────────────────────────────────────────────────

    void fire_once() {
        if (stop_flag_.load(std::memory_order_relaxed)) {
            queued_.store(false, std::memory_order_release);
            return;
        }

        // Record queue wait time (submission → now) and mark as executing
        auto t0 = clock_t::now();
        int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
            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>{});
                release_and_recheck();
                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>{});
                release_and_recheck();
                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) {
                release_and_recheck();
                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();
            stats_.record_queue_wait(duration_t(t1 - t0));
            auto cpu0 = NodeStats::cpu_now();

            if constexpr (std::is_void_v<return_raw>) {
                std::apply(Func, args);
            } else {
                auto result = std::apply(Func, args);
                push_outputs(normalise(std::move(result)),
                             std::make_index_sequence<output_count>{});
            }

            auto cpu1 = NodeStats::cpu_now();
            auto t2   = clock_t::now();
            // blocked_time = 0 for pool nodes (we don't block waiting for inputs)
            stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
        } catch (const ChannelClosedError&) {
            fire_callbacks(closed_callbacks_);
            self_stop();
            return;
        } catch (const ChannelOverflowError&) {
            fire_callbacks(event_callbacks_);
        } catch (...) {
            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_);
                self_stop();
                return;
            }
        }

        stats_.exec_start_us.store(0, std::memory_order_relaxed);
        release_and_recheck();

        if (stop_flag_.load(std::memory_order_relaxed)) return;

        // Parked by the push above. Same situation as the retry path at the top
        // of fire_once — and the same lost-wakeup race, which that path closes
        // and this one did not. A space callback that fired while queued_ was
        // still up got swallowed by try_submit's CAS, and the resubmit below
        // cannot cover it: this firing consumed its input, so inputs are empty
        // and on_input_ready() will not resubmit. The node would then hold its
        // value forever while its consumer waits for exactly that value and its
        // producer parks on an input channel that never drains. Re-check now
        // that the flag is down.
        if constexpr (!std::is_void_v<return_raw>) {
            if (pending_) {
                if (outputs_have_space(std::make_index_sequence<output_count>{}))
                    try_submit(0.5f);
                return;                    // parked
            }
        }

        // Source nodes always resubmit; others resubmit only if inputs are ready.
        if constexpr (input_count == 0) {
            try_submit(0.5f);
        } else {
            on_input_ready();
        }
    }

    // Pop all inputs — safe because we're the sole consumer and fire_once
    // is guarded by queued_ (only one fire_once runs at a time).
    template<std::size_t... Is>
    args_tuple pop_inputs(std::index_sequence<Is...>) {
        return {pop_one<Is>()...};
    }

    template<std::size_t I>
    std::tuple_element_t<I, args_tuple> pop_one() {
        auto& ch = *std::get<I>(input_channels_);
        std::tuple_element_t<I, args_tuple> val;
        if (!ch.try_pop_now(val))
            throw ChannelClosedError{};
        return val;
    }

    template<typename R = return_raw>
    static return_tuple normalise(R&& r) {
        if constexpr (is_tuple_v<R>) return std::move(r);
        else                         return std::make_tuple(std::move(r));
    }

    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...>) {
        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>
    bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
        auto* ch = std::get<I>(output_channels_);
        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 true;
        }
        // Backpressure without parking the worker. A full channel means the
        // consumer is behind; the value is kept and this node stops running
        // until the channel signals space (set_space_callback re-submits it).
        //
        // Blocking here instead would sleep inside a scheduler worker, and
        // nodes are pinned to workers — park enough of them and nothing is left
        // to run the consumer that would drain the channel. That is the
        // hold-and-wait deadlock channel.hpp warns about for sentinels; it
        // applies to data pushes just as much.
        return ch->try_push(val);
    }

    template<std::size_t I>
    static std::string output_port_label() {
        if constexpr (sizeof...(OutNames) > 0) {
            constexpr std::array<std::string_view, sizeof...(OutNames)> names{OutNames.view()...};
            return std::string("output['") + std::string(names[I]) + "']";
        } else {
            return "output[" + std::to_string(I) + "]";
        }
    }

    // ── State ─────────────────────────────────────────────────────────────────

    std::shared_ptr<IScheduler>              scheduler_;
    std::string                              name_;
    std::size_t                              fifo_capacity_;
    input_channels_t                         input_channels_;
    output_channels_t                        output_channels_{};
    std::atomic<bool>                        stop_flag_{true};
    std::atomic<bool>                        queued_{false};
    /// A wake that arrived while queued_ was up. See try_submit.
    std::atomic<bool>                        wake_pending_{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_{};
};

// ── PoolObjectNode ────────────────────────────────────────────────────────────
//
// Same as PoolNode but wraps a stateful callable object (functor / class with
// operator()). The object must outlive the PoolObjectNode.

template<typename Obj,
         typename InputTag  = in<>,
         typename OutputTag = out<>,
         fixed_string Label    = "",
         std::size_t UniqueTag = 0>
class PoolObjectNode;

template<typename Obj, fixed_string... InNames, fixed_string... OutNames,
         fixed_string Label, std::size_t UniqueTag>
class PoolObjectNode<Obj, in<InNames...>, out<OutNames...>, Label, UniqueTag> : public INode {
public:
    using F            = decltype(&Obj::operator());
    using args_tuple   = args_t<F>;
    using return_raw   = return_t<F>;
    using return_tuple = normalised_return_t<return_raw>;

    static constexpr std::string_view label()    { return Label.view(); }
    static constexpr std::size_t      unique_tag = UniqueTag;

    static constexpr std::size_t input_count  = arity_v<F>;
    static constexpr std::size_t output_count = std::tuple_size_v<return_tuple>;

    static_assert(
        sizeof...(InNames) == 0 || sizeof...(InNames) == input_count,
        "make_pool_node: number of input names must match operator() arity, or provide none"
    );
    static_assert(
        sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count,
        "make_pool_node: number of output names must match return tuple size, or provide none"
    );

    explicit PoolObjectNode(Obj& obj, std::shared_ptr<IScheduler> sched,
                            std::size_t fifo_capacity = 5)
        : obj_(obj), scheduler_(std::move(sched)), fifo_capacity_(fifo_capacity)
    {
        init_input_channels(std::make_index_sequence<input_count>{});
    }

    ~PoolObjectNode() override { stop(); }

    void start() override {
        enable_inputs(std::make_index_sequence<input_count>{});
        stop_flag_.store(false, std::memory_order_relaxed);
        queued_.store(false, std::memory_order_relaxed);
        register_callbacks(std::make_index_sequence<input_count>{});
        if constexpr (input_count == 0)
            try_submit(0.5f);
        else
            // Never start with a wake already outstanding — see PoolNode::start().
            on_input_ready();
    }

    void stop() override {
        stop_flag_.store(true, std::memory_order_seq_cst);
        disable_inputs(std::make_index_sequence<input_count>{});
    }

    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); }
    void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); }
    void set_closed_callback(NodeEventCallback cb)   { closed_callbacks_[0] = std::move(cb); }
    void set_network_closed_callback(NodeEventCallback cb)   override { closed_callbacks_[1] = std::move(cb); }

    const NodeStats& stats() const override { return stats_; }

    NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override {
        uint64_t frames   = stats_.frames_processed.load(std::memory_order_relaxed);
        double exec_ms    = stats_.ema_exec_us.load(std::memory_order_relaxed) / 1000.0;
        double blocked_ms = stats_.total_blocked_us.load(std::memory_order_relaxed) / 1000.0;
        double qwait_ms   = stats_.queue_wait_us.load(std::memory_order_relaxed) / 1000.0;
        double total_ms   = exec_ms + blocked_ms;
        return {
            name, frames, exec_ms,
            stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0,
            blocked_ms,
            elapsed_s > 0 ? frames / elapsed_s : 0.0,
            stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0,
            total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0,
            qwait_ms,
            stats_.total_exec_us.load(std::memory_order_relaxed) / 1000.0,
            queued_.load(std::memory_order_relaxed),
            wake_pending_.load(std::memory_order_relaxed),
        };
    }

    template<std::size_t I> InputPort<PoolObjectNode, I>  input()  { return {*this}; }
    template<std::size_t I> OutputPort<PoolObjectNode, I> output() { return {*this}; }

    template<fixed_string Name>
    auto input() {
        constexpr std::size_t idx = index_of<Name, InNames...>();
        static_assert(idx != npos, "unknown input port name");
        return input<idx>();
    }
    template<fixed_string Name>
    auto output() {
        constexpr std::size_t idx = index_of<Name, OutNames...>();
        static_assert(idx != npos, "unknown output port name");
        return output<idx>();
    }

    template<std::size_t I>
    Channel<std::tuple_element_t<I, args_tuple>>& input_channel() {
        return *std::get<I>(input_channels_);
    }
    template<std::size_t I>
    void set_input_channel(std::shared_ptr<Channel<std::tuple_element_t<I, args_tuple>>> ch) {
        std::get<I>(input_channels_) = std::move(ch);
    }
    template<std::size_t I>
    void set_output_channel(Channel<std::tuple_element_t<I, return_tuple>>* ch) {
        std::get<I>(output_channels_) = ch;
    }

private:
    template<std::size_t... Is>
    void init_input_channels(std::index_sequence<Is...>) {
        ((std::get<Is>(input_channels_) =
              std::make_shared<Channel<std::tuple_element_t<Is, args_tuple>>>(fifo_capacity_)),
         ...);
    }
    template<std::size_t... Is> void enable_inputs(std::index_sequence<Is...>)  { (std::get<Is>(input_channels_)->enable(), ...); }
    template<std::size_t... Is> void disable_inputs(std::index_sequence<Is...>) { (std::get<Is>(input_channels_)->disable(), ...); }
    template<std::size_t... Is>
    void disable_outputs(std::index_sequence<Is...>) {
        auto disable_one = [](auto* ch) { if (ch) ch->disable(); };
        (disable_one(std::get<Is>(output_channels_)), ...);
    }
    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(); }), ...);
    }

    static void fire_callbacks(const std::array<NodeEventCallback, 2>& cbs) {
        const auto ts = std::chrono::steady_clock::now();
        for (auto& cb : cbs) if (cb) cb(ts);
    }

     template<std::size_t... Os>
     bool outputs_have_space(std::index_sequence<Os...>) const {
         return (... && (!std::get<Os>(output_channels_) ||
                         std::get<Os>(output_channels_)->has_space()));
     }

     template<std::size_t... Os>
     void register_space_callbacks(std::index_sequence<Os...>) {
         ((std::get<Os>(output_channels_)
             ? (void)std::get<Os>(output_channels_)->set_space_callback(
                   [this] { try_submit(0.5f); })
             : (void)0), ...);
     }

    void self_stop() {
        disable_inputs(std::make_index_sequence<input_count>{});
        disable_outputs(std::make_index_sequence<output_count>{});
        stats_.exec_start_us.store(0, std::memory_order_relaxed);
        // Plain store, not release_and_recheck(): this node is stopping, and
        // honouring a pending wake here would resubmit a dead node.
        queued_.store(false, std::memory_order_release);
        stop_flag_.store(true, std::memory_order_relaxed);
    }

    template<typename Tup, std::size_t... Is>
    static auto make_input_channel_tuple(std::index_sequence<Is...>)
        -> std::tuple<std::shared_ptr<Channel<std::tuple_element_t<Is, Tup>>>...>;
    using input_channels_t = decltype(make_input_channel_tuple<args_tuple>(
        std::make_index_sequence<input_count>{}));

    template<typename Tup, std::size_t... Is>
    static auto make_output_channel_tuple(std::index_sequence<Is...>)
        -> std::tuple<Channel<std::tuple_element_t<Is, Tup>>*...>;
    using output_channels_t = decltype(make_output_channel_tuple<return_tuple>(
        std::make_index_sequence<output_count>{}));

    void on_input_ready() {
        if (stop_flag_.load(std::memory_order_relaxed)) return;
        std::size_t ready = count_ready(std::make_index_sequence<input_count>{});
        if (ready == input_count) try_submit(compute_priority());
    }

    template<std::size_t... Is>
    std::size_t count_ready(std::index_sequence<Is...>) {
        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 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...>) {
        ((sum += std::get<Is>(input_channels_)->capacity() > 0
            ? float(std::get<Is>(input_channels_)->approx_size())
              / float(std::get<Is>(input_channels_)->capacity())
            : 0.5f), ...);
    }

    /// Submit unless already queued. A wake that arrives while this node is
    /// queued or running is *recorded*, never dropped.
    ///
    /// Wakes are edge-triggered: a channel fires its space callback on the
    /// transition, once. If that lands while queued_ is up, the CAS below fails
    /// and — before wake_pending_ — the wake was gone. A node could then park a
    /// value, release its worker, and sleep forever holding output its consumer
    /// was waiting for, with every worker idle in cond_wait and nothing left to
    /// re-trigger it. Recording the drop turns the signal level-triggered: the
    /// invariant is that a node never sleeps with a wake outstanding, enforced
    /// by release_and_recheck() at every point that releases the node.
    void try_submit(float priority) {
        bool expected = false;
        if (queued_.compare_exchange_strong(expected, true, std::memory_order_acq_rel))
            scheduler_->submit([this] { fire_once(); }, priority);
        else
            wake_pending_.store(true, std::memory_order_release);
    }

    /// Clear queued_, then honour any wake that was dropped while it was up.
    /// Every path that finishes or parks a firing must release the node through
    /// here rather than storing queued_ directly.
    void release_and_recheck(float priority = 0.5f) {
        queued_.store(false, std::memory_order_release);
        if (wake_pending_.exchange(false, std::memory_order_acq_rel))
            try_submit(priority);
    }

    void fire_once() {
        if (stop_flag_.load(std::memory_order_relaxed)) {
            queued_.store(false, std::memory_order_release);
            return;
        }
        auto t0 = clock_t::now();
        int64_t now_us = std::chrono::duration_cast<std::chrono::microseconds>(
            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>{});
                release_and_recheck();
                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) {
                release_and_recheck();
                on_input_ready();
                return;
            }
        }

        try {
            auto args = pop_inputs(std::make_index_sequence<input_count>{});
            auto t1   = clock_t::now();
            stats_.record_queue_wait(duration_t(t1 - t0));
            auto cpu0 = NodeStats::cpu_now();

            if constexpr (std::is_void_v<return_raw>) {
                std::apply([this](auto&&... a) { obj_(std::forward<decltype(a)>(a)...); }, args);
            } else {
                auto result = std::apply([this](auto&&... a) { return obj_(std::forward<decltype(a)>(a)...); }, args);
                push_outputs(normalise(std::move(result)), std::make_index_sequence<output_count>{});
            }

            auto cpu1 = NodeStats::cpu_now();
            auto t2   = clock_t::now();
            stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1);
        } catch (const ChannelClosedError&) {
            fire_callbacks(closed_callbacks_);
            self_stop();
            return;
        } catch (const ChannelOverflowError&) {
            fire_callbacks(event_callbacks_);
        } catch (...) {
            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();
                return;
            }
        }

        stats_.exec_start_us.store(0, std::memory_order_relaxed);
        release_and_recheck();
        if (stop_flag_.load(std::memory_order_relaxed)) return;

        // Parked by the push above. Same situation as the retry path at the top
        // of fire_once — and the same lost-wakeup race, which that path closes
        // and this one did not. A space callback that fired while queued_ was
        // still up got swallowed by try_submit's CAS, and the resubmit below
        // cannot cover it: this firing consumed its input, so inputs are empty
        // and on_input_ready() will not resubmit. The node would then hold its
        // value forever while its consumer waits for exactly that value and its
        // producer parks on an input channel that never drains. Re-check now
        // that the flag is down.
        if constexpr (!std::is_void_v<return_raw>) {
            if (pending_) {
                if (outputs_have_space(std::make_index_sequence<output_count>{}))
                    try_submit(0.5f);
                return;                    // parked
            }
        }

        if constexpr (input_count == 0) try_submit(0.5f);
        else                            on_input_ready();
    }

    template<std::size_t... Is>
    args_tuple pop_inputs(std::index_sequence<Is...>) { return {pop_one<Is>()...}; }

    template<std::size_t I>
    std::tuple_element_t<I, args_tuple> pop_one() {
        auto& ch = *std::get<I>(input_channels_);
        std::tuple_element_t<I, args_tuple> val;
        if (!ch.try_pop_now(val)) throw ChannelClosedError{};
        return val;
    }

    template<typename R = return_raw>
    static return_tuple normalise(R&& r) {
        if constexpr (is_tuple_v<R>) return std::move(r);
        else                         return std::make_tuple(std::move(r));
    }

    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...>) {
        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>
    bool push_one_out(std::tuple_element_t<I, return_tuple>&& val) {
        auto* ch = std::get<I>(output_channels_);
        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 true;
        }
        // See the note on the typed overload above: park rather than block.
        return ch->try_push(val);
    }

    Obj&                                     obj_;
    std::shared_ptr<IScheduler>              scheduler_;
    std::string                              name_;
    std::size_t                              fifo_capacity_;
    input_channels_t                         input_channels_;
    output_channels_t                        output_channels_{};
    std::atomic<bool>                        stop_flag_{true};
    std::atomic<bool>                        queued_{false};
    /// A wake that arrived while queued_ was up. See try_submit.
    std::atomic<bool>                        wake_pending_{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_{};
};

// ── make_pool_node factory (NTTP) ─────────────────────────────────────────────

template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0>
auto make_pool_node(std::shared_ptr<IScheduler> sched, std::size_t fifo_capacity = 5) {
    return PoolNode<Func, in<>, out<>, Label, UniqueTag>(std::move(sched), fifo_capacity);
}

template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
         fixed_string... InNames>
auto make_pool_node(std::shared_ptr<IScheduler> sched, in<InNames...>,
                    std::size_t fifo_capacity = 5) {
    return PoolNode<Func, in<InNames...>, out<>, Label, UniqueTag>(std::move(sched), fifo_capacity);
}

template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
         fixed_string... OutNames>
auto make_pool_node(std::shared_ptr<IScheduler> sched, out<OutNames...>,
                    std::size_t fifo_capacity = 5) {
    return PoolNode<Func, in<>, out<OutNames...>, Label, UniqueTag>(std::move(sched), fifo_capacity);
}

template<auto Func, fixed_string Label = "", std::size_t UniqueTag = 0,
         fixed_string... InNames, fixed_string... OutNames>
auto make_pool_node(std::shared_ptr<IScheduler> sched, in<InNames...>, out<OutNames...>,
                    std::size_t fifo_capacity = 5) {
    return PoolNode<Func, in<InNames...>, out<OutNames...>, Label, UniqueTag>(
        std::move(sched), fifo_capacity);
}

// ── make_pool_node factory (callable object) ──────────────────────────────────

template<typename Obj>
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched,
                    std::size_t fifo_capacity = 5) {
    return PoolObjectNode<Obj, in<>, out<>>(obj, std::move(sched), fifo_capacity);
}

template<typename Obj, fixed_string... InNames>
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched, in<InNames...>,
                    std::size_t fifo_capacity = 5) {
    return PoolObjectNode<Obj, in<InNames...>, out<>>(obj, std::move(sched), fifo_capacity);
}

template<typename Obj, fixed_string... OutNames>
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched, out<OutNames...>,
                    std::size_t fifo_capacity = 5) {
    return PoolObjectNode<Obj, in<>, out<OutNames...>>(obj, std::move(sched), fifo_capacity);
}

template<typename Obj, fixed_string... InNames, fixed_string... OutNames>
auto make_pool_node(Obj& obj, std::shared_ptr<IScheduler> sched,
                    in<InNames...>, out<OutNames...>,
                    std::size_t fifo_capacity = 5) {
    return PoolObjectNode<Obj, in<InNames...>, out<OutNames...>>(
        obj, std::move(sched), fifo_capacity);
}

} // namespace kpn