補齊 PChome controlled apply drift verifier

routes/ai_routes.py

@@ -2515,6 +2515,38 @@ def api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_
         }), 500
 
 
+@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-drift-verifier-package')
+@login_required
+def api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package():
+    """P2 AI-controlled drift verifier for applied retry exception product matches."""
+    try:
+        from config import DATABASE_PATH
+        from services.pchome_mapping_backlog_service import (
+            build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package,
+        )
+
+        run_id = str(request.args.get('run_id') or '').strip() or None
+
+        engine = _create_icaim_dashboard_engine(DATABASE_PATH)
+        try:
+            package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package(
+                run_id=run_id,
+                engine=engine,
+            )
+        finally:
+            engine.dispose()
+        package["source_endpoint"] = (
+            "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-receipt-replay-package"
+        )
+        return jsonify(package)
+    except Exception as exc:
+        logger.error("[PChomeGrowth] direct mapping retry candidate exception controlled apply drift verifier 讀取失敗: %s", exc, exc_info=True)
+        return jsonify({
+            "success": False,
+            "error": "PChome 商品對應 retry 例外 controlled apply drift verifier 暫時無法讀取，請稍後再試。",
+        }), 500
+
+
 @ai_bp.route('/api/ai/pchome-growth/ai-automation-readiness')
 @login_required
 def api_pchome_growth_ai_automation_readiness():
@@ -2523,6 +2555,7 @@ def api_pchome_growth_ai_automation_readiness():
         from config import DATABASE_PATH
         from services.pchome_revenue_growth_service import build_pchome_growth_opportunities
         from services.pchome_mapping_backlog_service import (
+            build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package,
             build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package,
             build_pchome_growth_ai_automation_readiness,
         )
@@ -2531,6 +2564,7 @@ def api_pchome_growth_ai_automation_readiness():
         execute_search = str(request.args.get('execute_search') or '').strip().lower() in {'1', 'true', 'yes'}
         execute_fetch = str(request.args.get('execute_fetch') or '').strip().lower() in {'1', 'true', 'yes'}
         include_receipt_replay = str(request.args.get('include_receipt_replay', 'true') or '').strip().lower() in {'1', 'true', 'yes'}
+        include_drift_verifier = str(request.args.get('include_drift_verifier', 'true') or '').strip().lower() in {'1', 'true', 'yes'}
         limit = request.args.get('limit', 20, type=int)
         batch_size = request.args.get('batch_size', 8, type=int)
         limit = max(5, min(limit, 50))
@@ -2549,13 +2583,19 @@ def api_pchome_growth_ai_automation_readiness():
             _set_pchome_growth_cache(payload)
 
         receipt_replay = None
-        if include_receipt_replay:
+        drift_verifier = None
+        if include_receipt_replay or include_drift_verifier:
             replay_engine = _create_icaim_dashboard_engine(DATABASE_PATH)
             try:
                 receipt_replay = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package(
                     materialize_artifacts=False,
                     engine=replay_engine,
                 )
+                if include_drift_verifier:
+                    drift_verifier = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package(
+                        engine=replay_engine,
+                        source_receipt_replay=receipt_replay,
+                    )
             finally:
                 replay_engine.dispose()
 
@@ -2565,6 +2605,7 @@ def api_pchome_growth_ai_automation_readiness():
             execute_search=execute_search,
             execute_fetch=execute_fetch,
             controlled_apply_receipt_replay=receipt_replay,
+            controlled_apply_drift_verifier=drift_verifier,
         )
         readiness["source_endpoint"] = "/api/ai/pchome-growth/opportunities"
         return jsonify(readiness)

services/pchome_mapping_backlog_service.py

@@ -74,6 +74,9 @@ DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_EXECUTOR_POLICY = (
 DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY = (
     "ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay"
 )
+DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY = (
+    "ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier"
+)
 AI_AUTOMATION_READINESS_POLICY = "read_only_pchome_growth_ai_automation_readiness"
 EVIDENCE_ENRICHMENT_PREVIEW_POLICY = "read_only_pchome_growth_evidence_enrichment_preview"
 EVIDENCE_SOURCE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_source_preview"
@@ -4507,6 +4510,95 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_recei
     }
 
 
