diff --git a/README.md b/README.md index a855ca90..ead59627 100755 --- a/README.md +++ b/README.md @@ -716,6 +716,12 @@ MCP Client can access the following tools to interact with Windows: drag, set `from_loc=[x, y]` with `drag=True` to press at an explicit start point and release at `loc` in one tool call. Optional `duration` adds bounded intermediate movement. +- `Pointer`: Build a stateful gesture across calls with `down`, one or more `move` + actions, and `up`. Use `cancel` to release only the tracked button for recovery; + held buttons are automatically released after 30 seconds by default (maximum + 120 seconds). If Windows rejects an automatic release, it retries twice and + retains the tracked state so `cancel` can retry manually. Prefer `Move(drag=True)` + for a simple atomic drag. - `Shortcut`: Press keyboard shortcuts (`Ctrl+c`, `Alt+Tab`, etc). - `Wait`: Pause for a defined duration. - `WaitFor`: Wait until text, an active window, an element, or a focused element appears by polling UI state inside one tool call. diff --git a/SECURITY.md b/SECURITY.md index 1a7d9d28..50c0aa51 100755 --- a/SECURITY.md +++ b/SECURITY.md @@ -99,6 +99,7 @@ These tools can make permanent changes to your system: |------|------|-------------| | **Shell** | Critical | Can execute arbitrary PowerShell commands, including system modifications, file deletions, and network operations | | **Click** | High | Can trigger destructive UI actions (delete confirmations, system dialogs) | +| **Pointer** | High | Can hold and move mouse buttons across calls; use `up` or `cancel` to finish gestures | | **Type** | High | Can overwrite text, potentially destroying data when `clear=True` | | **Drag** | High | Can move/reorganize files, potentially overwriting existing files | | **Shortcut** | High | Can execute destructive keyboard shortcuts (Ctrl+D delete, Alt+F4 close) | @@ -318,4 +319,4 @@ Users are solely responsible for: ## License -This security policy is part of the Windows-MCP project, licensed under the MIT License. See [LICENSE](LICENSE.md) for details. \ No newline at end of file +This security policy is part of the Windows-MCP project, licensed under the MIT License. See [LICENSE](LICENSE.md) for details. diff --git a/manifest.json b/manifest.json index ac608fd4..8ab5a148 100755 --- a/manifest.json +++ b/manifest.json @@ -123,6 +123,10 @@ "name": "Move", "description": "Moves mouse cursor to coordinates [x, y] or passing a UI element's label/id. Set drag=True to perform a drag-and-drop operation from the current mouse position to the target coordinates. Default (drag=False) is a simple cursor move (hover). Provide either loc or label." }, + { + "name": "Pointer", + "description": "Controls one stateful mouse-button gesture across calls with down, move, up, and cancel actions. Cancel and timeout release only the tracked button. Held buttons are automatically released after 30 seconds by default (maximum 120 seconds). Prefer Move with drag=True for a simple atomic drag." + }, { "name": "Shortcut", "description": "Executes keyboard shortcuts using key combinations separated by +. Examples: \"ctrl+c\" (copy), \"ctrl+v\" (paste), \"alt+tab\" (switch apps), \"win+r\" (Run dialog), \"win\" (Start menu), \"ctrl+shift+esc\" (Task Manager). Use for quick actions and system commands." diff --git a/src/windows_mcp/__main__.py b/src/windows_mcp/__main__.py index 0ef5877c..b54f5e9e 100755 --- a/src/windows_mcp/__main__.py +++ b/src/windows_mcp/__main__.py @@ -340,19 +340,27 @@ async def lifespan(app: FastMCP): logger.debug("Server started, entering main loop") yield finally: - await notifier.close() try: - controller.stop() + await notifier.close() finally: try: - control_overlay.stop() + # Release Pointer before the controller's ledger retry. + if desktop: + try: + desktop.close() + except Exception: + logger.exception("Failed to release desktop input during shutdown") + controller.stop() finally: - logger.debug("Shutting down: stopping watchdog and analytics") - if watchdog: - watchdog.stop() - if analytics: - await analytics.close() - control_loop = None + try: + control_overlay.stop() + finally: + logger.debug("Shutting down: stopping watchdog and analytics") + if watchdog: + watchdog.stop() + if analytics: + await analytics.close() + control_loop = None _mcp = FastMCP(name="windows-mcp", instructions=instructions, lifespan=lifespan) _mcp.add_middleware(ControlToolGate(controller, notifier)) diff --git a/src/windows_mcp/desktop/control.py b/src/windows_mcp/desktop/control.py index b1ccb7f5..2f217f13 100644 --- a/src/windows_mcp/desktop/control.py +++ b/src/windows_mcp/desktop/control.py @@ -223,6 +223,14 @@ def _tick_locked(self, now: float) -> bool: "ai" if self._active_calls or now < self._lease_until else "ready" ) if self._state == "ai" and not self._active_calls and now >= self._lease_until: + pointer_hold = self.input_ledger.pointer_hold_state() + if pointer_hold is not None: + if pointer_hold: + # Pointer's bounded timer or physical takeover ends this + # cross-call hold; the ordinary 15-second lease does not. + return False + self._fail_open() + return self._set_locked("unavailable") if now - self._last_physical_event < 10.0: self._last_user = self._last_physical_event return self._set_locked("user") @@ -250,12 +258,25 @@ def begin_call(self, name: str) -> int: status["state"], "CONTROL_UNAVAILABLE" ) raise ControlBlocked(code, status) - try: - self.input_ledger.enable() # Any failed release keeps the lease closed. - except RuntimeError: - self._fail_open() - self._set_locked("unavailable") - raise ControlBlocked("CONTROL_UNAVAILABLE", self._snapshot_locked(now)) from None + pointer_hold = self.input_ledger.pointer_hold_state() + if pointer_hold is not None: + if self._state != "ai" or not pointer_hold: + self._fail_open() + self._set_locked("unavailable") + raise ControlBlocked("CONTROL_UNAVAILABLE", self._snapshot_locked(now)) + if name != "Pointer": + raise ControlBlocked("AI_INPUT_HELD", self._snapshot_locked(now)) + # The existing AI lease already enabled the ledger. Do not + # call enable(), which correctly rejects an outstanding hold. + else: + try: + self.input_ledger.enable() # Failed releases keep the lease closed. + except RuntimeError: + self._fail_open() + self._set_locked("unavailable") + raise ControlBlocked( + "CONTROL_UNAVAILABLE", self._snapshot_locked(now) + ) from None changed = self._set_locked("ai") or changed self._active_calls += 1 token = self._generation diff --git a/src/windows_mcp/desktop/control_ledger.py b/src/windows_mcp/desktop/control_ledger.py index 4bc8c923..f19a9997 100644 --- a/src/windows_mcp/desktop/control_ledger.py +++ b/src/windows_mcp/desktop/control_ledger.py @@ -15,7 +15,10 @@ class InputUnavailable(ToolError): class InputLedger: def __init__(self) -> None: self._lock = threading.RLock() - self._holds: dict[str, tuple[Callable[[], None], Callable[[], bool]]] = {} + # The optional receipt binds a cross-call hold to one exact gesture. + self._holds: dict[ + str, tuple[Callable[[], None], Callable[[], bool], str | None, object | None] + ] = {} self._blocked = False def block_new(self) -> None: @@ -33,36 +36,70 @@ def press( send_down: Callable[[], None], send_up: Callable[[], None], physical_down_delivered: Callable[[], bool], + *, + owner: str | None = None, + receipt: object | None = None, ) -> None: """Record before injection: a send routine may inject and then raise.""" + if (owner is None) != (receipt is None): + raise ValueError("owner and receipt must be provided together") with self._lock: if self._blocked: raise InputUnavailable("AI input is unavailable") if name in self._holds: raise RuntimeError(f"AI input already held: {name}") - self._holds[name] = (send_up, physical_down_delivered) + self._holds[name] = (send_up, physical_down_delivered, owner, receipt) try: # Keep the lock through injection so a watchdog cannot send # up before the matching down has actually reached Windows. send_down() except BaseException: try: - self.release(name) + self.release(name, owner=owner, receipt=receipt) except Exception: pass # Keep the entry for shutdown's second attempt. raise - def release(self, name: str) -> None: + def release( + self, name: str, *, owner: str | None = None, receipt: object | None = None + ) -> bool: + """Release this hold only if an optional gesture receipt still matches.""" + if (owner is None) != (receipt is None): + raise ValueError("owner and receipt must be provided together") with self._lock: entry = self._holds.get(name) if entry is None: - return - send_up, physical_down_delivered = entry + return False + send_up, physical_down_delivered, held_owner, held_receipt = entry + if owner is not None and (held_owner != owner or held_receipt is not receipt): + return False # If a physical down reached the app after takeover, its own up # will clear the state. Injecting another up would steal the key. if not physical_down_delivered(): send_up() self._holds.pop(name, None) + return True + + def owns(self, name: str, owner: str, receipt: object) -> bool: + """Check whether an exact gesture still owns its injected down.""" + with self._lock: + entry = self._holds.get(name) + return entry is not None and entry[2] == owner and entry[3] is receipt + + def pending_entries(self) -> tuple[tuple[str, str | None], ...]: + """Return a compact ownership snapshot for the desktop control gate.""" + with self._lock: + return tuple((name, entry[2]) for name, entry in self._holds.items()) + + def pointer_hold_state(self) -> bool | None: + """None: no hold; True: one Pointer mouse hold; False: unsafe hold.""" + with self._lock: + if not self._holds: + return None + if len(self._holds) != 1: + return False + name, entry = next(iter(self._holds.items())) + return name in {"mouse:left", "mouse:right", "mouse:middle"} and entry[2] == "Pointer" def release_all(self) -> None: with self._lock: diff --git a/src/windows_mcp/desktop/pointer.py b/src/windows_mcp/desktop/pointer.py new file mode 100644 index 00000000..320ac276 --- /dev/null +++ b/src/windows_mcp/desktop/pointer.py @@ -0,0 +1,333 @@ +"""Stateful mouse pointer control with bounded automatic release.""" + +from collections.abc import Callable +import logging +import math +from threading import RLock, Timer +from time import sleep +from typing import Literal + +import windows_mcp.uia as uia +from windows_mcp.desktop.control import get_controller + + +MouseButton = Literal["left", "right", "middle"] +DEFAULT_POINTER_TIMEOUT = 30.0 +MAX_POINTER_TIMEOUT = 120.0 +MAX_POINTER_MOVE_DURATION = 10.0 +RELEASE_RETRY_DELAY = 1.0 +MAX_RELEASE_ATTEMPTS = 3 + +logger = logging.getLogger(__name__) + + +def normalize_pointer_button(value: object, *, allow_none: bool = False) -> MouseButton | None: + """Validate a mouse button name.""" + if value is None and allow_none: + return None + if value not in {"left", "right", "middle"}: + raise ValueError("button must be one of: left, right, middle") + return value + + +def normalize_pointer_point(value: object, name: str = "loc") -> tuple[int, int]: + """Validate a screen point without accepting booleans or fractional coordinates.""" + if not isinstance(value, (list, tuple)) or len(value) != 2: + raise ValueError(f"{name} must be a list or tuple of exactly 2 integers [x, y]") + x, y = value + if any(isinstance(item, bool) or not isinstance(item, int) for item in (x, y)): + raise ValueError(f"{name} must contain exactly 2 integers") + return x, y + + +def normalize_pointer_duration(value: object | None) -> float | None: + """Validate an optional bounded pointer movement duration.""" + if value is None: + return None + if isinstance(value, bool): + raise ValueError("duration must be a finite number of seconds") + try: + duration = float(value) + except (TypeError, ValueError) as exc: + raise ValueError("duration must be a finite number of seconds") from exc + if not math.isfinite(duration): + raise ValueError("duration must be a finite number of seconds") + if duration < 0 or duration > MAX_POINTER_MOVE_DURATION: + raise ValueError(f"duration must be between 0 and {MAX_POINTER_MOVE_DURATION:g} seconds") + return duration + + +def normalize_pointer_timeout(value: object | None) -> float: + """Validate the automatic mouse-button release timeout.""" + if value is None: + return DEFAULT_POINTER_TIMEOUT + if isinstance(value, bool): + raise ValueError("timeout must be a finite number of seconds") + try: + timeout = float(value) + except (TypeError, ValueError) as exc: + raise ValueError("timeout must be a finite number of seconds") from exc + if not math.isfinite(timeout): + raise ValueError("timeout must be a finite number of seconds") + if timeout <= 0 or timeout > MAX_POINTER_TIMEOUT: + raise ValueError( + f"timeout must be greater than 0 and at most {MAX_POINTER_TIMEOUT:g} seconds" + ) + return timeout + + +class PointerController: + """Serialize stateful mouse input and prevent indefinitely held buttons.""" + + def __init__( + self, + timer_factory: Callable[[float, Callable[[], None]], Timer] = Timer, + *, + control=None, + sleeper: Callable[[float], None] = sleep, + ) -> None: + self._lock = RLock() + self._button: MouseButton | None = None + self._receipt: object | None = None + self._timer: Timer | None = None + self._generation = 0 + self._timer_factory = timer_factory + self._control = control if control is not None else get_controller() + self._sleep = sleeper + + @property + def held_button(self) -> MouseButton | None: + """Return the button currently held by this controller.""" + with self._lock: + self._owns_locked() + return self._button + + def _owns_locked(self) -> bool: + """Discard a gesture already released by takeover or the watchdog.""" + if self._button is None: + return False + receipt = self._receipt + if receipt is not None and self._control.input_ledger.owns( + f"mouse:{self._button}", "Pointer", receipt + ): + return True + self._clear_locked(cancel_timer=True) + return False + + def _owns_receipt_locked(self, button: MouseButton, receipt: object) -> bool: + """A delayed move must never attach itself to a newer gesture.""" + return self._button == button and self._receipt is receipt and self._owns_locked() + + @staticmethod + def _press(button: MouseButton, x: int, y: int) -> None: + if button == "left": + uia.PressMouse(x, y, waitTime=0.05) + elif button == "right": + uia.RightPressMouse(x, y, waitTime=0.05) + else: + uia.MiddlePressMouse(x, y, waitTime=0.05) + + @staticmethod + def _release(button: MouseButton) -> None: + if button == "left": + uia.ReleaseMouse(waitTime=0.05) + elif button == "right": + uia.RightReleaseMouse(waitTime=0.05) + else: + uia.MiddleReleaseMouse(waitTime=0.05) + + def _clear_locked(self, *, cancel_timer: bool) -> None: + timer = self._timer + self._timer = None + self._button = None + self._receipt = None + self._generation += 1 + if cancel_timer and timer is not None: + timer.cancel() + + def _release_owned_locked(self) -> None: + """Release only this controller's button, not unrelated physical input.""" + button = self._button + receipt = self._receipt + if button is not None and receipt is not None: + # A failed Win32 release may leave the button down. Keep ownership + # and the timer so cancel or a later timeout can retry safely. + self._control.input_ledger.release(f"mouse:{button}", owner="Pointer", receipt=receipt) + self._clear_locked(cancel_timer=True) + + def _timeout_release(self, generation: int, attempt: int = 1) -> None: + with self._lock: + if self._button is None or self._generation != generation: + return + try: + self._release_owned_locked() + except Exception: + logger.exception("Automatic pointer release failed") + if attempt >= MAX_RELEASE_ATTEMPTS: + # Keep the ledger entry for the watchdog and stop physical + # interception rather than holding the desktop forever. + self._control._fail_open() + return + try: + timer = self._timer_factory( + RELEASE_RETRY_DELAY, + lambda: self._timeout_release(generation, attempt + 1), + ) + timer.daemon = True + self._timer = timer + timer.start() + except Exception: + logger.exception("Failed to schedule pointer release retry") + # A missing retry timer cannot justify keeping physical + # input intercepted while this AI hold remains unresolved. + self._control._fail_open() + + def down( + self, + loc: tuple[int, int] | list[int], + button: MouseButton = "left", + timeout: float | int | str | None = None, + ) -> dict[str, object]: + """Press one mouse button and schedule a bounded automatic release.""" + x, y = normalize_pointer_point(loc) + normalized_button = normalize_pointer_button(button) + normalized_timeout = normalize_pointer_timeout(timeout) + + with self._lock: + self._control.checkpoint_current() + self._owns_locked() + if self._button is not None: + raise RuntimeError( + f"Cannot press {normalized_button}; {self._button} mouse button is already held" + ) + self._button = normalized_button + receipt = object() + self._receipt = receipt + self._generation += 1 + generation = self._generation + try: + # Arm recovery before injecting input: even an ambiguous press + # failure retains a timed path to release the tracked button. + timer = self._timer_factory( + normalized_timeout, + lambda: self._timeout_release(generation), + ) + timer.daemon = True + self._timer = timer + timer.start() + except BaseException: + self._clear_locked(cancel_timer=True) + raise + + try: + self._control.input_ledger.press( + f"mouse:{normalized_button}", + lambda: self._press(normalized_button, x, y), + lambda: self._release(normalized_button), + lambda: self._control.physical_mouse_down(normalized_button), + owner="Pointer", + receipt=receipt, + ) + except BaseException: + if not self._control.input_ledger.owns( + f"mouse:{normalized_button}", "Pointer", receipt + ): + self._clear_locked(cancel_timer=True) + raise + + return { + "action": "down", + "button": normalized_button, + "loc": [x, y], + "timeout": normalized_timeout, + } + + def move( + self, + loc: tuple[int, int] | list[int], + duration: float | int | str | None = None, + ) -> dict[str, object]: + """Move the pointer while the tracked mouse button remains held.""" + x, y = normalize_pointer_point(loc) + normalized_duration = normalize_pointer_duration(duration) + + with self._lock: + if not self._owns_locked(): + raise RuntimeError("Cannot move pointer because no mouse button is held") + button, receipt = self._button, self._receipt + assert button is not None and receipt is not None + try: + if normalized_duration is None: + with self._lock: + self._control.checkpoint_current() + if not self._owns_receipt_locked(button, receipt): + raise RuntimeError("Pointer gesture was released") + # An unsmoothed move is one bounded input step. + uia.SetCursorPos(x, y) + self._control.record_step_current() + else: + start_x, start_y = uia.GetCursorPos() + steps = max(1, math.ceil(normalized_duration / 0.02)) + for step in range(1, steps + 1): + with self._lock: + self._control.checkpoint_current() + if not self._owns_receipt_locked(button, receipt): + raise RuntimeError("Pointer gesture was released") + # Do not hold the Pointer lock during sleep: its timer must + # be able to release the button at the configured deadline. + self._sleep(normalized_duration / steps) + with self._lock: + self._control.checkpoint_current() + if not self._owns_receipt_locked(button, receipt): + raise RuntimeError("Pointer gesture was released") + uia.SetCursorPos( + start_x + (x - start_x) * step // steps, + start_y + (y - start_y) * step // steps, + ) + self._control.record_step_current() + except BaseException as move_error: + with self._lock: + if self._receipt is receipt: + try: + self._release_owned_locked() + except BaseException as release_error: + raise release_error from move_error + raise + + return { + "action": "move", + "button": button, + "loc": [x, y], + "duration": normalized_duration, + } + + def up(self, button: MouseButton | None = None) -> dict[str, object]: + """Release the tracked mouse button, optionally asserting its identity.""" + normalized_button = normalize_pointer_button(button, allow_none=True) + with self._lock: + if not self._owns_locked(): + raise RuntimeError("Cannot release pointer because no mouse button is held") + if normalized_button is not None and normalized_button != self._button: + raise RuntimeError( + f"Cannot release {normalized_button}; {self._button} mouse button is held" + ) + held_button = self._button + self._release_owned_locked() + return {"action": "up", "button": held_button} + + def cancel(self) -> dict[str, object]: + """Release the tracked button and clear state without touching other buttons.""" + with self._lock: + self._owns_locked() + held_button = self._button + self._release_owned_locked() + return {"action": "cancel", "button": held_button} + + def close(self) -> None: + """Release tracked input during a normal server shutdown.""" + with self._lock: + self._owns_locked() + if self._button is None: + self._clear_locked(cancel_timer=True) + return + self._release_owned_locked() diff --git a/src/windows_mcp/desktop/service.py b/src/windows_mcp/desktop/service.py index ae3bd2ba..9e9f3d4d 100644 --- a/src/windows_mcp/desktop/service.py +++ b/src/windows_mcp/desktop/service.py @@ -16,6 +16,7 @@ from windows_mcp.desktop import screenshot as screenshot_capture from windows_mcp.desktop.control import get_controller from windows_mcp.desktop import flash_overlay +from windows_mcp.desktop.pointer import MouseButton, PointerController, normalize_pointer_point from windows_mcp.infrastructure import safe_get from windows_mcp.desktop import control_overlay from urllib.parse import urljoin @@ -83,6 +84,11 @@ def __init__(self): self.encoding = getpreferredencoding() self.tree = Tree(self) self.desktop_state = None + self._pointer = PointerController() + + def close(self) -> None: + """Release stateful input owned by this desktop service.""" + self._pointer.close() def get_state( self, @@ -985,6 +991,33 @@ def move(self, loc: tuple[int, int]): owner.record_step_current() sleep(0.003) + def pointer_down( + self, + loc: tuple[int, int] | list[int], + button: MouseButton = "left", + timeout: float | int | str | None = None, + ) -> dict[str, object]: + """Press a mouse button for a bounded stateful gesture.""" + self._validate_screen_point(*normalize_pointer_point(loc)) + return self._pointer.down(loc, button, timeout) + + def pointer_move( + self, + loc: tuple[int, int] | list[int], + duration: float | int | str | None = None, + ) -> dict[str, object]: + """Move the pointer while its tracked button remains held.""" + self._validate_screen_point(*normalize_pointer_point(loc)) + return self._pointer.move(loc, duration) + + def pointer_up(self, button: MouseButton | None = None) -> dict[str, object]: + """Release the tracked mouse button.""" + return self._pointer.up(button) + + def pointer_cancel(self) -> dict[str, object]: + """Release only the tracked mouse button and clear pointer state.""" + return self._pointer.cancel() + def shortcut(self, shortcut: str): owner = get_controller() owner.checkpoint_current() diff --git a/src/windows_mcp/tools/__init__.py b/src/windows_mcp/tools/__init__.py index 01d9a8f7..9e0f230a 100644 --- a/src/windows_mcp/tools/__init__.py +++ b/src/windows_mcp/tools/__init__.py @@ -9,6 +9,7 @@ input, multi, notification, + pointer, process, registry, scrape, @@ -24,6 +25,7 @@ filesystem, snapshot, input, + pointer, scrape, multi, clipboard, diff --git a/src/windows_mcp/tools/pointer.py b/src/windows_mcp/tools/pointer.py new file mode 100644 index 00000000..2d179a1c --- /dev/null +++ b/src/windows_mcp/tools/pointer.py @@ -0,0 +1,118 @@ +"""Pointer tool — stateful mouse down, move, up, and cancel actions.""" + +import json +from typing import Any, Literal, cast + +from fastmcp import Context +from mcp.types import ToolAnnotations + +from windows_mcp.desktop.pointer import ( + MouseButton, + normalize_pointer_button, + normalize_pointer_duration, + normalize_pointer_point, + normalize_pointer_timeout, +) +from windows_mcp.infrastructure import with_analytics + + +PointerAction = Literal["down", "move", "up", "cancel"] + + +def _normalize_action(value: object) -> PointerAction: + if not isinstance(value, str): + raise ValueError("action must be one of: down, move, up, cancel") + normalized = value.strip().lower() + if normalized not in {"down", "move", "up", "cancel"}: + raise ValueError("action must be one of: down, move, up, cancel") + return cast(PointerAction, normalized) + + +def _as_point(value: list[int] | str | None) -> tuple[int, int]: + if value is None: + raise ValueError("loc is required for Pointer down and move actions") + if isinstance(value, str): + try: + value = json.loads(value) + except json.JSONDecodeError as exc: + raise ValueError("loc must be a JSON list of exactly 2 integers [x, y]") from exc + return normalize_pointer_point(value) + + +def register(mcp: Any, *, get_desktop, get_analytics) -> None: + @mcp.tool( + name="Pointer", + description=( + "Controls one stateful mouse-button gesture across calls. Use action='down' with loc, " + "an optional left/right/middle button, and an automatic-release timeout that defaults " + "to 30 seconds and is limited to 120 seconds; then action='move' one or more times with " + "loc and optional duration; finish with action='up'. Use action='cancel' to release " + "the tracked button and clear held state without releasing unrelated mouse buttons. " + "Prefer Move with drag=True for a simple atomic drag " + "in one call." + ), + annotations=ToolAnnotations( + title="Pointer", + readOnlyHint=False, + destructiveHint=True, + idempotentHint=False, + openWorldHint=False, + ), + ) + @with_analytics(get_analytics(), "Pointer-Tool") + def pointer_tool( + action: Literal["down", "move", "up", "cancel"], + loc: list[int] | str | None = None, + button: Literal["left", "right", "middle"] | None = None, + duration: float | int | str | None = None, + timeout: float | int | str | None = None, + ctx: Context = None, + ) -> str: + normalized_action = _normalize_action(action) + + if normalized_action == "down": + if duration is not None: + raise ValueError("duration is only supported for action='move'") + point = _as_point(loc) + requested_button = "left" if button is None else button + normalized_button = cast( + MouseButton, + normalize_pointer_button(requested_button), + ) + normalized_timeout = normalize_pointer_timeout(timeout) + result = get_desktop().pointer_down(point, normalized_button, normalized_timeout) + x, y = result["loc"] + return ( + f"Pressed {result['button']} mouse button at ({x},{y}); " + f"automatic release timeout is {result['timeout']:g} seconds." + ) + + if normalized_action == "move": + if button is not None or timeout is not None: + raise ValueError("button and timeout are only supported for action='down'") + point = _as_point(loc) + normalized_duration = normalize_pointer_duration(duration) + result = get_desktop().pointer_move(point, normalized_duration) + x, y = result["loc"] + if result["duration"] is None: + return f"Moved held {result['button']} mouse button to ({x},{y})." + return ( + f"Moved held {result['button']} mouse button to ({x},{y}) " + f"over {result['duration']:.3f} seconds." + ) + + if normalized_action == "up": + if loc is not None or duration is not None or timeout is not None: + raise ValueError("loc, duration, and timeout are not supported for action='up'") + normalized_button = normalize_pointer_button(button, allow_none=True) + result = get_desktop().pointer_up(normalized_button) + return f"Released {result['button']} mouse button." + + if any(value is not None for value in (loc, button, duration, timeout)): + raise ValueError( + "loc, button, duration, and timeout are not supported for action='cancel'" + ) + result = get_desktop().pointer_cancel() + if result["button"] is None: + return "No tracked mouse button was held." + return f"Released tracked {result['button']} mouse button and cleared held state." diff --git a/tests/test_pointer_control_integration.py b/tests/test_pointer_control_integration.py new file mode 100644 index 00000000..50f254e1 --- /dev/null +++ b/tests/test_pointer_control_integration.py @@ -0,0 +1,371 @@ +"""Cross-call Pointer ownership against the merged desktop takeover gate.""" + +from __future__ import annotations + +import threading +import time +from collections.abc import Callable + +import pytest +from fastmcp import Client, FastMCP + +from windows_mcp.desktop import pointer +from windows_mcp.desktop.control import ControlBlocked, ControlCoordinator +from windows_mcp.desktop.control_context import current_token +from windows_mcp.desktop.pointer import PointerController +from windows_mcp.tools.control_notifications import ControlNotifier, ControlToolGate + + +class FakeTimer: + def __init__(self, interval: float, callback: Callable[[], None]) -> None: + self.interval = interval + self.callback = callback + self.daemon = False + + def start(self) -> None: + pass + + def cancel(self) -> None: + pass + + def fire(self) -> None: + self.callback() + + +@pytest.fixture +def setup_pointer(monkeypatch: pytest.MonkeyPatch): + owner = ControlCoordinator() + owner._state = "ready" + events: list[str] = [] + timers: list[FakeTimer] = [] + + monkeypatch.setattr(pointer.uia, "PressMouse", lambda *a, **k: events.append("down")) + monkeypatch.setattr(pointer.uia, "ReleaseMouse", lambda *a, **k: events.append("up")) + monkeypatch.setattr(pointer.uia, "GetCursorPos", lambda: (0, 0)) + monkeypatch.setattr(pointer.uia, "SetCursorPos", lambda *a: events.append("move")) + + def make_timer(interval: float, callback: Callable[[], None]) -> FakeTimer: + timer = FakeTimer(interval, callback) + timers.append(timer) + return timer + + return owner, events, timers, make_timer + + +def _gesture_call(owner: ControlCoordinator, action: Callable[[], object]) -> object: + lease = owner.begin_call("Pointer") + marker = current_token.set(lease) + try: + return action() + finally: + current_token.reset(marker) + owner.end_call(lease) + + +@pytest.mark.asyncio +async def test_pointer_gesture_crosses_real_mcp_gate_without_reopening_other_tools(setup_pointer): + owner, events, _, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + mcp = FastMCP("pointer-integration") + mcp.add_middleware(ControlToolGate(owner, ControlNotifier(owner))) + + @mcp.tool(name="Pointer") + def pointer_tool(action: str) -> str: + if action == "down": + gesture.down([1, 2]) + elif action == "move": + gesture.move([3, 4]) + else: + gesture.up() + return action + + @mcp.tool(name="App") + def app_tool() -> str: + return "unexpected" + + async with Client(mcp) as client: + await client.call_tool("Pointer", {"action": "down"}) + blocked = await client.call_tool("App", {}, raise_on_error=False) + assert blocked.is_error and "AI_INPUT_HELD" in str(blocked.content) + await client.call_tool("Pointer", {"action": "move"}) + await client.call_tool("Pointer", {"action": "up"}) + assert (await client.call_tool("App", {})).is_error is False + assert events == ["down", "move", "up"] + + +@pytest.mark.parametrize("timeout", [30, 120]) +def test_pointer_hold_survives_lease_but_blocks_other_tools(setup_pointer, timeout): + owner, events, timers, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2], timeout=timeout)) + + assert timers[0].interval == timeout + owner._lease_until = time.monotonic() + 0.01 + assert owner.status()["state"] == "ai" + owner._lease_until = time.monotonic() - 1 # Beyond the normal 15-second lease. + assert owner.status()["state"] == "ai" + with pytest.raises(ControlBlocked, match="AI_INPUT_HELD"): + owner.begin_call("App") + + _gesture_call(owner, lambda: gesture.move([3, 4])) + _gesture_call(owner, gesture.up) + owner._lease_until = time.monotonic() - 1 + assert owner.status()["state"] == "ready" + assert events == ["down", "move", "up"] + + +def test_pointer_timeout_releases_hold_before_ai_lease_ends(setup_pointer): + owner, events, timers, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2])) + owner._lease_until = time.monotonic() - 1 + assert owner.status()["state"] == "ai" + + timers[0].fire() + assert owner.input_ledger.pending() == () + assert gesture.held_button is None + assert owner.status()["state"] == "ready" + assert events == ["down", "up"] + + +def test_unowned_legacy_hold_fails_closed_after_lease(setup_pointer): + owner, events, _, _ = setup_pointer + lease = owner.begin_call("App") + owner.input_ledger.press( + "mouse:left", lambda: events.append("down"), lambda: events.append("up"), lambda: False + ) + owner.end_call(lease) + owner._lease_until = time.monotonic() - 1 + + assert owner.status()["state"] == "unavailable" + assert owner._suppress is False + owner.status() # The emergency cleanup must retry the pending ledger hold. + assert owner.input_ledger.pending() == () + assert events == ["down", "up"] + + +def test_old_receipt_and_timer_cannot_release_new_same_button(setup_pointer): + owner, events, timers, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2])) + old_receipt = gesture._receipt + old_timer = timers[0] + owner.input_ledger.release_all() # Physical takeover owns this cleanup. + assert gesture.held_button is None + + _gesture_call(owner, lambda: gesture.down([3, 4])) + assert not owner.input_ledger.release("mouse:left", owner="Pointer", receipt=old_receipt) + old_timer.fire() + assert gesture.held_button == "left" + assert events == ["down", "up", "down"] + _gesture_call(owner, gesture.cancel) + assert events == ["down", "up", "down", "up"] + + +def test_takeover_during_timed_move_stops_before_next_injection(setup_pointer): + owner, events, _, make_timer = setup_pointer + + def takeover(_seconds: float) -> None: + owner._fast_takeover = True + owner.input_ledger.release_all() + + gesture = PointerController(control=owner, timer_factory=make_timer, sleeper=takeover) + _gesture_call(owner, lambda: gesture.down([1, 2])) + with pytest.raises(ControlBlocked, match="CONTROL_PREEMPTED"): + _gesture_call(owner, lambda: gesture.move([30, 40], duration=0.1)) + assert events == ["down", "up"] + assert owner.input_ledger.pending() == () + + +def test_move_and_takeover_barrier_finish_without_post_takeover_move(setup_pointer): + owner, events, _, make_timer = setup_pointer + sleeping = threading.Event() + resume = threading.Event() + errors: list[BaseException] = [] + + def pause(_seconds: float) -> None: + sleeping.set() + assert resume.wait(2) + + gesture = PointerController(control=owner, timer_factory=make_timer, sleeper=pause) + _gesture_call(owner, lambda: gesture.down([1, 2])) + + def worker() -> None: + try: + _gesture_call(owner, lambda: gesture.move([30, 40], duration=0.1)) + except BaseException as exc: + errors.append(exc) + + thread = threading.Thread(target=worker) + thread.start() + try: + assert sleeping.wait(2) + owner._fast_takeover = True + owner.input_ledger.release_all() + finally: + resume.set() + thread.join(2) + assert not thread.is_alive() + assert len(errors) == 1 and isinstance(errors[0], ControlBlocked) + assert events == ["down", "up"] + + +def test_timeout_can_release_during_timed_move_sleep(setup_pointer): + owner, events, timers, make_timer = setup_pointer + sleeping = threading.Event() + resume = threading.Event() + errors: list[BaseException] = [] + + def pause(_seconds: float) -> None: + sleeping.set() + assert resume.wait(2) + + gesture = PointerController(control=owner, timer_factory=make_timer, sleeper=pause) + _gesture_call(owner, lambda: gesture.down([1, 2])) + + def moving() -> None: + try: + _gesture_call(owner, lambda: gesture.move([30, 40], duration=0.1)) + except BaseException as exc: + errors.append(exc) + + move_thread = threading.Thread(target=moving) + timer_thread = threading.Thread(target=timers[0].fire) + move_thread.start() + try: + assert sleeping.wait(2) + timer_thread.start() + timer_thread.join(1) + assert not timer_thread.is_alive(), "timeout release waited for the move sleeper" + finally: + resume.set() + move_thread.join(2) + if timer_thread.ident is not None: + timer_thread.join(2) + assert not move_thread.is_alive() + assert not timer_thread.is_alive() + assert len(errors) == 1 and isinstance(errors[0], RuntimeError) + assert events == ["down", "up"] + + +def test_timer_and_next_call_finish_without_deadlock(setup_pointer, monkeypatch): + owner, events, timers, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2])) + releasing = threading.Event() + resume = threading.Event() + + def slow_up(*args, **kwargs) -> None: + releasing.set() + assert resume.wait(2) + events.append("up") + + monkeypatch.setattr(pointer.uia, "ReleaseMouse", slow_up) + timer_thread = threading.Thread(target=timers[0].fire) + next_thread = threading.Thread( + target=lambda: _gesture_call(owner, lambda: gesture.down([3, 4])) + ) + timer_thread.start() + try: + assert releasing.wait(2) + next_thread.start() + finally: + resume.set() + timer_thread.join(2) + next_thread.join(2) + assert not timer_thread.is_alive() and not next_thread.is_alive() + assert gesture.held_button == "left" + assert events == ["down", "up", "down"] + _gesture_call(owner, gesture.cancel) + + +def test_shutdown_retries_failed_pointer_release(setup_pointer, monkeypatch): + owner, events, _, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2])) + attempts = 0 + + def flaky_up(*args, **kwargs) -> None: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise OSError("first release failed") + events.append("up") + + monkeypatch.setattr(pointer.uia, "ReleaseMouse", flaky_up) + with pytest.raises(OSError, match="first release failed"): + gesture.close() + assert owner.input_ledger.pending() == ("mouse:left",) + owner.stop() + assert owner.input_ledger.pending() == () + assert attempts == 2 + assert owner._thread is None and owner._watchdog is None + + +def test_failed_retry_schedule_releases_physical_interception(setup_pointer, monkeypatch): + owner, _, timers, make_timer = setup_pointer + created = 0 + + def first_timer_only(interval: float, callback: Callable[[], None]) -> FakeTimer: + nonlocal created + created += 1 + if created > 1: + raise OSError("timer unavailable") + return make_timer(interval, callback) + + gesture = PointerController(control=owner, timer_factory=first_timer_only) + _gesture_call(owner, lambda: gesture.down([1, 2])) + monkeypatch.setattr( + pointer.uia, + "ReleaseMouse", + lambda *a, **k: (_ for _ in ()).throw(OSError("release failed")), + ) + + timers[0].fire() + assert owner._emergency is True + assert owner._suppress is False + assert owner.input_ledger.pending_entries() == (("mouse:left", "Pointer"),) + + +def test_shutdown_and_timer_finish_with_one_release(setup_pointer, monkeypatch): + owner, events, timers, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2])) + releasing = threading.Event() + resume = threading.Event() + + def slow_up(*args, **kwargs) -> None: + releasing.set() + assert resume.wait(2) + events.append("up") + + monkeypatch.setattr(pointer.uia, "ReleaseMouse", slow_up) + timer_thread = threading.Thread(target=timers[0].fire) + shutdown_thread = threading.Thread(target=lambda: (gesture.close(), owner.stop())) + timer_thread.start() + try: + assert releasing.wait(2) + shutdown_thread.start() + finally: + resume.set() + timer_thread.join(2) + shutdown_thread.join(2) + assert not timer_thread.is_alive() and not shutdown_thread.is_alive() + assert owner.input_ledger.pending() == () + assert events == ["down", "up"] + + +def test_failed_shutdown_keeps_ledger_receipt_for_audit(setup_pointer, monkeypatch): + owner, _, _, make_timer = setup_pointer + gesture = PointerController(control=owner, timer_factory=make_timer) + _gesture_call(owner, lambda: gesture.down([1, 2])) + monkeypatch.setattr( + pointer.uia, "ReleaseMouse", lambda *a, **k: (_ for _ in ()).throw(OSError("up failed")) + ) + with pytest.raises(OSError, match="up failed"): + gesture.close() + with pytest.raises(OSError, match="up failed"): + owner.stop() + assert owner.input_ledger.pending_entries() == (("mouse:left", "Pointer"),) + assert owner._stop.is_set() + assert owner._thread is None and owner._watchdog is None diff --git a/tests/test_pointer_controller.py b/tests/test_pointer_controller.py new file mode 100644 index 00000000..d1d978d1 --- /dev/null +++ b/tests/test_pointer_controller.py @@ -0,0 +1,483 @@ +from collections.abc import Callable + +import pytest + +from windows_mcp.desktop import pointer +from windows_mcp.desktop.control import ControlCoordinator +from windows_mcp.desktop.pointer import PointerController +from windows_mcp.desktop.service import Desktop + + +class FakeTimer: + def __init__(self, interval: float, function: Callable[[], None]) -> None: + self.interval = interval + self.function = function + self.daemon = False + self.started = False + self.cancelled = False + + def start(self) -> None: + self.started = True + + def cancel(self) -> None: + self.cancelled = True + + def fire(self) -> None: + self.function() + + +class TimerFactory: + def __init__(self) -> None: + self.timers: list[FakeTimer] = [] + + def __call__(self, interval: float, function: Callable[[], None]) -> FakeTimer: + timer = FakeTimer(interval, function) + self.timers.append(timer) + return timer + + +class FailingTimerFactory: + def __call__(self, interval: float, function: Callable[[], None]) -> FakeTimer: + raise RuntimeError("timer setup failed") + + +@pytest.fixture(autouse=True) +def isolated_input_owner(monkeypatch: pytest.MonkeyPatch) -> None: + """A held gesture in one test must never leak into another test's ledger.""" + owner = ControlCoordinator() + monkeypatch.setattr(pointer, "get_controller", lambda: owner) + + +def _mouse_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[object, ...]]: + calls: list[tuple[object, ...]] = [] + + def record(name: str) -> Callable: + return lambda *args, **kwargs: calls.append((name, *args, kwargs)) + + monkeypatch.setattr(pointer.uia, "PressMouse", record("press-left")) + monkeypatch.setattr(pointer.uia, "RightPressMouse", record("press-right")) + monkeypatch.setattr(pointer.uia, "MiddlePressMouse", record("press-middle")) + monkeypatch.setattr(pointer.uia, "ReleaseMouse", record("release-left")) + monkeypatch.setattr(pointer.uia, "RightReleaseMouse", record("release-right")) + monkeypatch.setattr(pointer.uia, "MiddleReleaseMouse", record("release-middle")) + monkeypatch.setattr(pointer.uia, "GetCursorPos", lambda: (1, 2)) + monkeypatch.setattr(pointer.uia, "SetCursorPos", record("set-pos")) + return calls + + +@pytest.mark.parametrize("button", ["left", "right", "middle"]) +def test_pointer_down_and_up_use_matching_button( + button: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + + down_result = controller.down([10, 20], button=button, timeout="12.5") + up_result = controller.up(button) + + assert down_result == { + "action": "down", + "button": button, + "loc": [10, 20], + "timeout": 12.5, + } + assert up_result == {"action": "up", "button": button} + assert [call[0] for call in calls] == [f"press-{button}", f"release-{button}"] + assert timers.timers[0].interval == 12.5 + assert timers.timers[0].daemon is True + assert timers.timers[0].started is True + assert timers.timers[0].cancelled is True + assert controller.held_button is None + + +def test_pointer_rejects_second_down_without_extra_input( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + controller.down([1, 2]) + + with pytest.raises(RuntimeError, match="already held"): + controller.down([3, 4], button="right") + + assert [call[0] for call in calls] == ["press-left"] + assert controller.held_button == "left" + + +def test_pointer_move_supports_immediate_and_duration_paths( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory(), sleeper=lambda _: None) + controller.down([1, 2]) + + immediate = controller.move([3, 4]) + bounded = controller.move([5, 6], duration="0.25") + + assert immediate["duration"] is None + assert bounded["duration"] == 0.25 + assert [call[0] for call in calls] == ["press-left"] + ["set-pos"] * 14 + assert calls[-1][1:3] == (5, 6) + assert controller.held_button == "left" + + +def test_pointer_move_requires_held_button_before_input( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + + with pytest.raises(RuntimeError, match="no mouse button is held"): + controller.move([3, 4]) + + assert calls == [] + + +def test_pointer_up_rejects_button_mismatch_without_releasing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + controller.down([1, 2], button="right") + + with pytest.raises(RuntimeError, match="right mouse button is held"): + controller.up("left") + + assert [call[0] for call in calls] == ["press-right"] + assert controller.held_button == "right" + + +def test_pointer_cancel_is_repeatable_and_releases_only_owned_button( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + controller.down([1, 2], button="middle") + + first = controller.cancel() + second = controller.cancel() + + assert first == {"action": "cancel", "button": "middle"} + assert second == {"action": "cancel", "button": None} + assert [call[0] for call in calls] == ["press-middle", "release-middle"] + assert controller.held_button is None + + +def test_pointer_timeout_releases_only_owned_button_and_ignores_stale_timer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + controller.down([1, 2]) + first_timer = timers.timers[0] + controller.up() + controller.down([3, 4], button="right") + + first_timer.fire() + assert controller.held_button == "right" + assert [call[0] for call in calls] == ["press-left", "release-left", "press-right"] + + timers.timers[1].fire() + assert controller.held_button is None + assert [call[0] for call in calls][-1:] == ["release-right"] + + +def test_pointer_close_releases_only_owned_button( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + controller.down([1, 2], button="right") + + controller.close() + controller.close() + + assert [call[0] for call in calls] == ["press-right", "release-right"] + assert controller.held_button is None + + +def test_pointer_move_failure_releases_tracked_button( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + controller.down([1, 2]) + monkeypatch.setattr( + pointer.uia, + "SetCursorPos", + lambda *args, **kwargs: (_ for _ in ()).throw(OSError("move failed")), + ) + + with pytest.raises(OSError, match="move failed"): + controller.move([3, 4]) + + assert [call[0] for call in calls] == ["press-left", "release-left"] + assert controller.held_button is None + + +def test_pointer_timer_setup_failure_does_not_press_button( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=FailingTimerFactory()) + + with pytest.raises(RuntimeError, match="timer setup failed"): + controller.down([1, 2]) + + assert calls == [] + assert controller.held_button is None + + +def test_pointer_cancel_retries_after_release_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + controller.down([1, 2]) + release = pointer.uia.ReleaseMouse + attempts = 0 + + def flaky_release(*args: object, **kwargs: object) -> None: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise OSError("release failed") + release(*args, **kwargs) + + monkeypatch.setattr(pointer.uia, "ReleaseMouse", flaky_release) + + with pytest.raises(OSError, match="release failed"): + controller.cancel() + + assert controller.held_button == "left" + assert timers.timers[0].cancelled is False + assert [call[0] for call in calls] == ["press-left"] + + controller.cancel() + assert controller.held_button is None + assert timers.timers[0].cancelled is True + assert [call[0] for call in calls] == ["press-left", "release-left"] + + +def test_pointer_timeout_retries_after_cursor_lookup_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + controller.down([1, 2], button="right") + release = pointer.uia.RightReleaseMouse + + def release_after_cursor_lookup(*args: object, **kwargs: object) -> None: + pointer.uia.GetCursorPos() + release(*args, **kwargs) + + monkeypatch.setattr(pointer.uia, "RightReleaseMouse", release_after_cursor_lookup) + monkeypatch.setattr( + pointer.uia, + "GetCursorPos", + lambda: (_ for _ in ()).throw(OSError("GetCursorPos failed")), + ) + + timers.timers[0].fire() + assert controller.held_button == "right" + assert len(timers.timers) == 2 + assert timers.timers[1].interval == pointer.RELEASE_RETRY_DELAY + assert [call[0] for call in calls] == ["press-right"] + + monkeypatch.setattr(pointer.uia, "GetCursorPos", lambda: (1, 2)) + timers.timers[1].fire() + assert controller.held_button is None + assert [call[0] for call in calls] == ["press-right", "release-right"] + + +def test_pointer_timeout_exhaustion_keeps_ownership_for_manual_cancel( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + controller.down([1, 2]) + release = pointer.uia.ReleaseMouse + monkeypatch.setattr( + pointer.uia, + "ReleaseMouse", + lambda *args, **kwargs: (_ for _ in ()).throw(OSError("release failed")), + ) + + for attempt in range(pointer.MAX_RELEASE_ATTEMPTS): + timers.timers[attempt].fire() + + assert len(timers.timers) == pointer.MAX_RELEASE_ATTEMPTS + assert controller.held_button == "left" + assert [call[0] for call in calls] == ["press-left"] + + monkeypatch.setattr(pointer.uia, "ReleaseMouse", release) + controller.cancel() + assert controller.held_button is None + assert [call[0] for call in calls] == ["press-left", "release-left"] + + +def test_pointer_press_and_compensation_failure_retains_retriable_ownership( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + release = pointer.uia.ReleaseMouse + monkeypatch.setattr( + pointer.uia, + "PressMouse", + lambda *args, **kwargs: (_ for _ in ()).throw(OSError("press failed")), + ) + monkeypatch.setattr( + pointer.uia, + "ReleaseMouse", + lambda *args, **kwargs: (_ for _ in ()).throw(OSError("release failed")), + ) + + with pytest.raises(OSError, match="press failed"): + controller.down([1, 2]) + + # The ledger reports the original press error and retains the failed + # compensation for the timer/watchdog to retry. + assert controller.held_button == "left" + assert timers.timers[0].started is True + assert timers.timers[0].cancelled is False + + monkeypatch.setattr(pointer.uia, "ReleaseMouse", release) + timers.timers[0].fire() + assert controller.held_button is None + assert [call[0] for call in calls] == ["release-left"] + + +def test_pointer_press_failure_with_successful_compensation_clears_ownership( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + timers = TimerFactory() + controller = PointerController(timer_factory=timers) + monkeypatch.setattr( + pointer.uia, + "PressMouse", + lambda *args, **kwargs: (_ for _ in ()).throw(OSError("press failed")), + ) + + with pytest.raises(OSError, match="press failed"): + controller.down([1, 2]) + + assert [call[0] for call in calls] == ["release-left"] + assert controller.held_button is None + assert timers.timers[0].cancelled is True + + +@pytest.mark.parametrize( + ("method", "value", "message"), + [ + ("down", [1], "loc"), + ("down", [1, True], "integers"), + ("down", [1.0, 2], "integers"), + ("timeout", True, "finite"), + ("timeout", 0, "greater than 0"), + ("timeout", 121, "at most 120"), + ("duration", False, "finite"), + ("duration", -1, "between 0 and 10"), + ("duration", 11, "between 0 and 10"), + ], +) +def test_pointer_rejects_invalid_values_before_input( + method: str, + value: object, + message: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = _mouse_calls(monkeypatch) + controller = PointerController(timer_factory=TimerFactory()) + + with pytest.raises(ValueError, match=message): + if method == "down": + controller.down(value) + elif method == "timeout": + controller.down([1, 2], timeout=value) + else: + controller.down([1, 2]) + calls.clear() + controller.move([3, 4], duration=value) + + assert calls == [] + + +class FakePointerController: + def __init__(self) -> None: + self.calls: list[tuple[object, ...]] = [] + + def down(self, *args: object) -> dict[str, object]: + self.calls.append(("down", *args)) + return {"action": "down"} + + def move(self, *args: object) -> dict[str, object]: + self.calls.append(("move", *args)) + return {"action": "move"} + + def up(self, *args: object) -> dict[str, object]: + self.calls.append(("up", *args)) + return {"action": "up"} + + def cancel(self) -> dict[str, object]: + self.calls.append(("cancel",)) + return {"action": "cancel"} + + def close(self) -> None: + self.calls.append(("close",)) + + +def test_desktop_delegates_pointer_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + desktop = Desktop.__new__(Desktop) + controller = FakePointerController() + desktop._pointer = controller + validated: list[tuple[int, int]] = [] + monkeypatch.setattr( + Desktop, + "_validate_screen_point", + staticmethod(lambda x, y: validated.append((x, y))), + ) + + desktop.pointer_down([1, 2], "right", 10) + desktop.pointer_move([3, 4], 0.2) + desktop.pointer_up("right") + desktop.pointer_cancel() + desktop.close() + + assert controller.calls == [ + ("down", [1, 2], "right", 10), + ("move", [3, 4], 0.2), + ("up", "right"), + ("cancel",), + ("close",), + ] + assert validated == [(1, 2), (3, 4)] + + +@pytest.mark.parametrize("action", ["down", "move"]) +@pytest.mark.parametrize("point", [[-1, 2], [100, 2], [1, 100]]) +def test_desktop_rejects_out_of_bounds_pointer_before_controller( + action: str, + point: list[int], + monkeypatch: pytest.MonkeyPatch, +) -> None: + desktop = Desktop.__new__(Desktop) + controller = FakePointerController() + desktop._pointer = controller + monkeypatch.setattr(pointer.uia, "GetVirtualScreenRect", lambda: (0, 0, 100, 100)) + + with pytest.raises(ValueError, match="outside the desktop bounds"): + if action == "down": + desktop.pointer_down(point) + else: + desktop.pointer_move(point) + + assert controller.calls == [] diff --git a/tests/test_pointer_lifespan.py b/tests/test_pointer_lifespan.py new file mode 100644 index 00000000..726bda9d --- /dev/null +++ b/tests/test_pointer_lifespan.py @@ -0,0 +1,119 @@ +"""The server must release stateful pointer input without skipping other cleanup.""" + +import asyncio + +import pytest + +from windows_mcp import __main__ as server +from windows_mcp import infrastructure, tools +from windows_mcp.desktop import control, control_overlay, service +from windows_mcp.tools import control_notifications + + +@pytest.mark.parametrize("close_fails", [False, True]) +@pytest.mark.parametrize("notifier_close_fails", [False, True]) +def test_lifespan_closes_pointer_before_watchdog_and_analytics( + close_fails: bool, + notifier_close_fails: bool, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + events: list[str] = [] + + class FakeDesktop: + def get_screen_size(self) -> tuple[int, int]: + return (100, 100) + + def close(self) -> None: + events.append("desktop.close") + if close_fails: + raise OSError("release failed") + + class FakeWatchdog: + def stop(self) -> None: + events.append("watchdog.stop") + + class FakeAnalytics: + async def close(self) -> None: + events.append("analytics.close") + + class FakeMCP: + def __init__(self, **kwargs: object) -> None: + self.lifespan = kwargs["lifespan"] + + def add_middleware(self, middleware: object) -> None: + pass + + class FakeControl: + mouse_takeover_units = 120 + mouse_takeover_pixels = 120 + _thread = None + + def subscribe(self, callback: object) -> None: + pass + + def set_health_probe(self, callback: object) -> None: + pass + + def start(self) -> None: + events.append("controller.start") + + def stop(self) -> None: + events.append("controller.stop") + + class FakeNotifier: + def __init__(self, controller: object) -> None: + pass + + def start(self) -> None: + events.append("notifier.start") + + async def close(self) -> None: + events.append("notifier.close") + if notifier_close_fails: + raise OSError("notifier close failed") + + desktop = FakeDesktop() + watchdog = FakeWatchdog() + analytics = FakeAnalytics() + monkeypatch.setenv("ANONYMIZED_TELEMETRY", "true") + monkeypatch.setattr(server, "_mcp", None) + monkeypatch.setattr(server, "desktop", None) + monkeypatch.setattr(server, "watchdog", None) + monkeypatch.setattr(server, "analytics", None) + monkeypatch.setattr(server, "FastMCP", FakeMCP) + monkeypatch.setattr(server, "_start_watchdog", lambda _: watchdog) + monkeypatch.setattr(infrastructure, "PostHogAnalytics", lambda: analytics) + monkeypatch.setattr(service, "Desktop", lambda: desktop) + monkeypatch.setattr(tools, "register_all", lambda *args, **kwargs: None) + monkeypatch.setattr(control, "get_controller", lambda: FakeControl()) + monkeypatch.setattr(control_notifications, "ControlNotifier", FakeNotifier) + monkeypatch.setattr(control_overlay, "start", lambda: events.append("overlay.start")) + monkeypatch.setattr(control_overlay, "stop", lambda: events.append("overlay.stop")) + monkeypatch.setattr(control_overlay, "is_healthy", lambda: True) + + mcp = server._build_mcp() + + async def run_lifespan() -> None: + async with mcp.lifespan(mcp): + pass + + if notifier_close_fails: + with pytest.raises(OSError, match="notifier close failed"): + asyncio.run(run_lifespan()) + else: + asyncio.run(run_lifespan()) + + assert events == [ + "overlay.start", + "controller.start", + "notifier.start", + "notifier.close", + "desktop.close", + "controller.stop", + "overlay.stop", + "watchdog.stop", + "analytics.close", + ] + if close_fails: + assert "Failed to release desktop input during shutdown" in caplog.text diff --git a/tests/test_pointer_tool.py b/tests/test_pointer_tool.py new file mode 100644 index 00000000..eb22a535 --- /dev/null +++ b/tests/test_pointer_tool.py @@ -0,0 +1,130 @@ +import asyncio +from collections.abc import Callable + +import pytest + +from windows_mcp.tools.pointer import register + + +class FakeMCP: + def __init__(self) -> None: + self.tools: dict[str, Callable] = {} + self.options: dict[str, dict[str, object]] = {} + + def tool(self, *, name: str, **kwargs: object) -> Callable: + self.options[name] = kwargs + + def decorator(func: Callable) -> Callable: + self.tools[name] = func + return func + + return decorator + + +class FakeDesktop: + def __init__(self) -> None: + self.calls: list[tuple[object, ...]] = [] + + def pointer_down(self, loc, button, timeout) -> dict[str, object]: + self.calls.append(("down", loc, button, timeout)) + return {"action": "down", "loc": list(loc), "button": button, "timeout": timeout} + + def pointer_move(self, loc, duration) -> dict[str, object]: + self.calls.append(("move", loc, duration)) + return {"action": "move", "loc": list(loc), "button": "left", "duration": duration} + + def pointer_up(self, button) -> dict[str, object]: + self.calls.append(("up", button)) + return {"action": "up", "button": button or "left"} + + def pointer_cancel(self) -> dict[str, object]: + self.calls.append(("cancel",)) + return {"action": "cancel", "button": None} + + +def _registered(desktop: FakeDesktop) -> tuple[FakeMCP, Callable]: + mcp = FakeMCP() + register(mcp, get_desktop=lambda: desktop, get_analytics=lambda: None) + return mcp, mcp.tools["Pointer"] + + +def test_pointer_registers_one_consolidated_tool() -> None: + mcp, _ = _registered(FakeDesktop()) + + assert set(mcp.tools) == {"Pointer"} + assert mcp.options["Pointer"]["annotations"].destructiveHint is True + assert mcp.options["Pointer"]["annotations"].idempotentHint is False + + +def test_pointer_down_accepts_json_loc_and_defaults() -> None: + desktop = FakeDesktop() + _, tool = _registered(desktop) + + result = asyncio.run(tool(action="down", loc="[10, 20]")) + + assert result == ( + "Pressed left mouse button at (10,20); automatic release timeout is 30 seconds." + ) + assert desktop.calls == [("down", (10, 20), "left", 30.0)] + + +def test_pointer_move_accepts_bounded_duration() -> None: + desktop = FakeDesktop() + _, tool = _registered(desktop) + + result = asyncio.run(tool(action=" MOVE ", loc=[30, 40], duration="0.25")) + + assert result == "Moved held left mouse button to (30,40) over 0.250 seconds." + assert desktop.calls == [("move", (30, 40), 0.25)] + + +def test_pointer_up_asserts_optional_button() -> None: + desktop = FakeDesktop() + _, tool = _registered(desktop) + + result = asyncio.run(tool(action="up", button="right")) + + assert result == "Released right mouse button." + assert desktop.calls == [("up", "right")] + + +def test_pointer_cancel_has_no_extra_arguments() -> None: + desktop = FakeDesktop() + _, tool = _registered(desktop) + + result = asyncio.run(tool(action="cancel")) + + assert result == "No tracked mouse button was held." + assert desktop.calls == [("cancel",)] + + +@pytest.mark.parametrize( + ("kwargs", "message"), + [ + ({"action": "invalid"}, "action must be one of"), + ({"action": "down"}, "loc is required"), + ({"action": "down", "loc": [1, 2], "duration": 1}, "only supported.*move"), + ({"action": "move", "loc": [1, 2], "button": "left"}, "only supported.*down"), + ({"action": "move", "loc": [1, 2], "timeout": 1}, "only supported.*down"), + ({"action": "up", "loc": [1, 2]}, "not supported.*up"), + ({"action": "up", "duration": 1}, "not supported.*up"), + ({"action": "cancel", "button": "left"}, "not supported.*cancel"), + ({"action": "down", "loc": "not json"}, "JSON list"), + ({"action": "down", "loc": [1, True]}, "integers"), + ({"action": "down", "loc": [1, 2], "button": ""}, "button must be"), + ({"action": "down", "loc": [1, 2], "button": "primary"}, "button must be"), + ({"action": "down", "loc": [1, 2], "timeout": True}, "finite"), + ({"action": "move", "loc": [1, 2], "duration": "nan"}, "finite"), + ], +) +def test_pointer_rejects_invalid_combinations_before_desktop( + kwargs: dict[str, object], + message: str, +) -> None: + desktop = FakeDesktop() + _, tool = _registered(desktop) + + with pytest.raises(ValueError, match=message): + asyncio.run(tool(**kwargs)) + + assert desktop.calls == [] diff --git a/tests/test_stdio_handshake.py b/tests/test_stdio_handshake.py index 847ea7a7..0f01f37b 100644 --- a/tests/test_stdio_handshake.py +++ b/tests/test_stdio_handshake.py @@ -54,6 +54,7 @@ "MultiEdit", "MultiSelect", "Notification", + "Pointer", "PowerShell", "Process", "Registry",