initial real project snapshot

host.py

@@ -0,0 +1,869 @@
+#!/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 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 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"
+
+
+class Server:
+    """单个 app-server 进程，多 thread 并发。"""
+
+    def __init__(self, codex_bin=None, codex_home=None, approval="deny"):
+        self.codex_bin = codex_bin or default_codex_bin()
+        env = dict(os.environ)
+        if codex_home:
+            os.makedirs(codex_home, exist_ok=True)
+            env["CODEX_HOME"] = codex_home
+        self.proc = subprocess.Popen(
+            [self.codex_bin, "app-server"],
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL, env=env,
+        )
+        self.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.approval = approval
+        self.reader = threading.Thread(target=self._read_loop, daemon=True)
+        self.reader.start()
+
+    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 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):
+        rid = self.next_id
+        self.next_id += 1
+        q = queue.Queue()
+        with self.pending_lock:
+            self.pending[rid] = q
+        try:
+            self._send({"jsonrpc": "2.0", "id": rid, "method": method,
+                        "params": params or {}})
+            return q.get(timeout=timeout)
+        finally:
+            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 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):
+        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:
+                done = q.get(timeout=timeout)
+            except queue.Empty:
+                return {"status": "timeout", "text": "", "turn_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"]
+            return {"status": turn.get("status", "completed"), "text": text,
+                    "turn_id": turn.get("id") or turn_id,
+                    "root_turn_id": turn.get("rootTurnId"),
+                    "error": turn.get("error")}
+        finally:
+            self.turn_queues.pop(thread_id, None)
+
+    def close(self):
+        try:
+            self.proc.stdin.close()
+        except Exception:
+            pass
+        try:
+            self.proc.wait(timeout=5)
+        except Exception:
+            self.proc.kill()
+
+
+# ---------------- 宿主状态 / 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:
+            return json.load(open(path))
+        except Exception:
+            pass
+    return {"tasks": {}, "automations": []}
+
+
+def load_state(path):
+    with _state_lock(path, exclusive=False):
+        return _read_state_file(path)
+
+
+def save_state(path, state):
+    with _state_lock(path, exclusive=True):
+        # 锁内重读 + 合并：内存任务覆盖同名磁盘任务，磁盘上
+        # 别的进程新增的任务保留；自动化按 name 合并，内存优先。
+        disk = _read_state_file(path)
+        merged_tasks = dict(disk.get("tasks") or {})
+        merged_tasks.update(state.get("tasks") or {})
+        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"):
+                merged[k] = v
+        tmp = f"{path}.tmp.{os.getpid()}"
+        json.dump(merged, open(tmp, "w"), indent=2, ensure_ascii=False)
+        os.replace(tmp, path)
+        # 让调用方内存里的 state 也跟上合并结果，避免下次保存
+        # 又把别的进程的任务挤掉（调用方继续用这份 state 时）。
+        state.clear()
+        state.update(merged)
+
+
+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 dispatch(server, state, name, prompt, repo, model, effort, sandbox,
+             approval_policy, use_worktree=True, parent_task=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)
+    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,
+            "created_at": time.time(), "turns": []}
+    state["tasks"][name] = task
+    return task
+
+
+def run_task_turn(server, state, task, prompt, model, effort,
+                  sandbox=None):
+    # task 自带 sandbox 时优先用它（只读评审任务不能被全局默认覆盖），
+    # 否则用调用方显式传的 sandbox；两者都没有时不传 sandboxPolicy，
+    # 保留 thread 原有设置（旧任务没有 sandbox 记录时尤其重要）。
+    use_sandbox = task.get("sandbox") or sandbox
+    result = server.run_turn(task["thread_id"], prompt, model=model,
+                             effort=effort, sandbox=use_sandbox)
+    task["turns"].append({"prompt": prompt, "result": result})
+    task["status"] = "completed" if result["status"] == "completed" \
+        else "failed"
+    task["last_text"] = result.get("text", "")
+    return result
+
+
+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("--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"])
+    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("--no-worktree", action="store_true")
+
+    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 = 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("run"); p.add_argument("--task", required=True); p.add_argument("--prompt", 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("--verify", required=True)
+    p.add_argument("--max-rounds", type=int, default=3)
+    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")
+
+    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-6.1-sol", "effort": "low",
+            "sandbox": "workspace-write", "approval_policy": "never",
+            "approval": "deny", "codex_home": "",
+            "state": "host-state.json",
+            "_note": "API key 不放这里，只走环境变量",
+        }
+        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-6.1-sol"
+    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"
+    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)
+    srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home,
+                 approval=args.approval)
+    try:
+        srv.initialize()
+        if args.cmd == "dispatch":
+            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)
+            # 生产教训：先落盘任务登记再跑 turn，中途被杀/超时后
+            # 任务还在 host-state 里，可用 run 续跑；否则 turn 中断
+            # 就等于任务丢了（2026-10-25 委派评审 300 秒超时实测）。
+            save_state(args.state, state)
+            result = run_task_turn(srv, state, task, args.task,
+                                   args.model, args.effort, args.sandbox)
+            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", "")},
+                             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}")
+        elif args.cmd == "merge":
+            t = state["tasks"][args.task]
+            if not t.get("repo") or not t.get("branch"):
+                print("非 git worktree 任务，无法 merge，直接看 diff 后手动处理")
+                return
+            # 关键一步：worktree 里的改动默认是未提交的，直接 merge 分支
+            # 合不到任何东西。先在 worktree 里提交，再把分支合并进主仓。
+            git(["add", "-A"], t["cwd"])
+            c = git(["commit", "-m", f"host: {args.task}"], t["cwd"])
+            if c.returncode != 0 and "nothing to commit" not in c.stdout + c.stderr:
+                print(f"worktree 提交失败：{c.stderr}")
+                return
+            r = git(["merge", "--no-ff", t["branch"], "-m", f"host: {args.task}"],
+                    t["repo"])
+            if r.returncode != 0:
+                git(["merge", "--abort"], t["repo"])
+                print(f"合并冲突，已 abort：{r.stderr}")
+                return
+            git(["worktree", "remove", "--force", t["cwd"]], t["repo"])
+            git(["branch", "-d", t["branch"]], t["repo"])
+            t["status"] = "merged"
+            save_state(args.state, state)
+            print(f"merged {t['branch']} into {t['repo']}")
+        elif args.cmd == "discard":
+            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 == "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(srv, state, t, args.prompt,
+                                   args.model, args.effort, args.sandbox)
+            save_state(args.state, state)
+            print(json.dumps(result, indent=2, ensure_ascii=False))
+        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)
+            result = run_task_turn(srv, state, sub, args.task,
+                                   args.model, args.effort, args.sandbox)
+            save_state(args.state, state)
+            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)
+                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)
+            save_state(args.state, state)
+            ok = False
+            last_verify = ""
+            for round_no in range(1, args.max_rounds + 1):
+                if round_no == 1:
+                    prompt = args.task
+                else:
+                    # 生产教训：只说“验收未通过” agent 不知道改什么，
+                    # 2026-10-25 实测 prod-retry 3 轮都原地不动。
+                    # 必须把验收命令和失败输出喂回去，才是可用的目标循环。
+                    prompt = (f"上一轮验收未通过，请针对性修复，不要重复已完成的部分。\n"
+                              f"目标：{args.task}\n"
+                              f"验收命令：{args.verify}\n"
+                              f"上一轮验收输出：\n{last_verify}")
+                result = run_task_turn(srv, state, task, prompt,
+                                       args.model, args.effort, args.sandbox)
+                v = subprocess.run(args.verify, shell=True, cwd=task["cwd"],
+                                   capture_output=True, text=True)
+                last_verify = (v.stdout + v.stderr).strip()[:1000] or \
+                    f"exit code {v.returncode}"
+                print(f"round {round_no}: turn={result['status']} "
+                      f"verify_rc={v.returncode} verify_out={last_verify[:300]}")
+                if v.returncode == 0:
+                    ok = True
+                    break
+            task["status"] = "goal-met" if ok else "goal-unmet"
+            save_state(args.state, state)
+            print(f"goal {'met' if ok else 'unmet'} after rounds")
+        elif args.cmd == "parallel":
+            jobs = json.load(open(args.jobs))
+            # 先全部派发（建 thread + worktree），再并发跑 turn
+            tasks = []
+            for job in jobs:
+                t = dispatch(srv, state, job["name"], job["task"],
+                             job.get("repo", ""), args.model, args.effort,
+                             job.get("sandbox", args.sandbox),
+                             args.approval_policy,
+                             use_worktree=bool(job.get("repo")))
+                tasks.append((job, t))
+            save_state(args.state, state)
+
+            def _run(job_task):
+                job, t = job_task
+                return job["name"], run_task_turn(srv, state, t, job["task"],
+                                                  args.model, args.effort, args.sandbox)
+            with ThreadPoolExecutor(max_workers=len(tasks)) as ex:
+                for name, result in ex.map(_run, tasks):
+                    print(f"{name}: {result['status']} "
+                          f"{result.get('text', '')[:300]}")
+            save_state(args.state, state)
+        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)
+                auto["last_run"] = now
+                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()