refac

open_terminal/browser/chrome.py

@@ -1,8 +1,9 @@
-"""Chrome tabs. CDP stays on private pipes, never a shared TCP port."""
+"""Chrome tabs using private CDP pipes on POSIX and loopback CDP on Windows."""
 
 from __future__ import annotations
 
 import asyncio
+import base64
 import contextlib
 import hashlib
 import json
@@ -12,6 +13,8 @@ from pathlib import Path
 import shutil
 import sys
 
+import websockets
+
 from fastapi import WebSocket, WebSocketDisconnect
 from open_terminal.env import sanitized_environ
 from .rewrite import target_url
@@ -26,12 +29,24 @@ def find_browser():
         shutil.which(name)
         for name in (
             "chromium",
+            "chrome",
             "chromium-browser",
             "google-chrome",
             "google-chrome-stable",
             "msedge",
         )
     ]
+    if os.name == "nt":
+        for root in ("PROGRAMFILES", "PROGRAMFILES(X86)", "LOCALAPPDATA"):
+            if directory := os.environ.get(root):
+                candidates += [
+                    str(Path(directory) / path)
+                    for path in (
+                        "Google/Chrome/Application/chrome.exe",
+                        "Microsoft/Edge/Application/msedge.exe",
+                        "Chromium/Application/chrome.exe",
+                    )
+                ]
     candidates += [
         "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
         "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
@@ -77,8 +92,9 @@ def lock_profile(profile):
 
 
 class Host:
-    def __init__(self, process, display=None):
+    def __init__(self, process, display=None, channel=None):
         self.process = process
+        self.channel = channel
         self.display = display
         self.viewer = None
         self.display_mode = None
@@ -91,8 +107,12 @@ class Host:
     async def read(self):
         try:
             while True:
-                raw = await self.process.stdout.readuntil(b"\0")
-                message = json.loads(raw[:-1])
+                raw = (
+                    await self.channel.recv()
+                    if self.channel
+                    else (await self.process.stdout.readuntil(b"\0"))[:-1]
+                )
+                message = json.loads(raw)
                 if "id" in message:
                     future = self.pending.pop(message["id"], None)
                     if future and not future.done():
@@ -124,8 +144,11 @@ class Host:
         if session:
             message["sessionId"] = session
         try:
-            self.process.stdin.write(json.dumps(message).encode() + b"\0")
-            await self.process.stdin.drain()
+            if self.channel:
+                await self.channel.send(json.dumps(message))
+            else:
+                self.process.stdin.write(json.dumps(message).encode() + b"\0")
+                await self.process.stdin.drain()
             try:
                 return await asyncio.wait_for(future, 15)
             except TimeoutError as exc:
@@ -146,6 +169,8 @@ class Host:
                 await self.process.wait()
         self.reader.cancel()
         await asyncio.gather(self.reader, return_exceptions=True)
+        if self.channel:
+            await self.channel.close()
         if self.display:
             await self.display.close()
 
@@ -154,6 +179,44 @@ hosts = {}
 host_lock = asyncio.Lock()
 
 
+async def windows_host(args, profile, home):
+    # Chrome on Windows exposes CDP on a random loopback port, scoped to this profile.
+    path = Path(profile) / "DevToolsActivePort"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.unlink(missing_ok=True)
+    process = await asyncio.create_subprocess_exec(
+        *args,
+        "--headless=new",
+        "--remote-debugging-port=0",
+        "--remote-debugging-address=127.0.0.1",
+        "about:blank",
+        stdout=asyncio.subprocess.DEVNULL,
+        stderr=asyncio.subprocess.DEVNULL,
+        env=sanitized_environ({"HOME": home}),
+    )
+    try:
+        for _ in range(150):
+            if process.returncode is not None:
+                raise RuntimeError(
+                    "Chrome exited during startup. Close any other process using its profile."
+                )
+            if path.exists():
+                lines = path.read_text().splitlines()
+                if len(lines) == 2 and lines[1].startswith("/devtools/browser/"):
+                    channel = await websockets.connect(
+                        f"ws://127.0.0.1:{int(lines[0])}{lines[1]}",
+                        max_size=16 * 1024 * 1024,
+                    )
+                    return Host(process, channel=channel)
+            await asyncio.sleep(0.1)
+        raise RuntimeError("Chrome did not open its debugging connection")
+    except BaseException:
+        if process.returncode is None:
+            process.kill()
+        await process.wait()
+        raise
+
+
 async def get_host(session):
     async with host_lock:
         key = (session.owner, session.fs.home)
@@ -166,8 +229,6 @@ async def get_host(session):
             raise RuntimeError(
                 "Chrome mode requires Chromium. Install it or set OPEN_TERMINAL_BROWSER_EXECUTABLE."
             )
-        if os.name == "nt":
-            raise RuntimeError("Managed Chrome currently requires a POSIX host")
         # Let Chrome create its own profile as the same OS user as the terminal.
         suffix = hashlib.sha256(session.owner.encode()).hexdigest()[:24]
         profile = str(
@@ -175,12 +236,16 @@ async def get_host(session):
         )
         args = [
             binary,
-            "--remote-debugging-pipe",
             f"--user-data-dir={profile}",
             "--no-first-run",
             "--no-default-browser-check",
             "--disable-dev-shm-usage",
         ]
+        if os.name == "nt":
+            host = await windows_host(args, profile, session.fs.home)
+            hosts[key] = host
+            return host
+        args.append("--remote-debugging-pipe")
         if Path("/.dockerenv").exists():
             args.append("--no-sandbox")
         display = None
@@ -279,6 +344,7 @@ class Chrome:
         self.navigation = None
         self.error = ""
         self.vnc_peer = None
+        self.acks = set()
         host.handlers[cdp_session] = self.event
 
     @classmethod
@@ -341,7 +407,18 @@ class Chrome:
 
     def event(self, method, params):
         if method == "Page.screencastFrame":
-            # Chrome applies backpressure until the frame is acknowledged.
+            # A slow viewer must not block Chrome; retain only its newest full frame.
+            task = asyncio.create_task(
+                self.send("Page.screencastFrameAck", {"sessionId": params["sessionId"]})
+            )
+            self.acks.add(task)
+
+            def done(task):
+                self.acks.discard(task)
+                if not task.cancelled():
+                    task.exception()
+
+            task.add_done_callback(done)
             if self.frame.full():
                 self.frame.get_nowait()
             self.frame.put_nowait(params)
@@ -375,7 +452,7 @@ class Chrome:
             "error": self.error,
             "canGoBack": index > 0,
             "canGoForward": index < len(entries) - 1,
-            "transport": "vnc" if self.host.display else "jpeg",
+            "transport": "vnc" if self.host.display else "frames",
         }
 
     async def capture(self):
