diff --git a/apps/api/src/jobs/capacity_forecaster_job.py b/apps/api/src/jobs/capacity_forecaster_job.py
index 1d0e252c..f2f5c813 100644
--- a/apps/api/src/jobs/capacity_forecaster_job.py
+++ b/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 = [
"📈 容量預測 (Phase 4 AI 升級版)",
- 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"🟡 {html.escape(host)}")
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:
diff --git a/apps/api/src/jobs/compliance_scanner_job.py b/apps/api/src/jobs/compliance_scanner_job.py
index 4848ba1d..b6f1bea2 100644
--- a/apps/api/src/jobs/compliance_scanner_job.py
+++ b/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))
diff --git a/apps/api/src/jobs/coverage_evaluator_job.py b/apps/api/src/jobs/coverage_evaluator_job.py
index 186dbdd9..e4ffe710 100644
--- a/apps/api/src/jobs/coverage_evaluator_job.py
+++ b/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))
diff --git a/apps/api/src/jobs/hermes_rule_quality_job.py b/apps/api/src/jobs/hermes_rule_quality_job.py
index 21752b71..d721ffd5 100644
--- a/apps/api/src/jobs/hermes_rule_quality_job.py
+++ b/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 = [
"🔍 Hermes 規則品質檢測 (AI 分析)",
@@ -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:
diff --git a/apps/api/src/services/ai_advisory_helpers.py b/apps/api/src/services/ai_advisory_helpers.py
new file mode 100644
index 00000000..d9c65589
--- /dev/null
+++ b/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
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 9f28f5de..ea8b0902 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/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