From 1e9c1b36eea1ee1b29a7242b5f75529bdb7c0548 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 08:25:00 +0000 Subject: [PATCH] =?UTF-8?q?coverage85-step2c:=20host=20=E5=89=A9=E4=BD=99?= =?UTF-8?q?=E7=BC=BA=E5=8F=A3=E7=9C=9F=E5=AE=9E=E8=A1=8C=E4=B8=BA=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=EF=BC=88+18=20=E9=A1=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 793 ++++++++++++++++++++++++++++++++ 1 file changed, 793 insertions(+) create mode 100644 tests/test_host_coverage85_2.py diff --git a/tests/test_host_coverage85_2.py b/tests/test_host_coverage85_2.py new file mode 100644 index 0000000..cb00e3c --- /dev/null +++ b/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()