Merge pull request #103 from openqj/feat-thread-wait-ops

feat: 线程等待与状态读取 8 项（wait/read/message/handoff/worktree/pr-checks/automation/sleep）

cloud/automations.py

@@ -324,6 +324,65 @@ def cancel_automation(conn, automation_id, now=None):
     return True, automation_to_dict(row, now)
 
 
+def update_automation(conn, automation_id, fields, now=None):
+    """更新自动化定义（automation_update）：fields 只认
+    name / prompt / interval_seconds / repo_url 四个键，
+    其它键一律忽略（不报错、不存入，避免调用方拼错字段
+    名后以为已改）；四个键全没给时抛 ValueError（空更新
+    不是成功）。校验与创建同款（_validate_fields，名字
+    仍按“同项目 active 名字唯一”查重，改名撞名时 400）。
+
+    返回 (ok, auto)：行不存在时 (False, None)；已取消时
+    (False, 旧 dict)——取消是终态，改已取消的定义必须
+    重建，不能靠 update 复活（与 cancel 的软删除语义
+    一致）。成功时 commit。注意：更新不重置 last_run_at /
+    last_claimed_at——改间隔只影响下一次到期判断（next_run
+    由 last_run_at + 新间隔算），不立即触发一轮；想立即
+    跑一轮应走执行器的 claim 路径，不是改定义。"""
+    if now is None:
+        now = time.time()
+    ensure_automations_tables(conn)
+    row = conn.execute(
+        "SELECT * FROM automations WHERE id = ?", (automation_id,)
+    ).fetchone()
+    if row is None:
+        return False, None
+    row = _row_dict(row, AUTOMATION_COLUMNS)
+    if row["status"] != ACTIVE:
+        return False, automation_to_dict(row, now)
+    if not isinstance(fields, dict):
+        raise ValueError("fields must be an object")
+    current = automation_to_dict(row, now)
+    name = fields.get("name", current["name"])
+    prompt = fields.get("prompt", current["prompt"])
+    interval = fields.get("interval_seconds",
+                          current["interval_seconds"])
+    repo_url = fields.get("repo_url", current["repo_url"])
+    if not any(k in fields for k in (
+            "name", "prompt", "interval_seconds", "repo_url")):
+        raise ValueError("no updatable field given")
+    name, prompt, interval, repo_url = _validate_fields(
+        name, prompt, interval, repo_url)
+    if name != current["name"]:
+        clash = conn.execute(
+            "SELECT id FROM automations WHERE project_id = ?"
+            " AND name = ? AND status = ? AND id != ?",
+            (row["project_id"], name, ACTIVE, automation_id)
+        ).fetchone()
+        if clash is not None:
+            raise ValueError(
+                "automation name already exists in this project")
+    conn.execute(
+        "UPDATE automations SET name = ?, prompt = ?,"
+        " interval_seconds = ?, repo_url = ? WHERE id = ?",
+        (name, prompt, interval, repo_url, automation_id))
+    conn.commit()
+    row = conn.execute(
+        "SELECT * FROM automations WHERE id = ?", (automation_id,)
+    ).fetchone()
+    return True, automation_to_dict(row, now)
+
+
 def due_automations(conn, project_id, now=None):
     """列出某项目当前到期的 active 定义(纯读，不抢单)。
     SQL 层算到期：从未跑过(last_run_at 为空)或上次执行距

cloud/migrations.py

@@ -104,6 +104,7 @@ UNMIGRATED_TABLES = (
     "github_fix_review_dedup",
     "external_identities",
     "pr_jobs",
+    "task_handoffs",
     "rate_limit_hits",
     "user_gateway_keys",
     "webhook_deliveries",

cloud/server.py

@@ -73,6 +73,24 @@ from cloud.peer_messages import peer_message_prefix as _peer_prefix
 from cloud.peer_messages import row_get as _row_get
 from cloud.automations import cancel_automation as _auto_cancel
 from cloud.automations import create_automation as _auto_create
+from cloud.automations import get_automation as _auto_get
+from cloud.automations import list_automation_runs as _auto_runs
+from cloud.automations import update_automation as _auto_update
+from cloud.wait_ops import build_pr_diagnostics as _wait_build_diag
+from cloud.wait_ops import create_handoff as _handoff_create
+from cloud.wait_ops import decode_wait_cursor as _wait_decode_cursor
+from cloud.wait_ops import encode_wait_cursor as _wait_encode_cursor
+from cloud.wait_ops import ensure_handoffs_table as _handoff_ensure
+from cloud.wait_ops import poll_for_first_change as _wait_poll_first
+from cloud.wait_ops import significant_now as _wait_significant_now
+from cloud.wait_ops import get_handoff as _handoff_get
+from cloud.wait_ops import interruptible_sleep as _wait_sleep
+from cloud.wait_ops import latest_handoff as _handoff_latest
+from cloud.wait_ops import poll_task_statuses as _wait_poll  # noqa: F401
+# poll_task_statuses 保留给“等全部终态”的调用方与单测；
+# /wait 本身走首变轮询（_wait_poll_first），见 _task_wait。
+from cloud.wait_ops import report_handoff as _handoff_report
+from cloud.wait_ops import worktree_status as _wait_worktree_status
 from cloud.automations import ensure_automations_tables as _auto_ensure
 from cloud.automations import list_automations as _auto_list
 from cloud.automations import triage_inbox as _auto_triage
@@ -1878,6 +1896,33 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             elif rest.endswith("/thread/compact") and method == "POST":
                 self._task_thread_compact(
                     user_id, rest[: -len("/thread/compact")])
+            elif rest.endswith("/thread/read") and method == "GET":
+                self._task_thread_read(user_id,
+                                       rest[: -len("/thread/read")])
+            elif rest.endswith("/thread/message") and method == "POST":
+                self._task_thread_message(
+                    user_id, rest[: -len("/thread/message")])
+            elif rest.endswith("/handoff") and method == "GET":
+                self._task_handoff_get(user_id,
+                                       rest[: -len("/handoff")])
+            elif rest.endswith("/handoff") and method == "POST":
+                self._task_handoff_create(user_id,
+                                          rest[: -len("/handoff")])
+            elif "/handoff/" in rest and rest.endswith("/report") \
+                    and method == "POST":
+                _task_part, _hid_part = rest.split("/handoff/", 1)
+                self._task_handoff_report(
+                    user_id, _task_part, _hid_part[: -len("/report")])
+            elif rest.endswith("/worktree") and method == "GET":
+                self._task_worktree_status(user_id,
+                                           rest[: -len("/worktree")])
+            elif rest.endswith("/pr/checks") and method == "GET":
+                self._task_pr_checks(user_id,
+                                     rest[: -len("/pr/checks")])
+            elif rest.endswith("/sleep") and method == "POST":
+                self._task_sleep(user_id, rest[: -len("/sleep")])
+            elif rest.endswith("/wait") and method == "POST":
+                self._task_wait(user_id, rest[: -len("/wait")])
             elif rest.endswith("/fork") and method == "POST":
                 self._fork_task(user_id, rest[: -len("/fork")])
             elif rest.endswith("/approvals") and method == "GET":
@@ -2694,6 +2739,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
             else:
                 self._send_json(404, {"error": "not found"})
             return
+        # /projects/{id}/automations/{aid}：GET 单条（含最近产出）
+        # / PATCH 更新（automation_update）
+        if len(parts) == 3 and parts[1] == "automations":
+            if method == "GET":
+                err = self._check_project(user_id, project_id, "read")
+                if err:
+                    self._send_json(err[0], {"error": err[1]})
+                    return
+                self._automation_get(user_id, project_id, parts[2])
+            elif method == "PATCH":
+                err = self._check_project(user_id, project_id, "create_task")
+                if err:
+                    self._send_json(err[0], {"error": err[1]})
+                    return
+                self._automation_update(user_id, project_id, parts[2])
+            else:
+                self._send_json(404, {"error": "not found"})
+            return
         # /projects/{id}/automations/{aid}/cancel
         if len(parts) == 4 and parts[1] == "automations" \
                 and parts[3] == "cancel":
@@ -4401,6 +4464,728 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         response["server_name"] = name
         self._send_json(200, response)
 
+    # -- 线程等待与移交操作（对照 codex app 8 项） -----------------
+    def _task_thread_read(self, user_id, task_id):
+        """GET /tasks/{id}/thread/read（read_thread）：合成摘要，
+        不是 /thread 的原样转发。任务状态、结果、心跳来自任务行；
+        进度来自链路快照（条目数 + 最后一条）；thread 元数据
+        尽力取（guard 挡住的 Docker / cloud / 异机任务、以及
+        thread-read 失败时，thread 回 null + note，仍回 200——
+        摘要不能因为 thread 读不到就把任务状态也藏起来，这
+        正是它与 /thread（409 / 502）的区别）。
+        详情来自链路条目本身：kind=file 是改动文件、
+        kind=exec 是执行命令（含退出码）、kind=message /
+        plan / reasoning 是最近动态；测试结果在链路里没有
+        单独记录（test_results 回 null + note，不编造“测试
+        通过”——测试通常就是命令，见 commands 的 exitCode
+        与 recent_activity 的 agent 消息）。阻塞判断来自
+        待审批（decision）与待处理后续消息（processing），
+        再加上“终态且可复活”时的 needs_instruction。读权限
+        read；审批扫荡（过期变 expired）与 /approvals 列表
+        同款，读摘要时一并做。"""
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        entries, trace_note = self._trace_snapshot(task)
+        last = entries[-1] if entries else None
+        last_text = ""
+        if isinstance(last, dict):
+            last_text = str(last.get("text") or last.get("message")
+                            or "")[:500]
+        now = time.time()
+        heartbeat_age = None
+        if task.get("last_heartbeat"):
+            heartbeat_age = now - task["last_heartbeat"]
+        thread = None
+        thread_id = None
+        note = trace_note
+        guard = self._thread_task_guard(task)
+        if guard is not None:
+            note = guard[1]
+        else:
+            resolved, _home, _cwd, rerr = self._resolve_task_thread(task)
+            if rerr is not None:
+                note = rerr[1]
+            else:
+                thread_id = resolved.get("id")
+                ok, result = _run_host_cli(
+                    ["thread-read", "--thread-id", thread_id],
+                    _tenant_codex_home(self.server, task["project_id"]))
+                if not ok:
+                    note = result
+                else:
+                    full = _extract_thread(result)
+                    if full is None:
+                        note = ("host.py thread-read output has"
+                                " no thread")
+                    else:
+                        thread = full
+        # 改动文件 / 命令 / 最近动态：只取链路里已有的事实，
+        # 去重与截断在服务端做，不回整条链路原文。
+        changed = {}
+        for entry in entries:
+            if isinstance(entry, dict) and entry.get("kind") == "file" \
+                    and entry.get("path"):
+                changed[entry["path"]] = {
+                    "path": entry["path"],
+                    "change": entry.get("change") or "",
+                    "status": entry.get("status") or ""}
+        changed_files = list(changed.values())[-50:]
+        commands = [{
+            "command": (e.get("command") or "")[:500],
+            "exit_code": e.get("exitCode"),
+            "duration_ms": e.get("durationMs"),
+        } for e in entries if isinstance(e, dict)
+            and e.get("kind") == "exec"][-20:]
+
+        def _activity(entry):
+            if not isinstance(entry, dict):
+                return {"kind": "unknown"}
+            text = (entry.get("text") or entry.get("command")
+                    or entry.get("path") or entry.get("summary")
+                    or entry.get("tool") or "")
+            return {"kind": entry.get("kind"),
+                    "id": entry.get("id"),
+                    "text": str(text)[:300]}
+
+        recent_activity = [_activity(e) for e in entries[-10:]]
+        # 待审批与待处理后续消息（阻塞与“需不需要人”的来源）。
+        conn = self._db()
+        try:
+            _appr_ensure(conn)
+            _appr_expire(conn)
+            pending_approvals = [
+                {"id": a["id"], "kind": a["kind"],
+                 "summary": (a["summary"] or "")[:300]}
+                for a in _appr_list(conn, task_id=task_id,
+                                    status="pending")]
+            _peer_ensure_followups(conn)
+            pending_followups = [{
+                "id": r[0], "mode": r[1],
+                "body": (r[2] or "")[:200]}
+                for r in conn.execute(
+                    "SELECT id, mode, body FROM task_followups"
+                    " WHERE task_id = ? AND status = 'pending'"
+                    " ORDER BY created_at, id",
+                    (task_id,)).fetchall()]
+        finally:
+            conn.close()
+        worktree_host = task.get("worktree_host")
+        remote = worktree_host and worktree_host != socket.gethostname()
+        agents = []
+        if not remote:
+            try:
+                agents = [{
+                    "id": a.get("id"), "kind": a.get("kind"),
+                    "status": a.get("status")}
+                    for a in _build_agent_chain(
+                        task, _trace_dirs_for_task(self.server, task))]
+            except Exception:
+                agents = []
+        state = {"queued": "waiting", "running": "running",
+                 "completed": "completed", "merged": "completed",
+                 "failed": "failed", "merge-failed": "failed",
+                 "interrupted": "interrupted"}.get(
+                     task["status"], "unknown")
+        needs_instruction = task["status"] in (
+            "completed", "failed", "interrupted") \
+            and not task.get("resume_message")
+        reasons = []
+        if pending_approvals:
+            reasons.append("pending approvals need a decision")
+        if pending_followups:
+            reasons.append("pending followups need processing")
+        if needs_instruction:
+            reasons.append("terminal task can be continued"
+                           " with a new instruction")
+        if task.get("error_text"):
+            reasons.append("task has an error_text")
+        self._send_json(200, {
+            "task_id": task_id,
+            "project_id": task["project_id"],
+            "status": task["status"],
+            "state": state,
+            "thread_id": thread_id,
+            "thread": thread,
+            "progress": {
+                "trace_entries": len(entries),
+                "last_trace": ({"kind": last.get("kind"),
+                                "id": last.get("id"),
+                                "text": last_text}
+                               if isinstance(last, dict) else None),
+                "heartbeat_age_seconds": heartbeat_age,
+                "worktree_host": worktree_host,
+            },
+            "result": {
+                "text": task.get("result_text"),
+                "error_text": task.get("error_text"),
+                "source": ("result_text" if task.get("result_text")
+                           else None),
+            },
+            "changed_files": changed_files,
+            "commands": commands,
+            "recent_activity": recent_activity,
+            "agents": agents,
+            "test_results": None,
+            "test_results_note": (
+                "no separate test-result record in the trace;"
+                " tests usually appear as commands (see exit_code)"
+                " and agent messages (see recent_activity)"),
+            "pending_approvals": pending_approvals,
+            "pending_followups": pending_followups,
+            "attention": {
+                "blocked": bool(pending_approvals),
+                "needs_decision": bool(pending_approvals),
+                "needs_processing": bool(pending_followups),
+                "needs_instruction": needs_instruction,
+                "reasons": reasons,
+            },
+            "note": note,
+        })
+
+    def _task_thread_message(self, user_id, task_id):
+        """POST /tasks/{id}/thread/message（send_message_to_thread）：
+        调用方以自己名义给另一会话的 thread 发后续指令。
+        授权模型：调用方须对目标项目有 create_task（项目写
+        权限）——“用户在对话里让我协调”本身不是 API 权限，
+        API 只认 key + 项目权限，跨项目写一律 403，审计
+        thread_message_sent 留痕；这就是“向其他任务发消息
+        需要明确授权”在本系统的落地（当前会话的协调仍须
+        持写权限的 key，不能凭一句话越权）。
+        投递分两路：进行中任务复用 /followups 的队列
+        （queued / direct，direct 同样立即进链路、不中断
+        本轮）；终态任务（completed / failed / interrupted /
+        merge-failed）用 resume_message 复活——与 /continue
+        同款 CAS（状态 + resume_message 为空才改行），并
+        清掉旧 result/error（新一轮的指令不应顶着旧结果，
+        这点与 /messages 的 peer 复活不同，后者不清旧结果）；
+        复活后 worker 认领即在同一 thread 上再跑一轮，这
+        正是“对失败任务发修复指令”的路径。merged 一律
+        409（worktree 已删，请分支）；goal 任务一律 409
+        （goal 循环自有轮次，终态 goal 走 /continue）。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        task, err = self._thread_task(user_id, task_id, "create_task")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        if self._reject_if_over_quota_for_task(user_id, task):
+            return
+        text = body.get("body")
+        if not text or not isinstance(text, str) or not text.strip():
+            self._send_json(400, {"error": "body required"})
+            return
+        text = text.strip()
+        if len(text) > 20000:
+            self._send_json(400, {"error": "body too long"})
+            return
+        mode = body.get("mode", "queued")
+        if mode not in ("queued", "direct"):
+            self._send_json(400, {"error": "mode must be queued or direct"})
+            return
+        if task.get("verify_command") or task.get("goal_mode"):
+            self._send_json(409, {"error": "goal 任务暂不支持后续消息"})
+            return
+        if task["status"] == "merged":
+            self._send_json(409, {
+                "error": "已合并任务不能继续（worktree 已删），请分支",
+                "status": task["status"]})
+            return
+        if task["status"] in TERMINAL_STATUSES:
+            # 复活路：与 /continue 同款终态集（merged 已在上
+            # 面排除），CAS 防“已有待续跑消息”时重复复活。
+            now = time.time()
+            conn = self._db()
+            try:
+                cur = conn.execute(
+                    "UPDATE tasks SET status = 'queued',"
+                    " resume_message = ?, result_text = NULL,"
+                    " error_text = NULL, actual_model = NULL,"
+                    " updated_at = ? WHERE id = ? AND status IN"
+                    " ('completed', 'failed', 'interrupted',"
+                    " 'merge-failed') AND resume_message IS NULL",
+                    (text, now, task_id))
+                conn.commit()
+                if cur.rowcount != 1:
+                    self._send_json(409, {
+                        "error": "target has a pending resume message"
+                                 " or status changed",
+                        "status": task["status"]})
+                    return
+            finally:
+                conn.close()
+            with _auth_store(self.server.db_path) as store:
+                store.audit(user_id, task["project_id"],
+                            "thread_message_sent",
+                            "%s:revived" % task_id)
+            self._send_json(201, {
+                "task_id": task_id, "delivery": "revived",
+                "followup_id": None, "status": "queued",
+                "mode": mode})
+            return
+        now = time.time()
+        conn = self._db()
+        try:
+            _peer_ensure_followups(conn)
+            fu_id = uuid.uuid4().hex
+            conn.execute(
+                "INSERT INTO task_followups(id, task_id, project_id,"
+                " body, mode, status, created_by, created_at,"
+                " updated_at) VALUES(?, ?, ?, ?, ?, 'pending', ?, ?, ?)",
+                (fu_id, task_id, task["project_id"], text, mode,
+                 user_id, now, now))
+            conn.commit()
+            fu = _followup_to_dict(
+                conn.execute(
+                    "SELECT * FROM task_followups WHERE id = ?",
+                    (fu_id,)).fetchone())
+        finally:
+            conn.close()
+        if mode == "direct":
+            self._append_direct_trace(task, fu_id, text)
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"],
+                        "thread_message_sent",
+                        "%s:%s" % (task_id, fu_id))
+        self._send_json(201, dict(fu, delivery="queued"
+                                  if mode == "queued" else "direct"))
+
+    def _task_handoff_create(self, user_id, task_id):
+        """POST /tasks/{id}/handoff：登记一次移交意图
+        （get_handoff_status 的写侧）。target_host 必填；
+        同一任务已有未终态移交时 409（见 wait_ops）。
+        权限 create_task；记 handoff_created 审计。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        task, err = self._thread_task(user_id, task_id, "create_task")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        conn = self._db()
+        try:
+            try:
+                handoff = _handoff_create(
+                    conn, task_id, task["project_id"],
+                    body.get("target_host"), user_id,
+                    source_host=task.get("worktree_host")
+                    or socket.gethostname(),
+                    detail=body.get("detail") or "")
+            except ValueError as exc:
+                self._send_json(409 if "already has an open" in str(exc)
+                                else 400, {"error": str(exc)})
+                return
+        finally:
+            conn.close()
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"],
+                        "handoff_created",
+                        "%s -> %s" % (task_id, handoff["id"]))
+        self._send_json(201, handoff)
+
+    def _task_handoff_get(self, user_id, task_id):
+        """GET /tasks/{id}/handoff（get_handoff_status）：轮询
+        最新移交。无移交记录时 handoff 回 null，同时回任务
+        当前状态与当前主机——轮询方据此判断“还没交”还是
+        “已交但对方未回报”。读权限 read。"""
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        conn = self._db()
+        try:
+            handoff = _handoff_latest(conn, task_id)
+        finally:
+            conn.close()
+        current_host = task.get("worktree_host") or socket.gethostname()
+        self._send_json(200, {
+            "task_id": task_id,
+            "task_status": task["status"],
+            "current_host": current_host,
+            "handoff": handoff,
+            "arrived": bool(
+                handoff is not None
+                and current_host == handoff["target_host"]),
+            "note": ("" if handoff is not None
+                     else "no handoff recorded for this task"),
+        })
+
+    def _task_handoff_report(self, user_id, task_id, handoff_id):
+        """POST /tasks/{id}/handoff/{hid}/report：对方主机
+        回报移交进度。body：status（必填）、progress（可选
+        0-100）、detail（可选）。跨项目移交一律 404（不
+        泄露存在性）；非法状态 / 进度 400；非法转移与终态
+        重报 409。权限 create_task（与任务心跳同款，回报
+        方持项目写权限）；记 handoff_reported 审计。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        task, err = self._thread_task(user_id, task_id, "create_task")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        conn = self._db()
+        try:
+            existing = _handoff_get(conn, handoff_id)
+            if existing is None or existing["project_id"] != \
+                    task["project_id"] or existing["task_id"] != task_id:
+                self._send_json(404, {"error": "handoff not found"})
+                return
+            ok, result = _handoff_report(
+                conn, handoff_id, body.get("status"),
+                progress=body.get("progress"),
+                detail=body.get("detail"))
+        finally:
+            conn.close()
+        if not ok:
+            if result == "handoff not found":
+                self._send_json(404, {"error": result})
+            elif result.startswith("cannot move"):
+                self._send_json(409, {"error": result})
+            else:
+                self._send_json(400, {"error": result})
+            return
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, task["project_id"],
+                        "handoff_reported",
+                        "%s:%s" % (task_id, handoff_id))
+        self._send_json(200, result)
+
+    def _task_worktree_status(self, user_id, task_id):
+        """GET /tasks/{id}/worktree（get_worktree_creation_status）：
+        派生状态，不是异步创建任务的轮询——云端 worktree 由
+        worker 认领后同步建，建的过程没有可查的 job 行。
+        worktree 在别机时 status=remote 且 exists=null（本机
+        查不到，不谎报不存在）；任务绑 environment 时一并回
+        environment_status（那才是真异步的创建过程）。读权限
+        read。"""
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        worktree_host = task.get("worktree_host")
+        remote = worktree_host if (
+            worktree_host and worktree_host != socket.gethostname()
+        ) else None
+        path = _thread_worktree_for_task(self.server, task)
+        exists = None
+        is_git = None
+        if remote is None:
+            exists = os.path.isdir(path)
+            is_git = os.path.exists(os.path.join(path, ".git")) \
+                if exists else False
+        env_status = None
+        if task.get("environment_id"):
+            env_row = self._fetch_environment(task["environment_id"])
+            if env_row is not None:
+                env_status = _environment_to_dict(env_row)["status"]
+        status = _wait_worktree_status(
+            task["status"], path, exists, is_git,
+            environment_status=env_status, remote_host=remote)
+        branch = None
+        if remote is None and exists and is_git:
+            # 尽力取分支：git 不在、不是真仓库、超时都回 null，
+            # 不让分支查询失败拖垮整个 worktree 状态。
+            try:
+                proc = subprocess.run(
+                    ["git", "-C", path, "branch", "--show-current"],
+                    capture_output=True, text=True, timeout=5)
+                if proc.returncode == 0:
+                    branch = (proc.stdout or "").strip() or None
+            except Exception:
+                branch = None
+        self._send_json(200, dict(status, task_id=task_id,
+                                  branch=branch,
+                                  environment_id=task.get(
+                                      "environment_id")))
+
+    def _task_pr_checks(self, user_id, task_id):
+        """GET /tasks/{id}/pr/checks（read_pull_request_check_diagnostics）：
+        只读诊断，绝不重跑检查、不批准、不合并（与 /pr 的
+        建 PR、/merge 的合并门是三条线）。无 pr_number 时
+        409（还没建 PR，查无可查）；worktree 在别机时 409
+        （gh 必须在 worktree 所在主机跑，与 /pr 的队列路
+        不同——诊断是只读拉取，不排队等对方 worker 顺带跑，
+        避免“查一次 CI 就排一个 PR 任务”的副作用）。
+        取数分两层：先 `gh pr checks --json`（workflow、
+        link、description 即“哪个 Action 失败”与失败摘要），
+        JSON 失败时回落制表符解析；有仓库 + head sha 且有
+        失败项时，再走 check-runs API 拿失败日志原文
+        （output.text，截 4000 字）与文件行号注释
+        （annotations：path / start_line / message，上限
+        20 条）。check-runs 一层失败时不 502 整单——
+        checks 本身已够回答“哪个检查失败”，日志是加分项，
+        此时 check_runs 回 available=False + note。读权限
+        read。"""
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        if not task.get("pr_number"):
+            self._send_json(409, {
+                "error": "task has no pull request yet",
+                "pr_number": None})
+            return
+        worktree_host = task.get("worktree_host")
+        if worktree_host and worktree_host != socket.gethostname():
+            self._send_json(409, {
+                "error": "task worktree is on host %s;"
+                         " pr checks must run there" % worktree_host,
+                "worktree_host": worktree_host})
+            return
+        from cloud.pr import _run as _pr_run
+        from cloud.pr import parse_git_remote_owner_repo as _parse_repo
+        worktree_path = _thread_worktree_for_task(self.server, task)
+        runner = getattr(self.server, "pr_gh_runner", None)
+
+        def _run(cmd):
+            result = _pr_run(cmd, cwd=worktree_path, runner=runner)
+            return result.returncode, result.stdout or "", \
+                result.stderr or ""
+
+        owner_repo = None
+        conn = self._db()
+        try:
+            row = conn.execute(
+                "SELECT repo_url FROM projects WHERE id = ?",
+                (task["project_id"],)).fetchone()
+            repo_url = row[0] if row is not None else None
+        finally:
+            conn.close()
+        if repo_url:
+            parsed = _parse_repo(repo_url)
+            if parsed is not None:
+                owner_repo = "%s/%s" % (parsed[0], parsed[1])
+        head_sha = None
+        if owner_repo:
+            rc, out, _e = _run(["gh", "pr", "view",
+                                str(task["pr_number"]),
+                                "--json", "headRefOid",
+                                "--jq", ".headRefOid"])
+            head_sha = (out or "").strip() or None
+            if not head_sha:
+                rc, out, _e = _run(["git", "-C", worktree_path,
+                                    "rev-parse", "HEAD"])
+                head_sha = (out or "").strip() or None
+        try:
+            diag = _wait_build_diag(_run, task["pr_number"],
+                                    owner_repo=owner_repo,
+                                    head_sha=head_sha)
+        except RuntimeError as exc:
+            self._send_json(502, {"error": str(exc)[:500]})
+            return
+        pending = [c["name"] for c in diag["checks"]
+                   if c["state"] == "pending"]
+        self._send_json(200, {
+            "task_id": task_id,
+            "pr_url": task.get("pr_url"),
+            "pr_number": task["pr_number"],
+            "checks": diag["checks"],
+            "all_pass": diag["all_pass"],
+            "failed": diag["failed"],
+            "pending": pending,
+            "diagnostics": diag["diagnostics"],
+            "check_runs": diag["check_runs"],
+        })
+
+    def _task_sleep(self, user_id, task_id):
+        """POST /tasks/{id}/sleep（clock.sleep 的控制面形态）：
+        在服务端睡一段（默认 5 秒，上限 60 秒）再回任务快照，
+        供轮询客户端“等一会儿再继续”而不用自己计时。睡的
+        是调用方的等待，不是任务的执行——任务仍由 worker
+        跑，sleep 绝不暂停、加速或改动任务。读权限 read；
+        seconds 非法时 400（不是静默钳制，调用方应知道
+        自己传错了）。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        seconds = body.get("seconds", 5)
+        if isinstance(seconds, bool) or not isinstance(
+                seconds, (int, float)):
+            self._send_json(400, {"error": "seconds must be a number"})
+            return
+        if seconds < 0 or seconds > 60:
+            self._send_json(400, {
+                "error": "seconds must be between 0 and 60"})
+            return
+        slept, _interrupted = _wait_sleep(seconds)
+        row = self._fetch_task(task_id)
+        task = _task_to_dict(row) if row is not None else task
+        self._send_json(200, {
+            "task_id": task_id, "slept_seconds": slept,
+            "status": task["status"],
+            "result_text": task.get("result_text"),
+        })
+
+    def _task_wait(self, user_id, task_id):
+        """POST /tasks/{id}/wait（wait_threads）：等第一个
+        重要变化，不是等全部终态。一次最多 8 个线程（超限
+        400，调用方自己选子集）；body 可选 task_ids（缺省
+        时等该任务的全部 coordinator 子任务，超过 8 个同样
+        400）；timeout_seconds 缺省 30、上限 120；轮询间
+        隔 0.5 秒。
+        重要变化（见 wait_ops.detect_significant_change）：
+        终态（completed / failed / interrupted / merged /
+        merge-failed）、待用户输入（kind=elicitation 的待
+        审批新增）、待审批（其它 kind 的待审批新增）、待
+        继续处理（待处理后续消息新增）。running / queued
+        流转与链路条目增加不算变化，不会把等待刷屏。
+        after_cursor：上一轮返回的游标；传了它，本轮只报
+        相对游标的新变化（已终态的线程不会再报一遍）；
+        非法游标 400（不静默当无游标）。不传游标时，已
+        处重要状态的线程立即返回（不用等一轮）。
+        超时回 timed_out=true + 当时进度 + 新游标，不是
+        错误——调用方拿新游标再等一轮，或先处理进度里
+        已终态 / 已阻塞的线程。返回里同时保留 children
+        （= progress 的别名）与 all_terminal，方便“等
+        全部完成”的调用方直接判断。
+        权限：锚任务 read，且每个等待任务都要 read（跨
+        项目任务无 read 时 403，任务不存在时 404，不
+        泄露存在性以外的信息）；只读轮询，不改任何任务，
+        不创建任务、不发指令。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        task, err = self._thread_task(user_id, task_id, "read")
+        if err is not None:
+            self._send_json(err[0], {"error": err[1]})
+            return
+        timeout = body.get("timeout_seconds", 30)
+        if isinstance(timeout, bool) or not isinstance(
+                timeout, (int, float)):
+            self._send_json(400, {"error": "timeout_seconds must be a number"})
+            return
+        if timeout < 0 or timeout > 120:
+            self._send_json(400, {
+                "error": "timeout_seconds must be between 0 and 120"})
+            return
+        requested = body.get("task_ids")
+        if requested is not None and (
+                not isinstance(requested, list) or any(
+                    not isinstance(x, str) for x in requested)):
+            self._send_json(400, {"error": "task_ids must be a list"})
+            return
+        if requested is not None and len(requested) == 0:
+            self._send_json(400, {"error": "task_ids must not be empty"})
+            return
+        if requested is not None and len(requested) > 8:
+            self._send_json(400, {
+                "error": "at most 8 threads per wait",
+                "count": len(requested)})
+            return
+        conn = self._db()
+        try:
+            if requested is None:
+                rows = conn.execute(
+                    "SELECT id FROM tasks WHERE parent_task_id = ?",
+                    (task_id,)).fetchall()
+                child_ids = [r[0] if not isinstance(r, dict)
+                             and not hasattr(r, "keys") else r["id"]
+                             for r in rows]
+            else:
+                child_ids = list(requested)
+        finally:
+            conn.close()
+        if len(child_ids) > 8:
+            self._send_json(400, {
+                "error": "at most 8 threads per wait;"
+                         " this task has %d children, pass task_ids"
+                         " to select a subset" % len(child_ids),
+                "count": len(child_ids)})
+            return
+        # 每个等待任务先验存在与 read 权限（顺序即 task_ids
+        # 顺序，第一个不合规的先报错，不静默跳过）。
+        for cid in child_ids:
+            _t, terr = self._thread_task(user_id, cid, "read")
+            if terr is not None:
+                self._send_json(terr[0], {"error": terr[1]})
+                return
+        baseline = None
+        if body.get("after_cursor") is not None:
+            try:
+                baseline = _wait_decode_cursor(body["after_cursor"])
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+
+        def _fetch_summaries():
+            conn = self._db()
+            try:
+                _appr_ensure(conn)
+                _peer_ensure_followups(conn)
+                out = {}
+                for cid in child_ids:
+                    row = conn.execute(
+                        "SELECT * FROM tasks WHERE id = ?",
+                        (cid,)).fetchone()
+                    if row is None:
+                        # 验过存在后理论上不会走到这里；走到
+                        # 时按“无状态”处理，不当成终态。
+                        out[cid] = {"status": "not_found",
+                                    "pending_approvals": 0,
+                                    "pending_input": 0,
+                                    "pending_followups": 0}
+                        continue
+                    status = _task_to_dict(row)["status"]
+                    appr_rows = conn.execute(
+                        "SELECT kind FROM task_approvals"
+                        " WHERE task_id = ? AND status = 'pending'",
+                        (cid,)).fetchall()
+                    kinds = [r[0] if not isinstance(r, dict)
+                             and not hasattr(r, "keys") else r["kind"]
+                             for r in appr_rows]
+                    followups = conn.execute(
+                        "SELECT COUNT(*) FROM task_followups"
+                        " WHERE task_id = ? AND status = 'pending'",
+                        (cid,)).fetchone()[0]
+                    out[cid] = {
+                        "status": status,
+                        "pending_approvals": sum(
+                            1 for k in kinds if k != "elicitation"),
+                        "pending_input": sum(
+                            1 for k in kinds if k == "elicitation"),
+                        "pending_followups": int(followups or 0),
+                    }
+                return out
+            finally:
+                conn.close()
+
+        changed, summaries, waited, timed_out = _wait_poll_first(
+            _fetch_summaries, child_ids, baseline=baseline,
+            timeout_seconds=timeout)
+        progress = []
+        for cid in child_ids:
+            summary = summaries.get(cid, {})
+            status = summary.get("status")
+            progress.append({
+                "task_id": cid, "status": status,
+                "terminal": status in TERMINAL_STATUSES,
+                "pending_approvals": summary.get("pending_approvals", 0),
+                "pending_input": summary.get("pending_input", 0),
+                "pending_followups": summary.get("pending_followups", 0),
+                "significant_now": _wait_significant_now(summary),
+            })
+        self._send_json(200, {
+            "task_id": task_id, "changed_task": changed,
+            "progress": progress, "children": progress,
+            "all_terminal": all(
+                p["terminal"] for p in progress),
+            "timed_out": timed_out,
+            "waited_seconds": waited,
+            "cursor": _wait_encode_cursor(summaries),
+        })
+
     def _fork_task(self, user_id, task_id):
         """会话分支：从已结束任务的 thread 分叉出新任务。
 
@@ -6491,6 +7276,65 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                         automation_id)
         self._send_json(200, auto)
 
+    def _automation_get(self, user_id, project_id, automation_id):
+        """GET 单条自动化（轮询用）：定义 + 最近产出记录。
+        行不存在或属于别的项目时一律 404；产出记录读失败
+        时不让整条定义也 502（runs 回空列表 + note）。"""
+        conn = self._db()
+        try:
+            auto = _auto_get(conn, automation_id)
+            runs = []
+            runs_note = ""
+            if auto is not None:
+                try:
+                    runs = _auto_runs(conn, automation_id)
+                except Exception:
+                    runs_note = "runs unavailable"
+        finally:
+            conn.close()
+        if auto is None or auto["project_id"] != project_id:
+            self._send_json(404, {"error": "automation not found"})
+            return
+        self._send_json(200, dict(auto, runs=runs,
+                                  runs_note=runs_note))
+
+    def _automation_update(self, user_id, project_id, automation_id):
+        """PATCH 单条自动化（automation_update）：改名、改
+        prompt、改间隔、改 repo_url，至少给一个字段。注意
+        语义边界：更新不重置 last_run_at（改间隔只影响下次
+        到期，不立即触发）；已取消的定义 409（取消是终态，
+        改它必须重建）；改名撞同项目 active 名时 400；本
+        端点不改“变化时通知”本身——到期产出的任务终态仍
+        走 worker 现有 push 派发（见 automations 模块文档）。
+        权限 create_task；记 automation_updated 审计。"""
+        body, jerr = self._read_json()
+        if jerr:
+            self._send_json(400, {"error": jerr})
+            return
+        conn = self._db()
+        try:
+            try:
+                ok, auto = _auto_update(conn, automation_id, body)
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
+            except Exception as exc:
+                self._send_json(502, {"error": str(exc)[:500]})
+                return
+        finally:
+            conn.close()
+        if auto is None or auto["project_id"] != project_id:
+            self._send_json(404, {"error": "automation not found"})
+            return
+        if not ok:
+            self._send_json(409, {
+                "error": "automation already cancelled"})
+            return
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, project_id, "automation_updated",
+                        automation_id)
+        self._send_json(200, auto)
+
     def _triage_inbox(self, project_id):
         """分诊收件箱(只读)：到期自动化、失败/中断任务、
         待处理后续消息、最近自动化产出。见 automations 模块

