digest_observer.hpp

#pragma once

// ============================================================================
// DigestObserver — LLM-Friendly Taskflow Profiling
// ============================================================================
//
// Replaces 321MB raw traces with a ~2KB structured digest that AI agents
// (and humans) can immediately act on. Uses O(1) memory per worker via
// streaming accumulators — no individual task spans are stored.
//
// Usage:
//   tf::Executor executor(8);
//   auto obs = executor.make_observer<tf::DigestObserver>();
//   executor.run(taskflow).wait();
//   obs->digest(std::cerr);          // print to stderr
//   std::string s = obs->digest();   // or capture as string
//
// Memory: ~2KB total (vs ~450MB for TFProfObserver with 5.6M tasks)
// Output: ~2KB structured text diagnosing 7 performance patterns
//
// ============================================================================

#include "../core/observer.hpp"
#include <mutex>
#include <queue>
#include <stack>
#include <array>
#include <algorithm>
#include <cmath>
#include <iomanip>
#include <sstream>
#include <limits>
#include <optional>

namespace tf {

// ─── Per-Worker Accumulator (lock-free, each worker writes its own) ─────────

struct DigestAccumulator {

  // task count & duration stats
  size_t task_count = 0;
  size_t total_duration_us = 0;
  size_t min_duration_us = std::numeric_limits<size_t>::max();
  size_t max_duration_us = 0;

  // duration histogram (5 buckets)
  size_t bucket_0us = 0;          // sub-microsecond (<1μs)
  size_t bucket_1us = 0;          // exactly 1μs
  size_t bucket_2_10us = 0;       // 2–10μs
  size_t bucket_11_100us = 0;     // 11–100μs
  size_t bucket_over_100us = 0;   // >100μs

  // time bounds (for wall-time and parallelism shape)
  observer_stamp_t first_start;
  observer_stamp_t last_end;
  bool has_tasks = false;

  // idle-gap attribution (5-bucket histogram on log scale)
  // NOTE: these are MICRO-gaps between consecutive tasks on the SAME worker.
  // They do NOT capture macro-idle periods where a worker had no work at all —
  // that's computed at digest time as (wall - busy).
  // The "long-gap" boundary is computed at digest time relative to measured
  // scheduling cost (100 × sched_cost), then snapped to the nearest decade.
  size_t gap_sub_1us = 0;         // < 1μs — quick steal; ≈ scheduling cost
  size_t gap_1_10us = 0;          // [1, 10μs)
  size_t gap_10_100us = 0;        // [10, 100μs)
  size_t gap_100us_1ms = 0;       // [100μs, 1ms)
  size_t gap_over_1ms = 0;        // ≥ 1ms
  size_t total_gap_us = 0;
  // sum/count of sub-1μs gap durations in ns — used to calibrate scheduling cost from data
  size_t gap_quick_ns_sum = 0;
  size_t gap_quick_ns_count = 0;
  observer_stamp_t prev_end;

  // parallelism shape — busy time in 7 log-time buckets
  //   0: 0–1ms   1: 1–10ms   2: 10–100ms   3: 100ms–1s
  //   4: 1–10s   5: 10–100s  6: 100s+
  std::array<size_t, 7> time_bucket_busy_us = {};
};

// ─── Top-K Outlier Reservoir ────────────────────────────────────────────────

struct OutlierEntry {
  std::string name;
  size_t duration_us;
  size_t worker_id;
  bool operator>(const OutlierEntry& o) const {
    return duration_us > o.duration_us;
  }
};

// ─── DigestObserver Class ───────────────────────────────────────────────────

/**
@class DigestObserver

@brief Observer that produces an LLM-friendly profiling digest in O(1) memory.

Instead of storing every task span (which can reach 300+ MB), DigestObserver
maintains lightweight streaming accumulators per worker and outputs a ~2KB
structured summary that diagnoses 7 common performance patterns:
  1. Scheduling overhead (tasks too fine-grained)
  2. Serial bottleneck (low parallelism)
  3. Load imbalance (uneven worker utilization)
  4. Straggler tasks (outliers dominating runtime)
  5. Under-decomposition (too few tasks)
  6. Recursive overlap (efficiency > 100%)
  7. Healthy (no issues)
*/
class DigestObserver : public ObserverInterface {

