Merge remote-tracking branch 'origin/release/v3.8.51' into fix/v3851-streamhandler-public-errors

CHANGELOG.md

@@ -97,6 +97,10 @@ _Living section — cycle opened at the v3.8.50 freeze (parallel-cycle model). B
 
 ### 🐛 Bug Fixes
 
+- **security(streaming):** sanitize generic mid-stream error messages before emitting OpenAI,
+  Responses, or Claude SSE failure frames and before diagnostic logging, while preserving raw
+  failures for internal classification and keeping client disconnects out of provider failure state.
+
 ### 📝 Maintenance
 
 ---

changelog.d/fixes/zed-hosted-stream-error-boundary.md

@@ -1 +0,0 @@
-- **fix(providers):** Zed Hosted streaming failures now trigger fallback before content and end partial streams with a sanitized structured error instead of fake assistant text and a normal-success stop.

open-sse/executors/zed-hosted.ts

@@ -44,8 +44,6 @@ import {
   zedLlmFetch,
   type ZedCredentials,
 } from "../shared/zedAuth.ts";
-import { buildErrorBody } from "../utils/error.ts";
-import { hasUsefulStreamContent } from "../utils/streamReadiness.ts";
 import { resolveSuppressThinkClose, THINKING_MARKER_HEADER } from "../utils/thinkCloseMarker.ts";
 
 // Wire values for the `provider` field of POST /completions. These are NOT
@@ -124,72 +122,37 @@ function convertProviderEvent(
   return event;
 }
 
