mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 09:07:46 +02:00
fix: flush attempt receipts and surface checkpoint failures
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3e77a70622
commit
f4100b256e
12 files changed
+198
-43
No files matched your search
@@ -0,0 +1,56 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createUsageCheckpointLog } from "./usage-checkpoint.js";
|
||||
import { runChildProcess } from "./server-utils.js";
|
||||
import type { AdapterUsageCheckpoint } from "./types.js";
|
||||
|
||||
const parse = (stdout: string): AdapterUsageCheckpoint => {
|
||||
const events = stdout.trim().split("\n").map(line => JSON.parse(line));
|
||||
return { costUsd: events.reduce((sum, event) => sum + event.usage.costUsd, 0), complete: false };
|
||||
};
|
||||
const event = JSON.stringify({ type: "usage", usage: { inputTokens: 2, costUsd: 0.25 }, text: "private content" });
|
||||
|
||||
describe("usage checkpoint stream", () => {
|
||||
it("flushes an unterminated Pi price and closes each attempt independently", async () => {
|
||||
const saved = vi.fn(); const parser = vi.fn(parse);
|
||||
const first = createUsageCheckpointLog(vi.fn(), saved, parser);
|
||||
await first("stdout", event);
|
||||
expect(saved).not.toHaveBeenCalled();
|
||||
await first.flush({ complete: true });
|
||||
const second = createUsageCheckpointLog(vi.fn(), saved, parse);
|
||||
await second("stdout", event + "\n");
|
||||
await second.flush({ complete: true });
|
||||
expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true });
|
||||
expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: true });
|
||||
expect(saved.mock.calls.at(-1)![0].attemptId).not.toBe(saved.mock.calls[0][0].attemptId);
|
||||
expect(parser.mock.calls[0][0]).not.toContain("private content");
|
||||
});
|
||||
|
||||
it("surfaces a persistence failure outside the local process's best-effort log path", async () => {
|
||||
const failure = new Error("Receipt storage unavailable");
|
||||
const output = vi.fn(); const saved = vi.fn().mockRejectedValue(failure);
|
||||
const log = createUsageCheckpointLog(output, saved, parse);
|
||||
const result = await runChildProcess("checkpoint-test", process.execPath, ["-e", `process.stdout.write(${JSON.stringify(event + "\n")})`], {
|
||||
cwd: process.cwd(), env: {}, timeoutSec: 5, graceSec: 1, onLog: log,
|
||||
});
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(output).toHaveBeenCalledWith("stdout", event + "\n");
|
||||
await expect(log.flush()).rejects.toBe(failure);
|
||||
});
|
||||
|
||||
it("does not promote partial usage to complete merely because the stream was flushed", async () => {
|
||||
const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, parse);
|
||||
await log("stdout", event); await log.flush();
|
||||
expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false });
|
||||
});
|
||||
|
||||
it("bounds replay work by stream size while retaining the exact final total", async () => {
|
||||
const parser = vi.fn(parse); const saved = vi.fn();
|
||||
const log = createUsageCheckpointLog(vi.fn(), saved, parser);
|
||||
for (let i = 0; i < 1024; i++) await log("stdout", event + "\n");
|
||||
await log.flush({ complete: true });
|
||||
expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 256, complete: true });
|
||||
expect(parser.mock.calls.length).toBeLessThanOrEqual(12);
|
||||
const finalSize = parser.mock.calls.at(-1)![0].length;
|
||||
expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThan(3 * finalSize);
|
||||
});
|
||||
});
|
||||
@@ -9,7 +9,7 @@ const fields = new Set([
|
||||
"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",
|
||||
"thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD", "costUsd",
|
||||
]);
|
||||
const stringFields = new Set(["type", "subtype", "role", "id", "session_id", "sessionID", "thread_id", "model", "modelID", "providerID"]);
|
||||
function project(value: unknown, depth = 0, field = ""): unknown {
|
||||
@@ -24,36 +24,63 @@ function project(value: unknown, depth = 0, field = ""): unknown {
|
||||
: 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. */
|
||||
/** A fresh instance belongs to one CLI attempt. Call flush() after its process
|
||||
* exits, before retrying or final result handling. Checkpoint failures are
|
||||
* retained outside runChildProcess's best-effort onLog error handler. */
|
||||
export function createUsageCheckpointLog(
|
||||
onLog: (stream: "stdout" | "stderr", chunk: string) => Promise<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; }
|
||||
let nextReplaySize = 0;
|
||||
let failure: unknown;
|
||||
let failed = false;
|
||||
function retain(line: string) {
|
||||
let raw: unknown;
|
||||
try { raw = JSON.parse(line); } catch { return; }
|
||||
const compact = JSON.stringify(project(raw));
|
||||
if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"|"model"|"modelID"/.test(compact)) return;
|
||||
accounting += compact + "\n";
|
||||
if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB");
|
||||
}
|
||||
async function publish(final = false, complete = false) {
|
||||
if (!onUsage || !accounting || (!final && accounting.length < nextReplaySize)) return;
|
||||
const parsed = parse(accounting);
|
||||
// Geometric replay bounds total parsing work by stream size. The first
|
||||
// observation is saved immediately; flush always saves the final snapshot.
|
||||
nextReplaySize = accounting.length * 2;
|
||||
if (!parsed) return;
|
||||
if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null &&
|
||||
!Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return;
|
||||
const receipt = { ...parsed, complete: parsed.complete || complete };
|
||||
const serialized = JSON.stringify(receipt);
|
||||
if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; }
|
||||
}
|
||||
const log = async (stream: "stdout" | "stderr", chunk: string) => {
|
||||
if (onUsage && !failed && stream === "stdout") {
|
||||
try {
|
||||
remainder += chunk;
|
||||
const lines = remainder.split("\n"); remainder = lines.pop() ?? "";
|
||||
if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB");
|
||||
for (const line of lines) retain(line);
|
||||
await publish();
|
||||
} catch (error) {
|
||||
failed = true; failure = error; remainder = ""; accounting = "";
|
||||
}
|
||||
}
|
||||
// Receipt persistence and ordinary logs have separate failure contracts.
|
||||
// A failed receipt must not hide the provider's diagnostic output.
|
||||
await onLog(stream, chunk);
|
||||
};
|
||||
return Object.assign(log, {
|
||||
async flush(options: { complete?: boolean } = {}) {
|
||||
if (failed) throw failure;
|
||||
if (!onUsage) return;
|
||||
if (remainder) { retain(remainder); remainder = ""; }
|
||||
await publish(true, options.complete ?? false);
|
||||
},
|
||||
});
|
||||
}
|
||||
import { randomUUID } from "node:crypto";
|
||||
@@ -964,6 +964,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const accountingLog = 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: Object.keys(parseObject(parsed.resultJson?.modelUsage)).length > 1 ? "mixed" : parsed.model || model,
|
||||
usageByModel: claudeModelReceipts(parsed.resultJson?.modelUsage), complete: parsed.resultJson !== null };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -973,12 +980,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
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 };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
terminalResultCleanup: {
|
||||
@@ -987,6 +989,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
},
|
||||
localProcessSandbox,
|
||||
});
|
||||
await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
|
||||
|
||||
const parsedStream = parseClaudeStreamJson(proc.stdout);
|
||||
const parsed = parsedStream.resultJson ?? parseJson(proc.stdout);
|
||||
@@ -1019,11 +1022,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: proc.exitCode,
|
||||
signal: proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: parsedStream.resultJson !== null,
|
||||
usage: parsedStream.usage ?? undefined,
|
||||
usageBasis: "per_run",
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || model,
|
||||
model: Object.keys(parseObject(parsedStream.resultJson?.modelUsage)).length > 1 ? "mixed" : parsedStream.model || model,
|
||||
usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage),
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
@@ -1091,7 +1096,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
usageBasis: "per_run",
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || model,
|
||||
model: Object.keys(parseObject(parsedStream.resultJson?.modelUsage)).length > 1 ? "mixed" : parsedStream.model || model,
|
||||
usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage),
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
errorMessage: fallbackErrorMessage,
|
||||
|
||||
@@ -1347,6 +1347,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
localProcessSandbox,
|
||||
});
|
||||
await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
|
||||
const cleanedStderr = stripCodexRolloutNoise(proc.stderr);
|
||||
return {
|
||||
proc: {
|
||||
@@ -1428,6 +1429,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawProtocolTerminalEvent,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: "openai",
|
||||
|
||||
@@ -630,7 +630,7 @@ 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 };
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult };
|
||||
});
|
||||
let stdoutLineBuffer = "";
|
||||
const emitNormalizedStdoutLine = async (rawLine: string) => {
|
||||
@@ -676,6 +676,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await flushStdoutChunk("", true);
|
||||
await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
|
||||
|
||||
return {
|
||||
proc,
|
||||
@@ -704,6 +705,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawResult,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: providerFromModel,
|
||||
|
||||
@@ -62,6 +62,7 @@ function readSessionId(event: Record<string, unknown>): string | null {
|
||||
|
||||
export function parseCursorJsonl(stdout: string) {
|
||||
let sessionId: string | null = null;
|
||||
let sawResult = false;
|
||||
const messages: string[] = [];
|
||||
let errorMessage: string | null = null;
|
||||
let totalCostUsd = 0;
|
||||
@@ -97,6 +98,7 @@ export function parseCursorJsonl(stdout: string) {
|
||||
}
|
||||
|
||||
if (type === "result") {
|
||||
sawResult = true;
|
||||
const usageObj = parseObject(event.usage);
|
||||
usage.inputTokens += asNumber(
|
||||
usageObj.input_tokens,
|
||||
@@ -160,6 +162,7 @@ export function parseCursorJsonl(stdout: string) {
|
||||
}
|
||||
|
||||
return {
|
||||
sawResult,
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage,
|
||||
|
||||
@@ -643,6 +643,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const accountingLog = 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 };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -651,13 +655,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
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 };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
|
||||
return {
|
||||
proc,
|
||||
parsed: parseGeminiJsonl(proc.stdout),
|
||||
@@ -691,6 +693,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.resultEvent !== null,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: "google",
|
||||
|
||||
@@ -648,6 +648,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: false,
|
||||
usageBasis: "per_run",
|
||||
provider: "moonshot",
|
||||
biller: "moonshot",
|
||||
|
||||
@@ -646,6 +646,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const accountingLog = 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 };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -655,14 +660,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
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 };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
|
||||
return {
|
||||
proc,
|
||||
rawStderr: proc.stderr,
|
||||
@@ -683,6 +685,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: false,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: parseModelProvider(model || null),
|
||||
|
||||
@@ -737,6 +737,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
};
|
||||
|
||||
const accountingLog = 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: parsed.sawAgentEnd };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -745,13 +749,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
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 };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush({ complete: !proc.timedOut && !proc.signal });
|
||||
|
||||
// Flush any remaining buffer content
|
||||
if (stdoutBuffer) {
|
||||
@@ -778,6 +780,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawAgentEnd,
|
||||
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,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { asNumber, asString, parseJson, parseObject } from "@paperclipai/adapter-utils/server-utils";
|
||||
|
||||
interface ParsedPiOutput {
|
||||
sawAgentEnd: boolean;
|
||||
sessionId: string | null;
|
||||
messages: string[];
|
||||
errors: string[];
|
||||
@@ -30,6 +31,7 @@ function extractTextContent(content: string | Array<{ type: string; text?: strin
|
||||
|
||||
export function parsePiJsonl(stdout: string): ParsedPiOutput {
|
||||
const result: ParsedPiOutput = {
|
||||
sawAgentEnd: false,
|
||||
sessionId: null,
|
||||
messages: [],
|
||||
errors: [],
|
||||
@@ -84,6 +86,7 @@ export function parsePiJsonl(stdout: string): ParsedPiOutput {
|
||||
}
|
||||
|
||||
if (eventType === "agent_end") {
|
||||
result.sawAgentEnd = true;
|
||||
const messages = event.messages as Array<Record<string, unknown>> | undefined;
|
||||
if (messages && messages.length > 0) {
|
||||
const lastMessage = messages[messages.length - 1];
|
||||
|
||||
@@ -31,6 +31,51 @@ const cases: Array<{ name: string; execute: () => Promise<Execute>; event: unkno
|
||||
describe("CLI adapter accounting on timeout", () => {
|
||||
const directories: string[] = [];
|
||||
afterEach(async () => { vi.clearAllMocks(); for (const dir of directories.splice(0)) await rm(dir, { recursive: true, force: true }); });
|
||||
it.each(cases.filter(fixture => fixture.name !== "kimi"))("$name flushes its attempt and surfaces checkpoint persistence failures", async (fixture) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir);
|
||||
const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 });
|
||||
const stdout = JSON.stringify(fixture.event);
|
||||
processResult.mockImplementation(async (_run, _target, _command, _args, options) => {
|
||||
// Local process logging catches callback failures. Finalization must
|
||||
// independently enforce receipt durability despite that contract.
|
||||
await options.onLog("stdout", stdout).catch(() => {});
|
||||
return { exitCode: 0, signal: null, timedOut: false, stdout, stderr: "", pid: 123, startedAt: new Date().toISOString() };
|
||||
});
|
||||
const onUsage = vi.fn();
|
||||
const context: AdapterExecutionContext = {
|
||||
runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
|
||||
runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
|
||||
config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [],
|
||||
env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } },
|
||||
context: {}, onLog: async () => {}, onUsage,
|
||||
};
|
||||
const execute = await fixture.execute();
|
||||
await execute(context);
|
||||
expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete: true });
|
||||
if (fixture.name === "claude") {
|
||||
const mixed = [
|
||||
{ type: "system", subtype: "init", model: "actual-model" },
|
||||
{ type: "result", total_cost_usd: 0.003, usage: { input_tokens: 3, output_tokens: 3 }, modelUsage: {
|
||||
"model-a": { inputTokens: 1, outputTokens: 1, costUSD: 0.001 },
|
||||
"model-b": { inputTokens: 2, outputTokens: 2, costUSD: 0.002 },
|
||||
} },
|
||||
];
|
||||
processResult.mockImplementation(async (_run, _target, _command, _args, options) => {
|
||||
const output = mixed.map(item => JSON.stringify(item)).join("\n");
|
||||
await options.onLog("stdout", output).catch(() => {});
|
||||
return { exitCode: 0, signal: null, timedOut: false, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() };
|
||||
});
|
||||
await execute(context);
|
||||
expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "mixed", costUsd: 0.003, complete: true, usageByModel: [
|
||||
{ model: "model-a", costUsd: 0.001 }, { model: "model-b", costUsd: 0.002 },
|
||||
] });
|
||||
delete (mixed[1] as { modelUsage?: unknown }).modelUsage;
|
||||
await execute(context);
|
||||
expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "actual-model", complete: true });
|
||||
}
|
||||
onUsage.mockRejectedValue(new Error("Receipt persistence failed"));
|
||||
await expect(execute(context)).rejects.toThrow("Receipt persistence failed");
|
||||
});
|
||||
it.each(cases)("$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 });
|
||||
@@ -44,6 +89,7 @@ describe("CLI adapter accounting on timeout", () => {
|
||||
context: {}, onLog: async () => {},
|
||||
});
|
||||
expect(result.timedOut).toBe(true);
|
||||
expect(result.usageComplete).toBe(["codex", "cursor", "gemini"].includes(fixture.name));
|
||||
expect(result.usageBasis).toBe("per_run");
|
||||
expect(result.provider).toBeTruthy();
|
||||
expect(result.billingType).toBeTruthy();
|
||||
|
||||
Reference in new issue
Block a user