Fix Codex Responses compression analytics (#7273)

* fix compression analytics for Codex responses

* preserve Responses tool output fields

* Test compression analytics cost failure isolation

open-sse/handlers/chatCore.ts

@@ -1445,6 +1445,7 @@ export async function handleChatCore({
               cavemanOutputModeIntensity,
               log,
             });
+            await compressionAnalyticsWritePromise;
           } else {
             // Compression was attempted (mode active, engines ran) but produced no
             // recordable saving — e.g. a Stacked RTK→Caveman pipeline on already-compact
@@ -1467,6 +1468,7 @@ export async function handleChatCore({
               },
               "no_savings"
             );
+            await compressionAnalyticsWritePromise;
           }
 
           if (result.compressed) {
@@ -1497,6 +1499,7 @@ export async function handleChatCore({
           cavemanOutputModeIntensity,
           log,
         });
+        await compressionAnalyticsWritePromise;
       }
       emitOutputStyleTelemetry({
         outputStyleResult,

open-sse/handlers/chatCore/cavemanOutputAnalytics.ts

@@ -5,11 +5,11 @@
  * Extracted from handleChatCore's request-setup compression path: when only the caveman output
  * mode was applied (no upstream compression run recorded a row), persist a single analytics row so
  * output-caveman runs still surface in compression analytics. Best-effort — returns the write
- * promise (the caller assigns it to compressionAnalyticsWritePromise) and swallows its own errors.
- * Behaviour is byte-identical to the previous inline block.
+ * promise so the caller can finish persistence before dispatch; errors remain non-fatal but are
+ * logged at warning level.
  */
 
-type LoggerLike = { debug?: (...args: unknown[]) => void } | null | undefined;
+type LoggerLike = { warn?: (...args: unknown[]) => void } | null | undefined;
 
 export function writeCavemanOutputAnalytics(args: {
   comboName: string | null | undefined;
@@ -37,7 +37,7 @@ export function writeCavemanOutputAnalytics(args: {
         output_mode: args.cavemanOutputModeIntensity,
       });
     } catch (err) {
-      args.log?.debug?.(
+      args.log?.warn?.(
         "COMPRESSION",
         "Caveman output analytics write skipped: " +
           (err instanceof Error ? err.message : String(err))

open-sse/handlers/chatCore/compressionAnalyticsWrite.ts

@@ -4,15 +4,21 @@
  *
  * Extracted from handleChatCore's request-setup compression path: persist the per-run compression
  * analytics row (cost saved, RTK raw-output pointers) plus the per-engine breakdown of a stacked
- * run. Returns the write promise (the caller assigns it to compressionAnalyticsWritePromise) and
- * swallows its own errors — best-effort, off the hot path, never throws into a request. Behaviour
- * is byte-identical to the previous inline block. Split into small builders so each stays under the
- * complexity cap.
+ * run. Returns the write promise so the caller can finish persistence before dispatching the
+ * upstream request. Errors remain best-effort and never throw into a request, but they are logged
+ * at warning level so a broken analytics path is observable. Split into small builders so each
+ * stays under the complexity cap.
  */
 
 import { type CompressionStats } from "../../services/compression/stats.ts";
 
-type LoggerLike = { debug?: (...args: unknown[]) => void } | null | undefined;
+type LoggerLike =
+  | {
+      debug?: (...args: unknown[]) => void;
+      warn?: (...args: unknown[]) => void;
+    }
+  | null
+  | undefined;
 
 type WriteOpts = {
   stats: CompressionStats;
@@ -30,6 +36,12 @@ type WriteOpts = {
 
 type RtkPointer = { id?: string | null; bytes?: number | null };
 
+type CalculateCost = typeof import("@/lib/usage/costCalculator").calculateCost;
+
+type WriteDependencies = {
+  calculateCost?: CalculateCost;
+};
+
 function buildRtkPointerFields(rtkPointers: RtkPointer[]) {
   return {
     rtk_raw_output_pointer: rtkPointers[0]?.id ?? null,
@@ -109,7 +121,7 @@ export function writeCompressionSkip(opts: WriteOpts, skipReason: string): Promi
         skip_reason: skipReason,
       });
     } catch (err) {
-      opts.log?.debug?.(
+      opts.log?.warn?.(
         "COMPRESSION",
         "Compression skip-analytics write skipped: " +
           (err instanceof Error ? err.message : String(err))
@@ -118,29 +130,42 @@ export function writeCompressionSkip(opts: WriteOpts, skipReason: string): Promi
   })();
 }
 
-export function writeCompressionAnalytics(opts: WriteOpts): Promise<void> {
+export function writeCompressionAnalytics(
+  opts: WriteOpts,
+  dependencies: WriteDependencies = {}
+): Promise<void> {
   return (async () => {
     try {
-      const { insertCompressionAnalyticsRow, insertCompressionEngineBreakdown } = await import(
-        "@/lib/db/compressionAnalytics"
-      );
-      const { calculateCost } = await import("@/lib/usage/costCalculator");
+      const { insertCompressionAnalyticsRow, insertCompressionEngineBreakdown } =
+        await import("@/lib/db/compressionAnalytics");
       const { stats } = opts;
       const tokensSaved = Math.max(0, stats.originalTokens - stats.compressedTokens);
       const rtkPointers = (stats.rtkRawOutputPointers ?? []) as RtkPointer[];
-      const estimatedUsdSaved = await calculateCost(
-        opts.provider ?? "",
-        opts.effectiveModel ?? "",
-        { input: tokensSaved },
-        { serviceTier: opts.effectiveServiceTier }
+      let estimatedUsdSaved = 0;
+      try {
+        const calculateCost =
+          dependencies.calculateCost ?? (await import("@/lib/usage/costCalculator")).calculateCost;
+        estimatedUsdSaved = await calculateCost(
+          opts.provider ?? "",
+          opts.effectiveModel ?? "",
+          { input: tokensSaved },
+          { serviceTier: opts.effectiveServiceTier }
+        );
+      } catch (err) {
+        opts.log?.debug?.(
+          "COMPRESSION",
+          "Compression cost estimate skipped: " + (err instanceof Error ? err.message : String(err))
+        );
+      }
+      insertCompressionAnalyticsRow(
+        buildAnalyticsRow(opts, tokensSaved, rtkPointers, estimatedUsdSaved)
       );
-      insertCompressionAnalyticsRow(buildAnalyticsRow(opts, tokensSaved, rtkPointers, estimatedUsdSaved));
       const breakdownRows = buildEngineBreakdownRows(stats, opts.skillRequestId);
       if (breakdownRows.length > 0) {
         insertCompressionEngineBreakdown(breakdownRows);
       }
     } catch (err) {
-      opts.log?.debug?.(
+      opts.log?.warn?.(
         "COMPRESSION",
         "Compression analytics write skipped: " + (err instanceof Error ? err.message : String(err))
       );

open-sse/services/compression/bodyAdapter.ts

@@ -14,7 +14,11 @@ type ResponsesItem = {
   [key: string]: unknown;
 };
 
-const RESPONSES_MESSAGE_TYPES = new Set(["message", "function_call_output"]);
+const RESPONSES_MESSAGE_TYPES = new Set([
+  "message",
+  "function_call_output",
+  "custom_tool_call_output",
+]);
 const COMPRESSION_INPUT_INDEX = Symbol("compressionInputIndex");
 
 // Kiro envelope path back to the original tool-result text inside
@@ -60,11 +64,47 @@ function fromChatContent(nextContent: unknown, originalContent: unknown): unknow
   return nextContent;
 }
 
+function customToolOutputToChatContent(rawOutput: unknown): unknown {
+  if (typeof rawOutput !== "string") {
+    if (isRecord(rawOutput) && typeof rawOutput.output === "string") return rawOutput.output;
+    return rawOutput;
+  }
+
+  try {
+    const parsed = JSON.parse(rawOutput) as unknown;
+    if (isRecord(parsed) && typeof parsed.output === "string") return parsed.output;
+  } catch {
+    // Plain-text custom tool output is already in the form compression engines expect.
+  }
+  return rawOutput;
+}
+
+function restoreCustomToolOutput(nextContent: unknown, originalOutput: unknown): unknown {
+  if (typeof originalOutput === "string") {
+    try {
+      const parsed = JSON.parse(originalOutput) as unknown;
+      if (isRecord(parsed) && typeof parsed.output === "string") {
+        return JSON.stringify({ ...parsed, output: nextContent });
+      }
+    } catch {
+      // Preserve the original plain-text representation below.
+    }
+  }
+  if (isRecord(originalOutput) && typeof originalOutput.output === "string") {
+    return { ...originalOutput, output: nextContent };
+  }
+  return fromChatContent(nextContent, originalOutput);
+}
+
+function responsesToolOutputField(item: ResponsesItem): "output" | "content" {
+  return item.output !== null && item.output !== undefined ? "output" : "content";
+}
+
 function responsesItemToMessage(item: ResponsesItem): MessageLike | null {
   const type = typeof item.type === "string" ? item.type : "message";
   if (!RESPONSES_MESSAGE_TYPES.has(type)) return null;
 
-  if (type === "function_call_output") {
+  if (type === "function_call_output" || type === "custom_tool_call_output") {
     const rawOutput = item.output ?? item.content;
     // OpenAI Responses shape (Codex): body.input holds Responses items. When
     // output is a JSON object (not a string or content array), serialise it so
@@ -77,7 +117,12 @@ function responsesItemToMessage(item: ResponsesItem): MessageLike | null {
       !Array.isArray(rawOutput);
     return {
       role: "tool",
-      content: isObjectOutput ? JSON.stringify(rawOutput) : toChatContent(rawOutput),
+      content:
+        type === "custom_tool_call_output"
+          ? customToolOutputToChatContent(rawOutput)
+          : isObjectOutput
+            ? JSON.stringify(rawOutput)
+            : toChatContent(rawOutput),
     };
   }
 
@@ -89,10 +134,15 @@ function responsesItemToMessage(item: ResponsesItem): MessageLike | null {
 
 function messageToResponsesItem(message: MessageLike, originalItem: ResponsesItem): ResponsesItem {
   const type = typeof originalItem.type === "string" ? originalItem.type : "message";
-  if (type === "function_call_output") {
+  if (type === "function_call_output" || type === "custom_tool_call_output") {
+    const outputField = responsesToolOutputField(originalItem);
+    const originalOutput = originalItem[outputField];
     return {
       ...originalItem,
-      output: fromChatContent(message.content, originalItem.output),
+      [outputField]:
+        type === "custom_tool_call_output"
+          ? restoreCustomToolOutput(message.content, originalOutput)
+          : fromChatContent(message.content, originalOutput),
     };
   }
 
@@ -332,7 +382,12 @@ function rewriteKiroEntry(
     let trChanged = false;
     const nextContent = content.map((part, partIdx) => {
       if (!isRecord(part) || typeof part.text !== "string") return part;
-      const key = kiroPathKey({ scope, historyIndex, toolResultIndex: trIdx, contentIndex: partIdx });
+      const key = kiroPathKey({
+        scope,
+        historyIndex,
+        toolResultIndex: trIdx,
+        contentIndex: partIdx,
+      });
       const rewritten = rewrites.get(key);
       if (rewritten === undefined || rewritten === part.text) return part;
       trChanged = true;

src/app/(dashboard)/dashboard/analytics/CompressionAnalyticsTab.tsx

@@ -83,7 +83,10 @@ function ModeBar({
           {skipped > 0 && (
             // #4268: attempted-but-no-op runs (e.g. Stacked saved nothing) are
             // recorded now, so this mode is visible even when count is 0.
-            <span className="text-text-muted/70"> · {skipped.toLocaleString()} skipped (no-op)</span>
+            <span className="text-text-muted/70">
+              {" "}
+              · {skipped.toLocaleString()} skipped (no-op)
+            </span>
           )}
         </span>
       </div>
@@ -174,6 +177,7 @@ export default function CompressionAnalyticsTab() {
 
   const modes = Object.entries(stats.byMode).sort(([, a], [, b]) => b.count - a.count);
   const providers = Object.entries(stats.byProvider).sort(([, a], [, b]) => b.count - a.count);
+  const totalAttempts = stats.totalRequests + (stats.totalSkipped ?? 0);
 
   // Calculate max tokens for hourly chart scaling
   const maxTokensPerHour = Math.max(...stats.last24h.map((h) => h.tokensSaved), 1);
@@ -209,7 +213,7 @@ export default function CompressionAnalyticsTab() {
         <StatCard
           icon="compress"
           label={t("compressionAnalyticsTotalRequests")}
-          value={stats.totalRequests.toLocaleString()}
+          value={totalAttempts.toLocaleString()}
         />
         <StatCard
           icon="token"
@@ -365,7 +369,7 @@ export default function CompressionAnalyticsTab() {
       )}
 
       {/* Empty state */}
-      {stats.totalRequests === 0 && (
+      {totalAttempts === 0 && (
         <div className="card p-8 text-center text-text-muted">
           <span className="material-symbols-outlined text-[48px] mb-3 block text-primary opacity-50">
             compress

tests/unit/chatcore-compression-analytics-write.test.ts

@@ -12,9 +12,8 @@ const testDataDir = fs.mkdtempSync(path.join(os.tmpdir(), "omni-comp-analytics-t
 process.env.DATA_DIR = testDataDir;
 
 const coreDb = await import("../../src/lib/db/core.ts");
-const { writeCompressionAnalytics } = await import(
-  "../../open-sse/handlers/chatCore/compressionAnalyticsWrite.ts"
-);
+const { writeCompressionAnalytics } =
+  await import("../../open-sse/handlers/chatCore/compressionAnalyticsWrite.ts");
 
 type Stats = Parameters<typeof writeCompressionAnalytics>[0]["stats"];
 
@@ -37,7 +36,7 @@ function analyticsRow(requestId: string): Record<string, unknown> | undefined {
     return coreDb
       .getDbInstance()
       .prepare(
-        "SELECT mode, engine, original_tokens AS orig, compressed_tokens AS comp, tokens_saved AS saved, validation_fallback AS vf FROM compression_analytics WHERE request_id = ?"
+        "SELECT mode, engine, original_tokens AS orig, compressed_tokens AS comp, tokens_saved AS saved, estimated_usd_saved AS usd, validation_fallback AS vf FROM compression_analytics WHERE request_id = ?"
       )
       .get(requestId) as Record<string, unknown> | undefined;
   } catch {
@@ -122,6 +121,19 @@ test("inserts a per-engine breakdown when present", async () => {
   assert.equal(breakdownCount("ca-req-3"), 2);
 });
 
+test("inserts the analytics row when calculateCost throws", async () => {
+  await writeCompressionAnalytics(baseOpts("ca-req-cost-error"), {
+    calculateCost: async () => {
+      throw new Error("forced cost-estimate failure");
+    },
+  });
+
+  const row = analyticsRow("ca-req-cost-error");
+  assert.ok(row, "expected a compression_analytics row despite the cost-estimate failure");
+  assert.equal(row!.saved, 120);
+  assert.equal(row!.usd, null);
+});
+
 test("never rejects even on a bad write (fail-open)", async () => {
   coreDb.resetDbInstance();
   await assert.doesNotReject(writeCompressionAnalytics(baseOpts("ca-req-4")));

tests/unit/compression/body-adapter.test.ts

@@ -78,6 +78,92 @@ describe("compression body adapter", () => {
     assert.equal(input[0].call_id, "call_1");
   });
 
+  it("applies RTK compression to Codex custom_tool_call_output items", () => {
+    const repeatedOutput = Array.from({ length: 20 }, () => "same noisy line").join("\n");
+    const body = {
+      input: [
+        {
+          type: "custom_tool_call_output",
+          call_id: "call_patch_1",
+          output: repeatedOutput,
+        },
+      ],
+    };
+
+    const result = applyRtkCompression(body);
+
+    assert.equal(result.compressed, true);
+    assert.ok(!("messages" in result.body), "Responses body must not leak synthetic messages");
+    const input = result.body.input as typeof body.input;
+    assert.equal(input[0].type, "custom_tool_call_output");
+    assert.equal(input[0].call_id, "call_patch_1");
+    assert.match(input[0].output, /\[rtk:dropped/);
+  });
+
+  it("preserves wrapped custom tool output metadata while compressing its output", () => {
+    const repeatedOutput = Array.from({ length: 20 }, () => "same noisy line").join("\n");
+    const body = {
+      input: [
+        {
+          type: "custom_tool_call_output",
+          call_id: "call_exec_1",
+          output: JSON.stringify({ output: repeatedOutput, metadata: { exitCode: 0 } }),
+        },
+      ],
+    };
+
+    const result = applyRtkCompression(body);
+    const input = result.body.input as typeof body.input;
+    const restoredOutput = JSON.parse(input[0].output) as {
+      output: string;
+      metadata: { exitCode: number };
+    };
+
+    assert.equal(result.compressed, true);
+    assert.match(restoredOutput.output, /\[rtk:dropped/);
+    assert.deepEqual(restoredOutput.metadata, { exitCode: 0 });
+  });
+
+  it("restores custom tool output to content when that was the source field", () => {
+    const repeatedOutput = Array.from({ length: 20 }, () => "same noisy line").join("\n");
+    const body = {
+      input: [
+        {
+          type: "custom_tool_call_output",
+          call_id: "call_exec_2",
+          content: repeatedOutput,
+        },
+      ],
+    };
+
+    const result = applyRtkCompression(body);
+    const input = result.body.input as Array<Record<string, unknown>>;
+
+    assert.equal(result.compressed, true);
+    assert.match(input[0].content as string, /\[rtk:dropped/);
+    assert.ok(!("output" in input[0]), "restore must not add a conflicting output field");
+  });
+
+  it("restores function call output to content when that was the source field", () => {
+    const repeatedOutput = Array.from({ length: 20 }, () => "same noisy line").join("\n");
+    const body = {
+      input: [
+        {
+          type: "function_call_output",
+          call_id: "call_2",
+          content: repeatedOutput,
+        },
+      ],
+    };
+
+    const result = applyRtkCompression(body);
+    const input = result.body.input as Array<Record<string, unknown>>;
+
+    assert.equal(result.compressed, true);
+    assert.match(input[0].content as string, /\[rtk:dropped/);
+    assert.ok(!("output" in input[0]), "restore must not add a conflicting output field");
+  });
+
   it("restores compressed array output on Responses function_call_output items", () => {
     const repeatedOutput = Array.from({ length: 20 }, () => "same noisy line").join("\n");
     const body = {