"""Headless Chromium automation for Codex tasks, via Playwright. One Browser object owns one page for the lifetime of an MCP process. It is deliberately small: navigate, snapshot (page text), click, fill, evaluate, screenshot, close. Anything fancier the agent can do with the shell tool, which has the same sandbox privileges as this browser (tenant Docker container, egress proxy whitelist). Browser launch facts that matter in production (verified 2026-10-30 in a runsc + user 1003 + docker-codex AppArmor container, the same shape as a production Docker task): - Chromium must run --no-sandbox (rootless container user, no userns sandbox) and --disable-dev-shm-usage (small /dev/shm). - Browsers live at PLAYWRIGHT_BROWSERS_PATH=/opt/ms-playwright in the image (the build installs them there, NOT in a home directory, because the container user has no home and the build ran as root). - Network goes through the egress proxy when HTTP_PROXY / HTTPS_PROXY is set in the MCP env (worker writes it from proxy_env, same as websearch): Playwright gets an explicit proxy server, because Chromium does not reliably pick up env proxies on Linux. Domains outside the Tinyproxy whitelist then fail with a proxy 403, which the tool reports plainly instead of bypassing. - Screenshots and downloads land in the working directory (the task worktree), never outside it. """ import json import os import time BROWSERS_PATH = "/opt/ms-playwright" SNAPSHOT_LIMIT = 8000 NAV_TIMEOUT_MS = 60000 class BrowserError(RuntimeError): pass def _validate_basename(name, what): if (not name or os.path.basename(name) != name or name in (".", "..") or "/" in name or "\\" in name): raise BrowserError( "%s must be a plain basename, got %r" % (what, name)) return name class Browser: """A single Chromium page. ``browser_factory`` is injectable: tests pass a fake returning a fake sync_playwright object.""" def __init__(self, cwd=None, proxy=None, browser_factory=None): self.cwd = os.path.abspath(cwd or os.getcwd()) self.proxy = (proxy or os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "") self._factory = browser_factory self._pw = None 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: return try: if self._factory is not None: pw = self._factory() if hasattr(pw, "start"): pw = pw.start() else: from playwright.sync_api import sync_playwright pw = sync_playwright().start() # Extra flags skip first-run/setup work only; they # change nothing about page rendering or tool # behaviour. Measured 2026-10-30 (runsc container, # launch + new_page, 3 runs each, alternating): # old median 0.889s, new median 0.819s -- ~70ms, # inside run-to-run noise (0.70-0.99s either way). # Kept anyway: zero cost, directionally faster. launch = {"args": ["--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu", "--no-first-run", "--no-default-browser-check", "--disable-default-apps", "--disable-extensions", "--disable-background-networking", "--disable-sync", "--disable-translate"]} if self.proxy: launch["proxy"] = {"server": self.proxy} browser = pw.chromium.launch(**launch) page = browser.new_page( viewport={"width": 1280, "height": 800}) except Exception as exc: # noqa: BLE001 - surfaced as tool error raise BrowserError("browser launch failed: %s" % exc) self._pw = pw self._browser = browser self._page = page def _info(self): try: 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") self._ensure() try: self._page.goto(str(url), timeout=NAV_TIMEOUT_MS, wait_until="domcontentloaded") self._current_url = self._page.url self._start_screencast() self._eager_frame() except Exception as exc: # noqa: BLE001 raise BrowserError("navigate failed: %s" % exc) result = self._info() result["ok"] = True return result def snapshot(self): self._ensure() try: text = self._page.locator("body").inner_text(timeout=10000) except Exception as exc: # noqa: BLE001 raise BrowserError("snapshot failed: %s" % exc) truncated = len(text) > SNAPSHOT_LIMIT if truncated: text = text[:SNAPSHOT_LIMIT] result = self._info() result["text"] = text result["truncated"] = truncated return result def click(self, selector): if not selector or not str(selector).strip(): raise BrowserError("selector is required") self._ensure() try: self._page.click(str(selector), timeout=10000) self._current_url = self._page.url except Exception as exc: # noqa: BLE001 raise BrowserError("click failed: %s" % exc) result = self._info() result["ok"] = True return result def fill(self, selector, text): if not selector or not str(selector).strip(): raise BrowserError("selector is required") self._ensure() try: self._page.fill(str(selector), "" if text is None else str(text), timeout=10000) except Exception as exc: # noqa: BLE001 raise BrowserError("fill failed: %s" % exc) result = self._info() 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) # everyNthFrame 1: full frame rate during motion. # Static pages cost nothing extra (frames only arrive # on repaint, and the bridge writes only changed # bytes), so dropping the frame-skip buys # responsiveness exactly when it matters -- scrolling, # dragging, animation -- without a static-page tax. # quality 50 (was 60): measured 2026-10-30 on a # text+gradient page, 62068 -> 56428 bytes (-9%). # Less than the 30% rule of thumb, but free: same # 1280x800 pixels, so click coordinates are # unaffected. Revert to 60 if text looks soft. cdp.send("Page.startScreencast", {"format": "jpeg", "quality": 50, "maxWidth": 1280, "maxHeight": 800, "everyNthFrame": 1}) self._cdp = cdp self._cdp_page = self._page except Exception: # noqa: BLE001 - screencast is best-effort self._cdp = None self._cdp_page = None def _eager_frame(self): """Push one current-viewport frame into the sink, best-effort. Screencast frames only arrive on repaint, and the page may already have painted before startScreencast ran -- so a static page would otherwise show nothing until its next repaint. Capturing the viewport once right after navigate guarantees the bridge file (and any open stream) has a picture within milliseconds of the page loading. """ if self._cdp is None or self._frame_sink is None: return try: result = self._cdp.send( "Page.captureScreenshot", {"format": "jpeg", "quality": 50}) data = result.get("data") if isinstance(result, dict) \ else None if data: # Empty session id: pump stores it but must not ack # it (there is no screencast frame to acknowledge). self._frame_sink.put(("", data)) except Exception: # noqa: BLE001 - screencast still flows pass def pump_screencast(self): """Main-thread only: store the latest frame and ack every real screencast frame (no ack = CDP stops sending after the first). Eager frames carry no session id and skip ack. 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 and sid: 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 console live view drives: the user clicks on a screenshot and the coordinates arrive here. Bounds-checked so a mis-scaled click cannot scroll the page into nowhere.""" try: fx = float(x) fy = float(y) except (TypeError, ValueError): raise BrowserError("click_at needs numeric x/y, got %r/%r" % (x, y)) if not (0 <= fx <= 1280 and 0 <= fy <= 800): raise BrowserError( "click_at out of viewport 1280x800: %s,%s" % (x, y)) self._ensure() try: self._page.mouse.click(fx, fy) self._current_url = self._page.url except Exception as exc: # noqa: BLE001 raise BrowserError("click_at failed: %s" % exc) result = self._info() result["ok"] = True result["x"] = fx result["y"] = fy return result def live_shot(self, output_filename=None): """Screenshot to a FIXED live-view file plus a JSON status sidecar, so a watcher (console, user) can poll one stable path instead of chasing per-shot filenames. The agent should call this after every navigate/click/fill while a live view is wanted. Returns path, bytes, url, title and a unix timestamp.""" self._ensure() name = _validate_basename( output_filename or "browser-live.png", "output_filename") path = os.path.join(self.cwd, name) try: self._page.screenshot(path=path) except Exception as exc: # noqa: BLE001 raise BrowserError("live_shot failed: %s" % exc) info = self._info() status = {"image": name, "path": path, "bytes": os.path.getsize(path), "url": info["url"], "title": info["title"], "timestamp": int(time.time())} status_path = os.path.join(self.cwd, "browser-live.json") with open(status_path, "w", encoding="utf-8") as fh: json.dump(status, fh, ensure_ascii=False) status["status_path"] = status_path return status def evaluate(self, expression): if not expression or not str(expression).strip(): raise BrowserError("expression is required") self._ensure() try: value = self._page.evaluate(str(expression)) except Exception as exc: # noqa: BLE001 raise BrowserError("evaluate failed: %s" % exc) return {"result": str(value)[:4000]} def screenshot(self, output_filename=None): self._ensure() name = _validate_basename( output_filename or ("shot-%d.png" % int(time.time())), "output_filename") path = os.path.join(self.cwd, name) try: self._page.screenshot(path=path) except Exception as exc: # noqa: BLE001 raise BrowserError("screenshot failed: %s" % exc) return {"path": path, "filename": name, "bytes": os.path.getsize(path)} 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()): try: closer() except Exception as exc: # noqa: BLE001 - close is best-effort errors.append(str(exc)) self._page = None self._browser = None self._pw = None return {"closed": True, "errors": errors} # ------------------------------------------------------- codex-home glue def ensure_codex_home_playwright(codex_home_dir, command_args=None, script_dir=None): """Seed playwright MCP files + config into a tenant codex-home. 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. """ import shutil home = os.path.abspath(codex_home_dir) os.makedirs(home, exist_ok=True) here = os.path.abspath(script_dir or os.path.dirname(__file__)) script_changed = False script_differs = [] 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): continue if not os.path.exists(dst): shutil.copyfile(src, dst) script_changed = True else: with open(src, "rb") as fh: src_bytes = fh.read() with open(dst, "rb") as fh: dst_bytes = fh.read() if src_bytes != dst_bytes: 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: command_args = ["python3", os.path.join(home, "playwright_mcp.py")] section = ("\n[mcp_servers.playwright]\n" "command = %s\n" "args = [%s]\n" % (json.dumps(command_args[0]), ", ".join(json.dumps(a) for a in command_args[1:]))) config_changed = False existing = "" if os.path.exists(config_path): with open(config_path, encoding="utf-8") as fh: existing = fh.read() if "[mcp_servers.playwright]" not in existing: with open(config_path, "a", encoding="utf-8") as fh: if existing and not existing.endswith("\n"): fh.write("\n") fh.write(section) config_changed = True return {"config_path": config_path, "script_path": os.path.join(home, "playwright_mcp.py"), "config_changed": config_changed, "script_changed": script_changed, "script_differs": script_differs, "script_refreshed": script_refreshed} def ensure_playwright_mcp_env(codex_home_dir, env): """Write [mcp_servers.playwright.env] into a tenant config.toml. Same rationale as the websearch/imagegen env writers: Codex's MCP subprocess does not inherit the Codex process env, so the proxy vars and PLAYWRIGHT_BROWSERS_PATH must live in config.toml. Idempotent in-place replacement of the env subsection; other sections untouched. Returns True when the file was written. """ if not env: return False home = os.path.abspath(codex_home_dir) config_path = os.path.join(home, "config.toml") try: existing = "" if os.path.exists(config_path): with open(config_path, encoding="utf-8") as fh: existing = fh.read() except OSError: return False start = existing.find("[mcp_servers.playwright.env]") if start >= 0: end = existing.find("\n[", start + 1) if end < 0: end = len(existing) existing = existing[:start] + existing[end:] lines = ["\n[mcp_servers.playwright.env]\n"] for key in sorted(env): lines.append("%s = %s\n" % (key, json.dumps(str(env[key])))) try: with open(config_path, "w", encoding="utf-8") as fh: fh.write(existing) if existing and not existing.endswith("\n"): fh.write("\n") fh.writelines(lines) except OSError: return False return True