@@ -384,6 +461,7 @@ class Chrome:
         await self.send("Page.stopScreencast")
         if not self.visible:
             return
+        await self.send("Page.bringToFront")
         await self.send(
             "Page.startScreencast",
             {
@@ -452,25 +530,48 @@ class Chrome:
                 )
         self.peer = ws
         tasks = []
+        binary_frames = False
+        ready = asyncio.Event()
+        ready.set()
+        pending_mouse = {}
+        mouse_changed = asyncio.Event()
         try:
             if state := await self.state():
                 await ws.send_json(state)
             await self.capture()
 
             async def frames():
+                from open_terminal.frames import Frames
+
+                encoder = Frames()
                 while True:
+                    await ready.wait()
                     frame = await self.frame.get()
-                    await ws.send_json(
-                        {
-                            "type": "frame",
-                            "data": frame["data"],
-                            "width": self.width,
-                            "height": self.height,
-                        }
-                    )
-                    await self.send(
-                        "Page.screencastFrameAck", {"sessionId": frame["sessionId"]}
-                    )
+                    if binary_frames:
+                        packet = await asyncio.to_thread(
+                            encoder.jpeg, base64.b64decode(frame["data"]), self.quality
+                        )
+                        if packet is not None:
+                            ready.clear()
+                            await ws.send_bytes(packet)
+                    else:
+                        await ws.send_json(
+                            {
+                                "type": "frame",
+                                "data": frame["data"],
+                                "width": self.width,
+                                "height": self.height,
+                            }
+                        )
+
+            async def mouse_commands():
+                while True:
+                    await mouse_changed.wait()
+                    events = list(pending_mouse.values())
+                    pending_mouse.clear()
+                    mouse_changed.clear()
+                    for event in events:
+                        await self.send("Input.dispatchMouseEvent", event)
 
             async def states():
                 while True:
@@ -480,12 +581,19 @@ class Chrome:
                         await ws.send_json(state)
 
             async def commands():
+                nonlocal binary_frames
                 while True:
                     msg = await ws.receive_json()
                     if not isinstance(msg, dict):
                         continue
                     kind = msg.get("type")
-                    if kind == "navigate":
+                    if kind == "frames":
+                        binary_frames = True
+                        await self.capture()
+                    elif kind == "frameAck":
+                        ready.set()
+                    elif kind == "navigate":
+                        pending_mouse.clear()
                         self.navigate(str(msg.get("url", "")))
                     elif kind in ("back", "forward"):
                         history = await self.send("Page.getNavigationHistory")
@@ -559,7 +667,21 @@ class Chrome:
                             "mouseMoved",
                             "mouseWheel",
                         ):
