feat: 线程生命周期对齐 15 项（bindings/refs/快照/心跳绑定/移交落点）

对照 Codex app 第二批：GET /tasks/{id}/bindings（项目/
环境/主机/worktree/PR/附件/血缘总表）、/refs 外部引用
增删查、/worktree/archive 快照（branch/head/status/diff/
untracked 路径）与 restore（目录在才解归档、目录删了
409 附快照）、自动化 bind + heartbeat（走 thread/message
同款投递路，cron 绑定不唤醒）、移交 completed 后任务
worktree_host 跟随切换、pr/checks 响应补 head_sha、
thread/read 的 tool 条目补 server/status。

诚实边界见 docs/thread-lifecycle-align.md：已有 worktree
附加、自动化变化静默、建议器、服务端定时触发器未实现；
Memory 对位是待合的 PR #102，本批不重复实现。

三套件：cloud 948 OK（4 跳过）、host 91 OK、deploy 29 OK

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

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",

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）：改名、改

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()