diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index fe985686b95..24feeca3dbd 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1352,6 +1352,13 @@ fresh native run identities. Unmarked preparations from older servers remain held during rolling upgrades, because those servers cannot safely yield to a recovery owner. Revisiting an existing preparation cannot add that marker. +Restart recovery checks the heartbeat process's live execution registry before +claiming an observed run. Native input can exist before workspace synchronization +has saved its descriptor or the executor has claimed its lease. A live preparation +keeps ownership throughout that interval. Recovery leaves its coordinator unchanged +and does not dispatch a second executor. An inactive run still requires the normal +ownership, workspace integrity, and provider evidence checks. + A preparation that resumes after another execution acquired the coordinator lease yields through the controller-detachment path. It does not fail the shared run, release its environment, or start another provider. Lease expiry alone does diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 50699dee967..995ab7c4711 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -19,6 +19,9 @@ Agent access: [companion](../agent-access-map/recovery.md). progress, and next action without requiring a raw-log interpretation. - `bounded-continuation`: `recovery/bounded-continuation` — supported interrupted work continues with preserved conversation context and bounded attempts, without blindly replaying tools. + Restart recovery respects live heartbeat preparation before workspace + synchronization finishes. PostgreSQL regressions cover unchanged ownership, + no duplicate dispatch, and later recovery after preparation leaves the registry. - `source-action`: `recovery/source-action` — source-scoped recovery actions remain attached to the task; independent repair work may have its own issue and return owner. - `exhaustion`: `recovery/exhaustion` — an exhausted infrastructure attempt offers ordinary Retry. diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index 34139757468..6b5eb940732 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -11,6 +11,7 @@ import { createHeartbeatRunState } from "./run-state.js"; import type { HotRestartIntent } from "../hot-restart.js"; import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js"; import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js"; +import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-recovery.js"; import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js"; import { logger } from "../../middleware/logger.js"; @@ -23,7 +24,7 @@ const native = vi.hoisted(() => ({ finalizations: vi.fn(async (_db: Db, _ids?: string[], _options?: unknown) => undefined), cleanups: vi.fn(async () => undefined), drainMaintenance: vi.fn<() => Promise>(async () => undefined), - claims: vi.fn<() => Promise>(async () => []), + claims: vi.fn(async () => []), })); const hotRestart = vi.hoisted(() => ({ read: vi.fn<() => Promise>(async () => null), @@ -608,6 +609,33 @@ describePostgres("heartbeat recovery module ownership", () => { await vi.waitFor(() => expect(deps.activeRunExecutionPromises.size).toBe(0)); }); + it("does not dispatch recovery for a run still preparing in this heartbeat process", async () => { + const run = await seedRun(); + await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, run.agentId)); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, run.issueId!)); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "observed" }); + const deps = callbacks(db); + deps.activeRunExecutions.add(run.id); + deps.getRun.mockImplementation(async id => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0] ?? null); + native.claims.mockImplementationOnce(claimNativeRestartRecoveries); + const before = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + try { + const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(); + expect(result.awaitingEvidenceRunIds).toEqual([run.id]); + expect(result.claims).toEqual([]); + expect(deps.executeRun).not.toHaveBeenCalled(); + expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id))).toEqual(before); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, run.issueId!))).toHaveLength(0); + expect((await deps.getRun(run.id))?.status).toBe("running"); + } finally { + deps.activeRunExecutions.delete(run.id); + await Promise.all([...deps.activeRunExecutionPromises]); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + } + }); + it("does not claim or execute recovery when persisted finalization fails", async () => { const deps = callbacks(db); native.finalizations.mockRejectedValueOnce(new Error("result settlement failed")); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index a5197485881..af13219764c 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -959,6 +959,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } const dispositions = await claimNativeRestartRecoveries({ db, + liveRunExecutions: activeRunExecutions, restartKind, recoveryRequestId: intent?.recoveryRequestId ?? null, coordinatedPreviousController: intent diff --git a/server/src/services/heartbeat/runtime-selection.test.ts b/server/src/services/heartbeat/runtime-selection.test.ts index 737f69beffa..8badce2038a 100644 --- a/server/src/services/heartbeat/runtime-selection.test.ts +++ b/server/src/services/heartbeat/runtime-selection.test.ts @@ -11,6 +11,7 @@ import { buildNativeRuntimeContext } from "../native-runtime/runtime-context.js" import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js"; import { prepareNativeWorkspaceSync } from "../native-runtime/native-workspace-sync.js"; import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mode.js"; +import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-recovery.js"; import { canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js"; import { selectHeartbeatRuntime, type HeartbeatRuntimeSelectionInput } from "./runtime-selection.js"; @@ -74,6 +75,31 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () = const coordinators = (id: string) => db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, id)); const sandbox = { kind: "remote" as const, transport: "sandbox" as const, remoteCwd: "/workspace", reusableLeaseConfigured: false }; + it("keeps recovery out of live preparation before the workspace descriptor exists", async () => { + const { input, run, issue } = await fixture(); + input.workspace.executionTarget = sandbox; + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, issue.id)); + const liveRunExecutions = new Set([run.id]); + const recoveryInput = { db, restartKind: "hard" as const, runIds: [run.id], liveRunExecutions }; + vi.mocked(prepareNativeWorkspaceSync).mockImplementationOnce(async () => { + const saved = await read(run.id); + expect(saved.runnerProfileJson).toHaveProperty("nativeExecutionInput"); + expect(saved.runnerProfileJson).not.toHaveProperty("nativeWorkspaceSync"); + const before = await coordinators(run.id); + expect(await claimNativeRestartRecoveries(recoveryInput)).toEqual([ + { kind: "awaiting_evidence", runId: run.id, reason: "current_controller_preparing" }, + ]); + expect(await coordinators(run.id)).toEqual(before); + expect((await read(run.id)).status).toBe("running"); + return null; + }); + expect((await selectHeartbeatRuntime(db, input)).selected).toBe(true); + liveRunExecutions.delete(run.id); + expect(await claimNativeRestartRecoveries(recoveryInput)).toEqual([ + expect.objectContaining({ kind: "bootstrap_incomplete", runId: run.id, controllerGeneration: 1 }), + ]); + }); + it("persists the native input, completion contract, and coordinator before transferring cleanup ownership", async () => { const { input, run } = await fixture(); input.workspace.onCheckpoint = vi.fn(async () => {}); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 25e29408bef..1e9006be369 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -465,6 +465,7 @@ export async function claimNativeRestartRecoveries(input: { restartKind: NativeRestartKind; recoveryRequestId?: string | null; runIds?: string[]; + liveRunExecutions?: Pick, "has">; now?: Date; limit?: number; coordinatedPreviousController?: { @@ -580,6 +581,14 @@ export async function claimNativeRestartRecoveries(input: { } as const; } + if (input.liveRunExecutions?.has(row.run.id)) { + return { + kind: "awaiting_evidence", + runId: row.run.id, + reason: "current_controller_preparing", + } as const; + } + const takeover = await evaluateNativeControllerTakeover({ owner: row.coordinator, now,