diff --git a/README.md b/README.md index f043c10..878ea96 100644 --- a/README.md +++ b/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 上是“会话内常驻 + + 回收后重拉”的定位。 ## 开源没给到的（别指望照搬） diff --git a/TEST-REPORT.md b/TEST-REPORT.md index 99f426f..bfe8b2c 100644 --- a/TEST-REPORT.md +++ b/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 health`（无模型成本）和 / + 或 `run-due`（到期才跑模型），需用户定间隔和是否允许 + 自动跑模型后才装。 diff --git a/host.py b/host.py index 9976107..dcad452 100644 --- a/host.py +++ b/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: diff --git a/tests/test_host.py b/tests/test_host.py index 0c5bdcf..ead48f8 100644 --- a/tests/test_host.py +++ b/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()