feat(ea-hitl): ADR-021 EA 升級審核 pre-fetch + 競價告警金額影響量化

根治 2026-05-02 統帥反映的三層 EA escalation 訊息空泛問題：

1. _escalate_to_human 對 price_drop_alert / market_opportunity /
   threat_escalation 三類觸發，送 Telegram 前先 await Hermes 取具體
   SKU 清單覆蓋 plan 元流程文字（5s 短超時，失敗 fallback 原 plan）
2. NemoTron 競價告警新增 _compute_business_impact helper：
   過去 7 日營收流失（gap_pct>0 才算）+ 跟進競品建議價，
   dispatch 主路徑 / 防線二 / Hermes rule fallback 三條全部 Python
   獨裁注入，告警含「📉 NT$ X」「🎯 NT$ Y」具體金額
3. 補實 telegram_bot_service.handle_callback 的 momo:eig: prefix
   handler，HITL「🛑 忽略此事件」按鈕首次有對應 audit 寫入

Critic 審查通過（5 項必修全綠）：
- Critical-1: user_label HTML escape 防 Telegram username XSS
- High-1: pre-fetch 改 asyncio.wait_for(5s) 防阻塞 escalation
- High-2: 全部行缺金額時 return None 觸發 plan fallback
- Medium-2: 空 event_id callback 拒絕避免 audit 污染
- Medium-3: gap_pct≤0 時 revenue_loss_7d 強制歸 0 不誤導降價

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

docs/adr/ADR-021-ea-hitl-prefetch-and-alert-impact.md

