host: fix state-loss from tmp collision and cross-container flock failure

2026-10-29 task 2 lost its host-state entry (version 390 -> single
digits) while running under Docker/gVisor. Root cause, proven live:
- save_state tmp name contained only pid, and every containerized
  host.py runs as pid 1, so concurrent writers shared one tmp file
  and interleaved writes produced corrupt JSON;
- fcntl.flock does not exclude across runsc containers sharing a
  bind-mounted lock file (both entered at 0.0s in a live probe);
- _read_state_file silently treated corrupt JSON as empty state, so
  the next save legitimately rewrote the file with only the
  surviving writer's tasks.

Fix:
- tmp name now hostname + pid + thread + uuid4 (backup tmp same);
- mkdir atomic lock replaces flock (atomic on the host fs, holds
  across bind mounts/gofer), with stale-lock recovery after 120s
  and a loud timeout instead of unlocked writes; reads need no lock
  because publication is tmp + os.replace;
- save path uses a strict disk read: corrupt JSON, non-object top
  level, or wrong-shaped tasks/automations now raise instead of
  overwriting (recovery stays with restore_state + .bak).

Tests: tests/test_host_state_lock.py, 7 items (tmp uniqueness in
process and across processes, 8-process concurrent saves all
survive with version 8, stale lock recovery, fresh lock timeout
raises, corrupt/wrong-shaped disk refuses to overwrite).
Full suites: cloud 1168 (skipped 4), host 104, deploy 39, all green.

host.py

