mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-22 00:30:19 +02:00
Fence idle computer retirement before closing native transport
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
6b0bf2f791
commit
8479b2caf0
5 files changed
+72
-7
No files matched your search
@@ -246,7 +246,8 @@ export interface AdapterComputerExecutionTarget extends Omit<AdapterSandboxExecu
|
||||
launch(input: { command: string; args?: string[]; cwd?: string; env?: Record<string, string> }): Promise<unknown>;
|
||||
inspectProcess(): Promise<{ running: boolean; claim: unknown }>;
|
||||
retainWarm(idleTimeoutMs: number): Promise<void>;
|
||||
retire(): Promise<void>;
|
||||
/** False means admission superseded this owner; callers must not close its process. */
|
||||
retire(): Promise<boolean>;
|
||||
computerTool: { command: string; args: string[] };
|
||||
}
|
||||
|
||||
|
||||
@@ -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]) {
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user