#pragma once
/// TRACES: VR-015 | PR-004
///
/// Pipeline throughput benchmark — how much of a run is spent in each node.
///
/// The KPN network already counts most of what an optimiser needs, and
/// `print_diagnostics()` throws nearly all of it away: it prints frames and
/// `ema` per node, and passes `elapsed_s = 0`, which zeroes throughput. Two
/// things had to change before "time per node" could be answered honestly.
///
/// **`ema` is not a total.** It is an exponentially weighted average, so
/// `frames * ema` tracks the end of the run rather than the whole of it. On a
/// film that is a real difference — a detector costs one thing in a crowd scene
/// and another over a landscape. `NodeStats::total_exec_us` (added alongside
/// this) is the true sum.
///
/// **Wall time inside a node is not all work.** `PoolObjectNode::fire_once`
/// times the functor *and* `push_outputs`, and `push_outputs` parks on a full
/// downstream channel (AR-004). A node that is merely backpressured therefore
/// bills the time it spent waiting to whoever is ahead of it: SuperHero's
/// `frame_source` reported 141.9 ms/frame against a decoder logging 12-18 ms.
/// Optimising against that number means optimising the fastest node in the
/// graph.
///
/// So each node is reported three ways, and the three together are what
/// identify a cost:
///
///   - `exec_ms`  — cumulative wall time in the node, work *and* parked pushes
///   - `cpu_ms`   — thread CPU time (CLOCK_THREAD_CPUTIME_ID). Backpressure
///                  cannot inflate it, because a parked node holds no thread.
///   - `pressure` — mean fill of its input channels minus that of its outputs
///
/// **`cpu_ms` cannot tell real work from a spinning GPU wait.** CUDA's default
/// sync policy (`cudaDeviceScheduleAuto`) spin-waits before yielding, so
/// `cudaStreamSynchronize` burns the calling thread's CPU while the GPU works.
/// A node that is purely GPU-bound can therefore report a high `cpu_ms` and read
/// as CPU-bound. `cudaSetDeviceFlags(cudaDeviceScheduleBlockingSync)` settles it
/// in one line: if a node's `cpu_ms` collapses under blocking sync, that CPU was
/// spin, not work.
///
/// **`cpu_ms` counts one thread only.** `CLOCK_THREAD_CPUTIME_ID` is per-thread,
/// and OpenCV here is built against TBB, so any node whose functor goes through
/// `cv::parallel_for_` (histogram compare, `warpAffine`, colour conversion) has
/// that work executed on TBB's arena — 19 workers on a 20-core box — and billed
/// to those threads rather than to the node. Such a node reads *cheaper* than it
/// is, and the difference shows up in `stall/f` instead, indistinguishable from a
/// GPU wait. `exec_ms` does capture it, since the functor does not return until
/// the parallel region joins: a node where `exec/f` greatly exceeds `cpu/f`
/// while its output channel is empty is fanning out, not waiting.
///
/// Work piles up *in front of* a bottleneck and starves everything *after* it,
/// so `pressure` is maximal at the node setting the pace. `cpu_ms` then says
/// which repair applies: high pressure with a saturated thread is CPU-bound and
/// the work must get cheaper, while high pressure with an idle thread is
/// waiting on a device, where batch size and engine precision are the knobs.
///
/// Occupancy has to be sampled during the run. `current_fill` is instantaneous
/// and every channel has drained by the time the network stops, so a single
/// read at the end reports an idle pipeline however congested it was.
///
/// Nothing here is specific to this pipeline's topology: the node graph is
/// recovered from KPN's channel names, so it keeps working when the graph
/// changes.

#include <kpn/diagnostics.hpp>

#include <nlohmann/json.hpp>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <functional>
#include <iomanip>
#include <map>
#include <ostream>
#include <sstream>
#include <string>
#include <thread>
#include <vector>

