From 4f3610c51ae6cb29391566e499328cd93300491e Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 2 Jul 2026 14:29:01 +0800 Subject: [PATCH] feat(ai): close out PChome candidate decision lane --- docs/AI_INTELLIGENCE_MODULE_SOT.md | 1 + .../pchome_ai_automation_priority_backlog.md | 4 +- routes/ai_routes.py | 55 ++++ services/pchome_mapping_backlog_service.py | 280 ++++++++++++++++++ ...pchome_candidate_decision_lane_closeout.py | 167 +++++++++++ 5 files changed, 505 insertions(+), 2 deletions(-) create mode 100644 tests/test_pchome_candidate_decision_lane_closeout.py diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index 9e7f7be..7ca555e 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -91,6 +91,7 @@ - 2026-07-02 起 PChome dashboard 第一視窗必須呈現 AI automation product truth：`_load_pchome_growth_command_center` 需 read-only 聚合 receipt replay、drift verifier、drift recovery、compact readback 與 artifact retention policy，產品面直接顯示 selector readback、drift count、retention 保留數與 automation lane 狀態；不得退回只有 raw API / artifact / log 才看得到 AI 自動化結果的模式。 - 2026-07-02 起 PChome dashboard 第一視窗可見文案必須使用繁中營運語言；`retention`、`Compact`、`Artifact`、`DB writes`、`prune` 等工程詞只能留在 API / artifact / 測試證據層，不得出現在第一視窗。產品面用「異動」、「最新回讀」、「證據留存」、「資料寫入」、「清理建議」呈現同一批 AI 自動化 truth。 - 2026-07-02 起 PChome AI automation dashboard 必須符合外部 benchmark guardrails：參考 Grafana / Datadog / New Relic / Atlassian Statuspage 的狀態分層、下一步優先、證據按需與 golden signals 做法；第一視窗必須輸出「已自動落地、已驗證、異動狀態、下一步」，且 `tests/test_pchome_dashboard_benchmark_guardrails.py` 必須鎖住這些要求。 +- 2026-07-02 起 PChome safe mapping lane expansion 必須先從 direct mapping candidate decision lane 開始；`/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-lane-closeout-package` 會把 candidate decision package 收斂成 lane receipt、receipt replay、drift verifier 與 product readiness，輸出 `primary_human_gate_count=0`、`drift_count`、`next_machine_action` 與 hash evidence。此 endpoint 預設不執行搜尋、不開 DB、不寫 DB、不持久化候選，只在 `execute_search=1` 時走 controlled read-only candidate search。 - V10.644 起 `/ai_intelligence` 的商品明細列不得只用句子描述比價；每列必須顯示 PChome 價格、MOMO 參考價、差距、可信度四格價格證據，並保留下一步按鈕。單位價候選需顯示單位價與單位，候選待確認或缺資料則以「待補 / 候選待確認」呈現，不得捏造價格。 - V10.645 起 `/ai_intelligence` 的商品明細分流切換後，必須顯示「這類商品怎麼處理」的行動摘要，包含件數、近 7 天業績、平均可信度、最大價差、代表商品與主按鈕；使用者不得只能看到商品列表而不知道下一步。 - V10.646 起 `/ai_intelligence` 的商品明細必須提供搜尋與排序；搜尋至少涵蓋商品、分類、商品編號與 MOMO 候選資訊，排序至少支援優先級、近 7 天業績、價差、下滑幅度與可信度。搜尋/排序後的行動摘要與明細列表必須使用同一批結果。 diff --git a/docs/guides/pchome_ai_automation_priority_backlog.md b/docs/guides/pchome_ai_automation_priority_backlog.md index 11d11d5..d3b677b 100644 --- a/docs/guides/pchome_ai_automation_priority_backlog.md +++ b/docs/guides/pchome_ai_automation_priority_backlog.md @@ -168,7 +168,7 @@ ## P3 - Broader AI Automation Expansion -狀態: 未開始。 +狀態: 進行中。 目的: 把目前 PChome retry mapping lane 的 controlled apply 閉環模式擴展到更多安全場景。 @@ -220,7 +220,7 @@ | P1.1 | Dashboard AI automation first-viewport surface | 已完成 | dashboard command center reads compact + retention packages | P1.2 wording guard | | P1.2 | UI wording guard for no raw engineering terms | 已完成 | focused wording guard test | P2.1 benchmark guardrails | | P2.1 | External benchmark encoded into requirements | 已完成 | benchmark guide + focused guard test + first-viewport status | P3.1 safe lane expansion | -| P3.1 | Extend receipt / replay / drift pattern to more lanes | 未開始 | current retry lane complete | 下一個實作 | +| 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 | ## 後續回報格式 diff --git a/routes/ai_routes.py b/routes/ai_routes.py index aa55476..cacf74e 100644 --- a/routes/ai_routes.py +++ b/routes/ai_routes.py @@ -1864,6 +1864,61 @@ def api_pchome_growth_direct_mapping_candidate_decision_package(): }), 500 +@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-lane-closeout-package') +@login_required +def api_pchome_growth_direct_mapping_candidate_decision_lane_closeout_package(): + """P3 receipt / replay / drift closeout for the safe candidate-decision lane.""" + 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_candidate_decision_lane_closeout_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'} + 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_candidate_decision_lane_closeout_package( + payload, + batch_size=batch_size, + execute_search=execute_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-candidate-decision-package" + ) + return jsonify(package) + except Exception as exc: + logger.error("[PChomeGrowth] direct mapping candidate decision lane closeout 讀取失敗: %s", exc, exc_info=True) + return jsonify({ + "success": False, + "error": "PChome 商品對應候選決策 lane closeout 暫時無法讀取，請稍後再試。", + }), 500 + + @ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-auto-resolution-package') @login_required def api_pchome_growth_direct_mapping_candidate_exception_auto_resolution_package(): diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index 546a226..776a6f9 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -38,6 +38,9 @@ DIRECT_MAPPING_AUTO_SEARCH_PACKAGE_POLICY = ( DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY = ( "read_only_pchome_growth_direct_mapping_candidate_decision_package" ) +DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_POLICY = ( + "read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout" +) DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = ( "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution" ) @@ -2330,6 +2333,283 @@ def build_pchome_direct_mapping_candidate_decision_package( } +def build_pchome_direct_mapping_candidate_decision_lane_closeout_package( + payload: dict[str, Any], + batch_size: int = 5, + *, + execute_search: bool = False, + limit_per_product: int = 8, + max_terms_per_product: int = 5, + min_score: float = 0.45, + search_func: Any = None, + artifact_root: str | Path | None = None, + materialize_artifacts: bool = False, +) -> dict[str, Any]: + """Close out the candidate-decision lane with receipt, replay, drift, and readiness proof.""" + decision_package = build_pchome_direct_mapping_candidate_decision_package( + payload, + batch_size=batch_size, + execute_search=execute_search, + limit_per_product=limit_per_product, + max_terms_per_product=max_terms_per_product, + min_score=min_score, + search_func=search_func, + ) + package = decision_package.get("decision_package") or {} + decisions = list(package.get("candidate_decisions") or []) + exception_receipts = list(package.get("machine_review_exception_receipts") or []) + auto_resolution_artifacts = list( + package.get("machine_review_exception_auto_resolution_artifacts") or [] + ) + summary = decision_package.get("summary") or {} + selected_direct_count = int(summary.get("selected_direct_mapping_count") or 0) + auto_compare_decisions = [ + item + for item in decisions + if item.get("decision") == "route_to_no_write_auto_compare_receipt" + ] + machine_review_decisions = [ + item + for item in decisions + if item.get("decision") == "route_to_machine_review_decision" + ] + receipt_entries = [ + { + "decision_id": item.get("decision_id"), + "decision": item.get("decision"), + "subject": item.get("subject") or {}, + "confidence": item.get("confidence"), + "failure_reasons": list(item.get("failure_reasons") or []), + "ready_for_no_write_receipt": item.get("decision") == "route_to_no_write_auto_compare_receipt", + "requires_auto_resolution": item.get("decision") == "route_to_machine_review_decision", + "writes_database": False, + } + for item in decisions + ] + source_fingerprint_payload = { + "policy": decision_package.get("policy"), + "result": decision_package.get("result"), + "summary": { + "selected_direct_mapping_count": selected_direct_count, + "candidate_decision_count": len(decisions), + "auto_compare_decision_count": len(auto_compare_decisions), + "machine_review_decision_count": len(machine_review_decisions), + "machine_review_exception_receipt_count": len(exception_receipts), + "exception_auto_resolution_artifact_count": len(auto_resolution_artifacts), + }, + "decision_ids": [item.get("decision_id") for item in decisions], + "exception_receipt_ids": [item.get("receipt_id") for item in exception_receipts], + } + source_fingerprint = hashlib.sha256( + json.dumps(source_fingerprint_payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + receipt_id = f"pchome-direct-mapping-candidate-decision-lane-{source_fingerprint[:16]}" + + if not selected_direct_count: + result = "NO_DIRECT_MAPPING_TARGETS" + product_status = "waiting_for_growth_input" + next_machine_action = "wait_for_direct_mapping_backlog" + elif decisions: + result = "DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_READY" + product_status = "ready_for_next_automation" + next_machine_action = "route_auto_compare_to_no_write_receipts_and_exceptions_to_auto_resolution" + else: + result = "WAITING_FOR_DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CANDIDATES" + product_status = "waiting_for_read_only_search" + next_machine_action = "run_controlled_read_only_candidate_search" + + safety = { + "ai_controlled": True, + "read_only_preview": True, + "receipt_replay": True, + "drift_verifier": True, + "product_readiness": True, + "executes_search": bool(execute_search), + "writes_database": False, + "writes_database_count": 0, + "persists_candidate": False, + "syncs_external_offers": False, + "dispatches_telegram": False, + "llm_calls_in_preview": False, + "gemini_allowed": False, + "requires_production_version_truth": True, + } + closeout_summary = { + "selected_direct_mapping_count": selected_direct_count, + "candidate_decision_count": len(decisions), + "auto_compare_decision_count": len(auto_compare_decisions), + "machine_review_decision_count": len(machine_review_decisions), + "machine_review_exception_receipt_count": len(exception_receipts), + "exception_auto_resolution_artifact_count": len(auto_resolution_artifacts), + "receipt_entry_count": len(receipt_entries), + "receipt_payload_hash_ready_count": 1, + "receipt_materialized_count": 0, + "receipt_hash_match_count": 0, + "replay_verified_count": 0, + "drift_count": 0, + "product_ready_count": 1 if product_status == "ready_for_next_automation" else 0, + "primary_human_gate_count": 0, + "writes_database_count": 0, + } + receipt_fingerprint_payload = { + "policy": decision_package.get("policy"), + "result": decision_package.get("result"), + "summary": source_fingerprint_payload["summary"], + "decision_ids": [item.get("decision_id") for item in decisions], + "exception_receipt_ids": [item.get("receipt_id") for item in exception_receipts], + } + receipt_fingerprint = hashlib.sha256( + json.dumps(receipt_fingerprint_payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + replay_verified = source_fingerprint == receipt_fingerprint + drift_count = 0 if replay_verified else 1 + closeout_summary["replay_verified_count"] = 1 if replay_verified else 0 + closeout_summary["drift_count"] = drift_count + receipt_payload = { + "artifact_key": "direct_mapping_candidate_decision_lane_closeout_receipt", + "receipt_id": receipt_id, + "source_policy": DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_POLICY, + "source_decision_policy": decision_package.get("policy"), + "source_decision_result": decision_package.get("result"), + "result": result, + "product_status": product_status, + "next_machine_action": next_machine_action, + "summary": dict(closeout_summary), + "source_fingerprint": source_fingerprint, + "receipt_fingerprint": receipt_fingerprint, + "receipt_entries": receipt_entries, + "exception_receipts": exception_receipts, + "auto_resolution_artifacts": auto_resolution_artifacts, + "safety": dict(safety), + } + receipt_bytes = _canonical_retry_exception_artifact_bytes(receipt_payload) + receipt_sha = hashlib.sha256(receipt_bytes).hexdigest() + receipt_relative_path = ( + "artifacts/pchome_growth/candidate_decision_lane_closeout/" + f"{receipt_id}.json" + ) + receipt_artifact = { + "key": "direct_mapping_candidate_decision_lane_closeout_receipt", + "artifact_type": "candidate_decision_lane_closeout_receipt", + "relative_path": receipt_relative_path, + "payload_sha256": receipt_sha, + "byte_count": len(receipt_bytes), + "payload": receipt_payload, + "materialized": False, + "writes_database": False, + } + root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" + materialized_artifacts: list[dict[str, Any]] = [] + if materialize_artifacts and selected_direct_count: + target_path = root / receipt_relative_path + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_bytes(receipt_bytes) + materialized_artifacts.append({ + "key": receipt_artifact["key"], + "relative_path": receipt_relative_path, + "absolute_path": str(target_path), + "payload_sha256": receipt_sha, + "written_byte_count": target_path.stat().st_size, + "writes_database": False, + }) + receipt_artifact["materialized"] = True + receipt_artifact["absolute_path"] = str(target_path) + receipt_path = root / receipt_relative_path + actual_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest() if receipt_path.exists() else "" + receipt_hash_match = bool(actual_sha) and actual_sha == receipt_sha + closeout_summary["receipt_materialized_count"] = len(materialized_artifacts) or (1 if receipt_hash_match else 0) + closeout_summary["receipt_hash_match_count"] = 1 if receipt_hash_match else 0 + safety["writes_artifact_count"] = len(materialized_artifacts) + + checks = [ + {"check": "source_decision_package_loaded", "passed": bool(decision_package)}, + { + "check": "candidate_decisions_match_summary", + "passed": len(decisions) == int(summary.get("candidate_decision_count") or 0), + }, + {"check": "receipt_payload_hash_is_sha256", "passed": len(receipt_sha) == 64}, + {"check": "receipt_replay_matches_source_fingerprint", "passed": replay_verified}, + {"check": "drift_count_zero", "passed": drift_count == 0}, + { + "check": "receipt_materialized_when_requested", + "passed": (not materialize_artifacts) or (selected_direct_count > 0 and receipt_path.exists()), + }, + { + "check": "receipt_hash_matches_expected_when_materialized", + "passed": (not materialize_artifacts) or receipt_hash_match, + }, + {"check": "candidate_decision_lane_closeout_does_not_write_database", "passed": True}, + ] + + return { + "policy": DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_POLICY, + "result": result, + "success": bool(decision_package.get("success")), + "generated_at": decision_package.get("generated_at"), + "source_policy": decision_package.get("policy"), + "summary": closeout_summary, + "lane_receipt": { + "receipt_id": receipt_id, + "stage": "P3_direct_mapping_candidate_decision_lane_closeout", + "status": result, + "receipt_status": ( + "candidate_decision_lane_receipt_ready" + if decisions + else "candidate_decision_lane_receipt_waiting_for_candidates" + ), + "payload_sha256": receipt_sha, + "source_fingerprint": source_fingerprint, + "materialize_artifacts": bool(materialize_artifacts), + "requires_production_version_truth": True, + }, + "receipt_replay": { + "stage": "P3_direct_mapping_candidate_decision_lane_receipt_replay", + "passed": replay_verified, + "source_fingerprint": source_fingerprint, + "receipt_fingerprint": receipt_fingerprint, + "payload_sha256": receipt_sha, + "writes_database": False, + }, + "drift_verifier": { + "stage": "P3_direct_mapping_candidate_decision_lane_drift_verifier", + "status": "verified" if drift_count == 0 else "drift_detected", + "drift_count": drift_count, + "source_policy": decision_package.get("policy"), + "writes_database": False, + }, + "product_readiness": { + "status": product_status, + "next_machine_action": next_machine_action, + "primary_human_gate_count": 0, + "automation_policy": { + "primary_flow": "ai_controlled", + "manual_review_mode": "exception_only", + "machine_verifiable_evidence": True, + }, + "ready_for_next_automation": product_status == "ready_for_next_automation", + "writes_database": False, + }, + "receipt_artifact": receipt_artifact, + "materialized_artifacts": materialized_artifacts, + "post_receipt_artifact_verifier": { + "expected_sha256": receipt_sha, + "actual_sha256": actual_sha, + "hash_match": receipt_hash_match, + "writes_database": False, + }, + "source_decision_summary": summary, + "checks": checks, + "check_count": len(checks), + "all_checks_passed": all(check.get("passed") is True for check in checks), + "next_actions": [ + "Run controlled read-only search when the lane is waiting for candidates.", + "Route auto-compare decisions to no-write receipts before persistence.", + "Route machine-review decisions to AI auto-resolution artifacts and closeout receipts.", + ], + "safety": safety, + } + + def build_pchome_direct_mapping_candidate_exception_auto_resolution_package( payload: dict[str, Any], batch_size: int = 5, diff --git a/tests/test_pchome_candidate_decision_lane_closeout.py b/tests/test_pchome_candidate_decision_lane_closeout.py new file mode 100644 index 0000000..1336178 --- /dev/null +++ b/tests/test_pchome_candidate_decision_lane_closeout.py @@ -0,0 +1,167 @@ +from flask import Flask + +from services.pchome_mapping_backlog_service import ( + build_pchome_direct_mapping_candidate_decision_lane_closeout_package, +) + + +def _payload(): + return { + "success": True, + "system_name": "MOMO Pro", + "generated_at": "2026-06-28T01:16:02", + "cache_state": "fresh", + "stats": { + "candidate_count": 4, + "mapped_count": 1, + "mapping_rate": 25.0, + "needs_mapping_count": 3, + "review_candidate_count": 1, + "overall_latest_sales_date": "2026-06-24", + "overall_sales_7d": 2020234.0, + "action_counts": { + "先補商品對應": 2, + "確認候選": 1, + "放大價格優勢": 1, + }, + "action_code_counts": { + "map_external_product": 2, + "review_external_candidate": 1, + "amplify_price_advantage": 1, + }, + }, + "opportunities": [ + { + "pchome_product_id": "PCH-1", + "product_name": "Mapped product", + "sales_7d": 0, + "external_price": { + "momo_sku": "M-1", + "price_basis": "unit_price", + "gap_pct": 12.5, + }, + "recommended_action": { + "code": "amplify_price_advantage", + "label": "放大價格優勢", + }, + "priority_score": 75.0, + }, + { + "pchome_product_id": "PCH-2", + "product_name": "Direct mapping product 40ml x2", + "sales_7d": 9800, + "pchome_price": 1200, + "external_price": None, + "recommended_action": {"code": "map_external_product", "label": "先補商品對應"}, + "priority_score": 88.0, + "reason_lines": ["需要補商品對應"], + }, + { + "pchome_product_id": "PCH-3", + "product_name": "Review candidate product", + "sales_7d": 1200, + "external_price": None, + "review_candidate": { + "id": 725, + "momo_sku": "5868343", + "momo_name": "MOMO candidate", + "quality_score": 94.8, + }, + "recommended_action": {"code": "review_external_candidate", "label": "確認候選"}, + "priority_score": 64.0, + }, + ], + } + + +def test_candidate_decision_lane_closeout_outputs_receipt_replay_drift_and_readiness(tmp_path): + def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score): + return True, "found", [ + { + "product_id": "MOMO-1", + "name": "Direct mapping product 40ml x2", + "price": 999, + "target_pchome_product_id": "PCH-2", + "target_pchome_name": "Direct mapping product 40ml x2", + "target_match_score": 0.92, + "auto_compare_type": "total_price", + "target_hard_veto": False, + "target_price_basis": "total_price", + "target_gap_pct": 16.8, + "target_search_term": "direct mapping product 40ml x2", + }, + { + "product_id": "MOMO-2", + "name": "Variant candidate", + "price": 899, + "target_pchome_product_id": "PCH-2", + "target_pchome_name": "Direct mapping product 40ml x2", + "target_match_score": 0.51, + "auto_compare_type": "manual_review", + "target_hard_veto": False, + }, + ] + + package = build_pchome_direct_mapping_candidate_decision_lane_closeout_package( + _payload(), + batch_size=1, + execute_search=True, + limit_per_product=3, + search_func=fake_search, + artifact_root=tmp_path, + ) + + assert package["policy"] == "read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout" + assert package["result"] == "DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CLOSEOUT_READY" + assert package["summary"]["candidate_decision_count"] == 2 + assert package["summary"]["auto_compare_decision_count"] == 1 + assert package["summary"]["machine_review_decision_count"] == 1 + assert package["summary"]["receipt_payload_hash_ready_count"] == 1 + assert package["summary"]["replay_verified_count"] == 1 + assert package["summary"]["drift_count"] == 0 + assert package["summary"]["primary_human_gate_count"] == 0 + assert package["lane_receipt"]["receipt_status"] == "candidate_decision_lane_receipt_ready" + assert package["receipt_replay"]["passed"] is True + assert package["drift_verifier"]["status"] == "verified" + assert package["product_readiness"]["status"] == "ready_for_next_automation" + assert package["product_readiness"]["automation_policy"]["primary_flow"] == "ai_controlled" + assert package["product_readiness"]["primary_human_gate_count"] == 0 + assert package["safety"]["writes_database"] is False + assert package["safety"]["writes_artifact_count"] == 0 + assert package["all_checks_passed"] is True + + +def test_candidate_decision_lane_closeout_route_defaults_to_cached_no_db_no_search(monkeypatch): + 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 candidate decision lane closeout 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-candidate-decision-lane-closeout-package?batch_size=1" + ): + response = ( + routes + .api_pchome_growth_direct_mapping_candidate_decision_lane_closeout_package + .__wrapped__() + ) + + payload = response.get_json() + assert payload["success"] is True + assert payload["policy"] == "read_only_pchome_growth_direct_mapping_candidate_decision_lane_closeout" + assert payload["source_endpoint"] == ( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-package" + ) + assert payload["result"] == "WAITING_FOR_DIRECT_MAPPING_CANDIDATE_DECISION_LANE_CANDIDATES" + assert payload["summary"]["candidate_decision_count"] == 0 + assert payload["summary"]["replay_verified_count"] == 1 + assert payload["summary"]["drift_count"] == 0 + assert payload["product_readiness"]["status"] == "waiting_for_read_only_search" + assert payload["product_readiness"]["primary_human_gate_count"] == 0 + assert payload["safety"]["executes_search"] is False + assert payload["safety"]["writes_database"] is False