diff --git a/server/src/__tests__/environment-pending-cleanup-queue-owner.test.ts b/server/src/__tests__/environment-pending-cleanup-queue-owner.test.ts new file mode 100644 index 00000000000..d1557b480a0 --- /dev/null +++ b/server/src/__tests__/environment-pending-cleanup-queue-owner.test.ts @@ -0,0 +1,99 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { companies, createDb, environmentLeases } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; +const KIND = "environment_pending_cleanup"; + +suite("pending-cleanup obligation owned by the cleanup queue", () => { + let db!: ReturnType; + let temporary: Awaited> | null = null; + let companyId = ""; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-pending-cleanup-queue-owner-"); + db = createDb(temporary.connectionString); + companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "PCQ", + requireBoardApprovalForNewAgents: false }); + }, 30_000); + + afterEach(async () => { + await db.delete(environmentLeases); + }); + + afterAll(async () => { + await temporary?.cleanup(); + }); + + async function item(updatedAt: Date, metadata: Record = {}) { + const id = randomUUID(); + await db.insert(environmentLeases).values({ id, companyId, status: "pending_cleanup", cleanupStatus: "failed", + provider: "fake", providerLeaseId: `sandbox://fake/${id}`, updatedAt, metadata }); + return id; + } + + async function pass() { + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + return { kind: report.kinds.find(entry => entry.kind === KIND)!, + violations: report.violations.filter(violation => violation.kind === KIND) }; + } + + it("names the cleanup queue as owner while the item's retry deadline is current, so I1 reports no gap", async () => { + await item(new Date()); + await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 5, + pendingCleanupRetryAfterMs: Date.now() + 20 * 60_000 }); + const { kind, violations } = await pass(); + expect(kind).toMatchObject({ inProgress: 2, verdicts: { alive: 2 }, reasons: { queue_scheduled: 2 } }); + expect(violations).toEqual([]); + }); + + it("reports an item the queue has not acted on long past its deadline as a visible-next-action gap", async () => { + const overdue = await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 1 }); + const { kind, violations } = await pass(); + expect(kind).toMatchObject({ verdicts: { unprovable: 1 }, reasons: { queue_overdue: 1 } }); + expect(violations).toEqual([expect.objectContaining({ rule: "I7_visible_next_action", recordId: overdue, + reason: "queue_overdue" })]); + }); + + it("never reports a settled orphan's first attempt as overdue just because the worker's deadline is 'now'", async () => { + // The orphaned-run-lease settle keeps updatedAt and saves when it queued the item. + await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 0, + pendingCleanupRecoveredAtMs: Date.now() }); + await db.update(environmentLeases).set({ failureReason: "orphaned_active_lease_recovered" }); + const { kind, violations } = await pass(); + expect(kind).toMatchObject({ verdicts: { alive: 1 }, reasons: { queue_scheduled: 1 } }); + expect(violations).toEqual([]); + }); + + it("wakes the reconciler when a scheduled item would become overdue", async () => { + const updatedAt = new Date(Date.now() - 60_000); + await item(updatedAt); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + // Due at updatedAt + one retry interval; overdue after two more. + const expected = updatedAt.getTime() + 5 * 60_000 + 10 * 60_000; + expect(Math.abs((report.nextDueAt?.getTime() ?? 0) - expected)).toBeLessThan(5_000); + }); + + it("keeps an escalated item with the sweep's 30-minute cooldown scheduled, due at its saved deadline", async () => { + // The sweep escalates well after boot: let the clock pass the reader's 1 s slack. + await new Promise(resolve => setTimeout(resolve, 1_500)); + const retryAfterMs = Date.now() + 30 * 60_000; + await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupRetryAttempts: 5, + pendingCleanupRetryAfterMs: retryAfterMs }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.kinds.find(entry => entry.kind === KIND)).toMatchObject({ verdicts: { alive: 1 }, + reasons: { queue_scheduled: 1 } }); + expect(report.violations.filter(violation => violation.kind === KIND)).toEqual([]); + expect(Math.abs((report.nextDueAt?.getTime() ?? 0) - (retryAfterMs + 10 * 60_000))).toBeLessThan(5_000); + }); + + it("does not list an item a claimant holds in flight (that is the teardown claim)", async () => { + await item(new Date(Date.now() - 60 * 60_000), { pendingCleanupInFlight: true, pendingCleanupAttemptId: "claim" }); + const { kind } = await pass(); + expect(kind).toMatchObject({ inProgress: 0 }); + }); +}); diff --git a/server/src/__tests__/environment-run-lease-ownership.test.ts b/server/src/__tests__/environment-run-lease-ownership.test.ts index 32d517d41aa..a8a666a0ebc 100644 --- a/server/src/__tests__/environment-run-lease-ownership.test.ts +++ b/server/src/__tests__/environment-run-lease-ownership.test.ts @@ -117,6 +117,7 @@ suite("environment_run_lease_active on the shared ownership path", () => { expect(report.kinds.find(kind => kind.kind === "environment_run_lease_active")).toMatchObject({ settled: 1, reasons: { pid_dead: 1 } }); expect(await row(leaseId)).toMatchObject({ status: "pending_cleanup", failureReason: "orphaned_active_lease_recovered" }); + expect(typeof (await row(leaseId)).metadata?.pendingCleanupRecoveredAtMs).toBe("number"); }); it("stops and keeps a dead holder's sandbox instead of destroying it", async () => { diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index b1986fa9880..e303af8eaef 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -158,15 +158,16 @@ describe("report-only ownership reconciliation", () => { verdicts: { alive: 1, gone: 0, unprovable: 0 }, reasons: { current_process: 1 } }); expect(kindReport(first, "environment_run_lease_active")).toMatchObject({ inProgress: 1, verdicts: { gone: 1 }, reasons: { owner_run_terminal: 1 } }); + // The cleanup queue owns an unclaimed item while its retry deadline is current. expect(kindReport(first, "environment_pending_cleanup")).toMatchObject({ inProgress: 1, - verdicts: { unprovable: 1 }, reasons: { no_owner_stamp: 1 } }); + verdicts: { alive: 1 }, reasons: { queue_scheduled: 1 } }); // A different PID space is never proof. expect(currentProcessSpaceHash()).not.toBe("another-pid-space"); expect(kindReport(first, "sandbox_creation_intent")).toMatchObject({ inProgress: 1, verdicts: { unprovable: 1 }, reasons: { different_process_space: 1 } }); expect(kindReport(first, "run_accounting_debt")).toMatchObject({ inProgress: 2 }); - expect(rules(first, "environment_pending_cleanup")).toEqual(["I1_owned"]); + expect(rules(first, "environment_pending_cleanup")).toEqual([]); // I2 needs the owner proven gone on two consecutive passes. expect(rules(first, "environment_run_lease_active")).toEqual([]); // Phase 3: pending accounting is exposure, not a gate. Only a refusal that diff --git a/server/src/services/environment-run-lease-settlement.ts b/server/src/services/environment-run-lease-settlement.ts index 0c1596bebc7..fb1c79a471e 100644 --- a/server/src/services/environment-run-lease-settlement.ts +++ b/server/src/services/environment-run-lease-settlement.ts @@ -75,7 +75,10 @@ export async function settleEnvironmentRunLease(db: Db, record: InProgressRecord // Keep the row's updatedAt, so the pending_cleanup sweep accepts it in the // same tick instead of waiting out its own retry backoff. const flipped = await writeEnvironmentCleanup(tx, write => write.update(environmentLeases) - .set({ status: "pending_cleanup", failureReason: ORPHANED_RUN_LEASE_FAILURE_REASON }) + .set({ status: "pending_cleanup", failureReason: ORPHANED_RUN_LEASE_FAILURE_REASON, + // When the first cleanup attempt became due (the row keeps updatedAt). + metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ + pendingCleanupRecoveredAtMs: context.now.getTime() })}::jsonb` }) .where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.status, "active"), eq(environmentLeases.heartbeatRunId, runId), sql`${environmentLeases.metadata} is not distinct from ${lease.metadata === null ? null : JSON.stringify(lease.metadata)}::jsonb`)) diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 7f68031765f..ff314c14c41 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -165,6 +165,10 @@ async function classify(owner: RecordOwner, runProofs: Map owner.graceMs + ? { verdict: "unprovable", reason: "queue_overdue" } : { verdict: "alive", reason: "queue_scheduled" }; case "none": return { verdict: "unprovable", reason: "no_owner_stamp" }; } } @@ -462,6 +466,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: { const kindReports: KindReport[] = []; const violations: InvariantViolation[] = []; const goneThisPass = new Set(); + let queueDueAt: Date | null = null; const settleRecords = options.settle ?? options.record !== false; let retryNeeded = false; @@ -501,6 +506,16 @@ export async function runOwnershipReconciliationPass(db: Db, options: { if (kind.invariantChecks.includes(1) && record.owner.kind === "none") { violations.push({ ...base, rule: "I1_owned", reason: "no_owner_stamp" }); } + if (kind.invariantChecks.includes(7) && record.owner.kind === "queue") { + if (proof.reason === "queue_overdue") { + violations.push({ ...base, rule: "I7_visible_next_action", reason: "queue_overdue" }); + } else { + // Re-check when this item would become overdue, so a queue that stops + // acting without another write is still reported. + const overdueAt = new Date(record.owner.dueAt.getTime() + record.owner.graceMs + 1); + if (!queueDueAt || overdueAt < queueDueAt) queueDueAt = overdueAt; + } + } if (settleRecords && kind.settle) { const outcome = await settleOrSurface(db, kind, record, proof, now); if (outcome === "settled") { report.settled += 1; continue; } @@ -517,7 +532,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: { } } - let nextDueAt: Date | null = null; + let nextDueAt: Date | null = queueDueAt; for (const check of [ () => checkAdmissionIndependence(db, state, now, limit), () => checkExactlyOnceIntents(db, state, now, limit), diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 787dd8160a2..aee032246e4 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -42,6 +42,11 @@ import { } from "../native-runtime/native-token-lease-settlement.js"; import { ephemeralServerEnabled } from "../ephemeral-server-mode.js"; import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../environment-lease-controller.js"; +import { + ENVIRONMENT_CLEANUP_RETRY_MS, + pendingCleanupDeadlineSql, + recoveredOrphanFirstAttemptSql, +} from "../environment-cleanup-state.js"; import { settleEnvironmentRunLease } from "../environment-run-lease-settlement.js"; import { settleEnvironmentTeardownClaim } from "../environment-teardown-claim-settlement.js"; @@ -71,6 +76,10 @@ export type RecordOwner = | { kind: "run"; runId: string } /** A person or agent assignment, such as an operator-owned recovery action. */ | { kind: "operator"; ownerType: string | null } + /** A durable work queue that acts on the record by its saved deadline. The + * queue owns the record while that deadline is current; a deadline long past + * is a gap (the queue is not acting), never a release. */ + | { kind: "queue"; queue: string; dueAt: Date; graceMs: number } /** Nothing on the record names an owner (invariant 1). */ | { kind: "none" }; @@ -362,6 +371,50 @@ function teardownClaimOwner(row: LeaseRow): RecordOwner { }; } +/** A cleanup worker pass that misses a due item by more than this is a gap. */ +const PENDING_CLEANUP_OVERDUE_GRACE_MS = 2 * ENVIRONMENT_CLEANUP_RETRY_MS; + +/** + * Pending-cleanup obligation (not in flight). An unclaimed queue item names no + * process by design: the environment cleanup queue owns it, and its next + * action is the queue's next attempt, due at the saved retry deadline (the + * same deadline the cleanup worker sleeps until). While that deadline is + * current the queue owns the item; once it is long past, the item is reported + * as a gap (I7), never released. Invariant 1 then measures only real gaps. + */ +function pendingCleanupKind(): OwnableKind { + const where = and(eq(environmentLeases.status, "pending_cleanup"), + sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`); + // The first attempt of a settled orphan is due at once: the worker's "run + // now" deadline is 0, so its due time is when the settle queued it (saved + // by the settle; a row from before that is due since its last write). + // Built per listing: the deadline reader bounds saved deadlines relative to + // the current time, so it must not be frozen at module load. + const dueMs = () => sql`case when ${recoveredOrphanFirstAttemptSql()} + then coalesce(case when jsonb_typeof(${environmentLeases.metadata}->'pendingCleanupRecoveredAtMs') = 'number' + then (${environmentLeases.metadata}->>'pendingCleanupRecoveredAtMs')::numeric end, + extract(epoch from ${environmentLeases.updatedAt}) * 1000) + else greatest(extract(epoch from ${environmentLeases.updatedAt}) * 1000 + ${ENVIRONMENT_CLEANUP_RETRY_MS}, + ${pendingCleanupDeadlineSql()}) end`; + return { + id: "environment_pending_cleanup", title: "Pending-cleanup obligation (not in flight)", obligation: "ownership", + storage: "database", settle: null, invariantChecks: [1, 2, 7], + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: environmentLeases.id, companyId: environmentLeases.companyId, + heartbeatRunId: environmentLeases.heartbeatRunId, issueId: environmentLeases.issueId, + updatedAt: environmentLeases.updatedAt, dueMs: dueMs() }) + .from(environmentLeases).where(where).orderBy(asc(environmentLeases.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: row.heartbeatRunId, issueId: row.issueId, + owner: { kind: "queue", queue: "environment_cleanup", dueAt: new Date(Number(row.dueMs)), + graceMs: PENDING_CLEANUP_OVERDUE_GRACE_MS }, + updatedAt: row.updatedAt, + })) }; + }, + }; +} + /** `ControllerProcess` metadata ({bootId, pid, hostHash, processStartedAt}). */ function controllerStamp(value: unknown, scope: string): RecordOwner { const controller = readControllerProcess(value); @@ -604,10 +657,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ teardownClaimOwner), settle: settleEnvironmentTeardownClaim, }, - leaseKind("environment_pending_cleanup", "Pending-cleanup obligation (not in flight)", "ownership", - and(eq(environmentLeases.status, "pending_cleanup"), - sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`), - () => ({ kind: "none" }), [1, 2, 7]), + pendingCleanupKind(), leaseKind("environment_stop_and_retain_intent", "Stop-and-retain intent", "workspace_preservation", and(eq(environmentLeases.status, "pending_cleanup"), sql`${environmentLeases.metadata} ? 'sandboxStopAndRetain'`), row => runOwner(row.heartbeatRunId)),