merge(chv-muse-round12): 合入 round-11 修复轮 HEAD 50cae2a (RETEST6 全绿)

按 hunk 合入，不整文件覆盖。冲突 3 处：
- chv-guest/build.sh：取 P4 的 copy-in（cdp-broker/guard/blocklist/
  chv-memory）+ cdp-broker service/enable 三件；取 R11 的 git
  config（去掉 credential.helper store，层 5）；tmpfiles 两边
  并存（/run/chv/ask-queue + /run/chv-browser）。
- chv-guest/chv-browser.py：P4 架构保留（CDP broker 客户端 +
  guard 模块）；取 R11 两处仍适用的运行期修复：BROWSER_SOCK
  移进 /run/chv-browser（缺陷 4，合入后的 broker 已指该路径）
  与 handle_request 认 op=browser/action 透传；R11 的直连
  Playwright 代码（validate_url/truncate_aria/_load_playwright/
  _page snapshot）由 P4 取代，不带回。
- cloud/tests/test_chv_muse_p2.py：truncate 三测取 P4 的
  GUARD.truncate_a11y（dict 树）；FakeCdpBroker 取 P4 版，删
  R11 的 FakePage 系列；干跑夹具取 R11 的 _fixture 调用，并
  给 _fixture 补 P4 四件（否则合入后 build.sh fail-fast）。

合入内移植（R11 修复在 P4 里移了位置，不移植即静默丢失）：
- chv-guest/chv-cdp-broker.py：_load_playwright 返回
  sync_playwright()（原 R11 在 chv-browser.py 修过的同形
  bug）；回归测试由 test_p2 移植到 test_chv_muse_p4.py。
- test_chv_muse_p2.py：HOST 由模块级 eager 改懒加载（_host()）：
  host.py 顶层 import queue 会把 stdlib queue 缓进 sys.modules，
  使同进程后加载的 test_queue_sandbox 取不到 cloud/queue.py
  遮蔽而 ImportError（全量唯一失败）。

未移植（见报告）：R11 的 Page.aria_snapshot() 纠正未进
chv-cdp-broker.py（仍调 accessibility.snapshot()，且 guard
的 dict 树 API 须配套改，属 P4 返工，不在本轮合入做）。

全量 cloud 套件：2522 tests OK (skipped=10)，基线 a6f9597
为 2474 OK (skipped=12)；+48 来自 R11 新增测试（p2 +36、
p3 +7、closeup +3、gaps +1、p4 +1 移植）。

cloud/chv/executor.py

@@ -250,8 +250,9 @@ class CloudHypervisorExecutor(Executor):
       3-4294967295 内，0/1/2 为保留 CID）
     - CHV_CELL_ENABLED：nspawn 单元开关（默认 1 开；只认
       0/1，非法值 ValueError）。开时客内 host.py 改走
-      chv-cell run（P2 差距表 §4）；关时走原命令，给
-      lite10 及更早无单元的旧镜像留逃生口
+      sudo chv-cell run（P2 差距表 §4；round-11 真机修复：
+      chv-cell 须 root，执行器以 chv 身份调，须 sudo）；
+      关时走原命令，给 lite10 及更早无单元的旧镜像留逃生口
 
     前置校验（preflight，执行入口必调，缺一即 fail-fast）：
     收集全部缺失项一次报出具体缺哪一项，绝不笼统报错：
@@ -1226,6 +1227,19 @@ class CloudHypervisorExecutor(Executor):
         return normalize_api_result(
             self.api_client(socket_path, method, path, payload))
 
