feat(chv): align Muse P8 background work (crons/swarm/presentation/per-task browser)

- crons: pause/resume, exponential-backoff retry (cap 1h), retry claim
  single-winner reusing quota+task build, run_cron_tick cap 5/project
- swarm: per-agent task id + isolated workspace (resolve_inside gate),
  tool register + invocation audit (no execution in swarm layer)
- presentation: worth-presenting rules (needs_input/failed/new
  content only) + notification ledger dedup 1h per (task, reason)
- browser_tasks: one BrowserTransport instance per browser task,
  single-use (reuse after destroy refused), shots dir per instance
- automations: paused status + retry columns (max_retries,
  retry_backoff_seconds, consecutive_failures, next_retry_at)
- gap table docs/chv-muse-p8-gap.md; self-defined values listed there
  (not Meta official): tick cap 5, swarm max 32/default 8, retry
  base 60s cap 3600s, ledger window 3600s, snapshot trunc 200/64KB
- tests: 49 new in cloud/tests/test_chv_muse_p8.py (incl. negatives:
  cron unauthorized trigger, cross-workspace access, no-content
  silence, dedup); full cloud suite 1916 OK (skipped=8)

cloud/automations.py

@@ -54,8 +54,19 @@ MAX_NAME_LEN = 100
 MAX_PROMPT_LEN = 20000
 MAX_REPO_URL_LEN = 1000
 TRIAGE_LIMIT = 50
+# 单项目自动化定义上限（本项目自定，非 Meta 官方值：Meta 未公开
+# cron 的数量上限；超限建定义即 ValueError）。
+MAX_AUTOMATIONS_PER_PROJECT = 100
+# 失败重试默认值（本项目自定，非 Meta 官方值：Meta 未公开失败
+# 重试策略；退避 = 基数 × 2^(n-1)，上限见 crons 模块）。
+DEFAULT_MAX_RETRIES = 3
+DEFAULT_RETRY_BACKOFF_SECONDS = 60
+MAX_RETRIES_LIMIT = 10
+MIN_RETRY_BACKOFF_SECONDS = 10
+MAX_RETRY_BACKOFF_SECONDS = 3600
 
 ACTIVE = "active"
+PAUSED = "paused"
 CANCELLED = "cancelled"
 
 
@@ -73,7 +84,9 @@ AUTOMATION_COLUMNS = (
     "id", "project_id", "name", "prompt", "repo_url",
     "interval_seconds", "status", "last_run_at", "last_claimed_at",
     "created_by", "created_at", "cancelled_at",
-    "first_run_at", "run_once")
+    "first_run_at", "run_once",
+    "max_retries", "retry_backoff_seconds",
+    "consecutive_failures", "next_retry_at")
 RUN_COLUMNS = (
     "id", "automation_id", "project_id", "task_id", "status",
     "detail", "created_at")
@@ -104,8 +117,11 @@ def _existing_columns(conn, table):
 def ensure_automations_tables(conn):
     """幂等建自动化表与产出记录表，并补齐新增列。
 
-    automations：一条定义；status 只取 active / cancelled，
-    取消是软删除(行保留、审计可查，名字可重用)。
+    automations：一条定义；status 只取 active / paused /
+    cancelled，取消是软删除(行保留、审计可查，名字可重用)；
+    paused 是停用(可恢复，不到期、不抢单，见 crons 模块)。
+    max_retries / retry_backoff_seconds / consecutive_failures /
+    next_retry_at 是失败重试列（P8 crons，见该模块）。
     automation_runs：一次执行产出；task_id 为空表示执行器
     抢单后尚未建任务(只应短暂存在，执行器建任务后补记)。
     调用方在写成功后 commit(server 的 handler 与 peer 模式
@@ -125,7 +141,11 @@ def ensure_automations_tables(conn):
         "created_at DOUBLE PRECISION, "
         "cancelled_at DOUBLE PRECISION, "
         "first_run_at DOUBLE PRECISION, "
-        "run_once INTEGER)"
+        "run_once INTEGER, "
+        "max_retries INTEGER, "
+        "retry_backoff_seconds INTEGER, "
+        "consecutive_failures INTEGER, "
+        "next_retry_at DOUBLE PRECISION)"
     )
     conn.execute(
         "CREATE TABLE IF NOT EXISTS automation_runs("
@@ -143,7 +163,11 @@ def ensure_automations_tables(conn):
             ("last_claimed_at", "DOUBLE PRECISION"),
             ("cancelled_at", "DOUBLE PRECISION"),
             ("first_run_at", "DOUBLE PRECISION"),
-            ("run_once", "INTEGER")):
+            ("run_once", "INTEGER"),
+            ("max_retries", "INTEGER"),
+            ("retry_backoff_seconds", "INTEGER"),
+            ("consecutive_failures", "INTEGER"),
+            ("next_retry_at", "DOUBLE PRECISION")):
         if col not in existing:
             try:
                 conn.execute(
@@ -232,11 +256,30 @@ def automation_to_dict(row, now=None):
         "cancelled_at": row_get(row, "cancelled_at"),
         "first_run_at": row_get(row, "first_run_at"),
         "run_once": bool(row_get(row, "run_once")),
+        "max_retries": _retry_int(row, "max_retries",
+                                  DEFAULT_MAX_RETRIES),
+        "retry_backoff_seconds": _retry_int(
+            row, "retry_backoff_seconds",
+            DEFAULT_RETRY_BACKOFF_SECONDS),
+        "consecutive_failures": _retry_int(
+            row, "consecutive_failures", 0),
+        "next_retry_at": row_get(row, "next_retry_at"),
         "next_run_at": nxt,
         "due": is_due(row, now),
     }
 
 
+def _retry_int(row, name, default):
+    """重试列取值：老库行缺列回 None 时用默认值；脏值（非
+    int/负数）同样回默认值，绝不因此炸掉整页。"""
+    value = row_get(row, name)
+    if isinstance(value, bool) or not isinstance(value, int):
+        return default
+    if value < 0:
+        return default
+    return value
+
+
 def _validate_fields(name, prompt, interval_seconds, repo_url,
                      first_run_at=None, run_once=False):
     if not name or not isinstance(name, str) or not name.strip():
@@ -276,18 +319,45 @@ def _validate_fields(name, prompt, interval_seconds, repo_url,
         first_run_at, run_once
 
 
+def _validate_retry(max_retries, retry_backoff_seconds):
+    """重试参数校验（本项目自定范围）：max_retries 0–10，
+    retry_backoff_seconds 10–3600 秒；非法即 ValueError。"""
+    if isinstance(max_retries, bool) \
+            or not isinstance(max_retries, int):
+        raise ValueError("max_retries must be an integer")
+    if not 0 <= max_retries <= MAX_RETRIES_LIMIT:
+        raise ValueError("max_retries must be between 0 and %d"
+                         % MAX_RETRIES_LIMIT)
+    if isinstance(retry_backoff_seconds, bool) \
+            or not isinstance(retry_backoff_seconds, int):
+        raise ValueError("retry_backoff_seconds must be an integer")
+    if not MIN_RETRY_BACKOFF_SECONDS <= retry_backoff_seconds \
+            <= MAX_RETRY_BACKOFF_SECONDS:
+        raise ValueError("retry_backoff_seconds must be between %d"
+                         " and %d" % (MIN_RETRY_BACKOFF_SECONDS,
+                                      MAX_RETRY_BACKOFF_SECONDS))
+    return max_retries, retry_backoff_seconds
+
+
 def create_automation(conn, project_id, name, prompt, created_by,
                       interval_seconds=86400, repo_url=None, now=None,
-                      first_run_at=None, run_once=False):
+                      first_run_at=None, run_once=False,
+                      max_retries=DEFAULT_MAX_RETRIES,
+                      retry_backoff_seconds=DEFAULT_RETRY_BACKOFF_SECONDS):
     """创建自动化定义：同项目内 active 的名字唯一(取消过的
     名字可重用，与环境名唯一规则同款语义)。成功时 commit
     并返回 dict；校验失败抛 ValueError(调用方转 400)。
     first_run_at 为首次运行时刻（epoch 秒，过去则建完即
-    到期）；run_once 为 True 时跑过一次后不再到期。"""
+    到期）；run_once 为 True 时跑过一次后不再到期。
+    max_retries / retry_backoff_seconds 是失败重试策略
+    （P8 crons，本项目自定）；单项目定义上限
+    MAX_AUTOMATIONS_PER_PROJECT，超限即 ValueError。"""
     (name, prompt, interval_seconds, repo_url,
      first_run_at, run_once) = _validate_fields(
         name, prompt, interval_seconds, repo_url,
         first_run_at, run_once)
+    (max_retries, retry_backoff_seconds) = _validate_retry(
+        max_retries, retry_backoff_seconds)
     if now is None:
         now = time.time()
     ensure_automations_tables(conn)
@@ -297,16 +367,30 @@ def create_automation(conn, project_id, name, prompt, created_by,
         (project_id, name, ACTIVE)).fetchone()
     if existing is not None:
         raise ValueError("automation name already exists in this project")
+    count_row = conn.execute(
+        "SELECT COUNT(*) FROM automations WHERE project_id = ?"
+        " AND status IN (?, ?)",
+        (project_id, ACTIVE, PAUSED)).fetchone()
+    try:
+        count = int(count_row[0])
+    except (TypeError, ValueError, IndexError):
+        count = 0
+    if count >= MAX_AUTOMATIONS_PER_PROJECT:
+        raise ValueError("too many automations in this project")
     auto_id = uuid.uuid4().hex
     conn.execute(
         "INSERT INTO automations(id, project_id, name, prompt,"
         " repo_url, interval_seconds, status, last_run_at,"
         " last_claimed_at, created_by, created_at, cancelled_at,"
-        " first_run_at, run_once)"
-        " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL, ?, ?)",
+        " first_run_at, run_once, max_retries,"
+        " retry_backoff_seconds, consecutive_failures,"
+        " next_retry_at)"
+        " VALUES(?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, NULL, ?, ?,"
+        " ?, ?, 0, NULL)",
         (auto_id, project_id, name, prompt, repo_url,
          interval_seconds, ACTIVE, created_by, now,
-         first_run_at, 1 if run_once else 0))
+         first_run_at, 1 if run_once else 0,
+         max_retries, retry_backoff_seconds))
     conn.commit()
     row = conn.execute(
         "SELECT * FROM automations WHERE id = ?", (auto_id,)).fetchone()
@@ -338,7 +422,7 @@ def get_automation(conn, automation_id):
 
 
 def cancel_automation(conn, automation_id, now=None):
-    """取消自动化(软删除)：仅 active 的行能取消。
+    """取消自动化(软删除)：仅 active / paused 的行能取消。
 
     返回 (ok, auto)：ok 为 True 时 auto 是取消后的 dict；
     行不存在时 (False, None)；已取消时 (False, 旧 dict)，
@@ -352,12 +436,12 @@ def cancel_automation(conn, automation_id, now=None):
     if row is None:
         return False, None
     row = _row_dict(row, AUTOMATION_COLUMNS)
-    if row["status"] != ACTIVE:
+    if row["status"] not in (ACTIVE, PAUSED):
         return False, automation_to_dict(row, now)
     cur = conn.execute(
         "UPDATE automations SET status = ?, cancelled_at = ?"
-        " WHERE id = ? AND status = ?",
-        (CANCELLED, now, automation_id, ACTIVE))
+        " WHERE id = ? AND status IN (?, ?)",
+        (CANCELLED, now, automation_id, ACTIVE, PAUSED))
     conn.commit()
     if cur.rowcount != 1:
         return False, automation_to_dict(row, now)
