dial9-viewer 0.3.7

#!/usr/bin/env node
// dial9 trace analysis script — generated by `dial9-viewer agents analyze`
// Usage: node analyze.js <trace.bin or directory>
//   --force    Ignore cached results and re-parse all files
//   --sample N Only analyze N evenly-spaced files from a directory
// Modify this script to drill deeper into specific findings.

const fs = require('fs');
const path = require('path');
const { createHistogram } = require('perf_hooks');

// Resolve toolkit files: sibling (toolkit copy) or ui/ (source tree)
function resolve(name) {
  const sibling = path.resolve(__dirname, name);
  if (fs.existsSync(sibling)) return sibling;
  return path.resolve(__dirname, '..', 'ui', name);
}

const { parseTrace, EVENT_TYPES, formatFrame, symbolizeChain, deduplicateSamples } = require(resolve('trace_parser.js'));
const { buildWorkerSpans, attachCpuSamples, buildActiveTaskTimeline,
        computeSchedulingDelays, filterPointsOfInterest, buildSpanData } = require(resolve('trace_analysis.js'));

// ── Helpers ──

function maxBy(arr, fn) { return arr.reduce((m, x) => Math.max(m, fn(x)), -Infinity); }
function minBy(arr, fn) { return arr.reduce((m, x) => Math.min(m, fn(x)), Infinity); }

/** Classify a symbol as kernel, libc, allocator, tokio-runtime, or application code */
function frameKind(sym) {
  if (!sym || sym.startsWith('0x')) return 'unknown';
  // C symbols (no :: separator) — check libc first to avoid __GI_ matching kernel __
  if (!sym.includes('::')) {
    if (/^(__GI_|__libc_|__pthread_|__clone|start_thread|__poll|__epoll|__futex|malloc|free|mmap|munmap|brk|sbrk)/.test(sym)) return 'libc';
    if (/^(je_|arena_|tcache_|extent_|large_|slab_|jemalloc)/.test(sym) || /^(mi_|_mi_|mi_heap_|mimalloc)/.test(sym)) return 'allocator';
    if (/^(__|do_|sys_|entry_|exit_to|irq_|page_|rcu_|synchronize_)/.test(sym)) return 'kernel';
    if (/^(schedule|expand_fd|expand_files|alloc_fd|ep_poll|ep_send|futex_|sock_|inet_|tcp_|vfs_|filp_|fd_install|ksys_|fput|fget)/.test(sym)) return 'kernel';
  }
  // Tokio runtime internals
  if (/^tokio::runtime::(scheduler|task|blocking|context|coop)/.test(sym)) return 'runtime';
  if (/^std::(panicking|sys|thread)::/.test(sym)) return 'runtime';
  return 'app';
}

/** Walk a symbolized stack to find the interesting frames:
 *  - The syscall (first libc frame)
 *  - The blocking kernel operation (kernel frame that isn't generic plumbing)
 *  - The first application frame that triggered it
 *  Returns { syscall, blocker, trigger, classified } */
function diagnoseStack(frames) {
  let syscall = null;
  let trigger = null;
  let blocker = null;
  const classified = frames.map(f => ({ ...f, kind: frameKind(f.symbol), display: formatFrame(f).text }));

  // Generic kernel plumbing — not the interesting part
  const PLUMBING = /^(__schedule|schedule$|schedule_timeout|do_syscall_64|entry_SYSCALL|__x64_sys_|__sys_|__wait_for_common|__wait_rcu_gp)/;

  for (let i = 0; i < classified.length; i++) {
    const f = classified[i];
    if (!syscall && f.kind === 'libc') syscall = f;
    if (!blocker && f.kind === 'kernel' && !PLUMBING.test(f.symbol)) blocker = f;
    if (!trigger && f.kind === 'app') { trigger = f; break; }
  }
  return { syscall, blocker, trigger, classified };
}

function fmtDur(ns) {
  if (ns >= 1e9) return (ns / 1e9).toFixed(2) + 's';
  if (ns >= 1e6) return (ns / 1e6).toFixed(2) + 'ms';
  if (ns >= 1e3) return (ns / 1e3).toFixed(1) + 'µs';
  return ns + 'ns';
}

function fmtRel(ns, minTs) { return fmtDur(ns - minTs); }

// ── Main ──

/** Create an empty accumulator for multi-trace analysis. */
function createAccumulator() {
  return {
    workerIds: new Set(),
    minTs: Infinity, maxTs: -Infinity,
    eventCount: 0, cpuSampleCount: 0,
    onCpuSampleCount: 0, offCpuSampleCount: 0,
    taskSpawnCount: 0, taskAliveAtEnd: 0,
    maxLocalQueue: 0,
    workerStats: {},
    longPolls: [],
    schedDelayTotal: 0, schedDelayHighCount: 0, schedDelayWorst: [],
    queueMax: 0, queueSum: 0, queueCount: 0,
    taskTimelineSamples: [],
    taskSpawnLocs: new Map(),
    taskSpawnTimes: new Map(),
    taskTerminateTimes: new Map(),
    callframeSymbols: new Map(),
    cpuGroupMap: new Map(),
    schedGroupMap: new Map(),
    spanStats: new Map(), // spanName -> Histogram
    pollDurationByLoc: new Map(), // spawnLoc -> Histogram
    schedDelayHist: null, // Histogram
  };
}