cloud/tests/test_thread_wait_ops.py

@@ -0,0 +1,668 @@
+"""线程等待与移交操作单测：cloud/wait_ops 纯逻辑 + server 端点。
+
+不跑真实模型、不联网、不调真实 gh：thread-read 用 stub
+host.py（与 test_thread_ops.py 同款契约），pr checks 用
+注入的 fake gh runner，worktree 状态用临时目录。覆盖鉴权、
+权限、校验、状态机（移交转移、终态不可动）、超时语义与
+诚实降级（异机 worktree / Docker 任务的 thread-read 仍回
+200，只 thread 为空）。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+    from cloud.wait_ops import (build_pr_diagnostics, create_handoff,
+                                decode_wait_cursor,
+                                detect_significant_change,
+                                encode_wait_cursor, interruptible_sleep,
+                                latest_handoff, parse_pr_checks,
+                                poll_for_first_change, poll_task_statuses,
+                                report_handoff, worktree_status)
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+    from cloud.wait_ops import (build_pr_diagnostics, create_handoff,
+                                decode_wait_cursor,
+                                detect_significant_change,
+                                encode_wait_cursor, interruptible_sleep,
+                                latest_handoff, parse_pr_checks,
+                                poll_for_first_change, poll_task_statuses,
+                                report_handoff, worktree_status)
+
+STUB = r'''import json
+import os
+import sys
+
+def main():
+    argv = sys.argv[1:]
+    cmd = None
+    rest = []
+    i = 0
+    while i < len(argv):
+        a = argv[i]
+        if a == "--codex-home":
+            i += 2
+            continue
+        if a.startswith("--codex-home="):
+            i += 1
+            continue
+        if cmd is None:
+            cmd = a
+            i += 1
+            continue
+        rest.append(a)
+        i += 1
+    if cmd == "thread-list":
+        print(json.dumps({"data": [
+            {"id": "t-old", "updatedAt": 100, "createdAt": 90},
+            {"id": "t-new", "updatedAt": 200, "createdAt": 190},
+        ]}))
+    elif cmd == "thread-read":
+        print(json.dumps({"thread": {"id": "t-new", "updatedAt": 200}}))
+    else:
+        print(json.dumps({"error": "unknown subcommand"}))
+        sys.exit(1)
+
+main()
+'''
+
+
+class WaitOpsPureTestCase(unittest.TestCase):
+    def test_parse_pr_checks_tabular_and_human(self):
+        checks, all_pass, failed, pending = parse_pr_checks(
+            "build\tpass\ntest\tfail\nlint\tpending\n")
+        self.assertEqual([c["state"] for c in checks],
+                         ["pass", "fail", "pending"])
+        self.assertFalse(all_pass)
+        self.assertEqual(failed, ["test"])
+        self.assertEqual(pending, ["lint"])
+        checks, all_pass, _, _ = parse_pr_checks(
+            "\u2713 build pass\n\u2713 test pass\n")
+        self.assertTrue(all_pass)
+        checks, all_pass, _, _ = parse_pr_checks("no checks reported")
+        self.assertEqual(checks, [])
+        self.assertFalse(all_pass)
+
+    def test_worktree_status_matrix(self):
+        self.assertEqual(
+            worktree_status("running", "/tmp/x", False, False)["status"],
+            "creating")
+        self.assertEqual(
+            worktree_status("completed", "/tmp/x", False, False)["status"],
+            "missing")
+        self.assertEqual(
+            worktree_status("running", "/tmp/x", True, True)["status"],
+            "ready")
+        self.assertEqual(
+            worktree_status("running", "/tmp/x", True, False)["status"],
+            "present_no_git")
+        remote = worktree_status("running", "/tmp/x", False, False,
+                                 remote_host="host-b")
+        self.assertEqual(remote["status"], "remote")
+        self.assertIsNone(remote["exists"])
+
+    def test_handoff_transitions(self):
+        conn = sqlite3.connect(":memory:")
+        conn.row_factory = sqlite3.Row
+        h = create_handoff(conn, "t1", "p1", "host-b", "u1",
+                           source_host="host-a")
+        self.assertEqual(h["status"], "pending")
+        with self.assertRaises(ValueError):
+            create_handoff(conn, "t1", "p1", "host-c", "u1")
+        ok, h = report_handoff(conn, h["id"], "running", progress=40)
+        self.assertTrue(ok)
+        self.assertEqual((h["status"], h["progress"]), ("running", 40))
+        ok, h = report_handoff(conn, h["id"], "completed")
+        self.assertTrue(ok)
+        ok, err = report_handoff(conn, h["id"], "running")
+        self.assertFalse(ok)
+        self.assertIn("cannot move", err)
+        ok, err = report_handoff(conn, h["id"], "bogus")
+        self.assertFalse(ok)
+        self.assertEqual(latest_handoff(conn, "t1")["status"], "completed")
+        conn.close()
+
+    def test_wait_cursor_and_first_change(self):
+        base = {"a": {"status": "running", "pending_approvals": 0,
+                       "pending_input": 0, "pending_followups": 0},
+                "b": {"status": "running", "pending_approvals": 0,
+                      "pending_input": 0, "pending_followups": 0}}
+        cursor = encode_wait_cursor(base)
+        self.assertEqual(decode_wait_cursor(cursor), base)
+        with self.assertRaises(ValueError):
+            decode_wait_cursor("not-a-cursor!!")
+        cur = dict(base)
+        cur["b"] = {"status": "running", "pending_approvals": 0,
+                    "pending_input": 1, "pending_followups": 0}
+        found = detect_significant_change(base, cur, ["a", "b"])
+        self.assertEqual(found[0], "b")
+        self.assertEqual(found[1], "needs_input")
+        cur2 = dict(base)
+        cur2["a"] = {"status": "failed", "pending_approvals": 0,
+                     "pending_input": 0, "pending_followups": 0}
+        found = detect_significant_change(base, cur2, ["a", "b"])
+        self.assertEqual(found[1], "terminal")
+        # 基线里没有的任务不误报；无游标时已终态立即算变化。
+        found = detect_significant_change(
+            base, {"c": {"status": "completed", "pending_approvals": 0,
+                         "pending_input": 0, "pending_followups": 0}},
+            ["c"])
+        self.assertIsNone(found)
+        found = detect_significant_change(
+            None, {"c": {"status": "completed", "pending_approvals": 0,
+                         "pending_input": 0, "pending_followups": 0}},
+            ["c"])
+        self.assertEqual(found[1], "terminal")
+        changed, summaries, _w, timed_out = poll_for_first_change(
+            lambda: {"a": {"status": "completed",
+                           "pending_approvals": 0, "pending_input": 0,
+                           "pending_followups": 0}},
+            ["a"], baseline=None, timeout_seconds=5)
+        self.assertFalse(timed_out)
+        self.assertEqual(changed["kind"], "terminal")
+
+    def test_build_pr_diagnostics_json_and_check_runs(self):
+        def run(cmd):
+            if "--json" in cmd:
+                return (0, json.dumps([
+                    {"name": "build", "bucket": "pass",
+                     "workflow": "CI",
+                     "link": "https://example.invalid/ci/1",
+                     "description": "ok"},
+                    {"name": "test", "bucket": "fail",
+                     "workflow": "CI",
+                     "link": "https://example.invalid/ci/2",
+                     "description": "1 failed"}]), "")
+            if "view" in cmd:
+                return (0, "abc123\n", "")
+            if any("commits" in c for c in cmd):
+                return (0, json.dumps({"check_runs": [
+                    {"id": 99, "name": "test",
+                     "conclusion": "failure"}]}), "")
+            if any("check-runs" in c for c in cmd):
+                return (0, json.dumps({
+                    "id": 99, "conclusion": "failure",
+                    "output": {
+                        "title": "Test failed",
+                        "summary": "1 test failed",
+                        "text": "FAILED test_x: assert failed",
+                        "annotations": [{
+                            "path": "src/a.py", "start_line": 12,
+                            "annotation_level": "failure",
+                            "message": "assert failed"}]}}), "")
+            return (1, "", "unknown cmd")
+        diag = build_pr_diagnostics(run, 42, owner_repo="o/r",
+                                    head_sha="abc123")
+        self.assertFalse(diag["all_pass"])
+        self.assertEqual(diag["failed"], ["test"])
+        self.assertEqual(diag["checks"][1]["workflow"], "CI")
+        self.assertTrue(diag["check_runs"]["available"])
+        run_detail = diag["check_runs"]["runs"][0]
+        self.assertIn("FAILED test_x", run_detail["log_text"])
+        self.assertEqual(run_detail["annotations"][0]["path"], "src/a.py")
+        self.assertEqual(run_detail["annotations"][0]["start_line"], 12)
+        # 制表符回落：--json 失败时仍能解析，且无仓库时不查 API。
+        diag2 = build_pr_diagnostics(
+            lambda cmd: (1, "", "boom") if "--json" in cmd
+            else (0, "build\tpass\ntest\tfail\n", ""),
+            7)
+        self.assertEqual(diag2["failed"], ["test"])
+        self.assertFalse(diag2["check_runs"]["available"])
+
+    def test_poll_and_sleep(self):
+        calls = {"n": 0}
+
+        def fetch():
+            calls["n"] += 1
+            return {"a": "running"} if calls["n"] < 3 else {"a": "completed"}
+
+        ok, statuses, waited, timed_out = poll_task_statuses(
+            fetch, timeout_seconds=5)
+        self.assertTrue(ok)
+        self.assertFalse(timed_out)
+        self.assertEqual(statuses, {"a": "completed"})
+        ok, _, _, timed_out = poll_task_statuses(
+            lambda: {"a": "running"}, timeout_seconds=0)
+        self.assertFalse(ok)
+        self.assertTrue(timed_out)
+        ok, _, _, _ = poll_task_statuses(lambda: {}, timeout_seconds=5)
+        self.assertTrue(ok)
+        slept, interrupted = interruptible_sleep(0)
+        self.assertEqual((slept, interrupted), (0.0, False))
+
+
+class ThreadWaitServerTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.reader_id, self.reader_key = self.auth.create_user("reader")
+        self.outsider_id, self.outsider_key = self.auth.create_user(
+            "outsider")
+        self.stub_path = os.path.join(self.tmp.name, "stub_host.py")
+        with open(self.stub_path, "w", encoding="utf-8") as f:
+            f.write(STUB)
+        self._old_host_py = os.environ.get("CODEX_HOST_PY")
+        os.environ["CODEX_HOST_PY"] = self.stub_path
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.server.tenants_root = os.path.join(self.tmp.name, "tenants")
+        self.server.host_worktrees_dir = os.path.join(
+            self.tmp.name, "worktrees")
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201, self.project)
+        self.auth.set_member(self.project["id"], self.owner_id,
+                             self.reader_id, "viewer")
+        status, self.project2 = self._req(
+            "POST", "/projects", {"name": "other"}, self.owner_key)
+        self.assertEqual(status, 201, self.project2)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        if self._old_host_py is None:
+            os.environ.pop("CODEX_HOST_PY", None)
+        else:
+            os.environ["CODEX_HOST_PY"] = self._old_host_py
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _create_task(self, project_id=None, **extra):
+        body = {"project_id": project_id or self.project["id"],
+                "prompt": "做事"}
+        body.update(extra)
+        status, task = self._req("POST", "/tasks", body, self.owner_key)
+        self.assertEqual(status, 201, task)
+        return task
+
+    def _set_task(self, task_id, **fields):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            for key, value in fields.items():
+                conn.execute(
+                    "UPDATE tasks SET %s = ? WHERE id = ?" % key,
+                    (value, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def test_thread_read_summary_and_docker_downgrade(self):
+        # 默认执行器是 docker（线程守卫会挡），读摘要的 thread
+        # 部分必须用显式 local 任务测；docker 任务测降级分支。
+        task = self._create_task(executor="local")
+        self._set_task(task["id"], status="running")
+        status, body = self._req(
+            "GET", "/tasks/%s/thread/read" % task["id"],
+            None, self.reader_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["status"], "running")
+        self.assertEqual(body["state"], "running")
+        self.assertEqual(body["thread_id"], "t-new")
+        self.assertEqual(body["thread"]["id"], "t-new")
+        self.assertEqual(body["progress"]["trace_entries"], 0)
+        self.assertIsNone(body["result"]["text"])
+        self.assertEqual(body["changed_files"], [])
+        self.assertEqual(body["commands"], [])
+        self.assertEqual(body["recent_activity"], [])
+        self.assertFalse(body["attention"]["blocked"])
+        self.assertFalse(body["attention"]["needs_instruction"])
+        self.assertIsNone(body["test_results"])
+        docker = self._create_task()
+        status, body = self._req(
+            "GET", "/tasks/%s/thread/read" % docker["id"],
+            None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertIsNone(body["thread"])
+        self.assertIn("docker", body["note"])
+        status, _ = self._req(
+            "GET", "/tasks/%s/thread/read" % task["id"],
+            None, self.outsider_key)
+        self.assertEqual(status, 403)
+        status, _ = self._req(
+            "GET", "/tasks/nope/thread/read", None, self.owner_key)
+        self.assertEqual(status, 404)
+
+    def test_thread_message_followup_semantics(self):
+        task = self._create_task()
+        self._set_task(task["id"], status="running")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/message" % task["id"],
+            {"body": "下一步改 README"}, self.owner_key)
+        self.assertEqual(status, 201, body)
+        self.assertEqual(body["delivery"], "queued")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/message" % task["id"],
+            {"body": "立刻改", "mode": "direct"}, self.owner_key)
+        self.assertEqual(status, 201, body)
+        self.assertEqual(body["delivery"], "direct")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/message" % task["id"],
+            {"body": "坏模式", "mode": "bogus"}, self.owner_key)
+        self.assertEqual(status, 400, body)
+        # 失败任务发修复指令：复活（不是 409），状态回 queued。
+        done = self._create_task()
+        self._set_task(done["id"], status="failed",
+                       error_text="compile error")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/message" % done["id"],
+            {"body": "继续修复编译错误并重跑测试"}, self.owner_key)
+        self.assertEqual(status, 201, body)
+        self.assertEqual(body["delivery"], "revived")
+        status, task_after = self._req(
+            "GET", "/tasks/%s" % done["id"], None, self.owner_key)
+        self.assertEqual(task_after["status"], "queued")
+        self.assertIsNone(task_after["error_text"])
+        # 已合并任务不能复活（worktree 已删），应 409。
+        merged = self._create_task()
+        self._set_task(merged["id"], status="merged")
+        status, body = self._req(
+            "POST", "/tasks/%s/thread/message" % merged["id"],
+            {"body": "再改一下"}, self.owner_key)
+        self.assertEqual(status, 409, body)
+
+    def test_handoff_lifecycle_and_cross_project(self):
+        task = self._create_task()
+        status, body = self._req(
+            "GET", "/tasks/%s/handoff" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertIsNone(body["handoff"])
+        status, h = self._req(
+            "POST", "/tasks/%s/handoff" % task["id"],
+            {"target_host": "host-b"}, self.owner_key)
+        self.assertEqual(status, 201, h)
+        self.assertEqual(h["status"], "pending")
+        status, body = self._req(
+            "POST", "/tasks/%s/handoff" % task["id"],
+            {"target_host": "host-c"}, self.owner_key)
+        self.assertEqual(status, 409, body)
+        status, body = self._req(
+            "POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]),
+            {"status": "running", "progress": 40}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual((body["status"], body["progress"]),
+                         ("running", 40))
+        status, body = self._req(
+            "GET", "/tasks/%s/handoff" % task["id"], None, self.owner_key)
+        self.assertEqual(body["handoff"]["status"], "running")
+        other = self._create_task(self.project2["id"])
+        status, body = self._req(
+            "POST", "/tasks/%s/handoff/%s/report" % (other["id"], h["id"]),
+            {"status": "completed"}, self.owner_key)
+        self.assertEqual(status, 404, body)
+        status, body = self._req(
+            "POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]),
+            {"status": "completed"}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        status, body = self._req(
+            "POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]),
+            {"status": "running"}, self.owner_key)
+        self.assertEqual(status, 409, body)
+
+    def test_worktree_status_derived(self):
+        task = self._create_task()
+        status, body = self._req(
+            "GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["status"], "creating")
+        import re
+        safe = re.sub(r"[^A-Za-z0-9_.-]", "-", task["id"])
+        path = os.path.join(self.server.host_worktrees_dir, safe)
+        os.makedirs(os.path.join(path, ".git"), exist_ok=True)
+        status, body = self._req(
+            "GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key)
+        self.assertEqual(body["status"], "ready")
+        self.assertTrue(body["exists"])
+        self._set_task(task["id"], worktree_host="other-host")
+        status, body = self._req(
+            "GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key)
+        self.assertEqual(body["status"], "remote")
+        self.assertIsNone(body["exists"])
+
+    def test_pr_checks_and_no_pr(self):
+        task = self._create_task()
+        status, body = self._req(
+            "GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 409, body)
+        self._set_task(task["id"], pr_number=42,
+                       pr_url="https://example.invalid/pr/42")
+        # 制表符回落 + 无仓库时不查 check-runs API。
+        self.server.pr_gh_runner = lambda cmd, cwd: SimpleNamespace(
+            returncode=0, stdout="build\tpass\ntest\tfail\n",
+            stderr="")
+        status, body = self._req(
+            "GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertFalse(body["all_pass"])
+        self.assertEqual(body["failed"], ["test"])
+        self.assertIn("test", body["diagnostics"])
+        self.assertFalse(body["check_runs"]["available"])
+        # 有仓库时：JSON checks + check-runs 日志与行号注释。
+        status, repo_proj = self._req(
+            "POST", "/projects",
+            {"name": "repo-proj",
+             "repo_url": "https://github.com/openqj/demo.git"},
+            self.owner_key)
+        self.assertEqual(status, 201, repo_proj)
+        repo_task = self._create_task(repo_proj["id"])
+        self._set_task(repo_task["id"], pr_number=43,
+                       pr_url="https://example.invalid/pr/43")
+
+        def _gh(cmd, cwd):
+            if "--json" in cmd:
+                return SimpleNamespace(returncode=0, stdout=json.dumps([
+                    {"name": "build", "bucket": "pass",
+                     "workflow": "CI",
+                     "link": "https://example.invalid/ci/1",
+                     "description": "ok"},
+                    {"name": "test", "bucket": "fail",
+                     "workflow": "CI",
+                     "link": "https://example.invalid/ci/2",
+                     "description": "1 failed"}]), stderr="")
+            if "view" in cmd:
+                return SimpleNamespace(returncode=0, stdout="abc123\n",
+                                       stderr="")
+            if any("commits" in c for c in cmd):
+                return SimpleNamespace(returncode=0, stdout=json.dumps(
+                    {"check_runs": [{"id": 99, "name": "test",
+                                     "conclusion": "failure"}]}),
+                    stderr="")
+            if any("check-runs" in c for c in cmd):
+                return SimpleNamespace(returncode=0, stdout=json.dumps({
+                    "id": 99, "conclusion": "failure",
+                    "output": {
+                        "title": "Test failed",
+                        "summary": "1 test failed",
+                        "text": "FAILED test_x: assert failed",
+                        "annotations": [{
+                            "path": "src/a.py", "start_line": 12,
+                            "annotation_level": "failure",
+                            "message": "assert failed"}]}}),
+                    stderr="")
+            return SimpleNamespace(returncode=1, stdout="", stderr="boom")
+
+        self.server.pr_gh_runner = _gh
+        status, body = self._req(
+            "GET", "/tasks/%s/pr/checks" % repo_task["id"],
+            None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["checks"][1]["workflow"], "CI")
+        self.assertIn("example.invalid/ci/2", body["checks"][1]["link"])
+        self.assertTrue(body["check_runs"]["available"])
+        run_detail = body["check_runs"]["runs"][0]
+        self.assertIn("FAILED test_x", run_detail["log_text"])
+        self.assertEqual(run_detail["annotations"][0]["path"], "src/a.py")
+        self.assertEqual(run_detail["annotations"][0]["start_line"], 12)
+        self._set_task(task["id"], worktree_host="other-host")
+        status, body = self._req(
+            "GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key)
+        self.assertEqual(status, 409, body)
+
+    def test_sleep_and_wait(self):
+        task = self._create_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/sleep" % task["id"],
+            {"seconds": 0}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["slept_seconds"], 0)
+        status, body = self._req(
+            "POST", "/tasks/%s/sleep" % task["id"],
+            {"seconds": 61}, self.owner_key)
+        self.assertEqual(status, 400, body)
+        parent = self._create_task()
+        child = self._create_task()
+        self._set_task(child["id"], status="completed",
+                       parent_task_id=parent["id"])
+        # 无游标：已终态的子任务立即作为首变返回，不等满超时。
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"timeout_seconds": 30}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["all_terminal"])
+        self.assertFalse(body["timed_out"])
+        self.assertEqual(body["children"][0]["task_id"], child["id"])
+        self.assertEqual(body["changed_task"]["task_id"], child["id"])
+        self.assertEqual(body["changed_task"]["kind"], "terminal")
+        cursor = body["cursor"]
+        # 同一游标再等：已终态不再重复报，超时回进度 + 新游标。
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"timeout_seconds": 0, "after_cursor": cursor},
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["timed_out"])
+        self.assertIsNone(body["changed_task"])
+        # 非法游标 400；超 8 个 400；运行中 + 超时 0 回超时。
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"timeout_seconds": 0, "after_cursor": "bogus!!"},
+            self.owner_key)
+        self.assertEqual(status, 400, body)
+        running = self._create_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"task_ids": [running["id"]], "timeout_seconds": 0},
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["timed_out"])
+        self.assertFalse(body["all_terminal"])
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"task_ids": ["x"] * 9, "timeout_seconds": 0},
+            self.owner_key)
+        self.assertEqual(status, 400, body)
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"timeout_seconds": 121}, self.owner_key)
+        self.assertEqual(status, 400, body)
+
+    def test_wait_cursor_approval_trigger(self):
+        # 基线游标（运行中、无待审批）-> 新建 elicitation 审批
+        # -> 再等同一游标，立即报 needs_input（不是等满超时）。
+        parent = self._create_task()
+        target = self._create_task()
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"task_ids": [target["id"]], "timeout_seconds": 0},
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertTrue(body["timed_out"])
+        cursor = body["cursor"]
+        status, appr = self._req(
+            "POST", "/tasks/%s/approvals" % target["id"],
+            {"kind": "elicitation", "summary": "请确认是否继续"},
+            self.owner_key)
+        self.assertEqual(status, 201, appr)
+        status, body = self._req(
+            "POST", "/tasks/%s/wait" % parent["id"],
+            {"task_ids": [target["id"]], "timeout_seconds": 30,
+             "after_cursor": cursor},
+            self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertFalse(body["timed_out"])
+        self.assertEqual(body["changed_task"]["kind"], "needs_input")
+        self.assertEqual(body["changed_task"]["task_id"], target["id"])
+
+    def test_automation_update_and_get(self):
+        status, auto = self._req(
+            "POST", "/projects/%s/automations" % self.project["id"],
+            {"name": "daily", "prompt": "做每日检查"}, self.owner_key)
+        self.assertEqual(status, 201, auto)
+        status, body = self._req(
+            "PATCH", "/projects/%s/automations/%s"
+            % (self.project["id"], auto["id"]),
+            {"interval_seconds": 7200}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["interval_seconds"], 7200)
+        status, body = self._req(
+            "GET", "/projects/%s/automations/%s"
+            % (self.project["id"], auto["id"]),
+            None, self.owner_key)
+        self.assertEqual(status, 200, body)
+        self.assertEqual(body["runs"], [])
+        status, body = self._req(
+            "PATCH", "/projects/%s/automations/%s"
+            % (self.project["id"], auto["id"]),
+            {}, self.owner_key)
+        self.assertEqual(status, 400, body)
+        status, body = self._req(
+            "PATCH", "/projects/%s/automations/%s"
+            % (self.project2["id"], auto["id"]),
+            {"interval_seconds": 3600}, self.owner_key)
+        self.assertEqual(status, 404, body)
+        status, body = self._req(
+            "POST", "/projects/%s/automations/%s/cancel"
+            % (self.project["id"], auto["id"]),
+            {}, self.owner_key)
+        self.assertEqual(status, 200, body)
+        status, body = self._req(
+            "PATCH", "/projects/%s/automations/%s"
+            % (self.project["id"], auto["id"]),
+            {"interval_seconds": 3600}, self.owner_key)
+        self.assertEqual(status, 409, body)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/wait_ops.py

@@ -0,0 +1,707 @@
+"""线程等待与移交操作（对照 codex app 的 8 项工具面）。
+
+本模块只做控制面纯逻辑，不跑模型、不联网、不 import cloud 其它
+模块（与 automations / peer_messages 同款，避免循环 import）；
+server.py 只负责鉴权、路由与错误码映射。
+
+8 项在控制面的落点（详见 docs/thread-wait-ops.md 对照表）：
+
+- read_thread：server 用任务行 + 链路快照 + 尽力 thread-read
+  合成摘要，本模块不重复造 thread 解析。
+- get_handoff_status：task_handoffs 表（本模块）。移交本身
+  不是本模块发明的传输通道：记录只登记“某任务被交到某主机”
+  的意图与对方回报的进度；真正的执行仍在目标主机的 worker
+  上，状态靠 report 回写，绝不凭登记就谎称已交接。
+- get_worktree_creation_status：派生状态（本模块
+  worktree_status）。云端任务的 worktree 由 worker 在认领后
+  同步建，没有独立的异步创建任务可轮询——本函数如实报告
+  目录存在与否、任务状态与环境状态，不编造百分比进度。
+- read_pull_request_check_diagnostics：本模块只解析
+  `gh pr checks` 的输出；日志原文 gh 本身不给，失败原因
+  即未通过的 check 名单，见 parse_pr_checks。
+- automation_update：cloud/automations.py 的 update_automation
+  （本批新增），本模块不另起一套自动化定义。
+- send_message_to_thread：server 复用 followup 投递（queued /
+  direct），本模块不另建队列。
+- clock.sleep：interruptible_sleep（步进睡眠，可被 stop_event
+  打断；server 的 /sleep 端点与 host.py 的 sleep 命令共用它）。
+- wait_threads：poll_task_statuses（轮询回调取状态，直到
+  全部终态或超时；server 的 /wait 端点用 DB 回读作回调）。
+
+表迁移：task_handoffs 是 SQLite 懒建表（与 pr_jobs /
+automations 同款），不在 POSTGRES_SCHEMA 里，因此必须同步
+进 cloud/migrations.py 的 UNMIGRATED_TABLES（test_p2_migrations
+会实测“源码里真有 DDL 且不在 PG schema 里”）。
+"""
+
+import base64
+import json
+import os
+import re
+import time
+import uuid
+
+HANDOFF_STATUSES = ("pending", "accepted", "running",
+                    "completed", "failed")
+# report 允许的状态转移：终态（completed / failed）不可再动；
+# pending 可进 accepted / running / 终态，accepted 可进 running /
+# 终态，running 只能进终态。跳过中间态不算错（对方可能直接报
+# completed），但终态回退、终态重报一律拒绝。
+_HANDOFF_TRANSITIONS = {
+    "pending": ("accepted", "running", "completed", "failed"),
+    "accepted": ("running", "completed", "failed"),
+    "running": ("completed", "failed"),
+    "completed": (),
+    "failed": (),
+}
+
+HANDOFF_COLUMNS = (
+    "id", "task_id", "project_id", "source_host", "target_host",
+    "status", "progress", "detail", "created_by", "created_at",
+    "updated_at")
+
+TERMINAL_TASK_STATUSES = frozenset(
+    {"completed", "failed", "interrupted", "merged", "merge-failed"})
+
+MAX_SLEEP_SECONDS = 60
+MAX_WAIT_SECONDS = 120
+WAIT_POLL_SECONDS = 0.5
+MAX_WAIT_THREADS = 8
+# wait_threads 的重要变化只认这几类：终态、待审批、待用户输入、
+# 待继续处理。running / queued 之间的流转、链路条目增加都
+# 不算——否则一次等待会被每条进度刷屏，协调方永远等不到
+# “第一个需要我处理”的线程。状态映射见 detect_significant_change。
+SIGNIFICANT_KINDS = ("terminal", "needs_approval", "needs_input",
+                     "needs_continue")
+
+
+def row_get(row, name):
+    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键报
+    KeyError、空行报 TypeError，一律回 None（与 automations
+    的 row_get 同款，老库行缺列时读移交列表不能因此炸掉）。"""
+    try:
+        return row[name]
+    except (KeyError, IndexError, TypeError):
+        return None
+
+
+def _row_dict(row, columns):
+    if row is None or isinstance(row, dict):
+        return row
+    try:
+        row["id"]
+        return row
+    except TypeError:
+        return dict(zip(columns, row))
+
+
+def ensure_handoffs_table(conn):
+    """幂等建 task_handoffs 表。调用方在写成功后 commit
+    （server 的 handler 与 peer 模式同款：建表后由业务写
+    操作一并 commit；只读的 GET 也调它建表再读，空库回空）。"""
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS task_handoffs("
+        "id TEXT PRIMARY KEY, task_id TEXT NOT NULL, "
+        "project_id TEXT NOT NULL, source_host TEXT, "
+        "target_host TEXT NOT NULL, status TEXT NOT NULL, "
+        "progress INTEGER NOT NULL DEFAULT 0, detail TEXT, "
+        "created_by TEXT, created_at REAL, updated_at REAL)"
+    )
+
+
+def handoff_to_dict(row, now=None):
+    """行 -> 对外 dict；progress 一律钳到 0-100，读侧也钳，
+    防止旧脏行把进度写成 120 或 -1 后原样漏到客户端。"""
+    r = _row_dict(row, HANDOFF_COLUMNS)
+    progress = row_get(r, "progress")
+    try:
+        progress = int(progress)
+    except (TypeError, ValueError):
+        progress = 0
+    progress = max(0, min(100, progress))
+    return {
+        "id": row_get(r, "id"),
+        "task_id": row_get(r, "task_id"),
+        "project_id": row_get(r, "project_id"),
+        "source_host": row_get(r, "source_host"),
+        "target_host": row_get(r, "target_host"),
+        "status": row_get(r, "status"),
+        "progress": progress,
+        "detail": row_get(r, "detail") or "",
+        "created_by": row_get(r, "created_by"),
+        "created_at": row_get(r, "created_at"),
+        "updated_at": row_get(r, "updated_at"),
+    }
+
+
+def create_handoff(conn, task_id, project_id, target_host, created_by,
+                   source_host=None, detail="", now=None):
+    """登记一次移交：同一任务同时只能有一个未终态移交
+    （pending / accepted / running），否则抛 ValueError——
+    两个进行中的移交指向不同主机时，轮询方分不清该信哪一个。
+    成功时 commit 并返回 dict；校验失败抛 ValueError。"""
+    if not target_host or not isinstance(target_host, str) \
+            or not target_host.strip():
+        raise ValueError("target_host required")
+    target_host = target_host.strip()
+    if len(target_host) > 200:
+        raise ValueError("target_host too long")
+    if detail is not None and not isinstance(detail, str):
+        raise ValueError("detail must be a string")
+    detail = (detail or "").strip()
+    if len(detail) > 2000:
+        raise ValueError("detail too long")
+    if now is None:
+        now = time.time()
+    ensure_handoffs_table(conn)
+    existing = conn.execute(
+        "SELECT id FROM task_handoffs WHERE task_id = ?"
+        " AND status IN ('pending', 'accepted', 'running')"
+        " LIMIT 1",
+        (task_id,)).fetchone()
+    if existing is not None:
+        raise ValueError("task already has an open handoff")
+    handoff_id = uuid.uuid4().hex
+    conn.execute(
+        "INSERT INTO task_handoffs(id, task_id, project_id,"
+        " source_host, target_host, status, progress, detail,"
+        " created_by, created_at, updated_at)"
+        " VALUES(?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?)",
+        (handoff_id, task_id, project_id, source_host, target_host,
+         detail, created_by, now, now))
+    conn.commit()
+    row = conn.execute(
+        "SELECT * FROM task_handoffs WHERE id = ?", (handoff_id,)
+    ).fetchone()
+    return handoff_to_dict(row)
+
+
+def get_handoff(conn, handoff_id):
+    """按 id 取移交（跨项目时调用方自己比对 project_id 后
+    回 404，不在本函数里泄露跨项目存在性）。"""
+    ensure_handoffs_table(conn)
+    row = conn.execute(
+        "SELECT * FROM task_handoffs WHERE id = ?", (handoff_id,)
+    ).fetchone()
+    if row is None:
+        return None
+    return handoff_to_dict(row)
+
+
+def latest_handoff(conn, task_id):
+    """某任务最新的一次移交（按 created_at、id 排序，时间
+    戳相同时 id 兜底，保证同一库两次读到同一条“最新”）。"""
+    ensure_handoffs_table(conn)
+    row = conn.execute(
+        "SELECT * FROM task_handoffs WHERE task_id = ?"
+        " ORDER BY created_at DESC, id DESC LIMIT 1",
+        (task_id,)).fetchone()
+    if row is None:
+        return None
+    return handoff_to_dict(row)
+
+
+def report_handoff(conn, handoff_id, status, progress=None, detail=None,
+                   now=None):
+    """对方主机回报进度：返回 (ok, handoff_or_error)。
+    ok 为 False 时第二个值是错误文本（调用方按 400 / 409
+    映射：非法状态与非法进度是 400，非法转移是 409）。
+    progress 不传时保持旧值；传了则钳到 0-100 且终态时
+    progress 不强制改 100（对方可能以 failed 终态报 40%，
+    那是真实进度，不是“完成度”）。"""
+    if status not in HANDOFF_STATUSES:
+        return False, "status must be one of %s" % (
+            ", ".join(HANDOFF_STATUSES),)
+    if progress is not None:
+        if isinstance(progress, bool) or not isinstance(progress, int):
+            return False, "progress must be an integer"
+        progress = max(0, min(100, progress))
+    if detail is not None and not isinstance(detail, str):
+        return False, "detail must be a string"
+    if detail is not None and len(detail) > 2000:
+        return False, "detail too long"
+    if now is None:
+        now = time.time()
+    ensure_handoffs_table(conn)
+    row = conn.execute(
+        "SELECT * FROM task_handoffs WHERE id = ?", (handoff_id,)
+    ).fetchone()
+    if row is None:
+        return False, "handoff not found"
+    current = handoff_to_dict(row)
+    if status not in _HANDOFF_TRANSITIONS[current["status"]] \
+            and status != current["status"]:
+        # 同状态重报（例如 running 再报 running + 新进度）允许，
+        # 其它非法转移拒绝。
+        return False, ("cannot move handoff from %s to %s"
+                       % (current["status"], status))
+    new_progress = current["progress"] if progress is None else progress
+    new_detail = current["detail"] if detail is None else detail.strip()
+    conn.execute(
+        "UPDATE task_handoffs SET status = ?, progress = ?,"
+        " detail = ?, updated_at = ? WHERE id = ?",
+        (status, new_progress, new_detail, now, handoff_id))
+    conn.commit()
+    row = conn.execute(
+        "SELECT * FROM task_handoffs WHERE id = ?", (handoff_id,)
+    ).fetchone()
+    return True, handoff_to_dict(row)
+
+
+def parse_pr_checks(output):
+    """解析 `gh pr checks` 输出（制表符与人类可读两种格式，
+    与 host.py 的 parse_pr_checks 同规则，server 不能 import
+    host.py——host 不是包，server 侧只能自带一份解析）。
+
+    返回 (checks, all_pass, failed, pending)：checks 为
+    [{name, state}]，state 统一成 pass / fail / pending /
+    skipped / cancelled / neutral；没有任何 check 时 all_pass
+    为 False（空 PR 不能算全过）；failed 是未通过的 check 名，
+    即“失败原因”的全部内容——gh checks 本身不给日志原文，
+    日志要到 GitHub 上该 check 的详情页看，本函数不编造
+    日志链接（check 行里没有 run id，拼不出可靠链接）。"""
+    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(" \u2713\u2717\u25cf\u25cb-").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)
+    failed = [c["name"] for c in checks if c["state"] == "fail"]
+    pending = [c["name"] for c in checks if c["state"] == "pending"]
+    return checks, all_pass, failed, pending
+
+
+def worktree_status(task_status, worktree_path, exists, is_git,
+                    environment_status=None, remote_host=None):
+    """派生 worktree 创建状态（纯函数，server 负责把文件系统
+    事实传进来，单测不用真的建目录）。
+
+    status 取值：
+    - remote：worktree 在别机，本机查不到，exists 一律 None。
+    - ready：目录在且含 .git（git worktree 或仓库检出）。
+    - present_no_git：目录在但无 .git（非 git 任务的独立目录
+      拷贝，host.py make_worktree 的退化形态，不是失败）。
+    - creating：任务还在 queued / running 且目录未出现——
+      worker 尚未建出来或正在建，轮询方应继续等。
+    - missing：终态任务且目录不在——merged 后 worktree 已删
+      是预期行为，不是“创建失败”；queued / running 之外的
+      终态也同此理，调用方看 task_status 区分。
+    - unknown：其它组合（例如目录不在且任务状态不在上述
+      两类），不硬编一个结论。
+    environment_status（任务绑 environment 时）原样透传，
+    它是真正异步的创建过程（creating -> ready / failed）。"""
+    if remote_host:
+        return {"status": "remote", "path": worktree_path,
+                "exists": None, "is_git": None,
+                "task_status": task_status,
+                "environment_status": environment_status,
+                "note": "worktree is on host %s; this host cannot"
+                        " see it" % remote_host}
+    if exists and is_git:
+        status, note = "ready", ""
+    elif exists:
+        status, note = "present_no_git", (
+            "directory exists without .git (non-git task copy)")
+    elif task_status in ("queued", "running"):
+        status, note = "creating", (
+            "worktree not created yet; worker creates it on claim")
+    elif task_status in TERMINAL_TASK_STATUSES:
+        status, note = "missing", (
+            "worktree not present for a terminal task"
+            " (merged tasks delete it by design)")
+    else:
+        status, note = "unknown", ""
+    return {"status": status, "path": worktree_path,
+            "exists": bool(exists), "is_git": bool(is_git),
+            "task_status": task_status,
+            "environment_status": environment_status,
+            "note": note}
+
+
+def interruptible_sleep(seconds, stop_event=None, step=0.5):
+    """步进睡眠（clock.sleep 的控制面原语）：seconds 钳到
+    [0, MAX_SLEEP_SECONDS]；stop_event 被置位时提前返回
+    (slept, True)，否则睡满返回 (slept, False)。server 的
+    /sleep 端点不传 stop_event（客户端断开不中断 handler，
+    上限 60 秒）；host.py 的 sleep 命令用 SIGINT 转 stop。"""
+    try:
+        seconds = float(seconds)
+    except (TypeError, ValueError):
+        seconds = 0.0
+    seconds = max(0.0, min(float(MAX_SLEEP_SECONDS), seconds))
+    slept = 0.0
+    while slept < seconds:
+        if stop_event is not None and stop_event.is_set():
+            return slept, True
+        chunk = min(step, seconds - slept)
+        time.sleep(chunk)
+        slept += chunk
+    return slept, False
+
+
+def poll_task_statuses(fetch_statuses, timeout_seconds=30.0,
+                       poll_seconds=WAIT_POLL_SECONDS, now=None):
+    """wait_threads 的轮询核：fetch_statuses() 回一个
+    {task_id: status}，直到全部终态或超时。
+
+    返回 (all_terminal, statuses, waited_seconds, timed_out)：
+    空任务集立即回 all_terminal=True（没有可等的东西，
+    不是错误）；fetch 抛异常时不吞——调用方（server）按
+    502 处理，轮询核绝不能把 DB 故障伪装成“超时”。"""
+    try:
+        timeout_seconds = float(timeout_seconds)
+    except (TypeError, ValueError):
+        timeout_seconds = 30.0
+    timeout_seconds = max(0.0, min(float(MAX_WAIT_SECONDS),
+                                   timeout_seconds))
+    start = time.time() if now is None else now()
+    statuses = fetch_statuses()
+    if all(s in TERMINAL_TASK_STATUSES for s in statuses.values()):
+        return True, statuses, 0.0, False
+    while True:
+        elapsed = (time.time() - start) if now is None else (now() - start)
+        if elapsed >= timeout_seconds:
+            return False, statuses, elapsed, True
+        time.sleep(min(poll_seconds, max(0.0, timeout_seconds - elapsed)))
+        statuses = fetch_statuses()
+        if all(s in TERMINAL_TASK_STATUSES for s in statuses.values()):
+            elapsed = (time.time() - start) if now is None \
+                else (now() - start)
+            return True, statuses, elapsed, False
+
+
+# -- wait_threads：首变 + 游标 ------------------------------------
+# 摘要形态（server 的 fetch 回调产出，键名固定）：
+#   {"status": str, "pending_approvals": int,
+#    "pending_input": int, "pending_followups": int}
+# pending_input 是 kind=elicitation 的待审批（向用户要输入），
+# 其余 kind（command / file_change / permissions 等）计入
+# pending_approvals（要审批）。server 不读链路文件做变化
+# 检测：每 0.5 秒全量读 8 个任务的 jsonl 是浪费，状态、
+# 审批、后续消息三张表已足够判断“需不需要人处理”。
+
+def encode_wait_cursor(summaries):
+    """摘要 -> 不透明游标（base64 的 JSON，调用方原样回传，
+    不解析、不拼接）。游标只含状态计数，不含任何正文。"""
+    raw = json.dumps(summaries, ensure_ascii=False,
+                     sort_keys=True).encode("utf-8")
+    return base64.urlsafe_b64encode(raw).decode("ascii")
+
+
+def decode_wait_cursor(cursor):
+    """游标 -> 摘要；非法游标抛 ValueError（调用方转 400，
+    绝不静默当成“无游标”——那样会把旧变化再报一遍，
+    违背 afterCursor“只取新变化”的契约）。"""
+    if not isinstance(cursor, str) or not cursor:
+        raise ValueError("after_cursor must be a non-empty string")
+    try:
+        raw = base64.urlsafe_b64decode(cursor.encode("ascii"))
+        data = json.loads(raw.decode("utf-8"))
+    except (ValueError, UnicodeDecodeError):
+        raise ValueError("after_cursor is not a valid cursor")
+    if not isinstance(data, dict):
+        raise ValueError("after_cursor is not a valid cursor")
+    for value in data.values():
+        if not isinstance(value, dict):
+            raise ValueError("after_cursor is not a valid cursor")
+    return data
+
+
+def _summary_numbers(summary):
+    """摘要里的计数统一成 int（缺键、脏值一律按 0 计，
+    不让旧脏行把等待逻辑炸掉）。"""
+    def _int(value):
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return 0
+    return (_int(summary.get("pending_approvals")),
+            _int(summary.get("pending_input")),
+            _int(summary.get("pending_followups")))
+
+
+def significant_now(summary):
+    """当前摘要是否本身就是“重要状态”（无游标首次等待时用）：
+    终态、待审批、待输入、待继续处理任一成立即算。"""
+    status = (summary or {}).get("status")
+    approvals, inputs, followups = _summary_numbers(summary or {})
+    if status in TERMINAL_TASK_STATUSES:
+        return "terminal"
+    if inputs > 0:
+        return "needs_input"
+    if approvals > 0:
+        return "needs_approval"
+    if followups > 0:
+        return "needs_continue"
+    return None
+
+
+def detect_significant_change(baseline, current, task_ids):
+    """按 task_ids 顺序找第一个重要变化，返回
+    (task_id, kind, detail) 或 None。
+
+    - baseline 为 None：只认 significant_now（首次等待，
+      已终态 / 已阻塞的线程应立即返回，不用等一轮）。
+    - baseline 为 dict：只认相对 baseline 的新增变化；
+      baseline 里没有的任务（调用方在新一轮里加进来的）
+      按“无变化”处理，不把它的当前状态误报成新变化。
+    - 优先级按检查顺序：终态 > 待输入 > 待审批 > 待继续，
+      同一任务多类变化时只报最先的一类；不同任务时按
+      task_ids 顺序，第一个有变化的胜出（“第一个发生
+      重要变化的线程”按列表顺序定先后，轮询本身没有
+      跨任务的精确时间戳可比）。"""
+    for task_id in task_ids:
+        cur = current.get(task_id)
+        if not isinstance(cur, dict):
+            continue
+        if baseline is None:
+            kind = significant_now(cur)
+            if kind is not None:
+                return task_id, kind, {"status": cur.get("status")}
+            continue
+        base = baseline.get(task_id)
+        if not isinstance(base, dict):
+            continue
+        cur_status = cur.get("status")
+        base_status = base.get("status")
+        if cur_status in TERMINAL_TASK_STATUSES \
+                and cur_status != base_status:
+            return task_id, "terminal", {"status": cur_status}
+        _ba, bi, _bf = _summary_numbers(base)
+        ca, ci, cf = _summary_numbers(cur)
+        if ci > bi:
+            return task_id, "needs_input", {
+                "pending_input": ci, "status": cur_status}
+        if ca > _ba:
+            return task_id, "needs_approval", {
+                "pending_approvals": ca, "status": cur_status}
+        if cf > _bf:
+            return task_id, "needs_continue", {
+                "pending_followups": cf, "status": cur_status}
+    return None
+
+
+def poll_for_first_change(fetch_summaries, task_ids, baseline=None,
+                          timeout_seconds=30.0,
+                          poll_seconds=WAIT_POLL_SECONDS):
+    """wait_threads 的轮询核：fetch_summaries() 回
+    {task_id: 摘要}，直到第一个重要变化或超时。
+
+    返回 (changed, summaries, waited_seconds, timed_out)：
+    changed 为 None 表示超时（此时 summaries 是超时那一刻
+    的进度，调用方用它 + 新游标继续等）；changed 非空时
+    为 {"task_id": ..., "kind": ..., "detail": {...}}。
+    fetch 抛异常不吞——调用方按 502 处理，不能把 DB 故障
+    伪装成“超时无变化”。"""
+    try:
+        timeout_seconds = float(timeout_seconds)
+    except (TypeError, ValueError):
+        timeout_seconds = 30.0
+    timeout_seconds = max(0.0, min(float(MAX_WAIT_SECONDS),
+                                   timeout_seconds))
+    start = time.time()
+    summaries = fetch_summaries()
+    found = detect_significant_change(baseline, summaries, task_ids)
+    if found is not None:
+        task_id, kind, detail = found
+        return {"task_id": task_id, "kind": kind,
+                "detail": detail}, summaries, 0.0, False
+    if baseline is None:
+        baseline = summaries
+    while True:
+        elapsed = time.time() - start
+        if elapsed >= timeout_seconds:
+            return None, summaries, elapsed, True
+        time.sleep(min(poll_seconds,
+                       max(0.0, timeout_seconds - elapsed)))
+        summaries = fetch_summaries()
+        found = detect_significant_change(baseline, summaries, task_ids)
+        if found is not None:
+            task_id, kind, detail = found
+            return {"task_id": task_id, "kind": kind,
+                    "detail": detail}, summaries, time.time() - start, False
+
+
+# -- read_pull_request_check_diagnostics：check-runs --------------
+# gh pr checks --json 给出的字段：name / state / bucket /
+# workflow / link / description / event / startedAt /
+# completedAt。state 与 bucket 都可能出现，统一成
+# pass / fail / pending / skipped / cancelled / neutral
+# （bucket 的 skipping -> skipped、cancel -> cancelled）。
+
+def _normalize_check_state(raw):
+    state = str(raw or "").lower()
+    return {"passing": "pass", "failing": "fail", "failed": "fail",
+            "queued": "pending", "waiting": "pending",
+            "in_progress": "pending", "skipping": "skipped",
+            "cancel": "cancelled"}.get(state, state)
+
+
+def _normalize_json_checks(payload):
+    """gh pr checks --json 的输出（list）-> checks 列表。
+    非 list、缺 name 的行直接跳过，不因此让整批诊断失败。"""
+    checks = []
+    if not isinstance(payload, list):
+        return checks
+    for item in payload:
+        if not isinstance(item, dict) or not item.get("name"):
+            continue
+        state = _normalize_check_state(
+            item.get("bucket") or item.get("state"))
+        checks.append({
+            "name": item["name"], "state": state,
+            "workflow": item.get("workflow") or "",
+            "link": item.get("link") or "",
+            "description": item.get("description") or "",
+            "event": item.get("event") or "",
+        })
+    return checks
+
+
+def build_pr_diagnostics(run, pr_number, owner_repo=None, head_sha=None):
+    """PR 诊断的组装核：run(cmd) -> (returncode, stdout, stderr)，
+    server 用 cloud.pr._run 适配它（cwd=worktree，runner 可注入）。
+
+    步骤（任一步失败都不让已拿到的 checks 白费）：
+    1. `gh pr checks --json ...` 拿 workflow / link / description；
+       JSON 解析失败时回落制表符解析（老版 gh / 假 runner）。
+       两路都失败才抛 RuntimeError（调用方转 502）。
+    2. 有 owner_repo + head_sha 且有失败项时，走 check-runs API
+       拿失败日志原文与文件行号注释：先 list 找失败 run 的 id，
+       再逐个（上限 5 个）调单条 check-run 取 output.text 与
+       annotations（path / start_line / message，上限 20 条，
+       text 截 4000 字）。list 或单条失败时 check_runs 回
+       available=False + note，不因此 502——checks 本身已够
+       回答“哪个检查失败”，日志是加分项不是前置条件。
+    返回 dict：checks / all_pass / failed / diagnostics /
+    check_runs（list 或 {available: False, note}）。"""
+    fields = ("bucket,completedAt,description,event,link,name,"
+              "startedAt,state,workflow")
+    rc, out, err = run(["gh", "pr", "checks", str(pr_number),
+                        "--json", fields])
+    checks = []
+    if rc == 0:
+        try:
+            checks = _normalize_json_checks(json.loads(out or ""))
+        except ValueError:
+            checks = []
+    if not checks:
+        rc2, out2, err2 = run(["gh", "pr", "checks", str(pr_number)])
+        if rc2 != 0:
+            raise RuntimeError((err or err2 or "gh pr checks failed")[:500])
+        parsed, _ap, _f, _p = parse_pr_checks(out2 or "")
+        checks = [{"name": c["name"], "state": c["state"],
+                   "workflow": "", "link": "",
+                   "description": "", "event": ""}
+                  for c in parsed]
+    all_pass = bool(checks) and all(
+        c["state"] == "pass" for c in checks)
+    failed = [c["name"] for c in checks if c["state"] == "fail"]
+    diagnostics = ("all checks pass" if all_pass else
+                   ("failed checks: %s" % ", ".join(failed) if failed
+                    else "no checks reported"))
+    check_runs = {"available": False,
+                  "note": "check-runs unavailable: "
+                          "no owner/repo or head sha"}
+    if owner_repo and head_sha and failed:
+        try:
+            check_runs = _fetch_failed_check_runs(
+                run, owner_repo, head_sha, failed)
+        except Exception as exc:
+            check_runs = {"available": False,
+                          "note": "check-runs fetch failed: %s"
+                                  % str(exc)[:200]}
+    return {"checks": checks, "all_pass": all_pass,
+            "failed": failed, "diagnostics": diagnostics,
+            "check_runs": check_runs}
+
+
+def _fetch_failed_check_runs(run, owner_repo, head_sha, failed_names):
+    """check-runs API 的两步取数（只查失败项，上限 5 个）。
+    list 端点回 {check_runs: [...]}；单条回 {id, name,
+    conclusion, output: {title, summary, text, annotations}}。
+    annotations 缺省为空列表，不因缺字段而失败。"""
+    rc, out, _err = run(["gh", "api",
+                         "repos/%s/commits/%s/check-runs"
+                         % (owner_repo, head_sha)])
+    if rc != 0:
+        raise RuntimeError("gh api check-runs list failed")
+    try:
+        payload = json.loads(out or "")
+    except ValueError:
+        raise RuntimeError("gh api check-runs list is not JSON")
+    runs = payload.get("check_runs") if isinstance(payload, dict) else None
+    if not isinstance(runs, list):
+        raise RuntimeError("gh api check-runs list has no check_runs")
+    by_name = {}
+    for item in runs:
+        if isinstance(item, dict) and item.get("name"):
+            by_name.setdefault(item["name"], item)
+    details = []
+    for name in failed_names[:5]:
+        item = by_name.get(name)
+        if item is None or not item.get("id"):
+            details.append({"name": name, "available": False,
+                            "note": "no check-run found for this check"})
+            continue
+        rc2, out2, _e2 = run(["gh", "api",
+                              "repos/%s/check-runs/%s"
+                              % (owner_repo, item["id"])])
+        if rc2 != 0:
+            details.append({"name": name, "available": False,
+                            "note": "check-run fetch failed"})
+            continue
+        try:
+            single = json.loads(out2 or "")
+        except ValueError:
+            details.append({"name": name, "available": False,
+                            "note": "check-run is not JSON"})
+            continue
+        output = single.get("output") if isinstance(single, dict) else None
+        if not isinstance(output, dict):
+            output = {}
+        annotations = []
+        for ann in output.get("annotations") or []:
+            if not isinstance(ann, dict):
+                continue
+            annotations.append({
+                "path": ann.get("path") or "",
+                "start_line": ann.get("start_line"),
+                "end_line": ann.get("end_line"),
+                "level": ann.get("annotation_level") or "",
+                "message": (ann.get("message") or "")[:500],
+            })
+            if len(annotations) >= 20:
+                break
+        details.append({
+            "name": name, "available": True,
+            "check_run_id": item["id"],
+            "conclusion": single.get("conclusion") or "",
+            "title": output.get("title") or "",
+            "summary": (output.get("summary") or "")[:1000],
+            "log_text": (output.get("text") or "")[:4000],
+            "annotations": annotations,
+        })
+    return {"available": True, "runs": details}
+