-                            await self.send("Input.dispatchMouseEvent", event)
+                            if event["type"] in ("mouseMoved", "mouseWheel"):
+                                if event["type"] == "mouseWheel" and (
+                                    previous := pending_mouse.get("mouseWheel")
+                                ):
+                                    event["deltaX"] = previous.get(
+                                        "deltaX", 0
+                                    ) + event.get("deltaX", 0)
+                                    event["deltaY"] = previous.get(
+                                        "deltaY", 0
+                                    ) + event.get("deltaY", 0)
+                                pending_mouse[event["type"]] = event
+                                mouse_changed.set()
+                            else:
+                                pending_mouse.pop("mouseMoved", None)
+                                await self.send("Input.dispatchMouseEvent", event)
                     elif kind == "key":
                         event = {
                             k: msg[k]
@@ -586,6 +708,7 @@ class Chrome:
                 asyncio.create_task(frames()),
                 asyncio.create_task(states()),
                 asyncio.create_task(commands()),
+                asyncio.create_task(mouse_commands()),
             ]
             await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
         except (WebSocketDisconnect, RuntimeError, ValueError, TypeError):
@@ -603,6 +726,9 @@ class Chrome:
                     await self.send("Page.stopScreencast")
 
     async def close(self):
+        for task in list(self.acks):
+            task.cancel()
+        await asyncio.gather(*self.acks, return_exceptions=True)
         if self.navigation:
             self.navigation.cancel()
             await asyncio.gather(self.navigation, return_exceptions=True)

open_terminal/desktop.py

@@ -6,6 +6,8 @@ import os
 from pathlib import Path
 import secrets
 import shutil
+import sys
+import importlib.util
 import signal
 import tempfile
 from typing import Literal
@@ -32,6 +34,10 @@ startup_lock = asyncio.Lock()
 
 
 def available():
