openchamber/packages/ui/src/sync/__tests__/event-pipeline.bench.js

/**
 * Synthetic benchmark for the event pipeline.
 *
 * Measures how much per-directory queueing + delta coalescing shrinks the
 * delivered event stream and how long enqueue/flush takes for realistic
 * multi-session workloads (parent + subagent token streaming).
 *
 * Run with:
 *   bun packages/ui/src/sync/__tests__/event-pipeline.bench.js
 *
 * This is NOT a bun:test file — it prints a report and exits. Nothing here
 * asserts; it exists purely to give you intuition about the optimization
 * impact at varying concurrency levels.
 */

import { createEventPipeline } from '../event-pipeline.ts';

// ---------------------------------------------------------------------------
// Minimal DOM stubs (same approach as the unit tests)
// ---------------------------------------------------------------------------

globalThis.document = {
  visibilityState: 'visible',
  addEventListener() {},
  removeEventListener() {},
};
globalThis.window = {
  addEventListener() {},
  removeEventListener() {},
};

// ---------------------------------------------------------------------------
// SDK mock that replays a pre-generated event list
// ---------------------------------------------------------------------------

function createReplaySdk(events, hold) {
  return {
    global: {
      event: async () => ({
        stream: (async function* () {
          for (const e of events) yield e;
          await hold;
        })(),
      }),
    },
  };
}

// ---------------------------------------------------------------------------
// Workload generators
// ---------------------------------------------------------------------------

/**
 * Token-stream workload: N sessions in `directoryCount` directories, each
 * emitting `tokensPerSession` text deltas plus a few framing events.
 *
 * Shape is intentionally close to a real opencode session:
 *   session.status(busy)
 *   message.part.delta × tokensPerSession   (coalescible)
 *   message.part.updated                    (final state)
 *   session.status(idle)
 */
function buildTokenStreamWorkload({
  directoryCount,
  sessionsPerDirectory,
  tokensPerSession,
}) {
  const events = [];
  for (let d = 0; d < directoryCount; d++) {
    const directory = `dir-${d}`;
    for (let s = 0; s < sessionsPerDirectory; s++) {
      const sessionID = `dir-${d}-s${s}`;
      const messageID = `${sessionID}-m1`;
      const partID = `${messageID}-p1`;

      events.push({
        directory,
        payload: {
          type: 'session.status',
          properties: { sessionID, status: { type: 'busy' } },
        },
      });

      for (let t = 0; t < tokensPerSession; t++) {
        events.push({
          directory,
          payload: {
            type: 'message.part.delta',
            properties: {
              messageID,
              partID,
              field: 'text',
              delta: 'x',
            },
          },
        });
      }

      events.push({
        directory,
        payload: {
          type: 'message.part.updated',
          properties: {
            part: {
              id: partID,
              type: 'text',
              messageID,
              text: 'x'.repeat(tokensPerSession),
            },
          },
        },
      });

      events.push({
        directory,
        payload: {
          type: 'session.status',
          properties: { sessionID, status: { type: 'idle' } },
        },
      });
    }
  }

  // Interleave events across directories/sessions so we exercise the real
  // "parent + subagent" arrival pattern instead of one session at a time.
  return interleave(events);
}

// Shuffle events within each directory bucket so arrivals are interleaved but
// still ordered within a single (sessionID, partID) stream (deltas must stay
// ordered relative to each other for append semantics to remain correct).
function interleave(events) {
  const buckets = new Map(); // directory -> list of events (in original order)
  for (const e of events) {
    const bucket = buckets.get(e.directory) ?? [];
    bucket.push(e);
    buckets.set(e.directory, bucket);
  }
  const out = [];
  let more = true;
  while (more) {
    more = false;
    for (const bucket of buckets.values()) {
      if (bucket.length > 0) {
        out.push(bucket.shift());
        more = true;
      }
    }
  }
  return out;
}

// ---------------------------------------------------------------------------
// Runner — pushes a workload through the pipeline and measures
// ---------------------------------------------------------------------------

async function runScenario(label, workload) {
  let release;
  const hold = new Promise((resolve) => {
    release = resolve;
  });

  let delivered = 0;
  let deliveredDeltas = 0;
  let deliveredDeltaBytes = 0;

  const sdk = createReplaySdk(workload, hold);

  const startWall = performance.now();
  const { cleanup } = createEventPipeline({
    sdk,
    onEvent: (_directory, payload) => {
      delivered++;
      if (payload.type === 'message.part.delta') {
        deliveredDeltas++;
        deliveredDeltaBytes += payload.properties.delta.length;
      }
    },
  });

  // Give the pipeline enough time to finish enqueueing AND flush. Scale wait
  // time with workload so large stress scenarios have room to drain — the SSE
  // loop yields every 8ms and the flush fires every 16ms, so in the worst case
  // we need ~(workload / STREAM_YIELD) * 16ms of wall clock.
  const waitMs = Math.max(200, Math.ceil(workload.length / 100));
  await new Promise((resolve) => setTimeout(resolve, waitMs));
  const endWall = performance.now();

  cleanup();
  release();

  // Count input-side delta events for comparison
  const inputDeltas = workload.filter((e) => e.payload.type === 'message.part.delta').length;
  const inputDeltaBytes = workload
    .filter((e) => e.payload.type === 'message.part.delta')
    .reduce((n, e) => n + e.payload.properties.delta.length, 0);

  const wallMs = endWall - startWall;
  const reductionPct = inputDeltas === 0 ? 0 : (1 - deliveredDeltas / inputDeltas) * 100;

  return {
    label,
    inputEvents: workload.length,
    inputDeltas,
    inputDeltaBytes,
    deliveredEvents: delivered,
    deliveredDeltas,
    deliveredDeltaBytes,
    reductionPct,
    wallMs,
  };
}

