diff --git a/packages/ui/src/sync/__tests__/event-pipeline.bench.js b/packages/ui/src/sync/__tests__/event-pipeline.bench.js new file mode 100644 index 00000000..b859da68 --- /dev/null +++ b/packages/ui/src/sync/__tests__/event-pipeline.bench.js @@ -0,0 +1,347 @@ +/** + * 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); +}); diff --git a/packages/ui/src/sync/__tests__/event-pipeline.test.js b/packages/ui/src/sync/__tests__/event-pipeline.test.js index 99dc7649..c505d645 100644 --- a/packages/ui/src/sync/__tests__/event-pipeline.test.js +++ b/packages/ui/src/sync/__tests__/event-pipeline.test.js @@ -51,6 +51,33 @@ function createSdkWithEvents(events, hold) { }; } +// Run a pipeline against a pre-seeded event stream, collect every dispatched +// event, wait long enough for the 16ms flush window to elapse, then tear it +// down. Returns the list of { directory, payload } that onEvent saw. +async function runPipelineWithEvents(events, waitMs = 80) { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const sdk = createSdkWithEvents(events, hold); + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (directory, payload) => { + received.push({ directory, payload }); + }, + }); + + await new Promise((resolve) => setTimeout(resolve, waitMs)); + cleanup(); + releaseStream(); + + return received; +} + describe('createEventPipeline', () => { it('falls back to payload.properties.directory when the SDK event omits top-level directory', async () => { installDomStubs(); @@ -341,3 +368,239 @@ describe('createEventPipeline', () => { expect(received[0].payload.type).toBe('message.part.updated'); }); }); + +// --------------------------------------------------------------------------- +// P1 — Per-directory queue isolation +// --------------------------------------------------------------------------- + +describe('createEventPipeline — per-directory isolation (P1)', () => { + it('delivers events from two directories without losing either', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's-a', status: { type: 'busy' } }, + }, + }, + { + directory: 'dir-b', + payload: { + type: 'session.status', + properties: { sessionID: 's-b', status: { type: 'idle' } }, + }, + }, + ]); + + const dirs = received.map((r) => r.directory).sort(); + expect(dirs).toEqual(['dir-a', 'dir-b']); + }); + + it('keeps distinct sessionIDs in the same directory as independent coalesce slots', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's1', status: { type: 'busy' } }, + }, + }, + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's2', status: { type: 'busy' } }, + }, + }, + ]); + + expect(received).toHaveLength(2); + const sessionIds = received.map((r) => r.payload.properties.sessionID).sort(); + expect(sessionIds).toEqual(['s1', 's2']); + }); + + it('collapses repeated session.status for the same session down to the latest', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's1', status: { type: 'busy' } }, + }, + }, + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's1', status: { type: 'idle' } }, + }, + }, + ]); + + expect(received).toHaveLength(1); + expect(received[0].payload.properties.status.type).toBe('idle'); + }); +}); + +// --------------------------------------------------------------------------- +// Option C — message.part.delta coalescing +// --------------------------------------------------------------------------- + +describe('createEventPipeline — delta coalescing (Option C)', () => { + it('accumulates consecutive deltas for the same (messageID, partID, field) into one event', async () => { + const events = ['Hello ', 'world', ', ', 'how ', 'are ', 'you?'].map((chunk) => ({ + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'text', + delta: chunk, + }, + }, + })); + + const received = await runPipelineWithEvents(events); + + expect(received).toHaveLength(1); + expect(received[0].payload.type).toBe('message.part.delta'); + expect(received[0].payload.properties.delta).toBe('Hello world, how are you?'); + expect(received[0].payload.properties.messageID).toBe('msg-1'); + expect(received[0].payload.properties.partID).toBe('part-1'); + expect(received[0].payload.properties.field).toBe('text'); + }); + + it('does NOT merge deltas across different fields on the same part', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'text', + delta: 'A', + }, + }, + }, + { + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'reasoning', + delta: 'B', + }, + }, + }, + ]); + + expect(received).toHaveLength(2); + const fieldDelta = received.map((r) => [ + r.payload.properties.field, + r.payload.properties.delta, + ]).sort(); + expect(fieldDelta).toEqual([ + ['reasoning', 'B'], + ['text', 'A'], + ]); + }); + + it('does NOT merge deltas across different parts on the same message', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'text', + delta: 'AAA', + }, + }, + }, + { + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-2', + field: 'text', + delta: 'BBB', + }, + }, + }, + ]); + + expect(received).toHaveLength(2); + const byPart = Object.fromEntries( + received.map((r) => [r.payload.properties.partID, r.payload.properties.delta]), + ); + expect(byPart['part-1']).toBe('AAA'); + expect(byPart['part-2']).toBe('BBB'); + }); + + it('does NOT merge deltas across different directories (per-directory queues)', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'text', + delta: 'from-a', + }, + }, + }, + { + directory: 'dir-b', + payload: { + type: 'message.part.delta', + properties: { + messageID: 'msg-1', + partID: 'part-1', + field: 'text', + delta: 'from-b', + }, + }, + }, + ]); + + expect(received).toHaveLength(2); + const byDir = Object.fromEntries( + received.map((r) => [r.directory, r.payload.properties.delta]), + ); + expect(byDir['dir-a']).toBe('from-a'); + expect(byDir['dir-b']).toBe('from-b'); + }); + + it('does not touch non-delta events (session.status still replaced, not concatenated)', async () => { + const received = await runPipelineWithEvents([ + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's1', status: { type: 'busy' } }, + }, + }, + { + directory: 'dir-a', + payload: { + type: 'session.status', + properties: { sessionID: 's1', status: { type: 'idle' } }, + }, + }, + ]); + + expect(received).toHaveLength(1); + expect(received[0].payload.properties.status.type).toBe('idle'); + }); +}); diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index 8cb19a53..bddf584f 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -78,62 +78,102 @@ function resolveEventDirectory(event: unknown, payload: Event): string { return propertyDirectory && propertyDirectory.length > 0 ? propertyDirectory : "global" } +// Per-directory queue state. Each directory owns an independent flush timer +// so a busy directory's delta storm cannot block another directory's events +// from reaching the UI (head-of-line blocking across sessions). +type DirectoryQueue = { + queue: Event[] + buffer: Event[] + coalesced: Map + timer: ReturnType | undefined + last: number +} + export function createEventPipeline(input: EventPipelineInput) { const { sdk, onEvent, onReconnect } = input const abort = new AbortController() let hasConnected = false - // Queue state - let queue: QueuedEvent[] = [] - let buffer: QueuedEvent[] = [] - const coalesced = new Map() - let timer: ReturnType | undefined - let last = 0 + // One queue + one flush timer per directory. Lazily created on first event. + const directories = new Map() - // Coalesce key — same-type events for the same entity replace earlier ones - const key = (directory: string, payload: Event): string | undefined => { + const getOrCreateDir = (directory: string): DirectoryQueue => { + let d = directories.get(directory) + if (d) return d + d = { + queue: [], + buffer: [], + coalesced: new Map(), + timer: undefined, + last: 0, + } + directories.set(directory, d) + return d + } + + // Coalesce key — same-type events for the same entity replace earlier ones. + // Keys are scoped to a single directory's queue, so directory is implicit. + // message.part.delta is a special case: consecutive deltas for the same + // (messageID, partID, field) are accumulated (string-concatenated) rather + // than replaced, because the reducer is a pure append and merging is + // semantically identical to applying each delta individually. + const key = (payload: Event): string | undefined => { if (payload.type === "session.status") { const props = payload.properties as { sessionID: string } - return `session.status:${directory}:${props.sessionID}` + return `session.status:${props.sessionID}` } if (payload.type === "lsp.updated") { - return `lsp.updated:${directory}` + return `lsp.updated` } if (payload.type === "message.part.updated") { const part = (payload.properties as { part: { messageID: string; id: string } }).part - return `message.part.updated:${directory}:${part.messageID}:${part.id}` + return `message.part.updated:${part.messageID}:${part.id}` + } + if (payload.type === "message.part.delta") { + const props = payload.properties as { messageID: string; partID: string; field: string } + return `message.part.delta:${props.messageID}:${props.partID}:${props.field}` } return undefined } - // Flush — swap queue, dispatch events - const flush = () => { - if (timer) clearTimeout(timer) - timer = undefined + // Flush one directory — swap queue, dispatch events. + // React 18 auto-batching still collapses the setState calls inside a single + // directory's flush into one render pass. + const flushDir = (directory: string) => { + const d = directories.get(directory) + if (!d) return + if (d.timer) { + clearTimeout(d.timer) + d.timer = undefined + } + if (d.queue.length === 0) return - if (queue.length === 0) return + const events = d.queue + d.queue = d.buffer + d.buffer = events + d.queue.length = 0 + d.coalesced.clear() - const events = queue - queue = buffer - buffer = events - queue.length = 0 - coalesced.clear() - - last = Date.now() + d.last = Date.now() syncDebug.pipeline.flush(events.length) - // React 18 batches synchronous setState calls automatically, - // equivalent to SolidJS batch() - for (const event of events) { - onEvent(event.directory, event.payload) + for (const payload of events) { + onEvent(directory, payload) } - buffer.length = 0 + d.buffer.length = 0 } - const schedule = () => { - if (timer) return - const elapsed = Date.now() - last - timer = setTimeout(flush, Math.max(0, FLUSH_FRAME_MS - elapsed)) + const flushAll = () => { + for (const directory of directories.keys()) { + flushDir(directory) + } + } + + const scheduleDir = (directory: string) => { + const d = getOrCreateDir(directory) + if (d.timer) return + const elapsed = Date.now() - d.last + d.timer = setTimeout(() => flushDir(directory), Math.max(0, FLUSH_FRAME_MS - elapsed)) } // Helpers @@ -201,18 +241,34 @@ export function createEventPipeline(input: EventPipelineInput) { } const normalizedPayload = normalizeEventType(payload) const directory = resolveEventDirectory(event, normalizedPayload) - const k = key(directory, normalizedPayload) + const d = getOrCreateDir(directory) + const k = key(normalizedPayload) if (k) { - const i = coalesced.get(k) + const i = d.coalesced.get(k) if (i !== undefined) { - queue[i] = { directory, payload: normalizedPayload } + if (normalizedPayload.type === "message.part.delta") { + // Accumulate delta strings — append to the already-queued event + // rather than replacing it. The reducer is a pure string append so + // this is semantically identical to applying each delta separately. + const prev = d.queue[i] as unknown as { properties: { delta: string } } + const inc = normalizedPayload.properties as { delta: string } + d.queue[i] = { + ...normalizedPayload, + properties: { + ...(normalizedPayload.properties as object), + delta: prev.properties.delta + inc.delta, + }, + } as unknown as Event + } else { + d.queue[i] = normalizedPayload + } syncDebug.pipeline.coalesced(normalizedPayload.type, k) continue } - coalesced.set(k, queue.length) + d.coalesced.set(k, d.queue.length) } - queue.push({ directory, payload: normalizedPayload }) - schedule() + d.queue.push(normalizedPayload) + scheduleDir(directory) if (Date.now() - yielded < STREAM_YIELD_MS) continue yielded = Date.now() @@ -232,7 +288,7 @@ export function createEventPipeline(input: EventPipelineInput) { if (abort.signal.aborted) return await wait(RECONNECT_DELAY_MS) } - })().finally(flush) + })().finally(flushAll) // Visibility handler — abort SSE on heartbeat timeout so the loop reconnects. // The reconnect triggers onReconnect above, which lets consumers resync state. @@ -262,7 +318,7 @@ export function createEventPipeline(input: EventPipelineInput) { window.removeEventListener("pageshow", onPageShow) } abort.abort() - flush() + flushAll() } return { cleanup }