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

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

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<Uint8Array> {
   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<Uint8Array>) => {
+    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<Uint8Array>) => {
+    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<Uint8Array>) => {
+    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<Uint8Array>({
+    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<Uint8Array>({
+    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(

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);

open-sse/utils/streamReadiness.ts

@@ -421,29 +421,57 @@ function prependBufferedChunks(
   chunks: Uint8Array[],
   reader: ReadableStreamDefaultReader<Uint8Array>
 ): ReadableStream<Uint8Array> {
+  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<Uint8Array>({
-    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);
     },
   });
 }

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<Record<string, unknown>>
+): ReadableStream<Uint8Array> {
+  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<Uint8Array>({
+    start(controller) {
+      controller.enqueue(encoder.encode(payload));
+      controller.close();
+    },
+  });
+}
+
+async function executeWithUpstreamBody(
+  body: ReadableStream<Uint8Array>,
+  prompt = "hi"
+): Promise<Response> {
+  __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<Record<string, unknown>>,
+  prompt = "hi"
+): Promise<Response> {
+  return executeWithUpstreamBody(createPerplexityStream(events), prompt);
+}
+
+async function executeThroughChatCore(
+  events: Array<Record<string, unknown>>,
+  prompt = "hi",
+  onStreamFailure?: (failure: StreamFailure) => void,
+  onRequestSuccess?: () => Promise<void>,
+  model = "pplx-auto"
+) {
+  return executeBodyThroughChatCore(
+    createPerplexityStream(events),
+    prompt,
+    onStreamFailure,
+    onRequestSuccess,
+    model
+  );
+}
+
+async function executeBodyThroughChatCore(
+  upstreamBody: ReadableStream<Uint8Array>,
+  prompt = "hi",
+  onStreamFailure?: (failure: StreamFailure) => void,
+  onRequestSuccess?: () => Promise<void>,
+  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<string, unknown>);
+  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<Record<string, unknown>>) {
+      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<boolean> {
+  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<Uint8Array>({
+      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<Uint8Array>({
+    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<string, unknown> };
+  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<string, unknown> };
+  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<Uint8Array>({
+    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<void>(() => {});
+    },
+    cancel() {
+      upstreamCancelCount += 1;
+      return new Promise<void>(() => {});
+    },
+  });
+
+  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<string, unknown> };
+  assert.equal(
+    sent.params?.last_backend_uuid,
+    undefined,
+    "a cancelled response must not create a reusable session"
+  );
+});

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

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<Uint8Array>({
+      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] <path>"
@@ -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/);
   }
 });