Merge remote-tracking branch 'origin/release/v3.8.51' into fix/v3851-zai-web-stream-error-boundary

CHANGELOG.md

@@ -97,10 +97,6 @@ _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-zai-web-stream-error-boundary.md

@@ -0,0 +1 @@
+- **Z.ai Web:** HTTP 200 streams carrying an upstream error now terminate with a structured failure instead of assistant text plus a normal stop, preserving partial output while allowing pre-content combo fallback.

open-sse/executors/zai-web/stream.ts

@@ -1,4 +1,4 @@
-import { sanitizeErrorMessage } from "../../utils/error.ts";
+import { buildErrorBody, sanitizeErrorMessage } from "../../utils/error.ts";
 
 export interface ZaiDelta {
   content: string;
@@ -113,22 +113,73 @@ function parseSsePayload(data: string): ZaiDelta | null {
   }
 }
 
+type ZaiDeltaSource = {
+  deltas: AsyncGenerator<ZaiDelta, void, void>;
+  cancel: (reason?: unknown) => void;
+};
+
+function createZaiDeltaSource(sourceBody: ReadableStream<Uint8Array>): ZaiDeltaSource {
+  const decoder = new TextDecoder();
+  const reader = sourceBody.getReader();
+  const buffer = { text: "" };
+  let upstreamDone = false;
+  let cancelRequested = false;
+  let readerReleased = false;
+
+  const releaseReader = () => {
+    if (readerReleased) return;
+    readerReleased = true;
+    try {
+      reader.releaseLock();
+    } catch {
+      // A concurrent read cancellation owns the final release.
+    }
+  };
+
+  const cancel = (reason?: unknown) => {
+    if (upstreamDone || cancelRequested) return;
+    cancelRequested = true;
+    try {
+      // Do not await an upstream cancel hook: a stalled provider is allowed to
+      // ignore cancellation, but it must never keep the client cancellation open.
+      void reader.cancel(reason).catch(() => {});
+    } catch {
+      // The reader may already have closed or released concurrently.
+    }
+  };
+
+  async function* iterate(): AsyncGenerator<ZaiDelta, void, void> {
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) {
+          upstreamDone = true;
+          return;
+        }
+        const payloads = extractSseDataPayloads(buffer, decoder.decode(value, { stream: true }));
+        for (const raw of payloads) {
+          const delta = parseSsePayload(raw);
+          if (delta) yield delta;
+        }
+      }
+    } finally {
+      if (!upstreamDone && !cancelRequested) cancel("Z.ai delta iteration ended");
+      releaseReader();
+    }
+  }
+
+  return { deltas: iterate(), cancel };
+}
+
 async function drainSseDeltas(
   sourceBody: ReadableStream<Uint8Array>,
   onDelta: (delta: ZaiDelta) => boolean
 ): Promise<boolean> {
-  const decoder = new TextDecoder();
-  const reader = sourceBody.getReader();
-  const buffer = { text: "" };
-  while (true) {
-    const { done, value } = await reader.read();
-    if (done) return false;
-    const payloads = extractSseDataPayloads(buffer, decoder.decode(value, { stream: true }));
-    for (const raw of payloads) {
-      const delta = parseSsePayload(raw);
-      if (delta && onDelta(delta)) return true;
-    }
+  const { deltas } = createZaiDeltaSource(sourceBody);
+  for await (const delta of deltas) {
+    if (onDelta(delta)) return true;
   }
+  return false;
 }
 
 function emitDeltaChunks(
@@ -137,17 +188,32 @@ function emitDeltaChunks(
   emitChunk: ZaiChunkEmitter,
   roleState: { emitted: boolean }
 ): boolean {
-  if (!roleState.emitted && (delta.content || delta.reasoning || delta.error)) {
+  if (delta.error) {
+    const errorBody = buildErrorBody(502, `Z.ai stream failed: ${delta.error}`, undefined, {
+      type: "upstream_error",
+      code: "zai_stream_error",
+    });
+
+    if (!roleState.emitted) {
+      // Keep a pre-content failure as an error-only Chat frame. Stream readiness
+      // rejects it before response headers are committed, so fallback receives a 502.
+      controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(errorBody)}\n\n`));
+      controller.close();
+    } else {
+      // Once content is public, error the protocol-neutral producer. The shared
+      // pipeline preserves prior chunks, records the failure, and emits the terminal
+      // error in the client's native Chat, Claude, or Responses wire format.
+      controller.error(Object.assign(new Error(errorBody.error.message), { statusCode: 502 }));
+    }
+    return true;
+  }
+
+  if (!roleState.emitted && (delta.content || delta.reasoning)) {
     roleState.emitted = true;
     emitChunk(controller, { role: "assistant", content: "" });
   }
   if (delta.reasoning) emitChunk(controller, { reasoning_content: delta.reasoning });
   if (delta.content) emitChunk(controller, { content: delta.content });
-  // Surfaced as visible content, matching the other web executors' mid-stream
-  // error convention (see zed-hosted's createErrorChunk): the 200 is already on
-  // the wire, so the status cannot change — but the caller must not be left
-  // reading an empty success. Any content streamed before the failure is kept.
-  if (delta.error) emitChunk(controller, { content: `[Z.ai error] ${delta.error}` });
   if (delta.done) {
     emitChunk(controller, {}, "stop");
     controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n"));
@@ -162,19 +228,32 @@ export function buildZaiStreamingBody(
   emitChunk: ZaiChunkEmitter,
   signal: AbortSignal | null | undefined
 ): ReadableStream {
+  const deltaSource = createZaiDeltaSource(sourceBody);
+  const { deltas } = deltaSource;
+  const roleState = { emitted: false };
+  let terminated = false;
+
   return new ReadableStream({
-    async start(controller) {
-      const roleState = { emitted: false };
+    async pull(controller) {
+      if (terminated) return;
       try {
-        const ended = await drainSseDeltas(sourceBody, (delta) =>
-          emitDeltaChunks(controller, delta, emitChunk, roleState)
-        );
-        if (ended) return;
+        const next = await deltas.next();
+        if (terminated) return;
+        if (next.done === false) {
+          if (emitDeltaChunks(controller, next.value, emitChunk, roleState)) {
+            terminated = true;
+            await deltas.return(undefined);
+          }
+          return;
+        }
+
+        terminated = true;
         if (!roleState.emitted) emitChunk(controller, { role: "assistant", content: "" });
         emitChunk(controller, {}, "stop");
         controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n"));
         controller.close();
       } catch (error) {
+        terminated = true;
         if (!signal?.aborted) {
           try {
             controller.error(error);
@@ -184,6 +263,11 @@ export function buildZaiStreamingBody(
         }
       }
     },
+    cancel(reason) {
+      terminated = true;
+      deltaSource.cancel(reason);
+      void deltas.return(undefined).catch(() => {});
+    },
   });
 }
 

open-sse/utils/streamHandler.ts

@@ -1,7 +1,6 @@
 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";
@@ -188,10 +187,6 @@ 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 &&
@@ -411,7 +406,7 @@ export function createStreamController({
       }
 
       if (error instanceof Error) {
-        logStream(`error: ${getPublicErrorMessage(error.message, getErrorStatusCode(error))}`);
+        logStream(`error: ${error.message}`);
         return;
       }
       logStream("error: unknown");
@@ -457,7 +452,6 @@ export function buildStreamErrorChunks(
   clientResponseFormat?: string | null
 ) {
   const statusMapping = getStreamErrorStatusMapping(statusCode);
-  const publicErrorMessage = getPublicErrorMessage(errorMsg, statusCode);
 
   if (isResponsesClientFormat(clientResponseFormat)) {
     const errorEvent = {
@@ -466,7 +460,7 @@ export function buildStreamErrorChunks(
         id: null,
         status: "failed",
         error: {
-          message: publicErrorMessage,
+          message: errorMsg,
           type: statusMapping.responses.type,
           code: statusMapping.responses.code,
         },
@@ -481,7 +475,7 @@ export function buildStreamErrorChunks(
       type: "error",
       error: {
         type: statusMapping.claude.type,
-        message: publicErrorMessage,
+        message: errorMsg,
       },
     };
 
@@ -504,7 +498,7 @@ export function buildStreamErrorChunks(
       },
     ],
     error: {
-      message: publicErrorMessage,
+      message: errorMsg,
       type: statusMapping.responses.type,
       code: statusMapping.responses.code,
     },

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

@@ -1,211 +0,0 @@
-// 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/zai-web-stream-error-boundary.fixture.ts

@@ -0,0 +1,253 @@
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import test from "node:test";
+
+assert.ok(process.env.DATA_DIR, "the parent wrapper must provide a synthetic DATA_DIR");
+assert.ok(
+  process.env.OMNIROUTE_PLUGINS_DIR,
+  "the parent wrapper must provide a synthetic plugin directory"
+);
+assert.ok(process.env.API_KEY_SECRET, "the parent wrapper must provide a synthetic API secret");
+
+fs.mkdirSync(process.env.DATA_DIR, { recursive: true });
+fs.mkdirSync(process.env.OMNIROUTE_PLUGINS_DIR, { recursive: true });
+
+const [
+  { buildZaiStreamingBody },
+  { ensureStreamReadiness },
+  { createSSEStream },
+  { createStreamController, pipeWithDisconnect },
+  { createStreamFailureFinalizers },
+  { FORMATS },
+  dbCore,
+  { closeSharedLoggerResource },
+] = await Promise.all([
+  import("../../open-sse/executors/zai-web/stream.ts"),
+  import("../../open-sse/utils/streamReadiness.ts"),
+  import("../../open-sse/utils/stream.ts"),
+  import("../../open-sse/utils/streamHandler.ts"),
+  import("../../open-sse/utils/streamFailureFinalization.ts"),
+  import("../../open-sse/translator/formats.ts"),
+  import("../../src/lib/db/core.ts"),
+  import("../../src/shared/utils/loggerResource.ts"),
+]);
+
+test.after(async () => {
+  await closeSharedLoggerResource();
+  dbCore.resetDbInstance();
+});
+
+const encoder = new TextEncoder();
+
+function upstreamSse(...payloads: Record<string, unknown>[]): ReadableStream<Uint8Array> {
+  return new ReadableStream({
+    start(controller) {
+      for (const payload of payloads) {
+        controller.enqueue(encoder.encode(`data: ${JSON.stringify(payload)}\n\n`));
+      }
+      controller.close();
+    },
+  });
+}
+
+function emitOpenAiChunk(
+  controller: ReadableStreamDefaultController,
+  delta: Record<string, unknown>,
+  finish: string | null = null
+): void {
+  const chunk = {
+    id: "chatcmpl-zai-test",
+    object: "chat.completion.chunk",
+    created: 1,
+    model: "glm-5.2",
+    choices: [{ index: 0, delta, finish_reason: finish }],
+  };
+  controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`));
+}
+
+type PipelineResult = {
+  output: string;
+  completions: Array<{ status: number; errorCode?: string | null; error?: string | null }>;
+  persisted: Array<{ status: number; errorCode?: string }>;
+  failures: Array<{ status: number; message: string; code?: string; type?: string }>;
+};
+
+function jsonDataPayloads(output: string): Array<Record<string, unknown>> {
+  return output
+    .split(/\r?\n/)
+    .filter((line) => line.startsWith("data: ") && line !== "data: [DONE]")
+    .map((line) => JSON.parse(line.slice(6)) as Record<string, unknown>);
+}
+
+async function runPartialFailurePipeline(clientResponseFormat: string): Promise<PipelineResult> {
+  const completions: PipelineResult["completions"] = [];
+  const persisted: PipelineResult["persisted"] = [];
+  const failures: PipelineResult["failures"] = [];
+  const { onPipelineStreamError } = createStreamFailureFinalizers({
+    isFailureCompletionRecorded: () => false,
+    isStreamCompletionRecorded: () => false,
+    onStreamComplete(payload) {
+      completions.push({
+        status: payload.status,
+        errorCode: payload.errorCode,
+        error: payload.error,
+      });
+    },
+    persistFailureUsage(status, errorCode) {
+      persisted.push({ status, errorCode });
+    },
+    onStreamFailure(failure) {
+      failures.push(failure);
+    },
+  });
+
+  const zaiStream = buildZaiStreamingBody(
+    upstreamSse(
+      {
+        type: "chat:completion",
+        data: { delta_content: "partial answer", phase: "answer" },
+      },
+      { error: { message: "stream aborted upstream" } }
+    ),
+    emitOpenAiChunk,
+    null
+  );
+  const passthrough = clientResponseFormat === FORMATS.OPENAI;
+  const transform = createSSEStream({
+    mode: passthrough ? "passthrough" : "translate",
+    targetFormat: FORMATS.OPENAI,
+    sourceFormat: passthrough ? FORMATS.OPENAI : clientResponseFormat,
+    clientResponseFormat,
+    provider: "zai-web",
+    model: "glm-5.2",
+    body: { messages: [{ role: "user", content: "hello" }] },
+  });
+  const streamController = createStreamController({
+    provider: "zai-web",
+    model: "glm-5.2",
+    clientResponseFormat,
+    onError: onPipelineStreamError,
+  });
+  const output = await new Response(
+    pipeWithDisconnect(
+      new Response(zaiStream, { headers: { "Content-Type": "text/event-stream" } }),
+      transform,
+      streamController,
+      { stallTimeoutMs: 0 }
+    )
+  ).text();
+
+  return { output, completions, persisted, failures };
+}
+
+function assertFailureWasPersisted(result: PipelineResult): void {
+  assert.deepEqual(result.completions, [
+    {
+      status: 502,
+      errorCode: "stream_pipeline_error",
+      error: "Z.ai stream failed: stream aborted upstream",
+    },
+  ]);
+  assert.deepEqual(result.persisted, [{ status: 502, errorCode: "stream_pipeline_error" }]);
+  assert.deepEqual(result.failures, [
+    {
+      status: 502,
+      message: "Z.ai stream failed: stream aborted upstream",
+      code: "stream_pipeline_error",
+      type: "stream_error",
+    },
+  ]);
+}
+
+test("a pre-content Z.ai error fails stream readiness with a sanitized 502", async () => {
+  const rawFailure =
+    'signature invalid at /srv/omniroute/open-sse/auth.ts:17:9 api_key="sk-private"\n' +
+    "    at verify (/srv/omniroute/open-sse/auth.ts:17:9)";
+  const stream = buildZaiStreamingBody(
+    upstreamSse({ error: { detail: rawFailure } }),
+    emitOpenAiChunk,
+    null
+  );
+
+  const readiness = await ensureStreamReadiness(
+    new Response(stream, { headers: { "Content-Type": "text/event-stream" } }),
+    { timeoutMs: 100, provider: "zai-web", model: "glm-5.2" }
+  );
+
+  if (readiness.ok) {
+    await readiness.response.body?.cancel();
+    assert.fail("an error-only Z.ai stream must not be accepted as ready model output");
+  }
+
+  assert.equal(readiness.response.status, 502);
+  assert.equal(readiness.code, "STREAM_EARLY_EOF");
+  assert.match(readiness.upstreamDiagnostic ?? "", /Z\.ai stream failed: signature invalid/);
+
+  const publicBody = JSON.stringify(await readiness.response.json());
+  assert.doesNotMatch(publicBody, /sk-private|\/srv\/omniroute|auth\.ts/);
+  assert.match(publicBody, /<path>/);
+});
+
+test("a partial Z.ai failure stays strict Chat and persists as pipeline failure", async () => {
+  const result = await runPartialFailurePipeline(FORMATS.OPENAI);
+  const payloads = jsonDataPayloads(result.output);
+  const terminal = payloads.find((payload) => "error" in payload);
+
+  assert.match(result.output, /partial answer/, "content before the failure is preserved");
+  assert.ok(terminal, "the Chat client receives a terminal structured error chunk");
+  assert.equal(terminal.object, "chat.completion.chunk");
+  assert.deepEqual(Object.keys(terminal).sort(), ["choices", "error", "object"]);
+  assert.deepEqual(terminal.choices, [{ index: 0, delta: {}, finish_reason: "error" }]);
+  assert.deepEqual(terminal.error, {
+    message: "Z.ai stream failed: stream aborted upstream",
+    type: "server_error",
+    code: "server_error",
+  });
+  assert.match(result.output, /data: \[DONE\]/);
+  assert.doesNotMatch(result.output, /response\.failed|event: response\.failed/);
+  assert.doesNotMatch(result.output, /"finish_reason":"stop"/);
+  assertFailureWasPersisted(result);
+});
+
+test("a partial Z.ai failure is translated to Claude and persists as failure", async () => {
+  const result = await runPartialFailurePipeline(FORMATS.CLAUDE);
+
+  assert.match(result.output, /partial answer/, "translated partial content is preserved");
+  assert.match(result.output, /event: error\r?\n/);
+  assert.match(result.output, /"type":"error"/);
+  assert.match(result.output, /"message":"Z\.ai stream failed: stream aborted upstream"/);
+  assert.match(result.output, /event: message_stop\r?\n/);
+  assert.doesNotMatch(result.output, /response\.failed|event: response\.failed/);
+  assert.doesNotMatch(result.output, /finish_reason|data: \[DONE\]/);
+  assertFailureWasPersisted(result);
+});
+
+test("client cancellation stays non-blocking and cancels a stalled Z.ai body", async () => {
+  let markPullStarted: (() => void) | null = null;
+  const pullStarted = new Promise<void>((resolve) => {
+    markPullStarted = resolve;
+  });
+  let upstreamCancelCalls = 0;
+  const stalledUpstream = new ReadableStream<Uint8Array>({
+    pull() {
+      markPullStarted?.();
+      return new Promise<void>(() => {});
+    },
+    cancel() {
+      upstreamCancelCalls += 1;
+      return new Promise<void>(() => {});
+    },
+  });
+  const reader = buildZaiStreamingBody(stalledUpstream, emitOpenAiChunk, null).getReader();
+  const pendingRead = reader.read();
+  void pendingRead.catch(() => {});
+  await pullStarted;
+
+  const outcome = await Promise.race([
+    reader.cancel("client closed").then(() => "resolved" as const),
+    new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 100)),
+  ]);
+
+  assert.equal(outcome, "resolved", "consumer cancel cannot wait for a stalled upstream body");
+  assert.equal(upstreamCancelCalls, 1, "the locked upstream reader receives one cancel request");
+});

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

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