+    if sys.platform in ("darwin", "win32"):
+        return all(
+            importlib.util.find_spec(name) for name in ("mss", "pyautogui", "pyperclip")
+        )
     return all(
         shutil.which(name)
         for name in (
@@ -47,6 +53,9 @@ def available():
 
 
 class Desktop:
+    transport = "vnc"
+    width, height = WIDTH, HEIGHT
+
     def __init__(self, fs):
         self.fs = fs
         self.runtime = Path(tempfile.mkdtemp(prefix="open-terminal-desktop-"))
@@ -257,6 +266,10 @@ class Desktop:
 async def get_desktop(request, fs):
     if not available():
         raise HTTPException(503, "Desktop is not installed on this terminal")
+    if sys.platform in ("darwin", "win32"):
+        from open_terminal.desktop_native import check_permissions
+
+        check_permissions()
     key = (request.headers.get("x-user-id", ""), fs.home)
     async with startup_lock:
         desktop = desktops.get(key)
@@ -265,7 +278,12 @@ async def get_desktop(request, fs):
         if desktop:
             await desktop.close()
             desktops.pop(key, None)
-        desktop = Desktop(fs)
+        if sys.platform in ("darwin", "win32"):
+            from open_terminal.desktop_native import NativeDesktop
+
+            desktop = NativeDesktop(fs)
+        else:
+            desktop = Desktop(fs)
         try:
             await desktop.start()
         except BaseException:
@@ -284,8 +302,13 @@ async def shutdown():
 
 @router.get("", include_in_schema=False, dependencies=[Depends(verify_api_key)])
 async def connect(request: Request, fs: UserFS = Depends(get_filesystem)):
-    await get_desktop(request, fs)
-    return {"width": WIDTH, "height": HEIGHT}
+    desktop = await get_desktop(request, fs)
+    return {
+        "width": desktop.width,
+        "height": desktop.height,
+        "transport": desktop.transport,
+        "platform": sys.platform,
+    }
 
 
 @router.get(
@@ -296,8 +319,8 @@ async def connect(request: Request, fs: UserFS = Depends(get_filesystem)):
 )
 async def screenshot(request: Request, fs: UserFS = Depends(get_filesystem)):
     """Return an image of the same desktop the user sees. Starts it if needed.
-    Coordinates are pixels on a 1280x800 screen. Inspect before interacting.
-    Desktop applications and files belong to this terminal workspace.
+    Coordinates are pixels in the returned image. Inspect before interacting.
+    On macOS and Windows this is the logged-in user's primary screen, shared across workspaces.
     """
     desktop = await get_desktop(request, fs)
     async with desktop.actions:
@@ -306,10 +329,10 @@ async def screenshot(request: Request, fs: UserFS = Depends(get_filesystem)):
 
 class Action(BaseModel):
     action: Literal["click", "double_click", "move", "drag", "scroll", "type", "key"]
-    x: int | None = Field(default=None, ge=0, lt=WIDTH)
-    y: int | None = Field(default=None, ge=0, lt=HEIGHT)
-    to_x: int | None = Field(default=None, ge=0, lt=WIDTH)
-    to_y: int | None = Field(default=None, ge=0, lt=HEIGHT)
+    x: int | None = Field(default=None, ge=0, le=32767)
+    y: int | None = Field(default=None, ge=0, le=32767)
+    to_x: int | None = Field(default=None, ge=0, le=32767)
+    to_y: int | None = Field(default=None, ge=0, le=32767)
     button: Literal["left", "middle", "right"] = "left"
     direction: Literal["up", "down", "left", "right"] = "down"
     amount: int = Field(default=3, ge=1, le=20)
@@ -318,7 +341,7 @@ class Action(BaseModel):
         default=None,
         pattern=r"^[A-Za-z0-9_+]+$",
         max_length=100,
-        description="X11 key or combination, e.g. Return, ctrl+c, alt+F2, Super_L",
+        description="Key or combination, e.g. Return, ctrl+c, command+c on macOS, alt+F2 on Linux",
     )
 
     @model_validator(mode="after")
@@ -347,11 +370,25 @@ class Action(BaseModel):
 async def action(request: Request, body: Action, fs: UserFS = Depends(get_filesystem)):
     """Click, double-click, move, drag, scroll, type text or press a key combination.
     Returns a screenshot after the action. Use desktop_screenshot first and work
-    from its 1280x800 pixel coordinates. Shares the user's desktop and workspace.
-    Use alt+F2 to launch an application. Type pastes literal text into the focused application.
+    from that image's pixel coordinates. Native macOS/Windows share the logged-in desktop.
+    Type pastes literal text into the focused application.
     """
     desktop = await get_desktop(request, fs)
     async with desktop.actions:
+        for x, y in ((body.x, body.y), (body.to_x, body.to_y)):
+            if x is not None and (
+                x >= desktop.width or y is None or y >= desktop.height
+            ):
+                raise HTTPException(
+                    400, "Coordinates are outside the current desktop screenshot"
+                )
+        if desktop.transport == "frames":
+            try:
+                await asyncio.to_thread(desktop.perform, body)
+            except ValueError as exc:
+                raise HTTPException(400, str(exc)) from exc
+            await asyncio.sleep(0.1)
+            return Response(await desktop.screenshot(), media_type="image/png")
         button = str({"left": 1, "middle": 2, "right": 3}[body.button])
         if body.x is not None:
             await desktop.run(

open_terminal/desktop_native.py

@@ -0,0 +1,302 @@
+"""Capture and control the logged-in macOS or Windows desktop."""
+
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+import ctypes
+import io
+import sys
+import time
+
+from fastapi import HTTPException, WebSocketDisconnect
+from PIL import Image
+from pydantic import BaseModel, Field
+from typing import Literal
+
+from open_terminal.frames import Frames
+
+
+class Input(BaseModel):
+    type: Literal["mouse", "key", "paste", "release", "frameAck"]
+    x: int = Field(default=0, ge=0, le=32767)
+    y: int = Field(default=0, ge=0, le=32767)
+    buttons: int = Field(default=0, ge=0, le=7)
+    deltaX: int = Field(default=0, ge=-10000, le=10000)
+    deltaY: int = Field(default=0, ge=-10000, le=10000)
+    code: str = Field(default="", max_length=50)
+    down: bool = False
+    text: str = Field(default="", max_length=20000)
+
+
+KEYS = {
+    "Enter": "enter",
+    "NumpadEnter": "enter",
+    "Escape": "esc",
+    "Space": "space",
+    "Backspace": "backspace",
+    "Tab": "tab",
+    "Delete": "delete",
+    "Insert": "insert",
+    "Home": "home",
+    "End": "end",
+    "PageUp": "pageup",
+    "PageDown": "pagedown",
+    "ArrowLeft": "left",
+    "ArrowRight": "right",
+    "ArrowUp": "up",
+    "ArrowDown": "down",
+    "ShiftLeft": "shiftleft",
+    "ShiftRight": "shiftright",
+    "ControlLeft": "ctrlleft",
+    "ControlRight": "ctrlright",
+    "AltLeft": "altleft",
+    "AltRight": "altright",
+    "MetaLeft": "command" if sys.platform == "darwin" else "winleft",
+    "MetaRight": "command" if sys.platform == "darwin" else "winright",
+    "CapsLock": "capslock",
+    "Minus": "-",
+    "Equal": "=",
+    "BracketLeft": "[",
+    "BracketRight": "]",
+    "Backslash": "\\",
+    "Semicolon": ";",
+    "Quote": "'",
+    "Comma": ",",
+    "Period": ".",
+    "Slash": "/",
+    "Backquote": "`",
+    **{f"Key{chr(n)}": chr(n).lower() for n in range(65, 91)},
+    **{f"Digit{n}": str(n) for n in range(10)},
+    **{f"F{n}": f"f{n}" for n in range(1, 25)},
+}
+ALIASES = {
+    "return": "enter",
+    "escape": "esc",
+    "super_l": "command" if sys.platform == "darwin" else "win",
+    "cmd": "command",
+    "control": "ctrl",
+    "page_up": "pageup",
+    "page_down": "pagedown",
+}
+
+
+def check_permissions():
+    if sys.platform == "darwin":
+        graphics = ctypes.CDLL(
+            "/System/Library/Frameworks/CoreGraphics.framework/CoreGraphics"
+        )
+        graphics.CGPreflightScreenCaptureAccess.restype = ctypes.c_bool
+        access = ctypes.CDLL(
+            "/System/Library/Frameworks/ApplicationServices.framework/ApplicationServices"
+        )
+        access.AXIsProcessTrusted.restype = ctypes.c_bool
+        missing = []
+        if not graphics.CGPreflightScreenCaptureAccess():
+            missing.append("Screen Recording")
+        if not access.AXIsProcessTrusted():
+            missing.append("Accessibility")
+        if missing:
+            raise HTTPException(
+                403,
+                "Allow "
+                + " and ".join(missing)
+                + " for the app running Open Terminal in System Settings → Privacy & Security, then restart Open Terminal.",
+            )
+
+
+class NativeDesktop:
+    transport = "frames"
+
+    def __init__(self, fs):
+        if fs.username:
+            raise HTTPException(
+                403,
+                "Native desktop requires Open Terminal to run as the logged-in desktop user.",
+            )
+        self.actions = asyncio.Lock()
+        self.processes = []
+        self.peers = set()
+        self.worker = ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix="desktop-capture"
+        )
+        self.capture = None
+        self.width = self.height = 0
+
+    async def work(self, fn, *args):
+        return await asyncio.get_running_loop().run_in_executor(self.worker, fn, *args)
+
+    def setup(self):
+        check_permissions()
+        import mss
+        import pyautogui
+
+        self.input = pyautogui
+        self.input.PAUSE = 0
+        # A remote viewer must be able to use screen corners, like the VNC viewer.
+        self.input.FAILSAFE = False
+        self.capture = mss.mss()
+        self.grab()
+
+    async def start(self):
+        try:
+            await self.work(self.setup)
+        except HTTPException:
+            raise
+        except Exception as exc:
+            raise HTTPException(
+                503,
+                "Desktop capture requires a logged-in, unlocked graphical session: "
+                + str(exc),
+            ) from exc
+
+    def grab(self):
+        screen = next(
+            (m for m in self.capture.monitors[1:] if m["left"] == 0 and m["top"] == 0),
+            self.capture.monitors[1],
+        )
+        shot = self.capture.grab(screen)
+        image = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX")
+        self.screen_width, self.screen_height = self.input.size()
+        scale = min(1, 1920 / self.screen_width, 1080 / self.screen_height)
+        self.width, self.height = (
+            round(self.screen_width * scale),
+            round(self.screen_height * scale),
+        )
+        return image.resize((self.width, self.height), Image.Resampling.BILINEAR)
+
+    def position(self, x, y):
+        if not (0 <= x < self.width and 0 <= y < self.height):
+            raise ValueError("Coordinates are outside the current desktop screenshot")
+        self.input.moveTo(
+            round(x * self.screen_width / self.width),
+            round(y * self.screen_height / self.height),
+        )
+
+    def paste(self, text):
+        import pyperclip
+
+        pyperclip.copy(text)
+        self.input.hotkey("command" if sys.platform == "darwin" else "ctrl", "v")
+
+    def release(self, keys, buttons):
+        for key in list(keys):
+            self.input.keyUp(key)
+        keys.clear()
+        for button in list(buttons):
+            self.input.mouseUp(button=button)
+        buttons.clear()
+
+    def handle(self, message, keys, buttons):
+        if message.type == "mouse":
+            self.position(message.x, message.y)
+            for flag, button in ((1, "left"), (2, "right"), (4, "middle")):
+                down = bool(message.buttons & flag)
+                if down and button not in buttons:
+                    self.input.mouseDown(button=button)
+                    buttons.add(button)
+                elif not down and button in buttons:
+                    self.input.mouseUp(button=button)
+                    buttons.remove(button)
+            if message.deltaY:
+                self.input.scroll(-message.deltaY)
+            if message.deltaX:
+                self.input.hscroll(message.deltaX)
+        elif message.type == "key" and (key := KEYS.get(message.code)):
+            if message.down:
+                self.input.keyDown(key)
+                keys.add(key)
+            else:
+                self.input.keyUp(key)
+                keys.discard(key)
+        elif message.type == "paste":
+            self.release(keys, buttons)
+            self.paste(message.text)
+        elif message.type == "release":
+            self.release(keys, buttons)
+
+    def perform(self, body):
+        if body.x is not None:
+            self.position(body.x, body.y)
+        if body.action in ("click", "double_click"):
+            self.input.click(
+                button=body.button,
+                clicks=2 if body.action == "double_click" else 1,
+                interval=0.1 if body.action == "double_click" else 0,
+            )
+        elif body.action == "drag":
+            try:
+                self.input.mouseDown(button=body.button)
+                self.position(body.to_x, body.to_y)
+            finally:
+                self.input.mouseUp(button=body.button)
+        elif body.action == "scroll":
+            amount = body.amount * (-1 if body.direction in ("down", "left") else 1)
+            (
+                self.input.hscroll
+                if body.direction in ("left", "right")
+                else self.input.scroll
+            )(amount)
+        elif body.action == "type":
+            self.paste(body.text)
+        elif body.action == "key":
+            keys = [
+                ALIASES.get(key.lower(), key.lower()) for key in body.key.split("+")
+            ]
+            if any(key not in self.input.KEYBOARD_KEYS for key in keys):
+                raise ValueError("Unknown key combination")
+            self.input.hotkey(*keys)
+
+    async def screenshot(self):
+        def encode():
+            output = io.BytesIO()
+            self.grab().save(output, "PNG")
+            return output.getvalue()
+
+        return await self.work(encode)
+
+    async def stream(self, ws):
+        self.peers.add(ws)
+        ready = asyncio.Event()
+        ready.set()
+        keys, buttons = set(), set()
+        encoder = Frames()
+
+        async def frames():
+            while True:
+                await ready.wait()
+                start = time.monotonic()
+                packet = await self.work(lambda: encoder.encode(self.grab()))
+                if packet is not None:
+                    ready.clear()
+                    await ws.send_bytes(packet)
+                await asyncio.sleep(max(0, 1 / 30 - (time.monotonic() - start)))
+
+        async def commands():
+            while True:
+                message = Input.model_validate(await ws.receive_json())
+                if message.type == "frameAck":
+                    ready.set()
+                else:
+                    async with self.actions:
+                        await asyncio.to_thread(self.handle, message, keys, buttons)
+
+        tasks = [asyncio.create_task(frames()), asyncio.create_task(commands())]
+        try:
+            done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+            for task in done:
+                task.result()
+        except WebSocketDisconnect:
+            pass
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            async with self.actions:
+                await asyncio.to_thread(self.release, keys, buttons)
+            self.peers.discard(ws)
+
+    async def close(self):
+        for peer in list(self.peers):
+            await peer.close(code=1012)
+        if self.capture:
+            await self.work(self.capture.close)
+        self.worker.shutdown(wait=False, cancel_futures=True)

open_terminal/frames.py

@@ -0,0 +1,35 @@
+"""Bounded binary image updates shared by native desktops and Chrome."""
+
+import io
+import struct
+
+from PIL import Image, ImageChops
+
+
+class Frames:
+    def __init__(self):
+        self.previous = None
+
+    def encode(self, image, quality=75, original=None):
+        image = image.convert("RGB")
+        width, height = image.size
+        box = (0, 0, width, height)
+        if self.previous is not None and self.previous.size == image.size:
+            box = ImageChops.difference(self.previous, image).getbbox()
+            if box is None:
+                return None
+        self.previous = image
+        left, top, right, bottom = box
+        if original and (right - left) * (bottom - top) > width * height // 2:
+            # Reuse Chrome's JPEG for large updates instead of compressing it twice.
+            return struct.pack("!6H", width, height, 0, 0, width, height) + original
+        output = io.BytesIO()
+        image.crop(box).save(output, "JPEG", quality=quality)
+        return (
+            struct.pack("!6H", width, height, left, top, right - left, bottom - top)
+            + output.getvalue()
+        )
+
+    def jpeg(self, data, quality=75):
+        with Image.open(io.BytesIO(data)) as image:
+            return self.encode(image, quality, original=data)