@@ -0,0 +1,124 @@
+# ADR-021: EA HITL Pre-fetch + 競價告警必填金額影響量化
+
+- **Status**: Accepted
+- **Date**: 2026-05-03
+- **Deciders**: 統帥
+- **Related**: 補強 ADR-012（Agent Action Ladder L3 HITL）對 escalation 訊息內容的要求；不取代任何既有 ADR
+- **Affects**: `services/elephant_alpha_autonomous_engine.py`、`services/nemoton_dispatcher_service.py`、`services/hermes_analyst_service.py`、`services/telegram_bot_service.py`
+
+## Context
+
+2026-05-02 統帥收到一條 EA 升級審核 Telegram 告警（commit `b5a2b094` 時段），內容為：
+
+> 自主決策信心度 0.83 低於門檻，需人工批准
+> AI 摘要：競爭對手價格溢價 >5% 且有庫存充足…
+> 步驟 1：[OpenClaw] 基於市區拉貨區市場特性，生成包含動態定價模型與競爭對手分析的策略建議
+> 步驟 2：[Hermes] 識別具體競爭商品並量化價格差異（目標確認至少 20 個高價溢價 SKU）
+> 步驟 3：[NemoTron] 將生成的定價策略與競爭分析結果同步至人工審核管道
+
+統帥指出此告警**毫無決策價值**，原因：
+
+1. 沒有具體 SKU、沒有金額影響、沒有可批准/駁回的具體動作
+2. 三步驟「建議行動」實為「申請開始分析」的元流程描述（OpenClaw 生成策略 → Hermes 識別 SKU → NemoTron 同步），人類審核者沒有判斷依據
+3. HITL 攔截點放錯位置：應在「具體動作出爐後、執行前」攔，而非「分析計畫產出但尚未跑」
+
+### 程式碼層證據
+
+**根因 A — `_escalate_to_human` 的 ai_actions 來源是 plan 階段**：[`services/elephant_alpha_autonomous_engine.py:528-557`](services/elephant_alpha_autonomous_engine.py#L528-L557) 的 `_execute_autonomous_decision` 在 `elephant_orchestrator.analyze_and_coordinate` 產出 `StrategicDecision` 後，**信心度不足即 escalate**——此時 `decision.execution_plan` 還只是 Gemini 寫的 plan 文字，Hermes/NemoTron 從未實際跑過取得具體 SKU。`_escalate_to_human` 把 plan 的前 3 個 step description 直接灌進 `triaged_alert.ai_actions` 是空泛元流程的根源。
+
+**根因 B — Hermes/NemoTron 已有具體告警，但缺金額影響量化**：[`services/nemoton_dispatcher_service.py:683-700`](services/nemoton_dispatcher_service.py#L683-L700) 的 `_exec_trigger_price_alert` 已經能輸出「[SKU] 商品｜MOMO $X / PChome $Y｜價差 ±%｜銷量 ±%」，但**缺絕對金額影響**——人類看到「價差 22.4%、銷量 -35%」仍需自己換算「我這週實際少賺多少」才能決策。
+
+**根因 C — `momo:eig:` callback 按鈕從未實作**：[`services/telegram_templates.py:466-467`](services/telegram_templates.py#L466-L467) 的 `triaged_alert` 鍵盤產出「🛑 忽略此事件」按鈕（callback_data = `momo:eig:{event_id}`），但 [`services/telegram_bot_service.py:512`](services/telegram_bot_service.py#L512) 的 `handle_callback` 只 dispatch `menu:/cmd:/await:` 三個 prefix，**`momo:` prefix 完全沒處理**——統帥點擊忽略按鈕後永遠沒反應，HITL 流程閉環缺一環。
+
+## Decision
+
+對 EA L3 HITL 升級審核訊息與 NemoTron 競價告警採取三項根治措施：
+
+### 規則 1 — EA 升級審核 pre-fetch Hermes 具體威脅清單
+
+對 `_PRICE_RELATED_TRIGGERS = {price_drop_alert, market_opportunity, threat_escalation}` 三類觸發，`_escalate_to_human` **送 Telegram 前先呼叫 Hermes 取得具體 SKU 清單**，將前 5 筆格式化為：
+
+```
+[SKU] 商品名稱｜MOMO $X vs PChome $Y (±%)｜近 7 日流失 NT$ Z｜建議跟進 NT$ W
+```
+
+蓋掉原本的 plan 元流程文字。**強制配套限制**：
+
+- `asyncio.wait_for(timeout=5)` 短超時：Hermes 熱駐留 < 10s，但冷啟動會拖到 30s+，HITL 訊息延遲不可大於 10s
+- Pre-fetch 失敗（timeout / 0 threats / 全部缺金額）→ fallback 回原 plan 文字，**不中斷 escalation 主流程**
+- 「全部行皆缺金額」也視同無料 fallback，避免「乾巴巴兩行 MOMO/PChome 比價」比 plan 文字更空泛
+
+### 規則 2 — NemoTron 告警必填金額影響量化
+
+新增模組級 helper `_compute_business_impact(threat) -> {revenue_loss_7d, recommended_price}`：
+
+- **`revenue_loss_7d`** = `max(0, sales_7d_prev_amount - sales_7d_curr_amount)` 且**僅在 `gap_pct > 0` 時計算**
+  - 語意：「我方比競品貴，且過去 7 日銷量金額下滑 → 推估價格因素導致的流失」
+  - `gap_pct ≤ 0`（我方便宜或持平）即使銷量下滑亦歸 0，避免把季節性/商品壽命終結等非價格因素誤標為「流失」誘導降價
+- **`recommended_price`** = `round(pchome_price)` 當 `gap_pct > 0`；否則 `None`
+  - 語意：「跟進競品的最低調價金額」；統帥可基於此再依毛利策略加溢價
+
+`_fmt_price_alert` / `_fmt_human_review` 加入「📉 過去 7 日營收流失：NT$ X」「🎯 跟進競品建議價：NT$ Y（毛利策略可再加溢價）」區塊。dispatch() 主路徑、防線二強制覆核、`_hermes_rule_fallback` 三條路徑**全部走 Python 獨裁注入**（同 Bug-1 防線二原則：客觀數字不過 LLM）。
+
+為支援此計算，`PriceThreat` dataclass 新增 `sales_7d_curr_amount` / `sales_7d_prev_amount` 兩個欄位，由 `_batch_analyze` 從 SQL 結果直接帶出（**不**餵給 LLM 推理，避免 token 爆炸與 LLM 嘗試引用幻覺）。
+
+### 規則 3 — 補實 `momo:eig:` callback handler
+
+`telegram_bot_service.py.handle_callback` 在既有 `menu:/cmd:/await:` dispatch 之前加 `momo:eig:` 處理，呼叫 `_handle_event_ignore_callback`：
+
+1. 解析 `event_id`，**空 id 立即拒絕**（防 audit 污染）
+2. 寫入 `ai_insights`：`status='ignored', insight_type='human_review', metadata_json` 含 `event_id / decided_by / decided_at`（ADR-012 §③ audit trail）
+3. 編輯原訊息加「🛑 已忽略 by `<user>` @ `<ts>`」尾註
+4. **`user_label` / `ts_label` 寫入 HTML 前必須 `html.escape()`**（user-controlled Telegram username 透過 `<a href>/<pre>` 注入超連結與破版的 XSS 防線）
+5. 任何例外都 best-effort，不阻斷 UI
+
+### Critic 審查（必修）
+
+| 編號 | 嚴重度 | 內容 | 修復 |
+|------|--------|------|------|
+| Critical-1 | CRITICAL | `user_label` 直接 HTML 拼接 → username 注入 `<a>/<pre>` 破版 | `html.escape()` 雙重 escape |
+| High-1 | HIGH | Pre-fetch Hermes 同步阻塞 escalation cooldown 視窗（30-60s） | `asyncio.wait_for(timeout=5)` |
+| High-2 | HIGH | Hermes 有 threats 但全部缺金額時 → 兩行乾巴巴比價反而更空泛 | `any_concrete` 判斷，全缺則 `return None` 觸發 plan fallback |
+| Medium-2 | MEDIUM | 空 `event_id` callback 寫入 `'unknown'` 污染 audit | prefix 解析後即拒絕 |
+| Medium-3 | MEDIUM | `gap_pct ≤ 0` 但 `prev > curr` 仍顯示「流失」誤導降價 | `revenue_loss_7d` 條件改為 `if gap_pct > 0` |
+
+## Alternatives Considered
+
+| 方案 | 為何不選 |
+|------|---------|
+| A. 廢掉 `market_opportunity` 觸發類型，全交 NemoTron pipeline | EA 同時負責 `price_drop_alert / threat_escalation / resource_optimization / code_exception` 多類觸發，不可單刪一類；且 EA orchestrator 是 SOT，廢觸發類型需大改 |
+| B. 把 EA 信心度門檻降到 0.5，讓「market_opportunity」全部自主執行 | 違反 ADR-012 L3 HITL 對「實際修改價格」的安全網；統帥未授權自動調價 |
+| C. Pre-fetch 改為背景 task，escalation 訊息分兩次發 | 兩次告警分裂語境，使用者體驗差；且 Hermes 結果晚到時 escalation 已 cooldown |
+| D. 不 pre-fetch，要求人類自己用 `/menu` 查 SKU | 違反「告警必須 actionable」精神，且增加人類認知負荷 |
+
+## Consequences
+
+### 正面
+
+- EA 升級審核 Telegram 內容從元流程描述變為「具體 SKU + 價格 + 金額流失 + 建議調價」，HITL 真正可決策
+- NemoTron 既有告警再升級，每筆都帶可批准/駁回的金額判斷依據
+- `momo:eig:` 按鈕首次有對應 handler，HITL 流程閉環完整
+- pre-fetch 改用 5s 短超時 + fallback，最壞情況退回原 plan 文字，不破壞既有行為
+
+### 負面 / 風險
+
+- 每次價格類 escalation 多花 ≤ 5s（Hermes 熱駐留實測 < 10s 但有 timeout），整體告警延遲略增
+- Hermes 在 5s 內若沒回應，告警內容降級回 plan 文字（仍維持原行為，無新增風險）
+- `gap_pct ≤ 0` 案例的銷量下滑（非價格因素）將完全不顯示流失金額——若統帥需追蹤「非價格流失」需另開告警類型（待後續 ADR）
+
+### 監控指標
+
+- Telegram 「EA 升級審核」訊息含「📉 過去 7 日營收流失」比例（部署後應 → 每筆價格類觸發都有，除非 5s timeout）
+- `_handle_event_ignore_callback` audit 寫入頻率（觀察人類實際使用 HITL 比例）
+- Hermes pre-fetch timeout rate（>10% 代表 Ollama 冷啟動嚴重，需檢查 keep_alive）
+
+## 實作 checklist
+
+- [x] `services/hermes_analyst_service.py` PriceThreat 新增絕對金額欄位
+- [x] `services/nemoton_dispatcher_service.py` `_compute_business_impact` helper + 三條 dispatch 路徑注入
+- [x] `services/elephant_alpha_autonomous_engine.py` `_fetch_hermes_threats_summary` + 5s timeout + fallback
+- [x] `services/telegram_bot_service.py` `_handle_event_ignore_callback` + HTML escape + 空 id 拒絕
+- [x] Critic 審查通過（Critical-1 / High-1 / High-2 / Medium-2 / Medium-3 全修）
+- [x] Smoke test：`_compute_business_impact` 對 gap≤0 / gap=0 / 銷量回升 / bogus type 四案例驗證
+- [x] `docs/adr/README.md` 索引加 ADR-021
+- [ ] 部署到 188 + 觀察首日 EA escalation 內容（人工驗證）

docs/adr/README.md

@@ -42,6 +42,7 @@
 | [018](ADR-018-four-agent-ai-automation-control-plane.md) | 四 AI Agent 自動化控制面（Hermes/NemoTron/OpenClaw/ElephantAlpha） | Accepted | 2026-04-29 |
 | [019](ADR-019-telegram-bot-agentic-conversation-layer.md) | Telegram Bot Agentic Conversation Layer（菜單→Agent 決策統一入口） | Accepted | 2026-05-02 |
 | [020](ADR-020-code-review-full-autoheal.md) | Code Review 全自動修復政策（局部覆寫 ADR-012 HITL） | Accepted | 2026-05-02 |
+| [021](ADR-021-ea-hitl-prefetch-and-alert-impact.md) | EA HITL Pre-fetch + 競價告警必填金額影響量化 | Accepted | 2026-05-03 |
 
 ## 規範
 

services/elephant_alpha_autonomous_engine.py

@@ -115,6 +115,14 @@ _PRICE_ADJUSTMENT_REVIEW_ACTIONS = frozenset({
     "dispatch_price_updates",
 })
 
+# A' 軌：價格相關觸發類型，HITL 前需 pre-fetch Hermes 具體威脅清單
+# 取代 Gemini plan 階段的元流程文字（「步驟 1:[OpenClaw] 生成策略」這類）
+_PRICE_RELATED_TRIGGERS = frozenset({
+    "price_drop_alert",
+    "market_opportunity",
+    "threat_escalation",
+})
+
 
 def _zh_trigger(trigger_type: str) -> str:
     return _TRIGGER_ZH.get(trigger_type, trigger_type)
@@ -623,6 +631,78 @@ class ElephantAlphaAutonomousEngine:
         from services.hermes_analyst_service import HermesAnalystService
         return await self._run_with_timeout(HermesAnalystService().run, timeout=SSH_COMMAND_TIMEOUT)
 
+    async def _fetch_hermes_threats_summary(self, top_n: int = 5) -> Optional[List[str]]:
+        """A' 軌：HITL escalation 前 pre-fetch Hermes 具體威脅清單，
+        將「步驟 1: [OpenClaw] 生成策略」這類元流程文字換成
+        「[SKU] 商品｜MOMO $X / PChome $Y｜流失 NT$ Z｜建議 NT$ W」具體可決策行動。
+
+        失敗回 None，由呼叫端 fallback 至既有 execution_plan 文字。
+        本方法為 best-effort：任何例外都不阻斷 escalation 主流程。
+
+        Critic High-1 fix: 加 5 秒短超時防止阻塞 escalation cooldown 視窗
+          （Hermes 完整 run 可能 30-60s，HITL 訊息應快速送出）
+        Critic High-2 fix: 若每筆都缺 loss/rec_price，視同無料、return None 觸發 fallback
+        """
+        # 使用 5s 短超時：Hermes 熱駐留時實測 < 10s，但若需冷啟動會拖到 30s+
+        # HITL 訊息延遲不可大於 10s（影響統帥決策時效性），寧可 fallback 到原 plan 文字
+        try:
+            result = await asyncio.wait_for(self._hermes_analyze(), timeout=5)
+        except asyncio.TimeoutError:
+            self._log.warning("Pre-fetch Hermes 5s timeout; falling back to plan text")
+            return None
+        except Exception as e:
+            self._log.warning("Pre-fetch Hermes threats failed (non-blocking): %s", e)
+            return None
+
+        threats = getattr(result, "threats", None) or []
+        if not threats:
+            self._log.info("Pre-fetch Hermes returned 0 threats; falling back to plan text")
+            return None
+
+        # 模組頂部 import 較乾淨，但這裡保留 lazy import 避免兩服務循環依賴
+        # （nemoton 也需 hermes_analyst 的 PriceThreat dataclass）
+        try:
+            from services.nemoton_dispatcher_service import _compute_business_impact
+        except Exception as imp_err:
+            self._log.error("import _compute_business_impact failed: %s", imp_err)
+            _compute_business_impact = None
+
+        lines: List[str] = []
+        any_concrete = False  # Critic High-2: 至少一筆有金額才算具體
+        for t in threats[:top_n]:
+            sku = getattr(t, "sku", "?")
+            name = getattr(t, "name", "")[:24]
+            momo = float(getattr(t, "momo_price", 0) or 0)
+            pchome = float(getattr(t, "pchome_price", 0) or 0)
+            gap_pct = float(getattr(t, "gap_pct", 0) or 0)
+
+            impact = _compute_business_impact(t) if _compute_business_impact else {
+                "revenue_loss_7d": 0.0, "recommended_price": None,
+            }
+            loss = impact.get("revenue_loss_7d", 0.0) or 0.0
+            rec_price = impact.get("recommended_price")
+
+            parts = [
+                f"[{sku}] {name}",
+                f"MOMO ${momo:,.0f} vs PChome ${pchome:,.0f} ({gap_pct:+.1f}%)",
+            ]
+            if loss > 0:
+                parts.append(f"近 7 日流失 NT$ {loss:,.0f}")
+                any_concrete = True
+            if rec_price is not None and rec_price > 0:
+                parts.append(f"建議跟進 NT$ {rec_price:,.0f}")
+                any_concrete = True
+            lines.append("｜".join(parts))
+
+        if not any_concrete:
+            # Critic High-2: 全部都只有「MOMO $X vs PChome $Y」乾巴巴兩行，
+            # 比原本「步驟 1:OpenClaw 生成策略」更空泛。返回 None 觸發 plan fallback
+            self._log.info("Pre-fetch threats lacked impact figures on all rows; falling back")
+            return None
+
+        self._log.info("Pre-fetch Hermes threats produced %d concrete actions", len(lines))
+        return lines
+
     async def _dispatch_alerts(self, threats: List[Any]) -> Any:
         from services.nemoton_dispatcher_service import NemotronDispatcher
         return await self._run_with_timeout(
@@ -789,6 +869,26 @@ class ElephantAlphaAutonomousEngine:
         cooldown_min = self._get_cooldown_min(trigger.trigger_type)
         if not dedup_ts or (datetime.now().timestamp() - dedup_ts) / 60 >= cooldown_min:
             self._store_escalation(trigger.trigger_type)
+
+            # A' 軌：價格類觸發前 pre-fetch Hermes 具體威脅清單，
+            # 取代「步驟 1:[OpenClaw] 生成策略」這類元流程文字。
+            # — Claude Opus 4.7 (2026-05-02)
+            concrete_actions: Optional[List[str]] = None
+            if trigger.trigger_type in _PRICE_RELATED_TRIGGERS:
+                try:
+                    concrete_actions = await self._fetch_hermes_threats_summary(top_n=5)
+                except Exception as e:
+                    self._log.warning("Pre-fetch threats raised (non-blocking): %s", e)
+                    concrete_actions = None
+
+            if concrete_actions:
+                ai_actions_payload = concrete_actions
+            else:
+                ai_actions_payload = [
+                    f"步驟 {s.get('step', i+1)}：{_zh_step(s)}"
+                    for i, s in enumerate(decision.execution_plan[:3])
+                ] or ["無具體執行計畫"]
+
             try:
                 from services.telegram_templates import triaged_alert, _send_telegram_raw
                 msg, keyboard = triaged_alert(
@@ -807,13 +907,13 @@ class ElephantAlphaAutonomousEngine:
                         f"信心度：{decision.confidence:.2f} | "
                         f"參與模組：{', '.join(_AGENT_LABEL.get(a.lower(), a) for a in decision.agents_required)}"
                     ),
-                    ai_actions=[
-                        f"步驟 {s.get('step', i+1)}：{_zh_step(s)}"
-                        for i, s in enumerate(decision.execution_plan[:3])
-                    ] or ["無具體執行計畫"],
+                    ai_actions=ai_actions_payload,
                 )
                 await self._run_with_timeout(_send_telegram_raw, msg, timeout=10, reply_markup=keyboard)
-                self._log.info("Human escalation Telegram sent: %s", trigger.trigger_type)
+                self._log.info(
+                    "Human escalation Telegram sent: %s (concrete=%s)",
+                    trigger.trigger_type, bool(concrete_actions),
+                )
             except Exception as e:
                 self._log.error("Telegram escalation failed (non-blocking): %s", e)
 

services/hermes_analyst_service.py

@@ -45,6 +45,8 @@ class PriceThreat:
     risk: str                   # HIGH / MED / LOW
     recommended_action: str
     confidence: float
+    sales_7d_curr_amount: float = 0.0   # 過去 7 日營收金額（NT$），供下游金額影響量化
+    sales_7d_prev_amount: float = 0.0   # 前 7 日營收金額（NT$），供「可挽回營收」估算
 
 
 @dataclass
@@ -375,6 +377,9 @@ class HermesAnalystService:
                 "pchome":      pchome_price,
                 "gap_pct":     gap_pct,              # Python 預算好，Hermes 只做分類
                 "sales_delta": delta_pct,
+                # 絕對營收金額（不傳給 Hermes 推理，只在 Python 端保留供下游金額影響量化）
+                "_sales_curr": sales_curr,
+                "_sales_prev": sales_prev,
             }
             if raw_tags:
                 item["competitor_tags"] = raw_tags   # 語意情境給 Hermes 加分
@@ -386,10 +391,15 @@ class HermesAnalystService:
 
         mcp_ctx = build_mcp_context(topics=["market_trends", "holiday_calendar", "seasonal_insights"])
 
+        # 餵給 LLM 的版本：剝除底線開頭的 Python-only 欄位（避免 token 爆炸 + 防 LLM 嘗試引用）
+        items_for_llm = [
+            {k: v for k, v in item.items() if not k.startswith("_")}
+            for item in items
+        ]
         prompt = (
             f"【市場外部情報 (MCP)】\n{mcp_ctx}\n\n"
-            f"分析以下 {len(items)} 支商品的競價威脅，回傳前 {TOP_N} 個最高風險商品。\n\n"
-            f"資料：{json.dumps(items, ensure_ascii=False)}\n\n"
+            f"分析以下 {len(items_for_llm)} 支商品的競價威脅，回傳前 {TOP_N} 個最高風險商品。\n\n"
+            f"資料：{json.dumps(items_for_llm, ensure_ascii=False)}\n\n"
             f"輸出格式（JSON 陣列，每筆含）：\n"
             f'[{{"sku": string, "name": string, "category": string, '
             f'"momo_price": number, "pchome_price": number, '
@@ -488,6 +498,9 @@ class HermesAnalystService:
                     risk=t.get("risk", "LOW"),                        # LLM 分類
                     recommended_action=t.get("recommended_action", ""),  # LLM 洞察
                     confidence=float(t.get("confidence", 0.5)),       # LLM 信心度
+                    # 絕對營收金額：純 Python truth，供下游金額影響量化（B' 軌）
+                    sales_7d_curr_amount=float(ground.get("_sales_curr", 0) or 0),
+                    sales_7d_prev_amount=float(ground.get("_sales_prev", 0) or 0),
                 ))
 
             hermes_stats = getattr(self, "_last_stats", {})

services/nemoton_dispatcher_service.py

@@ -259,6 +259,51 @@ TOOLS = [
 ]
 
 
+# ── 金額影響量化（B' 軌：告警必須攜帶可決策的金額數字） ──
+def _compute_business_impact(threat) -> dict:
+    """從 PriceThreat 計算「過去 7 日營收流失」與「建議調價金額」。
+
+    回傳純 Python 客觀計算結果，由 dispatcher 強制注入告警 — LLM 不得碰觸這些數字。
+
+    revenue_loss_7d
+      = max(0, sales_7d_prev_amount - sales_7d_curr_amount) **僅在 gap_pct > 0 時**
+      語意：「我方比競品貴，且過去 7 日銷量金額下滑 → 推估價格因素導致的流失」
+      若 gap_pct ≤ 0（我方已便宜或持平）即使銷量下滑亦歸 0，避免把
+      季節性/商品壽命終結等非價格因素誤標為「流失」誘導降價（Critic Medium-3 fix）
+
+    recommended_price
+      = round(pchome_price)
+      語意：「跟進競品的最低調價金額」；統帥可基於此再依毛利策略加溢價
+      gap_pct ≤ 0（我方已便宜或持平）→ recommended_price=None（不需調價）
+    """
+    try:
+        gap_pct = float(getattr(threat, "gap_pct", 0) or 0)
+    except (TypeError, ValueError):
+        gap_pct = 0.0
+
+    revenue_loss_7d = 0.0
+    if gap_pct > 0:
+        try:
+            prev = float(getattr(threat, "sales_7d_prev_amount", 0) or 0)
+            curr = float(getattr(threat, "sales_7d_curr_amount", 0) or 0)
+            revenue_loss_7d = max(0.0, prev - curr)
+        except (TypeError, ValueError):
+            revenue_loss_7d = 0.0
+
+    recommended_price = None
+    try:
+        pchome = float(getattr(threat, "pchome_price", 0) or 0)
+        if pchome > 0 and gap_pct > 0:
+            recommended_price = round(pchome)
+    except (TypeError, ValueError):
+        pass
+
+    return {
+        "revenue_loss_7d":   revenue_loss_7d,
+        "recommended_price": recommended_price,
+    }
+
+
 # ── 語意化 Emoji 字典 ──────────────────────────────────
 # 身份識別
 HEADER_DISPATCHER = "⚡ NemoTron 派發器"
@@ -515,6 +560,10 @@ class NemotronDispatcher:
 
         for t in threats:
             try:
+                # B' 軌：每個 threat 預先算金額影響，所有路徑統一注入
+                impact = _compute_business_impact(t)
+                rl, rp = impact["revenue_loss_7d"], impact["recommended_price"]
+
                 if t.gap_pct < 5 and t.sales_7d_delta_pct < -30:
                     # Rule 1：價差微小但銷量大跌 → 非定價問題，人工確認
                     self._exec_flag_for_human_review(
@@ -528,6 +577,7 @@ class NemotronDispatcher:
                         footprint=footprint,
                         momo_price=t.momo_price, comp_price=t.pchome_price,
                         gap_pct=t.gap_pct, sales_delta=t.sales_7d_delta_pct,
+                        revenue_loss_7d=rl, recommended_price=rp,
                     )
                 elif t.gap_pct >= 5 and t.risk == "HIGH":
                     # Rule 2：高價差 HIGH 風險 → 競價告警
@@ -538,6 +588,7 @@ class NemotronDispatcher:
                         t.confidence,
                         momo_price=t.momo_price, comp_price=t.pchome_price,
                         footprint=footprint,
+                        revenue_loss_7d=rl, recommended_price=rp,
                     )
                 elif t.gap_pct < 0 and t.sales_7d_delta_pct > 0:
                     # Rule 3：我方具競爭力 + 銷量正成長 → 推薦
