From 79c34058cdea4aa4de09f5e2455d98ba066f1685 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Wed, 14 Oct 2026 23:07:15 -0700 Subject: [PATCH] fix(native): retain unexported workspaces after execution failure (#16397) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native agents can edit files in a remote sandbox before they submit a result. > - A process identity or startup failure can end the run before file export. > - Cleanup can then delete the sandbox and its unpublished files. > - This pull request retains the exact source and records a file repair action before cleanup. > - Operators can recover the files without replaying the failed turn or changing the task owner. ## Linked Issues or Issue Description **What happened?** A native remote process had completed local edits when process identity recovery failed. The run still had a prepared workspace reference and no accepted result. Cleanup deleted the source sandbox. No file recovery card appeared. A restarted run can hit the same path during setup, before the native executor starts. **Expected behavior** Retain the exact unexported source. Show the existing file recovery card. Require source stop and file recovery evidence. Keep task ownership, status, and external action reconciliation separate from file repair. **Steps to reproduce** 1. Run a native agent in a remote ephemeral workspace and make unpublished edits. 2. Fail process identity recovery before result acceptance or file export. 3. Allow failure handling and lease cleanup to run. 4. Before this change, cleanup can destroy the source without a file repair action. Refs: #16341 for related process identity diagnostics. Refs: #16349 for startup ownership fencing. This change addresses file retention after failure. ## What Changed - Record source retention and a file recovery action in the terminal native failure transaction. - Cover restarted preparation failures before the terminal setup writer and cleanup run. - Require the exact company, run, lease, provider allocation, and workspace binding. Require the exact live recovery claim to settle an owned preparation failure. Refuse stale or competing claims. - Reuse the existing stop-only cleanup and file-only reconciliation path. Leave accepted results and scheduled same-run recovery on their existing paths. - Document the retention rule and add database and runtime boundary regressions. ## Verification - All 1,242 tests passed across the full heartbeat recovery suite, native executor, native source retention, runtime teardown, run completion, and legacy file recovery. - The integrated missing-descriptor regression fails before the inner-handler fix and passes after it. The runtime retention and owned restart regressions also reproduce their prior failures. - The tests cover exact source identity, current and stale claims, atomic rollback, cancellation, idempotence, and file-only repair on a completed task. - `pnpm -r typecheck` passed on the final code. - `pnpm build` passed on the final code. - Full repository CI passed on `3945eb2ba5c228c5d4992da02b765dcf2e634cb8`: 54 successful checks and two skipped Storybook checks. - Apex scored 5/5 on that exact commit. Both earlier findings are fixed and all review threads are resolved. - The unsharded local `pnpm test:run` is still in progress. It is not counted as a local full-suite pass; complete suite coverage is verified in CI. ## Risks - Failed runs can retain a sandbox that needs operator cleanup after file recovery. This uses the existing retention policy and recovery controls. - This does not recover files from allocations that were already deleted. - No schema migration, UI change, or telemetry change is required. ## Model Used - OpenAI GPT-6 through Codex. Tool use, code execution, and reasoning assisted this change. The environment identifies the model as GPT-6; it does not expose a more specific serving snapshot or context window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/native-workspace-finalization-recovery.md | 16 ++ .../heartbeat-process-recovery.test.ts | 51 +++++ server/src/services/heartbeat.ts | 22 +- .../heartbeat/runtime-execution.test.ts | 19 ++ .../services/heartbeat/runtime-execution.ts | 3 + .../src/services/legacy-execution-recovery.ts | 19 +- .../legacy-workspace-restore-recovery.ts | 9 +- .../native-failed-workspace-recovery.test.ts | 202 ++++++++++++++++++ .../native-failed-workspace-recovery.ts | 74 +++++++ .../native-runtime/native-session-executor.ts | 8 +- 10 files changed, 410 insertions(+), 13 deletions(-) create mode 100644 server/src/services/native-runtime/native-failed-workspace-recovery.test.ts create mode 100644 server/src/services/native-runtime/native-failed-workspace-recovery.ts diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 86438c3ca48..3d861bb82a9 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -1,5 +1,21 @@ # Native workspace finalization ownership and recovery +A native provider can fail before it submits a result while its workspace still +contains unpublished edits. A prepared workspace reference is not an export +receipt. Terminal provider failure pins the exact bound sandbox and records a +file-recovery obligation in the same transaction as the failed run. Both setup +failure handlers record that obligation before their terminal writer and provider +cleanup run. This includes descriptor reads that fail inside runtime selection. Only its exact live recovery claim can settle an owned +preparation failure; another owner or a stale claim prevents cleanup. Settling +that claim and recording file retention share a transaction. + +The recovery card names the original run and requires evidence that its exact +allocation stopped and its files were recovered. Recording that evidence does +not retry the old turn, certify unknown external actions, change the current task +owner or status, or require an accepted native result. Accepted-result export +recovery and same-run provider recovery keep their existing paths. An export +that has already finalized does not create this file obligation. + Native workspace export and merge acquire a PostgreSQL advisory lock scoped to company and run before the first physical copyback. The live heartbeat and the reconciler share that lock. Recovery skips a busy owner without recording another diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index e1c7c6acd8b..9d24caf061d 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -3,6 +3,7 @@ import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import * as conversationContinuation from "../services/conversation-continuation.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; import * as workspaceRuntime from "../services/workspace-runtime.js"; +import { nativeWorkspaceSyncInternals } from "../services/native-runtime/native-workspace-sync.js"; import { ensureNativeCompletionContract } from "../services/native-runtime/completion-contracts.js"; import { readNativePlanWait, hasCommittedNativePlanWait } from "../services/native-runtime/native-plan-wait.js"; import { nativeSha256 } from "../services/native-runtime/canonical.js"; @@ -3520,6 +3521,56 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } } + it("retains the source when a saved native descriptor fails inside runtime selection", async () => { + await withTempPaperclipHome(async () => { + const f = await seedQueuedIssueRunFixture(); + await db.update(agents).set({ adapterType: "paperclip_runner", + adapterConfig: { provider: "codex", model: "gpt-5.6-luna" }, + }).where(eq(agents.id, f.agentId)); + const originalFactory = environmentOrchestrator.environmentRunOrchestrator; + const release = vi.fn(async () => ({ released: [], errors: [] })); + const factory = vi.spyOn(environmentOrchestrator, "environmentRunOrchestrator").mockImplementation((...args) => ({ + ...originalFactory(...args), releaseForRun: release, + })); + const provider = vi.fn(() => { throw new Error("provider must not start"); }); + let leaseId: string | undefined; + try { + const heartbeat = heartbeatService(db, { + nativeSessionBackendFactory: provider, + beforeNativeRuntimeSelection: async runId => { + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: runId, provider: "daytona", providerLeaseId: randomUUID(), status: "active", leasePolicy: "ephemeral", + metadata: { driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true, remoteCwd: "/work" }, + }).returning(); + leaseId = lease!.id; + const reference = { schema: "paperclip.native-workspace-sync/v3" as const, state: "prepared" as const, + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: randomUUID(), leaseId: lease!.id, providerLeaseId: lease!.providerLeaseId!, remoteCwd: "/work", resourceDisposition: "destroy" as const }; + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: f.issueId, nativePhase: "observed", + runnerProfileJson: { nativeWorkspaceSync: reference }, + }).where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ runId, companyId: f.companyId, issueId: f.issueId, phase: "observed" }); + await nativeWorkspaceSyncInternals.readDescriptor({ companyId: f.companyId, runId, workspaceId: reference.workspaceId, reference }); + }, + }); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + expect(leaseId).toBeDefined(); + expect(await heartbeat.getRun(f.runId, { includeExecutionEvidence: true })).toMatchObject({ status: "failed", + runtimeMode: "native", nativePhase: "terminal_failure", resultJson: { + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { leaseIds: [leaseId] }, + }, + }); + expect(await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId!))) + .toMatchObject([{ status: "pending_cleanup", leasePolicy: "retain_on_failure" }]); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ workspaceRepairRequired: true, canRetry: false }); + expect(release).toHaveBeenCalledWith(expect.objectContaining({ providerResourceDisposition: "stop_and_retain" })); + expect(provider).not.toHaveBeenCalled(); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + } finally { factory.mockRestore(); } + }); + }); + it("fences native selection when cancellation wins during preparation", async () => { await withTempPaperclipHome(async () => { const { agentId, issueId, runId } = await seedQueuedIssueRunFixture(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c7ab34fa48e..c8544c5bdb9 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -8,6 +8,7 @@ import { prepareHeartbeatTaskContext } from "./heartbeat/task-context.js"; import { selectHeartbeatEnvironment } from "./heartbeat/environment-selection.js"; import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js"; import { executeHeartbeatRuntime, NativeSessionResumeScheduledError, NativeWorkspaceFinalizeScheduledError } from "./heartbeat/runtime-execution.js"; +import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; import { selectHeartbeatRuntime } from "./heartbeat/runtime-selection.js"; import { createHeartbeatRunCompletion } from "./heartbeat/run-completion.js"; export { @@ -1935,6 +1936,23 @@ export function heartbeatService( let requiredWorkspaceRestoreEvidence: Record | null = null; let providerResourceDispositionForRun: ProviderResourceDisposition | undefined; + const preserveNativeWorkspaceOnFailure = async () => { + const failureRun = await getRun(run.id); + if (failureRun?.runtimeMode === "native" + && parseObject(failureRun.runnerProfileJson?.nativeWorkspaceSync).state === "prepared") { + providerResourceDispositionForRun = "stop_and_retain"; + try { + await preserveNativeFailedWorkspace(db, failureRun, { + allowPreterminal: true, + restartRecovery: runOptions.nativeRestartRecovery, + }); + } catch (recoveryError) { + if (recoveryError instanceof NativeWorkspaceFinalizationBusyError) nativeOwnershipHeld = true; + else logger.error({ err: recoveryError, runId: run.id }, "failed to record retained native workspace recovery"); + } + } + return !nativeOwnershipHeld; + }; let nativeLifecycleTelemetryForRun: | { provider: string; @@ -3592,6 +3610,7 @@ export function heartbeatService( } return; } + if (!(await preserveNativeWorkspaceOnFailure())) return; await runCompletion.failRun({ run, agent, issueId, issueRef, signal: executionControl.controller.signal, output: runOutput, @@ -3638,7 +3657,8 @@ export function heartbeatService( }, ); } else { - await runCompletion.failRunSetup({ + await preserveNativeWorkspaceOnFailure(); + if (!nativeOwnershipHeld) await runCompletion.failRunSetup({ run, runId, outerErr, signal: executionControl.controller.signal, providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted, identityRedactor, readFailureReportSecrets, diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts index 4cee3f2734c..20f2ff1ec4d 100644 --- a/server/src/services/heartbeat/runtime-execution.test.ts +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -234,6 +234,25 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = }]); }); + it("retains an unexported native source before provider-error teardown", async () => { + const { input, run } = await fixture(true); + input.workspace.nativeWorkspaceSync = { + mode: "host_current", + reference: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: run.id, leaseId: randomUUID(), providerLeaseId: "sandbox", remoteCwd: "/workspace", resourceDisposition: "destroy" }, + restoreWorkspace: vi.fn(), cleanup: vi.fn(), + }; + const error = new Error("runner_remote_process_identity_unavailable"); + vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error); + input.instructions.releaseInstructionCopy = vi.fn(async () => { + expect(input.effects.onProviderResourceDisposition).toHaveBeenCalledWith("stop_and_retain"); + }); + await expect(executeHeartbeatRuntime(db, input)).rejects.toBe(error); + expect(input.workspace.nativeWorkspaceSync.restoreWorkspace).not.toHaveBeenCalled(); + expectCleanup(input); + }); + it("records a failed barrier if the successful finalization write fails", async () => { const { input } = await fixture(); const error = new Error("finalization write failed"); diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index 931b81edc59..fe601c505b0 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -994,6 +994,9 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe input.effects.onNativeOwnershipHeld(); throw adapterErr; } + if (nativeRuntimeResolution.kind === "native" && nativeWorkspaceSync?.reference.state === "prepared") { + input.effects.onProviderResourceDisposition("stop_and_retain"); + } await db .update(heartbeatRuns) .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) }) diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 828ad951553..b3b3609f31b 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -79,7 +79,7 @@ export async function terminalizeLegacyExecution(input: { const { db, run, status } = input; let patch = input.patch; const issueId = - run.nativeIssueId ?? + run.nativeIssueId ?? run.issueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null); @@ -99,7 +99,7 @@ export async function terminalizeLegacyExecution(input: { const [current] = await tx.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), )).for("update"); - if (!current || current.runtimeMode !== "legacy" || + if (!current || (current.runtimeMode !== "legacy" && !(input.recordRestoreFailureOnly && current.runtimeMode === "native")) || (!input.recordRestoreFailureOnly && !(input.fromStatuses ?? [run.status]).includes(current.status))) return null; if (patch?.resultJson !== undefined) patch = { ...patch, resultJson: preserveWorkspaceRestoreRecoveryMetadata(current.resultJson, @@ -154,9 +154,10 @@ export async function terminalizeLegacyExecution(input: { // newer owner or conversation generation start before copy-back settled. // A resolved no-replay hold can coexist with another active incident; the // one-active-action index must never make us supersede either obligation. + const fingerprint = `${run.runtimeMode === "native" ? "native-workspace" : "legacy-execution"}:${run.id}`; const [existing] = await tx.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, task.id), - eq(issueRecoveryActions.fingerprint, `legacy-execution:${run.id}`), + eq(issueRecoveryActions.fingerprint, fingerprint), )).orderBy(desc(issueRecoveryActions.createdAt)).limit(1).for("update"); const decision = existing?.evidence.executionReconciliation as Record | undefined; if (hasRequiredWorkspaceRecovery(existing?.evidence) && decision?.runId === run.id) return updated; @@ -179,8 +180,9 @@ export async function terminalizeLegacyExecution(input: { }; if (existing) await tx.update(issueRecoveryActions).set(values).where(eq(issueRecoveryActions.id, existing.id)); else await tx.insert(issueRecoveryActions).values({ ...values, companyId: run.companyId, - sourceIssueId: task.id, kind: "active_run_watchdog", cause: LEGACY_RECOVERY_CAUSE, - fingerprint: `legacy-execution:${run.id}`, attemptCount: 1 }); + sourceIssueId: task.id, kind: "active_run_watchdog", + cause: run.runtimeMode === "native" ? "native_continuation_requires_reconciliation" : LEGACY_RECOVERY_CAUSE, + fingerprint, attemptCount: 1 }); return updated; } const review = task?.status === "in_review" ? parseIssueExecutionState(task.executionState) : null; @@ -314,7 +316,12 @@ export async function settleInterruptedNativeBootstrap( * execution ownership while the adapter's host finalization is still running. */ export async function recordLegacyWorkspaceRestoreFailure(db: Db, run: Run, evidence: Record, source?: LegacyWorkspaceRestoreSource) { if (run.runtimeMode !== "legacy" || !hasWorkspaceRestoreFailure(evidence)) return; - await terminalizeLegacyExecution({ db, run, status: run.status, + return recordRequiredWorkspaceRestoreFailure(db, run, evidence, source); +} + +export async function recordRequiredWorkspaceRestoreFailure(db: Db, run: Run, evidence: Record, source?: LegacyWorkspaceRestoreSource) { + if (!hasWorkspaceRestoreFailure(evidence)) return; + return terminalizeLegacyExecution({ db, run, status: run.status, recordRestoreFailureOnly: true, patch: { resultJson: evidence }, workspaceRestoreSource: source, }); } diff --git a/server/src/services/legacy-workspace-restore-recovery.ts b/server/src/services/legacy-workspace-restore-recovery.ts index fc1419ddfb8..47ce54f6f84 100644 --- a/server/src/services/legacy-workspace-restore-recovery.ts +++ b/server/src/services/legacy-workspace-restore-recovery.ts @@ -14,7 +14,7 @@ export type LegacyWorkspaceRestoreSource = Pick; function needsRemoteRestore(run: Run): boolean { - return run.runtimeMode === "legacy" && ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status) + return ["failed", "timed_out", "interrupted", "cancelled"].includes(run.status) && hasWorkspaceRestoreFailure(run.resultJson); } @@ -54,7 +54,7 @@ function matchesOriginalSource(lease: typeof environmentLeases.$inferSelect, run /** A local lock timeout has no remote source to retain and keeps its old policy. */ export async function hasUnrestoredRemoteWorkspace(db: Db, run: Run): Promise { - if (!needsRemoteRestore(run)) return false; + if (run.runtimeMode !== "legacy" || !needsRemoteRestore(run)) return false; if (hasRequiredWorkspaceRecovery(run.resultJson)) return true; const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases) .leftJoin(environments, eq(environments.id, environmentLeases.environmentId)).where(leaseScope(run)).limit(1); @@ -65,10 +65,11 @@ export async function hasUnrestoredRemoteWorkspace(db: Db, run: Run): Promise { - if (!needsRemoteRestore(run)) return []; + if (!needsRemoteRestore(run) || (run.runtimeMode === "native" && !source)) return []; const rows = await db.select({ lease: environmentLeases }).from(environmentLeases) .leftJoin(environments, eq(environments.id, environmentLeases.environmentId)) - .where(leaseScope(run, source)).for("update", { of: environmentLeases }); + .where(and(leaseScope(run, source), run.runtimeMode === "native" && source ? eq(environmentLeases.id, source.id) : undefined)) + .for("update", { of: environmentLeases }); const retained: string[] = []; for (const { lease } of rows) { if (lease.status !== "active") { diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts new file mode 100644 index 00000000000..bd48f561e43 --- /dev/null +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -0,0 +1,202 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { getExecutionBlocker } from "../execution-blocker.js"; +import { markExecutionReconciliation, validateExecutionReconciliation } from "../execution-recovery-resolution.js"; +import { settleStopOnlyCleanup } from "../sandbox-stop-and-retain.js"; +import { preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; +import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; +import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; +import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; + +const externalUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim(); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("failed native workspace retention", () => { + let db: ReturnType; + let temporary: Awaited>; + beforeAll(async () => { + if (externalUrl) db = createDb(externalUrl); + else { + temporary = await startEmbeddedPostgresTestDatabase("native-failed-workspace-"); + db = createDb(temporary.connectionString); + } + }, 30_000); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + + async function seed() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Native workspace", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Engineer", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Preserve unfinished files", status: "in_progress", assigneeAgentId: agentId }); + const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, + runtimeMode: "native", status: "failed", nativePhase: "terminal_failure", errorCode: "adapter_failed", + error: "runner_remote_process_identity_unavailable", resultJson: { originalDiagnostic: true }, contextSnapshot: { issueId }, + }).returning(); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "terminal_failure" }); + const [environment] = await db.insert(environments).values({ name: `Native recovery ${runId}`, driver: "sandbox" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), status: "active", leasePolicy: "ephemeral", + metadata: { driver: "sandbox", pluginId: randomUUID(), sandboxProviderPlugin: true, remoteCwd: "/work" }, + }).returning(); + const [bound] = await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64), + baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: runId, + leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work", resourceDisposition: "destroy", + } } }).where(eq(heartbeatRuns.id, runId)).returning(); + return { companyId, agentId, issueId, runId, run: bound, lease, environment }; + } + const readRun = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + const readLease = async (id: string) => (await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)))[0]!; + const actions = async (issueId: string) => db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + + it("pins only the exact unexported source and exposes a file-only recovery card", async () => { + const f = await seed(); + const [other] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.runId, + issueId: f.issueId, environmentId: f.environment.id, provider: "daytona", providerLeaseId: randomUUID(), + status: "active", metadata: f.lease.metadata, + }).returning(); + await db.transaction(tx => preserveNativeFailedWorkspace(tx as unknown as typeof db, f.run)); + expect(await readRun(f.runId)).toMatchObject({ status: "failed", errorCode: "adapter_failed", resultJson: { + originalDiagnostic: true, workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { leaseIds: [f.lease.id] }, + } }); + expect(await readLease(f.lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure", metadata: { + sandboxStopAndRetain: { runId: f.runId, leaseId: f.lease.id, providerLeaseId: f.lease.providerLeaseId }, + } }); + expect(await readLease(other.id)).toMatchObject({ status: "active" }); + expect(await actions(f.issueId)).toMatchObject([{ status: "resolved", outcome: "blocked", evidence: { runId: f.runId } }]); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, + cause: "native_continuation_requires_reconciliation", workspaceRepairRequired: true, + canRecordWorkspaceRepair: true, canRetry: false, canContinue: false, + }); + await preserveNativeFailedWorkspace(db, await readRun(f.runId)); + expect(await actions(f.issueId)).toHaveLength(1); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.action, "issue.workspace_recovery_required")))).toHaveLength(1); + }); + + it("records verified file repair without retrying the native run or changing its task", async () => { + const f = await seed(); + await db.update(issues).set({ status: "done", assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + await preserveNativeFailedWorkspace(db, f.run); + const lease = await readLease(f.lease.id); + await settleStopOnlyCleanup(db, lease, { attemptId: String(lease.metadata?.pendingCleanupAttemptId), + receipt: { providerLeaseId: lease.providerLeaseId, state: "stopped" }, + }); + const [action] = await actions(f.issueId); + const decision = { runId: f.runId, providerStopped: true as const, actionOutcome: "mixed" as const, + outcomeEvidence: "Files verified; no external action is replayed.", workspaceRepairEvidence: "Exact retained allocation exported and every source hash checked." }; + await validateExecutionReconciliation({ db, companyId: f.companyId, issueId: f.issueId, + agentId: null, sourceRunId: f.runId, decision, workspaceRepairOnly: true }); + await markExecutionReconciliation(db, action!, decision, "board-operator", undefined, { workspaceRepairOnly: true }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + expect(await db.select().from(issues).where(eq(issues.id, f.issueId))).toMatchObject([{ status: "done", assigneeAgentId: null }]); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + expect(await actions(f.issueId)).toMatchObject([{ evidence: { continuationDelivery: "not_requested" } }]); + }); + + async function claimRestart(f: Awaited>) { + const claim: NativeRestartRecoveryClaim = { kind: "reattach_remote_runner", runId: f.runId, + leaseOwner: randomUUID(), controllerGeneration: 2, providerAttempt: 3, + restartKind: "hard", recoveryRequestId: null, + remote: { providerLeaseId: f.lease.providerLeaseId!, remoteCwd: "/work" } }; + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed" }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: claim.leaseOwner, + leaseExpiresAt: new Date(Date.now() + 60_000), controllerGeneration: claim.controllerGeneration, attempt: claim.providerAttempt, + }).where(eq(nativeRunFinalizations.runId, f.runId)); + return claim; + } + + it.each([false, true])("captures a preparation failure before the terminal writer, with recovery claim: %s", async claimed => { + const f = await seed(); + const restartRecovery = claimed ? await claimRestart(f) : undefined; + if (!claimed) { + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed" }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + } + await preserveNativeFailedWorkspace(db, await readRun(f.runId), { allowPreterminal: true, restartRecovery }); + expect(await readRun(f.runId)).toMatchObject({ status: "running", resultJson: { workspaceRestoreRecovery: { leaseIds: [f.lease.id] } } }); + if (claimed) expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))) + .toMatchObject([{ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: null }]); + await db.update(heartbeatRuns).set({ status: "failed", resultJson: preserveWorkspaceRestoreRecoveryMetadataSql({ stopReason: "adapter_failed" }) }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await readRun(f.runId)).toMatchObject({ status: "failed", resultJson: { + stopReason: "adapter_failed", workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { leaseIds: [f.lease.id] }, + } }); + expect(await readLease(f.lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ workspaceRepairRequired: true }); + }); + + it("records terminal file retention after the executor retires its recovery claim", async () => { + const f = await seed(); + const restartRecovery = await claimRestart(f); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ status: "failed", nativePhase: "terminal_failure" }).where(eq(heartbeatRuns.id, f.runId)); + await preserveNativeFailedWorkspace(db, await readRun(f.runId), { allowPreterminal: true, restartRecovery }); + expect(await readRun(f.runId)).toMatchObject({ status: "failed", nativePhase: "terminal_failure" }); + expect(await readLease(f.lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ workspaceRepairRequired: true }); + }); + + it.each(["token", "generation", "attempt", "run", "remote", "expired", "released"])("refuses a stale %s recovery claim", async kind => { + const f = await seed(); + const restartRecovery = await claimRestart(f); + if (kind === "token") restartRecovery.leaseOwner = randomUUID(); + if (kind === "generation") restartRecovery.controllerGeneration += 1; + if (kind === "attempt") restartRecovery.providerAttempt += 1; + if (kind === "run") restartRecovery.runId = randomUUID(); + if (kind === "remote" && restartRecovery.kind === "reattach_remote_runner") restartRecovery.remote.remoteCwd = "/other"; + if (kind === "expired") await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (kind === "released") await db.update(nativeRunFinalizations).set({ leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await expect(preserveNativeFailedWorkspace(db, await readRun(f.runId), { allowPreterminal: true, restartRecovery })) + .rejects.toBeInstanceOf(NativeWorkspaceFinalizationBusyError); + expect(await readLease(f.lease.id)).toMatchObject({ status: "active" }); + expect(await actions(f.issueId)).toHaveLength(0); + expect(await readRun(f.runId)).toMatchObject({ status: "running", nativePhase: "observed" }); + }); + + it("refuses cleanup when another execution still owns the coordinator", async () => { + const f = await seed(); + await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: randomUUID() }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await expect(preserveNativeFailedWorkspace(db, f.run, { allowPreterminal: true })) + .rejects.toBeInstanceOf(NativeWorkspaceFinalizationBusyError); + expect(await readLease(f.lease.id)).toMatchObject({ status: "active" }); + expect(await actions(f.issueId)).toHaveLength(0); + }); + + it.each(["finalized", "retryable_failure", "result_accepted", "wrong_provider", "wrong_company"])("leaves %s outside failed-source retention", async kind => { + const f = await seed(); + if (kind === "retryable_failure" || kind === "result_accepted") { + await db.update(nativeRunFinalizations).set({ phase: kind }).where(eq(nativeRunFinalizations.runId, f.runId)); + } else { + const reference = f.run.runnerProfileJson!.nativeWorkspaceSync as Record; + if (kind === "wrong_company") { + const foreign = await seed(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + ...reference, leaseId: foreign.lease.id, providerLeaseId: foreign.lease.providerLeaseId, + } } }).where(eq(heartbeatRuns.id, f.runId)); + } else await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + ...reference, ...(kind === "finalized" ? { state: "finalized", finalHostSha256: "c".repeat(64) } : { providerLeaseId: randomUUID() }), + } } }).where(eq(heartbeatRuns.id, f.runId)); + } + await preserveNativeFailedWorkspace(db, await readRun(f.runId)); + expect(await actions(f.issueId)).toHaveLength(0); + expect(await readLease(f.lease.id)).toMatchObject({ status: "active" }); + }); + + it("rolls back terminal status, source retention and the recovery card together", async () => { + const f = await seed(); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.runId)); + await expect(db.transaction(async tx => { + const [run] = await tx.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId)).returning(); + await preserveNativeFailedWorkspace(tx as unknown as typeof db, run!); + throw new Error("rollback fixture"); + })).rejects.toThrow("rollback fixture"); + expect(await readRun(f.runId)).toMatchObject({ status: "running", resultJson: { originalDiagnostic: true } }); + expect(await readLease(f.lease.id)).toMatchObject({ status: "active", leasePolicy: "ephemeral" }); + expect(await actions(f.issueId)).toHaveLength(0); + }); +}); diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts new file mode 100644 index 00000000000..348c016309d --- /dev/null +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -0,0 +1,74 @@ +import { and, eq, gt, isNull, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; +import { hasRequiredWorkspaceRecovery } from "../workspace-restore-recovery-state.js"; +import { logActivity } from "../activity-log.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; +import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; +import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; + +export async function preserveNativeFailedWorkspace(db: Db, selected: typeof heartbeatRuns.$inferSelect, options: { allowPreterminal?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) { + if (readNativeWorkspaceSyncReference(selected.runnerProfileJson?.nativeWorkspaceSync)?.state !== "prepared") return; + const issueId = selected.nativeIssueId ?? selected.issueId ?? selected.contextSnapshot?.issueId; + if (typeof issueId !== "string") return; + return db.transaction(async tx => { + const [task] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.id, issueId), eq(issues.companyId, selected.companyId), + )).for("update"); + if (!task) return; + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, selected.id), eq(nativeRunFinalizations.companyId, selected.companyId), + eq(nativeRunFinalizations.issueId, issueId), + )).for("update"); + const claim = options.restartRecovery; + const reference = readNativeWorkspaceSyncReference(selected.runnerProfileJson?.nativeWorkspaceSync)!; + const ownsClaim = Boolean(options.allowPreterminal && coordinator?.phase === "observed" && claim + && claim.runId === selected.id && claim.leaseOwner === coordinator.leaseOwner + && claim.controllerGeneration === coordinator.controllerGeneration && claim.providerAttempt === coordinator.attempt + && coordinator.leaseExpiresAt && coordinator.leaseExpiresAt.getTime() > Date.now() + && (claim.kind !== "reattach_remote_runner" || (claim.remote.providerLeaseId === reference.providerLeaseId + && claim.remote.remoteCwd === reference.remoteCwd))); + if ((coordinator?.leaseOwner || (claim && coordinator?.phase === "observed")) && !ownsClaim) throw new NativeWorkspaceFinalizationBusyError(); + if (!coordinator || coordinator.resultId + || !(options.allowPreterminal ? ["observed", "terminal_failure"] : ["terminal_failure"]).includes(coordinator.phase)) return; + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, selected.id), eq(heartbeatRuns.companyId, selected.companyId), + )).for("update"); + if (!run || run.runtimeMode !== "native" || (!options.allowPreterminal && run.status !== "failed")) return; + const currentReference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); + if (!currentReference || currentReference.state !== "prepared" + || (claim?.kind === "reattach_remote_runner" && (claim.remote.providerLeaseId !== currentReference.providerLeaseId + || claim.remote.remoteCwd !== currentReference.remoteCwd))) return; + const [source] = await tx.select().from(environmentLeases).where(and( + eq(environmentLeases.id, currentReference.leaseId), eq(environmentLeases.companyId, run.companyId), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, currentReference.providerLeaseId), + )).limit(1); + if (!source || source.issueId !== issueId || source.metadata?.remoteCwd !== currentReference.remoteCwd + || (source.executionWorkspaceId && source.executionWorkspaceId !== currentReference.workspaceId)) return; + const alreadyRetained = hasRequiredWorkspaceRecovery(run.resultJson); + let retained = alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source); + if (!retained || !hasRequiredWorkspaceRecovery(retained.resultJson)) return; + if (options.allowPreterminal && coordinator.phase === "observed") { + const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", + leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null, + recoveryState: "blocked", failureCode: coordinator.failureCode ?? "native_continuation_requires_reconciliation", updatedAt: new Date(), + }).where(and(eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.companyId, run.companyId), + eq(nativeRunFinalizations.issueId, issueId), eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId), + ownsClaim && claim ? and(eq(nativeRunFinalizations.leaseOwner, claim.leaseOwner), + eq(nativeRunFinalizations.controllerGeneration, claim.controllerGeneration), eq(nativeRunFinalizations.attempt, claim.providerAttempt), + gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`)) : isNull(nativeRunFinalizations.leaseOwner), + )).returning({ id: nativeRunFinalizations.runId }); + if (!settled) throw new NativeWorkspaceFinalizationBusyError(); + [retained] = await tx.update(heartbeatRuns).set({ nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date() }) + .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))).returning(); + } + if (!alreadyRetained) await logActivity(tx as unknown as Db, { + companyId: run.companyId, actorType: "system", actorId: "native-workspace-recovery", + action: "issue.workspace_recovery_required", entityType: "issue", + entityId: issueId, + agentId: run.agentId, runId: run.id, + details: { runId: run.id, leaseId: source.id, reason: "native_execution_failed_before_workspace_export" }, + }); + return retained; + }); +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index dedc7ad65ca..9e88e5b6314 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -37,6 +37,7 @@ import { readNativeLocalProcessStop, PROCESS_START_REQUESTED } from "../native-l import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { resolveConnectorAssignments, isConnectorSkill } from "../connector-runtime.js"; import { activeNativeExecutionRunIds } from "./native-active-runs.js"; +import { preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; import { boundedExecutionCleanup, EXECUTION_CONTROL_DEADLINE_MS, @@ -9298,7 +9299,7 @@ async function executePaperclipNativeSessionWithinScope( gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), ), ) - .returning({ runId: nativeRunFinalizations.runId }) + .returning({ runId: nativeRunFinalizations.runId, resultId: nativeRunFinalizations.resultId }) .then((rows) => rows[0] ?? null); if (!updated) throw new Error("native_session_lease_lost"); const [runBeforeWrite] = await tx @@ -9306,7 +9307,7 @@ async function executePaperclipNativeSessionWithinScope( .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.execution.binding.runId)) .for("update"); - const [updatedRun] = await tx + let [updatedRun] = await tx .update(heartbeatRuns) .set({ // An authentication timeout does not prove the retained runner or @@ -9329,6 +9330,9 @@ async function executePaperclipNativeSessionWithinScope( }) .where(eq(heartbeatRuns.id, input.execution.binding.runId)) .returning(); + if (updatedRun && exhausted && !ownershipUnverified && !updated.resultId) { + updatedRun = await preserveNativeFailedWorkspace(tx as unknown as Db, updatedRun) ?? updatedRun; + } if ( updatedRun && runBeforeWrite &&