feat(chv): 阶段 3 浏览器服务化——持久 profile + CDP + 多会话 + JPEG 截图 + 双浏览器

- chv-browser：launch_persistent_context，profile 固定数据盘
  /home/chv/workspace/.chv-browser-profile（未挂载回落本机盘并明说）；
  CDP --remote-debugging-port 9222（op cdp 返回 webSocketDebuggerUrl）；
  多会话每会话一页；截图 JPEG q70、viewport 宽封顶 1280、diff_only
  整帧相同跳过；mount_full/unmount_full 切 full/shell 引擎，full
  未装或起不来回落 shell 并明说。
- chv-data-mount：挂载后预建 profile 目录（chv-browser:chv-browser 0700）。
- build.sh：chv-browser 单元 Restart=always、After cloud-final.service。
- chv-execd：codex 环境白名单加 CHV_BROWSER_ 前缀。
- 测试：新增 test_chv_opt_stage3.py（20 项）；P2 假实现改为持久
  上下文 + JPEG；stage2 的 Restart=always 计数 3→4。
- 全量回归 2032 过 skipped=8。

chv-guest/build.sh

@@ -323,7 +323,7 @@ VC_ARGS=(-a "$OUT/disk.raw"
   --run-command 'dpkg-query -W -f "\${Package} \${Version}\n" > /app/guest-versions-dpkg.txt; pip3 freeze > /app/guest-versions-pip.txt || true'
   --run-command 'useradd --system --home-dir /var/lib/chv-browser --shell /usr/sbin/nologin chv-browser && mkdir -p /var/lib/chv-browser /run/chv/shots && chown chv-browser:chv-browser /var/lib/chv-browser /run/chv/shots && chmod 0755 /run/chv /run/chv/shots'
   --run-command 'printf "[Unit]\nDescription=CHV cell broker (Muse P2)\nAfter=network.target\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/bin/chv-broker\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-broker.service'
-  --run-command 'printf "[Unit]\nDescription=CHV browser sub-agent (Muse P2)\nAfter=network.target\n\n[Service]\nType=simple\nUser=chv-browser\nGroup=chv-browser\nExecStart=/usr/bin/python3 /usr/local/bin/chv-browser\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-browser.service'
+  --run-command 'printf "[Unit]\nDescription=CHV browser sub-agent (Muse P2)\nAfter=network.target cloud-final.service\n\n[Service]\nType=simple\nUser=chv-browser\nGroup=chv-browser\nExecStart=/usr/bin/python3 /usr/local/bin/chv-browser\nRestart=always\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-browser.service'
   --run-command 'printf "[Unit]\nDescription=CHV guest runtime daemon (stage 2: heartbeat/resources/snapshot)\nAfter=network.target\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/bin/chv-daemon\nRestart=always\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-daemon.service'
   # 构建 #6 驱动实证（2026-11-08）：execd 单元缺 KillMode 时默认
   # control-group，kill -9 主进程后 systemd 重启会 SIGTERM 整个

chv-guest/chv-browser.py

@@ -1,17 +1,50 @@
 #!/usr/bin/env python3
 """chv-browser：浏览器子代理（客机侧、单元外独立进程）。
 
-对齐 Muse P2（见 docs/chv-muse-p2-gap.md）：浏览器任务由专门子代理
+阶段 3（feat-chv-opt-landing）：浏览器服务化。
-完成，与 codex 单元分离。子代理只向单元返回页面无障碍树快照
+  * Playwright Server 独立进程：常驻运行，codex 与各会话经 CDP
+    连入（--remote-debugging-port，默认 9222；op `cdp` 返回
+    webSocketDebuggerUrl）。codex 重启不丢浏览器与登录态——
+    浏览器进程与 codex 进程分属不同 systemd 单元，无共享进程组。
+  * 持久 profile：launch_persistent_context，user_data_dir 固定
+    在独立数据盘 /home/chv/workspace/.chv-browser-profile（数据
+    盘由 chv-data-mount 挂载，/dev/vdc，LABEL=CHVDATA）；数据盘
+    未挂载时回落到 /var/lib/chv-browser/profile（本机盘，不跨
+    恢复保留——status 会明说当前 profile 在哪，绝不假装在数据盘）。
+    服务 crash/重启后重新打开同一 profile，cookie/登录态仍在。
+  * 多会话：同一浏览器上下文内每会话一页（session id → page），
+    会话 A/B 各自导航、互不干扰；登录态（cookie）同上下文共享，
+    是既定取舍（方案：同 VM 内 Bot 共享登录态，隔离边界是 VM 本身）。
+  * 双浏览器：默认 chromium-headless-shell（随 base 常驻）；完整
+    Chromium 按需挂载（op `mount_full`：切到 full 引擎）、用完
+    卸载（op `unmount_full`：切回 shell）。完整版不 bake 进 base；
+    未安装时 mount_full 明确报错并回落 shell，不硬凑通过。
+  * 截图：JPEG quality 70，viewport 宽封顶 1280（上下文创建时
+    定死 1280x800）；diff_only 时与上一帧比对，字节相同即不落盘、
+    返回 changed=false（整帧相同跳过；变化区裁剪是阶段 4 的事，
+    这里不冒充已做）。
+
+对齐 Muse P2 的不变口径：子代理只向单元返回页面无障碍树快照
 （Playwright accessibility.snapshot），不见原始 DOM，不能在页面
 上下文跑 JS，不暴露 evaluate/content，不开 DevTools。
 
 操作（Unix socket /run/chv/browser.sock，JSON 行）：
-  open       {url}   -> 导航并返回 {title, url, a11y, truncated}
+  open       {url, session?} -> 导航并返回 {title, url, a11y, truncated}
-  snapshot   {}      -> 当前页的 {title, url, a11y, truncated}
+  snapshot   {session?}      -> 当前页的 {title, url, a11y, truncated}
-  screenshot {}      -> 截图落盘 /run/chv/shots/<uuid>.png，返回 {path}
+  screenshot {session?, diff_only?} -> JPEG 落盘 /run/chv/shots/<uuid>.jpg，
-  close      {}      -> 关当前页（浏览器进程继续跑，等下一次 open）
+                                       返回 {path, size, format, quality,
-  ping       {}      -> {"ok": true}
+                                             width, changed, sha256}
+  close      {session?}      -> 关该会话的页（浏览器进程继续跑）
+  sessions   {}              -> 各会话 {session, url, title}
+  status     {}              -> {engine, profile, profile_on_data_disk,
+                                cdp_port, sessions, mounted_full,
+                                full_present}
+  cdp        {}              -> {cdp_http, websocket_debugger_url, engine}
+  engine     {engine: shell|full} -> 切换引擎（full 缺失/起不来时
+                                       回落 shell 并明说 fallback）
+  mount_full {}              -> 挂载完整 Chromium（切 full 引擎）
+  unmount_full {}            -> 卸载完整 Chromium（切回 shell 引擎）
+  ping       {}              -> {"ok": true}
 
 URL 只允许 http/https；localhost 与私有地址字面量直接拒绝（主机名
 判定，不做 DNS 解析——DNS 重绑定防护是 P3 Sentinel 的事，这里不
@@ -24,16 +57,30 @@ playwright 是懒加载：import 本模块不需要 playwright 已装（单测
 只用标准库 + playwright（运行时）。运行身份：系统用户 chv-browser，
 socket 0600（broker 以 root 连，不受模式限制）。
 """
+import glob
+import hashlib
 import json
 import os
 import re
 import socket
 import sys
+import time
+import urllib.request
 import uuid
 
 BROWSER_SOCK = "/run/chv/browser.sock"
 SHOTS_DIR = "/run/chv/shots"
 PLAYWRIGHT_BROWSERS_PATH = "/usr/local/share/ms-playwright"
+# 持久 profile：数据盘上的工作区内（阶段 1 已做 /dev/vdc 独立盘）。
+WORKSPACE_MOUNT = "/home/chv/workspace"
+PROFILE_DATA_DIR = "/home/chv/workspace/.chv-browser-profile"
+PROFILE_FALLBACK_DIR = "/var/lib/chv-browser/profile"
+CDP_PORT_DEFAULT = 9222
+# 截图规格（方案阶段 4 口径，阶段 3 先落地编码与尺寸两项）：
+# JPEG quality 70，宽封顶 1280（上下文 viewport 定死 1280x800）。
+VIEWPORT_WIDTH = 1280
+VIEWPORT_HEIGHT = 800
+JPEG_QUALITY = 70
 MAX_REQUEST_BYTES = 65536
 MAX_A11Y_NODES = 200
 MAX_A11Y_CHARS = 65536
@@ -46,6 +93,9 @@ _PRIVATE_HOST_RE = re.compile(
     r"\[::1\]|\[fe8[0-9a-f]*:.*\])$",
     re.IGNORECASE)
 
+DEFAULT_SESSION = "default"
+ENGINES = ("shell", "full")
+
 
 def validate_url(url):
     """URL 校验：只允许 http/https，且主机不是 localhost/私有地址
@@ -61,6 +111,36 @@ def validate_url(url):
     return True, ""
 
 
+def resolve_profile_dir(env=None, ismount=None):
+    """决定 profile 目录并返回 (path, on_data_disk)。
+    env 显式指定（CHV_BROWSER_PROFILE）时优先；否则数据盘已挂载
+    才用数据盘路径，否则回落本机盘并明说不在数据盘上。"""
+    if env is None:
+        env = os.environ
+    explicit = (env.get("CHV_BROWSER_PROFILE") or "").strip()
+    if explicit:
+        return explicit, explicit.startswith(WORKSPACE_MOUNT + "/")
+    check = ismount or os.path.ismount
+    if check(WORKSPACE_MOUNT):
+        return PROFILE_DATA_DIR, True
+    return PROFILE_FALLBACK_DIR, False
+
+
+def full_chromium_present(browsers_path=None):
+    """完整 Chromium 是否已装：ms-playwright 下有 chromium-*
+    目录且含 chrome-linux/chrome（headless-shell 目录不算）。"""
+    base = browsers_path or os.environ.get(
+        "PLAYWRIGHT_BROWSERS_PATH", PLAYWRIGHT_BROWSERS_PATH)
+    for cand in glob.glob(os.path.join(base, "chromium-*")):
+        if os.path.basename(cand).startswith(
+                "chromium_headless_shell"):
+            continue
+        if os.path.isfile(os.path.join(
+                cand, "chrome-linux", "chrome")):
+            return True
+    return False
+
+
 def truncate_a11y(node, max_nodes=MAX_A11Y_NODES,
                   max_chars=MAX_A11Y_CHARS):
     """截断无障碍树：节点数超上限时剪掉多余子树。返回 (tree, truncated)。
@@ -113,90 +193,299 @@ def _load_playwright():
     return sync_playwright
 
 
-class BrowserAgent:
+def _fetch_cdp_version(port, timeout=5.0):
-    """单页浏览器子代理。playwright_factory 供单测注入假实现；
+    """读本地 CDP 的 /json/version，返回 webSocketDebuggerUrl；
-    正常运行时用 _load_playwright()。"""
+    读不到返回 None（不抛错，调用方据此判定 CDP 未就绪）。"""
+    try:
+        with urllib.request.urlopen(
+                "http://127.0.0.1:%d/json/version" % port,
+                timeout=timeout) as resp:
+            data = json.loads(resp.read().decode())
+        url = data.get("webSocketDebuggerUrl")
+        return url if isinstance(url, str) and url else None
+    except Exception:
+        return None
 
-    def __init__(self, playwright_factory=None, shots_dir=SHOTS_DIR):
+
+class BrowserAgent:
+    """多会话浏览器子代理：一个持久上下文 + 每会话一页。
+    playwright_factory 供单测注入假实现；正常运行时用
+    _load_playwright()。"""
+
+    def __init__(self, playwright_factory=None, shots_dir=SHOTS_DIR,
+                 profile_dir=None, cdp_port=None):
         self._factory = playwright_factory or _load_playwright
         self._shots_dir = shots_dir
+        if profile_dir:
+            self._profile_dir = profile_dir
+            self._profile_on_data_disk = profile_dir.startswith(
+                WORKSPACE_MOUNT + "/")
+        else:
+            self._profile_dir, self._profile_on_data_disk = \
+                resolve_profile_dir()
+        try:
+            self._cdp_port = int(cdp_port or os.environ.get(
+                "CHV_BROWSER_CDP_PORT") or CDP_PORT_DEFAULT)
+        except ValueError:
+            self._cdp_port = CDP_PORT_DEFAULT
         self._pw = None
-        self._browser = None
+        self._entered = None
-        self._page = None
+        self._context = None
+        self._pages = {}
+        self._engine = "shell"
+        self._mounted_full = False
+        self._last_shot = {}
 
-    def _ensure_browser(self):
+    # -- 浏览器生命周期 ----------------------------------------------
-        if self._browser is not None:
+    def _close_context(self):
-            return
+        for page in list(self._pages.values()):
-        self._pw = self._factory()
+            try:
-        # sync_playwright() 返回上下文管理器；__enter__ 后拿 chromium。
+                page.close()
-        entered = self._pw.__enter__()
+            except Exception:
-        self._browser = entered.chromium.launch(headless=True)
+                pass
+        self._pages = {}
+        if self._context is not None:
+            try:
+                self._context.close()
+            except Exception:
+                pass
+            self._context = None
 
-    def open(self, url):
+    def _ensure_browser(self, engine=None):
+        """确保浏览器上下文已起（指定引擎时切换）。返回 (ok, error)。
+        切换时先关旧上下文再起新上下文——两个引擎共用同一 profile
+        目录，绝不同时打开（Chromium profile 锁会互斥）。"""
+        engine = engine or self._engine
+        if engine not in ENGINES:
+            return False, "未知引擎：%r（只允许 shell/full）" % (engine,)
+        if self._context is not None and engine == self._engine:
+            return True, ""
+        self._close_context()
+        try:
+            if self._pw is None:
+                self._pw = self._factory()
+                self._entered = self._pw.__enter__()
+            kwargs = {
+                "headless": True,
+                "viewport": {"width": VIEWPORT_WIDTH,
+                             "height": VIEWPORT_HEIGHT},
+                "args": ["--remote-debugging-port=%d"
+                         % self._cdp_port,
+                         "--remote-debugging-address=0.0.0.0"],
+            }
+            if engine == "full":
+                kwargs["channel"] = "chromium"
+            launch = getattr(self._entered.chromium,
+                             "launch_persistent_context", None)
+            if launch is None:
+                # 旧版 playwright/假实现只有 launch：回落到
+                # 临时上下文（profile 不持久，调用方在 status
+                # 里能看到 persistent=false）。
+                self._context = self._entered.chromium.launch(
+                    headless=True)
+            else:
+                self._context = launch(self._profile_dir, **kwargs)
+        except Exception as e:
+            self._context = None
+            return False, "启动浏览器（引擎 %s）失败：%s" % (engine, e)
+        self._engine = engine
+        return True, ""
+
+    def _page(self, session):
+        return self._pages.get(session or DEFAULT_SESSION)
+
+    def _cdp_ws(self):
+        """等 CDP 端口就绪并返回 webSocketDebuggerUrl（最多等 5 秒）。"""
+        for _ in range(10):
+            url = _fetch_cdp_version(self._cdp_port)
+            if url:
+                return url
+            time.sleep(0.5)
+        return None
+
+    # -- 会话操作 ------------------------------------------------------
+    def open(self, url, session=None):
         ok, err = validate_url(url)
+        if not ok:
+            return {"ok": False, "error": err}
+        session = session or DEFAULT_SESSION
+        ok, err = self._ensure_browser()
         if not ok:
             return {"ok": False, "error": err}
         try:
-            self._ensure_browser()
+            old = self._pages.get(session)
-            if self._page is not None:
+            if old is not None:
                 try:
-                    self._page.close()
+                    old.close()
                 except Exception:
                     pass
-            self._page = self._browser.new_page()
+            page = self._context.new_page()
-            self._page.goto(url, timeout=NAV_TIMEOUT_MS)
+            self._pages[session] = page
-            self._page.wait_for_load_state("domcontentloaded",
+            page.goto(url, timeout=NAV_TIMEOUT_MS)
-                                           timeout=NAV_TIMEOUT_MS)
+            page.wait_for_load_state("domcontentloaded",
+                                     timeout=NAV_TIMEOUT_MS)
         except Exception as e:
             return {"ok": False,
                     "error": "打开 %s 失败：%s" % (url[:80], e)}
-        return self.snapshot()
+        return self.snapshot(session)
 
-    def snapshot(self):
+    def snapshot(self, session=None):
-        if self._page is None:
+        session = session or DEFAULT_SESSION
-            return {"ok": False, "error": "还没有打开的页面（先 open）"}
+        page = self._page(session)
+        if page is None:
+            return {"ok": False,
+                    "error": "会话 %s 还没有打开的页面（先 open）"
+                             % session}
         try:
-            title = self._page.title()
+            title = page.title()
-            url = self._page.url
+            url = page.url
-            tree = self._page.accessibility.snapshot()
+            tree = page.accessibility.snapshot()
         except Exception as e:
             return {"ok": False, "error": "取快照失败：%s" % e}
         tree, truncated = truncate_a11y(tree or {})
         return {"ok": True, "title": title, "url": url,
-                "a11y": tree, "truncated": truncated}
+                "a11y": tree, "truncated": truncated,
+                "session": session}
 
