codex-app-server/cloud/playwright_browser.py

"""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 = ""

    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()
            launch = {"args": ["--no-sandbox",
                               "--disable-dev-shm-usage",
                               "--disable-gpu"]}
            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 = ""
        return {"url": self._current_url, "title": title}

    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
        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 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 = []
        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.

    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
    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 = []
    for name in ("playwright_browser.py", "playwright_mcp.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:
                script_differs.append(name)
                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}


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