feat(ai_advisory): P0 修 leader lock + inline keyboard + callback handler

統帥 2026-04-19 截圖反饋:
  1. 同一告警 22:44 連推 2 則 (多 Pod 都跑 daily loop)
  2. 純文字無按鈕 (無 feedback 閉環 / AI 只建議不執行)

新增 services/ai_advisory_helpers.py (~240 行):
  - try_acquire_daily_lock(job_name): Redis SETNX key 'aiops:daily_lock:{job}:{date}',
    TTL 25h,fail-open (Redis 掛照推,不阻塞).
  - try_acquire_hourly_lock(job_name): 同上 hourly 版 (coverage_evaluator 用).
  - is_snoozed / set_snooze: Redis key 'aiops:snooze:{type}:{target}' TTL 24h.
  - build_ai_advisory_keyboard: 統一 4 按鈕
      ✅ 已處理 / 😴 忽略 24h / 🔍 查看詳情 / 📋 產 kubectl 指令
    callback_data 格式: 'ai_advisory_{action}:{type}:{id}'
  - handle_ai_advisory_callback: 處理 handled/snooze 兩個 action 寫 aol.output.human_feedback,
    view/produce_cmd 留 P1.

4 個 LLM scanner 改用 helper:
  - capacity_forecaster: daily_lock + snooze check per host + 按鈕
  - compliance_scanner: daily_lock (cron only) + snooze per date + 按鈕
  - coverage_evaluator: hourly_lock + snooze per worst_dimension + 按鈕
  - hermes_rule_quality: daily_lock + snooze per primary rule + 按鈕

telegram_gateway.py:
  handle_callback 加 'ai_advisory_*' 路由 (step 1.85 drift 後)
  新增 _handle_ai_advisory_action 方法:
    解析 payload 'type:id' → 呼叫 handle_ai_advisory_callback
    → answer_callback (Telegram toast 回饋)
    → 返回 dict (info_action=True for view/produce_cmd)

統帥鐵律對齊:
  ✅ 多 Pod 場景只 leader 推 (Redis SETNX 保證冪等)
  ✅ 失敗 fail-open 不阻塞主業務 (Redis 掛仍能運作)
  ✅ aol.output 加 human_feedback 供 AI 學習
  ✅ snooze 避免重複告警 (24h TTL)
  ✅ 原 drift 按鈕 pattern 複用 (non-breaking)

明早 AI 將收到:
  - 單一訊息 (非重複)
  - 含 4 按鈕 (手動 feedback 閉環)
  - snooze 後同主題 24h 不再推

view/produce_cmd P1 留下 session (AI 主動 MCP 蒐證 + LLM 產 kubectl command).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/api/src/jobs/capacity_forecaster_job.py

@@ -95,7 +95,17 @@ async def run_capacity_forecaster_loop() -> None:
 
 
 async def forecast_once() -> dict[str, Any]:
