refactor(chatCore): extrai stageTrace + compressionUsageReceipt para leaves (#3501)

open-sse/handlers/chatCore.ts

@@ -143,6 +143,8 @@ import {
   persistAttemptLogs as persistAttemptLogsFor,
   type PersistAttemptLogsArgs,
 } from "./chatCore/attemptLogging.ts";
+import { stageTrace } from "./chatCore/stageTrace.ts";
+import { attachCompressionUsageReceiptAfterAnalytics as attachCompressionUsageReceiptAfterAnalyticsFor } from "./chatCore/compressionUsageReceipt.ts";
 
 import {
   getCallLogPipelineCaptureStreamChunks,
@@ -368,19 +370,10 @@ export async function handleChatCore({
     });
   });
   const traceEnabled = process.env.OMNIROUTE_TRACE === "true" || process.env.DEBUG === "true";
-  const trace = (label: string, extra?: Record<string, unknown>) => {
-    if (!traceEnabled) return;
-    const elapsed = Date.now() - startTime;
-    let suffix = "";
-    if (extra) {
-      try {
-        suffix = ` ${JSON.stringify(extra)}`;
-      } catch {
-        suffix = " [unserializable]";
-      }
-    }
-    log?.info?.("STAGE_TRACE", `${traceId} ${label} t=${elapsed}ms${suffix}`);
-  };
+  // Stage trace extracted to chatCore/stageTrace.ts (#3501); bind the per-request inputs once so the
+  // call sites stay byte-identical.
+  const trace = (label: string, extra?: Record<string, unknown>) =>
+    stageTrace(label, extra, { traceEnabled, startTime, traceId, log });
   let tokensCompressed: number | null = null;
   body = injectSystemPrompt(body);
   // ── Plugin onRequest hook ──
