diff --git a/src/app/api/insights/chat/route.ts b/src/app/api/insights/chat/route.ts index 4db7cb8d1..64a59e068 100644 --- a/src/app/api/insights/chat/route.ts +++ b/src/app/api/insights/chat/route.ts @@ -61,6 +61,7 @@ import { detectRefusal } from "@/lib/ai/coach/refusal"; import { getCoachSystemPrompt } from "@/lib/ai/coach/system-prompt"; import { buildCoachSnapshot } from "@/lib/ai/coach/snapshot"; import { parseKeyValuesSentinel } from "@/lib/ai/coach/keyvalues"; +import { createSseStream } from "@/lib/sse/create-stream"; /** * Hard cap on total turns kept inside the per-call prompt window. @@ -340,29 +341,23 @@ Reply now as the assistant, in ${locale === "de" ? "German" : "English"}.`; }); // ── Stream the body to the client ──────────────────────────── - const stream = new ReadableStream({ - start(controller) { - try { - for (const tok of tokeniseForStreaming(replyText)) { - controller.enqueue(encodeFrame({ type: "token", token: tok })); - } - controller.enqueue( - encodeFrame({ - type: "provenance", - metricSource: enrichedProvenance, - }), - ); - controller.enqueue( - encodeFrame({ - type: "done", - conversationId: workingConversationId, - messageId: assistantMessage.id, - }), - ); - } finally { - controller.close(); - } - }, + const stream = createSseStream((controller) => { + for (const tok of tokeniseForStreaming(replyText)) { + controller.enqueue(encodeFrame({ type: "token", token: tok })); + } + controller.enqueue( + encodeFrame({ + type: "provenance", + metricSource: enrichedProvenance, + }), + ); + controller.enqueue( + encodeFrame({ + type: "done", + conversationId: workingConversationId, + messageId: assistantMessage.id, + }), + ); }); return new Response(stream, { status: 200, headers: SSE_HEADERS }); @@ -412,43 +407,31 @@ async function streamRefusal(args: { promptVersion: PROMPT_VERSION, }); - const stream = new ReadableStream({ - start(controller) { - try { - controller.enqueue( - encodeFrame({ type: "token", token: args.refusalText }), - ); - controller.enqueue( - encodeFrame({ - type: "done", - conversationId, - messageId: refusalMessage.id, - }), - ); - } finally { - controller.close(); - } - }, + const stream = createSseStream((controller) => { + controller.enqueue( + encodeFrame({ type: "token", token: args.refusalText }), + ); + controller.enqueue( + encodeFrame({ + type: "done", + conversationId, + messageId: refusalMessage.id, + }), + ); }); return new Response(stream, { status: 200, headers: SSE_HEADERS }); } function streamProviderError(args: { code: string }): Response { - const stream = new ReadableStream({ - start(controller) { - try { - controller.enqueue( - encodeFrame({ - type: "error", - code: args.code, - message: args.code, - }), - ); - } finally { - controller.close(); - } - }, + const stream = createSseStream((controller) => { + controller.enqueue( + encodeFrame({ + type: "error", + code: args.code, + message: args.code, + }), + ); }); // Status 200 so the streaming client reads the SSE body and parses // the structured `error` frame (HTTP-status branches drop the diff --git a/src/lib/sse/__tests__/create-stream.test.ts b/src/lib/sse/__tests__/create-stream.test.ts new file mode 100644 index 000000000..10b631d14 --- /dev/null +++ b/src/lib/sse/__tests__/create-stream.test.ts @@ -0,0 +1,48 @@ +import { describe, it, expect } from "vitest"; +import { createSseStream } from "../create-stream"; + +/** + * v1.4.22 W5 reconcile (Sr-H2) — extracted SSE-stream constructor. + * + * Pin the contract: emit can be sync or async, the controller is + * always closed (success or thrown), and the resulting stream reads + * as a real `ReadableStream`. + */ +async function readAll(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let out = ""; + for (;;) { + const { value, done } = await reader.read(); + if (done) break; + out += decoder.decode(value, { stream: true }); + } + return out; +} + +describe("createSseStream", () => { + it("enqueues frames from a synchronous emit callback and closes", async () => { + const stream = createSseStream((controller) => { + controller.enqueue(new TextEncoder().encode("data: one\n\n")); + controller.enqueue(new TextEncoder().encode("data: two\n\n")); + }); + expect(await readAll(stream)).toBe("data: one\n\ndata: two\n\n"); + }); + + it("awaits an async emit callback before closing", async () => { + const stream = createSseStream(async (controller) => { + await Promise.resolve(); + controller.enqueue(new TextEncoder().encode("data: async\n\n")); + }); + expect(await readAll(stream)).toBe("data: async\n\n"); + }); + + it("closes the stream cleanly even when emit throws", async () => { + const stream = createSseStream(() => { + throw new Error("emit-failure"); + }); + // Should not hang — the finally block always closes the + // controller. The reader sees an empty body. + expect(await readAll(stream)).toBe(""); + }); +}); diff --git a/src/lib/sse/create-stream.ts b/src/lib/sse/create-stream.ts new file mode 100644 index 000000000..78181a7c2 --- /dev/null +++ b/src/lib/sse/create-stream.ts @@ -0,0 +1,43 @@ +/** + * v1.4.22 W5 reconcile (Sr-H2) — shared `ReadableStream` constructor + * for Server-Sent-Events handlers. + * + * The Coach chat route used to construct three separate + * `new ReadableStream({ start(controller) { try { … } finally { + * controller.close(); } } })` blocks — one for the refusal path, one + * for the provider-error path, one for the happy path. Each emitted a + * different mix of frames but the construction shape was identical. + * + * Extract the boilerplate so future SSE consumers (the v1.5 iOS + * daily-briefing live-regenerate endpoint is the next one) inherit the + * same try/finally close semantics without copy-paste. + * + * The `emit` callback receives the controller and is awaited inside + * the `start` body. Any error inside the callback re-enters the + * `finally` and closes the stream cleanly so the client sees the + * connection terminate (and never sees a hung response). + */ + +export interface SseController { + enqueue: (chunk: T) => void; +} + +/** + * Build a `ReadableStream` whose `start` callback runs the + * supplied `emit` function inside a try/finally that always closes the + * controller. Use the returned stream with the standard SSE response + * headers (`Content-Type: text/event-stream`, `Cache-Control: no-cache`). + */ +export function createSseStream( + emit: (controller: SseController) => void | Promise, +): ReadableStream { + return new ReadableStream({ + async start(controller) { + try { + await emit(controller); + } finally { + controller.close(); + } + }, + }); +}