From 6bcdeb8da16529f8c76b2169870dd2f8d205a011 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 30 Oct 2026 07:28:32 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=B5=8F=E8=A7=88=E5=99=A8=E8=BF=9C?= =?UTF-8?q?=E6=8E=A7=E4=BC=A0=E8=BE=93=E5=B1=82=E6=94=B9=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E6=A1=A5=EF=BC=88Unix=20socket=20=E5=9C=A8=20runsc=20=E4=B8=8B?= =?UTF-8?q?=E5=AE=BF=E4=B8=BB=E4=B8=8D=E5=8F=AF=E8=A7=81=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因（2026-10-30 活体实证）：#181 的 .browser-remote.sock 假设 宿主能连容器内 socket，但生产 Docker 任务走 runsc/gVisor， 其 gofer 挂载下容器内建的 socket 只在容器内可见，宿主目录 里只有普通文件（browser-live.png 一直能透传）。后果：server 代理对真实任务一律 503，会话根本没活过。 修法：传输层改四个普通文件（worktree 内，原子写 + 追加日志）： - .browser-remote-state.json：MCP 每轮重写，mtime 即存活信号， server 判 30 秒无心跳为会话死亡（503）； - .browser-remote-frame.jpg：最新 screencast 帧（base64 文本）； - .browser-remote-input.jsonl：server 追加，MCP 按偏移量消费 完整行，执行时再判 holder，半行等下一轮； - .browser-remote-control.json：{holder, seq}，MCP 首轮即应用， 接管可先于浏览器启动。 - holder 竞态：control 写下后 state 文件约 50ms 才刷新，server 的 input/stream 改判 effective_holder（控制文件 mtime 新者胜）， 否则接管后立刻点会误 409。 - 线程契约与控制权模型不变；MCP 主循环仍是唯一触页线程。 - 播种刷新（#181）让老租户自动拿到新版脚本。 测试：socket 测试改文件桥测试（桥轮询/偏移量/半行/陈旧态/ effective_holder + server 代理全链），全量 1357 项通过（skipped 4）。 --- cloud/playwright_mcp.py | 83 ++--- cloud/playwright_remote.py | 414 +++++++++++++---------- cloud/server.py | 230 +++++++------ cloud/tests/test_playwright.py | 2 +- cloud/tests/test_playwright_bridge.py | 166 +++++++++ cloud/tests/test_playwright_remote.py | 156 --------- cloud/tests/test_server_browser_proxy.py | 87 +++-- 7 files changed, 628 insertions(+), 510 deletions(-) create mode 100644 cloud/tests/test_playwright_bridge.py delete mode 100644 cloud/tests/test_playwright_remote.py diff --git a/cloud/playwright_mcp.py b/cloud/playwright_mcp.py index 369ab93..cd31761 100644 --- a/cloud/playwright_mcp.py +++ b/cloud/playwright_mcp.py @@ -230,7 +230,7 @@ def _err(code, message): _BROWSER = None -_REMOTE = None +_BRIDGE = None def _get_browser(browser=None): @@ -242,23 +242,23 @@ def _get_browser(browser=None): return _BROWSER -def _get_remote(browser=None): - """Remote control state, lazily created. Keyed by browser +def _get_bridge(browser=None): + """File-bridge remote control, lazily created. Keyed by browser identity so tests passing different fake browsers don't share one holder flag.""" - global _REMOTE + global _BRIDGE target = _get_browser(browser) - if _REMOTE is not None and _REMOTE.browser is target: - return _REMOTE + if _BRIDGE is not None and _BRIDGE.browser is target: + return _BRIDGE try: - from playwright_remote import RemoteState + from playwright_remote import RemoteBridge except ImportError: try: - from cloud.playwright_remote import RemoteState + from cloud.playwright_remote import RemoteBridge except ImportError: return None - _REMOTE = RemoteState(target) - return _REMOTE + _BRIDGE = RemoteBridge(target, getattr(target, "cwd", None)) + return _BRIDGE def handle_request(request, browser=None): @@ -299,7 +299,7 @@ def handle_request(request, browser=None): name = params.get("name") args = params.get("arguments") or {} target = _get_browser(browser) - remote = _get_remote(browser) + remote = _get_bridge(browser) if remote is not None and remote.get_holder() == "user": return {"jsonrpc": "2.0", "id": req_id, "result": {"content": [{"type": "text", @@ -359,61 +359,28 @@ def handle_request(request, browser=None): def _drain_remote(browser): - """Main-thread only: run queued user inputs and pump screencast - frames (acks). Called on every loop iteration so remote input + """Main-thread only: poll the file bridge (control, inputs, + frame, state). 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: + bridge = _get_bridge(browser) + if bridge 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) + bridge.poll(browser) + except Exception as exc: # noqa: BLE001 - bridge best-effort + print("playwright_mcp: bridge poll error: %s" % exc, + file=sys.stderr) def main(): 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) + # Frame sink must be set before any navigate so pump_screencast + # has frames for the bridge to publish. No bridge (missing + # module) means no sink either, so frames are dropped instead + # of accumulating unboundedly in a queue nobody drains. + if _get_bridge(browser) is not None: + browser.set_frame_sink(_frame_queue()) buffer = "" while True: _drain_remote(browser) diff --git a/cloud/playwright_remote.py b/cloud/playwright_remote.py index 373beec..26f317b 100644 --- a/cloud/playwright_remote.py +++ b/cloud/playwright_remote.py @@ -1,47 +1,60 @@ -"""Remote-control HTTP service for the Playwright MCP server. +"""File-bridge remote control 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. +Why files, not a Unix socket (corrected 2026-10-30): the first +version of this module ran an HTTP service on ``.browser-remote.sock`` +in the worktree, assuming the host control plane could connect to +it. That assumption is false under runsc/gVisor, which is what +production Docker tasks run under: a socket created inside a +runsc container on a bind mount is visible inside the container +only -- the host never sees the file (verified live: the socket +listed inside the container, the host directory showed only the +regular file). Regular files DO propagate through the same mount +(browser-live.png has always been host-visible). So the bridge +is four regular files in the worktree, all written atomically +(temp file + os.replace) except the append-only input log: -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). +- ``.browser-remote-state.json``: {url, title, started, holder, + streaming, updated_at} -- rewritten by the MCP drain loop on + every iteration, so its mtime is the session-liveness signal. + The server treats a state file older than STATE_STALE_SECONDS + as a dead session (503). +- ``.browser-remote-frame.jpg``: latest screencast frame, base64 + text, rewritten only when a new frame arrives. +- ``.browser-remote-input.jsonl``: the server appends one JSON + object per line. The MCP tracks a byte offset and executes + complete lines only (a trailing partial line waits for the + next poll); a line is executed only if the holder is "user" + at execution time, otherwise it is dropped. +- ``.browser-remote-control.json``: {holder, seq} -- written by + the server on takeover/release; the MCP applies it when seq + changes. Writing control ahead of browser start is fine: the + MCP picks it up on its first poll. -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. +Threading contract (unchanged from the socket version, measured +2026-10-30: Playwright sync calls from any thread but the +creating one raise "Cannot switch to a different thread"): +only the MCP main thread touches the browser. It calls +RemoteBridge.poll() on every select() iteration (~50ms), so +control/input/frames flow even while the agent is idle, and CDP +frame acks keep flowing (no ack = CDP stops after the first +frame). + +Control model (unchanged): holder "agent" by default; holder +"user" while the console user has taken over, during which +agent browser_* tools fail fast with a plain error. """ import collections import json import os -import socket import threading -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import time -SOCKET_NAME = ".browser-remote.sock" -STREAM_FPS = 5 +STATE_NAME = ".browser-remote-state.json" +FRAME_NAME = ".browser-remote-frame.jpg" +INPUT_NAME = ".browser-remote-input.jsonl" +CONTROL_NAME = ".browser-remote-control.json" +STATE_STALE_SECONDS = 30 class SimpleQueue: @@ -70,52 +83,15 @@ class SimpleQueue: 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 _atomic_write(path, data): + tmp = path + ".tmp" + with open(tmp, "wb") as fh: + fh.write(data) + os.replace(tmp, path) def execute_input(browser, item): - """Main-thread only: run one queued user input.""" + """Main-thread only: run one user input.""" kind = item.get("type") if kind == "mouse_click": return browser.click_at(item.get("x"), item.get("y")) @@ -128,107 +104,205 @@ def execute_input(browser, item): raise ValueError("unknown input type %r" % (kind,)) -def _make_handler(state): - class Handler(BaseHTTPRequestHandler): - protocol_version = "HTTP/1.1" +class RemoteBridge: + """MCP-side file bridge. Main thread only (see module doc).""" - def log_message(self, *args): # noqa: D102 - quiet + def __init__(self, browser, workdir=None): + self.browser = browser + self.workdir = workdir or getattr(browser, "cwd", None) or "." + self.holder = "agent" + self._input_offset = 0 + self._control_seq = None + self._frame_written = False + + def _path(self, name): + return os.path.join(self.workdir, name) + + def get_holder(self): + return self.holder + + def set_holder(self, holder): + if holder not in ("agent", "user"): + raise ValueError("holder must be 'agent' or 'user'") + self.holder = holder + return self.holder + + def state_dict(self): + info = self.browser.cached_info() + info["holder"] = self.holder + info["streaming"] = self._frame_written + info["updated_at"] = time.time() + return info + + def _poll_control(self): + try: + with open(self._path(CONTROL_NAME), "rb") as fh: + raw = fh.read() + ctrl = json.loads(raw.decode("utf-8")) + except (OSError, ValueError): + return + seq = ctrl.get("seq") + if seq is None or seq == self._control_seq: + return + self._control_seq = seq + try: + self.set_holder(ctrl.get("holder")) + except ValueError: 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"}) + def _poll_inputs(self): + try: + with open(self._path(INPUT_NAME), "rb") as fh: + fh.seek(self._input_offset) + data = fh.read() + except OSError: 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() + if not data: + return + # Only complete lines (ending with \n) are consumed; a + # trailing partial line stays for the next poll. + lines = data.split(b"\n") + complete = lines[:-1] + consumed = sum(len(line) + 1 for line in complete) + self._input_offset += consumed + for raw in complete: + if not raw.strip(): + continue 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): + item = json.loads(raw.decode("utf-8")) + except ValueError: + continue + if self.holder != "user": + continue + try: + execute_input(self.browser, item) + except Exception as exc: # noqa: BLE001 - input errors + print("playwright_remote: input error: %s" % exc) + + def poll(self, browser=None): + """One drain-loop iteration: control, inputs, frame, state.""" + target = browser if browser is not None else self.browser + self._poll_control() + self._poll_inputs() + try: + frame = target.pump_screencast() + except Exception: # noqa: BLE001 - screencast best-effort + frame = "" + if frame: + try: + _atomic_write(self._path(FRAME_NAME), + frame.encode("utf-8")) + self._frame_written = True + except OSError: 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) + _atomic_write( + self._path(STATE_NAME), + json.dumps(self.state_dict(), + ensure_ascii=False).encode("utf-8")) 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 +def bridge_paths(workdir): + """Server-side helper: the four bridge file paths.""" + return { + "state": os.path.join(workdir, STATE_NAME), + "frame": os.path.join(workdir, FRAME_NAME), + "input": os.path.join(workdir, INPUT_NAME), + "control": os.path.join(workdir, CONTROL_NAME), + } + + +def read_state(workdir): + """Server-side: read + freshness-check the state file. + + Returns (state_dict, None), or (None, (status, message)): + missing file -> 503 session not active; stale (no MCP + heartbeat for STATE_STALE_SECONDS) -> 503; unparseable -> + 502 (atomic writes mean this should not happen). + """ + path = os.path.join(workdir, STATE_NAME) + try: + with open(path, "rb") as fh: + raw = fh.read() + except OSError: + return None, (503, "browser session not active") + try: + state = json.loads(raw.decode("utf-8")) + except ValueError: + return None, (502, "bad browser response") + updated = state.get("updated_at") or 0 + if time.time() - updated > STATE_STALE_SECONDS: + return None, (503, "browser session not active") + return state, None + + +def effective_holder(workdir, state): + """Server-side holder with the control race closed. + + The MCP rewrites the state file every ~50ms, so right after + a takeover the state file still shows the old holder. The + control file is the newer truth when its mtime is at or + after the state's updated_at; otherwise the state file + (already refreshed by a poll) wins. Both agree once the + MCP has polled, so this only closes the ~50ms window. + """ + try: + mtime = os.path.getmtime(os.path.join(workdir, CONTROL_NAME)) + with open(os.path.join(workdir, CONTROL_NAME), + "rb") as fh: + ctrl = json.loads(fh.read().decode("utf-8")) + except (OSError, ValueError): + return state.get("holder") + holder = ctrl.get("holder") + if holder not in ("agent", "user"): + return state.get("holder") + if mtime >= (state.get("updated_at") or 0): + return holder + return state.get("holder") + + +def validate_input(body): + """Server-side input shape check. Returns (item, None) or + (None, (status, message)). Bounds mirror Browser.click_at / + mouse_move so a bad coordinate is a 400, not a queued no-op.""" + if not isinstance(body, dict): + return None, (400, "input must be a JSON object") + kind = body.get("type") + if kind in ("mouse_click", "mouse_move"): + x, y = body.get("x"), body.get("y") + if (isinstance(x, bool) or isinstance(y, bool) + or not isinstance(x, int) or not isinstance(y, int)): + return None, (400, "x and y must be integers") + if not (0 <= x <= 1280 and 0 <= y <= 800): + return None, (400, "coordinates out of viewport") + return {"type": kind, "x": x, "y": y}, None + if kind == "key": + key = body.get("key") + if not isinstance(key, str) or not key: + return None, (400, "key must be a non-empty string") + return {"type": "key", "key": key}, None + if kind == "type": + text = body.get("text") + if not isinstance(text, str): + return None, (400, "text must be a string") + return {"type": "type", "text": text}, None + return None, (400, "unknown input type %r" % (kind,)) + + +def append_input(workdir, item): + path = os.path.join(workdir, INPUT_NAME) + with open(path, "ab") as fh: + fh.write((json.dumps(item, ensure_ascii=False) + "\n") + .encode("utf-8")) + + +def write_control(workdir, holder): + if holder not in ("agent", "user"): + raise ValueError("holder must be 'agent' or 'user'") + _atomic_write( + os.path.join(workdir, CONTROL_NAME), + json.dumps({"holder": holder, "seq": time.time_ns()}, + ensure_ascii=False).encode("utf-8")) + return holder diff --git a/cloud/server.py b/cloud/server.py index 04a309f..9a67689 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -482,55 +482,6 @@ 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 后 @@ -2364,17 +2315,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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) + self._browser_state(user_id, + rest[: -len("/browser/state")]) elif rest.endswith("/browser/input") and method == "POST": - self._browser_proxy(user_id, + self._browser_input(user_id, rest[: -len("/browser/input")], - "POST", "/input", self._read_body()) + 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()) + self._browser_control(user_id, + rest[: -len("/browser/control")], + self._read_body()) elif rest.endswith("/browser/stream") and method == "GET": self._browser_stream(user_id, rest[: -len("/browser/stream")]) @@ -5118,13 +5068,11 @@ 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. + def _browser_worktree_for(self, user_id, task_id): + """Auth + worktree resolution shared by browser endpoints. - 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)). + Same rules as /files. Returns (worktree, None) or + (None, (status, msg)). """ task, err = self._thread_task(user_id, task_id, "read") if err is not None: @@ -5132,65 +5080,147 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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 + return worktree, 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) + def _browser_body_dict(self, body): + """Parse a raw POST body (bytes from _read_body) into a + dict; None for empty/invalid JSON (handlers map it to + their own 400).""" + if isinstance(body, dict): + return body + if not body: + return None + try: + obj = json.loads(body.decode("utf-8")) + except (ValueError, UnicodeDecodeError): + return None + return obj if isinstance(obj, dict) else None + + def _browser_state(self, user_id, task_id): + """GET /tasks/{id}/browser/state: fresh bridge state file.""" + from cloud.playwright_remote import read_state + worktree, err = self._browser_worktree_for(user_id, task_id) if err is not None: self._send_json(err[0], {"error": err[1]}) return + state, serr = read_state(worktree) + if serr is not None: + self._send_json(serr[0], {"error": serr[1]}) + return + self._send_json(200, state) + + def _browser_input(self, user_id, task_id, body): + """POST /tasks/{id}/browser/input: append to the bridge log. + + Holder is enforced here (409 unless the state file says + "user") and again by the MCP at execution time, so a + release racing an append can only drop the input, never + execute it under the wrong holder. + """ + from cloud.playwright_remote import ( + append_input, effective_holder, read_state, validate_input) + worktree, err = self._browser_worktree_for(user_id, task_id) + if err is not None: + self._send_json(err[0], {"error": err[1]}) + return + state, serr = read_state(worktree) + if serr is not None: + self._send_json(serr[0], {"error": serr[1]}) + return + holder = effective_holder(worktree, state) + if holder != "user": + self._send_json(409, {"error": "user does not hold control " + "(holder=%s); take over first" + % holder}) + return + item, ierr = validate_input(self._browser_body_dict(body)) + if ierr is not None: + self._send_json(ierr[0], {"error": ierr[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: " + append_input(worktree, item) + except OSError as exc: + self._send_json(503, {"error": "browser input 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"}) + self._send_json(200, {"queued": True}) + + def _browser_control(self, user_id, task_id, body): + """POST /tasks/{id}/browser/control: write the control file. + + Works even before the browser starts: the MCP applies the + file on its first poll, so an early takeover sticks. + """ + from cloud.playwright_remote import write_control + worktree, err = self._browser_worktree_for(user_id, task_id) + if err is not None: + self._send_json(err[0], {"error": err[1]}) return - self._send_json(status, obj) + parsed = self._browser_body_dict(body) + holder = parsed.get("holder") if parsed else None + try: + holder = write_control(worktree, holder) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + except OSError as exc: + self._send_json(503, {"error": "browser control unreachable: " + "%s" % exc}) + return + self._send_json(200, {"holder": holder}) 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) + """GET /tasks/{id}/browser/stream: SSE from bridge files. + + Polls the frame file every 200ms and emits a data line + whenever its bytes change (plus a heartbeat with the + current holder at least every 5s). Ends the stream when + the state file goes stale, so a dead session reads as a + disconnect, not a frozen picture. + """ + import time + from cloud.playwright_remote import ( + bridge_paths, effective_holder, read_state) + worktree, err = self._browser_worktree_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}) + _state, serr = read_state(worktree) + if serr is not None: + self._send_json(serr[0], {"error": serr[1]}) 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 + paths = bridge_paths(worktree) + last_frame = None + last_heartbeat = 0.0 try: - stream = sock.makefile("rb") - for line in stream: - self.wfile.write(line) - self.wfile.flush() + while True: + state, serr = read_state(worktree) + if serr is not None: + break + try: + with open(paths["frame"], "rb") as fh: + frame = fh.read().decode("utf-8") + except (OSError, UnicodeDecodeError): + frame = "" + now = time.time() + if frame != last_frame or now - last_heartbeat >= 5: + last_frame = frame + last_heartbeat = now + payload = json.dumps( + {"image": frame, + "holder": effective_holder(worktree, state), + "ts": int(now)}, ensure_ascii=False) + self.wfile.write( + ("data: %s\n\n" % payload).encode("utf-8")) + self.wfile.flush() + time.sleep(0.2) 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/tests/test_playwright.py b/cloud/tests/test_playwright.py index 72f8cc3..f6ec81a 100644 --- a/cloud/tests/test_playwright.py +++ b/cloud/tests/test_playwright.py @@ -335,7 +335,7 @@ class MCPTest(unittest.TestCase): def test_holder_user_blocks_agent_tools(self): import playwright_mcp fake = FakeMCPBrowser() - remote = playwright_mcp._get_remote(fake) + remote = playwright_mcp._get_bridge(fake) remote.set_holder("user") try: resp = handle_request( diff --git a/cloud/tests/test_playwright_bridge.py b/cloud/tests/test_playwright_bridge.py new file mode 100644 index 0000000..5c5b223 --- /dev/null +++ b/cloud/tests/test_playwright_bridge.py @@ -0,0 +1,166 @@ +"""File-bridge remote control tests (no Chromium, no sockets). + +The bridge is four regular files in the worktree because runsc +containers cannot publish a Unix socket to the host (verified +2026-10-30). These tests drive RemoteBridge.poll() with a stub +browser and plain files. +""" + +import json +import os +import sys +import tempfile +import time +import unittest +from pathlib import Path + +try: + from playwright_remote import ( + RemoteBridge, append_input, effective_holder, read_state, + validate_input, write_control) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.playwright_remote import ( + RemoteBridge, append_input, effective_holder, read_state, + validate_input, write_control) + + +class _StubBrowser: + def __init__(self): + self.calls = [] + self.frame = "" + + 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", key)) + return {"ok": True} + + def type_text(self, text): + self.calls.append(("type", text)) + return {"ok": True} + + def pump_screencast(self): + return self.frame + + +class BridgeTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.browser = _StubBrowser() + self.bridge = RemoteBridge(self.browser, self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_control_and_input_flow(self): + # Agent holder: inputs are dropped, not executed. + append_input(self.tmp.name, + {"type": "mouse_click", "x": 10, "y": 20}) + self.bridge.poll() + self.assertEqual(self.browser.calls, []) + # Takeover via control file, then a fresh input executes. + write_control(self.tmp.name, "user") + self.bridge.poll() + self.assertEqual(self.bridge.get_holder(), "user") + append_input(self.tmp.name, + {"type": "mouse_click", "x": 30, "y": 40}) + self.bridge.poll() + self.assertEqual(self.browser.calls, [("click_at", 30, 40)]) + # Second poll re-executes nothing (offset tracking). + self.bridge.poll() + self.assertEqual(self.browser.calls, [("click_at", 30, 40)]) + # Release drops later inputs until the next takeover. + write_control(self.tmp.name, "agent") + self.bridge.poll() + append_input(self.tmp.name, {"type": "key", "key": "Enter"}) + self.bridge.poll() + self.assertEqual(self.browser.calls, [("click_at", 30, 40)]) + + def test_partial_line_waits(self): + write_control(self.tmp.name, "user") + self.bridge.poll() + with open(os.path.join(self.tmp.name, + ".browser-remote-input.jsonl"), + "ab") as fh: + fh.write(b'{"type": "key", "key": "En') + self.bridge.poll() + self.assertEqual(self.browser.calls, []) + with open(os.path.join(self.tmp.name, + ".browser-remote-input.jsonl"), + "ab") as fh: + fh.write(b'ter"}\n') + self.bridge.poll() + self.assertEqual(self.browser.calls, [("key", "Enter")]) + + def test_state_and_frame_files(self): + self.browser.frame = "FRAMEBYTES" + self.bridge.poll() + state, err = read_state(self.tmp.name) + self.assertIsNone(err) + self.assertEqual(state["holder"], "agent") + self.assertTrue(state["streaming"]) + self.assertGreater(state["updated_at"], time.time() - 5) + with open(os.path.join(self.tmp.name, + ".browser-remote-frame.jpg"), + "rb") as fh: + self.assertEqual(fh.read(), b"FRAMEBYTES") + + def test_read_state_missing_and_stale(self): + state, err = read_state(self.tmp.name) + self.assertEqual(err[0], 503) + self.bridge.poll() + path = os.path.join(self.tmp.name, ".browser-remote-state.json") + with open(path, "rb") as fh: + doc = json.loads(fh.read()) + doc["updated_at"] = time.time() - 120 + with open(path, "wb") as fh: + fh.write(json.dumps(doc).encode()) + state, err = read_state(self.tmp.name) + self.assertEqual(err[0], 503) + + def test_effective_holder_control_race(self): + # State file says agent; a newer control file wins. + self.bridge.poll() + state, _ = read_state(self.tmp.name) + self.assertEqual(effective_holder(self.tmp.name, state), + "agent") + write_control(self.tmp.name, "user") + self.assertEqual(effective_holder(self.tmp.name, state), + "user") + # Once the MCP polls, the state file catches up and both + # agree again. + self.bridge.poll() + state, _ = read_state(self.tmp.name) + self.assertEqual(state["holder"], "user") + self.assertEqual(effective_holder(self.tmp.name, state), + "user") + + def test_validate_input(self): + item, err = validate_input({"type": "mouse_click", "x": 1, + "y": 2}) + self.assertIsNone(err) + _, err = validate_input({"type": "mouse_click", "x": 2000, + "y": 2}) + self.assertEqual(err[0], 400) + _, err = validate_input({"type": "mouse_click", "x": 1.5, + "y": 2}) + self.assertEqual(err[0], 400) + _, err = validate_input({"type": "teleport"}) + self.assertEqual(err[0], 400) + _, err = validate_input("nope") + self.assertEqual(err[0], 400) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_playwright_remote.py b/cloud/tests/test_playwright_remote.py deleted file mode 100644 index f09b9ef..0000000 --- a/cloud/tests/test_playwright_remote.py +++ /dev/null @@ -1,156 +0,0 @@ -"""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 index 07abc10..ade345b 100644 --- a/cloud/tests/test_server_browser_proxy.py +++ b/cloud/tests/test_server_browser_proxy.py @@ -1,9 +1,10 @@ -"""Browser proxy endpoints: /tasks/{id}/browser/{state,input,control}. +"""Browser endpoints over the file bridge: /tasks/{id}/browser/ +{state,input,control,stream}. -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. +The control plane never talks to the MCP process directly: it +reads/writes the bridge files in the task worktree (runsc +containers cannot publish a Unix socket to the host). Here the +worktree files are written by hand; no Chromium, no worker. """ import http.client @@ -13,26 +14,19 @@ import sqlite3 import sys import tempfile import threading +import time 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() @@ -52,17 +46,13 @@ class BrowserProxyTestCase(unittest.TestCase): "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.wt = os.path.join(self.worktrees, "task-b1") + os.makedirs(self.wt) + self._write_state("agent") 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) @@ -72,6 +62,14 @@ class BrowserProxyTestCase(unittest.TestCase): pass self.tmp.cleanup() + def _write_state(self, holder, fresh=True): + doc = {"url": "https://example.com", "title": "T", + "started": True, "holder": holder, + "streaming": False, + "updated_at": time.time() if fresh else time.time() - 120} + Path(self.wt, ".browser-remote-state.json").write_text( + json.dumps(doc), encoding="utf-8") + def _task(self, task_id): conn = sqlite3.connect(self.db_path) try: @@ -83,8 +81,9 @@ class BrowserProxyTestCase(unittest.TestCase): finally: conn.close() - def _req(self, method, path, body=None, key=None): - conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + def _req(self, method, path, body=None, key=None, raw=False): + conn = http.client.HTTPConnection("127.0.0.1", self.port, + timeout=10) hdrs = {} payload = None if body is not None: @@ -98,6 +97,8 @@ class BrowserProxyTestCase(unittest.TestCase): data = resp.read() finally: conn.close() + if raw: + return resp.status, data try: parsed = json.loads(data.decode("utf-8")) if data else None except ValueError: @@ -109,31 +110,67 @@ class BrowserProxyTestCase(unittest.TestCase): "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/input", + {"type": "mouse_click", "x": 9999, "y": 2}, self.owner_key) + self.assertEqual(status, 409, body) # holder checked first status, body = self._req( "POST", "/tasks/task-b1/browser/control", {"holder": "user"}, self.owner_key) self.assertEqual((status, body["holder"]), (200, "user")) + ctrl = json.loads(Path( + self.wt, ".browser-remote-control.json").read_text()) + self.assertEqual(ctrl["holder"], "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)) + lines = Path(self.wt, + ".browser-remote-input.jsonl").read_text() + self.assertEqual(json.loads(lines.strip()), + {"type": "key", "key": "Enter"}) + status, body = self._req( + "POST", "/tasks/task-b1/browser/input", + {"type": "mouse_click", "x": 9999, "y": 2}, self.owner_key) + self.assertEqual(status, 400, body) 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): + def test_forbidden_stale_and_no_session(self): status, _ = self._req( "GET", "/tasks/task-b1/browser/state", key=self.outsider_key) self.assertEqual(status, 403) + self._write_state("agent", fresh=False) status, body = self._req( - "GET", "/tasks/task-nosock/browser/state", key=self.owner_key) + "GET", "/tasks/task-b1/browser/state", key=self.owner_key) self.assertEqual(status, 503, body) + status, body = self._req( + "GET", "/tasks/task-nosock/browser/state", + key=self.owner_key) + self.assertEqual(status, 503, body) + + def test_stream_emits_frame(self): + Path(self.wt, ".browser-remote-frame.jpg").write_bytes(b"JPEG1") + self._write_state("agent") + conn = http.client.HTTPConnection("127.0.0.1", self.port, + timeout=10) + try: + conn.request("GET", "/tasks/task-b1/browser/stream", + headers={"X-API-Key": self.owner_key}) + resp = conn.getresponse() + self.assertEqual(resp.status, 200) + line = resp.readline().decode("utf-8") + self.assertTrue(line.startswith("data: "), line) + payload = json.loads(line[6:]) + self.assertEqual(payload["image"], "JPEG1") + finally: + conn.close() if __name__ == "__main__":