mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
fix(hermes): settle closed retry work with unknown usage
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
f4ae618d38
commit
d968e1d002
15 files changed
+261
-31
No files matched your search
@@ -25697,10 +25697,14 @@ export function heartbeatService(
|
||||
const cacheAdjustedCostUsd = adapterResult.costUsdExact != null && adapterResult.cacheAdjustedCostUsd == null
|
||||
? null : resolveCacheAdjustedCostUsd(adapterResult);
|
||||
const usageJson: Record<string, unknown> = {
|
||||
accountingReceiptReady: adapterResult.usageComplete !== false,
|
||||
accountingReceiptReady: adapterResult.usageComplete !== false || adapterResult.settlement?.providerWorkEnded === true,
|
||||
accountingSettlement: adapterResult.settlement ?? null,
|
||||
accountingUsageComplete: adapterResult.usageComplete !== false,
|
||||
costUsdExact: adapterResult.costUsdExact ?? null,
|
||||
providerRequestId: adapterResult.providerRequestId ?? null,
|
||||
...(normalizedUsage ?? {}),
|
||||
...(adapterResult.settlement?.usageComplete === false
|
||||
? { inputTokens: null, outputTokens: null, cachedInputTokens: null, cacheWriteTokens: null }
|
||||
: normalizedUsage ?? {}),
|
||||
...(adapterResult.usageByModel ? { usageByModel: adapterResult.usageByModel } : {}),
|
||||
...(rawUsage
|
||||
? {
|
||||
|
||||
@@ -5462,17 +5462,19 @@ describe("native terminal-turn accounting", () => {
|
||||
if (scenario === "late_update") expect(result.usage).toMatchObject({ inputTokens: 32, outputTokens: 8 });
|
||||
});
|
||||
|
||||
it.each(["reported", "partial", "legacy", "wrong_account", "zero"] as const)("settles bound Hermes billing through the executor (%s)", async scenario => {
|
||||
it.each(["reported", "partial", "partial_missing_tokens", "reported_missing_tokens", "legacy", "wrong_account", "zero"] as const)("settles bound Hermes billing through the executor (%s)", async scenario => {
|
||||
const previous = process.env.PAPERCLIP_RUNNER_ACPX_QUALIFICATION;
|
||||
process.env.PAPERCLIP_RUNNER_ACPX_QUALIFICATION = JSON.stringify([{ agent: "hermes", model: "fixture-model" }]);
|
||||
try {
|
||||
const onUsage = vi.fn(async (_receipt: import("@paperclipai/adapter-utils").AdapterUsageCheckpoint) => {});
|
||||
const amount = scenario === "zero" ? 0 : 0.25;
|
||||
const tokensMissing = scenario.endsWith("missing_tokens");
|
||||
const partialPrice = scenario.startsWith("partial");
|
||||
const billing = { schema: "paperclip.usage.billing/v1", source: "provider_reported", biller: "openrouter", currency: "USD",
|
||||
complete: scenario !== "partial", requestCount: 2, reportedRequestCount: scenario === "partial" ? 1 : 2,
|
||||
complete: !partialPrice, requestCount: 2, reportedRequestCount: partialPrice ? 1 : 2,
|
||||
amountUsd: amount, amountUsdExact: amount.toFixed(9) };
|
||||
const notification = (await import("@paperclipai/paperclip-runner/live")).rehydrateRunnerdUsageNotification({
|
||||
runDeltaAvailable: true, runDelta: { inputTokens: 20, outputTokens: 5, cacheReadTokens: 3, cacheWriteTokens: 2, providerCostUsd: amount },
|
||||
runDeltaAvailable: !tokensMissing, runDelta: { inputTokens: 20, outputTokens: 5, cacheReadTokens: 3, cacheWriteTokens: 2, providerCostUsd: amount },
|
||||
...(scenario === "legacy" ? {} : { billing }),
|
||||
}, "session", "turn");
|
||||
const event = (eventType: string, sourceSeq: number, payload: object = {}) => ({ eventType, turnId: "turn", payload,
|
||||
@@ -5482,6 +5484,7 @@ describe("native terminal-turn accounting", () => {
|
||||
const usage = event("item.completed", 2, { kind: "usage", usage: notification.tokenUsage });
|
||||
await accountingEvents.committed!(usage);
|
||||
expect(onUsage.mock.calls.at(-1)![0].complete).toBe(false);
|
||||
expect(onUsage.mock.calls.at(-1)![0].settlement).toBeUndefined();
|
||||
await accountingEvents.committed!(event("turn.completed", 3));
|
||||
await accountingEvents.duplicate!(usage);
|
||||
return { result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" }, turnId: "turn",
|
||||
@@ -5492,9 +5495,12 @@ describe("native terminal-turn accounting", () => {
|
||||
const result = await executePaperclipNativeSession({ db: leaseDb(), runnerInstanceId: "runner", onUsage,
|
||||
billingIdentity: { provider: "deepseek", biller: scenario === "wrong_account" ? "anthropic" : "openrouter", billingType: "metered_api" },
|
||||
execution: { ...execution, provider: { kind: "acpx", agent: "hermes", model: "fixture-model", permissionMode: "approve-all", profile: QUALIFIED_ACPX_PROFILES.hermes } } as NativeExecutionInput });
|
||||
expect(result).toMatchObject({ usageComplete: true, costUsd: known ? amount : null,
|
||||
costStatus: known && scenario !== "partial" ? "reported" : "unpriced" });
|
||||
expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: true, costUsd: result.costUsd, costStatus: result.costStatus }));
|
||||
expect(result).toMatchObject({ usageComplete: !tokensMissing, costUsd: known ? amount : null,
|
||||
costStatus: known && !partialPrice ? "reported" : "unpriced" });
|
||||
expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: !tokensMissing, costUsd: result.costUsd, costStatus: result.costStatus }));
|
||||
if (known) expect(result.settlement).toEqual({ schema: "paperclip.accounting.settlement/v1", providerWorkEnded: true, usageComplete: !tokensMissing });
|
||||
else expect(result.settlement).toBeUndefined();
|
||||
if (tokensMissing) expect(result.usage).toBeUndefined();
|
||||
if (known) expect(result.pricingProvenance).toMatchObject({ source: "provider_reported", version: "hermes-openrouter-wire/v1" });
|
||||
} finally {
|
||||
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_ACPX_QUALIFICATION;
|
||||
|
||||
@@ -7929,6 +7929,7 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
let observedAccountingUsage: Record<string, unknown> | null = null;
|
||||
let observedAccountingTurn: string | undefined;
|
||||
let nativeAccountingComplete = false;
|
||||
let nativeSettlementReady = false;
|
||||
// OpenCode and ACPX report their current turn; Codex reports a cumulative run delta.
|
||||
// Rebuild per-turn state from the durable event log when a controller resumes.
|
||||
const selectedBiller = resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity).biller;
|
||||
@@ -7941,22 +7942,26 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
)).orderBy(heartbeatRunEvents.seq);
|
||||
for (const row of history) turnAccounting.observe(record(row.payload).prpEvent as PrpEvent);
|
||||
}
|
||||
const persistAccountingUsage = async (usage: Record<string, unknown> | null, complete: boolean) => {
|
||||
const persistAccountingUsage = async (usage: Record<string, unknown> | null, complete: boolean, settlementReady = false) => {
|
||||
const accountingUsage = normalizeNativeUsage(usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" });
|
||||
const accountingCost = accountingUsage ? nativeUsageCostUsd(usage, input.execution.provider, selectedBiller) : undefined;
|
||||
const hermes = input.execution.provider.kind === "acpx" && input.execution.provider.agent === "hermes";
|
||||
const accountingCost = accountingUsage || hermes ? nativeUsageCostUsd(usage, input.execution.provider, selectedBiller) : undefined;
|
||||
complete = complete && accountingUsage !== undefined;
|
||||
const costStatus = complete && usage?.accountingCostIncomplete !== true
|
||||
const settlement = hermes && settlementReady ? { schema: "paperclip.accounting.settlement/v1" as const, providerWorkEnded: true as const, usageComplete: complete } : undefined;
|
||||
const costStatus = (complete || settlementReady) && usage?.accountingCostIncomplete !== true
|
||||
? input.execution.provider.kind === "aws_agentcore" ? "estimated" as const : nativeHermesPriceEvidence(usage, input.execution.provider, selectedBiller) ? "reported" as const : undefined
|
||||
: "unpriced" as const;
|
||||
const billing = resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity);
|
||||
const costUsdExact = typeof usage?.accountingCostUsdExact === "string" ? usage.accountingCostUsdExact : undefined;
|
||||
await input.onUsage?.({ usage: accountingUsage, ...billing, complete, usageBasis: "per_run", costUsdExact,
|
||||
await input.onUsage?.({ usage: accountingUsage, ...billing, complete, settlement, usageBasis: "per_run", costUsdExact,
|
||||
model: input.execution.provider.model ?? "unknown", costUsd: accountingCost ?? null, costStatus,
|
||||
...nativeHermesPriceEvidence(usage, input.execution.provider, selectedBiller) });
|
||||
if (!input.onUsage) await input.db.update(heartbeatRuns).set({
|
||||
costAccountingPending: true,
|
||||
usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({
|
||||
...accountingUsage, ...billing, accountingReceiptReady: complete, model: input.execution.provider.model ?? "unknown",
|
||||
...(settlement?.usageComplete === false ? { inputTokens: null, outputTokens: null, cachedInputTokens: null, cacheWriteTokens: null } : accountingUsage),
|
||||
...billing, accountingReceiptReady: complete || !!settlement, accountingSettlement: settlement ?? null,
|
||||
accountingUsageComplete: complete, model: input.execution.provider.model ?? "unknown",
|
||||
costUsd: accountingCost ?? null, costUsdExact: costUsdExact ?? null, costStatus: costStatus ?? null, usageSource: "per_run",
|
||||
...nativeHermesPriceEvidence(usage, input.execution.provider, selectedBiller),
|
||||
})}::jsonb`,
|
||||
@@ -8785,10 +8790,11 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
const snapshot = turnAccounting.finish(native.turnId);
|
||||
observedAccountingUsage = snapshot.usage;
|
||||
nativeAccountingComplete = snapshot.complete && snapshot.turnId === native.turnId;
|
||||
nativeSettlementReady = snapshot.settlementReady && snapshot.turnId === native.turnId;
|
||||
} else nativeAccountingComplete = observedAccountingUsage !== null
|
||||
&& observedAccountingTurn !== undefined && observedAccountingTurn === native.turnId;
|
||||
native = { ...native, usage: observedAccountingUsage };
|
||||
await persistAccountingUsage(native.usage, nativeAccountingComplete);
|
||||
await persistAccountingUsage(native.usage, nativeAccountingComplete, nativeSettlementReady);
|
||||
try {
|
||||
await completeManagedNativeCredentialTurn(managedCredentialSession);
|
||||
} catch {
|
||||
@@ -9430,13 +9436,14 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
...resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity),
|
||||
model: input.execution.provider.model,
|
||||
usage: normalizeNativeUsage(native.usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" }),
|
||||
costUsd: normalizeNativeUsage(native.usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" })
|
||||
costUsd: (input.execution.provider.kind === "acpx" && input.execution.provider.agent === "hermes") || normalizeNativeUsage(native.usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" })
|
||||
? nativeUsageCostUsd(native.usage, input.execution.provider, selectedBiller) ?? null : null,
|
||||
costUsdExact: typeof native.usage?.accountingCostUsdExact === "string" ? native.usage.accountingCostUsdExact : undefined,
|
||||
costStatus: nativeAccountingComplete && native.usage?.accountingCostIncomplete !== true
|
||||
costStatus: (nativeAccountingComplete || nativeSettlementReady) && native.usage?.accountingCostIncomplete !== true
|
||||
? input.execution.provider.kind === "aws_agentcore" ? "estimated" : nativeHermesPriceEvidence(native.usage, input.execution.provider, selectedBiller) ? "reported" : undefined
|
||||
: "unpriced",
|
||||
usageComplete: nativeAccountingComplete,
|
||||
...(nativeSettlementReady ? { settlement: { schema: "paperclip.accounting.settlement/v1" as const, providerWorkEnded: true as const, usageComplete: nativeAccountingComplete } } : {}),
|
||||
...nativeHermesPriceEvidence(native.usage, input.execution.provider, selectedBiller),
|
||||
usageBasis: "per_run",
|
||||
nativeFinalization: finalization,
|
||||
@@ -9507,17 +9514,21 @@ export function nativeUsageBiller(provider: NativeExecutionInput["provider"]): s
|
||||
}
|
||||
|
||||
export function createNativeTurnAccounting(provider: NativeExecutionInput["provider"], selectedBiller?: string | null) {
|
||||
const hermes = provider.kind === "acpx" && provider.agent === "hermes";
|
||||
const turns = new Map<string, { usage: Record<string, unknown> | null; terminal: boolean }>();
|
||||
const sequences = new Map<string, number>();
|
||||
let currentTurn: string | undefined;
|
||||
const terminalTypes = ["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"];
|
||||
const snapshot = () => {
|
||||
if (turns.size === 0) return { usage: null, turnId: undefined, complete: false };
|
||||
if (turns.size === 0) return { usage: null, turnId: undefined, complete: false, settlementReady: false };
|
||||
let inputTokens = 0, outputTokens = 0, cacheReadTokens = 0, cacheWriteTokens = 0;
|
||||
let cost = 0n, costKnown = false, costIncomplete = false, complete = true;
|
||||
let requestCount = 0, reportedRequestCount = 0;
|
||||
let tokensComplete = true;
|
||||
let settlementReady = hermes;
|
||||
for (const [turnId, turn] of turns) {
|
||||
const normalized = normalizeNativeUsage(turn.usage);
|
||||
tokensComplete &&= normalized !== undefined;
|
||||
complete &&= turnId !== "unattributed" && turn.terminal && normalized !== undefined;
|
||||
inputTokens += (normalized?.inputTokens ?? 0) - (normalized?.cacheWriteTokens ?? 0);
|
||||
outputTokens += normalized?.outputTokens ?? 0;
|
||||
@@ -9526,6 +9537,8 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi
|
||||
const price = nativeUsageCostUsd(turn.usage, provider, selectedBiller);
|
||||
const billing = provider.kind === "acpx" && provider.agent === "hermes"
|
||||
? readProviderUsageBilling(turn.usage?.billing) : null;
|
||||
settlementReady &&= turnId !== "unattributed" && turn.terminal && billing !== null
|
||||
&& selectedBiller === billing.biller && turn.usage !== null && nativeUsageMeasurement(turn.usage).providerCostUsd === billing.amountUsd;
|
||||
if (price === undefined) costIncomplete = true;
|
||||
else {
|
||||
cost += usdToUnits(billing?.amountUsdExact ?? price); costKnown = true;
|
||||
@@ -9544,14 +9557,15 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi
|
||||
...(retainCost ? { providerCostUsd: Number(cost) / 1_000_000_000 } : {}) };
|
||||
const current = currentTurn ? turns.get(currentTurn) : undefined;
|
||||
const costExact = `${cost / 1_000_000_000n}.${String(cost % 1_000_000_000n).padStart(9, "0")}`;
|
||||
return { usage: { runDelta, accountingCostIncomplete: costIncomplete,
|
||||
return { usage: { runDelta, runDeltaComplete: hermes ? tokensComplete : true, accountingCostIncomplete: costIncomplete,
|
||||
...(retainCost && provider.kind === "acpx" && provider.agent === "hermes" ? { billing: {
|
||||
schema: "paperclip.usage.billing/v1", source: "provider_reported", biller: "openrouter", currency: "USD",
|
||||
complete: !costIncomplete, requestCount, reportedRequestCount,
|
||||
amountUsd: Number(costExact), amountUsdExact: costExact,
|
||||
} } : {}),
|
||||
...(retainCost ? { accountingCostUsdExact: costExact } : {}) },
|
||||
turnId: current && normalizeNativeUsage(current.usage) ? currentTurn : undefined, complete };
|
||||
turnId: current && (normalizeNativeUsage(current.usage) || (settlementReady && readProviderUsageBilling(current.usage?.billing))) ? currentTurn : undefined,
|
||||
complete, settlementReady };
|
||||
};
|
||||
return {
|
||||
observe(event: PrpEvent) {
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agents, companies, costEvents, createDb, heartbeatRuns, issues, nativeRunFinalizations, runUsageReceipts } from "@paperclipai/db";
|
||||
import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils";
|
||||
import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { accountRunCost } from "./run-cost-accounting.js";
|
||||
import { createRunUsageRecorder, persistUsageReceipt, usageReceiptSpoolPath } from "./usage-receipts.js";
|
||||
|
||||
describe("closed native usage settlement", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let home: string;
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
const companyId = randomUUID(), agentId = randomUUID();
|
||||
const settlement = { schema: "paperclip.accounting.settlement/v1", providerWorkEnded: true, usageComplete: false } as const;
|
||||
const partial: AdapterUsageCheckpoint = { provider: "deepseek", biller: "openrouter", model: "fixture-model",
|
||||
billingType: "metered_api", usageBasis: "per_run", complete: false, costStatus: "unpriced",
|
||||
costUsd: 0.25, costUsdExact: "0.250000000", settlement };
|
||||
|
||||
beforeAll(async () => {
|
||||
home = await mkdtemp(join(tmpdir(), "native-usage-settlement-"));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
temporary = await startEmbeddedPostgresTestDatabase("native-usage-settlement-");
|
||||
db = createDb(temporary.connectionString);
|
||||
await db.insert(companies).values({ id: companyId, name: "Usage settlement", issuePrefix: "USE" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Hermes", adapterType: "paperclip_runner", status: "idle" });
|
||||
});
|
||||
afterAll(async () => {
|
||||
await temporary?.cleanup();
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME;
|
||||
else process.env.PAPERCLIP_HOME = previousHome;
|
||||
if (home) await rm(home, { recursive: true, force: true });
|
||||
});
|
||||
async function createRun() {
|
||||
const runId = randomUUID(), issueId = randomUUID();
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Settle attempted usage", status: "in_progress", assigneeAgentId: agentId });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, runtimeMode: "native",
|
||||
status: "failed", finishedAt: new Date(), costAccountingPending: true,
|
||||
usageJson: { ledgerScope: { issueId }, inputTokens: 99, outputTokens: 5 } });
|
||||
return { runId, issueId, recorder: await createRunUsageRecorder(db, { companyId, runId, adapterType: "paperclip_runner" }) };
|
||||
}
|
||||
async function readRun(runId: string) {
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
return run;
|
||||
}
|
||||
|
||||
it("settles a retry's known subtotal exactly once after provider closure without inventing tokens", async () => {
|
||||
const { runId, issueId, recorder } = await createRun();
|
||||
// A previous accepted request's counters are not the complete retry total.
|
||||
await recorder.capture({ ...partial, settlement: undefined, usage: { inputTokens: 20, outputTokens: 2 } });
|
||||
const receipt = await recorder.complete({ exitCode: 1, signal: null, timedOut: false, ...partial, usageComplete: false });
|
||||
expect(receipt).toMatchObject({ complete: false, settlement, costUsdExact: "0.250000000" });
|
||||
expect(receipt.usage).toBeUndefined();
|
||||
const pending = await readRun(runId);
|
||||
expect(pending.usageJson).toMatchObject({ inputTokens: null, outputTokens: null, accountingUsageComplete: false, accountingReceiptReady: true });
|
||||
await db.insert(nativeRunFinalizations).values({ runId, issueId, companyId, phase: "observed" });
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect(await accountRunCost(db, runId)).toBe(true);
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
const [journal] = await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.id, String(pending.usageJson?.accountingReceiptId)));
|
||||
await persistUsageReceipt(db, journal.receiptJson as Parameters<typeof persistUsageReceipt>[1]);
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
const events = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId));
|
||||
expect(events).toHaveLength(1);
|
||||
expect(Number(events[0].costCents)).toBe(25);
|
||||
expect(events[0].costStatus).toBe("unpriced");
|
||||
expect(await readRun(runId)).toMatchObject({ costAccountingPending: false, costAccountedAt: expect.any(Date) });
|
||||
});
|
||||
|
||||
it("keeps a legacy incomplete checkpoint pending without closure authority", async () => {
|
||||
const { runId, recorder } = await createRun();
|
||||
await recorder.complete({ exitCode: 1, signal: null, timedOut: false, ...partial, settlement: undefined, usageComplete: false });
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
expect(await readRun(runId)).toMatchObject({ costAccountingPending: true, costAccountedAt: null });
|
||||
});
|
||||
|
||||
it("does not use a closed later attempt to settle an unfinished attempt", async () => {
|
||||
const { runId, recorder } = await createRun();
|
||||
await recorder.capture({ ...partial, attemptId: randomUUID(), settlement: undefined });
|
||||
const receipt = await recorder.capture({ ...partial, attemptId: randomUUID() });
|
||||
expect(receipt.settlement).toBeUndefined();
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
});
|
||||
|
||||
it("cannot mint closure from an invalid settlement or a later capture failure", async () => {
|
||||
const { runId, recorder } = await createRun();
|
||||
await expect(recorder.capture({ ...partial, settlement: { ...settlement, schema: "unknown/v1" } } as unknown as AdapterUsageCheckpoint)).rejects.toThrow();
|
||||
await recorder.complete({ exitCode: 1, signal: null, timedOut: false, ...partial, usageComplete: false });
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
expect((await readRun(runId)).usageJson).toMatchObject({ accountingCaptureFailed: true, accountingReceiptReady: false });
|
||||
});
|
||||
|
||||
it("preserves v1 replay while rejecting settlement that disagrees with completeness", async () => {
|
||||
const { runId, recorder } = await createRun();
|
||||
await expect(recorder.capture({ ...partial, complete: true })).rejects.toThrow("Settlement must preserve usage completeness");
|
||||
expect(await accountRunCost(db, runId)).toBe(false);
|
||||
expect(usageReceiptSpoolPath().startsWith(home)).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -29,7 +29,10 @@ const checkpointSchema = z.object({
|
||||
pricingProvenance: pricingProvenanceSchema.optional(),
|
||||
costUsdExact: z.string().max(160).refine(value => { try { return !value.startsWith("-") && usdToUnits(value) >= 0n; } catch { return false; } }, "Invalid exact USD amount").nullable().optional(),
|
||||
providerRequestId: z.string().max(250).nullable().optional(), complete: z.boolean(),
|
||||
});
|
||||
settlement: z.object({ schema: z.literal("paperclip.accounting.settlement/v1"),
|
||||
providerWorkEnded: z.literal(true), usageComplete: z.boolean() }).strict().optional(),
|
||||
}).refine(receipt => !receipt.settlement || receipt.settlement.usageComplete === receipt.complete,
|
||||
"Settlement must preserve usage completeness");
|
||||
const envelopeSchema = z.object({
|
||||
schema: z.literal("paperclip/accounting-receipt/v1"), id: z.string().uuid(), companyId: z.string().uuid(),
|
||||
runId: z.string().uuid(), sourceId: z.string().uuid(), sequence: z.number().int().positive(), receivedAt: z.iso.datetime(),
|
||||
@@ -73,11 +76,13 @@ export async function persistUsageReceipt(db: Db, raw: UsageReceiptEnvelope) {
|
||||
await tx.update(heartbeatRuns).set({
|
||||
costAccountingPending: true,
|
||||
accountingLastError: previous.accountingCaptureFailed === true ? "usage_capture_failed" : null,
|
||||
usageJson: { ...previous, ...receipt.usage,
|
||||
usageJson: { ...previous, ...(receipt.settlement?.usageComplete === false
|
||||
? { inputTokens: null, outputTokens: null, cachedInputTokens: null, cacheWriteTokens: null } : receipt.usage),
|
||||
provider: receipt.provider ?? "unknown", biller: receipt.biller ?? receipt.provider ?? "unknown", model: receipt.model ?? "unknown", billingType,
|
||||
costUsd: receipt.costUsd ?? null, costUsdExact: receipt.costUsdExact ?? null, cacheAdjustedCostUsd: receipt.cacheAdjustedCostUsd ?? null,
|
||||
usageByModel: receipt.usageByModel ?? null, usageSource: receipt.usageBasis ?? "per_run", providerRequestId: receipt.providerRequestId ?? null,
|
||||
accountingReceiptReady: previous.accountingCaptureFailed !== true && receipt.complete && receipt.usageBasis !== "session_cumulative", accountingReceiptId: envelope.id,
|
||||
accountingReceiptReady: previous.accountingCaptureFailed !== true && (receipt.complete || receipt.settlement?.providerWorkEnded === true) && receipt.usageBasis !== "session_cumulative", accountingReceiptId: envelope.id,
|
||||
accountingSettlement: receipt.settlement ?? null, accountingUsageComplete: receipt.complete,
|
||||
accountingReceiptSequence: envelope.sequence, accountingReceiptReceivedAt: envelope.receivedAt,
|
||||
costStatus: receipt.costStatus ?? null,
|
||||
pricingProvenance: receipt.pricingProvenance ?? { source: receipt.costUsd != null || receipt.costUsdExact != null ? "provider_reported" : "unknown", version: "accounting-receipt/v1" },
|
||||
@@ -237,7 +242,7 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string;
|
||||
const attempt = parsed.attemptId ?? currentAttempt;
|
||||
if (!attempts.has(attempt)) currentAttempt = attempt;
|
||||
const previous = attempts.get(attempt);
|
||||
const priced = priceCodexReceipt({ ...parsed, usage: parsed.usage ?? previous?.usage,
|
||||
const priced = priceCodexReceipt({ ...parsed, usage: parsed.settlement?.usageComplete === false ? undefined : parsed.usage ?? previous?.usage,
|
||||
complete: parsed.complete && !(previous?.usage && !parsed.usage) });
|
||||
priced.pricingProvenance ??= {
|
||||
source: priced.costUsd != null || priced.costUsdExact != null || priced.cacheAdjustedCostUsd != null ? "provider_reported" : "unknown",
|
||||
@@ -280,6 +285,14 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string;
|
||||
// can be considered complete, even if a later attempt succeeded.
|
||||
receipt.complete = !captureFailed && parts.every(part => part.complete);
|
||||
}
|
||||
// A closed provider boundary can settle a known subtotal while token or
|
||||
// price totals remain unknown. Old incomplete checkpoints have no such
|
||||
// authority; capture failure and an unfinished attempt still keep debt open.
|
||||
if (parts.some(part => part.settlement)) {
|
||||
receipt.settlement = !captureFailed && parts.every(part => part.complete || part.settlement?.providerWorkEnded === true)
|
||||
? { schema: "paperclip.accounting.settlement/v1", providerWorkEnded: true, usageComplete: receipt.complete } : undefined;
|
||||
if (parts.some(part => part.settlement?.usageComplete === false)) receipt.usage = undefined;
|
||||
}
|
||||
const envelope: UsageReceiptEnvelope = { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), ...input, sourceId, sequence: ++sequence, receivedAt: new Date().toISOString(), receipt };
|
||||
const file = await spoolUsageReceipt(envelope, directory);
|
||||
try { await persistUsageReceipt(db, envelope); await fs.rm(file, { force: true }); }
|
||||
@@ -297,7 +310,7 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string;
|
||||
// A bounded stdout tail can lose a terminal record already saved from the
|
||||
// full stream. Keep that stronger evidence, but never hide additional
|
||||
// usage/charges or certify a snapshot inherited from a replaced recorder.
|
||||
if (sequence > 0 && previous?.complete && result.usageComplete === false) {
|
||||
if (sequence > 0 && previous?.complete && result.usageComplete === false && !result.settlement) {
|
||||
const moreTokens = (["inputTokens", "cachedInputTokens", "cacheWriteTokens", "outputTokens"] as const)
|
||||
.some(key => (final.usage?.[key] ?? 0) > (previous.usage?.[key] ?? 0));
|
||||
const finalPrice = final.cacheAdjustedCostUsd ?? final.costUsdExact ?? final.costUsd;
|
||||
|
||||
Reference in new issue
Block a user