"""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: """把损坏的账本挪到 .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}