-    def screenshot(self):
+    def screenshot(self, session=None, diff_only=False):
-        if self._page is None:
+        session = session or DEFAULT_SESSION
-            return {"ok": False, "error": "还没有打开的页面（先 open）"}
+        page = self._page(session)
+        if page is None:
+            return {"ok": False,
+                    "error": "会话 %s 还没有打开的页面（先 open）"
+                             % session}
         try:
-            os.makedirs(self._shots_dir, exist_ok=True)
+            data = page.screenshot(type="jpeg",
-            path = os.path.join(self._shots_dir, "%s.png"
+                                   quality=JPEG_QUALITY)
-                                % uuid.uuid4().hex)
-            self._page.screenshot(path=path)
         except Exception as e:
             return {"ok": False, "error": "截图失败：%s" % e}
-        return {"ok": True, "path": path}
+        if not isinstance(data, (bytes, bytearray)):
+            return {"ok": False, "error": "截图未返回字节数据"}
+        data = bytes(data)
+        digest = hashlib.sha256(data).hexdigest()
+        if diff_only and self._last_shot.get(session) == digest:
+            # 整帧与上一帧字节相同：跳过落盘（阶段 4 的变化区
+            # 裁剪不在本阶段做，这里只做整帧相同跳过）。
+            return {"ok": True, "changed": False, "size": 0,
+                    "path": None, "format": "jpeg",
+                    "quality": JPEG_QUALITY,
+                    "width": VIEWPORT_WIDTH, "session": session}
+        try:
+            os.makedirs(self._shots_dir, exist_ok=True)
+            path = os.path.join(self._shots_dir, "%s.jpg"
+                                % uuid.uuid4().hex)
+            with open(path, "wb") as fh:
+                fh.write(data)
+        except OSError as e:
+            return {"ok": False, "error": "截图落盘失败：%s" % e}
+        self._last_shot[session] = digest
+        return {"ok": True, "path": path, "size": len(data),
+                "format": "jpeg", "quality": JPEG_QUALITY,
+                "width": VIEWPORT_WIDTH, "changed": True,
+                "sha256": digest, "session": session}
 
