diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 65be57ed92d..84e5bc00b2c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2475,6 +2475,20 @@ when Stop found no live controller. The worker retains the exact prepared source and obtains independent provider-stop evidence through normal recovery. It does not convert the no-op acknowledgement into a dispatched interrupt receipt or permission to continue the cancelled turn. +Terminal release of a warm (`keep_running`) run that is not successful stops and +retains its reusable sandbox. This release can finish before the preservation +worker sees a terminal coordinator. A released reusable lease with a confirmed +stop-and-retain receipt for the same run, lease and allocation therefore +remains a preservation source. The worker records it as file recovery work and +keeps its release status, policy and receipts. It dispatches no second stop and +grants no execute right on the reusable sandbox. A released reusable lease +stays ineligible without that valid receipt and the exact remote-workspace binding. +Native-sync descriptors have no such binding proof, so they also stay ineligible. +Ineligible sources cause no run writes or notifications. Recovery still rejects +the source while a task writer is live or a later lease shares the allocation. +Until this recovery saves the files, the +task's remote workspace keeps the ended writer's hold. Successor runs wait for that +recovery instead of overwriting unsaved files. A reusable lease can also owe native copyback: the export-repair action must select its exact immutable descriptor and the finalizer's accepted successful result under the same company, task, run and completion contract. Lease reuse diff --git a/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts b/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts index 91579705ac4..2acdb9912dc 100644 --- a/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts +++ b/server/src/services/native-runtime/native-cancelled-workspace-preservation.test.ts @@ -283,6 +283,134 @@ import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; } finally { await worker.stop(); } }); + /** A reassignment Stop of a warm (keep_running) run: the executor settles its + * coordinator, then terminal release stops and retains the reusable sandbox + * before the preservation worker's next check. */ + async function stopAndRetainReusableSource(f: Awaited>, options: { stopReceipt?: boolean } = {}) { + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + const confirmedAt = new Date(lease.acquiredAt.getTime() + 1_000); + const releasedAt = new Date(confirmedAt.getTime() + 1); + const termination = { schema: "paperclip.remote-termination.v1", companyId: f.companyId, runId: f.runId, + leaseId: f.leaseId, provider: "daytona", providerLeaseId: f.reference.providerLeaseId, state: "stopped", + confirmedAt: confirmedAt.toISOString() }; + const receipt = { schema: "paperclip.sandbox-stop-and-retain-receipt.v1", companyId: f.companyId, runId: f.runId, + leaseId: f.leaseId, provider: "daytona", providerLeaseId: f.reference.providerLeaseId, requestId: randomUUID(), + method: "environmentStopLease", pluginId: lease.metadata?.pluginId, confirmedAt: confirmedAt.toISOString() }; + await db.update(environmentLeases).set({ status: "released", leasePolicy: "reuse_by_environment", + cleanupStatus: "success", releasedAt, metadata: { ...lease.metadata, pendingCleanupInFlight: false, + ...(options.stopReceipt === false ? {} : { remoteExecutionTermination: termination, sandboxStopAndRetainReceipt: receipt }) }, + }).where(eq(environmentLeases.id, f.leaseId)); + await db.update(executionWorkspaces).set({ mode: "shared_workspace" }).where(eq(executionWorkspaces.id, f.workspaceId)); + await db.update(heartbeatRuns).set({ errorCode: "issue_reassigned", error: "Cancelled before issue reassignment", + runnerProfileJson: { remoteWorkspace: f.reference, nativeResourceDisposition: "keep_running" }, + }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ failureCode: "native_retry_cancelled", recoveryState: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + return (await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]; + } + + it.each(["worker", "startup-scan"] as const)("preserves files of a cancelled run whose reusable sandbox was already stopped and retained (%s)", async kind => { + const f = await seed(); + const stopped = await stopAndRetainReusableSource(f); + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + if (kind === "startup-scan") await preserveCancelledNativeWorkspaces(db); + else await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + const [retained] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(retained.resultJson).toMatchObject({ nativeCancellation: f.nativeCancellation, workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.leaseId] } }); + expect(retained).toMatchObject({ status: "cancelled", errorCode: "issue_reassigned", finishedAt: f.run.finishedAt }); + // The stopped allocation keeps its receipts. Preservation dispatches no + // second stop and grants no execute right on the reusable sandbox. + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]).toEqual(stopped); + const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId)); + expect(actions).toHaveLength(1); + expect(actions[0].evidence).toMatchObject({ runId: f.runId, workspaceRestoreRecovery: { leaseIds: [f.leaseId] } }); + expect((await inspectWorkspaceRecoverySource(db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, actionId: actions[0].id })).sources).toHaveLength(1); + if (kind === "worker") expect(worker.nextWakeAt()).toBeNull(); + // A repeated pass records nothing again and starts no work. + await preserveCancelledNativeWorkspaces(db); + const repeated = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + try { + await registerCancelledWorkspacePreservationWork(db, repeated, () => true, async () => true).ready; + expect(repeated.nextWakeAt()).toBeNull(); + } finally { await repeated.stop(); } + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toEqual(actions); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson).toEqual(retained.resultJson); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", leaseOwner: null, resultId: null }); + } finally { await worker.stop(); } + }, 120_000); + + it("fences recovery of a stopped reusable source behind later allocation owners and live writers", async () => { + const f = await seed(); + await stopAndRetainReusableSource(f); + await preserveCancelledNativeWorkspaces(db); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId)); + const source = { companyId: f.companyId, issueId: f.issueId, runId: f.runId, actionId: action.id }; + expect((await inspectWorkspaceRecoverySource(db, source)).sources).toHaveLength(1); + // A live writer on the task blocks recovery from the retained files. + const writerId = randomUUID(); + fixtures.add(writerId); + await db.insert(heartbeatRuns).values({ id: writerId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + status: "running", contextSnapshot: { issueId: f.issueId } }); + await expect(inspectWorkspaceRecoverySource(db, source)).rejects.toMatchObject({ code: "workspace_writer_active" }); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, writerId)); + // A later lease on the same reusable allocation is not a retired + // predecessor. Ownership is ambiguous, so recovery stops. + const [stopped] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: stopped.environmentId, + heartbeatRunId: writerId, issueId: f.issueId, executionWorkspaceId: f.workspaceId, provider: "daytona", + providerLeaseId: stopped.providerLeaseId, status: "active", leasePolicy: "reuse_by_environment", + metadata: { ...stopped.metadata, remoteExecutionTermination: undefined, sandboxStopAndRetainReceipt: undefined } }); + await expect(inspectWorkspaceRecoverySource(db, source)).rejects.toMatchObject({ code: "allocation_ownership_ambiguous" }); + }, 120_000); + + it.each(["live-owner", "pending-stop", "no-stop-receipt", "invalid-receipt", "expired-lease", "native-sync", "running"] as const) + ("does not preserve or rewrite a stopped reusable sandbox for %s", async invalid => { + const f = await seed(); + let stopped = await stopAndRetainReusableSource(f, { stopReceipt: invalid !== "no-stop-receipt" }); + if (invalid === "live-owner") await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: randomUUID(), + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (invalid === "pending-stop") await db.update(heartbeatRuns).set({ resultJson: { + nativeCancellation: { ...f.nativeCancellation, dispatchState: "pending" } } }).where(eq(heartbeatRuns.id, f.runId)); + if (invalid === "running") await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.runId)); + if (invalid === "invalid-receipt" || invalid === "expired-lease") { + [stopped] = await db.update(environmentLeases).set(invalid === "expired-lease" + ? { expiresAt: new Date(Date.now() - 1_000) } + : { metadata: { ...stopped.metadata, sandboxStopAndRetainReceipt: { + ...stopped.metadata?.sandboxStopAndRetainReceipt as Record, runId: randomUUID() } } }) + .where(eq(environmentLeases.id, f.leaseId)).returning(); + } + if (invalid === "native-sync") { + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeResourceDisposition: "keep_running", nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v3", state: "prepared", workspaceId: f.workspaceId, + leaseId: f.leaseId, providerLeaseId: f.reference.providerLeaseId, remoteCwd: f.reference.remoteCwd, + baselineSha256: "a".repeat(64), descriptorSha256: "b".repeat(64), finalHostSha256: null, + resourceDisposition: "keep_running" } } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(executionWorkspaces).set({ metadata: {} }).where(eq(executionWorkspaces.id, f.workspaceId)); + } + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const errors: unknown[] = []; + const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { errors.push(error); } }); + try { + await registerCancelledWorkspacePreservationWork(db, worker, () => true, async () => true).ready; + await preserveCancelledNativeWorkspaces(db); + await preserveCancelledNativeWorkspaces(db); + expect(errors).toEqual([]); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run.resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + // No pass rewrites the run: no restore-failure marker, no new status delivery. + expect(run).toEqual(before); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]).toEqual(stopped); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(0); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + } finally { await worker.stop(); } + }); + it("rolls preservation back with its outer transaction", async () => { const f = await seed(); await expect(db.transaction(async tx => { diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index 1c95720902a..38b06bacafd 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -21,6 +21,21 @@ import { preparedRemoteWorkspaceSource } from "../remote-workspace-lifecycle.js" const reusableLeasePolicies = ["reuse_by_environment", "reuse_by_execution_workspace"]; +/** The exact source the restore recorder can retain. A released reusable lease + * qualifies only with a confirmed stop-and-retain receipt and the remote + * workspace binding proof. Native-sync descriptors have no such proof here. */ +async function cancelledWorkspaceSource(db: Db, run: Pick) { + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + cancelledWorkspaceSourceLease(), + )); + for (const lease of leases) { + if (lease.status !== "released" || lease.leasePolicy === "ephemeral") return lease; + if (run.runnerProfileJson?.nativeWorkspaceSync == null && hasConfirmedSandboxStopAndRetain(lease) + && await preparedRemoteWorkspaceSource(db, lease)) return lease; + } + return null; +} function workspaceRecoveryReference(profile: Record | null | undefined) { const remote = remoteWorkspaceRunReferenceSchema.safeParse(profile?.remoteWorkspace); @@ -215,7 +230,7 @@ export async function preserveCancelledNativeWorkspaces(db: Db) { for (const candidate of candidates) { cancelledWorkspaceCursors.set(db, candidate.runId); const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, candidate.runId)); - if (run && hasAcknowledgedNativeRunStopRequest(run)) { + if (run && hasAcknowledgedNativeRunStopRequest(run) && await cancelledWorkspaceSource(db, run)) { try { await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); } catch (error) { if (error instanceof NativeWorkspaceFinalizationBusyError) continue; @@ -277,11 +292,7 @@ export function registerCancelledWorkspacePreservationWork(db: Db, if (finalization?.resultId) return null; if (finalization?.phase !== "terminal_failure" || finalization.leaseOwner) return { ...entry, dueAt: Date.now() + 30_000 }; - const [source] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( - eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), - cancelledWorkspaceSourceLease(), - )).limit(1); - if (source) await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); + if (await cancelledWorkspaceSource(db, run)) await preserveNativeFailedWorkspace(db, run, { allowPreterminal: true, requireCancelledStop: true }); const [current] = await db.select().from(heartbeatRuns) .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))); const currentControllerDeadline = current?.controllerLeaseExpiresAt?.getTime();