wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 4 Packages Projects Releases Wiki Activity

feat(flywheel): W2 三件 + KMWriter critic 修法（1635 tests 全綠）
Some checks failed
CD Pipeline / build-and-deploy (push) Failing after 1m38s
Details

Browse Source 
W2 (onboarder 4 週飛輪 80→90 路徑第二週) + critic PR review 5 個 critical/major
全部修完，default flag=false 安全無爆炸風險。

## W2 三件 PR

### PR-R2 — AOL → catalog confidence EWMA 回灌（修飛輪斷鏈 C2）
- 新檔 `apps/api/src/jobs/aol_to_catalog_writeback_job.py`
- 邏輯：每小時掃 AOL 計算 EWMA confidence (alpha=0.3) 回灌 alert_rule_catalog
- 失敗閾值 N=5 連續低成功率 → review_status='draft'
- Hermes _fetch_noisy_rules SQL 加 OR review_status='draft'
- ENABLE_AOL_WRITEBACK_JOB=false (default)
- 8 個測試（mock path 修正：lazy import → patch src.db.base.get_db_context）

### PR-V1 — self_healing_validator 串接 (修飛輪斷鏈 C6)
- 新檔 `apps/api/src/services/self_healing_validator.py`（純函數 assess_self_healing）
- post_execution_verifier.py step 5 串接（feature flag gate）
- evidence_snapshot.py 加 self_healing_score / self_healing_detail 欄位
- db/models.py + base.py ALTER IF NOT EXISTS
- score < 0.5 → 觸發 rollback 提案 Telegram alert（不自動執行）
- ENABLE_SELF_HEALING_VALIDATOR=false (default)
- 7 個測試

### PR-L1 — KM ↔ Playbook 雙向回路 (修飛輪斷鏈 C3+C4)
- learning_service.py 三條新邏輯：
  1. _write_playbook_evolution_km：promote/demote 寫 KM 演化條目
  2. _check_and_mark_playbook_review：N=5 累積觸發 review_required
  3. _demote_alert_rule_catalog_confidence：DEPRECATED → confidence×=0.5
- PlaybookRecord 加 review_required 欄位（schema migration via base.py）
- ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP=false (default)
- KM_PLAYBOOK_REVIEW_THRESHOLD=5 可調
- 6 個測試

## KMWriter Critic 5 個 Critical/Major 修復（之前 critic PR review 發現）
之前 push commit c5753e1c 已修，本 commit 補回 stash 中的對應檔案：
- C1 km_writer.py:194 backfill 自打臉（已修：同步 await + DLQ）
- C2 km_writer.py:391 KM_WRITE_AWAIT=false 路徑收緊
- M1 decision_manager.py:2178/2203 移除 _fire_and_forget
- M2 incident_service.py:1099 自製 path 加 retry+DLQ
- M3 km_writer.py:166 冪等聲明對齊（UPSERT + partial unique index）

## 驗證
- 1635 unit tests 全綠（+27 from 1608）
- 與 fb0c72db (推翻 A2 Ollama primary) 共存無衝突
- 所有新 Job/Service default flag=false（不爆炸）

## 期望影響
飛輪斷鏈 C2 + C3 + C4 + C6 全修
飛輪自主化評分：65 → 85 預估（W2 完成後）

啟用順序（待 prod fb0c72db 驗證 OLLAMA primary 跑得起來後）：
1. ENABLE_AOL_WRITEBACK_JOB=true
2. ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP=true
3. ENABLE_SELF_HEALING_VALIDATOR=true

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-04-29 19:44:04 +08:00
parent fb0c72db42
commit 3668d49f2f
13 changed files with 2294 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
apps/api/src/services/evidence_snapshot.py
View File