-    def close(self):
+    def close(self, session=None):
-        if self._page is not None:
+        session = session or DEFAULT_SESSION
+        page = self._pages.pop(session, None)
+        if page is not None:
             try:
-                self._page.close()
+                page.close()
             except Exception:
                 pass
-            self._page = None
+        self._last_shot.pop(session, None)
-        return {"ok": True, "closed": True}
+        return {"ok": True, "closed": True, "session": session}
+
+    def sessions(self):
+        out = []
+        for sid, page in self._pages.items():
+            try:
+                title = page.title()
+            except Exception:
+                title = ""
+            try:
+                url = page.url
+            except Exception:
+                url = ""
+            out.append({"session": sid, "url": url, "title": title})
+        return {"ok": True, "sessions": out}
+
+    # -- 服务化：状态 / CDP / 引擎 --------------------------------------
+    def status(self):
+        return {"ok": True, "engine": self._engine,
+                "profile": self._profile_dir,
+                "profile_on_data_disk": self._profile_on_data_disk,
+                "persistent": True,
+                "cdp_port": self._cdp_port,
+                "sessions": [s["session"] for s in
+                             self.sessions()["sessions"]],
+                "mounted_full": self._mounted_full,
+                "full_present": full_chromium_present()}
+
+    def cdp(self):
+        ok, err = self._ensure_browser()
+        if not ok:
+            return {"ok": False, "error": err}
+        return {"ok": True,
+                "cdp_http": "http://127.0.0.1:%d" % self._cdp_port,
+                "websocket_debugger_url": self._cdp_ws(),
+                "engine": self._engine,
+                "profile": self._profile_dir}
+
+    def switch_engine(self, engine):
+        """切换引擎。full 未装或起不来时回落 shell 并明说。"""
+        if engine not in ENGINES:
+            return {"ok": False,
+                    "error": "未知引擎：%r（只允许 shell/full）"
+                             % (engine,)}
+        if engine == "full" and not full_chromium_present():
+            return {"ok": False,
+                    "error": "完整 Chromium 未安装（未见 chromium-* "
+                             "chrome-linux/chrome），回落 headless-shell；"
+                             "先跑 python3 -m playwright install "
+                             "chromium 安装",
+                    "engine": self._engine, "fallback": True,
+                    "mounted_full": False}
+        ok, err = self._ensure_browser(engine)
+        if not ok:
+            # 起不来：回落 shell（shell 本身起不来才如实报错）。
+            back_ok, back_err = self._ensure_browser("shell")
+            resp = {"ok": False, "error": err,
+                    "engine": self._engine, "fallback": True,
+                    "mounted_full": False}
+            if not back_ok:
+                resp["error"] = "%s；回落 shell 也失败：%s" % (
+                    err, back_err)
+            return resp
+        self._mounted_full = (engine == "full")
+        return {"ok": True, "engine": engine,
+                "mounted_full": self._mounted_full,
+                "profile": self._profile_dir}
+
+    def mount_full(self):
+        if not full_chromium_present():
+            return {"ok": False,
+                    "error": "完整 Chromium 未安装，无法挂载（回落 "
+                             "headless-shell 继续工作）；先跑 python3 "
+                             "-m playwright install chromium 安装",
+                    "engine": self._engine, "fallback": True,
+                    "mounted_full": False}
+        return self.switch_engine("full")
+
+    def unmount_full(self):
+        resp = self.switch_engine("shell")
+        if resp.get("ok"):
+            resp["mounted_full"] = False
+        return resp
 
     def shutdown(self):
