From d40218e23f7ece20d89faaa28485e0e438d2a65b Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Wed, 2 Sep 2026 06:46:53 -0300 Subject: [PATCH] fix(huggingchat): surface JSONL generation failures --- ...nding-huggingchat-stream-error-boundary.md | 1 + open-sse/executors/huggingchat.ts | 108 +++- open-sse/executors/huggingchat/jsonlStream.ts | 63 +- .../huggingchat-stream-error-boundary.test.ts | 551 ++++++++++++++++++ 4 files changed, 696 insertions(+), 27 deletions(-) create mode 100644 changelog.d/fixes/pending-huggingchat-stream-error-boundary.md create mode 100644 tests/unit/huggingchat-stream-error-boundary.test.ts diff --git a/changelog.d/fixes/pending-huggingchat-stream-error-boundary.md b/changelog.d/fixes/pending-huggingchat-stream-error-boundary.md new file mode 100644 index 0000000000..7285276d10 --- /dev/null +++ b/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. diff --git a/open-sse/executors/huggingchat.ts b/open-sse/executors/huggingchat.ts index 7b7557ce02..a37a965094 100644 --- a/open-sse/executors/huggingchat.ts +++ b/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({ + 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 { diff --git a/open-sse/executors/huggingchat/jsonlStream.ts b/open-sse/executors/huggingchat/jsonlStream.ts index b09bcb2c30..3d4980aebb 100644 --- a/open-sse/executors/huggingchat/jsonlStream.ts +++ b/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): void { + try { + void reader.cancel().catch(() => undefined); + } catch { + // The error event is authoritative; transport cleanup is best effort. + } +} + +function bindReaderCancellation( + reader: ReadableStreamDefaultReader, + 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 { 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(); diff --git a/tests/unit/huggingchat-stream-error-boundary.test.ts b/tests/unit/huggingchat-stream-error-boundary.test.ts new file mode 100644 index 0000000000..ed76a59ac6 --- /dev/null +++ b/tests/unit/huggingchat-stream-error-boundary.test.ts @@ -0,0 +1,551 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, test } from "node:test"; + +const originalDataDir = process.env.DATA_DIR; +const originalPluginsDir = process.env.OMNIROUTE_PLUGINS_DIR; +const testRoot = mkdtempSync(join(tmpdir(), "omniroute-huggingchat-error-boundary-")); +const testDataDir = join(testRoot, "data"); +const testPluginsDir = join(testRoot, "plugins"); + +mkdirSync(testDataDir, { recursive: true }); +mkdirSync(testPluginsDir, { recursive: true }); +process.env.DATA_DIR = testDataDir; +process.env.OMNIROUTE_PLUGINS_DIR = testPluginsDir; + +const [ + { HuggingChatExecutor }, + { HuggingChatStreamError, streamJsonlToOpenAi }, + { createPassthroughStreamWithLogger }, + { createStreamController, pipeWithDisconnect }, + { createStreamFailureFinalizers, finalizeStreamRequestLog }, + { ensureStreamReadiness }, + { FORMATS }, + usageHistory, + coreDb, +] = 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"), +]); + +after(() => { + usageHistory.clearPendingRequests(); + coreDb.resetDbInstance(); + + if (originalDataDir === undefined) delete process.env.DATA_DIR; + else process.env.DATA_DIR = originalDataDir; + + if (originalPluginsDir === undefined) delete process.env.OMNIROUTE_PLUGINS_DIR; + else process.env.OMNIROUTE_PLUGINS_DIR = originalPluginsDir; + + rmSync(testRoot, { recursive: true, force: true }); +}); + +function jsonlBody(lines: string[], trailingNewline = true): ReadableStream { + const encoded = new TextEncoder().encode(`${lines.join("\n")}${trailingNewline ? "\n" : ""}`); + return new ReadableStream({ + start(controller) { + controller.enqueue(encoded); + controller.close(); + }, + }); +} + +async function collectStream(body: ReadableStream): Promise { + 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"; + const provider = "huggingchat"; + const connectionId = "huggingchat-stream-error-boundary"; + 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(); + 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/); + } 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({ + start(controller) { + controller.enqueue(encoded); + }, + cancel() { + cancelCalled = true; + return new Promise(() => 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({ + pull(controller) { + upstreamPullCount += 1; + if (upstreamPullCount === 1) { + controller.enqueue(token); + return; + } + return new Promise(() => undefined); + }, + cancel() { + upstreamCancelCalled = true; + return new Promise(() => 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((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({ + pull(controller) { + upstreamPullCount += 1; + if (upstreamPullCount === 1) { + controller.enqueue(token); + return; + } + return new Promise(() => undefined); + }, + cancel() { + upstreamCancelCalled = true; + return new Promise(() => 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((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((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":\{/); +});