@@ -110,6 +110,11 @@ class EvidenceSnapshot:
post_execution_state: dict[str, Any] | None = None
verification_result: str | None = None
# W2 PR-V1: SelfHealingValidator 自愈品質評估 (2026-04-28 ogt + Claude Sonnet 4.6)
# ENABLE_SELF_HEALING_VALIDATOR=false 時永 None
self_healing_score: float | None = None
self_healing_detail: dict[str, Any] | None = None
# Phase 3 填充（目前永 null）
matched_playbook_id: str | None = None
@@ -292,6 +297,55 @@ class EvidenceSnapshot:
)
raise
async def update_self_healing(
self,
score: float,
detail: dict[str, Any],
) -> None:
"""
W2 PR-V1: SelfHealingValidator 評估結果補填。
在 PostExecutionVerifier.verify() 完成 update_post_execution() 之後呼叫。
僅在 ENABLE_SELF_HEALING_VALIDATOR=True 且 snapshot 已持久化時有效。
Args:
score: 自愈品質分數（0.0-1.0）
detail: SelfHealingValidator.assess_self_healing() 返回的明細 dict
2026-04-28 ogt + Claude Sonnet 4.6: W2 PR-V1 初始建立
"""
self.self_healing_score = score
self.self_healing_detail = detail
try:
async with get_db_context() as db:
stmt_result = await db.execute(
update(IncidentEvidence)
.where(IncidentEvidence.id == self.snapshot_id)
.values(
self_healing_score=score,
self_healing_detail=detail,
)
)
if stmt_result.rowcount < 1:
logger.warning(
"evidence_snapshot_self_healing_update_no_rows",
snapshot_id=self.snapshot_id,
score=score,
)
else:
logger.info(
"evidence_snapshot_self_healing_updated",
snapshot_id=self.snapshot_id,
score=score,
)
except Exception:
logger.exception(
"evidence_snapshot_self_healing_update_error",
snapshot_id=self.snapshot_id,
)
raise
async def get_latest_snapshot(incident_id: str) -> EvidenceSnapshot | None:
"""

286
apps/api/src/services/learning_service.py
View File

@@ -389,13 +389,40 @@ class LearningService:
playbook_id: str,
success: bool,
) -> None:
"""更新 Playbook 統計"""
"""
更新 Playbook 統計
W2 PR-L1: 統計更新後，取 Playbook 的 symptom_pattern hash 觸發邏輯 2
（KM 累積門檻檢查 → review_required 標記）。
"""
try:
from src.services.playbook_service import get_playbook_service
service = get_playbook_service()
await service.record_execution(playbook_id, success)
# W2 PR-L1 邏輯 2: 取得 Playbook symptom_pattern hash，觸發 KM 累積檢查
from src.core.config import settings
if settings.ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP:
try:
from src.repositories.playbook_repository import get_playbook_repository
from src.models.playbook import SymptomPattern
repo = get_playbook_repository()
playbook = await repo.get_by_id(playbook_id)
if playbook and playbook.symptom_pattern:
sp = playbook.symptom_pattern
# symptom_pattern 可能是 Pydantic model 或 dict（ORM 載入）
if isinstance(sp, dict):
sp = SymptomPattern.model_validate(sp)
symptoms_hash = sp.compute_hash()
await self._check_and_mark_playbook_review(symptoms_hash)
except Exception as inner_e:
logger.warning(
"playbook_review_check_failed",
playbook_id=playbook_id,
error=str(inner_e),
)
except Exception as e:
logger.warning(
"playbook_stats_update_error",
@@ -459,6 +486,7 @@ class LearningService:
- 尋找 source_incident_ids 包含此 incident_id 的 Playbooks
- 提升 ai_confidence +0.1 (上限 1.0)
- 若信心度 >= 0.9 且 status == DRAFT → 自動升級為 APPROVED
- W2 PR-L1: 寫 KM 演化條目（ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP 開啟時）
"""
try:
from src.repositories.playbook_repository import get_playbook_repository
@@ -478,6 +506,7 @@ class LearningService:
updated_count = 0
for playbook in playbooks:
previous_trust = playbook.trust_score
result = await repo.adjust_confidence(
playbook_id=playbook.playbook_id,
delta=CONFIDENCE_BOOST,
@@ -485,6 +514,13 @@ class LearningService:
)
if result:
updated_count += 1
# W2 PR-L1: promote 觸發 → 寫 KM 演化條目
await self._write_playbook_evolution_km(
playbook=playbook,
previous_trust=previous_trust,
evolution_type="promote",
incident_id=incident_id,
)
logger.info(
"playbook_promoted",
@@ -513,6 +549,7 @@ class LearningService:
- 尋找 source_incident_ids 包含此 incident_id 的 Playbooks
- 降低 ai_confidence -0.15 (下限 0.0)
- 若信心度 < 0.3 且 failure_rate > 50% → 自動降級為 DEPRECATED
- W2 PR-L1: 寫 KM 演化條目；DEPRECATED 時回灌 alert_rule_catalog（飛輪 C4 修復）
"""
try:
from src.repositories.playbook_repository import get_playbook_repository
@@ -532,6 +569,7 @@ class LearningService:
updated_count = 0
for playbook in playbooks:
previous_trust = playbook.trust_score
result = await repo.adjust_confidence(
playbook_id=playbook.playbook_id,
delta=CONFIDENCE_PENALTY,
@@ -539,6 +577,17 @@ class LearningService:
)
if result:
updated_count += 1
# W2 PR-L1: demote 觸發 → 寫 KM 演化條目
await self._write_playbook_evolution_km(
playbook=playbook,
previous_trust=previous_trust,
evolution_type="demote",
incident_id=incident_id,
)
# W2 PR-L1 邏輯 3: DEPRECATED 時回灌 alert_rule_catalog（飛輪 C4 修復）
from src.models.playbook import PlaybookStatus
if playbook.status == PlaybookStatus.DEPRECATED:
await self._demote_alert_rule_catalog_confidence(playbook)
logger.info(
"playbook_demoted",
@@ -557,6 +606,241 @@ class LearningService:
)
return False
# =========================================================================
# W2 PR-L1: KM → Playbook 互饋回路私有方法
# 飛輪斷鏈 C3 + C4 修復
# 2026-04-28 ogt + Claude Sonnet 4.6
# =========================================================================
async def _write_playbook_evolution_km(
self,
playbook: Any,
previous_trust: float,
evolution_type: str,
incident_id: str,
) -> None:
"""
邏輯 1: promote/demote 觸發 → 寫 KM 演化條目（飛輪 C3）
KM 條目 metadata 含：playbook_id, previous_trust, new_trust,
success_count, failure_count, decision_chain
path_type='playbook_evolution'，供冪等 key 使用
(incident_id, path_type) = (incident_id, 'playbook_evolution') 可能重複，
但 playbook_id 不同的演化各自獨立，所以 path_type 加 playbook_id 作為識別。
"""
from src.core.config import settings
if not settings.ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP:
return
try:
import json
from src.services.km_writer import KMWritePayload, km_write_with_flag
from src.utils.timezone import now_taipei
new_trust = getattr(playbook, "trust_score", previous_trust)
success_count = getattr(playbook, "success_count", 0)
failure_count = getattr(playbook, "failure_count", 0)
path_type = f"playbook_evolution:{playbook.playbook_id}"
payload = KMWritePayload(
path_type=path_type,
incident_id=incident_id,
entry_create_kwargs={
"title": f"Playbook {evolution_type}: {playbook.name} [{playbook.playbook_id}]",
"content": (
f"Playbook {evolution_type} 事件記錄\n"
f"Playbook ID: {playbook.playbook_id}\n"
f"名稱: {playbook.name}\n"
f"trust_score 變化: {previous_trust:.3f} → {new_trust:.3f}\n"
f"成功次數: {success_count} / 失敗次數: {failure_count}\n"
f"觸發來源: incident {incident_id}\n"
f"記錄時間: {now_taipei().isoformat()}"
),
"entry_type": "best_practice",
"category": "AI系統",
"tags": ["playbook_evolution", evolution_type, playbook.playbook_id],
"source": "ai_extracted",
"related_playbook_id": playbook.playbook_id,
"related_incident_id": incident_id,
"path_type": path_type,
},
metadata={
"playbook_id": playbook.playbook_id,
"previous_trust": previous_trust,
"new_trust": new_trust,
"success_count": success_count,
"failure_count": failure_count,
"evolution_type": evolution_type,
},
)
await km_write_with_flag(payload)
logger.info(
"playbook_evolution_km_written",
playbook_id=playbook.playbook_id,
evolution_type=evolution_type,
trust_change=f"{previous_trust:.3f} → {new_trust:.3f}",
)
except Exception as e:
logger.warning(
"playbook_evolution_km_write_failed",
playbook_id=getattr(playbook, "playbook_id", "unknown"),
evolution_type=evolution_type,
error=str(e),
)
async def _check_and_mark_playbook_review(self, symptoms_hash: str) -> None:
"""
邏輯 2: KM 累積 N=5 條同 symptom_pattern_hash → 觸發 Playbook review_required 標記（飛輪 C3）
每次 KM 寫入後由 _update_playbook_stats 呼叫端觸發此檢查。
若同 symptoms_hash 在 knowledge_entries 已有 >= threshold 條，
則 UPDATE playbooks SET review_required=true WHERE 症狀 hash 相符。
比對策略：從 KnowledgeEntry 讀 symptoms_hash 計數，
再透過 playbook.symptom_pattern 的 hash 比對 Playbook。
"""
from src.core.config import settings
if not settings.ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP:
return
if not symptoms_hash:
return
try:
from sqlalchemy import text as sa_text
from src.db.base import get_db_context
async with get_db_context() as db:
# 計算同 symptoms_hash 的 KM 條目數
count_result = await db.execute(
sa_text(
"SELECT COUNT(*) FROM knowledge_entries "
"WHERE symptoms_hash = :hash"
),
{"hash": symptoms_hash},
)
count = count_result.scalar() or 0
if count < settings.KM_PLAYBOOK_REVIEW_THRESHOLD:
return
# 累積達到門檻 → 標記相關 Playbook 需要 review
# Playbook 的 symptom_pattern 存為 JSONB，無直接 hash 欄位
# 透過 knowledge_entries.related_playbook_id 關聯找到要標記的 Playbook
updated = await db.execute(
sa_text(
"UPDATE playbooks pb "
"SET review_required = true, updated_at = NOW() "
"FROM knowledge_entries ke "
"WHERE ke.symptoms_hash = :hash "
" AND ke.related_playbook_id = pb.playbook_id "
" AND pb.review_required = false "
"RETURNING pb.playbook_id"
),
{"hash": symptoms_hash},
)
marked_ids = [row[0] for row in updated.fetchall()]
await db.commit()
if marked_ids:
logger.info(
"playbook_review_required_marked",
symptoms_hash=symptoms_hash,
km_count=count,
threshold=settings.KM_PLAYBOOK_REVIEW_THRESHOLD,
playbook_ids=marked_ids,
)
except Exception as e:
logger.warning(
"playbook_review_mark_failed",
symptoms_hash=symptoms_hash,
error=str(e),
)
async def _demote_alert_rule_catalog_confidence(self, playbook: Any) -> None:
"""
邏輯 3: Playbook DEPRECATED 時回灌 alert_rule_catalog（飛輪 C4 修復）
UPDATE alert_rule_catalog
SET confidence = confidence * 0.5,
review_status = 'draft' -- CHECK constraint 允許 draft/approved/deprecated/retired
WHERE rule_name LIKE pattern(symptom_pattern.alert_names)
注意：alert_rule_catalog.review_status CHECK 限制只允許：
draft | approved | deprecated | retired
任務描述的 'needs_review' 不合法，改用 'draft'（語意等效：需要人工審核）
失敗容忍：不影響 demote 主流程。
"""
from src.core.config import settings
if not settings.ENABLE_KM_PLAYBOOK_FEEDBACK_LOOP:
return
try:
import json
from sqlalchemy import text as sa_text
from src.db.base import get_db_context
# 從 playbook symptom_pattern 取出 alert_names 作為比對鍵
symptom = getattr(playbook, "symptom_pattern", None)
if symptom is None:
return
# symptom_pattern 可能是 Pydantic model 或 dict（從 ORM 載入為 dict）
if hasattr(symptom, "alert_names"):
alert_names: list[str] = symptom.alert_names or []
elif isinstance(symptom, dict):
alert_names = symptom.get("alert_names") or []
else:
return
if not alert_names:
logger.debug(
"playbook_demote_no_alert_names",
playbook_id=playbook.playbook_id,
)
return
async with get_db_context() as db:
updated_count = 0
for alert_name in alert_names:
# rule_name 完全匹配或前綴匹配（去掉 * suffix）
match_name = alert_name.rstrip("*")
result = await db.execute(
sa_text(
"UPDATE alert_rule_catalog "
"SET confidence = CASE "
" WHEN confidence IS NOT NULL "
" THEN GREATEST(0.01, confidence * 0.5) "
" ELSE 0.5 "
" END, "
" review_status = 'draft', "
" updated_at = NOW() "
"WHERE rule_name LIKE :pattern "
" AND (review_status IS NULL OR review_status NOT IN "
" ('deprecated', 'retired')) "
"RETURNING rule_id"
),
{"pattern": f"{match_name}%"},
)
affected = result.rowcount or 0
updated_count += affected
await db.commit()
if updated_count > 0:
logger.info(
"alert_rule_catalog_confidence_demoted",
playbook_id=playbook.playbook_id,
alert_names=alert_names,
rules_updated=updated_count,
)
except Exception as e:
logger.warning(
"alert_rule_catalog_demote_failed",
playbook_id=getattr(playbook, "playbook_id", "unknown"),
error=str(e),
)
# =========================================================================
# 🆕 Phase D-G P0 修正: 新增方法
# =========================================================================

154
apps/api/src/services/post_execution_verifier.py
View File

@@ -21,6 +21,11 @@ AWOOOI AIOps Phase 1 — 執行後驗證器
- 超時不 raise，標記 "timeout" 並繼續流程
- 不阻塞原始執行路徑（await，但結果不影響執行本身是否成功）
W2 PR-V1: SelfHealingValidator 串接 (2026-04-28 ogt + Claude Sonnet 4.6)
- ENABLE_SELF_HEALING_VALIDATOR=True 時，verify() 完成後呼叫 assess_self_healing()
- self_healing_score < 0.5 → Telegram 警示 rollback 提案（不自動執行）
- 驗證失敗不阻塞主流程（try/except 全包）
ADR-081: PreDecisionInvestigator + EvidenceSnapshot
MASTER §3.1 L6×D1
2026-04-15 ogt + Claude Sonnet 4.6 (亞太): Phase 1 初始建立
@@ -37,6 +42,9 @@ import structlog
from src.services.evidence_snapshot import EvidenceSnapshot
from src.services.mcp_tool_registry import SensorDimension, get_mcp_tool_registry
from src.services.sanitization_service import sanitize_dict_values
# W2 PR-V1: 頂層 import 讓測試 patch 路徑固定（延遲 import 無法被 patch）
# ENABLE_SELF_HEALING_VALIDATOR=False 時此 import 不影響效能（純 python 模組）
from src.services import self_healing_validator as _shv_module
if TYPE_CHECKING:
from src.models.incident import Incident
@@ -136,6 +144,26 @@ class PostExecutionVerifier:
result=result,
action=action_taken,
)
# 5. W2 PR-V1: SelfHealingValidator 串接（ENABLE_SELF_HEALING_VALIDATOR gate）
# 在 post_state 已補填後評估自愈品質，不阻塞主流程
# 外層 try/except 確保任何 validator 失敗不影響 verify() 返回值
try:
await _run_self_healing_validator(
incident_id=incident_id,
snapshot=snapshot,
pre_state=pre_state,
post_state=post_state,
verification_result=result,
action_taken=action_taken,
)
except Exception:
logger.warning(
"self_healing_validator_uncaught",
incident_id=incident_id,
exc_info=True,
)
return result
async def capture_pre_execution_state(
@@ -209,6 +237,132 @@ class PostExecutionVerifier:
return state
# ─────────────────────────────────────────────────────────────────────────────
# W2 PR-V1: SelfHealingValidator 串接
# 2026-04-28 ogt + Claude Sonnet 4.6: C6 飛輪斷鏈修復
# ─────────────────────────────────────────────────────────────────────────────
async def _run_self_healing_validator(
incident_id: str,
snapshot: EvidenceSnapshot | None,
pre_state: dict[str, Any] | None,
post_state: dict[str, Any],
verification_result: str,
action_taken: str,
) -> None:
"""
SelfHealingValidator 串接入口。
Feature gate: ENABLE_SELF_HEALING_VALIDATOR（預設 False）。
驗證失敗全程 try/except 保護，不影響主流程。
評估後：
- 補填 snapshot.self_healing_score + self_healing_detail
- score < 0.5 → 發送 Telegram rollback 提案警示
"""
try:
from src.core.config import get_settings
_settings = get_settings()
if not _settings.ENABLE_SELF_HEALING_VALIDATOR:
return
assessment = _shv_module.assess_self_healing(
pre_state=pre_state,
post_state=post_state,
verification_result=verification_result,
action_taken=action_taken,
)
score: float = assessment["score"]
logger.info(
"self_healing_assessed",
incident_id=incident_id,
score=score,
regressions=assessment.get("regressions", []),
root_cause_cleared=assessment.get("root_cause_cleared"),
detail=assessment.get("detail"),
)
# 補填 EvidenceSnapshot
if snapshot:
try:
await snapshot.update_self_healing(score=score, detail=assessment)
except Exception as _snap_err:
logger.warning(
"self_healing_snapshot_update_failed",
incident_id=incident_id,
error=str(_snap_err),
)
# score < 0.5 → Telegram rollback 提案警示
if score < 0.5:
await _send_rollback_proposal_alert(
incident_id=incident_id,
score=score,
assessment=assessment,
action_taken=action_taken,
)
except Exception:
logger.warning(
"self_healing_validator_error",
incident_id=incident_id,
exc_info=True,
)
async def _send_rollback_proposal_alert(
incident_id: str,
score: float,
assessment: dict[str, Any],
action_taken: str,
) -> None:
"""
自愈品質分數 < 0.5 時，發送 Telegram rollback 提案警示。
不自動執行 rollback，僅通知人工評估。
"""
try:
from src.core.config import get_settings
from src.services.telegram_gateway import get_telegram_gateway
_settings = get_settings()
gateway = get_telegram_gateway()
regressions = assessment.get("regressions", [])
reg_str = ", ".join(regressions[:5]) if regressions else "無"
root_cleared = "是" if assessment.get("root_cause_cleared") else "否"
text = (
f"⚠️ <b>自愈品質警示 — 建議人工評估 Rollback</b>\n"
f"Incident: <code>{incident_id}</code>\n"
f"動作: <code>{action_taken[:120]}</code>\n"
f"自愈分數: <b>{score:.2f}</b> (門檻 0.5)\n"
f"Root Cause 解除: {root_cleared}\n"
f"Regression 信號: {reg_str}\n"
f"<i>此為提案，不會自動執行 Rollback</i>"
)
await gateway._http_client.post(
f"https://api.telegram.org/bot{_settings.OPENCLAW_TG_BOT_TOKEN}/sendMessage",
json={
"chat_id": _settings.OPENCLAW_TG_CHAT_ID,
"text": text,
"parse_mode": "HTML",
},
)
logger.info(
"rollback_proposal_sent",
incident_id=incident_id,
score=score,
)
except Exception:
logger.warning(
"rollback_proposal_send_failed",
incident_id=incident_id,
exc_info=True,
)
# ─────────────────────────────────────────────────────────────────────────────
# Recovery Assessment
# ─────────────────────────────────────────────────────────────────────────────

163
apps/api/src/services/self_healing_validator.py Normal file
View File

@@ -0,0 +1,163 @@
"""
AWOOOI AIOps — 自愈品質驗證器
================================
W2 PR-V1: 飛輪斷鏈 C6 修復 — PostExecutionVerifier 串接自愈品質評估
職責：
1. 評估系統是否真的「自愈」（root cause 解除 vs 只是 metric 暫時恢復）
2. Regression Detection（修完一個指標但其他指標惡化）
3. 修復品質分數（0.0 ~ 1.0）
評分邏輯：
- base_score 由 verification_result 決定（success=1.0 / degraded=0.4 / failed=0.0 / timeout=0.2）
- regression_penalty 由 pre/post state diff 中惡化指標數量決定
- 最終 score = max(0.0, base_score - regression_penalty)
閾值：
- score < 0.5 → rollback 提案（Telegram 警示，不自動執行）
- score >= 0.5 → 認可自愈，無額外動作
設計原則：
- 不修改 self_healing_validator 內部邏輯（外部串接層）
- 驗證失敗不阻塞主流程（容錯 try/except 全包）
- Feature Flag: ENABLE_SELF_HEALING_VALIDATOR=false（預設關閉）
ADR-081 Phase 1 延伸
2026-04-28 ogt + Claude Sonnet 4.6: W2 PR-V1 初始建立（C6 修復）
"""
from __future__ import annotations
import re
from typing import TYPE_CHECKING, Any
import structlog
if TYPE_CHECKING:
pass
logger = structlog.get_logger(__name__)
# 修復品質分數基準（by verification_result）
_BASE_SCORES: dict[str, float] = {
"success": 1.0,
"degraded": 0.4,
"failed": 0.0,
"timeout": 0.2,
}
# 每個惡化指標的扣分
_REGRESSION_PENALTY_PER_METRIC = 0.15
# 扣分上限（避免 over-penalty）
_MAX_REGRESSION_PENALTY = 0.4
# root cause 解除信號（post_state 出現這些 → root cause 已清除）
_ROOT_CAUSE_CLEARED_SIGNALS = ["running", "ready", "1/1", "2/2", "3/3", "healthy"]
# regression 惡化信號（post_state 新出現但 pre_state 不存在 → regression）
_REGRESSION_SIGNALS = [
"crashloopbackoff",
"oomkilled",
"oomkill",
"pending",
"terminating",
"error",
"failed",
"timeout",
"evicted",
"imagepullbackoff",
"errimagepull",
]
# 數值指標惡化偵測（regex 找 %、數字，比較增幅）
_NUMERIC_THRESHOLD_RATIO = 0.2 # 超過 20% 增幅算惡化
def assess_self_healing(
pre_state: dict[str, Any] | None,
post_state: dict[str, Any] | None,
verification_result: str,
action_taken: str,
) -> dict[str, Any]:
"""
評估自愈品質，返回結構化評估結果。
Args:
pre_state: 執行前環境狀態（可為 None）
post_state: 執行後環境狀態（可為 None）
verification_result: PostExecutionVerifier 的判斷結果（success/degraded/failed/timeout）
action_taken: 執行的動作描述
Returns:
dict 包含：
score (float 0.0-1.0)
root_cause_cleared (bool)
regressions (list[str] — 惡化的指標名稱)
detail (str — 人類可讀說明)
"""
base_score = _BASE_SCORES.get(verification_result, 0.0)
pre_str = str(pre_state).lower() if pre_state else ""
post_str = str(post_state).lower() if post_state else ""
# 1. Root cause 是否真正解除
root_cause_cleared = any(sig in post_str for sig in _ROOT_CAUSE_CLEARED_SIGNALS)
if verification_result in ("failed", "timeout"):
root_cause_cleared = False
# 2. Regression detection — 新出現在 post 但 pre 沒有的惡化信號
regressions: list[str] = []
for sig in _REGRESSION_SIGNALS:
if sig in post_str and sig not in pre_str:
regressions.append(sig)
# 3. 數值指標惡化偵測（簡單版：找百分比值增幅）
pre_nums = _extract_percentages(pre_str)
post_nums = _extract_percentages(post_str)
for key, pre_val in pre_nums.items():
if key in post_nums:
post_val = post_nums[key]
if pre_val > 0 and (post_val - pre_val) / pre_val > _NUMERIC_THRESHOLD_RATIO:
regressions.append(f"metric_increase:{key}")
# 4. 計算最終分數
regression_penalty = min(
len(regressions) * _REGRESSION_PENALTY_PER_METRIC,
_MAX_REGRESSION_PENALTY,
)
score = max(0.0, base_score - regression_penalty)
# 5. 組裝說明
detail_parts = [f"base={base_score:.2f}"]
if regressions:
detail_parts.append(f"regression_penalty={regression_penalty:.2f} ({','.join(regressions[:5])})")
if not root_cause_cleared and verification_result == "success":
detail_parts.append("root_cause_unclear")
detail = "; ".join(detail_parts)
return {
"score": round(score, 4),
"root_cause_cleared": root_cause_cleared,
"regressions": regressions,
"detail": detail,
"verification_result": verification_result,
"action_taken": action_taken,
}
def _extract_percentages(text: str) -> dict[str, float]:
"""
從狀態字串中提取數值百分比。
例如 "cpu_usage: 85%" → {"cpu_usage": 85.0}
用於偵測指標惡化（簡單啟發式，Phase 1 版本）。
"""
result: dict[str, float] = {}
# 格式：word_key: N% 或 word_key=N%
pattern = re.compile(r"(\w+)[:\s=]+(\d+(?:\.\d+)?)\s*%")
for match in pattern.finditer(text):
key = match.group(1)
val = float(match.group(2))
result[key] = val
return result