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/PENDING-perplexity-stream-error-boundary.md

@@ -1 +0,0 @@
-- **fix(providers):** Perplexity Web no longer turns upstream stream failures into successful assistant text; pre-content failures remain eligible for fallback, partial output ends with a structured sanitized error, and failed sessions are not persisted

open-sse/executors/perplexity-web.ts

@@ -17,7 +17,6 @@ import { prepareToolMessages } from "../translator/webTools.ts";
 import { buildToolModeResponse } from "./chatgptWebTools.ts";
 import { sanitizeErrorMessage } from "../utils/error.ts";
 import { buildSessionCookieHeader, mergeRefreshedCookie } from "../utils/nextAuthCookie.ts";
-import { formatTranslatedStreamError } from "../utils/streamErrorFormat.ts";
 import {
   PPLX_SSE_ENDPOINT,
   PPLX_USER_AGENT,
@@ -36,8 +35,6 @@ import {
 
 const SESSION_MAX_AGE_MS = 3600_000;
 const SESSION_MAX_ENTRIES = 200;
-const PPLX_STREAM_ERROR_MESSAGE = "Perplexity upstream stream failed";
-const PPLX_STREAM_ERROR_CODE = "PPLX_STREAM_ERROR";
 
 interface SessionEntry {
   backendUuid: string;
@@ -105,162 +102,36 @@ function buildStreamingResponse(
   signal?: AbortSignal | null
 ): ReadableStream<Uint8Array> {
   const encoder = new TextEncoder();
-  const streamAbortController = new AbortController();
-  const forwardInputAbort = () =>
-    streamAbortController.abort(signal?.reason ?? "perplexity_request_aborted");
-  if (signal?.aborted) forwardInputAbort();
-  else signal?.addEventListener("abort", forwardInputAbort, { once: true });
-  let inputAbortListenerAttached = Boolean(signal && !signal.aborted);
-  const removeInputAbortListener = () => {
-    if (!inputAbortListenerAttached) return;
-    inputAbortListenerAttached = false;
-    signal?.removeEventListener("abort", forwardInputAbort);
-  };
-  const abortEventStream = (reason: unknown) => {
-    removeInputAbortListener();
-    if (!streamAbortController.signal.aborted) streamAbortController.abort(reason);
-  };
-  const contentIterator = extractContent(eventStream, streamAbortController.signal)[
-    Symbol.asyncIterator
-  ]();
-  let fullAnswer = "";
-  let respBackendUuid: string | null = null;
-  let roleEmitted = false;
-  let finished = false;
-  let pendingFailure: (Error & { statusCode: number }) | null = null;
 
-  const enqueuePreContentFailure = (controller: ReadableStreamDefaultController<Uint8Array>) => {
-    controller.enqueue(
-      encoder.encode(
-        formatTranslatedStreamError({
-          status: 502,
-          message: PPLX_STREAM_ERROR_MESSAGE,
-          type: "upstream_error",
-          code: PPLX_STREAM_ERROR_CODE,
-        })
-      )
-    );
-  };
-
-  const takeAssistantRoleChunk = (): string => {
-    if (roleEmitted) return "";
-    roleEmitted = true;
-    return sseChunk({
-      id: cid,
-      object: "chat.completion.chunk",
-      created,
-      model,
-      system_fingerprint: null,
-      choices: [
-        {
-          index: 0,
-          delta: { role: "assistant" },
-          finish_reason: null,
-          logprobs: null,
-        },
-      ],
-    });
-  };
-
-  const completeStream = (controller: ReadableStreamDefaultController<Uint8Array>) => {
-    if (finished) return;
-    finished = true;
-    controller.enqueue(
-      encoder.encode(
-        sseChunk({
-          id: cid,
-          object: "chat.completion.chunk",
-          created,
-          model,
-          system_fingerprint: null,
-          choices: [{ index: 0, delta: {}, finish_reason: "stop", logprobs: null }],
-        })
-      )
-    );
-    controller.enqueue(encoder.encode("data: [DONE]\n\n"));
-    sessionStore(history, currentMsg, cleanResponse(fullAnswer), respBackendUuid);
-    removeInputAbortListener();
-    controller.close();
-  };
-
-  const failStream = (controller: ReadableStreamDefaultController<Uint8Array>) => {
-    if (roleEmitted) {
-      pendingFailure = Object.assign(new Error(PPLX_STREAM_ERROR_MESSAGE), {
-        statusCode: 502,
-      });
-      finished = true;
-      controller.close();
-      return;
-    }
-    finished = true;
-    enqueuePreContentFailure(controller);
-    controller.close();
-  };
-
-  const providerStream = new ReadableStream<Uint8Array>({
-    async pull(controller) {
-      if (finished) return;
-
-      try {
-        const next = await contentIterator.next();
-        if (streamAbortController.signal.aborted) {
-          finished = true;
-          controller.close();
-          return;
-        }
-        if (next.done === true) {
-          completeStream(controller);
-          return;
-        }
-
-        const chunk = next.value;
-        if (chunk.backendUuid) respBackendUuid = chunk.backendUuid;
-
-        if (chunk.error) {
-          failStream(controller);
-          removeInputAbortListener();
-          void contentIterator.return?.(undefined).catch(() => undefined);
-          return;
-        }
-
-        if (chunk.thinking) {
+  return new ReadableStream(
+    {
+      async start(controller) {
+        try {
+          // Initial role chunk
           controller.enqueue(
             encoder.encode(
-              takeAssistantRoleChunk() +
-                sseChunk({
-                  id: cid,
-                  object: "chat.completion.chunk",
-                  created,
-                  model,
-                  system_fingerprint: null,
-                  choices: [
-                    {
-                      index: 0,
-                      delta: { reasoning_content: chunk.thinking + "\n" },
-                      finish_reason: null,
-                      logprobs: null,
-                    },
-                  ],
-                })
+              sseChunk({
+                id: cid,
+                object: "chat.completion.chunk",
+                created,
+                model,
+                system_fingerprint: null,
+                choices: [
+                  { index: 0, delta: { role: "assistant" }, finish_reason: null, logprobs: null },
+                ],
+              })
             )
           );
-          return;
-        }
 
-        if (chunk.done) {
-          fullAnswer = chunk.answer || fullAnswer;
-          completeStream(controller);
-          await contentIterator.return?.(undefined);
-          return;
-        }
+          let fullAnswer = "";
+          let respBackendUuid: string | null = null;
 
-        let dt = chunk.delta || "";
-        if (dt) {
-          dt = cleanResponse(dt, false);
-          if (dt) {
-            controller.enqueue(
-              encoder.encode(
-                takeAssistantRoleChunk() +
+          for await (const chunk of extractContent(eventStream, signal)) {
+            if (chunk.backendUuid) respBackendUuid = chunk.backendUuid;
+
+            if (chunk.error) {
+              controller.enqueue(
+                encoder.encode(
                   sseChunk({
                     id: cid,
                     object: "chat.completion.chunk",
@@ -268,60 +139,118 @@ function buildStreamingResponse(
                     model,
                     system_fingerprint: null,
                     choices: [
-                      { index: 0, delta: { content: dt }, finish_reason: null, logprobs: null },
+                      {
+                        index: 0,
+                        delta: { content: `[Error: ${chunk.error}]` },
+                        finish_reason: null,
+                        logprobs: null,
+                      },
                     ],
                   })
-              )
-            );
+                )
+              );
+              break;
+            }
+
+            if (chunk.thinking) {
+              controller.enqueue(
+                encoder.encode(
+                  sseChunk({
+                    id: cid,
+                    object: "chat.completion.chunk",
+                    created,
+                    model,
+                    system_fingerprint: null,
+                    choices: [
+                      {
+                        index: 0,
+                        delta: { reasoning_content: chunk.thinking + "\n" },
+                        finish_reason: null,
+                        logprobs: null,
+                      },
+                    ],
+                  })
+                )
+              );
+              continue;
+            }
+
+            if (chunk.done) {
+              fullAnswer = chunk.answer || fullAnswer;
+              break;
+            }
+
+            let dt = chunk.delta || "";
+            if (dt) {
+              dt = cleanResponse(dt, false);
+              if (dt) {
+                controller.enqueue(
+                  encoder.encode(
+                    sseChunk({
+                      id: cid,
+                      object: "chat.completion.chunk",
+                      created,
+                      model,
+                      system_fingerprint: null,
+                      choices: [
+                        { index: 0, delta: { content: dt }, finish_reason: null, logprobs: null },
+                      ],
+                    })
+                  )
+                );
+              }
+            }
+            if (chunk.answer) fullAnswer = chunk.answer;
           }
-        }
-        if (chunk.answer) fullAnswer = chunk.answer;
-      } catch {
-        failStream(controller);
-        removeInputAbortListener();
-        void contentIterator.return?.(undefined).catch(() => undefined);
-      }
-    },
 
-    cancel(reason) {
-      finished = true;
-      abortEventStream(reason);
-      void contentIterator.return?.(undefined).catch(() => undefined);
-    },
-  });
+          // Stop chunk
+          controller.enqueue(
+            encoder.encode(
+              sseChunk({
+                id: cid,
+                object: "chat.completion.chunk",
+                created,
+                model,
+                system_fingerprint: null,
+                choices: [{ index: 0, delta: {}, finish_reason: "stop", logprobs: null }],
+              })
+            )
+          );
+          controller.enqueue(encoder.encode("data: [DONE]\n\n"));
 
-  // Erroring the provider stream immediately would discard output buffered by the readiness
-  // handoff. Drain each provider chunk through a backpressure-aware reader, then reject only the
-  // read after the last legitimate chunk. The outer pipeline converts that fixed public error to
-  // the client's canonical terminal frame and records the stream failure.
-  const providerReader = providerStream.getReader();
-  let cancelled = false;
-  return new ReadableStream<Uint8Array>({
-    async pull(controller) {
-      try {
-        const next = await providerReader.read();
-        if (cancelled) return;
-        if (next.done === false) {
-          controller.enqueue(next.value);
-          return;
+          sessionStore(history, currentMsg, cleanResponse(fullAnswer), respBackendUuid);
+        } catch (err) {
+          controller.enqueue(
+            encoder.encode(
+              sseChunk({
+                id: cid,
+                object: "chat.completion.chunk",
+                created,
+                model,
+                system_fingerprint: null,
+                choices: [
+                  {
+                    index: 0,
+                    delta: {
+                      content: `[Stream error: ${err instanceof Error ? err.message : String(err)}]`,
+                    },
+                    finish_reason: "stop",
+                    logprobs: null,
+                  },
+                ],
+              })
+            )
+          );
+          controller.enqueue(encoder.encode("data: [DONE]\n\n"));
+        } finally {
+          try {
+            controller.close();
+          } catch {}
         }
-        if (pendingFailure) {
-          controller.error(pendingFailure);
-          return;
-        }
-        controller.close();
-      } catch (error) {
-        if (!cancelled) controller.error(error);
-      }
+      },
     },
-
-    cancel(reason) {
-      if (cancelled) return;
-      cancelled = true;
-      abortEventStream(reason);
-      void providerReader.cancel(reason).catch(() => undefined);
-    },
-  });
+    { highWaterMark: 16384 }
+  );
 }
 
 async function buildNonStreamingResponse(

open-sse/executors/perplexity-web/protocol.ts

@@ -213,19 +213,6 @@ export async function* readPplxSseEvents(
   const decoder = new TextDecoder();
   let buffer = "";
   let dataLines: string[] = [];
-  let readerFinished = false;
-  let readerCancelRequested = false;
-
-  const cancelReader = (reason: unknown) => {
-    if (readerFinished || readerCancelRequested) return;
-    readerCancelRequested = true;
-    // Cancellation is a client-facing latency boundary. Request upstream cleanup once, but never
-    // await a hostile underlying source whose cancel hook does not settle.
-    void reader.cancel(reason).catch(() => undefined);
-  };
-  const handleAbort = () => cancelReader(signal?.reason ?? "perplexity_stream_aborted");
-  if (signal?.aborted) handleAbort();
-  else signal?.addEventListener("abort", handleAbort, { once: true });
 
   function flush(): PplxStreamEvent | null | "done" {
     if (dataLines.length === 0) return null;
@@ -244,10 +231,7 @@ export async function* readPplxSseEvents(
     while (true) {
       if (signal?.aborted) return;
       const { value, done } = await reader.read();
-      if (done) {
-        readerFinished = true;
-        break;
-      }
+      if (done) break;
       buffer += decoder.decode(value, { stream: true });
 
       while (true) {
@@ -279,13 +263,7 @@ export async function* readPplxSseEvents(
     const tail = flush();
     if (tail && tail !== "done") yield tail;
   } finally {
-    signal?.removeEventListener("abort", handleAbort);
-    cancelReader(signal?.reason ?? "perplexity_stream_reader_closed");
-    try {
-      reader.releaseLock();
-    } catch {
-      // A hostile source may keep its cancel promise pending; the lock can be released later by GC.
-    }
+    reader.releaseLock();
   }
 }
 
@@ -937,10 +915,6 @@ export async function* extractContent(
     }
   }
 
-  // Cancellation is not a successful terminal event. In particular, do not synthesize the final
-  // `done` chunk: streaming callers use that signal to emit stop/[DONE] and persist the session.
-  if (signal?.aborted) return;
-
   // End-of-stream without a COMPLETED frame still try the last text blob.
   if (!fullAnswer.trim() && lastEventText) {
     const fromText = extractAnswerFromFinalText(lastEventText);

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,
     },

open-sse/utils/streamReadiness.ts

@@ -421,57 +421,29 @@ function prependBufferedChunks(
   chunks: Uint8Array[],
   reader: ReadableStreamDefaultReader<Uint8Array>
 ): ReadableStream<Uint8Array> {
-  let bufferedIndex = 0;
-  let cancelled = false;
-  let readerReleased = false;
-
-  const releaseReader = () => {
-    if (readerReleased) return;
-    readerReleased = true;
-    try {
-      reader.releaseLock();
-    } catch {
-      // A hostile source can keep a read/cancel pending forever. The public stream must
-      // remain cancellable even when its abandoned source cannot release immediately.
-    }
-  };
-
   return new ReadableStream<Uint8Array>({
-    async pull(controller) {
-      if (cancelled) return;
-
-      // Replay exactly one readiness chunk per pull. Keeping the first buffered chunk at
-      // the stream's default high-water mark prevents an eager read of a later upstream
-      // failure from discarding that legitimate prefix before the caller attaches.
-      if (bufferedIndex < chunks.length) {
-        controller.enqueue(chunks[bufferedIndex]);
-        bufferedIndex += 1;
-        return;
-      }
-
+    async start(controller) {
       try {
-        const { done, value } = await reader.read();
-        if (cancelled) return;
-        if (done) {
-          releaseReader();
-          controller.close();
-          return;
+        for (const chunk of chunks) {
+          controller.enqueue(chunk);
         }
-        if (value) controller.enqueue(value);
+
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) break;
+          if (value) controller.enqueue(value);
+        }
+
+        controller.close();
       } catch (error) {
-        releaseReader();
-        if (!cancelled) controller.error(error);
+        controller.error(error);
+      } finally {
+        reader.releaseLock();
       }
     },
-    cancel(reason) {
-      if (cancelled) return;
-      cancelled = true;
-      // Do not await a provider's cancel hook: a hostile or stalled source must not make
-      // downstream cancellation hang. Release the lock once cancellation actually settles.
-      void reader
-        .cancel(reason)
-        .catch(() => {})
-        .finally(releaseReader);
+    async cancel(reason) {
+      await reader.cancel(reason).catch(() => {});
+      reader.releaseLock();
     },
   });
 }

tests/fixtures/perplexity-web-stream-error-boundary.fixture.ts

@@ -1,648 +0,0 @@
-import assert from "node:assert/strict";
-import test from "node:test";
-
-assert.ok(process.env.DATA_DIR, "the subprocess fixture requires an isolated DATA_DIR");
-assert.ok(
-  process.env.OMNIROUTE_PLUGINS_DIR,
-  "the subprocess fixture requires an isolated OMNIROUTE_PLUGINS_DIR"
-);
-
-const core = await import("../../src/lib/db/core.ts");
-const { getUsageHistory } = await import("../../src/lib/usage/usageHistory.ts");
-const { waitForCallLogSaves } = await import("../../src/lib/usage/callLogs.ts");
-const { closeCallLogArtifactWriter } = await import("../../src/lib/usage/callLogArtifactWriter.ts");
-const { PerplexityWebExecutor } = await import("../../open-sse/executors/perplexity-web.ts");
-const { __setTlsFetchOverrideForTesting } =
-  await import("../../open-sse/services/perplexityTlsClient.ts");
-const { handleChatCore } = await import("../../open-sse/handlers/chatCore.ts");
-
-type StreamFailure = { status: number; message: string; code?: string; type?: string };
-
-function createPerplexityStream(
-  events: Array<Record<string, unknown>>
-): ReadableStream<Uint8Array> {
-  const encoder = new TextEncoder();
-  const payload =
-    events.map((event) => `event: message\r\ndata: ${JSON.stringify(event)}\r\n\r\n`).join("") +
-    "event: end_of_stream\r\n\r\n";
-  return new ReadableStream<Uint8Array>({
-    start(controller) {
-      controller.enqueue(encoder.encode(payload));
-      controller.close();
-    },
-  });
-}
-
-async function executeWithUpstreamBody(
-  body: ReadableStream<Uint8Array>,
-  prompt = "hi"
-): Promise<Response> {
-  __setTlsFetchOverrideForTesting(async () => ({
-    status: 200,
-    headers: new Headers({ "Content-Type": "text/event-stream" }),
-    text: null,
-    body,
-  }));
-
-  const executor = new PerplexityWebExecutor();
-  const result = await executor.execute({
-    model: "pplx-auto",
-    body: { messages: [{ role: "user", content: prompt }], stream: true },
-    stream: true,
-    credentials: { apiKey: "test-cookie" },
-    signal: AbortSignal.timeout(10_000),
-    log: null,
-  });
-  return result.response;
-}
-
-function executeStreaming(
-  events: Array<Record<string, unknown>>,
-  prompt = "hi"
-): Promise<Response> {
-  return executeWithUpstreamBody(createPerplexityStream(events), prompt);
-}
-
-async function executeThroughChatCore(
-  events: Array<Record<string, unknown>>,
-  prompt = "hi",
-  onStreamFailure?: (failure: StreamFailure) => void,
-  onRequestSuccess?: () => Promise<void>,
-  model = "pplx-auto"
-) {
-  return executeBodyThroughChatCore(
-    createPerplexityStream(events),
-    prompt,
-    onStreamFailure,
-    onRequestSuccess,
-    model
-  );
-}
-
-async function executeBodyThroughChatCore(
-  upstreamBody: ReadableStream<Uint8Array>,
-  prompt = "hi",
-  onStreamFailure?: (failure: StreamFailure) => void,
-  onRequestSuccess?: () => Promise<void>,
-  model = "pplx-auto"
-) {
-  __setTlsFetchOverrideForTesting(async () => ({
-    status: 200,
-    headers: new Headers({ "Content-Type": "text/event-stream" }),
-    text: null,
-    body: upstreamBody,
-  }));
-  const body = {
-    model,
-    messages: [{ role: "user", content: prompt }],
-    stream: true,
-  };
-  return handleChatCore({
-    body: structuredClone(body),
-    modelInfo: { provider: "perplexity-web", model, extendedContext: false },
-    credentials: { apiKey: "test-cookie", providerSpecificData: {} },
-    log: { debug() {}, info() {}, warn() {}, error() {} },
-    onRequestSuccess,
-    onStreamFailure,
-    clientRawRequest: {
-      endpoint: "/v1/chat/completions",
-      body: structuredClone(body),
-      headers: new Headers({ accept: "text/event-stream" }),
-    },
-    userAgent: "perplexity-stream-error-boundary-test",
-    skipResourcePressureGuard: true,
-  });
-}
-
-function assertNoSensitiveDetail(value: string): void {
-  assert.doesNotMatch(value, /private-runtime\.ts/);
-  assert.doesNotMatch(value, /sk-pplx-secret/);
-  assert.doesNotMatch(value, /api_key/);
-}
-
-function assertChatCompletionWire(value: string): void {
-  assert.doesNotMatch(value, /^event:/m, "Chat Completions must not receive Responses framing");
-  const payloads = value
-    .split(/\r?\n/)
-    .filter((line) => line.startsWith("data:") && line.slice(5).trim() !== "[DONE]")
-    .map((line) => JSON.parse(line.slice(5).trim()) as Record<string, unknown>);
-  assert.ok(payloads.length > 0);
-  for (const payload of payloads) {
-    assert.equal(payload.object, "chat.completion.chunk");
-    assert.ok(Array.isArray(payload.choices));
-    for (const choice of payload.choices as Array<Record<string, unknown>>) {
-      assert.equal(choice.index, 0);
-      assert.equal(typeof choice.delta, "object");
-      assert.ok(
-        choice.finish_reason === null || typeof choice.finish_reason === "string",
-        "finish_reason must remain Chat Completions-compatible"
-      );
-    }
-  }
-  assert.match(value, /data: \[DONE\]/);
-}
-
-async function waitForPersistedStreamFailure(startedAt: Date, model = "pplx-auto") {
-  for (let attempt = 0; attempt < 80; attempt++) {
-    const rows = await getUsageHistory({
-      provider: "perplexity-web",
-      model,
-      startDate: startedAt,
-    });
-    const failure = rows.find(
-      (row) =>
-        row.success === false && row.status === "502" && row.errorCode === "stream_pipeline_error"
-    );
-    if (failure) return failure;
-    await new Promise((resolve) => setTimeout(resolve, 25));
-  }
-  return null;
-}
-
-async function waitForCondition(predicate: () => boolean, timeoutMs = 500): Promise<boolean> {
-  const deadline = Date.now() + timeoutMs;
-  while (Date.now() < deadline) {
-    if (predicate()) return true;
-    await new Promise((resolve) => setTimeout(resolve, 10));
-  }
-  return predicate();
-}
-
-test.afterEach(() => {
-  __setTlsFetchOverrideForTesting(null);
-});
-
-test.after(async () => {
-  __setTlsFetchOverrideForTesting(null);
-  assert.equal(
-    await waitForCallLogSaves(3_000),
-    true,
-    "call-log writes must drain before the isolated DATA_DIR is removed"
-  );
-  await closeCallLogArtifactWriter();
-  core.resetDbInstance();
-});
-
-test("pre-content Perplexity failures remain unready and return a sanitized 502", async () => {
-  const result = await executeThroughChatCore([
-    {
-      error_code: "PPLX_ERROR",
-      error_message:
-        "failed at /srv/omniroute/private-runtime.ts:42:7 token=sk-pplx-secret-123456 api_key=hidden",
-    },
-  ]);
-
-  assert.equal(result.success, false, "the handler must expose a fallback-eligible failure");
-  assert.equal(result.status, 502);
-  assert.equal(result.response.status, 502);
-  assert.equal(result.errorCode, "STREAM_EARLY_EOF");
-  const body = await result.response.text();
-  assert.match(body, /"error"/);
-  assertNoSensitiveDetail(body);
-});
-
-test("thrown Perplexity stream failures remain unready and return a sanitized 502", async () => {
-  const result = await executeBodyThroughChatCore(
-    new ReadableStream<Uint8Array>({
-      start(controller) {
-        controller.error(
-          new Error(
-            "socket failed at /srv/omniroute/private-runtime.ts:51:9 token=sk-pplx-secret-catch"
-          )
-        );
-      },
-    }),
-    "throw before content"
-  );
-
-  assert.equal(result.success, false, "the handler must expose a fallback-eligible failure");
-  assert.equal(result.status, 502);
-  assert.equal(result.response.status, 502);
-  const responseBody = await result.response.text();
-  assert.match(responseBody, /"error"/);
-  assertNoSensitiveDetail(responseBody);
-});
-
-test("thrown failures after content preserve the prefix and terminate as a safe error", async () => {
-  const encoder = new TextEncoder();
-  const partialAnswer = "partial before transport failure";
-  let upstreamRead = false;
-  let finalizedFailure: StreamFailure | null = null;
-  const upstreamBody = new ReadableStream<Uint8Array>({
-    pull(controller) {
-      if (!upstreamRead) {
-        upstreamRead = true;
-        controller.enqueue(
-          encoder.encode(
-            `event: message\r\ndata: ${JSON.stringify({
-              backend_uuid: "uuid-thrown-must-not-store",
-              blocks: [
-                {
-                  intended_usage: "markdown",
-                  markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" },
-                },
-              ],
-              status: "PENDING",
-            })}\r\n\r\n`
-          )
-        );
-        return;
-      }
-      controller.error(
-        new Error(
-          "transport failed at /srv/omniroute/private-runtime.ts:79 token=sk-pplx-secret-after"
-        )
-      );
-    },
-  });
-
-  const result = await executeBodyThroughChatCore(
-    upstreamBody,
-    "post-content transport failure",
-    (failure) => {
-      finalizedFailure = failure;
-    }
-  );
-
-  assert.equal(result.success, true);
-  const output = await result.response.text();
-  assert.match(output, new RegExp(partialAnswer));
-  assert.match(output, /"finish_reason":"error"/);
-  assert.doesNotMatch(output, /"finish_reason":"stop"/);
-  assert.doesNotMatch(output, /response\.failed/);
-  assertNoSensitiveDetail(output);
-  assertChatCompletionWire(output);
-  assert.ok(finalizedFailure);
-  assert.equal(finalizedFailure.status, 502);
-  assert.equal(finalizedFailure.message, "Perplexity upstream stream failed");
-});
-
-test("partial content is preserved before a terminal error and the failed session is not stored", async () => {
-  const firstPrompt = "partial-boundary-first-prompt";
-  const partialAnswer = "safe partial answer";
-  const requestStartedAt = new Date(Date.now() - 1_000);
-  let finalizedFailure: StreamFailure | null = null;
-  const firstResult = await executeThroughChatCore(
-    [
-      {
-        backend_uuid: "uuid-must-not-be-stored",
-        blocks: [
-          {
-            intended_usage: "markdown",
-            markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" },
-          },
-        ],
-        status: "PENDING",
-      },
-      {
-        error_code: "PPLX_ERROR",
-        error_message:
-          "later failure at /srv/omniroute/private-runtime.ts:66:2 token=sk-pplx-secret-partial",
-      },
-    ],
-    firstPrompt,
-    (failure) => {
-      finalizedFailure = failure;
-    }
-  );
-
-  assert.equal(firstResult.success, true, "legitimate partial output must satisfy readiness");
-  const output = await firstResult.response.text();
-  assert.match(output, /"role":"assistant"/);
-  assert.match(output, new RegExp(partialAnswer));
-  assert.match(output, /"error":\{/);
-  assert.match(output, /"finish_reason":"error"/);
-  assert.doesNotMatch(output, /"finish_reason":"stop"/);
-  assert.doesNotMatch(output, /response\.failed/);
-  assert.doesNotMatch(output, /event:\s*response\.failed/);
-  assert.doesNotMatch(output, /"type":"response\.failed"/);
-  assert.doesNotMatch(output, /\[Error:/);
-  assertNoSensitiveDetail(output);
-  assertChatCompletionWire(output);
-  assert.ok(finalizedFailure, "the downstream pipeline must finalize the stream failure");
-  assert.equal(finalizedFailure.status, 502);
-  assert.equal(finalizedFailure.message, "Perplexity upstream stream failed");
-  const persistedFailure = await waitForPersistedStreamFailure(requestStartedAt);
-  assert.ok(persistedFailure, "the handler must persist the terminal stream failure");
-
-  let followUpRequestBody: string | undefined;
-  __setTlsFetchOverrideForTesting(async (_url, options) => {
-    followUpRequestBody = String(options.body ?? "");
-    return {
-      status: 200,
-      headers: new Headers({ "Content-Type": "text/event-stream" }),
-      text: null,
-      body: createPerplexityStream([
-        {
-          backend_uuid: "next-success-uuid",
-          blocks: [
-            {
-              intended_usage: "markdown",
-              markdown_block: { chunks: ["next answer"], progress: "DONE" },
-            },
-          ],
-          status: "COMPLETED",
-        },
-      ]),
-    };
-  });
-
-  const executor = new PerplexityWebExecutor();
-  const followUp = await executor.execute({
-    model: "pplx-auto",
-    body: {
-      messages: [
-        { role: "user", content: firstPrompt },
-        { role: "assistant", content: partialAnswer },
-        { role: "user", content: "continue" },
-      ],
-      stream: false,
-    },
-    stream: false,
-    credentials: { apiKey: "test-cookie" },
-    signal: AbortSignal.timeout(10_000),
-    log: null,
-  });
-  assert.equal(followUp.response.status, 200);
-  assert.ok(followUpRequestBody);
-  const sent = JSON.parse(followUpRequestBody) as { params?: Record<string, unknown> };
-  assert.equal(
-    sent.params?.last_backend_uuid,
-    undefined,
-    "a failed partial response must not create a reusable session"
-  );
-});
-
-test("same-packet content and error preserve the prefix across repeated readiness handoffs", async () => {
-  for (let attempt = 0; attempt < 8; attempt++) {
-    const partialAnswer = `same-packet partial ${attempt}`;
-    let finalizedFailure: StreamFailure | null = null;
-    const result = await executeThroughChatCore(
-      [
-        {
-          backend_uuid: `uuid-same-packet-${attempt}`,
-          blocks: [
-            {
-              intended_usage: "markdown",
-              markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" },
-            },
-          ],
-          status: "PENDING",
-        },
-        {
-          error_code: "PPLX_ERROR",
-          error_message: `same packet private failure ${attempt} token=sk-pplx-secret-repeat`,
-        },
-      ],
-      `same-packet prompt ${attempt}`,
-      (failure) => {
-        finalizedFailure = failure;
-      }
-    );
-
-    assert.equal(result.success, true);
-    const output = await result.response.text();
-    assert.match(output, new RegExp(partialAnswer));
-    assert.match(output, /"finish_reason":"error"/);
-    assert.doesNotMatch(output, /"finish_reason":"stop"/);
-    assert.doesNotMatch(output, /response\.failed/);
-    assertNoSensitiveDetail(output);
-    assertChatCompletionWire(output);
-    assert.ok(finalizedFailure);
-    assert.equal(finalizedFailure.status, 502);
-  }
-});
-
-test("a delayed success hook cannot erase same-packet content before terminal failure", async () => {
-  const partialAnswer = "prefix must survive delayed success bookkeeping";
-  const model = "pplx-auto-delayed-success-proof";
-  const requestStartedAt = new Date(Date.now() - 1_000);
-  let finalizedFailure: StreamFailure | null = null;
-  const result = await executeThroughChatCore(
-    [
-      {
-        backend_uuid: "uuid-delayed-success-hook-must-not-store",
-        blocks: [
-          {
-            intended_usage: "markdown",
-            markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" },
-          },
-        ],
-        status: "PENDING",
-      },
-      {
-        error_code: "PPLX_ERROR",
-        error_message: "private same-packet failure token=sk-pplx-secret-delayed-hook",
-      },
-    ],
-    "delayed success hook prompt",
-    (failure) => {
-      finalizedFailure = failure;
-    },
-    async () => {
-      await new Promise((resolve) => setTimeout(resolve, 25));
-    },
-    model
-  );
-
-  assert.equal(result.success, true, "the legitimate prefix must satisfy readiness");
-  const output = await result.response.text();
-  assert.match(output, new RegExp(partialAnswer));
-  assert.match(output, /"finish_reason":"error"/);
-  assert.doesNotMatch(output, /"finish_reason":"stop"/);
-  assert.doesNotMatch(output, /response\.failed/);
-  assertNoSensitiveDetail(output);
-  assertChatCompletionWire(output);
-  assert.ok(finalizedFailure);
-  assert.equal(finalizedFailure.status, 502);
-  assert.equal(finalizedFailure.message, "Perplexity upstream stream failed");
-  assert.ok(
-    await waitForPersistedStreamFailure(requestStartedAt, model),
-    "the delayed handoff must still persist the terminal stream failure"
-  );
-});
-
-test("successful streamed completions still store their Perplexity session", async () => {
-  const firstPrompt = "successful-session-first-prompt";
-  const firstAnswer = "successful session answer";
-  const firstResponse = await executeStreaming(
-    [
-      {
-        backend_uuid: "uuid-success-is-stored",
-        blocks: [
-          {
-            intended_usage: "markdown",
-            markdown_block: { chunks: [firstAnswer], progress: "DONE" },
-          },
-        ],
-        status: "COMPLETED",
-      },
-    ],
-    firstPrompt
-  );
-  const firstOutput = await firstResponse.text();
-  assert.match(firstOutput, new RegExp(firstAnswer));
-  assert.match(firstOutput, /"finish_reason":"stop"/);
-
-  let followUpRequestBody: string | undefined;
-  __setTlsFetchOverrideForTesting(async (_url, options) => {
-    followUpRequestBody = String(options.body ?? "");
-    return {
-      status: 200,
-      headers: new Headers({ "Content-Type": "text/event-stream" }),
-      text: null,
-      body: createPerplexityStream([
-        {
-          backend_uuid: "uuid-next-success",
-          blocks: [
-            {
-              intended_usage: "markdown",
-              markdown_block: { chunks: ["continued"], progress: "DONE" },
-            },
-          ],
-          status: "COMPLETED",
-        },
-      ]),
-    };
-  });
-
-  const executor = new PerplexityWebExecutor();
-  const followUp = await executor.execute({
-    model: "pplx-auto",
-    body: {
-      messages: [
-        { role: "user", content: firstPrompt },
-        { role: "assistant", content: firstAnswer },
-        { role: "user", content: "continue successful session" },
-      ],
-      stream: false,
-    },
-    stream: false,
-    credentials: { apiKey: "test-cookie" },
-    signal: AbortSignal.timeout(10_000),
-    log: null,
-  });
-  assert.equal(followUp.response.status, 200);
-  assert.ok(followUpRequestBody);
-  const sent = JSON.parse(followUpRequestBody) as { params?: Record<string, unknown> };
-  assert.equal(sent.params?.last_backend_uuid, "uuid-success-is-stored");
-});
-
-test("downstream cancellation reaches a stalled Perplexity reader exactly once", async () => {
-  const encoder = new TextEncoder();
-  const partialAnswer = "cancel after this prefix";
-  let firstPull = true;
-  let upstreamCancelCount = 0;
-  let finalizedFailure: StreamFailure | null = null;
-  const upstreamBody = new ReadableStream<Uint8Array>({
-    pull(controller) {
-      if (firstPull) {
-        firstPull = false;
-        controller.enqueue(
-          encoder.encode(
-            `event: message\r\ndata: ${JSON.stringify({
-              backend_uuid: "uuid-cancel-must-not-store",
-              blocks: [
-                {
-                  intended_usage: "markdown",
-                  markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" },
-                },
-              ],
-              status: "PENDING",
-            })}\r\n\r\n`
-          )
-        );
-        return;
-      }
-      return new Promise<void>(() => {});
-    },
-    cancel() {
-      upstreamCancelCount += 1;
-      return new Promise<void>(() => {});
-    },
-  });
-
-  const result = await executeBodyThroughChatCore(
-    upstreamBody,
-    "cancel stalled stream",
-    (failure) => {
-      finalizedFailure = failure;
-    }
-  );
-  assert.equal(result.success, true);
-  assert.ok(result.response.body);
-  const reader = result.response.body.getReader();
-  const decoder = new TextDecoder();
-  let prefix = "";
-  for (let readCount = 0; readCount < 4 && !prefix.includes(partialAnswer); readCount += 1) {
-    const next = await Promise.race([
-      reader.read(),
-      new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 500)),
-    ]);
-    if (next === "timeout" || next.done) break;
-    prefix += decoder.decode(next.value, { stream: true });
-  }
-
-  const cancelResult = await Promise.race([
-    reader.cancel("client stopped reading").then(() => "settled"),
-    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 500)),
-  ]);
-  assert.match(prefix, new RegExp(partialAnswer));
-  assert.doesNotMatch(prefix, /"finish_reason":"stop"/);
-  assert.doesNotMatch(prefix, /data: \[DONE\]/);
-  assert.equal(cancelResult, "settled", "client cancellation must not await a hostile upstream");
-  assert.equal(
-    await waitForCondition(() => upstreamCancelCount === 1),
-    true,
-    "cancellation must reach the real upstream reader"
-  );
-  assert.equal(upstreamCancelCount, 1);
-  await new Promise((resolve) => setTimeout(resolve, 25));
-  assert.equal(finalizedFailure, null, "client cancellation must not finalize as provider failure");
-
-  let followUpRequestBody: string | undefined;
-  __setTlsFetchOverrideForTesting(async (_url, options) => {
-    followUpRequestBody = String(options.body ?? "");
-    return {
-      status: 200,
-      headers: new Headers({ "Content-Type": "text/event-stream" }),
-      text: null,
-      body: createPerplexityStream([
-        {
-          backend_uuid: "uuid-after-cancel",
-          blocks: [
-            {
-              intended_usage: "markdown",
-              markdown_block: { chunks: ["answer after cancel"], progress: "DONE" },
-            },
-          ],
-          status: "COMPLETED",
-        },
-      ]),
-    };
-  });
-  const executor = new PerplexityWebExecutor();
-  const followUp = await executor.execute({
-    model: "pplx-auto",
-    body: {
-      messages: [
-        { role: "user", content: "cancel stalled stream" },
-        { role: "assistant", content: partialAnswer },
-        { role: "user", content: "continue after cancellation" },
-      ],
-      stream: false,
-    },
-    stream: false,
-    credentials: { apiKey: "test-cookie" },
-    signal: AbortSignal.timeout(10_000),
-    log: null,
-  });
-  assert.equal(followUp.response.status, 200);
-  assert.ok(followUpRequestBody);
-  const sent = JSON.parse(followUpRequestBody) as { params?: Record<string, unknown> };
-  assert.equal(
-    sent.params?.last_backend_uuid,
-    undefined,
-    "a cancelled response must not create a reusable session"
-  );
-});

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/unit/perplexity-web-stream-error-boundary.test.ts

@@ -1,90 +0,0 @@
-import assert from "node:assert/strict";
-import { spawn } from "node:child_process";
-import fs from "node:fs";
-import os from "node:os";
-import path from "node:path";
-import { fileURLToPath } from "node:url";
-import test from "node:test";
-
-const FIXTURE = fileURLToPath(
-  new URL("../fixtures/perplexity-web-stream-error-boundary.fixture.ts", import.meta.url)
-);
-
-type FixtureResult = {
-  code: number | null;
-  signal: NodeJS.Signals | null;
-  stdout: string;
-  stderr: string;
-};
-
-function runIsolatedFixture(testRoot: string): Promise<FixtureResult> {
-  const dataDir = path.join(testRoot, "data");
-  const pluginsDir = path.join(testRoot, "plugins");
-  fs.mkdirSync(dataDir, { recursive: true });
-  fs.mkdirSync(pluginsDir, { recursive: true });
-  const childEnv: NodeJS.ProcessEnv = {
-    PATH: process.env.PATH,
-    NODE_PATH: process.env.NODE_PATH,
-    LANG: process.env.LANG ?? "C.UTF-8",
-    LC_ALL: process.env.LC_ALL,
-    TZ: process.env.TZ ?? "UTC",
-    TMPDIR: process.env.TMPDIR ?? os.tmpdir(),
-    NODE_ENV: "test",
-    APP_LOG_TO_FILE: "false",
-    API_KEY_SECRET: "perplexity-stream-boundary-test-secret-00000000000000000000000000000000",
-    DATA_DIR: dataDir,
-    OMNIROUTE_PLUGINS_DIR: pluginsDir,
-  };
-  delete childEnv.NODE_TEST_CONTEXT;
-
-  return new Promise((resolve, reject) => {
-    const child = spawn(process.execPath, ["--import", "tsx/esm", "--test", FIXTURE], {
-      cwd: process.cwd(),
-      env: childEnv,
-      stdio: ["ignore", "pipe", "pipe"],
-      timeout: 90_000,
-    });
-    let stdout = "";
-    let stderr = "";
-    child.stdout.setEncoding("utf8").on("data", (chunk: string) => {
-      stdout += chunk;
-    });
-    child.stderr.setEncoding("utf8").on("data", (chunk: string) => {
-      stderr += chunk;
-    });
-    child.once("error", reject);
-    child.once("close", (code, signal) => resolve({ code, signal, stdout, stderr }));
-  });
-}
-
-test("Perplexity stream failures preserve protocol semantics in an isolated full pipeline", async () => {
-  const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-pplx-boundary-parent-"));
-  const originalDataDir = process.env.DATA_DIR;
-  const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR;
-  const eventBusOwner = globalThis as { __omnirouteEventBus?: unknown };
-  const originalEventBus = eventBusOwner.__omnirouteEventBus;
-
-  try {
-    const result = await runIsolatedFixture(testRoot);
-    assert.equal(
-      result.code,
-      0,
-      `isolated Perplexity fixture failed (signal=${result.signal ?? "none"})\n` +
-        `stdout:\n${result.stdout}\nstderr:\n${result.stderr}`
-    );
-    assert.equal(result.signal, null);
-    assert.match(result.stdout, /ℹ tests 8/);
-    assert.match(result.stdout, /ℹ pass 8/);
-    assert.match(result.stdout, /ℹ fail 0/);
-
-    assert.equal(process.env.DATA_DIR, originalDataDir);
-    assert.equal(process.env.OMNIROUTE_PLUGINS_DIR, originalPluginsDir);
-    assert.equal(
-      eventBusOwner.__omnirouteEventBus,
-      originalEventBus,
-      "the subprocess fixture must not replace the parent event bus singleton"
-    );
-  } finally {
-    fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
-  }
-});

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/stream-readiness.test.ts

@@ -451,43 +451,6 @@ test("ensureStreamReadiness preserves buffered chunks when stream starts", async
   assert.match(text, / world/);
 });
 
-test("ensureStreamReadiness preserves its buffered prefix until a delayed consumer observes a later error", async () => {
-  const prefix = `data: ${JSON.stringify({
-    object: "chat.completion.chunk",
-    choices: [
-      {
-        index: 0,
-        delta: { role: "assistant", content: "prefix before failure" },
-        finish_reason: null,
-      },
-    ],
-  })}\n\n`;
-  let pullCount = 0;
-  const response = new Response(
-    new ReadableStream<Uint8Array>({
-      pull(controller) {
-        pullCount += 1;
-        if (pullCount === 1) {
-          controller.enqueue(encoder.encode(prefix));
-          return;
-        }
-        controller.error(new Error("later upstream failure"));
-      },
-    }),
-    { status: 200, headers: { "Content-Type": "text/event-stream" } }
-  );
-
-  const result = await ensureStreamReadiness(response, { timeoutMs: 100 });
-  assert.equal(result.ok, true);
-  await new Promise((resolve) => setTimeout(resolve, 25));
-
-  const reader = result.response.body!.getReader();
-  const first = await reader.read();
-  assert.equal(first.done, false);
-  assert.match(new TextDecoder().decode(first.value), /prefix before failure/);
-  await assert.rejects(() => reader.read(), /later upstream failure/);
-});
-
 test("ensureStreamReadiness honors configured timeouts above 2000ms", async () => {
   const response = new Response(
     streamFromChunks(
@@ -653,7 +616,10 @@ test("ensureStreamReadiness preserves sanitized error-only diagnostics on early
   assert.equal(result.response.status, 502);
   assert.equal(result.code, "STREAM_EARLY_EOF");
   assert.equal(result.type, "stream_early_eof");
-  assert.equal(result.classificationReason, "Stream ended before producing a non-ping SSE event");
+  assert.equal(
+    result.classificationReason,
+    "Stream ended before producing a non-ping SSE event"
+  );
   assert.equal(
     result.upstreamDiagnostic,
     "UPSTREAM_DETAIL quota exhausted; retry after 2s; empty content Bearer [REDACTED] <path>"
@@ -670,9 +636,16 @@ test("ensureStreamReadiness preserves sanitized error-only diagnostics on early
   assert.equal(body.upstream_details.error.message, result.upstreamDiagnostic);
   assert.equal(warnings.length, 1);
 
-  for (const surfaced of [result.reason, body.upstream_details.error.message, warnings[0]]) {
+  for (const surfaced of [
+    result.reason,
+    body.upstream_details.error.message,
+    warnings[0],
+  ]) {
     assert.match(surfaced, /UPSTREAM_DETAIL/);
-    assert.doesNotMatch(surfaced, /SECOND_DETAIL|TOP_SECRET|\/srv\/omniroute\/handler\.ts/);
+    assert.doesNotMatch(
+      surfaced,
+      /SECOND_DETAIL|TOP_SECRET|\/srv\/omniroute\/handler\.ts/
+    );
   }
 });