@@ -683,22 +676,16 @@ export async function handleChatCore({
     detailedLoggingEnabled && getCallLogPipelineCaptureStreamChunks();
   const skillRequestId = generateRequestId();
   let compressionAnalyticsWritePromise: Promise<void> | null = null;
+  // Compression usage-receipt attachment extracted to chatCore/compressionUsageReceipt.ts (#3501);
+  // pass the in-flight analytics write + request id so behaviour stays byte-identical.
   const attachCompressionUsageReceiptAfterAnalytics = (
     usage: Record<string, unknown>,
     source: "provider" | "estimated" | "stream"
-  ) => {
-    const pendingWrite = compressionAnalyticsWritePromise;
-    void (async () => {
-      try {
-        if (pendingWrite) await pendingWrite;
-        const { attachCompressionUsageReceipt } =
-          await import("../../src/lib/db/compressionAnalytics.ts");
-        attachCompressionUsageReceipt(skillRequestId, usage, source);
-      } catch {
-        // Compression analytics are best-effort and must never affect responses.
-      }
-    })();
-  };
+  ) =>
+    attachCompressionUsageReceiptAfterAnalyticsFor(usage, source, {
+      pendingWrite: compressionAnalyticsWritePromise,
+      skillRequestId,
+    });
   const pipelineSessionId =
     (clientRawRequest?.headers && typeof clientRawRequest.headers.get === "function"
       ? clientRawRequest.headers.get("x-omniroute-session-id")

open-sse/handlers/chatCore/compressionUsageReceipt.ts

@@ -0,0 +1,27 @@
+/**
+ * chatCore compression-usage receipt attachment (Quality Gate v2 / Fase 9 — chatCore god-file
+ * decomposition, #3501).
+ *
+ * Extracted from handleChatCore: best-effort, fire-and-forget attachment of a usage receipt to the
+ * stored compression-analytics row, ordered after any in-flight analytics write (`pendingWrite`) so
+ * the receipt lands on the persisted row. Errors are swallowed — analytics must never affect the
+ * response. The per-request inputs are passed via `ctx`; behaviour is byte-identical to the previous
+ * inline closure.
+ */
+
+export function attachCompressionUsageReceiptAfterAnalytics(
+  usage: Record<string, unknown>,
+  source: "provider" | "estimated" | "stream",
+  ctx: { pendingWrite: Promise<void> | null; skillRequestId: string }
+) {
+  const { pendingWrite, skillRequestId } = ctx;
+  void (async () => {
+    try {
+      if (pendingWrite) await pendingWrite;
+      const { attachCompressionUsageReceipt } = await import("@/lib/db/compressionAnalytics.ts");
+      attachCompressionUsageReceipt(skillRequestId, usage, source);
+    } catch {
+      // Compression analytics are best-effort and must never affect responses.
+    }
+  })();
+}

open-sse/handlers/chatCore/stageTrace.ts

@@ -0,0 +1,30 @@
+/**
+ * chatCore per-request stage trace (Quality Gate v2 / Fase 9 — chatCore god-file decomposition,
+ * #3501).
+ *
+ * Extracted from handleChatCore: emits a `[STAGE_TRACE]` checkpoint log (trace id + elapsed ms +
+ * optional serialized extra) when tracing is enabled, so a hung request reveals which await it was
+ * sitting on. Per-request inputs (enable flag, start time, trace id, logger) are threaded via `ctx`
+ * so the call sites stay byte-identical; behaviour is unchanged.
+ */
+
+type LoggerLike = { info?: (...args: unknown[]) => void } | null | undefined;
+
+export function stageTrace(
+  label: string,
+  extra: Record<string, unknown> | undefined,
+  ctx: { traceEnabled: boolean; startTime: number; traceId: string; log: LoggerLike }
+) {
+  const { traceEnabled, startTime, traceId, log } = ctx;
+  if (!traceEnabled) return;
+  const elapsed = Date.now() - startTime;
+  let suffix = "";
+  if (extra) {
+    try {
+      suffix = ` ${JSON.stringify(extra)}`;
+    } catch {
+      suffix = " [unserializable]";
+    }
+  }
+  log?.info?.("STAGE_TRACE", `${traceId} ${label} t=${elapsed}ms${suffix}`);
+}

tests/unit/chatcore-compression-usage-receipt.test.ts

@@ -0,0 +1,78 @@
+// tests/unit/chatcore-compression-usage-receipt.test.ts
+// Characterization of attachCompressionUsageReceiptAfterAnalytics — the fire-and-forget compression
+// usage-receipt attachment extracted from handleChatCore (chatCore god-file decomposition, #3501).
+// Uses a real temp DB: inserts a compression_analytics row, attaches a receipt after pendingWrite,
+// and asserts the aggregated realUsage. Also locks the best-effort error swallowing.
+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-compression-receipt-test-"));
+process.env.DATA_DIR = testDataDir;
+
+const coreDb = await import("../../src/lib/db/core.ts");
+const { insertCompressionAnalyticsRow, getCompressionAnalyticsSummary } = await import(
+  "../../src/lib/db/compressionAnalytics.ts"
+);
+const { attachCompressionUsageReceiptAfterAnalytics } = await import(
+  "../../open-sse/handlers/chatCore/compressionUsageReceipt.ts"
+);
+
+const tick = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+before(async () => {
+  await coreDb.ensureDbInitialized();
+});
+
+after(() => {
+  coreDb.resetDbInstance();
+  fs.rmSync(testDataDir, { recursive: true, force: true });
+});
+
+test("attaches the usage receipt only after pendingWrite resolves", async () => {
+  insertCompressionAnalyticsRow({
+    timestamp: new Date().toISOString(),
+    mode: "test",
+    original_tokens: 100,
+    compressed_tokens: 60,
+    tokens_saved: 40,
+    request_id: "req-1",
+  });
+
+  let pendingResolved = false;
+  const pendingWrite = new Promise<void>((res) =>
+    setTimeout(() => {
+      pendingResolved = true;
+      res();
+    }, 20)
+  );
+
+  attachCompressionUsageReceiptAfterAnalytics(
+    { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 },
+    "provider",
+    { pendingWrite, skillRequestId: "req-1" }
+  );
+
+  await tick(80);
+  assert.equal(pendingResolved, true, "pendingWrite should have resolved first");
+
+  const summary = getCompressionAnalyticsSummary();
+  assert.equal(summary.realUsage.requestsWithReceipts, 1);
+  assert.equal(summary.realUsage.promptTokens, 10);
+  assert.equal(summary.realUsage.completionTokens, 5);
+});
+
+test("swallows the no-matching-row case without throwing or recording a receipt", async () => {
+  assert.doesNotThrow(() =>
+    attachCompressionUsageReceiptAfterAnalytics(
+      { prompt_tokens: 1, total_tokens: 1 },
+      "provider",
+      { pendingWrite: null, skillRequestId: "does-not-exist" }
+    )
+  );
+  await tick(40);
+  const summary = getCompressionAnalyticsSummary();
+  assert.equal(summary.realUsage.requestsWithReceipts, 1);
+});

tests/unit/chatcore-stage-trace.test.ts

@@ -0,0 +1,50 @@
+// tests/unit/chatcore-stage-trace.test.ts
+// Characterization of stageTrace — the per-request [STAGE_TRACE] checkpoint logger extracted from
+// handleChatCore (chatCore god-file decomposition, #3501). Locks: the disabled no-op, the
+// `${traceId} ${label} t=${elapsed}ms` format, serialized extra, and the [unserializable] fallback.
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { stageTrace } from "../../open-sse/handlers/chatCore/stageTrace.ts";
+
+function makeLog() {
+  const calls: Array<[string, string]> = [];
+  return { calls, log: { info: (tag: string, msg: string) => calls.push([tag, msg]) } };
+}
+
+test("is a no-op when tracing is disabled", () => {
+  const { calls, log } = makeLog();
+  stageTrace("post_translation", undefined, { traceEnabled: false, startTime: 0, traceId: "abc", log });
+  assert.equal(calls.length, 0);
+});
+
+test("emits a STAGE_TRACE line with trace id, label and elapsed ms", () => {
+  const { calls, log } = makeLog();
+  stageTrace("post_executor", undefined, {
+    traceEnabled: true,
+    startTime: Date.now() - 5,
+    traceId: "abc123",
+    log,
+  });
+  assert.equal(calls.length, 1);
+  assert.equal(calls[0][0], "STAGE_TRACE");
+  assert.match(calls[0][1], /^abc123 post_executor t=\d+ms$/);
+});
+
+test("appends serialized extra context", () => {
+  const { calls, log } = makeLog();
+  stageTrace("pre_executor", { attempt: 2 }, {
+    traceEnabled: true,
+    startTime: Date.now(),
+    traceId: "id",
+    log,
+  });
+  assert.match(calls[0][1], /\{"attempt":2\}$/);
+});
+
+test("falls back to [unserializable] for circular extra", () => {
+  const { calls, log } = makeLog();
+  const circular: Record<string, unknown> = {};
+  circular.self = circular;
+  stageTrace("x", circular, { traceEnabled: true, startTime: Date.now(), traceId: "id", log });
+  assert.match(calls[0][1], /\[unserializable\]$/);
+});