@@ -563,6 +614,7 @@ class NemotronDispatcher:
                         footprint=footprint,
                         momo_price=t.momo_price, comp_price=t.pchome_price,
                         gap_pct=t.gap_pct, sales_delta=t.sales_7d_delta_pct,
+                        revenue_loss_7d=rl, recommended_price=rp,
                     )
                 dispatched += 1
             except Exception as e:
@@ -585,16 +637,22 @@ class NemotronDispatcher:
         gap_pct: float, sales_delta: float,
         action: str, confidence: float,
         footprint: str,
+        revenue_loss_7d: float = 0.0,
+        recommended_price: Optional[float] = None,
     ) -> str:
         """
         類別一：緊急告警
-        倒金字塔：結論先行 → 核心數據 → AI 洞察 → 建議行動 → 運算足跡
+        倒金字塔：結論先行 → 核心數據 → 金額影響 → AI 洞察 → 運算足跡
 
         [2026-04-18 台北] Bug-3 防線三 UI 物理隔離：
           - 核心問題 = Python 客觀組字（價差 X% / 銷量 Y%），不碰 AI 文字
           - 關鍵數據 = Python 獨裁注入，None/0 降級為 N/A 而非 $0
           - AI 洞察 = action 唯一使用位置；移除假冒「Hermes 分析師研判」標籤
             （action 實為 NemoTron 輸出，非 Hermes） — Claude Opus 4.7
+
+        [2026-05-02 台北] B' 軌：金額影響量化 — Claude Opus 4.7
+          - revenue_loss_7d / recommended_price 純 Python 計算（_compute_business_impact）
+          - 解決「告警內容空泛、人類無可批准的具體動作」根因
         """
         conf_pct = int(confidence * 100)
 
