diff --git a/docs/adr/ADR-021-ea-hitl-prefetch-and-alert-impact.md b/docs/adr/ADR-021-ea-hitl-prefetch-and-alert-impact.md new file mode 100644 index 0000000..1d2f902 --- /dev/null +++ b/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 `` @ ``」尾註 +4. **`user_label` / `ts_label` 寫入 HTML 前必須 `html.escape()`**(user-controlled Telegram username 透過 `/
` 注入超連結與破版的 XSS 防線)
+5. 任何例外都 best-effort,不阻斷 UI
+
+### Critic 審查(必修)
+
+| 編號 | 嚴重度 | 內容 | 修復 |
+|------|--------|------|------|
+| Critical-1 | CRITICAL | `user_label` 直接 HTML 拼接 → username 注入 `/
` 破版 | `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 內容(人工驗證)
diff --git a/docs/adr/README.md b/docs/adr/README.md
index 50faece..3ad0530 100644
--- a/docs/adr/README.md
+++ b/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 |
 
 ## 規範
 
diff --git a/services/elephant_alpha_autonomous_engine.py b/services/elephant_alpha_autonomous_engine.py
index 9912b2f..8a8e211 100644
--- a/services/elephant_alpha_autonomous_engine.py
+++ b/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)
 
diff --git a/services/hermes_analyst_service.py b/services/hermes_analyst_service.py
index ff594b6..7edbabe 100644
--- a/services/hermes_analyst_service.py
+++ b/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", {})
diff --git a/services/nemoton_dispatcher_service.py b/services/nemoton_dispatcher_service.py
index 182cad8..e1d88e0 100644
--- a/services/nemoton_dispatcher_service.py
+++ b/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("=== 類別二:人工覆核 ===")
diff --git a/services/telegram_bot_service.py b/services/telegram_bot_service.py
index e557db4..6cb5927 100644
--- a/services/telegram_bot_service.py
+++ b/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:
+        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:
+        動作:
+          1. 寫入 ai_insights 一筆 status='ignored' 紀錄(含 event_id 供 audit)
+          2. 編輯原訊息:標記「已忽略 by  @ 」
+          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 注入 /
/破版標籤
+        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🛑 已忽略 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