@@ -370,16 +454,16 @@ def cancel_automation(conn, automation_id, now=None):
 def update_automation(conn, automation_id, fields, now=None):
     """更新自动化定义（automation_update）：fields 只认
     name / prompt / interval_seconds / repo_url / first_run_at /
-    run_once 六个键，其它键一律忽略（不报错、不存入，避免
-    调用方拼错字段名后以为已改）；六个键全没给时抛
-    ValueError（空更新不是成功）。校验与创建同款
+    run_once / max_retries / retry_backoff_seconds 八个键，其它键
+    一律忽略（不报错、不存入，避免调用方拼错字段名后以为已
+    改）；八个键全没给时抛 ValueError（空更新不是成功）。校验与创建同款
     （_validate_fields，名字仍按“同项目 active 名字唯一”
     查重，改名撞名时 400）。
 
     返回 (ok, auto)：行不存在时 (False, None)；已取消时
     (False, 旧 dict)——取消是终态，改已取消的定义必须
     重建，不能靠 update 复活（与 cancel 的软删除语义
-    一致）。成功时 commit。注意：更新不重置 last_run_at /
+    一致）。paused 的行可改（改完仍是 paused，不自动恢复）。成功时 commit。注意：更新不重置 last_run_at /
     last_claimed_at——改间隔只影响下一次到期判断（next_run
     由 last_run_at + 新间隔算），不立即触发一轮；想立即
     跑一轮应走执行器的 claim 路径，不是改定义。"""
@@ -392,7 +476,7 @@ def update_automation(conn, automation_id, fields, now=None):
     if row is None:
         return False, None
     row = _row_dict(row, AUTOMATION_COLUMNS)
-    if row["status"] != ACTIVE:
+    if row["status"] not in (ACTIVE, PAUSED):
         return False, automation_to_dict(row, now)
     if not isinstance(fields, dict):
         raise ValueError("fields must be an object")
@@ -407,11 +491,16 @@ def update_automation(conn, automation_id, fields, now=None):
     run_once = fields.get("run_once", current["run_once"])
     if not any(k in fields for k in (
             "name", "prompt", "interval_seconds", "repo_url",
-            "first_run_at", "run_once")):
+            "first_run_at", "run_once",
+            "max_retries", "retry_backoff_seconds")):
         raise ValueError("no updatable field given")
     (name, prompt, interval, repo_url,
      first_run_at, run_once) = _validate_fields(
         name, prompt, interval, repo_url, first_run_at, run_once)
