#!/usr/bin/env python3
"""
宿主编排层（无 GUI）：对齐 ChatGPT / Codex 桌面版宿主该干的事，
而不是再写一个 agent 循环。

ChatGPT / Codex 桌面版作为宿主，核心就这几件（见 README 的对齐表）：
  1. 线程生命周期：建、续跑、fork、归档、删除、列表、持久化
  2. 并行派发：多 thread 同时跑，每任务一个 git worktree 隔离
  3. 子代理委派：主 thread 派子 thread，子 thread 结果汇总回主 thread
     （模型内子代理是另一层，见 examples/subagent-prompt.txt；
      这里是宿主级委派，用 parentTurnId / rootTurnId 把委派链串起来）
  4. 长任务 / 目标循环：同一 thread 多回合，直到验收命令通过或到上限
  5. 评审队列：diff + 状态 + 合并 / 丢弃，人工在合并前把关
  6. 自动化：按间隔到期的任务自动派发，结果进评审队列（无推送，
     这是云端 ChatGPT 才有的部分，见 README 未对齐表）
  7. 边界：宿主定 cwd、沙箱、审批、模型、MCP/skills，agent 只在边界内跑

实现方式：一个 app-server 进程托管多个 thread（官方 thread manager
就是这么设计的），本文件用后台读线程按 id / threadId 分发响应和
turn/completed 通知，所以多个 thread 的 turn 可以并发跑。
线程状态本身持久化在 CODEX_HOME，host-state.json 只存宿主自己的
任务登记（task -> thread/worktree/状态/diff），不是 agent 状态的
第二份拷贝。

用法示例：
  python3 host.py dispatch --repo work/host-repo --task "给 sample.py 加 add2 函数" --name add2
  python3 host.py list
  python3 host.py status add2
  python3 host.py review add2
  python3 host.py merge add2        # 评审通过后才合并，冲突会 abort
  python3 host.py discard add2      # 评审不通过，删 worktree 和分支
  python3 host.py run --task add2 --prompt "再补一个测试"
  python3 host.py fork --task add2 --name add2-v2
  python3 host.py delegate --parent-task add2 --task "只读检查 add2 有没有边界问题" --name add2-review
  python3 host.py goal --repo work/host-repo --task "创建 result.txt，内容只有一行 goal-ok" \
      --verify "test -f result.txt" --name goal1
  python3 host.py parallel --jobs examples/host-jobs.json
  python3 host.py automations --add --name daily-check --repo work/host-repo \
      --task "只读检查 sample.py 有没有明显问题" --interval 86400
  python3 host.py run-due

默认 CODEX_HOME 是 ~/.codex；测试时用 --codex-home 指向隔离目录。
默认用 PATH 里的 codex；本机 VM 需用配对二进制时设
CODEX_BIN=.../bin/codex（start.sh 已自动优先用 bin/codex）。
"""
import argparse
import json
import os
import queue
import re
import shlex
import subprocess
import sys
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor

HERE = os.path.dirname(os.path.abspath(__file__))
DEFAULT_STATE = os.path.join(HERE, "host-state.json")


def _sum_turn_usage(events, thread_id, turn_id):
    """从事件窗口里汇总一回合的上游真实用量。

    app-server 在每个上游补全结束时发 thread/tokenUsage/updated，
    params.tokenUsage.last 是该次补全的用量（2026-10-26 中继实测：
    同一 turn 的多次 last 之和等于最后一次的 total，input/output/
    cached/reasoning 四路都对得上）。按 threadId + turnId 过滤后
    累加 last，得到本回合的真实用量；无匹配通知时返回 None（老
    版本 server 或回合未产生补全时调用方回落估算，绝不把 0 当成
    真实用量）。注意 totalTokens = inputTokens + outputTokens，
    reasoning 包含在 output 里，累加时不另算。
    """
    fields = (("input_tokens", "inputTokens"),
              ("cached_input_tokens", "cachedInputTokens"),
              ("cache_write_input_tokens", "cacheWriteInputTokens"),
              ("output_tokens", "outputTokens"),
              ("reasoning_output_tokens", "reasoningOutputTokens"),
              ("total_tokens", "totalTokens"))
    total = None
    for ev in events or []:
        if ev.get("method") != "thread/tokenUsage/updated":
            continue
        p = ev.get("params") or {}
        if p.get("threadId") != thread_id:
            continue
        if turn_id is not None and p.get("turnId") != turn_id:
            continue
        last = (p.get("tokenUsage") or {}).get("last") or {}
        if total is None:
            total = {k: 0 for k, _ in fields}
            total["completions"] = 0
        for key, wire in fields:
            total[key] += int(last.get(wire) or 0)
        total["completions"] += 1
    return total


def _partial_text_from_events(events):
    """超时回合的 best-effort 部分文本：从事件窗口里已完成的
    agentMessage 收集文本（不含流式 delta，避免重复拼接）。
    找不到时返回空串，调用方不得把部分文本当成最终结论。"""
    parts = []
    for ev in events or []:
        p = ev.get("params") or {}
        item = p.get("item") or {}
        if item.get("type") == "agentMessage" and item.get("text"):
            text = item["text"]
            if not parts or parts[-1] != text:
                parts.append(text)
    return "\n".join(parts)


def _interrupt_request_path(state_path, task_name):
    """跨进程中断请求文件路径：<state_dir>/interrupt-requests/<task>.json。

    属主进程在等 turn 时每秒轮询这个文件，另一个进程的
    `host.py interrupt --task` 直接协议中断不生效时就写这个文件，
    属主用自己的 live 连接发 turn/interrupt——只有属主的连接
    一定拥有这个 turn，这是跨进程中断可靠的原因。文件本身不
    由 run_task_turn 创建，只由 interrupt 命令写、属主读到后删。
    """
    safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_name))
    return os.path.join(
        os.path.dirname(os.path.abspath(state_path)),
        "interrupt-requests", safe + ".json")


def _wait_turn(q, timeout, progress_interval=30, on_progress=None,
               interrupt_check=None):
    """等 turn/completed 的轮询等待（对齐 Codex App 的长等待语义）。

    原来是一次性 q.get(timeout) / q.get()，长等待全程静默、另一
    进程无法插手；现在按 1 秒切片轮询，每片做两件事：
    1) interrupt_check() 为真时立刻返回 ("interrupted-requested", None)，
       由调用方用自己的 live 连接发 turn/interrupt 再记 interrupted；
    2) progress_interval > 0 且距上次进度已满一个间隔时调
       on_progress(elapsed_seconds)，让调用方打流式进度。
    返回：("completed", done) / ("timeout", None) /
    ("interrupted-requested", None)。timeout 为 None 或 <=0 时
    不限时，只等 done 或中断请求；正数时到时返回 timeout。
    """
    start = time.monotonic()
    next_progress = progress_interval if progress_interval and \
        progress_interval > 0 else None
    while True:
        elapsed = time.monotonic() - start
        if timeout and timeout > 0:
            remaining = timeout - elapsed
            if remaining <= 0:
                return "timeout", None
            slice_s = min(1.0, remaining)
        else:
            slice_s = 1.0
        # 切片还要受下一个进度点约束，否则一个长切片会把
        # 中间的进度检查全吞掉（timeout=0.35、间隔=0.1 时只会
        # 打 1 次进度，而不是 3 次；单测抓住了这个 bug）。
        if next_progress is not None:
            slice_s = min(slice_s, max(0.01, next_progress - elapsed))
        try:
            done = q.get(timeout=slice_s)
            return "completed", done
        except queue.Empty:
            pass
        elapsed = time.monotonic() - start
        if interrupt_check is not None:
            try:
                if interrupt_check():
                    return "interrupted-requested", None
            except Exception:
                pass
        if next_progress is not None and elapsed >= next_progress:
            if on_progress is not None:
                try:
                    on_progress(elapsed)
                except Exception:
                    pass
            next_progress += progress_interval
        if timeout and timeout > 0 and elapsed >= timeout:
            return "timeout", None


def _add_usage(a, b):
    """两份用量相加（任一为空时返回另一份；都为空返回 None）。"""
    if not a:
        return dict(b) if b else None
    if not b:
        return dict(a)
    out = dict(a)
    for k, v in b.items():
        if isinstance(v, int):
            out[k] = out.get(k, 0) + v
    return out


def default_codex_bin():
    if os.environ.get("CODEX_BIN"):
        return os.environ["CODEX_BIN"]
    paired = os.path.join(HERE, "bin", "codex")
    if os.path.exists(paired):
        return paired
    return "codex"


# 实时链路：单条通知里 item 的上限行数（防失控回合把文件写爆）。
TRACE_MAX_LINES = 5000

# agent shell 绝不能看到的变量（codex shell_environment_policy.exclude）。
# 实弹证实：不设此策略时 agent shell 直接继承 server 进程 env，
# XCM_API_KEY 在 local 路可被模型拿到；server 自己的中转调用走
# 进程 env（env_key），不受 shell 策略影响，所以只排 shell 不影响
# 模型调用。inherit 保持默认 all（PATH、代理变量等工作流变量不
# 动）；gh 用文件登录不用 env，故不排 *_TOKEN；四个精确名是当前
# worker env 里的全部密钥类变量，三条模式兜底将来新增的同类变量。
SHELL_ENV_EXCLUDE = [
    "XCM_API_KEY",
    "CLOUD_DB_PASSWORD",
    "GITHUB_OAUTH_CLIENT_SECRET",
    "GITHUB_WEBHOOK_SECRET",
    "*_KEY",
    "*_SECRET",
    "*_PASSWORD",
]


def _summarize_trace_item(method, params):
    """app-server 通知 → 链路条目列表（实时链路用）。

    只取过程事实，不取大文本：命令截 500 字、消息截 300 字、
    思考摘要截 200 字；fileChange 一项一变化拆多条。
    返回 [] 表示该通知不进链路（userMessage 等）。
    ts 取通知的 completedAtMs/startedAtMs（毫秒 epoch），
    没有时为 None，服务端展示时按无时间处理。
    thread 取通知的 threadId（模型内子代理与主任务同文件时，
    服务端凭它把条目归属到不同代理；没有时为 None，归主任务）。
    """
    item = (params or {}).get("item")
    if not isinstance(item, dict):
        return []
    item_type = item.get("type")
    item_id = item.get("id") or ""
    ts = params.get("completedAtMs") or params.get("startedAtMs")
    thread = params.get("threadId")
    if not thread and isinstance(params.get("thread"), dict):
        thread = params["thread"].get("id")
    if item_type == "commandExecution":
        return [{"kind": "exec", "id": item_id, "ts": ts, "thread": thread,
                 "command": (item.get("command") or "")[:500],
                 "cwd": item.get("cwd") or "",
                 "exitCode": item.get("exitCode"),
                 "durationMs": item.get("durationMs")}]
    if item_type == "fileChange":
        entries = []
        for ch in item.get("changes") or []:
            if not isinstance(ch, dict):
                continue
            path = ch.get("path") or ""
            entries.append({"kind": "file",
                            "id": item_id + ":" + path, "ts": ts,
                            "thread": thread,
                            "path": path, "change": ch.get("kind") or "",
                            "status": item.get("status") or ""})
        return entries
    if item_type == "agentMessage":
        text = item.get("text") or ""
        if not text:
            return []
        return [{"kind": "message", "id": item_id, "ts": ts, "thread": thread,
                 "phase": item.get("phase") or "", "text": text[:300]}]
    if item_type == "reasoning":
        summary = item.get("summary") or item.get("content") or []
        text = " ".join(s for s in summary if isinstance(s, str)).strip()
        if not text:
            return []
        return [{"kind": "reasoning", "id": item_id, "ts": ts, "thread": thread,
                 "summary": text[:200]}]
    if item_type == "mcpToolCall":
        return [{"kind": "tool", "id": item_id, "ts": ts, "thread": thread,
                 "server": item.get("server") or "",
                 "tool": item.get("tool") or "",
                 "status": item.get("status") or ""}]
    if item_type == "functionCallOutput":
        return [{"kind": "call", "id": item_id, "ts": ts, "thread": thread,
                 "name": item.get("name") or ""}]
    if item_type == "plan":
        text = item.get("text") or ""
        if not text:
            return []
        return [{"kind": "plan", "id": item_id, "ts": ts, "thread": thread,
                 "text": text[:300]}]
    return []


def _trace_file_path(state_path, task_name):
    """链路文件路径：state 文件同目录下的 host-traces/<清洗后名字>.jsonl。

    与 Server.set_trace_file、cloud 的 _trace_file_name 同款清洗，
    三边算出同一名字，delegate 的父链路标记才能落进父任务文件。
    """
    safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_name))
    return os.path.join(
        os.path.dirname(os.path.abspath(state_path)),
        "host-traces", safe + ".jsonl")


