""" Governance API Models — /governance 頁面 Pydantic Schemas ========================================================= PR 1 後端 3 endpoint 的 request/response schema. Endpoints: GET /api/v1/ai/governance/events — ai_governance_events 查詢 GET /api/v1/ai/governance/queue — governance_remediation_dispatch 隊列（Track D 依賴表） GET /api/v1/ai/governance/summary — 30d SLO 違反時序統計 設計原則: - Pydantic V2，對齊 models/ 目錄 - Severity 映射邏輯集中於此，Router / Service 直接用 - 禁止硬編碼 IP 或內網位址 2026-05-02 ogt + Claude Sonnet 4.6 Asia/Taipei """ from __future__ import annotations from datetime import datetime from typing import Literal from pydantic import BaseModel, Field # ============================================================================= # Severity 映射 # ============================================================================= # critical: slo_violation / conservative_mode / governance_slo_data_gap # warning: trust_drift / kb_stale / knowledge_degradation / execution_blast_radius # info: 其他（含 replay_degraded / self_demotion / llm_hallucination 等） _CRITICAL_TYPES: frozenset[str] = frozenset({ "slo_violation", "conservative_mode", "governance_slo_data_gap", }) _WARNING_TYPES: frozenset[str] = frozenset({ "trust_drift", "kb_stale", "knowledge_degradation", "execution_blast_radius", }) def map_severity(event_type: str) -> Literal["critical", "warning", "info"]: """將 event_type 映射為 severity 等級.""" if event_type in _CRITICAL_TYPES: return "critical" if event_type in _WARNING_TYPES: return "warning" return "info" # ============================================================================= # Endpoint 1: events # ============================================================================= class GovernanceEvent(BaseModel): id: str event_type: str severity: Literal["critical", "warning", "info"] triggered_at: datetime resolved: bool resolved_at: datetime | None = None impact: str = Field(description="≤80 字摘要，從 details 抽取") details: dict remediation: str | None = None dispatch_ids: list[str] = Field(default_factory=list) class GovernanceEventsResponse(BaseModel): items: list[GovernanceEvent] total: int page: int size: int # ============================================================================= # Endpoint 2: queue # ============================================================================= class DispatchItem(BaseModel): id: str governance_event_id: str event_type: str dispatch_status: str executor_type: str | None = None proposed_action: str = Field(description="≤120 字動作摘要") playbook_id: str | None = None playbook_trust: float | None = Field(default=None, ge=0.0, le=1.0) created_at: datetime dispatched_at: datetime | None = None started_at: datetime | None = None completed_at: datetime | None = None operator_note: str | None = None decision_path: str | None = None workflow_stage: str | None = None workflow_steps: list[str] = Field(default_factory=list) next_action: str | None = None lead_agent: str | None = None support_agents: list[str] = Field(default_factory=list) human_owner: str | None = None kb_draft_entry_id: str | None = None worker_status: str | None = None dry_run_plan_fingerprint: str | None = None archived_count: int | None = None stale_ratio_snapshot: dict | None = None class GovernanceQueueResponse(BaseModel): items: list[DispatchItem] total: int page: int size: int table_pending: bool = Field( default=False, description="dispatch 表尚未建立時為 True", ) # ============================================================================= # Endpoint 2B: KM review draft dedupe # ============================================================================= class KnowledgeReviewDraftDedupeGroup(BaseModel): governance_event_id: str canonical_entry_id: str canonical_title: str canonical_updated_at: datetime | None = None preferred_source: Literal["dispatch_context", "latest_review_draft"] duplicate_entry_ids: list[str] = Field(default_factory=list) duplicate_count: int total_entries: int suggested_action: str owner_action: str writes_on_read: bool = False can_archive_without_owner_approval: bool = False archive_history: list[DispatchItem] = Field(default_factory=list) class KnowledgeReviewDraftDedupeResponse(BaseModel): schema_version: str = "km_review_draft_dedupe_v1" total_review_drafts: int event_group_total: int duplicate_draft_total: int groups: list[KnowledgeReviewDraftDedupeGroup] generated_at: datetime class KnowledgeReviewDraftArchiveRequest(BaseModel): canonical_entry_id: str = Field(min_length=1, max_length=120) duplicate_entry_ids: list[str] = Field(min_length=1, max_length=100) owner: str = Field(default="operator_console", min_length=1, max_length=100) owner_approved: bool = False dry_run: bool = False dry_run_plan_fingerprint: str | None = Field( default=None, max_length=80, description="Dry-run response fingerprint that must be echoed before a write.", ) class KnowledgeReviewDraftStaleRatioSnapshot(BaseModel): stale_count: int total_count: int stale_ratio: float threshold: float stale_days: int class KnowledgeReviewDraftArchiveResponse(BaseModel): schema_version: str = "km_review_draft_archive_v1" governance_event_id: str canonical_entry_id: str requested_duplicate_entry_ids: list[str] archived_entry_ids: list[str] = Field(default_factory=list) skipped_entry_ids: list[str] = Field(default_factory=list) would_archive_entry_ids: list[str] = Field(default_factory=list) status: Literal["dry_run", "archived", "noop_already_archived"] owner: str owner_approved: bool dry_run: bool writes_km: bool writes_governance_audit: bool audit_dispatch_id: str | None = None stale_ratio_snapshot: KnowledgeReviewDraftStaleRatioSnapshot | None = None stale_ratio_recheck_status: Literal[ "dry_run", "completed", "already_active", "not_requested", ] = "not_requested" stale_ratio_recheck_dispatch_id: str | None = None dry_run_plan_fingerprint: str | None = None next_action: str = "stale_ratio_recheck" generated_at: datetime # ============================================================================= # Endpoint 2C: KM stale candidates # ============================================================================= class KnowledgeStaleCandidate(BaseModel): entry_id: str project_id: str title: str entry_type: str category: str | None = None status: str source: str | None = None updated_at: datetime | None = None stale_days: int view_count: int priority_score: int priority_tier: Literal["P0", "P1", "P2"] recommended_action: Literal[ "refresh_with_evidence", "owner_review", "archive_or_supersede", ] reasons: list[str] = Field(default_factory=list) correlation_sources: list[str] = Field(default_factory=list) related_incident_id: str | None = None related_playbook_id: str | None = None related_approval_id: str | None = None tags: list[str] = Field(default_factory=list) class KnowledgeStaleCandidatesResponse(BaseModel): schema_version: str = "km_stale_candidates_v1" project_id: str total_stale: int returned: int threshold_days: int writes_on_read: bool = False manual_review_required: bool = True items: list[KnowledgeStaleCandidate] generated_at: datetime # ============================================================================= # Endpoint 3: summary # ============================================================================= class DailyCount(BaseModel): date: str = Field(description="YYYY-MM-DD") total: int by_type: dict[str, int] = Field(description="{event_type: count}") class GovernanceSummaryResponse(BaseModel): compliance_rate: float = Field(description="0.0-1.0，1 - unresolved/total") total_events: int unresolved_count: int daily_counts: list[DailyCount]