+    (max_retries, retry_backoff_seconds) = _validate_retry(
+        fields.get("max_retries", current["max_retries"]),
+        fields.get("retry_backoff_seconds",
+                   current["retry_backoff_seconds"]))
     if name != current["name"]:
         clash = conn.execute(
             "SELECT id FROM automations WHERE project_id = ?"
@@ -424,9 +513,11 @@ def update_automation(conn, automation_id, fields, now=None):
     conn.execute(
         "UPDATE automations SET name = ?, prompt = ?,"
         " interval_seconds = ?, repo_url = ?, first_run_at = ?,"
-        " run_once = ? WHERE id = ?",
+        " run_once = ?, max_retries = ?,"
+        " retry_backoff_seconds = ? WHERE id = ?",
         (name, prompt, interval, repo_url, first_run_at,
-         1 if run_once else 0, automation_id))
+         1 if run_once else 0, max_retries, retry_backoff_seconds,
+         automation_id))
     conn.commit()
     row = conn.execute(
         "SELECT * FROM automations WHERE id = ?", (automation_id,)
@@ -669,9 +760,13 @@ def _audit_row(conn, project_id, action, detail, now):
                 pass
 
 
-def run_due_automation(conn, automation_id, now=None):
+def run_due_automation(conn, automation_id, now=None, claim_fn=None):
     """云端自动化执行器的唯一入口：抢单 -> 配额门 -> 建任务 -> 记产出。
 
+    claim_fn 可注入抢单函数（默认 claim_due_automation）：
+    crons 模块的重试入口传自己的重试抢单，两条路共用配额门
+    与建任务逻辑，抢单语义各守各的（间隔抢单改 last_run_at，
+    重试抢单只清 next_retry_at、不动间隔时钟）。
     抢不到（不存在、已取消、未到期、已被别的执行器抢走）时
     返回 None，不记产出、不建任务。抢到后先过配额门：创建人
     超配额时不建任务，产出记 skipped_quota（不是失败，不进
@@ -685,7 +780,8 @@ def run_due_automation(conn, automation_id, now=None):
     if now is None:
         now = time.time()
     ensure_automations_tables(conn)
-    if not claim_due_automation(conn, automation_id, now):
+    claim = claim_fn if claim_fn is not None else claim_due_automation
+    if not claim(conn, automation_id, now):
         return None
     auto = conn.execute(
         "SELECT id, project_id, prompt, created_by FROM automations"

cloud/chv/browser_tasks.py

@@ -0,0 +1,211 @@
+"""P8 browser_tasks：每浏览器任务一个实例——建/绑/毁生命周期。
+
+对齐 Muse B4（官方安全文）：每个浏览器任务由专门浏览器子代理
+完成。P2 已把浏览器子代理独立成进程（chv-browser.py，常驻、
+close 只关页不杀进程）；本轨改为每任务一个实例：建实例、绑
+定一个浏览器任务、用完即毁，状态绝不跨任务复用。
+
+transport 接口：真实 transport 走 P2 的 broker 协议
+（open/snapshot/screenshot/close/ping，经 /run/chv/browser.sock）；
+客机侧“每任务起一个全新 chv-browser.py、close 后杀进程”
+是差距表 §3/§6 的跟进项（P2 的 chv-broker.py 里改，本轨不
+复制 P2 的文件）。本层只管生命周期、单实例单任务与状态隔离，
+transport 注入（测试用 fake）。
+
+存储：state_dir/browser-tasks.json（原子写 + 损坏 quarantine，
+orphans 同款口径）；P7 状态存储落地后迁移。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+import uuid
+
+from cloud.chv.atomic_file import atomic_write_text
+from cloud.chv.compat215 import resolve_inside
+
+_LOGGER = logging.getLogger("cloud.cloud_hypervisor")
+
+_REGISTRY_FILENAME = "browser-tasks.json"
+
+
+class BrowserTransport:
+    """浏览器实例 transport 接口：真实实现走 P2 的 broker 协议。
+
+    spawn 建实例、close 毁实例、ping 探活。子类须实现三者；
+    基类方法一律 NotImplementedError（绝不静默假装成功）。"""
+
+    def spawn(self, instance: dict) -> None:
+        """建一个浏览器实例（instance 含 instance_id/task_id）。"""
+        raise NotImplementedError
+
+    def close(self, instance: dict) -> None:
+        """毁一个浏览器实例（进程退出、状态清理）。"""
+        raise NotImplementedError
+
+    def ping(self, instance: dict) -> bool:
+        """实例是否还活着。"""
+        raise NotImplementedError
+
+
+def _registry_path(state_dir: str) -> str:
+    """实例登记文件路径。"""
+    return os.path.join(state_dir, _REGISTRY_FILENAME)
+
+
+def _quarantine(path: str) -> None:
+    """把损坏的登记文件挪到 <path>.corrupt 旁（尽力而为）。"""
+    try:
+        os.replace(path, path + ".corrupt")
+    except OSError:
+        _LOGGER.warning("chv_browser_tasks_quarantine_failed "
+                        "path=%s", path)
+
+
+def _load_registry(state_dir: str) -> dict:
+    """读登记：无文件回空结构；损坏则 quarantine 后回空结构。"""
+    path = _registry_path(state_dir)
+    try:
+        with open(path, "r") as fh:
+            data = json.load(fh)
+    except FileNotFoundError:
+        return {"instances": {}}
+    except (OSError, ValueError):
+        _quarantine(path)
+        return {"instances": {}}
+    if not isinstance(data, dict) \
+            or not isinstance(data.get("instances"), dict):
+        _quarantine(path)
+        return {"instances": {}}
+    return data
+
+
+def _save_registry(state_dir: str, data: dict) -> None:
+    """原子落盘登记；写失败只记日志。"""
+    try:
+        os.makedirs(state_dir, exist_ok=True)
+        atomic_write_text(_registry_path(state_dir), json.dumps(data))
+    except OSError:
+        _LOGGER.warning("chv_browser_tasks_save_failed "
+                        "state_dir=%s", state_dir)
+
+
+def _get_or_raise(data: dict, task_id: str) -> dict:
+    """取某任务的实例；不存在即 ValueError。"""
+    instance = data.get("instances", {}).get(task_id)
+    if not isinstance(instance, dict):
+        raise ValueError("browser instance not found")
+    return instance
+
+
+def create_instance(state_dir: str, task_id: str,
+                    transport: BrowserTransport,
+                    now: float | None = None) -> dict:
+    """为一个浏览器任务建一个实例：单任务单实例（该任务已有
+    任何实例记录——active、destroyed 乃至 failed——即
+    ValueError；一个任务一生只有一个实例，毁后不复建，审计
+    轨迹留在登记里）；先落盘 creating、再调 transport.spawn、
+    成功后置 active；spawn 抛错时记 failed 并原样抛出
+    （不留半个实例）。返回实例 dict。"""
+    if now is None:
+        now = time.time()
+    if not task_id or not isinstance(task_id, str):
+        raise ValueError("task_id required")
+    data = _load_registry(state_dir)
+    existing = data["instances"].get(task_id)
+    if isinstance(existing, dict):
+        raise ValueError("browser instance already exists")
+    instance_id = uuid.uuid4().hex
+    instance = {
+        "instance_id": instance_id,
+        "task_id": task_id,
+        "status": "creating",
+        "shots_dir": "browser-tasks/%s/shots" % instance_id,
+        "created_at": now,
+    }
+    data["instances"][task_id] = instance
+    _save_registry(state_dir, data)
+    try:
+        transport.spawn(instance)
+    except Exception:
+        instance["status"] = "failed"
+        _save_registry(state_dir, data)
+        raise
+    instance["status"] = "active"
+    _save_registry(state_dir, data)
+    return dict(instance)
+
+
+def get_instance(state_dir: str, task_id: str) -> dict | None:
+    """取某任务的实例；不存在回 None。"""
+    data = _load_registry(state_dir)
+    instance = data.get("instances", {}).get(task_id)
+    if not isinstance(instance, dict):
+        return None
+    return dict(instance)
+
+
+def list_instances(state_dir: str) -> list[dict]:
+    """列出全部实例（按创建时间排序）。"""
+    data = _load_registry(state_dir)
+    out = [dict(v) for v in data["instances"].values()
+           if isinstance(v, dict)]
+    out.sort(key=lambda i: (i.get("created_at") or 0,
+                            i.get("task_id") or ""))
+    return out
+
+
+def destroy_instance(state_dir: str, task_id: str,
+                     transport: BrowserTransport,
+                     now: float | None = None) -> dict:
+    """毁一个实例：须是 active（不存在/已毁即 ValueError）；
+    调 transport.close，close 抛错也记下并照常置 destroyed
+    （清理不能因一次 close 失败卡住），返回实例 dict。"""
+    if now is None:
+        now = time.time()
+    data = _load_registry(state_dir)
+    instance = _get_or_raise(data, task_id)
+    if instance.get("status") != "active":
+        raise ValueError("browser instance is not active")
+    close_error = ""
+    try:
+        transport.close(instance)
+    except Exception as exc:
+        close_error = str(exc)[:300]
+    instance["status"] = "destroyed"
+    instance["destroyed_at"] = now
+    instance["close_error"] = close_error
+    _save_registry(state_dir, data)
+    return dict(instance)
+
+
+def ping_instance(state_dir: str, task_id: str,
+                  transport: BrowserTransport) -> bool:
+    """实例是否还活着：非 active 一律 False；active 时问
+    transport。"""
+    data = _load_registry(state_dir)
+    instance = data.get("instances", {}).get(task_id)
+    if not isinstance(instance, dict):
+        return False
+    if instance.get("status") != "active":
+        return False
+    try:
+        return bool(transport.ping(instance))
+    except Exception:
+        return False
+
+
+def instance_shots_dir(state_dir: str, task_id: str) -> str | None:
+    """取某任务实例的截图目录（state_dir 内、resolve_inside
+    校验，越界回 None）；实例不存在回 None。"""
+    data = _load_registry(state_dir)
+    instance = data.get("instances", {}).get(task_id)
+    if not isinstance(instance, dict):
+        return None
+    rel = instance.get("shots_dir")
+    if not isinstance(rel, str) or not rel:
+        return None
+    return resolve_inside(state_dir, rel)

cloud/chv/crons.py

@@ -0,0 +1,281 @@
+"""P8 crons：定时任务的启用/停用、失败重试、tick 驱动。
+
+建在 cloud.automations 之上（不改 host.py run-due：它是 host
+侧唯一的执行器，见 automations 模块 docstring）。本模块只依赖
+标准库 + cloud.automations，不 import 其它 cloud 模块。
+
+语义（数值全部是本项目自定，见 docs/chv-muse-p8-gap.md §4；
+Meta 未公开 cron 的调度粒度、数量上限与失败重试策略）：
+- 停用（paused）：不到期、不抢单、可恢复；恢复时若同项目已有
+  同名 active 定义则恢复失败（调用方据 ok=False 区分）。
+- 失败重试：执行终态为 failed/interrupted 时记一次失败；
+  失败次数 <= max_retries 时按下次重试时刻 next_retry_at =
+  now + min(backoff × 2^(n-1), 3600) 安排重试；超限后不再
+  重试，间隔时钟照常走（下个间隔仍会正常触发）。
+  skipped_quota 不是失败，不进重试。
+- 抢单：间隔抢单与重试抢单各守各的门（claim_due_automation
+  改 last_run_at；claim_retry_automation 只清 next_retry_at、
+  不动间隔时钟），同一执行同一时刻只一个执行器拿到。
+- tick：只列出本轮到期的定义（间隔到期 + 重试到期，单项目
+  单轮上限 5 个），不抢单；执行器对每条分别调
+  run_due_automation / run_retry_automation（抢单在入口内）。
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import time
+from typing import TYPE_CHECKING
+
+from cloud.automations import ACTIVE
+from cloud.automations import AUTOMATION_COLUMNS
+from cloud.automations import DEFAULT_MAX_RETRIES
+from cloud.automations import PAUSED
+from cloud.automations import _audit_row
+from cloud.automations import _row_dict
+from cloud.automations import automation_to_dict
+from cloud.automations import due_automations
+from cloud.automations import ensure_automations_tables
+from cloud.automations import get_automation
+from cloud.automations import row_get
+from cloud.automations import run_due_automation
+
+if TYPE_CHECKING:
+    from cloud.db import PgConnection
+
+# 单项目单轮 tick 最多认领的执行数（本项目自定）。
+MAX_CONCURRENT_CRON_RUNS_PER_PROJECT = 5
+# 重试退避上限（秒，本项目自定）。
+RETRY_BACKOFF_CAP_SECONDS = 3600
+# 计入失败重试的执行终态（skipped_quota 不是失败，不进重试）。
+RETRYABLE_STATUSES = ("failed", "interrupted")
+
+
+def retry_delay_seconds(failures: int, backoff_seconds: int) -> int:
+    """第 failures 次失败后的重试等待秒数：基数 × 2^(n-1)，
+    上限 3600 秒。failures < 1 时按 1 算。"""
+    n = failures if failures >= 1 else 1
+    delay = backoff_seconds * (2 ** (n - 1))
+    return min(delay, RETRY_BACKOFF_CAP_SECONDS)
+
+
+def _fetch(conn: sqlite3.Connection | PgConnection,
+           automation_id: str) -> dict | None:
+    """取单条定义的原始行 dict；不存在回 None。"""
+    ensure_automations_tables(conn)
+    row = conn.execute(
+        "SELECT * FROM automations WHERE id = ?",
+        (automation_id,)).fetchone()
+    if row is None:
+        return None
+    return _row_dict(row, AUTOMATION_COLUMNS)
+
+
+def pause_automation(conn: sqlite3.Connection | PgConnection,
+                     automation_id: str,
+                     now: float | None = None
+                     ) -> tuple[bool, dict | None]:
+    """停用自动化：仅 active 的行能停用。
+
+    返回 (ok, auto)：行不存在时 (False, None)；已停用/已取消
+    时 (False, 旧 dict)，调用方据此区分 404 与 409。成功时
+    commit 并记审计。停用后不到期、不抢单，last_run_at 不动
+    （恢复后按原间隔时钟继续，不立即补跑一轮）。"""
+    if now is None:
+        now = time.time()
+    row = _fetch(conn, automation_id)
+    if row is None:
+        return False, None
+    if row_get(row, "status") != ACTIVE:
+        return False, get_automation(conn, automation_id)
+    cur = conn.execute(
+        "UPDATE automations SET status = ?"
+        " WHERE id = ? AND status = ?",
+        (PAUSED, automation_id, ACTIVE))
+    conn.commit()
+    if cur.rowcount != 1:
+        return False, get_automation(conn, automation_id)
+    _audit_row(conn, row_get(row, "project_id"),
+               "automation_paused", automation_id, now)
+    return True, get_automation(conn, automation_id)
+
+
+def resume_automation(conn: sqlite3.Connection | PgConnection,
+                      automation_id: str,
+                      now: float | None = None
+                      ) -> tuple[bool, dict | None]:
+    """恢复已停用的自动化：仅 paused 的行能恢复。
+
+    同项目已有同名 active 定义时恢复失败（False, 旧 dict）：
+    名字唯一规则只认 active，停用期间建了同名新定义是允许
+    的，恢复撞名时必须先改名或取消那一个。成功时 commit 并
+    记审计；last_run_at 不动（恢复后若间隔已过，下一轮 tick
+    即到期，不立即触发）。"""
+    if now is None:
+        now = time.time()
+    row = _fetch(conn, automation_id)
+    if row is None:
+        return False, None
+    if row_get(row, "status") != PAUSED:
+        return False, get_automation(conn, automation_id)
+    clash = conn.execute(
+        "SELECT id FROM automations WHERE project_id = ?"
+        " AND name = ? AND status = ? AND id != ?",
+        (row_get(row, "project_id"), row_get(row, "name"),
+         ACTIVE, automation_id)).fetchone()
+    if clash is not None:
+        return False, get_automation(conn, automation_id)
+    cur = conn.execute(
+        "UPDATE automations SET status = ?"
+        " WHERE id = ? AND status = ?",
+        (ACTIVE, automation_id, PAUSED))
+    conn.commit()
+    if cur.rowcount != 1:
+        return False, get_automation(conn, automation_id)
+    _audit_row(conn, row_get(row, "project_id"),
+               "automation_resumed", automation_id, now)
+    return True, get_automation(conn, automation_id)
+
+
+def retry_due_automations(conn: sqlite3.Connection | PgConnection,
+                          project_id: str,
+                          now: float | None = None) -> list[dict]:
+    """列出某项目当前重试到期的定义（纯读，不抢单）：
+    active、next_retry_at 已到、失败次数 > 0 且未超 max_retries
+    （老库 NULL 列按默认值 3 算）。按创建时间排序。"""
+    if now is None:
+        now = time.time()
+    ensure_automations_tables(conn)
+    rows = conn.execute(
+        "SELECT * FROM automations WHERE project_id = ?"
+        " AND status = ? AND next_retry_at IS NOT NULL"
+        " AND next_retry_at <= ?"
+        " AND COALESCE(consecutive_failures, 0) > 0"
+        " AND COALESCE(consecutive_failures, 0)"
+        " <= COALESCE(max_retries, ?)"
+        " ORDER BY created_at, id",
+        (project_id, ACTIVE, now, DEFAULT_MAX_RETRIES)).fetchall()
+    return [automation_to_dict(r, now) for r in rows]
+
+
+def claim_retry_automation(conn: sqlite3.Connection | PgConnection,
+                           automation_id: str,
+                           now: float | None = None) -> bool:
+    """重试抢单：重试执行的唯一入口（与间隔抢单分开）。
+
+    条件 UPDATE 只有一行能成功；抢到即清 next_retry_at 并写
+    last_claimed_at，**不动 last_run_at**（间隔时钟不受重试
+    影响）。返回 True 表示抢到，False 表示不存在、非 active、
+    未到重试时刻、失败次数已超限或已被别的执行器抢走。"""
+    if now is None:
+        now = time.time()
+    ensure_automations_tables(conn)
+    cur = conn.execute(
+        "UPDATE automations SET last_claimed_at = ?,"
+        " next_retry_at = NULL"
+        " WHERE id = ? AND status = ?"
+        " AND next_retry_at IS NOT NULL AND next_retry_at <= ?"
+        " AND COALESCE(consecutive_failures, 0) > 0"
+        " AND COALESCE(consecutive_failures, 0)"
+        " <= COALESCE(max_retries, ?)",
+        (now, automation_id, ACTIVE, now, DEFAULT_MAX_RETRIES))
+    conn.commit()
+    return cur.rowcount == 1
+
+
+def record_automation_outcome(conn: sqlite3.Connection | PgConnection,
+                              automation_id: str, status: str,
+                              now: float | None = None
+                              ) -> dict | None:
+    """记一次自动化执行的终态，更新失败计数与下次重试时刻。
+
+    - failed/interrupted：失败次数 +1；未超 max_retries 时
+      next_retry_at = now + 退避，超限后清 next_retry_at
+      （不再重试，间隔时钟照常走）；记审计。
+    - completed：失败次数清零、next_retry_at 清空。
+    - 其它终态（created/skipped_quota 等）：不改计数。
+    定义不存在时回 None。成功时 commit 并返回最新 dict。"""
+    if now is None:
+        now = time.time()
+    auto = get_automation(conn, automation_id)
+    if auto is None:
+        return None
+    if status in RETRYABLE_STATUSES:
+        return _record_failure(conn, auto, now)
+    if status == "completed":
+        return _record_success(conn, auto, now)
+    return auto
+
+
+def _record_failure(conn: sqlite3.Connection | PgConnection,
+                    auto: dict, now: float) -> dict | None:
+    """失败分支：失败次数 +1，未超限则安排下次重试时刻。"""
+    failures = int(auto["consecutive_failures"]) + 1
+    max_retries = int(auto["max_retries"])
+    if failures <= max_retries:
+        delay = retry_delay_seconds(
+            failures, int(auto["retry_backoff_seconds"]))
+        next_retry = now + delay
+    else:
+        next_retry = None
+    conn.execute(
+        "UPDATE automations SET consecutive_failures = ?,"
+        " next_retry_at = ? WHERE id = ?",
+        (failures, next_retry, auto["id"]))
+    conn.commit()
+    _audit_row(conn, auto["project_id"], "automation_run_failed",
+               "%s: failures=%d next_retry_at=%s"
+               % (auto["id"], failures, next_retry), now)
+    return get_automation(conn, auto["id"])
+
+
+def _record_success(conn: sqlite3.Connection | PgConnection,
+                    auto: dict, now: float) -> dict | None:
+    """成功分支：失败次数清零、下次重试时刻清空。"""
+    conn.execute(
+        "UPDATE automations SET consecutive_failures = 0,"
+        " next_retry_at = NULL WHERE id = ?",
+        (auto["id"],))
+    conn.commit()
+    return get_automation(conn, auto["id"])
+
+
+def run_retry_automation(conn: sqlite3.Connection | PgConnection,
+                         automation_id: str,
+                         now: float | None = None) -> dict | None:
+    """重试执行的唯一入口：重试抢单 -> 配额门 -> 建任务 -> 记产出。
+
+    与 run_due_automation 共用配额门与建任务逻辑，只把抢单换
+    成重试抢单；抢不到时回 None。调用方（定时执行器）只调本
+    函数与 run_due_automation，绝不自己拼 claim + INSERT。"""
+    if now is None:
+        now = time.time()
+    return run_due_automation(conn, automation_id, now,
+                              claim_fn=claim_retry_automation)
+
+
+def run_cron_tick(conn: sqlite3.Connection | PgConnection,
+                  project_id: str,
+                  now: float | None = None) -> dict:
+    """一轮 tick 列出本项目当前到期的定义（纯读，不抢单）：
+    间隔到期在前、重试到期在后，单轮总数上限
+    MAX_CONCURRENT_CRON_RUNS_PER_PROJECT（本项目自定）。
+
+    返回 {"project_id", "due", "retry_due", "capped"}；执行器
+    对 due 的每条调 run_due_automation、对 retry_due 的每条调
+    run_retry_automation（抢单在入口内，两个执行器同时 tick
+    也只一个能抢到）。"""
+    if now is None:
+        now = time.time()
+    due = due_automations(conn, project_id, now)
+    retry = retry_due_automations(conn, project_id, now)
+    cap = MAX_CONCURRENT_CRON_RUNS_PER_PROJECT
+    kept_due = due[:cap]
+    kept_retry = retry[:max(0, cap - len(kept_due))]
+    capped = len(due) + len(retry) > cap
+    return {
+        "project_id": project_id,
+        "due": kept_due,
+        "retry_due": kept_retry,
+        "capped": capped,
+    }

cloud/chv/presentation.py

@@ -0,0 +1,172 @@
+"""P8 presentation：后台任务完成后“是否值得呈现”的评估，与
+通知去重节流账本。
+
+对齐 Muse B2（官方设计文）：完成后台工作后评估是否值得呈现，
+只在真正有新内容或需要输入时通知用户。
+
+判定是确定性规则（不是模型判断）：
+- needs_input：需要用户输入 -> 通知。
+- failed/interrupted：失败、需要决策 -> 通知。
+- completed 且有实质新内容（新产物、改动文件、或显式 notify
+  标志）-> 通知；否则常规完成只记日志。
+- cancelled / 未知终态 -> 只记日志（未知终态 fail-closed，
+  不通知）。
+
+账本：state_dir/notification-ledger.json（原子写 + 损坏
+quarantine，orphans 同款口径）；同一任务同一原因在节流窗口
+（3600 秒，本项目自定）内只通知一次。P7 状态存储落地后迁移，
+本模块 docstring 与差距表已标注；账本丢了只影响去重（可能
+重复通知一次），不影响任务执行本身。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+
+from cloud.chv.atomic_file import atomic_write_text
+
+_LOGGER = logging.getLogger("cloud.cloud_hypervisor")
+
+# 同一任务同一原因的通知节流窗口（秒，本项目自定）。
+NOTIFY_THROTTLE_SECONDS = 3600
+_LEDGER_FILENAME = "notification-ledger.json"
+_MAX_LEDGER_ENTRIES = 500
+
+
+def evaluate_presentation(result: dict) -> dict:
+    """评估一次后台任务终态是否值得呈现。
+
+    result 键：status（completed/failed/interrupted/cancelled/
+    needs_input）、new_artifacts（list）、changed_files（list）、
+    notify（bool，任务显式要求通知）。返回 {"present", "reason"}；
+    present 为 False 时调用方只记日志、不通知。"""
+    if not isinstance(result, dict):
+        return {"present": False, "reason": "unknown_status"}
+    status = result.get("status")
+    if status == "needs_input":
+        return {"present": True, "reason": "needs_input"}
+    if status in ("failed", "interrupted"):
+        return {"present": True, "reason": "failed_needs_decision"}
+    if status == "completed":
+        return _evaluate_completed(result)
+    if status == "cancelled":
+        return {"present": False, "reason": "cancelled"}
+    return {"present": False, "reason": "unknown_status"}
+
+
+def _evaluate_completed(result: dict) -> dict:
+    """完成态分支：有实质新内容才通知。"""
+    artifacts = result.get("new_artifacts")
+    changed = result.get("changed_files")
+    has_new = bool(result.get("notify"))
+    if isinstance(artifacts, list) and artifacts:
+        has_new = True
+    if isinstance(changed, list) and changed:
+        has_new = True
+    if has_new:
+        return {"present": True, "reason": "substantive_new_content"}
+    return {"present": False, "reason": "routine_completed"}
+
+
+def _ledger_path(state_dir: str) -> str:
+    """通知账本文件路径。"""
+    return os.path.join(state_dir, _LEDGER_FILENAME)
+
+
+def _quarantine(path: str) -> None:
+    """把损坏的账本挪到 <path>.corrupt 旁（尽力而为）。"""
+    try:
+        os.replace(path, path + ".corrupt")
+    except OSError:
+        _LOGGER.warning("chv_ledger_quarantine_failed path=%s", path)
+
+
+def _load_ledger(state_dir: str) -> dict:
+    """读账本：无文件回空结构；损坏则 quarantine 后回空结构
+    （去重信息丢了只会导致一次重复通知，不影响任务本身）。"""
+    path = _ledger_path(state_dir)
+    try:
+        with open(path, "r") as fh:
+            data = json.load(fh)
+    except FileNotFoundError:
+        return {"notified": {}}
+    except (OSError, ValueError):
+        _quarantine(path)
+        return {"notified": {}}
+    if not isinstance(data, dict) \
+            or not isinstance(data.get("notified"), dict):
+        _quarantine(path)
+        return {"notified": {}}
+    return data
+
+
+def _save_ledger(state_dir: str, data: dict) -> None:
+    """原子落盘账本（截断到上限）；写失败只记日志。"""
+    notified = data.get("notified", {})
+    if len(notified) > _MAX_LEDGER_ENTRIES:
+        items = sorted(notified.items(), key=lambda kv: kv[1])
+        data["notified"] = dict(items[-_MAX_LEDGER_ENTRIES:])
+    try:
+        os.makedirs(state_dir, exist_ok=True)
+        atomic_write_text(_ledger_path(state_dir), json.dumps(data))
+    except OSError:
+        _LOGGER.warning("chv_ledger_save_failed state_dir=%s", state_dir)
+
+
+class NotificationLedger:
+    """通知去重节流账本（每任务每原因节流）。
+
+    should_notify(task_id, reason) 在窗口内未通知过时回 True
+    并记下这次通知；窗口内已通知过回 False。账本先走原子
+    JSON 文件，待迁移到 P7 状态存储（见模块 docstring）。
+    """
+
+    def __init__(self, state_dir: str) -> None:
+        """建账本：state_dir 为空即 ValueError。"""
+        if not state_dir or not isinstance(state_dir, str):
+            raise ValueError("state_dir required")
+        self.state_dir = state_dir
+
+    def should_notify(self, task_id: str, reason: str,
+                      now: float | None = None) -> bool:
+        """该任务该原因现在能不能通知：窗口内没通知过就记下
+        并回 True，否则回 False。task_id/reason 为空即
+        ValueError。"""
+        if now is None:
+            now = time.time()
+        if not task_id or not isinstance(task_id, str):
+            raise ValueError("task_id required")
+        if not reason or not isinstance(reason, str):
+            raise ValueError("reason required")
+        key = "%s:%s" % (task_id, reason)
+        data = _load_ledger(self.state_dir)
+        notified = data.get("notified", {})
+        last = notified.get(key)
+        if isinstance(last, (int, float)) \
+                and now - last < NOTIFY_THROTTLE_SECONDS:
+            return False
+        notified[key] = now
+        _save_ledger(self.state_dir, data)
+        return True
+
+
+def present_or_log(result: dict, ledger: NotificationLedger,
+                   task_id: str,
+                   now: float | None = None) -> dict:
+    """评估 + 去重的一站式入口（worker/push 终态派发处调它）。
+
+    返回 {"present", "reason", "notified"}：present 是评估结论，
+    notified 是本次是否真的发出通知（评估通过但窗口内已通知
+    过时 notified 为 False、只记日志）。"""
+    if now is None:
+        now = time.time()
+    decision = evaluate_presentation(result)
+    if not decision["present"]:
+        return {"present": False, "reason": decision["reason"],
+                "notified": False}
+    notified = ledger.should_notify(task_id, decision["reason"], now)
+    return {"present": True, "reason": decision["reason"],
+            "notified": notified}

cloud/chv/swarm.py

@@ -0,0 +1,383 @@
+"""P8 swarm：子代理群——建群、派代理、隔离、工具注册与审计。
+
+对齐 Muse B3（官方安全文）：启动成群的子代理、构建自己的工具；
+VM 需要足够资源处理并发子代理与 crons。资源上限是本项目自定
+（见 docs/chv-muse-p8-gap.md §4），不是 Meta 官方值。
+
+隔离模型：每代理一个独立任务 id + 独立工作区（父任务工作区下
+按代理 id 分目录，`resolve_inside` 门守着）；代理之间互不可见
+对方工作区（`check_agent_path` 否定测试钉住）。真正执行在代
+理自己的 VM/单元里（P1 常驻池 + P2 单元是底座），本层只做登
+记、派发与审计，不执行工具代码。
+
+存储：state_dir/swarms.json（原子写 + 损坏 quarantine，orphans
+同款口径）；P7 状态存储落地后迁移，本模块 docstring 与差距表
+已标注。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+import uuid
+from collections.abc import Callable
+
+from cloud.chv.atomic_file import atomic_write_text
+from cloud.chv.compat215 import resolve_inside
+
+_LOGGER = logging.getLogger("cloud.cloud_hypervisor")
+
+MAX_AGENTS_PER_SWARM = 8
+MAX_AGENTS_LIMIT = 32
+MAX_SWARMS_PER_PROJECT = 10
+MAX_TOOL_NAME_LEN = 100
+MAX_AGENT_PROMPT_LEN = 20000
+MAX_AUDIT_ENTRIES = 200
+MAX_INVOCATION_ENTRIES = 200
+AGENT_STATUSES = ("pending", "running", "done", "failed")
+_REGISTRY_FILENAME = "swarms.json"
+
+
+def _registry_path(state_dir: str) -> str:
+    """群登记文件路径。"""
+    return os.path.join(state_dir, _REGISTRY_FILENAME)
+
+
+def _quarantine(path: str) -> None:
+    """把损坏的登记文件挪到 <path>.corrupt 旁（尽力而为）。"""
+    try:
+        os.replace(path, path + ".corrupt")
+    except OSError:
+        _LOGGER.warning("chv_swarm_registry_quarantine_failed "
+                        "path=%s", path)
+
+
+def _empty_registry() -> dict:
+    """空登记结构。"""
+    return {"swarms": {}, "tools": {}, "invocations": [],
+            "audit": []}
+
+
+def _load_registry(state_dir: str) -> dict:
+    """读登记：无文件回空结构；损坏则 quarantine 后回空结构
+    （损坏丢状态是 orphans 同款 fail-safe 口径，见差距表）。"""
+    path = _registry_path(state_dir)
+    try:
+        with open(path, "r") as fh:
+            data = json.load(fh)
+    except FileNotFoundError:
+        return _empty_registry()
+    except (OSError, ValueError):
+        _quarantine(path)
+        return _empty_registry()
+    if not isinstance(data, dict):
+        _quarantine(path)
+        return _empty_registry()
+    data.setdefault("swarms", {})
+    data.setdefault("tools", {})
+    data.setdefault("invocations", [])
+    data.setdefault("audit", [])
+    return data
+
+
+def _save_registry(state_dir: str, data: dict) -> None:
+    """原子落盘登记（审计与调用记录各截断到上限）；写失败
+    只记日志，不炸调用方。"""
+    data["audit"] = data.get("audit", [])[-MAX_AUDIT_ENTRIES:]
+    data["invocations"] = data.get("invocations", [])[
+        -MAX_INVOCATION_ENTRIES:]
+    try:
+        os.makedirs(state_dir, exist_ok=True)
+        atomic_write_text(_registry_path(state_dir),
+                          json.dumps(data))
+    except OSError:
+        _LOGGER.warning("chv_swarm_registry_save_failed "
+                        "state_dir=%s", state_dir)
+
+
+def _audit(data: dict, action: str, detail: str, now: float) -> None:
+    """记一条审计（调用方负责落盘）。"""
+    data.setdefault("audit", []).append(
+        {"action": action, "detail": detail, "created_at": now})
+
+
+def _get_swarm_or_raise(data: dict, swarm_id: str) -> dict:
+    """取群，不存在即 ValueError（调用方转 404）。"""
+    swarm = data.get("swarms", {}).get(swarm_id)
+    if not isinstance(swarm, dict):
+        raise ValueError("swarm not found")
+    return swarm
+
+
+def create_swarm(state_dir: str, project_id: str,
+                 parent_task_id: str, max_agents: int = 8,
+                 now: float | None = None) -> dict:
+    """建群：每项目群上限 10（本项目自定），每群代理上限
+    1–32（默认 8，本项目自定）。返回群 dict。"""
+    if now is None:
+        now = time.time()
+    if isinstance(max_agents, bool) \
+            or not isinstance(max_agents, int):
+        raise ValueError("max_agents must be an integer")
+    if not 1 <= max_agents <= MAX_AGENTS_LIMIT:
+        raise ValueError("max_agents must be between 1 and %d"
+                         % MAX_AGENTS_LIMIT)
+    if not project_id or not parent_task_id:
+        raise ValueError("project_id and parent_task_id required")
+    data = _load_registry(state_dir)
+    count = sum(1 for s in data["swarms"].values()
+                if isinstance(s, dict)
+                and s.get("project_id") == project_id)
+    if count >= MAX_SWARMS_PER_PROJECT:
+        raise ValueError("too many swarms in this project")
+    swarm_id = uuid.uuid4().hex
+    swarm = {
+        "id": swarm_id,
+        "project_id": project_id,
+        "parent_task_id": parent_task_id,
+        "max_agents": max_agents,
+        "status": "active",
+        "agents": {},
+        "created_at": now,
+    }
+    data["swarms"][swarm_id] = swarm
+    _audit(data, "swarm_created",
+           "%s parent=%s" % (swarm_id, parent_task_id), now)
+    _save_registry(state_dir, data)
+    return dict(swarm)
+
+
+def get_swarm(state_dir: str, swarm_id: str) -> dict | None:
+    """取群；不存在回 None。"""
+    data = _load_registry(state_dir)
+    swarm = data.get("swarms", {}).get(swarm_id)
+    if not isinstance(swarm, dict):
+        return None
+    return dict(swarm)
+
+
+def spawn_agent(state_dir: str, swarm_id: str, prompt: str,
+                workspace_root: str,
+                now: float | None = None) -> dict:
+    """派一个子代理：群须 active、人数未满上限；代理得独立
+    任务 id + 独立工作区（父工作区下按代理 id 分目录，
+    resolve_inside 门守着，越界即 ValueError）。返回代理 dict。"""
+    if now is None:
+        now = time.time()
+    if not prompt or not isinstance(prompt, str) \
+            or not prompt.strip():
+        raise ValueError("prompt required")
+    if len(prompt) > MAX_AGENT_PROMPT_LEN:
+        raise ValueError("prompt too long")
+    if not workspace_root or not isinstance(workspace_root, str):
+        raise ValueError("workspace_root required")
+    data = _load_registry(state_dir)
+    swarm = _get_swarm_or_raise(data, swarm_id)
+    if swarm.get("status") != "active":
+        raise ValueError("swarm is not active")
+    agents = swarm.setdefault("agents", {})
+    if len(agents) >= int(swarm["max_agents"]):
+        raise ValueError("swarm agent limit reached")
+    agent_id = uuid.uuid4().hex
+    workspace = resolve_inside(workspace_root, agent_id)
+    if workspace is None:
+        raise ValueError("workspace_root is not usable")
+    agent = {
+        "id": agent_id,
+        "task_id": uuid.uuid4().hex,
+        "prompt": prompt.strip(),
+        "workspace": workspace,
+        "status": "pending",
+        "created_at": now,
+    }
+    agents[agent_id] = agent
+    _audit(data, "swarm_agent_spawned",
+           "%s agent=%s" % (swarm_id, agent_id), now)
+    _save_registry(state_dir, data)
+    return dict(agent)
+
+
+def list_agents(state_dir: str, swarm_id: str) -> list[dict]:
+    """列出某群的全部代理（按创建时间排序）。群不存在即
+    ValueError。"""
+    data = _load_registry(state_dir)
+    swarm = _get_swarm_or_raise(data, swarm_id)
+    agents = [dict(a) for a in
+              swarm.get("agents", {}).values()
+              if isinstance(a, dict)]
+    agents.sort(key=lambda a: (a.get("created_at") or 0,
+                               a.get("id") or ""))
+    return agents
+
+
+def agent_workspace(state_dir: str, swarm_id: str,
+                    agent_id: str) -> str | None:
+    """取某代理的工作区路径；群或代理不存在回 None。"""
+    data = _load_registry(state_dir)
+    swarm = data.get("swarms", {}).get(swarm_id)
+    if not isinstance(swarm, dict):
+        return None
+    agent = swarm.get("agents", {}).get(agent_id)
+    if not isinstance(agent, dict):
+        return None
+    workspace = agent.get("workspace")
+    return workspace if isinstance(workspace, str) else None
+
+
+def check_agent_path(state_dir: str, swarm_id: str, agent_id: str,
+                     path: str) -> bool:
+    """隔离门：path 是否在该代理自己的工作区内（realpath +
+    commonpath 校验，符号链接逃逸、.. 逃逸、跨盘符一律 False）。
+    群或代理不存在时同样 False（fail-closed）。"""
+    workspace = agent_workspace(state_dir, swarm_id, agent_id)
+    if not workspace or not isinstance(path, str) or not path:
+        return False
+    try:
+        rel = os.path.relpath(path, workspace)
+    except ValueError:
+        return False
+    return resolve_inside(workspace, rel) is not None
+
+
+def _get_agent_or_raise(data: dict, swarm_id: str,
+                        agent_id: str) -> tuple[dict, dict]:
+    """取（群，代理），任一不存在即 ValueError。"""
+    swarm = _get_swarm_or_raise(data, swarm_id)
+    agent = swarm.get("agents", {}).get(agent_id)
+    if not isinstance(agent, dict):
+        raise ValueError("agent not found")
+    return swarm, agent
+
+
+def _refresh_swarm_status(swarm: dict) -> None:
+    """群终态推导：代理全终态后，全部 done 即 done、有 failed
+    即 failed；否则保持 active。"""
+    agents = [a for a in swarm.get("agents", {}).values()
+              if isinstance(a, dict)]
+    if not agents:
+        return
+    if any(a.get("status") not in ("done", "failed")
+           for a in agents):
+        return
+    if any(a.get("status") == "failed" for a in agents):
+        swarm["status"] = "failed"
+    else:
+        swarm["status"] = "done"
+
+
+def set_agent_status(state_dir: str, swarm_id: str, agent_id: str,
+                     status: str,
+                     now: float | None = None) -> dict:
+    """改代理状态（pending/running/done/failed）；非法状态或
+    代理不存在即 ValueError。改完推导群终态，返回代理 dict。"""
+    if now is None:
+        now = time.time()
+    if status not in AGENT_STATUSES:
+        raise ValueError("invalid agent status")
+    data = _load_registry(state_dir)
+    swarm, agent = _get_agent_or_raise(data, swarm_id, agent_id)
+    agent["status"] = status
+    _refresh_swarm_status(swarm)
+    _audit(data, "swarm_agent_status",
+           "%s agent=%s status=%s" % (swarm_id, agent_id, status),
+           now)
+    _save_registry(state_dir, data)
+    return dict(agent)
+
+
+def register_tool(state_dir: str, swarm_id: str, agent_id: str,
+                  name: str, description: str,
+                  now: float | None = None) -> dict:
+    """子代理构建自己的工具：登记进群内工具表（名字在该代理
+    内唯一，1–100 字符），带审计。登记只登记，不执行。"""
+    if now is None:
+        now = time.time()
+    if not name or not isinstance(name, str) or not name.strip():
+        raise ValueError("tool name required")
+    name = name.strip()
+    if len(name) > MAX_TOOL_NAME_LEN:
+        raise ValueError("tool name too long")
+    if not isinstance(description, str):
+        raise ValueError("description must be a string")
+    data = _load_registry(state_dir)
+    _get_agent_or_raise(data, swarm_id, agent_id)
+    tools = data.setdefault("tools", {})
+    by_agent = tools.setdefault(swarm_id, {}).setdefault(agent_id, {})
+    if name in by_agent:
+        raise ValueError("tool already registered for this agent")
+    tool = {"name": name, "description": description,
+            "agent_id": agent_id, "swarm_id": swarm_id,
+            "created_at": now}
+    by_agent[name] = tool
+    _audit(data, "swarm_tool_registered",
+           "%s agent=%s tool=%s" % (swarm_id, agent_id, name), now)
+    _save_registry(state_dir, data)
+    return dict(tool)
+
+
+def record_tool_invocation(state_dir: str, swarm_id: str,
+                           agent_id: str, name: str,
+                           args_summary: str,
+                           now: float | None = None) -> dict:
+    """记一次工具调用意图（登记 + 审计）：工具须是该代理自己
+    注册的，别代理的工具名在这里叫不动。
+
+    注意：本函数**不执行工具代码**——执行在子代理自己的 VM/
+    单元里（执行器派发代理任务时发生）。这里只留下“谁、在
+    哪群、调了哪个工具、参数摘要”的审计记录。"""
+    if now is None:
+        now = time.time()
+    if not isinstance(args_summary, str):
+        raise ValueError("args_summary must be a string")
+    data = _load_registry(state_dir)
+    _get_agent_or_raise(data, swarm_id, agent_id)
+    by_agent = data.get("tools", {}).get(swarm_id, {}).get(agent_id,
+                                                          {})
+    if name not in by_agent:
+        raise ValueError("tool not registered for this agent")
+    record = {
+        "id": uuid.uuid4().hex,
+        "swarm_id": swarm_id,
+        "agent_id": agent_id,
+        "name": name,
+        "args_summary": args_summary[:500],
+        "created_at": now,
+    }
+    data.setdefault("invocations", []).append(record)
+    _audit(data, "swarm_tool_invoked",
+           "%s agent=%s tool=%s" % (swarm_id, agent_id, name), now)
+    _save_registry(state_dir, data)
+    return dict(record)
+
+
+def dispatch_agent(state_dir: str, swarm_id: str, agent_id: str,
+                   runner: Callable[[dict], dict],
+                   now: float | None = None) -> dict:
+    """派发代理去执行：代理须是 pending；runner 是调用方注入
+    的执行函数（生产上接执行器 dispatch，测试里传 fake），
+    收到任务 dict {"id", "prompt", "swarm_id", "agent_id",
+    "workspace"}，返回其执行结果。
+
+    派发前把代理置为 running；runner 抛错时代理状态不动
+    （调用方按失败处理，不在这里把异常吞成 done）。"""
+    if now is None:
+        now = time.time()
+    data = _load_registry(state_dir)
+    swarm, agent = _get_agent_or_raise(data, swarm_id, agent_id)
+    if agent.get("status") != "pending":
+        raise ValueError("agent is not pending")
+    task = {
+        "id": agent["task_id"],
+        "prompt": agent["prompt"],
+        "swarm_id": swarm_id,
+        "agent_id": agent_id,
+        "workspace": agent["workspace"],
+    }
+    agent["status"] = "running"
+    _audit(data, "swarm_agent_dispatched",
+           "%s agent=%s task=%s"
+           % (swarm_id, agent_id, agent["task_id"]), now)
+    _save_registry(state_dir, data)
+    return runner(task)

cloud/tests/test_chv_muse_p8.py

@@ -0,0 +1,699 @@
+"""对齐 Muse P8：后台工作（crons / 子代理群 / 回访通知 /
+每任务浏览器子代理）单测。
+
+不跑真实模型、不联网、不碰真机：临时 SQLite 文件 + 临时
+state_dir + 模块函数直调 + fake transport/runner。
+覆盖：cron 停用/恢复/重试退避/重试到期与抢单/tick 上限/
+定义上限；群建群/派代理/工作区隔离/工具注册与调用/派发；
+值得呈现评估/通知账本去重节流；浏览器实例单任务单实例/
+用完即毁。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.automations import cancel_automation
+    from cloud.automations import create_automation
+    from cloud.automations import ensure_automations_tables
+    from cloud.automations import get_automation
+    from cloud.automations import update_automation
+    from cloud.chv.browser_tasks import BrowserTransport
+    from cloud.chv.browser_tasks import create_instance
+    from cloud.chv.browser_tasks import destroy_instance
+    from cloud.chv.browser_tasks import get_instance
+    from cloud.chv.browser_tasks import instance_shots_dir
+    from cloud.chv.browser_tasks import ping_instance
+    from cloud.chv.crons import claim_retry_automation
+    from cloud.chv.crons import pause_automation
+    from cloud.chv.crons import record_automation_outcome
+    from cloud.chv.crons import resume_automation
+    from cloud.chv.crons import retry_delay_seconds
+    from cloud.chv.crons import retry_due_automations
+    from cloud.chv.crons import run_cron_tick
+    from cloud.chv.crons import run_retry_automation
+    from cloud.chv.presentation import NotificationLedger
+    from cloud.chv.presentation import evaluate_presentation
+    from cloud.chv.presentation import present_or_log
+    from cloud.chv.swarm import check_agent_path
+    from cloud.chv.swarm import create_swarm
+    from cloud.chv.swarm import dispatch_agent
+    from cloud.chv.swarm import get_swarm
+    from cloud.chv.swarm import list_agents
+    from cloud.chv.swarm import record_tool_invocation
+    from cloud.chv.swarm import register_tool
+    from cloud.chv.swarm import set_agent_status
+    from cloud.chv.swarm import spawn_agent
+    from cloud.server import init_db
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.automations import cancel_automation
+    from cloud.automations import create_automation
+    from cloud.automations import ensure_automations_tables
+    from cloud.automations import get_automation
+    from cloud.automations import update_automation
+    from cloud.chv.browser_tasks import BrowserTransport
+    from cloud.chv.browser_tasks import create_instance
+    from cloud.chv.browser_tasks import destroy_instance
+    from cloud.chv.browser_tasks import get_instance
+    from cloud.chv.browser_tasks import instance_shots_dir
+    from cloud.chv.browser_tasks import ping_instance
+    from cloud.chv.crons import claim_retry_automation
+    from cloud.chv.crons import pause_automation
+    from cloud.chv.crons import record_automation_outcome
+    from cloud.chv.crons import resume_automation
+    from cloud.chv.crons import retry_delay_seconds
+    from cloud.chv.crons import retry_due_automations
+    from cloud.chv.crons import run_cron_tick
+    from cloud.chv.crons import run_retry_automation
+    from cloud.chv.presentation import NotificationLedger
+    from cloud.chv.presentation import evaluate_presentation
+    from cloud.chv.presentation import present_or_log
+    from cloud.chv.swarm import check_agent_path
+    from cloud.chv.swarm import create_swarm
+    from cloud.chv.swarm import dispatch_agent
+    from cloud.chv.swarm import get_swarm
+    from cloud.chv.swarm import list_agents
+    from cloud.chv.swarm import record_tool_invocation
+    from cloud.chv.swarm import register_tool
+    from cloud.chv.swarm import set_agent_status
+    from cloud.chv.swarm import spawn_agent
+    from cloud.server import init_db
+
+T0 = 1800000000.0
+
+
+class CronTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.conn = sqlite3.connect(self.db_path)
+        self.conn.row_factory = sqlite3.Row
+        ensure_automations_tables(self.conn)
+
+    def tearDown(self):
+        self.conn.close()
+        self.tmp.cleanup()
+
+    def _auto(self, name="a", **kwargs):
+        kwargs.setdefault("interval_seconds", 3600)
+        return create_automation(self.conn, "p1", name, "do thing",
+                                 "tester", now=T0, **kwargs)
+
+    def test_pause_resume_roundtrip(self):
+        auto = self._auto()
+        ok, paused = pause_automation(self.conn, auto["id"], now=T0)
+        self.assertTrue(ok)
+        self.assertEqual(paused["status"], "paused")
+        self.assertFalse(paused["due"])
+        ok, resumed = resume_automation(self.conn, auto["id"], now=T0)
+        self.assertTrue(ok)
+        self.assertEqual(resumed["status"], "active")
+
+    def test_pause_twice_and_resume_active_fail(self):
+        auto = self._auto()
+        self.assertTrue(pause_automation(self.conn, auto["id"],
+                                         now=T0)[0])
+        ok, _ = pause_automation(self.conn, auto["id"], now=T0)
+        self.assertFalse(ok)
+        ok, _ = resume_automation(self.conn, auto["id"], now=T0)
+        self.assertTrue(ok)
+        ok, _ = resume_automation(self.conn, auto["id"], now=T0)
+        self.assertFalse(ok)
+
+    def test_pause_missing_returns_false_none(self):
+        ok, auto = pause_automation(self.conn, "nope", now=T0)
+        self.assertFalse(ok)
+        self.assertIsNone(auto)
+
+    def test_cancel_paused_works(self):
+        auto = self._auto()
+        pause_automation(self.conn, auto["id"], now=T0)
+        ok, cancelled = cancel_automation(self.conn, auto["id"],
+                                          now=T0)
+        self.assertTrue(ok)
+        self.assertEqual(cancelled["status"], "cancelled")
+
+    def test_resume_name_clash_fails(self):
+        first = self._auto("same")
+        pause_automation(self.conn, first["id"], now=T0)
+        self._auto("same")
+        ok, _ = resume_automation(self.conn, first["id"], now=T0)
+        self.assertFalse(ok)
+        self.assertEqual(get_automation(self.conn, first["id"])
+                         ["status"], "paused")
+
+    def test_retry_delay_backoff_and_cap(self):
+        self.assertEqual(retry_delay_seconds(1, 60), 60)
+        self.assertEqual(retry_delay_seconds(2, 60), 120)
+        self.assertEqual(retry_delay_seconds(3, 60), 240)
+        self.assertEqual(retry_delay_seconds(0, 60), 60)
+        self.assertEqual(retry_delay_seconds(10, 60), 3600)
+        self.assertEqual(retry_delay_seconds(2, 3600), 3600)
+
+    def test_record_failure_schedules_retry(self):
+        auto = self._auto()
+        out = record_automation_outcome(self.conn, auto["id"],
+                                        "failed", now=T0)
+        self.assertEqual(out["consecutive_failures"], 1)
+        self.assertEqual(out["next_retry_at"], T0 + 60)
+        out = record_automation_outcome(self.conn, auto["id"],
+                                        "failed", now=T0 + 60)
+        self.assertEqual(out["consecutive_failures"], 2)
+        self.assertEqual(out["next_retry_at"], T0 + 60 + 120)
+
+    def test_record_success_clears_failures(self):
+        auto = self._auto()
+        record_automation_outcome(self.conn, auto["id"], "failed",
+                                  now=T0)
+        out = record_automation_outcome(self.conn, auto["id"],
+                                        "completed", now=T0 + 60)
+        self.assertEqual(out["consecutive_failures"], 0)
+        self.assertIsNone(out["next_retry_at"])
+
+    def test_skipped_quota_is_not_a_failure(self):
+        auto = self._auto()
+        record_automation_outcome(self.conn, auto["id"], "failed",
+                                  now=T0)
+        out = record_automation_outcome(self.conn, auto["id"],
+                                        "skipped_quota", now=T0 + 10)
+        self.assertEqual(out["consecutive_failures"], 1)
+        self.assertEqual(out["next_retry_at"], T0 + 60)
+
+    def test_retry_exhausted_after_max_retries(self):
+        auto = self._auto(max_retries=1)
+        out = record_automation_outcome(self.conn, auto["id"],
+                                        "failed", now=T0)
+        self.assertEqual(out["next_retry_at"], T0 + 60)
+        out = record_automation_outcome(self.conn, auto["id"],
+                                        "failed", now=T0 + 60)
+        self.assertEqual(out["consecutive_failures"], 2)
+        self.assertIsNone(out["next_retry_at"])
+        self.assertEqual(retry_due_automations(self.conn, "p1",
+                                               now=T0 + 61), [])
+
+    def test_retry_due_boundary(self):
+        auto = self._auto()
+        record_automation_outcome(self.conn, auto["id"], "failed",
+                                  now=T0)
+        self.assertEqual(retry_due_automations(self.conn, "p1",
+                                               now=T0 + 59), [])
+        due = retry_due_automations(self.conn, "p1", now=T0 + 60)
+        self.assertEqual([d["id"] for d in due], [auto["id"]])
+
+    def test_claim_retry_single_winner_and_clock_untouched(self):
+        auto = self._auto()
+        record_automation_outcome(self.conn, auto["id"], "failed",
+                                  now=T0)
+        before = get_automation(self.conn, auto["id"])
+        self.assertTrue(claim_retry_automation(self.conn, auto["id"],
+                                               now=T0 + 60))
+        self.assertFalse(claim_retry_automation(self.conn, auto["id"],
+                                                now=T0 + 60))
+        after = get_automation(self.conn, auto["id"])
+        self.assertEqual(after["last_run_at"],
+                         before["last_run_at"])
+        self.assertIsNone(after["next_retry_at"])
+
+    def test_run_retry_automation_builds_task(self):
+        auto = self._auto()
+        record_automation_outcome(self.conn, auto["id"], "failed",
+                                  now=T0)
+        run = run_retry_automation(self.conn, auto["id"], now=T0 + 60)
+        self.assertIsNotNone(run)
+        self.assertEqual(run["status"], "created")
+        self.assertIsNotNone(run["task_id"])
+        self.assertIsNone(run_retry_automation(self.conn, auto["id"],
+                                               now=T0 + 60))
+
+    def test_run_retry_before_due_returns_none(self):
+        auto = self._auto()
+        record_automation_outcome(self.conn, auto["id"], "failed",
+                                  now=T0)
+        self.assertIsNone(run_retry_automation(self.conn, auto["id"],
+                                               now=T0 + 59))
+
+    def test_tick_caps_at_five_interval_first(self):
+        for i in range(4):
+            self._auto("due%d" % i, first_run_at=T0 - 10)
+        retry_ids = []
+        for i in range(2):
+            auto = self._auto("retry%d" % i,
+                              first_run_at=T0 + 100000)
+            record_automation_outcome(self.conn, auto["id"], "failed",
+                                      now=T0)
+            retry_ids.append(auto["id"])
+        tick = run_cron_tick(self.conn, "p1", now=T0 + 61)
+        self.assertEqual(len(tick["due"]), 4)
+        self.assertEqual(len(tick["retry_due"]), 1)
+        self.assertTrue(tick["capped"])
+        # 两重试任务 next_retry_at 与 created_at 完全相同，
+        # 排序按 id 排，不按创建顺序；只断言属于其中之一。
+        self.assertIn(tick["retry_due"][0]["id"], retry_ids)
+
+    def test_tick_empty_when_nothing_due(self):
+        self._auto(first_run_at=T0 + 100000)
+        tick = run_cron_tick(self.conn, "p1", now=T0)
+        self.assertEqual(tick["due"], [])
+        self.assertEqual(tick["retry_due"], [])
+        self.assertFalse(tick["capped"])
+
+    def test_definition_cap_100(self):
+        for i in range(100):
+            self._auto("cap%d" % i)
+        with self.assertRaises(ValueError):
+            self._auto("cap100")
+
+    def test_update_retry_params(self):
+        auto = self._auto()
+        ok, updated = update_automation(
+            self.conn, auto["id"],
+            {"max_retries": 5, "retry_backoff_seconds": 120},
+            now=T0)
+        self.assertTrue(ok)
+        self.assertEqual(updated["max_retries"], 5)
+        self.assertEqual(updated["retry_backoff_seconds"], 120)
+        with self.assertRaises(ValueError):
+            update_automation(self.conn, auto["id"],
+                              {"max_retries": 11}, now=T0)
+        with self.assertRaises(ValueError):
+            update_automation(self.conn, auto["id"],
+                              {"retry_backoff_seconds": 5}, now=T0)
+
+    def test_create_retry_params_validated(self):
+        with self.assertRaises(ValueError):
+            self._auto("bad1", max_retries=11)
+        with self.assertRaises(ValueError):
+            self._auto("bad2", retry_backoff_seconds=4000)
+        auto = self._auto("ok", max_retries=0,
+                          retry_backoff_seconds=10)
+        self.assertEqual(auto["max_retries"], 0)
+
+
+class SwarmTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state_dir = os.path.join(self.tmp.name, "state")
+        os.makedirs(self.state_dir)
+        self.root = os.path.join(self.tmp.name, "workspaces")
+        os.makedirs(self.root)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _swarm(self, max_agents=8):
+        return create_swarm(self.state_dir, "p1", "parent-1",
+                            max_agents=max_agents, now=T0)
+
+    def test_create_swarm_defaults_and_caps(self):
+        swarm = self._swarm()
+        self.assertEqual(swarm["max_agents"], 8)
+        self.assertEqual(swarm["status"], "active")
+        with self.assertRaises(ValueError):
+            create_swarm(self.state_dir, "p1", "parent-1",
+                         max_agents=0, now=T0)
+        with self.assertRaises(ValueError):
+            create_swarm(self.state_dir, "p1", "parent-1",
+                         max_agents=33, now=T0)
+
+    def test_project_swarm_cap_10(self):
+        for _ in range(10):
+            self._swarm()
+        with self.assertRaises(ValueError):
+            self._swarm()
+
+    def test_spawn_agent_own_workspace(self):
+        swarm = self._swarm()
+        a1 = spawn_agent(self.state_dir, swarm["id"], "task one",
+                         self.root, now=T0)
+        a2 = spawn_agent(self.state_dir, swarm["id"], "task two",
+                         self.root, now=T0)
+        self.assertNotEqual(a1["task_id"], a2["task_id"])
+        self.assertNotEqual(a1["workspace"], a2["workspace"])
+        self.assertTrue(a1["workspace"].startswith(self.root))
+        self.assertEqual(a1["status"], "pending")
+
+    def test_spawn_beyond_max_agents_fails(self):
+        swarm = self._swarm(max_agents=1)
+        spawn_agent(self.state_dir, swarm["id"], "one", self.root,
+                    now=T0)
+        with self.assertRaises(ValueError):
+            spawn_agent(self.state_dir, swarm["id"], "two", self.root,
+                        now=T0)
+
+    def test_check_agent_path_isolation(self):
+        swarm = self._swarm()
+        a1 = spawn_agent(self.state_dir, swarm["id"], "one",
+                         self.root, now=T0)
+        a2 = spawn_agent(self.state_dir, swarm["id"], "two",
+                         self.root, now=T0)
+        self.assertTrue(check_agent_path(
+            self.state_dir, swarm["id"], a1["id"],
+            os.path.join(a1["workspace"], "out.txt")))
+        self.assertTrue(check_agent_path(
+            self.state_dir, swarm["id"], a1["id"], a1["workspace"]))
+        self.assertFalse(check_agent_path(
+            self.state_dir, swarm["id"], a1["id"], a2["workspace"]))
+        self.assertFalse(check_agent_path(
+            self.state_dir, swarm["id"], a1["id"], self.root))
+        self.assertFalse(check_agent_path(
+            self.state_dir, swarm["id"], a1["id"],
+            os.path.join(a1["workspace"], "..", "escape.txt")))
+        self.assertFalse(check_agent_path(
+            self.state_dir, swarm["id"], "nope", a1["workspace"]))
+
+    def test_swarm_status_derived_from_agents(self):
+        swarm = self._swarm()
+        a1 = spawn_agent(self.state_dir, swarm["id"], "one",
+                         self.root, now=T0)
+        a2 = spawn_agent(self.state_dir, swarm["id"], "two",
+                         self.root, now=T0)
+        set_agent_status(self.state_dir, swarm["id"], a1["id"],
+                         "done", now=T0)
+        self.assertEqual(get_swarm(self.state_dir, swarm["id"])
+                         ["status"], "active")
+        set_agent_status(self.state_dir, swarm["id"], a2["id"],
+                         "failed", now=T0)
+        self.assertEqual(get_swarm(self.state_dir, swarm["id"])
+                         ["status"], "failed")
+        with self.assertRaises(ValueError):
+            spawn_agent(self.state_dir, swarm["id"], "three",
+                        self.root, now=T0)
+
+    def test_set_agent_status_invalid(self):
+        swarm = self._swarm()
+        agent = spawn_agent(self.state_dir, swarm["id"], "one",
+                            self.root, now=T0)
+        with self.assertRaises(ValueError):
+            set_agent_status(self.state_dir, swarm["id"], agent["id"],
+                             "nope", now=T0)
+        with self.assertRaises(ValueError):
+            set_agent_status(self.state_dir, swarm["id"], "nope",
+                             "done", now=T0)
+
+    def test_list_agents_sorted(self):
+        swarm = self._swarm()
+        spawn_agent(self.state_dir, swarm["id"], "one", self.root,
+                    now=T0)
+        spawn_agent(self.state_dir, swarm["id"], "two", self.root,
+                    now=T0 + 1)
+        agents = list_agents(self.state_dir, swarm["id"])
+        self.assertEqual(len(agents), 2)
+        self.assertEqual(agents[0]["prompt"], "one")
+        with self.assertRaises(ValueError):
+            list_agents(self.state_dir, "nope")
+
+    def test_register_tool_namespaced_per_agent(self):
+        swarm = self._swarm()
+        a1 = spawn_agent(self.state_dir, swarm["id"], "one",
+                         self.root, now=T0)
+        a2 = spawn_agent(self.state_dir, swarm["id"], "two",
+                         self.root, now=T0)
+        tool = register_tool(self.state_dir, swarm["id"], a1["id"],
+                             "search", "web search", now=T0)
+        self.assertEqual(tool["name"], "search")
+        with self.assertRaises(ValueError):
+            register_tool(self.state_dir, swarm["id"], a1["id"],
+                          "search", "again", now=T0)
+        register_tool(self.state_dir, swarm["id"], a2["id"],
+                      "search", "same name other agent", now=T0)
+        with self.assertRaises(ValueError):
+            register_tool(self.state_dir, swarm["id"], a1["id"],
+                          "x" * 101, "too long", now=T0)
+
+    def test_record_tool_invocation_scoped(self):
+        swarm = self._swarm()
+        a1 = spawn_agent(self.state_dir, swarm["id"], "one",
+                         self.root, now=T0)
+        a2 = spawn_agent(self.state_dir, swarm["id"], "two",
+                         self.root, now=T0)
+        register_tool(self.state_dir, swarm["id"], a1["id"],
+                      "search", "web search", now=T0)
+        record = record_tool_invocation(
+            self.state_dir, swarm["id"], a1["id"], "search",
+            "q=example", now=T0)
+        self.assertEqual(record["name"], "search")
+        with self.assertRaises(ValueError):
+            record_tool_invocation(self.state_dir, swarm["id"],
+                                   a2["id"], "search", "q=x", now=T0)
+        with self.assertRaises(ValueError):
+            record_tool_invocation(self.state_dir, swarm["id"],
+                                   a1["id"], "nosuch", "q=x", now=T0)
+
+    def test_dispatch_agent_runs_runner_once(self):
+        swarm = self._swarm()
+        agent = spawn_agent(self.state_dir, swarm["id"], "do it",
+                            self.root, now=T0)
+        seen = []
+
+        def runner(task):
+            seen.append(task)
+            return {"status": "ok", "task_id": task["id"]}
+
+        result = dispatch_agent(self.state_dir, swarm["id"],
+                                agent["id"], runner, now=T0)
+        self.assertEqual(result["status"], "ok")
+        self.assertEqual(seen[0]["id"], agent["task_id"])
+        self.assertEqual(seen[0]["workspace"], agent["workspace"])
+        with self.assertRaises(ValueError):
+            dispatch_agent(self.state_dir, swarm["id"], agent["id"],
+                           runner, now=T0)
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class PresentationTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state_dir = os.path.join(self.tmp.name, "state")
+        os.makedirs(self.state_dir)
+        self.ledger = NotificationLedger(self.state_dir)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_needs_input_and_failure_present(self):
+        self.assertEqual(
+            evaluate_presentation({"status": "needs_input"}),
+            {"present": True, "reason": "needs_input"})
+        self.assertEqual(
+            evaluate_presentation({"status": "failed"}),
+            {"present": True, "reason": "failed_needs_decision"})
+        self.assertEqual(
+            evaluate_presentation({"status": "interrupted"}),
+            {"present": True, "reason": "failed_needs_decision"})
+
+    def test_completed_with_new_content_presents(self):
+        self.assertTrue(evaluate_presentation(
+            {"status": "completed",
+             "new_artifacts": ["report.pdf"]})["present"])
+        self.assertTrue(evaluate_presentation(
+            {"status": "completed",
+             "changed_files": ["a.py"]})["present"])
+        self.assertTrue(evaluate_presentation(
+            {"status": "completed", "notify": True})["present"])
+
+    def test_routine_completed_and_cancelled_log_only(self):
+        self.assertEqual(
+            evaluate_presentation({"status": "completed"}),
+            {"present": False, "reason": "routine_completed"})
+        self.assertEqual(
+            evaluate_presentation({"status": "completed",
+                                   "new_artifacts": [],
+                                   "changed_files": []}),
+            {"present": False, "reason": "routine_completed"})
+        self.assertEqual(
+            evaluate_presentation({"status": "cancelled"}),
+            {"present": False, "reason": "cancelled"})
+
+    def test_unknown_status_fail_closed(self):
+        self.assertEqual(
+            evaluate_presentation({"status": "weird"}),
+            {"present": False, "reason": "unknown_status"})
+        self.assertEqual(
+            evaluate_presentation("not-a-dict"),
+            {"present": False, "reason": "unknown_status"})
+
+    def test_ledger_dedups_same_reason(self):
+        self.assertTrue(self.ledger.should_notify("t1", "failed",
+                                                  now=T0))
+        self.assertFalse(self.ledger.should_notify("t1", "failed",
+                                                   now=T0 + 10))
+        self.assertTrue(self.ledger.should_notify("t1", "needs_input",
+                                                  now=T0 + 10))
+        self.assertTrue(self.ledger.should_notify("t2", "failed",
+                                                  now=T0 + 10))
+
+    def test_ledger_window_expires(self):
+        self.assertTrue(self.ledger.should_notify("t1", "failed",
+                                                  now=T0))
+        self.assertFalse(self.ledger.should_notify("t1", "failed",
+                                                   now=T0 + 3599))
+        self.assertTrue(self.ledger.should_notify("t1", "failed",
+                                                  now=T0 + 3600))
+
+    def test_ledger_rejects_empty_keys(self):
+        with self.assertRaises(ValueError):
+            self.ledger.should_notify("", "failed", now=T0)
+        with self.assertRaises(ValueError):
+            self.ledger.should_notify("t1", "", now=T0)
+
+    def test_ledger_corrupt_file_recovers(self):
+        self.assertTrue(self.ledger.should_notify("t1", "failed",
+                                                  now=T0))
+        with open(os.path.join(self.state_dir,
+                               "notification-ledger.json"),
+                   "w") as fh:
+            fh.write("{not json")
+        self.assertTrue(self.ledger.should_notify("t2", "failed",
+                                                  now=T0))
+
+    def test_present_or_log_combined(self):
+        out = present_or_log({"status": "failed"}, self.ledger,
+                             "t1", now=T0)
+        self.assertEqual(out, {"present": True,
+                               "reason": "failed_needs_decision",
+                               "notified": True})
+        out = present_or_log({"status": "failed"}, self.ledger,
+                             "t1", now=T0 + 5)
+        self.assertTrue(out["present"])
+        self.assertFalse(out["notified"])
+        out = present_or_log({"status": "completed"}, self.ledger,
+                             "t2", now=T0)
+        self.assertEqual(out, {"present": False,
+                               "reason": "routine_completed",
+                               "notified": False})
+
+
+class FakeTransport(BrowserTransport):
+    """测试用 transport：记录 spawn/close，可注入失败。"""
+
+    def __init__(self):
+        self.spawned = []
+        self.closed = []
+        self.fail_spawn = False
+        self.fail_close = False
+
+    def spawn(self, instance):
+        if self.fail_spawn:
+            raise RuntimeError("spawn exploded")
+        self.spawned.append(instance["instance_id"])
+
+    def close(self, instance):
+        if self.fail_close:
+            raise RuntimeError("close exploded")
+        self.closed.append(instance["instance_id"])
+
+    def ping(self, instance):
+        return instance["instance_id"] in self.spawned \
+            and instance["instance_id"] not in self.closed
+
+
+class BrowserTasksTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.state_dir = os.path.join(self.tmp.name, "state")
+        os.makedirs(self.state_dir)
+        self.transport = FakeTransport()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_create_and_get_instance(self):
+        inst = create_instance(self.state_dir, "task-1",
+                               self.transport, now=T0)
+        self.assertEqual(inst["status"], "active")
+        self.assertEqual(inst["task_id"], "task-1")
+        self.assertEqual(self.transport.spawned, [inst["instance_id"]])
+        got = get_instance(self.state_dir, "task-1")
+        self.assertEqual(got["instance_id"], inst["instance_id"])
+        self.assertIsNone(get_instance(self.state_dir, "nope"))
+
+    def test_two_tasks_get_distinct_instances(self):
+        a = create_instance(self.state_dir, "task-1", self.transport,
+                            now=T0)
+        b = create_instance(self.state_dir, "task-2", self.transport,
+                            now=T0)
+        self.assertNotEqual(a["instance_id"], b["instance_id"])
+        self.assertNotEqual(
+            instance_shots_dir(self.state_dir, "task-1"),
+            instance_shots_dir(self.state_dir, "task-2"))
+
+    def test_second_create_same_task_fails(self):
+        create_instance(self.state_dir, "task-1", self.transport,
+                        now=T0)
+        with self.assertRaises(ValueError):
+            create_instance(self.state_dir, "task-1", self.transport,
+                            now=T0)
+
+    def test_destroy_marks_destroyed_and_rejects_reuse(self):
+        inst = create_instance(self.state_dir, "task-1",
+                               self.transport, now=T0)
+        destroyed = destroy_instance(self.state_dir, "task-1",
+                                     self.transport, now=T0 + 5)
+        self.assertEqual(destroyed["status"], "destroyed")
+        self.assertEqual(destroyed["destroyed_at"], T0 + 5)
+        self.assertEqual(self.transport.closed, [inst["instance_id"]])
+        with self.assertRaises(ValueError):
+            destroy_instance(self.state_dir, "task-1", self.transport,
+                             now=T0 + 6)
+        with self.assertRaises(ValueError):
+            create_instance(self.state_dir, "task-1", self.transport,
+                            now=T0 + 6)
+        self.assertFalse(ping_instance(self.state_dir, "task-1",
+                                      self.transport))
+
+    def test_destroy_missing_fails(self):
+        with self.assertRaises(ValueError):
+            destroy_instance(self.state_dir, "nope", self.transport,
+                             now=T0)
+
+    def test_spawn_failure_records_failed(self):
+        self.transport.fail_spawn = True
+        with self.assertRaises(RuntimeError):
+            create_instance(self.state_dir, "task-1", self.transport,
+                            now=T0)
+        got = get_instance(self.state_dir, "task-1")
+        self.assertEqual(got["status"], "failed")
+
+    def test_close_failure_still_destroys(self):
+        create_instance(self.state_dir, "task-1", self.transport,
+                        now=T0)
+        self.transport.fail_close = True
+        destroyed = destroy_instance(self.state_dir, "task-1",
+                                     self.transport, now=T0)
+        self.assertEqual(destroyed["status"], "destroyed")
+        self.assertIn("close exploded", destroyed["close_error"])
+
+    def test_ping_active_and_missing(self):
+        self.assertFalse(ping_instance(self.state_dir, "task-1",
+                                      self.transport))
+        create_instance(self.state_dir, "task-1", self.transport,
+                        now=T0)
+        self.assertTrue(ping_instance(self.state_dir, "task-1",
+                                     self.transport))
+
+    def test_shots_dir_inside_state_dir(self):
+        create_instance(self.state_dir, "task-1", self.transport,
+                        now=T0)
+        shots = instance_shots_dir(self.state_dir, "task-1")
+        self.assertTrue(shots.startswith(self.state_dir))
+        self.assertIsNone(instance_shots_dir(self.state_dir, "nope"))
+
+    def test_base_transport_raises(self):
+        base = BrowserTransport()
+        with self.assertRaises(NotImplementedError):
+            base.spawn({})
+        with self.assertRaises(NotImplementedError):
+            base.close({})
+        with self.assertRaises(NotImplementedError):
+            base.ping({})

docs/chv-muse-p8-gap.md

@@ -0,0 +1,109 @@
+# 对齐 Muse P8 差距表：后台工作（crons / 子代理群 / 回访通知 / 每任务浏览器子代理）
+
+> 先写表再动手。口径分层：= Meta 安全文（security.muse.ai 2026-09-08）与产品
+> 设计文（introducing.muse.ai）原文；= 媒体/评测转述；=
+> 二进制拆解与用户 VM 观察。凡是 Meta 未公开的数值（cron 调度粒度、数量上限、失败重试
+> 策略），一律写清是**本项目自定**，不得写成 Meta 官方值。
+
+## 1. Muse 官方要求（出处：研究报告 §2.5）
+
+| # | 要求 | 来源 |
+|---|---|---|
+| B1 | 关应用后继续工作；按计划与相关事件推进 | 设计文 |
+| B2 | 完成后台工作后评估是否“值得呈现”，只在真正有新内容或需要输入时通知用户 | 设计文 |
+| B3 | 启动成群的子代理、构建自己的工具、编辑自身；VM 需要足够存储、CPU 与内存来做真正的工作——如编译代理写的代码、开发自定义 Skills、处理并发子代理与 crons | 安全文 |
+| B4 | 每个浏览器任务由专门浏览器子代理完成 | 安全文 |
+| B5 | 定时任务（crons）与并发子代理并存 | 安全文（只提存在） |
+| B6 | cron 的具体调度粒度、数量上限、失败重试策略 | Meta **未公开**——本项目自定，见 §4 |
+
+**不在本轨官方口径内的东西：** “目标工作器（objective workers）”一词只见于第三方
+VM 观察，未见于官方页面原文，本轨不把它当官方术语用；长期目标的分派与通知步骤
+用本轨的 swarm + presentation 实现，不挂这个名头。
+
+## 2. 现状（基线 06c3e17，实测代码为准）
+
+- **crons，两套现成的、都不完整：**
+- `host.py run-due`：host state JSON（name/task/repo/interval/last_run），
+`now - last_run >= interval` 到期即派发 host 任务，跑完才回写 last_run。
+无重试、无失败计数、无启用/停用（只有删定义）。
+- `cloud/automations.py`：控制面 automations 表（SQLite/Postgres），
+`claim_due_automation` 条件 UPDATE 抢单（同一间隔只一个执行器拿到），
+配额门、`automation_runs` 产出记录、分诊收件箱。无重试策略、无 paused
+状态、无单项目上限。`MIN_INTERVAL_SECONDS=60`、`MAX_INTERVAL_SECONDS=30d` 已有。
+- **子代理群：没有。** 只有模型层的子代理名（`cloud/harness2.py`
+`subagent_names`，prompt 树里的名字），没有执行层的并行派发、没有隔离、
+没有工具注册。
+- **后台回访通知：没有“值得呈现”评估。** `cloud/push.py dispatch_event`
+按订阅派发终态（completed/failed/goal-unmet 等）；自动化模块 docstring
+已写明“不新写通知代码”。无去重、无节流——同一终态可重复派发。
+- **每任务浏览器子代理：没有。** P2 的 `chv-browser.py` 是常驻进程，
+`close` 只关当前页、进程继续跑、等下一次 open；broker 固定转发到
+`/run/chv/browser.sock`。状态跨任务复用。
+
+## 3. 改什么（本轨四个新模块 + 一处扩展）
+
+| 模块 | 内容 | 归属 |
+|---|---|---|
+| `cloud/chv/crons.py`（新） | 在 `cloud/automations.py` 之上加：启用/停用（paused 状态）、失败重试（指数退避 + 上限）、单项目并发上限与定义上限、tick 驱动（到期 + 重试到期统一认领） | 本轨 |
+| `cloud/chv/swarm.py`（新） | 子代理群：建群/派代理/生命周期；每代理独立任务 id + 独立工作区（`resolve_inside` 门，越界拒绝）；工具注册与审计（只登记与审计调用意图，执行在代理自己的 VM 里走执行器）；上限可配 | 本轨 |
+| `cloud/chv/presentation.py`（新） | “值得呈现”评估（需要输入/失败需决策/有实质新内容才通知，常规完成只记日志）+ 通知账本（每任务每原因节流去重） | 本轨 |
+| `cloud/chv/browser_tasks.py`（新） | 每浏览器任务一个实例：建/绑/毁生命周期、单实例单任务（复用即错、毁后复用即错）、每实例独立状态目录；transport 接口注入（测试用 fake） | 本轨 |
+| `cloud/automations.py`（改） | 只加重试/停用所需的列与抢单入口（`max_retries`、`retry_backoff_seconds`、`consecutive_failures`、`next_retry_at`、`paused` 状态）；host.py run-due **不动**（host 侧唯一执行器，见该模块 docstring） | 本轨 |
+
+**P7 状态落点（更新 2026-11-07 06:39）：** P7 已完成并提交（分支
+`feat-chv-muse-p7`，提交 `29fa0c8`：每 VM 一个 SQLite 状态库
+AppStateStore，0600 WAL，tasks/goals/sessions/只追加 audit/meta）。
+本轨仍先用 `state_dir` 下的原子 JSON 文件（复用
+`cloud/chv/atomic_file.py`），模块 docstring 与本表明确标注“待迁移到
+P7 状态存储”；不重复造一套状态库。集成轨（round-12）把
+`swarms.json` / `notification-ledger.json` / `browser-tasks.json`
+迁到 AppStateStore，接口（load/save/原子写）已按同一模式写好，迁移
+只是换后端。crons 与 swarm 的状态复用 `automations` 表与 swarm 登记
+文件（同款原子写），同样标注待迁移。
+
+**客机侧跟进（本轨只写规约、不改 P2 的文件）：** 每任务浏览器实例的客机
+进程拉起/销毁在 P2 的 `chv-guest/chv-broker.py`（`browser` 操作固定转发到
+单例 socket）里改：每浏览器任务起一个全新的 `chv-browser.py`（每任务
+socket），`close` 后杀掉进程。P2 的文件不在本分支基线里，本轨不复制、不
+分叉，规约写进本表 §6，留给 round-12 集成时落在 P2 分支上。
+
+## 4. 自定数值清单（全部是本项目自定，非 Meta 官方值）
+
+| 数值 | 本项目自定值 | 说明 |
+|---|---|---|
+| cron 最小间隔 | 60 秒（沿用现有 `MIN_INTERVAL_SECONDS`） | 已有，不新定 |
+| cron 最大间隔 | 30 天（沿用现有 `MAX_INTERVAL_SECONDS`） | 已有，不新定 |
+| 失败重试默认次数 | 3 次 | 超过即停，不再重试 |
+| 重试退避 | 基数 60 秒 × 2^(n-1)，上限 3600 秒 | n 为第几次重试 |
+| 单项目并发 cron 执行 | 5 | tick 单轮最多认领 5 个 |
+| 单项目自动化定义上限 | 100 | 超过建定义即 ValueError |
+| 每群子代理上限 | 8 | 建群时可配，1–32 |
+| 每项目群上限 | 10 | 超过建群即 ValueError |
+| 通知节流窗口 | 同一任务同一原因 3600 秒内只通知一次 | 账本持久化 |
+| 工具名长度上限 | 100 字符 | 与自动化名上限同款 |
+
+## 5. 诚实边界
+
+- B3 的“VM 需要足够资源”是容量要求，不是本轨能代码保证的：本轨只做上限与
+隔离，宿主容量仍由部署方保障；群上限 8 是防炸宿主的默认值，不是性能承诺。
+- swarm 层**不执行工具**：工具调用意图只登记与审计，真正执行在子代理自己的
+VM/单元里（P1 常驻池 + P2 单元是底座）。“子代理构建自己的工具”在本轨指
+注册进群内工具表、可被审计、可被别的代理复用调用名——不是沙箱外任意执行。
+- “实质新内容”的判定是确定性规则（新产物/新改动文件/显式 notify 标志），
+不是模型判断；模型觉得重要但规则判无新内容时，以规则为准、只记日志。
+- cron 重试只针对 failed/interrupted 的执行；超配额跳过（skipped_quota）
+不是失败，不进重试（沿用 automations 模块现有口径）。
+- 通知账本先走原子 JSON 文件，P7 存储（`feat-chv-muse-p7` 提交
+`29fa0c8` 的 AppStateStore）已落地，集成轨再迁移；迁移前账本丢了只
+影响去重（可能重复通知一次），不影响任务执行本身。
+
+## 6. 真机待验证清单（round-11 与 P4/P5/P6 之后，等通知）
+
+1. cron 端到端：定义在测试机上按间隔触发、失败重试 3 次后停、停用后不再触发。
+2. 群端到端：3 个子代理在各自 VM 里并行跑、互不可见对方工作区。
+3. 回访通知端到端：后台任务完成后有新内容才通知、无新内容不通知、同一原因
+1 小时内不重复通知。
+4. 每任务浏览器实例端到端：连续两个浏览器任务各起全新实例、第二个看不到
+第一个的截图与快照状态（需 §3 客机侧跟进先落地）。
+5. 账本迁移：P7 存储已落地（AppStateStore，`feat-chv-muse-p7`
+提交 `29fa0c8`），集成轨把通知账本从 JSON 文件迁过去、去重语义不变。