namespace sae::bench {

// ── Edge naming ──────────────────────────────────────────────────────────────

/// TRACES: VR-015 | PR-004
/// KPN names a channel "<src>:<idx> → <dst>:<idx>" (static_network.hpp).
/// Recovering the two node names from it is what keeps attribution
/// topology-agnostic: the graph is read back out of the channel names rather
/// than hard-coded here, so a new node or a re-wired branch needs no change.
/// Leaves both outputs untouched if the name does not carry an arrow.
inline void split_edge_name(const std::string& name,
                            std::string& producer, std::string& consumer) {
    static const std::string kArrow = " \xe2\x86\x92 ";   // " → "
    const auto arrow = name.find(kArrow);
    if (arrow == std::string::npos) return;
    auto strip_port = [](std::string s) {
        const auto colon = s.rfind(':');
        return colon == std::string::npos ? s : s.substr(0, colon);
    };
    producer = strip_port(name.substr(0, arrow));
    consumer = strip_port(name.substr(arrow + kArrow.size()));
}

// ── Channel occupancy, time-averaged ─────────────────────────────────────────

/// TRACES: VR-015 | PR-004
/// One channel's fill level integrated over the run. `peak_fill` is already
/// cumulative in `ChannelStats`, but a peak cannot distinguish "full once" from
/// "full throughout", and those are opposite diagnoses. A mean can.
struct ChannelOccupancy {
    std::string   name;            // "src:0 → dst:0", as KPN names it
    std::string   producer;        // node name left of the arrow
    std::string   consumer;        // node name right of the arrow
    std::size_t   capacity{0};
    std::uint64_t samples{0};
    double        fill_sum{0.0};
    std::uint64_t samples_full{0};
    std::uint64_t samples_empty{0};

    // Final-snapshot totals (monotonic counters, so the last read is the total).
    std::size_t   peak_fill{0};
    std::uint64_t pushes{0};
    std::uint64_t pops{0};
    std::uint64_t drops{0};
    std::uint64_t overflows{0};
    std::uint64_t bytes_pushed{0};

    double mean_fill()     const { return samples ? fill_sum / static_cast<double>(samples) : 0.0; }
    double mean_fill_pct() const { return capacity ? 100.0 * mean_fill() / static_cast<double>(capacity) : 0.0; }
    double peak_pct()      const { return capacity ? 100.0 * static_cast<double>(peak_fill) / static_cast<double>(capacity) : 0.0; }
    double full_pct()      const { return samples ? 100.0 * static_cast<double>(samples_full)  / static_cast<double>(samples) : 0.0; }
    double empty_pct()     const { return samples ? 100.0 * static_cast<double>(samples_empty) / static_cast<double>(samples) : 0.0; }
    double bandwidth_mbs(double wall_s) const {
        return wall_s > 0.0 ? static_cast<double>(bytes_pushed) / wall_s / 1e6 : 0.0;
    }
};

// ── Per-node attributed cost ─────────────────────────────────────────────────

/// TRACES: VR-015 | PR-004
struct NodeCost {
    std::string   name;
    std::uint64_t frames{0};

    // Cumulative wall time in the node — the answer to "where did the run go",
    // but only for a node that is not backpressured; it includes parked pushes.
    double exec_ms{0.0};
    double exec_ms_per_frame{0.0};   // true mean, not the EMA
    double exec_share{0.0};          // exec_ms / wall_ms, 0..1
    double ema_exec_ms{0.0};         // KPN's EMA, kept for continuity with the old report
    double max_exec_ms{0.0};

    // Thread CPU time: excludes sleeping, parking and waiting on a device, so it
    // is the one number backpressure cannot inflate.
    double cpu_ms{0.0};
    double cpu_ms_per_frame{0.0};
    double cpu_share{0.0};           // cpu_ms / wall_ms — thread saturation, 0..1
    double cpu_pct_of_pipeline{0.0}; // this node's share of all nodes' CPU time

