補齊 PChome retry controlled apply preflight

routes/ai_routes.py

@@ -2361,6 +2361,63 @@ def api_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier
         }), 500
 
 
+@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-preflight-package')
+@login_required
+def api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight_package():
+    """P2 AI-controlled apply preflight for verified retry exception artifacts."""
+    try:
+        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_preflight_package,
+        )
+
+        force_refresh = str(request.args.get('refresh') or '').strip().lower() in {'1', 'true', 'yes'}
+        execute_search = str(request.args.get('execute_search') or '').strip().lower() in {'1', 'true', 'yes'}
+        execute_retry_search = str(request.args.get('execute_retry_search') or '').strip().lower() in {'1', 'true', 'yes'}
+        materialize_artifacts = str(request.args.get('materialize_artifacts') or '').strip().lower() in {'1', 'true', 'yes'}
+        limit = request.args.get('limit', 20, type=int)
+        batch_size = request.args.get('batch_size', 5, type=int)
+        limit_per_product = request.args.get('limit_per_product', 8, type=int)
+        max_terms_per_product = request.args.get('max_terms_per_product', 5, type=int)
+        min_score = request.args.get('min_score', 0.45, type=float)
+        limit = max(5, min(limit, 50))
+
+        payload = None
+        if not force_refresh:
+            payload = _get_cached_pchome_growth_payload()
+
+        if payload is None:
+            engine = _create_icaim_dashboard_engine(DATABASE_PATH)
+            try:
+                payload = build_pchome_growth_opportunities(engine, limit=limit)
+            finally:
+                engine.dispose()
+            payload["cache_state"] = "fresh"
+            _set_pchome_growth_cache(payload)
+
+        package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_preflight_package(
+            payload,
+            batch_size=batch_size,
+            execute_search=execute_search,
+            execute_retry_search=execute_retry_search,
+            limit_per_product=limit_per_product,
+            max_terms_per_product=max_terms_per_product,
+            min_score=min_score,
+            materialize_artifacts=materialize_artifacts,
+        )
+        package["source_endpoint"] = (
+            "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-closeout-verifier-artifact-preflight-verifier-package"
+        )
+        return jsonify(package)
+    except Exception as exc:
+        logger.error("[PChomeGrowth] direct mapping retry candidate exception controlled apply preflight 讀取失敗: %s", exc, exc_info=True)
+        return jsonify({
+            "success": False,
+            "error": "PChome 商品對應 retry 例外 controlled apply preflight 暫時無法讀取，請稍後再試。",
+        }), 500
+
+
 @ai_bp.route('/api/ai/pchome-growth/ai-automation-readiness')
 @login_required
 def api_pchome_growth_ai_automation_readiness():

services/pchome_mapping_backlog_service.py

@@ -65,6 +65,9 @@ DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_MATERIALIZAT
 DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CLOSEOUT_VERIFIER_ARTIFACT_PREFLIGHT_VERIFIER_POLICY = (
     "ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preflight_verifier"
 )
+DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_POLICY = (
+    "ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight"
+)
 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"
@@ -3621,6 +3624,259 @@ def build_pchome_direct_mapping_retry_candidate_exception_closeout_verifier_arti
     }
 
 
