[Refactor] Add cua-driver to replace the os_host module

[wangxingjun778]

No description provided.

[gemini-code-assist]

Code Review

This pull request migrates the project from the legacy Swift-based macOS host to a cross-platform, pluggable execution backend powered by cua-driver and the Model Control Protocol (MCP). It removes the legacy os_host module and introduces CuaDriverClient along with a suite of passive signal observers (FileSystemObserver, AppFocusObserver, ClipboardObserver, and InputTapObserver) managed by ObservationDaemon. The code review feedback identifies several critical issues: blocking the asyncio event loop with synchronous calls (like time.sleep and blocking I/O) in async contexts, potential task leaks due to uncancelled futures on timeout, highly inefficient polling loops that spawn subprocesses frequently, and incorrect handling of PermissionError when checking process status. Addressing these issues by utilizing executors, caching active window states, and using shutil.which will significantly improve the performance and robustness of the new architecture.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

If the lifecycle task shutdown times out, the task is left running. Cancel fut explicitly to ensure it is terminated.

Suggested change

	try:
	fut.result(timeout=5.0)
	except concurrent.futures.TimeoutError:
	logger.warning("cua-driver session shutdown timed out")
	try:
	fut.result(timeout=5.0)
	except concurrent.futures.TimeoutError:
	logger.warning("cua-driver session shutdown timed out")
	fut.cancel()

[gemini-code-assist]

The local handlers in _LOCAL_DISPATCH perform blocking disk I/O or spawn synchronous subprocesses (e.g., clipboard operations, file operations). Calling them directly inside async def call blocks the event loop. Run synchronous handlers in an executor.

Suggested change

	try:
	tool_name, tool_args = self._map_to_cua_tool(method, params)
	except _LocalResult as lr:
	handler = _LOCAL_DISPATCH.get(method)
	if handler is not None:
	return await asyncio.get_running_loop().run_in_executor(
	None, handler, params
	)

[gemini-code-assist]

Calling time.sleep(0.1) inside an async def function blocks the entire event loop. Use await asyncio.sleep(0.1) instead to keep the event loop responsive.

Suggested change

	# Wait up to 5s for process to exit
	for _ in range(50):
	time.sleep(0.1)
	try:
	os.kill(pid, 0)
	except OSError:
	break
	# Wait up to 5s for process to exit
	import asyncio
	for _ in range(50):
	await asyncio.sleep(0.1)
	try:
	os.kill(pid, 0)
	except OSError:
	break

[gemini-code-assist]

The polling loop calls get_active_app() every 0.5 seconds. Spawning four subprocesses (xdotool and xprop) every 0.5 seconds is extremely inefficient and will peg the CPU on Linux. Cache the last active window ID (wid) and only fetch the rest of the window details (name, PID, WM_CLASS) when the active window ID actually changes.

    def __init__(self) -> None:
        self._last_wid: Optional[str] = None
        self._cached_info: Optional[Dict[str, Any]] = None

    def get_active_app(self) -> Optional[Dict[str, Any]]:
        try:
            # Get active window ID
            wid_result = subprocess.run(
                ["xdotool", "getactivewindow"],
                capture_output=True, text=True, timeout=2,
            )
            if wid_result.returncode != 0:
                return None
            wid = wid_result.stdout.strip()
            if wid == self._last_wid:
                return self._cached_info

            # Get window name
            name_result = subprocess.run(
                ["xdotool", "getactivewindow", "getwindowname"],
                capture_output=True, text=True, timeout=2,
            )
            window_title = name_result.stdout.strip() if name_result.returncode == 0 else ""

            # Get PID
            pid_result = subprocess.run(
                ["xdotool", "getactivewindow", "getwindowpid"],
                capture_output=True, text=True, timeout=2,
            )
            pid = int(pid_result.stdout.strip()) if pid_result.returncode == 0 else 0

            # Get WM_CLASS for app identification
            class_result = subprocess.run(
                ["xprop", "-id", wid, "WM_CLASS"],
                capture_output=True, text=True, timeout=2,
            )
            app_id = ""
            app_name = ""
            if class_result.returncode == 0:
                # WM_CLASS(STRING) = "instance", "class"
                parts = class_result.stdout.split('"')
                if len(parts) >= 4:
                    app_id = parts[3]  # class name
                    app_name = parts[3]
                elif len(parts) >= 2:
                    app_id = parts[1]
                    app_name = parts[1]

            info = {
                "app_id": app_id,
                "app_name": app_name,
                "pid": pid,
                "window_title": window_title,
            }
            self._last_wid = wid
            self._cached_info = info
            return info