    // Per frame, time inside the node not spent on its own CPU: parked on a
    // full output channel, or waiting on the GPU. `pressure` separates those —
    // a backpressured node has a full output, a device-bound one does not.
    double stall_ms_per_frame{0.0};

    // Queue occupancy either side of the node, in percent of capacity.
    double in_fill_pct{0.0};
    double out_fill_pct{0.0};
    double pressure{0.0};            // in − out; maximal at the pacing node
    bool   has_input{false};
    bool   has_output{false};

    double queue_wait_ms{0.0};
    bool   is_bottleneck{false};

    /// TRACES: VR-015, AR-004 | PR-004
    /// Live scheduling state, so a wedged run says *why* it is wedged rather
    /// than only that it is. With `queued=0, wake=1` a wake was recorded and
    /// never consumed; with `queued=0, wake=0` and a full input, no wake was
    /// ever generated. Those are different bugs in different files.
    bool   queued{false};
    bool   wake_pending{false};
};

/// TRACES: VR-015 | PR-004
/// Attribute cost to nodes from a KPN node snapshot plus sampled channel
/// occupancy. Pure — no clocks, no threads, no network — so the ranking is
/// unit-testable on CI hardware that can never run the pipeline itself.
///
/// A node with several inputs takes the **minimum** input fill: it can only run
/// once every input has data, so the emptiest one gates it, and a full sibling
/// channel means that channel's producer is blocked rather than this node being
/// slow. A node with several outputs takes the **maximum** output fill, since
/// parking on any one branch stops the node.
///
/// Terminals are the infinite-reservoir limit of the same rule: a source has
/// unlimited work available (input treated as 100% full) and a sink unlimited
/// drain (output treated as empty), so both stay rankable against the interior
/// nodes instead of dropping out of the comparison.
inline std::vector<NodeCost> attribute_cost(
    const std::vector<kpn::NodeSnapshot>&  nodes,
    const std::vector<ChannelOccupancy>&   channels,
    double                                 wall_sec)
{
    const double wall_ms = wall_sec * 1000.0;

    double cpu_total = 0.0;
    for (const auto& n : nodes) cpu_total += n.total_cpu_ms;

    std::vector<NodeCost> out;
    out.reserve(nodes.size());

    for (const auto& n : nodes) {
        NodeCost c;
        c.name          = n.name;
        c.frames        = n.frames_processed;
        c.exec_ms       = n.total_exec_ms;
        c.ema_exec_ms   = n.ema_exec_ms;
        c.max_exec_ms   = n.max_exec_ms;
        c.cpu_ms        = n.total_cpu_ms;
        c.queue_wait_ms = n.queue_wait_ms;
        c.queued        = n.queued;
        c.wake_pending  = n.wake_pending;

        c.exec_ms_per_frame   = c.frames ? c.exec_ms / static_cast<double>(c.frames) : 0.0;
        c.cpu_ms_per_frame    = c.frames ? c.cpu_ms  / static_cast<double>(c.frames) : 0.0;
        c.exec_share          = wall_ms > 0.0 ? c.exec_ms / wall_ms : 0.0;
        c.cpu_share           = wall_ms > 0.0 ? c.cpu_ms  / wall_ms : 0.0;
        c.cpu_pct_of_pipeline = cpu_total > 0.0 ? 100.0 * c.cpu_ms / cpu_total : 0.0;
        c.stall_ms_per_frame  = c.exec_ms_per_frame - c.cpu_ms_per_frame;
        if (c.stall_ms_per_frame < 0.0) c.stall_ms_per_frame = 0.0;

        double in_min  = 0.0; bool have_in  = false;
        double out_max = 0.0; bool have_out = false;
        for (const auto& ch : channels) {
            if (ch.consumer == n.name) {
                const double f = ch.mean_fill_pct();
                if (!have_in || f < in_min) in_min = f;
                have_in = true;
            }
            if (ch.producer == n.name) {
                const double f = ch.mean_fill_pct();
                if (!have_out || f > out_max) out_max = f;
                have_out = true;
            }
        }
        c.has_input    = have_in;
        c.has_output   = have_out;
        c.in_fill_pct  = have_in  ? in_min  : 100.0;  // source: always has work
        c.out_fill_pct = have_out ? out_max :   0.0;  // sink: never blocks
        c.pressure     = c.in_fill_pct - c.out_fill_pct;
        out.push_back(std::move(c));
    }

    // Rank, but only among nodes that actually ran: a node with zero frames has
    // no cost to attribute and its neighbouring channels never moved.
    auto best = out.end();
    for (auto it = out.begin(); it != out.end(); ++it) {
        if (it->frames == 0) continue;
        if (best == out.end() || it->pressure > best->pressure) best = it;
    }
    if (best != out.end()) best->is_bottleneck = true;

    return out;
}

/// TRACES: VR-015 | PR-004
/// One line of plain English about the winning node, since the point of the
/// report is to say what to change next. A saturated thread means the node's
/// own work is the limit; an idle thread under pressure means it is waiting on
/// a device, and those are different repairs.
inline std::string verdict(const std::vector<NodeCost>& costs) {
    for (const auto& c : costs) {
        if (!c.is_bottleneck) continue;
        std::ostringstream os;
        os << std::fixed << c.name << " sets the pace: ";

        // A source's 100% input is the infinite-reservoir convention, not a
        // measured queue — saying "work is backed up in front of it" would be
        // asserting something no counter observed.
        if (!c.has_input)
            os << "nothing downstream is waiting on it (output "
               << std::setprecision(1) << c.out_fill_pct
               << "% full), so the pipeline is running as fast as this node can feed it. ";
        else if (!c.has_output)
            os << std::setprecision(1) << c.in_fill_pct
               << "% full input and nothing to block on, so it is the drain. ";
        else
            os << std::setprecision(1) << c.in_fill_pct << "% full input, "
               << c.out_fill_pct << "% full output. ";

        os << std::setprecision(2) << c.cpu_ms_per_frame << " ms/frame on CPU. ";

        if (c.cpu_share >= 0.85)
            os << "CPU-bound — its thread is busy " << std::setprecision(0)
               << (100.0 * c.cpu_share) << "% of the run, so the work itself has to get"
                  " cheaper or be split across more threads.";
        else if (c.cpu_share <= 0.35 && c.stall_ms_per_frame > c.cpu_ms_per_frame)
            os << "Device-bound — its thread is busy only " << std::setprecision(0)
               << (100.0 * c.cpu_share) << "% of the run and it spends "
               << std::setprecision(2) << c.stall_ms_per_frame
               << " ms/frame off-CPU, so it is waiting on the GPU or the disk: batch size,"
                  " engine precision and the decode path are the knobs, not the C++.";
        else
            os << "Mixed — thread busy " << std::setprecision(0) << (100.0 * c.cpu_share)
               << "% of the run, " << std::setprecision(2) << c.stall_ms_per_frame
               << " ms/frame off-CPU.";
        return os.str();
    }
    return "no node processed a frame — nothing to attribute";
}

// ── Recorder ─────────────────────────────────────────────────────────────────

/// TRACES: VR-015 | PR-004
/// Samples the live network on a timer and emits the report at the end.
///
/// The sampler only reads relaxed atomics, so it does not perturb what it
/// measures — which matters, since this exists to be trusted as a timing
/// measurement.
class BenchmarkRecorder {
public:
    using Sampler = std::function<kpn::NetworkSnapshot()>;

