codex-app-server/cloud/playwright_browser.py

"""Headless Chromium automation for Codex tasks, via Playwright.

One Browser object owns a list of pages for the lifetime of an
MCP process (multi-tab: new/list/switch/close tools; the
single-page tools act on the current page, and each new_page
gets a fresh Playwright context, so cookies are isolated
between pages). 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


def _allowed_hosts():
    """BROWSER_ALLOWED_HOSTS, or None when unset/empty.

    Comma list, case-insensitive. An entry with a leading dot
    (e.g. .python.org) matches the bare domain and any subdomain;
    a plain entry matches exactly. localhost/127.0.0.1 are NOT
    special-cased: they are allowed only when listed explicitly.
    Unset (or empty) means no allowlist check at all -- the old
    behaviour, where only the egress proxy whitelist applies.
    Read on every navigate so tests and env changes take effect
    without recreating the Browser."""
    raw = os.environ.get("BROWSER_ALLOWED_HOSTS")
    if raw is None:
        return None
    hosts = [entry.strip().lower() for entry in raw.split(",")]
    hosts = [entry for entry in hosts if entry]
    return hosts or None


def _check_allowed_url(url):
    """Raise BrowserError before launch when the allowlist is set
    and the URL is not allowed. file:// is always allowed; every
    other scheme except http(s) is blocked when the allowlist is
    set, because it has no host to match."""
    allowed = _allowed_hosts()
    if allowed is None:
        return
    from urllib.parse import urlparse
    parsed = urlparse(str(url))
    scheme = (parsed.scheme or "").lower()
    if scheme == "file":
        return
    if scheme not in ("http", "https"):
        raise BrowserError(
            "host allowlist blocks non-http(s) URL %r when "
            "BROWSER_ALLOWED_HOSTS is set" % (str(url),))
    host = (parsed.hostname or "").lower()
    if not host:
        raise BrowserError(
            "host allowlist blocks URL with no host %r" % (str(url),))
    for entry in allowed:
        if entry.startswith("."):
            if host == entry[1:] or host.endswith(entry):
                return
        elif host == entry:
            return
    raise BrowserError(
        "host %r is not in BROWSER_ALLOWED_HOSTS (%s); "
        "add it (or a leading-dot suffix entry) to allow this "
        "domain" % (host, ", ".join(allowed)))


def _check_navigated_url(url):
    """Post-navigation / click-target check: same allowlist as
    _check_allowed_url, but only for http(s) URLs. Anything else
    is not a page navigation (file:// is always allowed anyway,
    about:blank / data: / mailto: / javascript: never navigate
    the page to a host), so it is skipped here instead of
    raising a misleading non-http(s) error. Unset allowlist
    means no check at all, same as _check_allowed_url."""
    if _allowed_hosts() is None:
        return
    from urllib.parse import urlparse
    scheme = (urlparse(str(url)).scheme or "").lower()
    if scheme not in ("http", "https"):
        return
    _check_allowed_url(url)


class Browser:
    """Chromium pages (multi-tab). ``browser_factory`` is
    injectable: tests pass a fake returning a fake
    sync_playwright object. Single-page methods act on the
    current page; new_page/switch_page/close_page manage the
    page list, and each new_page gets a fresh Playwright
    context, so cookies are isolated between pages."""

    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._pages = []
        self._current_index = 0
        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
        self._pages = [page]
        self._current_index = 0

    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")
        # Before _ensure on purpose: a blocked host must fail
        # without launching Chromium at all.
        _check_allowed_url(url)
        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 _parse_page_index(self, index):
        """Shared switch/close index validation. Raises
        BrowserError with the page count, never IndexError."""
        if isinstance(index, bool):
            raise BrowserError(
                "page index must be an integer, got %r" % (index,))
        try:
            idx = int(index)
        except (TypeError, ValueError):
            raise BrowserError(
                "page index must be an integer, got %r" % (index,))
        if isinstance(index, float) and idx != index:
            raise BrowserError(
                "page index must be an integer, got %r" % (index,))
        if not self._pages:
            raise BrowserError(
                "no pages open: call browser_new_page or navigate "
                "first")
        if not 0 <= idx < len(self._pages):
            raise BrowserError(
                "no such page: index %d (%d page(s) open, valid "
                "0-%d)" % (idx, len(self._pages),
                            len(self._pages) - 1))
        return idx

    def _stop_screencast(self):
        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

    def _sync_current_url(self):
        try:
            self._current_url = self._page.url or ""
        except Exception:  # noqa: BLE001 - url is best-effort
            self._current_url = ""

    def new_page(self):
        """Open a new page and make it current. Every new_page
        gets a fresh Playwright context (browser.new_page()),
        so cookies and storage are isolated from other pages.
        Single-page tools then act on this page until a switch."""
        self._ensure()
        try:
            page = self._browser.new_page(
                viewport={"width": 1280, "height": 800})
        except Exception as exc:  # noqa: BLE001
            raise BrowserError("new_page failed: %s" % exc)
        self._stop_screencast()
        self._pages.append(page)
        self._current_index = len(self._pages) - 1
        self._page = page
        self._ref_paths = {}
        self._sync_current_url()
        result = self._info()
        result["ok"] = True
        result["index"] = self._current_index
        result["count"] = len(self._pages)
        return result

    def list_pages(self):
        """List all open pages without launching a browser when
        none is open. Titles/URLs are best-effort per page."""
        if not self._pages:
            return {"pages": [], "current": None, "count": 0}
        pages = []
        for i, page in enumerate(self._pages):
            try:
                url = page.url or ""
            except Exception:  # noqa: BLE001 - best-effort
                url = ""
            try:
                title = page.title() or ""
            except Exception:  # noqa: BLE001 - best-effort
                title = ""
            pages.append({"index": i, "url": url, "title": title,
                          "current": i == self._current_index})
        return {"pages": pages, "current": self._current_index,
                "count": len(pages)}

    def switch_page(self, index):
        """Make page ``index`` current. Single-page tools then
        act on it; refs from the previous page are dropped,
        because they map to that page's DOM."""
        idx = self._parse_page_index(index)
        if idx != self._current_index:
            self._stop_screencast()
            self._current_index = idx
            self._page = self._pages[idx]
            self._ref_paths = {}
            self._sync_current_url()
        result = self._info()
        result["ok"] = True
        result["index"] = self._current_index
        result["count"] = len(self._pages)
        return result

    def close_page(self, index=None):
        """Close one page. Closing the last page is an error --
        use browser_close for the whole browser. Closing the
        current page switches to the previous page (index-1, or
        0 when the first page was closed); closing a page before
        the current one keeps the same page current (its index
        shifts down by one)."""
        if not self._pages:
            raise BrowserError(
                "no pages open: call browser_new_page or navigate "
                "first")
        if len(self._pages) == 1:
            raise BrowserError(
                "cannot close the last page (index 0): use "
                "browser_close to close the whole browser")
        idx = (self._current_index if index is None
               else self._parse_page_index(index))
        page = self._pages[idx]
        try:
            page.close()
        except Exception as exc:  # noqa: BLE001
            raise BrowserError("close_page failed: %s" % exc)
        del self._pages[idx]
        if self._cdp_page is page:
            self._cdp = None
            self._cdp_page = None
        if idx < self._current_index:
            self._current_index -= 1
        elif idx == self._current_index:
            self._current_index = idx - 1 if idx > 0 else 0
        self._page = self._pages[self._current_index]
        self._ref_paths = {}
        self._sync_current_url()
        result = self._info()
        result["ok"] = True
        result["closed_index"] = idx
        result["index"] = self._current_index
        result["count"] = len(self._pages)
        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_href(self, selector):
        """Best-effort anchor href for a click target, or None.

        Reads the first matching element's href (the same element
        page.click would click). Returns the href resolved
        against the current page URL, or None when the target has
        no href, is not an anchor, or the locator API cannot
        answer (fake pages, non-anchor selectors). A None here
        does NOT mean allowed: it means the target is unknown
        before the click, so only the post-click URL check can
        catch it. Fragment-only hrefs resolve to the current
        page and are returned as None for the same reason: they
        never navigate to another host."""
        try:
            locator = self._page.locator(str(selector))
        except Exception:  # noqa: BLE001 - unknown target
            return None
        target = getattr(locator, "first", locator)
        get_attr = getattr(target, "get_attribute", None)
        if get_attr is None:
            return None
        try:
            href = get_attr("href")
        except Exception:  # noqa: BLE001 - unknown target
            return None
        if not href or not isinstance(href, str):
            return None
        href = href.strip()
        if not href or href.startswith("#"):
            return None
        from urllib.parse import urljoin
        try:
            return urljoin(self._current_url or "", href)
        except Exception:  # noqa: BLE001 - unknown target
            return None

    def _check_click_target(self, selector):
        """Pre-click allowlist gate for click/click_ref.

        When the allowlist is unset this is a no-op, so click
        behaviour is byte-identical to before. When set and the
        target href resolves to an http(s) URL, the same check
        as navigate runs BEFORE the click: a miss raises and
        the click is never executed. Unknown targets (no href,
        button/JS handler) skip this gate and are caught, if at
        all, by _check_navigated_current after the click."""
        if _allowed_hosts() is None:
            return
        target = self._click_href(selector)
        if target is not None:
            _check_navigated_url(target)

    def _check_navigated_current(self):
        """Post-click allowlist detection (not prevention).

        The click already happened, so this can only report a
        blocked landing host, never undo the navigation. It
        exists because some click targets cannot be resolved
        before the click (coordinates, JS handlers, forms): for
        those, a raised error here is still better than silently
        continuing on a blocked host, and the egress proxy 403
        is the layer that actually stops the page load when the
        allowlist equals the proxy whitelist."""
        if _allowed_hosts() is None or not self._current_url:
            return
        _check_navigated_url(self._current_url)

    def click(self, selector):
        if not selector or not str(selector).strip():
            raise BrowserError("selector is required")
        self._ensure()
        self._check_click_target(selector)
        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)
        self._check_navigated_current()
        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]
        # The ref path is a CSS path, so the same href gate as
        # click applies: an anchor ref to a blocked host is
        # refused before the click, never executed.
        self._check_click_target(path)
        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)
        self._check_navigated_current()
        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()
        # Deliberately NO pre-click gate here: coordinates do not
        # identify an element, so there is no href to check
        # before the click. Probing elementFromPoint would be
        # viewport/scroll-dependent, would miss JS-driven
        # navigations (the common click_at case: buttons, canvas,
        # SPA controls), and would run page JS on every user
        # click. What click_at does get is the post-click URL
        # check below, plus the egress proxy 403 backstop for
        # any host outside the proxy whitelist.
        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)
        self._check_navigated_current()
        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 page in list(self._pages):
            try:
                page.close()
            except Exception as exc:  # noqa: BLE001 - close is best-effort
                errors.append(str(exc))
        if self._page is not None and self._page not in self._pages:
            try:
                self._page.close()
            except Exception as exc:  # noqa: BLE001 - close is best-effort
                errors.append(str(exc))
        for closer in (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
        self._pages = []
        self._current_index = 0
        self._ref_paths = {}
        # Live-view files are per-browser, not per-shot: closing
        # the browser must not leave a stale browser-live.png /
        # browser-live.json for a watcher to mistake for a live
        # page. Failures are ignored on purpose (close is
        # best-effort; a missing file is the normal case).
        for name in ("browser-live.png", "browser-live.json"):
            try:
                os.remove(os.path.join(self.cwd, name))
            except OSError:
                pass
        return {"closed": True, "errors": errors}


# ------------------------------------------------------- codex-home glue

PLAYWRIGHT_SCRIPT_NAMES = ("playwright_browser.py", "playwright_mcp.py",
                           "playwright_remote.py")

# SHA-256 of the three MCP scripts as shipped at origin/main
# 2a569ac (PR #272), verified against the git blobs. A tenant
# script matching one of these hashes is PROVEN unmodified
# (it is exactly what a previous seeding wrote), so the safe
# upgrade below may replace it even when no manifest exists
# (legacy tenants seeded before manifests were written).
# A tenant script matching neither a manifest entry nor a hash
# here is treated as tenant-modified and is NEVER overwritten.
KNOWN_SHIPPED_SCRIPT_HASHES = {
    "playwright_browser.py":
        "0c36a286da3f8733fef4fec2dd18dec003fa68f54385fd4dc8da4b38e7571828",
    "playwright_mcp.py":
        "8455e67c2eb2ec6eb07bc9645292e2323b587e01d5a75fd8a985bbcd04a11198",
    "playwright_remote.py":
        "15a8026a02dfed6c630ea1614f75d54764a851df9d977c2380e15d0ed4f8a173",
}

_MANIFEST_NAME = "playwright_scripts_manifest.json"
_AUDIT_NAME = "playwright_scripts_audit.jsonl"


def _sha256_bytes(data):
    import hashlib
    return hashlib.sha256(data).hexdigest()


def _read_script_manifest(home):
    """name -> sha256 last written by seeding, or {}.

    The manifest is this helper's own metadata file in the
    tenant codex-home (not tenant data): it records what the
    last seeding wrote, so a later upgrade can tell an
    untouched script from a tenant-modified one. Missing or
    corrupt manifests read as empty -- the caller then falls
    back to KNOWN_SHIPPED_SCRIPT_HASHES, never to overwriting
    an unknown file."""
    try:
        with open(os.path.join(home, _MANIFEST_NAME),
                  encoding="utf-8") as fh:
            data = json.load(fh)
    except Exception:  # noqa: BLE001 - missing/corrupt = no proof
        return {}
    if not isinstance(data, dict):
        return {}
    return {k: v for k, v in data.items()
            if isinstance(k, str) and isinstance(v, str)}


def _write_script_manifest(home, hashes):
    """Best-effort manifest write; failures never block seeding."""
    try:
        with open(os.path.join(home, _MANIFEST_NAME), "w",
                  encoding="utf-8") as fh:
            json.dump({k: hashes[k] for k in sorted(hashes)},
                      fh, ensure_ascii=False, sort_keys=True)
            fh.write("\n")
    except OSError:
        pass


def _audit_script_skipped(home, name, dst_hash, src_hash):
    """Append one audit record for a skipped modified script.

    Idempotent: a record with the same script + tenant hash is
    never appended twice, so re-running an upgrade does not
    grow the log. The audit file is append-only metadata; the
    skipped script itself is left byte-identical."""
    path = os.path.join(home, _AUDIT_NAME)
    try:
        with open(path, encoding="utf-8") as fh:
            for line in fh:
                try:
                    rec = json.loads(line)
                except ValueError:
                    continue
                if (isinstance(rec, dict)
                        and rec.get("script") == name
                        and rec.get("dst_sha256") == dst_hash
                        and rec.get("action") == "skipped_modified"):
                    return path
    except OSError:
        pass
    record = {"ts": int(time.time()),
              "action": "skipped_modified",
              "script": name,
              "dst_sha256": dst_hash,
              "src_sha256": src_hash,
              "note": "tenant-modified script left untouched; "
                      "manual review required before upgrade"}
    try:
        with open(path, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(record, ensure_ascii=False) + "\n")
    except OSError:
        pass
    return path


def ensure_codex_home_playwright(codex_home_dir, command_args=None,
                                 script_dir=None):
    """Seed playwright MCP files + config into a tenant codex-home.

    SAFE UPGRADE RULE (changed from the old refresh-anything
    rule): a tenant script is auto-merged to the deployed
    version ONLY when it is proven unmodified -- its bytes
    match the seeding manifest, or match a
    KNOWN_SHIPPED_SCRIPT_HASHES entry (legacy tenants), or
    the file is missing (fresh seed) or already current. A
    script that differs AND matches neither proof is
    tenant-modified: it is NEVER overwritten, it is reported
    in ``script_skipped_modified`` (alias ``script_differs``),
    and one deduped audit record is appended to
    ``playwright_scripts_audit.jsonl`` for manual review.
    The upgrade never deletes a tenant file and never rewrites
    existing config semantics: ``[mcp_servers.playwright]``
    is appended only when absent, and the manifest/audit files
    are this helper's own additive metadata. 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__))
    manifest = _read_script_manifest(home)
    script_changed = False
    script_differs = []
    script_refreshed = []
    script_skipped = []
    audit_path = os.path.join(home, _AUDIT_NAME)
    new_hashes = {}
    for name in PLAYWRIGHT_SCRIPT_NAMES:
        src = os.path.join(here, name)
        dst = os.path.join(home, name)
        if not os.path.exists(src):
            continue
        with open(src, "rb") as fh:
            src_bytes = fh.read()
        src_hash = _sha256_bytes(src_bytes)
        if not os.path.exists(dst):
            shutil.copyfile(src, dst)
            script_changed = True
            new_hashes[name] = src_hash
            continue
        with open(dst, "rb") as fh:
            dst_bytes = fh.read()
        dst_hash = _sha256_bytes(dst_bytes)
        if dst_bytes == src_bytes:
            new_hashes[name] = src_hash
            continue
        proven_unmodified = (
            manifest.get(name) == dst_hash
            or KNOWN_SHIPPED_SCRIPT_HASHES.get(name) == dst_hash)
        if proven_unmodified:
            shutil.copyfile(src, dst)
            script_refreshed.append(name)
            script_changed = True
            new_hashes[name] = src_hash
        else:
            script_differs.append(name)
            script_skipped.append(name)
            _audit_script_skipped(home, name, dst_hash, src_hash)
            # NEVER record the tenant-modified hash as proven:
            # writing dst_hash into the manifest would make the
            # NEXT upgrade mistake this modified file for an
            # untouched one and overwrite it. Keep any older
            # proof entry (so a revert to that version can still
            # auto-merge later); otherwise drop the entry, which
            # fails closed to skip again.
            if name in manifest:
                new_hashes[name] = manifest[name]
    if new_hashes:
        _write_script_manifest(home, new_hashes)
    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,
            "script_skipped_modified": script_skipped,
            "audit_path": audit_path}


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, False without rewriting when the resulting
    file would be identical to the existing one (so a second
    task prep for an unchanged tenant leaves the file alone).
    """
    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
    original = existing
    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]))))
    # Canonical join: the old write (stripped body + lines[0]'s
    # leading blank line) grew one blank line per call, so a
    # second prep never produced identical bytes. Normalise
    # instead: body without trailing newlines, exactly one blank
    # line, then the block; an empty body keeps a single leading
    # blank line. This converges in one rewrite and is stable
    # afterwards, which the no-op check below relies on.
    block = "".join(lines).lstrip("\n")
    if existing.strip():
        new_text = existing.rstrip("\n") + "\n\n" + block
    else:
        new_text = "\n" + block
    if new_text == original:
        return False
    try:
        with open(config_path, "w", encoding="utf-8") as fh:
            fh.write(new_text)
    except OSError:
        return False
    return True