fix(auto-repair): prefer exact playbooks and fail failed steps
All checks were successful
Code Review / ai-code-review (push) Successful in 11s
CD Pipeline / tests (push) Successful in 1m3s
CD Pipeline / build-and-deploy (push) Successful in 3m31s
CD Pipeline / post-deploy-checks (push) Successful in 1m32s

This commit is contained in:
Your Name
2026-05-13 23:21:16 +08:00
parent ae643552e9
commit 7a8cbb3241
4 changed files with 155 additions and 11 deletions

View File

@@ -321,7 +321,16 @@ class AutoRepairService:
)
# 4. 檢查最佳匹配
best_match = recommendations[0]
best_match = self._select_best_recommendation(recommendations, symptoms)
if best_match is not recommendations[0]:
logger.warning(
"auto_repair_exact_match_prioritized",
incident_id=incident.incident_id,
selected_playbook_id=best_match.playbook.playbook_id,
original_playbook_id=recommendations[0].playbook.playbook_id,
selected_similarity=best_match.similarity_score,
original_similarity=recommendations[0].similarity_score,
)
# 2026-04-07 Claude Code: 統帥指令「直接全部跳成自動修復」
# 移除: 相似度門檻、is_high_quality 門檻、冷啟動機制、風險等級門檻
@@ -416,6 +425,8 @@ class AutoRepairService:
executed_steps.append(
f"Step {step.step_number}: {step.command[:50]}... -> {step_result}"
)
if self._is_step_failure_result(step_result):
raise RuntimeError(f"Step {step.step_number} failed: {step_result}")
# 更新 Playbook 統計
await self._playbook_service.record_execution(
@@ -697,6 +708,44 @@ class AutoRepairService:
keywords=keywords[:10],
)
def _select_best_recommendation(
self,
recommendations,
symptoms: SymptomPattern,
):
"""Prefer deterministic alert/service matches over fuzzy similarity only.
A higher fuzzy score must not outrank a playbook that explicitly names the
firing alert or affected service. Live-fire T16 proved that this can route
a safe K8s canary into an unrelated host diagnostic playbook.
"""
symptom_alerts = {str(name) for name in (symptoms.alert_names or []) if name}
symptom_services = {
str(service) for service in (symptoms.affected_services or []) if service
}
def _priority(recommendation) -> tuple[int, int, float]:
pattern = recommendation.playbook.symptom_pattern
playbook_alerts = {
str(name) for name in (pattern.alert_names or []) if name
}
playbook_services = {
str(service) for service in (pattern.affected_services or []) if service
}
alert_exact = int(bool(symptom_alerts & playbook_alerts))
service_exact = int(bool(symptom_services & playbook_services))
return (alert_exact, service_exact, float(recommendation.similarity_score or 0.0))
return max(recommendations, key=_priority)
@staticmethod
def _is_step_failure_result(step_result: str) -> bool:
"""Treat executor-declared failures as failed auto-repair executions."""
normalized = (step_result or "").strip().upper()
return normalized.startswith("FAILED:") or normalized == "UNKNOWN_ACTION_TYPE"
def _get_max_risk_level(self, playbook: Playbook) -> RiskLevel:
"""取得 Playbook 中最高的風險等級"""
risk_order = {