/** Accumulate one trace's analysis into the running accumulator. */
function accumulateTrace(acc, trace) {
  const workerIds = [...new Set(
    trace.events.filter(e => e.eventType !== EVENT_TYPES.QueueSample && e.eventType !== EVENT_TYPES.WakeEvent)
      .map(e => e.workerId)
  )].sort((a, b) => a - b);

  const minTs = minBy(trace.events, e => e.timestamp);
  const maxTs = maxBy(trace.events, e => e.timestamp);

  const spans = buildWorkerSpans(trace.events, workerIds, maxTs);
  attachCpuSamples(trace.cpuSamples, spans.workerSpans);
  const taskTimeline = buildActiveTaskTimeline(trace.taskSpawnTimes, trace.taskTerminateTimes);
  const schedDelays = computeSchedulingDelays(spans.workerSpans, workerIds, spans.wakesByTask);

  // Scalars
  for (const w of workerIds) acc.workerIds.add(w);
  if (minTs < acc.minTs) acc.minTs = minTs;
  if (maxTs > acc.maxTs) acc.maxTs = maxTs;
  acc.eventCount += trace.events.length;
  acc.cpuSampleCount += trace.cpuSamples.length;
  const onCpu = trace.cpuSamples.filter(s => s.source === 0);
  const offCpu = trace.cpuSamples.filter(s => s.source === 1);
  acc.onCpuSampleCount += onCpu.length;
  acc.offCpuSampleCount += offCpu.length;
  acc.taskSpawnCount += trace.taskSpawnTimes.size;
  acc.taskAliveAtEnd += trace.taskSpawnTimes.size - trace.taskTerminateTimes.size;
  if (spans.maxLocalQueue > acc.maxLocalQueue) acc.maxLocalQueue = spans.maxLocalQueue;

  // Per-worker stats
  for (const w of workerIds) {
    const ws = acc.workerStats[w] || (acc.workerStats[w] = {
      activeNs: 0, parkNs: 0, ratioSum: 0, activeCount: 0,
      pollCount: 0, parkCount: 0, schedWaits: [],
    });
    const s = spans.workerSpans[w];
    for (const a of s.actives) { ws.activeNs += a.end - a.start; ws.ratioSum += a.ratio; ws.activeCount++; }
    for (const p of s.parks) { ws.parkNs += p.end - p.start; ws.parkCount++; if (p.schedWait > 0) ws.schedWaits.push(p.schedWait); }
    ws.pollCount += s.polls.length;

    for (const p of s.polls) {
      const dur = p.end - p.start;
      // Poll duration histogram by spawn location
      const loc = p.spawnLoc || '(unknown)';
      let h = acc.pollDurationByLoc.get(loc);
      if (!h) { h = createHistogram(); acc.pollDurationByLoc.set(loc, h); }
      h.record(Math.max(1, Math.round(dur)));

      if (dur > 1e6) {
        acc.longPolls.push({ dur, poll: p, worker: w });
        if (acc.longPolls.length > 200) { acc.longPolls.sort((a, b) => b.dur - a.dur); acc.longPolls.length = 100; }
      }
    }
  }

  // Queue depth
  for (const q of spans.queueSamples) {
    if (q.global > acc.queueMax) acc.queueMax = q.global;
    acc.queueSum += q.global;
    acc.queueCount++;
  }

  // Scheduling delays
  for (const sd of schedDelays) {
    acc.schedDelayTotal++;
    // Histogram
    if (!acc.schedDelayHist) { acc.schedDelayHist = createHistogram(); }
    acc.schedDelayHist.record(Math.max(1, Math.round(sd.delay)));

    if (sd.delay > 1e6) {
      acc.schedDelayHighCount++;
      acc.schedDelayWorst.push(sd);
      if (acc.schedDelayWorst.length > 200) { acc.schedDelayWorst.sort((a, b) => b.delay - a.delay); acc.schedDelayWorst.length = 100; }
    }
  }

  // Task timeline
  for (const s of taskTimeline.activeTaskSamples) acc.taskTimelineSamples.push(s);

  // Maps
  for (const [k, v] of trace.taskSpawnLocs) acc.taskSpawnLocs.set(k, v);
  for (const [k, v] of trace.taskSpawnTimes) acc.taskSpawnTimes.set(k, v);
  for (const [k, v] of trace.taskTerminateTimes) acc.taskTerminateTimes.set(k, v);
  for (const [k, v] of trace.callframeSymbols) acc.callframeSymbols.set(k, v);

  // Sample groups
  for (const g of deduplicateSamples(onCpu, trace.callframeSymbols)) {
    const e = acc.cpuGroupMap.get(g.leaf);
    if (e) e.count += g.count; else acc.cpuGroupMap.set(g.leaf, { ...g });
  }
  for (const g of deduplicateSamples(offCpu, trace.callframeSymbols)) {
    const e = acc.schedGroupMap.get(g.leaf);
    if (e) e.count += g.count; else acc.schedGroupMap.set(g.leaf, { ...g });
  }

  // Span durations (native HDR histogram for bounded memory, exact percentiles)
  if (trace.customEvents && trace.customEvents.length > 0) {
    const { spansByWorker } = buildSpanData(trace.customEvents);
    for (const spans of Object.values(spansByWorker)) {
      for (const s of spans) {
        const dur = Math.max(1, Math.round(s.end - s.start));
        let h = acc.spanStats.get(s.spanName);
        if (!h) { h = createHistogram(); acc.spanStats.set(s.spanName, h); }
        h.record(dur);
      }
    }
  }
}

