fix(sse): preserve 1min.ai partial output before stream errors (#12466)

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.

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.

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<string, string> = {
   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<Uint8Array>, model: string, id: string, created: number): ReadableStream<Uint8Array> {
+function translateSseStream(
+  upstreamBody: ReadableStream<Uint8Array>,
+  model: string,
+  id: string,
+  created: number
+): ReadableStream<Uint8Array> {
   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<Uint8Array>({
-    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),

open-sse/utils/streamReadiness.ts

@@ -422,56 +422,66 @@ function prependBufferedChunks(
   reader: ReadableStreamDefaultReader<Uint8Array>
 ): ReadableStream<Uint8Array> {
   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<Uint8Array>({
     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);
     },
   });
 }

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<string, string>)["API-KEY"], "unit-test-key");
+
+    return responseFactory();
+  };
+  return () => calls;
+}
+
+function createStreamingResponse(events: string[]): Response {
+  return new Response(
+    new ReadableStream<Uint8Array>({
+      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<Response> {
+  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> | 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<T>(read: () => Promise<T | null>, timeoutMs = 5_000): Promise<T | null> {
+  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<Uint8Array>,
+  marker: string
+): Promise<string> {
+  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<Uint8Array>): Promise<string> {
+  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 <path>/);
+  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<Uint8Array> | null = null;
+  let cancelCalls = 0;
+  const getCalls = installFetchFactory(
+    () =>
+      new Response(
+        new ReadableStream<Uint8Array>({
+          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<Uint8Array>({
+          start(controller) {
+            controller.enqueue(
+              encoder.encode('event: error\ndata: {"message":"capacity unavailable"}\n\n')
+            );
+          },
+          cancel() {
+            cancelCalls += 1;
+            return new Promise<void>(() => {});
+          },
+        }),
+        { 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<never>((_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<Uint8Array> | null = null;
+  let cancelCalls = 0;
+  let markPullStarted: (() => void) | null = null;
+  const pullStarted = new Promise<void>((resolve) => {
+    markPullStarted = resolve;
+  });
+  const getCalls = installFetchFactory(
+    () =>
+      new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            upstreamController = controller;
+            controller.enqueue(
+              encoder.encode('event: content\ndata: {"content":"partial answer"}\n\n')
+            );
+          },
+          pull() {
+            markPullStarted?.();
+            return new Promise<void>(() => {});
+          },
+          cancel() {
+            cancelCalls += 1;
+            return new Promise<void>(() => {});
+          },
+        }),
+        { 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<never>((_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"/);
+});

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

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<Uint8Array>({
       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<Uint8Array>({
+      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<Uint8Array>({
+      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<void>(() => {});
+      },
+      cancel() {
+        cancelCalls += 1;
+        return new Promise<void>(() => {});
+      },
+    }),
+    { 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<never>((_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 () => {