From 755029f2202e94769e6b02ac126f161bc187efb6 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 28 Oct 2026 04:33:20 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=BA=BF=E7=A8=8B=E7=AD=89=E5=BE=85?= =?UTF-8?q?=E4=B8=8E=E7=8A=B6=E6=80=81=E8=AF=BB=E5=8F=96=208=20=E9=A1=B9?= =?UTF-8?q?=EF=BC=88wait/read/message/handoff/worktree/pr-checks/automatio?= =?UTF-8?q?n/sleep=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对照 Codex app：wait_threads 首变+afterCursor+8上限、 read_thread 详情摘要、send_message_to_thread 复活失败任务、 handoff 登记与上报轮询、worktree 派生状态、PR checks JSON+ check-runs 日志与注释、automation GET/PATCH、clock.sleep （API+host）。诚实边界见 docs/thread-wait-ops.md：无 worktree 阶段流、无 automation 变化静默、无独立测试结果 记录、handoff 服务端不执行迁移、wait 仅数据库级变化检测。 三套件：cloud 940 OK（4 跳过）、host 91 OK、deploy 29 OK --- cloud/automations.py | 59 ++ cloud/migrations.py | 1 + cloud/server.py | 844 ++++++++++++++++++++++++++++ cloud/tests/test_thread_wait_ops.py | 668 ++++++++++++++++++++++ cloud/wait_ops.py | 707 +++++++++++++++++++++++ docs/thread-wait-ops.md | 180 ++++++ host.py | 24 + tests/test_host_sleep.py | 52 ++ 8 files changed, 2535 insertions(+) create mode 100644 cloud/tests/test_thread_wait_ops.py create mode 100644 cloud/wait_ops.py create mode 100644 docs/thread-wait-ops.md create mode 100644 tests/test_host_sleep.py diff --git a/cloud/automations.py b/cloud/automations.py index 4c21cf1..8c109c5 100644 --- a/cloud/automations.py +++ b/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 为空)或上次执行距 diff --git a/cloud/migrations.py b/cloud/migrations.py index 468de77..1bf9655 100644 --- a/cloud/migrations.py +++ b/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", diff --git a/cloud/server.py b/cloud/server.py index 94af7a0..ce28528 100644 --- a/cloud/server.py +++ b/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 模块 diff --git a/cloud/tests/test_thread_wait_ops.py b/cloud/tests/test_thread_wait_ops.py new file mode 100644 index 0000000..c2ab98e --- /dev/null +++ b/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() diff --git a/cloud/wait_ops.py b/cloud/wait_ops.py new file mode 100644 index 0000000..d62b24c --- /dev/null +++ b/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} + diff --git a/docs/thread-wait-ops.md b/docs/thread-wait-ops.md new file mode 100644 index 0000000..1f0689b --- /dev/null +++ b/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）。 diff --git a/host.py b/host.py index b5420e2..eea399b 100644 --- a/host.py +++ b/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) diff --git a/tests/test_host_sleep.py b/tests/test_host_sleep.py new file mode 100644 index 0000000..44790f9 --- /dev/null +++ b/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()