diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index b69d63a9bc5..922c5cfbe17 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1884,7 +1884,10 @@ to block budget admission. After native recovery reaches terminal failure, an incomplete receipt holds new work as recovering accounting; it does not open a hard-stop incident. When the run no longer records a provider process (recovery clears it only after a -verified stop), no producer can deliver the receipt. Accounting then records +verified stop), no producer can deliver the receipt. A remote run reaches the +same point when every sandbox allocation it used carries a provider +termination receipt (stopped or destroyed); a sandbox that may still run keeps +the debt pending. Accounting then records its known part as an unpriced charge, so the unpriced-usage policy governs admission. A later receipt still posts through late-receipt reconciliation. A controller that detaches for server restart loses checkpoint and completion diff --git a/server/src/__tests__/budget-enforcement-worker.test.ts b/server/src/__tests__/budget-enforcement-worker.test.ts index f84d39d54d9..73a21c18e55 100644 --- a/server/src/__tests__/budget-enforcement-worker.test.ts +++ b/server/src/__tests__/budget-enforcement-worker.test.ts @@ -147,6 +147,25 @@ describe("budget accounting delivery", () => { expect(work.accounting).toHaveBeenCalledTimes(3); }); + // A sandbox stop or lease release can close a remote run's lost receipt. + it.each([DELIVERY_QUEUES.environmentCleanup, DELIVERY_QUEUES.retainedNativeSessionCleanup])( + "wakes accounting when %s work commits", async queue => { + const s = setup(); + const recover = vi.fn(async () => {}); + const worker = registerCostAccountingWork(s.db, s.coordinator, recover); + subscriptions.push(worker.stop); + await worker.ready; + const before = recover.mock.calls.length; + await s.owner.transaction(tx => notifyDeliveryWork(tx as object, queue)); + await vi.advanceTimersByTimeAsync(1); + expect(recover.mock.calls.length).toBeGreaterThan(before); + worker.stop(); + const stopped = recover.mock.calls.length; + await s.owner.transaction(tx => notifyDeliveryWork(tx as object, queue)); + await vi.advanceTimersByTimeAsync(1); + expect(recover).toHaveBeenCalledTimes(stopped); + }); + it("replays durable spool work after a pre-transaction failure and releases its local wake on stop", async () => { const s = setup(); const recover = vi.fn(async () => {}); diff --git a/server/src/__tests__/closed-native-receipt-accounting.test.ts b/server/src/__tests__/closed-native-receipt-accounting.test.ts index fde4b43b357..15bb4b01243 100644 --- a/server/src/__tests__/closed-native-receipt-accounting.test.ts +++ b/server/src/__tests__/closed-native-receipt-accounting.test.ts @@ -2,7 +2,8 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { - agents, budgetIncidents, budgetPolicies, budgetReservations, companies, costEvents, createDb, heartbeatRunEvents, heartbeatRuns, issues, + agents, budgetIncidents, budgetPolicies, budgetReservations, companies, costEvents, createDb, environmentLeases, environments, + heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, type Db, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -210,6 +211,57 @@ const support = await getEmbeddedPostgresTestSupport(); expect(policy.id).toBe(f.policy.id); }); + // Daytona wave: a remote run's lost receipt held all new company work even + // after its sandbox was destroyed. A remote run has no local stop event and + // keeps its recorded remote process identity. + async function remoteFixture(options: { status?: "failed" | "cancelled"; leases?: Array<{ + receipt?: "destroyed" | "stopped" | "none" | "foreign"; released?: boolean; + }> } = {}) { + const f = await fixture({ processPid: 443, stopEvidence: false }); + if (options.status) await db.update(heartbeatRuns).set({ status: options.status }).where(eq(heartbeatRuns.id, f.run.id)); + const [environment] = await db.insert(environments).values({ name: `Daytona ${f.run.id}`, driver: "sandbox", + config: { provider: "daytona" } }).returning(); + for (const lease of options.leases ?? [{ receipt: "destroyed" }]) { + const id = randomUUID(), providerLeaseId = randomUUID(), released = lease.released !== false; + const receipt = lease.receipt ?? "destroyed"; + await db.insert(environmentLeases).values({ id, environmentId: environment.id, companyId: f.companyId, + issueId: f.issue.id, heartbeatRunId: f.run.id, provider: "daytona", providerLeaseId, leasePolicy: "ephemeral", + status: released ? "expired" : "active", cleanupStatus: released ? "success" : null, + releasedAt: released ? new Date() : null, + metadata: receipt === "none" ? {} : { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1", + companyId: f.companyId, runId: f.run.id, leaseId: id, provider: "daytona", + providerLeaseId: receipt === "foreign" ? randomUUID() : providerLeaseId, + state: receipt === "foreign" ? "destroyed" : receipt, confirmedAt: new Date().toISOString() } } }); + } + return f; + } + + it.each(["failed", "cancelled"] as const)("settles a %s remote run once every sandbox has a termination receipt", async status => { + const f = await remoteFixture({ status, leases: [{ receipt: "destroyed" }, { receipt: "stopped" }] }); + expect(await f.block()).not.toBeNull(); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + expect((await f.readRun()).costAccountingPending).toBe(false); + expect(await f.charges()).toMatchObject([{ costStatus: "unpriced" }]); + expect((await f.reservation()).state).toBe("settled"); + expect(await f.block()).toBeNull(); + }); + + it.each([ + ["a sandbox that may still run", [{ receipt: "none", released: false }]], + // Older plugins release successfully without returning a receipt. + ["a released sandbox without a termination receipt", [{ receipt: "none" }]], + ["one of two sandboxes without a receipt", [{ receipt: "destroyed" }, { receipt: "none", released: false }]], + ["a receipt for another sandbox", [{ receipt: "foreign" }]], + ["no sandbox at all", []], + ] as const)("keeps a remote run's debt pending with %s", async (_label, leases) => { + const f = await remoteFixture({ leases: [...leases] }); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); + expect(await accountRunCost(db, f.run.id)).toBe(false); + expect((await f.readRun()).costAccountingPending).toBe(true); + expect(await f.charges()).toHaveLength(0); + }); + it("keeps the debt pending while native recovery can still produce a result", async () => { const f = await fixture({ phase: "observed" }); expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); diff --git a/server/src/services/budget-enforcement-worker.ts b/server/src/services/budget-enforcement-worker.ts index ad567f5527b..e5edb73609b 100644 --- a/server/src/services/budget-enforcement-worker.ts +++ b/server/src/services/budget-enforcement-worker.ts @@ -1,7 +1,7 @@ import type { Db } from "@paperclipai/db"; import { withBudgetSpan } from "./budget-tracing.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; -import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "./delivery-work-notifications.js"; import { deliverBudgetEnforcement, hasPendingBudgetEnforcement, @@ -87,8 +87,13 @@ export function registerCostAccountingWork(db: Db, coordinator: Coordinator, rec if (event.type === "heartbeat.run.status" && ["succeeded", "failed", "timed_out", "cancelled", "interrupted"].includes(String(event.payload.status))) worker.wake(); }); + // A released or stopped sandbox can close a terminal run's lost receipt. + // Those lease writers signal environment or retained-session cleanup. + const unsubscribeLeases = [DELIVERY_QUEUES.environmentCleanup, DELIVERY_QUEUES.retainedNativeSessionCleanup] + .map(queue => subscribeDeliveryWork(db, queue, worker.wake)); const stop = () => { unsubscribe(); + for (const unsubscribeLease of unsubscribeLeases) unsubscribeLease(); if (accountingWakes.get(db) === worker.wake) accountingWakes.delete(db); }; return { ...worker, stop }; diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index aaa192c08f0..cbd92ff10e7 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -1,7 +1,7 @@ import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { usdToUnits, unitsToCents } from "@paperclipai/shared"; import { and, asc, eq, inArray, or, sql } from "drizzle-orm"; -import { budgetReservations, agents, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; +import { budgetReservations, agents, environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; import { withOperationAccountingTransaction } from "./accounting-transaction.js"; import { withBudgetSpan } from "./budget-tracing.js"; import { conflict } from "../errors.js"; @@ -39,25 +39,49 @@ async function readAccountingClosure(db: Db, runId: string) { return closure; } -/** A failed native run's receipt source is closed only with positive stop - * evidence: recovery ended in terminal failure, the run no longer records a - * process identity, and its latest server-authored process event is the - * verified local stop. A deadline failure whose process still runs keeps its - * identity and has no stop event, so it stays pending. Missing identity alone is not proof. A cancelled run - * keeps its accounting drain: Stop recovery can still deliver the final - * receipt through the original or a reattached recorder. Remote runs keep - * their debt until a receipt arrives. One predicate serves the reconciler +/** Every remote allocation of the run carries a provider termination receipt + * bound to that exact lease and run (mirrors hasRemoteTerminationReceipt): the + * sandbox is stopped or destroyed, so no runner there can report usage. The + * per-lease check uses "is not true": a missing receipt makes it NULL, which + * must never count as proof. */ +const remoteExecutionTerminated = sql`exists (select 1 from ${environmentLeases} remote_lease + where remote_lease.company_id = ${heartbeatRuns.companyId} and remote_lease.heartbeat_run_id = ${heartbeatRuns.id}) + and not exists (select 1 from ${environmentLeases} remote_lease + where remote_lease.company_id = ${heartbeatRuns.companyId} and remote_lease.heartbeat_run_id = ${heartbeatRuns.id} + and (remote_lease.provider is not null and remote_lease.provider <> 'local' + and remote_lease.provider_lease_id is not null and remote_lease.released_at is not null + and remote_lease.cleanup_status = 'success' and remote_lease.status in ('released', 'expired', 'failed') + and remote_lease.metadata->'remoteExecutionTermination'->>'schema' = 'paperclip.remote-termination.v1' + and remote_lease.metadata->'remoteExecutionTermination'->>'companyId' = remote_lease.company_id::text + and remote_lease.metadata->'remoteExecutionTermination'->>'runId' = remote_lease.heartbeat_run_id::text + and remote_lease.metadata->'remoteExecutionTermination'->>'leaseId' = remote_lease.id::text + and remote_lease.metadata->'remoteExecutionTermination'->>'provider' = remote_lease.provider + and remote_lease.metadata->'remoteExecutionTermination'->>'providerLeaseId' = remote_lease.provider_lease_id + and remote_lease.metadata->'remoteExecutionTermination'->>'state' in ('stopped', 'destroyed')) is not true)`; + +/** A native run's receipt source is closed only with positive stop evidence, + * after recovery ended in terminal failure without a result: + * - Local: the run no longer records a process identity and its latest + * server-authored process event is the verified local stop. Only a failed + * run closes this way; a cancelled local run keeps its accounting drain, + * since Stop recovery can still reattach a recorder. + * - Remote: every allocation has a termination receipt for its sandbox, so + * neither the original nor a reattached runner can report usage. That proof + * does not clear the recorded remote process identity. + * Missing identity alone is never proof. One predicate serves the reconciler * scan and settlement so they cannot disagree. */ -const closedNativeReceiptSource = sql`${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed' - and ${heartbeatRuns.processPid} is null and ${heartbeatRuns.processGroupId} is null +const closedNativeReceiptSource = sql`${heartbeatRuns.runtimeMode} = 'native' and exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'terminal_failure') - and (select process_event.event_type from ${heartbeatRunEvents} process_event - where process_event.company_id = ${heartbeatRuns.companyId} and process_event.run_id = ${heartbeatRuns.id} - and process_event.source_event_id is null - and process_event.event_type in ('native.process_start_requested', 'native.process_identity_recorded', 'native.local_process_stopped') - order by process_event.seq desc limit 1) = 'native.local_process_stopped'`; + and ((${heartbeatRuns.status} = 'failed' + and ${heartbeatRuns.processPid} is null and ${heartbeatRuns.processGroupId} is null + and (select process_event.event_type from ${heartbeatRunEvents} process_event + where process_event.company_id = ${heartbeatRuns.companyId} and process_event.run_id = ${heartbeatRuns.id} + and process_event.source_event_id is null + and process_event.event_type in ('native.process_start_requested', 'native.process_identity_recorded', 'native.local_process_stopped') + order by process_event.seq desc limit 1) = 'native.local_process_stopped') + or (${heartbeatRuns.status} in ('failed', 'cancelled', 'timed_out', 'interrupted') and ${remoteExecutionTerminated}))`; async function nativeReceiptSourceClosed(db: Db, runId: string) { const [row] = await db.select({ closed: sql`${closedNativeReceiptSource}` }).from(heartbeatRuns)