""" heartbeat_helper.py — 高频 Agent 心跳辅助脚本 提供心跳脚本中所有通用功能，底层通过 multica_proxy 调用 multica CLI， 自动享受缓存和限流保护。 用法： from heartbeat_helper import check_my_tasks, check_timeouts, check_dependencies 作者：陆怀瑾（COO） 日期：2026-06-23 """ import os import sys import json import time from typing import Any, Dict, List, Optional _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) if _SCRIPT_DIR not in sys.path: sys.path.insert(0, _SCRIPT_DIR) from multica_proxy import ( run_multica, multica_issue_list_my_todo, multica_issue_list_in_progress, multica_issue_get, openclaw_workboard_list, openclaw_workboard_read, get_cache_stats, clear_cache, start_coordinated_poller, subscribe_to_poller, get_poller_status, health_check, ) # ============================================================================ # Agent 配置 # ============================================================================ AGENT_CONFIGS = { "coo": { "name": "陆怀瑾", "multica_uuid": "1c38b437-b54d-4784-bda3-29ce4c8a6722", "openclaw_agent_id": "coo", "is_coo": True, }, "secretary": { "name": "刘诗妮", "multica_uuid": "b024fcdc-30ff-420d-b289-498041466e1b", "openclaw_agent_id": "secretary", "is_coo": False, }, "projectmanager": { "name": "胡蓉", "multica_uuid": "d877b8c3-b230-4073-b3f7-80e148cfdb71", "openclaw_agent_id": "projectmanager", "is_coo": False, }, "costcodev": { "name": "徐聪", "multica_uuid": "46bdd4a6-5c64-475a-92ef-36a763602fa1", "openclaw_agent_id": "costcodev", "is_coo": False, }, "opengineer": { "name": "严维序", "multica_uuid": "d3804433-9e2e-4199-a92b-a153049b3bc9", "openclaw_agent_id": "opengineer", "is_coo": False, }, "productmanager": { "name": "沈路明", "multica_uuid": "a101fa88-d821-4839-9754-e04580d5fd68", "openclaw_agent_id": "productmanager", "is_coo": False, }, "architect": { "name": "梁思筑", "multica_uuid": "40abd41a-62d0-416d-bc44-92c1f758d87a", "openclaw_agent_id": "architect", "is_coo": False, }, "designer": { "name": "苏锦绘", "multica_uuid": "13bd8968-cc2a-4934-90c7-957a2d3c09c2", "openclaw_agent_id": "designer", "is_coo": False, }, "contentspecialist": { "name": "文墨言", "multica_uuid": "8321b0bf-7d89-4ece-927a-0780f42ad396", "openclaw_agent_id": "contentspecialist", "is_coo": False, }, "cvexpert": { "name": "程伯予", "multica_uuid": "4a8696fd-6531-40da-8956-ef84d7ea3c43", "openclaw_agent_id": "cvexpert", "is_coo": False, }, "prompt-engineer": { "name": "许言", "multica_uuid": "ece81d8e-8a24-4dd8-a7af-8adfc54b9d01", "openclaw_agent_id": "prompt-engineer", "is_coo": False, }, "mediaspecialist": { "name": "钟帧韵", "multica_uuid": "e2b587d4-1d16-447c-8ad9-e2a01358ff0a", "openclaw_agent_id": "mediaspecialist", "is_coo": False, }, "taobaospecialist": { "name": "陆云帆", "multica_uuid": "e0f62d8f-9568-4f41-8ad4-b73d79a163a7", "openclaw_agent_id": "taobaospecialist", "is_coo": False, }, "marketanalysis": { "name": "顾析策", "multica_uuid": "5ed91729-658f-4654-98f0-3e0313022002", "openclaw_agent_id": "marketanalysis", "is_coo": False, }, "lawyer": { "name": "苏慎", "multica_uuid": "6fb0fbd2-16a6-4566-ba7a-d2c136baec25", "openclaw_agent_id": "lawyer", "is_coo": False, }, } def get_agent_config(agent_id: str) -> Dict[str, Any]: """获取 Agent 配置""" config = AGENT_CONFIGS.get(agent_id) if config is None: raise ValueError(f"Unknown agent: {agent_id}. Known: {list(AGENT_CONFIGS.keys())}") return config # ============================================================================ # 三源任务检查 # ============================================================================ def check_workboard_tasks(agent_id: str) -> List[Dict[str, Any]]: """ 检查 WorkBoard 中分配给当前 Agent 的待办卡片 替代内联 bash 脚本 """ result = openclaw_workboard_list() if not result["success"]: print(f"[heartbeat] WorkBoard 查询失败: {result['error']}") return [] data = result["data"] my_cards = [ c for c in data.get("cards", []) if c.get("agentId") == agent_id and c.get("status") == "todo" ] return my_cards def check_multica_tasks(agent_id: str) -> List[Dict[str, Any]]: """ 检查 Multica 中分配给当前 Agent 的待办 Issue 替代内联 bash 脚本 """ config = get_agent_config(agent_id) result = multica_issue_list_my_todo(config["multica_uuid"]) if not result["success"]: print(f"[heartbeat] Multica 查询失败: {result['error']}") return [] data = result["data"] if isinstance(data, list): return data return [] def check_todo_docs(workspace_dir: str) -> List[str]: """ 检查工作区待办文档中的未完成项 """ items = [] for filename in ["TODO.md", "AGENTS.md"]: filepath = os.path.join(workspace_dir, filename) if os.path.exists(filepath): try: with open(filepath) as f: for i, line in enumerate(f, 1): if "[ ]" in line: items.append(f"{filename}:{i}: {line.strip()}") except Exception: pass return items def check_my_tasks(agent_id: str, workspace_dir: str) -> Dict[str, Any]: """ 三源合并检查：WorkBoard + Multica + 待办文档 """ wb_tasks = check_workboard_tasks(agent_id) mul_tasks = check_multica_tasks(agent_id) doc_tasks = check_todo_docs(workspace_dir) return { "workboard": wb_tasks, "multica": mul_tasks, "documents": doc_tasks, "total": len(wb_tasks) + len(mul_tasks) + len(doc_tasks), } # ============================================================================ # 超时检测 # ============================================================================ TIMEOUT_SECONDS = 1200 # 20 分钟 def check_workboard_timeouts() -> List[Dict[str, Any]]: """ 检查 WorkBoard 中超过 20 分钟无进展的进行中任务 """ result = openclaw_workboard_list() if not result["success"]: print(f"[heartbeat] WorkBoard 超时检测失败: {result['error']}") return [] data = result["data"] now = time.time() timeouts = [] for c in data.get("cards", []): if c.get("status") != "in_progress": continue updated = c.get("updated_at", "") if updated: try: age = now - time.mktime(time.strptime(updated[:19], "%Y-%m-%dT%H:%M:%S")) if age > TIMEOUT_SECONDS: timeouts.append(c) except (ValueError, OverflowError): pass return timeouts def check_multica_timeouts() -> List[Dict[str, Any]]: """ 检查 Multica 中超过 20 分钟无进展的进行中 Issue """ result = multica_issue_list_in_progress() if not result["success"]: print(f"[heartbeat] Multica 超时检测失败: {result['error']}") return [] data = result["data"] now = time.time() timeouts = [] if isinstance(data, list): for issue in data: updated = issue.get("updated_at", "") if updated: try: age = now - time.mktime(time.strptime(updated[:19], "%Y-%m-%dT%H:%M:%S")) if age > TIMEOUT_SECONDS: timeouts.append(issue) except (ValueError, OverflowError): pass return timeouts def check_timeouts() -> Dict[str, Any]: """ 跨平台超时检测 """ wb_timeouts = check_workboard_timeouts() mul_timeouts = check_multica_timeouts() return { "workboard_timeouts": wb_timeouts, "multica_timeouts": mul_timeouts, "total_timeouts": len(wb_timeouts) + len(mul_timeouts), } # ============================================================================ # 依赖检查 # ============================================================================ def check_workboard_dependencies(card_id: str) -> Dict[str, Any]: """ 检查 WorkBoard 卡片的依赖是否满足 """ result = openclaw_workboard_read(card_id) if not result["success"]: return {"satisfied": False, "error": result["error"], "unmet": []} card = result["data"] deps = card.get("dependsOn", []) unmet = [dep for dep in deps if dep.get("status") != "done"] return { "satisfied": len(unmet) == 0, "total_deps": len(deps), "unmet": unmet, } def check_multica_dependencies(issue_id: str) -> Dict[str, Any]: """ 检查 Multica Issue 的父 Issue 依赖是否满足 """ result = multica_issue_get(issue_id) if not result["success"]: return {"satisfied": False, "error": result["error"], "unmet": []} issue = result["data"] parent_id = issue.get("parent_issue_id") if not parent_id: return {"satisfied": True, "total_deps": 0, "unmet": []} parent_result = multica_issue_get(parent_id) if not parent_result["success"]: return {"satisfied": False, "error": f"Failed to check parent {parent_id}", "unmet": [parent_id]} parent = parent_result["data"] if parent.get("status") != "done": return {"satisfied": False, "total_deps": 1, "unmet": [{"id": parent_id, "identifier": parent.get("identifier"), "status": parent.get("status")}]} return {"satisfied": True, "total_deps": 1, "unmet": []} # ============================================================================ # 全局积压巡检（COO 专用） # ============================================================================ def check_global_backlog() -> Dict[str, Any]: """ 全平台积压巡检：WorkBoard + Multica 全局待办数 """ wb_result = openclaw_workboard_list() mul_result = multica_issue_list_in_progress() wb_stats = {"total": 0, "todo": 0, "in_progress": 0, "done": 0} if wb_result["success"]: cards = wb_result["data"].get("cards", []) wb_stats["total"] = len(cards) for c in cards: status = c.get("status", "") if status in wb_stats: wb_stats[status] += 1 mul_stats = {"total": 0, "in_progress": 0} if mul_result["success"] and isinstance(mul_result["data"], list): mul_stats["total"] = len(mul_result["data"]) mul_stats["in_progress"] = mul_stats["total"] return { "workboard": wb_stats, "multica": mul_stats, } # ============================================================================ # 心跳主入口 # ============================================================================ def run_heartbeat(agent_id: str, workspace_dir: str) -> Dict[str, Any]: """ 执行完整心跳检查 参数： agent_id: Agent ID（如 "coo", "secretary"） workspace_dir: 工作区目录路径 返回： 心跳结果字典 """ config = get_agent_config(agent_id) is_coo = config["is_coo"] result = { "agent": config["name"], "agent_id": agent_id, "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"), "tasks": check_my_tasks(agent_id, workspace_dir), "timeouts": check_timeouts(), } # COO 额外检查 if is_coo: result["global_backlog"] = check_global_backlog() result["cache_stats"] = get_cache_stats() result["poller_status"] = get_poller_status() return result def print_heartbeat_report(result: Dict[str, Any]) -> None: """打印格式化的心跳报告""" print(f"\n{'='*60}") print(f" 🫀 心跳报告 — {result['agent']} ({result['agent_id']})") print(f" ⏰ {result['timestamp']}") print(f"{'='*60}") tasks = result["tasks"] print(f"\n📋 任务检查:") print(f" WorkBoard 待办: {len(tasks['workboard'])}") for t in tasks["workboard"]: print(f" ⚠️ WB TODO: {t['id'][:8]} → {t.get('agentId','?')} - {t.get('title','?')[:50]}") print(f" Multica 待办: {len(tasks['multica'])}") for t in tasks["multica"]: print(f" ⚠️ MUL TODO: {t.get('identifier','?')} - {t.get('title','?')[:50]}") print(f" 文档待办: {len(tasks['documents'])}") for d in tasks["documents"]: print(f" 📝 {d}") timeouts = result["timeouts"] print(f"\n⏱️ 超时检测:") print(f" WorkBoard 超时: {len(timeouts['workboard_timeouts'])}") for t in timeouts["workboard_timeouts"]: print(f" ⏰ WB TIMEOUT: {t['id'][:8]} [{t.get('agentId','?')}] {t.get('title','?')[:50]}") print(f" Multica 超时: {len(timeouts['multica_timeouts'])}") for t in timeouts["multica_timeouts"]: print(f" ⏰ MUL TIMEOUT: {t.get('identifier','?')} {t.get('title','?')[:50]}") if "global_backlog" in result: gb = result["global_backlog"] print(f"\n📊 全局积压:") print(f" WorkBoard: {gb['workboard']}") print(f" Multica: {gb['multica']}") if "cache_stats" in result: print(f"\n💾 缓存: {result['cache_stats']}") print(f"\n{'='*60}\n") # ============================================================================ # CLI 入口 # ============================================================================ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Agent 心跳辅助脚本") parser.add_argument("agent_id", help="Agent ID (coo/secretary/projectmanager/costcodev/opengineer)") parser.add_argument("--workspace", "-w", default=os.getcwd(), help="工作区目录") parser.add_argument("--json", action="store_true", help="JSON 输出") parser.add_argument("--health", action="store_true", help="健康检查") parser.add_argument("--clear-cache", action="store_true", help="清理缓存") args = parser.parse_args() if args.health: print(json.dumps(health_check(), indent=2, ensure_ascii=False)) elif args.clear_cache: count = clear_cache() print(f"已清理 {count} 条缓存") else: result = run_heartbeat(args.agent_id, args.workspace) if args.json: print(json.dumps(result, indent=2, ensure_ascii=False, default=str)) else: print_heartbeat_report(result)