/** Finalize accumulator into the shape reportAnalysis expects. */
function finalizeAccumulator(acc) {
  acc.longPolls.sort((a, b) => b.dur - a.dur);
  acc.schedDelayWorst.sort((a, b) => b.delay - a.delay);
  acc.taskTimelineSamples.sort((a, b) => a.t - b.t);

  const workerIds = [...acc.workerIds].sort((a, b) => a - b);
  const workerSpans = {};
  for (const w of workerIds) {
    const ws = acc.workerStats[w] || { activeNs: 0, parkNs: 0, ratioSum: 0, activeCount: 0, pollCount: 0, parkCount: 0, schedWaits: [] };
    const total = ws.activeNs + ws.parkNs;
    ws.schedWaits.sort((a, b) => b - a);
    workerSpans[w] = {
      utilization: total > 0 ? ws.activeNs / total : 0,
      avgCpuRatio: ws.activeCount > 0 ? ws.ratioSum / ws.activeCount : 0,
      pollCount: ws.pollCount, parkCount: ws.parkCount, activeCount: ws.activeCount,
      schedWaits: ws.schedWaits,
    };
  }

  return {
    workerIds, minTs: acc.minTs, maxTs: acc.maxTs,
    durationMs: (acc.maxTs - acc.minTs) / 1e6,
    eventCount: acc.eventCount, cpuSampleCount: acc.cpuSampleCount,
    onCpuSampleCount: acc.onCpuSampleCount, offCpuSampleCount: acc.offCpuSampleCount,
    taskSpawnCount: acc.taskSpawnCount, taskAliveAtEnd: acc.taskAliveAtEnd,
    maxLocalQueue: acc.maxLocalQueue,
    workerSpans,
    longPolls: acc.longPolls,
    schedDelayStats: { total: acc.schedDelayTotal, highCount: acc.schedDelayHighCount, worst: acc.schedDelayWorst },
    schedDelays: acc.schedDelayWorst,
    queueDepthStats: { max: acc.queueMax, avg: acc.queueCount > 0 ? acc.queueSum / acc.queueCount : 0, samples: acc.queueCount },
    taskTimeline: { activeTaskSamples: acc.taskTimelineSamples },
    taskSpawnLocs: acc.taskSpawnLocs, taskSpawnTimes: acc.taskSpawnTimes,
    taskTerminateTimes: acc.taskTerminateTimes, callframeSymbols: acc.callframeSymbols,
    cpuGroups: [...acc.cpuGroupMap.values()].sort((a, b) => b.count - a.count),
    schedGroups: [...acc.schedGroupMap.values()].sort((a, b) => b.count - a.count),
    spanStats: acc.spanStats,
    pollDurationByLoc: acc.pollDurationByLoc,
    schedDelayHist: acc.schedDelayHist,
  };
}

