feat(passthrough): clean response.failed terminal on mid-stream SSE reset

An upstream reset after headers used to tear the client connection with a
raw socket error and no terminal SSE event. relaySseWithFailedTail closes
any partial block, emits a synthetic response.failed (code upstream_reset)
plus data: [DONE], and closes the stream. win32 keeps the pure native tee
relay (Bun#32111 JS-sink segfault) — the guard test now pins that
platform-gated invariant. Deliberately not a resend: the upstream already
committed the request (cursor committed=non-replayable policy).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: lidge-jun

src/server.ts

@@ -457,7 +457,13 @@ async function handleResponses(
         consumeForResponseLogMetadata(inspectBody, logCtx, turnAc.signal, () => unregisterTurn(turnAc));
       }
       if (!headers.has("content-type")) headers.set("content-type", "text/event-stream");
-      return markNativePassthroughSseResponse(new Response(nativeBody, {
+      // win32 must keep the pure native relay (Bun#32111 JS-sink segfault); elsewhere a JS pull
+      // relay is established practice (relayWithAbort, relaySseWithHeartbeat) and lets a
+      // mid-stream reset end with a clean response.failed terminal instead of a raw socket error.
+      const clientBody = process.platform === "win32"
+        ? nativeBody
+        : relaySseWithFailedTail(nativeBody, upstream);
+      return markNativePassthroughSseResponse(new Response(clientBody, {
         status: upstreamResponse.status,
         headers,
       }));
@@ -982,6 +988,54 @@ export function relayWithAbort(
   });
 }
 
+/**
+ * Relay a passthrough SSE body like relayWithAbort, but convert a MID-STREAM failure (upstream
+ * reset after headers) into a clean terminal: any partial block is closed off, then a synthetic
+ * `response.failed` event and `data: [DONE]` are emitted and the stream closes. Without this the
+ * client sees a raw socket teardown with no terminal SSE event. Deliberately NOT a resend: the
+ * upstream already committed the request (duplicate-completion risk — same policy as cursor's
+ * committed=non-replayable transport retry).
+ */
+export function relaySseWithFailedTail(
+  body: ReadableStream<Uint8Array>,
+  upstream: AbortController,
+): ReadableStream<Uint8Array> {
+  const reader = body.getReader();
+  const encoder = new TextEncoder();
+  return new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        const { done, value } = await reader.read();
+        if (done) {
+          controller.close();
+          return;
+        }
+        controller.enqueue(value);
+      } catch (err) {
+        const failure = {
+          type: "upstream_error",
+          code: "upstream_reset",
+          message: `Upstream stream terminated unexpectedly: ${err instanceof Error ? err.message : String(err)}`,
+        };
+        const payload = JSON.stringify({
+          type: "response.failed",
+          response: { status: "failed", error: failure, last_error: failure },
+        });
+        try {
+          // Leading blank line terminates a partial SSE block so the failed frame parses cleanly.
+          controller.enqueue(encoder.encode(`\n\nevent: response.failed\ndata: ${payload}\n\ndata: [DONE]\n\n`));
+          controller.close();
+        } catch { /* client already torn down */ }
+        upstream.abort();
+      }
+    },
+    cancel(reason) {
+      upstream.abort(reason);
+      reader.cancel(reason).catch(() => {});
+    },
+  });
+}
+
 function nextSseBlock(buffer: string): { block: string; rest: string } | null {
   const match = buffer.match(/\r?\n\r?\n/);
   if (!match || match.index === undefined) return null;

tests/passthrough-abort.test.ts

@@ -30,7 +30,7 @@ async function readAll(stream: ReadableStream<Uint8Array>): Promise<string> {
 }
 
 describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
-  test("native passthrough SSE response body avoids async-pull client wrappers", async () => {
+  test("native passthrough SSE keeps win32 on the pure native relay (Bun#32111)", async () => {
     const source = await readSource("src/server.ts");
     const sseBranch = source.slice(
       source.indexOf("if (isEventStream && upstreamResponse.body)"),
@@ -42,7 +42,12 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
     );
 
     expect(sseBranch).toContain("upstreamResponse.body.tee()");
-    expect(sseBranch).toContain("new Response(nativeBody");
+    // win32 must receive the tee'd body untouched — no JS pull wrapper (Bun#32111 segfault).
+    expect(sseBranch).toContain('process.platform === "win32"');
+    expect(sseBranch).toContain("? nativeBody");
+    // Elsewhere the failed-tail relay converts mid-stream resets into a clean response.failed.
+    expect(sseBranch).toContain("relaySseWithFailedTail(nativeBody, upstream)");
+    expect(sseBranch).toContain("new Response(clientBody");
     expect(sseBranch).toContain("markNativePassthroughSseResponse");
     expect(sseBranch).not.toContain("relaySseWithHeartbeat(");
     expect(sseBranch).not.toContain("trackStreamLifetime(");

tests/sse-failed-tail.test.ts

@@ -0,0 +1,79 @@
+import { describe, expect, test } from "bun:test";
+import { relaySseWithFailedTail } from "../src/server";
+
+const encoder = new TextEncoder();
+const decoder = new TextDecoder();
+
+function sourceStream(chunks: string[], opts: { failAfter?: boolean; error?: Error } = {}): ReadableStream<Uint8Array> {
+  let i = 0;
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (i < chunks.length) {
+        controller.enqueue(encoder.encode(chunks[i++]));
+        return;
+      }
+      if (opts.failAfter) {
+        controller.error(opts.error ?? Object.assign(new Error("The socket connection was closed unexpectedly."), { code: "ECONNRESET" }));
+        return;
+      }
+      controller.close();
+    },
+  });
+}
+
+async function drain(stream: ReadableStream<Uint8Array>): Promise<string> {
+  const reader = stream.getReader();
+  let out = "";
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) return out;
+    out += decoder.decode(value, { stream: true });
+  }
+}
+
+describe("relaySseWithFailedTail", () => {
+  test("relays a healthy stream verbatim with no injected frame", async () => {
+    const upstream = new AbortController();
+    const src = sourceStream(["event: response.completed\n", 'data: {"type":"response.completed"}\n\n', "data: [DONE]\n\n"]);
+    const out = await drain(relaySseWithFailedTail(src, upstream));
+    expect(out).toBe('event: response.completed\ndata: {"type":"response.completed"}\n\ndata: [DONE]\n\n');
+    expect(out).not.toContain("response.failed");
+    expect(upstream.signal.aborted).toBe(false);
+  });
+
+  test("mid-stream error keeps prior bytes and appends a clean failed terminal", async () => {
+    const upstream = new AbortController();
+    const src = sourceStream(['data: {"type":"response.output_text.delta","delta":"hel', ""], { failAfter: true });
+    const out = await drain(relaySseWithFailedTail(src, upstream));
+    // Prior (partial) bytes preserved, then blank-line boundary, then the failed frame.
+    expect(out.startsWith('data: {"type":"response.output_text.delta","delta":"hel')).toBe(true);
+    expect(out).toContain("\n\nevent: response.failed\ndata: ");
+    expect(out.endsWith("data: [DONE]\n\n")).toBe(true);
+    const dataLine = out.split("event: response.failed\ndata: ")[1]!.split("\n")[0]!;
+    const parsed = JSON.parse(dataLine) as { type: string; response: { status: string; error: { code: string; message: string } } };
+    expect(parsed.type).toBe("response.failed");
+    expect(parsed.response.status).toBe("failed");
+    expect(parsed.response.error.code).toBe("upstream_reset");
+    expect(parsed.response.error.message).toContain("socket connection was closed unexpectedly");
+    // Stream CLOSED (drain returned) rather than erroring, and the upstream fetch was aborted.
+    expect(upstream.signal.aborted).toBe(true);
+  });
+
+  test("error before any bytes yields only the failed terminal", async () => {
+    const upstream = new AbortController();
+    const src = sourceStream([], { failAfter: true });
+    const out = await drain(relaySseWithFailedTail(src, upstream));
+    expect(out).toContain("event: response.failed\ndata: ");
+    expect(out.endsWith("data: [DONE]\n\n")).toBe(true);
+  });
+
+  test("client cancel aborts the upstream controller", async () => {
+    const upstream = new AbortController();
+    // A source that never ends on its own.
+    const src = new ReadableStream<Uint8Array>({ pull() { /* stay pending */ } });
+    const relayed = relaySseWithFailedTail(src, upstream);
+    const reader = relayed.getReader();
+    await reader.cancel(new DOMException("client closed", "AbortError"));
+    expect(upstream.signal.aborted).toBe(true);
+  });
+});