diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index 1fd0bcb..4e44554 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -93,6 +93,7 @@ - 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。 - 2026-07-02 起 AI automation scheduled health summary 必須提供 machine-readable endpoint；`/api/ai-automation/scheduled-health-summary` 會只讀 smoke history，並可選擇 `include_current_smoke=1` 執行不寫 history 的 current smoke，收斂 AI smoke、PChome drift monitor、history freshness、daily summary delivery readiness 四個 family，輸出 `primary_human_gate_count=0`、`writes_database_count=0`、`next_machine_actions` 與 scheduled output endpoints。此 endpoint 不寄 Telegram、不寫 DB、不改排程，只提供排程/監控可消費的健康摘要。 +- 2026-07-02 起 PChome controlled apply rollback evidence 必須提供聚合 endpoint；`/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-rollback-evidence-package` 會聚合 receipt replay、drift verifier、drift recovery、compact readback、artifact retention 五類 evidence，輸出 rollback required / ready actions / protected chain / next machine action。此 endpoint 不執行 rollback、不執行 re-apply、不執行 SQL、不寫 DB；0 drift 時必須輸出 no-op evidence，drift detected 時才輸出 check-mode reapply action。 - 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 4784bd9..0921cf1 100644 --- a/docs/guides/pchome_ai_automation_priority_backlog.md +++ b/docs/guides/pchome_ai_automation_priority_backlog.md @@ -222,6 +222,7 @@ | 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 | 已完成 | direct mapping candidate decision lane closeout route + focused tests | P3.2 scheduled automation health summaries | | P3.2 | Scheduled automation health summaries | 已完成 | `/api/ai-automation/scheduled-health-summary` + smoke service focused tests | P3.3 rollback evidence packages | +| P3.3 | Rollback evidence packages | 已完成 | controlled apply rollback evidence route + focused tests | P3.4 observability metrics integration | ## 後續回報格式 diff --git a/routes/ai_routes.py b/routes/ai_routes.py index cacf74e..5178ba0 100644 --- a/routes/ai_routes.py +++ b/routes/ai_routes.py @@ -2709,6 +2709,40 @@ def api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_ }), 500 +@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-rollback-evidence-package') +@login_required +def api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package(): + """P3 read-only rollback evidence package for controlled-apply families.""" + try: + from config import DATABASE_PATH + from services.pchome_mapping_backlog_service import ( + build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package, + ) + + run_id = str(request.args.get('run_id') or '').strip() or None + materialize_artifacts = str(request.args.get('materialize_artifacts') or '').strip().lower() in {'1', 'true', 'yes'} + + engine = _create_icaim_dashboard_engine(DATABASE_PATH) + try: + package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package( + run_id=run_id, + materialize_artifacts=materialize_artifacts, + engine=engine, + ) + finally: + engine.dispose() + package["source_endpoint"] = ( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-artifact-retention-package" + ) + return jsonify(package) + except Exception as exc: + logger.error("[PChomeGrowth] direct mapping retry candidate exception controlled apply rollback evidence 讀取失敗: %s", exc, exc_info=True) + return jsonify({ + "success": False, + "error": "PChome 商品對應 retry 例外 controlled apply rollback evidence 暫時無法讀取，請稍後再試。", + }), 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 776a6f9..c72609e 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -89,6 +89,9 @@ DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_POLIC DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY = ( "read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention" ) +DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY = ( + "read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence" +) 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" @@ -5857,6 +5860,334 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_artif } +def _retry_exception_controlled_apply_rollback_evidence_id( + summary: dict[str, Any], + rollback_actions: list[dict[str, Any]], +) -> str: + payload = {"summary": summary, "rollback_actions": rollback_actions} + digest = hashlib.sha256( + json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + ).hexdigest()[:16] + return f"pchome-retry-exception-controlled-apply-rollback-evidence-{digest}" + + +def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package( + *, + artifact_root: str | Path | None = None, + run_id: str | None = None, + engine: Any = None, + source_receipt_replay: dict[str, Any] | None = None, + source_drift_verifier: dict[str, Any] | None = None, + source_drift_recovery: dict[str, Any] | None = None, + source_compact_readback: dict[str, Any] | None = None, + source_artifact_retention: dict[str, Any] | None = None, + materialize_artifacts: bool = False, +) -> dict[str, Any]: + """Aggregate rollback evidence for the controlled-apply family without executing rollback.""" + root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" + replay = source_receipt_replay or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package( + artifact_root=root, + run_id=run_id, + materialize_artifacts=False, + engine=engine, + ) + drift = source_drift_verifier or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package( + artifact_root=root, + run_id=run_id, + engine=engine, + source_receipt_replay=replay, + materialize_artifacts=False, + ) + recovery = source_drift_recovery or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_recovery_package( + artifact_root=root, + run_id=run_id, + engine=engine, + source_receipt_replay=replay, + source_drift_verifier=drift, + materialize_artifacts=False, + ) + compact = source_compact_readback or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_compact_readback_package( + artifact_root=root, + run_id=run_id, + engine=engine, + source_receipt_replay=replay, + source_drift_verifier=drift, + source_drift_recovery=recovery, + materialize_artifacts=False, + ) + retention = source_artifact_retention or build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_artifact_retention_package( + artifact_root=root, + run_id=run_id, + engine=engine, + source_compact_readback=compact, + materialize_artifacts=False, + ) + + replay_summary = replay.get("summary") or {} + drift_summary = drift.get("summary") or {} + recovery_summary = recovery.get("summary") or {} + compact_summary = compact.get("summary") or {} + retention_summary = retention.get("summary") or {} + drift_count = int(drift_summary.get("drift_count") or recovery_summary.get("drift_count") or compact_summary.get("drift_count") or 0) + recovery_actions = list(recovery.get("recovery_actions") or []) + ready_actions = [ + item for item in recovery_actions + if item.get("status") == "ready_for_controlled_reapply" + ] + rollback_actions = [] + for action in recovery_actions: + rollback_actions.append({ + "action_id": action.get("action_id"), + "selector_id": action.get("selector_id"), + "momo_icode": action.get("momo_icode"), + "expected_pchome_id": action.get("expected_pchome_id"), + "actual_pchome_id": action.get("actual_pchome_id"), + "status": action.get("status"), + "rollback_sql_shape": action.get("rollback_sql_shape"), + "controlled_reapply_sql_shape": action.get("controlled_reapply_sql_shape"), + "selector_bindings": action.get("selector_bindings") or {}, + "acceptance_gates": list(action.get("acceptance_gates") or []), + "executes_in_package": False, + "writes_database": False, + }) + + rollback_required = drift_count > 0 + rollback_ready = rollback_required and bool(ready_actions) and len(ready_actions) == drift_count + no_rollback_required = ( + not rollback_required + and ( + drift.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" + or recovery.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_NOT_REQUIRED" + or compact.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_VERIFIED" + ) + ) + missing_artifacts = list(replay.get("missing_artifacts") or []) + retention_ready = bool(retention_summary.get("artifact_count") or retention.get("success")) + family_evidence = [ + { + "family": "controlled_apply_receipt_replay", + "result": replay.get("result"), + "selector_count": int(replay_summary.get("target_selector_count") or 0), + "readback_pass_count": int(replay_summary.get("post_apply_readback_pass_count") or 0), + "hash_match_count": int(replay_summary.get("executor_receipt_hash_match_count") or 0), + "rollback_role": "baseline_expected_state", + "writes_database": False, + }, + { + "family": "controlled_apply_drift_verifier", + "result": drift.get("result"), + "drift_count": drift_count, + "hash_match_count": int(drift_summary.get("drift_verifier_artifact_hash_match_count") or 0), + "rollback_role": "current_state_delta_detector", + "writes_database": False, + }, + { + "family": "controlled_apply_drift_recovery", + "result": recovery.get("result"), + "action_count": len(recovery_actions), + "ready_action_count": len(ready_actions), + "hash_match_count": int(recovery_summary.get("recovery_artifact_hash_match_count") or 0), + "rollback_role": "rollback_and_reapply_action_source", + "writes_database": False, + }, + { + "family": "controlled_apply_compact_readback", + "result": compact.get("result"), + "product_status": (compact.get("compact_readback") or {}).get("status"), + "next_machine_action": (compact.get("compact_readback") or {}).get("next_machine_action"), + "hash_match_count": int(compact_summary.get("compact_readback_artifact_hash_match_count") or 0), + "rollback_role": "product_facing_status_source", + "writes_database": False, + }, + { + "family": "controlled_apply_artifact_retention", + "result": retention.get("result"), + "artifact_count": int(retention_summary.get("artifact_count") or 0), + "protected_active_chain_count": int(retention_summary.get("protected_active_chain_count") or 0), + "hash_match_count": int(retention_summary.get("retention_artifact_hash_match_count") or 0), + "rollback_role": "evidence_chain_protection_source", + "writes_database": False, + }, + ] + + if rollback_ready: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_READY" + product_status = "rollback_ready" + next_machine_action = "run_controlled_reapply_check_mode" + elif rollback_required: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_BLOCKED" + product_status = "rollback_blocked" + next_machine_action = "rebuild_selector_identity_before_rollback" + elif no_rollback_required: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED" + product_status = "rollback_not_required" + next_machine_action = "keep_monitoring_drift" + elif missing_artifacts: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_ARTIFACTS" + product_status = "waiting" + next_machine_action = "restore_or_materialize_source_receipts" + else: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_BASELINE" + product_status = "waiting" + next_machine_action = "run_receipt_replay_and_drift_verifier" + + summary = { + "controlled_apply_family_count": len(family_evidence), + "target_selector_count": int(replay_summary.get("target_selector_count") or compact_summary.get("target_selector_count") or 0), + "post_apply_readback_pass_count": int(replay_summary.get("post_apply_readback_pass_count") or compact_summary.get("post_apply_readback_pass_count") or 0), + "drift_count": drift_count, + "rollback_required_count": 1 if rollback_required else 0, + "rollback_action_count": len(rollback_actions), + "rollback_ready_action_count": len(ready_actions), + "rollback_blocked_action_count": len(rollback_actions) - len(ready_actions), + "retention_artifact_count": int(retention_summary.get("artifact_count") or 0), + "protected_active_chain_count": int(retention_summary.get("protected_active_chain_count") or 0), + "rollback_evidence_ready_count": 1 if (rollback_ready or no_rollback_required) else 0, + "rollback_evidence_artifact_materialized_count": 0, + "rollback_evidence_artifact_hash_match_count": 0, + "primary_human_gate_count": 0, + "writes_database_count": 0, + } + rollback_evidence_id = _retry_exception_controlled_apply_rollback_evidence_id(summary, rollback_actions) + safety = { + "ai_controlled_apply": True, + "rollback_evidence": True, + "reads_artifact_files": True, + "reads_database": engine is not None, + "executes_rollback": False, + "executes_reapply": False, + "executes_sql": False, + "deletes_artifacts": False, + "writes_database": False, + "writes_database_count": 0, + "writes_artifact_count": 0, + "syncs_external_offers": False, + "dispatches_telegram": False, + "gemini_allowed": False, + "requires_production_version_truth": True, + } + checks = [ + {"check": "receipt_replay_loaded", "passed": bool(replay)}, + {"check": "drift_verifier_loaded", "passed": bool(drift)}, + {"check": "drift_recovery_loaded", "passed": bool(recovery)}, + {"check": "compact_readback_loaded", "passed": bool(compact)}, + {"check": "artifact_retention_loaded", "passed": bool(retention)}, + {"check": "rollback_actions_cover_detected_drift", "passed": (not rollback_required) or len(ready_actions) == drift_count}, + {"check": "artifact_retention_protects_evidence_chain", "passed": retention_ready or not rollback_required}, + {"check": "rollback_evidence_has_no_primary_human_gate", "passed": True}, + {"check": "rollback_evidence_does_not_execute_sql", "passed": True}, + {"check": "rollback_evidence_does_not_write_database", "passed": True}, + ] + artifact_payload = { + "artifact_key": "retry_exception_controlled_apply_rollback_evidence_receipt", + "rollback_evidence_id": rollback_evidence_id, + "source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY, + "source_compact_readback_result": compact.get("result"), + "source_drift_recovery_result": recovery.get("result"), + "source_artifact_retention_result": retention.get("result"), + "result": result, + "product_status": product_status, + "next_machine_action": next_machine_action, + "summary": summary, + "family_evidence": family_evidence, + "rollback_actions": rollback_actions, + "checks": checks, + "safety": safety, + } + artifact_bytes = _canonical_retry_exception_artifact_bytes(artifact_payload) + artifact_relative_path = ( + f"artifacts/pchome_growth/retry_exception_closeout/" + f"controlled_apply_rollback_evidence/{rollback_evidence_id}.json" + ) + rollback_evidence_artifact = { + "key": "retry_exception_controlled_apply_rollback_evidence_receipt", + "artifact_type": "controlled_apply_rollback_evidence_receipt", + "relative_path": artifact_relative_path, + "payload_sha256": hashlib.sha256(artifact_bytes).hexdigest(), + "byte_count": len(artifact_bytes), + "payload": artifact_payload, + "materialized": False, + "writes_database": False, + } + materialized_rollback_evidence_artifacts: list[dict[str, Any]] = [] + if materialize_artifacts and (rollback_ready or no_rollback_required): + target_path = _resolve_retry_exception_artifact_path(root, artifact_relative_path) + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_bytes(artifact_bytes) + materialized_rollback_evidence_artifacts.append({ + "key": rollback_evidence_artifact["key"], + "relative_path": artifact_relative_path, + "absolute_path": str(target_path), + "payload_sha256": rollback_evidence_artifact["payload_sha256"], + "written_byte_count": target_path.stat().st_size, + "writes_database": False, + }) + rollback_evidence_artifact["materialized"] = True + rollback_evidence_artifact["absolute_path"] = str(target_path) + artifact_path = _resolve_retry_exception_artifact_path(root, artifact_relative_path) + artifact_sha = hashlib.sha256(artifact_path.read_bytes()).hexdigest() if artifact_path.exists() else "" + artifact_hash_match = bool(artifact_sha) and artifact_sha == rollback_evidence_artifact["payload_sha256"] + summary["rollback_evidence_artifact_materialized_count"] = ( + len(materialized_rollback_evidence_artifacts) or (1 if artifact_hash_match else 0) + ) + summary["rollback_evidence_artifact_hash_match_count"] = 1 if artifact_hash_match else 0 + safety["writes_artifact_count"] = len(materialized_rollback_evidence_artifacts) + checks.extend([ + { + "check": "rollback_evidence_artifact_materialized_when_requested", + "passed": (not materialize_artifacts) or ((rollback_ready or no_rollback_required) and artifact_path.exists()), + }, + { + "check": "rollback_evidence_artifact_hash_matches_expected", + "passed": (not materialize_artifacts) or artifact_hash_match, + }, + ]) + return { + "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLICY, + "result": result, + "success": result in { + "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_READY", + "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED", + }, + "summary": summary, + "rollback_evidence": { + "rollback_evidence_id": rollback_evidence_id, + "stage": "P3_retry_exception_controlled_apply_rollback_evidence", + "status": product_status, + "rollback_required": rollback_required, + "next_machine_action": next_machine_action, + "materialize_artifacts": bool(materialize_artifacts), + "requires_production_version_truth": True, + }, + "family_evidence": family_evidence, + "rollback_actions": rollback_actions, + "rollback_evidence_artifact": rollback_evidence_artifact, + "materialized_rollback_evidence_artifacts": materialized_rollback_evidence_artifacts, + "post_rollback_evidence_artifact_verifier": { + "expected_sha256": rollback_evidence_artifact["payload_sha256"], + "actual_sha256": artifact_sha, + "hash_match": artifact_hash_match, + "writes_database": False, + }, + "source_results": { + "receipt_replay": replay.get("result"), + "drift_verifier": drift.get("result"), + "drift_recovery": recovery.get("result"), + "compact_readback": compact.get("result"), + "artifact_retention": retention.get("result"), + }, + "checks": checks, + "check_count": len(checks), + "all_checks_passed": all(check.get("passed") is True for check in checks), + "next_actions": [ + "If rollback_required is false, keep monitoring drift using the scheduled health summary.", + "If rollback evidence is ready, run the controlled re-apply path in check-mode before any write.", + "After any controlled re-apply, replay receipts and regenerate this rollback evidence package.", + ], + "safety": safety, + } + + 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) diff --git a/tests/test_pchome_controlled_apply_rollback_evidence.py b/tests/test_pchome_controlled_apply_rollback_evidence.py new file mode 100644 index 0000000..611b91c --- /dev/null +++ b/tests/test_pchome_controlled_apply_rollback_evidence.py @@ -0,0 +1,233 @@ +from flask import Flask + +from services.pchome_mapping_backlog_service import ( + build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package, +) + + +def _replay_package(): + return { + "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED", + "summary": { + "target_selector_count": 2, + "post_apply_readback_pass_count": 2, + "executor_receipt_hash_match_count": 1, + "missing_artifact_count": 0, + "writes_database_count": 0, + }, + "missing_artifacts": [], + "receipt_replay": {"run_id": "run-test"}, + } + + +def _drift_package(drift_count=0): + drift_items = [] + if drift_count: + drift_items = [ + { + "selector_id": "receipt-1", + "momo_icode": "MOMO-1", + "expected_pchome_id": "PCH-EXPECTED", + "actual_pchome_id": "PCH-DRIFT", + "expected_momo_name": "Expected product", + "actual_momo_name": "Expected product", + } + ] + return { + "result": ( + "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED" + if drift_count + else "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" + ), + "summary": { + "target_selector_count": 2, + "post_apply_readback_pass_count": 2 - drift_count, + "drift_count": drift_count, + "drift_verified_count": 0 if drift_count else 1, + "drift_verifier_artifact_hash_match_count": 1, + "writes_database_count": 0, + }, + "drift_items": drift_items, + } + + +def _recovery_package(drift_count=0): + actions = [] + if drift_count: + actions = [ + { + "action_id": "rollback-action-1", + "selector_id": "receipt-1", + "momo_icode": "MOMO-1", + "expected_pchome_id": "PCH-EXPECTED", + "actual_pchome_id": "PCH-DRIFT", + "status": "ready_for_controlled_reapply", + "rollback_sql_shape": "UPDATE pchome_product_matches SET pchome_id = :actual_pchome_id", + "controlled_reapply_sql_shape": "UPDATE pchome_product_matches SET pchome_id = :expected_pchome_id", + "selector_bindings": { + "momo_icode": "MOMO-1", + "expected_pchome_id": "PCH-EXPECTED", + "actual_pchome_id": "PCH-DRIFT", + }, + "acceptance_gates": ["post_reapply_drift_verifier_returns_zero_drift"], + "writes_database": False, + } + ] + return { + "result": ( + "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_PACKAGE_READY" + if drift_count + else "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_RECOVERY_NOT_REQUIRED" + ), + "summary": { + "drift_count": drift_count, + "drift_recovery_action_count": len(actions), + "drift_reapply_ready_count": len(actions), + "recovery_artifact_hash_match_count": 1, + "writes_database_count": 0, + }, + "recovery_actions": actions, + "rollback_plan": { + "rollback_action_count": len(actions), + "executes_in_package": False, + "writes_database": False, + }, + } + + +def _compact_package(drift_count=0): + return { + "result": ( + "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_DRIFT_REQUIRES_RECOVERY" + if drift_count + else "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_COMPACT_READBACK_VERIFIED" + ), + "summary": { + "target_selector_count": 2, + "post_apply_readback_pass_count": 2 - drift_count, + "drift_count": drift_count, + "drift_recovery_action_count": drift_count, + "compact_readback_artifact_hash_match_count": 1, + "writes_database_count": 0, + }, + "compact_readback": { + "status": "blocked" if drift_count else "completed", + "next_machine_action": ( + "run_controlled_reapply_check_mode" + if drift_count + else "keep_monitoring_drift" + ), + }, + } + + +def _retention_package(): + return { + "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ARTIFACT_RETENTION_POLICY_READY", + "success": True, + "summary": { + "artifact_count": 8, + "protected_active_chain_count": 4, + "retention_artifact_hash_match_count": 1, + "writes_database_count": 0, + }, + } + + +def test_rollback_evidence_no_drift_builds_noop_machine_evidence(): + package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package( + source_receipt_replay=_replay_package(), + source_drift_verifier=_drift_package(), + source_drift_recovery=_recovery_package(), + source_compact_readback=_compact_package(), + source_artifact_retention=_retention_package(), + ) + + assert package["policy"] == ( + "read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence" + ) + assert package["result"] == ( + "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED" + ) + assert package["summary"]["controlled_apply_family_count"] == 5 + assert package["summary"]["drift_count"] == 0 + assert package["summary"]["rollback_required_count"] == 0 + assert package["summary"]["rollback_evidence_ready_count"] == 1 + assert package["summary"]["primary_human_gate_count"] == 0 + assert package["rollback_evidence"]["status"] == "rollback_not_required" + assert package["rollback_actions"] == [] + assert package["safety"]["executes_sql"] is False + assert package["safety"]["writes_database"] is False + assert package["all_checks_passed"] is True + + +def test_rollback_evidence_drift_ready_outputs_actions_without_db_write(): + package = build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package( + source_receipt_replay=_replay_package(), + source_drift_verifier=_drift_package(drift_count=1), + source_drift_recovery=_recovery_package(drift_count=1), + source_compact_readback=_compact_package(drift_count=1), + source_artifact_retention=_retention_package(), + ) + + assert package["result"] == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_READY" + assert package["summary"]["drift_count"] == 1 + assert package["summary"]["rollback_required_count"] == 1 + assert package["summary"]["rollback_action_count"] == 1 + assert package["summary"]["rollback_ready_action_count"] == 1 + assert package["rollback_evidence"]["status"] == "rollback_ready" + assert package["rollback_evidence"]["next_machine_action"] == "run_controlled_reapply_check_mode" + assert package["rollback_actions"][0]["momo_icode"] == "MOMO-1" + assert "UPDATE pchome_product_matches" in package["rollback_actions"][0]["rollback_sql_shape"] + assert package["rollback_actions"][0]["executes_in_package"] is False + assert package["rollback_actions"][0]["writes_database"] is False + assert package["safety"]["executes_rollback"] is False + assert package["safety"]["executes_reapply"] is False + assert package["safety"]["writes_database"] is False + assert package["all_checks_passed"] is True + + +def test_rollback_evidence_route_uses_engine_and_returns_source_endpoint(monkeypatch): + from routes import ai_routes as routes + + class FakeEngine: + disposed = False + + def dispose(self): + self.disposed = True + + fake_engine = FakeEngine() + monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", lambda _path: fake_engine) + + def fake_builder(**kwargs): + assert kwargs["engine"] is fake_engine + return { + "success": True, + "policy": "read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence", + "result": "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_NOT_REQUIRED", + "summary": {"writes_database_count": 0, "primary_human_gate_count": 0}, + "safety": {"writes_database": False}, + } + + monkeypatch.setattr( + "services.pchome_mapping_backlog_service.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package", + fake_builder, + ) + + app = Flask(__name__) + with app.test_request_context( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-rollback-evidence-package" + ): + response = ( + routes + .api_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence_package + .__wrapped__() + ) + + payload = response.get_json() + assert fake_engine.disposed is True + assert payload["source_endpoint"] == ( + "/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-exception-controlled-apply-artifact-retention-package" + ) + assert payload["summary"]["primary_human_gate_count"] == 0 + assert payload["safety"]["writes_database"] is False