coverage85-step2c: host 剩余缺口真实行为测试（+18 项）

- merge_task：成功、验测失败回滚、-X ours 自动解成功（分支
  新增行全在主仓）与丢改动回滚；run_health_check（好/坏/损坏
  state/损坏 pidfile/备份存在）；notify_failure（无命令/写标记/
  坏模板/失败 rc/注入文本不执行）。
- stub Server：dispatch（含 seed-dir 与无 worktree 拒绝）、
  dispatch_from_branch、run_task_turn 三态与重试预算门、
  run_goal_loop（验测达标/标记达标/未达标/预算中断/回合异常）。
- main() 进程内：init/config/doctor/failures/health/daemon-status
  /daemon-stop/sleep/github-workflow/pr/review-comment/discard/
  automations-cron-line/restore-state/recover/dispatch/status/list/
  thread-*/skills/apps/mcp/run/compact/archive/code-review/fork/
  delegate/interrupt/merge/goal-resume；daemonize 子进程双 fork
  写 pidfile 与 marker 验证。
生产代码零改动；dispatch 建的 worktree 在 tearDown 清理。

tests/test_host_coverage85_2.py

@@ -0,0 +1,793 @@
+"""覆盖率 78%->85% 第 2 步：host.py 剩余缺口的真实行为测试。
+
+分三类，全部用临时目录/临时 git 仓库，不跑真实模型：
+- 纯函数：merge_task（成功 / 验测回滚 / -X ours 自动解成功与丢改动
+  回滚）、run_health_check、notify_failure；
+- stub Server：dispatch / dispatch_from_branch / run_task_turn /
+  run_task_turn_with_retry / run_goal_loop（goal-met / unmet /
+  预算中断 / 回合异常）；
+- main() 进程内调用（--config/--state 均指临时文件，--codex-bin
+  指假 app-server 脚本，走真实 Server 子进程协议）；
+  daemonize 另用子进程脚本验证（双 fork 后写 marker）。
+dispatch 建的 worktree 落在 HERE/work/host-worktrees（已 gitignored），
+tearDown 里逐一清理。
+"""
+
+import io
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT))
+
+import host
+from host import (
+    Server,
+    dispatch,
+    dispatch_from_branch,
+    make_worktree,
+    merge_task,
+    notify_failure,
+    run_goal_loop,
+    run_health_check,
+    run_task_turn,
+    run_task_turn_with_retry,
+)
+
+HERE = os.path.dirname(os.path.abspath(host.__file__))
+
+FAKE_SCRIPT = r'''#!/usr/bin/env python3
+"""假 app-server：只讲 JSON-RPC 行协议，不跑模型。"""
+import json
+import sys
+
+
+def send(obj):
+    sys.stdout.write(json.dumps(obj) + "\n")
+    sys.stdout.flush()
+
+
+def result_for(method):
+    if method == "thread/start":
+        return {"thread": {"id": "th-1"}}
+    if method == "thread/list":
+        return {"threads": [{"id": "th-1"}]}
+    if method == "thread/resume":
+        return {"resumed": True}
+    if method == "review/start":
+        return {"review": "ok"}
+    if method == "thread/fork":
+        return {"thread": {"id": "th-2"}}
+    if method == "thread/archive":
+        return {}
+    if method == "turn/interrupt":
+        return {"interrupted": True}
+    if method == "initialize":
+        return {"server": "fake"}
+    if method == "turn/start":
+        return {"turn": {"id": "turn-1"}}
+    return {}
+
+
+def main():
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            msg = json.loads(line)
+        except ValueError:
+            continue
+        if "method" not in msg:
+            continue
+        send({"jsonrpc": "2.0", "id": msg.get("id"),
+              "result": result_for(msg["method"])})
+
+
+main()
+'''
+
+USAGE = {"input_tokens": 3, "cached_input_tokens": 0,
+         "cache_write_input_tokens": 0, "output_tokens": 1,
+         "reasoning_output_tokens": 0, "total_tokens": 4,
+         "completions": 1}
+
+
+class StubServer:
+    def __init__(self, results=None):
+        self.results = list(results) if results is not None else [
+            {"status": "completed", "text": "done\n", "usage": dict(USAGE)}]
+        self.created = []
+
+    def create_thread(self, cwd, model=None, sandbox=None,
+                      approval_policy=None):
+        self.created.append(cwd)
+        return "th-1"
+
+    def run_turn(self, thread_id, prompt, model=None, effort=None,
+                 sandbox=None, timeout=None, interrupt_file=None):
+        if not self.results:
+            return {"status": "completed", "text": "done\n",
+                    "usage": dict(USAGE)}
+        result = self.results.pop(0)
+        if isinstance(result, Exception):
+            raise result
+        return result
+
+
+def _git(repo, *args, check=True):
+    return subprocess.run(["git", "-C", repo] + list(args),
+                          check=check, capture_output=True, text=True)
+
+
+def _make_repo(path, content="l1\nl2\nl3\nl4\nl5\n"):
+    subprocess.run(["git", "init", "-q", path], check=True,
+                   capture_output=True)
+    _git(path, "config", "user.email", "t@t.local")
+    _git(path, "config", "user.name", "T")
+    with open(os.path.join(path, "f.txt"), "w") as f:
+        f.write(content)
+    _git(path, "add", ".")
+    _git(path, "commit", "-qm", "base")
+    return path
+
+
+class MergeTaskTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.repo = _make_repo(os.path.join(self.tmp.name, "repo"))
+
+    def _worktree_for_branch(self, branch):
+        wt = os.path.join(self.tmp.name, "wt-" + branch.replace("/", "_"))
+        _git(self.repo, "worktree", "add", wt, "-b", branch, "HEAD")
+        return wt
+
+    def _head(self):
+        return _git(self.repo, "rev-parse", "HEAD").stdout.strip()
+
+    def test_merge_success_and_verify_rollback(self):
+        wt = self._worktree_for_branch("host/t1")
+        with open(os.path.join(wt, "f.txt"), "a") as f:
+            f.write("task-line\n")
+        ok, msg = merge_task(self.repo, wt, "host/t1", "t1",
+                             verify_command="true")
+        self.assertTrue(ok, msg)
+        with open(os.path.join(self.repo, "f.txt")) as f:
+            self.assertIn("task-line", f.read())
+        self.assertEqual(_git(self.repo, "branch", "--list", "host/t1",
+                              check=False).stdout.strip(), "")
+        # 验测失败：回滚到合并前 HEAD，分支与 worktree 保留。
+        wt2 = self._worktree_for_branch("host/t2")
+        with open(os.path.join(wt2, "f.txt"), "a") as f:
+            f.write("bad-line\n")
+        pre = self._head()
+        ok, msg = merge_task(self.repo, wt2, "host/t2", "t2",
+                             verify_command="false")
+        self.assertFalse(ok)
+        self.assertIn("验测失败", msg)
+        self.assertEqual(self._head(), pre)
+        self.assertIn("host/t2",
+                      _git(self.repo, "branch", "--list").stdout)
+        _git(self.repo, "worktree", "remove", "--force", wt2)
+
+    def test_merge_conflict_auto_resolve_keeps_branch_lines(self):
+        # gap2 同款：主仓与分支在同一位置插入了不同第二行（冲突），
+        # 分支的第二行是文件别处已有的行（L09），远处独有改动是
+        # L11-task。-X ours 解后分支新增行全在主仓里 -> 成功。
+        # 注意 base 必须与两侧共享绝大多数行，否则整文件算一个
+        # 冲突块，-X ours 会把远处改动也一起丢掉。
+        lfile = "".join("L%02d\n" % i for i in range(1, 13))
+        repo2 = _make_repo(os.path.join(self.tmp.name, "repo2"),
+                           content=lfile)
+        with open(os.path.join(repo2, "f.txt"), "w") as f:
+            f.write("L01\nL02\nSH\nM1\nL03\nL04\nL05\nL06\nL07\nL08\n"
+                    "L09\nL10\nL11\nL12\n")
+        _git(repo2, "add", ".")
+        _git(repo2, "commit", "-qm", "main advance")
+        wt = os.path.join(self.tmp.name, "wt-t3")
+        _git(repo2, "worktree", "add", wt, "-b", "host/t3", "HEAD~1")
+        with open(os.path.join(wt, "f.txt"), "w") as f:
+            f.write("L01\nL02\nSH\nL09\nL03\nL04\nL05\nL06\nL07\nL08\n"
+                    "L09\nL10\nL11-task\nL12\n")
+        out = io.StringIO()
+        with redirect_stdout(out):
+            ok, msg = merge_task(repo2, wt, "host/t3", "t3",
+                                 verify_command="true")
+        self.assertTrue(ok, msg)
+        self.assertIn("-X ours", out.getvalue())
+        with open(os.path.join(repo2, "f.txt")) as f:
+            content = f.read()
+        self.assertIn("L11-task", content)
+        self.assertIn("SH", content)
+
+    def test_merge_conflict_loses_branch_line_rolls_back(self):
+        # 同一行互改、无其它改动：-X ours 解后分支行丢失 -> 回滚。
+        with open(os.path.join(self.repo, "f.txt"), "w") as f:
+            f.write("l1\nl2\nl3-main\nl4\nl5\n")
+        _git(self.repo, "add", ".")
+        _git(self.repo, "commit", "-qm", "main advance")
+        wt = os.path.join(self.tmp.name, "wt-t4")
+        _git(self.repo, "worktree", "add", wt, "-b", "host/t4", "HEAD~1")
+        with open(os.path.join(wt, "f.txt"), "w") as f:
+            f.write("l1\nl2\nl3-task\nl4\nl5\n")
+        pre = self._head()
+        ok, msg = merge_task(self.repo, wt, "host/t4", "t4",
+                             verify_command="true")
+        self.assertFalse(ok)
+        self.assertIn("改动被丢弃", msg)
+        self.assertEqual(self._head(), pre)
+
+
+class HealthNotifyTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.state_path = os.path.join(self.tmp.name, "host-state.json")
+        host.save_state(self.state_path, {"tasks": {}, "automations": []})
+
+    def test_health_check_variants(self):
+        class FakeSrv:
+            def __init__(self, ok=True):
+                self.ok = ok
+
+            def initialize(self):
+                if not self.ok:
+                    raise RuntimeError("init boom")
+
+            def list_threads(self):
+                return {"data": [{"id": "th-1"}, {"id": "th-2"}]}
+
+            def close(self):
+                pass
+
+        bin_path = os.path.join(self.tmp.name, "codex-bin")
+        with open(bin_path, "w") as f:
+            f.write("#!/bin/sh\n")
+        report = run_health_check(self.state_path, {}, bin_path, None,
+                                  "deny",
+                                  server_factory=lambda: FakeSrv(True))
+        self.assertTrue(report["ok"])
+        self.assertTrue(report["checks"]["codex_bin_exists"])
+        self.assertTrue(report["checks"]["state_valid"])
+        self.assertTrue(report["checks"]["server_ok"])
+        self.assertEqual(report["checks"]["server_threads"], 2)
+        self.assertFalse(report["checks"]["backup_exists"])
+        with open(self.state_path + ".bak", "w") as f:
+            f.write("{}")
+        report = run_health_check(self.state_path, {}, bin_path, None,
+                                  "deny",
+                                  server_factory=lambda: FakeSrv(True))
+        self.assertTrue(report["checks"]["backup_exists"])
+        report = run_health_check(self.state_path, {}, "/nonexistent-bin",
+                                  None, "deny",
+                                  server_factory=lambda: FakeSrv(False))
+        self.assertFalse(report["ok"])
+        self.assertFalse(report["checks"]["codex_bin_exists"])
+        self.assertFalse(report["checks"]["server_ok"])
+        self.assertIn("init boom", report["checks"]["server_error"])
+        # 损坏的 state：state_valid 为 False，failed 列表为空。
+        with open(self.state_path, "w") as f:
+            f.write("{broken")
+        report = run_health_check(self.state_path, {}, bin_path, None,
+                                  "deny",
+                                  server_factory=lambda: FakeSrv(True))
+        self.assertFalse(report["checks"]["state_valid"])
+        self.assertEqual(report["checks"]["failed_tasks"], [])
+        # 损坏的 pidfile：daemon_alive 为 False（先删掉坏 state 文件，
+        # save_state 按设计拒绝在坏文件上重写）。
+        pidfile = os.path.join(self.tmp.name, "host-daemon.pid")
+        with open(pidfile, "w") as f:
+            f.write("not-a-pid")
+        os.unlink(self.state_path)
+        host.save_state(self.state_path, {"tasks": {}, "automations": []})
+        report = run_health_check(self.state_path, {}, bin_path, None,
+                                  "deny",
+                                  server_factory=lambda: FakeSrv(True))
+        self.assertFalse(report["checks"]["daemon_alive"])
+        self.assertIsNone(report["checks"]["daemon_pid"])
+
+    def test_notify_failure_variants(self):
+        msg = notify_failure(self.state_path, {}, "t1",
+                             {"status": "failed", "text": "boom"})
+        self.assertTrue(msg.startswith("FAILED t1:"))
+        log_path = os.path.join(self.tmp.name, "host-run.log")
+        with open(log_path) as f:
+            self.assertIn("FAILED t1:", f.read())
+        marker = os.path.join(self.tmp.name, "notified.txt")
+        msg = notify_failure(
+            self.state_path, {"notify_command": "echo {task} >> " + marker},
+            "t2", {"status": "failed", "text": "x"})
+        with open(marker) as f:
+            self.assertIn("t2", f.read())
+        # 坏模板：不跑命令、记 template error。
+        marker2 = os.path.join(self.tmp.name, "notified2.txt")
+        notify_failure(
+            self.state_path,
+            {"notify_command": "echo hi >> " + marker2 + " {unknown}"},
+            "t3", {"status": "failed", "text": "x"})
+        self.assertFalse(os.path.exists(marker2))
+        with open(log_path) as f:
+            self.assertIn("template error", f.read())
+        # 失败命令：记 rc；注入文本不执行。
+        notify_failure(self.state_path, {"notify_command": "false"},
+                       "t4", {"status": "failed", "text": "x"})
+        with open(log_path) as f:
+            self.assertIn("notify command rc=1", f.read())
+        pwn = os.path.join(self.tmp.name, "pwned")
+        notify_failure(
+            self.state_path,
+            {"notify_command": "echo {text} >> " + marker},
+            "t5", {"status": "failed",
+                   "text": "$(touch %s)" % pwn})
+        self.assertFalse(os.path.exists(pwn))
+
+
+class DaemonizeTest(unittest.TestCase):
+    def test_daemonize_double_fork_writes_pidfile_and_marker(self):
+        script = os.path.join(tempfile.gettempdir(),
+                              "cov85_daemon_probe_%d.py" % os.getpid())
+        pidfile = os.path.join(tempfile.gettempdir(),
+                               "cov85_daemon_%d.pid" % os.getpid())
+        logfile = os.path.join(tempfile.gettempdir(),
+                               "cov85_daemon_%d.log" % os.getpid())
+        marker = os.path.join(tempfile.gettempdir(),
+                              "cov85_daemon_%d.marker" % os.getpid())
+        for p in (pidfile, logfile, marker):
+            if os.path.exists(p):
+                os.unlink(p)
+        code = (
+            "import os, sys\n"
+            "sys.path.insert(0, %r)\n"
+            "import host\n"
+            "cont = host.daemonize(%r, %r)\n"
+            "if not cont:\n"
+            "    sys.exit(42)\n"
+            "open(%r, 'w').write(str(os.getpid()))\n"
+            % (str(REPO_ROOT), pidfile, logfile, marker))
+        with open(script, "w") as f:
+            f.write(code)
+        try:
+            proc = subprocess.run([sys.executable, script],
+                                  capture_output=True, text=True, timeout=30)
+            self.assertEqual(proc.returncode, 42)
+            deadline = time.time() + 5
+            while not os.path.exists(marker) and time.time() < deadline:
+                time.sleep(0.05)
+            self.assertTrue(os.path.exists(marker))
+            with open(pidfile) as f:
+                pid_text = f.read().strip()
+            with open(marker) as f:
+                self.assertEqual(f.read().strip(), pid_text)
+            self.assertTrue(os.path.exists(logfile))
+        finally:
+            for p in (script, pidfile, logfile, marker):
+                if os.path.exists(p):
+                    os.unlink(p)
+
+
+class DispatchGoalTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.repo = _make_repo(os.path.join(self.tmp.name, "repo"))
+        self.state_path = os.path.join(self.tmp.name, "host-state.json")
+        self.state = {"tasks": {}, "automations": []}
+        self.created_names = []
+
+    def tearDown(self):
+        for name in self.created_names:
+            wt = os.path.join(HERE, "work", "host-worktrees", name)
+            if os.path.exists(wt):
+                subprocess.run(["git", "-C", self.repo, "worktree",
+                                "remove", "--force", wt],
+                               capture_output=True)
+            subprocess.run(["git", "-C", self.repo, "branch", "-D",
+                            "host/" + name], capture_output=True)
+        self.tmp.cleanup()
+
+    def test_dispatch_and_existing_and_seed(self):
+        srv = StubServer()
+        task = dispatch(srv, self.state, "t1", "do thing", self.repo,
+                        None, None, None, "never")
+        self.created_names.append("t1")
+        self.assertEqual(task["thread_id"], "th-1")
+        self.assertEqual(task["status"], "running")
+        self.assertTrue(os.path.isdir(task["cwd"]))
+        self.assertIs(self.state["tasks"]["t1"], task)
+        with self.assertRaises(RuntimeError):
+            dispatch(srv, self.state, "t1", "again", self.repo,
+                     None, None, None, "never")
+        seed = os.path.join(self.tmp.name, "seed")
+        os.makedirs(os.path.join(seed, "sub"))
+        with open(os.path.join(seed, "a.txt"), "w") as f:
+            f.write("seeded")
+        with open(os.path.join(seed, "sub", "b.txt"), "w") as f:
+            f.write("seeded-sub")
+        task2 = dispatch(srv, self.state, "t2", "seeded task", self.repo,
+                         None, None, None, "never", seed_dir=seed)
+        self.created_names.append("t2")
+        with open(os.path.join(task2["cwd"], "a.txt")) as f:
+            self.assertEqual(f.read(), "seeded")
+        with open(os.path.join(task2["cwd"], "sub", "b.txt")) as f:
+            self.assertEqual(f.read(), "seeded-sub")
+        # seed-dir 没有 worktree 时 loud fail。
+        with self.assertRaises(RuntimeError):
+            dispatch(srv, self.state, "t3", "x", self.repo,
+                     None, None, None, "never", use_worktree=False,
+                     seed_dir=seed)
+        with self.assertRaises(RuntimeError):
+            dispatch(srv, self.state, "t4", "x", None,
+                     None, None, None, "never", seed_dir=seed)
+
+    def test_dispatch_from_branch(self):
+        srv = StubServer()
+        dep = dispatch(srv, self.state, "dep", "dep work", self.repo,
+                       None, None, None, "never")
+        self.created_names.append("dep")
+        with open(os.path.join(dep["cwd"], "dep.txt"), "w") as f:
+            f.write("dep-change")
+        task, err = dispatch_from_branch(
+            srv, self.state, "child", "child work", self.repo,
+            dep["branch"], dep["cwd"], None, None, None, "never")
+        self.created_names.append("child")
+        self.assertEqual(err, "")
+        self.assertEqual(task["branch"], "host/child")
+        with open(os.path.join(task["cwd"], "dep.txt")) as f:
+            self.assertEqual(f.read(), "dep-change")
+        task2, err2 = dispatch_from_branch(
+            srv, self.state, "child", "again", self.repo,
+            dep["branch"], dep["cwd"], None, None, None, "never")
+        self.assertIsNone(task2)
+        self.assertIn("已存在", err2)
+
+    def test_run_task_turn_status_mapping_and_retry(self):
+        state = {"tasks": {}, "automations": []}
+        task = {"name": "r1", "thread_id": "th-1", "status": "running",
+                "turns": []}
+        srv = StubServer([{"status": "completed", "text": "ok",
+                           "usage": dict(USAGE)}])
+        result = run_task_turn(srv, state, task, "prompt", None, None)
+        self.assertEqual(task["status"], "completed")
+        self.assertEqual(result["status"], "completed")
+        task2 = {"name": "r2", "thread_id": "th-1", "status": "running",
+                 "turns": []}
+        srv2 = StubServer([{"status": "timeout", "text": "partial",
+                            "usage": dict(USAGE)}])
+        run_task_turn(srv2, state, task2, "prompt", None, None)
+        self.assertEqual(task2["status"], "interrupted")
+        task3 = {"name": "r3", "thread_id": "th-1", "status": "running",
+                 "turns": []}
+        srv3 = StubServer([{"status": "failed", "text": "",
+                            "usage": dict(USAGE)}])
+        run_task_turn(srv3, state, task3, "prompt", None, None)
+        self.assertEqual(task3["status"], "failed")
+        # 重试：先失败再成功，usage 汇总且 retry_count 为 1。
+        task4 = {"name": "r4", "thread_id": "th-1", "status": "running",
+                 "turns": [], "max_retries": 1}
+        srv4 = StubServer([{"status": "failed", "text": "",
+                            "usage": dict(USAGE)},
+                           {"status": "completed", "text": "ok",
+                            "usage": dict(USAGE)}])
+        result = run_task_turn_with_retry(srv4, state, task4, "prompt",
+                                          None, None)
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(task4["retry_count"], 1)
+        self.assertEqual(result["usage"]["input_tokens"], 6)
+        # 预算门：cfg 限 1 轮且已用 1 轮时直接失败，不跑 turn。
+        today = time.strftime("%Y-%m-%d", time.localtime())
+        state2 = {"tasks": {}, "automations": [],
+                  "usage": {"date": today, "turns": 1, "failed_turns": 0}}
+        task5 = {"name": "r5", "thread_id": "th-1", "status": "running",
+                 "turns": [], "max_retries": 0}
+        srv5 = StubServer()
+        result = run_task_turn_with_retry(
+            srv5, state2, task5, "prompt", None, None,
+            cfg={"max_daily_turns": 1})
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("budget exceeded", result["error"])
+        self.assertEqual(task5["status"], "failed")
+
+    def _goal_state(self, name):
+        task = {"name": name, "thread_id": "th-1", "status": "running",
+                "turns": [], "cwd": self.tmp.name}
+        state = {"tasks": {name: task}, "automations": []}
+        return state, task
+
+    def test_goal_loop_met_by_verify_and_by_marker(self):
+        state, task = self._goal_state("g1")
+        out = io.StringIO()
+        with redirect_stdout(out):
+            res = run_goal_loop(StubServer(), state, self.state_path,
+                                task, "do it", "true", 3, None, None, None)
+        self.assertTrue(res["ok"])
+        self.assertEqual(task["status"], "goal-met")
+        self.assertEqual(res["rounds_done"], 1)
+        self.assertIn("goal met after rounds", out.getvalue())
+        state, task = self._goal_state("g2")
+        srv = StubServer([{"status": "completed",
+                           "text": "work done\n目标已达成\n",
+                           "usage": dict(USAGE)}])
+        res = run_goal_loop(srv, state, self.state_path, task, "do it",
+                            None, 3, None, None, None)
+        self.assertTrue(res["ok"])
+        self.assertEqual(task["status"], "goal-met")
+
+    def test_goal_loop_unmet_budget_and_exception(self):
+        state, task = self._goal_state("g3")
+        res = run_goal_loop(StubServer(), state, self.state_path, task,
+                            "do it", "false", 1, None, None, None)
+        self.assertFalse(res["ok"])
+        self.assertEqual(task["status"], "goal-unmet")
+        self.assertEqual(res["rounds_done"], 1)
+        today = time.strftime("%Y-%m-%d", time.localtime())
+        state, task = self._goal_state("g4")
+        state["usage"] = {"date": today, "turns": 5, "failed_turns": 0}
+        res = run_goal_loop(StubServer(), state, self.state_path, task,
+                            "do it", "true", 3, None, None, None,
+                            cfg={"max_daily_turns": 3})
+        self.assertTrue(res["budget_stopped"])
+        self.assertEqual(task["status"], "interrupted")
+        state, task = self._goal_state("g5")
+        srv = StubServer([RuntimeError("turn boom")])
+        res = run_goal_loop(srv, state, self.state_path, task, "do it",
+                            "true", 3, None, None, None)
+        self.assertTrue(res["interrupted"])
+        self.assertEqual(task["status"], "interrupted")
+        self.assertIn("turn boom", task["goal_interrupted_reason"])
+
+
+class MainCommandsTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.repo = _make_repo(os.path.join(self.tmp.name, "repo"))
+        self.config_path = os.path.join(self.tmp.name, "host-config.json")
+        with open(self.config_path, "w") as f:
+            json.dump({}, f)
+        self.state_path = os.path.join(self.tmp.name, "host-state.json")
+        self.fake_bin = os.path.join(self.tmp.name, "fake-codex.py")
+        with open(self.fake_bin, "w") as f:
+            f.write(FAKE_SCRIPT)
+        os.chmod(self.fake_bin, 0o755)
+        self._old_argv = sys.argv
+        self.created_names = []
+
+    def tearDown(self):
+        sys.argv = self._old_argv
+        for name in self.created_names:
+            wt = os.path.join(HERE, "work", "host-worktrees", name)
+            if os.path.exists(wt):
+                subprocess.run(["git", "-C", self.repo, "worktree",
+                                "remove", "--force", wt],
+                               capture_output=True)
+            subprocess.run(["git", "-C", self.repo, "branch", "-D",
+                            "host/" + name], capture_output=True)
+        self.tmp.cleanup()
+
+    def _main(self, *argv):
+        sys.argv = ["host.py", "--config", self.config_path,
+                    "--state", self.state_path] + list(argv)
+        out = io.StringIO()
+        with redirect_stdout(out):
+            host.main()
+        return out.getvalue()
+
+    def _main_code(self, *argv):
+        sys.argv = ["host.py", "--config", self.config_path,
+                    "--state", self.state_path] + list(argv)
+        out = io.StringIO()
+        with redirect_stdout(out):
+            try:
+                host.main()
+            except SystemExit as exc:
+                return exc.code, out.getvalue()
+        return 0, out.getvalue()
+
+    def _load_state(self):
+        with open(self.state_path) as f:
+            return json.load(f)
+
+    def test_init_config_doctor_failures(self):
+        os.unlink(self.config_path)
+        out = self._main("init", "--repo", self.repo)
+        self.assertIn("配置文件已写", out)
+        out = self._main("init")
+        self.assertIn("配置文件已存在", out)
+        out = self._main("config")
+        cfg = json.loads(out)
+        self.assertEqual(cfg["model"], "gpt-5.6-terra")
+        out = self._main("doctor")
+        doc = json.loads(out)
+        self.assertEqual(doc["failed_tasks"], [])
+        out = self._main("failures")
+        self.assertEqual(json.loads(out), {"failed_tasks": []})
+        host.save_state(self.state_path,
+                        {"tasks": {"bad": {"name": "bad",
+                                           "status": "failed"}},
+                         "automations": []})
+        out = self._main("failures")
+        self.assertEqual(json.loads(out), {"failed_tasks": ["bad"]})
+
+    def test_health_good_and_bad(self):
+        out = self._main("--codex-bin", self.fake_bin, "health")
+        report = json.loads(out)
+        self.assertTrue(report["ok"])
+        self.assertTrue(report["checks"]["server_ok"])
+        code, out = self._main_code("--codex-bin", "/nonexistent-bin",
+                                    "health")
+        self.assertEqual(code, 1)
+        report = json.loads(out)
+        self.assertFalse(report["ok"])
+
+    def test_daemon_status_stop_and_sleep(self):
+        out = self._main("daemon-status")
+        doc = json.loads(out)
+        self.assertFalse(doc["alive"])
+        self.assertIsNone(doc["pid"])
+        out = self._main("daemon-stop")
+        self.assertIn("daemon 未运行", out)
+        pidfile = os.path.join(self.tmp.name, "host-daemon.pid")
+        with open(pidfile, "w") as f:
+            f.write("garbage")
+        out = self._main("daemon-stop")
+        self.assertIn("pidfile 损坏", out)
+        self.assertFalse(os.path.exists(pidfile))
+        out = self._main("sleep", "--seconds", "0")
+        self.assertEqual(json.loads(out)["slept_seconds"], 0)
+        code, _ = self._main_code("sleep", "--seconds", "61")
+        self.assertEqual(code, 1)
+
+    def test_github_workflow_pr_and_review_comment(self):
+        code, _ = self._main_code("github-workflow")
+        self.assertEqual(code, 1)
+        code, _ = self._main_code("pr", "--task", "missing")
+        self.assertEqual(code, 1)
+        host.save_state(self.state_path,
+                        {"tasks": {"t1": {"name": "t1",
+                                          "thread_id": "th-1",
+                                          "status": "completed",
+                                          "review_comments": []}},
+                         "automations": []})
+        out = self._main("review-comment", "--task", "t1",
+                         "--file", "a.py", "--line", "3",
+                         "--comment", "fix this")
+        comment = json.loads(out)
+        self.assertEqual(comment["file"], "a.py")
+        self.assertEqual(comment["line"], 3)
+        state = self._load_state()
+        self.assertEqual(len(state["tasks"]["t1"]["review_comments"]), 1)
+        out = self._main("discard", "--task", "missing")
+        self.assertIn("任务不存在", out)
+        out = self._main("discard", "--task", "t1")
+        self.assertIn("discarded t1", out)
+        state = self._load_state()
+        self.assertEqual(state["tasks"]["t1"]["status"], "discarded")
+
+    def test_automations_cron_line_restore_recover(self):
+        out = self._main("automations", "--cron-line")
+        self.assertIn("codex-host-run-due", out)
+        out = self._main("restore-state")
+        self.assertIn("state 完好", out)
+        out = self._main("recover")
+        self.assertEqual(json.loads(out), {"recovered": []})
+
+    def test_dispatch_status_list_threads_and_run(self):
+        self.created_names.append("dtask")
+        out = self._main("--codex-bin", self.fake_bin,
+                         "--turn-timeout", "1",
+                         "dispatch", "--name", "dtask",
+                         "--task", "do thing", "--repo", self.repo)
+        info = json.loads(out)
+        self.assertEqual(info["task"], "dtask")
+        self.assertEqual(info["thread_id"], "th-1")
+        state = self._load_state()
+        self.assertEqual(state["tasks"]["dtask"]["status"], "interrupted")
+        out = self._main("status", "--task", "dtask")
+        self.assertEqual(json.loads(out)["name"], "dtask")
+        out = self._main("--codex-bin", self.fake_bin, "list")
+        self.assertIn("dtask", out)
+        out = self._main("--codex-bin", self.fake_bin, "thread-list")
+        self.assertIn("th-1", out)
+        out = self._main("--codex-bin", self.fake_bin, "skills-list")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin, "apps-list")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin, "mcp-servers-list")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin, "mcp-servers-reload")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "thread-read", "--thread-id", "th-1")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "thread-rename", "--thread-id", "th-1",
+                         "--name", "renamed")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "thread-archive", "--thread-id", "th-1")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "thread-unarchive", "--thread-id", "th-1")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "thread-compact", "--thread-id", "th-1")
+        self.assertEqual(json.loads(out), {})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "--turn-timeout", "1",
+                         "run", "--task", "dtask", "--prompt", "again")
+        result = json.loads(out)
+        self.assertEqual(result["status"], "timeout")
+        out = self._main("--codex-bin", self.fake_bin,
+                         "compact", "--task", "dtask")
+        self.assertIn("compact dtask", out)
+        out = self._main("--codex-bin", self.fake_bin,
+                         "archive", "--task", "dtask")
+        self.assertIn("archived dtask", out)
+        out = self._main("--codex-bin", self.fake_bin,
+                         "code-review", "--task", "dtask",
+                         "--instructions", "look carefully")
+        self.assertIn("code-review dtask", out)
+        self.assertIn("review", out)
+
+    def test_fork_delegate_interrupt_and_merge(self):
+        self.created_names.append("ptask")
+        self.created_names.append("ftask")
+        self._main("--codex-bin", self.fake_bin,
+                   "--turn-timeout", "1",
+                   "dispatch", "--name", "ptask",
+                   "--task", "parent work", "--repo", self.repo)
+        out = self._main("--codex-bin", self.fake_bin,
+                         "fork", "--task", "ptask", "--name", "ftask")
+        self.assertIn("forked ptask -> ftask", out)
+        self.assertIn("thread=th-2", out)
+        out = self._main("--codex-bin", self.fake_bin,
+                         "--turn-timeout", "1",
+                         "delegate", "--parent-task", "ptask",
+                         "--task", "sub1", "--name", "sub1")
+        doc = json.loads(out)
+        self.assertEqual(doc["sub_task"], "sub1")
+        # 中断：无 turn_id 与有 turn_id 两路。
+        host.save_state(self.state_path,
+                        {"tasks": {"it": {"name": "it",
+                                          "thread_id": "th-1",
+                                          "status": "running",
+                                          "turns": []}},
+                         "automations": []})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "interrupt", "--task", "it")
+        self.assertIn("没有可中断的 turn_id", out)
+        host.save_state(self.state_path,
+                        {"tasks": {"it": {"name": "it",
+                                          "thread_id": "th-1",
+                                          "status": "running",
+                                          "turns": [{"result": {
+                                              "turn_id": "turn-1",
+                                              "status": "completed"}}]}},
+                         "automations": []})
+        out = self._main("--codex-bin", self.fake_bin,
+                         "interrupt", "--task", "it")
+        self.assertIn("turn=turn-1", out)
+        state = self._load_state()
+        self.assertEqual(state["tasks"]["it"]["status"], "interrupted")
+        # merge：ptask 的 worktree 无改动，合并即成功。
+        out = self._main("merge", "--task", "ptask")
+        self.assertIn("merged", out)
+        # goal-resume 对不存在的任务拒绝。
+        out = self._main("--codex-bin", self.fake_bin,
+                         "goal-resume", "--task", "missing")
+        self.assertIn("拒绝", out)
+        # pr 对有任务但无 gh 时退出 1。
+        code, _ = self._main_code("pr", "--task", "ptask")
+        self.assertEqual(code, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()