diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d41ae8d47..c718f610b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -333,6 +333,12 @@ jobs: env: BSK_COOP_CHROME: ${{ env.BSK_VIDEO_CHROME }} run: pnpm exec vitest run src/content/__tests__/cooperation-capture.browser.test.ts + - name: Validate thumbnail cancellation, synchronization and navigating clicks + working-directory: packages/dsh-plugin-browserskill + env: + BSK_COOP_CHROME: ${{ env.BSK_VIDEO_CHROME }} + BSK_COOP_BSK: ../../target/debug/bsk + run: pnpm exec vitest run tests/cooperation.browser.test.ts - name: Validate task termination on tab detachment working-directory: apps/extension env: diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b45b5eb4..10e930fb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,20 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/). Starting from 0.2.0, CLI / Extension / DSH Plugin share the same version number. +## [Unreleased] + +### Fixed + +- Shared-control synchronization waits for cancelled observation thumbnails to + settle. Viewport capture reads have a 10-second deadline with separate bounded + cleanup; reaching that deadline returns `timeout` (exit 4) with screenshot-specific + guidance. Browser/CDP capture failures remain `cdp_failed` (exit 3). +- Completed inputs retain their successful receipts when they navigate, including + clicks, Enter submissions, drag submissions, and forms submitted after filling + or selecting a value. A new synchronization is still required before the next action. +- CLI cancellation is armed before sending a browser command, so an immediate + Ctrl-C can cancel the operation and return its outcome instead of terminating silently. + ## [0.4.0] - 2026-10-30 ### Changed diff --git a/apps/extension/src/cooperation/controller.test.ts b/apps/extension/src/cooperation/controller.test.ts index c26bace8b..dc4eb65bf 100644 --- a/apps/extension/src/cooperation/controller.test.ts +++ b/apps/extension/src/cooperation/controller.test.ts @@ -1129,6 +1129,138 @@ describe("shared-tab dispatch fences", () => { const fresh = await f.command("sync"); expect(await f.act(fresh.context_id)).toEqual({}); }); + it.each([ + "before-release", + "after-release", + ])("preserves a completed click when navigation invalidates its document %s", async (navigation) => { + const f = await fixture(); + const receipt = { tab_id: 7, used_ref: "e1", x: 4, y: 5 }; + const result = await f.act(f.first.context_id, async (runner) => { + await runner.send(7, "Input.dispatchMouseEvent", { + type: "mousePressed", + button: "left", + x: 4, + y: 5, + }); + if (navigation === "before-release") f.controller.invalidate(7); + await runner.send(7, "Input.dispatchMouseEvent", { + type: "mouseReleased", + button: "left", + x: 4, + y: 5, + }); + if (navigation === "after-release") f.controller.invalidate(7); + return receipt; + }); + expect(result).toEqual(receipt); + expect(f.native).toHaveBeenCalledTimes(2); + expect(await f.act()).toMatchObject({ code: "permission_denied" }); + const fresh = await f.command("sync"); + expect(await f.act(fresh.context_id)).toEqual({}); + }); + it.each([ + "human", + "hold", + "stop", + "cancel", + "failed-release", + "more-input", + ])("does not hide %s after a navigating click", async (intervention) => { + const f = await fixture(); + const result = await f.act(f.first.context_id, async (runner) => { + await runner.send(7, "Input.dispatchMouseEvent", { + type: "mousePressed", + button: "left", + x: 4, + y: 5, + }); + f.controller.invalidate(7); + if (intervention === "human") f.change({ human: 1, lastInput: 10000 }); + if (intervention === "hold") + await f.controller.control(7, f.controller.view(7)!.revision, "hold"); + if (intervention === "stop") f.controller.stopSession("shared"); + if (intervention === "cancel") f.abort.abort(); + if (intervention === "failed-release") + return { code: "cdp_failed", message: "release failed" }; + await runner.send(7, "Input.dispatchMouseEvent", { + type: "mouseReleased", + button: "left", + x: 4, + y: 5, + }); + if (intervention === "more-input") + await runner.send(7, "Input.insertText", { text: "stale" }); + return { tab_id: 7, x: 4, y: 5 }; + }); + expect(result).toMatchObject({ + code: + intervention === "cancel" + ? "cancelled" + : intervention === "stop" + ? "user_aborted" + : "permission_denied", + data: { effect_state: "unknown" }, + }); + expect(f.native).toHaveBeenCalledTimes(intervention === "failed-release" ? 1 : 2); + }); + it("preserves a completed Enter submission after its document is replaced", async () => { + const f = await fixture(); + const receipt = { tab_id: 7, key: "Enter", code: "Enter", modifiers: [] }; + const result = await f.act( + f.first.context_id, + async (runner) => { + await runner.send(7, "Input.dispatchKeyEvent", { type: "rawKeyDown", key: "Enter" }); + await runner.send(7, "Input.dispatchKeyEvent", { type: "char", key: "Enter", text: "\r" }); + f.controller.invalidate(7); + await runner.send(7, "Input.dispatchKeyEvent", { type: "keyUp", key: "Enter" }); + return receipt; + }, + "tool.press", + ); + expect(result).toEqual(receipt); + expect(await f.act()).toMatchObject({ code: "permission_denied" }); + const fresh = await f.command("sync"); + expect(await f.act(fresh.context_id)).toEqual({}); + }); + it.each([ + "tool.drag", + "tool.select", + ])("preserves completed %s receipts after navigation", async (method) => { + const f = await fixture(); + const receipt = { tab_id: 7 }; + const result = await f.act( + f.first.context_id, + async (runner) => { + if (method === "tool.drag") { + await runner.send(7, "Input.dispatchMouseEvent", { + type: "mousePressed", + button: "left", + x: 4, + y: 5, + }); + await runner.send(7, "Input.dispatchMouseEvent", { + type: "mouseReleased", + button: "left", + x: 10, + y: 5, + }); + } else { + await runner.beginDomMutation?.(); + await runner.send(7, "Runtime.callFunctionOn", { + objectId: "select", + functionDeclaration: "function() { this.dispatchEvent(new Event('change')); }", + }); + } + f.controller.invalidate(7); + return receipt; + }, + method, + ); + expect(result).toEqual(receipt); + expect(await f.act()).toMatchObject({ code: "permission_denied" }); + const fresh = await f.command("sync"); + expect(await f.act(fresh.context_id)).toEqual({}); + }); it("still blocks input after a navigation invalidates the action's document", async () => { const f = await fixture(); const result = await f.act( diff --git a/apps/extension/src/cooperation/controller.ts b/apps/extension/src/cooperation/controller.ts index 0323c4af7..d29edd932 100644 --- a/apps/extension/src/cooperation/controller.ts +++ b/apps/extension/src/cooperation/controller.ts @@ -1321,15 +1321,16 @@ export class CooperationController implements ControlAuthority { data: { effect_state: sent ? "unknown" : "none" }, }; if (["tool.evaluate", "tool.upload"].includes(req.method)) await this.refresh(state); - // Navigation replaces its observed document. Return the handler's completed - // receipt without restoring that context or allowing any further input. + // A completed navigation or input can replace the observed + // document. Preserve the handler's receipt without authorizing further input. // Human intervention, cancellation and failed dispatch still take priority. const settledNavigation = sent && !isRpcError(result) && - ["tool.navigate", "tool.navigate_back", "tool.navigate_forward", "tool.reload"].includes( + (["tool.navigate", "tool.navigate_back", "tool.navigate_forward", "tool.reload"].includes( req.method, - ); + ) || + nativeInputStarted); const reason = revoked ?? interruption(settledNavigation); if (reason) return failure(reason, sent ? "unknown" : "none"); return result; diff --git a/apps/extension/src/tools/__tests__/viewport-screenshot.test.ts b/apps/extension/src/tools/__tests__/viewport-screenshot.test.ts index a4fa30da2..f5cd8225c 100644 --- a/apps/extension/src/tools/__tests__/viewport-screenshot.test.ts +++ b/apps/extension/src/tools/__tests__/viewport-screenshot.test.ts @@ -93,6 +93,84 @@ describe("controlled viewport capture", () => { expect(f.bridge.mock.calls.at(-1)?.[1].phase).toBe("end"); }); + it.each([ + "Page.getFrameTree", + "Page.createIsolatedWorld", + "Runtime.evaluate", + "Page.captureScreenshot", + ])("cancels a pending %s without waiting for Chrome's reply", async (method) => { + const f = fixture(); + let release!: () => void; + const pending = new Promise((resolve) => { + release = resolve; + }); + f.hook.command = async (name) => { + if (name === method) await pending; + }; + const abort = new AbortController(); + const result = f.run(abort.signal); + await vi.waitFor(() => + expect(f.send.mock.calls.some(([, name]) => name === method)).toBe(true), + ); + abort.abort(); + try { + await expect(result).resolves.toMatchObject({ code: "cancelled" }); + const commands = f.send.mock.calls.length; + release(); + await Promise.resolve(); + expect(f.send.mock.calls).toHaveLength(commands); + if (method === "Page.captureScreenshot") + expect(f.bridge.mock.calls.at(-1)?.[1].phase).toBe("end"); + f.hook.command = async () => {}; + expect(await f.run()).toMatchObject({ image_base64: PNG }); + } finally { + release(); + } + }); + it.each([ + "screenshot", + "before-capture identity", + "after-capture identity", + ])("times out hung %s and bounds overlay restoration", async (stage) => { + vi.useFakeTimers(); + const f = fixture(); + let identities = 0; + f.hook.command = async (method) => { + if (method === "Runtime.evaluate") identities++; + if ( + (stage === "screenshot" && method === "Page.captureScreenshot") || + (method === "Runtime.evaluate" && + identities === + (stage === "before-capture identity" ? 2 : stage === "after-capture identity" ? 3 : 0)) + ) + await new Promise(() => {}); + }; + f.hook.bridge = async (phase) => { + if (phase === "end") await new Promise(() => {}); + }; + try { + const result = f.run(); + const assertion = expect(result).resolves.toMatchObject({ code: "timeout" }); + await vi.advanceTimersByTimeAsync(11_000); + await assertion; + expect(f.bridge.mock.calls.at(-1)?.[1].phase).toBe("end"); + } finally { + vi.useRealTimers(); + } + }); + it("cancels a pending overlay begin and restores it without capturing", async () => { + const f = fixture(); + f.hook.bridge = async (phase) => { + if (phase === "begin") await new Promise(() => {}); + }; + const abort = new AbortController(); + const result = f.run(abort.signal); + await vi.waitFor(() => expect(f.bridge).toHaveBeenCalled()); + abort.abort(); + await expect(result).resolves.toMatchObject({ code: "cancelled" }); + expect(f.bridge.mock.calls.map(([, message]) => message.phase)).toEqual(["begin", "end"]); + expect(f.send.mock.calls.some(([, method]) => method === "Page.captureScreenshot")).toBe(false); + }); it("reports capture failure without retrying another target or backend", async () => { const f = fixture(); f.hook.command = async (method) => { diff --git a/apps/extension/src/tools/viewport-screenshot.ts b/apps/extension/src/tools/viewport-screenshot.ts index 4ff73e55c..9234bad35 100644 --- a/apps/extension/src/tools/viewport-screenshot.ts +++ b/apps/extension/src/tools/viewport-screenshot.ts @@ -6,6 +6,7 @@ import type { RpcError } from "@/transport/types"; import { rpcError } from "./errors"; import { parsePngDimensions } from "./png"; import type { CdpRunner } from "./shared"; +import { BoundedWaitError, waitBounded } from "./transfer-transaction"; import { isAbortError, throwIfAborted } from "./vom/capture-abort"; import { verifyDocumentIdentity } from "./vom/document-identity"; import type { DocumentIdentity } from "./vom/facts"; @@ -88,19 +89,59 @@ export async function captureControlledViewport( signal?: AbortSignal, sendToTab?: CaptureSuppressSendToTab, ): Promise<{ image_base64: string; width: number; height: number } | RpcError> { - try { + const deadline = Date.now() + 10_000; + const wait = (work: Promise, cleanup = false) => + waitBounded( + work, + cleanup ? Date.now() + 500 : deadline, + cleanup ? undefined : signal, + "Viewport capture timed out", + ); + const check = () => { throwIfAborted(signal); - if (!(await stillControlled())) return changed(); - const identity = await readDocumentIdentity(cdp, tabId, signal); + if (Date.now() >= deadline) throw new BoundedWaitError("timeout", "Viewport capture timed out"); + }; + // Chrome cannot cancel dispatched reads. Retire their waiters and consume late + // replies; never continue the pipeline after its cancellation or deadline. + const captureCdp: CdpRunner = { + getAttachmentId: (id) => cdp.getAttachmentId?.(id), + send: (id, method, params) => { + const cleanup = method === "Runtime.releaseObjectGroup"; + if (!cleanup) check(); + return wait(cdp.send(id, method, params), cleanup); + }, + }; + const controlled = () => { + check(); + return wait(stillControlled()); + }; + const send = sendToTab ?? ((id, message) => chrome.tabs.sendMessage(id, message)); + const overlay: CaptureSuppressSendToTab = async (id, message) => { + if (message.phase === "end") return wait(send(id, message), true); + check(); + try { + return await wait(send(id, message)); + } catch (error) { + // The content script hides the overlay before acknowledging begin. Restore + // it even if that acknowledgement never arrives, with a separate budget. + if (error instanceof BoundedWaitError) + await wait(send(id, { ...message, phase: "end" }), true).catch(() => {}); + throw error; + } + }; + try { + check(); + if (!(await controlled())) return changed(); + const identity = await readDocumentIdentity(captureCdp, tabId, signal); if (!identity) return changed(); - const bound = boundToAttachment(cdp, tabId, identity.attachmentId); + const bound = boundToAttachment(captureCdp, tabId, identity.attachmentId); const current = async () => { - throwIfAborted(signal); - return ( - (await stillControlled()) && - (await verifyDocumentIdentity(bound, identity, signal)) === "current" && - cdp.getAttachmentId?.(tabId) === identity.attachmentId - ); + if (!(await controlled())) return false; + const verified = await verifyDocumentIdentity(bound, identity, signal); + // Identity verification treats failed reads as unavailable. Preserve this + // transaction's timeout instead of reporting a document change. + check(); + return verified === "current" && cdp.getAttachmentId?.(tabId) === identity.attachmentId; }; const shot = await withExtensionOverlayHidden( tabId, @@ -112,11 +153,11 @@ export async function captureControlledViewport( captureBeyondViewport: false, }); }, - sendToTab, + overlay, ); throwIfAborted(signal); // Validate after overlay restoration as well: navigation/return can race it. - if (!shot || !(await current()) || !(await stillControlled())) return changed(); + if (!shot || !(await current()) || !(await controlled())) return changed(); throwIfAborted(signal); if (cdp.getAttachmentId?.(tabId) !== identity.attachmentId) return changed(); const dims = shot.data ? parsePngDimensions(shot.data) : null; @@ -133,7 +174,7 @@ export async function captureControlledViewport( return { code: "cancelled", message: "screenshot aborted" }; } return rpcError( - "cdp_failed", + error instanceof BoundedWaitError && error.kind === "timeout" ? "timeout" : "cdp_failed", "screenshot_capture_failed", error instanceof Error ? error.message : String(error), ); diff --git a/crates/bsk-cli/src/cli/business_rpc.rs b/crates/bsk-cli/src/cli/business_rpc.rs index 287c048e8..3d77512bc 100644 --- a/crates/bsk-cli/src/cli/business_rpc.rs +++ b/crates/bsk-cli/src/cli/business_rpc.rs @@ -150,7 +150,7 @@ where let outcome = tokio::select! { biased; - res = &mut main_fut => res?, + // Arm cancellation before the RPC can dispatch browser input. sig = wait_for_cancel(parent_cancel) => { sig.context("wait for cancellation").map_err(CliError::Local)?; CANCEL_REQUESTED.store(true, Ordering::Relaxed); @@ -171,6 +171,7 @@ where } } } + res = &mut main_fut => res?, }; outcome.map_err(CliError::from_rpc) diff --git a/crates/bsk-cli/src/cli/render_error.rs b/crates/bsk-cli/src/cli/render_error.rs index d2672033c..18495f9e2 100644 --- a/crates/bsk-cli/src/cli/render_error.rs +++ b/crates/bsk-cli/src/cli/render_error.rs @@ -576,6 +576,13 @@ pub fn info_for_error(code: ErrorCode, data: Option<&serde_json::Value>) -> Rend ), exit_code: 4, }, + (ErrorCode::Timeout, reason::SCREENSHOT_CAPTURE_FAILED) => RenderInfo { + summary: "viewport screenshot timed out", + hint: Some( + "check that the tab is responsive before capturing again; if capture keeps timing out, reload the tab or restart the browser", + ), + exit_code: base.exit_code, + }, (ErrorCode::CdpFailed, reason::SCREENSHOT_CAPTURE_FAILED) => RenderInfo { summary: "the browser could not capture the tab image", hint: Some( @@ -855,6 +862,20 @@ mod tests { assert_eq!(info.exit_code, 3); } + #[test] + fn screenshot_capture_timeout_has_specific_guidance_and_timeout_exit_code() { + let data = serde_json::json!({ "reason": reason::SCREENSHOT_CAPTURE_FAILED }); + let info = info_for_error(ErrorCode::Timeout, Some(&data)); + assert_eq!(info.summary, "viewport screenshot timed out"); + assert_eq!( + info.hint, + Some( + "check that the tab is responsive before capturing again; if capture keeps timing out, reload the tab or restart the browser" + ) + ); + assert_eq!(info.exit_code, 4); + } + #[test] fn control_authority_denial_does_not_request_a_blocked_renderer_sync() { let data = serde_json::json!({ "reason": "control_authority_required" }); diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts index 56cfff5fe..855f06b5b 100644 --- a/packages/dsh-plugin-browserskill/src/observation.ts +++ b/packages/dsh-plugin-browserskill/src/observation.ts @@ -88,6 +88,8 @@ export class ObservationService { private thumbnailViewers = 0; private readonly captureTimers = new Map(); private readonly captureInFlight = new Set(); + /** Only running thumbnails: shared-control reads must not wait behind actions. */ + private readonly activeCaptures = new Map>(); /** Captures intentionally cancelled to make way for foreground work. */ private readonly capturePreempted = new Set(); /** Number of queued/running model-facing calls that currently outrank thumbnails. */ @@ -253,6 +255,28 @@ export class ObservationService { }; } + /** Call after acquiring foreground priority, before bypassing the action queue. + * Killing the CLI only requests cancellation; wait for its bounded cleanup. */ + async waitForCapture(sessionId: string, signal: AbortSignal): Promise { + signal.throwIfAborted(); + const capture = this.activeCaptures.get(sessionId); + if (!capture) return; + let abort: (() => void) | undefined; + try { + await Promise.race([ + capture.catch(() => {}), + new Promise((_, reject) => { + abort = () => reject(signal.reason); + signal.addEventListener("abort", abort, { once: true }); + if (signal.aborted) abort(); + }), + ]); + signal.throwIfAborted(); + } finally { + if (abort) signal.removeEventListener("abort", abort); + } + } + /** Mark an action starting on a session (tool entry instrumentation). */ beginAction(sessionId: string, action: string): void { if (!this.deps.options.enabled || this.disposed) return; @@ -328,6 +352,7 @@ export class ObservationService { for (const sessionId of [...this.captureTimers.keys()]) this.cancelCapture(sessionId); this.captureTimers.clear(); this.captureInFlight.clear(); + this.activeCaptures.clear(); this.capturePreempted.clear(); this.foregroundDepth.clear(); this.captureFailures.clear(); @@ -406,13 +431,22 @@ export class ObservationService { const outPath = join(tmpdir(), `bsk-obs-${this.scratchNamespace}-${sessionKey}.png`); let writtenPath = outPath; try { - const result = await this.deps.queue.run(sessionId, () => { + const result = await this.deps.queue.run(sessionId, async () => { // The last viewer may have left, or foreground work arrived, while queued. if (!this.canCapture(sessionId)) return Promise.resolve(undefined); - return this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], { - timeoutMs: 15_000, - tag: `observation:${sessionId}`, - }); + const capture = this.deps.runner.run( + ["screenshot", "--session", sessionId, "--out", outPath], + { + timeoutMs: 15_000, + tag: `observation:${sessionId}`, + }, + ); + this.activeCaptures.set(sessionId, capture); + try { + return await capture; + } finally { + if (this.activeCaptures.get(sessionId) === capture) this.activeCaptures.delete(sessionId); + } }); if (result === undefined) return; if (result.code !== 0) { diff --git a/packages/dsh-plugin-browserskill/src/tools.ts b/packages/dsh-plugin-browserskill/src/tools.ts index 827f91e74..4afe31ff3 100644 --- a/packages/dsh-plugin-browserskill/src/tools.ts +++ b/packages/dsh-plugin-browserskill/src/tools.ts @@ -145,6 +145,10 @@ async function runBsk( ? invoke() : runWithSessionBusyRetry(invoke, exec.signal); }; + // Shared-control reads stay outside the action queue, but cannot overtake + // the cancellation of our own thumbnail in the extension's operation lane. + if (observeSession !== undefined && args[0] === "cooperate" && args[1] !== "enable") + await deps.observation.waitForCapture(observeSession, exec.signal); result = observeSession !== undefined && (args[0] !== "cooperate" || args[1] === "enable") ? await deps.queue.run(observeSession, runOnce, exec.signal) diff --git a/packages/dsh-plugin-browserskill/tests/cooperation.browser.test.ts b/packages/dsh-plugin-browserskill/tests/cooperation.browser.test.ts new file mode 100644 index 000000000..a1a0d1305 --- /dev/null +++ b/packages/dsh-plugin-browserskill/tests/cooperation.browser.test.ts @@ -0,0 +1,265 @@ +// Real CLI, daemon, extension and plugin; all browser/profile/runtime state is isolated. +import { execFile, spawn } from "node:child_process"; +import { once } from "node:events"; +import { cp, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import type { ToolDefinition, ToolRunContext } from "@deepseek-ai/dsh-tools"; +import { expect, it } from "vitest"; +import { + attachVideoBackground, + type BrowserSend, +} from "../../../apps/extension/src/video/browser-fixture"; +import { registerBrowserTools } from "../src/browser-tools"; +import { ObservationService } from "../src/observation"; +import { KeyedExecutor } from "../src/queue"; +import { createBskRunner } from "../src/runner"; +import { SessionRegistry } from "../src/sessions"; + +it + .skipIf(!process.env.BSK_COOP_CHROME || !process.env.BSK_COOP_BSK) + .each(["click", "press", "drag", "fill", "select"] as const)( + "settles cancelled thumbnails before sync and preserves navigating %s receipts", + async (action) => { + // Keep the daemon's Unix socket path below macOS's length limit. + const directory = await mkdtemp(path.join(tmpdir(), "bsk-coop-")); + const executable = path.resolve(process.env.BSK_COOP_BSK!); + const env = { + ...process.env, + BSK_HOME: path.join(directory, "state"), + BSK_AUTO_START: "0", + BSK_AUTO_UPDATE: "off", + }; + const daemon = spawn(executable, ["daemon", "start", "--foreground", "--port", "0"], { + env, + stdio: "ignore", + }); + let arrivals = 0; + const server = createServer((request, response) => { + response.setHeader("Content-Type", "text/html"); + if (request.url?.startsWith("/done")) { + arrivals++; + response.end("

