wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增 Gemini 出站 smoke sentinel

Browse Source
This commit is contained in:
OoO
2026-05-21 15:17:24 +08:00
committed by AiderHeal Bot
parent 1c4fcae5ca
commit 1f0d37a1fe
6 changed files with 218 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config.py
View File

@@ -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 # 用於模板顯示

1
docs/memory/history_logs.md
View File

@@ -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。

12
scheduler.py
View File

@@ -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

120
services/ai_automation_smoke_service.py
View File

@@ -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 += ["", "🚩 <b>最近異常檢查</b>"]
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(),

81
tests/test_ai_automation_smoke_service.py
View File

@@ -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 "<script>" not in message

8
tests/test_run_scheduler_embed_consistency.py
View File

@@ -161,3 +161,11 @@ def test_roi_ai_smoke_and_daily_report_schedules_stay_staggered():
assert 'schedule.every().day.at("09:05").do(run_roi_monthly_report_if_new_month)' in source
assert 'schedule.every().day.at("09:10").do(run_ai_smoke_daily_summary_task)' in source
assert "schedule.every(6).hours.do(run_action_plan_hygiene_task)" in source
def test_ai_smoke_daily_summary_refreshes_smoke_before_push(monkeypatch):
run_scheduler = _load_run_scheduler(monkeypatch)
source = inspect.getsource(run_scheduler.run_ai_smoke_daily_summary_task)
assert "collect_ai_automation_smoke(record_history=True" in source
assert "send_smoke_daily_summary()" in source