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 &&