fix(huggingchat): surface HTTP 200 JSONL failures

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 { 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 @@
   "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,26 +528,81 @@
 
     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) {
+      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 {
-            for await (const chunk of jsonlStream) {
-              controller.enqueue(encoder.encode(chunk));
+            const chunk = await jsonlStream.next();
+            if (streamCancelled) return;
+            if (chunk.done) {
+              controller.close();
+              return;
             }
+            controller.enqueue(encoder.encode(chunk.value));
           } catch (err) {
-            log?.error?.("HUGGINGCHAT", `Stream error: ${err}`);
-          } finally {
-            controller.close();
+            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);
+        },
       });
 
       return {
@@ -560,7 +620,29 @@
       };
     }
 
-    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,18 +189,21 @@ export async function* streamJsonlToOpenAi(
       }
     }
   } finally {
+    unbindReaderCancellation();
     reader.releaseLock();
   }
 
-  if (!signal?.aborted) {
+  if (!signal?.aborted && !cancellationSignal?.aborted) {
     yield sseChunk({
       id,
       object: "chat.completion.chunk",
       created,
       model,
       choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
     });
-    yield "data: [DONE]\n\n";
+    if (!signal?.aborted && !cancellationSignal?.aborted) {
+      yield "data: [DONE]\n\n";
+    }
   }
 }
 
@@ -204,14 +235,18 @@ 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);
+        }
       }
     }
 
     if (buffer.trim()) {
       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();