diff --git a/routes/ai_routes.py b/routes/ai_routes.py index b311b18..987c535 100644 --- a/routes/ai_routes.py +++ b/routes/ai_routes.py @@ -1864,6 +1864,59 @@ def api_pchome_growth_direct_mapping_candidate_decision_package(): }), 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(): + """P2 no-write auto-resolution package for direct mapping candidate exceptions.""" + 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_auto_resolution_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'} + 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_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, + ) + 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 exception auto-resolution 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 3754ffd..c52a04e 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_EXCEPTION_AUTO_RESOLUTION_POLICY = ( + "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution" +) 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" @@ -1455,6 +1458,182 @@ def _build_candidate_exception_receipt(decision: dict[str, Any]) -> dict[str, An } +def _unit_basis_search_terms_from_subject(subject: dict[str, Any], max_terms: int = 8) -> list[str]: + pchome_name = str(subject.get("pchome_product_name") or "").strip() + momo_name = str(subject.get("momo_product_name") or "").strip() + source_name = pchome_name or momo_name + terms = _build_direct_mapping_search_terms(source_name, max_terms=3) + basis = parse_unit_package_basis(source_name) if source_name else {} + unit_measure = basis.get("unit_pricing_measure") or {} + base_measure = basis.get("unit_pricing_base_measure") or {} + quantity_terms: list[str] = [] + if unit_measure.get("value") and unit_measure.get("unit"): + quantity_terms.append(f"{unit_measure.get('value'):g}{unit_measure.get('unit')}") + if base_measure.get("value") and base_measure.get("unit"): + quantity_terms.append(f"每{base_measure.get('value'):g}{base_measure.get('unit')}") + if basis.get("estimated_total_quantity") and basis.get("unit_label"): + quantity_terms.append(f"{basis.get('estimated_total_quantity'):g}{basis.get('unit_label')}") + if basis.get("multipliers"): + quantity_terms.append("x".join(str(item) for item in basis.get("multipliers") or [])) + + for quantity_term in quantity_terms: + if source_name: + terms.append(f"{source_name} {quantity_term}") + if source_name and ("bundle_or_promo" in (basis.get("risk_signals") or [])): + terms.append(f"{source_name} 單入") + if source_name and quantity_terms: + terms.append(f"{source_name} 單位價") + return list(dict.fromkeys(term for term in terms if term))[:max_terms] + + +def _build_variant_bundle_discriminator(subject: dict[str, Any], failure_reasons: list[str]) -> dict[str, Any]: + pchome_name = str(subject.get("pchome_product_name") or "").strip() + momo_name = str(subject.get("momo_product_name") or "").strip() + target_basis = parse_unit_package_basis(pchome_name) if pchome_name else {} + momo_basis = parse_unit_package_basis(momo_name) if momo_name else {} + signals = set(target_basis.get("risk_signals") or []) + signals.update(momo_basis.get("risk_signals") or []) + if "auto_compare_type_not_receipt_ready" in failure_reasons: + signals.add("needs_auto_compare_type_resolution") + if "target_hard_veto_true" in failure_reasons: + signals.add("target_identity_veto_blocks_receipt") + + return { + "resolver": "variant_bundle_discriminator", + "decision": "blocks_no_write_receipt_until_resolved" if signals else "identity_delta_only", + "risk_signals": sorted(signals), + "target_unit_package_basis": target_basis, + "momo_unit_package_basis": momo_basis, + "checks": [ + "same_brand_or_named_line", + "same_quantity_or_convertible_unit_basis", + "same_variant_or_color_scope", + "same_bundle_or_single_item_scope", + ], + "writes_database": False, + } + + +def _build_named_candidate_evidence_delta(subject: dict[str, Any], failure_reasons: list[str]) -> dict[str, Any]: + evidence_keys = [ + "target_pchome_product_id", + "pchome_product_name", + "momo_product_id", + "momo_product_name", + "confidence", + "failure_reasons", + ] + missing_keys = [ + key + for key in ("target_pchome_product_id", "pchome_product_name", "momo_product_id", "momo_product_name") + if not subject.get(key) + ] + return { + "resolver": "named_candidate_evidence_delta", + "named_evidence_keys": evidence_keys, + "missing_evidence_keys": missing_keys, + "failure_reasons": failure_reasons, + "resolution": "ready_for_retry_search" if not missing_keys else "drop_incomplete_candidate_and_retry_search", + "writes_database": False, + } + + +def _build_candidate_exception_auto_resolution_artifact(receipt: dict[str, Any]) -> dict[str, Any]: + subject = receipt.get("subject") or {} + failure_reasons = list(receipt.get("failure_reasons") or []) + next_actions = list(receipt.get("next_machine_actions") or []) + artifact_basis = { + "receipt_id": receipt.get("receipt_id"), + "failure_reasons": failure_reasons, + "next_actions": next_actions, + "subject": subject, + } + artifact_hash = hashlib.sha256( + json.dumps(artifact_basis, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + resolvers: dict[str, Any] = {} + if "run_variant_bundle_discriminator" in next_actions: + resolvers["variant_bundle_discriminator"] = _build_variant_bundle_discriminator( + subject, + failure_reasons, + ) + if "build_named_candidate_evidence_delta" in next_actions: + resolvers["named_candidate_evidence_delta"] = _build_named_candidate_evidence_delta( + subject, + failure_reasons, + ) + if "expand_search_terms_with_unit_basis" in next_actions: + resolvers["unit_basis_search_expansion"] = { + "resolver": "unit_basis_search_expansion", + "expanded_search_terms": _unit_basis_search_terms_from_subject(subject), + "retry_package": "direct_mapping_auto_search_package", + "writes_database": False, + } + if "expand_search_terms_with_brand_spec_anchors" in next_actions: + resolvers["brand_spec_search_expansion"] = { + "resolver": "brand_spec_search_expansion", + "expanded_search_terms": _build_direct_mapping_search_terms( + str(subject.get("pchome_product_name") or subject.get("momo_product_name") or ""), + max_terms=6, + ), + "retry_package": "direct_mapping_auto_search_package", + "writes_database": False, + } + + return { + "artifact_id": f"pchome-direct-mapping-exception-resolution-{artifact_hash[:16]}", + "source_receipt_id": receipt.get("receipt_id"), + "source_decision_id": receipt.get("source_decision_id"), + "stage": "P2_machine_verifiable_exception_auto_resolution", + "subject": subject, + "failure_reasons": failure_reasons, + "machine_actions": next_actions, + "resolvers": resolvers, + "resolution_status": "AUTO_RESOLUTION_PLANNED", + "next_package": "direct_mapping_candidate_decision_package_after_retry", + "guardrails": { + "machine_actionable": True, + "can_auto_execute_read_only": True, + "writes_database": False, + "persists_candidate": False, + "requires_verifier_before_persistence": True, + }, + } + + +def _build_candidate_exception_auto_resolution_artifacts( + exception_receipts: list[dict[str, Any]], +) -> list[dict[str, Any]]: + return [ + _build_candidate_exception_auto_resolution_artifact(receipt) + for receipt in exception_receipts + ] + + +def _summarize_exception_auto_resolution_artifacts(artifacts: list[dict[str, Any]]) -> dict[str, int]: + resolver_counts = { + "variant_bundle_discriminator_count": 0, + "named_candidate_evidence_delta_count": 0, + "unit_basis_search_expansion_count": 0, + "brand_spec_search_expansion_count": 0, + } + retry_search_action_count = 0 + for artifact in artifacts: + resolvers = artifact.get("resolvers") or {} + for key in resolver_counts: + resolver_name = key.removesuffix("_count") + if resolver_name in resolvers: + resolver_counts[key] += 1 + if any(key.endswith("search_expansion") for key in resolvers): + retry_search_action_count += 1 + return { + "exception_auto_resolution_artifact_count": len(artifacts), + "retry_search_action_count": retry_search_action_count, + **resolver_counts, + "writes_database_count": 0, + } + + def build_pchome_direct_mapping_auto_search_package( payload: dict[str, Any], batch_size: int = 5, @@ -1643,6 +1822,12 @@ def build_pchome_direct_mapping_candidate_decision_package( _build_candidate_exception_receipt(envelope) for envelope in machine_review_decisions ] + exception_auto_resolution_artifacts = _build_candidate_exception_auto_resolution_artifacts( + machine_review_exception_receipts + ) + exception_auto_resolution_summary = _summarize_exception_auto_resolution_artifacts( + exception_auto_resolution_artifacts + ) if not int((search_package.get("summary") or {}).get("selected_direct_mapping_count") or 0): result = "NO_DIRECT_MAPPING_TARGETS" @@ -1669,6 +1854,7 @@ def build_pchome_direct_mapping_candidate_decision_package( "auto_compare_decision_count": len(auto_compare_decisions), "machine_review_decision_count": len(machine_review_decisions), "machine_review_exception_receipt_count": len(machine_review_exception_receipts), + **exception_auto_resolution_summary, "can_auto_persist_now_count": 0, "writes_database_count": 0, "persists_candidate_count": 0, @@ -1678,6 +1864,7 @@ def build_pchome_direct_mapping_candidate_decision_package( "execute_search": bool(execute_search), "candidate_decisions": decision_envelopes, "machine_review_exception_receipts": machine_review_exception_receipts, + "machine_review_exception_auto_resolution_artifacts": exception_auto_resolution_artifacts, "manual_review_mode": "exception_only", }, "decision_acceptance_policy": { @@ -1694,7 +1881,85 @@ def build_pchome_direct_mapping_candidate_decision_package( "next_actions": [ "Run controlled read-only search first when candidate_decision_count is zero.", "Send auto-compare decisions to no-write receipt generation before any persistence.", - "Route machine-review decisions through exception receipts with named failure reasons and next machine actions.", + "Route machine-review decisions through exception receipts and auto-resolution artifacts.", + ], + "safety": { + "read_only_preview": True, + "executes_search": bool(execute_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_direct_mapping_candidate_exception_auto_resolution_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, +) -> dict[str, Any]: + """Build a no-write auto-resolution package for machine-review candidate exceptions.""" + 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 {} + exception_receipts = list(package.get("machine_review_exception_receipts") or []) + artifacts = list(package.get("machine_review_exception_auto_resolution_artifacts") or []) + summary = _summarize_exception_auto_resolution_artifacts(artifacts) + + selected_direct_count = int((decision_package.get("summary") or {}).get("selected_direct_mapping_count") or 0) + candidate_decision_count = int((decision_package.get("summary") or {}).get("candidate_decision_count") or 0) + if not selected_direct_count: + result = "NO_DIRECT_MAPPING_TARGETS" + elif artifacts: + result = "DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_READY" + elif candidate_decision_count: + result = "DIRECT_MAPPING_CANDIDATE_EXCEPTIONS_CLEAR" + else: + result = "WAITING_FOR_DIRECT_MAPPING_CANDIDATES" + + return { + "policy": DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY, + "result": result, + "success": bool(decision_package.get("success")), + "generated_at": decision_package.get("generated_at"), + "source_policy": decision_package.get("policy"), + "stats": decision_package.get("stats") or {}, + "backlog": decision_package.get("backlog") or {}, + "summary": { + "direct_mapping_count": int((decision_package.get("summary") or {}).get("direct_mapping_count") or 0), + "selected_direct_mapping_count": selected_direct_count, + "candidate_decision_count": candidate_decision_count, + "machine_review_exception_receipt_count": len(exception_receipts), + **summary, + }, + "auto_resolution_package": { + "stage": "P2_machine_verifiable_exception_auto_resolution", + "execute_search": bool(execute_search), + "exception_receipts": exception_receipts, + "auto_resolution_artifacts": artifacts, + "resolution_mode": "ai_controlled_read_only", + }, + "upstream_decision_summary": decision_package.get("summary") or {}, + "next_actions": [ + "Use variant_bundle_discriminator and named_candidate_evidence_delta before retry search.", + "Use unit_basis_search_expansion for hard-veto unit candidates before another candidate decision package.", + "Only route resolved candidates to no-write receipt and verifier packages before persistence.", ], "safety": { "read_only_preview": True, @@ -2330,6 +2595,10 @@ def build_pchome_growth_ai_automation_readiness( exception_receipt_count = int( decision_summary.get("machine_review_exception_receipt_count") or 0 ) + exception_auto_resolution_artifact_count = int( + decision_summary.get("exception_auto_resolution_artifact_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) ready_receipt_count = int(receipt_summary.get("ready_for_auto_persistence_count") or 0) @@ -2341,11 +2610,13 @@ def build_pchome_growth_ai_automation_readiness( PRIMARY_HUMAN_GATE_COUNT_KEY: 0, "ai_exception_count": exception_count, "exception_receipt_count": exception_receipt_count, + "exception_auto_resolution_artifact_count": exception_auto_resolution_artifact_count, + "retry_search_action_count": retry_search_action_count, "routes": [ { "source": "candidate_decision_package", "condition": "not_ready_for_no_write_receipt", - "auto_resolution": "build_failure_reasons_and_next_machine_action", + "auto_resolution": "build_exception_receipts_and_auto_resolution_artifacts", }, { "source": "evidence_receipts", @@ -2358,6 +2629,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_auto_resolution_artifact_count: + result = "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_READY" elif candidate_decision_count: result = "AI_AUTOMATION_CANDIDATE_DECISIONS_READY" elif direct_mapping_count and selected_search_targets: @@ -2392,6 +2665,14 @@ def build_pchome_growth_ai_automation_readiness( f"等待 {waiting_candidate_count} 筆候選" if waiting_candidate_count else "可輸出 decision envelope", "將候選分流到 no-write receipt 或 AI 例外決策", ), + _automation_lane( + "candidate_exception_auto_resolution", + "候選例外自動解法", + "ready" if exception_auto_resolution_artifact_count else ("planned" if exception_receipt_count else "waiting"), + exception_auto_resolution_artifact_count, + f"{retry_search_action_count} 組 retry search 動作", + "執行變體/組合判別、命名證據差分與單位基準擴搜尋", + ), _automation_lane( "evidence_receipts", "證據收據", @@ -2424,6 +2705,17 @@ def build_pchome_growth_ai_automation_readiness( "auto_compare_decision_count": int(decision_summary.get("auto_compare_decision_count") or 0), "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, + "retry_search_action_count": retry_search_action_count, + "variant_bundle_discriminator_count": int( + decision_summary.get("variant_bundle_discriminator_count") or 0 + ), + "named_candidate_evidence_delta_count": int( + decision_summary.get("named_candidate_evidence_delta_count") or 0 + ), + "unit_basis_search_expansion_count": int( + decision_summary.get("unit_basis_search_expansion_count") or 0 + ), "receipt_count": receipt_count, "ready_receipt_count": ready_receipt_count, "exception_count": exception_count, diff --git a/tests/test_pchome_mapping_backlog_report.py b/tests/test_pchome_mapping_backlog_report.py index b059157..4529fe2 100644 --- a/tests/test_pchome_mapping_backlog_report.py +++ b/tests/test_pchome_mapping_backlog_report.py @@ -72,6 +72,7 @@ from services.pchome_mapping_backlog_service import ( build_pchome_evidence_source_preview, 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_growth_ai_automation_readiness, build_pchome_mapping_operator_preview, parse_pchome_product_page_evidence_html, @@ -409,6 +410,58 @@ def test_direct_mapping_candidate_decision_package_routes_candidates_to_machine_ assert package["safety"]["persists_candidate"] is False +def test_direct_mapping_candidate_exception_auto_resolution_builds_machine_artifacts(): + def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score): + return True, "found", [ + { + "product_id": "MOMO-VARIANT", + "name": "Direct mapping product 40ml 多款任選", + "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, + }, + { + "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_auto_resolution_package( + _payload(), + batch_size=1, + execute_search=True, + search_func=fake_search, + ) + + artifacts = package["auto_resolution_package"]["auto_resolution_artifacts"] + assert package["policy"] == "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution" + assert package["result"] == "DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_READY" + assert package["summary"]["machine_review_exception_receipt_count"] == 2 + assert package["summary"]["exception_auto_resolution_artifact_count"] == 2 + assert package["summary"]["variant_bundle_discriminator_count"] == 1 + assert package["summary"]["named_candidate_evidence_delta_count"] == 1 + assert package["summary"]["unit_basis_search_expansion_count"] == 1 + assert package["summary"]["retry_search_action_count"] == 1 + assert artifacts[0]["artifact_id"].startswith("pchome-direct-mapping-exception-resolution-") + assert artifacts[0]["resolvers"]["variant_bundle_discriminator"]["writes_database"] is False + assert artifacts[0]["resolvers"]["named_candidate_evidence_delta"]["resolution"] == "ready_for_retry_search" + assert "unit_basis_search_expansion" in artifacts[1]["resolvers"] + assert any("40ml" in term.lower() for term in artifacts[1]["resolvers"]["unit_basis_search_expansion"]["expanded_search_terms"]) + assert artifacts[1]["guardrails"]["can_auto_execute_read_only"] is True + assert package["summary"]["writes_database_count"] == 0 + assert package["safety"]["writes_database"] is False + assert package["safety"]["persists_candidate"] is False + + def test_ai_automation_readiness_makes_automation_visible_without_manual_primary_flow(): readiness = build_pchome_growth_ai_automation_readiness(_payload(), batch_size=1) @@ -476,6 +529,43 @@ def test_ai_automation_readiness_reports_candidate_decisions_after_controlled_se assert call_count["search"] == 1 +def test_ai_automation_readiness_reports_exception_auto_resolution_ready(): + def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score): + return True, "found", [ + { + "product_id": "MOMO-VARIANT", + "name": "Direct mapping product 40ml 多款任選", + "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, + } + ] + + readiness = build_pchome_growth_ai_automation_readiness( + _payload(), + batch_size=1, + execute_search=True, + search_func=fake_search, + ) + + lanes = {lane["key"]: lane for lane in readiness["automation_lanes"]} + assert readiness["result"] == "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_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"]["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 lanes["candidate_exception_auto_resolution"]["status"] == "ready" + assert lanes["candidate_exception_auto_resolution"]["value"] == 1 + assert readiness["summary"]["primary_human_gate_count"] == 0 + assert readiness["summary"]["writes_database_count"] == 0 + assert readiness["safety"]["writes_database"] is False + + def test_unit_package_basis_parser_extracts_quantity_count_and_risk_signals(): single = parse_unit_package_basis("雅詩蘭黛 粉持久完美持妝粉底 30ml") assert single["package_basis"] == "single_unit_quantity_candidate" @@ -14604,6 +14694,36 @@ def test_direct_mapping_candidate_decision_package_route_defaults_to_no_search_a assert payload["safety"]["writes_database"] is False +def test_direct_mapping_candidate_exception_auto_resolution_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 auto-resolution 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-auto-resolution-package?batch_size=1" + ): + response = routes.api_pchome_growth_direct_mapping_candidate_exception_auto_resolution_package.__wrapped__() + + payload = response.get_json() + assert payload["success"] is True + assert payload["policy"] == "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution" + assert payload["source_endpoint"] == ( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-package" + ) + assert payload["result"] == "WAITING_FOR_DIRECT_MAPPING_CANDIDATES" + assert payload["summary"]["exception_auto_resolution_artifact_count"] == 0 + assert payload["auto_resolution_package"]["resolution_mode"] == "ai_controlled_read_only" + assert payload["safety"]["executes_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