From f79f913d04a39dc8dd5450057c5029465c74f61b Mon Sep 17 00:00:00 2001 From: Duy Nguyen Date: Mon, 12 Oct 2026 08:57:10 +0700 Subject: [PATCH 1/3] fix: cap SSE line and event buffer The stream parser held the unterminated line and the event being built with no limit, so a server streaming an endless line grew panel memory until the tab died. The parser now holds at most 1 Mi UTF-16 code units for the partial line plus the current event's fields. Past that it drops what it holds, cancels the body and throws SseProtocolError. The client reports it as DeweeApiError protocol_error with afterResponse, so the panel reconciles the turn like other faults after the response and never resends it. Closes #2 Co-Authored-By: Claude Opus 5.5 --- docs/architecture.md | 2 +- src/api/dewee-client.ts | 10 +- src/api/errors.ts | 4 +- src/chat/use-chat.ts | 2 +- src/protocol/sse-reader.ts | 85 ++++++++++++++--- src/shared/i18n.ts | 2 + tests/integration/chat/use-chat.test.tsx | 17 ++++ tests/unit/api/dewee-client.test.ts | 27 ++++++ tests/unit/protocol/sse-reader.test.ts | 112 ++++++++++++++++++++++- 9 files changed, 241 insertions(+), 20 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index b48cd0e..29de0dc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,7 +37,7 @@ The extension is a thin Chrome MV3 client. Dewee (Gateway, or the CCP app for De **Cloud connect.** The panel stores the profile, asks the background to run the OAuth flow, then calls `GET /me`. A failed first sign-in removes the new profile. -**Chat.** `use-chat` posts `chat/stream`. `chat.started` gives the session key, which is remembered per workspace and agent. The panel never re-sends a message by itself, and it offers "Send again" only when the server provably did not receive the message: (a) the panel failed before calling fetch (background or credentials unavailable), (b) the fetch was rejected while `navigator.onLine` was false, or (c) the server refused the turn with an explicit 4xx before accepting it (validation, auth, rate limit, or a busy 409 such as `run_in_progress`). Every other failure is uncertain: a 5xx (including a proxy's 502, 503 or 504), a stall, a drop after the response, a fetch rejected while online, a decoder fault, a history read that failed, or a turn not found after a reconcile. An uncertain turn stays visible once, marked `delivery_unconfirmed`, with "Reload conversation" (or "Show conversations" in a new chat, which has no session to check). It never gets "Send again"; the text stays on screen, so the person can copy it and send it again by hand. After an uncertain failure in a known session, or a break after the server answered (200, any frame or `chat.started`), the engine polls run status and reloads history; polling pauses while the browser is offline, slows down after about 3 minutes and keeps the composer closed until the run ends. Only one reconcile runs per conversation, and a switch, unmount or newer reconcile retires the old one. An unconfirmed turn is dropped only when the history provably holds it: the message that was last at send time (its `m` id, role and text) is still in place, and a user turn with the same text follows it (past the turns confirmed by `chat.started` since). After compaction, renumbered ids, or a highest position below the send-time one, the anchor is gone and the turn stays unconfirmed. A history that cannot be read (including one that keeps shrinking while it is read) is never treated as empty: the transcript stays and the turn stays unconfirmed. Unconfirmed turns survive later history reloads and new sends, and every reconcile or reload checks them again. Residual risks: the offline check is a heuristic, so if the connection drops after the request bytes left, `onLine` can already be false and "Send again" may start a second run; and an identical text sent again after the anchor can make a lost repeat look delivered. Stop calls `chat/abort` and aborts the fetch; pressed before `chat.started` names the run, the abort is sent as soon as it does (or the fetch is dropped after 5 s). +**Chat.** `use-chat` posts `chat/stream`. `chat.started` gives the session key, which is remembered per workspace and agent. The panel never re-sends a message by itself, and it offers "Send again" only when the server provably did not receive the message: (a) the panel failed before calling fetch (background or credentials unavailable), (b) the fetch was rejected while `navigator.onLine` was false, or (c) the server refused the turn with an explicit 4xx before accepting it (validation, auth, rate limit, or a busy 409 such as `run_in_progress`). Every other failure is uncertain: a 5xx (including a proxy's 502, 503 or 504), a stall, a drop after the response, a fetch rejected while online, a decoder fault (including `protocol_error`: the stream parser holds at most 1 Mi characters for the current line and event, and stops the stream past that), a history read that failed, or a turn not found after a reconcile. An uncertain turn stays visible once, marked `delivery_unconfirmed`, with "Reload conversation" (or "Show conversations" in a new chat, which has no session to check). It never gets "Send again"; the text stays on screen, so the person can copy it and send it again by hand. After an uncertain failure in a known session, or a break after the server answered (200, any frame or `chat.started`), the engine polls run status and reloads history; polling pauses while the browser is offline, slows down after about 3 minutes and keeps the composer closed until the run ends. Only one reconcile runs per conversation, and a switch, unmount or newer reconcile retires the old one. An unconfirmed turn is dropped only when the history provably holds it: the message that was last at send time (its `m` id, role and text) is still in place, and a user turn with the same text follows it (past the turns confirmed by `chat.started` since). After compaction, renumbered ids, or a highest position below the send-time one, the anchor is gone and the turn stays unconfirmed. A history that cannot be read (including one that keeps shrinking while it is read) is never treated as empty: the transcript stays and the turn stays unconfirmed. Unconfirmed turns survive later history reloads and new sends, and every reconcile or reload checks them again. Residual risks: the offline check is a heuristic, so if the connection drops after the request bytes left, `onLine` can already be false and "Send again" may start a second run; and an identical text sent again after the anchor can make a lost repeat look delivered. Stop calls `chat/abort` and aborts the fetch; pressed before `chat.started` names the run, the abort is sent as soon as it does (or the fetch is dropped after 5 s). **Cards.** `ui_action` frames and `GET /actions?status=pending` feed the card list. A response goes to `POST /actions/{id}/respond`. If the response is lost (network error or 409) the panel reloads the card. A resolved card's `continuation` is queued for its session, tagged with the profile and workspace captured when the respond call started (a switch during the call does not re-tag it). Queued continuations are sent one at a time once their own session is shown and idle, so two cards answered during a reply both continue after it, and a continuation never streams into another conversation. The queue lives in `storage.session` with one key per item (profile, workspace and setup action; the value holds only those IDs plus session and agent, a `held` flag and the time of the first in-progress answer, never a secret). Adding or removing an item writes only that key, so panels in two windows never remove each other's items, and each panel follows `storage.onChanged` to see what another panel added or delivered. A session or agent switch or a closed panel does not lose an item; reopening the session sends what is pending. Both behave the same while a continuation streams: the stream is dropped without holding the item (this is not a Stop), so it is sent again when its session is shown, here or in a reopened panel, and the Gateway answers `continuation_sent` if it was delivered meanwhile. Unlike user messages, continuations are sent automatically, because the Gateway delivers each at most once. An item leaves the queue only when the agent provably got it: a `message.completed` frame, a 409 `continuation_sent` (already delivered; a silent success that reloads the conversation, with no error bubble), or a `chat.error` with code `setup_result_delivered` (the result reached the agent's session but the run failed; its message is shown and the person continues in the chat). Earlier frames such as `chat.started` are not proof: if the run fails before the result is saved (provider error, rate limit, exhausted credit), the Gateway releases the send lease. It also leaves when it can never succeed (400, 404 or 410, or a 403 other than a key problem: the card or continuation is invalid, gone or not allowed); the panel then says once, with no "Continue", that the card's result could not be delivered and the person should ask the agent again in chat (`continuation_refused`), and logs a `console.warn` with the status and code only (a 400 is a client bug). A 403 `scope_required` or `owner_binding_required` can be fixed by the person, so that item is held with "Continue" and the key error. A 409 `continuation_in_progress` means another request holds a live send lease that may still fail and release, so it is not delivery: the item stays and is sent again after the server's `Retry-After` hint (5 s without one), then 15 s, 30 s and every 60 s, only while its session is shown and idle, for about 11 minutes after the first such answer (a lease goes stale after 10 minutes; the start is stored with the item, so reopening a panel does not restart the window); then it shows "Continue". Any other outcome keeps the item, held, and shows "Continue": any other `chat.error`, `chat.stopped` or Stop, a stream that closes without a terminal frame or drops after the response (the reconcile offers "Continue" once the run ends), and any other refusal, including an unknown 409 code. Resending is safe: the Gateway answers `continuation_sent` if it was delivered and `continuation_in_progress` while a run holds the lease. A held item is stored with a `held` flag, so it is not sent automatically again, neither after a conversation switch nor in a reopened panel; when its session is shown, it offers "Continue". The stored flag wins in every panel: a held copy stored by another panel stops this panel's automatic retries, and an unheld copy (another panel pressed "Continue") releases the item here and removes this panel's "Continue". When another panel delivers the item, its "Continue" here goes away too, and a click on a stale one does nothing. Remaining limit: a held item (including one past the in-progress retry window) stays in `storage.session` until the person presses "Continue" or the browser restarts. diff --git a/src/api/dewee-client.ts b/src/api/dewee-client.ts index 56878b6..6eeeb69 100644 --- a/src/api/dewee-client.ts +++ b/src/api/dewee-client.ts @@ -5,7 +5,7 @@ */ import { z } from 'zod'; import { mapRuntimeFrame, type ChatEvent, type RawSseFrame } from '@/protocol/chat-events'; -import { readSseFrames } from '@/protocol/sse-reader'; +import { readSseFrames, SseProtocolError } from '@/protocol/sse-reader'; import { DeweeApiError, errorFromResponse } from './errors'; import { abortResponse, @@ -160,9 +160,13 @@ export class DeweeClient { } } catch (error) { if (stalled) throw new DeweeApiError(0, 'stream_stalled', 'The stream stopped responding', undefined, responded); - // Any other fault after the 200 (a decoder or mapping bug) still means the server took the - // turn: mark it so callers reconcile instead of offering a resend. The original is the cause. + // Any other fault after the 200 (an oversized line or event, a decoder or mapping bug) still + // means the server took the turn: mark it so callers reconcile instead of offering a resend. + // The original is the cause. if (responded && !signal?.aborted && !(error instanceof DeweeApiError)) { + if (error instanceof SseProtocolError) { + throw new DeweeApiError(0, 'protocol_error', 'The reply broke the stream format', undefined, true, error); + } throw new DeweeApiError(0, 'runtime_request_failed', 'The reply could not be read', undefined, true, error); } throw error; diff --git a/src/api/errors.ts b/src/api/errors.ts index 4f04d91..411d49e 100644 --- a/src/api/errors.ts +++ b/src/api/errors.ts @@ -1,8 +1,8 @@ /** * One error type for every transport failure. `code` is the server's error code (see * docs/api-contract.md) or a client code: `network_error` (rejected fetch), `timeout`, - * `invalid_response`, `stream_stalled`, `connection_lost`, `runtime_request_failed` (a client - * fault after the stream answered), `history_changed` (the history kept shrinking while it was + * `invalid_response`, `stream_stalled`, `connection_lost`, `protocol_error` (a stream line or + * event over the size cap), `runtime_request_failed` (a client fault after the stream answered), `history_changed` (the history kept shrinking while it was * read), `not_connected`, or a background code such as * `background_unavailable`. Messages never contain request bodies or credentials. */ diff --git a/src/chat/use-chat.ts b/src/chat/use-chat.ts index ae38344..be2f5d0 100644 --- a/src/chat/use-chat.ts +++ b/src/chat/use-chat.ts @@ -94,7 +94,7 @@ const FIXABLE_REFUSAL_CODES = new Set(['scope_required', 'owner_binding_required const isOffline = () => typeof navigator !== 'undefined' && navigator.onLine === false; /** Status-0 codes of a request that may have reached the server. Other status-0 codes come from the panel. */ -const TRANSPORT_CODES = new Set(['network_error', 'timeout', 'stream_stalled', 'connection_lost', 'invalid_response', 'runtime_request_failed']); +const TRANSPORT_CODES = new Set(['network_error', 'timeout', 'stream_stalled', 'connection_lost', 'protocol_error', 'invalid_response', 'runtime_request_failed']); /** * True only when the server provably did not receive the request: the panel failed before the diff --git a/src/protocol/sse-reader.ts b/src/protocol/sse-reader.ts index db0483a..c706f8e 100644 --- a/src/protocol/sse-reader.ts +++ b/src/protocol/sse-reader.ts @@ -1,35 +1,87 @@ import type { RawSseFrame } from './chat-events'; +/** + * Upper bound, in UTF-16 code units (JavaScript string length), for what the parser holds at once: + * the unterminated line plus the fields of the event being built. 1 Mi code units is at most + * 2 MiB of string memory and at least 1 MiB of UTF-8 on the wire, far above any legitimate frame + * (respond bodies are capped at 160 KiB). Counting code units needs no re-encoding. + */ +export const MAX_SSE_PENDING = 1024 * 1024; + +/** The stream broke the event-stream contract (a line or event over the cap). It is never retried. */ +export class SseProtocolError extends Error { + constructor(message: string) { + super(message); + this.name = 'SseProtocolError'; + } +} + +const NEWLINE = /\r\n|\r|\n/g; + /** * Incremental Server-Sent Events parser (WHATWG event-stream format). Feed decoded text chunks; - * complete frames are returned as soon as their blank-line terminator arrives. + * complete frames are returned as soon as their blank-line terminator arrives. Once more than + * `MAX_SSE_PENDING` code units would be held, `push` drops what it holds and throws + * `SseProtocolError`, and every later `push` throws too. */ export class SseParser { private buffer = ''; private event = ''; private data: string[] = []; + private dataSize = 0; private id: string | undefined; + private failed = false; + + constructor(private readonly maxPending = MAX_SSE_PENDING) {} + + /** Code units currently held between pushes: the partial line and the event being built. */ + get pending(): number { + return this.buffer.length + this.held(); + } push(chunk: string): RawSseFrame[] { - this.buffer += chunk; + if (this.failed) throw new SseProtocolError('The stream already failed'); const frames: RawSseFrame[] = []; - let newline: number; - while ((newline = this.buffer.search(/\r\n|\r|\n/)) >= 0) { - const line = this.buffer.slice(0, newline); - const width = this.buffer.startsWith('\r\n', newline) ? 2 : 1; - this.buffer = this.buffer.slice(newline + width); + let start = 0; + NEWLINE.lastIndex = 0; + for (let match = NEWLINE.exec(chunk); match; match = NEWLINE.exec(chunk)) { + const line = this.buffer + chunk.slice(start, match.index); + this.buffer = ''; + start = match.index + match[0].length; + this.limit(line.length); const frame = this.line(line); if (frame) frames.push(frame); + this.limit(0); } + const rest = chunk.length - start; + this.limit(this.buffer.length + rest); + this.buffer += chunk.slice(start); return frames; } + private held(): number { + return this.event.length + this.dataSize + (this.id?.length ?? 0); + } + + /** Throws, dropping everything held, when `extra` more code units would pass the cap. */ + private limit(extra: number): void { + if (extra + this.held() <= this.maxPending) return; + this.failed = true; + this.buffer = ''; + this.event = ''; + this.data = []; + this.dataSize = 0; + this.id = undefined; + throw new SseProtocolError(`An event-stream line or event exceeded ${this.maxPending} characters`); + } + private line(line: string): RawSseFrame | null { if (line === '') { if (this.data.length === 0 && !this.event) return null; const frame: RawSseFrame = { event: this.event || 'message', data: this.data.join('\n'), ...(this.id ? { id: this.id } : {}) }; this.event = ''; this.data = []; + this.dataSize = 0; return frame; } if (line.startsWith(':')) return null; @@ -38,17 +90,23 @@ export class SseParser { let value = colon < 0 ? '' : line.slice(colon + 1); if (value.startsWith(' ')) value = value.slice(1); if (field === 'event') this.event = value; - else if (field === 'data') this.data.push(value); - else if (field === 'id') this.id = value; + else if (field === 'data') { + this.data.push(value); + // The joining newline counts too, so `dataSize` matches the joined `data` length plus one. + this.dataSize += value.length + 1; + } else if (field === 'id') this.id = value; return null; } } -/** Reads a fetch Response body as SSE frames until the stream ends or the signal aborts. */ -export async function* readSseFrames(body: ReadableStream): AsyncGenerator { +/** + * Reads a fetch Response body as SSE frames until the stream ends or the signal aborts. Throws + * `SseProtocolError` when a line or event passes the cap; the body is cancelled then. + */ +export async function* readSseFrames(body: ReadableStream, maxPending = MAX_SSE_PENDING): AsyncGenerator { const reader = body.getReader(); const decoder = new TextDecoder(); - const parser = new SseParser(); + const parser = new SseParser(maxPending); try { for (;;) { const { value, done } = await reader.read(); @@ -56,6 +114,9 @@ export async function* readSseFrames(body: ReadableStream): AsyncGen yield* parser.push(decoder.decode(value, { stream: true })); } yield* parser.push(decoder.decode() + '\n\n'); + } catch (error) { + if (error instanceof SseProtocolError) await reader.cancel(error).catch(() => {}); + throw error; } finally { reader.releaseLock(); } diff --git a/src/shared/i18n.ts b/src/shared/i18n.ts index a766484..5e2766a 100644 --- a/src/shared/i18n.ts +++ b/src/shared/i18n.ts @@ -105,6 +105,7 @@ const en = { 'error.timeout': 'The server did not respond in time.', 'error.stream_stalled': 'The reply stopped arriving.', 'error.connection_lost': 'The connection dropped during the reply.', + 'error.protocol_error': 'The reply could not be read because a part of it was too large.', 'error.delivery_unconfirmed': 'Delivery could not be confirmed. The message may have reached Dewee and may still be answered, so check the conversation before sending it again.', 'error.runtime_unavailable': 'Dewee is temporarily unavailable.', 'error.run_in_progress': 'This conversation is still running.', @@ -293,6 +294,7 @@ const vi: Record = { 'error.timeout': 'Máy chủ không phản hồi kịp.', 'error.stream_stalled': 'Câu trả lời đã ngừng truyền về.', 'error.connection_lost': 'Kết nối bị ngắt trong lúc nhận câu trả lời.', + 'error.protocol_error': 'Không đọc được câu trả lời vì một phần quá lớn.', 'error.delivery_unconfirmed': 'Không thể xác nhận tin nhắn đã được gửi. Tin nhắn có thể đã đến Dewee và vẫn sẽ có câu trả lời, nên hãy kiểm tra cuộc trò chuyện trước khi gửi lại.', 'error.runtime_unavailable': 'Dewee tạm thời không khả dụng.', 'error.run_in_progress': 'Cuộc trò chuyện này vẫn đang chạy.', diff --git a/tests/integration/chat/use-chat.test.tsx b/tests/integration/chat/use-chat.test.tsx index e3d160e..2093b0e 100644 --- a/tests/integration/chat/use-chat.test.tsx +++ b/tests/integration/chat/use-chat.test.tsx @@ -6,6 +6,7 @@ import type { ChatMessage } from '../../../src/chat/chat-state'; import { addPendingContinuation, loadPendingContinuations, removePendingContinuation } from '../../../src/chat/pending-continuations'; import { useChat, type ChatContext } from '../../../src/chat/use-chat'; import type { ChatEvent } from '../../../src/protocol/chat-events'; +import { SseProtocolError } from '../../../src/protocol/sse-reader'; import { parseUiAction } from '../../../src/protocol/ui-action'; import { readFileSync } from 'node:fs'; import { join } from 'node:path'; @@ -990,6 +991,22 @@ describe('useChat delivery proof', () => { expect(logged).toHaveBeenCalledWith('Chat stream failed', bug); }); + it('treats an oversized stream (protocol_error) as uncertain: no "Send again" and no automatic resend', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + // eslint-disable-next-line require-yield -- the body fails before the first frame + const { client, methods } = fakeClient(async function* () { + throw new DeweeApiError(0, 'protocol_error', 'The reply broke the stream format', undefined, true, new SseProtocolError('too large')); + }); + const { result } = renderHook(() => useChat(context(client), hooks())); + + void result.current.send('Hello'); + + await waitFor(() => expect(lastMessage(result.current.state.messages)?.error).toBeDefined()); + expect(lastMessage(result.current.state.messages)).toMatchObject({ error: { code: 'delivery_unconfirmed' }, recover: 'sessions' }); + expect(result.current.state.messages.some((m) => m.retryText)).toBe(false); + expect(methods.streamChat).toHaveBeenCalledTimes(1); + }); + it('keeps a manual retry for a continuation lost before delivery in a known session, and resends it once', async () => { let call = 0; const { result, methods } = await openSession(async function* () { diff --git a/tests/unit/api/dewee-client.test.ts b/tests/unit/api/dewee-client.test.ts index e9c5f70..3ec77ff 100644 --- a/tests/unit/api/dewee-client.test.ts +++ b/tests/unit/api/dewee-client.test.ts @@ -3,6 +3,7 @@ import { join } from 'node:path'; import { DeweeClient, type AuthAdapter } from '../../../src/api/dewee-client'; import { DeweeApiError } from '../../../src/api/errors'; import type { ChatEvent } from '../../../src/protocol/chat-events'; +import { SseProtocolError } from '../../../src/protocol/sse-reader'; const BASE = 'https://gw.example.test/v1/extension'; const card = JSON.parse( @@ -356,6 +357,32 @@ describe('streamChat', () => { expect(error).toMatchObject({ status: 0, code: 'runtime_request_failed', afterResponse: true, cause: bug }); }); + it('stops an endless stream line as protocol_error after the response, so the turn is reconciled', async () => { + const encoder = new TextEncoder(); + const filler = encoder.encode('x'.repeat(64 * 1024)); + let first = true; + const fetch = vi.fn(async () => { + const body = new ReadableStream({ + pull(controller) { + if (first) { + first = false; + controller.enqueue(encoder.encode(sse('chat.started', { run_id: 'r1', session_key: 's1', agent_key: 'a' }) + 'data: ')); + } else controller.enqueue(filler); + }, + }); + return new Response(body, { status: 200 }); + }); + const received: ChatEvent[] = []; + const run = async () => { + for await (const event of client(fetch, { streamIdleMs: 1000 }).streamChat('ws_1', { agent_key: 'a', message: 'Hi' })) received.push(event); + }; + const error = await run().catch((e: unknown) => e); + expect(error).toBeInstanceOf(DeweeApiError); + expect(error).toMatchObject({ status: 0, code: 'protocol_error', afterResponse: true }); + expect((error as DeweeApiError).cause).toBeInstanceOf(SseProtocolError); + expect(received.map((e) => e.type)).toEqual(['chat.started']); + }); + it('surfaces an auth adapter bug unchanged', async () => { const bug = new Error('adapter bug'); const fetch = vi.fn(); diff --git a/tests/unit/protocol/sse-reader.test.ts b/tests/unit/protocol/sse-reader.test.ts index 693792e..2123705 100644 --- a/tests/unit/protocol/sse-reader.test.ts +++ b/tests/unit/protocol/sse-reader.test.ts @@ -1,4 +1,5 @@ -import { readSseFrames, SseParser } from '../../../src/protocol/sse-reader'; +import type { RawSseFrame } from '../../../src/protocol/chat-events'; +import { MAX_SSE_PENDING, readSseFrames, SseParser, SseProtocolError } from '../../../src/protocol/sse-reader'; describe('SseParser', () => { it('parses frames split across chunks with mixed line endings', () => { @@ -13,6 +14,91 @@ describe('SseParser', () => { it('joins multi-line data', () => { expect(new SseParser().push('data: a\ndata: b\n\n')).toEqual([{ event: 'message', data: 'a\nb' }]); }); + + it('caps the default at 1 Mi characters', () => { + expect(MAX_SSE_PENDING).toBe(1024 * 1024); + }); + + it('stops an endless line fed in chunks at the cap, holding no more than the cap', () => { + const parser = new SseParser(); + const chunk = 'x'.repeat(64 * 1024); + let pushed = 0; + let failure: unknown; + for (let i = 0; i < 64; i++) { + try { + parser.push(i === 0 ? `data: ${chunk}` : chunk); + } catch (error) { + failure = error; + break; + } + pushed += i === 0 ? chunk.length + 6 : chunk.length; + expect(parser.pending).toBe(pushed); + expect(parser.pending).toBeLessThanOrEqual(MAX_SSE_PENDING); + } + expect(failure).toBeInstanceOf(SseProtocolError); + // The push that would pass the cap fails; nothing before it was refused. + expect(pushed + chunk.length).toBeGreaterThan(MAX_SSE_PENDING); + expect(pushed).toBeLessThanOrEqual(MAX_SSE_PENDING); + // The oversized data is dropped, and the parser stays failed. + expect(parser.pending).toBe(0); + expect(() => parser.push('\n\n')).toThrow(SseProtocolError); + }); + + it('stops one chunk that alone holds an oversized line', () => { + const parser = new SseParser(100); + expect(() => parser.push(`data: ${'x'.repeat(200)}`)).toThrow(SseProtocolError); + expect(parser.pending).toBe(0); + }); + + it('stops an event whose data lines together pass the cap, though each line is small', () => { + const parser = new SseParser(1000); + const line = `data: ${'y'.repeat(90)}\n`; + let lines = 0; + expect(() => { + for (; lines < 20; lines++) { + parser.push(line); + expect(parser.pending).toBeLessThanOrEqual(1000); + } + }).toThrow(SseProtocolError); + // Each line holds 91 (value plus joining newline): the 11th line (96) on top of 910 passes 1000. + expect(lines).toBe(10); + expect(parser.pending).toBe(0); + }); + + it('stops an oversized multi-line event at the default cap', () => { + const parser = new SseParser(); + const line = `data: ${'z'.repeat(32 * 1024)}\n`; + expect(() => { + for (let i = 0; i < 64; i++) parser.push(line); + }).toThrow(SseProtocolError); + expect(parser.pending).toBe(0); + }); + + it('parses many small events whose total passes the cap, since each is released when it ends', () => { + const parser = new SseParser(1000); + const frame = `event: message.delta\ndata: ${'d'.repeat(500)}\n\n`; + for (let i = 0; i < 10; i++) expect(parser.push(frame)).toEqual([{ event: 'message.delta', data: 'd'.repeat(500) }]); + }); + + it('parses an event just below the cap split across chunks', () => { + const max = 1000; + const parser = new SseParser(max); + // While the data line is read, the held event name (13) plus the whole line ("data: " + value) + // must fit; afterwards the name plus the value and its joining newline. + const value = 'v'.repeat(max - 13 - 6); + const text = `event: message.delta\r\ndata: ${value}\r\n\r\n`; + const frames = []; + for (let i = 0; i < text.length; i += 97) frames.push(...parser.push(text.slice(i, i + 97))); + expect(frames).toEqual([{ event: 'message.delta', data: value }]); + expect(parser.pending).toBe(0); + }); + + it('refuses the same event one character over the cap', () => { + const max = 1000; + const parser = new SseParser(max); + const value = 'v'.repeat(max - 13 - 5); + expect(() => parser.push(`event: message.delta\ndata: ${value}\n\n`)).toThrow(SseProtocolError); + }); }); describe('readSseFrames', () => { @@ -29,4 +115,28 @@ describe('readSseFrames', () => { for await (const f of readSseFrames(body)) frames.push(f); expect(frames.map((f) => f.event)).toEqual(['chat.started', 'chat.stopped']); }); + + it('stops an endless line and cancels the body', async () => { + const chunk = new TextEncoder().encode('x'.repeat(64 * 1024)); + let pulls = 0; + let cancelled: unknown; + const body = new ReadableStream({ + pull(c) { + pulls++; + c.enqueue(chunk); + }, + cancel(reason) { + cancelled = reason; + }, + }); + const frames: RawSseFrame[] = []; + const error = await (async () => { + for await (const f of readSseFrames(body)) frames.push(f); + })().catch((e: unknown) => e); + expect(error).toBeInstanceOf(SseProtocolError); + expect(cancelled).toBe(error); + expect(frames).toEqual([]); + // 16 chunks fill the cap exactly; the 17th passes it. A few extra pulls are read-ahead. + expect(pulls).toBeLessThan(20); + }); }); From 34a9f150e8dafb8891bdca23228be2cd964f9d0a Mon Sep 17 00:00:00 2001 From: Duy Nguyen Date: Mon, 12 Oct 2026 09:07:39 +0700 Subject: [PATCH 2/3] fix: keep frames parsed before an SSE overflow When the cap tripped partway through a chunk, the parser threw and dropped the frames that chunk had already completed, such as chat.started or message.completed, so the panel could not learn the session key or settle a continuation. SseProtocolError now carries the frames completed before the overflow. readSseFrames cancels the body, yields those frames, then throws. The client maps SseProtocolError to protocol_error before it checks the abort signal, so a Stop that lands while the body is cancelled no longer turns the fault into a raw error that could offer "Send again". The joined data size no longer counts one newline too many, and the redundant post-line cap check is gone: a parsed line never holds more than its own length. Refs #2 Co-Authored-By: Claude Opus 5.5 --- docs/architecture.md | 2 +- src/api/dewee-client.ts | 13 ++-- src/protocol/sse-reader.ts | 39 ++++++++---- tests/integration/chat/use-chat.test.tsx | 51 +++++++++++++++ tests/unit/api/dewee-client.test.ts | 38 ++++++++++- tests/unit/protocol/sse-reader.test.ts | 81 +++++++++++++++++++++++- 6 files changed, 200 insertions(+), 24 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 29de0dc..01e2746 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,7 +37,7 @@ The extension is a thin Chrome MV3 client. Dewee (Gateway, or the CCP app for De **Cloud connect.** The panel stores the profile, asks the background to run the OAuth flow, then calls `GET /me`. A failed first sign-in removes the new profile. -**Chat.** `use-chat` posts `chat/stream`. `chat.started` gives the session key, which is remembered per workspace and agent. The panel never re-sends a message by itself, and it offers "Send again" only when the server provably did not receive the message: (a) the panel failed before calling fetch (background or credentials unavailable), (b) the fetch was rejected while `navigator.onLine` was false, or (c) the server refused the turn with an explicit 4xx before accepting it (validation, auth, rate limit, or a busy 409 such as `run_in_progress`). Every other failure is uncertain: a 5xx (including a proxy's 502, 503 or 504), a stall, a drop after the response, a fetch rejected while online, a decoder fault (including `protocol_error`: the stream parser holds at most 1 Mi characters for the current line and event, and stops the stream past that), a history read that failed, or a turn not found after a reconcile. An uncertain turn stays visible once, marked `delivery_unconfirmed`, with "Reload conversation" (or "Show conversations" in a new chat, which has no session to check). It never gets "Send again"; the text stays on screen, so the person can copy it and send it again by hand. After an uncertain failure in a known session, or a break after the server answered (200, any frame or `chat.started`), the engine polls run status and reloads history; polling pauses while the browser is offline, slows down after about 3 minutes and keeps the composer closed until the run ends. Only one reconcile runs per conversation, and a switch, unmount or newer reconcile retires the old one. An unconfirmed turn is dropped only when the history provably holds it: the message that was last at send time (its `m` id, role and text) is still in place, and a user turn with the same text follows it (past the turns confirmed by `chat.started` since). After compaction, renumbered ids, or a highest position below the send-time one, the anchor is gone and the turn stays unconfirmed. A history that cannot be read (including one that keeps shrinking while it is read) is never treated as empty: the transcript stays and the turn stays unconfirmed. Unconfirmed turns survive later history reloads and new sends, and every reconcile or reload checks them again. Residual risks: the offline check is a heuristic, so if the connection drops after the request bytes left, `onLine` can already be false and "Send again" may start a second run; and an identical text sent again after the anchor can make a lost repeat look delivered. Stop calls `chat/abort` and aborts the fetch; pressed before `chat.started` names the run, the abort is sent as soon as it does (or the fetch is dropped after 5 s). +**Chat.** `use-chat` posts `chat/stream`. `chat.started` gives the session key, which is remembered per workspace and agent. The panel never re-sends a message by itself, and it offers "Send again" only when the server provably did not receive the message: (a) the panel failed before calling fetch (background or credentials unavailable), (b) the fetch was rejected while `navigator.onLine` was false, or (c) the server refused the turn with an explicit 4xx before accepting it (validation, auth, rate limit, or a busy 409 such as `run_in_progress`). Every other failure is uncertain: a 5xx (including a proxy's 502, 503 or 504), a stall, a drop after the response, a fetch rejected while online, a decoder fault (including `protocol_error`: the stream parser holds at most 1 Mi characters for the current line and event, and stops the stream past that, after delivering the frames that were already complete), a history read that failed, or a turn not found after a reconcile. An uncertain turn stays visible once, marked `delivery_unconfirmed`, with "Reload conversation" (or "Show conversations" in a new chat, which has no session to check). It never gets "Send again"; the text stays on screen, so the person can copy it and send it again by hand. After an uncertain failure in a known session, or a break after the server answered (200, any frame or `chat.started`), the engine polls run status and reloads history; polling pauses while the browser is offline, slows down after about 3 minutes and keeps the composer closed until the run ends. Only one reconcile runs per conversation, and a switch, unmount or newer reconcile retires the old one. An unconfirmed turn is dropped only when the history provably holds it: the message that was last at send time (its `m` id, role and text) is still in place, and a user turn with the same text follows it (past the turns confirmed by `chat.started` since). After compaction, renumbered ids, or a highest position below the send-time one, the anchor is gone and the turn stays unconfirmed. A history that cannot be read (including one that keeps shrinking while it is read) is never treated as empty: the transcript stays and the turn stays unconfirmed. Unconfirmed turns survive later history reloads and new sends, and every reconcile or reload checks them again. Residual risks: the offline check is a heuristic, so if the connection drops after the request bytes left, `onLine` can already be false and "Send again" may start a second run; and an identical text sent again after the anchor can make a lost repeat look delivered. Stop calls `chat/abort` and aborts the fetch; pressed before `chat.started` names the run, the abort is sent as soon as it does (or the fetch is dropped after 5 s). **Cards.** `ui_action` frames and `GET /actions?status=pending` feed the card list. A response goes to `POST /actions/{id}/respond`. If the response is lost (network error or 409) the panel reloads the card. A resolved card's `continuation` is queued for its session, tagged with the profile and workspace captured when the respond call started (a switch during the call does not re-tag it). Queued continuations are sent one at a time once their own session is shown and idle, so two cards answered during a reply both continue after it, and a continuation never streams into another conversation. The queue lives in `storage.session` with one key per item (profile, workspace and setup action; the value holds only those IDs plus session and agent, a `held` flag and the time of the first in-progress answer, never a secret). Adding or removing an item writes only that key, so panels in two windows never remove each other's items, and each panel follows `storage.onChanged` to see what another panel added or delivered. A session or agent switch or a closed panel does not lose an item; reopening the session sends what is pending. Both behave the same while a continuation streams: the stream is dropped without holding the item (this is not a Stop), so it is sent again when its session is shown, here or in a reopened panel, and the Gateway answers `continuation_sent` if it was delivered meanwhile. Unlike user messages, continuations are sent automatically, because the Gateway delivers each at most once. An item leaves the queue only when the agent provably got it: a `message.completed` frame, a 409 `continuation_sent` (already delivered; a silent success that reloads the conversation, with no error bubble), or a `chat.error` with code `setup_result_delivered` (the result reached the agent's session but the run failed; its message is shown and the person continues in the chat). Earlier frames such as `chat.started` are not proof: if the run fails before the result is saved (provider error, rate limit, exhausted credit), the Gateway releases the send lease. It also leaves when it can never succeed (400, 404 or 410, or a 403 other than a key problem: the card or continuation is invalid, gone or not allowed); the panel then says once, with no "Continue", that the card's result could not be delivered and the person should ask the agent again in chat (`continuation_refused`), and logs a `console.warn` with the status and code only (a 400 is a client bug). A 403 `scope_required` or `owner_binding_required` can be fixed by the person, so that item is held with "Continue" and the key error. A 409 `continuation_in_progress` means another request holds a live send lease that may still fail and release, so it is not delivery: the item stays and is sent again after the server's `Retry-After` hint (5 s without one), then 15 s, 30 s and every 60 s, only while its session is shown and idle, for about 11 minutes after the first such answer (a lease goes stale after 10 minutes; the start is stored with the item, so reopening a panel does not restart the window); then it shows "Continue". Any other outcome keeps the item, held, and shows "Continue": any other `chat.error`, `chat.stopped` or Stop, a stream that closes without a terminal frame or drops after the response (the reconcile offers "Continue" once the run ends), and any other refusal, including an unknown 409 code. Resending is safe: the Gateway answers `continuation_sent` if it was delivered and `continuation_in_progress` while a run holds the lease. A held item is stored with a `held` flag, so it is not sent automatically again, neither after a conversation switch nor in a reopened panel; when its session is shown, it offers "Continue". The stored flag wins in every panel: a held copy stored by another panel stops this panel's automatic retries, and an unheld copy (another panel pressed "Continue") releases the item here and removes this panel's "Continue". When another panel delivers the item, its "Continue" here goes away too, and a click on a stale one does nothing. Remaining limit: a held item (including one past the in-progress retry window) stays in `storage.session` until the person presses "Continue" or the browser restarts. diff --git a/src/api/dewee-client.ts b/src/api/dewee-client.ts index 6eeeb69..6411af0 100644 --- a/src/api/dewee-client.ts +++ b/src/api/dewee-client.ts @@ -159,14 +159,15 @@ export class DeweeClient { if (event) yield event; } } catch (error) { + // An oversized line or event comes only from a body the server already answered, and it stays + // a protocol error even when the caller aborts while the body is being cancelled. + if (error instanceof SseProtocolError) { + throw new DeweeApiError(0, 'protocol_error', 'The reply broke the stream format', undefined, true, error); + } if (stalled) throw new DeweeApiError(0, 'stream_stalled', 'The stream stopped responding', undefined, responded); - // Any other fault after the 200 (an oversized line or event, a decoder or mapping bug) still - // means the server took the turn: mark it so callers reconcile instead of offering a resend. - // The original is the cause. + // Any other fault after the 200 (a decoder or mapping bug) still means the server took the + // turn: mark it so callers reconcile instead of offering a resend. The original is the cause. if (responded && !signal?.aborted && !(error instanceof DeweeApiError)) { - if (error instanceof SseProtocolError) { - throw new DeweeApiError(0, 'protocol_error', 'The reply broke the stream format', undefined, true, error); - } throw new DeweeApiError(0, 'runtime_request_failed', 'The reply could not be read', undefined, true, error); } throw error; diff --git a/src/protocol/sse-reader.ts b/src/protocol/sse-reader.ts index c706f8e..0b260a7 100644 --- a/src/protocol/sse-reader.ts +++ b/src/protocol/sse-reader.ts @@ -8,9 +8,15 @@ import type { RawSseFrame } from './chat-events'; */ export const MAX_SSE_PENDING = 1024 * 1024; -/** The stream broke the event-stream contract (a line or event over the cap). It is never retried. */ +/** + * The stream broke the event-stream contract (a line or event over the cap). It is never retried. + * `frames` holds the frames completed earlier in the same chunk, which callers still deliver. + */ export class SseProtocolError extends Error { - constructor(message: string) { + constructor( + message: string, + readonly frames: readonly RawSseFrame[] = [], + ) { super(message); this.name = 'SseProtocolError'; } @@ -22,7 +28,8 @@ const NEWLINE = /\r\n|\r|\n/g; * Incremental Server-Sent Events parser (WHATWG event-stream format). Feed decoded text chunks; * complete frames are returned as soon as their blank-line terminator arrives. Once more than * `MAX_SSE_PENDING` code units would be held, `push` drops what it holds and throws - * `SseProtocolError`, and every later `push` throws too. + * `SseProtocolError` carrying the frames that chunk completed before the overflow; every later + * `push` throws too. */ export class SseParser { private buffer = ''; @@ -48,13 +55,13 @@ export class SseParser { const line = this.buffer + chunk.slice(start, match.index); this.buffer = ''; start = match.index + match[0].length; - this.limit(line.length); + // A parsed line never holds more than its own length, so checking before it is enough. + this.limit(line.length, frames); const frame = this.line(line); if (frame) frames.push(frame); - this.limit(0); } const rest = chunk.length - start; - this.limit(this.buffer.length + rest); + this.limit(this.buffer.length + rest, frames); this.buffer += chunk.slice(start); return frames; } @@ -64,7 +71,7 @@ export class SseParser { } /** Throws, dropping everything held, when `extra` more code units would pass the cap. */ - private limit(extra: number): void { + private limit(extra: number, frames: RawSseFrame[]): void { if (extra + this.held() <= this.maxPending) return; this.failed = true; this.buffer = ''; @@ -72,7 +79,7 @@ export class SseParser { this.data = []; this.dataSize = 0; this.id = undefined; - throw new SseProtocolError(`An event-stream line or event exceeded ${this.maxPending} characters`); + throw new SseProtocolError(`An event-stream line or event exceeded ${this.maxPending} characters`, frames); } private line(line: string): RawSseFrame | null { @@ -91,17 +98,18 @@ export class SseParser { if (value.startsWith(' ')) value = value.slice(1); if (field === 'event') this.event = value; else if (field === 'data') { + // `dataSize` matches the joined `data` length: every line after the first adds a newline. + this.dataSize += value.length + (this.data.length > 0 ? 1 : 0); this.data.push(value); - // The joining newline counts too, so `dataSize` matches the joined `data` length plus one. - this.dataSize += value.length + 1; } else if (field === 'id') this.id = value; return null; } } /** - * Reads a fetch Response body as SSE frames until the stream ends or the signal aborts. Throws - * `SseProtocolError` when a line or event passes the cap; the body is cancelled then. + * Reads a fetch Response body as SSE frames until the stream ends or the signal aborts. When a + * line or event passes the cap it cancels the body, yields the frames completed before the + * overflow, then throws `SseProtocolError`. */ export async function* readSseFrames(body: ReadableStream, maxPending = MAX_SSE_PENDING): AsyncGenerator { const reader = body.getReader(); @@ -115,9 +123,12 @@ export async function* readSseFrames(body: ReadableStream, maxPendin } yield* parser.push(decoder.decode() + '\n\n'); } catch (error) { - if (error instanceof SseProtocolError) await reader.cancel(error).catch(() => {}); + if (error instanceof SseProtocolError) { + await reader.cancel(error).catch(() => {}); + yield* error.frames; + } throw error; } finally { reader.releaseLock(); } -} +} \ No newline at end of file diff --git a/tests/integration/chat/use-chat.test.tsx b/tests/integration/chat/use-chat.test.tsx index 2093b0e..165d5fd 100644 --- a/tests/integration/chat/use-chat.test.tsx +++ b/tests/integration/chat/use-chat.test.tsx @@ -1007,6 +1007,40 @@ describe('useChat delivery proof', () => { expect(methods.streamChat).toHaveBeenCalledTimes(1); }); + it('shows protocol_error in a known session and reconciles, with no "Send again"', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + let failed = false; + let release = () => {}; + const gate = new Promise((resolve) => (release = resolve)); + const { result, methods } = await openSession( + async function* () { + yield started(); + yield { type: 'message.delta', delta: 'partial' } as ChatEvent; + failed = true; + throw new DeweeApiError(0, 'protocol_error', 'The reply broke the stream format', undefined, true, new SseProtocolError('too large')); + }, + { + history: async () => { + if (failed) await gate; + return history(SERVER_HISTORY); + }, + }, + ); + + void result.current.send('Hello again'); + + // While the reconcile reads the history, the failed turn shows the protocol error. + await waitFor(() => expect(lastMessage(result.current.state.messages)?.error?.code).toBe('protocol_error')); + expect(result.current.state.phase).toBe('reconciling'); + expect(result.current.state.messages.some((m) => m.retryText || m.retryContinuation)).toBe(false); + + release(); + await waitFor(() => expect(loads(methods.history)).toBe(2)); + await waitFor(() => expect(result.current.state.phase).toBe('idle')); + expect(result.current.state.messages.some((m) => m.retryText || m.retryContinuation)).toBe(false); + expect(methods.streamChat).toHaveBeenCalledTimes(1); + }); + it('keeps a manual retry for a continuation lost before delivery in a known session, and resends it once', async () => { let call = 0; const { result, methods } = await openSession(async function* () { @@ -1351,6 +1385,23 @@ describe('useChat continuation delivery proof', () => { expect(methods.streamChat).toHaveBeenCalledTimes(1); }); + it('keeps a continuation whose stream broke the format after the response (protocol_error): held, with "Continue"', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + const { result, methods } = await openSession(async function* () { + yield started(); + throw new DeweeApiError(0, 'protocol_error', 'The reply broke the stream format', undefined, true, new SseProtocolError('too large')); + }); + + void result.current.continueWith(CONTINUATION); + + await waitFor(() => expect(lastMessage(result.current.state.messages)?.retryContinuation).toEqual(CONTINUATION)); + expect(lastMessage(result.current.state.messages)?.error?.code).toBe('protocol_error'); + expect(result.current.state.messages.filter((m) => m.retryContinuation)).toHaveLength(1); + await waitFor(async () => expect(await storedContinuations()).toEqual([HELD])); + await idle(); + expect(methods.streamChat).toHaveBeenCalledTimes(1); + }); + it.each([ [400, 'invalid_request'], [403, 'unauthorized'], diff --git a/tests/unit/api/dewee-client.test.ts b/tests/unit/api/dewee-client.test.ts index 3ec77ff..44cd948 100644 --- a/tests/unit/api/dewee-client.test.ts +++ b/tests/unit/api/dewee-client.test.ts @@ -3,7 +3,7 @@ import { join } from 'node:path'; import { DeweeClient, type AuthAdapter } from '../../../src/api/dewee-client'; import { DeweeApiError } from '../../../src/api/errors'; import type { ChatEvent } from '../../../src/protocol/chat-events'; -import { SseProtocolError } from '../../../src/protocol/sse-reader'; +import { MAX_SSE_PENDING, SseProtocolError } from '../../../src/protocol/sse-reader'; const BASE = 'https://gw.example.test/v1/extension'; const card = JSON.parse( @@ -383,6 +383,42 @@ describe('streamChat', () => { expect(received.map((e) => e.type)).toEqual(['chat.started']); }); + /** One chunk holding a complete chat.started frame followed by a line past the default cap. */ + function overflowAfterStarted(onCancel?: () => void): ReadableStream { + const text = sse('chat.started', { run_id: 'r1', session_key: 's1', agent_key: 'a' }) + `data: ${'x'.repeat(MAX_SSE_PENDING)}\n`; + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + cancel() { + onCancel?.(); + }, + }); + } + + it('delivers chat.started from the chunk that overflows, so the session key is learned', async () => { + const fetch = vi.fn(async () => new Response(overflowAfterStarted(), { status: 200 })); + const received: ChatEvent[] = []; + const run = async () => { + for await (const event of client(fetch, { streamIdleMs: 1000 }).streamChat('ws_1', { agent_key: 'a', message: 'Hi' })) received.push(event); + }; + const error = await run().catch((e: unknown) => e); + expect(received).toEqual([{ type: 'chat.started', runId: 'r1', sessionKey: 's1', agentKey: 'a' }]); + expect(error).toMatchObject({ status: 0, code: 'protocol_error', afterResponse: true }); + }); + + it('keeps protocol_error after the response when the caller aborts while the body is cancelled', async () => { + const stop = new AbortController(); + const fetch = vi.fn(async () => new Response(overflowAfterStarted(() => stop.abort()), { status: 200 })); + const run = async () => { + for await (const event of client(fetch, { streamIdleMs: 1000 }).streamChat('ws_1', { agent_key: 'a', message: 'Hi' }, stop.signal)) void event; + }; + const error = await run().catch((e: unknown) => e); + expect(stop.signal.aborted).toBe(true); + expect(error).toBeInstanceOf(DeweeApiError); + expect(error).toMatchObject({ status: 0, code: 'protocol_error', afterResponse: true }); + }); + it('surfaces an auth adapter bug unchanged', async () => { const bug = new Error('adapter bug'); const fetch = vi.fn(); diff --git a/tests/unit/protocol/sse-reader.test.ts b/tests/unit/protocol/sse-reader.test.ts index 2123705..edddbdb 100644 --- a/tests/unit/protocol/sse-reader.test.ts +++ b/tests/unit/protocol/sse-reader.test.ts @@ -60,7 +60,8 @@ describe('SseParser', () => { expect(parser.pending).toBeLessThanOrEqual(1000); } }).toThrow(SseProtocolError); - // Each line holds 91 (value plus joining newline): the 11th line (96) on top of 910 passes 1000. + // The first line holds 90, each later one 91 with its joining newline: the 11th line (96) on top + // of 909 passes 1000. expect(lines).toBe(10); expect(parser.pending).toBe(0); }); @@ -84,7 +85,7 @@ describe('SseParser', () => { const max = 1000; const parser = new SseParser(max); // While the data line is read, the held event name (13) plus the whole line ("data: " + value) - // must fit; afterwards the name plus the value and its joining newline. + // must fit; afterwards the name plus the value. const value = 'v'.repeat(max - 13 - 6); const text = `event: message.delta\r\ndata: ${value}\r\n\r\n`; const frames = []; @@ -99,6 +100,45 @@ describe('SseParser', () => { const value = 'v'.repeat(max - 13 - 5); expect(() => parser.push(`event: message.delta\ndata: ${value}\n\n`)).toThrow(SseProtocolError); }); + + it('counts joined data without an extra newline', () => { + const parser = new SseParser(); + parser.push('data: ab\ndata: cd\n'); + // "ab\ncd" is 5 code units. + expect(parser.pending).toBe(5); + }); + + it('keeps the frames a chunk completed before an oversized line', () => { + const parser = new SseParser(100); + let failure: unknown; + try { + parser.push(`event: chat.started\ndata: {"session_key":"s1"}\n\ndata: ${'x'.repeat(200)}\n`); + } catch (error) { + failure = error; + } + expect(failure).toBeInstanceOf(SseProtocolError); + expect((failure as SseProtocolError).frames).toEqual([{ event: 'chat.started', data: '{"session_key":"s1"}' }]); + expect(parser.pending).toBe(0); + }); + + it('stops an oversized id line', () => { + const parser = new SseParser(100); + expect(() => parser.push(`id: ${'i'.repeat(200)}\n`)).toThrow(SseProtocolError); + expect(parser.pending).toBe(0); + }); + + it('counts a held id toward the cap', () => { + const parser = new SseParser(100); + parser.push(`id: ${'i'.repeat(90)}\n`); + expect(parser.pending).toBe(90); + expect(() => parser.push(`data: ${'d'.repeat(20)}`)).toThrow(SseProtocolError); + }); + + it('stops an oversized comment line', () => { + const parser = new SseParser(100); + expect(() => parser.push(`: ${'c'.repeat(200)}\n`)).toThrow(SseProtocolError); + expect(parser.pending).toBe(0); + }); }); describe('readSseFrames', () => { @@ -139,4 +179,41 @@ describe('readSseFrames', () => { // 16 chunks fill the cap exactly; the 17th passes it. A few extra pulls are read-ahead. expect(pulls).toBeLessThan(20); }); + + it('yields the frames before an oversized line in the same chunk, then fails', async () => { + const text = `event: chat.started\ndata: {"session_key":"s1"}\n\nevent: message.completed\ndata: {}\n\ndata: ${'x'.repeat(200)}\n`; + let cancelled: unknown; + const body = new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode(text)); + }, + cancel(reason) { + cancelled = reason; + }, + }); + const frames: RawSseFrame[] = []; + const error = await (async () => { + for await (const f of readSseFrames(body, 100)) frames.push(f); + })().catch((e: unknown) => e); + expect(frames.map((f) => f.event)).toEqual(['chat.started', 'message.completed']); + expect(error).toBeInstanceOf(SseProtocolError); + expect(cancelled).toBe(error); + }); + + it('still throws the protocol error when cancelling the body rejects', async () => { + const body = new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode(`data: ${'x'.repeat(200)}`)); + }, + cancel() { + throw new Error('cancel failed'); + }, + }); + const frames: RawSseFrame[] = []; + const error = await (async () => { + for await (const f of readSseFrames(body, 100)) frames.push(f); + })().catch((e: unknown) => e); + expect(error).toBeInstanceOf(SseProtocolError); + expect(frames).toEqual([]); + }); }); From 955e8be04a10389fd7a883ab4e046f9f195dc32e Mon Sep 17 00:00:00 2001 From: Duy Nguyen Date: Mon, 12 Oct 2026 09:14:04 +0700 Subject: [PATCH 3/3] fix: keep a finished turn when the stream fails after its terminal frame When the chunk that overflowed the SSE cap also completed message.completed, the reply finished normally and the protocol_error that followed then added an error bubble and a reconcile to a turn that had already succeeded. use-chat now ignores a stream error once the turn has seen its terminal frame. A continuation that saw message.completed is already acknowledged and out of the queue, and stays so. The client keeps reading after a terminal frame, so frames the server may still send are not cut off. Refs #2 Co-Authored-By: Claude Opus 5.5 --- src/chat/use-chat.ts | 4 ++ tests/integration/chat/use-chat.test.tsx | 55 +++++++++++++++++++++++- tests/unit/protocol/sse-reader.test.ts | 6 +-- 3 files changed, 60 insertions(+), 5 deletions(-) diff --git a/src/chat/use-chat.ts b/src/chat/use-chat.ts index be2f5d0..f7795e8 100644 --- a/src/chat/use-chat.ts +++ b/src/chat/use-chat.ts @@ -567,6 +567,10 @@ export function useChat(ctx: ChatContext | null, hooks: { onSessionKey(key: stri } } catch (error) { if (!live()) return; + // The turn already ended with its terminal frame (a continuation that saw `message.completed` + // already left the queue): a fault in what follows, such as an oversized line in the same + // chunk, must not turn a finished reply into a failure. + if (terminal) return; if (error instanceof DOMException && error.name === 'AbortError') { if (item && !settled) { // Stopped before the agent provably got it: it stays queued for "Continue" (the Gateway diff --git a/tests/integration/chat/use-chat.test.tsx b/tests/integration/chat/use-chat.test.tsx index 165d5fd..6c0c4b9 100644 --- a/tests/integration/chat/use-chat.test.tsx +++ b/tests/integration/chat/use-chat.test.tsx @@ -1,12 +1,12 @@ import { act, renderHook, waitFor } from '@testing-library/react'; -import type { DeweeClient } from '../../../src/api/dewee-client'; +import { DeweeClient } from '../../../src/api/dewee-client'; import { DeweeApiError } from '../../../src/api/errors'; import type { HistoryMessage } from '../../../src/api/schemas'; import type { ChatMessage } from '../../../src/chat/chat-state'; import { addPendingContinuation, loadPendingContinuations, removePendingContinuation } from '../../../src/chat/pending-continuations'; import { useChat, type ChatContext } from '../../../src/chat/use-chat'; import type { ChatEvent } from '../../../src/protocol/chat-events'; -import { SseProtocolError } from '../../../src/protocol/sse-reader'; +import { MAX_SSE_PENDING, SseProtocolError } from '../../../src/protocol/sse-reader'; import { parseUiAction } from '../../../src/protocol/ui-action'; import { readFileSync } from 'node:fs'; import { join } from 'node:path'; @@ -1402,6 +1402,57 @@ describe('useChat continuation delivery proof', () => { expect(methods.streamChat).toHaveBeenCalledTimes(1); }); + /** + * The real client over one body chunk holding chat.started, message.completed and then a line + * past the parser cap, so the turn ends normally before the stream fails with protocol_error. + */ + const completedThenOverflow = (signal?: AbortSignal) => { + const frame = (event: string, data: unknown) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; + const text = + frame('chat.started', { run_id: 'r1', session_key: SESSION, agent_key: 'super-agent' }) + + frame('message.completed', { message: { role: 'assistant', content: 'Done', created_at: '2026-10-12T09:00:00Z' } }) + + `data: ${'x'.repeat(MAX_SSE_PENDING)}\n`; + const fetch = async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + }), + { status: 200 }, + ); + const real = new DeweeClient({ baseUrl: 'https://gw.example.test/v1/extension', auth: { authorization: async () => 'Bearer test-api-key' }, fetch, streamIdleMs: 1000 }); + return real.streamChat(WORKSPACE, { agent_key: 'super-agent', message: 'Hi', session_key: SESSION }, signal); + }; + + it('keeps a reply completed in the chunk that overflows: no error bubble and no "Send again"', async () => { + const logged = vi.spyOn(console, 'error').mockImplementation(() => {}); + const { result, methods } = await openSession(completedThenOverflow); + + void result.current.send('Hello again'); + + await waitFor(() => expect(lastMessage(result.current.state.messages)?.content).toBe('Done')); + await idle(); + expect(result.current.state.phase).toBe('idle'); + expect(result.current.state.messages.some((m) => m.error || m.retryText || m.retryContinuation || m.recover)).toBe(false); + expect(methods.streamChat).toHaveBeenCalledTimes(1); + expect(logged).not.toHaveBeenCalledWith('Chat stream failed', expect.anything()); + }); + + it('acknowledges a continuation whose message.completed came in the chunk that overflows', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + const { result, methods } = await openSession(completedThenOverflow); + + void result.current.continueWith(CONTINUATION); + + await waitFor(() => expect(lastMessage(result.current.state.messages)?.content).toBe('Done')); + await waitFor(async () => expect(await storedContinuations()).toEqual([])); + await idle(); + expect(result.current.state.phase).toBe('idle'); + expect(result.current.state.messages.some((m) => m.error || m.retryContinuation)).toBe(false); + expect(methods.streamChat).toHaveBeenCalledTimes(1); + }); + it.each([ [400, 'invalid_request'], [403, 'unauthorized'], diff --git a/tests/unit/protocol/sse-reader.test.ts b/tests/unit/protocol/sse-reader.test.ts index edddbdb..a4b451d 100644 --- a/tests/unit/protocol/sse-reader.test.ts +++ b/tests/unit/protocol/sse-reader.test.ts @@ -200,10 +200,10 @@ describe('readSseFrames', () => { expect(cancelled).toBe(error); }); - it('still throws the protocol error when cancelling the body rejects', async () => { + it('still yields the earlier frame and throws the protocol error when cancelling the body rejects', async () => { const body = new ReadableStream({ start(c) { - c.enqueue(new TextEncoder().encode(`data: ${'x'.repeat(200)}`)); + c.enqueue(new TextEncoder().encode(`event: chat.started\ndata: {}\n\ndata: ${'x'.repeat(200)}`)); }, cancel() { throw new Error('cancel failed'); @@ -214,6 +214,6 @@ describe('readSseFrames', () => { for await (const f of readSseFrames(body, 100)) frames.push(f); })().catch((e: unknown) => e); expect(error).toBeInstanceOf(SseProtocolError); - expect(frames).toEqual([]); + expect(frames).toEqual([{ event: 'chat.started', data: '{}' }]); }); });