@@ -605,6 +663,17 @@ class NemotronDispatcher:
         # 核心問題：Python 客觀組字，不碰 AI 文字
         core_issue = f"價差 {gap_pct:+.1f}% / 近七天銷量 {sales_delta:+.1f}%"
 
+        # 金額影響區塊（B' 軌新增）
+        impact_lines = []
+        if revenue_loss_7d and revenue_loss_7d > 0:
+            impact_lines.append(f"📉 過去 7 日營收流失：NT$ {revenue_loss_7d:,.0f}")
+        if recommended_price is not None and recommended_price > 0:
+            impact_lines.append(
+                f"🎯 跟進競品建議價：NT$ {recommended_price:,.0f}"
+                f"（毛利策略可再加溢價）"
+            )
+        impact_block = ("\n".join(impact_lines) + "\n\n") if impact_lines else ""
+
         # AI 洞察：唯一允許 LLM 文字進入的欄位
         ai_insight = _sanitize_text(action, fallback="請人工評估議價空間")
 
@@ -616,6 +685,7 @@ class NemotronDispatcher:
             f"• 我方價格：{mp_str}\n"
             f"• 競品價格:{cp_str}\n"
             f"• 銷量變化：{sales_delta:+.1f}%\n\n"
+            f"{impact_block}"
             f"{ICON_AI} AI 洞察（信心度 {conf_pct}%）：\n"
             f"{ai_insight}\n\n"
             f"{footprint}"
