diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index bfe21aaa54b..ec70df8e56d 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -106,6 +106,25 @@ returns `409` without reopening work. A duplicate queued request is idempotent. An export that is still unsafe creates another explicit repair hold. Generic recovery's **Retry source task** does not substitute for export-only retry. +Before resuming the retained sandbox, the controller durably records a stop-only +cleanup intent on that exact lease and removes the old stopped receipt. If the +resume reply is lost, probing fails, or admission changes, it stops and retains +the sandbox. A failed stop remains `pending_cleanup`; a restarted controller's +bounded cleanup sweep retries the verified stop without destroying saved files. +An in-flight request holds a 15-minute cleanup claim, so a controller crash may +delay that sweep until the claim expires. Neither an unconfirmed stop nor a +stale receipt grants admission, and a changed lease or competing sandbox owner +prevents cleanup from taking ownership. Retry export after the lease has a new +confirmed stopped receipt. No provider turn is created by this recovery. + +The opt-in `native-workspace-export-resume.live.test.ts` is a provider-boundary +fault integration, separate from the browser Product E2E. After building the +Daytona plugin, run that exact Vitest file with `PAPERCLIP_LIVE_EXPORT_RESUME=1`, +`DAYTONA_API_KEY`, and `PAPERCLIP_LIVE_EXPORT_RESUME_IMAGE` set to an immutable +image digest. It creates one disposable sandbox and database, injects probe and +stop transport failures, and verifies a fresh runtime can stop the sandbox while +preserving exact nonce bytes. It deletes only that owned fixture after proof. + The reconciliation sweep also restores a repair notice that an older generic sweeper incorrectly resolved as `new_source_execution_path`, but only for the current blocked task with its accepted result and exact stopped lease. It does diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 6a1dc8b833a..35cc856b74f 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -5037,6 +5037,46 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(await db.select().from(environmentLeases)).toHaveLength(1); }); + it.each(["stopped", "unconfirmed", "foreign_marker", "competing_owner", "live_request", "late_receipt"])("recovers a pending export resume with stop-only cleanup after restart: %s", async outcome => { + const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); + const lease = seeded.reusableLease; + const requestId = randomUUID(); + await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", metadata: { + ...lease.metadata, + nativeWorkspaceExportResume: { schema: "paperclip.workspace-export-resume.v1", requestId, companyId: seeded.companyId, + runId: outcome === "foreign_marker" ? randomUUID() : lease.heartbeatRunId, leaseId: lease.id, + provider: lease.provider, providerLeaseId: lease.providerLeaseId, resultId: randomUUID() }, + pendingCleanupAttemptId: requestId, pendingCleanupInFlight: true, + pendingCleanupLeaseExpiresAtMs: Date.now() + (outcome === "live_request" ? 60_000 : -1), + } }).where(eq(environmentLeases.id, lease.id)); + if (outcome === "competing_owner") await db.insert(environmentLeases).values({ companyId: seeded.companyId, + environmentId: seeded.environment.id, status: "active", provider: lease.provider, providerLeaseId: lease.providerLeaseId }); + const call = vi.fn(async (_id: string, method: string) => { + if (method !== "environmentReleaseLease") throw new Error("Saved workspace must never be destroyed"); + if (outcome === "late_receipt") await db.update(environmentLeases).set({ status: "active", heartbeatRunId: null, + cleanupStatus: null, metadata: { newOwner: true } }).where(eq(environmentLeases.id, lease.id)); + return outcome === "unconfirmed" ? undefined : { providerLeaseId: lease.providerLeaseId, state: "stopped" }; + }); + const workerManager = { isRunning: () => true, call, + getWorker: () => ({ supportedMethods: ["environmentReleaseLease", "environmentDestroyLease"] }), + } as unknown as PluginWorkerManager; + const restarted = environmentRuntimeService(db, { pluginWorkerManager: workerManager }); + await heartbeatService(db, { environmentRuntime: restarted }).sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(call.mock.calls.some((entry) => entry[1] === "environmentDestroyLease")).toBe(false); + const persisted = (await environmentService(db).getLeaseById(lease.id))!; + if (outcome === "stopped") { + expect(call).toHaveBeenCalledWith(seeded.pluginId, "environmentReleaseLease", expect.objectContaining({ providerLeaseId: lease.providerLeaseId, cancelActiveWork: true }), expect.any(Number)); + expect(persisted).toMatchObject({ status: "expired", cleanupStatus: "success", metadata: { remoteExecutionTermination: { state: "stopped", runId: lease.heartbeatRunId } } }); + expect(persisted.metadata?.nativeWorkspaceExportResume).toBeUndefined(); + } else if (outcome === "late_receipt") { + expect(persisted).toMatchObject({ status: "active", heartbeatRunId: null, cleanupStatus: null, metadata: { newOwner: true } }); + } else { + expect(persisted).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed" }); + expect(persisted.metadata?.remoteExecutionTermination).toBeUndefined(); + if (["foreign_marker", "competing_owner", "live_request"].includes(outcome)) expect(call).not.toHaveBeenCalled(); + } + }); + it("does not allocate a native-runner replacement without a verified backup", async () => { const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); const workerManager = { diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index da961aa6f65..3e3ca76e1f7 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,7 +1,8 @@ import { readEnvironmentCreationCleanupError } from "@paperclipai/plugin-sdk"; import { remoteTerminationReceipt } from "./remote-execution-termination.js"; +import { hasNativeWorkspaceExportResume, readNativeWorkspaceExportResume } from "./native-runtime/native-workspace-export-resume.js"; import { createHash, randomUUID } from "node:crypto"; -import { and, eq, inArray, sql } from "drizzle-orm"; +import { and, eq, inArray, ne, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { companySecrets, companySecretVersions, environmentLeases, heartbeatRuns } from "@paperclipai/db"; import type { @@ -2564,6 +2565,22 @@ function createSandboxEnvironmentDriver( }, async retryPendingSandboxTeardown(input) { + const exportResume = hasNativeWorkspaceExportResume(input.lease); + const resumeIntent = readNativeWorkspaceExportResume(input.lease); + const assertExportResumeOwnership = async () => { + if (!resumeIntent) throw new Error("Workspace export resume ownership is invalid."); + const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, input.lease.id)).limit(1); + const [otherOwner] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + ne(environmentLeases.id, input.lease.id), eq(environmentLeases.provider, input.lease.provider!), + eq(environmentLeases.providerLeaseId, input.lease.providerLeaseId!), inArray(environmentLeases.status, ["active", "pending_cleanup"]), + )).limit(1); + if (!current || current.status !== "pending_cleanup" || current.companyId !== input.lease.companyId + || current.metadata?.pendingCleanupAttemptId !== input.lease.metadata?.pendingCleanupAttemptId + || readNativeWorkspaceExportResume(current)?.requestId !== resumeIntent.requestId || otherOwner) { + throw new Error("Workspace export resume ownership changed before cleanup."); + } + }; + if (exportResume) await assertExportResumeOwnership(); // Resolve the teardown from the immutable orphan lease row, not from the // current environment. The row keeps the provider, the provider lease id, // and the sandbox config in its metadata. A provider change re-points the @@ -2614,6 +2631,23 @@ function createSandboxEnvironmentDriver( { issueId: input.lease.issueId, heartbeatRunId: input.lease.heartbeatRunId }, ); const workerConfig = stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig); + if (exportResume) { + const pluginId = pluginProvider.resolved.plugin.id; + if (!pluginWorkerVerifiesLifecycleMethod(pluginId, "environmentReleaseLease")) { + throw new Error("Workspace export recovery requires verified stop-only cleanup."); + } + await assertExportResumeOwnership(); + const receipt = await pluginWorkerManager.call(pluginId, "environmentReleaseLease", { + driverKey: recordedProvider, companyId: input.lease.companyId, + environmentId: input.lease.environmentId ?? "", issueId: input.lease.issueId, + config: workerConfig, providerLeaseId: input.lease.providerLeaseId, + leaseMetadata: input.lease.metadata ?? {}, cancelActiveWork: true, + }, Math.min(resolvePluginSandboxRpcTimeoutMs(workerConfig) ?? 60_000, 60_000)); + if (remoteTerminationReceipt(input.lease, receipt)?.state !== "stopped") { + throw new Error("Workspace export recovery did not confirm the retained sandbox stopped."); + } + return receipt; + } const failedCreation = readEnvironmentCreationCleanupError({ data: { schema: "paperclip/environment-creation-cleanup/v1", cleanup: input.lease.metadata?.failedCreateCleanup, @@ -2649,6 +2683,7 @@ function createSandboxEnvironmentDriver( // durable orphan record, not the environment binding, for the same reason // as the plugin path above. The teardown targets the recorded provider, // never the current environment provider. + if (exportResume) throw new Error("Workspace export recovery requires verified stop-only cleanup."); const cleanupConfig = await resolveSandboxCleanupConfigSecrets( db, input.lease.companyId, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 3c03788c0d9..88abaef7bbd 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -5,6 +5,7 @@ import { NativeWorkspaceFinalizationOwnershipLostError, type NativeWorkspaceFinalizationOwnership, } from "./native-runtime/native-workspace-finalization-ownership.js"; +import { hasNativeWorkspaceExportResume, settleNativeWorkspaceExportResume } from "./native-runtime/native-workspace-export-resume.js"; import { applyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-result"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; import { externalConversationStateSql, nonIdleSlackIssueCondition } from "./slack-conversation-state.js"; @@ -18580,8 +18581,10 @@ export function heartbeatService( // environment still exists tears down through `destroyRunLease`. That // path uses the provider and configuration recorded on the lease first; // the environment is lifecycle context and only a legacy fallback. + // An interrupted export resume also uses recorded cleanup, but its tagged + // intent permits verified stop-and-retain only, never sandbox destruction. const isOrphanEphemeralLease = lease.leasePolicy === "ephemeral"; - const useRecordedTeardown = isOrphanEphemeralLease || !environment; + const useRecordedTeardown = isOrphanEphemeralLease || !environment || hasNativeWorkspaceExportResume(lease); // Do not consume a finite cleanup attempt while the provider plugin is // briefly unavailable. A plugin worker restart, a plugin reload, or a @@ -18650,7 +18653,9 @@ export function heartbeatService( environment, lease, }); - const released = await environmentsSvc.releaseLease(lease.id, "expired", { + const released = hasNativeWorkspaceExportResume(lease) + ? await settleNativeWorkspaceExportResume(db, lease, { attemptId: claimed, receipt, status: "expired" }) + : await environmentsSvc.releaseLease(lease.id, "expired", { expectedPendingCleanupAttemptId: claimed, cleanupStatus: "success", failureReason: "pending_cleanup_retry", diff --git a/server/src/services/native-runtime/native-workspace-export-resume.live.test.ts b/server/src/services/native-runtime/native-workspace-export-resume.live.test.ts new file mode 100644 index 00000000000..ab232ed9b58 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-export-resume.live.test.ts @@ -0,0 +1,158 @@ +// Opt-in provider-boundary fault integration, not a Product E2E workflow. +// Real Daytona acquire/resume/stop/execute + production retry/reaper services; +// only the probe transport and first stop reply are deliberately unavailable. +import { randomUUID, createHash } from "node:crypto"; +import { writeFile } from "node:fs/promises"; +import { createRequire } from "node:module"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { eq } from "drizzle-orm"; +import { agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, agentWakeupRequests, plugins } from "@paperclipai/db"; +import type { Environment, EnvironmentLease } from "@paperclipai/shared"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { environmentRuntimeService } from "../environment-runtime.js"; +import type { PluginWorkerManager } from "../plugin-worker-manager.js"; +import { heartbeatService } from "../heartbeat.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { retryNativeWorkspaceExport } from "./native-workspace-export-retry.js"; + +const enabled = process.env.PAPERCLIP_LIVE_EXPORT_RESUME === "1"; +const describeLive = enabled ? describe : describe.skip; +const image = process.env.PAPERCLIP_LIVE_EXPORT_RESUME_IMAGE ?? ""; +const quote = (s: string) => `'${s.replaceAll("'", "'\\''")}'`; + +describeLive("live Daytona export-resume failure recovery", () => { + let temporary: Awaited>; + let db: ReturnType; + let plugin: any; + let manifest: any; + let sandbox: any; + let lease: EnvironmentLease; + let environment: Environment; + const ids = { company: randomUUID(), agent: randomUUID(), environment: randomUUID(), run: randomUUID(), issue: randomUUID(), + result: randomUUID(), contract: randomUUID(), lease: randomUUID(), action: randomUUID(), plugin: randomUUID() }; + const nonce = randomUUID(); + const bytes = `preserved-export-resume-${nonce}\n`; + const digest = createHash("sha256").update(bytes).digest("hex"); + const config = { provider: "daytona", image, reuseLease: true, timeoutMs: 120_000, autoStopInterval: 10, autoDeleteInterval: 30 }; + const evidence: Record = { kind: "live_provider_boundary_fault_integration", image, nonceSha256: digest }; + const methods: string[] = []; + let failProbe = false; + let failStop = false; + const handlers: Record = { environmentResumeLease: "onEnvironmentResumeLease", environmentReleaseLease: "onEnvironmentReleaseLease", environmentExecute: "onEnvironmentExecute" }; + const manager = () => ({ isRunning: () => true, + getWorker: () => ({ supportedMethods: Object.keys(handlers) }), + call: async (_id: string, method: string, input: any) => { + methods.push(method); + if (method === "environmentExecute" && failProbe) { failProbe = false; throw new Error("Injected probe transport failure"); } + if (method === "environmentReleaseLease" && failStop) { failStop = false; throw new Error("Injected stop transport failure"); } + if (!handlers[method]) throw new Error("Unexpected provider operation"); + try { return await plugin[handlers[method]](input); } + catch { throw new Error("Live Daytona operation failed; provider details withheld"); } + }, + }) as unknown as PluginWorkerManager; + const scope = () => ({ driverKey: "daytona", companyId: ids.company, environmentId: ids.environment, issueId: ids.issue, + config, providerLeaseId: lease.providerLeaseId, leaseMetadata: lease.metadata }); + async function persistedLease() { + return (await db.select().from(environmentLeases).where(eq(environmentLeases.id, ids.lease)))[0]; + } + async function confirmStoppedAndPreserved() { + await sandbox.refreshData(); + expect(sandbox.state).toBe("stopped"); + await sandbox.start(60); + const result = await sandbox.process.executeCommand(`sha256sum ${quote(`${lease.metadata!.remoteCwd}/preserved-${nonce}.txt`)}`, "/", undefined, 15); + expect(result.exitCode).toBe(0); + expect(result.result.split(/\s/)[0]).toBe(digest); + await sandbox.stop(60); + await sandbox.refreshData(); + expect(sandbox.state).toBe("stopped"); + } + beforeAll(async () => { + if (!process.env.DAYTONA_API_KEY || !/@sha256:[a-f0-9]{64}$/.test(image)) throw new Error("Explicit Daytona credentials and immutable image required"); + temporary = await startEmbeddedPostgresTestDatabase("live-export-resume-"); db = createDb(temporary.connectionString); + ({ default: plugin } = await import(new URL("../../../../packages/plugins/sandbox-providers/daytona/dist/plugin.js", import.meta.url).href)); + plugin = plugin.definition; + ({ default: manifest } = await import(new URL("../../../../packages/plugins/sandbox-providers/daytona/dist/manifest.js", import.meta.url).href)); + await db.insert(companies).values({ id: ids.company, name: "Disposable export lifecycle", issuePrefix: "LXR" }); + await db.insert(agents).values({ id: ids.agent, companyId: ids.company, name: "No provider turn", adapterType: "paperclip_runner" }); + await db.insert(environments).values({ id: ids.environment, name: `Lifecycle ${nonce}`, driver: "sandbox", config }); + environment = (await db.select().from(environments).where(eq(environments.id, ids.environment)))[0] as unknown as Environment; + await db.insert(plugins).values({ id: ids.plugin, pluginKey: manifest.id, packageName: "@paperclipai/plugin-daytona", version: manifest.version, + apiVersion: 1, categories: ["automation"], manifestJson: manifest, status: "ready", installOrder: 1 }); + await db.insert(issues).values({ id: ids.issue, companyId: ids.company, title: "Preserved accepted result", status: "blocked", assigneeAgentId: ids.agent }); + await db.insert(completionContracts).values({ id: ids.contract, companyId: ids.company, issueId: ids.issue, revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "live-test", risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { objective: "Preserve saved work" }, canonicalSha256: digest, createdByActorType: "system", createdByActorId: "live-test" }); + await db.insert(heartbeatRuns).values({ id: ids.run, companyId: ids.company, agentId: ids.agent, status: "failed", runtimeMode: "native", nativeIssueId: ids.issue, nativePhase: "terminal_failure", completionContractId: ids.contract }); + await db.insert(nativeRunResults).values({ id: ids.result, companyId: ids.company, issueId: ids.issue, runId: ids.run, completionContractId: ids.contract, serverFingerprint: digest, schemaStatus: "accepted", resultJson: { work: "preserve saved work" }, canonicalSha256: digest }); + await db.insert(nativeRunFinalizations).values({ runId: ids.run, companyId: ids.company, issueId: ids.issue, phase: "terminal_failure", resultId: ids.result, failureCode: "native_workspace_sync_out_unsafe_archive" }); + await db.insert(issueRecoveryActions).values({ id: ids.action, companyId: ids.company, sourceIssueId: ids.issue, kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: ids.agent, cause: "native_workspace_sync_out_unsafe_archive", fingerprint: ids.run, evidence: { runId: ids.run }, nextAction: "Repair saved files" }); + let setupPhase = "acquire"; + try { + const acquired = await plugin.onEnvironmentAcquireLease({ driverKey: "daytona", companyId: ids.company, environmentId: ids.environment, + issueId: ids.issue, runId: ids.run, agentId: ids.agent, adapterType: "paperclip_runner", config }); + setupPhase = "sdk_lookup"; + const require = createRequire(new URL("../../../../packages/plugins/sandbox-providers/daytona/package.json", import.meta.url)); + const { Daytona } = require("@daytonaio/sdk"); + sandbox = await new Daytona({ apiKey: process.env.DAYTONA_API_KEY }).get(acquired.providerLeaseId); + const metadata = { ...acquired.metadata, provider: "daytona", pluginId: ids.plugin, sandboxProviderPlugin: true }; + await db.insert(environmentLeases).values({ id: ids.lease, companyId: ids.company, environmentId: ids.environment, issueId: ids.issue, + heartbeatRunId: ids.run, status: "active", leasePolicy: "reuse_by_environment", provider: "daytona", providerLeaseId: acquired.providerLeaseId, metadata }); + lease = await persistedLease() as unknown as EnvironmentLease; + setupPhase = "write"; + const wrote = await sandbox.process.executeCommand(`printf %s ${quote(bytes)} > ${quote(`${metadata.remoteCwd}/preserved-${nonce}.txt`)}`, "/", undefined, 15); + if (wrote.exitCode !== 0) throw new Error("Write failed"); + setupPhase = "initial_stop"; + const receipt = await plugin.onEnvironmentReleaseLease({ ...scope(), cancelActiveWork: true }); + expect(receipt.state).toBe("stopped"); + await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(), + metadata: { ...metadata, remoteExecutionTermination: remoteTerminationReceipt(lease, receipt) } }).where(eq(environmentLeases.id, ids.lease)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: randomUUID(), leaseId: ids.lease, providerLeaseId: lease.providerLeaseId, remoteCwd: metadata.remoteCwd, resourceDisposition: "keep_running" } } }).where(eq(heartbeatRuns.id, ids.run)); + } catch (error) { + evidence.setupFailure = { phase: setupPhase, errorType: (error as Error).name }; + throw new Error(`Live fixture setup failed at ${setupPhase}; provider details withheld`); + } + }, 180_000); + afterAll(async () => { + try { + if (sandbox) { + if (sandbox.labels?.["paperclip-company-id"] !== ids.company || sandbox.labels?.["paperclip-environment-id"] !== ids.environment || sandbox.labels?.["paperclip-run-id"] !== ids.run) throw new Error("Fixture cleanup ownership mismatch"); + await sandbox.delete(60); + evidence.cleanupPassed = true; + } + } catch { throw new Error("Exact live fixture cleanup failed; provider details withheld"); } + finally { + if (temporary) await temporary.cleanup(); + if (process.env.PAPERCLIP_EXPORT_RESUME_EVIDENCE_PATH) await writeFile(process.env.PAPERCLIP_EXPORT_RESUME_EVIDENCE_PATH, JSON.stringify(evidence, null, 2), { mode: 0o600 }); + } + }, 90_000); + it("stops after a failed probe, and recovers a failed stop through a fresh runtime", async () => { + const request = () => ({ db, companyId: ids.company, issueId: ids.issue, actionId: ids.action, runId: ids.run, actorId: "live-test", + repairNote: "Provider-boundary fault integration preserves exact synthetic nonce bytes", environmentRuntime: environmentRuntimeService(db, { pluginWorkerManager: manager() }) }); + failProbe = true; + await expect(retryNativeWorkspaceExport(request())).rejects.toThrow("Resume and repair"); + expect(await persistedLease()).toMatchObject({ status: "released", cleanupStatus: "success", metadata: { remoteExecutionTermination: { state: "stopped" } } }); + await confirmStoppedAndPreserved(); + evidence.failedProbeCompensated = true; + failProbe = true; failStop = true; + await expect(retryNativeWorkspaceExport(request())).rejects.toThrow("Resume and repair"); + const pending = await persistedLease(); + expect(pending).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed", metadata: { pendingCleanupInFlight: false, nativeWorkspaceExportResume: { runId: ids.run, leaseId: ids.lease } } }); + expect(pending.metadata?.remoteExecutionTermination).toBeUndefined(); + await sandbox.refreshData(); expect(sandbox.state).toBe("started"); + evidence.failedStopDurablyTracked = true; + // Load a fresh worker module as well as a fresh controller service. The + // reaper must recover from durable rows with no cached provider handle. + const restartedPlugin = (await import(`${new URL("../../../../packages/plugins/sandbox-providers/daytona/dist/plugin.js", import.meta.url).href}?restart=${nonce}`)).default.definition; + expect(restartedPlugin).not.toBe(plugin); + plugin = restartedPlugin; + const fresh = environmentRuntimeService(db, { pluginWorkerManager: manager() }); + await heartbeatService(db, { environmentRuntime: fresh }).sweepPendingCleanupLeases({ backoffMs: 0 }); + expect(await persistedLease()).toMatchObject({ status: "expired", cleanupStatus: "success", metadata: { remoteExecutionTermination: { state: "stopped" } } }); + expect((await persistedLease()).metadata?.nativeWorkspaceExportResume).toBeUndefined(); + await confirmStoppedAndPreserved(); + expect(await db.select().from(agentWakeupRequests)).toHaveLength(0); + expect(await db.select().from(heartbeatRuns)).toHaveLength(1); + expect((await db.select().from(nativeRunFinalizations))[0]).toMatchObject({ phase: "terminal_failure", resultId: ids.result }); + expect(methods.filter(m => m === "environmentResumeLease")).toHaveLength(2); + expect(methods.filter(m => m === "environmentReleaseLease")).toHaveLength(3); + evidence.restartStopPreservedBytes = true; evidence.noNewProviderTurn = true; evidence.noDestroyBeforeFixtureCleanup = true; + }, 180_000); +}); diff --git a/server/src/services/native-runtime/native-workspace-export-resume.ts b/server/src/services/native-runtime/native-workspace-export-resume.ts new file mode 100644 index 00000000000..e225fc65eba --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-export-resume.ts @@ -0,0 +1,56 @@ +import { and, eq, sql } from "drizzle-orm"; +import { environmentLeases, type Db } from "@paperclipai/db"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; + +type Lease = { + id: string; companyId: string; heartbeatRunId: string | null; + provider: string | null; providerLeaseId: string | null; + metadata: Record | null; +}; + +export const NATIVE_WORKSPACE_EXPORT_RESUME_KEY = "nativeWorkspaceExportResume"; + +export function hasNativeWorkspaceExportResume(lease: Pick): boolean { + return Object.prototype.hasOwnProperty.call(lease.metadata ?? {}, NATIVE_WORKSPACE_EXPORT_RESUME_KEY); +} + +/** This intent is written before resuming a retained sandbox. Recovery may stop + * this exact allocation, but must never fall through to destructive cleanup. */ +export function readNativeWorkspaceExportResume(lease: Lease) { + const value = lease.metadata?.[NATIVE_WORKSPACE_EXPORT_RESUME_KEY]; + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const marker = value as Record; + if (marker.schema !== "paperclip.workspace-export-resume.v1" + || marker.companyId !== lease.companyId || marker.runId !== lease.heartbeatRunId || !lease.heartbeatRunId + || marker.leaseId !== lease.id || marker.provider !== lease.provider || !lease.provider + || marker.providerLeaseId !== lease.providerLeaseId || !lease.providerLeaseId + || typeof marker.requestId !== "string" || !marker.requestId + || typeof marker.resultId !== "string" || !marker.resultId) return null; + return marker; +} + +/** A late cleanup receipt cannot rewrite a rebound lease or a newer attempt. */ +export async function settleNativeWorkspaceExportResume(db: Db, lease: Lease, options: { + attemptId: string; receipt?: unknown; status?: "released" | "expired"; +}) { + const marker = readNativeWorkspaceExportResume(lease); + if (!marker) return null; + const receipt = remoteTerminationReceipt(lease, options.receipt); + const stopped = receipt?.state === "stopped"; + const now = new Date(); + const [row] = await db.update(environmentLeases).set({ + status: stopped ? options.status ?? "released" : "pending_cleanup", cleanupStatus: stopped ? "success" : "failed", + releasedAt: now, lastUsedAt: now, updatedAt: now, + metadata: sql`(${environmentLeases.metadata} - 'remoteExecutionTermination' + ${stopped ? sql`- 'nativeWorkspaceExportResume'` : sql``}) || ${JSON.stringify({ + ...(stopped ? { remoteExecutionTermination: receipt } : {}), + pendingCleanupInFlight: false, pendingCleanupLeaseExpiresAtMs: 0, + })}::jsonb`, + }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId), + eq(environmentLeases.heartbeatRunId, lease.heartbeatRunId!), eq(environmentLeases.provider, lease.provider!), + eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), eq(environmentLeases.status, "pending_cleanup"), + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.attemptId}`, + sql`${environmentLeases.metadata}->'nativeWorkspaceExportResume'->>'requestId' = ${marker.requestId}`, + )).returning(); + return row ?? null; +} diff --git a/server/src/services/native-runtime/native-workspace-export-retry.test.ts b/server/src/services/native-runtime/native-workspace-export-retry.test.ts index 8fc16d7e1db..b771f648676 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.test.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.test.ts @@ -35,7 +35,10 @@ describe("board retry of accepted workspace export", () => { const identity = { id: leaseId, companyId, heartbeatRunId: runId, provider: "daytona", providerLeaseId }; await db.insert(environmentLeases).values({ ...identity, environmentId, issueId, status: "released", releasedAt: new Date(), cleanupStatus: "success", leasePolicy: "reuse_by_environment", metadata: { remoteExecutionTermination: remoteTerminationReceipt(identity, { providerLeaseId, state: "stopped" }) } }); await db.insert(issueRecoveryActions).values({ id: actionId, companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId, cause: "native_workspace_sync_out_unsafe_archive", fingerprint: runId, evidence: { runId }, nextAction: "Repair saved files" }); - const request = { db, companyId, issueId, actionId, runId, actorId: "board", repairNote: "Removed only the known unsafe fixture link and preserved the saved work.", environmentRuntime: { resumeRunLease: vi.fn().mockResolvedValue({ providerLeaseId, metadata: { remoteCwd: "/work" } }) } as never }; + const request = { db, companyId, issueId, actionId, runId, actorId: "board", repairNote: "Removed only the known unsafe fixture link and preserved the saved work.", environmentRuntime: { + resumeRunLease: vi.fn().mockResolvedValue({ providerLeaseId, metadata: { remoteCwd: "/work" } }), + retryPendingSandboxTeardown: vi.fn().mockResolvedValue({ providerLeaseId, state: "stopped" }), + } as never }; return { ...request, request, resultId, leaseId }; } it("queues only the existing result and lease, audits once, and deduplicates a pending click", async () => { @@ -62,6 +65,75 @@ describe("board retry of accepted workspace export", () => { expect(resume).toHaveBeenCalledOnce(); expect(resume).toHaveBeenCalledWith(expect.objectContaining({ lease: expect.objectContaining({ id: f.leaseId, providerLeaseId: lease.providerLeaseId }) })); }); + it("records a recoverable stop-only intent before the provider can resume", async () => { + const f = await seed(); + const runtime = f.request.environmentRuntime as { resumeRunLease: ReturnType; retryPendingSandboxTeardown: ReturnType }; + runtime.resumeRunLease.mockImplementation(async () => { + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + expect(lease).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed", metadata: { + nativeWorkspaceExportResume: { runId: f.runId, leaseId: f.leaseId, resultId: f.resultId }, + pendingCleanupInFlight: true, + } }); + expect(lease.metadata?.remoteExecutionTermination).toBeUndefined(); + return { providerLeaseId: lease.providerLeaseId, metadata: { remoteCwd: "/work" } }; + }); + await expect(retryNativeWorkspaceExport(f.request)).resolves.toMatchObject({ status: "queued" }); + expect(runtime.retryPendingSandboxTeardown).not.toHaveBeenCalled(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + expect(lease.status).toBe("active"); + expect(lease.metadata?.nativeWorkspaceExportResume).toBeUndefined(); + }); + it.each(["resume_reply_lost", "probe_failed", "admission_changed", "cancelled", "completed", "stop_failed"])("tracks and compensates a failed export resume: %s", async kind => { + const f = await seed(); + const runtime = f.request.environmentRuntime as { resumeRunLease: ReturnType; retryPendingSandboxTeardown: ReturnType }; + if (kind === "resume_reply_lost") runtime.resumeRunLease.mockRejectedValueOnce(new Error("reply lost after provider resumed")); + else probe.mockImplementationOnce(async () => { + if (kind === "admission_changed") { + await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: f.issueId, status: "failed", createdAt: new Date(Date.now() + 1000) }); + return { exitCode: 0, timedOut: false }; + } + if (kind === "cancelled" || kind === "completed") { + await db.update(heartbeatRuns).set({ status: kind === "cancelled" ? "cancelled" : "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + return { exitCode: 0, timedOut: false }; + } + throw new Error("probe unavailable"); + }); + if (kind === "stop_failed") runtime.retryPendingSandboxTeardown.mockRejectedValueOnce(new Error("provider stop unavailable")); + await expect(retryNativeWorkspaceExport(f.request)).rejects.toThrow(); + expect(runtime.retryPendingSandboxTeardown).toHaveBeenCalledOnce(); + expect(runtime.retryPendingSandboxTeardown).toHaveBeenCalledWith(expect.objectContaining({ lease: expect.objectContaining({ id: f.leaseId, heartbeatRunId: f.runId }) })); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + expect(lease).toMatchObject(kind === "stop_failed" ? { status: "pending_cleanup", cleanupStatus: "failed" } : { status: "released", cleanupStatus: "success" }); + if (kind === "stop_failed") { + expect(lease.metadata?.remoteExecutionTermination).toBeUndefined(); + expect(lease.metadata?.pendingCleanupInFlight).toBe(false); + expect(lease.metadata?.nativeWorkspaceExportResume).toMatchObject({ runId: f.runId, leaseId: f.leaseId }); + } else expect(lease.metadata?.remoteExecutionTermination).toMatchObject({ runId: f.runId, leaseId: f.leaseId, state: "stopped" }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toMatchObject({ phase: "terminal_failure", resultId: f.resultId }); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).toHaveLength(0); + }); + it.each(["rebound_lease", "competing_owner", "late_stop_receipt"])("does not stop or overwrite a newer sandbox owner: %s", async kind => { + const f = await seed(); + const runtime = f.request.environmentRuntime as { retryPendingSandboxTeardown: ReturnType }; + const rebound = async () => db.update(environmentLeases).set({ status: "active", heartbeatRunId: null, + metadata: { newOwner: true }, cleanupStatus: null, releasedAt: null }).where(eq(environmentLeases.id, f.leaseId)); + probe.mockImplementationOnce(async () => { + if (kind === "rebound_lease") await rebound(); + if (kind === "competing_owner") { + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + await db.insert(environmentLeases).values({ companyId, environmentId, status: "active", provider: lease.provider, providerLeaseId: lease.providerLeaseId }); + } + throw new Error("probe failed after ownership changed"); + }); + if (kind === "late_stop_receipt") runtime.retryPendingSandboxTeardown.mockImplementationOnce(async ({ lease }) => { + await rebound(); return { providerLeaseId: lease.providerLeaseId, state: "stopped" }; + }); + await expect(retryNativeWorkspaceExport(f.request)).rejects.toThrow(); + expect(runtime.retryPendingSandboxTeardown).toHaveBeenCalledTimes(kind === "late_stop_receipt" ? 1 : 0); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + if (kind !== "competing_owner") expect(lease).toMatchObject({ status: "active", heartbeatRunId: null, metadata: { newOwner: true }, cleanupStatus: null }); + expect(lease.metadata?.remoteExecutionTermination).toBeUndefined(); + }); it.each(["missing", "replacement", "wrong_root"])("refuses an unproven resume without probing or acquiring replacement: %s", async kind => { const f = await seed(); const resume = (f.request.environmentRuntime as { resumeRunLease: ReturnType }).resumeRunLease; diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index bbb14d6e157..6d527a4383e 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -4,7 +4,8 @@ import { and, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; import { environmentLeases, environments, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { conflict } from "../../errors.js"; import { persistActivity, publishActivity } from "../activity-log.js"; -import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; +import { hasRemoteTerminationReceipt, remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { NATIVE_WORKSPACE_EXPORT_RESUME_KEY, readNativeWorkspaceExportResume, settleNativeWorkspaceExportResume } from "./native-workspace-export-resume.js"; import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; import type { EnvironmentRuntimeService } from "../environment-runtime.js"; @@ -19,7 +20,7 @@ export async function retryNativeWorkspaceExport(input: { db: Db; companyId: string; issueId: string; actionId: string; runId: string; actorId: string; repairNote: string; environmentRuntime: EnvironmentRuntimeService; }) { - async function inspect(db: Db, lock = false) { + async function inspect(db: Db, lock = false, resumeRequestId?: string) { const one = async (query: PromiseLike & { for(mode: "update"): PromiseLike }): Promise => (await (lock ? query.for("update") : query))[0] ?? null; // Keep the same issue -> coordinator -> run order as status commitment. @@ -39,7 +40,7 @@ export async function retryNativeWorkspaceExport(input: { if (!lease || lease.heartbeatRunId !== run.id || lease.issueId !== issue.id || lease.providerLeaseId !== reference.providerLeaseId || !lease.environmentId) throw changed(); const [otherLease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( - ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider!), eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), eq(environmentLeases.status, "active"), + ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider!), eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), inArray(environmentLeases.status, ["active", "pending_cleanup"]), )).limit(1); if (otherLease) throw changed(); const [result] = await db.select().from(nativeRunResults).where(and(eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.issueId, issue.id), eq(nativeRunResults.runId, run.id), eq(nativeRunResults.id, coordinator.resultId ?? "00000000-0000-0000-0000-000000000000"))).limit(1); @@ -53,9 +54,16 @@ export async function retryNativeWorkspaceExport(input: { const pending = record(action.evidence?.workspaceExportRetry); const alreadyQueued = coordinator.phase !== "terminal_failure" && pending.runId === run.id && pending.resultId === result.id && pending.leaseId === lease.id; + const resumeIntent = readNativeWorkspaceExportResume(lease); + const ownsResume = Boolean(resumeRequestId && lease.status === "pending_cleanup" + && lease.metadata?.pendingCleanupAttemptId === resumeRequestId + && lease.metadata?.pendingCleanupInFlight === true && resumeIntent?.requestId === resumeRequestId + && Number(lease.metadata?.pendingCleanupLeaseExpiresAtMs) > Date.now() + && resumeIntent.resultId === result.id); if (!alreadyQueued && (coordinator.phase !== "terminal_failure" || coordinator.failureCode !== CAUSE - || run.status !== "failed" || !hasRemoteTerminationReceipt(lease) - || record(lease.metadata?.remoteExecutionTermination).state !== "stopped")) throw changed(); + || run.status !== "failed" || (!ownsResume && (!hasRemoteTerminationReceipt(lease) + || record(lease.metadata?.remoteExecutionTermination).state !== "stopped")))) throw changed(); + if (resumeRequestId && !ownsResume) throw changed(); const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId)).limit(1); if (!environment) throw changed(); return { issue, coordinator, run, action, reference, lease, result, environment, alreadyQueued }; @@ -63,58 +71,112 @@ export async function retryNativeWorkspaceExport(input: { const owned = await withNativeWorkspaceFinalizationOwnership(input, async (ownership) => { const initial = await inspect(input.db); - if (!initial.alreadyQueued) { - // The operator resumes and repairs the exact retained sandbox through its - // provider console. Never acquire a replacement or seed over saved work. - const target = await resolveEnvironmentExecutionTarget({ db: input.db, companyId: input.companyId, - adapterType: "paperclip_runner", environment: initial.environment, leaseId: initial.lease.id, - leaseMetadata: initial.lease.metadata, lease: initial.lease as EnvironmentLease, environmentRuntime: input.environmentRuntime }); - if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner - || target.remoteCwd !== initial.reference.remoteCwd) throw changed(); - try { - // A provider stop also closes its controller-side activity gate. The - // operator's external repair does not reopen that gate. Resume only the - // recorded lease through its provider so sentinel and drain checks run. - const resumed = await input.environmentRuntime.resumeRunLease({ environment: initial.environment as unknown as Environment, - lease: initial.lease as EnvironmentLease }); - if (resumed?.providerLeaseId !== initial.lease.providerLeaseId - || resumed.metadata?.remoteCwd !== initial.reference.remoteCwd) throw new Error("sandbox_resume_identity_unproven"); - const probe = await target.runner.execute({ command: target.shellCommand ?? "sh", args: ["-c", "true"], cwd: "/", timeoutMs: 10_000, bypassSession: true }); - if (probe.timedOut || probe.exitCode !== 0) throw new Error("sandbox_not_running"); - } catch { - throw conflict("Resume and repair the retained sandbox before retrying workspace export. No provider turn was started.", { code: "workspace_export_sandbox_unavailable" }); + let resumeRequestId: string | undefined; + try { + if (!initial.alreadyQueued) { + // The operator resumes and repairs the exact retained sandbox through its + // provider console. Never acquire a replacement or seed over saved work. + const target = await resolveEnvironmentExecutionTarget({ db: input.db, companyId: input.companyId, + adapterType: "paperclip_runner", environment: initial.environment, leaseId: initial.lease.id, + leaseMetadata: initial.lease.metadata, lease: initial.lease as EnvironmentLease, environmentRuntime: input.environmentRuntime }); + if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner + || target.remoteCwd !== initial.reference.remoteCwd) throw changed(); + await ownership.assertHeld(); + const requestId = randomUUID(); + const claimedLease = await input.db.transaction(async tx => { + const current = await inspect(tx as unknown as Db, true); + if (current.alreadyQueued || current.result.id !== initial.result.id || current.lease.id !== initial.lease.id + || current.reference.descriptorSha256 !== initial.reference.descriptorSha256) throw changed(); + const now = new Date(); + // Persist cleanup authority before any provider call. A lost resume + // reply or controller restart must not leave a running sandbox behind + // an old stopped receipt. The pending claim fences the cleanup sweeper. + const [lease] = await tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", + releasedAt: now, failureReason: "workspace_export_resume_pending", updatedAt: now, + metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined, + [NATIVE_WORKSPACE_EXPORT_RESUME_KEY]: { schema: "paperclip.workspace-export-resume.v1", requestId, + companyId: input.companyId, runId: input.runId, leaseId: current.lease.id, + provider: current.lease.provider, providerLeaseId: current.lease.providerLeaseId, resultId: current.result.id }, + pendingCleanupAttemptId: requestId, pendingCleanupInFlight: true, + pendingCleanupLeaseExpiresAtMs: now.getTime() + 15 * 60_000 }, + }).where(eq(environmentLeases.id, current.lease.id)).returning(); + return lease; + }); + resumeRequestId = requestId; + try { + await ownership.assertHeld(); + await inspect(input.db, false, requestId); + // A provider stop also closes its controller-side activity gate. The + // operator's external repair does not reopen that gate. Resume only the + // recorded lease through its provider so sentinel and drain checks run. + const resumed = await input.environmentRuntime.resumeRunLease({ environment: initial.environment as unknown as Environment, + lease: claimedLease as EnvironmentLease }); + if (resumed?.providerLeaseId !== initial.lease.providerLeaseId + || resumed.metadata?.remoteCwd !== initial.reference.remoteCwd) throw new Error("sandbox_resume_identity_unproven"); + const probe = await target.runner.execute({ command: target.shellCommand ?? "sh", args: ["-c", "true"], cwd: "/", timeoutMs: 10_000, bypassSession: true }); + if (probe.timedOut || probe.exitCode !== 0) throw new Error("sandbox_not_running"); + } catch { + throw conflict("Resume and repair the retained sandbox before retrying workspace export. No provider turn was started.", { code: "workspace_export_sandbox_unavailable" }); + } } + await ownership.assertHeld(); + const admitted = await input.db.transaction(async tx => { + const current = await inspect(tx as unknown as Db, true, resumeRequestId); + if (current.result.id !== initial.result.id || current.lease.id !== initial.lease.id + || current.reference.descriptorSha256 !== initial.reference.descriptorSha256) throw changed(); + const receipt = { runId: current.run.id, resultId: current.result.id, leaseId: current.lease.id, status: "queued" as const }; + if (current.alreadyQueued) return { receipt, publication: null }; + const now = new Date(); + const retry = { ...receipt, requestId: randomUUID(), actorId: input.actorId, repairNote: input.repairNote, requestedAt: now.toISOString(), + stoppedProvider: initial.lease.metadata?.remoteExecutionTermination }; + await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, + failureDetail: { ...current.coordinator.failureDetail, workspaceFinalizeAttempt: 0, workspaceExportRetry: retry }, updatedAt: now }).where(eq(nativeRunFinalizations.runId, current.run.id)); + await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null, + nativePhase: "result_accepted", nativePhaseUpdatedAt: now, + resultJson: { ...current.run.resultJson, workspaceExportRetry: retry, finalizationPhase: "result_accepted", failureCode: null, originalFailureCode: null, nextAttemptAt: null }, updatedAt: now }).where(eq(heartbeatRuns.id, current.run.id)); + await tx.update(issues).set({ executionRunId: current.run.id, updatedAt: now }).where(eq(issues.id, current.issue.id)); + // Reclaim the same lease only for control-plane copyback. The normal + // finalization settlement will stop/retain it again on success or failure. + await tx.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, failureReason: null, + metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined, + [NATIVE_WORKSPACE_EXPORT_RESUME_KEY]: undefined, pendingCleanupAttemptId: undefined, + pendingCleanupInFlight: undefined, pendingCleanupLeaseExpiresAtMs: undefined }, updatedAt: now }).where(eq(environmentLeases.id, current.lease.id)); + await tx.update(issueRecoveryActions).set({ evidence: { ...current.action.evidence, workspaceExportRetry: retry }, + nextAction: "Workspace export is queued for the accepted result. No provider turn will run.", + wakePolicy: { kind: "resume_native_run", runId: current.run.id }, updatedAt: now }).where(eq(issueRecoveryActions.id, current.action.id)); + const publication = await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "user", actorId: input.actorId, + action: "issue.workspace_export_retry_requested", entityType: "issue", entityId: input.issueId, runId: input.runId, + details: { recoveryActionId: input.actionId, resultId: current.result.id, leaseId: current.lease.id, repairNote: input.repairNote } }); + return { receipt, publication }; + }); + if (admitted.publication) publishActivity(admitted.publication.publication); + return admitted.receipt; + } catch (error) { + if (resumeRequestId) { + // Admission may have committed before a connection failed. Stop only + // while our durable intent is still present; never stop a later owner. + try { + const [lease] = await input.db.select().from(environmentLeases).where(and(eq(environmentLeases.id, initial.lease.id), + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.heartbeatRunId, input.runId), + eq(environmentLeases.status, "pending_cleanup"), + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${resumeRequestId}`)).limit(1); + const [otherOwner] = await input.db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + ne(environmentLeases.id, initial.lease.id), eq(environmentLeases.provider, initial.lease.provider!), + eq(environmentLeases.providerLeaseId, initial.lease.providerLeaseId!), inArray(environmentLeases.status, ["active", "pending_cleanup"]), + )).limit(1); + if (lease && !otherOwner && readNativeWorkspaceExportResume(lease)?.requestId === resumeRequestId) { + let termination: ReturnType; + try { + const receipt = await input.environmentRuntime.retryPendingSandboxTeardown({ environment: initial.environment as unknown as Environment, lease: lease as EnvironmentLease }); + termination = remoteTerminationReceipt(lease, receipt); + if (termination?.state !== "stopped") termination = undefined; + } catch { /* The durable stop-only intent remains recoverable. */ } + await settleNativeWorkspaceExportResume(input.db, lease, { attemptId: resumeRequestId, receipt: termination }); + } + } catch { /* A database outage leaves the pre-resume intent for recovery. */ } + } + throw error; } - await ownership.assertHeld(); - const admitted = await input.db.transaction(async tx => { - const current = await inspect(tx as unknown as Db, true); - if (current.result.id !== initial.result.id || current.lease.id !== initial.lease.id - || current.reference.descriptorSha256 !== initial.reference.descriptorSha256) throw changed(); - const receipt = { runId: current.run.id, resultId: current.result.id, leaseId: current.lease.id, status: "queued" as const }; - if (current.alreadyQueued) return { receipt, publication: null }; - const now = new Date(); - const retry = { ...receipt, requestId: randomUUID(), actorId: input.actorId, repairNote: input.repairNote, requestedAt: now.toISOString(), - stoppedProvider: current.lease.metadata?.remoteExecutionTermination }; - await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, - failureDetail: { ...current.coordinator.failureDetail, workspaceFinalizeAttempt: 0, workspaceExportRetry: retry }, updatedAt: now }).where(eq(nativeRunFinalizations.runId, current.run.id)); - await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null, - nativePhase: "result_accepted", nativePhaseUpdatedAt: now, - resultJson: { ...current.run.resultJson, workspaceExportRetry: retry, finalizationPhase: "result_accepted", failureCode: null, originalFailureCode: null, nextAttemptAt: null }, updatedAt: now }).where(eq(heartbeatRuns.id, current.run.id)); - await tx.update(issues).set({ executionRunId: current.run.id, updatedAt: now }).where(eq(issues.id, current.issue.id)); - // Reclaim the same lease only for control-plane copyback. The normal - // finalization settlement will stop/retain it again on success or failure. - await tx.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, - metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined }, updatedAt: now }).where(eq(environmentLeases.id, current.lease.id)); - await tx.update(issueRecoveryActions).set({ evidence: { ...current.action.evidence, workspaceExportRetry: retry }, - nextAction: "Workspace export is queued for the accepted result. No provider turn will run.", - wakePolicy: { kind: "resume_native_run", runId: current.run.id }, updatedAt: now }).where(eq(issueRecoveryActions.id, current.action.id)); - const publication = await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "user", actorId: input.actorId, - action: "issue.workspace_export_retry_requested", entityType: "issue", entityId: input.issueId, runId: input.runId, - details: { recoveryActionId: input.actionId, resultId: current.result.id, leaseId: current.lease.id, repairNote: input.repairNote } }); - return { receipt, publication }; - }); - if (admitted.publication) publishActivity(admitted.publication.publication); - return admitted.receipt; }); if (!owned.acquired) throw conflict("Workspace finalization is still owned by another operation. Wait for it to stop before retrying.", { code: "workspace_export_retry_busy" }); return owned.value;