From ec0ccb28d386fea08c501562472ea9aa1fd85628 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 2 Sep 2026 08:02:34 -0300 Subject: [PATCH] fix(sse): preserve Perplexity stream failures --- ...ENDING-perplexity-stream-error-boundary.md | 1 + open-sse/executors/perplexity-web.ts | 353 ++++++---- open-sse/executors/perplexity-web/protocol.ts | 30 +- open-sse/utils/streamReadiness.ts | 62 +- ...exity-web-stream-error-boundary.fixture.ts | 648 ++++++++++++++++++ ...rplexity-web-stream-error-boundary.test.ts | 90 +++ tests/unit/stream-readiness.test.ts | 53 +- 7 files changed, 1064 insertions(+), 173 deletions(-) create mode 100644 changelog.d/fixes/PENDING-perplexity-stream-error-boundary.md create mode 100644 tests/fixtures/perplexity-web-stream-error-boundary.fixture.ts create mode 100644 tests/unit/perplexity-web-stream-error-boundary.test.ts diff --git a/changelog.d/fixes/PENDING-perplexity-stream-error-boundary.md b/changelog.d/fixes/PENDING-perplexity-stream-error-boundary.md new file mode 100644 index 0000000000..0d9a14a2db --- /dev/null +++ b/changelog.d/fixes/PENDING-perplexity-stream-error-boundary.md @@ -0,0 +1 @@ +- **fix(providers):** Perplexity Web no longer turns upstream stream failures into successful assistant text; pre-content failures remain eligible for fallback, partial output ends with a structured sanitized error, and failed sessions are not persisted diff --git a/open-sse/executors/perplexity-web.ts b/open-sse/executors/perplexity-web.ts index 51f774ea87..8451fe8379 100644 --- a/open-sse/executors/perplexity-web.ts +++ b/open-sse/executors/perplexity-web.ts @@ -17,6 +17,7 @@ import { prepareToolMessages } from "../translator/webTools.ts"; import { buildToolModeResponse } from "./chatgptWebTools.ts"; import { sanitizeErrorMessage } from "../utils/error.ts"; import { buildSessionCookieHeader, mergeRefreshedCookie } from "../utils/nextAuthCookie.ts"; +import { formatTranslatedStreamError } from "../utils/streamErrorFormat.ts"; import { PPLX_SSE_ENDPOINT, PPLX_USER_AGENT, @@ -35,6 +36,8 @@ import { const SESSION_MAX_AGE_MS = 3600_000; const SESSION_MAX_ENTRIES = 200; +const PPLX_STREAM_ERROR_MESSAGE = "Perplexity upstream stream failed"; +const PPLX_STREAM_ERROR_CODE = "PPLX_STREAM_ERROR"; interface SessionEntry { backendUuid: string; @@ -102,155 +105,223 @@ function buildStreamingResponse( signal?: AbortSignal | null ): ReadableStream { const encoder = new TextEncoder(); + const streamAbortController = new AbortController(); + const forwardInputAbort = () => + streamAbortController.abort(signal?.reason ?? "perplexity_request_aborted"); + if (signal?.aborted) forwardInputAbort(); + else signal?.addEventListener("abort", forwardInputAbort, { once: true }); + let inputAbortListenerAttached = Boolean(signal && !signal.aborted); + const removeInputAbortListener = () => { + if (!inputAbortListenerAttached) return; + inputAbortListenerAttached = false; + signal?.removeEventListener("abort", forwardInputAbort); + }; + const abortEventStream = (reason: unknown) => { + removeInputAbortListener(); + if (!streamAbortController.signal.aborted) streamAbortController.abort(reason); + }; + const contentIterator = extractContent(eventStream, streamAbortController.signal)[ + Symbol.asyncIterator + ](); + let fullAnswer = ""; + let respBackendUuid: string | null = null; + let roleEmitted = false; + let finished = false; + let pendingFailure: (Error & { statusCode: number }) | null = null; - return new ReadableStream( - { - async start(controller) { - 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 }, - ], - }) - ) - ); + const enqueuePreContentFailure = (controller: ReadableStreamDefaultController) => { + controller.enqueue( + encoder.encode( + formatTranslatedStreamError({ + status: 502, + message: PPLX_STREAM_ERROR_MESSAGE, + type: "upstream_error", + code: PPLX_STREAM_ERROR_CODE, + }) + ) + ); + }; - let fullAnswer = ""; - let respBackendUuid: string | null = null; + const takeAssistantRoleChunk = (): string => { + if (roleEmitted) return ""; + roleEmitted = true; + return sseChunk({ + id: cid, + object: "chat.completion.chunk", + created, + model, + system_fingerprint: null, + choices: [ + { + index: 0, + delta: { role: "assistant" }, + finish_reason: null, + logprobs: null, + }, + ], + }); + }; - for await (const chunk of extractContent(eventStream, signal)) { - if (chunk.backendUuid) respBackendUuid = chunk.backendUuid; + const completeStream = (controller: ReadableStreamDefaultController) => { + if (finished) return; + finished = true; + controller.enqueue( + encoder.encode( + sseChunk({ + id: cid, + object: "chat.completion.chunk", + created, + model, + system_fingerprint: null, + choices: [{ index: 0, delta: {}, finish_reason: "stop", logprobs: null }], + }) + ) + ); + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + sessionStore(history, currentMsg, cleanResponse(fullAnswer), respBackendUuid); + removeInputAbortListener(); + controller.close(); + }; - if (chunk.error) { - controller.enqueue( - encoder.encode( - sseChunk({ - id: cid, - object: "chat.completion.chunk", - created, - model, - system_fingerprint: null, - choices: [ - { - index: 0, - delta: { content: `[Error: ${chunk.error}]` }, - finish_reason: null, - logprobs: null, - }, - ], - }) - ) - ); - break; - } + const failStream = (controller: ReadableStreamDefaultController) => { + if (roleEmitted) { + pendingFailure = Object.assign(new Error(PPLX_STREAM_ERROR_MESSAGE), { + statusCode: 502, + }); + finished = true; + controller.close(); + return; + } + finished = true; + enqueuePreContentFailure(controller); + controller.close(); + }; - if (chunk.thinking) { - controller.enqueue( - encoder.encode( - sseChunk({ - id: cid, - object: "chat.completion.chunk", - created, - model, - system_fingerprint: null, - choices: [ - { - index: 0, - delta: { reasoning_content: chunk.thinking + "\n" }, - finish_reason: null, - logprobs: null, - }, - ], - }) - ) - ); - continue; - } + const providerStream = new ReadableStream({ + async pull(controller) { + if (finished) return; - if (chunk.done) { - fullAnswer = chunk.answer || fullAnswer; - break; - } - - let dt = chunk.delta || ""; - if (dt) { - dt = cleanResponse(dt, false); - if (dt) { - controller.enqueue( - encoder.encode( - sseChunk({ - id: cid, - object: "chat.completion.chunk", - created, - model, - system_fingerprint: null, - choices: [ - { index: 0, delta: { content: dt }, finish_reason: null, logprobs: null }, - ], - }) - ) - ); - } - } - if (chunk.answer) fullAnswer = chunk.answer; - } - - // Stop chunk - controller.enqueue( - encoder.encode( - sseChunk({ - id: cid, - object: "chat.completion.chunk", - created, - model, - system_fingerprint: null, - choices: [{ index: 0, delta: {}, finish_reason: "stop", logprobs: null }], - }) - ) - ); - controller.enqueue(encoder.encode("data: [DONE]\n\n")); - - sessionStore(history, currentMsg, cleanResponse(fullAnswer), respBackendUuid); - } catch (err) { - controller.enqueue( - encoder.encode( - sseChunk({ - id: cid, - object: "chat.completion.chunk", - created, - model, - system_fingerprint: null, - choices: [ - { - index: 0, - delta: { - content: `[Stream error: ${err instanceof Error ? err.message : String(err)}]`, - }, - finish_reason: "stop", - logprobs: null, - }, - ], - }) - ) - ); - controller.enqueue(encoder.encode("data: [DONE]\n\n")); - } finally { - try { - controller.close(); - } catch {} + try { + const next = await contentIterator.next(); + if (streamAbortController.signal.aborted) { + finished = true; + controller.close(); + return; } - }, + if (next.done === true) { + completeStream(controller); + return; + } + + const chunk = next.value; + if (chunk.backendUuid) respBackendUuid = chunk.backendUuid; + + if (chunk.error) { + failStream(controller); + removeInputAbortListener(); + void contentIterator.return?.(undefined).catch(() => undefined); + return; + } + + if (chunk.thinking) { + controller.enqueue( + encoder.encode( + takeAssistantRoleChunk() + + sseChunk({ + id: cid, + object: "chat.completion.chunk", + created, + model, + system_fingerprint: null, + choices: [ + { + index: 0, + delta: { reasoning_content: chunk.thinking + "\n" }, + finish_reason: null, + logprobs: null, + }, + ], + }) + ) + ); + return; + } + + if (chunk.done) { + fullAnswer = chunk.answer || fullAnswer; + completeStream(controller); + await contentIterator.return?.(undefined); + return; + } + + let dt = chunk.delta || ""; + if (dt) { + dt = cleanResponse(dt, false); + if (dt) { + controller.enqueue( + encoder.encode( + takeAssistantRoleChunk() + + sseChunk({ + id: cid, + object: "chat.completion.chunk", + created, + model, + system_fingerprint: null, + choices: [ + { index: 0, delta: { content: dt }, finish_reason: null, logprobs: null }, + ], + }) + ) + ); + } + } + if (chunk.answer) fullAnswer = chunk.answer; + } catch { + failStream(controller); + removeInputAbortListener(); + void contentIterator.return?.(undefined).catch(() => undefined); + } }, - { highWaterMark: 16384 } - ); + + cancel(reason) { + finished = true; + abortEventStream(reason); + void contentIterator.return?.(undefined).catch(() => undefined); + }, + }); + + // Erroring the provider stream immediately would discard output buffered by the readiness + // handoff. Drain each provider chunk through a backpressure-aware reader, then reject only the + // read after the last legitimate chunk. The outer pipeline converts that fixed public error to + // the client's canonical terminal frame and records the stream failure. + const providerReader = providerStream.getReader(); + let cancelled = false; + return new ReadableStream({ + async pull(controller) { + try { + const next = await providerReader.read(); + if (cancelled) return; + if (next.done === false) { + controller.enqueue(next.value); + return; + } + if (pendingFailure) { + controller.error(pendingFailure); + return; + } + controller.close(); + } catch (error) { + if (!cancelled) controller.error(error); + } + }, + + cancel(reason) { + if (cancelled) return; + cancelled = true; + abortEventStream(reason); + void providerReader.cancel(reason).catch(() => undefined); + }, + }); } async function buildNonStreamingResponse( diff --git a/open-sse/executors/perplexity-web/protocol.ts b/open-sse/executors/perplexity-web/protocol.ts index 12e98ccdc4..f3f8a318c5 100644 --- a/open-sse/executors/perplexity-web/protocol.ts +++ b/open-sse/executors/perplexity-web/protocol.ts @@ -213,6 +213,19 @@ export async function* readPplxSseEvents( const decoder = new TextDecoder(); let buffer = ""; let dataLines: string[] = []; + let readerFinished = false; + let readerCancelRequested = false; + + const cancelReader = (reason: unknown) => { + if (readerFinished || readerCancelRequested) return; + readerCancelRequested = true; + // Cancellation is a client-facing latency boundary. Request upstream cleanup once, but never + // await a hostile underlying source whose cancel hook does not settle. + void reader.cancel(reason).catch(() => undefined); + }; + const handleAbort = () => cancelReader(signal?.reason ?? "perplexity_stream_aborted"); + if (signal?.aborted) handleAbort(); + else signal?.addEventListener("abort", handleAbort, { once: true }); function flush(): PplxStreamEvent | null | "done" { if (dataLines.length === 0) return null; @@ -231,7 +244,10 @@ export async function* readPplxSseEvents( while (true) { if (signal?.aborted) return; const { value, done } = await reader.read(); - if (done) break; + if (done) { + readerFinished = true; + break; + } buffer += decoder.decode(value, { stream: true }); while (true) { @@ -263,7 +279,13 @@ export async function* readPplxSseEvents( const tail = flush(); if (tail && tail !== "done") yield tail; } finally { - reader.releaseLock(); + signal?.removeEventListener("abort", handleAbort); + cancelReader(signal?.reason ?? "perplexity_stream_reader_closed"); + try { + reader.releaseLock(); + } catch { + // A hostile source may keep its cancel promise pending; the lock can be released later by GC. + } } } @@ -915,6 +937,10 @@ export async function* extractContent( } } + // Cancellation is not a successful terminal event. In particular, do not synthesize the final + // `done` chunk: streaming callers use that signal to emit stop/[DONE] and persist the session. + if (signal?.aborted) return; + // End-of-stream without a COMPLETED frame still try the last text blob. if (!fullAnswer.trim() && lastEventText) { const fromText = extractAnswerFromFinalText(lastEventText); diff --git a/open-sse/utils/streamReadiness.ts b/open-sse/utils/streamReadiness.ts index 4bbeef1e0e..3774cc0260 100644 --- a/open-sse/utils/streamReadiness.ts +++ b/open-sse/utils/streamReadiness.ts @@ -421,29 +421,57 @@ function prependBufferedChunks( chunks: Uint8Array[], reader: ReadableStreamDefaultReader ): ReadableStream { + let bufferedIndex = 0; + let cancelled = false; + let readerReleased = false; + + const releaseReader = () => { + if (readerReleased) return; + readerReleased = true; + try { + reader.releaseLock(); + } 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. + } + }; + return new ReadableStream({ - async start(controller) { + async pull(controller) { + if (cancelled) 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. + if (bufferedIndex < chunks.length) { + controller.enqueue(chunks[bufferedIndex]); + bufferedIndex += 1; + return; + } + try { - for (const chunk of chunks) { - controller.enqueue(chunk); + const { done, value } = await reader.read(); + if (cancelled) return; + if (done) { + releaseReader(); + controller.close(); + return; } - - while (true) { - const { done, value } = await reader.read(); - if (done) break; - if (value) controller.enqueue(value); - } - - controller.close(); + if (value) controller.enqueue(value); } catch (error) { - controller.error(error); - } finally { - reader.releaseLock(); + releaseReader(); + if (!cancelled) controller.error(error); } }, - async cancel(reason) { - await reader.cancel(reason).catch(() => {}); - reader.releaseLock(); + 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); }, }); } diff --git a/tests/fixtures/perplexity-web-stream-error-boundary.fixture.ts b/tests/fixtures/perplexity-web-stream-error-boundary.fixture.ts new file mode 100644 index 0000000000..8d02adbbad --- /dev/null +++ b/tests/fixtures/perplexity-web-stream-error-boundary.fixture.ts @@ -0,0 +1,648 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +assert.ok(process.env.DATA_DIR, "the subprocess fixture requires an isolated DATA_DIR"); +assert.ok( + process.env.OMNIROUTE_PLUGINS_DIR, + "the subprocess fixture requires an isolated OMNIROUTE_PLUGINS_DIR" +); + +const core = await import("../../src/lib/db/core.ts"); +const { getUsageHistory } = await import("../../src/lib/usage/usageHistory.ts"); +const { waitForCallLogSaves } = await import("../../src/lib/usage/callLogs.ts"); +const { closeCallLogArtifactWriter } = await import("../../src/lib/usage/callLogArtifactWriter.ts"); +const { PerplexityWebExecutor } = await import("../../open-sse/executors/perplexity-web.ts"); +const { __setTlsFetchOverrideForTesting } = + await import("../../open-sse/services/perplexityTlsClient.ts"); +const { handleChatCore } = await import("../../open-sse/handlers/chatCore.ts"); + +type StreamFailure = { status: number; message: string; code?: string; type?: string }; + +function createPerplexityStream( + events: Array> +): ReadableStream { + const encoder = new TextEncoder(); + const payload = + events.map((event) => `event: message\r\ndata: ${JSON.stringify(event)}\r\n\r\n`).join("") + + "event: end_of_stream\r\n\r\n"; + return new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(payload)); + controller.close(); + }, + }); +} + +async function executeWithUpstreamBody( + body: ReadableStream, + prompt = "hi" +): Promise { + __setTlsFetchOverrideForTesting(async () => ({ + status: 200, + headers: new Headers({ "Content-Type": "text/event-stream" }), + text: null, + body, + })); + + const executor = new PerplexityWebExecutor(); + const result = await executor.execute({ + model: "pplx-auto", + body: { messages: [{ role: "user", content: prompt }], stream: true }, + stream: true, + credentials: { apiKey: "test-cookie" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + return result.response; +} + +function executeStreaming( + events: Array>, + prompt = "hi" +): Promise { + return executeWithUpstreamBody(createPerplexityStream(events), prompt); +} + +async function executeThroughChatCore( + events: Array>, + prompt = "hi", + onStreamFailure?: (failure: StreamFailure) => void, + onRequestSuccess?: () => Promise, + model = "pplx-auto" +) { + return executeBodyThroughChatCore( + createPerplexityStream(events), + prompt, + onStreamFailure, + onRequestSuccess, + model + ); +} + +async function executeBodyThroughChatCore( + upstreamBody: ReadableStream, + prompt = "hi", + onStreamFailure?: (failure: StreamFailure) => void, + onRequestSuccess?: () => Promise, + model = "pplx-auto" +) { + __setTlsFetchOverrideForTesting(async () => ({ + status: 200, + headers: new Headers({ "Content-Type": "text/event-stream" }), + text: null, + body: upstreamBody, + })); + const body = { + model, + messages: [{ role: "user", content: prompt }], + stream: true, + }; + return handleChatCore({ + body: structuredClone(body), + modelInfo: { provider: "perplexity-web", model, extendedContext: false }, + credentials: { apiKey: "test-cookie", providerSpecificData: {} }, + log: { debug() {}, info() {}, warn() {}, error() {} }, + onRequestSuccess, + onStreamFailure, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: structuredClone(body), + headers: new Headers({ accept: "text/event-stream" }), + }, + userAgent: "perplexity-stream-error-boundary-test", + skipResourcePressureGuard: true, + }); +} + +function assertNoSensitiveDetail(value: string): void { + assert.doesNotMatch(value, /private-runtime\.ts/); + assert.doesNotMatch(value, /sk-pplx-secret/); + assert.doesNotMatch(value, /api_key/); +} + +function assertChatCompletionWire(value: string): void { + assert.doesNotMatch(value, /^event:/m, "Chat Completions must not receive Responses framing"); + const payloads = value + .split(/\r?\n/) + .filter((line) => line.startsWith("data:") && line.slice(5).trim() !== "[DONE]") + .map((line) => JSON.parse(line.slice(5).trim()) as Record); + assert.ok(payloads.length > 0); + for (const payload of payloads) { + assert.equal(payload.object, "chat.completion.chunk"); + assert.ok(Array.isArray(payload.choices)); + for (const choice of payload.choices as Array>) { + assert.equal(choice.index, 0); + assert.equal(typeof choice.delta, "object"); + assert.ok( + choice.finish_reason === null || typeof choice.finish_reason === "string", + "finish_reason must remain Chat Completions-compatible" + ); + } + } + assert.match(value, /data: \[DONE\]/); +} + +async function waitForPersistedStreamFailure(startedAt: Date, model = "pplx-auto") { + for (let attempt = 0; attempt < 80; attempt++) { + const rows = await getUsageHistory({ + provider: "perplexity-web", + model, + startDate: startedAt, + }); + const failure = rows.find( + (row) => + row.success === false && row.status === "502" && row.errorCode === "stream_pipeline_error" + ); + if (failure) return failure; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return null; +} + +async function waitForCondition(predicate: () => boolean, timeoutMs = 500): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return true; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return predicate(); +} + +test.afterEach(() => { + __setTlsFetchOverrideForTesting(null); +}); + +test.after(async () => { + __setTlsFetchOverrideForTesting(null); + assert.equal( + await waitForCallLogSaves(3_000), + true, + "call-log writes must drain before the isolated DATA_DIR is removed" + ); + await closeCallLogArtifactWriter(); + core.resetDbInstance(); +}); + +test("pre-content Perplexity failures remain unready and return a sanitized 502", async () => { + const result = await executeThroughChatCore([ + { + error_code: "PPLX_ERROR", + error_message: + "failed at /srv/omniroute/private-runtime.ts:42:7 token=sk-pplx-secret-123456 api_key=hidden", + }, + ]); + + assert.equal(result.success, false, "the handler must expose a fallback-eligible failure"); + assert.equal(result.status, 502); + assert.equal(result.response.status, 502); + assert.equal(result.errorCode, "STREAM_EARLY_EOF"); + const body = await result.response.text(); + assert.match(body, /"error"/); + assertNoSensitiveDetail(body); +}); + +test("thrown Perplexity stream failures remain unready and return a sanitized 502", async () => { + const result = await executeBodyThroughChatCore( + new ReadableStream({ + start(controller) { + controller.error( + new Error( + "socket failed at /srv/omniroute/private-runtime.ts:51:9 token=sk-pplx-secret-catch" + ) + ); + }, + }), + "throw before content" + ); + + assert.equal(result.success, false, "the handler must expose a fallback-eligible failure"); + assert.equal(result.status, 502); + assert.equal(result.response.status, 502); + const responseBody = await result.response.text(); + assert.match(responseBody, /"error"/); + assertNoSensitiveDetail(responseBody); +}); + +test("thrown failures after content preserve the prefix and terminate as a safe error", async () => { + const encoder = new TextEncoder(); + const partialAnswer = "partial before transport failure"; + let upstreamRead = false; + let finalizedFailure: StreamFailure | null = null; + const upstreamBody = new ReadableStream({ + pull(controller) { + if (!upstreamRead) { + upstreamRead = true; + controller.enqueue( + encoder.encode( + `event: message\r\ndata: ${JSON.stringify({ + backend_uuid: "uuid-thrown-must-not-store", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" }, + }, + ], + status: "PENDING", + })}\r\n\r\n` + ) + ); + return; + } + controller.error( + new Error( + "transport failed at /srv/omniroute/private-runtime.ts:79 token=sk-pplx-secret-after" + ) + ); + }, + }); + + const result = await executeBodyThroughChatCore( + upstreamBody, + "post-content transport failure", + (failure) => { + finalizedFailure = failure; + } + ); + + assert.equal(result.success, true); + const output = await result.response.text(); + assert.match(output, new RegExp(partialAnswer)); + assert.match(output, /"finish_reason":"error"/); + assert.doesNotMatch(output, /"finish_reason":"stop"/); + assert.doesNotMatch(output, /response\.failed/); + assertNoSensitiveDetail(output); + assertChatCompletionWire(output); + assert.ok(finalizedFailure); + assert.equal(finalizedFailure.status, 502); + assert.equal(finalizedFailure.message, "Perplexity upstream stream failed"); +}); + +test("partial content is preserved before a terminal error and the failed session is not stored", async () => { + const firstPrompt = "partial-boundary-first-prompt"; + const partialAnswer = "safe partial answer"; + const requestStartedAt = new Date(Date.now() - 1_000); + let finalizedFailure: StreamFailure | null = null; + const firstResult = await executeThroughChatCore( + [ + { + backend_uuid: "uuid-must-not-be-stored", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" }, + }, + ], + status: "PENDING", + }, + { + error_code: "PPLX_ERROR", + error_message: + "later failure at /srv/omniroute/private-runtime.ts:66:2 token=sk-pplx-secret-partial", + }, + ], + firstPrompt, + (failure) => { + finalizedFailure = failure; + } + ); + + assert.equal(firstResult.success, true, "legitimate partial output must satisfy readiness"); + const output = await firstResult.response.text(); + assert.match(output, /"role":"assistant"/); + assert.match(output, new RegExp(partialAnswer)); + assert.match(output, /"error":\{/); + assert.match(output, /"finish_reason":"error"/); + assert.doesNotMatch(output, /"finish_reason":"stop"/); + assert.doesNotMatch(output, /response\.failed/); + assert.doesNotMatch(output, /event:\s*response\.failed/); + assert.doesNotMatch(output, /"type":"response\.failed"/); + assert.doesNotMatch(output, /\[Error:/); + assertNoSensitiveDetail(output); + assertChatCompletionWire(output); + assert.ok(finalizedFailure, "the downstream pipeline must finalize the stream failure"); + assert.equal(finalizedFailure.status, 502); + assert.equal(finalizedFailure.message, "Perplexity upstream stream failed"); + const persistedFailure = await waitForPersistedStreamFailure(requestStartedAt); + assert.ok(persistedFailure, "the handler must persist the terminal stream failure"); + + let followUpRequestBody: string | undefined; + __setTlsFetchOverrideForTesting(async (_url, options) => { + followUpRequestBody = String(options.body ?? ""); + return { + status: 200, + headers: new Headers({ "Content-Type": "text/event-stream" }), + text: null, + body: createPerplexityStream([ + { + backend_uuid: "next-success-uuid", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: ["next answer"], progress: "DONE" }, + }, + ], + status: "COMPLETED", + }, + ]), + }; + }); + + const executor = new PerplexityWebExecutor(); + const followUp = await executor.execute({ + model: "pplx-auto", + body: { + messages: [ + { role: "user", content: firstPrompt }, + { role: "assistant", content: partialAnswer }, + { role: "user", content: "continue" }, + ], + stream: false, + }, + stream: false, + credentials: { apiKey: "test-cookie" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + assert.equal(followUp.response.status, 200); + assert.ok(followUpRequestBody); + const sent = JSON.parse(followUpRequestBody) as { params?: Record }; + assert.equal( + sent.params?.last_backend_uuid, + undefined, + "a failed partial response must not create a reusable session" + ); +}); + +test("same-packet content and error preserve the prefix across repeated readiness handoffs", async () => { + for (let attempt = 0; attempt < 8; attempt++) { + const partialAnswer = `same-packet partial ${attempt}`; + let finalizedFailure: StreamFailure | null = null; + const result = await executeThroughChatCore( + [ + { + backend_uuid: `uuid-same-packet-${attempt}`, + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" }, + }, + ], + status: "PENDING", + }, + { + error_code: "PPLX_ERROR", + error_message: `same packet private failure ${attempt} token=sk-pplx-secret-repeat`, + }, + ], + `same-packet prompt ${attempt}`, + (failure) => { + finalizedFailure = failure; + } + ); + + assert.equal(result.success, true); + const output = await result.response.text(); + assert.match(output, new RegExp(partialAnswer)); + assert.match(output, /"finish_reason":"error"/); + assert.doesNotMatch(output, /"finish_reason":"stop"/); + assert.doesNotMatch(output, /response\.failed/); + assertNoSensitiveDetail(output); + assertChatCompletionWire(output); + assert.ok(finalizedFailure); + assert.equal(finalizedFailure.status, 502); + } +}); + +test("a delayed success hook cannot erase same-packet content before terminal failure", async () => { + const partialAnswer = "prefix must survive delayed success bookkeeping"; + const model = "pplx-auto-delayed-success-proof"; + const requestStartedAt = new Date(Date.now() - 1_000); + let finalizedFailure: StreamFailure | null = null; + const result = await executeThroughChatCore( + [ + { + backend_uuid: "uuid-delayed-success-hook-must-not-store", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" }, + }, + ], + status: "PENDING", + }, + { + error_code: "PPLX_ERROR", + error_message: "private same-packet failure token=sk-pplx-secret-delayed-hook", + }, + ], + "delayed success hook prompt", + (failure) => { + finalizedFailure = failure; + }, + async () => { + await new Promise((resolve) => setTimeout(resolve, 25)); + }, + model + ); + + assert.equal(result.success, true, "the legitimate prefix must satisfy readiness"); + const output = await result.response.text(); + assert.match(output, new RegExp(partialAnswer)); + assert.match(output, /"finish_reason":"error"/); + assert.doesNotMatch(output, /"finish_reason":"stop"/); + assert.doesNotMatch(output, /response\.failed/); + assertNoSensitiveDetail(output); + assertChatCompletionWire(output); + assert.ok(finalizedFailure); + assert.equal(finalizedFailure.status, 502); + assert.equal(finalizedFailure.message, "Perplexity upstream stream failed"); + assert.ok( + await waitForPersistedStreamFailure(requestStartedAt, model), + "the delayed handoff must still persist the terminal stream failure" + ); +}); + +test("successful streamed completions still store their Perplexity session", async () => { + const firstPrompt = "successful-session-first-prompt"; + const firstAnswer = "successful session answer"; + const firstResponse = await executeStreaming( + [ + { + backend_uuid: "uuid-success-is-stored", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: [firstAnswer], progress: "DONE" }, + }, + ], + status: "COMPLETED", + }, + ], + firstPrompt + ); + const firstOutput = await firstResponse.text(); + assert.match(firstOutput, new RegExp(firstAnswer)); + assert.match(firstOutput, /"finish_reason":"stop"/); + + let followUpRequestBody: string | undefined; + __setTlsFetchOverrideForTesting(async (_url, options) => { + followUpRequestBody = String(options.body ?? ""); + return { + status: 200, + headers: new Headers({ "Content-Type": "text/event-stream" }), + text: null, + body: createPerplexityStream([ + { + backend_uuid: "uuid-next-success", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: ["continued"], progress: "DONE" }, + }, + ], + status: "COMPLETED", + }, + ]), + }; + }); + + const executor = new PerplexityWebExecutor(); + const followUp = await executor.execute({ + model: "pplx-auto", + body: { + messages: [ + { role: "user", content: firstPrompt }, + { role: "assistant", content: firstAnswer }, + { role: "user", content: "continue successful session" }, + ], + stream: false, + }, + stream: false, + credentials: { apiKey: "test-cookie" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + assert.equal(followUp.response.status, 200); + assert.ok(followUpRequestBody); + const sent = JSON.parse(followUpRequestBody) as { params?: Record }; + assert.equal(sent.params?.last_backend_uuid, "uuid-success-is-stored"); +}); + +test("downstream cancellation reaches a stalled Perplexity reader exactly once", async () => { + const encoder = new TextEncoder(); + const partialAnswer = "cancel after this prefix"; + let firstPull = true; + let upstreamCancelCount = 0; + let finalizedFailure: StreamFailure | null = null; + const upstreamBody = new ReadableStream({ + pull(controller) { + if (firstPull) { + firstPull = false; + controller.enqueue( + encoder.encode( + `event: message\r\ndata: ${JSON.stringify({ + backend_uuid: "uuid-cancel-must-not-store", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: [partialAnswer], progress: "IN_PROGRESS" }, + }, + ], + status: "PENDING", + })}\r\n\r\n` + ) + ); + return; + } + return new Promise(() => {}); + }, + cancel() { + upstreamCancelCount += 1; + return new Promise(() => {}); + }, + }); + + const result = await executeBodyThroughChatCore( + upstreamBody, + "cancel stalled stream", + (failure) => { + finalizedFailure = failure; + } + ); + assert.equal(result.success, true); + assert.ok(result.response.body); + const reader = result.response.body.getReader(); + const decoder = new TextDecoder(); + let prefix = ""; + for (let readCount = 0; readCount < 4 && !prefix.includes(partialAnswer); readCount += 1) { + const next = await Promise.race([ + reader.read(), + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 500)), + ]); + if (next === "timeout" || next.done) break; + prefix += decoder.decode(next.value, { stream: true }); + } + + const cancelResult = await Promise.race([ + reader.cancel("client stopped reading").then(() => "settled"), + new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), 500)), + ]); + assert.match(prefix, new RegExp(partialAnswer)); + assert.doesNotMatch(prefix, /"finish_reason":"stop"/); + assert.doesNotMatch(prefix, /data: \[DONE\]/); + assert.equal(cancelResult, "settled", "client cancellation must not await a hostile upstream"); + assert.equal( + await waitForCondition(() => upstreamCancelCount === 1), + true, + "cancellation must reach the real upstream reader" + ); + assert.equal(upstreamCancelCount, 1); + await new Promise((resolve) => setTimeout(resolve, 25)); + assert.equal(finalizedFailure, null, "client cancellation must not finalize as provider failure"); + + let followUpRequestBody: string | undefined; + __setTlsFetchOverrideForTesting(async (_url, options) => { + followUpRequestBody = String(options.body ?? ""); + return { + status: 200, + headers: new Headers({ "Content-Type": "text/event-stream" }), + text: null, + body: createPerplexityStream([ + { + backend_uuid: "uuid-after-cancel", + blocks: [ + { + intended_usage: "markdown", + markdown_block: { chunks: ["answer after cancel"], progress: "DONE" }, + }, + ], + status: "COMPLETED", + }, + ]), + }; + }); + const executor = new PerplexityWebExecutor(); + const followUp = await executor.execute({ + model: "pplx-auto", + body: { + messages: [ + { role: "user", content: "cancel stalled stream" }, + { role: "assistant", content: partialAnswer }, + { role: "user", content: "continue after cancellation" }, + ], + stream: false, + }, + stream: false, + credentials: { apiKey: "test-cookie" }, + signal: AbortSignal.timeout(10_000), + log: null, + }); + assert.equal(followUp.response.status, 200); + assert.ok(followUpRequestBody); + const sent = JSON.parse(followUpRequestBody) as { params?: Record }; + assert.equal( + sent.params?.last_backend_uuid, + undefined, + "a cancelled response must not create a reusable session" + ); +}); diff --git a/tests/unit/perplexity-web-stream-error-boundary.test.ts b/tests/unit/perplexity-web-stream-error-boundary.test.ts new file mode 100644 index 0000000000..57daec0c15 --- /dev/null +++ b/tests/unit/perplexity-web-stream-error-boundary.test.ts @@ -0,0 +1,90 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import test from "node:test"; + +const FIXTURE = fileURLToPath( + new URL("../fixtures/perplexity-web-stream-error-boundary.fixture.ts", import.meta.url) +); + +type FixtureResult = { + code: number | null; + signal: NodeJS.Signals | null; + stdout: string; + stderr: string; +}; + +function runIsolatedFixture(testRoot: string): Promise { + const dataDir = path.join(testRoot, "data"); + const pluginsDir = path.join(testRoot, "plugins"); + fs.mkdirSync(dataDir, { recursive: true }); + fs.mkdirSync(pluginsDir, { recursive: true }); + const childEnv: NodeJS.ProcessEnv = { + PATH: process.env.PATH, + NODE_PATH: process.env.NODE_PATH, + LANG: process.env.LANG ?? "C.UTF-8", + LC_ALL: process.env.LC_ALL, + TZ: process.env.TZ ?? "UTC", + TMPDIR: process.env.TMPDIR ?? os.tmpdir(), + NODE_ENV: "test", + APP_LOG_TO_FILE: "false", + API_KEY_SECRET: "perplexity-stream-boundary-test-secret-00000000000000000000000000000000", + DATA_DIR: dataDir, + OMNIROUTE_PLUGINS_DIR: pluginsDir, + }; + delete childEnv.NODE_TEST_CONTEXT; + + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, ["--import", "tsx/esm", "--test", FIXTURE], { + cwd: process.cwd(), + env: childEnv, + stdio: ["ignore", "pipe", "pipe"], + timeout: 90_000, + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", (chunk: string) => { + stdout += chunk; + }); + child.stderr.setEncoding("utf8").on("data", (chunk: string) => { + stderr += chunk; + }); + child.once("error", reject); + child.once("close", (code, signal) => resolve({ code, signal, stdout, stderr })); + }); +} + +test("Perplexity stream failures preserve protocol semantics in an isolated full pipeline", async () => { + const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-pplx-boundary-parent-")); + const originalDataDir = process.env.DATA_DIR; + const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR; + const eventBusOwner = globalThis as { __omnirouteEventBus?: unknown }; + const originalEventBus = eventBusOwner.__omnirouteEventBus; + + try { + const result = await runIsolatedFixture(testRoot); + assert.equal( + result.code, + 0, + `isolated Perplexity fixture failed (signal=${result.signal ?? "none"})\n` + + `stdout:\n${result.stdout}\nstderr:\n${result.stderr}` + ); + assert.equal(result.signal, null); + assert.match(result.stdout, /ℹ tests 8/); + assert.match(result.stdout, /ℹ pass 8/); + assert.match(result.stdout, /ℹ fail 0/); + + assert.equal(process.env.DATA_DIR, originalDataDir); + assert.equal(process.env.OMNIROUTE_PLUGINS_DIR, originalPluginsDir); + assert.equal( + eventBusOwner.__omnirouteEventBus, + originalEventBus, + "the subprocess fixture must not replace the parent event bus singleton" + ); + } finally { + fs.rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); + } +}); diff --git a/tests/unit/stream-readiness.test.ts b/tests/unit/stream-readiness.test.ts index b2ea196818..7e2ea9c242 100644 --- a/tests/unit/stream-readiness.test.ts +++ b/tests/unit/stream-readiness.test.ts @@ -451,6 +451,43 @@ 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; + const response = new Response( + new ReadableStream({ + pull(controller) { + pullCount += 1; + if (pullCount === 1) { + controller.enqueue(encoder.encode(prefix)); + return; + } + controller.error(new Error("later upstream failure")); + }, + }), + { status: 200, headers: { "Content-Type": "text/event-stream" } } + ); + + const result = await ensureStreamReadiness(response, { timeoutMs: 100 }); + assert.equal(result.ok, true); + await new Promise((resolve) => setTimeout(resolve, 25)); + + 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/); +}); + test("ensureStreamReadiness honors configured timeouts above 2000ms", async () => { const response = new Response( streamFromChunks( @@ -616,10 +653,7 @@ test("ensureStreamReadiness preserves sanitized error-only diagnostics on early assert.equal(result.response.status, 502); assert.equal(result.code, "STREAM_EARLY_EOF"); assert.equal(result.type, "stream_early_eof"); - assert.equal( - result.classificationReason, - "Stream ended before producing a non-ping SSE event" - ); + assert.equal(result.classificationReason, "Stream ended before producing a non-ping SSE event"); assert.equal( result.upstreamDiagnostic, "UPSTREAM_DETAIL quota exhausted; retry after 2s; empty content Bearer [REDACTED] " @@ -636,16 +670,9 @@ test("ensureStreamReadiness preserves sanitized error-only diagnostics on early assert.equal(body.upstream_details.error.message, result.upstreamDiagnostic); assert.equal(warnings.length, 1); - for (const surfaced of [ - result.reason, - body.upstream_details.error.message, - warnings[0], - ]) { + for (const surfaced of [result.reason, body.upstream_details.error.message, warnings[0]]) { assert.match(surfaced, /UPSTREAM_DETAIL/); - assert.doesNotMatch( - surfaced, - /SECOND_DETAIL|TOP_SECRET|\/srv\/omniroute\/handler\.ts/ - ); + assert.doesNotMatch(surfaced, /SECOND_DETAIL|TOP_SECRET|\/srv\/omniroute\/handler\.ts/); } });