From 94718ce168033cf43059e9b79e2f16c9a26436d2 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sat, 7 Nov 2026 14:51:32 -0800 Subject: [PATCH] fix(native): continue on a replacement when a retained Runner retires before claim (#19897) --- .../native-session-executor.test.ts | 2 +- .../native-runtime/native-session-executor.ts | 96 ++++++++++++------- ...plugin-runner-eviction.integration.test.ts | 46 +++++++-- .../native-runtime/plugin-runner-task.test.ts | 20 +++- .../native-runtime/plugin-runner-task.ts | 6 ++ 5 files changed, 123 insertions(+), 47 deletions(-) diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index cc914a1265f..20078cd1f0f 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -16086,7 +16086,7 @@ describe("runnerd provider runtime wiring", () => { state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []); const readWorkspaceFile = vi.fn(async () => Buffer.from("proof")); const release = vi.fn(async () => {}), resume = vi.fn(async () => "resumed" as const); - const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, rebind:vi.fn() }); + const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retiredBeforeClaim: () => false, rebind:vi.fn() }); const task = vi.fn(async (operation: unknown) => { environmentTaskOperationSchema.parse(operation); return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, releaseResumeVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index db064de7157..51704b21a96 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10811,40 +10811,38 @@ async function executePaperclipNativeSessionWithinScope( const resolveWake = () => timeController("wake_binding", () => resolveCurrentWakeCommentsBinding(input.db, input.execution.binding)); const expectedCurrentWakeComments = await (executionContext ? executionContext.wait(resolveWake) : resolveWake()); let remoteWarmRecoveryPending = remoteWarmRecovery !== undefined; + const createExecutionBackend = () => createRunnerdBackend({ + ...input, + onDiagnosticLease: ownDiagnosticLease, + providerPackAdmissionSource, + recoveryPurpose: claimedStopCleanup ? "stop_cleanup" : "execution", + executionContext, + onAccountingEvent: producerUsage?.capture, + controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { + bootId: controller.bootId, + runnerArtifact: restoredController ? { version: restoredController.runnerVersion, digest: restoredController.runnerDigest } : undefined, + } : undefined, + ...(overlapGitHubPreparation ? { prepareRunnerEnvironment } : {}), + // Durable scope and prior-run verification use the controller's + // canonical workspace identity. createRunnerdBackend separately + // projects remoteCwd into the provider execution boundary. + execution: input.execution, + runnerInstanceId: effectiveRunnerInstanceId, + pluginTaskDescription, + retainedPluginTaskOwner: existingWarmSession ? pluginTaskSessionOwners.get(existingWarmSession) : undefined, + durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, + remoteWarmRecovery: remoteWarmRecovery ? { + ...remoteWarmRecovery, + revalidate: async () => { + if (remoteWarmRecoveryPending) await remoteWarmRecovery.revalidate(leaseOwner); + }, + } : undefined, + trace, + toolTrace, + }); let runnerdBackend = input.useRunnerd && input.backend === undefined - ? await timeController("backend_setup", () => { - const create = () => createRunnerdBackend({ - ...input, - onDiagnosticLease: ownDiagnosticLease, - providerPackAdmissionSource, - recoveryPurpose: claimedStopCleanup ? "stop_cleanup" : "execution", - executionContext, - onAccountingEvent: producerUsage?.capture, - controllerPersistence: input.runnerExecutionTarget?.kind === "remote" ? { - bootId: controller.bootId, - runnerArtifact: restoredController ? { version: restoredController.runnerVersion, digest: restoredController.runnerDigest } : undefined, - } : undefined, - ...(overlapGitHubPreparation ? { prepareRunnerEnvironment } : {}), - // Durable scope and prior-run verification use the controller's - // canonical workspace identity. createRunnerdBackend separately - // projects remoteCwd into the provider execution boundary. - execution: input.execution, - runnerInstanceId: effectiveRunnerInstanceId, - pluginTaskDescription, - retainedPluginTaskOwner: existingWarmSession ? pluginTaskSessionOwners.get(existingWarmSession) : undefined, - durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, - remoteWarmRecovery: remoteWarmRecovery ? { - ...remoteWarmRecovery, - revalidate: async () => { - if (remoteWarmRecoveryPending) await remoteWarmRecovery.revalidate(leaseOwner); - }, - } : undefined, - trace, - toolTrace, - }); - return executionContext ? executionContext.wait(create) : create(); - }) + ? await timeController("backend_setup", () => executionContext ? executionContext.wait(createExecutionBackend) : createExecutionBackend()) : null; if (remoteWarmRecovery) { if (executionContext) await executionContext.wait(() => remoteWarmRecovery.revalidate(leaseOwner)); @@ -10877,7 +10875,7 @@ async function executePaperclipNativeSessionWithinScope( durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, toolTrace }); } } - const remoteCleanupLease = input.runnerExecutionTarget?.kind === "remote" && + const readRemoteCleanupLease = async () => input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.leaseId ? await input.db.select({ provider: environmentLeases.provider, providerLeaseId: environmentLeases.providerLeaseId }) .from(environmentLeases).where(and( @@ -10885,6 +10883,26 @@ async function executePaperclipNativeSessionWithinScope( eq(environmentLeases.id, input.runnerExecutionTarget.leaseId), )).then(rows => rows[0]) : null; + let remoteCleanupLease = await readRemoteCleanupLease(); + const continueAfterUnavailableReuse = async (error: unknown) => { + const target = input.runnerExecutionTarget; + if (!(error instanceof Error) || error.message !== "native_runner_reuse_unavailable" || !existingWarmSession + || pluginTaskSessionOwners.get(existingWarmSession)?.retiredBeforeClaim() !== true + || remoteWarmRecovery || warmConfigDigest === null || target?.kind !== "remote" || target.transport !== "sandbox" + || !target.runnerTask || runnerdBackend === null) return false; + const retainedTaskId = target.sandboxLeaseAcquisition?.providerLeaseId; + await existingWarmSession.close({ reason: "retained Runner was unavailable for reuse" }); + const inspect = () => target.runnerTask!({ kind: "describe" }, executionContext ? { executionContext } : undefined); + const description = environmentTaskResultSchema.parse(await (executionContext ? executionContext.wait(inspect) : inspect())); + if (description.kind !== "description" || description.taskId === retainedTaskId) return false; + existingWarmSession = undefined; + pluginTaskDescription = undefined; + await finishStartupContextForFallback(); + persistedWarmSession = loadWarmNativeCheckpoint(input.execution, warmConfigDigest, target.kind); + runnerdBackend = await createExecutionBackend(); + remoteCleanupLease = await readRemoteCleanupLease(); + return true; + }; nativeSessionExecuteStartedAtMs = Date.now(); native = await trace.measure( "native.session.execute", @@ -10894,7 +10912,7 @@ async function executePaperclipNativeSessionWithinScope( startedAtMs: nativeSessionExecuteStartedAtMs, }); trace.activate(runnerSessionStartupScope); - const result = await trace.run(runnerSessionStartupScope, () => + const runExecution = () => executeNativeSession({ resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner", getFreshSessionHandoff: input.getFreshSessionHandoff, @@ -11274,8 +11292,14 @@ async function executePaperclipNativeSessionWithinScope( ); } }, - }), - ); + }); + let result: Awaited>; + try { + result = await trace.run(runnerSessionStartupScope, runExecution); + } catch (error) { + if (!await continueAfterUnavailableReuse(error)) throw error; + result = await trace.run(runnerSessionStartupScope, runExecution); + } await trace.end(runnerSessionStartupScope, { outcome: result.terminal.runTerminalState === "succeeded" ? "ok" : "failed", diff --git a/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts b/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts index 079ad9e768a..f5c55284379 100644 --- a/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts +++ b/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts @@ -45,10 +45,12 @@ const parkId = (source: DurableRecoveryIdentity) => `park_${sha(JSON.stringify(O describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execution entry", () => { const environment = nativeRestartFixture(); const nativeIt = process.platform === "win32" ? it.skip : realProcessIt; - nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "detached", "reused lease", "prepared copy", "missing stop", "missing state", "foreign state", "working state"] as const).map(mode => [harness, mode] as const))])("continues %s after graceful eviction with a %s Core owner", { timeout: 60_000 }, async ([harness, mode], context) => { + nativeIt.for([...(["codex", "claude"] as const).flatMap(harness => (["cached", "detached", "reused lease", "after inspection", "bare unavailable", "prepared copy", "missing stop", "missing state", "foreign state", "working state"] as const).map(mode => [harness, mode] as const))])("continues %s after graceful eviction with a %s Core owner", { timeout: 60_000 }, async ([harness, mode], context) => { if (harness === "claude" && !existsSync(fakeAcpxSidecar)) context.skip(); const detached = mode === "detached" || mode === "missing state" || mode === "foreign state"; const denied = mode.startsWith("missing") || mode.endsWith("state"); + const unavailableResume = mode === "after inspection" || mode === "bare unavailable"; + const reusesLease = mode === "reused lease" || unavailableResume; const instructionCopy = (id: string) => ({ runId: id, root: workspace, preparationKey: "unchanged-fixture-instructions", hasChanges: async () => false, collectStopped: async () => {} }); const first = await environment.seedRun(`RELEASE-ENTRY-${harness}-${mode}`); const { db, runId, issueId, native } = first; @@ -132,6 +134,7 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu const releases: DurableRecoveryIdentity[] = [], claims: DurableRecoveryIdentity[] = []; let handle: RunnerProcessHandle | undefined, submissions = 0, cleaning = false, sourceReadRequests = 0, sourceReadResponses = 0; let pendingFollowup: Promise | undefined; + let inspectedLive = false, resumeAttempts = 0, acquisitions = 0, evictRetained: (() => Promise) | undefined; const providerStarts = async () => (await readFile(calls, "utf8")).trim().split(/\r?\n/) .filter(line => harness === "codex" ? line === "process-start" : JSON.parse(line).processStart).length; const state = async (taskId: string = taskA) => JSON.parse(await readFile(join(runnerRoot(taskId), "runner-state.json"), "utf8")); @@ -139,8 +142,14 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu const task = async (raw: unknown, authorityRunId: string, taskId: string): Promise => { const host = hosts.get(taskId)!; const operation = environmentTaskOperationSchema.parse(raw); - if (operation.kind === "describe") return { kind: "description", taskId, launchPreparationVersion: 1, - projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact }; + if (operation.kind === "describe") { + if (unavailableResume && authorityRunId !== runId && !inspectedLive) { + expect(handles.get(taskA)?.isExitConfirmed?.()).toBe(false); + inspectedLive = true; + } + return { kind: "description", taskId, launchPreparationVersion: 1, + projectAuthCheckVersion: 1, warmTaskVersion: 1, releaseResumeVersion: 1, runner: artifact }; + } if (operation.kind === "prepare") { expect(operation.requirements.identity.runId).toBe(authorityRunId); expect(operation.accountIdentity).toBe("fixture-account"); @@ -205,6 +214,13 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu if (reply.status !== "parked") throw new Error("Fixture native release refused"); releases.push(operation.identity); return { kind: "released", taskId }; } + if (operation.kind === "resume" && unavailableResume) { + resumeAttempts++; + expect(inspectedLive).toBe(true); + if (mode === "after inspection") await evictRetained!(); + expect(handles.get(taskA)?.isExitConfirmed?.()).toBe(mode === "after inspection"); + return { kind: "resume", taskId, outcome: "unavailable" }; + } if (operation.kind === "resume") { const retainedProviderCount = await providerStarts(); const reply = await host.request({ op: "claim", id: `claim_${sha(operation.bootstrapTicket)}`, parkId: parkId(operation.identity), bootstrapTicket: operation.bootstrapTicket }); @@ -213,7 +229,10 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu claims.push(operation.identity); return { kind: "resume", taskId, outcome: "resumed" }; } - if (operation.kind === "stop") { await host.close(); } + if (operation.kind === "stop") { + if (mode === "after inspection" && taskId === taskA) await new Promise(done => setTimeout(done, 300)); + await host.close(); + } if (operation.kind === "complete") { expect(handles.get(taskId)?.isExitConfirmed?.()).toBe(true); } return { kind: "accepted", taskId }; }; @@ -227,6 +246,7 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu }; const proof = async () => (await supervisor()).releasedRunner; const workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }), call: async (_id: string, method: string, params: any, _timeout?: number, _unused?: unknown, settled?: () => void) => { + if (method === "environmentAcquireLease") acquisitions++; if (method === "environmentAcquireLease") return { providerLeaseId: taskB, metadata: { creationOperationId: params.creationOperationId } }; try { return await task(params.operation, params.runId, params.taskId); } finally { settled?.(); } } } as unknown as PluginWorkerManager; @@ -237,7 +257,7 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu const execute = async (current: typeof execution, leaseId: string) => { await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: current }, runnerInstanceId: native.runnerInstanceId, completionContractId: contract!.id, completionContractSha256: contract!.canonicalSha256, nativeSessionId: current.session.normalizedSessionId }) .where(eq(heartbeatRuns.id, current.binding.runId)); - const target = mode === "reused lease" ? await resolveEnvironmentExecutionTarget({ db, companyId, adapterType: "paperclip_runner", + const target = reusesLease ? await resolveEnvironmentExecutionTarget({ db, companyId, adapterType: "paperclip_runner", environment: { id: environmentId, driver: "plugin", config: { pluginKey: pluginId, driverKey: "tasks" } }, leaseId, lease: await environmentService(db).getLeaseById(leaseId), leaseMetadata: metadata, environmentRuntime: environmentRuntimeService(db, { pluginWorkerManager: workers }) }) : { @@ -274,7 +294,8 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success" }).where(eq(environmentLeases.id, leaseA)); } }; - if (mode !== "reused lease") await evict(); + evictRetained = evict; + if (!reusesLease) await evict(); const admitNext = async () => { const runB = randomUUID(), leaseB = randomUUID(); await db.insert(heartbeatRuns).values({ id: runB, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native" }); @@ -282,8 +303,8 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu await db.update(issues).set({ executionRunId: runB, status: "in_progress" }).where(eq(issues.id, issueId)); const [oldLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseA)); await db.insert(environmentLeases).values({ id: leaseB, companyId, environmentId, heartbeatRunId: runB, issueId, executionWorkspaceId: workspaceId, - provider: providerKey, providerLeaseId: mode === "reused lease" ? taskA : taskB, status: "active", leasePolicy: "reuse_by_environment", - metadata: mode === "reused lease" ? { ...oldLease!.metadata, runnerTaskReused: true, runnerTaskPreviousLeaseId: leaseA, + provider: providerKey, providerLeaseId: reusesLease ? taskA : taskB, status: "active", leasePolicy: "reuse_by_environment", + metadata: reusesLease ? { ...oldLease!.metadata, runnerTaskReused: true, runnerTaskPreviousLeaseId: leaseA, runnerTaskExecutionStopped: false } : { ...metadata, remoteCwd: workspace } }); if (mode === "reused lease") { expect(handles.get(taskA)!.isExitConfirmed?.()).toBe(false); @@ -295,6 +316,12 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu preparationKey: "unchanged-fixture-instructions", adopt: async () => instructionCopy(runB) }) : null; if (mode === "prepared copy") expect(releasePreparation).toBeTypeOf("function"); pendingFollowup = execute(next, leaseB); + if (mode === "bare unavailable") { + await expect(pendingFollowup).rejects.toThrow("native_runner_reuse_unavailable"); + expect({ resumeAttempts, acquisitions, submissions, claims }).toEqual({ resumeAttempts: 1, acquisitions: 0, submissions: 1, claims: [] }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runB))).toHaveLength(0); + return runB; + } if (denied) { const refusal = mode === "working state" ? "checkpoint_not_settled" : mode === "foreign state" ? "Plugin Runner checkpoint identity mismatch" @@ -316,12 +343,13 @@ describeEmbeddedPostgres("plugin Runner graceful eviction through the Core execu return runB; }; const runB = await admitNext(); - if (denied) return; + if (denied || mode === "bare unavailable") return; await waitForCondition("automatic release B", () => releases.length === 2); expect(await proof()).toMatchObject({ source: releases[1], taskId: taskB }); expect((await supervisor()).snapshot.providerSessionId).toBe(providerSessionId); const methods = (await readFile(calls, "utf8")).trim().split(/\r?\n/); expect(submissions).toBe(2); + if (mode === "after inspection") expect({ resumeAttempts, acquisitions }).toEqual({ resumeAttempts: 1, acquisitions: 1 }); expect(releases[1]!.runId).toBe(runB); expect(claims).toEqual([]); if (harness === "codex") { diff --git a/server/src/services/native-runtime/plugin-runner-task.test.ts b/server/src/services/native-runtime/plugin-runner-task.test.ts index 34b7de9bbe0..4132b12532a 100644 --- a/server/src/services/native-runtime/plugin-runner-task.test.ts +++ b/server/src/services/native-runtime/plugin-runner-task.test.ts @@ -653,7 +653,25 @@ describe("plugin Runner release/resume facade", () => { const owner = createPluginRunnerTaskOwner({ ...input("codex"), task }); expect(await owner.driver.adoptExistingRunner!.isAlive()).toBe(false); await expect(owner.resume(requirements.identity, "transient-ticket")).resolves.toBe("unavailable"); - expect(task.mock.calls.map(([operation]) => operation.kind)).toEqual(["status", "resume"]); + expect(task.mock.calls.map(([operation]) => operation.kind)).toEqual(["status", "resume", "status"]); + expect(owner.retiredBeforeClaim()).toBe(false); // stopped, but no saved session state + }); + + it("records a refused resume as retired only when the exact task stopped with saved session state", async () => { + for (const [status, retired] of [ + [{ kind: "status", taskId: "task", phase: "cancelled", executionStopped: true, sessionStateAvailable: true }, true], + [{ kind: "status", taskId: "task", phase: "running", executionStopped: false, sessionStateAvailable: false }, false], + [new Error("fixture status reply lost"), false], + ] as const) { + const task = vi.fn(async (operation: any): Promise => { + if (operation.kind === "resume") return { kind: "resume", taskId: "task", outcome: "unavailable" }; + if (status instanceof Error) throw status; + return status; + }); + const owner = createPluginRunnerTaskOwner({ ...input("codex"), task }); + await expect(owner.resume(requirements.identity, "transient-ticket")).resolves.toBe("unavailable"); + expect(owner.retiredBeforeClaim()).toBe(retired); + } }); it.each(["release", "resume"] as const)("validates exact task and closed %s result", async operation => { diff --git a/server/src/services/native-runtime/plugin-runner-task.ts b/server/src/services/native-runtime/plugin-runner-task.ts index cd020ceb0a1..f8d87ac1554 100644 --- a/server/src/services/native-runtime/plugin-runner-task.ts +++ b/server/src/services/native-runtime/plugin-runner-task.ts @@ -73,6 +73,7 @@ export function createPluginRunnerTaskOwner(input: { let confirmStopped: (code: number) => void = () => {}; let stoppedState: Extract | undefined; let completed = false; + let retiredBeforeClaim = false; let rejectFailure!: (error: unknown) => void; const failure = new Promise((_resolve, reject) => { rejectFailure = reject; }); void failure.catch(() => undefined); @@ -413,12 +414,17 @@ export function createPluginRunnerTaskOwner(input: { const result = await call(environmentTaskOperationSchema.parse({ kind: "resume", identity, bootstrapTicket })); requireReleaseResume(); if (result.kind !== "resume") throw new Error("Plugin Runner resume result unavailable"); + if (result.outcome === "unavailable") { + const state = await call({ kind: "status" }).catch(() => undefined); + retiredBeforeClaim = state?.kind === "status" && state.executionStopped === true && state.sessionStateAvailable === true; + } return result.outcome; }; return { launcher, registration, driver, readWorkspaceFile, release, resume, // This is a lease-bound Launcher receipt, not a terminal phase or a sent // stop request. Completion also proves credential/writeback obligations ran. isRetired: () => stoppedState?.executionStopped === true && completed && !detached, + retiredBeforeClaim: () => retiredBeforeClaim, rebind(next: typeof input) { if (!launched || stopRequested || stoppedState || detached || next.taskId !== taskId || next.harness !== input.harness || input.accountIdentity !== next.accountIdentity || input.credentialIdentity !== next.credentialIdentity ||