host: fill second-class gaps (merge verify, deps, review comments, reconnect, budget, notify, state version)

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']}")