host: phase2 daemon + health + auto-recovery (restore, recover, double-fork)

README.md

@@ -168,6 +168,7 @@ turn 可以真正并发（不是轮流排队）。
 | 失败通知 | 失败只有 state 里的 status，跑完没人知道 | 已修：失败时写 host-run.log + 跑配置的 notify_command（本地 shell，模板 {task} {status}）；failures 命令列失败任务，doctor 列失败/缺失 worktree/open 评论 |
 | 多机器 state 冲突 | 锁只防同一台机器，多机器共享 state 文件会静默覆盖 | 部分修：state 加 version/last_writer/updated_at，保存时磁盘版本比加载时新就记 STATE-CONFLICT 日志并合并保留双方任务，不静默丢；真多机同步仍需中心服务（见下“没对齐”） |
 | 合并时提交缓存文件 | merge 的 git add -A 会把 __pycache__/*.pyc 也提交进主仓，下次合并被 .pyc 改动挡住 | 已修：merge 前清 worktree 的 __pycache__/*.pyc，主仓的 .pyc 改动先还原；单测覆盖 pyc 不入库 |
+| 守护进程 + 健康检查 + 自动恢复 | host.py 跑完就退出，不常驻；state 坏了静默当空 state；中断任务一直停在 running | 已修：daemon（双 fork 到 PID 1，循环 recover + run-due）、health（不跑模型，查二进制/配置/state/server/daemon）、restore-state（坏了从 .bak 恢复并留底）、recover（超过阈值的 running 标 interrupted，可用 run 续跑，不自动重跑模型回合） |
 | 边界：cwd、沙箱、审批、模型 | dispatch / turn 参数里定，见 `config.example.toml` | 已实测 |
 | 任务登记 | `host-state.json`：task -> thread/worktree/状态/diff；agent 状态本身仍在 CODEX_HOME，不另存第二份 | 已实测 |
 
@@ -190,6 +191,11 @@ turn 可以真正并发（不是轮流排队）。
   任务，但两台机器共享一个 state 文件仍不是 ChatGPT 云端
   那种中心同步——要真多机，得另起一个中心 state 服务，
   这不在本地 host 的范围里。
+- 守护进程的边界：daemon 本身也活不过 VM 平台回收（和
+  微信桥 serve 一样，回收后要重拉），所以 VM 上更稳的定时
+  仍是 Muse cron 定时调 run-due/health，而不是只靠 daemon
+  或 crontab；daemon 适合常驻机器，VM 上是“会话内常驻 +
+  回收后重拉”的定位。
 
 ## 开源没给到的（别指望照搬）
 

TEST-REPORT.md

@@ -394,3 +394,47 @@ tests/ 标准库 unittest，验收命令就是 `python3 -m unittest discover -s
   包括 topo 排序、merge 验测通过/失败回滚/重复加自动解/
   pyc 不入库、评审评论、重试成功/预算拦截/通知落盘、
   state 版本冲突、server kill 重拉。
+
+## 第二阶段：守护进程 + 健康检查 + 自动恢复（2026-10-26 凌晨，用户选“先做第二阶段”）
+
+- **先堵的洞**：旧 `_read_state_file` 会把 JSON 解析错误
+  吞掉、当成空 state 返回——state 损坏一次就等于全部任务
+  静默丢失。新加 `load_state_strict`（文件存在但解析失败
+  时返回错误，不当空 state），health/recover 都用严格版。
+- **health（不跑模型）**：查 codex 二进制存在、配置可读、
+  state 可解析、备份存在、daemon pid 存活/过期、server
+  能 initialize + list、缺失 worktree。实测：完好时 ok=true、
+  server_threads=7、exit 0；故意写坏 state 后 ok=false、
+  state_valid=false、带解析错误信息、exit 1。
+- **restore-state**：state 损坏或丢失时按 .bak、.bak.1、
+  .bak.2 顺序找第一个能解析的备份，先把坏文件改名成
+  .corrupt.<时间戳> 留底，再把备份拷回。实测：坏 state +
+  version=9 的 .bak，恢复后 version=9，health 回到 ok=true。
+- **recover（只标记，不自动重跑模型）**：status=running
+  且最后回合超过 --older-than（默认 3600 秒）未更新的
+  任务标 interrupted 并记日志，用户随后可用 run 续跑
+  （thread 在 CODEX_HOME 里不丢）。不自动重跑是刻意的
+  安全边界：自动重跑会不经用户同意花中转站额度。
+  实测：2 小时前的 stuck -> interrupted，1 分钟前的
+  fresh 保持 running；单测覆盖只标旧任务。
+- **daemon（双 fork）**：daemonize 用 fork -> setsid ->
+  再 fork，孙进程 reparent 到 PID 1、写 pidfile、stdout/
+  stderr 重定向到 host-daemon.log，循环每 interval（默认
+  300 秒）跑 recover + run-due，每轮都是独立子进程调
+  host.py CLI，一轮崩了不带走 daemon 本体。实测（interval
+  5 秒）：daemon pid 10776、PPID=1（活过 launcher 会话），
+  daemon-status alive=true，日志每 5 秒一个 tick；
+  daemon-stop 后 pid 消失、status alive=false、pidfile
+  删除、日志有 host daemon stopped。
+- **VM 边界（诚实说）**：daemon 本身活不过平台回收，
+  和微信桥 serve 同理——回收后要重拉，Muse cron 定时调
+  run-due/health 才是 VM 上更稳的定时。本轮没有自动装
+  Muse cron 或 crontab，因为 run-due 到期会跑模型花额度，
+  装定时这一步留给用户显式下令（见下一条待办）。
+- **回归**：tests 从 15 个涨到 19 个，全部通过，新增
+  corrupt 检测 + 从 .bak 恢复、无备份时不覆盖、recover
+  只标旧任务、health 用假 server 全通过。
+- **待办（不在本轮自动做）**：Muse cron 定时调
+  `host.py --config <config> health`（无模型成本）和 /
+  或 `run-due`（到期才跑模型），需用户定间隔和是否允许
+  自动跑模型后才装。

host.py

@@ -381,6 +381,172 @@ def _read_state_file(path):
     return {"tasks": {}, "automations": []}
 
 
+def load_state_strict(path):
+    """严格读 state：文件不存在返回 (空 state, None)；
+    文件存在但 JSON 坏了返回 (None, 错误信息)，绝不静默
+    当成空 state——否则损坏一次就等于全部任务丢了，
+    这是自动恢复要先堵住的洞（旧 _read_state_file 会吞掉
+    解析错误，health/recover 必须用这个严格版）。"""
+    if not os.path.exists(path):
+        return {"tasks": {}, "automations": []}, None
+    try:
+        with open(path) as f:
+            return json.load(f), None
+    except Exception as e:
+        return None, str(e)
+
+
+def restore_state(state_path):
+    """state 损坏或丢失时从备份恢复：按 .bak、.bak.1、.bak.2
+    顺序找第一个能解析的备份，先把坏文件（如果存在）改名
+    成 .corrupt.<时间戳> 留底，再把备份拷回主文件。
+    返回 (ok, message)。主文件完好时不做任何事。"""
+    strict, err = load_state_strict(state_path)
+    if err is None and strict is not None:
+        return True, "state 完好，无需恢复"
+    if os.path.exists(state_path):
+        corrupt = f"{state_path}.corrupt.{int(time.time())}"
+        try:
+            os.replace(state_path, corrupt)
+        except Exception as e:
+            return False, f"坏 state 改名留底失败：{e}"
+    else:
+        corrupt = None
+    for suffix in (".bak", ".bak.1", ".bak.2"):
+        cand = state_path + suffix
+        if not os.path.exists(cand):
+            continue
+        try:
+            with open(cand) as f:
+                data = json.load(f)
+        except Exception:
+            continue
+        import shutil
+        shutil.copy2(cand, state_path)
+        return True, (f"已从 {suffix} 恢复，version="
+                      f"{data.get('version')}，"
+                      f"坏文件留底：{corrupt}")
+    return False, "没有可用的备份（.bak/.bak.1/.bak.2 均缺失或损坏）"
+
+
+def recover_interrupted(state, state_path, older_than=3600, cfg=None):
+    """自动恢复（只标记，不自动重跑模型回合）：
+    status=running 且最后回合时间超过 older_than 秒的任务，
+    判定为进程被杀/VM 回收导致的中断，标 interrupted 并记日志，
+    用户随后可用 run 续跑（thread 在 CODEX_HOME 里不丢）。
+    不自动重跑是刻意的安全边界：自动重跑会不经用户同意
+    花中转站额度，daemon 只负责标记 + 日志 + 通知。
+    返回被标记的任务名列表。"""
+    now = time.time()
+    recovered = []
+    for name, t in (state.get("tasks") or {}).items():
+        if t.get("status") != "running":
+            continue
+        last = t.get("last_turn_at") or t.get("created_at") or 0
+        if now - last < older_than:
+            continue
+        t["status"] = "interrupted"
+        recovered.append(name)
+        try:
+            log_event(state_path, f"recover {name} -> interrupted "
+                                  f"(last_turn {int(now - last)}s ago)")
+        except Exception:
+            pass
+        if cfg is not None:
+            notify_failure(state_path, cfg, name,
+                           {"status": "interrupted",
+                            "text": "process killed / VM reclaimed"})
+    return recovered
+
+
+def run_health_check(state_path, cfg, codex_bin, codex_home, approval,
+                     server_factory=None):
+    """健康检查（不跑模型）：返回 report dict，ok 为全部通过。
+    检查项：二进制存在、配置可读、state 可解析、备份存在、
+    daemon pid 是否存活/过期、server 能 initialize + list、
+    缺失 worktree（doctor 的逻辑）。server_factory 供单测注入
+    假 server，避免健康检查本身要拉真进程。"""
+    checks = {}
+    checks["codex_bin_exists"] = bool(codex_bin) and os.path.exists(codex_bin)
+    checks["config_loads"] = isinstance(cfg, dict)
+    strict, err = load_state_strict(state_path)
+    checks["state_valid"] = err is None
+    checks["state_error"] = err
+    checks["backup_exists"] = os.path.exists(state_path + ".bak")
+    if strict is not None:
+        doc = check_state(state_path, strict)
+        checks["failed_tasks"] = doc["failed_tasks"]
+        checks["missing_worktrees"] = doc["missing_worktrees"]
+        checks["open_review_comments"] = doc["open_review_comments"]
+    else:
+        checks["failed_tasks"] = []
+        checks["missing_worktrees"] = []
+        checks["open_review_comments"] = 0
+    # daemon pid
+    pidfile = os.path.join(os.path.dirname(os.path.abspath(state_path)),
+                           "host-daemon.pid")
+    checks["daemon_pidfile"] = pidfile
+    if os.path.exists(pidfile):
+        try:
+            pid = int(open(pidfile).read().strip())
+            os.kill(pid, 0)
+            checks["daemon_alive"] = True
+            checks["daemon_pid"] = pid
+        except Exception:
+            checks["daemon_alive"] = False
+            checks["daemon_pid"] = None
+    else:
+        checks["daemon_alive"] = False
+        checks["daemon_pid"] = None
+    # server（无模型成本，只 initialize + list）
+    try:
+        if server_factory is not None:
+            srv = server_factory()
+        else:
+            srv = Server(codex_bin=codex_bin, codex_home=codex_home,
+                         approval=approval)
+        try:
+            srv.initialize()
+            threads = srv.list_threads()
+            checks["server_ok"] = True
+            checks["server_threads"] = len(
+                (threads or {}).get("data") or []) if isinstance(threads, dict) else None
+        finally:
+            try:
+                srv.close()
+            except Exception:
+                pass
+    except Exception as e:
+        checks["server_ok"] = False
+        checks["server_error"] = str(e)[:500]
+    ok = bool(checks["codex_bin_exists"] and checks["state_valid"]
+              and checks["server_ok"])
+    return {"ok": ok, "checks": checks}
+
+
+def daemonize(pidfile, logfile):
+    """双 fork 守护进程（WeChat 桥 daemon_serve.py 同款教训：
+    setsid nohup 会在 launcher 会话结束时被 runtime 按进程树
+    回收杀掉，双 fork 到 PID 1 才能活过会话结束）。
+    父进程返回 False（应退出），子进程返回 True（继续跑循环）。"""
+    if os.fork() > 0:
+        return False
+    os.setsid()
+    if os.fork() > 0:
+        os._exit(0)
+    sys.stdout.flush()
+    sys.stderr.flush()
+    log_f = open(logfile, "a", encoding="utf-8")
+    os.dup2(log_f.fileno(), sys.stdout.fileno())
+    os.dup2(log_f.fileno(), sys.stderr.fileno())
+    with open("/dev/null", "r") as devnull:
+        os.dup2(devnull.fileno(), sys.stdin.fileno())
+    with open(pidfile, "w") as f:
+        f.write(str(os.getpid()))
+    return True
+
+
+
 def load_state(path):
     with _state_lock(path, exclusive=False):
         st = _read_state_file(path)
@@ -843,10 +1009,14 @@ def run_task_turn(server, state, task, prompt, model, effort,
     # 否则用调用方显式传的 sandbox；两者都没有时不传 sandboxPolicy，
     # 保留 thread 原有设置（旧任务没有 sandbox 记录时尤其重要）。
     use_sandbox = task.get("sandbox") or sandbox
+    started = time.time()
     result = server.run_turn(task["thread_id"], prompt, model=model,
                              effort=effort, sandbox=use_sandbox)
+    finished = time.time()
     record_turn_usage(state, failed=(result["status"] != "completed"))
-    task["turns"].append({"prompt": prompt, "result": result})
+    task["turns"].append({"prompt": prompt, "result": result,
+                          "started_at": started, "finished_at": finished})
+    task["last_turn_at"] = finished
     task["status"] = "completed" if result["status"] == "completed" \
         else "failed"
     task["last_text"] = result.get("text", "")
@@ -939,6 +1109,18 @@ def main():
     p = sub.add_parser("fix-review"); p.add_argument("--task", required=True)
     sub.add_parser("doctor")
     sub.add_parser("failures")
+    sub.add_parser("health")
+    sub.add_parser("restore-state")
+    p = sub.add_parser("recover")
+    p.add_argument("--older-than", type=int, default=3600,
+                   help="最后回合超过多少秒未更新才判定中断，默认 3600")
+    p = sub.add_parser("daemon")
+    p.add_argument("--interval", type=int, default=300,
+                   help="每轮间隔秒数，默认 300")
+    p.add_argument("--foreground", action="store_true",
+                   help="前台跑循环（测试用，不双 fork）")
+    sub.add_parser("daemon-status")
+    sub.add_parser("daemon-stop")
     p = sub.add_parser("discard"); p.add_argument("--task", required=True)
     p = sub.add_parser("archive"); p.add_argument("--task", required=True)
     p = sub.add_parser("fork"); p.add_argument("--task", required=True); p.add_argument("--name", required=True); p.add_argument("--sandbox", default=None)
@@ -1053,6 +1235,131 @@ def main():
         save_state(args.state, state)
         print(json.dumps(c, indent=2, ensure_ascii=False))
         return
+    if args.cmd == "health":
+        report = run_health_check(args.state, cfg, args.codex_bin or default_codex_bin(),
+                                  args.codex_home, args.approval)
+        print(json.dumps(report, indent=2, ensure_ascii=False))
+        if not report["ok"]:
+            raise SystemExit(1)
+        return
+    if args.cmd == "restore-state":
+        ok, msg = restore_state(args.state)
+        print(msg)
+        if not ok:
+            raise SystemExit(1)
+        return
+    if args.cmd == "recover":
+        strict, err = load_state_strict(args.state)
+        if err is not None:
+            print(f"state 已损坏，recover 不会覆盖它，先跑 restore-state：{err}")
+            raise SystemExit(1)
+        recovered = recover_interrupted(state, args.state,
+                                        older_than=args.older_than, cfg=cfg)
+        if recovered:
+            save_state(args.state, state)
+        print(json.dumps({"recovered": recovered}, indent=2, ensure_ascii=False))
+        return
+    if args.cmd == "daemon-status":
+        pidfile = os.path.join(os.path.dirname(os.path.abspath(args.state)),
+                               "host-daemon.pid")
+        logfile = os.path.join(os.path.dirname(os.path.abspath(args.state)),
+                               "host-daemon.log")
+        alive = False
+        pid = None
+        if os.path.exists(pidfile):
+            try:
+                pid = int(open(pidfile).read().strip())
+                os.kill(pid, 0)
+                alive = True
+            except Exception:
+                alive = False
+        tail = ""
+        if os.path.exists(logfile):
+            try:
+                tail = "\n".join(open(logfile).read().splitlines()[-5:])
+            except Exception:
+                pass
+        print(json.dumps({"alive": alive, "pid": pid, "pidfile": pidfile,
+                          "log_tail": tail}, indent=2, ensure_ascii=False))
+        return
+    if args.cmd == "daemon-stop":
+        pidfile = os.path.join(os.path.dirname(os.path.abspath(args.state)),
+                               "host-daemon.pid")
+        if not os.path.exists(pidfile):
+            print("daemon 未运行（无 pidfile）")
+            return
+        try:
+            pid = int(open(pidfile).read().strip())
+        except Exception:
+            print("pidfile 损坏，直接删除")
+            os.remove(pidfile)
+            return
+        try:
+            os.kill(pid, 15)
+            for _ in range(10):
+                time.sleep(0.5)
+                try:
+                    os.kill(pid, 0)
+                except Exception:
+                    break
+            else:
+                os.kill(pid, 9)
+        except Exception:
+            pass
+        try:
+            os.remove(pidfile)
+        except Exception:
+            pass
+        print(f"daemon 已停：pid {pid}")
+        return
+    if args.cmd == "daemon":
+        state_dir = os.path.dirname(os.path.abspath(args.state))
+        pidfile = os.path.join(state_dir, "host-daemon.pid")
+        logfile = os.path.join(state_dir, "host-daemon.log")
+        if not args.foreground:
+            if os.path.exists(pidfile):
+                try:
+                    old_pid = int(open(pidfile).read().strip())
+                    os.kill(old_pid, 0)
+                    print(f"daemon 已在运行：pid {old_pid}，先 daemon-stop 再起")
+                    return
+                except Exception:
+                    pass
+            if not daemonize(pidfile, logfile):
+                print(f"daemon 已在后台启动：pidfile {pidfile}，log {logfile}")
+                return
+        # 子进程 / 前台：循环跑 recover + run-due，每轮都是独立
+        # 子进程调 host.py CLI，这样一轮崩了不带走 daemon 本体，
+        # 且 run-due 的模型回合数受自动化本身和预算控制。
+        import signal
+        stop = {"flag": False}
+
+        def _on_term(signum, frame):
+            stop["flag"] = True
+        signal.signal(signal.SIGTERM, _on_term)
+        signal.signal(signal.SIGINT, _on_term)
+        print(f"host daemon started pid={os.getpid()} interval={args.interval}s "
+              f"config={config_path} state={args.state}", flush=True)
+        while not stop["flag"]:
+            ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+            print(f"{ts} daemon tick: recover + run-due", flush=True)
+            for subcmd in (["recover", "--older-than", "3600"], ["run-due"]):
+                try:
+                    subprocess.run([sys.executable, os.path.join(HERE, "host.py"),
+                                    "--config", config_path] + subcmd,
+                                   timeout=3600)
+                except Exception as e:
+                    print(f"{ts} daemon subcommand {subcmd} 失败：{e}", flush=True)
+            for _ in range(args.interval):
+                if stop["flag"]:
+                    break
+                time.sleep(1)
+        try:
+            os.remove(pidfile)
+        except Exception:
+            pass
+        print("host daemon stopped", flush=True)
+        return
     srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home,
                  approval=args.approval)
     try:

tests/test_host.py

@@ -297,6 +297,82 @@ class ServerRestartTest(unittest.TestCase):
                     pass
 
 
+class HealthRestoreRecoverTest(unittest.TestCase):
+    class FakeServer:
+        def initialize(self):
+            return {}
+
+        def list_threads(self):
+            return {"data": []}
+
+        def close(self):
+            pass
+
+    def test_corrupt_state_detected_and_restored_from_bak(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            good = {"tasks": {"a": {"status": "completed"}},
+                    "automations": [], "version": 3}
+            with open(path + ".bak", "w") as f:
+                json.dump(good, f)
+            with open(path, "w") as f:
+                f.write("{not json!!!")
+            strict, err = host.load_state_strict(path)
+            self.assertIsNone(strict)
+            self.assertIsNotNone(err)
+            ok, msg = host.restore_state(path)
+            self.assertTrue(ok, msg)
+            with open(path) as f:
+                restored = json.load(f)
+            self.assertEqual(restored["version"], 3)
+            self.assertTrue(os.path.exists(path + ".corrupt.") or
+                            any(fn.startswith("host-state.json.corrupt.")
+                                for fn in os.listdir(tmp)))
+
+    def test_restore_without_backup_fails_without_overwriting(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            with open(path, "w") as f:
+                f.write("{bad")
+            ok, msg = host.restore_state(path)
+            self.assertFalse(ok)
+            # 坏文件被改名留底，主文件已不在（不是被空 state 覆盖）
+            self.assertFalse(os.path.exists(path))
+
+    def test_recover_marks_old_running_only(self):
+        import time as _time
+        now = _time.time()
+        state = {"tasks": {
+            "old": {"status": "running", "created_at": now - 7200,
+                    "last_turn_at": now - 7200},
+            "fresh": {"status": "running", "created_at": now - 60,
+                      "last_turn_at": now - 60},
+            "done": {"status": "completed", "created_at": now - 7200},
+        }, "automations": []}
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            recovered = host.recover_interrupted(state, path,
+                                                 older_than=3600, cfg={})
+            self.assertEqual(recovered, ["old"])
+            self.assertEqual(state["tasks"]["old"]["status"], "interrupted")
+            self.assertEqual(state["tasks"]["fresh"]["status"], "running")
+
+    def test_health_with_fake_server(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "host-state.json")
+            with open(path, "w") as f:
+                json.dump({"tasks": {}, "automations": [], "version": 1}, f)
+            with open(path + ".bak", "w") as f:
+                json.dump({"tasks": {}, "automations": [], "version": 1}, f)
+            report = host.run_health_check(
+                path, {}, "/bin/true", tmp, "deny",
+                server_factory=lambda: self.FakeServer())
+            self.assertTrue(report["ok"], report)
+            self.assertTrue(report["checks"]["state_valid"])
+            self.assertTrue(report["checks"]["backup_exists"])
+            self.assertTrue(report["checks"]["server_ok"])
+
+
 if __name__ == "__main__":
     unittest.main()