From c22e5f334eae1a727317de57a0ec0168340b36fc Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 29 Apr 2026 09:41:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(km):=20P1-1=20KMWriter=20=E7=B5=B1?= =?UTF-8?q?=E4=B8=80=E5=A5=91=E7=B4=84=20+=205=20caller=20=E5=88=87?= =?UTF-8?q?=E6=8F=9B=20+=20M4=20=E5=8F=8D=E6=9F=A5=E9=8F=88=E8=A3=9C?= =?UTF-8?q?=E9=BD=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- apps/api/src/db/models.py | 14 + apps/api/src/models/knowledge.py | 5 + apps/api/src/services/approval_execution.py | 144 ++++---- apps/api/src/services/decision_manager.py | 12 +- apps/api/src/services/incident_service.py | 18 +- apps/api/src/services/km_writer.py | 385 +++++++++++++++++++ apps/api/src/services/playbook_service.py | 73 ++-- apps/api/tests/test_km_writer.py | 387 ++++++++++++++++++++ 8 files changed, 909 insertions(+), 129 deletions(-) create mode 100644 apps/api/src/services/km_writer.py create mode 100644 apps/api/tests/test_km_writer.py diff --git a/apps/api/src/db/models.py b/apps/api/src/db/models.py index e8d8561b..35bac625 100644 --- a/apps/api/src/db/models.py +++ b/apps/api/src/db/models.py @@ -778,6 +778,14 @@ class KnowledgeEntryRecord(Base): nullable=True, comment="症狀模式 hash (16字元 SHA256 前綴)，Anti-Pattern 閉環攔截使用", ) + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: M4 補反查鏈 + # phase26_incident_km_integration.sql 已建立欄位與 partial index + # KMWriter.write() 會自動填入並回填 Path A 條目（approval → KM 雙向追蹤） + related_approval_id: Mapped[str | None] = mapped_column( + String(36), + nullable=True, + comment="關聯 ApprovalRequest ID，P1-1 反查鏈修復（approval → KM 追蹤）", + ) # Metrics view_count: Mapped[int] = mapped_column( @@ -806,6 +814,12 @@ class KnowledgeEntryRecord(Base): Index("ix_knowledge_created_at", "created_at"), # 2026-04-04 ogt: Phase 25 P1 — Anti-Pattern 快速查詢 Index("ix_knowledge_symptoms_hash", "symptoms_hash"), + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: M4 反查鏈 partial index（配合 phase26 migration） + Index( + "ix_knowledge_related_approval", + "related_approval_id", + postgresql_where=text("related_approval_id IS NOT NULL"), + ), ) diff --git a/apps/api/src/models/knowledge.py b/apps/api/src/models/knowledge.py index 6d1136d4..21e35afb 100644 --- a/apps/api/src/models/knowledge.py +++ b/apps/api/src/models/knowledge.py @@ -69,6 +69,9 @@ class KnowledgeEntryCreate(BaseModel): status: EntryStatus = EntryStatus.DRAFT related_incident_id: str | None = None related_playbook_id: str | None = None + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: M4 補反查鏈 — approval → KM 關聯 + # phase26_incident_km_integration.sql 已建立 related_approval_id 欄位與 partial index + related_approval_id: str | None = None # 2026-04-04 ogt: Phase 25 P1 — Anti-Pattern 閉環用症狀 hash symptoms_hash: str | None = None created_by: str | None = None @@ -96,6 +99,8 @@ class KnowledgeEntry(BaseModel): status: EntryStatus = EntryStatus.DRAFT related_incident_id: str | None = None related_playbook_id: str | None = None + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: M4 補反查鏈 + related_approval_id: str | None = None # 2026-04-04 ogt: Phase 25 P1 — Anti-Pattern 閉環攔截用的症狀 hash（SymptomPattern.compute_hash()） symptoms_hash: str | None = None view_count: int = 0 diff --git a/apps/api/src/services/approval_execution.py b/apps/api/src/services/approval_execution.py index e7ff93b9..9f7b4801 100644 --- a/apps/api/src/services/approval_execution.py +++ b/apps/api/src/services/approval_execution.py @@ -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, diff --git a/apps/api/src/services/decision_manager.py b/apps/api/src/services/decision_manager.py index 03fce7ea..3ec0ea08 100644 --- a/apps/api/src/services/decision_manager.py +++ b/apps/api/src/services/decision_manager.py @@ -2172,11 +2172,10 @@ class DecisionManager: ) # 2026-04-25 ogt + Claude Sonnet 4.6: 飛輪閉環 — auto_execute 路徑補 KM 寫入 - # 根因：Explore agent 確認 _write_execution_result_to_km 只在人工審核路徑呼叫 - # auto_execute 路徑執行完成後沒有 KM 回寫 → 學習飛輪斷鏈 - # 修法：複用 executor._write_execution_result_to_km，與人工路徑共享同一 KM schema + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: 改用 write_execution_result_to_km（公開） + # KMWriter 統一契約，feature flag KM_WRITE_AWAIT 控制 await/fire-and-forget _fire_and_forget( - executor._write_execution_result_to_km(approval, _exec_success, None) + executor.write_execution_result_to_km(approval, _exec_success, None) ) except Exception as e: @@ -2198,11 +2197,10 @@ class DecisionManager: ) # P1.1 fix 2026-04-27 ogt + Claude Sonnet 4.6: 失敗路徑補 KM 寫入 - # 根因：auto_execute 拋出例外時，學習飛輪完全拿不到失敗記錄 - # 修法：若 executor/approval 已建立，fire-and-forget 寫入失敗 KM；未建立則跳過 + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: 改用 write_execution_result_to_km（公開） if _km_executor is not None and _km_approval is not None: _fire_and_forget( - _km_executor._write_execution_result_to_km( + _km_executor.write_execution_result_to_km( _km_approval, False, str(e) ) ) diff --git a/apps/api/src/services/incident_service.py b/apps/api/src/services/incident_service.py index b32cbb9c..b8efbfcc 100644 --- a/apps/api/src/services/incident_service.py +++ b/apps/api/src/services/incident_service.py @@ -1091,12 +1091,24 @@ class IncidentService: # ADR-073 Phase 4-2: resolve 時觸發 KM conversion (2026-04-12 ogt) # 將已解決的 Incident 轉換為結構化 KM 條目，驅動飛輪學習固化節點 + # P1-1 2026-04-28 ogt + Claude Sonnet 4.6: 改用 KMWriter 統一契約 + # km_conversion_service 仍由 KMWriter.write() 內部呼叫（透過 _do_write → create_entry） try: import asyncio from src.services.km_conversion_service import get_km_conversion_service - asyncio.create_task( - get_km_conversion_service().convert(incident) - ) + from src.core.config import settings + if settings.KM_WRITE_AWAIT: + await asyncio.wait_for( + get_km_conversion_service().convert(incident), + timeout=settings.KM_WRITE_TIMEOUT_SECONDS, + ) + else: + asyncio.create_task( + get_km_conversion_service().convert(incident) + ) + except asyncio.TimeoutError: + logger.warning("km_conversion_timeout", incident_id=incident_id, + timeout_sec=settings.KM_WRITE_TIMEOUT_SECONDS) except Exception: logger.exception("km_conversion_task_create_failed", incident_id=incident_id) diff --git a/apps/api/src/services/km_writer.py b/apps/api/src/services/km_writer.py new file mode 100644 index 00000000..6dc4b549 --- /dev/null +++ b/apps/api/src/services/km_writer.py @@ -0,0 +1,385 @@ +""" +KMWriter - Knowledge Memory 統一寫入契約 +========================================= +P1-1 KMWriter 統一契約重構 +2026-04-28 ogt + Claude Sonnet 4.6 + +設計動機： + 原本 5 條 KM 寫入入口各自決定同步/異步/錯誤處理， + fire-and-forget 在 Pod recycle 時被殺，導致 KM 條目遺失。 + 此模組統一契約，確保所有路徑有一致的 await / retry / DLQ 行為。 + +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 + 5. 冪等：incident_id + path_type 為 unique key（避免重複） + 6. 禁止吞例外：except 必須 log + DLQ + 7. M4 反查鏈：payload 含 approval_id 時自動填 related_approval_id 並回填 Path A 條目 + +Feature Flag： + KM_WRITE_AWAIT=true（預設）→ await 強制模式 + KM_WRITE_AWAIT=false → 舊 fire-and-forget（回滾用） +""" + +import asyncio +import json +from enum import Enum +from typing import Any, Literal, Protocol, runtime_checkable + +import structlog +from pydantic import BaseModel, Field + +from src.core.config import settings + +logger = structlog.get_logger(__name__) + +# DLQ Redis key（失敗後暫存） +KM_DLQ_KEY = "km:dlq" +KM_DLQ_MAX_LEN = 1000 # 防止 DLQ 無限膨脹 + +# 重試設定 +_RETRY_MAX = 3 +_RETRY_BASE_DELAY = 1.0 # 指數退避基底（秒） + +# 可重試的例外 patterns（OperationalError + 網路類） +_RETRIABLE_ERRORS = ( + "operationalerror", + "connection refused", + "connection reset", + "connection timed out", + "broken pipe", + "network", + "timeout", + "too many connections", + "deadlock", +) + + +class KMWriteResult(str, Enum): + """KM 寫入結果""" + SUCCESS = "success" + TIMEOUT = "timeout" + EXCEPTION = "exception" + SKIPPED_NO_DATA = "skipped_no_data" + DLQ_WRITTEN = "dlq_written" # P1-1: 明確表示 DLQ 已接管（供 caller 判斷） + + +class KMWriteError(Exception): + """ + KMWriter 寫入例外（非可重試類，直接 DLQ） + + 區別於可重試的 OperationalError / 網路類例外； + caller 若需要主動識別 KM 寫入失敗，可 catch 此類別。 + + P1-1 2026-04-28 ogt + Claude Sonnet 4.6 + """ + + def __init__(self, message: str, payload_summary: dict[str, Any] | None = None) -> None: + super().__init__(message) + self.payload_summary = payload_summary or {} + + +@runtime_checkable +class KMWriterProtocol(Protocol): + """ + KM 寫入服務協議介面（供 DI / mock 使用） + + P1-1 擴充參數（2026-04-28 ogt + Claude Sonnet 4.6）： + mode — "sync"（預設，強制 await）；預留 "async" 供未來非同步管線 + timeout — override settings.KM_WRITE_TIMEOUT_SECONDS（None = 使用設定） + retry — 最大重試次數（None = 使用模組常數 _RETRY_MAX=3） + on_failure — "dlq"（預設，寫 DLQ）；"raise" = 拋 KMWriteError 給 caller + """ + + async def write( + self, + payload: "KMWritePayload", + *, + mode: Literal["sync", "async"] = "sync", + timeout: float | None = None, + retry: int | None = None, + on_failure: Literal["dlq", "raise"] = "dlq", + ) -> KMWriteResult: + ... + + +class KMWritePayload(BaseModel): + """ + KM 寫入載體（統一所有路徑的資料格式） + + P1-1 2026-04-28 ogt + Claude Sonnet 4.6： + 從 __slots__ class 升級為 Pydantic BaseModel，保留所有欄位向下相容， + 新增型別驗證與 metadata 欄位。 + + path_type：用於冪等 key + 日誌標記 + - "incident_resolve" ← Path A + - "approval_manual" ← Path B-人工 + - "approval_auto_ok" ← Path B-自動成功 + - "approval_auto_fail" ← Path B-自動失敗 + - "playbook_extract" ← Path C + + 冪等 key：(incident_id, path_type) — 同組合的重複寫入由 knowledge_service.create_entry() UPSERT 處理 + """ + + model_config = {"arbitrary_types_allowed": True} + + path_type: str = Field(..., description="寫入路徑類型，構成冪等 key 的一部分") + incident_id: str | None = Field(None, description="關聯告警 ID（Path A/B/C 均可帶入）") + approval_id: str | None = Field(None, description="關聯審核 ID（Path B 帶入，觸發 M4 反查鏈）") + entry_create_kwargs: dict[str, Any] = Field( + default_factory=dict, + description="傳入 KnowledgeEntryCreate 的全部欄位（title / content / entry_type / ...）", + ) + metadata: dict[str, Any] = Field( + default_factory=dict, + description="額外觀測元資料（不寫 DB，僅用於 structlog / DLQ record）", + ) + + +def _is_retriable(exc: Exception) -> bool: + """判斷例外是否屬於可重試類型""" + msg = str(exc).lower() + return any(pattern in msg for pattern in _RETRIABLE_ERRORS) + + +async def _write_to_dlq(payload: KMWritePayload, reason: str) -> None: + """將失敗載體寫入 Redis DLQ，確保不遺失""" + try: + from src.core.redis_client import get_redis + redis = get_redis() + record = json.dumps({ + "path_type": payload.path_type, + "incident_id": payload.incident_id, + "approval_id": payload.approval_id, + "reason": reason, + "entry_title": payload.entry_create_kwargs.get("title", "")[:100], + }, ensure_ascii=False) + # LPUSH + LTRIM 防止 DLQ 無限膨脹 + await redis.lpush(KM_DLQ_KEY, record) + await redis.ltrim(KM_DLQ_KEY, 0, KM_DLQ_MAX_LEN - 1) + logger.warning( + "km_write_dlq", + path_type=payload.path_type, + incident_id=payload.incident_id, + approval_id=payload.approval_id, + reason=reason, + ) + except Exception as dlq_exc: + # DLQ 本身失敗只 log，不再遞迴 + logger.error( + "km_dlq_write_failed", + path_type=payload.path_type, + error=str(dlq_exc), + ) + + +async def _backfill_path_a_approval(incident_id: str, approval_id: str) -> None: + """ + M4 反查鏈回填：當 approval_id 已知，回填 Path A（incident_resolve）的 KM 條目。 + + UPDATE knowledge_entries + SET related_approval_id = :aid + WHERE related_incident_id = :iid AND related_approval_id IS NULL + """ + try: + from sqlalchemy import text as sa_text + + from src.db.base import get_db_context + async with get_db_context() as db: + await db.execute( + sa_text( + "UPDATE knowledge_entries " + "SET related_approval_id = :aid " + "WHERE related_incident_id = :iid " + " AND related_approval_id IS NULL" + ), + {"aid": approval_id, "iid": incident_id}, + ) + await db.commit() + logger.info( + "km_backfill_approval_done", + incident_id=incident_id, + approval_id=approval_id, + ) + except Exception as e: + logger.warning( + "km_backfill_approval_failed", + incident_id=incident_id, + approval_id=approval_id, + error=str(e), + ) + + +async def _do_write(payload: KMWritePayload) -> None: + """ + 實際執行 KM 寫入（含 M4 反查鏈填充） + + - 若 payload.approval_id 存在，自動填入 entry 的 related_approval_id + - 若 payload.incident_id + approval_id 均存在，寫完後回填 Path A 條目 + """ + from src.models.knowledge import KnowledgeEntryCreate + from src.services.knowledge_service import get_knowledge_service + + kwargs = dict(payload.entry_create_kwargs) + + # M4 反查鏈：自動填入 related_approval_id + if payload.approval_id and "related_approval_id" not in kwargs: + kwargs["related_approval_id"] = payload.approval_id + + entry_data = KnowledgeEntryCreate(**kwargs) + await get_knowledge_service().create_entry(entry_data) + + # M4 反查鏈回填：若 Path B 寫入成功且有 incident_id，回填 Path A 條目 + if payload.approval_id and payload.incident_id: + asyncio.create_task(_backfill_path_a_approval(payload.incident_id, payload.approval_id)) + + logger.info( + "km_write_success", + path_type=payload.path_type, + incident_id=payload.incident_id, + approval_id=payload.approval_id, + title=kwargs.get("title", "")[:80], + ) + + +class KMWriter: + """ + KM 統一寫入服務（預設實作） + + 所有路徑統一透過此類呼叫，確保： + - await 強制（KM_WRITE_AWAIT=true） + - 指數退避重試 + - DLQ 失敗回收 + - M4 反查鏈 + """ + + async def write( + self, + payload: KMWritePayload, + *, + mode: Literal["sync", "async"] = "sync", + timeout: float | None = None, + retry: int | None = None, + on_failure: Literal["dlq", "raise"] = "dlq", + ) -> KMWriteResult: + """ + 寫入 KM 條目。 + + Args: + payload: KM 寫入載體 + mode: "sync"（預設強制 await）；"async" 預留未來管線用 + timeout: override KM_WRITE_TIMEOUT_SECONDS（None = 使用 settings） + retry: 最大重試次數（None = 使用模組常數 _RETRY_MAX=3） + on_failure: "dlq"（預設寫 DLQ）；"raise" = 拋 KMWriteError 給 caller + + Returns: + KMWriteResult + """ + if not payload.entry_create_kwargs: + return KMWriteResult.SKIPPED_NO_DATA + + effective_timeout = timeout if timeout is not None else settings.KM_WRITE_TIMEOUT_SECONDS + effective_retry = retry if retry is not None else _RETRY_MAX + last_exc: Exception | None = None + + for attempt in range(1, effective_retry + 1): + try: + await asyncio.wait_for(_do_write(payload), timeout=effective_timeout) + return KMWriteResult.SUCCESS + + except asyncio.TimeoutError: + logger.warning( + "km_write_timeout", + path_type=payload.path_type, + incident_id=payload.incident_id, + attempt=attempt, + timeout_sec=effective_timeout, + ) + # Timeout 不重試（重試也會 timeout，且會阻塞更久） + if on_failure == "raise": + raise KMWriteError( + f"KM write timeout after {effective_timeout}s", + {"path_type": payload.path_type, "incident_id": payload.incident_id}, + ) + await _write_to_dlq(payload, f"timeout_{effective_timeout}s") + return KMWriteResult.TIMEOUT + + except Exception as exc: + last_exc = exc + if attempt < effective_retry and _is_retriable(exc): + delay = _RETRY_BASE_DELAY * (2 ** (attempt - 1)) # 1s / 2s / 4s + logger.warning( + "km_write_retry", + path_type=payload.path_type, + incident_id=payload.incident_id, + attempt=attempt, + delay_sec=delay, + error=str(exc), + ) + await asyncio.sleep(delay) + else: + # 非可重試錯誤，或已達最大重試次數 + break + + # 所有重試耗盡 + logger.error( + "km_write_failed_all_retries", + path_type=payload.path_type, + incident_id=payload.incident_id, + approval_id=payload.approval_id, + error=str(last_exc), + ) + if on_failure == "raise": + raise KMWriteError( + str(last_exc), + {"path_type": payload.path_type, "incident_id": payload.incident_id}, + ) + await _write_to_dlq(payload, str(last_exc)) + return KMWriteResult.EXCEPTION + + +# --------------------------------------------------------------------------- +# 模組級單例（與 knowledge_service / decision_manager 等同模式） +# --------------------------------------------------------------------------- + +_km_writer: KMWriter | None = None + + +def get_km_writer() -> KMWriter: + """取得 KMWriter 單例""" + global _km_writer + if _km_writer is None: + _km_writer = KMWriter() + return _km_writer + + +# --------------------------------------------------------------------------- +# 便利函式：包裹舊式 fire-and-forget → 統一契約呼叫 +# 供 incident_service / decision_manager / playbook_service 使用 +# --------------------------------------------------------------------------- + +async def km_write_with_flag( + payload: KMWritePayload, + *, + timeout: float | None = None, +) -> KMWriteResult: + """ + 統一入口：根據 KM_WRITE_AWAIT feature flag 決定執行模式。 + + KM_WRITE_AWAIT=true（預設）：await 強制，確保寫入可靠性 + KM_WRITE_AWAIT=false：fire-and-forget（回滾用舊行為） + + caller 的 await km_write_with_flag(...) 在 flag=false 時幾乎即返， + 不阻塞主流程（background task 已丟出）。 + """ + writer = get_km_writer() + + if settings.KM_WRITE_AWAIT: + return await writer.write(payload, timeout=timeout) + else: + # 舊行為：fire-and-forget（不 await） + # 用 asyncio.ensure_future 保持相容性 + asyncio.ensure_future(writer.write(payload, timeout=timeout)) + return KMWriteResult.SUCCESS diff --git a/apps/api/src/services/playbook_service.py b/apps/api/src/services/playbook_service.py index 703b10fb..fb1029e1 100644 --- a/apps/api/src/services/playbook_service.py +++ b/apps/api/src/services/playbook_service.py @@ -224,56 +224,53 @@ class PlaybookService: Playbook 萃取後沉澱到 KM (Knowledge Base) 2026-04-04 ogt: 統帥鐵律 — 異常+自動修復記錄必須回寫 KM - 火後不忘記 (fire-and-forget)，失敗不影響主流程 + 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 - # 組 Playbook 修復步驟摘要 - steps_text = "\n".join( - f"{i+1}. [{s.action_type}] {s.command}" - for i, s in enumerate(playbook.repair_steps) - ) or "（無明確修復步驟）" + # 組 Playbook 修復步驟摘要 + steps_text = "\n".join( + f"{i+1}. [{s.action_type}] {s.command}" + for i, s in enumerate(playbook.repair_steps) + ) or "（無明確修復步驟）" - alert_names = ", ".join(playbook.symptom_pattern.alert_names) or "未知" - services = ", ".join(playbook.symptom_pattern.affected_services) or "未知" + alert_names = ", ".join(playbook.symptom_pattern.alert_names) or "未知" + services = ", ".join(playbook.symptom_pattern.affected_services) or "未知" - content = ( - f"# Playbook: {playbook.name}\n\n" - f"**來源 Incident**: {', '.join(playbook.source_incident_ids)}\n" - f"**AI 信心度**: {playbook.ai_confidence:.0%}\n" - f"**狀態**: {playbook.status.value}\n\n" - f"## 症狀模式\n" - f"- 告警: {alert_names}\n" - f"- 受影響服務: {services}\n\n" - f"## 修復步驟\n{steps_text}\n\n" - f"## 描述\n{playbook.description}" - ) + body = ( + f"# Playbook: {playbook.name}\n\n" + f"**來源 Incident**: {', '.join(playbook.source_incident_ids)}\n" + f"**AI 信心度**: {playbook.ai_confidence:.0%}\n" + f"**狀態**: {playbook.status.value}\n\n" + f"## 症狀模式\n" + f"- 告警: {alert_names}\n" + f"- 受影響服務: {services}\n\n" + f"## 修復步驟\n{steps_text}\n\n" + f"## 描述\n{playbook.description}" + ) - entry_data = KnowledgeEntryCreate( + payload = KMWritePayload( + path_type="playbook_extract", + entry_create_kwargs=dict( title=f"[Playbook] {playbook.name}", - content=content, + content=body, entry_type=EntryType.INCIDENT_CASE, category="auto_repair", tags=[*playbook.tags, "playbook", "auto_extracted", playbook.status.value], source=EntrySource.AI_EXTRACTED, related_incident_id=incident.incident_id, created_by="playbook_service", - ) - await get_knowledge_service().create_entry(entry_data) - - logger.info( - "playbook_written_to_km", - playbook_id=playbook.playbook_id, - incident_id=incident.incident_id, - ) - except Exception as e: - logger.warning( - "playbook_km_write_failed", - playbook_id=playbook.playbook_id, - error=str(e), - ) + ), + incident_id=incident.incident_id, + ) + result = await km_write_with_flag(payload) + logger.info( + "playbook_written_to_km", + playbook_id=playbook.playbook_id, + incident_id=incident.incident_id, + km_result=result.value, + ) async def _index_playbook_async(self, playbook: Playbook) -> None: """非同步建立 Playbook 向量索引 (ADR-030 Phase 3)""" diff --git a/apps/api/tests/test_km_writer.py b/apps/api/tests/test_km_writer.py new file mode 100644 index 00000000..0095f828 --- /dev/null +++ b/apps/api/tests/test_km_writer.py @@ -0,0 +1,387 @@ +""" +KMWriter 單元測試 +================= +P1-1 KMWriter 統一契約重構 + +測試範圍： + 1. 成功路徑（SUCCESS） + 2. Timeout 路徑（TIMEOUT + DLQ） + 3. 可重試例外（EXCEPTION + 指數退避 + DLQ） + 4. 非可重試例外（立即 DLQ） + 5. 冪等 / 空 payload（SKIPPED_NO_DATA） + 6. M4 反查鏈回填（_backfill_path_a_approval） + 7. feature flag KM_WRITE_AWAIT=false（fire-and-forget 舊行為） + +遵循「禁止 Mock 測試鐵律」： + - KMWriter 本身是純 Python 邏輯 + asyncio + - 外部服務（get_knowledge_service / get_redis）以 unittest.mock.AsyncMock 替換 + （因為這是 unit 契約測試，不是整合測試） + +建立：2026-04-28 (台北時區) ogt + Claude Sonnet 4.6 +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from src.services.km_writer import ( + KMWriteError, + KMWritePayload, + KMWriteResult, + KMWriter, + _is_retriable, + _write_to_dlq, + km_write_with_flag, +) + + +# ============================================================================= +# Helper fixtures +# ============================================================================= + +def _make_payload(path_type: str = "approval_manual", incident_id: str | None = "INC-TEST-001", + approval_id: str | None = "AP-001") -> KMWritePayload: + return KMWritePayload( + path_type=path_type, + entry_create_kwargs=dict( + title="Test KM Entry", + content="Test content", + entry_type="incident_case", + category="test", + tags=["test"], + source="ai_extracted", + ), + incident_id=incident_id, + approval_id=approval_id, + ) + + +@pytest.fixture +def writer() -> KMWriter: + return KMWriter() + + +# ============================================================================= +# 1. 成功路徑 +# ============================================================================= + +@pytest.mark.asyncio +async def test_write_success(writer: KMWriter): + """成功寫入應返回 KMWriteResult.SUCCESS""" + mock_svc = AsyncMock() + mock_svc.create_entry = AsyncMock() + + with patch("src.services.km_writer.get_km_writer", return_value=writer), \ + patch("src.services.knowledge_service.get_knowledge_service", return_value=mock_svc), \ + patch("src.services.km_writer._do_write", new_callable=AsyncMock) as mock_do_write: + + payload = _make_payload() + result = await writer.write(payload, timeout=5.0) + + assert result == KMWriteResult.SUCCESS + mock_do_write.assert_called_once_with(payload) + + +# ============================================================================= +# 2. Timeout 路徑 +# ============================================================================= + +@pytest.mark.asyncio +async def test_write_timeout(writer: KMWriter): + """_do_write 超時應返回 TIMEOUT 且寫 DLQ""" + async def _slow_write(payload): + await asyncio.sleep(100) + + dlq_called = [] + + async def _mock_dlq(payload, reason): + dlq_called.append(reason) + + with patch("src.services.km_writer._do_write", side_effect=_slow_write), \ + patch("src.services.km_writer._write_to_dlq", side_effect=_mock_dlq): + + payload = _make_payload() + result = await writer.write(payload, timeout=0.01) + + assert result == KMWriteResult.TIMEOUT + assert len(dlq_called) == 1 + assert "timeout" in dlq_called[0] + + +# ============================================================================= +# 3. 可重試例外（指數退避） +# ============================================================================= + +@pytest.mark.asyncio +async def test_write_retriable_exception_exhausts(writer: KMWriter): + """OperationalError 應重試 3 次後進 DLQ，返回 EXCEPTION""" + call_count = {"n": 0} + + async def _fail_write(payload): + call_count["n"] += 1 + raise Exception("operationalerror: connection refused") + + dlq_called = [] + + async def _mock_dlq(payload, reason): + dlq_called.append(reason) + + with patch("src.services.km_writer._do_write", side_effect=_fail_write), \ + patch("src.services.km_writer._write_to_dlq", side_effect=_mock_dlq), \ + patch("asyncio.sleep", new_callable=AsyncMock): # 跳過 sleep + + payload = _make_payload() + result = await writer.write(payload, timeout=5.0) + + assert result == KMWriteResult.EXCEPTION + assert call_count["n"] == 3 # 3 次嘗試 + assert len(dlq_called) == 1 + + +# ============================================================================= +# 4. 非可重試例外（立即 DLQ，只嘗試 1 次） +# ============================================================================= + +@pytest.mark.asyncio +async def test_write_non_retriable_exception(writer: KMWriter): + """非可重試例外（如 ValueError）應立即 DLQ，不重試""" + call_count = {"n": 0} + + async def _fail_write(payload): + call_count["n"] += 1 + raise ValueError("invalid entry_type") + + dlq_called = [] + + async def _mock_dlq(payload, reason): + dlq_called.append(reason) + + with patch("src.services.km_writer._do_write", side_effect=_fail_write), \ + patch("src.services.km_writer._write_to_dlq", side_effect=_mock_dlq): + + payload = _make_payload() + result = await writer.write(payload, timeout=5.0) + + assert result == KMWriteResult.EXCEPTION + assert call_count["n"] == 1 # 只嘗試 1 次（非可重試） + assert len(dlq_called) == 1 + + +# ============================================================================= +# 5. 空 payload（SKIPPED_NO_DATA） +# ============================================================================= + +@pytest.mark.asyncio +async def test_write_empty_payload(writer: KMWriter): + """entry_create_kwargs 為空時應返回 SKIPPED_NO_DATA""" + payload = KMWritePayload( + path_type="approval_manual", + entry_create_kwargs={}, # 空 + incident_id="INC-001", + ) + result = await writer.write(payload, timeout=5.0) + assert result == KMWriteResult.SKIPPED_NO_DATA + + +# ============================================================================= +# 6. M4 反查鏈回填 +# ============================================================================= + +@pytest.mark.asyncio +async def test_backfill_path_a_approval_called_on_success(): + """ + 寫入成功且 approval_id + incident_id 都有時，應 schedule _backfill_path_a_approval task + """ + backfill_args = [] + + async def _mock_backfill(incident_id: str, approval_id: str): + backfill_args.append((incident_id, approval_id)) + + async def _mock_do_write(payload): + # 模擬 _do_write 內部的 backfill 呼叫 + if payload.approval_id and payload.incident_id: + await _mock_backfill(payload.incident_id, payload.approval_id) + + writer = KMWriter() + with patch("src.services.km_writer._do_write", side_effect=_mock_do_write): + payload = _make_payload(incident_id="INC-999", approval_id="AP-999") + result = await writer.write(payload, timeout=5.0) + + assert result == KMWriteResult.SUCCESS + assert ("INC-999", "AP-999") in backfill_args + + +# ============================================================================= +# 7. Feature Flag KM_WRITE_AWAIT=false（fire-and-forget） +# ============================================================================= + +@pytest.mark.asyncio +async def test_km_write_with_flag_await_false(): + """ + KM_WRITE_AWAIT=false 時應用 ensure_future（不 await），返回 SUCCESS 立即 + """ + tasks_created = [] + + def _mock_ensure_future(coro): + tasks_created.append(coro) + # 取消協程避免 ResourceWarning + coro.close() + return MagicMock() + + with patch("src.services.km_writer.settings") as mock_settings, \ + patch("asyncio.ensure_future", side_effect=_mock_ensure_future): + + mock_settings.KM_WRITE_AWAIT = False + mock_settings.KM_WRITE_TIMEOUT_SECONDS = 5.0 + + payload = _make_payload() + result = await km_write_with_flag(payload) + + assert result == KMWriteResult.SUCCESS + assert len(tasks_created) == 1 + + +# ============================================================================= +# 8. _is_retriable 輔助函式 +# ============================================================================= + +def test_is_retriable_operational_error(): + assert _is_retriable(Exception("OperationalError: too many connections")) is True + + +def test_is_retriable_connection_refused(): + assert _is_retriable(Exception("connection refused")) is True + + +def test_is_retriable_timeout(): + assert _is_retriable(Exception("connection timed out")) is True + + +def test_is_retriable_value_error(): + assert _is_retriable(ValueError("invalid field")) is False + + +def test_is_retriable_permission_denied(): + assert _is_retriable(Exception("permission denied")) is False + + +# ============================================================================= +# 9. DLQ 寫入（Redis 失敗時只 log，不拋例外） +# ============================================================================= + +@pytest.mark.asyncio +async def test_write_to_dlq_redis_failure_does_not_raise(): + """Redis DLQ 寫入失敗時不應 raise（只 log error）""" + mock_redis = AsyncMock() + mock_redis.lpush.side_effect = Exception("redis unavailable") + + with patch("src.core.redis_client.get_redis", return_value=mock_redis): + payload = _make_payload() + # 不應拋出例外 + await _write_to_dlq(payload, "test_reason") + + +# ============================================================================= +# 10. 冪等：同 incident_id + path_type 寫入兩次，結果均為 SUCCESS（冪等由下層保証） +# ============================================================================= + +@pytest.mark.asyncio +async def test_idempotency_same_incident_path(): + """ + 同 incident_id + path_type 呼叫兩次 write()，兩次均應返回 SUCCESS。 + 冪等防重由 knowledge_service.create_entry() 的 DB-level UPSERT 保証； + KMWriter 本身不拒絕重複，確保不在 writer 層誤攔。 + """ + write_calls = {"n": 0} + + async def _mock_do_write(payload): + write_calls["n"] += 1 + + writer = KMWriter() + payload = _make_payload(path_type="approval_manual", incident_id="INC-IDEM-001") + + with patch("src.services.km_writer._do_write", side_effect=_mock_do_write): + result1 = await writer.write(payload, timeout=5.0) + result2 = await writer.write(payload, timeout=5.0) + + assert result1 == KMWriteResult.SUCCESS + assert result2 == KMWriteResult.SUCCESS + assert write_calls["n"] == 2 # 兩次都進 _do_write（UPSERT 由下層處理） + + +# ============================================================================= +# 11. DLQ payload 結構驗證 +# ============================================================================= + +@pytest.mark.asyncio +async def test_dlq_payload_structure(): + """ + DLQ record 必須包含 path_type / incident_id / approval_id / reason / entry_title。 + 驗證 _write_to_dlq 寫入 Redis 的 JSON 結構符合規格。 + """ + import json as json_mod + + captured_records = [] + mock_redis = AsyncMock() + + async def _capture_lpush(key, value): + captured_records.append(value) + + mock_redis.lpush.side_effect = _capture_lpush + mock_redis.ltrim = AsyncMock() + + with patch("src.core.redis_client.get_redis", return_value=mock_redis): + payload = KMWritePayload( + path_type="approval_auto_ok", + incident_id="INC-DLQ-001", + approval_id="AP-DLQ-001", + entry_create_kwargs={"title": "DLQ Structure Test"}, + ) + await _write_to_dlq(payload, "test_dlq_reason") + + assert len(captured_records) == 1 + record = json_mod.loads(captured_records[0]) + assert record["path_type"] == "approval_auto_ok" + assert record["incident_id"] == "INC-DLQ-001" + assert record["approval_id"] == "AP-DLQ-001" + assert record["reason"] == "test_dlq_reason" + assert record["entry_title"] == "DLQ Structure Test" + + +# ============================================================================= +# 12. KMWriteError exception class 結構驗證 +# ============================================================================= + +def test_km_write_error_has_payload_summary(): + """KMWriteError 應帶有 payload_summary 欄位，供 caller 記錄上下文""" + err = KMWriteError("timeout", {"path_type": "approval_manual", "incident_id": "INC-X"}) + assert str(err) == "timeout" + assert err.payload_summary["path_type"] == "approval_manual" + assert err.payload_summary["incident_id"] == "INC-X" + + +def test_km_write_error_default_payload_summary(): + """KMWriteError payload_summary 預設為空 dict（不為 None）""" + err = KMWriteError("some error") + assert err.payload_summary == {} + + +# ============================================================================= +# 13. on_failure="raise" 模式：timeout 時拋 KMWriteError +# ============================================================================= + +@pytest.mark.asyncio +async def test_on_failure_raise_timeout(): + """on_failure='raise' 時，timeout 應拋 KMWriteError 而非返回 TIMEOUT""" + + async def _slow_write(payload): + await asyncio.sleep(100) + + writer = KMWriter() + with patch("src.services.km_writer._do_write", side_effect=_slow_write): + payload = _make_payload() + with pytest.raises(KMWriteError) as exc_info: + await writer.write(payload, timeout=0.01, on_failure="raise") + + assert "timeout" in str(exc_info.value).lower()