refactor(chatCore): extrai recordStreamingUsageStats (analytics de usage streaming, #3501)

open-sse/handlers/chatCore.ts

@@ -172,6 +172,7 @@ import { scheduleQuotaShareConsumption } from "./chatCore/quotaShareConsumption.
 import { emitRequestGamificationEvent } from "./chatCore/gamificationEvent.ts";
 import { runPluginOnResponseHook } from "./chatCore/pluginOnResponse.ts";
 import { scheduleStreamingQuotaShareConsumption } from "./chatCore/streamingQuotaShare.ts";
+import { recordStreamingUsageStats } from "./chatCore/streamingUsageStats.ts";
 import {
   appendNonStreamingSseTerminalSignal,
   type NonStreamingSseTerminalState,
@@ -3974,37 +3975,20 @@ export async function handleChatCore({
     // Track cache token metrics for streaming responses
     if (streamUsage && typeof streamUsage === "object") {
       attachCompressionUsageReceiptAfterAnalytics(streamUsage as Record<string, unknown>, "stream");
-
-      saveRequestUsage({
-        provider: provider || "unknown",
-        model: model || "unknown",
-        tokens: streamUsage,
-        status: String(normalizedStreamStatus),
-        success: normalizedStreamStatus === 200,
-        latencyMs: Date.now() - startTime,
-        timeToFirstTokenMs: ttft,
-        errorCode:
-          normalizedStreamStatus === 200 ? null : streamErrorCode || String(normalizedStreamStatus),
-        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 && normalizedStreamStatus === 200) {
-        try {
-          const billable = computeBillableTokens(streamUsage);
-          if (billable > 0)
-            recordTokenUsage(apiKeyInfo.id, provider || "unknown", model || "unknown", billable);
-        } catch {
-          // never block the stream on counter recording
-        }
-      }
     }
+    recordStreamingUsageStats(streamUsage, {
+      provider,
+      model,
+      streamStatus: normalizedStreamStatus,
+      startTime,
+      ttft,
+      streamErrorCode,
+      connectionId,
+      apiKeyInfo,
+      effectiveServiceTier,
+      isCombo,
+      comboStrategy,
+    });
 
     persistAttemptLogs({
       status: normalizedStreamStatus,

open-sse/handlers/chatCore/streamingUsageStats.ts

@@ -0,0 +1,77 @@
+/**
+ * chatCore streaming usage-stats persistence (Quality Gate v2 / Fase 9 — chatCore god-file
+ * decomposition, #3501).
+ *
+ * Extracted from handleChatCore's onStreamComplete: records per-request usage analytics for a
+ * completed streaming response — the fire-and-forget `saveRequestUsage` row and the per-api-key
+ * billable-token counter (only on a 200 stream). 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. The
+ * compression usage-receipt attach stays in the handler (it is a handler-bound closure).
+ */
+
+import { saveRequestUsage } from "@/lib/usageDb";
+import { recordTokenUsage } from "../../services/tokenLimitCounter.ts";
+import { computeBillableTokens } from "./upstreamTimeouts.ts";
+import { type EffectiveServiceTier } from "./serviceTier.ts";
+
+export type RecordStreamingUsageStatsContext = {
+  provider: string | null | undefined;
+  model: string | null | undefined;
+  streamStatus: number;
+  startTime: number;
+  ttft: number;
+  streamErrorCode: string | null | undefined;
+  connectionId: string | null | undefined;
+  apiKeyInfo: { id?: string | null; name?: string | null } | null | undefined;
+  effectiveServiceTier: EffectiveServiceTier;
+  isCombo: boolean;
+  comboStrategy: string | null | undefined;
+};
+
+function persistStreamingUsageRow(usage: object, ctx: RecordStreamingUsageStatsContext): void {
+  const { provider, model, streamStatus, startTime, ttft, streamErrorCode } = ctx;
+  saveRequestUsage({
+    provider: provider || "unknown",
+    model: model || "unknown",
+    tokens: usage,
+    status: String(streamStatus),
+    success: streamStatus === 200,
+    latencyMs: Date.now() - startTime,
+    timeToFirstTokenMs: ttft,
+    errorCode: streamStatus === 200 ? null : streamErrorCode || String(streamStatus),
+    timestamp: new Date().toISOString(),
+    connectionId: ctx.connectionId || undefined,
+    apiKeyId: ctx.apiKeyInfo?.id || undefined,
+    apiKeyName: ctx.apiKeyInfo?.name || undefined,
+    serviceTier: ctx.effectiveServiceTier,
+    comboStrategy: ctx.isCombo ? ctx.comboStrategy || undefined : undefined,
+  }).catch((err) => {
+    console.error("Failed to save usage stats:", err.message);
+  });
+}
+
+function recordStreamingBillableTokens(usage: object, ctx: RecordStreamingUsageStatsContext): void {
+  if (!ctx.apiKeyInfo?.id || ctx.streamStatus !== 200) return;
+  try {
+    const billable = computeBillableTokens(usage);
+    if (billable > 0)
+      recordTokenUsage(
+        ctx.apiKeyInfo.id,
+        ctx.provider || "unknown",
+        ctx.model || "unknown",
+        billable
+      );
+  } catch {
+    // never block the stream on counter recording
+  }
+}
+
+export function recordStreamingUsageStats(
+  usage: unknown,
+  ctx: RecordStreamingUsageStatsContext
+): void {
+  if (!usage || typeof usage !== "object") return;
+  persistStreamingUsageRow(usage, ctx);
+  recordStreamingBillableTokens(usage, ctx);
+}

tests/unit/chatcore-streaming-usage-stats.test.ts

@@ -0,0 +1,104 @@
+// Characterization of recordStreamingUsageStats — the per-request usage-stats persistence extracted
+// from handleChatCore's onStreamComplete (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, the streaming field mapping (status/success/ttft/errorCode), and a non-200 stream.
+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-stream-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 { recordStreamingUsageStats } = await import(
+  "../../open-sse/handlers/chatCore/streamingUsageStats.ts"
+);
+
+function baseCtx(overrides: Record<string, unknown> = {}) {
+  return {
+    provider: "openai",
+    model: "gpt-x",
+    streamStatus: 200,
+    startTime: Date.now() - 50,
+    ttft: 12,
+    streamErrorCode: null,
+    connectionId: "conn-1",
+    apiKeyInfo: { id: "key-1", name: "Key One" },
+    effectiveServiceTier: "standard",
+    isCombo: false,
+    comboStrategy: null,
+    ...overrides,
+  } as Parameters<typeof recordStreamingUsageStats>[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", async () => {
+  const before = (await rowsFor("guard-stream")).length;
+  for (const bad of [null, undefined, 7, "x"]) {
+    recordStreamingUsageStats(bad, baseCtx({ provider: "guard-stream" }));
+  }
+  await new Promise((r) => setTimeout(r, 100));
+  assert.equal((await rowsFor("guard-stream")).length, before);
+});
+
+test("200 stream persists a success row with ttft and status 200", async () => {
+  recordStreamingUsageStats(
+    { prompt_tokens: 20, completion_tokens: 9 },
+    baseCtx({ provider: "smap-prov", model: "gpt-stream", streamStatus: 200 })
+  );
+  const rows = await waitForRows("smap-prov", 1);
+  const row = rows[0] as {
+    model: string;
+    success: boolean;
+    status: string;
+    timeToFirstTokenMs: number;
+    tokens: { input: number; output: number };
+  };
+  assert.equal(row.model, "gpt-stream");
+  assert.equal(row.success, true);
+  assert.equal(row.status, "200");
+  assert.equal(row.timeToFirstTokenMs, 12);
+  assert.equal(row.tokens.input, 20);
+  assert.equal(row.tokens.output, 9);
+});
+
+test("non-200 stream persists a failure row (success=false, status string)", async () => {
+  recordStreamingUsageStats(
+    { prompt_tokens: 1, completion_tokens: 0 },
+    baseCtx({ provider: "sfail-prov", streamStatus: 503, streamErrorCode: "upstream_5xx" })
+  );
+  const rows = await waitForRows("sfail-prov", 1);
+  const row = rows[0] as { success: boolean; status: string; errorCode: string };
+  assert.equal(row.success, false);
+  assert.equal(row.status, "503");
+  assert.equal(row.errorCode, "upstream_5xx");
+});