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]))); + }); +});