"""No-write drift verifier for controlled PChome mapping receipts.""" 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.policies import ( DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, ) def build_controlled_apply_drift_verifier_from_replay( replay: dict[str, Any], *, root: Path, run_id: str | None = None, engine: Any = None, source_receipt_replay: dict[str, Any] | None = None, materialize_artifacts: bool = False, ) -> dict[str, Any]: """Verify replay readbacks and emit a bounded drift receipt.""" replay_summary = replay.get("summary") or {} readbacks = list(replay.get("post_apply_readbacks") or []) drift_items = [item for item in readbacks if item.get("passed") is not True] selector_count = int(replay_summary.get("target_selector_count") or 0) pass_count = int(replay_summary.get("post_apply_readback_pass_count") or 0) receipt_hash_match_count = int(replay_summary.get("executor_receipt_hash_match_count") or 0) source_ready = ( replay.get("result") == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_RECEIPT_REPLAYED" and selector_count > 0 and receipt_hash_match_count > 0 ) drift_verified = source_ready and not drift_items and pass_count == selector_count if drift_items: result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_DETECTED" elif drift_verified: result = "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED" elif replay.get("missing_artifacts"): result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_ARTIFACTS" else: result = "WAITING_FOR_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_BASELINE" checks = [ {"check": "source_replay_loaded", "passed": bool(replay)}, {"check": "source_receipt_hash_matches", "passed": receipt_hash_match_count > 0}, {"check": "target_selectors_present", "passed": selector_count > 0}, {"check": "all_current_readbacks_match_receipt", "passed": not drift_items and pass_count == selector_count}, {"check": "drift_verifier_does_not_write_database", "passed": True}, ] verifier_id_payload = { "run_id": (replay.get("receipt_replay") or {}).get("run_id") or run_id or "", "result": result, "post_apply_readbacks": readbacks, } verifier_id = ( "pchome-retry-exception-controlled-apply-drift-verifier-" + hashlib.sha256( json.dumps(verifier_id_payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") ).hexdigest()[:16] ) summary = { "target_selector_count": selector_count, "post_apply_readback_count": int(replay_summary.get("post_apply_readback_count") or 0), "post_apply_readback_pass_count": pass_count, "drift_count": len(drift_items), "drift_verified_count": 1 if drift_verified else 0, "receipt_hash_match_count": receipt_hash_match_count, "missing_artifact_count": int(replay_summary.get("missing_artifact_count") or 0), "drift_verifier_artifact_materialized_count": 0, "drift_verifier_artifact_hash_match_count": 0, "writes_database_count": 0, } safety = { "ai_controlled_apply": True, "reads_artifact_files": True, "reads_database": engine is not None or bool(source_receipt_replay), "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, } artifact_payload = { "artifact_key": "retry_exception_controlled_apply_drift_verifier_receipt", "verifier_id": verifier_id, "run_id": verifier_id_payload["run_id"], "source_policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, "source_receipt_replay_result": replay.get("result"), "result": result, "summary": summary, "drift_items": drift_items, "post_apply_readbacks": readbacks, "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_drift_verifier/{verifier_id}.json" ) drift_verifier_artifact = { "key": "retry_exception_controlled_apply_drift_verifier_receipt", "artifact_type": "controlled_apply_drift_verifier_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_drift_artifacts: list[dict[str, Any]] = [] if materialize_artifacts and selector_count: 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_drift_artifacts.append({ "key": drift_verifier_artifact["key"], "relative_path": artifact_relative_path, "absolute_path": str(target_path), "payload_sha256": drift_verifier_artifact["payload_sha256"], "written_byte_count": target_path.stat().st_size, "writes_database": False, }) drift_verifier_artifact["materialized"] = True drift_verifier_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 == drift_verifier_artifact["payload_sha256"] summary["drift_verifier_artifact_materialized_count"] = len(materialized_drift_artifacts) or (1 if artifact_hash_match else 0) summary["drift_verifier_artifact_hash_match_count"] = 1 if artifact_hash_match else 0 safety["writes_artifact_count"] = len(materialized_drift_artifacts) checks.extend([ { "check": "drift_artifact_materialized_when_requested", "passed": (not materialize_artifacts) or (selector_count > 0 and artifact_path.exists()), }, { "check": "drift_artifact_hash_matches_expected", "passed": (not materialize_artifacts) or artifact_hash_match, }, ]) return { "policy": DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIER_POLICY, "result": result, "success": result == "DIRECT_MAPPING_RETRY_EXCEPTION_CONTROLLED_APPLY_DRIFT_VERIFIED", "summary": summary, "drift_verifier": { "verifier_id": verifier_id, "stage": "P2_retry_exception_controlled_apply_drift_verifier", "status": result, "source_receipt_replay_result": replay.get("result"), "ready": drift_verified, "materialize_artifacts": bool(materialize_artifacts), "requires_production_version_truth": True, }, "drift_items": drift_items, "post_apply_readbacks": readbacks, "source_receipt_replay_summary": replay_summary, "drift_verifier_artifact": drift_verifier_artifact, "materialized_drift_artifacts": materialized_drift_artifacts, "post_drift_verifier_artifact_verifier": { "expected_sha256": drift_verifier_artifact["payload_sha256"], "actual_sha256": artifact_sha, "hash_match": artifact_hash_match, "writes_database": False, }, "checks": checks, "check_count": len(checks), "all_checks_passed": all(check.get("passed") is True for check in checks), "next_actions": [ "Keep this verifier on the readiness surface so DB drift is visible without manual table review.", "If drift is detected, use the receipt replay readbacks as rollback or re-apply evidence.", ], "safety": safety, } __all__ = ( "build_controlled_apply_drift_verifier_from_replay", )