Merge pull request #529 from Tencent/codex/fix-527-cooperation

fix: settle observation captures and preserve completed input receipts

Author: iuyo5678

.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:

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

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(

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;

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<void>((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) => {

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 = <T>(work: Promise<T>, 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),
     );

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)