@@ -44,6 +44,7 @@ turn/completed 通知，所以多个 thread 的 turn 可以并发跑。
 CODEX_BIN=.../bin/codex（start.sh 已自动优先用 bin/codex）。
 """
 import argparse
+import contextlib
 import json
 import os
 import queue
@@ -852,31 +853,93 @@ class Server:
 # 覆盖、自动化按名合并，这样两个进程各改各的任务时，后保存
 # 的不会把先保存的覆盖掉。tmp 文件带 pid，避免两个进程用
 # 同一个 .tmp 互相踩。锁文件在 state 文件旁边，名如
-# host-state.json.lock，用 fcntl.flock（Linux/macOS 都有）。
-import contextlib
-try:
-    import fcntl
-except ImportError:  # Windows 等无 fcntl 的平台：退化为无锁
-    fcntl = None
+# 跨进程互斥绝不能再靠 fcntl.flock：2026-10-29 实测，runsc
+# （gVisor）容器里 flock 只在各自 Sentry 内模拟，两个共享
+# bind 挂载锁文件的容器能同时拿到排他锁；而 tmp 文件名又只
+# 含 pid（容器内 host.py 全是 pid 1），并发保存写同一个 tmp，
+# 交错出坏 JSON，读到坏 JSON 的下一次保存静默当成空 state
+# 重写，全部任务条目一起丢（任务 2 实弹，版本 390 跌回个位数）。
+# 改用 mkdir 原子锁：mkdir 在宿主文件系统上原子成功才算拿到，
+# 经 bind 挂载/gofer 也一样成立；持有者被杀留下的陈旧锁按
+# mtime 超龄清理；超时未拿到则大声失败，绝不无锁裸写。
+_STATE_LOCK_TIMEOUT = 30.0
+_STATE_LOCK_STALE = 120.0
+
+
+def _state_tmp_path(path):
+    # tmp 文件名必须每个写者唯一：旧写法只含 pid，容器内 host.py
+    # 全是 pid 1，并发保存写同一个 tmp 是断点丢失的直接原因。
+    # hostname + pid + 线程 + uuid 四重区分；备份 tmp 同款。
+    try:
+        import socket
+        host_name = socket.gethostname()
+    except Exception:
+        host_name = "unknown"
+    return (f"{path}.tmp.{host_name}.{os.getpid()}."
+            f"{threading.get_ident()}.{uuid.uuid4().hex}")
 
 
 @contextlib.contextmanager
 def _state_lock(path, exclusive=True):
-    lock_path = path + ".lock"
-    os.makedirs(os.path.dirname(os.path.abspath(lock_path)), exist_ok=True)
-    f = open(lock_path, "a+")
+    if not exclusive:
+        # 读不加锁：保存是 tmp + os.replace 原子发布，读到的
+        # 永远是完整旧版或完整新版，不会读到半截文件。
+        yield
+        return
+    lock_dir = path + ".lockdir"
+    os.makedirs(os.path.dirname(os.path.abspath(lock_dir)), exist_ok=True)
+    deadline = time.time() + _STATE_LOCK_TIMEOUT
+    while True:
+        try:
+            os.mkdir(lock_dir)
+            break
+        except FileExistsError:
+            try:
+                age = time.time() - os.path.getmtime(lock_dir)
+            except OSError:
+                age = 0  # 刚被别人删掉，重试
+            if age > _STATE_LOCK_STALE:
+                try:
+                    os.rmdir(lock_dir)
+                except OSError:
+                    pass  # 别人先清掉了，重试
+                continue
+            if time.time() >= deadline:
+                raise RuntimeError(
+                    f"state 锁超时未拿到（{lock_dir}），拒绝无锁写入，"
+                    "防止并发保存互相覆盖")
+            time.sleep(0.05)
     try:
-        if fcntl is not None:
-            fcntl.flock(f.fileno(),
-                        fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
         yield
     finally:
-        if fcntl is not None:
-            try:
-                fcntl.flock(f.fileno(), fcntl.LOCK_UN)
-            except Exception:
-                pass
-        f.close()
+        try:
+            os.rmdir(lock_dir)
+        except OSError:
+            pass
+
+
+def _read_state_file_for_save(path):
+    # 保存路径的严格读：文件存在但 JSON 损坏、顶层不是对象、
+    # tasks/automations 类型不对时一律抛错，绝不当成空 state
+    # 去覆盖——旧的静默读空是断点丢失的最后一环：坏文件一旦
+    # 被当成空 state 保存，全部任务条目就被合法地重写没了。
+    # 文件不存在仍当空 state（首次保存）；.bak 恢复走
+    # restore_state，不走这里。
+    if not os.path.exists(path):
+        return {"tasks": {}, "automations": []}
+    try:
+        with open(path, encoding="utf-8") as f:
+            data = json.load(f)
+    except (ValueError, UnicodeDecodeError) as e:
+        raise RuntimeError(
+            f"state 文件 JSON 已损坏，拒绝覆盖（先用 .bak 恢复）：{path}：{e}")
+    if not isinstance(data, dict):
+        raise RuntimeError(f"state 文件顶层不是对象，拒绝覆盖：{path}")
+    if not isinstance(data.get("tasks", {}), dict):
+        raise RuntimeError(f"state 文件 tasks 不是对象，拒绝覆盖：{path}")
+    if not isinstance(data.get("automations", []), list):
+        raise RuntimeError(f"state 文件 automations 不是列表，拒绝覆盖：{path}")
+    return data
 
 
 def _read_state_file(path):
@@ -1069,7 +1132,10 @@ def save_state(path, state):
     with _state_lock(path, exclusive=True):
         # 锁内重读 + 合并：内存任务覆盖同名磁盘任务，磁盘上
         # 别的进程新增的任务保留；自动化按名合并，内存优先。
-        disk = _read_state_file(path)
+        # 注意：这里必须用保存路径的严格读——磁盘文件已损坏时
+        # 抛错而不是当成空 state，否则就是 2026-10-29 断点丢失
+        # 的最后一环（坏文件被合法重写，全部条目消失）。
+        disk = _read_state_file_for_save(path)
         loaded_version = state.get("_loaded_version", disk.get("version", 0))
         if disk.get("version", 0) > loaded_version:
             # 别的写者（另一进程 / 另一台机器共享文件时）已先写过，
@@ -1106,7 +1172,7 @@ def save_state(path, state):
         except Exception:
             merged["last_writer"] = str(os.getpid())
         merged["updated_at"] = time.time()
-        tmp = f"{path}.tmp.{os.getpid()}"
+        tmp = _state_tmp_path(path)
         with open(tmp, "w") as f:
             json.dump(merged, f, indent=2, ensure_ascii=False)
         if os.path.exists(path):
@@ -1114,7 +1180,7 @@ def save_state(path, state):
             backup = path + ".bak"
             backup_1 = backup + ".1"
             backup_2 = backup + ".2"
-            backup_tmp = f"{backup}.tmp.{os.getpid()}.{threading.get_ident()}"
+            backup_tmp = _state_tmp_path(backup)
             shutil.copy2(path, backup_tmp)
             if os.path.exists(backup_2):
                 os.remove(backup_2)

tests/test_host_state_lock.py

@@ -0,0 +1,123 @@
+"""host-state 并发保存回归测试（2026-10-29 断点丢失实弹）。
+
+旧写法两处同时失效：tmp 文件名只含 pid（容器内 host.py 全是
+pid 1，并发保存写同一个 tmp 交错出坏 JSON）；flock 在 runsc
+容器间不互斥（实测双双 0 秒拿到排他锁）。坏 JSON 又被
+_read_state_file 静默当成空 state，下一次保存合法地重写，
+全部任务条目一起丢、版本从 390 跌回个位数。
+
+本文件锁死三条修法：tmp 名每个写者唯一、mkdir 原子锁跨进程
+互斥、磁盘已损坏时保存直接拒绝覆盖。
+"""
+
+import glob
+import json
+import multiprocessing
+import os
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT))
+
+import host
+
+
+def _save_one(args):
+    path, name = args
+    state = {"tasks": {name: {"status": "running"}}, "automations": []}
+    host.save_state(path, state)
+    return name
+
+
+def _tmp_name(_ignored):
+    return host._state_tmp_path("/tmp/x/host-state.json")
+
+
+class StateTmpNameTest(unittest.TestCase):
+    def test_tmp_names_unique_in_process(self):
+        names = {host._state_tmp_path("/tmp/x/host-state.json")
+                 for _ in range(100)}
+        self.assertEqual(len(names), 100)
+
+    def test_tmp_names_unique_across_processes(self):
+        ctx = multiprocessing.get_context("fork")
+        with ctx.Pool(8) as pool:
+            names = pool.map(_tmp_name, range(8))
+        self.assertEqual(len(set(names)), 8)
+
+
+class StateMkdirLockTest(unittest.TestCase):
+    def test_concurrent_saves_all_tasks_survive(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            ctx = multiprocessing.get_context("fork")
+            names = [f"task-{i}" for i in range(8)]
+            with ctx.Pool(8) as pool:
+                pool.map(_save_one, [(path, n) for n in names])
+            with open(path, encoding="utf-8") as f:
+                data = json.load(f)
+            self.assertEqual(set(data["tasks"]), set(names))
+            self.assertEqual(data["version"], 8)
+            self.assertFalse(glob.glob(path + ".tmp.*"))
+            self.assertFalse(os.path.exists(path + ".lockdir"))
+
+    def test_stale_lock_is_recovered(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            lock_dir = path + ".lockdir"
+            os.mkdir(lock_dir)
+            old = time.time() - 1000
+            os.utime(lock_dir, (old, old))
+            host.save_state(path, {"tasks": {"a": {}}, "automations": []})
+            self.assertFalse(os.path.exists(lock_dir))
+            with open(path, encoding="utf-8") as f:
+                self.assertIn("a", json.load(f)["tasks"])
+
+    def test_fresh_lock_timeout_raises_instead_of_writing_unlocked(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            os.mkdir(path + ".lockdir")
+            old_timeout = host._STATE_LOCK_TIMEOUT
+            host._STATE_LOCK_TIMEOUT = 0.3
+            try:
+                with self.assertRaises(RuntimeError):
+                    host.save_state(path, {"tasks": {"a": {}},
+                                           "automations": []})
+            finally:
+                host._STATE_LOCK_TIMEOUT = old_timeout
+            self.assertFalse(os.path.exists(path))
+
+
+class StateCorruptRefusalTest(unittest.TestCase):
+    def test_corrupt_disk_refuses_to_overwrite(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            garbage = b"{not json, torn write"
+            with open(path, "wb") as f:
+                f.write(garbage)
+            with self.assertRaises(RuntimeError):
+                host.save_state(path, {"tasks": {"a": {}},
+                                       "automations": []})
+            with open(path, "rb") as f:
+                self.assertEqual(f.read(), garbage)
+            self.assertFalse(os.path.exists(path + ".bak"))
+            self.assertFalse(os.path.exists(path + ".lockdir"))
+
+    def test_wrong_shaped_disk_refuses_to_overwrite(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            with open(path, "w", encoding="utf-8") as f:
+                json.dump({"tasks": ["not", "an", "object"],
+                           "automations": []}, f)
+            with self.assertRaises(RuntimeError):
+                host.save_state(path, {"tasks": {"a": {}},
+                                       "automations": []})
+
+
+if __name__ == "__main__":
+    unittest.main()