mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 09:07:46 +02:00
Merge full-stream accounting fixes into quota stack
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
beb28fef65
3 files changed
+59
-14
No files matched your search
@@ -84,7 +84,7 @@ function parseModelProvider(model: string | null): string | null {
|
||||
}
|
||||
|
||||
function resolveOpenCodeBiller(env: Record<string, string>, provider: string | null): string {
|
||||
return inferOpenAiCompatibleBiller(env, null) ?? provider ?? "unknown";
|
||||
return provider === "openai" ? inferOpenAiCompatibleBiller(env, "openai") ?? "unknown" : provider ?? "unknown";
|
||||
}
|
||||
|
||||
const REMOTE_OPENCODE_MODELS_PROBE_DEFAULT_TIMEOUT_SEC = 20;
|
||||
@@ -647,7 +647,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
|
||||
const consumeAccounting = createOpenCodeJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
let hasAccounting = false;
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage ?? (async () => {}), stdout => {
|
||||
hasAccounting = true;
|
||||
const parsed = consumeAccounting(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 };
|
||||
@@ -666,11 +668,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush({ complete: proc.exitCode === 0 && !proc.timedOut && !proc.signal });
|
||||
return {
|
||||
proc,
|
||||
rawStderr: proc.stderr,
|
||||
parsed: parseOpenCodeJsonl(proc.stdout),
|
||||
};
|
||||
// Display output is capped by the process transport. Keep accounting
|
||||
// from the full stream, including when no checkpoint callback is installed.
|
||||
const parsed = parseOpenCodeJsonl(proc.stdout);
|
||||
if (hasAccounting) {
|
||||
const retained = consumeAccounting("");
|
||||
parsed.usage = retained.usage;
|
||||
parsed.costUsd = retained.costUsd;
|
||||
}
|
||||
return { proc, rawStderr: proc.stderr, parsed };
|
||||
};
|
||||
|
||||
const toResult = (
|
||||
|
||||
@@ -142,7 +142,7 @@ async function buildPiSkillsDir(config: Record<string, unknown>): Promise<string
|
||||
}
|
||||
|
||||
function resolvePiBiller(env: Record<string, string>, provider: string | null): string {
|
||||
return inferOpenAiCompatibleBiller(env, null) ?? provider ?? "unknown";
|
||||
return provider === "openai" ? inferOpenAiCompatibleBiller(env, "openai") ?? "unknown" : provider ?? "unknown";
|
||||
}
|
||||
|
||||
async function ensureSessionsDir(): Promise<string> {
|
||||
@@ -738,7 +738,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
};
|
||||
|
||||
const consumeAccounting = createPiJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(bufferedOnLog, ctx.onUsage, stdout => {
|
||||
let hasAccounting = false;
|
||||
const accountingLog = createUsageCheckpointLog(bufferedOnLog, ctx.onUsage ?? (async () => {}), stdout => {
|
||||
hasAccounting = true;
|
||||
const parsed = consumeAccounting(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: parsed.sawAgentEnd };
|
||||
});
|
||||
@@ -761,11 +763,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
await onLog("stdout", stdoutBuffer);
|
||||
}
|
||||
|
||||
return {
|
||||
proc,
|
||||
rawStderr: proc.stderr,
|
||||
parsed: parsePiJsonl(proc.stdout),
|
||||
};
|
||||
// Display output is capped by the process transport. Keep accounting
|
||||
// from the full stream, including when no checkpoint callback is installed.
|
||||
const parsed = parsePiJsonl(proc.stdout);
|
||||
if (hasAccounting) {
|
||||
const retained = consumeAccounting("");
|
||||
parsed.usage = retained.usage;
|
||||
parsed.sawAgentEnd = retained.sawAgentEnd;
|
||||
}
|
||||
return { proc, rawStderr: proc.stderr, parsed };
|
||||
};
|
||||
|
||||
const toResult = (
|
||||
|
||||
@@ -86,6 +86,39 @@ describe("CLI adapter accounting on timeout", () => {
|
||||
onUsage.mockRejectedValue(new Error("Receipt persistence failed"));
|
||||
await expect(execute(context)).rejects.toThrow("Receipt persistence failed");
|
||||
});
|
||||
it.each(cases.filter(f => ["pi", "opencode"].includes(f.name)))("$name preserves full-stream accounting after stdout truncation", async (fixture) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-tail-")); directories.push(dir);
|
||||
const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'anthropic test\\n'\n", { mode: 0o755 });
|
||||
const execute = await fixture.execute();
|
||||
for (const checkpoint of [true, false]) for (const timedOut of [false, true]) for (const missingPrice of [false, true]) {
|
||||
const early = structuredClone(fixture.event) as any;
|
||||
if (missingPrice) {
|
||||
if (fixture.name === "pi") delete early.message.usage.cost;
|
||||
else delete early.part.cost;
|
||||
}
|
||||
const end = fixture.name === "pi" && !timedOut ? '\n{"type":"agent_end","messages":[]}' : "";
|
||||
const first = JSON.stringify(early) + "\n";
|
||||
const noise = JSON.stringify({ type: "text", part: { text: "x".repeat(5 * 1024 * 1024) } }) + "\n";
|
||||
const last = JSON.stringify(fixture.event) + end;
|
||||
processResult.mockImplementation(async (_run, _target, _command, _args, options) => {
|
||||
for (const chunk of [first, noise, last]) await options.onLog("stdout", chunk);
|
||||
return { exitCode: timedOut ? null : 0, signal: timedOut ? "SIGTERM" : null, timedOut, stdout: (first + noise + last).slice(-4 * 1024 * 1024), stderr: "", pid: 123, startedAt: new Date().toISOString() };
|
||||
});
|
||||
const onUsage = checkpoint ? vi.fn() : undefined;
|
||||
const result = await execute({
|
||||
runId: "tail-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
|
||||
runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
|
||||
config: { engine: "cli", command, cwd: dir, model: "anthropic/test", paperclipRuntimeSkills: [], env: { ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } },
|
||||
context: {}, onLog: async () => {}, onUsage,
|
||||
});
|
||||
expect(result.usage).toMatchObject({ inputTokens: 40, outputTokens: 20, cachedInputTokens: 200 });
|
||||
expect(result.costUsd).toBe(missingPrice ? null : 0.008);
|
||||
expect(result.biller).toBe("anthropic");
|
||||
expect(result.usageComplete).toBe(!timedOut);
|
||||
if (onUsage) expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: result.usage, costUsd: result.costUsd });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(cases)("$name retains observed accounting when the process times out", async (fixture) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir);
|
||||
const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 });
|
||||
|
||||
Reference in new issue
Block a user