fix: cap SSE line and event buffer

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<position>` 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<position>` 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.
 

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.

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.
  */

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

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,25 +97,38 @@ 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<Uint8Array>): AsyncGenerator<RawSseFrame> {
+/**
+ * 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<Uint8Array>, maxPending = MAX_SSE_PENDING): AsyncGenerator<RawSseFrame> {
   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();
       if (done) break;
       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();
   }
-}
+}

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<MessageKey, string> = {
   '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.',