From 2e8ed32dad5603ba9e9b02c3d22b1ddec7d1973f Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 13:33:53 +0800 Subject: [PATCH] refactor(pchome): extract controlled receipt replay --- config.py | 2 +- docs/AI_INTELLIGENCE_MODULE_SOT.md | 2 +- .../code_modularization_inventory_20260430.md | 2 +- .../controlled_apply_receipt.py | 294 ++++++++++++++++++ services/pchome_mapping_backlog_service.py | 271 +--------------- ...t_pchome_mapping_backlog_modularization.py | 30 +- 6 files changed, 331 insertions(+), 270 deletions(-) create mode 100644 services/pchome_mapping_backlog/controlled_apply_receipt.py diff --git a/config.py b/config.py index ec69a6f..9270bbe 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.782" +SYSTEM_VERSION = "V10.783" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index cfd23b4..d0cbc18 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -2,7 +2,7 @@ > **最後更新**: 2026-07-11 (台北時間) > **狀態**: 🟠 Partial。四 Agent、PixelRAG、MCP/RAG registry、PChome controlled-preview families 與多項 smoke/readback 已建立；但 access control、database identity/RBAC、full asset reconciliation、software supply chain、same-run controlled apply、restore drill、internal RAG canary 與 MCP/RAG runtime closure 尚未全部完成。任何局部 receipt 不得代表整體閉環完成。 -> **適用版本**: V10.782 +> **適用版本**: V10.783 --- diff --git a/docs/memory/code_modularization_inventory_20260430.md b/docs/memory/code_modularization_inventory_20260430.md index b2ae6f2..d7f8922 100644 --- a/docs/memory/code_modularization_inventory_20260430.md +++ b/docs/memory/code_modularization_inventory_20260430.md @@ -119,7 +119,7 @@ | 行數 | 檔案 | 分類 | 拆分方向 | |---:|---|---|---| -| 43802 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor 與 no-write drift verifier 已移到 `services/pchome_mapping_backlog/`；下一步依序拆 receipt、reporter | +| 43543 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor、drift verifier 與 receipt replay 已移到 `services/pchome_mapping_backlog/`；下一步拆 reporter | | 14289 | `services/ai_automation_smoke_service.py` | P0 smoke 巨型 service | 拆 family registry、collectors、health projection、metrics adapter | | 9383 | `routes/openclaw_bot_routes.py` | P0 巨型 Blueprint、拆分進行中 | scheduler hook 已移到 `services/openclaw_bot/scheduled_jobs.py`；下一步拆 report、command | | 7641 | `routes/ai_routes.py` | P0 巨型 Blueprint | PChome growth、AI automation、recommendation route extension 分離 | diff --git a/services/pchome_mapping_backlog/controlled_apply_receipt.py b/services/pchome_mapping_backlog/controlled_apply_receipt.py new file mode 100644 index 0000000..a518f75 --- /dev/null +++ b/services/pchome_mapping_backlog/controlled_apply_receipt.py @@ -0,0 +1,294 @@ +"""Replayable controlled-apply receipts with bounded DB readback.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +from services.pchome_mapping_backlog.artifacts import ( + _canonical_retry_exception_artifact_bytes, + _resolve_retry_exception_artifact_path, +) +from services.pchome_mapping_backlog.controlled_apply_executor import ( + _fetch_pchome_product_match_by_momo_icode, +) +from services.pchome_mapping_backlog.policies import ( + DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY, +) + + +def _find_retry_exception_artifact_file(root: Path, subdir: str, run_id: str | None = None) -> Path | None: + artifact_dir = root / "artifacts" / "pchome_growth" / "retry_exception_closeout" / subdir + if run_id: + candidate = artifact_dir / f"{run_id}.json" + return candidate if candidate.exists() else None + candidates = sorted( + artifact_dir.glob("*.json"), + key=lambda path: path.stat().st_mtime, + reverse=True, + ) + return candidates[0] if candidates else None + + +def _load_retry_exception_json_artifact(path: Path | None) -> dict[str, Any]: + if path is None or not path.exists(): + return {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return payload if isinstance(payload, dict) else {} + + +def _retry_exception_controlled_apply_receipt_replay_id( + run_id: str, + post_apply_readbacks: list[dict[str, Any]], +) -> str: + payload = { + "run_id": run_id, + "post_apply_readbacks": post_apply_readbacks, + } + 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-receipt-replay-{digest}" + + +def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package( + *, + artifact_root: str | Path | None = None, + run_id: str | None = None, + materialize_artifacts: bool = False, + engine: Any = None, +) -> dict[str, Any]: + """Rebuild a machine-verifiable executor closeout receipt from artifacts and DB readback.""" + root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" + verifier_path = _find_retry_exception_artifact_file(root, "verifier_inputs", run_id) + identity_path = _find_retry_exception_artifact_file(root, "identity_readback", run_id) + preflight_path = _find_retry_exception_artifact_file(root, "controlled_apply_preflight", run_id) + verifier_payload = _load_retry_exception_json_artifact(verifier_path) + identity_payload = _load_retry_exception_json_artifact(identity_path) + preflight_payload = _load_retry_exception_json_artifact(preflight_path) + effective_run_id = ( + run_id + or verifier_payload.get("run_id") + or identity_payload.get("run_id") + or preflight_payload.get("run_id") + or "" + ) + receipts = list(verifier_payload.get("no_write_verifier_receipts") or []) + target_selectors = [ + { + "selector_id": receipt.get("receipt_id"), + "source_receipt_id": receipt.get("receipt_id"), + "source_closeout_receipt_id": receipt.get("source_closeout_receipt_id"), + "momo_product_id": (receipt.get("subject") or {}).get("momo_product_id"), + "momo_product_name": (receipt.get("subject") or {}).get("momo_product_name"), + "momo_price": (receipt.get("subject") or {}).get("momo_price"), + "target_pchome_product_id": (receipt.get("subject") or {}).get("target_pchome_product_id"), + "target_pchome_product_name": (receipt.get("subject") or {}).get("pchome_product_name"), + } + for receipt in receipts + if (receipt.get("subject") or {}).get("momo_product_id") + ] + missing_artifacts = [ + name + for name, payload in ( + ("verifier_inputs", verifier_payload), + ("identity_readback", identity_payload), + ("controlled_apply_preflight", preflight_payload), + ) + if not payload + ] + post_apply_readbacks: list[dict[str, Any]] = [] + if engine is not None and target_selectors: + with engine.connect() as conn: + for selector in target_selectors: + momo_icode = str(selector.get("momo_product_id") or "").strip() + expected_pchome_id = str(selector.get("target_pchome_product_id") or "").strip() + row = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode) + actual_pchome_id = str((row or {}).get("pchome_id") or "").strip() + post_apply_readbacks.append({ + "selector_id": selector.get("selector_id"), + "momo_icode": momo_icode, + "expected_pchome_id": expected_pchome_id, + "actual_pchome_id": actual_pchome_id, + "expected_momo_name": selector.get("momo_product_name"), + "actual_momo_name": (row or {}).get("momo_name"), + "passed": bool(row) and actual_pchome_id == expected_pchome_id, + "writes_database": False, + }) + + readback_pass_count = sum(1 for item in post_apply_readbacks if item.get("passed")) + receipt_ready = ( + not missing_artifacts + and bool(effective_run_id) + and bool(target_selectors) + and engine is not None + and readback_pass_count == len(target_selectors) + ) + if missing_artifacts: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_REPLAY_ARTIFACTS" + elif engine is None: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_BLOCKED_ENGINE_REQUIRED" + elif not target_selectors: + result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_REPLAY_SELECTORS" + elif receipt_ready: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED" + else: + result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_READBACK_MISMATCH" + + receipt_id = _retry_exception_controlled_apply_receipt_replay_id(effective_run_id, post_apply_readbacks) + summary = { + "source_receipt_count": len(receipts), + "target_selector_count": len(target_selectors), + "post_apply_readback_count": len(post_apply_readbacks), + "post_apply_readback_pass_count": readback_pass_count, + "post_apply_readback_fail_count": len(post_apply_readbacks) - readback_pass_count, + "missing_artifact_count": len(missing_artifacts), + "executor_receipt_ready_count": 1 if receipt_ready else 0, + "executor_receipt_materialized_count": 0, + "executor_receipt_hash_match_count": 0, + "writes_database_count": 0, + } + safety = { + "ai_controlled_apply": True, + "receipt_replay": True, + "reads_artifact_files": True, + "reads_database": engine is not None, + "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, + } + rollback_plan = { + "rollback_step_count": 1 if receipt_ready else 0, + "rollback_steps": [ + { + "action": "delete_materialized_executor_replay_receipt", + "receipt_id": receipt_id, + "executes_in_replay": False, + "writes_database": False, + } + ] if receipt_ready else [], + "writes_database": False, + } + receipt_relative_path = ( + f"artifacts/pchome_growth/retry_exception_closeout/" + f"controlled_apply_executor_replay/{receipt_id}.json" + ) + receipt_payload = { + "artifact_key": "retry_exception_controlled_apply_executor_replay_receipt", + "receipt_id": receipt_id, + "run_id": effective_run_id, + "source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY, + "source_artifacts": { + "verifier_inputs": str(verifier_path.relative_to(root)) if verifier_path else None, + "identity_readback": str(identity_path.relative_to(root)) if identity_path else None, + "controlled_apply_preflight": str(preflight_path.relative_to(root)) if preflight_path else None, + }, + "result": result, + "summary": summary, + "identity_readback": { + "momo_product_ids": identity_payload.get("momo_product_ids") or [], + "target_pchome_product_ids": identity_payload.get("target_pchome_product_ids") or [], + "identity_delta_status": identity_payload.get("identity_delta_status"), + }, + "target_selectors": target_selectors, + "post_apply_readbacks": post_apply_readbacks, + "rollback_plan": rollback_plan, + "safety": safety, + } + receipt_bytes = _canonical_retry_exception_artifact_bytes(receipt_payload) + receipt_artifact = { + "key": "retry_exception_controlled_apply_executor_replay_receipt", + "artifact_type": "controlled_apply_executor_replay_receipt", + "relative_path": receipt_relative_path, + "payload_sha256": hashlib.sha256(receipt_bytes).hexdigest(), + "byte_count": len(receipt_bytes), + "payload": receipt_payload, + "materialized": False, + "writes_database": False, + } + materialized_artifacts: list[dict[str, Any]] = [] + if materialize_artifacts and receipt_ready: + target_path = _resolve_retry_exception_artifact_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_artifact["payload_sha256"], + "written_byte_count": target_path.stat().st_size, + "writes_database": False, + }) + receipt_artifact["materialized"] = True + receipt_artifact["absolute_path"] = str(target_path) + + receipt_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path) + actual_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest() if receipt_path.exists() else "" + verifier_checks = [ + {"check": "source_artifacts_loaded", "passed": not missing_artifacts, "missing_artifacts": missing_artifacts}, + {"check": "target_selectors_present", "passed": bool(target_selectors)}, + {"check": "post_apply_readbacks_all_passed", "passed": readback_pass_count == len(target_selectors)}, + { + "check": "receipt_materialized_when_requested", + "passed": (not materialize_artifacts) or (receipt_ready and receipt_path.exists()), + }, + { + "check": "receipt_hash_matches_expected", + "passed": (not materialize_artifacts) or (bool(actual_sha) and actual_sha == receipt_artifact["payload_sha256"]), + }, + {"check": "replay_does_not_write_database", "passed": safety["writes_database"] is False}, + ] + existing_receipt_hash_match = bool(actual_sha) and actual_sha == receipt_artifact["payload_sha256"] + summary["executor_receipt_materialized_count"] = len(materialized_artifacts) or (1 if existing_receipt_hash_match else 0) + summary["executor_receipt_hash_match_count"] = 1 if existing_receipt_hash_match else 0 + summary["post_executor_receipt_verifier_check_count"] = len(verifier_checks) + safety["writes_artifact_count"] = len(materialized_artifacts) + return { + "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY, + "result": result, + "success": not missing_artifacts, + "summary": summary, + "receipt_replay": { + "receipt_id": receipt_id, + "run_id": effective_run_id, + "stage": "P2_retry_exception_controlled_apply_receipt_replay", + "status": result, + "materialize_artifacts": bool(materialize_artifacts), + "artifact_root": str(root), + "ready": receipt_ready, + }, + "target_selectors": target_selectors, + "post_apply_readbacks": post_apply_readbacks, + "executor_receipt_artifact": receipt_artifact, + "materialized_executor_artifacts": materialized_artifacts, + "post_executor_receipt_verifier": { + "checks": verifier_checks, + "check_count": len(verifier_checks), + "passed": all(check.get("passed") is True for check in verifier_checks), + "expected_sha256": receipt_artifact["payload_sha256"], + "actual_sha256": actual_sha, + "hash_match": bool(actual_sha) and actual_sha == receipt_artifact["payload_sha256"], + "writes_database": False, + }, + "rollback_plan": rollback_plan, + "source_artifacts": receipt_payload["source_artifacts"], + "missing_artifacts": missing_artifacts, + "safety": safety, + } + + +__all__ = ( + "_find_retry_exception_artifact_file", + "_load_retry_exception_json_artifact", + "_retry_exception_controlled_apply_receipt_replay_id", + "build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package", +) diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index 7ae267a..b80dc4a 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -142,6 +142,12 @@ from services.pchome_mapping_backlog.controlled_apply_executor import ( _upsert_pchome_product_match, build_controlled_apply_executor_from_preflight, ) +from services.pchome_mapping_backlog.controlled_apply_receipt import ( + _find_retry_exception_artifact_file, + _load_retry_exception_json_artifact, + _retry_exception_controlled_apply_receipt_replay_id, + build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package, +) from services.pchome_mapping_backlog.controlled_apply_verifier import ( build_controlled_apply_drift_verifier_from_replay, ) @@ -3599,271 +3605,6 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_execu -def _find_retry_exception_artifact_file(root: Path, subdir: str, run_id: str | None = None) -> Path | None: - artifact_dir = root / "artifacts" / "pchome_growth" / "retry_exception_closeout" / subdir - if run_id: - candidate = artifact_dir / f"{run_id}.json" - return candidate if candidate.exists() else None - candidates = sorted( - artifact_dir.glob("*.json"), - key=lambda path: path.stat().st_mtime, - reverse=True, - ) - return candidates[0] if candidates else None - - -def _load_retry_exception_json_artifact(path: Path | None) -> dict[str, Any]: - if path is None or not path.exists(): - return {} - try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {} - return payload if isinstance(payload, dict) else {} - - -def _retry_exception_controlled_apply_receipt_replay_id( - run_id: str, - post_apply_readbacks: list[dict[str, Any]], -) -> str: - payload = { - "run_id": run_id, - "post_apply_readbacks": post_apply_readbacks, - } - 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-receipt-replay-{digest}" - - -def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package( - *, - artifact_root: str | Path | None = None, - run_id: str | None = None, - materialize_artifacts: bool = False, - engine: Any = None, -) -> dict[str, Any]: - """Rebuild a machine-verifiable executor closeout receipt from artifacts and DB readback.""" - root = Path(artifact_root) if artifact_root is not None else Path.cwd() / "data" - verifier_path = _find_retry_exception_artifact_file(root, "verifier_inputs", run_id) - identity_path = _find_retry_exception_artifact_file(root, "identity_readback", run_id) - preflight_path = _find_retry_exception_artifact_file(root, "controlled_apply_preflight", run_id) - verifier_payload = _load_retry_exception_json_artifact(verifier_path) - identity_payload = _load_retry_exception_json_artifact(identity_path) - preflight_payload = _load_retry_exception_json_artifact(preflight_path) - effective_run_id = ( - run_id - or verifier_payload.get("run_id") - or identity_payload.get("run_id") - or preflight_payload.get("run_id") - or "" - ) - receipts = list(verifier_payload.get("no_write_verifier_receipts") or []) - target_selectors = [ - { - "selector_id": receipt.get("receipt_id"), - "source_receipt_id": receipt.get("receipt_id"), - "source_closeout_receipt_id": receipt.get("source_closeout_receipt_id"), - "momo_product_id": (receipt.get("subject") or {}).get("momo_product_id"), - "momo_product_name": (receipt.get("subject") or {}).get("momo_product_name"), - "momo_price": (receipt.get("subject") or {}).get("momo_price"), - "target_pchome_product_id": (receipt.get("subject") or {}).get("target_pchome_product_id"), - "target_pchome_product_name": (receipt.get("subject") or {}).get("pchome_product_name"), - } - for receipt in receipts - if (receipt.get("subject") or {}).get("momo_product_id") - ] - missing_artifacts = [ - name - for name, payload in ( - ("verifier_inputs", verifier_payload), - ("identity_readback", identity_payload), - ("controlled_apply_preflight", preflight_payload), - ) - if not payload - ] - post_apply_readbacks: list[dict[str, Any]] = [] - if engine is not None and target_selectors: - with engine.connect() as conn: - for selector in target_selectors: - momo_icode = str(selector.get("momo_product_id") or "").strip() - expected_pchome_id = str(selector.get("target_pchome_product_id") or "").strip() - row = _fetch_pchome_product_match_by_momo_icode(conn, momo_icode) - actual_pchome_id = str((row or {}).get("pchome_id") or "").strip() - post_apply_readbacks.append({ - "selector_id": selector.get("selector_id"), - "momo_icode": momo_icode, - "expected_pchome_id": expected_pchome_id, - "actual_pchome_id": actual_pchome_id, - "expected_momo_name": selector.get("momo_product_name"), - "actual_momo_name": (row or {}).get("momo_name"), - "passed": bool(row) and actual_pchome_id == expected_pchome_id, - "writes_database": False, - }) - - readback_pass_count = sum(1 for item in post_apply_readbacks if item.get("passed")) - receipt_ready = ( - not missing_artifacts - and bool(effective_run_id) - and bool(target_selectors) - and engine is not None - and readback_pass_count == len(target_selectors) - ) - if missing_artifacts: - result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_REPLAY_ARTIFACTS" - elif engine is None: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_BLOCKED_ENGINE_REQUIRED" - elif not target_selectors: - result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_REPLAY_SELECTORS" - elif receipt_ready: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED" - else: - result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_READBACK_MISMATCH" - - receipt_id = _retry_exception_controlled_apply_receipt_replay_id(effective_run_id, post_apply_readbacks) - summary = { - "source_receipt_count": len(receipts), - "target_selector_count": len(target_selectors), - "post_apply_readback_count": len(post_apply_readbacks), - "post_apply_readback_pass_count": readback_pass_count, - "post_apply_readback_fail_count": len(post_apply_readbacks) - readback_pass_count, - "missing_artifact_count": len(missing_artifacts), - "executor_receipt_ready_count": 1 if receipt_ready else 0, - "executor_receipt_materialized_count": 0, - "executor_receipt_hash_match_count": 0, - "writes_database_count": 0, - } - safety = { - "ai_controlled_apply": True, - "receipt_replay": True, - "reads_artifact_files": True, - "reads_database": engine is not None, - "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, - } - rollback_plan = { - "rollback_step_count": 1 if receipt_ready else 0, - "rollback_steps": [ - { - "action": "delete_materialized_executor_replay_receipt", - "receipt_id": receipt_id, - "executes_in_replay": False, - "writes_database": False, - } - ] if receipt_ready else [], - "writes_database": False, - } - receipt_relative_path = ( - f"artifacts/pchome_growth/retry_exception_closeout/" - f"controlled_apply_executor_replay/{receipt_id}.json" - ) - receipt_payload = { - "artifact_key": "retry_exception_controlled_apply_executor_replay_receipt", - "receipt_id": receipt_id, - "run_id": effective_run_id, - "source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY, - "source_artifacts": { - "verifier_inputs": str(verifier_path.relative_to(root)) if verifier_path else None, - "identity_readback": str(identity_path.relative_to(root)) if identity_path else None, - "controlled_apply_preflight": str(preflight_path.relative_to(root)) if preflight_path else None, - }, - "result": result, - "summary": summary, - "identity_readback": { - "momo_product_ids": identity_payload.get("momo_product_ids") or [], - "target_pchome_product_ids": identity_payload.get("target_pchome_product_ids") or [], - "identity_delta_status": identity_payload.get("identity_delta_status"), - }, - "target_selectors": target_selectors, - "post_apply_readbacks": post_apply_readbacks, - "rollback_plan": rollback_plan, - "safety": safety, - } - receipt_bytes = _canonical_retry_exception_artifact_bytes(receipt_payload) - receipt_artifact = { - "key": "retry_exception_controlled_apply_executor_replay_receipt", - "artifact_type": "controlled_apply_executor_replay_receipt", - "relative_path": receipt_relative_path, - "payload_sha256": hashlib.sha256(receipt_bytes).hexdigest(), - "byte_count": len(receipt_bytes), - "payload": receipt_payload, - "materialized": False, - "writes_database": False, - } - materialized_artifacts: list[dict[str, Any]] = [] - if materialize_artifacts and receipt_ready: - target_path = _resolve_retry_exception_artifact_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_artifact["payload_sha256"], - "written_byte_count": target_path.stat().st_size, - "writes_database": False, - }) - receipt_artifact["materialized"] = True - receipt_artifact["absolute_path"] = str(target_path) - - receipt_path = _resolve_retry_exception_artifact_path(root, receipt_relative_path) - actual_sha = hashlib.sha256(receipt_path.read_bytes()).hexdigest() if receipt_path.exists() else "" - verifier_checks = [ - {"check": "source_artifacts_loaded", "passed": not missing_artifacts, "missing_artifacts": missing_artifacts}, - {"check": "target_selectors_present", "passed": bool(target_selectors)}, - {"check": "post_apply_readbacks_all_passed", "passed": readback_pass_count == len(target_selectors)}, - { - "check": "receipt_materialized_when_requested", - "passed": (not materialize_artifacts) or (receipt_ready and receipt_path.exists()), - }, - { - "check": "receipt_hash_matches_expected", - "passed": (not materialize_artifacts) or (bool(actual_sha) and actual_sha == receipt_artifact["payload_sha256"]), - }, - {"check": "replay_does_not_write_database", "passed": safety["writes_database"] is False}, - ] - existing_receipt_hash_match = bool(actual_sha) and actual_sha == receipt_artifact["payload_sha256"] - summary["executor_receipt_materialized_count"] = len(materialized_artifacts) or (1 if existing_receipt_hash_match else 0) - summary["executor_receipt_hash_match_count"] = 1 if existing_receipt_hash_match else 0 - summary["post_executor_receipt_verifier_check_count"] = len(verifier_checks) - safety["writes_artifact_count"] = len(materialized_artifacts) - return { - "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAY_POLICY, - "result": result, - "success": not missing_artifacts, - "summary": summary, - "receipt_replay": { - "receipt_id": receipt_id, - "run_id": effective_run_id, - "stage": "P2_retry_exception_controlled_apply_receipt_replay", - "status": result, - "materialize_artifacts": bool(materialize_artifacts), - "artifact_root": str(root), - "ready": receipt_ready, - }, - "target_selectors": target_selectors, - "post_apply_readbacks": post_apply_readbacks, - "executor_receipt_artifact": receipt_artifact, - "materialized_executor_artifacts": materialized_artifacts, - "post_executor_receipt_verifier": { - "checks": verifier_checks, - "check_count": len(verifier_checks), - "passed": all(check.get("passed") is True for check in verifier_checks), - "expected_sha256": receipt_artifact["payload_sha256"], - "actual_sha256": actual_sha, - "hash_match": bool(actual_sha) and actual_sha == receipt_artifact["payload_sha256"], - "writes_database": False, - }, - "rollback_plan": rollback_plan, - "source_artifacts": receipt_payload["source_artifacts"], - "missing_artifacts": missing_artifacts, - "safety": safety, - } def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_drift_verifier_package( diff --git a/tests/test_pchome_mapping_backlog_modularization.py b/tests/test_pchome_mapping_backlog_modularization.py index 2a033ce..6d7b4d2 100644 --- a/tests/test_pchome_mapping_backlog_modularization.py +++ b/tests/test_pchome_mapping_backlog_modularization.py @@ -5,6 +5,7 @@ from services.pchome_mapping_backlog import ( artifacts, contracts, controlled_apply_executor, + controlled_apply_receipt, controlled_apply_verifier, evidence, policies, @@ -135,7 +136,32 @@ def test_controlled_apply_drift_verifier_core_is_no_write(tmp_path): assert result["safety"]["writes_artifact_count"] == 0 -def test_legacy_facade_is_smaller_after_third_p0_extraction(): +def test_controlled_apply_receipt_replay_keeps_identity_and_no_write_default(tmp_path): + assert ( + compatibility_facade.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package + is controlled_apply_receipt.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package + ) + assert ( + compatibility_facade._find_retry_exception_artifact_file + is controlled_apply_receipt._find_retry_exception_artifact_file + ) + + result = controlled_apply_receipt.build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_receipt_replay_package( + artifact_root=tmp_path, + run_id="missing-run", + ) + + assert result["result"] == "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_REPLAY_ARTIFACTS" + assert result["summary"]["missing_artifact_count"] == 3 + assert result["summary"]["writes_database_count"] == 0 + assert result["materialized_executor_artifacts"] == [] + assert result["safety"]["reads_database"] is False + assert result["safety"]["writes_database"] is False + assert result["safety"]["writes_artifact_count"] == 0 + assert len(result["executor_receipt_artifact"]["payload_sha256"]) == 64 + + +def test_legacy_facade_is_smaller_after_fourth_p0_extraction(): facade = ROOT / "services/pchome_mapping_backlog_service.py" - assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_850 + assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_600