107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
"""Governance timeline dashboard.
|
|
|
|
Turns governance events into a human-readable flow for Web UI so operators can
|
|
see why a change is blocked or ready for promotion.
|
|
"""
|
|
|
|
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()
|
|
|
|
|
|
EVENT_ORDER = {
|
|
"requirement_changed": 10,
|
|
"architecture_review_started": 20,
|
|
"architecture_review_completed": 30,
|
|
"test_gate_completed": 40,
|
|
"user_acceptance_completed": 50,
|
|
"version_decision_created": 60,
|
|
"owner_gate_completed": 70,
|
|
"promotion_completed": 80,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GovernanceTimelineStep:
|
|
id: str
|
|
kind: str
|
|
title: str
|
|
actor: str
|
|
status: str
|
|
timestamp: str
|
|
correlation_id: str | None = None
|
|
detail: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class GovernanceTimelineDashboard:
|
|
def build(self, events: list[dict[str, Any]] | None = None) -> dict[str, Any]:
|
|
events = events or []
|
|
steps = [self._event_to_step(event) for event in events]
|
|
steps.sort(key=lambda step: (EVENT_ORDER.get(step.kind, 999), step.timestamp))
|
|
blockers = [step for step in steps if step.status in {"blocked", "fail", "attention_required"}]
|
|
return {
|
|
"module": "governance-timeline",
|
|
"route": "/admin/ai-evolution/governance/timeline",
|
|
"title": "治理事件时间线",
|
|
"status": "attention_required" if blockers else "healthy",
|
|
"generated_at": utc_now(),
|
|
"steps": [step.to_dict() for step in steps],
|
|
"blockers": [step.to_dict() for step in blockers],
|
|
"flow": self._flow_summary(steps),
|
|
}
|
|
|
|
def _event_to_step(self, event: dict[str, Any]) -> GovernanceTimelineStep:
|
|
payload = event.get("payload", {})
|
|
kind = event.get("kind", "unknown")
|
|
return GovernanceTimelineStep(
|
|
id=event.get("id", f"event-{len(str(event))}"),
|
|
kind=kind,
|
|
title=self._title(kind, payload),
|
|
actor=event.get("source", payload.get("actor", "unknown")),
|
|
status=payload.get("status") or payload.get("decision") or event.get("status", "ok"),
|
|
timestamp=event.get("created_at") or payload.get("created_at") or utc_now(),
|
|
correlation_id=event.get("correlation_id") or payload.get("change_id") or payload.get("id"),
|
|
detail=payload,
|
|
)
|
|
|
|
def _title(self, kind: str, payload: dict[str, Any]) -> str:
|
|
titles = {
|
|
"requirement_changed": "需求变化",
|
|
"architecture_review_started": "架构审查开始",
|
|
"architecture_review_completed": "架构审查完成",
|
|
"test_gate_completed": "测试 Gate 完成",
|
|
"user_acceptance_completed": "用户验收完成",
|
|
"version_decision_created": "版本建议产生",
|
|
"owner_gate_completed": "Owner Gate 完成",
|
|
"promotion_completed": "Promotion 完成",
|
|
}
|
|
return payload.get("title") or payload.get("summary") or titles.get(kind, kind)
|
|
|
|
def _flow_summary(self, steps: list[GovernanceTimelineStep]) -> dict[str, Any]:
|
|
present = {step.kind for step in steps}
|
|
required = [
|
|
"requirement_changed",
|
|
"architecture_review_completed",
|
|
"test_gate_completed",
|
|
"user_acceptance_completed",
|
|
"version_decision_created",
|
|
]
|
|
missing = [kind for kind in required if kind not in present]
|
|
return {
|
|
"required_steps": required,
|
|
"missing_steps": missing,
|
|
"ready_for_owner_gate": not missing,
|
|
}
|
|
|
|
|
|
__all__ = ["GovernanceTimelineStep", "GovernanceTimelineDashboard"]
|