    explicit BenchmarkRecorder(int sample_interval_ms = 100)
        : interval_(std::chrono::milliseconds(sample_interval_ms)) {}

    ~BenchmarkRecorder() { stop(); }

    void start(Sampler sampler) {
        sampler_ = std::move(sampler);
        running_.store(true, std::memory_order_release);
        thread_ = std::thread([this] {
            while (running_.load(std::memory_order_acquire)) {
                accumulate(sampler_());
                std::this_thread::sleep_for(interval_);
            }
        });
    }

    /// Stops sampling and latches the final counter values. Call while the
    /// network object is still alive: the monotonic counters stay valid after
    /// `net.stop()`, but they die with the object.
    void stop() {
        if (!running_.exchange(false, std::memory_order_acq_rel)) return;
        if (thread_.joinable()) thread_.join();
        if (sampler_) {
            final_ = sampler_();
            // Occupancy is deliberately NOT accumulated from this last read:
            // the pipeline has drained by now, and folding an idle sample into
            // the mean biases every channel toward "never congested".
            for (const auto& ch : final_.channels) {
                auto& occ = occupancy_[ch.name];
                if (occ.name.empty()) {   // a channel that never moved
                    occ.name     = ch.name;
                    occ.capacity = ch.capacity;
                    split_edge_name(ch.name, occ.producer, occ.consumer);
                }
                occ.peak_fill    = ch.peak_fill;
                occ.pushes       = ch.pushes;
                occ.pops         = ch.pops;
                occ.drops        = ch.drops;
                occ.overflows    = ch.overflows;
                occ.bytes_pushed = ch.bytes_pushed;
            }
        }
        stopped_ = true;
    }

