Add visual operations flow graph model for AIOS Web UI

2026-07-08 21:00:33 +08:00
parent b866bb8faf
commit 9a0b113160
1 changed files with 134 additions and 0 deletions
@@ -0,0 +1,134 @@
+"""Visual operations flow model for AIOS Web UI.
+
+Provides graph/timeline/trend data so the admin UI can show system behavior as
+flows and operational diagrams instead of only numeric tables.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@dataclass(frozen=True)
+class FlowNode:
+    id: str
+    label: str
+    node_type: str
+    status: str = "ok"
+    group: str = "default"
+    detail: str = ""
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class FlowEdge:
+    source: str
+    target: str
+    label: str
+    status: str = "ok"
+    count: int = 0
+    latency_ms: int | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class TimelineEvent:
+    id: str
+    title: str
+    event_type: str
+    status: str
+    occurred_at: str
+    summary: str = ""
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class TrendPoint:
+    timestamp: str
+    values: dict[str, Any]
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+class AIOSVisualOpsFlow:
+    """Builds visual workflow data for `/admin/ai-evolution` dashboards."""
+
+    def build_flow(self, runtime: dict[str, Any] | None = None, counters: dict[str, int] | None = None) -> dict[str, Any]:
+        runtime = runtime or {}
+        counters = counters or {}
+        nodes = [
+            FlowNode("event", "事件/命令输入", "input", group="ingest"),
+            FlowNode("watcher", "Command Watcher", "service", runtime.get("watcher_status", "ok"), "control"),
+            FlowNode("controller", "Autonomous Controller", "service", runtime.get("controller_status", "ok"), "control"),
+            FlowNode("diagnosis", "诊断/策略判断", "process", group="analysis"),
+            FlowNode("patch", "Patch 生成", "process", group="gitops"),
+            FlowNode("owner_gate", "Owner Gate", "gate", runtime.get("owner_gate_status", "ok"), "governance"),
+            FlowNode("git", "Git Archive", "store", runtime.get("git_status", "ok"), "gitops"),
+            FlowNode("audit", "Audit Plane", "store", runtime.get("audit_status", "ok"), "audit"),
+            FlowNode("ui", "Web UI 可视化", "output", group="visibility"),
+            FlowNode("notify", "通知/日报", "output", runtime.get("notification_status", "ok"), "visibility"),
+        ]
+        edges = [
+            FlowEdge("event", "watcher", "发现命令", count=counters.get("commands_seen", 0)),
+            FlowEdge("watcher", "controller", "触发自治循环", count=counters.get("controller_cycles", 0)),
+            FlowEdge("controller", "diagnosis", "健康检查/差异检测", count=counters.get("health_checks", 0)),
+            FlowEdge("diagnosis", "patch", "生成修复计划", count=counters.get("patch_candidates", 0)),
+            FlowEdge("patch", "owner_gate", "提交审批", count=counters.get("pending_owner_gate", 0)),
+            FlowEdge("owner_gate", "git", "批准后归档", count=counters.get("archived_changes", 0)),
+            FlowEdge("git", "audit", "写入审计", count=counters.get("audit_events", 0)),
+            FlowEdge("audit", "ui", "展示证据", count=counters.get("ui_refresh", 0)),
+            FlowEdge("audit", "notify", "汇报状态", count=counters.get("notifications", 0)),
+        ]
+        return {
+            "type": "flow_graph",
+            "layout": "left_to_right",
+            "generated_at": utc_now(),
+            "nodes": [n.to_dict() for n in nodes],
+            "edges": [e.to_dict() for e in edges],
+        }
+
+    def build_timeline(self, events: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+        if not events:
+            events = [
+                {"id": "baseline", "title": "Baseline V1R3C1 生效", "event_type": "release", "status": "ok", "summary": "自治系统 baseline 已晋升"},
+                {"id": "watcher", "title": "Command Watcher 心跳", "event_type": "heartbeat", "status": "ok", "summary": "最近周期正常"},
+                {"id": "audit", "title": "GitOps 差异巡检", "event_type": "audit", "status": "ok", "summary": "无未归档变更"},
+            ]
+        timeline = [
+            TimelineEvent(
+                id=e.get("id", f"event-{idx}"),
+                title=e.get("title", "event"),
+                event_type=e.get("event_type", "generic"),
+                status=e.get("status", "ok"),
+                occurred_at=e.get("occurred_at", utc_now()),
+                summary=e.get("summary", ""),
+            ).to_dict()
+            for idx, e in enumerate(events)
+        ]
+        return {"type": "timeline", "events": timeline, "generated_at": utc_now()}
+
+    def build_trends(self, points: list[dict[str, Any]] | None = None) -> dict[str, Any]:
+        points = points or [
+            {"timestamp": utc_now(), "values": {"health_score": 100, "drift_count": 0, "pending_owner_gate": 0, "failed_commands": 0}}
+        ]
+        return {
+            "type": "trend_series",
+            "series": [TrendPoint(p["timestamp"], p.get("values", {})).to_dict() for p in points],
+            "recommended_charts": ["health_score_line", "drift_bar", "owner_gate_queue", "failure_rate"],
+        }
+
+
+__all__ = ["FlowNode", "FlowEdge", "TimelineEvent", "TrendPoint", "AIOSVisualOpsFlow"]