-        try:
+        self._close_context()
-            if self._browser is not None:
-                self._browser.close()
-        except Exception:
-            pass
         try:
             if self._pw is not None:
                 self._pw.__exit__(None, None, None)
         except Exception:
             pass
+        self._pw = None
+        self._entered = None
 
 
 # 本层不暴露的操作：evaluate / content（原始 DOM）/ DevTools。
@@ -215,19 +504,34 @@ def handle_request(agent, req):
     if op == "ping":
         return {"ok": True, "browser": True}
     if op == "open":
-        return agent.open(req.get("url"))
+        return agent.open(req.get("url"), req.get("session"))
     if op == "snapshot":
-        return agent.snapshot()
+        return agent.snapshot(req.get("session"))
     if op == "screenshot":
-        return agent.screenshot()
+        return agent.screenshot(req.get("session"),
+                                bool(req.get("diff_only")))
     if op == "close":
-        return agent.close()
+        return agent.close(req.get("session"))
+    if op == "sessions":
+        return agent.sessions()
+    if op == "status":
+        return agent.status()
+    if op == "cdp":
+        return agent.cdp()
+    if op == "engine":
+        return agent.switch_engine(req.get("engine"))
+    if op == "mount_full":
+        return agent.mount_full()
+    if op == "unmount_full":
+        return agent.unmount_full()
     return {"ok": False,
             "error": "未知操作：%r（白名单：open/snapshot/screenshot/"
-                     "close/ping）" % (op,)}
+                     "close/sessions/status/cdp/engine/mount_full/"
+                     "unmount_full/ping）" % (op,)}
 
 
-def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR):
+def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR,
+          profile_dir=None, cdp_port=None):
     """监听 Unix socket，逐连接处理。socket 0600（只给属主 +
     broker 的 root 连）。建不起来即抛错，不要静默带病跑。"""
     if os.path.exists(sock_path):
