PaperClipAI/server/src/services/native-runtime/native-usage-delivery.integration.test.ts

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 { activityLog, 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 { accountingIntegrityService } from "../accounting-integrity.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.each([
    { name: "a report whose terminal was already acknowledged", types: ["usage.reported"] },
    { name: "a report and its terminal", types: ["usage.reported", "turn.completed"] },
  ])("settles a finished run's outbox with no receipts yet: $name", async ({ types }) => {
    const input = await seed();
    await db.update(heartbeatRuns).set({ runnerInstanceId: input.runnerInstanceId }).where(eq(heartbeatRuns.id, input.runId));
    await createNativeUsageDelivery(db, input);
    const { billing, ...source } = input;
    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(),
      usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify(billing)}::jsonb` }).where(eq(heartbeatRuns.id, input.runId));
    const continuation = await createNativeUsageContinuationDelivery(db, source);
    for (const [index, type] of types.entries()) await continuation.capture(event(input, index + 1, type));
    expect(await accountRunCost(db, input.runId)).toBe(true);
    expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))
      .toEqual([expect.objectContaining({ biller: billing.biller, billingType: billing.billingType, outputTokens: 10 })]);
  });

  it("keeps a lone report open while the run has not finished", async () => {
    const input = await seed();
    await db.update(heartbeatRuns).set({ runnerInstanceId: input.runnerInstanceId }).where(eq(heartbeatRuns.id, input.runId));
    await createNativeUsageDelivery(db, input);
    const { billing, ...source } = input;
    await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify(billing)}::jsonb` })
      .where(eq(heartbeatRuns.id, input.runId));
    const continuation = await createNativeUsageContinuationDelivery(db, source);
    await continuation.capture(event(input, 1, "usage.reported"));
    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
    expect(run!.usageJson).toMatchObject({ accountingReceiptReady: false });
  });

  it("refuses to continue a run with no receipts and no recorded classification", async () => {
    const input = await seed();
    await db.update(heartbeatRuns).set({ runnerInstanceId: input.runnerInstanceId }).where(eq(heartbeatRuns.id, input.runId));
    await createNativeUsageDelivery(db, input);
    const { billing: _billing, ...source } = input;
    await expect(createNativeUsageContinuationDelivery(db, source)).rejects.toThrow("native_usage_continuation_receipts_missing");
  });

  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 durable debt without failing delivery: %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 continuation.capture(late);
      await continuation.capture(late);
      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 keeps the evidence owed.
      const restarted = await createNativeUsageContinuationDelivery(db, source);
      await restarted.capture(late);
      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(async () => expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0].usageJson)
        .toMatchObject({ accountingLateReceiptPending: true }), { timeout: 10_000 });
      // Late evidence is acknowledged: it must not wedge the producer's outbox.
      await vi.waitFor(() => expect(producerState().outbox).toEqual([]), { timeout: 10_000 });
      expect(failed).toBe(failedBeforeLateReport);
      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);

  describe("late usage reported on a later run", () => {
    const sessionId = randomUUID();
    const lateReport = (input: Awaited<ReturnType<typeof seed>>, priorOrigin: Record<string, unknown> | null, sequence = 7) => {
      const report = event(input, sequence, "usage.reported");
      report.normalizedSessionId = sessionId;
      report.payload = { runDeltaBasis: "run", runDeltaAvailable: false, runDelta: { inputTokens: 30, outputTokens: 3, providerCostUsd: 0.05 },
        reconciliation: { reason: "accounting_reconciliation_required", receivedTurnId: "provider-turn-1", cumulative: { inputTokens: 150, outputTokens: 13 }, priorOrigin } };
      return report;
    };
    const successor = async (owner: Awaited<ReturnType<typeof finalizedSource>>) => {
      const runId = randomUUID();
      await db.update(heartbeatRuns).set({ nativeSessionId: sessionId }).where(eq(heartbeatRuns.id, owner.input.runId));
      await db.insert(heartbeatRuns).values({ id: runId, companyId: owner.input.companyId, agentId: owner.input.agentId, invocationSource: "on_demand",
        status: "running", runtimeMode: "native", nativeSessionId: sessionId, costAccountingPending: true });
      return { ...owner.input, runId };
    };
    const ledger = (runId: string) => db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId));

    it.each([["exact", true], ["inferred", false]] as const)("posts the increment to the %s owner and leaves the delivering run alone", async (attribution, named) => {
      const owner = await finalizedSource();
      const current = await successor(owner);
      const delivery = await createNativeUsageDelivery(db, current);
      const report = lateReport(current, named ? { runId: owner.input.runId, normalizedSessionId: sessionId } : null);
      await delivery.capture(report);
      await delivery.capture(report);
      const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, current.runId));
      expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, current.runId))).toHaveLength(0);
      expect(before.usageJson).not.toHaveProperty("inputTokens");
      expect(await accountRunCost(db, owner.input.runId)).toBe(true);
      expect(await accountRunCost(db, owner.input.runId)).toBe(false);
      const charges = await ledger(owner.input.runId);
      expect(charges).toHaveLength(2);
      expect(charges.map(row => row.inputTokens).sort((a, b) => a - b)).toEqual([30, 120]);
      expect(charges.reduce((sum, row) => sum + row.costCents, 0)).toBeCloseTo(30);
      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.input.runId));
      expect(run.usageJson).toMatchObject({ inputTokens: 150, outputTokens: 13, accountingLateReceiptPending: false });
      expect(run.accountingLastError).toBeNull();
      expect(run.costAccountingPending).toBe(false);
      const [activity] = await db.select().from(activityLog).where(and(eq(activityLog.entityId, owner.input.runId), eq(activityLog.action, "cost.late_usage_posted")));
      expect(activity.details).toMatchObject({ attribution, reportedByRunId: current.runId, inputTokens: 30 });
      const inspection = await accountingIntegrityService(db).inspect(owner.input.companyId);
      expect(inspection.findings.filter(finding => finding.entityId === owner.input.runId)).toEqual([]);
    });

    it("waits for the owner's charge before posting its late usage", async () => {
      const ownerInput = await seed();
      await db.update(heartbeatRuns).set({ runnerInstanceId: ownerInput.runnerInstanceId, nativeSessionId: sessionId }).where(eq(heartbeatRuns.id, ownerInput.runId));
      const ownerDelivery = await createNativeUsageDelivery(db, ownerInput);
      await ownerDelivery.capture(event(ownerInput, 1, "usage.reported"));
      await ownerDelivery.capture(event(ownerInput, 2, "turn.completed"));
      await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, ownerInput.runId));
      const runId = randomUUID();
      await db.insert(heartbeatRuns).values({ id: runId, companyId: ownerInput.companyId, agentId: ownerInput.agentId, invocationSource: "on_demand",
        status: "running", runtimeMode: "native", nativeSessionId: sessionId, costAccountingPending: true });
      const current = { ...ownerInput, runId };
      await (await createNativeUsageDelivery(db, current)).capture(lateReport(current, { runId: ownerInput.runId, normalizedSessionId: sessionId }));
      expect(await accountRunCost(db, ownerInput.runId)).toBe(true);
      expect(await ledger(ownerInput.runId)).toHaveLength(1);
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ownerInput.runId)))[0].costAccountingPending).toBe(true);
      expect(await accountRunCost(db, ownerInput.runId)).toBe(true);
      expect(await ledger(ownerInput.runId)).toHaveLength(2);
    });

    it("keeps an older Runner's invalidation without an increment on the delivering run", async () => {
      const owner = await finalizedSource();
      const current = await successor(owner);
      const report = lateReport(current, { runId: owner.input.runId, normalizedSessionId: sessionId });
      (report.payload as Record<string, unknown>).runDelta = { inputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, outputTokens: 0, providerCostUsd: 0, requests: 0, activeSeconds: 0 };
      await (await createNativeUsageDelivery(db, current)).capture(report);
      expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, current.runId))).toHaveLength(1);
      expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.input.runId)))[0].usageJson).not.toHaveProperty("accountingLateReceiptPending", true);
    });

    it("charges the delivering run when its session has no finished run, never another session's", async () => {
      const other = await finalizedSource();
      await db.update(heartbeatRuns).set({ nativeSessionId: randomUUID() }).where(eq(heartbeatRuns.id, other.input.runId));
      const runId = randomUUID();
      await db.insert(heartbeatRuns).values({ id: runId, companyId: other.input.companyId, agentId: other.input.agentId, invocationSource: "on_demand",
        status: "running", runtimeMode: "native", nativeSessionId: sessionId, costAccountingPending: true });
      const current = { ...other.input, runId };
      const delivery = await createNativeUsageDelivery(db, current);
      await delivery.capture(lateReport(current, { runId: other.input.runId, normalizedSessionId: sessionId }));
      await delivery.capture(event(current, 8, "usage.reported"));
      await delivery.capture(event(current, 9, "turn.completed"));
      await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId));
      expect(await accountRunCost(db, runId)).toBe(true);
      expect(await accountRunCost(db, runId)).toBe(true);
      expect((await ledger(runId)).map(row => row.inputTokens).sort((a, b) => a - b)).toEqual([30, 120]);
      expect(await ledger(other.input.runId)).toHaveLength(1);
      const [activity] = await db.select().from(activityLog).where(and(eq(activityLog.entityId, runId), eq(activityLog.action, "cost.late_usage_posted")));
      expect(activity.details).toMatchObject({ attribution: "inferred", reportedByRunId: runId });
      expect((await accountingIntegrityService(db).inspect(current.companyId)).findings.filter(finding => finding.entityId === runId)).toEqual([]);
    });
  });

  it("posts a same-run token increase as a separate auditable charge", async () => {
    const { input, source } = await finalizedSource();
    const continuation = await createNativeUsageContinuationDelivery(db, source);
    const late = event(input, 3, "usage.reported");
    (late.payload as Record<string, unknown>).runDelta = { inputTokens: 150, outputTokens: 15, providerCostUsd: 0.5 };
    await continuation.capture(late);
    expect(await accountRunCost(db, input.runId)).toBe(true);
    const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId));
    expect(charges.map(row => row.inputTokens).sort((a, b) => a - b)).toEqual([30, 120]);
    expect(charges.reduce((sum, row) => sum + row.costCents, 0)).toBeCloseTo(50);
    expect((await accountingIntegrityService(db).inspect(input.companyId)).findings.filter(finding => finding.entityId === input.runId)).toEqual([]);
  });

  it("prices a same-run late report from the rate card instead of posting free tokens", 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, outputTokens: number) => {
      const reported = event(input, sequence, "usage.reported");
      reported.payload = { runDeltaBasis: "run", runDeltaAvailable: true, runDelta: { inputTokens: 1000, outputTokens } };
      return reported;
    };
    await delivery.capture(report(1, 100000));
    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, 120000));
    expect(await accountRunCost(db, input.runId)).toBe(true);
    const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId));
    expect(charges).toHaveLength(2);
    expect(charges.every(charge => charge.costStatus !== "unpriced")).toBe(true);
    const late = charges.find(charge => charge.outputTokens === 20000)!;
    expect(late.costCents).toBeGreaterThan(0);
    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
    expect(run.usageJson).toMatchObject({ outputTokens: 120000, accountingLateReceiptPending: false });
    expect((await accountingIntegrityService(db).inspect(input.companyId)).findings.filter(finding => finding.entityId === input.runId)).toEqual([]);
  });

  it("labels a rate-card late charge as an estimate when the original charge was reported", 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, runDelta: Record<string, unknown>) => {
      const reported = event(input, sequence, "usage.reported");
      reported.payload = { runDeltaBasis: "run", runDeltaAvailable: true, runDelta };
      return reported;
    };
    await delivery.capture(report(1, { inputTokens: 1000, outputTokens: 100000, providerCostUsd: 0.5 }));
    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, { inputTokens: 1000, outputTokens: 120000 }));
    expect(await accountRunCost(db, input.runId)).toBe(true);
    const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId));
    const original = charges.find(charge => charge.outputTokens === 100000)!;
    const late = charges.find(charge => charge.outputTokens === 20000)!;
    expect(original.costStatus).not.toBe("estimated");
    expect(late.costStatus).toBe("estimated");
    expect(late.pricingProvenance).toMatchObject({ source: "rate_card" });
  });

  it("keeps lower late usage as visible debt instead of guessing", async () => {
    const { input, source } = await finalizedSource();
    const continuation = await createNativeUsageContinuationDelivery(db, source);
    const late = event(input, 3, "usage.reported");
    (late.payload as Record<string, unknown>).runDelta = { inputTokens: 90, outputTokens: 15, providerCostUsd: 0.5 };
    await continuation.capture(late);
    expect(await accountRunCost(db, input.runId)).toBe(false);
    expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(1);
    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0].accountingLastError).toBe("late_usage_requires_reconciliation");
  });

  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 keeps changed %s as debt 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 delivery.capture(late);
    await delivery.capture(late);
    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 delivery.capture(correction);
  });

  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);
  });

  // A replacement Runner fails the turn it lost on restore before the
  // continuation turn runs. Remote delivery must keep that marker, or the
  // receipt stays incomplete until the closure bound.
  it("closes a replacement source's receipt past a turn lost on restore, as a lower bound", async () => {
    const input = { ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } };
    const lost = event(input, 26, "turn.failed", "turn-lost");
    lost.payload = { status: "failed", error: { code: "provider_turn_lost_on_restore", recoverable: true } };
    const started = event(input, 29, "turn.started", "turn-next");
    const report = event(input, 30, "usage.reported", "turn-next");
    const done = event(input, 31, "turn.completed", "turn-next");
    let delivery = await createNativeUsageDelivery(db, input);
    for (const value of [lost, started, report, done]) await delivery.capture(value);
    const read = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0];
    expect((await read()).usageJson).toMatchObject({ accountingReceiptReady: true, outputTokens: 10,
      pricingProvenance: { evidence: expect.stringContaining("lower bound") } });
    // The saved receipts replay into a reopened receiver without conflict.
    delivery = await createNativeUsageDelivery(db, input);
    for (const value of [lost, started, report, done]) await delivery.capture(value);
    expect((await read()).usageJson).toMatchObject({ accountingReceiptReady: true });
  });

  describe("a turn restored mid-turn on a remote Runner", () => {
    const restoredSeed = async () => ({ ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } });
    const partial = (input: Awaited<ReturnType<typeof restoredSeed>>, sequence: number) => {
      const value = event(input, sequence, "usage.reported");
      (value.payload as Record<string, unknown>).runDeltaAvailable = false;
      return value;
    };
    const read = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0];

    // The local Runner rule from #20517: K1 stopped the provider before the
    // turn's final report; the restored Runner reported the turn completed.
    it("closes the turn as a lower bound, also after the controller restarts between restore and terminal", async () => {
      const input = await restoredSeed();
      const history = [event(input, 1, "turn.started"), partial(input, 2), event(input, 3, "session.resumed", undefined as never)];
      const done = event(input, 4, "turn.completed");
      let delivery = await createNativeUsageDelivery(db, input);
      for (const value of history) await delivery.capture(value);
      // A replacement controller rebuilds the receiver from the saved journal.
      delivery = await createNativeUsageDelivery(db, input);
      await delivery.capture(done);
      expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: true,
        pricingProvenance: { evidence: expect.stringContaining("lower bound") } });
      // Replaying every event into another receiver is idempotent.
      delivery = await createNativeUsageDelivery(db, input);
      for (const value of [...history, done]) await delivery.capture(value);
      expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: true });
    });

    it("keeps the turn's own final report when it arrives after the restore", async () => {
      const input = await restoredSeed();
      const delivery = await createNativeUsageDelivery(db, input);
      for (const value of [event(input, 1, "turn.started"), partial(input, 2), event(input, 3, "session.resumed", undefined as never),
        event(input, 4, "usage.reported"), event(input, 5, "turn.completed")]) await delivery.capture(value);
      const usage = (await read(input.runId)).usageJson;
      expect(usage).toMatchObject({ accountingReceiptReady: true, outputTokens: 10 });
      expect(JSON.stringify(usage?.pricingProvenance ?? {})).not.toContain("lower bound");
    });

    it("keeps a turn started after the restore pending, and journals nothing for a restore with no open turn", async () => {
      const input = await restoredSeed();
      const delivery = await createNativeUsageDelivery(db, input);
      for (const value of [event(input, 1, "session.resumed", undefined as never), event(input, 2, "turn.started"), partial(input, 3),
        event(input, 4, "turn.completed")]) await delivery.capture(value);
      expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: false });
      expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId))).toHaveLength(3);
    });
  });

  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");
  });
});