docs/thread-wait-ops.md

@@ -0,0 +1,180 @@
+# 线程等待与状态读取：与 Codex app 的对照
+
+本批实现把 Codex app 的 8 个线程/等待类工具落到本控制面。
+原则：只读的绝不写（不重跑、不批准、不合并），等待的绝不
+创建任务或发指令；所有“状态”都来自已有记录（任务行、
+trace、审批表、后续表、git 目录），没有的字段明说没有，
+不编造。
+
+## 三类
+
+| 类 | 官方工具 | 本端点 | 说明 |
+|---|---|---|---|
+| 管理并行 Agent | wait_threads | POST /tasks/{id}/wait | 等一批线程的首个重要变化 |
+| 管理并行 Agent | read_thread | GET /tasks/{id}/thread/read | 深入查看单个线程的摘要 |
+| 管理并行 Agent | send_message_to_thread | POST /tasks/{id}/thread/message | 向已授权线程发后续指令 |
+| 读取任务状态 | get_handoff_status | GET /tasks/{id}/handoff | 移交到另一主机后的进度 |
+| 读取任务状态 | get_worktree_creation_status | GET /tasks/{id}/worktree | worktree 准备状态 |
+| 读取任务状态 | read_pull_request_check_diagnostics | GET /tasks/{id}/pr/checks | PR 的 CI 诊断 |
+| 持续监控/等待 | automation_update | GET/PATCH /projects/{id}/automations/{aid} | 长期自动化的查看与修改 |
+| 持续监控/等待 | clock.sleep | POST /tasks/{id}/sleep，host sleep | 一次性等待后继续 |
+
+配套：POST /tasks/{id}/handoff 与 POST /tasks/{id}/handoff/{hid}/report
+是移交的登记与上报口（官方是“先发起 handoff 再轮询”，
+本批把发起与上报也补齐，否则 GET 无数据可轮询）。
+
+## 对照表
+
+### wait_threads — POST /tasks/{id}/wait
+
+- 入参：`task_ids`（可选，缺省取锚任务的 coordinator 子任务），
+  `timeout_seconds`（缺省 30，上限 120），`after_cursor`（可选）。
+- 一次最多 8 个线程，超了回 400（不静默截断）。
+- 只返回第一个发生重要变化的线程：完成/失败等终态、
+  需要用户输入（elicitation 审批）、需要审批（其它审批）、
+  需要继续处理（待后续消息）。变化判断按 task_ids 顺序，
+  不是按真实时间戳排序。
+- 无变化时超时回当前进度（progress 全量 + timed_out=true）。
+- `after_cursor`：上一次返回的 cursor 只含各任务的状态计数，
+  不含正文；带它再等时，已报过的变化不再重复报。
+- 它不创建任务、不发指令，只等和收集状态。
+- 诚实边界：变化检测只读数据库（任务行、审批表、后续表），
+  不轮询 trace 文件；跨任务顺序是调用方给的顺序。
+
+### read_thread — GET /tasks/{id}/thread/read
+
+- 返回：状态映射（queued→waiting，completed/merged→completed，
+  failed/merge-failed→failed，interrupted→interrupted）、
+  线程摘要、最近活动（trace 末 10 条）、改动文件（trace 里
+  kind=file 的路径去重，上限 50）、执行命令（kind=exec 的
+  命令与退出码，末 20 条）、待审批、待后续、Agent 链、
+  阻塞/需要决策/需要处理/需要指令的 attention 汇总。
+- 线程取不到时 thread=null + note，仍回 200（与 /thread 的
+  409/502 不同：读摘要不该因为线程缺失而全单失败）。
+- `test_results` 恒为 null + 说明：trace 里没有独立的测试
+  结果记录，测试只能从命令退出码与 Agent 消息里看，不编
+  “测试通过”。
+- docker 执行器任务没有本机线程，thread=null（这是预期，
+  不是故障）。
+
+### send_message_to_thread — POST /tasks/{id}/thread/message
+
+- 入参：`body`（必填，上限 20000 字），`mode`（queued|direct，
+  缺省 queued）。
+- 授权模型：API 侧要求 API key + 项目 create_task 权限，
+  并记审计 thread_message_sent；“用户在聊天里让我协调”
+  本身不是 API 权限，调用方必须持 key。
+- 排队/运行中任务：写入 task_followups（pending），direct
+  模式同时追加一条直接 trace。
+- 已完成/失败/中断的任务：不是 409，而是复活——用 CAS
+  把状态改回 queued、清空旧结果与旧错误、写入 resume_message；
+  若已有待复活消息或状态已变，回 409（不双重复活）。
+- 已合并任务回 409：worktree 已删，应 fork 新任务而不是复活。
+- 目标任务（goal/verify）回 409：目标任务不接受后续消息。
+- 发给其它任务的消息需要用户明确授权；本任务内按用户指令
+  协调。跨项目发消息不在本端点（走 peer_messages 的
+  /messages，跨项目会被拒绝）。
+
+### get_handoff_status — GET /tasks/{id}/handoff
+
+- 移交状态：pending / accepted / running / completed / failed；
+  终态不可再变，同态重报只算进度更新；进度 0–100，不因
+  终态强制改成 100。
+- 同一任务同一时间只能有一个未关闭移交，否则建移交回 409。
+- 响应含 current_host 与 arrived（当前主机是否等于目标主机）：
+  这是“到了”的证据。
+- 诚实边界：服务端只登记意图与轮询上报，不执行 SSH 迁移、
+  不搬文件；实际迁移由目标主机的 worker 执行并回 report。
+
+### get_worktree_creation_status — GET /tasks/{id}/worktree
+
+- 状态：remote（worktree 在别机）/ ready（目录 + .git）/
+  present_no_git / creating（排队或运行中且目录未建）/
+  missing（终态且目录未建；已合并任务删 worktree 是设计，
+  不是丢失）/ unknown。
+- 本机且已就绪时附带当前分支（git branch --show-current，
+  失败则 null，不阻塞整单）。
+- 诚实边界：worktree 创建在本系统是同步的，没有可观测的
+  “接收对象 / 更新文件”阶段流，不编百分比。
+
+### read_pull_request_check_diagnostics — GET /tasks/{id}/pr/checks
+
+- 只读：不重跑检查、不批准、不合并。
+- 两层取数：`gh pr checks --json`（workflow、link、description
+  即哪个 Action 失败与失败摘要），JSON 失败时回落制表符解析；
+  有仓库 + head sha 且有失败项时，再走 check-runs API 拿
+  失败日志原文（output.text，截 4000 字）与文件行号注释
+  （annotations：path / start_line / message，上限 20 条，
+  失败 run 上限 5 个）。
+- check-runs 一层失败时不 502 整单：checks 本身已够回答
+  “哪个检查失败”，此时 check_runs 回 available=False + note。
+- 无 PR 时 409；worktree 在别机时 409（gh 必须在 worktree
+  所在主机跑；诊断是只读拉取，不排队等对方 worker）。
+
+### automation_update — GET/PATCH /projects/{id}/automations/{aid}
+
+- GET 回定义 + 近期运行（运行取不到时 runs=[] + note，
+  不 502 定义本身）。
+- PATCH 只接受 name / prompt / interval_seconds / repo_url，
+  其它键静默忽略；无可更新字段回 400；已取消的自动化
+  回 409（取消是终态，更新不能复活）；同项目同名冲突回 400。
+- 更新不重置 last_run_at：改间隔只影响下一次到期，绝不
+  触发一次立即运行。
+- 诚实边界：“变化才通知”未实现：每次运行终态仍按现有
+  worker 推送通道通知，没有“与上次输出比对、无变化静默”
+  的抑制逻辑。
+
+### clock.sleep — POST /tasks/{id}/sleep 与 host sleep
+
+- API：read 权限；seconds 缺省 5，只接受 0–60 的数，超出
+  回 400（API 层不静默截断）；睡的是调用方的等待，不暂停、
+  不加速、不改动任务本身；返回时重读任务状态。
+- host：`host.py sleep --seconds N`，0–60，超了退出码 1；
+  Ctrl-C 打印已睡秒数 + interrupted=true，退出码 130。
+
+## 10 步协调流程（映射到端点）
+
+1. 主 Agent 明确需求、影响客户端、验收标准
+   → 写进 POST /tasks 的 prompt（目标任务用 verify_command）。
+2. 为 Android、iOS、macOS、Web 或后端分别派发子任务
+   → POST /tasks，parent_task_id 指向主任务（coordinator）。
+3. 每个子任务在独立 worktree 或明确分支中开发
+   → 任务 worktree 由 worker 分配；GET /tasks/{id}/worktree 查。
+4. 使用 wait_threads 等待任务完成或遇到阻塞
+   → POST /tasks/{主}/wait（缺省即等 coordinator 子任务）。
+5. 使用 read_thread 查看详细结果
+   → GET /tasks/{子}/thread/read。
+6. 对失败任务使用 send_message_to_thread 要求修复
+   → POST /tasks/{子}/thread/message（失败任务会复活续跑）。
+7. 汇总各客户端改动并检查共享 API、事件和数据结构
+   → 各子任务的 read_thread 摘要 + result_text 人工/主 Agent 对照。
+8. 运行跨客户端构建、测试和集成验证
+   → 新建验证任务（或目标任务的 verify_command），再走 4–6。
+9. 读取 PR 的 CI 诊断并继续修复
+   → POST /tasks/{id}/pr 建 PR 后，GET /tasks/{id}/pr/checks，
+   失败项用 thread/message 发回修复。
+10. 最后由主 Agent 汇总变更、证据和剩余风险
+    → 各任务 result_text + trace + 本对照表的诚实边界。
+
+## 派发模板（映射到 POST /tasks 字段）
+
+- 任务：prompt 首行写清任务名与一句话目标。
+- 客户端：prompt 内写明（Android / iOS / macOS / Web / 后端），
+  本系统没有独立的客户端字段，不要另建字段。
+- 仓库：项目 repo_url（建项目时定）；子任务不另设仓库。
+- 基线提交：写进 prompt（如“基线 abc123”），或 environment
+  的分支字段；没有独立的基线字段。
+- 要求：写进 prompt（复用现有组件、不新建导航页等约束）。
+- 验收：写进 prompt；目标任务另用 verify_command 做自动验收。
+- 返回：任务完成时 result_text + trace；read_thread 再给出
+  改动文件、命令与退出码、待处理项。主 Agent 按“改动文件 /
+  执行命令与结果 / 未解决问题”三项核对，不只看一句“已做完”。
+
+## 未实现 / 未宣称
+
+- worktree 创建阶段流（接收对象、更新文件百分比）：无。
+- automation 的“无变化静默”抑制：无。
+- 测试结果独立记录：无（只有命令退出码与 Agent 消息）。
+- handoff 的实际迁移执行：服务端不做，靠目标主机 worker 上报。
+- wait 的 trace 级变化检测：无，只有数据库级变化。
+- 自动注入身份/记忆上下文到任务：不在本批（见 identity-memory）。

