feat(smoke): monitor pchome auto policy guard

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(),