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

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(() => {});
+    },
   });
 }
 

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/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 });
+  }
+});