diff --git a/routes/ai_routes.py b/routes/ai_routes.py index 987c535..c551002 100644 --- a/routes/ai_routes.py +++ b/routes/ai_routes.py @@ -1917,6 +1917,61 @@ def api_pchome_growth_direct_mapping_candidate_exception_auto_resolution_package }), 500 +@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-resolution-closeout-package') +@login_required +def api_pchome_growth_direct_mapping_candidate_exception_resolution_closeout_package(): + """P2 no-write closeout package for direct mapping candidate exception resolution.""" + 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_exception_resolution_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'} + execute_retry_search = str(request.args.get('execute_retry_search') 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_exception_resolution_closeout_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, + ) + package["source_endpoint"] = ( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-auto-resolution-package" + ) + return jsonify(package) + except Exception as exc: + logger.error("[PChomeGrowth] direct mapping candidate exception resolution closeout package 讀取失敗: %s", exc, exc_info=True) + return jsonify({ + "success": False, + "error": "PChome 商品對應候選例外解法收斂暫時無法讀取，請稍後再試。", + }), 500 + + @ai_bp.route('/api/ai/pchome-growth/ai-automation-readiness') @login_required def api_pchome_growth_ai_automation_readiness(): diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index 7b84bb9..cae9594 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -41,6 +41,9 @@ DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY = ( DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = ( "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution" ) +DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = ( + "read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout" +) 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" @@ -1646,6 +1649,197 @@ def _summarize_exception_auto_resolution_artifacts(artifacts: list[dict[str, Any } +def _expanded_retry_search_terms_from_artifact(artifact: dict[str, Any]) -> list[str]: + terms: list[str] = [] + resolvers = artifact.get("resolvers") or {} + for key in ("unit_basis_search_expansion", "brand_spec_search_expansion"): + resolver = resolvers.get(key) or {} + terms.extend(str(term).strip() for term in resolver.get("expanded_search_terms") or []) + return list(dict.fromkeys(term for term in terms if term)) + + +def _retry_search_targets_from_artifact( + artifact: dict[str, Any], + *, + max_terms_per_product: int, +) -> list[dict[str, Any]]: + subject = artifact.get("subject") or {} + terms = _expanded_retry_search_terms_from_artifact(artifact)[:max_terms_per_product] + return [ + { + "product_id": subject.get("target_pchome_product_id") or "", + "name": term, + "price": subject.get("pchome_price"), + "source_artifact_id": artifact.get("artifact_id"), + "source_resolution_receipt_id": artifact.get("source_receipt_id"), + } + for term in terms + if term + ] + + +def _run_retry_search_for_artifact( + artifact: dict[str, Any], + *, + execute_retry_search: bool, + limit_per_product: int, + max_terms_per_product: int, + min_score: float, + search_func: Any = None, +) -> dict[str, Any]: + targets = _retry_search_targets_from_artifact( + artifact, + max_terms_per_product=max_terms_per_product, + ) + if not execute_retry_search or not targets: + return { + "executed": False, + "search_success": None, + "search_message": "retry_search_not_executed" if targets else "no_retry_search_terms", + "targets": targets, + "candidate_count": 0, + "candidates": [], + } + + if search_func is None: + from services.momo_crawler import search_momo_products_for_pchome_products + + search_func = search_momo_products_for_pchome_products + search_success, search_message, candidates = search_func( + targets, + limit_per_product=limit_per_product, + max_products=len(targets), + max_terms_per_product=1, + min_score=min_score, + ) + candidates = list(candidates or []) + for candidate in candidates: + candidate["source_resolution_artifact_id"] = artifact.get("artifact_id") + candidate["source_resolution_receipt_id"] = artifact.get("source_receipt_id") + candidate["retry_search_source"] = "candidate_exception_resolution_closeout" + + return { + "executed": True, + "search_success": bool(search_success), + "search_message": search_message, + "targets": targets, + "candidate_count": len(candidates), + "candidates": candidates[:20], + } + + +def _build_candidate_exception_resolution_closeout_receipt( + artifact: dict[str, Any], + retry_search_result: dict[str, Any], +) -> dict[str, Any]: + resolvers = artifact.get("resolvers") or {} + evidence_delta = resolvers.get("named_candidate_evidence_delta") or {} + retry_search_ready = bool(_expanded_retry_search_terms_from_artifact(artifact)) + closeout_basis = { + "artifact_id": artifact.get("artifact_id"), + "resolver_keys": sorted(resolvers.keys()), + "retry_search_ready": retry_search_ready, + "retry_candidate_count": retry_search_result.get("candidate_count"), + } + closeout_hash = hashlib.sha256( + json.dumps(closeout_basis, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + retry_candidate_count = int(retry_search_result.get("candidate_count") or 0) + return { + "receipt_id": f"pchome-direct-mapping-exception-closeout-{closeout_hash[:16]}", + "source_artifact_id": artifact.get("artifact_id"), + "source_receipt_id": artifact.get("source_receipt_id"), + "source_decision_id": artifact.get("source_decision_id"), + "stage": "P2_machine_verifiable_exception_resolution_closeout", + "resolution_status": "AUTO_RESOLUTION_CLOSEOUT_READY", + "completed_resolvers": sorted(resolvers.keys()), + "evidence_delta": { + "ready": bool(evidence_delta), + "missing_evidence_keys": list(evidence_delta.get("missing_evidence_keys") or []), + "resolution": evidence_delta.get("resolution") or "not_applicable", + }, + "retry_search": { + "ready": retry_search_ready, + "executed": bool(retry_search_result.get("executed")), + "search_success": retry_search_result.get("search_success"), + "search_message": retry_search_result.get("search_message"), + "target_count": len(retry_search_result.get("targets") or []), + "candidate_count": retry_candidate_count, + "targets": retry_search_result.get("targets") or [], + "candidates": retry_search_result.get("candidates") or [], + }, + "next_package": ( + "direct_mapping_candidate_decision_package_after_retry" + if retry_search_ready + else "no_write_receipt_verifier_when_identity_delta_clears" + ), + "guardrails": { + "machine_actionable": True, + "can_auto_execute_read_only": retry_search_ready, + "writes_database": False, + "persists_candidate": False, + "requires_no_write_receipt": True, + "requires_verifier_before_persistence": True, + }, + } + + +def _build_candidate_exception_resolution_closeout_receipts( + artifacts: list[dict[str, Any]], + *, + 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, +) -> list[dict[str, Any]]: + receipts = [] + for artifact in artifacts: + retry_search_result = _run_retry_search_for_artifact( + artifact, + 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, + ) + receipts.append(_build_candidate_exception_resolution_closeout_receipt(artifact, retry_search_result)) + return receipts + + +def _summarize_exception_resolution_closeout_receipts(receipts: list[dict[str, Any]]) -> dict[str, int]: + retry_search_ready_count = 0 + retry_search_executed_count = 0 + retry_candidate_count = 0 + evidence_delta_closeout_count = 0 + ready_for_next_decision_count = 0 + for receipt in receipts: + retry_search = receipt.get("retry_search") or {} + evidence_delta = receipt.get("evidence_delta") or {} + retry_ready = bool(retry_search.get("ready")) + retry_executed = bool(retry_search.get("executed")) + retry_candidates = int(retry_search.get("candidate_count") or 0) + if retry_ready: + retry_search_ready_count += 1 + if retry_executed: + retry_search_executed_count += 1 + if evidence_delta.get("ready"): + evidence_delta_closeout_count += 1 + retry_candidate_count += retry_candidates + if retry_candidates: + ready_for_next_decision_count += 1 + + return { + "exception_resolution_closeout_receipt_count": len(receipts), + "retry_search_ready_count": retry_search_ready_count, + "retry_search_executed_count": retry_search_executed_count, + "retry_candidate_count": retry_candidate_count, + "evidence_delta_closeout_count": evidence_delta_closeout_count, + "ready_for_next_candidate_decision_count": ready_for_next_decision_count, + "writes_database_count": 0, + } + + def build_pchome_direct_mapping_auto_search_package( payload: dict[str, Any], batch_size: int = 5, @@ -1987,6 +2181,96 @@ def build_pchome_direct_mapping_candidate_exception_auto_resolution_package( } +def build_pchome_direct_mapping_candidate_exception_resolution_closeout_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, +) -> dict[str, Any]: + """Close out candidate exception auto-resolution artifacts into retry-ready receipts.""" + auto_resolution = build_pchome_direct_mapping_candidate_exception_auto_resolution_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 = auto_resolution.get("auto_resolution_package") or {} + artifacts = list(package.get("auto_resolution_artifacts") or []) + closeout_receipts = _build_candidate_exception_resolution_closeout_receipts( + artifacts, + 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, + ) + closeout_summary = _summarize_exception_resolution_closeout_receipts(closeout_receipts) + selected_direct_count = int((auto_resolution.get("summary") or {}).get("selected_direct_mapping_count") or 0) + artifact_count = int((auto_resolution.get("summary") or {}).get("exception_auto_resolution_artifact_count") or 0) + + if not selected_direct_count: + result = "NO_DIRECT_MAPPING_TARGETS" + elif closeout_receipts: + result = "DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_READY" + elif artifact_count: + result = "DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_READY" + else: + result = "WAITING_FOR_DIRECT_MAPPING_CANDIDATES" + + return { + "policy": DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY, + "result": result, + "success": bool(auto_resolution.get("success")), + "generated_at": auto_resolution.get("generated_at"), + "source_policy": auto_resolution.get("policy"), + "stats": auto_resolution.get("stats") or {}, + "backlog": auto_resolution.get("backlog") or {}, + "summary": { + "direct_mapping_count": int((auto_resolution.get("summary") or {}).get("direct_mapping_count") or 0), + "selected_direct_mapping_count": selected_direct_count, + "candidate_decision_count": int((auto_resolution.get("summary") or {}).get("candidate_decision_count") or 0), + "machine_review_exception_receipt_count": int( + (auto_resolution.get("summary") or {}).get("machine_review_exception_receipt_count") or 0 + ), + "exception_auto_resolution_artifact_count": artifact_count, + **closeout_summary, + }, + "closeout_package": { + "stage": "P2_machine_verifiable_exception_resolution_closeout", + "execute_search": bool(execute_search), + "execute_retry_search": bool(execute_retry_search), + "closeout_receipts": closeout_receipts, + "resolution_mode": "ai_controlled_read_only", + }, + "upstream_auto_resolution_summary": auto_resolution.get("summary") or {}, + "next_actions": [ + "Feed retry_search candidates back into the candidate decision package when retry_candidate_count is nonzero.", + "Feed evidence_delta closeouts into no-write verifier receipts before any persistence.", + "Keep database writes behind controlled apply, rollback, and production readback.", + ], + "safety": { + "read_only_preview": True, + "executes_search": bool(execute_search), + "executes_retry_search": bool(execute_retry_search), + "writes_database": False, + "persists_candidate": False, + "syncs_external_offers": False, + "dispatches_telegram": False, + "llm_calls_in_preview": 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) @@ -2599,6 +2883,19 @@ def build_pchome_growth_ai_automation_readiness( search_summary = decision_package.get("upstream_search_summary") or {} decision_summary = decision_package.get("summary") or {} receipt_summary = receipt_gate.get("summary") or {} + exception_artifacts = list( + (decision_package.get("decision_package") or {}).get( + "machine_review_exception_auto_resolution_artifacts" + ) + or [] + ) + exception_closeout_receipts = _build_candidate_exception_resolution_closeout_receipts( + exception_artifacts, + execute_retry_search=False, + ) + exception_closeout_summary = _summarize_exception_resolution_closeout_receipts( + exception_closeout_receipts + ) direct_mapping_count = int(backlog.get("direct_mapping_count") or 0) selected_search_targets = int(search_summary.get("selected_direct_mapping_count") or 0) @@ -2610,6 +2907,9 @@ def build_pchome_growth_ai_automation_readiness( exception_auto_resolution_artifact_count = int( decision_summary.get("exception_auto_resolution_artifact_count") or 0 ) + exception_resolution_closeout_receipt_count = int( + exception_closeout_summary.get("exception_resolution_closeout_receipt_count") or 0 + ) retry_search_action_count = int(decision_summary.get("retry_search_action_count") or 0) waiting_candidate_count = selected_search_targets if not candidate_decision_count else 0 receipt_count = int(receipt_summary.get("receipt_count") or 0) @@ -2623,6 +2923,7 @@ def build_pchome_growth_ai_automation_readiness( "ai_exception_count": exception_count, "exception_receipt_count": exception_receipt_count, "exception_auto_resolution_artifact_count": exception_auto_resolution_artifact_count, + "exception_resolution_closeout_receipt_count": exception_resolution_closeout_receipt_count, "retry_search_action_count": retry_search_action_count, "routes": [ { @@ -2641,6 +2942,8 @@ def build_pchome_growth_ai_automation_readiness( if not direct_mapping_count and ready_receipt_count: result = "AI_AUTOMATION_READY_FOR_CONTROLLED_APPLY" + elif exception_resolution_closeout_receipt_count: + result = "AI_AUTOMATION_EXCEPTION_RESOLUTION_CLOSEOUT_READY" elif exception_auto_resolution_artifact_count: result = "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_READY" elif candidate_decision_count: @@ -2685,6 +2988,14 @@ def build_pchome_growth_ai_automation_readiness( f"{retry_search_action_count} 組 retry search 動作", "執行變體/組合判別、命名證據差分與單位基準擴搜尋", ), + _automation_lane( + "candidate_exception_resolution_closeout", + "候選例外解法收斂", + "ready" if exception_resolution_closeout_receipt_count else "waiting", + exception_resolution_closeout_receipt_count, + f"{exception_closeout_summary.get('retry_search_ready_count', 0)} 筆可進 retry search", + "把 resolver artifacts 收斂成 retry search / verifier receipts", + ), _automation_lane( "evidence_receipts", "證據收據", @@ -2718,7 +3029,15 @@ def build_pchome_growth_ai_automation_readiness( "machine_review_decision_count": int(decision_summary.get("machine_review_decision_count") or 0), "machine_review_exception_receipt_count": exception_receipt_count, "exception_auto_resolution_artifact_count": exception_auto_resolution_artifact_count, + "exception_resolution_closeout_receipt_count": exception_resolution_closeout_receipt_count, "retry_search_action_count": retry_search_action_count, + "retry_search_ready_count": int(exception_closeout_summary.get("retry_search_ready_count") or 0), + "retry_search_executed_count": int(exception_closeout_summary.get("retry_search_executed_count") or 0), + "retry_candidate_count": int(exception_closeout_summary.get("retry_candidate_count") or 0), + "evidence_delta_closeout_count": int(exception_closeout_summary.get("evidence_delta_closeout_count") or 0), + "ready_for_next_candidate_decision_count": int( + exception_closeout_summary.get("ready_for_next_candidate_decision_count") or 0 + ), "variant_bundle_discriminator_count": int( decision_summary.get("variant_bundle_discriminator_count") or 0 ), diff --git a/tests/test_pchome_mapping_backlog_report.py b/tests/test_pchome_mapping_backlog_report.py index 8b3a065..04480f8 100644 --- a/tests/test_pchome_mapping_backlog_report.py +++ b/tests/test_pchome_mapping_backlog_report.py @@ -73,6 +73,7 @@ from services.pchome_mapping_backlog_service import ( build_pchome_direct_mapping_auto_search_package, build_pchome_direct_mapping_candidate_decision_package, build_pchome_direct_mapping_candidate_exception_auto_resolution_package, + build_pchome_direct_mapping_candidate_exception_resolution_closeout_package, build_pchome_growth_ai_automation_readiness, build_pchome_mapping_operator_preview, parse_pchome_product_page_evidence_html, @@ -466,6 +467,63 @@ def test_direct_mapping_candidate_exception_auto_resolution_builds_machine_artif assert package["safety"]["persists_candidate"] is False +def test_direct_mapping_candidate_exception_resolution_closeout_executes_retry_search_without_db_write(): + 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", + "name": "Direct mapping product 40ml 單入", + "price": 520, + "target_pchome_product_id": "PCH-2", + "target_match_score": 0.82, + "auto_compare_type": "unit_price", + "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_candidate_exception_resolution_closeout_package( + _payload(), + batch_size=1, + execute_search=True, + execute_retry_search=True, + max_terms_per_product=6, + search_func=fake_search, + ) + + receipts = package["closeout_package"]["closeout_receipts"] + assert package["policy"] == "read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout" + assert package["result"] == "DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_READY" + assert package["summary"]["exception_resolution_closeout_receipt_count"] == 1 + assert package["summary"]["retry_search_ready_count"] == 1 + assert package["summary"]["retry_search_executed_count"] == 1 + assert package["summary"]["retry_candidate_count"] == 1 + assert package["summary"]["ready_for_next_candidate_decision_count"] == 1 + assert receipts[0]["resolution_status"] == "AUTO_RESOLUTION_CLOSEOUT_READY" + assert receipts[0]["retry_search"]["executed"] is True + assert receipts[0]["retry_search"]["candidate_count"] == 1 + assert receipts[0]["retry_search"]["candidates"][0]["source_resolution_artifact_id"] == receipts[0]["source_artifact_id"] + assert receipts[0]["guardrails"]["writes_database"] is False + assert package["safety"]["executes_retry_search"] is True + 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) @@ -556,15 +614,20 @@ def test_ai_automation_readiness_reports_exception_auto_resolution_ready(): ) lanes = {lane["key"]: lane for lane in readiness["automation_lanes"]} - assert readiness["result"] == "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_READY" + assert readiness["result"] == "AI_AUTOMATION_EXCEPTION_RESOLUTION_CLOSEOUT_READY" assert readiness["summary"]["candidate_decision_count"] == 1 assert readiness["summary"]["machine_review_exception_receipt_count"] == 1 assert readiness["summary"]["exception_auto_resolution_artifact_count"] == 1 + assert readiness["summary"]["exception_resolution_closeout_receipt_count"] == 1 + assert readiness["summary"]["evidence_delta_closeout_count"] == 1 assert readiness["summary"]["variant_bundle_discriminator_count"] == 1 assert readiness["summary"]["named_candidate_evidence_delta_count"] == 1 assert readiness["ai_exception_auto_resolution"]["exception_auto_resolution_artifact_count"] == 1 + assert readiness["ai_exception_auto_resolution"]["exception_resolution_closeout_receipt_count"] == 1 assert lanes["candidate_exception_auto_resolution"]["status"] == "ready" assert lanes["candidate_exception_auto_resolution"]["value"] == 1 + assert lanes["candidate_exception_resolution_closeout"]["status"] == "ready" + assert lanes["candidate_exception_resolution_closeout"]["value"] == 1 assert readiness["summary"]["primary_human_gate_count"] == 0 assert readiness["summary"]["writes_database_count"] == 0 assert readiness["safety"]["writes_database"] is False @@ -14728,6 +14791,37 @@ def test_direct_mapping_candidate_exception_auto_resolution_route_uses_cached_pa assert payload["safety"]["writes_database"] is False +def test_direct_mapping_candidate_exception_resolution_closeout_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 exception resolution closeout package 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-exception-resolution-closeout-package?batch_size=1" + ): + response = routes.api_pchome_growth_direct_mapping_candidate_exception_resolution_closeout_package.__wrapped__() + + payload = response.get_json() + assert payload["success"] is True + assert payload["policy"] == "read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout" + assert payload["source_endpoint"] == ( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-auto-resolution-package" + ) + assert payload["result"] == "WAITING_FOR_DIRECT_MAPPING_CANDIDATES" + assert payload["summary"]["exception_resolution_closeout_receipt_count"] == 0 + assert payload["closeout_package"]["resolution_mode"] == "ai_controlled_read_only" + assert payload["safety"]["executes_search"] is False + assert payload["safety"]["executes_retry_search"] 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