@@ -627,10 +697,13 @@ class NemotronDispatcher:
         concern: str, footprint: str,
         momo_price: float = None, comp_price: float = None,
         gap_pct: float = None, sales_delta: float = None,
+        revenue_loss_7d: float = 0.0,
+        recommended_price: Optional[float] = None,
     ) -> str:
         """
         類別二：人工覆核
         客觀數據由 Python 注入（防幻覺），AI 診斷隔離在獨立欄位
+        B' 軌：補金額影響欄位
         """
         # 客觀數據快照（100% Python，不經 LLM）
         if momo_price is not None and comp_price is not None:
@@ -643,10 +716,19 @@ class NemotronDispatcher:
         else:
             data_block = f"{ICON_REPORT} 客觀數據：（無競品比價數據）\n"
 
+        # 金額影響（B' 軌新增）
+        impact_lines = []
+        if revenue_loss_7d and revenue_loss_7d > 0:
+            impact_lines.append(f"📉 過去 7 日營收流失：NT$ {revenue_loss_7d:,.0f}")
+        if recommended_price is not None and recommended_price > 0:
+            impact_lines.append(f"🎯 跟進競品建議價：NT$ {recommended_price:,.0f}")
+        impact_block = ("\n".join(f"• {l}" for l in impact_lines) + "\n") if impact_lines else ""
+
         return (
             f"{ICON_WARNING} [{HEADER_DISPATCHER}] 異常波動需人工覆核\n\n"
             f"🔍 待查商品：[{sku}] {name}\n\n"
-            f"{data_block}\n"
+            f"{data_block}"
+            f"{impact_block}\n"
             f"🧠 AI 診斷：\n"
             f"{concern}\n\n"
             f"👉 建議行動：請營運人員立即進行前台走查。\n\n"
@@ -686,26 +768,38 @@ class NemotronDispatcher:
         gap_pct: float, sales_delta: float, action: str, confidence: float,
         momo_price=None, comp_price=None,
         footprint: str = "",
+        revenue_loss_7d: float = 0.0,
+        recommended_price: Optional[float] = None,
     ):
         """發送語意化競價高危險預警
 
         [2026-04-18 台北] Bug-1 防線一 保險：default 改 None，避免 LLM 漏吐
         → 舊版 default=0 → Telegram 顯示 $0。Layer A Hermes 已根治，這層是第二道屏障
         — Claude Opus 4.7
+
+        [2026-05-02 台北] B' 軌：revenue_loss_7d / recommended_price 純 Python 注入
+        — Claude Opus 4.7
         """
         msg = self._fmt_price_alert(
             sku, name, momo_price, comp_price,
             gap_pct, sales_delta, action, confidence, footprint,
+            revenue_loss_7d=revenue_loss_7d,
+            recommended_price=recommended_price,
         )
         self._send_telegram(msg)