  static constexpr size_t K_OUTLIERS = 10;
  // scheduling cost is calibrated from observed sub-1μs gaps, not hardcoded.
  // fallback used only when no quick gaps were observed.
  static constexpr double SCHEDULING_COST_FALLBACK_US = 0.5;

  public:

  /** @brief prints the profiling digest to an output stream */
  void digest(std::ostream& os) const;

  /** @brief returns the profiling digest as a string */
  std::string digest() const;

  /** @brief resets all accumulators for a fresh profiling run */
  void clear();

  /** @brief returns total tasks observed across all workers */
  size_t num_tasks() const;

  /** @brief returns number of workers */
  size_t num_workers() const { return _accumulators.size(); }

  // ObserverInterface overrides (must be public for executor access)
  void set_up(size_t num_workers) override final;
  void on_entry(WorkerView wv, TaskView tv) override final;
  void on_exit(WorkerView wv, TaskView tv) override final;

  private:

  observer_stamp_t _origin;
  std::vector<DigestAccumulator> _accumulators;
  std::vector<std::stack<observer_stamp_t>> _stacks;

  // outlier reservoir — mutex-protected, rarely contended
  mutable std::mutex _outlier_mutex;
  std::priority_queue<
    OutlierEntry,
    std::vector<OutlierEntry>,
    std::greater<OutlierEntry>
  > _top_outliers;

