From 46fd97371a13cdb2672d10b408ebb07da7bc66a3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 10 Oct 2026 09:08:30 -0500 Subject: [PATCH] Wait for terminal computer turn cleanup before admitting follow-ups --- .../computers/application/service.test.ts | 56 +++++++++++++++++++ .../modules/computers/application/service.ts | 25 +++++++-- 2 files changed, 75 insertions(+), 6 deletions(-) diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index 03c6ea383c..978ca3210c 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -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(); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 0c8487b4b0..38cf67056a 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -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",