-        logger.info(f"[Dispatcher] 競價告警 → {sku} gap={gap_pct:.1f}% sales={sales_delta:.1f}%")
+        logger.info(
+            f"[Dispatcher] 競價告警 → {sku} gap={gap_pct:.1f}% sales={sales_delta:.1f}% "
+            f"loss=${revenue_loss_7d:,.0f} rec_price={recommended_price}"
+        )
         # ADR-007 雙寫：沉澱到 ai_insights 供日後 RAG
         self._sink_insight_to_km(
             insight_type="price_alert",
             sku=sku, name=name,
             content=f"[高危險告警] {name} 價差 {gap_pct:+.1f}% / 銷量 {sales_delta:+.1f}%。行動：{action}",
             metadata={"gap_pct": gap_pct, "sales_delta": sales_delta, "confidence": confidence,
-                      "momo_price": momo_price, "comp_price": comp_price},
+                      "momo_price": momo_price, "comp_price": comp_price,
+                      "revenue_loss_7d": revenue_loss_7d,
+                      "recommended_price": recommended_price},
         )
 
     def _exec_add_to_recommendation(
@@ -776,6 +870,8 @@ class NemotronDispatcher:
         footprint: str = "",
         momo_price: float = None, comp_price: float = None,
         gap_pct: float = None, sales_delta: float = None,
+        revenue_loss_7d: float = 0.0,
+        recommended_price: Optional[float] = None,
     ):
         """發送語意化人工覆核請求"""
         concern = _sanitize_text(concern, fallback=f"數據走勢違背常理，疑似缺貨或前台異常。")