Navigation completed

"); + } else { + response.end(`

Settlement regression

Continue +
+ + + +
+
Drag to submit
`); + } + }); + const cli = async (...args: string[]) => { + const result = await promisify(execFile)(executable, [...args, "--json"], { + env, + timeout: 20000, + }); + return JSON.parse(result.stdout); + }; + try { + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const url = `http://127.0.0.1:${(server.address() as { port: number }).port}/`; + let port = 0; + await expect + .poll( + async () => { + try { + port = JSON.parse( + await readFile(path.join(env.BSK_HOME!, "daemon.json"), "utf8"), + ).ws_port; + return port; + } catch { + return 0; + } + }, + { timeout: 15000 }, + ) + .toBeGreaterThan(0); + const extension = path.join(directory, "extension"); + await cp(path.resolve("../../apps/extension/dist/chrome-mv3"), extension, { + recursive: true, + }); + const manifestPath = path.join(extension, "manifest.json"); + const manifest = JSON.parse(await readFile(manifestPath, "utf8")); + manifest.name = "BrowserSkill cooperation settlement regression"; + await writeFile(manifestPath, JSON.stringify(manifest)); + const backgroundPath = path.join(extension, "background.js"); + const background = await readFile(backgroundPath, "utf8"); + expect(background).toContain("ws://127.0.0.1:52800"); + await writeFile( + backgroundPath, + background.replaceAll("ws://127.0.0.1:52800", `ws://127.0.0.1:${port}`), + ); + const { withChrome } = await import( + new URL( + "../../../evals/browser/cases/regression/snapshot-coordinates/chrome.mjs", + import.meta.url, + ).href + ); + await withChrome( + { + executable: process.env.BSK_COOP_CHROME, + extensionPath: extension, + deviceScale: 1, + zoom: 1, + }, + async (send: BrowserSend) => { + await expect + .poll(async () => (await cli("browsers")).length, { timeout: 15000 }) + .toBeGreaterThan(0); + const worker = await attachVideoBackground(send, manifest.name); + const evaluate = async (expression: string) => { + const reply = await send<{ result: { value: T }; exceptionDetails?: unknown }>( + "Runtime.evaluate", + { expression, returnByValue: true, awaitPromise: true }, + worker.sessionId, + ); + expect(reply.exceptionDetails).toBeUndefined(); + return reply.result.value; + }; + const tools = new Map(); + const ctx = { + tools: { + register: (tool: ToolDefinition) => { + tools.set(tool.name, tool); + return () => tools.delete(tool.name); + }, + }, + get: () => undefined, + }; + const runner = createBskRunner(executable, (command, args, options) => + spawn(command, args, { ...options, env: { ...options?.env, ...env } }), + ); + const registry = new SessionRegistry(1); + const queue = new KeyedExecutor(); + const config = { + bskPath: executable, + defaultTimeoutMs: 15000, + maxSessions: 1, + observationEnabled: true, + thumbnailIntervalMs: 1500, + idleIntervalMs: 8000, + lazyTools: false, + }; + const observation = new ObservationService({ + ctx: ctx as never, + runner, + registry, + queue, + options: { enabled: true, thumbnailIntervalMs: 1500, idleIntervalMs: 8000 }, + }); + registerBrowserTools({ ctx: ctx as never, runner, registry, queue, config, observation }); + const call = (name: string, args: object) => + tools.get(name)!.execute( + args as never, + { + callId: "regression", + name, + signal: new AbortController().signal, + } as ToolRunContext, + ) as Promise>; + let session: string | undefined; + let unsubscribe = () => {}; + try { + session = (await call("bsk_session", { action: "start", url, noFocus: true })) + .sessionId; + // Only this fixture holds one Chrome reply. The installed extension, + // dispatcher, daemon cancellation and plugin scheduler remain real. + await evaluate(`(() => { + const original = chrome.debugger.sendCommand.bind(chrome.debugger); + const committed = new Promise(resolve => { + const onCommitted = details => { + if (details.frameId !== 0 || new URL(details.url).pathname !== '/done') return; + chrome.webNavigation.onCommitted.removeListener(onCommitted); + resolve(); + }; + chrome.webNavigation.onCommitted.addListener(onCommitted); + }); + globalThis.holdNextShot = true; + chrome.debugger.sendCommand = async (target, method, params) => { + if (method === 'Page.captureScreenshot' && globalThis.holdNextShot) { + globalThis.holdNextShot = false; + return new Promise(resolve => { globalThis.finishHeldShot = () => resolve({data: ''}); }); + } + const result = await original(target, method, params); + const value = result?.result?.value; + // Hold only a completed input/verification reply until navigation commits. + if ((method === 'Input.dispatchKeyEvent' && params.type === 'keyUp' && params.key === 'Enter') || + (method === 'Input.dispatchMouseEvent' && params.type === 'mouseReleased') || + (method === 'Runtime.callFunctionOn' && (value?.selected_values || value?.connected && value?.matches))) + await committed; + return result; + }; + })()`); + unsubscribe = observation.subscribe(() => {}); + await expect + .poll(() => evaluate("typeof globalThis.finishHeldShot === 'function'"), { + timeout: 10000, + }) + .toBe(true); + const syncing = call("bsk_session", { action: "sync", session }); + unsubscribe(); + const synced = await syncing; + expect(synced.context_id).toEqual(expect.any(String)); + expect(observation.getState()[0].thumbnailAttachmentId).toBeUndefined(); + expect((await cli("debug", "activity", "--session", session!)).activity.state).toBe( + "idle", + ); + await evaluate("globalThis.finishHeldShot()"); + const fresh = await call("bsk_session", { action: "sync", session }); + expect(fresh.context_id).not.toBe(synced.context_id); + expect(fresh.kind).toBe("unchanged"); + const ref = synced.observation.text.match(/(@e\d+) link "Continue"/)?.[1]; + expect(ref).toBeDefined(); + const input = { + click: { target: ref }, + press: { target: "input", key: "Enter" }, + drag: { target: "#drag", deltaX: 80, durationMs: 50 }, + fill: { target: "input", value: "submitted" }, + select: { target: "select", values: ["b"] }, + }[action]; + const receipt = await call("bsk_interact", { + action, + session, + ...input, + contextId: fresh.context_id, + }); + expect(action === "drag" ? receipt.tab_id : receipt.tabId).toEqual(expect.any(Number)); + const destination = await call("bsk_session", { action: "sync", session }); + expect(destination.observation.text).toContain("Navigation completed"); + expect(arrivals).toBe(1); + await expect( + call("bsk_interact", { + action, + session, + ...input, + contextId: fresh.context_id, + }), + ).rejects.toThrow(); + expect(arrivals).toBe(1); + } finally { + unsubscribe(); + observation.dispose(); + if (session) await cli("session", "stop", session); + runner.killAll(); + } + }, + ); + } finally { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + if (daemon.exitCode === null && daemon.signalCode === null) { + const exited = once(daemon, "exit"); + daemon.kill(); + await exited; + } + await rm(directory, { recursive: true, force: true }); + } + }, + 60000, +); diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index a0fc4cea1..7a1692aa8 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -2280,6 +2280,117 @@ describe("screenshot scratch file lifecycle", () => { }); describe("observation action instrumentation timing", () => { + it("keeps shared-control reads usable while an ordinary action owns the queue", async () => { + const { ctx, tools } = makeCtx(); + const registry = new SessionRegistry(1); + const queue = new KeyedExecutor(); + const { runner, calls } = fakeRunner({ + "session start": START_REPLY("s1"), + cooperate: { ready: true }, + }); + const observation = disabledObservation({ ctx, runner, registry }); + registerBrowserTools({ + ctx: ctx as never, + runner, + registry, + queue, + observation, + config: CONFIG, + }); + await startSession(tools); + let release!: () => void; + const busy = queue.run( + "s1", + () => + new Promise((resolve) => { + release = resolve; + }), + ); + try { + await vi.waitFor(() => expect(release).toBeDefined()); + const sync = tools.get("session.cooperate.sync")!.execute({}, makeExec()); + await vi.waitFor(() => expect(calls.some((call) => call.args[0] === "cooperate")).toBe(true)); + await sync; + } finally { + release(); + await busy; + observation.dispose(); + } + }); + + it.each([ + "sync", + "wait", + "status", + "cancel-sync", + ])("waits for preempted thumbnail cleanup before cooperate %s", async (action) => { + const { ctx, tools } = makeCtx(); + const registry = new SessionRegistry(1); + const queue = new KeyedExecutor(); + const fake = fakeRunner({ "session start": START_REPLY("s1"), cooperate: { ready: true } }); + let finish!: (result: BskRunResult) => void; + const capture = new Promise((resolve) => { + finish = resolve; + }); + const shotStarted = vi.fn(); + const runner: BskRunner = { + ...fake.runner, + run: (args, options) => { + if (args[0] === "screenshot") { + shotStarted(); + return capture; + } + return fake.runner.run(args, options); + }, + killFor: vi.fn(() => 1), + }; + const observation = new ObservationService({ + ctx: ctx as never, + runner, + registry, + queue, + options: { enabled: true, thumbnailIntervalMs: 1500, idleIntervalMs: 8000 }, + }); + registerBrowserTools({ + ctx: ctx as never, + runner, + registry, + queue, + observation, + config: CONFIG, + }); + await startSession(tools); + const unsubscribe = observation.subscribe(() => {}); + const cancelled = { code: 2, stdout: "{}", stderr: "", timedOut: false, aborted: false }; + try { + await vi.waitFor(() => expect(shotStarted).toHaveBeenCalledOnce()); + const abort = new AbortController(); + const operation = action === "cancel-sync" ? "sync" : action; + const result = tools + .get(`session.cooperate.${operation}`)! + .execute({}, makeExec({ signal: abort.signal })); + const outcome = Promise.resolve(result).then( + (value) => ({ value }), + (error) => ({ error }), + ); + await vi.waitFor(() => expect(runner.killFor).toHaveBeenCalledWith("observation:s1")); + expect(fake.calls.some((call) => call.args[0] === "cooperate")).toBe(false); + if (action === "cancel-sync") { + abort.abort(); + expect(await outcome).toMatchObject({ error: expect.any(Error) }); + expect(fake.calls.some((call) => call.args[0] === "cooperate")).toBe(false); + } else { + finish(cancelled); + expect(await outcome).not.toHaveProperty("error"); + expect(fake.calls.filter((call) => call.args[0] === "cooperate")).toHaveLength(1); + } + } finally { + unsubscribe(); + observation.dispose(); + finish(cancelled); + } + }); + it("a queued call never overwrites the in-flight action label", async () => { const { ctx, tools } = makeCtx(); const registry = new SessionRegistry(5);