Show AI control, temporarily block physical input, and allow user takeover

README.md

@@ -719,6 +719,8 @@ MCP Client can access the following tools to interact with Windows:
 - `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.
+- `ControlStatus`: Read the desktop control owner and remaining user-idle wait. It stays available when user takeover blocks new tool calls. During AI control, a visible edge and cursor glow indicate ownership; moving the physical mouse substantially or pressing `Ctrl+Alt+Shift+Backspace` returns control to the user until 10 seconds of inactivity. During screenshot capture, the indicator is temporarily hidden and physical input is released; any physical input during this interval preempts AI control, although `ControlStatus` can briefly still report `ai` until that event is processed. Stepwise desktop input stops at its next checkpoint, but an already-running external command (for example `PowerShell` or an `App` launch) may continue after takeover; this control gate does not cancel processes or undo completed side effects. On a detected hook fault, injected keys and mouse buttons are released asynchronously, so physical input can briefly overlap a still-held AI key or button. Windows may also silently remove a low-level hook without notifying this process; periodic hook replacement reduces but does not eliminate that undetected window. User takeover is therefore best-effort, not a security boundary or a guarantee of exclusive control.
+  Sessionless MCP 2026-07-28 clients cannot receive unsolicited control-state logs; they should query `ControlStatus` when they need the current owner.
 - `DisplayInventory`: Read display layout, work areas, effective DPI, and scale metadata.
 - `Screenshot`: Fast screenshot-first desktop capture with cursor position, active/open windows, and an image. Skips UI tree extraction for speed and should be the default first call when you mainly need visual context. Supports `display=[0]` or `display=[0,1]` using zero-based active Windows display indices, and `region=[left, top, right, bottom]` (virtual-desktop pixel coordinates) to capture just that rectangle instead of the whole screen — cheaper on tokens when you already know which area matters. `region` takes precedence over `display` when both are given; an invalid or out-of-bounds region raises an error. After capture, a brief orange-red glowing border is drawn inside the captured area as a visual confirmation (set `WINDOWS_MCP_DISABLE_FLASH=1` to disable).
 - `Snapshot`: Full desktop state capture for workflows that need interactive element ids, scrollable regions, or `use_dom=True` browser extraction. Supports `use_vision=True` for including screenshots, `display=[0]` or `display=[0,1]` using zero-based active Windows display indices, and `region=[left, top, right, bottom]` (virtual-desktop pixel coordinates) to inspect just that rectangle instead of the whole screen; `region` takes precedence over `display` when both are given, and an invalid or out-of-bounds region raises an error.

manifest.json

@@ -79,6 +79,10 @@
     }
   },
   "tools": [
+    {
+      "name": "ControlStatus",
+      "description": "Reports whether AI or the user currently controls the desktop and the remaining user-idle wait before AI control can resume. This read-only status remains available during user takeover."
+    },
     {
       "name": "App",
       "description": "Manages Windows applications with three modes: 'launch' (opens the prescibed application), 'resize' (adjusts the size/position of a named window or the active window if name is omitted), 'switch' (brings specific window into focus)."

src/windows_mcp/__main__.py

@@ -1,5 +1,6 @@
 from contextlib import asynccontextmanager
 from windows_mcp.config import enable_debug
+from windows_mcp.infrastructure.config import ControlConfig
 from windows_mcp.infrastructure import (
     AuthKeyMiddleware,
     OAuthOnlyMiddleware,
@@ -29,6 +30,7 @@
 import shlex
 import secrets
 import subprocess
+import threading
 import click
 import os
 import sys
@@ -49,11 +51,14 @@ def _echo_section(title: str) -> None:
     Setting PYTHONIOENCODING=utf-8 is a workaround; this is the fix.
     """
     enc = getattr(sys.stdout, "encoding", None) or "ascii"
+    # Older Ruff releases drop the required parentheses around this exception tuple.
+    # fmt: off
     try:
         "─".encode(enc)
         bar = "─" * 3
     except (UnicodeEncodeError, LookupError):
         bar = "==="
+    # fmt: on
     click.echo(f"\n{bar} {title} {bar}")
 
 
@@ -62,6 +67,8 @@ def _echo_section(title: str) -> None:
 analytics: Any | None = None
 screen_size: Any | None = None
 _mcp: FastMCP | None = None
+_control_config = ControlConfig()
+_control_notifier: Any | None = None
 
 instructions = dedent("""
 Windows MCP server provides tools to interact directly with the Windows desktop,
@@ -240,7 +247,7 @@ def _start_watchdog(desktop):
 
 def _build_mcp() -> FastMCP:
     """Create the MCP server instance."""
-    global _mcp
+    global _mcp, _control_notifier
 
     if _mcp is not None:
         return _mcp
@@ -251,11 +258,68 @@ def _build_mcp() -> FastMCP:
         from windows_mcp.tools import register_all
     except ModuleNotFoundError as exc:
         _exit_missing_dependency(exc)
+    from windows_mcp.tools.control_notifications import ControlNotifier, ControlToolGate
+    from windows_mcp.desktop.control import get_controller
+    from windows_mcp.desktop import control_overlay
+
+    controller = get_controller()
+    notifier = ControlNotifier(controller)
+    _control_notifier = notifier
+    control_loop: asyncio.AbstractEventLoop | None = None
+
+    def apply_control_state(status: dict) -> None:
+        try:
+            # Queued input-thread updates can arrive after a later takeover.
+            # Never let an old AI event re-show the indicator in user mode.
+            if status.get("generation") is not None and status[
+                "generation"
+            ] != controller.status().get("generation"):
+                return
+            generation = status.get("generation")
+            if status["state"] == "ai":
+                applied = control_overlay.set_active(True, generation=generation)
+                if applied and generation is not None:
+                    controller.arm_visible(status["generation"])
+            elif status["state"] == "takeover_pending":
+                control_overlay.set_pending(True, generation=generation)
+            else:
+                control_overlay.set_active(False, generation=generation)
+        except Exception:
+            # A missing visual indicator invalidates the AI control lease.
+            controller._fail_open()
+            logger.exception("AI control indicator failed; releasing physical input")
+
+    def show_control_state(status: dict) -> None:
+        if threading.current_thread() is controller._thread or not (
+            status["state"] == "ai" and controller.visible_ack_required()
+        ):
+            # Never wait for a window-owner acknowledgement on the hook thread.
+            # A capture may hold the overlay lock for an unbounded backend call;
+            # only a new MCP lease may wait for the visible AI acknowledgement.
+            # Other state transitions must not block the MCP event loop.
+            try:
+                if control_loop is None or control_loop.is_closed():
+                    raise RuntimeError("server event loop unavailable")
+                control_loop.call_soon_threadsafe(
+                    lambda: asyncio.create_task(
+                        asyncio.to_thread(apply_control_state, status.copy())
+                    )
+                )
+            except RuntimeError:
+                controller._fail_open()
+                logger.exception("Could not schedule AI control indicator update")
+        else:
+            # begin_call waits for the visible AI indicator before any tool runs.
+            apply_control_state(status)
+
+    controller.subscribe(show_control_state)
 
     @asynccontextmanager
     async def lifespan(app: FastMCP):
         """Runs initialization code before the server starts and cleanup code after it shuts down."""
         global desktop, watchdog, analytics, screen_size
+        nonlocal control_loop
+        control_loop = asyncio.get_running_loop()
 
         if os.getenv("ANONYMIZED_TELEMETRY", "true").lower() != "false":
             analytics = PostHogAnalytics()
@@ -264,17 +328,34 @@ async def lifespan(app: FastMCP):
 
         watchdog = _start_watchdog(desktop)
 
+        # Set thresholds before installing input hooks; invalid TOML was rejected at load.
+        controller.mouse_takeover_units = _control_config.mouse_takeover_units
+        controller.mouse_takeover_pixels = _control_config.mouse_takeover_pixels
+
         try:
+            control_overlay.start()
+            controller.set_health_probe(control_overlay.is_healthy)
+            controller.start()
+            notifier.start()
             logger.debug("Server started, entering main loop")
             yield
         finally:
-            logger.debug("Shutting down: stopping watchdog and analytics")
-            if watchdog:
-                watchdog.stop()
-            if analytics:
-                await analytics.close()
+            await notifier.close()
+            try:
+                controller.stop()
+            finally:
+                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))
     register_all(_mcp, get_desktop=_get_desktop, get_analytics=_get_analytics)
     return _mcp
 
@@ -385,11 +466,12 @@ def _remove(name):
 
     if explicit_tools:
         keep = {t for t in explicit_tools if t in registered}
+        keep.add("ControlStatus")
         for name in registered - keep:
             _remove(name)
     elif exclude_tools:
         for name in exclude_tools:
-            if name in registered:
+            if name in registered and name != "ControlStatus":
                 _remove(name)
     logger.debug("Tool filter applied: explicit=%s exclude=%s", explicit_tools, exclude_tools)
 
@@ -410,6 +492,11 @@ def _run_server(
     stateless_http: bool = False,
 ) -> None:
     mcp = _build_mcp()
+    if _control_notifier is not None:
+        # stdio has one session until process exit; stateless HTTP has no
+        # durable session to which a later ownership event can be pushed.
+        _control_notifier.ttl = float("inf") if transport == "stdio" else 60.0
+        _control_notifier.push_enabled = not (transport == "streamable-http" and stateless_http)
     if explicit_tools or exclude_tools:
         _apply_tool_filter(mcp, explicit_tools, exclude_tools)
     match transport:
@@ -588,6 +675,7 @@ def serve(
     oauth_client_secret,
     stateless_http,
 ):
+    global _control_config
     asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
     install_selfpipe_guard()
     if transport == Transport.STDIO.value:
@@ -604,6 +692,7 @@ def serve(
         cfg = load_config(config_path)
     except (FileNotFoundError, ValueError) as exc:
         raise click.ClickException(str(exc))
+    _control_config = cfg.control
 
     transport = _choose_value(ctx, "transport", transport, cfg.server.transport, "stdio")
     host = _choose_value(ctx, "host", host, cfg.server.host, "localhost")