From f4100b256e15c4ad78f6d08279185cb9e670efe0 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:27:02 -0700 Subject: [PATCH 1/4] 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(); From 2a811ea1a9b93ee2996b7cf69eeb191242d805fe Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:33:49 -0700 Subject: [PATCH 2/4] test: forward stdin through the Cursor fixture runner Co-Authored-By: Paperclip --- packages/adapters/cursor-local/src/server/execute.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/adapters/cursor-local/src/server/execute.test.ts b/packages/adapters/cursor-local/src/server/execute.test.ts index 913d34cdc40..6aea2414d3d 100644 --- a/packages/adapters/cursor-local/src/server/execute.test.ts +++ b/packages/adapters/cursor-local/src/server/execute.test.ts @@ -288,7 +288,7 @@ printf '%s\\n' '{"type":"result","subtype":"success","session_id":"cursor-sessio finalPreparedCommand = preferredCommandPath; const runtimeEnv = { ...input.env, - PATH: `${path.join(systemHomeDir, ".local", "bin")}${path.delimiter}${input.env.PATH}`, + PATH: `${path.join(systemHomeDir, ".local", "bin")}${path.delimiter}${input.env.PATH ?? process.env.PATH ?? "/usr/bin:/bin"}`, }; await fs.mkdir(path.dirname(preferredCommandPath), { recursive: true }); await fs.writeFile(preferredCommandPath, preferredAgentScript); @@ -312,13 +312,14 @@ printf '%s\\n' '{"type":"result","subtype":"success","session_id":"cursor-sessio // shell seam faithful to that protocol instead of returning empty stdout // for every shell command. const runner = { - execute: async (input: { command: string; args?: string[]; env?: Record }) => { + execute: async (input: { command: string; args?: string[]; env?: Record; stdin?: string }) => { runnerState.commands.push(input.command); // Exercise actual bounded file reads during managed-home restoration; // reporting empty success for every shell command hides missing bytes. return runChildProcess(`cursor-fresh-lease-${runnerState.commands.length}`, input.command, input.args ?? [], { cwd: remoteWorkspace, env: { ...input.env, PATH: `${input.env?.PATH ?? ""}:/usr/bin:/bin` }, + stdin: input.stdin, timeoutSec: 30, graceSec: 5, onLog: async () => {}, From 32501493050743867eeb5e75ed84b1040c27f8f0 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:37:20 -0700 Subject: [PATCH 3/4] fix: invalidate quota caches when personal credentials are disabled Co-Authored-By: Paperclip --- server/src/__tests__/ai-connections.test.ts | 9 ++++++++- server/src/__tests__/company-quota-windows.test.ts | 2 ++ server/src/services/quota-windows.ts | 12 ++++++++---- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/ai-connections.test.ts b/server/src/__tests__/ai-connections.test.ts index 8be04f43fba..3cb74ed7039 100644 --- a/server/src/__tests__/ai-connections.test.ts +++ b/server/src/__tests__/ai-connections.test.ts @@ -8,7 +8,7 @@ import { mkdtemp, rm, access, readFile, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { and, eq, sql } from "drizzle-orm"; -import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, companySecretVersions, principalPermissionGrants } from "@paperclipai/db"; +import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, companySecretVersions, userSecretDefinitions, principalPermissionGrants } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "@paperclipai/db/test-embedded-postgres"; import { fetchCompanyQuotaWindows } from "../services/quota-windows.js"; import { aiConnectionService } from "../services/ai-connections.js"; @@ -69,6 +69,13 @@ describe("managed AI connections", () => { expect(rotated[0].accountKey).not.toBe(first[0].accountKey); expect(fetchSpy).toHaveBeenCalledTimes(2); expect(fetchSpy).toHaveBeenLastCalledWith(expect.any(String), expect.objectContaining({ headers: expect.objectContaining({ Authorization: "Bearer quota-cache-rotated" }) })); + await db.update(userSecretDefinitions).set({ status: "disabled" }).where(eq(userSecretDefinitions.id, before.userSecretDefinitionId!)); + const disabled = await fetchCompanyQuotaWindows(db, companyId, owner); + expect(disabled[0]).toMatchObject({ ok: false, errorFamily: "credentials_unavailable", windows: [] }); + expect(disabled[0].accountKey).not.toBe(rotated[0].accountKey); + expect(fetchSpy).toHaveBeenCalledTimes(2); + await db.update(userSecretDefinitions).set({ status: "active" }).where(eq(userSecretDefinitions.id, before.userSecretDefinitionId!)); + expect((await fetchCompanyQuotaWindows(db, companyId, owner))[0].ok).toBe(true); await db.update(companySecretVersions).set({ status: "disabled", revokedAt: new Date() }).where(and(eq(companySecretVersions.secretId, secretId), eq(companySecretVersions.version, 2))); const revoked = await fetchCompanyQuotaWindows(db, companyId, owner); expect(revoked[0].ok).toBe(false); diff --git a/server/src/__tests__/company-quota-windows.test.ts b/server/src/__tests__/company-quota-windows.test.ts index dd4da0a056c..d848986790a 100644 --- a/server/src/__tests__/company-quota-windows.test.ts +++ b/server/src/__tests__/company-quota-windows.test.ts @@ -28,6 +28,7 @@ let revision = 1; const db = { select: () => ({ from: () => ({ + leftJoin: () => ({ leftJoin: () => ({ where: async () => [ { @@ -38,6 +39,7 @@ const db = { }, ], }), + }), }), }), } as unknown as Db; diff --git a/server/src/services/quota-windows.ts b/server/src/services/quota-windows.ts index a897f59904e..c290969c96f 100644 --- a/server/src/services/quota-windows.ts +++ b/server/src/services/quota-windows.ts @@ -4,7 +4,7 @@ import { listServerAdapters } from "../adapters/registry.js"; import { logger } from "../middleware/logger.js"; import { createHash } from "node:crypto"; import { eq, inArray, and } from "drizzle-orm"; -import { companySecrets, companySecretVersions, type Db } from "@paperclipai/db"; +import { companySecrets, companySecretVersions, userSecretDefinitions, type Db } from "@paperclipai/db"; import { fetchCodexQuota } from "@paperclipai/adapter-codex-local/server"; import { fetchClaudeQuota } from "@paperclipai/adapter-claude-local/server"; import { aiConnectionService } from "./ai-connections.js"; @@ -25,15 +25,19 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId // Resolve revision metadata before consulting the cache, so rotation and // revocation cannot serve windows associated with an old credential. const secretIds = accounts.flatMap(row => row.grant.credentialSecretRefs.map(ref => ref.secretId)); - const revisions = secretIds.length ? await db.select({ id: companySecrets.id, latestVersion: companySecrets.latestVersion, status: companySecrets.status, versionStatus: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt }) - .from(companySecrets).leftJoin(companySecretVersions, and(eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, companySecrets.latestVersion))).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : []; + const revisions = secretIds.length ? await db.select({ id: companySecrets.id, latestVersion: companySecrets.latestVersion, status: companySecrets.status, versionStatus: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt, definitionId: companySecrets.userSecretDefinitionId, definitionStatus: userSecretDefinitions.status, definitionDeletedAt: userSecretDefinitions.deletedAt }) + .from(companySecrets).leftJoin(companySecretVersions, and(eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, companySecrets.latestVersion))).leftJoin(userSecretDefinitions, and(eq(userSecretDefinitions.id, companySecrets.userSecretDefinitionId), eq(userSecretDefinitions.companyId, companyId))).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : []; async function readAccount(row: (typeof accounts)[number]): Promise { const accountKey = createHash("sha256").update(JSON.stringify([companyId, row.connection.id, row.grant.id, row.connection.updatedAt, row.grant.updatedAt, row.grant.credentialSecretRefs.map(ref => [ref.secretId, revisions.find(r => r.id === ref.secretId)]), ])).digest("hex"); const base = { provider: row.summary.provider, accountKey, accountLabel: row.summary.name, source: "managed-connection" }; - if (row.summary.status !== "connected") { + const definitionUnavailable = row.grant.credentialSecretRefs.some(ref => { + const revision = revisions.find(value => value.id === ref.secretId); + return revision?.definitionId && (revision.definitionStatus !== "active" || revision.definitionDeletedAt !== null); + }); + if (row.summary.status !== "connected" || definitionUnavailable) { return { ...base, ok: false, errorFamily: "credentials_unavailable", error: publicError, windows: [] }; } const key = `${userId}:${accountKey}`; From 014810c58da90f3a69063187fc05e4da731b916a Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:45:06 -0700 Subject: [PATCH 4/4] fix: checkpoint every usage update with incremental protocol parsers Use provider terminal evidence for completeness; an OpenCode successful EOF closes its step-based stream. Preserve immutable snapshots and propagate partial completeness through final results. Co-Authored-By: Paperclip --- .../src/usage-checkpoint.test.ts | 43 ++- .../adapter-utils/src/usage-checkpoint.ts | 24 +- .../claude-local/src/server/execute.ts | 9 +- .../adapters/claude-local/src/server/parse.ts | 143 ++++---- .../codex-local/src/server/execute.ts | 11 +- .../adapters/codex-local/src/server/parse.ts | 105 +++--- .../cursor-local/src/server/execute.ts | 8 +- .../adapters/cursor-local/src/server/parse.ts | 159 ++++----- .../gemini-local/src/server/execute.ts | 8 +- .../adapters/gemini-local/src/server/parse.ts | 195 ++++++----- .../opencode-local/src/server/execute.ts | 10 +- .../opencode-local/src/server/parse.ts | 99 +++--- .../adapters/pi-local/src/server/execute.ts | 10 +- .../adapters/pi-local/src/server/parse.ts | 321 +++++++++--------- .../adapter-cost-failure-paths.test.ts | 47 ++- 15 files changed, 656 insertions(+), 536 deletions(-) diff --git a/packages/adapter-utils/src/usage-checkpoint.test.ts b/packages/adapter-utils/src/usage-checkpoint.test.ts index 79d8167460c..2b622ef60f0 100644 --- a/packages/adapter-utils/src/usage-checkpoint.test.ts +++ b/packages/adapter-utils/src/usage-checkpoint.test.ts @@ -3,20 +3,24 @@ 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 createParser = () => { + let costUsd = 0; + return (stdout: string): AdapterUsageCheckpoint => { + const events = stdout.trim().split("\n").map(line => JSON.parse(line)); + costUsd += events.reduce((sum, event) => sum + event.usage.costUsd, 0); + return { costUsd, complete: events.some(event => event.type === "result") }; + }; }; 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 saved = vi.fn(); const parser = vi.fn(createParser()); 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); + const second = createUsageCheckpointLog(vi.fn(), saved, createParser()); await second("stdout", event + "\n"); await second.flush({ complete: true }); expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true }); @@ -28,7 +32,7 @@ describe("usage checkpoint stream", () => { 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 log = createUsageCheckpointLog(output, saved, createParser()); 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, }); @@ -38,19 +42,32 @@ describe("usage checkpoint stream", () => { }); 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); + const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, createParser()); 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(); + it("persists each changed total and the final receipt before publishing its chunk", async () => { + const order: string[] = []; + const saved = vi.fn(async (receipt: AdapterUsageCheckpoint) => { order.push(`saved:${receipt.costUsd}:${receipt.complete}`); }); + const log = createUsageCheckpointLog(async () => { order.push("log"); }, saved, createParser()); + await log("stdout", event + "\n"); + await log("stdout", event + "\n"); + await log("stdout", JSON.stringify({ type: "result", usage: { costUsd: 0 } }) + "\n"); + expect(order).toEqual(["saved:0.25:false", "log", "saved:0.5:false", "log", "saved:0.5:true", "log"]); + }); + + it("saves every new receipt before logging with linear parsing work", async () => { + const parser = vi.fn(createParser()); const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parser); - for (let i = 0; i < 1024; i++) await log("stdout", event + "\n"); + for (let i = 0; i < 1024; i++) { + await log("stdout", event + "\n"); + expect(saved.mock.calls.at(-1)![0].costUsd).toBe((i + 1) * 0.25); + } 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); + expect(saved).toHaveBeenCalledTimes(1025); // all updates plus completion + expect(parser).toHaveBeenCalledTimes(1024); + expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThanOrEqual(1024 * (event.length + 1)); }); }); diff --git a/packages/adapter-utils/src/usage-checkpoint.ts b/packages/adapter-utils/src/usage-checkpoint.ts index f7ebe701203..1e0c44bd86e 100644 --- a/packages/adapter-utils/src/usage-checkpoint.ts +++ b/packages/adapter-utils/src/usage-checkpoint.ts @@ -24,17 +24,18 @@ function project(value: unknown, depth = 0, field = ""): unknown { : project(entry, depth + 1, key)])); } -/** A fresh instance belongs to one CLI attempt. Call flush() after its process +/** The consumer incrementally parses only new records (never the full history). + * 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, + consume: (records: string) => AdapterUsageCheckpoint | null, ) { const attemptId = randomUUID(); let remainder = "", accounting = "", previous = ""; - let nextReplaySize = 0; + let snapshot: AdapterUsageCheckpoint | null = null; let failure: unknown; let failed = false; function retain(line: string) { @@ -43,14 +44,15 @@ export function createUsageCheckpointLog( 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"); + if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint chunk 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; + async function publish(complete = false) { + if (!onUsage) return; + if (accounting) { + snapshot = consume(accounting); + accounting = ""; + } + const parsed = snapshot; if (!parsed) return; if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null && !Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return; @@ -79,7 +81,7 @@ export function createUsageCheckpointLog( if (failed) throw failure; if (!onUsage) return; if (remainder) { retain(remainder); remainder = ""; } - await publish(true, options.complete ?? false); + await publish(options.complete ?? false); }, }); } diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 2f407d05364..fedcafe5132 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -65,7 +65,7 @@ import { import { claudeModelUsageTotals, claudeModelReceipts, - parseClaudeStreamJson, + parseClaudeStreamJson, createClaudeStreamParser, describeClaudeFailure, detectClaudeLoginRequired, extractClaudeRetryNotBefore, @@ -964,8 +964,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseClaudeStreamJson(stdout); + const parsed = consumeAccounting(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, @@ -989,7 +990,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise | null = null; const assistantTexts: string[] = []; const messageUsage = new Map(); let anonymousMessage = 0; + const observedTotals = { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) continue; - const event = parseJson(line); - if (!event) continue; + return (stdout: string) => { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; + const event = parseJson(line); + if (!event) continue; - const type = asString(event.type, ""); - if (type === "system" && asString(event.subtype, "") === "init") { - sessionId = asString(event.session_id, sessionId ?? "") || sessionId; - model = asString(event.model, model); - continue; - } - - if (type === "assistant") { - sessionId = asString(event.session_id, sessionId ?? "") || sessionId; - const message = parseObject(event.message); - const observed = parseObject(message.usage); - if (Object.keys(observed).length > 0) { - const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`; - const previous = messageUsage.get(key); - messageUsage.set(key, { - inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)), - cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)), - outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)), - }); + const type = asString(event.type, ""); + if (type === "system" && asString(event.subtype, "") === "init") { + sessionId = asString(event.session_id, sessionId ?? "") || sessionId; + model = asString(event.model, model); + continue; } - const content = Array.isArray(message.content) ? message.content : []; - for (const entry of content) { - if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue; - const block = entry as Record; - if (asString(block.type, "") === "text") { - const text = asString(block.text, ""); - if (text) assistantTexts.push(text); + + if (type === "assistant") { + sessionId = asString(event.session_id, sessionId ?? "") || sessionId; + const message = parseObject(event.message); + const observed = parseObject(message.usage); + if (Object.keys(observed).length > 0) { + const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`; + const previous = messageUsage.get(key); + const next = { + inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)), + cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)), + outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)), + }; + observedTotals.inputTokens += next.inputTokens - (previous?.inputTokens ?? 0); + observedTotals.outputTokens += next.outputTokens - (previous?.outputTokens ?? 0); + observedTotals.cachedInputTokens += next.cachedInputTokens - (previous?.cachedInputTokens ?? 0); + messageUsage.set(key, next); } + const content = Array.isArray(message.content) ? message.content : []; + for (const entry of content) { + if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue; + const block = entry as Record; + if (asString(block.type, "") === "text") { + const text = asString(block.text, ""); + if (text) assistantTexts.push(text); + } + } + continue; + } + + if (type === "result") { + finalResult = event; + sessionId = asString(event.session_id, sessionId ?? "") || sessionId; } - continue; } - if (type === "result") { - finalResult = event; - sessionId = asString(event.session_id, sessionId ?? "") || sessionId; + if (!finalResult) { + return { + sessionId, + model, + costUsd: null as number | null, + usage: messageUsage.size ? { ...observedTotals } : null, + usageBasis: "per_run" as const, + summary: assistantTexts.join("\n\n").trim(), + resultJson: null as Record | null, + }; } - } - if (!finalResult) { + const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage); + const usageObj = parseObject(finalResult.usage); + const usage: UsageSummary = modelUsageTotals ?? { + inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0), + cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0), + outputTokens: asNumber(usageObj.output_tokens, 0), + }; + const costRaw = finalResult.total_cost_usd; + const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null; + const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim(); + return { sessionId, model, - costUsd: null as number | null, - usage: messageUsage.size ? [...messageUsage.values()].reduce((total, part) => ({ - inputTokens: total.inputTokens + part.inputTokens, outputTokens: total.outputTokens + part.outputTokens, - cachedInputTokens: (total.cachedInputTokens ?? 0) + (part.cachedInputTokens ?? 0), - }), { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }) : null, + costUsd, + usage, + // modelUsage covers exactly this CLI invocation, so mark it per-run to + // keep the server from applying its session-cumulative delta heuristic. usageBasis: "per_run" as const, - summary: assistantTexts.join("\n\n").trim(), - resultJson: null as Record | null, + summary, + resultJson: finalResult, }; - } - - const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage); - const usageObj = parseObject(finalResult.usage); - const usage: UsageSummary = modelUsageTotals ?? { - inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0), - cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0), - outputTokens: asNumber(usageObj.output_tokens, 0), - }; - const costRaw = finalResult.total_cost_usd; - const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null; - const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim(); - - return { - sessionId, - model, - costUsd, - usage, - // modelUsage covers exactly this CLI invocation, so mark it per-run to - // keep the server from applying its session-cumulative delta heuristic. - usageBasis: "per_run" as const, - summary, - resultJson: finalResult, }; } diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index 2069e55f957..630e82fb7da 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -62,7 +62,7 @@ import { type LocalProcessSandboxOptions, } from "@paperclipai/adapter-utils/local-process-sandbox"; import { - parseCodexJsonl, + parseCodexJsonl, createCodexJsonlParser, classifyCodexAuthRefreshFailure, extractCodexRetryNotBefore, isCodexHarnessCrash, @@ -1319,8 +1319,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseCodexJsonl(stdout); + const parsed = consumeAccounting(stdout); return { usage: parsed.usage, usageBasis: "per_run", provider: "openai", biller: resolveCodexBiller(effectiveEnv, billingType), billingType, model, costUsd: null, complete: parsed.sawProtocolTerminalEvent }; }); try { @@ -1347,7 +1348,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const type = asString(event.type, ""); - if (type) sawProtocolEvent = true; - if (type === "error" || type === "turn.completed" || type === "turn.failed") { - sawProtocolTerminalEvent = true; - } - if (type === "thread.started") { - sessionId = asString(event.thread_id, sessionId ?? "") || sessionId; - continue; - } + const type = asString(event.type, ""); + if (type) sawProtocolEvent = true; + if (type === "error" || type === "turn.completed" || type === "turn.failed") { + sawProtocolTerminalEvent = true; + } + if (type === "thread.started") { + sessionId = asString(event.thread_id, sessionId ?? "") || sessionId; + continue; + } - if (type === "error") { - const msg = asString(event.message, "").trim(); - if (msg) errorMessage = msg; - continue; - } + if (type === "error") { + const msg = asString(event.message, "").trim(); + if (msg) errorMessage = msg; + continue; + } - if (type === "item.completed") { - const item = parseObject(event.item); - if (asString(item.type, "") === "agent_message") { - const text = asString(item.text, ""); - if (text) finalMessage = text; + if (type === "item.completed") { + const item = parseObject(event.item); + if (asString(item.type, "") === "agent_message") { + const text = asString(item.text, ""); + if (text) finalMessage = text; + } + continue; + } + + if (type === "turn.completed") { + const usageObj = parseObject(event.usage); + usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens); + usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens); + usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens); + continue; + } + + if (type === "turn.failed") { + const err = parseObject(event.error); + const msg = asString(err.message, "").trim(); + if (msg) errorMessage = msg; } - continue; } - if (type === "turn.completed") { - const usageObj = parseObject(event.usage); - usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens); - usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens); - usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens); - continue; - } - - if (type === "turn.failed") { - const err = parseObject(event.error); - const msg = asString(err.message, "").trim(); - if (msg) errorMessage = msg; - } - } - - return { - sessionId, - summary: finalMessage?.trim() ?? "", - // Codex includes cache hits in input_tokens; Paperclip stores them separately. - usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) }, - usageBasis: "per_run" as const, - errorMessage, - sawProtocolEvent, - sawProtocolTerminalEvent, + return { + sessionId, + summary: finalMessage?.trim() ?? "", + // Codex includes cache hits in input_tokens; Paperclip stores them separately. + usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) }, + usageBasis: "per_run" as const, + errorMessage, + sawProtocolEvent, + sawProtocolTerminalEvent, + }; }; } diff --git a/packages/adapters/cursor-local/src/server/execute.ts b/packages/adapters/cursor-local/src/server/execute.ts index 2c0b49dbd8c..2c1e6698c6d 100644 --- a/packages/adapters/cursor-local/src/server/execute.ts +++ b/packages/adapters/cursor-local/src/server/execute.ts @@ -54,7 +54,7 @@ import { joinPromptSections, } from "@paperclipai/adapter-utils/server-utils"; import { DEFAULT_CURSOR_LOCAL_MODEL, SANDBOX_INSTALL_COMMAND } from "../index.js"; -import { firstCursorDiagnosticLine, parseCursorJsonl, isCursorUnknownSessionError } from "./parse.js"; +import { firstCursorDiagnosticLine, parseCursorJsonl, createCursorJsonlParser, isCursorUnknownSessionError } from "./parse.js"; import { prepareCursorSandboxCommand } from "./remote-command.js"; import { normalizeCursorStreamLine } from "../shared/stream.js"; import { hasCursorTrustBypassArg } from "../shared/trust.js"; @@ -627,8 +627,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseCursorJsonl(stdout); + const parsed = consumeAccounting(stdout); const provider = resolveProviderFromModel(model); return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult }; }); @@ -676,7 +677,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise): string | null { } export function parseCursorJsonl(stdout: string) { + return createCursorJsonlParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createCursorJsonlParser() { let sessionId: string | null = null; let sawResult = false; const messages: string[] = []; @@ -80,94 +85,96 @@ export function parseCursorJsonl(stdout: string) { outputTokens: 0, }; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = normalizeCursorStreamLine(rawLine).line; - if (!line) continue; + return (stdout: string) => { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = normalizeCursorStreamLine(rawLine).line; + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const foundSession = readSessionId(event); - if (foundSession) sessionId = foundSession; + const foundSession = readSessionId(event); + if (foundSession) sessionId = foundSession; - const type = asString(event.type, "").trim(); + const type = asString(event.type, "").trim(); - if (type === "assistant") { - messages.push(...collectAssistantText(event.message)); - continue; - } - - if (type === "result") { - sawResult = true; - const usageObj = parseObject(event.usage); - usage.inputTokens += asNumber( - usageObj.input_tokens, - asNumber(usageObj.inputTokens, 0), - ); - usage.cachedInputTokens += asNumber( - usageObj.cached_input_tokens, - asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)), - ); - usage.outputTokens += asNumber( - usageObj.output_tokens, - asNumber(usageObj.outputTokens, 0), - ); - addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost); - - const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error"; - const resultText = asString(event.result, "").trim(); - if (resultText && messages.length === 0) { - messages.push(resultText); + if (type === "assistant") { + messages.push(...collectAssistantText(event.message)); + continue; } - if (isError) { - const resultError = asErrorText(event.error ?? event.message ?? event.result).trim(); - if (resultError) errorMessage = resultError; + + if (type === "result") { + sawResult = true; + const usageObj = parseObject(event.usage); + usage.inputTokens += asNumber( + usageObj.input_tokens, + asNumber(usageObj.inputTokens, 0), + ); + usage.cachedInputTokens += asNumber( + usageObj.cached_input_tokens, + asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)), + ); + usage.outputTokens += asNumber( + usageObj.output_tokens, + asNumber(usageObj.outputTokens, 0), + ); + addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost); + + const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error"; + const resultText = asString(event.result, "").trim(); + if (resultText && messages.length === 0) { + messages.push(resultText); + } + if (isError) { + const resultError = asErrorText(event.error ?? event.message ?? event.result).trim(); + if (resultError) errorMessage = resultError; + } + continue; } - continue; - } - if (type === "error") { - const message = asErrorText(event.message ?? event.error ?? event.detail).trim(); - if (message) errorMessage = message; - continue; - } - - if (type === "system") { - const subtype = asString(event.subtype, "").trim().toLowerCase(); - if (subtype === "error") { + if (type === "error") { const message = asErrorText(event.message ?? event.error ?? event.detail).trim(); if (message) errorMessage = message; + continue; + } + + if (type === "system") { + const subtype = asString(event.subtype, "").trim().toLowerCase(); + if (subtype === "error") { + const message = asErrorText(event.message ?? event.error ?? event.detail).trim(); + if (message) errorMessage = message; + } + continue; + } + + // Compatibility with older stream-json shapes. + if (type === "text") { + const part = parseObject(event.part); + const text = asString(part.text, "").trim(); + if (text) messages.push(text); + continue; + } + + if (type === "step_finish") { + const part = parseObject(event.part); + const tokens = parseObject(part.tokens); + const cache = parseObject(tokens.cache); + usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); + usage.cachedInputTokens += asNumber(cache.read, 0); + usage.outputTokens += asNumber(tokens.output, 0); + addCost(part.cost); + continue; } - continue; } - // Compatibility with older stream-json shapes. - if (type === "text") { - const part = parseObject(event.part); - const text = asString(part.text, "").trim(); - if (text) messages.push(text); - continue; - } - - if (type === "step_finish") { - const part = parseObject(event.part); - const tokens = parseObject(part.tokens); - const cache = parseObject(tokens.cache); - usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); - usage.cachedInputTokens += asNumber(cache.read, 0); - usage.outputTokens += asNumber(tokens.output, 0); - addCost(part.cost); - continue; - } - } - - return { - sawResult, - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd: reportedCost && !missingCost ? totalCostUsd : null, - errorMessage, + return { + sawResult, + sessionId, + summary: messages.join("\n\n").trim(), + usage: { ...usage }, + costUsd: reportedCost && !missingCost ? totalCostUsd : null, + errorMessage, + }; }; } diff --git a/packages/adapters/gemini-local/src/server/execute.ts b/packages/adapters/gemini-local/src/server/execute.ts index 500d395a819..3e4b19b9ac3 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -63,7 +63,7 @@ import { isGeminiTransientNetworkError, isGeminiTurnLimitResult, isGeminiSessionUnrecoverableError, - parseGeminiJsonl, + parseGeminiJsonl, createGeminiJsonlParser, } from "./parse.js"; import { firstNonEmptyLine } from "./utils.js"; import { @@ -643,8 +643,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseGeminiJsonl(stdout); + const parsed = consumeAccounting(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, { @@ -659,7 +660,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const foundSessionId = readSessionId(event); - if (foundSessionId) sessionId = foundSessionId; + const foundSessionId = readSessionId(event); + if (foundSessionId) sessionId = foundSessionId; - const type = asString(event.type, "").trim(); + const type = asString(event.type, "").trim(); - if (type === "assistant") { - messages.push(...collectMessageText(event.message)); - const messageObj = parseObject(event.message); - const content = Array.isArray(messageObj.content) ? messageObj.content : []; - for (const partRaw of content) { - const part = parseObject(partRaw); - if (asString(part.type, "").trim() === "question") { - question = { - prompt: asString(part.prompt, "").trim(), - choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => { - const choice = parseObject(choiceRaw); - return { - key: asString(choice.key, "").trim(), - label: asString(choice.label, "").trim(), - description: asString(choice.description, "").trim() || undefined, - }; - }), - }; - break; // only one question per message + if (type === "assistant") { + messages.push(...collectMessageText(event.message)); + const messageObj = parseObject(event.message); + const content = Array.isArray(messageObj.content) ? messageObj.content : []; + for (const partRaw of content) { + const part = parseObject(partRaw); + if (asString(part.type, "").trim() === "question") { + question = { + prompt: asString(part.prompt, "").trim(), + choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => { + const choice = parseObject(choiceRaw); + return { + key: asString(choice.key, "").trim(), + label: asString(choice.label, "").trim(), + description: asString(choice.description, "").trim() || undefined, + }; + }), + }; + break; // only one question per message + } } + continue; } - continue; - } - // Gemini CLI v0.38+ stream-json schema emits assistant turns as: - // {"type":"message","role":"assistant","content":"...","delta":true} - // These are discrete final messages (one per assistant turn), not - // cumulative streaming tokens, so collecting all of them produces the - // expected concatenated turn-by-turn summary rather than duplicated text. - if (type === "message") { - const role = asString(event.role, "").trim().toLowerCase(); - if (role === "assistant") { - messages.push(...collectMessageText(event.content)); + // Gemini CLI v0.38+ stream-json schema emits assistant turns as: + // {"type":"message","role":"assistant","content":"...","delta":true} + // These are discrete final messages (one per assistant turn), not + // cumulative streaming tokens, so collecting all of them produces the + // expected concatenated turn-by-turn summary rather than duplicated text. + if (type === "message") { + const role = asString(event.role, "").trim().toLowerCase(); + if (role === "assistant") { + messages.push(...collectMessageText(event.content)); + } + continue; } - continue; - } - if (type === "result") { - resultEvent = event; - accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats); - const resultText = - asString(event.result, "").trim() || - asString(event.text, "").trim() || - asString(event.response, "").trim(); - if (resultText && messages.length === 0) messages.push(resultText); - const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); - if (typeof reportedCost === "number") costUsd = reportedCost; - const status = asString(event.status, "").toLowerCase(); - const isError = - event.is_error === true || - asString(event.subtype, "").toLowerCase() === "error" || - status === "error" || - status === "failed"; - if (isError) { - const text = asErrorText(event.error ?? event.message ?? event.result).trim(); - if (text) errorMessage = text; + if (type === "result") { + resultEvent = event; + accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats); + const resultText = + asString(event.result, "").trim() || + asString(event.text, "").trim() || + asString(event.response, "").trim(); + if (resultText && messages.length === 0) messages.push(resultText); + const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); + if (typeof reportedCost === "number") costUsd = reportedCost; + const status = asString(event.status, "").toLowerCase(); + const isError = + event.is_error === true || + asString(event.subtype, "").toLowerCase() === "error" || + status === "error" || + status === "failed"; + if (isError) { + const text = asErrorText(event.error ?? event.message ?? event.result).trim(); + if (text) errorMessage = text; + } + continue; } - continue; - } - if (type === "error") { - const text = asErrorText(event.error ?? event.message ?? event.detail).trim(); - if (text) errorMessage = text; - continue; - } - - if (type === "system") { - const subtype = asString(event.subtype, "").trim().toLowerCase(); - if (subtype === "error") { + if (type === "error") { const text = asErrorText(event.error ?? event.message ?? event.detail).trim(); if (text) errorMessage = text; + continue; + } + + if (type === "system") { + const subtype = asString(event.subtype, "").trim().toLowerCase(); + if (subtype === "error") { + const text = asErrorText(event.error ?? event.message ?? event.detail).trim(); + if (text) errorMessage = text; + } + continue; + } + + if (type === "text") { + const part = parseObject(event.part); + const text = asString(part.text, "").trim(); + if (text) messages.push(text); + continue; + } + + if (type === "step_finish" || event.usage || event.usageMetadata) { + accumulateUsage(usage, event.usage ?? event.usageMetadata); + const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); + if (typeof reportedCost === "number") costUsd = reportedCost; + continue; } - continue; } - if (type === "text") { - const part = parseObject(event.part); - const text = asString(part.text, "").trim(); - if (text) messages.push(text); - continue; - } - - if (type === "step_finish" || event.usage || event.usageMetadata) { - accumulateUsage(usage, event.usage ?? event.usageMetadata); - const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0); - if (typeof reportedCost === "number") costUsd = reportedCost; - continue; - } - } - - return { - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd, - errorMessage, - resultEvent, - question, + return { + sessionId, + summary: messages.join("\n\n").trim(), + usage: { ...usage }, + costUsd, + errorMessage, + resultEvent, + question, + }; }; } diff --git a/packages/adapters/opencode-local/src/server/execute.ts b/packages/adapters/opencode-local/src/server/execute.ts index 6576bafe37a..72dd350f6b4 100644 --- a/packages/adapters/opencode-local/src/server/execute.ts +++ b/packages/adapters/opencode-local/src/server/execute.ts @@ -53,7 +53,7 @@ import { readPaperclipIssueWorkModeFromContext, resolveLegacyPaperclipDesiredSkillNames, } from "@paperclipai/adapter-utils/server-utils"; -import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl } from "./parse.js"; +import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl, createOpenCodeJsonlParser } from "./parse.js"; import { ensureOpenCodeModelConfiguredAndAvailable, isTruthyEnvFlag, @@ -646,8 +646,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parseOpenCodeJsonl(stdout); + const parsed = consumeAccounting(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 }; }); @@ -664,7 +665,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const currentSessionId = asString(event.sessionID, "").trim(); - if (currentSessionId) sessionId = currentSessionId; + const currentSessionId = asString(event.sessionID, "").trim(); + if (currentSessionId) sessionId = currentSessionId; - const type = asString(event.type, ""); + const type = asString(event.type, ""); - if (type === "text") { - const part = parseObject(event.part); - const text = asString(part.text, "").trim(); - if (text) messages.push(text); - continue; - } + if (type === "text") { + const part = parseObject(event.part); + const text = asString(part.text, "").trim(); + if (text) messages.push(text); + continue; + } - if (type === "step_finish") { - const part = parseObject(event.part); - const tokens = parseObject(part.tokens); - const cache = parseObject(tokens.cache); - usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); - usage.cachedInputTokens += asNumber(cache.read, 0); - usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0); - if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost; - else missingCost = true; - continue; - } + if (type === "step_finish") { + const part = parseObject(event.part); + const tokens = parseObject(part.tokens); + const cache = parseObject(tokens.cache); + usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0); + usage.cachedInputTokens += asNumber(cache.read, 0); + usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0); + if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost; + else missingCost = true; + continue; + } - if (type === "tool_use") { - const part = parseObject(event.part); - const state = parseObject(part.state); - if (asString(state.status, "") === "error") { - const text = asString(state.error, "").trim(); - if (text) toolErrors.push(text); + if (type === "tool_use") { + const part = parseObject(event.part); + const state = parseObject(part.state); + if (asString(state.status, "") === "error") { + const text = asString(state.error, "").trim(); + if (text) toolErrors.push(text); + } + continue; + } + + if (type === "error") { + const text = errorText(event.error ?? event.message).trim(); + if (text) errors.push(text); + continue; } - continue; } - if (type === "error") { - const text = errorText(event.error ?? event.message).trim(); - if (text) errors.push(text); - continue; - } - } - - return { - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd: missingCost ? null : costUsd, - errorMessage: errors.length > 0 ? errors.join("\n") : null, - toolErrors, + return { + sessionId, + summary: messages.join("\n\n").trim(), + usage: { ...usage }, + costUsd: missingCost ? null : costUsd, + errorMessage: errors.length > 0 ? errors.join("\n") : null, + toolErrors, + }; }; } diff --git a/packages/adapters/pi-local/src/server/execute.ts b/packages/adapters/pi-local/src/server/execute.ts index 3a07b333e4a..0274d2810e5 100644 --- a/packages/adapters/pi-local/src/server/execute.ts +++ b/packages/adapters/pi-local/src/server/execute.ts @@ -55,7 +55,7 @@ import { runChildProcess, } from "@paperclipai/adapter-utils/server-utils"; import { shellQuote } from "@paperclipai/adapter-utils/ssh"; -import { isPiUnknownSessionError, parsePiJsonl } from "./parse.js"; +import { isPiUnknownSessionError, parsePiJsonl, createPiJsonlParser } from "./parse.js"; import { ensurePiModelConfiguredAndAvailable } from "./models.js"; import { preparePiRuntimeConfig } from "./runtime-config.js"; import { SANDBOX_INSTALL_COMMAND } from "../index.js"; @@ -737,8 +737,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - const parsed = parsePiJsonl(stdout); + const parsed = consumeAccounting(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, { @@ -753,7 +754,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line) continue; - const event = parseJson(line); - if (!event) continue; + const event = parseJson(line); + if (!event) continue; - const eventType = asString(event.type, ""); + const eventType = asString(event.type, ""); - // Pi can exit successfully after a provider failure. The terminal assistant - // message carries that failure in both message_end and turn_end envelopes. - const terminalMessages = eventType === "agent_end" - ? (Array.isArray(event.messages) ? event.messages : []) - : eventType === "message_end" || eventType === "turn_end" - ? [event.message] - : []; - for (const rawMessage of terminalMessages) { - const message = asRecord(rawMessage); - if (message?.role !== "assistant" || message.stopReason !== "error") continue; - const error = asString(message.errorMessage, "").trim() || "Pi provider request failed."; - if (!result.errors.includes(error)) result.errors.push(error); - } - - // RPC protocol messages - skip these (internal implementation detail) - if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") { - continue; - } - - // Agent lifecycle - if (eventType === "agent_start") { - continue; - } - - if (eventType === "agent_end") { - result.sawAgentEnd = true; - const messages = event.messages as Array> | undefined; - if (messages && messages.length > 0) { - const lastMessage = messages[messages.length - 1]; - if (lastMessage?.role === "assistant") { - const content = lastMessage.content as string | Array<{ type: string; text?: string }>; - result.finalMessage = extractTextContent(content); - } + // Pi can exit successfully after a provider failure. The terminal assistant + // message carries that failure in both message_end and turn_end envelopes. + const terminalMessages = eventType === "agent_end" + ? (Array.isArray(event.messages) ? event.messages : []) + : eventType === "message_end" || eventType === "turn_end" + ? [event.message] + : []; + for (const rawMessage of terminalMessages) { + const message = asRecord(rawMessage); + if (message?.role !== "assistant" || message.stopReason !== "error") continue; + const error = asString(message.errorMessage, "").trim() || "Pi provider request failed."; + if (!result.errors.includes(error)) result.errors.push(error); } - continue; - } - if (eventType === "auto_retry_end") { - const succeeded = event.success === true; - if (!succeeded) { - const finalError = asString(event.finalError, "").trim(); - result.errors.push(finalError || "Pi exhausted automatic retries without producing a response."); + // RPC protocol messages - skip these (internal implementation detail) + if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") { + continue; } - continue; - } - // Turn lifecycle - if (eventType === "turn_start") { - continue; - } - - if (eventType === "turn_end") { - const message = asRecord(event.message); - if (message) { - const content = message.content as string | Array<{ type: string; text?: string }>; - const text = extractTextContent(content); - if (text) { - result.finalMessage = text; - result.messages.push(text); - } - - // Extract usage and cost from assistant message - const usage = asRecord(message.usage); - if (usage) { - result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0); - result.usage.outputTokens += asNumber(usage.output, 0); - result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0); - - // Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.) - const cost = asRecord(usage.cost); - addCost(cost?.total); - } + // Agent lifecycle + if (eventType === "agent_start") { + continue; } - - // Tool results are in toolResults array - const toolResults = event.toolResults as Array> | undefined; - if (toolResults) { - for (const tr of toolResults) { - const toolCallId = asString(tr.toolCallId, ""); - const content = tr.content; - const isError = tr.isError === true; - - // Find matching tool call by toolCallId - const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); - if (existingCall) { - existingCall.result = typeof content === "string" ? content : JSON.stringify(content); - existingCall.isError = isError; + + if (eventType === "agent_end") { + result.sawAgentEnd = true; + const messages = event.messages as Array> | undefined; + if (messages && messages.length > 0) { + const lastMessage = messages[messages.length - 1]; + if (lastMessage?.role === "assistant") { + const content = lastMessage.content as string | Array<{ type: string; text?: string }>; + result.finalMessage = extractTextContent(content); } } + continue; } - continue; - } - // Message updates (streaming) - if (eventType === "message_update") { - const assistantEvent = asRecord(event.assistantMessageEvent); - if (assistantEvent) { - const msgType = asString(assistantEvent.type, ""); - if (msgType === "text_delta") { - const delta = asString(assistantEvent.delta, ""); - if (delta) { - // Append to last message or create new - if (result.messages.length === 0) { - result.messages.push(delta); - } else { - result.messages[result.messages.length - 1] += delta; + if (eventType === "auto_retry_end") { + const succeeded = event.success === true; + if (!succeeded) { + const finalError = asString(event.finalError, "").trim(); + result.errors.push(finalError || "Pi exhausted automatic retries without producing a response."); + } + continue; + } + + // Turn lifecycle + if (eventType === "turn_start") { + continue; + } + + if (eventType === "turn_end") { + const message = asRecord(event.message); + if (message) { + const content = message.content as string | Array<{ type: string; text?: string }>; + const text = extractTextContent(content); + if (text) { + result.finalMessage = text; + result.messages.push(text); + } + + // Extract usage and cost from assistant message + const usage = asRecord(message.usage); + if (usage) { + result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0); + result.usage.outputTokens += asNumber(usage.output, 0); + result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0); + + // Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.) + const cost = asRecord(usage.cost); + addCost(cost?.total); + } + } + + // Tool results are in toolResults array + const toolResults = event.toolResults as Array> | undefined; + if (toolResults) { + for (const tr of toolResults) { + const toolCallId = asString(tr.toolCallId, ""); + const content = tr.content; + const isError = tr.isError === true; + + // Find matching tool call by toolCallId + const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); + if (existingCall) { + existingCall.result = typeof content === "string" ? content : JSON.stringify(content); + existingCall.isError = isError; } } } + continue; } - continue; - } - if (eventType === "error") { - const message = asString(event.message, "").trim(); - if (message) { - result.errors.push(message); + // Message updates (streaming) + if (eventType === "message_update") { + const assistantEvent = asRecord(event.assistantMessageEvent); + if (assistantEvent) { + const msgType = asString(assistantEvent.type, ""); + if (msgType === "text_delta") { + const delta = asString(assistantEvent.delta, ""); + if (delta) { + // Append to last message or create new + if (result.messages.length === 0) { + result.messages.push(delta); + } else { + result.messages[result.messages.length - 1] += delta; + } + } + } + } + continue; } - continue; - } - // Tool execution - if (eventType === "tool_execution_start") { - const toolCallId = asString(event.toolCallId, ""); - const toolName = asString(event.toolName, ""); - const args = event.args; - currentToolCall = { toolCallId, toolName, args }; - result.toolCalls.push({ - toolCallId, - toolName, - args, - result: null, - isError: false, - }); - continue; - } - - if (eventType === "tool_execution_end") { - const toolCallId = asString(event.toolCallId, ""); - const toolName = asString(event.toolName, ""); - const toolResult = event.result; - const isError = event.isError === true; - - // Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool) - const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); - if (existingCall) { - existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult); - existingCall.isError = isError; + if (eventType === "error") { + const message = asString(event.message, "").trim(); + if (message) { + result.errors.push(message); + } + continue; } - currentToolCall = null; - continue; - } - // Usage tracking if available in the event (fallback for standalone usage events) - if (eventType === "usage" || event.usage) { - const usage = asRecord(event.usage); - if (usage) { - // Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens) - result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0); - result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0); - result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0); - - // Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format) - const cost = asRecord(usage.cost); - addCost(cost?.total ?? usage.costUsd); + // Tool execution + if (eventType === "tool_execution_start") { + const toolCallId = asString(event.toolCallId, ""); + const toolName = asString(event.toolName, ""); + const args = event.args; + currentToolCall = { toolCallId, toolName, args }; + result.toolCalls.push({ + toolCallId, + toolName, + args, + result: null, + isError: false, + }); + continue; + } + + if (eventType === "tool_execution_end") { + const toolCallId = asString(event.toolCallId, ""); + const toolName = asString(event.toolName, ""); + const toolResult = event.result; + const isError = event.isError === true; + + // Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool) + const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId); + if (existingCall) { + existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult); + existingCall.isError = isError; + } + currentToolCall = null; + continue; + } + + // Usage tracking if available in the event (fallback for standalone usage events) + if (eventType === "usage" || event.usage) { + const usage = asRecord(event.usage); + if (usage) { + // Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens) + result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0); + result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0); + result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0); + + // Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format) + const cost = asRecord(usage.cost); + addCost(cost?.total ?? usage.costUsd); + } } } - } - if (missingCost) result.usage.costUsd = null; - return result; + if (missingCost) result.usage.costUsd = null; + return { ...result, usage: { ...result.usage }, messages: [...result.messages], errors: [...result.errors], toolCalls: [...result.toolCalls] }; + }; } export function isPiUnknownSessionError(stdout: string, stderr: string): boolean { diff --git a/server/src/__tests__/adapter-cost-failure-paths.test.ts b/server/src/__tests__/adapter-cost-failure-paths.test.ts index 0ad2c8d0038..141dd5f16b7 100644 --- a/server/src/__tests__/adapter-cost-failure-paths.test.ts +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -50,8 +50,18 @@ describe("CLI adapter accounting on timeout", () => { 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 }); + const initial = await execute(context); + const complete = !["claude", "pi"].includes(fixture.name); + expect(initial.usageComplete).toBe(complete); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete }); + if (["claude", "pi", "opencode"].includes(fixture.name)) { + processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => { + await options.onLog("stdout", stdout).catch(() => {}); + return { exitCode: 1, signal: null, timedOut: false, stdout, stderr: "process failed", pid: 123, startedAt: new Date().toISOString() }; + }); + expect((await execute(context)).usageComplete).toBe(false); + expect(onUsage.mock.calls.at(-1)![0].complete).toBe(false); + } if (fixture.name === "claude") { const mixed = [ { type: "system", subtype: "init", model: "actual-model" }, @@ -97,3 +107,36 @@ describe("CLI adapter accounting on timeout", () => { expect(result.costUsd).toBe(fixture.price); }); }); + + +describe("incremental protocol accounting", () => { + it("keeps Claude message updates idempotent and preserves immutable snapshots", async () => { + const { createClaudeStreamParser } = await import("../../../packages/adapters/claude-local/src/server/parse.js"); + const consume = createClaudeStreamParser(); + consume(JSON.stringify({ type: "system", subtype: "init", model: "actual-model" })); + const message = (id: string, output: number) => JSON.stringify({ type: "assistant", message: { id, usage: { input_tokens: 2, output_tokens: output } } }); + const first = consume(message("one", 1)); + expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 }); + expect(consume(message("one", 3)).usage).toMatchObject({ inputTokens: 2, outputTokens: 3 }); + expect(consume(message("two", 2)).usage).toMatchObject({ inputTokens: 4, outputTokens: 5 }); + expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 }); + const final = consume(JSON.stringify({ type: "result", total_cost_usd: 1, usage: { input_tokens: 9, output_tokens: 8 } })); + expect(final).toMatchObject({ model: "actual-model", costUsd: 1, usage: { inputTokens: 9, outputTokens: 8 } }); + }); + it.each(cases.filter(item => !["claude", "kimi"].includes(item.name)))("updates $name totals once per newly received record", async fixture => { + const factories = { + codex: async () => (await import("../../../packages/adapters/codex-local/src/server/parse.js")).createCodexJsonlParser(), + cursor: async () => (await import("../../../packages/adapters/cursor-local/src/server/parse.js")).createCursorJsonlParser(), + gemini: async () => (await import("../../../packages/adapters/gemini-local/src/server/parse.js")).createGeminiJsonlParser(), + pi: async () => (await import("../../../packages/adapters/pi-local/src/server/parse.js")).createPiJsonlParser(), + opencode: async () => (await import("../../../packages/adapters/opencode-local/src/server/parse.js")).createOpenCodeJsonlParser(), + }; + const consume = await factories[fixture.name as keyof typeof factories](); + const line = JSON.stringify(fixture.event); + const first = consume(line); + const second = consume(line); + expect(first.usage).toMatchObject(fixture.tokens!); + const multiplier = fixture.name === "codex" ? 1 : 2; + expect(second.usage).toMatchObject(Object.fromEntries(Object.entries(fixture.tokens!).map(([key, count]) => [key, count * multiplier]))); + }); +});