/** Print analysis report from pre-computed results. */
function reportAnalysis(a, label) {
  const { workerIds, minTs, durationMs,
          callframeSymbols, taskSpawnLocs, taskSpawnTimes, taskTerminateTimes } = a;

  console.log(`\n${'='.repeat(60)}`);
  console.log(`TRACE SUMMARY: ${label}`);
  console.log(`${'='.repeat(60)}`);
  console.log(`Duration:       ${durationMs.toFixed(1)}ms`);
  console.log(`Workers:        ${workerIds.length} (IDs: ${workerIds.join(', ')})`);
  console.log(`Events:         ${a.eventCount.toLocaleString()}`);
  console.log(`CPU samples:    ${a.cpuSampleCount.toLocaleString()}`);
  console.log(`Tasks spawned:  ${a.taskSpawnCount}`);
  console.log(`Tasks alive at end: ${a.taskAliveAtEnd}`);

  // ── Worker utilization ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`WORKER UTILIZATION`);
  console.log(`${'─'.repeat(60)}`);
  for (const w of workerIds) {
    const ws = a.workerSpans[w];
    const total = ws.utilization;
    const util = (total * 100).toFixed(1);
    const avgCpu = ws.activeCount > 0 ? ws.avgCpuRatio.toFixed(3) : 'N/A';
    console.log(`  Worker ${w}: ${util}% active, ${ws.pollCount} polls, ${ws.parkCount} parks, avg CPU ratio: ${avgCpu}`);
  }

  // ── Long polls ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`LONG POLLS (>1ms)`);
  console.log(`${'─'.repeat(60)}`);
  const longPolls = a.longPolls;

  if (longPolls.length === 0) {
    console.log('  None found ✅');
  } else {
    console.log(`  Found ${longPolls.length} long poll(s)\n`);
    for (const lp of longPolls.slice(0, 10)) {
      const p = lp.poll;
      console.log(`  ▸ ${fmtDur(lp.dur)} on worker ${lp.worker} at ${fmtRel(p.start, minTs)}`);
      console.log(`    Task ${p.taskId}, spawn: ${p.spawnLoc || '(unknown)'}`);

      if (p.schedSamples?.length) {
        console.log(`    ${p.schedSamples.length} scheduling sample(s) — blocking call stacks:`);
        for (const s of p.schedSamples) {
          const frames = symbolizeChain(s.callchain, callframeSymbols);
          const { syscall, blocker, trigger, classified } = diagnoseStack(frames);
          if (blocker) console.log(`      Blocked by:  ${blocker.symbol} (kernel)`);
          if (syscall) console.log(`      Syscall:     ${syscall.display}`);
          if (trigger) {
            console.log(`      Called from:  ${trigger.display}`);
            if (trigger.location) console.log(`        at ${trigger.location}`);
          }
          const kernelPath = classified.filter(f => f.kind === 'kernel').map(f => f.symbol);
          if (kernelPath.length > 1) console.log(`      Kernel path: ${kernelPath.join(' → ')}`);
          const appFrames = classified.filter(f => f.kind === 'app');
          if (appFrames.length > 0) {
            console.log(`      App call chain (${appFrames.length} frames):`);
            const show = appFrames.length <= 8 ? appFrames : [
              ...appFrames.slice(0, 3),
              { display: `... ${appFrames.length - 6} more frames ...`, kind: 'ellipsis' },
              ...appFrames.slice(-3),
            ];
            for (const f of show) console.log(`        ${f.display}`);
          }
        }
      }
      if (p.cpuSamples?.length) {
        console.log(`    ${p.cpuSamples.length} CPU sample(s) — on-CPU stacks:`);
        for (const s of p.cpuSamples.slice(0, 3)) {
          const frames = symbolizeChain(s.callchain, callframeSymbols);
          const { trigger } = diagnoseStack(frames);
          if (trigger) console.log(`      ${trigger.display}`);
        }
      }
      if (!p.schedSamples?.length && !p.cpuSamples?.length) {
        console.log(`    (no profiling samples captured during this poll)`);
      }
      console.log();
    }
  }

  // ── Scheduling delays ──
  console.log(`${'─'.repeat(60)}`);
  console.log(`SCHEDULING DELAYS (wake → poll latency)`);
  console.log(`${'─'.repeat(60)}`);
  const sds = a.schedDelayStats;
  if (sds.highCount === 0) {
    console.log('  No delays > 1ms ✅');
  } else {
    const sorted = sds.worst.slice().sort((a, b) => b.delay - a.delay);
    console.log(`  ${sds.highCount} delays > 1ms`);
    if (a.schedDelayHist) {
      console.log(`  p50: ${fmtDur(a.schedDelayHist.percentile(50))}, p99: ${fmtDur(a.schedDelayHist.percentile(99))}, max: ${fmtDur(a.schedDelayHist.max)}`);
    }
    console.log(`\n  Top 5 worst:`);
    for (const d of sorted.slice(0, 5)) {
      console.log(`    ${fmtDur(d.delay)} — task ${d.taskId} (spawn: ${d.poll?.spawnLoc || '?'}) at ${fmtRel(d.wakeTime, minTs)}`);
    }
  }

  // ── Poll duration by spawn location ──
  if (a.pollDurationByLoc && a.pollDurationByLoc.size > 0) {
    console.log(`\n${'─'.repeat(60)}`);
    console.log(`POLL DURATION BY SPAWN LOCATION`);
    console.log(`${'─'.repeat(60)}`);
    for (const [loc, h] of [...a.pollDurationByLoc.entries()].sort((x, y) => y[1].max - x[1].max)) {
      console.log(`  ${loc}: count=${h.count} p50=${(h.percentile(50)/1e3).toFixed(1)}µs p99=${(h.percentile(99)/1e3).toFixed(1)}µs max=${(h.max/1e6).toFixed(2)}ms`);
    }
  }

  // ── Task lifecycle ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`TASK LIFECYCLE`);
  console.log(`${'─'.repeat(60)}`);
  const spawnCounts = new Map();
  for (const [, loc] of taskSpawnLocs) {
    spawnCounts.set(loc || '(unknown)', (spawnCounts.get(loc || '(unknown)') || 0) + 1);
  }
  console.log('  Spawns by location:');
  for (const [loc, count] of [...spawnCounts.entries()].sort((a, b) => b[1] - a[1])) {
    console.log(`    ${String(count).padStart(5)} × ${loc}`);
  }

  const atSamples = a.taskTimeline.activeTaskSamples;
  if (atSamples.length > 0) {
    const peak = maxBy(atSamples, s => s.count);
    const final_ = atSamples[atSamples.length - 1].count;
    console.log(`\n  Peak active tasks: ${peak}`);
    console.log(`  Active at trace end: ${final_}`);
    if (final_ > atSamples[0].count * 2 && final_ === peak) {
      console.log('  ⚠ POSSIBLE TASK LEAK — active count grew monotonically');
      const alive = new Map();
      for (const [taskId] of taskSpawnTimes) {
        if (!taskTerminateTimes.has(taskId)) {
          const loc = taskSpawnLocs.get(taskId) || '(unknown)';
          alive.set(loc, (alive.get(loc) || 0) + 1);
        }
      }
      for (const [loc, count] of [...alive.entries()].sort((a, b) => b[1] - a[1])) {
        console.log(`    ${count} leaked from ${loc}`);
      }
    }
  }

  // ── Blocking calls ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`BLOCKING CALLS (scheduling/off-CPU samples)`);
  console.log(`${'─'.repeat(60)}`);
  if (a.offCpuSampleCount === 0) {
    console.log('  No off-CPU samples in trace');
  } else {
    console.log(`  ${a.offCpuSampleCount} off-CPU sample(s)\n`);
    for (const g of a.schedGroups.slice(0, 10)) {
      const pct = (g.count / a.offCpuSampleCount * 100).toFixed(1);
      const { syscall, blocker, trigger } = diagnoseStack(g.frames);
      console.log(`  ${String(g.count).padStart(5)} samples (${pct}%) — leaf: ${g.leaf}`);
      if (blocker) console.log(`         blocked by: ${blocker.symbol} (kernel)`);
      if (syscall) console.log(`         syscall:    ${syscall.display}`);
      if (trigger) console.log(`         called from: ${trigger.display}`);
    }
  }

  // ── CPU hotspots ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`CPU HOTSPOTS (on-CPU samples)`);
  console.log(`${'─'.repeat(60)}`);
  if (a.onCpuSampleCount === 0) {
    console.log('  No CPU samples in trace');
  } else {
    console.log(`  ${a.onCpuSampleCount} CPU sample(s)\n`);
    for (const g of a.cpuGroups.slice(0, 10)) {
      const pct = (g.count / a.onCpuSampleCount * 100).toFixed(1);
      const { trigger } = diagnoseStack(g.frames);
      console.log(`  ${String(g.count).padStart(5)} samples (${pct}%) — leaf: ${g.leaf}`);
      if (trigger) console.log(`         app frame: ${trigger.display}`);
    }
  }

  // ── Queue depth ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`QUEUE DEPTH`);
  console.log(`${'─'.repeat(60)}`);
  const qd = a.queueDepthStats;
  if (qd.samples > 0) {
    console.log(`  Global queue: max=${qd.max}, avg=${qd.avg.toFixed(1)}, samples=${qd.samples}`);
    if (qd.max > 100) console.log('  ⚠ High queue depth — runtime may be overloaded');
  } else {
    console.log('  No queue samples');
  }
  console.log(`  Max local queue: ${a.maxLocalQueue}`);

  // ── Kernel scheduling wait ──
  console.log(`\n${'─'.repeat(60)}`);
  console.log(`KERNEL SCHEDULING WAIT`);
  console.log(`${'─'.repeat(60)}`);
  for (const w of workerIds) {
    const ws = a.workerSpans[w];
    if (ws.schedWaits.length === 0) continue;
    const waits = ws.schedWaits;
    const max = waits[0]; // already sorted descending
    const p50 = waits[Math.floor(waits.length * 0.5)];
    console.log(`  Worker ${w}: max=${(max/1e6).toFixed(1)}ms, p50=${(p50/1e6).toFixed(2)}ms, ${waits.filter(w => w > 1e6).length} events > 1ms`);
  }

  // ── Span durations ──
  if (a.spanStats && a.spanStats.size > 0) {
    console.log(`\n${'─'.repeat(60)}`);
    console.log(`SPAN DURATIONS`);
    console.log(`${'─'.repeat(60)}`);
    for (const [name, h] of [...a.spanStats.entries()].sort((x, y) => y[1].count - x[1].count)) {
      console.log(`  ${name}: count=${h.count} p50=${(h.percentile(50)/1e3).toFixed(1)}µs p99=${(h.percentile(99)/1e3).toFixed(1)}µs max=${(h.max/1e3).toFixed(1)}µs`);
    }
  }

  console.log(`\n${'='.repeat(60)}`);
}

