diff --git a/cloud/lifecycle.py b/cloud/lifecycle.py new file mode 100644 index 0000000..8e0eba3 --- /dev/null +++ b/cloud/lifecycle.py @@ -0,0 +1,288 @@ +"""线程生命周期补充：worktree 快照、任务外部引用、自动化心跳绑定。 + +本模块只补控制面已有能力的缺口，不发明执行能力： + +- worktree 快照（task_worktree_snapshots）：归档线程时真正要 + 保住的是“未提交修改 + 未推送 commit 的位置 + 非忽略文件 + 清单”。快照记 branch / head_sha / git status（截 4000 字）/ + git diff（截 20000 字）/ untracked 路径清单。注意诚实边界： + untracked 文件只记路径不拷内容，diff 只覆盖已跟踪文件的 + 修改；快照是事后可查的证据，不是可一键还原的备份。恢复 + （restore）只在 worktree 目录还在时解归档；目录已删时 + 回 409 并附快照（重建需要仓库路径，服务端没有，不编造）。 +- 任务外部引用（task_refs）：任务附件表只记文件包路径， + 交付物里的“生成的文档 / 构建结果 / 外部服务结果引用” + 没有地方放。refs 只记 kind + ref + note 的引用行，不存 + 文件内容、不校验外部链接可达——引用就是引用。 +- 自动化心跳绑定（automation_bindings）：自动化定义本身 + 只有 interval + prompt，没有“绑到哪个线程”。绑定记 + automation -> task + mode（heartbeat | cron）。heartbeat + 是“到期唤醒已有线程”（走 thread/message 的投递路）； + cron 是“到期独立跑新任务”（走自动化自身的产出路）。 + 本模块只记绑定与取绑定；到期触发仍由外部触发器 + （host run-due / 未来的 run_due_automation）调用 + /heartbeat 端点，服务端没有也不会偷偷起定时器。 + +三张表都走 UNMIGRATED_TABLES（见 cloud/migrations.py）： +运行时 ensure 建表，与 task_handoffs 同款，不进 +POSTGRES_SCHEMA 快照。""" + +import json +import time +import uuid + +SNAPSHOT_STATUS_LIMIT = 4000 +SNAPSHOT_DIFF_LIMIT = 20000 +REF_LIMIT = 2000 +NOTE_LIMIT = 1000 +REF_KINDS = ("doc", "build", "external", "pr", "file") +BINDING_MODES = ("heartbeat", "cron") + + +def _row_get(row, name): + if row is None: + return None + if isinstance(row, dict): + return row.get(name) + try: + return row[name] + except Exception: + return None + + +# -- worktree 快照 ------------------------------------------------------ + +def ensure_snapshots_table(conn): + conn.execute( + "CREATE TABLE IF NOT EXISTS task_worktree_snapshots(" + "id TEXT PRIMARY KEY, " + "task_id TEXT, " + "project_id TEXT, " + "worktree_path TEXT, " + "branch TEXT, " + "head_sha TEXT, " + "status_text TEXT, " + "diff_text TEXT, " + "untracked_json TEXT, " + "created_at DOUBLE PRECISION, " + "restored_at DOUBLE PRECISION, " + "restore_note TEXT)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_snapshots_task" + " ON task_worktree_snapshots(task_id, created_at)") + + +def snapshot_to_dict(row): + try: + untracked = json.loads(_row_get(row, "untracked_json") or "[]") + except ValueError: + untracked = [] + if not isinstance(untracked, list): + untracked = [] + return { + "id": _row_get(row, "id"), + "task_id": _row_get(row, "task_id"), + "project_id": _row_get(row, "project_id"), + "worktree_path": _row_get(row, "worktree_path"), + "branch": _row_get(row, "branch"), + "head_sha": _row_get(row, "head_sha"), + "status_text": _row_get(row, "status_text"), + "diff_text": _row_get(row, "diff_text"), + "untracked": untracked, + "created_at": _row_get(row, "created_at"), + "restored_at": _row_get(row, "restored_at"), + "restore_note": _row_get(row, "restore_note"), + } + + +def create_snapshot(conn, task_id, project_id, worktree_path, branch, + head_sha, status_text, diff_text, untracked, now=None): + """插一条快照并提交。status/diff 在调用方截断，这里再截 + 一次防直接调用方越界；untracked 只接受 list[str]。""" + ensure_snapshots_table(conn) + snap_id = uuid.uuid4().hex + now = time.time() if now is None else now + untracked = [str(p) for p in (untracked or []) if isinstance(p, str)] + conn.execute( + "INSERT INTO task_worktree_snapshots(id, task_id, project_id," + " worktree_path, branch, head_sha, status_text, diff_text," + " untracked_json, created_at, restored_at, restore_note)" + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL)", + (snap_id, task_id, project_id, worktree_path, branch, + head_sha, (status_text or "")[:SNAPSHOT_STATUS_LIMIT], + (diff_text or "")[:SNAPSHOT_DIFF_LIMIT], + json.dumps(untracked), now)) + conn.commit() + row = conn.execute( + "SELECT * FROM task_worktree_snapshots WHERE id = ?", + (snap_id,)).fetchone() + return snapshot_to_dict(row) + + +def list_snapshots(conn, task_id): + ensure_snapshots_table(conn) + rows = conn.execute( + "SELECT * FROM task_worktree_snapshots WHERE task_id = ?" + " ORDER BY created_at DESC, id", + (task_id,)).fetchall() + return [snapshot_to_dict(r) for r in rows] + + +def latest_snapshot(conn, task_id): + rows = list_snapshots(conn, task_id) + return rows[0] if rows else None + + +def mark_restored(conn, snapshot_id, note, now=None): + ensure_snapshots_table(conn) + now = time.time() if now is None else now + cur = conn.execute( + "UPDATE task_worktree_snapshots SET restored_at = ?," + " restore_note = ? WHERE id = ?", + (now, note, snapshot_id)) + conn.commit() + if cur.rowcount != 1: + return None + row = conn.execute( + "SELECT * FROM task_worktree_snapshots WHERE id = ?", + (snapshot_id,)).fetchone() + return snapshot_to_dict(row) + + +# -- 任务外部引用 --------------------------------------------------------- + +def ensure_refs_table(conn): + conn.execute( + "CREATE TABLE IF NOT EXISTS task_refs(" + "id TEXT PRIMARY KEY, " + "task_id TEXT, " + "project_id TEXT, " + "kind TEXT, " + "ref TEXT, " + "note TEXT, " + "created_by TEXT, " + "created_at DOUBLE PRECISION)") + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_refs_task" + " ON task_refs(task_id, created_at)") + + +def ref_to_dict(row): + return { + "id": _row_get(row, "id"), + "task_id": _row_get(row, "task_id"), + "project_id": _row_get(row, "project_id"), + "kind": _row_get(row, "kind"), + "ref": _row_get(row, "ref"), + "note": _row_get(row, "note"), + "created_by": _row_get(row, "created_by"), + "created_at": _row_get(row, "created_at"), + } + + +def add_ref(conn, task_id, project_id, kind, ref, note, created_by, + now=None): + """校验后插引用行。kind 不在白名单、ref 为空或超长、 + note 超长时抛 ValueError（调用方转 400）。""" + if kind not in REF_KINDS: + raise ValueError("kind must be one of %s" % ", ".join(REF_KINDS)) + if not ref or not isinstance(ref, str) or not ref.strip(): + raise ValueError("ref required") + ref = ref.strip() + if len(ref) > REF_LIMIT: + raise ValueError("ref too long") + if note is not None and len(str(note)) > NOTE_LIMIT: + raise ValueError("note too long") + ensure_refs_table(conn) + ref_id = uuid.uuid4().hex + now = time.time() if now is None else now + conn.execute( + "INSERT INTO task_refs(id, task_id, project_id, kind, ref," + " note, created_by, created_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?)", + (ref_id, task_id, project_id, kind, ref, note, created_by, + now)) + conn.commit() + row = conn.execute( + "SELECT * FROM task_refs WHERE id = ?", (ref_id,)).fetchone() + return ref_to_dict(row) + + +def list_refs(conn, task_id): + ensure_refs_table(conn) + rows = conn.execute( + "SELECT * FROM task_refs WHERE task_id = ?" + " ORDER BY created_at, id", + (task_id,)).fetchall() + return [ref_to_dict(r) for r in rows] + + +def delete_ref(conn, task_id, ref_id): + """只删属于该任务的引用行；行不存在或不属于该任务时 + 返回 False（调用方一律 404，不泄露跨任务存在性）。""" + ensure_refs_table(conn) + cur = conn.execute( + "DELETE FROM task_refs WHERE id = ? AND task_id = ?", + (ref_id, task_id)) + conn.commit() + return cur.rowcount == 1 + + +# -- 自动化心跳绑定 -------------------------------------------------------- + +def ensure_bindings_table(conn): + conn.execute( + "CREATE TABLE IF NOT EXISTS automation_bindings(" + "automation_id TEXT PRIMARY KEY, " + "project_id TEXT, " + "task_id TEXT, " + "mode TEXT, " + "created_at DOUBLE PRECISION)") + + +def binding_to_dict(row): + if row is None: + return None + return { + "automation_id": _row_get(row, "automation_id"), + "project_id": _row_get(row, "project_id"), + "task_id": _row_get(row, "task_id"), + "mode": _row_get(row, "mode"), + "created_at": _row_get(row, "created_at"), + } + + +def bind_automation(conn, automation_id, project_id, task_id, mode, + now=None): + if mode not in BINDING_MODES: + raise ValueError("mode must be heartbeat or cron") + ensure_bindings_table(conn) + now = time.time() if now is None else now + conn.execute( + "INSERT INTO automation_bindings(automation_id, project_id," + " task_id, mode, created_at) VALUES(?, ?, ?, ?, ?)" + " ON CONFLICT(automation_id) DO UPDATE SET" + " task_id = excluded.task_id, mode = excluded.mode," + " created_at = excluded.created_at", + (automation_id, project_id, task_id, mode, now)) + conn.commit() + row = conn.execute( + "SELECT * FROM automation_bindings WHERE automation_id = ?", + (automation_id,)).fetchone() + return binding_to_dict(row) + + +def get_binding(conn, automation_id): + ensure_bindings_table(conn) + row = conn.execute( + "SELECT * FROM automation_bindings WHERE automation_id = ?", + (automation_id,)).fetchone() + return binding_to_dict(row) + + +def unbind_automation(conn, automation_id): + ensure_bindings_table(conn) + cur = conn.execute( + "DELETE FROM automation_bindings WHERE automation_id = ?", + (automation_id,)) + conn.commit() + return cur.rowcount == 1 diff --git a/cloud/migrations.py b/cloud/migrations.py index 1bf9655..554152e 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -105,6 +105,9 @@ UNMIGRATED_TABLES = ( "external_identities", "pr_jobs", "task_handoffs", + "task_worktree_snapshots", + "task_refs", + "automation_bindings", "rate_limit_hits", "user_gateway_keys", "webhook_deliveries", diff --git a/cloud/server.py b/cloud/server.py index ce28528..1082662 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -76,6 +76,16 @@ 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.lifecycle import add_ref as _life_add_ref +from cloud.lifecycle import bind_automation as _life_bind +from cloud.lifecycle import create_snapshot as _life_snapshot +from cloud.lifecycle import delete_ref as _life_delete_ref +from cloud.lifecycle import get_binding as _life_get_binding +from cloud.lifecycle import latest_snapshot as _life_latest_snapshot +from cloud.lifecycle import list_refs as _life_list_refs +from cloud.lifecycle import list_snapshots as _life_list_snapshots +from cloud.lifecycle import mark_restored as _life_mark_restored +from cloud.lifecycle import unbind_automation as _life_unbind 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 @@ -1916,6 +1926,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): elif rest.endswith("/worktree") and method == "GET": self._task_worktree_status(user_id, rest[: -len("/worktree")]) + elif rest.endswith("/bindings") and method == "GET": + self._task_bindings(user_id, + rest[: -len("/bindings")]) + elif rest.endswith("/refs") and method == "GET": + self._task_refs_list(user_id, rest[: -len("/refs")]) + elif rest.endswith("/refs") and method == "POST": + self._task_refs_create(user_id, rest[: -len("/refs")]) + elif "/refs/" in rest and method == "DELETE": + _task_part, _ref_part = rest.split("/refs/", 1) + self._task_refs_delete(user_id, _task_part, _ref_part) + elif rest.endswith("/worktree/archive") and method == "POST": + self._task_worktree_archive( + user_id, rest[: -len("/worktree/archive")]) + elif rest.endswith("/worktree/restore") and method == "POST": + self._task_worktree_restore( + user_id, rest[: -len("/worktree/restore")]) + elif rest.endswith("/worktree/snapshots") and method == "GET": + self._task_worktree_snapshots( + user_id, rest[: -len("/worktree/snapshots")]) elif rest.endswith("/pr/checks") and method == "GET": self._task_pr_checks(user_id, rest[: -len("/pr/checks")]) @@ -2769,6 +2798,40 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): return self._automation_cancel(user_id, project_id, parts[2]) return + # /projects/{id}/automations/{aid}/bind（GET/POST/DELETE） + # / /heartbeat（POST）：心跳绑定与唤醒，见 handlers。 + if len(parts) == 4 and parts[1] == "automations" \ + and parts[3] in ("bind", "heartbeat"): + if parts[3] == "bind" and method == "GET": + err = self._check_project(user_id, project_id, "read") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._automation_bind_get(user_id, project_id, parts[2]) + elif parts[3] == "bind" and method == "POST": + err = self._check_project(user_id, project_id, + "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._automation_bind(user_id, project_id, parts[2]) + elif parts[3] == "bind" and method == "DELETE": + err = self._check_project(user_id, project_id, + "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._automation_unbind(user_id, project_id, parts[2]) + elif parts[3] == "heartbeat" and method == "POST": + err = self._check_project(user_id, project_id, + "create_task") + if err: + self._send_json(err[0], {"error": err[1]}) + return + self._automation_heartbeat(user_id, project_id, parts[2]) + else: + self._send_json(404, {"error": "not found"}) + return # /projects/{id}/triage if len(parts) == 2 and parts[1] == "triage": if method != "GET": @@ -4542,9 +4605,17 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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 "") + if entry.get("kind") == "tool": + # tool 条目没有 text/command，旧代码只取 tool 名， + # 丢了 server 与 status；拼成可读的一行。 + text = "%s/%s: %s" % ( + entry.get("server") or "", + entry.get("tool") or "", + entry.get("status") or "") + else: + 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]} @@ -4686,14 +4757,25 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if mode not in ("queued", "direct"): self._send_json(400, {"error": "mode must be queued or direct"}) return + code, payload = self._deliver_to_thread( + user_id, task, text, mode) + self._send_json(code, payload) + + def _deliver_to_thread(self, user_id, task, text, mode, + audit_action="thread_message_sent"): + """给线程投递后续指令的共用实现（thread/message 与 + 自动化 heartbeat 共用同一路，不各写一套复活逻辑）： + goal 任务 409、merged 409、终态任务 CAS 复活（清旧 + 结果/旧错误）、进行中任务写 followup（direct 再追加 + 直接 trace）。返回 (http_code, payload)，不直接 + _send_json；配额检查由调用方在调本方法前做。""" + task_id = task["id"] if task.get("verify_command") or task.get("goal_mode"): - self._send_json(409, {"error": "goal 任务暂不支持后续消息"}) - return + return 409, {"error": "goal 任务暂不支持后续消息"} if task["status"] == "merged": - self._send_json(409, { + return 409, { "error": "已合并任务不能继续（worktree 已删），请分支", - "status": task["status"]}) - return + "status": task["status"]} if task["status"] in TERMINAL_STATUSES: # 复活路：与 /continue 同款终态集（merged 已在上 # 面排除），CAS 防“已有待续跑消息”时重复复活。 @@ -4710,22 +4792,19 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): (text, now, task_id)) conn.commit() if cur.rowcount != 1: - self._send_json(409, { + return 409, { "error": "target has a pending resume message" " or status changed", - "status": task["status"]}) - return + "status": task["status"]} finally: conn.close() with _auth_store(self.server.db_path) as store: - store.audit(user_id, task["project_id"], - "thread_message_sent", + store.audit(user_id, task["project_id"], audit_action, "%s:revived" % task_id) - self._send_json(201, { + return 201, { "task_id": task_id, "delivery": "revived", "followup_id": None, "status": "queued", - "mode": mode}) - return + "mode": mode} now = time.time() conn = self._db() try: @@ -4747,11 +4826,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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", + store.audit(user_id, task["project_id"], audit_action, "%s:%s" % (task_id, fu_id)) - self._send_json(201, dict(fu, delivery="queued" - if mode == "queued" else "direct")) + return 201, dict(fu, delivery="queued" + if mode == "queued" else "direct") def _task_handoff_create(self, user_id, task_id): """POST /tasks/{id}/handoff：登记一次移交意图 @@ -4850,12 +4928,312 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): else: self._send_json(400, {"error": result}) return + if result.get("status") == "completed": + # 移交完成即执行位置切换的落点：任务的 worktree_host + # 改成目标主机，后续 /worktree、/pr/checks、线程解析 + # 都按新主机判断；失败/进行中的上报不改主机。 + conn = self._db() + try: + conn.execute( + "UPDATE tasks SET worktree_host = ?," + " updated_at = ? WHERE id = ?", + (result.get("target_host"), time.time(), task_id)) + conn.commit() + finally: + conn.close() 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 _worktree_summary(self, task): + """worktree 摘要（bindings 与 /worktree 共用的读逻辑）： + remote 时 exists/is_git/branch 一律 null（本机查不到， + 不谎报）；本机时读目录 + .git + 分支（尽力，失败 null）。""" + 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 + branch = None + if remote is None: + exists = os.path.isdir(path) + is_git = os.path.exists(os.path.join(path, ".git")) \ + if exists else False + if exists and is_git: + 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 + 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) + return dict(status, path=path, branch=branch, + environment_id=task.get("environment_id"), + remote_host=remote) + + def _task_bindings(self, user_id, task_id): + """GET /tasks/{id}/bindings（任务与 Git 状态绑定 + + 项目/主机上下文）：把散在任务行、项目行、环境行、 + worktree、PR、附件、外部引用里的一行拼成一张总表， + 供跨线程协调时一次取全。只读聚合，不产生新状态； + 读权限 read。local/remote 按 worktree_host 与本机 + 主机名判定；default_executor 是全局缺省，任务行 + 的 executor 为空时才生效。""" + 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: + proj = conn.execute( + "SELECT id, name, repo_url FROM projects WHERE id = ?", + (task["project_id"],)).fetchone() + atts = _attachments_map(conn, [task_id])[task_id] + refs = _life_list_refs(conn, task_id) + snaps = _life_list_snapshots(conn, task_id) + finally: + conn.close() + env = None + if task.get("environment_id"): + env_row = self._fetch_environment(task["environment_id"]) + if env_row is not None: + env = _environment_to_dict(env_row) + worktree = self._worktree_summary(task) + self._send_json(200, { + "task_id": task_id, + "task_status": task["status"], + "project": ({ + "id": _row_get(proj, "id"), + "name": _row_get(proj, "name"), + "repo_url": _row_get(proj, "repo_url"), + } if proj is not None else None), + "environment": env, + "host": { + "worktree_host": task.get("worktree_host"), + "local": worktree["remote_host"] is None, + "current_host": socket.gethostname(), + }, + "worktree": worktree, + "branch": worktree["branch"], + "pull_request": ({ + "url": task.get("pr_url"), + "number": task.get("pr_number"), + } if task.get("pr_number") else None), + "executor": task.get("executor") or _default_executor(), + "attachments": { + "files": atts, + "refs": refs, + }, + "worktree_snapshots": [{ + "id": s["id"], "branch": s["branch"], + "head_sha": s["head_sha"], + "created_at": s["created_at"], + "restored_at": s["restored_at"], + } for s in snaps], + "lineage": { + "parent_task_id": task.get("parent_task_id"), + "forked_from": task.get("forked_from"), + "coordinator_part": task.get("coordinator_part"), + }, + }) + + def _task_refs_list(self, user_id, task_id): + 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: + refs = _life_list_refs(conn, task_id) + finally: + conn.close() + self._send_json(200, {"task_id": task_id, "refs": refs}) + + def _task_refs_create(self, user_id, task_id): + """POST /tasks/{id}/refs：记一条外部引用（生成的文档 / + 构建结果 / 外部服务结果）。只记引用，不存内容、不 + 校验链接可达；kind 白名单见 lifecycle.REF_KINDS。 + 权限 create_task。""" + 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 + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + conn = self._db() + try: + try: + ref = _life_add_ref( + conn, task_id, task["project_id"], + body.get("kind"), body.get("ref"), + body.get("note"), user_id) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], "task_ref_added", + "%s:%s" % (task_id, ref["id"])) + self._send_json(201, ref) + + def _task_refs_delete(self, user_id, task_id, ref_id): + 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: + ok = _life_delete_ref(conn, task_id, ref_id) + finally: + conn.close() + if not ok: + self._send_json(404, {"error": "ref not found"}) + return + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], "task_ref_deleted", + "%s:%s" % (task_id, ref_id)) + self._send_json(200, {"task_id": task_id, "ref_id": ref_id, + "deleted": True}) + + def _git_text(self, path, args, limit): + """worktree 上尽力跑一条只读 git 命令；git 不在、 + 不是仓库、超时都回空串，不拖垮快照。""" + try: + proc = subprocess.run( + ["git", "-C", path] + args, + capture_output=True, text=True, timeout=10) + if proc.returncode != 0: + return "" + return (proc.stdout or "")[:limit] + except Exception: + return "" + + def _task_worktree_archive(self, user_id, task_id): + """POST /tasks/{id}/worktree/archive：给当前 worktree + 拍一张快照（branch / head / status / diff / untracked + 路径），行进 task_worktree_snapshots。快照不删目录、 + 不删分支——删目录是 merge/discard 的职责，归档只是 + “先把证据存下来”。worktree 在别机时 409（本机 + git 读不到对方目录）；目录不存在时 409（无可拍）。 + 权限 create_task；记 worktree_snapshot_created 审计。""" + 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 + summary = self._worktree_summary(task) + if summary["remote_host"] is not None: + self._send_json(409, { + "error": "task worktree is on host %s;" + " snapshot must run there" + % summary["remote_host"], + "worktree_host": summary["remote_host"]}) + return + path = summary["path"] + if not summary["exists"]: + self._send_json(409, { + "error": "worktree directory does not exist", + "path": path}) + return + status_text = self._git_text( + path, ["status", "--porcelain"], 4000) \ + if summary["is_git"] else "" + diff_text = self._git_text( + path, ["diff", "HEAD"], 20000) if summary["is_git"] else "" + head_sha = None + if summary["is_git"]: + head_sha = self._git_text( + path, ["rev-parse", "HEAD"], 100).strip() or None + untracked = [line[3:] for line in status_text.splitlines() + if line.startswith("?? ")] + conn = self._db() + try: + snap = _life_snapshot( + conn, task_id, task["project_id"], path, + summary["branch"], head_sha, status_text, diff_text, + untracked) + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], + "worktree_snapshot_created", + "%s:%s" % (task_id, snap["id"])) + self._send_json(201, { + "task_id": task_id, "snapshot_id": snap["id"], + "branch": snap["branch"], "head_sha": snap["head_sha"], + "untracked_count": len(snap["untracked"]), + "status_lines": len( + (snap["status_text"] or "").splitlines()), + "note": ("snapshot only: untracked files are listed," + " not copied; restore does not recreate a" + " deleted directory"), + }) + + def _task_worktree_snapshots(self, user_id, task_id): + 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: + snaps = _life_list_snapshots(conn, task_id) + finally: + conn.close() + self._send_json(200, {"task_id": task_id, "snapshots": snaps}) + + def _task_worktree_restore(self, user_id, task_id): + """POST /tasks/{id}/worktree/restore：解归档最新快照。 + 目录还在时只把快照标 restored（目录本身没动过，谈 + 不上“还原”）；目录已删时 409 并附快照（重建需要 + 仓库路径与分支检出，服务端没有，不编一个假恢复）。 + 权限 create_task。""" + 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: + snap = _life_latest_snapshot(conn, task_id) + finally: + conn.close() + if snap is None: + self._send_json(409, {"error": "no snapshot to restore"}) + return + if not os.path.isdir(snap["worktree_path"]): + self._send_json(409, { + "error": "worktree directory is gone; snapshot" + " cannot recreate it without the repo", + "snapshot": snap}) + return + conn = self._db() + try: + snap = _life_mark_restored( + conn, snap["id"], + "worktree directory still present; snapshot" + " unarchived only") + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, task["project_id"], + "worktree_snapshot_restored", + "%s:%s" % (task_id, snap["id"])) + self._send_json(200, {"task_id": task_id, "snapshot": snap}) + def _task_worktree_status(self, user_id, task_id): """GET /tasks/{id}/worktree（get_worktree_creation_status）： 派生状态，不是异步创建任务的轮询——云端 worktree 由 @@ -4868,41 +5246,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 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"))) + summary = self._worktree_summary(task) + self._send_json(200, dict(summary, task_id=task_id)) def _task_pr_checks(self, user_id, task_id): """GET /tasks/{id}/pr/checks（read_pull_request_check_diagnostics）： @@ -4985,6 +5330,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "task_id": task_id, "pr_url": task.get("pr_url"), "pr_number": task["pr_number"], + "head_sha": head_sha, "checks": diag["checks"], "all_pass": diag["all_pass"], "failed": diag["failed"], @@ -7295,8 +7641,146 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): if auto is None or auto["project_id"] != project_id: self._send_json(404, {"error": "automation not found"}) return + conn = self._db() + try: + binding = _life_get_binding(conn, automation_id) + finally: + conn.close() self._send_json(200, dict(auto, runs=runs, - runs_note=runs_note)) + runs_note=runs_note, + heartbeat_binding=binding)) + + def _automation_bind_get(self, user_id, project_id, automation_id): + conn = self._db() + try: + auto = _auto_get(conn, automation_id) + binding = _life_get_binding(conn, automation_id) + 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, {"automation_id": automation_id, + "binding": binding}) + + def _automation_bind(self, user_id, project_id, automation_id): + """POST .../automations/{aid}/bind {task_id, mode}： + 把自动化绑到已有线程。mode=heartbeat 表示到期唤醒 + 该线程（走 thread/message 投递路）；mode=cron 表示 + 到期独立跑新任务（绑定只作标记，不唤醒）。目标任务 + 必须在同一项目，否则 404（不泄露跨项目存在性）。 + 权限 create_task；记 automation_bound 审计。""" + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + conn = self._db() + try: + auto = _auto_get(conn, automation_id) + if auto is None or auto["project_id"] != project_id: + self._send_json(404, {"error": "automation not found"}) + return + if auto["status"] != "active": + self._send_json(409, { + "error": "cancelled automation cannot be bound"}) + return + target_id = body.get("task_id") + target = self._fetch_task(target_id) \ + if isinstance(target_id, str) else None + if target is None: + self._send_json(404, {"error": "task not found"}) + return + target = _task_to_dict(target) + if target["project_id"] != project_id: + self._send_json(404, {"error": "task not found"}) + return + try: + binding = _life_bind( + conn, automation_id, project_id, target_id, + body.get("mode", "heartbeat")) + except ValueError as exc: + self._send_json(400, {"error": str(exc)}) + return + finally: + conn.close() + with _auth_store(self.server.db_path) as store: + store.audit(user_id, project_id, "automation_bound", + "%s:%s" % (automation_id, target_id)) + self._send_json(200, {"automation_id": automation_id, + "binding": binding}) + + def _automation_unbind(self, user_id, project_id, automation_id): + conn = self._db() + try: + auto = _auto_get(conn, automation_id) + if auto is None or auto["project_id"] != project_id: + self._send_json(404, {"error": "automation not found"}) + return + ok = _life_unbind(conn, automation_id) + finally: + conn.close() + if not ok: + self._send_json(404, {"error": "no binding"}) + return + with _auth_store(self.server.db_path) as store: + store.audit(user_id, project_id, "automation_unbound", + automation_id) + self._send_json(200, {"automation_id": automation_id, + "deleted": True}) + + def _automation_heartbeat(self, user_id, project_id, automation_id): + """POST .../automations/{aid}/heartbeat {message?}： + 手动触发一次心跳唤醒——到期触发器（host run-due / + run_due_automation）调的就是这个口，服务端本身不 + 起定时器。无绑定 409；mode=cron 409（cron 不唤醒 + 线程）；message 缺省用自动化 prompt。投递走 + _deliver_to_thread，与 thread/message 同款复活路， + 配额门同款。记 automation_heartbeat_sent 审计。""" + body, jerr = self._read_json() + if jerr: + self._send_json(400, {"error": jerr}) + return + conn = self._db() + try: + auto = _auto_get(conn, automation_id) + binding = _life_get_binding(conn, automation_id) + finally: + conn.close() + if auto is None or auto["project_id"] != project_id: + self._send_json(404, {"error": "automation not found"}) + return + if binding is None: + self._send_json(409, { + "error": "automation has no heartbeat binding"}) + return + if binding["mode"] != "heartbeat": + self._send_json(409, { + "error": "cron binding does not wake a thread;" + " use the automation run path"}) + return + target_row = self._fetch_task(binding["task_id"]) + if target_row is None: + self._send_json(404, {"error": "bound task not found"}) + return + target = _task_to_dict(target_row) + if target["project_id"] != project_id: + self._send_json(404, {"error": "bound task not found"}) + return + if self._reject_if_over_quota_for_task(user_id, target): + return + message = body.get("message") or auto.get("prompt") or "" + if not isinstance(message, str) or not message.strip(): + self._send_json(400, {"error": "message required"}) + return + message = message.strip() + if len(message) > 20000: + self._send_json(400, {"error": "message too long"}) + return + code, payload = self._deliver_to_thread( + user_id, target, message, "queued", + audit_action="automation_heartbeat_sent") + self._send_json(code, dict(payload, + automation_id=automation_id)) def _automation_update(self, user_id, project_id, automation_id): """PATCH 单条自动化（automation_update）：改名、改 diff --git a/cloud/tests/test_thread_lifecycle.py b/cloud/tests/test_thread_lifecycle.py new file mode 100644 index 0000000..499c677 --- /dev/null +++ b/cloud/tests/test_thread_lifecycle.py @@ -0,0 +1,298 @@ +"""线程生命周期补充单测：cloud/lifecycle 纯逻辑 + server 端点。 + +覆盖：bindings 聚合、外部引用增删查、心跳绑定与唤醒、 +worktree 快照归档/恢复、移交完成更新 worktree_host、 +PR checks 补 head_sha。不跑真实模型、不联网、不调真实 +git（快照测试用空 .git 目录，git 命令失败时仍能建快照， +只 branch/head 为空——这是诚实降级分支本身）。""" + +import http.client +import json +import os +import shutil +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.server import create_server + from cloud.lifecycle import (add_ref, bind_automation, + create_snapshot, get_binding, + latest_snapshot, list_refs, + list_snapshots) +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.lifecycle import (add_ref, bind_automation, + create_snapshot, get_binding, + latest_snapshot, list_refs, + list_snapshots) + + +class LifecyclePureTestCase(unittest.TestCase): + def setUp(self): + self.conn = sqlite3.connect(":memory:") + self.conn.row_factory = sqlite3.Row + + def tearDown(self): + self.conn.close() + + def test_snapshot_truncates_and_lists(self): + snap = create_snapshot( + self.conn, "t1", "p1", "/tmp/wt", "main", "abc", + "M a.py", "diff " * 10000, ["untracked.txt"]) + self.assertLessEqual(len(snap["diff_text"]), 20000) + self.assertEqual(snap["untracked"], ["untracked.txt"]) + self.assertIsNone(snap["restored_at"]) + self.assertEqual(latest_snapshot(self.conn, "t1")["id"], + snap["id"]) + self.assertEqual(len(list_snapshots(self.conn, "t1")), 1) + + def test_ref_validation(self): + ref = add_ref(self.conn, "t1", "p1", "doc", + "https://example.invalid/spec.md", "规格", "u1") + self.assertEqual(ref["kind"], "doc") + with self.assertRaises(ValueError): + add_ref(self.conn, "t1", "p1", "bogus", "x", None, "u1") + with self.assertRaises(ValueError): + add_ref(self.conn, "t1", "p1", "doc", " ", None, "u1") + self.assertEqual(len(list_refs(self.conn, "t1")), 1) + + def test_binding_mode_and_overwrite(self): + b = bind_automation(self.conn, "a1", "p1", "t1", "heartbeat") + self.assertEqual(b["mode"], "heartbeat") + b = bind_automation(self.conn, "a1", "p1", "t2", "cron") + self.assertEqual((b["task_id"], b["mode"]), ("t2", "cron")) + self.assertEqual(get_binding(self.conn, "a1")["task_id"], "t2") + with self.assertRaises(ValueError): + bind_automation(self.conn, "a2", "p1", "t1", "weekly") + + +class LifecycleServerTestCase(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.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") + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + 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, **extra): + body = {"project_id": 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 _worktree_dir(self, task_id): + # _thread_worktree_for_task 的算法：host_worktrees_dir + + # 安全化后的任务 id（本测试任务 id 只含 hex，无需转义）。 + return os.path.join(self.server.host_worktrees_dir, task_id) + + def test_bindings_aggregates(self): + task = self._create_task() + self._set_task(task["id"], pr_number=7, + pr_url="https://example.invalid/pr/7") + status, body = self._req( + "GET", "/tasks/%s/bindings" % task["id"], None, + self.reader_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["project"]["name"], "demo") + self.assertEqual(body["pull_request"]["number"], 7) + self.assertEqual(body["worktree"]["status"], "creating") + self.assertTrue(body["host"]["local"]) + self.assertEqual(body["lineage"]["parent_task_id"], None) + self.assertEqual(body["attachments"]["files"], []) + status, _ = self._req( + "GET", "/tasks/%s/bindings" % task["id"], None, + self.outsider_key) + self.assertEqual(status, 403) + + def test_refs_crud(self): + task = self._create_task() + status, ref = self._req( + "POST", "/tasks/%s/refs" % task["id"], + {"kind": "doc", "ref": "https://example.invalid/spec.md", + "note": "规格"}, self.owner_key) + self.assertEqual(status, 201, ref) + status, body = self._req( + "POST", "/tasks/%s/refs" % task["id"], + {"kind": "bogus", "ref": "x"}, self.owner_key) + self.assertEqual(status, 400, body) + status, body = self._req( + "GET", "/tasks/%s/refs" % task["id"], None, self.reader_key) + self.assertEqual(len(body["refs"]), 1) + status, body = self._req( + "DELETE", "/tasks/%s/refs/%s" % (task["id"], ref["id"]), + None, self.owner_key) + self.assertEqual(status, 200, body) + status, body = self._req( + "DELETE", "/tasks/%s/refs/%s" % (task["id"], ref["id"]), + None, self.owner_key) + self.assertEqual(status, 404, body) + + def test_worktree_snapshot_and_restore(self): + task = self._create_task() + wt = self._worktree_dir(task["id"]) + os.makedirs(os.path.join(wt, ".git"), exist_ok=True) + status, body = self._req( + "POST", "/tasks/%s/worktree/archive" % task["id"], + {}, self.owner_key) + self.assertEqual(status, 201, body) + self.assertEqual(body["untracked_count"], 0) + status, body = self._req( + "GET", "/tasks/%s/worktree/snapshots" % task["id"], + None, self.reader_key) + self.assertEqual(len(body["snapshots"]), 1) + status, body = self._req( + "POST", "/tasks/%s/worktree/restore" % task["id"], + {}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertIsNotNone(body["snapshot"]["restored_at"]) + # 目录删了之后再恢复：409 并附快照，不编假恢复。 + shutil.rmtree(wt) + status, body = self._req( + "POST", "/tasks/%s/worktree/archive" % task["id"], + {}, self.owner_key) + self.assertEqual(status, 409, body) + # 先再拍一张是不可能的（目录已删），直接用旧快照恢复： + status, body = self._req( + "POST", "/tasks/%s/worktree/restore" % task["id"], + {}, self.owner_key) + self.assertEqual(status, 409, body) + self.assertIn("snapshot", body) + + def test_handoff_completed_moves_host(self): + task = self._create_task() + status, h = self._req( + "POST", "/tasks/%s/handoff" % task["id"], + {"target_host": "host-b"}, self.owner_key) + self.assertEqual(status, 201, h) + for st in ("running", "completed"): + status, body = self._req( + "POST", "/tasks/%s/handoff/%s/report" + % (task["id"], h["id"]), + {"status": st}, self.owner_key) + self.assertEqual(status, 200, body) + status, task_after = self._req( + "GET", "/tasks/%s" % task["id"], None, self.owner_key) + self.assertEqual(task_after["worktree_host"], "host-b") + + def test_automation_heartbeat_binding(self): + status, auto = self._req( + "POST", "/projects/%s/automations" % self.project["id"], + {"name": "watch", "prompt": "检查部署健康", + "interval_seconds": 3600}, self.owner_key) + self.assertEqual(status, 201, auto) + aid = auto["id"] + task = self._create_task() + self._set_task(task["id"], status="running") + # 未绑定时心跳 409。 + status, body = self._req( + "POST", "/projects/%s/automations/%s/heartbeat" + % (self.project["id"], aid), + {"message": "查一下"}, self.owner_key) + self.assertEqual(status, 409, body) + status, body = self._req( + "POST", "/projects/%s/automations/%s/bind" + % (self.project["id"], aid), + {"task_id": task["id"], "mode": "heartbeat"}, + self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["binding"]["task_id"], task["id"]) + status, body = self._req( + "GET", "/projects/%s/automations/%s" % (self.project["id"], + aid), + None, self.reader_key) + self.assertEqual(body["heartbeat_binding"]["mode"], "heartbeat") + status, body = self._req( + "POST", "/projects/%s/automations/%s/heartbeat" + % (self.project["id"], aid), + {"message": "检查部署健康并汇报"}, self.owner_key) + self.assertEqual(status, 201, body) + self.assertEqual(body["delivery"], "queued") + # cron 绑定不唤醒线程。 + status, body = self._req( + "POST", "/projects/%s/automations/%s/bind" + % (self.project["id"], aid), + {"task_id": task["id"], "mode": "cron"}, self.owner_key) + self.assertEqual(status, 200, body) + status, body = self._req( + "POST", "/projects/%s/automations/%s/heartbeat" + % (self.project["id"], aid), + {}, self.owner_key) + self.assertEqual(status, 409, body) + status, body = self._req( + "DELETE", "/projects/%s/automations/%s/bind" + % (self.project["id"], aid), None, self.owner_key) + self.assertEqual(status, 200, body) + status, body = self._req( + "GET", "/projects/%s/automations/%s/bind" + % (self.project["id"], aid), None, self.reader_key) + self.assertIsNone(body["binding"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/thread-lifecycle-align.md b/docs/thread-lifecycle-align.md new file mode 100644 index 0000000..3e2a78d --- /dev/null +++ b/docs/thread-lifecycle-align.md @@ -0,0 +1,214 @@ +# 与 Codex app 的功能对齐：15 项逐项对照 + +本批（feat-thread-lifecycle）之前已有 #103 的 8 项 +（wait/read/message/handoff/worktree/pr-checks/automation/ +sleep，见 docs/thread-wait-ops.md）。本文件逐项说明 15 项 +在控制面的现状：已实现 / 本批补齐 / 待合 PR / 未实现。 +“已实现”指端点真实存在且有单测，不是文档宣称。 + +## 1. Agent 线程管理 + +| 能力 | 落点 | 状态 | +|---|---|---| +| 创建独立线程 | POST /tasks | 已实现 | +| 从当前线程分叉，保留上下文 | POST /tasks/{id}/fork（终态任务；worker 调 host.py fork，分叉 thread 历史 + 拷 worktree 文件） | 已实现 | +| 继续执行暂停/空闲线程 | POST /tasks/{id}/continue（终态复活）；POST /tasks/{id}/thread/message（终态 CAS 复活、进行中写后续） | 已实现 | +| 向其他线程发后续指令 | POST /tasks/{id}/thread/message；跨任务 POST /tasks/{id}/messages（peer，跨项目拒绝） | 已实现 | +| 读取状态、最近结果、历史摘要 | GET /tasks/{id}/thread/read；GET /tasks/{id}/thread?include_turns=1 | 已实现（#103） | +| 设置线程标题 | POST /tasks/{id}/thread/rename {name}（终态任务 + 本机线程） | 已实现 | +| 归档和恢复线程 | POST /tasks/{id}/thread/archive、/thread/unarchive（终态任务 + 本机线程，经 host.py CLI） | 已实现 | +| 查看主机、项目、工作目录 | GET /tasks/{id}/bindings（本批） | 本批补齐 | + +注意：rename/archive/unarchive 走 `_task_thread_mutate`， +只允许终态任务且本机线程；docker 任务、进行中任务会 +409/守卫拦截——这是线程正在变、快照不可信的保护， +不是缺口。归档只归档线程本身，worktree 快照见第 5 项。 + +## 2. 并行任务等待 + +wait_threads = POST /tasks/{id}/wait（#103）：最多 8 个、 +首个重要变化返回（完成/失败/需输入/需审批/需处理）、 +超时回进度、afterCursor 只取新变化。事件等待机制， +不是固定间隔定时器；变化检测只读数据库（任务行、 +审批表、后续表），不轮询 trace。 + +## 3. 线程状态读取 + +read_thread = GET /tasks/{id}/thread/read（#103）：状态映射、 +最近活动、改动文件、命令与退出码、待审批/待后续、 +attention 汇总。本批补了一个小缺口：tool 条目原来只取 +tool 名，丢了 server 与 status，现在拼成 +`server/tool: status` 一行。`test_results` 仍为 null： +trace 里没有独立测试结果记录，不编“测试通过”。 + +## 4. 任务移交和远程主机切换 + +- 发起移交：POST /tasks/{id}/handoff {target_host}（#103） +- 查询进度/是否成功：GET /tasks/{id}/handoff（含 arrived） +- 对方主机上报：POST /tasks/{id}/handoff/{hid}/report +- 移交后继续发指令：POST /tasks/{id}/thread/message（与 + 主机无关，认领方按新主机判断） +- 本批补齐的落点：report 到 completed 时，任务的 + worktree_host 改成目标主机——之前上报只改移交行， + 任务行还指着旧主机，后续 /worktree、/pr/checks 会 + 按旧主机判断，这是真缺口，现已补上。 +- 未实现：服务端不执行 SSH 迁移、不搬文件；迁移本身 + 由目标主机 worker 执行并回报（见 thread-wait-ops.md）。 + +## 5. Git Worktree 生命周期管理 + +| 能力 | 现状 | +|---|---| +| 创建独立 worktree | 已实现：worker 认领后同步建（host.py make_worktree，一任务一分支 host/，非 git 仓库退化为目录拷贝） | +| 从指定 commit/分支/ref 创建 | 部分：任务绑 environment 时按 environment.branch 检出（environment 本身是异步创建过程，creating→ready/failed）；任务直接建时一律从 HEAD 建，不接受 commit/ref 参数 | +| 将已有 worktree 附加到任务 | 未实现：worker 派发路径按任务 id 确定 worktree 目录，附加需要改派发，不在本批编一个只记关联、不被执行的“附加” | +| 查看创建进度 | GET /tasks/{id}/worktree（派生状态；创建是同步的，没有阶段流，不编百分比） | +| 恢复已归档 worktree | 本批：POST /tasks/{id}/worktree/restore（见下） | +| 归档不再需要的 worktree | 本批：POST /tasks/{id}/worktree/archive 拍快照（见下）；真删目录仍是 merge/discard 的职责 | +| 保存未提交修改、未推送 commit、非忽略文件 | 本批快照记：branch、head_sha（即未推送 commit 的位置）、git status、git diff、untracked 路径清单。诚实边界：untracked 文件只记路径不拷内容，diff 只覆盖已跟踪文件；快照是证据，不是可一键还原的备份 | +| 保留 worktree 与线程的关联 | 已实现：快照行含 task_id + worktree_path；bindings 里 worktree 与任务同表返回 | + +Restore 语义（不编造）：目录还在时只把快照标 restored +（目录没动过，谈不上还原）；目录已删时 409 并附快照 +（重建需要仓库路径，服务端没有）。 + +## 6. 任务与 Git 状态绑定 + +本批 GET /tasks/{id}/bindings 一次返回：项目（名、 +repo_url）、环境（名、分支、执行器、状态）、主机 +（worktree_host、local/remote、本机名）、worktree +（状态、路径、分支）、分支、PR（url/number）、执行器 +（任务值或全局缺省）、附件（文件 + 外部引用）、 +worktree 快照清单、血缘（parent_task_id、forked_from、 +coordinator_part）。之前这些字段散在任务行、项目行、 +环境行、worktree 端点里，没有一张总表。 + +## 7. Pull Request 和 CI 诊断 + +read_pull_request_check_diagnostics = +GET /tasks/{id}/pr/checks（#103）：checks JSON（workflow、 +link、description）、失败 run 的 check-runs 日志原文 +（截 4000 字）与行号注释（上限 20 条）。本批只补一个 +字段：响应里加 head_sha（PR 当前 commit，之前内部 +算出来只用于查 check-runs，没返回）。只读：不重跑、 +不批准、不合并。外部 CI 状态即 check-runs 的 +conclusion；构建/测试失败日志即 output.text。 + +## 8. 自动化、心跳和定时任务 + +| 能力 | 现状 | +|---|---| +| 创建/查看/修改/删除长期自动化 | POST/GET/PATCH /projects/{id}/automations、cancel（#103 前已有 + #103 的 update） | +| 按时间周期运行 | 已实现：interval_seconds（60 秒–30 天），到期判断 last_run + interval | +| 定期唤醒已有线程（heartbeat） | 本批：POST .../automations/{aid}/bind {task_id, mode: heartbeat} + POST .../automations/{aid}/heartbeat {message?}（缺省用自动化 prompt）；投递走 thread/message 同款路（终态复活、进行中写后续、配额门同款） | +| cron（到期独立跑新任务） | 绑定 mode=cron 只作标记；heartbeat 口对 cron 绑定回 409（cron 不唤醒线程） | +| 复用已有对话上下文 | heartbeat 唤醒的是同一线程，上下文自然复用；cron 新任务不继承历史（与 fork 的语义一致） | +| 继续执行多日/多周任务 | 机制上可以（interval 上限 30 天 + 绑定线程长期存在），未做跨周的专项验证 | +| 只在变化/完成/失败/需处理时通知，状态不变保持安静 | 未实现：没有“与上次产出比对、无变化静默”的抑制逻辑；每次运行终态仍按现有推送通道通知 | +| 到期触发器 | 未实现（服务端本身不起定时器）：触发需外部调 /heartbeat，或等 PR #101 的 run_due_automation 落地；本批只给了绑定与唤醒口，没宣称“已自动跑” | + +## 9. 线程归档和恢复 + +线程归档/恢复见第 1 项（host.py CLI 透传）。归档时 +保留的东西分两处：对话记录与线程状态在 codex-home +的线程存档里；worktree 快照在本批的 +task_worktree_snapshots 里；PR 附件在任务行的 +pr_url/pr_number 与 bindings 里；未提交文件快照即 +第 5 项的快照（status/diff/untracked 路径）。长期 +项目、阶段性任务、暂时暂停的 Agent 走“归档线程 + +拍 worktree 快照”两步，不是同一个端点。 + +## 10. 任务附件和交付物管理 + +- 文件：已实现——task_attachments（上传包绑定，任务行 + attachments 字段，fork 时带走清单）。 +- PR、worktree：已实现——任务行 pr_url/pr_number、 + worktree 路径，bindings 同表返回。 +- 生成的文档、构建结果、外部服务结果引用：本批—— + POST/GET/DELETE /tasks/{id}/refs（kind: doc/build/ + external/pr/file，只记引用不存内容、不校验链接可达）。 +- 移除：refs 可 DELETE；文件附件的移除走上传包的 + used 状态与任务删除，不在本批另开删除口。 + +## 11. 项目和主机上下文 + +GET /tasks/{id}/bindings 即上下文总表（项目、仓库、 +主机、本地/远程、工作目录、默认执行器、环境、归属）。 +项目默认执行器：任务 executor 为空时取全局缺省 +（_default_executor），没有“项目级默认执行器”字段—— +环境行（executor）是更细的默认，任务绑 environment +时以环境为准。 + +## 12. 插件、MCP 和外部连接器 + +已实现（经 host.py CLI 透传，项目 scope）： + +- GET /projects/{id}/skills（skills/list） +- GET /projects/{id}/apps（app/list） +- GET /projects/{id}/mcp-servers（mcpServerStatus/list） +- POST /projects/{id}/mcp-servers/reload +- POST /projects/{id}/mcp-servers/{name}/oauth-login + +限制与官方一致：连接受账户、工作区、管理员权限、 +插件安装状态限制；oauth-login 只是发起流程，真正 +授权在宿主侧完成；容器内视图与宿主侧视图可能不同 +（skills/MCP 端点 docstring 已注明）。GitHub、Slack +等具体外部系统在本系统是另外的连接器（GitHub App、 +Slack webhook），不是 MCP 一个口。 + +## 13. Memory 和长期上下文 + +分两层，不要混淆： + +- 已实现（本系统自有）：KB（/projects/{id}/kb）与 + 任务 summaries——按任务/项目存的背景，不是跨任务 + 的用户长期记忆。 +- 待合 PR：PR #102（feat-cloud-memory-identity， + user_profiles + identity_memories，用户级仅本人、 + 项目级项目成员可见，上下文各取最新 5 条）—— + 这才是官方 Memory 的对位。本批不重复实现，避免 + 与 #102 合并冲突；自动注入任务/thread 是 #102 + 之后仍待显式决策的下一步（注入点、去重、成本上限）。 +- 可用性限制与官方一致：记忆的范围受账户与发布状态 + 影响，本系统里受项目成员权限限制。 + +## 14. 主动续作和建议性任务 + +未实现为独立服务。现有拼图：自动化（到期唤醒/独立 +跑）+ triage（/projects/{id}/triage，自动化到期与 +待处理汇总）+ #102 的记忆 + bindings 的 PR/任务状态。 +“根据未完成任务、关联 PR、插件内容、Memory、项目 +上下文建议下一步”需要一个建议器（读上述输入、产出 +建议任务），不在本批范围；本批的 heartbeat 与 +bindings 是给建议器准备的输入口与执行口，不是建议器 +本身。不要把 triage 的到期计数说成“已会建议”。 + +## 15. 使用量和连接状态管理 + +已实现（不新开端点，本批只做对照）： + +- 账户使用量：GET /users/me/quota、GET /tasks/{id}/cost +- 限制窗口剩余：quota 的美元配额与无价 token cap +- 连接的主机：GET /workers（心跳与主机名） +- 远程主机是否可用：worktree_host + /workers 心跳龄 + （_workers_heartbeat）+ /health（db_ok、worker_ok） +- App 更新状态：/health 的 commit 字段（部署提交） +- 已安装插件和连接状态：第 12 项的 skills/mcp-servers +- 项目和线程的运行环境：第 6 项的 bindings + +这些用于任务调度与故障判断，不直接跑任务。 + +## 本批新增端点一览 + +- GET /tasks/{id}/bindings +- POST /tasks/{id}/refs、GET /tasks/{id}/refs、 + DELETE /tasks/{id}/refs/{ref_id} +- POST /tasks/{id}/worktree/archive、 + GET /tasks/{id}/worktree/snapshots、 + POST /tasks/{id}/worktree/restore +- GET/POST/DELETE /projects/{id}/automations/{aid}/bind +- POST /projects/{id}/automations/{aid}/heartbeat +- GET /projects/{id}/automations/{aid} 响应加 + heartbeat_binding 字段 +- GET /tasks/{id}/pr/checks 响应加 head_sha 字段 +- 移交 completed 上报后任务 worktree_host 跟随切换