diff --git a/doc/task-recovery.md b/doc/task-recovery.md index cac5d7aff46..55cebac9f1b 100644 --- a/doc/task-recovery.md +++ b/doc/task-recovery.md @@ -38,7 +38,14 @@ repeating the failed step. A Retry never silently drops the agent's work: Either choice consumes the generation's one alternative attempt. Preparing the same attempt again after a restart reuses the directory that attempt already -created and bound; it does not ask for another fresh directory. +created and bound; it does not ask for another fresh directory. A native +restart re-prepares from its admitted workspace. When that is the task's +current task directory, it keeps the directory strategy; a task directory is +never the base of a new git worktree. A git worktree base is always the +project's checkout or the task's own worktree. If a restart's re-preparation +still fails before the provider starts, the run settles its native coordinator +as a terminal failure, so a later Retry is admitted instead of waiting on an +open recovery claim. The existing exact-run retry endpoint retains its source-bound contract. Connected conversations use their existing inbound authorization owner. A board diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c7bbd976b47..9c85c7515c5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -23,6 +23,7 @@ import { readUpgradeReadiness } from "./heartbeat/upgrade-readiness.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; +import { settleFailedNativeRecoveryPreparation } from "./native-runtime/native-recovery-preparation-failure.js"; import { promptInjectionPreventionService } from "./prompt-injection-prevention.js"; import { createHeartbeatAgentStatus } from "./heartbeat/agent-status.js"; import { heartbeatExternalProvider, nativeRuntimeHasInstructionFilesystem } from "./heartbeat/external-provider.js"; @@ -3810,6 +3811,11 @@ export function heartbeatService( }); legacyPreparationFailedBeforeDispatch = !legacyAdapterEntered && !nativeDispatchStarted; } + // A failed re-preparation must not keep its recovery claim open. + if (runOptions.nativeRestartRecovery && !legacyAdapterEntered && !nativeDispatchStarted) + await settleFailedNativeRecoveryPreparation(db, { companyId: run.companyId, runId: run.id, + claim: runOptions.nativeRestartRecovery }).catch(error => + logger.warn({ err: error, runId: run.id }, "Failed native recovery preparation could not be settled")); } } finally { let caseHandoffCleanupSettled = false; diff --git a/server/src/services/heartbeat/environment-selection.test.ts b/server/src/services/heartbeat/environment-selection.test.ts index b8a3981f7ad..d416429daea 100644 --- a/server/src/services/heartbeat/environment-selection.test.ts +++ b/server/src/services/heartbeat/environment-selection.test.ts @@ -13,6 +13,7 @@ import { instanceSettingsService } from "../instance-settings.js"; import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js"; import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js"; import { selectHeartbeatEnvironment, type HeartbeatEnvironmentSelectionInput } from "./environment-selection.js"; +import { materializeIsolatedTaskDirectory } from "../isolated-task-directory.js"; const restoration = vi.hoisted(() => ({ removed: false })); vi.mock("../git-workspace-restoration.js", async importOriginal => { @@ -96,6 +97,35 @@ describe.skipIf(!support.supported)("heartbeat environment selection boundary", expect(result.persistedNativeExecutionInput?.binding.executionWorkspaceId).toBe(original.id); }); + it.each(["bound", "issue_moved"] as const)( + "re-prepares a restarted fresh Retry attempt as its task directory, never as a git worktree base (RC-58, %s)", async binding => { + const { input, issue, workspace, recover } = await fixture(); + // A project-less task's fresh Retry attempt directory, created and bound by the first preparation. + await db.update(issues).set({ projectId: null }).where(eq(issues.id, issue.id)); + const attemptCwd = await materializeIsolatedTaskDirectory({ companyId: input.agent.companyId, issueId: issue.id, attemptId: input.run.id }); + const created = await executionWorkspaceService(db).create({ companyId: input.agent.companyId, projectId: null, + sourceIssueId: issue.id, name: "Fresh attempt", mode: "isolated_workspace", strategyType: "project_primary", + providerType: "local_fs", cwd: attemptCwd }); + if (!created) throw new Error("Attempt workspace fixture was not created"); + await db.update(issues).set({ executionWorkspaceId: created.id, executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "isolated_workspace" } }).where(eq(issues.id, issue.id)); + const issueRef = { ...input.task.issueRef!, projectId: null, executionWorkspaceId: created.id, + executionWorkspacePreference: "reuse_existing" }; + input.task.issueRef = issueRef; + input.task.executionProjectId = null; + input.policy.requestedExecutionWorkspaceMode = "isolated_workspace"; + input.policy.issueExecutionWorkspaceSettings = { mode: "isolated_workspace" }; + if (binding === "issue_moved") issueRef.executionWorkspaceId = (await workspace()).id; + // The restart re-prepares from the admitted native input; readiness is no longer "fresh". + recover(created.id); + input.task.freshWorkspace = null; + const result = await selectHeartbeatEnvironment(db, input); + expect(result.nativeRecoveryExecutionWorkspaceId).toBe(created.id); + expect(result.reusableExistingExecutionWorkspace?.id).toBe(created.id); + // Only the task's current binding may resume as a task directory. + expect(result.useIsolatedTaskDirectory).toBe(binding === "bound"); + }); + it("selects the saved execution workspace without inspecting live Git delivery state", async () => { const { input, workspace, recover } = await fixture(); const saved = await workspace(); diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index e4076e2c0eb..43a8422c90f 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -494,7 +494,11 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron ...(reusableExistingExecutionWorkspace?.providerType === "adapter_managed" ? [config.workspaceStrategy] : []), issueAssigneeOverrides?.adapterConfig?.workspaceStrategy, projectExecutionWorkspacePolicy?.workspaceStrategy, issueExecutionWorkspaceSettings?.workspaceStrategy]; - const boundTaskDirectory = !controllerOnly && !fresh && !nativeRecoveryExecutionWorkspaceId && issueId !== null && issueRef !== null && + // RC-58: a native restart re-prepares its admitted workspace. When that is + // the task's own bound directory (for example a fresh Retry attempt), it keeps + // its directory strategy; it never becomes the base of a new git worktree. + const boundTaskDirectory = !controllerOnly && !fresh && issueId !== null && issueRef !== null && + (!nativeRecoveryExecutionWorkspaceId || nativeRecoveryExecutionWorkspaceId === issueRef.executionWorkspaceId) && requestedShouldReuseExisting && requestedExecutionWorkspaceMode === "isolated_workspace" && !workspaceProjectContext?.hasWorkspace && !issueRef.projectWorkspaceId && taskWorkspaceIntent?.selection.kind !== "configured_source" && diff --git a/server/src/services/native-runtime/native-recovery-preparation-failure.test.ts b/server/src/services/native-runtime/native-recovery-preparation-failure.test.ts new file mode 100644 index 00000000000..11f26a5167d --- /dev/null +++ b/server/src/services/native-runtime/native-recovery-preparation-failure.test.ts @@ -0,0 +1,95 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, applyPendingMigrations, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, + issueRecoveryActions, issues, nativeRunFinalizations, type Db, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { admitExplicitNativeContinuation } from "../explicit-native-continuation.js"; +import { settleFailedNativeRecoveryPreparation } from "./native-recovery-preparation-failure.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RECOVERY_PREPARATION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +/** RC-58: a failed re-preparation after a restart must leave a working Retry. */ +describe.skipIf(!support.supported)("failed native recovery preparation", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + if (externalUrl) { + await applyPendingMigrations(externalUrl); + database = { connectionString: externalUrl, cleanup: async () => {} } as typeof database; + } else database = await startEmbeddedPostgresTestDatabase("native-recovery-preparation-"); + db = createDb(database.connectionString); + }, 90_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function seed() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), commentId = randomUUID(); + const leaseOwner = `boot:2:${randomUUID()}`; + await db.insert(companies).values({ id: companyId, name: "Recovery preparation", defaultResponsibleUserId: "board", + issuePrefix: `P${companyId.replaceAll("-", "").slice(0, 8)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native", role: "engineer", adapterType: "paperclip_runner", + status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Write the report", status: "blocked", assigneeAgentId: agentId }); + // The restart claimed the run (bootstrap_incomplete); its re-preparation then failed. + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, runtimeMode: "native", + status: "failed", errorCode: "workspace_validation_failed", nativePhase: "observed", contextSnapshot: { issueId }, + startedAt: new Date(Date.now() - 120_000), finishedAt: new Date(Date.now() - 60_000) }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed", attempt: 1, + leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), controllerGeneration: 2, recoveryState: "bootstrap_incomplete" }); + // Server-authored process trail: the first bootstrap's runner was verified stopped. + await db.insert(heartbeatRunEvents).values([ + { companyId, runId, agentId, seq: 1, eventType: "native.process_start_requested", stream: "system" }, + { companyId, runId, agentId, seq: 2, eventType: "native.process_identity_recorded", stream: "system" }, + { companyId, runId, agentId, seq: 3, eventType: "native.local_process_stopped", stream: "system" }, + ]); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", + cause: "native_continuation_requires_reconciliation", fingerprint: runId, status: "active", ownerType: "board", + returnOwnerAgentId: agentId, nextAction: "Retry the task.", evidence: { runId } }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user", authorUserId: "board", + body: "Please continue.", createdAt: new Date() }); + return { companyId, agentId, issueId, runId, commentId, leaseOwner }; + } + + async function admissionBlocks(f: Awaited>) { + const blocked: string[] = []; + await db.transaction(async tx => admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: f.companyId, + issueId: f.issueId, agentId: f.agentId, actorId: "board", actorType: "user", reason: "issue_commented", + commentId: f.commentId, successorRunId: randomUUID(), dryRun: true, onBlocked: reason => blocked.push(reason) })); + return blocked; + } + + it("settles its own recovery claim so a later Retry is no longer held by the controller", async () => { + const f = await seed(); + expect(await admissionBlocks(f)).toEqual(["controller_settling"]); + expect(await settleFailedNativeRecoveryPreparation(db, { companyId: f.companyId, runId: f.runId, + claim: { kind: "bootstrap_incomplete", leaseOwner: f.leaseOwner } })).toBe(true); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator).toMatchObject({ phase: "terminal_failure", leaseOwner: null, resultId: null, + failureCode: "workspace_validation_failed" }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run).toMatchObject({ status: "failed", nativePhase: "terminal_failure" }); + expect(await admissionBlocks(f)).not.toContain("controller_settling"); + // Settling is idempotent. + expect(await settleFailedNativeRecoveryPreparation(db, { companyId: f.companyId, runId: f.runId, + claim: { kind: "bootstrap_incomplete", leaseOwner: f.leaseOwner } })).toBe(false); + }, 30_000); + + it.each(["other_owner", "reattach_claim", "run_not_failed", "finalizing_elsewhere"] as const)( + "leaves a coordinator it does not own unchanged: %s", async kind => { + const f = await seed(); + if (kind === "run_not_failed") await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }) + .where(eq(heartbeatRuns.id, f.runId)); + if (kind === "finalizing_elsewhere") await db.update(nativeRunFinalizations).set({ phase: "committing" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const claim = { kind: kind === "reattach_claim" ? "reattach_existing_runner" as const : "resume_dead_runner" as const, + leaseOwner: kind === "other_owner" ? `boot:3:${randomUUID()}` : f.leaseOwner }; + const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await settleFailedNativeRecoveryPreparation(db, { companyId: f.companyId, runId: f.runId, claim })).toBe(false); + const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(after).toEqual(before); + }, 30_000); +}); diff --git a/server/src/services/native-runtime/native-recovery-preparation-failure.ts b/server/src/services/native-runtime/native-recovery-preparation-failure.ts new file mode 100644 index 00000000000..e82e7750b2b --- /dev/null +++ b/server/src/services/native-runtime/native-recovery-preparation-failure.ts @@ -0,0 +1,48 @@ +import { and, eq, isNull } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; +import type { NativeRestartRecoveryClaim } from "./native-restart-recovery.js"; + +/** Claims whose previous runner process is verified gone. A reattach claim + * still has a live runner, so its coordinator keeps its own recovery owner. */ +const SETTLEABLE_CLAIMS = new Set(["bootstrap_incomplete", "resume_dead_runner"]); + +/** + * RC-58: a restart recovery claims the run's native coordinator, then prepares + * the run again. When that preparation fails before any provider dispatch, the + * run is already failed but the claim would keep the coordinator `observed` + * and owned. That reads as unfinished native accounting and an unsettled + * controller, so a later Retry can never be admitted. Settle exactly this + * claim as a terminal failure. A coordinator that another owner holds, or one + * with a result, is left unchanged. Returns whether this call settled it. + */ +export async function settleFailedNativeRecoveryPreparation(db: Db, input: { + companyId: string; runId: string; claim: Pick; +}): Promise { + if (!SETTLEABLE_CLAIMS.has(input.claim.kind)) return false; + return db.transaction(async tx => { + const [run] = await tx.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))) + .for("update"); + if (run?.status !== "failed") return false; + // Register the work before the writes it announces. A spare wake when no + // coordinator matches below is harmless. + await notifyExecutionRunWork(tx as unknown as Db, { companyId: input.companyId }); + const now = new Date(); + const [settled] = await tx.update(nativeRunFinalizations).set({ + phase: "terminal_failure", leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: null, controlDeadlineAt: null, + failureCode: run.errorCode ?? "native_recovery_preparation_failed", + failureDetail: { recoveryPreparationFailed: { claim: input.claim.kind, errorCode: run.errorCode }, + nextAction: "Preparation failed before the provider was started again. Retry the task." }, + recoveryState: "blocked", updatedAt: now, + }).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.phase, "observed"), eq(nativeRunFinalizations.leaseOwner, input.claim.leaseOwner), + isNull(nativeRunFinalizations.resultId), + )).returning({ runId: nativeRunFinalizations.runId }); + if (!settled) return false; + await tx.update(heartbeatRuns).set({ nativePhase: "terminal_failure", nativePhaseUpdatedAt: now, executionControlDeadlineAt: null }) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))); + return true; + }); +}