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

changelog.d/fixes/pending-huggingchat-stream-error-boundary.md

@@ -0,0 +1 @@
+- HuggingChat now turns HTTP 200 JSONL generation failures into a sanitized 502 before content, or a fixed public stream failure after partial output, so fallback and request persistence no longer record a false successful stop.

open-sse/executors/huggingchat.ts

@@ -27,7 +27,11 @@ import {
 import { FETCH_TIMEOUT_MS } from "../config/constants.ts";
 import { buildErrorBody, sanitizeErrorMessage } from "../utils/error.ts";
 import { normalizeSessionCookieHeader } from "@/lib/providers/webCookieAuth";
-import { streamJsonlToOpenAi, readJsonlResponse } from "./huggingchat/jsonlStream.ts";
+import {
+  HuggingChatStreamError,
+  readJsonlResponse,
+  streamJsonlToOpenAi,
+} from "./huggingchat/jsonlStream.ts";
 
 const HUGGINGFACE_BASE = "https://huggingface.co";
 const CONVERSATION_URL = `${HUGGINGFACE_BASE}/chat/conversation`;
@@ -38,6 +42,7 @@ const USER_AGENT =
   "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36";
 
 const DEFAULT_MODEL = "baidu/ERNIE-4.5-VL-424B-A47B-Base-PT";
+const HUGGINGCHAT_PUBLIC_STREAM_ERROR = "HuggingChat generation failed";
 
 // -- Helpers -----------------------------------------------------------------
 
@@ -523,25 +528,80 @@ export class HuggingChatExecutor extends BaseExecutor {
 
     if (stream) {
       const encoder = new TextEncoder();
+      const streamCancellationController = new AbortController();
       const jsonlStream = streamJsonlToOpenAi(
         upstreamResponse.body,
         resolvedModel,
         id,
         created,
-        signal
+        signal,
+        streamCancellationController.signal
       );
 
-      const sseStream = new ReadableStream({
-        async start(controller) {
-          try {
-            for await (const chunk of jsonlStream) {
-              controller.enqueue(encoder.encode(chunk));
-            }
-          } catch (err) {
-            log?.error?.("HUGGINGCHAT", `Stream error: ${err}`);
-          } finally {
-            controller.close();
+      const primedChunks: string[] = [];
+      try {
+        const first = await jsonlStream.next();
+        if (!first.done) {
+          primedChunks.push(first.value);
+          if (first.value.includes('"role":"assistant"')) {
+            const content = await jsonlStream.next();
+            if (!content.done) primedChunks.push(content.value);
           }
+        }
+      } catch (err) {
+        if (!(err instanceof HuggingChatStreamError)) throw err;
+        const message = err instanceof Error ? err.message : String(err);
+        const safeMessage = sanitizeErrorMessage(message);
+        log?.error?.("HUGGINGCHAT", `Stream failed before content: ${safeMessage}`);
+        return {
+          response: new Response(
+            JSON.stringify(
+              buildErrorBody(502, message, undefined, {
+                type: "upstream_error",
+                code: "huggingchat_generation_error",
+              })
+            ),
+            { status: 502, headers: { "Content-Type": "application/json" } }
+          ),
+          url: messageUrl,
+          headers: baseHeaders,
+          transformedBody: sendDataPayload,
+        };
+      }
+
+      let primedChunkIndex = 0;
+      let streamCancelled = false;
+      const sseStream = new ReadableStream<Uint8Array>({
+        async pull(controller) {
+          if (streamCancelled) return;
+          if (primedChunkIndex < primedChunks.length) {
+            controller.enqueue(encoder.encode(primedChunks[primedChunkIndex]));
+            primedChunkIndex += 1;
+            return;
+          }
+
+          try {
+            const chunk = await jsonlStream.next();
+            if (streamCancelled) return;
+            if (chunk.done) {
+              controller.close();
+              return;
+            }
+            controller.enqueue(encoder.encode(chunk.value));
+          } catch (err) {
+            if (streamCancelled) return;
+            const message = err instanceof Error ? err.message : String(err);
+            const safeMessage = sanitizeErrorMessage(message);
+            log?.error?.("HUGGINGCHAT", `Stream error: ${safeMessage}`);
+            controller.error(
+              Object.assign(new Error(HUGGINGCHAT_PUBLIC_STREAM_ERROR), { statusCode: 502 })
+            );
+          }
+        },
+        cancel() {
+          streamCancelled = true;
+          streamCancellationController.abort();
+          void jsonlStream.return(undefined).catch(() => undefined);
         },
       });
 
@@ -560,7 +620,29 @@ export class HuggingChatExecutor extends BaseExecutor {
       };
     }
 
-    const fullText = await readJsonlResponse(upstreamResponse.body, signal);
+    let fullText: string;
+    try {
+      fullText = await readJsonlResponse(upstreamResponse.body, signal);
+    } catch (err) {
+      if (!(err instanceof HuggingChatStreamError)) throw err;
+      const message = err instanceof Error ? err.message : String(err);
+      const safeMessage = sanitizeErrorMessage(message);
+      log?.error?.("HUGGINGCHAT", `Generation error: ${safeMessage}`);
+      return {
+        response: new Response(
+          JSON.stringify(
+            buildErrorBody(502, message, undefined, {
+              type: "upstream_error",
+              code: "huggingchat_generation_error",
+            })
+          ),
+          { status: 502, headers: { "Content-Type": "application/json" } }
+        ),
+        url: messageUrl,
+        headers: baseHeaders,
+        transformedBody: sendDataPayload,
+      };
+    }
     const completionTokens = estimateTokens(fullText);
 
     return {

open-sse/executors/huggingchat/jsonlStream.ts

@@ -1,5 +1,36 @@
 // Pure JSONL stream translation (HuggingChat NDJSON -> OpenAI SSE). Verbatim from huggingchat.ts.
 
+export class HuggingChatStreamError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "HuggingChatStreamError";
+  }
+}
+
+function cancelReader(reader: ReadableStreamDefaultReader<Uint8Array>): void {
+  try {
+    void reader.cancel().catch(() => undefined);
+  } catch {
+    // The error event is authoritative; transport cleanup is best effort.
+  }
+}
+
+function bindReaderCancellation(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  signal?: AbortSignal | null
+): () => void {
+  if (!signal) return () => undefined;
+
+  const cancel = () => cancelReader(reader);
+  if (signal.aborted) {
+    cancel();
+    return () => undefined;
+  }
+
+  signal.addEventListener("abort", cancel, { once: true });
+  return () => signal.removeEventListener("abort", cancel);
+}
+
 export function sseChunk(data: unknown): string {
   return `data: ${JSON.stringify(data)}\n\n`;
 }
@@ -42,9 +73,11 @@ export async function* streamJsonlToOpenAi(
   model: string,
   id: string,
   created: number,
-  signal?: AbortSignal | null
+  signal?: AbortSignal | null,
+  cancellationSignal?: AbortSignal | null
 ): AsyncGenerator<string> {
   const reader = body.getReader();
+  const unbindReaderCancellation = bindReaderCancellation(reader, cancellationSignal);
   const decoder = new TextDecoder();
   let buffer = "";
   let emittedRole = false;
@@ -70,16 +103,8 @@ export async function* streamJsonlToOpenAi(
         const parsed = parseJsonlLine(trimmed);
 
         if (parsed.error) {
-          yield sseChunk({
-            id,
-            object: "chat.completion.chunk",
-            created,
-            model,
-            choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
-          });
-          yield "data: [DONE]\n\n";
-          finished = true;
-          return;
+          cancelReader(reader);
+          throw new HuggingChatStreamError(parsed.error);
         }
 
         if (parsed.token) {
@@ -140,6 +165,9 @@ export async function* streamJsonlToOpenAi(
 
     if (!finished && buffer.trim()) {
       const parsed = parseJsonlLine(buffer.trim());
+      if (parsed.error) {
+        throw new HuggingChatStreamError(parsed.error);
+      }
       if (parsed.token && !signal?.aborted) {
         if (!emittedRole) {
           emittedRole = true;
@@ -161,10 +189,11 @@ export async function* streamJsonlToOpenAi(
       }
     }
   } finally {
+    unbindReaderCancellation();
     reader.releaseLock();
   }
 
-  if (!signal?.aborted) {
+  if (!signal?.aborted && !cancellationSignal?.aborted) {
     yield sseChunk({
       id,
       object: "chat.completion.chunk",
@@ -172,7 +201,9 @@ export async function* streamJsonlToOpenAi(
       model,
       choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
     });
-    yield "data: [DONE]\n\n";
+    if (!signal?.aborted && !cancellationSignal?.aborted) {
+      yield "data: [DONE]\n\n";
+    }
   }
 }
 
@@ -204,7 +235,10 @@ export async function readJsonlResponse(
         const parsed = parseJsonlLine(trimmed);
         if (parsed.token) fullText += parsed.token;
         if (parsed.text) return parsed.text;
-        if (parsed.error) throw new Error(parsed.error);
+        if (parsed.error) {
+          cancelReader(reader);
+          throw new HuggingChatStreamError(parsed.error);
+        }
       }
     }
 
@@ -212,6 +246,7 @@ export async function readJsonlResponse(
       const parsed = parseJsonlLine(buffer.trim());
       if (parsed.text) return parsed.text;
       if (parsed.token) fullText += parsed.token;
+      if (parsed.error) throw new HuggingChatStreamError(parsed.error);
     }
   } finally {
     reader.releaseLock();

tests/unit/_fixtures/huggingchat-stream-error-boundary.fixture.ts

@@ -0,0 +1,592 @@
+import assert from "node:assert/strict";
+import { isAbsolute, relative } from "node:path";
+import { after, test } from "node:test";
+
+function requiredEnv(name: string): string {
+  const value = process.env[name];
+  assert.ok(value, `${name} must be supplied by the isolated parent wrapper`);
+  return value;
+}
+
+const testRoot = requiredEnv("OMNIROUTE_HUGGINGCHAT_TEST_ROOT");
+const fixtureRunId = requiredEnv("OMNIROUTE_HUGGINGCHAT_TEST_RUN_ID");
+const testDataDir = requiredEnv("DATA_DIR");
+const testPluginsDir = requiredEnv("OMNIROUTE_PLUGINS_DIR");
+const xdgConfigDir = requiredEnv("XDG_CONFIG_HOME");
+
+for (const [name, candidate] of [
+  ["DATA_DIR", testDataDir],
+  ["OMNIROUTE_PLUGINS_DIR", testPluginsDir],
+  ["XDG_CONFIG_HOME", xdgConfigDir],
+] as const) {
+  const fromRoot = relative(testRoot, candidate);
+  assert.equal(
+    isAbsolute(fromRoot) || fromRoot.startsWith(".."),
+    false,
+    `${name} escaped test root`
+  );
+}
+assert.equal(
+  process.env.NODE_TEST_CONTEXT,
+  undefined,
+  "nested node:test state must not be inherited"
+);
+assert.equal(process.env.HOME, undefined, "the child must not inherit the operator HOME");
+assert.equal(process.env.CODEX_HOME, undefined, "the child must not inherit CODEX_HOME");
+assert.match(requiredEnv("API_KEY_SECRET"), /^[0-9a-f]{64}$/);
+
+const [
+  { HuggingChatExecutor },
+  { HuggingChatStreamError, streamJsonlToOpenAi },
+  { createPassthroughStreamWithLogger },
+  { createStreamController, pipeWithDisconnect },
+  { createStreamFailureFinalizers, finalizeStreamRequestLog },
+  { ensureStreamReadiness },
+  { FORMATS },
+  usageHistory,
+  coreDb,
+  callLogs,
+  callLogArtifactWriter,
+  loggerResource,
+] = await Promise.all([
+  import("../../../open-sse/executors/huggingchat.ts"),
+  import("../../../open-sse/executors/huggingchat/jsonlStream.ts"),
+  import("../../../open-sse/utils/stream.ts"),
+  import("../../../open-sse/utils/streamHandler.ts"),
+  import("../../../open-sse/utils/streamFailureFinalization.ts"),
+  import("../../../open-sse/utils/streamReadiness.ts"),
+  import("../../../open-sse/translator/formats.ts"),
+  import("../../../src/lib/usage/usageHistory.ts"),
+  import("../../../src/lib/db/core.ts"),
+  import("../../../src/lib/usage/callLogs.ts"),
+  import("../../../src/lib/usage/callLogArtifactWriter.ts"),
+  import("../../../src/shared/utils/loggerResource.ts"),
+]);
+
+after(async () => {
+  assert.equal(
+    await callLogs.waitForCallLogSaves(10_000),
+    true,
+    "all asynchronous call-log writes must drain before DB teardown"
+  );
+  await callLogArtifactWriter.closeCallLogArtifactWriter();
+  usageHistory.clearPendingRequests();
+  await loggerResource.closeSharedLoggerResource();
+  coreDb.resetDbInstance();
+});
+
+function jsonlBody(lines: string[], trailingNewline = true): ReadableStream<Uint8Array> {
+  const encoded = new TextEncoder().encode(`${lines.join("\n")}${trailingNewline ? "\n" : ""}`);
+  return new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(encoded);
+      controller.close();
+    },
+  });
+}
+
+async function collectStream(body: ReadableStream<Uint8Array>): Promise<string> {
+  const chunks: string[] = [];
+  for await (const chunk of streamJsonlToOpenAi(
+    body,
+    "test/huggingchat-model",
+    "chatcmpl-huggingchat-test",
+    1_725_000_000
+  )) {
+    chunks.push(chunk);
+  }
+  return chunks.join("");
+}
+
+test("HuggingChat turns a pre-content JSONL generation error into a sanitized 502", async () => {
+  const rawError =
+    "generation failed at /srv/omniroute/providers/huggingchat.ts:44:9 api_key=super-secret\n" +
+    "    at provider (/srv/omniroute/runtime.ts:1:1)";
+  const realFetch = globalThis.fetch;
+  let callCount = 0;
+  const errorLogs: string[] = [];
+
+  globalThis.fetch = (async () => {
+    callCount += 1;
+    if (callCount === 1) {
+      return Response.json({ conversationId: "conversation-test" });
+    }
+    if (callCount === 2) {
+      return Response.json({ rootMessageId: "root-message-test" });
+    }
+    if (callCount === 3) {
+      return new Response(
+        jsonlBody(
+          [
+            JSON.stringify({ type: "status", status: "started" }),
+            JSON.stringify({ type: "status", status: "error", message: rawError }),
+          ],
+          false
+        ),
+        { status: 200, headers: { "Content-Type": "application/jsonl" } }
+      );
+    }
+    throw new Error(`Unexpected fetch call ${callCount}`);
+  }) as typeof globalThis.fetch;
+
+  try {
+    const result = await new HuggingChatExecutor().execute({
+      model: "test/huggingchat-model",
+      body: { messages: [{ role: "user", content: "hello" }] },
+      stream: true,
+      credentials: { apiKey: "hf-chat=fake-cookie" },
+      signal: null,
+      log: { error: (_tag, message) => errorLogs.push(message) },
+    });
+
+    assert.equal(callCount, 3, "the test must intercept every HuggingChat request");
+    assert.equal(result.response.status, 502);
+    assert.match(result.response.headers.get("content-type") || "", /application\/json/);
+
+    const payload = (await result.response.json()) as {
+      error: { message: string; type?: string; code?: string };
+    };
+    assert.equal(payload.error.type, "upstream_error");
+    assert.equal(payload.error.code, "huggingchat_generation_error");
+    assert.match(payload.error.message, /generation failed/);
+    assert.doesNotMatch(payload.error.message, /\/srv\/omniroute/);
+    assert.doesNotMatch(payload.error.message, /super-secret/);
+    assert.doesNotMatch(payload.error.message, /\n\s*at /);
+    assert.equal(errorLogs.length, 1);
+    assert.doesNotMatch(errorLogs[0], /\/srv\/omniroute/);
+    assert.doesNotMatch(errorLogs[0], /super-secret/);
+    assert.doesNotMatch(errorLogs[0], /\n\s*at /);
+  } finally {
+    globalThis.fetch = realFetch;
+  }
+});
+
+test("HuggingChat turns a terminal non-stream JSONL error into a sanitized 502", async () => {
+  const rawError =
+    "generation failed at /srv/omniroute/providers/huggingchat.ts:55:2 cookie=super-secret\n" +
+    "    at provider (/srv/omniroute/runtime.ts:1:1)";
+  const realFetch = globalThis.fetch;
+  let callCount = 0;
+
+  globalThis.fetch = (async () => {
+    callCount += 1;
+    if (callCount === 1) return Response.json({ conversationId: "conversation-test" });
+    if (callCount === 2) return Response.json({ rootMessageId: "root-message-test" });
+    if (callCount === 3) {
+      return new Response(
+        jsonlBody([JSON.stringify({ type: "status", status: "error", message: rawError })], false),
+        { status: 200, headers: { "Content-Type": "application/jsonl" } }
+      );
+    }
+    throw new Error(`Unexpected fetch call ${callCount}`);
+  }) as typeof globalThis.fetch;
+
+  try {
+    const result = await new HuggingChatExecutor().execute({
+      model: "test/huggingchat-model",
+      body: { messages: [{ role: "user", content: "hello" }] },
+      stream: false,
+      credentials: { apiKey: "hf-chat=fake-cookie" },
+      signal: null,
+    });
+
+    assert.equal(callCount, 3, "the test must intercept every HuggingChat request");
+    assert.equal(result.response.status, 502);
+    const payload = (await result.response.json()) as {
+      error: { message: string; type?: string; code?: string };
+    };
+    assert.equal(payload.error.type, "upstream_error");
+    assert.equal(payload.error.code, "huggingchat_generation_error");
+    assert.doesNotMatch(payload.error.message, /\/srv\/omniroute/);
+    assert.doesNotMatch(payload.error.message, /super-secret/);
+    assert.doesNotMatch(payload.error.message, /\n\s*at /);
+  } finally {
+    globalThis.fetch = realFetch;
+  }
+});
+
+test("HuggingChat rejects its JSONL generator after partial content instead of faking success", async () => {
+  const rawError =
+    "generation failed at /srv/omniroute/providers/huggingchat.ts:44:9 access_token=super-secret\n" +
+    "    at provider (/srv/omniroute/runtime.ts:1:1)";
+  const stream = streamJsonlToOpenAi(
+    jsonlBody([
+      JSON.stringify({ type: "stream", token: "partial answer" }),
+      JSON.stringify({ type: "status", status: "error", message: rawError }),
+    ]),
+    "test/huggingchat-model",
+    "chatcmpl-huggingchat-test",
+    1_725_000_000
+  );
+
+  const roleChunk = await stream.next();
+  const contentChunk = await stream.next();
+
+  assert.equal(roleChunk.done, false);
+  assert.match(roleChunk.value || "", /"role":"assistant"/);
+  assert.equal(contentChunk.done, false);
+  assert.match(contentChunk.value || "", /partial answer/);
+  await assert.rejects(() => stream.next(), HuggingChatStreamError);
+});
+
+test("HuggingChat partial failures reach stream finalization, persistence, and fallback", async () => {
+  const model = `test/huggingchat-model-${fixtureRunId}`;
+  const provider = "huggingchat";
+  const connectionId = `huggingchat-stream-error-boundary-${fixtureRunId}`;
+  const publicErrorMessage = "HuggingChat generation failed";
+  const rawError =
+    "generation failed at /srv/omniroute/providers/huggingchat.ts:44:9 access_token=super-secret\n" +
+    "    at provider (/srv/omniroute/runtime.ts:1:1)";
+  const realFetch = globalThis.fetch;
+  let callCount = 0;
+  const errorLogs: string[] = [];
+
+  globalThis.fetch = (async () => {
+    callCount += 1;
+    if (callCount === 1) return Response.json({ conversationId: "conversation-test" });
+    if (callCount === 2) return Response.json({ rootMessageId: "root-message-test" });
+    if (callCount === 3) {
+      return new Response(
+        jsonlBody([
+          JSON.stringify({ type: "stream", token: "partial answer" }),
+          JSON.stringify({ type: "status", status: "error", message: rawError }),
+        ]),
+        { status: 200, headers: { "Content-Type": "application/jsonl" } }
+      );
+    }
+    throw new Error(`Unexpected fetch call ${callCount}`);
+  }) as typeof globalThis.fetch;
+
+  usageHistory.clearPendingRequests();
+  assert.equal(usageHistory.getPendingById().size, 0, "the child must start without pending state");
+  assert.equal(
+    usageHistory.getCompletedDetails().size,
+    0,
+    "the child must start without completed state"
+  );
+  const previousPersistence = coreDb
+    .getDbInstance()
+    .prepare("SELECT COUNT(*) AS count FROM call_logs WHERE connection_id = ? AND model = ?")
+    .get(connectionId, model) as { count: number };
+  assert.equal(previousPersistence.count, 0, "the child must not reuse a prior persisted identity");
+  const requestId = usageHistory.trackPendingRequest(model, provider, connectionId, true);
+  assert.ok(requestId, "the full-pipeline test must own a real pending request");
+
+  type CompletionPayload = {
+    status: number;
+    usage: unknown;
+    providerPayload?: unknown;
+    clientPayload?: unknown;
+    error?: string | null;
+    errorCode?: string | null;
+  };
+  type FailurePayload = {
+    status: number;
+    message: string;
+    code?: string;
+    type?: string;
+  };
+
+  let completionPayload: CompletionPayload | null = null;
+  let streamCompletionRecorded = false;
+  let streamFailureCompletionRecorded = false;
+  const persistedFailures: Array<{ status: number; errorCode?: string }> = [];
+  const fallbackFailures: FailurePayload[] = [];
+
+  const onStreamComplete = (payload: CompletionPayload) => {
+    const normalizedStatus = payload.status || 200;
+    if (streamCompletionRecorded) return;
+    streamCompletionRecorded = true;
+    if (normalizedStatus !== 200) {
+      if (streamFailureCompletionRecorded) return;
+      streamFailureCompletionRecorded = true;
+    }
+    completionPayload = payload;
+    finalizeStreamRequestLog({
+      pendingRequestId: requestId,
+      model,
+      provider,
+      connectionId,
+      providerResponse: payload.providerPayload,
+      clientResponse: payload.clientPayload,
+      status: normalizedStatus,
+      error: payload.error,
+      errorCode: payload.errorCode,
+    });
+  };
+
+  const { handleStreamFailure, onPipelineStreamError } = createStreamFailureFinalizers({
+    isFailureCompletionRecorded: () => streamFailureCompletionRecorded,
+    isStreamCompletionRecorded: () => streamCompletionRecorded,
+    onStreamComplete,
+    persistFailureUsage: (status, errorCode) => {
+      persistedFailures.push({ status, errorCode });
+    },
+    onStreamFailure: (failure) => {
+      fallbackFailures.push(failure);
+    },
+  });
+
+  try {
+    const result = await new HuggingChatExecutor().execute({
+      model,
+      body: { messages: [{ role: "user", content: "hello" }] },
+      stream: true,
+      credentials: { apiKey: "hf-chat=fake-cookie" },
+      signal: null,
+      log: { error: (_tag, message) => errorLogs.push(message) },
+    });
+
+    assert.equal(callCount, 3, "the test must intercept every HuggingChat request");
+    assert.equal(result.response.status, 200, "partial output has already committed HTTP 200");
+    const readiness = await ensureStreamReadiness(result.response, {
+      timeoutMs: 1_000,
+      provider,
+      model,
+    });
+    if (!readiness.ok) assert.fail(`unexpected readiness failure: ${readiness.reason}`);
+
+    const transform = createPassthroughStreamWithLogger(
+      provider,
+      null,
+      null,
+      model,
+      connectionId,
+      { messages: [{ role: "user", content: "hello" }] },
+      onStreamComplete,
+      null,
+      handleStreamFailure,
+      FORMATS.OPENAI
+    );
+    const streamController = createStreamController({
+      onError: onPipelineStreamError,
+      provider,
+      model,
+      connectionId,
+      clientResponseFormat: FORMATS.OPENAI,
+    });
+    const clientStream = pipeWithDisconnect(readiness.response, transform, streamController, {
+      stallTimeoutMs: 0,
+    });
+    const wire = await new Response(clientStream).text();
+
+    assert.match(wire, /partial answer/);
+    assert.match(wire, /"finish_reason":"error"/);
+    assert.match(wire, new RegExp(publicErrorMessage));
+    assert.match(wire, /data: \[DONE\]/);
+    assert.doesNotMatch(wire, /"finish_reason":"stop"/);
+    assert.doesNotMatch(wire, /\/srv\/omniroute/);
+    assert.doesNotMatch(wire, /super-secret/);
+    assert.equal(errorLogs.length, 1);
+    assert.doesNotMatch(errorLogs[0], /\/srv\/omniroute/);
+    assert.doesNotMatch(errorLogs[0], /super-secret/);
+    assert.doesNotMatch(errorLogs[0], /\n\s*at /);
+
+    assert.ok(completionPayload, "the pipeline must record a terminal failure");
+    assert.equal(completionPayload.status, 502);
+    assert.equal(completionPayload.error, publicErrorMessage);
+    assert.equal(completionPayload.errorCode, "stream_pipeline_error");
+    assert.deepEqual(persistedFailures, [{ status: 502, errorCode: "stream_pipeline_error" }]);
+    assert.deepEqual(fallbackFailures, [
+      {
+        status: 502,
+        message: publicErrorMessage,
+        code: "stream_pipeline_error",
+        type: "stream_error",
+      },
+    ]);
+
+    assert.equal(usageHistory.getPendingById().has(requestId), false);
+    const completedDetail = usageHistory.getCompletedDetails().get(requestId);
+    assert.ok(completedDetail, "failure finalization must persist the completed request detail");
+    assert.equal(completedDetail.status, 502);
+    assert.equal(completedDetail.error, publicErrorMessage);
+    assert.equal(completedDetail.errorCode, "stream_pipeline_error");
+    assert.doesNotMatch(JSON.stringify(completedDetail), /\/srv\/omniroute|super-secret/);
+    assert.deepEqual(
+      [...usageHistory.getCompletedDetails().keys()],
+      [requestId],
+      "only this child run may own completed usage state"
+    );
+  } finally {
+    globalThis.fetch = realFetch;
+    usageHistory.clearPendingRequests();
+  }
+});
+
+test("HuggingChat reports an authoritative error without waiting for transport cancellation", async () => {
+  let cancelCalled = false;
+  const encoded = new TextEncoder().encode(
+    `${JSON.stringify({ type: "status", status: "error", message: "provider failed" })}\n`
+  );
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(encoded);
+    },
+    cancel() {
+      cancelCalled = true;
+      return new Promise<void>(() => undefined);
+    },
+  });
+  const stream = streamJsonlToOpenAi(
+    body,
+    "test/huggingchat-model",
+    "chatcmpl-huggingchat-test",
+    1_725_000_000
+  );
+
+  const outcome = await Promise.race([
+    stream.next().then(
+      () => ({ kind: "resolved" as const }),
+      (error: unknown) => ({ kind: "rejected" as const, error })
+    ),
+    new Promise<{ kind: "hung" }>((resolve) => {
+      setImmediate(() => resolve({ kind: "hung" }));
+    }),
+  ]);
+
+  assert.equal(cancelCalled, true);
+  assert.equal(outcome.kind, "rejected", "transport cleanup must not delay error delivery");
+  assert.ok(
+    outcome.kind === "rejected" && outcome.error instanceof HuggingChatStreamError,
+    "the authoritative HuggingChat error must remain classifiable"
+  );
+});
+
+test("HuggingChat cancellation suppresses final chunks after a pending JSONL read", async () => {
+  let upstreamCancelCalled = false;
+  let upstreamPullCount = 0;
+  const cancellationController = new AbortController();
+  const token = new TextEncoder().encode(
+    `${JSON.stringify({ type: "stream", token: "partial answer" })}\n`
+  );
+  const body = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      upstreamPullCount += 1;
+      if (upstreamPullCount === 1) {
+        controller.enqueue(token);
+        return;
+      }
+      return new Promise<void>(() => undefined);
+    },
+    cancel() {
+      upstreamCancelCalled = true;
+      return new Promise<void>(() => undefined);
+    },
+  });
+  const stream = streamJsonlToOpenAi(
+    body,
+    "test/huggingchat-model",
+    "chatcmpl-huggingchat-test",
+    1_725_000_000,
+    null,
+    cancellationController.signal
+  );
+
+  assert.match((await stream.next()).value || "", /"role":"assistant"/);
+  assert.match((await stream.next()).value || "", /partial answer/);
+  const pendingNext = stream.next();
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  cancellationController.abort();
+
+  const outcome = await Promise.race([
+    pendingNext.then((result) => ({ kind: "settled" as const, result })),
+    new Promise<{ kind: "hung" }>((resolve) => setImmediate(() => resolve({ kind: "hung" }))),
+  ]);
+
+  assert.equal(upstreamCancelCalled, true);
+  assert.equal(outcome.kind, "settled", "cancellation must settle the pending generator read");
+  assert.equal(
+    outcome.kind === "settled" ? outcome.result.done : false,
+    true,
+    "a cancelled generator must not emit stop or [DONE]"
+  );
+  void stream.return(undefined).catch(() => undefined);
+});
+
+test("HuggingChat client cancellation reaches a blocked upstream reader without waiting", async () => {
+  const realFetch = globalThis.fetch;
+  let callCount = 0;
+  let upstreamCancelCalled = false;
+  let upstreamPullCount = 0;
+  const errorLogs: string[] = [];
+  const token = new TextEncoder().encode(
+    `${JSON.stringify({ type: "stream", token: "partial answer" })}\n`
+  );
+  const blockedBody = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      upstreamPullCount += 1;
+      if (upstreamPullCount === 1) {
+        controller.enqueue(token);
+        return;
+      }
+      return new Promise<void>(() => undefined);
+    },
+    cancel() {
+      upstreamCancelCalled = true;
+      return new Promise<void>(() => undefined);
+    },
+  });
+
+  globalThis.fetch = (async () => {
+    callCount += 1;
+    if (callCount === 1) return Response.json({ conversationId: "conversation-test" });
+    if (callCount === 2) return Response.json({ rootMessageId: "root-message-test" });
+    if (callCount === 3) {
+      return new Response(blockedBody, {
+        status: 200,
+        headers: { "Content-Type": "application/jsonl" },
+      });
+    }
+    throw new Error(`Unexpected fetch call ${callCount}`);
+  }) as typeof globalThis.fetch;
+
+  try {
+    const result = await new HuggingChatExecutor().execute({
+      model: "test/huggingchat-model",
+      body: { messages: [{ role: "user", content: "hello" }] },
+      stream: true,
+      credentials: { apiKey: "hf-chat=fake-cookie" },
+      signal: null,
+      log: { error: (_tag, message) => errorLogs.push(message) },
+    });
+
+    assert.equal(callCount, 3, "the test must intercept every HuggingChat request");
+    assert.ok(result.response.body);
+    const reader = result.response.body.getReader();
+    const roleChunk = await reader.read();
+    const contentChunk = await reader.read();
+    assert.match(new TextDecoder().decode(roleChunk.value), /"role":"assistant"/);
+    assert.match(new TextDecoder().decode(contentChunk.value), /partial answer/);
+
+    const blockedRead = reader.read();
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    const cancelOutcome = await Promise.race([
+      reader.cancel("client disconnected").then(() => "resolved" as const),
+      new Promise<"hung">((resolve) => setImmediate(() => resolve("hung"))),
+    ]);
+    void blockedRead.catch(() => undefined);
+
+    assert.equal(cancelOutcome, "resolved", "downstream cancellation must remain non-blocking");
+    assert.equal(upstreamCancelCalled, true, "cancellation must reach the locked upstream reader");
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    assert.deepEqual(errorLogs, [], "client cancellation must not log a provider stream failure");
+  } finally {
+    globalThis.fetch = realFetch;
+  }
+});
+
+test("HuggingChat keeps the normal JSONL completion contract unchanged", async () => {
+  const output = await collectStream(
+    jsonlBody([
+      JSON.stringify({ type: "stream", token: "complete answer" }),
+      JSON.stringify({ type: "status", status: "finished" }),
+    ])
+  );
+
+  assert.match(output, /"role":"assistant"/);
+  assert.match(output, /complete answer/);
+  assert.match(output, /"finish_reason":"stop"/);
+  assert.match(output, /data: \[DONE\]/);
+  assert.doesNotMatch(output, /"error":\{/);
+});

tests/unit/huggingchat-stream-error-boundary.test.ts

@@ -0,0 +1,85 @@
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { mkdirSync, mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { basename, join } from "node:path";
+import { fileURLToPath } from "node:url";
+import test from "node:test";
+
+const REPO_ROOT = fileURLToPath(new URL("../..", import.meta.url));
+const FIXTURE = fileURLToPath(
+  new URL("./_fixtures/huggingchat-stream-error-boundary.fixture.ts", import.meta.url)
+);
+const SYNTHETIC_API_KEY_SECRET = "0".repeat(64);
+
+type ChildResult = {
+  code: number | null;
+  signal: NodeJS.Signals | null;
+  stdout: string;
+  stderr: string;
+};
+
+function runFixture(testRoot: string): Promise<ChildResult> {
+  const dataDir = join(testRoot, "data");
+  const pluginsDir = join(testRoot, "plugins");
+  // Keep config fallbacks inside the fixture root without inheriting or repurposing HOME.
+  const xdgConfigDir = join(testRoot, "xdg-config");
+
+  for (const dir of [dataDir, pluginsDir, xdgConfigDir]) {
+    mkdirSync(dir, { recursive: true });
+  }
+
+  const childEnv: NodeJS.ProcessEnv = {
+    API_KEY_SECRET: SYNTHETIC_API_KEY_SECRET,
+    APP_LOG_TO_FILE: "false",
+    DATA_DIR: dataDir,
+    DISABLE_SQLITE_AUTO_BACKUP: "true",
+    FORCE_COLOR: "0",
+    LANG: "C.UTF-8",
+    NODE_ENV: "test",
+    OMNIROUTE_HUGGINGCHAT_TEST_ROOT: testRoot,
+    OMNIROUTE_HUGGINGCHAT_TEST_RUN_ID: basename(testRoot),
+    OMNIROUTE_PLUGINS_DIR: pluginsDir,
+    TZ: "UTC",
+    XDG_CONFIG_HOME: xdgConfigDir,
+  };
+  if (process.env.PATH) childEnv.PATH = process.env.PATH;
+
+  return new Promise((resolve, reject) => {
+    const child = spawn(process.execPath, ["--import", "tsx/esm", FIXTURE], {
+      cwd: REPO_ROOT,
+      env: childEnv,
+      stdio: ["ignore", "pipe", "pipe"],
+    });
+    let stdout = "";
+    let stderr = "";
+    child.stdout.setEncoding("utf8").on("data", (chunk) => (stdout += chunk));
+    child.stderr.setEncoding("utf8").on("data", (chunk) => (stderr += chunk));
+    child.once("error", reject);
+    child.once("close", (code, signal) => resolve({ code, signal, stdout, stderr }));
+  });
+}
+
+function childDiagnostics(result: ChildResult): string {
+  return [
+    `exit=${String(result.code)} signal=${String(result.signal)}`,
+    "--- stdout ---",
+    result.stdout,
+    "--- stderr ---",
+    result.stderr,
+  ].join("\n");
+}
+
+test("HuggingChat stream error boundaries stay isolated from shared DB and usage state", async () => {
+  const testRoot = mkdtempSync(join(tmpdir(), "omniroute-huggingchat-boundary-child-"));
+  try {
+    const result = await runFixture(testRoot);
+    assert.equal(result.signal, null, childDiagnostics(result));
+    assert.equal(result.code, 0, childDiagnostics(result));
+    assert.match(result.stdout, /(?:#|ℹ) pass 8\b/, childDiagnostics(result));
+    assert.match(result.stdout, /(?:#|ℹ) fail 0\b/, childDiagnostics(result));
+    assert.doesNotMatch(result.stdout + result.stderr, /super-secret/);
+  } finally {
+    rmSync(testRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+  }
+});