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/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}