diff --git a/README.md b/README.md index 0bbbd90..f043c10 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,15 @@ turn 可以真正并发（不是轮流排队）。 | 跨进程 state 竞态 | host-state.json 单文件，多 host.py 进程同时写会丢任务，同一进程内并行安全 | 已修：锁文件 + 保存时合并，实测双进程并发保存不丢 | | 项目配置 | 每次都要传 model/repo/state 等参数 | 已修：host-config.json + init/config，CLI 优先于配置，密钥只走环境变量 | | 定时自动化 | run-due 要手动跑，跑完没日志 | 已修：host-run.log 记录派发/跳过，automations --cron-line 生成 crontab 行，--install-cron 才装机 | +| 合并后自动验测 + 回滚 | merge 合完不验主仓测试，合坏了靠人发现 | 已修：merge 默认仓库有 tests/ 就跑 unittest，验测失败 reset 回合并前 HEAD，分支/worktree 保留；--no-verify 跳过，--verify 可指定命令 | +| 任务依赖 + 优先级 | parallel 是扁平的，没有依赖和优先级 | 已修：jobs 里 depends_on/priority，拓扑排序 + 按波次跑，依赖失败时依赖任务标 skipped 不跑；依赖任务从依赖分支建 worktree（有血缘关系），不是文件拷贝 | +| 失败自动重试 | 失败任务就停在 failed，不重试 | 已修：dispatch/parallel 里 --max-retries，失败后同 prompt 再跑，重试前查预算；默认 0，行为不变 | +| 评审评论回流 | 评审只看 diff，修要另起 run prompt，评论没地方钉 | 已修：review-comment --file --line --comment 把评论存进任务，review 时列出 open 评论，fix-review 按 open 评论逐条修，修完标 resolved | +| server 崩溃重连 | app-server 死了，host 就 BrokenPipe 失败 | 已修：Server.ensure_alive，死了重拉并 resume thread（thread 在 CODEX_HOME 里不丢）；request 遇 BrokenPipe 重试一次；单测 kill 进程后重拉 + list 能通 | +| 成本预算 | 没有预算，模型跑多少全靠自觉 | 已修：配置 max_daily_turns，按 turn 数算（协议不暴露真实 token 数，这是诚实边界），超预算不派发/不重试；usage 记在 state 里，doctor 可看 | +| 失败通知 | 失败只有 state 里的 status，跑完没人知道 | 已修：失败时写 host-run.log + 跑配置的 notify_command（本地 shell，模板 {task} {status}）；failures 命令列失败任务，doctor 列失败/缺失 worktree/open 评论 | +| 多机器 state 冲突 | 锁只防同一台机器，多机器共享 state 文件会静默覆盖 | 部分修：state 加 version/last_writer/updated_at，保存时磁盘版本比加载时新就记 STATE-CONFLICT 日志并合并保留双方任务，不静默丢；真多机同步仍需中心服务（见下“没对齐”） | +| 合并时提交缓存文件 | merge 的 git add -A 会把 __pycache__/*.pyc 也提交进主仓，下次合并被 .pyc 改动挡住 | 已修：merge 前清 worktree 的 __pycache__/*.pyc，主仓的 .pyc 改动先还原；单测覆盖 pyc 不入库 | | 边界：cwd、沙箱、审批、模型 | dispatch / turn 参数里定，见 `config.example.toml` | 已实测 | | 任务登记 | `host-state.json`：task -> thread/worktree/状态/diff；agent 状态本身仍在 CODEX_HOME，不另存第二份 | 已实测 | @@ -172,6 +181,15 @@ turn 可以真正并发（不是轮流排队）。 - 并行改同一文件：worktree 能防互相踩，但两个分支都在文件末尾 追加时，合并第二支仍可能冲突，host 会 abort 等人工解冲突， 不会自动猜合并结果——这和桌面版一样，是正确做法，不是 bug。 + 2026-10-26 补了一层安全网：merge 冲突时先试 -X ours，解完 + 后逐行核对任务分支的新增行是否都在主仓里，缺一行就回滚 + （gap2 实测拦下了一次 -X ours 丢 mul2 的情况）；依赖任务 + 现在从依赖分支建 worktree，依赖先合、依赖任务后合时不再 + 重复加同一段。 +- 多机器同步：state 现在有 version 和冲突日志，不会静默丢 + 任务，但两台机器共享一个 state 文件仍不是 ChatGPT 云端 + 那种中心同步——要真多机，得另起一个中心 state 服务， + 这不在本地 host 的范围里。 ## 开源没给到的（别指望照搬） diff --git a/TEST-REPORT.md b/TEST-REPORT.md index ffca0c5..99f426f 100644 --- a/TEST-REPORT.md +++ b/TEST-REPORT.md @@ -210,7 +210,6 @@ tests/ 标准库 unittest，验收命令就是 `python3 -m unittest discover -s 不常驻。 ## 真实项目一轮（2026-10-26 凌晨，用户选“先在真实项目上跑一轮再打包”） - - **选项目**：没动生产敏感的 wechat-bridge（带 session 凭证）， 选 codex-app-server 本身做 dogfooding——这是真实代码，不是 玩具仓。为此先把本项目初始化成 git 仓（commit b49b9cf、 @@ -330,3 +329,68 @@ tests/ 标准库 unittest，验收命令就是 `python3 -m unittest discover -s - 审批请求的完整回包字段：各版本间字段有差异，client.py 默认 用 never 绕开审批；on-request + auto/deny 的回包是尽力而为， 生产交互客户端应以本版 protocol schema 为准再对齐一次。 + +## 第二类缺口补齐（2026-10-26 凌晨，用户选“继续补齐第二类所有缺口”） + +- **合并后自动验测 + 回滚**：merge_task 抽成可单测函数， + 默认仓库有 tests/ 就跑 unittest，验测失败 reset 回合并前 + HEAD，分支/worktree 保留，任务标 merge-failed 并走失败通知。 + 单测覆盖验测通过清 worktree、验测失败回滚主仓 intact。 +- **任务依赖 + 优先级 + 重试**：jobs 里 depends_on/priority/ + max_retries，topo_sort_jobs 拓扑排序（缺依赖/成环抛错）， + parallel 按波次跑，依赖失败时依赖任务标 skipped 且落盘 + （修过一次 skipped 未落盘导致 discard KeyError 的 bug）。 + 依赖任务现在从依赖分支建 worktree（dispatch_from_branch： + 先把依赖 worktree 提交到依赖分支，再从该分支建新分支）， + 这样依赖分支是依赖任务分支的祖先，合并时不再重复加同一段。 + 重试：run_task_turn_with_retry，默认 0 次行为不变，配了 + max_retries 时失败后同 prompt 再跑，重试前查预算。 +- **评审评论回流**：review-comment --file --line --comment + 存进任务，review 时列出 open 评论，fix-review 按 open + 评论逐条修，修完标 resolved。模型实测：gap2-b 的 + add2_mul2 docstring 评论，第一次 fix-review 因我给的 + 300 秒 shell 超时被杀（turn 本身没完成，评论正确地保持 + open），第二次 900 秒超时下 51 秒完成并标 resolved。 +- **server 崩溃重连**：Server 抽出 _start/ensure_alive， + 进程死了重拉并 resume thread（thread 在 CODEX_HOME 里）， + request 遇 BrokenPipe 重试一次。单测 kill 进程后重拉 + + list 能通。 +- **成本预算 + 失败通知**：配置 max_daily_turns（按 turn 数， + 协议不暴露真实 token 数，这是诚实边界），超预算不派发/ + 不重试；失败时写 host-run.log + 跑 notify_command（本地 + shell，模板 {task} {status}）；failures 列失败任务，doctor + 列 version/备份/失败/缺失 worktree/open 评论。模型实测： + 不存在的模型 + max_retries=1，跑了 2 个 turn（retry_count=1， + 均为 404），status failed，notify 日志里有 gap3-fail failed。 +- **多机器 state 冲突**：state 加 version/last_writer/ + updated_at，保存时磁盘版本比加载时新就记 STATE-CONFLICT + 日志并合并保留双方任务。单测：磁盘版本被另一写者改到 5 + 后再保存，合并后版本 6，双方任务都在，日志里有冲突行。 + 诚实边界：这只是“检测 + 不丢”，不是 ChatGPT 云端那种 + 中心同步，两台机器共享一个 state 文件仍不推荐。 +- **模型实测 gap2/gap3（依赖 + 合并，教训最多的一轮）**： + 第一轮 gap（/tmp）因 VM 回收丢了 /tmp 里的 repo 和 state， + 但 home 下的 worktree 还在，证明依赖拷贝已生效（gap-b 里 + 有 gap-a 的 add2 且未重复定义）；第二轮 gap2（state 放 + home）先因 codex_home 新建后没有中转站配置 401 失败两次， + 靠失败通知和依赖 skipped 门正确地没浪费后续调用，修配置 + （model_provider + [model_providers.xcmapi] 两部分都要有， + 只给其一时要么走 api.openai.com 401，要么报 provider not found） + 后第三轮并行全通：gap2-a（优先级 5）先跑，gap2-b 依赖它且 + 工作区里有 add2，gap2-c 独立；评审评论 + fix-review 通； + 合并时暴露两个真问题：① 依赖任务文件拷贝法建的分支没有 + 血缘关系，合并依赖任务时在同一位置重复加 add2，-X ours + 自动解会把 mul2 整段丢掉——幸好新增行核对把它拦下来回滚 + 了（这是安全网的价值，不是 bug）；② merge 的 git add -A + 会把 __pycache__/*.pyc 提交进主仓，下次合并被 .pyc 改动 + 挡住。① 用“从依赖分支建 worktree”修了，② 用“合并前清 + worktree 的 __pycache__/*.pyc，主仓 .pyc 改动先还原”修了； + gap3 用新方法跑依赖对 + 合并全通（add2/mul2/add2_mul2， + 测试通过，值 3/6/8），gap2 的遗留分支用手动取目标文件内容 + 的方式解完（主仓最终 add2/mul2/add2_mul2/sub2 全有，测试通过）。 + 独立任务在同一文件同一位置追加时仍会冲突，这是 git 本身 + 的行为，host 正确的做法是 abort/回滚 + 人工解，不是自动猜。 +- **回归**：tests/test_host.py 从 3 个涨到 15 个，全部通过， + 包括 topo 排序、merge 验测通过/失败回滚/重复加自动解/ + pyc 不入库、评审评论、重试成功/预算拦截/通知落盘、 + state 版本冲突、server kill 重拉。 diff --git a/host.py b/host.py index 807339e..9976107 100644 --- a/host.py +++ b/host.py @@ -72,15 +72,8 @@ class Server: def __init__(self, codex_bin=None, codex_home=None, approval="deny"): self.codex_bin = codex_bin or default_codex_bin() - env = dict(os.environ) - if codex_home: - os.makedirs(codex_home, exist_ok=True) - env["CODEX_HOME"] = codex_home - self.proc = subprocess.Popen( - [self.codex_bin, "app-server"], - stdin=subprocess.PIPE, stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, env=env, - ) + self.codex_home = codex_home + self.approval = approval self.send_lock = threading.Lock() self.next_id = 1 self.pending = {} @@ -88,10 +81,39 @@ class Server: self.turn_queues = {} self.events = [] self.events_lock = threading.Lock() - self.approval = approval + self.restart_count = 0 + self._start() + + def _start(self): + env = dict(os.environ) + if self.codex_home: + os.makedirs(self.codex_home, exist_ok=True) + env["CODEX_HOME"] = self.codex_home + self.proc = subprocess.Popen( + [self.codex_bin, "app-server"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, env=env, + ) self.reader = threading.Thread(target=self._read_loop, daemon=True) self.reader.start() + def is_alive(self): + return self.proc is not None and self.proc.poll() is None + + def ensure_alive(self): + """server 崩了就重拉一次，thread 存在 CODEX_HOME 里， + 重拉后靠 resume_thread 恢复，不丢 thread。 + 返回 True 表示 alive（原来 alive 或重拉成功）。""" + if self.is_alive(): + return True + try: + self._start() + self.initialize() + self.restart_count += 1 + return True + except Exception: + return False + def _answer_server_request(self, msg): # MCP 工具审批不是通用 approval：服务端发的是 # mcpServer/elicitation/request（mode=form，requestedSchema）， @@ -146,18 +168,31 @@ class Server: self.proc.stdin.flush() def request(self, method, params=None, timeout=120): - rid = self.next_id - self.next_id += 1 - q = queue.Queue() - with self.pending_lock: - self.pending[rid] = q - try: - self._send({"jsonrpc": "2.0", "id": rid, "method": method, - "params": params or {}}) - return q.get(timeout=timeout) - finally: + # BrokenPipe / 进程已死时重拉一次再重试（只重试一次， + # turn/start 这类有副作用的调用由上层 resume 后再跑， + # 这里重试的是 initialize/list/resume 等幂等请求为主； + # 即使是 turn/start，重试也只发生在“还没收到任何响应” + # 的 BrokenPipe 场景，server 死在 turn 中途时 _read_loop + # 已退出、q.get 会超时而不是走这条路）。 + for attempt in range(2): + rid = self.next_id + self.next_id += 1 + q = queue.Queue() with self.pending_lock: - self.pending.pop(rid, None) + self.pending[rid] = q + try: + self._send({"jsonrpc": "2.0", "id": rid, "method": method, + "params": params or {}}) + return q.get(timeout=timeout) + except (BrokenPipeError, OSError): + with self.pending_lock: + self.pending.pop(rid, None) + if attempt == 0 and self.ensure_alive(): + continue + raise + finally: + with self.pending_lock: + self.pending.pop(rid, None) def notify(self, method, params=None): self._send({"jsonrpc": "2.0", "method": method, "params": params or {}}) @@ -221,6 +256,15 @@ class Server: def run_turn(self, thread_id, prompt, model=None, effort=None, parent_turn_id=None, root_turn_id=None, timeout=600, sandbox=None): + # 开 turn 前先看 server 是否还活着，死了就重拉并 resume + # 这个 thread（thread 持久化在 CODEX_HOME 里，重拉不丢）， + # 否则 turn/start 会直接 BrokenPipe 失败。 + if not self.is_alive(): + if self.ensure_alive(): + try: + self.resume_thread(thread_id) + except Exception: + pass q = queue.Queue() self.turn_queues[thread_id] = q start_events = len(self.events) @@ -287,6 +331,10 @@ class Server: self.proc.wait(timeout=5) except Exception: self.proc.kill() + try: + self.proc.stdout.close() + except Exception: + pass # ---------------- 宿主状态 / worktree / diff ---------------- @@ -335,16 +383,36 @@ def _read_state_file(path): def load_state(path): with _state_lock(path, exclusive=False): - return _read_state_file(path) + st = _read_state_file(path) + # _loaded_version 只活在内存里，save 时不写回文件（见下）， + # 用来判断“保存时磁盘版本是否比加载时新”——多机器共享 + # state 时，版本跳变就是冲突信号，不静默覆盖。 + st["_loaded_version"] = st.get("version", 0) + return st def save_state(path, state): with _state_lock(path, exclusive=True): # 锁内重读 + 合并：内存任务覆盖同名磁盘任务，磁盘上 - # 别的进程新增的任务保留；自动化按 name 合并，内存优先。 + # 别的进程新增的任务保留；自动化按名合并，内存优先。 disk = _read_state_file(path) + loaded_version = state.get("_loaded_version", disk.get("version", 0)) + if disk.get("version", 0) > loaded_version: + # 别的写者（另一进程 / 另一台机器共享文件时）已先写过， + # 合并保留了它的任务，但这是一次冲突，记日志不静默。 + try: + log_path = os.path.join( + os.path.dirname(os.path.abspath(path)), "host-run.log") + ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + with open(log_path, "a", encoding="utf-8") as lf: + lf.write(f"{ts} STATE-CONFLICT disk_version=" + f"{disk.get('version')} loaded_version=" + f"{loaded_version} merged_tasks=" + f"{len(disk.get('tasks') or {})}\n") + except Exception: + pass merged_tasks = dict(disk.get("tasks") or {}) - merged_tasks.update(state.get("tasks") or {}) + merged_tasks.update({k: v for k, v in (state.get("tasks") or {}).items()}) disk_autos = {a.get("name"): a for a in (disk.get("automations") or []) if isinstance(a, dict)} for a in (state.get("automations") or []): @@ -352,10 +420,18 @@ def save_state(path, state): disk_autos[a["name"]] = a merged = {"tasks": merged_tasks, "automations": list(disk_autos.values())} - # 保留内存 state 里 tasks/automations 之外的顶层字段 + # 保留内存 state 里 tasks/automations 之外的顶层字段， + # 但 _ 开头的内存标记不写回文件。 for k, v in state.items(): - if k not in ("tasks", "automations"): + if k not in ("tasks", "automations") and not k.startswith("_"): merged[k] = v + merged["version"] = (disk.get("version", 0) or 0) + 1 + try: + import socket + merged["last_writer"] = f"{socket.gethostname()}:{os.getpid()}" + except Exception: + merged["last_writer"] = str(os.getpid()) + merged["updated_at"] = time.time() tmp = f"{path}.tmp.{os.getpid()}" with open(tmp, "w") as f: json.dump(merged, f, indent=2, ensure_ascii=False) @@ -378,6 +454,7 @@ def save_state(path, state): # 又把别的进程的任务挤掉（调用方继续用这份 state 时）。 state.clear() state.update(merged) + state["_loaded_version"] = merged["version"] def log_event(state_path, message): @@ -448,8 +525,293 @@ def task_diff(worktree): "untracked": "\n".join(sorted(os.listdir(worktree)))} +# ---------------- 第二类缺口补齐：依赖 / 验测 / 评审评论 / 预算 ---------------- + +def topo_sort_jobs(jobs): + """依赖拓扑 + 优先级：依赖先跑，同层按 priority 降序、名字升序。 + 缺依赖或成环时抛 ValueError，不静默跳过。""" + by_name = {} + for j in jobs: + by_name[j["name"]] = j + for j in jobs: + for d in (j.get("depends_on") or []): + if d not in by_name: + raise ValueError(f"任务 {j['name']} 依赖不存在的任务 {d}") + if d == j["name"]: + raise ValueError(f"任务 {j['name']} 不能依赖自己") + ordered = [] + done = set() + remaining = list(jobs) + while remaining: + ready = [j for j in remaining + if all(d in done for d in (j.get("depends_on") or []))] + if not ready: + names = ", ".join(j["name"] for j in remaining) + raise ValueError(f"依赖成环，无法排序：{names}") + ready.sort(key=lambda j: (-(j.get("priority") or 0), j["name"])) + # 一波只取优先级最高的那批？不，一波取全部 ready， + # 这样同层可以并行，优先级只决定同层内的启动顺序。 + # 但为了“高优先级先完成”可观测，返回顺序即启动顺序。 + for j in ready: + ordered.append(j) + done.add(j["name"]) + remaining = [j for j in remaining if j["name"] not in done] + return ordered + + +def detect_verify_command(repo): + """合并后验测的默认命令：仓库有 tests/ 就跑标准库 unittest， + 没有就返回 None（不猜别的测试框架，不强制跑）。""" + if repo and os.path.isdir(os.path.join(repo, "tests")): + return "python3 -m unittest discover -s tests -v" + return None + + +def run_verify(command, cwd): + if not command: + return 0, "" + v = subprocess.run(command, shell=True, cwd=cwd, + capture_output=True, text=True, timeout=600) + return v.returncode, (v.stdout + v.stderr).strip()[:2000] + + +def merge_task(repo, worktree, branch, task_name, verify_command=None): + """合并 + 验测 + 回滚（纯 git 操作，不依赖 server，可单测）。 + 返回 (ok, message)：验测失败时 reset 回合并前 HEAD，分支和 + worktree 保留给修复；验测通过时才删 worktree/分支。""" + if verify_command is None: + verify_command = detect_verify_command(repo) + pre_head = git(["rev-parse", "HEAD"], repo).stdout.strip() + # 合并前先清掉 worktree 里的 __pycache__ / *.pyc：agent 跑 + # 测试时会生成它们，git add -A 会把二进制缓存也提交进主仓， + # 下次合并时主仓的 .pyc 被改动就会挡住 merge（gap3 实测）。 + # .pyc 是缓存，删了可再生，安全。 + subprocess.run(["find", worktree, "-type", "d", "-name", "__pycache__", + "-exec", "rm", "-rf", "{}", "+"], + capture_output=True) + subprocess.run(["find", worktree, "-name", "*.pyc", "-delete"], + capture_output=True) + # 主仓里已入库的 .pyc 若被测试改动，也先还原，否则 merge + # 会报 local changes would be overwritten。 + git(["checkout", "--", "*.pyc"], repo) + git(["add", "-A"], worktree) + c = git(["commit", "-m", f"host: {task_name}"], worktree) + if c.returncode != 0 and "nothing to commit" not in c.stdout + c.stderr: + return False, f"worktree 提交失败：{c.stderr}" + r = git(["merge", "--no-ff", branch, "-m", f"host: {task_name}"], repo) + if r.returncode != 0: + git(["merge", "--abort"], repo) + # 依赖任务的常见冲突：依赖任务的分支里已经包含依赖的改动， + # 依赖先合并后，再合并依赖任务会在同一位置重复加同一段 + # 代码（gap2 实测：add2 在两边都出现）。这种“重复加同一段” + # 可以用 -X ours 自动解：主仓版本优先，任务的新增部分 + # （mul2、add2_mul2 等）仍会合进来；解完后必须逐行核对 + # 任务分支的新增行是否都在主仓里，缺一行就回滚，不静默丢改动。 + r2 = git(["merge", "--no-ff", "-X", "ours", branch, + "-m", f"host: {task_name}"], repo) + if r2.returncode != 0: + git(["merge", "--abort"], repo) + return False, f"合并冲突，已 abort：{r.stderr}" + # 核对新增行：以 merge-base 到分支的 diff 为准 + base = git(["merge-base", pre_head, branch], repo).stdout.strip() + diff = git(["diff", base, branch], repo).stdout + added = [ln[1:].strip() for ln in diff.splitlines() + if ln.startswith("+") and not ln.startswith("+++") + and ln[1:].strip()] + # 合并后的工作区文件内容（主仓最新） + missing = [] + for ln in added: + found = False + for root, dirs, files in os.walk(repo): + dirs[:] = [d for d in dirs if d not in (".git", "__pycache__")] + for fn in files: + if fn.endswith(".pyc"): + continue + try: + with open(os.path.join(root, fn), encoding="utf-8", + errors="ignore") as f: + if ln in f.read(): + found = True + break + except Exception: + continue + if found: + break + if not found: + missing.append(ln) + if missing: + git(["reset", "--hard", pre_head], repo) + return False, (f"合并冲突自动解后发现任务改动被丢弃 " + f"（缺 {len(missing)} 行，如 {missing[:3]}），" + f"已回滚到 {pre_head[:8]}，需手动解冲突。") + # 自动解成功，继续走下面的验测（不 return，落到验测分支） + print(f"注意：合并冲突用 -X ours 自动解（重复加同一段），" + f"新增行 {len(added)} 行已核对都在主仓里。") + if verify_command: + rc, out = run_verify(verify_command, repo) + if rc != 0: + git(["reset", "--hard", pre_head], repo) + return False, (f"合并后验测失败，已回滚到 {pre_head[:8]}，" + f"分支 {branch} 和 worktree 保留：\n{out[:1000]}") + git(["worktree", "remove", "--force", worktree], repo) + git(["branch", "-d", branch], repo) + return True, f"merged {branch} into {repo}" + + +def add_review_comment(task, file_path, line, comment): + task.setdefault("review_comments", []) + task["review_comments"].append({ + "file": file_path, "line": int(line), "comment": comment, + "status": "open", "created_at": time.time()}) + return task["review_comments"][-1] + + +def open_review_comments(task): + return [c for c in (task.get("review_comments") or []) + if c.get("status") == "open"] + + +def build_fix_prompt(task): + comments = open_review_comments(task) + lines = ["请按以下评审评论逐条修复，只改评论涉及的文件和行附近，" + "不要改无关文件，修完后简述每条评论的处理结果："] + for c in comments: + lines.append(f"- {c['file']}:{c['line']}: {c['comment']}") + return "\n".join(lines) + + +def record_turn_usage(state, failed=False): + today = time.strftime("%Y-%m-%d", time.localtime()) + usage = state.get("usage") or {} + if usage.get("date") != today: + usage = {"date": today, "turns": 0, "failed_turns": 0} + usage["turns"] = usage.get("turns", 0) + 1 + if failed: + usage["failed_turns"] = usage.get("failed_turns", 0) + 1 + state["usage"] = usage + return usage + + +def budget_exceeded(state, cfg): + """成本预算：app-server 协议不暴露真实 token 用量， + 所以预算按 turn 数算（不是 token 数），这是诚实边界。 + cfg 里 max_daily_turns 未配时不限流。""" + limit = None + try: + limit = int(cfg.get("max_daily_turns")) if cfg.get("max_daily_turns") else None + except Exception: + limit = None + if not limit: + return False + usage = state.get("usage") or {} + today = time.strftime("%Y-%m-%d", time.localtime()) + if usage.get("date") != today: + return False + return usage.get("turns", 0) >= limit + + +def notify_failure(state_path, cfg, task_name, result): + """失败通知：先写 host-run.log（一定有），再跑用户配置的 + notify_command（本地 shell 命令，只来自用户自己的配置， + 不是外部输入；模板里可用 {task} {status}）。""" + text = (result or {}).get("text", "") or "" + msg = f"FAILED {task_name}: {result.get('status')} {text[:200]}" + try: + log_event(state_path, msg) + except Exception: + pass + cmd = (cfg or {}).get("notify_command") + if cmd: + try: + subprocess.run(cmd.format(task=task_name, + status=result.get("status", ""), + text=text[:200]), + shell=True, timeout=30) + except Exception: + pass + return msg + + +def check_state(state_path, state): + """doctor：state 版本、备份、失败任务、缺失 worktree。 + 不跑模型，不改 state。""" + out = {"state_path": state_path, + "version": state.get("version"), + "last_writer": state.get("last_writer"), + "updated_at": state.get("updated_at"), + "usage": state.get("usage"), + "backup_exists": os.path.exists(state_path + ".bak"), + "failed_tasks": [n for n, t in (state.get("tasks") or {}).items() + if t.get("status") in ("failed", "goal-unmet", + "merge-failed")], + "missing_worktrees": [n for n, t in (state.get("tasks") or {}).items() + if t.get("cwd") and not os.path.isdir(t["cwd"]) + and t.get("status") not in ("merged", "discarded", "archived")], + "open_review_comments": sum( + len(open_review_comments(t)) + for t in (state.get("tasks") or {}).values())} + return out + + + +def copy_tree_excluding_git(src, dst): + """把 src 工作区的文件拷到 dst（不含 .git、__pycache__、.pyc）， + fork 和依赖任务都用它：依赖任务要站在依赖任务的改动上， + 而不是只站在主仓 HEAD 上，否则依赖只是“等一等”，代码上 + 根本看不到依赖的改动。""" + import shutil + for root, dirs, files in os.walk(src): + dirs[:] = [d for d in dirs if d not in (".git", "__pycache__")] + for fn in files: + if fn.endswith(".pyc") or fn == ".git": + continue + s = os.path.join(root, fn) + rel = os.path.relpath(s, src) + d = os.path.join(dst, rel) + os.makedirs(os.path.dirname(d), exist_ok=True) + shutil.copy2(s, d) + + +def dispatch_from_branch(server, state, name, prompt, repo, dep_branch, + dep_worktree, model, effort, sandbox, + approval_policy, depends_on=None, priority=0, + max_retries=0): + """依赖任务的正确派发：先把依赖任务 worktree 的改动提交到 + 它的分支，再从该分支建新 worktree。这样依赖任务的分支是 + 依赖分支的后代，合并时先合依赖、再合依赖任务就不会出现 + “两边在同一位置重复加同一段代码”的冲突（gap2 实测： + 文件拷贝法建的分支没有这个血缘关系，合并时必然冲突， + -X ours 还会把依赖任务独有的改动整段丢掉，幸好验测 + 核对把它拦下来回滚了）。""" + repo = os.path.abspath(repo) + git(["add", "-A"], dep_worktree) + c = git(["commit", "-m", f"host: {name} base from {dep_branch}"], + dep_worktree) + # nothing to commit 时分支里已有全部改动，继续用它当基底 + base = os.path.join(HERE, "work", "host-worktrees") + os.makedirs(base, exist_ok=True) + path = os.path.join(base, name) + if os.path.exists(path): + return None, "worktree 已存在" + branch = f"host/{name}" + r = git(["worktree", "add", path, "-b", branch, dep_branch], repo) + if r.returncode != 0: + return None, f"git worktree from {dep_branch} 失败: {r.stderr}" + thread_id = server.create_thread(path, model=model, sandbox=sandbox, + approval_policy=approval_policy) + task = {"name": name, "thread_id": thread_id, "cwd": path, + "repo": repo, "branch": branch, "sandbox": sandbox, + "status": "running", "prompt": prompt, "parent_task": None, + "depends_on": list(depends_on or []), "priority": int(priority or 0), + "max_retries": int(max_retries or 0), "retry_count": 0, + "created_at": time.time(), "turns": []} + state["tasks"][name] = task + return task, "" + + def dispatch(server, state, name, prompt, repo, model, effort, sandbox, - approval_policy, use_worktree=True, parent_task=None): + approval_policy, use_worktree=True, parent_task=None, + depends_on=None, priority=0, max_retries=0): if name in state["tasks"]: raise RuntimeError(f"task 已存在: {name}，先换名或 discard 它") if repo and not os.path.isabs(repo) and not os.path.exists(repo): @@ -468,6 +830,8 @@ def dispatch(server, state, name, prompt, repo, model, effort, sandbox, "repo": os.path.abspath(repo) if repo else None, "branch": branch, "sandbox": sandbox, "status": "running", "prompt": prompt, "parent_task": parent_task, + "depends_on": list(depends_on or []), "priority": int(priority or 0), + "max_retries": int(max_retries or 0), "retry_count": 0, "created_at": time.time(), "turns": []} state["tasks"][name] = task return task @@ -481,6 +845,7 @@ def run_task_turn(server, state, task, prompt, model, effort, use_sandbox = task.get("sandbox") or sandbox result = server.run_turn(task["thread_id"], prompt, model=model, effort=effort, sandbox=use_sandbox) + record_turn_usage(state, failed=(result["status"] != "completed")) task["turns"].append({"prompt": prompt, "result": result}) task["status"] = "completed" if result["status"] == "completed" \ else "failed" @@ -488,6 +853,34 @@ def run_task_turn(server, state, task, prompt, model, effort, return result +def run_task_turn_with_retry(server, state, task, prompt, model, effort, + sandbox=None, cfg=None): + """失败自动重试：默认 max_retries=0 时行为不变（只跑一次）； + 配了重试时，失败后同 prompt 再跑，最多 max_retries 次， + 每次重试前先查预算，超预算直接停，不烧钱。""" + max_retries = int(task.get("max_retries") or 0) + result = None + for attempt in range(max_retries + 1): + if attempt > 0: + task["retry_count"] = attempt + prompt_try = (f"{prompt}\n\n（第 {attempt} 次重试：" + f"上一轮状态 {result.get('status')}，" + f"请针对失败原因重试，不要重复已完成的部分。）") + else: + prompt_try = prompt + if cfg is not None and budget_exceeded(state, cfg): + result = {"status": "failed", "text": "", + "error": "budget exceeded: max_daily_turns"} + task["status"] = "failed" + task["last_text"] = "" + break + result = run_task_turn(server, state, task, prompt_try, + model, effort, sandbox) + if result["status"] == "completed": + break + return result + + def _load_config_file(path): if path and os.path.exists(path): try: @@ -528,11 +921,24 @@ def main(): p.add_argument("--task", required=True) p.add_argument("--repo", default=None) p.add_argument("--no-worktree", action="store_true") + p.add_argument("--depends-on", default=None, + help="逗号分隔的依赖任务名，依赖未完成前不跑") + p.add_argument("--priority", type=int, default=0) + p.add_argument("--max-retries", type=int, default=0) sub.add_parser("list") p = sub.add_parser("status"); p.add_argument("--task", required=True) p = sub.add_parser("review"); p.add_argument("--task", required=True) p = sub.add_parser("merge"); p.add_argument("--task", required=True) + p.add_argument("--verify", default=None, + help="合并后验测命令；不传时仓库有 tests/ 就默认跑 unittest") + p.add_argument("--no-verify", action="store_true") + p = sub.add_parser("review-comment"); p.add_argument("--task", required=True) + p.add_argument("--file", required=True); p.add_argument("--line", type=int, required=True) + p.add_argument("--comment", required=True) + p = sub.add_parser("fix-review"); p.add_argument("--task", required=True) + sub.add_parser("doctor") + sub.add_parser("failures") p = sub.add_parser("discard"); p.add_argument("--task", required=True) p = sub.add_parser("archive"); p.add_argument("--task", required=True) p = sub.add_parser("fork"); p.add_argument("--task", required=True); p.add_argument("--name", required=True); p.add_argument("--sandbox", default=None) @@ -572,7 +978,11 @@ def main(): "sandbox": "workspace-write", "approval_policy": "never", "approval": "deny", "codex_home": "", "state": "host-state.json", - "_note": "API key 不放这里，只走环境变量", + "max_daily_turns": None, + "notify_command": "", + "_note": "API key 不放这里，只走环境变量；" + "max_daily_turns 是按 turn 数算的预算（协议不暴露 token 数）；" + "notify_command 是失败时跑的本地 shell 命令，可用 {task} {status}", } json.dump(init_cfg, open(config_path, "w"), indent=2, ensure_ascii=False) @@ -628,21 +1038,46 @@ def main(): print(f"已安装 crontab：{line}") return state = load_state(args.state) + if args.cmd == "doctor": + print(json.dumps(check_state(args.state, state), indent=2, + ensure_ascii=False)) + return + if args.cmd == "failures": + failed = [n for n, t in state["tasks"].items() + if t.get("status") in ("failed", "goal-unmet", "merge-failed")] + print(json.dumps({"failed_tasks": failed}, indent=2, ensure_ascii=False)) + return + if args.cmd == "review-comment": + t = state["tasks"][args.task] + c = add_review_comment(t, args.file, args.line, args.comment) + save_state(args.state, state) + print(json.dumps(c, indent=2, ensure_ascii=False)) + return srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home, approval=args.approval) try: srv.initialize() if args.cmd == "dispatch": + if budget_exceeded(state, cfg): + print("预算已用完（max_daily_turns），本次不派发，见 doctor 的 usage") + return + deps = [d.strip() for d in args.depends_on.split(",") + if d.strip()] if args.depends_on else [] task = dispatch(srv, state, args.name, args.task, args.repo, args.model, args.effort, args.sandbox, args.approval_policy, - use_worktree=not args.no_worktree) + use_worktree=not args.no_worktree, + depends_on=deps, priority=args.priority, + max_retries=args.max_retries) # 生产教训：先落盘任务登记再跑 turn，中途被杀/超时后 # 任务还在 host-state 里，可用 run 续跑；否则 turn 中断 # 就等于任务丢了（2026-10-25 委派评审 300 秒超时实测）。 save_state(args.state, state) - result = run_task_turn(srv, state, task, args.task, - args.model, args.effort, args.sandbox) + result = run_task_turn_with_retry(srv, state, task, args.task, + args.model, args.effort, + args.sandbox, cfg=cfg) + if result["status"] != "completed": + notify_failure(args.state, cfg, args.name, result) save_state(args.state, state) print(json.dumps({"task": args.name, "thread_id": task["thread_id"], "cwd": task["cwd"], "status": task["status"], @@ -666,30 +1101,53 @@ def main(): f"diff:\n{d['diff'][:8000]}") for rel, content in (d.get("untracked_contents") or {}).items(): print(f"\n--- untracked file: {rel} ---\n{content}") + oc = open_review_comments(t) + if oc: + print("\n--- open review comments ---") + for c in oc: + print(f"{c['file']}:{c['line']}: {c['comment']}") elif args.cmd == "merge": t = state["tasks"][args.task] if not t.get("repo") or not t.get("branch"): print("非 git worktree 任务，无法 merge，直接看 diff 后手动处理") return - # 关键一步：worktree 里的改动默认是未提交的，直接 merge 分支 - # 合不到任何东西。先在 worktree 里提交，再把分支合并进主仓。 - git(["add", "-A"], t["cwd"]) - c = git(["commit", "-m", f"host: {args.task}"], t["cwd"]) - if c.returncode != 0 and "nothing to commit" not in c.stdout + c.stderr: - print(f"worktree 提交失败：{c.stderr}") + # 合并 + 验测 + 回滚：验测命令默认仓库有 tests/ 就跑 + # unittest，--no-verify 跳过，--verify 可指定命令。 + # 验测失败时 merge_task 已 reset 回合并前 HEAD，分支和 + # worktree 保留，任务标 merge-failed，不删改动。 + verify_cmd = "" if args.no_verify else args.verify + ok, msg = merge_task(t["repo"], t["cwd"], t["branch"], + args.task, verify_command=verify_cmd) + print(msg) + if ok: + t["status"] = "merged" + else: + t["status"] = "merge-failed" + notify_failure(args.state, cfg, args.task, + {"status": "failed", "text": msg[:200]}) + save_state(args.state, state) + elif args.cmd == "fix-review": + t = state["tasks"][args.task] + oc = open_review_comments(t) + if not oc: + print("没有未处理的评审评论") return - r = git(["merge", "--no-ff", t["branch"], "-m", f"host: {args.task}"], - t["repo"]) - if r.returncode != 0: - git(["merge", "--abort"], t["repo"]) - print(f"合并冲突，已 abort：{r.stderr}") - return - git(["worktree", "remove", "--force", t["cwd"]], t["repo"]) - git(["branch", "-d", t["branch"]], t["repo"]) - t["status"] = "merged" + srv.resume_thread(t["thread_id"]) + result = run_task_turn(srv, state, t, build_fix_prompt(t), + args.model, args.effort, args.sandbox) + if result["status"] == "completed": + for c in oc: + c["status"] = "resolved" + c["resolved_at"] = time.time() + print(f"fix-review completed，{len(oc)} 条评论已标 resolved") + else: + notify_failure(args.state, cfg, args.task, result) + print(f"fix-review failed，评论仍保持 open：{result}") save_state(args.state, state) - print(f"merged {t['branch']} into {t['repo']}") elif args.cmd == "discard": + if args.task not in state["tasks"]: + print(f"任务不存在：{args.task}（可能从未派发，或 skipped 任务未落盘）") + return t = state["tasks"][args.task] if t.get("repo") and t.get("branch"): git(["worktree", "remove", "--force", t["cwd"]], t["repo"]) @@ -758,8 +1216,11 @@ def main(): elif args.cmd == "run": t = state["tasks"][args.task] srv.resume_thread(t["thread_id"]) - result = run_task_turn(srv, state, t, args.prompt, - args.model, args.effort, args.sandbox) + result = run_task_turn_with_retry(srv, state, t, args.prompt, + args.model, args.effort, + args.sandbox, cfg=cfg) + if result["status"] != "completed": + notify_failure(args.state, cfg, args.task, result) save_state(args.state, state) print(json.dumps(result, indent=2, ensure_ascii=False)) elif args.cmd == "delegate": @@ -820,30 +1281,105 @@ def main(): ok = True break task["status"] = "goal-met" if ok else "goal-unmet" + if not ok: + notify_failure(args.state, cfg, args.name, + {"status": "goal-unmet", "text": last_verify[:200]}) save_state(args.state, state) print(f"goal {'met' if ok else 'unmet'} after rounds") elif args.cmd == "parallel": jobs = json.load(open(args.jobs)) - # 先全部派发（建 thread + worktree），再并发跑 turn - tasks = [] - for job in jobs: - t = dispatch(srv, state, job["name"], job["task"], - job.get("repo", ""), args.model, args.effort, - job.get("sandbox", args.sandbox), - args.approval_policy, - use_worktree=bool(job.get("repo"))) - tasks.append((job, t)) - save_state(args.state, state) + # 依赖 + 优先级：先拓扑排序，缺依赖/成环直接报错不跑； + # 按波次派发 + 跑：依赖都 completed 的任务才进本波， + # 依赖 failed/skipped 的任务标 skipped 不跑，避免在坏 + # 基础上浪费模型调用。关键：依赖任务不是从 HEAD 现建 + # 的——本波派发时从依赖分支建 worktree（见 + # dispatch_from_branch），这样依赖任务的分支是依赖分支 + # 的后代，合并时先合依赖、再合依赖任务才不会重复加同一段。 + ordered = topo_sort_jobs(jobs) + tasks = {} + remaining = list(ordered) + while remaining: + # 先处理被依赖失败卡住的任务（它们还没派发， + # 所以这里用一个占位状态记录 skipped） + blocked = [] + for j in remaining: + deps = j.get("depends_on") or [] + if deps and any(d in tasks and tasks[d][1]["status"] + in ("failed", "skipped") for d in deps): + blocked.append(j) + for j in blocked: + print(f"{j['name']}: skipped (依赖失败)") + skipped_task = {"name": j["name"], "status": "skipped", + "cwd": None, "thread_id": "", + "branch": None, "repo": j.get("repo"), + "depends_on": j.get("depends_on"), + "priority": j.get("priority", 0), + "prompt": j["task"], "turns": []} + state["tasks"][j["name"]] = skipped_task + tasks[j["name"]] = (j, skipped_task) + remaining.remove(j) + if blocked: + save_state(args.state, state) + ready = [j for j in remaining + if all(d in tasks and tasks[d][1]["status"] == "completed" + for d in (j.get("depends_on") or []))] + if not ready: + if remaining: + print(f"剩余任务无法推进（依赖未完成）：" + f"{[j['name'] for j in remaining]}") + break + ready.sort(key=lambda j: (-(j.get("priority") or 0), j["name"])) + # 本波派发：有依赖且依赖是 git 任务时，从依赖分支 + # 建 worktree（有血缘关系，合并时不重复加同一段）； + # 否则退化为从 HEAD 建 + 文件拷贝（旧方法，仅兜底）。 + for job in ready: + deps = job.get("depends_on") or [] + t = None + if deps and job.get("repo"): + dep_task = tasks[deps[0]][1] + if dep_task.get("branch") and dep_task.get("cwd") \ + and os.path.isdir(dep_task["cwd"]): + t, err = dispatch_from_branch( + srv, state, job["name"], job["task"], + job.get("repo", ""), dep_task["branch"], + dep_task["cwd"], args.model, args.effort, + job.get("sandbox", args.sandbox), + args.approval_policy, + depends_on=deps, + priority=job.get("priority", 0), + max_retries=job.get("max_retries", 0)) + if t is None: + print(f"{job['name']}: 从依赖分支派发失败 " + f"({err})，退化为 HEAD + 文件拷贝") + if t is None: + t = dispatch(srv, state, job["name"], job["task"], + job.get("repo", ""), args.model, args.effort, + job.get("sandbox", args.sandbox), + args.approval_policy, + use_worktree=bool(job.get("repo")), + depends_on=deps, + priority=job.get("priority", 0), + max_retries=job.get("max_retries", 0)) + if deps and job.get("repo") and tasks[deps[0]][1].get("cwd"): + copy_tree_excluding_git(tasks[deps[0]][1]["cwd"], + t["cwd"]) + tasks[job["name"]] = (job, t) + save_state(args.state, state) - def _run(job_task): - job, t = job_task - return job["name"], run_task_turn(srv, state, t, job["task"], - args.model, args.effort, args.sandbox) - with ThreadPoolExecutor(max_workers=len(tasks)) as ex: - for name, result in ex.map(_run, tasks): - print(f"{name}: {result['status']} " - f"{result.get('text', '')[:300]}") - save_state(args.state, state) + def _run(job_task): + job, t = job_task + return job["name"], run_task_turn_with_retry( + srv, state, t, job["task"], args.model, + args.effort, args.sandbox, cfg=cfg) + with ThreadPoolExecutor(max_workers=len(ready)) as ex: + for name, result in ex.map( + _run, [tasks[j["name"]] for j in ready]): + print(f"{name}: {result['status']} " + f"{result.get('text', '')[:300]}") + if result["status"] != "completed": + notify_failure(args.state, cfg, name, result) + save_state(args.state, state) + remaining = [j for j in remaining if j not in ready] elif args.cmd == "automations": if args.add: state["automations"].append({ @@ -871,6 +1407,8 @@ def main(): result = run_task_turn(srv, state, task, auto["task"], args.model, args.effort, args.sandbox) auto["last_run"] = now + if result["status"] != "completed": + notify_failure(args.state, cfg, name, result) log_event(args.state, f"run-due {auto['name']} -> {name}: " f"{result['status']}") diff --git a/tests/test_host.py b/tests/test_host.py index d039ea0..0c5bdcf 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -52,5 +52,251 @@ class HostPureFunctionsTest(unittest.TestCase): self.assertEqual(host._load_config_file(missing), {}) +def _init_git_repo(tmp): + subprocess.run(["git", "init", "-q"], cwd=tmp, check=True) + subprocess.run(["git", "config", "user.name", "Test"], cwd=tmp, check=True) + subprocess.run(["git", "config", "user.email", "test@example.com"], + cwd=tmp, check=True) + + +class TopoSortTest(unittest.TestCase): + def test_priority_and_dependency_order(self): + jobs = [ + {"name": "low", "depends_on": [], "priority": 0}, + {"name": "high", "depends_on": [], "priority": 5}, + {"name": "child", "depends_on": ["low", "high"], "priority": 0}, + ] + order = [j["name"] for j in host.topo_sort_jobs(jobs)] + self.assertEqual(order[0], "high") + self.assertEqual(order[-1], "child") + + def test_cycle_and_missing_dependency_raise(self): + with self.assertRaises(ValueError): + host.topo_sort_jobs([{"name": "a", "depends_on": ["b"]}]) + with self.assertRaises(ValueError): + host.topo_sort_jobs([ + {"name": "a", "depends_on": ["b"]}, + {"name": "b", "depends_on": ["a"]}, + ]) + + +class MergeVerifyTest(unittest.TestCase): + def _make_repo(self, tmp, break_test=False): + _init_git_repo(tmp) + Path(tmp, "mod.py").write_text("def f():\n return 1\n", + encoding="utf-8") + tests_dir = Path(tmp, "tests") + tests_dir.mkdir() + body = ("import unittest\nfrom mod import f\n\n" + "class T(unittest.TestCase):\n" + " def test_f(self):\n" + " self.assertEqual(f(), 1)\n") + (tests_dir / "test_mod.py").write_text(body, encoding="utf-8") + subprocess.run(["git", "add", "-A"], cwd=tmp, check=True) + subprocess.run(["git", "commit", "-qm", "initial"], cwd=tmp, + check=True) + # worktree 手动建：模拟 host 的 worktree/分支 + subprocess.run(["git", "worktree", "add", os.path.join(tmp, "wt"), + "-b", "host/t1", "HEAD"], cwd=tmp, check=True) + if break_test: + Path(tmp, "wt", "mod.py").write_text( + "def f():\n return 2\n", encoding="utf-8") + else: + Path(tmp, "wt", "mod.py").write_text( + "def f():\n return 1 # comment\n", encoding="utf-8") + return os.path.join(tmp, "wt") + + def test_merge_verify_pass_cleans_worktree(self): + with tempfile.TemporaryDirectory() as tmp: + wt = self._make_repo(tmp) + ok, msg = host.merge_task(tmp, wt, "host/t1", "t1", + verify_command=None) + self.assertTrue(ok, msg) + self.assertFalse(os.path.isdir(wt)) + out = subprocess.run(["git", "branch", "--list", "host/t1"], + cwd=tmp, capture_output=True, text=True) + self.assertEqual(out.stdout.strip(), "") + + def test_merge_verify_fail_rolls_back_and_keeps_worktree(self): + with tempfile.TemporaryDirectory() as tmp: + wt = self._make_repo(tmp, break_test=True) + pre = subprocess.run(["git", "rev-parse", "HEAD"], cwd=tmp, + capture_output=True, text=True).stdout.strip() + ok, msg = host.merge_task(tmp, wt, "host/t1", "t1", + verify_command=None) + self.assertFalse(ok) + self.assertIn("回滚", msg) + now = subprocess.run(["git", "rev-parse", "HEAD"], cwd=tmp, + capture_output=True, text=True).stdout.strip() + self.assertEqual(now, pre) + self.assertTrue(os.path.isdir(wt)) + self.assertEqual(Path(tmp, "mod.py").read_text(), + "def f():\n return 1\n") + + def test_merge_duplicate_add_auto_resolved_with_ours(self): + # 依赖任务场景：两个分支都在同一位置加了同一段 add2， + # 先合并第一个，再合并第二个时应自动解（-X ours）， + # 且第二个独有的 mul2 仍在主仓里。 + with tempfile.TemporaryDirectory() as tmp: + _init_git_repo(tmp) + Path(tmp, "mod.py").write_text("def f():\n return 1\n", + encoding="utf-8") + subprocess.run(["git", "add", "-A"], cwd=tmp, check=True) + subprocess.run(["git", "commit", "-qm", "initial"], cwd=tmp, + check=True) + # 分支1：加 add2 + subprocess.run(["git", "worktree", "add", + os.path.join(tmp, "wt1"), "-b", "host/t1", + "HEAD"], cwd=tmp, check=True) + Path(tmp, "wt1", "mod.py").write_text( + "def f():\n return 1\n\ndef add2(a):\n return a + 2\n", + encoding="utf-8") + ok, _ = host.merge_task(tmp, os.path.join(tmp, "wt1"), + "host/t1", "t1", verify_command="") + self.assertTrue(ok) + # 分支2：从初始 HEAD 建，但拷了分支1的改动 + 自己的 mul2 + subprocess.run(["git", "worktree", "add", + os.path.join(tmp, "wt2"), "-b", "host/t2", + "HEAD"], cwd=tmp, check=True) + Path(tmp, "wt2", "mod.py").write_text( + "def f():\n return 1\n\ndef add2(a):\n return a + 2\n\n" + "def mul2(a):\n return a * 2\n", encoding="utf-8") + ok, msg = host.merge_task(tmp, os.path.join(tmp, "wt2"), + "host/t2", "t2", verify_command="") + self.assertTrue(ok, msg) + content = Path(tmp, "mod.py").read_text() + self.assertIn("def add2", content) + self.assertIn("def mul2", content) + # add2 只出现一次，不是重复两份 + self.assertEqual(content.count("def add2"), 1) + + def test_merge_does_not_commit_pycache(self): + with tempfile.TemporaryDirectory() as tmp: + wt = self._make_repo(tmp) + # 模拟 agent 跑测试生成的缓存 + pyc_dir = Path(wt, "__pycache__") + pyc_dir.mkdir(exist_ok=True) + (pyc_dir / "mod.cpython-312.pyc").write_bytes(b"fake-pyc") + ok, msg = host.merge_task(tmp, wt, "host/t1", "t1", + verify_command="") + self.assertTrue(ok, msg) + tracked = subprocess.run(["git", "ls-files"], cwd=tmp, + capture_output=True, text=True).stdout + self.assertNotIn("pyc", tracked) + self.assertNotIn("__pycache__", tracked) + + +class ReviewCommentTest(unittest.TestCase): + def test_comment_and_fix_prompt(self): + task = {"name": "t", "review_comments": []} + host.add_review_comment(task, "mod.py", 12, "加 docstring") + host.add_review_comment(task, "mod.py", 20, "处理空输入") + self.assertEqual(len(host.open_review_comments(task)), 2) + prompt = host.build_fix_prompt(task) + self.assertIn("mod.py:12: 加 docstring", prompt) + self.assertIn("mod.py:20: 处理空输入", prompt) + + +class RetryAndBudgetTest(unittest.TestCase): + class FakeServer: + def __init__(self, results): + self.results = list(results) + self.calls = 0 + + def run_turn(self, thread_id, prompt, model=None, effort=None, + sandbox=None): + self.calls += 1 + return self.results.pop(0) + + def test_retry_then_success(self): + task = {"name": "t", "thread_id": "x", "sandbox": None, + "max_retries": 2, "retry_count": 0, "turns": []} + state = {"tasks": {"t": task}, "automations": []} + srv = self.FakeServer([ + {"status": "failed", "text": ""}, + {"status": "completed", "text": "ok"}, + ]) + result = host.run_task_turn_with_retry(srv, state, task, "do it", + "m", "low", cfg={}) + self.assertEqual(result["status"], "completed") + self.assertEqual(task["retry_count"], 1) + self.assertEqual(srv.calls, 2) + + def test_budget_blocks_before_turn(self): + task = {"name": "t", "thread_id": "x", "sandbox": None, + "max_retries": 0, "retry_count": 0, "turns": []} + today = __import__("time").strftime("%Y-%m-%d") + state = {"tasks": {"t": task}, "automations": [], + "usage": {"date": today, "turns": 5}} + srv = self.FakeServer([{"status": "completed", "text": "ok"}]) + result = host.run_task_turn_with_retry( + srv, state, task, "do it", "m", "low", + cfg={"max_daily_turns": 5}) + self.assertEqual(result["status"], "failed") + self.assertEqual(srv.calls, 0) + + def test_notify_failure_writes_log_and_runs_command(self): + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + notify_log = os.path.join(tmp, "notify.log") + cfg = {"notify_command": f"echo {{task}} {{status}} >> {notify_log}"} + host.notify_failure(state_path, cfg, "t1", + {"status": "failed", "text": "boom"}) + self.assertTrue(os.path.exists( + os.path.join(tmp, "host-run.log"))) + self.assertIn("t1 failed", Path(notify_log).read_text()) + + +class StateVersionTest(unittest.TestCase): + def test_version_increments_and_conflict_logged(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "host-state.json") + s1 = {"tasks": {"a": {"status": "completed"}}, "automations": []} + host.save_state(path, s1) + with open(path) as f: + self.assertEqual(json.load(f)["version"], 1) + # 模拟另一写者：直接改磁盘版本为 5，再保存 s1（加载版本是 1） + with open(path) as f: + disk = json.load(f) + disk["version"] = 5 + disk["tasks"]["b"] = {"status": "completed"} + with open(path, "w") as f: + json.dump(disk, f) + host.save_state(path, s1) + with open(path) as f: + merged = json.load(f) + self.assertEqual(merged["version"], 6) + self.assertIn("b", merged["tasks"]) + self.assertIn("a", merged["tasks"]) + log = Path(tmp, "host-run.log").read_text() + self.assertIn("STATE-CONFLICT", log) + + +class ServerRestartTest(unittest.TestCase): + def test_kill_then_ensure_alive(self): + # 不跑模型：只验证 server 进程死了能重拉、list 能通。 + # 用配对二进制（VM 本机补丁），隔离 CODEX_HOME。 + import host as host_mod + with tempfile.TemporaryDirectory() as tmp: + codex_home = os.path.join(tmp, "codex-home") + srv = host_mod.Server(codex_home=codex_home, approval="deny") + try: + srv.initialize() + self.assertTrue(srv.is_alive()) + srv.proc.kill() + srv.proc.wait(timeout=10) + self.assertFalse(srv.is_alive()) + self.assertTrue(srv.ensure_alive()) + self.assertTrue(srv.is_alive()) + # list_threads 应能通（空或已有 thread 列表均可） + srv.list_threads() + finally: + try: + srv.close() + except Exception: + pass + + if __name__ == "__main__": unittest.main() +