"""Unified Agent Event Bus. Unifies commands, GPT messages, replies, approvals, audit and reports into one state machine. Compatibility layers can filter events by kind instead of keeping separate queues. """ from __future__ import annotations from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from hashlib import sha256 from pathlib import Path import json from typing import Any def utc_now() -> str: return datetime.now(timezone.utc).isoformat() @dataclass class AgentEvent: id: str kind: str source: str target: str summary: str payload: dict[str, Any] status: str = "pending" correlation_id: str | None = None created_at: str = field(default_factory=utc_now) updated_at: str = field(default_factory=utc_now) actor: str | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) class AgentEventBus: STATES = ("pending", "running", "completed", "failed") def __init__(self, root: str = "agents/events"): self.root = Path(root) for state in self.STATES: (self.root / state).mkdir(parents=True, exist_ok=True) (self.root / "audit").mkdir(parents=True, exist_ok=True) def publish( self, kind: str, source: str, target: str, summary: str, payload: dict[str, Any], correlation_id: str | None = None, ) -> dict[str, Any]: event_id = "evt_" + sha256( json.dumps( {"kind": kind, "source": source, "summary": summary, "payload": payload}, sort_keys=True, ensure_ascii=False, ).encode("utf-8") ).hexdigest()[:16] event = AgentEvent(event_id, kind, source, target, summary, payload, correlation_id=correlation_id) self._write("pending", event.to_dict()) self._audit("created", event.to_dict()) return event.to_dict() def list_events(self, kind: str | None = None, status: str = "pending", limit: int = 100) -> list[dict[str, Any]]: result = [] for path in sorted((self.root / status).glob("*.json")): event = self._read(path) if kind and event.get("kind") != kind: continue result.append(event) return result[:limit] def get(self, event_id: str) -> dict[str, Any] | None: for state in self.STATES: path = self.root / state / f"{event_id}.json" if path.exists(): return self._read(path) return None def transition(self, event_id: str, status: str, actor: str | None = None) -> dict[str, Any]: event = self.get(event_id) if not event: return {"status": "not_found", "id": event_id} if status not in self.STATES: return {"status": "invalid_state"} old = event.get("status", "pending") old_path = self.root / old / f"{event_id}.json" event["status"] = status event["actor"] = actor event["updated_at"] = utc_now() self._write(status, event) if old_path.exists() and old != status: old_path.unlink() self._audit("transition", event, {"from": old, "to": status}) return event def stats(self) -> dict[str, Any]: return { "events": {state: len(list((self.root / state).glob("*.json"))) for state in self.STATES}, "audit_events": len(list((self.root / "audit").glob("*.json"))), "status": "healthy", } def _write(self, state: str, data: dict[str, Any]): (self.root / state / f"{data['id']}.json").write_text( json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" ) def _read(self, path: Path): return json.loads(path.read_text(encoding="utf-8")) def _audit(self, action: str, event: dict[str, Any], extra: dict[str, Any] | None = None): data = {"action": action, "event": event, "extra": extra or {}, "time": utc_now()} (self.root / "audit" / f"{event['id']}-{action}.json").write_text( json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" ) __all__ = ["AgentEvent", "AgentEventBus"]