feat(api): Sprint 4 Phase A+B — 告警處置統計資料層+寫入層

Phase A: 資料層
- A1: IncidentFrequencyStats 新增 4 欄位 (human_approved/manual_resolved/cold_start_trust/total_resolution)
- A2: AnomalyCounter.record_disposition() — Redis HINCRBY 原子遞增
- A3: get_disposition_stats() — HGETALL 回傳處置分佈
- AnomalyFrequency dataclass 擴充 + to_dict() 同步
- _record_anomaly_impl() 整合 disposition stats

Phase B: 寫入層觸發點接線
- B1: 自動修復成功 → record_disposition("auto_repair")
- B2: 冷啟動信任成功 → record_disposition("cold_start_trust")
  - AutoRepairDecision 新增 is_cold_start flag
  - execute_auto_repair() 接收並區分處置類型
- B3: 人工批准執行成功 → record_disposition("human_approved")
  - 新增 _get_anomaly_key_from_approval() helper
- B4: 手動處理推斷 → resolve_incident() 排除法判定
  - 若 resolved 且無 auto/human/cold_start 紀錄 → manual_resolved

安全設計: 所有 disposition 記錄走 try/except,失敗不阻塞主流程

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-07 11:54:46 +08:00
parent e82d3802c5
commit 9253281d46
6 changed files with 195 additions and 0 deletions

View File

@@ -160,6 +160,17 @@ class ApprovalExecutionService:
duration_seconds=result.duration_ms / 1000 if result.duration_ms else 0,
)
)
# 2026-04-07 Claude Code: Sprint 4 B3 — 記錄人工批准處置類型
try:
anomaly_key = await self._get_anomaly_key_from_approval(approval)
if anomaly_key:
from src.services.anomaly_counter import get_anomaly_counter
counter = get_anomaly_counter()
await counter.record_disposition(anomaly_key, "human_approved")
except Exception as _disp_e:
logger.warning("disposition_record_failed", error=str(_disp_e))
else:
logger.error(
"background_execution_failed",
@@ -206,6 +217,35 @@ class ApprovalExecutionService:
)
)
async def _get_anomaly_key_from_approval(self, approval: ApprovalRequest) -> str | None:
"""
從 approval → incident → anomaly_signature → hash。
2026-04-07 Claude Code: Sprint 4 B3
Returns:
anomaly_key or None if not derivable
"""
try:
if not approval.incident_id:
return None
from src.services.incident_service import get_incident_service
incident_service = get_incident_service()
incident = await incident_service.get_from_working_memory(approval.incident_id)
if not incident or not incident.signals:
return None
# 從第一個 signal 建立 anomaly signature
signal = incident.signals[0]
signature = {
"alert_name": signal.alert_name,
"service": incident.affected_services[0] if incident.affected_services else "",
"namespace": getattr(signal, "namespace", ""),
}
from src.services.anomaly_counter import AnomalyCounter
return AnomalyCounter.hash_signature(signature)
except Exception as e:
logger.warning("get_anomaly_key_from_approval_failed", error=str(e))
return None
async def _trigger_learning(
self,
approval: ApprovalRequest,