--- name: hermes-memory-bridge description: Hermes Agent 与 WorkBuddy 双向记忆互通 Skill。触发词：同步到hermes、读取hermes记忆、hermes会话历史、跨记忆搜索、记忆互通、bridge状态、hermes统计、环境变量、错误处理、信号事件、信号通知、发送任务、WorkBuddy执行、闭环命令、协同进化、co_evolution、auto_learn、bridge统计、sync_from_hermes、同步账本、进化状态 version: 2.1.0 --- # hermes-memory-bridge > v2.1 | WorkBuddy ↔ Hermes Agent 双向记忆桥梁（闭环命令版） > > **里程碑**：2026-04-19 协同进化评分达到 **8.7/10**（v3.4.6），系统进入稳定维护阶段。 WorkBuddy 与 Hermes Agent 之间的双向记忆桥梁，支持**信号事件**、**自适应轮询**和**命令任务闭环处理**。 ## 架构概览 ``` Hermes 侧共享目录 WorkBuddy 侧 │ │ │ │ event_signaler.py emit ──→ signals/ ←── event_watcher.py │ │ event_signaler.py ──────→ signals/ ←── FSEvents/Poller │ │ send_task ─────────────→ sig_task_*.json │ │ │ │ │ feedback poll ←────────── feedback/ ←── task_processor.py │ │ │ │ │ │ ← feedback_writer.py │ │ ←──────────────────── feedback/*.json │ ``` **v2.0 核心组件**： - `event_signaler.py` — Hermes 侧：发射信号 + **发送命令任务** + **轮询处理结果** - `event_watcher.py` — WorkBuddy 侧：监听 Hermes 信号 + **调用任务处理器** + **回写结果** - `task_processor.py` — **WorkBuddy 任务处理器**：执行 Hermes 发来的命令 - `feedback_writer.py` — **结果回写器**：WorkBuddy 执行结果 → Hermes 可读 - `communication_queue.py` — 消息队列 + 信号事件 + ACK 确认 ## 存储布局 | 路径 | 用途 | |------|------| | `~/.hermes/shared/signals/` | 信号目录（Hermes 发射命令/通知，WorkBuddy 读取并标记已处理） | | `~/.hermes/shared/feedback/` | **v2.0** 反馈目录（WorkBuddy 回写处理结果，Hermes 轮询读取） | | `~/.hermes/shared/queue/` | 消息队列目录（异步通信） | | `~/.hermes/shared/hermes.log` | Hermes 运行日志 | | `~/.hermes/memories/MEMORY.md` | Hermes 个人笔记（WorkBuddy 可写） | | `~/.hermes/memories/USER.md` | 用户画像 | ## 环境变量配置 | 变量 | 默认值 | 说明 | |------|--------|------| | `HERMES_HOME` | `~/.hermes` | Hermes 根目录 | | `WORKBUDDY_HOME` | `~/WorkBuddy` | WorkBuddy 根目录 | | `WORKBUDDY_MEMORY_DIR` | （自动发现） | 强制指定 WorkBuddy 记忆目录 | | `BRIDGE_LOG_LEVEL` | `INFO` | 日志级别 | ## v2.0 闭环命令系统 ### Hermes → WorkBuddy 命令流程 ```bash # 1. Hermes 发送命令 python3 ~/.hermes/event_signaler.py send_task '' # 2. WorkBuddy 处理（后台守护进程自动处理，或手动触发） python3 ~/.workbuddy/skills/hermes-memory-bridge/event_watcher.py --once # 3. Hermes 轮询结果 python3 ~/.hermes/event_signaler.py feedback ``` ### 支持的命令类型 | 命令 | 说明 | 参数示例 | |------|------|---------| | `search_memory` | 跨会话搜索 WorkBuddy 记忆 | `{"keyword": "辽望"}` | | `sync_session` | 将会话摘要写入 WorkBuddy 记忆 | `{"topic":"公众号","summary":"完成3月计划"}` | | `create_task` | 在滴答清单创建任务 | `{"title": "周五前完成"}` | | `complete_task` | 标记滴答任务完成 | `{"task_id": "xxx"}` | | `list_tasks` | 列出滴答清单任务 | `{}` | | `ack` | 确认信号已处理 | `{"signal_id": "xxx"}` | | `echo` | 回显测试 | `{"message": "ping"}` | ## v2.0 事件驱动命令 ```bash # ── 信号事件（实时通知）────────────────────────────── # 从 Hermes 发射信号（Hermes Agent 调用） python3 ~/.hermes/event_signaler.py emit task_done "完成XXX项目" python3 ~/.hermes/event_signaler.py emit learning_updated "更新了学习材料" # 从 Hermes 轮询来自 WorkBuddy 的信号 python3 ~/.hermes/event_signaler.py poll # 从 Hermes 确认收到某信号 python3 ~/.hermes/event_signaler.py ack # 查看信号统计 python3 ~/.hermes/event_signaler.py stats # ── WorkBuddy 侧事件监听（守护进程）──────────────── # 启动事件监听（FSEvents + 自适应轮询） python3 ~/.workbuddy/skills/hermes-memory-bridge/event_watcher.py # 强制使用轮询模式（禁用 FSEvents） python3 ~/.workbuddy/skills/hermes-memory-bridge/event_watcher.py --poll-only # ── 自适应轮询器（独立使用）───────────────────────── python3 ~/.workbuddy/skills/hermes-memory-bridge/adaptive_poller.py # ── 队列消息 ─────────────────────────────────────── python3 -c " from communication_queue import enqueue, dequeue, ack # 放入队列 qid = enqueue('wb_to_hm', {'action': 'sync', 'data': 'something'}) # 取出（Hermes 侧） msg = dequeue('wb_to_hm') # 确认完成 ack(qid, 'wb_to_hm') " ``` ## v1.x 原有命令（保持兼容） > **⚠️ 重要**：`bridge.py` 位于 `~/.workbuddy/skills/hermes-memory-bridge/bridge.py`， > 不在 memory 工作目录中。调用时须使用完整路径： > `python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge.py ` ```bash # 同步 WorkBuddy 工作到 Hermes 记忆 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py sync_to_hermes "完成了XXX" [tags...] # 拉取 Hermes 近 N 天上下文 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge.py sync_from_hermes [days] # 跨 WorkBuddy + Hermes 全文搜索 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py search [days] # 查看桥接状态 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py status # Hermes 使用统计 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge.py stats [days] # 列出最近会话 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py sessions [days] [limit] # 读取 Hermes 记忆 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py memory [memory|user] # 查看桥接事件历史 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py events [limit] # 同步学习材料 python3 ~/.workbuddy/skills/hermes-memory-bridge/bridge_enhanced.py learning_sync ``` ## 信号事件机制（v2.0 核心） ### 信号类型 | type | 说明 | 典型来源 | |------|------|---------| | `task_done` | WorkBuddy 完成任务 | WorkBuddy | | `sync` | 同步操作完成 | WorkBuddy/Hermes | | `config_change` | 配置变更 | WorkBuddy/Hermes | | `learning_updated` | 学习材料更新 | Hermes | | `ack` | 信号确认 | 接收方 | ### 事件流 ``` WorkBuddy 完成任务 → signal_event('task_done', {...}) # 写入 sig_task_done_xxx.json → event_watcher 检测到文件变化 # FSEvents 即时 / 轮询最多 60s → 触发回调（如有配置） → Hermes poll 轮询到信号 → Hermes ack_signal(sid) → WorkBuddy wait_for_ack(sid) ← 可选阻塞等待 ``` ### ACK 机制 - 信号发射后，接收方可随时 `ack` 确认 - `wait_for_ack(sid, timeout_sec=300)` 支持阻塞等待确认 - 超过 5 分钟未确认自动超时 - 信号文件保留 6 小时后自动清理 ## 自适应轮询策略 | 状态 | 间隔 | |------|------| | 有活动后 | 最短 60s | | 无活动后 | 每轮乘 1.15 倍 | | 出错时 | 强制回到 60s | | 最长间隔 | 300s（5分钟） | ## 守护进程部署（macOS） ```bash # 一键安装 bash ~/.workbuddy/skills/hermes-memory-bridge/install_v2.sh # 启动守护进程 launchctl load ~/Library/LaunchAgents/com.workbuddy.hermes-watcher.plist # 查看日志 tail -f /tmp/hermes-watcher.log ``` ## API 参考（Python 模块） ```python # ── 事件驱动 v2.0 ─────────────────────────────── from communication_queue import ( enqueue, # 放入消息队列 dequeue, # 取出消息（阻塞） ack, # 确认消息 signal_event, # 发射信号（实时通知） signal_ack, # 确认收到信号 wait_for_ack, # 等待信号确认（阻塞） list_pending_signals, # 列出待确认信号 get_queue_stats, # 队列统计 ) # 发射信号 sid = signal_event( signal_type="task_done", data={"summary": "完成XXX", "project": "ABC"}, source="WorkBuddy", priority="normal", ) # 等待 Hermes 确认（最多等 5 分钟） result = wait_for_ack(sid, timeout_sec=300) # ── v1.x 兼容 ─────────────────────────────────── from sync import ( sync_workbuddy_to_hermes, sync_hermes_to_workbuddy_context, search_both_memories, read_bridge_status, ) from memory_writer import ( append_hermes_memory, write_bridge_event, write_shared_log, read_shared_events, ) from queries import ( get_recent_sessions, get_session_stats, read_hermes_memory, search_messages, ) ``` ## 错误处理所有命令均有健壮错误处理： - **文件不存在**：优雅降级，返回空列表/空结果，不抛异常 - **权限不足**：记录警告日志，返回友好错误信息 - **数据库错误**：自动降级（如 FTS5 不可用 → 降级为 LIKE 搜索） - **同步部分失败**：返回 `status: partial`，仍输出成功写入的部分 - **FSEvents 不可用**：自动降级为自适应轮询，无需手动干预 **返回值约定**： - `exit code 0` = 成功 - `exit code 1` = 失败（含参数错误、全部写入失败等） ## 触发词 - "把今天的工作同步到 Hermes" - "搜索一下 Hermes 里关于 MCP 的记录" - "Hermes 最近有多少会话" - "两个系统的记忆里有没有关于 deepseek 的内容" - "查看 bridge 状态" - **"信号事件" / "信号通知"**（v2.0） - **"等待确认" / "wait_for_ack"**（v2.0） - **"协同进化" / "co_evolution" / "自动学习"**（协同进化流程） - **"sync_from_hermes" / "同步账本" / "进化状态"**（协同进化状态查询） - **"bridge统计" / "hermes统计" / "bridge stats"**（使用统计） FILE:adaptive_poller.py """ hermes-memory-bridge / adaptive_poller.py 自适应轮询器 — 根据活跃程度动态调整轮询间隔策略： - 启动间隔 60s（检测到活动后缩短） - 无活动时，每轮乘以 decay_factor (1.15)，最多延长到 max_interval (300s) - 有活动时，乘以 growth_factor (0.7)，最低不低于 min_interval (60s) - 出错时强制回到 min_interval """ from __future__ import annotations import json import logging import os import sys import threading import time from datetime import datetime from pathlib import Path from typing import Any, Optional SKILL_DIR = Path(__file__).parent try: from config import SHARED_DIR, _get_logger except ImportError: SHARED_DIR = Path.home() / ".hermes" / "shared" import logging logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) _get_logger = lambda n: logging.getLogger(n) logger = _get_logger("adaptive_poller") # ─── 配置 ────────────────────────────────────────────────────────── MIN_INTERVAL = 60.0 # 最短轮询间隔（秒） MAX_INTERVAL = 300.0 # 最长轮询间隔（秒） DECAY_FACTOR = 1.15 # 无活动时乘以这个因子 GROWTH_FACTOR = 0.7 # 有活动时乘以这个因子 SIGNAL_DIR = SHARED_DIR / "signals" QUEUE_DIR = SHARED_DIR / "queue" def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _poll_for_signals() -> list[dict]: """轮询新信号（来自 WorkBuddy 的信号）""" if not SIGNAL_DIR.exists(): return [] signals = [] try: for fpath in sorted(SIGNAL_DIR.glob("sig_*.json"), key=lambda f: f.stat().st_mtime, reverse=True): sig = _safe_read(fpath) if sig is None: continue if sig.get("status") != "pending": continue if sig.get("source") != "WorkBuddy": continue signals.append(sig) if len(signals) >= 5: # 每次最多处理 5 条 break except PermissionError: pass return signals def _poll_for_queue(direction: str = "wb_to_hm") -> list[dict]: """轮询新队列消息（来自 WorkBuddy 的队列）""" if not QUEUE_DIR.exists(): return [] prefix = "wb2hm" items = [] try: for fpath in sorted(QUEUE_DIR.glob(f"{prefix}_*.json"), key=lambda f: f.stat().st_mtime): item = _safe_read(fpath) if item is None: continue if item.get("status") not in ("pending", "processing"): continue items.append(item) if len(items) >= 5: break except PermissionError: pass return items def _adjust_interval(current: float, has_activity: bool) -> float: """根据是否有活动调整间隔""" if has_activity: new_interval = current * GROWTH_FACTOR return max(MIN_INTERVAL, new_interval) else: new_interval = current * DECAY_FACTOR return min(MAX_INTERVAL, new_interval) class AdaptivePoller: """ 自适应轮询器。使用示例： poller = AdaptivePoller() poller.start() # 注册回调（每收到一条新信号调用一次） poller.on_signal(lambda sig: print(f"收到信号: {sig['type']}")) # 停止 poller.stop() """ def __init__(self): self._interval = 60.0 self._running = False self._thread: Optional[threading.Thread] = None self._callbacks: list[Callable[[dict], None]] = [] self._lock = threading.Lock() def on_signal(self, cb: Callable[[dict], None]) -> "AdaptivePoller": """注册信号回调（支持链式调用）""" with self._lock: self._callbacks.append(cb) return self def start(self) -> None: if self._running: logger.warning("轮询器已在运行，忽略重复启动") return self._running = True self._thread = threading.Thread(target=self._run, daemon=True, name="adaptive-poller") self._thread.start() logger.info(f"自适应轮询已启动（初始间隔 {self._interval}s）") def stop(self) -> None: self._running = False if self._thread: self._thread.join(timeout=10) logger.info("自适应轮询已停止") def _run(self) -> None: while self._running: try: signals = _poll_for_signals() if signals: logger.info(f"[Poller] 检测到 {len(signals)} 条新信号") with self._lock: cbs = list(self._callbacks) for sig in signals: for cb in cbs: try: cb(sig) except Exception as e: logger.error(f"回调异常: {e}") self._interval = _adjust_interval(self._interval, has_activity=True) else: self._interval = _adjust_interval(self._interval, has_activity=False) logger.debug(f"[Poller] 下次轮询间隔: {self._interval:.1f}s") except Exception as e: logger.error(f"轮询异常: {e}") self._interval = MIN_INTERVAL time.sleep(self._interval) @property def current_interval(self) -> float: return self._interval def run_cli() -> None: """CLI 模式：直接轮询并打印结果""" logger.info("启动自适应轮询器（CLI 模式）...") poller = AdaptivePoller() poller.on_signal( lambda sig: print(f"📬 [{sig.get('id')}] {sig.get('type')} | {sig.get('summary', '')}") ) poller.start() try: while True: time.sleep(60) except KeyboardInterrupt: poller.stop() print("已停止") if __name__ == "__main__": run_cli() FILE:agent_communicator.py #!/usr/bin/env python3 """ agent_communicator.py 两个 AI 之间通过共享文件直接对话的通信工具。共享文件（~/.hermes/shared/agent_messages.json）: wb_to_hermes: WorkBuddy 发给 Hermes 的消息（待处理） hermes_to_wb: Hermes 回复 WorkBuddy 的消息（待处理）格式： { "wb_to_hermes": [ {"id": "uuid", "content": "...", "sent_at": "ISO8601", "status": "pending|read|replied"} ], "hermes_to_wb": [ {"id": "uuid", "content": "...", "sent_at": "ISO8601", "status": "pending|read"} ] } """ import json import os import sys import uuid from datetime import datetime, timezone SHARED_FILE = os.path.expanduser("~/.hermes/shared/agent_messages.json") def _ensure_file(): if not os.path.exists(SHARED_FILE): os.makedirs(os.path.dirname(SHARED_FILE), exist_ok=True) with open(SHARED_FILE, "w") as f: json.dump({"wb_to_hermes": [], "hermes_to_wb": []}, f, ensure_ascii=False, indent=2) def _read(): _ensure_file() with open(SHARED_FILE) as f: return json.load(f) def _write(data): _ensure_file() with open(SHARED_FILE, "w") as f: json.dump(data, f, ensure_ascii=False, indent=2) # ── WorkBuddy 端 API ────────────────────────────────────── def send_to_hermes(content: str) -> str: """WorkBuddy 发送消息给 Hermes，返回消息 ID""" data = _read() msg = { "id": str(uuid.uuid4())[:8], "content": content, "sent_at": datetime.now(timezone.utc).isoformat(), "status": "pending" } data["wb_to_hermes"].append(msg) _write(data) print(f"[agent_comm] 📤 WB→Hermes [{msg['id']}]: {content[:60]}...") return msg["id"] def read_from_hermes() -> list: """WorkBuddy 读取 Hermes 的所有回复，标记为已读""" data = _read() replies = [m for m in data["hermes_to_wb"] if m["status"] == "pending"] for r in replies: r["status"] = "read" if replies: _write(data) for r in replies: print(f"[agent_comm] 📥 Hermes→WB [{r['id']}]: {r['content'][:80]}...") return replies def get_pending_to_hermes() -> list: """查看发给 Hermes 但 Hermes 还未读的消息""" data = _read() return [m for m in data["wb_to_hermes"] if m["status"] == "pending"] # ── Hermes 端 API ───────────────────────────────────────── def read_from_wb() -> list: """Hermes 读取 WorkBuddy 发来的所有消息，标记为已读""" data = _read() msgs = [m for m in data["wb_to_hermes"] if m["status"] == "pending"] for m in msgs: m["status"] = "read" if msgs: _write(data) for m in msgs: print(f"[agent_comm] 📥 WB→Hermes [{m['id']}]: {m['content'][:80]}...") return msgs def reply_to_wb(content: str) -> str: """Hermes 回复 WorkBuddy""" data = _read() msg = { "id": str(uuid.uuid4())[:8], "content": content, "sent_at": datetime.now(timezone.utc).isoformat(), "status": "pending" } data["hermes_to_wb"].append(msg) _write(data) print(f"[agent_comm] 📤 Hermes→WB [{msg['id']}]: {content[:60]}...") return msg["id"] def get_pending_from_wb() -> list: """Hermes 查看还有哪些消息等待处理""" data = _read() return [m for m in data["wb_to_hermes"] if m["status"] == "pending"] # ── CLI 入口 ────────────────────────────────────────────── if __name__ == "__main__": cmd = sys.argv[1] if len(sys.argv) > 1 else "help" if cmd == "send": # python3 agent_communicator.py send "消息内容" content = sys.argv[2] if len(sys.argv) > 2 else input("消息内容: ") send_to_hermes(content) elif cmd == "read": # python3 agent_communicator.py read replies = read_from_hermes() if not replies: print("[agent_comm] 没有新消息") for r in replies: print(f"\n=== Hermes 回复 [{r['id']}] ===") print(r["content"]) elif cmd == "pending": msgs = get_pending_to_hermes() if msgs: print(f"[agent_comm] 还有 {len(msgs)} 条消息等待 Hermes 处理") for m in msgs: print(f" [{m['id']}] {m['content'][:60]}") else: print("[agent_comm] 无待处理消息") elif cmd == "hermes-read": msgs = read_from_wb() if not msgs: print("[agent_comm] 没有新消息") for m in msgs: print(f"\n=== WB 发来 [{m['id']}] ===") print(m["content"]) elif cmd == "hermes-reply": content = sys.argv[2] if len(sys.argv) > 2 else input("回复内容: ") reply_to_wb(content) elif cmd == "hermes-pending": msgs = get_pending_from_wb() if msgs: print(f"[agent_comm] 还有 {len(msgs)} 条消息等待处理") else: print("[agent_comm] 无待处理消息") else: print("用法:") print(" # WorkBuddy 端") print(" python3 agent_communicator.py send <消息> # 发消息给 Hermes") print(" python3 agent_communicator.py read # 读取 Hermes 回复") print(" python3 agent_communicator.py pending # 查看待处理消息") print() print(" # Hermes 端") print(" python3 agent_communicator.py hermes-read # 读取 WorkBuddy 消息") print(" python3 agent_communicator.py hermes-reply <回复> # 回复 WorkBuddy") print(" python3 agent_communicator.py hermes-pending # 查看待处理消息") FILE:auto_poller.py # DEPRECATED - 此文件已废弃 # # 自适应轮询功能已迁移到 adaptive_poller.py # auto_poller.py 在 v2.0 中不再使用，请使用： # # from adaptive_poller import AdaptivePoller # poller = AdaptivePoller() # poller.on_signal(lambda sig: print(f"收到: {sig['type']}")) # poller.start() # # 如有部署脚本引用本文件，请更新为 adaptive_poller.py FILE:bridge.py #!/usr/bin/env python3 """ hermes-memory-bridge / bridge.py WorkBuddy 与 Hermes Agent 双向记忆互通主引擎用法（作为 Skill 被 WorkBuddy 调用）： python3 bridge.py [args...] 命令： sync_to_hermes

