From f4100b256e15c4ad78f6d08279185cb9e670efe0 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:27:02 -0700 Subject: [PATCH] fix: flush attempt receipts and surface checkpoint failures Co-Authored-By: Paperclip --- .../src/usage-checkpoint.test.ts | 56 +++++++++++++++ .../adapter-utils/src/usage-checkpoint.ts | 69 +++++++++++++------ .../claude-local/src/server/execute.ts | 22 +++--- .../codex-local/src/server/execute.ts | 2 + .../cursor-local/src/server/execute.ts | 4 +- .../adapters/cursor-local/src/server/parse.ts | 3 + .../gemini-local/src/server/execute.ts | 11 +-- .../adapters/kimi-local/src/server/execute.ts | 1 + .../opencode-local/src/server/execute.ts | 13 ++-- .../adapters/pi-local/src/server/execute.ts | 11 +-- .../adapters/pi-local/src/server/parse.ts | 3 + .../adapter-cost-failure-paths.test.ts | 46 +++++++++++++ 12 files changed, 198 insertions(+), 43 deletions(-) create mode 100644 packages/adapter-utils/src/usage-checkpoint.test.ts diff --git a/packages/adapter-utils/src/usage-checkpoint.test.ts b/packages/adapter-utils/src/usage-checkpoint.test.ts new file mode 100644 index 00000000000..79d8167460c --- /dev/null +++ b/packages/adapter-utils/src/usage-checkpoint.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it, vi } from "vitest"; +import { createUsageCheckpointLog } from "./usage-checkpoint.js"; +import { runChildProcess } from "./server-utils.js"; +import type { AdapterUsageCheckpoint } from "./types.js"; + +const parse = (stdout: string): AdapterUsageCheckpoint => { + const events = stdout.trim().split("\n").map(line => JSON.parse(line)); + return { costUsd: events.reduce((sum, event) => sum + event.usage.costUsd, 0), complete: false }; +}; +const event = JSON.stringify({ type: "usage", usage: { inputTokens: 2, costUsd: 0.25 }, text: "private content" }); + +describe("usage checkpoint stream", () => { + it("flushes an unterminated Pi price and closes each attempt independently", async () => { + const saved = vi.fn(); const parser = vi.fn(parse); + const first = createUsageCheckpointLog(vi.fn(), saved, parser); + await first("stdout", event); + expect(saved).not.toHaveBeenCalled(); + await first.flush({ complete: true }); + const second = createUsageCheckpointLog(vi.fn(), saved, parse); + await second("stdout", event + "\n"); + await second.flush({ complete: true }); + expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true }); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: true }); + expect(saved.mock.calls.at(-1)![0].attemptId).not.toBe(saved.mock.calls[0][0].attemptId); + expect(parser.mock.calls[0][0]).not.toContain("private content"); + }); + + it("surfaces a persistence failure outside the local process's best-effort log path", async () => { + const failure = new Error("Receipt storage unavailable"); + const output = vi.fn(); const saved = vi.fn().mockRejectedValue(failure); + const log = createUsageCheckpointLog(output, saved, parse); + const result = await runChildProcess("checkpoint-test", process.execPath, ["-e", `process.stdout.write(${JSON.stringify(event + "\n")})`], { + cwd: process.cwd(), env: {}, timeoutSec: 5, graceSec: 1, onLog: log, + }); + expect(result.exitCode).toBe(0); + expect(output).toHaveBeenCalledWith("stdout", event + "\n"); + await expect(log.flush()).rejects.toBe(failure); + }); + + it("does not promote partial usage to complete merely because the stream was flushed", async () => { + const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parse); + await log("stdout", event); await log.flush(); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false }); + }); + + it("bounds replay work by stream size while retaining the exact final total", async () => { + const parser = vi.fn(parse); const saved = vi.fn(); + const log = createUsageCheckpointLog(vi.fn(), saved, parser); + for (let i = 0; i < 1024; i++) await log("stdout", event + "\n"); + await log.flush({ complete: true }); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 256, complete: true }); + expect(parser.mock.calls.length).toBeLessThanOrEqual(12); + const finalSize = parser.mock.calls.at(-1)![0].length; + expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThan(3 * finalSize); + }); +}); diff --git a/packages/adapter-utils/src/usage-checkpoint.ts b/packages/adapter-utils/src/usage-checkpoint.ts index 39b05c68a5f..f7ebe701203 100644 --- a/packages/adapter-utils/src/usage-checkpoint.ts +++ b/packages/adapter-utils/src/usage-checkpoint.ts @@ -9,7 +9,7 @@ const fields = new Set([ "cache_read_input_tokens", "cache_creation_input_tokens", "inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens", "cachedInputTokens", "promptTokenCount", "candidatesTokenCount", "cachedContentTokenCount", "reasoning", "cacheRead", "cacheWrite", "prompt", "candidates", "totalTokenCount", "total_tokens", "toolUsePromptTokenCount", "messages", - "thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD", + "thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD", "costUsd", ]); const stringFields = new Set(["type", "subtype", "role", "id", "session_id", "sessionID", "thread_id", "model", "modelID", "providerID"]); function project(value: unknown, depth = 0, field = ""): unknown { @@ -24,36 +24,63 @@ function project(value: unknown, depth = 0, field = ""): unknown { : project(entry, depth + 1, key)])); } -/** A fresh instance belongs to one CLI attempt. Retries must not concatenate - * cumulative usage from different attempts. Callback durability is awaited - * before ordinary log publication. */ +/** A fresh instance belongs to one CLI attempt. Call flush() after its process + * exits, before retrying or final result handling. Checkpoint failures are + * retained outside runChildProcess's best-effort onLog error handler. */ export function createUsageCheckpointLog( onLog: (stream: "stdout" | "stderr", chunk: string) => Promise, onUsage: ((receipt: AdapterUsageCheckpoint) => Promise) | undefined, parse: (stdout: string) => AdapterUsageCheckpoint | null, ) { - if (!onUsage) return onLog; const attemptId = randomUUID(); let remainder = "", accounting = "", previous = ""; - return async (stream: "stdout" | "stderr", chunk: string) => { - if (stream === "stdout") { - remainder += chunk; - const lines = remainder.split("\n"); remainder = lines.pop() ?? ""; - if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB"); - for (const line of lines) { - let raw: unknown; - try { raw = JSON.parse(line); } catch { continue; } - const compact = JSON.stringify(project(raw)); - if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"/.test(compact)) continue; - accounting += compact + "\n"; - if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB"); - const receipt = parse(accounting); - if (!receipt) continue; - const serialized = JSON.stringify(receipt); - if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; } + let nextReplaySize = 0; + let failure: unknown; + let failed = false; + function retain(line: string) { + let raw: unknown; + try { raw = JSON.parse(line); } catch { return; } + const compact = JSON.stringify(project(raw)); + if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"|"model"|"modelID"/.test(compact)) return; + accounting += compact + "\n"; + if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB"); + } + async function publish(final = false, complete = false) { + if (!onUsage || !accounting || (!final && accounting.length < nextReplaySize)) return; + const parsed = parse(accounting); + // Geometric replay bounds total parsing work by stream size. The first + // observation is saved immediately; flush always saves the final snapshot. + nextReplaySize = accounting.length * 2; + if (!parsed) return; + if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null && + !Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return; + const receipt = { ...parsed, complete: parsed.complete || complete }; + const serialized = JSON.stringify(receipt); + if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; } + } + const log = async (stream: "stdout" | "stderr", chunk: string) => { + if (onUsage && !failed && stream === "stdout") { + try { + remainder += chunk; + const lines = remainder.split("\n"); remainder = lines.pop() ?? ""; + if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB"); + for (const line of lines) retain(line); + await publish(); + } catch (error) { + failed = true; failure = error; remainder = ""; accounting = ""; } } + // Receipt persistence and ordinary logs have separate failure contracts. + // A failed receipt must not hide the provider's diagnostic output. await onLog(stream, chunk); }; + return Object.assign(log, { + async flush(options: { complete?: boolean } = {}) { + if (failed) throw failure; + if (!onUsage) return; + if (remainder) { retain(remainder); remainder = ""; } + await publish(true, options.complete ?? false); + }, + }); } import { randomUUID } from "node:crypto"; diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 378917ecadb..2f407d05364 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -964,6 +964,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parseClaudeStreamJson(stdout); + return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd, + provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic", + billingType, model: Object.keys(parseObject(parsed.resultJson?.modelUsage)).length > 1 ? "mixed" : parsed.model || model, + usageByModel: claudeModelReceipts(parsed.resultJson?.modelUsage), complete: parsed.resultJson !== null }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -973,12 +980,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseClaudeStreamJson(stdout); - return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd, - provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic", - billingType, model: parsed.model || model, complete: parsed.resultJson !== null }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, terminalResultCleanup: { @@ -987,6 +989,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise 1 ? "mixed" : parsedStream.model || model, + usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage), billingType, costUsd: parsedStream.costUsd, errorMessage: `Timed out after ${timeoutSec}s`, @@ -1091,7 +1096,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise 1 ? "mixed" : parsedStream.model || model, + usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage), billingType, costUsd: parsedStream.costUsd, errorMessage: fallbackErrorMessage, diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index a302bdbba86..2069e55f957 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1347,6 +1347,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { const parsed = parseCursorJsonl(stdout); const provider = resolveProviderFromModel(model); - return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: false }; + return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult }; }); let stdoutLineBuffer = ""; const emitNormalizedStdoutLine = async (rawLine: string) => { @@ -676,6 +676,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise): string | null { export function parseCursorJsonl(stdout: string) { let sessionId: string | null = null; + let sawResult = false; const messages: string[] = []; let errorMessage: string | null = null; let totalCostUsd = 0; @@ -97,6 +98,7 @@ export function parseCursorJsonl(stdout: string) { } if (type === "result") { + sawResult = true; const usageObj = parseObject(event.usage); usage.inputTokens += asNumber( usageObj.input_tokens, @@ -160,6 +162,7 @@ export function parseCursorJsonl(stdout: string) { } return { + sawResult, sessionId, summary: messages.join("\n\n").trim(), usage, diff --git a/packages/adapters/gemini-local/src/server/execute.ts b/packages/adapters/gemini-local/src/server/execute.ts index 1ba97244a2d..500d395a819 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -643,6 +643,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parseGeminiJsonl(stdout); + return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -651,13 +655,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseGeminiJsonl(stdout); - return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, }); + await accountingLog.flush({ complete: !proc.timedOut && !proc.signal }); return { proc, parsed: parseGeminiJsonl(proc.stdout), @@ -691,6 +693,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parseOpenCodeJsonl(stdout); + const provider = parseModelProvider(model || null); + return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -655,14 +660,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseOpenCodeJsonl(stdout); - const provider = parseModelProvider(model || null); - return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, }); + await accountingLog.flush({ complete: !proc.timedOut && !proc.signal }); return { proc, rawStderr: proc.stderr, @@ -683,6 +685,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = parsePiJsonl(stdout); + return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: parsed.sawAgentEnd }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -745,13 +749,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parsePiJsonl(stdout); - return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false }; - }), + onLog: accountingLog, runLogTail: paperclipBridge?.runLogTail, settleRunDisposition: paperclipBridge?.settleRunDisposition, }); + await accountingLog.flush({ complete: !proc.timedOut && !proc.signal }); // Flush any remaining buffer content if (stdoutBuffer) { @@ -778,6 +780,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise> | undefined; if (messages && messages.length > 0) { const lastMessage = messages[messages.length - 1]; diff --git a/server/src/__tests__/adapter-cost-failure-paths.test.ts b/server/src/__tests__/adapter-cost-failure-paths.test.ts index 6355240efbb..0ad2c8d0038 100644 --- a/server/src/__tests__/adapter-cost-failure-paths.test.ts +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -31,6 +31,51 @@ const cases: Array<{ name: string; execute: () => Promise; event: unkno describe("CLI adapter accounting on timeout", () => { const directories: string[] = []; afterEach(async () => { vi.clearAllMocks(); for (const dir of directories.splice(0)) await rm(dir, { recursive: true, force: true }); }); + it.each(cases.filter(fixture => fixture.name !== "kimi"))("$name flushes its attempt and surfaces checkpoint persistence failures", async (fixture) => { + const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir); + const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 }); + const stdout = JSON.stringify(fixture.event); + processResult.mockImplementation(async (_run, _target, _command, _args, options) => { + // Local process logging catches callback failures. Finalization must + // independently enforce receipt durability despite that contract. + await options.onLog("stdout", stdout).catch(() => {}); + return { exitCode: 0, signal: null, timedOut: false, stdout, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const onUsage = vi.fn(); + const context: AdapterExecutionContext = { + runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} }, + runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, + config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [], + env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, + context: {}, onLog: async () => {}, onUsage, + }; + const execute = await fixture.execute(); + await execute(context); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete: true }); + if (fixture.name === "claude") { + const mixed = [ + { type: "system", subtype: "init", model: "actual-model" }, + { type: "result", total_cost_usd: 0.003, usage: { input_tokens: 3, output_tokens: 3 }, modelUsage: { + "model-a": { inputTokens: 1, outputTokens: 1, costUSD: 0.001 }, + "model-b": { inputTokens: 2, outputTokens: 2, costUSD: 0.002 }, + } }, + ]; + processResult.mockImplementation(async (_run, _target, _command, _args, options) => { + const output = mixed.map(item => JSON.stringify(item)).join("\n"); + await options.onLog("stdout", output).catch(() => {}); + return { exitCode: 0, signal: null, timedOut: false, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + await execute(context); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "mixed", costUsd: 0.003, complete: true, usageByModel: [ + { model: "model-a", costUsd: 0.001 }, { model: "model-b", costUsd: 0.002 }, + ] }); + delete (mixed[1] as { modelUsage?: unknown }).modelUsage; + await execute(context); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "actual-model", complete: true }); + } + onUsage.mockRejectedValue(new Error("Receipt persistence failed")); + await expect(execute(context)).rejects.toThrow("Receipt persistence failed"); + }); it.each(cases)("$name retains observed accounting when the process times out", async (fixture) => { const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir); const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 }); @@ -44,6 +89,7 @@ describe("CLI adapter accounting on timeout", () => { context: {}, onLog: async () => {}, }); expect(result.timedOut).toBe(true); + expect(result.usageComplete).toBe(["codex", "cursor", "gemini"].includes(fixture.name)); expect(result.usageBasis).toBe("per_run"); expect(result.provider).toBeTruthy(); expect(result.billingType).toBeTruthy();