diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 571d07b81bf..b78e91122c5 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -195,6 +195,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 } @@ -233,10 +234,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 @@ -540,7 +542,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", @@ -554,6 +556,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).", ); @@ -1478,6 +1482,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"); @@ -1527,6 +1570,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"); @@ -2814,7 +2910,9 @@ describe("shared ACPX engine runtime behavior", () => { it.each(["claude", "codex", "grok", "gemini", "kimi"])("resumes %s with added and removed MCP servers and current credentials", async (agent) => { const root = await makeTempRoot(); - const config = { agent, cwd: root, stateDir: path.join(root, "state"), paperclipRuntimeSkills: [], paperclipSkillSync: { desiredSkills: [] } }; + // This fixture injects its runtime. Keep session identity independent of + // host CLI discovery/version probes while testing MCP credential changes. + const config = { agent, agentCommand: "node ./fake-acp.js", cwd: root, stateDir: path.join(root, "state"), paperclipRuntimeSkills: [], paperclipSkillSync: { desiredSkills: [] } }; const first = await runExecutor(config); const server = { name: "github", url: "https://example.test/github/mcp", connectionId: "github", token: "current-token" }; const second = await runExecutor(config, { @@ -6317,9 +6415,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: () => ({ @@ -6328,11 +6427,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, }); @@ -6343,7 +6438,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; }, }); @@ -6354,11 +6450,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); @@ -7316,6 +7420,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" }); @@ -7335,6 +7440,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 c0474eccb38..135126363fb 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -3942,6 +3942,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 @@ -4246,6 +4247,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; @@ -4751,8 +4753,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- @@ -4837,6 +4843,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, @@ -4953,6 +4962,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); } @@ -5039,6 +5052,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. @@ -5124,6 +5143,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, @@ -5257,6 +5277,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..00c2b34e743 100644 --- a/packages/adapter-utils/src/billing.test.ts +++ b/packages/adapter-utils/src/billing.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { inferOpenAiCompatibleBiller } from "./billing.js"; +import { inferOpenAiCompatibleBiller, resolveManagedOpenAiBilling } from "./billing.js"; describe("inferOpenAiCompatibleBiller", () => { it("returns openrouter when OPENROUTER_API_KEY is present", () => { @@ -25,4 +25,23 @@ 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(); + }); +}); + +describe("managed Responses routing", () => { + it.each([["openrouter", "api_key", "openrouter"], ["custom", "api_key", "unknown"], ["local", "none", "unknown"]])("keeps %s/%s usage outside direct OpenAI pricing", (kind, auth, biller) => { + expect(resolveManagedOpenAiBilling({ kind, auth })).toEqual({ provider: "openai", biller, billingType: "api" }); + }); + it("leaves ordinary direct API and subscription detection unchanged", () => { + expect(resolveManagedOpenAiBilling(undefined)).toBeUndefined(); + }); }); diff --git a/packages/adapter-utils/src/billing.ts b/packages/adapter-utils/src/billing.ts index 3ecd8ec3594..7d7fa87783d 100644 --- a/packages/adapter-utils/src/billing.ts +++ b/packages/adapter-utils/src/billing.ts @@ -14,7 +14,28 @@ 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; } + +/** Managed Responses routes use their own credentials and prices, including + * no-auth endpoints. Absence of OPENAI_API_KEY never makes them ChatGPT plans. */ +export function resolveManagedOpenAiBilling(routing: unknown) { + if (!routing || typeof routing !== "object") return undefined; + return { + provider: "openai", + biller: (routing as { kind?: unknown }).kind === "openrouter" ? "openrouter" : "unknown", + billingType: "api" as const, + }; +} diff --git a/packages/adapter-utils/src/index.ts b/packages/adapter-utils/src/index.ts index 1fb3046e1ed..3920143d8a0 100644 --- a/packages/adapter-utils/src/index.ts +++ b/packages/adapter-utils/src/index.ts @@ -3,6 +3,7 @@ export type { AgentRuntimeIdentity, AdapterRuntime, UsageSummary, + AdapterUsageCheckpoint, AdapterBillingType, AdapterRuntimeServiceReport, AdapterExecutionResult, @@ -88,7 +89,7 @@ export type { RuntimeStatusSink, RuntimeStatusUpdate, } from "./runtime-progress.js"; -export { inferOpenAiCompatibleBiller } from "./billing.js"; +export { inferOpenAiCompatibleBiller, resolveManagedOpenAiBilling } from "./billing.js"; export { ADAPTER_LOGIN_PANEL_MODES, ADAPTER_LOGIN_TIMEOUT_POLICIES, diff --git a/packages/adapter-utils/src/types.ts b/packages/adapter-utils/src/types.ts index 6b068adcd3b..6c05d918661 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 @@ -240,6 +272,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..669ccc952c5 --- /dev/null +++ b/packages/adapter-utils/src/usage-checkpoint.test.ts @@ -0,0 +1,132 @@ +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("redacts malformed counter text without making the counter absent", async () => { + const consume = vi.fn((_records: string) => null); + const log = createUsageCheckpointLog(vi.fn(), vi.fn(), consume); + await log("stdout", JSON.stringify({ type: "result", usage: { + input_tokens: 2, output_tokens: 1, cached_input_tokens: "private malformed counter", + } }) + "\n"); + const retained = consume.mock.calls[0][0]; + expect(retained).not.toContain("private malformed counter"); + const usage = JSON.parse(retained).usage; + expect(Object.hasOwn(usage, "cached_input_tokens")).toBe(true); + expect(typeof usage.cached_input_tokens).not.toBe("number"); + expect(usage.cached_input_tokens).not.toBeNull(); + }); + + it.each([false, true])("persists reported zero usage on confirmed completion (newline=%s)", async newline => { + const saved = vi.fn(); + const receipt = { usage: { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }, costUsd: null, complete: false }; + const log = createUsageCheckpointLog(vi.fn(), saved, () => receipt); + await log("stdout", JSON.stringify({ type: "step_finish", part: { tokens: { input: 0, output: 0 } } }) + (newline ? "\n" : "")); + expect(saved).not.toHaveBeenCalled(); + await log.flush({ complete: true }); + expect(saved).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ ...receipt, complete: true })); + await log.flush({ complete: true }); + expect(saved).toHaveBeenCalledTimes(1); + }); + + it("propagates a zero-receipt persistence failure before cleanup and never invents absent usage", async () => { + const failure = new Error("Cannot persist zero receipt"); + const saved = vi.fn().mockRejectedValue(failure); + const log = createUsageCheckpointLog(vi.fn(), saved, () => ({ usage: { inputTokens: 0, outputTokens: 0 }, costUsd: null, complete: false })); + await log.flush({ complete: true }); + expect(saved).not.toHaveBeenCalled(); + await log("stdout", '{"type":"step_finish","part":{"tokens":{"input":0,"output":0}}}'); + await expect(log.flush({ complete: true })).rejects.toBe(failure); + }); + + 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..274dfcc9eff --- /dev/null +++ b/packages/adapter-utils/src/usage-checkpoint.ts @@ -0,0 +1,126 @@ +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; + // Retain an invalid-type marker without retaining its contents. Omitting a + // malformed optional counter would let parsers default it to a valid zero. + if (typeof value === "string") return stringFields.has(field) ? value.slice(0, 250) : false; + 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.started"|"turn.completed"|"turn.failed"|"error"|"agent_end"|"step_finish"|"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; + const receipt = { ...parsed, complete: parsed.complete || complete }; + if (!receipt.complete && receipt.costStatus !== "unpriced" && receipt.costUsd == null && receipt.costUsdExact == null && + !Object.values(receipt.usage ?? {}).some(value => typeof value === "number" && value > 0)) return; + 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 988c2479f5a..ee1f3e35a76 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, @@ -964,6 +966,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, @@ -973,7 +983,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, @@ -1075,6 +1095,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, @@ -1237,19 +1266,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/acp.test.ts b/packages/adapters/codex-local/src/server/acp.test.ts index a1391bf6982..1927afca93f 100644 --- a/packages/adapters/codex-local/src/server/acp.test.ts +++ b/packages/adapters/codex-local/src/server/acp.test.ts @@ -1426,6 +1426,12 @@ describe("codex_local ACP lane", () => { }); describe("resolveCodexAcpBillingIdentity", () => { + it.each([["openrouter", "api_key", "openrouter"], ["custom", "api_key", "unknown"], ["local", "none", "unknown"]])("classifies managed %s/%s independently of host authentication", (kind, auth, biller) => { + expect(resolveCodexAcpBillingIdentity({ config: { + managedAiRouting: { kind, auth }, + env: { OPENAI_API_KEY: "", PAPERCLIP_AI_PROVIDER_KEY: auth === "none" ? "" : "fixture" }, + } })).toEqual({ provider: "openai", biller, billingType: "api" }); + }); const originalOpenAiKey = process.env.OPENAI_API_KEY; const originalOpenRouterKey = process.env.OPENROUTER_API_KEY; diff --git a/packages/adapters/codex-local/src/server/acp.ts b/packages/adapters/codex-local/src/server/acp.ts index c7715dc75ff..7eb2b87bfd6 100644 --- a/packages/adapters/codex-local/src/server/acp.ts +++ b/packages/adapters/codex-local/src/server/acp.ts @@ -13,7 +13,7 @@ import { parseLocalProcessFilesystemScope, parseLocalProcessNetworkScope, } from "@paperclipai/adapter-utils/local-process-sandbox"; -import { inferOpenAiCompatibleBiller } from "@paperclipai/adapter-utils"; +import { inferOpenAiCompatibleBiller, resolveManagedOpenAiBilling } from "@paperclipai/adapter-utils"; import { ensureAdapterExecutionTargetCommandResolvable, readAdapterExecutionTarget, @@ -330,6 +330,8 @@ export function resolveCodexAcpBillingIdentity( ctx: Pick & Partial>, ): { provider: string; biller: string; billingType: AdapterBillingType } { + const managedBilling = resolveManagedOpenAiBilling(ctx.config.managedAiRouting); + if (managedBilling) return managedBilling; const envConfig = parseObject(parseObject(ctx.config).env); const target = readAdapterExecutionTarget({ executionTarget: ctx.executionTarget, diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index 78e6594c519..d0c3ef76675 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1,8 +1,9 @@ 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"; -import { inferOpenAiCompatibleBiller, type AdapterExecutionContext, type AdapterExecutionResult } from "@paperclipai/adapter-utils"; +import { inferOpenAiCompatibleBiller, resolveManagedOpenAiBilling, type AdapterExecutionContext, type AdapterExecutionResult } from "@paperclipai/adapter-utils"; import { buildCodexAuthInboundProvision } from "./codex-auth-merge-scripts.js"; import { copyBackCodexAuth } from "./codex-auth-copyback.js"; import { @@ -61,7 +62,7 @@ import { type LocalProcessSandboxOptions, } from "@paperclipai/adapter-utils/local-process-sandbox"; import { - parseCodexJsonl, + parseCodexJsonl, createCodexJsonlParser, classifyCodexAuthRefreshFailure, extractCodexRetryNotBefore, isCodexHarnessCrash, @@ -993,7 +994,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise typeof entry[1] === "string", ), ); - const billingType = resolveCodexBillingType(effectiveEnv); + const managedBilling = resolveManagedOpenAiBilling(config.managedAiRouting); + const billingType = managedBilling?.billingType ?? resolveCodexBillingType(effectiveEnv); + const biller = managedBilling?.biller ?? resolveCodexBiller(effectiveEnv, billingType); const networkScope = parseLocalProcessNetworkScope(config.networkScope); const filesystemScope = parseLocalProcessFilesystemScope(config.filesystemScope); const localProcessSandbox: LocalProcessSandboxOptions | null = @@ -1228,6 +1231,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = consumeAccounting(stdout); + return { usage: parsed.usageReported ? parsed.usage : undefined, costStatus: parsed.usageComplete ? undefined : "unpriced", usageBasis: "per_run", provider: "openai", biller, billingType, model, pricingContext, costUsd: null, complete: parsed.usageComplete }; + }); try { const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { onProcessStopped: providerStop.beginInvocation(), @@ -1343,7 +1352,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); @@ -1354,6 +1363,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise; + pricingContext?: AdapterExecutionResult["pricingContext"]; monitor?: | { fired: false } | { fired: true; terminationSignal: NodeJS.Signals | null; elapsedMsSinceLastEvent: number; timeoutMs: number }; @@ -1403,19 +1415,22 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + it.each(["turn.failed", "error", "turn.completed"])("does not invent zero usage for %s without counters", (type) => { + expect(parseCodexJsonl(JSON.stringify({ type }))).toMatchObject({ + sawProtocolTerminalEvent: true, usageReported: false, usageComplete: false, + }); + }); + it.each([{}, { input_tokens: 0 }, { input_tokens: -1, output_tokens: 0 }, { input_tokens: 1, output_tokens: 0, cached_input_tokens: 2 }])("rejects missing or invalid completion counters: %j", (usage) => { + expect(parseCodexJsonl(JSON.stringify({ type: "turn.completed", usage })).usageComplete).toBe(false); + }); + it("distinguishes explicit zero usage from missing usage", () => { + expect(parseCodexJsonl(JSON.stringify({ type: "turn.completed", usage: { input_tokens: 0, output_tokens: 0 } }))).toMatchObject({ + usageReported: true, usageComplete: true, usage: { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }, + }); + }); it("captures session id, assistant summary, usage, and error message", () => { const stdout = [ JSON.stringify({ type: "thread.started", thread_id: "thread_123" }), @@ -28,7 +41,7 @@ describe("parseCodexJsonl", () => { sessionId: "thread_123", summary: "Recovered response", usage: { - inputTokens: 10, + inputTokens: 8, cachedInputTokens: 2, outputTokens: 4, }, @@ -36,6 +49,8 @@ describe("parseCodexJsonl", () => { errorMessage: "resume failed", sawProtocolEvent: true, sawProtocolTerminalEvent: true, + usageReported: true, + usageComplete: false, }); }); @@ -64,7 +79,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, }, @@ -72,6 +87,8 @@ describe("parseCodexJsonl", () => { errorMessage: null, sawProtocolEvent: true, sawProtocolTerminalEvent: true, + usageReported: true, + usageComplete: true, }); }); }); diff --git a/packages/adapters/codex-local/src/server/parse.ts b/packages/adapters/codex-local/src/server/parse.ts index 7698977f409..7db887d8107 100644 --- a/packages/adapters/codex-local/src/server/parse.ts +++ b/packages/adapters/codex-local/src/server/parse.ts @@ -1,6 +1,5 @@ import { asString, - asNumber, parseObject, parseJson, } from "@paperclipai/adapter-utils/server-utils"; @@ -28,72 +27,88 @@ 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; let sawProtocolEvent = false; let sawProtocolTerminalEvent = false; + let usageReported = false; + let usageComplete = false; const usage = { inputTokens: 0, cachedInputTokens: 0, 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; + const reported = parseObject(event.usage); + const validCount = (value: unknown): value is number => typeof value === "number" && Number.isSafeInteger(value) && value >= 0; + const cached = reported.cached_input_tokens ?? 0; + usageComplete = validCount(reported.input_tokens) && validCount(reported.output_tokens) + && validCount(cached) && cached <= reported.input_tokens; + if (usageComplete) { + usage.inputTokens = reported.input_tokens as number; + usage.cachedInputTokens = cached as number; + usage.outputTokens = reported.output_tokens as number; + usageReported = true; + } + } + if (type === "turn.started") usageComplete = false; + 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.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, + usageReported, + usageComplete, + }; }; } diff --git a/packages/adapters/cursor-local/src/server/execute.test.ts b/packages/adapters/cursor-local/src/server/execute.test.ts index 895b152e687..866cdd952e0 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 f0cbb4f4212..02b813fd40f 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,25 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const parsed = consumeAccounting(stdout); + const provider = resolveProviderFromModel(model); + const hasCompleteAccounting = parsed.usageComplete || parsed.costUsd != null; + return { + usage: parsed.usageReported ? parsed.usage : undefined, + costUsd: parsed.costUsd, + costStatus: hasCompleteAccounting ? undefined : "unpriced", + usageBasis: "per_run", provider, + biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, + complete: parsed.sawResult && hasCompleteAccounting, + }; + }); 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 +685,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + const hasCompleteAccounting = attempt.parsed.usageComplete || attempt.parsed.costUsd != null; + const usageComplete = hasCompleteAccounting && (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.usageReported ? attempt.parsed.usage : undefined, + usageBasis: "per_run", + provider: providerFromModel, + biller: resolveCursorBiller(effectiveEnv, billingType, providerFromModel), + model, + billingType, + costUsd: attempt.parsed.costUsd, + costStatus: hasCompleteAccounting ? undefined : "unpriced", errorMessage: `Timed out after ${timeoutSec}s`, clearSession: clearSessionOnMissingSession, }; @@ -729,6 +758,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; + let usageReported = false; + let missingUsage = 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; - - const event = parseJson(line); - if (!event) continue; - - const foundSession = readSessionId(event); - if (foundSession) sessionId = foundSession; - - const type = asString(event.type, "").trim(); - - if (type === "assistant") { - messages.push(...collectAssistantText(event.message)); - continue; + // Missing counters are not reported zeroes. Preserve earlier valid counts, + // but never present their partial sum as the complete invocation total. + const addUsage = (input: unknown, output: unknown, cached: unknown = 0, written: unknown = 0) => { + const counts = [input, output, cached, written]; + if (!counts.every(value => typeof value === "number" && Number.isSafeInteger(value) && value >= 0)) { + missingUsage = true; + return; } + const [inputTokens, outputTokens, cachedInputTokens, cacheWriteTokens] = counts as number[]; + const nextInput = usage.inputTokens + inputTokens + cacheWriteTokens; + const nextOutput = usage.outputTokens + outputTokens; + const nextCached = usage.cachedInputTokens + cachedInputTokens; + if (![nextInput, nextOutput, nextCached].every(Number.isSafeInteger)) { + missingUsage = true; + return; + } + usage.inputTokens = nextInput; + usage.outputTokens = nextOutput; + usage.cachedInputTokens = nextCached; + usageReported = true; + }; - 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))); + return (stdout: string) => { + for (const rawLine of stdout.split(/\r?\n/)) { + const line = normalizeCursorStreamLine(rawLine).line; + if (!line) continue; - 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); + const event = parseJson(line); + if (!event) continue; + + const foundSession = readSessionId(event); + if (foundSession) sessionId = foundSession; + + const type = asString(event.type, "").trim(); + + 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); + addUsage( + usageObj.input_tokens ?? usageObj.inputTokens, + usageObj.output_tokens ?? usageObj.outputTokens, + usageObj.cached_input_tokens ?? usageObj.cachedInputTokens ?? usageObj.cache_read_input_tokens, + ); + 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); + addUsage(tokens.input, tokens.output, cache.read, cache.write); + 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, + usageReported, + usageComplete: usageReported && !missingUsage, + 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 7f0e915709d..246f2a0a0ba 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 96a28c127b9..fbccf43638c 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,16 @@ export async function execute(ctx: AdapterExecutionContext): Promise {}), stdout => { + hasAccounting = true; + const parsed = consumeAccounting(stdout); + const provider = parseModelProvider(model || null); + return { usage: parsed.usageReported ? parsed.usage : undefined, costUsd: parsed.costUsd, + costStatus: parsed.usageComplete || parsed.costUsd != null ? undefined : "unpriced", + 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 +665,27 @@ 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(); + }); +}); + +describe("OpenCode reported usage completeness", () => { + it.each([undefined, {}, { input: 1 }, { input: -1, output: 2 }, { input: 1, output: 0.5 }, { input: 1, output: 2, cache: { read: -1 } }, { input: 1, output: 2, cache: { read: null } }, { input: 1, output: 2, reasoning: null }])("preserves unknown usage for incomplete counters %j", tokens => { + expect(parseOpenCodeJsonl(JSON.stringify({ type: "step_finish", part: { tokens } }))).toMatchObject({ usageReported: false, usageComplete: false, costUsd: null }); + }); + it("accepts explicit zeros but keeps a later missing step incomplete", () => { + const line = JSON.stringify({ type: "step_finish", part: { tokens: { input: 0, output: 0 } } }); + expect(parseOpenCodeJsonl(line)).toMatchObject({ usageReported: true, usageComplete: true, usage: { inputTokens: 0, outputTokens: 0 }, costUsd: null }); + expect(parseOpenCodeJsonl(line + '\n' + JSON.stringify({ type: "step_finish" }))).toMatchObject({ usageReported: true, usageComplete: false }); + }); +}); diff --git a/packages/adapters/opencode-local/src/server/parse.ts b/packages/adapters/opencode-local/src/server/parse.ts index ee7fd41279b..c244fa0e102 100644 --- a/packages/adapters/opencode-local/src/server/parse.ts +++ b/packages/adapters/opencode-local/src/server/parse.ts @@ -1,4 +1,4 @@ -import { asNumber, asString, parseJson, parseObject } from "@paperclipai/adapter-utils/server-utils"; +import { asString, parseJson, parseObject } from "@paperclipai/adapter-utils/server-utils"; function errorText(value: unknown): string { if (typeof value === "string") return value; @@ -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,75 @@ export function parseOpenCodeJsonl(stdout: string) { cachedInputTokens: 0, outputTokens: 0, }; - let costUsd = 0; + let costUsd: number | null = null; + let missingCost = false; + let usageReported = false; + let missingUsage = 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); + const counts = [tokens.input, tokens.output, cache.read === undefined ? 0 : cache.read, cache.write === undefined ? 0 : cache.write, tokens.reasoning === undefined ? 0 : tokens.reasoning]; + if (counts.every(value => typeof value === "number" && Number.isSafeInteger(value) && value >= 0)) { + const [input, output, cached, written, reasoning] = counts as number[]; + const next = { inputTokens: usage.inputTokens + input + written, + outputTokens: usage.outputTokens + output + reasoning, cachedInputTokens: usage.cachedInputTokens + cached }; + if (Object.values(next).every(Number.isSafeInteger)) { Object.assign(usage, next); usageReported = true; } + else missingUsage = true; + } else missingUsage = true; + 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 }, + usageReported, + usageComplete: usageReported && !missingUsage, + 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 dc00a9946f1..d2daf7e6a85 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/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts index ac5d7415700..fbf45dfea9f 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts @@ -294,6 +294,28 @@ describe("Codex app-server Codex driver", () => { expect(events.some((event) => event.eventType === "turn.interrupted")).toBe(true); }); + it.each(["output", "cache"])("does not certify partial provider usage using retained %s counters", async omitted => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-partial-usage", normalizedSessionId: "normalized-partial-usage", workingDirectory: WORKSPACE, + }); + const turn = await session.startTurn({ message: { role: "user", text: "Work." } }); + transport.push("turn/started", { threadId: "thread-1", turn: { id: turn.turnId, status: "inProgress" } }); + const first = { inputTokens: 10, outputTokens: 4, ...(omitted === "cache" ? { cachedInputTokens: 2 } : {}) }; + const second = { inputTokens: 20, ...(omitted === "cache" ? { outputTokens: 8 } : {}) }; + for (const total of [first, second]) { + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: turn.turnId, tokenUsage: { total } }); + } + transport.push("turn/completed", { threadId: "thread-1", turn: { id: turn.turnId, status: "completed", items: [] } }); + const events = await collectUntilTerminal(session.events()); + const usage = events.filter(event => event.payload.kind === "usage"); + expect(usage.map(event => event.payload.usage)).toMatchObject([ + { runDelta: first, runDeltaComplete: true }, + { runDelta: { ...first, ...second }, runDeltaComplete: false }, + ]); + expect(await session.usage()).toMatchObject({ runDeltaComplete: false }); + }); + it("accepts a thread usage snapshot replayed before a resumed turn starts", async () => { const transport = new FakeCodexTransport(); const session = await makeDriver([transport]).openSession({ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index 59370b07385..525feb8dce4 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -533,9 +533,19 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex } if (notification.method === "thread/tokenUsage/updated") { state.usageSnapshot = boundedPayload(record(params.tokenUsage)); - if (state.driverKind === "codex_app_server" && Object.keys(record(record(params.tokenUsage).total)).length > 0) { - state.codexUsageBaseline = observeCodexUsage(state.codexUsageBaseline, record(params.tokenUsage).total, false); - state.usageSnapshot = { ...state.usageSnapshot, ...codexRunUsage(state.codexUsageBaseline) }; + if (state.driverKind === "codex_app_server") { + const reported = record(record(params.tokenUsage).total); + const validCount = (value: unknown) => typeof value === "number" && Number.isSafeInteger(value) && value >= 0; + // The monotonic snapshot retains absent counters from older reports. + // Preserve whether this provider report can actually close accounting. + const runDeltaComplete = validCount(reported.inputTokens) && validCount(reported.outputTokens) + && Object.entries(reported).every(([key, value]) => !key.endsWith("Tokens") || validCount(value)) + && Object.keys(state.codexUsageBaseline?.latest ?? {}).every(key => !key.endsWith("Tokens") || validCount(reported[key])); + if (Object.keys(reported).length > 0) { + state.codexUsageBaseline = observeCodexUsage(state.codexUsageBaseline, reported, false); + state.usageSnapshot = { ...state.usageSnapshot, ...codexRunUsage(state.codexUsageBaseline) }; + } + state.usageSnapshot = { ...state.usageSnapshot, runDeltaComplete }; } // Codex can replay a thread-scoped usage snapshot while a resumed thread // is being attached, before the next turn has started. Keep the snapshot, 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..0c15c0081e4 --- /dev/null +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -0,0 +1,327 @@ +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(f => ["codex", "cursor", "opencode"].includes(f.name)))("$name preserves malformed optional counters through the checkpoint wrapper", async fixture => { + const dir = await mkdtemp(join(tmpdir(), "paperclip-invalid-counter-")); directories.push(dir); + const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 }); + const execute = await fixture.execute(); + for (const cached of ["private malformed counter", false, {}, 0, undefined]) for (const newline of [false, true]) { + const event = fixture.name === "opencode" + ? { type: "step_finish", part: { tokens: { input: 20, output: 10, cache: { read: cached } } } } + : { type: fixture.name === "codex" ? "turn.completed" : "result", usage: { input_tokens: 20, output_tokens: 10, cached_input_tokens: cached } }; + const stdout = JSON.stringify(event) + (newline ? "\n" : ""); + processResult.mockImplementation(async (_run, _target, _command, _args, options) => { + for (let offset = 0; offset < stdout.length; offset += 7) await options.onLog("stdout", stdout.slice(offset, offset + 7)); + return { exitCode: 0, signal: null, timedOut: false, stdout, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const onUsage = vi.fn(); + const result = await execute({ + runId: "invalid-counter-run", agent: { id: "agent", companyId: "company", name: "Worker", adapterType: `${fixture.name}_local`, adapterConfig: {} }, + runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, + config: { engine: "cli", command, cwd: dir, model: fixture.name === "opencode" ? "openai/test" : "gpt-6-astra", paperclipRuntimeSkills: [], + env: { OPENAI_API_KEY: "fixture", OPENCODE_ALLOW_ALL_MODELS: "1" } }, + context: {}, onLog: async () => {}, onUsage, + }); + const valid = cached === 0 || cached === undefined; + expect(result.usageComplete, `${fixture.name}: ${JSON.stringify(cached)}`).toBe(valid); + expect(result.usage).toEqual(valid ? { inputTokens: 20, outputTokens: 10, cachedInputTokens: 0 } : undefined); + expect(onUsage.mock.calls.at(-1)![0].complete).toBe(valid); + if (!valid) { + expect(result.costStatus).toBe("unpriced"); + expect(onUsage.mock.calls.every(([receipt]) => !receipt.complete && receipt.costStatus === "unpriced")).toBe(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 type of ["turn.failed", "error", "turn.completed"]) { + processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => { + const failedOutput = JSON.stringify({ type }) + "\n"; + await options.onLog("stdout", failedOutput); + return { exitCode: 1, signal: null, timedOut: false, stdout: failedOutput, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const failed = await execute({ ...context, config: { ...context.config, model: "gpt-6-astra" } }); + expect(failed).toMatchObject({ usage: undefined, usageComplete: false, costStatus: "unpriced", costUsd: null }); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: undefined, complete: false, costStatus: "unpriced", costUsd: null }); + } + for (const [kind, auth, biller] of [["openrouter", "api_key", "openrouter"], ["custom", "api_key", "unknown"], ["local", "none", "unknown"]]) { + const routed = await execute({ ...context, config: { ...context.config, + managedAiRouting: { kind, auth }, env: { ...(context.config.env as object), OPENAI_API_KEY: "", PAPERCLIP_AI_PROVIDER_KEY: auth === "none" ? "" : "fixture" }, + } }); + expect(routed).toMatchObject({ billingType: "api", biller, costUsd: null }); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ billingType: "api", biller, costUsd: null }); + } + 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 (fixture.name === "opencode") { + for (const kind of ["zero_usage", "zero_price", "missing", "partial"]) for (const timedOut of [false, true]) { + const part = kind === "zero_usage" ? { tokens: { input: 0, output: 0 } } : kind === "zero_price" ? { cost: 0 } : kind === "partial" ? { tokens: { input: 2 } } : {}; + processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => { + const output = JSON.stringify({ type: "step_finish", part }); + await options.onLog("stdout", output); + return { exitCode: timedOut ? null : 0, signal: timedOut ? "SIGTERM" : null, timedOut, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + onUsage.mockClear(); + const reported = await execute(context); + const complete = !timedOut && kind.startsWith("zero_"); + expect(reported.usageComplete).toBe(complete); + expect(reported.usage).toEqual(kind === "zero_usage" ? { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 } : undefined); + expect(reported.costUsd).toBe(kind === "zero_price" ? 0 : null); + if (complete || !kind.startsWith("zero_")) expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ complete }); + if (!kind.startsWith("zero_")) expect(reported.costStatus).toBe("unpriced"); + } + } + if (fixture.name === "cursor") { + for (const type of ["result", "step_finish"]) for (const timedOut of [false, true]) { + processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => { + const output = JSON.stringify({ type }) + "\n"; + await options.onLog("stdout", output); + return { exitCode: timedOut ? null : 0, signal: timedOut ? "SIGTERM" : null, timedOut, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const missing = await execute({ ...context, config: { ...context.config, model: "gpt-5.6-sol" } }); + expect(missing).toMatchObject({ usage: undefined, usageComplete: false, costStatus: "unpriced", costUsd: null }); + expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: undefined, complete: false, costStatus: "unpriced", costUsd: null }); + } + for (const explicitPrice of [false, true]) { + processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => { + const output = JSON.stringify({ type: "result", ...(explicitPrice ? { total_cost_usd: 0 } : { usage: { input_tokens: 0, output_tokens: 0 } }) }) + "\n"; + await options.onLog("stdout", output); + return { exitCode: 0, signal: null, timedOut: false, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() }; + }); + const reported = await execute(context); + expect(reported.usageComplete).toBe(true); + expect(reported.usage).toEqual(explicitPrice ? undefined : { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }); + expect(reported.costUsd).toBe(explicitPrice ? 0 : null); + expect(onUsage.mock.calls.at(-1)![0].complete).toBe(true); + } + } + 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.each(["turn.failed", "error", "turn.started"])("saves an unpriced Codex %s checkpoint before exposing its log", async (type) => { + const { createCodexJsonlParser } = await import("../../../packages/adapters/codex-local/src/server/parse.js"); + const { createUsageCheckpointLog } = await import("@paperclipai/adapter-utils/usage-checkpoint"); + const consume = createCodexJsonlParser(); + const order: string[] = []; + const saved = vi.fn(async () => { order.push("receipt"); }); + const parsed = vi.fn((records: string) => { + const result = consume(records); + return { usage: result.usageReported ? result.usage : undefined, complete: result.usageComplete, costStatus: "unpriced" as const, costUsd: null }; + }); + const log = createUsageCheckpointLog(async () => { order.push("log"); }, saved, parsed); + await log("stdout", JSON.stringify({ type, message: "private provider failure" }) + "\n"); + // No flush or adapter return has happened: a host crash after this point + // must preserve the unknown usage and keep the reservation pending. + expect(saved).toHaveBeenCalledWith(expect.objectContaining({ complete: false, costUsd: null, costStatus: "unpriced", usage: undefined })); + expect(order).toEqual(["receipt", "log"]); + expect(parsed.mock.calls[0][0]).not.toContain("private provider failure"); + }); + + 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..087a8667868 100644 --- a/server/src/__tests__/cursor-local-adapter.test.ts +++ b/server/src/__tests__/cursor-local-adapter.test.ts @@ -4,6 +4,39 @@ import { parseCursorStdoutLine } from "@paperclipai/adapter-cursor-local/ui"; import { printCursorStreamEvent } from "@paperclipai/adapter-cursor-local/cli"; describe("cursor parser", () => { + it.each([ + { type: "result" }, + { type: "result", usage: {} }, + { type: "result", usage: { input_tokens: 0 } }, + { type: "result", usage: { input_tokens: -1, output_tokens: 0 } }, + { type: "step_finish", part: {} }, + ])("does not treat absent or malformed counters as reported zeroes (%j)", event => { + const parsed = parseCursorJsonl(JSON.stringify(event)); + expect(parsed.usageReported).toBe(false); + expect(parsed.usageComplete).toBe(false); + expect(parsed.costUsd).toBeNull(); + }); + + it.each([ + { type: "result", usage: { input_tokens: 0, output_tokens: 0 } }, + { type: "result", usage: { inputTokens: 0, outputTokens: 0 } }, + { type: "step_finish", part: { tokens: { input: 0, output: 0 } } }, + ])("preserves explicitly reported zero counters (%j)", event => { + expect(parseCursorJsonl(JSON.stringify(event))).toMatchObject({ + usageReported: true, usageComplete: true, + usage: { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0 }, + }); + }); + + it("keeps known partial usage without claiming a complete total", () => { + const parsed = parseCursorJsonl([ + { type: "step_finish", part: { tokens: { input: 20, output: 5 } } }, + { type: "result" }, + ].map(event => JSON.stringify(event)).join("\n")); + expect(parsed).toMatchObject({ usageReported: true, usageComplete: false, + usage: { inputTokens: 20, outputTokens: 5 }, costUsd: null }); + }); + it("extracts session, summary, usage, cost, and terminal error message", () => { const stdout = [ JSON.stringify({ type: "system", subtype: "init", session_id: "chat_123", model: "gpt-5" }), @@ -403,3 +436,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..473a28807b1 100644 --- a/server/src/__tests__/cursor-local-execute.test.ts +++ b/server/src/__tests__/cursor-local-execute.test.ts @@ -1,10 +1,23 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { runChildProcess } from "@paperclipai/adapter-utils/server-utils"; import { execute } from "@paperclipai/adapter-cursor-local/server"; +// These fixtures verify command discovery, environment and real CLI execution. +// Callback bridge startup/teardown has its own execution-target coverage; starting +// that server here adds unrelated process and polling work to command assertions. +vi.mock("@paperclipai/adapter-utils/execution-target", async () => { + const actual = await vi.importActual( + "@paperclipai/adapter-utils/execution-target", + ); + return { + ...actual, + startAdapterExecutionTargetPaperclipBridge: async () => null, + }; +}); + async function writeFakeCursorCommand(commandPath: string): Promise { const script = `#!/usr/bin/env node const fs = require("node:fs"); @@ -88,6 +101,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/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index 99efb01d320..4711ad72c29 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -4,6 +4,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { agents, agentWakeupRequests, + agentRuntimeState, companies, costEvents, createDb, @@ -1227,6 +1228,41 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { ); }); + it.each(["codex_local", "gemini_local"].flatMap(adapterType => [ + { adapterType, inputTokens: 50_000, cachedInputTokens: 550_000, rotate: true, rawInputIncludesCached: false }, + { adapterType, inputTokens: 50_000, cachedInputTokens: 449_999, rotate: false, rawInputIncludesCached: false }, + { adapterType, inputTokens: 0, cachedInputTokens: 500_000, rotate: true, rawInputIncludesCached: false }, + { adapterType, inputTokens: 400_000, cachedInputTokens: 350_000, rotate: false, rawInputIncludesCached: undefined }, + { adapterType, inputTokens: 500_000, cachedInputTokens: 350_000, rotate: true, rawInputIncludesCached: undefined }, + ]))("counts cached prompt input for $adapterType session rotation ($inputTokens + $cachedInputTokens; raw includes cache: $rawInputIncludesCached)", async ({ adapterType, inputTokens, cachedInputTokens, rotate, rawInputIncludesCached }) => { + const { companyId, agentId } = await seedCompanyAndAgent({ heartbeatConfig: { + sessionCompaction: { enabled: true, maxSessionRuns: 0, maxRawInputTokens: 500_000, maxSessionAgeHours: 0 }, + } }); + const sessionId = randomUUID(); + await db.update(agents).set({ adapterType }).where(eq(agents.id, agentId)); + await db.insert(agentRuntimeState).values({ companyId, agentId, adapterType, sessionId }); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", sessionIdAfter: sessionId, + usageJson: { ...(rawInputIncludesCached === false ? { rawInputIncludesCached } : {}), inputTokens, cachedInputTokens, outputTokens: 10, rawInputTokens: inputTokens, rawCachedInputTokens: cachedInputTokens }, + resultJson: { summary: "Continue from this completed turn" }, startedAt: new Date(), finishedAt: new Date(), + }); + mockAdapterExecute.mockResolvedValueOnce({ exitCode: 0, signal: null, timedOut: false, errorMessage: null, + provider: "test", model: "test-model", costUsd: 0, usageComplete: true, usageBasis: "per_run", + usage: { inputTokens: 1, cachedInputTokens: 2, outputTokens: 3 }, + }); + const run = await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual" }); + expect(run).not.toBeNull(); + expect(await waitForCondition(() => Promise.resolve(countExecuteCallsForRun(run!.id) === 1))).toBe(true); + const [context] = mockAdapterExecute.mock.calls.find(([context]) => context?.runId === run!.id)!; + expect(context.runtime.sessionId).toBe(rotate ? null : sessionId); + if (rotate) { + expect(context.context.paperclipSessionRotationReason).toContain((inputTokens + (rawInputIncludesCached === false ? cachedInputTokens : 0)).toLocaleString("en-US")); + expect(context.context.paperclipSessionHandoffMarkdown).toContain("Continue from this completed turn"); + } else expect(context.context.paperclipSessionRotationReason).toBeUndefined(); + await heartbeat.drainActiveRunExecutions(); + const [finished] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id)); + expect(finished.usageJson).toMatchObject({ rawInputTokens: 1, rawCachedInputTokens: 2, rawInputIncludesCached: false }); + }); + it("skips wakes before queueing when per-agent daily cost cap is reached", async () => { const { companyId, agentId } = await seedCompanyAndAgent({ heartbeatConfig: { diff --git a/server/src/adapters/process/execute.ts b/server/src/adapters/process/execute.ts index 672a4827ab8..2c9fc5dac1c 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; } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 832fef85161..db56257f742 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -12251,6 +12251,10 @@ export function heartbeatService( ? await getOldestRunForSession(agent.id, sessionId) : (runs[runs.length - 1] ?? latestRun); const latestRawUsage = readRawUsageTotals(latestRun?.usageJson); + // Historical Codex/Gemini raw input includes cache reads. Only add the + // separate cache counter when the writer explicitly saved exclusive input. + const latestRawInputTokens = (latestRawUsage?.inputTokens ?? 0) + + (latestRun?.usageJson?.rawInputIncludesCached === false ? latestRawUsage?.cachedInputTokens ?? 0 : 0); const sessionAgeHours = latestRun && oldestRun ? Math.max( @@ -12267,10 +12271,10 @@ export function heartbeatService( } else if ( policy.maxRawInputTokens > 0 && latestRawUsage && - latestRawUsage.inputTokens >= policy.maxRawInputTokens + latestRawInputTokens >= policy.maxRawInputTokens ) { reason = - `session raw input reached ${formatCount(latestRawUsage.inputTokens)} tokens ` + + `session raw input reached ${formatCount(latestRawInputTokens)} tokens ` + `(threshold ${formatCount(policy.maxRawInputTokens)})`; } else if ( policy.maxSessionAgeHours > 0 && @@ -25624,6 +25628,7 @@ export function heartbeatService( ...(rawUsage ? { rawInputTokens: rawUsage.inputTokens, + rawInputIncludesCached: false, rawCachedInputTokens: rawUsage.cachedInputTokens, rawOutputTokens: rawUsage.outputTokens, }