fix(execution-feedback): 修復系統自動化反饋完全斷鏈的三層 P0 故障
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 8m57s

**背景**
用戶報告執行狀態卡在「 執行中...」永不回報,導致自動修復機制完全癱瘓
(信心度修復後,執行失敗但無法推送 Telegram 卡片通知)

**L1 — Post-verify AttributeError(2 處)**
- approval_execution.py:757, 1010 調用不存在方法 IncidentService.get_incident()
- 正確方法:get_from_working_memory() fallback get_from_episodic_memory()
- 影響:post-verify 邏輯被 exception 無聲吞掉,下游 Telegram 推送完全卡住

**L2 — Notification Provider 未配置**
- 新增 notifications/telegram.py:複用既有 TelegramGateway.send_notification()
- 修改 manager.py:初始化時註冊 TelegramWebhookProvider
- 影響:執行完成後無任何 provider 發送推送,導致 Telegram 看不到結果

**L3 — Solver Agent 語意合成生成殘缺指令**
- 舊邏輯:action_title="重啟服務" → 合成 "kubectl rollout restart deployment -n awoooi-prod"(缺名)
- 下游 operation_parser 無法解析(regex 要求 deployment/<name>)
- 修法:優先從 parsed 提取 target 欄位;無名則 return [],降級到唯讀調查指令
- 測試全部通過:35/35,含 11 個新安全測試

**驗證**
- 被阻擋的惡意 kubectl_command 現在正確 fall-through 到語意合成路徑
- 無 target 名稱時返回空列表,不再生成殘缺指令
- Telegram 執行結果推送鏈路已完整

**預期效果**
- 執行失敗 → 立即收到「 執行失敗」Telegram 卡片(L1 + L2 修復)
- 自動化決策遵循白名單,避免生成無法執行的指令(L3 修復)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-04-25 03:29:38 +08:00
parent 7b6df17dee
commit f9f2263c00
6 changed files with 179 additions and 20 deletions

View File

@@ -754,7 +754,11 @@ class ApprovalExecutionService:
from src.services.evidence_snapshot import EvidenceSnapshot
incident_svc = get_incident_service()
incident = await incident_svc.get_incident(approval.incident_id)
# 2026-04-25 修復 L1IncidentService 沒有 get_incident() 方法
# 應用正確方法 get_from_working_memory() 或 get_from_episodic_memory()
incident = await incident_svc.get_from_working_memory(approval.incident_id)
if incident is None:
incident = await incident_svc.get_from_episodic_memory(approval.incident_id)
if incident is None:
logger.warning(
"post_verify_incident_not_found",
@@ -1007,7 +1011,10 @@ class ApprovalExecutionService:
from src.services.incident_service import get_incident_service
incident_service = get_incident_service()
incident = await incident_service.get_incident(incident_id)
# 2026-04-25 修復 L1IncidentService 沒有 get_incident() 方法
incident = await incident_service.get_from_working_memory(incident_id)
if incident is None:
incident = await incident_service.get_from_episodic_memory(incident_id)
if not incident:
logger.info(

View File

@@ -16,6 +16,7 @@ from .base import (
NotificationResult,
)
from .discord import DiscordWebhookProvider
from .telegram import TelegramWebhookProvider # 2026-04-25 修復 L2
from .manager import NotificationManager, get_notification_manager
__all__ = [
@@ -24,6 +25,7 @@ __all__ = [
"NotificationResult",
"ExecutionStatus",
"DiscordWebhookProvider",
"TelegramWebhookProvider", # 2026-04-25 修復 L2
"NotificationManager",
"get_notification_manager",
]

View File

@@ -57,9 +57,11 @@ class NotificationManager:
discord = DiscordWebhookProvider()
self.register(discord)
# TODO: 註冊其他 Provider
# slack = SlackWebhookProvider()
# self.register(slack)
# 2026-04-25 修復 L2註冊 Telegram provider
# 根本原因:執行完成後無 provider 發送 Telegram 通知
from .telegram import TelegramWebhookProvider
telegram = TelegramWebhookProvider()
self.register(telegram)
self._initialized = True
logger.info(

View File

@@ -0,0 +1,96 @@
"""Telegram Notification Provider — 接線 TelegramGateway 到 NotificationManager
2026-04-25 修復 L2系統執行反饋完全丟失
根本原因:執行完成後的推送通知未配置任何 providerTelegram 曾有 Gateway 實作
但從未註冊為 NotificationProvider導致執行失敗的卡片無法推送回 Telegram
本模組:直接複用既有 TelegramGateway.send_notification(),將執行結果格式化並推送
"""
from src.core.config import settings
from src.core.logging import get_logger
from .base import (
ExecutionStatus,
NotificationMessage,
NotificationProvider,
NotificationResult,
NotificationStatus,
)
logger = get_logger("awoooi.notifications.telegram")
class TelegramWebhookProvider(NotificationProvider):
"""透過既有 TelegramGateway 發送執行結果卡片"""
@property
def name(self) -> str:
return "telegram"
@property
def enabled(self) -> bool:
"""檢查 Telegram bot token 與 chat ID 是否配置"""
return bool(settings.OPENCLAW_TG_BOT_TOKEN) and bool(settings.OPENCLAW_TG_CHAT_ID)
def _format(self, msg: NotificationMessage) -> str:
"""格式化執行結果為 Telegram 訊息"""
title = f"{msg.status_emoji} <b>{msg.status_text}</b>"
lines = [
title,
"━━━━━━━━━━━━━━━━━━━",
f"🎯 <code>{msg.action_title[:120]}</code>",
f"🧭 Namespace: <code>{msg.namespace}</code> | Op: <code>{msg.operation_type}</code>",
f"{msg.risk_emoji} 風險: {msg.risk_level.upper()} | Pods: {msg.affected_pods}",
f"📝 Approval: <code>{msg.approval_id[:12]}</code>",
]
if msg.duration_ms is not None:
lines.append(f"⏱️ 耗時: {msg.duration_ms}ms")
if msg.error_message:
lines.append(f"❗ 錯誤: <code>{msg.error_message[:200]}</code>")
if msg.signers:
lines.append(f"👥 簽核: {msg.signers_display}")
return "\n".join(lines)
async def send(self, message: NotificationMessage) -> NotificationResult:
"""推送執行結果到 Telegram"""
if not self.enabled:
return NotificationResult(
status=NotificationStatus.SKIPPED,
provider=self.name,
message="Telegram bot token or chat_id not configured",
)
try:
from src.services.telegram_gateway import get_telegram_gateway
gateway = get_telegram_gateway()
text = self._format(message)
# send_notification 已處理 chat_id default + HTML parse_mode
resp = await gateway.send_notification(text=text, parse_mode="HTML")
return NotificationResult(
status=NotificationStatus.SUCCESS,
provider=self.name,
message="Telegram notification sent",
response_data=resp if isinstance(resp, dict) else None,
)
except Exception as e:
logger.exception("telegram_notification_exception", error=str(e))
return NotificationResult(
status=NotificationStatus.FAILED,
provider=self.name,
message="Exception during send",
error=str(e)[:300],
)
async def test_connection(self) -> bool:
"""測試 Telegram 連接"""
if not self.enabled:
return False
try:
from src.services.telegram_gateway import get_telegram_gateway
gw = get_telegram_gateway()
await gw.send_notification(text="🔔 AWOOOI Telegram provider 連線測試")
return True
except Exception as e:
logger.error("telegram_connection_test_failed", error=str(e))
return False