From 887d8f7afb442221da9e6bc0a7f1449608991256 Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Sat, 31 Oct 2026 09:40:15 +0000
Subject: [PATCH] =?UTF-8?q?coverage85-step4f:=20host=20=E7=BA=AF=E5=87=BD?=
 =?UTF-8?q?=E6=95=B0=E4=B8=8E=E5=B0=8F=E6=96=B9=E6=B3=95=E9=95=BF=E5=B0=BE?=
 =?UTF-8?q?=E6=B5=8B=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tests/test_host_coverage85_4.py | 685 ++++++++++++++++++++++++++++++++
 1 file changed, 685 insertions(+)
 create mode 100644 tests/test_host_coverage85_4.py

diff --git a/tests/test_host_coverage85_4.py b/tests/test_host_coverage85_4.py
new file mode 100644
index 0000000..088daf3
--- /dev/null
+++ b/tests/test_host_coverage85_4.py
@@ -0,0 +1,685 @@
+"""覆盖率 85% 第 4 步第 6 批：host.py 纯函数与小方法长尾。
+
+只动测试。全部是直接调用生产函数/方法：用量汇总、链路
+摘要、等待轮询、状态锁与恢复、git diff/合并/验测、PR 解析
+与正文、通知、doctor、Server 的 trace/close（不拉起真实
+app-server 进程）。
+"""
+
+import json
+import os
+import queue
+import subprocess
+import tempfile
+import time
+import unittest
+from unittest import mock
+
+import host
+
+
+def _git(repo, *args, **kw):
+    return subprocess.run(["git", "-C", repo] + list(args),
+                          capture_output=True, text=True, **kw)
+
+
+def _make_repo(base):
+    repo = os.path.join(base, "repo")
+    os.makedirs(repo)
+    _git(repo, "init", "-q", "-b", "main")
+    _git(repo, "config", "user.email", "t@t.t")
+    _git(repo, "config", "user.name", "t")
+    with open(os.path.join(repo, "a.py"), "w") as f:
+        f.write("x = 1\n")
+    _git(repo, "add", ".")
+    _git(repo, "commit", "-q", "-m", "init")
+    return repo
+
+
+class UsageAndEventsTest(unittest.TestCase):
+    def test_sum_turn_usage(self):
+        events = [
+            {"method": "other", "params": {}},
+            {"method": "thread/tokenUsage/updated",
+             "params": {"threadId": "th-1", "turnId": "tu-1",
+                        "tokenUsage": {"last": {
+                            "inputTokens": 10, "outputTokens": 5,
+                            "cachedInputTokens": 2,
+                            "totalTokens": 15}}}},
+            {"method": "thread/tokenUsage/updated",
+             "params": {"threadId": "th-1", "turnId": "tu-2",
+                        "tokenUsage": {"last": {
+                            "inputTokens": 7, "outputTokens": 3}}}},
+            {"method": "thread/tokenUsage/updated",
+             "params": {"threadId": "th-2", "turnId": "tu-1",
+                        "tokenUsage": {"last": {"inputTokens": 99}}}},
+        ]
+        total = host._sum_turn_usage(events, "th-1", "tu-1")
+        self.assertEqual(total["input_tokens"], 10)
+        self.assertEqual(total["output_tokens"], 5)
+        self.assertEqual(total["cached_input_tokens"], 2)
+        self.assertEqual(total["completions"], 1)
+        total = host._sum_turn_usage(events, "th-1", None)
+        self.assertEqual(total["input_tokens"], 17)
+        self.assertEqual(total["completions"], 2)
+        self.assertIsNone(host._sum_turn_usage(events, "th-9", None))
+        self.assertIsNone(host._sum_turn_usage([], "th-1", None))
+        self.assertIsNone(host._sum_turn_usage(None, "th-1", None))
+
+    def test_partial_text_from_events(self):
+        events = [
+            {"params": {"item": {"type": "agentMessage",
+                                 "text": "first"}}},
+            {"params": {"item": {"type": "agentMessage",
+                                 "text": "first"}}},
+            {"params": {"item": {"type": "delta", "text": "skip"}}},
+            {"params": {"item": {"type": "agentMessage",
+                                 "text": "second"}}},
+            {"params": {}},
+        ]
+        self.assertEqual(host._partial_text_from_events(events),
+                         "first\nsecond")
+        self.assertEqual(host._partial_text_from_events([]), "")
+        self.assertEqual(host._partial_text_from_events(None), "")
+
+    def test_add_usage(self):
+        self.assertIsNone(host._add_usage(None, None))
+        self.assertEqual(host._add_usage(None, {"a": 1}), {"a": 1})
+        self.assertEqual(host._add_usage({"a": 1}, None), {"a": 1})
+        out = host._add_usage({"a": 1, "s": "x"}, {"a": 2, "b": 3})
+        self.assertEqual(out, {"a": 3, "s": "x", "b": 3})
+
+    def test_interrupt_request_path_sanitizes(self):
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            p = host._interrupt_request_path(
+                os.path.join(tmp.name, "host-state.json"), "a/b c")
+            self.assertEqual(os.path.basename(p), "a-b-c.json")
+            self.assertEqual(os.path.basename(os.path.dirname(p)),
+                             "interrupt-requests")
+        finally:
+            tmp.cleanup()
+
+
+class WaitTurnTest(unittest.TestCase):
+    def test_completed_immediately(self):
+        q = queue.Queue()
+        q.put({"done": True})
+        status, done = host._wait_turn(q, timeout=5)
+        self.assertEqual((status, done), ("completed", {"done": True}))
+
+    def test_timeout_and_progress(self):
+        q = queue.Queue()
+        seen = []
+        status, done = host._wait_turn(
+            q, timeout=0.35, progress_interval=0.1,
+            on_progress=lambda e: seen.append(round(e, 1)))
+        self.assertEqual(status, "timeout")
+        self.assertIsNone(done)
+        self.assertGreaterEqual(len(seen), 2)
+
+    def test_interrupt_requested_and_swallow(self):
+        q = queue.Queue()
+        calls = {"n": 0}
+
+        def _check():
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise RuntimeError("check 炸了")
+            return True
+        status, _ = host._wait_turn(q, timeout=5,
+                                    interrupt_check=_check)
+        self.assertEqual(status, "interrupted-requested")
+        # on_progress 抛错也被吞掉，不影响 timeout 判定。
+        status, _ = host._wait_turn(
+            q, timeout=0.25, progress_interval=0.1,
+            on_progress=lambda e: (_ for _ in ()).throw(
+                RuntimeError("progress 炸了")))
+        self.assertEqual(status, "timeout")
+
+    def test_unlimited_waits_for_done(self):
+        q = queue.Queue()
+
+        def _put():
+            time.sleep(0.2)
+            q.put("x")
+        import threading
+        threading.Thread(target=_put, daemon=True).start()
+        status, done = host._wait_turn(q, timeout=0)
+        self.assertEqual((status, done), ("completed", "x"))
+
+
+class SummarizeTraceTest(unittest.TestCase):
+    def _params(self, item, **kw):
+        p = {"item": item}
+        p.update(kw)
+        return p
+
+    def test_all_item_types(self):
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "commandExecution", "id": "c1",
+                               "command": "ls", "cwd": "/tmp",
+                               "exitCode": 0, "durationMs": 5},
+                              completedAtMs=123, threadId="th-1"))
+        self.assertEqual(out[0]["kind"], "exec")
+        self.assertEqual(out[0]["ts"], 123)
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "fileChange", "id": "f1",
+                               "status": "done",
+                               "changes": [{"path": "a.py", "kind": "edit"},
+                                           "bad",
+                                           {"path": "b.py"}]},
+                              thread={"id": "th-9"}))
+        self.assertEqual([e["id"] for e in out],
+                         ["f1:a.py", "f1:b.py"])
+        self.assertEqual(out[0]["thread"], "th-9")
+        long_text = "z" * 400
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "agentMessage", "id": "m1",
+                               "text": long_text}))
+        self.assertEqual(len(out[0]["text"]), 300)
+        self.assertTrue(out[0]["truncated"])
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "reasoning", "id": "r1",
+                               "summary": ["a", 5, "b"]}))
+        self.assertEqual(out[0]["summary"], "a b")
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "mcpToolCall", "id": "t1",
+                               "server": "s", "tool": "t",
+                               "status": "ok"}))
+        self.assertEqual(out[0]["kind"], "tool")
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "functionCallOutput", "id": "o1",
+                               "name": "f"}))
+        self.assertEqual(out[0]["kind"], "call")
+        out = host._summarize_trace_item(
+            "m", self._params({"type": "plan", "id": "p1",
+                               "text": "plan text"}))
+        self.assertEqual(out[0]["kind"], "plan")
+
+    def test_empty_and_unknown(self):
+        self.assertEqual(host._summarize_trace_item("m", {}), [])
+        self.assertEqual(host._summarize_trace_item(
+            "m", {"item": "not-a-dict"}), [])
+        self.assertEqual(host._summarize_trace_item(
+            "m", {"item": {"type": "userMessage", "text": "hi"}}), [])
+        self.assertEqual(host._summarize_trace_item(
+            "m", {"item": {"type": "agentMessage", "text": ""}}), [])
+        self.assertEqual(host._summarize_trace_item(
+            "m", {"item": {"type": "reasoning", "summary": []}}), [])
+        self.assertEqual(host._summarize_trace_item(
+            "m", {"item": {"type": "plan", "text": ""}}), [])
+
+    def test_trace_path_and_marker(self):
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            state_path = os.path.join(tmp.name, "host-state.json")
+            path = host._trace_file_path(state_path, "a/b")
+            self.assertEqual(os.path.basename(path), "a-b.jsonl")
+            host._append_trace_marker(state_path, "a/b",
+                                      {"kind": "mark"})
+            with open(path) as f:
+                row = json.loads(f.readline())
+            self.assertEqual(row["kind"], "mark")
+        finally:
+            tmp.cleanup()
+
+
+class StateLockAndReadTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_tmp_path_unique(self):
+        a = host._state_tmp_path(self.path)
+        b = host._state_tmp_path(self.path)
+        self.assertNotEqual(a, b)
+        self.assertTrue(a.startswith(self.path + ".tmp."))
+
+    def test_lock_exclusive_and_shared(self):
+        with host._state_lock(self.path):
+            self.assertTrue(os.path.exists(self.path + ".lockdir"))
+        self.assertFalse(os.path.exists(self.path + ".lockdir"))
+        with host._state_lock(self.path, exclusive=False):
+            self.assertFalse(os.path.exists(self.path + ".lockdir"))
+
+    def test_lock_stale_removed_and_timeout(self):
+        lockdir = self.path + ".lockdir"
+        os.makedirs(lockdir)
+        old = time.time() - 10000
+        os.utime(lockdir, (old, old))
+        with host._state_lock(self.path):
+            pass
+        self.assertFalse(os.path.exists(lockdir))
+        os.makedirs(lockdir)
+        with mock.patch.object(host, "_STATE_LOCK_TIMEOUT", 0.01):
+            with self.assertRaises(RuntimeError):
+                with host._state_lock(self.path):
+                    pass
+        os.rmdir(lockdir)
+
+    def test_read_state_file_variants(self):
+        self.assertEqual(host._read_state_file(self.path),
+                         {"tasks": {}, "automations": []})
+        with open(self.path, "w") as f:
+            f.write("{bad")
+        self.assertEqual(host._read_state_file(self.path),
+                         {"tasks": {}, "automations": []})
+        with self.assertRaises(RuntimeError):
+            host._read_state_file_for_save(self.path)
+        with open(self.path, "w") as f:
+            json.dump(["not", "a", "dict"], f)
+        with self.assertRaises(RuntimeError):
+            host._read_state_file_for_save(self.path)
+        with open(self.path, "w") as f:
+            json.dump({"tasks": [], "automations": []}, f)
+        with self.assertRaises(RuntimeError):
+            host._read_state_file_for_save(self.path)
+        with open(self.path, "w") as f:
+            json.dump({"tasks": {}, "automations": {}}, f)
+        with self.assertRaises(RuntimeError):
+            host._read_state_file_for_save(self.path)
+        with open(self.path, "w") as f:
+            json.dump({"tasks": {}, "automations": []}, f)
+        self.assertEqual(host._read_state_file_for_save(self.path),
+                         {"tasks": {}, "automations": []})
+
+    def test_load_state_strict(self):
+        data, err = host.load_state_strict(self.path)
+        self.assertIsNone(err)
+        with open(self.path, "w") as f:
+            f.write("{bad")
+        data, err = host.load_state_strict(self.path)
+        self.assertIsNone(data)
+        self.assertIsNotNone(err)
+
+
+class RestoreAndRecoverTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _write(self, path, obj):
+        with open(path, "w") as f:
+            json.dump(obj, f)
+
+    def test_restore_good_is_noop(self):
+        self._write(self.path, {"tasks": {}, "automations": [],
+                                "version": 5})
+        ok, msg = host.restore_state(self.path)
+        self.assertTrue(ok)
+        self.assertIn("完好", msg)
+
+    def test_restore_from_bak1_when_bak_bad(self):
+        with open(self.path, "w") as f:
+            f.write("{bad")
+        with open(self.path + ".bak", "w") as f:
+            f.write("{also bad")
+        self._write(self.path + ".bak.1",
+                    {"tasks": {}, "automations": [], "version": 7})
+        ok, msg = host.restore_state(self.path)
+        self.assertTrue(ok)
+        self.assertIn(".bak.1", msg)
+        self.assertEqual(host.load_state(self.path)["version"], 7)
+        leftovers = [n for n in os.listdir(self.tmp.name)
+                     if n.startswith("host-state.json.corrupt.")]
+        self.assertEqual(len(leftovers), 1)
+
+    def test_restore_no_backup(self):
+        with open(self.path, "w") as f:
+            f.write("{bad")
+        ok, msg = host.restore_state(self.path)
+        self.assertFalse(ok)
+        self.assertIn("没有可用的备份", msg)
+
+    def test_recover_marks_only_old_running(self):
+        now = time.time()
+        state = {"tasks": {
+            "old": {"name": "old", "status": "running",
+                    "created_at": now - 5000},
+            "fresh": {"name": "fresh", "status": "running",
+                      "last_turn_at": now - 10},
+            "done": {"name": "done", "status": "completed",
+                     "created_at": now - 5000}},
+            "automations": []}
+        got = host.recover_interrupted(state, self.path,
+                                       older_than=3600)
+        self.assertEqual(got, ["old"])
+        self.assertEqual(state["tasks"]["old"]["status"],
+                         "interrupted")
+        self.assertEqual(state["tasks"]["fresh"]["status"],
+                         "running")
+        self.assertEqual(state["tasks"]["done"]["status"],
+                         "completed")
+
+
+class GitHelpersTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.repo = _make_repo(self.tmp.name)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_task_diff_tracked_and_untracked(self):
+        with open(os.path.join(self.repo, "a.py"), "a") as f:
+            f.write("y = 2\n")
+        with open(os.path.join(self.repo, "new.txt"), "w") as f:
+            f.write("hello\n")
+        os.makedirs(os.path.join(self.repo, "__pycache__"),
+                    exist_ok=True)
+        with open(os.path.join(self.repo, "__pycache__",
+                               "x.pyc"), "wb") as f:
+            f.write(b"\x00\x01")
+        out = host.task_diff(self.repo)
+        self.assertIn("y = 2", out["diff"])
+        self.assertIn("new.txt", out["untracked"])
+        self.assertEqual(out["untracked_contents"]["new.txt"],
+                         "hello\n")
+        self.assertNotIn("__pycache__/x.pyc",
+                         out["untracked_contents"])
+
+    def test_task_diff_non_git_dir(self):
+        plain = os.path.join(self.tmp.name, "plain")
+        os.makedirs(plain)
+        with open(os.path.join(plain, "f.txt"), "w") as f:
+            f.write("x")
+        out = host.task_diff(plain)
+        self.assertEqual(out["diff"], "")
+        self.assertIn("f.txt", out["untracked"])
+
+    def test_topo_sort_and_errors(self):
+        jobs = [{"name": "b", "depends_on": ["a"], "priority": 1},
+                {"name": "a", "priority": 5},
+                {"name": "c", "depends_on": ["a"], "priority": 9}]
+        ordered = [j["name"] for j in host.topo_sort_jobs(jobs)]
+        self.assertEqual(ordered, ["a", "c", "b"])
+        with self.assertRaises(ValueError):
+            host.topo_sort_jobs([{"name": "a",
+                                  "depends_on": ["ghost"]}])
+        with self.assertRaises(ValueError):
+            host.topo_sort_jobs([{"name": "a",
+                                  "depends_on": ["a"]}])
+        with self.assertRaises(ValueError):
+            host.topo_sort_jobs([{"name": "a",
+                                  "depends_on": ["b"]},
+                                 {"name": "b",
+                                  "depends_on": ["a"]}])
+
+    def test_detect_verify_and_run_verify(self):
+        self.assertIsNone(host.detect_verify_command(self.repo))
+        os.makedirs(os.path.join(self.repo, "tests"))
+        self.assertIn("unittest",
+                      host.detect_verify_command(self.repo))
+        self.assertIsNone(host.detect_verify_command(None))
+        self.assertEqual(host.run_verify("", self.repo), (0, ""))
+        rc, out = host.run_verify("echo hi", self.repo)
+        self.assertEqual((rc, out), (0, "hi"))
+        rc, _ = host.run_verify("false", self.repo)
+        self.assertNotEqual(rc, 0)
+
+    def test_merge_task_success_and_verify_fail(self):
+        wt = os.path.join(self.tmp.name, "wt1")
+        _git(self.repo, "worktree", "add", wt, "-b", "host/t1")
+        with open(os.path.join(wt, "a.py"), "a") as f:
+            f.write("z = 3\n")
+        ok, msg = host.merge_task(self.repo, wt, "host/t1", "t1",
+                                  verify_command="true")
+        self.assertTrue(ok, msg)
+        self.assertFalse(os.path.exists(wt))
+        self.assertNotIn("host/t1",
+                         _git(self.repo, "branch").stdout)
+        with open(os.path.join(self.repo, "a.py")) as f:
+            self.assertIn("z = 3", f.read())
+        # 验测失败：回滚主仓、分支与 worktree 保留。
+        wt2 = os.path.join(self.tmp.name, "wt2")
+        _git(self.repo, "worktree", "add", wt2, "-b", "host/t2")
+        with open(os.path.join(wt2, "a.py"), "a") as f:
+            f.write("w = 4\n")
+        ok, msg = host.merge_task(self.repo, wt2, "host/t2", "t2",
+                                  verify_command="false")
+        self.assertFalse(ok)
+        self.assertIn("验测失败", msg)
+        with open(os.path.join(self.repo, "a.py")) as f:
+            self.assertNotIn("w = 4", f.read())
+        self.assertIn("host/t2", _git(self.repo, "branch").stdout)
+        _git(self.repo, "worktree", "remove", "--force", wt2)
+        _git(self.repo, "branch", "-D", "host/t2")
+
+
+class BudgetAndNotifyTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = os.path.join(self.tmp.name, "host-state.json")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_record_usage_and_budget(self):
+        state = {}
+        u = host.record_turn_usage(state)
+        self.assertEqual(u["turns"], 1)
+        u = host.record_turn_usage(state, failed=True)
+        self.assertEqual((u["turns"], u["failed_turns"]), (2, 1))
+        # 跨天重置。
+        state["usage"] = {"date": "2000-01-01", "turns": 9,
+                          "failed_turns": 9}
+        u = host.record_turn_usage(state)
+        self.assertEqual((u["turns"], u["failed_turns"]), (1, 0))
+        today = time.strftime("%Y-%m-%d", time.localtime())
+        self.assertFalse(host.budget_exceeded(
+            {"usage": {"date": today, "turns": 1}}, {}))
+        self.assertFalse(host.budget_exceeded(
+            {"usage": {"date": today, "turns": 99}},
+            {"max_daily_turns": "bogus"}))
+        self.assertFalse(host.budget_exceeded(
+            {"usage": {"date": "2000-01-01", "turns": 99}},
+            {"max_daily_turns": 5}))
+        self.assertTrue(host.budget_exceeded(
+            {"usage": {"date": today, "turns": 5}},
+            {"max_daily_turns": 5}))
+
+    def test_notify_failure_log_only_and_command(self):
+        msg = host.notify_failure(self.path, {}, "t1",
+                                  {"status": "failed", "text": "boom"})
+        self.assertIn("FAILED t1", msg)
+        log_path = os.path.join(self.tmp.name, "host-run.log")
+        self.assertTrue(os.path.exists(log_path))
+        out_file = os.path.join(self.tmp.name, "notified.txt")
+        cfg = {"notify_command":
+               "echo {task} {status} > " + out_file}
+        host.notify_failure(self.path, cfg, "t1",
+                            {"status": "failed", "text": "boom"})
+        with open(out_file) as f:
+            self.assertIn("t1", f.read())
+        # 坏模板：记 template error，不跑命令。
+        before = open(log_path).read()
+        host.notify_failure(self.path, {"notify_command": "echo {nope}"},
+                            "t1", {"status": "failed", "text": "x"})
+        after = open(log_path).read()
+        self.assertIn("template error", after[len(before):])
+
+    def test_build_fix_prompt_and_pr_body(self):
+        task = {"name": "t1", "prompt": "do it",
+                "review_comments": [
+                    {"file": "a.py", "line": 3, "comment": "fix",
+                     "status": "open"},
+                    {"file": "b.py", "line": 1, "comment": "old",
+                     "status": "resolved"}]}
+        prompt = host.build_fix_prompt(task)
+        self.assertIn("a.py:3: fix", prompt)
+        self.assertNotIn("old", prompt)
+        body = host.build_pr_body(task)
+        self.assertIn("host 任务：t1", body)
+        self.assertIn("a.py:3: fix", body)
+        body = host.build_pr_body({"name": "t2"})
+        self.assertIn("（无提示）", body)
+        self.assertIn("无未解决评论", body)
+
+    def test_check_state(self):
+        state = {"version": 5, "tasks": {
+            "bad": {"status": "failed", "cwd": "/nonexistent-x"},
+            "ok": {"status": "completed"},
+            "gone": {"status": "running", "cwd": "/nonexistent-x",
+                     "review_comments": [
+                         {"status": "open", "file": "a.py",
+                          "line": 1, "comment": "c"}]}},
+            "automations": []}
+        out = host.check_state(self.path, state)
+        self.assertEqual(out["version"], 5)
+        self.assertFalse(out["backup_exists"])
+        self.assertEqual(out["failed_tasks"], ["bad"])
+        self.assertEqual(sorted(out["missing_worktrees"]),
+                         ["bad", "gone"])
+        self.assertEqual(out["open_review_comments"], 1)
+
+
+class ParsePrChecksTest(unittest.TestCase):
+    def test_tab_and_human_formats(self):
+        checks, all_pass = host.parse_pr_checks(
+            "name\tstate\nbuild\tpass\nlint\tfail\n")
+        self.assertEqual(checks, [{"name": "build", "state": "pass"},
+                                  {"name": "lint", "state": "fail"}])
+        self.assertFalse(all_pass)
+        checks, all_pass = host.parse_pr_checks(
+            "✓ build pass\n✓ lint passing\n")
+        self.assertTrue(all_pass)
+        self.assertEqual([c["state"] for c in checks],
+                         ["pass", "pass"])
+        checks, all_pass = host.parse_pr_checks(
+            "no checks reported\n")
+        self.assertEqual((checks, all_pass), ([], False))
+        checks, all_pass = host.parse_pr_checks("")
+        self.assertEqual((checks, all_pass), ([], False))
+        checks, _ = host.parse_pr_checks("build queued\ntest skipped\n"
+                                         "job cancelled\nx neutral\n"
+                                         "y in_progress\nz failing\n")
+        self.assertEqual([c["state"] for c in checks],
+                         ["pending", "skipped", "cancelled",
+                          "neutral", "pending", "fail"])
+
+
+class CliHelpersTest(unittest.TestCase):
+    def test_load_config_and_effective_cfg(self):
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            self.assertEqual(host._load_config_file(None), {})
+            self.assertEqual(host._load_config_file(
+                os.path.join(tmp.name, "nope.json")), {})
+            bad = os.path.join(tmp.name, "bad.json")
+            with open(bad, "w") as f:
+                f.write("{bad")
+            with self.assertRaises(RuntimeError):
+                host._load_config_file(bad)
+            cfg = {"max_daily_turns": 5, "other": 1}
+            self.assertIs(host._effective_cfg(cfg, False), cfg)
+            out = host._effective_cfg(cfg, True)
+            self.assertIsNone(out["max_daily_turns"])
+            self.assertEqual(out["other"], 1)
+            self.assertEqual(cfg["max_daily_turns"], 5)
+        finally:
+            tmp.cleanup()
+
+    def test_resolve_path_and_thread_result(self):
+        self.assertEqual(host._resolve_path("/base", ""), "")
+        self.assertEqual(host._resolve_path("/base", "/abs"), "/abs")
+        self.assertEqual(host._resolve_path("/base", "rel"),
+                         "/base/rel")
+        self.assertEqual(host._thread_cli_result({"result": {"a": 1}},
+                                                 "thread-list"),
+                         {"a": 1})
+        self.assertEqual(host._thread_cli_result({"result": None},
+                                                 "thread-list"), {})
+        with self.assertRaises(RuntimeError):
+            host._thread_cli_result({"error": "boom"}, "thread-list")
+        with self.assertRaises(RuntimeError):
+            host._thread_cli_result(None, "thread-list")
+
+    def test_server_command(self):
+        cmd = host._server_command("/bin/codex")
+        self.assertEqual(cmd[-1], "app-server")
+        self.assertIn("-c", cmd)
+        with self.assertRaises(ValueError):
+            host._server_command("/bin/codex", provider="other")
+        with mock.patch.dict(os.environ, {"SUB2API_BASE_URL": ""}):
+            with self.assertRaises(RuntimeError):
+                host._server_command("/bin/codex",
+                                     provider="sub2api")
+        with mock.patch.dict(os.environ,
+                             {"SUB2API_BASE_URL":
+                              "https://gw.example.com/"}):
+            cmd = host._server_command("/bin/codex",
+                                       provider="sub2api")
+            self.assertTrue(any("gw.example.com/v1" in c
+                                for c in cmd))
+
+
+class ServerTraceCloseTest(unittest.TestCase):
+    def test_append_trace_and_close(self):
+        tmp = tempfile.TemporaryDirectory()
+        try:
+            srv = host.Server(codex_bin="/bin/false")
+            state_path = os.path.join(tmp.name, "host-state.json")
+            srv.set_trace_file(state_path, "t1")
+            self.assertIsNotNone(srv.trace_path)
+            srv._append_trace({"method": "x", "params": {
+                "item": {"type": "agentMessage", "id": "m1",
+                         "text": "hello"}}})
+            srv._append_trace({"method": "x", "params": {
+                "item": {"type": "userMessage", "text": "skip"}}})
+            with open(srv.trace_path) as f:
+                lines = f.readlines()
+            self.assertEqual(len(lines), 1)
+            self.assertIn("hello", lines[0])
+            # trace_path 为 None 时 _append_trace 是空操作。
+            srv.trace_path = None
+            srv._append_trace({"method": "x", "params": {
+                "item": {"type": "agentMessage", "id": "m2",
+                         "text": "nope"}}})
+        finally:
+            tmp.cleanup()
+
+    def test_close_with_fake_proc(self):
+        class _Stdin:
+            def __init__(self):
+                self.closed = False
+
+            def close(self):
+                self.closed = True
+
+        class _Proc:
+            def __init__(self, wait_raises=False):
+                self.stdin = _Stdin()
+                self.stdout = _Stdin()
+                self.wait_raises = wait_raises
+                self.killed = False
+
+            def wait(self, timeout=None):
+                if self.wait_raises:
+                    raise RuntimeError("not dead")
+                return 0
+
+            def kill(self):
+                self.killed = True
+
+        srv = host.Server(codex_bin="/bin/false")
+        proc = _Proc()
+        srv.proc = proc
+        srv.close()
+        self.assertTrue(proc.stdin.closed)
+        self.assertFalse(proc.killed)
+        proc2 = _Proc(wait_raises=True)
+        srv.proc = proc2
+        srv.close()
+        self.assertTrue(proc2.killed)
+        self.assertTrue(proc2.stdout.closed)
+
+
+if __name__ == "__main__":
+    unittest.main()