  // helper: assign a timestamp to a log-time bucket index
  static size_t _time_bucket(size_t offset_us);
};

// ─── Implementation ─────────────────────────────────────────────────────────

inline void DigestObserver::set_up(size_t num_workers) {
  _origin = observer_stamp_t::clock::now();
  _accumulators.resize(num_workers);
  _stacks.resize(num_workers);
}

inline void DigestObserver::on_entry(WorkerView wv, TaskView) {
  _stacks[wv.id()].push(observer_stamp_t::clock::now());
}

inline void DigestObserver::on_exit(WorkerView wv, TaskView tv) {

  using namespace std::chrono;

  size_t w = wv.id();
  assert(!_stacks[w].empty());

  auto beg = _stacks[w].top();
  _stacks[w].pop();
  auto end = observer_stamp_t::clock::now();
  size_t dur = duration_cast<microseconds>(end - beg).count();

  auto& a = _accumulators[w];

  // ── basic stats ──
  a.task_count++;
  a.total_duration_us += dur;
  a.min_duration_us = (std::min)(a.min_duration_us, dur);
  a.max_duration_us = (std::max)(a.max_duration_us, dur);

  // ── duration histogram ──
  if      (dur == 0)   a.bucket_0us++;
  else if (dur <= 1)   a.bucket_1us++;
  else if (dur <= 10)  a.bucket_2_10us++;
  else if (dur <= 100) a.bucket_11_100us++;
  else                 a.bucket_over_100us++;

  // ── time bounds ──
  if (!a.has_tasks) { a.first_start = beg; a.has_tasks = true; }
  a.last_end = end;

  // ── idle-gap attribution (ns precision, so we don't round sub-μs gaps to 0) ──
  if (a.task_count > 1) {
    size_t gap_ns = duration_cast<nanoseconds>(beg - a.prev_end).count();
    if (gap_ns < 1000) {
      a.gap_sub_1us++;
      a.gap_quick_ns_sum += gap_ns;
      a.gap_quick_ns_count++;
    }
    else if (gap_ns < 10000)    a.gap_1_10us++;
    else if (gap_ns < 100000)   a.gap_10_100us++;
    else if (gap_ns < 1000000)  a.gap_100us_1ms++;
    else                        a.gap_over_1ms++;
    a.total_gap_us += gap_ns / 1000;
  }
  a.prev_end = end;

  // ── parallelism shape (log-time bucket) ──
  size_t mid_us = duration_cast<microseconds>(
    beg + (end - beg) / 2 - _origin
  ).count();
  a.time_bucket_busy_us[_time_bucket(mid_us)] += dur;

  // ── top-K outlier reservoir ──
  // only lock when this task is a plausible outlier (>3× running mean)
  if (a.task_count <= K_OUTLIERS ||
      dur * a.task_count > a.total_duration_us * 3) {
    std::lock_guard<std::mutex> lk(_outlier_mutex);
    if (_top_outliers.size() < K_OUTLIERS) {
      _top_outliers.push({std::string(tv.name()), dur, w});
    } else if (dur > _top_outliers.top().duration_us) {
      _top_outliers.pop();
      _top_outliers.push({std::string(tv.name()), dur, w});
    }
  }
}

inline size_t DigestObserver::_time_bucket(size_t us) {
  if (us < 1000)         return 0;   // 0–1ms
  if (us < 10000)        return 1;   // 1–10ms
  if (us < 100000)       return 2;   // 10–100ms
  if (us < 1000000)      return 3;   // 100ms–1s
  if (us < 10000000)     return 4;   // 1–10s
  if (us < 100000000)    return 5;   // 10–100s
  return 6;                          // 100s+
}

inline size_t DigestObserver::num_tasks() const {
  size_t n = 0;
  for (auto& a : _accumulators) n += a.task_count;
  return n;
}

inline void DigestObserver::clear() {
  for (auto& a : _accumulators) a = DigestAccumulator{};
  for (auto& s : _stacks) while (!s.empty()) s.pop();
  std::lock_guard<std::mutex> lk(_outlier_mutex);
  while (!_top_outliers.empty()) _top_outliers.pop();
}

// ─── Digest Output ──────────────────────────────────────────────────────────

inline void DigestObserver::digest(std::ostream& os) const {

  using namespace std::chrono;

  const size_t W = _accumulators.size();
  if (W == 0) { os << "(no workers)\n"; return; }

  // ── aggregate across workers ──
  size_t total_tasks = 0, total_busy = 0;
  size_t global_max_dur = 0;
  size_t hist[5] = {};
  size_t quick_gap_ns_sum = 0, quick_gap_ns_count = 0;
  std::optional<observer_stamp_t> wall_beg, wall_end;

  for (auto& a : _accumulators) {
    total_tasks += a.task_count;
    total_busy  += a.total_duration_us;
    global_max_dur = (std::max)(global_max_dur, a.max_duration_us);
    hist[0] += a.bucket_0us;
    hist[1] += a.bucket_1us;
    hist[2] += a.bucket_2_10us;
    hist[3] += a.bucket_11_100us;
    hist[4] += a.bucket_over_100us;
    quick_gap_ns_sum   += a.gap_quick_ns_sum;
    quick_gap_ns_count += a.gap_quick_ns_count;
    if (a.has_tasks) {
      wall_beg = wall_beg ? (std::min)(*wall_beg, a.first_start) : a.first_start;
      wall_end = wall_end ? (std::max)(*wall_end, a.last_end)     : a.last_end;
    }
  }

  // data-driven scheduling cost: mean of sub-1μs inter-task gaps ≈ dispatch cost
  double sched_cost_us = quick_gap_ns_count > 0
    ? (quick_gap_ns_sum / static_cast<double>(quick_gap_ns_count)) / 1000.0
    : SCHEDULING_COST_FALLBACK_US;

  if (total_tasks == 0) { os << "(no tasks observed)\n"; return; }

  size_t wall_us = (wall_beg && wall_end)
    ? duration_cast<microseconds>(*wall_end - *wall_beg).count()
    : 0;

  double mean_dur = static_cast<double>(total_busy) / total_tasks;
  double efficiency = (wall_us > 0)
    ? static_cast<double>(total_busy) / (wall_us * W) * 100.0
    : 0.0;
  double overhead_est = total_tasks * sched_cost_us;
  double overhead_pct = (total_busy + overhead_est > 0)
    ? overhead_est / (total_busy + overhead_est) * 100.0
    : 0.0;

  // ── header ──
  os << "\n"
     << "╔══════════════════════════════════════════════════════════════╗\n"
     << "║            TASKFLOW  PROFILING  DIGEST                     ║\n"
     << "╚══════════════════════════════════════════════════════════════╝\n"
     << "\n"
     << "  Workers : " << W << "\n"
     << "  Tasks   : " << total_tasks << "\n"
     << "  Wall    : " << wall_us << " μs"
     << " (" << std::fixed << std::setprecision(2)
     << wall_us / 1e6 << " s)\n"
     << "\n";

  // ── task duration ──
  os << "┌─ TASK DURATION ─────────────────────────────────────────────┐\n";
  os << "│  Mean : " << std::setprecision(3) << mean_dur << " μs"
     << "    Max : " << global_max_dur << " μs\n";
  os << "│\n";
  os << "│  Histogram:\n";

  const char* bucket_labels[] = {"<1μs", "1μs", "2–10μs", "11–100μs", ">100μs"};
  for (int i = 0; i < 5; ++i) {
    double pct = total_tasks > 0 ? hist[i] * 100.0 / total_tasks : 0;
    if (pct < 0.05 && hist[i] == 0) continue;
    int bar = static_cast<int>(pct / 2.5);  // 40 chars = 100%
    os << "│    " << std::setw(8) << bucket_labels[i] << "  "
       << std::setw(5) << std::setprecision(1) << pct << "%  ";
    for (int b = 0; b < bar; ++b) os << "█";
    os << "\n";
  }
  os << "└─────────────────────────────────────────────────────────────┘\n\n";

  // ── efficiency ──
  os << "┌─ EFFICIENCY ────────────────────────────────────────────────┐\n";
  os << "│  Parallel efficiency : "
     << std::setprecision(1) << efficiency << "%\n";
  os << "│  Scheduling overhead : "
     << std::setprecision(1) << overhead_pct << "%"
     << "  (est. " << std::setprecision(0) << overhead_est << " μs"
     << " @ " << std::setprecision(3) << sched_cost_us << " μs/task";
  if (quick_gap_ns_count > 0) {
    os << ", calibrated from " << quick_gap_ns_count << " sub-1μs gaps";
  } else {
    os << ", fallback — no quick gaps observed";
  }
  os << ")\n";

  // efficiency bar
  os << "│\n│  ";
  int eff_bar = static_cast<int>(std::min(efficiency, 100.0) / 2.5);
  for (int b = 0; b < 40; ++b)
    os << (b < eff_bar ? "█" : "░");
  os << "  " << std::setprecision(1) << efficiency << "%\n";
  os << "└─────────────────────────────────────────────────────────────┘\n\n";

  // ── parallelism shape ──
  os << "┌─ PARALLELISM SHAPE (log-time buckets) ─────────────────────┐\n";
  const char* tbucket_labels[] = {
    "0–1ms", "1–10ms", "10–100ms", "100ms–1s", "1–10s", "10–100s", "100s+"
  };
  // aggregate busy time per time-bucket across workers
  std::array<size_t, 7> tb_busy = {};
  std::array<size_t, 7> tb_tasks = {};
  for (auto& a : _accumulators) {
    for (int b = 0; b < 7; ++b) {
      tb_busy[b] += a.time_bucket_busy_us[b];
    }
  }
  // bucket widths: clip the theoretical bucket [start, end] against the
  // actual run window [wall_beg_off, wall_end_off] (offsets from _origin).
  // Without this clipping we divide by the *theoretical* bucket width even
  // when the run only used a fraction of it — which under-reports avg_workers
  // in any partially-filled bucket (typically the tail bucket of the run).
  const size_t bucket_starts_us[7] = {     0,  1000, 10000, 100000, 1000000, 10000000, 100000000};
  const size_t bucket_ends_us[7]   = {  1000, 10000, 100000, 1000000, 10000000, 100000000,
                                        std::numeric_limits<size_t>::max() / 2};
  size_t wall_beg_off_us = wall_beg
    ? duration_cast<microseconds>(*wall_beg - _origin).count() : 0;
  size_t wall_end_off_us = wall_end
    ? duration_cast<microseconds>(*wall_end - _origin).count() : 0;
  size_t bucket_widths[7];
  for (int b = 0; b < 7; ++b) {
    size_t lo = (std::max)(bucket_starts_us[b], wall_beg_off_us);
    size_t hi = (std::min)(bucket_ends_us[b],   wall_end_off_us);
    bucket_widths[b] = (hi > lo) ? (hi - lo) : 1;  // 1 to avoid /0; empty buckets won't display
  }
  for (int b = 0; b < 7; ++b) {
    if (tb_busy[b] == 0) continue;
    double avg_workers = static_cast<double>(tb_busy[b]) / bucket_widths[b];
    if (avg_workers > W) avg_workers = W;  // cap at num workers
    int bar = static_cast<int>(avg_workers / W * 30);
    os << "│  " << std::setw(9) << tbucket_labels[b] << "  avg "
       << std::setw(4) << std::setprecision(1) << avg_workers
       << "/" << W << " workers  ";
    for (int i = 0; i < 30; ++i)
      os << (i < bar ? "█" : "░");
    os << "\n";
  }
  os << "└─────────────────────────────────────────────────────────────┘\n\n";

  // ── load balance ──
  os << "┌─ LOAD BALANCE ──────────────────────────────────────────────┐\n";
  size_t max_busy = 0, min_busy = std::numeric_limits<size_t>::max();
  size_t max_busy_worker = 0;
  for (size_t w = 0; w < W; ++w) {
    auto& a = _accumulators[w];
    if (!a.has_tasks) continue;
    if (a.total_duration_us > max_busy) { max_busy = a.total_duration_us; max_busy_worker = w; }
    min_busy = (std::min)(min_busy, a.total_duration_us);
  }
  double imbalance = (wall_us > 0)
    ? static_cast<double>(max_busy - min_busy) / wall_us * 100.0
    : 0.0;

  for (size_t w = 0; w < W; ++w) {
    auto& a = _accumulators[w];
    double busy_pct = (wall_us > 0)
      ? a.total_duration_us * 100.0 / wall_us
      : 0.0;
    int bar = static_cast<int>(std::min(busy_pct, 100.0) / 5.0);
    os << "│  W" << w << " " << std::setw(8) << a.task_count << " tasks  "
       << std::setw(5) << std::setprecision(1) << busy_pct << "%  ";
    for (int i = 0; i < 20; ++i)
      os << (i < bar ? "█" : "░");
    os << "\n";
  }
  os << "│\n│  Imbalance spread: "
     << std::setprecision(1) << imbalance << "%\n";
  os << "└─────────────────────────────────────────────────────────────┘\n\n";

  // ── idle gaps + macro-idle ──
  // macro_idle = wall - busy. This captures the real dependency-wait/starvation
  // signal: a worker with 900ms macro_idle had NO work for most of the run,
  // which is invisible to inter-task micro-gap tracking.
  //
  // Long-gap boundary: 100 × measured scheduling cost, snapped to the nearest
  // decade among {10μs, 100μs, 1ms} so the bucket counts (collected on a fixed
  // log binning at on_exit time) can be re-aggregated cleanly here.
  double long_threshold_us_target = 100.0 * sched_cost_us;
  size_t long_threshold_us;     // displayed boundary
  enum { CUT_10, CUT_100, CUT_1000 } cut;
  if (long_threshold_us_target < 31.6) {        // sched_cost < 0.316μs → cut at 10μs
    long_threshold_us = 10;   cut = CUT_10;
  } else if (long_threshold_us_target < 316) {  // sched_cost in [0.316, 3.16μs) → 100μs
    long_threshold_us = 100;  cut = CUT_100;
  } else {                                       // sched_cost ≥ 3.16μs → 1ms
    long_threshold_us = 1000; cut = CUT_1000;
  }

  os << "┌─ IDLE BREAKDOWN ────────────────────────────────────────────┐\n";
  os << "│  per-worker: macro-idle (wall - busy) + micro-gap attribution\n";
  os << "│  'long-gap' = >" << long_threshold_us << "μs (≈ 100× sched_cost"
     << ", calibrated " << std::setprecision(3) << sched_cost_us << "μs/task)\n";
  os << "│   long does NOT necessarily mean dep-wait — could be preemption,\n";
  os << "│   cache cold, NUMA.\n│\n";
  size_t global_macro_idle = 0;
  for (size_t w = 0; w < W; ++w) {
    auto& a = _accumulators[w];
    size_t macro_idle = (wall_us > a.total_duration_us) ? (wall_us - a.total_duration_us) : 0;
    global_macro_idle += macro_idle;
    double idle_pct = (wall_us > 0) ? macro_idle * 100.0 / wall_us : 0.0;
    os << "│  W" << w
       << "  idle " << std::setw(5) << std::setprecision(1) << idle_pct << "%";

    // Re-aggregate the 5 fixed-binning buckets into {quick, cont, long}
    // using the calibrated cut chosen above.
    size_t cont = 0, long_ = 0;
    switch (cut) {
      case CUT_10:
        cont  = a.gap_1_10us;
        long_ = a.gap_10_100us + a.gap_100us_1ms + a.gap_over_1ms;
        break;
      case CUT_100:
        cont  = a.gap_1_10us + a.gap_10_100us;
        long_ = a.gap_100us_1ms + a.gap_over_1ms;
        break;
      case CUT_1000:
        cont  = a.gap_1_10us + a.gap_10_100us + a.gap_100us_1ms;
        long_ = a.gap_over_1ms;
        break;
    }
    size_t total_gaps = a.gap_sub_1us + cont + long_;
    if (total_gaps == 0) {
      os << "  (no micro-gaps)\n";
      continue;
    }
    double pct_healthy    = a.gap_sub_1us * 100.0 / total_gaps;
    double pct_contention = cont          * 100.0 / total_gaps;
    double pct_long       = long_         * 100.0 / total_gaps;
    os << "  gaps: "
       << std::setw(3) << std::setprecision(0) << pct_healthy    << "% quick  "
       << std::setw(3) << pct_contention << "% cont  "
       << std::setw(3) << pct_long       << "% long\n";
  }
  double global_idle_pct = (wall_us > 0 && W > 0)
    ? global_macro_idle * 100.0 / (wall_us * W) : 0.0;
  os << "│\n│  Aggregate macro-idle: " << std::setprecision(1)
     << global_idle_pct << "% of worker-time\n";
  os << "└─────────────────────────────────────────────────────────────┘\n\n";

  // ── top outliers ──
  os << "┌─ TOP OUTLIERS ──────────────────────────────────────────────┐\n";
  {
    // copy the heap to iterate (it's small — max 10 entries)
    auto heap_copy = _top_outliers;
    std::vector<OutlierEntry> outliers;
    while (!heap_copy.empty()) {
      outliers.push_back(heap_copy.top());
      heap_copy.pop();
    }
    std::sort(outliers.begin(), outliers.end(),
      [](const auto& a, const auto& b) { return a.duration_us > b.duration_us; });
    for (auto& o : outliers) {
      double ratio = (mean_dur > 0) ? o.duration_us / mean_dur : 0;
      os << "│  W" << o.worker_id
         << "  " << std::setw(20) << o.name
         << "  " << std::setw(10) << o.duration_us << " μs"
         << "  (" << std::setprecision(0) << ratio << "× mean)\n";
    }
    if (outliers.empty()) os << "│  (none)\n";
  }
  os << "└─────────────────────────────────────────────────────────────┘\n\n";

  // ── auto-diagnosis ──
  os << "┌─ DIAGNOSIS ─────────────────────────────────────────────────┐\n";

  int diag_count = 0;

  // SCHEDULING_OVERHEAD trigger: tasks whose mean wall is comparable to the
  // scheduling cost we measured this run. Threshold scales with measured
  // dispatch — on a fast scheduler we flag smaller tasks; on a slow scheduler
  // we flag larger ones. Old behavior used a hardcoded 1.0μs cliff.
  if (mean_dur < 2.0 * sched_cost_us && overhead_pct > 20) {
    os << "│  " << ++diag_count << ". SCHEDULING_OVERHEAD\n"
       << "│     task_mean(" << std::setprecision(3) << mean_dur << "μs)"
       << " < 2×scheduling_cost(~" << std::setprecision(3) << 2.0 * sched_cost_us << "μs)\n"
       << "│     → Coarsen: tile loops, batch tasks, increase chunk size\n";
  }

  // weighted serial fraction — continuous, no arbitrary cliff.
  // serial_frac = 1 - (busy-time-weighted average parallelism / W)
  // A bucket running at W workers contributes 1.0; a bucket running at 1
  // worker contributes 1/W. No binary < 2.0 threshold.
  size_t total_tb_busy = 0;
  for (int b = 0; b < 7; ++b) total_tb_busy += tb_busy[b];
  double weighted_parallelism = 0.0;
  if (total_tb_busy > 0 && W > 0) {
    for (int b = 0; b < 7; ++b) {
      if (tb_busy[b] == 0) continue;
      double avg_w = static_cast<double>(tb_busy[b]) / bucket_widths[b];
      if (avg_w > W) avg_w = W;
      double weight = static_cast<double>(tb_busy[b]) / total_tb_busy;
      weighted_parallelism += (avg_w / W) * weight;
    }
  }
  double serial_frac = (1.0 - weighted_parallelism) * 100.0;

  // relaxed OR-gate: either condition alone can signal a serial bottleneck.
  // The old AND-gate missed cases like 80% serial + 55% efficiency.
  if (serial_frac > 70 || (serial_frac > 50 && efficiency < 50)) {
    os << "│  " << ++diag_count << ". SERIAL_BOTTLENECK\n"
       << "│     ~" << std::setprecision(0) << serial_frac
       << "% busy-weighted serial fraction"
       << "  (efficiency=" << std::setprecision(1) << efficiency << "%)\n"
       << "│     → Restructure DAG, add level-parallel for_each, pipeline\n";
  }

  // identify worst-outlier worker to decide whether LOAD_IMBALANCE and
  // STRAGGLER_TASKS are the same root cause (worst task lives on busiest worker).
  size_t worst_outlier_worker = std::numeric_limits<size_t>::max();
  size_t worst_outlier_dur = 0;
  {
    auto copy = _top_outliers;
    while (!copy.empty()) {
      if (copy.top().duration_us > worst_outlier_dur) {
        worst_outlier_dur = copy.top().duration_us;
        worst_outlier_worker = copy.top().worker_id;
      }
      copy.pop();
    }
  }
  bool straggler_explains_imbalance =
    (worst_outlier_worker == max_busy_worker) &&
    (max_busy > 0) &&
    (worst_outlier_dur * 2 > max_busy);  // outlier dominates busiest worker

  // trimmed-mean straggler ratio: exclude top-K outliers from the denominator
  // so one 890ms task in a 795K-task pool doesn't self-dilute the mean.
  size_t outlier_busy_sum = 0;
  size_t outlier_count = 0;
  {
    auto copy = _top_outliers;
    while (!copy.empty()) {
      outlier_busy_sum += copy.top().duration_us;
      outlier_count++;
      copy.pop();
    }
  }
  double trimmed_mean = mean_dur;
  if (total_tasks > outlier_count && total_busy > outlier_busy_sum) {
    trimmed_mean = static_cast<double>(total_busy - outlier_busy_sum)
                 / static_cast<double>(total_tasks - outlier_count);
  }
  double outlier_ratio = (trimmed_mean > 0) ? global_max_dur / trimmed_mean : 0;

  if (imbalance > 30) {
    os << "│  " << ++diag_count << ". LOAD_IMBALANCE\n"
       << "│     " << std::setprecision(1) << imbalance << "% spread"
       << " (busiest=W" << max_busy_worker << " vs least busy)\n";
    if (straggler_explains_imbalance) {
      os << "│     root cause: single straggler task ("
         << worst_outlier_dur << "μs on W" << worst_outlier_worker
         << ") dominates that worker\n";
    }
    os << "│     → Switch to DynamicPartitioner, rebalance recursion\n";
  }

  // Only fire STRAGGLER_TASKS independently when it ISN'T already explained
  // by LOAD_IMBALANCE — otherwise it's double-counting the same root cause.
  if (outlier_ratio > 50 && !(imbalance > 30 && straggler_explains_imbalance)) {
    os << "│  " << ++diag_count << ". STRAGGLER_TASKS\n"
       << "│     worst task " << std::setprecision(0) << outlier_ratio
       << "× trimmed mean"
       << " (mean excluding top-" << outlier_count << " outliers)\n"
       << "│     → Equalize task sizes, add sequential cutoff\n";
  }

  if (efficiency > 100) {
    os << "│  " << ++diag_count << ". RECURSIVE_OVERLAP\n"
       << "│     efficiency=" << std::setprecision(1) << efficiency
       << "% (nested parallel tasks)\n"
       << "│     → Check base-case cutoff depth\n";
  }

  if (total_tasks < W * 5 && efficiency < 30) {
    os << "│  " << ++diag_count << ". UNDER_DECOMPOSITION\n"
       << "│     only " << total_tasks << " tasks for " << W << " workers\n"
       << "│     → Split work finer, use for_each_index / reduce\n";
  }

  if (diag_count == 0) {
    os << "│  ✓ HEALTHY — no performance issues detected\n"
       << "│    efficiency=" << std::setprecision(1) << efficiency
       << "%  overhead=" << std::setprecision(1) << overhead_pct << "%\n";
  }

  os << "└─────────────────────────────────────────────────────────────┘\n\n";
}

inline std::string DigestObserver::digest() const {
  std::ostringstream oss;
  digest(oss);
  return oss.str();
}

}  // namespace tf