tests/unit/zai-web-silent-empty-repro.test.ts

@@ -1,9 +1,8 @@
 import test from "node:test";
 import assert from "node:assert/strict";
 
-const { buildZaiStreamingBody, parseZaiFrame, collectZaiNonStreaming } = await import(
-  "../../open-sse/executors/zai-web/stream.ts"
-);
+const { buildZaiStreamingBody, parseZaiFrame, collectZaiNonStreaming } =
+  await import("../../open-sse/executors/zai-web/stream.ts");
 
 /**
  * Hard Rule #6 — "never silently swallow errors in SSE streams".
@@ -47,6 +46,23 @@ async function readAll(stream: ReadableStream): Promise<string> {
   return out;
 }
 
+async function readUntilError(stream: ReadableStream): Promise<{ output: string; error: unknown }> {
+  const reader = stream.getReader();
+  const decoder = new TextDecoder();
+  let output = "";
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) return { output, error: null };
+      output += decoder.decode(value as Uint8Array, { stream: true });
+    }
+  } catch (error) {
+    return { output, error };
+  } finally {
+    reader.releaseLock();
+  }
+}
+
 const emitChunk = (
   controller: ReadableStreamDefaultController,
   delta: Record<string, unknown>,
@@ -61,6 +77,14 @@ const emitChunk = (
 const contentOf = (sse: string) =>
   [...sse.matchAll(/"content":"([^"]*)"/g)].map((m) => m[1]).join("");
 
+function errorPayloads(sse: string): Array<Record<string, unknown>> {
+  return sse
+    .split(/\r?\n/)
+    .filter((line) => line.startsWith("data: ") && line !== "data: [DONE]")
+    .map((line) => JSON.parse(line.slice(6)) as Record<string, unknown>)
+    .filter((payload) => "error" in payload);
+}
+
 test("parseZaiFrame classifies an error-shaped frame instead of discarding it", () => {
   assert.equal(parseZaiFrame({ error: "captcha expired" })?.error, "captcha expired");
   assert.equal(
@@ -88,25 +112,42 @@ test("REGRESSION GUARD: contentless frames are still skipped, not reported as er
   assert.equal(parseZaiFrame("not-an-object"), null);
 });
 
-test("a 200 stream carrying an error frame surfaces it instead of finishing empty", async () => {
+test("a 200 stream carrying an error frame emits a terminal error instead of false success", async () => {
   const upstream = sseStream(JSON.stringify({ error: { detail: "signature invalid" } }));
   const out = await readAll(buildZaiStreamingBody(upstream, emitChunk, null));
 
-  assert.match(contentOf(out), /signature invalid/, "the upstream's diagnosis must reach the caller");
-  assert.match(contentOf(out), /\[Z\.ai error\]/, "tagged like the other web executors");
-  assert.ok(out.includes('"finish_reason":"stop"'));
-  assert.ok(out.includes("[DONE]"), "the stream still terminates cleanly for the client");
+  assert.equal(contentOf(out), "", "an upstream failure must not become assistant content");
+  assert.deepEqual(errorPayloads(out), [
+    {
+      error: {
+        message: "Z.ai stream failed: signature invalid",
+        type: "upstream_error",
+        code: "zai_stream_error",
+      },
+    },
+  ]);
+  assert.ok(!out.includes("response.failed"), "Chat streams cannot emit Responses events");
+  assert.ok(!out.includes('"finish_reason":"stop"'), "a failure must not report a normal stop");
+  assert.ok(!out.includes("[DONE]"), "readiness must see an error-only pre-content stream");
 });
 
-test("an error frame after partial content still surfaces, keeping what was streamed", async () => {
+test("an error after partial content preserves it, then errors the producer stream", async () => {
   const upstream = sseStream(
-    JSON.stringify({ type: "chat:completion", data: { delta_content: "partial", phase: "answer" } }),
+    JSON.stringify({
+      type: "chat:completion",
+      data: { delta_content: "partial", phase: "answer" },
+    }),
     JSON.stringify({ error: "stream aborted upstream" })
   );
-  const out = await readAll(buildZaiStreamingBody(upstream, emitChunk, null));
+  const { output, error } = await readUntilError(buildZaiStreamingBody(upstream, emitChunk, null));
 
-  assert.match(contentOf(out), /partial/, "already-streamed content is preserved");
-  assert.match(contentOf(out), /stream aborted upstream/, "and the failure is appended, not dropped");
+  assert.match(contentOf(output), /partial/, "already-streamed content is preserved");
+  assert.match(String(error), /Z\.ai stream failed: stream aborted upstream/);
+  assert.ok(!output.includes("response.failed"), "the producer stays protocol-neutral");
+  assert.ok(
+    !output.includes('"finish_reason":"stop"'),
+    "partial output does not make failure success"
+  );
 });
 
 test("control: a well-formed stream is untouched", async () => {

tests/unit/zai-web-stream-error-boundary.test.ts

@@ -0,0 +1,43 @@
+import assert from "node:assert/strict";
+import { spawnSync } from "node:child_process";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import test from "node:test";
+import { fileURLToPath } from "node:url";
+
+const fixture = fileURLToPath(
+  new URL("../fixtures/zai-web-stream-error-boundary.fixture.ts", import.meta.url)
+);
+
+test("Z.ai stream error boundaries pass in a process-isolated fixture", () => {
+  const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-zai-stream-boundary-"));
+  const childEnv: NodeJS.ProcessEnv = {
+    API_KEY_SECRET: "zai-stream-boundary-test-only-secret",
+    DATA_DIR: path.join(testRoot, "data"),
+    OMNIROUTE_PLUGINS_DIR: path.join(testRoot, "plugins"),
+  };
+
+  // The parent itself is a node:test process. Never forward its runner identity to the child;
+  // `node --test` owns the child context and creates a fresh value for its fixture process.
+  delete childEnv.NODE_TEST_CONTEXT;
+
+  try {
+    const result = spawnSync(process.execPath, ["--import", "tsx/esm", "--test", fixture], {
+      cwd: fileURLToPath(new URL("../..", import.meta.url)),
+      encoding: "utf8",
+      env: childEnv,
+      timeout: 60_000,
+    });
+    const diagnostics = [result.stdout, result.stderr].filter(Boolean).join("\n");
+
+    assert.equal(result.error, undefined, diagnostics);
+    assert.equal(result.signal, null, diagnostics);
+    assert.equal(result.status, 0, diagnostics);
+    assert.match(result.stdout, /tests 4\b/);
+    assert.match(result.stdout, /pass 4\b/);
+    assert.match(result.stdout, /fail 0\b/);
+  } finally {
+    fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+  }
+});