[tags...] sync_from_hermes [days] search [days] status stats [days] sessions [days] [limit] memory [memory|user] events [limit] help """ from __future__ import annotations import json import logging import sys import textwrap from datetime import datetime from pathlib import Path from config import _get_logger, logger logger = _get_logger("bridge") def cmd_sync_to_hermes(args: list) -> bool: """同步 WorkBuddy 工作到 Hermes，返回是否成功""" from sync import sync_workbuddy_to_hermes summary = args[0] if args else "" work_type = args[1] if len(args) > 1 else "task" tags = args[2:] if len(args) > 2 else [] if not summary: print("用法: sync_to_hermes

[work_type] [tags...]", file=sys.stderr) return False try: result = sync_workbuddy_to_hermes(summary, work_type, tags) except Exception as e: logger.error(f"同步失败: {e}") print(f"❌ 同步失败: {e}") return False if result["status"] in ("synced", "partial"): icon = "✅" if result["status"] == "synced" else "⚠️" print(f"{icon} 已同步到 Hermes（{result['status']}）") if result.get("entry"): print(f" 记忆: {result['entry'][:80]}...") if result.get("log_path"): print(f" 日志: {result['log_path']}") return True else: print(f"❌ 同步失败（全部写入操作均未成功）") return False def cmd_sync_from_hermes(args: list) -> bool: """拉取 Hermes 最新上下文到 WorkBuddy""" from sync import sync_hermes_to_workbuddy_context days = int(args[0]) if args else 7 try: result = sync_hermes_to_workbuddy_context(days=days) print(result["summary_text"]) return True except Exception as e: logger.error(f"拉取 Hermes 上下文失败: {e}") print(f"❌ 拉取失败: {e}") return False def cmd_search(args: list) -> bool: """跨 WorkBuddy + Hermes 全文搜索""" from sync import search_both_memories keyword = args[0] if args else "" days = int(args[1]) if len(args) > 1 else 30 if not keyword: print("用法: search [days]", file=sys.stderr) return False try: results = search_both_memories(keyword, days=days) except Exception as e: logger.error(f"搜索失败: {e}") print(f"❌ 搜索失败: {e}") return False _print_search_results(results, keyword) return True def cmd_status(args: list) -> bool: """桥接状态总览""" from sync import read_bridge_status try: status = read_bridge_status() except Exception as e: logger.error(f"读取状态失败: {e}") print(f"❌ 读取状态失败: {e}") return False print("=" * 48) print(" Hermes-Memory-Bridge 状态总览") print("=" * 48) print(f" 数据库: {'✅ 存在' if status['db_exists'] else '❌ 不存在'}") print(f" WorkBuddy 记忆: {status.get('workbuddy_memory_dir') or '（未找到）'}") print(f" 记忆文件: {', '.join(status['hermes_memory_files']) or '（暂无）'}") print(f" 共用文件: {', '.join(status['shared_files']) or '（暂无）'}") print(f" 近期事件: {len(status['recent_events'])} 条") print("=" * 48) for ev in status["recent_events"][-5:]: ts = ev.get("timestamp", "")[:16] etype = ev.get("type", "") # 去除 timestamp 字段避免重复显示 ev_clean = {k: v for k, v in ev.items() if k != "timestamp"} info = json.dumps(ev_clean, ensure_ascii=False)[:60] print(f" [{ts}] {etype}: {info}") return True def cmd_stats(args: list) -> bool: """Hermes 使用统计""" from queries import get_session_stats days = int(args[0]) if args else 30 try: stats = get_session_stats(days=days) except Exception as e: logger.error(f"获取统计失败: {e}") print(f"❌ 获取统计失败: {e}") return False print(f"📊 Hermes 近 {days} 天统计") print(f" 会话数: {stats.get('total_sessions', 0)}") print(f" 消息数: {stats.get('total_messages', 0)}") print(f" Token数: {stats.get('total_tokens', 0):,}") print(f" 估算费用: ${stats.get('estimated_cost_usd', 0):.4f}") return True def cmd_sessions(args: list) -> bool: """列出最近会话""" from queries import get_recent_sessions days = int(args[0]) if args else 7 limit = int(args[1]) if len(args) > 1 else 10 try: sessions = get_recent_sessions(days=days, limit=limit) except Exception as e: logger.error(f"获取会话列表失败: {e}") print(f"❌ 获取会话失败: {e}") return False if not sessions: print(f"近 {days} 天无会话记录") return True print(f"📋 近 {days} 天会话（共 {len(sessions)} 条）") for s in sessions: try: ts = datetime.fromtimestamp(s["started_at"]).strftime("%m-%d %H:%M") except (ValueError, KeyError, TypeError): ts = "??-?? ??:??" title = s.get("title") or s.get("source") or "无标题" msgs = s.get("message_count", 0) cost = s.get("estimated_cost_usd") or 0 print(f" [{ts}] {title} | {msgs}条消息 | ${cost:.4f}") return True def cmd_memory(args: list) -> bool: """读取 Hermes 记忆文件""" from queries import read_hermes_memory target = args[0] if args else "memory" try: mem = read_hermes_memory() except Exception as e: logger.error(f"读取 Hermes 记忆失败: {e}") print(f"❌ 读取失败: {e}") return False key = "MEMORY.md" if target == "memory" else "USER.md" data = mem.get(key, {}) entries = data.get("entries", []) print(f"🧠 Hermes {key}（共 {len(entries)} 条记忆）") print("-" * 48) if not entries: print(" （记忆为空）") else: for i, entry in enumerate(entries, 1): print(f" [{i}] {entry[:300]}") print() return True def cmd_events(args: list) -> bool: """读取桥接事件历史""" from memory_writer import read_shared_events limit = int(args[0]) if args else 20 try: events = read_shared_events(limit=limit) except Exception as e: logger.error(f"读取事件历史失败: {e}") print(f"❌ 读取失败: {e}") return False print(f"🔗 桥接事件历史（共 {len(events)} 条）") for ev in events[-limit:]: ts = ev.get("timestamp", "")[:16] etype = ev.get("type", "") ev_clean = {k: v for k, v in ev.items() if k != "timestamp"} info = json.dumps(ev_clean, ensure_ascii=False)[:100] print(f" [{ts}] {etype}: {info}") return True def _print_search_results(results: dict, keyword: str) -> None: print(f"\n🔍 搜索「{keyword}」\n") hermes = results.get("hermes", []) workbuddy = results.get("workbuddy", []) if hermes: print(f"**Hermes 会话**（{len(hermes)} 条）") for r in hermes[:5]: try: ts = ( datetime.fromtimestamp(r["timestamp"]).strftime("%m-%d %H:%M") if isinstance(r.get("timestamp"), float) else "?" ) except (ValueError, TypeError): ts = "?" role = r.get("role", "") content = (r.get("content") or "")[:120] print(f" [{ts}] [{role}]: {content}...") else: print("**Hermes 会话**：无匹配结果") print() if workbuddy: print(f"**WorkBuddy 记忆文件**（{len(workbuddy)} 条）") for r in workbuddy[:5]: print(f" [{r['file']}:{r['line']}] {r['snippet']}") else: print("**WorkBuddy 记忆文件**：无匹配结果") def main() -> int: """返回退出码：0=成功，1=失败""" if len(sys.argv) < 2 or sys.argv[1] == "help": print(textwrap.dedent(""" hermes-memory-bridge v1.1.0 用法: python3 bridge.py [args...] 命令: sync_to_hermes

[work_type] [tags...] sync_from_hermes [days] search [days] status stats [days] sessions [days] [limit] memory [memory|user] events [limit] 环境变量: HERMES_HOME Hermes 根目录（默认 ~/.hermes） WORKBUDDY_HOME WorkBuddy 根目录（默认 ~/WorkBuddy） BRIDGE_LOG_LEVEL DEBUG|INFO|WARNING|ERROR（默认 INFO） """.strip())) return 0 cmd = sys.argv[1] args = sys.argv[2:] commands: dict[str, tuple] = { "sync_to_hermes": (cmd_sync_to_hermes, "同步工作到 Hermes"), "sync_from_hermes": (cmd_sync_from_hermes, "拉取 Hermes 上下文"), "search": (cmd_search, "跨系统搜索"), "status": (cmd_status, "桥接状态"), "stats": (cmd_stats, "Hermes 统计"), "sessions": (cmd_sessions, "会话列表"), "memory": (cmd_memory, "读取记忆"), "events": (cmd_events, "事件历史"), } if cmd not in commands: print(f"未知命令: {cmd}", file=sys.stderr) print(f"可用命令: {', '.join(commands)}", file=sys.stderr) return 1 handler, description = commands[cmd] logger.info(f"执行命令: {cmd} {args}") try: success = handler(args) except Exception as e: logger.exception(f"命令 {cmd} 异常: {e}") print(f"❌ 命令执行异常: {e}", file=sys.stderr) return 1 return 0 if success else 1 if __name__ == "__main__": sys.exit(main()) FILE:bridge_enhanced.py #!/usr/bin/env python3 """ hermes-memory-bridge / bridge.py - 增强版添加学习材料同步功能 """ import json import sys import textwrap from datetime import datetime from config import HERMES_DB from memory_writer import read_shared_events as read_bridge_events, read_workbuddy_log from queries import ( get_recent_sessions, get_session_messages, get_session_stats, read_hermes_memory, search_fts, search_messages, ) from sync import ( read_bridge_status, search_both_memories, sync_hermes_to_workbuddy_context, sync_workbuddy_to_hermes, ) # 导入学习材料同步模块 try: from hermes_learning_sync import sync_learning_materials, get_learning_stats HAS_LEARNING_SYNC = True except ImportError: HAS_LEARNING_SYNC = False def cmd_sync_to_hermes(args: list) -> None: """同步 WorkBuddy 工作到 Hermes""" summary = args[0] if args else "" work_type = args[1] if len(args) > 1 else "task" tags = args[2:] if len(args) > 2 else [] if not summary: print("❌ 需要提供工作摘要") print("用法: sync_to_hermes

[tags...]") return result = sync_workbuddy_to_hermes(summary, work_type, tags) print("✅ 已同步到 Hermes") print(f" 记忆条目: {result['entry'][:80]}...") print(f" 日志路径: {result['log_path']}") def cmd_sync_from_hermes(args: list) -> None: """从 Hermes 同步上下文到 WorkBuddy""" days = int(args[0]) if args else 7 result = sync_hermes_to_workbuddy_context(days) print(result["summary_text"]) def cmd_search(args: list) -> None: """跨系统搜索""" if not args: print("❌ 需要提供搜索关键词") print("用法: search [days]") return keyword = args[0] days = int(args[1]) if len(args) > 1 else 30 results = search_both_memories(keyword, days) from sync import _format_results_for_user print(_format_results_for_user(results, keyword)) def cmd_status(args: list) -> None: """显示桥接状态""" status = read_bridge_status() print("=" * 48) print(" Hermes-Memory-Bridge 状态总览") print("=" * 48) print(f" 数据库: {'✅ 存在' if status['db_exists'] else '❌ 缺失'}") print(f" 记忆文件: {', '.join(status['hermes_memory_files']) or '无'}") print(f" 共用文件: {', '.join(status['shared_files'])[:40]}...") print(f" 近期事件: {len(status['recent_events'])} 条") print("=" * 48) for ev in status["recent_events"][:3]: ts = ev.get("timestamp", "").split("T")[0] ev_type = ev.get("type", "unknown") summary = ev.get("summary", "")[:40] print(f" [{ts}] {ev_type}: {summary}...") def cmd_stats(args: list) -> None: """显示统计信息""" days = int(args[0]) if args else 7 stats = get_session_stats(days) print(f"📊 Hermes 近 {days} 天统计") print(f" 会话数: {stats.get('total_sessions', 0)}") print(f" 消息数: {stats.get('total_messages', 0)}") print(f" Token数: {stats.get('total_tokens', 0):,}") print(f" 估算费用: ${stats.get('estimated_cost_usd', 0):.4f}") def cmd_sessions(args: list) -> None: """显示会话列表""" days = int(args[0]) if args else 7 limit = int(args[1]) if len(args) > 1 else 10 sessions = get_recent_sessions(days, limit) print(f"📋 近 {days} 天会话（共 {len(sessions)} 条）") for s in sessions: ts = datetime.fromtimestamp(s["started_at"]).strftime("%m-%d %H:%M") title = s.get("title") or s["source"] or "无标题" msg_count = s.get("message_count", 0) cost = s.get("estimated_cost_usd", 0) print(f" [{ts}] {title} | {msg_count}条消息 | ${cost:.4f}") def cmd_memory(args: list) -> None: """读取 Hermes 记忆""" target = args[0] if args else "memory" mem = read_hermes_memory() if target == "memory": entries = mem.get("MEMORY.md", {}).get("entries", []) print(f"🧠 Hermes MEMORY.md（共 {len(entries)} 条记忆）") print("-" * 48) for i, entry in enumerate(entries[-10:], 1): print(f" [{i}] {entry[:80]}...") elif target == "user": entries = mem.get("USER.md", {}).get("entries", []) print(f"👤 Hermes USER.md（共 {len(entries)} 条用户记忆）") print("-" * 48) for i, entry in enumerate(entries[-10:], 1): print(f" [{i}] {entry[:80]}...") else: print("❌ 未知目标，使用 'memory' 或 'user'") def cmd_events(args: list) -> None: """显示桥接事件""" limit = int(args[0]) if args else 10 events = read_bridge_events(limit=limit) print(f"🔗 桥接事件历史（共 {len(events)} 条）") for ev in events: ts = ev.get("timestamp", "").split("T")[1][:8] ev_type = ev.get("type", "unknown") summary = ev.get("summary", "")[:60] print(f" [{ts}] {ev_type}: {summary}...") def cmd_learning_sync(args: list) -> None: """同步学习材料到 WorkBuddy""" if not HAS_LEARNING_SYNC: print("❌ 学习材料同步模块未找到") print("请确保 hermes_learning_sync.py 在技能目录中") return print("🔄 开始同步 Hermes 学习材料到 WorkBuddy...") success = sync_learning_materials() if success: print("✅ 学习材料同步完成！") stats = get_learning_stats() print(f"📊 同步统计:") print(f" 学习材料: {stats.get('materials_count', 0)} 类") print(f" 记忆条目: {stats.get('summary_entries', 0)} 条") print(f" 关键学习点: {stats.get('key_insights', 0)} 个") else: print("❌ 学习材料同步失败") def cmd_learning_stats(args: list) -> None: """显示学习材料统计""" if not HAS_LEARNING_SYNC: print("❌ 学习材料同步模块未找到") return stats = get_learning_stats() print("📚 Hermes 学习材料统计") print("=" * 40) print(f"最后同步: {stats.get('last_sync_time', '从未同步')}") print(f"学习材料: {stats.get('materials_count', 0)} 类") print(f"记忆条目: {stats.get('summary_entries', 0)} 条") print(f"关键学习点: {stats.get('key_insights', 0)} 个") print(f"同步状态: {stats.get('status', 'unknown')}") def cmd_help() -> None: """显示帮助信息""" help_text = """ hermes-memory-bridge 命令列表: 基础命令: sync_to_hermes

