refactor(api): extract createSseStream helper from chat route

The Coach chat route had three identical `new ReadableStream({
start(controller) { try { ... } finally { controller.close(); } } })`
blocks (refusal / provider-error / happy path), differing only in
which frames they enqueued. Extract the boilerplate into
`src/lib/sse/create-stream.ts` so the v1.5 iOS streaming endpoints
inherit the same try/finally close semantics without copy-paste.

The helper accepts sync or async emit callbacks; the controller is
always closed in `finally` so a thrown emit never hangs the response.

Three unit tests pin the sync, async, and throw paths.

Closes Sr-H2.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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<Uint8Array>({
-    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<Uint8Array>({
-    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<Uint8Array>({
-    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

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<Uint8Array>`.
+ */
+async function readAll(stream: ReadableStream<Uint8Array>): Promise<string> {
+  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("");
+  });
+});

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<T = Uint8Array> {
+  enqueue: (chunk: T) => void;
+}
+
+/**
+ * Build a `ReadableStream<Uint8Array>` 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<void>,
+): ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    async start(controller) {
+      try {
+        await emit(controller);
+      } finally {
+        controller.close();
+      }
+    },
+  });
+}