/**
 * Analyze traces from a file or directory. Returns a finalized accumulator
 * with aggregated stats, histograms, and top-N results.
 * For directories, parsing and analysis run in parallel subprocesses.
 * @param {string} tracePath - file or directory path
 * @param {Object} [opts] - { force, sample, onParseProgress }
 */
async function analyzeTraces(tracePath, opts) {
  opts = opts || {};
  const isDir = fs.statSync(tracePath).isDirectory();
  const acc = createAccumulator();

  if (!isDir) {
    for await (const trace of parseTrace(tracePath, opts)) {
      accumulateTrace(acc, trace);
    }
    return finalizeAccumulator(acc);
  }

  // Phase 1: parallel parse (populate cache)
  const parseOpts = { ...opts };
  parseOpts.onParseProgress = opts.onParseProgress || null;
  const iter = parseTrace(tracePath, parseOpts);
  if (iter.allCached) await iter.allCached;
  else { for await (const _ of iter) {} }
  if (opts.onParseComplete) opts.onParseComplete();

  // Phase 2: parallel analysis via accumulate workers
  const { execFile } = require('child_process');
  const os = require('os');
  const cacheDir = path.join(tracePath, '.d9-cache');
  const cacheFiles = iter.files.map(f => f.replace(/\.(bin|bin\.gz)$/, '') + '.json');

  const accWorkerCandidate = path.resolve(__dirname, 'analyze.js');
  const accWorkerFallback = path.resolve(__dirname, '..', 'skills', 'analyze.js');
  const accWorkerScript = fs.existsSync(accWorkerCandidate) ? accWorkerCandidate : accWorkerFallback;

  const concurrency = Math.min(os.cpus().length, 32);
  let analyzed = 0;

  await new Promise((resolveAll, rejectAll) => {
    let nextIdx = 0;
    let active = 0;
    let failed = false;

    function dispatch() {
      while (active < concurrency && nextIdx < cacheFiles.length) {
        const cf = cacheFiles[nextIdx++];
        active++;
        execFile(process.execPath, [accWorkerScript, '--analyze-worker', path.join(cacheDir, cf)], { maxBuffer: 256 * 1024 * 1024 }, (err, stdout, stderr) => {
          if (failed) return;
          if (err) { failed = true; rejectAll(new Error(`Analysis failed for ${cf}: ${stderr || err.message}`)); return; }
          try {
            mergePartial(acc, JSON.parse(stdout));
            analyzed++;
            if (opts.onAnalysisProgress) opts.onAnalysisProgress({ done: analyzed, total: cacheFiles.length, file: cf });
          } catch (e) { failed = true; rejectAll(e); return; }
          active--;
          if (analyzed === cacheFiles.length) resolveAll();
          else dispatch();
        });
      }
    }
    dispatch();
  });

  return finalizeAccumulator(acc);
}

