diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index fcf339efa67..f77b10f3b04 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1478,6 +1478,30 @@ differed, for example `lease_status`, `adapter_type`, `provider_lease_id`, `remo `identity_run_id`. The detail contains no identifiers. Consumers match the code exactly or as the `code: ` prefix. +A continuation run starts with its predecessor's saved provider session. For a +warm session it also carries the predecessor's runner process identity. When the +run fails before it asks for a runner of its own (for example while it stages +its provider pack), a due same-run retry does not reattach: there is no runner +of this run to reattach to. The server retires the inherited process identity +and uses the ordinary same-run resume, which is the path the failed attempt +took. The resume prepares the run again in its own retained sandbox when that +sandbox lease is still active, and acquires one as an ordinary resume does when +it is gone. The server does this only when all of these are true: + +- the failed attempt is a result-less exact-checkpoint retry that recorded + `providerEventsExist: false`, and no controller owns it; +- the run has no runner start, process identity, adapter invocation, tool start, + runner event, controller checkpoint, provider event or provider session of + its own; +- the saved session has no active turn and no pending runtime request; and +- a settled predecessor of the same native session recorded the exact same + process identity. + +It records an `inherited_runner_never_launched` lifecycle event. Any other run, +and any run saved by an older release without these facts, keeps verified +reattachment. A reattachment that loses its runner still requests the user's +Retry as described above. + When a native attempt fails before launch because its predecessor's session is quarantined, explicit task Retry verifies the retained journal's actual owner. That owner must belong to the same company, task, agent, workspace, native session, diff --git a/server/src/__tests__/environment-run-orchestrator.test.ts b/server/src/__tests__/environment-run-orchestrator.test.ts index a5c368142e4..c922db7ef04 100644 --- a/server/src/__tests__/environment-run-orchestrator.test.ts +++ b/server/src/__tests__/environment-run-orchestrator.test.ts @@ -1,3 +1,4 @@ +import { RetainedSandboxAbsentError } from "../services/retained-sandbox-absent-error.js"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { randomUUID } from "node:crypto"; import * as remoteWorkspaces from "../services/remote-workspaces.js"; @@ -1015,6 +1016,13 @@ describe("live remote runner lease reattachment", () => { expect(mockUpdateLeaseMetadata).not.toHaveBeenCalled(); }, ); + it("names a provider-confirmed missing sandbox as a lease mismatch without acquiring a replacement", async () => { + const runtime = makeMockRuntime({ resumeRunLease: vi.fn(async () => { throw new RetainedSandboxAbsentError(); }) }); + await expect(environmentRunOrchestrator({} as never, { environmentRuntime: runtime }).acquireForRun(input)) + .rejects.toThrow(new Error("native_remote_recovery_lease_mismatch: stage=environment_resume field=lease_expired")); + expect(runtime.acquireRunLease).not.toHaveBeenCalled(); + expect(runtime.execute).not.toHaveBeenCalled(); + }); it("preserves a retained sandbox startup failure without acquiring a replacement", async () => { const error = new Error("retained sandbox startup timed out"); const runtime = makeMockRuntime({ resumeRunLease: vi.fn(async () => { throw error; }) }); diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 08441642ecc..79cc2b987b1 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -3956,6 +3956,28 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }); }); + it.each([ + [{ providerLeaseId: null, metadata: { expired: true } }, "RetainedSandboxAbsentError"], + [{ providerLeaseId: null, metadata: { expired: true, workspaceSentinel: { result: "mismatch" } } }, "RetainedSandboxAbsentError"], + [{ providerLeaseId: "another-sandbox", metadata: {} }, "Error"], + [{ providerLeaseId: null, metadata: {} }, "Error"], + ] as const)("names only a provider-confirmed missing sandbox as absent on exact resume (%j)", async (reply, name) => { + const seeded = await seedReusablePluginSandboxLease("paperclip_runner", { timeoutMs: 1234 }); + const lease = (await environmentService(db).getLeaseById(seeded.reusableLease.id))!; + const call = vi.fn(async (_id: string, method: string) => { + if (method !== "environmentResumeLease") throw new Error("Only exact resume is permitted"); + return reply; + }); + const workerManager = { isRunning: () => true, call, + getWorker: () => ({ supportedMethods: ["environmentResumeLease", "environmentReleaseLease", "environmentDestroyLease"] }), + } as unknown as PluginWorkerManager; + const runtime = environmentRuntimeService(db, { pluginWorkerManager: workerManager }); + const error = await runtime.resumeRunLease({ environment: seeded.environment, lease }).catch(caught => caught); + expect(error).toBeInstanceOf(Error); + expect(error.name).toBe(name); + expect(call).toHaveBeenCalledOnce(); + }); + it.each([ { timeoutMs: 300_000, rpcTimeoutMs: 330_000 }, { timeoutMs: 1234, rpcTimeoutMs: 31_234 }, diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index de083eed785..2cc5c05d702 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -1,3 +1,4 @@ +import { RetainedSandboxAbsentError } from "./retained-sandbox-absent-error.js"; import { freshWorkspaceSchema } from "@paperclipai/plugin-sdk"; import { prepareRemoteWorkspace, readRemoteWorkspaceAuthority, remoteWorkspaceRepositoryPath } from "./remote-workspaces.js"; /** @@ -385,6 +386,11 @@ export function environmentRunOrchestrator( environment, lease, requestedCwd: expected.remoteCwd, ...(warmRunnerIdleTimeoutMs !== undefined ? { warmRunnerIdleTimeoutMs } : {}), ...(runnerReconnectGraceMs !== undefined ? { runnerReconnectGraceMs } : {}), + }).catch((error: unknown) => { + // Confirmed provider absence is a named lease mismatch; other + // failures (for example a transient refusal) keep their own error. + if (error instanceof RetainedSandboxAbsentError) throw remoteRecoveryLeaseMismatch("environment_resume", "lease_expired"); + throw error; }); if (!resumed) throw remoteRecoveryLeaseMismatch("environment_resume", "lease"); if (resumed.providerLeaseId !== expected.providerLeaseId) { diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 4fc672b52dc..bc9f3b4df0f 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,3 +1,4 @@ +import { RetainedSandboxAbsentError, resumeReplyConfirmsAbsence } from "./retained-sandbox-absent-error.js"; import { readNativeWorkspaceMigration } from "./native-runtime/native-workspace-migration.js"; import { ReusableSandboxResumeError } from "./reusable-sandbox-resume-error.js"; import { PluginWorkerUnavailableError } from "./plugin-worker-unavailable-error.js"; @@ -2617,6 +2618,7 @@ function createSandboxEnvironmentDriver( ...(input.requestedCwd !== undefined ? { requestedCwd: input.requestedCwd } : {}), }, resolvePluginSandboxRpcTimeoutMs(config) ?? 60_000); if (resumed?.providerLeaseId !== input.lease.providerLeaseId) { + if (resumeReplyConfirmsAbsence(resumed)) throw new RetainedSandboxAbsentError(); throw new Error("The provider did not confirm the exact retained sandbox. No replacement was acquired."); } return resumed; diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index ddc034f7259..d8586a3b6b1 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -1,3 +1,4 @@ +import { retireInheritedRemoteRunnerIdentity } from "../native-runtime/native-inherited-runner-retry.js"; import { randomUUID } from "node:crypto"; import { spawn } from "node:child_process"; import { once } from "node:events"; @@ -369,6 +370,137 @@ describePostgres("heartbeat recovery module ownership", () => { } finally { kill.mockRestore(); try { await f.cleanup(); } finally { await child.stop(); } } }); + // Staging shape: a chat continuation inherited its predecessor's saved + // provider session and warm runner identity, then failed while staging its + // provider pack, before it ever asked for a runner of its own. Its workspace + // reference is saved and its own sandbox lease is still active. + type InheritedCase = "staging" | "own_turn" | "own_launch" | "own_checkpoint" | "no_predecessor" + | "pending_side_effect" | "legacy_detail"; + async function inheritedRunnerRetry(pid: number, condition: InheritedCase) { + const f = await remoteRetryFixture(pid, "remote_workspace"); + const run = await f.current(); + // The PID came from the predecessor, so this run recorded no process identity. + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); + if (condition !== "own_checkpoint") await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.run.id)); + const profile = run.runnerProfileJson ?? {}; + const checkpoint = profile.sessionCheckpoint as Record; + await db.update(agents).set({ status: "active" }).where(eq(agents.id, f.run.agentId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeExecutionInput: { fixture: true }, + remoteWorkspace: { ...(profile.remoteWorkspace as Record), state: "saved" }, + sessionCheckpoint: { ...checkpoint, activeTurnId: null, + pendingRuntimeRequests: condition === "pending_side_effect" ? [{ requestId: "tool-1", kind: "tool_call" }] : [] }, + } }).where(eq(heartbeatRuns.id, f.run.id)); + await db.update(environmentLeases).set({ executionWorkspaceId: null }).where(eq(environmentLeases.id, f.lease.id)); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, failureDetail: { + message: "runner_remote_provider_archive_invalid", recoveryMode: "exact_checkpoint_resume", checkpointExists: true, + providerSessionEstablished: true, ...(condition === "legacy_detail" ? {} : { providerEventsExist: false }), + } }).where(eq(nativeRunFinalizations.runId, f.run.id)); + const predecessorId = randomUUID(); + if (condition !== "no_predecessor") await db.insert(heartbeatRuns).values({ id: predecessorId, + companyId: f.run.companyId, agentId: f.run.agentId, issueId: f.run.issueId, invocationSource: "automation", + status: "succeeded", runtimeMode: "native", nativeSessionId: run.nativeSessionId, createdAt: new Date(Date.now() - 600_000), + finishedAt: new Date(Date.now() - 300_000), processPid: run.processPid, processGroupId: run.processGroupId, + processStartedAt: run.processStartedAt }); + if (condition === "own_turn") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id, + agentId: f.run.agentId, eventType: "turn.started", stream: "system", level: "info", message: "turn.started" }); + if (condition === "own_launch") await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id, + agentId: f.run.agentId, eventType: PROCESS_START_REQUESTED, stream: "system", level: "info" }); + return { ...f, predecessorId, async cleanupAll() { + try { await f.cleanup(); } finally { await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, predecessorId)); } + } }; + } + + it("resumes a continuation that never launched its own runner in its retained sandbox, without reattaching", async () => { + const child = await ownedProcess(); + const f = await inheritedRunnerRetry(child.pid, "staging"); + try { + native.claims.mockClear(); + await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id }); + await Promise.all([...f.deps.activeRunExecutionPromises]); + expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(false); + expect(f.deps.executeRun).toHaveBeenCalledExactlyOnceWith(f.run.id, { nativeLeaseOwner: expect.any(String) }); + const current = await f.current(); + expect(current).toMatchObject({ processPid: null, processGroupId: null, processStartedAt: null }); + expect(current.runnerProfileJson?.nativeInheritedRunnerRetirement).toEqual({ + schema: "paperclip.native-inherited-runner-retirement.v1", predecessorRunId: f.predecessorId, + retainedSandbox: { leaseId: f.lease.id, providerLeaseId: f.lease.providerLeaseId, remoteCwd: "/work" } }); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); + expect(events.find(event => event.payload?.reason === "inherited_runner_never_launched")?.payload) + .toMatchObject({ predecessorRunId: f.predecessorId, retainedSandbox: true }); + // The retained sandbox lease stays the run's only lease; nothing new was acquired. + expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, f.run.id))) + .toMatchObject([{ id: f.lease.id, status: "active" }]); + expect(f.deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled(); + expect(child.alive()).toBe(true); + } finally { try { await f.cleanupAll(); } finally { await child.stop(); } } + }); + + it("routes a run another sweep already retired to the same resume", async () => { + const child = await ownedProcess(); + const f = await inheritedRunnerRetry(child.pid, "staging"); + try { + expect(await retireInheritedRemoteRunnerIdentity(db, f.run.id)).toBe("retired"); + expect(await retireInheritedRemoteRunnerIdentity(db, f.run.id)).toBe("retired"); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); + expect(events.filter(event => event.payload?.reason === "inherited_runner_never_launched")).toHaveLength(1); + } finally { try { await f.cleanupAll(); } finally { await child.stop(); } } + }); + + it("defers instead of reattaching when the retirement cannot take its locks", async () => { + const child = await ownedProcess(); + const f = await inheritedRunnerRetry(child.pid, "staging"); + let release!: () => void; + const released = new Promise(resolve => { release = resolve; }); + let locked!: () => void; + const holding = new Promise(resolve => { locked = resolve; }); + const holder = db.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, f.run.issueId!)).for("update"); + locked(); + await released; + }); + try { + await holding; + native.claims.mockClear(); + const deferred = vi.fn(); + await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id, onDeferredDeadline: deferred }); + expect(deferred).toHaveBeenCalled(); + expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(false); + expect(f.deps.executeRun).not.toHaveBeenCalled(); + expect((await f.current()).processPid).toBe(child.pid); + } finally { + release(); await holder; + try { await f.cleanupAll(); } finally { await child.stop(); } + } + }, 30_000); + + it("keeps the existing recovery path for a non-transient retirement error", async () => { + const failing = { transaction: async () => { throw new Error("unexpected retirement fault"); } } as unknown as Db; + expect(await retireInheritedRemoteRunnerIdentity(failing, randomUUID())).toBe("not_applicable"); + const lockTimeout = Object.assign(new Error("Failed query"), { cause: Object.assign(new Error("lock timeout"), { code: "55P03" }) }); + const contended = { transaction: async () => { throw lockTimeout; } } as unknown as Db; + expect(await retireInheritedRemoteRunnerIdentity(contended, randomUUID())).toBe("transient"); + const statementTimeout = Object.assign(new Error("Failed query"), { cause: Object.assign(new Error("statement timeout"), { code: "57014" }) }); + const slow = { transaction: async () => { throw statementTimeout; } } as unknown as Db; + expect(await retireInheritedRemoteRunnerIdentity(slow, randomUUID())).toBe("transient"); + }); + + it.each(["own_turn", "own_launch", "own_checkpoint", "no_predecessor", "pending_side_effect", "legacy_detail"] as const)( + "keeps verified reattachment when the run's own runner may hold work or the proof is incomplete: %s", async condition => { + const child = await ownedProcess(); + const f = await inheritedRunnerRetry(child.pid, condition); + try { + native.claims.mockClear(); + await createHeartbeatRecovery(db, f.deps).resumeRetryableNativeProcessSessions({ runId: f.run.id }); + await Promise.all([...f.deps.activeRunExecutionPromises]); + expect(native.claims.mock.calls.some(([input]) => input?.runIds?.includes(f.run.id))).toBe(true); + expect(f.deps.executeRun).not.toHaveBeenCalledWith(f.run.id, { nativeLeaseOwner: expect.any(String) }); + const current = await f.current(); + expect(current.processPid).toBe(child.pid); + expect(current.runnerProfileJson).not.toHaveProperty("nativeInheritedRunnerRetirement"); + } finally { try { await f.cleanupAll(); } finally { await child.stop(); } } + }, + ); + it("routes a due plugin-task retry through its existing remote lease", async () => { const child = await ownedProcess(); const f = await remoteRetryFixture(child.pid, "plugin_task"); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index a12a74bfbfb..06c35f3d206 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -28,6 +28,7 @@ import { } from "../sandbox-stop-and-retain.js"; import { discoverSettledNativeRunProcesses, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, recordNativeLocalProcessStopInTransaction, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; +import { retireInheritedRemoteRunnerIdentity } from "../native-runtime/native-inherited-runner-retry.js"; import { legacyControllerBootId, liveLegacyControllerExpiresAt, @@ -2202,7 +2203,13 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD if (activeRunExecutions.has(run.id) || runningProcesses.has(run.id)) continue; const profile = parseObject(run.runnerProfileJson); if (profile.nativeWorkspaceSync != null || profile.remoteWorkspace != null) { - remoteRetryRunIds.push(run.id); + // A continuation that never started its own runner holds only its + // predecessor's identity: resume it the way its failed attempt ran. + const retirement = await retireInheritedRemoteRunnerIdentity(db, run.id); + if (retirement === "retired") claimableNativeRunIds.add(run.id); + // A lock or database error is not evidence: retry on the next sweep. + else if (retirement === "transient") scope.onDeferredDeadline?.(Date.now() + 30_000); + else remoteRetryRunIds.push(run.id); } else { const [taskLease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index 1edc7bfe73d..802934c00df 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -2065,6 +2065,76 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace expect(input.nativeRunnerPreparationSpans).toMatchObject([{ name: "environment.acquire", outcome: "failed" }]); }); + // A same-run resume whose inherited runner identity was retired keeps the + // run's own retained sandbox. When that sandbox is gone, it acquires as an + // ordinary resume does. + function retiredInheritedRunner(input: Awaited>["input"]) { + const location = { leaseId: randomUUID(), providerLeaseId: "retained-sandbox", remoteCwd: "/retained" }; + input.run.runnerProfileJson = { ...(input.run.runnerProfileJson ?? {}), + remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", workspaceId: randomUUID(), epoch: 2, + state: "saved", snapshotId: randomUUID(), ...location }, + nativeInheritedRunnerRetirement: { schema: "paperclip.native-inherited-runner-retirement.v1", + predecessorRunId: randomUUID(), retainedSandbox: location } }; + input.reuse.runOptions = { nativeLeaseOwner: "resume-owner" }; + return location; + } + + it("resumes a retired inherited runner in its own retained sandbox", async () => { + const { input, envOrchestrator } = await fixture(); + const location = retiredInheritedRunner(input); + const reached = new Error("Reached the retained sandbox"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ reattachRemoteLease: location })); + }); + + it("acquires as an ordinary resume when the retained sandbox is gone", async () => { + const { input, envOrchestrator } = await fixture(); + retiredInheritedRunner(input); + const reached = new Error("Reached ordinary acquisition"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun") + .mockRejectedValueOnce(new Error("native_remote_recovery_lease_mismatch: stage=environment_acquire field=lease_status")) + .mockRejectedValueOnce(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire).toHaveBeenCalledTimes(2); + expect(acquire.mock.calls[1]![0]).not.toHaveProperty("reattachRemoteLease"); + }); + + it("keeps a live retained lease's scope failure instead of acquiring another sandbox", async () => { + const { input, envOrchestrator } = await fixture(); + retiredInheritedRunner(input); + const scope = new Error("native_remote_recovery_lease_mismatch: stage=environment_acquire field=lease_scope"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(scope); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(scope); + expect(acquire).toHaveBeenCalledOnce(); + }); + + it("resumes a retired inherited runner's native workspace sync sandbox", async () => { + const { input, envOrchestrator } = await fixture(); + const location = { leaseId: randomUUID(), providerLeaseId: "retained-sync-sandbox", remoteCwd: "/retained-sync" }; + input.run.runnerProfileJson = { ...(input.run.runnerProfileJson ?? {}), + nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", workspaceId: randomUUID(), + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + resourceDisposition: "destroy", ...location }, + nativeInheritedRunnerRetirement: { schema: "paperclip.native-inherited-runner-retirement.v1", + predecessorRunId: randomUUID(), retainedSandbox: location } }; + input.reuse.runOptions = { nativeLeaseOwner: "resume-owner" }; + const reached = new Error("Reached the retained sync sandbox"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ reattachRemoteLease: location })); + }); + + it("does not reuse a retained sandbox without a retired inherited runner", async () => { + const { input, envOrchestrator } = await fixture(); + retiredInheritedRunner(input); + delete (input.run.runnerProfileJson as Record).nativeInheritedRunnerRetirement; + const reached = new Error("Reached ordinary acquisition"); + const acquire = vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(reached); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(reached); + expect(acquire.mock.calls[0]![0]).not.toHaveProperty("reattachRemoteLease"); + }); + it("binds a replacement workspace from realization and snapshots the restore lease", async () => { const { input, envOrchestrator, issue } = await fixture(); const realize = envOrchestrator.realizeForRun; diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 7ff85efacfd..86acdbd3d45 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -1,4 +1,5 @@ -import { remoteRecoveryLeaseMismatch } from "../native-runtime/remote-recovery-lease-mismatch.js"; +import { NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH, remoteRecoveryLeaseMismatch } from "../native-runtime/remote-recovery-lease-mismatch.js"; +import { readInheritedRunnerRetainedSandbox } from "../native-runtime/native-inherited-runner-retry.js"; import { resolveProviderApiPlacement } from "../native-runtime/provider-api-placement.js"; import { assertEphemeralWorkspaceExecution, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, shouldUseRemoteWorkspaces } from "../remote-workspaces.js"; import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WorkspaceBusyDeferral } from "./retries.js"; @@ -97,6 +98,9 @@ import type { type WorkspaceResolver = ReturnType; type WorkspaceMetadataInput = Parameters[0]; +/** Exact-lease failures that prove the retained sandbox can no longer be used. */ +const RETAINED_SANDBOX_GONE = new RegExp(`^${NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH}: stage=environment_(?:acquire|resume) field=(?:lease|lease_status|lease_expired|provider_lease_id)$`); + /** Prepares the workspace and environment; dispatch and terminal lease release stay with the caller. */ export interface HeartbeatWorkspacePreparationInput { run: typeof heartbeatRuns.$inferSelect; @@ -1306,7 +1310,23 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace : saved.remoteCwd !== sandboxRecovery.remoteCwd ? "remote_cwd" as const : null; if (mismatch) throw remoteRecoveryLeaseMismatch("workspace_preparation", mismatch); } - acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => envOrchestrator.acquireForRun({ + // A same-run resume whose inherited runner identity was retired keeps the + // run's own retained sandbox, so its workspace and allocation carry over. + const retainedSandbox = runOptions.nativeLeaseOwner && !runOptions.nativeRestartRecovery + ? readInheritedRunnerRetainedSandbox(parseObject(run.runnerProfileJson)) : null; + const acquireRetainedOrNew = async (input: Parameters[0]) => { + if (!retainedSandbox) return envOrchestrator.acquireForRun(input); + try { + return await envOrchestrator.acquireForRun({ ...input, reattachRemoteLease: retainedSandbox }); + } catch (error) { + // Only proof that the retained sandbox is gone permits ordinary + // acquisition. A live lease with another scope keeps the failure. + if (!(error instanceof Error) || !RETAINED_SANDBOX_GONE.test(error.message)) throw error; + logger.warn({ runId: run.id, detail: error.message }, "retained sandbox unavailable for same-run resume; acquiring"); + return envOrchestrator.acquireForRun(input); + } + }; + acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => acquireRetainedOrNew({ companyId: agent.companyId, selectedEnvironmentId: selectedEnvironmentId ?? localEnvironment.id, localEnvironmentId: localEnvironment.id, diff --git a/server/src/services/native-runtime/native-inherited-runner-retry.ts b/server/src/services/native-runtime/native-inherited-runner-retry.ts new file mode 100644 index 00000000000..63c686304b8 --- /dev/null +++ b/server/src/services/native-runtime/native-inherited-runner-retry.ts @@ -0,0 +1,172 @@ +import { and, desc, eq, inArray, isNull, lt, ne, or, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; +import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; +import { hasNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { nativeSessionProviderEvidence } from "./native-session-resume.js"; +import { hasNativeExecutionStartEvidence } from "./native-restart-recovery.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; +import { logger } from "../../middleware/logger.js"; + +/** + * A continuation run starts with its predecessor's saved provider session and, + * for a warm session, the predecessor's runner process identity. When such a + * run fails before it ever asks for its own runner (for example while staging + * its provider pack), the recorded process identity is the predecessor's, not + * its own. A due same-run retry used to reattach to "this run's" runner, which + * never existed, and always failed with `native_remote_recovery_lease_mismatch`. + * + * This module proves that shape from durable state and retires the inherited + * identity, so the retry takes the ordinary same-run resume: the same path the + * failed attempt took. It also records the run's own retained sandbox lease, + * so preparation resumes that exact sandbox instead of acquiring another one. + */ +export const NATIVE_INHERITED_RUNNER_RETIREMENT_KEY = "nativeInheritedRunnerRetirement"; +const RETIREMENT_SCHEMA = "paperclip.native-inherited-runner-retirement.v1"; + +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; + +export type InheritedRunnerRetainedSandbox = { leaseId: string; providerLeaseId: string; remoteCwd: string }; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +/** Lock unavailable, statement timeout, serialization failure, deadlock, shutdown, or a connection class error. */ +const TRANSIENT_SQLSTATES = new Set(["55P03", "57014", "40001", "40P01", "57P01", "57P03"]); + +function isTransientDatabaseError(error: unknown): boolean { + for (let current = error, depth = 0; current && typeof current === "object" && depth < 5; + current = (current as { cause?: unknown }).cause, depth++) { + const code = (current as { code?: unknown }).code; + if (typeof code === "string" && (TRANSIENT_SQLSTATES.has(code) || code.startsWith("08"))) return true; + if (typeof code === "string" && ["ECONNRESET", "ECONNREFUSED", "ETIMEDOUT", "CONNECTION_CLOSED", "CONNECTION_ENDED"].includes(code)) return true; + } + return false; +} + +function sameTime(left: Date | null, right: Date | null) { + return (left?.getTime() ?? null) === (right?.getTime() ?? null); +} + +/** The run's saved remote location: a native workspace sync or a remote workspace reference. */ +function savedRemoteLocation(profile: Record): InheritedRunnerRetainedSandbox | null { + const sync = readNativeWorkspaceSyncReference(profile.nativeWorkspaceSync); + if (sync) return { leaseId: sync.leaseId, providerLeaseId: sync.providerLeaseId, remoteCwd: sync.remoteCwd }; + const remote = remoteWorkspaceRunReferenceSchema.safeParse(profile.remoteWorkspace); + return remote.success + ? { leaseId: remote.data.leaseId, providerLeaseId: remote.data.providerLeaseId, remoteCwd: remote.data.remoteCwd } + : null; +} + +/** The retained sandbox recorded when the inherited identity was retired, if it still matches the saved location. */ +export function readInheritedRunnerRetainedSandbox(profile: Record | null | undefined): InheritedRunnerRetainedSandbox | null { + const marker = record(profile?.[NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]); + const sandbox = record(marker.retainedSandbox); + const location = savedRemoteLocation(record(profile)); + if (marker.schema !== RETIREMENT_SCHEMA || !location || sandbox.leaseId !== location.leaseId + || sandbox.providerLeaseId !== location.providerLeaseId || sandbox.remoteCwd !== location.remoteCwd) return null; + return location; +} + +/** + * Retires the inherited remote runner identity of a due same-run retry when + * durable state proves all of these: + * + * - the failed attempt is a result-less exact-checkpoint retry that recorded + * `providerEventsExist: false`, and no controller owns it; + * - this run never requested or identified a runner, invoked the adapter, + * started a tool, or committed a runner event, has no controller checkpoint + * of its own, and has no provider event or provider session of its own; + * - its saved session has no active turn and no pending runtime request; and + * - its exact process identity is the one a settled predecessor of the same + * native session recorded. + * + * Anything else, including rows saved by a release that did not record these + * facts, keeps the existing verified reattachment path. + */ +export type InheritedRunnerRetirement = "retired" | "not_applicable" | "transient"; + +export async function retireInheritedRemoteRunnerIdentity(db: Db, runId: string): Promise { + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + await tx.execute(sql`select set_config('lock_timeout', '2000', true)`); + const [candidate] = await tx.select({ companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.nativeIssueId }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1); + if (!candidate?.issueId) return "not_applicable"; + // Same lock order as same-run resumption: issue, coordinator, run. + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, candidate.companyId), eq(issues.id, candidate.issueId))).for("update").limit(1); + if (!issue) return "not_applicable"; + const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, candidate.companyId), eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.issueId, issue.id))).for("update").limit(1); + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, runId))).for("update").limit(1); + // A concurrent sweep already retired it: route it to the same resume. + if (coordinator && run && run.processPid === null && run.processGroupId === null + && record(record(run.runnerProfileJson)[NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]).schema === RETIREMENT_SCHEMA + && coordinator.phase === "retryable_failure" && coordinator.resultId === null) return "retired"; + if (!coordinator || !run || run.runtimeMode !== "native" || isNativeRunnerOwnershipHeld(run) + || !["running", "failed"].includes(run.status) || run.sessionIdAfter + || (run.processPid === null && run.processGroupId === null) || !run.nativeSessionId) return "not_applicable"; + const detail = record(coordinator.failureDetail); + if (coordinator.phase !== "retryable_failure" || coordinator.resultId !== null || coordinator.leaseOwner !== null + || detail.recoveryMode !== "exact_checkpoint_resume" || detail.providerEventsExist !== false + || typeof detail.successorRunId === "string") return "not_applicable"; + + const profile = record(run.runnerProfileJson); + const location = savedRemoteLocation(profile); + const checkpoint = record(profile.sessionCheckpoint); + const pending = checkpoint.pendingRuntimeRequests; + if (!location || checkpoint.activeTurnId != null + || (pending !== undefined && pending !== null && (!Array.isArray(pending) || pending.length > 0))) return "not_applicable"; + if (await hasNativeExecutionStartEvidence(tx, run)) return "not_applicable"; + if ((await nativeSessionProviderEvidence(tx, [run.id])).has(run.id)) return "not_applicable"; + if (await hasNativeRunnerCheckpoint(tx, { companyId: run.companyId, issueId: issue.id, runId: run.id })) return "not_applicable"; + + // The identity must be exactly the one a settled predecessor recorded. + const [predecessor] = await tx.select({ id: heartbeatRuns.id, processStartedAt: heartbeatRuns.processStartedAt }) + .from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), ne(heartbeatRuns.id, run.id), + lt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES), + run.processPid === null ? isNull(heartbeatRuns.processPid) : eq(heartbeatRuns.processPid, run.processPid), + run.processGroupId === null ? isNull(heartbeatRuns.processGroupId) : eq(heartbeatRuns.processGroupId, run.processGroupId), + )).orderBy(desc(heartbeatRuns.createdAt)).limit(1); + if (!predecessor || !sameTime(predecessor.processStartedAt, run.processStartedAt)) return "not_applicable"; + + // Keep the run's own sandbox when its lease is still live. + const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, location.leaseId), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, location.providerLeaseId), + eq(environmentLeases.status, "active"), isNull(environmentLeases.releasedAt), isNull(environmentLeases.cleanupStatus), + or(isNull(environmentLeases.expiresAt), sql`${environmentLeases.expiresAt} > now()`))).limit(1); + + await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id, agentId: run.agentId }); + await tx.update(heartbeatRuns).set({ + processPid: null, processGroupId: null, processStartedAt: null, + runnerProfileJson: { ...profile, [NATIVE_INHERITED_RUNNER_RETIREMENT_KEY]: { + schema: RETIREMENT_SCHEMA, predecessorRunId: predecessor.id, + ...(lease ? { retainedSandbox: location } : {}), + } }, + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + await appendHeartbeatRunEvent(tx, { + companyId: run.companyId, runId: run.id, agentId: run.agentId, + eventType: "lifecycle", stream: "system", level: "info", + message: "Retrying this run in its own sandbox: it never started a runner, so there was nothing to reattach to.", + payload: { reason: "inherited_runner_never_launched", predecessorRunId: predecessor.id, + retainedSandbox: Boolean(lease) }, + }); + return "retired" as const; + }).catch((error): InheritedRunnerRetirement => { + logger.warn({ err: error, runId }, "failed to retire an inherited remote runner identity"); + // A busy lock, a serialization conflict or a lost connection proves + // nothing: keep the run for the next sweep. Any other error keeps the + // existing recovery path, which reports its own cause. + return isTransientDatabaseError(error) ? "transient" : "not_applicable"; + }); +} diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 98d8444ee24..a5ed318b216 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -727,7 +727,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: { /** Any record that execution reached a runner: a requested or identified * process, an adapter invocation, a tool start, or a runner-sourced event. */ -async function hasNativeExecutionStartEvidence( +export async function hasNativeExecutionStartEvidence( tx: Pick, run: { companyId: string; id: string }, ): Promise { const [executionEvent] = await tx diff --git a/server/src/services/retained-sandbox-absent-error.ts b/server/src/services/retained-sandbox-absent-error.ts new file mode 100644 index 00000000000..45a09675356 --- /dev/null +++ b/server/src/services/retained-sandbox-absent-error.ts @@ -0,0 +1,19 @@ +/** + * The provider confirmed that a retained sandbox no longer exists: its resume + * reply named no provider lease and marked the lease expired. Resume never + * acquires a replacement; callers decide what confirmed absence permits. + */ +export class RetainedSandboxAbsentError extends Error { + constructor() { + super("The provider confirmed the retained sandbox no longer exists. No replacement was acquired."); + this.name = "RetainedSandboxAbsentError"; + } +} + +/** A resume reply that confirms the sandbox is gone, as sandbox providers report it. */ +export function resumeReplyConfirmsAbsence(reply: unknown): boolean { + if (!reply || typeof reply !== "object") return false; + const value = reply as { providerLeaseId?: unknown; metadata?: unknown }; + const metadata = value.metadata && typeof value.metadata === "object" ? value.metadata as Record : {}; + return (value.providerLeaseId === null || value.providerLeaseId === undefined) && metadata.expired === true; +}