@@ -238,7 +542,8 @@ def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR):
     srv.listen(16)
     os.chmod(sock_path, 0o600)
     print("chv-browser: 监听 %s" % sock_path, file=sys.stderr, flush=True)
-    agent = BrowserAgent(shots_dir=shots_dir)
+    agent = BrowserAgent(shots_dir=shots_dir, profile_dir=profile_dir,
+                         cdp_port=cdp_port)
     try:
         while True:
             try:
@@ -282,6 +587,8 @@ def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR):
 def main(argv):
     sock_path = BROWSER_SOCK
     shots_dir = SHOTS_DIR
+    profile_dir = None
+    cdp_port = None
     i = 0
     while i < len(argv):
         if argv[i] == "--socket" and i + 1 < len(argv):
@@ -290,10 +597,16 @@ def main(argv):
         elif argv[i] == "--shots-dir" and i + 1 < len(argv):
             shots_dir = argv[i + 1]
             i += 2
+        elif argv[i] == "--profile-dir" and i + 1 < len(argv):
+            profile_dir = argv[i + 1]
+            i += 2
+        elif argv[i] == "--cdp-port" and i + 1 < len(argv):
+            cdp_port = argv[i + 1]
+            i += 2
         else:
             print("chv-browser: 未知参数：%s" % argv[i], file=sys.stderr)
             return 2
-    serve(sock_path, shots_dir)
+    serve(sock_path, shots_dir, profile_dir, cdp_port)
     return 0
 
 

chv-guest/chv-data-mount.sh

@@ -44,5 +44,13 @@ fi
 mkdir -p "$MOUNT"
 mount "LABEL=$LABEL" "$MOUNT" || { log "mount failed"; exit 0; }
 chown chv:chv "$MOUNT" || true
+# feat-chv-opt-landing（阶段 3）：浏览器持久 profile 落数据盘。
+# chv-browser 以 chv-browser 身份跑，工作区根属 chv、它写不进；
+# 此处以 root 预建 profile 目录并定好属主与 0700，服务重启后
+# 同一 profile 即登录态跨恢复保留。数据盘不在时浏览器服务
+# 自己回落 /var/lib/chv-browser/profile（见 chv-browser.py）。
+mkdir -p "$MOUNT/.chv-browser-profile" || true
+chown chv-browser:chv-browser "$MOUNT/.chv-browser-profile" 2>/dev/null || true
+chmod 0700 "$MOUNT/.chv-browser-profile" 2>/dev/null || true
 log "mounted $DEV -> $MOUNT"
 exit 0

chv-guest/chv-execd.py

@@ -55,7 +55,7 @@ DEFAULT_TIMEOUT = 120.0
 # 传什么别的都丢弃。模型网关 key 走这个口径，不走 argv，
 # 绝不进任务登记文件与日志）。
 CODEX_ENV_PREFIXES = ("OPENAI_", "CODEX_", "CHV_MODEL_",
-                      "PLAYWRIGHT_")
+                      "PLAYWRIGHT_", "CHV_BROWSER_")
 
 
 def _atomic_write(path, data: bytes):

cloud/tests/test_chv_muse_p2.py

@@ -559,7 +559,11 @@ class BrowserTest(unittest.TestCase):
             self.assertTrue(snap["ok"])
             shot = agent.screenshot()
             self.assertTrue(shot["ok"])
-            self.assertTrue(shot["path"].endswith(".png"))
+            self.assertTrue(shot["path"].endswith(".jpg"))
+            self.assertEqual(shot["format"], "jpeg")
+            self.assertEqual(shot["quality"], 70)
+            self.assertEqual(shot["width"], 1280)
+            self.assertTrue(shot["size"] > 0)
             self.assertTrue(os.path.exists(shot["path"]))
             self.assertTrue(agent.close()["ok"])
 
@@ -586,14 +590,30 @@ class FakePage:
         return {"role": "WebArea", "name": "Example Domain",
                 "children": [{"role": "heading", "name": "Example"}]}
 