def _append_trace_marker(state_path, task_name, entry):
    """往另一任务的链路文件追加一条标记（delegate 向父链路报到用）。

    尽力而为：目录建不出、写失败都不影响回合本身。父链路的
    行数上限由属主进程自己的计数器管，这里不重复计数（标记
    一轮至多两条，超限风险可忽略）。
    """
    try:
        path = _trace_file_path(state_path, task_name)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception:
        pass


def _server_command(codex_bin, provider=None):
    """app-server 启动命令：-c 必须放在子命令之前。

    shell_environment_policy.exclude 只过滤 agent shell 看到的
    变量，不影响 server 进程自己的中转调用（env_key 直读进程 env）。
    provider="sub2api" 时再叠加 -c 覆盖：model_provider 与
    provider 表都按任务进程生效，不改任何租户的 config.toml
    （多网关任务同项目并发时互不干扰）。网关地址取环境变量
    SUB2API_BASE_URL（origin，不带 /v1）；未配则 loud fail，
    绝不静默跑回默认网关。
    """
    cmd = [
        codex_bin,
        "-c",
        "shell_environment_policy.exclude=" + json.dumps(SHELL_ENV_EXCLUDE),
    ]
    if provider is not None:
        if provider != "sub2api":
            raise ValueError("未知模型 provider: %r" % provider)
        origin = os.environ.get("SUB2API_BASE_URL", "").strip().rstrip("/")
        if not origin:
            raise RuntimeError(
                "provider=sub2api 但 SUB2API_BASE_URL 未配置")
        cmd += [
            "-c", 'model_provider="sub2api"',
            "-c", 'model_providers.sub2api.name="Sub2API"',
            "-c", 'model_providers.sub2api.base_url="%s/v1"' % origin,
            "-c", 'model_providers.sub2api.env_key="SUB2API_API_KEY"',
            "-c", 'model_providers.sub2api.wire_api="responses"',
        ]
    cmd.append("app-server")
    return cmd


