"""Inbox JSONL Watcher — 秒级事件推送 Agent 完成任务后写 JSONL 到 inbox/daemon.jsonl，Inbox watcher 秒级消费。 用 truncate（清空内容但不删除文件），避免并发写入时文件不存在。 支持的事件类型： - agent_output: Agent 写入产出 - agent_claim: Agent 认领任务 - agent_status: Agent 状态变更（working/done/failed） - agent_heartbeat: Agent 心跳 """ from __future__ import annotations import asyncio import json import logging from pathlib import Path from typing import Any, Callable, Coroutine, Dict, List, Optional logger = logging.getLogger("moziplus-v2.inbox") class InboxWatcher: """Inbox JSONL 文件监听器""" def __init__( self, inbox_path: Path, process_callback: Optional[Callable[[ Dict[str, Any]], Coroutine[Any, Any, None]]] = None, watch_interval: float = 1.0, ): """ Args: inbox_path: daemon.jsonl 文件路径 process_callback: 收到事件后的异步回调 watch_interval: 轮询间隔秒数 """ self.inbox_path = inbox_path self.process_callback = process_callback self.watch_interval = watch_interval self._running: bool = False self._task: Optional[asyncio.Task] = None self._total_processed: int = 0 self._total_errors: int = 0 # ------------------------------------------------------------------ # 生命周期 # ------------------------------------------------------------------ async def start(self) -> None: """启动监听""" if self._running: return self._running = True self._task = asyncio.create_task(self._loop()) logger.info("Inbox watcher started (path=%s, interval=%.1fs)", self.inbox_path, self.watch_interval) async def stop(self) -> None: """停止监听""" self._running = False if self._task: self._task.cancel() try: await self._task except asyncio.CancelledError: pass logger.info("Inbox watcher stopped (processed=%d, errors=%d)", self._total_processed, self._total_errors) @property def is_running(self) -> bool: return self._running @property def total_processed(self) -> int: return self._total_processed @property def total_errors(self) -> int: return self._total_errors # ------------------------------------------------------------------ # 主循环 # ------------------------------------------------------------------ async def _loop(self) -> None: while self._running: try: await self.poll() except asyncio.CancelledError: raise except Exception: logger.exception("Inbox poll error") self._total_errors += 1 try: await asyncio.sleep(self.watch_interval) except asyncio.CancelledError: return # ------------------------------------------------------------------ # 消费逻辑 # ------------------------------------------------------------------ async def poll(self) -> List[Dict[str, Any]]: """读取并处理所有新事件，然后 truncate 文件 Returns: 本次处理的事件列表 """ events = self._read_events() if not events: return [] processed = [] for event in events: try: if self.process_callback: await self.process_callback(event) processed.append(event) self._total_processed += 1 except Exception: logger.exception("Error processing inbox event: %s", json.dumps(event, default=str)) self._total_errors += 1 # 继续处理后续事件 # truncate（清空内容，不删除文件） self._truncate() return processed def _read_events(self) -> List[Dict[str, Any]]: """读取 JSONL 文件中的所有有效行""" if not self.inbox_path.exists(): return [] try: content = self.inbox_path.read_text(encoding="utf-8").strip() except OSError: return [] if not content: return [] events: List[Dict[str, Any]] = [] for line_no, line in enumerate(content.split("\n"), 1): line = line.strip() if not line: continue try: event = json.loads(line) if isinstance(event, dict): events.append(event) else: logger.warning("Inbox line %d: expected dict, got %s", line_no, type(event).__name__) self._total_errors += 1 except json.JSONDecodeError: logger.warning( "Inbox line %d: invalid JSON, skipping", line_no) self._total_errors += 1 return events def _truncate(self) -> None: """清空文件内容（不删除文件）""" try: with open(self.inbox_path, "w", encoding="utf-8"): pass # "w" mode already truncates except OSError: logger.exception("Failed to truncate inbox: %s", self.inbox_path) # ------------------------------------------------------------------ # 写入（供 Agent / 测试使用） # ------------------------------------------------------------------ @staticmethod def write_event(inbox_path: Path, event: Dict[str, Any]) -> None: """原子写入一个事件到 JSONL 文件 使用 "追加" 模式写入，多个 Agent 并发写安全（OS 保证 append 原子性 对于小行 < PIPE_BUF，通常 512-4096 字节） """ inbox_path.parent.mkdir(parents=True, exist_ok=True) line = json.dumps(event, ensure_ascii=False) + "\n" with open(inbox_path, "a", encoding="utf-8") as f: f.write(line) @staticmethod def write_events(inbox_path: Path, events: List[Dict[str, Any]]) -> None: """批量写入事件""" inbox_path.parent.mkdir(parents=True, exist_ok=True) with open(inbox_path, "a", encoding="utf-8") as f: for event in events: f.write(json.dumps(event, ensure_ascii=False) + "\n") # --------------------------------------------------------------------------- # 事件处理回调 — 将 Inbox 事件翻译为黑板操作 # --------------------------------------------------------------------------- async def default_inbox_callback( event: Dict[str, Any], registry_root: Path, initialized_dbs: set, ) -> None: """默认回调：根据事件类型更新黑板""" from src.blackboard.db import init_db, get_connection event_type = event.get("type", "") project_id = event.get("project_id", "") task_id = event.get("task_id") agent = event.get("agent", "unknown") if not project_id: logger.warning("Inbox event missing project_id: %s", event_type) return db_path = registry_root / project_id / "blackboard.db" db_key = str(db_path) # 确保 DB 已初始化 if db_key not in initialized_dbs: init_db(db_path) initialized_dbs.add(db_key) if event_type == "agent_output": # Agent 写入产出 → 写 outputs 表 + 事件 conn = get_connection(db_path) try: conn.execute("BEGIN IMMEDIATE") conn.execute( "INSERT INTO outputs (task_id, agent, output_type, title, summary, metadata) " "VALUES (?,?,?,?,?,?)", (task_id, agent, event.get("output_type", "other"), event.get("title", "output"), event.get("summary"), json.dumps(event.get("metadata", {}))), ) conn.execute( "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)", (task_id, agent, "output_written", json.dumps({"type": event.get("output_type")})), ) conn.commit() finally: conn.close() elif event_type == "agent_status": # Agent 状态变更 → 通过 Blackboard 状态机校验 new_status = event.get("status") if new_status and task_id: from src.blackboard.operations import Blackboard bb = Blackboard(db_path) ok = bb.update_task_status(task_id, new_status, agent=agent) if not ok: logger.warning("Inbox: invalid status transition to '%s' for task %s", new_status, task_id) elif event_type == "agent_claim": # Agent 认领任务 if task_id: conn = get_connection(db_path) try: conn.execute("BEGIN IMMEDIATE") cursor = conn.execute( "UPDATE tasks SET status='claimed', assignee=?, claimed_at=datetime('now'), " "updated_at=datetime('now') WHERE id=? AND status='pending'", (agent, task_id), ) if cursor.rowcount > 0: conn.execute( "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)", (task_id, agent, "task_claimed", json.dumps({"by": agent})), ) conn.commit() finally: conn.close() elif event_type == "agent_heartbeat": # Agent 心跳 → 更新 agents 表 conn = get_connection(db_path) try: conn.execute("BEGIN IMMEDIATE") conn.execute( "INSERT OR REPLACE INTO agents (agent_id, role, current_status, last_active) " "VALUES (?,?,?,datetime('now'))", (agent, event.get("role", ""), event.get("agent_status", "busy")), ) conn.execute( "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)", (task_id, agent, "daemon_tick", json.dumps({"heartbeat": True})), ) conn.commit() finally: conn.close() else: logger.debug("Unhandled inbox event type: %s", event_type)