diff --git a/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py b/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py index ebcf64843..074767c40 100644 --- a/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py +++ b/packages/agentenv-protocol/src/agentenv_protocol/computer/conformance/checks.py @@ -13,6 +13,7 @@ from collections.abc import Awaitable, Callable from datetime import datetime from importlib import resources +from typing import Any from mcp import ClientSession from mcp.types import CallToolResult @@ -152,6 +153,10 @@ async def shot(self, **args) -> tuple[CallToolResult, object]: raise AssertionError(f"computer_screenshot failed: {result.content[0].text}") return result, image(result) + async def pixel(self, x: int, y: int) -> tuple[int, int, int]: + """The pixel at frame pixel (x, y) of a new frame.""" + return (await self.shot())[1].getpixel((x, y)) + async def act(self, name: str, args: dict) -> None: result = await self.tool(name, args) if result.isError: @@ -161,13 +166,15 @@ async def state(self) -> dict: out = await self.driver.run("cat", f"{STAGE}/state/state.json") return json.loads(out["stdout"]) - async def until(self, holds: Callable[[dict], bool]) -> tuple[bool, dict]: - """Whether the fixture's state comes to satisfy ``holds`` within SETTLE_S, and the state it last had.""" + async def until(self, holds: Callable[[Any], bool], + read: Callable[[], Awaitable[Any]] | None = None) -> tuple[bool, Any]: + """Whether what ``read`` gives, the fixture's state unless it's given, comes to satisfy ``holds`` within + SETTLE_S, and what it last gave.""" deadline = time.monotonic() + SETTLE_S while True: - state = await self.state() - if holds(state) or time.monotonic() > deadline: - return holds(state), state + seen = await (read or self.state)() + if holds(seen) or time.monotonic() > deadline: + return holds(seen), seen await asyncio.sleep(POLL_S) async def settled(self, holds: Callable[[dict], bool], show: Callable[[dict], object]) -> str | None: @@ -588,8 +595,8 @@ async def frame_space() -> str | None: problem = await self.settled(lambda s: s["cells"].get("2,3") == 1, lambda s: s["cells"]) if problem: return problem - _, img = await self.shot() - return None if img.getpixel((x, y)) == (255, 0, 0) else f"the next frame shows {img.getpixel((x, y))}" + shows, pixel = await self.until(lambda p: p == (255, 0, 0), lambda: self.pixel(x, y)) + return None if shows else f"no frame within {SETTLE_S:g} s shows it: the last shows {pixel}" async def norm_space() -> str | None: x, y = cell(5, 5) @@ -636,7 +643,7 @@ async def widgets() -> str | None: return await self.settled(lambda s: s["saved_count"] == 1 and s["subscribed"], lambda s: (s["saved_count"], s["subscribed"])) - await self.check("P4", "a click in frame space hits its cell, and the next frame shows it", frame_space) + await self.check("P4", "a click in frame space hits its cell, and a frame shows it", frame_space) if self.driver.declares(COMPUTER_SPACES): await self.check("P4", "a click in norm space hits its cell", norm_space, COMPUTER_SPACES) if "right" in buttons: diff --git a/packages/agentenv-protocol/tests/test_computer_conformance.py b/packages/agentenv-protocol/tests/test_computer_conformance.py index 3f2f8ddd6..d4ff28957 100644 --- a/packages/agentenv-protocol/tests/test_computer_conformance.py +++ b/packages/agentenv-protocol/tests/test_computer_conformance.py @@ -85,12 +85,43 @@ async def test_a_driver_given_a_control_token_passes_with_it_and_refuses_calls_w assert not unauthorized.passed and "401 unauthorized" in _failed(unauthorized)[0] +class StaleFrame(FixtureDriver): + """Its first ``stale`` captures after a click are from before the click, as when an app records a click before it + repaints.""" + + def __init__(self, stale: int = 1, **kwargs) -> None: + super().__init__(**kwargs) + self.stale, self.left, self.before_click = stale, 0, None + + @click + async def click_at(self, x, y, button, count, modifiers): + self.before_click, self.left = self.screen.image.copy(), self.stale + return await super().click_at(x, y, button, count, modifiers) + + @capture + async def grab(self): + if not self.left: + return self.screen.image.copy() + self.left -= 1 + return self.before_click.copy() + + @pytest.mark.asyncio -async def test_a_driver_whose_default_space_is_norm_passes_at_full(): - report = await _probe(FixtureDriver(settings=ComputerSettings(default_space="norm"))) +@pytest.mark.parametrize("driver", [lambda: FixtureDriver(settings=ComputerSettings(default_space="norm")), StaleFrame], + ids=["default-space-norm", "stale-first-frame"]) +async def test_a_driver_whose_default_space_is_norm_or_whose_first_frame_after_a_click_is_stale_passes_at_full(driver): + report = await _probe(driver()) assert report.passed, _failed(report) +@pytest.mark.asyncio +async def test_frames_that_never_show_a_click_the_fixture_saw_fail_p4_with_the_last_pixel(monkeypatch): + monkeypatch.setattr(checks, "SETTLE_S", 0.5) + report = await _probe(StaleFrame(stale=10**6)) + failed = next(c for c in report.checks if c.name == "a click in frame space hits its cell, and a frame shows it") + assert (failed.status, failed.detail) == ("failed", "no frame within 0.5 s shows it: the last shows (255, 255, 255)") + + @pytest.mark.asyncio async def test_a_tool_already_hidden_from_the_probes_role_stays_hidden(): with served(FixtureDriver()) as url: @@ -195,7 +226,7 @@ async def run(self, argv, cwd, env, timeout_s): @pytest.mark.asyncio @pytest.mark.parametrize("driver, broken", [ (NeverReady, "computer-ready reports ready"), - (ClicksOff, "a click in frame space hits its cell, and the next frame shows it"), + (ClicksOff, "a click in frame space hits its cell, and a frame shows it"), (ShiftedCapture, "markers at odd and even pixels read back in frame pixels"), (AsciiOnly, "Unicode and mixed-case text arrives whole"), (SwappedScroll, "scrolls up, down, left and right each arrive that many times"), diff --git a/packages/agentenv-protocol/tests/test_computer_conformance_checks.py b/packages/agentenv-protocol/tests/test_computer_conformance_checks.py index b2a142e04..f44521b13 100644 --- a/packages/agentenv-protocol/tests/test_computer_conformance_checks.py +++ b/packages/agentenv-protocol/tests/test_computer_conformance_checks.py @@ -229,7 +229,7 @@ async def test_effects_without_spaces_right_or_double_clicks_leave_those_out_and assert "a click in norm space hits its cell" not in got and "a right click arrives as one" not in got assert "a double click arrives as one" not in got and "norm points land on the same pixels" not in got assert got["16 clicks sent at once all land"] == ("failed", "16 of 16 were refused") - assert got["a click in frame space hits its cell, and the next frame shows it"][1].startswith("AssertionError") + assert got["a click in frame space hits its cell, and a frame shows it"][1].startswith("AssertionError") @pytest.mark.asyncio diff --git a/src/agent_env/drivers/envs/reference/driver.py b/src/agent_env/drivers/envs/reference/driver.py index a8de1b3b0..ff3b4965d 100644 --- a/src/agent_env/drivers/envs/reference/driver.py +++ b/src/agent_env/drivers/envs/reference/driver.py @@ -1,6 +1,7 @@ """The reference computer driver, for an X11 desktop: mss captures the screen and XTEST, through python-xlib, sends the input. X calls run off the event loop, one at a time, on a connection checked before each call and reopened when it -drops, so the driver outlives a restarted X server. It records the display through FFmpeg's ``x11grab``, by PyAV. It +drops, so the driver outlives a restarted X server, and input that lets go of a button returns once the server has let +go of it. It records the display through FFmpeg's ``x11grab``, by PyAV. It needs ``agentenv-framework-protocol[computer]``, ``mss`` and ``python-xlib`` where it runs.""" from __future__ import annotations @@ -51,6 +52,8 @@ ) BUTTONS = {"left": 1, "middle": 2, "right": 3, "back": 8, "forward": 9} +#: The pointer state's bit for each button that has one. +BUTTON_MASKS = {"left": X.Button1Mask, "middle": X.Button2Mask, "right": X.Button3Mask} SCROLL = {"up": 4, "down": 5, "left": 6, "right": 7} #: W3C key values and their X keysym names; any other single character is typed as its own keysym. KEYSYMS = { @@ -63,6 +66,9 @@ TYPED = {"\n": "Return", "\t": "Tab"} #: The pause between the presses of a double or triple click, in seconds. CLICK_GAP_S = 0.04 +#: How long a button's release may take to reach the app, and how often the driver looks. +RELEASE_WAIT_S = 5.0 +RELEASE_POLL_S = 0.005 #: How far apart a drag's pointer moves are, in pixels. DRAG_STEP_PX = 20 DROPPED = (ConnectionClosedError, DisplayConnectionError, OSError) @@ -130,6 +136,7 @@ def act(c): if i + 1 < count: c.d.sync() time.sleep(CLICK_GAP_S) + c.released(button) c.release(reversed(held)) await self._send(act) @@ -147,6 +154,7 @@ def start(c): def finish(c): c.button(button, False) + c.released(button) c.release(reversed(held)) try: @@ -183,7 +191,10 @@ async def down(self, button) -> None: @pointer_up async def up(self, button) -> None: - await self._send(lambda c: c.button(button, False)) + def act(c): + c.button(button, False) + c.released(button) + await self._send(act) @pointer_position async def where(self) -> tuple[int, int]: @@ -287,6 +298,14 @@ def move(self, x, y) -> None: def button(self, button: str, down: bool) -> None: xtest.fake_input(self.d, X.ButtonPress if down else X.ButtonRelease, BUTTONS[button]) + def released(self, button: str) -> None: + """Wait, up to RELEASE_WAIT_S, until the server has let go of ``button``. A window manager's passive grab holds + a click on a window until it replays it to the app, while keys reach the app at once, so a key sent before then + overtakes the click, and so do the next click's moves.""" + deadline = time.monotonic() + RELEASE_WAIT_S + while self.d.screen().root.query_pointer().mask & BUTTON_MASKS.get(button, 0) and time.monotonic() < deadline: + time.sleep(RELEASE_POLL_S) + def hold(self, keysyms: list[int]) -> list[int]: """``hold_each``'s keycodes, in the order they were pressed.""" return [code for pressed in self.hold_each(keysyms) for code in pressed] diff --git a/tst/unit/drivers/reference_driver_test.py b/tst/unit/drivers/reference_driver_test.py index cbed28526..281306d72 100644 --- a/tst/unit/drivers/reference_driver_test.py +++ b/tst/unit/drivers/reference_driver_test.py @@ -26,6 +26,7 @@ TOKYO = [0x01000000 | ord(ch) for ch in "東京"] #: The fake keymap puts the named keys, such as Shift_L, from this keycode up; printable ASCII is its keysym + 8. NAMED_KEYS = 200 +BUTTON1_MASK = 1 << 8 class ConnectionClosedError(Exception): @@ -34,8 +35,9 @@ class ConnectionClosedError(Exception): class FakeX: """An X display: its screen and pointer, a US-like keymap whose capital letters are the second level of their - keys, the keys down, the XTEST events sent, ``drops`` syncs and the ``fail_at``-th event that fail, and the - ``spare`` keycodes the keymap leaves empty.""" + keys, the keys down, the XTEST events sent, ``drops`` syncs and the ``fail_at``-th event that fail, the ``spare`` + keycodes the keymap leaves empty, and ``held``, how many more queries of the pointer find the left button down + after its release, as while a window manager's grab holds a click before replaying it.""" def __init__(self) -> None: self.events: list = [] @@ -46,12 +48,18 @@ def __init__(self) -> None: self.drops = 0 self.fail_at = None self.spare = {8} + self.held = 0 self.display = types.SimpleNamespace(info=types.SimpleNamespace(min_keycode=8, max_keycode=255)) def screen(self): - query = lambda: types.SimpleNamespace(root_x=self.pointer[0], root_y=self.pointer[1]) # noqa: E731 return types.SimpleNamespace(width_in_pixels=WIDTH, height_in_pixels=HEIGHT, - root=types.SimpleNamespace(query_pointer=query)) + root=types.SimpleNamespace(query_pointer=self.query_pointer)) + + def query_pointer(self): + held, self.held = self.held > 0, max(0, self.held - 1) + if held and not self.held: + self.events.append(("replayed", 1)) + return types.SimpleNamespace(root_x=self.pointer[0], root_y=self.pointer[1], mask=BUTTON1_MASK if held else 0) def sync(self) -> None: if self.drops: @@ -156,7 +164,8 @@ def module(name: str, **attrs) -> types.ModuleType: return m X = module("Xlib.X", MotionNotify="move", ButtonPress="button-down", ButtonRelease="button-up", - KeyPress="key-down", KeyRelease="key-up") + KeyPress="key-down", KeyRelease="key-up", Button1Mask=BUTTON1_MASK, Button2Mask=BUTTON1_MASK << 1, + Button3Mask=BUTTON1_MASK << 2) XK = module("Xlib.XK", string_to_keysym=lambda name: keysyms.setdefault(name, 0xff00 + len(keysyms))) display = module("Xlib.display", Display=open_display) module("Xlib.error", ConnectionClosedError=ConnectionClosedError, DisplayConnectionError=ConnectionClosedError) @@ -230,6 +239,24 @@ async def test_tool_calls_become_the_xtest_events_they_mean(x11): assert not fake.down +@pytest.mark.asyncio +async def test_input_that_lets_go_of_a_button_returns_once_a_window_manager_holding_it_has_replayed_it(x11): + driver, fake, code = x11 + shift = code("Shift_L") + async with mcp_session(driver) as session: + await session.call_tool("computer_screenshot", {}) + fake.held = 3 + clicked = await _events(fake, session, "computer_click", {"x": 10, "y": 20, "modifiers": ["Shift"]}) + fake.held = 3 + dragged = await _events(fake, session, "computer_drag", {"path": [[0, 0], [40, 0]], "duration_ms": 0}) + await session.call_tool("computer_pointer_down", {"x": 5, "y": 5}) + fake.held = 3 + up = await _events(fake, session, "computer_pointer_up", {}) + assert clicked == [("key-down", shift), ("move", (10, 20)), ("button-down", 1), ("button-up", 1), ("replayed", 1), + ("key-up", shift)] + assert dragged[-2:] == up == [("button-up", 1), ("replayed", 1)] and fake.held == 0 + + @pytest.mark.asyncio async def test_keys_held_by_keyboard_down_stay_down_through_other_input(x11): driver, fake, code = x11