From 619c2eaaeb358cf447a97838eaedc774976046f1 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Tue, 23 Jun 2026 07:11:37 -0300 Subject: [PATCH] =?UTF-8?q?refactor(chatCore):=20extrai=2011=20helpers=20d?= =?UTF-8?q?e=20n=C3=ADvel=20superior=20para=206=20leaves=20puros=20(#3501)?= =?UTF-8?q?=20(#4571)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chatCore god-file decomposition (#3501): extract 6 pure leaves (cacheUsageMeta, executorClientHeaders, nonStreamingResponseBody, skillsFormat, streamErrorResult, streamFinalize) from chatCore.ts. Rebased onto release/v3.8.35 tip (resolved single chatCore.ts conflict — removed now-extracted inline buildExecutorClientHeaders). 265/265 chatcore tests, 26/26 new leaf tests, typecheck:core, cycles, file-size all green. Integrated into release/v3.8.35. --- open-sse/handlers/chatCore.ts | 253 +----------------- open-sse/handlers/chatCore/cacheUsageMeta.ts | 65 +++++ .../chatCore/executorClientHeaders.ts | 36 +++ .../chatCore/nonStreamingResponseBody.ts | 68 +++++ open-sse/handlers/chatCore/skillsFormat.ts | 33 +++ .../handlers/chatCore/streamErrorResult.ts | 58 ++++ open-sse/handlers/chatCore/streamFinalize.ts | 48 ++++ tests/unit/chatcore-cache-usage-meta.test.ts | 58 ++++ .../chatcore-executor-client-headers.test.ts | 42 +++ ...atcore-non-streaming-response-body.test.ts | 33 +++ tests/unit/chatcore-skills-format.test.ts | 32 +++ .../unit/chatcore-stream-error-result.test.ts | 52 ++++ tests/unit/chatcore-stream-finalize.test.ts | 47 ++++ 13 files changed, 582 insertions(+), 243 deletions(-) create mode 100644 open-sse/handlers/chatCore/cacheUsageMeta.ts create mode 100644 open-sse/handlers/chatCore/executorClientHeaders.ts create mode 100644 open-sse/handlers/chatCore/nonStreamingResponseBody.ts create mode 100644 open-sse/handlers/chatCore/skillsFormat.ts create mode 100644 open-sse/handlers/chatCore/streamErrorResult.ts create mode 100644 open-sse/handlers/chatCore/streamFinalize.ts create mode 100644 tests/unit/chatcore-cache-usage-meta.test.ts create mode 100644 tests/unit/chatcore-executor-client-headers.test.ts create mode 100644 tests/unit/chatcore-non-streaming-response-body.test.ts create mode 100644 tests/unit/chatcore-skills-format.test.ts create mode 100644 tests/unit/chatcore-stream-error-result.test.ts create mode 100644 tests/unit/chatcore-stream-finalize.test.ts diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 6268c77d6f..0a3bc7b399 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -126,6 +126,16 @@ import { updateProviderConnection, getProviderConnectionById } from "@/lib/db/pr import { wasRefreshTokenRotated } from "@omniroute/open-sse/services/refreshSerializer.ts"; import { connectionHasExtraKeys } from "../services/apiKeyRotator.ts"; import { recordKeyHealthStatus as recordKeyHealthStatusFor } from "./chatCore/keyHealth.ts"; +import { getSkillsModelIdForFormat } from "./chatCore/skillsFormat.ts"; +import { readNonStreamingResponseBody } from "./chatCore/nonStreamingResponseBody.ts"; +import { + isSemaphoreCapacityError, + createStreamingErrorResult, + getUpstreamErrorIdentifier, +} from "./chatCore/streamErrorResult.ts"; +import { wrapReadableStreamWithFinalize } from "./chatCore/streamFinalize.ts"; +import { buildCacheUsageLogMeta, attachLogMeta } from "./chatCore/cacheUsageMeta.ts"; +import { buildExecutorClientHeaders } from "./chatCore/executorClientHeaders.ts"; import { getCallLogPipelineCaptureStreamChunks, @@ -278,222 +288,6 @@ import { incrementRequestCount } from "../services/geminiRateLimitTracker.ts"; import { isSmallEnoughForSemanticCache } from "../utils/estimateSize.ts"; -function getSkillsProviderForFormat(format: string): "openai" | "anthropic" | "google" | "other" { - switch (format) { - case FORMATS.CLAUDE: - return "anthropic"; - case FORMATS.GEMINI: - return "google"; - default: - return "openai"; - } -} - -function getSkillsModelIdForFormat(format: string): string { - switch (format) { - case FORMATS.CLAUDE: - return "claude"; - case FORMATS.GEMINI: - return "gemini"; - default: - return "openai"; - } -} - -async function readNonStreamingResponseBody( - response: Response, - contentType: string, - upstreamStream: boolean -): Promise { - if ( - !upstreamStream || - !response.body || - (!contentType.includes("text/event-stream") && !contentType.includes("application/x-ndjson")) - ) { - return withBodyTimeout(response.text()); - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - const terminalState: NonStreamingSseTerminalState = { - currentEvent: "", - pendingLine: "", - }; - let rawBody = ""; - const deadline = FETCH_BODY_TIMEOUT_MS > 0 ? Date.now() + FETCH_BODY_TIMEOUT_MS : 0; - - try { - while (true) { - const timeoutMs = deadline > 0 ? deadline - Date.now() : 0; - if (deadline > 0 && timeoutMs <= 0) { - throw createBodyTimeoutError(FETCH_BODY_TIMEOUT_MS); - } - - const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs); - if (done) break; - if (!value) continue; - - const decodedChunk = decoder.decode(value, { stream: true }); - rawBody += decodedChunk; - if (appendNonStreamingSseTerminalSignal(terminalState, decodedChunk)) { - await reader.cancel("non-streaming bridge consumed terminal SSE event").catch(() => {}); - break; - } - } - } catch (error) { - await reader.cancel(error).catch(() => {}); - throw error; - } finally { - rawBody += decoder.decode(); - reader.releaseLock(); - } - - return rawBody; -} - -function isSemaphoreCapacityError(error: unknown): error is Error & { code: string } { - return ( - !!error && - typeof error === "object" && - ((error as { code?: unknown }).code === "SEMAPHORE_TIMEOUT" || - (error as { code?: unknown }).code === "SEMAPHORE_QUEUE_FULL") - ); -} - -function createStreamingErrorResult( - statusCode: number, - message: string, - code?: string, - type?: string -) { - const errorBody = buildErrorBody(statusCode, message); - if (code) { - errorBody.error.code = code; - } - if (type) { - errorBody.error.type = type; - } - - const body = `data: ${JSON.stringify(errorBody)}\n\ndata: [DONE]\n\n`; - - return { - success: false as const, - status: statusCode, - error: message, - response: new Response(body, { - status: statusCode, - headers: { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, - }), - }; -} - -function getUpstreamErrorIdentifier(error: unknown): string | undefined { - if (!error || typeof error !== "object") return undefined; - const value = (error as { code?: unknown }).code; - return typeof value === "string" && value.length > 0 ? value : undefined; -} - -function wrapReadableStreamWithFinalize( - readable: ReadableStream, - finalize: () => void -): ReadableStream { - const reader = readable.getReader(); - let finalized = false; - - const runFinalize = () => { - if (finalized) return; - finalized = true; - finalize(); - }; - - return new ReadableStream({ - async pull(controller) { - try { - const { done, value } = await reader.read(); - if (done) { - runFinalize(); - controller.close(); - return; - } - controller.enqueue(value); - } catch (error) { - runFinalize(); - controller.error(error); - } - }, - - async cancel(reason) { - runFinalize(); - try { - await reader.cancel(reason); - } catch (error) { - // Ignored - } - }, - }); -} - -function toPositiveNumber(value: unknown) { - return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : 0; -} - -function buildCacheUsageLogMeta(usage: Record | null | undefined) { - if (!usage || typeof usage !== "object") return null; - const promptTokenDetails = - usage.prompt_tokens_details && typeof usage.prompt_tokens_details === "object" - ? (usage.prompt_tokens_details as Record) - : undefined; - const hasCacheFields = - "cache_read_input_tokens" in usage || - "cached_tokens" in usage || - "cache_creation_input_tokens" in usage || - (!!promptTokenDetails && - ("cached_tokens" in promptTokenDetails || "cache_creation_tokens" in promptTokenDetails)); - const cacheReadTokens = toPositiveNumber( - usage.cache_read_input_tokens ?? usage.cached_tokens ?? promptTokenDetails?.cached_tokens - ); - const cacheCreationTokens = toPositiveNumber( - usage.cache_creation_input_tokens ?? promptTokenDetails?.cache_creation_tokens - ); - if (!hasCacheFields) return null; - return { - cacheReadTokens, - cacheCreationTokens, - }; -} - -function attachLogMeta( - payload: Record | null | undefined, - meta: Record | null | undefined -) { - if (!meta || typeof meta !== "object") return payload; - const compactMeta = Object.fromEntries( - Object.entries(meta).filter(([, value]) => value !== null && value !== undefined) - ); - if (Object.keys(compactMeta).length === 0) return payload; - if (!payload || typeof payload !== "object" || Array.isArray(payload)) { - return { _omniroute: compactMeta, _payload: payload ?? null }; - } - const existing = - payload._omniroute && - typeof payload._omniroute === "object" && - !Array.isArray(payload._omniroute) - ? payload._omniroute - : {}; - return { - ...payload, - _omniroute: { - ...existing, - ...compactMeta, - }, - }; -} - /** * Core chat handler - shared between SSE and Worker * Returns { success, response, status, error } for caller to handle fallback @@ -514,33 +308,6 @@ function attachLogMeta( * @param {string} options.connectionId - Connection ID for settings lookup */ -function buildExecutorClientHeaders( - headers: Headers | Record | null | undefined, - userAgent?: string | null -) { - const normalized: Record = {}; - - if (headers instanceof Headers) { - headers.forEach((value, key) => { - normalized[key] = value; - }); - } else if (headers && typeof headers === "object") { - for (const [key, value] of Object.entries(headers)) { - if (typeof value === "string") { - normalized[key] = value; - } - } - } - - const normalizedUserAgent = typeof userAgent === "string" ? userAgent.trim() : ""; - if (normalizedUserAgent && !normalized["user-agent"] && !normalized["User-Agent"]) { - normalized["user-agent"] = normalizedUserAgent; - normalized["User-Agent"] = normalizedUserAgent; - } - - return Object.keys(normalized).length > 0 ? normalized : null; -} - // extractSystemRoleMessages extracted to chatCore/claudeSystemRole.ts (#3501); re-exported above so // existing importers (e.g. tests/unit/system-role-extraction.test.ts) keep resolving it from here. diff --git a/open-sse/handlers/chatCore/cacheUsageMeta.ts b/open-sse/handlers/chatCore/cacheUsageMeta.ts new file mode 100644 index 0000000000..ccb3b5707b --- /dev/null +++ b/open-sse/handlers/chatCore/cacheUsageMeta.ts @@ -0,0 +1,65 @@ +/** + * chatCore cache-usage log meta helpers (Quality Gate v2 / Fase 9 — chatCore god-file decomposition, + * #3501). + * + * Pure helpers extracted from chatCore: coerce an unknown to a positive number, derive cache + * read/creation token counts from a usage object (handling both top-level and prompt_tokens_details + * shapes), and attach an `_omniroute` meta blob to a log payload. Side-effect-free; behaviour is + * byte-identical to the previous module-level functions. + */ + +export function toPositiveNumber(value: unknown) { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : 0; +} + +export function buildCacheUsageLogMeta(usage: Record | null | undefined) { + if (!usage || typeof usage !== "object") return null; + const promptTokenDetails = + usage.prompt_tokens_details && typeof usage.prompt_tokens_details === "object" + ? (usage.prompt_tokens_details as Record) + : undefined; + const hasCacheFields = + "cache_read_input_tokens" in usage || + "cached_tokens" in usage || + "cache_creation_input_tokens" in usage || + (!!promptTokenDetails && + ("cached_tokens" in promptTokenDetails || "cache_creation_tokens" in promptTokenDetails)); + const cacheReadTokens = toPositiveNumber( + usage.cache_read_input_tokens ?? usage.cached_tokens ?? promptTokenDetails?.cached_tokens + ); + const cacheCreationTokens = toPositiveNumber( + usage.cache_creation_input_tokens ?? promptTokenDetails?.cache_creation_tokens + ); + if (!hasCacheFields) return null; + return { + cacheReadTokens, + cacheCreationTokens, + }; +} + +export function attachLogMeta( + payload: Record | null | undefined, + meta: Record | null | undefined +) { + if (!meta || typeof meta !== "object") return payload; + const compactMeta = Object.fromEntries( + Object.entries(meta).filter(([, value]) => value !== null && value !== undefined) + ); + if (Object.keys(compactMeta).length === 0) return payload; + if (!payload || typeof payload !== "object" || Array.isArray(payload)) { + return { _omniroute: compactMeta, _payload: payload ?? null }; + } + const existing = + payload._omniroute && + typeof payload._omniroute === "object" && + !Array.isArray(payload._omniroute) + ? payload._omniroute + : {}; + return { + ...payload, + _omniroute: { + ...existing, + ...compactMeta, + }, + }; +} diff --git a/open-sse/handlers/chatCore/executorClientHeaders.ts b/open-sse/handlers/chatCore/executorClientHeaders.ts new file mode 100644 index 0000000000..e2a77bf36e --- /dev/null +++ b/open-sse/handlers/chatCore/executorClientHeaders.ts @@ -0,0 +1,36 @@ +/** + * chatCore executor client-header normalizer (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Pure helper extracted from chatCore: normalizes a Headers instance or a plain header object into a + * lowercased-tolerant Record, and backfills the client User-Agent (both casings) when + * one is supplied and not already present. Returns null when nothing was collected. Side-effect-free; + * behaviour is byte-identical to the previous module-level function. + */ + +export function buildExecutorClientHeaders( + headers: Headers | Record | null | undefined, + userAgent?: string | null +) { + const normalized: Record = {}; + + if (headers instanceof Headers) { + headers.forEach((value, key) => { + normalized[key] = value; + }); + } else if (headers && typeof headers === "object") { + for (const [key, value] of Object.entries(headers)) { + if (typeof value === "string") { + normalized[key] = value; + } + } + } + + const normalizedUserAgent = typeof userAgent === "string" ? userAgent.trim() : ""; + if (normalizedUserAgent && !normalized["user-agent"] && !normalized["User-Agent"]) { + normalized["user-agent"] = normalizedUserAgent; + normalized["User-Agent"] = normalizedUserAgent; + } + + return Object.keys(normalized).length > 0 ? normalized : null; +} diff --git a/open-sse/handlers/chatCore/nonStreamingResponseBody.ts b/open-sse/handlers/chatCore/nonStreamingResponseBody.ts new file mode 100644 index 0000000000..c55c7c0261 --- /dev/null +++ b/open-sse/handlers/chatCore/nonStreamingResponseBody.ts @@ -0,0 +1,68 @@ +/** + * chatCore non-streaming response-body reader (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Extracted from chatCore: reads an upstream response body to a string. When the upstream is an SSE / + * NDJSON stream consumed in non-streaming mode, it drains the reader chunk-by-chunk under the body + * timeout and cancels early once a terminal SSE signal is observed; otherwise it falls back to a + * timeout-bounded response.text(). Behaviour is byte-identical to the previous module-level function. + */ + +import { withBodyTimeout } from "../../utils/stream.ts"; +import { FETCH_BODY_TIMEOUT_MS } from "../../config/constants.ts"; +import { createBodyTimeoutError, readStreamChunkWithTimeout } from "./upstreamTimeouts.ts"; +import { + appendNonStreamingSseTerminalSignal, + type NonStreamingSseTerminalState, +} from "./nonStreamingSse.ts"; + +export async function readNonStreamingResponseBody( + response: Response, + contentType: string, + upstreamStream: boolean +): Promise { + if ( + !upstreamStream || + !response.body || + (!contentType.includes("text/event-stream") && !contentType.includes("application/x-ndjson")) + ) { + return withBodyTimeout(response.text()); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const terminalState: NonStreamingSseTerminalState = { + currentEvent: "", + pendingLine: "", + }; + let rawBody = ""; + const deadline = FETCH_BODY_TIMEOUT_MS > 0 ? Date.now() + FETCH_BODY_TIMEOUT_MS : 0; + + try { + while (true) { + const timeoutMs = deadline > 0 ? deadline - Date.now() : 0; + if (deadline > 0 && timeoutMs <= 0) { + throw createBodyTimeoutError(FETCH_BODY_TIMEOUT_MS); + } + + const { done, value } = await readStreamChunkWithTimeout(reader, timeoutMs); + if (done) break; + if (!value) continue; + + const decodedChunk = decoder.decode(value, { stream: true }); + rawBody += decodedChunk; + if (appendNonStreamingSseTerminalSignal(terminalState, decodedChunk)) { + await reader.cancel("non-streaming bridge consumed terminal SSE event").catch(() => {}); + break; + } + } + } catch (error) { + await reader.cancel(error).catch(() => {}); + throw error; + } finally { + rawBody += decoder.decode(); + reader.releaseLock(); + } + + return rawBody; +} diff --git a/open-sse/handlers/chatCore/skillsFormat.ts b/open-sse/handlers/chatCore/skillsFormat.ts new file mode 100644 index 0000000000..8007d99564 --- /dev/null +++ b/open-sse/handlers/chatCore/skillsFormat.ts @@ -0,0 +1,33 @@ +/** + * chatCore skills-format mappers (Quality Gate v2 / Fase 9 — chatCore god-file decomposition, #3501). + * + * Pure mappers extracted from chatCore: translate the request wire format into the skills provider + * bucket and the skills model id used when injecting skill context. Side-effect-free; behaviour is + * byte-identical to the previous module-level functions. + */ + +import { FORMATS } from "../../translator/formats.ts"; + +export function getSkillsProviderForFormat( + format: string +): "openai" | "anthropic" | "google" | "other" { + switch (format) { + case FORMATS.CLAUDE: + return "anthropic"; + case FORMATS.GEMINI: + return "google"; + default: + return "openai"; + } +} + +export function getSkillsModelIdForFormat(format: string): string { + switch (format) { + case FORMATS.CLAUDE: + return "claude"; + case FORMATS.GEMINI: + return "gemini"; + default: + return "openai"; + } +} diff --git a/open-sse/handlers/chatCore/streamErrorResult.ts b/open-sse/handlers/chatCore/streamErrorResult.ts new file mode 100644 index 0000000000..77244b611d --- /dev/null +++ b/open-sse/handlers/chatCore/streamErrorResult.ts @@ -0,0 +1,58 @@ +/** + * chatCore streaming error-result helpers (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Extracted from chatCore: identify semaphore capacity errors, build a sanitized SSE error result + * (an `data: {...}\n\ndata: [DONE]\n\n` body wrapped in an event-stream Response), and pull a string + * error code off an unknown error. Side-effect-free; behaviour is byte-identical to the previous + * module-level functions. + */ + +import { buildErrorBody } from "../../utils/error.ts"; + +export function isSemaphoreCapacityError(error: unknown): error is Error & { code: string } { + return ( + !!error && + typeof error === "object" && + ((error as { code?: unknown }).code === "SEMAPHORE_TIMEOUT" || + (error as { code?: unknown }).code === "SEMAPHORE_QUEUE_FULL") + ); +} + +export function createStreamingErrorResult( + statusCode: number, + message: string, + code?: string, + type?: string +) { + const errorBody = buildErrorBody(statusCode, message); + if (code) { + errorBody.error.code = code; + } + if (type) { + errorBody.error.type = type; + } + + const body = `data: ${JSON.stringify(errorBody)}\n\ndata: [DONE]\n\n`; + + return { + success: false as const, + status: statusCode, + error: message, + response: new Response(body, { + status: statusCode, + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", + }, + }), + }; +} + +export function getUpstreamErrorIdentifier(error: unknown): string | undefined { + if (!error || typeof error !== "object") return undefined; + const value = (error as { code?: unknown }).code; + return typeof value === "string" && value.length > 0 ? value : undefined; +} diff --git a/open-sse/handlers/chatCore/streamFinalize.ts b/open-sse/handlers/chatCore/streamFinalize.ts new file mode 100644 index 0000000000..83ef8dec7e --- /dev/null +++ b/open-sse/handlers/chatCore/streamFinalize.ts @@ -0,0 +1,48 @@ +/** + * chatCore stream finalize wrapper (Quality Gate v2 / Fase 9 — chatCore god-file decomposition, + * #3501). + * + * Extracted from chatCore: wraps a ReadableStream so a `finalize` callback runs exactly once when the + * stream is fully drained, errors, or is cancelled. Side-effect-free other than the wrapped stream's + * own lifecycle; behaviour is byte-identical to the previous module-level function. + */ + +export function wrapReadableStreamWithFinalize( + readable: ReadableStream, + finalize: () => void +): ReadableStream { + const reader = readable.getReader(); + let finalized = false; + + const runFinalize = () => { + if (finalized) return; + finalized = true; + finalize(); + }; + + return new ReadableStream({ + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (done) { + runFinalize(); + controller.close(); + return; + } + controller.enqueue(value); + } catch (error) { + runFinalize(); + controller.error(error); + } + }, + + async cancel(reason) { + runFinalize(); + try { + await reader.cancel(reason); + } catch (error) { + // Ignored + } + }, + }); +} diff --git a/tests/unit/chatcore-cache-usage-meta.test.ts b/tests/unit/chatcore-cache-usage-meta.test.ts new file mode 100644 index 0000000000..0a247c0e7e --- /dev/null +++ b/tests/unit/chatcore-cache-usage-meta.test.ts @@ -0,0 +1,58 @@ +// tests/unit/chatcore-cache-usage-meta.test.ts +// Characterization of toPositiveNumber / buildCacheUsageLogMeta / attachLogMeta — cache-usage log +// meta helpers extracted from handleChatCore (chatCore god-file decomposition, #3501). Locks: the +// positive-number coercion, the cache-token derivation across top-level and prompt_tokens_details +// shapes (null when no cache fields), and the _omniroute meta attachment/merge. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + toPositiveNumber, + buildCacheUsageLogMeta, + attachLogMeta, +} from "../../open-sse/handlers/chatCore/cacheUsageMeta.ts"; + +test("toPositiveNumber keeps finite positives, zeros everything else", () => { + assert.equal(toPositiveNumber(5), 5); + assert.equal(toPositiveNumber(0), 0); + assert.equal(toPositiveNumber(-3), 0); + assert.equal(toPositiveNumber(Infinity), 0); + assert.equal(toPositiveNumber("5"), 0); + assert.equal(toPositiveNumber(null), 0); +}); + +test("buildCacheUsageLogMeta returns null when there are no cache fields", () => { + assert.equal(buildCacheUsageLogMeta(null), null); + assert.equal(buildCacheUsageLogMeta({ prompt_tokens: 10 }), null); +}); + +test("buildCacheUsageLogMeta reads top-level cache fields", () => { + const meta = buildCacheUsageLogMeta({ + cache_read_input_tokens: 12, + cache_creation_input_tokens: 4, + }); + assert.deepEqual(meta, { cacheReadTokens: 12, cacheCreationTokens: 4 }); +}); + +test("buildCacheUsageLogMeta reads prompt_tokens_details shapes", () => { + const meta = buildCacheUsageLogMeta({ + prompt_tokens_details: { cached_tokens: 7, cache_creation_tokens: 2 }, + }); + assert.deepEqual(meta, { cacheReadTokens: 7, cacheCreationTokens: 2 }); +}); + +test("attachLogMeta returns the payload untouched when meta is empty", () => { + const payload = { a: 1 }; + assert.equal(attachLogMeta(payload, null), payload); + assert.equal(attachLogMeta(payload, {}), payload); + assert.equal(attachLogMeta(payload, { x: null, y: undefined }), payload); +}); + +test("attachLogMeta merges compact meta into _omniroute", () => { + const out = attachLogMeta({ a: 1, _omniroute: { keep: true } }, { added: 2, drop: null }); + assert.deepEqual(out, { a: 1, _omniroute: { keep: true, added: 2 } }); +}); + +test("attachLogMeta wraps non-object payloads", () => { + const out = attachLogMeta(null, { added: 2 }); + assert.deepEqual(out, { _omniroute: { added: 2 }, _payload: null }); +}); diff --git a/tests/unit/chatcore-executor-client-headers.test.ts b/tests/unit/chatcore-executor-client-headers.test.ts new file mode 100644 index 0000000000..92ff2a8c5b --- /dev/null +++ b/tests/unit/chatcore-executor-client-headers.test.ts @@ -0,0 +1,42 @@ +// tests/unit/chatcore-executor-client-headers.test.ts +// Characterization of buildExecutorClientHeaders — the executor client-header normalizer extracted +// from handleChatCore (chatCore god-file decomposition, #3501). Locks: Headers and plain-object +// normalization, non-string value skipping, User-Agent backfill (both casings, only when absent), +// and the null-when-empty return. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { buildExecutorClientHeaders } from "../../open-sse/handlers/chatCore/executorClientHeaders.ts"; + +test("returns null for empty / nullish inputs", () => { + assert.equal(buildExecutorClientHeaders(null), null); + assert.equal(buildExecutorClientHeaders(undefined), null); + assert.equal(buildExecutorClientHeaders({}), null); +}); + +test("normalizes a Headers instance", () => { + const h = new Headers({ "X-Test": "1", "content-type": "application/json" }); + const out = buildExecutorClientHeaders(h); + assert.equal(out?.["content-type"], "application/json"); + assert.equal(out?.["x-test"], "1"); +}); + +test("normalizes a plain object and skips non-string values", () => { + const out = buildExecutorClientHeaders({ a: "1", b: 2, c: null } as Record); + assert.deepEqual(out, { a: "1" }); +}); + +test("backfills the User-Agent in both casings when absent", () => { + const out = buildExecutorClientHeaders({ a: "1" }, " MyAgent/1.0 "); + assert.equal(out?.["user-agent"], "MyAgent/1.0"); + assert.equal(out?.["User-Agent"], "MyAgent/1.0"); +}); + +test("does not overwrite an existing user-agent header", () => { + const out = buildExecutorClientHeaders({ "user-agent": "Existing/9" }, "MyAgent/1.0"); + assert.equal(out?.["user-agent"], "Existing/9"); + assert.equal(out?.["User-Agent"], undefined); +}); + +test("a trimmed-empty user agent does not create headers on its own", () => { + assert.equal(buildExecutorClientHeaders({}, " "), null); +}); diff --git a/tests/unit/chatcore-non-streaming-response-body.test.ts b/tests/unit/chatcore-non-streaming-response-body.test.ts new file mode 100644 index 0000000000..41d52a2e37 --- /dev/null +++ b/tests/unit/chatcore-non-streaming-response-body.test.ts @@ -0,0 +1,33 @@ +// tests/unit/chatcore-non-streaming-response-body.test.ts +// Characterization of readNonStreamingResponseBody — the non-streaming body reader extracted from +// handleChatCore (chatCore god-file decomposition, #3501). Locks: the response.text() fallback path +// (non-stream, or non-SSE content type) and the SSE-drain path that concatenates chunks until the +// stream closes. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { readNonStreamingResponseBody } from "../../open-sse/handlers/chatCore/nonStreamingResponseBody.ts"; + +test("falls back to response.text() when upstream is not streaming", async () => { + const out = await readNonStreamingResponseBody(new Response("hello"), "application/json", false); + assert.equal(out, "hello"); +}); + +test("falls back to response.text() for a non-SSE content type even when streaming", async () => { + const out = await readNonStreamingResponseBody(new Response("plain"), "application/json", true); + assert.equal(out, "plain"); +}); + +test("drains an SSE stream chunk-by-chunk and concatenates until close", async () => { + const enc = new TextEncoder(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(enc.encode("data: {\"a\":1}\n\n")); + controller.enqueue(enc.encode("data: {\"b\":2}\n\n")); + controller.close(); + }, + }); + const response = new Response(body, { headers: { "Content-Type": "text/event-stream" } }); + const out = await readNonStreamingResponseBody(response, "text/event-stream", true); + assert.ok(out.includes('"a":1')); + assert.ok(out.includes('"b":2')); +}); diff --git a/tests/unit/chatcore-skills-format.test.ts b/tests/unit/chatcore-skills-format.test.ts new file mode 100644 index 0000000000..63f5f6d555 --- /dev/null +++ b/tests/unit/chatcore-skills-format.test.ts @@ -0,0 +1,32 @@ +// tests/unit/chatcore-skills-format.test.ts +// Characterization of getSkillsProviderForFormat / getSkillsModelIdForFormat — the skills-format +// mappers extracted from handleChatCore (chatCore god-file decomposition, #3501). Locks the +// claude→anthropic/claude, gemini→google/gemini and default→openai/openai mappings. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + getSkillsProviderForFormat, + getSkillsModelIdForFormat, +} from "../../open-sse/handlers/chatCore/skillsFormat.ts"; + +test("claude format maps to anthropic provider and claude model id", () => { + assert.equal(getSkillsProviderForFormat("claude"), "anthropic"); + assert.equal(getSkillsModelIdForFormat("claude"), "claude"); +}); + +test("gemini format maps to google provider and gemini model id", () => { + assert.equal(getSkillsProviderForFormat("gemini"), "google"); + assert.equal(getSkillsModelIdForFormat("gemini"), "gemini"); +}); + +test("openai format maps to openai provider and openai model id", () => { + assert.equal(getSkillsProviderForFormat("openai"), "openai"); + assert.equal(getSkillsModelIdForFormat("openai"), "openai"); +}); + +test("unknown / responses formats fall back to openai for both mappers", () => { + assert.equal(getSkillsProviderForFormat("openai-responses"), "openai"); + assert.equal(getSkillsModelIdForFormat("openai-responses"), "openai"); + assert.equal(getSkillsProviderForFormat("gemini-cli"), "openai"); + assert.equal(getSkillsModelIdForFormat("totally-unknown"), "openai"); +}); diff --git a/tests/unit/chatcore-stream-error-result.test.ts b/tests/unit/chatcore-stream-error-result.test.ts new file mode 100644 index 0000000000..352877046a --- /dev/null +++ b/tests/unit/chatcore-stream-error-result.test.ts @@ -0,0 +1,52 @@ +// tests/unit/chatcore-stream-error-result.test.ts +// Characterization of isSemaphoreCapacityError / createStreamingErrorResult / +// getUpstreamErrorIdentifier — streaming error-result helpers extracted from handleChatCore +// (chatCore god-file decomposition, #3501). Locks the semaphore code matching, the SSE error +// envelope shape (status, headers, `data: {...}\n\ndata: [DONE]\n\n` body, optional code/type), and +// the string-code extraction. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + isSemaphoreCapacityError, + createStreamingErrorResult, + getUpstreamErrorIdentifier, +} from "../../open-sse/handlers/chatCore/streamErrorResult.ts"; + +test("isSemaphoreCapacityError matches the two semaphore codes only", () => { + assert.equal(isSemaphoreCapacityError({ code: "SEMAPHORE_TIMEOUT" }), true); + assert.equal(isSemaphoreCapacityError({ code: "SEMAPHORE_QUEUE_FULL" }), true); + assert.equal(isSemaphoreCapacityError({ code: "OTHER" }), false); + assert.equal(isSemaphoreCapacityError(null), false); + assert.equal(isSemaphoreCapacityError("SEMAPHORE_TIMEOUT"), false); +}); + +test("createStreamingErrorResult builds an SSE error envelope with [DONE] terminator", async () => { + const result = createStreamingErrorResult(503, "boom"); + assert.equal(result.success, false); + assert.equal(result.status, 503); + assert.equal(result.error, "boom"); + assert.equal(result.response.status, 503); + assert.equal(result.response.headers.get("Content-Type"), "text/event-stream"); + assert.equal(result.response.headers.get("X-Accel-Buffering"), "no"); + const body = await result.response.text(); + assert.ok(body.startsWith("data: ")); + assert.ok(body.endsWith("data: [DONE]\n\n")); + const json = JSON.parse(body.slice("data: ".length, body.indexOf("\n\n"))); + assert.equal(json.error.message, "boom"); +}); + +test("createStreamingErrorResult attaches optional code and type", async () => { + const result = createStreamingErrorResult(429, "slow down", "rate_limited", "rate_limit_error"); + const body = await result.response.text(); + const json = JSON.parse(body.slice("data: ".length, body.indexOf("\n\n"))); + assert.equal(json.error.code, "rate_limited"); + assert.equal(json.error.type, "rate_limit_error"); +}); + +test("getUpstreamErrorIdentifier returns a non-empty string code or undefined", () => { + assert.equal(getUpstreamErrorIdentifier({ code: "ECONNRESET" }), "ECONNRESET"); + assert.equal(getUpstreamErrorIdentifier({ code: "" }), undefined); + assert.equal(getUpstreamErrorIdentifier({ code: 123 }), undefined); + assert.equal(getUpstreamErrorIdentifier(null), undefined); + assert.equal(getUpstreamErrorIdentifier("ECONNRESET"), undefined); +}); diff --git a/tests/unit/chatcore-stream-finalize.test.ts b/tests/unit/chatcore-stream-finalize.test.ts new file mode 100644 index 0000000000..4b7ede3c22 --- /dev/null +++ b/tests/unit/chatcore-stream-finalize.test.ts @@ -0,0 +1,47 @@ +// tests/unit/chatcore-stream-finalize.test.ts +// Characterization of wrapReadableStreamWithFinalize — the stream finalize wrapper extracted from +// handleChatCore (chatCore god-file decomposition, #3501). Locks: finalize runs exactly once on +// full drain, on cancel, and is not double-invoked. +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { wrapReadableStreamWithFinalize } from "../../open-sse/handlers/chatCore/streamFinalize.ts"; + +function streamOf(chunks: unknown[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) { + controller.enqueue(chunks[i++]); + } else { + controller.close(); + } + }, + }); +} + +test("finalize runs exactly once after the stream is fully drained", async () => { + let calls = 0; + const wrapped = wrapReadableStreamWithFinalize(streamOf(["a", "b"]), () => { + calls++; + }); + const reader = wrapped.getReader(); + const seen: unknown[] = []; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + seen.push(value); + } + assert.deepEqual(seen, ["a", "b"]); + assert.equal(calls, 1); +}); + +test("finalize runs exactly once on cancel", async () => { + let calls = 0; + const wrapped = wrapReadableStreamWithFinalize(streamOf(["a", "b", "c"]), () => { + calls++; + }); + const reader = wrapped.getReader(); + await reader.read(); + await reader.cancel("done early"); + assert.equal(calls, 1); +});