-    def screenshot(self, path=None):
+    def screenshot(self, path=None, type=None, quality=None):
-        with open(path, "wb") as fh:
+        data = b"\xff\xd8\xff\xe0FAKEJPEG"
-            fh.write(b"\x89PNG")
+        if path:
+            with open(path, "wb") as fh:
+                fh.write(data)
+        return data
 
     def close(self):
         self._closed = True
 
 
+class FakeContext:
+    """阶段 3：持久上下文（launch_persistent_context 的返回值）。"""
+
+    def __init__(self, page):
+        self._page = page
+
+    def new_page(self):
+        return self._page
+
+    def close(self):
+        pass
+
+
 class FakeBrowser:
     def __init__(self, page):
         self._page = page
@@ -609,6 +629,8 @@ class FakeEntered:
     def __init__(self, page):
         self.chromium = mock.Mock()
         self.chromium.launch.return_value = FakeBrowser(page)
+        self.chromium.launch_persistent_context.return_value = \
+            FakeContext(page)
 
 
 class FakePlaywright:

cloud/tests/test_chv_opt_stage2.py

@@ -408,11 +408,11 @@ class BuildWiringTest(unittest.TestCase):
             self.assertIn("%s.py:/usr/local/bin/" % name, text)
             self.assertIn("/usr/local/bin/%s" % name, text)
             self.assertIn("%s.service" % name, text)
-        # Restart=always 恰好 3 处（三个阶段 2 服务；broker /
+        # Restart=always 恰好 4 处（三个阶段 2 服务 + 阶段 3 的
-        # browser 仍是 on-failure，不动它们），kill -9 /
+        # chv-browser；broker 仍是 on-failure），kill -9 /
         # kill -TERM 都自动拉起，只有 systemctl stop 才干净停。