host.py

@@ -2211,6 +2211,8 @@ def main():
     p.add_argument("--repo", default=None)
     p = sub.add_parser("pr-merge"); p.add_argument("--task", required=True)
     p.add_argument("--repo", default=None)
+    p = sub.add_parser("sleep"); p.add_argument("--seconds", type=float,
+                                                default=5)
 
     args = ap.parse_args()
     # 配置解析：CLI > 配置文件 > 内置默认。配置文件里绝不放密钥，
@@ -2358,6 +2360,28 @@ def main():
             print(str(e))
             raise SystemExit(1)
         return
+    if args.cmd == "sleep":
+        # clock.sleep 的 host 形态：纯等待，不碰任务、不跑模型、
+        # 不改 state。0-60 秒（与服务端 /sleep 上限同款），
+        # 超限是调用方传错，报错而不是静默钳制；Ctrl-C 打断
+        # 时按惯例 exit 130 并报告已睡时长，不谎称睡满。
+        seconds = args.seconds
+        if seconds is None or seconds < 0 or seconds > 60:
+            print("seconds 必须在 0 到 60 之间")
+            raise SystemExit(1)
+        slept = 0.0
+        try:
+            while slept < seconds:
+                chunk = min(0.5, seconds - slept)
+                time.sleep(chunk)
+                slept += chunk
+        except KeyboardInterrupt:
+            print(json.dumps({"slept_seconds": slept,
+                              "interrupted": True}, ensure_ascii=False))
+            raise SystemExit(130)
+        print(json.dumps({"slept_seconds": slept,
+                          "interrupted": False}, ensure_ascii=False))
+        return
     if args.cmd == "health":
         report = run_health_check(args.state, cfg, args.codex_bin or default_codex_bin(),
                                   args.codex_home, args.approval)