    bool   has_data() const { return stopped_ && !final_.nodes.empty(); }
    double wall_sec() const { return final_.elapsed_s; }

    std::vector<ChannelOccupancy> channels() const {
        std::vector<ChannelOccupancy> v;
        v.reserve(occupancy_.size());
        for (const auto& [_, occ] : occupancy_) v.push_back(occ);
        return v;
    }

    std::vector<NodeCost> costs() const {
        return attribute_cost(final_.nodes, channels(), final_.elapsed_s);
    }

    /// TRACES: VR-015 | PR-004
    /// Machine-readable report, for sweeping configurations and diffing runs.
    /// `film_sec` is the last timestamp the pipeline reached, so
    /// `realtime_factor` answers what the optimiser is really asking: seconds
    /// of film per second of wall clock. It is 0 for a topology with no result
    /// sink (the dump-only path), and the field is then omitted rather than
    /// reported as zero throughput.
    nlohmann::json to_json(const nlohmann::json& run_config, double film_sec) const {
        using nlohmann::json;
        const double wall  = final_.elapsed_s;
        const auto   chans = channels();
        const auto   cost  = attribute_cost(final_.nodes, chans, wall);

        json j;
        j["schema_version"] = 1;
        j["config"]         = run_config;

        json summary;
        summary["wall_sec"]           = wall;
        summary["sample_count"]       = sample_count_;
        summary["sample_interval_ms"] = interval_.count();
        if (film_sec > 0.0) {
            summary["film_sec"]        = film_sec;
            summary["realtime_factor"] = wall > 0.0 ? film_sec / wall : 0.0;
        }
        for (const auto& c : cost)
            if (c.is_bottleneck) { summary["bottleneck"] = c.name; break; }
        summary["verdict"] = verdict(cost);
        j["summary"] = summary;

        json jnodes = json::array();
        for (const auto& c : cost) {
            jnodes.push_back({
                {"name",                c.name},
                {"frames",              c.frames},
                {"fps",                 wall > 0.0 ? c.frames / wall : 0.0},
                {"exec_ms",             c.exec_ms},
                {"exec_ms_per_frame",   c.exec_ms_per_frame},
                {"exec_share",          c.exec_share},
                {"ema_exec_ms",         c.ema_exec_ms},
                {"max_exec_ms",         c.max_exec_ms},
                {"cpu_ms",              c.cpu_ms},
                {"cpu_ms_per_frame",    c.cpu_ms_per_frame},
                {"cpu_share",           c.cpu_share},
                {"cpu_pct_of_pipeline", c.cpu_pct_of_pipeline},
                {"stall_ms_per_frame",  c.stall_ms_per_frame},
                {"queue_wait_ms",       c.queue_wait_ms},
                {"in_fill_pct",         c.in_fill_pct},
                {"out_fill_pct",        c.out_fill_pct},
                {"pressure",            c.pressure},
                {"is_bottleneck",       c.is_bottleneck},
                {"queued",              c.queued},
                {"wake_pending",        c.wake_pending},
            });
        }
        j["nodes"] = std::move(jnodes);

        json jch = json::array();
        for (const auto& ch : chans) {
            jch.push_back({
                {"name",          ch.name},
                {"producer",      ch.producer},
                {"consumer",      ch.consumer},
                {"capacity",      ch.capacity},
                {"mean_fill",     ch.mean_fill()},
                {"mean_fill_pct", ch.mean_fill_pct()},
                {"peak_fill",     ch.peak_fill},
                {"peak_pct",      ch.peak_pct()},
                {"full_pct",      ch.full_pct()},
                {"empty_pct",     ch.empty_pct()},
                {"pushes",        ch.pushes},
                {"pops",          ch.pops},
                {"drops",         ch.drops},
                {"overflows",     ch.overflows},
                {"mb_per_sec",    ch.bandwidth_mbs(wall)},
            });
        }
        j["channels"] = std::move(jch);
        return j;
    }