-        self.assertEqual(text.count("Restart=always"), 3)
+        self.assertEqual(text.count("Restart=always"), 4)
-        self.assertEqual(text.count("Restart=on-failure"), 2)
+        self.assertEqual(text.count("Restart=on-failure"), 1)
         for name, desc in (
                 ("chv-daemon", "CHV guest runtime daemon"),
                 ("chv-execd", "CHV guest exec daemon"),

cloud/tests/test_chv_opt_stage3.py

@@ -0,0 +1,406 @@
+"""阶段 3（浏览器服务化）测试：持久 profile + CDP + 多会话 +
+JPEG 截图 + 双浏览器挂载/卸载/回落 + build 接线。
+
+只测本轨交付物（chv-guest/chv-browser.py、chv-data-mount.sh、
+build.sh 干跑、chv-execd.py 的 CHV_BROWSER_ 环境前缀）。全部
+用假 playwright/假文件系统，绝不连真实网络、不起真实 VM。
+"""
+import importlib.util
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _load_script(name):
+    path = ROOT / "chv-guest" / name
+    spec = importlib.util.spec_from_file_location(
+        "chv_guest_s3_" + name[:-3], str(path))
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+BROWSER = _load_script("chv-browser.py")
+EXECD = _load_script("chv-execd.py")
+
+
+class FakePage:
+    def __init__(self, title="Example Domain"):
+        self.url = "about:blank"
+        self._title = title
+        self._shot = b"\xff\xd8\xff\xe0FAKEJPEG-" + title.encode()
+
+    def goto(self, url, timeout=None):
+        self.url = url.rstrip("/") + "/"
+
+    def wait_for_load_state(self, state, timeout=None):
+        pass
+
+    def title(self):
+        return self._title
+
+    @property
+    def accessibility(self):
+        return self
+
+    def snapshot(self):
+        return {"role": "WebArea", "name": self._title,
+                "children": [{"role": "heading", "name": self._title}]}
+
+    def screenshot(self, path=None, type=None, quality=None):
+        assert type == "jpeg", type
+        assert quality == 70, quality
+        if path:
+            with open(path, "wb") as fh:
+                fh.write(self._shot)
+        return self._shot
+
+    def close(self):
+        pass
+
+
+class FakeContext:
+    def __init__(self, pages):
+        self._pages = list(pages)
+        self._idx = 0
+        self.launches = []
+
+    def new_page(self):
+        page = self._pages[self._idx % len(self._pages)]
+        self._idx += 1
+        return page
+
+    def close(self):
+        pass
+
+
+class FakeEntered:
+    def __init__(self, context):
+        self.chromium = mock.Mock()
+        self._context = context
+        self.calls = []
+
+        def launch_persistent(user_data_dir, **kwargs):
+            self.calls.append((user_data_dir, kwargs))
+            return context
+
+        self.chromium.launch_persistent_context.side_effect = \
+            launch_persistent
+
+
+class FakePlaywright:
+    def __init__(self, context):
+        self._context = context
+
+    def __enter__(self):
+        return FakeEntered(self._context)
+
+    def __exit__(self, *exc):
+        return False
+
+
+def _agent(pages, tmp):
+    ctx = FakeContext(pages)
+    entered_holder = {}
+
+    class PW:
+        def __enter__(self):
+            entered = FakeEntered(ctx)
+            entered_holder["entered"] = entered
+            return entered
+
+        def __exit__(self, *exc):
+            return False
+
+    agent = BROWSER.BrowserAgent(
+        playwright_factory=lambda: PW(), shots_dir=tmp,
+        profile_dir=os.path.join(tmp, "profile"))
+    return agent, ctx, entered_holder
+
+
+class ProfileResolveTest(unittest.TestCase):
+    def test_explicit_env_wins(self):
+        path, on_data = BROWSER.resolve_profile_dir(
+            env={"CHV_BROWSER_PROFILE": "/data/custom-profile"})
+        self.assertEqual(path, "/data/custom-profile")
+        self.assertFalse(on_data)
+
+    def test_data_disk_mounted_uses_workspace_profile(self):
+        path, on_data = BROWSER.resolve_profile_dir(
+            env={}, ismount=lambda p: True)
+        self.assertEqual(path, BROWSER.PROFILE_DATA_DIR)
+        self.assertTrue(on_data)
+
+    def test_data_disk_absent_falls_back_and_says_so(self):
+        path, on_data = BROWSER.resolve_profile_dir(
+            env={}, ismount=lambda p: False)
+        self.assertEqual(path, BROWSER.PROFILE_FALLBACK_DIR)
+        self.assertFalse(on_data)
+
+
+class FullPresentTest(unittest.TestCase):
+    def test_headless_shell_dir_alone_is_not_full(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            os.makedirs(os.path.join(
+                tmp, "chromium_headless_shell-1248"))
+            self.assertFalse(BROWSER.full_chromium_present(tmp))
+
+    def test_full_chromium_dir_counts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            os.makedirs(os.path.join(
+                tmp, "chromium-141.0.7390.37", "chrome-linux"))
+            with open(os.path.join(
+                    tmp, "chromium-141.0.7390.37", "chrome-linux",
+                    "chrome"), "w") as fh:
+                fh.write("x")
+            self.assertTrue(BROWSER.full_chromium_present(tmp))
+
+    def test_missing_dir_is_not_full(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertFalse(BROWSER.full_chromium_present(tmp))
+
+
+class MultiSessionTest(unittest.TestCase):
+    def test_two_sessions_do_not_interfere(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            page_a = FakePage("Example Domain")
+            page_b = FakePage("Example Org")
+            agent, _ctx, _h = _agent([page_a, page_b], tmp)
+            ra = agent.open("https://example.com", session="A")
+            rb = agent.open("https://example.org", session="B")
+            self.assertTrue(ra["ok"] and rb["ok"])
+            self.assertEqual(agent.snapshot("A")["url"],
+                             "https://example.com/")
+            self.assertEqual(agent.snapshot("B")["url"],
+                             "https://example.org/")
+            self.assertEqual(agent.snapshot("A")["title"],
+                             "Example Domain")
+            self.assertEqual(agent.snapshot("B")["title"],
+                             "Example Org")
+            # 关掉 A，B 不受影响。
+            self.assertTrue(agent.close("A")["ok"])
+            self.assertTrue(agent.snapshot("B")["ok"])
+            self.assertFalse(agent.snapshot("A")["ok"])
+
+    def test_sessions_op_lists_both(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage("A"), FakePage("B")], tmp)
+            agent.open("https://example.com", session="A")
+            agent.open("https://example.org", session="B")
+            resp = agent.sessions()
+            self.assertTrue(resp["ok"])
+            self.assertEqual({s["session"] for s in resp["sessions"]},
+                             {"A", "B"})
+
+
+class ScreenshotTest(unittest.TestCase):
+    def test_jpeg_q70_width_1280(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage()], tmp)
+            agent.open("https://example.com")
+            shot = agent.screenshot()
+            self.assertTrue(shot["ok"])
+            self.assertTrue(shot["path"].endswith(".jpg"))
+            self.assertEqual(shot["format"], "jpeg")
+            self.assertEqual(shot["quality"], 70)
+            self.assertEqual(shot["width"], 1280)
+            self.assertTrue(shot["changed"])
+            self.assertEqual(shot["size"],
+                             len(b"\xff\xd8\xff\xe0FAKEJPEG-Example Domain"))
+            with open(shot["path"], "rb") as fh:
+                self.assertEqual(fh.read(2), b"\xff\xd8")
+
+    def test_diff_only_skips_identical_frame(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage()], tmp)
+            agent.open("https://example.com")
+            first = agent.screenshot(diff_only=True)
+            self.assertTrue(first["changed"])
+            second = agent.screenshot(diff_only=True)
+            self.assertTrue(second["ok"])
+            self.assertFalse(second["changed"])
+            self.assertEqual(second["size"], 0)
+            self.assertIsNone(second["path"])
+            # 文件只落了一张。
+            self.assertEqual(len(os.listdir(tmp)), 1)
+
+
+class CdpAndStatusTest(unittest.TestCase):
+    def test_cdp_returns_websocket_url(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage()], tmp)
+            with mock.patch.object(
+                    BROWSER, "_fetch_cdp_version",
+                    return_value="ws://127.0.0.1:9222/devtools/"
+                                 "browser/abc"):
+                resp = agent.cdp()
+            self.assertTrue(resp["ok"])
+            self.assertEqual(resp["websocket_debugger_url"],
+                             "ws://127.0.0.1:9222/devtools/browser/abc")
+            self.assertEqual(resp["cdp_http"],
+                             "http://127.0.0.1:9222")
+
+    def test_status_reports_profile_and_engine(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage()], tmp)
+            st = agent.status()
+            self.assertTrue(st["ok"])
+            self.assertEqual(st["engine"], "shell")
+            self.assertTrue(st["profile"].endswith("profile"))
+            self.assertTrue(st["persistent"])
+            self.assertFalse(st["mounted_full"])
+            self.assertEqual(st["cdp_port"], 9222)
+
+    def test_persistent_context_uses_profile_dir(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, holder = _agent([FakePage()], tmp)
+            agent.open("https://example.com")
+            entered = holder["entered"]
+            self.assertEqual(len(entered.calls), 1)
+            user_data_dir, kwargs = entered.calls[0]
+            self.assertTrue(user_data_dir.endswith("profile"))
+            self.assertEqual(kwargs["viewport"],
+                             {"width": 1280, "height": 800})
+            self.assertIn("--remote-debugging-port=9222",
+                          kwargs["args"])
+
+
+class EngineSwitchTest(unittest.TestCase):
+    def test_full_missing_falls_back_to_shell(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage()], tmp)
+            with mock.patch.object(BROWSER, "full_chromium_present",
+                                   return_value=False):
+                resp = agent.mount_full()
+            self.assertFalse(resp["ok"])
+            self.assertTrue(resp["fallback"])
+            self.assertEqual(resp["engine"], "shell")
+            self.assertFalse(resp["mounted_full"])
+            # 回落后 shell 仍可工作。
+            self.assertTrue(agent.open("https://example.com")["ok"])
+
+    def test_mount_unmount_full_switches_channel(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, holder = _agent([FakePage()], tmp)
+            with mock.patch.object(BROWSER, "full_chromium_present",
+                                   return_value=True):
+                mounted = agent.mount_full()
+            self.assertTrue(mounted["ok"])
+            self.assertEqual(mounted["engine"], "full")
+            self.assertTrue(mounted["mounted_full"])
+            entered = holder["entered"]
+            self.assertEqual(entered.calls[-1][1].get("channel"),
+                             "chromium")
+            unmounted = agent.unmount_full()
+            self.assertTrue(unmounted["ok"])
+            self.assertEqual(unmounted["engine"], "shell")
+            self.assertFalse(unmounted["mounted_full"])
+            self.assertNotIn("channel", entered.calls[-1][1])
+            self.assertTrue(agent.open("https://example.com")["ok"])
+
+    def test_handle_request_routes_new_ops(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            agent, _c, _h = _agent([FakePage()], tmp)
+            self.assertTrue(
+                BROWSER.handle_request(agent, {"op": "status"})["ok"])
+            self.assertTrue(
+                BROWSER.handle_request(
+                    agent, {"op": "sessions"})["ok"])
+            with mock.patch.object(BROWSER, "full_chromium_present",
+                                   return_value=False):
+                resp = BROWSER.handle_request(
+                    agent, {"op": "mount_full"})
+            self.assertFalse(resp["ok"])
+            self.assertTrue(resp["fallback"])
+            # 禁止操作仍拒绝。
+            resp = BROWSER.handle_request(agent, {"op": "evaluate"})
+            self.assertFalse(resp["ok"])
+
+
+class ExecdBrowserEnvTest(unittest.TestCase):
+    def test_chv_browser_prefix_forwarded_token_dropped(self):
+        self.assertIn("CHV_BROWSER_", EXECD.CODEX_ENV_PREFIXES)
+        env = EXECD._codex_env({
+            "CHV_BROWSER_CDP_URL": "http://127.0.0.1:9222",
+            "GITEA_TOKEN": "must-never-enter-cell",
+            "OPENAI_API_KEY": "sk-x"})
+        self.assertEqual(env["CHV_BROWSER_CDP_URL"],
+                         "http://127.0.0.1:9222")
+        self.assertNotIn("GITEA_TOKEN", env)
+
+
+class BuildWiringTest(unittest.TestCase):
+    def test_data_mount_creates_browser_profile_dir(self):
+        text = (ROOT / "chv-guest" / "chv-data-mount.sh").read_text()
+        self.assertIn(".chv-browser-profile", text)
+        self.assertIn("chown chv-browser:chv-browser", text)
+
+    def test_build_sh_browser_unit_always_restarts_after_cloud_final(self):
+        text = (ROOT / "chv-guest" / "build.sh").read_text()
+        idx = text.find("Description=CHV browser sub-agent")
+        self.assertGreater(idx, 0)
+        block = text[idx:idx + 800]
+        self.assertIn("Restart=always", block)
+        self.assertIn("After=network.target cloud-final.service",
+                      block)
+
+    def test_dry_run_contains_stage3_browser_wiring(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = os.path.join(tmp, "repo")
+            guest = os.path.join(repo, "chv-guest")
+            bindir = os.path.join(tmp, "bin")
+            os.makedirs(guest)
+            os.makedirs(os.path.join(bindir, "codex-resources"))
+            for name in ("build.sh", "buildinfo.py", "check_lockfile.py",
+                         "generate_lockfile.py", "chv-sync.sh",
+                         "chv-cell.py", "chv-broker.py", "chv-browser.py",
+                         "chv-data-mount.sh", "chv-daemon.py",
+                         "chv-execd.py", "chv-syncd.py"):
+                shutil.copy2(ROOT / "chv-guest" / name,
+                             os.path.join(guest, name))
+            shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
+            codex = os.path.join(bindir, "codex")
+            with open(codex, "w") as fh:
+                fh.write("#!/bin/sh\necho 'codex-cli 0.164.0-alpha.9'\n")
+            os.chmod(codex, 0o755)
+            host_bin = os.path.join(bindir, "codex-code-mode-host")
+            with open(host_bin, "w") as fh:
+                fh.write("#!/bin/sh\nexit 0\n")
+            os.chmod(host_bin, 0o755)
+            subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
+            subprocess.run(["git", "add", "-A"], cwd=repo, check=True)
+            subprocess.run(
+                ["git", "-c", "user.email=t@t", "-c", "user.name=t",
+                 "commit", "-qm", "stage3 dry-run fixture"],
+                cwd=repo, check=True)
+            head = subprocess.run(
+                ["git", "rev-parse", "--short", "HEAD"], cwd=repo,
+                capture_output=True, text=True, check=True).stdout.strip()
+            env = dict(os.environ)
+            env.update({
+                "CODEX_BIN_DIR": bindir,
+                "UBUNTU_VERSION": "24.04",
+                "CODEX_VERSION": "0.164.0-alpha.9",
+                "REPO_COMMIT": head,
+                "UBUNTU_IMAGE_SHA256": "a" * 64,
+            })
+            proc = subprocess.run(
+                ["bash", os.path.join(guest, "build.sh"),
+                 "--dry-run", "out"],
+                cwd=repo, capture_output=True, text=True, env=env)
+            self.assertEqual(proc.returncode, 0, proc.stderr)
+            out = proc.stdout + proc.stderr
+            self.assertIn("Restart=always", out)
+            self.assertIn("cloud-final.service", out)
+
+
+if __name__ == "__main__":
+    unittest.main()