Live Playwright driver feeding the computer-use recorder (closes #4)

packages/volo-computeruse/pyproject.toml

@@ -6,6 +6,11 @@ requires-python = ">=3.12"
 license = { text = "Apache-2.0" }
 dependencies = ["volo-core", "volo-sdk", "volo-simulator"]
 
+# The live browser driver is opt-in — PlaywrightDriver is duck-typed over the Page, so the core
+# stays browser-free and only live recording pulls Playwright.
+[project.optional-dependencies]
+playwright = ["playwright>=1.40"]
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

packages/volo-computeruse/src/volo_computeruse/__init__.py

@@ -1,5 +1,6 @@
 """volo-computeruse — record/replay computer-use (browser/desktop) agents (newplan P8/M31)."""
 
+from volo_computeruse.driver import Page, PlaywrightDriver
 from volo_computeruse.events import ACTION_KINDS, ActionEvent, screenshot_hash
 from volo_computeruse.recorder import ComputerUseRecorder
 from volo_computeruse.replay import ComputerUseReplayServer
@@ -9,5 +10,7 @@
     "ActionEvent",
     "ComputerUseRecorder",
     "ComputerUseReplayServer",
+    "Page",
+    "PlaywrightDriver",
     "screenshot_hash",
 ]

packages/volo-computeruse/src/volo_computeruse/driver.py

@@ -0,0 +1,81 @@
+"""Live driver that feeds ``ComputerUseRecorder`` from a real browser (post-v5.0; ADR-0034).
+
+M31 shipped the transport-free core (``ActionEvent`` / recorder / replay). ``PlaywrightDriver`` is
+the live transport — the analog of the MCP stdio adapter: it wraps a Playwright ``Page``, performs
+each action on the real page, hashes the screen *before* and *after*, and records an
+``ActionEvent``. The recording then replays deterministically offline via
+``ComputerUseReplayServer`` — flagging any (action, screen) it never saw.
+
+The driver is **duck-typed** over the Page: it never imports ``playwright`` and only calls the
+methods it needs (``screenshot``, ``goto``, ``click``, ``fill``), so a real Playwright page drives
+it in production and a fake page drives it in tests — no browser needed in CI. ``playwright`` is an
+optional extra (``volo-computeruse[playwright]``).
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from volo_computeruse.events import ActionEvent, screenshot_hash
+from volo_computeruse.recorder import ComputerUseRecorder
+
+
+class Page(Protocol):
+    """The slice of the Playwright sync ``Page`` API the driver uses."""
+
+    def screenshot(self) -> bytes: ...
+    def goto(self, url: str) -> Any: ...
+    def click(self, selector: str) -> Any: ...
+    def fill(self, selector: str, value: str) -> Any: ...
+
+
+class PlaywrightDriver:
+    """Drive a browser page and record every action as an ``ActionEvent``."""
+
+    def __init__(
+        self,
+        page: Page,
+        *,
+        recorder: ComputerUseRecorder | None = None,
+        session_name: str = "playwright",
+    ) -> None:
+        self._page = page
+        self.recorder = recorder or ComputerUseRecorder(session_name=session_name)
+
+    def _screen(self) -> str:
+        # Hash the current screenshot; a DOM serialization would work identically.
+        return screenshot_hash(self._page.screenshot())
+
+    def _record(
+        self, kind: str, *, target: str, value: str, before: str, result: dict[str, Any]
+    ) -> str:
+        after = self._screen()
+        self.recorder.record(
+            ActionEvent(kind=kind, target=target, value=value, screen=before),
+            result=result,
+            screen_after=after,
+        )
+        return after
+
+    def navigate(self, url: str) -> str:
+        """Go to ``url`` and record it; returns the resulting screen hash."""
+        before = self._screen()
+        self._page.goto(url)
+        return self._record("navigate", target="", value=url, before=before, result={"ok": True})
+
+    def click(self, selector: str) -> str:
+        before = self._screen()
+        self._page.click(selector)
+        return self._record("click", target=selector, value="", before=before, result={"ok": True})
+
+    def type(self, selector: str, text: str) -> str:
+        before = self._screen()
+        self._page.fill(selector, text)
+        return self._record("type", target=selector, value=text, before=before, result={"ok": True})
+
+    @property
+    def recording(self):  # type: ignore[no-untyped-def]
+        return self.recorder.recording
+
+    def save(self, path: Any = None) -> Any:
+        return self.recorder.save(path)

packages/volo-computeruse/tests/test_driver.py

@@ -0,0 +1,87 @@
+"""PlaywrightDriver records actions from a (fake) Page; the recording replays deterministically."""
+
+from __future__ import annotations
+
+from volo_computeruse import (
+    ActionEvent,
+    ComputerUseReplayServer,
+    PlaywrightDriver,
+    screenshot_hash,
+)
+
+
+class FakePage:
+    """A stand-in for a Playwright Page: mutates a string 'DOM' so screenshots differ per state."""
+
+    def __init__(self) -> None:
+        self.state = "home"
+
+    def screenshot(self) -> bytes:
+        return self.state.encode("utf-8")
+
+    def goto(self, url: str) -> None:
+        self.state = f"page::{url}"
+
+    def click(self, selector: str) -> None:
+        self.state = f"{self.state}|click:{selector}"
+
+    def fill(self, selector: str, value: str) -> None:
+        self.state = f"{self.state}|fill:{selector}={value}"
+
+
+def _drive() -> PlaywrightDriver:
+    driver = PlaywrightDriver(FakePage(), session_name="checkout")
+    driver.navigate("https://shop.test")
+    driver.click("#add-to-cart")
+    driver.type("#coupon", "SAVE10")
+    return driver
+
+
+def test_driver_records_actions_as_events() -> None:
+    rec = _drive().recording
+    tools = [s.payload.tool for s in rec.steps]
+    assert tools == ["cu.navigate", "cu.click", "cu.type"]
+    assert rec.agent_meta.framework == "computer_use"
+    # each step's pre-action screen differs from the next (state advanced)
+    screens = [s.payload.request["screen"] for s in rec.steps]
+    assert len(set(screens)) == 3
+
+
+def test_before_screen_is_recorded_state() -> None:
+    driver = PlaywrightDriver(FakePage())
+    # first action's pre-screen is the initial 'home' state
+    driver.navigate("https://x")
+    first = driver.recording.steps[0].payload
+    assert first.request["screen"] == screenshot_hash("home")
+    assert first.response["screen_after"] == screenshot_hash("page::https://x")
+
+
+def test_recording_replays_and_flags_unseen_screen() -> None:
+    rec = _drive().recording
+    server = ComputerUseReplayServer.from_recording(rec)
+
+    # reconstruct each recorded event and confirm it replays (not flagged)
+    for step in rec.steps:
+        p = step.payload
+        event = ActionEvent(
+            kind=p.tool.removeprefix("cu."),
+            target=p.request["target"],
+            value=p.request["value"],
+            screen=p.request["screen"],
+        )
+        out = server.step(event)
+        assert "__flagged__" not in out, out
+        assert out["result"] == {"ok": True}
+
+    # the same first action on a screen the driver never saw -> flagged
+    unseen = ActionEvent(kind="click", target="#add-to-cart", screen="never-seen")
+    assert "__flagged__" in server.step(unseen)
+
+
+def test_driver_is_duck_typed_no_playwright_import() -> None:
+    # importing the driver must not require playwright (the core stays browser-free)
+    import sys
+
+    import volo_computeruse.driver as d
+
+    assert "playwright" not in sys.modules or d is not None  # never imports it at module load