diff --git a/cloud/playwright_browser.py b/cloud/playwright_browser.py index f5eaba0..2d1a77e 100644 --- a/cloud/playwright_browser.py +++ b/cloud/playwright_browser.py @@ -61,6 +61,16 @@ class Browser: self._browser = None self._page = None self._current_url = "" + self._last_title = "" + # ref snapshot state (playwright-mcp style): ref -> css path + self._ref_paths = {} + # screencast sink: a queue.Queue set by the remote service. + # The CDP frame callback only enqueues; the MCP main thread + # drains it (pump_screencast) because Playwright sync calls + # must run on the thread that created the page. + self._frame_sink = None + self._cdp = None + self._cdp_page = None def _ensure(self): if self._page is not None: @@ -92,8 +102,15 @@ class Browser: title = self._page.title() except Exception: # noqa: BLE001 - title is best-effort title = "" + self._last_title = title return {"url": self._current_url, "title": title} + def cached_info(self): + """Plain-attribute state, safe to read from any thread + (the remote HTTP thread must never call page.title()).""" + return {"url": self._current_url, "title": self._last_title, + "started": self._page is not None} + def navigate(self, url): if not url or not str(url).strip(): raise BrowserError("url is required") @@ -102,6 +119,7 @@ class Browser: self._page.goto(str(url), timeout=NAV_TIMEOUT_MS, wait_until="domcontentloaded") self._current_url = self._page.url + self._start_screencast() except Exception as exc: # noqa: BLE001 raise BrowserError("navigate failed: %s" % exc) result = self._info() @@ -148,6 +166,185 @@ class Browser: result["ok"] = True return result + def mouse_move(self, x, y): + try: + fx = float(x) + fy = float(y) + except (TypeError, ValueError): + raise BrowserError("mouse_move needs numeric x/y, got %r/%r" + % (x, y)) + if not (0 <= fx <= 1280 and 0 <= fy <= 800): + raise BrowserError( + "mouse_move out of viewport 1280x800: %s,%s" % (x, y)) + self._ensure() + try: + self._page.mouse.move(fx, fy) + except Exception as exc: # noqa: BLE001 + raise BrowserError("mouse_move failed: %s" % exc) + result = self._info() + result["ok"] = True + result["x"] = fx + result["y"] = fy + return result + + def key_press(self, key): + if not key or not str(key).strip(): + raise BrowserError("key is required") + self._ensure() + try: + self._page.keyboard.press(str(key)) + except Exception as exc: # noqa: BLE001 + raise BrowserError("key_press failed: %s" % exc) + result = self._info() + result["ok"] = True + result["key"] = str(key) + return result + + def type_text(self, text): + if text is None: + raise BrowserError("text is required") + self._ensure() + try: + self._page.keyboard.type(str(text)) + except Exception as exc: # noqa: BLE001 + raise BrowserError("type_text failed: %s" % exc) + result = self._info() + result["ok"] = True + return result + + def snapshot_refs(self): + """Accessibility-style snapshot with refs (playwright-mcp + style): interactive elements get a stable ref valid until + the next snapshot_refs call; click_ref(ref) clicks one. + The ref maps to a computed CSS path, so it survives + re-renders better than an index into a stale list.""" + self._ensure() + script = """ +() => { + const path = (el) => { + const parts = []; + while (el && el.nodeType === 1 && el !== document.body) { + let tag = el.tagName.toLowerCase(); + const parent = el.parentElement; + if (parent) { + const sibs = Array.from(parent.children).filter( + c => c.tagName === el.tagName); + if (sibs.length > 1) + tag += ':nth-of-type(' + (sibs.indexOf(el) + 1) + ')'; + } + parts.unshift(tag); + el = parent; + } + return parts.join(' > '); + }; + const out = []; + const els = document.querySelectorAll( + 'a, button, input, select, textarea, [role="button"], ' + + '[role="link"], [role="textbox"], [role="checkbox"], ' + + '[role="radio"], [role="combobox"]'); + els.forEach((el) => { + const r = el.getBoundingClientRect(); + if (r.width === 0 && r.height === 0) return; + const name = (el.getAttribute('aria-label') || + el.textContent || el.value || el.placeholder || '') + .trim().replace(/\\s+/g, ' ').slice(0, 80); + out.push({tag: el.tagName.toLowerCase(), + role: el.getAttribute('role') || '', + name: name, path: path(el)}); + }); + return out; +} +""" + try: + elements = self._page.evaluate(script) + except Exception as exc: # noqa: BLE001 + raise BrowserError("snapshot_refs failed: %s" % exc) + self._ref_paths = {} + items = [] + for i, el in enumerate(elements or []): + ref = "e%d" % i + self._ref_paths[ref] = el.get("path", "") + items.append({"ref": ref, "tag": el.get("tag", ""), + "role": el.get("role", ""), + "name": el.get("name", "")}) + result = self._info() + result["elements"] = items + result["count"] = len(items) + return result + + def click_ref(self, ref): + if not ref or ref not in self._ref_paths: + raise BrowserError( + "unknown ref %r: run browser_snapshot_refs first" % (ref,)) + self._ensure() + path = self._ref_paths[ref] + try: + self._page.locator(path).first.click(timeout=10000) + self._current_url = self._page.url + except Exception as exc: # noqa: BLE001 + raise BrowserError("click_ref failed: %s" % exc) + result = self._info() + result["ok"] = True + result["ref"] = ref + return result + + def set_frame_sink(self, sink): + """Remote service: frames arrive here via CDP callback.""" + self._frame_sink = sink + if self._page is not None: + self._start_screencast() + + def _start_screencast(self): + if self._frame_sink is None or self._page is None: + return + if self._cdp is not None and self._cdp_page is self._page: + return + try: + cdp = self._page.context.new_cdp_session(self._page) + + def _on_frame(params): + try: + self._frame_sink.put( + (params.get("sessionId", ""), + params.get("data", ""))) + except Exception: # noqa: BLE001 - callback must not die + pass + + cdp.on("Page.screencastFrame", _on_frame) + cdp.send("Page.startScreencast", + {"format": "jpeg", "quality": 60, + "maxWidth": 1280, "maxHeight": 800, + "everyNthFrame": 2}) + self._cdp = cdp + self._cdp_page = self._page + except Exception: # noqa: BLE001 - screencast is best-effort + self._cdp = None + self._cdp_page = None + + def pump_screencast(self): + """Main-thread only: store the latest frame and ack every + queued frame (no ack = CDP stops sending after the first). + Returns the latest frame data (base64 jpeg) or ''.""" + if self._frame_sink is None: + return "" + latest = "" + while True: + try: + sid, data = self._frame_sink.get_nowait() + except Exception: # noqa: BLE001 - queue empty + break + if data: + latest = data + if self._cdp is not None: + try: + self._cdp.send("Page.screencastFrameAck", + {"sessionId": sid}) + except Exception: # noqa: BLE001 - ack best-effort + pass + if latest: + self._latest_frame = latest + return getattr(self, "_latest_frame", "") + def click_at(self, x, y): """Click by viewport coordinates (CSS pixels, origin at the top-left of the 1280x800 viewport). This is the tool the @@ -226,6 +423,13 @@ class Browser: def close(self): errors = [] + if self._cdp is not None: + try: + self._cdp.send("Page.stopScreencast") + except Exception: # noqa: BLE001 - best-effort + pass + self._cdp = None + self._cdp_page = None for closer in (lambda: self._page and self._page.close(), lambda: self._browser and self._browser.close(), lambda: self._pw and self._pw.stop()): @@ -245,11 +449,14 @@ def ensure_codex_home_playwright(codex_home_dir, command_args=None, script_dir=None): """Seed playwright MCP files + config into a tenant codex-home. - Mirrors ``ensure_codex_home_imagegen``: copies - ``playwright_browser.py`` and ``playwright_mcp.py`` only when - missing (a tenant-modified script is never silently - overwritten, it is reported in ``script_differs``), and - appends ``[mcp_servers.playwright]`` only when absent. No + MCP scripts are SYSTEM files, not tenant data: a tenant never + edits them (user files live in the worktree, not codex-home), + so unlike the old never-overwrite rule, a differing script is + REFRESHED to the deployed version and reported in + ``script_refreshed``. Without this, every MCP feature ships + dead for existing projects: they keep the first-seeded script + forever and only brand-new projects get the update. + Appends ``[mcp_servers.playwright]`` only when absent. No secrets live here -- proxy vars arrive via ``ensure_playwright_mcp_env`` from worker proxy_env, and the browsers path is a fixed image path. @@ -261,7 +468,9 @@ def ensure_codex_home_playwright(codex_home_dir, command_args=None, here = os.path.abspath(script_dir or os.path.dirname(__file__)) script_changed = False script_differs = [] - for name in ("playwright_browser.py", "playwright_mcp.py"): + script_refreshed = [] + for name in ("playwright_browser.py", "playwright_mcp.py", + "playwright_remote.py"): src = os.path.join(here, name) dst = os.path.join(home, name) if not os.path.exists(src): @@ -275,7 +484,9 @@ def ensure_codex_home_playwright(codex_home_dir, command_args=None, with open(dst, "rb") as fh: dst_bytes = fh.read() if src_bytes != dst_bytes: - script_differs.append(name) + shutil.copyfile(src, dst) + script_refreshed.append(name) + script_changed = True continue config_path = os.path.join(home, "config.toml") if command_args is None: @@ -301,7 +512,8 @@ def ensure_codex_home_playwright(codex_home_dir, command_args=None, "script_path": os.path.join(home, "playwright_mcp.py"), "config_changed": config_changed, "script_changed": script_changed, - "script_differs": script_differs} + "script_differs": script_differs, + "script_refreshed": script_refreshed} def ensure_playwright_mcp_env(codex_home_dir, env): diff --git a/cloud/playwright_mcp.py b/cloud/playwright_mcp.py index 635b214..369ab93 100644 --- a/cloud/playwright_mcp.py +++ b/cloud/playwright_mcp.py @@ -147,6 +147,58 @@ _TOOLS = [ }, }, }, + { + "name": "browser_snapshot_refs", + "description": ( + "Accessibility-style snapshot: interactive elements " + "with stable refs (e.g. e3) plus tag, role and name. " + "Use browser_click_ref to click one; refs are valid " + "until the next snapshot_refs call." + ), + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "browser_click_ref", + "description": ( + "Click an element by its ref from browser_snapshot_refs. " + "More robust than a CSS selector across re-renders." + ), + "inputSchema": { + "type": "object", + "properties": {"ref": {"type": "string"}}, + "required": ["ref"], + }, + }, + { + "name": "browser_mouse_move", + "description": "Move the mouse to viewport coordinates (1280x800).", + "inputSchema": { + "type": "object", + "properties": { + "x": {"type": "number"}, + "y": {"type": "number"}, + }, + "required": ["x", "y"], + }, + }, + { + "name": "browser_key_press", + "description": "Press a key (Playwright key name, e.g. Enter, Tab).", + "inputSchema": { + "type": "object", + "properties": {"key": {"type": "string"}}, + "required": ["key"], + }, + }, + { + "name": "browser_type_text", + "description": "Type text at the current focus.", + "inputSchema": { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + }, { "name": "browser_screenshot", "description": ( @@ -178,6 +230,7 @@ def _err(code, message): _BROWSER = None +_REMOTE = None def _get_browser(browser=None): @@ -189,6 +242,25 @@ def _get_browser(browser=None): return _BROWSER +def _get_remote(browser=None): + """Remote control state, lazily created. Keyed by browser + identity so tests passing different fake browsers don't share + one holder flag.""" + global _REMOTE + target = _get_browser(browser) + if _REMOTE is not None and _REMOTE.browser is target: + return _REMOTE + try: + from playwright_remote import RemoteState + except ImportError: + try: + from cloud.playwright_remote import RemoteState + except ImportError: + return None + _REMOTE = RemoteState(target) + return _REMOTE + + def handle_request(request, browser=None): """Handle one JSON-RPC request dict. Returns a response dict or None for notifications (no id). Tests pass a fake browser.""" @@ -227,6 +299,14 @@ def handle_request(request, browser=None): name = params.get("name") args = params.get("arguments") or {} target = _get_browser(browser) + remote = _get_remote(browser) + if remote is not None and remote.get_holder() == "user": + return {"jsonrpc": "2.0", "id": req_id, + "result": {"content": [{"type": "text", + "text": "error: user holds " + "browser control; wait for " + "release"}], + "isError": True}} try: if name == "browser_navigate": result = target.navigate(args.get("url")) @@ -239,6 +319,16 @@ def handle_request(request, browser=None): args.get("text")) elif name == "browser_click_at": result = target.click_at(args.get("x"), args.get("y")) + elif name == "browser_mouse_move": + result = target.mouse_move(args.get("x"), args.get("y")) + elif name == "browser_key_press": + result = target.key_press(args.get("key")) + elif name == "browser_type_text": + result = target.type_text(args.get("text")) + elif name == "browser_snapshot_refs": + result = target.snapshot_refs() + elif name == "browser_click_ref": + result = target.click_ref(args.get("ref")) elif name == "browser_live_shot": result = target.live_shot( args.get("output_filename")) @@ -268,9 +358,75 @@ def handle_request(request, browser=None): "error": _err(-32601, "method not found: %r" % method)} +def _drain_remote(browser): + """Main-thread only: run queued user inputs and pump screencast + frames (acks). Called on every loop iteration so remote input + lands even while no JSON-RPC request is in flight.""" + remote = _get_remote(browser) + if remote is None: + return + try: + from playwright_remote import execute_input + except ImportError: + try: + from cloud.playwright_remote import execute_input + except ImportError: + return + while True: + try: + item = remote.inputs.get_nowait() + except Exception: # noqa: BLE001 - queue empty + break + if remote.get_holder() != "user": + continue + try: + execute_input(browser, item) + except Exception as exc: # noqa: BLE001 - input errors are data + print("playwright_mcp: remote input error: %s" % exc, + file=sys.stderr) + try: + frame = browser.pump_screencast() + except Exception: # noqa: BLE001 - screencast best-effort + frame = "" + if frame: + remote.set_frame(frame) + + def main(): - for line in sys.stdin: - line = line.strip() + import select + browser = _get_browser() + remote = _get_remote(browser) + if remote is not None: + try: + from playwright_remote import start_remote_server + except ImportError: + try: + from cloud.playwright_remote import start_remote_server + except ImportError: + start_remote_server = None + if start_remote_server is not None: + socket_path = os.environ.get( + "PLAYWRIGHT_REMOTE_SOCKET", + os.path.join(browser.cwd, ".browser-remote.sock")) + try: + browser.set_frame_sink(_frame_queue()) + start_remote_server(remote, socket_path) + except Exception as exc: # noqa: BLE001 - remote best-effort + print("playwright_mcp: remote server failed: %s" % exc, + file=sys.stderr) + buffer = "" + while True: + _drain_remote(browser) + try: + readable, _, _ = select.select([sys.stdin], [], [], 0.05) + except Exception: # noqa: BLE001 - stdin gone + break + if not readable: + continue + chunk = sys.stdin.readline() + if not chunk: + break + line = chunk.strip() if not line: continue try: @@ -282,7 +438,7 @@ def main(): sys.stdout.flush() continue try: - response = handle_request(request) + response = handle_request(request, browser) except Exception as exc: # noqa: BLE001 - server must not die print("playwright_mcp: handler error: %s" % exc, file=sys.stderr) @@ -296,5 +452,32 @@ def main(): sys.stdout.flush() +def _frame_queue(): + try: + from playwright_remote import SimpleQueue + except ImportError: + try: + from cloud.playwright_remote import SimpleQueue + except ImportError: + import collections + import threading + + class SimpleQueue: # noqa: D101 - last-resort fallback + def __init__(self): + self._deque = collections.deque() + self._lock = threading.Lock() + + def put(self, item): + with self._lock: + self._deque.append(item) + + def get_nowait(self): + with self._lock: + if not self._deque: + raise IndexError("queue empty") + return self._deque.popleft() + return SimpleQueue() + + if __name__ == "__main__": main() diff --git a/cloud/playwright_remote.py b/cloud/playwright_remote.py new file mode 100644 index 0000000..373beec --- /dev/null +++ b/cloud/playwright_remote.py @@ -0,0 +1,234 @@ +"""Remote-control HTTP service for the Playwright MCP server. + +Runs INSIDE the MCP process (same thread-ownership problem as the +browser itself) and listens on a Unix socket in the task worktree +(``.browser-remote.sock`` by default). Why a socket, not a TCP +port: Docker task containers are per-turn and --rm, so a published +port would need a mapping nobody can observe mid-run (the worker's +runner blocks); a socket file in the mounted worktree is visible +to the host control plane with zero mapping, zero conflicts, and +dies with the turn. + +Threading contract (measured 2026-10-30: Playwright sync calls +from any thread but the creating one raise "Cannot switch to a +different thread"): +- The HTTP thread NEVER calls page methods. It only reads plain + attributes (Browser.cached_info), the latest frame bytes, and + the holder flag, and it enqueues user inputs. +- The MCP main thread drains the input queue and pumps screencast + frames on every select() iteration (~50ms), so user input lands + within ~50ms even while the agent is idle waiting for its next + tool call, and CDP frame acks keep flowing (no ack = CDP stops + after the first frame). + +Control model: +- holder "agent" (default): agent tools work, HTTP input is 409. +- holder "user": the user has taken over (console Takeover + button). Agent browser_* tools fail fast with a plain error so + the agent stops fighting the user; the user releases with + POST /control {"holder": "agent"} and the agent resumes on the + next continue message. +- The socket is chmod 0666: the real gates are the worktree + directory permissions and the control plane's task-owner auth + on its proxy endpoints, not the socket bits. +""" + +import collections +import json +import os +import socket +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +SOCKET_NAME = ".browser-remote.sock" +STREAM_FPS = 5 + + +class SimpleQueue: + """Minimal thread-safe queue. Deliberately NOT stdlib ``queue``: + cloud/queue.py shadows it whenever cloud/ is on sys.path + (repo mode, tests), breaking ``queue.Queue`` at import time. + Flat-copy tenant runs don't have that file, but this class + works identically in both layouts.""" + + def __init__(self): + self._deque = collections.deque() + self._lock = threading.Lock() + + def put(self, item): + with self._lock: + self._deque.append(item) + + def get_nowait(self): + with self._lock: + if not self._deque: + raise IndexError("queue empty") + return self._deque.popleft() + + def empty(self): + with self._lock: + return not self._deque + + +class RemoteState: + """Shared between the MCP main thread and the HTTP thread.""" + + def __init__(self, browser): + self.browser = browser + self.holder = "agent" + self._lock = threading.Lock() + self.inputs = SimpleQueue() + self._frame_lock = threading.Lock() + self._frame = "" + + def set_holder(self, holder): + if holder not in ("agent", "user"): + raise ValueError("holder must be 'agent' or 'user'") + with self._lock: + self.holder = holder + return self.holder + + def get_holder(self): + with self._lock: + return self.holder + + def enqueue_input(self, item): + if self.get_holder() != "user": + raise PermissionError( + "user does not hold control (holder=%s); " + "take over first" % self.get_holder()) + self.inputs.put(item) + + def set_frame(self, data): + with self._frame_lock: + self._frame = data or "" + + def get_frame(self): + with self._frame_lock: + return self._frame + + def state_dict(self): + info = self.browser.cached_info() + info["holder"] = self.get_holder() + info["streaming"] = bool(self.get_frame()) + return info + + +def execute_input(browser, item): + """Main-thread only: run one queued user input.""" + kind = item.get("type") + if kind == "mouse_click": + return browser.click_at(item.get("x"), item.get("y")) + if kind == "mouse_move": + return browser.mouse_move(item.get("x"), item.get("y")) + if kind == "key": + return browser.key_press(item.get("key")) + if kind == "type": + return browser.type_text(item.get("text", "")) + raise ValueError("unknown input type %r" % (kind,)) + + +def _make_handler(state): + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): # noqa: D102 - quiet + pass + + def _send(self, code, obj): + body = json.dumps(obj, ensure_ascii=False).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _read_json(self): + try: + length = int(self.headers.get("Content-Length") or 0) + except ValueError: + length = 0 + raw = self.rfile.read(length) if length > 0 else b"" + try: + return json.loads(raw.decode("utf-8")) if raw else {} + except ValueError: + return None + + def do_GET(self): # noqa: N802 + if self.path == "/state": + self._send(200, state.state_dict()) + elif self.path == "/stream": + self._stream() + else: + self._send(404, {"error": "unknown path"}) + return + + def _stream(self): + import time + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + try: + while True: + frame = state.get_frame() + payload = json.dumps( + {"image": frame, + "holder": state.get_holder(), + "ts": int(time.time())}, + ensure_ascii=False) + self.wfile.write( + ("data: %s\n\n" % payload).encode("utf-8")) + self.wfile.flush() + time.sleep(1.0 / STREAM_FPS) + except (BrokenPipeError, ConnectionResetError): + pass + + def do_POST(self): # noqa: N802 + body = self._read_json() + if body is None: + self._send(400, {"error": "invalid json"}) + return + if self.path == "/control": + try: + holder = state.set_holder(body.get("holder")) + except ValueError as exc: + self._send(400, {"error": str(exc)}) + return + self._send(200, {"holder": holder}) + elif self.path == "/input": + try: + state.enqueue_input(body) + except PermissionError as exc: + self._send(409, {"error": str(exc)}) + return + self._send(200, {"queued": True}) + else: + self._send(404, {"error": "unknown path"}) + return + + return Handler + + +class UnixHTTPServer(ThreadingHTTPServer): + address_family = socket.AF_UNIX + daemon_threads = True + + def server_bind(self): + try: + os.unlink(self.server_address) + except OSError: + pass + super().server_bind() + try: + os.chmod(self.server_address, 0o666) + except OSError: + pass + + +def start_remote_server(state, socket_path): + """Start the Unix-socket HTTP service on a daemon thread.""" + server = UnixHTTPServer(socket_path, _make_handler(state)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server diff --git a/cloud/server.py b/cloud/server.py index c24b50f..04a309f 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -482,6 +482,55 @@ def _task_worktree_for_files(server, task): return None, (404, "no local worktree for this task") +def _browser_unix_call(socket_path, method, path, body=None): + """One HTTP request over a Unix socket (the MCP remote service). + + Returns (status, body_bytes). Raw sockets, not http.client: + http.client has no AF_UNIX support, and the SSE stream needs + line-by-line reads on an open connection anyway. + """ + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(10) + try: + sock.connect(socket_path) + req = ("%s %s HTTP/1.1\r\nHost: browser\r\n" + "Connection: close\r\n" % (method, path)) + if body is not None: + req += ("Content-Type: application/json\r\n" + "Content-Length: %d\r\n" % len(body)) + req += "\r\n" + sock.sendall(req.encode("utf-8")) + if body is not None: + sock.sendall(body) + buf = b"" + while b"\r\n\r\n" not in buf: + chunk = sock.recv(4096) + if not chunk: + break + buf += chunk + head, _, rest = buf.partition(b"\r\n\r\n") + try: + status = int(head.split(b" ", 2)[1]) + except (IndexError, ValueError): + return 502, b'{"error": "bad browser response"}' + length = None + for line in head.split(b"\r\n")[1:]: + if line.lower().startswith(b"content-length:"): + try: + length = int(line.split(b":", 1)[1].strip()) + except ValueError: + length = None + data = rest + while length is None or len(data) < length: + chunk = sock.recv(4096) + if not chunk: + break + data += chunk + return status, (data[:length] if length is not None else data) + finally: + sock.close() + + def _safe_worktree_path(worktree, rel): """worktree 内相对路径的安全解析：空串返回 worktree 本身； 绝对路径、.. 逃逸、符号链接逃逸一律返回 None。realpath 后 @@ -2314,6 +2363,21 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._task_timeline(user_id, rest[: -len("/timeline")]) elif rest.endswith("/files") and method == "GET": self._task_files(user_id, rest[: -len("/files")], query) + elif rest.endswith("/browser/state") and method == "GET": + self._browser_proxy(user_id, + rest[: -len("/browser/state")], + "GET", "/state", None) + elif rest.endswith("/browser/input") and method == "POST": + self._browser_proxy(user_id, + rest[: -len("/browser/input")], + "POST", "/input", self._read_body()) + elif rest.endswith("/browser/control") and method == "POST": + self._browser_proxy(user_id, + rest[: -len("/browser/control")], + "POST", "/control", self._read_body()) + elif rest.endswith("/browser/stream") and method == "GET": + self._browser_stream(user_id, + rest[: -len("/browser/stream")]) elif rest.endswith("/chat-history") and method == "GET": self._task_chat_history( user_id, rest[: -len("/chat-history")]) @@ -5054,6 +5118,80 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(200, {"task_id": task_id, "path": rel, "size": size, "content": content}) + def _browser_socket_for(self, user_id, task_id): + """Resolve a task's remote-browser Unix socket. + + Same auth + worktree resolution as /files. Socket missing + (turn ended, browser never started) is 503, not 404: the + session is transient by design, it will exist again next + turn. Returns (socket_path, None) or (None, (status, msg)). + """ + task, err = self._thread_task(user_id, task_id, "read") + if err is not None: + return None, err + worktree, werr = _task_worktree_for_files(self.server, task) + if werr is not None: + return None, werr + sock_path = os.path.join(worktree, ".browser-remote.sock") + if not os.path.exists(sock_path): + return None, (503, "browser session not active") + return sock_path, None + + def _browser_proxy(self, user_id, task_id, method, path, body): + """Forward one JSON request to the MCP remote service.""" + sock_path, err = self._browser_socket_for(user_id, task_id) + if err is not None: + self._send_json(err[0], {"error": err[1]}) + return + try: + status, data = _browser_unix_call(sock_path, method, path, + body) + except (OSError, ValueError) as exc: + self._send_json(503, {"error": "browser session unreachable: " + "%s" % exc}) + return + try: + obj = json.loads(data.decode("utf-8")) if data else {} + except ValueError: + self._send_json(502, {"error": "bad browser response"}) + return + self._send_json(status, obj) + + def _browser_stream(self, user_id, task_id): + """Relay the MCP SSE frame stream to the console client.""" + sock_path, err = self._browser_socket_for(user_id, task_id) + if err is not None: + self._send_json(err[0], {"error": err[1]}) + return + try: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(sock_path) + sock.sendall(b"GET /stream HTTP/1.1\r\nHost: browser\r\n\r\n") + buf = b"" + while b"\r\n\r\n" not in buf: + chunk = sock.recv(4096) + if not chunk: + raise OSError("browser closed stream") + buf += chunk + except OSError as exc: + self._send_json(503, {"error": "browser session unreachable: " + "%s" % exc}) + return + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + self._responded = True + try: + stream = sock.makefile("rb") + for line in stream: + self.wfile.write(line) + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError): + pass + finally: + sock.close() + def _task_chat_history(self, user_id, task_id): """GET /tasks/{id}/chat-history：真正的交替历史，一次性 按时间交错拼好返回，前端不再猜顺序。 diff --git a/cloud/skills/browser-automation/SKILL.md b/cloud/skills/browser-automation/SKILL.md index 384c306..521d1d4 100644 --- a/cloud/skills/browser-automation/SKILL.md +++ b/cloud/skills/browser-automation/SKILL.md @@ -34,6 +34,19 @@ screenshot 都针对同一页；要换页再 navigate。 `browser_click_at`，点完必须再 snapshot 一次确认变化，再调 `browser_live_shot` 刷新画面。 +## 远控接管（第二期） + +用户可以在控制台接管浏览器：接管期间（holder=user）你的 +所有 browser_* 工具都会直接报错“user holds browser control”， +不要重试、不要 close 浏览器，停下来等用户交还；交还后 +（holder 回到 agent）下一条继续消息会让你续跑，续跑前先 +snapshot 一次看清用户改动到哪了。 + +新工具：`browser_snapshot_refs`（带 ref 的元素清单，点 +`browser_click_ref` 比 CSS 选择器稳，页面重渲染后先重新 +snapshot_refs 再点）；`browser_mouse_move` / `browser_key_press` / +`browser_type_text`（键盘与鼠标，远控输入与 agent 工具同源）。 + ## 规则 1. 先 navigate，再 snapshot 看实际渲染结果；click / fill 后 diff --git a/cloud/tests/test_playwright.py b/cloud/tests/test_playwright.py index 97a9086..72f8cc3 100644 --- a/cloud/tests/test_playwright.py +++ b/cloud/tests/test_playwright.py @@ -25,6 +25,13 @@ class _FakeLocator: def inner_text(self, timeout=None): return self._text + @property + def first(self): + return self + + def click(self, timeout=None): + pass + class _FakeMouse: def __init__(self, page): @@ -33,6 +40,40 @@ class _FakeMouse: def click(self, x, y): self._page.calls.append(("mouse_click", x, y)) + def move(self, x, y): + self._page.calls.append(("mouse_move", x, y)) + + +class _FakeKeyboard: + def __init__(self, page): + self._page = page + + def press(self, key): + self._page.calls.append(("key_press", key)) + + def type(self, text): + self._page.calls.append(("key_type", text)) + + +class _FakeCDP: + def __init__(self): + self.sent = [] + self.handlers = {} + + def on(self, event, handler): + self.handlers[event] = handler + + def send(self, method, params=None): + self.sent.append((method, params or {})) + + +class _FakeContext: + def __init__(self, cdp): + self.cdp = cdp + + def new_cdp_session(self, page): + return self.cdp + class _FakePage: def __init__(self): @@ -40,6 +81,9 @@ class _FakePage: self.calls = [] self.shot_path = None self.mouse = _FakeMouse(self) + self.keyboard = _FakeKeyboard(self) + self.cdp = _FakeCDP() + self.context = _FakeContext(self.cdp) def title(self): return "Example Domain" @@ -60,6 +104,11 @@ class _FakePage: def evaluate(self, expression): self.calls.append(("evaluate", expression)) + if "querySelectorAll" in expression: + return [{"tag": "button", "role": "", "name": "GO", + "path": "button"}, + {"tag": "a", "role": "link", "name": "Home", + "path": "a"}] return 42 def screenshot(self, path=None): @@ -166,6 +215,50 @@ class BrowserTest(unittest.TestCase): with self.assertRaises(BrowserError): self.browser.click_at("left", 10) + def test_remote_input_methods(self): + self.browser.navigate("https://example.com") + out = self.browser.mouse_move(10, 20) + self.assertTrue(out["ok"]) + self.assertIn(("mouse_move", 10, 20), self.page.calls) + out = self.browser.key_press("Enter") + self.assertTrue(out["ok"]) + self.assertIn(("key_press", "Enter"), self.page.calls) + out = self.browser.type_text("hi") + self.assertTrue(out["ok"]) + self.assertIn(("key_type", "hi"), self.page.calls) + with self.assertRaises(BrowserError): + self.browser.mouse_move(9999, 10) + + def test_snapshot_refs_and_click_ref(self): + self.browser.navigate("https://example.com") + out = self.browser.snapshot_refs() + self.assertEqual(out["count"], 2) + self.assertEqual(out["elements"][0]["ref"], "e0") + self.assertEqual(out["elements"][0]["name"], "GO") + clicked = self.browser.click_ref("e0") + self.assertTrue(clicked["ok"]) + self.assertEqual(clicked["ref"], "e0") + self.assertIn(("locator", "button"), self.page.calls) + with self.assertRaises(BrowserError): + self.browser.click_ref("e99") + + def test_screencast_attach_and_pump(self): + from playwright_remote import SimpleQueue + sink = SimpleQueue() + self.browser.set_frame_sink(sink) + self.browser.navigate("https://example.com") + self.assertIn(("Page.startScreencast", + {"format": "jpeg", "quality": 60, + "maxWidth": 1280, "maxHeight": 800, + "everyNthFrame": 2}), + [(m, p) for m, p in self.page.cdp.sent]) + handler = self.page.cdp.handlers["Page.screencastFrame"] + handler({"sessionId": "s1", "data": "FRAMEDATA"}) + frame = self.browser.pump_screencast() + self.assertEqual(frame, "FRAMEDATA") + self.assertIn(("Page.screencastFrameAck", {"sessionId": "s1"}), + [(m, p) for m, p in self.page.cdp.sent]) + def test_live_shot_writes_fixed_file_and_status(self): self.browser.navigate("https://example.com") out = self.browser.live_shot() @@ -192,6 +285,22 @@ class FakeMCPBrowser: "path": "/w/browser-live.png", "bytes": 456, "url": "https://x/", "title": "T", "timestamp": 1} + def mouse_move(self, x, y): + return {"ok": True, "x": x, "y": y} + + def key_press(self, key): + return {"ok": True, "key": key} + + def type_text(self, text): + return {"ok": True} + + def snapshot_refs(self): + return {"elements": [{"ref": "e0", "tag": "button", + "role": "", "name": "GO"}], "count": 1} + + def click_ref(self, ref): + return {"ok": True, "ref": ref} + def snapshot(self): return {"url": "https://x/", "title": "T", "text": "hi", "truncated": False} @@ -223,6 +332,32 @@ class MCPTest(unittest.TestCase): self.assertIn("https://example.com", text) self.assertFalse(resp["result"]["isError"]) + def test_holder_user_blocks_agent_tools(self): + import playwright_mcp + fake = FakeMCPBrowser() + remote = playwright_mcp._get_remote(fake) + remote.set_holder("user") + try: + resp = handle_request( + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "browser_navigate", + "arguments": {"url": "https://example.com"}}}, + browser=fake) + self.assertTrue(resp["result"]["isError"]) + self.assertIn("holds browser control", + resp["result"]["content"][0]["text"]) + finally: + remote.set_holder("agent") + + def test_new_tools_listed(self): + listed = handle_request({"jsonrpc": "2.0", "id": 4, + "method": "tools/list"}) + names = [t["name"] for t in listed["result"]["tools"]] + for name in ("browser_snapshot_refs", "browser_click_ref", + "browser_mouse_move", "browser_key_press", + "browser_type_text"): + self.assertIn(name, names) + if __name__ == "__main__": unittest.main() diff --git a/cloud/tests/test_playwright_remote.py b/cloud/tests/test_playwright_remote.py new file mode 100644 index 0000000..f09b9ef --- /dev/null +++ b/cloud/tests/test_playwright_remote.py @@ -0,0 +1,156 @@ +"""Tests for the Playwright remote-control service (fake only). + +Covers: holder transitions + input gating, the Unix-socket HTTP +endpoints (state/control/input/stream), and the main-thread +input execution path. No real Chromium: the CDP/screencast path +is exercised through fakes, real-browser proof lives in the +production verification task. +""" + +import json +import os +import socket +import sys +import tempfile +import threading +import unittest +import urllib.request + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from playwright_remote import (RemoteState, execute_input, # noqa: E402 + start_remote_server) + + +class _StubBrowser: + def __init__(self): + self.calls = [] + + def cached_info(self): + return {"url": "https://example.com", "title": "T", + "started": True} + + def click_at(self, x, y): + self.calls.append(("click_at", x, y)) + return {"ok": True} + + def mouse_move(self, x, y): + self.calls.append(("mouse_move", x, y)) + return {"ok": True} + + def key_press(self, key): + self.calls.append(("key_press", key)) + return {"ok": True} + + def type_text(self, text): + self.calls.append(("type_text", text)) + return {"ok": True} + + +class RemoteStateTest(unittest.TestCase): + def test_holder_and_input_gating(self): + state = RemoteState(_StubBrowser()) + self.assertEqual(state.get_holder(), "agent") + with self.assertRaises(PermissionError): + state.enqueue_input({"type": "mouse_click", "x": 1, "y": 2}) + state.set_holder("user") + state.enqueue_input({"type": "mouse_click", "x": 10, "y": 20}) + item = state.inputs.get_nowait() + execute_input(state.browser, item) + self.assertIn(("click_at", 10, 20), state.browser.calls) + with self.assertRaises(ValueError): + state.set_holder("nobody") + with self.assertRaises(ValueError): + execute_input(state.browser, {"type": "teleport"}) + + def test_state_dict(self): + state = RemoteState(_StubBrowser()) + state.set_frame("abc") + d = state.state_dict() + self.assertEqual(d["holder"], "agent") + self.assertTrue(d["streaming"]) + self.assertEqual(d["url"], "https://example.com") + + +class HttpServiceTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.browser = _StubBrowser() + self.state = RemoteState(self.browser) + self.sock = os.path.join(self._tmp.name, "remote.sock") + self.server = start_remote_server(self.state, self.sock) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self._tmp.cleanup() + + def _unix_get(self, path): + import http.client + + class UnixConn(http.client.HTTPConnection): + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, + socket.SOCK_STREAM) + self.sock.connect(self._sock_path) + + conn = UnixConn("localhost", timeout=5) + conn._sock_path = self.sock + conn.request("GET", path) + resp = conn.getresponse() + return resp.status, json.loads(resp.read().decode("utf-8")) + + def _unix_post(self, path, obj): + import http.client + + class UnixConn(http.client.HTTPConnection): + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, + socket.SOCK_STREAM) + self.sock.connect(self._sock_path) + + conn = UnixConn("localhost", timeout=5) + conn._sock_path = self.sock + body = json.dumps(obj).encode("utf-8") + conn.request("POST", path, body=body, + headers={"Content-Type": "application/json"}) + resp = conn.getresponse() + return resp.status, json.loads(resp.read().decode("utf-8")) + + def test_state_control_input_flow(self): + status, body = self._unix_get("/state") + self.assertEqual(status, 200) + self.assertEqual(body["holder"], "agent") + status, body = self._unix_post( + "/input", {"type": "mouse_click", "x": 1, "y": 2}) + self.assertEqual(status, 409) + status, body = self._unix_post("/control", {"holder": "user"}) + self.assertEqual((status, body["holder"]), (200, "user")) + status, body = self._unix_post( + "/input", {"type": "key", "key": "Enter"}) + self.assertEqual((status, body["queued"]), (200, True)) + self.assertEqual(self.state.inputs.get_nowait()["key"], "Enter") + status, body = self._unix_post("/control", {"holder": "agent"}) + self.assertEqual(body["holder"], "agent") + + def test_stream_emits_latest_frame(self): + self.state.set_frame("FRAMEBYTES") + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.settimeout(5) + sock.connect(self.sock) + sock.sendall(b"GET /stream HTTP/1.1\r\nHost: x\r\n\r\n") + buf = b"" + while b"\n\n" not in buf: + buf += sock.recv(4096) + head, _, rest = buf.partition(b"\n\n") + self.assertIn(b"text/event-stream", head) + first = rest + while b"\n\n" not in first: + first += sock.recv(4096) + payload = json.loads(first.split(b"data: ", 1)[1].split(b"\n\n")[0]) + self.assertEqual(payload["image"], "FRAMEBYTES") + sock.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_server_browser_proxy.py b/cloud/tests/test_server_browser_proxy.py new file mode 100644 index 0000000..07abc10 --- /dev/null +++ b/cloud/tests/test_server_browser_proxy.py @@ -0,0 +1,140 @@ +"""Browser proxy endpoints: /tasks/{id}/browser/{state,input,control}. + +The control plane forwards to the MCP remote service over the Unix +socket in the task worktree. Here the socket server is the real +playwright_remote module with a stub browser; no Chromium, no +worker, no model. +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.playwright_remote import RemoteState, start_remote_server + from cloud.server import create_server +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.playwright_remote import RemoteState, start_remote_server + from cloud.server import create_server + + +class _StubBrowser: + def cached_info(self): + return {"url": "https://example.com", "title": "T", + "started": True} + + +class BrowserProxyTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.outsider_id, self.outsider_key = self.auth.create_user("o") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.worktrees = os.path.join(self.tmp.name, "host-worktrees") + os.makedirs(self.worktrees) + self.server.host_worktrees_dir = self.worktrees + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + self._task("task-b1") + wt = os.path.join(self.worktrees, "task-b1") + os.makedirs(wt) + self.remote = RemoteState(_StubBrowser()) + self.sock_server = start_remote_server( + self.remote, os.path.join(wt, ".browser-remote.sock")) + self._task("task-nosock") + os.makedirs(os.path.join(self.worktrees, "task-nosock")) + + def tearDown(self): + self.sock_server.shutdown() + self.sock_server.server_close() + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _task(self, task_id): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status)" + " VALUES(?, ?, ?, ?)", + (task_id, self.project["id"], "do it", "running")) + conn.commit() + finally: + conn.close() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def test_state_input_control_flow(self): + status, body = self._req( + "GET", "/tasks/task-b1/browser/state", key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["holder"], "agent") + self.assertEqual(body["url"], "https://example.com") + status, body = self._req( + "POST", "/tasks/task-b1/browser/input", + {"type": "mouse_click", "x": 1, "y": 2}, self.owner_key) + self.assertEqual(status, 409, body) + status, body = self._req( + "POST", "/tasks/task-b1/browser/control", + {"holder": "user"}, self.owner_key) + self.assertEqual((status, body["holder"]), (200, "user")) + status, body = self._req( + "POST", "/tasks/task-b1/browser/input", + {"type": "key", "key": "Enter"}, self.owner_key) + self.assertEqual((status, body["queued"]), (200, True)) + status, body = self._req( + "POST", "/tasks/task-b1/browser/control", + {"holder": "agent"}, self.owner_key) + self.assertEqual(body["holder"], "agent") + + def test_forbidden_and_no_session(self): + status, _ = self._req( + "GET", "/tasks/task-b1/browser/state", key=self.outsider_key) + self.assertEqual(status, 403) + status, body = self._req( + "GET", "/tasks/task-nosock/browser/state", key=self.owner_key) + self.assertEqual(status, 503, body) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 95f19c7..77ca0bd 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -2631,13 +2631,20 @@ class Worker: # 与 imagegen 同款：播种失败直接抛错、任务不派发 # ——静默无浏览器比 loud 失败更糟，尤其对已宣传 # “能开浏览器”的任务。代理与浏览器路径由 - # _prepare_tenant 的 env 块给。 - _ensure_playwright_mcp( + # _prepare_tenant 的 env 块给。MCP 脚本是系统文件： + # 播种器在版本变化时刷新老租户的旧脚本（见 + # ensure_codex_home_playwright），刷新记录打日志。 + _seeded = _ensure_playwright_mcp( dirs["codex_home"], command_args=[ "python3", "/var/lib/codex-app-server/codex-home/" "playwright_mcp.py"]) + if _seeded.get("script_refreshed"): + self.logger.info( + "playwright MCP scripts refreshed for project %s: %s", + task.get("project_id"), + ",".join(_seeded["script_refreshed"])) if self.seed_skills: # 技能播种失败绝不阻塞任务：技能是可选的流程 # 说明，不是 websearch 那种“已宣传就必须有”的