fix(arch-review): 首席架構師審查 S1×3 S2×3 S3×3 全修復 + ADR-064

S1 Critical:
- S1-1: asyncio 觸發移至 _call_with_fallback async 上下文，移除 sync 中的 get_event_loop()
- S1-2: _append_rule_to_yaml 加 textwrap.dedent() 正規化 LLM 輸出縮排
- S1-3: _matches() 對 alertname=["*"] 直接回傳 False，防意外命中

S2 Major:
- S2-1: auto_generate_rule() 改為 DI 參數注入 (ollama_url/model/gemini_api_key)，移除 import settings
- S2-4: _generate_mock_response docstring 澄清為規則引擎生產路徑，非假數據
- S2-5: suggested_action .strip() 防空白字串繞過 or

S3 Minor:
- S3-2: priority 上界 min(next, 890)
- S3-3: alertname sanitize re.sub([{}]) 防 format KeyError
- S3-4: model_registry.py 最後修改時間戳更新

文件:
- ADR-064: Alert Rule Engine YAML 驅動 + AI 自動學習
- Skills 02: 告警規則引擎 DI 規範 + asyncio 禁止事項
- Skills 03: _generate_mock_response 語意澄清 + 規則引擎降級流程
- LOGBOOK: 本次 Session 完整記錄

2026-04-09 ogt: 首席架構師審查修正

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/api/src/api/v1/webhooks.py