class Server:
    """单个 app-server 进程，多 thread 并发。"""

    def __init__(self, codex_bin=None, codex_home=None, approval="deny",
                 progress_interval=30, model_provider=None):
        self.codex_bin = codex_bin or default_codex_bin()
        self.codex_home = codex_home
        self.approval = approval
        self.model_provider = model_provider
        # 流式进度间隔（秒）：run_turn 等待时每满一个间隔打一次
        # 进度；0 表示关闭进度输出，但中断轮询仍每秒进行。
        self.progress_interval = progress_interval
        self.send_lock = threading.Lock()
        self.next_id = 1
        self.pending = {}
        self.pending_lock = threading.Lock()
        self.turn_queues = {}
        self.events = []
        self.events_lock = threading.Lock()
        self.restart_count = 0
        # 实时链路文件：set_trace_file 未调时为 None，回合本身
        # 不受影响（写链路失败也绝不打断 _read_loop）。
        self.trace_path = None
        self._trace_lines = 0
        self._start()

    def set_trace_file(self, state_path, task_name):
        """链路文件定到 <state_dir>/host-traces/<task_name>.jsonl。

        state_dir 与 state 文件同目录：本地 worker 是
        /opt/codex-app-server/work，容器内是 /app/work（经挂载
        落到宿主同一目录），服务端按同一规则读。目录建不出
        时 trace_path 保持 None，不影响回合。
        """
        try:
            self.trace_path = _trace_file_path(state_path, task_name)
            os.makedirs(os.path.dirname(self.trace_path), exist_ok=True)
            self._trace_lines = 0
        except Exception:
            self.trace_path = None

    def _append_trace(self, msg):
        """通知 → 摘要条目 → 追加写链路文件（每事件开合一次文件，
        进程被杀也不丢已写行；行数到 TRACE_MAX_LINES 即停）。"""
        if self.trace_path is None or \
                self._trace_lines >= TRACE_MAX_LINES:
            return
        entries = _summarize_trace_item(msg.get("method"),
                                        msg.get("params") or {})
        if not entries:
            return
        with open(self.trace_path, "a", encoding="utf-8") as f:
            for entry in entries:
                if self._trace_lines >= TRACE_MAX_LINES:
                    break
                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
                self._trace_lines += 1

    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(
            _server_command(self.codex_bin,
                            provider=self.model_provider),
            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），
        # 按 MCP elicitation 协议回 action=accept/decline + content，
        # 回通用 {"decision": ...} 会被当成拒绝（2026-10-25 实测）。
        if msg.get("method") == "mcpServer/elicitation/request":
            if self.approval == "auto":
                result = {"action": "accept", "content": {}}
            else:
                result = {"action": "decline"}
            self._send({"jsonrpc": "2.0", "id": msg["id"], "result": result})
            return
        decision = "approve" if self.approval == "auto" else "deny"
        self._send({"jsonrpc": "2.0", "id": msg["id"],
                    "result": {"decision": decision}})

    def _read_loop(self):
        try:
            for raw in self.proc.stdout:
                if not raw.strip():
                    continue
                try:
                    msg = json.loads(raw)
                except Exception:
                    continue
                if "method" in msg and "id" in msg:
                    self._answer_server_request(msg)
                    continue
                if "id" in msg:
                    with self.pending_lock:
                        q = self.pending.get(msg["id"])
                    if q is not None:
                        q.put(msg)
                    continue
                # 通知
                params = msg.get("params") or {}
                thread_id = params.get("threadId")
                if not thread_id and isinstance(params.get("thread"), dict):
                    thread_id = params["thread"].get("id")
                with self.events_lock:
                    self.events.append(msg)
                if self.trace_path is not None:
                    try:
                        self._append_trace(msg)
                    except Exception:
                        pass
                if msg.get("method") == "turn/completed" and thread_id:
                    q = self.turn_queues.get(thread_id)
                    if q is not None:
                        q.put(msg)
        except Exception:
            pass

    def _send(self, obj):
        with self.send_lock:
            self.proc.stdin.write((json.dumps(obj) + "\n").encode())
            self.proc.stdin.flush()

    def request(self, method, params=None, timeout=120):
        # 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[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 {}})

    def initialize(self):
        resp = self.request("initialize", {
            "clientInfo": {"name": "codex-host", "version": "0.1"},
        }, timeout=60)
        if not resp or "error" in resp:
            raise RuntimeError(f"initialize 失败: {resp}")
        self.notify("initialized", {})
        return resp["result"]

    def create_thread(self, cwd, model=None, sandbox="workspace-write",
                      approval_policy="never", instructions=None):
        params = {"cwd": os.path.abspath(cwd),
                  "approvalPolicy": approval_policy, "sandbox": sandbox}
        if model:
            params["model"] = model
        if instructions:
            params["developerInstructions"] = instructions
        resp = self.request("thread/start", params, timeout=120)
        if not resp or "error" in resp:
            raise RuntimeError(f"thread/start 失败: {resp}")
        return resp["result"]["thread"]["id"]

    def list_threads(self):
        resp = self.request("thread/list", {}, timeout=60)
        if not resp or "error" in resp:
            raise RuntimeError(f"thread/list 失败: {resp}")
        return resp["result"]

    def resume_thread(self, thread_id):
        resp = self.request("thread/resume", {"threadId": thread_id}, timeout=120)
        if not resp or "error" in resp:
            raise RuntimeError(f"thread/resume 失败: {resp}")
        return resp["result"]

    def start_review(self, thread_id, target, timeout=600):
        # 协议 review/start：params 必填 threadId 与 target，
        # target 四种见 protocol v2 的 ReviewTarget（uncommitted
        # Changes / baseBranch / commit / custom）；delivery 缺
        # 省即 inline，不发 detached（已废弃）。失败直接抛错，
        # 调用方不得把 error 响应当成功打印。
        params = {"threadId": thread_id, "target": target}
        resp = self.request("review/start", params, timeout=timeout)
        if not resp or "error" in resp:
            raise RuntimeError(f"review/start 失败: {resp}")
        return resp["result"]

    def fork_thread(self, thread_id, cwd=None, sandbox=None):
        params = {"threadId": thread_id}
        if cwd:
            params["cwd"] = os.path.abspath(cwd)
        if sandbox:
            params["sandbox"] = sandbox
        resp = self.request("thread/fork", params, timeout=120)
        if not resp or "error" in resp:
            raise RuntimeError(f"thread/fork 失败: {resp}")
        # fork 响应里 thread 字段形态随版本略有差异，尽量都取
        r = resp["result"]
        if isinstance(r, dict):
            t = r.get("thread") or {}
            return t.get("id") or r.get("threadId") or r.get("id")
        return None

    def archive_thread(self, thread_id):
        resp = self.request("thread/archive", {"threadId": thread_id}, timeout=60)
        if resp and "error" in resp:
            raise RuntimeError(f"thread/archive 失败: {resp}")
        return resp

    def run_turn(self, thread_id, prompt, model=None, effort=None,
                 parent_turn_id=None, root_turn_id=None, timeout=600,
                 sandbox=None, progress_interval=None, interrupt_file=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)
        try:
            params = {"threadId": thread_id,
                      "input": [{"type": "text", "text": prompt}]}
            if model:
                params["model"] = model
            if effort:
                params["effort"] = effort
            if sandbox:
                # turn 级 sandboxPolicy 会覆盖 thread 级设置，且对后续
                # turn 也生效。这是修复 fork 换 cwd 后被重置为 read-only
                # 的关键：fork 时传 sandbox 参数服务端不认，turn 里传才认。
                tmap = {"workspace-write": "workspaceWrite",
                        "read-only": "readOnly",
                        "danger-full-access": "dangerFullAccess"}
                params["sandboxPolicy"] = {"type": tmap.get(sandbox, sandbox)}
            if parent_turn_id:
                params["parentTurnId"] = parent_turn_id
            if root_turn_id:
                params["rootTurnId"] = root_turn_id
            resp = self.request("turn/start", params, timeout=120)
            if not resp or "error" in resp:
                return {"status": "failed", "text": "", "error": resp}
            turn_id = None
            if isinstance(resp.get("result"), dict):
                t = resp["result"].get("turn") or {}
                turn_id = t.get("id")
            try:
                # 对齐 Codex App 的长任务语义：timeout 为 None 或 0
                # 表示不限时、一直等到 turn/completed（由用户中断
                # 或模型自己结束）；正数才按秒计时。默认仍是 600，
                # 由调用方（CLI --turn-timeout / 配置 turn_timeout）
                # 显式改，不改默认就不会悄悄挂起自动化。
                # 等待不再是一次性 q.get()：按 1 秒切片轮询，每满
                # progress_interval 打一次流式进度（stderr，绝不
                # 走 stdout——dispatch/run 的 stdout 是 JSON、goal
                # 的 stdout 要被 worker 解析结论行与用量行，进度
                # 进 stdout 会把这两种解析都打坏；链路文件里也
                # 同步记一条 kind=progress，console 按 fallback
                # 把它当 JSON 行展示），并每秒检查跨进程中断
                # 请求文件（见 _interrupt_request_path）。
                interval = self.progress_interval if \
                    progress_interval is None else progress_interval

                def _on_progress(elapsed):
                    sys.stderr.write(
                        "turn progress: elapsed=%ds\n"
                        % int(elapsed))
                    sys.stderr.flush()
                    if self.trace_path is not None and \
                            self._trace_lines < TRACE_MAX_LINES:
                        try:
                            with self.events_lock:
                                n_events = len(self.events)
                            with open(self.trace_path, "a",
                                      encoding="utf-8") as f:
                                f.write(json.dumps(
                                    {"kind": "progress",
                                     "elapsed_s": int(elapsed),
                                     "events": n_events},
                                    ensure_ascii=False) + "\n")
                            self._trace_lines += 1
                        except Exception:
                            pass

                def _interrupt_check():
                    return bool(interrupt_file) and \
                        os.path.exists(interrupt_file)

                wait_status, done = _wait_turn(
                    q, timeout, progress_interval=interval,
                    on_progress=_on_progress,
                    interrupt_check=_interrupt_check)
                if wait_status == "interrupted-requested":
                    # 属主用自己的 live 连接发中断：只有这个连接
                    # 一定拥有该 turn，另一进程的直接中断未必生效。
                    # 中断请求本身已明确，属主不再继续等：即使
                    # 服务端回 error 也记 interrupted（任务可续跑，
                    # 不是失败），并把请求文件删掉避免重复触发；
                    # 服务端侧的停止是 best-effort，stderr 留一行。
                    try:
                        resp = self.interrupt_turn(thread_id, turn_id)
                        if resp and "error" in resp:
                            sys.stderr.write(
                                "turn interrupt request returned error: "
                                "%s\n" % (resp,))
                            sys.stderr.flush()
                    except Exception as exc:
                        sys.stderr.write(
                            "turn interrupt request failed: %s\n" % (exc,))
                        sys.stderr.flush()
                    try:
                        if interrupt_file:
                            os.remove(interrupt_file)
                    except Exception:
                        pass
                    with self.events_lock:
                        evs = list(self.events[start_events:])
                    return {"status": "interrupted",
                            "text": _partial_text_from_events(evs),
                            "turn_id": turn_id,
                            "usage": _sum_turn_usage(evs, thread_id, turn_id)}
                if wait_status == "timeout":
                    with self.events_lock:
                        evs = list(self.events[start_events:])
                    return {"status": "timeout",
                            "text": _partial_text_from_events(evs),
                            "turn_id": turn_id,
                            "usage": _sum_turn_usage(evs, thread_id, turn_id)}
            except queue.Empty:
                with self.events_lock:
                    evs = list(self.events[start_events:])
                return {"status": "timeout",
                        "text": _partial_text_from_events(evs),
                        "turn_id": turn_id,
                        "usage": _sum_turn_usage(evs, thread_id, turn_id)}
            turn = (done.get("params") or {}).get("turn") or {}
            text = ""
            for item in turn.get("items") or []:
                if item.get("type") == "agentMessage" and \
                        item.get("phase") == "final_answer":
                    text = item.get("text") or text
            if not text:
                # 兜底：从本回合事件里找 final_answer
                with self.events_lock:
                    evs = list(self.events[start_events:])
                for ev in evs:
                    p = ev.get("params") or {}
                    item = p.get("item") or {}
                    if item.get("type") == "agentMessage" and \
                            item.get("phase") == "final_answer" and \
                            item.get("text"):
                        text = item["text"]
            status = turn.get("status", "completed")
            error = turn.get("error")
            if status != "completed" and not error:
                # app-server 的 turn/completed 事件常常只给
                # status=failed，不给失败原因（采样失败的 404
                # 只写在它本地的 thread_history 库里，事件里没有）。
                # 此时 best-effort 从 codex_home 的 thread_history
                # 库里按 thread_id + turn_id 查 error_json，把
                # message 捞回来给调用方（worker 靠它判断是否可
                # fallback）；查不到、库不存在、表结构变了都只当
                # 没有原因，绝不因此把回合本身搞失败。
                error = self._turn_error_from_history(
                    thread_id, turn.get("id") or turn_id)
            # 真实用量：本回合窗口内的 tokenUsage/updated 通知累加。
            # 失败/超时回合也一样记（已产生的补全就是已花的钱）。
            with self.events_lock:
                evs = list(self.events[start_events:])
            usage = _sum_turn_usage(
                evs, thread_id, turn.get("id") or turn_id)
            return {"status": status, "text": text,
                    "turn_id": turn.get("id") or turn_id,
                    "root_turn_id": turn.get("rootTurnId"),
                    "error": error, "usage": usage}
        finally:
            self.turn_queues.pop(thread_id, None)

    def interrupt_turn(self, thread_id, turn_id):
        """对齐 Codex App 的停止按钮：向 app-server 发 turn/interrupt。
        协议要求 threadId + turnId；turn_id 缺失时无法中断，返回 None。
        注意：直接中断只对本 server 进程正在跑的 turn 可靠，另一进程的
        turn 未必能被本进程的中断请求停下；跨进程的可靠路径是
        中断请求文件（见 _interrupt_request_path）：interrupt 命令
        直接中断不生效时写文件，属主进程在等 turn 时每秒轮询到后
        用自己的 live 连接发中断（run_turn 的 interrupted-requested 分支）。"""
        if not thread_id or not turn_id:
            return None
        return self.request("turn/interrupt",
                            {"threadId": thread_id, "turnId": turn_id},
                            timeout=120)

    def _turn_error_from_history(self, thread_id, turn_id):
        """从 codex thread_history 库里取失败回合的 error message。

        只读、只在回合失败且事件本身没给原因时调用；任何异常
        都返回 None。库路径：codex_home/thread_history_1.sqlite，
        codex_home 未指定时用 ~/.codex。
        """
        if not thread_id or not turn_id:
            return None
        try:
            import sqlite3
            home = self.codex_home or os.path.expanduser("~/.codex")
            db_path = os.path.join(home, "thread_history_1.sqlite")
            if not os.path.exists(db_path):
                return None
            conn = sqlite3.connect("file:%s?mode=ro" % db_path, uri=True)
            try:
                row = conn.execute(
                    "SELECT error_json FROM thread_turns"
                    " WHERE thread_id = ? AND turn_id = ?",
                    (thread_id, turn_id)).fetchone()
            finally:
                conn.close()
            if not row or not row[0]:
                return None
            parsed = json.loads(row[0])
            msg = parsed.get("message") if isinstance(parsed, dict) else None
            if not msg:
                return None
            return str(msg)[:1000]
        except Exception:
            return None

    def close(self):
        try:
            self.proc.stdin.close()
        except Exception:
            pass
        try:
            self.proc.wait(timeout=5)
        except Exception:
            self.proc.kill()
        try:
            self.proc.stdout.close()
        except Exception:
            pass


# ---------------- 宿主状态 / worktree / diff ----------------
# 生产加固：state 锁。锁只包住“读/写”这两个短临界区，不包住
# 整个模型 turn（turn 可能跑几分钟，包住它会把别的 host 进程
# 活活饿死）。保存时先在锁里重读磁盘，再把内存里的任务按名
# 覆盖、自动化按名合并，这样两个进程各改各的任务时，后保存
# 的不会把先保存的覆盖掉。tmp 文件带 pid，避免两个进程用
# 同一个 .tmp 互相踩。锁文件在 state 文件旁边，名如
# host-state.json.lock，用 fcntl.flock（Linux/macOS 都有）。
import contextlib
try:
    import fcntl
except ImportError:  # Windows 等无 fcntl 的平台：退化为无锁
    fcntl = None


@contextlib.contextmanager
def _state_lock(path, exclusive=True):
    lock_path = path + ".lock"
    os.makedirs(os.path.dirname(os.path.abspath(lock_path)), exist_ok=True)
    f = open(lock_path, "a+")
    try:
        if fcntl is not None:
            fcntl.flock(f.fileno(),
                        fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
        yield
    finally:
        if fcntl is not None:
            try:
                fcntl.flock(f.fileno(), fcntl.LOCK_UN)
            except Exception:
                pass
        f.close()


def _read_state_file(path):
    if os.path.exists(path):
        try:
            with open(path) as f:
                return json.load(f)
        except Exception:
            pass
    return {"tasks": {}, "automations": []}


def load_state_strict(path):
    """严格读 state：文件不存在返回 (空 state, None)；
    文件存在但 JSON 坏了返回 (None, 错误信息)，绝不静默
    当成空 state——否则损坏一次就等于全部任务丢了，
    这是自动恢复要先堵住的洞（旧 _read_state_file 会吞掉
    解析错误，health/recover 必须用这个严格版）。"""
    if not os.path.exists(path):
        return {"tasks": {}, "automations": []}, None
    try:
        with open(path) as f:
            return json.load(f), None
    except Exception as e:
        return None, str(e)


def restore_state(state_path):
    """state 损坏或丢失时从备份恢复：按 .bak、.bak.1、.bak.2
    顺序找第一个能解析的备份，先把坏文件（如果存在）改名
    成 .corrupt.<时间戳> 留底，再把备份拷回主文件。
    返回 (ok, message)。主文件完好时不做任何事。"""
    strict, err = load_state_strict(state_path)
    if err is None and strict is not None:
        return True, "state 完好，无需恢复"
    if os.path.exists(state_path):
        corrupt = f"{state_path}.corrupt.{int(time.time())}"
        try:
            os.replace(state_path, corrupt)
        except Exception as e:
            return False, f"坏 state 改名留底失败：{e}"
    else:
        corrupt = None
    for suffix in (".bak", ".bak.1", ".bak.2"):
        cand = state_path + suffix
        if not os.path.exists(cand):
            continue
        try:
            with open(cand) as f:
                data = json.load(f)
        except Exception:
            continue
        import shutil
        shutil.copy2(cand, state_path)
        return True, (f"已从 {suffix} 恢复，version="
                      f"{data.get('version')}，"
                      f"坏文件留底：{corrupt}")
    return False, "没有可用的备份（.bak/.bak.1/.bak.2 均缺失或损坏）"


def recover_interrupted(state, state_path, older_than=3600, cfg=None):
    """自动恢复（只标记，不自动重跑模型回合）：
    status=running 且最后回合时间超过 older_than 秒的任务，
    判定为进程被杀/VM 回收导致的中断，标 interrupted 并记日志，
    用户随后可用 run 续跑（thread 在 CODEX_HOME 里不丢）。
    不自动重跑是刻意的安全边界：自动重跑会不经用户同意
    花中转站额度，daemon 只负责标记 + 日志 + 通知。
    返回被标记的任务名列表。"""
    now = time.time()
    recovered = []
    for name, t in (state.get("tasks") or {}).items():
        if t.get("status") != "running":
            continue
        last = t.get("last_turn_at") or t.get("created_at") or 0
        if now - last < older_than:
            continue
        t["status"] = "interrupted"
        recovered.append(name)
        try:
            log_event(state_path, f"recover {name} -> interrupted "
                                  f"(last_turn {int(now - last)}s ago)")
        except Exception:
            pass
        if cfg is not None:
            notify_failure(state_path, cfg, name,
                           {"status": "interrupted",
                            "text": "process killed / VM reclaimed"})
    return recovered


def run_health_check(state_path, cfg, codex_bin, codex_home, approval,
                     server_factory=None):
    """健康检查（不跑模型）：返回 report dict，ok 为全部通过。
    检查项：二进制存在、配置可读、state 可解析、备份存在、
    daemon pid 是否存活/过期、server 能 initialize + list、
    缺失 worktree（doctor 的逻辑）。server_factory 供单测注入
    假 server，避免健康检查本身要拉真进程。"""
    checks = {}
    checks["codex_bin_exists"] = bool(codex_bin) and os.path.exists(codex_bin)
    checks["config_loads"] = isinstance(cfg, dict)
    strict, err = load_state_strict(state_path)
    checks["state_valid"] = err is None
    checks["state_error"] = err
    checks["backup_exists"] = os.path.exists(state_path + ".bak")
    if strict is not None:
        doc = check_state(state_path, strict)
        checks["failed_tasks"] = doc["failed_tasks"]
        checks["missing_worktrees"] = doc["missing_worktrees"]
        checks["open_review_comments"] = doc["open_review_comments"]
    else:
        checks["failed_tasks"] = []
        checks["missing_worktrees"] = []
        checks["open_review_comments"] = 0
    # daemon pid
    pidfile = os.path.join(os.path.dirname(os.path.abspath(state_path)),
                           "host-daemon.pid")
    checks["daemon_pidfile"] = pidfile
    if os.path.exists(pidfile):
        try:
            pid = int(open(pidfile).read().strip())
            os.kill(pid, 0)
            checks["daemon_alive"] = True
            checks["daemon_pid"] = pid
        except Exception:
            checks["daemon_alive"] = False
            checks["daemon_pid"] = None
    else:
        checks["daemon_alive"] = False
        checks["daemon_pid"] = None
    # server（无模型成本，只 initialize + list）
    try:
        if server_factory is not None:
            srv = server_factory()
        else:
            srv = Server(codex_bin=codex_bin, codex_home=codex_home,
                         approval=approval)
        try:
            srv.initialize()
            threads = srv.list_threads()
            checks["server_ok"] = True
            checks["server_threads"] = len(
                (threads or {}).get("data") or []) if isinstance(threads, dict) else None
        finally:
            try:
                srv.close()
            except Exception:
                pass
    except Exception as e:
        checks["server_ok"] = False
        checks["server_error"] = str(e)[:500]
    ok = bool(checks["codex_bin_exists"] and checks["state_valid"]
              and checks["server_ok"])
    return {"ok": ok, "checks": checks}


def daemonize(pidfile, logfile):
    """双 fork 守护进程（WeChat 桥 daemon_serve.py 同款教训：
    setsid nohup 会在 launcher 会话结束时被 runtime 按进程树
    回收杀掉，双 fork 到 PID 1 才能活过会话结束）。
    父进程返回 False（应退出），子进程返回 True（继续跑循环）。"""
    if os.fork() > 0:
        return False
    os.setsid()
    if os.fork() > 0:
        os._exit(0)
    sys.stdout.flush()
    sys.stderr.flush()
    log_f = open(logfile, "a", encoding="utf-8")
    os.dup2(log_f.fileno(), sys.stdout.fileno())
    os.dup2(log_f.fileno(), sys.stderr.fileno())
    with open("/dev/null", "r") as devnull:
        os.dup2(devnull.fileno(), sys.stdin.fileno())
    with open(pidfile, "w") as f:
        f.write(str(os.getpid()))
    return True



def load_state(path):
    with _state_lock(path, exclusive=False):
        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):
        # 锁内重读 + 合并：内存任务覆盖同名磁盘任务，磁盘上
        # 别的进程新增的任务保留；自动化按名合并，内存优先。
        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({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 []):
            if isinstance(a, dict) and a.get("name"):
                disk_autos[a["name"]] = a
        merged = {"tasks": merged_tasks,
                  "automations": list(disk_autos.values())}
        # 保留内存 state 里 tasks/automations 之外的顶层字段，
        # 但 _ 开头的内存标记不写回文件。
        for k, v in state.items():
            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)
        if os.path.exists(path):
            import shutil
            backup = path + ".bak"
            backup_1 = backup + ".1"
            backup_2 = backup + ".2"
            backup_tmp = f"{backup}.tmp.{os.getpid()}.{threading.get_ident()}"
            shutil.copy2(path, backup_tmp)
            if os.path.exists(backup_2):
                os.remove(backup_2)
            if os.path.exists(backup_1):
                os.replace(backup_1, backup_2)
            if os.path.exists(backup):
                os.replace(backup, backup_1)
            os.replace(backup_tmp, backup)
        os.replace(tmp, path)
        # 让调用方内存里的 state 也跟上合并结果，避免下次保存
        # 又把别的进程的任务挤掉（调用方继续用这份 state 时）。
        state.clear()
        state.update(merged)
        state["_loaded_version"] = merged["version"]


def log_event(state_path, message):
    """run-due / merge / discard 等运维事件写日志，不依赖标准输出。
    日志在 state 文件同目录的 host-run.log，定时任务跑完后
    看这个文件就知道派发了什么、跳过了什么。"""
    try:
        log_path = os.path.join(os.path.dirname(os.path.abspath(state_path)),
                                "host-run.log")
        ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        with open(log_path, "a", encoding="utf-8") as f:
            f.write(f"{ts} {message}\n")
    except Exception:
        pass


def git(args, cwd):
    return subprocess.run(["git"] + args, cwd=cwd, capture_output=True,
                          text=True)


def make_worktree(repo, task_name):
    """一任务一 worktree；不是 git 仓库时退化为独立目录拷贝。"""
    repo = os.path.abspath(repo)
    base = os.path.join(HERE, "work", "host-worktrees")
    os.makedirs(base, exist_ok=True)
    path = os.path.join(base, task_name)
    if os.path.exists(path):
        return path, None
    if os.path.isdir(os.path.join(repo, ".git")):
        branch = f"host/{task_name}"
        r = git(["worktree", "add", path, "-b", branch, "HEAD"], repo)
        if r.returncode != 0:
            # 分支已存在时直接用它
            r = git(["worktree", "add", path, branch], repo)
            if r.returncode != 0:
                raise RuntimeError(f"git worktree 失败: {r.stderr}")
        return path, branch
    # 非 git：复制一份
    import shutil
    shutil.copytree(repo, path)
    return path, None


def task_diff(worktree):
    if os.path.isdir(os.path.join(worktree, ".git")) or \
            subprocess.run(["git", "-C", worktree, "rev-parse"],
                           capture_output=True).returncode == 0:
        st = git(["status", "--porcelain"], worktree).stdout
        diff = git(["diff", "HEAD"], worktree).stdout
        untracked = git(["ls-files", "--others", "--exclude-standard"],
                        worktree).stdout
        # 生产评审要点：新文件默认是 untracked，git diff 里看不到内容，
        # 只列文件名等于没评审。把未追踪文本文件的内容也带上（截断）。
        contents = {}
        for rel in untracked.split():
            p = os.path.join(worktree, rel)
            if rel.endswith(".pyc") or "__pycache__" in rel:
                continue
            try:
                with open(p, encoding="utf-8") as f:
                    contents[rel] = f.read(8000)
            except Exception:
                contents[rel] = "<binary or unreadable>"
        return {"status": st, "diff": diff, "untracked": untracked,
                "untracked_contents": contents}
    return {"status": "", "diff": "",
            "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}）。

    插进命令串的值一律 shlex.quote：模型文本可能含 shell 元
    字符（引号、分号、$() 等），不加 quote 就等于把模型输出
    当命令执行。模板本身有问题（未知占位符、未成对的大括号）
    时不再静默吞掉：记一条 template error 到 host-run.log 并
    且不跑这条命令。命令的返回值 / 超时 / 异常同样记日志，
    否则通知丢了都无从发现（旧行为是异常全吞、无任何记录）。
    没有配 notify_command 时只写 host-run.log，不外发。"""
    result = result or {}
    text = result.get("text", "") or ""
    status = result.get("status", "") or ""
    msg = f"FAILED {task_name}: {status} {text[:200]}"
    try:
        log_event(state_path, msg)
    except Exception:
        pass
    cmd = (cfg or {}).get("notify_command")
    if not cmd:
        return msg
    try:
        formatted = cmd.format(task=shlex.quote(str(task_name)),
                               status=shlex.quote(str(status)),
                               text=shlex.quote(str(text[:200])))
    except (KeyError, IndexError, ValueError) as exc:
        # 模板写错（例如 {unknown} 或单个 {）时 .format 抛错：
        # 明确记日志、不跑命令、不让调用方以为通知已发出。
        try:
            log_event(state_path,
                      "notify command template error: "
                      f"{type(exc).__name__}: {exc}")
        except Exception:
            pass
        return msg
    try:
        proc = subprocess.run(formatted, shell=True, timeout=30)
        try:
            log_event(state_path,
                      f"notify command rc={proc.returncode} "
                      f"task={task_name}")
        except Exception:
            pass
    except subprocess.TimeoutExpired:
        try:
            log_event(state_path,
                      f"notify command timeout task={task_name}")
        except Exception:
            pass
    except Exception as exc:
        try:
            log_event(state_path,
                      "notify command failed: "
                      f"{type(exc).__name__} task={task_name}")
        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


# ---------------- GitHub PR / CI ----------------
# 这一层只做 host -> GitHub 的桥，不做本地合并：本地合并仍是
# merge_task()，PR 合并由 GitHub 完成，两条线互不干扰。
# 真实 PR 需要两样外部条件，缺一不可，且都不由本轮代码提供：
#   1. 仓库有 git remote origin（本仓目前没有 remote）；
#   2. gh 已登录（gh auth login，PAT 走 Secure Vault，
#      绝不写进文件、state、日志或聊天记录）。
# gh 调用全部走 run_gh()，单测注入假 gh，不依赖网络和登录。

HOST_VERIFY_WORKFLOW = """name: host-verify
on:
  pull_request:
  push:
    branches: [main]
jobs:
  verify:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: python3 -m unittest discover -s tests -v
      - run: python3 -m unittest discover -s cloud/tests -v
"""


def generate_github_workflow(repo):
    """生成 .github/workflows/host-verify.yml（幂等）：
    返回 (changed, path)。文件已存在且内容相同则不改写。"""
    path = os.path.join(repo, ".github", "workflows", "host-verify.yml")
    os.makedirs(os.path.dirname(path), exist_ok=True)
    if os.path.exists(path):
        with open(path, encoding="utf-8") as f:
            if f.read() == HOST_VERIFY_WORKFLOW:
                return False, path
    with open(path, "w", encoding="utf-8") as f:
        f.write(HOST_VERIFY_WORKFLOW)
    return True, path


def run_gh(args, cwd=None, runner=None):
    """gh 调用唯一入口。runner 供单测注入：runner(args) 返回
    一个带 returncode/stdout/stderr 的对象；真实路径走 gh CLI，
    cwd 传给 gh 找仓库上下文（假 runner 忽略 cwd）。"""
    if runner is not None:
        return runner(args)
    return subprocess.run(["gh"] + list(args), cwd=cwd,
                          capture_output=True, text=True)


def _github_remote_url(repo):
    r = git(["remote", "get-url", "origin"], repo)
    url = (r.stdout or "").strip()
    return url if r.returncode == 0 and url else None


def build_pr_body(task):
    """PR 正文：任务名、提示摘要、未解决评审评论。
    只含任务本身的信息，不含密钥。"""
    prompt = (task.get("prompt") or "")[:500]
    lines = [f"host 任务：{task.get('name')}",
             "",
             "任务提示摘要：",
             prompt or "（无提示）",
             ""]
    comments = open_review_comments(task)
    if comments:
        lines.append("评审评论（未解决）：")
        for c in comments:
            lines.append(f"- {c['file']}:{c['line']}: {c['comment']}")
    else:
        lines.append("评审评论：无未解决评论")
    return "\n".join(lines)


def create_pr(repo, task, gh_runner=None):
    """建 PR：先校验任务，再 push 分支，再 gh pr create。
    成功返回 {pr_url, pr_number, branch} 并写回任务记录；
    任何一步失败都抛异常，不编造 URL、不静默失败。
    调用方负责 save_state。"""
    branch = task.get("branch")
    if not branch:
        raise ValueError(f"任务 {task.get('name')} 没有 branch，"
                         f"无法建 PR（可能是没用 worktree 派发的任务）")
    if task.get("status") not in ("completed", "merged"):
        raise ValueError(f"任务 {task.get('name')} 状态是 "
                         f"{task.get('status')}，只有 completed/merged "
                         f"才能建 PR")
    if not _github_remote_url(repo):
        raise RuntimeError(
            "仓库没有 git remote origin，无法 push 分支和建 PR："
            "请先给仓库加 remote（git remote add origin <url>）"
            "并用 gh auth login 登录，PAT 走 Secure Vault，"
            "不要写进文件或聊天记录")
    push = git(["push", "origin", branch], repo)
    if push.returncode != 0:
        raise RuntimeError(f"git push origin {branch} 失败："
                           f"{push.stderr or push.stdout}")
    title = f"host: {task.get('name')}"
    r = run_gh(["pr", "create", "--title", title,
                "--body", build_pr_body(task), "--head", branch],
               cwd=repo, runner=gh_runner)
    out = (r.stdout or "").strip()
    if r.returncode != 0 or not out:
        # 失败时打印 gh 的错误原文，不编造 URL。
        raise RuntimeError(f"gh pr create 失败：{r.stderr or out}")
    pr_url = out.splitlines()[0].strip()
    m = re.search(r"/(\d+)\s*$", pr_url)
    pr_number = int(m.group(1)) if m else None
    task["pr_url"] = pr_url
    task["pr_number"] = pr_number
    return {"pr_url": pr_url, "pr_number": pr_number, "branch": branch}


def parse_pr_checks(output):
    """解析 gh pr checks 输出，兼容两种常见格式：
    制表符格式（name<TAB>state…）和人类可读格式（✓ build pass）。
    返回 (checks, all_pass)：checks 为 [{name, state}]，state
    统一成 pass/fail/pending/skipped/cancelled/neutral；
    没有任何 check 时 all_pass 为 False（空 PR 不能算全过）。"""
    checks = []
    for raw in (output or "").splitlines():
        line = raw.strip()
        if not line or "no checks" in line.lower():
            continue
        if line.lower().startswith("name"):
            continue
        if "\t" in line:
            parts = [p.strip() for p in line.split("\t") if p.strip()]
            if len(parts) < 2:
                continue
            name, state_raw = parts[0], parts[1].lower()
        else:
            m = re.search(r"\b(pass|passing|fail|failing|failed|pending|"
                          r"queued|waiting|in_progress|skipped|cancelled|"
                          r"neutral)\b", line.lower())
            if not m:
                continue
            state_raw = m.group(1)
            name = line[:m.start()].strip(" ✓✗●○-").strip()
            if not name:
                name = line
        state = {"passing": "pass", "failing": "fail", "failed": "fail",
                 "queued": "pending", "waiting": "pending",
                 "in_progress": "pending"}.get(state_raw, state_raw)
        checks.append({"name": name, "state": state})
    all_pass = bool(checks) and all(c["state"] == "pass" for c in checks)
    return checks, all_pass


def get_pr_checks(task, gh_runner=None, repo=None):
    """查 PR 的 CI 状态。任务须已有 pr_number，否则报错。"""
    number = task.get("pr_number")
    if not number:
        raise ValueError(f"任务 {task.get('name')} 还没有 PR 编号，"
                         f"先跑 pr 建 PR")
    r = run_gh(["pr", "checks", str(number)], cwd=repo, runner=gh_runner)
    if r.returncode != 0:
        raise RuntimeError(f"gh pr checks 失败：{r.stderr or r.stdout}")
    checks, all_pass = parse_pr_checks(r.stdout or "")
    return {"pr_url": task.get("pr_url"), "checks": checks,
            "all_pass": all_pass}


def merge_pr(task, gh_runner=None, repo=None):
    """PR 合并门：先查 checks，未全过拒绝且不调 merge；
    全过后调 gh pr merge --merge，成功才把任务标 merged。
    不动本地主分支——合并由 GitHub 完成，本地分支/主仓
    不由这条命令改（和本地 merge_task() 是两条线）。"""
    info = get_pr_checks(task, gh_runner=gh_runner, repo=repo)
    if not info["all_pass"]:
        bad = [c for c in info["checks"] if c["state"] != "pass"]
        return False, (f"PR checks 未全过，拒绝合并："
                       f"{[(c['name'], c['state']) for c in bad]}")
    r = run_gh(["pr", "merge", str(task["pr_number"]), "--merge"],
               cwd=repo, runner=gh_runner)
    if r.returncode != 0:
        return False, f"gh pr merge 失败：{r.stderr or r.stdout}"
    task["status"] = "merged"
    return True, f"PR #{task['pr_number']} 已合并：{task.get('pr_url')}"




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 _seed_worktree(cwd, seed_dir):
    """把暂存附件拷进 worktree（dispatch/goal 共用，turn 开始前）。

    seed_dir 是服务端写好的暂存目录（只含常规文件，无符号链接）：
    顶层 .git 与符号链接一律跳过；子目录整树拷入，同名文件覆盖
    （worktree 刚从 repo 建出来，覆盖即用户文件优先）。"""
    import shutil
    seed_dir = os.path.abspath(seed_dir)
    if not os.path.isdir(seed_dir):
        raise RuntimeError(f"seed-dir 不是目录: {seed_dir}")
    for entry in sorted(os.listdir(seed_dir)):
        if entry == ".git":
            continue
        src = os.path.join(seed_dir, entry)
        dst = os.path.join(cwd, entry)
        if os.path.islink(src):
            continue
        if os.path.isdir(src):
            shutil.copytree(src, dst, dirs_exist_ok=True, symlinks=False)
        else:
            shutil.copy2(src, dst)


def dispatch(server, state, name, prompt, repo, model, effort, sandbox,
             approval_policy, use_worktree=True, parent_task=None,
             depends_on=None, priority=0, max_retries=0, seed_dir=None):
    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):
        cand = os.path.join(HERE, repo)
        if os.path.exists(cand):
            repo = cand
    if not sandbox:
        sandbox = "workspace-write"
    cwd = repo
    branch = None
    if use_worktree and repo:
        cwd, branch = make_worktree(repo, name)
    if seed_dir:
        # 附件只能进 worktree：直接以 repo 为 cwd 时拷进去等于污染
        # 仓库，必须 loud fail，绝不静默放行。
        if not use_worktree or not repo or not cwd:
            raise RuntimeError(
                "seed-dir 需要 worktree，当前派发没有建 worktree")
        _seed_worktree(cwd, seed_dir)
    thread_id = server.create_thread(cwd, model=model, sandbox=sandbox,
                                     approval_policy=approval_policy)
    task = {"name": name, "thread_id": thread_id, "cwd": cwd,
            "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


def run_task_turn(server, state, task, prompt, model, effort,
                  sandbox=None, timeout=None, interrupt_file=None):
    # task 自带 sandbox 时优先用它（只读评审任务不能被全局默认覆盖），
    # 否则用调用方显式传的 sandbox；两者都没有时不传 sandboxPolicy，
    # 保留 thread 原有设置（旧任务没有 sandbox 记录时尤其重要）。
    # timeout 为 None 时不传给 server（沿用 server 默认 600），
    # 这样旧 FakeServer（不接受 timeout 参数）不受影响；显式
    # 传了才透传，0 由 server 侧解释为不限时。interrupt_file 同理：
    # 不传时不进 server 参数，旧 FakeServer 不受影响；传了时
    # 属主在等 turn 时每秒轮询它（见 _interrupt_request_path）。
    use_sandbox = task.get("sandbox") or sandbox
    started = time.time()
    turn_kwargs = {}
    if timeout is not None:
        turn_kwargs["timeout"] = timeout
    if interrupt_file is not None:
        turn_kwargs["interrupt_file"] = interrupt_file
    try:
        result = server.run_turn(task["thread_id"], prompt, model=model,
                                 effort=effort, sandbox=use_sandbox,
                                 **turn_kwargs)
    except TypeError as exc:
        # 兼容旧 server：不接受 interrupt_file 参数时退回不传它
        # 再跑一次（goal 单测的 FakeServer 就是这款）；其它
        # TypeError 一律不吞，原样抛出。
        if "interrupt_file" not in str(exc):
            raise
        turn_kwargs.pop("interrupt_file", None)
        result = server.run_turn(task["thread_id"], prompt, model=model,
                                 effort=effort, sandbox=use_sandbox,
                                 **turn_kwargs)
    finished = time.time()
    record_turn_usage(state, failed=(result["status"] != "completed"))
    task["turns"].append({"prompt": prompt, "result": result,
                          "started_at": started, "finished_at": finished})
    task["last_turn_at"] = finished
    # 对齐 Codex App：超时 / 被中断不是失败，是可续跑的中断态；
    # 只有真正的 failed 才记 failed，recover 才能区分两者。
    if result["status"] == "completed":
        task["status"] = "completed"
    elif result["status"] in ("timeout", "interrupted"):
        task["status"] = "interrupted"
    else:
        task["status"] = "failed"
    task["last_text"] = result.get("text", "")
    return result


def run_task_turn_with_retry(server, state, task, prompt, model, effort,
                             sandbox=None, cfg=None, timeout=None,
                             interrupt_file=None):
    """失败自动重试：默认 max_retries=0 时行为不变（只跑一次）；
    配了重试时，失败后同 prompt 再跑，最多 max_retries 次，
    每次重试前先查预算，超预算直接停，不烧钱。"""
    max_retries = int(task.get("max_retries") or 0)
    result = None
    usage = 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, timeout=timeout,
                               interrupt_file=interrupt_file)
        # 各次尝试的真实用量都要算（失败尝试花的钱也是钱），
        # 最后一次的结果里带汇总，不只带最后一次。
        usage = _add_usage(usage, result.get("usage"))
        if result["status"] == "completed":
            break
    if usage is not None and result is not None:
        result = dict(result, usage=usage)
    return result


GOAL_DONE_MARKER = "目标已达成"


def _llm_declared_done(text):
    """模型是否宣告目标达成：回合文本里单独一行是标记行。

    只有完整跑完的回合才查标记（调用方保证）：超时/中断回合的
    部分文本即使含标记也不算，否则没做完也会提前结束。

    行尾句号要容忍：提示本身让模型输出“目标已达成。”（带句
    号），若检测要求整行恰好等于无句号的标记，模型照做也永
    远不算达成、循环空转（2026-10-27 实弹 292 轮教训）。所以
    比对前先去行尾空白与句末标点，但行内其它文字一律不算
    （“xx目标已达成”“目标已达成，继续”仍不算，单独一行要
    求不动）。"""
    if not text:
        return False
    punct = "。！!？?；;，,、.．"
    for line in text.splitlines():
        if line.strip().rstrip(punct).strip() == GOAL_DONE_MARKER:
            return True
    return False


def run_goal_loop(server, state, state_path, task, goal_text, verify,
                  max_rounds, model, effort, sandbox, cfg=None,
                  timeout=None, verify_timeout=600, start_round=1,
                  last_verify="", last_turn_status=None, last_turn_text=""):
    """目标循环（长任务版，对齐 Codex App 的可续跑语义）。

    两种验收形态（2026-10-27 起，console goal 模式走新形态）：
    旧形态 verify 非空：每轮 turn → 跑验收命令 → v_rc==0 达标，
    max_rounds 穷尽则 unmet。
    新形态 verify 为空：不跑验收命令，达标与否由模型自己在回合
    末尾单独一行输出“目标已达成”宣告；max_rounds=0 表示不限轮次，
    一直跑到模型宣告、预算耗尽或用户取消为止（与 turn_timeout=0
    不限时同约定）。轮次提示里会把标记约定写明，模型才知道怎么
    宣告；同一 thread 保留历史，续轮提示可以很短。
    每轮结束立即落盘（检查点），进程被杀 / VM 回收后可用
    goal-resume 从 goal_rounds_done 的下一轮续跑，而不是从头重来。
    与云端 worker 的解析契约（cloud/sandbox.py）：
    结论行必须是 "goal met after rounds" / "goal unmet after rounds"，
    每轮行必须是 "round N: turn=... verify_rc=... verify_out=..."，
    （无验收时 verify_rc 记 -，verify_out 为空），
    用量行必须是 "usage input=... cached=... ..." 格式，改动时
    不能动这三行；两种结局都 exit 0（main 不抛异常即 0）。
    预算 stop 是第三种结局：打印 "goal stopped after rounds: ..."
    且任务记 interrupted（明天预算重置后可续跑），不冒充 unmet。
    goal-unmet 穷尽轮次后不允许 goal-resume（按既有规则须新建
    任务），goal-resume 只接受 goal-running / interrupted / running。
    """
    task["goal"] = goal_text
    task["verify_command"] = verify
    task["goal_max_rounds"] = int(max_rounds)
    task["goal_rounds_done"] = int(start_round) - 1
    task["goal_last_verify"] = last_verify
    # 循环前先落盘一次：否则第一轮 turn 进行中被杀时，磁盘上的
    # 任务还没有 goal/verify_command，goal-resume 会因“没有记录”
    # 而拒绝——断点续跑必须在第一轮开始前就已可续跑。
    task["status"] = "goal-running"
    save_state(state_path, state)
    ok = False
    budget_stopped = False
    goal_usage = None
    # max_rounds=0 是不限轮次：while 循环只在模型宣告、预算耗尽
    # 或取消时结束；非 0 时按旧语义穷尽轮次即 unmet。
    round_no = int(start_round) - 1
    while True:
        round_no += 1
        if int(max_rounds) != 0 and round_no > int(max_rounds):
            break
        if cfg is not None and budget_exceeded(state, cfg):
            budget_stopped = True
            print(f"goal stopped: 预算已用完（max_daily_turns），"
                  f"第 {round_no} 轮未开始，任务可续跑")
            break
        if round_no == 1 and not last_verify and not last_turn_status:
            prompt = goal_text
            if verify is None:
                # 新形态首轮必须把标记约定写进提示，否则模型不知
                # 道怎么宣告完成，循环永远等不到结束信号。
                prompt = (f"{goal_text}\n\n"
                          f"完成方式：这是目标循环，你可以多轮工作；"
                          f"当你确认目标已完全达成时，在回复末尾单独一行输出："
                          f"{GOAL_DONE_MARKER}。未达成时不要输出这一行。")
        elif last_turn_status in ("timeout", "interrupted"):
            # 上轮回合根本没跑完：说清是回合中断，不是验收失败，
            # 否则 agent 会对着验收输出修一个不存在的问题。
            prompt = (f"上一轮回合未完成（状态：{last_turn_status}），"
                      f"请继续完成目标，不要重复已完成的部分。\n"
                      f"目标：{goal_text}\n")
            if verify is not None:
                prompt += (f"验收命令：{verify}\n"
                           f"上一轮验收输出：\n{last_verify}\n")
            else:
                prompt += (f"确认目标已完全达成时，在回复末尾单独一行输出："
                           f"{GOAL_DONE_MARKER}。\n")
            prompt += f"上一轮部分文本：\n{last_turn_text[:1000]}"
        elif verify is None:
            # 新形态续轮：同一 thread 已有历史，只需说明“还没
            # 宣告达成、继续做”，并重申标记约定。
            prompt = (f"上一轮未宣告目标达成，请继续完成目标，"
                      f"不要重复已完成的部分。\n"
                      f"目标：{goal_text}\n"
                      f"确认目标已完全达成时，在回复末尾单独一行输出："
                      f"{GOAL_DONE_MARKER}。")
        else:
            # 生产教训：只说“验收未通过” agent 不知道改什么，
            # 2026-10-25 实测 prod-retry 3 轮都原地不动。
            # 必须把验收命令和失败输出喂回去，才是可用的目标循环。
            prompt = (f"上一轮验收未通过，请针对性修复，不要重复已完成的部分。\n"
                      f"目标：{goal_text}\n"
                      f"验收命令：{verify}\n"
                      f"上一轮验收输出：\n{last_verify}")
        result = run_task_turn(server, state, task, prompt,
                               model, effort, sandbox, timeout=timeout,
                               interrupt_file=_interrupt_request_path(
                                   state_path, task["name"])
                               if task.get("name") else None)
        goal_usage = _add_usage(goal_usage, result.get("usage"))
        last_turn_status = result["status"]
        last_turn_text = result.get("text", "") or ""
        if verify is None:
            # 新形态没有验收命令可跑，v_rc 记 -（轮次行格式不变，
            # 云端解析只读 verify_out 做 unmet 详情，不受影响）。
            v_rc, v_out = None, ""
        else:
            try:
                v_rc, v_out = None, None
                if verify_timeout and verify_timeout > 0:
                    v = subprocess.run(verify, shell=True, cwd=task["cwd"],
                                       capture_output=True, text=True,
                                       timeout=verify_timeout)
                else:
                    v = subprocess.run(verify, shell=True, cwd=task["cwd"],
                                       capture_output=True, text=True)
                v_rc = v.returncode
                v_out = (v.stdout + v.stderr).strip()[:1000] or \
                    f"exit code {v.returncode}"
            except subprocess.TimeoutExpired as exc:
                # 验测本身也可能很长（测试套件）：超时是验测失败，
                # 不是进程挂死，喂回下一轮让 agent 缩小验测范围。
                v_rc = 124
                out = (exc.stdout or "") if isinstance(exc.stdout, str) \
                    else ""
                err = (exc.stderr or "") if isinstance(exc.stderr, str) \
                    else ""
                v_out = (f"verify timeout after {verify_timeout}s: "
                         f"{(out + err).strip()[:900]}")
        last_verify = v_out
        task["goal_rounds_done"] = round_no
        task["goal_last_verify"] = last_verify
        task["goal_last_turn_status"] = last_turn_status
        # 检查点：非终局轮次先记 goal-running，终局轮次由下面
        # 统一改写；这样被杀后 state 里能看出循环没跑完。
        task["status"] = "goal-running"
        save_state(state_path, state)
        print(f"round {round_no}: turn={result['status']} "
              f"verify_rc={v_rc if v_rc is not None else '-'} "
              f"verify_out={last_verify[:300]}")
        if verify is None:
            # 模型自决：只有完整跑完的回合里的标记行才算数。
            if (result["status"] == "completed"
                    and _llm_declared_done(last_turn_text)):
                ok = True
                break
        elif v_rc == 0:
            ok = True
            break
    if budget_stopped:
        task["status"] = "interrupted"
        task["goal_stopped_reason"] = "budget exceeded: max_daily_turns"
        # goal 预算中断以前只打印：与 dispatch 前置门同款，
        # 补一次 budget-exceeded 通知（没配命令时只写日志）。
        notify_failure(state_path, cfg, task.get("name") or "goal",
                       {"status": "budget-exceeded",
                        "text": "budget exceeded: max_daily_turns"})
    else:
        task["status"] = "goal-met" if ok else "goal-unmet"
    if task["status"] == "goal-unmet":
        notify_failure(state_path, cfg, task["name"],
                       {"status": "goal-unmet", "text": last_verify[:200]})
    save_state(state_path, state)
    if budget_stopped:
        print("goal stopped after rounds: budget exceeded")
    else:
        print(f"goal {'met' if ok else 'unmet'} after rounds")
    if goal_usage:
        print(f"usage input={goal_usage['input_tokens']} "
              f"cached={goal_usage['cached_input_tokens']} "
              f"cache_write={goal_usage['cache_write_input_tokens']} "
              f"output={goal_usage['output_tokens']} "
              f"reasoning={goal_usage['reasoning_output_tokens']} "
              f"total={goal_usage['total_tokens']} "
              f"completions={goal_usage['completions']}")
    return {"ok": ok, "budget_stopped": budget_stopped,
            "rounds_done": task["goal_rounds_done"],
            "last_verify": last_verify, "usage": goal_usage}


def _load_config_file(path):
    if path and os.path.exists(path):
        try:
            return json.load(open(path))
        except Exception as e:
            raise RuntimeError(f"配置文件读失败 {path}: {e}")
    return {}


def _resolve_path(base_dir, p):
    if not p:
        return p
    if os.path.isabs(p):
        return p
    return os.path.join(base_dir, p)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--codex-home", default=None)
    ap.add_argument("--codex-bin", default=None)
    ap.add_argument("--config", default=None,
                    help="项目配置文件，默认 host.py 同目录的 host-config.json（若存在）")
    ap.add_argument("--state", default=None)
    ap.add_argument("--model", default=None)
    ap.add_argument("--model-provider", default=None,
                    help="模型 provider 覆盖：sub2api 时以 -c 注入网关"
                         " provider（不改 config.toml）；不传则走配置")
    ap.add_argument("--effort", default=None)
    ap.add_argument("--sandbox", default=None)
    ap.add_argument("--approval-policy", default=None)
    ap.add_argument("--approval", default=None, choices=["auto", "deny"])
    ap.add_argument("--turn-timeout", type=int, default=None,
                    help="单轮等待上限（秒）；0 表示不限时、对齐 Codex App "
                         "等 turn/completed 的语义；不传时用配置 "
                         "turn_timeout，未配则默认 600")
    ap.add_argument("--progress-interval", type=float, default=None,
                    help="长等待流式进度间隔（秒）；每满一个间隔往 stderr "
                         "打一行 turn progress 并往链路文件记一条 "
                         "kind=progress；0 表示关闭进度输出；不传时用配置 "
                         "progress_interval，未配则默认 30")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("init")
    p.add_argument("--repo", default=None)
    p.add_argument("--force", action="store_true")
    sub.add_parser("config")
    p = sub.add_parser("dispatch")
    p.add_argument("--name", required=True)
    p.add_argument("--task", required=True)
    p.add_argument("--repo", default=None)
    p.add_argument("--seed-dir", default=None,
                   help="附件暂存目录：派发时拷进 worktree 再跑 turn")
    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")
    sub.add_parser("health")
    sub.add_parser("restore-state")
    p = sub.add_parser("recover")
    p.add_argument("--older-than", type=int, default=3600,
                   help="最后回合超过多少秒未更新才判定中断，默认 3600")
    p = sub.add_parser("daemon")
    p.add_argument("--interval", type=int, default=300,
                   help="每轮间隔秒数，默认 300")
    p.add_argument("--foreground", action="store_true",
                   help="前台跑循环（测试用，不双 fork）")
    sub.add_parser("daemon-status")
    sub.add_parser("daemon-stop")
    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)
    p = sub.add_parser("compact"); p.add_argument("--task", required=True)
    p = sub.add_parser("code-review"); p.add_argument("--task", required=True)
    p.add_argument("--base-branch", default=None,
                   help="评审目标：与该分支的差异（baseBranch）")
    p.add_argument("--commit", default=None,
                   help="评审目标：指定提交 sha（commit）")
    p.add_argument("--instructions", default=None,
                   help="评审目标：自定义评审指令（custom）")
    p = sub.add_parser("run"); p.add_argument("--task", required=True); p.add_argument("--prompt", required=True)
    p = sub.add_parser("interrupt"); p.add_argument("--task", required=True)
    p = sub.add_parser("delegate")
    p.add_argument("--parent-task", required=True); p.add_argument("--task", required=True)
    p.add_argument("--name", required=True); p.add_argument("--report-to-parent", action="store_true")
    p = sub.add_parser("goal")
    p.add_argument("--name", required=True); p.add_argument("--task", required=True)
    p.add_argument("--repo", default=None)
    p.add_argument("--seed-dir", default=None,
                   help="附件暂存目录：派发时拷进 worktree 再跑 turn")
    p.add_argument("--verify", default=None,
                   help="验收命令；不传时走模型自决：模型在回合末尾"
                        "单独一行输出“目标已达成”即达标")
    p.add_argument("--max-rounds", type=int, default=3,
                   help="轮次上限；0 表示不限轮次（模型宣告完成才结束），"
                        "默认 3")
    p.add_argument("--verify-timeout", type=int, default=600,
                   help="验测命令上限（秒）；0 表示不限时，默认 600")
    p = sub.add_parser("goal-resume")
    p.add_argument("--task", required=True)
    p.add_argument("--goal-text", default=None,
                   help="新目标文本：旧任务接续 goal 模式时用——跳过状态与"
                        "goal 记录守卫，在旧线程上从第 1 轮开新目标循环"
                        "（新语义：无验收、不传 --max-rounds 时不限轮次）")
    p.add_argument("--max-rounds", type=int, default=None,
                   help="不传时沿用原任务的 goal_max_rounds；传了则覆盖总轮次")
    p.add_argument("--verify-timeout", type=int, default=600,
                   help="验测命令上限（秒）；0 表示不限时，默认 600")
    p = sub.add_parser("parallel"); p.add_argument("--jobs", required=True)
    p = sub.add_parser("automations"); p.add_argument("--add", action="store_true")
    p.add_argument("--name"); p.add_argument("--task"); p.add_argument("--repo")
    p.add_argument("--interval", type=int, default=86400)
    p.add_argument("--cron-line", action="store_true",
                   help="只打印 run-due 的 crontab 行，不装机")
    p.add_argument("--install-cron", action="store_true",
                   help="把 run-due 写进当前用户 crontab（幂等，已有则不重复）")
    p.add_argument("--cron-every-minutes", type=int, default=30)
    sub.add_parser("run-due")
    sub.add_parser("github-workflow")
    p = sub.add_parser("pr"); p.add_argument("--task", required=True)
    p.add_argument("--repo", default=None)
    p = sub.add_parser("pr-checks"); p.add_argument("--task", required=True)
    p.add_argument("--repo", default=None)
    p = sub.add_parser("pr-merge"); p.add_argument("--task", required=True)
    p.add_argument("--repo", default=None)

    args = ap.parse_args()
    # 配置解析：CLI > 配置文件 > 内置默认。配置文件里绝不放密钥，
    # API key 只走环境变量（中转站用 XCM_API_KEY 那种），这是硬规则。
    config_path = args.config or os.path.join(HERE, "host-config.json")
    cfg = _load_config_file(config_path) if os.path.exists(config_path) else {}
    cfg_dir = os.path.dirname(os.path.abspath(config_path))
    if args.cmd == "init":
        if os.path.exists(config_path) and not args.force:
            print(f"配置文件已存在：{config_path}，加 --force 才会覆盖")
            return
        init_cfg = {
            "repo": args.repo or cfg.get("repo") or "",
            "model": "gpt-5.6-terra", "effort": "low",
            "sandbox": "workspace-write", "approval_policy": "never",
            "approval": "deny", "codex_home": "",
            "state": "host-state.json",
            "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)
        print(f"配置文件已写：{config_path}，请按需改 repo/model 后再用")
        return
    # 有效配置（init/config 之外的命令都用它）
    args.model = args.model or cfg.get("model") or "gpt-5.6-terra"
    args.effort = args.effort or cfg.get("effort") or "low"
    args.sandbox = args.sandbox or cfg.get("sandbox")
    args.approval_policy = args.approval_policy or cfg.get("approval_policy") or "never"
    args.approval = args.approval or cfg.get("approval") or "deny"
    # turn-timeout 解析：CLI > 配置 > 默认 600；0 保持为 0（不限时），
    # 不能用 or 串联，否则 0 会被当成缺失而吞掉。
    if args.turn_timeout is None:
        args.turn_timeout = cfg.get("turn_timeout")
    if args.turn_timeout is None:
        args.turn_timeout = 600
    # progress-interval 解析：同款 CLI > 配置 > 默认 30；0 保持为 0
    #（关闭进度输出），不能用 or 串联。
    if args.progress_interval is None:
        args.progress_interval = cfg.get("progress_interval")
    if args.progress_interval is None:
        args.progress_interval = 30
    args.codex_home = args.codex_home or _resolve_path(cfg_dir, cfg.get("codex_home"))
    args.codex_bin = args.codex_bin or _resolve_path(cfg_dir, cfg.get("codex_bin"))
    args.state = args.state or _resolve_path(cfg_dir, cfg.get("state")) or DEFAULT_STATE
    if args.cmd == "config":
        print(json.dumps({
            "config_file": config_path if os.path.exists(config_path) else None,
            "model": args.model, "effort": args.effort,
            "sandbox": args.sandbox, "approval_policy": args.approval_policy,
            "approval": args.approval, "codex_home": args.codex_home,
            "codex_bin": args.codex_bin, "state": args.state,
            "repo": cfg.get("repo"),
        }, indent=2, ensure_ascii=False))
        return
    # dispatch/goal/automations 的 repo 缺省用配置里的 repo
    if args.cmd in ("dispatch", "goal"):
        if not args.repo:
            args.repo = cfg.get("repo")
        if not args.repo:
            raise RuntimeError("缺少 --repo，也没有在 host-config.json 里配 repo")
    if args.cmd == "automations" and (args.cron_line or args.install_cron):
        # cron 行生成：只打印 / 装机，不跑模型，不花钱。
        bin_path = args.codex_bin or default_codex_bin()
        log_path = os.path.join(os.path.dirname(os.path.abspath(args.state)),
                                "host-run.log")
        line = (f"*/{args.cron_every_minutes} * * * * cd {HERE} && "
                f"CODEX_BIN={bin_path} python3 host.py --config {config_path} "
                f"run-due >> {log_path} 2>&1  # codex-host-run-due")
        if args.cron_line:
            print(line)
            print("注意：这行本身不生效，要装进 crontab 才生效；"
                  "VM 会被平台回收，crontab 不一定能活过回收，"
                  "常驻机器用 crontab，VM 上更稳的是 Muse cron 定时调 run-due。")
            return
        if args.install_cron:
            cur = subprocess.run(["crontab", "-l"], capture_output=True,
                                 text=True).stdout or ""
            if "codex-host-run-due" in cur:
                print("crontab 里已有 codex-host-run-due，未重复安装")
                return
            new = (cur.rstrip("\n") + "\n" + line + "\n") if cur.strip() else line + "\n"
            subprocess.run(["crontab", "-"], input=new, text=True, check=True)
            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
    if args.cmd == "github-workflow":
        repo = cfg.get("repo")
        if not repo:
            print("缺少 repo：请在配置里配 repo")
            raise SystemExit(1)
        changed, path = generate_github_workflow(repo)
        print(json.dumps({"path": path,
                          "changed": changed}, indent=2, ensure_ascii=False))
        return
    if args.cmd in ("pr", "pr-checks", "pr-merge"):
        if args.task not in state["tasks"]:
            print(f"任务不存在：{args.task}")
            raise SystemExit(1)
        t = state["tasks"][args.task]
        repo = args.repo or t.get("repo") or cfg.get("repo")
        try:
            if args.cmd == "pr":
                if not repo:
                    print("缺少 repo：任务和配置里都没有 repo")
                    raise SystemExit(1)
                info = create_pr(repo, t)
                save_state(args.state, state)
                print(json.dumps(info, indent=2, ensure_ascii=False))
            elif args.cmd == "pr-checks":
                info = get_pr_checks(t, repo=repo)
                print(json.dumps(info, indent=2, ensure_ascii=False))
            else:
                ok, msg = merge_pr(t, repo=repo)
                if ok:
                    save_state(args.state, state)
                    print(json.dumps({"merged": True, "message": msg},
                                     indent=2, ensure_ascii=False))
                else:
                    print(msg)
                    raise SystemExit(1)
        except SystemExit:
            raise
        except Exception as e:
            # 失败时打印错误原文（gh/git 的 stderr 已包含在异常里），
            # 不编造 PR URL，不静默失败。
            print(str(e))
            raise SystemExit(1)
        return
    if args.cmd == "health":
        report = run_health_check(args.state, cfg, args.codex_bin or default_codex_bin(),
                                  args.codex_home, args.approval)
        print(json.dumps(report, indent=2, ensure_ascii=False))
        if not report["ok"]:
            raise SystemExit(1)
        return
    if args.cmd == "restore-state":
        ok, msg = restore_state(args.state)
        print(msg)
        if not ok:
            raise SystemExit(1)
        return
    if args.cmd == "recover":
        strict, err = load_state_strict(args.state)
        if err is not None:
            print(f"state 已损坏，recover 不会覆盖它，先跑 restore-state：{err}")
            raise SystemExit(1)
        recovered = recover_interrupted(state, args.state,
                                        older_than=args.older_than, cfg=cfg)
        if recovered:
            save_state(args.state, state)
        print(json.dumps({"recovered": recovered}, indent=2, ensure_ascii=False))
        return
    if args.cmd == "daemon-status":
        pidfile = os.path.join(os.path.dirname(os.path.abspath(args.state)),
                               "host-daemon.pid")
        logfile = os.path.join(os.path.dirname(os.path.abspath(args.state)),
                               "host-daemon.log")
        alive = False
        pid = None
        if os.path.exists(pidfile):
            try:
                pid = int(open(pidfile).read().strip())
                os.kill(pid, 0)
                alive = True
            except Exception:
                alive = False
        tail = ""
        if os.path.exists(logfile):
            try:
                tail = "\n".join(open(logfile).read().splitlines()[-5:])
            except Exception:
                pass
        print(json.dumps({"alive": alive, "pid": pid, "pidfile": pidfile,
                          "log_tail": tail}, indent=2, ensure_ascii=False))
        return
    if args.cmd == "daemon-stop":
        pidfile = os.path.join(os.path.dirname(os.path.abspath(args.state)),
                               "host-daemon.pid")
        if not os.path.exists(pidfile):
            print("daemon 未运行（无 pidfile）")
            return
        try:
            pid = int(open(pidfile).read().strip())
        except Exception:
            print("pidfile 损坏，直接删除")
            os.remove(pidfile)
            return
        try:
            os.kill(pid, 15)
            for _ in range(10):
                time.sleep(0.5)
                try:
                    os.kill(pid, 0)
                except Exception:
                    break
            else:
                os.kill(pid, 9)
        except Exception:
            pass
        try:
            os.remove(pidfile)
        except Exception:
            pass
        print(f"daemon 已停：pid {pid}")
        return
    if args.cmd == "daemon":
        state_dir = os.path.dirname(os.path.abspath(args.state))
        pidfile = os.path.join(state_dir, "host-daemon.pid")
        logfile = os.path.join(state_dir, "host-daemon.log")
        if not args.foreground:
            if os.path.exists(pidfile):
                try:
                    old_pid = int(open(pidfile).read().strip())
                    os.kill(old_pid, 0)
                    print(f"daemon 已在运行：pid {old_pid}，先 daemon-stop 再起")
                    return
                except Exception:
                    pass
            if not daemonize(pidfile, logfile):
                print(f"daemon 已在后台启动：pidfile {pidfile}，log {logfile}")
                return
        # 子进程 / 前台：循环跑 recover + run-due，每轮都是独立
        # 子进程调 host.py CLI，这样一轮崩了不带走 daemon 本体，
        # 且 run-due 的模型回合数受自动化本身和预算控制。
        import signal
        stop = {"flag": False}

        def _on_term(signum, frame):
            stop["flag"] = True
        signal.signal(signal.SIGTERM, _on_term)
        signal.signal(signal.SIGINT, _on_term)
        print(f"host daemon started pid={os.getpid()} interval={args.interval}s "
              f"config={config_path} state={args.state}", flush=True)
        while not stop["flag"]:
            ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
            print(f"{ts} daemon tick: recover + run-due", flush=True)
            for subcmd in (["recover", "--older-than", "3600"], ["run-due"]):
                try:
                    subprocess.run([sys.executable, os.path.join(HERE, "host.py"),
                                    "--config", config_path] + subcmd,
                                   timeout=3600)
                except Exception as e:
                    print(f"{ts} daemon subcommand {subcmd} 失败：{e}", flush=True)
            for _ in range(args.interval):
                if stop["flag"]:
                    break
                time.sleep(1)
        try:
            os.remove(pidfile)
        except Exception:
            pass
        print("host daemon stopped", flush=True)
        return
    srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home,
                 approval=args.approval,
                 progress_interval=args.progress_interval,
                 model_provider=args.model_provider)
    # 实时链路：只给跑 turn 的命令设链路文件（dispatch/run/goal/
    # delegate）；只读命令不设。名字映射与 worker 派发一致：
    # dispatch/goal/delegate 用 --name，run 用 --task。
    _trace_name = None
    if args.cmd in ("dispatch", "goal", "delegate"):
        _trace_name = getattr(args, "name", None)
    elif args.cmd == "run":
        _trace_name = getattr(args, "task", None)
    if _trace_name:
        srv.set_trace_file(args.state, _trace_name)
    try:
        srv.initialize()
        if args.cmd == "dispatch":
            if budget_exceeded(state, cfg):
                print("预算已用完（max_daily_turns），本次不派发，见 doctor 的 usage")
                # 前置预算门以前只打印：配了 notify_command 的
                # 用户收不到任何通知。补一次 budget-exceeded
                # 通知；没配命令时 notify_failure 只写日志，
                # 不会无配置刷屏。
                notify_failure(args.state, cfg, args.name,
                               {"status": "budget-exceeded",
                                "text": "budget exceeded: max_daily_turns"})
                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,
                            depends_on=deps, priority=args.priority,
                            max_retries=args.max_retries,
                            seed_dir=args.seed_dir)
            # 生产教训：先落盘任务登记再跑 turn，中途被杀/超时后
            # 任务还在 host-state 里，可用 run 续跑；否则 turn 中断
            # 就等于任务丢了（2026-10-25 委派评审 300 秒超时实测）。
            save_state(args.state, state)
            result = run_task_turn_with_retry(srv, state, task, args.task,
                                              args.model, args.effort,
                                              args.sandbox, cfg=cfg,
                                              timeout=args.turn_timeout,
                                              interrupt_file=_interrupt_request_path(args.state, args.name))
            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"],
                              "text": result.get("text", ""),
                              "error": result.get("error"),
                              "turn_id": result.get("turn_id"),
                              "root_turn_id": result.get("root_turn_id"),
                              "usage": result.get("usage")},
                             indent=2, ensure_ascii=False))
        elif args.cmd == "list":
            for name, t in state["tasks"].items():
                print(f"{name}: thread={t['thread_id'][:8]} status={t['status']} "
                      f"cwd={t['cwd']} branch={t.get('branch')}")
            print(f"app-server threads: "
                  f"{json.dumps(srv.list_threads(), ensure_ascii=False)[:500]}")
        elif args.cmd == "status":
            t = state["tasks"][args.task]
            print(json.dumps(t, indent=2, ensure_ascii=False))
        elif args.cmd == "review":
            t = state["tasks"][args.task]
            d = task_diff(t["cwd"])
            t["diff"] = d
            save_state(args.state, state)
            print(f"status:\n{d['status']}\nuntracked:\n{d['untracked']}\n"
                  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
            # 合并 + 验测 + 回滚：验测命令默认仓库有 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
            srv.resume_thread(t["thread_id"])
            result = run_task_turn(srv, state, t, build_fix_prompt(t),
                                   args.model, args.effort, args.sandbox,
                                   timeout=args.turn_timeout,
                                   interrupt_file=_interrupt_request_path(args.state, args.task))
            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)
        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"])
                git(["branch", "-D", t["branch"]], t["repo"])
            t["status"] = "discarded"
            save_state(args.state, state)
            print(f"discarded {args.task}")
        elif args.cmd == "compact":
            t = state["tasks"][args.task]
            # 注意：compact 要求 thread 已在当前 server 里加载，
            # 直接对未加载的旧 thread 调会报 thread not found，
            # 所以先 resume 再 compact，这是协议层面的先后顺序。
            srv.resume_thread(t["thread_id"])
            resp = srv.request("thread/compact/start",
                               {"threadId": t["thread_id"]}, timeout=120)
            print(f"compact {args.task}: {resp}")
        elif args.cmd == "code-review":
            # 注意命名：host.py 已有另一条线的 review 子命令
            # （PR 评审评论），本命令换名 code-review，不覆盖它。
            # 目标优先级：commit > instructions > base-branch >
            # 缺省 uncommittedChanges；与 cloud/review.py 的
            # build_review_target 同款组合，非法组合直接报错。
            # compact 同款原因先 resume 再 review（未加载的旧
            # thread 直接 review 会报 thread not found）。
            # app-server 返回 error 时只打印失败并以非零退出，
            # 绝不把失败结果当成评审结论。
            t = state["tasks"][args.task]
            if args.commit:
                target = {"type": "commit", "sha": args.commit}
            elif args.instructions:
                target = {"type": "custom",
                          "instructions": args.instructions}
            elif args.base_branch:
                target = {"type": "baseBranch",
                          "branch": args.base_branch}
            else:
                target = {"type": "uncommittedChanges"}
            try:
                srv.resume_thread(t["thread_id"])
                result = srv.start_review(t["thread_id"], target)
            except Exception as exc:
                print(f"code-review failed: {exc}")
                raise SystemExit(1)
            print(f"code-review {args.task}: {result}")
        elif args.cmd == "archive":
            t = state["tasks"][args.task]
            srv.archive_thread(t["thread_id"])
            t["status"] = "archived"
            save_state(args.state, state)
            print(f"archived {args.task}")
        elif args.cmd == "fork":
            t = state["tasks"][args.task]
            # 生产教训：fork 只分叉 thread 历史，不自动隔离工作区。
            # 旧实现复用原 worktree，合并后 worktree 已删时 fork 出来的
            # thread 连 cwd 都没了，只能只读。正确做法：有 repo 时给
            # fork 单独建一个新 worktree，并把 cwd 传给 thread/fork。
            new_cwd = t["cwd"]
            new_branch = t.get("branch")
            if t.get("repo"):
                old_cwd = t["cwd"]
                new_cwd, new_branch = make_worktree(t["repo"], args.name)
                if os.path.isdir(old_cwd):
                    # 未合并任务：新 worktree 只有主仓内容，没有原任务的
                    # 改动。把原 worktree 的文件（不含 .git）拷过去，
                    # fork 出来的 thread 历史和工作区才对得上。
                    import shutil
                    for root, dirs, files in os.walk(old_cwd):
                        dirs[:] = [d for d in dirs
                                   if d not in (".git", "__pycache__")]
                        for fn in files:
                            if fn.endswith(".pyc") or fn == ".git":
                                # .git 在 worktree 里是一个指向 gitdir 的文件，
                                # 拷过去会把新 worktree 的指针指回旧 worktree，
                                # 导致 git worktree 校验失败（2026-10-25 实测）。
                                continue
                            src = os.path.join(root, fn)
                            rel = os.path.relpath(src, old_cwd)
                            dst = os.path.join(new_cwd, rel)
                            os.makedirs(os.path.dirname(dst), exist_ok=True)
                            shutil.copy2(src, dst)
            # fork 不传 sandbox 时服务端默认 read-only（实测），
            # 所以必须显式传原任务的沙箱，旧任务没存沙箱时默认 workspace-write。
            fork_sandbox = args.sandbox or t.get("sandbox") or "workspace-write"
            new_id = srv.fork_thread(t["thread_id"], cwd=new_cwd,
                                     sandbox=fork_sandbox)
            nt = dict(t)
            nt.update({"name": args.name, "thread_id": new_id,
                       "cwd": new_cwd, "branch": new_branch,
                       "status": "forked", "parent_task": args.task})
            state["tasks"][args.name] = nt
            save_state(args.state, state)
            print(f"forked {args.task} -> {args.name} thread={new_id} "
                  f"cwd={new_cwd}")
        elif args.cmd == "run":
            t = state["tasks"][args.task]
            srv.resume_thread(t["thread_id"])
            result = run_task_turn_with_retry(srv, state, t, args.prompt,
                                              args.model, args.effort,
                                              args.sandbox, cfg=cfg,
                                              timeout=args.turn_timeout,
                                              interrupt_file=_interrupt_request_path(args.state, args.task))
            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 == "interrupt":
            t = state["tasks"][args.task]
            turn_id = None
            for turn in reversed(t.get("turns", [])):
                turn_id = (turn.get("result") or {}).get("turn_id")
                if turn_id:
                    break
            if not turn_id:
                print(f"interrupt {args.task}: 没有可中断的 turn_id")
            else:
                srv.resume_thread(t["thread_id"])
                resp = srv.interrupt_turn(t["thread_id"], turn_id)
                # 直接中断生效（快路径）时沿用老行为：只有服务端
                # 接受中断才改状态；它说“没有活动回合”时说明该轮
                # 已结束，硬标 interrupted 会掩盖真实终态。
                if resp and "error" not in resp:
                    t["status"] = "interrupted"
                    save_state(args.state, state)
                    req = _interrupt_request_path(args.state, args.task)
                    try:
                        if os.path.exists(req):
                            os.remove(req)
                    except Exception:
                        pass
                    print(f"interrupt {args.task}: turn={turn_id} resp={resp}")
                else:
                    # 直接中断不生效（典型是另一进程在等这个 turn，
                    # 本进程的 server 并不拥有它）：写中断请求文件，
                    # 属主进程在等 turn 时每秒轮询到后用自己的
                    # live 连接发 turn/interrupt 并记 interrupted。
                    # 写完后最多等 10 秒看属主是否已响应，超时也
                    # 不删文件——属主下次轮询时仍会处理，调用方
                    # 也可 Ctrl-C 属主进程 + recover 兜底。
                    req = _interrupt_request_path(args.state, args.task)
                    try:
                        os.makedirs(os.path.dirname(req), exist_ok=True)
                        with open(req, "w", encoding="utf-8") as f:
                            json.dump({"task": args.task,
                                       "thread_id": t["thread_id"],
                                       "turn_id": turn_id,
                                       "requested_at": time.time()},
                                      f, ensure_ascii=False)
                    except Exception as exc:
                        print(f"interrupt {args.task}: 写中断请求失败：{exc}")
                        return
                    print(f"interrupt {args.task}: 直接中断未生效 "
                          f"(resp={resp})，已写中断请求 {req}，"
                          f"等待属主进程处理")
                    for _ in range(10):
                        time.sleep(1)
                        try:
                            cur = json.load(open(args.state,
                                                 encoding="utf-8"))
                            if (cur.get("tasks", {}).get(args.task, {})
                                    .get("status") == "interrupted"):
                                print(f"interrupt {args.task}: 属主已响应，"
                                      f"任务已记 interrupted")
                                break
                        except Exception:
                            pass
                    else:
                        print(f"interrupt {args.task}: 属主 10 秒内未响应，"
                              f"请求文件仍在 {req}，属主下次轮询时会处理")
        elif args.cmd == "delegate":
            parent = state["tasks"][args.parent_task]
            # 子 thread：cwd 与父任务相同 worktree 时只读更安全，
            # 这里默认沿用父 cwd + read-only，避免子任务改坏父任务的改动
            sub_thread = srv.create_thread(parent["cwd"], model=args.model,
                                           sandbox="read-only",
                                           approval_policy="never")
            sub = {"name": args.name, "thread_id": sub_thread,
                   "cwd": parent["cwd"], "repo": parent.get("repo"),
                   "branch": None, "sandbox": "read-only", "status": "running",
                   "prompt": args.task, "parent_task": args.parent_task,
                   "created_at": time.time(), "turns": []}
            state["tasks"][args.name] = sub
            save_state(args.state, state)
            # 子代理链路：向父任务链路文件报到（开始 + 结束各一条），
            # 服务端凭标记把子链路文件拼进父任务的代理链；子任务自己
            # 的过程条目仍写它自己的链路文件（main 起动时按 --name 定）。
            child_trace = (os.path.basename(srv.trace_path)
                           if srv.trace_path else None)
            _append_trace_marker(
                args.state, args.parent_task,
                {"kind": "agent-start", "id": "agent:" + args.name,
                 "name": args.name, "prompt": (args.task or "")[:300],
                 "trace": child_trace, "ts": int(time.time() * 1000)})
            result = run_task_turn(srv, state, sub, args.task,
                                   args.model, args.effort, args.sandbox,
                                   timeout=args.turn_timeout,
                                   interrupt_file=_interrupt_request_path(args.state, args.name))
            save_state(args.state, state)
            _append_trace_marker(
                args.state, args.parent_task,
                {"kind": "agent-done", "id": "agent:" + args.name + ":done",
                 "name": args.name, "status": sub["status"],
                 "trace": child_trace, "ts": int(time.time() * 1000)})
            print(json.dumps({"sub_task": args.name, "status": sub["status"],
                              "text": result.get("text", "")},
                             indent=2, ensure_ascii=False))
            if args.report_to_parent:
                srv.resume_thread(parent["thread_id"])
                rep = run_task_turn(srv, state, parent,
                                    f"子任务 {args.name} 的结论如下，请汇总到主任务结论里，"
                                    f"不要重复做子任务的工作：\n{result.get('text', '')}",
                                    args.model, args.effort, args.sandbox,
                                    timeout=args.turn_timeout,
                                    interrupt_file=_interrupt_request_path(args.state, args.parent_task))
                save_state(args.state, state)
                print(f"reported to parent: {rep.get('text', '')[:1000]}")
        elif args.cmd == "goal":
            task = dispatch(srv, state, args.name, args.task, args.repo,
                            args.model, args.effort, args.sandbox,
                            args.approval_policy,
                            seed_dir=args.seed_dir)
            save_state(args.state, state)
            run_goal_loop(srv, state, args.state, task, args.task,
                          args.verify, args.max_rounds, args.model,
                          args.effort, args.sandbox, cfg=cfg,
                          timeout=args.turn_timeout,
                          verify_timeout=args.verify_timeout)
        elif args.cmd == "goal-resume":
            t = state["tasks"][args.task]
            if args.goal_text:
                # 旧任务接续 goal 模式（云端 /continue 带 goal_mode=1
                # 时走这里）：旧任务已终局、没有 goal 记录是正常情况，
                # 不能按断点续跑的守卫拒绝。只要线程还在，就在旧线程
                # 上从第 1 轮开新目标循环（新语义：无验收、不传轮次
                # 时不限轮次，模型宣告完成才结束）。
                if not t.get("thread_id"):
                    print(f"goal-resume 拒绝：{args.task} 没有会话线程，"
                          f"不能接续")
                    return
                max_rounds = args.max_rounds \
                    if args.max_rounds is not None else 0
                srv.resume_thread(t["thread_id"])
                run_goal_loop(srv, state, args.state, t, args.goal_text,
                              None, max_rounds,
                              args.model, args.effort, args.sandbox, cfg=cfg,
                              timeout=args.turn_timeout,
                              verify_timeout=args.verify_timeout,
                              start_round=1,
                              last_verify="",
                              last_turn_status=None,
                              last_turn_text="")
                return
            if t.get("status") not in ("goal-running", "interrupted",
                                       "running"):
                print(f"goal-resume 拒绝：{args.task} 当前状态是 "
                      f"{t.get('status')}，已终局或已达标；"
                      f"goal-unmet / goal-met 后请新建任务，不要同任务续跑")
                return
            if not t.get("goal"):
                print(f"goal-resume 拒绝：{args.task} 没有 goal 记录")
                return
            # 新形态 verify_command 为空、goal_max_rounds 为 0（不限轮次）：
            # 0 是合法值不能用 or 兜底，否则续跑会被悄悄改成 3 轮上限。
            if args.max_rounds is not None:
                max_rounds = args.max_rounds
            elif t.get("goal_max_rounds") is not None:
                max_rounds = t["goal_max_rounds"]
            else:
                max_rounds = 3
            srv.resume_thread(t["thread_id"])
            last_turn = (t.get("turns") or [{}])[-1]
            last_result = last_turn.get("result") or {}
            run_goal_loop(srv, state, args.state, t, t["goal"],
                          t["verify_command"], max_rounds,
                          args.model, args.effort, args.sandbox, cfg=cfg,
                          timeout=args.turn_timeout,
                          verify_timeout=args.verify_timeout,
                          start_round=int(t.get("goal_rounds_done") or 0) + 1,
                          last_verify=t.get("goal_last_verify") or "",
                          last_turn_status=t.get("goal_last_turn_status"),
                          last_turn_text=last_result.get("text", "") or "")
        elif args.cmd == "parallel":
            jobs = json.load(open(args.jobs))
            # 依赖 + 优先级：先拓扑排序，缺依赖/成环直接报错不跑；
            # 按波次派发 + 跑：依赖都 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_with_retry(
                        srv, state, t, job["task"], args.model,
                        args.effort, args.sandbox, cfg=cfg,
                        timeout=args.turn_timeout,
                        interrupt_file=_interrupt_request_path(args.state, job["name"]))
                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({
                    "name": args.name, "task": args.task,
                    "repo": args.repo or cfg.get("repo"),
                    "interval": args.interval, "last_run": 0})
                save_state(args.state, state)
                print(f"automation added: {args.name} every {args.interval}s")
            else:
                print(json.dumps(state["automations"], indent=2,
                                 ensure_ascii=False))
        elif args.cmd == "run-due":
            now = time.time()
            for auto in state["automations"]:
                if now - auto.get("last_run", 0) < auto["interval"]:
                    log_event(args.state,
                              f"run-due skip {auto['name']} "
                              f"(interval {auto['interval']}s, not due)")
                    continue
                name = f"{auto['name']}-{int(now)}"
                task = dispatch(srv, state, name, auto["task"], auto["repo"],
                                args.model, args.effort, args.sandbox,
                                args.approval_policy)
                save_state(args.state, state)
                result = run_task_turn(srv, state, task, auto["task"],
                                       args.model, args.effort, args.sandbox,
                                       timeout=args.turn_timeout,
                                       interrupt_file=_interrupt_request_path(args.state, name))
                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']}")
                print(f"automation {auto['name']} -> {name}: "
                      f"{result['status']} (进评审队列，见 host.py status/review)")
            save_state(args.state, state)
    finally:
        srv.close()


if __name__ == "__main__":
    main()