@@ -783,16 +879,22 @@ class NemotronDispatcher:
             sku, name, concern, footprint,
             momo_price=momo_price, comp_price=comp_price,
             gap_pct=gap_pct, sales_delta=sales_delta,
+            revenue_loss_7d=revenue_loss_7d,
+            recommended_price=recommended_price,
         )
         self._send_telegram(msg)
-        logger.info(f"[Dispatcher] 人工覆核請求 → {sku}")
+        logger.info(
+            f"[Dispatcher] 人工覆核請求 → {sku} loss=${revenue_loss_7d:,.0f}"
+        )
         # ADR-007 雙寫
         self._sink_insight_to_km(
             insight_type="human_review",
             sku=sku, name=name,
             content=f"[人工覆核] {name}。疑慮：{concern}",
             metadata={"confidence": confidence, "gap_pct": gap_pct, "sales_delta": sales_delta,
-                      "momo_price": momo_price, "comp_price": comp_price},
+                      "momo_price": momo_price, "comp_price": comp_price,
+                      "revenue_loss_7d": revenue_loss_7d,
+                      "recommended_price": recommended_price},
         )
 
     def _exec_route_to_km(
@@ -1059,6 +1161,7 @@ class NemotronDispatcher:
                         "疑似缺貨、下架、或前台異常，請營運人員立即走查。"
                     )
                 try:
+                    impact = _compute_business_impact(t)
                     self._exec_flag_for_human_review(
                         sku=t.sku,
                         name=t.name,
@@ -1069,6 +1172,8 @@ class NemotronDispatcher:
                         comp_price=t.pchome_price,
                         gap_pct=t.gap_pct,
                         sales_delta=t.sales_7d_delta_pct,
+                        revenue_loss_7d=impact["revenue_loss_7d"],
+                        recommended_price=impact["recommended_price"],
                     )
                     dispatched += 1
                 except Exception as e:
@@ -1174,17 +1279,24 @@ class NemotronDispatcher:
             # [2026-04-18 台北] Bug-1 防線一 保險：所有客觀數字強制由 Python 從 threat_map 注入，
             # 覆蓋 LLM 可能回吐的幻覺數字（例如 $0）。Layer A Hermes 根治是主防線，
             # 此處為二道屏障（萬一 ground_items 有漏網，或未來走 bypass） — Claude Opus 4.7
+            # [2026-05-02 台北] B' 軌：金額影響量化亦走 Python 獨裁注入 — Claude Opus 4.7
             t = threat_map.get(args.get("sku"))
             if tool_name == "trigger_price_alert" and t:
                 args["momo_price"]  = getattr(t, "momo_price",         None)
                 args["comp_price"]  = getattr(t, "pchome_price",       None)
                 args["gap_pct"]     = getattr(t, "gap_pct",            None)
                 args["sales_delta"] = getattr(t, "sales_7d_delta_pct", None)
+                impact = _compute_business_impact(t)
+                args["revenue_loss_7d"]   = impact["revenue_loss_7d"]
+                args["recommended_price"] = impact["recommended_price"]
             elif tool_name == "flag_for_human_review" and t:
                 args["momo_price"]  = getattr(t, "momo_price",         None)
                 args["comp_price"]  = getattr(t, "pchome_price",       None)
                 args["gap_pct"]     = getattr(t, "gap_pct",            None)
                 args["sales_delta"] = getattr(t, "sales_7d_delta_pct", None)
+                impact = _compute_business_impact(t)
+                args["revenue_loss_7d"]   = impact["revenue_loss_7d"]
+                args["recommended_price"] = impact["recommended_price"]
             elif tool_name == "add_to_recommendation":
                 args["footprint_data"] = footprint_data
                 args["threat"]         = t
@@ -1235,17 +1347,22 @@ if __name__ == "__main__":
         risk: str
         recommended_action: str
         confidence: float
+        sales_7d_curr_amount: float = 0.0
+        sales_7d_prev_amount: float = 0.0
 
     fake_threats = [
         FakeThreat("A003", "舒特膚AD乳液200ml", "美妝保養",
                    1200, 980, 22.4, -35.0, "HIGH",
-                   "建議立即降價至 $1,000 迎戰，或發放 $200 專屬折價券", 0.85),
+                   "建議立即降價至 $1,000 迎戰，或發放 $200 專屬折價券", 0.85,
+                   sales_7d_curr_amount=78000, sales_7d_prev_amount=120000),
         FakeThreat("A001", "玻尿酸面膜10片裝", "美妝保養",
                    320, 280, 14.3, -42.0, "HIGH",
-                   "建議跟進降價至 $285，配合限時加購活動", 0.78),
+                   "建議跟進降價至 $285，配合限時加購活動", 0.78,
+                   sales_7d_curr_amount=58000, sales_7d_prev_amount=100000),
         FakeThreat("A009", "美白化妝水150ml", "美妝保養",
                    420, 350, 20.0, -22.0, "HIGH",
-                   "價格差距過大，建議優先調降或捆包促銷", 0.45),
+                   "價格差距過大，建議優先調降或捆包促銷", 0.45,
+                   sales_7d_curr_amount=78000, sales_7d_prev_amount=100000),
     ]
 
     # 模擬 Hermes 運算足跡
