From 16c615c9eec275ca81cab1623d9b3af2148fe01b Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 3 Jul 2026 08:00:01 +0800 Subject: [PATCH] feat(smoke): monitor pchome auto policy guard --- config.py | 2 +- docs/AI_INTELLIGENCE_MODULE_SOT.md | 7 +- .../pchome_ai_automation_priority_backlog.md | 25 +- services/ai_automation_smoke_service.py | 218 +++++++++++++++++- tests/test_ai_automation_metrics.py | 11 +- tests/test_ai_automation_smoke_service.py | 127 +++++++++- 6 files changed, 367 insertions(+), 23 deletions(-) diff --git a/config.py b/config.py index 5998e71..6c11ecd 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.725" +SYSTEM_VERSION = "V10.726" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index ac44d71..23d2200 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -1,8 +1,8 @@ # PChome 業績成長自動化作戰系統 — AI 競價情報模組 Single Source of Truth -> **最後更新**: 2026-07-02 (台北時間) -> **狀態**: 🟢 四 AI Agent 自動化閉環已落地；LLM 路由紅線升級為 Ollama-first 三主機級聯；PChome 後台業績匯入韌性已補強；產品定位正名為「PChome 業績成長自動化作戰系統」；外部市場來源正規化層、自動同步、作戰清單與價格參考表優先讀取、CSV 備援預檢、前台操作入口、高可見頁面繁中化守門、比價/作戰 UI 工作台化、AI 密集工作台文字密度守門、跨平台來源治理與商品身份 UI 契約已建立，GCP embedding 熔斷延後處理、110 proxy rescue 與 direct host health skip 已建立 -> **適用版本**: V10.725 +> **最後更新**: 2026-07-03 (台北時間) +> **狀態**: 🟢 四 AI Agent 自動化閉環已落地；LLM 路由紅線升級為 Ollama-first 三主機級聯；PChome 後台業績匯入韌性已補強；產品定位正名為「PChome 業績成長自動化作戰系統」；外部市場來源正規化層、自動同步、作戰清單與價格參考表優先讀取、CSV 備援預檢、前台操作入口、高可見頁面繁中化守門、比價/作戰 UI 工作台化、AI 密集工作台文字密度守門、跨平台來源治理、商品身份 UI 契約、sitewide visual QA runtime readback 與 PChome auto-policy authorization guard monitoring 已建立，GCP embedding 熔斷延後處理、110 proxy rescue 與 direct host health skip 已建立 +> **適用版本**: V10.726 --- @@ -241,6 +241,7 @@ SQL漏斗(~300筆) - `/metrics` 在尚無事件時仍輸出 `momo_ai_*` zero-baseline series，讓 Prometheus/Grafana 重啟後可立即看到 metric names。 - `/ai_automation_smoke` 提供登入後 AI 自動化健康檢查頁。 - `/api/ai-automation/smoke` 提供 read-only JSON 狀態，不做外部網路呼叫。 +- `/api/ai-automation/smoke` 必須包含 `PChome auto-policy authorization guard`：read-only 組裝 PChome auto-policy DB apply guard lane，檢查 no-secret、no-SQL、no-DB-write、machine-verifiable contract、`manual_review_mode=exception_only` 與 `primary_human_gate_count=0`。此 check 不做外部 fetch、不寫 DB、不發 authorization；`/api/ai-automation/scheduled-health-summary` 需輸出 `pchome_auto_policy_authorization_guard` family，`/metrics` 需暴露對應 `momo_ai_automation_scheduled_health_family_status`。 - 健康檢查 API 會將最近檢查結果保存到 JSONL，頁面顯示最近狀態趨勢。 - 健康檢查歷史支援 JSONL 匯出、清理與每日「正常 / 注意 / 嚴重」摘要。 - 健康檢查每日摘要支援手動 Telegram 推播，並由 `momo-scheduler` 每日 09:10 呼叫 `run_ai_smoke_daily_summary_task()`。 diff --git a/docs/guides/pchome_ai_automation_priority_backlog.md b/docs/guides/pchome_ai_automation_priority_backlog.md index 68853d5..f880495 100644 --- a/docs/guides/pchome_ai_automation_priority_backlog.md +++ b/docs/guides/pchome_ai_automation_priority_backlog.md @@ -9,7 +9,7 @@ 以下是使用者在主線推進期間插入、且必須保留在完整工作項目裡的要求。後續不得再把這些需求漏掉或變成口頭提醒。 1. 正式環境才是最新版本真相；不得用本機檔案、舊筆記、舊分支蓋過 production truth。 -2. 版本不得搞錯；目前正式環境 `/health` 版本維持 `V10.725`，除非有明確版本 bump、部署與 readback。 +2. 版本不得搞錯；目前正式環境 `/health` 版本維持 `V10.726`，除非有明確版本 bump、部署與 readback。 3. GitHub 全面 freeze；不得使用 GitHub、`gh`、GitHub API、GitHub Actions、PR、issue、mirror 或 read-only GitHub 流程。 4. 實作結果必須推到 Gitea `dev` 與 `main`；若改到 runtime 行為，還必須部署到正式環境並回讀。 5. 主方向是 AI 自動化，不是人工審核；人工欄位只能當 evidence / ledger / UI truth，不得阻擋低爆炸半徑、可驗證、可回滾的 controlled apply。 @@ -30,7 +30,7 @@ | P0 | Runtime truth + AI controlled apply closure | 進行中，核心閉環已完成多段 | 先證明 AI 真的在正式環境可執行、可回讀、可驗證、可回滾 | | P1 | 產品可視化 + 專業網站體驗 | 進行中 | 使用者要看得到實作結果，AI 自動化不能只藏在 artifact / log | | P2 | 外部主流 benchmark 落成 guardrails | 已完成第一版，持續擴充 | 把主流專業做法變成可測規則，不只是參考口號 | -| P3 | 擴大 AI 自動化覆蓋面 | 未開始 | 把目前 PChome retry lane 的閉環模式複製到更多安全場景 | +| P3 | 擴大 AI 自動化覆蓋面 | 進行中 | 把目前 PChome retry lane 的閉環模式複製到更多安全場景 | | P4 | Source / deployment governance | 持續進行 | 保證版本、Gitea、正式環境、DB 安全紅線不再混亂 | ## P0 - Runtime Truth And AI Controlled Apply Closure @@ -41,7 +41,7 @@ 已完成: -- 正式環境版本真相 guard：`/health` 是最高真相，目前維持 `V10.725`。 +- 正式環境版本真相 guard：`/health` 是最高真相，目前維持 `V10.726`。 - Gitea `dev` / `main` source truth 對齊。 - Retry exception controlled apply executor 已落地，目標表為 `pchome_product_matches`。 - 正式 DB 已 controlled apply 4 個 selectors: @@ -86,10 +86,16 @@ - `PRODUCT_SURFACE_CLEAR` - `finding_count=0` - `primary_human_gate_count=0` +- PChome auto-policy authorization guard runtime monitoring 已完成: + - `/api/ai-automation/smoke` 已加入 `PChome auto-policy authorization guard` + - read-only 組裝 auto-policy DB apply guard lane，檢查 no-secret / no-SQL / no-DB-write / machine-verifiable / exception-only + - `/api/ai-automation/scheduled-health-summary` 已輸出 `pchome_auto_policy_authorization_guard` family + - `/metrics` 已輸出 `momo_ai_automation_scheduled_health_family_status{family="pchome_auto_policy_authorization_guard",...}` + - 不做外部 fetch、不寫 DB、不發 apply authorization，下一個機器動作是補 machine fetch evidence 與後續 controlled authorization lane 進行中 / 下一步，必須照順序: -1. 將 PChome retry lane 的 receipt / replay / drift / product dashboard pattern 複製到下一條 safe automation lane。 +1. 將 auto-policy authorization guard 往下一條 machine-fetch evidence / authorization decision preflight lane 推進。 完成標準: @@ -254,12 +260,12 @@ 目的: 把目前 PChome retry mapping lane 的 controlled apply 閉環模式擴展到更多安全場景。 -未開始 / 下一步，必須照順序: +進行中 / 下一步，必須照順序: -1. 對其他 safe mapping backlog lanes 套用 receipt / replay / drift verifier 模式。 -2. 增加 scheduled automation health summaries。 +1. 將 auto-policy authorization guard 補齊 machine fetch evidence，推進到 authorization decision preflight lane。 +2. 對其他 safe mapping backlog lanes 套用 receipt / replay / drift verifier 模式。 3. 為每一類 controlled apply 自動產生 rollback evidence package。 -4. 把 AI automation metrics 整合進既有 observability surfaces。 +4. 把 AI automation metrics 持續整合進既有 observability surfaces。 完成標準: @@ -327,7 +333,8 @@ | P3.1 | Extend receipt / replay / drift pattern to more lanes | 已完成 | direct mapping candidate decision lane closeout route + focused tests | P3.2 scheduled automation health summaries | | P3.2 | Scheduled automation health summaries | 已完成 | `/api/ai-automation/scheduled-health-summary` + smoke service focused tests | P3.3 rollback evidence packages | | P3.3 | Rollback evidence packages | 已完成 | controlled apply rollback evidence route + focused tests | P3.4 observability metrics integration | -| P3.4 | Observability metrics integration | 已完成 | `/metrics` exports scheduled health summary gauges + focused tests; AI surface HTML readback now feeds smoke and scheduled health | P4 source / deployment governance ongoing | +| P3.4 | Observability metrics integration | 已完成 | `/metrics` exports scheduled health summary gauges + focused tests; AI surface HTML readback now feeds smoke and scheduled health | P3.5 auto-policy guard monitoring | +| P3.5 | PChome auto-policy authorization guard monitoring | 已完成 | AI smoke + scheduled family + metrics family read no-write/no-secret/exception-only guard; focused tests | machine fetch evidence / authorization decision preflight lane | | P4.1 | Source / deployment / runtime truth package | 已完成 | `scripts/ops/report_source_deploy_runtime_truth.py` + focused tests | 每次 Gitea push / production deploy 後執行 P4 report | | P2.2 | Benchmark guardrails applied to core AI surfaces | 已完成 | `/ai_intelligence` + `/observability/overview` golden-signal strips + focused tests | 後續 safe automation lanes 套同樣 first-viewport summary | diff --git a/services/ai_automation_smoke_service.py b/services/ai_automation_smoke_service.py index 6cef2b8..92030aa 100644 --- a/services/ai_automation_smoke_service.py +++ b/services/ai_automation_smoke_service.py @@ -30,6 +30,15 @@ _HISTORY_LOCK = threading.Lock() _PCHOME_DRIFT_MONITOR_DB_STATEMENT_TIMEOUT_MS = int( os.getenv("MOMO_PCHOME_DRIFT_MONITOR_DB_STATEMENT_TIMEOUT_MS", "5000") ) +_PCHOME_AUTO_POLICY_GUARD_LIMIT = int( + os.getenv("MOMO_PCHOME_AUTO_POLICY_GUARD_LIMIT", "20") +) +_PCHOME_AUTO_POLICY_GUARD_BATCH_SIZE = int( + os.getenv("MOMO_PCHOME_AUTO_POLICY_GUARD_BATCH_SIZE", "12") +) +_PCHOME_AUTO_POLICY_GUARD_TIMEOUT_SECONDS = int( + os.getenv("MOMO_PCHOME_AUTO_POLICY_GUARD_TIMEOUT_SECONDS", "5") +) def _check(name: str, status: str, summary: str, details: Dict[str, Any] | None = None) -> Dict[str, Any]: @@ -208,6 +217,11 @@ def build_scheduled_automation_health_summary( source_kind = "current_smoke" if current else "latest_history" pchome_drift = _find_check(source_result, "PChome 受控落地 drift monitor") pchome_details = pchome_drift.get("details") or {} + auto_policy_guard = _find_check(source_result, "PChome auto-policy authorization guard") + auto_policy_details = auto_policy_guard.get("details") or {} + if not auto_policy_guard or not auto_policy_details: + auto_policy_guard = _pchome_auto_policy_authorization_guard_check() + auto_policy_details = auto_policy_guard.get("details") or {} surface_readback = _find_check(source_result, "AI surface HTML readback") surface_details = surface_readback.get("details") or {} sitewide_readback = _find_check(source_result, "Sitewide UI/UX Agent readback") @@ -264,6 +278,56 @@ def build_scheduled_automation_health_summary( "writes_database": False, }, }, + { + "key": "pchome_auto_policy_authorization_guard", + "label": "PChome auto-policy authorization guard", + "status": auto_policy_guard.get("status") or "warning", + "summary": ( + auto_policy_guard.get("summary") + or "PChome auto-policy authorization guard has no latest check." + ), + "next_machine_action": auto_policy_details.get("next_machine_action") + or ( + "continue_pchome_auto_policy_guard_monitoring" + if auto_policy_guard.get("status") == "ok" + else "refresh_pchome_auto_policy_authorization_guard" + ), + "details": { + "policy": auto_policy_details.get("policy"), + "result": auto_policy_details.get("result"), + "guard_ready_count": int(auto_policy_details.get("guard_ready_count") or 0), + "lane_guard_check_count": int(auto_policy_details.get("lane_guard_check_count") or 0), + "lane_guard_pass_count": int(auto_policy_details.get("lane_guard_pass_count") or 0), + "lane_guard_waiting_count": int(auto_policy_details.get("lane_guard_waiting_count") or 0), + "authorization_request_closeout_ready_count": int( + auto_policy_details.get("authorization_request_closeout_ready_count") or 0 + ), + "exact_request_payload_field_count": int( + auto_policy_details.get("exact_request_payload_field_count") or 0 + ), + "machine_request_manifest_step_count": int( + auto_policy_details.get("machine_request_manifest_step_count") or 0 + ), + "lane_entry_requirement_count": int( + auto_policy_details.get("lane_entry_requirement_count") or 0 + ), + "reads_secret_count": int(auto_policy_details.get("reads_secret_count") or 0), + "executes_sql_count": int(auto_policy_details.get("executes_sql_count") or 0), + "writes_database_count": int(auto_policy_details.get("writes_database_count") or 0), + "primary_human_gate_count": int( + auto_policy_details.get("primary_human_gate_count") or 0 + ), + "manual_review_mode": auto_policy_details.get("manual_review_mode"), + "machine_verifiable": bool(auto_policy_details.get("machine_verifiable")), + "ready_for_database_apply_now": bool( + auto_policy_details.get("ready_for_database_apply_now") + ), + "issues_database_apply_authorization": bool( + auto_policy_details.get("issues_database_apply_authorization") + ), + "writes_database": False, + }, + }, { "key": "ai_surface_html_readback", "label": "AI surface HTML readback", @@ -353,14 +417,22 @@ def build_scheduled_automation_health_summary( }, ] worst = max(families, key=lambda item: STATUS_RANK.get(item["status"], 2))["status"] + primary_human_gate_count = sum( + int((item.get("details") or {}).get("primary_human_gate_count") or 0) + for item in families + ) + writes_database_count = sum( + int((item.get("details") or {}).get("writes_database_count") or 0) + for item in families + ) summary = { "ok": sum(1 for item in families if item["status"] == "ok"), "warning": sum(1 for item in families if item["status"] == "warning"), "critical": sum(1 for item in families if item["status"] == "critical"), "total": len(families), "history_record_count": history_count, - "primary_human_gate_count": 0, - "writes_database_count": 0, + "primary_human_gate_count": primary_human_gate_count, + "writes_database_count": writes_database_count, } problem_families = [ item for item in families if item.get("status") in {"warning", "critical"} @@ -858,6 +930,147 @@ def _pchome_controlled_apply_drift_monitor_check() -> Dict[str, Any]: engine.dispose() +def _pchome_auto_policy_authorization_guard_check() -> Dict[str, Any]: + """Read-only monitor for the PChome auto-policy DB apply guard lane.""" + engine = None + try: + from config import DATABASE_PATH + from services import pchome_mapping_backlog_service as backlog + from services.ai_exception_contract import ( + LEGACY_REVIEW_REQUIRED_COUNT_KEY, + PRIMARY_HUMAN_GATE_COUNT_KEY, + ) + from services.pchome_revenue_growth_service import build_pchome_growth_opportunities + + engine = _create_pchome_drift_monitor_engine(DATABASE_PATH) + payload = build_pchome_growth_opportunities( + engine, + limit=max(5, min(_PCHOME_AUTO_POLICY_GUARD_LIMIT, 50)), + ) + payload["cache_state"] = "smoke_read_only" + guard = backlog.build_pchome_auto_policy_db_apply_authorization_lane_guard( + payload, + batch_size=max(1, min(_PCHOME_AUTO_POLICY_GUARD_BATCH_SIZE, 50)), + execute_fetch=False, + timeout_seconds=max(1, min(_PCHOME_AUTO_POLICY_GUARD_TIMEOUT_SECONDS, 30)), + ) + summary = guard.get("summary") or {} + lane = guard.get("future_authorization_lane_guard") or {} + contract = guard.get("lane_transfer_contract") or {} + safety = guard.get("safety") or {} + result = str(guard.get("result") or "UNKNOWN") + + guard_ready_count = int(summary.get("authorization_lane_guard_ready_count") or 0) + check_count = int(summary.get("lane_guard_check_count") or 0) + pass_count = int(summary.get("lane_guard_pass_count") or 0) + waiting_count = int(summary.get("lane_guard_waiting_count") or 0) + request_ready_count = int(summary.get("authorization_request_closeout_ready_count") or 0) + exact_field_count = int(summary.get("exact_request_payload_field_count") or 0) + manifest_step_count = int(summary.get("machine_request_manifest_step_count") or 0) + entry_requirement_count = int(summary.get("lane_entry_requirement_count") or 0) + reads_secret_count = int(summary.get("reads_secret_count") or 0) + executes_sql_count = int(summary.get("executes_sql_count") or 0) + writes_database_count = int(summary.get("writes_database_count") or 0) + primary_human_gate_count = int(summary.get(PRIMARY_HUMAN_GATE_COUNT_KEY) or 0) + manual_review_required_count = int(summary.get(LEGACY_REVIEW_REQUIRED_COUNT_KEY) or 0) + side_effect_count = reads_secret_count + executes_sql_count + writes_database_count + risk_detected = any( + [ + side_effect_count, + primary_human_gate_count, + manual_review_required_count, + safety.get("reads_secret_in_preview") is not False, + safety.get("executes_sql") is not False, + safety.get("writes_database") is not False, + contract.get("writes_database") is not False, + contract.get("issues_database_apply_authorization") is not False, + lane.get("ready_for_database_apply_now") is not False, + lane.get("issues_database_apply_authorization") is not False, + ] + ) + guard_contract_present = ( + check_count >= 12 + and pass_count >= max(check_count - 1, 1) + and waiting_count <= 1 + and exact_field_count == 10 + and manifest_step_count == 6 + and entry_requirement_count == 6 + and contract.get("machine_verifiable") is True + and safety.get("manual_review_mode") == "exception_only" + ) + + if risk_detected: + status = "critical" + summary_text = "PChome auto-policy guard 偵測到 secret、SQL、DB write 或人工主 gate 風險" + next_machine_action = "halt_pchome_auto_policy_guard_and_inspect_risk" + elif guard_contract_present: + status = "ok" + summary_text = ( + "PChome auto-policy guard 已自動讀回 no-write/no-secret package，" + f"{pass_count}/{check_count} guard checks 通過，人工主 gate=0" + ) + next_machine_action = ( + "collect_machine_fetch_evidence_for_authorization_guard" + if guard_ready_count == 0 or request_ready_count == 0 + else "continue_pchome_auto_policy_guard_monitoring" + ) + else: + status = "warning" + summary_text = "PChome auto-policy guard 尚未達自動監控契約" + next_machine_action = "refresh_pchome_auto_policy_authorization_guard" + + return _check( + "PChome auto-policy authorization guard", + status, + summary_text, + { + "policy": guard.get("policy"), + "result": result, + "source_policy": guard.get("source_policy"), + "guard_ready_count": guard_ready_count, + "authorization_request_closeout_ready_count": request_ready_count, + "lane_guard_check_count": check_count, + "lane_guard_pass_count": pass_count, + "lane_guard_waiting_count": waiting_count, + "exact_request_payload_field_count": exact_field_count, + "machine_request_manifest_step_count": manifest_step_count, + "lane_entry_requirement_count": entry_requirement_count, + "reads_secret_count": reads_secret_count, + "executes_sql_count": executes_sql_count, + "writes_database_count": writes_database_count, + "primary_human_gate_count": primary_human_gate_count, + "manual_review_required_count": manual_review_required_count, + "manual_review_mode": safety.get("manual_review_mode"), + "machine_verifiable": bool(contract.get("machine_verifiable")), + "ready_for_database_apply_now": bool(lane.get("ready_for_database_apply_now")), + "issues_database_apply_authorization": bool( + lane.get("issues_database_apply_authorization") + ), + "requires_fresh_production_truth": bool( + lane.get("requires_fresh_production_truth_in_same_run") + ), + "execute_fetch": False, + "writes_database": False, + "next_machine_action": next_machine_action, + }, + ) + except Exception as exc: + return _check( + "PChome auto-policy authorization guard", + "warning", + f"PChome auto-policy guard 例行監控暫時無法讀取：{exc}", + { + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + "next_machine_action": "refresh_pchome_auto_policy_authorization_guard", + }, + ) + finally: + if engine is not None: + engine.dispose() + + def _ai_surface_html_readback_check() -> Dict[str, Any]: """Read-only AI product surface guardrail readback.""" try: @@ -1085,6 +1298,7 @@ def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: i _embedding_queue_check(), _elephant_hitl_check(), _pchome_controlled_apply_drift_monitor_check(), + _pchome_auto_policy_authorization_guard_check(), _ai_surface_html_readback_check(), _sitewide_ui_ux_agent_check(), _sitewide_visual_qa_check(), diff --git a/tests/test_ai_automation_metrics.py b/tests/test_ai_automation_metrics.py index 0971f09..d4ac408 100644 --- a/tests/test_ai_automation_metrics.py +++ b/tests/test_ai_automation_metrics.py @@ -107,16 +107,17 @@ def test_system_metrics_exports_scheduled_health_summary(): Gauge, { "summary": { - "ok": 4, + "ok": 5, "warning": 1, "critical": 0, - "total": 5, + "total": 6, "primary_human_gate_count": 0, "writes_database_count": 0, }, "families": [ {"key": "ai_automation_smoke", "status": "ok"}, {"key": "pchome_controlled_apply_drift_monitor", "status": "warning"}, + {"key": "pchome_auto_policy_authorization_guard", "status": "ok"}, {"key": "sitewide_ui_ux_agent", "status": "ok"}, {"key": "sitewide_visual_qa", "status": "ok"}, ], @@ -124,7 +125,7 @@ def test_system_metrics_exports_scheduled_health_summary(): ) output = generate_latest(registry).decode("utf-8") - assert 'momo_ai_automation_scheduled_health_summary_total{status="ok"} 4.0' in output + assert 'momo_ai_automation_scheduled_health_summary_total{status="ok"} 5.0' in output assert 'momo_ai_automation_scheduled_health_summary_total{status="warning"} 1.0' in output assert ( 'momo_ai_automation_scheduled_health_family_status{family="ai_automation_smoke",status="ok"} 1.0' @@ -134,6 +135,10 @@ def test_system_metrics_exports_scheduled_health_summary(): 'momo_ai_automation_scheduled_health_family_status{family="pchome_controlled_apply_drift_monitor",status="warning"} 1.0' in output ) + assert ( + 'momo_ai_automation_scheduled_health_family_status{family="pchome_auto_policy_authorization_guard",status="ok"} 1.0' + in output + ) assert ( 'momo_ai_automation_scheduled_health_family_status{family="sitewide_ui_ux_agent",status="ok"} 1.0' in output diff --git a/tests/test_ai_automation_smoke_service.py b/tests/test_ai_automation_smoke_service.py index 82b45b7..6b28bc7 100644 --- a/tests/test_ai_automation_smoke_service.py +++ b/tests/test_ai_automation_smoke_service.py @@ -1,6 +1,35 @@ import json +def _auto_policy_guard_history_check(status="ok"): + return { + "name": "PChome auto-policy authorization guard", + "status": status, + "summary": "PChome auto-policy guard 已自動讀回 no-write/no-secret package", + "details": { + "policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard", + "result": "WAITING_FOR_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT", + "guard_ready_count": 0, + "lane_guard_check_count": 12, + "lane_guard_pass_count": 11, + "lane_guard_waiting_count": 1, + "authorization_request_closeout_ready_count": 0, + "exact_request_payload_field_count": 10, + "machine_request_manifest_step_count": 6, + "lane_entry_requirement_count": 6, + "reads_secret_count": 0, + "executes_sql_count": 0, + "writes_database_count": 0, + "primary_human_gate_count": 0, + "manual_review_mode": "exception_only", + "machine_verifiable": True, + "ready_for_database_apply_now": False, + "issues_database_apply_authorization": False, + "next_machine_action": "collect_machine_fetch_evidence_for_authorization_guard", + }, + } + + def test_event_router_smoke_reports_queued_deliveries(tmp_path, monkeypatch): from services import ai_automation_metrics as metrics from services import ai_automation_smoke_service as smoke @@ -36,6 +65,7 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch): monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "critical", "boom")) monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok")) monkeypatch.setattr(smoke, "_pchome_controlled_apply_drift_monitor_check", lambda: smoke._check("pchome", "ok", "ok")) + monkeypatch.setattr(smoke, "_pchome_auto_policy_authorization_guard_check", lambda: smoke._check("auto-policy", "ok", "ok")) monkeypatch.setattr(smoke, "_ai_surface_html_readback_check", lambda: smoke._check("surface", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_ui_ux_agent_check", lambda: smoke._check("sitewide", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check("visual", "ok", "ok")) @@ -43,7 +73,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": 8, "warning": 1, "critical": 1, "total": 10} + assert result["summary"] == {"ok": 9, "warning": 1, "critical": 1, "total": 11} def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch): @@ -97,6 +127,71 @@ def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monke assert fake_engine.disposed is True +def test_pchome_auto_policy_authorization_guard_reports_no_write_exception_only(monkeypatch): + from services import ai_automation_smoke_service as smoke + from services import pchome_mapping_backlog_service as backlog + from services import pchome_revenue_growth_service as growth + + class FakeEngine: + disposed = False + + def dispose(self): + self.disposed = True + + fake_engine = FakeEngine() + monkeypatch.setattr(smoke, "_create_pchome_drift_monitor_engine", lambda _path: fake_engine) + monkeypatch.setattr(growth, "build_pchome_growth_opportunities", lambda _engine, limit: {"success": True}) + monkeypatch.setattr( + backlog, + "build_pchome_auto_policy_db_apply_authorization_lane_guard", + lambda *_args, **_kwargs: { + "policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_lane_guard", + "result": "WAITING_FOR_DB_APPLY_AUTHORIZATION_REQUEST_CLOSEOUT", + "source_policy": "read_only_pchome_growth_auto_policy_db_apply_authorization_request_closeout", + "summary": { + "authorization_lane_guard_ready_count": 0, + "authorization_request_closeout_ready_count": 0, + "lane_guard_check_count": 12, + "lane_guard_pass_count": 11, + "lane_guard_waiting_count": 1, + "exact_request_payload_field_count": 10, + "machine_request_manifest_step_count": 6, + "lane_entry_requirement_count": 6, + "reads_secret_count": 0, + "executes_sql_count": 0, + "writes_database_count": 0, + "primary_human_gate_count": 0, + "manual_review_required_count": 0, + }, + "future_authorization_lane_guard": { + "ready_for_database_apply_now": False, + "issues_database_apply_authorization": False, + "requires_fresh_production_truth_in_same_run": True, + }, + "lane_transfer_contract": { + "machine_verifiable": True, + "writes_database": False, + "issues_database_apply_authorization": False, + }, + "safety": { + "reads_secret_in_preview": False, + "executes_sql": False, + "writes_database": False, + "manual_review_mode": "exception_only", + }, + }, + ) + + result = smoke._pchome_auto_policy_authorization_guard_check() + + assert result["status"] == "ok" + assert result["details"]["lane_guard_pass_count"] == 11 + assert result["details"]["primary_human_gate_count"] == 0 + assert result["details"]["writes_database_count"] == 0 + assert result["details"]["next_machine_action"] == "collect_machine_fetch_evidence_for_authorization_guard" + assert fake_engine.disposed is True + + def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypatch): from services import ai_automation_smoke_service as smoke @@ -110,6 +205,7 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "ok", "ok")) monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok")) monkeypatch.setattr(smoke, "_pchome_controlled_apply_drift_monitor_check", lambda: smoke._check("pchome", "ok", "ok")) + monkeypatch.setattr(smoke, "_pchome_auto_policy_authorization_guard_check", lambda: smoke._check("auto-policy", "ok", "ok")) monkeypatch.setattr(smoke, "_ai_surface_html_readback_check", lambda: smoke._check("surface", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_ui_ux_agent_check", lambda: smoke._check("sitewide", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check("visual", "ok", "ok")) @@ -168,7 +264,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( json.dumps({ "generated_at": datetime.now().isoformat(timespec="seconds"), "status": "ok", - "summary": {"ok": 10, "warning": 0, "critical": 0, "total": 10}, + "summary": {"ok": 11, "warning": 0, "critical": 0, "total": 11}, "checks": [ { "name": "PChome 受控落地 drift monitor", @@ -182,6 +278,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( "drift_count": 0, }, }, + _auto_policy_guard_history_check(), { "name": "AI surface HTML readback", "status": "ok", @@ -243,11 +340,19 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( ) assert summary["policy"] == "read_only_ai_automation_scheduled_health_summary" assert summary["status"] == "ok" - assert summary["summary"]["total"] == 7 + assert summary["summary"]["total"] == 8 assert summary["summary"]["primary_human_gate_count"] == 0 assert summary["summary"]["writes_database_count"] == 0 assert pchome_family["status"] == "ok" assert pchome_family["details"]["drift_count"] == 0 + auto_policy_family = next( + item for item in summary["families"] + if item["key"] == "pchome_auto_policy_authorization_guard" + ) + assert auto_policy_family["status"] == "ok" + assert auto_policy_family["details"]["lane_guard_pass_count"] == 11 + assert auto_policy_family["details"]["primary_human_gate_count"] == 0 + assert auto_policy_family["details"]["writes_database_count"] == 0 surface_family = next( item for item in summary["families"] if item["key"] == "ai_surface_html_readback" @@ -282,7 +387,7 @@ def test_scheduled_automation_health_summary_can_use_current_smoke_without_recor current_smoke = { "generated_at": "2026-07-02T12:00:00", "status": "critical", - "summary": {"ok": 9, "warning": 0, "critical": 1, "total": 10}, + "summary": {"ok": 10, "warning": 0, "critical": 1, "total": 11}, "checks": [ { "name": "PChome 受控落地 drift monitor", @@ -290,6 +395,7 @@ def test_scheduled_automation_health_summary_can_use_current_smoke_without_recor "summary": "PChome controlled apply 偵測到 1 筆 drift", "details": {"drift_count": 1, "selector_count": 4, "readback_pass_count": 3}, }, + _auto_policy_guard_history_check(), { "name": "AI surface HTML readback", "status": "ok", @@ -351,6 +457,11 @@ def test_scheduled_automation_health_summary_falls_back_to_visual_qa_artifact(tm encoding="utf-8", ) monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path)) + monkeypatch.setattr( + smoke, + "_pchome_auto_policy_authorization_guard_check", + lambda: _auto_policy_guard_history_check(), + ) monkeypatch.setattr( surface_service, "build_sitewide_visual_qa_readback", @@ -405,6 +516,11 @@ def test_scheduled_automation_health_summary_route_returns_compact_payload(tmp_p encoding="utf-8", ) monkeypatch.setattr(smoke, "_HISTORY_PATH", str(history_path)) + monkeypatch.setattr( + smoke, + "_pchome_auto_policy_authorization_guard_check", + lambda: _auto_policy_guard_history_check(), + ) app = Flask(__name__) with app.test_request_context( @@ -628,6 +744,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch): monkeypatch.setattr(smoke, "_embedding_queue_check", lambda: smoke._check("embedding", "ok", "ok")) monkeypatch.setattr(smoke, "_elephant_hitl_check", lambda: smoke._check("elephant", "ok", "ok")) monkeypatch.setattr(smoke, "_pchome_controlled_apply_drift_monitor_check", lambda: smoke._check("pchome", "ok", "ok")) + monkeypatch.setattr(smoke, "_pchome_auto_policy_authorization_guard_check", lambda: smoke._check("auto-policy", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check( "Sitewide visual QA readback", "ok", @@ -649,7 +766,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch): item for item in result["checks"] if item["name"] == "Sitewide visual QA readback" ) - assert result["summary"]["total"] == 10 + assert result["summary"]["total"] == 11 assert surface_check["status"] == "ok" assert surface_check["details"]["checked_surface_count"] == 10 assert sitewide_check["status"] == "ok"