mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 12:33:05 +01:00
DottaandPaperclip
0eaabe5c0b
Settle paid operations through operation-scoped transactions. Preserve exact ledger evidence and replay while moving aggregate budget decisions to durable deferred work. Co-Authored-By: Paperclip <noreply@paperclip.ing>
377 lines
28 KiB
TypeScript
377 lines
28 KiB
TypeScript
import { compareCents, usdToUnits } from "@paperclipai/shared";
|
|
import { spawn, execFileSync, type ChildProcess } from "node:child_process";
|
|
import { once } from "node:events";
|
|
import { randomUUID } from "node:crypto";
|
|
import { mkdtempSync, readFileSync, writeFileSync, rmSync, existsSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { resolve } from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
import { and, eq, sql } from "drizzle-orm";
|
|
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
|
import { agents, budgetPolicies, companies, costEvents, createDb, heartbeatRuns, runUsageReceipts, type Db } from "@paperclipai/db";
|
|
import { type PrpEvent } from "../../vendor/paperclip-runner/index.js";
|
|
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
|
import { flushBudgetEvaluationWork } from "../budgets.js";
|
|
import { accountRunCost } from "../run-cost-accounting.js";
|
|
import { usageReceiptSpoolPath, type UsageReceiptEnvelope } from "../usage-receipts.js";
|
|
import { createNativeUsageContinuationDelivery, createNativeUsageDelivery } from "./native-usage-delivery.js";
|
|
|
|
const runnerRoot = fileURLToPath(new URL("../../../../packages/paperclip-runner/runner/", import.meta.url));
|
|
const fixtureProfile = process.env.CI ? "release" : "debug";
|
|
const billing = { provider: "openai", biller: "openai", billingType: "metered_api" as const };
|
|
|
|
describe("producer-owned native usage", () => {
|
|
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
|
let db: Db;
|
|
beforeAll(async () => {
|
|
execFileSync("cargo", ["build", ...(fixtureProfile === "release" ? ["--release"] : []), "--locked",
|
|
"--manifest-path", resolve(runnerRoot, "Cargo.toml"), "-p", "paperclip-runner-core", "--example", "recovery-fixture"],
|
|
{ timeout: 600_000, stdio: "pipe" });
|
|
database = await startEmbeddedPostgresTestDatabase("native-usage-delivery-");
|
|
db = createDb(database.connectionString);
|
|
}, 660_000);
|
|
afterAll(async () => { await database?.cleanup(); });
|
|
const seed = async () => {
|
|
const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID();
|
|
await db.insert(companies).values({ id: companyId, name: "Usage delivery", issuePrefix: randomUUID().slice(0, 8) });
|
|
await db.insert(agents).values({ id: agentId, companyId, name: "Producer", adapterType: "paperclip_runner" });
|
|
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "running", runtimeMode: "native", costAccountingPending: true });
|
|
return { companyId, agentId, runId, runnerInstanceId: randomUUID(), provider: { kind: "codex" as const, model: "test-model" }, billing };
|
|
};
|
|
const event = (input: Pick<Awaited<ReturnType<typeof seed>>, "runId" | "runnerInstanceId">, sequence: number, type: string, turnId = "turn-1", amount = 0.25): PrpEvent => ({
|
|
schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceKind: "runner", sourceInstanceId: input.runnerInstanceId,
|
|
sourceEventId: `usage-${sequence}`, sourceSeq: sequence, priority: 0, runId: input.runId,
|
|
normalizedSessionId: "session-1", turnId, emittedAt: new Date().toISOString(), eventType: type,
|
|
payload: type === "usage.reported" ? { runDeltaAvailable: true, runDeltaBasis: "run", runDelta: { inputTokens: 120, outputTokens: 10, providerCostUsd: amount } } : {},
|
|
} as PrpEvent);
|
|
|
|
const finalizedSource = async () => {
|
|
const input = await seed();
|
|
await db.update(heartbeatRuns).set({ runnerInstanceId: input.runnerInstanceId }).where(eq(heartbeatRuns.id, input.runId));
|
|
const delivery = await createNativeUsageDelivery(db, input);
|
|
const report = event(input, 1, "usage.reported"), done = event(input, 2, "turn.completed");
|
|
await delivery.capture(report);
|
|
await delivery.capture(done);
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(await accountRunCost(db, input.runId)).toBe(true);
|
|
const { billing: _billing, ...source } = input;
|
|
return { input, source, report, done };
|
|
};
|
|
|
|
it("continues exact finalized source receipts without acquiring ownership or resetting accounting", async () => {
|
|
const { input, source, report, done } = await finalizedSource();
|
|
const before = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
await expect(createNativeUsageDelivery(db, input)).rejects.toThrow("native_usage_run_not_available");
|
|
const continuation = await createNativeUsageContinuationDelivery(db, source);
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId))).toEqual(before);
|
|
await continuation.capture(report);
|
|
await continuation.capture(done);
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId))).toEqual(before);
|
|
expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId))).toHaveLength(2);
|
|
expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(1);
|
|
await expect(continuation.capture({ ...report, sourceInstanceId: randomUUID() })).rejects.toThrow("native_usage_source_mismatch");
|
|
await expect(continuation.capture({ ...report, payload: { ...report.payload as object, runDeltaAvailable: false } }))
|
|
.rejects.toThrow("native_usage_source_conflict");
|
|
});
|
|
|
|
it("journals an equal finalized source tail using the source billing without repricing", async () => {
|
|
const { input, source } = await finalizedSource();
|
|
const before = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
// Even a structurally supplied extra current-run classification cannot own
|
|
// the old receipt: the continuation derives billing from saved source data.
|
|
const withCurrentBilling = { ...source, billing: { provider: "anthropic", biller: "current-grant", billingType: "subscription_included" } };
|
|
const continuation = await createNativeUsageContinuationDelivery(db, withCurrentBilling);
|
|
await continuation.capture(event(input, 3, "usage.reported"));
|
|
const receipts = await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId));
|
|
expect(receipts).toHaveLength(3);
|
|
expect(receipts.find(row => row.sequence === 3)?.receiptJson.receipt).toMatchObject(billing);
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId))).toEqual(before);
|
|
expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(25);
|
|
});
|
|
|
|
it.each(["tokens", "cost", "incomplete", "unpriceable", "basis"])(
|
|
"retains a differing finalized source tail as unaccepted durable reconciliation debt: %s", async change => {
|
|
const { input, source } = await finalizedSource();
|
|
const continuation = await createNativeUsageContinuationDelivery(db, source);
|
|
const late = event(input, 3, "usage.reported");
|
|
if (change === "incomplete") (late.payload as Record<string, unknown>).runDeltaAvailable = false;
|
|
if (change === "basis") delete (late.payload as Record<string, unknown>).runDeltaBasis;
|
|
if (["tokens", "cost", "unpriceable"].includes(change)) (late.payload as Record<string, unknown>).runDelta = {
|
|
inputTokens: change === "tokens" ? 150 : change === "unpriceable" ? "unknown" : 120,
|
|
outputTokens: 10, providerCostUsd: change === "cost" ? 0.5 : 0.25,
|
|
};
|
|
await expect(continuation.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation");
|
|
await expect(continuation.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation");
|
|
expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId))).toHaveLength(3);
|
|
const [pending] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(pending.usageJson).toMatchObject({ accountingLateReceiptPending: true });
|
|
// Loading the durable journal on restart cannot turn rejected evidence
|
|
// into an acknowledged producer delivery.
|
|
const restarted = await createNativeUsageContinuationDelivery(db, source);
|
|
await expect(restarted.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation");
|
|
expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(25);
|
|
});
|
|
|
|
it.each(["company", "run", "runner", "source"])("rejects a changed continuation binding without mutation: %s", async change => {
|
|
const { input, source } = await finalizedSource();
|
|
if (change === "source") await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify({ accountingReceiptSourceId: randomUUID() })}::jsonb` })
|
|
.where(eq(heartbeatRuns.id, input.runId));
|
|
const before = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
await expect(createNativeUsageContinuationDelivery(db, { ...source,
|
|
...(change === "company" ? { companyId: randomUUID() } : {}),
|
|
...(change === "run" ? { runId: randomUUID() } : {}),
|
|
...(change === "runner" ? { runnerInstanceId: randomUUID() } : {}),
|
|
})).rejects.toThrow("native_usage_continuation_source_mismatch");
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId))).toEqual(before);
|
|
});
|
|
|
|
it("rejects changed immutable source receipt facts during hydration", async () => {
|
|
const { input, source } = await finalizedSource();
|
|
const [receipt] = await db.select().from(runUsageReceipts).where(and(eq(runUsageReceipts.runId, input.runId), eq(runUsageReceipts.sequence, 1)));
|
|
const changed = structuredClone(receipt.receiptJson) as UsageReceiptEnvelope;
|
|
changed.producerEvent!.usage!.runDelta = { inputTokens: 999, outputTokens: 10 };
|
|
await db.update(runUsageReceipts).set({ receiptJson: changed }).where(eq(runUsageReceipts.id, receipt.id));
|
|
await expect(createNativeUsageContinuationDelivery(db, source)).rejects.toThrow("native_usage_source_conflict");
|
|
});
|
|
|
|
it("retains the same Runner report through a failed Postgres receipt and an empty replacement receiver", async () => {
|
|
const input = await seed();
|
|
expect(existsSync(usageReceiptSpoolPath())).toBe(false);
|
|
const root = mkdtempSync(resolve(tmpdir(), "native-usage-runner-"));
|
|
const receiverA = resolve(root, "receiver-a"), receiverB = resolve(root, "receiver-b");
|
|
const identity = { runnerInstanceId: input.runnerInstanceId, environmentLeaseId: randomUUID(), runId: input.runId,
|
|
normalizedSessionId: randomUUID(), turnId: "turn-1", itemId: "item-1" };
|
|
const expectedRunnerVersion = "0.3.0", expectedRunnerDigest = `sha256:${"c".repeat(64)}`;
|
|
let receiver: ChildProcess | undefined, child: ChildProcess | undefined;
|
|
let exit: Promise<unknown[]> | undefined, failed = 0, lastCaptureError = "";
|
|
const receiptFile = resolve(root, "producer.json"), releaseFile = resolve(root, "release"), scriptFile = resolve(root, "script.json");
|
|
writeFileSync(scriptFile, JSON.stringify({ steps: 3, waitAfter: 2, retentionLimit: 8, receiptFile, releaseFile, accounting: true }));
|
|
const receipts = () => db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId));
|
|
try {
|
|
|
|
await db.execute(sql`CREATE FUNCTION fail_native_usage_receipt() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'receipt database unavailable'; END $$`);
|
|
await db.execute(sql`CREATE TRIGGER fail_native_usage BEFORE INSERT ON run_usage_receipts FOR EACH ROW EXECUTE FUNCTION fail_native_usage_receipt()`);
|
|
const startReceiver = () => new Promise<ChildProcess>((resolve, reject) => {
|
|
const process = spawn(globalThis.process.execPath, ["--import", fileURLToPath(new URL("../../../node_modules/tsx/dist/loader.mjs", import.meta.url)), fileURLToPath(new URL("./__fixtures__/usage-receiver.ts", import.meta.url))], {
|
|
cwd: fileURLToPath(new URL("../../../../", import.meta.url)), stdio: ["ignore", "ignore", "pipe", "ipc"], env: { ...globalThis.process.env },
|
|
});
|
|
let diagnostics = "";
|
|
process.stderr!.on("data", bytes => { diagnostics = (diagnostics + String(bytes)).slice(-16000); });
|
|
const timer = setTimeout(() => { process.kill("SIGKILL"); reject(new Error(`receiver startup timeout: ${diagnostics}`)); }, 60_000);
|
|
process.once("exit", code => { clearTimeout(timer); reject(new Error(`receiver exited ${code}: ${diagnostics}`)); });
|
|
process.on("message", message => {
|
|
const kind = (message as { kind?: string }).kind;
|
|
if (kind === "capture-failed") { failed++; lastCaptureError = String((message as { cause?: string }).cause); }
|
|
if (kind === "ready") { clearTimeout(timer); resolve(process); }
|
|
});
|
|
});
|
|
let requestId = 0;
|
|
const request = (target: ChildProcess, value: Record<string, unknown>) => new Promise<Record<string, unknown>>((resolve, reject) => {
|
|
const id = ++requestId;
|
|
const timer = setTimeout(() => { target.off("message", listener); reject(new Error("receiver response timeout")); }, 30_000);
|
|
const listener = (message: unknown) => {
|
|
const result = message as Record<string, unknown>;
|
|
if (result.requestId !== id) return;
|
|
clearTimeout(timer); target.off("message", listener);
|
|
if (result.error) reject(new Error(String(result.error))); else resolve(result);
|
|
};
|
|
target.on("message", listener);
|
|
target.send({ ...value, requestId: id });
|
|
});
|
|
receiver = await startReceiver();
|
|
const firstReceiver = await request(receiver, { kind: "start", connectionString: database.connectionString, directory: receiverA, identity, input });
|
|
const connectUrl = String(firstReceiver.connectUrl), port = Number(new URL(connectUrl).port), ticket = String(firstReceiver.ticket);
|
|
child = spawn(resolve(runnerRoot, `target/${fixtureProfile}/examples/recovery-fixture`), ["--fixture-script", scriptFile, "--connect-url", connectUrl,
|
|
"--state-dir", resolve(root, "producer"), "--runner-id", identity.runnerInstanceId, "--environment-lease-id", identity.environmentLeaseId,
|
|
"--run-id", identity.runId, "--session-id", identity.normalizedSessionId, "--turn-id", identity.turnId, "--item-id", identity.itemId,
|
|
"--runner-version", expectedRunnerVersion, "--runner-digest", expectedRunnerDigest, "--max-outbox-bytes", "65536", "--p0-reserve-bytes", "16384", "--reconnect-grace-ms", "120000"],
|
|
{ env: { ...process.env, PAPERCLIP_RUNNER_BOOTSTRAP_TICKET: ticket }, stdio: ["ignore", "ignore", "pipe"] });
|
|
exit = once(child, "exit");
|
|
await request(receiver, { kind: "command", type: "turn.start", commandId: "start-once" });
|
|
writeFileSync(releaseFile, "produce");
|
|
await vi.waitFor(() => expect(failed).toBeGreaterThan(0), { timeout: 10_000 });
|
|
expect(await receipts()).toHaveLength(0);
|
|
const producerState = () => JSON.parse(readFileSync(resolve(root, "producer/runner-state.json"), "utf8")) as {
|
|
outbox: Array<{ eventType: string; envelope: { payload: { sourceEventId: string } } }>;
|
|
};
|
|
const pendingUsage = producerState().outbox.find(report => report.eventType === "usage.reported");
|
|
expect(pendingUsage).toBeDefined();
|
|
const pendingSourceEventId = pendingUsage!.envelope.payload.sourceEventId;
|
|
const stoppedReceiver = await request(receiver, { kind: "stop" });
|
|
rmSync(receiverA, { recursive: true, force: true });
|
|
await db.execute(sql`DROP TRIGGER fail_native_usage ON run_usage_receipts`);
|
|
await db.execute(sql`DROP FUNCTION fail_native_usage_receipt()`);
|
|
|
|
receiver = await startReceiver();
|
|
await request(receiver, { kind: "start", connectionString: database.connectionString, directory: receiverB, identity, input, port, checkpoint: stoppedReceiver.checkpoint });
|
|
await vi.waitFor(async () => expect((await receipts()).filter(row => row.complete)).toHaveLength(1), { timeout: 15_000 });
|
|
await vi.waitFor(() => expect(producerState().outbox).toEqual([]), { timeout: 10_000 });
|
|
expect((await receipts()).some(row => (row.receiptJson.producerEvent as Record<string, unknown>).sourceEventId === pendingSourceEventId)).toBe(true);
|
|
expect((await receipts()).filter(row => row.receiptJson.producerEvent && (row.receiptJson.producerEvent as Record<string, unknown>).eventType === "usage.reported")).toHaveLength(1);
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(await accountRunCost(db, input.runId)).toBe(true);
|
|
expect(await accountRunCost(db, input.runId)).toBe(false);
|
|
const ledger = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId));
|
|
expect(ledger).toHaveLength(1);
|
|
expect(ledger[0].costCents).toBe(25);
|
|
const failedBeforeLateReport = failed;
|
|
await request(receiver, { kind: "command", type: "turn.steer", commandId: "release-late-usage" });
|
|
await vi.waitFor(() => expect(failed).toBeGreaterThan(failedBeforeLateReport), { timeout: 10_000 });
|
|
expect(lastCaptureError).toContain("native_usage_after_finalization_requires_reconciliation");
|
|
const retainedLate = producerState().outbox.find(report => report.eventType === "usage.reported");
|
|
expect(retainedLate).toBeDefined();
|
|
expect(retainedLate!.envelope.payload).toMatchObject({ runId: input.runId, sourceInstanceId: input.runnerInstanceId });
|
|
expect((await receipts()).filter(row => (row.receiptJson.producerEvent as Record<string, unknown>)?.eventType === "usage.reported")).toHaveLength(2);
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0].usageJson)
|
|
.toMatchObject({ accountingLateReceiptPending: true });
|
|
expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(1);
|
|
expect(existsSync(usageReceiptSpoolPath())).toBe(false);
|
|
} finally {
|
|
await db.execute(sql`DROP TRIGGER IF EXISTS fail_native_usage ON run_usage_receipts`);
|
|
await db.execute(sql`DROP FUNCTION IF EXISTS fail_native_usage_receipt()`);
|
|
if (receiver && receiver.exitCode === null && receiver.signalCode === null) receiver.kill("SIGKILL");
|
|
if (child && child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
|
|
await exit;
|
|
rmSync(root, { force: true, recursive: true });
|
|
}
|
|
}, 240_000);
|
|
|
|
it("preserves source takeover fencing, budget holds, and missing-usage debt", async () => {
|
|
const input = await seed();
|
|
const first = await createNativeUsageDelivery(db, input);
|
|
await db.insert(budgetPolicies).values({ companyId: input.companyId, scopeType: "agent", scopeId: input.agentId,
|
|
metric: "billed_cents", windowKind: "lifetime", amount: 10, notifyEnabled: false });
|
|
await first.capture(event(input, 1, "usage.reported"));
|
|
await flushBudgetEvaluationWork(db, { companyId: input.companyId });
|
|
const [agent] = await db.select().from(agents).where(eq(agents.id, input.agentId));
|
|
expect(agent.lifecycleHolds).toContain("budget");
|
|
const replacement = { ...input, runnerInstanceId: randomUUID() };
|
|
const second = await createNativeUsageDelivery(db, replacement);
|
|
await second.capture(event(replacement, 1, "usage.reported", "turn-2", 0.5));
|
|
await first.capture(event(input, 2, "turn.completed"));
|
|
const [pending] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(pending.usageJson).toMatchObject({ costUsd: 0.5, accountingReceiptReady: false });
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(await accountRunCost(db, input.runId)).toBe(false);
|
|
expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(0);
|
|
await second.capture(event(replacement, 2, "turn.completed", "turn-2"));
|
|
expect(await accountRunCost(db, input.runId)).toBe(true);
|
|
expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(50);
|
|
});
|
|
|
|
it.each(["tokens", "cost", "incomplete"])("accepts equal finalized snapshots and rejects changed %s without repricing", async change => {
|
|
const input = await seed();
|
|
const delivery = await createNativeUsageDelivery(db, input);
|
|
await delivery.capture(event(input, 1, "usage.reported"));
|
|
await delivery.capture(event(input, 2, "turn.completed"));
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(await accountRunCost(db, input.runId)).toBe(true);
|
|
await delivery.capture(event(input, 3, "usage.reported"));
|
|
const late = event(input, 4, "usage.reported");
|
|
(late.payload as Record<string, unknown>).runDelta = { inputTokens: change === "tokens" ? 150 : 120, outputTokens: change === "tokens" ? 15 : 10, providerCostUsd: change === "incomplete" ? 0.25 : 0.5 };
|
|
if (change === "incomplete") (late.payload as Record<string, unknown>).runDeltaAvailable = false;
|
|
await expect(delivery.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation");
|
|
await expect(delivery.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation");
|
|
expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(25);
|
|
});
|
|
|
|
it("accepts equal finalized rate-card reports without requiring terminal-only derived pricing", async () => {
|
|
const input = { ...await seed(), provider: { kind: "codex" as const, model: "gpt-6.1-sol" } };
|
|
const delivery = await createNativeUsageDelivery(db, input);
|
|
const report = (sequence: number) => {
|
|
const reported = event(input, sequence, "usage.reported");
|
|
reported.payload = { runDeltaBasis: "run", runDeltaAvailable: true, runDelta: { inputTokens: 1000, outputTokens: 100000 } };
|
|
return reported;
|
|
};
|
|
await delivery.capture(report(1));
|
|
await delivery.capture(event(input, 2, "turn.completed"));
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(await accountRunCost(db, input.runId)).toBe(true);
|
|
await delivery.capture(report(3));
|
|
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(run.usageJson).toMatchObject({ pricingProvenance: { source: "rate_card" }, accountingReceiptReady: true });
|
|
expect(usdToUnits(run.usageJson!.costUsdExact as string)).toBe(usdToUnits("1.002"));
|
|
const ledger = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId));
|
|
expect(ledger).toHaveLength(1);
|
|
expect(ledger[0].costCents).toBe(100.2);
|
|
const correction = report(4);
|
|
(correction.payload as Record<string, unknown>).runDelta = { inputTokens: 1000, outputTokens: 100000, providerCostUsd: 2 };
|
|
await expect(delivery.capture(correction)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation");
|
|
});
|
|
|
|
it("keeps successor exposure pending after prior-turn accounting invalidates its live snapshot", async () => {
|
|
const input = await seed();
|
|
const delivery = await createNativeUsageDelivery(db, input);
|
|
const paid = event(input, 1, "usage.reported");
|
|
paid.payload = { runDeltaBasis: "run", runDeltaAvailable: true, runDelta: { inputTokens: 50, outputTokens: 5, providerCostUsd: 0.5 } };
|
|
await delivery.capture(paid);
|
|
const invalidated = event(input, 2, "usage.reported");
|
|
invalidated.payload = { runDeltaBasis: "run", runDeltaAvailable: false, runDelta: {}, reason: "accounting_reconciliation_required",
|
|
reconciliation: { receivedTurnId: "old-provider-turn", cumulative: { inputTokens: 120, outputTokens: 12 }, priorOrigin: { runId: "old-run" } } };
|
|
await delivery.capture(invalidated);
|
|
await delivery.capture(event(input, 3, "turn.failed"));
|
|
await db.update(heartbeatRuns).set({ status: "failed", errorCode: "codex_usage_accounting_integrity", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(run).toMatchObject({ costAccountingPending: true, costAccountedAt: null,
|
|
usageJson: { accountingReceiptReady: false, costStatus: "unpriced" } });
|
|
expect(compareCents((run.usageJson?.budgetExposure as { costCents: string }).costCents, "50")).toBe(0);
|
|
const [required] = await db.select().from(runUsageReceipts).where(and(eq(runUsageReceipts.runId, input.runId), eq(runUsageReceipts.sequence, 2)));
|
|
expect(required.receiptJson.producerEvent).toMatchObject({ usage: { reconciliation: { receivedTurnId: "old-provider-turn", cumulative: { inputTokens: 120 }, priorOrigin: { runId: "old-run" } } } });
|
|
expect(await accountRunCost(db, input.runId)).toBe(false);
|
|
expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(0);
|
|
});
|
|
|
|
it("retains unmarked legacy Codex reports as incomplete unpriced debt", async () => {
|
|
const input = { ...await seed(), provider: { kind: "codex" as const, model: "gpt-6.1-sol" } };
|
|
const delivery = await createNativeUsageDelivery(db, input);
|
|
const legacy = event(input, 1, "usage.reported");
|
|
delete (legacy.payload as Record<string, unknown>).runDeltaBasis;
|
|
await delivery.capture(legacy);
|
|
await delivery.capture(event(input, 2, "turn.completed"));
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId));
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0]).toMatchObject({
|
|
costAccountingPending: true, costAccountedAt: null, usageJson: { accountingReceiptReady: false, costStatus: "unpriced" } });
|
|
expect(await accountRunCost(db, input.runId)).toBe(false);
|
|
expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(0);
|
|
});
|
|
|
|
it("replays partial and final usage once after a lost commit response and restores turn aggregation", async () => {
|
|
const input = { ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } };
|
|
let loseResponse = false;
|
|
const failingDb = new Proxy(db, { get(target, property) {
|
|
if (property === "transaction") return async (...args: Parameters<Db["transaction"]>) => {
|
|
const result = await db.transaction(...args);
|
|
if (loseResponse) { loseResponse = false; throw new Error("lost commit response"); }
|
|
return result;
|
|
};
|
|
const value = Reflect.get(target, property);
|
|
return typeof value === "function" ? value.bind(target) : value;
|
|
} }) as Db;
|
|
let delivery = await createNativeUsageDelivery(failingDb, input);
|
|
const first = event(input, 1, "usage.reported"), done = event(input, 2, "turn.completed");
|
|
loseResponse = true;
|
|
await expect(delivery.capture(first)).rejects.toThrow("lost commit response");
|
|
await delivery.capture(first);
|
|
await delivery.capture(done);
|
|
delivery = await createNativeUsageDelivery(db, input);
|
|
await delivery.capture(first);
|
|
await delivery.capture(done);
|
|
const nextTurn = event(input, 3, "turn.started", "turn-1");
|
|
nextTurn.payload = { providerTurnId: "provider-turn-2" };
|
|
await delivery.capture(nextTurn);
|
|
const partial = event(input, 4, "usage.reported", "turn-1", 0.5);
|
|
(partial.payload as Record<string, unknown>).runDeltaAvailable = false;
|
|
await delivery.capture(partial);
|
|
const final = event(input, 5, "usage.reported", "turn-1", 0.5);
|
|
await delivery.capture(final);
|
|
const terminal = event(input, 6, "turn.completed", "turn-1");
|
|
terminal.payload = { providerTurnId: "provider-turn-2" };
|
|
await delivery.capture(terminal);
|
|
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
|
|
expect(run.usageJson).toMatchObject({ inputTokens: 240, outputTokens: 20, accountingReceiptReady: true, costUsd: 0.75 });
|
|
expect(run.usageJson?.accountingCaptureFailed).not.toBe(true);
|
|
expect(await db.select().from(runUsageReceipts).where(and(eq(runUsageReceipts.runId, input.runId), eq(runUsageReceipts.sequence, 1)))).toHaveLength(1);
|
|
await expect(delivery.capture({ ...final, payload: { ...final.payload as object, runDeltaAvailable: false } })).rejects.toThrow("native_usage_source_conflict");
|
|
});
|
|
});
|