@@ -1396,7 +1396,7 @@ async def alertmanager_webhook(
                 risk_level=risk_level.value,
                 resource_name=target_resource,
                 root_cause=root_cause,
-                suggested_action=analysis_result.kubectl_command or analysis_result.suggested_action.value,
+                suggested_action=(analysis_result.kubectl_command or "").strip() or analysis_result.suggested_action.value,
                 estimated_downtime=estimated_downtime,
                 hit_count=1,
                 primary_responsibility=primary_responsibility,

apps/api/src/services/alert_rule_engine.py

@@ -99,14 +99,17 @@ def _load_rules() -> list[dict]:
 
 
 def _matches(rule: dict, alertname: str, alert_type: str, message: str) -> bool:
-    """判斷規則是否匹配"""
+    """判斷規則是否匹配。通用兜底規則（alertname=["*"]）永遠回傳 False，由 match_rule 單獨處理。"""
     match = rule.get("match", {})
 
-    # alertname 完全匹配
+    # S1-3 修正: 通用兜底規則不參與 _matches，防止其 alert_type/message 關鍵字意外命中
     alertnames = match.get("alertname", [])
-    if alertnames and alertnames != ["*"]:
-        if alertname in alertnames:
-            return True
+    if alertnames == ["*"]:
+        return False
+
+    # alertname 完全匹配
+    if alertnames and alertname in alertnames:
+        return True
 
     # alert_type 部分匹配
     for kw in match.get("alert_type", []):
@@ -279,12 +282,15 @@ def _append_rule_to_yaml(rule_yaml: str, alertname: str) -> bool:
             logger.warning("auto_rule_empty_response", alertname=alertname)
             return False
 
+        # S1-2 修正: dedent 正規化 LLM 可能輸出的前置空格，再加 2 spaces 縮排到 rules: 下
+        import textwrap
+        normalized = textwrap.dedent(rule_yaml.strip())
+
         # append 到 YAML 檔
         with RULES_FILE.open("a", encoding="utf-8") as f:
             now = datetime.now().strftime("%Y-%m-%d %H:%M")
             f.write(f"\n  # AUTO-GENERATED {now} — alertname={alertname}\n")
-            # indent list item under rules:
-            for line in rule_yaml.strip().splitlines():
+            for line in normalized.splitlines():
                 f.write(f"  {line}\n")
 
         # 清除 lru_cache 讓新規則立即生效
@@ -340,64 +346,84 @@ def _extract_yaml_block(text: str) -> str:
     return match.group(1).strip() if match else text
 
 
-async def auto_generate_rule(alert_context: dict) -> None:
+async def auto_generate_rule(
+    alert_context: dict,
+    ollama_url: str,
+    model: str,
+    gemini_api_key: str = "",
+) -> None:
     """
     非同步背景任務：呼叫 AI 為未知告警自動生成規則並寫入 alert_rules.yaml。
 
     觸發條件: match_rule() 命中 generic_fallback
-    流程: Ollama (deepseek-r1:14b) → 失敗則 Gemini → 驗證 → append YAML → 清除 cache
-    """
-    from src.core.config import settings
+    流程: Ollama → 失敗則 Gemini → 驗證格式 → append YAML → 清除 lru_cache 立即生效
 
+    Args:
+        alert_context: 告警上下文
+        ollama_url: Ollama endpoint（由呼叫方從 settings 注入，S2-1 DI 修正）
+        model: Ollama 模型名稱
+        gemini_api_key: Gemini API Key（空字串則跳過 Gemini 備援）
+
+    限制:
+        - 進程級去重 (_generating set)，多 Pod 環境可能重複生成（ADR-064 已記錄）
+        - 寫入後清除 lru_cache，同 Pod 立即生效；其他 Pod 需重啟
+    """
     labels = alert_context.get("labels", {})
     alertname = labels.get("alertname", alert_context.get("alert_type", "custom"))
 
+    # S3-3 修正: sanitize alertname，防止含 {/} 的 alertname 在 format() 中拋出 KeyError
+    alertname_safe = re.sub(r"[{}]", "", alertname)
+
     # 去重：同一 alertname 同時只跑一次
-    if alertname in _generating:
+    if alertname_safe in _generating:
         return
-    if _rule_id_exists(alertname):
-        logger.debug("auto_rule_skip_exists", alertname=alertname)
+    if _rule_id_exists(alertname_safe):
+        logger.debug("auto_rule_skip_exists", alertname=alertname_safe)
         return
 
-    _generating.add(alertname)
+    _generating.add(alertname_safe)
     try:
-        rule_id = re.sub(r"[^a-z0-9_]", "_", alertname.lower()).strip("_")
-        # priority: 500~899 給 AI 生成規則，不干擾手寫規則 (1-499)
+        rule_id = re.sub(r"[^a-z0-9_]", "_", alertname_safe.lower()).strip("_")
+
+        # S3-2 修正: priority 上界 890，防止超出 AI 生成範圍
         existing = [r.get("priority", 0) for r in _load_rules() if not _is_generic(r)]
-        priority = max((p for p in existing if 500 <= p < 900), default=499) + 10
+        next_priority = max((p for p in existing if 500 <= p < 900), default=499) + 10
+        priority = min(next_priority, 890)
 
         prompt = _AUTO_RULE_PROMPT.format(
-            alertname=alertname,
+            alertname=alertname_safe,
             alert_type=alert_context.get("alert_type", "custom"),
             message=alert_context.get("message", "")[:200],
-            labels=json.dumps({k: v for k, v in labels.items() if k in
-                               ("job", "instance", "severity", "namespace", "container", "name")},
-                              ensure_ascii=False),
+            labels=json.dumps(
+                {k: v for k, v in labels.items()
+                 if k in ("job", "instance", "severity", "namespace", "container", "name")},
+                ensure_ascii=False,
+            ),
             rule_id=rule_id,
             priority=priority,
         )
 
-        logger.info("auto_rule_generating", alertname=alertname, rule_id=rule_id)
+        logger.info("auto_rule_generating", alertname=alertname_safe, rule_id=rule_id)
 
         # 1. 先試 Ollama
-        raw = await _call_ollama(prompt, settings.OLLAMA_URL, settings.OPENCLAW_DEFAULT_MODEL)
+        raw = await _call_ollama(prompt, ollama_url, model)
 
         # 2. Ollama 失敗 → Gemini
-        if not raw and settings.GEMINI_API_KEY:
-            raw = await _call_gemini(prompt, settings.GEMINI_API_KEY)
+        if not raw and gemini_api_key:
+            raw = await _call_gemini(prompt, gemini_api_key)
 
         if not raw:
-            logger.warning("auto_rule_no_response", alertname=alertname)
+            logger.warning("auto_rule_no_response", alertname=alertname_safe)
             return
 
         yaml_block = _extract_yaml_block(raw)
-        success = _append_rule_to_yaml(yaml_block, alertname)
+        success = _append_rule_to_yaml(yaml_block, alertname_safe)
         if success:
-            logger.info("auto_rule_success", alertname=alertname, rule_id=rule_id)
+            logger.info("auto_rule_success", alertname=alertname_safe, rule_id=rule_id)
         else:
-            logger.warning("auto_rule_failed_validation", alertname=alertname)
+            logger.warning("auto_rule_failed_validation", alertname=alertname_safe)
 
     except Exception as e:
-        logger.error("auto_rule_exception", alertname=alertname, error=str(e))
+        logger.error("auto_rule_exception", alertname=alertname_safe, error=str(e))
     finally:
-        _generating.discard(alertname)
+        _generating.discard(alertname_safe)

apps/api/src/services/model_registry.py

@@ -12,7 +12,7 @@ Model Registry - Phase 12 P1 修復
 版本: v1.0
 建立: 2026-03-26 23:00 (台北時區)
 建立者: Claude Code
-最後修改: 2026-03-26 23:00 (台北時區)
+最後修改: 2026-04-09 10:00 (台北時區) — ogt: fallback config 更新為 deepseek-r1:14b + gemma3:4b
 修改者: Claude Code
 """
 

apps/api/src/services/openclaw.py

@@ -572,12 +572,17 @@ class OpenClawService:
         signoz_metrics: GoldMetrics | None = None,
     ) -> str:
         """
-        Mock LLM 回應生成器 - 規則引擎降級 (v8.0)
+        規則引擎降級回應 (v8.0) — 生產用途，不是假數據
 
-        從 alert_rules.yaml 載入規則，取代硬編碼 if/elif。
-        新增規則只需修改 YAML，不需要改代碼重新部署。
+        從 alert_rules.yaml 載入規則進行匹配，AI 分析失敗時的正式降級路徑。
+        命中 generic_fallback 時會回傳 rule_id="generic_fallback"，
+        由上層 async 方法（_call_with_fallback）觸發 auto_generate_rule() 學習新規則。
+
+        Returns:
+            (json_str, rule_id) tuple
 
         2026-04-09 ogt: 重構為規則引擎，移除 if/elif 硬編碼
+        2026-04-09 ogt: S2-4 架構師審查 — 修正 Mock 語意混淆，澄清為規則引擎生產路徑
         """
         from src.services.alert_rule_engine import match_rule
 
