From a85fa3af31b269dc5008aef6f03fd36dd8d4cc26 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Tue, 6 Oct 2026 11:28:19 -0700 Subject: [PATCH] fix: checkpoint adapter usage and preserve unknown prices Co-Authored-By: Paperclip --- .../src/acpx-engine/execute.test.ts | 126 ++++++- .../adapter-utils/src/acpx-engine/execute.ts | 22 ++ packages/adapter-utils/src/billing.test.ts | 10 + packages/adapter-utils/src/billing.ts | 12 +- packages/adapter-utils/src/index.ts | 1 + packages/adapter-utils/src/types.ts | 35 +- .../src/usage-checkpoint.test.ts | 96 +++++ .../adapter-utils/src/usage-checkpoint.ts | 124 +++++++ .../claude-local/src/server/execute.ts | 37 +- .../claude-local/src/server/parse.test.ts | 14 + .../adapters/claude-local/src/server/parse.ts | 146 +++++--- .../codex-local/src/server/execute.ts | 27 +- .../codex-local/src/server/parse.test.ts | 4 +- .../adapters/codex-local/src/server/parse.ts | 104 +++--- .../cursor-local/src/server/execute.test.ts | 30 ++ .../cursor-local/src/server/execute.ts | 26 +- .../adapters/cursor-local/src/server/parse.ts | 169 +++++---- .../gemini-local/src/server/execute.ts | 21 +- .../gemini-local/src/server/parse.test.ts | 40 ++- .../adapters/gemini-local/src/server/parse.ts | 229 ++++++------ .../kimi-local/src/server/execute.test.ts | 3 + .../adapters/kimi-local/src/server/execute.ts | 13 + .../opencode-local/src/server/execute.ts | 40 ++- .../opencode-local/src/server/parse.test.ts | 9 + .../opencode-local/src/server/parse.ts | 101 +++--- .../adapters/pi-local/src/server/execute.ts | 33 +- .../pi-local/src/server/parse.test.ts | 9 + .../adapters/pi-local/src/server/parse.ts | 330 +++++++++--------- .../adapter-cost-failure-paths.test.ts | 218 ++++++++++++ .../src/__tests__/codex-local-adapter.test.ts | 2 +- .../__tests__/cursor-local-adapter.test.ts | 9 + .../__tests__/cursor-local-execute.test.ts | 5 + .../environment-execution-target.test.ts | 17 + .../__tests__/gemini-local-adapter.test.ts | 2 +- server/src/adapters/process/execute.ts | 18 + .../services/environment-execution-target.ts | 3 + 36 files changed, 1551 insertions(+), 534 deletions(-) create mode 100644 packages/adapter-utils/src/usage-checkpoint.test.ts create mode 100644 packages/adapter-utils/src/usage-checkpoint.ts create mode 100644 server/src/__tests__/adapter-cost-failure-paths.test.ts diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 4296a110e79..bda5233b743 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -185,6 +185,7 @@ async function runExecutor( const meta: Record[] = []; const logs: Array<{ stream: string; text: string }> = []; const events: Array<{ eventType: string; payload?: Record }> = []; + const receipts: unknown[] = []; const execute = createAcpxEngineExecutor({ ...(options.prepareRemoteManagedHome ? { prepareRemoteManagedHome: options.prepareRemoteManagedHome } @@ -222,10 +223,11 @@ async function runExecutor( onEvent: async (event: { eventType: string; payload?: Record }) => { events.push(event); }, + onUsage: async (receipt: unknown) => { receipts.push(receipt); }, } as never); expect(result.exitCode).toBe(0); - return { logs, meta, events, runtimeOptions, configOptions, sessionInputs, turnInputs, result }; + return { logs, meta, events, receipts, runtimeOptions, configOptions, sessionInputs, turnInputs, result }; } // Under `vi.useFakeTimers()`, setup before `ensureSession` still performs real @@ -529,7 +531,7 @@ describe("shared ACPX engine runtime behavior", () => { }); it("sets Codex model, effort, and fast mode through CODEX_CONFIG without session config calls", async () => { - const { configOptions, meta } = await runExecutor({ + const { configOptions, meta, receipts, result } = await runExecutor({ agent: "codex", model: "gpt-5.6-sol", modelReasoningEffort: "high", @@ -543,6 +545,8 @@ describe("shared ACPX engine runtime behavior", () => { features: { fast_mode: true }, }); expect(configOptions).toEqual([]); + expect(result.pricingContext).toEqual({ serviceTier: "fast" }); + expect(receipts).toContainEqual(expect.objectContaining({ pricingContext: { serviceTier: "fast" }, complete: true })); expect(meta[0]?.commandNotes).toContain( "Requested ACPX model: gpt-5.6-sol (set via CODEX_CONFIG at startup).", ); @@ -1452,6 +1456,45 @@ describe("shared ACPX engine runtime behavior", () => { expect(statusLine?.text).toContain('"cost"'); }); + it.each(["checkpoint", "terminal"])("returns received usage after the %s receipt sink rejects and still closes the runtime", async (failedSave) => { + const root = await makeTempRoot(); + const close = vi.fn(async () => {}); + const cancel = vi.fn(async () => {}); + let reads = 0; + const execute = createAcpxEngineExecutor({ + resolveBillingIdentity: () => ({ provider: "anthropic", biller: "anthropic", billingType: "api" }), + createRuntime: () => ({ + ...buildRuntime(), + getStatus: async () => ++reads === 1 + ? { usage: { cost: { amount: 0.4, currency: "USD" } } } + : { usage: { cumulative: { inputTokens: 120, outputTokens: 4500, cachedReadTokens: 900, cachedWriteTokens: 30 }, cost: { amount: 1.15, currency: "USD" } } }, + startTurn: () => ({ + events: (async function* () { + yield { type: "status", text: "usage", tag: "usage_update", + breakdown: { inputTokens: 100, outputTokens: 4000, cachedReadTokens: 800, cachedWriteTokens: 20 }, cost: { amount: 1.1, currency: "USD" } }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), cancel, + }), close, + }) as never, + }); + const onUsage = vi.fn(async (receipt: { complete: boolean }) => { + if (failedSave === "checkpoint" || receipt.complete) throw new Error("Receipt storage unavailable"); + }); + const result = await execute({ runId: `failed-${failedSave}-save`, agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, onMeta: async () => {}, onLog: async () => {}, onUsage } as never); + expect(result).toMatchObject({ exitCode: 1, usageBasis: "per_run", usageComplete: failedSave === "terminal", + provider: "anthropic", biller: "anthropic", billingType: "api" }); + expect(result.errorMessage).toContain("Receipt storage unavailable"); + expect(result.usage).toEqual(failedSave === "terminal" + ? { inputTokens: 150, outputTokens: 4500, cachedInputTokens: 900 } + : { inputTokens: 120, outputTokens: 4000, cachedInputTokens: 800 }); + expect(result.costUsd).toBeCloseTo(failedSave === "terminal" ? 0.75 : 0.7); + expect(onUsage).toHaveBeenCalledTimes(failedSave === "terminal" ? 2 : 1); + expect(cancel).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalledTimes(1); + }); + it("falls back to usage_update events when the runtime lacks getStatus", async () => { const root = await makeTempRoot(); const stateDir = path.join(root, "state"); @@ -1501,6 +1544,59 @@ describe("shared ACPX engine runtime behavior", () => { expect(result.billingType).toBe("unknown"); }); + it.each(["failed", "cancelled"] as const)("keeps %s in-stream usage provisional without a final receipt", async (status) => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const execute = createAcpxEngineExecutor({ + createRuntime: () => ({ + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => ({ + events: (async function* () { + yield { + type: "status", + text: "usage", + tag: "usage_update", + cost: { amount: 0.31, currency: "USD" }, + breakdown: { inputTokens: 40, outputTokens: 700, cachedReadTokens: 60 }, + }; + + })(), + result: Promise.resolve({ status, error: { code: "INTERNAL", message: "fixture failure" } }), + cancel: async () => {}, + }), + close: async () => {}, + }) as never, + }); + + const onUsage = vi.fn(); + const result = await execute({ + runId: "run-usage-event-fallback", + agent: { + id: "agent-1", + companyId: "company-1", + }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async () => {}, + onMeta: async () => {}, + onUsage, + } as never); + + expect(result.exitCode).not.toBe(0); + expect(result.usageComplete).toBe(false); + expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: false, costUsd: 0.31 })); + expect(result.usage).toEqual({ inputTokens: 40, outputTokens: 700, cachedInputTokens: 60 }); + expect(result.usageBasis).toBe("per_run"); + expect(result.costUsd).toBeCloseTo(0.31); + expect(result.provider).toBe("acpx"); + expect(result.billingType).toBe("unknown"); + }); + it.skipIf(process.platform === "win32")("materializes ACPX Claude skills without symlinked descendants", async () => { const root = await makeTempRoot(); const skillRoot = path.join(root, "skills"); @@ -6243,9 +6339,10 @@ describe("ACPX engine run lifecycle corrections (F1: settle every failure after expect(stagingLocks.size).toBe(0); }); - it("test_prompt_build_failure_returns_error_result_with_phase_prepare_turn", async () => { + it.each(["prompt", "metadata", "dispatch"])("records whether provider work could have started on %s failure", async (failure) => { const root = await makeTempRoot(); const close = vi.fn(async () => {}); + const startTurn = vi.fn(() => { throw new Error("dispatch failed"); }); const execute = createAcpxEngineExecutor({ createRuntime: () => ({ @@ -6254,11 +6351,7 @@ describe("ACPX engine run lifecycle corrections (F1: settle every failure after agentSessionId: "agent-session", runtimeSessionName: "runtime-session", }), - startTurn: () => ({ - events: (async function* () {})(), - result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), - cancel: async () => {}, - }), + startTurn, close, }) as never, }); @@ -6269,7 +6362,8 @@ describe("ACPX engine run lifecycle corrections (F1: settle every failure after Object.defineProperty(context, "paperclipSessionHandoffMarkdown", { enumerable: false, get() { - throw new Error("prompt build boom"); + if (failure === "prompt") throw new Error("prompt build boom"); + return undefined; }, }); @@ -6280,11 +6374,19 @@ describe("ACPX engine run lifecycle corrections (F1: settle every failure after config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, context, onLog: async () => {}, - onMeta: async () => {}, + onMeta: async () => { if (failure === "metadata") throw new Error("metadata failed"); }, } as never); expect(result.exitCode).toBe(1); - expect(result.resultJson?.phase).toBe("prepare_turn"); + if (failure === "dispatch") { + expect(startTurn).toHaveBeenCalledTimes(1); + expect(result.executionRecovery).toBeUndefined(); + expect(result.usageComplete).toBe(false); + } else { + expect(result.resultJson?.phase).toBe("prepare_turn"); + expect(startTurn).not.toHaveBeenCalled(); + expect(result.executionRecovery).toEqual({ kind: "bootstrap", providerWorkStarted: false }); + } // The established runtime is closed on the pre-turn failure, so no session // leaks. expect(close).toHaveBeenCalledTimes(1); @@ -7242,6 +7344,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => // The lost channel overrides the nominally completed terminal to a failure. expect(result.exitCode).not.toBe(0); expect(result.errorCode).toBe("duplex_channel_lost"); + expect(result.usageComplete).toBe(false); // The message carries only the typed loss reason, not raw provider text. expect(result.errorMessage).toContain("provider_exit"); expect(result.resultJson).toMatchObject({ status: "failed" }); @@ -7261,6 +7364,7 @@ describe("ACPX engine sandbox bridge run-disposition seam (fail-closed)", () => expect(result.exitCode).toBe(0); expect(result.errorCode ?? null).toBeNull(); + expect(result.usageComplete).toBe(true); // The atomic settle step marked the orderly completion for the // success-eligible terminal. expect(fake.settleRunDisposition).toHaveBeenCalledTimes(1); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index 28e5b620259..3caec3364f3 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -3896,6 +3896,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { provider: billingIdentity?.provider ?? "acpx", ...(billingIdentity?.biller ? { biller: billingIdentity.biller } : {}), billingType: billingIdentity?.billingType ?? ("unknown" as const), + pricingContext: undefined as AdapterExecutionResult["pricingContext"], }; const warmIdleMs = asNumber(ctx.config.warmHandleIdleMs, DEFAULT_ACP_ENGINE_WARM_HANDLE_IDLE_MS); // The host run site owns the warm-handle store on this run. It operates over @@ -4200,6 +4201,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }), ); buildRuntimeSettled = true; + billingFields.pricingContext = prepared.fastMode ? { serviceTier: "fast" } : undefined; // Capture acquired resources before the cancellation boundary so the // normal settlement path also releases a just-completed build. releaseStagingLease = prepared.sessionStagingLeaseRelease; @@ -4705,8 +4707,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }; let eventBreakdown: AcpRuntimeUsageBreakdown | null = null; let eventCostUsd: number | null = null; + // Receipt persistence is fallible; retain the provider evidence before + // calling the sink so the failure result can still return it to the host. + let observedUsage: Pick = { usageComplete: false }; let lastRuntimeEventAt: number | null = null; let observedRuntimeEvents = 0; + let turnDispatchAttempted = false; // The turn-local state the sequence steps share. `promptBuild` sets the // prompt, `preTurnUsage` sets the pre-turn status, `turnStart` sets the // active turn, and `turnFinalize` reads all three. `activeTurn` is the run- @@ -4789,6 +4795,9 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }; const stepTurnStart = (signal: AbortSignal, startTimeoutMs: number | undefined): StartedTurn => { ctx.signal?.throwIfAborted(); + // After entering startTurn, a thrown error cannot prove the provider + // received no work. Only failures before this boundary are free to settle. + turnDispatchAttempted = true; const turn = runtime.startTurn({ handle: sessionHandle, text: runPrompt, @@ -4905,6 +4914,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { if (event.type === "status" && event.tag === "usage_update") { eventBreakdown = event.breakdown ?? eventBreakdown; eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd; + const checkpoint = summarizeAcpxTurnUsage({ preStatus: preTurnStatus, postStatus: null, eventBreakdown, eventCostUsd }); + observedUsage = { usage: checkpoint.usage ?? undefined, costUsd: checkpoint.costUsd, usageBasis: "per_run", usageComplete: false }; + await ctx.onUsage?.({ ...billingFields, usage: checkpoint.usage ?? undefined, costUsd: checkpoint.costUsd, + model: prepared.requestedModel, usageBasis: "per_run", complete: false }); } await emitRuntimeEvent(ctx, event, toolTitles, prepared.coalescePlaceholderToolUpdates); } @@ -4991,6 +5004,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { eventBreakdown, eventCostUsd, }); + // An interrupted stream is provisional unless the runtime supplies its + // post-terminal receipt. An in-stream usage_update alone cannot close it. + const usageComplete = !channelLost && (turnSucceeded || postTurnStatus?.usage != null); + observedUsage = { usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd, usageBasis: "per_run", usageComplete }; + await ctx.onUsage?.({ ...billingFields, usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd, + model: prepared.requestedModel, usageBasis: "per_run", complete: usageComplete }); const failedTurn = terminal.status === "failed" || terminal.status === "cancelled" || timedOut; // ACPX can defer session/load until runTurn. Forget an unavailable // session so the next bounded turn receives the full task conversation. @@ -5076,6 +5095,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { model: prepared.requestedModel || null, ...(turnUsage.usage ? { usage: turnUsage.usage, usageBasis: "per_run" as const } : {}), costUsd: turnUsage.costUsd, + usageComplete, resultJson: { status: channelLost ? "failed" : terminal.status, ...activityDiagnostics, @@ -5209,6 +5229,8 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { errorMessage: message, errorCode: timedOut ? "acpx_timeout" : (emitted?.classified.errorCode ?? null), errorMeta: emitted?.classified.errorMeta, + ...observedUsage, + ...(!turnDispatchAttempted ? { executionRecovery: { kind: "bootstrap" as const, providerWorkStarted: false as const } } : {}), ...billingFields, ...referencedProjectStagingFailuresField, model: prepared.requestedModel || null, diff --git a/packages/adapter-utils/src/billing.test.ts b/packages/adapter-utils/src/billing.test.ts index 1b85340f8f7..76e147c0deb 100644 --- a/packages/adapter-utils/src/billing.test.ts +++ b/packages/adapter-utils/src/billing.test.ts @@ -25,4 +25,14 @@ describe("inferOpenAiCompatibleBiller", () => { ), ).toBe("openai"); }); + + it.each(["https://proxy.example/v1", "https://openrouter.ai.example/v1", "https://example.org/openrouter.ai", "http://api.openai.com/v1", "invalid"])("does not apply direct provider prices to %s", (baseUrl) => { + expect(inferOpenAiCompatibleBiller({ OPENAI_BASE_URL: baseUrl }, "openai")).toBe("unknown"); + }); + + it("recognizes each supported base URL override", () => { + expect(inferOpenAiCompatibleBiller({ OPENAI_API_BASE: "https://proxy.example" })).toBe("unknown"); + expect(inferOpenAiCompatibleBiller({ OPENAI_API_BASE_URL: "https://proxy.example" })).toBe("unknown"); + expect(inferOpenAiCompatibleBiller({}, null)).toBeNull(); + }); }); diff --git a/packages/adapter-utils/src/billing.ts b/packages/adapter-utils/src/billing.ts index 3ecd8ec3594..b57ab8666ad 100644 --- a/packages/adapter-utils/src/billing.ts +++ b/packages/adapter-utils/src/billing.ts @@ -14,7 +14,17 @@ export function inferOpenAiCompatibleBiller( readEnv(env, "OPENAI_BASE_URL") ?? readEnv(env, "OPENAI_API_BASE") ?? readEnv(env, "OPENAI_API_BASE_URL"); - if (baseUrl && /openrouter\.ai/i.test(baseUrl)) return "openrouter"; + if (baseUrl) { + try { + const url = new URL(baseUrl); + if (url.protocol === "https:" && url.hostname === "openrouter.ai") return "openrouter"; + // An OpenAI-compatible endpoint does not imply OpenAI prices. Do not + // assign direct-provider estimates to a proxy or an unknown endpoint. + if (url.protocol !== "https:" || url.hostname !== "api.openai.com") return "unknown"; + } catch { + return "unknown"; + } + } return fallback; } diff --git a/packages/adapter-utils/src/index.ts b/packages/adapter-utils/src/index.ts index eb02fb4efc4..c14eebfa25e 100644 --- a/packages/adapter-utils/src/index.ts +++ b/packages/adapter-utils/src/index.ts @@ -2,6 +2,7 @@ export type { AdapterAgent, AdapterRuntime, UsageSummary, + AdapterUsageCheckpoint, AdapterBillingType, AdapterRuntimeServiceReport, AdapterExecutionResult, diff --git a/packages/adapter-utils/src/types.ts b/packages/adapter-utils/src/types.ts index c11343cae4e..5ba1f93fedf 100644 --- a/packages/adapter-utils/src/types.ts +++ b/packages/adapter-utils/src/types.ts @@ -31,9 +31,32 @@ export interface AdapterRuntime { // --------------------------------------------------------------------------- export interface UsageSummary { + /** Uncached input plus cache writes; excludes cache reads. Total tokens are input + cachedInput + output. */ inputTokens: number; outputTokens: number; cachedInputTokens?: number; + /** Subset of inputTokens, not an additional token count. */ + cacheWriteTokens?: number; +} + +/** Accounting-only snapshot. Never include prompt, response, or credentials. */ +export interface AdapterUsageCheckpoint { + attemptId?: string; + usage?: UsageSummary; + usageByModel?: Array<{ model: string; usage: UsageSummary; costUsd: number }>; + usageBasis?: "per_run" | "session_cumulative" | null; + provider?: string | null; + biller?: string | null; + model?: string | null; + billingType?: AdapterBillingType | null; + costUsd?: number | null; + costUsdExact?: string | null; + costStatus?: "reported" | "estimated" | "unpriced"; + pricingContext?: { serviceTier?: string; contextTier?: "short" | "long" }; + pricingProvenance?: { source: "provider_reported" | "provider_invoice" | "operator" | "rate_card" | "unknown"; version?: string; evidence?: string; inputCentsPerMillion?: string; cachedInputCentsPerMillion?: string; cacheWriteCentsPerMillion?: string; outputCentsPerMillion?: string; serviceTier?: string; contextTier?: "short" | "long" }; + cacheAdjustedCostUsd?: number | null; + providerRequestId?: string | null; + complete: boolean; } export type AdapterBillingType = @@ -93,11 +116,14 @@ export interface AdapterExecutionResult { retryNotBefore?: string | null; errorMeta?: Record; usage?: UsageSummary; + /** Complete per-model receipts, when supplied by the runtime. Their sums must match the run totals. */ + usageByModel?: Array<{ model: string; usage: UsageSummary; costUsd: number }>; /** * How `usage` totals are scoped. "per_run" means the tokens cover only this * execution; "session_cumulative" means they are running totals for the * persisted session, and the server must delta consecutive runs. Absent - * means unknown — the server applies its legacy session-delta heuristic. + * defaults to per-run; the server must never infer cumulative usage from a + * reused session ID. */ usageBasis?: "per_run" | "session_cumulative" | null; /** @@ -111,6 +137,12 @@ export interface AdapterExecutionResult { model?: string | null; billingType?: AdapterBillingType | null; costUsd?: number | null; + costUsdExact?: string | null; + costStatus?: "reported" | "estimated" | "unpriced"; + pricingContext?: { serviceTier?: string; contextTier?: "short" | "long" }; + pricingProvenance?: { source: "provider_reported" | "provider_invoice" | "operator" | "rate_card" | "unknown"; version?: string; evidence?: string; inputCentsPerMillion?: string; cachedInputCentsPerMillion?: string; cacheWriteCentsPerMillion?: string; outputCentsPerMillion?: string; serviceTier?: string; contextTier?: "short" | "long" }; + providerRequestId?: string | null; + usageComplete?: boolean; /** * Provider-billed cost after prompt-cache discounts. Adapters should set * this when they expose it separately; otherwise the server treats a @@ -232,6 +264,7 @@ export interface AdapterExecutionContext { onLog: (stream: "stdout" | "stderr", chunk: string) => Promise; onMeta?: (meta: AdapterInvocationMeta) => Promise; onEvent?: (event: AdapterRuntimeEvent) => Promise; + onUsage?: (receipt: AdapterUsageCheckpoint) => Promise; onRuntimeProgress?: RuntimeStatusSink; /** * Reports that execution has crossed the adapter's dispatch boundary. 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..a158f587c15 --- /dev/null +++ b/packages/adapter-utils/src/usage-checkpoint.test.ts @@ -0,0 +1,96 @@ +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 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(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, createParser()); + 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("bounds large content strings before buffering and preserves usage on either side", async () => { + const saved = vi.fn(); const parser = vi.fn(createParser()); const output = vi.fn(); + const log = createUsageCheckpointLog(output, saved, parser); + const line = JSON.stringify({ type: "result", usage: { inputTokens: 2, costUsd: 0.25 }, messages: [{ content: '"\\\n'.repeat(2 * 1024 * 1024) }], model: "after-large-content" }); + for (let offset = 0; offset < line.length; offset += 4093) await log("stdout", line.slice(offset, offset + 4093)); + await log.flush(); + expect(saved).toHaveBeenLastCalledWith(expect.objectContaining({ costUsd: 0.25, complete: true })); + expect(parser.mock.calls[0][0]).toContain("after-large-content"); + expect(parser.mock.calls[0][0].length).toBeLessThan(1000); + expect(output.mock.calls.map(call => call[1]).join("")).toBe(line); + }); + + it("preserves Unicode and escaped quotes across the string limit and chunk boundaries", async () => { + for (const suffix of ['\\u1234', '\\"', '\\\\']) { + const saved = vi.fn(); const parser = vi.fn(createParser()); + const log = createUsageCheckpointLog(vi.fn(), saved, parser); + const line = '{"type":"result","text":"' + 'a'.repeat(249) + suffix + 'discarded","usage":{"costUsd":0.25}}\n'; + for (const character of line) await log("stdout", character); + await log.flush(); + expect(saved).toHaveBeenCalledWith(expect.objectContaining({ costUsd: 0.25 })); + } + }); + + 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, 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, + }); + 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, createParser()); + await log("stdout", event); await log.flush(); + expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false }); + }); + + 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"); + 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(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 new file mode 100644 index 00000000000..123ae66615e --- /dev/null +++ b/packages/adapter-utils/src/usage-checkpoint.ts @@ -0,0 +1,124 @@ +import type { AdapterUsageCheckpoint } from "./types.js"; + +/** Project accounting protocol fields before retaining a stream snapshot. + * Content-bearing strings are never retained in the accounting buffer. */ +const fields = new Set([ + "type", "subtype", "role", "id", "session_id", "sessionID", "thread_id", "model", "modelID", "providerID", + "usage", "usageMetadata", "modelUsage", "message", "part", "info", "stats", "tokens", "cost", "total", + "input", "output", "cache", "read", "write", "input_tokens", "output_tokens", "cached_input_tokens", + "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", "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 { + if (depth > 8) return undefined; + if (typeof value === "number" || typeof value === "boolean" || value === null) return value; + if (typeof value === "string") return stringFields.has(field) ? value.slice(0, 250) : undefined; + if (Array.isArray(value)) return value.slice(0, 500).map(item => project(item, depth + 1)); + if (!value || typeof value !== "object") return undefined; + return Object.fromEntries(Object.entries(value).filter(([key]) => fields.has(key)) + .map(([key, entry]) => [key, key === "modelUsage" && entry && typeof entry === "object" + ? Object.fromEntries(Object.entries(entry).map(([model, usage]) => [model.slice(0, 250), project(usage, depth + 1)])) + : project(entry, depth + 1, key)])); +} + +/** Bound strings before buffering whole JSON lines. Provider records can carry + * many MiB of conversation text alongside a small usage object. Preserve JSON + * escapes across chunks so discarded text cannot corrupt the accounting fields. */ +function createStringCompactor() { + let inString = false, length = 0, escapeRemaining = 0; + let escapeCode = false, keepEscape = false; + return (chunk: string) => { + let output = ""; + for (const character of chunk) { + if (character === "\n") { + inString = false; escapeRemaining = 0; escapeCode = false; + output += character; + } else if (!inString) { + output += character; + if (character === '"') { inString = true; length = 0; } + } else if (escapeRemaining > 0) { + if (keepEscape) output += character; + escapeRemaining = escapeCode && character === "u" ? 4 : escapeRemaining - 1; + escapeCode = false; + length++; + } else if (character === '"') { + output += character; inString = false; + } else if (character === "\\") { + keepEscape = length < 250; + if (keepEscape) output += character; + escapeRemaining = 1; escapeCode = true; length++; + } else { + if (length < 250) output += character; + length++; + } + } + return output; + }; +} + +/** 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, + consume: (records: string) => AdapterUsageCheckpoint | null, +) { + const attemptId = randomUUID(); + const compactStrings = createStringCompactor(); + let remainder = "", accounting = "", previous = ""; + let snapshot: AdapterUsageCheckpoint | null = null; + 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 chunk exceeds 8 MiB"); + } + 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; + 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 += compactStrings(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(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 114cad4fa43..fedcafe5132 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -1,4 +1,5 @@ import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; +import { createUsageCheckpointLog } from "@paperclipai/adapter-utils/usage-checkpoint"; import fs from "node:fs/promises"; import path from "node:path"; import { fileURLToPath } from "node:url"; @@ -63,7 +64,8 @@ import { } from "@paperclipai/adapter-utils/local-process-sandbox"; import { claudeModelUsageTotals, - parseClaudeStreamJson, + claudeModelReceipts, + parseClaudeStreamJson, createClaudeStreamParser, describeClaudeFailure, detectClaudeLoginRequired, extractClaudeRetryNotBefore, @@ -962,6 +964,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + 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, + usageByModel: claudeModelReceipts(parsed.resultJson?.modelUsage), complete: parsed.resultJson !== null }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -971,7 +981,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`, errorCode: "timeout", errorMeta, @@ -1073,6 +1093,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise 1 ? "mixed" : parsedStream.model || model, + usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage), + billingType, + costUsd: parsedStream.costUsd, errorMessage: fallbackErrorMessage, errorCode, errorFamily, @@ -1235,19 +1264,21 @@ export async function execute(ctx: AdapterExecutionContext): Promise 1 ? "mixed" : parsedStream.model || asString(parsed.model, model), billingType, costUsd: parsedStream.costUsd, resultJson: mergedResultJson, diff --git a/packages/adapters/claude-local/src/server/parse.test.ts b/packages/adapters/claude-local/src/server/parse.test.ts index 5b291d4056e..b84651f051d 100644 --- a/packages/adapters/claude-local/src/server/parse.test.ts +++ b/packages/adapters/claude-local/src/server/parse.test.ts @@ -567,3 +567,17 @@ describe("parseClaudeStreamJson usage extraction", () => { expect(parsed.usageBasis).toBe("per_run"); }); }); + +describe("interrupted Claude accounting", () => { + it("retains partial usage and replaces repeated message snapshots", () => { + const events = [ + { type: "assistant", message: { id: "m1", usage: { input_tokens: 10, cache_creation_input_tokens: 20, cache_read_input_tokens: 40, output_tokens: 3 } } }, + { type: "assistant", message: { id: "m1", usage: { input_tokens: 10, cache_creation_input_tokens: 20, cache_read_input_tokens: 40, output_tokens: 7 } } }, + { type: "assistant", message: { id: "m2", usage: { input_tokens: 5, output_tokens: 2 } } }, + ]; + const parsed = parseClaudeStreamJson(events.map((event) => JSON.stringify(event)).join("\n")); + expect(parsed.usage).toEqual({ inputTokens: 35, cachedInputTokens: 40, outputTokens: 9 }); + expect(parsed.usageBasis).toBe("per_run"); + expect(parsed.costUsd).toBeNull(); + }); +}); diff --git a/packages/adapters/claude-local/src/server/parse.ts b/packages/adapters/claude-local/src/server/parse.ts index 3e636ac2cfa..8004885c7b5 100644 --- a/packages/adapters/claude-local/src/server/parse.ts +++ b/packages/adapters/claude-local/src/server/parse.ts @@ -54,79 +54,115 @@ export function claudeModelUsageTotals(modelUsage: unknown): UsageSummary | null return { inputTokens, outputTokens, cachedInputTokens }; } +export function claudeModelReceipts(modelUsage: unknown) { + const entries = Object.entries(parseObject(modelUsage)); + if (!entries.length) return undefined; + const receipts = entries.map(([model, raw]) => { + const entry = parseObject(raw); + const usage = claudeModelUsageTotals({ [model]: entry }); + const costUsd = entry.costUSD; + return usage && typeof costUsd === "number" && Number.isFinite(costUsd) && costUsd >= 0 ? { model, usage, costUsd } : null; + }); + return receipts.every((part) => part !== null) ? receipts : undefined; +} + export function parseClaudeStreamJson(stdout: string) { + return createClaudeStreamParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createClaudeStreamParser() { let sessionId: string | null = null; let model = ""; let finalResult: Record | 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; - } + 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 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: null as UsageSummary | null, - usageBasis: null as "per_run" | null, - summary: assistantTexts.join("\n\n").trim(), - resultJson: null as Record | 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, + resultJson: finalResult, }; - } - - const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage); - const usageObj = parseObject(finalResult.usage); - const usage: UsageSummary = modelUsageTotals ?? { - inputTokens: asNumber(usageObj.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 fe2fa38b588..8ef3c6bb383 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1,4 +1,5 @@ import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; +import { createUsageCheckpointLog } from "@paperclipai/adapter-utils/usage-checkpoint"; import fs from "node:fs/promises"; import path from "node:path"; import { fileURLToPath } from "node:url"; @@ -61,7 +62,7 @@ import { type LocalProcessSandboxOptions, } from "@paperclipai/adapter-utils/local-process-sandbox"; import { - parseCodexJsonl, + parseCodexJsonl, createCodexJsonlParser, classifyCodexAuthRefreshFailure, extractCodexRetryNotBefore, isCodexHarnessCrash, @@ -1228,6 +1229,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = consumeAccounting(stdout); + return { usage: parsed.usage, usageBasis: "per_run", provider: "openai", biller: resolveCodexBiller(effectiveEnv, billingType), billingType, model, pricingContext, costUsd: null, complete: parsed.sawProtocolTerminalEvent }; + }); try { const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), @@ -1331,7 +1338,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { monitor?.noteOutputChunk(stream, chunk); if (stream === "stdout") { - await onLog(stream, chunk); + await accountingLog(stream, chunk); return; } const cleaned = stripCodexRolloutNoise(chunk); @@ -1342,6 +1349,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise; + pricingContext?: AdapterExecutionResult["pricingContext"]; monitor?: | { fired: false } | { fired: true; terminationSignal: NodeJS.Signals | null; elapsedMsSinceLastEvent: number; timeoutMs: number }; @@ -1391,6 +1401,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { sessionId: "thread_123", summary: "Recovered response", usage: { - inputTokens: 10, + inputTokens: 8, cachedInputTokens: 2, outputTokens: 4, }, @@ -64,7 +64,7 @@ describe("parseCodexJsonl", () => { sessionId: "thread_123", summary: "Fixed the issue and verified the targeted tests pass.", usage: { - inputTokens: 10, + inputTokens: 8, cachedInputTokens: 2, outputTokens: 4, }, diff --git a/packages/adapters/codex-local/src/server/parse.ts b/packages/adapters/codex-local/src/server/parse.ts index 7698977f409..a2e9dd812a2 100644 --- a/packages/adapters/codex-local/src/server/parse.ts +++ b/packages/adapters/codex-local/src/server/parse.ts @@ -28,6 +28,11 @@ export type CodexAuthRefreshFailureClass = | "refresh_token_invalidated"; export function parseCodexJsonl(stdout: string) { + return createCodexJsonlParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createCodexJsonlParser() { let sessionId: string | null = null; let finalMessage: string | null = null; let errorMessage: string | null = null; @@ -39,61 +44,64 @@ export function parseCodexJsonl(stdout: string) { outputTokens: 0, }; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) 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 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() ?? "", - usage, - 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.test.ts b/packages/adapters/cursor-local/src/server/execute.test.ts index 6aea2414d3d..7628a53624a 100644 --- a/packages/adapters/cursor-local/src/server/execute.test.ts +++ b/packages/adapters/cursor-local/src/server/execute.test.ts @@ -138,6 +138,36 @@ function createFreshLeaseSandboxRunner(options: { } describe("cursor execute", () => { + it.each([0, 7])("settles legacy step usage only after a clean process exit (%s)", async (exitCode) => { + setPrepareCursorSandboxCommand.mockReset(); + setPrepareCursorSandboxCommand.mockImplementation(async (input) => ({ + command: input.command, env: input.env, remoteSystemHomeDir: null, + addedPathEntry: null, preferredCommandPath: null, + })); + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-cursor-legacy-")); + const command = path.join(root, "agent.sh"); + await fs.writeFile(command, `#!/bin/sh +cat >/dev/null +printf '%s\\n' '{"type":"step_finish","part":{"tokens":{"input":20,"output":5},"cost":0.01}}' +exit ${exitCode} +`, { mode: 0o755 }); + const onUsage = vi.fn(); + try { + const result = await execute({ + runId: "run-legacy", agent: { id: "agent-1", companyId: "company-1", name: "Cursor", adapterType: "cursor", adapterConfig: {} }, + runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, + config: { command, cwd: root }, context: createPromptContextFixture(), + authToken: "fixture-run-token", onLog: async () => {}, onUsage, + }); + expect(result.usage).toMatchObject({ inputTokens: 20, outputTokens: 5 }); + expect(result.costUsd).toBe(0.01); + expect(result.usageComplete).toBe(exitCode === 0); + expect(onUsage).toHaveBeenCalledWith(expect.objectContaining({ complete: false, costUsd: 0.01 })); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } + }); + it.each([ { detail: "Authentication failed", structured: "", expected: "Authentication failed" }, { detail: "", structured: "", expected: "Cursor exited with code 7" }, diff --git a/packages/adapters/cursor-local/src/server/execute.ts b/packages/adapters/cursor-local/src/server/execute.ts index cfaa9949a9e..88a9c8ca273 100644 --- a/packages/adapters/cursor-local/src/server/execute.ts +++ b/packages/adapters/cursor-local/src/server/execute.ts @@ -1,4 +1,5 @@ import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; +import { createUsageCheckpointLog } from "@paperclipai/adapter-utils/usage-checkpoint"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -53,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"; @@ -626,11 +627,17 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + 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 }; + }); let stdoutLineBuffer = ""; const emitNormalizedStdoutLine = async (rawLine: string) => { const normalized = normalizeCursorStreamLine(rawLine); if (!normalized.line) return; - await onLog(normalized.stream ?? "stdout", `${normalized.line}\n`); + await accountingLog(normalized.stream ?? "stdout", `${normalized.line}\n`); }; const flushStdoutChunk = async (chunk: string, finalize = false) => { const combined = `${stdoutLineBuffer}${chunk}`; @@ -670,6 +677,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const usageComplete = attempt.parsed.sawResult || (attempt.parsed.sawLegacyStep + && attempt.proc.exitCode === 0 && !attempt.proc.signal && !attempt.proc.timedOut + && !attempt.parsed.errorMessage); if (attempt.proc.timedOut) { return { exitCode: attempt.proc.exitCode, signal: attempt.proc.signal, timedOut: true, + usageComplete, + usage: attempt.parsed.usage, + usageBasis: "per_run", + provider: providerFromModel, + biller: resolveCursorBiller(effectiveEnv, + billingType, providerFromModel), + model, + billingType, + costUsd: attempt.parsed.costUsd, errorMessage: `Timed out after ${timeoutSec}s`, clearSession: clearSessionOnMissingSession, }; @@ -729,6 +749,8 @@ 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; + let sawLegacyStep = false; const messages: string[] = []; let errorMessage: string | null = null; let totalCostUsd = 0; + let reportedCost = false; + let missingCost = false; + const addCost = (value: unknown) => { + if (typeof value === "number" && Number.isFinite(value) && value >= 0) { + totalCostUsd += value; + reportedCost = true; + } else missingCost = true; + }; const usage = { inputTokens: 0, cachedInputTokens: 0, 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") { - 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), - ); - totalCostUsd += asNumber(event.total_cost_usd, asNumber(event.cost_usd, asNumber(event.cost, 0))); - - 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") { + sawLegacyStep = true; + 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); - usage.cachedInputTokens += asNumber(cache.read, 0); - usage.outputTokens += asNumber(tokens.output, 0); - totalCostUsd += asNumber(part.cost, 0); - continue; - } - } - - return { - sessionId, - summary: messages.join("\n\n").trim(), - usage, - costUsd: totalCostUsd > 0 ? totalCostUsd : null, - errorMessage, + return { + sawResult, + sawLegacyStep, + 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 720c384d433..3e4b19b9ac3 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -1,4 +1,5 @@ import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; +import { createUsageCheckpointLog } from "@paperclipai/adapter-utils/usage-checkpoint"; import fs from "node:fs/promises"; import type { Dirent } from "node:fs"; import os from "node:os"; @@ -62,7 +63,7 @@ import { isGeminiTransientNetworkError, isGeminiTurnLimitResult, isGeminiSessionUnrecoverableError, - parseGeminiJsonl, + parseGeminiJsonl, createGeminiJsonlParser, } from "./parse.js"; import { firstNonEmptyLine } from "./utils.js"; import { @@ -642,6 +643,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + 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, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -650,10 +656,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { expect(result.summary).toBe("hello."); expect(result.sessionId).toBe("session-abc"); expect(result.errorMessage).toBeNull(); - expect(result.usage.inputTokens).toBe(9095); - expect(result.usage.outputTokens).toBe(29); + expect(result.usage.inputTokens).toBe(963); + expect(result.usage.outputTokens).toBe(373); expect(result.usage.cachedInputTokens).toBe(8132); }); @@ -213,3 +214,38 @@ describe("isGeminiTransientNetworkError", () => { ).toBe(false); }); }); + + +describe("Gemini accounting semantics", () => { + it.each([ + [0, "result"], [1.25, "result"], [0, "step_finish"], [1.25, "step_finish"], [0, "usage"], [1.25, "usage"], + ] as const)("invalidates an earlier $%s price when a later %s adds unpriced usage", (cost, type) => { + const consume = createGeminiJsonlParser(); + const priced = JSON.stringify({ type: "step_finish", usage: { input_tokens: 10, output_tokens: 2 }, cost }); + const unpriced = JSON.stringify({ type, usage: { input_tokens: 20, output_tokens: 3 } }); + expect(consume(priced).costUsd).toBe(cost); + const parsed = consume(unpriced); + expect(parsed.usage).toEqual({ inputTokens: 30, cachedInputTokens: 0, outputTokens: 5 }); + expect(parsed.costUsd).toBeNull(); + expect(parseGeminiJsonl(`${priced}\n${unpriced}`)).toEqual(parsed); + }); + it.each([0, 2.5])("accepts a final reported total of $%s after earlier unpriced usage", (cost) => { + const consume = createGeminiJsonlParser(); + expect(consume(JSON.stringify({ type: "step_finish", usage: { input_tokens: 10 } })).costUsd).toBeNull(); + expect(consume(JSON.stringify({ type: "result", total_cost_usd: cost })).costUsd).toBe(cost); + }); + it("retains a reported price when later events add no usage", () => { + const consume = createGeminiJsonlParser(); + consume(JSON.stringify({ type: "step_finish", usage: { input_tokens: 10 }, cost: 1.25 })); + expect(consume(JSON.stringify({ type: "message", role: "assistant", content: "Done" })).costUsd).toBe(1.25); + expect(consume(JSON.stringify({ type: "result", status: "success" })).costUsd).toBe(1.25); + }); + it("keeps cache reads disjoint and includes thinking tokens in API metadata", () => { + const parsed = parseGeminiJsonl(JSON.stringify({ type: "result", usageMetadata: { promptTokenCount: 100, cachedContentTokenCount: 80, candidatesTokenCount: 10, thoughtsTokenCount: 5, totalTokenCount: 115 } })); + expect(parsed.usage).toEqual({ inputTokens: 20, cachedInputTokens: 80, outputTokens: 15 }); + }); + it("preserves explicit zero and distinguishes absent prices", () => { + expect(parseGeminiJsonl(JSON.stringify({ type: "result", total_cost_usd: 0 })).costUsd).toBe(0); + expect(parseGeminiJsonl(JSON.stringify({ type: "result" })).costUsd).toBeNull(); + }); +}); diff --git a/packages/adapters/gemini-local/src/server/parse.ts b/packages/adapters/gemini-local/src/server/parse.ts index 008f6522d05..87a5a63969c 100644 --- a/packages/adapters/gemini-local/src/server/parse.ts +++ b/packages/adapters/gemini-local/src/server/parse.ts @@ -58,24 +58,27 @@ function accumulateUsage( const usageMetadata = parseObject(usage.usageMetadata); const source = Object.keys(usageMetadata).length > 0 ? usageMetadata : usage; - target.inputTokens += asNumber( - source.input_tokens, - asNumber(source.inputTokens, asNumber(source.promptTokenCount, 0)), - ); - target.cachedInputTokens += asNumber( - source.cached_input_tokens, - asNumber( - source.cachedInputTokens, - asNumber(source.cachedContentTokenCount, asNumber(source.cached, 0)), - ), - ); - target.outputTokens += asNumber( - source.output_tokens, - asNumber(source.outputTokens, asNumber(source.candidatesTokenCount, 0)), - ); + // Gemini CLI's input_tokens is the full prompt; input is uncached input. + // https://github.com/google-gemini/gemini-cli/blob/main/packages/core/src/output/stream-json-formatter.ts + const cached = asNumber(source.cached_input_tokens, + asNumber(source.cachedInputTokens, asNumber(source.cachedContentTokenCount, asNumber(source.cached, 0)))); + const prompt = asNumber(source.input_tokens, asNumber(source.inputTokens, asNumber(source.promptTokenCount, 0))); + const toolInput = asNumber(source.toolUsePromptTokenCount, 0); + target.inputTokens += Math.max(0, asNumber(source.input, prompt - cached)) + toolInput; + target.cachedInputTokens += cached; + // The streaming CLI omits a separate thought count, but includes it in total_tokens. + const response = asNumber(source.output_tokens, asNumber(source.outputTokens, asNumber(source.candidatesTokenCount, 0))); + const total = asNumber(source.total_tokens, asNumber(source.totalTokenCount, 0)); + target.outputTokens += Math.max(response + asNumber(source.thoughtsTokenCount, 0), total - prompt - toolInput); + } export function parseGeminiJsonl(stdout: string) { + return createGeminiJsonlParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createGeminiJsonlParser() { let sessionId: string | null = null; const messages: string[] = []; let errorMessage: string | null = null; @@ -87,115 +90,127 @@ export function parseGeminiJsonl(stdout: string) { cachedInputTokens: 0, outputTokens: 0, }; + const recordAccounting = (event: Record, usageRaw: unknown) => { + const before = { ...usage }; + accumulateUsage(usage, usageRaw); + 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; + else if (usage.inputTokens !== before.inputTokens || usage.cachedInputTokens !== before.cachedInputTokens || usage.outputTokens !== before.outputTokens) { + // An earlier price, including explicit zero, cannot cover new unpriced + // usage. Text-only or terminal events with no new usage leave it intact. + costUsd = null; + } + }; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) 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 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); - costUsd = asNumber(event.total_cost_usd, asNumber(event.cost_usd, asNumber(event.cost, costUsd ?? 0))) || costUsd; - 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; + recordAccounting(event, 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 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) { + recordAccounting(event, event.usage ?? event.usageMetadata); + 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); - costUsd = asNumber(event.total_cost_usd, asNumber(event.cost_usd, asNumber(event.cost, costUsd ?? 0))) || costUsd; - 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/kimi-local/src/server/execute.test.ts b/packages/adapters/kimi-local/src/server/execute.test.ts index ea899c98cd3..93ea4dc2b15 100644 --- a/packages/adapters/kimi-local/src/server/execute.test.ts +++ b/packages/adapters/kimi-local/src/server/execute.test.ts @@ -326,6 +326,9 @@ describe("kimi_local execute", () => { const result = await execute(makeContext(root, { config: { cwd: root, timeoutSec: 5 } })); expect(result.timedOut).toBe(true); + expect(result.usageComplete).toBe(true); + expect(result.costStatus).toBe("unpriced"); + expect(result.costUsd).toBeNull(); expect(result.errorMessage).toContain("5s"); }); diff --git a/packages/adapters/kimi-local/src/server/execute.ts b/packages/adapters/kimi-local/src/server/execute.ts index b02ca0c02da..19be487cacd 100644 --- a/packages/adapters/kimi-local/src/server/execute.ts +++ b/packages/adapters/kimi-local/src/server/execute.ts @@ -648,6 +648,16 @@ export async function execute(ctx: AdapterExecutionContext): Promise, provider: string | null): string { - return inferOpenAiCompatibleBiller(env, null) ?? provider ?? "unknown"; + return provider === "openai" ? inferOpenAiCompatibleBiller(env, "openai") ?? "unknown" : provider ?? "unknown"; } const REMOTE_OPENCODE_MODELS_PROBE_DEFAULT_TIMEOUT_SEC = 20; @@ -645,6 +646,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise {}), stdout => { + hasAccounting = true; + 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 }; + }); const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -654,15 +663,20 @@ export async function execute(ctx: AdapterExecutionContext): Promise { expect(isOpenCodeUnknownSessionError("all good", "")).toBe(false); }); }); + +describe("OpenCode price availability", () => { + it("distinguishes missing cost from reported zero and counts cache writes", () => { + const event = { type: "step_finish", part: { tokens: { input: 10, output: 3, cache: { read: 100, write: 20 } } } }; + expect(parseOpenCodeJsonl(JSON.stringify(event))).toMatchObject({ costUsd: null, usage: { inputTokens: 30, cachedInputTokens: 100, outputTokens: 3 } }); + expect(parseOpenCodeJsonl(JSON.stringify({ ...event, part: { ...event.part, cost: 0 } })).costUsd).toBe(0); + expect(parseOpenCodeJsonl([event, { ...event, part: { ...event.part, cost: 1 } }].map((row) => JSON.stringify(row)).join("\n")).costUsd).toBeNull(); + }); +}); diff --git a/packages/adapters/opencode-local/src/server/parse.ts b/packages/adapters/opencode-local/src/server/parse.ts index ee7fd41279b..eb36576063a 100644 --- a/packages/adapters/opencode-local/src/server/parse.ts +++ b/packages/adapters/opencode-local/src/server/parse.ts @@ -20,6 +20,11 @@ function errorText(value: unknown): string { } export function parseOpenCodeJsonl(stdout: string) { + return createOpenCodeJsonlParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createOpenCodeJsonlParser() { let sessionId: string | null = null; const messages: string[] = []; const errors: string[] = []; @@ -29,62 +34,66 @@ export function parseOpenCodeJsonl(stdout: string) { cachedInputTokens: 0, outputTokens: 0, }; - let costUsd = 0; + let costUsd: number | null = null; + let missingCost = false; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) 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 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); - usage.cachedInputTokens += asNumber(cache.read, 0); - usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0); - costUsd += asNumber(part.cost, 0); - 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, - 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 3149642b323..f978290e4cc 100644 --- a/packages/adapters/pi-local/src/server/execute.ts +++ b/packages/adapters/pi-local/src/server/execute.ts @@ -1,4 +1,5 @@ import { createProviderStoppedBoundary } from "@paperclipai/adapter-utils/provider-stopped-boundary"; +import { createUsageCheckpointLog } from "@paperclipai/adapter-utils/usage-checkpoint"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -54,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"; @@ -141,7 +142,7 @@ async function buildPiSkillsDir(config: Record): Promise, provider: string | null): string { - return inferOpenAiCompatibleBiller(env, null) ?? provider ?? "unknown"; + return provider === "openai" ? inferOpenAiCompatibleBiller(env, "openai") ?? "unknown" : provider ?? "unknown"; } async function ensureSessionsDir(): Promise { @@ -736,6 +737,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise {}), stdout => { + hasAccounting = true; + 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, { onProcessStopped: providerStop.beginInvocation(), cwd, @@ -744,21 +752,26 @@ export async function execute(ctx: AdapterExecutionContext): Promise { .toEqual(["Pi provider request failed."]); }); }); + +describe("Pi price availability", () => { + it("retains unpriced usage and counts cache writes without treating a missing price as free", () => { + const event = { type: "turn_end", message: { role: "assistant", content: [], usage: { input: 10, output: 3, cacheRead: 100, cacheWrite: 20 } } }; + expect(parsePiJsonl(JSON.stringify(event)).usage).toEqual({ inputTokens: 30, cachedInputTokens: 100, outputTokens: 3, costUsd: null }); + event.message.usage = { ...event.message.usage, cost: { total: 0 } } as typeof event.message.usage; + expect(parsePiJsonl(JSON.stringify(event)).usage.costUsd).toBe(0); + }); +}); diff --git a/packages/adapters/pi-local/src/server/parse.ts b/packages/adapters/pi-local/src/server/parse.ts index b4d769e1596..f0911d78d1a 100644 --- a/packages/adapters/pi-local/src/server/parse.ts +++ b/packages/adapters/pi-local/src/server/parse.ts @@ -1,6 +1,7 @@ import { asNumber, asString, parseJson, parseObject } from "@paperclipai/adapter-utils/server-utils"; interface ParsedPiOutput { + sawAgentEnd: boolean; sessionId: string | null; messages: string[]; errors: string[]; @@ -8,7 +9,7 @@ interface ParsedPiOutput { inputTokens: number; outputTokens: number; cachedInputTokens: number; - costUsd: number; + costUsd: number | null; }; finalMessage: string | null; toolCalls: Array<{ toolCallId: string; toolName: string; args: unknown; result: string | null; isError: boolean }>; @@ -28,8 +29,14 @@ function extractTextContent(content: string | Array<{ type: string; text?: strin .join(""); } -export function parsePiJsonl(stdout: string): ParsedPiOutput { +export function parsePiJsonl(stdout: string) { + return createPiJsonlParser()(stdout); +} + +/** Consume complete JSONL records once, retaining protocol accounting state. */ +export function createPiJsonlParser() { const result: ParsedPiOutput = { + sawAgentEnd: false, sessionId: null, messages: [], errors: [], @@ -37,198 +44,201 @@ export function parsePiJsonl(stdout: string): ParsedPiOutput { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0, - costUsd: 0, + costUsd: null, }, finalMessage: null, toolCalls: [], }; + let missingCost = false; + function addCost(value: unknown) { + if (typeof value === "number" && Number.isFinite(value) && value >= 0) result.usage.costUsd = (result.usage.costUsd ?? 0) + value; + else missingCost = true; + } let currentToolCall: { toolCallId: string; toolName: string; args: unknown } | null = null; - for (const rawLine of stdout.split(/\r?\n/)) { - const line = rawLine.trim(); - if (!line) 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 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") { - 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; - } + // Agent lifecycle + if (eventType === "agent_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); - 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); - if (cost) { - result.usage.costUsd += asNumber(cost.total, 0); + 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; } - - // 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 === "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); } } - } - 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; + // 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); - 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); - if (cost) { - result.usage.costUsd += asNumber(cost.total ?? usage.costUsd, 0); - } else { - result.usage.costUsd += asNumber(usage.costUsd, 0); + // 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); } } } - } - 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 new file mode 100644 index 00000000000..ba463537c92 --- /dev/null +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -0,0 +1,218 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { AdapterExecutionContext, AdapterExecutionResult } from "@paperclipai/adapter-utils"; + +const processResult = vi.hoisted(() => vi.fn()); +vi.mock("@paperclipai/adapter-utils/execution-target", async (original) => ({ + ...await original(), + runAdapterExecutionTargetProcess: processResult, +})); + +type Execute = (context: AdapterExecutionContext) => Promise; +const cases: Array<{ name: string; execute: () => Promise; event: unknown; tokens?: { inputTokens: number; outputTokens: number; cachedInputTokens: number }; price: number | null }> = [ + { name: "codex", execute: async () => (await import("../../../packages/adapters/codex-local/src/server/execute.js")).execute, + event: { type: "turn.completed", usage: { input_tokens: 120, cached_input_tokens: 100, output_tokens: 10 } }, tokens: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 }, price: null }, + { name: "claude", execute: async () => (await import("../../../packages/adapters/claude-local/src/server/execute.js")).execute, + event: { type: "assistant", message: { id: "m1", usage: { input_tokens: 20, cache_read_input_tokens: 100, output_tokens: 10 }, content: [] } }, tokens: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 }, price: null }, + { name: "opencode", execute: async () => (await import("../../../packages/adapters/opencode-local/src/server/execute.js")).execute, + event: { type: "step_finish", part: { tokens: { input: 20, output: 10, cache: { read: 100 } }, cost: 0.004 } }, tokens: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 }, price: 0.004 }, + { name: "pi", execute: async () => (await import("../../../packages/adapters/pi-local/src/server/execute.js")).execute, + event: { type: "turn_end", message: { role: "assistant", content: [], usage: { input: 20, output: 10, cacheRead: 100, cost: { total: 0.004 } } } }, tokens: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 }, price: 0.004 }, + { name: "cursor", execute: async () => (await import("../../../packages/adapters/cursor-local/src/server/execute.js")).execute, + event: { type: "result", usage: { input_tokens: 20, cache_read_input_tokens: 100, output_tokens: 10 }, total_cost_usd: 0.004 }, tokens: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 }, price: 0.004 }, + { name: "gemini", execute: async () => (await import("../../../packages/adapters/gemini-local/src/server/execute.js")).execute, + event: { type: "result", usage: { input_tokens: 120, cached_input_tokens: 100, output_tokens: 10 }, total_cost_usd: 0.004 }, tokens: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 }, price: 0.004 }, + { name: "kimi", execute: async () => (await import("../../../packages/adapters/kimi-local/src/server/execute.js")).execute, + event: { type: "assistant", content: "partial work" }, price: null }, +]; + +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(); + 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 (fixture.name === "codex") { + for (const safer of [false, true]) { + const priced = await execute({ ...context, + config: { ...context.config, model: "gpt-5.4", fastMode: true }, + context: safer ? { codexTransientFallbackMode: "fresh_session_safer_invocation" } : {}, + }); + const expected = safer ? undefined : { serviceTier: "fast" }; + expect(priced.pricingContext).toEqual(expected); + expect(onUsage.mock.calls.at(-1)![0].pricingContext).toEqual(expected); + } + } + 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" }, + { 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.filter(f => ["pi", "opencode"].includes(f.name)))("$name preserves full-stream accounting after stdout truncation", async (fixture) => { + const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-tail-")); directories.push(dir); + const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'anthropic test\\n'\n", { mode: 0o755 }); + const execute = await fixture.execute(); + for (const checkpoint of [true, false]) for (const timedOut of [false, true]) for (const missingPrice of [false, true]) { + const early = structuredClone(fixture.event) as any; + if (missingPrice) { + if (fixture.name === "pi") delete early.message.usage.cost; + else delete early.part.cost; + } + const end = fixture.name === "pi" && !timedOut ? '\n{"type":"agent_end","messages":[]}' : ""; + const first = JSON.stringify(early) + "\n"; + const noise = JSON.stringify({ type: "text", part: { text: "x".repeat(9 * 1024 * 1024) } }) + "\n"; + const last = JSON.stringify(fixture.event) + end; + processResult.mockImplementation(async (_run, _target, _command, _args, options) => { + for (const record of [first, noise, last]) { + for (let offset = 0; offset < record.length; offset += 65521) await options.onLog("stdout", record.slice(offset, offset + 65521)); + } + return { exitCode: timedOut ? null : 0, signal: timedOut ? "SIGTERM" : null, timedOut, stdout: (first + noise + last).slice(-4 * 1024 * 1024), stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const onUsage = checkpoint ? vi.fn() : undefined; + const result = await execute({ + runId: "tail-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: "anthropic/test", paperclipRuntimeSkills: [], env: { ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } }, + context: {}, onLog: async () => {}, onUsage, + }); + expect(result.usage).toMatchObject({ inputTokens: 40, outputTokens: 20, cachedInputTokens: 200 }); + expect(result.costUsd).toBe(missingPrice ? null : 0.008); + expect(result.biller).toBe("anthropic"); + expect(result.usageComplete).toBe(!timedOut); + if (onUsage) expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: result.usage, costUsd: result.costUsd }); + } + }); + + 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 }); + processResult.mockResolvedValue({ exitCode: null, signal: "SIGTERM", timedOut: true, stdout: JSON.stringify(fixture.event), stderr: "", pid: 123, startedAt: new Date().toISOString() }); + const execute = await fixture.execute(); + const result = await execute({ + 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", MOONSHOT_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, + context: {}, onLog: async () => {}, + }); + expect(result.timedOut).toBe(true); + expect(result.usageComplete).toBe(["codex", "cursor", "gemini", "kimi"].includes(fixture.name)); + if (fixture.name === "kimi") expect(result.costStatus).toBe("unpriced"); + expect(result.usageBasis).toBe("per_run"); + expect(result.provider).toBeTruthy(); + expect(result.billingType).toBeTruthy(); + if (fixture.tokens) expect(result.usage).toMatchObject(fixture.tokens); + 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]))); + }); +}); + + +describe("process adapter bootstrap accounting proof", () => { + const context = (config: Record): AdapterExecutionContext => ({ + runId: `bootstrap-${crypto.randomUUID()}`, + agent: { id: "agent", companyId: "company", name: "Process", adapterType: "process", adapterConfig: {} }, + runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, + config, context: {}, onLog: async () => {}, + }); + it.each([{}, { command: "/nonexistent-paperclip-accounting-command" }, + { command: process.execPath, cwd: "/nonexistent-paperclip-accounting-directory" }])( + "proves no provider work for a failure before spawn: %j", async (config) => { + const { execute } = await import("../adapters/process/execute.js"); + const onSpawn = vi.fn(); + const result = await execute({ ...context(config), onSpawn }); + expect(result).toMatchObject({ exitCode: 1, executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }); + expect(result.errorMessage).toBeTruthy(); + expect(onSpawn).not.toHaveBeenCalled(); + }, + ); + it("does not invent bootstrap proof when a spawned process fails", async () => { + const { execute } = await import("../adapters/process/execute.js"); + const onSpawn = vi.fn(async () => { throw new Error("Metadata persistence failed"); }); + const result = await execute({ ...context({ command: process.execPath, args: ["-e", "process.exit(7)"] }), onSpawn }); + expect(onSpawn).toHaveBeenCalledOnce(); + expect(result.exitCode).toBe(7); + expect(result.executionRecovery).toBeUndefined(); + }); +}); diff --git a/server/src/__tests__/codex-local-adapter.test.ts b/server/src/__tests__/codex-local-adapter.test.ts index 84c806cd7a0..d1eb49ff817 100644 --- a/server/src/__tests__/codex-local-adapter.test.ts +++ b/server/src/__tests__/codex-local-adapter.test.ts @@ -16,7 +16,7 @@ describe("codex_local parser", () => { expect(parsed.sessionId).toBe("thread-123"); expect(parsed.summary).toBe("hello"); expect(parsed.usage).toEqual({ - inputTokens: 10, + inputTokens: 8, cachedInputTokens: 2, outputTokens: 4, }); diff --git a/server/src/__tests__/cursor-local-adapter.test.ts b/server/src/__tests__/cursor-local-adapter.test.ts index a1949af19c3..808acad3ba8 100644 --- a/server/src/__tests__/cursor-local-adapter.test.ts +++ b/server/src/__tests__/cursor-local-adapter.test.ts @@ -403,3 +403,12 @@ describe("cursor cli formatter", () => { } }); }); + + +describe("Cursor price completeness", () => { + it("preserves reported zero and keeps a partially priced stream unpriced", () => { + expect(parseCursorJsonl(JSON.stringify({ type: "result", cost_usd: 0 })).costUsd).toBe(0); + expect(parseCursorJsonl(JSON.stringify({ type: "result" })).costUsd).toBeNull(); + expect(parseCursorJsonl([JSON.stringify({ type: "step_finish", part: { cost: 1 } }), JSON.stringify({ type: "step_finish", part: {} })].join("\n")).costUsd).toBeNull(); + }); +}); diff --git a/server/src/__tests__/cursor-local-execute.test.ts b/server/src/__tests__/cursor-local-execute.test.ts index ac0c8480611..ac240f10eae 100644 --- a/server/src/__tests__/cursor-local-execute.test.ts +++ b/server/src/__tests__/cursor-local-execute.test.ts @@ -88,6 +88,11 @@ function createLocalSandboxRunner() { onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; }) => { counter += 1; + // Both sandbox fixtures install their fake CLI up front. Never fall back + // to a network install if discovery regresses on another host or shell. + if (input.args?.some(arg => arg.includes("curl https://cursor.com/install"))) { + throw new Error("Sandbox fixture did not discover its preinstalled Cursor CLI"); + } return await runChildProcess(`cursor-sandbox-execute-${counter}`, input.command, input.args ?? [], { cwd: input.cwd ?? process.cwd(), env: input.env ?? {}, diff --git a/server/src/__tests__/environment-execution-target.test.ts b/server/src/__tests__/environment-execution-target.test.ts index b9d96aabc34..d06f505ef62 100644 --- a/server/src/__tests__/environment-execution-target.test.ts +++ b/server/src/__tests__/environment-execution-target.test.ts @@ -713,6 +713,23 @@ describe("resolveEnvironmentExecutionTarget", () => { expect(delivered).toEqual([["stdout", "full"]]); }); + it("does not replay a capped provider tail after streaming more than 4 MiB", async () => { + const receipt = JSON.stringify({ type: "turn_end", message: { usage: { input: 20, output: 10, cost: { total: 0.004 } } } }) + "\n"; + const output = "x".repeat(5 * 1024 * 1024) + "\n" + receipt; + const { tracer } = createRecordingExecTracer(); + const runner = await runnerWithExecute({ provider: "createos", tracer, + execute: async (input: unknown) => { + const typed = input as { onLog?: (s: "stdout" | "stderr", c: string) => Promise }; + for (let offset = 0; offset < output.length; offset += 65536) await typed.onLog?.("stdout", output.slice(offset, offset + 65536)); + await typed.onLog?.("stderr", "diagnostic prefix\nlast warning\n"); + return { exitCode: 0, signal: null, timedOut: false, stdout: output.slice(-4 * 1024 * 1024), stderr: "last warning\n" }; + }, + }); + const delivered = await runExecuteCollectingLogs(runner as { execute(input: unknown): Promise }); + expect(delivered.filter(([stream]) => stream === "stdout").map(([, chunk]) => chunk).join("")).toBe(output); + expect(delivered.filter(([stream]) => stream === "stderr")).toEqual([["stderr", "diagnostic prefix\nlast warning\n"]]); + }); + it("delivers the full captured output when the provider streams nothing incrementally", async () => { // The provider streams no incremental chunk, so the whole final result is // the suffix and reaches the sink once. diff --git a/server/src/__tests__/gemini-local-adapter.test.ts b/server/src/__tests__/gemini-local-adapter.test.ts index bc2d4643d8b..860605516ce 100644 --- a/server/src/__tests__/gemini-local-adapter.test.ts +++ b/server/src/__tests__/gemini-local-adapter.test.ts @@ -34,7 +34,7 @@ describe("gemini_local parser", () => { expect(parsed.sessionId).toBe("gemini-session-1"); expect(parsed.summary).toBe("hello"); expect(parsed.usage).toEqual({ - inputTokens: 12, + inputTokens: 9, cachedInputTokens: 3, outputTokens: 7, }); diff --git a/server/src/adapters/process/execute.ts b/server/src/adapters/process/execute.ts index 0f08381574f..f9b03a6d1d7 100644 --- a/server/src/adapters/process/execute.ts +++ b/server/src/adapters/process/execute.ts @@ -18,6 +18,24 @@ import { } from "../utils.js"; export async function execute(ctx: AdapterExecutionContext): Promise { + let spawned = false; + try { + return await executeProcess({ ...ctx, onSpawn: async (meta) => { + // Set this before metadata persistence: a failed callback cannot prove + // that the already-created process did no billable work. + spawned = true; + await ctx.onSpawn?.(meta); + } }); + } catch (error) { + if (spawned) throw error; + return { exitCode: 1, signal: null, timedOut: false, + errorMessage: error instanceof Error ? error.message : String(error), + executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + }; + } +} + +async function executeProcess(ctx: AdapterExecutionContext): Promise { const { runId, agent, config, onLog, onMeta, authToken } = ctx; const command = asString(config.command, ""); if (!command) throw new Error("Process adapter missing command"); diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index d1aefd34f20..d719535f032 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -125,6 +125,9 @@ function undeliveredSuffix(delivered: string, final: string): string { if (!final) return ""; if (delivered.length === 0) return final; if (final.startsWith(delivered)) return final.slice(delivered.length); + // Some providers retain only the last 4 MiB while streaming the full output. + // Replaying that captured tail would count its usage receipts a second time. + if (delivered.endsWith(final)) return ""; return final; }