diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index c54ccbac1..359959c56 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -247,6 +247,16 @@ runtime cost control, and shared AI transport surfaces. Pulse's browser drawer adapts that model by keeping the transcript rows typed and stable while the active-turn footer remains a replacing live status slot for provider and workflow progress between visible parts. + The same OpenCode commit applies each streamed session event as its own + state mutation in `packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx` + (`apply(event)` and the `session.next.tool.*` cases), so Pulse's shared + browser SSE consumer must treat opted-in non-text Assistant progress events + as paint checkpoints. `frontend-modern/src/api/streaming.ts` must not drain + queued workflow/tool/model events from separate stream reads without yielding + to the browser, or the drawer will show those steps only after a batch has + already finished. Token content and hidden reasoning may continue to opt out + of those checkpoints through the caller predicate so answer streaming remains + fast. OpenCode-parity Assistant UX work must reference OpenCode's actual source implementation for message parts, tool-state mutation, progress rendering, and model/session selection before changing Pulse behavior; parity means diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index a192f9f70..606e66715 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -464,12 +464,12 @@ payload shape change when the portal presents compact client rows. visible tool row from pending to running/waiting to completed/error, or cancel a runtime-policy-hidden pending row, without inventing a browser-only event shape. The shared frontend SSE consumer in - `frontend-modern/src/api/streaming.ts` may insert a bounded task break - between already-buffered Assistant progress/tool/status events so coalesced - HTTP chunks preserve visible event order, but that pacing is an API-client - consumption rule only: it must not change event payload shape, event order, - timeout handling, reader cleanup, parse-error handling, or ordinary - content-token throughput. + `frontend-modern/src/api/streaming.ts` may insert a bounded task break after + opted-in Assistant progress/tool/status events, including events that arrive + through separate queued stream reads rather than the same decoded HTTP chunk. + That pacing is an API-client consumption rule only: it must not change event + payload shape, event order, timeout handling, reader cleanup, parse-error + handling, or ordinary content-token throughput. Cold Assistant streams must include a typed `session` event backed by `internal/ai/chat.SessionData` as soon as the HTTP SSE writer is ready and an immediate neutral `workflow_state` preparation event before backend diff --git a/frontend-modern/src/api/__tests__/streaming.test.ts b/frontend-modern/src/api/__tests__/streaming.test.ts index b8de19cdb..db61fd8b9 100644 --- a/frontend-modern/src/api/__tests__/streaming.test.ts +++ b/frontend-modern/src/api/__tests__/streaming.test.ts @@ -14,6 +14,23 @@ const makeEventStreamResponse = (body: string) => }, }) as unknown as Response; +const makeChunkedEventStreamResponse = (chunks: string[]) => { + const read = vi.fn(); + for (const chunk of chunks) { + read.mockResolvedValueOnce({ done: false, value: new TextEncoder().encode(chunk) }); + } + read.mockResolvedValueOnce({ done: true, value: undefined }); + + return { + body: { + getReader: () => ({ + read, + releaseLock: vi.fn(), + }), + }, + } as unknown as Response; +}; + const flushMicrotasks = async () => { for (let attempt = 0; attempt < 20; attempt += 1) { await Promise.resolve(); @@ -100,4 +117,38 @@ describe('consumeJSONEventStream', () => { expect(events).toEqual(['content', 'tool_progress', 'done']); expect(vi.getTimerCount()).toBe(0); }); + + it('yields to the browser between matching events that arrive in separate queued reads', async () => { + vi.useFakeTimers(); + + const events: string[] = []; + const streamPromise = consumeJSONEventStream<{ type: string }>( + makeChunkedEventStreamResponse([ + 'data: {"type":"tool_start"}\n\n', + 'data: {"type":"tool_progress"}\n\n', + 'data: {"type":"done"}\n\n', + ]), + { + onEvent: (event) => { + events.push(event.type); + return event.type === 'done'; + }, + yieldBetweenEvents: (event) => event.type !== 'content', + }, + ); + + await flushMicrotasks(); + expect(events).toEqual(['tool_start']); + expect(vi.getTimerCount()).toBeGreaterThan(0); + + await vi.advanceTimersByTimeAsync(1); + await flushMicrotasks(); + expect(events).toEqual(['tool_start', 'tool_progress']); + expect(vi.getTimerCount()).toBeGreaterThan(0); + + await vi.advanceTimersByTimeAsync(1); + await streamPromise; + expect(events).toEqual(['tool_start', 'tool_progress', 'done']); + expect(vi.getTimerCount()).toBe(0); + }); }); diff --git a/frontend-modern/src/api/streaming.ts b/frontend-modern/src/api/streaming.ts index 575aea60f..53a3e8a54 100644 --- a/frontend-modern/src/api/streaming.ts +++ b/frontend-modern/src/api/streaming.ts @@ -18,9 +18,8 @@ const yieldToBrowser = () => const shouldYieldAfterEvent = ( event: T, options: JSONEventStreamOptions, - hasBufferedEventAfterThisOne: boolean, ) => { - if (!hasBufferedEventAfterThisOne || !options.yieldBetweenEvents) return false; + if (!options.yieldBetweenEvents) return false; if (typeof options.yieldBetweenEvents === 'function') { return options.yieldBetweenEvents(event); } @@ -62,11 +61,6 @@ export async function consumeJSONEventStream( const messages = normalizedBuffer.split('\n\n'); buffer = messages.pop() || ''; - const messageHasData = (message: string) => - !!message - .split('\n') - .find((line) => line.startsWith('data: ') && !!line.slice(6).trim()); - for (let messageIndex = 0; messageIndex < messages.length; messageIndex += 1) { const message = messages[messageIndex]; if (!message.trim() || message.trim().startsWith(':')) { @@ -90,10 +84,7 @@ export async function consumeJSONEventStream( if (options.onEvent(event)) { return true; } - const hasBufferedEventAfterThisOne = - dataLines.slice(lineIndex + 1).some((nextLine) => !!nextLine.slice(6).trim()) || - messages.slice(messageIndex + 1).some(messageHasData); - if (shouldYieldAfterEvent(event, options, hasBufferedEventAfterThisOne)) { + if (shouldYieldAfterEvent(event, options)) { await yieldToBrowser(); } }