mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
fix: checkpoint adapter usage and preserve unknown prices
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
7eb38f881e
commit
b827836835
30 files changed
+411
-58
No files matched your search
@@ -7119,6 +7119,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" });
|
||||
@@ -7138,6 +7139,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);
|
||||
|
||||
@@ -4906,6 +4906,9 @@ 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 });
|
||||
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);
|
||||
}
|
||||
@@ -4992,6 +4995,8 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
||||
eventBreakdown,
|
||||
eventCostUsd,
|
||||
});
|
||||
await ctx.onUsage?.({ ...billingFields, usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd,
|
||||
model: prepared.requestedModel, usageBasis: "per_run", complete: !channelLost });
|
||||
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.
|
||||
@@ -5077,6 +5082,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
|
||||
model: prepared.requestedModel || null,
|
||||
...(turnUsage.usage ? { usage: turnUsage.usage, usageBasis: "per_run" as const } : {}),
|
||||
costUsd: turnUsage.costUsd,
|
||||
usageComplete: !channelLost,
|
||||
resultJson: {
|
||||
status: channelLost ? "failed" : terminal.status,
|
||||
...activityDiagnostics,
|
||||
|
||||
@@ -25,4 +25,14 @@ describe("inferOpenAiCompatibleBiller", () => {
|
||||
),
|
||||
).toBe("openai");
|
||||
});
|
||||
|
||||
it.each(["https://proxy.example/v1", "https://openrouter.ai.example/v1", "https://example.org/openrouter.ai", "http://api.openai.com/v1", "invalid"])("does not apply direct provider prices to %s", (baseUrl) => {
|
||||
expect(inferOpenAiCompatibleBiller({ OPENAI_BASE_URL: baseUrl }, "openai")).toBe("unknown");
|
||||
});
|
||||
|
||||
it("recognizes each supported base URL override", () => {
|
||||
expect(inferOpenAiCompatibleBiller({ OPENAI_API_BASE: "https://proxy.example" })).toBe("unknown");
|
||||
expect(inferOpenAiCompatibleBiller({ OPENAI_API_BASE_URL: "https://proxy.example" })).toBe("unknown");
|
||||
expect(inferOpenAiCompatibleBiller({}, null)).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -14,7 +14,17 @@ export function inferOpenAiCompatibleBiller(
|
||||
readEnv(env, "OPENAI_BASE_URL") ??
|
||||
readEnv(env, "OPENAI_API_BASE") ??
|
||||
readEnv(env, "OPENAI_API_BASE_URL");
|
||||
if (baseUrl && /openrouter\.ai/i.test(baseUrl)) return "openrouter";
|
||||
if (baseUrl) {
|
||||
try {
|
||||
const url = new URL(baseUrl);
|
||||
if (url.protocol === "https:" && url.hostname === "openrouter.ai") return "openrouter";
|
||||
// An OpenAI-compatible endpoint does not imply OpenAI prices. Do not
|
||||
// assign direct-provider estimates to a proxy or an unknown endpoint.
|
||||
if (url.protocol !== "https:" || url.hostname !== "api.openai.com") return "unknown";
|
||||
} catch {
|
||||
return "unknown";
|
||||
}
|
||||
}
|
||||
|
||||
return fallback;
|
||||
}
|
||||
@@ -2,6 +2,7 @@ export type {
|
||||
AdapterAgent,
|
||||
AdapterRuntime,
|
||||
UsageSummary,
|
||||
AdapterUsageCheckpoint,
|
||||
AdapterBillingType,
|
||||
AdapterRuntimeServiceReport,
|
||||
AdapterExecutionResult,
|
||||
|
||||
@@ -30,9 +30,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 =
|
||||
@@ -92,11 +115,14 @@ export interface AdapterExecutionResult {
|
||||
retryNotBefore?: string | null;
|
||||
errorMeta?: Record<string, unknown>;
|
||||
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;
|
||||
/**
|
||||
@@ -110,6 +136,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
|
||||
@@ -229,6 +261,7 @@ export interface AdapterExecutionContext {
|
||||
onLog: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
onMeta?: (meta: AdapterInvocationMeta) => Promise<void>;
|
||||
onEvent?: (event: AdapterRuntimeEvent) => Promise<void>;
|
||||
onUsage?: (receipt: AdapterUsageCheckpoint) => Promise<void>;
|
||||
onRuntimeProgress?: RuntimeStatusSink;
|
||||
/**
|
||||
* Reports that execution has crossed the adapter's dispatch boundary.
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
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",
|
||||
]);
|
||||
const stringFields = new Set(["type", "subtype", "role", "id", "session_id", "sessionID", "thread_id", "model", "modelID", "providerID"]);
|
||||
function project(value: unknown, depth = 0, field = ""): unknown {
|
||||
if (depth > 8) return undefined;
|
||||
if (typeof value === "number" || typeof value === "boolean" || value === null) return value;
|
||||
if (typeof value === "string") return stringFields.has(field) ? value.slice(0, 250) : undefined;
|
||||
if (Array.isArray(value)) return value.slice(0, 500).map(item => project(item, depth + 1));
|
||||
if (!value || typeof value !== "object") return undefined;
|
||||
return Object.fromEntries(Object.entries(value).filter(([key]) => fields.has(key))
|
||||
.map(([key, entry]) => [key, key === "modelUsage" && entry && typeof entry === "object"
|
||||
? Object.fromEntries(Object.entries(entry).map(([model, usage]) => [model.slice(0, 250), project(usage, depth + 1)]))
|
||||
: project(entry, depth + 1, key)]));
|
||||
}
|
||||
|
||||
/** A fresh instance belongs to one CLI attempt. Retries must not concatenate
|
||||
* cumulative usage from different attempts. Callback durability is awaited
|
||||
* before ordinary log publication. */
|
||||
export function createUsageCheckpointLog(
|
||||
onLog: (stream: "stdout" | "stderr", chunk: string) => Promise<void>,
|
||||
onUsage: ((receipt: AdapterUsageCheckpoint) => Promise<void>) | undefined,
|
||||
parse: (stdout: string) => AdapterUsageCheckpoint | null,
|
||||
) {
|
||||
if (!onUsage) return onLog;
|
||||
const attemptId = randomUUID();
|
||||
let remainder = "", accounting = "", previous = "";
|
||||
return async (stream: "stdout" | "stderr", chunk: string) => {
|
||||
if (stream === "stdout") {
|
||||
remainder += 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) {
|
||||
let raw: unknown;
|
||||
try { raw = JSON.parse(line); } catch { continue; }
|
||||
const compact = JSON.stringify(project(raw));
|
||||
if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"/.test(compact)) continue;
|
||||
accounting += compact + "\n";
|
||||
if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB");
|
||||
const receipt = parse(accounting);
|
||||
if (!receipt) continue;
|
||||
const serialized = JSON.stringify(receipt);
|
||||
if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; }
|
||||
}
|
||||
}
|
||||
await onLog(stream, chunk);
|
||||
};
|
||||
}
|
||||
import { randomUUID } from "node:crypto";
|
||||
@@ -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,6 +64,7 @@ import {
|
||||
} from "@paperclipai/adapter-utils/local-process-sandbox";
|
||||
import {
|
||||
claudeModelUsageTotals,
|
||||
claudeModelReceipts,
|
||||
parseClaudeStreamJson,
|
||||
describeClaudeFailure,
|
||||
detectClaudeLoginRequired,
|
||||
@@ -971,7 +973,12 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
onLog: createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseClaudeStreamJson(stdout);
|
||||
return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd,
|
||||
provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
billingType, model: parsed.model || model, complete: parsed.resultJson !== null };
|
||||
}),
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
terminalResultCleanup: {
|
||||
@@ -1012,6 +1019,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: proc.exitCode,
|
||||
signal: proc.signal,
|
||||
timedOut: true,
|
||||
usage: parsedStream.usage ?? undefined,
|
||||
usageBasis: "per_run",
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || model,
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
errorCode: "timeout",
|
||||
errorMeta,
|
||||
@@ -1073,6 +1087,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: proc.exitCode,
|
||||
signal: proc.signal,
|
||||
timedOut: false,
|
||||
usage: parsedStream.usage ?? undefined,
|
||||
usageBasis: "per_run",
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || model,
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
errorMessage: fallbackErrorMessage,
|
||||
errorCode,
|
||||
errorFamily,
|
||||
@@ -1241,13 +1262,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
retryNotBefore: transientRetryNotBefore ? transientRetryNotBefore.toISOString() : null,
|
||||
errorMeta,
|
||||
usage,
|
||||
usageByModel: claudeModelReceipts(parsed.modelUsage),
|
||||
...(usageBasis ? { usageBasis } : {}),
|
||||
sessionId: resolvedSessionId,
|
||||
sessionParams: resolvedSessionParams,
|
||||
sessionDisplayId: resolvedSessionId,
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || asString(parsed.model, model),
|
||||
model: Object.keys(parseObject(parsed.modelUsage)).length > 1 ? "mixed" : parsedStream.model || asString(parsed.model, model),
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
resultJson: mergedResultJson,
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -54,11 +54,25 @@ 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) {
|
||||
let sessionId: string | null = null;
|
||||
let model = "";
|
||||
let finalResult: Record<string, unknown> | null = null;
|
||||
const assistantTexts: string[] = [];
|
||||
const messageUsage = new Map<string, UsageSummary>();
|
||||
let anonymousMessage = 0;
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
@@ -76,6 +90,16 @@ export function parseClaudeStreamJson(stdout: string) {
|
||||
if (type === "assistant") {
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
const message = parseObject(event.message);
|
||||
const observed = parseObject(message.usage);
|
||||
if (Object.keys(observed).length > 0) {
|
||||
const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`;
|
||||
const previous = messageUsage.get(key);
|
||||
messageUsage.set(key, {
|
||||
inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)),
|
||||
cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)),
|
||||
outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)),
|
||||
});
|
||||
}
|
||||
const content = Array.isArray(message.content) ? message.content : [];
|
||||
for (const entry of content) {
|
||||
if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue;
|
||||
@@ -99,8 +123,11 @@ export function parseClaudeStreamJson(stdout: string) {
|
||||
sessionId,
|
||||
model,
|
||||
costUsd: null as number | null,
|
||||
usage: null as UsageSummary | null,
|
||||
usageBasis: null as "per_run" | null,
|
||||
usage: messageUsage.size ? [...messageUsage.values()].reduce<UsageSummary>((total, part) => ({
|
||||
inputTokens: total.inputTokens + part.inputTokens, outputTokens: total.outputTokens + part.outputTokens,
|
||||
cachedInputTokens: (total.cachedInputTokens ?? 0) + (part.cachedInputTokens ?? 0),
|
||||
}), { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }) : null,
|
||||
usageBasis: "per_run" as const,
|
||||
summary: assistantTexts.join("\n\n").trim(),
|
||||
resultJson: null as Record<string, unknown> | null,
|
||||
};
|
||||
@@ -109,7 +136,7 @@ export function parseClaudeStreamJson(stdout: string) {
|
||||
const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage);
|
||||
const usageObj = parseObject(finalResult.usage);
|
||||
const usage: UsageSummary = modelUsageTotals ?? {
|
||||
inputTokens: asNumber(usageObj.input_tokens, 0),
|
||||
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),
|
||||
};
|
||||
|
||||
@@ -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";
|
||||
@@ -1318,6 +1319,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
};
|
||||
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseCodexJsonl(stdout);
|
||||
return { usage: parsed.usage, usageBasis: "per_run", provider: "openai", biller: resolveCodexBiller(effectiveEnv, billingType), billingType, model, costUsd: null, complete: parsed.sawProtocolTerminalEvent };
|
||||
});
|
||||
try {
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
@@ -1331,7 +1336,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
onLog: async (stream, chunk) => {
|
||||
monitor?.noteOutputChunk(stream, chunk);
|
||||
if (stream === "stdout") {
|
||||
await onLog(stream, chunk);
|
||||
await accountingLog(stream, chunk);
|
||||
return;
|
||||
}
|
||||
const cleaned = stripCodexRolloutNoise(chunk);
|
||||
@@ -1423,6 +1428,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: "openai",
|
||||
biller: resolveCodexBiller(effectiveEnv, billingType),
|
||||
model,
|
||||
billingType,
|
||||
costUsd: null,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
clearSession: clearSessionOnMissingSession,
|
||||
};
|
||||
|
||||
@@ -28,7 +28,7 @@ describe("parseCodexJsonl", () => {
|
||||
sessionId: "thread_123",
|
||||
summary: "Recovered response",
|
||||
usage: {
|
||||
inputTokens: 10,
|
||||
inputTokens: 8,
|
||||
cachedInputTokens: 2,
|
||||
outputTokens: 4,
|
||||
},
|
||||
@@ -64,7 +64,7 @@ describe("parseCodexJsonl", () => {
|
||||
sessionId: "thread_123",
|
||||
summary: "Fixed the issue and verified the targeted tests pass.",
|
||||
usage: {
|
||||
inputTokens: 10,
|
||||
inputTokens: 8,
|
||||
cachedInputTokens: 2,
|
||||
outputTokens: 4,
|
||||
},
|
||||
|
||||
@@ -89,7 +89,8 @@ export function parseCodexJsonl(stdout: string) {
|
||||
return {
|
||||
sessionId,
|
||||
summary: finalMessage?.trim() ?? "",
|
||||
usage,
|
||||
// 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,
|
||||
|
||||
@@ -288,7 +288,7 @@ printf '%s\\n' '{"type":"result","subtype":"success","session_id":"cursor-sessio
|
||||
finalPreparedCommand = preferredCommandPath;
|
||||
const runtimeEnv = {
|
||||
...input.env,
|
||||
PATH: `${path.join(systemHomeDir, ".local", "bin")}${path.delimiter}${input.env.PATH}`,
|
||||
PATH: `${path.join(systemHomeDir, ".local", "bin")}${path.delimiter}${input.env.PATH ?? process.env.PATH ?? "/usr/bin:/bin"}`,
|
||||
};
|
||||
await fs.mkdir(path.dirname(preferredCommandPath), { recursive: true });
|
||||
await fs.writeFile(preferredCommandPath, preferredAgentScript);
|
||||
@@ -312,13 +312,14 @@ printf '%s\\n' '{"type":"result","subtype":"success","session_id":"cursor-sessio
|
||||
// shell seam faithful to that protocol instead of returning empty stdout
|
||||
// for every shell command.
|
||||
const runner = {
|
||||
execute: async (input: { command: string; args?: string[]; env?: Record<string, string> }) => {
|
||||
execute: async (input: { command: string; args?: string[]; env?: Record<string, string>; stdin?: string }) => {
|
||||
runnerState.commands.push(input.command);
|
||||
// Exercise actual bounded file reads during managed-home restoration;
|
||||
// reporting empty success for every shell command hides missing bytes.
|
||||
return runChildProcess(`cursor-fresh-lease-${runnerState.commands.length}`, input.command, input.args ?? [], {
|
||||
cwd: remoteWorkspace,
|
||||
env: { ...input.env, PATH: `${input.env?.PATH ?? ""}:/usr/bin:/bin` },
|
||||
stdin: input.stdin,
|
||||
timeoutSec: 30,
|
||||
graceSec: 5,
|
||||
onLog: async () => {},
|
||||
|
||||
@@ -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";
|
||||
@@ -626,11 +627,16 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseCursorJsonl(stdout);
|
||||
const provider = resolveProviderFromModel(model);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: false };
|
||||
});
|
||||
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}`;
|
||||
@@ -698,6 +704,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: providerFromModel,
|
||||
biller: resolveCursorBiller(effectiveEnv,
|
||||
billingType, providerFromModel),
|
||||
model,
|
||||
billingType,
|
||||
costUsd: attempt.parsed.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
clearSession: clearSessionOnMissingSession,
|
||||
};
|
||||
@@ -729,6 +743,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageBasis: "per_run",
|
||||
errorMessage:
|
||||
(attempt.proc.exitCode ?? 0) === 0
|
||||
? null
|
||||
|
||||
@@ -65,6 +65,14 @@ export function parseCursorJsonl(stdout: string) {
|
||||
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,
|
||||
@@ -102,7 +110,7 @@ export function parseCursorJsonl(stdout: string) {
|
||||
usageObj.output_tokens,
|
||||
asNumber(usageObj.outputTokens, 0),
|
||||
);
|
||||
totalCostUsd += asNumber(event.total_cost_usd, asNumber(event.cost_usd, asNumber(event.cost, 0)));
|
||||
addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost);
|
||||
|
||||
const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error";
|
||||
const resultText = asString(event.result, "").trim();
|
||||
@@ -143,10 +151,10 @@ export function parseCursorJsonl(stdout: string) {
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
usage.inputTokens += asNumber(tokens.input, 0);
|
||||
usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
|
||||
usage.cachedInputTokens += asNumber(cache.read, 0);
|
||||
usage.outputTokens += asNumber(tokens.output, 0);
|
||||
totalCostUsd += asNumber(part.cost, 0);
|
||||
addCost(part.cost);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
@@ -155,7 +163,7 @@ export function parseCursorJsonl(stdout: string) {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage,
|
||||
costUsd: totalCostUsd > 0 ? totalCostUsd : null,
|
||||
costUsd: reportedCost && !missingCost ? totalCostUsd : null,
|
||||
errorMessage,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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";
|
||||
@@ -650,7 +651,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
onLog: createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseGeminiJsonl(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null };
|
||||
}),
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
@@ -687,6 +691,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: "google",
|
||||
biller: "google",
|
||||
model,
|
||||
billingType,
|
||||
costUsd: attempt.parsed.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
errorCode: authMeta.requiresAuth
|
||||
? "gemini_auth_required"
|
||||
@@ -743,6 +754,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: failed ? fallbackErrorMessage : null,
|
||||
// Forward the transport-level error code from the run-disposition seam
|
||||
// first. A lost duplex control channel surfaces the typed
|
||||
|
||||
@@ -88,8 +88,8 @@ describe("parseGeminiJsonl", () => {
|
||||
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 +213,15 @@ describe("isGeminiTransientNetworkError", () => {
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("Gemini accounting semantics", () => {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -58,21 +58,19 @@ 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) {
|
||||
@@ -145,7 +143,8 @@ export function parseGeminiJsonl(stdout: string) {
|
||||
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 reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
|
||||
if (typeof reportedCost === "number") costUsd = reportedCost;
|
||||
const status = asString(event.status, "").toLowerCase();
|
||||
const isError =
|
||||
event.is_error === true ||
|
||||
@@ -183,7 +182,8 @@ export function parseGeminiJsonl(stdout: string) {
|
||||
|
||||
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;
|
||||
const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
|
||||
if (typeof reportedCost === "number") costUsd = reportedCost;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -648,6 +648,12 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageBasis: "per_run",
|
||||
provider: "moonshot",
|
||||
biller: "moonshot",
|
||||
model: model || null,
|
||||
billingType,
|
||||
costUsd: null,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
errorCode: authMeta.requiresAuth
|
||||
? "kimi_auth_required"
|
||||
@@ -704,6 +710,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: failed ? fallbackErrorMessage : null,
|
||||
// Forward the transport-level error code from the run-disposition seam
|
||||
// first. A lost duplex control channel surfaces the typed
|
||||
|
||||
@@ -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";
|
||||
@@ -654,7 +655,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
onLog: createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseOpenCodeJsonl(stdout);
|
||||
const provider = parseModelProvider(model || null);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false };
|
||||
}),
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
@@ -678,6 +683,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: parseModelProvider(model || null),
|
||||
biller: resolveOpenCodeBiller(runtimeEnv, parseModelProvider(model || null)),
|
||||
model,
|
||||
billingType: "unknown",
|
||||
costUsd: attempt.parsed.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
clearSession: clearSessionOnMissingSession,
|
||||
};
|
||||
@@ -715,6 +727,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: synthesizedExitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: (synthesizedExitCode ?? 0) === 0 ? null : fallbackErrorMessage,
|
||||
// Forward the transport-level error code from the run-disposition seam.
|
||||
// A lost duplex control channel surfaces the typed `duplex_channel_lost`
|
||||
|
||||
@@ -75,3 +75,12 @@ describe("parseOpenCodeJsonl", () => {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -29,7 +29,8 @@ export function parseOpenCodeJsonl(stdout: string) {
|
||||
cachedInputTokens: 0,
|
||||
outputTokens: 0,
|
||||
};
|
||||
let costUsd = 0;
|
||||
let costUsd: number | null = null;
|
||||
let missingCost = false;
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
@@ -54,10 +55,11 @@ export function parseOpenCodeJsonl(stdout: string) {
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
usage.inputTokens += asNumber(tokens.input, 0);
|
||||
usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
|
||||
usage.cachedInputTokens += asNumber(cache.read, 0);
|
||||
usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0);
|
||||
costUsd += asNumber(part.cost, 0);
|
||||
if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost;
|
||||
else missingCost = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -82,7 +84,7 @@ export function parseOpenCodeJsonl(stdout: string) {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage,
|
||||
costUsd,
|
||||
costUsd: missingCost ? null : costUsd,
|
||||
errorMessage: errors.length > 0 ? errors.join("\n") : null,
|
||||
toolErrors,
|
||||
};
|
||||
|
||||
@@ -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";
|
||||
@@ -744,7 +745,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: bufferedOnLog,
|
||||
onLog: createUsageCheckpointLog(bufferedOnLog, ctx.onUsage, stdout => {
|
||||
const parsed = parsePiJsonl(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false };
|
||||
}),
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
@@ -774,6 +778,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usage: attempt.parsed.usage, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), model, billingType: "unknown", costUsd: attempt.parsed.usage.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
clearSession: clearSessionOnMissingSession,
|
||||
};
|
||||
@@ -805,6 +810,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: effectiveExitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: (effectiveExitCode ?? 0) === 0 ? null : fallbackErrorMessage,
|
||||
// Forward the transport-level error code from the run-disposition seam.
|
||||
// A lost duplex control channel surfaces the typed `duplex_channel_lost`
|
||||
|
||||
@@ -288,3 +288,12 @@ describe("terminal provider failures", () => {
|
||||
.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);
|
||||
});
|
||||
});
|
||||
@@ -8,7 +8,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 }>;
|
||||
@@ -37,12 +37,17 @@ 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/)) {
|
||||
@@ -117,15 +122,13 @@ export function parsePiJsonl(stdout: string): ParsedPiOutput {
|
||||
// Extract usage and cost from assistant message
|
||||
const usage = asRecord(message.usage);
|
||||
if (usage) {
|
||||
result.usage.inputTokens += asNumber(usage.input, 0);
|
||||
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);
|
||||
if (cost) {
|
||||
result.usage.costUsd += asNumber(cost.total, 0);
|
||||
}
|
||||
addCost(cost?.total);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -213,21 +216,18 @@ export function parsePiJsonl(stdout: string): ParsedPiOutput {
|
||||
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.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);
|
||||
if (cost) {
|
||||
result.usage.costUsd += asNumber(cost.total ?? usage.costUsd, 0);
|
||||
} else {
|
||||
result.usage.costUsd += asNumber(usage.costUsd, 0);
|
||||
}
|
||||
addCost(cost?.total ?? usage.costUsd);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (missingCost) result.usage.costUsd = null;
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
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<typeof import("@paperclipai/adapter-utils/execution-target")>(),
|
||||
runAdapterExecutionTargetProcess: processResult,
|
||||
}));
|
||||
|
||||
type Execute = (context: AdapterExecutionContext) => Promise<AdapterExecutionResult>;
|
||||
const cases: Array<{ name: string; execute: () => Promise<Execute>; 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)("$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.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);
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
@@ -403,3 +403,12 @@ describe("cursor cli formatter", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("Cursor price completeness", () => {
|
||||
it("preserves reported zero and keeps a partially priced stream unpriced", () => {
|
||||
expect(parseCursorJsonl(JSON.stringify({ type: "result", cost_usd: 0 })).costUsd).toBe(0);
|
||||
expect(parseCursorJsonl(JSON.stringify({ type: "result" })).costUsd).toBeNull();
|
||||
expect(parseCursorJsonl([JSON.stringify({ type: "step_finish", part: { cost: 1 } }), JSON.stringify({ type: "step_finish", part: {} })].join("\n")).costUsd).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user