diff --git a/config.py b/config.py
index f9183e7..85ff431 100644
--- a/config.py
+++ b/config.py
@@ -325,7 +325,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
-SYSTEM_VERSION = "V10.369"
+SYSTEM_VERSION = "V10.370"
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
public_url = PUBLIC_URL # 用於模板顯示
diff --git a/docs/memory/history_logs.md b/docs/memory/history_logs.md
index 3cf4318..832cbcd 100644
--- a/docs/memory/history_logs.md
+++ b/docs/memory/history_logs.md
@@ -13,6 +13,7 @@
## 📅 詳細更新日誌 (考古存檔)
### 2026-05-21：瀏覽器測試守門與 PChome 熱路徑優化
+- **V10.370 Gemini runtime sentinel**: AI automation smoke 新增 `Gemini 出站費用 sentinel`，每天檢查近 24h `ai_calls.provider='gemini'` 的 calls/tokens/cost/top callers；若 `GEMINI_API_HARD_DISABLED=true` 仍有 Gemini 記錄，smoke 直接升為 critical。scheduler 09:10 摘要推播前會先執行一次只讀 smoke，讓 Gemini 費用異常不再依賴人工打開 `/ai_automation_smoke` 才被發現。
- **V10.369 Gemini 防復發測試與極端價差同款放行**: 新增靜態測試禁止 production code 在 `services.gemini_guard` / `config.py` 之外直接讀 `GEMINI_API_KEY`，並要求所有 Gemini SDK/REST 出站點必須經 `get_gemini_api_key()`；比價 matcher 針對「同品牌 + 明確 identity anchor + 規格完全一致」但競品價格極端偏低的原生露/眉筆案例抑制價格懲罰，避免真同款因價格差被錯降級，同時補回既有 hard-veto 安全斷言。
- **V10.368 比價搜尋錨點強化**: marketplace matcher 補 LUDEYA 蜂王玫瑰外泌微臻霜、雅詩蘭黛微分子肌底原生露、Za / PERIPERA 眉筆眉彩等低信心邊界品牌的 identity anchor，並把「兩入組 / 任選色號 / 多色可選 / 櫻花輕盈版」歸為搜尋噪音，讓 MOMO → PChome 搜尋詞更聚焦於同款身份與規格，不被包裝組合或色號選項帶偏。
- **V10.367 Gemini hard egress kill switch**: 新增 `GEMINI_API_HARD_DISABLED=true` 預設硬封鎖，中央 `services.gemini_guard` 會在 hard switch 未解鎖時拒絕 `GEMINI_API_KEY`，即使 `GEMINI_FALLBACK_ENABLED=true` 也不會初始化 SDK 或 REST 出站。Code Review/OpenClaw/MCP/通用 AI fallback 保留 emergency path，但必須同時設 `GEMINI_API_HARD_DISABLED=false` 與 `GEMINI_FALLBACK_ENABLED=true`，必要時再用 `GEMINI_ALLOWED_CONTEXTS` 限定 caller。
diff --git a/scheduler.py b/scheduler.py
index 7daec97..c64e6b0 100644
--- a/scheduler.py
+++ b/scheduler.py
@@ -2857,16 +2857,22 @@ def run_ppt_auto_generation_task(schedule_kind=None):
def run_ai_smoke_daily_summary_task():
- """每日 AI 自動化 Smoke trend 摘要推播（只讀 history，不重新執行 smoke）。"""
+ """每日 AI 自動化 Smoke 摘要推播；先執行一次只讀 smoke 以刷新 sentinel。"""
try:
- from services.ai_automation_smoke_service import send_smoke_daily_summary
+ from services.ai_automation_smoke_service import collect_ai_automation_smoke, send_smoke_daily_summary
+ smoke_result = collect_ai_automation_smoke(record_history=True, history_limit=20)
result = send_smoke_daily_summary()
logging.info(
- "[Scheduler] [AISmokeSummary] 完成 | status=%s sent=%s failed=%s",
+ "[Scheduler] [AISmokeSummary] 完成 | smoke=%s send_status=%s sent=%s failed=%s",
+ smoke_result.get("status"),
result.get("status"),
result.get("telegram", {}).get("sent", 0),
result.get("telegram", {}).get("failed", 0),
)
+ result["smoke"] = {
+ "status": smoke_result.get("status"),
+ "summary": smoke_result.get("summary"),
+ }
_save_stats('ai_smoke_daily_summary', result)
except Exception as e:
import traceback as _tb
diff --git a/services/ai_automation_smoke_service.py b/services/ai_automation_smoke_service.py
index 44a51e9..c205adb 100644
--- a/services/ai_automation_smoke_service.py
+++ b/services/ai_automation_smoke_service.py
@@ -9,7 +9,7 @@ from __future__ import annotations
import os
import json
import threading
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
from html import escape
from typing import Any, Dict, List
@@ -187,6 +187,19 @@ def build_smoke_daily_summary_message(history_limit: int = 200) -> str:
else:
lines += ["", "🗓️ 尚無 smoke history；請先開啟 /ai_automation_smoke 執行快檢。"]
+ problem_checks = [
+ item for item in latest.get("checks", [])
+ if item.get("status") in {"warning", "critical"}
+ ]
+ if problem_checks:
+ lines += ["", "🚩 最近異常檢查"]
+ for item in problem_checks[:5]:
+ lines.append(
+ f"{escape(str(item.get('status', '')).upper())}｜"
+ f"{escape(str(item.get('name', 'unknown')))}："
+ f"{escape(str(item.get('summary', '')))}"
+ )
+
lines += [
"━━━━━━━━━━━━━━━━━━━━",
"入口：/ai_automation_smoke",
@@ -234,6 +247,110 @@ def _event_router_check() -> Dict[str, Any]:
return _check("EventRouter 通知鏈", "critical", f"EventRouter smoke 失敗：{exc}")
+def _row_mapping(row: Any) -> Dict[str, Any]:
+ if row is None:
+ return {}
+ if isinstance(row, dict):
+ return row
+ mapping = getattr(row, "_mapping", None)
+ if mapping is not None:
+ return dict(mapping)
+ return {}
+
+
+def _gemini_egress_check(window_hours: int = 24) -> Dict[str, Any]:
+ """Read-only runtime sentinel for unexpected Gemini spend."""
+ session = None
+ try:
+ from services.gemini_guard import gemini_disabled_message, is_gemini_hard_disabled
+
+ session = get_session()
+ since_at = datetime.now(timezone.utc) - timedelta(hours=int(window_hours))
+ summary_row = session.execute(
+ text("""
+ SELECT
+ COUNT(*) AS calls,
+ COALESCE(SUM(COALESCE(input_tokens, 0) + COALESCE(output_tokens, 0)), 0) AS tokens,
+ COALESCE(SUM(COALESCE(cost_usd, 0)), 0) AS cost_usd,
+ MAX(called_at) AS last_called
+ FROM ai_calls
+ WHERE lower(provider) = 'gemini'
+ AND called_at >= :since_at
+ """),
+ {"since_at": since_at},
+ ).fetchone()
+ summary = _row_mapping(summary_row)
+ calls = int(summary.get("calls") or 0)
+ tokens = int(summary.get("tokens") or 0)
+ cost_usd = float(summary.get("cost_usd") or 0.0)
+ hard_disabled = is_gemini_hard_disabled()
+
+ top_rows = []
+ if calls:
+ top_rows = [
+ _row_mapping(row) for row in session.execute(
+ text("""
+ SELECT
+ caller,
+ model,
+ COUNT(*) AS calls,
+ COALESCE(SUM(COALESCE(input_tokens, 0) + COALESCE(output_tokens, 0)), 0) AS tokens,
+ COALESCE(SUM(COALESCE(cost_usd, 0)), 0) AS cost_usd,
+ MAX(called_at) AS last_called
+ FROM ai_calls
+ WHERE lower(provider) = 'gemini'
+ AND called_at >= :since_at
+ GROUP BY caller, model
+ ORDER BY calls DESC, last_called DESC
+ LIMIT 5
+ """),
+ {"since_at": since_at},
+ ).fetchall()
+ ]
+
+ details = {
+ "window_hours": int(window_hours),
+ "since_at": since_at.isoformat(timespec="seconds"),
+ "hard_disabled": hard_disabled,
+ "calls": calls,
+ "tokens": tokens,
+ "cost_usd": round(cost_usd, 6),
+ "last_called": str(summary.get("last_called") or ""),
+ "top_callers": top_rows,
+ "guard_reason": gemini_disabled_message("ai_smoke_gemini_egress"),
+ }
+
+ if calls == 0:
+ return _check(
+ "Gemini 出站費用 sentinel",
+ "ok",
+ f"最近 {window_hours}h Gemini 出站 0 次，費用 $0",
+ details,
+ )
+ if hard_disabled:
+ return _check(
+ "Gemini 出站費用 sentinel",
+ "critical",
+ f"Hard-disabled 狀態仍偵測到 Gemini {calls} 次 / ${cost_usd:.6f}",
+ details,
+ )
+ return _check(
+ "Gemini 出站費用 sentinel",
+ "warning",
+ f"Gemini emergency fallback 近 {window_hours}h 使用 {calls} 次 / ${cost_usd:.6f}",
+ details,
+ )
+ except Exception as exc:
+ return _check(
+ "Gemini 出站費用 sentinel",
+ "warning",
+ f"Gemini 出站紀錄無法讀取：{exc}",
+ )
+ finally:
+ if session is not None:
+ session.close()
+
+
def _autoheal_check() -> Dict[str, Any]:
try:
import services.auto_heal_service as autoheal
@@ -360,6 +477,7 @@ def _elephant_hitl_check() -> Dict[str, Any]:
def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: int = 20) -> Dict[str, Any]:
checks: List[Dict[str, Any]] = [
_event_router_check(),
+ _gemini_egress_check(),
_autoheal_check(),
_nemotron_check(),
_embedding_queue_check(),
diff --git a/tests/test_ai_automation_smoke_service.py b/tests/test_ai_automation_smoke_service.py
index dd1d203..d79c219 100644
--- a/tests/test_ai_automation_smoke_service.py
+++ b/tests/test_ai_automation_smoke_service.py
@@ -27,6 +27,7 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
from services import ai_automation_smoke_service as smoke
monkeypatch.setattr(smoke, "_event_router_check", lambda: smoke._check("event", "ok", "ok"))
+ monkeypatch.setattr(smoke, "_gemini_egress_check", lambda: smoke._check("gemini", "ok", "ok"))
monkeypatch.setattr(smoke, "_autoheal_check", lambda: smoke._check("autoheal", "warning", "warn"))
monkeypatch.setattr(smoke, "_nemotron_check", lambda: smoke._check("nemotron", "ok", "ok"))
monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "critical", "boom"))
@@ -35,7 +36,7 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch):
result = smoke.collect_ai_automation_smoke(record_history=False)
assert result["status"] == "critical"
- assert result["summary"] == {"ok": 3, "warning": 1, "critical": 1, "total": 5}
+ assert result["summary"] == {"ok": 4, "warning": 1, "critical": 1, "total": 6}
def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypatch):
@@ -45,6 +46,7 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat
monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path))
monkeypatch.setattr(smoke, "_HISTORY_LIMIT", 2)
monkeypatch.setattr(smoke, "_event_router_check", lambda: smoke._check("event", "ok", "ok"))
+ monkeypatch.setattr(smoke, "_gemini_egress_check", lambda: smoke._check("gemini", "ok", "ok"))
monkeypatch.setattr(smoke, "_autoheal_check", lambda: smoke._check("autoheal", "ok", "ok"))
monkeypatch.setattr(smoke, "_nemotron_check", lambda: smoke._check("nemotron", "ok", "ok"))
monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "ok", "ok"))
@@ -95,6 +97,82 @@ def test_smoke_history_daily_summary():
]
+def test_gemini_egress_check_ok_when_no_calls(monkeypatch):
+ from services import ai_automation_smoke_service as smoke
+
+ class FakeResult:
+ def fetchone(self):
+ return {"calls": 0, "tokens": 0, "cost_usd": 0, "last_called": None}
+
+ def fetchall(self):
+ return []
+
+ class FakeSession:
+ closed = False
+
+ def execute(self, *_args, **_kwargs):
+ return FakeResult()
+
+ def close(self):
+ self.closed = True
+
+ fake_session = FakeSession()
+ monkeypatch.setattr(smoke, "get_session", lambda: fake_session)
+ monkeypatch.delenv("GEMINI_API_HARD_DISABLED", raising=False)
+
+ result = smoke._gemini_egress_check()
+
+ assert result["status"] == "ok"
+ assert result["details"]["calls"] == 0
+ assert result["details"]["hard_disabled"] is True
+ assert fake_session.closed is True
+
+
+def test_gemini_egress_check_critical_when_hard_disabled_has_calls(monkeypatch):
+ from services import ai_automation_smoke_service as smoke
+
+ class FakeSummaryResult:
+ def fetchone(self):
+ return {
+ "calls": 2,
+ "tokens": 1234,
+ "cost_usd": 0.012345,
+ "last_called": "2026-05-21 07:00:00+00",
+ }
+
+ class FakeTopResult:
+ def fetchall(self):
+ return [{
+ "caller": "code_review_openclaw_gemini",
+ "model": "gemini-2.5-flash",
+ "calls": 2,
+ "tokens": 1234,
+ "cost_usd": 0.012345,
+ "last_called": "2026-05-21 07:00:00+00",
+ }]
+
+ class FakeSession:
+ def __init__(self):
+ self.calls = 0
+
+ def execute(self, *_args, **_kwargs):
+ self.calls += 1
+ return FakeSummaryResult() if self.calls == 1 else FakeTopResult()
+
+ def close(self):
+ pass
+
+ monkeypatch.setattr(smoke, "get_session", lambda: FakeSession())
+ monkeypatch.delenv("GEMINI_API_HARD_DISABLED", raising=False)
+
+ result = smoke._gemini_egress_check()
+
+ assert result["status"] == "critical"
+ assert "Hard-disabled" in result["summary"]
+ assert result["details"]["calls"] == 2
+ assert result["details"]["top_callers"][0]["caller"] == "code_review_openclaw_gemini"
+
+
def test_build_smoke_daily_summary_message_escapes_history(tmp_path, monkeypatch):
from services import ai_automation_smoke_service as smoke
@@ -111,6 +189,7 @@ def test_build_smoke_daily_summary_message_escapes_history(tmp_path, monkeypatch
assert "AI 自動化 Smoke 每日摘要" in message
assert "WARNING" in message
+ assert "最近異常檢查" in message
assert "