[tags...] sync_from_hermes [days] search [days] status stats [days] sessions [days] [limit] memory [memory|user] events [limit] 学习材料命令 (新增): learning_sync 同步 Hermes 学习材料到 WorkBuddy learning_stats 显示学习材料统计帮助: help 显示此帮助信息示例: python3 bridge.py sync_to_hermes "完成XXX任务" work "tag1,tag2" python3 bridge.py search "MCP" 7 python3 bridge.py learning_sync python3 bridge.py status """ print(textwrap.dedent(help_text).strip()) def main(): if len(sys.argv) < 2: cmd_help() return command = sys.argv[1] args = sys.argv[2:] commands = { "sync_to_hermes": cmd_sync_to_hermes, "sync_from_hermes": cmd_sync_from_hermes, "search": cmd_search, "status": cmd_status, "stats": cmd_stats, "sessions": cmd_sessions, "memory": cmd_memory, "events": cmd_events, "learning_sync": cmd_learning_sync, "learning_stats": cmd_learning_stats, "help": cmd_help, } if command in commands: try: commands[command](args) except Exception as e: print(f"❌ 命令执行错误: {e}") import traceback traceback.print_exc() else: print(f"❌ 未知命令: {command}") cmd_help() if __name__ == "__main__": main() FILE:bridge_enhanced_v2.py #!/usr/bin/env python3 """ hermes-memory-bridge / bridge_enhanced.py - v1.1.0 兼容版添加学习材料同步功能，与 v1.1.0 改进完全兼容 """ import sys import textwrap from datetime import datetime from config import _get_logger, logger logger = _get_logger("bridge_enhanced") # 导入基础模块 try: from memory_writer import read_shared_events as read_bridge_events, read_workbuddy_log from queries import ( get_recent_sessions, get_session_messages, get_session_stats, read_hermes_memory, search_fts, search_messages, ) from sync import ( read_bridge_status, search_both_memories, sync_hermes_to_workbuddy_context, sync_workbuddy_to_hermes, ) HAS_BASE_MODULES = True except ImportError as e: logger.error(f"导入基础模块失败: {e}") HAS_BASE_MODULES = False # 导入学习材料同步模块 try: from hermes_learning_sync import sync_learning_materials, get_learning_stats HAS_LEARNING_SYNC = True except ImportError: logger.warning("学习材料同步模块未找到，相关功能将不可用") HAS_LEARNING_SYNC = False def cmd_sync_to_hermes(args: list) -> bool: """同步 WorkBuddy 工作到 Hermes，返回是否成功""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False summary = args[0] if args else "" work_type = args[1] if len(args) > 1 else "task" tags = args[2:] if len(args) > 2 else [] if not summary: print("用法: sync_to_hermes

[work_type] [tags...]", file=sys.stderr) return False try: result = sync_workbuddy_to_hermes(summary, work_type, tags) except Exception as e: logger.error(f"同步失败: {e}") print(f"❌ 同步失败: {e}") return False if result["status"] in ("synced", "partial"): icon = "✅" if result["status"] == "synced" else "⚠️" print(f"{icon} 已同步到 Hermes（{result['status']}）") if result.get("entry"): print(f" 记忆: {result['entry'][:80]}...") if result.get("log_path"): print(f" 日志: {result['log_path']}") return True else: print(f"❌ 同步失败（全部写入操作均未成功）") return False def cmd_sync_from_hermes(args: list) -> bool: """拉取 Hermes 最新上下文到 WorkBuddy""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False days = int(args[0]) if args else 7 try: result = sync_hermes_to_workbuddy_context(days=days) print(result["summary_text"]) return True except Exception as e: logger.error(f"拉取 Hermes 上下文失败: {e}") print(f"❌ 拉取失败: {e}") return False def cmd_search(args: list) -> bool: """跨 WorkBuddy + Hermes 全文搜索""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False keyword = args[0] if args else "" days = int(args[1]) if len(args) > 1 else 30 if not keyword: print("用法: search [days]", file=sys.stderr) return False try: results = search_both_memories(keyword, days=days) except Exception as e: logger.error(f"搜索失败: {e}") print(f"❌ 搜索失败: {e}") return False from sync import _format_results_for_user print(_format_results_for_user(results, keyword)) return True def cmd_status(args: list) -> bool: """显示桥接状态""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False try: status = read_bridge_status() except Exception as e: logger.error(f"读取状态失败: {e}") print(f"❌ 读取状态失败: {e}") return False print("=" * 48) print(" Hermes-Memory-Bridge 状态总览 (增强版)") print("=" * 48) print(f" 数据库: {'✅ 存在' if status['db_exists'] else '❌ 缺失'}") # 显示 WorkBuddy 记忆目录（如果找到） wb_mem_dir = status.get('workbuddy_memory_dir') if wb_mem_dir: print(f" WorkBuddy 记忆: {wb_mem_dir}") print(f" 记忆文件: {', '.join(status['hermes_memory_files']) or '无'}") print(f" 共用文件: {', '.join(status['shared_files'])[:40]}...") print(f" 近期事件: {len(status['recent_events'])} 条") print("=" * 48) for ev in status["recent_events"][:3]: ts = ev.get("timestamp", "").split("T")[0] ev_type = ev.get("type", "unknown") summary = ev.get("summary", "")[:40] print(f" [{ts}] {ev_type}: {summary}...") return True def cmd_stats(args: list) -> bool: """显示统计信息""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False days = int(args[0]) if args else 7 try: stats = get_session_stats(days) except Exception as e: logger.error(f"获取统计失败: {e}") print(f"❌ 获取统计失败: {e}") return False print(f"📊 Hermes 近 {days} 天统计") print(f" 会话数: {stats.get('total_sessions', 0)}") print(f" 消息数: {stats.get('total_messages', 0)}") print(f" Token数: {stats.get('total_tokens', 0):,}") print(f" 估算费用: ${stats.get('estimated_cost_usd', 0):.4f}") return True def cmd_sessions(args: list) -> bool: """显示会话列表""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False days = int(args[0]) if args else 7 limit = int(args[1]) if len(args) > 1 else 10 try: sessions = get_recent_sessions(days, limit) except Exception as e: logger.error(f"获取会话列表失败: {e}") print(f"❌ 获取会话列表失败: {e}") return False print(f"📋 近 {days} 天会话（共 {len(sessions)} 条）") for s in sessions: ts = datetime.fromtimestamp(s["started_at"]).strftime("%m-%d %H:%M") title = s.get("title") or s["source"] or "无标题" msg_count = s.get("message_count", 0) cost = s.get("estimated_cost_usd", 0) print(f" [{ts}] {title} | {msg_count}条消息 | ${cost:.4f}") return True def cmd_memory(args: list) -> bool: """读取 Hermes 记忆""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False target = args[0] if args else "memory" try: mem = read_hermes_memory() except Exception as e: logger.error(f"读取记忆失败: {e}") print(f"❌ 读取记忆失败: {e}") return False if target == "memory": entries = mem.get("MEMORY.md", {}).get("entries", []) print(f"🧠 Hermes MEMORY.md（共 {len(entries)} 条记忆）") print("-" * 48) for i, entry in enumerate(entries[-10:], 1): print(f" [{i}] {entry[:80]}...") elif target == "user": entries = mem.get("USER.md", {}).get("entries", []) print(f"👤 Hermes USER.md（共 {len(entries)} 条用户记忆）") print("-" * 48) for i, entry in enumerate(entries[-10:], 1): print(f" [{i}] {entry[:80]}...") else: print("❌ 未知目标，使用 'memory' 或 'user'") return False return True def cmd_events(args: list) -> bool: """显示桥接事件""" if not HAS_BASE_MODULES: print("❌ 基础模块加载失败") return False limit = int(args[0]) if args else 10 try: events = read_bridge_events(limit=limit) except Exception as e: logger.error(f"读取事件失败: {e}") print(f"❌ 读取事件失败: {e}") return False print(f"🔗 桥接事件历史（共 {len(events)} 条）") for ev in events: ts = ev.get("timestamp", "").split("T")[1][:8] ev_type = ev.get("type", "unknown") summary = ev.get("summary", "")[:60] print(f" [{ts}] {ev_type}: {summary}...") return True def cmd_learning_sync(args: list) -> bool: """同步学习材料到 WorkBuddy""" if not HAS_LEARNING_SYNC: print("❌ 学习材料同步模块未找到") print("请确保 hermes_learning_sync.py 在技能目录中") return False print("🔄 开始同步 Hermes 学习材料到 WorkBuddy...") try: success = sync_learning_materials() except Exception as e: logger.error(f"学习材料同步失败: {e}") print(f"❌ 学习材料同步失败: {e}") return False if success: print("✅ 学习材料同步完成！") try: stats = get_learning_stats() print(f"📊 同步统计:") print(f" 学习材料: {stats.get('materials_count', 0)} 类") print(f" 记忆条目: {stats.get('summary_entries', 0)} 条") print(f" 关键学习点: {stats.get('key_insights', 0)} 个") except Exception as e: logger.warning(f"获取统计失败: {e}") else: print("❌ 学习材料同步失败") return success def cmd_learning_stats(args: list) -> bool: """显示学习材料统计""" if not HAS_LEARNING_SYNC: print("❌ 学习材料同步模块未找到") return False try: stats = get_learning_stats() except Exception as e: logger.error(f"获取学习统计失败: {e}") print(f"❌ 获取学习统计失败: {e}") return False print("📚 Hermes 学习材料统计") print("=" * 40) print(f"最后同步: {stats.get('last_sync_time', '从未同步')}") print(f"学习材料: {stats.get('materials_count', 0)} 类") print(f"记忆条目: {stats.get('summary_entries', 0)} 条") print(f"关键学习点: {stats.get('key_insights', 0)} 个") print(f"同步状态: {stats.get('status', 'unknown')}") return True def cmd_help() -> bool: """显示帮助信息""" help_text = """ hermes-memory-bridge 增强版命令列表 (v1.1.0 兼容) 基础命令: sync_to_hermes

[tags...] sync_from_hermes [days] search [days] status stats [days] sessions [days] [limit] memory [memory|user] events [limit] 学习材料命令 (新增): learning_sync 同步 Hermes 学习材料到 WorkBuddy learning_stats 显示学习材料统计帮助: help 显示此帮助信息环境变量: HERMES_HOME Hermes 根目录（默认 ~/.hermes） WORKBUDDY_HOME WorkBuddy 根目录（默认 ~/WorkBuddy） WORKBUDDY_MEMORY_DIR 强制指定 WorkBuddy 记忆目录 BRIDGE_LOG_LEVEL 日志级别 DEBUG|INFO|WARNING|ERROR 示例: python3 bridge_enhanced.py sync_to_hermes "完成XXX任务" work "tag1,tag2" python3 bridge_enhanced.py search "MCP" 7 python3 bridge_enhanced.py learning_sync python3 bridge_enhanced.py status """ print(textwrap.dedent(help_text).strip()) return True def main(): if len(sys.argv) < 2: cmd_help() sys.exit(0) command = sys.argv[1] args = sys.argv[2:] commands = { "sync_to_hermes": cmd_sync_to_hermes, "sync_from_hermes": cmd_sync_from_hermes, "search": cmd_search, "status": cmd_status, "stats": cmd_stats, "sessions": cmd_sessions, "memory": cmd_memory, "events": cmd_events, "learning_sync": cmd_learning_sync, "learning_stats": cmd_learning_stats, "help": cmd_help, } if command in commands: try: success = commands[command](args) sys.exit(0 if success else 1) except Exception as e: logger.error(f"命令执行错误: {e}") import traceback traceback.print_exc() sys.exit(1) else: print(f"❌ 未知命令: {command}") cmd_help() sys.exit(1) if __name__ == "__main__": main() FILE:communication_queue.py """ hermes-memory-bridge / communication_queue.py 事件驱动队列 v1.2 — 支持信号事件与 ACK 确认机制新增 v1.2： - signal_event / signal_ack / wait_for_ack / list_pending_signals - 支持 Hermes ↔ WorkBuddy 之间的实时信号通知 """ from __future__ import annotations import json import logging import os import threading import time import uuid from datetime import datetime, timedelta from pathlib import Path from typing import Any, Optional from config import SHARED_DIR, _get_logger logger = _get_logger("communication_queue") # ─── 路径 ────────────────────────────────────────────────────────── QUEUE_DIR = SHARED_DIR / "queue" SIGNAL_DIR = SHARED_DIR / "signals" # ─── ACK 超时（秒）───────────────────────────────────────────────── ACK_TIMEOUT_SEC = 300 # 5 分钟超时 CLEANUP_INTERVAL_SEC = 60 # 清理线程运行间隔 MAX_QUEUE_AGE_DAYS = 7 # 超过 7 天的队列文件删除 MAX_SIGNAL_AGE_HOURS = 6 # 信号文件保留 6 小时 # ─── 初始化 ──────────────────────────────────────────────────────── def _ensure_dirs() -> bool: for d in [QUEUE_DIR, SIGNAL_DIR]: try: d.mkdir(parents=True, exist_ok=True) except PermissionError: logger.error(f"权限不足，无法创建目录 {d}") return False return True # ─── 工具函数 ────────────────────────────────────────────────────── def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _safe_write(path: Path, data: Any) -> bool: try: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return True except (PermissionError, OSError) as e: logger.error(f"写入失败 {path}: {e}") return False def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _queue_fname(prefix: str, qid: str) -> Path: return QUEUE_DIR / f"{prefix}_{qid}.json" # ─── 原有队列功能 ────────────────────────────────────────────────── def enqueue(direction: str, payload: dict) -> str | None: """ 将消息放入队列。 Args: direction: 'wb_to_hm' | 'hm_to_wb' payload: 任意 JSON 可序列化数据 Returns: 队列 ID，或 None（失败时） """ if not _ensure_dirs(): return None qid = str(uuid.uuid4())[:8] item = { "id": qid, "direction": direction, "payload": payload, "created_at": _ts(), "status": "pending", } prefix = "wb2hm" if direction == "wb_to_hm" else "hm2wb" fpath = _queue_fname(prefix, qid) if _safe_write(fpath, item): logger.debug(f"入队 [{direction}]: {qid}") return qid return None def dequeue(direction: str) -> dict | None: """ 按 FIFO 顺序取出一条待处理消息，并标记为 processing。 Returns: 消息 dict 或 None（队列为空） """ if not QUEUE_DIR.exists(): return None prefix = "wb2hm" if direction == "wb_to_hm" else "hm2wb" try: files = sorted(QUEUE_DIR.glob(f"{prefix}_*.json"), key=lambda f: f.name) except PermissionError: return None for fpath in files: item = _safe_read(fpath) if item is None: continue if item.get("status") != "pending": continue item["status"] = "processing" item["processed_at"] = _ts() _safe_write(fpath, item) logger.debug(f"出队 [{direction}]: {item['id']}") return item return None def ack(queue_id: str, direction: str) -> bool: """ 确认一条消息处理完成（幂等操作）。 Args: queue_id: 消息 ID direction: 'wb_to_hm' | 'hm_to_wb' Returns: 是否成功确认 """ prefix = "wb2hm" if direction == "wb_to_hm" else "hm2wb" fpath = _queue_fname(prefix, queue_id) item = _safe_read(fpath) if item is None: logger.warning(f"ACK 找不到消息: {queue_id}") return False item["status"] = "done" item["acked_at"] = _ts() _safe_write(fpath, item) logger.debug(f"ACK [{direction}]: {queue_id}") return True def get_queue_stats() -> dict[str, Any]: """返回队列统计信息""" stats: dict[str, Any] = {"pending": 0, "processing": 0, "done": 0, "total": 0} if not QUEUE_DIR.exists(): return stats for fpath in QUEUE_DIR.glob("*.json"): item = _safe_read(fpath) if item is None: continue status = item.get("status", "unknown") if status in stats: stats[status] += 1 stats["total"] += 1 return stats # ─── v1.2 新增：信号事件 ─────────────────────────────────────────── def signal_event( signal_type: str, data: dict, source: str = "unknown", priority: str = "normal", ) -> str | None: """ 发射一个信号事件（跨 Agent 通知）。与普通队列消息不同，信号是"立即可感知"的事件，配合 event_watcher.py 的 FSEvents 监听可实现近实时通知。 Args: signal_type: 'task_done' | 'sync' | 'config_change' | 'ack' | 自定义 data: 事件数据 source: 'WorkBuddy' | 'Hermes' priority: 'low' | 'normal' | 'high' Returns: 信号 ID 或 None """ if not _ensure_dirs(): return None sid = str(uuid.uuid4())[:12] signal = { "id": sid, "type": signal_type, "source": source, "priority": priority, "data": data, "created_at": _ts(), "status": "pending", "ack_id": None, } fname = SIGNAL_DIR / f"sig_{signal_type}_{sid}.json" if _safe_write(fname, signal): logger.info(f"信号发射 [{source}→{signal_type}]: {sid}") return sid return None def signal_ack(signal_id: str, from_source: str) -> bool: """ 确认收到某信号。用于接收方告知发送方"已处理"。实际上是在对应信号文件中写入 ack 记录。 Args: signal_id: 信号 ID（不含前缀） from_source: 'WorkBuddy' | 'Hermes' """ if not SIGNAL_DIR.exists(): return False for fpath in SIGNAL_DIR.glob("sig_*.json"): sig = _safe_read(fpath) if sig is None: continue # 通过文件名中的 id 部分匹配 if sig.get("id") != signal_id: continue sig["status"] = "acknowledged" sig["acked_by"] = from_source sig["acked_at"] = _ts() _safe_write(fpath, sig) logger.debug(f"信号 ACK [{from_source}]: {signal_id}") return True logger.warning(f"signal_ack 找不到信号: {signal_id}") return False def wait_for_ack( signal_id: str, timeout_sec: float = ACK_TIMEOUT_SEC, poll_interval: float = 1.0, ) -> dict | None: """ 等待某个信号被对方 ACK（阻塞轮询）。适用于 WorkBuddy 主动发射信号后，等待 Hermes 处理完成的场景。 Args: signal_id: 信号 ID timeout_sec: 超时秒数 poll_interval: 轮询间隔（秒） Returns: 信号 dict（已确认）或 None（超时） """ deadline = time.time() + timeout_sec while time.time() < deadline: if not SIGNAL_DIR.exists(): time.sleep(poll_interval) continue for fpath in SIGNAL_DIR.glob("sig_*.json"): sig = _safe_read(fpath) if sig is None: continue if sig.get("id") != signal_id: continue if sig.get("status") == "acknowledged": logger.debug(f"收到信号 ACK: {signal_id}") return sig # pending 或其他状态，继续等待 break time.sleep(poll_interval) logger.warning(f"等待信号 ACK 超时: {signal_id}") return None def list_pending_signals( signal_type: str | None = None, source: str | None = None, limit: int = 20, ) -> list[dict]: """ 列出待确认的信号。 Args: signal_type: 过滤类型，None 表示全部 source: 过滤来源，None 表示全部 limit: 返回条数上限 Returns: 信号列表（按时间倒序） """ if not SIGNAL_DIR.exists(): return [] signals: list[dict] = [] for fpath in sorted(SIGNAL_DIR.glob("sig_*.json"), key=lambda f: f.stat().st_mtime, reverse=True): sig = _safe_read(fpath) if sig is None: continue if sig.get("status") == "done": continue # 已归档，跳过 if signal_type and sig.get("type") != signal_type: continue if source and sig.get("source") != source: continue signals.append(sig) if len(signals) >= limit: break return signals # ─── 清理线程 ────────────────────────────────────────────────────── _cleanup_lock = threading.Lock() _cleanup_running = False def start_cleanup_thread() -> None: """启动后台清理线程（幂等调用）""" global _cleanup_running with _cleanup_lock: if _cleanup_running: return _cleanup_running = True def run(): while True: time.sleep(CLEANUP_INTERVAL_SEC) try: cleanup_stale_files() except Exception as e: logger.error(f"清理线程异常: {e}") t = threading.Thread(target=run, daemon=True, name="queue-cleanup") t.start() logger.info("队列清理线程已启动") def cleanup_stale_files() -> int: """ 删除过期的队列和信号文件。 Returns: 删除的文件数量 """ removed = 0 now = datetime.now() for d in [QUEUE_DIR, SIGNAL_DIR]: if not d.exists(): continue for fpath in d.glob("*.json"): try: mtime = datetime.fromtimestamp(fpath.stat().st_mtime) if d == QUEUE_DIR: age_days = (now - mtime).days if age_days > MAX_QUEUE_AGE_DAYS: fpath.unlink(missing_ok=True) removed += 1 else: # SIGNAL_DIR age_hours = (now - mtime).total_seconds() / 3600 sig = _safe_read(fpath) # 删除超过 6 小时或已完成的信号 if age_hours > MAX_SIGNAL_AGE_HOURS or ( sig and sig.get("status") in ("acknowledged", "done") ): fpath.unlink(missing_ok=True) removed += 1 except OSError: pass if removed > 0: logger.debug(f"清理完成，删除 {removed} 个过期文件") return removed FILE:config.py """ hermes-memory-bridge / config.py 路径、常量与日志配置支持环境变量覆盖： HERMES_HOME - Hermes 根目录（默认 ~/.hermes） WORKBUDDY_HOME - WorkBuddy 根目录（默认 ~/WorkBuddy） BRIDGE_LOG_LEVEL - 日志级别 DEBUG|INFO|WARNING|ERROR（默认 INFO） """ from __future__ import annotations import logging import os import sys from pathlib import Path from typing import Optional # ─── 版本 ──────────────────────────────────────────────────────────── __version__ = "1.1.0" # ─── 日志初始化（所有模块共享）─────────────────────────────────────── _log_level = os.getenv("BRIDGE_LOG_LEVEL", "INFO").upper() _log_format = "%(asctime)s [%(levelname)s] %(name)s: %(message)s" logging.basicConfig( level=getattr(logging, _log_level, logging.INFO), format=_log_format, stream=sys.stderr, ) logger = logging.getLogger("hermes-memory-bridge") def _get_logger(name: str) -> logging.Logger: """获取子模块 logger""" return logging.getLogger(f"hermes-memory-bridge.{name}") # ─── Hermes 路径（支持 HERMES_HOME 环境变量）──────────────────────── def _resolve_hermes_home() -> Path: env = os.getenv("HERMES_HOME", "").strip() if env: p = Path(env) if not p.is_dir(): logger.warning(f"HERMES_HOME={env} 不存在或不是目录，使用默认值 ~/.hermes") p = Path.home() / ".hermes" else: p = Path.home() / ".hermes" return p HERMES_HOME: Path = _resolve_hermes_home() HERMES_MEMORIES_DIR: Path = HERMES_HOME / "memories" HERMES_DB: Path = HERMES_HOME / "state.db" # ─── 共用互通目录 ──────────────────────────────────────────────────── SHARED_DIR: Path = HERMES_HOME / "shared" WORKBUDDY_LOG: Path = SHARED_DIR / "workbuddy.log" HERMES_LOG: Path = SHARED_DIR / "hermes.log" BRIDGE_META: Path = SHARED_DIR / "meta.json" # ─── WorkBuddy 路径（智能查找，支持 WORKBUDDY_HOME 环境变量）────────── def get_workbuddy_memory_dir() -> Optional[Path]: """ 动态查找 WorkBuddy 记忆目录。优先级： 1. WORKBUDDY_MEMORY_DIR 环境变量（完整路径） 2. WORKBUDDY_HOME 环境变量 → 其下找最新时间戳子目录 3. ~/WorkBuddy → 找最新时间戳子目录 → .workbuddy/memory Returns: Path 或 None（均找不到时返回 None，不抛异常） """ # 优先级 1：直接指定 env_mem = os.getenv("WORKBUDDY_MEMORY_DIR", "").strip() if env_mem: p = Path(env_mem) if p.is_dir(): logger.debug(f"使用 WORKBUDDY_MEMORY_DIR={p}") return p logger.warning(f"WORKBUDDY_MEMORY_DIR={env_mem} 不存在或不是目录") # 优先级 2：WORKBUDDY_HOME wb_home = os.getenv("WORKBUDDY_HOME", "").strip() if not wb_home: wb_home = str(Path.home() / "WorkBuddy") wb_root = Path(wb_home) if not wb_root.is_dir(): logger.warning(f"WorkBuddy 根目录不存在: {wb_root}，记忆搜索功能将受限") return None # 找最新子目录（格式为纯数字时间戳） try: subdirs = [ d for d in wb_root.iterdir() if d.is_dir() and d.name.isdigit() and len(d.name) >= 10 ] except PermissionError as e: logger.warning(f"读取 {wb_root} 权限不足: {e}") return None if not subdirs: logger.warning(f"{wb_root} 下未找到 WorkBuddy 项目子目录，记忆搜索功能将受限") return None latest = max(subdirs, key=lambda d: d.name) memory_dir = latest / ".workbuddy" / "memory" logger.debug(f"自动发现 WorkBuddy 目录: {latest.name}，记忆目录: {memory_dir}") return memory_dir WORKBUDDY_MEMORY_DIR: Optional[Path] = get_workbuddy_memory_dir() # ─── 常量 ──────────────────────────────────────────────────────────── ENTRY_DELIMITER = "\n§\n" MAX_ENTRY_CHARS = 4000 MAX_EVENTS = 100 # meta.json 中保留的最大事件条数 FILE:event_signaler.py #!/usr/bin/env python3 """ hermes-memory-bridge / event_signaler.py Hermes 侧信号发射器 — 将 Hermes 的重要操作主动通知 WorkBuddy v1.3 新增（闭环命令系统）： - send_task — 向 WorkBuddy 发送命令任务 - feedback — 轮询 WorkBuddy 的处理结果 - mark_read — 标记反馈为已读用法（从 Hermes Agent 调用）： python3 event_signaler.py [args...] 命令： emit

发射通知信号 send_task [params] 发送一条命令给 WorkBuddy 执行 feedback [limit] [--unread] 轮询 WorkBuddy 的处理结果 mark_read 标记反馈为已读 ack 确认收到某信号 stats 显示信号统计 """ from __future__ import annotations import json import os import sys import time import uuid from datetime import datetime from pathlib import Path from typing import Any HERMES_HOME = Path.home() / ".hermes" SHARED_DIR = HERMES_HOME / "shared" SIGNAL_DIR = SHARED_DIR / "signals" QUEUE_DIR = SHARED_DIR / "queue" FEEDBACK_DIR = SHARED_DIR / "feedback" # 日志配置 _log_level = os.getenv("BRIDGE_LOG_LEVEL", "INFO").upper() import logging logging.basicConfig( level=getattr(logging, _log_level, logging.INFO), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", stream=sys.stderr, ) logger = logging.getLogger("event_signaler") def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _safe_write(path: Path, data: Any) -> bool: try: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return True except (PermissionError, OSError) as e: logger.error(f"写入失败 {path}: {e}") return False def _ensure_dir(p: Path) -> None: try: p.mkdir(parents=True, exist_ok=True) except PermissionError: logger.error(f"权限不足: {p}") def _write_log(message: str) -> None: """写入 Hermes 运行日志""" log_path = SHARED_DIR / "hermes.log" _ensure_dir(SHARED_DIR) timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") entry = f"[{timestamp}] {message}\n" try: log_path.write_text( log_path.read_text(encoding="utf-8") + entry, encoding="utf-8" ) lines = log_path.read_text(encoding="utf-8").strip().split("\n") if len(lines) > 500: log_path.write_text("\n".join(lines[-500:]) + "\n", encoding="utf-8") except OSError: pass # ─── 信号发射 ────────────────────────────────────────────────────── def emit_signal(signal_type: str, summary: str, data: dict | None = None) -> str | None: """ 从 Hermes 发射信号，通知 WorkBuddy。 """ _ensure_dir(SIGNAL_DIR) sid = str(uuid.uuid4())[:12] signal = { "id": sid, "type": signal_type, "source": "Hermes", "priority": "normal", "summary": summary, "data": data or {}, "created_at": _ts(), "status": "pending", } fname = SIGNAL_DIR / f"sig_{signal_type}_{sid}.json" if _safe_write(fname, signal): logger.info(f"[Hermes→] 信号发射: {signal_type} ({sid})") _write_log(f"[SIGNAL→WorkBuddy] {signal_type}: {summary}") return sid return None # ─── 命令任务发送（v1.3 新增）────────────────────────────────────── def send_task(command: str, params: dict | None = None) -> str | None: """ 向 WorkBuddy 发送一条命令任务（带完整参数）。 WorkBuddy 的 task_processor 会自动解析并执行。 Args: command: 命令类型 search_memory | sync_session | create_task | complete_task | list_tasks | ack | echo params: 命令参数字典 Returns: 信号 ID 或 None """ _ensure_dir(SIGNAL_DIR) tid = str(uuid.uuid4())[:12] task_signal = { "id": tid, "type": "task", "source": "Hermes", "priority": "normal", "summary": f"[命令] {command}", "data": { "command": command, "params": params or {}, }, "created_at": _ts(), "status": "pending", } fname = SIGNAL_DIR / f"sig_task_{tid}.json" if _safe_write(fname, task_signal): _write_log(f"[TASK→WorkBuddy] command={command}, id={tid}, params={params}") logger.info(f"[Hermes→WorkBuddy] 命令已发送: {command} ({tid})") return tid return None # ─── 结果反馈轮询（v1.3 新增）───────────────────────────────────── def _is_feedback_read(feedback_id: str) -> bool: return (FEEDBACK_DIR / f".read_{feedback_id}").exists() def _mark_feedback_read(feedback_id: str) -> bool: try: _ensure_dir(FEEDBACK_DIR) marker = FEEDBACK_DIR / f".read_{feedback_id}" marker.write_text(_ts(), encoding="utf-8") return True except OSError: return False def poll_feedback(limit: int = 10, mark_read: bool = False) -> list[dict]: """ 轮询 WorkBuddy 发来的处理结果反馈。 """ if not FEEDBACK_DIR.exists(): return [] feedback_list: list[dict] = [] for fpath in sorted(FEEDBACK_DIR.glob("fb_*.json"), key=lambda f: f.stat().st_mtime): fb = _safe_read(fpath) if fb is None: continue if fb.get("source") != "WorkBuddy": continue feedback_list.append(fb) if len(feedback_list) >= limit: break if mark_read: for fb in feedback_list: _mark_feedback_read(fb["id"]) return feedback_list # ─── 原有功能 ────────────────────────────────────────────────────── def poll_signals(direction: str = "wb_to_hm", limit: int = 10) -> list[dict]: if not SIGNAL_DIR.exists(): return [] signals = [] for fpath in sorted(SIGNAL_DIR.glob("sig_*.json"), key=lambda f: f.stat().st_mtime, reverse=True): sig = _safe_read(fpath) if sig is None: continue if sig.get("status") != "pending": continue if sig.get("source") != "WorkBuddy": continue signals.append(sig) if len(signals) >= limit: break return signals def ack_signal(signal_id: str) -> bool: if not SIGNAL_DIR.exists(): return False for fpath in SIGNAL_DIR.glob("sig_*.json"): sig = _safe_read(fpath) if sig is None: continue if sig.get("id") != signal_id: continue sig["status"] = "acknowledged" sig["acked_by"] = "Hermes" sig["acked_at"] = _ts() _safe_write(fpath, sig) _write_log(f"[ACK] 确认信号 {signal_id}，type={sig.get('type')}") logger.info(f"[Hermes] ACK 信号: {signal_id}") return True logger.warning(f"[Hermes] ACK 找不到信号: {signal_id}") return False def get_signal_stats() -> dict[str, Any]: stats = {"total": 0, "pending": 0, "acknowledged": 0, "by_type": {}} if not SIGNAL_DIR.exists(): return stats for fpath in SIGNAL_DIR.glob("sig_*.json"): sig = _safe_read(fpath) if sig is None: continue stats["total"] += 1 status = sig.get("status", "unknown") if status == "pending": stats["pending"] += 1 elif status == "acknowledged": stats["acknowledged"] += 1 stype = sig.get("type", "unknown") stats["by_type"][stype] = stats["by_type"].get(stype, 0) + 1 return stats # ─── CLI 入口 ────────────────────────────────────────────────────── def main(): if len(sys.argv) < 2: _print_help() sys.exit(0) cmd = sys.argv[1] if cmd == "emit": if len(sys.argv) < 4: print("用法: emit

[json_data]", file=sys.stderr) sys.exit(1) sig_type = sys.argv[2] summary = sys.argv[3] data = None if len(sys.argv) >= 5: try: data = json.loads(sys.argv[4]) except json.JSONDecodeError: print(f"❌ JSON 解析失败: {sys.argv[4]}", file=sys.stderr) sys.exit(1) sid = emit_signal(sig_type, summary, data) print(f"✅ 信号已发射: {sid}" if sid else "❌ 发射失败") elif cmd == "send_task": if len(sys.argv) < 3: print("用法: send_task [params_json]", file=sys.stderr) sys.exit(1) task_cmd = sys.argv[2] task_params = None if len(sys.argv) >= 4: try: task_params = json.loads(sys.argv[3]) except json.JSONDecodeError: print(f"❌ JSON 解析失败: {sys.argv[3]}", file=sys.stderr) sys.exit(1) tid = send_task(task_cmd, task_params) print(f"✅ 任务已发送: {task_cmd} → {tid}" if tid else "❌ 发送失败") elif cmd == "feedback": unread_only = "--unread" in sys.argv limit = 10 for a in sys.argv[2:]: if a.isdigit(): limit = int(a) break fb_list = poll_feedback(limit=limit, mark_read=not unread_only) if not fb_list: print("📭 没有待读反馈（WorkBuddy 尚未处理任务）") else: print(f"📬 WorkBuddy 处理结果 ({len(fb_list)} 条)：") for fb in fb_list: read_mark = "✅" if _is_feedback_read(fb["id"]) else "📩" result_ok = fb.get("result", {}).get("success", False) ok_mark = "✅成功" if result_ok else "❌失败" print(f"\n [{read_mark}] {fb['id']} | {fb['command']} | {ok_mark} | {fb['created_at'][:16]}") print(f" → {json.dumps(fb.get('result', {}), ensure_ascii=False)[:300]}") if not unread_only: print(f"\n（已自动标记 {len(fb_list)} 条为已读）") elif cmd == "mark_read": if len(sys.argv) < 3: print("用法: mark_read ", file=sys.stderr) sys.exit(1) ok = _mark_feedback_read(sys.argv[2]) print(f"{'✅' if ok else '❌'} 标记已读 {sys.argv[2]}") elif cmd == "poll": signals = poll_signals(limit=int(sys.argv[2]) if len(sys.argv) >= 3 and sys.argv[2].isdigit() else 10) if not signals: print("📭 没有待处理信号") else: print(f"📬 待处理信号 ({len(signals)} 条)：") for sig in signals: print(f" [{sig['id']}] {sig['type']} | {sig.get('summary', '')} | {sig.get('created_at', '')[:16]}") elif cmd == "ack": if len(sys.argv) < 3: print("用法: ack ", file=sys.stderr) sys.exit(1) ok = ack_signal(sys.argv[2]) print(f"{'✅' if ok else '❌'} ACK {sys.argv[2]}") elif cmd == "stats": stats = get_signal_stats() print(f"📊 信号统计") print(f" 总数: {stats['total']}") print(f" 待确认: {stats['pending']}") print(f" 已确认: {stats['acknowledged']}") print(f" 按类型:") for t, n in stats.get("by_type", {}).items(): print(f" {t}: {n}") # 反馈统计 _ensure_dir(FEEDBACK_DIR) fb_count = len(list(FEEDBACK_DIR.glob("fb_*.json"))) if FEEDBACK_DIR.exists() else 0 print(f" 反馈: {fb_count} 条") else: print(f"❌ 未知命令: {cmd}", file=sys.stderr) _print_help() sys.exit(1) def _print_help(): print(""" Hermes Event Signaler (v1.3) — 闭环命令任务系统 emit

[json_data] 发射通知信号到 WorkBuddy send_task [params_json] 发送一条命令给 WorkBuddy 执行 feedback [limit] [--unread] 轮询 WorkBuddy 的处理结果 mark_read 标记反馈为已读 poll [limit] 轮询来自 WorkBuddy 的待处理信号 ack 确认收到某信号 stats 显示信号统计命令类型（send_task）： search_memory {"keyword": "..."} sync_session {"topic": "...", "summary": "..."} create_task {"title": "..."} complete_task {"task_id": "..."} list_tasks {} ack {"signal_id": "...", "message": "..."} echo {"message": "..."} 示例: # 通知 WorkBuddy 某事完成 python3 event_signaler.py emit task_done "完成项目A的开发" # 让 WorkBuddy 执行 echo 测试 python3 event_signaler.py send_task echo '{"message":"ping"}' # 让 WorkBuddy 搜索记忆 python3 event_signaler.py send_task search_memory '{"keyword":"辽望客户端"}' # 轮询 WorkBuddy 的处理结果 python3 event_signaler.py feedback # 轮询（只看未读） python3 event_signaler.py feedback 5 --unread """) if __name__ == "__main__": main() FILE:event_watcher.py #!/usr/bin/env python3 """ hermes-memory-bridge / event_watcher.py WorkBuddy 侧事件监听器 — FSEvents + 自适应轮询双模式 v2.0 新增： - 内置任务处理器（task_processor）集成 - 收到 Hermes 命令后自动处理并回写结果（feedback_writer） - CLI 参数：--dry-run（仅处理，不回写）、--once（单次轮询）用法（独立运行）： python3 event_watcher.py # 持续监听 python3 event_watcher.py --poll-only # 强制轮询模式 python3 event_watcher.py --once # 单次处理现有信号后退出 python3 event_watcher.py --dry-run # 处理但不回写（测试用）用法（作为模块导入）： from event_watcher import watch, set_callback, process_hermes_signals watch() """ from __future__ import annotations import json import os import signal import sys import threading import time from datetime import datetime from pathlib import Path from typing import Any, Callable, Optional SKILL_DIR = Path(__file__).parent sys.path.insert(0, str(SKILL_DIR)) try: from config import SHARED_DIR, _get_logger except ImportError: SHARED_DIR = Path.home() / ".hermes" / "shared" import logging logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) _get_logger = lambda n: logging.getLogger(n) logger = _get_logger("event_watcher") # ─── 路径 ────────────────────────────────────────────────────────── SIGNAL_DIR = SHARED_DIR / "signals" WATCH_DIR = SHARED_DIR # 监听整个 shared 目录 # ─── 配置 ────────────────────────────────────────────────────────── POLL_FALLBACK_INTERVAL_SEC = 5.0 LAST_SEEN_FILE = SIGNAL_DIR / ".last_seen_signals" PROCESSED_FILE = SIGNAL_DIR / ".processed_signals" # ─── 回调机制（v1 兼容）──────────────────────────────────────────── _callback: Optional[Callable[[str, dict], None]] = None _callback_lock = threading.Lock() def set_callback(cb: Callable[[str, dict], None]) -> None: global _callback with _callback_lock: _callback = cb def _dispatch(event_type: str, data: dict) -> None: with _callback_lock: cb = _callback if cb is None: logger.debug(f"无回调，分发跳过: {event_type}") return try: cb(event_type, data) logger.info(f"事件已分发: {event_type}") except Exception as e: logger.error(f"回调执行异常: {e}") # ─── 工具函数 ────────────────────────────────────────────────────── def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _read_processed() -> set[str]: data = _safe_read(PROCESSED_FILE, {"processed": []}) return set(data.get("processed", [])) def _write_processed(processed: set[str]) -> None: SIGNAL_DIR.mkdir(parents=True, exist_ok=True) try: items = list(processed)[-200:] PROCESSED_FILE.write_text( json.dumps({"processed": items, "updated_at": _ts()}, ensure_ascii=False), encoding="utf-8", ) except OSError: pass def _poll_new_signals(source_filter: str = "Hermes") -> list[dict]: if not SIGNAL_DIR.exists(): return [] processed = _read_processed() new_signals = [] for fpath in sorted(SIGNAL_DIR.glob("sig_*.json"), key=lambda f: f.stat().st_mtime): sig = _safe_read(fpath) if sig is None: continue sid = sig.get("id", "") if sid in processed: continue if sig.get("source") != source_filter: continue new_signals.append(sig) processed.add(sid) if new_signals: _write_processed(processed) return new_signals # ─── v2.0 新增：任务处理集成 ────────────────────────────────────── _dry_run = False def set_dry_run(val: bool = True) -> None: global _dry_run _dry_run = val def _process_single_signal(sig: dict) -> None: """ 处理单条 Hermes 信号： 1. 提取命令类型和参数 2. 调用 task_processor 执行 3. 调用 feedback_writer 回写结果 4. 对原始信号发 ACK """ import importlib signal_id = sig.get("id", "") signal_type = sig.get("type", "") signal_data = sig.get("data", {}) logger.info( f"[处理] 信号 {signal_id[:12]} ({signal_type}): " f"{signal_data.get('summary', signal_data.get('message', ''))[:60]}" ) # 提取命令类型（优先从 data.command 获取，其次从 type 推断） command_type = signal_data.get("command", "") if not command_type: type_to_command = { "task_done": "echo", "sync": "sync_session", "config_change": "ack", "ack": "ack", "feedback": "ack", } command_type = type_to_command.get(signal_type, signal_type) # 提取参数 params = signal_data.get("params", signal_data) # 1. 执行命令 try: task_processor = importlib.import_module("task_processor") result = task_processor.process_command(command_type, params, signal_id) except Exception as e: logger.error(f"task_processor 执行失败: {e}") result = {"success": False, "error": str(e)} # 2. 回写结果（dry_run 模式下跳过） if not _dry_run: try: feedback_writer = importlib.import_module("feedback_writer") feedback_id = feedback_writer.write_feedback( signal_id, command_type, result, source_signal=sig ) if feedback_id: logger.info(f"[回写] 反馈 {feedback_id} 已写入 feedback 目录") except Exception as e: logger.error(f"feedback_writer 回写失败: {e}") # 3. 对原始信号发 ACK if not _dry_run: try: comm_queue = importlib.import_module("communication_queue") comm_queue.signal_ack(signal_id, "WorkBuddy") logger.info(f"[ACK] 已确认信号 {signal_id[:12]}") except Exception as e: logger.error(f"ACK 失败: {e}") def process_hermes_signals(dry_run: bool = False) -> int: """ 处理所有待处理的 Hermes 信号（一次性，不阻塞）。 Args: dry_run: True = 处理但不回写结果（测试用） Returns: 处理的信号数量 """ global _dry_run _dry_run = dry_run signals = _poll_new_signals(source_filter="Hermes") if not signals: return 0 for sig in signals: try: _process_single_signal(sig) except Exception as e: logger.error(f"处理信号异常: {e}") return len(signals) # ─── FSEvents 模式 ──────────────────────────────────────────────── def _try_fsevents() -> bool: try: import MacOS.FSEvents as FSEvents # type: ignore return True except ImportError: return False class FSEventsWatcher: def __init__(self, path: Path, latency: float = 0.5): self.path = str(path) self.latency = latency self._stream_ref: Optional[Any] = None self._running = False self._watch_thread: Optional[threading.Thread] = None def _create_stream(self): import MacOS.FSEvents as FSEvents # type: ignore self._stream_ref = FSEvents.EventStream( [self.path], self.latency, FSEvents.kFSEventStreamCreateFlagFileEvents, ) def callback( stream_ref, client_callback_info, num_events, event_paths, event_flags, event_ids, ): for i in range(num_events): path = event_paths[i] flag = event_flags[i] if flag & 0x0100: # kFSEventStreamEventFlagItemCreated self._on_create(Path(path)) self._stream_ref.start() self._running = True logger.info(f"FSEvents 监听已启动: {self.path}") def _on_create(self, fpath: Path) -> None: if fpath.suffix != ".json": return sig = _safe_read(fpath) if sig and sig.get("source") == "Hermes": logger.info(f"[FSEvents] 检测到 Hermes 命令: {sig.get('id')} ({sig.get('type')})") try: _process_single_signal(sig) except Exception as e: logger.error(f"FSEvents 处理异常: {e}") def start(self) -> None: if not _try_fsevents(): logger.warning("FSEvents 不可用，降级为轮询模式") return try: self._create_stream() except Exception as e: logger.warning(f"FSEvents 启动失败，降级为轮询: {e}") return self._watch_thread = threading.Thread(target=self._run_loop, daemon=True) self._watch_thread.start() def _run_loop(self) -> None: try: while self._running: time.sleep(1) except Exception: pass def stop(self) -> None: self._running = False if self._stream_ref: try: self._stream_ref.stop() except Exception: pass logger.info("FSEvents 监听已停止") # ─── 自适应轮询模式 ──────────────────────────────────────────────── class AdaptivePoller: def __init__(self): self._interval = 60.0 self._min_interval = 60.0 self._max_interval = 300.0 self._decay_factor = 1.15 self._growth_factor = 0.7 self._running = False self._thread: Optional[threading.Thread] = None def start(self) -> None: self._running = True self._thread = threading.Thread(target=self._run, daemon=True, name="adaptive-poller") self._thread.start() logger.info(f"自适应轮询已启动（初始间隔 {self._interval}s）") def stop(self) -> None: self._running = False if self._thread: self._thread.join(timeout=5) logger.info("自适应轮询已停止") def _run(self) -> None: while self._running: try: processed = process_hermes_signals(dry_run=False) if processed > 0: logger.info(f"[Poller] 批处理 {processed} 条 Hermes 信号") self._interval = max( self._min_interval, self._interval * self._growth_factor, ) else: self._interval = min( self._max_interval, self._interval * self._decay_factor, ) except Exception as e: logger.error(f"轮询异常: {e}") self._interval = self._min_interval time.sleep(self._interval) # ─── 统一入口 ────────────────────────────────────────────────────── _watcher: Optional[FSEventsWatcher] = None _poller: Optional[AdaptivePoller] = None # ─── Hermes 协调任务监控器 ──────────────────────────────────────── # 检测 Hermes 发来的协调任务，主动通知用户确认后再执行 COORD_DIR = SHARED_DIR / "coordination" COORD_ALERT_FILE = SIGNAL_DIR / ".coord_alert_state" _coord_alert_state: dict[str, Any] = {"last_seen_task": None} def _load_coord_state() -> dict[str, Any]: data = _safe_read(COORD_ALERT_FILE, {"last_seen_task": None, "notified_ids": []}) return data def _save_coord_state(state: dict) -> None: try: COORD_ALERT_FILE.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8") except OSError: pass def _get_pending_coord_tasks() -> list[dict]: """找出 Hermes 发来的、未通知过用户的协调任务""" if not COORD_DIR.exists(): return [] state = _load_coord_state() notified = set(state.get("notified_ids", [])) pending = [] for fpath in sorted(COORD_DIR.glob("*.md"), key=lambda f: f.stat().st_mtime, reverse=True): try: content = fpath.read_text(encoding="utf-8") # 提取任务 ID（文件名格式: type_taskid.md） fname = fpath.stem parts = fname.split("_", 1) if len(parts) < 2: continue task_id = parts[1] if task_id in notified: continue # 解析协调任务元信息（从文件内容提取） title = "" lines = content.strip().split("\n") for line in lines[:10]: if line.startswith("# "): title = line[2:].strip() break pending.append({ "id": task_id, "title": title or fname, "file": str(fpath), "preview": content[:200].strip(), }) notified.add(task_id) except (OSError, UnicodeDecodeError): continue if pending: state["notified_ids"] = list(notified) _save_coord_state(state) return pending def check_hermes_coordination_tasks() -> list[dict]: """ 主入口：检查 Hermes 是否有新的协调任务。发现新任务后写入 feedback 目录，通知用户确认后再执行。 Returns: 新发现的任务列表（每次最多返回 1 条，优先最新） """ tasks = _get_pending_coord_tasks() if not tasks: return [] # 只通知最新一条（避免刷屏） task = tasks[0] alert_msg = ( f"📋 **Hermes 新协调任务**\n\n" f"**任务 ID**: {task['id'][:12]}\n" f"**任务标题**: {task['title']}\n\n" f"请确认是否执行此任务。回复「确认」开始执行，或「跳过」忽略。\n" f"> 查看详情：`cat {task['file']}`" ) # 写入 feedback，让 Hermes 下次轮询时读到并推送给用户 _write_coord_alert_feedback(task, alert_msg) logger.info(f"[HermesAlert] 发现新协调任务: {task['id'][:12]} - {task['title']}") return [task] def _write_coord_alert_feedback(task: dict, message: str) -> None: """将协调任务提醒写入 feedback 目录""" feedback_dir = SHARED_DIR / "feedback" feedback_dir.mkdir(exist_ok=True) feedback_id = f"coord_{task['id'][:12]}" fb_file = feedback_dir / f"fb_{feedback_id}.json" payload = { "id": f"fb_{feedback_id}", "type": "coordination_alert", "source": "WorkBuddy", "target": "user", "timestamp": _ts(), "task_id": task["id"], "task_title": task["title"], "message": message, "status": "awaiting_confirmation", "requires_user_action": True, "action_options": ["confirm", "skip"], "content_preview": task["preview"], } try: fb_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") logger.info(f"[HermesAlert] 已写入反馈: fb_{feedback_id}.json") except OSError as e: logger.error(f"[HermesAlert] 写入反馈失败: {e}") # ─── 在主轮询循环中集成 ────────────────────────────────────────── def _process_adaptive_with_alert() -> int: """ 轮询 Hermes 信号 + 检查协调任务。发现协调任务时主动通知用户（不自动处理）。 """ # 1. 检查协调任务（优先） try: new_tasks = check_hermes_coordination_tasks() if new_tasks: # 有新协调任务：写入通知，跳过自动处理，等用户确认 logger.info(f"[HermesAlert] {len(new_tasks)} 条协调任务已通知用户，等待确认") except Exception as e: logger.error(f"[HermesAlert] 检查协调任务异常: {e}") # 2. 处理普通 Hermes 信号 try: processed = process_hermes_signals(dry_run=False) return processed except Exception as e: logger.error(f"处理信号异常: {e}") return 0 class AdaptivePollerWithAlert(AdaptivePoller): """带协调任务监控的自适应轮询器""" def _run(self) -> None: while self._running: try: # 检查协调任务（不触发普通信号处理，保留给下一步） processed_signals = _process_adaptive_with_alert() if processed_signals > 0: logger.info(f"[Poller] 批处理 {processed_signals} 条 Hermes 信号") self._interval = max( self._min_interval, self._interval * self._growth_factor, ) else: self._interval = min( self._max_interval, self._interval * self._decay_factor, ) except Exception as e: logger.error(f"轮询异常: {e}") self._interval = self._min_interval time.sleep(self._interval) # ─── 覆盖原 watch() 函数的轮询器 ───────────────────────────────── def watch() -> None: global _watcher, _poller logger.info("启动 WorkBuddy 事件监听（v2.0 含任务处理 + 主动通知）...") if _try_fsevents(): _watcher = FSEventsWatcher(WATCH_DIR) _watcher.start() logger.info("✅ 使用 FSEvents 模式") # FSEvents 检测到信号后也需要检查协调任务 else: logger.info("⚠️ FSEvents 不可用，使用自适应轮询模式") _poller = AdaptivePollerWithAlert() _poller.start() def signal_handler(signum, frame): logger.info("收到退出信号，正在停止...") stop() sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) try: while True: time.sleep(3600) except KeyboardInterrupt: stop() def stop() -> None: if _watcher: _watcher.stop() if _poller: _poller.stop() logger.info("事件监听已停止") # ─── CLI 入口 ────────────────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="WorkBuddy 事件监听器（含任务处理闭环）" ) parser.add_argument("--poll-only", action="store_true", help="强制轮询模式") parser.add_argument("--once", action="store_true", help="单次处理现有信号后退出") parser.add_argument("--dry-run", action="store_true", help="处理但不回写结果（测试用）") args = parser.parse_args() if args.once: count = process_hermes_signals(dry_run=args.dry_run) print(f"处理了 {count} 条 Hermes 信号（dry_run={args.dry_run}）") else: if args.poll_only: _poller = AdaptivePoller() _poller.start() print("轮询模式已启动，按 Ctrl+C 退出") try: while True: time.sleep(3600) except KeyboardInterrupt: _poller.stop() print("已停止") else: print("启动 WorkBuddy 事件监听（FSEvents + 自适应轮询）...") if args.dry_run: set_dry_run(True) print("⚠️ DRY-RUN 模式：处理但不回写结果") watch() FILE:event_watcher_extended.py #!/usr/bin/env python3 """ hermes-memory-bridge / event_watcher_extended.py WorkBuddy 侧事件监听器扩展版 — 使用扩展版任务处理器基于原始event_watcher.py，但使用task_processor_extended.py """ from __future__ import annotations import json import os import signal import sys import threading import time from datetime import datetime from pathlib import Path from typing import Any, Callable, Optional SKILL_DIR = Path(__file__).parent sys.path.insert(0, str(SKILL_DIR)) try: from config import SHARED_DIR, _get_logger except ImportError: SHARED_DIR = Path.home() / ".hermes" / "shared" import logging logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) _get_logger = lambda n: logging.getLogger(n) logger = _get_logger("event_watcher_extended") # ─── 路径 ────────────────────────────────────────────────────────── SIGNAL_DIR = SHARED_DIR / "signals" WATCH_DIR = SHARED_DIR # 监听整个 shared 目录 # ─── 全局状态 ────────────────────────────────────────────────────── _dry_run = False _running = True _callback: Optional[Callable[[dict], None]] = None # ─── 工具函数 ────────────────────────────────────────────────────── def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _poll_new_signals(source_filter: Optional[str] = None) -> list[dict]: """轮询新信号（状态为 pending）""" if not SIGNAL_DIR.exists(): return [] signals = [] for fpath in SIGNAL_DIR.glob("sig_*.json"): sig = _safe_read(fpath) if sig is None: continue if sig.get("status") != "pending": continue if source_filter and sig.get("source") != source_filter: continue signals.append(sig) # 按创建时间排序（旧→新） signals.sort(key=lambda s: s.get("created_at", "")) return signals # ─── 信号处理核心 ────────────────────────────────────────────────── def _process_signal(sig: dict) -> None: """处理单个信号""" signal_id = sig.get("id", "") sig_type = sig.get("type", "") source = sig.get("source", "") logger.info(f"[处理] 信号 {signal_id[:12]} ({sig_type}): {sig.get('summary', '')[:60]}") # 只处理 task 类型的 Hermes 信号 if sig_type != "task" or source != "Hermes": return # 提取命令和参数 data = sig.get("data", {}) command_type = data.get("command", "") params = data.get("params", {}) if not command_type: logger.warning(f"信号 {signal_id} 缺少 command 字段") return # 1. 执行命令（使用扩展版任务处理器） try: import importlib task_processor = importlib.import_module("task_processor_extended") result = task_processor.process_command(command_type, params, signal_id) except Exception as e: logger.error(f"task_processor_extended 执行失败: {e}") result = {"success": False, "error": str(e)} # 2. 回写结果（dry_run 模式下跳过） if not _dry_run: try: feedback_writer = importlib.import_module("feedback_writer") feedback_id = feedback_writer.write_feedback( signal_id, command_type, result, source_signal=sig ) if feedback_id: logger.info(f"[回写] 反馈 {feedback_id} 已写入 feedback 目录") except Exception as e: logger.error(f"feedback_writer 回写失败: {e}") # 3. 对原始信号发 ACK if not _dry_run: try: comm_queue = importlib.import_module("communication_queue") comm_queue.signal_ack(signal_id, "WorkBuddy") logger.info(f"[ACK] 已确认信号 {signal_id[:12]}") except Exception as e: logger.error(f"ACK 失败: {e}") def process_hermes_signals(dry_run: bool = False) -> int: """ 处理所有待处理的 Hermes 信号（一次性，不阻塞）。 Args: dry_run: True = 处理但不回写结果（测试用） Returns: 处理的信号数量 """ global _dry_run _dry_run = dry_run signals = _poll_new_signals(source_filter="Hermes") if not signals: return 0 for sig in signals: try: _process_signal(sig) except Exception as e: logger.error(f"处理信号 {sig.get('id', '?')} 失败: {e}") return len(signals) # ─── 自适应轮询器（FSEvents 不可用时的降级方案）──────────────────── class AdaptivePoller: """自适应轮询器，根据信号频率动态调整轮询间隔""" def __init__(self, base_interval: float = 60.0): self.base_interval = base_interval self.current_interval = base_interval self.min_interval = 5.0 self.max_interval = 300.0 self.last_signal_count = 0 self.last_poll_time = time.time() def poll(self) -> int: """执行一次轮询，返回处理的信号数量""" processed = process_hermes_signals() now = time.time() # 自适应调整间隔 if processed > 0: # 有信号，加快轮询 self.current_interval = max(self.min_interval, self.current_interval * 0.7) else: # 无信号，减慢轮询（但不超过最大值） self.current_interval = min(self.max_interval, self.current_interval * 1.1) self.last_signal_count = processed self.last_poll_time = now return processed def get_next_interval(self) -> float: """获取下一次轮询的间隔（秒）""" return self.current_interval # ─── 主监听循环 ──────────────────────────────────────────────────── def watch(poll_only: bool = False, once: bool = False, dry_run: bool = False) -> None: """ 启动事件监听。 Args: poll_only: 强制使用轮询模式（即使 FSEvents 可用） once: 单次处理现有信号后退出 dry_run: 处理但不回写结果（测试用） """ global _running, _dry_run _dry_run = dry_run logger.info("启动 WorkBuddy 事件监听（扩展版，含天气查询）...") # 检查 FSEvents 可用性 fsevents_available = False if not poll_only: try: import fsevents fsevents_available = True logger.info("✅ FSEvents 可用，使用文件系统事件监听") except ImportError: logger.warning("⚠️ FSEvents 不可用，使用自适应轮询模式") if fsevents_available and not poll_only: # FSEvents 模式 try: import fsevents from fsevents import Stream def fsevents_callback(event): if event.name.endswith(".json") and "signals" in event.name: process_hermes_signals(dry_run) stream = Stream(fsevents_callback, str(WATCH_DIR), file_events=True) stream.start() logger.info("FSEvents 监听已启动") # 等待退出信号 def stop(signum, frame): global _running _running = False stream.stop() signal.signal(signal.SIGINT, stop) signal.signal(signal.SIGTERM, stop) while _running: time.sleep(1) except Exception as e: logger.error(f"FSEvents 启动失败: {e}") fsevents_available = False if not fsevents_available or poll_only: # 自适应轮询模式 poller = AdaptivePoller(base_interval=30.0) # 初始间隔30秒 def stop(signum, frame): global _running _running = False signal.signal(signal.SIGINT, stop) signal.signal(signal.SIGTERM, stop) logger.info(f"自适应轮询已启动（初始间隔 {poller.base_interval:.1f}s）") if once: # 单次模式 processed = poller.poll() logger.info(f"单次处理完成，处理了 {processed} 个信号") return # 持续轮询 while _running: processed = poller.poll() if processed > 0: logger.info(f"[Poller] 批处理 {processed} 条 Hermes 信号") # 等待下一次轮询 interval = poller.get_next_interval() for _ in range(int(interval)): if not _running: break time.sleep(1) # ─── CLI 入口 ────────────────────────────────────────────────────── def main(): import argparse parser = argparse.ArgumentParser( description="WorkBuddy 事件监听器（扩展版）", epilog="示例：\n" " python3 event_watcher_extended.py # 持续监听\n" " python3 event_watcher_extended.py --poll-only # 强制轮询模式\n" " python3 event_watcher_extended.py --once # 单次处理现有信号\n" " python3 event_watcher_extended.py --dry-run # 处理但不回写（测试）" ) parser.add_argument("--poll-only", action="store_true", help="强制使用轮询模式（即使 FSEvents 可用）") parser.add_argument("--once", action="store_true", help="单次处理现有信号后退出") parser.add_argument("--dry-run", action="store_true", help="处理但不回写结果（测试用）") args = parser.parse_args() watch( poll_only=args.poll_only, once=args.once, dry_run=args.dry_run ) if __name__ == "__main__": main() FILE:feedback_writer.py """ hermes-memory-bridge / feedback_writer.py 结果回写器 — WorkBuddy 处理完 Hermes 命令后，将结果写入共享目录工作原理： 1. task_processor.py 执行完命令后，调用 write_feedback() 回写结果 2. 结果以 signal 形式写入 ~/.hermes/shared/feedback/ 目录 3. Hermes 通过 event_signaler.py feedback poll 命令读取结果用法（作为模块导入）： from feedback_writer import write_feedback write_feedback(signal_id, command_type, result) """ from __future__ import annotations import json import os import sys import uuid from datetime import datetime from pathlib import Path from typing import Any, Optional SKILL_DIR = Path(__file__).parent sys.path.insert(0, str(SKILL_DIR)) try: from config import SHARED_DIR, _get_logger except ImportError: SHARED_DIR = Path.home() / ".hermes" / "shared" import logging logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) _get_logger = lambda n: logging.getLogger(n) logger = _get_logger("feedback_writer") # ─── 路径 ────────────────────────────────────────────────────────── FEEDBACK_DIR = SHARED_DIR / "feedback" # WorkBuddy 结果 → Hermes 读取 # ─── 工具函数 ────────────────────────────────────────────────────── def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _safe_write(path: Path, data: Any) -> bool: try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return True except (PermissionError, OSError) as e: logger.error(f"写入失败 {path}: {e}") return False def _ensure_dir() -> bool: try: FEEDBACK_DIR.mkdir(parents=True, exist_ok=True) return True except PermissionError: logger.error(f"权限不足，无法创建目录 {FEEDBACK_DIR}") return False # ─── 核心函数 ────────────────────────────────────────────────────── def write_feedback( signal_id: str, command_type: str, result: dict, source_signal: Optional[dict] = None, ) -> Optional[str]: """ 将命令执行结果写入反馈文件。 Args: signal_id: Hermes 原始信号的 ID（用于关联） command_type: 命令类型 result: task_processor.process_command() 的返回结果 source_signal: 原始信号 dict（可选，用于复制元数据） Returns: feedback_id 或 None（失败时） """ if not _ensure_dir(): return None fid = str(uuid.uuid4())[:12] feedback = { "id": f"fb_{fid}", "ref_signal_id": signal_id, "command": command_type, "source": "WorkBuddy", "target": "Hermes", "status": "done" if result.get("success") else "error", "result": result, "created_at": _ts(), "processed_at": result.get("_meta", {}).get("processed_at", _ts()), "elapsed_ms": result.get("_meta", {}).get("elapsed_ms", 0), } # 复制原始信号的摘要信息 if source_signal: feedback["source_summary"] = source_signal.get("data", {}).get( "summary", source_signal.get("summary", "") ) feedback["source_type"] = source_signal.get("type", command_type) fname = FEEDBACK_DIR / f"fb_{command_type}_{fid}.json" if _safe_write(fname, feedback): logger.info(f"结果回写成功 [{signal_id[:8]}]: fb_{fid} ({command_type})") return f"fb_{fid}" return None def read_feedback(feedback_id: str) -> Optional[dict]: """按 feedback_id 读取单条反馈（Hermes 侧用）""" try: fpath = FEEDBACK_DIR / f"fb_*{feedback_id}*.json" matches = list(FEEDBACK_DIR.glob(f"fb_*{feedback_id}*.json")) if not matches: return None return json.loads(matches[0].read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return None def list_pending_feedback(source: str = "WorkBuddy", limit: int = 20) -> list[dict]: """ 列出未读的反馈（供 Hermes 轮询）。 Args: source: 过滤来源，"WorkBuddy" = WorkBuddy 发给 Hermes 的结果 limit: 返回条数上限 Returns: 反馈列表（按时间升序 = 最早的在前） """ if not FEEDBACK_DIR.exists(): return [] feedback_list: list[dict] = [] for fpath in sorted(FEEDBACK_DIR.glob("fb_*.json"), key=lambda f: f.stat().st_mtime): try: fb = json.loads(fpath.read_text(encoding="utf-8")) if fb.get("source") == source: feedback_list.append(fb) if len(feedback_list) >= limit: break except (json.JSONDecodeError, OSError): continue return feedback_list def mark_feedback_read(feedback_id: str) -> bool: """标记反馈为已读（写入 .read 标记文件，幂等）""" try: marker = FEEDBACK_DIR / f".read_{feedback_id}" marker.write_text(_ts(), encoding="utf-8") return True except OSError: return False def is_feedback_read(feedback_id: str) -> bool: """检查反馈是否已被 Hermes 读取""" return (FEEDBACK_DIR / f".read_{feedback_id}").exists() def cleanup_old_feedback(max_age_hours: int = 24) -> int: """ 删除超过 max_age_hours 的反馈文件（默认 24h）。 Returns: 删除数量 """ if not FEEDBACK_DIR.exists(): return 0 import time as time_module cutoff = time_module.time() - max_age_hours * 3600 removed = 0 for fpath in FEEDBACK_DIR.glob("fb_*.json"): if fpath.stat().st_mtime < cutoff: try: fpath.unlink(missing_ok=True) removed += 1 except OSError: pass if removed > 0: logger.debug(f"清理反馈文件 {removed} 条") return removed # ─── CLI 入口 ────────────────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="WorkBuddy → Hermes 反馈写入器") sub = parser.add_subparsers(dest="cmd", help="子命令") p_list = sub.add_parser("list", help="列出待读反馈") p_list.add_argument("--limit", type=int, default=20) p_list.add_argument("--unread", action="store_true", help="仅显示未读") p_write = sub.add_parser("write", help="写入反馈") p_write.add_argument("--signal-id", required=True) p_write.add_argument("--command", required=True) p_write.add_argument("--result", required=True, help="JSON 字符串") p_write.add_argument("--ref-summary", default="") p_cleanup = sub.add_parser("cleanup", help="清理旧反馈") p_cleanup.add_argument("--max-hours", type=int, default=24) args = parser.parse_args() if args.cmd == "list": fbs = list_pending_feedback(limit=args.limit) print(f"待读反馈共 {len(fbs)} 条：\n") for fb in fbs: read_mark = "✅已读" if is_feedback_read(fb["id"]) else "📩未读" print(f"[{read_mark}] {fb['id']} | {fb['command']} | {fb['created_at'][:16]} | {fb['result'].get('success', '?')}") print(f" → {json.dumps(fb['result'], ensure_ascii=False)[:200]}") print() elif args.cmd == "write": try: result_data = json.loads(args.result) except json.JSONDecodeError: print(f"JSON 解析失败: {args.result}") sys.exit(1) source_signal = {"summary": args.ref_summary} if args.ref_summary else None fid = write_feedback(args.signal_id, args.command, result_data, source_signal) print(f"写入成功: {fid}" if fid else "写入失败") elif args.cmd == "cleanup": n = cleanup_old_feedback(args.max_hours) print(f"清理完成，删除 {n} 条") else: parser.print_help() FILE:hermes_learning_sync.py #!/usr/bin/env python3 """ hermes-memory-bridge / hermes_learning_sync.py Hermes 学习材料同步模块 """ import json import os import shutil from datetime import datetime from pathlib import Path # 配置路径 HERMES_HOME = Path.home() / ".hermes" SHARED_DIR = HERMES_HOME / "shared" PROCESSED_DIR = SHARED_DIR / "processed" WORKBUDDY_SKILLS_DIR = Path.home() / ".workbuddy" / "skills" WORKBUDDY_LEARNING_DIR = WORKBUDDY_SKILLS_DIR / "hermes-learning" def ensure_dirs(): """确保所有必要的目录存在""" SHARED_DIR.mkdir(parents=True, exist_ok=True) PROCESSED_DIR.mkdir(parents=True, exist_ok=True) WORKBUDDY_LEARNING_DIR.mkdir(parents=True, exist_ok=True) def get_latest_learning_materials(): """获取最新的学习材料""" materials = {} # 1. 读取记忆摘要 summary_path = SHARED_DIR / "memory_summary.json" if summary_path.exists(): with open(summary_path, 'r', encoding='utf-8') as f: materials["summary"] = json.load(f) # 2. 读取完整学习材料 learning_path = PROCESSED_DIR / "learning_materials.json" if learning_path.exists(): with open(learning_path, 'r', encoding='utf-8') as f: materials["full_learning"] = json.load(f) # 3. 读取记忆反馈 feedback_path = SHARED_DIR / "memory_feedback.json" if feedback_path.exists(): with open(feedback_path, 'r', encoding='utf-8') as f: materials["feedback"] = json.load(f) return materials def create_workbuddy_learning_files(materials): """为 WorkBuddy 创建学习文件""" # 1. 创建技能目录结构 skill_dir = WORKBUDDY_LEARNING_DIR skill_dir.mkdir(parents=True, exist_ok=True) # 2. 创建简化的学习应用脚本 learning_script = '''#!/usr/bin/env python3 """ Hermes 学习材料应用脚本 - 简化版 """ import json import sys from pathlib import Path HERMES_SHARED = Path.home() / ".hermes" / "shared" def load_summary(): """加载学习摘要""" summary_path = HERMES_SHARED / "memory_summary.json" if summary_path.exists(): with open(summary_path, 'r', encoding='utf-8') as f: return json.load(f) return {} def apply_learnings(): """应用学习材料""" summary = load_summary() print("🚀 应用 Hermes 学习材料...") # 应用成功模式 for insight in summary.get("key_insights", []): if insight.get("title") == "成功任务模式": examples = insight.get("examples", []) print(f"✅ 学习 {len(examples)} 个成功案例") # 在实际应用中，这里会更新 WorkBuddy 的策略 print("🎉 学习材料应用完成！") def show_summary(): """显示摘要""" summary = load_summary() print("📚 Hermes 学习摘要") print("=" * 40) print(f"最后更新: {summary.get('last_update', '未知')}") print(f"总记忆: {summary.get('total_memories', 0)}") for insight in summary.get("key_insights", []): print(f"\\n{insight.get('title', '未知')}:") print(f" {insight.get('description', '')}") print(f" 示例: {len(insight.get('examples', []))}") if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "apply": apply_learnings() else: show_summary() ''' script_path = skill_dir / "apply_learning.py" with open(script_path, 'w', encoding='utf-8') as f: f.write(learning_script) # 3. 使脚本可执行 os.chmod(script_path, 0o755) # 4. 创建状态文件 sync_status = { "last_sync_time": datetime.now().isoformat(), "materials_count": len(materials), "summary_entries": materials.get('summary', {}).get('total_memories', 0), "key_insights": len(materials.get('summary', {}).get('key_insights', [])), "status": "synced" } status_path = skill_dir / "sync_status.json" with open(status_path, 'w', encoding='utf-8') as f: json.dump(sync_status, f, ensure_ascii=False, indent=2) return skill_dir, sync_status def sync_learning_materials(): """同步学习材料到 WorkBuddy""" ensure_dirs() # 1. 获取学习材料 materials = get_latest_learning_materials() if not materials: print("📭 未找到学习材料") return False print(f"📚 找到 {len(materials)} 类学习材料") # 2. 创建 WorkBuddy 学习文件 skill_dir, sync_status = create_workbuddy_learning_files(materials) # 3. 复制关键文件 print("📋 复制学习文件...") files_to_copy = ["memory_summary.json", "memory_feedback.json"] for file_name in files_to_copy: src = SHARED_DIR / file_name dst = skill_dir / file_name if src.exists(): shutil.copy2(src, dst) print(f" ✅ {file_name}") # 4. 记录桥接事件 update_bridge_event({ "type": "learning_sync", "materials_count": len(materials), "summary_entries": sync_status["summary_entries"], "key_insights": sync_status["key_insights"] }) print(f"\\n🎉 同步完成！") print(f"📁 位置: {skill_dir}") return True def get_learning_stats(): """获取学习材料统计""" status_path = WORKBUDDY_LEARNING_DIR / "sync_status.json" if status_path.exists(): with open(status_path, 'r', encoding='utf-8') as f: return json.load(f) # 返回默认值 return { "last_sync_time": "从未同步", "materials_count": 0, "summary_entries": 0, "key_insights": 0, "status": "not_synced" } def update_bridge_event(data): """更新桥接事件""" meta_path = SHARED_DIR / "meta.json" if meta_path.exists(): with open(meta_path, 'r', encoding='utf-8') as f: meta = json.load(f) else: meta = {"events": []} event = { "type": "learning_sync", "timestamp": datetime.now().isoformat(), **data } meta["events"].append(event) # 保留最近100条 meta["events"] = meta["events"][-100:] with open(meta_path, 'w', encoding='utf-8') as f: json.dump(meta, f, ensure_ascii=False, indent=2) def main(): """独立运行""" success = sync_learning_materials() if success: print("✅ Hermes 学习材料同步成功！") else: print("❌ 同步失败") if __name__ == "__main__": main() FILE:install_v2.sh #!/bin/bash # # hermes-memory-bridge / install_v2.sh # Hermes-WorkBuddy 事件驱动架构 v2.0 部署脚本 # # 用法： # bash install_v2.sh # 安装到 ~/.hermes/shared/（默认） # bash install_v2.sh /path/to # 指定 Hermes 根目录 # # 执行内容： # 1. 创建必要的目录结构 # 2. 设置文件权限 # 3. 配置 launchd 守护进程（macOS） # 4. 验证安装 # set -e HERMES_HOME="${1:-$HOME/.hermes}" SHARED_DIR="$HERMES_HOME/shared" SIGNAL_DIR="$SHARED_DIR/signals" QUEUE_DIR="$SHARED_DIR/queue" SKILL_DIR="$(cd "$(dirname "$0")" && pwd)" echo "========================================" echo " Hermes-WorkBuddy v2.0 安装脚本" echo "========================================" echo "Hermes 目录: $HERMES_HOME" echo "" # ─── 1. 创建目录 ─────────────────────────────────────────────────── echo "📁 创建目录结构..." mkdir -p "$SIGNAL_DIR" mkdir -p "$QUEUE_DIR" mkdir -p "$SHARED_DIR/feedback" mkdir -p "$HERMES_HOME/memories" echo " ✅ $SIGNAL_DIR" echo " ✅ $QUEUE_DIR" echo " ✅ $SHARED_DIR/feedback" echo " ✅ $HERMES_HOME/memories" # ─── 2. 复制/链接事件驱动模块到 Hermes 侧 ───────────────────────── echo "" echo "📦 部署事件驱动模块到 Hermes 侧..." cp "$SKILL_DIR/event_signaler.py" "$HERMES_HOME/event_signaler.py" cp "$SKILL_DIR/communication_queue.py" "$HERMES_HOME/communication_queue.py" cp "$SKILL_DIR/task_processor.py" "$HERMES_HOME/task_processor.py" cp "$SKILL_DIR/feedback_writer.py" "$HERMES_HOME/feedback_writer.py" chmod +x "$HERMES_HOME/event_signaler.py" echo " ✅ event_signaler.py → $HERMES_HOME/" echo " ✅ communication_queue.py → $HERMES_HOME/" echo " ✅ task_processor.py → $HERMES_HOME/" echo " ✅ feedback_writer.py → $HERMES_HOME/" # ─── 3. 创建 Hermes 环境变量配置（可选）──────────────────────────── echo "" echo "⚙️ 配置环境变量..." ENV_FILE="$HERMES_HOME/.env" if [ -f "$ENV_FILE" ]; then echo " ℹ️ $ENV_FILE 已存在，跳过" else cat > "$ENV_FILE" << 'EOF' # Hermes-WorkBuddy Bridge v2.0 环境变量 # 由 install_v2.sh 自动生成 # Hermes 根目录（通常无需修改） HERMES_HOME=~/.hermes # WorkBuddy 根目录 WORKBUDDY_HOME=~/WorkBuddy # 桥接日志级别：DEBUG | INFO | WARNING | ERROR BRIDGE_LOG_LEVEL=INFO # 信号/队列文件保留时长（秒） SIGNAL_TTL=21600 QUEUE_TTL=604800 EOF echo " ✅ $ENV_FILE" fi # ─── 4. 配置 launchd 守护进程（macOS）───────────────────────────── echo "" echo "🚀 配置守护进程（macOS launchd）..." LAUNCHD_DIR="$HOME/Library/LaunchAgents" PLIST_NAME="com.workbuddy.hermes-watcher.plist" PLIST_PATH="$LAUNCHD_DIR/$PLIST_NAME" mkdir -p "$LAUNCHD_DIR" cat > "$PLIST_PATH" << EOF Label com.workbuddy.hermes-watcher ProgramArguments /usr/bin/python3 ${SKILL_DIR}/event_watcher.py WorkingDirectory ${SHARED_DIR} EnvironmentVariables BRIDGE_LOG_LEVEL INFO RunAtLoad KeepAlive StandardOutPath /tmp/hermes-watcher.log StandardErrorPath /tmp/hermes-watcher.err ProcessType Background EOF echo " ✅ $PLIST_PATH" echo " ℹ️ 启动守护进程：launchctl load $PLIST_PATH" echo " ℹ️ 停止守护进程：launchctl unload $PLIST_PATH" # ─── 5. 验证安装 ─────────────────────────────────────────────────── echo "" echo "🔍 验证安装..." ERRORS=0 for f in "$SIGNAL_DIR" "$QUEUE_DIR" "$SHARED_DIR/feedback" "$HERMES_HOME/memories" \ "$HERMES_HOME/event_signaler.py" "$HERMES_HOME/communication_queue.py" \ "$HERMES_HOME/task_processor.py" "$HERMES_HOME/feedback_writer.py"; do if [ -e "$f" ]; then echo " ✅ $f" else echo " ❌ 缺失: $f" ERRORS=$((ERRORS + 1)) fi done # ─── 6. 完成 ────────────────────────────────────────────────────── echo "" echo "========================================" if [ $ERRORS -eq 0 ]; then echo " ✅ 安装完成！" else echo " ⚠️ 安装完成，但有 $ERRORS 个问题" fi echo "========================================" echo "" echo "下一步：" echo " 1. 启动守护进程（WorkBuddy 后台任务处理器）：" echo " launchctl load $PLIST_PATH" echo "" echo " 2. 测试信号发射（从 Hermes）：" echo " python3 $HERMES_HOME/event_signaler.py emit task_done '测试信号'" echo "" echo " 3. 测试命令任务（让 WorkBuddy 执行命令）：" echo " python3 $HERMES_HOME/event_signaler.py send_task echo '{\"message\":\"ping\"}'" echo "" echo " 4. 轮询 WorkBuddy 处理结果：" echo " python3 $HERMES_HOME/event_signaler.py feedback" echo "" echo " 5. 查看信号统计：" echo " python3 $HERMES_HOME/event_signaler.py stats" echo "" echo " 6. 查看 WorkBuddy 守护日志：" echo " tail -f /tmp/hermes-watcher.log" echo "" FILE:memory_writer.py """ hermes-memory-bridge / memory_writer.py 写入 Hermes 记忆文件（带健壮错误处理） """ from __future__ import annotations import json import logging import re from datetime import datetime from pathlib import Path from config import ( BRIDGE_META, ENTRY_DELIMITER, HERMES_MEMORIES_DIR, MAX_ENTRY_CHARS, MAX_EVENTS, SHARED_DIR, WORKBUDDY_LOG, _get_logger, ) logger = _get_logger("memory_writer") # ─── 内部工具 ─────────────────────────────────────────────────────── def _ensure_dir(path: Path) -> bool: """确保目录存在，返回是否成功""" try: path.parent.mkdir(parents=True, exist_ok=True) return True except PermissionError as e: logger.error(f"权限不足，无法创建目录 {path.parent}: {e}") return False except OSError as e: logger.error(f"创建目录失败 {path.parent}: {e}") return False def _safe_read(path: Path, default: str = "") -> str: """安全读取文件，失败时返回默认值""" try: return path.read_text(encoding="utf-8") except FileNotFoundError: return default except PermissionError as e: logger.warning(f"权限不足，无法读取 {path}: {e}") return default except OSError as e: logger.warning(f"读取文件失败 {path}: {e}") return default def _safe_write(path: Path, content: str) -> bool: """安全写入文件，返回是否成功""" try: path.write_text(content, encoding="utf-8") return True except PermissionError as e: logger.error(f"权限不足，无法写入 {path}: {e}") return False except OSError as e: logger.error(f"写入文件失败 {path}: {e}") return False def _sanitize(content: str) -> str: """去除潜在注入风险内容""" patterns = [ r"ignore\s+(previous|all|above|prior)\s+instructions", r"you\s+are\s+now\s+", r"system\s+prompt\s+override", r"<\|im_start\|>|<\|im_end\|>", ] for pat in patterns: content = re.sub(pat, "[FILTERED]", content, flags=re.IGNORECASE) return content[:MAX_ENTRY_CHARS] # ─── 公开接口 ─────────────────────────────────────────────────────── def append_hermes_memory( target: str, content: str, source: str = "WorkBuddy", ) -> str | None: """ 向 Hermes 的 MEMORY.md 或 USER.md 追加一条记忆条目。 Args: target: 'memory' | 'user' content: 要写入的内容 source: 来源标记（默认为 WorkBuddy） Returns: 写入的完整条目，或 None（失败时） """ if not _ensure_dir(HERMES_MEMORIES_DIR): return None fname = "MEMORY.md" if target == "memory" else "USER.md" fpath = HERMES_MEMORIES_DIR / fname timestamp = datetime.now().strftime("%Y-%m-%d") safe_content = _sanitize(content) entry = f"[{timestamp} · {source}]\n{safe_content}" existing = _safe_read(fpath) if existing.strip(): new_content = existing.rstrip() + ENTRY_DELIMITER + entry else: new_content = entry if _safe_write(fpath, new_content): logger.info(f"写入记忆: [{fname}] {timestamp} · {source}") return entry return None def write_shared_log( content: str, log_type: str = "workbuddy", ) -> Path | None: """ 写入共用互通日志（Hermes 可通过 on_delegation hook 读取）。 Args: content: 日志内容 log_type: 'workbuddy' | 'hermes' Returns: 写入的日志文件路径，或 None（失败时） """ if not _ensure_dir(SHARED_DIR): return None log_file = SHARED_DIR / f"{log_type}.log" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M") entry = f"[{timestamp}] {content}\n" existing = _safe_read(log_file) # 保留最近 500 行，防止日志无限增长 lines = (existing + entry).strip().split("\n") lines = lines[-500:] combined = "\n".join(lines) + "\n" if _safe_write(log_file, combined): logger.debug(f"写入日志: {log_file.name}") return log_file return None def write_bridge_event(event_type: str, data: dict) -> bool: """ 写入桥接元事件，供两边 Agent 在下次启动时读取。 Args: event_type: 'task_done' | 'config_change' | 'sync' | 'error' data: 事件附加数据 Returns: 是否写入成功 """ if not _ensure_dir(SHARED_DIR): return False meta_file = BRIDGE_META try: raw = _safe_read(meta_file) meta = json.loads(raw) if raw else {"events": []} except (json.JSONDecodeError, TypeError) as e: logger.warning(f"meta.json 格式损坏，重置: {e}") meta = {"events": []} event = { "type": event_type, "timestamp": datetime.now().isoformat(), **data, } meta["events"].append(event) # 保留最近 MAX_EVENTS 条 meta["events"] = meta["events"][-MAX_EVENTS:] if _safe_write(meta_file, json.dumps(meta, ensure_ascii=False, indent=2)): logger.debug(f"写入事件: {event_type}") return True return False def read_shared_events( event_type: str | None = None, limit: int = 20, ) -> list[dict]: """ 读取共用互通事件。 Args: event_type: 过滤特定类型，None 则返回全部 limit: 返回最近 N 条 Returns: 事件列表 """ meta_file = BRIDGE_META if not meta_file.exists(): return [] try: meta = json.loads(meta_file.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError) as e: logger.warning(f"读取 meta.json 失败: {e}") return [] events = meta.get("events", []) if event_type: events = [e for e in events if e.get("type") == event_type] return events[-limit:] def read_workbuddy_log(lines: int = 20) -> list[str]: """读取 WorkBuddy 写给 Hermes 的最新日志""" log_file = SHARED_DIR / "workbuddy.log" if not log_file.exists(): return [] try: all_lines = log_file.read_text(encoding="utf-8").strip().split("\n") return all_lines[-lines:] except (OSError, UnicodeDecodeError) as e: logger.warning(f"读取 workbuddy.log 失败: {e}") return [] FILE:queries.py """ hermes-memory-bridge / queries.py Hermes state.db 查询封装（带健壮错误处理） """ from __future__ import annotations import json import logging import sqlite3 from datetime import datetime, timedelta from typing import Any from config import HERMES_DB, _get_logger logger = _get_logger("queries") # ─── 异常类 ───────────────────────────────────────────────────────── class HermesDBError(Exception): """Hermes 数据库操作异常""" pass class MemoryFileError(Exception): """记忆文件读写异常""" pass # ─── 内部工具 ─────────────────────────────────────────────────────── def _row2dict(row: sqlite3.Row) -> dict: return dict(row) def _safe_connect(path: str) -> sqlite3.Connection: """建立数据库连接，带异常处理""" try: conn = sqlite3.connect(path, timeout=5.0) conn.row_factory = sqlite3.Row return conn except sqlite3.Error as e: logger.error(f"数据库连接失败: {e}") raise HermesDBError(f"无法连接数据库: {e}") from e # ─── 查询接口 ─────────────────────────────────────────────────────── def get_recent_sessions(days: int = 7, limit: int = 20) -> list[dict]: """返回最近 N 天的会话列表""" if not HERMES_DB.exists(): logger.debug("Hermes 数据库不存在，返回空列表") return [] try: cutoff = datetime.now() - timedelta(days=days) with _safe_connect(str(HERMES_DB)) as conn: cur = conn.execute( """ SELECT id, source, model, title, started_at, ended_at, end_reason, message_count, tool_call_count, estimated_cost_usd, actual_cost_usd FROM sessions WHERE started_at > ? ORDER BY started_at DESC LIMIT ? """, (cutoff.timestamp(), limit), ) rows = [_row2dict(r) for r in cur.fetchall()] logger.debug(f"查询会话: {len(rows)} 条（近 {days} 天）") return rows except HermesDBError: raise except Exception as e: logger.error(f"查询会话失败: {e}") return [] def get_session_messages(session_id: str, limit: int = 50) -> list[dict]: """返回指定会话的消息历史""" if not HERMES_DB.exists(): return [] try: with _safe_connect(str(HERMES_DB)) as conn: cur = conn.execute( """ SELECT role, content, tool_name, timestamp, finish_reason FROM messages WHERE session_id = ? ORDER BY timestamp ASC LIMIT ? """, (session_id, limit), ) return [_row2dict(r) for r in cur.fetchall()] except HermesDBError: raise except Exception as e: logger.error(f"查询消息历史失败（session={session_id}）: {e}") return [] def search_messages(keyword: str, days: int = 30) -> list[dict]: """全文模糊搜索 Hermes 会话""" if not HERMES_DB.exists(): return [] if not keyword or len(keyword) < 2: logger.debug("关键词过短，跳过搜索") return [] try: cutoff = datetime.now() - timedelta(days=days) with _safe_connect(str(HERMES_DB)) as conn: cur = conn.execute( """ SELECT m.session_id, m.role, m.content, m.tool_name, m.timestamp, s.title, s.source FROM messages m JOIN sessions s ON m.session_id = s.id WHERE m.timestamp > ? AND m.content LIKE ? ORDER BY m.timestamp DESC LIMIT 30 """, (cutoff.timestamp(), f"%{keyword}%"), ) rows = [_row2dict(r) for r in cur.fetchall()] logger.debug(f"搜索「{keyword}」: {len(rows)} 条匹配") return rows except HermesDBError: raise except Exception as e: logger.error(f"搜索消息失败: {e}") return [] def search_fts(keyword: str, limit: int = 20) -> list[dict]: """使用 FTS5 全文索引搜索（降级为 LIKE 搜索）""" if not HERMES_DB.exists(): return [] try: with _safe_connect(str(HERMES_DB)) as conn: cur = conn.execute( """ SELECT m.session_id, m.role, m.content, m.tool_name, s.title, s.started_at, highlight(messages_fts, 0, '**', '**') AS hl_content FROM messages_fts JOIN messages m ON messages_fts.rowid = m.id JOIN sessions s ON m.session_id = s.id WHERE messages_fts MATCH ? ORDER BY rank LIMIT ? """, (keyword, limit), ) return [_row2dict(r) for r in cur.fetchall()] except sqlite3.OperationalError: logger.debug("FTS5 表不可用，降级为 LIKE 搜索") return search_messages(keyword, 30) except HermesDBError: raise except Exception as e: logger.error(f"FTS 搜索失败: {e}") return [] def get_session_stats(days: int = 30) -> dict[str, Any]: """获取 Hermes 使用统计""" if not HERMES_DB.exists(): return {"period_days": days, "total_sessions": 0, "total_messages": 0, "total_tokens": 0, "estimated_cost_usd": 0.0} try: cutoff = datetime.now() - timedelta(days=days) with _safe_connect(str(HERMES_DB)) as conn: def count(sql: str, params: tuple) -> int: try: return conn.execute(sql, params).fetchone()[0] or 0 except Exception: return 0 total = count( "SELECT COUNT(*) FROM sessions WHERE started_at > ?", (cutoff.timestamp(),), ) total_messages = count( """ SELECT COUNT(*) FROM messages m JOIN sessions s ON m.session_id = s.id WHERE s.started_at > ? """, (cutoff.timestamp(),), ) total_tokens = count( "SELECT SUM(input_tokens + output_tokens) FROM sessions WHERE started_at > ?", (cutoff.timestamp(),), ) total_cost = count( """ SELECT SUM(actual_cost_usd) FROM sessions WHERE started_at > ? AND actual_cost_usd IS NOT NULL """, (cutoff.timestamp(),), ) stats = { "period_days": days, "total_sessions": total, "total_messages": total_messages, "total_tokens": total_tokens, "estimated_cost_usd": round(float(total_cost), 4), } logger.debug(f"统计: {stats}") return stats except HermesDBError: raise except Exception as e: logger.error(f"获取统计失败: {e}") return {"period_days": days, "total_sessions": 0, "total_messages": 0, "total_tokens": 0, "estimated_cost_usd": 0.0} def read_hermes_memory() -> dict[str, str]: """读取 Hermes 内置记忆文件（MEMORY.md / USER.md）""" from config import HERMES_MEMORIES_DIR result: dict[str, dict] = {} for fname in ("MEMORY.md", "USER.md"): fpath = HERMES_MEMORIES_DIR / fname if not fpath.exists(): result[fname] = {"entries": [], "raw": ""} continue try: content = fpath.read_text(encoding="utf-8") entries = [e.strip() for e in content.split("\n§\n") if e.strip()] result[fname] = {"entries": entries, "raw": content} logger.debug(f"读取 {fname}: {len(entries)} 条记忆") except (OSError, UnicodeDecodeError) as e: logger.warning(f"读取 {fpath} 失败: {e}") result[fname] = {"entries": [], "raw": ""} return result FILE:sync.py """ hermes-memory-bridge / sync.py WorkBuddy ↔ Hermes 双向同步引擎（带健壮错误处理） """ from __future__ import annotations import logging from datetime import datetime from pathlib import Path from typing import Any, Optional from config import ( BRIDGE_META, HERMES_DB, HERMES_MEMORIES_DIR, SHARED_DIR, WORKBUDDY_LOG, WORKBUDDY_MEMORY_DIR, _get_logger, ) logger = _get_logger("sync") # ─── 延迟导入（避免循环依赖）──────────────────────────────────────── # 注意：以下导入在运行时通过 bridge.py 的 import 链完成 def sync_workbuddy_to_hermes( work_summary: str, work_type: str = "task", tags: list[str] | None = None, ) -> dict[str, Any]: """ 将 WorkBuddy 完成的工作同步到 Hermes 记忆系统。三步操作（任一步失败均记录日志并继续）： 1. 写入 Hermes MEMORY.md 2. 写入共用互通日志 3. 写入桥接元事件 Returns: dict，含 status: 'synced' | 'partial' | 'failed' """ from memory_writer import ( append_hermes_memory, write_bridge_event, write_shared_log, ) tags = tags or [] result: dict[str, Any] = {"status": "failed", "entry": None, "log_path": None} success_count = 0 # ① 写入 Hermes MEMORY.md try: entry = append_hermes_memory( target="memory", content=work_summary, source="WorkBuddy", ) if entry: result["entry"] = entry success_count += 1 else: logger.error("写入 Hermes MEMORY.md 失败") except Exception as e: logger.error(f"写入 Hermes MEMORY.md 异常: {e}") # ② 写入共用日志 try: log_path = write_shared_log( f"[{work_type}] {work_summary}", log_type="workbuddy" ) if log_path: result["log_path"] = str(log_path) success_count += 1 else: logger.warning("写入共用日志失败") except Exception as e: logger.error(f"写入共用日志异常: {e}") # ③ 写入桥接元事件 try: write_bridge_event("task_done", { "summary": work_summary, "work_type": work_type, "tags": tags, }) success_count += 1 except Exception as e: logger.error(f"写入桥接事件异常: {e}") # 综合状态 if success_count == 3: result["status"] = "synced" elif success_count > 0: result["status"] = "partial" logger.warning(f"同步部分成功（{success_count}/3 步）") else: logger.error("同步全部失败") logger.info(f"sync_workbuddy_to_hermes: {result['status']} - {work_summary[:50]}") return result def sync_hermes_to_workbuddy_context(days: int = 7) -> dict[str, Any]: """ 将 Hermes 最近的重要上下文同步到 WorkBuddy 可读格式，用于在 WorkBuddy 启动时了解 Hermes 侧的最新动态。 """ from memory_writer import write_bridge_event # 延迟导入 queries from queries import ( get_recent_sessions, get_session_stats, read_hermes_memory, ) try: sessions = get_recent_sessions(days=days) stats = get_session_stats(days=days) hermes_mem = read_hermes_memory() except Exception as e: logger.error(f"拉取 Hermes 数据失败: {e}") return { "sessions": [], "stats": {}, "summary_text": f"[错误] 无法读取 Hermes 数据: {e}", "workbuddy_entries": [], } summary_lines = [ f"## Hermes 近 {days} 天动态", "", f"**会话数**: {stats.get('total_sessions', 0)}", f"**消息数**: {stats.get('total_messages', 0)}", "", "### 最近会话", ] for s in sessions[:5]: try: ts = datetime.fromtimestamp(s["started_at"]).strftime("%m-%d %H:%M") except (ValueError, KeyError, TypeError): ts = "未知时间" title = s.get("title") or s.get("source") or "无标题" summary_lines.append(f"- [{ts}] {title}") # Hermes 记忆中关于 WorkBuddy 的条目 wb_entries: list[str] = [] for entry in hermes_mem.get("MEMORY.md", {}).get("entries", []): if "WorkBuddy" in entry: wb_entries.append(entry) if wb_entries: summary_lines.extend(["", "### Hermes 中关于 WorkBuddy 的记忆"]) for e in wb_entries[-5:]: summary_lines.append(f"- {e[:200]}") summary_text = "\n".join(summary_lines) # 记录本次同步事件 try: write_bridge_event("sync", { "direction": "hermes_to_workbuddy", "days": days, "sessions_found": len(sessions), "workbuddy_entries_found": len(wb_entries), }) except Exception as e: logger.warning(f"写入同步事件失败: {e}") logger.debug(f"sync_hermes_to_workbuddy: {len(sessions)} sessions, {len(wb_entries)} WB entries") return { "sessions": sessions, "stats": stats, "summary_text": summary_text, "workbuddy_entries": wb_entries, } def search_both_memories(keyword: str, days: int = 30) -> dict[str, Any]: """ 跨 WorkBuddy 和 Hermes 记忆的全文搜索。 """ from queries import search_messages if not keyword or len(keyword) < 2: logger.debug("关键词过短，跳过搜索") return {"keyword": keyword, "hermes": [], "workbuddy": []} # Hermes 侧搜索 hermes_results: list[dict] = [] try: hermes_results = search_messages(keyword, days=days) except Exception as e: logger.warning(f"Hermes 搜索失败: {e}") # WorkBuddy 侧搜索（读日志文件） wb_results: list[dict] = [] try: wb_results = _search_workbuddy_memory(keyword) except Exception as e: logger.warning(f"WorkBuddy 记忆搜索失败: {e}") return { "keyword": keyword, "hermes": hermes_results, "workbuddy": wb_results, } def _search_workbuddy_memory(keyword: str) -> list[dict]: """在 WorkBuddy 记忆文件中搜索（安全读取）""" results: list[dict] = [] if WORKBUDDY_MEMORY_DIR is None: logger.debug("WORKBUDDY_MEMORY_DIR 未找到，跳过搜索") return results if not WORKBUDDY_MEMORY_DIR.exists(): logger.debug(f"WorkBuddy 记忆目录不存在: {WORKBUDDY_MEMORY_DIR}") return results try: md_files = list(WORKBUDDY_MEMORY_DIR.glob("*.md")) except PermissionError as e: logger.warning(f"权限不足，无法扫描 {WORKBUDDY_MEMORY_DIR}: {e}") return results except OSError as e: logger.warning(f"扫描 WorkBuddy 记忆目录失败: {e}") return results kw_lower = keyword.lower() for fpath in md_files: try: content = fpath.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError) as e: logger.warning(f"读取 {fpath} 失败: {e}") continue if kw_lower not in content.lower(): continue lines = content.split("\n") for i, line in enumerate(lines): if kw_lower in line.lower(): context = lines[max(0, i - 1):i + 3] snippet = " ... ".join(c.strip() for c in context if c.strip())[:200] results.append({ "file": fpath.name, "line": i + 1, "snippet": snippet, }) logger.debug(f"WorkBuddy 记忆搜索「{keyword}」: {len(results)} 条") return results def read_bridge_status() -> dict[str, Any]: """读取桥接状态总览（健壮版本）""" status: dict[str, Any] = { "hermes_memory_files": [], "shared_files": [], "recent_events": [], "db_exists": HERMES_DB.exists(), "workbuddy_memory_dir": str(WORKBUDDY_MEMORY_DIR) if WORKBUDDY_MEMORY_DIR else None, } # 确保目录存在（不抛异常） try: HERMES_MEMORIES_DIR.mkdir(parents=True, exist_ok=True) SHARED_DIR.mkdir(parents=True, exist_ok=True) except PermissionError: logger.warning("权限不足，无法创建 Hermes 目录") # 列出记忆文件 try: status["hermes_memory_files"] = [ f.name for f in HERMES_MEMORIES_DIR.glob("*.md") ] except OSError as e: logger.warning(f"无法列出记忆文件: {e}") # 列出共用文件 try: status["shared_files"] = [f.name for f in SHARED_DIR.glob("*") if f.is_file()] except OSError as e: logger.warning(f"无法列出共用文件: {e}") # 读取近期事件 from memory_writer import read_shared_events try: status["recent_events"] = read_shared_events(limit=10) except Exception as e: logger.warning(f"读取桥接事件失败: {e}") logger.debug(f"read_bridge_status: db={status['db_exists']}, " f"memories={len(status['hermes_memory_files'])}, " f"shared={len(status['shared_files'])}") return status FILE:task_processor.py """ hermes-memory-bridge / task_processor.py 任务处理器核心 — 接收 Hermes 命令，执行并回写结果支持的命令类型： search_memory — 搜索 WorkBuddy 记忆 sync_session — 同步会话记忆 create_task — 在滴答清单创建任务 complete_task — 标记任务完成 list_tasks — 列出滴答清单任务 research — 执行深度调研 write_note — 写入 IMA 笔记 ack — 确认收到信号（ACK） echo — 回显测试用法（作为模块导入）： from task_processor import process_command result = process_command(command_type, params) """ from __future__ import annotations import json import os import subprocess import sys import time from datetime import datetime from pathlib import Path from typing import Any, Optional SKILL_DIR = Path(__file__).parent sys.path.insert(0, str(SKILL_DIR)) try: from config import SHARED_DIR, _get_logger, WORKBUDDY_MEMORY_DIR except ImportError: SHARED_DIR = Path.home() / ".hermes" / "shared" import logging logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) _get_logger = lambda n: logging.getLogger(n) WORKBUDDY_MEMORY_DIR = None logger = _get_logger("task_processor") # ─── 路径 ────────────────────────────────────────────────────────── FEEDBACK_DIR = SHARED_DIR / "feedback" # WorkBuddy 结果回写目录 # ─── 工具函数 ────────────────────────────────────────────────────── def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _safe_write(path: Path, data: Any) -> bool: try: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return True except (PermissionError, OSError) as e: logger.error(f"写入失败 {path}: {e}") return False def _run_cmd(cmd: list[str], timeout: int = 30) -> dict[str, Any]: """通用命令执行""" try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, ) return { "success": result.returncode == 0, "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), "returncode": result.returncode, } except subprocess.TimeoutExpired: return {"success": False, "error": f"命令超时（{timeout}s）"} except FileNotFoundError as e: return {"success": False, "error": f"命令不存在: {e}"} except Exception as e: return {"success": False, "error": str(e)} # ─── 任务处理器映射 ──────────────────────────────────────────────── def _search_memory(params: dict) -> dict: """搜索 WorkBuddy 记忆目录（跨所有会话目录）""" keyword = params.get("keyword", "") if not keyword: return {"success": False, "error": "缺少 keyword 参数"} results: list[dict] = [] wb_root = Path.home() / "WorkBuddy" if not wb_root.exists(): return {"success": False, "error": f"找不到 WorkBuddy 根目录: {wb_root}"} # 收集所有会话的记忆目录（当前会话优先） memory_dirs: list[Path] = [] try: for d in wb_root.iterdir(): if not (d.is_dir() and d.name.isdigit() and len(d.name) >= 10): continue mem_dir = d / ".workbuddy" / "memory" if mem_dir.exists(): memory_dirs.append(mem_dir) except PermissionError: return {"success": False, "error": "权限不足，无法读取 WorkBuddy 目录"} if not memory_dirs: return {"success": False, "error": "找不到任何 WorkBuddy 记忆目录"} for search_dir in memory_dirs: try: for fpath in search_dir.rglob("*.md"): try: content = fpath.read_text(encoding="utf-8") if keyword in content: lines = content.split("\n") matched = [l.strip() for l in lines if keyword in l] results.append({ "dir": search_dir.parent.parent.name, # 会话ID "file": fpath.name, "snippet": matched[0][:200] if matched else content[:200], }) except (OSError, UnicodeDecodeError): continue except PermissionError: continue return { "success": True, "keyword": keyword, "count": len(results), "results": results[:10], } def _sync_session(params: dict) -> dict: """同步会话记忆（生成会话摘要写到 WorkBuddy 记忆）""" topic = params.get("topic", "通用会话") summary = params.get("summary", "") notes = params.get("notes", "") if not summary: return {"success": False, "error": "缺少 summary 参数"} memory_dir = WORKBUDDY_MEMORY_DIR if not memory_dir or not memory_dir.exists(): # 回退到当前最新会话目录 wb_root = Path.home() / "WorkBuddy" if wb_root.exists(): try: latest = max( [d for d in wb_root.iterdir() if d.is_dir() and d.name.isdigit()], key=lambda d: d.name, ) memory_dir = latest / ".workbuddy" / "memory" except (ValueError, PermissionError): pass if not memory_dir: return {"success": False, "error": "无法定位 WorkBuddy 记忆目录"} today = datetime.now().strftime("%Y-%m-%d") fname = memory_dir / f"{today}.md" entry = f"\n## [{_ts()}] {topic}\n\n{summary}\n" if notes: entry += f"\n**备注**: {notes}\n" try: FEEDBACK_DIR.mkdir(parents=True, exist_ok=True) existing = "" if fname.exists(): existing = fname.read_text(encoding="utf-8") fname.write_text(existing + entry, encoding="utf-8") return {"success": True, "file": str(fname), "entry_preview": summary[:100]} except PermissionError: return {"success": False, "error": "权限不足，无法写入记忆文件"} except Exception as e: return {"success": False, "error": str(e)} def _create_task(params: dict) -> dict: """在滴答清单创建任务""" title = params.get("title", "") if not title: return {"success": False, "error": "缺少 title 参数"} # 优先使用 ticktickpower skill try: result = _run_cmd([ sys.executable, "-c", f"from ticktickpower import TickTick; t = TickTick(); " f"print(t.add_task(title='{title}', project_id='5a4ba4bce775913530602288'))" ], timeout=15) if result.get("success"): try: return {"success": True, "task": json.loads(result["stdout"])} except json.JSONDecodeError: return {"success": True, "raw": result["stdout"]} return {"success": False, "error": result.get("error") or result.get("stderr") or "创建失败"} except Exception: pass # fallback：使用 dida365 web API 方式（简化为返回提示） return { "success": False, "error": "ticktickpower 未安装或不可用", "hint": "请在 WorkBuddy 中手动创建任务或确保 ticktickpower skill 已激活", "title": title, } def _complete_task(params: dict) -> dict: """标记滴答清单任务完成""" task_id = params.get("task_id", "") if not task_id: return {"success": False, "error": "缺少 task_id 参数"} try: result = _run_cmd([ sys.executable, "-c", f"from ticktickpower import TickTick; t = TickTick(); t.complete_task('{task_id}')" ], timeout=10) return {"success": result.get("success", False), "raw": result} except Exception as e: return {"success": False, "error": str(e)} def _list_tasks(params: dict) -> dict: """列出滴答清单今日任务""" try: result = _run_cmd([ sys.executable, "-c", "from ticktickpower import TickTick; t = TickTick(); print(t.list_tasks(limit=10))" ], timeout=15) if result.get("success"): try: tasks = json.loads(result["stdout"]) return {"success": True, "tasks": tasks, "count": len(tasks)} except json.JSONDecodeError: return {"success": True, "raw": result["stdout"]} return {"success": False, "error": result.get("error") or result.get("stderr")} except Exception as e: return {"success": False, "error": str(e)} def _ack(params: dict) -> dict: """确认信号（对 Hermes 发来的 ack 类型指令）""" signal_id = params.get("signal_id", "") message = params.get("message", "已确认") return { "success": True, "signal_id": signal_id, "message": message, "acknowledged_at": _ts(), } def _echo(params: dict) -> dict: """回显测试""" return { "success": True, "echo": params.get("message", "pong"), "received_at": _ts(), } def _unknown(params: dict) -> dict: """未知命令""" return { "success": False, "error": f"未知命令类型", "hint": "支持的命令类型：search_memory, sync_session, create_task, complete_task, list_tasks, ack, echo", } # ─── 命令注册表 ──────────────────────────────────────────────────── _COMMAND_HANDLERS: dict[str, callable] = { "search_memory": _search_memory, "sync_session": _sync_session, "create_task": _create_task, "complete_task": _complete_task, "list_tasks": _list_tasks, "ack": _ack, "echo": _echo, } # ─── 核心处理函数 ────────────────────────────────────────────────── def process_command(command_type: str, params: dict, signal_id: str = "") -> dict: """ 处理来自 Hermes 的命令。 Args: command_type: 命令类型（见上方注册表） params: 命令参数字典 signal_id: 对应的信号 ID（用于回写） Returns: 执行结果字典（包含 success、data、error 等字段） """ handler = _COMMAND_HANDLERS.get(command_type, _unknown) logger.info(f"处理命令 [{signal_id[:8] if signal_id else '?'}]: {command_type}") start = time.time() result: dict[str, Any] try: result = handler(params) except Exception as e: logger.error(f"命令 [{command_type}] 执行异常: {e}") result = {"success": False, "error": str(e)} elapsed_ms = round((time.time() - start) * 1000, 1) result["_meta"] = { "command": command_type, "signal_id": signal_id, "processed_at": _ts(), "elapsed_ms": elapsed_ms, "processor": "WorkBuddy", } logger.info( f"命令 [{command_type}] 完成: {'✅' if result.get('success') else '❌'} " f"({elapsed_ms}ms)" ) return result # ─── CLI 入口（手动触发） ───────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Hermes → WorkBuddy 任务处理器") parser.add_argument("command", help="命令类型") parser.add_argument("--params", default="{}", help="JSON 参数字符串") parser.add_argument("--signal-id", default="", help="关联信号 ID") args = parser.parse_args() try: params = json.loads(args.params) except json.JSONDecodeError: print(f"参数 JSON 解析失败: {args.params}") sys.exit(1) result = process_command(args.command, params, args.signal_id) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0 if result.get("success") else 1) FILE:task_processor_extended.py """ hermes-memory-bridge / task_processor_extended.py 任务处理器扩展版 — 添加天气查询功能支持的命令类型： search_memory — 搜索 WorkBuddy 记忆 sync_session — 同步会话记忆 create_task — 在滴答清单创建任务 complete_task — 标记任务完成 list_tasks — 列出滴答清单任务 research — 执行深度调研 write_note — 写入 IMA 笔记 ack — 确认收到信号（ACK） echo — 回显测试 weather_query — 查询天气（新增）用法（作为模块导入）： from task_processor_extended import process_command result = process_command(command_type, params) """ from __future__ import annotations import json import os import subprocess import sys import time from datetime import datetime from pathlib import Path from typing import Any, Optional SKILL_DIR = Path(__file__).parent sys.path.insert(0, str(SKILL_DIR)) try: from config import SHARED_DIR, _get_logger, WORKBUDDY_MEMORY_DIR except ImportError: SHARED_DIR = Path.home() / ".hermes" / "shared" import logging logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) _get_logger = lambda n: logging.getLogger(n) WORKBUDDY_MEMORY_DIR = None logger = _get_logger("task_processor_extended") # ─── 路径 ────────────────────────────────────────────────────────── FEEDBACK_DIR = SHARED_DIR / "feedback" # WorkBuddy 结果回写目录 # ─── 工具函数 ────────────────────────────────────────────────────── def _ts() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def _safe_read(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError, OSError): return default def _safe_write(path: Path, data: Any) -> bool: try: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") return True except (PermissionError, OSError) as e: logger.error(f"写入失败 {path}: {e}") return False def _run_cmd(cmd: list[str], timeout: int = 30) -> dict[str, Any]: """通用命令执行""" try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout, ) return { "success": result.returncode == 0, "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), "returncode": result.returncode, } except subprocess.TimeoutExpired: return {"success": False, "error": f"命令超时（{timeout}s）"} except FileNotFoundError as e: return {"success": False, "error": f"命令不存在: {e}"} except Exception as e: return {"success": False, "error": str(e)} # ─── 任务处理器映射 ──────────────────────────────────────────────── def _search_memory(params: dict) -> dict: """搜索 WorkBuddy 记忆目录（跨所有会话目录）""" keyword = params.get("keyword", "") if not keyword: return {"success": False, "error": "缺少 keyword 参数"} # 导入原始模块中的函数 try: from task_processor import _search_memory as original_search_memory return original_search_memory(params) except ImportError: # 如果无法导入，返回简化版本 return {"success": False, "error": "无法导入原始搜索函数"} def _sync_session(params: dict) -> dict: """同步会话摘要到 WorkBuddy 记忆""" topic = params.get("topic", "") summary = params.get("summary", "") if not topic or not summary: return {"success": False, "error": "缺少 topic 或 summary 参数"} try: from task_processor import _sync_session as original_sync_session return original_sync_session(params) except ImportError: return {"success": False, "error": "无法导入原始同步函数"} def _create_task(params: dict) -> dict: """在滴答清单创建任务""" title = params.get("title", "") if not title: return {"success": False, "error": "缺少 title 参数"} try: from task_processor import _create_task as original_create_task return original_create_task(params) except ImportError: return {"success": False, "error": "无法导入原始创建任务函数"} def _complete_task(params: dict) -> dict: """标记滴答清单任务完成""" task_id = params.get("task_id", "") if not task_id: return {"success": False, "error": "缺少 task_id 参数"} try: from task_processor import _complete_task as original_complete_task return original_complete_task(params) except ImportError: return {"success": False, "error": "无法导入原始完成任务函数"} def _list_tasks(params: dict) -> dict: """列出滴答清单今日任务""" try: from task_processor import _list_tasks as original_list_tasks return original_list_tasks(params) except ImportError: return {"success": False, "error": "无法导入原始列出任务函数"} def _ack(params: dict) -> dict: """确认信号（对 Hermes 发来的 ack 类型指令）""" signal_id = params.get("signal_id", "") message = params.get("message", "已确认") return { "success": True, "signal_id": signal_id, "message": message, "acknowledged_at": _ts(), } def _echo(params: dict) -> dict: """回显测试""" return { "success": True, "echo": params.get("message", "pong"), "received_at": _ts(), } def _weather_query(params: dict) -> dict: """查询天气""" location = params.get("location", "Beijing") format_type = params.get("format", "3") try: # 使用wttr.in查询天气 cmd = ["curl", "-s", f"wttr.in/{location}?format={format_type}"] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: return { "success": True, "location": location, "weather": result.stdout.strip(), "source": "wttr.in" } else: return { "success": False, "error": f"天气查询失败: {result.stderr.strip()}", "location": location } except subprocess.TimeoutExpired: return {"success": False, "error": "查询超时", "location": location} except Exception as e: return {"success": False, "error": str(e), "location": location} def _unknown(params: dict) -> dict: """未知命令""" return { "success": False, "error": f"未知命令类型", "hint": "支持的命令类型：search_memory, sync_session, create_task, complete_task, list_tasks, ack, echo, weather_query", } # ─── 命令注册表 ──────────────────────────────────────────────────── _COMMAND_HANDLERS: dict[str, callable] = { "search_memory": _search_memory, "sync_session": _sync_session, "create_task": _create_task, "complete_task": _complete_task, "list_tasks": _list_tasks, "ack": _ack, "echo": _echo, "weather_query": _weather_query, } # ─── 核心处理函数 ────────────────────────────────────────────────── def process_command(command_type: str, params: dict, signal_id: str = "") -> dict: """ 处理来自 Hermes 的命令。 Args: command_type: 命令类型（见上方注册表） params: 命令参数字典 signal_id: 对应的信号 ID（用于回写） Returns: 执行结果字典（包含 success、data、error 等字段） """ handler = _COMMAND_HANDLERS.get(command_type, _unknown) logger.info(f"处理命令 [{signal_id[:8] if signal_id else '?'}]: {command_type}") start = time.time() result: dict[str, Any] try: result = handler(params) except Exception as e: logger.error(f"命令 [{command_type}] 执行异常: {e}") result = {"success": False, "error": str(e)} elapsed_ms = round((time.time() - start) * 1000, 1) result["_meta"] = { "command": command_type, "signal_id": signal_id, "processed_at": _ts(), "elapsed_ms": elapsed_ms, "processor": "WorkBuddy", } logger.info( f"命令 [{command_type}] 完成: {'✅' if result.get('success') else '❌'} " f"({elapsed_ms}ms)" ) return result # ─── CLI 入口（手动触发） ───────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Hermes → WorkBuddy 任务处理器（扩展版）") parser.add_argument("command", help="命令类型") parser.add_argument("--params", default="{}", help="JSON 参数字符串") parser.add_argument("--signal-id", default="", help="信号 ID（可选）") args = parser.parse_args() try: params = json.loads(args.params) except json.JSONDecodeError: print(f"❌ JSON 解析失败: {args.params}", file=sys.stderr) sys.exit(1) result = process_command(args.command, params, args.signal_id) print(json.dumps(result, ensure_ascii=False, indent=2))

Hermes Memory Bridge

Comments (0)