From c14100de12b9b32d3082e60485f6bfa708d574ff Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 2 Sep 2026 06:55:02 -0300 Subject: [PATCH] fix(grok-web): fail streaming errors safely --- .../pending-grok-web-stream-error-boundary.md | 4 + open-sse/executors/grok-web.ts | 216 +++++--- .../grok-web-stream-error-boundary.test.ts | 509 ++++++++++++++++++ 3 files changed, 663 insertions(+), 66 deletions(-) create mode 100644 changelog.d/fixes/pending-grok-web-stream-error-boundary.md create mode 100644 tests/unit/grok-web-stream-error-boundary.test.ts diff --git a/changelog.d/fixes/pending-grok-web-stream-error-boundary.md b/changelog.d/fixes/pending-grok-web-stream-error-boundary.md new file mode 100644 index 0000000000..28e4292504 --- /dev/null +++ b/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. diff --git a/open-sse/executors/grok-web.ts b/open-sse/executors/grok-web.ts index a99bc2590a..33d364d1a3 100644 --- a/open-sse/executors/grok-web.ts +++ b/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, encoder: TextEncoder, @@ -349,63 +394,77 @@ function buildStreamingResponse( signal?: AbortSignal | null ): ReadableStream { 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((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, diff --git a/tests/unit/grok-web-stream-error-boundary.test.ts b/tests/unit/grok-web-stream-error-boundary.test.ts new file mode 100644 index 0000000000..6ef06abe68 --- /dev/null +++ b/tests/unit/grok-web-stream-error-boundary.test.ts @@ -0,0 +1,509 @@ +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 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, + { handleChatCore }, + usageHistory, + accountSemaphore, + requestDedup, + accountFallback, +] = 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("../../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"), +]); + +function grokEventStream(events: unknown[]): ReadableStream { + 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 { + 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(() => {}); + }, + cancel() { + onCancel(); + return new Promise(() => {}); + }, + }); +} + +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, + requestBody: Record = { + messages: [{ role: "user", content: "hello" }], + stream: true, + }, + options: { log?: TestExecutorLog | null; signal?: AbortSignal | null } = {} +): Promise { + __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 { + 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(read: () => Promise, timeoutMs = 3_000): Promise { + 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, timeoutMs = 500): Promise { + let timeout: ReturnType | undefined; + const settled = await Promise.race([ + promise.then(() => true), + new Promise((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); + globalThis.fetch = originalFetch; + await callLogs.waitForCallLogSaves(3_000); + usageHistory.clearPendingRequests(); + accountSemaphore.resetAll(); + requestDedup.clearInflight(); + accountFallback.clearModelLock(); + dbCore.resetDbInstance(); + + 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 }); +}); + +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>; + 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({ + 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>; + 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({ + start(controller) { + controller.enqueue( + encoder.encode( + `${JSON.stringify({ result: { response: { token: "cancel-safe partial" } } })}\n` + ) + ); + }, + pull() { + return new Promise(() => {}); + }, + cancel() { + upstreamCancelCalls += 1; + return new Promise(() => {}); + }, + }); + 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((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> = []; + 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) { + 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> = []; + 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) { + 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: + '{"name":"memory_context_tool","arguments":{"query":"grok"}}', + }, + }, + }, + }, + ], + () => { + 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); +});