diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index c405b8be27f..14dec9fe6ca 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -548,8 +548,14 @@ an active or ephemeral one. It records the save obligation for the exact writer run while the workspace is still bound to that writer. A Stop during an accepted result's copy-back retries ends that copy-back, and no finalization exports those files afterwards. Preservation admits that writer as well and -leaves the accepted result uncommitted. Other accepted results keep their own -export repair path. Paperclip's workspace recovery then saves those files, which +leaves the accepted result uncommitted. A failed remote writer whose accepted +result exhausted its copy-back retries can also retain its files through this +path. It needs the exact stopped sandbox receipt, accepted completion contract, +and unchanged dirty workspace binding. This works even when a newer setup +attempt has failed on the old writer's hold. File recovery preserves both +attempts and the accepted result; it does not apply that result or repeat the +provider work. Other accepted results keep their own export repair path. +Paperclip's workspace recovery then saves those files, which releases the hold. It never replays the stopped run. While that save is in progress, its last progress counts as the ended writer's activity for the holder cutoff. A waiter keeps waiting instead of failing setup. The save's retry budget bounds that wait. A save that exhausts 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 9ec7ac330fe..19b87205a18 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -4,7 +4,7 @@ import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; -import { environmentLeases, executionWorkspaces, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, executionWorkspaces, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; import { logActivity } from "../activity-log.js"; @@ -102,7 +102,10 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick Date.now() + || run.runnerProfileJson?.nativeWorkspaceSync != null || !remote.success || remote.data.state !== "prepared" + || !hasConfirmedSandboxStopAndRetain(source) + || !await preparedRemoteWorkspaceSource(tx as unknown as Db, source)) return; + const [accepted] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.issueId, issueId), + eq(nativeRunResults.runId, run.id), eq(nativeRunResults.id, coordinator.resultId!), + eq(nativeRunResults.completionContractId, run.completionContractId), eq(nativeRunResults.schemaStatus, "accepted"), + )).limit(1); + if (!accepted) return; + } // A stopped reusable sandbox is retained only with its exact stop receipt // while the workspace is still bound to this writer. Otherwise record nothing. - if (source.status !== "active" && reusableLeasePolicies.includes(source.leasePolicy) && (!hasConfirmedSandboxStopAndRetain(source) + if (!exhaustedAcceptedCopyback && source.status !== "active" && reusableLeasePolicies.includes(source.leasePolicy) && (!hasConfirmedSandboxStopAndRetain(source) || !await preparedRemoteWorkspaceSource(tx as unknown as Db, source))) return; // Ownership settlement still applies to workspace-less external sessions. // Their Runner descriptor is not evidence that user files need exporting. diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 20f7b40be8e..1a8b87ca3a9 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -33,7 +33,7 @@ import { workspaceOperationService } from "./workspace-operations.js"; import { instanceSettingsService } from "./instance-settings.js"; import { projectTaskInterruption, taskInterruptionPresentation } from "./task-interruption-presentation.js"; import * as nativeWorkspaceSync from "./native-runtime/native-workspace-sync.js"; -import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js"; +import { registerCancelledWorkspacePreservationWork, preserveTaskRemoteWorkspaceWriter, preserveOrphanedNativeWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspace-sync"; import { readSystemRecovery } from "./system-recovery-state.js"; import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; @@ -974,6 +974,126 @@ describe("automatic retained workspace recovery", () => { return { ...f, ...(sourceLayout === "published" ? { remote: remoteCwd } : {}), source, workspaceId, resultId, git, commit }; } + async function exhaustedAcceptedRemoteWriter(leasePolicy: "ephemeral" | "reuse_by_environment" = "ephemeral") { + const f = await acceptedRemoteExport(); + await db.update(issueRecoveryActions).set({ status: "cancelled", cause: "native_workspace_sync_out_failed", + evidence: { runId: f.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } }, + }).where(eq(issueRecoveryActions.id, f.actionId)); + await db.update(heartbeatRuns).set({ errorCode: "native_workspace_sync_out_retry_exhausted" }) + .where(eq(heartbeatRuns.id, f.runId)); + const [lease] = await db.update(environmentLeases).set({ leasePolicy }) + .where(eq(environmentLeases.id, f.lease.id)).returning(); + const stopping = (await prepareStoppedSandboxForRecovery(db, lease))!; + const stopped = (await settleStopOnlyCleanup(db, stopping, { + attemptId: String(stopping.metadata!.pendingCleanupAttemptId), + receipt: { providerLeaseId: f.providerLeaseId, state: "stopped" }, + }))!; + const source = await run(f.runId); + const [preparation] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, + issueId: f.issueId, runtimeMode: "legacy", status: "failed", executionStage: "preparing", + startedAt: new Date(), finishedAt: new Date(), errorCode: "setup_failed", + error: "The workspace still has an unfinished writer or save operation.", + contextSnapshot: { issueId: f.issueId, executionWorkspaceId: f.workspaceId }, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).returning(); + return { ...f, stopped, preparation }; + } + + it.each(["bound writer", "orphaned lease", "reusable writer"] as const)("retains an exhausted accepted remote copy-back through its %s", async entry => { + const f = await exhaustedAcceptedRemoteWriter(entry === "reusable writer" ? "reuse_by_environment" : "ephemeral"); + const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const historical = await action(f.actionId); + if (entry === "orphaned lease") expect(await preserveOrphanedNativeWorkspace(db, f.stopped)).toBe(true); + else await preserveTaskRemoteWorkspaceWriter(db, f.companyId, f.issueId); + expect((await run(f.runId)).resultJson).toMatchObject({ workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { leaseIds: [f.lease.id] } }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", resultId: result.id, leaseOwner: null }); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(result); + expect(await action(f.actionId)).toEqual(historical); + expect(await run(f.preparation.id)).toEqual(f.preparation); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(task); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n"); + }, 30_000); + + it.each(["active-finalizer", "retryable", "other-failure", "live-controller", "unfinished-run", "missing-stop", + "rejected-result", "new-writer", "saved-workspace", "wrong-epoch", "native-sync", "source-binding"] as const)( + "keeps accepted remote copy-back held when its %s proof changes", async gate => { + const f = await exhaustedAcceptedRemoteWriter(); + const source = await run(f.runId); + if (gate === "active-finalizer") await db.update(nativeRunFinalizations).set({ leaseOwner: "live-finalizer" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + if (gate === "retryable") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + if (gate === "other-failure") await db.update(nativeRunFinalizations).set({ failureCode: "native_workspace_sync_out_unsafe_archive" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + if (gate === "live-controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, f.runId)); + if (gate === "unfinished-run") await db.update(heartbeatRuns).set({ finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "missing-stop") await db.update(environmentLeases).set({ metadata: { ...f.stopped.metadata, + remoteExecutionTermination: null } }).where(eq(environmentLeases.id, f.lease.id)); + if (gate === "rejected-result") await db.update(nativeRunResults).set({ schemaStatus: "rejected" }) + .where(eq(nativeRunResults.id, f.resultId)); + if (gate === "new-writer" || gate === "saved-workspace") { + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + const authority = readRemoteWorkspaceAuthority(workspace.metadata)!; + await db.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspace: { ...authority, + writerRunId: gate === "new-writer" ? randomUUID() : null, dirty: gate === "new-writer" } } }) + .where(eq(executionWorkspaces.id, f.workspaceId)); + } + if (gate === "wrong-epoch") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, + remoteWorkspace: { ...(source.runnerProfileJson!.remoteWorkspace as object), epoch: 99 } } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "native-sync") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...source.runnerProfileJson, + nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3" } } }).where(eq(heartbeatRuns.id, f.runId)); + if (gate === "source-binding") await db.update(environmentLeases).set({ metadata: { ...f.stopped.metadata, remoteCwd: "/other" } }) + .where(eq(environmentLeases.id, f.lease.id)); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); + await preserveTaskRemoteWorkspaceWriter(db, f.companyId, f.issueId); + expect((await run(f.runId)).resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(lease); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(result); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId))).toHaveLength(1); + expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n"); + }, 30_000); + + it("repairs an older accepted remote writer before one HTTP Retry continues a newer failed preparation", async () => { + const f = await exhaustedAcceptedRemoteWriter(); + const [result] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.preparation.agentId, status: "running" }); + const receipt = (await request(receiptOnlyRecoveryApp()).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.preparation.id, idempotencyKey: randomUUID() }).expect(202)).body; + expect((await run(f.runId)).resultJson).toHaveProperty("workspaceRestoreRecovery"); + const records = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, f.companyId)); + const preservation = records.find(record => record.id !== f.actionId && record.evidence.runId === f.runId + && record.evidence.workspaceRestoreRecovery)!; + expect(preservation).toBeDefined(); + const input = { ...f.input, actionId: preservation.id }; + await adoptSystemRecovery(db, input); + await recoverRetainedWorkspace(input); + expect(readWorkspaceRecovery((await action(preservation.id)).evidence)?.status).toBe("recovered"); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId), + heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "queued", contextSnapshot: { previousRunId: f.preparation.id } }); + const recovered = await run(f.runId), reference = recovered.runnerProfileJson!.remoteWorkspace as { snapshotId: string }; + const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId, + snapshotId: reference.snapshotId }); + try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); } + finally { await saved.cleanup(); } + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(result); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", resultId: result.id }); + expect(await run(f.preparation.id)).toEqual(f.preparation); + }, 30_000); + async function pressuredToolCache(pressure: "transfer" | "sandbox-write" = "transfer") { const f = await acceptedRemoteExport(); const runner = path.join(f.remote, ".paperclip-runtime", "paperclip-runner");