@@ -640,20 +645,9 @@ class OpenClawService:
             is_mock=True,
         )
 
-        # 2026-04-09 ogt: 命中通用兜底時，背景自動生成專屬規則
-        if rule_id == "generic_fallback":
-            from src.services.alert_rule_engine import auto_generate_rule
-            import asyncio
-            try:
-                loop = asyncio.get_event_loop()
-                if loop.is_running():
-                    loop.create_task(auto_generate_rule(alert_context))
-                else:
-                    asyncio.run(auto_generate_rule(alert_context))
-            except Exception as _e:
-                logger.warning("auto_rule_trigger_failed", error=str(_e))
-
-        return json.dumps(mock_response)
+        # 2026-04-09 ogt: rule_id 回傳給上層 async 方法觸發自動規則生成
+        # 不在此 sync 方法中呼叫 asyncio，避免 event loop 混用問題 (S1-1 架構師審查)
+        return json.dumps(mock_response), rule_id
 
     # =========================================================================
     # LLM Cache Layer (憲法要求: 嚴禁無快取裸奔)
@@ -871,7 +865,20 @@ class OpenClawService:
         # Mock Mode: 開發測試用
         if settings.MOCK_MODE:
             logger.info("mock_mode_enabled", using="mock_llm")
-            return self._generate_mock_response(alert_context or {}, signoz_metrics), "mock", True, 0, 0.0
+            _mock_json, _rule_id = self._generate_mock_response(alert_context or {}, signoz_metrics)
+            if _rule_id == "generic_fallback":
+                import asyncio
+                from src.services.alert_rule_engine import auto_generate_rule
+                try:
+                    asyncio.create_task(auto_generate_rule(
+                        alert_context or {},
+                        ollama_url=settings.OLLAMA_URL,
+                        model=settings.OPENCLAW_DEFAULT_MODEL,
+                        gemini_api_key=getattr(settings, "GEMINI_API_KEY", ""),
+                    ))
+                except Exception as _e:
+                    logger.warning("auto_rule_trigger_failed", error=str(_e))
+            return _mock_json, "mock", True, 0, 0.0
 
         # Phase 15.1 + 15.3: Langfuse 追蹤整合 + SignOz Deep Linking
         with langfuse_trace(
@@ -978,7 +985,20 @@ class OpenClawService:
             # 所有 Provider 失敗時，fallback 到 Mock (優雅降級)
             logger.warning("all_providers_failed_using_mock", fallback="mock_llm")
             trace.score(name="provider_success", value=0.0, comment="All providers failed, using mock")
-            return self._generate_mock_response(alert_context or {}, signoz_metrics), "mock_fallback", True, 0, 0.0
+            _mock_json, _rule_id = self._generate_mock_response(alert_context or {}, signoz_metrics)
+            if _rule_id == "generic_fallback":
+                import asyncio
+                from src.services.alert_rule_engine import auto_generate_rule
+                try:
+                    asyncio.create_task(auto_generate_rule(
+                        alert_context or {},
+                        ollama_url=settings.OLLAMA_URL,
+                        model=settings.OPENCLAW_DEFAULT_MODEL,
+                        gemini_api_key=getattr(settings, "GEMINI_API_KEY", ""),
+                    ))
+                except Exception as _e:
+                    logger.warning("auto_rule_trigger_failed", error=str(_e))
+            return _mock_json, "mock_fallback", True, 0, 0.0
 
     def _get_model_name(self, provider: str) -> str:
         """取得 provider 對應的模型名稱 (從 ModelRegistry)"""