From 20a67d445669da6b6ca2f44aadfac815e08f9245 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 2 Jul 2026 22:18:12 -0300 Subject: [PATCH] refactor(executors): extract pure JSONL stream translation from huggingchat (#6016) Extract the pure JSONL->OpenAI-SSE translation (sseChunk, parseJsonlLine, streamJsonlToOpenAi, readJsonlResponse) verbatim into the leaf huggingchat/jsonlStream.ts. They consume a passed-in ReadableStream (no fetch/network/state). Host imports back the two it uses; all module-private (no re-export). Host 812 -> 594 LOC. Byte-identical bodies (verbatim), leaf has zero imports (no cycle). Cookie/auth/multipart/execute untouched. Adds a split-guard; consumer tests stay green (executor-huggingchat 6, huggingchat-model-catalog 3). --- open-sse/executors/huggingchat.ts | 221 +----------------- open-sse/executors/huggingchat/jsonlStream.ts | 221 ++++++++++++++++++ tests/unit/huggingchat-jsonl-split.test.ts | 32 +++ 3 files changed, 254 insertions(+), 220 deletions(-) create mode 100644 open-sse/executors/huggingchat/jsonlStream.ts create mode 100644 tests/unit/huggingchat-jsonl-split.test.ts diff --git a/open-sse/executors/huggingchat.ts b/open-sse/executors/huggingchat.ts index 46db72840e..7b7557ce02 100644 --- a/open-sse/executors/huggingchat.ts +++ b/open-sse/executors/huggingchat.ts @@ -27,6 +27,7 @@ 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"; const HUGGINGFACE_BASE = "https://huggingface.co"; const CONVERSATION_URL = `${HUGGINGFACE_BASE}/chat/conversation`; @@ -108,10 +109,6 @@ function buildConversationPrompt(messages: Array>): { }; } -function sseChunk(data: unknown): string { - return `data: ${JSON.stringify(data)}\n\n`; -} - function estimateTokens(text: string): number { return Math.max(1, Math.ceil((text || "").length / 4)); } @@ -245,222 +242,6 @@ function mergeCookieHeaderWithSetCookie(cookieHeader: string, setCookieHeaders: return [...cookieMap.entries()].map(([name, value]) => `${name}=${value}`).join("; "); } -function parseJsonlLine(line: string): { - token?: string; - done?: boolean; - error?: string; - text?: string; -} { - try { - const event = JSON.parse(line); - - if (event.type === "stream" && typeof event.token === "string") { - const token = event.token.replace(/\0/g, ""); - if (token) return { token }; - } - - if (event.type === "finalAnswer" && typeof event.text === "string") { - return { text: event.text, done: true }; - } - - if (event.type === "status") { - if (event.status === "error") { - return { error: event.message || "HuggingChat generation error" }; - } - if (event.status === "finished") { - return { done: true }; - } - } - } catch { - // Skip non-JSON lines - } - - return {}; -} - -async function* streamJsonlToOpenAi( - body: ReadableStream, - model: string, - id: string, - created: number, - signal?: AbortSignal | null -): AsyncGenerator { - const reader = body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - let emittedRole = false; - let fullText = ""; - let finished = false; - - try { - while (true) { - if (signal?.aborted) break; - - const { value, done } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - - const lines = buffer.split("\n"); - buffer = lines.pop() || ""; - - for (const line of lines) { - const trimmed = line.trim(); - if (!trimmed) continue; - - 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; - } - - if (parsed.token) { - if (!emittedRole) { - emittedRole = true; - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], - }); - } - - fullText += parsed.token; - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }], - }); - } - - if (parsed.text) { - const remaining = parsed.text.slice(fullText.length); - if (remaining) { - if (!emittedRole) { - emittedRole = true; - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], - }); - } - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { content: remaining }, finish_reason: null }], - }); - } - finished = true; - break; - } - - if (parsed.done) { - finished = true; - break; - } - } - - if (finished) break; - } - - if (!finished && buffer.trim()) { - const parsed = parseJsonlLine(buffer.trim()); - if (parsed.token && !signal?.aborted) { - if (!emittedRole) { - emittedRole = true; - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], - }); - } - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }], - }); - } - } - } finally { - reader.releaseLock(); - } - - if (!signal?.aborted) { - yield sseChunk({ - id, - object: "chat.completion.chunk", - created, - model, - choices: [{ index: 0, delta: {}, finish_reason: "stop" }], - }); - yield "data: [DONE]\n\n"; - } -} - -async function readJsonlResponse( - body: ReadableStream, - signal?: AbortSignal | null -): Promise { - const reader = body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - let fullText = ""; - - try { - while (true) { - if (signal?.aborted) break; - - const { value, done } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - - const lines = buffer.split("\n"); - buffer = lines.pop() || ""; - - for (const line of lines) { - const trimmed = line.trim(); - if (!trimmed) continue; - - 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 (buffer.trim()) { - const parsed = parseJsonlLine(buffer.trim()); - if (parsed.text) return parsed.text; - if (parsed.token) fullText += parsed.token; - } - } finally { - reader.releaseLock(); - } - - return fullText; -} - // -- Executor ---------------------------------------------------------------- export class HuggingChatExecutor extends BaseExecutor { diff --git a/open-sse/executors/huggingchat/jsonlStream.ts b/open-sse/executors/huggingchat/jsonlStream.ts new file mode 100644 index 0000000000..b09bcb2c30 --- /dev/null +++ b/open-sse/executors/huggingchat/jsonlStream.ts @@ -0,0 +1,221 @@ +// Pure JSONL stream translation (HuggingChat NDJSON -> OpenAI SSE). Verbatim from huggingchat.ts. + +export function sseChunk(data: unknown): string { + return `data: ${JSON.stringify(data)}\n\n`; +} + +export function parseJsonlLine(line: string): { + token?: string; + done?: boolean; + error?: string; + text?: string; +} { + try { + const event = JSON.parse(line); + + if (event.type === "stream" && typeof event.token === "string") { + const token = event.token.replace(/\0/g, ""); + if (token) return { token }; + } + + if (event.type === "finalAnswer" && typeof event.text === "string") { + return { text: event.text, done: true }; + } + + if (event.type === "status") { + if (event.status === "error") { + return { error: event.message || "HuggingChat generation error" }; + } + if (event.status === "finished") { + return { done: true }; + } + } + } catch { + // Skip non-JSON lines + } + + return {}; +} + +export async function* streamJsonlToOpenAi( + body: ReadableStream, + model: string, + id: string, + created: number, + signal?: AbortSignal | null +): AsyncGenerator { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let emittedRole = false; + let fullText = ""; + let finished = false; + + try { + while (true) { + if (signal?.aborted) break; + + const { value, done } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + + const lines = buffer.split("\n"); + buffer = lines.pop() || ""; + + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) continue; + + 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; + } + + if (parsed.token) { + if (!emittedRole) { + emittedRole = true; + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], + }); + } + + fullText += parsed.token; + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }], + }); + } + + if (parsed.text) { + const remaining = parsed.text.slice(fullText.length); + if (remaining) { + if (!emittedRole) { + emittedRole = true; + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], + }); + } + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { content: remaining }, finish_reason: null }], + }); + } + finished = true; + break; + } + + if (parsed.done) { + finished = true; + break; + } + } + + if (finished) break; + } + + if (!finished && buffer.trim()) { + const parsed = parseJsonlLine(buffer.trim()); + if (parsed.token && !signal?.aborted) { + if (!emittedRole) { + emittedRole = true; + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }], + }); + } + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: { content: parsed.token }, finish_reason: null }], + }); + } + } + } finally { + reader.releaseLock(); + } + + if (!signal?.aborted) { + yield sseChunk({ + id, + object: "chat.completion.chunk", + created, + model, + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + }); + yield "data: [DONE]\n\n"; + } +} + +export async function readJsonlResponse( + body: ReadableStream, + signal?: AbortSignal | null +): Promise { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let fullText = ""; + + try { + while (true) { + if (signal?.aborted) break; + + const { value, done } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + + const lines = buffer.split("\n"); + buffer = lines.pop() || ""; + + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) continue; + + 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 (buffer.trim()) { + const parsed = parseJsonlLine(buffer.trim()); + if (parsed.text) return parsed.text; + if (parsed.token) fullText += parsed.token; + } + } finally { + reader.releaseLock(); + } + + return fullText; +} diff --git a/tests/unit/huggingchat-jsonl-split.test.ts b/tests/unit/huggingchat-jsonl-split.test.ts new file mode 100644 index 0000000000..e6d952683a --- /dev/null +++ b/tests/unit/huggingchat-jsonl-split.test.ts @@ -0,0 +1,32 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, join } from "node:path"; + +// Split-guard for the huggingchat executor JSONL-stream extraction. +// The pure JSONL->OpenAI-SSE translation lives in huggingchat/jsonlStream.ts +// (consumes a passed-in ReadableStream; no host state/fetch/auth). Host imports it back. +const HERE = dirname(fileURLToPath(import.meta.url)); +const EXE = join(HERE, "../../open-sse/executors"); +const HOST = join(EXE, "huggingchat.ts"); +const LEAF = join(EXE, "huggingchat/jsonlStream.ts"); + +test("leaf hosts the jsonl translators and does not import the host", () => { + const src = readFileSync(LEAF, "utf8"); + assert.match(src, /export async function\* streamJsonlToOpenAi\b/); + assert.match(src, /export async function readJsonlResponse\b/); + assert.match(src, /export function (sseChunk|parseJsonlLine)\b/); + assert.doesNotMatch(src, /from "\.\.\/huggingchat\.ts"/); +}); + +test("host imports the jsonl translators back from the leaf", () => { + const host = readFileSync(HOST, "utf8"); + assert.match(host, /from "\.\/huggingchat\/jsonlStream\.ts"/); +}); + +test("parseJsonlLine parses a jsonl data line", async () => { + const { parseJsonlLine } = await import("../../open-sse/executors/huggingchat/jsonlStream.ts"); + assert.equal(typeof parseJsonlLine, "function"); + assert.doesNotThrow(() => parseJsonlLine("not json")); +});