+    @staticmethod
+    def _truncate_api_body(body, limit=500):
+        """把 API 响应体压成单行短串（放进失败消息里定位用）。
+        vm.create/vm.boot 的错误体是 VMM 的报错文本，不含秘密；
+        超长即截断并标明，避免日志被巨体淹没。"""
+        if body is None:
+            return ""
+        text = body if isinstance(body, str) else repr(body)
+        text = " ".join(text.split())
+        if len(text) > limit:
+            return text[:limit] + "…（已截断）"
+        return text
+
     def _call_runner(self, cmd, timeout=None):
         """runner 调用（共享 helper 在 cloud/sandbox.py）：
         timeout 给定时优先传 timeout，假 runner 只接受 cmd 时
@@ -1462,11 +1476,15 @@ class CloudHypervisorExecutor(Executor):
         放在该路径（与 Docker 的 /app:ro 同款约定，只是载体
         从容器换成 VM 镜像）。
         round-11 集成补洞（P2 差距表 §4）：单元开时（默认）
-        命令改为 chv-cell run -- python3 /app/host.py --state
+        命令改为 sudo chv-cell run -- python3 /app/host.py --state
         /home/chv/.codex/host-state.json --codex-home
         /home/chv/.codex … --sandbox danger-full-access [子命令
-        ...]。--state/--codex-home 须显式给（见模块顶 _CELL_*
-        注释）；--sandbox 放全局参数最后（argparse 后胜出），
+        ...]。sudo 不可省：chv-cell 须 root（nspawn + 配网），
+        执行器经 SSH 以 chv 身份调，chv 的 NOPASSWD sudo 来自
+        seed users 块（见 test_chv_gitea_sync.py 的 users 夹具；
+        探针 p2.log 的手动调用也是 sudo chv-cell run）。
+        --state/--codex-home 须显式给（见模块顶 _CELL_* 注释）；
+        --sandbox 放全局参数最后（argparse 后胜出），
         确保单元内恒为 danger-full-access：单元即沙箱，调用
         方传的 read-only（plan 任务）与 approval=full 自带的
         值都不得把它盖回 bwrap。单元关时
@@ -1476,7 +1494,7 @@ class CloudHypervisorExecutor(Executor):
         cell = self._cell_enabled()
         cmd = ["python3", self.guest_host_py]
         if cell:
-            cmd = ["chv-cell", "run", "--"] + cmd
+            cmd = ["sudo", "chv-cell", "run", "--"] + cmd
             cmd += ["--state", _CELL_STATE_PATH,
                     "--codex-home", _CELL_CODEX_HOME]
         cmd += _host_command_options(
@@ -1864,13 +1882,14 @@ class CloudHypervisorExecutor(Executor):
         self._launch_api_only(task)
         self.wait_api_ready(task)
         self._check_api_socket_permissions(socket_path)
-        status, _ = self._api_call(
+        status, body = self._api_call(
             socket_path, "PUT", "/vm.create",
             self.build_vm_config(task_disk, seed_disk=seed_disk))
         if status not in (200, 204):
             raise ChvFailure(
                 "vm_create_failed",
-                "vm.create 失败：status=%s" % status)
+                "vm.create 失败：status=%s body=%s"
+                % (status, self._truncate_api_body(body)))
         self._emit("vm_create", task,
                    "vm=%s disk=%s" % (name, task_disk))
         if net_entries:
@@ -1881,11 +1900,12 @@ class CloudHypervisorExecutor(Executor):
                     "net_registry_write_failed",
                     "vm=%s net 登记写失败，拒绝启动（缺登记会"
                     "漏判 tap/mac/ip 冲突）" % name)
-        status, _ = self._api_call(socket_path, "PUT", "/vm.boot")
+        status, body = self._api_call(socket_path, "PUT", "/vm.boot")
         if status not in (200, 204):
             raise ChvFailure(
                 "vm_boot_failed",
-                "vm.boot 失败：status=%s" % status)
+                "vm.boot 失败：status=%s body=%s"
+                % (status, self._truncate_api_body(body)))
         self._emit("vm_boot", task, "vm=%s" % name)
         self.wait_guest_ready(task)
         # V-9 移植：密钥已投递时等 guest-env 对 SSH 用户可读再报
@@ -2622,7 +2642,7 @@ class CloudHypervisorExecutor(Executor):
         ValueError（_to_switch 的 CHV_BALLOON/CHV_VSOCK 同款
         口径，绝不静默 fallback）。关时 _guest_host_command
         走原命令，给 lite10 及更早无单元的旧镜像留逃生口；
-        开时走 chv-cell run -- python3 /app/host.py …。读
+        开时走 sudo chv-cell run -- python3 /app/host.py …。读
         self._env（与 _resident_enabled 同款惰性读法），不
         经 config：开关只影响命令构造，不影响建池/起 VM。"""
         flag = _to_switch(

cloud/chv/hostkeys.py

@@ -193,10 +193,16 @@ def build_seed_user_data(private_key_text: str, public_key_text: str) -> str:
 
 def _authd_env_entry(authd_env: dict) -> list:
     """authd-seed 的 write_files 条目行（0600 root，值经
-    shlex.quote 引用）：只装 AUTHD_SECRET_KEYS 命中的密钥，
-    与 guest-env 条目并列写进同一 write_files 列表（绝不另写
-    一个 write_files: 键，见 _guest_env_entry 的 YAML 重复键
-    禁令）。"""
+    shlex.quote 引用）：装 AUTHD_SECRET_KEYS 命中的密钥
+    （本期只有 GITEA_TOKEN），外加与 guest-env 同源抄进来的
+    三个非密钥同步变量（GITEA_URL/GITEA_USER/SYNC_REPO——
+    round-11 真机修复层 2：authd 的 build_server 要求 seed
+    里同时有这三键，只 GITEA_TOKEN 会 ValueError；探针
+    fixprobe3 即从 guest-env 补这三键）。三键只在调用方
+    guest-env 里有才抄，没有不补、不编默认值；与 guest-env
+    条目并列写进同一 write_files 列表（绝不另写一个
+    write_files: 键，见 _guest_env_entry 的 YAML 重复键禁令）。
+    """
     lines = [
         "  - path: %s" % AUTHD_SEED_PATH,
         "    permissions: '0600'",
@@ -296,12 +302,18 @@ def _write_seed_meta(seed_dir: str, vm_name: str) -> str:
 def _seed_env_entries(merged_env: dict,
                       authd_env: dict | None) -> list:
     """guest-env 与 authd-seed 两路的 write_files 条目行
-    （皆空即空列表，不写条目）。"""
+    （皆空即空列表，不写条目）。authd-seed 除了密钥还抄
+    guest-env 里的 GITEA_URL/GITEA_USER/SYNC_REPO（层 2，
+    见 _authd_env_entry）——build_server 要求四键齐全。"""
     extra_lines = []
     if merged_env:
         extra_lines += _guest_env_entry(merged_env)
     if authd_env:
-        extra_lines += _authd_env_entry(authd_env)
+        seed_env = dict(authd_env)
+        for key in ("GITEA_URL", "GITEA_USER", "SYNC_REPO"):
+            if key in merged_env:
+                seed_env[key] = merged_env[key]
+        extra_lines += _authd_env_entry(seed_env)
     return extra_lines
 
 

cloud/chv/sentinel.py

@@ -93,7 +93,9 @@ _BLOCKED_NETWORKS = tuple(
 
 # 每连接器的 L7 路径策略：(方法, 匹配方式, 模式)。
 # gitea：API 只放 /api/v1/；git 智能 HTTP 只放三条（clone/pull/push
-# 的实际路径，见 git http-protocol）。其它路径（Web UI、登录页、
+# 的实际路径，见 git http-protocol）加裸 .git（round-11 真机修复
+# 层 4：git 智能 HTTP 的发现请求只发裸 .git 路径，裸路径只放 GET、
+# 一律当 pull；POST 裸 .git 仍拒绝）。其它路径（Web UI、登录页、
 # 任意管理接口）一律不在本连接器的能力内。
 # feat-chv-muse-p9 的 inference 条目：推理连接器只放
 # POST /v1/responses 与 GET /v1/models（/v1/models 是否必需待
@@ -104,6 +106,7 @@ _CONNECTOR_PATH_RULES = {
         ("GET", "prefix", "/api/v1/"),
         ("POST", "prefix", "/api/v1/"),
         ("GET", "suffix", ".git/info/refs"),
+        ("GET", "suffix", ".git"),
         ("POST", "suffix", ".git/git-upload-pack"),
         ("POST", "suffix", ".git/git-receive-pack"),
     ),

cloud/tests/test_chv_closeup.py

@@ -224,6 +224,42 @@ class WakeFailureTest(CloseupCase):
         with self.assertRaises(ChvFailure) as ctx:
             ex.wake_vm(self._task())
         self.assertEqual(ctx.exception.kind, "vm_boot_failed")
+
+    def test_wake_vm_boot_failed_records_body(self):
+        # part 4：vm.boot 500 只留 status 即无法定位——body 要进
+        # 失败消息（截断后，见 _truncate_api_body）。
+        def api(socket_path, method, path, payload=None):
+            if method == "PUT" and path == "/vm.boot":
+                return (500, {"error": "boom-detail-500"})
+            return (200, {})
+        ex = self.make_exec(api_client=api)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_boot_failed")
+        self.assertIn("boom-detail-500", str(ctx.exception))
+        self.assertIn("status=500", str(ctx.exception))
+
+    def test_wake_vm_create_failed_records_body(self):
+        def api(socket_path, method, path, payload=None):
+            if method == "PUT" and path == "/vm.create":
+                return (500, "No bootitem named default")
+            return (200, {})
+        ex = self.make_exec(api_client=api)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertEqual(ctx.exception.kind, "vm_create_failed")
+        self.assertIn("No bootitem named default", str(ctx.exception))
+
+    def test_wake_vm_boot_failed_body_truncated(self):
+        def api(socket_path, method, path, payload=None):
+            if method == "PUT" and path == "/vm.boot":
+                return (500, {"error": "x" * 2000})
+            return (200, {})
+        ex = self.make_exec(api_client=api)
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm(self._task())
+        self.assertIn("已截断", str(ctx.exception))
+        self.assertLess(len(str(ctx.exception)), 1200)
         self.assertIn("vm_failed", self._audit_events())
         self.assertEqual(ex._vms, {})
 

cloud/tests/test_chv_gaps.py

@@ -743,6 +743,31 @@ class ChvSyncTest(unittest.TestCase):
         r = self._sync("boot", env_extra={"CHV_GUEST_ENV": env_file})
         self.assertNotEqual(r.returncode, 0)
 
+    def test_boot_without_authd_fails_closed_no_credentials_file(self):
+        # round-11 真机修复（层 5）：authd 不在时 boot 直接失败，
+        # 绝不把 token 写进 ~/.git-credentials（旧落盘回落已废止；
+        # 即使 guest-env 里有 token 也不写）。
+        if os.getuid() != 0:
+            self.skipTest("boot 须以 root 运行")
+        env_file = os.path.join(self.dir, "guest-env")
+        with open(env_file, "w") as fh:
+            fh.write("GITEA_URL=https://gitea.tose.top\n"
+                     "GITEA_USER=admin\n"
+                     "SYNC_REPO=admin/chv-sync-test\n"
+                     "GITEA_TOKEN=fake-token-for-test\n")
+        home = os.path.join(self.dir, "home")
+        os.makedirs(home, exist_ok=True)
+        r = self._sync("boot", env_extra={
+            "CHV_GUEST_ENV": env_file,
+            "CHV_USER_HOME": home,
+            "CHV_USER": "root",
+            "CHV_AUTHD_SOCKET": os.path.join(self.dir, "no-authd.sock"),
+        })
+        self.assertNotEqual(r.returncode, 0)
+        self.assertIn("chv-authd 未运行", r.stderr)
+        self.assertFalse(
+            os.path.exists(os.path.join(home, ".git-credentials")))
+
 
 # -- (6) 中途死亡 → interrupted → 断点续跑闭环 ---------------------------
 class MidDeathLoopTest(ChvGapCase):

cloud/tests/test_chv_muse_p2.py

@@ -37,6 +37,34 @@ GUARD = _load_script("chv_browser_guard.py")
 BUILDINFO = _load_script("buildinfo.py")
 
 
+def _load_host():
+    """把仓库根的 host.py 当模块加载（只含定义，__main__ 才跑
+    CLI，导入无副作用），供 worktree 根选择逻辑单测。"""
+    path = ROOT / "host.py"
+    spec = importlib.util.spec_from_file_location(
+        "chv_host_py", str(path))
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+_HOST = None
+
+
+def _host():
+    """懒加载 host.py（缓存）。合入 round-12 时由模块级 eager
+    改懒加载：host.py 顶层 `import queue` 会把 stdlib queue
+    缓进 sys.modules，使同进程后加载的 test_queue_sandbox
+    的 `from queue import TaskQueue`（依赖 cloud/queue.py
+    遮蔽 stdlib）取到 stdlib 而 ImportError——全量 2476 项
+    里唯一因此失败的模块。host.py 只在 worktree 测试里用，
+    首次使用时再加载即可。"""
+    global _HOST
+    if _HOST is None:
+        _HOST = _load_host()
+    return _HOST
+
+
 class CellArgvTest(unittest.TestCase):
     def test_private_users_maps_cell_root_to_chv(self):
         argv = CELL.build_nspawn_argv(["true"])
@@ -44,7 +72,15 @@ class CellArgvTest(unittest.TestCase):
         self.assertIn("--machine=chv-cell", argv)
         self.assertIn("--directory=/", argv)
         self.assertIn("--as-pid2", argv)
-        self.assertIn("--no-new-privileges", argv)
+        # round-11 真机修复：nspawn 255 拒裸 --no-new-privileges
+        #（当作带必填参数的选项）与裸 --directory=/（须
+        # --volatile/--ephemeral）。复测纠正：--ephemeral 在客内
+        # ext4 根盘上做整盘快照拷贝，2.4G 盘只剩 593M 即 ENOSPC，
+        # 改 --volatile=yes（tmpfs 根 + 只读 /usr，不拷盘）。
+        self.assertIn("--no-new-privileges=yes", argv)
+        self.assertNotIn("--no-new-privileges", argv)
+        self.assertIn("--volatile=yes", argv)
+        self.assertNotIn("--ephemeral", argv)
 
     def test_muse_named_capabilities_dropped(self):
         argv = CELL.build_nspawn_argv(["true"])
@@ -71,10 +107,17 @@ class CellArgvTest(unittest.TestCase):
             playwright_package_dir="/usr/local/lib/python3.12/dist-packages/playwright")
         hidden = [a.split("=", 1)[1] for a in argv
                   if a.startswith("--inaccessible=")]
-        for path in ("/etc/chv", "/home/chv/.git-credentials",
-                     "/usr/local/share/ms-playwright",
+        # /usr 下的浏览器路径仍遮蔽（/usr 只读挂进单元，遮蔽有效）。
+        for path in ("/usr/local/share/ms-playwright",
                      "/usr/local/lib/python3.12/dist-packages/playwright"):
             self.assertIn(path, hidden)
+        # round-11 复测纠正（第二轮）：/etc/chv 与
+        # /home/chv/.git-credentials 不再列出——--volatile=yes 下
+        # 容器根是基本空的 tmpfs，这两处在容器根里根本不存在，
+        # 凭证文件按构造即不可达；列出反而让 nspawn 解析失败
+        #（Failed to resolve）而起不来。
+        for path in ("/etc/chv", "/home/chv/.git-credentials"):
+            self.assertNotIn(path, hidden)
 
     def test_writable_binds_and_tmpfs(self):
         argv = CELL.build_nspawn_argv(["true"])
@@ -396,7 +439,12 @@ class BrowserTest(unittest.TestCase):
             self.assertTrue(resp["ok"])
             self.assertEqual(resp["title"], "Example Domain")
             self.assertEqual(resp["url"], "https://example.com/")
+            # P4：a11y 是 broker 回包的 dict 树（经 guard 截断）。
+            # round-11 的 ARIA 字符串纠正（Page.aria_snapshot()）
+            # 在 P4 架构里应落在 chv-cdp-broker.py 的 snapshot，
+            # 本轮合入未移植（见合入报告未决问题），此处不断言 str。
             self.assertEqual(resp["a11y"]["role"], "WebArea")
+            self.assertFalse(resp["truncated"])
             snap = agent.snapshot()
             self.assertTrue(snap["ok"])
             shot = agent.screenshot()
@@ -405,6 +453,32 @@ class BrowserTest(unittest.TestCase):
             self.assertTrue(os.path.exists(shot["path"]))
             self.assertTrue(agent.close()["ok"])
 
+    def test_browser_op_dispatches_on_action(self):
+        # round-11 复测回归：broker 原样转发 {"op": "browser", …}，
+        # chv-browser 白名单却不认 "browser"。内层 op 在 action 里；
+        # 无 action 但带 url 时默认 open（驱动只发 url）。
+        agent = mock.Mock()
+        agent.open.return_value = {"ok": True}
+        resp = BROWSER.handle_request(
+            agent, {"op": "browser", "action": "open",
+                    "url": "https://example.com"})
+        self.assertTrue(resp["ok"])
+        agent.open.assert_called_once_with("https://example.com")
+        agent.reset_mock()
+        resp = BROWSER.handle_request(
+            agent, {"op": "browser", "url": "https://example.com"})
+        self.assertTrue(resp["ok"])
+        agent.open.assert_called_once_with("https://example.com")
+        # 内层 forbidden 同样拒绝。
+        agent.reset_mock()
+        resp = BROWSER.handle_request(
+            agent, {"op": "browser", "action": "evaluate"})
+        self.assertFalse(resp["ok"])
+        agent.open.assert_not_called()
+        # 无 action 也无 url：无法推断内层 op。
+        resp = BROWSER.handle_request(agent, {"op": "browser"})
+        self.assertFalse(resp["ok"])
+
 
 class FakeCdpBroker:
     """假 CDP broker：收到 req dict 返回 resp dict；open 后才有页。
@@ -537,73 +611,621 @@ class BuildInfoTest(unittest.TestCase):
 
 
 class BuildDryRunTest(unittest.TestCase):
+    def _fixture(self, tmp, with_vsock=False):
+        # 隔离的临时 git 仓：工作树干净、HEAD 与 REPO_COMMIT
+        # 一致、假 codex 二进制报对版本。with_vsock 时一并拷入
+        # vsock_notify.py（round-11 真机修复：服务须随文件建）。
+        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",
+                     "authd.py", "git-credential-chv-authd",
+                     "chv-authd.service", "authd_init_stub.py",
+                     # round-12 P4：build.sh 缺一即 fail-fast，夹具
+                     # 须一并拷入（合入时补：R11 夹具本无此四件）。
+                     "chv-cdp-broker.py", "chv_browser_guard.py",
+                     "browser-blocklist.txt", "chv-memory.py"):
+            shutil.copy2(ROOT / "chv-guest" / name,
+                         os.path.join(guest, name))
+        shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
+        # round-11：build.sh 要求 P3 云端叶子模块在仓里（fail-fast），
+        # 夹具须一并拷入，否则干跑在 P3 预检就失败。
+        for rel in ("cloud/chv/authd.py", "cloud/chv/sentinel.py",
+                    "cloud/chv/errors.py",
+                    "cloud/__init__.py", "cloud/chv/__init__.py"):
+            dest = os.path.join(repo, rel)
+            os.makedirs(os.path.dirname(dest), exist_ok=True)
+            shutil.copy2(ROOT / rel, dest)
+        if with_vsock:
+            rel = "cloud/chv/guest/vsock_notify.py"
+            dest = os.path.join(repo, rel)
+            os.makedirs(os.path.dirname(dest), exist_ok=True)
+            shutil.copy2(ROOT / rel, dest)
+        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", "p2 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,
+        })
+        return repo, guest, env
+
+    def _dry_run(self, repo, guest, env):
+        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)
+        return proc.stdout + proc.stderr
+
     def test_dry_run_contains_p2_steps(self):
-        # 在隔离的临时 git 仓里跑 build.sh --dry-run：工作树干净、
-        # HEAD 与 REPO_COMMIT 一致、假 codex 二进制报对版本。
+        # 在隔离的临时 git 仓里跑 build.sh --dry-run。
         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",
-                         "authd.py", "git-credential-chv-authd",
-                         "chv-authd.service",
-                         "chv-cdp-broker.py", "chv_browser_guard.py",
-                         "browser-blocklist.txt", "chv-memory.py"):
-                shutil.copy2(ROOT / "chv-guest" / name,
-                             os.path.join(guest, name))
-            shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
-            # round-11：build.sh 要求 P3 云端叶子模块在仓里（fail-fast），
-            # 夹具须一并拷入，否则干跑在 P3 预检就失败。
-            for rel in ("cloud/chv/authd.py", "cloud/chv/sentinel.py",
-                        "cloud/__init__.py", "cloud/chv/__init__.py"):
-                dest = os.path.join(repo, rel)
-                os.makedirs(os.path.dirname(dest), exist_ok=True)
-                shutil.copy2(ROOT / rel, dest)
-            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", "p2 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
+            repo, guest, env = self._fixture(tmp)
+            out = self._dry_run(repo, guest, env)
             for needle in ("systemd-container", "chv-cell", "chv-broker",
                            "chv-browser", "chv-browser.service",
                            "chv-broker.service", "useradd --system",
                            "systemctl enable chv-broker chv-browser",
                            "chv-authd", "git-credential-chv-authd",
                            "systemctl enable chv-authd",
-                           "systemctl start chv-authd"):
+                           "systemctl start chv-authd",
+                           # round-11 真机修复（缺陷 4）：browser.sock
+                           # 移进 chv-browser 自有目录，build.sh 须
+                           # 建目录 + tmpfiles 登记。
+                           "/run/chv-browser",
+                           # round-11 真机修复（层 1）：叶子树含
+                           # errors.py + stub __init__，不拷真实的
+                           # cloud/chv/__init__.py。
+                           "errors.py", "authd_init_stub.py"):
                 self.assertIn(needle, out, needle)
             # P1 文件不在本分支：干跑只警告、不失败。
             self.assertIn("vsock_notify.py 不在仓库里", out)
+            # round-11 真机修复（层 5）：镜像不再设 store helper
+            #（authd helper 是唯一凭证 helper），git user 仍配。
+            self.assertNotIn("credential.helper store", out)
+            self.assertIn("git config --system user.name chv", out)
+
+    def test_dry_run_with_vsock_creates_service(self):
+        # round-11 真机修复（vsock 自动启动）：vsock_notify.py
+        # 在场时建 vsock-notify.service 并 enable，不再只警告。
+        with tempfile.TemporaryDirectory() as tmp:
+            repo, guest, env = self._fixture(tmp, with_vsock=True)
+            out = self._dry_run(repo, guest, env)
+            self.assertNotIn("vsock_notify.py 不在仓库里", out)
+            for needle in ("vsock-notify.service",
+                           "systemctl enable vsock-notify",
+                           "ExecStart=/usr/local/bin/vsock-notify"):
+                self.assertIn(needle, out, needle)
+
+
+class CellGateTest(unittest.TestCase):
+    """part 4：起步门包装。"""
+
+    def test_gate_wrapper_waits_then_execs(self):
+        wrapped = CELL.gate_wrapper_command(
+            ["python3", "/app/host.py", "status"],
+            "/run/chv/.cell-gate-chv-cell-00")
+        self.assertEqual(wrapped[:4], [
+            "/usr/bin/sh", "-c",
+            "while [ ! -e \"$1\" ]; do sleep 0.05; done; "
+            "shift; exec \"$@\"",
+            "chv-cell-gate"])
+        self.assertEqual(wrapped[4],
+                         "/run/chv/.cell-gate-chv-cell-00")
+        self.assertEqual(wrapped[5:],
+                         ["python3", "/app/host.py", "status"])
+        # 绝对路径：volatile 单元根里 /bin 未必有，/usr 必有。
+        self.assertTrue(wrapped[0].startswith("/usr/bin/"))
+
+
+class CellInitPidPollTest(unittest.TestCase):
+    """part 4：find_cell_init_pid 带超时轮询。"""
+
+    def _proc_root_with_child(self, tmp, pid=777, child=778):
+        task_dir = os.path.join(tmp, str(pid), "task", str(pid))
+        os.makedirs(task_dir)
+        with open(os.path.join(task_dir, "children"), "w") as fh:
+            fh.write(str(child))
+
+    def test_single_shot_default_unchanged(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._proc_root_with_child(tmp)
+            self.assertEqual(
+                CELL.find_cell_init_pid(777, proc_root=tmp), 778)
+            self.assertIsNone(
+                CELL.find_cell_init_pid(999, proc_root=tmp))
+
+    def test_poll_finds_late_child(self):
+        import threading
+        with tempfile.TemporaryDirectory() as tmp:
+            def _late():
+                import time as _time
+                _time.sleep(0.3)
+                self._proc_root_with_child(tmp)
+            threading.Thread(target=_late, daemon=True).start()
+            self.assertEqual(
+                CELL.find_cell_init_pid(777, proc_root=tmp,
+                                        timeout=5.0), 778)
+
+    def test_poll_returns_none_on_timeout(self):
+        import time as _time
+        with tempfile.TemporaryDirectory() as tmp:
+            start = _time.time()
+            self.assertIsNone(
+                CELL.find_cell_init_pid(999, proc_root=tmp,
+                                        timeout=0.3))
+            self.assertLess(_time.time() - start, 5.0)
+
+    def test_poll_stops_when_nspawn_dead(self):
+        import time as _time
+        with tempfile.TemporaryDirectory() as tmp:
+            start = _time.time()
+            self.assertIsNone(
+                CELL.find_cell_init_pid(999, proc_root=tmp,
+                                        timeout=10.0,
+                                        proc_alive=lambda: False))
+            # nspawn 已死即立刻回，不空等 10 秒。
+            self.assertLess(_time.time() - start, 5.0)
+
+
+class CellPart4ArgvTest(unittest.TestCase):
+    """part 4：volatile 空根补 bind + 唯一身份进 argv。"""
+
+    def test_app_and_ca_bundle_bound_read_only(self):
+        argv = CELL.build_nspawn_argv(["true"])
+        self.assertIn("--bind-ro=/app", argv)
+        self.assertIn("--bind-ro=/etc/ssl/certs", argv)
+        # /app 是只读绑：单元内 python3 /app/host.py 只读，不写。
+        self.assertNotIn("--bind=/app", argv)
+
+    def test_extra_ro_binds(self):
+        argv = CELL.build_nspawn_argv(
+            ["true"], extra_ro_binds=[
+                ("/run/chv/resolv.conf.chv-cell-00", "/etc/resolv.conf"),
+                ("/run/chv/hosts.chv-cell-00", "/etc/hosts")])
+        self.assertIn(
+            "--bind-ro=/run/chv/resolv.conf.chv-cell-00:/etc/resolv.conf",
+            argv)
+        self.assertIn(
+            "--bind-ro=/run/chv/hosts.chv-cell-00:/etc/hosts", argv)
+
+
+
+    def test_cell_net_machine_and_no_legacy_name(self):
+        net = CELL.CellNet(index=3, machine="chv-cell-03",
+                           host_iface="ve-chv-cell-03",
+                           host_ip="10.200.3.1", cell_ip="10.200.3.2",
+                           subnet="10.200.3.0/24")
+        argv = CELL.build_nspawn_argv(["true"], cell_net=net)
+        self.assertIn("--machine=chv-cell-03", argv)
+        self.assertNotIn("--machine=chv-cell", argv)
+        # 旧常量身份仍是默认（argv 子命令与旧测试不改）。
+        self.assertIn("--machine=chv-cell",
+                      CELL.build_nspawn_argv(["true"]))
+
+    def test_legacy_cell_net(self):
+        net = CELL.legacy_cell_net()
+        self.assertEqual(net.machine, "chv-cell")
+        self.assertEqual(net.host_iface, "ve-chv-cell")
+        self.assertEqual(net.subnet, "10.200.0.0/24")
+
+
+class CellPart6WorktreeTest(unittest.TestCase):
+    """part 6 修复（RETEST5 新发现）：worktree 根移出 /app。
+
+    part 5 的旧测试钉住的是 argv 顺序（/app/work 绑定排在
+    --bind-ro=/app 之后），RETEST5 真机证明该顺序本身不可行：
+    后挂的 bind 只有在挂载点已存在时才能盖住，挂载点不存在时
+    nspawn 在只读 /app 下连建都建不起来（全量 73 次同一错误，
+    CELL-RC=0 共 0 次）。本轮单测不再钉顺序，改钉语义：argv 里
+    不得出现任何指向只读 /app 之下的 bind（挂载点不存在时不发
+    致命 bind），worktree 根改经 CHV_WORKTREE_ROOT 环境变量传
+    进单元，落点在已可写绑进单元的 workspace 之下。"""
+
+    @staticmethod
+    def _bind_dests(argv):
+        """argv 里全部 bind/bind-ro 的单元内目的路径。"""
+        dests = []
+        for token in argv:
+            for prefix in ("--bind=", "--bind-ro="):
+                if token.startswith(prefix):
+                    spec = token[len(prefix):]
+                    dests.append(spec.split(":", 1)[1]
+                                 if ":" in spec else spec)
+        return dests
+
+    def test_no_bind_targets_anything_under_readonly_app(self):
+        argv = CELL.build_nspawn_argv(["true"])
+        # RETEST5 的致命绑定本身必须消失。
+        self.assertNotIn(
+            "--bind=/home/chv/workspace/.chv-worktrees:/app/work",
+            argv)
+        # 更一般的语义：任何 bind 的目的路径都不得落在只读的
+        # /app 之下（/app 本身是 --bind-ro 的挂载点，允许；其
+        # 子路径挂载点在镜像里不存在，nspawn 建不了即 EROFS）。
+        for dest in self._bind_dests(argv):
+            self.assertFalse(
+                dest.startswith("/app/"),
+                "bind 目的路径在只读 /app 之下：%s" % dest)
+
+    def test_worktree_root_passed_via_setenv(self):
+        argv = CELL.build_nspawn_argv(["true"])
+        self.assertIn(
+            "--setenv=CHV_WORKTREE_ROOT="
+            "/home/chv/workspace/.chv-worktrees",
+            argv)
+
+    def test_worktree_dir_reachable_via_writable_workspace_bind(self):
+        # worktree 根在 workspace 之下，且 workspace 本身已以
+        # 可写 bind（非 --bind-ro）进单元——根无需额外 bind，
+        # 挂载点问题按构造即不存在。
+        self.assertTrue(
+            CELL.WORKTREE_DIR.startswith(CELL.WORKSPACE + "/"))
+        argv = CELL.build_nspawn_argv(["true"])
+        self.assertIn("--bind=%s" % CELL.WORKSPACE, argv)
+        self.assertNotIn("--bind-ro=%s" % CELL.WORKSPACE, argv)
+        # .chv-worktrees 只应出现在 --setenv 里，不应出现在
+        # 任何 bind/bind-ro 参数里。
+        for token in argv:
+            if ".chv-worktrees" in token:
+                self.assertTrue(
+                    token.startswith("--setenv=CHV_WORKTREE_ROOT="),
+                    "worktree 目录不应再经 bind 进单元：%s" % token)
+
+    def test_worktree_dir_matches_host_cell_fallback(self):
+        # chv-cell 的 WORKTREE_DIR 与 host.py 在 CHV_CELL=1 且
+        # 环境变量未设时的回落值须是同一路径，否则两边各说各话。
+        self.assertEqual(CELL.WORKTREE_DIR, _host().CELL_WORKTREE_ROOT)
+
+
+class HostWorktreeBaseTest(unittest.TestCase):
+    """part 6：host.py 的 worktree 根选择（worktree_base）与
+    make_worktree 落点——单元内永不在 /app 之下，单元外不变。"""
+
+    def test_env_override_wins_over_cell_fallback(self):
+        with mock.patch.dict(
+                os.environ,
+                {"CHV_WORKTREE_ROOT": "/tmp/wt-root",
+                 "CHV_CELL": "1"},
+                clear=True):
+            self.assertEqual(
+                _host().worktree_base(), "/tmp/wt-root/host-worktrees")
+
+    def test_cell_without_env_falls_back_to_workspace(self):
+        with mock.patch.dict(os.environ, {"CHV_CELL": "1"},
+                             clear=True):
+            base = _host().worktree_base()
+            self.assertEqual(
+                base,
+                "/home/chv/workspace/.chv-worktrees/host-worktrees")
+            self.assertFalse(base.startswith("/app/"))
+
+    def test_outside_cell_uses_here_work(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(
+                _host().worktree_base(),
+                os.path.join(_host().HERE, "work", "host-worktrees"))
+
+    def _make_git_repo(self, tmp):
+        repo = os.path.join(tmp, "repo")
+        os.makedirs(repo)
+        for args in (["init", "-q", repo],
+                     ["-C", repo, "config", "user.email",
+                      "test@example.invalid"],
+                     ["-C", repo, "config", "user.name", "test"]):
+            subprocess.run(["git"] + args, check=True,
+                           capture_output=True)
+        with open(os.path.join(repo, "README.md"), "w") as fh:
+            fh.write("hi\n")
+        subprocess.run(["git", "-C", repo, "add", "."],
+                       check=True, capture_output=True)
+        subprocess.run(
+            ["git", "-C", repo, "commit", "-qm", "init"],
+            check=True, capture_output=True)
+        return repo
+
+    def test_make_worktree_uses_override_not_app(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = self._make_git_repo(tmp)
+            wt_root = os.path.join(tmp, "wt")
+            with mock.patch.dict(
+                    os.environ,
+                    {"CHV_WORKTREE_ROOT": wt_root, "CHV_CELL": ""},
+                    clear=False):
+                path, branch = _host().make_worktree(repo, "t1")
+            self.assertEqual(
+                path, os.path.join(wt_root, "host-worktrees", "t1"))
+            self.assertEqual(branch, "host/t1")
+            self.assertFalse(path.startswith("/app/"))
+            # 真是 git worktree，不是空目录拷贝。
+            proc = subprocess.run(
+                ["git", "-C", path, "rev-parse", "--show-toplevel"],
+                capture_output=True, text=True)
+            self.assertEqual(proc.returncode, 0)
+            self.assertEqual(proc.stdout.strip(), path)
+
+    def test_dispatch_from_branch_uses_same_base(self):
+        # 第二个 HERE/work 调用方（依赖派发）须与 make_worktree
+        # 走同一根，否则 goal/fork 的依赖链仍撞只读 /app。
+        with tempfile.TemporaryDirectory() as tmp:
+            repo = self._make_git_repo(tmp)
+            wt_root = os.path.join(tmp, "wt")
+            with mock.patch.dict(
+                    os.environ,
+                    {"CHV_WORKTREE_ROOT": wt_root, "CHV_CELL": ""},
+                    clear=False):
+                dep_path, _ = _host().make_worktree(repo, "dep")
+                created = {}
+
+                class _FakeServer:
+                    def create_thread(self, path, **kwargs):
+                        created["path"] = path
+                        return "th-1"
+
+                state = {"tasks": {}}
+                task, err = _host().dispatch_from_branch(
+                    _FakeServer(), state, "child", "prompt", repo,
+                    "host/dep", dep_path, "m", "low",
+                    "danger-full-access", "ask")
+            self.assertEqual(err, "")
+            self.assertEqual(
+                task["cwd"],
+                os.path.join(wt_root, "host-worktrees", "child"))
+            self.assertFalse(task["cwd"].startswith("/app/"))
+
+
+class CellFilesTest(unittest.TestCase):
+    """part 4：单元专属 resolv.conf/hosts 生成。"""
+
+    def _write(self, path, content):
+        with open(path, "w") as fh:
+            fh.write(content)
+        return path
+
+    def test_resolv_filters_stub_and_falls_back(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            stub = self._write(os.path.join(tmp, "resolv.conf"),
+                               "nameserver 127.0.0.53\n"
+                               "options edns0\n")
+            real = self._write(os.path.join(tmp, "resolved.conf"),
+                               "nameserver 8.8.8.8\n"
+                               "nameserver 1.1.1.1\n")
+            files = CELL.prepare_cell_files(
+                "chv-cell-00", run_dir=tmp, resolv_conf=stub,
+                resolved_conf=real,
+                hosts_file=os.path.join(tmp, "no-hosts"))
+            with open(files["resolv"]) as fh:
+                content = fh.read()
+            self.assertEqual(content,
+                             "nameserver 8.8.8.8\nnameserver 1.1.1.1\n")
+            self.assertNotIn("127.0.0.53", content)
+            with open(files["hosts"]) as fh:
+                self.assertIn("127.0.0.1 localhost", fh.read())
+            self.assertEqual(oct(os.stat(files["resolv"]).st_mode
+                                 & 0o777), "0o644")
+            self.assertEqual(files["gate"],
+                             os.path.join(tmp, ".cell-gate-chv-cell-00"))
+            # gate 文件本身不在这里创建（run_cell 配网后才开门）。
+            self.assertFalse(os.path.exists(files["gate"]))
+
+    def test_resolv_real_servers_used_directly(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            resolv = self._write(os.path.join(tmp, "resolv.conf"),
+                                 "nameserver 8.8.4.4\n")
+            files = CELL.prepare_cell_files(
+                "chv-cell-01", run_dir=tmp, resolv_conf=resolv,
+                resolved_conf=os.path.join(tmp, "missing"),
+                hosts_file=os.path.join(tmp, "missing-hosts"))
+            with open(files["resolv"]) as fh:
+                self.assertEqual(fh.read(), "nameserver 8.8.4.4\n")
+
+    def test_no_dns_anywhere_is_loud(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            with self.assertRaises(RuntimeError):
+                CELL.prepare_cell_files(
+                    "chv-cell-00", run_dir=tmp,
+                    resolv_conf=os.path.join(tmp, "missing"),
+                    resolved_conf=os.path.join(tmp, "missing2"),
+                    hosts_file=os.path.join(tmp, "missing-hosts"))
+
+
+class CellPart4NetworkTest(unittest.TestCase):
+    """part 4：配网命令表走唯一身份。"""
+
+    def _net(self, index=5):
+        return CELL.CellNet(
+            index=index, machine="chv-cell-%02x" % index,
+            host_iface="ve-chv-cell-%02x" % index,
+            host_ip="10.200.%d.1" % index,
+            cell_ip="10.200.%d.2" % index,
+            subnet="10.200.%d.0/24" % index)
+
+    def test_host_commands_use_cell_net(self):
+        net = self._net(5)
+        flat = [" ".join(c)
+                for c in CELL.host_network_commands("enp0s5", net)]
+        self.assertIn("ip addr add 10.200.5.1/24 dev ve-chv-cell-05",
+                      flat)
+        self.assertTrue(any(
+            "iptables -t nat -A POSTROUTING -s 10.200.5.0/24 "
+            "-o enp0s5 -j MASQUERADE" in line for line in flat))
+        # 旧常量仍是默认。
+        legacy = [" ".join(c)
+                  for c in CELL.host_network_commands("enp0s5")]
+        self.assertIn("ip addr add 10.200.0.1/24 dev ve-chv-cell",
+                      legacy)
+
+    def test_cell_commands_use_cell_net(self):
+        net = self._net(7)
+        flat = [" ".join(c)
+                for c in CELL.cell_network_commands(4321, net)]
+        self.assertIn("nsenter -t 4321 -n ip addr add 10.200.7.2/24 "
+                      "dev host0", flat)
+        self.assertIn("nsenter -t 4321 -n ip route add default via "
+                      "10.200.7.1", flat)
+
+
+class CellAllocTest(unittest.TestCase):
+    """part 4：锁下分配唯一 cell 身份。"""
+
+    class _FakeProc:
+        def __init__(self, returncode=0, stdout=""):
+            self.returncode = returncode
+            self.stdout = stdout
+
+    def _fake_run(self, link_out="", addr_out="", machinectl_out="",
+                  machinectl_rc=0):
+        def run(cmd, **kwargs):
+            if cmd[:3] == ["ip", "-o", "link"]:
+                return self._FakeProc(0, link_out)
+            if cmd[:3] == ["ip", "-o", "addr"]:
+                return self._FakeProc(0, addr_out)
+            if cmd[0] == "machinectl":
+                return self._FakeProc(machinectl_rc, machinectl_out)
+            raise AssertionError("意外命令：%r" % (cmd,))
+        return run
+
+    def test_allocates_lowest_free_index(self):
+        link = ("2: ve-chv-cell-00@if2: <UP> mtu 1500\n"
+                "3: ve-chv-cell-01@if3: <UP> mtu 1500\n")
+        addr = ("2: ve-chv-cell-00 inet 10.200.0.1/24 scope global\n"
+                "3: ve-chv-cell-01 inet 10.200.1.1/24 scope global\n")
+        with tempfile.TemporaryDirectory() as tmp:
+            lock = os.path.join(tmp, "lock")
+            net = CELL.allocate_cell_net(
+                run=self._fake_run(link, addr), lock_path=lock)
+        self.assertEqual(net.index, 2)
+        self.assertEqual(net.machine, "chv-cell-02")
+        self.assertEqual(net.host_iface, "ve-chv-cell-02")
+        self.assertEqual(net.host_ip, "10.200.2.1")
+        self.assertEqual(net.cell_ip, "10.200.2.2")
+        self.assertEqual(net.subnet, "10.200.2.0/24")
+        self.assertLessEqual(len(net.host_iface), 15)
+
+    def test_machinectl_registered_index_skipped(self):
+        # 接口已消失、machine 仍在注销：避让该下标。
+        run = self._fake_run(
+            "", "", machinectl_out="chv-cell-00  1234  ...\n")
+        with tempfile.TemporaryDirectory() as tmp:
+            net = CELL.allocate_cell_net(
+                run=run, lock_path=os.path.join(tmp, "lock"))
+        self.assertEqual(net.index, 1)
+
+    def test_machinectl_missing_proceeds(self):
+        def run(cmd, **kwargs):
+            if cmd[0] == "machinectl":
+                raise OSError("无此命令")
+            return self._FakeProc(0, "")
+        with tempfile.TemporaryDirectory() as tmp:
+            net = CELL.allocate_cell_net(
+                run=run, lock_path=os.path.join(tmp, "lock"))
+        self.assertEqual(net.index, 0)
+
+    def test_ip_failure_is_loud(self):
+        def run(cmd, **kwargs):
+            return self._FakeProc(1, "")
+        with tempfile.TemporaryDirectory() as tmp:
+            with self.assertRaises(RuntimeError):
+                CELL.allocate_cell_net(
+                    run=run, lock_path=os.path.join(tmp, "lock"))
+
+
+class BrokerMultiTest(unittest.TestCase):
+    """part 4：broker 对多 cell 的 pid ns 对照。"""
+
+    def _fake_proc(self, root, mapping):
+        for pid, ns in mapping.items():
+            d = os.path.join(root, str(pid), "ns")
+            os.makedirs(d)
+            os.symlink("pid:[%s]" % ns, os.path.join(d, "pid"))
+
+    def test_peer_allowed_if_matches_any_cell(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._fake_proc(tmp, {100: "111", 101: "222", 200: "222"})
+            ok, _ = BROKER.peer_allowed_multi(
+                1000, 200, [100, 101], proc_root=tmp)
+            self.assertTrue(ok)
+
+    def test_peer_denied_if_matches_none(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._fake_proc(tmp, {100: "111", 101: "222", 200: "333"})
+            ok, reason = BROKER.peer_allowed_multi(
+                1000, 200, [100, 101], proc_root=tmp)
+            self.assertFalse(ok)
+            self.assertIn("pid ns", reason)
+
+    def test_stale_pid_file_skipped_not_fatal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            # 100 已退出（无 /proc 项），101 活着且同 ns。
+            self._fake_proc(tmp, {101: "222", 200: "222"})
+            ok, _ = BROKER.peer_allowed_multi(
+                1000, 200, [100, 101], proc_root=tmp)
+            self.assertTrue(ok)
+
+    def test_all_ns_unreadable_rejects(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._fake_proc(tmp, {200: "222"})
+            ok, _ = BROKER.peer_allowed_multi(
+                1000, 200, [100, 101], proc_root=tmp)
+            self.assertFalse(ok)
+
+    def test_no_pid_files_uid_only(self):
+        ok, reason = BROKER.peer_allowed_multi(1000, 200, [])
+        self.assertTrue(ok)
+        self.assertIn("未做 pid ns 对照", reason)
+        # part 5：reason 须保留“cell.pid 缺失”字样——P2-9 驱动
+        # grep 该串认降级日志；part 4 切到 multi 版时改写成
+        # “无 cell pid 文件”，该串再也没进过 journal，
+        # RETEST4 的 degraded_logged 即因此为 False。
+        self.assertIn("cell.pid 缺失", reason)
+
+    def test_degraded_warning_carries_probe_string(self):
+        # serve() 打的警告行就是 _degraded_warning 的返回值：
+        # reason 原样进日志，journalctl grep 'cell.pid 缺失'
+        # 才能命中；pid/uid 也在行里（排障用）。
+        _, reason = BROKER.peer_allowed_multi(1000, 200, [])
+        line = BROKER._degraded_warning(reason, 123, 1000)
+        self.assertIn("cell.pid 缺失", line)
+        self.assertIn("pid=123", line)
+        self.assertIn("uid=1000", line)
+
+    def test_read_cell_pids_glob_and_legacy(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            with open(os.path.join(tmp, "cell-chv-cell-00.pid"),
+                      "w") as fh:
+                fh.write("100")
+            with open(os.path.join(tmp, "cell-chv-cell-01.pid"),
+                      "w") as fh:
+                fh.write("101")
+            with open(os.path.join(tmp, "cell.pid"), "w") as fh:
+                fh.write("101")
+            with open(os.path.join(tmp, "cell-chv-cell-02.pid"),
+                      "w") as fh:
+                fh.write("not-a-pid")
+            self.assertEqual(BROKER._read_cell_pids(tmp), [100, 101])
 
 
 if __name__ == "__main__":

cloud/tests/test_chv_muse_p3.py

@@ -271,14 +271,18 @@ class SentinelTest(unittest.TestCase):
             (self._req(port=80), "port_not_allowed"),
             (self._req(host="bad.example"), "host_denied"),
             (self._req(path="/login"), "path_not_allowed"),
+            # round-11 真机修复（层 4）：裸 .git 只放 GET，POST 仍拒。
+            (self._req(method="POST", path="/admin/r.git"),
+             "path_not_allowed"),
         ]
         for req, reason in cases:
             dec = policy.evaluate(req, resolve=_public_resolver)
             self.assertEqual(dec.verdict, "deny", reason)
             self.assertEqual(dec.reasons, [reason])
-        # git 智能 HTTP 三条路径放行。
+        # git 智能 HTTP 三条路径放行，裸 .git 只放 GET（层 4）。
         for method, path in (
                 ("GET", "/admin/r.git/info/refs"),
+                ("GET", "/admin/r.git"),
                 ("POST", "/admin/r.git/git-upload-pack"),
                 ("POST", "/admin/r.git/git-receive-pack")):
             dec = policy.evaluate(
@@ -367,6 +371,49 @@ class SeedSplitTest(unittest.TestCase):
         self.assertNotIn("GITEA_TOKEN",
                          entries["/etc/chv/guest-env"]["content"])
 
+    def test_write_seed_authd_entry_carries_sync_keys(self):
+        # round-11 真机修复（层 2）：authd-seed 须同时有
+        # GITEA_URL/GITEA_USER/SYNC_REPO，build_server 才会放行
+        #（只 GITEA_TOKEN 时 ValueError，探针 fixprobe3 即从
+        # guest-env 补这三键）；三键只在 guest-env 里有才抄。
+        import yaml
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_dir = os.path.join(tmp, "seed")
+            write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
+                       guest_env={
+                           "GITEA_URL": "https://gitea.tose.top",
+                           "GITEA_USER": "admin",
+                           "SYNC_REPO": "admin/chv-sync-test",
+                           "SYNC_BRANCH": "main"},
+                       authd_env={"GITEA_TOKEN": TOKEN})
+            with open(os.path.join(seed_dir, "user-data")) as fh:
+                parsed = yaml.safe_load(fh.read())
+        entries = {e["path"]: e for e in parsed["write_files"]}
+        content = entries[AUTHD_SEED_PATH]["content"]
+        for key in ("GITEA_URL=https://gitea.tose.top",
+                    "GITEA_USER=admin",
+                    "SYNC_REPO=admin/chv-sync-test",
+                    "GITEA_TOKEN="):
+            self.assertIn(key, content)
+        # 非同步键不进 authd-seed。
+        self.assertNotIn("SYNC_BRANCH", content)
+
+    def test_write_seed_authd_entry_without_sync_keys(self):
+        # guest-env 里没有三键时 authd-seed 只装密钥，不编默认值。
+        import yaml
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_dir = os.path.join(tmp, "seed")
+            write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
+                       authd_env={"GITEA_TOKEN": TOKEN})
+            with open(os.path.join(seed_dir, "user-data")) as fh:
+                parsed = yaml.safe_load(fh.read())
+        entries = {e["path"]: e for e in parsed["write_files"]}
+        content = entries[AUTHD_SEED_PATH]["content"]
+        self.assertIn("GITEA_TOKEN=", content)
+        self.assertNotIn("GITEA_URL", content)
+        self.assertNotIn("GITEA_USER", content)
+        self.assertNotIn("SYNC_REPO", content)
+
 
 # -- (4) wake 预审 -----------------------------------------------------------
 class _FakeResult:
@@ -555,6 +602,28 @@ class GuestAuthdTest(unittest.TestCase):
         self.assertEqual(resp["password"], TOKEN)
         self.assertGreater(resp["expires_at"], 0)
 
+    def test_bare_git_path_is_pull(self):
+        # round-11 真机修复（层 4）：裸 .git 路径一律当 pull。
+        resp = self._ask({"connector": "gitea",
+                          "host": "gitea.tose.top",
+                          "path": "/admin/chv-sync-test.git"})
+        self.assertTrue(resp["ok"])
+        self.assertEqual(resp["username"], "admin")
+        self.assertEqual(resp["password"], TOKEN)
+
+    def test_socket_uses_group_perms(self):
+        # round-11 真机修复（层 3）：socket 用 group 权限（目录
+        # 0750、socket 0660），不用探针的 0666；放行 uid 无
+        # passwd 条目时才回落 root-only（见 SocketGroupTest）。
+        import stat
+        mod = _load_guest_authd()
+        if mod._socket_group_gid([os.getuid()]) is None:
+            self.skipTest("放行 uid 无 passwd 条目，组权限不可解析")
+        self.assertEqual(
+            stat.S_IMODE(os.stat(self.socket_path).st_mode), 0o660)
+        self.assertEqual(
+            stat.S_IMODE(os.stat(self.dir).st_mode), 0o750)
+
     def test_second_request_gets_fresh_release(self):
         for _ in range(2):
             resp = self._ask({"connector": "gitea",
@@ -612,6 +681,60 @@ class GuestAuthdTest(unittest.TestCase):
                              resolve=_public_resolver)
 
 
+class SocketGroupTest(unittest.TestCase):
+    """round-11 真机修复（层 3）：socket 共享组 gid 的派生与回落。"""
+
+    def test_explicit_group_and_derivation(self):
+        mod = _load_guest_authd()
+        # 显式 int 即 gid；显式组名解析；不存在的组名 → None。
+        self.assertEqual(mod._socket_group_gid([0, 1000], 1234), 1234)
+        self.assertEqual(mod._socket_group_gid([0, 1000], "root"), 0)
+        self.assertIsNone(
+            mod._socket_group_gid([0, 1000], "no-such-group-xyz"))
+        # 只剩 root → 无组可派，回落 root-only。
+        self.assertIsNone(mod._socket_group_gid([0]))
+        self.assertIsNone(mod._socket_group_gid([]))
+        # 派生：第一个非 root 放行 uid 的主组（用 passwd 里第一个
+        # 非 root 条目，不依赖当前测试进程的 uid——本机测试以 root
+        # 跑，os.getuid() 本身就是 0，派生不到组）。
+        import pwd
+        others = [p for p in pwd.getpwall() if p.pw_uid != 0]
+        if not others:
+            self.skipTest("passwd 里无非 root 用户")
+        first = sorted(others, key=lambda p: p.pw_uid)[0]
+        self.assertEqual(
+            mod._socket_group_gid([0, first.pw_uid]), first.pw_gid)
+
+    def test_server_applies_group_perms_with_explicit_group(self):
+        # 显式给组时服务器落地 0750/0660（本机以 root 跑测试，
+        # allowed_uids 只有 root，派生不到组，须显式给才能覆到
+        # AuthdServer 的落盘路径）。
+        import stat
+        mod = _load_guest_authd()
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_path = os.path.join(tmp, "authd-seed")
+            with open(seed_path, "w") as fh:
+                fh.write("GITEA_URL=https://gitea.tose.top\n"
+                         "GITEA_USER=admin\n"
+                         "GITEA_TOKEN=" + TOKEN + "\n"
+                         "SYNC_REPO=admin/chv-sync-test\n")
+            socket_path = os.path.join(tmp, "authd.sock")
+            log_path = os.path.join(tmp, "authd.log")
+            server = mod.build_server(
+                seed_path=seed_path, socket_path=socket_path,
+                log_path=log_path, lib_dir=str(REPO_ROOT),
+                allowed_uids=[os.getuid()], resolve=_public_resolver,
+                socket_group=os.getgid())
+            try:
+                self.assertEqual(server.socket_gid, os.getgid())
+                self.assertEqual(
+                    stat.S_IMODE(os.stat(socket_path).st_mode), 0o660)
+                self.assertEqual(
+                    stat.S_IMODE(os.stat(tmp).st_mode), 0o750)
+            finally:
+                server.server_close()
+
+
 class HelperE2ETest(unittest.TestCase):
     """helper + 真 git credential fill（假 authd socket 在后台）。"""
 
@@ -656,6 +779,19 @@ class HelperE2ETest(unittest.TestCase):
         self.assertEqual(proc.returncode, 1)
         self.assertIn("path_not_allowed", proc.stderr)
 
+    def test_helper_returns_credential_for_bare_git_path(self):
+        # round-11 真机修复（层 4）：裸 .git 路径一律当 pull，
+        # helper 须照常返回凭证（git 智能 HTTP 发现请求只发裸路径）。
+        proc = subprocess.run(
+            [sys.executable, str(self.HELPER)],
+            input="protocol=https\nhost=gitea.tose.top\n"
+                  "path=/admin/chv-sync-test.git\n\n",
+            capture_output=True, text=True, env=self._helper_env(),
+            check=False)
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn("username=admin", proc.stdout)
+        self.assertIn("password=" + TOKEN, proc.stdout)
+
     def test_git_credential_fill_end_to_end(self):
         env = self._helper_env()
         env["GIT_CONFIG_NOSYSTEM"] = "1"

cloud/tests/test_chv_muse_p4.py

@@ -509,6 +509,40 @@ class CdpBrokerTest(unittest.TestCase):
             self.assertTrue(broker.close()["ok"])
             self.assertFalse(broker.snapshot()["ok"])
 
+    def test_load_playwright_returns_called_cm(self):
+        # round-11 复测回归，由合入 round-12 时从 test_chv_muse_p2
+        # 移植到此处：_load_playwright 曾返回裸 sync_playwright
+        # 函数（未调用），_ensure_browser 对它调 __enter__ 即
+        # AttributeError，open 永远失败；单测的假工厂直接返回带
+        # __enter__ 的对象，盖不住这个形状。现 _load_playwright
+        # 必须返回调用后的上下文管理器。
+        import types
+        calls = []
+
+        class FakeCM:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *exc):
+                return False
+
+        cm = FakeCM()
+
+        def sync_playwright():
+            calls.append("called")
+            return cm
+
+        fake_api = types.ModuleType("playwright.sync_api")
+        fake_api.sync_playwright = sync_playwright
+        fake_pkg = types.ModuleType("playwright")
+        fake_pkg.sync_api = fake_api
+        sys.modules["playwright"] = fake_pkg
+        sys.modules["playwright.sync_api"] = fake_api
+        self.addCleanup(sys.modules.pop, "playwright.sync_api", None)
+        self.addCleanup(sys.modules.pop, "playwright", None)
+        self.assertIs(CDP._load_playwright(), cm)
+        self.assertEqual(calls, ["called"])
+
 
 class FakeCdpBroker:
     """假 CDP broker：按 canned 响应表回包，记录收到的请求。"""
@@ -718,14 +752,18 @@ class BuildDryRunTest(unittest.TestCase):
                          "chv-cdp-broker.py", "chv_browser_guard.py",
                          "browser-blocklist.txt",
                          "authd.py", "git-credential-chv-authd",
-                         "chv-authd.service", "chv-memory.py"):
+                         # 合入 round-11 时补：build.sh 的 P3 预检硬要求
+                         # authd_init_stub.py 在仓（缺一即 exit 1）。
+                         "chv-authd.service", "authd_init_stub.py",
+                         "chv-memory.py"):
                 shutil.copy2(ROOT / "chv-guest" / name,
                              os.path.join(guest, name))
             shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py"))
-            # round-12：合入后的 build.sh 要求 P3 客机三件套在仓里
+            # round-12：合入后的 build.sh 要求 P3 客机四件套在仓里
             # （硬失败），云端叶子模块只警告、不失败——夹具一并拷入，
-            # 与 P2 夹具同口径。
+            # 与 P2 夹具同口径（含 errors.py，使 P3 叶子段落跑到）。
             for rel in ("cloud/chv/authd.py", "cloud/chv/sentinel.py",
+                        "cloud/chv/errors.py",
                         "cloud/__init__.py", "cloud/chv/__init__.py"):
                 dest = os.path.join(repo, rel)
                 os.makedirs(os.path.dirname(dest), exist_ok=True)

cloud/tests/test_chv_muse_round11_cell.py

@@ -1,9 +1,10 @@
 """round-11 集成补洞：执行器接 nspawn 单元（P2 差距表 §4）。
 
 只测 _guest_host_command 的命令构造与 CHV_CELL_ENABLED 开关：
-单元开（默认）时 host.py 调用改为 chv-cell run -- python3
+单元开（默认）时 host.py 调用改为 sudo chv-cell run -- python3
 /app/host.py --state … --codex-home … --sandbox
-danger-full-access …；单元关时走原命令；非法开关值即
+danger-full-access …（sudo 不可省：chv-cell 须 root，执行器以
+chv 身份调；round-11 真机修复）；单元关时走原命令；非法开关值即
 ValueError。只用 mock，不启动真实 VM、不连真实网络。
 """
 
@@ -83,7 +84,7 @@ class CellCommandTest(CellCase):
         ex = self.make_exec()
         cmd = ex._guest_host_command("status", "--task", "t1")
         self.assertEqual(cmd, [
-            "chv-cell", "run", "--",
+            "sudo", "chv-cell", "run", "--",
             "python3", "/app/host.py",
             "--state", "/home/chv/.codex/host-state.json",
             "--codex-home", "/home/chv/.codex",
@@ -93,7 +94,7 @@ class CellCommandTest(CellCase):
     def test_cell_on_explicit_1(self):
         ex = self.make_exec(env_extra={"CHV_CELL_ENABLED": "1"})
         cmd = ex._guest_host_command("status", "--task", "t1")
-        self.assertEqual(cmd[:3], ["chv-cell", "run", "--"])
+        self.assertEqual(cmd[:4], ["sudo", "chv-cell", "run", "--"])
         self.assertIn("--sandbox", cmd)
         self.assertEqual(
             cmd[cmd.index("--sandbox") + 1], "danger-full-access")
@@ -120,7 +121,7 @@ class CellCommandTest(CellCase):
             model="m1", effort="low", approval="ask",
             model_provider="sub2api", sandbox="read-only")
         self.assertEqual(cmd, [
-            "chv-cell", "run", "--",
+            "sudo", "chv-cell", "run", "--",
             "python3", "/app/host.py",
             "--state", "/home/chv/.codex/host-state.json",
             "--codex-home", "/home/chv/.codex",
@@ -164,12 +165,14 @@ class CellCommandTest(CellCase):
 
     def test_cell_wiring_reaches_ssh_command(self):
         # 接线落到 SSH 层：build_ssh_command 的远端命令里
-        # chv-cell 前缀在、python3 在其后。
+        # sudo chv-cell 前缀在、python3 在其后。
         ex = self.make_exec()
         inner = ex._guest_host_command("status", "--task", "t1")
         cmd = ex.build_ssh_command({"id": "t1"}, inner)
+        self.assertIn("sudo", cmd)
         self.assertIn("chv-cell", cmd)
         self.assertIn("danger-full-access", cmd)
+        self.assertLess(cmd.index("sudo"), cmd.index("chv-cell"))
         self.assertLess(cmd.index("chv-cell"), cmd.index("python3"))
 
 

cloud/tests/test_chv_residual.py

@@ -418,7 +418,8 @@ class LockfileTest(unittest.TestCase):
                 "openssh-client 1:9.9", "cloud-init 9.9",
                 "python3 9.9", "python3-pip 9.9", "curl 9.9",
                 "ca-certificates 9.9", "xz-utils 9.9",
-                "fonts-wqy-microhei 9.9", "fontconfig 9.9"]
+                "fonts-wqy-microhei 9.9", "fontconfig 9.9",
+                "systemd-container 9.9"]
         info = {
             "tag": "chv-guest-24.04-0.164.0-alpha.9-abc1234",
             "ubuntu": "24.04",
@@ -432,6 +433,19 @@ class LockfileTest(unittest.TestCase):
                          "fonts-wqy-microhei", "fontconfig", "codex",
                          "codex-code-mode-host", "codex-resources/bwrap",
                          "/app/host.py", "gt", "chv-sync",
+                         # 与 717efc4 真实 BUILDINFO.contents 对齐：
+                         # P2 三件套 + P3 authd 三件套 + 叶子模块 +
+                         # vsock_notify（缺 DECLARED_SOURCES 登记时
+                         # generate 记 unresolved、verify/check 失败，
+                         # 合成清单须覆盖才拦得住回归）。round-11
+                         # 真机修复（层 1）后叶子模块还含 errors.py
+                         # 与 stub __init__.py。
+                         "systemd-container", "chv-cell", "chv-broker",
+                         "chv-browser", "chv-authd",
+                         "git-credential-chv-authd", "chv-authd.service",
+                         "cloud/chv/authd.py", "cloud/chv/sentinel.py",
+                         "cloud/chv/errors.py", "cloud/chv/__init__.py",
+                         "vsock_notify",
                          "playwright", "chromium-headless-shell"],
             "versions": {"dpkg": dpkg,
                          "pip": ["playwright==9.9.9"],