tests/test_host_sleep.py

@@ -0,0 +1,52 @@
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest import mock
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT))
+
+import host
+
+
+class HostSleepTest(unittest.TestCase):
+    def _run(self, cli_args):
+        with tempfile.TemporaryDirectory() as tmp:
+            state_path = os.path.join(tmp, "host-state.json")
+            with open(state_path, "w", encoding="utf-8") as f:
+                json.dump({"tasks": {}, "automations": []}, f)
+            cfg_path = os.path.join(tmp, "cfg.json")
+            with open(cfg_path, "w", encoding="utf-8") as f:
+                json.dump({"state": state_path}, f)
+            argv = ["host.py", "--config", cfg_path,
+                    "--state", state_path] + cli_args
+            out = io.StringIO()
+            code = 0
+            with mock.patch.object(host, "Server",
+                                   lambda **kw: mock.MagicMock()):
+                with mock.patch.object(sys, "argv", argv):
+                    with redirect_stdout(out):
+                        try:
+                            host.main()
+                        except SystemExit as e:
+                            code = e.code if isinstance(e.code, int) else 1
+            return code, out.getvalue()
+
+    def test_sleep_zero(self):
+        code, out = self._run(["sleep", "--seconds", "0"])
+        self.assertEqual(code, 0)
+        self.assertEqual(json.loads(out),
+                         {"slept_seconds": 0.0, "interrupted": False})
+
+    def test_sleep_out_of_range(self):
+        code, _out = self._run(["sleep", "--seconds", "61"])
+        self.assertEqual(code, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()