fix(governance): normalize knowledge degradation payloads

apps/api/src/services/failover_alerter.py

@@ -352,6 +352,34 @@ _IMPACT_PROFILES: dict[str, list[tuple[str, str]]] = {
     ],
 }
 
+_TOP_LEVEL_IMPACT_ALIASES: dict[str, dict[str, tuple[str, ...]]] = {
+    "knowledge_degradation": {
+        "stale_count": ("stale_count", "stale", "stale_km"),
+        "total_count": ("total_count", "total", "total_km"),
+        "stale_ratio": ("stale_ratio", "ratio"),
+        "threshold": ("threshold",),
+        "stale_days": ("stale_days",),
+    },
+}
+
+_TOP_LEVEL_FALLBACK_KEEP: dict[str, set[str]] = {
+    "knowledge_degradation": {
+        "automatable_work",
+        "next_action",
+        "next_step",
+        "ratio",
+        "stale",
+        "stale_count",
+        "stale_days",
+        "stale_km",
+        "stale_ratio",
+        "threshold",
+        "total",
+        "total_count",
+        "total_km",
+    },
+}
+
 
 def _event_display_name(event_type: str) -> str:
     if event_type in _EVENT_DISPLAY_NAMES:
@@ -421,6 +449,47 @@ def _governance_summary_lines(event_type: str, impact: dict[str, Any]) -> str:
     return _tree_lines(rows)
 
 
+def _normalized_impact(event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
+    impact = dict(_as_dict(payload.get("impact")))
+    for canonical_key, aliases in _TOP_LEVEL_IMPACT_ALIASES.get(event_type, {}).items():
+        if canonical_key in impact:
+            continue
+        for alias in aliases:
+            if alias in payload:
+                impact[canonical_key] = payload[alias]
+                break
+    return impact
+
+
+def _section_payload(
+    payload: dict[str, Any],
+    canonical_key: str,
+    *,
+    item_aliases: tuple[str, ...] = (),
+    next_action_aliases: tuple[str, ...] = (),
+) -> dict[str, Any]:
+    raw = payload.get(canonical_key)
+    section = dict(raw) if isinstance(raw, dict) else {}
+    if isinstance(raw, list) and "items" not in section:
+        section["items"] = raw
+
+    if "items" not in section:
+        for alias in item_aliases:
+            value = payload.get(alias)
+            if isinstance(value, list):
+                section["items"] = value
+                break
+
+    if "next_action" not in section:
+        for alias in next_action_aliases:
+            value = payload.get(alias)
+            if value:
+                section["next_action"] = value
+                break
+
+    return section
+
+
 def _governance_operator_context(event_type: str, impact: dict[str, Any]) -> list[str]:
     """Return operator-facing guidance for governance alerts.
 
@@ -481,9 +550,17 @@ def format_governance_alert_card(event_type: str, payload: dict[str, Any]) -> st
     轉成可掃描卡片，避免大量純文字欄位洗版。
     """
     payload = payload if isinstance(payload, dict) else {}
-    impact = _as_dict(payload.get("impact"))
-    remediation = _as_dict(payload.get("remediation"))
-    actionable = _as_dict(payload.get("actionable"))
+    impact = _normalized_impact(event_type, payload)
+    remediation = _section_payload(
+        payload,
+        "remediation",
+        next_action_aliases=("next_step", "next_action"),
+    )
+    actionable = _section_payload(
+        payload,
+        "actionable",
+        item_aliases=("automatable_work",),
+    )
     status = payload.get("status", "warning")
 
     sections: list[str] = [
@@ -516,9 +593,17 @@ def format_governance_alert_card(event_type: str, payload: dict[str, Any]) -> st
         sections.extend(["", "🤖 *可自動化工作*", actionable_lines])
 
     profiled_keys = {key for key, _label in _IMPACT_PROFILES.get(event_type, [])}
+    top_level_keep = _TOP_LEVEL_FALLBACK_KEEP.get(event_type, set())
     fallback_items = _fallback_pairs(
         payload,
-        keep={"status", "impact", "remediation", "actionable", *profiled_keys},
+        keep={
+            "status",
+            "impact",
+            "remediation",
+            "actionable",
+            *profiled_keys,
+            *top_level_keep,
+        },
         max_items=4,
     )
     if fallback_items:

apps/api/tests/test_failover_alerter.py

@@ -294,6 +294,38 @@ def test_governance_alert_card_formats_knowledge_degradation() -> None:
     assert "欄位快覽" not in card
 
 
+def test_governance_alert_card_accepts_legacy_knowledge_degradation_payload() -> None:
+    card = format_governance_alert_card(
+        "knowledge_degradation",
+        {
+            "status": "warning",
+            "stale_count": 1425,
+            "total": 1856,
+            "stale_ratio": 0.768,
+            "threshold": 0.2,
+            "stale_days": 7,
+            "remediation": [
+                "啟動 KM 反查與自動補齊流程",
+                "關鍵服務告警自動同步到 KM 任務",
+            ],
+            "next_step": "run_kb_growth_healthcheck",
+            "automatable_work": [
+                "每日檢查 ANTI_PATTERN 更新結果",
+                "安排至少 2 位 owner 對 stale 條目做快速人工審核",
+            ],
+        },
+    )
+
+    assert "1425 / 1856 筆 KM" in card
+    assert "陳舊 KM：1425" in card
+    assert "總 KM：1856" in card
+    assert "陳舊比例：76\\.8%" in card
+    assert "▶️ 下一步：run\\_kb\\_growth\\_healthcheck" in card
+    assert "每日檢查 ANTI\\_PATTERN 更新結果" in card
+    assert "📎 *補充欄位*" not in card
+    assert "? / ?" not in card
+
+
 def test_governance_alert_card_limits_fallback_fields() -> None:
     card = format_governance_alert_card(
         "custom_signal",