async function main() {
  const TRACE_PATH = process.argv[2];
  if (!TRACE_PATH) { console.error('Usage: node analyze.js <trace.bin or directory>'); process.exit(1); }
  const FORCE = process.argv.includes('--force');
  const sampleIdx = process.argv.indexOf('--sample');
  const SAMPLE = sampleIdx !== -1 ? Number(process.argv[sampleIdx + 1]) : null;

  const isDir = fs.statSync(TRACE_PATH).isDirectory();

  if (isDir) process.stderr.write(`Analyzing directory: ${TRACE_PATH}\n`);
  else process.stderr.write(`Loading ${TRACE_PATH}...\n`);

  const result = await analyzeTraces(TRACE_PATH, {
    force: FORCE,
    sample: SAMPLE || undefined,
    onParseProgress: isDir ? ({ done, total, cached }) => {
      if (done === 0) process.stderr.write(`Found ${total} trace file(s)\n`);
      else if (done === total || done % Math.max(1, Math.floor(total / 100)) === 0) {
        const cachedStr = cached > 0 ? ` (${cached} cached)` : '';
        process.stderr.write(`\r  parsing: [${done}/${total}]${cachedStr}\x1b[K`);
      }
    } : undefined,
    onParseComplete: isDir ? () => {
      process.stderr.write('\n');
    } : undefined,
    onAnalysisProgress: isDir ? ({ done, total }) => {
      if (done === total || done % Math.max(1, Math.floor(total / 100)) === 0) process.stderr.write(`\r  analyzing: [${done}/${total}]\x1b[K`);
    } : undefined,
  });
  if (isDir) process.stderr.write('\n');

  const label = isDir ? `${result.files ? result.files.length : ''} files in ${path.basename(TRACE_PATH)}` : path.basename(TRACE_PATH);
  reportAnalysis(result, label);
}

