diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index bdadebda32..23505da191 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -12,6 +12,7 @@ import { geminiVersionSupportsNativeAcpFlag, parseGeminiVersionParts, rewriteGeminiAcpFlagForVersion, + summarizeAcpxTurnUsage, } from "./execute.js"; import { runChildProcess } from "../server-utils.js"; @@ -252,6 +253,139 @@ describe("shared ACPX engine runtime behavior", () => { }); }); + it("captures per-run usage, cost deltas, and billing identity from the ACP runtime", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const logs: Array<{ stream: string; text: string }> = []; + let statusCalls = 0; + const execute = createAcpxEngineExecutor({ + createRuntime: () => ({ + ensureSession: async () => ({ + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + getStatus: async () => { + statusCalls += 1; + return statusCalls === 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", + used: 5550, + size: 200000, + cost: { amount: 1.1, currency: "USD" }, + }; + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: async () => {}, + }) as never, + resolveBillingIdentity: () => ({ provider: "anthropic", biller: "anthropic", billingType: "api" }), + }); + + const result = await execute({ + runId: "run-usage-capture", + agent: { + id: "agent-1", + companyId: "company-1", + }, + runtime: {}, + config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir }, + context: {}, + onLog: async (stream: "stdout" | "stderr", text: string) => { + logs.push({ stream, text }); + }, + onMeta: async () => {}, + } as never); + + expect(result.exitCode).toBe(0); + expect(statusCalls).toBe(2); + // Cache-write tokens count as input tokens; cached reads stay separate. + expect(result.usage).toEqual({ inputTokens: 150, outputTokens: 4500, cachedInputTokens: 900 }); + expect(result.usageBasis).toBe("per_run"); + // Agent-reported cost is cumulative; this run pays the delta. + expect(result.costUsd).toBeCloseTo(0.75); + expect(result.provider).toBe("anthropic"); + expect(result.biller).toBe("anthropic"); + expect(result.billingType).toBe("api"); + expect((result.resultJson as Record)?.cumulativeCostUsd).toBeCloseTo(1.15); + expect((result.resultJson as Record)?.usage).toEqual({ + inputTokens: 120, + outputTokens: 4500, + cachedReadTokens: 900, + cachedWriteTokens: 30, + }); + const statusLine = logs.find((entry) => entry.text.includes('"acpx.status"')); + expect(statusLine?.text).toContain('"cost"'); + }); + + it("falls back to usage_update events when the runtime lacks getStatus", async () => { + 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 }, + }; + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }), + close: async () => {}, + }) as never, + }); + + 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 () => {}, + } as never); + + expect(result.exitCode).toBe(0); + 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"); @@ -1311,3 +1445,117 @@ describe("shared ACP engine execution timeouts", () => { expect(cancelReasons).toContain(expectedMessage); }, 15_000); }); + +describe("summarizeAcpxTurnUsage", () => { + it("uses the post-turn amount alone when the cumulative cost counter reset", () => { + const summary = summarizeAcpxTurnUsage({ + preStatus: { usage: { cost: { amount: 2.5, currency: "USD" } } }, + postStatus: { + usage: { + cumulative: { inputTokens: 10, outputTokens: 20 }, + cost: { amount: 0.3, currency: "USD" }, + }, + }, + eventBreakdown: null, + eventCostUsd: null, + }); + expect(summary.costUsd).toBeCloseTo(0.3); + expect(summary.cumulativeCostUsd).toBeCloseTo(0.3); + }); + + it("ignores non-USD cost amounts", () => { + const summary = summarizeAcpxTurnUsage({ + preStatus: null, + postStatus: { usage: { cost: { amount: 4, currency: "EUR" } } }, + eventBreakdown: null, + eventCostUsd: null, + }); + expect(summary.costUsd).toBeNull(); + expect(summary.cumulativeCostUsd).toBeNull(); + }); + + it("returns no usage when nothing was reported", () => { + const summary = summarizeAcpxTurnUsage({ + preStatus: null, + postStatus: null, + eventBreakdown: null, + eventCostUsd: null, + }); + expect(summary.usage).toBeNull(); + expect(summary.costUsd).toBeNull(); + }); +}); + +describe("summarizeAcpxTurnUsage no-report turns", () => { + it("suppresses usage when the turn reported nothing and the persisted breakdown is unchanged", () => { + const stale = { inputTokens: 10, outputTokens: 500, cachedReadTokens: 30 }; + const summary = summarizeAcpxTurnUsage({ + preStatus: { usage: { cumulative: stale, cost: { amount: 0.5, currency: "USD" } } }, + postStatus: { usage: { cumulative: { ...stale }, cost: { amount: 0.5, currency: "USD" } } }, + eventBreakdown: null, + eventCostUsd: null, + }); + expect(summary.usage).toBeNull(); + expect(summary.usageDetail).toBeNull(); + expect(summary.costUsd).toBeCloseTo(0); + }); + + it("prefers current event usage when the persisted breakdown is stale", () => { + const stale = { inputTokens: 10, outputTokens: 500, cachedReadTokens: 30 }; + const current = { inputTokens: 25, outputTokens: 75, cachedReadTokens: 5 }; + const summary = summarizeAcpxTurnUsage({ + preStatus: { usage: { cumulative: stale } }, + postStatus: { usage: { cumulative: { ...stale } } }, + eventBreakdown: current, + eventCostUsd: null, + }); + expect(summary.usage).toEqual({ + inputTokens: 25, + outputTokens: 75, + cachedInputTokens: 5, + }); + expect(summary.usageDetail).toMatchObject(current); + }); + + it("treats omitted and explicit zero fields as the same stale breakdown", () => { + const current = { inputTokens: 25, outputTokens: 75, cachedReadTokens: 5 }; + const summary = summarizeAcpxTurnUsage({ + preStatus: { usage: { cumulative: { inputTokens: 10, outputTokens: 500 } } }, + postStatus: { + usage: { + cumulative: { + inputTokens: 10, + outputTokens: 500, + cachedReadTokens: 0, + cachedWriteTokens: 0, + thoughtTokens: 0, + totalTokens: 0, + }, + }, + }, + eventBreakdown: current, + eventCostUsd: null, + }); + expect(summary.usage).toEqual({ + inputTokens: 25, + outputTokens: 75, + cachedInputTokens: 5, + }); + }); + + it("does not reuse stale tokens when the turn reports cost only", () => { + const stale = { inputTokens: 10, outputTokens: 500, cachedReadTokens: 30 }; + const summary = summarizeAcpxTurnUsage({ + preStatus: { usage: { cumulative: stale, cost: { amount: 0.5, currency: "USD" } } }, + postStatus: { + usage: { cumulative: { ...stale }, cost: { amount: 0.5, currency: "USD" } }, + }, + eventBreakdown: null, + eventCostUsd: 0.75, + }); + expect(summary.usage).toBeNull(); + expect(summary.usageDetail).toBeNull(); + expect(summary.costUsd).toBeCloseTo(0.25); + expect(summary.cumulativeCostUsd).toBeCloseTo(0.75); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index 32dbac0877..ccf232e07f 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -5,7 +5,12 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; import { createHash, randomUUID } from "node:crypto"; import { fileURLToPath } from "node:url"; -import type { AdapterExecutionContext, AdapterExecutionResult } from "@paperclipai/adapter-utils"; +import type { + AdapterBillingType, + AdapterExecutionContext, + AdapterExecutionResult, + UsageSummary, +} from "@paperclipai/adapter-utils"; import { adapterExecutionTargetSessionIdentity, formatAdapterExecutionTimeoutErrorMessage, @@ -54,8 +59,11 @@ import { type AcpRuntimeEvent, type AcpRuntimeHandle, type AcpRuntimeOptions, + type AcpRuntimeStatus, type AcpRuntimeTurn, type AcpRuntimeTurnResult, + type AcpRuntimeUsageBreakdown, + type AcpRuntimeUsageCost, } from "acpx/runtime"; import { DEFAULT_ACP_ENGINE_AGENT, @@ -86,6 +94,12 @@ interface AcpxEngineSettings { packageRootDir: string; } +export interface AcpxEngineBillingIdentity { + provider?: string | null; + biller?: string | null; + billingType?: AdapterBillingType | null; +} + export interface AcpxEngineExecutorOptions { createRuntime?: AcpxRuntimeFactory; now?: () => number; @@ -93,6 +107,14 @@ export interface AcpxEngineExecutorOptions { adapterType?: string; moduleDir?: string; packageRootDir?: string; + /** + * Adapter-specific billing classification (provider/biller/billingType) for + * cost-ledger attribution. Without it, results fall back to the opaque + * "acpx" provider and an "unknown" billing type. + */ + resolveBillingIdentity?: ( + ctx: AdapterExecutionContext, + ) => AcpxEngineBillingIdentity | null | Promise; } interface AcpxPreparedRuntime { @@ -1428,6 +1450,8 @@ async function emitRuntimeEvent(ctx: AdapterExecutionContext, event: AcpRuntimeE tag: event.tag, used: event.used, size: event.size, + ...(event.cost ? { cost: event.cost } : {}), + ...(event.breakdown ? { breakdown: event.breakdown } : {}), }); return; } @@ -1454,6 +1478,116 @@ function resultErrorMessage(result: AcpRuntimeTurnResult): string | null { return result.error.message; } +function usageBreakdownsEqual( + left: AcpRuntimeUsageBreakdown, + right: AcpRuntimeUsageBreakdown, +): boolean { + return ( + asNumber(left.inputTokens, 0) === asNumber(right.inputTokens, 0) && + asNumber(left.outputTokens, 0) === asNumber(right.outputTokens, 0) && + asNumber(left.cachedReadTokens, 0) === asNumber(right.cachedReadTokens, 0) && + asNumber(left.cachedWriteTokens, 0) === asNumber(right.cachedWriteTokens, 0) && + asNumber(left.thoughtTokens, 0) === asNumber(right.thoughtTokens, 0) && + asNumber(left.totalTokens, 0) === asNumber(right.totalTokens, 0) + ); +} + +function usdCostAmount(cost: AcpRuntimeUsageCost | null | undefined): number | null { + if (!cost || typeof cost.amount !== "number" || !Number.isFinite(cost.amount)) return null; + if (cost.currency && cost.currency.trim().toUpperCase() !== "USD") return null; + return cost.amount; +} + +async function readRuntimeStatus( + runtime: AcpRuntime, + handle: AcpRuntimeHandle, +): Promise { + if (!runtime.getStatus) return null; + try { + return (await runtime.getStatus({ handle })) ?? null; + } catch { + return null; + } +} + +/** + * Fold the ACP runtime's post-turn usage into the adapter execution result + * shape. The runtime persists the latest turn's token breakdown (adapters like + * claude-agent-acp report per-turn accumulated usage in the prompt response), + * so tokens are per-run. Cost is reported by agents as a cumulative session + * amount, so the per-run cost is the delta against the pre-turn snapshot; a + * decrease means the agent process restarted and its counter reset, in which + * case the post-turn amount alone covers this run. + */ +export function summarizeAcpxTurnUsage(input: { + preStatus: AcpRuntimeStatus | null; + postStatus: AcpRuntimeStatus | null; + eventBreakdown: AcpRuntimeUsageBreakdown | null; + eventCostUsd: number | null; +}): { + usage: UsageSummary | null; + usageDetail: Record | null; + costUsd: number | null; + cumulativeCostUsd: number | null; +} { + // The persisted breakdown is overwritten per turn, so an unchanged value + // is stale for this turn. Prefer an in-turn event breakdown when available; + // otherwise suppress the stale value so it cannot be double-counted. + const preBreakdown = input.preStatus?.usage?.cumulative ?? null; + const postBreakdown = input.postStatus?.usage?.cumulative ?? null; + const postBreakdownIsStale = + preBreakdown != null && + postBreakdown != null && + usageBreakdownsEqual(preBreakdown, postBreakdown); + const breakdown = postBreakdownIsStale + ? input.eventBreakdown + : postBreakdown ?? input.eventBreakdown ?? null; + const inputTokens = Math.max(0, Math.floor(asNumber(breakdown?.inputTokens, 0))); + const outputTokens = Math.max(0, Math.floor(asNumber(breakdown?.outputTokens, 0))); + const cachedReadTokens = Math.max(0, Math.floor(asNumber(breakdown?.cachedReadTokens, 0))); + const cachedWriteTokens = Math.max(0, Math.floor(asNumber(breakdown?.cachedWriteTokens, 0))); + const hasTokens = inputTokens > 0 || outputTokens > 0 || cachedReadTokens > 0 || cachedWriteTokens > 0; + // Cache-write tokens are prompt tokens the provider billed to create cache + // entries; UsageSummary has no dedicated field, so count them as input. + const usage: UsageSummary | null = hasTokens + ? { + inputTokens: inputTokens + cachedWriteTokens, + outputTokens, + cachedInputTokens: cachedReadTokens, + } + : null; + const usageDetail = breakdown + ? Object.fromEntries( + Object.entries({ + inputTokens: breakdown.inputTokens, + outputTokens: breakdown.outputTokens, + cachedReadTokens: breakdown.cachedReadTokens, + cachedWriteTokens: breakdown.cachedWriteTokens, + thoughtTokens: breakdown.thoughtTokens, + totalTokens: breakdown.totalTokens, + }).filter((entry): entry is [string, number] => typeof entry[1] === "number"), + ) + : null; + + const previousCostUsd = usdCostAmount(input.preStatus?.usage?.cost); + const postCostUsd = usdCostAmount(input.postStatus?.usage?.cost); + const postCostIsStale = + input.eventCostUsd != null && + previousCostUsd != null && + postCostUsd != null && + postCostUsd === previousCostUsd; + const cumulativeCostUsd = postCostIsStale ? input.eventCostUsd : postCostUsd ?? input.eventCostUsd; + let costUsd: number | null = null; + if (cumulativeCostUsd != null) { + costUsd = + previousCostUsd != null && cumulativeCostUsd >= previousCostUsd + ? cumulativeCostUsd - previousCostUsd + : cumulativeCostUsd; + } + + return { usage, usageDetail, costUsd, cumulativeCostUsd }; +} + type AcpxExecutionPhase = "ensure_session" | "configure_session" | "turn"; function describeErrorDiagnostics(err: unknown): { @@ -1695,6 +1829,17 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const engine = resolveEngineSettings(deps); return async function executeAcpxEngine(ctx: AdapterExecutionContext): Promise { + let billingIdentity: AcpxEngineBillingIdentity | null = null; + try { + billingIdentity = (await deps.resolveBillingIdentity?.(ctx)) ?? null; + } catch { + billingIdentity = null; + } + const billingFields = { + provider: billingIdentity?.provider ?? "acpx", + ...(billingIdentity?.biller ? { biller: billingIdentity.biller } : {}), + billingType: billingIdentity?.billingType ?? ("unknown" as const), + }; const prepared = await buildRuntime({ ctx, engine }); // State the effective wall-clock timeout and its source up front so a // later timeout is diagnosable from the run log alone. Goes to stderr: @@ -1776,7 +1921,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { timedOut: false, errorMessage: message, ...classified, - provider: "acpx", + ...billingFields, model: prepared.requestedModel || null, clearSession, resultJson: { phase: "ensure_session" }, @@ -1792,7 +1937,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { timedOut: false, errorMessage: "ACPX did not return a runtime session handle.", errorCode: "acpx_runtime_error", - provider: "acpx", + ...billingFields, model: prepared.requestedModel || null, resultJson: { phase: "ensure_session" }, summary: "ACPX did not return a runtime session handle.", @@ -1830,7 +1975,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { timedOut: false, errorMessage: message, ...classified, - provider: "acpx", + ...billingFields, model: prepared.requestedModel || null, clearSession, resultJson: { @@ -1892,7 +2037,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { let timeout: NodeJS.Timeout | null = null; let timedOut = false; const textParts: string[] = []; + let eventBreakdown: AcpRuntimeUsageBreakdown | null = null; + let eventCostUsd: number | null = null; try { + // Snapshot pre-turn usage so cumulative agent-reported cost can be + // attributed to this run alone. + const preTurnStatus = await readRuntimeStatus(runtime, sessionHandle); const timeoutMs = prepared.timeoutSec > 0 ? prepared.timeoutSec * 1000 : undefined; controller = new AbortController(); if (timeoutMs) { @@ -1915,10 +2065,22 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }; for await (const event of turn.events) { if (event.type === "text_delta") textParts.push(event.text); + if (event.type === "status" && event.tag === "usage_update") { + eventBreakdown = event.breakdown ?? eventBreakdown; + eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd; + } await emitRuntimeEvent(ctx, event); } const terminal = await turn.result; if (timeout) clearTimeout(timeout); + // Read usage before the close/warm-handle paths below can discard state. + const postTurnStatus = await readRuntimeStatus(runtime, sessionHandle); + const turnUsage = summarizeAcpxTurnUsage({ + preStatus: preTurnStatus, + postStatus: postTurnStatus, + eventBreakdown, + eventCostUsd, + }); if (terminal.status === "failed" || terminal.status === "cancelled" || timedOut) { const existing = warmHandles.get(prepared.sessionKey); if (warmHandleMatches(existing, runtime, sessionHandle) && existing) { @@ -1998,10 +2160,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { sessionId: sessionHandle.backendSessionId ?? sessionHandle.runtimeSessionName, sessionParams: buildSessionParams({ prepared, handle: sessionHandle }), sessionDisplayId: sessionHandle.agentSessionId ?? sessionHandle.backendSessionId ?? sessionHandle.runtimeSessionName, - provider: "acpx", + ...billingFields, model: prepared.requestedModel || null, - billingType: "unknown", - costUsd: null, + ...(turnUsage.usage ? { usage: turnUsage.usage, usageBasis: "per_run" as const } : {}), + costUsd: turnUsage.costUsd, resultJson: { status: terminal.status, stopReason: terminalStopReason, @@ -2010,6 +2172,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { requestedModel: prepared.requestedModel || null, requestedThinkingEffort: prepared.requestedThinkingEffort || null, fastMode: prepared.fastMode, + ...(turnUsage.usageDetail ? { usage: turnUsage.usageDetail } : {}), + ...(turnUsage.cumulativeCostUsd != null + ? { cumulativeCostUsd: turnUsage.cumulativeCostUsd } + : {}), }, summary: textParts.join("").trim() || terminalStopReason || terminal.status, clearSession, @@ -2048,7 +2214,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { errorMessage: message, errorCode: timedOut ? "acpx_timeout" : classified.errorCode, errorMeta: classified.errorMeta, - provider: "acpx", + ...billingFields, model: prepared.requestedModel || null, clearSession: clearSession || timedOut, resultJson: { phase: "turn" }, diff --git a/packages/adapter-utils/src/types.ts b/packages/adapter-utils/src/types.ts index 7bdfbc4264..360276b8b4 100644 --- a/packages/adapter-utils/src/types.ts +++ b/packages/adapter-utils/src/types.ts @@ -77,6 +77,13 @@ export interface AdapterExecutionResult { retryNotBefore?: string | null; errorMeta?: Record; usage?: UsageSummary; + /** + * 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. + */ + usageBasis?: "per_run" | "session_cumulative" | null; /** * Legacy single session id output. Prefer `sessionParams` + `sessionDisplayId`. */ diff --git a/packages/adapters/claude-local/src/server/acp.test.ts b/packages/adapters/claude-local/src/server/acp.test.ts index ba8cb5bc34..6c2a46ab54 100644 --- a/packages/adapters/claude-local/src/server/acp.test.ts +++ b/packages/adapters/claude-local/src/server/acp.test.ts @@ -7,6 +7,7 @@ import { buildClaudeAcpConfig, createClaudeAcpExecutor, nodeVersionMeetsClaudeAcpMinimum, + resolveClaudeAcpBillingIdentity, resolveClaudeExecutionEngine, resolveClaudeExecutionEngineForRun, testClaudeAcpEnvironment, @@ -446,3 +447,51 @@ describe("claude_local ACP lane", () => { expect(runtimes[1]?.ensureInputs[0]?.resumeSessionId).toBe("acp-1"); }); }); + +describe("resolveClaudeAcpBillingIdentity", () => { + const originalApiKey = process.env.ANTHROPIC_API_KEY; + const originalBedrock = process.env.CLAUDE_CODE_USE_BEDROCK; + const originalBedrockBase = process.env.ANTHROPIC_BEDROCK_BASE_URL; + + afterEach(() => { + if (originalApiKey === undefined) delete process.env.ANTHROPIC_API_KEY; + else process.env.ANTHROPIC_API_KEY = originalApiKey; + if (originalBedrock === undefined) delete process.env.CLAUDE_CODE_USE_BEDROCK; + else process.env.CLAUDE_CODE_USE_BEDROCK = originalBedrock; + if (originalBedrockBase === undefined) delete process.env.ANTHROPIC_BEDROCK_BASE_URL; + else process.env.ANTHROPIC_BEDROCK_BASE_URL = originalBedrockBase; + }); + + it("classifies an adapter-config API key as api billing", () => { + expect( + resolveClaudeAcpBillingIdentity({ config: { env: { ANTHROPIC_API_KEY: "sk-ant-test" } } }), + ).toEqual({ provider: "anthropic", biller: "anthropic", billingType: "api" }); + }); + + it("classifies Bedrock auth as metered_api billed to aws_bedrock", () => { + expect( + resolveClaudeAcpBillingIdentity({ config: { env: { CLAUDE_CODE_USE_BEDROCK: "1" } } }), + ).toEqual({ provider: "anthropic", biller: "aws_bedrock", billingType: "metered_api" }); + }); + + it("falls back to subscription without API-key or Bedrock auth", () => { + delete process.env.ANTHROPIC_API_KEY; + delete process.env.CLAUDE_CODE_USE_BEDROCK; + delete process.env.ANTHROPIC_BEDROCK_BASE_URL; + expect(resolveClaudeAcpBillingIdentity({ config: {} })).toEqual({ + provider: "anthropic", + biller: "anthropic", + billingType: "subscription", + }); + }); + + it("ignores host env for remote execution targets", () => { + process.env.ANTHROPIC_API_KEY = "sk-ant-host-only"; + expect( + resolveClaudeAcpBillingIdentity({ + config: {}, + executionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/work" }, + } as never).billingType, + ).toBe("subscription"); + }); +}); diff --git a/packages/adapters/claude-local/src/server/acp.ts b/packages/adapters/claude-local/src/server/acp.ts index 8b74ddfdef..ef8c4d0c8e 100644 --- a/packages/adapters/claude-local/src/server/acp.ts +++ b/packages/adapters/claude-local/src/server/acp.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { fileURLToPath } from "node:url"; import type { + AdapterBillingType, AdapterEnvironmentCheck, AdapterEnvironmentTestContext, AdapterEnvironmentTestResult, @@ -111,8 +112,45 @@ export function buildClaudeAcpConfig(config: Record): Record & + Partial>, +): { provider: string; biller: string; billingType: AdapterBillingType } { + const envConfig = parseObject(parseObject(ctx.config).env); + const target = readAdapterExecutionTarget({ + executionTarget: ctx.executionTarget, + legacyRemoteExecution: ctx.executionTransport?.remoteExecution, + }); + const considerHostEnv = target?.kind !== "remote"; + const readEnvValue = (key: string): string => { + const fromConfig = envConfig[key]; + if (typeof fromConfig === "string" && fromConfig.trim()) return fromConfig.trim(); + const fromHost = considerHostEnv ? process.env[key] : undefined; + return typeof fromHost === "string" ? fromHost.trim() : ""; + }; + const bedrockFlag = readEnvValue("CLAUDE_CODE_USE_BEDROCK"); + const bedrock = bedrockFlag === "1" || bedrockFlag === "true" || Boolean(readEnvValue("ANTHROPIC_BEDROCK_BASE_URL")); + const billingType: AdapterBillingType = bedrock + ? "metered_api" + : readEnvValue("ANTHROPIC_API_KEY") + ? "api" + : "subscription"; + return { + provider: "anthropic", + biller: bedrock ? "aws_bedrock" : "anthropic", + billingType, + }; +} + function withClaudeAcpDefaults(options: ClaudeAcpExecutorOptions): AcpxEngineExecutorOptions { return { + resolveBillingIdentity: resolveClaudeAcpBillingIdentity, ...options, adapterType: "claude_local", moduleDir, diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 97ca1994ca..ea4e2759eb 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -45,6 +45,7 @@ import { DEFAULT_PAPERCLIP_AGENT_PROMPT_TEMPLATE, } from "@paperclipai/adapter-utils/server-utils"; import { + claudeModelUsageTotals, parseClaudeStreamJson, describeClaudeFailure, detectClaudeLoginRequired, @@ -927,8 +928,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise { const usageObj = parseObject(parsed.usage); return { @@ -937,6 +940,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise { ).toBeNull(); }); }); + +describe("claudeModelUsageTotals", () => { + it("sums per-model usage across models and counts cache writes as input", () => { + const totals = claudeModelUsageTotals({ + "claude-fable-5": { + inputTokens: 100, + outputTokens: 70_000, + cacheReadInputTokens: 250_000, + cacheCreationInputTokens: 4_000, + costUSD: 1.2, + }, + "claude-haiku-4-5": { + inputTokens: 50, + outputTokens: 7_000, + cacheReadInputTokens: 10_000, + cacheCreationInputTokens: 500, + costUSD: 0.05, + }, + }); + expect(totals).toEqual({ + inputTokens: 4_650, + outputTokens: 77_000, + cachedInputTokens: 260_000, + }); + }); + + it("returns null for missing or empty modelUsage", () => { + expect(claudeModelUsageTotals(undefined)).toBeNull(); + expect(claudeModelUsageTotals({})).toBeNull(); + }); +}); + +describe("parseClaudeStreamJson usage extraction", () => { + const resultEvent = (extra: Record) => + JSON.stringify({ + type: "result", + subtype: "success", + session_id: "sess-1", + result: "done", + total_cost_usd: 1.25, + usage: { input_tokens: 10, output_tokens: 1_800, cache_read_input_tokens: 20 }, + ...extra, + }); + + it("prefers modelUsage totals over the main-loop usage block and marks them per-run", () => { + const parsed = parseClaudeStreamJson( + `${resultEvent({ + modelUsage: { + "claude-fable-5": { + inputTokens: 90, + outputTokens: 77_000, + cacheReadInputTokens: 300_000, + cacheCreationInputTokens: 2_000, + }, + }, + })}\n`, + ); + expect(parsed.usage).toEqual({ + inputTokens: 2_090, + outputTokens: 77_000, + cachedInputTokens: 300_000, + }); + expect(parsed.usageBasis).toBe("per_run"); + expect(parsed.costUsd).toBeCloseTo(1.25); + }); + + it("falls back to the result usage block when modelUsage is absent", () => { + const parsed = parseClaudeStreamJson(`${resultEvent({})}\n`); + expect(parsed.usage).toEqual({ + inputTokens: 10, + outputTokens: 1_800, + cachedInputTokens: 20, + }); + expect(parsed.usageBasis).toBeNull(); + }); +}); diff --git a/packages/adapters/claude-local/src/server/parse.ts b/packages/adapters/claude-local/src/server/parse.ts index 9b279e974d..852a7f1f01 100644 --- a/packages/adapters/claude-local/src/server/parse.ts +++ b/packages/adapters/claude-local/src/server/parse.ts @@ -16,6 +16,31 @@ const CLAUDE_PROVIDER_QUOTA_RE = const CLAUDE_EXTRA_USAGE_RESET_RE = /(?:you(?:'|’)ve\s+hit\s+your\s+session\s+limit|session\s+limit\s+(?:reached|exceeded)|out\s+of\s+extra\s+usage|extra\s+usage|usage\s+limit\s+reached|usage\s+cap\s+reached|5[-\s]?hour\s+limit\s+reached|weekly\s+limit\s+reached|claude\s+usage\s+limit\s+reached)[\s\S]{0,120}?\bresets?\s+(?:at\s+)?([^\n()]+?)(?:\s*\(([^)]+)\))?(?:[.!]|\n|$)/i; +/** + * Sum the per-model usage ledger from a Claude CLI result event. The result + * event's top-level `usage` reflects only the main-loop message chain, so it + * undercounts output tokens whenever subagents or sidechains ran; `modelUsage` + * is the CLI's authoritative per-model accounting (it is what backs /cost). + * Cache-creation tokens are billed prompt tokens, so they count as input. + */ +export function claudeModelUsageTotals(modelUsage: unknown): UsageSummary | null { + const byModel = parseObject(modelUsage); + let inputTokens = 0; + let outputTokens = 0; + let cachedInputTokens = 0; + let sawEntry = false; + for (const value of Object.values(byModel)) { + const entry = parseObject(value); + if (Object.keys(entry).length === 0) continue; + sawEntry = true; + inputTokens += asNumber(entry.inputTokens, 0) + asNumber(entry.cacheCreationInputTokens, 0); + outputTokens += asNumber(entry.outputTokens, 0); + cachedInputTokens += asNumber(entry.cacheReadInputTokens, 0); + } + if (!sawEntry) return null; + return { inputTokens, outputTokens, cachedInputTokens }; +} + export function parseClaudeStreamJson(stdout: string) { let sessionId: string | null = null; let model = ""; @@ -62,13 +87,15 @@ export function parseClaudeStreamJson(stdout: string) { 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, }; } + const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage); const usageObj = parseObject(finalResult.usage); - const usage: UsageSummary = { + const usage: UsageSummary = modelUsageTotals ?? { inputTokens: asNumber(usageObj.input_tokens, 0), cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0), outputTokens: asNumber(usageObj.output_tokens, 0), @@ -82,6 +109,9 @@ export function parseClaudeStreamJson(stdout: string) { 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: modelUsageTotals ? ("per_run" as const) : null, 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 100e147400..2e72298b8f 100644 --- a/packages/adapters/codex-local/src/server/acp.test.ts +++ b/packages/adapters/codex-local/src/server/acp.test.ts @@ -7,6 +7,7 @@ import { buildCodexAcpConfig, createCodexAcpExecutor, nodeVersionMeetsCodexAcpMinimum, + resolveCodexAcpBillingIdentity, resolveCodexExecutionEngine, resolveCodexExecutionEngineForRun, testCodexAcpEnvironment, @@ -477,3 +478,50 @@ describe("codex_local ACP lane", () => { expect(runtimes[1]?.ensureInputs[0]?.resumeSessionId).toBe("acp-1"); }); }); + +describe("resolveCodexAcpBillingIdentity", () => { + const originalOpenAiKey = process.env.OPENAI_API_KEY; + const originalOpenRouterKey = process.env.OPENROUTER_API_KEY; + + afterEach(() => { + if (originalOpenAiKey === undefined) delete process.env.OPENAI_API_KEY; + else process.env.OPENAI_API_KEY = originalOpenAiKey; + if (originalOpenRouterKey === undefined) delete process.env.OPENROUTER_API_KEY; + else process.env.OPENROUTER_API_KEY = originalOpenRouterKey; + }); + + it("classifies an adapter-config API key as api billing to openai", () => { + delete process.env.OPENROUTER_API_KEY; + expect( + resolveCodexAcpBillingIdentity({ config: { env: { OPENAI_API_KEY: "sk-test" } } }), + ).toEqual({ provider: "openai", biller: "openai", billingType: "api" }); + }); + + it("falls back to chatgpt subscription without an API key", () => { + delete process.env.OPENAI_API_KEY; + delete process.env.OPENROUTER_API_KEY; + expect(resolveCodexAcpBillingIdentity({ config: {} })).toEqual({ + provider: "openai", + biller: "chatgpt", + billingType: "subscription", + }); + }); + + it("bills OpenRouter-backed runs to openrouter", () => { + expect( + resolveCodexAcpBillingIdentity({ + config: { env: { OPENAI_API_KEY: "sk-test", OPENROUTER_API_KEY: "or-test" } }, + }), + ).toEqual({ provider: "openai", biller: "openrouter", billingType: "api" }); + }); + + it("ignores host env for remote execution targets", () => { + process.env.OPENAI_API_KEY = "sk-host-only"; + expect( + resolveCodexAcpBillingIdentity({ + config: {}, + executionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/work" }, + } as never).billingType, + ).toBe("subscription"); + }); +}); diff --git a/packages/adapters/codex-local/src/server/acp.ts b/packages/adapters/codex-local/src/server/acp.ts index 6b8ae974ca..1f317b5557 100644 --- a/packages/adapters/codex-local/src/server/acp.ts +++ b/packages/adapters/codex-local/src/server/acp.ts @@ -2,12 +2,14 @@ import fs from "node:fs/promises"; import path from "node:path"; import { fileURLToPath } from "node:url"; import type { + AdapterBillingType, AdapterEnvironmentCheck, AdapterEnvironmentTestContext, AdapterEnvironmentTestResult, AdapterExecutionContext, AdapterExecutionResult, } from "@paperclipai/adapter-utils"; +import { inferOpenAiCompatibleBiller } from "@paperclipai/adapter-utils"; import { ensureAdapterExecutionTargetCommandResolvable, readAdapterExecutionTarget, @@ -113,6 +115,7 @@ export function buildCodexAcpConfig(config: Record): Record & + Partial>, +): { provider: string; biller: string; billingType: AdapterBillingType } { + const envConfig = parseObject(parseObject(ctx.config).env); + const target = readAdapterExecutionTarget({ + executionTarget: ctx.executionTarget, + legacyRemoteExecution: ctx.executionTransport?.remoteExecution, + }); + const considerHostEnv = target?.kind !== "remote"; + const mergedEnv: NodeJS.ProcessEnv = { + ...(considerHostEnv ? process.env : {}), + ...Object.fromEntries( + Object.entries(envConfig).filter((entry): entry is [string, string] => typeof entry[1] === "string"), + ), + }; + const apiKey = typeof mergedEnv.OPENAI_API_KEY === "string" && mergedEnv.OPENAI_API_KEY.trim().length > 0; + const billingType: AdapterBillingType = apiKey ? "api" : "subscription"; + const openAiCompatibleBiller = inferOpenAiCompatibleBiller(mergedEnv, "openai"); + const biller = + openAiCompatibleBiller === "openrouter" + ? "openrouter" + : billingType === "subscription" + ? "chatgpt" + : openAiCompatibleBiller ?? "openai"; + return { provider: "openai", biller, billingType }; +} + export function createCodexAcpExecutor(options: CodexAcpExecutorOptions = {}): CodexAcpExecutor { let executor: CodexAcpExecutor | null = null; return async (ctx) => { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7d6992f853..2a7095194f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -6560,9 +6560,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) runId: string; sessionId: string | null; rawUsage: UsageTotals | null; + usageBasis?: "per_run" | "session_cumulative" | null; }) { - const { agentId, runId, sessionId, rawUsage } = input; - if (!sessionId || !rawUsage) { + const { agentId, runId, sessionId, rawUsage, usageBasis } = input; + // Adapters that declare per-run usage (e.g. the ACPX lane reports each + // turn's tokens, not session totals) must not be session-delta'd, or + // consecutive runs would be undercounted. + if (!sessionId || !rawUsage || usageBasis === "per_run") { return { normalizedUsage: rawUsage, previousRawUsage: null as UsageTotals | null, @@ -12886,6 +12890,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) runId: run.id, sessionId: nextSessionState.displayId ?? nextSessionState.legacySessionId, rawUsage, + usageBasis: adapterResult.usageBasis ?? null, }); const normalizedUsage = sessionUsageResolution.normalizedUsage; const runErrorMessage = @@ -12936,7 +12941,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) rawCachedInputTokens: rawUsage.cachedInputTokens, rawOutputTokens: rawUsage.outputTokens, } : {}), - ...(sessionUsageResolution.derivedFromSessionTotals ? { usageSource: "session_delta" } : {}), + ...(sessionUsageResolution.derivedFromSessionTotals + ? { usageSource: "session_delta" } + : adapterResult.usageBasis === "per_run" + ? { usageSource: "per_run" } + : {}), ...((nextSessionState.displayId ?? nextSessionState.legacySessionId) ? { persistedSessionId: nextSessionState.displayId ?? nextSessionState.legacySessionId } : {}),