From 9d31396f7a6f67de3d8d79fa0e072ed335c0dcf9 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 13 Aug 2026 17:44:02 -0700 Subject: [PATCH] FN-9031: strengthen computer snapshot action fencing Serialize computer snapshots and actions through a durable app-scoped fence. - Hold the fence from snapshot acquisition or element resolution through OS action consumption. - Report snapshot consumption only when the expected latest pointer was invalidated. - Heartbeat and token-check directory locks for safe crash recovery and add command-fence coverage. Files changed: .../commands/__tests__/computer-commands.test.ts | 26 +++++++++- .../__tests__/computer-snapshot-index.test.ts | 2 +- packages/cli/src/commands/computer.ts | 37 ++++++++----- .../cli/src/commands/computer/snapshot-store.ts | 60 ++++++++++++++++------ 4 files changed, 93 insertions(+), 32 deletions(-) Fusion-Task-Id: FN-9031 Fusion-Task-Lineage: f196b7ca-ce3f-4a50-98d2-d6ac8bf184ac Co-authored-by: Fusion (runfusion.ai) --- .../__tests__/computer-commands.test.ts | 26 +++++++- .../__tests__/computer-snapshot-index.test.ts | 2 +- packages/cli/src/commands/computer.ts | 37 ++++++++---- .../src/commands/computer/snapshot-store.ts | 60 +++++++++++++------ 4 files changed, 93 insertions(+), 32 deletions(-) diff --git a/packages/cli/src/commands/__tests__/computer-commands.test.ts b/packages/cli/src/commands/__tests__/computer-commands.test.ts index a259c2ab13..27a973d639 100644 --- a/packages/cli/src/commands/__tests__/computer-commands.test.ts +++ b/packages/cli/src/commands/__tests__/computer-commands.test.ts @@ -145,7 +145,7 @@ describe("computer commands", () => { resolveLocator: async (_window, locator) => ({ element: locator.path === from.locator.path ? from : to, handle: locator.path }), drag: async (value) => { input = value; return { action: "drag", app: value.app, snapshotId: value.snapshotId, elementIndex: null, fromElementIndex: value.from?.element.index ?? null, toElementIndex: value.to?.element.index ?? null, performed: true, snapshotConsumed: false }; }, }; - const store = { resolve, consume: vi.fn(), getElement: (_record: typeof record, index: number) => _record.elements[String(index)] } as unknown as import("../computer/snapshot-store.js").ComputerSnapshotStore; + const store = { resolve, consume: vi.fn(async () => true), withAppFence: async (_app: typeof app, operation: () => Promise) => operation(), getElement: (_record: typeof record, index: number) => _record.elements[String(index)] } as unknown as import("../computer/snapshot-store.js").ComputerSnapshotStore; expect(await runComputer(["drag", "--app", "App", "--from-element-index", "7", "--to-element-index", "9", "--json"], { adapter: dragAdapter, store, stdout: () => undefined })).toBe(0); expect(resolve).toHaveBeenCalledTimes(1); expect(input).toMatchObject({ snapshotId: record.snapshotId, from: { element: { index: 7 } }, to: { element: { index: 9 } } }); @@ -222,6 +222,30 @@ describe("computer commands", () => { } finally { await (await import("node:fs/promises")).rm(root, { recursive: true, force: true }); } }); + it("holds the command fence through consumption and makes a re-capture wait", async () => { + const root = await mkdtemp(join(tmpdir(), "fusion-computer-command-fence-")); + const clock = { now: () => new Date(0) }; + let releaseAction!: () => void; + const actionStarted = new Promise((resolve) => { releaseAction = resolve; }); + let completeAction!: () => void; + const actionComplete = new Promise((resolve) => { completeAction = resolve; }); + const events: string[] = []; + let captures = 0; + const blockingAdapter: ComputerAdapter = { ...adapter, captureState: async (target, options) => { captures += 1; if (captures > 1) events.push("capture"); return adapter.captureState(target, options); }, click: async (input) => { releaseAction(); await actionComplete; events.push("action-consumed"); return { action: "click", app: input.app, snapshotId: input.snapshotId, elementIndex: 7, fromElementIndex: null, toElementIndex: null, performed: true, snapshotConsumed: false }; } }; + try { + await runComputer(["get-app-state", "--app", "App", "--no-screenshot", "--json"], { adapter: blockingAdapter, projectRoot: root, clock, stdout: () => undefined }); + const action = runComputer(["click", "--app", "App", "--element-index", "7", "--json"], { adapter: blockingAdapter, projectRoot: root, clock, stdout: () => undefined }); + await actionStarted; + let recaptured = false; + const recapture = runComputer(["get-app-state", "--app", "App", "--no-screenshot", "--json"], { adapter: blockingAdapter, projectRoot: root, clock, stdout: () => { recaptured = true; } }); + completeAction(); + expect(await action).toBe(0); + expect(await recapture).toBe(0); + expect(recaptured).toBe(true); + expect(events).toEqual(["action-consumed", "capture"]); + } finally { await rm(root, { recursive: true, force: true }); } + }); + it("uses group-level INVALID_ARGUMENTS for unknown commands", async () => { const output: string[] = []; expect(await runComputer(["nope", "--json"], { adapter, stdout: (x) => output.push(x) })).toBe(1); expect(JSON.parse(output[0])).toMatchObject({ command: "computer", error: { code: "INVALID_ARGUMENTS" } }); }); it("returns the required JSON envelope for a missing subcommand", async () => { const output: string[] = []; diff --git a/packages/cli/src/commands/__tests__/computer-snapshot-index.test.ts b/packages/cli/src/commands/__tests__/computer-snapshot-index.test.ts index c0205ddba9..bf8d1453f3 100644 --- a/packages/cli/src/commands/__tests__/computer-snapshot-index.test.ts +++ b/packages/cli/src/commands/__tests__/computer-snapshot-index.test.ts @@ -86,7 +86,7 @@ describe("ComputerSnapshotStore", () => { it("consumes the latest pointer until a fresh capture re-arms it", async () => { const { store } = await fixture(); - await expect(store.consume(app)).resolves.toBeUndefined(); + await expect(store.consume(app)).resolves.toBe(false); const first = await store.persist({ app, window, elementCount: 1, elements: [element(7)] }); await store.consume(app); diff --git a/packages/cli/src/commands/computer.ts b/packages/cli/src/commands/computer.ts index a311366181..f2e93720e4 100644 --- a/packages/cli/src/commands/computer.ts +++ b/packages/cli/src/commands/computer.ts @@ -61,19 +61,27 @@ async function optionalElement(args: string[], adapter: ComputerAdapter, store: * after the first action changed focus, navigation, scrolling, or the rendered accessibility tree. */ async function finishAction(result: ActionResult, store: ComputerSnapshotStore, app: AppRef): Promise { - await store.consume(app, result.snapshotId ?? undefined); - return { ...result, snapshotConsumed: true }; + const snapshotConsumed = await store.consume(app, result.snapshotId ?? undefined, true); + return { ...result, snapshotConsumed }; +} + +/** + * FNXC:ComputerUse 2026-08-14-00:35: + * Resolution, locator replay, OS action, and consumption must remain under one app-scoped fence. + */ +async function fencedAction(store: ComputerSnapshotStore, app: AppRef, operation: () => Promise): Promise { + return store.withAppFence(app, async () => finishAction(await operation(), store, app)); } export const COMPUTER_HANDLERS: Record = { capabilities: async (_args, o) => adapterFor(o).capabilities(), permissions: async (_args, o) => adapterFor(o).permissions(), "list-apps": async (_args, o) => adapterFor(o).listApps(), "list-windows": async (args, o) => { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); const adapter = adapterFor(o); return adapter.listWindows(parseAppTarget(raw)); }, - "get-app-state": async (args, o) => { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); if (value(args, "--window-id") && value(args, "--window-index")) throw new ComputerUseError("INVALID_ARGUMENTS", "Window flags are mutually exclusive."); const adapter = adapterFor(o); const state = await adapter.captureState(parseAppTarget(raw), { windowId: value(args, "--window-id"), windowIndex: number(args, "--window-index"), screenshot: !args.includes("--no-screenshot"), restoreWindow: args.includes("--restore-window") }); const record = await storeFor(o).persist({ app: state.app, window: state.window, elementCount: state.snapshot.elementCount, elements: state.snapshot.elements, capturedAt: state.snapshot.capturedAt }); state.snapshot.snapshotId = record.snapshotId; state.snapshot.targetKey = record.targetKey; state.snapshot.windowKey = record.windowKey; state.snapshot.capturedAt = record.capturedAt; state.snapshot.expiresAt = record.expiresAt; return state; }, - click: async (args, o) => { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o), item = await requireElement(args, adapter, store, app); return finishAction(await adapter.click({ app, window: item.window, element: item.element, snapshotId: item.record.snapshotId }), store, app); }, - "set-value": async (args, o) => { const raw = value(args, "--app"), text = value(args, "--value"); if (!raw || (!text && !args.includes("--value-stdin")) || (text && args.includes("--value-stdin"))) throw new ComputerUseError("INVALID_ARGUMENTS", "--app and exactly one value source are required."); const secret = args.includes("--value-stdin") ? await (o.stdin ?? (async () => ""))() : text!; const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o), item = await requireElement(args, adapter, store, app); return finishAction(await adapter["set-value"]({ app, window: item.window, element: item.element, snapshotId: item.record.snapshotId, value: secret }), store, app); }, + "get-app-state": async (args, o) => { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); if (value(args, "--window-id") && value(args, "--window-index")) throw new ComputerUseError("INVALID_ARGUMENTS", "Window flags are mutually exclusive."); const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o); return store.withAppFence(app, async () => { const state = await adapter.captureState(parseAppTarget(raw), { windowId: value(args, "--window-id"), windowIndex: number(args, "--window-index"), screenshot: !args.includes("--no-screenshot"), restoreWindow: args.includes("--restore-window") }); const record = await store.persist({ app: state.app, window: state.window, elementCount: state.snapshot.elementCount, elements: state.snapshot.elements, capturedAt: state.snapshot.capturedAt }, true); state.snapshot.snapshotId = record.snapshotId; state.snapshot.targetKey = record.targetKey; state.snapshot.windowKey = record.windowKey; state.snapshot.capturedAt = record.capturedAt; state.snapshot.expiresAt = record.expiresAt; return state; }); }, + click: async (args, o) => { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o); return fencedAction(store, app, async () => { const item = await requireElement(args, adapter, store, app); return adapter.click({ app, window: item.window, element: item.element, snapshotId: item.record.snapshotId }); }); }, + "set-value": async (args, o) => { const raw = value(args, "--app"), text = value(args, "--value"); if (!raw || (!text && !args.includes("--value-stdin")) || (text && args.includes("--value-stdin"))) throw new ComputerUseError("INVALID_ARGUMENTS", "--app and exactly one value source are required."); const secret = args.includes("--value-stdin") ? await (o.stdin ?? (async () => ""))() : text!; const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o); return fencedAction(store, app, async () => { const item = await requireElement(args, adapter, store, app); return adapter["set-value"]({ app, window: item.window, element: item.element, snapshotId: item.record.snapshotId, value: secret }); }); }, "type-text": async (args, o) => targetedOrUntargeted("type-text", args, o), "press-key": async (args, o) => targetedOrUntargeted("press-key", args, o), scroll: async (args, o) => targetedOrUntargeted("scroll", args, o), - hotkey: async (args, o) => { const raw = value(args, "--app"), keys = value(args, "--keys"); if (!raw || !keys) throw new ComputerUseError("INVALID_ARGUMENTS", "--app and --keys are required."); const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o); return finishAction(await adapter.hotkey({ app, keys: keys.split("+") }), store, app); }, + hotkey: async (args, o) => { const raw = value(args, "--app"), keys = value(args, "--keys"); if (!raw || !keys) throw new ComputerUseError("INVALID_ARGUMENTS", "--app and --keys are required."); const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o); return fencedAction(store, app, () => adapter.hotkey({ app, keys: keys.split("+") })); }, drag: async (args, o) => { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); const coordinateFlags = ["--from-x", "--from-y", "--to-x", "--to-y"]; @@ -85,13 +93,13 @@ export const COMPUTER_HANDLERS: Record = { if (hasCoordinates) { const coordinates = coordinateFlags.map((flag) => number(args, flag)); if (coordinates.some((item) => item === undefined) || value(args, "--snapshot-id") || value(args, "--window-id") || value(args, "--window-index")) throw new ComputerUseError("INVALID_ARGUMENTS", "Coordinate drag requires all coordinates and takes no snapshot or window flags."); - return finishAction(await adapter.drag({ app, snapshotId: null, fromX: coordinates[0]!, fromY: coordinates[1]!, toX: coordinates[2]!, toY: coordinates[3]! }), storeFor(o), app); + const store = storeFor(o); + return fencedAction(store, app, () => adapter.drag({ app, snapshotId: null, fromX: coordinates[0]!, fromY: coordinates[1]!, toX: coordinates[2]!, toY: coordinates[3]! })); } if (from === undefined || to === undefined) throw new ComputerUseError("INVALID_ARGUMENTS", "Drag requires either all coordinates or both element indexes."); // Both endpoints share one resolved record and one replayed window, even if another capture updates latest mid-action. const store = storeFor(o); - const resolved = await requireElements(args, [from, to], adapter, store, app); - return finishAction(await adapter.drag({ app, snapshotId: resolved.record.snapshotId, window: resolved.window, from: resolved.elements[0]!, to: resolved.elements[1]! }), store, app); + return fencedAction(store, app, async () => { const resolved = await requireElements(args, [from, to], adapter, store, app); return adapter.drag({ app, snapshotId: resolved.record.snapshotId, window: resolved.window, from: resolved.elements[0]!, to: resolved.elements[1]! }); }); }, }; function validateFlags(name: ComputerSubcommand, args: string[]): void { @@ -159,17 +167,20 @@ function validateFlags(name: ComputerSubcommand, args: string[]): void { async function targetedOrUntargeted(kind: "type-text" | "press-key" | "scroll", args: string[], o: ComputerCommandOptions): Promise { const raw = value(args, "--app"); if (!raw) throw new ComputerUseError("INVALID_ARGUMENTS", "--app is required."); - const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o), item = await optionalElement(args, adapter, store, app); + const adapter = adapterFor(o), app = await appFor(adapter, raw), store = storeFor(o); + return fencedAction(store, app, async () => { + const item = await optionalElement(args, adapter, store, app); if (!item && (value(args, "--snapshot-id") || value(args, "--window-id") || value(args, "--window-index"))) throw new ComputerUseError("INVALID_ARGUMENTS", "Snapshot and window flags require --element-index."); if (kind === "type-text") { const direct = value(args, "--text"), fromStdin = args.includes("--text-stdin"); if ((direct === undefined && !fromStdin) || (direct !== undefined && fromStdin)) throw new ComputerUseError("INVALID_ARGUMENTS", "Exactly one text source is required."); const text = fromStdin ? await (o.stdin ?? (async () => ""))() : direct!; - return finishAction(await adapter["type-text"]({ app, text, ...(item ? { window: item.window, element: item.element, snapshotId: item.record.snapshotId } : {}) }), store, app); + return adapter["type-text"]({ app, text, ...(item ? { window: item.window, element: item.element, snapshotId: item.record.snapshotId } : {}) }); } - if (kind === "press-key") { const key = value(args, "--key"); if (!key) throw new ComputerUseError("INVALID_ARGUMENTS", "--key is required."); return finishAction(await adapter["press-key"]({ app, key, ...(item ? { window: item.window, element: item.element, snapshotId: item.record.snapshotId } : {}) }), store, app); } + if (kind === "press-key") { const key = value(args, "--key"); if (!key) throw new ComputerUseError("INVALID_ARGUMENTS", "--key is required."); return adapter["press-key"]({ app, key, ...(item ? { window: item.window, element: item.element, snapshotId: item.record.snapshotId } : {}) }); } const direction = value(args, "--direction"); if (!direction || !["up", "down", "left", "right"].includes(direction)) throw new ComputerUseError("INVALID_ARGUMENTS", "A valid --direction is required."); - return finishAction(await adapter.scroll({ app, direction: direction as "up" | "down" | "left" | "right", amount: number(args, "--amount") ?? 3, ...(item ? { window: item.window, element: item.element, snapshotId: item.record.snapshotId } : {}) }), store, app); + return adapter.scroll({ app, direction: direction as "up" | "down" | "left" | "right", amount: number(args, "--amount") ?? 3, ...(item ? { window: item.window, element: item.element, snapshotId: item.record.snapshotId } : {}) }); + }); } export async function runComputer(args: string[], options: ComputerCommandOptions = {}): Promise { const json = args.includes("--json"); diff --git a/packages/cli/src/commands/computer/snapshot-store.ts b/packages/cli/src/commands/computer/snapshot-store.ts index 01d1306d2b..6754c6f325 100644 --- a/packages/cli/src/commands/computer/snapshot-store.ts +++ b/packages/cli/src/commands/computer/snapshot-store.ts @@ -1,4 +1,4 @@ -import { mkdir, readFile, readdir, rename, rm, stat, writeFile } from "node:fs/promises"; +import { mkdir, readFile, readdir, rename, rm, stat, utimes, writeFile } from "node:fs/promises"; import { join, resolve } from "node:path"; import { randomBytes } from "node:crypto"; import { @@ -70,7 +70,7 @@ export class ComputerSnapshotStore { return join(this.projectRoot, ".fusion", "computer-use", "latest"); } - async persist(input: PersistSnapshotInput): Promise { + async persist(input: PersistSnapshotInput, alreadyFenced = false): Promise { const capturedAt = input.capturedAt ?? this.now().toISOString(); const expiresAt = input.expiresAt ?? new Date(Date.parse(capturedAt) + this.ttlMs).toISOString(); const targetKey = targetKeyForApp(input.app); @@ -89,9 +89,9 @@ export class ComputerSnapshotStore { await mkdir(this.snapshotsDirectory, { recursive: true }); await mkdir(this.latestDirectory, { recursive: true }); await writeJsonAtomically(this.snapshotPath(record.snapshotId), record); - await this.mutateLatestPointer(targetKey, async () => { - await writeJsonAtomically(this.latestPath(targetKey), { snapshotId: record.snapshotId }); - }); + const writeLatest = async () => { await writeJsonAtomically(this.latestPath(targetKey), { snapshotId: record.snapshotId }); }; + if (alreadyFenced) await writeLatest(); + else await this.mutateLatestPointer(targetKey, writeLatest); await this.prune({ preserveSnapshotId: record.snapshotId }); return record; } @@ -101,13 +101,28 @@ export class ComputerSnapshotStore { * A successful action consumes only this app's latest pointer, retaining the record for clear * stale details and pruning. The next element replay must capture a new accessibility tree. */ - async consume(app: AppRef, expectedSnapshotId?: string): Promise { + async consume(app: AppRef, expectedSnapshotId?: string, alreadyFenced = false): Promise { const targetKey = targetKeyForApp(app); - await this.mutateLatestPointer(targetKey, async () => { + let consumed = false; + const consumeLatest = async () => { const latest = await this.readLatest(targetKey); if (!latest || (expectedSnapshotId !== undefined && latest.snapshotId !== expectedSnapshotId)) return; await writeJsonAtomically(this.latestPath(targetKey), { snapshotId: latest.snapshotId, consumedAt: this.now().toISOString() }); - }); + consumed = true; + }; + if (alreadyFenced) await consumeLatest(); + else await this.mutateLatestPointer(targetKey, consumeLatest); + return consumed; + } + + /** + * FNXC:ComputerUse 2026-08-14-00:35: + * Capture acquisition and element replay share this app fence, not merely their pointer writes. + * This prevents an action resolved from S or a tree captured before it from racing to re-arm S. + */ + async withAppFence(app: AppRef, operation: () => Promise): Promise { + await mkdir(this.latestDirectory, { recursive: true }); + return withDirectoryLock(`${this.latestPath(targetKeyForApp(app))}.lock`, operation); } /** Read the current record for the resolved app and enforce the C9 fence order. */ @@ -198,9 +213,7 @@ export class ComputerSnapshotStore { * while holding the same per-app lock that persist uses to re-arm the pointer. */ private async mutateLatestPointer(targetKey: string, mutation: () => Promise): Promise { - await mkdir(this.latestDirectory, { recursive: true }); - const lockPath = `${this.latestPath(targetKey)}.lock`; - await withDirectoryLock(lockPath, mutation); + await this.withAppFence({ bundleId: targetKey.startsWith("bundle:") ? targetKey.slice("bundle:".length) : null, name: "snapshot-pointer", pid: targetKey.startsWith("pid:") ? Number(targetKey.slice("pid:".length)) : 0 }, mutation); } private snapshotPath(snapshotId: string): string { @@ -278,28 +291,41 @@ async function readDirectory(path: string): Promise { const POINTER_LOCK_RETRY_MS = 5; const POINTER_LOCK_STALE_MS = 60_000; +const LOCK_HEARTBEAT_MS = 1_000; -/** Serialize a tiny pointer rewrite across independently-invoked CLI processes. */ -async function withDirectoryLock(lockPath: string, operation: () => Promise): Promise { +/** + * FNXC:ComputerUse 2026-08-14-00:35: + * An app fence lives through OS work, so its heartbeat—not a fixed maximum action duration—proves + * liveness. Stale locks are atomically quarantined and token-checked on release: a crashed holder + * cannot block forever, and a recovered holder cannot delete a successor's lock after a PID reuse. + */ +async function withDirectoryLock(lockPath: string, operation: () => Promise): Promise { + const ownerPath = join(lockPath, "owner.json"); + const token = randomBytes(16).toString("hex"); for (;;) { try { await mkdir(lockPath); + await writeJsonAtomically(ownerPath, { token, pid: process.pid }); break; } catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; const lockInfo = await stat(lockPath).catch(() => undefined); const age = lockInfo === undefined ? Number.NaN : Date.now() - lockInfo.mtimeMs; if (Number.isFinite(age) && age > POINTER_LOCK_STALE_MS) { - await rm(lockPath, { recursive: true, force: true }); - continue; + const quarantined = `${lockPath}.stale-${randomBytes(8).toString("hex")}`; + try { await rename(lockPath, quarantined); await rm(quarantined, { recursive: true, force: true }); continue; } + catch { continue; } } await new Promise((resolveRetry) => setTimeout(resolveRetry, POINTER_LOCK_RETRY_MS)); } } + const heartbeat = setInterval(() => { void utimes(lockPath, new Date(), new Date()).catch(() => undefined); }, LOCK_HEARTBEAT_MS); try { - await operation(); + return await operation(); } finally { - await rm(lockPath, { recursive: true, force: true }); + clearInterval(heartbeat); + const owner = await readJson(ownerPath) as { token?: unknown } | undefined; + if (owner?.token === token) await rm(lockPath, { recursive: true, force: true }); } }