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

对照 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
This commit is contained in:
host committed 2026-10-28 04:33:20 +08:00
1 parent 38c0797cb3
commit 755029f220
8 files changed
+2535

No files matched your search

+844
View File
@@ -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 模块