diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e4473596965..c60dd58e74c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3175,8 +3175,17 @@ these differences: cannot be compared keeps the rejection, for example `retirement_refused=runner_process_alive`. A later Retry checks again. +A warm sandbox owner gets the same retirement when the next run starts in a +new or replacement sandbox. A warm Runner stays `ready` in its own sandbox +after its turn, so that the next turn can adopt it live. When a restart or +deploy replaces its controller and the next run receives a different sandbox, +nothing can adopt that Runner any more. Live adoption then steps aside, and +admission applies the same checks as for a per-turn owner. The journal of a +warm owner has no `runner.suspend` command. A local warm Runner and a resumed +sandbox that still holds the Runner keep the live adoption path. + The provider session then follows the normal resume-or-replace policy. A -resumed sandbox, a warm session, a plugin task, a restart claim, or any +resumed sandbox, a local warm session, a plugin task, a restart claim, or any missing proof keeps the existing `prior_owner_terminal_state_indeterminate` rejection. When admission attempted the retirement, the error names the first proof that failed, for example @@ -3327,6 +3336,14 @@ is unknown. The completed copyback receipt then releases the hold for the stoppe run's own agent as well, so the user's redirect continues once. The hold row and its evidence are unchanged. +A run can restore a committed save and then fail before it saves anything else, +for example before provider work after a restart. Its restore hold then may +have no recovery receipt. When its saved reference names the exact snapshot +that its own succeeded `remote_workspace_restore` loaded, at the same epoch, it +left no unsaved files. A later continuation into another workspace skips that +source instead of failing with `remote_workspace_recovery_receipt_invalid`. A +source that saved any other snapshot still needs the canonical receipt. + A queued turn for the same task that never started is not a workspace writer for automatic recovery. The hold gates its dispatch until the recovery receipt is complete, so recovery and that turn cannot both wait for each other. Running @@ -4581,7 +4598,18 @@ backup verification and provider resume checks still run; this refresh does not start a new provider session or declare a continuity break. A marker that already names the exact selected identity only rehydrates its own controller state. The existing retained-sandbox binding and suspended Runner read still -govern that same-source resume. A current-run marker rehydrates only that run's +govern that same-source resume. + +The Runner saves `suspended` and a terminal fence for its `runner.suspend` +result before it sends that result. When a restart stops the controller first, +the fence stays in the Runner state although the suspension finished. The +suspended Runner read accepts this one fence: the fence names +`runner.suspend` with lifecycle `suspended`, it is the Runner's last controller +command, and the restored journal holds that exact command once and no later +command. A +later authorized Runner process reconciles the receipt. Any other fence, a +provider cleanup, unsent events or a warm transition still fail with +`native_controller_recovery_missing_evidence:runner_not_settled`. A current-run marker rehydrates only that run's own checkpoint before the existing bootstrap retry checks. A missing current checkpoint never selects an older turn instead. 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 56c2e491475..c40cb4dd51e 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 @@ -321,7 +321,14 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d await expect(readSandbox()).resolves.toEqual({ remoteCwd: "/workspace" }); await expect(inspect({ readControlState: () => ({ identity, commands: [{ status: "pending" }], committedEvents: [{ eventType: "run.terminal", envelope: identity }] }) })).rejects.toThrow(denied); - const recovered = await inspect(); + // A created or replacement sandbox holds no Runner to adopt. Live adoption + // steps aside so ordinary admission can retire the finished owner, instead + // of denying with prior_owner_terminal_state_indeterminate (Nov 10 staging). + for (const outcome of ["created", "replacement"] as const) { + await expect(inspect({ target: { ...target, sandboxLeaseAcquisition: { outcome, providerLeaseId: "new-sandbox" } } as never })) + .resolves.toBeNull(); + } + const recovered = (await inspect())!; expect(recovered.identity).toEqual(identity); expect(recovered.checkpoint).toBe(checkpoint); expect(recovered.reuseProcess).toBe(true); @@ -436,6 +443,85 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d } }); + // Nov 10 staging shape: a restart stopped the controller right after it sent + // runner.suspend to an idle warm Runner. The Runner saved `suspended` with a + // terminal fence for that result, which nobody acknowledged. The next warm + // run resumed the same sandbox and directory and failed with runner_not_settled. + it.each([ + { name: "undelivered suspend receipt", accepted: true }, + { name: "undelivered turn terminal", accepted: false, pending: { commandType: "turn.stop", lifecycle: "suspended" } }, + { name: "suspend receipt with provider cleanup", accepted: false, runner: { pendingProviderCleanup: { commandType: "runner.suspend" } } }, + { name: "suspend receipt with unsent events", accepted: false, runner: { outbox: [{ sourceSeq: 9 }] } }, + { name: "suspend receipt behind a later command", accepted: false, runner: { lastControllerCommandSeq: 8 } }, + { name: "journal names the command differently", accepted: false, journal: "turn.stop" }, + { name: "journal without the suspend command", accepted: false, journalCommands: [] }, + { name: "journal with a later command", accepted: false, laterCommand: true }, + ] as const)("reads a resumed sandbox's suspended Runner with an $name", async variant => { + const fixture = await seed(), environmentId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(); + const providerLeaseId = randomUUID(), base = join(root, `sandbox-${runId}`, "paperclip-workspace"); + const cwd = join(base, ".paperclip-workspaces", `${workspaceId}-1-${randomUUID()}`); + await mkdir(cwd, { recursive: true }); + await db.insert(environments).values({ id: environmentId, name: `Suspended ${environmentId}`, driver: "sandbox" }); + await db.insert(environmentLeases).values({ companyId, environmentId, issueId: fixture.scope.issueId, + heartbeatRunId: fixture.scope.runId, provider: "daytona", providerLeaseId, status: "expired", + leasePolicy: "reuse_by_environment", metadata: { remoteCwd: cwd } }); + 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 runnerDirectory = join(cwd, ".paperclip-runtime", "paperclip-runner", "sessions", + createHash("sha256").update(fixture.scope.normalizedSessionId).digest("hex"), "runner"); + const suspend = { commandId: randomUUID(), controllerSeq: 7 }; + await mkdir(runnerDirectory, { recursive: true }); + await writeFile(join(runnerDirectory, "runner-state.json"), JSON.stringify({ schema: "paperclip.runner.durable.state.v1", + ...identity, lifecycle: "suspended", outbox: [], lastControllerCommandSeq: 7, + pendingTerminalDelivery: { ...suspend, commandType: "runner.suspend", lifecycle: "suspended", ...("pending" in variant ? variant.pending : {}) }, + pendingProviderCleanup: null, warmTransition: null, ...("runner" in variant ? variant.runner : {}) })); + const execute = (request: { command: string; args?: string[] }) => { + const script = (request.args ?? []).join(" "); + const paths = script.match(/'(\/[^']*)'/g)?.map(path => path.slice(1, -1)) ?? []; + if (paths.some(path => !path.startsWith(`${base}/`))) { + return Promise.resolve({ stdout: "", stderr: "outside sandbox", exitCode: 126, timedOut: false }); + } + return new Promise<{ stdout: string; stderr: string; exitCode: number; timedOut: boolean }>(resolve => { + execFile(request.command, request.args ?? [], { cwd }, (error, stdout, stderr) => resolve({ stdout, stderr, + exitCode: error ? typeof error.code === "number" ? error.code : 1 : 0, timedOut: false })); + }); + }; + const execution = parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId, agentId, issueId: fixture.scope.issueId, runId, executionWorkspaceId: workspaceId }, + task: { identifier: "NCD-1", title: "Continue retained work", description: null, + prompt: "Continue the saved work.", workMode: "standard" }, + workspace: { cwd, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: fixture.scope.normalizedSessionId, driverKind: "codex_app_server", + protocolVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 900_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: [], + }); + // The restored journal still holds the controller's unanswered suspend. + const controller = join(root, `controller-${runId}`); + await mkdir(join(controller, "control-plane"), { recursive: true }); + await writeFile(join(controller, "control-plane", "control-plane-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.control-plane-state.v2", identity, + commands: "journalCommands" in variant ? variant.journalCommands : [ + { ...suspend, type: "journal" in variant ? variant.journal : "runner.suspend", status: "pending" }, + ...("laterCommand" in variant ? [{ commandId: randomUUID(), controllerSeq: 8, type: "turn.stop", status: "pending" }] : []), + ] })); + const target = { kind: "remote", transport: "sandbox", environmentId, leaseId: randomUUID(), providerKey: "daytona", + remoteCwd: cwd, sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId }, runner: { execute } } as const; + const restore = () => restoreRemoteSuspendedRunnerCheckpoint({ db, execution, runnerExecutionTarget: target as never }, controller); + + if (variant.accepted) { + await restore(); + expect(JSON.parse(await readFile(join(controller, "plugin-runner-state.json"), "utf8"))) + .toEqual({ schema: "paperclip.runner.durable.state.v1", lifecycle: "suspended", ...identity }); + } else { + await expect(restore()).rejects.toThrow("native_controller_recovery_missing_evidence:runner_not_settled"); + await expect(readFile(join(controller, "plugin-runner-state.json"))).rejects.toMatchObject({ code: "ENOENT" }); + } + }); + 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-harness-filesystem-independence-continuation.integration.test.ts b/server/src/services/native-runtime/native-harness-filesystem-independence-continuation.integration.test.ts index 3d3ae2327ef..4a9ad221791 100644 --- a/server/src/services/native-runtime/native-harness-filesystem-independence-continuation.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-filesystem-independence-continuation.integration.test.ts @@ -92,19 +92,19 @@ describeEmbeddedPostgres("legacy harness continuation without provider-home file const describe = vi.fn(async () => ({ kind: "runner", name: "controlled provider boundary", version: "fixture", capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } })); const backend: NativeSessionBackend = { descriptor: describe, recoverSession: recover, openSession: fresh, openReplacementSession: fresh }; const failure = await executePaperclipNativeSession({ db: f.db, execution: current, runnerInstanceId: native.runnerInstanceId, useRunnerd: true, backend, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: runtimeRoot, environmentId: randomUUID(), leaseId: randomUUID(), providerKey: "fixture", reusableLeaseConfigured: true, runner: { execute: async () => { throw new Error("unexpected remote process execution"); } }, sandboxLeaseAcquisition: { outcome: "created", providerLeaseId: "new-fixture-sandbox" } } as never }).catch(error => error); + // A warm owner whose controller a restart replaced, on a newly created + // sandbox, is retired like a per-turn owner and resumes the same way. + expect(recover, String(failure)).toHaveBeenCalledOnce(); + expect(recover.mock.calls[0]![0]).toMatchObject({ providerSessionId: snapshot.providerSessionId, identity: { runId } }); if (lifecycle === "warm") { - expect(String(failure)).toContain("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"); - expect(describe).not.toHaveBeenCalled(); expect(recover).not.toHaveBeenCalled(); expect(fresh).not.toHaveBeenCalled(); - const [retained] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); - const [unchanged] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); - expect(retained!.runnerProfileJson).toEqual(sourceProfile); - expect(unchanged!.runnerProfileJson?.sessionCheckpoint).toEqual(rebound); - observations.push({ refusal: String(failure), providerSessionId: rebound!.providerSessionId, - policy: rebound!.providerRecoveryPolicy, replacementAttempts: 0 }); + // Warm resumes from its own warm checkpoint, which carries no + // replacement policy: a failed resume is reported, never replaced. + expect(String(failure)).toContain("native_session_recovery_failed"); + expect(fresh).not.toHaveBeenCalled(); + observations.push({ providerSessionId: recover.mock.calls[0]![0].providerSessionId, replacementAttempts: 0 }); continue; } - expect(recover, String(failure)).toHaveBeenCalledOnce(); - expect(recover.mock.calls[0]![0]).toMatchObject({ providerSessionId: snapshot.providerSessionId, providerRecoveryPolicy: policy, identity: { runId } }); + expect(recover.mock.calls[0]![0]).toMatchObject({ providerRecoveryPolicy: policy }); expect(fresh).toHaveBeenCalledTimes(policy === "allow_replacement_after_resume_failure" ? 1 : 0); const digest = `sha256:${"a".repeat(64)}`; const artifacts = Object.fromEntries(Object.entries({ grokLauncher: "dist/providers/grok/launcher.cjs", nodeCommand: "node_modules/node/bin/node", diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index f27f169fdd5..271da7146e4 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -93,7 +93,8 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" | "failed-turn-start" | "rejected-turn-start" | "indeterminate-turn-start" | "unknown-outcome-turn-start" | "unsuperseded-unknown-outcome-turn-start" | "failed-suspend" = "settled", - options: { controller?: { bootId: string; pid: number; processStartedAt: Date }; provider?: "codex" | "acpx" | "opencode" } = {}) { + options: { controller?: { bootId: string; pid: number; processStartedAt: Date }; provider?: "codex" | "acpx" | "opencode"; + lifecycle?: "warm" } = {}) { const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(); await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" }); const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", @@ -108,7 +109,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered taskPrompt: "Continue the conversation.", workspace: { id: workspaceId, cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, normalizedSessionId: native.normalizedSessionId, - lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null }, + lifecyclePolicy: options.lifecycle === "warm" ? { mode: "warm", idleTimeoutMs: 900_000 } : { mode: "per_turn", idleTimeoutMs: null }, runtimeContext: nativeRuntimeContextFixture(), completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: contract.schemaVersion, contract: contract.contractJson as never }, @@ -138,8 +139,9 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered "unknown-outcome-turn-start": unknownOutcome, } as const; const answered = answers as Record; - const commands = ["run.prepare", "session.open", ...(control in answered ? ["turn.start"] : []), "turn.start", "runner.suspend"] - .map(type => controller.queueCommand(type)); + // A warm Runner stays ready after its turn: its controller sends no suspend. + const commands = ["run.prepare", "session.open", ...(control in answered ? ["turn.start"] : []), "turn.start", + ...(options.lifecycle === "warm" ? [] : ["runner.suspend"])].map(type => controller.queueCommand(type)); for (const command of control === "pending-command" ? commands.slice(0, -1) : commands) { Object.assign(command, { status: "completed", result: { status: "completed" } }); } @@ -620,6 +622,47 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(await archivedOwners(await sessionRoot())).toEqual([prior.runId]); }, 60_000); + // Staging shape: a warm Codex owner committed under a controller that a + // deploy then replaced. The next run got a newly created sandbox, so the + // prior Runner (left `ready` in its own retained sandbox) can never be + // adopted here. The follow-up failed with + // `prior_owner_terminal_state_indeterminate` and the task stayed blocked. + it.each(["created", "replacement"] as const)( + "retires a finished warm owner left behind by a restart when this run's sandbox is %s", async outcome => { + const prior = await seedRecoveredPriorRun("settled", { provider: "codex", lifecycle: "warm" }); + const followUp = await admitFollowUp(prior); + expect(await followUp.start(outcome)).toBe(ADMITTED); + expect(await archivedOwners(await sessionRoot())).toEqual([prior.runId]); + expect(await quarantined()).toEqual([]); + expect(followUp.logs.join("")).toContain("finished without a saved runner checkpoint; starting a new runner"); + expect(followUp.remoteExecute).not.toHaveBeenCalled(); + }, 60_000); + + it("keeps a finished warm owner for live adoption when this run resumed its sandbox", async () => { + const prior = await seedRecoveredPriorRun("settled", { provider: "codex", lifecycle: "warm" }); + const followUp = await admitFollowUp(prior); + // A resumed sandbox still holds that Runner; its own recovery path reads it. + expect(await followUp.start("resumed")).not.toBe(ADMITTED); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + }, 60_000); + + it.each(["pending-command", "unanswered-tool"] as const)( + "keeps a warm owner held when its journal is unsettled (%s), even on a new sandbox", async control => { + const prior = await seedRecoveredPriorRun(control, { provider: "codex", lifecycle: "warm" }); + const followUp = await admitFollowUp(prior); + expect(await followUp.start("created")).toMatch(/^runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; retirement_refused=journal_/); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + }, 60_000); + + it("keeps a warm owner held while its controller still holds the finalization lease", async () => { + const prior = await seedRecoveredPriorRun("settled", { provider: "codex", lifecycle: "warm" }); + await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID(), leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, prior.runId)); + const followUp = await admitFollowUp(prior); + expect(await followUp.start("created")).not.toBe(ADMITTED); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + }, 60_000); + it("skips every unstarted follow-up after repeated failed retries", async () => { const prior = await seedRecoveredPriorRun(); const failed = await failUnstartedFollowUp(prior); 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 dc06ac192ac..71e32b5a3cf 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 @@ -644,10 +644,9 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); (controller as ReturnType).kill("SIGKILL"); await exited; controller = undefined; - if (!resumedAfterSuspension) await expect(executePaperclipNativeSession({ ...nextOptions, - runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, - sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "unrelated-sandbox" } }, - })).rejects.toThrow("prior_owner_terminal_state_indeterminate"); + // A replacement sandbox would now retire this finished owner and start + // a new Runner (native-recovered-owner-followup.test.ts), which this + // continuity scenario must not do before resuming the same sandbox. } if (scenario === "legacy") { const checkpoints = resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner", "sessions"); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 21d39a8995f..dc548592fe3 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5603,6 +5603,15 @@ function retirementReasonName(kind: keyof typeof RETIREMENT_REASON_NAMES, value: * receipt is a claim of one recorded Runner process. Once that process and its * recorded provider processes are proven stopped by exact PID and start time, * the claim has no owner left and the same retirement applies. + * + * A warm sandbox owner keeps its Runner `ready` in its own sandbox after the + * turn, waiting for live adoption. A restart or deploy that replaces its + * controller leaves that Runner without anyone who could adopt it, and when + * this run received a different sandbox (created or replacement) it can never + * be adopted here. Its terminal state is then as knowable as a per-turn + * owner's: the same committed result, released lease, stopped controller and + * settled journal prove it finished, so the same retirement applies. A local + * warm Runner, or a resumed sandbox that still holds it, keeps the live path. */ async function inspectFinishedPerTurnRunnerdOwner(input: { db: Db; @@ -5618,8 +5627,9 @@ async function inspectFinishedPerTurnRunnerdOwner(input: { // A resumed sandbox still holds the prior Runner's own lifecycle, which its // existing recovery path reads. Plugin tasks own a separate saved checkpoint. const refuse = (refused: string) => ({ finished: false as const, refused }); - if (input.execution.session.lifecyclePolicy.mode !== "per_turn") return refuse("lifecycle_not_per_turn"); + const lifecycle = input.execution.session.lifecyclePolicy.mode; const localRunner = !target || target.kind === "local"; + if (lifecycle !== "per_turn" && !(lifecycle === "warm" && !localRunner)) return refuse("lifecycle_not_per_turn"); if (!localRunner) { if (target.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) { return refuse("target_not_sandbox_runner"); @@ -7028,12 +7038,40 @@ export async function restoreRemoteSuspendedRunnerCheckpoint(input: { // controller-fencing and physical-process checks. Only suspension is cached. if (!movedAllocation && input.execution.session.lifecyclePolicy.mode === "warm" && state.lifecycle !== "suspended") return; if (state.lifecycle !== "suspended" || !Array.isArray(state.outbox) || state.outbox.length !== 0 - || state.pendingTerminalDelivery != null || state.pendingProviderCleanup != null || state.warmTransition != null) { + || (state.pendingTerminalDelivery != null && !isUndeliveredSuspendReceipt(state, root)) + || state.pendingProviderCleanup != null || state.warmTransition != null) { throw new Error("native_controller_recovery_missing_evidence:runner_not_settled"); } savePluginRunnerCheckpoint(root, state, prior); } +/** The Runner saves `suspended` and a terminal fence for its suspend result + * before it sends that result. When a restart stops the controller first, the + * fence stays: the suspension finished, and only its acknowledgement is + * missing. A later authorized Runner process reconciles that receipt, so it + * does not hold the suspended state. Any other fence (a turn's terminal, a + * provider cleanup, unsent events), or a journal that lacks the command, + * names it differently or holds a later command still refuses. */ +function isUndeliveredSuspendReceipt(state: Record, root: string): boolean { + const pending = record(state.pendingTerminalDelivery); + if (pending.commandType !== "runner.suspend" || pending.lifecycle !== "suspended" + || typeof pending.commandId !== "string" || !pending.commandId + || !Number.isSafeInteger(pending.controllerSeq) || Number(pending.controllerSeq) <= 0 + || pending.controllerSeq !== state.lastControllerCommandSeq) return false; + try { + const control = record(JSON.parse(readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8"))); + const commands = Array.isArray(control.commands) ? control.commands.map(record) : []; + const matching = commands.filter(command => command.commandId === pending.commandId); + // The suspend must be journaled exactly once and be the controller's last command. + return matching.length === 1 && matching[0]!.type === "runner.suspend" + && matching[0]!.controllerSeq === pending.controllerSeq + && commands.every(command => !(Number(command.controllerSeq) > Number(pending.controllerSeq))); + } catch { + return false; + } +} + /** Materialize opaque controller state before ordinary warm admission reads it. * A retained owner or server checkpoint identifies the last completed run; * the original external marker can precede several in-process warm turns. */ @@ -9969,8 +10007,8 @@ async function executePaperclipNativeSessionAttempt( canReuseProcess: previous => !needsCoreGitHubAccess(input) && canReuseNativeProcess(input, record(JSON.parse(readBoundedNativeFile(nativeSessionCheckpointPath(previous), NATIVE_WARM_CHECKPOINT_MAX_BYTES, "native_session_supervisor_checkpoint_too_large").toString("utf8"))).processConfigDigest), - }); - remoteWarmRecovery.releasedSource = releasedSource; + }) ?? undefined; + if (remoteWarmRecovery) remoteWarmRecovery.releasedSource = releasedSource; } } if (input.useRunnerd) { 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 d1759466ed7..e2ce1bcbc9c 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -123,7 +123,7 @@ export async function inspectRemoteWarmSessionRecovery(input: { matchesExecution: (previous: NativeExecutionInput) => boolean; loadCheckpoint: (previous: NativeExecutionInput) => PersistedNativeSession | null; canReuseProcess: (previous: NativeExecutionInput) => boolean; -}): Promise { +}): Promise { const deny = (): never => { throw new Error("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"); }; const { binding } = input.execution; const { target } = input; @@ -133,6 +133,13 @@ export async function inspectRemoteWarmSessionRecovery(input: { if (typeof value !== "string" || !value) return deny(); identity[key] = value; } + // A Runner lives in the sandbox it started in. A created or replacement + // sandbox (for example after a restart or deploy replaced the controller) + // holds no Runner to adopt, so live adoption steps aside. Ordinary admission + // then retires the finished owner or keeps it held with a named reason. + if (target.transport === "sandbox" && ["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) { + return null; + } if (target.transport !== "sandbox" || !target.leaseId || !target.environmentId || !target.providerKey || !target.sandboxLeaseAcquisition || target.sandboxLeaseAcquisition.outcome !== "resumed" || identity.runId === binding.runId) return deny(); diff --git a/server/src/services/remote-workspace-recovery-attachments.ts b/server/src/services/remote-workspace-recovery-attachments.ts index 9ee0360a040..046f11dea3b 100644 --- a/server/src/services/remote-workspace-recovery-attachments.ts +++ b/server/src/services/remote-workspace-recovery-attachments.ts @@ -57,7 +57,9 @@ export async function attachPreservedRemoteWorkspaceSnapshots(input: { if (remote.workspaceId === input.workspaceId || remote.state !== "saved" || !remote.snapshotId) continue; const proof = await immutableRemoteWorkspaceRecoveryAttachment(input.db, source); if (!proof) { - if (hasRequiredWorkspaceRecovery(source.resultJson)) throw new Error("remote_workspace_recovery_receipt_invalid"); + if (hasRequiredWorkspaceRecovery(source.resultJson) && !await savedOnlyItsRestoredSnapshot(input.db, source, remote)) { + throw new Error("remote_workspace_recovery_receipt_invalid"); + } continue; } const operationId = proof.operationId; @@ -101,6 +103,29 @@ export async function attachPreservedRemoteWorkspaceSnapshots(input: { return result; } +/** A run that restored a committed save and then failed (for example before + * provider work, after a restart) can keep a restore hold that never received a + * recovery receipt. When its saved reference names the very snapshot that its + * own succeeded restore loaded, at the same epoch, it saved nothing else: no + * unsaved file exists to attach, and the last save already holds its files. + * Any other saved snapshot still needs the canonical receipt. */ +async function savedOnlyItsRestoredSnapshot(db: Db, source: typeof heartbeatRuns.$inferSelect, + remote: { workspaceId: string; snapshotId?: string | null; epoch: number }): Promise { + if (!remote.snapshotId) return false; + const [restore] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, source.companyId), eq(workspaceOperations.heartbeatRunId, source.id), + eq(workspaceOperations.phase, "remote_workspace_restore"), eq(workspaceOperations.status, "succeeded"), + sql`${workspaceOperations.metadata}->>'snapshotId' = ${remote.snapshotId}`, + sql`${workspaceOperations.metadata}->>'epoch' = ${String(remote.epoch)}`, + )).limit(1); + if (!restore) return false; + const [snapshot] = await db.select({ id: executionWorkspaceSnapshots.id }).from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, source.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, remote.workspaceId), + eq(executionWorkspaceSnapshots.id, remote.snapshotId), inArray(executionWorkspaceSnapshots.status, ["committed", "superseded"]), + )).limit(1); + return Boolean(snapshot); +} + /** Verify a sidecar already inside an authorized shared workspace. This only * reconstructs its immutable identity; it cannot import files or admit work. */ async function retainedRecoveryAttachmentProvenance(input: { diff --git a/server/src/services/workspace-recovery-preparation.test.ts b/server/src/services/workspace-recovery-preparation.test.ts index 1327cf5194b..44924d4ba27 100644 --- a/server/src/services/workspace-recovery-preparation.test.ts +++ b/server/src/services/workspace-recovery-preparation.test.ts @@ -541,6 +541,44 @@ describe("automatic retained workspace recovery", () => { } finally { failingRead?.mockRestore(); } }, 30_000); + // Nov 10 staging (PAP-3881): a continuation into a fresh workspace failed + // with remote_workspace_recovery_receipt_invalid. Its predecessor restored the + // committed save, failed before provider work, and kept a restore hold with + // no recovery receipt. It saved the very snapshot it restored, so it left no + // unsaved files to attach. A source that saved anything else stays strict. + it.each(["restored_unchanged", "saved_other_snapshot", "restore_not_succeeded", "restored_other_epoch"] as const)( + "decides a receipt-less restore hold of a source that %s", async variant => { + const f = await failedRemotePreparation(); + const template = await run(f.runId); + const restoredSnapshotId = randomUUID(), sourceId = randomUUID(), leaseId = randomUUID(); + await db.insert(executionWorkspaceSnapshots).values({ id: restoredSnapshotId, companyId: f.companyId, + executionWorkspaceId: f.workspaceId, sourceOperationId: randomUUID(), sourceRunId: randomUUID(), + generation: 4, epoch: 3, status: "committed", sha256: "a".repeat(64) }); + await db.insert(heartbeatRuns).values({ id: sourceId, companyId: f.companyId, agentId: template.agentId, + nativeIssueId: f.issueId, issueId: f.issueId, status: "failed", runtimeMode: "native", finishedAt: new Date(), + contextSnapshot: { issueId: f.issueId }, + errorCode: "native_continuation_requires_reconciliation", + error: "native_controller_recovery_missing_evidence:retained_sandbox_binding", + runnerProfileJson: { remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", epoch: 6, state: "saved", + leaseId, providerLeaseId: randomUUID(), remoteCwd: `/home/daytona/paperclip-workspace/.paperclip-workspaces/${f.workspaceId}-6-${leaseId}`, + workspaceId: f.workspaceId, snapshotId: variant === "saved_other_snapshot" ? randomUUID() : restoredSnapshotId } }, + resultJson: { stopReason: "adapter_failed", executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] } } }); + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: sourceId, + executionWorkspaceId: f.workspaceId, phase: "remote_workspace_restore", + status: variant === "restore_not_succeeded" ? "failed" : "succeeded", + metadata: { epoch: variant === "restored_other_epoch" ? 5 : 6, snapshotId: restoredSnapshotId } }); + const [writer] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: template.agentId, + issueId: f.issueId, status: "running", responsibleUserId: "local-board", contextSnapshot: { issueId: f.issueId } }).returning(); + const destination = path.join(root, writer.id); await mkdir(destination); + const attach = attachPreservedRemoteWorkspaceSnapshots({ db, companyId: f.companyId, issueId: f.issueId, + runId: writer.id, workspaceId: randomUUID(), sourceRunIds: [sourceId], cwd: destination }); + if (variant === "restored_unchanged") await expect(attach).resolves.toEqual([]); + else await expect(attach).rejects.toThrow("remote_workspace_recovery_receipt_invalid"); + expect(await readdir(destination)).toEqual([]); + }, 30_000); + it.each(["unchanged", "advanced"])("deduplicates immutable canonical sidecars when the source workspace is %s", async state => { const f = await failedRemotePreparation(); await recoverRetainedWorkspace(f.input); const source = await run(f.runId), destination = path.join(root, randomUUID()); await mkdir(destination);