+def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package(
+    *,
+    artifact_root: str | Path | None = None,
+    run_id: str | None = None,
+    engine: Any = None,
+    source_receipt_replay: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Verify that applied PChome product matches still agree with the replay receipt."""
+    replay = source_receipt_replay or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package(
+        artifact_root=artifact_root,
+        run_id=run_id,
+        materialize_artifacts=False,
+        engine=engine,
+    )
+    replay_summary = replay.get("summary") or {}
+    readbacks = list(replay.get("post_apply_readbacks") or [])
+    drift_items = [item for item in readbacks if item.get("passed") is not True]
+    selector_count = int(replay_summary.get("target_selector_count") or 0)
+    pass_count = int(replay_summary.get("post_apply_readback_pass_count") or 0)
+    receipt_hash_match_count = int(replay_summary.get("executor_receipt_hash_match_count") or 0)
+    source_ready = (
+        replay.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED"
+        and selector_count > 0
+        and receipt_hash_match_count > 0
+    )
+    drift_verified = source_ready and not drift_items and pass_count == selector_count
+
+    if drift_items:
+        result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED"
+    elif drift_verified:
+        result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED"
+    elif replay.get("missing_artifacts"):
+        result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_ARTIFACTS"
+    else:
+        result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_BASELINE"
+
+    checks = [
+        {"check": "source_replay_loaded", "passed": bool(replay)},
+        {"check": "source_receipt_hash_matches", "passed": receipt_hash_match_count > 0},
+        {"check": "target_selectors_present", "passed": selector_count > 0},
+        {"check": "all_current_readbacks_match_receipt", "passed": not drift_items and pass_count == selector_count},
+        {"check": "drift_verifier_does_not_write_database", "passed": True},
+    ]
+    return {
+        "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY,
+        "result": result,
+        "success": result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED",
+        "summary": {
+            "target_selector_count": selector_count,
+            "post_apply_readback_count": int(replay_summary.get("post_apply_readback_count") or 0),
+            "post_apply_readback_pass_count": pass_count,
+            "drift_count": len(drift_items),
+            "drift_verified_count": 1 if drift_verified else 0,
+            "receipt_hash_match_count": receipt_hash_match_count,
+            "missing_artifact_count": int(replay_summary.get("missing_artifact_count") or 0),
+            "writes_database_count": 0,
+        },
+        "drift_verifier": {
+            "stage": "P2_retry_exception_controlled_apply_drift_verifier",
+            "status": result,
+            "source_receipt_replay_result": replay.get("result"),
+            "ready": drift_verified,
+            "requires_production_version_truth": True,
+        },
+        "drift_items": drift_items,
+        "post_apply_readbacks": readbacks,
+        "source_receipt_replay_summary": replay_summary,
+        "checks": checks,
+        "check_count": len(checks),
+        "all_checks_passed": all(check.get("passed") is True for check in checks),
+        "next_actions": [
+            "Keep this verifier on the readiness surface so DB drift is visible without manual table review.",
+            "If drift is detected, use the receipt replay readbacks as rollback or re-apply evidence.",
+        ],
+        "safety": {
+            "ai_controlled_apply": True,
+            "reads_artifact_files": True,
+            "reads_database": engine is not None or bool(source_receipt_replay),
+            "writes_database": False,
+            "writes_database_count": 0,
+            "writes_artifact_count": 0,
+            "syncs_external_offers": False,
+            "dispatches_telegram": False,
+            "gemini_allowed": False,
+            "requires_production_version_truth": True,
+        },
+    }
+
+
 def build_pchome_evidence_enrichment_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]:
     """Build a read-only evidence enrichment package for mapping targets."""
     operator_preview = build_pchome_mapping_operator_preview(payload, batch_size=batch_size)
@@ -5102,6 +5194,7 @@ def build_pchome_growth_ai_automation_readiness(
     execute_fetch: bool = False,
     search_func: Any = None,
     controlled_apply_receipt_replay: dict[str, Any] | None = None,
+    controlled_apply_drift_verifier: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build a single read-only product-facing AI automation readiness view."""
     mapping_summary = summarize_pchome_mapping_backlog(payload)
@@ -5161,6 +5254,10 @@ def build_pchome_growth_ai_automation_readiness(
         and receipt_replay_readback_pass_count == receipt_replay_selector_count
         and int(receipt_replay_summary.get("executor_receipt_ready_count") or 0) > 0
     )
+    drift_summary = (controlled_apply_drift_verifier or {}).get("summary") or {}
+    controlled_apply_drift_count = int(drift_summary.get("drift_count") or 0)
+    controlled_apply_drift_verified_count = int(drift_summary.get("drift_verified_count") or 0)
+    controlled_apply_drift_selector_count = int(drift_summary.get("target_selector_count") or 0)
     exception_count = _summary_exception_count(receipt_summary) + int(
         decision_summary.get("machine_review_decision_count") or 0
     )
@@ -5187,7 +5284,9 @@ def build_pchome_growth_ai_automation_readiness(
         "writes_database": False,
     }
 
-    if controlled_apply_closeout_verified:
+    if controlled_apply_drift_count:
+        result = "AI_AUTOMATION_CONTROLLED_APPLY_DRIFT_DETECTED"
+    elif controlled_apply_closeout_verified:
         result = "AI_AUTOMATION_CONTROLLED_APPLY_CLOSEOUT_VERIFIED"
     elif not direct_mapping_count and ready_receipt_count:
         result = "AI_AUTOMATION_READY_FOR_CONTROLLED_APPLY"
@@ -5272,6 +5371,18 @@ def build_pchome_growth_ai_automation_readiness(
             f"hash match {receipt_replay_hash_match_count}",
             "從 artifact + DB readback 自動證明 apply 已收斂",
         ),
+        _automation_lane(
+            "controlled_apply_drift_verifier",
+            "落地漂移偵測",
+            "blocked" if controlled_apply_drift_count else ("completed" if controlled_apply_drift_verified_count else "waiting"),
+            controlled_apply_drift_count,
+            (
+                f"verified {controlled_apply_drift_selector_count}/{controlled_apply_drift_selector_count}"
+                if controlled_apply_drift_verified_count
+                else f"drift {controlled_apply_drift_count}"
+            ),
+            "持續比對 receipt 與正式 DB，偵測後進 rollback / re-apply",
+        ),
     ]
 
     return {
@@ -5314,6 +5425,9 @@ def build_pchome_growth_ai_automation_readiness(
             "controlled_apply_receipt_materialized_count": receipt_replay_materialized_count,
             "controlled_apply_receipt_hash_match_count": receipt_replay_hash_match_count,
             "controlled_apply_closeout_verified_count": 1 if controlled_apply_closeout_verified else 0,
+            "controlled_apply_drift_count": controlled_apply_drift_count,
+            "controlled_apply_drift_verified_count": controlled_apply_drift_verified_count,
+            "controlled_apply_drift_selector_count": controlled_apply_drift_selector_count,
             "exception_count": exception_count,
             "ai_exception_count": exception_count,
             AI_EXCEPTION_REQUIRED_COUNT_KEY: exception_count,
@@ -5330,6 +5444,7 @@ def build_pchome_growth_ai_automation_readiness(
             "exception_resolution": "ai_machine_verifiable",
             "machine_verifiable_decision_required": True,
             "controlled_apply_closeout": "receipt_replay_machine_verified" if controlled_apply_closeout_verified else "waiting_for_verifier",
+            "controlled_apply_drift": "drift_detected" if controlled_apply_drift_count else ("drift_verified" if controlled_apply_drift_verified_count else "waiting_for_drift_verifier"),
         },
         "ai_exception_auto_resolution": ai_exception_auto_resolution,
         "manual_policy": {
@@ -5353,6 +5468,7 @@ def build_pchome_growth_ai_automation_readiness(
             "llm_calls_in_preview": False,
             "gemini_allowed": False,
             "reads_database_for_receipt_replay": bool(controlled_apply_receipt_replay),
+            "reads_database_for_drift_verifier": bool(controlled_apply_drift_verifier),
         },
     }
 

tests/test_pchome_mapping_backlog_report.py

@@ -76,6 +76,7 @@ from services.pchome_mapping_backlog_service import (
     build_pchome_direct_mapping_candidate_exception_resolution_closeout_package,
     build_pchome_direct_mapping_retry_candidate_decision_package,
     build_pchome_direct_mapping_retry_candidate_exception_auto_resolution_package,
+    build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package,
     build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_executor_package,
     build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_preflight_package,
     build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package,
@@ -1411,6 +1412,32 @@ def test_direct_mapping_retry_candidate_exception_controlled_apply_receipt_repla
     assert read_only_package["summary"]["executor_receipt_materialized_count"] == 1
     assert read_only_package["summary"]["executor_receipt_hash_match_count"] == 1
     assert read_only_package["post_executor_receipt_verifier"]["hash_match"] is True
+    verifier = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package(
+        artifact_root=tmp_path,
+        run_id=run_id,
+        engine=engine,
+        source_receipt_replay=read_only_package,
+    )
+    assert verifier["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED"
+    assert verifier["summary"]["target_selector_count"] == 2
+    assert verifier["summary"]["drift_count"] == 0
+    assert verifier["summary"]["drift_verified_count"] == 1
+    assert verifier["safety"]["writes_database"] is False
+    with engine.begin() as conn:
+        conn.execute(text("""
+            UPDATE pchome_product_matches
+            SET pchome_id = 'PCH-DRIFT'
+            WHERE momo_icode = 'MOMO-RETRY-REVIEW'
+        """))
+    drift_package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package(
+        artifact_root=tmp_path,
+        run_id=run_id,
+        engine=engine,
+    )
+    assert drift_package["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED"
+    assert drift_package["summary"]["drift_count"] == 1
+    assert drift_package["drift_items"][0]["momo_icode"] == "MOMO-RETRY-REVIEW"
+    assert drift_package["drift_items"][0]["actual_pchome_id"] == "PCH-DRIFT"
     assert call_count["search"] == 2
 
 
@@ -1460,6 +1487,16 @@ def test_ai_automation_readiness_surfaces_controlled_apply_receipt_replay_closeo
                 "writes_database": False,
             },
         },
+        controlled_apply_drift_verifier={
+            "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED",
+            "summary": {
+                "target_selector_count": 4,
+                "post_apply_readback_pass_count": 4,
+                "drift_count": 0,
+                "drift_verified_count": 1,
+                "receipt_hash_match_count": 1,
+            },
+        },
     )
 
     lanes = {lane["key"]: lane for lane in readiness["automation_lanes"]}
@@ -1468,12 +1505,53 @@ def test_ai_automation_readiness_surfaces_controlled_apply_receipt_replay_closeo
     assert readiness["summary"]["controlled_apply_replay_readback_pass_count"] == 4
     assert readiness["summary"]["controlled_apply_receipt_materialized_count"] == 1
     assert readiness["summary"]["controlled_apply_closeout_verified_count"] == 1
+    assert readiness["summary"]["controlled_apply_drift_count"] == 0
+    assert readiness["summary"]["controlled_apply_drift_verified_count"] == 1
     assert readiness["automation_policy"]["controlled_apply_closeout"] == "receipt_replay_machine_verified"
+    assert readiness["automation_policy"]["controlled_apply_drift"] == "drift_verified"
     assert lanes["controlled_apply"]["status"] == "completed"
     assert lanes["controlled_apply"]["value"] == 4
     assert lanes["controlled_apply_receipt_replay"]["status"] == "completed"
     assert lanes["controlled_apply_receipt_replay"]["value"] == 1
+    assert lanes["controlled_apply_drift_verifier"]["status"] == "completed"
+    assert lanes["controlled_apply_drift_verifier"]["value"] == 0
     assert readiness["safety"]["reads_database_for_receipt_replay"] is True
+    assert readiness["safety"]["reads_database_for_drift_verifier"] is True
+    assert readiness["safety"]["writes_database"] is False
+
+
+def test_ai_automation_readiness_surfaces_controlled_apply_drift_detected():
+    readiness = build_pchome_growth_ai_automation_readiness(
+        _payload(),
+        batch_size=1,
+        controlled_apply_receipt_replay={
+            "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_READBACK_MISMATCH",
+            "summary": {
+                "target_selector_count": 4,
+                "post_apply_readback_pass_count": 3,
+                "executor_receipt_ready_count": 0,
+                "executor_receipt_materialized_count": 1,
+                "executor_receipt_hash_match_count": 0,
+            },
+        },
+        controlled_apply_drift_verifier={
+            "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED",
+            "summary": {
+                "target_selector_count": 4,
+                "post_apply_readback_pass_count": 3,
+                "drift_count": 1,
+                "drift_verified_count": 0,
+                "receipt_hash_match_count": 0,
+            },
+        },
+    )
+
+    lanes = {lane["key"]: lane for lane in readiness["automation_lanes"]}
+    assert readiness["result"] == "AI_AUTOMATION_CONTROLLED_APPLY_DRIFT_DETECTED"
+    assert readiness["summary"]["controlled_apply_drift_count"] == 1
+    assert readiness["automation_policy"]["controlled_apply_drift"] == "drift_detected"
+    assert lanes["controlled_apply_drift_verifier"]["status"] == "blocked"
+    assert lanes["controlled_apply_drift_verifier"]["value"] == 1
     assert readiness["safety"]["writes_database"] is False
 
 
@@ -16057,7 +16135,9 @@ def test_ai_automation_readiness_route_defaults_to_no_search_and_uses_cached_pay
     monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", fail_engine)
 
     app = Flask(__name__)
-    with app.test_request_context("/api/ai/pchome-growth/ai-automation-readiness?batch_size=1&include_receipt_replay=false"):
+    with app.test_request_context(
+        "/api/ai/pchome-growth/ai-automation-readiness?batch_size=1&include_receipt_replay=false&include_drift_verifier=false"
+    ):
         response = routes.api_pchome_growth_ai_automation_readiness.__wrapped__()
 
     payload = response.get_json()