"""Admin dashboard data model for the unified Agent Event Bus.""" from __future__ import annotations from collections import Counter from typing import Any from .agent_event_bus import AgentEventBus class AgentEventDashboard: def __init__(self, bus: AgentEventBus | None = None): self.bus = bus or AgentEventBus() def overview(self) -> dict[str, Any]: stats = self.bus.stats() pending = self.bus.list_events(status="pending") running = self.bus.list_events(status="running") completed = self.bus.list_events(status="completed") failed = self.bus.list_events(status="failed") all_visible = pending + running + completed + failed by_kind = Counter(event.get("kind", "unknown") for event in all_visible) by_target = Counter(event.get("target", "unknown") for event in all_visible) return { "module": "agent-event-bus", "title": "Unified Agent Event Bus", "route": "/admin/ai-evolution/events", "metrics": { **stats, "by_kind": dict(by_kind), "by_target": dict(by_target), "gpt_inbox_pending": by_kind.get("gpt_inbox", 0), "openclaw_pending": len([e for e in pending if e.get("target") == "openclaw"]), }, "panels": [ "event timeline", "pending commands", "GPT inbox", "agent replies", "failure reports", "audit trail", ], "audit": { "enabled": True, "location": "agents/events/audit", }, } __all__ = ["AgentEventDashboard"]