diff --git a/src/node-host/runtime.computer.test.ts b/src/node-host/runtime.computer.test.ts index 58ef558bb716..0ff1e1c1e376 100644 --- a/src/node-host/runtime.computer.test.ts +++ b/src/node-host/runtime.computer.test.ts @@ -120,6 +120,45 @@ async function startComputer(ephemeral = true, prepare?: () => Promise) { } describe("private worker computer runtime", () => { + it("joins watcher and disconnect cleanup until physical computer close settles", async () => { + const host = await startComputer(); + const physicalClose = createDeferredCore(); + let physicalCloseFinished = false; + host.close.mockImplementationOnce(async () => { + await physicalClose.promise; + physicalCloseFinished = true; + }); + let closing: Promise | undefined; + try { + expect( + await host.invoke({ + operation: "snapshot", + providerGeneration: descriptor.provider.generation, + params: { executionId }, + }), + ).toMatchObject({ ok: true }); + let closed = false; + closing = host.runtime.close().then(() => { + closed = true; + }); + await vi.waitFor(() => expect(host.close).toHaveBeenCalledOnce()); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(physicalCloseFinished).toBe(false); + expect(closed).toBe(false); + + physicalClose.resolve(); + await closing; + expect(physicalCloseFinished).toBe(true); + expect(host.close).toHaveBeenCalledOnce(); + } finally { + physicalClose.resolve(); + await closing; + await host.runtime.close(); + } + }); + it("awaits the registered provider preparation before publishing the first manifest", async () => { const gate = createDeferredCore(); const prepare = vi.fn(() => gate.promise); diff --git a/src/plugins/computer-use-contract.test.ts b/src/plugins/computer-use-contract.test.ts index 5fd7f7a2da6a..e5e85f2d6cd8 100644 --- a/src/plugins/computer-use-contract.test.ts +++ b/src/plugins/computer-use-contract.test.ts @@ -1,5 +1,6 @@ import { readFileSync } from "node:fs"; import { describe, expect, it, vi } from "vitest"; +import { createDeferredCore } from "../shared/deferred.js"; import { COMPUTER_STALE_OBSERVATION, COMPUTER_USE_V2_ACTION_NAMES, @@ -289,6 +290,166 @@ describe("Computer Use wire contract", () => { }); describe("Computer Use provider registration", () => { + it.each(["same", "different", "same-close", "different-close"] as const)( + "preserves queued acquisition and close order for %s", + async (laterOwner) => { + const firstId = "123e4567-e89b-42d3-a456-426614174000"; + const nextId = "223e4567-e89b-42d3-a456-426614174000"; + const laterId = laterOwner.startsWith("same") + ? nextId + : "323e4567-e89b-42d3-a456-426614174000"; + const closeLater = laterOwner.endsWith("-close"); + const commands: OpenClawPluginNodeHostCommand[] = []; + const firstClose = createDeferredCore(); + const nextClose = createDeferredCore(); + const close = vi + .fn(async () => {}) + .mockImplementationOnce(() => firstClose.promise) + .mockImplementationOnce(() => nextClose.promise); + const openExecution = vi.fn(async () => ({ + snapshot: async () => "snapshot", + act: async () => "act", + close, + })); + registerComputerUseProvider( + { registerNodeHostCommand: (command) => commands.push(command) }, + { + id: "fixture", + label: "Fixture", + capabilities: () => ({ + contractVersion: 2, + provider: { id: "fixture", label: "Fixture", generation: "generation-1" }, + actions: ["screenshot"], + targets: ["screen"], + deliveryModes: ["foreground"], + observations: ["image"], + features: { recording: false, agentCursor: false, multiDisplay: false }, + }), + isAvailable: () => true, + openExecution, + }, + ); + const snapshot = commands[0]!; + const computer = commands[1]!; + await snapshot.handle(JSON.stringify({ executionId: firstId })); + const retiringFirst = computer.handle( + JSON.stringify({ executionId: firstId, action: "__close_execution" }), + ); + const openingNext = snapshot.handle(JSON.stringify({ executionId: nextId })); + const retiringNext = computer.handle( + JSON.stringify({ executionId: nextId, action: "__close_execution" }), + ); + const laterSnapshot = snapshot.handle(JSON.stringify({ executionId: laterId })); + const retiringLater = closeLater + ? computer.handle(JSON.stringify({ executionId: laterId, action: "__close_execution" })) + : undefined; + const laterSettled = vi.fn(); + const observedLater = laterSnapshot.then(laterSettled, laterSettled); + const laterCloseSettled = vi.fn(); + const observedLaterClose = retiringLater?.then(laterCloseSettled, laterCloseSettled); + const operations = Promise.allSettled([ + retiringFirst, + openingNext, + retiringNext, + laterSnapshot, + ...(retiringLater ? [retiringLater] : []), + ]); + try { + firstClose.resolve(); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(openExecution).toHaveBeenCalledTimes(2); + expect(close).toHaveBeenCalledTimes(2); + expect(laterSettled).not.toHaveBeenCalled(); + expect(laterCloseSettled).not.toHaveBeenCalled(); + nextClose.resolve(); + expect(await operations).toEqual([ + { status: "fulfilled", value: '{"ok":true}' }, + { status: "fulfilled", value: "snapshot" }, + { status: "fulfilled", value: '{"ok":true}' }, + { status: "fulfilled", value: "snapshot" }, + ...(closeLater ? [{ status: "fulfilled", value: '{"ok":true}' }] : []), + ]); + expect(openExecution).toHaveBeenCalledTimes(3); + expect(close).toHaveBeenCalledTimes(closeLater ? 3 : 2); + } finally { + firstClose.resolve(); + nextClose.resolve(); + await operations; + await observedLater; + await observedLaterClose; + await snapshot.onDisconnect?.(); + } + }, + ); + + it("retains terminal close failure without confusing it with failed-open recovery", async () => { + const executionId = "123e4567-e89b-42d3-a456-426614174000"; + const otherId = "223e4567-e89b-42d3-a456-426614174000"; + const commands: OpenClawPluginNodeHostCommand[] = []; + const openFailure = new Error("native driver startup failed"); + const closeFailure = new Error("native driver shutdown failed"); + const failedOpening = + createDeferredCore>>(); + const physicalClose = vi.fn(async () => { + throw closeFailure; + }); + let terminalClose: Promise | undefined; + const close = vi.fn(() => (terminalClose ??= physicalClose())); + const openExecution = vi + .fn(async () => ({ + snapshot: async () => "snapshot", + act: async () => "act", + close, + })) + .mockImplementationOnce(() => failedOpening.promise); + registerComputerUseProvider( + { registerNodeHostCommand: (command) => commands.push(command) }, + { + id: "fixture", + label: "Fixture", + capabilities: () => ({ + contractVersion: 2, + provider: { id: "fixture", label: "Fixture", generation: "generation-1" }, + actions: ["screenshot"], + targets: ["screen"], + deliveryModes: ["foreground"], + observations: ["image"], + features: { recording: false, agentCursor: false, multiDisplay: false }, + }), + isAvailable: () => true, + openExecution, + }, + ); + const snapshot = commands[0]!; + const computer = commands[1]!; + const params = JSON.stringify({ executionId }); + const closeParams = JSON.stringify({ executionId, action: "__close_execution" }); + const opening = snapshot.handle(params); + const closingFailedOpen = computer.handle(closeParams); + const failedOpenResults = Promise.allSettled([opening, closingFailedOpen]); + failedOpening.reject(openFailure); + expect(await failedOpenResults).toEqual([ + { status: "rejected", reason: openFailure }, + { status: "rejected", reason: openFailure }, + ]); + await expect(snapshot.handle(params)).resolves.toBe("snapshot"); + await expect(computer.handle(closeParams)).rejects.toBe(closeFailure); + await expect( + computer.handle(JSON.stringify({ executionId: otherId, action: "__close_execution" })), + ).resolves.toBe('{"ok":true}'); + expect(close).toHaveBeenCalledOnce(); + + await expect(computer.handle(closeParams)).rejects.toBe(closeFailure); + await expect(snapshot.handle(JSON.stringify({ executionId: otherId }))).rejects.toBe( + closeFailure, + ); + expect(openExecution).toHaveBeenCalledTimes(2); + expect(close).toHaveBeenCalledTimes(2); + expect(physicalClose).toHaveBeenCalledOnce(); + }); + it("registers one command pair and dispatches both through one execution", async () => { const executionId = "123e4567-e89b-42d3-a456-426614174000"; const commands: OpenClawPluginNodeHostCommand[] = []; diff --git a/src/plugins/computer-use-contract.ts b/src/plugins/computer-use-contract.ts index 2da9014d6ff0..58c7f72cc5df 100644 --- a/src/plugins/computer-use-contract.ts +++ b/src/plugins/computer-use-contract.ts @@ -569,7 +569,8 @@ export function registerComputerUseProvider( provider: ComputerUseProvider, ): void { let execution: { id: string; promise: Promise } | undefined; - let closingPromise: Promise = Promise.resolve(); + let closingPromise: Promise | undefined; + let pendingClose: Promise | undefined; const executionEnvelopeFromParams = (paramsJSON: string | null | undefined) => { let value: unknown; @@ -601,7 +602,10 @@ export function registerComputerUseProvider( if (!executionId) { throw new Error("COMPUTER_INVALID_REQUEST: executionId is required"); } - await closingPromise; + // An earlier queued close can replace the barrier while this acquisition resumes. + for (let barrier = closingPromise; barrier !== undefined; barrier = closingPromise) { + await barrier; + } if (execution && execution.id !== executionId) { throw new Error("COMPUTER_HOST_BUSY: another provider execution owns this computer"); } @@ -620,18 +624,61 @@ export function registerComputerUseProvider( } return execution.promise; }; - const closeExecution = async (executionId: string | undefined, reason: string) => { - await closingPromise; + const closeCurrentExecution = ( + executionId: string | undefined, + reason: string, + ): Promise => { const current = execution; if (!current || (executionId !== undefined && current.id !== executionId)) { - return; + return Promise.resolve(); } - execution = undefined; - if (current) { - const close = current.promise.then(async (opened) => await opened.close(reason)); - closingPromise = close.catch(() => {}); - await close; + if (pendingClose) { + return pendingClose; } + // Watcher stop and disconnect must join the same physical close before either yields. + const close = current.promise.then(async (opened) => await opened.close(reason)); + pendingClose = close; + closingPromise = close; + void close.then( + () => { + if (execution === current) { + execution = undefined; + } + pendingClose = undefined; + closingPromise = undefined; + }, + () => { + pendingClose = undefined; + // Failed open owns nothing; failed physical close stays owned for an explicit close. + if (execution !== current) { + closingPromise = undefined; + } + }, + ); + return close; + }; + const closeExecution = (executionId: string | undefined, reason: string): Promise => { + if (!pendingClose) { + return closeCurrentExecution(executionId, reason); + } + const joined = (async () => { + // Earlier queued operations can publish another close after each barrier settles. + for (let barrier = pendingClose; barrier !== undefined; barrier = pendingClose) { + const matchesClosingOwner = executionId === undefined || execution?.id === executionId; + try { + await barrier; + } catch (error) { + if (matchesClosingOwner) { + throw error; + } + return; + } + } + await closeCurrentExecution(executionId, reason); + })(); + // Watcher cleanup may initiate an unawaited close; joiners still receive the actual failure. + void joined.catch(() => {}); + return joined; }; api.registerNodeHostCommand({