    /// TRACES: VR-015 | PR-004
    /// Human-readable form of the same data, so a run is legible without
    /// opening the JSON.
    void print(std::ostream& os, double film_sec) const {
        print_impl(os, final_, film_sec);
    }

    /// TRACES: VR-015, AR-004 | PR-004
    /// Dump the report from a LIVE snapshot, mid-run, without stopping anything.
    ///
    /// A report that only exists at shutdown is no use against the failure this
    /// pipeline actually has: a wedged run never reaches shutdown, so the one
    /// moment the numbers matter most is the one moment they were unavailable.
    /// Channel occupancy names the stalled node directly — it is the one whose
    /// input is full and whose output is empty — which is otherwise a debug-build
    /// and a gdb session away.
    ///
    /// Safe to call from the wait loop while the pipeline is running or hung: it
    /// takes the same lock-free snapshot the sampler does.
    void dump_live(std::ostream& os, double film_sec) const {
        if (!sampler_) { os << "[benchmark] no sampler — run with --benchmark\n"; return; }
        print_impl(os, sampler_(), film_sec);
    }

private:
    void print_impl(std::ostream& os, const kpn::NetworkSnapshot& snap,
                    double film_sec) const {
        const double wall  = snap.elapsed_s;
        const auto   chans = channels();
        const auto   cost  = attribute_cost(snap.nodes, chans, wall);

        os << "\n┌─ Pipeline benchmark (VR-015) ──────────────────────────────────────────────\n";
        os << "│  wall " << std::fixed << std::setprecision(1) << wall << "s";
        if (film_sec > 0.0)
            os << "   film " << film_sec << "s   realtime x" << std::setprecision(2)
               << (wall > 0.0 ? film_sec / wall : 0.0);
        os << "   samples " << sample_count_ << "\n│\n";

        os << "│  node              frames     cpu_s  cpu%run  cpu%tot   cpu/f"
              "  exec/f  stall/f    in%   out%   press  q/w\n";
        for (const auto& c : cost) {
            os << "│  " << (c.is_bottleneck ? "▶ " : "  ") << std::left << std::setw(16)
               << c.name << std::right
               << std::setw(7) << c.frames
               << std::setw(10) << std::setprecision(1) << (c.cpu_ms / 1000.0)
               << std::setw(9)  << std::setprecision(0) << (100.0 * c.cpu_share)
               << std::setw(9)  << std::setprecision(0) << c.cpu_pct_of_pipeline
               << std::setw(8)  << std::setprecision(2) << c.cpu_ms_per_frame
               << std::setw(8)  << std::setprecision(2) << c.exec_ms_per_frame
               << std::setw(9)  << std::setprecision(2) << c.stall_ms_per_frame
               << std::setw(7)  << std::setprecision(0) << c.in_fill_pct
               << std::setw(7)  << std::setprecision(0) << c.out_fill_pct
               << std::setw(8)  << std::setprecision(1) << c.pressure
               << "  " << int(c.queued) << "/" << int(c.wake_pending)
               << "\n";
        }

        /// TRACES: VR-015, AR-004 | PR-004
        // Fires only when the scheduling state is actually wrong, so a healthy
        // run stays quiet and a wedged one names the fault — instead of leaving
        // it to be reconstructed under a debugger that suppresses the bug.
        for (const auto& c : cost) {
            if (c.queued || !c.has_input) continue;
            if (c.wake_pending)
                os << "│  !! " << c.name << " idle with a wake outstanding"
                      " (queued=0 wake=1): the wake was recorded and never"
                      " consumed — submit/release handshake.\n";
            else if (c.in_fill_pct > 50.0)
                os << "│  !! " << c.name << " idle with a "
                   << std::setprecision(0) << c.in_fill_pct
                   << "% full input and no wake pending: the wake was never"
                      " generated — channel edge detection.\n";
        }

        os << "│\n│  channel                              cap  mean%  peak%  full%"
              " empty%     MB/s\n";
        for (const auto& ch : chans) {
            os << "│  " << std::left << std::setw(36) << ch.name << std::right
               << std::setw(5) << ch.capacity
               << std::setw(7) << std::setprecision(1) << ch.mean_fill_pct()
               << std::setw(7) << ch.peak_pct()
               << std::setw(7) << ch.full_pct()
               << std::setw(7) << ch.empty_pct()
               << std::setw(9) << std::setprecision(1) << ch.bandwidth_mbs(wall)
               << "\n";
        }
        os << "│\n│  " << verdict(cost) << "\n";
        os << "└────────────────────────────────────────────────────────────────────────────\n";
        os << "  cpu_s / cpu%tot is where the run's compute actually went. exec/f is wall\n"
              "  time in the node INCLUDING time parked on a full output channel, so it\n"
              "  overstates a backpressured node — compare it against cpu/f, which cannot\n"
              "  be inflated that way. press = input fill − output fill, and locates the\n"
              "  node that work is queueing up in front of.\n"
              "  cpu_s counts THIS node's thread only: work OpenCV fans out via TBB is\n"
              "  billed to the TBB arena, so a node using cv::parallel_for_ reads cheaper\n"
              "  than it is and the difference surfaces in stall/f.\n";
    }

private:
    void accumulate(const kpn::NetworkSnapshot& snap) {
        ++sample_count_;
        for (const auto& ch : snap.channels) {
            auto& occ = occupancy_[ch.name];
            if (occ.name.empty()) {
                occ.name     = ch.name;
                occ.capacity = ch.capacity;
                split_edge_name(ch.name, occ.producer, occ.consumer);
            }
            occ.fill_sum += static_cast<double>(ch.current_fill);
            ++occ.samples;
            if (ch.capacity && ch.current_fill >= ch.capacity) ++occ.samples_full;
            if (ch.current_fill == 0)                          ++occ.samples_empty;
        }
    }

    std::chrono::milliseconds               interval_;
    Sampler                                 sampler_;
    std::thread                             thread_;
    std::atomic<bool>                       running_{false};
    bool                                    stopped_{false};
    std::uint64_t                           sample_count_{0};
    std::map<std::string, ChannelOccupancy> occupancy_;
    kpn::NetworkSnapshot                    final_{};
};

} // namespace sae::bench