function formatRow(r) {
  const cols = [
    r.label.padEnd(44),
    String(r.inputEvents).padStart(8),
    String(r.deliveredEvents).padStart(8),
    String(r.inputDeltas).padStart(8),
    String(r.deliveredDeltas).padStart(8),
    `${r.reductionPct.toFixed(1)}%`.padStart(8),
    `${r.wallMs.toFixed(1)}ms`.padStart(10),
    r.inputDeltaBytes === r.deliveredDeltaBytes ? 'bytes ✓' : `bytes ${r.inputDeltaBytes}→${r.deliveredDeltaBytes}`,
  ];
  return cols.join('  ');
}

function header() {
  const cols = [
    'scenario'.padEnd(44),
    'in'.padStart(8),
    'out'.padStart(8),
    'in Δ'.padStart(8),
    'out Δ'.padStart(8),
    'reduce'.padStart(8),
    'wall'.padStart(10),
    'integrity',
  ];
  return cols.join('  ');
}

// ---------------------------------------------------------------------------
// Scenarios
// ---------------------------------------------------------------------------

const scenarios = [
  {
    label: 'single project, 1 session, 500 tokens',
    workload: buildTokenStreamWorkload({
      directoryCount: 1,
      sessionsPerDirectory: 1,
      tokensPerSession: 500,
    }),
  },
  {
    label: 'single project, parent + 1 subagent, 500 tokens each',
    workload: buildTokenStreamWorkload({
      directoryCount: 1,
      sessionsPerDirectory: 2,
      tokensPerSession: 500,
    }),
  },
  {
    label: 'single project, parent + 3 subagents, 500 tokens each',
    workload: buildTokenStreamWorkload({
      directoryCount: 1,
      sessionsPerDirectory: 4,
      tokensPerSession: 500,
    }),
  },
  {
    label: 'single project, parent + 9 subagents, 200 tokens each',
    workload: buildTokenStreamWorkload({
      directoryCount: 1,
      sessionsPerDirectory: 10,
      tokensPerSession: 200,
    }),
  },
  {
    label: '3 projects, 1 session each, 500 tokens',
    workload: buildTokenStreamWorkload({
      directoryCount: 3,
      sessionsPerDirectory: 1,
      tokensPerSession: 500,
    }),
  },
  {
    label: '3 projects × (parent + subagent), 500 tokens each',
    workload: buildTokenStreamWorkload({
      directoryCount: 3,
      sessionsPerDirectory: 2,
      tokensPerSession: 500,
    }),
  },
  {
    label: '5 projects × parent + 3 subagents, 200 tokens',
    workload: buildTokenStreamWorkload({
      directoryCount: 5,
      sessionsPerDirectory: 4,
      tokensPerSession: 200,
    }),
  },
  {
    label: 'stress: 10 projects × 5 sessions × 1000 tokens',
    workload: buildTokenStreamWorkload({
      directoryCount: 10,
      sessionsPerDirectory: 5,
      tokensPerSession: 1000,
    }),
  },
];

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

async function main() {
  console.log('event-pipeline synthetic benchmark\n');
  console.log(header());
  console.log('-'.repeat(header().length + 8));

  const results = [];
  for (const { label, workload } of scenarios) {
    const r = await runScenario(label, workload);
    results.push(r);
    console.log(formatRow(r));
  }

  console.log('\nLegend:');
  console.log('  in       — total events fed into the pipeline');
  console.log('  out      — total events dispatched via onEvent after coalescing + flush');
  console.log('  in Δ     — input events of type message.part.delta');
  console.log('  out Δ    — delta events that actually made it to onEvent (after merging)');
  console.log('  reduce   — (1 − outΔ / inΔ) × 100, i.e. how much delta traffic shrunk');
  console.log('  wall     — total wall-clock time for the scenario (bounded by 200ms wait)');
  console.log('  integrity — "bytes ✓" means the concatenated delta bytes match the input total');
  console.log('');
  console.log('Interpretation:');
  console.log('  Higher reduce % = fewer reducer invocations, fewer React setState calls,');
  console.log('  fewer allocations inside the flush loop. The integrity check confirms no');
  console.log('  text was dropped during coalescing.');
}

main().catch((error) => {
  console.error('benchmark failed:', error);
  process.exit(1);
});