@@ -1259,11 +1376,13 @@ if __name__ == "__main__":
         footprint    = _build_footprint_block(fake_hermes_stats, fake_nim_stats)
 
         # 測試三種訊息格式
-        print("=== 類別一：緊急告警 ===")
+        print("=== 類別一：緊急告警（含 B' 金額影響） ===")
         print(NemotronDispatcher._fmt_price_alert(
             "A003", "舒特膚AD乳液200ml", 1200, 980,
             22.4, -35.0, "建議立即降價至 $1,000 迎戰，或發放 $200 專屬折價券",
             0.85, footprint,
+            revenue_loss_7d=42000.0,    # B' 軌驗證：120k - 78k = 42k 流失
+            recommended_price=980,       # B' 軌驗證：跟進競品價
         ))
         print()
         print("=== 類別二：人工覆核 ===")

services/telegram_bot_service.py

@@ -509,6 +509,12 @@ class TrendTelegramBot:
 
         await query.answer()
 
+        # ADR-012 / A' 軌：HITL escalation 按鈕（momo:eig:{event_id} = event_ignore）
+        # triaged_alert 鍵盤按鈕「🛑 忽略此事件」會 callback momo:eig:<id>
+        if data.startswith('momo:eig:'):
+            await self._handle_event_ignore_callback(query, data)
+            return
+
         if data.startswith(('menu:', 'cmd:', 'await:')):
             await self._handle_openclaw_callback(query, context, data)
             return
@@ -579,6 +585,88 @@ class TrendTelegramBot:
         elif data.startswith("settings_"):
             await self._handle_settings_callback(query, data)
 
+    async def _handle_event_ignore_callback(self, query, data: str):
+        """A' 軌：HITL 升級審核「🛑 忽略此事件」按鈕處理。
+
+        callback_data 格式：momo:eig:<event_id>
+        動作：
+          1. 寫入 ai_insights 一筆 status='ignored' 紀錄（含 event_id 供 audit）
+          2. 編輯原訊息：標記「已忽略 by <user> @ <ts>」
+          3. 失敗 best-effort，不阻斷 user 體驗
+
+        ADR-012 §③：人工決策必有 audit trail；ai_insights 為 SOT。
+        Critic Critical-1 fix: user_label / ts_label 寫入 HTML 前須 escape；
+        Critic Medium-2 fix: 空 event_id 直接拒絕，避免污染 audit 資料。
+        """
+        from datetime import datetime as _dt
+        from html import escape as _html_escape
+
+        # Critic Medium-2: 空 event_id 即拒絕（避免攻擊者送 `momo:eig:` 污染 audit）
+        parts_in = data.split(':', 2)
+        event_id = parts_in[2].strip() if len(parts_in) >= 3 else ''
+        if not event_id:
+            try:
+                await query.answer("event_id 缺失，忽略動作未生效", show_alert=False)
+            except Exception:
+                pass
+            logger.warning("[EA HITL] empty event_id callback rejected: %r", data)
+            return
+
+        user = getattr(query, 'from_user', None)
+        user_label_raw = (
+            getattr(user, 'username', None)
+            or getattr(user, 'first_name', None)
+            or str(getattr(user, 'id', '?'))
+        )
+        ts_label_raw = _dt.now().strftime('%Y-%m-%d %H:%M')
+
+        try:
+            from database.manager import get_session
+            from sqlalchemy import text
+            session = get_session()
+            try:
+                session.execute(
+                    text("""
+                        INSERT INTO ai_insights
+                            (insight_type, content, confidence, created_by, status, metadata_json)
+                        VALUES (:type, :content, :conf, :by, :status, :meta)
+                    """),
+                    {
+                        "type": "human_review",
+                        "content": f"[EA HITL] 事件 {event_id} 由 {user_label_raw} 忽略",
+                        "conf": 1.0,
+                        "by": f"telegram:{user_label_raw}",
+                        "status": "ignored",
+                        "meta": json.dumps({
+                            "event_id": event_id,
+                            "decided_by": user_label_raw,
+                            "decided_at": ts_label_raw,
+                            "decision": "ignored",
+                        }, ensure_ascii=False),
+                    },
+                )
+                session.commit()
+            finally:
+                session.close()
+        except Exception as audit_err:
+            logger.warning(f"[EA HITL] ai_insights audit 寫入失敗（不阻斷 UI）: {audit_err}")
+
+        # Critic Critical-1: user_label / ts_label 須 HTML escape，避免攻擊者
+        # 透過 Telegram username 注入 <a>/<pre>/破版標籤
+        user_label_safe = _html_escape(str(user_label_raw))
+        ts_label_safe = _html_escape(ts_label_raw)
+        try:
+            original = query.message.text_html if query.message and getattr(query.message, 'text_html', None) else (query.message.text if query.message else "")
+            footer = f"\n\n🛑 <b>已忽略</b> by {user_label_safe} @ {ts_label_safe}"
+            await query.edit_message_text(
+                (original or '事件已忽略') + footer,
+                parse_mode='HTML',
+            )
+        except Exception as ui_err:
+            logger.debug(f"[EA HITL] edit_message 失敗（不阻斷）: {ui_err}")
+
+        logger.info(f"[EA HITL] event_ignore event_id={event_id} by={user_label_raw}")
+
     async def _handle_openclaw_callback(self, query, context, data: str):
         """轉接 OpenClaw 完整菜單 callback，避免長輪詢 Bot 吃掉 /menu。"""
         chat_id = query.message.chat_id