part 4 缺陷 1：cell 启动竞态——起步门 + init pid 轮询

RETEST3 缺陷 1：run_cell 里配网与 payload 零同步，快 payload
（run -- true）在配网完成前退出即撞墙，11 次只成 6 次。
本提交先落 helper（接线在缺陷 3 的 run_cell 重写里做）：
- gate_wrapper_command()：payload 包一层 sh 起步门，等
  /run/chv/.cell-gate-<machine> 出现再 exec（绝对路径
  /usr/bin/sh：volatile 单元根里 /bin 未必有）。
- find_cell_init_pid() 改带超时轮询（timeout 默认 0.0 = 单次，
  旧调用行为不变；缺目录不再直接回 None；proc_alive 回调让
  nspawn 已死时立刻回，不空等）。
新测试：CellGateTest、CellInitPidPollTest（5 个）。

chv-guest/chv-cell.py

@@ -235,10 +235,18 @@ def host_network_cleanup_commands(uplink_iface):
     return cmds
 
 
-def find_cell_init_pid(nspawn_pid, proc_root="/proc"):
+def find_cell_init_pid(nspawn_pid, proc_root="/proc", timeout=0.0,
+                       interval=0.1, proc_alive=None):
     """找单元 init 的 pid：nspawn 进程的第一个子进程
    （--as-pid2 时它是 stub init，payload 是 PID 2；两者同 netns）。
+    timeout=0（默认）即只查一次（旧行为）；timeout>0 时带超时轮询
+    （part 4 竞态修复：init 可能还没 fork，单次查找即误判）。
+    proc_alive 供调用方传“nspawn 是否还活着”（如 proc.poll() is
+    None 的 lambda）：nspawn 已退出即立刻回 None，不再空等超时。
     找不到返回 None。"""
+    deadline = time.time() + timeout
+    while True:
+        found = None
         try:
             task_dir = os.path.join(proc_root, str(nspawn_pid), "task")
             for tid in os.listdir(task_dir):
@@ -249,10 +257,23 @@ def find_cell_init_pid(nspawn_pid, proc_root="/proc"):
                 except OSError:
                     continue
                 if children:
-                return int(children[0])
-    except (OSError, ValueError):
+                    try:
+                        found = int(children[0])
+                    except ValueError:
+                        continue
+                    break
+        except OSError:
+            # 任务目录还没出现（init 未 fork）即当作“这次没找到”，
+            # 继续轮询，不直接回 None（旧代码这里直接 return None，
+            # 正是单次查找误判的根因之一）。
+            found = None
+        if found is not None:
+            return found
+        if proc_alive is not None and not proc_alive():
             return None
+        if time.time() >= deadline:
             return None
+        time.sleep(min(interval, max(0.0, deadline - time.time())))
 
 
 def cell_network_commands(cell_pid):
@@ -282,6 +303,20 @@ def _ensure_bind_source(path, uid=CELL_UID, gid=CELL_GID):
         raise
 
 
+def gate_wrapper_command(command, gate_path):
+    """把单元内命令包一层起步门（纯函数）：payload 先在单元内
+    等 gate 文件出现再 exec 原命令。gate 文件由 run_cell 在配网
+    完成后才创建（见 run_cell）——payload 与配网零同步是 RETEST3
+    新缺陷 1 的根因（快 payload 在配网完成前退出，nspawn 随之
+    退出，包装器再去配网即撞墙）。/usr/bin/sh 用绝对路径：
+    volatile 单元根里 /bin 未必有，/usr 是只读挂进来的，必有。
+    注意：$1 是 gate 路径，shift 后 "$@" 即原命令。"""
+    script = ("while [ ! -e \"$1\" ]; do sleep 0.05; done; "
+              "shift; exec \"$@\"")
+    return ["/usr/bin/sh", "-c", script, "chv-cell-gate",
+            gate_path] + list(command)
+
+
 def _wait_for_iface(iface, timeout=10.0):
     deadline = time.time() + timeout
     while time.time() < deadline:

cloud/tests/test_chv_muse_p2.py

@@ -742,5 +742,75 @@ class BuildDryRunTest(unittest.TestCase):
                 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)
+
+
 if __name__ == "__main__":
     unittest.main()