/** Merge a partial accumulator (from accumulate_worker) into the main accumulator. */
function mergePartial(acc, p) {
  for (const w of p.workerIds) acc.workerIds.add(w);
  if (p.minTs < acc.minTs) acc.minTs = p.minTs;
  if (p.maxTs > acc.maxTs) acc.maxTs = p.maxTs;
  acc.eventCount += p.eventCount;
  acc.cpuSampleCount += p.cpuSampleCount;
  acc.onCpuSampleCount += p.onCpuSampleCount;
  acc.offCpuSampleCount += p.offCpuSampleCount;
  acc.taskSpawnCount += p.taskSpawnCount;
  acc.taskAliveAtEnd += p.taskAliveAtEnd;
  if (p.maxLocalQueue > acc.maxLocalQueue) acc.maxLocalQueue = p.maxLocalQueue;

  // Worker stats
  for (const [w, ws] of Object.entries(p.workerStats)) {
    const dst = acc.workerStats[w] || (acc.workerStats[w] = {
      activeNs: 0, parkNs: 0, ratioSum: 0, activeCount: 0, pollCount: 0, parkCount: 0, schedWaits: [],
    });
    dst.activeNs += ws.activeNs; dst.parkNs += ws.parkNs;
    dst.ratioSum += ws.ratioSum; dst.activeCount += ws.activeCount;
    dst.pollCount += ws.pollCount; dst.parkCount += ws.parkCount;
    for (const sw of ws.schedWaits) dst.schedWaits.push(sw);
  }

  // Long polls
  for (const lp of p.longPolls) acc.longPolls.push(lp);
  if (acc.longPolls.length > 200) { acc.longPolls.sort((a, b) => b.dur - a.dur); acc.longPolls.length = 100; }

  // Queue
  if (p.queueMax > acc.queueMax) acc.queueMax = p.queueMax;
  acc.queueSum += p.queueSum;
  acc.queueCount += p.queueCount;

  // Sched delays
  acc.schedDelayTotal += p.schedDelayTotal;
  acc.schedDelayHighCount += p.schedDelayHighCount;
  for (const sd of p.schedDelayWorst) acc.schedDelayWorst.push(sd);
  if (acc.schedDelayWorst.length > 200) { acc.schedDelayWorst.sort((a, b) => b.delay - a.delay); acc.schedDelayWorst.length = 100; }
  // Feed delay values into histogram
  if (p.schedDelayValues) {
    if (!acc.schedDelayHist) acc.schedDelayHist = createHistogram();
    for (const v of p.schedDelayValues) acc.schedDelayHist.record(v);
  }

  // Task timeline
  for (const s of p.taskTimelineSamples) acc.taskTimelineSamples.push(s);

  // Maps
  if (p.taskSpawnLocs) for (const [k, v] of p.taskSpawnLocs) acc.taskSpawnLocs.set(k, v);
  if (p.taskSpawnTimes) for (const [k, v] of p.taskSpawnTimes) acc.taskSpawnTimes.set(k, v);
  if (p.taskTerminateTimes) for (const [k, v] of p.taskTerminateTimes) acc.taskTerminateTimes.set(k, v);
  if (p.callframeSymbols) for (const [k, v] of p.callframeSymbols) acc.callframeSymbols.set(k, v);

  // Sample groups
  for (const g of (p.cpuGroups || [])) {
    const e = acc.cpuGroupMap.get(g.leaf);
    if (e) e.count += g.count; else acc.cpuGroupMap.set(g.leaf, { ...g });
  }
  for (const g of (p.schedGroups || [])) {
    const e = acc.schedGroupMap.get(g.leaf);
    if (e) e.count += g.count; else acc.schedGroupMap.set(g.leaf, { ...g });
  }

  // Poll durations by location -> histogram
  for (const [loc, durations] of Object.entries(p.pollDurationsByLoc || {})) {
    let h = acc.pollDurationByLoc.get(loc);
    if (!h) { h = createHistogram(); acc.pollDurationByLoc.set(loc, h); }
    for (const d of durations) h.record(d);
  }

  // Span durations -> histogram
  for (const [name, durations] of Object.entries(p.spanDurations || {})) {
    let h = acc.spanStats.get(name);
    if (!h) { h = createHistogram(); acc.spanStats.set(name, h); }
    for (const d of durations) h.record(d);
  }
}

// Run CLI if executed directly, export if required as module
// ── Worker modes (invoked as subprocesses) ──

function mapToEntries(m) { return m instanceof Map ? [...m.entries()] : m; }

async function parseWorkerMain(traceFile, cachePath) {
  const trace = await parseTrace(fs.readFileSync(traceFile));
  const tmpPath = cachePath + '.tmp';
  const stream = fs.createWriteStream(tmpPath);
  function writeLine(obj) { stream.write(JSON.stringify(obj) + '\n'); }

  writeLine({ t: 'm', d: {
    magic: trace.magic, version: trace.version,
    truncated: trace.truncated, timeFiltered: trace.timeFiltered,
    filterStartTime: trace.filterStartTime, filterEndTime: trace.filterEndTime,
    hasCpuTime: trace.hasCpuTime, hasSchedWait: trace.hasSchedWait, hasTaskTracking: trace.hasTaskTracking,
    spawnLocations: mapToEntries(trace.spawnLocations),
    taskSpawnLocs: mapToEntries(trace.taskSpawnLocs),
    taskSpawnTimes: mapToEntries(trace.taskSpawnTimes),
    taskTerminateTimes: mapToEntries(trace.taskTerminateTimes),
    callframeSymbols: mapToEntries(trace.callframeSymbols),
    threadNames: mapToEntries(trace.threadNames),
    runtimeWorkers: mapToEntries(trace.runtimeWorkers),
    clockSyncAnchors: trace.clockSyncAnchors, clockOffsetNs: trace.clockOffsetNs,
  }});
  for (const e of trace.events) writeLine({ t: 'e', d: e });
  for (const s of trace.cpuSamples) writeLine({ t: 'c', d: s });
  if (trace.customEvents) for (const x of trace.customEvents) writeLine({ t: 'x', d: x });

  await new Promise((res, rej) => { stream.end(() => { fs.renameSync(tmpPath, cachePath); res(); }); stream.on('error', rej); });
  process.stdout.write('OK\n');
}

function loadCacheFile(cachePath) {
  const buf = fs.readFileSync(cachePath);
  let pos = 0;
  function nextLine() {
    const nl = buf.indexOf(10, pos);
    if (nl === -1) { if (pos < buf.length) { const s = buf.toString('utf8', pos, buf.length); pos = buf.length; return s; } return null; }
    const s = buf.toString('utf8', pos, nl); pos = nl + 1; return s;
  }
  let raw = null;
  const events = [], cpuSamples = [], customEvents = [];
  let line;
  while ((line = nextLine()) !== null) {
    if (!line) continue;
    const rec = JSON.parse(line);
    switch (rec.t) {
      case 'm': raw = rec.d;
        for (const k of ['spawnLocations','taskSpawnLocs','taskSpawnTimes','taskTerminateTimes','callframeSymbols','threadNames','runtimeWorkers'])
          if (raw[k]) raw[k] = new Map(raw[k]);
        break;
      case 'e': events.push(rec.d); break;
      case 'c': cpuSamples.push(rec.d); break;
      case 'x': customEvents.push(rec.d); break;
    }
  }
  raw.events = events; raw.cpuSamples = cpuSamples; raw.customEvents = customEvents;
  return raw;
}

