diff --git a/docs/architecture.md b/docs/architecture.md index b48cd0e..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, 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 56878b6..6411af0 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, @@ -159,6 +159,11 @@ 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 (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. 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..f7795e8 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 @@ -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/src/protocol/sse-reader.ts b/src/protocol/sse-reader.ts index db0483a..0b260a7 100644 --- a/src/protocol/sse-reader.ts +++ b/src/protocol/sse-reader.ts @@ -1,35 +1,94 @@ 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. + * `frames` holds the frames completed earlier in the same chunk, which callers still deliver. + */ +export class SseProtocolError extends Error { + constructor( + message: string, + readonly frames: readonly RawSseFrame[] = [], + ) { + 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` carrying the frames that chunk completed before the overflow; 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; + // 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); } + const rest = chunk.length - start; + this.limit(this.buffer.length + rest, frames); + 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, frames: RawSseFrame[]): 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`, frames); + } + 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 +97,24 @@ 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') { + // `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); + } 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. 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(); const decoder = new TextDecoder(); - const parser = new SseParser(); + const parser = new SseParser(maxPending); try { for (;;) { const { value, done } = await reader.read(); @@ -56,7 +122,13 @@ 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(() => {}); + yield* error.frames; + } + throw error; } finally { reader.releaseLock(); } -} +} \ No newline at end of file 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..6c0c4b9 100644 --- a/tests/integration/chat/use-chat.test.tsx +++ b/tests/integration/chat/use-chat.test.tsx @@ -1,11 +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 { 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'; @@ -990,6 +991,56 @@ 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('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* () { @@ -1334,6 +1385,74 @@ 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); + }); + + /** + * 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/api/dewee-client.test.ts b/tests/unit/api/dewee-client.test.ts index e9c5f70..44cd948 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 { MAX_SSE_PENDING, SseProtocolError } from '../../../src/protocol/sse-reader'; const BASE = 'https://gw.example.test/v1/extension'; const card = JSON.parse( @@ -356,6 +357,68 @@ 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']); + }); + + /** 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 693792e..a4b451d 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,131 @@ 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); + // 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); + }); + + 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. + 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); + }); + + 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', () => { @@ -29,4 +155,65 @@ 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); + }); + + 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 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(`event: chat.started\ndata: {}\n\ndata: ${'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([{ event: 'chat.started', data: '{}' }]); + }); });