diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7d50d479125..a67dda11b8d 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3030,7 +3030,12 @@ resumption path. This does not consume a provider attempt or stop the retained sandbox. A locked run/coordinator check rejects Stop, changed ownership or any provider progress. Startup restores even an overdue wait to the same scheduler; ordinary resumption rechecks current permissions and task authority. No idle -scan is added. Per-turn continuation reconstructs a missing suspended-state +scan is added. The same wait applies when a compatible successful predecessor +still has a live finalization lease, even with full local controller projections. +An expired or undated owner, incomplete canonical commands and incompatible +execution bindings retain their existing reconciliation holds. Once the owner +releases its lease, the existing resumption claim admits the same run once. +Per-turn continuation reconstructs a missing suspended-state receipt from the admitted sandbox or prepared-task owner, including after a previous read failed with only a PRP projection written. diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 0f06c319e23..c8a2c5037b2 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -23,6 +23,10 @@ The `warm-suspended` case suspends the first turn, resumes the second turn, then keeps that controller warm for the third turn. It must retire the first turn's local suspension receipt so the third turn can reuse the resumed Runner and provider history without a second process or repeated turn. +The `finalizer-wait` case gives the completed predecessor a live finalization +lease. The next run must wait without starting a provider turn or consuming an +attempt. After lease release, one durable resumption claim continues that same +run and preserves provider history through the third turn. The same-server case also starts preparation lease renewal during recovery and holds instruction collection open past a renewal tick after coordinator release. The executor must acquire renewal before preparation releases its reference, then diff --git a/server/src/__tests__/external-object-refresh-work.test.ts b/server/src/__tests__/external-object-refresh-work.test.ts index b07e4c3cf8c..d1d9b7a0d01 100644 --- a/server/src/__tests__/external-object-refresh-work.test.ts +++ b/server/src/__tests__/external-object-refresh-work.test.ts @@ -16,6 +16,7 @@ import { instanceSettingsService } from "../services/instance-settings.js"; import { notifyExternalObjectRefresh } from "../services/external-object-refresh-notifications.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -92,7 +93,7 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { } throw new Error("refresh worker did not settle"); } - async function quiet() { await settle(() => coordinator!.nextWakeAt() === null); } + async function quiet() { await settle(() => coordinator!.nextWakeAt() === null && idleWorkSnapshot().active === 0); } function clock() { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); } async function change(companyId: string, patch: Partial) { await db.transaction(async tx => { @@ -601,7 +602,10 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: { await db.insert(externalObjectMentions).values(batch.map(objectId => ({ companyId: f.companyId, sourceIssueId: f.issueId, sourceKind: "description" as const, objectId }))); } - await start(svc); await settle(() => coordinator!.nextWakeAt() === future.getTime()); + await start(svc); + await vi.waitFor(() => { + expect(coordinator!.nextWakeAt()).toBe(future.getTime()); + }, { timeout: 10_000, interval: 5 }); const [incoming] = await db.insert(issues).values({ companyId: f.companyId, title: "Incoming", description: "https://example.com/new-object" }).returning(); const reads: Array<{ query: string; params: unknown[] }> = []; diff --git a/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts b/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts index 160be9a352e..16676581539 100644 --- a/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts +++ b/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts @@ -1,6 +1,8 @@ import { createDb } from "@paperclipai/db"; import { closeIdleWarmNativeSessionsForRestart, executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js"; import { dockerCommandRunner } from "./docker-command-runner.js"; +import { createServer } from "node:http"; +import { registerLocalRuntimeApiServer } from "../../runtime-api.js"; // Separate TypeScript fixture import time from actual controller admission. process.send?.({ ready: true }); @@ -15,7 +17,10 @@ process.once("message", async (input: { }) => { const target = input.options.runnerExecutionTarget; if (target?.kind !== "remote") throw new Error("Expected remote worker target"); + const api = createServer((_request, response) => { response.writeHead(404).end(); }); + registerLocalRuntimeApiServer(api); try { + await new Promise(resolve => api.listen(0, "127.0.0.1", resolve)); const runner = dockerCommandRunner(input.container); let inspectionOverridden = false; const execute = runner.execute.bind(runner); @@ -55,6 +60,9 @@ process.once("message", async (input: { process.send?.({ result: { ...result, inspectionOverridden } }); } catch (error) { process.send?.({ error: error instanceof Error ? error.stack : String(error) }); + } finally { + api.closeAllConnections(); + await new Promise(resolve => api.close(() => resolve())); } }); setInterval(() => undefined, 60_000); diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index 2accfe2a6ca..c4315273e81 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -4,16 +4,17 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agentTaskSessions, agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; -import { DurablePrpControlPlane, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { agentTaskSessions, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { DurablePrpControlPlane, parseNativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres.js"; import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; -import { createNativeControllerRecovery, deferNativeControllerRecovery, importNativeControllerRecovery, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { assertNativeRetainedSandboxBinding, createNativeControllerRecovery, deferNativeControllerRecovery, importNativeControllerRecovery, NativeControllerRecoveryWaitingError, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; import { claimNativeRestartRecoveries } from "./native-restart-recovery.js"; import { readNativeControllerRecoveryInventory } from "./native-controller-recovery-inventory.js"; +import { inspectRemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable discovery and import", () => { let database: Awaited>; @@ -132,6 +133,91 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now: nextDeadline })).toEqual([]); }); + it.each(["committed", "workspace_finalizing"])("defers a remote warm continuation while its %s predecessor has a finalization lease", async phase => { + const fixture = await seed(), runId = randomUUID(), environmentId = randomUUID(); + const previous = parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId, agentId, issueId: fixture.scope.issueId, runId: fixture.scope.runId, + executionWorkspaceId: randomUUID() }, + task: { identifier: "NCD-1", title: "Continue retained execution", description: null, + prompt: "Continue the saved work.", workMode: "standard" }, + workspace: { cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: fixture.scope.normalizedSessionId, driverKind: "codex_app_server", + protocolVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } }, + completionContract: { id: randomUUID(), sha256: "fixture", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Continue the saved work.", criteria: [{ id: "objective", requirement: "Save the result." }] } }, + credentialBindings: [], interactionResponses: [], + }); + const current = { ...previous, binding: { ...previous.binding, runId } }; + const checkpoint: PersistedNativeSession = { ...fixture.snapshot, driverKind: "codex_app_server", + providerSessionId: "retained-provider-session", activeTurnId: null, pendingRuntimeRequests: [] }; + const metadata = createNativeControllerRecovery(checkpoint, compatibility, { runId: fixture.scope.runId, + runnerInstanceId: fixture.native.runnerInstanceId, normalizedSessionId: fixture.scope.normalizedSessionId }); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.profile, nativeExecutionInput: previous, + sessionCheckpoint: checkpoint, nativeControllerRecovery: metadata } }).where(eq(heartbeatRuns.id, fixture.scope.runId)); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: fixture.scope.issueId, + nativeSessionId: fixture.scope.normalizedSessionId, status: "running", runtimeMode: "native", + runnerProfileJson: { nativeExecutionInput: current, sessionCheckpoint: { ...checkpoint, + identity: { ...checkpoint.identity, runId } } } }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: fixture.scope.issueId, runId, phase: "observed" }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, fixture.scope.issueId)); + await db.insert(environments).values({ id: environmentId, name: `Retained ${environmentId}`, driver: "sandbox" }); + await db.insert(environmentLeases).values({ companyId, environmentId, issueId: fixture.scope.issueId, + heartbeatRunId: fixture.scope.runId, provider: "daytona", providerLeaseId: "retained-sandbox", status: "released", + metadata: { remoteCwd: "/workspace" } }); + const deadline = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ phase, leaseOwner: "finalizer", leaseExpiresAt: deadline, + controllerPid: process.pid, controllerProcessStartedAt: new Date() }).where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + const identity = { runId: fixture.scope.runId, normalizedSessionId: fixture.scope.normalizedSessionId, + runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId, + turnId: fixture.native.turnId, itemId: fixture.native.itemId }; + const target = { kind: "remote", transport: "sandbox", environmentId, leaseId: randomUUID(), providerKey: "daytona", + remoteCwd: "/workspace", sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "retained-sandbox" } } as const; + const readSandbox = (override: Record = {}) => assertNativeRetainedSandboxBinding(db, + { ...fixture.scope, target: { ...target, ...override } as never }); + const inspect = (overrides: Partial[0]> = {}) => inspectRemoteWarmSessionRecovery({ db, execution: current, + target: target as never, + identity, readControlState: () => ({ identity, commands: [{ type: "turn.start", status: "completed" }], + committedEvents: [{ eventType: "run.terminal", envelope: identity }] }), + matchesExecution: saved => saved.binding.runId === fixture.scope.runId, + loadCheckpoint: () => checkpoint, canReuseProcess: () => true, ...overrides, + }); + const denied = "runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"; + await expect(inspect({ execution: { ...current, binding: { ...current.binding, companyId: randomUUID() } } })) + .rejects.toThrow(denied); + await expect(inspect({ matchesExecution: () => false })).rejects.toThrow(denied); + await expect(readSandbox({ remoteCwd: "/unrelated" })).rejects.toThrow("retained_sandbox_binding"); + for (const leaseExpiresAt of [null, new Date(Date.now() - 1_000)]) { + await db.update(nativeRunFinalizations).set({ leaseExpiresAt }).where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + await expect(inspect()).rejects.toThrow(denied); + await expect(readSandbox()).rejects.toThrow("retained_sandbox_binding"); + } + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: deadline }).where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, fixture.scope.runId)); + await expect(inspect()).rejects.toThrow(denied); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, fixture.scope.runId)); + await expect(readSandbox()).rejects.toMatchObject({ retryAt: deadline }); + const waiting = await inspect().catch(error => error); + expect(waiting).toBeInstanceOf(NativeControllerRecoveryWaitingError); + expect(waiting.retryAt).toEqual(deadline); + await deferNativeControllerRecovery(db, { ...current.binding, retryAt: waiting.retryAt }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]) + .toMatchObject({ phase: "retryable_failure", attempt: 0, nextAttemptAt: deadline, resultId: null }); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId] })).toEqual([]); + await db.update(nativeRunFinalizations).set({ phase: "committed", leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + await expect(readSandbox()).resolves.toBeUndefined(); + await expect(inspect({ readControlState: () => ({ identity, commands: [{ status: "pending" }], + committedEvents: [{ eventType: "run.terminal", envelope: identity }] }) })).rejects.toThrow(denied); + const recovered = await inspect(); + expect(recovered.identity).toEqual(identity); + expect(recovered.checkpoint).toBe(checkpoint); + expect(recovered.reuseProcess).toBe(true); + const claims = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now: deadline }); + expect(claims).toHaveLength(1); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "duplicate", runIds: [runId], now: deadline })).toEqual([]); + }); + it.each(["live-owner", "expired-owner", "provider-attempt", "stopped", "operator-stop"])("does not publish an admission wait after authority changes: %s", async scenario => { const fixture = await seed(), runId = randomUUID(); const scope = { ...fixture.scope, runId }; diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts index 4c5fd4ac419..223e5f69bdc 100644 --- a/server/src/services/native-runtime/native-controller-recovery.ts +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -138,11 +138,17 @@ export async function assertNativeRetainedSandboxBinding(db: Db, input: { eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.issueId, input.issueId), eq(environmentLeases.heartbeatRunId, input.runId), eq(environmentLeases.environmentId, target.environmentId), eq(environmentLeases.provider, target.providerKey), eq(environmentLeases.providerLeaseId, target.sandboxLeaseAcquisition.providerLeaseId))); - if (!owner || owner.leaseOwner || owner.phase !== "committed" || !owner.resultId || owner.recoveryState + if (!owner || !owner.resultId || owner.recoveryState || leases.length !== 1 || leases[0].metadata?.remoteCwd !== target.remoteCwd || !["active", "released", "expired"].includes(leases[0].status)) { throw new Error("native_controller_recovery_missing_evidence:retained_sandbox_binding"); } + if (owner.leaseOwner && owner.leaseExpiresAt && owner.leaseExpiresAt > new Date()) { + throw new NativeControllerRecoveryWaitingError(owner.leaseExpiresAt); + } + if (owner.leaseOwner || owner.phase !== "committed") { + throw new Error("native_controller_recovery_missing_evidence:retained_sandbox_binding"); + } } /** Only the existing bounded selector chooses history. Never search backwards diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index b7d3403eb9a..c90619e12c7 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -1,5 +1,7 @@ import { renewNativeSessionExecutionLease, startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js"; import { enableRunFailureReporting } from "../../__tests__/helpers/run-failure-reporting.js"; +import { NativeControllerRecoveryWaitingError } from "./native-controller-recovery.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; import { hasNativeLocalProcessStop } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; import { createRequire } from "node:module"; @@ -150,7 +152,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( expect(next.kind).toBe("bootstrap_incomplete"); }); - it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "warm-suspended", "stale-projection", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry", "diskless-warm-suspended", "diskless-warm-suspended-read-retry", "diskless-live-inspection"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { + it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "finalizer-wait", "warm-suspended", "stale-projection", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry", "diskless-warm-suspended", "diskless-warm-suspended-read-retry", "diskless-live-inspection"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { const hardRestart = !["warm", "same-server", "warm-suspended"].includes(scenario); const diskless = scenario.startsWith("diskless"); const perTurn = scenario.startsWith("diskless-per-turn"); @@ -408,6 +410,23 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( ...(perTurn ? { sessionCheckpoint: admission.checkpoint } : {}) } }).where(eq(heartbeatRuns.id, row.id)); }; await admitContinuation(nextOptions.execution); + let continuationLeaseOwner: string | undefined; + if (scenario === "finalizer-wait") { + const retryAt = new Date(Date.now() + 60_000); + await fixture.db.update(nativeRunFinalizations).set({ leaseOwner: "fixture-finalizer", leaseExpiresAt: retryAt }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + await expect(executePaperclipNativeSession(nextOptions)).rejects.toBeInstanceOf(NativeControllerRecoveryWaitingError); + const [waiting] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, nextRunId)); + expect(waiting).toMatchObject({ phase: "retryable_failure", attempt: 0, nextAttemptAt: retryAt }); + const calls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] }); + expect(calls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(1); + await fixture.db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + const [claim] = await claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId: next.runnerInstanceId, + runIds: [nextRunId], now: retryAt }); + expect(claim).toBeDefined(); + continuationLeaseOwner = claim.leaseOwner; + } if (scenario.endsWith("-read-retry")) { let rejectedReads = 0; await expect(executePaperclipNativeSession({ ...nextOptions, @@ -425,7 +444,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( // Retry with the partial PRP projection left by the failed read. No // fixture writes a suspended receipt or changes the surviving Runner. } - const second = diskless ? await executeInController(nextOptions) : await executePaperclipNativeSession(nextOptions); + const second = diskless ? await executeInController(nextOptions) : await executePaperclipNativeSession({ ...nextOptions, + ...(continuationLeaseOwner ? { leaseOwner: continuationLeaseOwner } : {}) }); expect(second.exitCode).toBe(0); if (scenario === "diskless-live-inspection") expect("inspectionOverridden" in second && second.inspectionOverridden).toBe(true); const historyAfter = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] }); diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index 893dfefaef6..ec63a72325e 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -3,7 +3,7 @@ import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunF import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { currentNativeControllerIdentity, evaluateNativeControllerTakeover } from "./native-restart-recovery.js"; -import { readNativeControllerRecovery } from "./native-controller-recovery.js"; +import { NativeControllerRecoveryWaitingError, readNativeControllerRecovery } from "./native-controller-recovery.js"; export type RemoteWarmSessionRecovery = { identity: Record; @@ -71,11 +71,14 @@ export async function inspectRemoteWarmSessionRecovery(input: { .from(heartbeatRuns).innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) .where(and(eq(heartbeatRuns.id, identity.runId), eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId))); - if (!prior || prior.run.status !== "succeeded" || prior.owner.phase !== "committed" || - !prior.owner.resultId || prior.owner.leaseOwner || !prior.owner.controllerPid || + if (!prior || prior.run.status !== "succeeded" || !prior.owner.resultId || !prior.owner.controllerPid || !prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny(); const previous = parseNativeExecutionInput(prior.run.runnerProfileJson?.nativeExecutionInput); if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny(); + if (prior.owner.leaseOwner && prior.owner.leaseExpiresAt && prior.owner.leaseExpiresAt > new Date()) { + throw new NativeControllerRecoveryWaitingError(prior.owner.leaseExpiresAt); + } + if (prior.owner.phase !== "committed" || prior.owner.leaseOwner) return deny(); const checkpoint = input.loadCheckpoint(previous); if (!checkpoint || checkpoint.identity.runId !== identity.runId || checkpoint.identity.companyId !== binding.companyId || checkpoint.identity.issueId !== binding.issueId ||