refactor(chatCore): extrai parse + usage-stats não-streaming do executeProviderRequest (#3501) (#4762)

chatCore #3501: extract parseNonStreamingResponseBody + recordNonStreamingUsageStats. Integrated into release/v3.8.35.

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)

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<string, unknown> | Headers | null | undefined;
+  finalBody: unknown;
+  targetFormat: string;
+  model: string;
+  log?: LoggerLike;
+}): Promise<NonStreamingParseResult> {
+  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,
+    };
+  }
+}

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);
+}

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);
+});

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<string, unknown> = {}) {
+  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<typeof recordNonStreamingUsageStats>[1];
+}
+
+async function rowsFor(provider: string): Promise<Array<Record<string, unknown>>> {
+  return (await getUsageHistory({ provider })) as Array<Record<string, unknown>>;
+}
+
+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/);
+});