From 89aa09f6dfc0b25d33238dbb2394e3968e8ab756 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 10:07:24 -0300 Subject: [PATCH] =?UTF-8?q?refactor(chatCore):=20extrai=20parse=20+=20usag?= =?UTF-8?q?e-stats=20n=C3=A3o-streaming=20do=20executeProviderRequest=20(#?= =?UTF-8?q?3501)=20(#4762)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chatCore #3501: extract parseNonStreamingResponseBody + recordNonStreamingUsageStats. Integrated into release/v3.8.35. --- open-sse/handlers/chatCore.ts | 191 ++++++------------ .../chatCore/nonStreamingResponseParse.ts | 135 +++++++++++++ .../chatCore/nonStreamingUsageStats.ts | 88 ++++++++ ...tcore-non-streaming-response-parse.test.ts | 157 ++++++++++++++ ...chatcore-non-streaming-usage-stats.test.ts | 129 ++++++++++++ 5 files changed, 576 insertions(+), 124 deletions(-) create mode 100644 open-sse/handlers/chatCore/nonStreamingResponseParse.ts create mode 100644 open-sse/handlers/chatCore/nonStreamingUsageStats.ts create mode 100644 tests/unit/chatcore-non-streaming-response-parse.test.ts create mode 100644 tests/unit/chatcore-non-streaming-usage-stats.test.ts diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index f823275daa..5d0c502cf4 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -157,7 +157,6 @@ import { ensureEngineBreakdown } from "../services/compression/engineBreakdown.t import { handleBypassRequest } from "../utils/bypassHandler.ts"; import { saveRequestUsage, trackPendingRequest, appendRequestLog } from "@/lib/usageDb"; import { finalizePendingScope, updatePendingScope } from "@/lib/usage/pendingRequestScope"; -import { formatUsageLog } from "@/lib/usage/tokenAccounting"; import { recordCost } from "@/domain/costRules"; import { calculateCost } from "@/lib/usage/costCalculator"; import { attachOmniRouteMetaHeaders } from "@/domain/omnirouteResponseMeta"; @@ -167,12 +166,11 @@ import { mergeResponseToolNameMap, } from "./chatCore/passthroughToolNames.ts"; import { - parseNonStreamingSSEPayload, - normalizeNonStreamingEventPayload, - shouldTreatBufferedEventResponseAsExpected, appendNonStreamingSseTerminalSignal, type NonStreamingSseTerminalState, } from "./chatCore/nonStreamingSse.ts"; +import { parseNonStreamingResponseBody } from "./chatCore/nonStreamingResponseParse.ts"; +import { recordNonStreamingUsageStats } from "./chatCore/nonStreamingUsageStats.ts"; import { createBodyTimeoutError, readStreamChunkWithTimeout, @@ -207,7 +205,6 @@ import { buildCodexQuotaPersistence } from "./chatCore/codexQuota.ts"; import { invalidateCodexQuotaCache } from "../services/codexQuotaFetcher.ts"; import { translateNonStreamingResponse } from "./responseTranslator.ts"; import { extractUsageFromResponse } from "./usageExtractor.ts"; -import { extractSSEErrorMessage } from "./sseParser.ts"; import { sanitizeOpenAIResponse, sanitizeResponsesApiResponse } from "./responseSanitizer.ts"; import { withRateLimit, @@ -254,7 +251,6 @@ import { stripMarkdownCodeFence, } from "../utils/aiSdkCompat.ts"; import { generateRequestId } from "@/shared/utils/requestId"; -import { normalizePayloadForLog } from "@/lib/logPayloads"; import { extractFacts } from "@/lib/memory/extraction"; import { handleToolCallExecution } from "@/lib/skills/interception"; import { OMNIROUTE_RESPONSE_HEADERS } from "@/shared/constants/headers"; @@ -3382,93 +3378,64 @@ export async function handleChatCore({ // Non-streaming response if (!stream) { - const contentType = (providerResponse.headers.get("content-type") || "").toLowerCase(); - let responseBody; - let responsePayloadFormat = targetFormat; - const rawBody = await readNonStreamingResponseBody( + const parsed = await parseNonStreamingResponseBody({ providerResponse, - contentType, - upstreamStream - ); - const normalizedProviderPayload = normalizePayloadForLog(rawBody); - const looksLikeSSE = - contentType.includes("text/event-stream") || - contentType.includes("application/x-ndjson") || - /(^|\n)\s*(event|data):/m.test(rawBody); + upstreamStream, + providerHeaders, + finalBody, + targetFormat, + model, + log, + }); + const normalizedProviderPayload = parsed.normalizedProviderPayload; + const looksLikeSSE = parsed.looksLikeSSE; - if (looksLikeSSE) { - const streamPayload = normalizeNonStreamingEventPayload(rawBody, contentType); - const streamKind = contentType.includes("application/x-ndjson") ? "NDJSON" : "SSE"; - if (shouldTreatBufferedEventResponseAsExpected(upstreamStream, providerHeaders, finalBody)) { - log?.debug?.( - "STREAM", - `Buffering upstream ${streamKind} response for non-streaming client request` - ); - } else { - log?.warn?.( - "STREAM", - `Unexpected ${streamKind} response for non-streaming request — buffering` - ); - } - // Upstream returned an event stream for a non-streaming client; convert best-effort to JSON. - const parsedFromSSE = parseNonStreamingSSEPayload(streamPayload, targetFormat, model); - - if (!parsedFromSSE) { - appendRequestLog({ - model, - provider, - connectionId, - status: `FAILED ${HTTP_STATUS.BAD_GATEWAY}`, - }).catch(() => {}); - // Some executors (e.g. the Devin/Windsurf CLI) always emit - // text/event-stream, signalling failure with an error-only chunk - // (`data: {"error":{"message":"Devin CLI not found..."}}`) that carries - // no `choices`. Surface that real, sanitized message instead of the - // generic 502 so the actionable error is not swallowed (#3324). - const surfacedSseError = extractSSEErrorMessage(streamPayload); - const invalidSseMessage = - surfacedSseError || "Invalid SSE response for non-streaming request"; - persistAttemptLogs({ - status: HTTP_STATUS.BAD_GATEWAY, - error: invalidSseMessage, - providerRequest: finalBody || translatedBody, - providerResponse: normalizedProviderPayload, - clientResponse: buildErrorBody(HTTP_STATUS.BAD_GATEWAY, invalidSseMessage), - cacheSource: "upstream", - }); - persistFailureUsage(HTTP_STATUS.BAD_GATEWAY, "invalid_sse_payload"); - trackPendingRequest(model, provider, pendingConnId, false); - return createErrorResult(HTTP_STATUS.BAD_GATEWAY, invalidSseMessage); - } - - responseBody = parsedFromSSE.body; - responsePayloadFormat = parsedFromSSE.format; - } else { - try { - responseBody = rawBody ? JSON.parse(rawBody) : {}; - } catch (err) { - appendRequestLog({ - model, - provider, - connectionId, - status: `FAILED ${HTTP_STATUS.BAD_GATEWAY}`, - }).catch(() => {}); - const detailedError = `Invalid JSON response from provider (error: ${err instanceof Error ? err.message : String(err)}): ${rawBody.substring(0, 1000)}`; - const invalidJsonMessage = "Invalid JSON response from provider"; - persistAttemptLogs({ - status: HTTP_STATUS.BAD_GATEWAY, - error: detailedError, - providerRequest: finalBody || translatedBody, - providerResponse: normalizedProviderPayload, - clientResponse: buildErrorBody(HTTP_STATUS.BAD_GATEWAY, invalidJsonMessage), - cacheSource: "upstream", - }); - persistFailureUsage(HTTP_STATUS.BAD_GATEWAY, "invalid_json_payload"); - trackPendingRequest(model, provider, connectionId, false); - return createErrorResult(HTTP_STATUS.BAD_GATEWAY, invalidJsonMessage); - } + if (parsed.kind === "invalid_sse") { + appendRequestLog({ + model, + provider, + connectionId, + status: `FAILED ${HTTP_STATUS.BAD_GATEWAY}`, + }).catch(() => {}); + const invalidSseMessage = parsed.message; + persistAttemptLogs({ + status: HTTP_STATUS.BAD_GATEWAY, + error: invalidSseMessage, + providerRequest: finalBody || translatedBody, + providerResponse: normalizedProviderPayload, + clientResponse: buildErrorBody(HTTP_STATUS.BAD_GATEWAY, invalidSseMessage), + cacheSource: "upstream", + }); + persistFailureUsage(HTTP_STATUS.BAD_GATEWAY, "invalid_sse_payload"); + trackPendingRequest(model, provider, pendingConnId, false); + return createErrorResult(HTTP_STATUS.BAD_GATEWAY, invalidSseMessage); } + if (parsed.kind === "invalid_json") { + appendRequestLog({ + model, + provider, + connectionId, + status: `FAILED ${HTTP_STATUS.BAD_GATEWAY}`, + }).catch(() => {}); + const detailedError = parsed.detailedError; + const invalidJsonMessage = parsed.message; + persistAttemptLogs({ + status: HTTP_STATUS.BAD_GATEWAY, + error: detailedError, + providerRequest: finalBody || translatedBody, + providerResponse: normalizedProviderPayload, + clientResponse: buildErrorBody(HTTP_STATUS.BAD_GATEWAY, invalidJsonMessage), + cacheSource: "upstream", + }); + persistFailureUsage(HTTP_STATUS.BAD_GATEWAY, "invalid_json_payload"); + trackPendingRequest(model, provider, connectionId, false); + return createErrorResult(HTTP_STATUS.BAD_GATEWAY, invalidJsonMessage); + } + + let responseBody = parsed.responseBody; + let responsePayloadFormat = parsed.responsePayloadFormat; + // Check for empty content response (fake success) - trigger fallback if (isEmptyContentResponse(responseBody)) { appendRequestLog({ @@ -3598,41 +3565,17 @@ export async function handleChatCore({ // Save structured call log with full payloads const cacheUsageLogMeta = buildCacheUsageLogMeta(usage); - if (usage && typeof usage === "object") { - if (traceEnabled) { - const msg = `[${new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit" })}] 📊 [USAGE] ${provider?.toUpperCase()} | ${formatUsageLog(usage)}${connectionId ? ` | account=${connectionId.slice(0, 8)}...` : ""}`; - console.log(`${COLORS.green}${msg}${COLORS.reset}`); - } - - saveRequestUsage({ - provider: provider || "unknown", - model: model || "unknown", - tokens: usage, - status: "200", - success: true, - latencyMs: Date.now() - startTime, - timeToFirstTokenMs: Date.now() - startTime, - errorCode: null, - timestamp: new Date().toISOString(), - connectionId: connectionId || undefined, - apiKeyId: apiKeyInfo?.id || undefined, - apiKeyName: apiKeyInfo?.name || undefined, - serviceTier: effectiveServiceTier, - comboStrategy: isCombo ? comboStrategy || undefined : undefined, - }).catch((err) => { - console.error("Failed to save usage stats:", err.message); - }); - - if (apiKeyInfo?.id) { - try { - const billable = computeBillableTokens(usage); - if (billable > 0) - recordTokenUsage(apiKeyInfo.id, provider || "unknown", model || "unknown", billable); - } catch { - // never block the response on counter recording - } - } - } + recordNonStreamingUsageStats(usage, { + traceEnabled, + provider, + connectionId, + model, + startTime, + apiKeyInfo, + effectiveServiceTier, + isCombo, + comboStrategy, + }); // Translate response to client's expected format (usually OpenAI) // Pass toolNameMap so Claude OAuth proxy_ prefix is stripped in tool_use blocks (#605) diff --git a/open-sse/handlers/chatCore/nonStreamingResponseParse.ts b/open-sse/handlers/chatCore/nonStreamingResponseParse.ts new file mode 100644 index 0000000000..fcdf4d072e --- /dev/null +++ b/open-sse/handlers/chatCore/nonStreamingResponseParse.ts @@ -0,0 +1,135 @@ +/** + * chatCore non-streaming response parsing/classification (Quality Gate v2 / Fase 9 — chatCore + * god-file decomposition, #3501 — response-handling slice of executeProviderRequest). + * + * Extracted from handleChatCore's `if (!stream)` block: reads the upstream non-streaming body, + * detects whether it is an event stream (SSE / NDJSON) buffered for a non-streaming client, and + * parses it into a JSON response body. Returns a discriminated union describing the outcome + * (`ok` | `invalid_sse` | `invalid_json`) and leaves every persistence side-effect + * (appendRequestLog / persistAttemptLogs / persistFailureUsage / trackPendingRequest / + * createErrorResult) to the handler, so behaviour is observably identical to the previous inline + * block. Pure with respect to handler state (only buffering debug/warn logs as a side effect). + */ + +import { normalizePayloadForLog } from "@/lib/logPayloads"; +import { extractSSEErrorMessage } from "../sseParser.ts"; +import { readNonStreamingResponseBody } from "./nonStreamingResponseBody.ts"; +import { + normalizeNonStreamingEventPayload, + parseNonStreamingSSEPayload, + shouldTreatBufferedEventResponseAsExpected, +} from "./nonStreamingSse.ts"; + +type LoggerLike = + | { + debug?: (...args: unknown[]) => void; + warn?: (...args: unknown[]) => void; + } + | null + | undefined; + +export type NonStreamingParseResult = + | { + kind: "ok"; + responseBody: unknown; + responsePayloadFormat: string; + looksLikeSSE: boolean; + normalizedProviderPayload: unknown; + } + | { + kind: "invalid_sse"; + message: string; + looksLikeSSE: true; + normalizedProviderPayload: unknown; + } + | { + kind: "invalid_json"; + message: string; + detailedError: string; + looksLikeSSE: false; + normalizedProviderPayload: unknown; + }; + +export async function parseNonStreamingResponseBody(opts: { + providerResponse: Response; + upstreamStream: boolean; + providerHeaders: Record | Headers | null | undefined; + finalBody: unknown; + targetFormat: string; + model: string; + log?: LoggerLike; +}): Promise { + const { providerResponse, upstreamStream, providerHeaders, finalBody, targetFormat, model, log } = + opts; + + const contentType = (providerResponse.headers.get("content-type") || "").toLowerCase(); + const rawBody = await readNonStreamingResponseBody(providerResponse, contentType, upstreamStream); + const normalizedProviderPayload = normalizePayloadForLog(rawBody); + const looksLikeSSE = + contentType.includes("text/event-stream") || + contentType.includes("application/x-ndjson") || + /(^|\n)\s*(event|data):/m.test(rawBody); + + if (looksLikeSSE) { + const streamPayload = normalizeNonStreamingEventPayload(rawBody, contentType); + const streamKind = contentType.includes("application/x-ndjson") ? "NDJSON" : "SSE"; + if (shouldTreatBufferedEventResponseAsExpected(upstreamStream, providerHeaders, finalBody)) { + log?.debug?.( + "STREAM", + `Buffering upstream ${streamKind} response for non-streaming client request` + ); + } else { + log?.warn?.( + "STREAM", + `Unexpected ${streamKind} response for non-streaming request — buffering` + ); + } + // Upstream returned an event stream for a non-streaming client; convert best-effort to JSON. + const parsedFromSSE = parseNonStreamingSSEPayload(streamPayload, targetFormat, model); + + if (!parsedFromSSE) { + // Some executors (e.g. the Devin/Windsurf CLI) always emit text/event-stream, signalling + // failure with an error-only chunk (`data: {"error":{"message":"Devin CLI not found..."}}`) + // that carries no `choices`. Surface that real, sanitized message instead of the generic 502 + // so the actionable error is not swallowed (#3324). + const surfacedSseError = extractSSEErrorMessage(streamPayload); + const invalidSseMessage = + surfacedSseError || "Invalid SSE response for non-streaming request"; + return { + kind: "invalid_sse", + message: invalidSseMessage, + looksLikeSSE: true, + normalizedProviderPayload, + }; + } + + return { + kind: "ok", + responseBody: parsedFromSSE.body, + responsePayloadFormat: parsedFromSSE.format, + looksLikeSSE: true, + normalizedProviderPayload, + }; + } + + try { + const responseBody = rawBody ? JSON.parse(rawBody) : {}; + return { + kind: "ok", + responseBody, + responsePayloadFormat: targetFormat, + looksLikeSSE: false, + normalizedProviderPayload, + }; + } catch (err) { + const detailedError = `Invalid JSON response from provider (error: ${err instanceof Error ? err.message : String(err)}): ${rawBody.substring(0, 1000)}`; + const invalidJsonMessage = "Invalid JSON response from provider"; + return { + kind: "invalid_json", + message: invalidJsonMessage, + detailedError, + looksLikeSSE: false, + normalizedProviderPayload, + }; + } +} diff --git a/open-sse/handlers/chatCore/nonStreamingUsageStats.ts b/open-sse/handlers/chatCore/nonStreamingUsageStats.ts new file mode 100644 index 0000000000..d7514e5d44 --- /dev/null +++ b/open-sse/handlers/chatCore/nonStreamingUsageStats.ts @@ -0,0 +1,88 @@ +/** + * chatCore non-streaming usage-stats persistence (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501 — response-handling slice of executeProviderRequest). + * + * Extracted from handleChatCore's non-streaming success path: records per-request usage analytics + * for a successful non-streaming response — an optional trace console line, the fire-and-forget + * `saveRequestUsage` row, and the per-api-key billable-token counter. Side-effect only (no handler + * state is mutated, nothing is returned); best-effort, every write swallows its own errors. The + * per-request context is threaded via `ctx` so the call site stays byte-identical; behaviour is + * unchanged. + */ + +import { saveRequestUsage } from "@/lib/usageDb"; +import { formatUsageLog } from "@/lib/usage/tokenAccounting"; +import { COLORS } from "../../utils/stream.ts"; +import { recordTokenUsage } from "../../services/tokenLimitCounter.ts"; +import { computeBillableTokens } from "./upstreamTimeouts.ts"; +import { type EffectiveServiceTier } from "./serviceTier.ts"; + +export type RecordNonStreamingUsageStatsContext = { + traceEnabled: boolean; + provider: string | null | undefined; + connectionId: string | null | undefined; + model: string | null | undefined; + startTime: number; + apiKeyInfo: { id?: string | null; name?: string | null } | null | undefined; + effectiveServiceTier: EffectiveServiceTier; + isCombo: boolean; + comboStrategy: string | null | undefined; +}; + +function logUsageTrace( + usage: object, + provider: string | null | undefined, + connectionId: string | null | undefined +): void { + const msg = `[${new Date().toLocaleTimeString("en-US", { hour12: false, hour: "2-digit", minute: "2-digit" })}] 📊 [USAGE] ${provider?.toUpperCase()} | ${formatUsageLog(usage)}${connectionId ? ` | account=${connectionId.slice(0, 8)}...` : ""}`; + console.log(`${COLORS.green}${msg}${COLORS.reset}`); +} + +function persistUsageRow(usage: object, ctx: RecordNonStreamingUsageStatsContext): void { + const { provider, connectionId, model, startTime, apiKeyInfo, effectiveServiceTier } = ctx; + saveRequestUsage({ + provider: provider || "unknown", + model: model || "unknown", + tokens: usage, + status: "200", + success: true, + latencyMs: Date.now() - startTime, + timeToFirstTokenMs: Date.now() - startTime, + errorCode: null, + timestamp: new Date().toISOString(), + connectionId: connectionId || undefined, + apiKeyId: apiKeyInfo?.id || undefined, + apiKeyName: apiKeyInfo?.name || undefined, + serviceTier: effectiveServiceTier, + comboStrategy: ctx.isCombo ? ctx.comboStrategy || undefined : undefined, + }).catch((err) => { + console.error("Failed to save usage stats:", err.message); + }); +} + +function recordBillableTokens( + usage: object, + apiKeyInfo: RecordNonStreamingUsageStatsContext["apiKeyInfo"], + provider: string | null | undefined, + model: string | null | undefined +): void { + if (!apiKeyInfo?.id) return; + try { + const billable = computeBillableTokens(usage); + if (billable > 0) + recordTokenUsage(apiKeyInfo.id, provider || "unknown", model || "unknown", billable); + } catch { + // never block the response on counter recording + } +} + +export function recordNonStreamingUsageStats( + usage: unknown, + ctx: RecordNonStreamingUsageStatsContext +): void { + if (!usage || typeof usage !== "object") return; + + if (ctx.traceEnabled) logUsageTrace(usage, ctx.provider, ctx.connectionId); + persistUsageRow(usage, ctx); + recordBillableTokens(usage, ctx.apiKeyInfo, ctx.provider, ctx.model); +} diff --git a/tests/unit/chatcore-non-streaming-response-parse.test.ts b/tests/unit/chatcore-non-streaming-response-parse.test.ts new file mode 100644 index 0000000000..21b761fa04 --- /dev/null +++ b/tests/unit/chatcore-non-streaming-response-parse.test.ts @@ -0,0 +1,157 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { parseNonStreamingResponseBody } from "@omniroute/open-sse/handlers/chatCore/nonStreamingResponseParse.ts"; + +// Minimal Response stub: only the surface parseNonStreamingResponseBody touches +// (headers.get + text()). upstreamStream is passed false so readNonStreamingResponseBody +// always takes the buffered response.text() path. +function makeResponse(body: string, contentType: string): Response { + return { + headers: { + get: (name: string) => + name.toLowerCase() === "content-type" ? contentType : null, + }, + text: async () => body, + body: null, + } as unknown as Response; +} + +const baseOpts = { + upstreamStream: false, + providerHeaders: null, + finalBody: null, + targetFormat: "openai", + model: "gpt-4o-mini", +}; + +test("valid JSON body → ok with parsed object and targetFormat", async () => { + const payload = { id: "x", choices: [{ message: { content: "hi" } }] }; + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse(JSON.stringify(payload), "application/json"), + }); + assert.equal(res.kind, "ok"); + if (res.kind !== "ok") return; + assert.deepEqual(res.responseBody, payload); + assert.equal(res.responsePayloadFormat, "openai"); + assert.equal(res.looksLikeSSE, false); +}); + +test("empty body (non-SSE) → ok with empty object", async () => { + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse("", "application/json"), + }); + assert.equal(res.kind, "ok"); + if (res.kind !== "ok") return; + assert.deepEqual(res.responseBody, {}); + assert.equal(res.looksLikeSSE, false); +}); + +test("invalid JSON → invalid_json with short message + detailed error", async () => { + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse("{not json", "application/json"), + }); + assert.equal(res.kind, "invalid_json"); + if (res.kind !== "invalid_json") return; + assert.equal(res.message, "Invalid JSON response from provider"); + assert.match(res.detailedError, /^Invalid JSON response from provider \(error: /); + assert.match(res.detailedError, /\{not json/); + assert.equal(res.looksLikeSSE, false); +}); + +test("valid SSE payload (by content-type) → ok with SSE-derived format", async () => { + const sse = + 'data: {"id":"c1","object":"chat.completion.chunk","choices":[{"delta":{"content":"hello"},"index":0,"finish_reason":null}]}\n\n' + + 'data: {"id":"c1","object":"chat.completion.chunk","choices":[{"delta":{},"index":0,"finish_reason":"stop"}]}\n\n' + + "data: [DONE]\n\n"; + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse(sse, "text/event-stream"), + }); + assert.equal(res.kind, "ok"); + if (res.kind !== "ok") return; + assert.equal(res.looksLikeSSE, true); + assert.ok(res.responseBody && typeof res.responseBody === "object"); + assert.equal(typeof res.responsePayloadFormat, "string"); +}); + +test("SSE detected by body heuristic even with non-stream content-type", async () => { + const sse = + 'data: {"id":"c1","object":"chat.completion.chunk","choices":[{"delta":{"content":"hi"},"index":0,"finish_reason":"stop"}]}\n\n' + + "data: [DONE]\n\n"; + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse(sse, "text/plain"), + }); + assert.equal(res.looksLikeSSE, true); +}); + +test("error-only SSE (no choices) → invalid_sse surfacing the upstream error (#3324)", async () => { + const sse = 'data: {"error":{"message":"Devin CLI not found in PATH"}}\n\n'; + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse(sse, "text/event-stream"), + }); + assert.equal(res.kind, "invalid_sse"); + if (res.kind !== "invalid_sse") return; + assert.equal(res.message, "Devin CLI not found in PATH"); + assert.equal(res.looksLikeSSE, true); +}); + +test("unparseable SSE with no embedded error → generic invalid_sse message", async () => { + const sse = "data: not-json-at-all\n\n"; + const res = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse(sse, "text/event-stream"), + }); + assert.equal(res.kind, "invalid_sse"); + if (res.kind !== "invalid_sse") return; + assert.equal(res.message, "Invalid SSE response for non-streaming request"); +}); + +test("normalizedProviderPayload is present on every branch", async () => { + const ok = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse('{"a":1}', "application/json"), + }); + assert.ok("normalizedProviderPayload" in ok); + const bad = await parseNonStreamingResponseBody({ + ...baseOpts, + providerResponse: makeResponse("{bad", "application/json"), + }); + assert.ok("normalizedProviderPayload" in bad); +}); + +test("buffering log fires debug when stream was expected, warn otherwise", async () => { + const sse = + 'data: {"id":"c1","object":"chat.completion.chunk","choices":[{"delta":{"content":"hi"},"index":0,"finish_reason":"stop"}]}\n\n' + + "data: [DONE]\n\n"; + const debugCalls: string[] = []; + const warnCalls: string[] = []; + const log = { + debug: (_tag: string, msg: string) => debugCalls.push(msg), + warn: (_tag: string, msg: string) => warnCalls.push(msg), + }; + + // upstreamStream=true → expected → debug path + await parseNonStreamingResponseBody({ + ...baseOpts, + upstreamStream: true, + providerResponse: makeResponse(sse, "text/event-stream"), + log, + }); + assert.equal(debugCalls.length, 1); + assert.equal(warnCalls.length, 0); + + // upstreamStream=false + no accept/stream hints → unexpected → warn path + await parseNonStreamingResponseBody({ + ...baseOpts, + upstreamStream: false, + providerResponse: makeResponse(sse, "text/event-stream"), + log, + }); + assert.equal(warnCalls.length, 1); +}); diff --git a/tests/unit/chatcore-non-streaming-usage-stats.test.ts b/tests/unit/chatcore-non-streaming-usage-stats.test.ts new file mode 100644 index 0000000000..0065a83a75 --- /dev/null +++ b/tests/unit/chatcore-non-streaming-usage-stats.test.ts @@ -0,0 +1,129 @@ +// Characterization of recordNonStreamingUsageStats — the per-request usage-stats persistence +// extracted from handleChatCore's non-streaming success path (chatCore god-file decomposition, +// #3501). Uses a real temp DB and polls usage_history (saveRequestUsage is async + +// fire-and-forget). Locks: the non-object usage guard (no-op), the field mapping +// (provider/model/connection/api-key/tokens/serviceTier), and the trace-log line format. +import { test, before, after } from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const testDataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omni-nonstream-usage-test-")); +process.env.DATA_DIR = testDataDir; + +const coreDb = await import("../../src/lib/db/core.ts"); +const { getUsageHistory } = await import("../../src/lib/usage/usageHistory.ts"); +const { recordNonStreamingUsageStats } = await import( + "../../open-sse/handlers/chatCore/nonStreamingUsageStats.ts" +); + +function baseCtx(overrides: Record = {}) { + return { + traceEnabled: false, + provider: "openai", + connectionId: "conn-12345678abc", + model: "gpt-x", + startTime: Date.now() - 50, + apiKeyInfo: { id: "key-1", name: "Key One" }, + effectiveServiceTier: "standard", + isCombo: false, + comboStrategy: null, + ...overrides, + } as Parameters[1]; +} + +async function rowsFor(provider: string): Promise>> { + return (await getUsageHistory({ provider })) as Array>; +} + +async function waitForRows(provider: string, min: number, timeoutMs = 3000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const rows = await rowsFor(provider); + if (rows.length >= min) return rows; + await new Promise((r) => setTimeout(r, 25)); + } + return rowsFor(provider); +} + +before(async () => { + await coreDb.ensureDbInitialized(); +}); + +after(() => { + coreDb.resetDbInstance(); + try { + fs.rmSync(testDataDir, { recursive: true, force: true }); + } catch { + // best-effort cleanup + } +}); + +test("non-object usage is a no-op (null, undefined, number, string)", async () => { + const before = (await rowsFor("guard-prov")).length; + for (const bad of [null, undefined, 42, "usage", true]) { + recordNonStreamingUsageStats(bad, baseCtx({ provider: "guard-prov" })); + } + // give any (erroneous) async write a chance to land, then assert nothing was persisted + await new Promise((r) => setTimeout(r, 100)); + const after = (await rowsFor("guard-prov")).length; + assert.equal(after, before); +}); + +test("valid usage persists a row with the mapped fields", async () => { + recordNonStreamingUsageStats( + { prompt_tokens: 11, completion_tokens: 7 }, + baseCtx({ provider: "map-prov", model: "gpt-map", connectionId: "conn-abc", isCombo: false }) + ); + const rows = await waitForRows("map-prov", 1); + assert.equal(rows.length, 1); + const row = rows[0] as { + provider: string; + model: string; + connectionId: string; + apiKeyId: string; + apiKeyName: string; + success: boolean; + status: string; + tokens: { input: number; output: number }; + }; + assert.equal(row.provider, "map-prov"); + assert.equal(row.model, "gpt-map"); + assert.equal(row.connectionId, "conn-abc"); + assert.equal(row.apiKeyId, "key-1"); + assert.equal(row.apiKeyName, "Key One"); + assert.equal(row.success, true); + assert.equal(row.status, "200"); + assert.equal(row.tokens.input, 11); + assert.equal(row.tokens.output, 7); +}); + +test("falls back to 'unknown' provider/model when absent", async () => { + recordNonStreamingUsageStats( + { prompt_tokens: 1, completion_tokens: 1 }, + baseCtx({ provider: null, model: null, apiKeyInfo: null }) + ); + const rows = await waitForRows("unknown", 1); + const mine = rows.find((r) => (r as { model?: string }).model === "unknown"); + assert.ok(mine, "expected a row with provider/model 'unknown'"); +}); + +test("trace log emits a [USAGE] line with the upper-cased provider when traceEnabled", () => { + const original = console.log; + const captured: string[] = []; + console.log = (...args: unknown[]) => { + captured.push(args.map(String).join(" ")); + }; + try { + recordNonStreamingUsageStats( + { prompt_tokens: 3, completion_tokens: 2 }, + baseCtx({ provider: "trace-prov", traceEnabled: true }) + ); + } finally { + console.log = original; + } + const usageLine = captured.find((l) => l.includes("[USAGE]")); + assert.ok(usageLine, "expected a [USAGE] trace line"); + assert.match(usageLine as string, /TRACE-PROV/); +});