From 6ebc035e6175b4052a3a39ee649e022cd1d00abc Mon Sep 17 00:00:00 2001 From: Nickk'z Date: Tue, 10 Nov 2026 15:11:27 +0530 Subject: [PATCH] Share one server-sent events helper across the four streams Closes #325 --- backend/AGENTS.md | 1 + backend/src/chat/routes.ts | 146 +++--- backend/src/runs/changes-route.ts | 58 +-- backend/src/runs/routes.ts | 734 +++++++++++++----------------- backend/src/util/sse.test.ts | 193 ++++++++ backend/src/util/sse.ts | 127 ++++++ backend/test/ratchet.test.ts | 2 +- 7 files changed, 705 insertions(+), 556 deletions(-) create mode 100644 backend/src/util/sse.test.ts create mode 100644 backend/src/util/sse.ts diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 0905db781..a9d5d2c7b 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -42,6 +42,7 @@ below are relative to `backend/`. | Untrusted text in prompts | `src/bots/prompt-safe.ts` `promptSafeJson` | | Secrets in text | `src/secrets/redact.ts` | | Messaging channels | `src/connectors/types.ts` (`Transport`, `Renderer`), `src/connectors/email/` as the worked example | +| An event stream to the browser | `src/util/sse.ts` `sseResponse`: headers, opening comment, heartbeat and teardown on disconnect; the route only writes frames | The guard tests: `test/ratchet.test.ts` (an 800-line file cap with shrink-only baselines, the dependency law, the product-name ratchet) and diff --git a/backend/src/chat/routes.ts b/backend/src/chat/routes.ts index 1569c618a..be50b746a 100644 --- a/backend/src/chat/routes.ts +++ b/backend/src/chat/routes.ts @@ -5,6 +5,7 @@ import { isMemoryScope, type MemoryScope } from "../memory/scope"; import { providerCredentialMissingMessage } from "../engines/provider-credential-gate"; import { credentialWaitSignal, resolveChatProviderCredential } from "../provider-gateway/credentials"; import { awaitWithSignal } from "../util/abortable-operation"; +import { sseResponse } from "../util/sse"; import { buildResourceAccessSnapshot, formatResourceAccessContext, @@ -66,9 +67,9 @@ chatRoutes.get("/models", async (c) => { // Emits `event: context` (citations) once, then a burst of `event: delta` text // tokens, then `event: done`. A failure surfaces as `event: error`. NO sandbox. // -// Built as a raw ReadableStream (not hono streamSSE) so we own every header - -// `no-transform` + `X-Accel-Buffering: no` stop proxies buffering the stream -// (the same SSE-hygiene the runs `/events` route relies on). +// Served through `sseResponse` (util/sse.ts), the same stream helper the runs +// `/events` route relies on for headers, heartbeat and teardown when the client +// aborts, cancels the body or a write to it fails. chatRoutes.post("/", async (c) => { if (!chatLlmEnabled()) return c.json({ error: "chat is turned off" }, 503); let body: { messages?: unknown; model?: unknown; memoryScope?: unknown }; @@ -127,101 +128,58 @@ chatRoutes.post("/", async (c) => { const query = [...messages].reverse().find((m) => m.role === "user")?.content ?? ""; const threadId = `chat:${orgId}`; - const encoder = new TextEncoder(); const signal = c.req.raw.signal; - const stream = new ReadableStream({ - start(controller) { - let closed = false; - const send = (frame: string): void => { - if (closed) return; - try { - controller.enqueue(encoder.encode(frame)); - } catch { - /* controller already closed (client gone) */ + return sseResponse({ signal }, (stream) => { + void (async () => { + try { + // Read-only retrieval first (best-effort, never throws) so the UI can + // show honest Sources before the answer streams. + const [context, resourceSnapshot] = await Promise.all([ + retrieveChatContext({ orgId, userId, query, memoryScope, threadId }), + userId + ? buildResourceAccessSnapshot( + { + orgId, + userId, + runId: threadId, + resources: [], + repos: [], + }, + undefined, + { inlineLimit: 500, exactInventoryTool: null }, + ) + : Promise.resolve(null), + ]); + if (stream.closed) return; + stream.sendEvent("context", { citations: context.citations }); + + const system = [ + CHAT_SYSTEM_PROMPT, + resourceSnapshot ? formatResourceAccessContext(resourceSnapshot) : "", + context.block, + ].filter(Boolean).join("\n\n"); + const llmMessages: ChatMessage[] = [{ role: "system", content: system }, ...messages]; + let answer = ""; + for await (const delta of chatTurnStream({ model, orgId, userId }, llmMessages, resolved, signal)) { + if (stream.closed) return; + answer += delta; + stream.sendEvent("delta", { delta }); } - }; - const sendEvent = (event: string, data: unknown): void => - send(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); - - // Prime headers/first bytes, then heartbeat idle streams. - send(": open\n\n"); - const heartbeat = setInterval(() => send(": ping\n\n"), 25_000); - heartbeat.unref?.(); - - const cleanup = (): void => { - if (closed) return; - closed = true; - clearInterval(heartbeat); - signal.removeEventListener("abort", cleanup); - try { - controller.close(); - } catch { - /* already closed */ + if (!stream.closed) { + stream.sendEvent("done", {}); + // Governed capture parity (item 7): a COMPLETED exchange (never an + // aborted stream) enqueues through the same outbox + salience gate + // as runs, marked with the chat origin. Best-effort by contract — + // captureChatExchange never throws into the stream. + void captureChatExchange({ orgId, userId, memoryScope, prompt: query, summary: answer, model }); } - }; - if (signal.aborted) return cleanup(); - signal.addEventListener("abort", cleanup); - - void (async () => { - try { - // Read-only retrieval first (best-effort, never throws) so the UI can - // show honest Sources before the answer streams. - const [context, resourceSnapshot] = await Promise.all([ - retrieveChatContext({ orgId, userId, query, memoryScope, threadId }), - userId - ? buildResourceAccessSnapshot( - { - orgId, - userId, - runId: threadId, - resources: [], - repos: [], - }, - undefined, - { inlineLimit: 500, exactInventoryTool: null }, - ) - : Promise.resolve(null), - ]); - if (closed) return; - sendEvent("context", { citations: context.citations }); - - const system = [ - CHAT_SYSTEM_PROMPT, - resourceSnapshot ? formatResourceAccessContext(resourceSnapshot) : "", - context.block, - ].filter(Boolean).join("\n\n"); - const llmMessages: ChatMessage[] = [{ role: "system", content: system }, ...messages]; - let answer = ""; - for await (const delta of chatTurnStream({ model, orgId, userId }, llmMessages, resolved, signal)) { - if (closed) return; - answer += delta; - sendEvent("delta", { delta }); - } - if (!closed) { - sendEvent("done", {}); - // Governed capture parity (item 7): a COMPLETED exchange (never an - // aborted stream) enqueues through the same outbox + salience gate - // as runs, marked with the chat origin. Best-effort by contract — - // captureChatExchange never throws into the stream. - void captureChatExchange({ orgId, userId, memoryScope, prompt: query, summary: answer, model }); - } - } catch (error) { - if (!closed) sendEvent("error", { error: chatFailure(error).reason }); - } finally { - cleanup(); - } - })(); - }, - }); - - return new Response(stream, { - headers: { - "Content-Type": "text/event-stream; charset=utf-8", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, + } catch (error) { + if (!stream.closed) stream.sendEvent("error", { error: chatFailure(error).reason }); + } finally { + stream.close(); + } + })(); }); }); diff --git a/backend/src/runs/changes-route.ts b/backend/src/runs/changes-route.ts index ab4b32e8c..a0468aa68 100644 --- a/backend/src/runs/changes-route.ts +++ b/backend/src/runs/changes-route.ts @@ -1,5 +1,6 @@ import type { Hono } from "hono"; import type { AppEnv } from "../http"; +import { sseResponse } from "../util/sse"; import { clientOrgChangeForUser, subscribeOrg } from "./org-signals"; /** GET /api/runs/changes - the tenant-scoped invalidation stream. */ @@ -11,54 +12,13 @@ export function registerRunChangesRoute(routes: Hono): void { routes.get("/changes", (c) => { const orgId = c.get("orgId"); const userId = c.get("userId"); - const encoder = new TextEncoder(); - const signal = c.req.raw.signal; - - const body = new ReadableStream({ - start(controller) { - let closed = false; - const send = (frame: string): void => { - if (closed) return; - try { - controller.enqueue(encoder.encode(frame)); - } catch { - cleanup(); - } - }; - const unsubscribe = subscribeOrg(orgId, (change) => { - const clientChange = clientOrgChangeForUser(change, userId); - if (!clientChange) return; - send(`event: change\ndata: ${JSON.stringify(clientChange)}\n\n`); - }); - const heartbeat = setInterval(() => send(": ping\n\n"), 25_000); - heartbeat.unref?.(); - - function cleanup(): void { - if (closed) return; - closed = true; - clearInterval(heartbeat); - unsubscribe(); - signal.removeEventListener("abort", cleanup); - try { - controller.close(); - } catch { - // The browser may already have closed the stream. - } - } - - send(": open\nretry: 1500\n\n"); - if (signal.aborted) cleanup(); - else signal.addEventListener("abort", cleanup); - }, - }); - - return new Response(body, { - headers: { - "Content-Type": "text/event-stream; charset=utf-8", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, - }); + // `retry` has the browser reconnect quickly after a drop; the subscription + // is the only thing to undo when it leaves. + return sseResponse({ signal: c.req.raw.signal, retryMs: 1500 }, (stream) => + subscribeOrg(orgId, (change) => { + const clientChange = clientOrgChangeForUser(change, userId); + if (clientChange) stream.sendEvent("change", clientChange); + }), + ); }); } diff --git a/backend/src/runs/routes.ts b/backend/src/runs/routes.ts index b966dd79f..52c6b0803 100644 --- a/backend/src/runs/routes.ts +++ b/backend/src/runs/routes.ts @@ -51,6 +51,7 @@ import { import { bus, channel, pumpThread, type BusEvent } from "../worker"; import { turnStream, type DeltaKind } from "./turn-stream"; import { assertNever } from "../util/exhaustive"; +import { sseResponse } from "../util/sse"; import { getNativeFramesSince, subscribeNative, type NativeFrame } from "./native-events"; import { parseResumeCursor, resolveNativeResume, resolveResumeCursor, resumeFramePayload } from "./thread-resume"; import { @@ -496,13 +497,9 @@ registerProviderSessionRoutes(runsRoutes); // SSE trace stream: replay existing steps, then live-push new ones. // -// Built as a raw ReadableStream (not Hono's streamSSE) so we own every response -// header — this is the QM port's key SSE-hygiene requirement: `no-transform` + -// `X-Accel-Buffering: no` to stop proxies buffering/transforming frames. Hono's -// streamSSE hardcodes `Cache-Control: no-cache` and re-merges it over any -// override, so it can't express `no-transform`. Speed comes from delta -// publishing + these anti-buffering headers + immediate enqueue (no compression -// middleware wraps this app — only cors — so nothing buffers the body). +// Served through `sseResponse` (util/sse.ts): anti-buffering headers, opening +// comment, heartbeat, and teardown when the client aborts, cancels the body or +// a write fails. Speed comes from delta publishing + immediate enqueue. runsRoutes.get("/:id/events", async (c) => { const id = c.req.param("id"); // Authorize by org first — a cross-org (or missing) id is a 404, never a stream. @@ -517,8 +514,6 @@ runsRoutes.get("/:id/events", async (c) => { | BusEvent | { type: "delta"; delta: string; kind?: DeltaKind } | { type: "native"; frame: NativeFrame }; - const encoder = new TextEncoder(); - const signal = c.req.raw.signal; // Native-frame lane cursor: replay lossless native events with seq strictly // greater than `?cursor=` (default -1 → from the start; seq begins at 0). A @@ -528,155 +523,113 @@ runsRoutes.get("/:id/events", async (c) => { ? Number(cursorRaw) : -1; - const body = new ReadableStream({ - start(controller) { - // idx → content fingerprint of the LAST version sent. Updates (same idx, - // enriched code_json) must pass; only true duplicates are suppressed. - const emitted = new Map(); - // Native-frame dedupe: eventId → the highest seq already sent. A revision - // (same eventId, higher seq) passes; a replay/live overlap is suppressed. - const nativeSeen = new Map(); - const queue: OutEvent[] = []; - let wake: (() => void) | null = null; - let closed = false; - - const send = (frame: string): void => { - if (closed) return; - try { - controller.enqueue(encoder.encode(frame)); - } catch { - /* controller already closed (client gone) */ - } - }; - const sendEvent = (event: string, data: unknown): void => - send(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + return sseResponse({ signal: c.req.raw.signal }, (stream) => { + // idx → content fingerprint of the LAST version sent. Updates (same idx, + // enriched code_json) must pass; only true duplicates are suppressed. + const emitted = new Map(); + // Native-frame dedupe: eventId → the highest seq already sent. A revision + // (same eventId, higher seq) passes; a replay/live overlap is suppressed. + const nativeSeen = new Map(); + const queue: OutEvent[] = []; + let wake: (() => void) | null = null; + + const wakeUp = (): void => { + if (wake) { + wake(); + wake = null; + } + }; + const push = (ev: OutEvent): void => { + queue.push(ev); + wakeUp(); + }; + bus.on(channel(id), push); + // Live-typing narration: each delta an engine publishes is forwarded as a + // distinct `event: delta` frame. Old clients (which only listen for `step` + // / `done`) ignore it — contract-compatible. + const unsubscribeDeltas = turnStream.subscribe(id, (delta, kind) => + push({ type: "delta", delta, kind }), + ); + // Live native frames (lossless capture projection). Old clients ignore the + // `event: native` type — additive, contract-compatible. + const unsubscribeNative = subscribeNative(id, (frame) => + push({ type: "native", frame }), + ); - const wakeUp = (): void => { - if (wake) { - wake(); - wake = null; - } - }; - const push = (ev: OutEvent): void => { - queue.push(ev); - wakeUp(); - }; - bus.on(channel(id), push); - // Live-typing narration: each delta an engine publishes is forwarded as a - // distinct `event: delta` frame. Old clients (which only listen for `step` - // / `done`) ignore it — contract-compatible. - const unsubscribeDeltas = turnStream.subscribe(id, (delta, kind) => - push({ type: "delta", delta, kind }), - ); - // Live native frames (lossless capture projection). Old clients ignore the - // `event: native` type — additive, contract-compatible. - const unsubscribeNative = subscribeNative(id, (frame) => - push({ type: "native", frame }), - ); + // Emit a native frame if it advances its eventId's seq (dedupe overlap). + const sendNative = (frame: NativeFrame): void => { + if ((nativeSeen.get(frame.eventId) ?? -1) >= frame.seq) return; + nativeSeen.set(frame.eventId, frame.seq); + stream.sendEvent("native", frame); + }; - // Emit a native frame if it advances its eventId's seq (dedupe overlap). - const sendNative = (frame: NativeFrame): void => { - if ((nativeSeen.get(frame.eventId) ?? -1) >= frame.seq) return; - nativeSeen.set(frame.eventId, frame.seq); - sendEvent("native", frame); - }; + // Drive replay + the live loop OUTSIDE start() so the response streams + // immediately (start resolving isn't gated on the whole run finishing). + void (async () => { + // Replay everything already persisted (subscribed first, so no gap). + for (const step of await getStepsApi(id)) { + if (stream.closed) return; + emitted.set(step.idx, `${step.id}|${step.code_json ?? ""}`); + stream.sendEvent("step", step); + } - // Prime the stream so headers/first bytes flush immediately. - send(": open\n\n"); - - // Heartbeat comment keeps proxies/browsers from dropping an idle stream. - // unref so it never keeps the process alive on its own. - const heartbeat = setInterval(() => send(": ping\n\n"), 25_000); - heartbeat.unref?.(); - - const cleanup = (): void => { - if (closed) return; - closed = true; - clearInterval(heartbeat); - bus.off(channel(id), push); - unsubscribeDeltas(); - unsubscribeNative(); - signal.removeEventListener("abort", cleanup); - wakeUp(); - try { - controller.close(); - } catch { - /* already closed */ - } - }; - if (signal.aborted) return cleanup(); - signal.addEventListener("abort", cleanup); - - // Drive replay + the live loop OUTSIDE start() so the response streams - // immediately (start resolving isn't gated on the whole run finishing). - void (async () => { - // Replay everything already persisted (subscribed first, so no gap). - for (const step of await getStepsApi(id)) { - if (closed) return; - emitted.set(step.idx, `${step.id}|${step.code_json ?? ""}`); - sendEvent("step", step); - } + // Replay the native lane from the client's cursor (ordered by seq; + // already deduped by native id in the store). Subscribed before this, so + // a frame persisted during replay arrives on the live queue and the + // seq-dedupe suppresses the overlap. + for (const frame of await getNativeFramesSince(id, cursorSeq)) { + if (stream.closed) return; + sendNative(frame); + } - // Replay the native lane from the client's cursor (ordered by seq; - // already deduped by native id in the store). Subscribed before this, so - // a frame persisted during replay arrives on the live queue and the - // seq-dedupe suppresses the overlap. - for (const frame of await getNativeFramesSince(id, cursorSeq)) { - if (closed) return; - sendNative(frame); - } + // If the run already finished, close immediately after replay. + const snapshot = await getRun(id); + if (snapshot && (snapshot.status === "completed" || snapshot.status === "failed")) { + stream.sendEvent("done", { id, status: snapshot.status }); + return stream.close(); + } - // If the run already finished, close immediately after replay. - const snapshot = await getRun(id); - if (snapshot && (snapshot.status === "completed" || snapshot.status === "failed")) { - sendEvent("done", { id, status: snapshot.status }); - return cleanup(); + // Live loop. + while (!stream.closed) { + if (queue.length === 0) { + await new Promise((resolve) => { + wake = resolve; + }); } - - // Live loop. - while (!closed) { - if (queue.length === 0) { - await new Promise((resolve) => { - wake = resolve; - }); - } - while (queue.length > 0 && !closed) { - const ev = queue.shift()!; - switch (ev.type) { - case "step": { - // Same idx may arrive again with enriched code_json (tool output - // attached in place) — forward it; skip only true duplicates. - const fp = `${ev.step.id}|${ev.step.code_json ?? ""}`; - if (emitted.get(ev.step.idx) === fp) continue; - emitted.set(ev.step.idx, fp); - sendEvent("step", ev.step); - continue; - } - case "delta": - sendEvent("delta", { delta: ev.delta, kind: ev.kind }); - continue; - case "native": - sendNative(ev.frame); - continue; - case "end": - sendEvent("done", { id, status: ev.status }); - return cleanup(); - default: - assertNever(ev); + while (queue.length > 0 && !stream.closed) { + const ev = queue.shift()!; + switch (ev.type) { + case "step": { + // Same idx may arrive again with enriched code_json (tool output + // attached in place) — forward it; skip only true duplicates. + const fp = `${ev.step.id}|${ev.step.code_json ?? ""}`; + if (emitted.get(ev.step.idx) === fp) continue; + emitted.set(ev.step.idx, fp); + stream.sendEvent("step", ev.step); + continue; } + case "delta": + stream.sendEvent("delta", { delta: ev.delta, kind: ev.kind }); + continue; + case "native": + sendNative(ev.frame); + continue; + case "end": + stream.sendEvent("done", { id, status: ev.status }); + return stream.close(); + default: + assertNever(ev); } } - })(); - }, - }); + } + })(); - return new Response(body, { - headers: { - "Content-Type": "text/event-stream; charset=utf-8", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, + return () => { + bus.off(channel(id), push); + unsubscribeDeltas(); + unsubscribeNative(); + wakeUp(); + }; }); }); @@ -716,9 +669,6 @@ runsRoutes.get("/:rootRunId/thread-events", async (c) => { const threadId = rootRun.threadId; const requested = parseResumeCursor(c.req.query("canonicalAfter"), c.req.query("canonicalId"), c.req.query("epoch"), c.req.queries("nativeAfter")); - const encoder = new TextEncoder(); - const signal = c.req.raw.signal; - type ThreadOut = | { type: "signal"; runId: string } | { type: "step"; runId: string; step: ApiStep } @@ -728,282 +678,242 @@ runsRoutes.get("/:rootRunId/thread-events", async (c) => { | { type: "canonical"; event: DeliveredCanonicalEvent } | { type: "canonical-complete"; complete: CanonicalizationComplete }; - const body = new ReadableStream({ - start(controller) { - const queue: ThreadOut[] = []; - let wake: (() => void) | null = null; - let closed = false; - let overflowed = false; - - // Per-run dedupe state (keyed by runId): step idx→content fingerprint, and - // native eventId→highest seq. Mirrors the per-run route's dedupe, one map - // per run so replay/live overlap collapses without cross-run interference. - const emittedByRun = new Map>(); - const nativeSeenByRun = new Map>(); - // Per-run live-source listeners, torn down together on disconnect. - const attached = new Set(); - const perRunCleanups = new Map void>(); - - const send = (frame: string): void => { - if (closed) return; - try { - controller.enqueue(encoder.encode(frame)); - } catch { - /* controller already closed (client gone) */ - } - }; - const sendFrame = (event: string, data: unknown): void => - send(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + return sseResponse({ signal: c.req.raw.signal }, (stream) => { + const queue: ThreadOut[] = []; + let wake: (() => void) | null = null; + let overflowed = false; + + // Per-run dedupe state (keyed by runId): step idx→content fingerprint, and + // native eventId→highest seq. Mirrors the per-run route's dedupe, one map + // per run so replay/live overlap collapses without cross-run interference. + const emittedByRun = new Map>(); + const nativeSeenByRun = new Map>(); + // Per-run live-source listeners, torn down together on disconnect. + const attached = new Set(); + const perRunCleanups = new Map void>(); + + const wakeUp = (): void => { + if (wake) { + wake(); + wake = null; + } + }; + const push = (ev: ThreadOut): void => { + if (stream.closed) return; + if (queue.length >= MAX_QUEUE) { + // Bound memory: drop the connection so the browser reconnects to a fresh + // authoritative snapshot instead of buffering without limit. + overflowed = true; + stream.close(); + return; + } + queue.push(ev); + wakeUp(); + }; - const wakeUp = (): void => { - if (wake) { - wake(); - wake = null; - } - }; - const push = (ev: ThreadOut): void => { - if (closed) return; - if (queue.length >= MAX_QUEUE) { - // Bound memory: drop the connection so the browser reconnects to a fresh - // authoritative snapshot instead of buffering without limit. - overflowed = true; - cleanup(); - return; - } - queue.push(ev); - wakeUp(); + // Attach the EXISTING per-run live sources for one run (idempotent). The + // worker step/end bus, the transient delta stream, and the native bus are all + // relayed with the run's id; a settled run stays attached (late native + // revisions still land) until the browser disconnects. + const attachRun = (runId: string): void => { + if (attached.has(runId)) return; + attached.add(runId); + const onBus = (ev: BusEvent): void => { + if (ev.type === "step") push({ type: "step", runId, step: ev.step }); + else push({ type: "end", runId, status: ev.status }); }; + bus.on(channel(runId), onBus); + const offDelta = turnStream.subscribe(runId, (delta, kind) => + push({ type: "delta", runId, delta, kind }), + ); + const offNative = subscribeNative(runId, (frame) => + push({ type: "native", runId, frame }), + ); + perRunCleanups.set(runId, () => { + bus.off(channel(runId), onBus); + offDelta(); + offNative(); + }); + }; - // Attach the EXISTING per-run live sources for one run (idempotent). The - // worker step/end bus, the transient delta stream, and the native bus are all - // relayed with the run's id; a settled run stays attached (late native - // revisions still land) until the browser disconnects. - const attachRun = (runId: string): void => { - if (attached.has(runId)) return; - attached.add(runId); - const onBus = (ev: BusEvent): void => { - if (ev.type === "step") push({ type: "step", runId, step: ev.step }); - else push({ type: "end", runId, status: ev.status }); - }; - bus.on(channel(runId), onBus); - const offDelta = turnStream.subscribe(runId, (delta, kind) => - push({ type: "delta", runId, delta, kind }), - ); - const offNative = subscribeNative(runId, (frame) => - push({ type: "native", runId, frame }), - ); - perRunCleanups.set(runId, () => { - bus.off(channel(runId), onBus); - offDelta(); - offNative(); - }); - }; + // Emit a step if its (idx, fingerprint) is new or enriched (same idx with new + // code_json passes; a pure duplicate is suppressed). + const sendStep = (runId: string, step: ApiStep): void => { + let m = emittedByRun.get(runId); + if (!m) { + m = new Map(); + emittedByRun.set(runId, m); + } + const fp = `${step.id}|${step.code_json ?? ""}`; + if (m.get(step.idx) === fp) return; + m.set(step.idx, fp); + stream.sendEvent("step", { threadId, runId, step }); + }; + const seedStepDedupe = (runId: string, steps: readonly ApiStep[]): void => { + let m = emittedByRun.get(runId); + if (!m) { + m = new Map(); + emittedByRun.set(runId, m); + } + for (const s of steps) m.set(s.idx, `${s.id}|${s.code_json ?? ""}`); + }; - // Emit a step if its (idx, fingerprint) is new or enriched (same idx with new - // code_json passes; a pure duplicate is suppressed). - const sendStep = (runId: string, step: ApiStep): void => { - let m = emittedByRun.get(runId); - if (!m) { - m = new Map(); - emittedByRun.set(runId, m); - } - const fp = `${step.id}|${step.code_json ?? ""}`; - if (m.get(step.idx) === fp) return; - m.set(step.idx, fp); - sendFrame("step", { threadId, runId, step }); - }; - const seedStepDedupe = (runId: string, steps: readonly ApiStep[]): void => { - let m = emittedByRun.get(runId); - if (!m) { - m = new Map(); - emittedByRun.set(runId, m); - } - for (const s of steps) m.set(s.idx, `${s.id}|${s.code_json ?? ""}`); - }; + const nativeSeen = (runId: string): Map => + nativeSeenByRun.get(runId) ?? nativeSeenByRun.set(runId, new Map()).get(runId)!; + // Emit a native frame if it advances its eventId's seq (dedupe replay/live). + const sendNative = (runId: string, frame: NativeFrame): void => { + const m = nativeSeen(runId); + if ((m.get(frame.eventId) ?? -1) >= frame.seq) return; + m.set(frame.eventId, frame.seq); + stream.sendEvent("native", { threadId, runId, frame }); + }; - const nativeSeen = (runId: string): Map => - nativeSeenByRun.get(runId) ?? nativeSeenByRun.set(runId, new Map()).get(runId)!; - // Emit a native frame if it advances its eventId's seq (dedupe replay/live). - const sendNative = (runId: string, frame: NativeFrame): void => { - const m = nativeSeen(runId); - if ((m.get(frame.eventId) ?? -1) >= frame.seq) return; - m.set(frame.eventId, frame.seq); - sendFrame("native", { threadId, runId, frame }); - }; + // Canonical lane: the provider-neutral events, deduped + // by the IMMUTABLE thread delivery cursor - replay + live never re-send a cursor. + // Thread-scoped (threadId was authorized at route entry), so no per-run attach. + let canonicalCursor = 0; + const sendCanonical = (event: DeliveredCanonicalEvent): void => { + if (event.deliverySeq <= canonicalCursor) return; + canonicalCursor = event.deliverySeq; + stream.sendEvent("canonical", { threadId, event }); + }; + // Canonicalization-complete (H2): the per-run signal that its canonical projection + // is trustworthy. Deduped per run so replay + live never re-announce a run. + const canonicalCompleteSeen = new Map(); // runId -> degraded (see admitCanonicalComplete) + const sendCanonicalComplete = (complete: CanonicalizationComplete): void => { + if (!admitCanonicalComplete(canonicalCompleteSeen, complete)) return; + stream.sendEvent("canonical-complete", { threadId, complete }); + }; - // Canonical lane: the provider-neutral events, deduped - // by the IMMUTABLE thread delivery cursor - replay + live never re-send a cursor. - // Thread-scoped (threadId was authorized at route entry), so no per-run attach. - let canonicalCursor = 0; - const sendCanonical = (event: DeliveredCanonicalEvent): void => { - if (event.deliverySeq <= canonicalCursor) return; - canonicalCursor = event.deliverySeq; - sendFrame("canonical", { threadId, event }); - }; - // Canonicalization-complete (H2): the per-run signal that its canonical projection - // is trustworthy. Deduped per run so replay + live never re-announce a run. - const canonicalCompleteSeen = new Map(); // runId -> degraded (see admitCanonicalComplete) - const sendCanonicalComplete = (complete: CanonicalizationComplete): void => { - if (!admitCanonicalComplete(canonicalCompleteSeen, complete)) return; - sendFrame("canonical-complete", { threadId, complete }); - }; + // Subscribe to the thread-change signal BEFORE loading the snapshot, so a run + // accepted mid-load is discovered (no window where a new run is missed). The + // handler only enqueues; the async drain re-reads durable state. + const unsubscribeThread = subscribeThread(threadId, (change) => + push({ type: "signal", runId: change.runId }), + ); + // Live canonical events for the whole thread (all runs, incl. later ones). + const unsubscribeCanonical = subscribeCanonicalThread(threadId, (event) => + push({ type: "canonical", event }), + ); + // Live canonicalization-complete signals (one per run as its outbox reaches complete). + const unsubscribeCanonicalComplete = subscribeCanonicalizationComplete(threadId, (complete) => + push({ type: "canonical-complete", complete }), + ); - // Prime headers/first bytes. - send(": open\n\n"); - const heartbeat = setInterval(() => send(": ping\n\n"), 25_000); - heartbeat.unref?.(); - - function cleanup(): void { - if (closed) return; - closed = true; - clearInterval(heartbeat); - unsubscribeThread(); - unsubscribeCanonical(); - unsubscribeCanonicalComplete(); - for (const off of perRunCleanups.values()) off(); - perRunCleanups.clear(); - attached.clear(); - signal.removeEventListener("abort", cleanup); - wakeUp(); - try { - controller.close(); - } catch { - /* already closed */ - } + // Reload one run's durable projection and emit it as a `run` frame, ensuring + // its live sources are attached and its native frames replayed. AUTHORIZE + + // verify thread membership BEFORE attaching any live listeners: a signal that + // ever carried a runId from a different thread/org (a future/misrouted + // publisher, or a bug) must never subscribe this connection to that run's + // channels. Attaching first left a fail-closed hole - the durable `run` frame + // was withheld but the live step/delta/native listeners stayed bound, leaking + // frames cross-thread/cross-org (Codex review finding 1). Org-scoped read fails + // closed; a run outside this thread is ignored without ever attaching. + const projectRun = async (runId: string): Promise => { + const run = await getRunWithSteps(orgId, runId); + if (!run || run.thread_id !== threadId) return; + attachRun(runId); + seedStepDedupe(runId, run.steps); + stream.sendEvent("run", { threadId, run }); + for (const frame of await getNativeFramesSince(runId, -1)) { + if (stream.closed) return; + sendNative(runId, frame); } + }; - // Subscribe to the thread-change signal BEFORE loading the snapshot, so a run - // accepted mid-load is discovered (no window where a new run is missed). The - // handler only enqueues; the async drain re-reads durable state. - const unsubscribeThread = subscribeThread(threadId, (change) => - push({ type: "signal", runId: change.runId }), - ); - // Live canonical events for the whole thread (all runs, incl. later ones). - const unsubscribeCanonical = subscribeCanonicalThread(threadId, (event) => - push({ type: "canonical", event }), - ); - // Live canonicalization-complete signals (one per run as its outbox reaches complete). - const unsubscribeCanonicalComplete = subscribeCanonicalizationComplete(threadId, (complete) => - push({ type: "canonical-complete", complete }), - ); - - if (signal.aborted) return cleanup(); - signal.addEventListener("abort", cleanup); - - // Reload one run's durable projection and emit it as a `run` frame, ensuring - // its live sources are attached and its native frames replayed. AUTHORIZE + - // verify thread membership BEFORE attaching any live listeners: a signal that - // ever carried a runId from a different thread/org (a future/misrouted - // publisher, or a bug) must never subscribe this connection to that run's - // channels. Attaching first left a fail-closed hole - the durable `run` frame - // was withheld but the live step/delta/native listeners stayed bound, leaking - // frames cross-thread/cross-org (Codex review finding 1). Org-scoped read fails - // closed; a run outside this thread is ignored without ever attaching. - const projectRun = async (runId: string): Promise => { - const run = await getRunWithSteps(orgId, runId); - if (!run || run.thread_id !== threadId) return; - attachRun(runId); - seedStepDedupe(runId, run.steps); - sendFrame("run", { threadId, run }); - for (const frame of await getNativeFramesSince(runId, -1)) { - if (closed) return; - sendNative(runId, frame); + void (async () => { + // 1. Load the authoritative thread (oldest→newest), attaching every run's live + // sources FIRST so frames produced during replay queue up. + const thread = await getThreadForRun(orgId, rootRunId); + if (stream.closed) return; + if (!thread) return stream.close(); // resolved above; defensive + for (const run of thread) attachRun(run.id); + + // 2. Say whether the browser's canonical resume cursor is honoured (a refused one + // replays from zero, and the browser drops what it retained), then the snapshot. + const resume = await resolveResumeCursor(threadId, requested); + canonicalCursor = resume.canonicalAfter; + stream.sendEvent("resume", resumeFramePayload(threadId, resume)); + stream.sendEvent("snapshot", { threadId, runs: thread }); + for (const run of thread) seedStepDedupe(run.id, run.steps); + + // 3. Runs whose canonicalization is COMPLETE (H2), read BEFORE the canonical rows so a + // run finalized between the reads announces completion via the live loop. A sealed + // run the browser proved it holds (thread-resume.ts) counts as sent up to its cursor + // and replays only what is above it; every other run replays every frame. + const completes = await completeCanonicalRuns(threadId); + const native = await resolveNativeResume(requested.native, new Map(completes.map((c) => [c.runId, c.sourceFrameMax])), { epoch: requested.epoch, reset: resume.reset }); + for (const run of thread) { + for (const held of native.retained(run.id)) nativeSeen(run.id).set(held.eventId, held.seq); + for (const frame of await native.replay(run.id)) { + if (stream.closed) return; + sendNative(run.id, frame); } - }; + } - void (async () => { - // 1. Load the authoritative thread (oldest→newest), attaching every run's live - // sources FIRST so frames produced during replay queue up. - const thread = await getThreadForRun(orgId, rootRunId); - if (closed) return; - if (!thread) return cleanup(); // resolved above; defensive - for (const run of thread) attachRun(run.id); - - // 2. Say whether the browser's canonical resume cursor is honoured (a refused one - // replays from zero, and the browser drops what it retained), then the snapshot. - const resume = await resolveResumeCursor(threadId, requested); - canonicalCursor = resume.canonicalAfter; - sendFrame("resume", resumeFramePayload(threadId, resume)); - sendFrame("snapshot", { threadId, runs: thread }); - for (const run of thread) seedStepDedupe(run.id, run.steps); - - // 3. Runs whose canonicalization is COMPLETE (H2), read BEFORE the canonical rows so a - // run finalized between the reads announces completion via the live loop. A sealed - // run the browser proved it holds (thread-resume.ts) counts as sent up to its cursor - // and replays only what is above it; every other run replays every frame. - const completes = await completeCanonicalRuns(threadId); - const native = await resolveNativeResume(requested.native, new Map(completes.map((c) => [c.runId, c.sourceFrameMax])), { epoch: requested.epoch, reset: resume.reset }); - for (const run of thread) { - for (const held of native.retained(run.id)) nativeSeen(run.id).set(held.eventId, held.seq); - for (const frame of await native.replay(run.id)) { - if (closed) return; - sendNative(run.id, frame); - } - } + // 3b. Canonical rows above the cursor (deduped by deliverySeq), then the completions. + for (const event of await loadCanonicalThread(threadId, resume.canonicalAfter)) { + if (stream.closed) return; + sendCanonical(event); + } + for (const complete of completes) { + if (stream.closed) return; + sendCanonicalComplete({ ...complete, threadId }); + } - // 3b. Canonical rows above the cursor (deduped by deliverySeq), then the completions. - for (const event of await loadCanonicalThread(threadId, resume.canonicalAfter)) { - if (closed) return; - sendCanonical(event); - } - for (const complete of completes) { - if (closed) return; - sendCanonicalComplete({ ...complete, threadId }); + // 4. Live loop — never closes on a single run settling; only the browser + // disconnect (abort) or a queue overflow closes it. + while (!stream.closed) { + if (queue.length === 0) { + await new Promise((resolve) => { + wake = resolve; + }); } - - // 4. Live loop — never closes on a single run settling; only the browser - // disconnect (abort) or a queue overflow closes it. - while (!closed) { - if (queue.length === 0) { - await new Promise((resolve) => { - wake = resolve; - }); - } - while (queue.length > 0 && !closed) { - const ev = queue.shift()!; - switch (ev.type) { - case "signal": - await projectRun(ev.runId); - continue; - case "step": - sendStep(ev.runId, ev.step); - continue; - case "delta": - sendFrame("delta", { threadId, runId: ev.runId, delta: ev.delta, kind: ev.kind }); - continue; - case "native": - sendNative(ev.runId, ev.frame); - continue; - case "canonical": - sendCanonical(ev.event); - continue; - case "canonical-complete": - sendCanonicalComplete(ev.complete); - continue; - case "end": - // Settle ONE run; keep the thread connection open for queued/future - // turns. The `settled` thread signal re-emits a `run` frame with the - // final summary/status right after. - sendFrame("done", { threadId, runId: ev.runId, status: ev.status }); - continue; - default: - assertNever(ev); - } + while (queue.length > 0 && !stream.closed) { + const ev = queue.shift()!; + switch (ev.type) { + case "signal": + await projectRun(ev.runId); + continue; + case "step": + sendStep(ev.runId, ev.step); + continue; + case "delta": + stream.sendEvent("delta", { threadId, runId: ev.runId, delta: ev.delta, kind: ev.kind }); + continue; + case "native": + sendNative(ev.runId, ev.frame); + continue; + case "canonical": + sendCanonical(ev.event); + continue; + case "canonical-complete": + sendCanonicalComplete(ev.complete); + continue; + case "end": + // Settle ONE run; keep the thread connection open for queued/future + // turns. The `settled` thread signal re-emits a `run` frame with the + // final summary/status right after. + stream.sendEvent("done", { threadId, runId: ev.runId, status: ev.status }); + continue; + default: + assertNever(ev); } } - })().catch((err) => { - if (!overflowed) console.error(`[thread-events] stream ${rootRunId} failed:`, err); - cleanup(); - }); - }, - }); + } + })().catch((err) => { + if (!overflowed) console.error(`[thread-events] stream ${rootRunId} failed:`, err); + stream.close(); + }); - return new Response(body, { - headers: { - "Content-Type": "text/event-stream; charset=utf-8", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, + return () => { + unsubscribeThread(); + unsubscribeCanonical(); + unsubscribeCanonicalComplete(); + for (const off of perRunCleanups.values()) off(); + perRunCleanups.clear(); + attached.clear(); + wakeUp(); + }; }); }); diff --git a/backend/src/util/sse.test.ts b/backend/src/util/sse.test.ts new file mode 100644 index 000000000..8d7226663 --- /dev/null +++ b/backend/src/util/sse.test.ts @@ -0,0 +1,193 @@ +import { describe, expect, spyOn, test } from "bun:test"; +import { SSE_HEARTBEAT_MS, sseResponse } from "./sse"; + +/** Read raw text off the response until `until` returns true or the body ends. */ +async function readUntil(res: Response, until: (text: string) => boolean): Promise<{ text: string; done: boolean }> { + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let text = ""; + for (;;) { + if (until(text)) { + reader.releaseLock(); + return { text, done: false }; + } + const { value, done } = await reader.read(); + if (done) return { text, done: true }; + text += decoder.decode(value, { stream: true }); + } +} + +async function readAll(res: Response): Promise { + return (await readUntil(res, () => false)).text; +} + +describe("sseResponse", () => { + test("sends the four anti-buffering headers and the opening comment", async () => { + const res = sseResponse({ signal: new AbortController().signal }, (stream) => { + stream.close(); + }); + expect(res.status).toBe(200); + expect(res.headers.get("content-type")).toBe("text/event-stream; charset=utf-8"); + expect(res.headers.get("cache-control")).toBe("no-cache, no-transform"); + expect(res.headers.get("connection")).toBe("keep-alive"); + expect(res.headers.get("x-accel-buffering")).toBe("no"); + expect([...res.headers.keys()].sort()).toEqual([ + "cache-control", + "connection", + "content-type", + "x-accel-buffering", + ]); + expect(await readAll(res)).toBe(": open\n\n"); + }); + + test("puts the reconnect delay in the opening comment", async () => { + const res = sseResponse({ signal: new AbortController().signal, retryMs: 1500 }, (stream) => { + stream.close(); + }); + expect(await readAll(res)).toBe(": open\nretry: 1500\n\n"); + }); + + test("sendEvent writes event and JSON data, and send writes a frame as given", async () => { + const res = sseResponse({ signal: new AbortController().signal }, (stream) => { + stream.sendEvent("step", { idx: 0, text: "a\nb" }); + stream.send(": comment\n\n"); + stream.close(); + }); + expect(await readAll(res)).toBe( + ': open\n\nevent: step\ndata: {"idx":0,"text":"a\\nb"}\n\n: comment\n\n', + ); + }); + + test("heartbeats an idle stream and clears the timer when closed", async () => { + expect(SSE_HEARTBEAT_MS).toBe(25_000); + const started = spyOn(globalThis, "setInterval"); + const cleared = spyOn(globalThis, "clearInterval"); + try { + let close!: () => void; + const res = sseResponse({ signal: new AbortController().signal, heartbeatMs: 5 }, (stream) => { + close = stream.close; + }); + expect(started).toHaveBeenCalledTimes(1); + expect(started.mock.calls[0]?.[1]).toBe(5); + const handle = started.mock.results[0]?.value as ReturnType; + const { text } = await readUntil(res, (t) => t.split(": ping\n\n").length > 2); + expect(text.startsWith(": open\n\n: ping\n\n: ping\n\n")).toBe(true); + expect(cleared).not.toHaveBeenCalled(); + close(); + // The interval itself is cleared, not just its frames dropped. + expect(cleared).toHaveBeenCalledTimes(1); + expect(cleared.mock.calls[0]?.[0]).toBe(handle); + const rest = await readAll(res); + // Everything after the close is the tail already buffered, then the end of the body. + expect(rest.replaceAll(": ping\n\n", "")).toBe(""); + } finally { + started.mockRestore(); + cleared.mockRestore(); + } + }); + + test("a route that throws while starting leaves no timer or abort hook behind", () => { + const cleared = spyOn(globalThis, "clearInterval"); + const controller = new AbortController(); + try { + expect(() => + sseResponse({ signal: controller.signal }, () => { + throw new Error("route failed"); + }), + ).toThrow("route failed"); + expect(cleared).toHaveBeenCalledTimes(1); + } finally { + cleared.mockRestore(); + } + }); + + test("aborting the request runs the teardown once and ends the body", async () => { + const controller = new AbortController(); + let torn = 0; + let stream!: { closed: boolean; sendEvent(event: string, data: unknown): void }; + const res = sseResponse({ signal: controller.signal }, (s) => { + stream = s; + return () => { + torn += 1; + }; + }); + expect(stream.closed).toBe(false); + controller.abort(); + expect(stream.closed).toBe(true); + expect(torn).toBe(1); + controller.abort(); + stream.sendEvent("late", {}); + expect(torn).toBe(1); + const { text, done } = await readUntil(res, () => false); + expect(done).toBe(true); + expect(text).toBe(": open\n\n"); + }); + + test("a request that is already gone never starts the route", async () => { + const controller = new AbortController(); + controller.abort(); + let started = false; + const res = sseResponse({ signal: controller.signal }, () => { + started = true; + }); + expect(started).toBe(false); + expect((await readUntil(res, () => false)).done).toBe(true); + }); + + test("close is idempotent and a write after it is dropped", async () => { + let torn = 0; + const res = sseResponse({ signal: new AbortController().signal }, (stream) => { + stream.sendEvent("one", 1); + stream.close(); + stream.close(); + stream.sendEvent("two", 2); + return () => { + torn += 1; + }; + }); + expect(torn).toBe(1); + expect(await readAll(res)).toBe(": open\n\nevent: one\ndata: 1\n\n"); + }); + + test("a client cancelling the body closes the stream and runs the teardown", async () => { + let torn = 0; + let stream!: { closed: boolean }; + const res = sseResponse({ signal: new AbortController().signal }, (s) => { + stream = s; + return () => { + torn += 1; + }; + }); + await res.body!.cancel(); + expect(stream.closed).toBe(true); + expect(torn).toBe(1); + }); +}); + +describe("the stream literals live in one place", () => { + test("the heartbeat interval and the full header block appear only in util/sse.ts", async () => { + const { readdirSync, readFileSync, statSync } = await import("node:fs"); + const { join, relative } = await import("node:path"); + const root = join(import.meta.dir, ".."); + const files: string[] = []; + const walk = (dir: string): void => { + for (const name of readdirSync(dir)) { + const abs = join(dir, name); + if (statSync(abs).isDirectory()) walk(abs); + else if (abs.endsWith(".ts") && !abs.endsWith(".test.ts")) files.push(abs); + } + }; + walk(root); + const heartbeat: string[] = []; + const headers: string[] = []; + for (const file of files) { + const text = readFileSync(file, "utf8"); + if (text.includes("25_000")) heartbeat.push(relative(root, file)); + if (text.includes("text/event-stream; charset=utf-8")) headers.push(relative(root, file)); + } + // A new stream goes through sseResponse; a new 25 s wait reuses SSE_HEARTBEAT_MS + // or gets its own named constant. + expect(heartbeat).toEqual(["util/sse.ts"]); + expect(headers).toEqual(["util/sse.ts"]); + }); +}); diff --git a/backend/src/util/sse.ts b/backend/src/util/sse.ts new file mode 100644 index 000000000..c965d549a --- /dev/null +++ b/backend/src/util/sse.ts @@ -0,0 +1,127 @@ +/** The one server-sent events response behind every stream the backend serves + * (the run, thread, chat and changes streams). It owns the ReadableStream, the + * opening comment, the heartbeat, the anti-buffering headers and the teardown + * when the connection ends, so a fix here reaches every stream. A route only + * writes frames and hands back what to undo when the stream closes. */ + +/** An idle stream gets a `: ping` comment this often so proxies and browsers + * do not drop it. */ +export const SSE_HEARTBEAT_MS = 25_000; + +/** `no-transform` and `X-Accel-Buffering: no` stop proxies buffering or + * rewriting frames; the raw Response is used instead of Hono's streamSSE + * because that pins its own Cache-Control and cannot express `no-transform`. + * No compression middleware wraps the app (only cors), so nothing buffers + * the body. */ +export const SSE_HEADERS = Object.freeze({ + "Content-Type": "text/event-stream; charset=utf-8", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", +}); + +export interface SseStream { + /** True once the stream has closed; a write after that is dropped. */ + readonly closed: boolean; + /** Write one raw frame, already terminated (such as `: ping\n\n`). */ + send(frame: string): void; + /** Write `event: \ndata: \n\n`. */ + sendEvent(event: string, data: unknown): void; + /** End the stream: stop the heartbeat, run the route's teardown once, close + * the body. Safe to call more than once. */ + close(): void; +} + +export interface SseOptions { + /** The request's signal: the client leaving closes the stream. */ + signal: AbortSignal; + /** Sent in the opening comment as `retry: `, the browser's delay before + * it reconnects. */ + retryMs?: number; + /** How often the heartbeat comment goes out; tests shorten it. */ + heartbeatMs?: number; +} + +/** + * Open a stream, send the opening comment, then hand the route a writer. + * `start` runs synchronously once, never for a request that is already gone, + * and may return a teardown that runs exactly once when the stream closes: + * the client aborting, the body being cancelled, a write failing because the + * client left, or the route calling `close()`. The last three close the + * stream on every route now; before this helper only the changes stream + * closed on a failed write and no route handled a cancelled body. + */ +export function sseResponse( + options: SseOptions, + start: (stream: SseStream) => (() => void) | void, +): Response { + const { signal, retryMs, heartbeatMs = SSE_HEARTBEAT_MS } = options; + const encoder = new TextEncoder(); + let controller: ReadableStreamDefaultController | undefined; + let heartbeat: ReturnType | undefined; + let teardown: (() => void) | undefined; + let closed = false; + + const close = (): void => { + if (closed) return; + closed = true; + clearInterval(heartbeat); + signal.removeEventListener("abort", close); + const undo = teardown; + teardown = undefined; + if (typeof undo === "function") undo(); + try { + controller?.close(); + } catch { + /* the body is already closed */ + } + }; + const send = (frame: string): void => { + if (closed) return; + try { + controller?.enqueue(encoder.encode(frame)); + } catch { + // The body is closed or errored underneath us: the client is gone, so + // stop the heartbeat and the route's subscriptions now rather than + // waiting for the abort. + close(); + } + }; + const stream: SseStream = { + get closed() { + return closed; + }, + send, + sendEvent: (event, data) => send(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`), + close, + }; + + const body = new ReadableStream({ + start(c) { + controller = c; + // Prime the stream so headers and first bytes flush immediately. + send(retryMs === undefined ? ": open\n\n" : `: open\nretry: ${retryMs}\n\n`); + heartbeat = setInterval(() => send(": ping\n\n"), heartbeatMs); + // Never keep the process alive on its own. + heartbeat.unref?.(); + if (signal.aborted) return close(); + signal.addEventListener("abort", close); + let undo: (() => void) | void; + try { + undo = start(stream); + } catch (error) { + // The route failed before it could stream: stop the heartbeat and the + // abort hook here, then let the request fail as it would have. + close(); + throw error; + } + if (typeof undo !== "function") return; + // A route may close the stream from inside start; its teardown still runs once. + if (closed) undo(); + else teardown = undo; + }, + cancel: close, + }); + + return new Response(body, { headers: SSE_HEADERS }); +} diff --git a/backend/test/ratchet.test.ts b/backend/test/ratchet.test.ts index c6c8f6c3a..bda7ba70b 100644 --- a/backend/test/ratchet.test.ts +++ b/backend/test/ratchet.test.ts @@ -67,7 +67,7 @@ function lineCount(text: string): number { // out of scope (they grow with fixtures and cases). const BACKEND_SIZE_BASELINE: Record = { // Read-only and windowed thread routes live in runs/read-routes.ts. - "runs/routes.ts": 1009, + "runs/routes.ts": 919, "memory/team-memory.ts": 1005, // Scripted mock execution and worker events live in focused worker modules. };