Wait for terminal computer turn cleanup before admitting follow-ups

This commit is contained in:
Dotta committed 2026-10-10 09:09:12 -05:00
1 parent dd8a5c38af
commit 46fd97371a
2 files changed
+75 -6

No files matched your search

@@ -305,6 +305,62 @@ describe("computer ownership", () => {
await Promise.all([first, second]);
expect(f.backend.stop).toHaveBeenCalledOnce();
});
it.each(["retire", "warm"] as const)("waits for terminal predecessor %s before admitting its follow-up", async (transition) => {
const f = fixture();
await f.attach();
const first = await f.admit();
const claim = await first.launch({ command: "runnerd" });
vi.mocked(f.repository.runState).mockResolvedValue("terminal");
const wait = vi.fn(async () => {
const incumbent = (await f.repository.get(f.scope)).ledger.owners.find(owner => owner.id === first.owner.ownerId)!;
expect(incumbent.phase).toBe("active");
expect(incumbent.generation).toBe(first.owner.generation);
expect(f.backend.retire).not.toHaveBeenCalled();
if (transition === "retire") await f.service.retire({ ...f.scope, owner: first.owner });
else await f.service.retainWarm({ ...f.scope, owner: first.owner, idleTimeoutMs: 60_000 });
});
const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait });
const next = await waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 });
expect(wait).toHaveBeenCalledOnce();
expect(f.repository.runState).toHaveBeenCalledWith(expect.objectContaining(f.scope), "run", "agent");
if (transition === "warm") {
expect(next.owner.ownerId).toBe(first.owner.ownerId);
expect(next.owner.generation).toBe(first.owner.generation + 1);
expect((await next.inspectProcess()).claim?.nonce).toBe(claim.nonce);
expect(f.backend.retire).not.toHaveBeenCalled();
} else {
expect(next.owner.ownerId).not.toBe(first.owner.ownerId);
expect(f.backend.retire).toHaveBeenCalledOnce();
}
});
it.each(["active", "missing"] as const)("does not wait on or supersede a %s predecessor", async (state) => {
const f = fixture();
await f.attach();
const first = await f.admit();
vi.mocked(f.repository.runState).mockResolvedValue(state);
const wait = vi.fn(async () => {});
const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait });
await expect(waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 }))
.rejects.toMatchObject({ code: "conflict" });
expect(wait).not.toHaveBeenCalled();
expect(f.backend.retire).not.toHaveBeenCalled();
expect((await f.repository.get(f.scope)).ledger.owners[0]?.generation).toBe(first.owner.generation);
});
it("bounds the wait for terminal predecessor teardown without forcing retirement", async () => {
const f = fixture();
await f.attach();
await f.admit();
vi.mocked(f.repository.runState).mockResolvedValue("terminal");
const wait = vi.fn(async () => {});
const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait });
await expect(waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 }))
.rejects.toMatchObject({ code: "conflict" });
expect(wait).toHaveBeenCalledTimes(2);
expect(f.backend.retire).not.toHaveBeenCalled();
});
it("waits for a normal provider snapshot stop before admitting the next turn", async () => {
const f = fixture();
await f.attach();
@@ -15,6 +15,14 @@ import {
} from "../domain/ledger.js";
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
// A terminal run can enqueue its successor before remote teardown has finished.
// Keep the incumbent owner fenced until its own cleanup or warm handoff settles.
class ComputerSessionBusyError extends ComputerError {
constructor(readonly runId: string | undefined, readonly agentId: string | undefined) {
super("conflict", "Computer session already has an active turn");
}
}
type Scope = { companyId: string; environmentId: string };
export function createComputerService(
repository: ComputerRepository,
@@ -54,8 +62,16 @@ export function createComputerService(
)
throw error;
const current = await repository.get(scope);
if (current.ledger.status !== "attached" || !current.ledger.action)
throw error;
if (current.ledger.status !== "attached") throw error;
if (error instanceof ComputerSessionBusyError && error.runId && error.agentId) {
// A terminal database status does not prove the process has stopped.
// Wait for the existing owner transition; never retire it here or
// advance its generation while final commands might still run.
if (await repository.runState(scope, error.runId, error.agentId) !== "terminal") throw error;
await wait(1000);
continue;
}
if (!current.ledger.action) throw error;
await reconcileRecord(current);
if ((await repository.get(scope)).ledger.action) await wait(1000);
}
@@ -366,10 +382,7 @@ finally:
o.phase !== "retired",
);
if (owner && owner.phase !== "warm")
throw new ComputerError(
"conflict",
"Computer session already has an active turn",
);
throw new ComputerSessionBusyError(owner.runId, owner.agentId);
if (owner && expired(owner, now()))
throw new ComputerError(
"conflict",