From 350ac8c12ddccbd7bd2c28df234f98d4de43e694 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Thu, 3 Sep 2026 21:06:28 -0300 Subject: [PATCH] fix(sse): preserve 1min.ai partial output before stream errors (#12466) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Validado após reconciliar com o #12465, que entrou primeiro e criou o mesmo arquivo novo `open-sse/utils/streamReadiness.ts` com desenho divergente de cancelamento. Mantive a versão desta branch, que defere o release do lock para quando a leitura em voo termina e faz `reader.cancel()` fire-and-forget — assim uma promise de provider que nunca resolve não torna o cancelamento ilimitado. A escolha não foi por preferência: rodei as suítes dos **dois** PRs contra ela, 21/21 no readiness compartilhado e **22/22** incluindo o boundary do Perplexity do próprio #12465. typecheck:core limpo. --- .../pending-onemin-stream-error-boundary.md | 1 + open-sse/executors/oneminai.ts | 341 +++++++--- open-sse/utils/streamReadiness.ts | 50 +- .../oneminai-stream-error-boundary.fixture.ts | 620 ++++++++++++++++++ .../oneminai-stream-error-boundary.test.ts | 91 +++ tests/unit/stream-readiness.test.ts | 120 +++- 6 files changed, 1083 insertions(+), 140 deletions(-) create mode 100644 changelog.d/fixes/pending-onemin-stream-error-boundary.md create mode 100644 tests/fixtures/oneminai-stream-error-boundary.fixture.ts create mode 100644 tests/unit/oneminai-stream-error-boundary.test.ts diff --git a/changelog.d/fixes/pending-onemin-stream-error-boundary.md b/changelog.d/fixes/pending-onemin-stream-error-boundary.md new file mode 100644 index 0000000000..818b124a2f --- /dev/null +++ b/changelog.d/fixes/pending-onemin-stream-error-boundary.md @@ -0,0 +1 @@ +- **fix(providers):** keep 1min.ai HTTP 200 stream errors out of assistant content, preserve partial output, and expose sanitized terminal errors so pre-content failures can fall back. diff --git a/open-sse/executors/oneminai.ts b/open-sse/executors/oneminai.ts index 023f6ce425..5b3e2f4dda 100644 --- a/open-sse/executors/oneminai.ts +++ b/open-sse/executors/oneminai.ts @@ -16,6 +16,8 @@ type OpenAIMessage = { }; const CHAT_URL = "https://api.1min.ai/api/chat-with-ai"; +const MAX_STREAM_ERROR_DATA_CHARS = 64 * 1024; +const STREAM_ERROR_FALLBACK = "1min.ai upstream stream failed"; const ROLE_LABELS: Record = { system: "System", developer: "System", @@ -69,7 +71,35 @@ function buildSseChunk(data: unknown): string { return `data: ${JSON.stringify(data)}\n\n`; } -function buildOpenAiJsonCompletion(content: string, model: string, id: string, created: number): Response { +function parseStreamErrorMessage(data: string): string { + if (!data || data.length > MAX_STREAM_ERROR_DATA_CHARS) return STREAM_ERROR_FALLBACK; + + try { + const parsed = asRecord(JSON.parse(data)); + const directMessage = typeof parsed.message === "string" ? parsed.message.trim() : ""; + if (directMessage) return directMessage; + + if (typeof parsed.error === "string") { + const errorMessage = parsed.error.trim(); + if (errorMessage) return errorMessage; + } + + const nestedError = asRecord(parsed.error); + const nestedMessage = typeof nestedError.message === "string" ? nestedError.message.trim() : ""; + if (nestedMessage) return nestedMessage; + } catch { + // Malformed and over-complex payloads use the fixed public fallback below. + } + + return STREAM_ERROR_FALLBACK; +} + +function buildOpenAiJsonCompletion( + content: string, + model: string, + id: string, + created: number +): Response { return new Response( JSON.stringify({ id, @@ -84,7 +114,11 @@ function buildOpenAiJsonCompletion(content: string, model: string, id: string, c ); } -function toOpenAiErrorResponse(status: number, message: string, upstreamDetails?: unknown): Response { +function toOpenAiErrorResponse( + status: number, + message: string, + upstreamDetails?: unknown +): Response { return new Response(JSON.stringify(buildErrorBody(status, message, upstreamDetails)), { status, headers: { "Content-Type": "application/json" }, @@ -96,109 +130,214 @@ function toOpenAiErrorResponse(status: number, message: string, upstreamDetails? * data: {...}) from the upstream Response body and re-emit them as standard * OpenAI chat.completion.chunk SSE. */ -function translateSseStream(upstreamBody: ReadableStream, model: string, id: string, created: number): ReadableStream { +function translateSseStream( + upstreamBody: ReadableStream, + model: string, + id: string, + created: number +): ReadableStream { const decoder = new TextDecoder(); const encoder = new TextEncoder(); + const reader = upstreamBody.getReader(); + const pendingChunks: Uint8Array[] = []; + let buffer = ""; + let finished = false; + let roleEmitted = false; + let terminalError: Error | null = null; + let upstreamCancelRequested = false; + let downstreamCancelled = false; + let readInFlight = false; + let readerReleased = false; + + const releaseReader = () => { + if (readerReleased) return; + readerReleased = true; + reader.releaseLock(); + }; + + const cancelUpstream = (reason: unknown) => { + if (upstreamCancelRequested) return; + upstreamCancelRequested = true; + try { + // Upstream cleanup is provider-controlled and may never settle. The + // translated stream owns the reader lock and releases it independently. + void reader.cancel(reason).catch(() => {}); + } catch { + // Cancellation is cleanup-only; the terminal state is already fixed. + } + }; + + const queueChunk = (text: string) => { + pendingChunks.push(encoder.encode(text)); + }; + + const emitRole = () => { + if (roleEmitted) return; + roleEmitted = true; + queueChunk( + buildSseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], + }) + ); + }; + + const finish = () => { + if (finished) return; + finished = true; + queueChunk( + buildSseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + }) + ); + queueChunk("data: [DONE]\n\n"); + }; + + const emitContent = (text: string) => { + if (!text) return; + emitRole(); + queueChunk( + buildSseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { content: text }, finish_reason: null }], + }) + ); + }; + + const emitError = (data: string) => { + if (finished) return; + finished = true; + cancelUpstream("1min.ai upstream stream error"); + + if (!roleEmitted) { + const message = parseStreamErrorMessage(data); + queueChunk(buildSseChunk(buildErrorBody(502, message))); + queueChunk("data: [DONE]\n\n"); + return; + } + + // A bare `{ error }` frame is dropped by the OpenAI passthrough sanitizer. + // Preserve every content delta already queued, then error the source with + // a fixed public message. pipeWithDisconnect() converts it into a native + // terminal error frame and drives usage, call-log, and fallback finalizers. + terminalError = Object.assign(new Error(STREAM_ERROR_FALLBACK), { + statusCode: 502, + }); + }; + + // SSE event framing: "event:"/"data:" lines, blank-line separated records. + const processEvent = (eventText: string) => { + let eventType = "message"; + const dataLines: string[] = []; + for (const rawLine of eventText.split("\n")) { + if (rawLine.startsWith("event:")) { + eventType = rawLine.slice(6).trim(); + } else if (rawLine.startsWith("data:")) { + dataLines.push(rawLine.slice(5).trim()); + } + } + const data = dataLines.join("\n"); + if (eventType === "content") { + try { + const parsed = asRecord(JSON.parse(data)); + if (typeof parsed.content === "string") emitContent(parsed.content); + } catch { + // Ignore malformed content events rather than surfacing partial JSON. + } + } else if (eventType === "error") { + emitError(data); + } else if (eventType === "done") { + finish(); + } + // "result" carries the final full aiRecord, redundant with the content + // events already streamed — intentionally ignored. + }; + + const processBufferedEvents = () => { + let separatorIndex = buffer.indexOf("\n\n"); + while (separatorIndex !== -1 && !finished) { + processEvent(buffer.slice(0, separatorIndex)); + buffer = buffer.slice(separatorIndex + 2); + separatorIndex = buffer.indexOf("\n\n"); + } + }; return new ReadableStream({ - async start(controller) { - controller.enqueue( - encoder.encode( - buildSseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], - }) - ) - ); + async pull(controller) { + if (downstreamCancelled) return; - const reader = upstreamBody.getReader(); - let buffer = ""; - let finished = false; - - const finish = () => { - if (finished) return; - finished = true; - controller.enqueue( - encoder.encode( - buildSseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: {}, finish_reason: "stop" }], - }) - ) - ); - controller.enqueue(encoder.encode("data: [DONE]\n\n")); - controller.close(); - }; - - const emitContent = (text: string) => { - if (!text) return; - controller.enqueue( - encoder.encode( - buildSseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { content: text }, finish_reason: null }], - }) - ) - ); - }; - - // SSE event framing: "event:"/"data:" lines, blank-line separated records. - const processEvent = (eventText: string) => { - let eventType = "message"; - const dataLines: string[] = []; - for (const rawLine of eventText.split("\n")) { - if (rawLine.startsWith("event:")) { - eventType = rawLine.slice(6).trim(); - } else if (rawLine.startsWith("data:")) { - dataLines.push(rawLine.slice(5).trim()); - } - } - const data = dataLines.join("\n"); - if (eventType === "content") { - try { - const parsed = asRecord(JSON.parse(data)); - if (typeof parsed.content === "string") emitContent(parsed.content); - } catch { - // Ignore malformed content events rather than surfacing partial JSON. - } - } else if (eventType === "error") { - emitContent(`\n[1min.ai error: ${data}]`); - finish(); - } else if (eventType === "done") { - finish(); - } - // "result" carries the final full aiRecord, redundant with the content - // events already streamed — intentionally ignored. - }; - - try { - while (!finished) { - const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - let separatorIndex = buffer.indexOf("\n\n"); - while (separatorIndex !== -1) { - processEvent(buffer.slice(0, separatorIndex)); - buffer = buffer.slice(separatorIndex + 2); - separatorIndex = buffer.indexOf("\n\n"); - } - } - if (!finished && buffer.trim()) processEvent(buffer); - finish(); - } catch (error) { - controller.error(error); - } finally { - reader.releaseLock(); + if (pendingChunks.length > 0) { + controller.enqueue(pendingChunks.shift()!); + return; } + + if (terminalError) { + releaseReader(); + controller.error(terminalError); + return; + } + + if (finished) { + releaseReader(); + controller.close(); + return; + } + + readInFlight = true; + try { + while (pendingChunks.length === 0 && !finished && !downstreamCancelled) { + const { done, value } = await reader.read(); + if (downstreamCancelled) return; + if (done) { + buffer += decoder.decode(); + if (buffer.trim()) processEvent(buffer); + finish(); + break; + } + + buffer += decoder.decode(value, { stream: true }); + // Process the complete upstream chunk, even after it queues output. + // One network read may contain multiple content events followed by + // an error; the internal queue preserves all of them in order. + processBufferedEvents(); + } + + if (downstreamCancelled) return; + if (pendingChunks.length > 0) { + controller.enqueue(pendingChunks.shift()!); + } else if (terminalError) { + releaseReader(); + controller.error(terminalError); + } else if (finished) { + releaseReader(); + controller.close(); + } + } catch (error) { + releaseReader(); + if (!downstreamCancelled) controller.error(error); + } finally { + readInFlight = false; + if (downstreamCancelled) releaseReader(); + } + }, + cancel(reason) { + downstreamCancelled = true; + pendingChunks.length = 0; + // A client disconnect must release the upstream reader even when its + // next pull never settles. Do not await provider cleanup here: the + // downstream cancellation contract must remain bounded. + cancelUpstream(reason ?? "1min.ai downstream cancelled"); + if (!readInFlight) releaseReader(); }, }); } @@ -290,7 +429,9 @@ export class OneMinAiExecutor extends BaseExecutor { const aiRecord = asRecord(json.aiRecord); const detail = asRecord(aiRecord.aiRecordDetail); const resultObject = Array.isArray(detail.resultObject) ? detail.resultObject : []; - const content = resultObject.filter((part): part is string => typeof part === "string").join(""); + const content = resultObject + .filter((part): part is string => typeof part === "string") + .join(""); return { response: buildOpenAiJsonCompletion(content, model, id, created), diff --git a/open-sse/utils/streamReadiness.ts b/open-sse/utils/streamReadiness.ts index 3774cc0260..908c18725c 100644 --- a/open-sse/utils/streamReadiness.ts +++ b/open-sse/utils/streamReadiness.ts @@ -422,56 +422,66 @@ function prependBufferedChunks( reader: ReadableStreamDefaultReader ): ReadableStream { let bufferedIndex = 0; - let cancelled = false; + let readInFlight = false; + let cancelRequested = false; let readerReleased = false; const releaseReader = () => { if (readerReleased) return; readerReleased = true; + reader.releaseLock(); + }; + + const cancelReader = (reason: unknown) => { + if (cancelRequested) return; + cancelRequested = true; + try { - reader.releaseLock(); + // The provider controls this promise and may never settle. Cancellation + // of the replay stream must remain bounded, so cleanup is deliberately + // fire-and-forget while the in-flight read releases the lock in `pull`. + void reader.cancel(reason).catch(() => {}); } catch { - // A hostile source can keep a read/cancel pending forever. The public stream must - // remain cancellable even when its abandoned source cannot release immediately. + // A synchronous cancellation failure is cleanup-only; the downstream + // stream has already been cancelled by its consumer. } + + if (!readInFlight) releaseReader(); }; return new ReadableStream({ async pull(controller) { - if (cancelled) return; + if (cancelRequested) return; - // Replay exactly one readiness chunk per pull. Keeping the first buffered chunk at - // the stream's default high-water mark prevents an eager read of a later upstream - // failure from discarding that legitimate prefix before the caller attaches. + // Replay exactly one readiness chunk per demand. Reading the source + // eagerly here would let a subsequent source error clear this queue + // before the consumer has observed the buffered prefix. if (bufferedIndex < chunks.length) { controller.enqueue(chunks[bufferedIndex]); bufferedIndex += 1; return; } + readInFlight = true; try { const { done, value } = await reader.read(); - if (cancelled) return; + if (cancelRequested) return; if (done) { releaseReader(); controller.close(); - return; + } else if (value) { + controller.enqueue(value); } - if (value) controller.enqueue(value); } catch (error) { releaseReader(); - if (!cancelled) controller.error(error); + if (!cancelRequested) controller.error(error); + } finally { + readInFlight = false; + if (cancelRequested) releaseReader(); } }, cancel(reason) { - if (cancelled) return; - cancelled = true; - // Do not await a provider's cancel hook: a hostile or stalled source must not make - // downstream cancellation hang. Release the lock once cancellation actually settles. - void reader - .cancel(reason) - .catch(() => {}) - .finally(releaseReader); + cancelReader(reason); }, }); } diff --git a/tests/fixtures/oneminai-stream-error-boundary.fixture.ts b/tests/fixtures/oneminai-stream-error-boundary.fixture.ts new file mode 100644 index 0000000000..5af1b6be9c --- /dev/null +++ b/tests/fixtures/oneminai-stream-error-boundary.fixture.ts @@ -0,0 +1,620 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +assert.ok(process.env.DATA_DIR, "the parent harness must provide an isolated DATA_DIR"); +assert.ok( + process.env.OMNIROUTE_PLUGINS_DIR, + "the parent harness must provide an isolated OMNIROUTE_PLUGINS_DIR" +); + +const [ + { OneMinAiExecutor }, + { ensureStreamReadiness }, + dbCore, + settingsDb, + callLogs, + usageHistory, + accountSemaphore, + readCache, + { handleChatCore }, +] = await Promise.all([ + import("../../open-sse/executors/oneminai.ts"), + import("../../open-sse/utils/streamReadiness.ts"), + import("../../src/lib/db/core.ts"), + import("../../src/lib/db/settings.ts"), + import("../../src/lib/usage/callLogs.ts"), + import("../../src/lib/usage/usageHistory.ts"), + import("../../open-sse/services/accountSemaphore.ts"), + import("../../src/lib/db/readCache.ts"), + import("../../open-sse/handlers/chatCore.ts"), +]); + +const originalFetch = globalThis.fetch; +const encoder = new TextEncoder(); +const STREAM_URL = "https://api.1min.ai/api/chat-with-ai?isStreaming=true"; + +type PersistenceIdentity = { + model: string; + connectionId: string; +}; + +const PRE_CONTENT_IDENTITY: PersistenceIdentity = { + model: "gpt-4o-mini-onemin-pre-content-boundary", + connectionId: "onemin-stream-pre-content-boundary", +}; +const BATCHED_IDENTITY: PersistenceIdentity = { + model: "gpt-4o-mini-onemin-batched-boundary", + connectionId: "onemin-stream-batched-boundary", +}; +const PARTIAL_IDENTITY: PersistenceIdentity = { + model: "gpt-4o-mini-onemin-partial-boundary", + connectionId: "onemin-stream-partial-boundary", +}; + +function installFetchFactory(responseFactory: () => Response): () => number { + let calls = 0; + globalThis.fetch = async (input, init = {}) => { + calls += 1; + assert.equal(String(input), STREAM_URL, "the test must never permit another network target"); + assert.equal(init.method, "POST"); + assert.equal((init.headers as Record)["API-KEY"], "unit-test-key"); + + return responseFactory(); + }; + return () => calls; +} + +function createStreamingResponse(events: string[]): Response { + return new Response( + new ReadableStream({ + start(controller) { + for (const event of events) controller.enqueue(encoder.encode(event)); + controller.close(); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); +} + +function installStreamingFetch(events: string[]): () => number { + return installFetchFactory(() => createStreamingResponse(events)); +} + +async function executeStreaming(events: string[]): Promise { + const getCalls = installStreamingFetch(events); + const result = await new OneMinAiExecutor().execute({ + model: "gpt-4o-mini", + body: { messages: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { apiKey: "unit-test-key" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + assert.equal(getCalls(), 1); + return result.response; +} + +function noopLog() { + return { debug() {}, info() {}, warn() {}, error() {} }; +} + +async function invokeStreamingChatCore( + identity: PersistenceIdentity, + onStreamFailure?: (failure: { + status: number; + message: string; + code?: string; + type?: string; + }) => void, + onRequestSuccess?: () => Promise | void +) { + await settingsDb.updateSettings({ call_log_pipeline_enabled: true }); + readCache.invalidateDbCache("settings"); + const body = { + model: identity.model, + stream: true, + messages: [{ role: "user", content: "hello" }], + }; + + return handleChatCore({ + body: structuredClone(body), + modelInfo: { provider: "oneminai", model: identity.model, extendedContext: false }, + credentials: { + apiKey: "unit-test-key", + connectionId: identity.connectionId, + providerSpecificData: {}, + }, + connectionId: identity.connectionId, + log: noopLog(), + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: structuredClone(body), + headers: new Headers({ + accept: "text/event-stream", + "x-omniroute-session-id": identity.connectionId, + }), + }, + userAgent: identity.connectionId, + onRequestSuccess, + onStreamFailure, + } as never); +} + +async function waitFor(read: () => Promise, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const value = await read(); + if (value) return value; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return null; +} + +async function getOneMinCallLog(identity: PersistenceIdentity) { + assert.equal( + await callLogs.waitForCallLogSaves(5_000), + true, + "call-log persistence must drain before inspection" + ); + const rows = await callLogs.getCallLogs({ + provider: "oneminai", + model: identity.model, + limit: 20, + }); + const row = Array.isArray(rows) + ? rows.find( + (candidate) => + candidate.connectionId === identity.connectionId && + (candidate.model === identity.model || candidate.requestedModel === identity.model) + ) + : null; + return row ? callLogs.getCallLogById(row.id) : null; +} + +async function getOneMinUsage(identity: PersistenceIdentity) { + const rows = await usageHistory.getUsageHistory({ + provider: "oneminai", + model: identity.model, + }); + return rows.find((row) => row.connectionId === identity.connectionId) ?? null; +} + +async function assertUnusedPersistenceIdentity(identity: PersistenceIdentity) { + assert.equal( + await getOneMinCallLog(identity), + null, + `call-log identity must be unused before scenario: ${identity.connectionId}` + ); + assert.equal( + await getOneMinUsage(identity), + null, + `usage identity must be unused before scenario: ${identity.connectionId}` + ); +} + +async function readUntil( + reader: ReadableStreamDefaultReader, + marker: string +): Promise { + const decoder = new TextDecoder(); + let text = ""; + while (!text.includes(marker)) { + const { done, value } = await reader.read(); + assert.equal(done, false, `stream ended before ${marker}`); + if (value) text += decoder.decode(value, { stream: true }); + } + return text; +} + +async function readRemaining(reader: ReadableStreamDefaultReader): Promise { + const decoder = new TextDecoder(); + let text = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) return text + decoder.decode(); + if (value) text += decoder.decode(value, { stream: true }); + } +} + +test.afterEach(async () => { + const drained = await callLogs.waitForCallLogSaves(5_000); + globalThis.fetch = originalFetch; + usageHistory.clearPendingRequests(); + accountSemaphore.resetAll(); + assert.equal(drained, true, "all call-log saves must drain before the next test"); +}); + +test.after(async () => { + const drained = await callLogs.waitForCallLogSaves(5_000); + try { + await callLogs.closeCallLogSaves(5_000); + } finally { + globalThis.fetch = originalFetch; + usageHistory.clearPendingRequests(); + accountSemaphore.resetAll(); + dbCore.resetDbInstance(); + } + assert.equal(drained, true, "all call-log saves must drain before teardown"); +}); + +test("1min.ai pre-content stream errors stay errors and permit readiness fallback", async () => { + const rawMessage = + "quota lookup failed at /srv/omniroute/open-sse/executors/oneminai.ts:170\n" + + " at translateSseStream (/srv/omniroute/open-sse/executors/oneminai.ts:99:5)"; + const response = await executeStreaming([ + `event: error\ndata: ${JSON.stringify({ error: { message: rawMessage } })}\n\n`, + ]); + const clientCopy = response.clone(); + + const readiness = await ensureStreamReadiness(response, { + timeoutMs: 2_000, + provider: "oneminai", + model: "gpt-4o-mini", + }); + assert.equal(readiness.ok, false); + if (readiness.ok) assert.fail("an error-only stream must not become ready"); + assert.equal(readiness.response.status, 502); + const fallbackBody = await readiness.response.text(); + assert.match(fallbackBody, /STREAM_EARLY_EOF/); + assert.doesNotMatch(fallbackBody, /\/srv\/omniroute/); + assert.doesNotMatch(fallbackBody, /translateSseStream/); + + const clientText = await clientCopy.text(); + assert.match(clientText, /^data: \{"error":/); + assert.match(clientText, /quota lookup failed at /); + assert.match(clientText, /data: \[DONE\]/); + assert.doesNotMatch(clientText, /"role":"assistant"/); + assert.doesNotMatch(clientText, /"finish_reason":"stop"/); + assert.doesNotMatch(clientText, /\/srv\/omniroute/); + assert.doesNotMatch(clientText, /translateSseStream/); +}); + +test("chatCore turns a pre-content 1min.ai stream error into persisted HTTP 502", async () => { + await assertUnusedPersistenceIdentity(PRE_CONTENT_IDENTITY); + installStreamingFetch([ + `event: error\ndata: ${JSON.stringify({ + error: { + message: + "quota lookup failed at /srv/omniroute/open-sse/executors/oneminai.ts:230 api_key=pre-content-secret\nstack tail", + }, + })}\n\n`, + ]); + + const result = await invokeStreamingChatCore(PRE_CONTENT_IDENTITY); + assert.equal(result.success, false); + if (result.success) assert.fail("a pre-content error must not commit HTTP 200"); + assert.equal(result.status, 502); + assert.equal(result.response.status, 502); + const clientBody = await result.response.text(); + assert.match(clientBody, /STREAM_EARLY_EOF/); + assert.doesNotMatch(clientBody, /pre-content-secret/); + assert.doesNotMatch(clientBody, /\/srv\/omniroute/); + assert.doesNotMatch(clientBody, /stack tail/); + + const detail = await waitFor(() => getOneMinCallLog(PRE_CONTENT_IDENTITY)); + assert.ok(detail, "the failed pre-content attempt must be persisted"); + assert.equal(detail.status, 502); + const persisted = JSON.stringify(detail); + assert.doesNotMatch(persisted, /pre-content-secret/); + assert.doesNotMatch(persisted, /\/srv\/omniroute/); + assert.doesNotMatch(persisted, /stack tail/); + + const usage = await waitFor(() => getOneMinUsage(PRE_CONTENT_IDENTITY)); + assert.ok(usage, "the failed pre-content usage record must be persisted"); + assert.equal(usage.success, false); + assert.equal(usage.status, "502"); + assert.equal(usage.errorCode, "STREAM_EARLY_EOF"); +}); + +test("chatCore preserves batched 1min.ai content before its terminal stream error", async () => { + await assertUnusedPersistenceIdentity(BATCHED_IDENTITY); + installStreamingFetch([ + 'event: content\ndata: {"content":"batched partial one"}\n\n' + + 'event: content\ndata: {"content":"batched partial two"}\n\n' + + `event: error\ndata: ${JSON.stringify({ + message: + "provider failed at /srv/omniroute/open-sse/executors/oneminai.ts:230 api_key=batched-secret", + })}\n\n`, + ]); + const failures: Array<{ + status: number; + message: string; + code?: string; + type?: string; + }> = []; + const requestSuccessPhases: string[] = []; + + const result = await invokeStreamingChatCore( + BATCHED_IDENTITY, + (failure) => failures.push(failure), + async () => { + requestSuccessPhases.push("started"); + await new Promise((resolve) => setTimeout(resolve, 30)); + requestSuccessPhases.push("finished"); + } + ); + assert.equal(result.success, true, "batched real content must cross the readiness boundary"); + assert.deepEqual(requestSuccessPhases, ["started", "finished"]); + assert.ok(result.response.body); + const clientText = await result.response.text(); + const firstContentIndex = clientText.indexOf("batched partial one"); + const secondContentIndex = clientText.indexOf("batched partial two"); + const errorIndex = clientText.indexOf('"error":'); + const doneIndex = clientText.indexOf("data: [DONE]"); + + assert.ok(firstContentIndex >= 0, "the first queued content delta must not be discarded"); + assert.ok(secondContentIndex >= 0, "the second queued content delta must not be discarded"); + assert.ok(firstContentIndex < secondContentIndex, "batched content must retain upstream order"); + assert.ok(secondContentIndex < errorIndex, "all batched content must precede its terminal error"); + assert.ok( + errorIndex < doneIndex, + `the terminal error must precede [DONE]: ${JSON.stringify(clientText)}` + ); + assert.match(clientText, /"finish_reason":"error"/); + assert.doesNotMatch(clientText, /"finish_reason":"stop"/); + assert.doesNotMatch(clientText, /response\.failed/); + assert.doesNotMatch(clientText, /batched-secret/); + assert.doesNotMatch(clientText, /\/srv\/omniroute/); + assert.deepEqual(failures, [ + { + status: 502, + message: "1min.ai upstream stream failed", + code: "stream_pipeline_error", + type: "stream_error", + }, + ]); + + const pending = usageHistory.getPendingRequests(); + assert.deepEqual(Object.keys(pending.byModel), []); + assert.deepEqual(Object.keys(pending.byAccount), []); + + const completed = [...usageHistory.getCompletedDetails().values()]; + assert.equal(completed.length, 1); + assert.equal(completed[0].status, 502); + assert.equal(completed[0].error, "1min.ai upstream stream failed"); + assert.equal(completed[0].errorCode, "stream_pipeline_error"); + + const detail = await waitFor(() => getOneMinCallLog(BATCHED_IDENTITY)); + assert.ok(detail, "the batched terminal stream failure must be persisted"); + assert.equal(detail.status, 502); + assert.equal(detail.error, "1min.ai upstream stream failed"); + const persisted = JSON.stringify(detail); + assert.doesNotMatch(persisted, /batched-secret/); + assert.doesNotMatch(persisted, /\/srv\/omniroute/); + + const usage = await waitFor(() => getOneMinUsage(BATCHED_IDENTITY)); + assert.ok(usage, "the batched terminal failure usage record must be persisted"); + assert.equal(usage.success, false); + assert.equal(usage.status, "502"); + assert.equal(usage.errorCode, "stream_pipeline_error"); +}); + +test("chatCore preserves partial 1min.ai content then finalizes and persists a stream failure", async () => { + await assertUnusedPersistenceIdentity(PARTIAL_IDENTITY); + let upstreamController: ReadableStreamDefaultController | null = null; + let cancelCalls = 0; + const getCalls = installFetchFactory( + () => + new Response( + new ReadableStream({ + start(controller) { + upstreamController = controller; + controller.enqueue( + encoder.encode('event: content\ndata: {"content":"partial answer"}\n\n') + ); + }, + cancel() { + cancelCalls += 1; + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ) + ); + const failures: Array<{ + status: number; + message: string; + code?: string; + type?: string; + }> = []; + + const result = await invokeStreamingChatCore(PARTIAL_IDENTITY, (failure) => + failures.push(failure) + ); + assert.equal(getCalls(), 1); + assert.equal(result.success, true, "real content must cross the readiness boundary"); + assert.ok(result.response.body); + const reader = result.response.body.getReader(); + let clientText = await readUntil(reader, "partial answer"); + + assert.ok(upstreamController); + upstreamController.enqueue( + encoder.encode( + `event: error\ndata: ${JSON.stringify({ + message: + "provider failed at /srv/omniroute/open-sse/executors/oneminai.ts:230 api_key=post-content-secret\nstack tail", + })}\n\n` + ) + ); + clientText += await readRemaining(reader); + + const roleIndex = clientText.indexOf('"role":"assistant"'); + const contentIndex = clientText.indexOf("partial answer"); + const errorIndex = clientText.indexOf('"error":'); + const doneIndex = clientText.indexOf("data: [DONE]"); + + assert.ok(roleIndex >= 0 && roleIndex < contentIndex, "the role must precede real content"); + assert.ok(contentIndex < errorIndex, "partial content must remain before the terminal error"); + assert.ok(errorIndex < doneIndex, "the pipeline error must precede [DONE]"); + assert.equal(clientText.match(/"role":"assistant"/g)?.length, 1); + assert.match(clientText, /"finish_reason":"error"/); + assert.match(clientText, /1min\.ai upstream stream failed/); + assert.doesNotMatch(clientText, /"finish_reason":"stop"/); + assert.doesNotMatch(clientText, /response\.failed/); + assert.doesNotMatch(clientText, /post-content-secret/); + assert.doesNotMatch(clientText, /\/srv\/omniroute/); + assert.doesNotMatch(clientText, /stack tail/); + + assert.equal(cancelCalls, 1, "the upstream source must be cancelled after its terminal error"); + assert.equal(failures.length, 1); + assert.deepEqual(failures[0], { + status: 502, + message: "1min.ai upstream stream failed", + code: "stream_pipeline_error", + type: "stream_error", + }); + const pending = usageHistory.getPendingRequests(); + assert.deepEqual(Object.keys(pending.byModel), []); + assert.deepEqual(Object.keys(pending.byAccount), []); + + const completed = [...usageHistory.getCompletedDetails().values()]; + assert.equal(completed.length, 1); + assert.equal(completed[0].status, 502); + assert.equal(completed[0].error, "1min.ai upstream stream failed"); + assert.equal(completed[0].errorCode, "stream_pipeline_error"); + + const detail = await waitFor(() => getOneMinCallLog(PARTIAL_IDENTITY)); + assert.ok(detail, "the post-content stream failure must be persisted"); + assert.equal(detail.status, 502); + assert.equal(detail.error, "1min.ai upstream stream failed"); + const persisted = JSON.stringify(detail); + assert.match(persisted, /1min\.ai upstream stream failed/); + assert.doesNotMatch(persisted, /post-content-secret/); + assert.doesNotMatch(persisted, /\/srv\/omniroute/); + assert.doesNotMatch(persisted, /stack tail/); + + const usage = await waitFor(() => getOneMinUsage(PARTIAL_IDENTITY)); + assert.ok(usage, "the post-content failure usage record must be persisted"); + assert.equal(usage.success, false); + assert.equal(usage.status, "502"); + assert.equal(usage.errorCode, "stream_pipeline_error"); +}); + +test("1min.ai error completion does not wait for an upstream cancel promise", async () => { + let cancelCalls = 0; + const getCalls = installFetchFactory( + () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode('event: error\ndata: {"message":"capacity unavailable"}\n\n') + ); + }, + cancel() { + cancelCalls += 1; + return new Promise(() => {}); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ) + ); + + const result = await new OneMinAiExecutor().execute({ + model: "gpt-4o-mini", + body: { messages: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { apiKey: "unit-test-key" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + const clientText = await Promise.race([ + result.response.text(), + new Promise((_resolve, reject) => + setTimeout(() => reject(new Error("translated stream stayed pending on cancel")), 500) + ), + ]); + + assert.equal(getCalls(), 1); + assert.equal(cancelCalls, 1); + assert.match(clientText, /capacity unavailable/); + assert.match(clientText, /data: \[DONE\]/); +}); + +test("1min.ai propagates downstream cancellation without awaiting upstream cleanup", async () => { + let upstreamController: ReadableStreamDefaultController | null = null; + let cancelCalls = 0; + let markPullStarted: (() => void) | null = null; + const pullStarted = new Promise((resolve) => { + markPullStarted = resolve; + }); + const getCalls = installFetchFactory( + () => + new Response( + new ReadableStream({ + start(controller) { + upstreamController = controller; + controller.enqueue( + encoder.encode('event: content\ndata: {"content":"partial answer"}\n\n') + ); + }, + pull() { + markPullStarted?.(); + return new Promise(() => {}); + }, + cancel() { + cancelCalls += 1; + return new Promise(() => {}); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ) + ); + + const result = await new OneMinAiExecutor().execute({ + model: "gpt-4o-mini", + body: { messages: [{ role: "user", content: "hello" }] }, + stream: true, + credentials: { apiKey: "unit-test-key" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + assert.ok(result.response.body); + const reader = result.response.body.getReader(); + + try { + const clientText = await readUntil(reader, "partial answer"); + assert.match(clientText, /"role":"assistant"/); + await pullStarted; + await Promise.race([ + reader.cancel("client disconnected"), + new Promise((_resolve, reject) => + setTimeout(() => reject(new Error("downstream cancellation stayed pending")), 500) + ), + ]); + + assert.equal(getCalls(), 1); + assert.equal(cancelCalls, 1, "downstream cancellation must reach the upstream reader once"); + assert.deepEqual(await reader.read(), { value: undefined, done: true }); + } finally { + try { + upstreamController?.close(); + } catch { + // The fixed path has already cancelled and closed the upstream stream. + } + } +}); + +test("1min.ai accepts the bounded error-string shape without exposing a success chunk", async () => { + const response = await executeStreaming([ + 'event: error\ndata: {"error":"billing temporarily unavailable"}\n\n', + ]); + const clientText = await response.text(); + + assert.match(clientText, /"error":\{"message":"billing temporarily unavailable"/); + assert.doesNotMatch(clientText, /"role":"assistant"/); + assert.doesNotMatch(clientText, /"finish_reason":"stop"/); +}); + +test("1min.ai replaces oversized stream-error payloads with a fixed public fallback", async () => { + const oversizedMessage = `private-prefix-${"x".repeat(70 * 1024)}`; + const response = await executeStreaming([ + `event: error\ndata: ${JSON.stringify({ message: oversizedMessage })}\n\n`, + ]); + const clientText = await response.text(); + + assert.match(clientText, /1min\.ai upstream stream failed/); + assert.ok(clientText.length < 1_024, "the oversized upstream payload must not be reflected"); + assert.doesNotMatch(clientText, /private-prefix/); + assert.doesNotMatch(clientText, /"role":"assistant"/); + assert.doesNotMatch(clientText, /"finish_reason":"stop"/); +}); diff --git a/tests/unit/oneminai-stream-error-boundary.test.ts b/tests/unit/oneminai-stream-error-boundary.test.ts new file mode 100644 index 0000000000..88cc402e6e --- /dev/null +++ b/tests/unit/oneminai-stream-error-boundary.test.ts @@ -0,0 +1,91 @@ +import assert from "node:assert/strict"; +import { execFile } from "node:child_process"; +import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; + +const execFileAsync = promisify(execFile); +const fixturePath = fileURLToPath( + new URL("../fixtures/oneminai-stream-error-boundary.fixture.ts", import.meta.url) +); + +type ChildFailure = Error & { + stdout?: string | Buffer; + stderr?: string | Buffer; +}; + +test( + "1min.ai stream-error boundary passes in an isolated persistence subprocess", + { timeout: 180_000 }, + async () => { + const originalDataDir = process.env.DATA_DIR; + const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR; + const originalFetch = globalThis.fetch; + const testRoot = mkdtempSync(join(tmpdir(), "omniroute-onemin-stream-error-child-")); + const testDataDir = join(testRoot, "data"); + const testPluginsDir = join(testRoot, "plugins"); + + mkdirSync(testDataDir, { recursive: true }); + mkdirSync(testPluginsDir, { recursive: true }); + const childEnv: NodeJS.ProcessEnv = { + APP_LOG_TO_FILE: "false", + DATA_DIR: testDataDir, + DISABLE_SQLITE_AUTO_BACKUP: "true", + NODE_ENV: "test", + OMNIROUTE_PLUGINS_DIR: testPluginsDir, + }; + for (const name of ["PATH", "NODE_PATH", "LANG", "LC_ALL", "TZ", "TMPDIR"] as const) { + const value = process.env[name]; + if (value !== undefined) childEnv[name] = value; + } + // A nested `node --test` must create its own runner context instead of + // inheriting the parent's private reporter channel. + delete childEnv.NODE_TEST_CONTEXT; + + try { + let stdout = ""; + let stderr = ""; + try { + const child = await execFileAsync( + process.execPath, + ["--import", "tsx/esm", "--test", "--test-concurrency=1", fixturePath], + { + cwd: process.cwd(), + encoding: "utf8", + env: childEnv, + maxBuffer: 2 * 1024 * 1024, + timeout: 170_000, + } + ); + stdout = child.stdout; + stderr = child.stderr; + } catch (error) { + const failure = error as ChildFailure; + assert.fail( + [ + `isolated 1min.ai fixture failed: ${failure.message}`, + failure.stdout ? String(failure.stdout) : "", + failure.stderr ? String(failure.stderr) : "", + ] + .filter(Boolean) + .join("\n") + ); + } + + const childOutput = `${stdout}\n${stderr}`; + assert.match(childOutput, /tests 8/); + assert.match(childOutput, /pass 8/); + assert.match(childOutput, /fail 0/); + assert.doesNotMatch(childOutput, /not ok|failed to drain|stayed pending/i); + } finally { + rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); + } + + assert.equal(process.env.DATA_DIR, originalDataDir); + assert.equal(process.env.OMNIROUTE_PLUGINS_DIR, originalPluginsDir); + assert.equal(globalThis.fetch, originalFetch); + } +); diff --git a/tests/unit/stream-readiness.test.ts b/tests/unit/stream-readiness.test.ts index 7e2ea9c242..04432fc62c 100644 --- a/tests/unit/stream-readiness.test.ts +++ b/tests/unit/stream-readiness.test.ts @@ -451,27 +451,24 @@ test("ensureStreamReadiness preserves buffered chunks when stream starts", async assert.match(text, / world/); }); -test("ensureStreamReadiness preserves its buffered prefix until a delayed consumer observes a later error", async () => { - const prefix = `data: ${JSON.stringify({ - object: "chat.completion.chunk", - choices: [ - { - index: 0, - delta: { role: "assistant", content: "prefix before failure" }, - finish_reason: null, - }, - ], - })}\n\n`; - let pullCount = 0; +test("ensureStreamReadiness replays buffered chunks before a subsequent source error", async () => { + let reads = 0; const response = new Response( new ReadableStream({ pull(controller) { - pullCount += 1; - if (pullCount === 1) { - controller.enqueue(encoder.encode(prefix)); + reads += 1; + if (reads === 1) { + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { role: "assistant", content: "prefix" } }], + })}\n\n` + ) + ); return; } - controller.error(new Error("later upstream failure")); + controller.error(Object.assign(new Error("terminal source failure"), { statusCode: 502 })); }, }), { status: 200, headers: { "Content-Type": "text/event-stream" } } @@ -479,13 +476,96 @@ test("ensureStreamReadiness preserves its buffered prefix until a delayed consum const result = await ensureStreamReadiness(response, { timeoutMs: 100 }); assert.equal(result.ok, true); - await new Promise((resolve) => setTimeout(resolve, 25)); + assert.ok(result.response.body); + const reader = result.response.body.getReader(); + const first = await reader.read(); - const reader = result.response.body!.getReader(); + assert.equal(first.done, false); + assert.match(new TextDecoder().decode(first.value), /prefix/); + await assert.rejects(reader.read(), /terminal source failure/); +}); + +test("ensureStreamReadiness replays multiple buffered chunks in order before an error", async () => { + let reads = 0; + const response = new Response( + new ReadableStream({ + pull(controller) { + reads += 1; + if (reads === 1) { + controller.enqueue(encoder.encode(": keepalive\n\n")); + return; + } + if (reads === 2) { + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { role: "assistant", content: "ready" } }], + })}\n\n` + ) + ); + return; + } + controller.error(new Error("failure after buffered prefix")); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); + + const result = await ensureStreamReadiness(response, { timeoutMs: 100 }); + assert.equal(result.ok, true); + assert.ok(result.response.body); + const reader = result.response.body.getReader(); + const first = await reader.read(); + const second = await reader.read(); + + assert.equal(first.done, false); + assert.equal(second.done, false); + assert.match(new TextDecoder().decode(first.value), /keepalive/); + assert.match(new TextDecoder().decode(second.value), /ready/); + await assert.rejects(reader.read(), /failure after buffered prefix/); +}); + +test("ensureStreamReadiness cancellation is bounded when upstream cancel never settles", async () => { + let cancelCalls = 0; + const response = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { role: "assistant", content: "prefix" } }], + })}\n\n` + ) + ); + }, + pull() { + return new Promise(() => {}); + }, + cancel() { + cancelCalls += 1; + return new Promise(() => {}); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); + + const result = await ensureStreamReadiness(response, { timeoutMs: 100 }); + assert.equal(result.ok, true); + assert.ok(result.response.body); + const reader = result.response.body.getReader(); const first = await reader.read(); assert.equal(first.done, false); - assert.match(new TextDecoder().decode(first.value), /prefix before failure/); - await assert.rejects(() => reader.read(), /later upstream failure/); + + await Promise.race([ + reader.cancel("client disconnected"), + new Promise((_resolve, reject) => + setTimeout(() => reject(new Error("readiness cancellation stayed pending")), 500) + ), + ]); + await reader.cancel("duplicate cancellation"); + assert.equal(cancelCalls, 1); }); test("ensureStreamReadiness honors configured timeouts above 2000ms", async () => {