-const MAX_ZED_FAILURE_MESSAGE_LENGTH = 512;
-const MAX_PENDING_ZED_OUTPUT_LENGTH = 64 * 1024;
-const ZED_STREAM_FAILURE_PUBLIC_MESSAGE = "Zed upstream stream failed";
-
-function boundedFailureText(value: unknown): string | null {
-  if (typeof value !== "string" && typeof value !== "number") return null;
-  const text = String(value).trim();
-  return text ? text.slice(0, MAX_ZED_FAILURE_MESSAGE_LENGTH) : null;
-}
-
-function extractZedFailureMessage(failed: Record<string, unknown>): string {
-  const nestedError =
-    failed.error && typeof failed.error === "object" && !Array.isArray(failed.error)
-      ? (failed.error as Record<string, unknown>)
-      : null;
-  const candidates = [
-    failed.message,
-    nestedError?.message,
-    typeof failed.error === "object" ? undefined : failed.error,
-    failed.code,
-    nestedError?.code,
-  ];
-  for (const candidate of candidates) {
-    const text = boundedFailureText(candidate);
-    if (text) return text;
-  }
-  return "request failed";
-}
-
-function createErrorChunk(message: string): ReturnType<typeof buildErrorBody> {
-  return buildErrorBody(502, `Zed stream failed: ${message}`, undefined, {
-    type: "upstream_error",
-    code: "ZED_STREAM_FAILED",
-  });
+function createErrorChunk(model: string, message: string): Record<string, unknown> {
+  return {
+    id: `chatcmpl-zed-error-${Date.now()}`,
+    object: "chat.completion.chunk",
+    created: Math.floor(Date.now() / 1000),
+    model,
+    choices: [{ index: 0, delta: { content: `[Zed error] ${message}` }, finish_reason: "stop" }],
+  };
 }
 
 /**
- * The controller capabilities these SSE helpers use. Normal frames only enqueue;
- * terminal failures also terminate so they do not depend on the upstream socket
- * eventually reaching EOF. Narrow controller types keep the helpers honest. The wider
+ * The single controller capability these SSE helpers use. They only ever enqueue —
+ * never `close()`, never read `desiredSize` — so typing them by that one method lets
+ * the same code serve both stream kinds. The wider
  * `ReadableStreamDefaultController` annotation rejected every call site, because the
  * helpers are driven from a TransformStream and `TransformStreamDefaultController`
  * has no `close()`.
  */
 type SseEnqueueTarget = Pick<ReadableStreamDefaultController<Uint8Array>, "enqueue">;
-type SseProcessTarget = Pick<TransformStreamDefaultController<Uint8Array>, "enqueue" | "terminate">;
-
-function serializeSseObject(chunk: unknown): string {
-  if (!chunk) return "";
-  let serialized = "";
-  const items = Array.isArray(chunk) ? chunk : [chunk];
-  for (const item of items) {
-    if (!item) continue;
-    serialized += `data: ${JSON.stringify(item)}\n\n`;
-  }
-  return serialized;
-}
 
 function enqueueSseObject(
   controller: SseEnqueueTarget,
   encoder: TextEncoder,
   chunk: unknown
 ): void {
-  const serialized = serializeSseObject(chunk);
-  if (!serialized) return;
-  controller.enqueue(encoder.encode(serialized));
+  if (!chunk) return;
+  const items = Array.isArray(chunk) ? chunk : [chunk];
+  for (const item of items) {
+    if (!item) continue;
+    controller.enqueue(encoder.encode(`data: ${JSON.stringify(item)}\n\n`));
+  }
 }
 
 type ZedLine = { done?: true; status?: unknown; event?: unknown } | null;
@@ -263,47 +226,16 @@ function wrapZedCompletionStream(
   }
   let buffer = "";
   let done = false;
-  let providerOutputForwarded = false;
-  let pendingProviderOutput = "";
-  let pendingFailure: (Error & { statusCode: number }) | null = null;
-
-  const forwardProviderOutput = (controller: SseEnqueueTarget, chunk: unknown) => {
-    const serialized = serializeSseObject(chunk);
-    if (!serialized) return;
-    if (providerOutputForwarded) {
-      controller.enqueue(encoder.encode(serialized));
-      return;
-    }
-
-    // A role/bootstrap-only chunk makes ensureStreamReadiness release the response before any
-    // model output exists. If the next chunk is status.failed, downstream read-ahead can discard
-    // the first real content while propagating the error. Hold structural frames until the first
-    // substantive text/reasoning/tool delta, then release them atomically with that output.
-    const outputWithBootstrap = pendingProviderOutput + serialized;
-    if (!hasUsefulStreamContent(outputWithBootstrap)) {
-      pendingProviderOutput =
-        outputWithBootstrap.length <= MAX_PENDING_ZED_OUTPUT_LENGTH
-          ? outputWithBootstrap
-          : serialized.length <= MAX_PENDING_ZED_OUTPUT_LENGTH
-            ? serialized
-            : "";
-      return;
-    }
-    controller.enqueue(encoder.encode(outputWithBootstrap));
-    pendingProviderOutput = "";
-    providerOutputForwarded = true;
-  };
 
   const finish = (controller: SseEnqueueTarget) => {
     if (done) return;
     const finalChunk = convertProviderEvent(provider, null, state);
-    const finalOutput = `${pendingProviderOutput}${serializeSseObject(finalChunk)}data: [DONE]\n\n`;
-    pendingProviderOutput = "";
-    controller.enqueue(encoder.encode(finalOutput));
+    enqueueSseObject(controller, encoder, finalChunk);
+    controller.enqueue(encoder.encode("data: [DONE]\n\n"));
     done = true;
   };
 
-  const processLine = (line: string, controller: SseProcessTarget) => {
+  const processLine = (line: string, controller: SseEnqueueTarget) => {
     if (done) return;
     const payload = unwrapZedLine(line);
     if (!payload) return;
@@ -314,29 +246,17 @@ function wrapZedCompletionStream(
     if (payload.status) {
       const status = normalizeStatus(payload.status);
       if (status?.type === "failed" || status?.failed) {
-        const failed =
-          status.failed && typeof status.failed === "object" && !Array.isArray(status.failed)
-            ? (status.failed as Record<string, unknown>)
-            : status;
-        if (providerOutputForwarded) {
-          pendingFailure = Object.assign(new Error(ZED_STREAM_FAILURE_PUBLIC_MESSAGE), {
-            statusCode: 502,
-          });
-          done = true;
-          controller.terminate();
-          return;
-        }
-        pendingProviderOutput = "";
-        enqueueSseObject(controller, encoder, createErrorChunk(extractZedFailureMessage(failed)));
-        done = true;
-        controller.terminate();
+        const failed = (status.failed as Record<string, unknown>) || status;
+        const message = String(failed.message || failed.error || failed.code || "request failed");
+        enqueueSseObject(controller, encoder, createErrorChunk(model, message));
+        finish(controller);
       } else if (status?.type === "stream_ended" || status === ("stream_ended" as unknown)) {
         finish(controller);
       }
       return;
     }
     const converted = convertProviderEvent(provider, payload.event, state);
-    forwardProviderOutput(controller, converted);
+    enqueueSseObject(controller, encoder, converted);
   };
 
   const transformed = response.body.pipeThrough(
@@ -361,47 +281,7 @@ function wrapZedCompletionStream(
     })
   );
 
-  // `TransformStreamDefaultController.error()` discards already-enqueued output. A failed
-  // status can share one upstream network chunk with the last content delta, so erroring the
-  // transform immediately would erase that partial answer. Drain the transformed chunks through
-  // a backpressure-aware reader first, then reject the next read with the fixed public error.
-  // The normal chat pipeline turns that rejection into its client-format terminal frame and
-  // records the 502 through the existing failure finalizers.
-  const transformedReader = transformed.getReader();
-  let guardedStreamCancelled = false;
-  const cancelTransformedReader = (reason: unknown) => {
-    if (guardedStreamCancelled) return;
-    guardedStreamCancelled = true;
-    // Client cancellation must settle independently of an upstream body whose cancel hook hangs.
-    // Request cancellation once, but do not await provider cleanup on the client-facing boundary.
-    void transformedReader.cancel(reason).catch(() => {
-      console.debug("[ZED] upstream stream cancellation rejected");
-    });
-  };
-  const guardedStream = new ReadableStream<Uint8Array>({
-    async pull(controller) {
-      try {
-        const next = await transformedReader.read();
-        if (guardedStreamCancelled) return;
-        if (!next.done) {
-          controller.enqueue(next.value);
-          return;
-        }
-        if (pendingFailure) {
-          controller.error(pendingFailure);
-          return;
-        }
-        controller.close();
-      } catch (error) {
-        if (!guardedStreamCancelled) controller.error(error);
-      }
-    },
-    cancel(reason) {
-      cancelTransformedReader(reason);
-    },
-  });
-
-  return new Response(guardedStream, {
+  return new Response(transformed, {
     status: response.status,
     statusText: response.statusText,
     headers: {

open-sse/utils/streamHandler.ts

@@ -1,6 +1,7 @@
 import { trackPendingRequest } from "@/lib/usageDb";
 import { STREAM_IDLE_TIMEOUT_MS } from "../config/constants.ts";
 import { FORMATS } from "../translator/formats.ts";
+import { buildErrorBody } from "./error.ts";
 import { PENDING_REQUEST_CLEARED_MARKER } from "./stream.ts";
 import { createCompletedResponsesToolHandoffWatcher } from "./responsesToolHandoff.ts";
 import { createStreamContentWatcher, type StreamContentWatcher } from "./streamReadiness.ts";
@@ -187,6 +188,10 @@ function getErrorStatusCode(error: unknown): number {
   return 502;
 }
 
+function getPublicErrorMessage(errorMsg: string, statusCode: number): string {
+  return buildErrorBody(statusCode, errorMsg).error.message;
+}
+
 function isDeadlineAbortReason(reason: unknown): reason is Error {
   return (
     reason instanceof Error &&
@@ -406,7 +411,7 @@ export function createStreamController({
       }
 
       if (error instanceof Error) {
-        logStream(`error: ${error.message}`);
+        logStream(`error: ${getPublicErrorMessage(error.message, getErrorStatusCode(error))}`);
         return;
       }
       logStream("error: unknown");
@@ -452,6 +457,7 @@ export function buildStreamErrorChunks(
   clientResponseFormat?: string | null
 ) {
   const statusMapping = getStreamErrorStatusMapping(statusCode);
+  const publicErrorMessage = getPublicErrorMessage(errorMsg, statusCode);
 
   if (isResponsesClientFormat(clientResponseFormat)) {
     const errorEvent = {
@@ -460,7 +466,7 @@ export function buildStreamErrorChunks(
         id: null,
         status: "failed",
         error: {
-          message: errorMsg,
+          message: publicErrorMessage,
           type: statusMapping.responses.type,
           code: statusMapping.responses.code,
         },
@@ -475,7 +481,7 @@ export function buildStreamErrorChunks(
       type: "error",
       error: {
         type: statusMapping.claude.type,
-        message: errorMsg,
+        message: publicErrorMessage,
       },
     };
 
@@ -498,7 +504,7 @@ export function buildStreamErrorChunks(
       },
     ],
     error: {
-      message: errorMsg,
+      message: publicErrorMessage,
       type: statusMapping.responses.type,
       code: statusMapping.responses.code,
     },

tests/fixtures/stream-handler-public-error-boundary.fixture.ts

@@ -0,0 +1,211 @@
+// This suite owns process-wide DATA_DIR, plugin, logger, and DB state. It must run only inside
+// the subprocess launched by tests/unit/stream-handler-public-error-boundary.test.ts.
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import test from "node:test";
+
+const originalDataDir = process.env.DATA_DIR;
+const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR;
+const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-stream-public-error-"));
+const TEST_DATA_DIR = path.join(testRoot, "data");
+const TEST_PLUGINS_DIR = path.join(testRoot, "plugins");
+fs.mkdirSync(TEST_DATA_DIR, { recursive: true });
+fs.mkdirSync(TEST_PLUGINS_DIR, { recursive: true });
+process.env.DATA_DIR = TEST_DATA_DIR;
+process.env.OMNIROUTE_PLUGINS_DIR = TEST_PLUGINS_DIR;
+
+const [core, callLogs, artifactWriter, loggerResource, streamHandler, { FORMATS }] =
+  await Promise.all([
+    import("../../src/lib/db/core.ts"),
+    import("../../src/lib/usage/callLogs.ts"),
+    import("../../src/lib/usage/callLogArtifactWriter.ts"),
+    import("../../src/shared/utils/loggerResource.ts"),
+    import("../../open-sse/utils/streamHandler.ts"),
+    import("../../open-sse/translator/formats.ts"),
+  ]);
+const { createStreamController, pipeWithDisconnect } = streamHandler;
+
+const SECRET = "sk-live-streamhandler-secret-123456";
+const API_KEY = "provider-key-streamhandler-654321";
+const PRIVATE_PATH = "/srv/omniroute/private/provider.ts:42:9";
+const RAW_MESSAGE =
+  `Upstream failed at ${PRIVATE_PATH} Authorization: Bearer ${SECRET} api_key=${API_KEY}` +
+  `\n    at dispatch (/srv/omniroute/private/dispatcher.ts:88:3)`;
+
+test.after(async () => {
+  assert.equal(await callLogs.waitForCallLogSaves(3_000), true);
+  await artifactWriter.closeCallLogArtifactWriter();
+  core.resetDbInstance();
+  await loggerResource.closeSharedLoggerResource();
+
+  if (originalDataDir === undefined) delete process.env.DATA_DIR;
+  else process.env.DATA_DIR = originalDataDir;
+  if (originalPluginsDir === undefined) delete process.env.OMNIROUTE_PLUGINS_DIR;
+  else process.env.OMNIROUTE_PLUGINS_DIR = originalPluginsDir;
+
+  fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+});
+
+test("fixture binds all persistent state to its process-owned directories", () => {
+  assert.equal(core.DATA_DIR, TEST_DATA_DIR);
+  assert.equal(core.SQLITE_FILE, path.join(TEST_DATA_DIR, "storage.sqlite"));
+  assert.equal(process.env.DATA_DIR, TEST_DATA_DIR);
+  assert.equal(process.env.OMNIROUTE_PLUGINS_DIR, TEST_PLUGINS_DIR);
+  assert.equal(fs.existsSync(TEST_DATA_DIR), true);
+  assert.equal(fs.existsSync(TEST_PLUGINS_DIR), true);
+});
+
+test("OpenAI stream failures keep raw diagnostics internal and sanitize the public wire", async () => {
+  const upstreamError = Object.assign(new Error(RAW_MESSAGE), { statusCode: 502 });
+  const source = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.error(upstreamError);
+    },
+  });
+  let internalMessage = "";
+
+  const stream = pipeWithDisconnect(
+    new Response(source),
+    new TransformStream<Uint8Array, Uint8Array>(),
+    createStreamController({
+      clientResponseFormat: FORMATS.OPENAI,
+      onError(event) {
+        internalMessage = event.message;
+        return true;
+      },
+    }),
+    { stallTimeoutMs: 0 }
+  );
+  const publicWire = await new Response(stream).text();
+
+  assert.equal(internalMessage, RAW_MESSAGE, "failure classification must retain the raw message");
+  assert.match(publicWire, /"finish_reason":"error"/);
+  assert.match(publicWire, /"code":"server_error"/);
+  assert.match(publicWire, /\[DONE\]/);
+  assert.doesNotMatch(publicWire, new RegExp(SECRET));
+  assert.doesNotMatch(publicWire, new RegExp(API_KEY));
+  assert.doesNotMatch(publicWire, /\/srv\/omniroute\/private/);
+  assert.doesNotMatch(publicWire, /dispatcher\.ts/);
+  assert.match(publicWire, /Authorization: \[REDACTED\]/);
+  assert.match(publicWire, /<path>/);
+});
+
+test("Responses stream failures preserve the failure event shape without leaking diagnostics", async () => {
+  const upstreamError = Object.assign(new Error(RAW_MESSAGE), { statusCode: 429 });
+  const source = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.error(upstreamError);
+    },
+  });
+  let internalError: unknown;
+
+  const stream = pipeWithDisconnect(
+    new Response(source),
+    new TransformStream<Uint8Array, Uint8Array>(),
+    createStreamController({
+      clientResponseFormat: FORMATS.OPENAI_RESPONSES,
+      onError(event) {
+        internalError = event.error;
+        return true;
+      },
+    }),
+    { stallTimeoutMs: 0 }
+  );
+  const publicWire = await new Response(stream).text();
+
+  assert.equal(internalError, upstreamError, "the original error object must reach classification");
+  assert.match(publicWire, /event: response\.failed/);
+  assert.match(publicWire, /"type":"response\.failed"/);
+  assert.match(publicWire, /"type":"rate_limit_error"/);
+  assert.match(publicWire, /"code":"rate_limit_exceeded"/);
+  assert.doesNotMatch(publicWire, new RegExp(SECRET));
+  assert.doesNotMatch(publicWire, new RegExp(API_KEY));
+  assert.doesNotMatch(publicWire, /\/srv\/omniroute\/private/);
+  assert.doesNotMatch(publicWire, /dispatcher\.ts/);
+  assert.match(publicWire, /Authorization: \[REDACTED\]/);
+  assert.match(publicWire, /<path>/);
+});
+
+test("Claude stream failures preserve error and stop events without leaking diagnostics", async () => {
+  const upstreamError = Object.assign(new Error(RAW_MESSAGE), { statusCode: 403 });
+  const source = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.error(upstreamError);
+    },
+  });
+  let internalStatusCode = 0;
+
+  const stream = pipeWithDisconnect(
+    new Response(source),
+    new TransformStream<Uint8Array, Uint8Array>(),
+    createStreamController({
+      clientResponseFormat: FORMATS.CLAUDE,
+      onError(event) {
+        internalStatusCode = event.statusCode;
+        return true;
+      },
+    }),
+    { stallTimeoutMs: 0 }
+  );
+  const publicWire = await new Response(stream).text();
+
+  assert.equal(internalStatusCode, 403);
+  assert.match(publicWire, /event: error/);
+  assert.match(publicWire, /"type":"permission_error"/);
+  assert.match(publicWire, /event: message_stop/);
+  assert.doesNotMatch(publicWire, new RegExp(SECRET));
+  assert.doesNotMatch(publicWire, new RegExp(API_KEY));
+  assert.doesNotMatch(publicWire, /\/srv\/omniroute\/private/);
+  assert.doesNotMatch(publicWire, /dispatcher\.ts/);
+  assert.match(publicWire, /Authorization: \[REDACTED\]/);
+  assert.match(publicWire, /<path>/);
+});
+
+test("stream diagnostics sanitize logs while callbacks retain the original failure", () => {
+  const upstreamError = Object.assign(new Error(RAW_MESSAGE), { statusCode: 502 });
+  const originalLog = console.log;
+  const logLines: string[] = [];
+  let internalError: unknown;
+  console.log = (...args: unknown[]) => {
+    logLines.push(args.map(String).join(" "));
+  };
+
+  try {
+    createStreamController({
+      provider: "test-provider",
+      model: "test-model",
+      onError(event) {
+        internalError = event.error;
+        return true;
+      },
+    }).handleError(upstreamError);
+  } finally {
+    console.log = originalLog;
+  }
+
+  const logs = logLines.join("\n");
+  assert.equal(internalError, upstreamError);
+  assert.match(logs, /error: Upstream failed at <path>/);
+  assert.match(logs, /Authorization: \[REDACTED\]/);
+  assert.doesNotMatch(logs, new RegExp(SECRET));
+  assert.doesNotMatch(logs, new RegExp(API_KEY));
+  assert.doesNotMatch(logs, /\/srv\/omniroute\/private/);
+  assert.doesNotMatch(logs, /dispatcher\.ts/);
+});
+
+test("client disconnects stay outside the provider-failure callback", () => {
+  let providerFailureRecorded = false;
+  const controller = createStreamController({
+    onError() {
+      providerFailureRecorded = true;
+      return true;
+    },
+  });
+
+  controller.handleError(new DOMException("request_signal_aborted", "AbortError"));
+
+  assert.equal(providerFailureRecorded, false);
+  assert.equal(controller.signal.aborted, false);
+});

tests/fixtures/zed-hosted-stream-error-boundary-child.ts

@@ -1,341 +0,0 @@
-import test from "node:test";
-import assert from "node:assert/strict";
-import fs from "node:fs";
-import os from "node:os";
-import path from "node:path";
-
-// This file is executed only by the process-isolated unit-test wrapper. State
-// mutations and repository imports must remain here, never in the parent test.
-const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-zed-stream-data-"));
-const TEST_PLUGINS_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-zed-stream-plugins-"));
-const originalDataDir = process.env.DATA_DIR;
-const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR;
-const originalFetch = globalThis.fetch;
-let networkCalls = 0;
-
-process.env.DATA_DIR = TEST_DATA_DIR;
-process.env.OMNIROUTE_PLUGINS_DIR = TEST_PLUGINS_DIR;
-globalThis.fetch = async () => {
-  networkCalls += 1;
-  throw new Error("Unexpected network access in Zed stream boundary test");
-};
-
-const core = await import("../../src/lib/db/core.ts");
-const loggerResource = await import("../../src/shared/utils/loggerResource.ts");
-const { __test__ } = await import("../../open-sse/executors/zed-hosted.ts");
-const { FORMATS } = await import("../../open-sse/translator/formats.ts");
-const { assembleStreamingPipeline } =
-  await import("../../open-sse/handlers/chatCore/streamingPipeline.ts");
-const { createPassthroughStreamWithLogger } = await import("../../open-sse/utils/stream.ts");
-const { createStreamFailureFinalizers } =
-  await import("../../open-sse/utils/streamFailureFinalization.ts");
-const { createStreamController } = await import("../../open-sse/utils/streamHandler.ts");
-const { ensureStreamReadiness } = await import("../../open-sse/utils/streamReadiness.ts");
-const { wrapZedCompletionStream } = __test__;
-
-type StreamCompletionEvent = Parameters<
-  Parameters<typeof createStreamFailureFinalizers>[0]["onStreamComplete"]
->[0];
-
-const RAW_FAILURE = "Bearer TOP_SECRET /srv/omniroute/zed-handler.ts:42 api_key=zed-secret";
-const TEST_MODEL = "grok-test-zed-stream-boundary";
-const TEST_CONNECTION_ID = "zed-stream-boundary-partial-connection";
-
-function failedStatusLine(): string {
-  return JSON.stringify({ status: { failed: { message: RAW_FAILURE } } });
-}
-
-function nestedFailedStatusLine(): string {
-  return JSON.stringify({
-    status: {
-      type: "failed",
-      error: { message: `${RAW_FAILURE} ${"x".repeat(2_000)}` },
-    },
-  });
-}
-
-function wrapOpenNdjson(lines: unknown[]): Response {
-  const encoder = new TextEncoder();
-  const body = lines
-    .map((line) => (typeof line === "string" ? line : JSON.stringify(line)))
-    .join("\n");
-  const stream = new ReadableStream<Uint8Array>({
-    start(controller) {
-      controller.enqueue(encoder.encode(`${body}\n`));
-      // Keep the upstream open: status.failed must terminate the wrapped stream itself.
-    },
-    cancel() {},
-  });
-  return wrapZedCompletionStream(
-    new Response(stream, {
-      status: 200,
-      headers: { "Content-Type": "application/x-ndjson" },
-    }),
-    "x_ai",
-    TEST_MODEL
-  );
-}
-
-function wrapOpenFailedNdjson(): Response {
-  const encoder = new TextEncoder();
-  const body = new ReadableStream<Uint8Array>({
-    start(controller) {
-      controller.enqueue(encoder.encode(`${failedStatusLine()}\n`));
-      // Deliberately stay open: status.failed is terminal by itself and must not
-      // depend on the upstream socket eventually reaching EOF.
-    },
-    cancel() {},
-  });
-  return wrapZedCompletionStream(
-    new Response(body, {
-      status: 200,
-      headers: { "Content-Type": "application/x-ndjson" },
-    }),
-    "x_ai",
-    TEST_MODEL
-  );
-}
-
-function wrapStalledNdjson(onCancel: () => void): Response {
-  const body = new ReadableStream<Uint8Array>({
-    cancel() {
-      onCancel();
-      return new Promise<void>(() => {});
-    },
-  });
-  return wrapZedCompletionStream(
-    new Response(body, {
-      status: 200,
-      headers: { "Content-Type": "application/x-ndjson" },
-    }),
-    "x_ai",
-    TEST_MODEL
-  );
-}
-
-async function resolvesWithin(promise: Promise<unknown>, timeoutMs: number): Promise<void> {
-  let timeout: ReturnType<typeof setTimeout> | undefined;
-  try {
-    await Promise.race([
-      promise,
-      new Promise<never>((_, reject) => {
-        timeout = setTimeout(
-          () => reject(new Error(`operation exceeded ${timeoutMs}ms`)),
-          timeoutMs
-        );
-      }),
-    ]);
-  } finally {
-    if (timeout) clearTimeout(timeout);
-  }
-}
-
-async function waitFor(predicate: () => boolean, timeoutMs: number): Promise<void> {
-  const deadline = Date.now() + timeoutMs;
-  while (!predicate() && Date.now() < deadline) {
-    await new Promise((resolve) => setTimeout(resolve, 1));
-  }
-  assert.equal(predicate(), true, `condition was not met within ${timeoutMs}ms`);
-}
-
-function parseSsePayloads(text: string): Array<Record<string, unknown>> {
-  return text
-    .split(/\r?\n/)
-    .filter((line) => line.startsWith("data: ") && line.slice(6) !== "[DONE]")
-    .map((line) => JSON.parse(line.slice(6)) as Record<string, unknown>);
-}
-
-function assertNoSensitiveFailureText(text: string): void {
-  assert.doesNotMatch(text, /TOP_SECRET|zed-secret|\/srv\/omniroute\/zed-handler\.ts/);
-}
-
-test.after(async () => {
-  core.resetDbInstance();
-  await loggerResource.closeSharedLoggerResource();
-  globalThis.fetch = originalFetch;
-  if (originalDataDir === undefined) delete process.env.DATA_DIR;
-  else process.env.DATA_DIR = originalDataDir;
-  if (originalPluginsDir === undefined) delete process.env.OMNIROUTE_PLUGINS_DIR;
-  else process.env.OMNIROUTE_PLUGINS_DIR = originalPluginsDir;
-  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
-  fs.rmSync(TEST_PLUGINS_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
-});
-
-test("zed-hosted pre-content status.failed becomes a sanitized 502 readiness failure", async () => {
-  const readiness = await ensureStreamReadiness(wrapOpenFailedNdjson(), {
-    timeoutMs: 100,
-    provider: "zed-hosted",
-    model: TEST_MODEL,
-  });
-
-  assert.equal(readiness.ok, false, "the structured error must remain eligible for fallback");
-  if (readiness.ok) assert.fail("pre-content Zed failure must not make the stream ready");
-  assert.equal(readiness.response.status, 502);
-  assert.equal(readiness.code, "STREAM_EARLY_EOF");
-
-  const bodyText = await readiness.response.text();
-  const body = JSON.parse(bodyText) as {
-    error: { message: string; type: string; code: string };
-    upstream_details?: { error?: { message?: string } };
-  };
-  assert.equal(body.error.type, "stream_early_eof");
-  assert.equal(body.error.code, "STREAM_EARLY_EOF");
-  assert.match(body.upstream_details?.error?.message ?? "", /Zed stream failed/i);
-  assertNoSensitiveFailureText(bodyText);
-  assert.equal(networkCalls, 0);
-});
-
-test("zed-hosted partial failure reaches stream finalization and persistence as 502", async () => {
-  const roleChunk = {
-    event: {
-      id: "chatcmpl-zed-partial",
-      object: "chat.completion.chunk",
-      choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }],
-    },
-  };
-  const contentChunk = {
-    event: {
-      id: "chatcmpl-zed-partial",
-      object: "chat.completion.chunk",
-      choices: [{ index: 0, delta: { content: "partial answer" }, finish_reason: null }],
-    },
-  };
-  const readiness = await ensureStreamReadiness(
-    wrapOpenNdjson([
-      roleChunk,
-      contentChunk,
-      nestedFailedStatusLine(),
-      { event: { ignored: "after failure" } },
-    ]),
-    {
-      timeoutMs: 100,
-      provider: "zed-hosted",
-      model: TEST_MODEL,
-    }
-  );
-
-  assert.equal(readiness.ok, true, "partial model output must remain deliverable");
-  const completionEvents: StreamCompletionEvent[] = [];
-  const persistedFailures: Array<{
-    connectionId: string;
-    model: string;
-    status: number;
-    code?: string;
-  }> = [];
-  const streamFailures: Array<{ status: number; message: string; code?: string; type?: string }> =
-    [];
-  const pipelineErrors: Array<{ message: string; statusCode: number }> = [];
-  let streamCompletionRecorded = false;
-  let failureCompletionRecorded = false;
-
-  const recordCompletion = (payload: StreamCompletionEvent): void => {
-    if (streamCompletionRecorded) return;
-    streamCompletionRecorded = true;
-    if (payload.status !== 200) failureCompletionRecorded = true;
-    completionEvents.push(payload);
-  };
-  const finalizers = createStreamFailureFinalizers({
-    isFailureCompletionRecorded: () => failureCompletionRecorded,
-    isStreamCompletionRecorded: () => streamCompletionRecorded,
-    onStreamComplete: recordCompletion,
-    persistFailureUsage: (status, code) =>
-      persistedFailures.push({
-        connectionId: TEST_CONNECTION_ID,
-        model: TEST_MODEL,
-        status,
-        code,
-      }),
-    onStreamFailure: (failure) => streamFailures.push(failure),
-  });
-  const streamController = createStreamController({
-    onError: (event) => {
-      pipelineErrors.push({ message: event.message, statusCode: event.statusCode });
-      return finalizers.onPipelineStreamError(event);
-    },
-    provider: "zed-hosted",
-    model: TEST_MODEL,
-    connectionId: TEST_CONNECTION_ID,
-    clientResponseFormat: FORMATS.OPENAI,
-  });
-  const transformStream = createPassthroughStreamWithLogger(
-    "zed-hosted",
-    null,
-    null,
-    TEST_MODEL,
-    TEST_CONNECTION_ID,
-    { messages: [{ role: "user", content: "test" }] },
-    recordCompletion,
-    null,
-    finalizers.handleStreamFailure,
-    FORMATS.OPENAI
-  );
-  const responseHeaders: Record<string, string> = {};
-  const finalStream = assembleStreamingPipeline({
-    providerResponse: readiness.response,
-    transformStream,
-    streamController,
-    createPiiTransform: null,
-    clientRawRequestHeaders: null,
-    clientResponseFormat: FORMATS.OPENAI,
-    echoModel: null,
-    responseHeaders,
-  });
-  const text = await new Response(finalStream, { headers: responseHeaders }).text();
-  const payloads = parseSsePayloads(text);
-  const errorPayload = payloads.find((payload) => "error" in payload) as
-    { error: { message: string; type: string; code: string } } | undefined;
-
-  assert.match(text, /partial answer/);
-  assert.ok(errorPayload, "the stream handler must emit its format-safe terminal error");
-  assert.equal(errorPayload.error.type, "server_error");
-  assert.equal(errorPayload.error.code, "server_error");
-  assert.equal(errorPayload.error.message, "Zed upstream stream failed");
-  assert.match(text, /"finish_reason":"error"/);
-  assert.doesNotMatch(text, /\[Zed error\]|"finish_reason":"stop"|response\.failed/);
-  assert.doesNotMatch(text, /"ignored":"after failure"/);
-  assertNoSensitiveFailureText(text);
-
-  assert.equal(completionEvents.length, 1, "the failure must finalize exactly once");
-  assert.equal(completionEvents[0].status, 502);
-  assert.equal(completionEvents[0].error, "Zed upstream stream failed");
-  assert.equal(completionEvents[0].errorCode, "stream_pipeline_error");
-  assert.deepEqual(persistedFailures, [
-    {
-      connectionId: TEST_CONNECTION_ID,
-      model: TEST_MODEL,
-      status: 502,
-      code: "stream_pipeline_error",
-    },
-  ]);
-  assert.deepEqual(streamFailures, [
-    {
-      status: 502,
-      message: "Zed upstream stream failed",
-      code: "stream_pipeline_error",
-      type: "stream_error",
-    },
-  ]);
-  assert.deepEqual(pipelineErrors, [{ message: "Zed upstream stream failed", statusCode: 502 }]);
-  assertNoSensitiveFailureText(JSON.stringify(completionEvents));
-  assert.equal(networkCalls, 0);
-});
-
-test("zed-hosted client cancellation does not await a stalled upstream cancel hook", async () => {
-  let upstreamCancelCalls = 0;
-  const response = wrapStalledNdjson(() => {
-    upstreamCancelCalls += 1;
-  });
-  assert.ok(response.body);
-
-  const reader = response.body.getReader();
-  const pendingRead = reader.read();
-  await resolvesWithin(reader.cancel("client disconnected"), 100);
-  const readResult = await pendingRead;
-  assert.equal(readResult.done, true);
-  await waitFor(() => upstreamCancelCalls === 1, 100);
-
-  await resolvesWithin(reader.cancel("duplicate cancel"), 100);
-  await new Promise((resolve) => setTimeout(resolve, 10));
-  assert.equal(upstreamCancelCalls, 1, "the upstream cancel hook must be requested exactly once");
-  assert.equal(networkCalls, 0);
-});

tests/unit/stream-handler-public-error-boundary.test.ts

@@ -0,0 +1,62 @@
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import { fileURLToPath } from "node:url";
+import test from "node:test";
+
+const REPO_ROOT = fileURLToPath(new URL("../..", import.meta.url));
+const FIXTURE = fileURLToPath(
+  new URL("../fixtures/stream-handler-public-error-boundary.fixture.ts", import.meta.url)
+);
+
+const CHILD_RUNTIME_ENV_KEYS = [
+  "PATH",
+  "TMPDIR",
+  "TMP",
+  "TEMP",
+  "SystemRoot",
+  "ComSpec",
+  "PATHEXT",
+  "LANG",
+  "LC_ALL",
+  "TZ",
+] as const;
+
+function buildFixtureEnv(): NodeJS.ProcessEnv {
+  const env: NodeJS.ProcessEnv = {
+    NODE_ENV: "test",
+    APP_LOG_TO_FILE: "false",
+    API_KEY_SECRET: "stream-handler-boundary-fixture-secret-20260902",
+    DISABLE_SQLITE_AUTO_BACKUP: "true",
+    NO_COLOR: "1",
+  };
+
+  for (const key of CHILD_RUNTIME_ENV_KEYS) {
+    const value = process.env[key];
+    if (value !== undefined) env[key] = value;
+  }
+
+  // Nested test runners must not inherit the parent runner's recursion marker.
+  delete env.NODE_TEST_CONTEXT;
+  return env;
+}
+
+test("generic stream public error boundaries pass in an isolated process", () => {
+  const result = spawnSync(
+    process.execPath,
+    ["--import", "tsx/esm", "--import", "./open-sse/utils/setupPolyfill.ts", "--test", FIXTURE],
+    {
+      cwd: REPO_ROOT,
+      encoding: "utf8",
+      env: buildFixtureEnv(),
+      timeout: 120_000,
+    }
+  );
+  const output = `${result.stdout}\n${result.stderr}`;
+
+  assert.ifError(result.error);
+  assert.equal(result.signal, null, output.slice(-12_000));
+  assert.equal(result.status, 0, output.slice(-12_000));
+  assert.match(output, /(?:^|\s)tests\s+6(?:\s|$)/m);
+  assert.match(output, /(?:^|\s)pass\s+6(?:\s|$)/m);
+  assert.match(output, /(?:^|\s)fail\s+0(?:\s|$)/m);
+});

tests/unit/stream-handler.test.ts

@@ -256,7 +256,8 @@ test("createDisconnectAwareStream emits Responses API failure events for Respons
 
   assert.match(text, /event: response\.failed/);
   assert.match(text, /"type":"response\.failed"/);
-  assert.match(text, /"message":"responses stream\\ndied"/);
+  assert.match(text, /"message":"responses stream"/);
+  assert.doesNotMatch(text, /died/);
   assert.match(text, /"type":"server_error"/);
   assert.match(text, /"code":"server_error"/);
   assert.doesNotMatch(text, /chat\.completion\.chunk/);
@@ -264,7 +265,7 @@ test("createDisconnectAwareStream emits Responses API failure events for Respons
   assert.doesNotMatch(text, /\[DONE\]/);
 });
 
-test("createDisconnectAwareStream keeps newlines escaped inside SSE data fields", async () => {
+test("createDisconnectAwareStream strips multiline diagnostic tails from Responses errors", async () => {
   const upstreamError = Object.assign(new Error("line one\nline two\rline three"), {
     statusCode: 400,
   });
@@ -290,9 +291,9 @@ test("createDisconnectAwareStream keeps newlines escaped inside SSE data fields"
   const text = await readStreamText(stream);
 
   assert.match(text, /^event: response\.failed\ndata: \{"type":"response\.failed"/);
-  assert.match(text, /"message":"line one\\nline two\\rline three"/);
-  assert.doesNotMatch(text, /^line two/m);
-  assert.doesNotMatch(text, /^line three/m);
+  assert.match(text, /"message":"line one"/);
+  assert.doesNotMatch(text, /line two/);
+  assert.doesNotMatch(text, /line three/);
 });
 
 test("createDisconnectAwareStream treats legacy OpenAI response format alias as Responses", async () => {
@@ -360,7 +361,7 @@ test("createDisconnectAwareStream emits Claude SSE errors for Claude clients", a
   assert.doesNotMatch(text, /\[DONE\]/);
 });
 
-test("createDisconnectAwareStream keeps newlines escaped for Claude SSE errors", async () => {
+test("createDisconnectAwareStream strips multiline diagnostic tails from Claude errors", async () => {
   const upstreamError = Object.assign(new Error("claude line one\nclaude line two"), {
     statusCode: 502,
   });
@@ -386,8 +387,8 @@ test("createDisconnectAwareStream keeps newlines escaped for Claude SSE errors",
   const text = await readStreamText(stream);
 
   assert.match(text, /^event: error\ndata: \{"type":"error"/);
-  assert.match(text, /"message":"claude line one\\nclaude line two"/);
-  assert.doesNotMatch(text, /^claude line two/m);
+  assert.match(text, /"message":"claude line one"/);
+  assert.doesNotMatch(text, /claude line two/);
 });
 
 // #7699/#7816 — heuristic is scoped to FORMATS.CLAUDE (/v1/messages); a

tests/unit/zed-hosted-stream-error-boundary.test.ts

@@ -1,84 +0,0 @@
-import assert from "node:assert/strict";
-import { spawn } from "node:child_process";
-import { fileURLToPath } from "node:url";
-import test from "node:test";
-
-const repoRoot = fileURLToPath(new URL("../../", import.meta.url));
-const fixturePath = fileURLToPath(
-  new URL("../fixtures/zed-hosted-stream-error-boundary-child.ts", import.meta.url)
-);
-
-type FixtureResult = {
-  code: number | null;
-  signal: NodeJS.Signals | null;
-  stdout: string;
-  stderr: string;
-};
-
-function runFixture(): Promise<FixtureResult> {
-  // Keep the parent process pristine: the fast unit suite can run files with
-  // --test-isolation=none, so all stateful imports and mutations live in the child.
-  const childEnv: NodeJS.ProcessEnv = {
-    PATH: process.env.PATH,
-    NODE_PATH: process.env.NODE_PATH,
-    LANG: process.env.LANG,
-    LC_ALL: process.env.LC_ALL,
-    TZ: process.env.TZ,
-    TMPDIR: process.env.TMPDIR,
-    NODE_ENV: "test",
-    API_KEY_SECRET: "zed-boundary-test-only-secret-with-32-plus-characters",
-    DISABLE_SQLITE_AUTO_BACKUP: "true",
-    NO_COLOR: "1",
-  };
-  // Inheriting this marker makes Node silently skip the nested --test run.
-  delete childEnv.NODE_TEST_CONTEXT;
-
-  return new Promise((resolve, reject) => {
-    const child = spawn(process.execPath, ["--import", "tsx/esm", "--test", fixturePath], {
-      cwd: repoRoot,
-      env: childEnv,
-      stdio: ["ignore", "pipe", "pipe"],
-    });
-    let stdout = "";
-    let stderr = "";
-    let timedOut = false;
-
-    child.stdout.setEncoding("utf8");
-    child.stderr.setEncoding("utf8");
-    child.stdout.on("data", (chunk: string) => {
-      stdout += chunk;
-    });
-    child.stderr.on("data", (chunk: string) => {
-      stderr += chunk;
-    });
-
-    const timeout = setTimeout(() => {
-      timedOut = true;
-      child.kill("SIGKILL");
-    }, 120_000);
-
-    child.once("error", (error) => {
-      clearTimeout(timeout);
-      reject(error);
-    });
-    child.once("close", (code, signal) => {
-      clearTimeout(timeout);
-      if (timedOut) {
-        reject(new Error("Zed stream error boundary fixture timed out after 120 seconds"));
-        return;
-      }
-      resolve({ code, signal, stdout, stderr });
-    });
-  });
-}
-
-test("Zed stream error boundary passes in a process-isolated runtime", async () => {
-  const result = await runFixture();
-  const output = `${result.stdout}\n${result.stderr}`;
-
-  assert.equal(result.signal, null, output.slice(-12_000));
-  assert.equal(result.code, 0, output.slice(-12_000));
-  assert.match(output, /(?:^|\s)tests\s+3(?:\s|$)/m);
-  assert.match(output, /(?:^|\s)pass\s+3(?:\s|$)/m);
-  assert.match(output, /(?:^|\s)fail\s+0(?:\s|$)/m);
-});