From 3bb892a0ad6d5832e12bb2414f0be657a22ea335 Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Mon, 13 Apr 2026 18:26:01 +0530 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20add=20Windows=20VDD=20confinement?= =?UTF-8?q?=20=E2=80=94=20isolate=20agent=20to=20virtual=20monitor=20[#29]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds VirtualDisplayManager (Windows-only) that creates a dedicated virtual monitor for the agent using the Parsec Virtual Display Driver, providing spatial isolation at the display level. All ctypes/Win32 calls are guarded so the module imports cleanly on macOS and Linux. 28 tests, all passing. --- .../computer/windows/virtual_display.py | 416 ++++++++++++++++++ tests/test_virtual_display.py | 412 +++++++++++++++++ 2 files changed, 828 insertions(+) create mode 100644 operator_use/computer/windows/virtual_display.py create mode 100644 tests/test_virtual_display.py diff --git a/operator_use/computer/windows/virtual_display.py b/operator_use/computer/windows/virtual_display.py new file mode 100644 index 0000000..105f5ab --- /dev/null +++ b/operator_use/computer/windows/virtual_display.py @@ -0,0 +1,416 @@ +"""Windows Virtual Display Manager — Parsec VDD confinement layer. + +Creates a dedicated virtual monitor for the agent using the Parsec Virtual +Display Driver (VDD), providing spatial isolation at the display level. +Agent windows live on a separate monitor the user never sees. + +Requires Parsec Virtual Display Driver: https://github.com/nomi-san/parsec-vdd + +Installation: + Download and install ParsecVDA from the repository above, or via the + Parsec desktop application (Settings → Displays → Enable Virtual Display). + +Usage:: + + mgr = VirtualDisplayManager() + if mgr.is_available(): + mgr.create_virtual_display(width=1920, height=1080, refresh_rate=60) + mgr.move_window_to_virtual_display("Notepad") + # ... agent does work ... + mgr.remove_virtual_display() + +All public methods return False / None on failure rather than raising, so the +caller can treat VDD as an optional enhancement without defensive try/except. +""" + +from __future__ import annotations + +import logging +import subprocess +import sys +from typing import Any + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Win32 constants (kept at module level so tests can patch them easily) +# --------------------------------------------------------------------------- + +SWP_NOSIZE = 0x0001 +SWP_NOZORDER = 0x0004 +SWP_SHOWWINDOW = 0x0040 + +MONITOR_DEFAULTTONEAREST = 0x00000002 + +# --------------------------------------------------------------------------- +# Lazy ctypes imports — succeed on any platform; actual Win32 calls are +# guarded behind sys.platform checks at call time. +# --------------------------------------------------------------------------- + +_user32: Any = None +_kernel32: Any = None +_ctypes_loaded = False + + +def _load_win32() -> bool: + """Attempt to load user32 / kernel32 via ctypes. Returns True on success.""" + global _user32, _kernel32, _ctypes_loaded + if _ctypes_loaded: + return _user32 is not None + _ctypes_loaded = True + if sys.platform != "win32": + return False + try: + import ctypes + + _user32 = ctypes.windll.user32 # type: ignore[attr-defined] + _kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + return True + except Exception as exc: + logger.debug("ctypes win32 load failed: %s", exc) + return False + + +# --------------------------------------------------------------------------- +# Registry helper — thin wrapper so tests can mock winreg cleanly +# --------------------------------------------------------------------------- + + +def _registry_key_exists(hive: int | None, subkey: str) -> bool: + """Return True if *subkey* exists under *hive* in the registry. + + Returns False on ImportError (non-Windows), OSError (key missing), or if + *hive* is None. + """ + if hive is None: + return False + try: + import winreg # type: ignore[import] + + handle = winreg.OpenKey(hive, subkey) + winreg.CloseKey(handle) + return True + except (ImportError, OSError, FileNotFoundError): + return False + + +# --------------------------------------------------------------------------- +# VirtualDisplayManager +# --------------------------------------------------------------------------- + + +class VirtualDisplayManager: + """Windows-only plugin that manages a Parsec Virtual Display Driver monitor. + + All methods are safe to call on non-Windows platforms and when VDD is not + installed — they return ``False`` or ``None`` without raising exceptions. + """ + + # Registry path written by the Parsec VDA installer + _VDD_REGISTRY_KEY = r"SOFTWARE\Parsec\vdd" + # Driver INF file name (alternative detection path) + _VDD_INF_NAME = "ParsecVDA.inf" + # Executable name for subprocess fallback + _VDD_EXE = "ParsecVDA.exe" + + def __init__(self) -> None: + self._virtual_monitor_index: int | None = None + self._virtual_monitor_handle: int | None = None + + # ------------------------------------------------------------------ + # Detection + # ------------------------------------------------------------------ + + def is_available(self) -> bool: + """Return True only on win32 with the Parsec VDD installed.""" + if sys.platform != "win32": + return False + return self.is_vdd_installed() + + def is_vdd_installed(self) -> bool: + """Check whether the Parsec VDD is present on this machine. + + Tries two detection strategies in order: + 1. Registry key ``HKLM\\SOFTWARE\\Parsec\\vdd`` + 2. Presence of ``ParsecVDA.inf`` in the Windows driver store + + Returns True if either check succeeds. + """ + try: + import winreg # type: ignore[import] + + hklm: int | None = winreg.HKEY_LOCAL_MACHINE + except ImportError: + # Not on Windows at all; _registry_key_exists handles None gracefully. + hklm = None + + # Strategy 1 — registry + if _registry_key_exists(hklm, self._VDD_REGISTRY_KEY): + logger.debug("Parsec VDD detected via registry key") + return True + + # Strategy 2 — driver store + try: + result = subprocess.run( + ["pnputil", "/enum-drivers", "/class", "Display"], + capture_output=True, + text=True, + timeout=10, + ) + if self._VDD_INF_NAME.lower() in result.stdout.lower(): + logger.debug("Parsec VDD detected via driver store (%s)", self._VDD_INF_NAME) + return True + except Exception as exc: + logger.debug("pnputil driver check failed: %s", exc) + + return False + + # ------------------------------------------------------------------ + # Display lifecycle + # ------------------------------------------------------------------ + + def create_virtual_display( + self, + width: int = 1920, + height: int = 1080, + refresh_rate: int = 60, + ) -> bool: + """Add a virtual monitor via the Parsec VDA API. + + Tries ``ctypes.windll.ParsecVDA`` first; falls back to launching + ``ParsecVDA.exe add`` as a subprocess. Returns True on success. + """ + if sys.platform != "win32": + logger.debug("create_virtual_display: not on win32, skipping") + return False + + # Attempt ctypes direct call first + if self._create_via_ctypes(width, height, refresh_rate): + return True + + # Subprocess fallback + return self._create_via_subprocess(width, height, refresh_rate) + + def _create_via_ctypes(self, width: int, height: int, refresh_rate: int) -> bool: + """Try to call ParsecVDA.dll directly via ctypes.""" + try: + import ctypes + + vda = ctypes.windll.ParsecVDA # type: ignore[attr-defined] + # ParsecVDAAddDisplay(width, height, hz) — unofficial API + result = vda.ParsecVDAAddDisplay( + ctypes.c_int(width), + ctypes.c_int(height), + ctypes.c_int(refresh_rate), + ) + if result == 0: + logger.info( + "Virtual display created via ctypes (%dx%d@%dHz)", width, height, refresh_rate + ) + self._virtual_monitor_index = 0 + return True + logger.debug("ParsecVDAAddDisplay returned %s", result) + except (AttributeError, OSError, Exception) as exc: + logger.debug("ctypes VDA call failed: %s", exc) + return False + + def _create_via_subprocess(self, width: int, height: int, refresh_rate: int) -> bool: + """Launch ParsecVDA.exe as a subprocess fallback.""" + try: + result = subprocess.run( + [ + self._VDD_EXE, + "add", + f"--width={width}", + f"--height={height}", + f"--hz={refresh_rate}", + ], + capture_output=True, + text=True, + timeout=15, + ) + if result.returncode == 0: + logger.info( + "Virtual display created via subprocess (%dx%d@%dHz)", + width, + height, + refresh_rate, + ) + self._virtual_monitor_index = 0 + return True + logger.debug("ParsecVDA.exe add returned %d: %s", result.returncode, result.stderr) + except (FileNotFoundError, subprocess.TimeoutExpired, Exception) as exc: + logger.debug("Subprocess VDA call failed: %s", exc) + return False + + def remove_virtual_display(self) -> bool: + """Remove the virtual display created by this manager instance. + + Returns True on success, False if removal failed or no display was + previously created. + """ + if sys.platform != "win32": + return False + if self._virtual_monitor_index is None: + logger.debug("remove_virtual_display: no virtual display to remove") + return False + + removed = self._remove_via_ctypes() or self._remove_via_subprocess() + if removed: + self._virtual_monitor_index = None + self._virtual_monitor_handle = None + return removed + + def _remove_via_ctypes(self) -> bool: + try: + import ctypes + + vda = ctypes.windll.ParsecVDA # type: ignore[attr-defined] + result = vda.ParsecVDARemoveDisplay(ctypes.c_int(self._virtual_monitor_index or 0)) + if result == 0: + logger.info("Virtual display removed via ctypes") + return True + logger.debug("ParsecVDARemoveDisplay returned %s", result) + except (AttributeError, OSError, Exception) as exc: + logger.debug("ctypes VDA remove failed: %s", exc) + return False + + def _remove_via_subprocess(self) -> bool: + try: + result = subprocess.run( + [self._VDD_EXE, "remove"], + capture_output=True, + text=True, + timeout=15, + ) + if result.returncode == 0: + logger.info("Virtual display removed via subprocess") + return True + logger.debug("ParsecVDA.exe remove returned %d: %s", result.returncode, result.stderr) + except (FileNotFoundError, subprocess.TimeoutExpired, Exception) as exc: + logger.debug("Subprocess VDA remove failed: %s", exc) + return False + + # ------------------------------------------------------------------ + # Window placement + # ------------------------------------------------------------------ + + def move_window_to_virtual_display(self, window_title: str) -> bool: + """Move a window to the virtual monitor. + + Uses ``EnumWindows`` to find the target HWND by title, then + ``EnumDisplayMonitors`` to locate the virtual monitor handle, and + finally ``SetWindowPos`` to move the window. + + Returns True if the window was successfully moved. + """ + if sys.platform != "win32": + return False + if not _load_win32(): + return False + + hwnd = self._find_window(window_title) + if hwnd is None: + logger.debug("move_window_to_virtual_display: window %r not found", window_title) + return False + + monitor = self._get_virtual_monitor() + if monitor is None: + logger.debug("move_window_to_virtual_display: virtual monitor not found") + return False + + return self._set_window_pos(hwnd, monitor) + + def _find_window(self, title: str) -> int | None: + """Return the HWND of the first visible top-level window whose title + contains *title* (case-insensitive). Returns None if not found.""" + if _user32 is None: + return None + try: + import ctypes + from ctypes.wintypes import HWND, LPARAM + + found: list[int] = [] + title_lower = title.lower() + + EnumWindowsProc = ctypes.WINFUNCTYPE( # type: ignore[attr-defined] + ctypes.c_bool, HWND, LPARAM + ) + + def _callback(hwnd: int, _lparam: int) -> bool: + buf = ctypes.create_unicode_buffer(512) + _user32.GetWindowTextW(hwnd, buf, 512) + if title_lower in buf.value.lower(): + found.append(hwnd) + return False # stop enumeration + return True + + _user32.EnumWindows(EnumWindowsProc(_callback), 0) + return found[0] if found else None + except Exception as exc: + logger.debug("_find_window failed: %s", exc) + return None + + def _get_virtual_monitor(self) -> Any | None: + """Return the HMONITOR for the virtual display. + + Uses ``EnumDisplayMonitors`` and picks the last monitor (heuristic: + the virtual display is typically the last one enumerated after the + physical displays). + """ + if _user32 is None: + return None + try: + import ctypes + from ctypes.wintypes import HDC, LPRECT + + monitors: list[Any] = [] + + MonitorEnumProc = ctypes.WINFUNCTYPE( # type: ignore[attr-defined] + ctypes.c_bool, + ctypes.c_ulong, + HDC, + LPRECT, + ctypes.c_double, + ) + + def _mon_callback(hmon: Any, _hdc: Any, _rect: Any, _data: Any) -> bool: + monitors.append(hmon) + return True + + _user32.EnumDisplayMonitors(None, None, MonitorEnumProc(_mon_callback), 0) + + # Virtual display is the last monitor; need at least 2 (physical + virtual) + if len(monitors) >= 2: + self._virtual_monitor_handle = monitors[-1] + return monitors[-1] + logger.debug("_get_virtual_monitor: only %d monitor(s) found", len(monitors)) + except Exception as exc: + logger.debug("_get_virtual_monitor failed: %s", exc) + return None + + def _set_window_pos(self, hwnd: int, monitor: Any) -> bool: + """Move *hwnd* onto *monitor* using ``SetWindowPos``.""" + if _user32 is None: + return False + try: + import ctypes + + # MONITORINFO layout: cbSize(4) + rcMonitor(16) + rcWork(16) + dwFlags(4) + # We pack it as an array of 10 LONGs for simplicity. + moninfo_array = (ctypes.c_long * 10)() + moninfo_array[0] = 40 # cbSize + _user32.GetMonitorInfoW(monitor, ctypes.byref(moninfo_array)) + # rcMonitor is at offset 4 (4 LONGs = left, top, right, bottom) + left = moninfo_array[1] + top = moninfo_array[2] + + flags = SWP_NOSIZE | SWP_NOZORDER | SWP_SHOWWINDOW + ret = _user32.SetWindowPos(hwnd, None, left, top, 0, 0, flags) + if ret: + logger.info("Window HWND=%d moved to virtual monitor (%d, %d)", hwnd, left, top) + return True + logger.debug("SetWindowPos returned 0 for HWND=%d", hwnd) + except Exception as exc: + logger.debug("_set_window_pos failed: %s", exc) + return False diff --git a/tests/test_virtual_display.py b/tests/test_virtual_display.py new file mode 100644 index 0000000..8de6998 --- /dev/null +++ b/tests/test_virtual_display.py @@ -0,0 +1,412 @@ +"""Tests for VirtualDisplayManager — Windows VDD confinement layer. + +All Win32 / ctypes calls are mocked so the suite runs on macOS and Linux. +""" + +from __future__ import annotations + +import subprocess +import sys +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_mgr(): + """Return a fresh VirtualDisplayManager with internal state reset.""" + import operator_use.computer.windows.virtual_display as vd_mod + + # Reset cached ctypes state between tests + vd_mod._user32 = None + vd_mod._kernel32 = None + vd_mod._ctypes_loaded = False + + from operator_use.computer.windows.virtual_display import VirtualDisplayManager + + return VirtualDisplayManager() + + +# --------------------------------------------------------------------------- +# 1. is_available() — platform gate +# --------------------------------------------------------------------------- + + +def test_is_available_false_on_non_windows(): + mgr = _make_mgr() + with patch.object(sys, "platform", "darwin"): + assert mgr.is_available() is False + + +def test_is_available_false_on_linux(): + mgr = _make_mgr() + with patch.object(sys, "platform", "linux"): + assert mgr.is_available() is False + + +def test_is_available_true_when_win32_and_vdd_installed(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch.object(mgr, "is_vdd_installed", return_value=True), + ): + assert mgr.is_available() is True + + +def test_is_available_false_when_win32_but_vdd_missing(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch.object(mgr, "is_vdd_installed", return_value=False), + ): + assert mgr.is_available() is False + + +# --------------------------------------------------------------------------- +# 2. is_vdd_installed() — registry + driver store checks +# --------------------------------------------------------------------------- + + +def test_is_vdd_installed_true_via_registry(): + mgr = _make_mgr() + with patch.object(sys, "platform", "win32"): + with patch( + "operator_use.computer.windows.virtual_display._registry_key_exists", + return_value=True, + ): + assert mgr.is_vdd_installed() is True + + +def test_is_vdd_installed_false_when_winreg_missing(): + """On non-Windows, winreg is absent so is_vdd_installed returns False.""" + mgr = _make_mgr() + # We're already on macOS/Linux — winreg import will fail naturally. + # Registry check returns False; subprocess finds no VDA INF either. + with patch( + "operator_use.computer.windows.virtual_display._registry_key_exists", + return_value=False, + ), patch( + "subprocess.run", + return_value=MagicMock(returncode=0, stdout="SomeOtherDriver.inf\n", stderr=""), + ): + result = mgr.is_vdd_installed() + # Should return False, not raise + assert result is False + + +def test_is_vdd_installed_fallback_to_driver_store(): + """When registry check fails, falls back to pnputil and finds VDA INF.""" + mgr = _make_mgr() + with ( + patch( + "operator_use.computer.windows.virtual_display._registry_key_exists", + return_value=False, + ), + patch( + "subprocess.run", + return_value=MagicMock( + returncode=0, stdout="ParsecVDA.inf\nSome other driver\n", stderr="" + ), + ), + ): + assert mgr.is_vdd_installed() is True + + +def test_is_vdd_installed_false_when_neither_check_succeeds(): + mgr = _make_mgr() + with patch.object(sys, "platform", "win32"): + with ( + patch( + "operator_use.computer.windows.virtual_display._registry_key_exists", + return_value=False, + ), + patch( + "subprocess.run", + return_value=MagicMock(returncode=0, stdout="SomeOtherDriver.inf\n", stderr=""), + ), + ): + assert mgr.is_vdd_installed() is False + + +# --------------------------------------------------------------------------- +# 3. create_virtual_display() — ctypes path +# --------------------------------------------------------------------------- + + +def test_create_virtual_display_calls_vda_with_correct_params(): + mgr = _make_mgr() + mock_vda = MagicMock() + mock_vda.ParsecVDAAddDisplay.return_value = 0 # success + + with patch.object(sys, "platform", "win32"): + import ctypes as _ctypes + + with patch.object(_ctypes, "windll", create=True) as mock_windll: + mock_windll.ParsecVDA = mock_vda + + result = mgr._create_via_ctypes(1920, 1080, 60) + + assert result is True + mock_vda.ParsecVDAAddDisplay.assert_called_once() + # Verify correct dimension args were used (ctypes wraps them, so inspect call) + args = mock_vda.ParsecVDAAddDisplay.call_args + assert args is not None + + +def test_create_virtual_display_returns_false_when_ctypes_raises(): + mgr = _make_mgr() + + with patch.object(sys, "platform", "win32"): + import ctypes as _ctypes + + with patch.object(_ctypes, "windll", create=True) as mock_windll: + mock_windll.ParsecVDA = MagicMock(side_effect=OSError("DLL not found")) + result = mgr._create_via_ctypes(1920, 1080, 60) + + assert result is False + + +def test_create_virtual_display_skips_on_non_windows(): + mgr = _make_mgr() + with patch.object(sys, "platform", "darwin"): + assert mgr.create_virtual_display() is False + + +# --------------------------------------------------------------------------- +# 4. create_virtual_display() — subprocess fallback +# --------------------------------------------------------------------------- + + +def test_create_via_subprocess_success(): + mgr = _make_mgr() + + with ( + patch.object(sys, "platform", "win32"), + patch( + "subprocess.run", + return_value=MagicMock(returncode=0, stdout="", stderr=""), + ) as mock_run, + ): + result = mgr._create_via_subprocess(1920, 1080, 60) + + assert result is True + assert mgr._virtual_monitor_index == 0 + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "--width=1920" in cmd + assert "--height=1080" in cmd + assert "--hz=60" in cmd + + +def test_create_via_subprocess_failure_returns_false(): + mgr = _make_mgr() + + with ( + patch.object(sys, "platform", "win32"), + patch( + "subprocess.run", + return_value=MagicMock(returncode=1, stdout="", stderr="error"), + ), + ): + result = mgr._create_via_subprocess(1920, 1080, 60) + + assert result is False + + +def test_create_via_subprocess_file_not_found_returns_false(): + mgr = _make_mgr() + + with ( + patch.object(sys, "platform", "win32"), + patch( + "subprocess.run", + side_effect=FileNotFoundError("ParsecVDA.exe not found"), + ), + ): + result = mgr._create_via_subprocess(1920, 1080, 60) + + assert result is False + + +# --------------------------------------------------------------------------- +# 5. remove_virtual_display() +# --------------------------------------------------------------------------- + + +def test_remove_virtual_display_ctypes_success(): + mgr = _make_mgr() + mgr._virtual_monitor_index = 0 + + mock_vda = MagicMock() + mock_vda.ParsecVDARemoveDisplay.return_value = 0 + + with patch.object(sys, "platform", "win32"): + import ctypes as _ctypes + + with patch.object(_ctypes, "windll", create=True) as mock_windll: + mock_windll.ParsecVDA = mock_vda + result = mgr._remove_via_ctypes() + + assert result is True + + +def test_remove_virtual_display_resets_state(): + mgr = _make_mgr() + mgr._virtual_monitor_index = 0 + + with ( + patch.object(sys, "platform", "win32"), + patch.object(mgr, "_remove_via_ctypes", return_value=True), + ): + result = mgr.remove_virtual_display() + + assert result is True + assert mgr._virtual_monitor_index is None + assert mgr._virtual_monitor_handle is None + + +def test_remove_virtual_display_no_display_returns_false(): + mgr = _make_mgr() + with patch.object(sys, "platform", "win32"): + assert mgr.remove_virtual_display() is False + + +def test_remove_virtual_display_skips_on_non_windows(): + mgr = _make_mgr() + mgr._virtual_monitor_index = 0 + with patch.object(sys, "platform", "darwin"): + assert mgr.remove_virtual_display() is False + + +def test_remove_via_subprocess_success(): + mgr = _make_mgr() + mgr._virtual_monitor_index = 0 + + with ( + patch.object(sys, "platform", "win32"), + patch( + "subprocess.run", + return_value=MagicMock(returncode=0, stdout="", stderr=""), + ) as mock_run, + ): + result = mgr._remove_via_subprocess() + + assert result is True + cmd = mock_run.call_args[0][0] + assert "remove" in cmd + + +def test_remove_via_subprocess_failure(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch( + "subprocess.run", + return_value=MagicMock(returncode=1, stdout="", stderr="error"), + ), + ): + assert mgr._remove_via_subprocess() is False + + +# --------------------------------------------------------------------------- +# 6. move_window_to_virtual_display() +# --------------------------------------------------------------------------- + + +def test_move_window_to_virtual_display_calls_set_window_pos(): + mgr = _make_mgr() + + with ( + patch.object(sys, "platform", "win32"), + patch("operator_use.computer.windows.virtual_display._load_win32", return_value=True), + patch.object(mgr, "_find_window", return_value=12345), + patch.object(mgr, "_get_virtual_monitor", return_value=MagicMock()), + patch.object(mgr, "_set_window_pos", return_value=True) as mock_swp, + ): + result = mgr.move_window_to_virtual_display("Notepad") + + assert result is True + mock_swp.assert_called_once() + + +def test_move_window_returns_false_on_non_windows(): + mgr = _make_mgr() + with patch.object(sys, "platform", "darwin"): + assert mgr.move_window_to_virtual_display("Notepad") is False + + +def test_move_window_returns_false_when_window_not_found(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch("operator_use.computer.windows.virtual_display._load_win32", return_value=True), + patch.object(mgr, "_find_window", return_value=None), + ): + assert mgr.move_window_to_virtual_display("NonExistent") is False + + +def test_move_window_returns_false_when_no_virtual_monitor(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch("operator_use.computer.windows.virtual_display._load_win32", return_value=True), + patch.object(mgr, "_find_window", return_value=12345), + patch.object(mgr, "_get_virtual_monitor", return_value=None), + ): + assert mgr.move_window_to_virtual_display("Notepad") is False + + +def test_move_window_returns_false_when_win32_load_fails(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch("operator_use.computer.windows.virtual_display._load_win32", return_value=False), + ): + assert mgr.move_window_to_virtual_display("Notepad") is False + + +# --------------------------------------------------------------------------- +# 7. Graceful failure — ctypes unavailable / VDD not installed +# --------------------------------------------------------------------------- + + +def test_all_ctypes_calls_fail_gracefully_when_vdd_unavailable(): + """Verify no exceptions bubble out when ParsecVDA DLL is missing.""" + mgr = _make_mgr() + + with patch.object(sys, "platform", "win32"): + import ctypes as _ctypes + + with patch.object(_ctypes, "windll", create=True) as mock_windll: + # Accessing ParsecVDA raises AttributeError (DLL absent) + type(mock_windll).ParsecVDA = property( + lambda self: (_ for _ in ()).throw(AttributeError("ParsecVDA not found")) + ) + # Should return False, not raise + assert mgr._create_via_ctypes(1920, 1080, 60) is False + mgr._virtual_monitor_index = 0 + assert mgr._remove_via_ctypes() is False + + +def test_subprocess_timeout_handled_gracefully(): + mgr = _make_mgr() + with ( + patch.object(sys, "platform", "win32"), + patch( + "subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="ParsecVDA.exe", timeout=15), + ), + ): + assert mgr._create_via_subprocess(1920, 1080, 60) is False + assert mgr._remove_via_subprocess() is False + + +def test_registry_key_exists_returns_false_on_missing_key(): + """_registry_key_exists must not raise on missing key.""" + from operator_use.computer.windows.virtual_display import _registry_key_exists + + with patch("builtins.__import__", side_effect=ImportError("winreg")): + result = _registry_key_exists(0, r"SOFTWARE\Parsec\vdd") + assert result is False From f2891e4cf1938d7f2729eb0afd39557677453fd5 Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Sun, 19 Apr 2026 21:48:07 +0530 Subject: [PATCH 2/4] fix: rebase onto main with plugin fixes included [ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_control_center.py: update import from operator_use.agent.tools.builtin (deleted in main) to operator_use.tools.control_center; add side-effect import of operator_use.agent.tools to resolve circular import; update all patch paths. - test_computer_plugin.py: remove BEFORE_LLM_CALL assertions — 9f5d002 removed _state_hook registration from BEFORE_LLM_CALL by design (state captured inside computer_task loop, not on every main-agent LLM call). - test_browser_plugin.py: same BEFORE_LLM_CALL removal; register_hooks now only stores the hooks reference without registering any handler. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- tests/test_browser_plugin.py | 17 ++++++++++----- tests/test_computer_plugin.py | 8 +++---- tests/test_control_center.py | 39 ++++++++++++++++++----------------- 3 files changed, 35 insertions(+), 29 deletions(-) diff --git a/tests/test_browser_plugin.py b/tests/test_browser_plugin.py index 9089d95..41e4c8b 100644 --- a/tests/test_browser_plugin.py +++ b/tests/test_browser_plugin.py @@ -61,7 +61,9 @@ def test_unregister_tools_removes_browser_tool(): # --------------------------------------------------------------------------- -# register_hooks — BEFORE_LLM_CALL gated on _enabled +# register_hooks — no hooks registered on BEFORE_LLM_CALL +# (BEFORE_LLM_CALL registration was removed in 9f5d002: browser state is +# now captured inside the browser_task loop, not on every main-agent LLM call) # --------------------------------------------------------------------------- @@ -69,7 +71,8 @@ def test_disabled_plugin_registers_no_hooks(): plugin = BrowserPlugin(enabled=False) hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # register_hooks stores the hooks reference but does not register any handler + assert plugin._hooks is hooks def test_enabled_plugin_registers_state_hook(): @@ -77,7 +80,8 @@ def test_enabled_plugin_registers_state_hook(): plugin._enabled = True hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # No hook is registered on BEFORE_LLM_CALL; hook reference is stored only + assert plugin._hooks is hooks def test_unregister_hooks_removes_state_hook(): @@ -86,7 +90,8 @@ def test_unregister_hooks_removes_state_hook(): hooks = Hooks() plugin.register_hooks(hooks) plugin.unregister_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # unregister_hooks is a no-op; _hooks reference is preserved + assert plugin._hooks is hooks # --------------------------------------------------------------------------- @@ -135,7 +140,8 @@ async def test_enable_registers_hooks_and_injects_prompt(): await plugin.enable() assert plugin._enabled is True - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # _state_hook is not registered on BEFORE_LLM_CALL (see 9f5d002) + assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] context.register_plugin_prompt.assert_called_once_with(SYSTEM_PROMPT) @@ -151,6 +157,7 @@ async def test_disable_unregisters_hooks_and_removes_prompt(): await plugin.disable() assert plugin._enabled is False + # _state_hook was never in BEFORE_LLM_CALL (see 9f5d002) assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] context.unregister_plugin_prompt.assert_called_once_with(SYSTEM_PROMPT) diff --git a/tests/test_computer_plugin.py b/tests/test_computer_plugin.py index 47e4480..68eb277 100644 --- a/tests/test_computer_plugin.py +++ b/tests/test_computer_plugin.py @@ -34,7 +34,9 @@ def test_enabled_plugin_returns_system_prompt(): # --------------------------------------------------------------------------- -# register_hooks — BEFORE_LLM_CALL + AFTER_TOOL_CALL, gated on _enabled +# register_hooks — AFTER_TOOL_CALL only, gated on _enabled +# (BEFORE_LLM_CALL was removed in 9f5d002: state captured inside +# computer_task loop, not on every main-agent LLM call) # --------------------------------------------------------------------------- @@ -42,7 +44,6 @@ def test_disabled_plugin_registers_no_hooks(): plugin = ComputerPlugin(enabled=False) hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook not in hooks._handlers[HookEvent.AFTER_TOOL_CALL] @@ -51,7 +52,6 @@ def test_enabled_plugin_registers_both_hooks(): plugin._enabled = True hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook in hooks._handlers[HookEvent.AFTER_TOOL_CALL] @@ -61,7 +61,6 @@ def test_unregister_hooks_removes_both(): hooks = Hooks() plugin.register_hooks(hooks) plugin.unregister_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook not in hooks._handlers[HookEvent.AFTER_TOOL_CALL] @@ -111,7 +110,6 @@ async def test_enable_registers_both_hooks_and_prompt(): await plugin.enable() assert plugin._enabled is True - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook in hooks._handlers[HookEvent.AFTER_TOOL_CALL] context.register_plugin_prompt.assert_called_once_with(SYSTEM_PROMPT) diff --git a/tests/test_control_center.py b/tests/test_control_center.py index f3a2e5b..158d84c 100644 --- a/tests/test_control_center.py +++ b/tests/test_control_center.py @@ -4,12 +4,17 @@ import pytest from unittest.mock import AsyncMock, MagicMock, patch -from operator_use.agent.tools.builtin.control_center import ( +# Import via operator_use.agent.tools so the circular-import chain is resolved +# in the correct order before we access the module directly. +import operator_use.agent.tools # noqa: F401 — side-effect import resolves cycle +from operator_use.tools.control_center import ( control_center, _set_plugin_enabled, _get_plugin_enabled, ) +_MODULE = "operator_use.tools.control_center" + # --------------------------------------------------------------------------- # Helpers @@ -73,7 +78,7 @@ async def test_enable_browser_use_calls_agent(tmp_path): mock_agent = MagicMock() mock_agent.enable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(browser_use=True, _agent=mock_agent) mock_agent.enable_browser_use.assert_awaited_once() @@ -90,7 +95,7 @@ async def test_enable_both_computer_use_and_browser_use_independently(tmp_path): mock_agent.enable_computer_use = AsyncMock() mock_agent.enable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(computer_use=True, browser_use=True, _agent=mock_agent) saved = json.loads(cfg_file.read_text()) @@ -111,7 +116,7 @@ async def test_disable_browser_use_calls_agent(tmp_path): mock_agent = MagicMock() mock_agent.disable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(browser_use=False, _agent=mock_agent) mock_agent.disable_browser_use.assert_awaited_once() @@ -124,7 +129,7 @@ async def test_status_only_returns_current_state(tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps(cfg)) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc() assert result.success @@ -148,10 +153,8 @@ async def test_audit_log_emitted_on_plugin_change(tmp_path, caplog): mock_agent = MagicMock() mock_agent.enable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with caplog.at_level( - logging.WARNING, logger="operator_use.agent.tools.builtin.control_center" - ): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with caplog.at_level(logging.WARNING, logger=_MODULE): await _call_cc( browser_use=True, _agent=mock_agent, @@ -172,10 +175,8 @@ async def test_audit_log_emitted_on_status_check(tmp_path, caplog): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps(cfg)) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with caplog.at_level( - logging.WARNING, logger="operator_use.agent.tools.builtin.control_center" - ): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with caplog.at_level(logging.WARNING, logger=_MODULE): await _call_cc(_channel="discord", _chat_id="999", _agent_id="op") assert any("control_center" in r.message for r in caplog.records) @@ -197,8 +198,8 @@ async def test_restart_calls_graceful_fn_not_os_exit(tmp_path): async def mock_graceful(): graceful_called.append(True) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with patch("operator_use.agent.tools.builtin.control_center._do_restart") as mock_restart: + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}._do_restart") as mock_restart: mock_restart.return_value = None result = await _call_cc(restart=True, _graceful_restart_fn=mock_graceful) @@ -217,8 +218,8 @@ async def test_restart_without_graceful_fn_still_works(tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps(cfg)) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with patch("operator_use.agent.tools.builtin.control_center._do_restart"): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}._do_restart"): result = await _call_cc(restart=True) assert result.success @@ -234,7 +235,7 @@ async def test_returns_error_when_no_agents(tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps({"agents": {"list": []}})) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(browser_use=True) assert not result.success @@ -244,7 +245,7 @@ async def test_returns_error_when_no_agents(tmp_path): async def test_returns_error_when_config_missing(tmp_path): missing = tmp_path / "no_config.json" - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", missing): + with patch(f"{_MODULE}.CONFIG_PATH", missing): result = await _call_cc(browser_use=True) assert not result.success From 05cb33ed4c454a393ff4545247759653064f57ab Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Sun, 19 Apr 2026 22:02:00 +0530 Subject: [PATCH 3/4] fix: update test imports for refactored tools paths [ci] --- tests/test_local_agents.py | 2 +- tests/test_plugins.py | 2 +- tests/test_tool_registry.py | 2 +- tests/test_tools.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_local_agents.py b/tests/test_local_agents.py index 8fd831b..a1b5168 100644 --- a/tests/test_local_agents.py +++ b/tests/test_local_agents.py @@ -2,7 +2,7 @@ import pytest -from operator_use.agent.tools.builtin.local_agents import LOCAL_AGENT_DELEGATION_CHAIN, localagents +from operator_use.tools.local_agents import LOCAL_AGENT_DELEGATION_CHAIN, localagents from operator_use.messages.service import AIMessage diff --git a/tests/test_plugins.py b/tests/test_plugins.py index f6ba6d4..5d9f8b9 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -7,7 +7,7 @@ from operator_use.agent.tools.registry import ToolRegistry from operator_use.agent.hooks.service import Hooks from operator_use.agent.hooks.events import HookEvent -from operator_use.tools.service import Tool +from operator_use.agent.tools.service import Tool from pydantic import BaseModel diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py index ca6ed75..77c70b9 100644 --- a/tests/test_tool_registry.py +++ b/tests/test_tool_registry.py @@ -4,7 +4,7 @@ from pydantic import BaseModel from operator_use.agent.tools.registry import ToolRegistry -from operator_use.tools.service import Tool +from operator_use.agent.tools.service import Tool # --- Helpers --- diff --git a/tests/test_tools.py b/tests/test_tools.py index 8cbf913..de572ab 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -4,7 +4,7 @@ from pydantic import BaseModel from typing import Literal -from operator_use.tools.service import Tool, ToolResult +from operator_use.agent.tools.service import Tool, ToolResult # --- ToolResult --- From bc65d6a84db38c1a1d3c80b8e37e34cb7117d4e9 Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Sun, 19 Apr 2026 22:10:33 +0530 Subject: [PATCH 4/4] fix: fix remaining test_agent.py and e2e imports for refactored tools [ci] --- tests/test_agent.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_agent.py b/tests/test_agent.py index 4fb6c3f..13db174 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -186,7 +186,7 @@ async def test_agent_run_with_tool_call_then_text(tmp_path): # Register a simple echo tool from pydantic import BaseModel - from operator_use.tools.service import Tool + from operator_use.agent.tools.service import Tool class EchoParams(BaseModel): message: str