[gemini-code-assist]

Calling the synchronous, blocking _resolve_mcp_invocation function directly inside an async function blocks the entire event loop. Run it in an executor to keep the loop responsive.

            command, args = await asyncio.get_running_loop().run_in_executor(
                None, _resolve_mcp_invocation, _get_cua_driver_cmd()
            )

[gemini-code-assist]

On Unix/macOS, if the daemon process is running under a different user or has restricted privileges, os.kill(pid, 0) can raise a PermissionError. In the current implementation, catching OSError generally will treat this as a stale/invalid PID file and delete it, which is incorrect because the process is actually running. Explicitly catch PermissionError and treat the process as alive to prevent starting duplicate daemons.

Suggested change

	try:
	pid = int(pid_file.read_text().strip())
	# Check if process is alive
	os.kill(pid, 0)
	return pid
	except (ValueError, OSError):
	# Stale or invalid PID file
	try:
	pid_file.unlink()
	except OSError:
	pass
	return None
	try:
	pid = int(pid_file.read_text().strip())
	# Check if process is alive
	os.kill(pid, 0)
	return pid
	except PermissionError:
	# Process is alive but we don't have permission to signal it
	return pid
	except (ValueError, OSError):
	# Stale or invalid PID file
	try:
	pid_file.unlink()
	except OSError:
	pass
	return None

[gemini-code-assist]

Use the dynamic _get_cua_driver_cmd() helper to ensure any runtime configuration overrides are respected.

Suggested change

	return bool(shutil.which(_CUA_DRIVER_CMD))
	return bool(shutil.which(_get_cua_driver_cmd()))

[gemini-code-assist]

Use the dynamic _get_cua_driver_cmd() helper to ensure any runtime configuration overrides are respected.

Suggested change

	"cua_driver_cmd": _CUA_DRIVER_CMD,
	"cua_driver_cmd": _get_cua_driver_cmd(),

[gemini-code-assist]

Evaluating _CUA_DRIVER_CMD at import time prevents it from picking up configuration overrides loaded from .env or settings files later. Define a helper function to read the command dynamically at runtime.

Suggested change

	_CUA_DRIVER_CMD = os.environ.get("LEAPFLOW_CUA_DRIVER_CMD", "cua-driver")
	def _get_cua_driver_cmd() -> str:
	return os.environ.get("LEAPFLOW_CUA_DRIVER_CMD", "cua-driver")

[gemini-code-assist]

Spawning a subprocess (subprocess.run) during object initialization just to check if a command exists is inefficient and can cause side effects or hang. Use shutil.which to check if the binaries are available on the system PATH instead.

Suggested change

	def __init__(self) -> None:
	self._cmd: Optional[list[str]] = None
	# Detect available tool
	for cmd in [
	["xclip", "-selection", "clipboard", "-o"],
	["xsel", "--clipboard", "--output"],
	]:
	try:
	subprocess.run(
	cmd, capture_output=True, timeout=1,
	)
	self._cmd = cmd
	break
	except FileNotFoundError:
	continue
	def __init__(self) -> None:
	self._cmd: Optional[list[str]] = None
	import shutil
	if shutil.which("xclip"):
	self._cmd = ["xclip", "-selection", "clipboard", "-o"]
	elif shutil.which("xsel"):
	self._cmd = ["xsel", "--clipboard", "--output"]