+def _retry_exception_controlled_apply_preflight_id(verifier_package: dict[str, Any]) -> str:
+    verifier = verifier_package.get("artifact_preflight_verifier") or {}
+    summary = verifier_package.get("summary") or {}
+    payload = {
+        "run_id": verifier.get("run_id") or "",
+        "source_preview_id": verifier.get("source_preview_id") or "",
+        "artifact_hash_match_count": summary.get("artifact_hash_match_count") or 0,
+        "artifact_readback_pass_count": summary.get("artifact_readback_pass_count") or 0,
+    }
+    digest = hashlib.sha256(
+        json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
+    ).hexdigest()
+    return f"pchome-retry-exception-controlled-apply-preflight-{digest[:16]}"
+
+
+def _load_retry_exception_artifact_payload(root: Path, readback: dict[str, Any]) -> dict[str, Any]:
+    path = _resolve_retry_exception_artifact_path(root, str(readback.get("relative_path") or ""))
+    if not path.exists():
+        return {}
+    try:
+        loaded = json.loads(path.read_text(encoding="utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return {}
+    return loaded if isinstance(loaded, dict) else {}
+
+
+def _build_retry_exception_controlled_apply_selectors(verifier_input_payload: dict[str, Any]) -> list[dict[str, Any]]:
+    selectors: list[dict[str, Any]] = []
+    for receipt in verifier_input_payload.get("no_write_verifier_receipts") or []:
+        subject = receipt.get("subject") or {}
+        momo_product_id = subject.get("momo_product_id") or subject.get("product_id")
+        target_pchome_product_id = subject.get("target_pchome_product_id")
+        if not momo_product_id or not target_pchome_product_id:
+            continue
+        selectors.append({
+            "selector_id": receipt.get("receipt_id"),
+            "momo_product_id": momo_product_id,
+            "target_pchome_product_id": target_pchome_product_id,
+            "target_match_score": subject.get("target_match_score"),
+            "auto_compare_type": subject.get("auto_compare_type"),
+            "source_closeout_receipt_id": receipt.get("source_closeout_receipt_id"),
+            "source_artifact_id": receipt.get("source_artifact_id"),
+            "source_decision_id": receipt.get("source_decision_id"),
+            "ready_for_controlled_apply_preflight": True,
+        })
+    return selectors
+
+
+def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_preflight_package(
+    payload: dict[str, Any],
+    batch_size: int = 5,
+    *,
+    execute_search: bool = False,
+    execute_retry_search: bool = False,
+    limit_per_product: int = 8,
+    max_terms_per_product: int = 5,
+    min_score: float = 0.45,
+    search_func: Any = None,
+    materialize_artifacts: bool = False,
+    artifact_root: str | Path | None = None,
+) -> dict[str, Any]:
+    """Build a machine-verifiable controlled apply preflight from verified retry artifacts."""
+    verifier_package = (
+        build_pchome_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preflight_verifier_package(
+            payload,
+            batch_size=batch_size,
+            execute_search=execute_search,
+            execute_retry_search=execute_retry_search,
+            limit_per_product=limit_per_product,
+            max_terms_per_product=max_terms_per_product,
+            min_score=min_score,
+            search_func=search_func,
+            materialize_artifacts=materialize_artifacts,
+            artifact_root=artifact_root,
+        )
+    )
+    verifier = verifier_package.get("artifact_preflight_verifier") or {}
+    verifier_ready = bool(verifier.get("ready_for_controlled_apply_preflight"))
+    root = Path(verifier.get("artifact_root") or (Path.cwd() / "data"))
+    readbacks = list(verifier_package.get("artifact_readbacks") or [])
+    payload_by_key = {
+        str(readback.get("key") or ""): _load_retry_exception_artifact_payload(root, readback)
+        for readback in readbacks
+        if readback.get("passed")
+    }
+    verifier_input_payload = payload_by_key.get("retry_exception_closeout_verifier_input_artifact") or {}
+    selectors = _build_retry_exception_controlled_apply_selectors(verifier_input_payload)
+    preflight_id = _retry_exception_controlled_apply_preflight_id(verifier_package)
+    mutation_plan = [
+        {
+            "mutation_id": f"{preflight_id}-{index + 1:03d}",
+            "action": "upsert_retry_exception_direct_mapping_candidate",
+            "selector_id": selector.get("selector_id"),
+            "momo_product_id": selector.get("momo_product_id"),
+            "target_pchome_product_id": selector.get("target_pchome_product_id"),
+            "write_mode": "future_controlled_executor_only",
+            "executes_in_preflight": False,
+            "writes_database": False,
+        }
+        for index, selector in enumerate(selectors)
+    ]
+    guard_checks = [
+        {
+            "check": "artifact_preflight_verifier_ready",
+            "passed": verifier_ready,
+        },
+        {
+            "check": "target_selector_count_positive",
+            "passed": bool(selectors),
+        },
+        {
+            "check": "all_artifact_readbacks_passed",
+            "passed": int((verifier_package.get("summary") or {}).get("artifact_readback_fail_count") or 0) == 0,
+        },
+        {
+            "check": "artifact_hashes_all_match",
+            "passed": int((verifier_package.get("summary") or {}).get("artifact_hash_match_count") or 0) == len(readbacks),
+        },
+        {
+            "check": "identity_readback_artifact_ready",
+            "passed": (payload_by_key.get("retry_exception_identity_readback_artifact") or {}).get("identity_delta_status") == "ready",
+        },
+        {
+            "check": "controlled_apply_artifact_requires_rollback",
+            "passed": (payload_by_key.get("retry_exception_controlled_apply_preflight_artifact") or {}).get("rollback_plan_required") is True,
+        },
+        {
+            "check": "controlled_apply_artifact_requires_post_apply_readback",
+            "passed": (
+                payload_by_key.get("retry_exception_controlled_apply_preflight_artifact") or {}
+            ).get("production_readback_required") is True,
+        },
+        {
+            "check": "preflight_does_not_execute_database_write",
+            "passed": True,
+        },
+        {
+            "check": "executor_still_requires_fresh_production_truth",
+            "passed": True,
+        },
+    ]
+    preflight_ready = all(check["passed"] for check in guard_checks)
+    result = (
+        "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_READY"
+        if preflight_ready
+        else "WAITING_FOR_RETRY_EXCEPTION_ARTIFACT_PREFLIGHT_VERIFIER"
+    )
+    rollback_steps = [
+        {
+            "mutation_id": mutation.get("mutation_id"),
+            "action": "delete_or_restore_retry_exception_mapping_candidate",
+            "selector_id": mutation.get("selector_id"),
+            "executes_in_preflight": False,
+            "writes_database": False,
+        }
+        for mutation in mutation_plan
+    ]
+    readback_checks = [
+        "mapping_candidate_exists_for_selector",
+        "mapping_candidate_source_receipt_matches",
+        "target_pchome_product_id_matches_selector",
+        "momo_product_id_matches_selector",
+        "post_apply_artifact_hashes_still_match",
+    ]
+    return {
+        "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_POLICY,
+        "result": result,
+        "success": bool(verifier_package.get("success")),
+        "generated_at": verifier_package.get("generated_at"),
+        "source_policy": verifier_package.get("policy"),
+        "stats": verifier_package.get("stats") or {},
+        "backlog": verifier_package.get("backlog") or {},
+        "summary": {
+            "controlled_apply_preflight_ready_count": 1 if preflight_ready else 0,
+            "source_artifact_preflight_ready_count": int(
+                (verifier_package.get("summary") or {}).get("artifact_preflight_verifier_ready_count") or 0
+            ),
+            "target_selector_count": len(selectors),
+            "mutation_plan_count": len(mutation_plan),
+            "rollback_step_count": len(rollback_steps),
+            "post_apply_readback_check_count": len(readback_checks),
+            "guard_check_count": len(guard_checks),
+            "guard_pass_count": sum(1 for check in guard_checks if check.get("passed")),
+            "guard_fail_count": sum(1 for check in guard_checks if not check.get("passed")),
+            "writes_artifact_count": int((verifier_package.get("summary") or {}).get("writes_artifact_count") or 0),
+            "executes_database_write_count": 0,
+            "writes_database_count": 0,
+            "persists_candidate_count": 0,
+        },
+        "controlled_apply_preflight": {
+            "preflight_id": preflight_id,
+            "run_id": verifier.get("run_id"),
+            "source_preview_id": verifier.get("source_preview_id"),
+            "stage": "P2_retry_exception_controlled_apply_preflight",
+            "status": result,
+            "artifact_root": str(root),
+            "ready_for_controlled_apply_executor": preflight_ready,
+            "ready_for_database_apply_now": False,
+            "requires_fresh_production_truth_before_executor": True,
+            "executes_database_write_in_preflight": False,
+            "writes_database": False,
+        },
+        "target_selectors": selectors,
+        "mutation_plan": {
+            "mode": "dry_run_preflight_only",
+            "mutation_plan_count": len(mutation_plan),
+            "mutations": mutation_plan,
+            "executes_in_preflight": False,
+            "writes_database": False,
+        },
+        "rollback_plan": {
+            "rollback_step_count": len(rollback_steps),
+            "rollback_steps": rollback_steps,
+            "executes_in_preflight": False,
+            "writes_database": False,
+        },
+        "post_apply_readback_plan": {
+            "readback_checks": readback_checks,
+            "readback_check_count": len(readback_checks),
+            "executes_in_preflight": False,
+            "writes_database": False,
+        },
+        "executor_guard": {
+            "guard_checks": guard_checks,
+            "guard_check_count": len(guard_checks),
+            "all_passed": preflight_ready,
+            "requires_fresh_production_truth": True,
+            "allows_database_write_now": False,
+            "writes_database": False,
+        },
+        "source_artifact_preflight_summary": verifier_package.get("summary") or {},
+        "next_actions": [
+            "Feed this preflight into the retry exception controlled apply executor only after fresh production truth.",
+            "Executor must write one receipt per selector and then run post-apply readback checks.",
+            "Abort executor if any guard check, rollback plan, or artifact hash readback drifts.",
+        ],
+        "safety": {
+            "ai_controlled_apply": True,
+            "materialize_artifacts": bool(materialize_artifacts),
+            "reads_artifact_files": True,
+            "writes_artifact_count": int((verifier_package.get("summary") or {}).get("writes_artifact_count") or 0),
+            "executes_database_write_in_preflight": False,
+            "writes_database": False,
+            "persists_candidate": False,
+            "syncs_external_offers": False,
+            "dispatches_telegram": False,
+            "llm_calls_in_preflight": 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)

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_preflight_package,
     build_pchome_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_materialization_package,
     build_pchome_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preflight_verifier_package,
     build_pchome_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_preview_package,
@@ -1061,6 +1062,90 @@ def test_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_pre
     assert call_count["search"] == 4
 
 
+def test_direct_mapping_retry_candidate_exception_controlled_apply_preflight_builds_executor_guard(tmp_path):
+    call_count = {"search": 0}
+
+    def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score):
+        call_count["search"] += 1
+        if targets[0].get("source_artifact_id"):
+            return True, "retry_found", [
+                {
+                    "product_id": "MOMO-RETRY-REVIEW",
+                    "name": "Direct mapping product 40ml 多款任選",
+                    "price": 499,
+                    "target_pchome_product_id": "PCH-2",
+                    "target_pchome_name": "Direct mapping product 40ml x2",
+                    "target_match_score": 0.74,
+                    "auto_compare_type": "manual_review",
+                    "target_hard_veto": False,
+                },
+                {
+                    "product_id": "MOMO-RETRY-REVIEW-2",
+                    "name": "Direct mapping product 40ml 限量組",
+                    "price": 520,
+                    "target_pchome_product_id": "PCH-2",
+                    "target_pchome_name": "Direct mapping product 40ml x2",
+                    "target_match_score": 0.91,
+                    "auto_compare_type": "manual_review",
+                    "target_hard_veto": False,
+                },
+            ]
+        return True, "found", [
+            {
+                "product_id": "MOMO-UNIT",
+                "name": "Direct mapping product 40ml",
+                "price": 499,
+                "target_pchome_product_id": "PCH-2",
+                "target_pchome_name": "Direct mapping product 40ml x2",
+                "target_match_score": 0.91,
+                "auto_compare_type": "unit_price",
+                "target_hard_veto": True,
+            }
+        ]
+
+    package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_preflight_package(
+        _payload(),
+        batch_size=1,
+        execute_search=True,
+        execute_retry_search=True,
+        max_terms_per_product=6,
+        search_func=fake_search,
+        materialize_artifacts=True,
+        artifact_root=tmp_path,
+    )
+
+    preflight = package["controlled_apply_preflight"]
+    assert package["policy"] == (
+        "ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight"
+    )
+    assert package["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_PREFLIGHT_READY"
+    assert package["summary"]["controlled_apply_preflight_ready_count"] == 1
+    assert package["summary"]["target_selector_count"] == 2
+    assert package["summary"]["mutation_plan_count"] == 2
+    assert package["summary"]["rollback_step_count"] == 2
+    assert package["summary"]["post_apply_readback_check_count"] == 5
+    assert package["summary"]["guard_check_count"] == 9
+    assert package["summary"]["guard_fail_count"] == 0
+    assert package["summary"]["executes_database_write_count"] == 0
+    assert package["summary"]["writes_database_count"] == 0
+    assert preflight["preflight_id"].startswith("pchome-retry-exception-controlled-apply-preflight-")
+    assert preflight["ready_for_controlled_apply_executor"] is True
+    assert preflight["ready_for_database_apply_now"] is False
+    assert package["mutation_plan"]["executes_in_preflight"] is False
+    assert package["mutation_plan"]["writes_database"] is False
+    assert package["rollback_plan"]["writes_database"] is False
+    assert package["post_apply_readback_plan"]["readback_check_count"] == 5
+    assert package["executor_guard"]["all_passed"] is True
+    assert package["executor_guard"]["allows_database_write_now"] is False
+    assert {selector["momo_product_id"] for selector in package["target_selectors"]} == {
+        "MOMO-RETRY-REVIEW",
+        "MOMO-RETRY-REVIEW-2",
+    }
+    assert package["safety"]["executes_database_write_in_preflight"] is False
+    assert package["safety"]["writes_database"] is False
+    assert call_count["search"] == 2
+
+
 def test_ai_automation_readiness_makes_automation_visible_without_manual_primary_flow():
     readiness = build_pchome_growth_ai_automation_readiness(_payload(), batch_size=1)
 
@@ -15587,6 +15672,42 @@ def test_direct_mapping_retry_candidate_exception_closeout_verifier_artifact_pre
     assert payload["safety"]["writes_database"] is False
 
 
+def test_direct_mapping_retry_candidate_exception_controlled_apply_preflight_route_uses_cached_payload(monkeypatch):
+    from flask import Flask
+    from routes import ai_routes as routes
+
+    monkeypatch.setattr(routes, "_get_cached_pchome_growth_payload", lambda: _payload())
+
+    def fail_engine(database_path):
+        raise AssertionError("cached retry exception controlled apply preflight should not open a DB engine")
+
+    monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", fail_engine)
+
+    app = Flask(__name__)
+    with app.test_request_context(
+        "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-preflight-package?batch_size=1"
+    ):
+        response = routes.api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight_package.__wrapped__()
+
+    payload = response.get_json()
+    assert payload["success"] is True
+    assert payload["policy"] == (
+        "ai_controlled_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_preflight"
+    )
+    assert payload["source_endpoint"] == (
+        "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-closeout-verifier-artifact-preflight-verifier-package"
+    )
+    assert payload["result"] == "WAITING_FOR_RETRY_EXCEPTION_ARTIFACT_PREFLIGHT_VERIFIER"
+    assert payload["summary"]["controlled_apply_preflight_ready_count"] == 0
+    assert payload["summary"]["target_selector_count"] == 0
+    assert payload["summary"]["executes_database_write_count"] == 0
+    assert payload["controlled_apply_preflight"]["ready_for_controlled_apply_executor"] is False
+    assert payload["controlled_apply_preflight"]["ready_for_database_apply_now"] is False
+    assert payload["executor_guard"]["allows_database_write_now"] is False
+    assert payload["safety"]["executes_database_write_in_preflight"] is False
+    assert payload["safety"]["writes_database"] is False
+
+
 def test_ai_automation_readiness_route_defaults_to_no_search_and_uses_cached_payload(monkeypatch):
     from flask import Flask
     from routes import ai_routes as routes