diff --git a/doc/plans/2026-10-30-live-recovery-sleep.md b/doc/plans/2026-10-30-live-recovery-sleep.md index f0dbff1e446..fac3d192b6c 100644 --- a/doc/plans/2026-10-30-live-recovery-sleep.md +++ b/doc/plans/2026-10-30-live-recovery-sleep.md @@ -55,3 +55,7 @@ The actual private-pool control prewarms the connection and requires exactly fiv Agent settlement retains a bounded retry only after its actual `FOR UPDATE SKIP LOCKED` attempt misses an existing candidate. Releasing a transaction that only held the lock emits no eligibility hint; the retained entry still settles exactly once and becomes quiet. A native question marker can commit before execution changes the run to failed/retryable failure. Both reconstruction and consumption select this exact unfinished class. A Stop may also advance the coordinator to `terminal_failure/native_retry_cancelled` before the final run-status commit fails. The owning pending predicate includes only the existing exactly bound, audited pending/acknowledged cancellation receipt; cancellation candidate discovery uses `nativeRetryCancellationEligible` again. Existing locked native dispatch and the unchanged acknowledged-receipt final CAS still authorize all effects. A one-shot PostgreSQL failure proves this continuation completes with the same intent/audit receipt and no provider execution, while marked failed history with misbound receipts remains unchanged and quiet. + +Shutdown workspace cleanup publishes resource-work intent only after its competing-allocation and in-flight-cleanup refusal checks, immediately before the existing write. An unchanged allocation conflict or cleanup owner without a finite expiry parks until an owning resource write; inspecting that gate cannot wake the worker again. A finite pendingCleanupLeaseExpiresAtMs remains a concrete obligation: the shutdown owner returns that exact stored deadline through its existing callback and re-reads a moved expiry under the original issue/run/lease lock order. + +Actual registered-owner controls use a confirmed stop-only released lease and a pending ordinary-cleanup lease. Both refusal states retain the pending shutdown marker while twenty virtual minutes produce zero driver SQL. A rolled-back owning gate release preserves that state; a committed release reactivates the original preservation path, retains the allocation and recovery metadata, and expires the marker without certifying file export. Separate controls cross the original and moved cleanup-claim deadlines and the existing active shutdown-marker deadline. Transaction uncertainty is joined through the queue-specific work-fence reconciliation before the idle measurement; no timeout or recurring recovery scan is added. diff --git a/server/src/__tests__/live-recovery-work.test.ts b/server/src/__tests__/live-recovery-work.test.ts index 2e765167475..fefe4bb0b2a 100644 --- a/server/src/__tests__/live-recovery-work.test.ts +++ b/server/src/__tests__/live-recovery-work.test.ts @@ -7,7 +7,7 @@ import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { agents, authUsers, companies, companyMemberships, createDb, closeLivenessDb, closeCheckpointDb, heartbeatRuns, heartbeatRunEvents, nativeRunFinalizations, environmentLeases, issueThreadInteractions, applyPendingMigrations, - issueRecoveryActions, issues, activityLog, agentWakeupRequests } from "@paperclipai/db"; + issueRecoveryActions, issues, activityLog, agentWakeupRequests, reconcileDatabaseWork } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -16,6 +16,7 @@ import { notifyLiveRunRecovery, notifyLiveResourceRecovery, notifyStrandedAgentS import { notifyExecutionRunWork } from "../services/execution-control-notifications.js"; import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { prepareSandboxStopAndRetain, settleStopOnlyCleanup, hasConfirmedSandboxStopAndRetain } from "../services/sandbox-stop-and-retain.js"; import { beginLegacyShutdownWorkspaceSettlement } from "../services/legacy-shutdown-workspace-settlement.js"; import { renewLegacyControllerLease, legacyControllerBootId } from "../services/legacy-controller-lease.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -388,6 +389,79 @@ describe("residual recovery production matrix", () => { await vi.waitFor(async () => expect((await getLease(orphan.id)).status).toBe("pending_cleanup"), { timeout: 40_000 }); await quiet(); }); + async function shutdownSource(released: boolean, cleanupExpiry?: number) { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id, + status: "failed", runtimeMode: "legacy", contextSnapshot: { issueId: f.issue.id }, finishedAt: old }).returning(); + let [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id, + issueId: f.issue.id, status: "active", provider: "fixture", providerLeaseId: randomUUID(), acquiredAt: old, updatedAt: old, + metadata: { driver: "sandbox", sandboxProviderPlugin: true, pluginId: "shutdown-fixture" } }).returning(); + await beginLegacyShutdownWorkspaceSettlement(db, run, new Date(Date.now() - 1000)); + if (released) { + const prepared = await prepareSandboxStopAndRetain(db, await getLease(lease.id)); + expect(prepared).toBeTruthy(); + const settled = await settleStopOnlyCleanup(db, prepared!, { attemptId: String(prepared!.metadata?.pendingCleanupAttemptId), + receipt: { providerLeaseId: lease.providerLeaseId, state: "stopped" } }); + expect(settled).toBeTruthy(); lease = settled!; + expect(hasConfirmedSandboxStopAndRetain(lease)).toBe(true); + } else { + [lease] = await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", + metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ pendingCleanupInFlight: true, + ...(cleanupExpiry === undefined ? {} : { pendingCleanupLeaseExpiresAtMs: cleanupExpiry }) })}::jsonb`, + }).where(eq(environmentLeases.id, lease.id)).returning(); + } + return { ...f, run, lease }; + } + it.each(["competing allocation", "cleanup in flight"] as const)("parks unchanged shutdown refusal for %s until its owning commit", async gate => { + const f = await shutdownSource(gate === "competing allocation"); + const [competing] = gate === "competing allocation" ? await db.insert(environmentLeases).values({ companyId: f.companyId, + status: "pending_cleanup", provider: f.lease.provider, providerLeaseId: f.lease.providerLeaseId, + acquiredAt: new Date(f.lease.acquiredAt.getTime() - 1000) }).returning() : []; + await start(); + expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" }); + await quiet(); + const releaseGate = async (database: typeof db) => writeEnvironmentCleanup(database, tx => tx.update(environmentLeases).set( + competing ? { status: "released", releasedAt: new Date() } + : { metadata: sql`${environmentLeases.metadata} || '{"pendingCleanupInFlight":false}'::jsonb` }, + ).where(eq(environmentLeases.id, competing?.id ?? f.lease.id)), { companyId: f.companyId }); + await expect(db.transaction(async tx => { + await releaseGate(tx as unknown as typeof db); + throw new Error("roll back shutdown gate release"); + })).rejects.toThrow("roll back shutdown gate release"); + await reconcileDatabaseWork(db, "legacy-shutdown-settlement"); await Promise.allSettled([...passes]); await immediate(); + expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" }); + await quiet(); + await releaseGate(db); + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); + const current = await getLease(f.lease.id); + expect(current.metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "expired" }); + expect(current.leasePolicy).toBe("retain_on_failure"); + expect(current.metadata?.workspaceRestoreRecovery).toMatchObject({ runId: f.run.id }); + expect(current.providerLeaseId).toBe(f.lease.providerLeaseId); + }, { timeout: 15_000 }); + await reconcileDatabaseWork(db, "legacy-shutdown-settlement"); await Promise.allSettled([...passes]); await immediate(); + await quiet(); + }); + it("retains the actual cleanup claim deadline after shutdown expiry and honors a moved expiry", async () => { + const initial = Date.now() + 1500, f = await shutdownSource(false, initial); + await start(); + expect(deadlines.get("legacy-shutdown-settlement")!()).toBe(initial); + const moved = Date.now() + 4000; + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ + metadata: sql`jsonb_set(${environmentLeases.metadata}, '{pendingCleanupLeaseExpiresAtMs}', ${JSON.stringify(moved)}::jsonb)`, + }).where(eq(environmentLeases.id, f.lease.id)), { companyId: f.companyId, leaseId: f.lease.id }); + await advance(initial - Date.now()); + expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" }); + expect(deadlines.get("legacy-shutdown-settlement")!()).toBe(moved); + await advance(moved - Date.now() - 1); + expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" }); + await advance(1); + await vi.waitFor(async () => expect((await getLease(f.lease.id)).metadata?.legacyShutdownWorkspaceSettlement) + .toMatchObject({ state: "expired" })); + expect((await getLease(f.lease.id)).metadata?.workspaceRestoreRecovery).toMatchObject({ runId: f.run.id }); + await quiet(); + }); it("expires the exact shutdown settlement marker into preserved recovery instead of inferring file export", async () => { const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id, diff --git a/server/src/services/legacy-shutdown-workspace-settlement.ts b/server/src/services/legacy-shutdown-workspace-settlement.ts index 4a3576f997d..c4a4451c364 100644 --- a/server/src/services/legacy-shutdown-workspace-settlement.ts +++ b/server/src/services/legacy-shutdown-workspace-settlement.ts @@ -75,7 +75,7 @@ export async function legacyShutdownWorkspaceResourceProtected(db: Db, } /** Released stop receipts still need reconciliation if export never settled. */ -export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date(), scope: { companyId?: string; leaseId?: string } = {}) { +export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date(), scope: { companyId?: string; leaseId?: string; onDeferredDeadline?: (at: number) => void } = {}) { const rows = await db.select().from(environmentLeases).where(and( scope.companyId ? eq(environmentLeases.companyId, scope.companyId) : undefined, scope.leaseId ? eq(environmentLeases.id, scope.leaseId) : undefined, @@ -83,7 +83,7 @@ export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`, sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt' <= ${now.toISOString()}`, )).orderBy(sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt'`).limit(50); - for (const lease of rows) await allowLegacyShutdownWorkspaceCleanup(db, lease, now); + for (const lease of rows) await allowLegacyShutdownWorkspaceCleanup(db, lease, now, { onDeferredDeadline: scope.onDeferredDeadline }); } /** Commit the fence while the run is still live, before exposing interruption. */ @@ -115,7 +115,7 @@ export async function finishLegacyShutdownWorkspaceSettlement(db: Db, run: Run, /** A restart or another cleanup worker must never infer saved files from timeout. */ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, now = new Date(), - options: { ownerStopOnly?: boolean } = {}) { + options: { ownerStopOnly?: boolean; onDeferredDeadline?: (at: number) => void } = {}) { if (!pendingLegacyShutdownWorkspaceSettlement(lease)) return true; return db.transaction(async tx => { const [hint] = await tx.select().from(environmentLeases).where(and( @@ -148,7 +148,6 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, if (!Number.isFinite(deadline) || deadline > now.getTime()) return false; // A killed container cannot certify export. Preserve source and hold in // this same transaction before any worker may perform provider cleanup. - await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); if (current.status !== "active") { const { hasConfirmedSandboxStopAndRetain, readStopOnlyCleanup } = await import("./sandbox-stop-and-retain.js"); if (!(current.status === "released" && hasConfirmedSandboxStopAndRetain(current)) @@ -165,12 +164,17 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, if (current.status === "pending_cleanup" && !readStopOnlyCleanup(current)) { const expiresAt = current.metadata?.pendingCleanupLeaseExpiresAtMs; if (current.metadata?.pendingCleanupInFlight === true - && (typeof expiresAt !== "number" || !Number.isFinite(expiresAt) || expiresAt > now.getTime())) return false; + && (typeof expiresAt !== "number" || !Number.isFinite(expiresAt) || expiresAt > now.getTime())) { + if (typeof expiresAt === "number" && Number.isFinite(expiresAt)) options.onDeferredDeadline?.(expiresAt); + return false; + } // A failed ordinary release can race the shutdown marker. Re-enter the // recorder's active-source path within this claim transaction so it // replaces destructive retry with an exact stop-only recovery intent. + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); await tx.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, current.id)); } else { + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); await tx.update(environmentLeases).set({ leasePolicy: "retain_on_failure", metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ workspaceRestoreRecovery: { schema: LEGACY_WORKSPACE_RECOVERY_SCHEMA, runId: run.id, @@ -178,6 +182,7 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, }).where(eq(environmentLeases.id, current.id)); } } + if (current.status === "active") await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); const { terminalizeLegacyExecution } = await import("./legacy-execution-recovery.js"); const preserved = await terminalizeLegacyExecution({ db: tx as unknown as Db, run, status: run.status, recordRestoreFailureOnly: true, patch: { resultJson: { workspaceRestoreFailure: "restore_failed" } } });