function analyzeWorkerMain(cachePath) {
  const trace = loadCacheFile(cachePath);
  const wids = [...new Set(trace.events.filter(e => e.eventType !== EVENT_TYPES.QueueSample && e.eventType !== EVENT_TYPES.WakeEvent).map(e => e.workerId))].sort((a, b) => a - b);
  const minTs = trace.events.reduce((m, e) => Math.min(m, e.timestamp), Infinity);
  const maxTs = trace.events.reduce((m, e) => Math.max(m, e.timestamp), -Infinity);
  const spans = buildWorkerSpans(trace.events, wids, maxTs);
  attachCpuSamples(trace.cpuSamples, spans.workerSpans);
  const taskTimeline = buildActiveTaskTimeline(trace.taskSpawnTimes, trace.taskTerminateTimes);
  const schedDelays = computeSchedulingDelays(spans.workerSpans, wids, spans.wakesByTask);
  const onCpu = trace.cpuSamples.filter(s => s.source === 0);
  const offCpu = trace.cpuSamples.filter(s => s.source === 1);

  const partial = {
    workerIds: wids, minTs, maxTs,
    eventCount: trace.events.length, cpuSampleCount: trace.cpuSamples.length,
    onCpuSampleCount: onCpu.length, offCpuSampleCount: offCpu.length,
    taskSpawnCount: trace.taskSpawnTimes.size,
    taskAliveAtEnd: trace.taskSpawnTimes.size - trace.taskTerminateTimes.size,
    maxLocalQueue: spans.maxLocalQueue,
    workerStats: {}, longPolls: [],
    queueMax: 0, queueSum: 0, queueCount: 0,
    schedDelayTotal: schedDelays.length, schedDelayHighCount: 0, schedDelayWorst: [],
    schedDelayValues: schedDelays.map(sd => Math.max(1, Math.round(sd.delay))),
    taskTimelineSamples: taskTimeline.activeTaskSamples,
    taskSpawnLocs: mapToEntries(trace.taskSpawnLocs),
    taskSpawnTimes: mapToEntries(trace.taskSpawnTimes),
    taskTerminateTimes: mapToEntries(trace.taskTerminateTimes),
    callframeSymbols: mapToEntries(trace.callframeSymbols),
    cpuGroups: deduplicateSamples(onCpu, trace.callframeSymbols),
    schedGroups: deduplicateSamples(offCpu, trace.callframeSymbols),
    pollDurationsByLoc: {}, spanDurations: {},
  };

  for (const w of wids) {
    const s = spans.workerSpans[w];
    const ws = { activeNs: 0, parkNs: 0, ratioSum: 0, activeCount: 0, pollCount: s.polls.length, parkCount: s.parks.length, schedWaits: [] };
    for (const a of s.actives) { ws.activeNs += a.end - a.start; ws.ratioSum += a.ratio; ws.activeCount++; }
    for (const p of s.parks) { ws.parkNs += p.end - p.start; if (p.schedWait > 0) ws.schedWaits.push(p.schedWait); }
    partial.workerStats[w] = ws;
    for (const p of s.polls) {
      const dur = p.end - p.start;
      const loc = p.spawnLoc || '(unknown)';
      (partial.pollDurationsByLoc[loc] || (partial.pollDurationsByLoc[loc] = [])).push(Math.max(1, Math.round(dur)));
      if (dur > 1e6) partial.longPolls.push({ dur, poll: p, worker: w });
    }
  }
  partial.longPolls.sort((a, b) => b.dur - a.dur);
  partial.longPolls.length = Math.min(partial.longPolls.length, 100);
  for (const q of spans.queueSamples) { if (q.global > partial.queueMax) partial.queueMax = q.global; partial.queueSum += q.global; partial.queueCount++; }
  for (const sd of schedDelays) { if (sd.delay > 1e6) { partial.schedDelayHighCount++; partial.schedDelayWorst.push(sd); } }
  partial.schedDelayWorst.sort((a, b) => b.delay - a.delay);
  partial.schedDelayWorst.length = Math.min(partial.schedDelayWorst.length, 100);
  if (trace.customEvents && trace.customEvents.length > 0) {
    const { spansByWorker } = buildSpanData(trace.customEvents);
    for (const ss of Object.values(spansByWorker)) for (const s of ss)
      (partial.spanDurations[s.spanName] || (partial.spanDurations[s.spanName] = [])).push(Math.max(1, Math.round(s.end - s.start)));
  }
  process.stdout.write(JSON.stringify(partial) + '\n');
}

// ── Entry point ──

if (require.main === module) {
  if (process.argv[2] === '--parse-worker') {
    parseWorkerMain(process.argv[3], process.argv[4]).catch(err => { process.stderr.write(err.stack + '\n'); process.exit(1); });
  } else if (process.argv[2] === '--analyze-worker') {
    analyzeWorkerMain(process.argv[3]);
  } else {
    main().catch(err => { console.error(err); process.exit(1); });
  }
}

module.exports = { analyzeTraces };