diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 98f4c44681..3e0c90fe69 100644 --- a/open-sse/handlers/chatCore.ts +++ b/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, diff --git a/open-sse/handlers/chatCore/cavemanOutputAnalytics.ts b/open-sse/handlers/chatCore/cavemanOutputAnalytics.ts index fa3a778d9c..18307628ce 100644 --- a/open-sse/handlers/chatCore/cavemanOutputAnalytics.ts +++ b/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)) diff --git a/open-sse/handlers/chatCore/compressionAnalyticsWrite.ts b/open-sse/handlers/chatCore/compressionAnalyticsWrite.ts index 21b21b33c0..ec68aff23c 100644 --- a/open-sse/handlers/chatCore/compressionAnalyticsWrite.ts +++ b/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 { +export function writeCompressionAnalytics( + opts: WriteOpts, + dependencies: WriteDependencies = {} +): Promise { 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)) ); diff --git a/open-sse/services/compression/bodyAdapter.ts b/open-sse/services/compression/bodyAdapter.ts index 9c42fbdd6f..af407c510d 100644 --- a/open-sse/services/compression/bodyAdapter.ts +++ b/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; diff --git a/src/app/(dashboard)/dashboard/analytics/CompressionAnalyticsTab.tsx b/src/app/(dashboard)/dashboard/analytics/CompressionAnalyticsTab.tsx index 1ad0ea9926..c5e9c4d7f1 100644 --- a/src/app/(dashboard)/dashboard/analytics/CompressionAnalyticsTab.tsx +++ b/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. - · {skipped.toLocaleString()} skipped (no-op) + + {" "} + · {skipped.toLocaleString()} skipped (no-op) + )} @@ -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() { compress diff --git a/tests/unit/chatcore-compression-analytics-write.test.ts b/tests/unit/chatcore-compression-analytics-write.test.ts index 15768b22e1..ab3d056b8f 100644 --- a/tests/unit/chatcore-compression-analytics-write.test.ts +++ b/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[0]["stats"]; @@ -37,7 +36,7 @@ function analyticsRow(requestId: string): Record | 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 | 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"))); diff --git a/tests/unit/compression/body-adapter.test.ts b/tests/unit/compression/body-adapter.test.ts index be584bf211..4bc3103d3d 100644 --- a/tests/unit/compression/body-adapter.test.ts +++ b/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>; + + 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>; + + 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 = {