From 7d1e9b3477f12780e987c32fb8191d476c210fda Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:47:42 -0600 Subject: [PATCH] fix(recovery): retain exact warm native export sources (#17916) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native task recovery must save unfinished workspace exports before continuation. > - A prepared export can still refer to a sandbox with a warm reuse policy. > - The strict recovery inspector rejects that policy and exhausts source verification. > - Ordinary reuse must not give the same unfinished allocation to another run. > - An existing user Retry can authorize bounded file maintenance after exact ownership checks. > - This change retains that exact source and preserves the original failure record. > - Recovery completes only when files, stop evidence and policy settlement agree. ## Linked Issues or Issue Description **What happened?** A failed native run has a prepared `keep_running` workspace export. Its exact sandbox lease still has `reuse_by_environment` policy. Native export cleanup can preserve that policy while it stops the sandbox. Generic workspace recovery then rejects the source with `lease_policy_not_retained`. A saved user Retry remains blocked before file export. **Expected behavior** Keep unfinished exports out of ordinary warm reuse. With current user authority and an unchanged saved Retry, retain the exact source for one bounded maintenance generation. Preserve files and return continuation to normal admission. Keep all ambiguous sources held. **Steps to reproduce** 1. Prepare a native workspace export with a warm sandbox and `keep_running` disposition. 2. Save a normal board Retry for the failed source run. 3. Persist its first exhausted source-verification failure before file export. 4. Run the normal maintenance worker. Before this change, source policy rejection leaves the saved files unrestored. Refs #17773 for source diagnostics. Related #17839 records a missing preservation obligation during blocked restart. This change handles an existing prepared warm export whose verification already failed. ## What Changed - Exclude unfinished physical allocations from reuse and obsolete-lease cleanup across every historical lease row. Reuse a clean candidate or acquire a fresh sandbox through normal selection. Preserve exact same-run reattachment, workspace-less execution and ordinary task admission checks. - Preserve the newer prepared remote-workspace writer proof for native and legacy runs. Share its binding check between single-source recovery and one batched allocation read. - Keep persisted recovery-action IDs when constructing continuation source checks. Synthetic sources cannot grant snapshot receipt authority. - Permit one audited maintenance adoption for the first exhausted source-verification attempt. Require an unadmitted canonical saved board Retry, current user authority, unchanged assignment and latest source run. - Keep all company, source, allocation, workspace writer, controller and physical finalization ownership checks. The public read-only inspector keeps its strict policy check. - Pin the exact source for file maintenance. Preserve the previous generation and reuse the existing bounded recovery worker. Create no provider turn or new user request. - Restore the warm policy in the same transaction as verified native copyback completion. A failed write or changed settlement keeps a recovery obligation. - When local merge metadata is missing, settle the exact verified snapshot while keeping its unfinished allocation retained. Deliver the saved Retry through normal fresh isolation and attach the verified files. Keep the accepted result as history. - Require native copyback proof before reopening an accepted result's original workspace finalizer. - Validate snapshot receipts through one shared boundary for warm, retained and ephemeral sources. Bind the receipt to the current generation's operation IDs, company, issue, run, recovery action, lease and provider allocation. - Retire fixture input before removing recovery proof, and use the same configured heartbeat for HTTP and worker dispatch. Keep the exact catalog refusal and no-provider-work assertions. - Document the recovery contract in technical documentation. Leave capability maps unchanged for this bug fix. ## Verification - Candidate-selection RED: all six clean-candidate and fresh-allocation controls fail on unchanged `f54e7baf538` because a pending export aborts the whole acquisition. GREEN: the focused suite passes **15/15 cases** after candidate filtering. Both provider paths preserve the protected lease and never resume or destroy its allocation. Obsolete fingerprints cannot send that allocation to cleanup. Historical-row fencing, modern native/legacy writers, exact same-run reuse, workspace-less execution and the single-read allocation-history check remain covered. - The existing reusable-runtime selection cases passed **40/40**. Driver/capability contract suites passed **5/5**. Reusable service controls passed **2/2**. After merging current master, the combined lease/provider suites pass 61 cases and the recovery integration selection passes 36 cases, including the upstream inherited-session and resolved-interruption assertions. This gives 97 passing targeted cases on the final source. A duplicate test-app declaration from the automatic merge was removed while retaining the upstream assertions. - The source-history repair has meaningful red/green evidence: three valid retired-handoff cases could not complete under the old blanket history-row rejection. Settlement now uses the same full source inspector as adoption. It accepts exact retired handoffs and still rejects a predecessor that becomes active after export. The 26 focused history, rollback, snapshot and ambiguity cases passed before this last merge. - Server typecheck passes on the final merged source. Module boundaries and `git diff --check` pass. Full local repository tests and a successful current-head full build are not claimed. Current-head CI is complete on `0ef974d76483e11fb0b0304bcb05bbd3b56e2ef9`: both required gates pass, with 54 successful checks, three expected skips, one neutral Paperclip review check and a successful security status. [Full CI evidence](https://github.com/paperclipai/paperclip/actions/runs/41904339370). Greptile reviewed that exact head at **5/5** with no actionable findings. Its check completed successfully. [Ponytail also reviewed the exact head at 5/5](https://github.com/paperclipai/paperclip/pull/17916#issuecomment-6640083408). No review threads remain unresolved, and the branch is mergeable. - Earlier local broad runs exposed a dropped recovery-action ID in the merged source projection and deferred test input being revisited after its records were removed. Both were repaired. Later local disk exhaustion caused manifest and saved-file reserve failures; generated caches were removed only from these checkouts. The affected saved-file control passed unchanged after cleanup. The original logs are retained. No disk reserve or timeout was lowered. - On the prior head, the first [full CI attempt](https://github.com/paperclipai/paperclip/actions/runs/41899038530) received simultaneous runner shutdowns. Four jobs were cancelled, and two late jobs have lost-runner annotations with unavailable logs. The Runner job had five earlier ACPX failures whose full assertion report was lost on interruption. Their cause remains unproven. Identical Runner source passed the [#19004 control](https://github.com/paperclipai/paperclip/actions/runs/41898305521/job/126814429725), including all seven enabled tests in that file. - That attempt's [browser shard](https://github.com/paperclipai/paperclip/actions/runs/41899038530/job/126817404595) returned to the exact saved task URL, but its conversation was still loading at the unchanged heading deadline. The [artifact](https://github.com/paperclipai/paperclip/actions/runs/41899038530/artifacts/13358840931) is retained. All five Cloud-auth cases passed locally on the unchanged head and in the [same-source CI control](https://github.com/paperclipai/paperclip/actions/runs/41898305521/job/126814429966). The delay remains unexplained. One rerun of failed/interrupted jobs was submitted after the workflow completed. This new source revision requires fresh CI. - The earlier [original-head workflow](https://github.com/paperclipai/paperclip/actions/runs/40552228175) passed all 48 jobs, and prior head `f54e7baf538` received Greptile 5/5. These results do not establish readiness for the new head. The earlier Superagent scan was neutral after a pinned-base checkout failure; its external analysis remains unverified. - These tests use real PostgreSQL and controlled provider boundaries. Hosted rollout and original-task acceptance remain unverified. ## Risks - Maintenance can resume and stop the exact sandbox for file export. It cannot admit a provider turn, choose another allocation or bypass normal continuation gates. - A saved Retry alone is insufficient. A changed source, newer run, active writer, competing lease, lost user authority or previous adoption keeps the hold. - Policy restoration can fail. Completion and restoration now commit together, so normal recovery remains responsible for a retry. - Snapshot preservation cannot repair missing local merge metadata. Its old allocation remains explicitly retained. A snapshot receipt grants no permission to reuse or destroy that allocation, or to replay the accepted result. - The allocation check uses one projected join per candidate over the existing provider lease index. It adds no polling timer or recurring scan. A fresh fallback can create another sandbox while the unfinished source remains retained; it grants no authority to execute a provider turn. - This change does not establish the cause of a prior provider failure or prove that any specific hosted sandbox still contains its original files. ## Model Used OpenAI Codex, GPT-6 family, for the original change. OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), xhigh reasoning, for the merge and integration repairs. Both used repository tools and PostgreSQL verification. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Devin Foley --- doc/execution-semantics.md | 22 ++ server/src/services/environment-runtime.ts | 12 +- .../services/explicit-native-continuation.ts | 11 +- .../native-runtime/native-source-retention.ts | 138 +++++++++ .../native-workspace-finalization-state.ts | 61 +++- .../native-workspace-warm-reuse.test.ts | 280 ++++++++++++++++++ .../services/preserved-result-finalization.ts | 2 + .../services/remote-workspace-lifecycle.ts | 24 +- .../services/workspace-recovery-snapshot.ts | 24 +- .../src/services/workspace-recovery-source.ts | 7 +- .../src/services/workspace-recovery.test.ts | 253 +++++++++++++++- server/src/services/workspace-recovery.ts | 9 +- 12 files changed, 816 insertions(+), 27 deletions(-) create mode 100644 server/src/services/native-runtime/native-source-retention.ts create mode 100644 server/src/services/native-runtime/native-workspace-warm-reuse.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b7dcbe9e8e7..3dad5628de7 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1975,6 +1975,28 @@ remains protected. Cleanup never interprets this registration as a stop receipt or permission to replay provider work. Repeated cleanup visits register the source once; settled sources add no continuing discovery or new maintenance timer. +Unfinished prepared native exports also fence ordinary warm reuse of the same +physical allocation, including a different historical lease row. Candidate +selection excludes that allocation from reuse and obsolete-lease cleanup. It may +reuse another clean allocation or acquire a fresh sandbox; workspace recovery and +task admission still decide whether execution is allowed. Exact same-run +reattachment keeps its existing controller checks. A saved board Retry can adopt +one files-only maintenance generation when its first source-verification attempt +failed before export because the exact warm source lacked retention policy. +The old generation remains in the audit record. Adoption requires current user +authority, unchanged source identity, no competing allocation or workspace writer, +and the existing physical finalization owner. It grants no provider turn and does +not repeat a consumed Retry. Other failures and an already adopted source remain +held. Warm policy returns only after verified file persistence and exact stopped +allocation evidence. Policy restoration and the recovered receipt commit together; +failed settlement keeps the existing recovery obligation. Task continuation still +passes ordinary admission. +If local merge metadata is missing, a verified snapshot can settle preservation +while the unfinished allocation stays retained. Its exact snapshot operation +remains bound to the source and maintenance generation. Only verified native +copyback can reopen an accepted result's original workspace finalizer. A saved +Retry can instead use fresh isolation and attach the verified snapshot. + Retry and authenticated new messages also inspect the task's bound remote workspace for its exact failed writer. A newer setup failure cannot hide that writer's file obligation. A released reusable source can register preservation diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 31983fa5ac7..c75a714ef37 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -7,7 +7,7 @@ import { flushDeferredEnvironmentCleanup, registerDeferredEnvironmentCleanup, wr import { executeEnvironmentTask, pluginEnvironmentSupportsTasks, settleDispatchedEnvironmentTask } from "./environment-task-runtime.js"; import type { PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import { allowLegacyShutdownWorkspaceCleanup, legacyShutdownWorkspaceResourceProtected } from "./legacy-shutdown-workspace-settlement.js"; -import { nativeWorkspaceExportPendingForLease } from "./native-runtime/native-workspace-finalization-state.js"; +import { nativeWorkspaceExportPendingForLease, nativeWorkspaceExportPendingForResource } from "./native-runtime/native-workspace-finalization-state.js"; import { beginIdleTrackedWork } from "./task-admission.js"; import { historicalProviderCredentials } from "./historical-provider-credentials.js"; import { ownedWorkspaceRecoveryLease, type WorkspaceRecoverySourceIdentity } from "./workspace-recovery-source.js"; @@ -1720,7 +1720,10 @@ function createSandboxEnvironmentDriver( ) : []; for (let index = reusableCandidateLeases.length - 1; index >= 0; index--) { - if (await legacyShutdownWorkspaceResourceProtected(db, reusableCandidateLeases[index]!)) reusableCandidateLeases.splice(index, 1); + // Protected allocations must stay out of both selection and obsolete cleanup. + if (await nativeWorkspaceExportPendingForResource(db, reusableCandidateLeases[index]!, input.heartbeatRunId) + || await legacyShutdownWorkspaceResourceProtected(db, reusableCandidateLeases[index]!)) + reusableCandidateLeases.splice(index, 1); } const reusableExistingLeases = reusableCandidateLeases.filter((lease) => reusableSandboxLeaseScopeMatches({ @@ -2056,7 +2059,10 @@ function createSandboxEnvironmentDriver( ) : []; for (let index = reusableCandidateLeases.length - 1; index >= 0; index--) { - if (await legacyShutdownWorkspaceResourceProtected(db, reusableCandidateLeases[index]!)) reusableCandidateLeases.splice(index, 1); + // Protected allocations must stay out of both selection and obsolete cleanup. + if (await nativeWorkspaceExportPendingForResource(db, reusableCandidateLeases[index]!, input.heartbeatRunId) + || await legacyShutdownWorkspaceResourceProtected(db, reusableCandidateLeases[index]!)) + reusableCandidateLeases.splice(index, 1); } const reusableExistingLeases = reusableCandidateLeases.filter((lease) => reusableSandboxLeaseScopeMatches({ diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 247c260a236..1bcda896860 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -2,6 +2,7 @@ import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-n import { assertDurableChatWakeupRequest, assertDurableChatWakeupReceipt, type DurableChatWakeupRequest } from "./durable-chat-wakeup.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { canSelectFreshWorkspace, type ContinuationReadiness } from "./fresh-workspace-continuation.js"; +import { verifiedWorkspaceSnapshotReceipt } from "./workspace-recovery-snapshot.js"; import { readTaskRetryIntent } from "./task-retry.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; @@ -240,7 +241,8 @@ export async function admitExplicitNativeContinuation(input: { and result.schema_status = 'accepted' and result.result_json#>>'{terminal,runTerminalState}' = 'succeeded')`, )), )).for("update"); - const sourceChecks = actions.map(action => ({ cause: action.cause, evidence: action.evidence })); + const sourceChecks: Array<{ id?: string; cause: string; evidence: Record }> = + actions.map(action => ({ id: action.id, cause: action.cause, evidence: action.evidence })); if (nativeQueuedSend) { const [latest] = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode }) .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), @@ -355,7 +357,12 @@ export async function admitExplicitNativeContinuation(input: { eq(nativeRunResults.schemaStatus, "accepted"), sql`${nativeRunResults.resultJson}#>>'{terminal,runTerminalState}' = 'succeeded'`, )).limit(1); - if (completedResult) return blocked("finalizing_saved_result", "Paperclip is finishing the saved result."); + const recovery = readSystemRecovery(action.evidence); + const snapshotSettled = completedResult && freshIntent && action.id && recovery?.state === "complete" + && await verifiedWorkspaceSnapshotReceipt(db, { companyId, issueId, runId: run.id, + actionId: action.id, operationIds: recovery.preservationOperationIds }); + if (completedResult && !(freshIntent && snapshotSettled)) + return blocked("finalizing_saved_result", "Paperclip is finishing the saved result."); } // A repaired export no longer blocks execution. Its machine receipt still // binds the saved user turn; finalized file state alone is insufficient. diff --git a/server/src/services/native-runtime/native-source-retention.ts b/server/src/services/native-runtime/native-source-retention.ts new file mode 100644 index 00000000000..d4b18b7b786 --- /dev/null +++ b/server/src/services/native-runtime/native-source-retention.ts @@ -0,0 +1,138 @@ +import { and, desc, eq, or, sql } from "drizzle-orm"; +import { agentWakeupRequests, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { persistActivity } from "../activity-log.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; +import { createSystemRecovery, readSystemRecovery, SystemRecoveryFailure, type SystemRecoveryState } from "../system-recovery-state.js"; +import { updateSystemRecovery } from "../system-recovery.js"; +import { readTaskRetryIntent, taskRetryMutationAllowed } from "../task-retry.js"; +import { inspectWorkspaceRecoverySource, type WorkspaceRecoverySourceIdentity } from "../workspace-recovery-source.js"; +import { hasVerifiedWorkspaceRepair } from "../workspace-recovery-receipt.js"; +import { verifiedWorkspaceSnapshotReceipt } from "../workspace-recovery-snapshot.js"; +import type { NativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; + +export function canAdoptNativeSourceRetention(state: SystemRecoveryState) { + return state.state === "exhausted" && state.phase === "verify_execution" && state.attempts === 1 + && state.retriesUsed === 0 && state.failure?.code === "source_binding_invalid" + && (!state.failure.sourceVerification || state.failure.sourceVerification.reason === "lease_policy_not_retained") + && state.preservationOperationIds.length === 0 && !state.freshAttemptRunId; +} + +export async function retainNativeRecoverySource(db: Db, input: WorkspaceRecoverySourceIdentity, + previous: SystemRecoveryState, ownership: NativeWorkspaceFinalizationOwnership): Promise { + if (!canAdoptNativeSourceRetention(previous)) return null; + await ownership.assertHeld(); + try { + return await db.transaction(async tx => { + const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.issueId, input.issueId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))).for("update"); + const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId), + eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, input.actionId))).for("update"); + const current = readSystemRecovery(action?.evidence); + if (!issue || !run || !action || !current || current.generationId !== previous.generationId || current.revision !== previous.revision + || action.evidence.nativeSourceRetention || action.status === "cancelled" || action.evidence.executionReconciliation + || !coordinator || coordinator.phase !== "terminal_failure" || coordinator.leaseOwner + || coordinator.failureDetail?.successorRunId || issue.assigneeAgentId !== run.agentId || issue.executionRunId || issue.checkoutRunId + || ["done", "cancelled"].includes(issue.status)) return null; + const [latest] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + if (latest?.id !== run.id) return null; + const [request] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.reason, "task_retry"), eq(agentWakeupRequests.status, "deferred_issue_execution"), + eq(agentWakeupRequests.requestedByActorType, "user"), sql`${agentWakeupRequests.runId} is null`, + sql`${agentWakeupRequests.payload}->>'issueId' = ${input.issueId}`, + sql`${agentWakeupRequests.payload}#>>'{taskRetry,sourceRunId}' = ${input.runId}`, + )).limit(1).for("update"); + const intent = readTaskRetryIntent(request?.payload); + if (!request || !intent || intent.generationId !== request.id || intent.actorId !== request.requestedByActorId + || intent.chatActionId || intent.boardMessageOrigin || intent.canonicalIntentId + || !await taskRetryMutationAllowed(tx as unknown as Db, { companyId: input.companyId, + issueId: input.issueId, actorId: intent.actorId, actorSource: intent.actorSource })) return null; + const inspected = await inspectWorkspaceRecoverySource(tx as unknown as Db, input, { maximumSources: 1, prepareNativeRetention: true }); + const source = inspected.sources[0]; + if (!source?.reference || source.reference.state !== "prepared" || source.reference.resourceDisposition !== "keep_running" + || source.lease.leasePolicy !== "reuse_by_environment" || !["released", "retained"].includes(source.lease.status)) return null; + const marker = { schema: "paperclip.native-source-retention.v1", sourceRunId: run.id, actionId: action.id, + requestId: request.id, leaseId: source.lease.id, providerLeaseId: source.lease.providerLeaseId, + descriptorSha256: source.reference.descriptorSha256, baselineSha256: source.reference.baselineSha256, + workspaceId: source.reference.workspaceId, originalLeasePolicy: "reuse_by_environment" }; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); + const [retained] = await tx.update(environmentLeases).set({ leasePolicy: "retain_on_failure", updatedAt: new Date(), + metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ nativeSourceRetention: marker })}::jsonb`, + }).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, source.lease.id), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.leasePolicy, "reuse_by_environment"), + eq(environmentLeases.status, source.lease.status), + sql`${environmentLeases.metadata} is not distinct from ${JSON.stringify(source.lease.metadata)}::jsonb`, + )).returning({ id: environmentLeases.id }); + if (!retained) return null; + const next = { ...createSystemRecovery(run.id), intentId: request.id, revision: previous.revision + 1, + providerCooldownAt: previous.providerCooldownAt, nextAttemptAt: previous.providerCooldownAt }; + if (!await updateSystemRecovery(tx as unknown as Db, input, previous, next)) throw new Error("workspace_recovery_generation_changed"); + await tx.update(issueRecoveryActions).set({ evidence: sql`${issueRecoveryActions.evidence} || + ${JSON.stringify({ nativeSourceRetention: { ...marker, previousGeneration: previous } })}::jsonb`, + }).where(eq(issueRecoveryActions.id, action.id)); + await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "native-source-retention", + action: "issue.native_workspace_source_retained", entityType: "issue", entityId: input.issueId, runId: run.id, + details: { recoveryActionId: action.id, requestId: request.id, leaseId: source.lease.id, + previousGeneration: previous, nextGenerationId: next.generationId } }); + return next; + }); + } catch (error) { + if (error instanceof SystemRecoveryFailure) return null; + throw error; + } +} + +export async function settleRecoveredNativeSourceRetention(db: Db, input: WorkspaceRecoverySourceIdentity, operationIds: string[]) { + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))); + if (!run) return false; + const reference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); + if (!reference || reference.resourceDisposition !== "keep_running") return false; + const [lease] = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.id, reference.leaseId), eq(environmentLeases.heartbeatRunId, run.id))); + const marker = lease?.metadata?.nativeSourceRetention as Record | undefined; + if (!lease || marker?.schema !== "paperclip.native-source-retention.v1" || marker.sourceRunId !== run.id + || marker.actionId !== input.actionId || marker.leaseId !== lease.id || marker.providerLeaseId !== reference.providerLeaseId + || lease.providerLeaseId !== reference.providerLeaseId || lease.metadata?.remoteCwd !== reference.remoteCwd + || marker.baselineSha256 !== reference.baselineSha256 || marker.workspaceId !== reference.workspaceId + || marker.originalLeasePolicy !== "reuse_by_environment" + || !hasRemoteTerminationReceipt(lease) || lease.metadata?.pendingCleanupInFlight === true + || (lease.metadata?.remoteExecutionTermination as { state?: string })?.state !== "stopped") return false; + // Recheck the same retired-predecessor and live-owner proof used at + // adoption. Valid allocation history must not prevent policy settlement. + const inspected = await inspectWorkspaceRecoverySource(db, input, { maximumSources: 1 }); + if (inspected.sources[0]?.lease.id !== lease.id) return false; + if (reference.state === "prepared") { + const operation = await verifiedWorkspaceSnapshotReceipt(db, { ...input, operationIds, + leaseId: lease.id, providerLeaseId: reference.providerLeaseId }); + if (!operation) return false; + const [retained] = await db.update(environmentLeases).set({ updatedAt: new Date(), + metadata: { ...lease.metadata, nativeSourceRetention: { ...marker, settlement: "snapshot_retained", snapshotOperationId: operation.id } }, + }).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, lease.id), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.leasePolicy, "retain_on_failure"), + eq(environmentLeases.status, "released"), sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`, + )).returning({ id: environmentLeases.id }); + if (!retained) return false; + const [settled] = await db.update(issueRecoveryActions).set({ evidence: sql`jsonb_set(${issueRecoveryActions.evidence}, + '{nativeSourceRetention}', (${issueRecoveryActions.evidence}->'nativeSourceRetention') || + ${JSON.stringify({ settlement: "snapshot_retained", snapshotOperationId: operation.id })}::jsonb)` }).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId), + eq(issueRecoveryActions.id, input.actionId), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, + )).returning({ id: issueRecoveryActions.id }); + return Boolean(settled); + } + if (!await hasVerifiedWorkspaceRepair(db, run)) return false; + const restored = await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment", updatedAt: new Date() }).where(and( + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, lease.id), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.leasePolicy, "retain_on_failure"), + eq(environmentLeases.status, "released"), sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`, + )).returning({ id: environmentLeases.id }); + return restored.length === 1; +} diff --git a/server/src/services/native-runtime/native-workspace-finalization-state.ts b/server/src/services/native-runtime/native-workspace-finalization-state.ts index 250c070d968..d3c0994a4f4 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-state.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-state.ts @@ -1,10 +1,9 @@ import { environmentLeases, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { and, eq, exists, sql } from "drizzle-orm"; -import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; -import { preparedRemoteWorkspaceSource, type WorkspaceSourceLease } from "../remote-workspace-lifecycle.js"; +import { matchPreparedRemoteWorkspaceSource, preparedRemoteWorkspaceSource, type WorkspaceSourceLease } from "../remote-workspace-lifecycle.js"; /** File-only recovery for setup that never reached native admission. The live * preparer must still own its controller; orphan repair accepts terminal runs @@ -44,10 +43,9 @@ export async function preserveRemoteWorkspacePreparationFailure(db: Db, lease: W }); } -export async function nativeWorkspaceExportPendingForLease(db: Db, lease: WorkspaceSourceLease): Promise { - if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return false; +function nativeWorkspaceRunProjection() { const input = sql`${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput'`; - const [run] = await db.select({ + return { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, runtimeMode: heartbeatRuns.runtimeMode, driverKind: heartbeatRuns.driverKind, @@ -62,17 +60,58 @@ export async function nativeWorkspaceExportPendingForLease(db: Db, lease: Worksp and ${nativeRunFinalizations.phase} = 'observed' and ${nativeRunFinalizations.resultId} is null)`, executionScope: sql>`jsonb_build_object( 'schema', ${input}->'schema', 'binding', ${input}->'binding', 'workspace', ${input}->'workspace', - 'provider', jsonb_build_object('kind', ${input}#>'{provider,kind}'))`, - }).from(heartbeatRuns).where(and( + 'provider', jsonb_build_object('kind', ${input}#>'{provider,kind}', 'bridgeRevision', ${input}#>'{provider,bridgeRevision}'))`, + controllerBootId: heartbeatRuns.controllerBootId, + contextIssueId: sql`${heartbeatRuns.contextSnapshot}->>'issueId'`, + remoteReference: sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`, + }; +} + +function nativeWorkspaceSourcePending(run: Pick & { + reference: unknown; executionScope: Record; observedRemoteWorkspace: boolean; + }, lease: WorkspaceSourceLease) { + if (run.runtimeMode !== "native") return false; + if (run.observedRemoteWorkspace) return true; + if (nativeRunHasNoWorkspaceAccess({ ...run, runnerProfileJson: { nativeExecutionInput: run.executionScope } })) return false; + const reference = readNativeWorkspaceSyncReference(run.reference); + return reference?.state === "prepared" && reference.leaseId === lease.id + && reference.providerLeaseId === lease.providerLeaseId; +} + +export async function nativeWorkspaceExportPendingForLease(db: Db, lease: WorkspaceSourceLease): Promise { + if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return false; + const [run] = await db.select(nativeWorkspaceRunProjection()).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, lease.heartbeatRunId), eq(heartbeatRuns.companyId, lease.companyId), eq(heartbeatRuns.runtimeMode, "native"), )).limit(1); if (run?.observedRemoteWorkspace) return true; if (await preparedRemoteWorkspaceSource(db, lease)) return true; - if (!run || nativeRunHasNoWorkspaceAccess({ ...run, runnerProfileJson: { nativeExecutionInput: run.executionScope } })) return false; - const reference = readNativeWorkspaceSyncReference(run.reference); - return reference?.state === "prepared" && reference.leaseId === lease.id - && reference.providerLeaseId === lease.providerLeaseId; + return Boolean(run && nativeWorkspaceSourcePending(run, lease)); +} + +export async function nativeWorkspaceExportPendingForResource(db: Db, lease: WorkspaceSourceLease, resumingRunId: string | null) { + if (!lease.provider || !lease.providerLeaseId || lease.provider === "local") return false; + const sources = await db.select({ lease: { id: environmentLeases.id, companyId: environmentLeases.companyId, + heartbeatRunId: environmentLeases.heartbeatRunId, provider: environmentLeases.provider, providerLeaseId: environmentLeases.providerLeaseId, + issueId: environmentLeases.issueId, executionWorkspaceId: environmentLeases.executionWorkspaceId, + environmentId: environmentLeases.environmentId, metadata: environmentLeases.metadata }, + run: nativeWorkspaceRunProjection(), + workspace: { id: executionWorkspaces.id, companyId: executionWorkspaces.companyId, sourceIssueId: executionWorkspaces.sourceIssueId, + mode: executionWorkspaces.mode, metadata: executionWorkspaces.metadata }, + }).from(environmentLeases).innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, environmentLeases.companyId), + eq(heartbeatRuns.id, environmentLeases.heartbeatRunId))).leftJoin(executionWorkspaces, and( + eq(executionWorkspaces.companyId, heartbeatRuns.companyId), + sql`${executionWorkspaces.id}::text = ${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,workspaceId}'`, + )).where(and(eq(environmentLeases.provider, lease.provider), eq(environmentLeases.providerLeaseId, lease.providerLeaseId))); + for (const source of sources) { + if (source.lease.id === lease.id && source.lease.heartbeatRunId === resumingRunId) continue; + const remote = matchPreparedRemoteWorkspaceSource({ lease: source.lease, run: { ...source.run, + issueId: source.run.nativeIssueId ?? source.run.issueId ?? source.run.contextIssueId, + reference: source.run.remoteReference, legacyReference: source.run.reference } }, source.workspace); + if (remote || nativeWorkspaceSourcePending(source.run, source.lease)) return true; + } + return false; } /** diff --git a/server/src/services/native-runtime/native-workspace-warm-reuse.test.ts b/server/src/services/native-runtime/native-workspace-warm-reuse.test.ts new file mode 100644 index 00000000000..defc8da40e9 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-warm-reuse.test.ts @@ -0,0 +1,280 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { agents, agentWakeupRequests, applyPendingMigrations, companies, createDb, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, plugins } from "@paperclipai/db"; +import type { Environment } from "@paperclipai/shared"; +import type { PluginWorkerManager } from "../plugin-worker-manager.js"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { environmentRuntimeService } from "../environment-runtime.js"; +import * as leases from "../environments.js"; +import * as providers from "../sandbox-provider-runtime.js"; +import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "../system-recovery-state.js"; +import { recoverRetainedWorkspace } from "../workspace-recovery.js"; +import { nativeWorkspaceExportPendingForLease, nativeWorkspaceExportPendingForResource } from "./native-workspace-finalization-state.js"; + +describe("native workspace export warm reuse fence", () => { + let db: ReturnType; + let temporary: Awaited>; + beforeAll(async () => { + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_RECOVERY_TEST_DATABASE_URL"); + if (external) { await applyPendingMigrations(external); temporary = { connectionString: external, cleanup: async () => {} }; } + else temporary = await startEmbeddedPostgresTestDatabase("native-warm-reuse-"); + db = createDb(temporary.connectionString); + }, 30_000); + afterEach(() => vi.restoreAllMocks()); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + + async function seed(plugin: boolean) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), pluginId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Native warm fence", issuePrefix: `W${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Keep the source", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, runtimeMode: "native", status: "running" }); + const provider = plugin ? "native-fence-plugin" : "fake"; + if (plugin) await db.insert(plugins).values({ id: pluginId, pluginKey: `native-fence.${pluginId}`, + packageName: "native-fence-provider", version: "1.0.0", apiVersion: 1, categories: ["automation"], status: "ready", + manifestJson: { id: `native-fence.${pluginId}`, apiVersion: 1, version: "1.0.0", displayName: "Native fence provider", + description: "Controlled lifecycle boundary", author: "Paperclip", categories: ["automation"], + capabilities: ["environment.drivers.register"], entrypoints: { worker: "dist/worker.js" }, + environmentDrivers: [{ driverKey: provider, kind: "sandbox_provider", displayName: "Native fence", supportsReusableLeases: true, + configSchema: { type: "object", properties: { image: { type: "string" }, reuseLease: { type: "boolean" } } } }] } }); + const [environment] = await db.insert(environments).values({ name: runId, driver: "sandbox", status: "active", + config: { provider, image: "fake:test", reuseLease: true } }).returning(); + const providerLeaseId: string = randomUUID(); + let acquisitions = 0; + const call = vi.fn(async (_id: string, method: string, input?: Record) => { + if (method === "environmentAcquireLease") return { providerLeaseId: acquisitions++ ? randomUUID() : providerLeaseId, + metadata: { remoteCwd: "/work" } }; + if (method === "environmentResumeLease") return { providerLeaseId: String(input?.providerLeaseId ?? providerLeaseId), metadata: { remoteCwd: "/work" } }; + if (["environmentReleaseLease", "environmentStopLease"].includes(method)) return { providerLeaseId, state: "stopped" }; + if (method === "environmentDestroyLease") return { providerLeaseId, state: "destroyed" }; + throw new Error(method); + }); + const worker = { isRunning: () => true, call, getWorker: () => ({ supportedMethods: ["environmentResumeLease", + "environmentReleaseLease", "environmentStopLease", "environmentDestroyLease"] }) } as unknown as PluginWorkerManager; + if (!plugin) { + const original = providers.getSandboxProvider; + vi.spyOn(providers, "getSandboxProvider").mockImplementation(key => { + const value = original(key); + return value && key === "fake" ? Object.assign(Object.create(value), { supportsReusableLeases: true }) : value; + }); + } + const runtime = environmentRuntimeService(db, { pluginWorkerManager: worker }); + const input = { companyId, agentId, issueId, environment: environment as unknown as Environment, + heartbeatRunId: runId, persistedExecutionWorkspace: null }; + const acquired = await runtime.acquireRunLease(input); + await runtime.releaseRunLeases(runId, "released", undefined, "keep_running"); + const reference = { schema: "paperclip.native-workspace-sync/v3", state: "prepared", leaseId: acquired.lease.id, + providerLeaseId: acquired.lease.providerLeaseId, workspaceId: issueId, remoteCwd: "/work", resourceDisposition: "keep_running", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null }; + await db.update(heartbeatRuns).set({ status: "failed", runnerProfileJson: { nativeWorkspaceSync: reference } }).where(eq(heartbeatRuns.id, runId)); + const nextRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: nextRunId, companyId, agentId, issueId, nativeIssueId: issueId, runtimeMode: "native", status: "running" }); + call.mockClear(); + return { runtime, input, runId, nextRunId, reference, lease: acquired.lease, call }; + } + + async function expectFreshAllocation(f: Awaited>, runtime = f.runtime) { + const [protectedLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + const fresh = await runtime.acquireRunLease({ ...f.input, heartbeatRunId: f.nextRunId }); + expect(fresh.lease.providerLeaseId).not.toBe(f.lease.providerLeaseId); + expect(fresh.lease.heartbeatRunId).toBe(f.nextRunId); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(protectedLease); + expect(f.call.mock.calls.map(call => call[1])).not.toContain("environmentResumeLease"); + expect(f.call.mock.calls.map(call => call[1])).not.toContain("environmentDestroyLease"); + return fresh; + } + + it.each([false, true])("allocates a fresh sandbox while retaining its unfinished source for plugin=%s", async plugin => { + const f = await seed(plugin); + const acquire = vi.spyOn(providers, "acquireSandboxProviderLease"); + const destroy = vi.spyOn(providers, "destroySandboxProviderLease"); + await expectFreshAllocation(f); + expect(destroy).not.toHaveBeenCalled(); + if (plugin) expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentAcquireLease"]); + else expect(acquire).toHaveBeenCalledWith(expect.objectContaining({ reusableProviderLeaseId: null })); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].runnerProfileJson) + .toMatchObject({ nativeWorkspaceSync: { state: "prepared", leaseId: f.lease.id } }); + }); + + it.each([{ plugin: false, obsolete: false }, { plugin: true, obsolete: false }, + { plugin: false, obsolete: true }, { plugin: true, obsolete: true }])( + "reuses a clean candidate without touching a pending allocation: plugin=$plugin obsolete=$obsolete", async ({ plugin, obsolete }) => { + const f = await seed(plugin), cleanRunId = randomUUID(), cleanLeaseId = randomUUID(), cleanProviderLeaseId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: cleanRunId, companyId: f.input.companyId, agentId: f.input.agentId, + issueId: f.input.issueId, runtimeMode: "legacy", status: "succeeded", finishedAt: new Date() }); + await db.insert(environmentLeases).values({ id: cleanLeaseId, companyId: f.input.companyId, environmentId: f.input.environment.id, + heartbeatRunId: cleanRunId, issueId: f.input.issueId, provider: f.lease.provider, providerLeaseId: cleanProviderLeaseId, + leasePolicy: "reuse_by_environment", status: "released", cleanupStatus: "success", metadata: f.lease.metadata }); + if (obsolete) await db.update(environmentLeases).set({ metadata: { ...f.lease.metadata, reusableSandboxLease: null } }) + .where(eq(environmentLeases.id, f.lease.id)); + const [protectedLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + // The provider boundary returns the exact clean allocation selected by + // the real policy. No source inspection or database reads are mocked. + const acquire = vi.spyOn(providers, "acquireSandboxProviderLease"); + if (!plugin) acquire.mockImplementation(async input => ({ providerLeaseId: input.reusableProviderLeaseId!, + metadata: { provider: "fake", image: "fake:test", reuseLease: true } })); + const destroy = vi.spyOn(providers, "destroySandboxProviderLease"); + const resumed = await f.runtime.acquireRunLease({ ...f.input, heartbeatRunId: f.nextRunId }); + expect(resumed.lease).toMatchObject({ heartbeatRunId: f.nextRunId, providerLeaseId: cleanProviderLeaseId }); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(protectedLease); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, cleanLeaseId)))[0].status).toBe("expired"); + expect(destroy).not.toHaveBeenCalled(); + if (plugin) { + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentResumeLease"]); + expect(f.call.mock.calls[0]?.[2]).toMatchObject({ providerLeaseId: cleanProviderLeaseId }); + } else expect(acquire).toHaveBeenCalledWith(expect.objectContaining({ reusableProviderLeaseId: cleanProviderLeaseId })); + }); + + it("checks the allocation even when the candidate snapshot contains only a rebound row", async () => { + const f = await seed(true); + const service = leases.environmentService(db); + const [rebound] = await db.insert(environmentLeases).values({ companyId: f.input.companyId, environmentId: f.input.environment.id, + issueId: f.input.issueId, heartbeatRunId: f.nextRunId, provider: f.lease.provider, providerLeaseId: f.lease.providerLeaseId, + leasePolicy: "reuse_by_environment", status: "retained", metadata: f.lease.metadata }).returning(); + const list = service.listLeases; + service.listLeases = async (...args) => (await list(...args)).filter(lease => lease.id === rebound.id); + vi.spyOn(leases, "environmentService").mockReturnValue(service); + const runtime = environmentRuntimeService(db, { pluginWorkerManager: { isRunning: () => true, call: f.call, + getWorker: () => ({ supportedMethods: ["environmentResumeLease", "environmentReleaseLease", "environmentDestroyLease"] }) } as unknown as PluginWorkerManager }); + await expectFreshAllocation(f, runtime); + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentAcquireLease"]); + }); + + it.each(["native", "legacy"])("fences a historical modern %s writer in one allocation read", async runtimeMode => { + const f = await seed(true), workspaceId = randomUUID(); + const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId: f.lease.id, + providerLeaseId: f.lease.providerLeaseId!, remoteCwd: "/work", epoch: 1, state: "prepared", snapshotId: null }; + const authority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: null, source: { kind: "empty" }, + writerRunId: f.runId, dirty: true, location: { environmentId: f.input.environment.id, + leaseId: f.lease.id, providerLeaseId: f.lease.providerLeaseId!, remoteCwd: "/work" } }; + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.input.companyId, + sourceIssueId: f.input.issueId, name: "Unsaved remote source", cwd: "/work", providerRef: "/work", + providerType: "cloud_sandbox", mode: "isolated_workspace", strategyType: "git_worktree", + status: "active", metadata: { remoteWorkspace: authority } }); + await db.update(environmentLeases).set({ executionWorkspaceId: workspaceId }).where(eq(environmentLeases.id, f.lease.id)); + await db.update(heartbeatRuns).set({ runtimeMode, nativeIssueId: runtimeMode === "native" ? f.input.issueId : null, + runnerProfileJson: { remoteWorkspace: reference } }).where(eq(heartbeatRuns.id, f.runId)); + const [rebound] = await db.insert(environmentLeases).values({ companyId: f.input.companyId, environmentId: f.input.environment.id, + issueId: f.input.issueId, heartbeatRunId: f.nextRunId, provider: f.lease.provider, providerLeaseId: f.lease.providerLeaseId, + leasePolicy: "reuse_by_environment", status: "retained", metadata: f.lease.metadata }).returning(); + const reads = vi.spyOn(db, "select"); + expect(await nativeWorkspaceExportPendingForResource(db, rebound, f.nextRunId)).toBe(true); + expect(reads).toHaveBeenCalledTimes(1); + reads.mockRestore(); + await expectFreshAllocation(f); + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentAcquireLease"]); + // A saved descriptor or a changed writer cannot keep fencing an allocation. + await db.update(heartbeatRuns).set({ runnerProfileJson: { remoteWorkspace: { ...reference, state: "saved" } } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await nativeWorkspaceExportPendingForResource(db, rebound, f.nextRunId)).toBe(false); + await db.update(heartbeatRuns).set({ runnerProfileJson: { remoteWorkspace: reference } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(executionWorkspaces).set({ metadata: { remoteWorkspace: { ...authority, writerRunId: f.nextRunId } } }) + .where(eq(executionWorkspaces.id, workspaceId)); + expect(await nativeWorkspaceExportPendingForResource(db, rebound, f.nextRunId)).toBe(false); + }); + + it("rechecks a prepared export after selecting a stale candidate and before resuming it", async () => { + const f = await seed(true); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, + state: "finalized", finalHostSha256: "c".repeat(64) } } }).where(eq(heartbeatRuns.id, f.runId)); + const service = leases.environmentService(db), list = service.listLeases; + service.listLeases = async (...args) => { + const snapshot = await list(...args); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: f.reference } }).where(eq(heartbeatRuns.id, f.runId)); + return snapshot; + }; + vi.spyOn(leases, "environmentService").mockReturnValue(service); + const runtime = environmentRuntimeService(db, { pluginWorkerManager: { isRunning: () => true, call: f.call, + getWorker: () => ({ supportedMethods: ["environmentResumeLease", "environmentReleaseLease", "environmentDestroyLease"] }) } as unknown as PluginWorkerManager }); + await expectFreshAllocation(f, runtime); + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentAcquireLease"]); + }); + + it("preserves exact same-run reattachment without allocating a replacement", async () => { + const f = await seed(true); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.runId)); + const resumed = await f.runtime.acquireRunLease(f.input); + expect(resumed.lease.id).toBe(f.lease.id); + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentResumeLease"]); + }); + + it("checks a long allocation history in one read and allows finalized workspace reuse", async () => { + const f = await seed(true); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, + state: "finalized", finalHostSha256: "c".repeat(64) } } }).where(eq(heartbeatRuns.id, f.runId)); + const history = Array.from({ length: 64 }, () => randomUUID()); + await db.insert(heartbeatRuns).values(history.map(id => ({ id, companyId: f.input.companyId, agentId: f.input.agentId, + issueId: f.input.issueId, nativeIssueId: f.input.issueId, runtimeMode: "native", status: "succeeded", + runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, state: "finalized", finalHostSha256: "c".repeat(64) } } }))); + await db.insert(environmentLeases).values(history.map(heartbeatRunId => ({ companyId: f.input.companyId, + environmentId: f.input.environment.id, heartbeatRunId, provider: f.lease.provider, + providerLeaseId: f.lease.providerLeaseId, leasePolicy: "reuse_by_environment", status: "released" }))); + const reads = vi.spyOn(db, "select"); + expect(await nativeWorkspaceExportPendingForResource(db, f.lease, f.nextRunId)).toBe(false); + expect(reads).toHaveBeenCalledTimes(1); + reads.mockRestore(); + const resumed = await f.runtime.acquireRunLease({ ...f.input, heartbeatRunId: f.nextRunId }); + expect(resumed.lease.id).not.toBe(f.lease.id); + expect(resumed.lease).toMatchObject({ heartbeatRunId: f.nextRunId, providerLeaseId: f.lease.providerLeaseId }); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]) + .toMatchObject({ heartbeatRunId: f.runId, status: "expired" }); + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentResumeLease"]); + }); + + it.each(["grok_bot", "muse"])("does not fence an exactly bound workspace-less %s run", async kind => { + const f = await seed(true); + const binding = { runId: f.runId, companyId: f.input.companyId, agentId: f.input.agentId, issueId: f.input.issueId }; + const profile = { nativeWorkspaceSync: f.reference, nativeExecutionInput: { + schema: kind === "muse" ? "paperclip.native-execution-input.v8" : "paperclip.native-execution-input.v6", + binding, workspace: { access: "none", cwd: null }, provider: { kind, bridgeRevision: "muse-v1" } } }; + await db.update(heartbeatRuns).set({ driverKind: kind === "muse" ? "muse_external" : "grok_bot_mcp", + runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + expect(await nativeWorkspaceExportPendingForLease(db, f.lease)).toBe(false); + expect(await nativeWorkspaceExportPendingForResource(db, f.lease, f.nextRunId)).toBe(false); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeExecutionInput: { + ...profile.nativeExecutionInput, binding: { ...binding, runId: randomUUID() } } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await nativeWorkspaceExportPendingForResource(db, f.lease, f.nextRunId)).toBe(true); + }); + + it("does not take over a source while an ordinary provider resume is awaiting its database handoff", async () => { + const f = await seed(true), requestId = randomUUID(), actionId = randomUUID(); + await db.update(heartbeatRuns).set({ runnerProfileJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + let entered!: () => void, release!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const pending = new Promise(resolve => { release = resolve; }); + f.call.mockImplementation(async (_id: string, method: string) => { + if (method !== "environmentResumeLease") throw new Error(method); + entered(); await pending; + return { providerLeaseId: f.lease.providerLeaseId!, metadata: { remoteCwd: "/work" } }; + }); + const ordinary = f.runtime.acquireRunLease({ ...f.input, heartbeatRunId: f.nextRunId }); + try { + await started; + const failure = failSystemRecovery(reserveSystemRecoveryAttempt(createSystemRecovery(f.runId, requestId))!, + new SystemRecoveryFailure("source_binding_invalid", "service_required")); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: f.reference }, + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] } } }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunFinalizations).values({ companyId: f.input.companyId, issueId: f.input.issueId, + runId: f.runId, phase: "terminal_failure" }); + await db.insert(issueRecoveryActions).values({ id: actionId, companyId: f.input.companyId, sourceIssueId: f.input.issueId, + kind: "active_run_watchdog", ownerType: "system", cause: "native_workspace_sync_out_retry_exhausted", fingerprint: f.runId, + nextAction: "Preserve the source", evidence: { runId: f.runId, systemRecovery: failure, workspaceRecovery: { + schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "blocked", stage: "verifying_stop", + attempt: 1, requestIds: [requestId] } } }); + await db.insert(agentWakeupRequests).values({ id: requestId, companyId: f.input.companyId, agentId: f.input.agentId, + source: "on_demand", reason: "task_retry", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "board", + payload: { issueId: f.input.issueId, taskRetry: { schema: "paperclip.task-retry.v1", generationId: requestId, + sourceRunId: f.runId, actorId: "board", actorSource: "local_implicit", strategy: "evaluate" } } }); + await recoverRetainedWorkspace({ db, companyId: f.input.companyId, issueId: f.input.issueId, + runId: f.runId, actionId, environmentRuntime: f.runtime }); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, actionId)); + expect(action.evidence.systemRecovery).toEqual(failure); + expect(action.evidence.nativeSourceRetention).toBeUndefined(); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].leasePolicy).toBe("reuse_by_environment"); + expect(f.call.mock.calls.map(call => call[1])).toEqual(["environmentResumeLease"]); + } finally { release(); await ordinary; } + }); +}); diff --git a/server/src/services/preserved-result-finalization.ts b/server/src/services/preserved-result-finalization.ts index 69c47de6ac5..627f0e87bd5 100644 --- a/server/src/services/preserved-result-finalization.ts +++ b/server/src/services/preserved-result-finalization.ts @@ -2,6 +2,7 @@ import { notifyNativeFinalizationWork } from "./native-finalization-notification import { and, desc, eq, gte, ne, or, sql } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { readSystemRecovery } from "./system-recovery-state.js"; +import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; /** Re-enter the existing finalizer only after preservation. This does not * recreate a result, grant an executor budget, or apply it over newer work. */ @@ -37,6 +38,7 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin )).orderBy(desc(issueRecoveryActions.createdAt)).limit(1); const recovery = readSystemRecovery(action?.evidence); if (!result || recovery?.state !== "complete" || !recovery.preservationOperationIds.length) return false; + if (!await hasVerifiedWorkspaceRepair(tx as unknown as Db, run)) return false; const [intent] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.reason, "task_retry"), diff --git a/server/src/services/remote-workspace-lifecycle.ts b/server/src/services/remote-workspace-lifecycle.ts index aa4d4097086..b4d24873fe0 100644 --- a/server/src/services/remote-workspace-lifecycle.ts +++ b/server/src/services/remote-workspace-lifecycle.ts @@ -31,20 +31,36 @@ export async function preparedRemoteWorkspaceSource(db: Db, lease: WorkspaceSour if (!bound) return null; const parsed = remoteWorkspaceRunReferenceSchema.safeParse(bound.run.reference); if (!parsed.success || parsed.data.state !== "prepared" || bound.run.legacyReference != null) return null; + const [workspace] = await db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, lease.companyId), eq(executionWorkspaces.id, parsed.data.workspaceId))).limit(1); + return matchPreparedRemoteWorkspaceSource(bound, workspace); +} + +type PreparedSourceBinding = { + lease: Pick; + run: { id: string; runtimeMode: string; controllerBootId: string | null; issueId: string | null; + reference: unknown; legacyReference: unknown }; +}; + +/** Shared binding proof for single-source inspection and batched allocation fencing. */ +export function matchPreparedRemoteWorkspaceSource(bound: T, + workspace: Pick | null | undefined) { + const parsed = remoteWorkspaceRunReferenceSchema.safeParse(bound.run.reference); + if (!parsed.success || parsed.data.state !== "prepared" || bound.run.legacyReference != null) return null; const reference = parsed.data; const issueId = bound.run.issueId; if (typeof issueId !== "string" || bound.lease.issueId !== issueId || reference.leaseId !== bound.lease.id || reference.providerLeaseId !== bound.lease.providerLeaseId || reference.remoteCwd !== bound.lease.metadata?.remoteCwd || bound.lease.executionWorkspaceId && bound.lease.executionWorkspaceId !== reference.workspaceId) return null; - const [workspace] = await db.select().from(executionWorkspaces).where(and( - eq(executionWorkspaces.companyId, lease.companyId), eq(executionWorkspaces.id, reference.workspaceId))).limit(1); const authority = readRemoteWorkspaceAuthority(workspace?.metadata); - if (!workspace || workspace.sourceIssueId !== issueId && workspace.mode !== "shared_workspace" + if (!workspace || workspace.companyId !== bound.lease.companyId || workspace.id !== reference.workspaceId + || workspace.sourceIssueId !== issueId && workspace.mode !== "shared_workspace" || !authority || !authority.dirty || authority.writerRunId !== bound.run.id || authority.epoch !== reference.epoch || authority.snapshotId !== reference.snapshotId || !authority.location || authority.location.environmentId !== bound.lease.environmentId || authority.location.leaseId !== bound.lease.id || authority.location.providerLeaseId !== bound.lease.providerLeaseId || authority.location.remoteCwd !== reference.remoteCwd) return null; - return { ...bound, workspace: workspace!, reference, issueId }; + return { ...bound, workspace, reference, issueId }; } export async function inspectRemoteWorkspacePersistence(db: Db | Transaction, workspace: Workspace) { diff --git a/server/src/services/workspace-recovery-snapshot.ts b/server/src/services/workspace-recovery-snapshot.ts index 37d4f9030d5..819136d17c8 100644 --- a/server/src/services/workspace-recovery-snapshot.ts +++ b/server/src/services/workspace-recovery-snapshot.ts @@ -6,8 +6,8 @@ import os from "node:os"; import path from "node:path"; import { Readable } from "node:stream"; import { pipeline } from "node:stream/promises"; -import { and, desc, eq, sql } from "drizzle-orm"; -import { workspaceOperations, type Db } from "@paperclipai/db"; +import { and, desc, eq, inArray, sql } from "drizzle-orm"; +import { environmentLeases, workspaceOperations, type Db } from "@paperclipai/db"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import { createCommandManagedRuntimeClient } from "@paperclipai/adapter-utils/command-managed-runtime"; import { createTarballFromDirectory, exportSandboxWorkspaceSnapshot, restoreVerifiedSnapshotArchive } from "@paperclipai/adapter-utils/sandbox-managed-runtime"; @@ -33,6 +33,26 @@ export function readSnapshot(value: unknown, companyId: string, runId: string) { return s; } +export async function verifiedWorkspaceSnapshotReceipt(db: Db, input: { + companyId: string; issueId: string; runId: string; actionId: string; operationIds: string[]; + leaseId?: string; providerLeaseId?: string; +}) { + if (!input.operationIds.length) return null; + const [operation] = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata }) + .from(workspaceOperations).innerJoin(environmentLeases, and( + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.heartbeatRunId, input.runId), + sql`${environmentLeases.id}::text = ${workspaceOperations.metadata}->>'leaseId'`, + sql`${environmentLeases.providerLeaseId} = ${workspaceOperations.metadata}->>'providerLeaseId'`, + )).where(and(inArray(workspaceOperations.id, input.operationIds), + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.issueId, input.issueId), + eq(workspaceOperations.heartbeatRunId, input.runId), eq(workspaceOperations.phase, "workspace_recovery_snapshot"), + eq(workspaceOperations.status, "succeeded"), sql`${workspaceOperations.metadata}->>'recoveryActionId' = ${input.actionId}`, + input.leaseId ? eq(environmentLeases.id, input.leaseId) : undefined, + input.providerLeaseId ? eq(environmentLeases.providerLeaseId, input.providerLeaseId) : undefined, + )).limit(1); + return operation ? { id: operation.id, snapshot: readSnapshot(operation.metadata, input.companyId, input.runId) } : null; +} + async function digest(stream: AsyncIterable, limit = Number.MAX_SAFE_INTEGER) { const hash = createHash("sha256"); let byteSize = 0; diff --git a/server/src/services/workspace-recovery-source.ts b/server/src/services/workspace-recovery-source.ts index 87b27c4409d..b79eaac47e0 100644 --- a/server/src/services/workspace-recovery-source.ts +++ b/server/src/services/workspace-recovery-source.ts @@ -73,7 +73,7 @@ export async function ownedWorkspaceRecoveryLease(db: Db, input: WorkspaceRecove /** Resolve only the immutable source recorded by the failed run. Never discover * a candidate by recency, mutable adapter configuration, or a provider search. */ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRecoverySourceIdentity, - options: { maximumSources?: number } = {}) { + options: { maximumSources?: number; prepareNativeRetention?: boolean } = {}) { const invalid = workspaceSourceBindingFailure; const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))); const [action] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId), @@ -153,9 +153,12 @@ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRec if (!lease.environmentId) throw invalid("lease_environment_missing"); if (!lease.providerLeaseId) throw invalid("lease_allocation_missing"); if (!lease.provider || lease.provider === "local") throw invalid("lease_provider_invalid"); + const preparingRetention = options.prepareNativeRetention && run.runtimeMode === "native" && run.status === "failed" + && legacyDescriptor?.state === "prepared" && legacyDescriptor.resourceDisposition === "keep_running" + && lease.leasePolicy === "reuse_by_environment" && ["released", "retained"].includes(lease.status); const reusableExport = descriptor && (acceptedExportResult || remoteDescriptor && remoteAuthority) && ["reuse_by_environment", "reuse_by_execution_workspace"].includes(lease.leasePolicy); - if (!["retain_on_failure", "ephemeral"].includes(lease.leasePolicy) && !reusableExport) + if (!["retain_on_failure", "ephemeral"].includes(lease.leasePolicy) && !reusableExport && !preparingRetention) throw invalid("lease_policy_not_retained", lease.leasePolicy); if (typeof remoteCwd !== "string" || !remoteCwd.startsWith("/")) throw invalid("lease_cwd_invalid"); if (descriptor && lease.providerLeaseId !== descriptor.providerLeaseId) throw invalid("allocation_reference_mismatch"); diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index ae84b204504..a38c0881f22 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -37,7 +37,7 @@ import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js" import { getStorageService } from "../storage/index.js"; import * as storageBackend from "../storage/index.js"; import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; -import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; +import { attachPreservedWorkspaceSnapshots, verifiedWorkspaceSnapshotReceipt } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "./system-recovery-state.js"; import { requestTaskRetry } from "./task-retry.js"; @@ -62,7 +62,8 @@ import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import type { EnvironmentLease } from "@paperclipai/shared"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; -import { prepareNativeWorkspaceSync, resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js"; +import { nativeWorkspaceSyncStateRoot, prepareNativeWorkspaceSync, resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js"; +import { nativeWorkspaceExportPendingForResource } from "./native-runtime/native-workspace-finalization-state.js"; import { nativeWorkspaceFinalizationLockKey } from "./native-runtime/native-workspace-finalization-ownership.js"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { prepareStoppedSandboxForRecovery, prepareSandboxStopAndRetain, settleStopOnlyCleanup, readStopOnlyCleanup } from "./sandbox-stop-and-retain.js"; @@ -224,6 +225,205 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + async function failedWarmSource(exportOnly = false, hops = 0) { + const f = hops ? await warmHandoffSource(hops) : await seed("native", true, "keep_running"); + if (hops) await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment" }).where(eq(environmentLeases.id, f.lease.id)); + const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + const failed = failSystemRecovery(reserveSystemRecoveryAttempt(createSystemRecovery(f.runId, retry.requestId!))!, + new SystemRecoveryFailure("source_binding_invalid", "service_required"), { now: Date.now() - 86_400_000 }); + await db.update(issueRecoveryActions).set({ evidence: { ...(await action(f.actionId)).evidence, systemRecovery: failed, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "blocked", stage: "verifying_stop", + attempt: 1, attemptLimit: 6, requestIds: [retry.requestId!] } } }).where(eq(issueRecoveryActions.id, f.actionId)); + if (exportOnly) { + const contractId = randomUUID(), resultId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, + revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: {}, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ completionContractId: contractId, resultJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: contractId, schemaStatus: "accepted", serverFingerprint: resultId, canonicalSha256: resultId, + resultJson: { terminal: { runTerminalState: "succeeded" } } }); + await db.update(nativeRunFinalizations).set({ resultId }).where(eq(nativeRunFinalizations.runId, f.runId)); + const evidence = { ...(await action(f.actionId)).evidence }; + delete evidence.workspaceRestoreFailure; delete evidence.workspaceRestoreRecovery; + await db.update(issueRecoveryActions).set({ cause: "native_workspace_sync_out_retry_exhausted", evidence }).where(eq(issueRecoveryActions.id, f.actionId)); + } + return { ...f, retry, failed }; + } + + it.each([false, true])("recovers an exact warm native source for an already saved Retry without replaying its failed generation: exportOnly=%s", async exportOnly => { + const f = await failedWarmSource(exportOnly); + expect(f.runtime.resumeRunLease).not.toHaveBeenCalled(); + const resumed = vi.fn(); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, resumed) ?? []); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + const current = await action(f.actionId); + expect(current.evidence.nativeSourceRetention).toMatchObject({ schema: "paperclip.native-source-retention.v1", + sourceRunId: f.runId, requestId: f.retry.requestId, previousGeneration: f.failed }); + expect(readSystemRecovery(current.evidence)).toMatchObject({ state: "complete", intentId: f.retry.requestId }); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]) + .toMatchObject({ status: "released", leasePolicy: "reuse_by_environment", cleanupStatus: "success" }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + const snapshot = await action(f.actionId); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, resumed) ?? []); + expect(await action(f.actionId)).toEqual(snapshot); + expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce(); + const source = await run(f.runId); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, status: "running" }); + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + await Promise.all([service.resumeTaskRetry(f.companyId, f.retry.requestId!), service.resumeTaskRetry(f.companyId, f.retry.requestId!)]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + if (exportOnly) { + expect(successors).toHaveLength(0); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retry.requestId!)))[0]) + .toMatchObject({ status: "coalesced", runId: f.runId }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("workspace_finalizing"); + } else { + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ taskRetryRequestId: f.retry.requestId, previousRunId: f.runId }); + expect(await readFreshWorkspaceReadiness(db, successors[0])).toMatchObject({ mode: "resume_preserved_workspace" }); + } + }, 30_000); + + it.each(["throw", "skip"])("keeps completion and the saved Retry pending when policy settlement must %s", async failure => { + const f = await failedWarmSource(true), trigger = `native_retention_${randomUUID().replaceAll("-", "")}`; + await db.execute(sql.raw(`CREATE FUNCTION ${trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN + ${failure === "throw" ? "RAISE EXCEPTION 'test policy write failed';" : "RETURN NULL;"} END $$`)); + try { + await db.execute(sql.raw(`CREATE TRIGGER ${trigger} BEFORE UPDATE ON environment_leases FOR EACH ROW + WHEN (NEW.id = '${f.lease.id}'::uuid AND OLD.lease_policy = 'retain_on_failure' AND NEW.lease_policy = 'reuse_by_environment') + EXECUTE FUNCTION ${trigger}()`)); + await recoverRetainedWorkspace(f.input); + const held = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(held.state).toBe(failure === "throw" ? "scheduled" : "awaiting_evidence"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + expect(await resumePreservedResultFinalization(db, f.companyId, f.runId)).toBe(false); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].leasePolicy).toBe("retain_on_failure"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retry.requestId!)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + } finally { + await db.execute(sql.raw(`DROP TRIGGER IF EXISTS ${trigger} ON environment_leases`)); + await db.execute(sql.raw(`DROP FUNCTION ${trigger}()`)); + } + const pending = await action(f.actionId), state = readSystemRecovery(pending.evidence)!; + await db.update(issueRecoveryActions).set({ evidence: { ...pending.evidence, systemRecovery: { ...state, nextAttemptAt: 0 } } }) + .where(eq(issueRecoveryActions.id, f.actionId)); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].leasePolicy).toBe("reuse_by_environment"); + expect(await resumePreservedResultFinalization(db, f.companyId, f.runId)).toBe(true); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retry.requestId!)))[0]) + .toMatchObject({ status: "coalesced", runId: f.runId }); + }, 30_000); + + it.each([ + ["reuse_by_environment", false], ["reuse_by_environment", true], ["retain_on_failure", true], ["ephemeral", true], + ] as const)("settles snapshot-only source preservation without making its unfinished allocation reusable: policy=%s acceptedResult=%s", async (leasePolicy, acceptedResult) => { + const f = await failedWarmSource(acceptedResult); + if (leasePolicy !== "reuse_by_environment") { + await db.update(environmentLeases).set({ leasePolicy }).where(eq(environmentLeases.id, f.lease.id)); + const sourceAction = await action(f.actionId); + await db.update(issueRecoveryActions).set({ evidence: { ...sourceAction.evidence, + systemRecovery: createSystemRecovery(f.runId, f.retry.requestId!) } }).where(eq(issueRecoveryActions.id, f.actionId)); + } + targets.get(f.lease.id)!.runner!.syncIn = async () => ({ filesTransferred: 0, bytesTransferred: 0, operations: [] }); + targets.get(f.lease.id)!.runner!.syncOut = async operations => { + for (const operation of operations) for (const file of operation.files) + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; + }; + await rm(path.join(nativeWorkspaceSyncStateRoot(), f.runId), { recursive: true }); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, vi.fn()) ?? []); + expect(await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, f.runId), + eq(workspaceOperations.phase, "workspace_recovery_snapshot"), eq(workspaceOperations.status, "succeeded")))).toHaveLength(1); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + expect(lease).toMatchObject({ status: "released", leasePolicy: leasePolicy === "ephemeral" ? "ephemeral" : "retain_on_failure", cleanupStatus: "success" }); + if (leasePolicy === "reuse_by_environment") expect(lease.metadata?.nativeSourceRetention).toMatchObject({ settlement: "snapshot_retained" }); + else expect(lease.metadata?.nativeSourceRetention).toBeUndefined(); + expect(await nativeWorkspaceExportPendingForResource(db, lease, randomUUID())).toBe(true); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("terminal_failure"); + await recoverRetainedWorkspace(f.input); + expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce(); + const source = await run(f.runId); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, status: "running" }); + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + if (acceptedResult) { + const operationIds = readSystemRecovery((await action(f.actionId)).evidence)!.preservationOperationIds; + const receiptInput = { companyId: f.companyId, issueId: f.issueId, runId: f.runId, actionId: f.actionId, operationIds }; + expect(await verifiedWorkspaceSnapshotReceipt(db, receiptInput)).not.toBeNull(); + for (const patch of [{ companyId: randomUUID() }, { issueId: randomUUID() }, { runId: randomUUID() }, + { actionId: randomUUID() }, { operationIds: [randomUUID()] }]) + expect(await verifiedWorkspaceSnapshotReceipt(db, { ...receiptInput, ...patch })).toBeNull(); + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, operationIds[0])); + await db.update(workspaceOperations).set({ metadata: { ...operation.metadata, providerLeaseId: randomUUID() } }) + .where(eq(workspaceOperations.id, operation.id)); + try { + await service.resumeTaskRetry(f.companyId, f.retry.requestId!); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retry.requestId!)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + } finally { await db.update(workspaceOperations).set({ metadata: operation.metadata }).where(eq(workspaceOperations.id, operation.id)); } + } + await Promise.all([service.resumeTaskRetry(f.companyId, f.retry.requestId!), service.resumeTaskRetry(f.companyId, f.retry.requestId!)]); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(1); + const [successor] = successors; + expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "fresh_isolated_workspace", + retryIntentId: f.retry.requestId, sourceRunId: f.runId, retainedSourceIds: [f.lease.id] }); + const fresh = path.join(root, randomUUID()); await mkdir(fresh); + const attached = await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: f.issueId, + sourceRunIds: [successor.id], cwd: fresh }); + expect(attached).toHaveLength(1); + expect(await readFile(path.join(attached[0].path, "files", "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); + if (acceptedResult) expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)))[0]) + .toMatchObject({ schemaStatus: "accepted", resultJson: { terminal: { runTerminalState: "succeeded" } } }); + }, 30_000); + + it.each(["no_request", "unauthorized", "already_adopted", "prior_export", "other_failure", "wrong_cwd", "writer", "newer_terminal", "controller", "older_lease", "foreign_lease"])( + "does not adopt a held warm source with %s", async reason => { + const f = await failedWarmSource(); + if (reason === "no_request") await db.update(agentWakeupRequests).set({ status: "failed" }).where(eq(agentWakeupRequests.id, f.retry.requestId!)); + if (reason === "unauthorized") { + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retry.requestId!)); + await db.update(agentWakeupRequests).set({ payload: { ...wake.payload, taskRetry: { + ...(wake.payload!.taskRetry as object), actorSource: "session" } } }).where(eq(agentWakeupRequests.id, wake.id)); + } + if (["already_adopted", "prior_export", "other_failure"].includes(reason)) await db.update(issueRecoveryActions).set({ evidence: { + ...(await action(f.actionId)).evidence, + ...(reason === "already_adopted" ? { nativeSourceRetention: { schema: "paperclip.native-source-retention.v1" } } : {}), + ...(reason === "prior_export" ? { systemRecovery: { ...f.failed, preservationOperationIds: [randomUUID()] } } : {}), + ...(reason === "other_failure" ? { systemRecovery: { ...f.failed, failure: { ...f.failed.failure!, code: "source_unavailable" } } } : {}), + } }).where(eq(issueRecoveryActions.id, f.actionId)); + if (reason === "wrong_cwd") await db.update(environmentLeases).set({ metadata: { ...f.lease.metadata, remoteCwd: "/wrong" } }).where(eq(environmentLeases.id, f.lease.id)); + if (reason === "writer" || reason === "newer_terminal") await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: (await run(f.runId)).agentId, + issueId: f.issueId, nativeIssueId: f.issueId, status: reason === "writer" ? "queued" : "succeeded", runtimeMode: "native", + createdAt: new Date(Date.now() + 1000) }); + if (reason === "controller") await db.update(nativeRunFinalizations).set({ leaseOwner: "live-controller" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (reason === "older_lease" || reason === "foreign_lease") { + const companyId = reason === "foreign_lease" ? randomUUID() : f.companyId; + if (companyId !== f.companyId) await db.insert(companies).values({ id: companyId, name: "Other allocation owner", issuePrefix: `F${companyId.slice(0, 6)}` }); + await db.insert(environmentLeases).values({ companyId, provider: "daytona", providerLeaseId: f.providerLeaseId, + status: "released", leasePolicy: "reuse_by_environment" }); + } + const before = await action(f.actionId); + await recoverRetainedWorkspace(f.input); + expect(await action(f.actionId)).toEqual(before); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].leasePolicy).toBe("reuse_by_environment"); + expect(f.runtime.resumeRunLease).not.toHaveBeenCalled(); + expect(f.runtime.retryPendingSandboxTeardown).not.toHaveBeenCalled(); + }, + ); + it.each(["retained", "removed", "child", "expired-owner", "no-descriptor"])("preserves an old sandbox after storage migration without writing its old host (%s)", async mode => { vi.stubEnv("DATABASE_URL", connectionString); vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); @@ -1855,6 +2055,55 @@ describe("automatic retained workspace recovery", () => { if (original.home === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = original.home; } }); + it.each([{ hops: 1, snapshot: false }, { hops: 2, snapshot: false }, { hops: 2, snapshot: true }])( + "settles adopted warm source history: $hops handoffs, snapshot=$snapshot", async ({ hops, snapshot }) => { + const f = await failedWarmSource(false, hops); + if (snapshot) { + const runner = targets.get(f.lease.id)!.runner!; + runner.syncIn = async () => ({ filesTransferred: 0, bytesTransferred: 0, operations: [] }); + runner.syncOut = async operations => { + for (const operation of operations) for (const file of operation.files) + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; + }; + await rm(path.join(nativeWorkspaceSyncStateRoot(), f.runId), { recursive: true }); + } + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + expect(lease).toMatchObject({ status: "released", cleanupStatus: "success", + leasePolicy: snapshot ? "retain_on_failure" : "reuse_by_environment" }); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(!snapshot); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.retry.requestId!)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce(); + if (snapshot) { + const state = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(await verifiedWorkspaceSnapshotReceipt(db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, actionId: f.actionId, operationIds: state.preservationOperationIds })).not.toBeNull(); + expect(lease.metadata?.nativeSourceRetention).toMatchObject({ settlement: "snapshot_retained" }); + } else expect(await readFile(path.join(f.host, "after-warm-stop.txt"), "utf8")).toBe("stopped turn work\n"); + }, 30_000); + + it("keeps adopted warm history held when a predecessor changes after export", async () => { + const f = await failedWarmSource(false, 1); + const history = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, f.companyId), + eq(environmentLeases.providerLeaseId, f.providerLeaseId))); + const predecessor = history.find(lease => lease.id !== f.lease.id)!; + const stop = f.runtime.retryPendingSandboxTeardown.getMockImplementation()!; + f.runtime.retryPendingSandboxTeardown.mockImplementationOnce(async input => { + const receipt = await stop(input); + await db.update(environmentLeases).set({ status: "active" }).where(eq(environmentLeases.id, predecessor.id)); + return receipt; + }); + await recoverRetainedWorkspace(f.input); + const state = readSystemRecovery((await action(f.actionId)).evidence)!; + expect(state.state).not.toBe("complete"); + expect(state.failure?.code).toBe("allocation_ownership_ambiguous"); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].leasePolicy).toBe("retain_on_failure"); + expect(await readFile(path.join(f.host, "after-warm-stop.txt"), "utf8")).toBe("stopped turn work\n"); + }, 30_000); + it.each([1, 2])("recovers a stopped native source after %s verified warm predecessor handoffs", async hops => { const f = await warmHandoffSource(hops); expect((await run(f.priorRunId)).runnerProfileJson?.nativeWorkspaceSync).toMatchObject({ state: "finalized", leaseId: f.priorLeaseId }); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index dc8de2e9f1a..cf0cc0f1de5 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -31,6 +31,7 @@ import { SYSTEM_RECOVERY_MAX_RETRIES, createSystemRecovery, failSystemRecovery, import { inspectWorkspaceRecoverySource, workspaceRecoverySourceEligible, workspaceSourceBindingFailure } from "./workspace-recovery-source.js"; import { preserveWorkspaceSnapshot } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries } from "./workspace-recovery-backfill.js"; +import { canAdoptNativeSourceRetention, retainNativeRecoverySource, settleRecoveredNativeSourceRetention } from "./native-runtime/native-source-retention.js"; import { saveRemoteWorkspace } from "./remote-workspaces.js"; const exportCauses = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]; @@ -156,11 +157,15 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ const { db } = input; await adoptSystemRecovery(db, input); const initial = await loadSystemRecovery(db, input); - if (!initial || initial.state === "complete" || initial.state === "exhausted" || + if (!initial || initial.state === "complete" || (initial.state === "exhausted" && !canAdoptNativeSourceRetention(initial)) || (initial.nextAttemptAt !== null && initial.nextAttemptAt > Date.now())) return; const runAttempt = async (ownership: { token: string; assertHeld(): Promise }) => { let current = await loadSystemRecovery(db, input); if (!current) return; + if (current.state === "exhausted") { + current = await retainNativeRecoverySource(db, input, current, ownership); + if (!current) return; + } const reserved = reserveSystemRecoveryAttempt(current); if (!reserved) { if (current.state !== "exhausted" && current.deadlineAt !== null && Date.now() >= current.deadlineAt) @@ -358,6 +363,8 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ metadata: { owningService: "workspace_recovery", preservationOperationIds: next.preservationOperationIds } }); if (!await updateSystemRecovery(tx as unknown as Db, input, current!, next)) throw new Error("workspace_recovery_generation_changed"); + if (inspected.action.evidence.nativeSourceRetention && !await settleRecoveredNativeSourceRetention(tx as unknown as Db, input, next.preservationOperationIds)) + throw new SystemRecoveryFailure("effect_unresolved", "recheck_evidence"); return persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery", action: "issue.workspace_recovered", entityType: "issue", entityId: input.issueId, runId: input.runId, details: { recoveryActionId: input.actionId, summary: "Saved work recovered." } });