fix(adapters): propagate ACP-lane usage and cost into spend telemetry (#9471)
## Thinking Path > - Paperclip is the open source control plane for running and governing AI-agent companies. > - Adapter executions feed token usage, billing identity, and run cost into the control plane's spend telemetry. > - The default ACP execution lane for local Claude and Codex adapters did not propagate per-turn usage or cost, so paid runs could be recorded with zero spend and no tokens. > - Claude CLI result events could also undercount output tokens by reading only the main-loop usage block instead of the complete per-model ledger. > - The shared executor needs to distinguish per-run usage from session-cumulative usage so the server does not apply the wrong delta heuristic. > - This pull request captures ACP usage and cumulative-cost deltas, resolves adapter billing identity, uses Claude's complete model-usage ledger, and preserves per-run usage in server normalization. > - The benefit is accurate token and cost accounting across the default paid Claude and Codex execution paths. ## Linked Issues or Issue Description ### What happened? Paid `claude_local` and `codex_local` runs using the default ACP engine can complete successfully while the control plane records zero or null cost and missing token usage. Claude CLI result parsing can additionally undercount output tokens when subagent or sidechain usage is present. ### Steps to reproduce 1. Run a paid Claude or Codex local adapter through the ACP engine. 2. Complete a turn that reports usage and cumulative cost through ACP status/events. 3. Inspect the execution result and normalized run telemetry. ### Expected behavior The execution result contains per-turn token usage, a per-run USD cost delta, and the correct billing identity. Server normalization records those per-run values without applying a session-cumulative delta a second time. ### Actual behavior before this change ACP execution results returned no usage and `costUsd: null` with unknown billing. The server therefore recorded zero spend and no tokens for paid runs. Claude CLI parsing could use an incomplete usage block. ## What Changed - Capture ACP usage from runtime status and `usage_update` events, reporting it as `usageBasis: per_run`. - Convert agent-reported cumulative ACP cost into a per-turn delta, including counter-reset and no-report safeguards. - Add a shared billing-identity resolver and map Claude and Codex authentication/provider modes to control-plane billing types. - Prefer Claude result-event `modelUsage` totals so subagent and sidechain tokens are included. - Skip the server's session-cumulative usage delta when an adapter explicitly reports per-run usage. - Add regression coverage for usage capture, event fallback, cost resets, stale reports, billing identities, model-usage totals, and server spend normalization. ## Verification - `pnpm exec vitest run packages/adapter-utils/src/acpx-engine/execute.test.ts packages/adapters/claude-local/src/server/parse.test.ts packages/adapters/claude-local/src/server/acp.test.ts packages/adapters/codex-local/src/server/acp.test.ts server/src/__tests__/costs-service.test.ts server/src/__tests__/monthly-spend-service.test.ts` — 6 files, 126 tests passed. - `pnpm --filter @paperclipai/adapter-utils typecheck` — passed. - `pnpm --filter @paperclipai/adapter-claude-local typecheck` — passed. - `pnpm --filter @paperclipai/adapter-codex-local typecheck` — passed. - `pnpm --filter @paperclipai/server typecheck` — passed. - A broader Claude-local suite has a pre-existing rate-limit classification failure in `test.probe.test.ts`; it also fails on clean `master` and is unrelated to this change. ## Risks - Cost reporting depends on the agent's cumulative counter semantics; reset handling falls back to the post-turn amount and is covered by regression tests. - Incorrect billing-mode inference could misclassify spend; provider/auth mappings mirror each adapter's existing CLI behavior and have focused tests. - The new `usageBasis` contract changes server normalization only when adapters explicitly opt into `per_run`; existing adapters retain prior behavior. - No database migration, workflow, lockfile, or UI changes are included. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Implementation commit: Anthropic Claude Fable 5, tool-enabled coding workflow (exact context window and runtime configuration were not recorded in the commit metadata). - PR preparation and verification: OpenAI Codex, tool-enabled coding agent (runtime model ID and context window are not exposed to this session). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
4a40c0cb13
commit
9e7e84e3fe
|
|
@ -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<string, unknown>)?.cumulativeCostUsd).toBeCloseTo(1.15);
|
||||
expect((result.resultJson as Record<string, unknown>)?.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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<AcpxEngineBillingIdentity | null>;
|
||||
}
|
||||
|
||||
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<AcpRuntimeStatus | null> {
|
||||
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<string, number> | 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<AdapterExecutionResult> {
|
||||
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" },
|
||||
|
|
|
|||
|
|
@ -77,6 +77,13 @@ export interface AdapterExecutionResult {
|
|||
retryNotBefore?: string | null;
|
||||
errorMeta?: Record<string, unknown>;
|
||||
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`.
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>): Record<st
|
|||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify billing the same way the Claude CLI lane does so ACP runs land in
|
||||
* the cost ledger with a real provider/billingType instead of acpx/unknown.
|
||||
* Host env only counts for local execution targets; remote targets see just
|
||||
* the adapter-config env.
|
||||
*/
|
||||
export function resolveClaudeAcpBillingIdentity(
|
||||
ctx: Pick<AdapterExecutionContext, "config"> &
|
||||
Partial<Pick<AdapterExecutionContext, "executionTarget" | "executionTransport">>,
|
||||
): { 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,
|
||||
|
|
|
|||
|
|
@ -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<AdapterExec
|
|||
};
|
||||
}
|
||||
|
||||
const fallbackModelUsageTotals = parsedStream.usage ? null : claudeModelUsageTotals(parsed.modelUsage);
|
||||
const usage =
|
||||
parsedStream.usage ??
|
||||
fallbackModelUsageTotals ??
|
||||
(() => {
|
||||
const usageObj = parseObject(parsed.usage);
|
||||
return {
|
||||
|
|
@ -937,6 +940,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
outputTokens: asNumber(usageObj.output_tokens, 0),
|
||||
};
|
||||
})();
|
||||
const usageBasis = parsedStream.usage
|
||||
? parsedStream.usageBasis
|
||||
: fallbackModelUsageTotals
|
||||
? ("per_run" as const)
|
||||
: null;
|
||||
|
||||
const rawResolvedSessionId =
|
||||
parsedStream.sessionId ??
|
||||
|
|
@ -1051,6 +1059,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
retryNotBefore: transientRetryNotBefore ? transientRetryNotBefore.toISOString() : null,
|
||||
errorMeta,
|
||||
usage,
|
||||
...(usageBasis ? { usageBasis } : {}),
|
||||
sessionId: resolvedSessionId,
|
||||
sessionParams: resolvedSessionParams,
|
||||
sessionDisplayId: resolvedSessionId,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
claudeModelUsageTotals,
|
||||
parseClaudeStreamJson,
|
||||
detectClaudeLoginRequired,
|
||||
extractClaudeRetryNotBefore,
|
||||
isClaudeProviderQuotaError,
|
||||
|
|
@ -345,3 +347,79 @@ describe("extractClaudeRetryNotBefore", () => {
|
|||
).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<string, unknown>) =>
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> | 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,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>): Record<str
|
|||
|
||||
function withCodexAcpDefaults(options: CodexAcpExecutorOptions): AcpxEngineExecutorOptions {
|
||||
return {
|
||||
resolveBillingIdentity: resolveCodexAcpBillingIdentity,
|
||||
...options,
|
||||
adapterType: "codex_local",
|
||||
moduleDir,
|
||||
|
|
@ -120,6 +123,40 @@ function withCodexAcpDefaults(options: CodexAcpExecutorOptions): AcpxEngineExecu
|
|||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify billing the same way the Codex CLI lane does so ACP runs land in
|
||||
* the cost ledger with a real provider/billingType instead of acpx/unknown.
|
||||
* Host env only counts for local execution targets; remote targets see just
|
||||
* the adapter-config env.
|
||||
*/
|
||||
export function resolveCodexAcpBillingIdentity(
|
||||
ctx: Pick<AdapterExecutionContext, "config"> &
|
||||
Partial<Pick<AdapterExecutionContext, "executionTarget" | "executionTransport">>,
|
||||
): { 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) => {
|
||||
|
|
|
|||
|
|
@ -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 }
|
||||
: {}),
|
||||
|
|
|
|||
Loading…
Reference in New Issue