# Stream Module -- TRD ## Status Implemented ## Overview The SSE (Server-Sent Events) streaming protocol between `@coda/server` and `@coda/client` is currently split across three locations with no shared contract: the server formats frames in `sse-utils.ts` and emits stringly-typed events in `stream-handler.ts`, while the client parses the wire format in `readSSEStream`, routes events through an `SSE_HANDLERS` map full of `as` casts, and falls back to `dispatchSSEEvent`. Event names, payload shapes, and wire format are duplicated. Adding or renaming an event requires changes in multiple files with no compile-time safety. This TRD defines a single **stream module** (`api/src/stream.ts`) in the shared `@coda/core-api` package that provides: - **`StreamEventMap`** -- a TypeScript interface that is the single source of truth for every event name and its payload type. - **`StreamWriter`** (server) and **`StreamReader`** (client) -- typed classes that enforce the contract at compile time. - **`SSEWritable`** / **`SSEReadable`** -- thin transport adapters that handle SSE wire format, decoupled from the typed event layer. After migration, the server writes `writer.event("chunk", { chunk: text })` and the client reads `.on("chunk", (d) => ...)` with full type inference. No `as` casts, no string literals that can drift, no duplicated parsing logic. ## Goals ### Goals | # | Goal | | --- | -------------------------------------------------------------------------------------------------------------- | | G1 | Single source of truth for all SSE event names and payload shapes, enforced by the TypeScript compiler | | G2 | Eliminate duplicated SSE wire-format code between server and client | | G3 | Provide 140+ unit tests covering serialization, parsing, guards, abort, and edge cases | | G4 | Zero infrastructure cost change -- same HTTP/1.1 SSE transport, same deployment | | G5 | Incremental migration path -- server and client can be updated in separate commits without breaking the stream | | G6 | W3C EventSource-compliant SSE parsing (blank-line dispatch, multi-data concatenation) | ### Non-Goals | # | Non-Goal | Rationale | | --- | ----------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | N1 | Switch to WebSockets | SSE is unidirectional (server-to-client), which matches the streaming use case exactly. WebSockets add bidirectional complexity, connection upgrade negotiation, and proxy compatibility issues for no benefit here. | | N2 | Create a separate `@coda/stream` package | The stream module is tightly coupled to the types already in `@coda/core-api` (`UsageStats`, `SourceLink`, `ConversationStep`, etc.). A separate package would add a circular dependency or require extracting those types first. | | N3 | Implement keepalive / heartbeat | Long-running streams can be interrupted by proxies with idle timeouts. `X-Accel-Buffering: no` mitigates nginx buffering. A heartbeat mechanism may be added in a future iteration but is out of scope here. | | N4 | Support binary payloads or file streaming | All payloads are JSON. File attachments are sent as data URLs within JSON, not as binary streams. | ## Architecture ### Layer Diagram ```mermaid graph TB subgraph "@coda/core-api -- stream.ts" SEM["StreamEventMap
event name -> payload contract"] SW["StreamWriter
typed event emitter"] SR["StreamReader
typed event consumer"] SSEW["SSEWritable
wire format adapter"] SSER["SSEReadable
wire format parser"] end subgraph Server RES["Express res"] SH["stream-handler.ts"] end subgraph Client FETCH["fetch response.body"] HOOK["use-coda-orchestrator.ts"] end SEM -.->|types| SW SEM -.->|types| SR RES -->|createSSEWritable| SSEW SSEW --> SW SH -->|"writer.event() / writer.end()"| SW FETCH -->|createSSEReadable| SSER SSER --> SR HOOK -->|".on().consume()"| SR ``` ### Data Flow ```mermaid sequenceDiagram participant Server as stream-handler.ts participant SW as StreamWriter participant Wire as SSE Wire (HTTP) participant SR as StreamReader participant Client as React Hook Server->>SW: writer.event("message_start", {...}) SW->>Wire: event: message_start\ndata: {"userMessageId":...}\n\n Wire->>SR: async iterator yields {event, data} SR->>Client: on("message_start") handler called with typed payload loop Streaming tokens Server->>SW: writer.event("chunk", {chunk: text}) SW->>Wire: event: chunk\ndata: {"chunk":"..."}\n\n Wire->>SR: yield SR->>Client: on("chunk") handler end Server->>SW: writer.end("complete", msgId, userMsgId) SW->>Wire: event: done\ndata: {"done":true,"reason":"complete",...}\n\n Wire->>SR: yield SR->>Client: on("done") handler Note over SR: consume() promise resolves ``` ### Module Exports All exports live in `api/src/stream.ts` and are re-exported from `api/src/index.ts`: | Export | Kind | Used By | | ------------------- | ------------ | ---------------------------------------------- | | `StreamEventMap` | Type | Server, Client (type inference) | | `StreamEndReason` | Type | Server (end reason), Client (done handler) | | `SSEWritable` | Interface | Server (transport layer) | | `SSEReadable` | Interface | Client (transport layer) | | `StreamWriter` | Class | Server (`stream-handler.ts`) | | `StreamReader` | Class | Client (`ApiClient.consumeStream`) | | `createSSEWritable` | Function | Server (factory for Express `res`) | | `createSSEReadable` | Function | Client (factory for fetch `ReadableStream`) | | `isStreamEndReason` | Type guard | Server (abort signal validation) | | `SSE_HEADERS` | Const object | Server (reference; set by `createSSEWritable`) | ## Detailed Design ### StreamEventMap -- The Event Contract `StreamEventMap` is a TypeScript interface where each key is an event name and each value is the payload type. This is the **single source of truth** for the streaming protocol. ```ts interface StreamEventMap { // Streaming content chunk: { chunk: string }; reasoning: { chunk: string } | { done: true }; progress: { steps: ConversationStep[] } | { done: true } | { chunk: string }; clear: Record; // Metadata message_start: { userMessageId: string; assistantMessageId: string; model: { id: string; displayName: string; provider: string }; }; usage: UsageStats; sources: SourceLink[]; suggestions: string[]; attachments: ExtractedAttachment[]; title: { title: string }; warnings: string[]; compression: { messageCount: number } | { done: true }; // Interactive selection_required: SelectionRequired; // Terminal done: { done: true; messageId: string; userMessageId: string; reason: StreamEndReason; }; error: { error: string; message: string }; } ``` **Key design choices:** - **Discriminated unions** (`progress`, `reasoning`, `compression`) use `"done" in data`, `"steps" in data`, `"chunk" in data` for runtime discrimination. No enum, no `type` field -- the presence of a key is the discriminant. - **Unwrapped arrays** for `suggestions`, `warnings`, `sources`, `attachments`. The old protocol wrapped these (`{ suggestions: string[] }`); the new protocol sends the array directly. Less nesting, same JSON. - **`clear` uses `Record`** -- TypeScript's way of expressing "empty object, no fields allowed." - **`error` has both `error` and `message` fields** with the same value. `error` is the machine-readable key for error detection; `message` is the human-readable text. Both kept for protocol clarity. ### StreamEndReason ```ts type StreamEndReason = | "complete" | "max_rounds" | "cancelled" | "context_limit" | "timeout"; ``` | Bedrock `stop_reason` | Maps to `StreamEndReason` | Notes | | --------------------- | ------------------------- | ------------------------------ | | `end_turn` | `complete` | Normal completion | | `max_tokens` | `context_limit` | Token budget exhausted | | `stop_sequence` | `complete` | Stop sequence hit | | `tool_use` | _(not terminal)_ | Server continues the tool loop | The `isStreamEndReason` type guard validates abort signal reasons at runtime: ```ts const STREAM_END_REASONS: ReadonlySet = new Set([...]); function isStreamEndReason(value: unknown): value is StreamEndReason { return typeof value === "string" && STREAM_END_REASONS.has(value); } ``` ### Event Changes from Current Implementation | Current | New | Change | | ------------------------------------------- | ------------------------------------ | -------------------------------------------------------- | | Unnamed `data:` frames for chunk/done/error | Named `event:` frames for all events | All events use `event: name\ndata: json\n\n` | | `thinking` | `progress` | Renamed -- distinct from `reasoning` (extended thinking) | | `clear_partial` | `clear` | Shortened | | `suggestions: { suggestions: string[] }` | `suggestions: string[]` | Unwrapped | | `warnings: { warnings: string[] }` | `warnings: string[]` | Unwrapped | | `message_start` without model | `message_start` with `model` field | Supports variable model selection | | `progress: { clear: true }` | `progress: { done: true }` | Signals phase end, not UI directive | | `progress: { thinking: string }` | `progress: { chunk: string }` | Consistent `chunk` naming | | No `reason` on done | `done: { reason: StreamEndReason }` | Surfaces max_rounds, context_limit, timeout, cancelled | | No compression event | `compression` | Future-ready for context compression | ### SSE Wire Format Every event is serialized as a named SSE frame: ``` event: chunk data: {"chunk":"Hello, how can I help?"} ``` The frame format is `event: \ndata: \n\n`. The trailing blank line is the SSE dispatch boundary per the W3C EventSource specification. ### SSEWritable (Server Transport) ```ts interface SSEWritable { write(event: string, data: string): void; end(): void; } ``` `createSSEWritable(res)` is the factory for Express responses. It: 1. Calls `res.writeHead(200, SSE_HEADERS)` immediately. 2. Returns an `SSEWritable` that formats each call as `event: ${event}\ndata: ${data}\n\n`. SSE headers set: | Header | Value | Purpose | | ------------------- | ------------------- | ------------------------------ | | `Content-Type` | `text/event-stream` | SSE content type | | `Cache-Control` | `no-cache` | Prevent response caching | | `Connection` | `keep-alive` | Maintain persistent connection | | `X-Accel-Buffering` | `no` | Disable nginx proxy buffering | ### SSEReadable (Client Transport) ```ts interface SSEReadable { [Symbol.asyncIterator](): AsyncIterableIterator<{ event: string; data: string; }>; } ``` `createSSEReadable(body)` is the factory for a fetch `ReadableStream`. It returns an async iterable that yields `{ event, data }` pairs. **Parser implementation details:** - **W3C-compliant dispatch**: Events dispatch on blank-line boundaries. Multiple `data:` lines within a single event are concatenated with `\n`. The protocol uses single-line JSON, so concatenation is a no-op in practice, but the parser handles any valid SSE stream. - **Buffering strategy**: Uses a `string[]` pending buffer with cumulative length tracking (`pendingLen`) instead of string concatenation on every chunk. Chunks with no newline are pushed to the array without joining. - **indexOf optimization**: The first `indexOf("\n")` result from the decoded chunk is reused, adjusted by `pendingLen`, to avoid redundant scanning of the joined buffer. The inner loop uses `do...while` since the first newline position is already known. - **Cleanup**: `try/finally` releases the reader lock on all exit paths -- normal completion, early `break`, and errors. ### StreamWriter (Server-Side Typed Emitter) ```ts class StreamWriter { event(name: K, data: StreamEventMap[K]): void; end(reason: StreamEndReason, messageId: string, userMessageId: string): void; error(message: string): void; get closed(): boolean; } ``` **Behavioral contracts:** | Behavior | Detail | | ------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Write guard | After `end()` or `error()`, all subsequent `event()`, `end()`, `error()` calls are silent no-ops. The `closed` getter returns `true`. | | AbortSignal | Constructor accepts `{ signal?: AbortSignal }`. On abort: if `signal.reason` is a valid `StreamEndReason`, sends `done` with that reason; otherwise sends `error("Stream aborted unexpectedly")`. | | Pre-aborted signal | If the signal is already aborted at construction time, the abort handler fires immediately in the constructor. | | Message ID tracking | `message_start` events are intercepted to capture `messageId` and `userMessageId`. If abort fires, the `done` event includes these tracked IDs. If abort fires before `message_start`, IDs are empty strings. | **Server usage pattern:** ```ts const ac = new AbortController(); const timeoutId = setTimeout(() => ac.abort("timeout"), config.timeout); res.on("close", () => ac.abort("cancelled")); const writable = createSSEWritable(res); const writer = new StreamWriter(writable, { signal: ac.signal }); writer.event("message_start", { userMessageId, assistantMessageId, model }); // ... streaming loop ... writer.end("complete", assistantMsgId, userMsgId); clearTimeout(timeoutId); ``` ### StreamReader (Client-Side Typed Consumer) ```ts class StreamReader { on( event: K, handler: (data: StreamEventMap[K]) => void, ): this; onParseError( handler: (event: string, raw: string, error: unknown) => void, ): this; consume(): Promise; } ``` **Behavioral contracts:** | Behavior | Detail | | ------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Handler errors propagate | If a handler throws synchronously, the exception propagates out of `consume()` (promise rejects) and the stream stops. This is intentional -- a handler throwing is a consumer bug, not a transport issue. | | Late registration | Handlers registered after `consume()` is called receive subsequent events. The handler map is read live during dispatch, not snapshotted. | | Unregistered events | Silently ignored. No `JSON.parse` is called, no error is raised. Events can be added to the protocol without breaking existing consumers. | | Idempotent consume | Second `consume()` call returns the same promise. | | Parse errors | Malformed JSON is skipped. If `onParseError` handlers are registered, they receive the event name, raw string, and error. The stream continues processing subsequent frames. | **Client usage pattern:** ```ts const readable = createSSEReadable(response.body); await new StreamReader(readable) .on("message_start", ({ assistantMessageId, userMessageId, model }) => { setMessageIds(assistantMessageId, userMessageId); setActiveModel(model); }) .on("chunk", ({ chunk }) => session.bufferChunk(chunk)) .on("progress", (data) => { if ("steps" in data) updateSteps(data.steps); else if ("done" in data) clearProgress(); else updateThinkingText(data.chunk); }) .on("done", ({ reason }) => finalize(reason)) .on("error", ({ message }) => { throw new StreamingError(message); }) .onParseError((event, raw, err) => logger.warn("Malformed SSE", { event, raw, err }), ) .consume(); ``` ## Alternatives Explored ### Why not WebSockets? SSE is inherently unidirectional (server-to-client), which is exactly the streaming use case: the server pushes tokens, progress, and metadata to the client. WebSockets would add: - Bidirectional connection complexity (the client already sends requests via REST POST). - Connection upgrade negotiation, which some corporate proxies block. - A need to implement reconnection, heartbeat, and message framing that SSE provides for free. - No benefit -- the client never needs to push data mid-stream. ### Why not a separate `@coda/stream` package? The stream module depends on types already defined in `@coda/core-api`: `UsageStats`, `SourceLink`, `ConversationStep`, `SelectionRequired`, `ExtractedAttachment`. Creating a separate package would either: - Introduce a circular dependency (`stream` imports from `api`, `api` re-exports from `stream`), or - Require extracting all shared types into a fourth `@coda/types` package, which is over-engineering for the current monorepo size. Placing the module in `api/src/stream.ts` keeps it co-located with the types it depends on and the client that consumes it. ### Why typed discriminated unions over loose event strings? The current implementation uses string event names (`"thinking"`, `"clear_partial"`) with untyped payloads and `as` casts in the client. This has caused: - Runtime errors from payload shape mismatches after server changes. - Silent failures when event names are renamed on one side but not the other. - Boilerplate `as` casts that defeat the purpose of TypeScript. `StreamEventMap` enforces at compile time that `writer.event("chunk", { chunk: text })` matches the expected payload and that `.on("chunk", (d) => d.chunk)` has the correct type. Adding a new event requires adding it to one interface, and both sides get type errors until they handle it. ## Cost Analysis | Category | Impact | | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Engineering effort | ~3-5 days across 12 tasks (can be parallelized across server/client) | | Infrastructure cost | **Zero change** -- same HTTP/1.1 SSE transport, same Express server, same deployment | | Bundle size | Net reduction: removes ~200 lines of duplicated SSE parsing/formatting code from server and client; adds ~250 lines of shared module. Net delta is negligible after tree-shaking. | | Maintenance cost | **Reduction** -- one file to update when events change instead of three. Compile-time safety catches drift. 140+ tests provide regression coverage. | | Migration risk | Low -- incremental task-by-task migration with per-task commits. Functional tests verify no regression at each step. | ## Performance Analysis ### SSE Wire Format | Concern | Approach | Impact | | ---------------- | ---------------------------------------------------------------------------- | --------------------------------------------------------------------- | | Serialization | `JSON.stringify(data)` per event -- same as current implementation | No change | | Single-line JSON | All payloads are single-line JSON. No multi-line `data:` fields in practice. | No multi-line join overhead in the reader | | Parse overhead | `JSON.parse(raw)` per event in `StreamReader.dispatch()` -- same as current | No change | | Skipped events | Unregistered events skip `JSON.parse` entirely (handler map check first) | Minor improvement over current implementation which parses all events | ### SSE Parser Buffering Strategy The `createSSEReadable` parser avoids string concatenation on every chunk: | Technique | Benefit | | ------------------------- | -------------------------------------------------------------------------------------------------------------------------------------- | | `string[]` pending buffer | Chunks without newlines are pushed to an array, not concatenated. Join happens only when a newline is found. | | `pendingLen` tracking | Cumulative length avoids re-measuring the buffer array on every chunk. | | `indexOf` reuse | The first `indexOf("\n")` on the decoded chunk is adjusted by `pendingLen` to index into the joined buffer, avoiding a redundant scan. | | `do...while` inner loop | Since the first newline position is already known, the inner loop starts processing immediately without a redundant condition check. | | `string.substring()` | Used instead of `string.slice()` for line extraction -- both are O(n) but `substring` avoids negative-index edge cases. | ### Benchmark Expectations For a typical streaming response (500 chunks, ~10 metadata events): | Metric | Value | Notes | | --------------------------------- | -------- | ------------------------------------------------- | | Parse overhead per chunk | < 0.01ms | `indexOf` + `substring` on small strings | | JSON.parse per event | < 0.05ms | Small JSON payloads (< 1KB typical) | | Memory per pending buffer | < 1KB | Chunks are typically 10-100 bytes (single tokens) | | Total parse overhead per response | < 10ms | Negligible compared to LLM latency (seconds) | ## Scaling Characteristics | Dimension | Characteristic | Limit | | --------------------- | ------------------------------------------------------------------------- | ----------------------------------------------------------------------------- | | Event throughput | Bounded by Node.js `res.write()` throughput and network bandwidth | Typically 100-500 events/sec for token streaming; well within limits | | Memory per connection | One `StreamWriter` instance (~200 bytes) + Express response buffer | Minimal; Express handles backpressure via `res.write()` return value | | Concurrent streams | Bounded by Express server connection limit and available file descriptors | Current deployment handles ~50 concurrent streams; no change | | Payload size | JSON.stringify has no practical limit; SSE frames are unbounded | Large payloads (attachments as data URLs) can be 1-10MB; see Breakdown Points | | Client-side memory | One `StreamReader` instance + handler closures + pending buffer | Pending buffer is flushed on every newline; typically < 1KB resident | ## Breakdown Points & Mitigations | Breakdown Point | Impact | Mitigation | | -------------------------- | ---------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | **Large payloads (> 1MB)** | `JSON.stringify` / `JSON.parse` blocking the event loop for large attachment data URLs | Current behavior, not new. Future: chunked attachment streaming or out-of-band URLs. | | **Connection drops** | Client receives partial stream; `consume()` resolves without `done` event | Client should treat stream end without `done` as an error. `StreamReader` does not auto-reconnect (SSE reconnect is a browser EventSource feature, not applicable to fetch-based streams). | | **Malformed events** | Corrupted JSON from proxy interference or encoding issues | `onParseError` handler logs the event; stream continues processing subsequent valid frames. | | **Backpressure** | If the client cannot consume events fast enough, Node.js `res.write()` buffers in memory | Express handles TCP backpressure natively. If the client disconnects, `res.on("close")` fires the abort signal, which triggers `writer.end("cancelled")`. | | **Proxy idle timeout** | Long tool executions (30s+) produce no events, causing proxy disconnect | `X-Accel-Buffering: no` header mitigates nginx. No heartbeat mechanism in this iteration (documented non-goal). Progress events during tool execution partially mitigate. | | **Abort race conditions** | Signal fires between `writer.event()` and `writer.end()` | Write guard (`this.ended` flag) ensures at most one terminal event. All methods check `ended` before writing. | | **Multiple data: lines** | SSE spec allows multiple `data:` lines per event, joined with `\n` | Parser handles this correctly per W3C spec. Protocol uses single-line JSON, so this is a no-op in practice but the parser is correct for any valid SSE stream. | ## Decision Log | # | Decision | Rationale | | --- | --------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | D1 | **All events use named `event:` frames** | Current implementation uses unnamed `data:` frames for chunk/done/error. Named events enable `StreamReader` to route by event name without parsing the JSON payload first. | | D2 | **`reason` field on `done` event** | The current implementation has no way to distinguish normal completion from max_rounds, timeout, or cancellation. The client needs this to show appropriate UI (e.g., "Response truncated -- context limit reached"). | | D3 | **No keepalive / heartbeat** | Adds complexity for a problem that rarely occurs in practice (most streams complete in < 30s). `X-Accel-Buffering: no` handles the common case. Can be added later without protocol changes. | | D4 | **`error` payload has both `error` and `message` fields** | `error` is the machine-readable key clients check for error detection; `message` is human-readable. Both currently hold the same string, but separating them allows future divergence (e.g., error codes). | | D5 | **`suggestions` and `warnings` unwrapped** | The old `{ suggestions: string[] }` wrapper added a nesting level with no semantic value. Unwrapping aligns with `sources` and `attachments` which were already sent as bare arrays. | | D6 | **`clear` uses `Record`** | Signals "empty object, no fields" at the type level. `{}` in TypeScript means "any non-nullish value" which is too permissive. | | D7 | **`progress` renamed from `thinking`** | `thinking` was ambiguous -- it could mean extended thinking (reasoning) or tool-call progress. `progress` is clearer and distinct from the separate `reasoning` event. | | D8 | **Weak ETag approach to versioning** | The stream protocol has no version negotiation. Breaking changes are deployed atomically (server + client in the same monorepo). If external consumers emerge, a version header can be added. | | D9 | **`message_start` includes `model` field** | Supports variable model selection. The client needs to know which model is responding to display the correct model badge and capabilities. | | D10 | **Handler errors propagate (StreamReader)** | A handler throwing is a consumer bug. Swallowing the error would hide bugs. The stream stops on error, which is the safest behavior. | | D11 | **`onParseError` is separate from `on("error")`** | Parse errors are transport issues (malformed JSON from proxy interference). `on("error")` is for application-level errors sent by the server. Different concerns, different handlers. | | D12 | **Single file (`stream.ts`) not split by concern** | The module is ~250 lines total. Splitting into `stream-writer.ts`, `stream-reader.ts`, `stream-types.ts` would add import boilerplate for no readability gain at this size. | ## Dependencies | Dependency | Package | Purpose | | --------------------------------- | ----------------------------- | ------------------------------------------------------------------------------------------ | | `@coda/core-api` types | `api/src/types.ts` | `ConversationStep`, `UsageStats`, `SourceLink`, `ExtractedAttachment`, `SelectionRequired` | | Express `res` | `@types/express` | `createSSEWritable` accepts a duck-typed response object (not the full Express type) | | Fetch API `ReadableStream` | Built-in (browser + Node 18+) | `createSSEReadable` accepts `ReadableStream` | | `TextDecoder` | Built-in (browser + Node) | SSE parser decodes `Uint8Array` chunks to strings | | `AbortSignal` / `AbortController` | Built-in (browser + Node 16+) | `StreamWriter` abort handling | No new npm dependencies are introduced. ## Testing Strategy The test suite targets 140+ test cases across four test files, organized by component. ### Test Distribution | Test File | Component | Test Cases | Focus | | ----------------------------- | ------------------- | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `stream-writer.test.ts` | `StreamWriter` | ~45 | Event serialization (16 event types), terminal methods, write guard (9 cases), abort signal handling (12 cases), message ID tracking | | `stream-reader.test.ts` | `StreamReader` | ~30 | Handler registration, `consume()` lifecycle, dispatch with type safety (16 event types), parse error handling, handler error propagation, late handler registration | | `stream-sse-writable.test.ts` | `createSSEWritable` | ~9 | SSE header verification (4 headers), frame format, special character preservation, `end()` delegation | | `stream-sse-readable.test.ts` | `createSSEReadable` | ~35 | Basic parsing, blank-line dispatch (SSE spec compliance), buffering and fast path, `indexOf` optimization, split frames, cleanup (reader lock release), edge cases | | _(existing functional tests)_ | End-to-end | varies | Regression -- event names updated, payload shapes verified | ### Key Test Categories **Serialization correctness** -- Every event type in `StreamEventMap` has at least one serialization test verifying the JSON output matches expectations. Discriminated union variants (`progress`, `reasoning`, `compression`) each have separate tests for each variant. **Write guard** -- 9 tests verify that after `end()` or `error()`, all subsequent calls are no-ops. Covers every combination: `event` after `end`, `event` after `error`, `end` after `end`, `error` after `error`, `end` after `error`, `error` after `end`, plus `closed` getter state. **AbortSignal** -- 12 tests cover: valid reasons (`timeout`, `cancelled`, `context_limit`, `max_rounds`), no reason, non-string reason, invalid string reason, abort after `end`, abort after `error`, pre-aborted signal, no signal, and message ID tracking (before/after `message_start`). **SSE parser** -- Tests cover W3C spec compliance (blank-line dispatch, multi-data concatenation, event-only lines), chunked delivery (split across 1, 2, 3+ chunks), the `indexOf` optimization with `pendingLen`, reader lock cleanup on all exit paths (normal, early break, error), and edge cases (empty stream, `\r\n` endings, data without trailing blank line). **Parse error resilience** -- StreamReader continues processing valid frames after encountering malformed JSON. Tests verify the stream does not stop, `onParseError` handlers receive the event name and raw data, and no `JSON.parse` is attempted for unregistered event types. ### Running Tests ```bash # Stream module tests only cd api && pnpm test -- --testPathPattern=stream # Full monorepo test suite pnpm test:unit # Functional tests (requires local MySQL) cd server && pnpm test:functional ``` ## Rollout Plan ### Migration Sequence The migration is structured as 12 incremental tasks, each with its own commit. Tasks 1-6 are additive (new code only). Tasks 7-11 are migration (replacing old code with new). Task 12 is verification. ```mermaid gantt title Stream Module Migration dateFormat X axisFormat %s section Phase 1: Build StreamEventMap + isStreamEndReason :t1, 0, 1 SSEWritable + createSSEWritable :t2, 1, 2 SSEReadable + createSSEReadable :t3, 2, 3 StreamWriter :t4, 3, 4 StreamReader :t5, 4, 5 Export from index.ts + remove old types :t6, 5, 6 section Phase 2: Migrate server/types/ai.ts ThinkingData removal :t7, 6, 7 stream-handler.ts to StreamWriter :t8, 7, 8 ApiClient to StreamReader :t9, 8, 9 Client hooks + stream-session :t10, 9, 10 chat-routes docs + functional tests :t11, 10, 11 section Phase 3: Verify Full monorepo build + test + grep :t12, 11, 12 ``` ### Files Changed #### Created (Phase 1) | File | Contents | | ---------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `api/src/stream.ts` | `StreamEventMap`, `StreamEndReason`, `isStreamEndReason`, `SSEWritable`, `SSEReadable`, `createSSEWritable`, `createSSEReadable`, `StreamWriter`, `StreamReader` | | `api/src/__tests__/stream.test.ts` | 140+ test cases | #### Modified (Phase 2) | File | Change | | ------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | | `api/src/index.ts` | Add stream module re-exports; remove `ThinkingData`, `StreamDoneEvent` exports | | `api/src/types.ts` | Remove `ThinkingData`, `StreamDoneEvent` type definitions | | `api/src/client.ts` | Replace `SSE_HANDLERS`, `readSSEStream`, `dispatchSSEEvent` with `consumeStream` using `StreamReader`. Rename callbacks in `StreamQueryOptions`. | | `api/src/__tests__/client.test.ts` | Update SSE frames to named events; update callback names | | `server/src/routes/sse-utils.ts` | Strip to `ERROR_MESSAGES` only (remove `SSE_HEADERS`, `sendSSEData`, `sendSSEEvent`) | | `server/src/routes/stream-handler.ts` | Replace `sendSSEData`/`sendSSEEvent` with `writer.event()`/`writer.end()`/`writer.error()` | | `server/src/routes/chat-routes.ts` | Update SSE protocol docs in module JSDoc | | `server/src/types/ai.ts` | Replace `ThinkingData` import with `StreamEventMap["progress"]` | | `client/src/hooks/use-coda-orchestrator.ts` | Rename callbacks: `onThinking` -> `onProgress`, `onClearPartial` -> `onClear`; update `onMessageStart`/`onDone` to object args | | `client/src/lib/stream-session.ts` | Rename `handleStreamThinking` -> `handleStreamProgress`; update parameter types | | `server/src/__tests__/functional/helpers/sse-parser.ts` | Update event names (`thinking` -> `progress`, `clear_partial` -> `clear`) | #### Deleted (Phase 2) | File | Replaced By | | ----------------------------------------------- | ------------------------------------------------------------------------------ | | `server/src/routes/__tests__/sse-utils.test.ts` | `api/src/__tests__/stream.test.ts` (SSE writing tests moved to shared package) | #### Unchanged | File | Reason | | ------------------------------- | ---------------------------------------------------------------------------------------------------------- | | `server/src/ai/orchestrator.ts` | `converseWithTools` callbacks use `ConverseWithToolsOptions`, not `StreamQueryOptions` -- no rename needed | ### Callback Rename Summary | Old (`StreamQueryOptions`) | New (`StreamQueryOptions`) | Signature Change | | --------------------------------------------------- | -------------------------------------------------------------- | -------------------------------------------- | | `onThinking` | `onProgress` | `(data: StreamEventMap["progress"]) => void` | | `onClearPartial` | `onClear` | `() => void` (no change) | | `onMessageStart(assistantMessageId, userMessageId)` | `onMessageStart({ assistantMessageId, userMessageId, model })` | Positional args -> object with `model` | | `onDone(messageId, userMessageId)` | `onDone({ messageId, userMessageId, reason })` | Positional args -> object with `reason` | | `onSuggestions({ suggestions })` | `onSuggestions(suggestions)` | Unwrapped | | `onWarnings({ warnings })` | `onWarnings(warnings)` | Unwrapped | ### Verification Checklist (Task 12) ```bash # 1. Full monorepo build pnpm build # 2. Full monorepo typecheck pnpm typecheck # 3. Full unit test suite pnpm test:unit # 4. Verify no remaining references to deleted code grep -r "sendSSEData\|sendSSEEvent\|SSE_HANDLERS\|readSSEStream\|dispatchSSEEvent\|sse-utils" \ server/src/ api/src/ --include="*.ts" | grep -v node_modules | grep -v __tests__/stream # Expected: no results ``` ## Open Questions | # | Question | Context | | --- | -------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Q1 | Should `StreamReader` support reconnection? | Browser `EventSource` auto-reconnects with `Last-Event-ID`. The fetch-based reader does not. If connection reliability becomes an issue, a reconnection wrapper could be added around `createSSEReadable`. | | Q2 | Should `compression` events be included now or deferred? | The event type is defined in `StreamEventMap` but the server does not emit it yet. Including it now is forward-compatible but untested in production. | | Q3 | Should we add a protocol version header? | Currently no version negotiation. The monorepo deploys server + client atomically, so version mismatch is impossible. If external consumers (mobile app, third-party integrations) emerge, a `X-Stream-Protocol-Version` header should be added. | | Q4 | Should heartbeat/keepalive be added in a follow-up? | Long tool executions (30s+) can trigger proxy idle timeouts. A periodic comment frame (`: heartbeat\n\n`) would prevent this. Deferred as a non-goal for this iteration. |