-    """跑一次預測,對每個高風險 host 留痕 + LLM 分析 + 推 Telegram."""
+    """跑一次預測,對每個高風險 host 留痕 + LLM 分析 + 推 Telegram.
+
+    2026-04-19 P0 修 (統帥截圖反饋): 加 leader_lock 避免多 Pod 重複推.
+    """
+    from src.services.ai_advisory_helpers import try_acquire_daily_lock
+
+    # Leader lock: 只 leader Pod 跑,其他 skip
+    if not await try_acquire_daily_lock("capacity_forecaster"):
+        logger.info("capacity_forecast_skipped_not_leader")
+        return {"skipped": "not_leader"}
+
     started_ms = _time.time()
     stats: dict[str, Any] = {
         "queries_run": 0, "high_risk_hosts": 0, "recommendations": 0, "llm_analyzed": 0,
@@ -306,21 +316,40 @@ async def _send_telegram_forecast(
     risks: dict[str, list[dict[str, Any]]],
     llm_analyses: dict[str, dict[str, Any]] | None = None,
 ) -> bool:
-    """推 Telegram 預測摘要 (含 LLM 分析)."""
+    """推 Telegram 預測摘要 (含 LLM 分析 + 互動按鈕).
+
+    2026-04-19 P0 修 (統帥截圖反饋): 加 snooze check + inline_keyboard 4 按鈕
+    (✅ 已處理 / 😴 忽略 24h / 🔍 查看詳情 / 📋 產 kubectl 指令).
+    """
     try:
         import html
+        from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed
         from src.services.telegram_gateway import get_telegram_gateway
 
         if not settings.OPENCLAW_TG_CHAT_ID:
             return False
 
+        # Snooze check: 過濾掉被人工 snooze 的 host (按「忽略 24h」後)
+        active_risks = {}
+        skipped_hosts: list[str] = []
+        for host, findings in risks.items():
+            if await is_snoozed("capacity_forecast", host):
+                skipped_hosts.append(host)
+                continue
+            active_risks[host] = findings
+
+        if not active_risks:
+            logger.info("capacity_forecast_all_snoozed", total=len(risks))
+            return False
+
         llm_analyses = llm_analyses or {}
         lines = [
             "📈 <b>容量預測 (Phase 4 AI 升級版)</b>",
-            f"未來 7 天高風險 host: {len(risks)} 台",
+            f"未來 7 天高風險 host: {len(active_risks)} 台"
+            + (f" (含 {len(skipped_hosts)} 台已忽略)" if skipped_hosts else ""),
             "",
         ]
-        for host, findings in list(risks.items())[:8]:
+        for host, findings in list(active_risks.items())[:8]:
             lines.append(f"🟡 <code>{html.escape(host)}</code>")
             for f in findings[:3]:
                 lines.append(f"   ▸ {html.escape(f['reason'])} (value={f['value']:.2f})")
@@ -335,12 +364,19 @@ async def _send_telegram_forecast(
                     detail = html.escape(str(act.get("action", ""))[:100])
                     lines.append(f"   ▸ [{pri}] {detail}")
             else:
-                # LLM fallback: 用 hardcoded _derive_actions
                 actions = _derive_actions(findings)
                 if actions:
                     lines.append(f"   建議: {html.escape(actions[0])[:100]}")
             lines.append("")
-        lines.append("決策: 人工評估擴容/清理時機")
+
+        # advisory_id 用第一個 host (snooze / aol 對應用)
+        primary_host = next(iter(active_risks.keys()))
+        keyboard = build_ai_advisory_keyboard(
+            advisory_type="capacity_forecast",
+            advisory_id=primary_host,
+            include_view=True,
+            include_produce_cmd=True,
+        )
 
         msg = "\n".join(lines)
 
@@ -350,6 +386,7 @@ async def _send_telegram_forecast(
             "text": msg,
             "parse_mode": "HTML",
             "disable_web_page_preview": True,
+            "reply_markup": keyboard,
         })
         return True
     except Exception as e:

apps/api/src/jobs/compliance_scanner_job.py

@@ -80,7 +80,15 @@ async def scan_once(triggered_by: str = "cron") -> dict[str, int]:
 
     2026-04-19 Gap 3.2 LLM 升級: scan 完後若有 warnings/violations,
     用 LLM 分析整體 compliance posture + top 3 優先建議.
+    2026-04-19 P0 修: 加 leader_lock 避免多 Pod 重複推 Telegram.
     """
+    from src.services.ai_advisory_helpers import try_acquire_daily_lock
+
+    # Leader lock (cron 觸發才鎖,手動觸發不鎖)
+    if triggered_by == "cron" and not await try_acquire_daily_lock("compliance_scanner"):
+        logger.info("compliance_scan_skipped_not_leader")
+        return {"skipped": "not_leader"}
+
     started_ms = _time.time()
     stats: dict[str, Any] = {
         "assets_scanned": 0, "snapshots_written": 0, "violations": 0, "warnings": 0,
@@ -459,15 +467,23 @@ async def _send_telegram_posture(
     stats: dict[str, Any],
     analysis: dict[str, Any],
 ) -> None:
-    """推 Telegram 合規摘要."""
+    """推 Telegram 合規摘要 + 互動按鈕 (P0 修)."""
     try:
         import html
         from src.core.config import settings
+        from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed
         from src.services.telegram_gateway import get_telegram_gateway
 
         if not settings.OPENCLAW_TG_CHAT_ID:
             return
 
+        # Snooze check (advisory_id 用當日 date 即可,一天只能 snooze 一次)
+        from src.utils.timezone import now_taipei
+        today = now_taipei().date().isoformat()
+        if await is_snoozed("compliance_posture", today):
+            logger.info("compliance_posture_snoozed", date=today)
+            return
+
         grade = analysis.get("posture_grade", "?")
         grade_emoji = {"A": "🟢", "B": "🟡", "C": "🟠", "D": "🔴", "F": "⛔"}.get(grade, "⚠️")
         risk = analysis.get("risk_level", "?")
@@ -501,12 +517,19 @@ async def _send_telegram_posture(
         lines.append("決策: 人工評估各項修復優先")
 
         msg = "\n".join(lines)
+        keyboard = build_ai_advisory_keyboard(
+            advisory_type="compliance_posture",
+            advisory_id=today,
+            include_view=False,
+            include_produce_cmd=False,
+        )
         tg = get_telegram_gateway()
         await tg._send_request("sendMessage", {  # type: ignore[attr-defined]
             "chat_id": settings.OPENCLAW_TG_CHAT_ID,
             "text": msg,
             "parse_mode": "HTML",
             "disable_web_page_preview": True,
+            "reply_markup": keyboard,
         })
     except Exception as e:
         logger.warning("compliance_telegram_failed", error=str(e))

apps/api/src/jobs/coverage_evaluator_job.py

@@ -72,7 +72,15 @@ async def evaluate_once() -> dict[str, int]:
       + auto_remediation: remediation_events 過去 30d 有 target match asset.name
       + auto_rule_matching: incidents 過去 30d 有 asset match (alertname+affected_services)
       + auto_rule_creation: alert_rule_catalog source='ai_generated' 覆蓋 asset
+
+    2026-04-19 P0 修: 加 hourly_lock 避免多 Pod 重複推 + LLM 分析.
     """
+    from src.services.ai_advisory_helpers import try_acquire_hourly_lock
+
+    if not await try_acquire_hourly_lock("coverage_evaluator"):
+        logger.info("coverage_evaluate_skipped_not_leader")
+        return {"skipped": "not_leader"}
+
     started_ms = _time.time()
     stats = {
         "monitoring_updated": 0, "alerting_updated": 0, "km_updated": 0,
@@ -275,15 +283,22 @@ async def _send_telegram_gaps(
     red_summary: dict[str, Any],
     analysis: dict[str, Any],
 ) -> None:
-    """推 coverage 缺口 Telegram 摘要."""
+    """推 coverage 缺口 Telegram 摘要 + 互動按鈕 (P0 修)."""
     try:
         import html
         from src.core.config import settings
+        from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed
         from src.services.telegram_gateway import get_telegram_gateway
 
         if not settings.OPENCLAW_TG_CHAT_ID:
             return
 
+        # Snooze check: 以 worst_dimension 為 key
+        worst_dim = str(analysis.get("worst_dimension", "unknown"))
+        if await is_snoozed("coverage_gap", worst_dim):
+            logger.info("coverage_gap_snoozed", dim=worst_dim)
+            return
+
         worst = html.escape(str(analysis.get("worst_dimension", "")))
         cause = html.escape(str(analysis.get("root_cause", ""))[:200])
         weeks = analysis.get("estimated_weeks_to_close", "?")
@@ -309,12 +324,19 @@ async def _send_telegram_gaps(
         lines.append("決策: 人工評估補覆蓋排程")
 
         msg = "\n".join(lines)
+        keyboard = build_ai_advisory_keyboard(
+            advisory_type="coverage_gap",
+            advisory_id=worst_dim,
+            include_view=False,
+            include_produce_cmd=False,
+        )
         tg = get_telegram_gateway()
         await tg._send_request("sendMessage", {  # type: ignore[attr-defined]
             "chat_id": settings.OPENCLAW_TG_CHAT_ID,
             "text": msg,
             "parse_mode": "HTML",
             "disable_web_page_preview": True,
+            "reply_markup": keyboard,
         })
     except Exception as e:
         logger.warning("coverage_telegram_failed", error=str(e))

apps/api/src/jobs/hermes_rule_quality_job.py

@@ -64,7 +64,16 @@ async def run_hermes_rule_quality_loop() -> None:
 
 
 async def analyze_once() -> dict[str, int]:
-    """一次分析: 找噪音 rule + LLM 分析真因 + 推建議 + aol 留痕."""
+    """一次分析: 找噪音 rule + LLM 分析真因 + 推建議 + aol 留痕.
+
+    2026-04-19 P0 修: 加 daily leader_lock 避免多 Pod 重複推.
+    """
+    from src.services.ai_advisory_helpers import try_acquire_daily_lock
+
+    if not await try_acquire_daily_lock("hermes_rule_quality"):
+        logger.info("hermes_analyze_skipped_not_leader")
+        return {"skipped": "not_leader"}
+
     started_ms = _time.time()
     stats = {"noisy_rules": 0, "llm_analyzed": 0, "advisories_written": 0, "telegram_sent": 0}
     error_msg: str | None = None
@@ -289,16 +298,23 @@ async def _send_telegram_summary(
     noisy: list[dict[str, Any]],
     llm_analyses: dict[str, dict[str, Any]] | None = None,
 ) -> bool:
-    """推 Telegram 摘要訊息給 SRE group,含 LLM 分析結果."""
+    """推 Telegram 摘要訊息給 SRE group,含 LLM 分析結果 + 互動按鈕 (P0 修)."""
     try:
         import html
         from src.core.config import settings
+        from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed
         from src.services.telegram_gateway import get_telegram_gateway
 
         if not settings.OPENCLAW_TG_CHAT_ID:
             logger.info("hermes_telegram_skip_no_chat_id")
             return False
 
+        # Snooze check: 以第一條 noisy rule_name 為 key
+        primary_rule = noisy[0]["rule_name"] if noisy else "unknown"
+        if await is_snoozed("rule_quality", primary_rule):
+            logger.info("hermes_rule_snoozed", rule=primary_rule)
+            return False
+
         llm_analyses = llm_analyses or {}
         lines = [
             "🔍 <b>Hermes 規則品質檢測 (AI 分析)</b>",
@@ -327,14 +343,20 @@ async def _send_telegram_summary(
         lines.append("決策: 人工 UPDATE alert_rule_catalog SET review_status='deprecated' WHERE rule_name='...'")
 
         msg = "\n".join(lines)
+        keyboard = build_ai_advisory_keyboard(
+            advisory_type="rule_quality",
+            advisory_id=primary_rule,
+            include_view=False,
+            include_produce_cmd=False,
+        )
 
         tg = get_telegram_gateway()
-        # 直接用 telegram_gateway._send_request 送一般訊息
         await tg._send_request("sendMessage", {  # type: ignore[attr-defined]
             "chat_id": settings.OPENCLAW_TG_CHAT_ID,
             "text": msg,
             "parse_mode": "HTML",
             "disable_web_page_preview": True,
+            "reply_markup": keyboard,
         })
         return True
     except Exception as e:

apps/api/src/services/ai_advisory_helpers.py

@@ -0,0 +1,262 @@
+"""
+AI Advisory Helpers — Leader Lock + Snooze + Inline Keyboard (ADR-092 升級)
+=============================================================================
+統帥 2026-04-19 反饋:
+  - 多 Pod 重複推 Telegram (leader 機制缺)
+  - 純文字無按鈕 (feedback 無閉環)
+  - AI 只「建議」不「執行」(view_detail 缺)
+
+本 helper 提供 4 LLM scanner 共用:
+  1. daily_job_leader_lock: Redis SETNX,只 leader Pod 跑 daily job (避免 replica 重複)
+  2. is_snoozed: 檢查是否被 snooze 24h (避免同 host/rule 重複告警)
+  3. build_ai_advisory_keyboard: 統一 4 按鈕 (已處理/忽略 24h/查看詳情/產指令)
+
+callback_data 格式 (與 telegram_gateway 對接):
+  'ai_advisory_{action}:{advisory_type}:{advisory_id}'
+  action: handled / snooze / view / produce_cmd
+  advisory_type: capacity_forecast / compliance_posture / rule_quality / coverage_gap
+  advisory_id: aol.op_id (UUID) or 組合 key (host@date)
+
+2026-04-19 ogt + Claude Opus 4.7 (1M context) Asia/Taipei
+ADR-092 § AI Decision LLM 擴展層 P0 修復
+"""
+from __future__ import annotations
+
+from datetime import date as _date
+from typing import Any
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+
+# ============================================================================
+# Leader Lock — 多 Pod 場景只 leader 跑 daily job
+# ============================================================================
+
+async def try_acquire_daily_lock(job_name: str) -> bool:
+    """
+    用 Redis SETNX 搶當日 lock. 只搶到的 Pod 跑 daily job.
+
+    key: aiops:daily_lock:{job_name}:{YYYY-MM-DD-Taipei}
+    TTL: 25h (覆蓋 daily tick + 時差保險)
+
+    Returns:
+        True — 搶到 lock (此 Pod 跑)
+        False — 其他 Pod 已搶 (此 Pod skip)
+
+    失敗安全: Redis 掛 → 視為搶到 (fail-open,不阻塞主流程,可能造成一次多 Pod 推送,但好過不推)
+    """
+    try:
+        from src.core.redis_client import get_redis
+        from src.utils.timezone import now_taipei
+
+        today = now_taipei().date().isoformat()
+        key = f"aiops:daily_lock:{job_name}:{today}"
+
+        redis = await get_redis()
+        # NX: 只有不存在才設. EX 25h. 回 True/None
+        acquired = await redis.set(key, "1", nx=True, ex=25 * 3600)
+        if acquired:
+            logger.info("daily_lock_acquired", job=job_name, date=today, key=key)
+            return True
+        logger.info("daily_lock_skipped_other_pod", job=job_name, date=today)
+        return False
+    except Exception as e:
+        logger.warning("daily_lock_redis_failed_fail_open", job=job_name, error=str(e))
+        return True  # fail-open
+
+
+async def try_acquire_hourly_lock(job_name: str) -> bool:
+    """
+    類似 daily 但 hourly granularity. 適用於 coverage_evaluator 等每 1h 跑的 scanner.
+
+    key: aiops:hourly_lock:{job_name}:{YYYY-MM-DD-HH-Taipei}
+    TTL: 75 分鐘 (覆蓋 hourly tick + 緩衝)
+    """
+    try:
+        from src.core.redis_client import get_redis
+        from src.utils.timezone import now_taipei
+
+        now = now_taipei()
+        slot = f"{now.date().isoformat()}-{now.hour:02d}"
+        key = f"aiops:hourly_lock:{job_name}:{slot}"
+
+        redis = await get_redis()
+        acquired = await redis.set(key, "1", nx=True, ex=75 * 60)
+        if acquired:
+            logger.info("hourly_lock_acquired", job=job_name, slot=slot)
+            return True
+        logger.info("hourly_lock_skipped_other_pod", job=job_name, slot=slot)
+        return False
+    except Exception as e:
+        logger.warning("hourly_lock_redis_failed_fail_open", job=job_name, error=str(e))
+        return True
+
+
+# ============================================================================
+# Snooze — 按「忽略 24h」後避免重複告警
+# ============================================================================
+
+async def is_snoozed(advisory_type: str, target: str) -> bool:
+    """
+    檢查 advisory 是否被 snooze (人工按「忽略 24h」後).
+
+    key: aiops:snooze:{advisory_type}:{target}
+    target: host IP / rule_name / worst_dimension 等
+
+    Returns:
+        True — 被 snooze,應跳過 Telegram 推送
+        False — 未 snooze 或 Redis 失敗
+    """
+    try:
+        from src.core.redis_client import get_redis
+
+        redis = await get_redis()
+        key = f"aiops:snooze:{advisory_type}:{target}"
+        val = await redis.get(key)
+        return val is not None
+    except Exception as e:
+        logger.warning("snooze_check_failed", type=advisory_type, target=target, error=str(e))
+        return False
+
+
+async def set_snooze(advisory_type: str, target: str, hours: int = 24) -> None:
+    """設 snooze TTL — 供 callback handler 呼叫."""
+    try:
+        from src.core.redis_client import get_redis
+
+        redis = await get_redis()
+        key = f"aiops:snooze:{advisory_type}:{target}"
+        await redis.set(key, "1", ex=hours * 3600)
+        logger.info("snooze_set", type=advisory_type, target=target, hours=hours)
+    except Exception as e:
+        logger.warning("snooze_set_failed", type=advisory_type, target=target, error=str(e))
+
+
+# ============================================================================
+# Inline Keyboard — 統一 AI advisory 按鈕格式
+# ============================================================================
+
+def build_ai_advisory_keyboard(
+    advisory_type: str,
+    advisory_id: str,
+    include_view: bool = True,
+    include_produce_cmd: bool = False,
+) -> dict[str, Any]:
+    """
+    建 AI advisory 訊息的 inline_keyboard.
+
+    callback_data 格式: 'ai_advisory_{action}:{advisory_type}:{advisory_id}'
+
+    Args:
+        advisory_type: capacity_forecast / compliance_posture / rule_quality / coverage_gap
+        advisory_id: aol.op_id 或組合 key
+        include_view: 是否含「🔍 查看詳情」按鈕 (P1,目前多數 disabled)
+        include_produce_cmd: 是否含「📋 產指令」按鈕 (P1,需 LLM 產 kubectl command)
+
+    Returns:
+        dict: {"inline_keyboard": [[...]]} 可直接傳給 Telegram sendMessage reply_markup
+    """
+    cb = lambda action: f"ai_advisory_{action}:{advisory_type}:{advisory_id}"  # noqa: E731
+
+    row1 = [
+        {"text": "✅ 已處理", "callback_data": cb("handled")},
+        {"text": "😴 忽略 24h", "callback_data": cb("snooze")},
+    ]
+    row2 = []
+    if include_view:
+        row2.append({"text": "🔍 查看詳情", "callback_data": cb("view")})
+    if include_produce_cmd:
+        row2.append({"text": "📋 產 kubectl 指令", "callback_data": cb("produce_cmd")})
+
+    keyboard = {"inline_keyboard": [row1]}
+    if row2:
+        keyboard["inline_keyboard"].append(row2)
+    return keyboard
+
+
+# ============================================================================
+# Callback Handler — 處理「已處理」/「忽略 24h」按鈕
+# ============================================================================
+
+async def handle_ai_advisory_callback(
+    action: str,
+    advisory_type: str,
+    advisory_id: str,
+    username: str,
+) -> dict[str, Any]:
+    """
+    處理 ai_advisory_* callback.
+
+    Args:
+        action: handled / snooze / view / produce_cmd
+        advisory_type: 4 種 advisory 類型
+        advisory_id: 對應 aol.op_id 或組合 key
+
+    Returns:
+        {success: bool, feedback_text: str (給 Telegram answer_callback 顯示)}
+
+    邏輯:
+      - handled:   UPDATE aol.output jsonb += {human_feedback, by, at}
+      - snooze:    set_snooze(24h) + UPDATE aol
+      - view:      (P1 TODO — 未來 LLM 產詳情)
+      - produce_cmd: (P1 TODO — 未來 LLM 產 kubectl command)
+    """
+    from datetime import datetime
+    from src.utils.timezone import now_taipei
+
+    feedback_text = ""
+    success = False
+
+    if action == "handled":
+        success = await _write_human_feedback(advisory_id, "handled", username)
+        feedback_text = f"✅ 已記錄「已處理」by {username}"
+    elif action == "snooze":
+        # advisory_id 本身就是 target (host / rule_name 等)
+        await set_snooze(advisory_type, advisory_id, hours=24)
+        success = await _write_human_feedback(advisory_id, "snoozed_24h", username)
+        feedback_text = "😴 已忽略 24 小時"
+    elif action == "view":
+        feedback_text = "🔍 詳情功能下階段實作"  # P1
+    elif action == "produce_cmd":
+        feedback_text = "📋 產指令功能下階段實作"  # P1
+    else:
+        feedback_text = f"❓ 未知 action: {action}"
+
+    return {"success": success, "feedback_text": feedback_text}
+
+
+async def _write_human_feedback(advisory_id: str, feedback: str, username: str) -> bool:
+    """UPDATE aol.output.human_feedback 供 AI 學習."""
+    try:
+        from sqlalchemy import text as _sql
+        from src.db.base import get_db_context
+        from src.utils.timezone import now_taipei
+
+        async with get_db_context() as db:
+            await db.execute(
+                _sql("""
+                    UPDATE automation_operation_log
+                    SET output = output
+                        || jsonb_build_object(
+                            'human_feedback', :fb,
+                            'human_feedback_by', :who,
+                            'human_feedback_at', :at
+                        )
+                    WHERE op_id::text = :aid
+                       OR (input::jsonb->>'advisory_id' = :aid)
+                       OR (output::jsonb->>'host' = :aid)
+                       OR (output::jsonb->>'rule_name' = :aid)
+                """),
+                {
+                    "fb": feedback,
+                    "who": username[:50],
+                    "at": now_taipei().isoformat(),
+                    "aid": advisory_id,
+                },
+            )
+        return True
+    except Exception as e:
+        logger.warning("write_human_feedback_failed", advisory_id=advisory_id, error=str(e))
+        return False

apps/api/src/services/telegram_gateway.py

@@ -2153,6 +2153,67 @@ class TelegramGateway:
                 "parse_mode": "HTML",
             })
 
+    async def _handle_ai_advisory_action(
+        self,
+        action: str,
+        advisory_payload: str,  # 格式: '{type}:{id}'
+        callback_query_id: str,
+        user_id: int,
+        username: str,
+        user: dict,
+    ) -> dict:
+        """
+        2026-04-19 P0 修 (ADR-092): 處理 4 LLM scanner 的互動按鈕.
+
+        action: ai_advisory_handled / ai_advisory_snooze / ai_advisory_view / ai_advisory_produce_cmd
+        advisory_payload: '{advisory_type}:{advisory_id}' (nonce 解析後的 approval_id 位置)
+
+        流程:
+          1. 解析 payload → advisory_type + advisory_id
+          2. 呼叫 ai_advisory_helpers.handle_ai_advisory_callback
+          3. answer_callback (Telegram 按鈕回饋 toast)
+          4. 編輯原訊息尾部加「✅ 已處理 by user@時間」
+        """
+        try:
+            # 解析 '{type}:{id}'
+            if ":" in advisory_payload:
+                advisory_type, advisory_id = advisory_payload.split(":", 1)
+            else:
+                advisory_type, advisory_id = "unknown", advisory_payload
+
+            # action 去掉 'ai_advisory_' 前綴 → 得到純 action 名 (handled/snooze/view/produce_cmd)
+            pure_action = action.replace("ai_advisory_", "", 1)
+
+            logger.info(
+                "ai_advisory_callback",
+                action=pure_action, advisory_type=advisory_type,
+                advisory_id=advisory_id, user=username,
+            )
+
+            from src.services.ai_advisory_helpers import handle_ai_advisory_callback
+            result = await handle_ai_advisory_callback(
+                action=pure_action,
+                advisory_type=advisory_type,
+                advisory_id=advisory_id,
+                username=username,
+            )
+
+            feedback_text = result.get("feedback_text", "已收到")
+            await self._answer_callback(callback_query_id, action, text=feedback_text)
+
+            return {
+                "action": action, "advisory_type": advisory_type, "advisory_id": advisory_id,
+                "user": user, "success": result.get("success", False),
+                "info_action": pure_action in ("view", "produce_cmd"),
+            }
+        except Exception as _e:
+            logger.exception("ai_advisory_callback_error", action=action, error=str(_e))
+            try:
+                await self._answer_callback(callback_query_id, action, text="⚠️ 處理失敗")
+            except Exception:
+                pass
+            return {"action": action, "user": user, "success": False}
+
     async def _edit_drift_card_outcome(
         self, report_id: str, verb: str, by: str, ok: bool,
     ) -> None:
@@ -2986,6 +3047,22 @@ class TelegramGateway:
                     user=user,
                 )
 
+            # ===================================================================
+            # 2026-04-19 P0 修 (ADR-092): ai_advisory_* 按鈕路由
+            # 4 LLM scanner (capacity/compliance/coverage/rule_quality) 的互動按鈕
+            # callback_data 格式: 'ai_advisory_{handled|snooze|view|produce_cmd}:{type}:{id}'
+            # nonce 解析後 action = 'ai_advisory_handled' 等,approval_id 內嵌 type+id
+            # ===================================================================
+            if action.startswith("ai_advisory_"):
+                return await self._handle_ai_advisory_action(
+                    action=action,
+                    advisory_payload=approval_id,  # 格式: '{type}:{id}'
+                    callback_query_id=callback_query_id,
+                    user_id=user_id,
+                    username=username,
+                    user=user,
+                )
+
             # ===================================================================
             # Step 1.9: Phase 5 Sprint 5.3 — 分類按鈕寫類 action 路由
             # 2026-04-14 Claude Sonnet 4.6