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." } });