diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 3546241c639..fb4754c95e4 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1369,6 +1369,16 @@ recording the proof and successor lineage. Retained provider files are not edite Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Automatic attempts retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. +A failed, unowned native sandbox retry can outlive an explicit Retry that already +completed successfully. Startup recovery and ordinary retry admission retire that +old retry only with the same company, task and agent, a linked completed successor, +and the exact prepared source lease. They atomically register file preservation +without replaying provider work or changing the current task. A live owner, +recorded process or mismatched source remains held. Retirement does not prove +termination or saved files; the existing stop-and-save worker must establish both +before workspace selection can proceed. Both callers hold the task, coordinator +and source-run locks throughout retirement and preservation registration. + When a same-run retry acquires a replacement plugin task lease, the executor reloads the prior task's durable Runner checkpoint before checking local lifecycle evidence. This includes retries after a provider turn omitted its diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 48329cb29db..3456b5c5793 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -7,6 +7,9 @@ Feature ID: `recovery`. Native finalization deadline reconciliation records unfinished workspace preservation through the existing server-owned recovery path. It grants no additional agent API, provider-stop authority or permission to replay work. +The server also retires an unowned failed sandbox retry after its linked explicit +Retry completes. This only registers exact-source preservation; it neither changes +the current task nor supplies stop or saved-file proof. Remote process probes remain server-owned. Confirmed exit requires `ESRCH` or a zombie with the saved generation. Unknown adopted-process ownership rejects the diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 6969e047fce..90126dacd6c 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -46,6 +46,9 @@ Agent access: [companion](../agent-access-map/recovery.md). New messages and orphan cleanup can register missing preservation automatically and wake the existing worker. Saved input still waits for verified stop and file persistence, then enters ordinary admission once. + A completed explicit Retry also retires its failed, unowned sandbox predecessor's + stale retry and registers preservation atomically. Current task state and live or + mismatched owners remain protected; workspace switching still waits for cleanup. - `unknown-outcome`: `recovery/unknown-outcome` — uncertain external side effects require inspection and an explicit decision, not automatic replay. - `user-continuation`: `recovery/user-continuation` — saved user messages after a stop are reconsidered once 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 index 37492f913c7..eb9425f715e 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -10,6 +10,8 @@ import { preserveCancelledNativeWorkspaces, preserveNativeFailedWorkspace, prese import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-workspace-finalization-ownership.js"; import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; +import { claimNativeRestartRecoveries } from "./native-restart-recovery.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; const externalUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim(); const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); @@ -51,6 +53,82 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes 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)); + async function supersededRetry() { + const f = await seed(); + await db.update(heartbeatRuns).set({ nativePhase: "retryable_failure", startedAt: new Date(1), finishedAt: new Date(2) }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1, + recoveryState: "resuming_session", nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [successor] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, nativeIssueId: f.issueId, runtimeMode: "native", status: "succeeded", nativePhase: "committed", + startedAt: new Date(3), finishedAt: new Date(4), contextSnapshot: { issueId: f.issueId, + previousRunId: f.runId, taskRetryRequestId: randomUUID() } }).returning(); + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)); + return { ...f, successor }; + } + + it.each(["restart", "resume"])("retires a superseded failed retry and preserves its files during %s recovery", async path => { + const f = await supersededRetry(); + const beforeTask = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + const recover = () => path === "restart" ? claimNativeRestartRecoveries({ db, runIds: [f.runId], restartKind: "hard" }) + : claimNativeSessionResumptions({ db, runIds: [f.runId], runnerInstanceId: "recovery" }); + await recover(); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "terminal_failure", attempt: 1, leaseOwner: null, nextAttemptAt: null, + failureCode: "native_retry_superseded", failureDetail: { successorRunId: f.successor.id } }); + expect(await readRun(f.runId)).toMatchObject({ status: "failed", nativePhase: "terminal_failure", + resultJson: { originalDiagnostic: true, workspaceRestoreRecovery: { leaseIds: [f.lease.id] } } }); + expect(await readLease(f.lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(beforeTask); + const retainedActions = await actions(f.issueId); + expect(retainedActions).toHaveLength(1); + await recover(); + expect(await actions(f.issueId)).toEqual(retainedActions); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2); + }); + + it.each(["live_owner", "process", "current_execution", "unfinished_successor", "unlinked_successor", "foreign_successor", "mismatched_source"])( + "keeps an unverified superseded retry held: %s", async scenario => { + const f = await supersededRetry(); + if (scenario === "live_owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "live-controller", + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "process") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "current_execution") await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + if (scenario === "unfinished_successor") await db.update(heartbeatRuns).set({ status: "running", nativePhase: "observed", finishedAt: null }) + .where(eq(heartbeatRuns.id, f.successor.id)); + if (scenario === "unlinked_successor") await db.update(heartbeatRuns).set({ contextSnapshot: { previousRunId: randomUUID(), taskRetryRequestId: randomUUID() } }) + .where(eq(heartbeatRuns.id, f.successor.id)); + if (scenario === "foreign_successor") { + const foreign = await seed(); + await db.update(heartbeatRuns).set({ contextSnapshot: {} }).where(eq(heartbeatRuns.id, f.successor.id)); + await db.insert(heartbeatRuns).values({ companyId: foreign.companyId, agentId: foreign.agentId, + issueId: foreign.issueId, nativeIssueId: foreign.issueId, runtimeMode: "native", status: "succeeded", nativePhase: "committed", + startedAt: new Date(3), finishedAt: new Date(4), contextSnapshot: { previousRunId: f.runId, taskRetryRequestId: randomUUID() } }); + } + if (scenario === "mismatched_source") await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }) + .where(eq(environmentLeases.id, f.lease.id)); + const beforeRun = await readRun(f.runId), beforeLease = await readLease(f.lease.id); + await claimNativeSessionResumptions({ db, runIds: [f.runId], runnerInstanceId: "recovery" }); + expect(await readRun(f.runId)).toEqual(beforeRun); + expect(await readLease(f.lease.id)).toEqual(beforeLease); + expect(await actions(f.issueId)).toEqual([]); + }); + + it("rolls back retry retirement together with file preservation", async () => { + const f = await supersededRetry(); + const beforeRun = await readRun(f.runId), beforeLease = await readLease(f.lease.id); + const beforeCoordinator = (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]; + await expect(db.transaction(async tx => { + await claimNativeSessionResumptions({ db: tx as unknown as typeof db, runIds: [f.runId], runnerInstanceId: "recovery" }); + expect((await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("terminal_failure"); + throw new Error("rollback retry retirement"); + })).rejects.toThrow("rollback retry retirement"); + expect(await readRun(f.runId)).toEqual(beforeRun); + expect(await readLease(f.lease.id)).toEqual(beforeLease); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(beforeCoordinator); + expect(await actions(f.issueId)).toEqual([]); + }); + it("keeps another orphan recoverable while a task lock delays preservation", async () => { const busy = await seed(), free = await seed(); let unlock!: () => void, ready!: () => void; diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 283aac76b2d..fda887b8eb8 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -63,6 +63,7 @@ import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold-guard.js"; import { readRunCancellation } from "../run-cancellation.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; +import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; export type NativeReconciliationFacts = { equivalentAttentionFamily?: boolean; @@ -392,7 +393,7 @@ export async function claimNativeSessionResumptions(input: { const [issue] = await tx.select().from(issues).where(and( eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId), )).for("update").limit(1); - if (!issue || !["todo", "in_progress", "in_review"].includes(issue.status)) return false; + if (!issue) return false; const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, candidate.runId), eq(nativeRunFinalizations.issueId, issue.id), @@ -402,6 +403,10 @@ export async function claimNativeSessionResumptions(input: { eq(heartbeatRuns.nativeIssueId, issue.id), )).for("update").limit(1); if (!run || !coordinator) return false; + if (await retireSupersededNativeRetry(tx as unknown as Db, { + run, coordinator, issueExecutionRunId: issue.executionRunId, now, + })) return false; + if (!["todo", "in_progress", "in_review"].includes(issue.status)) return false; const row = { run, coordinator }; if ( row.run.runtimeMode !== "native" || diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 0c854b7fe13..a906c14c56e 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -26,6 +26,7 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; export type NativeControllerIdentity = { bootId: string; @@ -588,6 +589,9 @@ export async function claimNativeRestartRecoveries(input: { reason: "result_already_persisted", } as const; } + if (!input.liveRunExecutions?.has(row.run.id) && await retireSupersededNativeRetry(tx as unknown as Db, { + run: row.run, coordinator: row.coordinator, issueExecutionRunId: row.issueExecutionRunId, now, + })) return { kind: "blocked", runId: row.run.id, reason: "native_retry_superseded" } as const; if (row.issueExecutionRunId !== row.run.id) { return { kind: "blocked", diff --git a/server/src/services/native-runtime/native-superseded-retry.ts b/server/src/services/native-runtime/native-superseded-retry.ts new file mode 100644 index 00000000000..e98672826b0 --- /dev/null +++ b/server/src/services/native-runtime/native-superseded-retry.ts @@ -0,0 +1,56 @@ +import { and, desc, eq, gt, isNotNull, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; +import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; +import { preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; + +export async function retireSupersededNativeRetry(db: Db, input: { + run: typeof heartbeatRuns.$inferSelect; + coordinator: typeof nativeRunFinalizations.$inferSelect; + issueExecutionRunId: string | null; + now: Date; +}): Promise { + const { run, coordinator, now } = input; + if (run.runtimeMode !== "native" || run.status !== "failed" || !run.startedAt || !run.finishedAt + || run.processPid !== null || run.processGroupId !== null + || coordinator.phase !== "retryable_failure" || coordinator.resultId || coordinator.leaseOwner + || coordinator.companyId !== run.companyId || coordinator.issueId !== run.nativeIssueId + || coordinator.runId !== run.id || input.issueExecutionRunId === run.id + || run.runnerProfileJson?.nativeWorkspaceFinalizationOwner || nativeRunHasNoWorkspaceAccess(run)) return false; + const reference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); + if (reference?.state !== "prepared") return false; + + const [successor] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.nativeIssueId, coordinator.issueId), + eq(heartbeatRuns.agentId, run.agentId), eq(heartbeatRuns.runtimeMode, "native"), + eq(heartbeatRuns.status, "succeeded"), eq(heartbeatRuns.nativePhase, "committed"), + gt(heartbeatRuns.startedAt, run.finishedAt), isNotNull(heartbeatRuns.finishedAt), + sql`${heartbeatRuns.contextSnapshot}->>'previousRunId' = ${run.id}`, + sql`nullif(${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId', '') is not null`, + )).orderBy(desc(heartbeatRuns.startedAt)).limit(1); + if (!successor) return false; + const [source] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.id, reference.leaseId), eq(environmentLeases.companyId, run.companyId), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.issueId, coordinator.issueId), + eq(environmentLeases.providerLeaseId, reference.providerLeaseId), + sql`${environmentLeases.metadata}->>'remoteCwd' = ${reference.remoteCwd}`, + sql`(${environmentLeases.executionWorkspaceId} is null or ${environmentLeases.executionWorkspaceId}::text = ${reference.workspaceId})`, + )).limit(1); + if (!source) return false; + + await notifyExecutionRunWork(db); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null, + leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: "blocked", + failureCode: "native_retry_superseded", failureDetail: { ...coordinator.failureDetail, + reason: "completed_explicit_retry", successorRunId: successor.id }, updatedAt: now, + }).where(eq(nativeRunFinalizations.runId, run.id)); + const [retired] = await db.update(heartbeatRuns).set({ nativePhase: "terminal_failure", + nativePhaseUpdatedAt: now, updatedAt: now, + }).where(eq(heartbeatRuns.id, run.id)).returning(); + const preserved = await preserveNativeFailedWorkspace(db, retired!); + if (!preserved || !hasRequiredWorkspaceRecovery(preserved.resultJson)) { + throw new Error("superseded_native_retry_preservation_not_recorded"); + } + return true; +} diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index a0a462e6056..3cb3089b196 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -40,6 +40,8 @@ import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { registerWorkspaceRecoveryWork } from "./workspace-recovery-work.js"; import { heartbeatService } from "./heartbeat.js"; import { hasVerifiedWorkspaceRepair, readWorkspaceRecovery } from "./workspace-recovery-receipt.js"; +import { claimNativeSessionResumptions } from "./native-runtime/native-finalization-reconciler.js"; +import { executionWorkspaces } from "@paperclipai/db"; import { getExecutionBlocker } from "./execution-blocker.js"; import { issueRoutes } from "../routes/issues.js"; import { errorHandler } from "../middleware/index.js"; @@ -137,6 +139,42 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + it("unblocks a pending workspace selection only after a superseded retry's real files are preserved", async () => { + const f = await seed(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const source = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: null, nativePhase: "retryable_failure", + startedAt: new Date(1), finishedAt: new Date(2) }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1, + recoveryState: "resuming_session", nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(environmentLeases).set({ status: "active", leasePolicy: "ephemeral", + releasedAt: null, cleanupStatus: null, metadata: f.lease.metadata }).where(eq(environmentLeases.id, f.lease.id)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, + issueId: f.issueId, nativeIssueId: f.issueId, runtimeMode: "native", status: "succeeded", nativePhase: "committed", + startedAt: new Date(3), finishedAt: new Date(4), contextSnapshot: { previousRunId: f.runId, taskRetryRequestId: randomUUID() } }); + await db.insert(executionWorkspaces).values({ id: f.issueId, companyId: f.companyId, sourceIssueId: f.issueId, + name: "Shared files", mode: "shared_workspace", strategyType: "project_primary", cwd: f.host }); + const actor = { type: "board" as const, source: "local_implicit" as const, isInstanceAdmin: true }; + const workspaces = executionWorkspaceService(db); + await workspaces.selectTaskWorkspace({ companyId: f.companyId, issueId: f.issueId, actor, + selection: { kind: "existing", workspaceId: f.issueId }, expectedBindingRevision: 0, requestKey: randomUUID() }); + const admission = { companyId: f.companyId, issueId: f.issueId, actor, runId: randomUUID() }; + await expect(workspaces.applyPendingTaskWorkspaceSelection(admission)).rejects.toThrow("Previous task run still owns workspace recovery"); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "recovery", runIds: [f.runId] })).toEqual([]); + await expect(workspaces.applyPendingTaskWorkspaceSelection(admission)).rejects.toThrow("Previous task run still owns workspace recovery"); + const [preservation] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await recoverRetainedWorkspace({ ...f.input, actionId: preservation.id }); + expect(readSystemRecovery((await action(preservation.id)).evidence)?.state).toBe("complete"); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); + expect(await workspaces.applyPendingTaskWorkspaceSelection(admission)).toBe(true); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task).toMatchObject({ status: "blocked", assigneeAgentId: source.agentId, executionWorkspaceId: f.issueId, + workspacePendingSelection: null, executionWorkspaceSettings: { mode: "shared_workspace" } }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(2); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.companyId, f.companyId))).toHaveLength(0); + }, 30_000); + it.each(["HTTP Retry", "new comment", "automatic cleanup"] as const)("repairs an unregistered historical deadline failure through %s without admitting work before verified stop", async entry => { const f = await seed("native", true, "keep_running"); const source = await run(f.runId);