feat(km): P1-1 KMWriter 統一契約 + 5 caller 切換 + M4 反查鏈補齊
12-Agent 全景診斷揪出 KM 寫入鏈路 5 條入口無統一契約，fire-and-forget 在 Pod recycle 時會丟失條目。本次抽 KMWriter 強制 7 條契約。 ## 7 條契約強制 1. 同步底線：強制 await asyncio.wait_for(timeout) 2. 重試：3 次指數退避 1s/2s/4s（OperationalError / 網路類例外） 3. 失敗回收：3 次後寫 Redis DLQ km:dlq + log 4. 觀測：structlog event + 預留 metric hook（P1-3 補 emitter） 5. 冪等：incident_id + path_type 為 unique key 6. 禁止吞例外：except 必須 log + raise/DLQ 7. M4 反查鏈：payload 含 approval_id 時自動填 related_approval_id 並回填 Path A ## Caller 切換（5 條入口統一介面） - incident_service.py:1086 Path A（KB extractor + km_conversion） - approval_execution.py:771 Path B-人工 - decision_manager.py:2178 Path B-自動成功（消除跨類私有方法調用 M1） - decision_manager.py:2200 Path B-自動失敗（修 B2 早期吞例外） - playbook_service.py:210 PlaybookKM（兩份 T0 報告都漏的第三條） ## M4 反查鏈補齊 - knowledge.py + models.py: 補 related_approval_id ORM 欄位 - 對齊 phase26_incident_km_integration.sql:20 schema（partial index 已存在） - approval↔KM 雙向反查鏈完整（dual-path 縫合線） ## Feature Flag (rollback 保險) - KM_WRITE_AWAIT=true (default): await + timeout + DLQ 強制 - KM_WRITE_AWAIT=false: fire-and-forget（舊行為） ## 測試 - apps/api/tests/test_km_writer.py: 18 測試全綠 覆蓋 success / timeout / retry / DLQ / 冪等 / KMWriteError / on_failure=raise / 反查鏈回填 - 1552 unit tests 全綠（無回歸） ## 驗收 飛輪閉環核心 — KM 寫入不再靜默丟失，AI 學習鏈不斷裂。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Your Name
@@ -767,18 +767,9 @@ class ApprovalExecutionService:
|
||||
# 2026-04-04 ogt: 執行結果沉澱到 KM — 移出 try/except 確保 learning 失敗也寫入
|
||||
# 統帥鐵律: 所有異常與自動修復紀錄必須回寫 KM
|
||||
# P1.5 fix 2026-04-24 ogt + Claude Sonnet 4.6: fire-and-forget → await（30s 熔斷）
|
||||
# 根因：asyncio.create_task 在 Pod recycle 時被殺，KM 寫入遺失（audit D 每天+5）
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
self._write_execution_result_to_km(approval, success, error_message),
|
||||
timeout=30.0,
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
logger.warning(
|
||||
"km_write_timeout",
|
||||
approval_id=str(approval.id),
|
||||
timeout_sec=30.0,
|
||||
)
|
||||
# P1-1 2026-04-28 ogt + Claude Sonnet 4.6: 改用 write_execution_result_to_km（公開）
|
||||
# KMWriter 統一契約：timeout / retry / DLQ 由 km_writer.py 統一管理
|
||||
await self.write_execution_result_to_km(approval, success, error_message)
|
||||
|
||||
async def _run_post_execution_verify(
|
||||
self,
|
||||
@@ -862,7 +853,7 @@ class ApprovalExecutionService:
|
||||
error=str(_e),
|
||||
)
|
||||
|
||||
async def _write_execution_result_to_km(
|
||||
async def write_execution_result_to_km(
|
||||
self,
|
||||
approval: "ApprovalRequest",
|
||||
success: bool,
|
||||
@@ -874,89 +865,80 @@ class ApprovalExecutionService:
|
||||
2026-04-04 ogt: 統帥鐵律 — 成功/失敗執行記錄都必須回寫 KM
|
||||
2026-04-14 Claude Sonnet 4.6 (BP-1 B.1 精修): 區分 auto_approve vs 人工路徑，
|
||||
補齊 alert_category / alertname / affected_services 供 RAG 檢索。
|
||||
P1-1 2026-04-28 ogt + Claude Sonnet 4.6: 改名公開（去底線），委派 KMWriter 統一契約。
|
||||
"""
|
||||
try:
|
||||
from src.models.knowledge import EntrySource, EntryType, KnowledgeEntryCreate
|
||||
from src.services.knowledge_service import get_knowledge_service
|
||||
from src.models.knowledge import EntrySource, EntryType
|
||||
from src.services.km_writer import KMWritePayload, km_write_with_flag
|
||||
|
||||
# 來源辨識（B.1 精修）
|
||||
_is_auto = (approval.requested_by or "").lower() == "auto_approve"
|
||||
_mode_prefix = "[自動修復]" if _is_auto else "[人工修復]"
|
||||
_mode_tag = "auto_executed" if _is_auto else "human_approved"
|
||||
# 來源辨識（B.1 精修）
|
||||
_is_auto = (approval.requested_by or "").lower() == "auto_approve"
|
||||
_mode_prefix = "[自動修復]" if _is_auto else "[人工修復]"
|
||||
_mode_tag = "auto_executed" if _is_auto else "human_approved"
|
||||
|
||||
status_icon = "✅" if success else "❌"
|
||||
status_text = "成功" if success else f"失敗: {error_message or '未知原因'}"
|
||||
_status_tag = "success" if success else "failure"
|
||||
status_icon = "✅" if success else "❌"
|
||||
status_text = "成功" if success else f"失敗: {error_message or '未知原因'}"
|
||||
_status_tag = "success" if success else "failure"
|
||||
|
||||
# 從關聯 Incident 提取豐富元資料
|
||||
alertname = "unknown"
|
||||
alert_category = "general"
|
||||
affected_services: list[str] = []
|
||||
if approval.incident_id:
|
||||
try:
|
||||
from src.services.incident_service import get_incident_service
|
||||
_svc = get_incident_service()
|
||||
# get_from_working_memory (Redis) → fallback get_from_episodic_memory (PG)
|
||||
_inc = await _svc.get_from_working_memory(approval.incident_id)
|
||||
if _inc is None:
|
||||
_inc = await _svc.get_from_episodic_memory(approval.incident_id)
|
||||
if _inc:
|
||||
if _inc.signals:
|
||||
alertname = _inc.signals[0].labels.get("alertname", "unknown") or "unknown"
|
||||
alert_category = getattr(_inc, "alert_category", "") or "general"
|
||||
affected_services = list(_inc.affected_services or [])
|
||||
except Exception as _ie:
|
||||
logger.debug("km_incident_enrich_failed",
|
||||
incident_id=approval.incident_id, error=str(_ie))
|
||||
# 從關聯 Incident 提取豐富元資料
|
||||
alertname = "unknown"
|
||||
alert_category = "general"
|
||||
affected_services: list[str] = []
|
||||
if approval.incident_id:
|
||||
try:
|
||||
from src.services.incident_service import get_incident_service
|
||||
_svc = get_incident_service()
|
||||
# get_from_working_memory (Redis) → fallback get_from_episodic_memory (PG)
|
||||
_inc = await _svc.get_from_working_memory(approval.incident_id)
|
||||
if _inc is None:
|
||||
_inc = await _svc.get_from_episodic_memory(approval.incident_id)
|
||||
if _inc:
|
||||
if _inc.signals:
|
||||
alertname = _inc.signals[0].labels.get("alertname", "unknown") or "unknown"
|
||||
alert_category = getattr(_inc, "alert_category", "") or "general"
|
||||
affected_services = list(_inc.affected_services or [])
|
||||
except Exception as _ie:
|
||||
logger.debug("km_incident_enrich_failed",
|
||||
incident_id=approval.incident_id, error=str(_ie))
|
||||
|
||||
_services_str = ", ".join(affected_services) if affected_services else "未關聯"
|
||||
_services_str = ", ".join(affected_services) if affected_services else "未關聯"
|
||||
|
||||
content = (
|
||||
f"# {status_icon} {_mode_prefix} {alertname}\n\n"
|
||||
f"**告警名稱**: {alertname}\n"
|
||||
f"**告警類別**: {alert_category}\n"
|
||||
f"**受影響服務**: {_services_str}\n"
|
||||
f"**執行命令**: `{approval.action[:200]}`\n"
|
||||
f"**執行結果**: {status_text}\n"
|
||||
f"**風險等級**: {approval.risk_level.value if approval.risk_level else '未知'}\n"
|
||||
f"**執行路徑**: {'自動執行 (confidence >= 0.65)' if _is_auto else '人工審核批准'}\n"
|
||||
f"**Incident ID**: {approval.incident_id or '未關聯'}\n"
|
||||
f"**Approval ID**: {approval.id}\n\n"
|
||||
f"## 操作描述\n{approval.description or '無描述'}\n"
|
||||
)
|
||||
content = (
|
||||
f"# {status_icon} {_mode_prefix} {alertname}\n\n"
|
||||
f"**告警名稱**: {alertname}\n"
|
||||
f"**告警類別**: {alert_category}\n"
|
||||
f"**受影響服務**: {_services_str}\n"
|
||||
f"**執行命令**: `{approval.action[:200]}`\n"
|
||||
f"**執行結果**: {status_text}\n"
|
||||
f"**風險等級**: {approval.risk_level.value if approval.risk_level else '未知'}\n"
|
||||
f"**執行路徑**: {'自動執行 (confidence >= 0.65)' if _is_auto else '人工審核批准'}\n"
|
||||
f"**Incident ID**: {approval.incident_id or '未關聯'}\n"
|
||||
f"**Approval ID**: {approval.id}\n\n"
|
||||
f"## 操作描述\n{approval.description or '無描述'}\n"
|
||||
)
|
||||
|
||||
# Tags: 模式 + 狀態 + 類別（供 RAG 多維度檢索）
|
||||
tags = [_mode_tag, _status_tag, alert_category, "execution"]
|
||||
if not success:
|
||||
tags.append("execution_failed")
|
||||
# Tags: 模式 + 狀態 + 類別（供 RAG 多維度檢索）
|
||||
tags = [_mode_tag, _status_tag, alert_category, "execution"]
|
||||
if not success:
|
||||
tags.append("execution_failed")
|
||||
|
||||
entry_data = KnowledgeEntryCreate(
|
||||
payload = KMWritePayload(
|
||||
path_type="approval_auto_ok" if (_is_auto and success) else
|
||||
"approval_auto_fail" if (_is_auto and not success) else
|
||||
"approval_manual",
|
||||
entry_create_kwargs=dict(
|
||||
title=f"{_mode_prefix} {alertname}: {approval.action[:50]}",
|
||||
content=content,
|
||||
entry_type=EntryType.INCIDENT_CASE,
|
||||
category=alert_category, # 用真實類別取代硬編 "execution_result"
|
||||
category=alert_category,
|
||||
tags=tags,
|
||||
source=EntrySource.AI_EXTRACTED,
|
||||
related_incident_id=approval.incident_id or None,
|
||||
created_by="auto_execute" if _is_auto else "approval_execution",
|
||||
)
|
||||
await get_knowledge_service().create_entry(entry_data)
|
||||
|
||||
logger.info(
|
||||
"execution_result_written_to_km",
|
||||
approval_id=str(approval.id),
|
||||
incident_id=approval.incident_id,
|
||||
alertname=alertname,
|
||||
alert_category=alert_category,
|
||||
mode=_mode_tag,
|
||||
success=success,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
"execution_result_km_write_failed",
|
||||
approval_id=str(approval.id),
|
||||
error=str(e),
|
||||
)
|
||||
),
|
||||
incident_id=approval.incident_id or None,
|
||||
approval_id=str(approval.id),
|
||||
)
|
||||
await km_write_with_flag(payload)
|
||||
|
||||
async def _send_execution_notification(
|
||||
self,
|
||||
|
||||
Reference in New Issue
Block a user