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

changelog.d/fixes/pending-grok-web-stream-error-boundary.md

@@ -0,0 +1,4 @@
+- **fix(grok-web):** treat upstream streaming failures as failures instead of successful
+  assistant text: error-only streams now fail readiness with HTTP 502, while failures after
+  legitimate content preserve that partial output and terminate through the sanitized stream
+  failure path without a normal `stop` completion.

open-sse/executors/grok-web.ts

@@ -19,7 +19,7 @@ import {
   type ExecuteInput,
   type ExecutorLog,
 } from "./base.ts";
-import { FETCH_TIMEOUT_MS } from "../config/constants.ts";
+import { FETCH_TIMEOUT_MS, STREAM_READINESS_TIMEOUT_MS } from "../config/constants.ts";
 import { buildGrokCookieHeader } from "@/lib/providers/webCookieAuth";
 import {
   tlsFetchGrok,
@@ -27,7 +27,8 @@ import {
   isCloudflareChallenge,
   type TlsFetchResult,
 } from "../services/grokTlsClient.ts";
-import { sanitizeErrorMessage } from "../utils/error.ts";
+import { buildErrorBody, sanitizeErrorMessage } from "../utils/error.ts";
+import { ensureStreamReadiness } from "../utils/streamReadiness.ts";
 import {
   shouldUseGrokBrowserBacked,
   acquireFreshGrokClearance,
@@ -119,12 +120,29 @@ async function* readGrokNdjsonEvents(
   const reader = body.getReader();
   const decoder = new TextDecoder();
   let buffer = "";
+  let reachedEnd = false;
+  let cancelRequested = false;
+
+  const requestReaderCancel = (reason?: unknown) => {
+    if (cancelRequested || reachedEnd) return;
+    cancelRequested = true;
+    // Cancellation must release the upstream promptly even when a provider's
+    // underlying cancel promise never settles.
+    void reader.cancel(reason).catch(() => {});
+  };
+  const handleAbort = () => requestReaderCancel(signal?.reason);
+
+  if (signal?.aborted) requestReaderCancel(signal.reason);
+  else signal?.addEventListener("abort", handleAbort, { once: true });
 
   try {
     while (true) {
       if (signal?.aborted) return;
       const { value, done } = await reader.read();
-      if (done) break;
+      if (done) {
+        reachedEnd = true;
+        break;
+      }
       buffer += decoder.decode(value, { stream: true });
 
       while (true) {
@@ -142,6 +160,8 @@ async function* readGrokNdjsonEvents(
       }
     }
 
+    if (signal?.aborted) return;
+
     // Flush remaining buffer
     buffer += decoder.decode();
     const remaining = buffer.trim();
@@ -153,7 +173,11 @@ async function* readGrokNdjsonEvents(
       }
     }
   } finally {
-    reader.releaseLock();
+    signal?.removeEventListener("abort", handleAbort);
+    if (!reachedEnd) requestReaderCancel(signal?.reason ?? "Grok stream reader closed early");
+    try {
+      reader.releaseLock();
+    } catch {}
   }
 }
 
@@ -271,6 +295,8 @@ async function* extractContent(
     }
   }
 
+  if (signal?.aborted) return;
+
   const trailingThinking =
     suppressThinkingAfterVisibleContent && emittedVisibleContent ? "" : thinkingFilter.flush();
   if (trailingThinking) {
@@ -292,6 +318,25 @@ function sseChunk(data: unknown): string {
   return `data: ${JSON.stringify(data)}\n\n`;
 }
 
+const GROK_STREAM_FAILURE_MESSAGE = "Grok upstream stream failed";
+const GROK_STREAM_FAILURE_CODE = "GROK_STREAM_ERROR";
+
+function grokStreamErrorChunk(): string {
+  return sseChunk(
+    buildErrorBody(502, GROK_STREAM_FAILURE_MESSAGE, undefined, {
+      type: "upstream_error",
+      code: GROK_STREAM_FAILURE_CODE,
+    })
+  );
+}
+
+function grokStreamFailure(): Error & { statusCode: number; code: string } {
+  return Object.assign(new Error(GROK_STREAM_FAILURE_MESSAGE), {
+    statusCode: 502,
+    code: GROK_STREAM_FAILURE_CODE,
+  });
+}
+
 function enqueueStreamingToolCalls(
   controller: ReadableStreamDefaultController<Uint8Array>,
   encoder: TextEncoder,
@@ -349,63 +394,77 @@ function buildStreamingResponse(
   signal?: AbortSignal | null
 ): ReadableStream<Uint8Array> {
   const encoder = new TextEncoder();
+  const streamAbortController = new AbortController();
+  const requestStreamCancel = (reason?: unknown) => {
+    if (!streamAbortController.signal.aborted) streamAbortController.abort(reason);
+  };
+  const handleParentAbort = () => requestStreamCancel(signal?.reason);
+
+  if (signal?.aborted) requestStreamCancel(signal.reason);
+  else signal?.addEventListener("abort", handleParentAbort, { once: true });
 
   return new ReadableStream(
     {
       async start(controller) {
+        let roleSent = false;
+        let firstOutputHandedOff = false;
         try {
-          // Initial role chunk
-          controller.enqueue(
-            encoder.encode(
-              sseChunk({
-                id: cid,
-                object: "chat.completion.chunk",
-                created,
-                model,
-                system_fingerprint: null,
-                choices: [
-                  { index: 0, delta: { role: "assistant" }, finish_reason: null, logprobs: null },
-                ],
-              })
-            )
-          );
-
           let fp = "";
           let buffered = "";
 
+          const enqueueRole = () => {
+            if (roleSent) return;
+            controller.enqueue(
+              encoder.encode(
+                sseChunk({
+                  id: cid,
+                  object: "chat.completion.chunk",
+                  created,
+                  model,
+                  system_fingerprint: fp || null,
+                  choices: [
+                    {
+                      index: 0,
+                      delta: { role: "assistant" },
+                      finish_reason: null,
+                      logprobs: null,
+                    },
+                  ],
+                })
+              )
+            );
+            roleSent = true;
+          };
+
+          const handOffFirstOutput = async () => {
+            if (firstOutputHandedOff) return;
+            firstOutputHandedOff = true;
+            // Give readiness/finalization wrappers one turn to attach before a later
+            // upstream failure errors the stream and invalidates queued chunks.
+            await new Promise<void>((resolve) => setImmediate(resolve));
+          };
+
           for await (const chunk of extractContent(
             eventStream,
             isThinkingModel,
             toolRegistry,
-            signal,
+            streamAbortController.signal,
             true
           )) {
             if (chunk.fingerprint) fp = chunk.fingerprint;
 
             if (chunk.error) {
-              controller.enqueue(
-                encoder.encode(
-                  sseChunk({
-                    id: cid,
-                    object: "chat.completion.chunk",
-                    created,
-                    model,
-                    system_fingerprint: fp || null,
-                    choices: [
-                      {
-                        index: 0,
-                        delta: { content: `[Error: ${chunk.error}]` },
-                        finish_reason: null,
-                        logprobs: null,
-                      },
-                    ],
-                  })
-                )
-              );
-              break;
+              if (roleSent) {
+                controller.error(grokStreamFailure());
+                return;
+              }
+              controller.enqueue(encoder.encode(grokStreamErrorChunk()));
+              controller.enqueue(encoder.encode("data: [DONE]\n\n"));
+              return;
             }
 
             if (chunk.thinking) {
+              enqueueRole();
               controller.enqueue(
                 encoder.encode(
                   sseChunk({
@@ -425,10 +484,12 @@ function buildStreamingResponse(
                   })
                 )
               );
+              await handOffFirstOutput();
               continue;
             }
 
             if (chunk.toolCalls) {
+              enqueueRole();
               enqueueStreamingToolCalls(controller, encoder, {
                 id: cid,
                 created,
@@ -444,6 +505,7 @@ function buildStreamingResponse(
             if (chunk.fullMessage) {
               const toolCalls = parseClientToolCallMarkup(chunk.fullMessage, toolRegistry);
               if (toolCalls) {
+                enqueueRole();
                 enqueueStreamingToolCalls(controller, encoder, {
                   id: cid,
                   created,
@@ -453,6 +515,30 @@ function buildStreamingResponse(
                 });
                 return;
               }
+              if (!buffered) {
+                enqueueRole();
+                buffered = chunk.fullMessage;
+                controller.enqueue(
+                  encoder.encode(
+                    sseChunk({
+                      id: cid,
+                      object: "chat.completion.chunk",
+                      created,
+                      model,
+                      system_fingerprint: fp || null,
+                      choices: [
+                        {
+                          index: 0,
+                          delta: { content: chunk.fullMessage },
+                          finish_reason: null,
+                          logprobs: null,
+                        },
+                      ],
+                    })
+                  )
+                );
+                await handOffFirstOutput();
+              }
             }
 
             if (chunk.delta) {
@@ -469,6 +555,7 @@ function buildStreamingResponse(
                 return;
               }
               if (hasOpenToolCallMarkup(buffered)) continue;
+              enqueueRole();
               controller.enqueue(
                 encoder.encode(
                   sseChunk({
@@ -488,10 +575,13 @@ function buildStreamingResponse(
                   })
                 )
               );
+              await handOffFirstOutput();
             }
           }
 
-          // Stop chunk
+          if (streamAbortController.signal.aborted || !roleSent) return;
+
+          // Stop chunk — only after legitimate content/reasoning/tool output.
           controller.enqueue(
             encoder.encode(
               sseChunk({
@@ -505,37 +595,24 @@ function buildStreamingResponse(
             )
           );
           controller.enqueue(encoder.encode("data: [DONE]\n\n"));
-        } catch (err) {
-          controller.enqueue(
-            encoder.encode(
-              sseChunk({
-                id: cid,
-                object: "chat.completion.chunk",
-                created,
-                model,
-                system_fingerprint: null,
-                choices: [
-                  {
-                    index: 0,
-                    delta: {
-                      content: sanitizeErrorMessage(
-                        `[Stream error: ${err instanceof Error ? err.message : String(err)}]`
-                      ),
-                    },
-                    finish_reason: "stop",
-                    logprobs: null,
-                  },
-                ],
-              })
-            )
-          );
+        } catch {
+          if (streamAbortController.signal.aborted) return;
+          if (roleSent) {
+            controller.error(grokStreamFailure());
+            return;
+          }
+          controller.enqueue(encoder.encode(grokStreamErrorChunk()));
           controller.enqueue(encoder.encode("data: [DONE]\n\n"));
         } finally {
+          signal?.removeEventListener("abort", handleParentAbort);
           try {
             controller.close();
           } catch {}
         }
       },
+      cancel(reason) {
+        requestStreamCancel(reason);
+      },
     },
     { highWaterMark: 16384 }
   );
@@ -1026,6 +1103,13 @@ export class GrokWebExecutor extends BaseExecutor {
           "X-Accel-Buffering": "no",
         },
       });
+      const readiness = await ensureStreamReadiness(finalResponse, {
+        timeoutMs: STREAM_READINESS_TIMEOUT_MS,
+        provider: this.provider,
+        model,
+        log,
+      });
+      finalResponse = readiness.response;
     } else {
       finalResponse = await buildNonStreamingResponse(
         tlsResult.body,

tests/fixtures/grok-web-stream-error-boundary-child.ts

@@ -0,0 +1,517 @@
+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";
+
+// This file is executed only by the process-isolated unit-test wrapper. State
+// mutations and repository imports must remain here, never in the parent test.
+const originalDataDir = process.env.DATA_DIR;
+const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR;
+const originalFetch = globalThis.fetch;
+const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-grok-web-stream-error-"));
+
+process.env.DATA_DIR = path.join(testRoot, "data");
+process.env.OMNIROUTE_PLUGINS_DIR = path.join(testRoot, "plugins");
+fs.mkdirSync(process.env.DATA_DIR, { recursive: true });
+fs.mkdirSync(process.env.OMNIROUTE_PLUGINS_DIR, { recursive: true });
+globalThis.fetch = async () => {
+  throw new Error("Unexpected network request in Grok stream error boundary test");
+};
+
+const [
+  { GrokWebExecutor },
+  { __setTlsFetchOverrideForTesting },
+  dbCore,
+  settingsDb,
+  callLogs,
+  artifactWriter,
+  { handleChatCore },
+  usageHistory,
+  accountSemaphore,
+  requestDedup,
+  accountFallback,
+  loggerResource,
+] = await Promise.all([
+  import("../../open-sse/executors/grok-web.ts"),
+  import("../../open-sse/services/grokTlsClient.ts"),
+  import("../../src/lib/db/core.ts"),
+  import("../../src/lib/db/settings.ts"),
+  import("../../src/lib/usage/callLogs.ts"),
+  import("../../src/lib/usage/callLogArtifactWriter.ts"),
+  import("../../open-sse/handlers/chatCore.ts"),
+  import("../../src/lib/usage/usageHistory.ts"),
+  import("../../open-sse/services/accountSemaphore.ts"),
+  import("../../open-sse/services/requestDedup.ts"),
+  import("../../open-sse/services/accountFallback.ts"),
+  import("../../src/shared/utils/loggerResource.ts"),
+]);
+
+function grokEventStream(events: unknown[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(`${events.map((event) => JSON.stringify(event)).join("\n")}\n`)
+      );
+      controller.close();
+    },
+  });
+}
+
+function stalledGrokEventStream(
+  events: unknown[],
+  onCancel: () => void
+): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder();
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(`${events.map((event) => JSON.stringify(event)).join("\n")}\n`)
+      );
+    },
+    pull() {
+      return new Promise<void>(() => {});
+    },
+    cancel() {
+      onCancel();
+      return new Promise<void>(() => {});
+    },
+  });
+}
+
+type TestExecutorLog = {
+  debug?: (tag: string, message: string) => void;
+  info?: (tag: string, message: string) => void;
+  warn?: (tag: string, message: string) => void;
+  error?: (tag: string, message: string) => void;
+};
+
+async function executeStreamingBody(
+  upstreamBody: ReadableStream<Uint8Array>,
+  requestBody: Record<string, unknown> = {
+    messages: [{ role: "user", content: "hello" }],
+    stream: true,
+  },
+  options: { log?: TestExecutorLog | null; signal?: AbortSignal | null } = {}
+): Promise<Response> {
+  __setTlsFetchOverrideForTesting(async () => ({
+    status: 200,
+    headers: new Headers({ "Content-Type": "application/x-ndjson" }),
+    text: null,
+    body: upstreamBody,
+  }));
+
+  const result = await new GrokWebExecutor().execute({
+    model: "grok-4.1-fast",
+    body: requestBody,
+    stream: true,
+    credentials: { apiKey: "sso=test-only-cookie" },
+    signal: options.signal ?? AbortSignal.timeout(10_000),
+    log: options.log ?? null,
+  });
+  return result.response;
+}
+
+function executeStreaming(events: unknown[]): Promise<Response> {
+  return executeStreamingBody(grokEventStream(events));
+}
+
+function parseSseData(text: string): unknown[] {
+  return text
+    .split(/\r?\n/)
+    .filter((line) => line.startsWith("data: ") && line !== "data: [DONE]")
+    .map((line) => JSON.parse(line.slice("data: ".length)) as unknown);
+}
+
+async function readUntilFailure(response: Response): Promise<{ text: string; error: unknown }> {
+  assert.ok(response.body, "expected a streaming response body");
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let text = "";
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) return { text, error: null };
+      text += decoder.decode(value, { stream: true });
+    }
+  } catch (error) {
+    text += decoder.decode();
+    return { text, error };
+  }
+}
+
+async function waitFor<T>(read: () => Promise<T | null>, timeoutMs = 3_000): Promise<T | null> {
+  const startedAt = Date.now();
+  while (Date.now() - startedAt < timeoutMs) {
+    const value = await read();
+    if (value) return value;
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+  return null;
+}
+
+async function settlesWithin(promise: Promise<unknown>, timeoutMs = 500): Promise<boolean> {
+  let timeout: ReturnType<typeof setTimeout> | undefined;
+  const settled = await Promise.race([
+    promise.then(() => true),
+    new Promise<boolean>((resolve) => {
+      timeout = setTimeout(() => resolve(false), timeoutMs);
+    }),
+  ]);
+  if (timeout) clearTimeout(timeout);
+  return settled;
+}
+
+test.afterEach(() => {
+  __setTlsFetchOverrideForTesting(null);
+  usageHistory.clearPendingRequests();
+  accountSemaphore.resetAll();
+  requestDedup.clearInflight();
+  accountFallback.clearModelLock();
+});
+
+test.after(async () => {
+  __setTlsFetchOverrideForTesting(null);
+  assert.equal(await callLogs.waitForCallLogSaves(3_000), true);
+  await artifactWriter.closeCallLogArtifactWriter();
+  usageHistory.clearPendingRequests();
+  accountSemaphore.resetAll();
+  requestDedup.clearInflight();
+  accountFallback.clearModelLock();
+  dbCore.resetDbInstance();
+  await loggerResource.closeSharedLoggerResource();
+  globalThis.fetch = originalFetch;
+
+  if (originalDataDir === undefined) delete process.env.DATA_DIR;
+  else process.env.DATA_DIR = originalDataDir;
+  if (originalPluginsDir === undefined) delete process.env.OMNIROUTE_PLUGINS_DIR;
+  else process.env.OMNIROUTE_PLUGINS_DIR = originalPluginsDir;
+
+  fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+});
+
+test("Grok Web rejects an error-only upstream stream before advertising HTTP 200 success", async () => {
+  const response = await executeStreaming([
+    {
+      error: {
+        code: "UPSTREAM_PRIVATE_CODE",
+        message:
+          "UPSTREAM_PRIVATE_DETAIL Bearer top-secret-token /srv/grok/handler.ts:42\n" +
+          "    at internal (/srv/grok/handler.ts:42:7)",
+      },
+    },
+  ]);
+
+  assert.equal(response.status, 502);
+  assert.match(response.headers.get("Content-Type") ?? "", /application\/json/);
+
+  const body = (await response.json()) as {
+    error: { message: string; type?: string; code?: string };
+    upstream_details?: { error?: { message?: string } };
+  };
+  assert.equal(body.error.code, "STREAM_EARLY_EOF");
+  assert.equal(body.error.type, "stream_early_eof");
+  assert.equal(body.upstream_details?.error?.message, "Grok upstream stream failed");
+
+  const publicBody = JSON.stringify(body);
+  assert.doesNotMatch(publicBody, /UPSTREAM_PRIVATE/);
+  assert.doesNotMatch(publicBody, /top-secret-token/);
+  assert.doesNotMatch(publicBody, /\/srv\/grok/);
+  assert.doesNotMatch(publicBody, /\bat internal\b/);
+});
+
+test("Grok Web preserves partial content then rejects with a fixed public error", async () => {
+  let upstreamCancelCalls = 0;
+  const response = await executeStreamingBody(
+    stalledGrokEventStream(
+      [
+        { result: { response: { token: "partial answer" } } },
+        {
+          error: {
+            code: "UPSTREAM_PRIVATE_CODE",
+            message: "UPSTREAM_PRIVATE_DETAIL secret=never-public /srv/grok/stream.ts:99",
+          },
+        },
+      ],
+      () => {
+        upstreamCancelCalls += 1;
+      }
+    )
+  );
+
+  assert.equal(response.status, 200);
+  const { text, error } = await readUntilFailure(response);
+  assert.ok(error instanceof Error);
+  assert.equal(error.message, "Grok upstream stream failed");
+  const payloads = parseSseData(text) as Array<Record<string, unknown>>;
+  const content = payloads.find((payload) => {
+    const choices = payload.choices as Array<{ delta?: { content?: string } }> | undefined;
+    return choices?.[0]?.delta?.content === "partial answer";
+  });
+  assert.ok(content, "the valid content preceding the upstream failure must be retained");
+
+  assert.doesNotMatch(text, /UPSTREAM_PRIVATE/);
+  assert.doesNotMatch(text, /never-public/);
+  assert.doesNotMatch(text, /\/srv\/grok/);
+  assert.doesNotMatch(text, /\[Error:/);
+  assert.doesNotMatch(text, /"finish_reason":"stop"/);
+  assert.equal(upstreamCancelCalls, 1);
+});
+
+test("Grok Web converts a reader failure after content into the same safe terminal error", async () => {
+  const encoder = new TextEncoder();
+  const upstreamBody = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(`${JSON.stringify({ result: { response: { token: "kept" } } })}\n`)
+      );
+      setTimeout(() => {
+        controller.error(
+          new Error("READER_PRIVATE_DETAIL Bearer stream-token /srv/grok/reader.ts:12")
+        );
+      }, 0);
+    },
+  });
+
+  const response = await executeStreamingBody(upstreamBody);
+  assert.equal(response.status, 200);
+  const { text, error } = await readUntilFailure(response);
+  assert.ok(error instanceof Error);
+  assert.equal(error.message, "Grok upstream stream failed");
+  const payloads = parseSseData(text) as Array<Record<string, unknown>>;
+  assert.ok(
+    payloads.some((payload) => {
+      const choices = payload.choices as Array<{ delta?: { content?: string } }> | undefined;
+      return choices?.[0]?.delta?.content === "kept";
+    })
+  );
+  assert.doesNotMatch(text, /READER_PRIVATE/);
+  assert.doesNotMatch(text, /stream-token/);
+  assert.doesNotMatch(text, /\/srv\/grok/);
+  assert.doesNotMatch(text, /"finish_reason":"stop"/);
+});
+
+test("Grok Web propagates downstream cancellation once without awaiting a stuck upstream", async () => {
+  const encoder = new TextEncoder();
+  let upstreamCancelCalls = 0;
+  const upstreamBody = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(
+          `${JSON.stringify({ result: { response: { token: "cancel-safe partial" } } })}\n`
+        )
+      );
+    },
+    pull() {
+      return new Promise<void>(() => {});
+    },
+    cancel() {
+      upstreamCancelCalls += 1;
+      return new Promise<void>(() => {});
+    },
+  });
+  const logMessages: string[] = [];
+  const recordLog = (tag: string, message: string) => {
+    logMessages.push(`${tag}: ${message}`);
+  };
+
+  const response = await executeStreamingBody(upstreamBody, undefined, {
+    log: { debug: recordLog, info: recordLog, warn: recordLog, error: recordLog },
+  });
+  assert.equal(response.status, 200);
+  assert.ok(response.body);
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let text = "";
+  while (!text.includes("cancel-safe partial")) {
+    const { done, value } = await reader.read();
+    assert.equal(done, false);
+    if (value) text += decoder.decode(value, { stream: true });
+  }
+  const logCountBeforeCancel = logMessages.length;
+
+  assert.equal(await settlesWithin(reader.cancel("client stopped reading")), true);
+  assert.equal(await settlesWithin(reader.cancel("duplicate cancel")), true);
+  await new Promise<void>((resolve) => setImmediate(resolve));
+
+  assert.equal(upstreamCancelCalls, 1);
+  assert.doesNotMatch(text, /"finish_reason":"stop"|data: \[DONE\]/);
+  assert.equal(logMessages.length, logCountBeforeCancel);
+});
+
+test("chatCore returns a pre-content Grok failure to the outer fallback contract", async () => {
+  const streamFailures: Array<Record<string, unknown>> = [];
+  let requestSucceeded = false;
+  const requestBody = {
+    model: "grok-4.1-fast",
+    messages: [{ role: "user", content: "fallback proof" }],
+    stream: true,
+  };
+
+  __setTlsFetchOverrideForTesting(async () => ({
+    status: 200,
+    headers: new Headers({ "Content-Type": "application/x-ndjson" }),
+    text: null,
+    body: grokEventStream([
+      {
+        error: {
+          code: "FALLBACK_PRIVATE_CODE",
+          message: "FALLBACK_PRIVATE_DETAIL secret=never-public /srv/grok/fallback.ts:5",
+        },
+      },
+    ]),
+  }));
+
+  const result = await handleChatCore({
+    body: structuredClone(requestBody),
+    modelInfo: { provider: "grok-web", model: "grok-4.1-fast", extendedContext: false },
+    credentials: { apiKey: "sso=test-only-cookie", providerSpecificData: {} },
+    connectionId: "grok-stream-error-fallback",
+    log: { debug() {}, info() {}, warn() {}, error() {} },
+    clientRawRequest: {
+      endpoint: "/v1/chat/completions",
+      body: structuredClone(requestBody),
+      headers: new Headers({ accept: "text/event-stream" }),
+    },
+    userAgent: "grok-stream-error-boundary-test",
+    onRequestSuccess() {
+      requestSucceeded = true;
+    },
+    onStreamFailure(failure: Record<string, unknown>) {
+      streamFailures.push(failure);
+    },
+  } as never);
+
+  assert.equal(result.success, false);
+  assert.equal(result.status, 502);
+  assert.equal(requestSucceeded, false);
+  assert.deepEqual(streamFailures, []);
+
+  const publicBody = await result.response.text();
+  assert.match(publicBody, /Grok upstream stream failed/);
+  assert.doesNotMatch(publicBody, /FALLBACK_PRIVATE|never-public|\/srv\/grok/);
+  assert.doesNotMatch(publicBody, /"role":"assistant"|"finish_reason":"stop"/);
+});
+
+test("chatCore converts a Grok post-content failure into terminal wire error and failed persistence", async () => {
+  await settingsDb.updateSettings({ call_log_pipeline_enabled: true });
+  const streamFailures: Array<Record<string, unknown>> = [];
+  const requestBody = {
+    model: "grok-4.1-fast",
+    messages: [{ role: "user", content: "pipeline proof" }],
+    stream: true,
+  };
+
+  __setTlsFetchOverrideForTesting(async () => ({
+    status: 200,
+    headers: new Headers({ "Content-Type": "application/x-ndjson" }),
+    text: null,
+    body: grokEventStream([
+      { result: { response: { token: "pipeline partial" } } },
+      {
+        error: {
+          code: "PIPELINE_PRIVATE_CODE",
+          message: "PIPELINE_PRIVATE_DETAIL secret=never-public /srv/grok/pipeline.ts:7",
+        },
+      },
+    ]),
+  }));
+
+  const result = await handleChatCore({
+    body: structuredClone(requestBody),
+    modelInfo: { provider: "grok-web", model: "grok-4.1-fast", extendedContext: false },
+    credentials: { apiKey: "sso=test-only-cookie", providerSpecificData: {} },
+    connectionId: "grok-stream-error-boundary",
+    log: { debug() {}, info() {}, warn() {}, error() {} },
+    clientRawRequest: {
+      endpoint: "/v1/chat/completions",
+      body: structuredClone(requestBody),
+      headers: new Headers({ accept: "text/event-stream" }),
+    },
+    userAgent: "grok-stream-error-boundary-test",
+    onStreamFailure(failure: Record<string, unknown>) {
+      streamFailures.push(failure);
+    },
+  } as never);
+
+  assert.equal(result.success, true);
+  const wire = await result.response.text();
+  assert.match(wire, /"content":"pipeline partial"/);
+  assert.match(wire, /"finish_reason":"error"/);
+  assert.match(wire, /"message":"Grok upstream stream failed"/);
+  assert.match(wire, /"type":"server_error"/);
+  assert.match(wire, /"code":"server_error"/);
+  assert.match(wire, /data: \[DONE\]/);
+  assert.doesNotMatch(wire, /"finish_reason":"stop"/);
+  assert.doesNotMatch(wire, /PIPELINE_PRIVATE|never-public|\/srv\/grok/);
+
+  assert.equal(streamFailures.length, 1);
+  assert.deepEqual(streamFailures[0], {
+    status: 502,
+    message: "Grok upstream stream failed",
+    code: "stream_pipeline_error",
+    type: "stream_error",
+  });
+
+  assert.equal(await callLogs.waitForCallLogSaves(3_000), true);
+  const persisted = await waitFor(async () => {
+    const rows = await callLogs.getCallLogs({ provider: "grok-web", status: "error", limit: 5 });
+    return rows.find((row) => row.connectionId === "grok-stream-error-boundary") ?? null;
+  });
+  assert.ok(persisted, "expected the pipeline failure to be persisted");
+  assert.equal(persisted.status, 502);
+  assert.equal(persisted.error, "Grok upstream stream failed");
+
+  const detail = await callLogs.getCallLogById(persisted.id);
+  assert.ok(detail?.pipelinePayloads, "expected failed pipeline payloads in the call log");
+  const persistedPayload = JSON.stringify(detail.pipelinePayloads);
+  assert.match(persistedPayload, /Grok upstream stream failed/);
+  assert.doesNotMatch(persistedPayload, /PIPELINE_PRIVATE|never-public|\/srv\/grok/);
+});
+
+test("Grok Web still emits streaming tool calls after delaying the assistant role", async () => {
+  let upstreamCancelCalls = 0;
+  const response = await executeStreamingBody(
+    stalledGrokEventStream(
+      [
+        {
+          result: {
+            response: {
+              modelResponse: {
+                message:
+                  '<tool_call>{"name":"memory_context_tool","arguments":{"query":"grok"}}</tool_call>',
+              },
+            },
+          },
+        },
+      ],
+      () => {
+        upstreamCancelCalls += 1;
+      }
+    ),
+    {
+      messages: [{ role: "user", content: "search memory" }],
+      stream: true,
+      tools: [
+        {
+          type: "function",
+          function: {
+            name: "memory_context_tool",
+            parameters: { type: "object", properties: { query: { type: "string" } } },
+          },
+        },
+      ],
+    }
+  );
+
+  assert.equal(response.status, 200);
+  const text = await response.text();
+  assert.match(text, /"role":"assistant"/);
+  assert.match(text, /"tool_calls"/);
+  assert.match(text, /"name":"memory_context_tool"/);
+  assert.match(text, /"finish_reason":"tool_calls"/);
+  assert.doesNotMatch(text, /"error"/);
+  assert.equal(upstreamCancelCalls, 1);
+});

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

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