diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index d682f9aad02..dbffa195aa9 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -246,7 +246,8 @@ export interface AdapterComputerExecutionTarget extends Omit }): Promise; inspectProcess(): Promise<{ running: boolean; claim: unknown }>; retainWarm(idleTimeoutMs: number): Promise; - retire(): Promise; + /** False means admission superseded this owner; callers must not close its process. */ + retire(): Promise; computerTool: { command: string; args: string[] }; } diff --git a/packages/adapter-utils/src/runner-connectivity.test.ts b/packages/adapter-utils/src/runner-connectivity.test.ts index 3bc7315a004..e345a7ccc82 100644 --- a/packages/adapter-utils/src/runner-connectivity.test.ts +++ b/packages/adapter-utils/src/runner-connectivity.test.ts @@ -222,7 +222,7 @@ it("uses each computer owner's allocated port without changing the sandbox defau resourceAuthority: { kind: "computer-owner" as const, computerId: "machine", ownerId: "owner-a", generation: 2 }, fileAuthority: { kind: "remote-persistent" as const, placementId: "placement", root: "/home/user/project", agentHome: "/home/user/agent-a" }, launch: async () => ({}), inspectProcess: async () => ({ running: true, claim: {} }), - retainWarm: async () => {}, retire: async () => {}, computerTool: { command: "cua-driver", args: ["mcp"] }, + retainWarm: async () => {}, retire: async () => true, computerTool: { command: "cua-driver", args: ["mcp"] }, }; expect(adapterExecutionTargetIsCommandBacked(base)).toBe(true); for (const port of [45101, 45102]) { diff --git a/server/src/services/computer-environment-driver.ts b/server/src/services/computer-environment-driver.ts index 3f36fa0a205..1a1fea64e73 100644 --- a/server/src/services/computer-environment-driver.ts +++ b/server/src/services/computer-environment-driver.ts @@ -122,6 +122,7 @@ export async function computerExecutionTarget(db: Db, lease: EnvironmentLease, i const result = await computers.retire({ ...scope, owner }); if (result.retired) await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success", remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) }); + return result.retired; }, getRunnerIngressEndpoint: ({ port, path }) => { if (port !== binding.listenerPort) throw new Error("computer_runner_port_mismatch"); 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 a9fd53f651e..79c6e7b0efe 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -7405,7 +7405,7 @@ describe("native warm session supervision", () => { const name = `computer-owner-${suffix}`; const current = { ...execution, binding: { ...execution.binding, runId: name, agentId: name, executionWorkspaceId: name }, session: { ...execution.session, normalizedSessionId: name, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1; - const close = vi.fn(async () => undefined), retainWarm = vi.fn(async () => undefined), retire = vi.fn(async () => undefined); + const close = vi.fn(async () => undefined), retainWarm = vi.fn(async () => undefined), retire = vi.fn(async () => true); const target = { kind: "remote" as const, transport: "computer" as const, environmentId: "shared-computer", remoteCwd: `/home/user/${name}`, listenerPort: suffix === "a" ? 45001 : 45002, resourceAuthority: { kind: "computer-owner", computerId: "computer", ownerId: name, generation: 1 }, @@ -7427,12 +7427,52 @@ describe("native warm session supervision", () => { expect(result.closed).toBe(1); expect(owners[0]!.close).toHaveBeenCalledOnce(); expect(owners[0]!.retire).toHaveBeenCalledOnce(); - expect(owners[0]!.close.mock.invocationCallOrder[0]).toBeLessThan(owners[0]!.retire.mock.invocationCallOrder[0]!); + expect(owners[0]!.retire.mock.invocationCallOrder[0]).toBeLessThan(owners[0]!.close.mock.invocationCallOrder[0]!); expect(owners[1]!.close).not.toHaveBeenCalled(); expect(owners[1]!.retire).not.toHaveBeenCalled(); await closeWarmNativeSessionsForRun({ runId: owners[1]!.runId, reason: "fixture cleanup" }); }); + it("keeps a computer transport alive when its old idle timer loses to warm admission", async () => { + const name = "computer-idle-admission-race"; + const current = { ...execution, binding: { ...execution.binding, runId: `${name}-one`, agentId: name, executionWorkspaceId: name }, + session: { ...execution.session, normalizedSessionId: name, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1; + const close = vi.fn(async () => undefined); + const detachControllerForRestart = vi.fn(async () => undefined); + const session = { close, detachControllerForRestart }; + let generation = 1; + const retire = vi.fn(async () => generation === 1); + const target = { kind: "remote", transport: "computer", environmentId: name, remoteCwd: `/home/user/${name}`, + resourceAuthority: { kind: "computer-owner", computerId: name, ownerId: name, generation: 1 }, + retainWarm: vi.fn(async () => undefined), retire }; + const result = { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, turnId: name, + normalizedSessionId: name, providerSessionId: name, driverKind: "test", driverVersion: "1", + nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession?.(session); return result; + }).mockImplementationOnce(async options => { + expect(options.existingSession).toBe(session); + await options.onSession?.(session); return result; + }); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + try { + await executePaperclipNativeSession({ db: leaseDb(current), execution: current, runnerInstanceId: name, runnerExecutionTarget: target as never }); + generation = 2; // The durable admission committed before local reservation. + await vi.advanceTimersByTimeAsync(60_000); + expect(retire).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalled(); + expect(detachControllerForRestart).not.toHaveBeenCalled(); + const next = { ...current, binding: { ...current.binding, runId: `${name}-two` } }; + const nextRetire = vi.fn(async () => true); + await executePaperclipNativeSession({ db: leaseDb(next), execution: next, runnerInstanceId: name, + runnerExecutionTarget: { ...target, resourceAuthority: { ...target.resourceAuthority, generation: 2 }, retire: nextRetire } as never }); + await closeWarmNativeSessionsForRun({ runId: next.binding.runId, reason: "fixture cleanup" }); + expect(nextRetire).toHaveBeenCalledOnce(); + expect(detachControllerForRestart).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalled(); + } finally { vi.useRealTimers(); } + }); + describe("managed directory warm checkpoints", () => { const result = { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, turnId: "turn", normalizedSessionId: "managed", providerSessionId: "provider", driverKind: "test", driverVersion: "1", diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 7c4e4c627f8..d5155806671 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -470,17 +470,31 @@ type WarmNativeSession = { lastActivityAt: string; }; +class ComputerOwnerSupersededError extends Error { + constructor() { super("computer_owner_retirement_superseded"); } +} + async function closeWarmNativeSession(entry: WarmNativeSession, reason: string, preserveInstructionsForRunId?: string) { + // Computer admission can advance while an old idle timer is firing. Only + // the exact durable owner may stop the runner or its controller transport. + if (entry.computerTarget && !(await entry.computerTarget.retire())) { + throw new ComputerOwnerSupersededError(); + } // Revoke before awaiting process retirement/checkpoint IO. const stopping = entry.githubAccess?.stop(); try { - try { await entry.session.close({ reason }); } + try { + if (entry.computerTarget && entry.session.detachControllerForRestart) { + await entry.session.detachControllerForRestart(); + } else { + await entry.session.close({ reason }); + } + } catch (error) { try { await entry.instructionWorkingCopy?.retirementFailed?.(); } catch (receiptError) { throw new AggregateError([error, receiptError], "Warm instruction retirement and receipt both failed"); } throw error; } - await entry.computerTarget?.retire(); await entry.instructionWorkingCopy?.collectStopped(); if (entry.instructionCopy?.runId !== preserveInstructionsForRunId) await entry.instructionCopy?.collectStopped(); } @@ -6191,7 +6205,16 @@ async function releaseWarmNativeSession( current.closeOnReleaseReason = "warm native session idle timeout"; void closeWarmNativeSession(current, current.closeOnReleaseReason) .then(() => { if (warmNativeSessions.get(sessionId) === current) warmNativeSessions.delete(sessionId); }) - .catch(() => { if (!current.instructionWorkingCopy?.runId && warmNativeSessions.get(sessionId) === current) warmNativeSessions.delete(sessionId); }) + .catch((error) => { + if (error instanceof ComputerOwnerSupersededError) { + // Admission owns the next generation. Leave this transport available + // for its imminent warm reservation instead of closing or deleting it. + current.closeOnReleaseReason = undefined; + current.idleTimer = null; + return; + } + if (!current.instructionWorkingCopy?.runId && warmNativeSessions.get(sessionId) === current) warmNativeSessions.delete(sessionId); + }) .finally(() => { current.busy = false; }); }, idleTimeoutMs); entry.idleTimer.unref();