#!/usr/bin/env python3 from __future__ import annotations import argparse import json import re from pathlib import Path from typing import Any from incremental_sync_state import sanitize_error_category_counts ALLOWED_OBJECTS = {"Lead", "LeadHistory", "LeadShare"} ALLOWED_WINDOW_STATUSES = {"planned", "validated", "completed", "failed", "cancelled"} ALLOWED_SYNC_OUTCOMES = {"success", "failure", "cancelled", "skipped"} ALLOWED_IDENTITY_STATUSES = {"completed", "skipped", "conflict"} ALLOWED_REFERENCE_HYDRATION_STATUSES = {"completed", "skipped"} ALLOWED_PARENT_PULL_FORWARD_STATUSES = { "completed", "skipped", "blocked_source_parent_unavailable", } ALLOWED_DEDUPE_STATUSES = { "completed", "skipped", "conflict", "blocked_unmatched", "blocked_unmatched_reference", } ALLOWED_LOAD_STATUSES = { "loaded", "missing_table", "missing_file", "no_common_columns", } ALLOWED_REPLAY_STATUSES = { "success", "not_required", "disabled", "pending", "failed", "dead_letter", } ALLOWED_CURSOR_PERSISTENCE_STATUSES = { "advanced", "preserved_current_cursor", "preserved_newer_cursor", "skipped_validate_only", } ALLOWED_FAILURE_STAGES = { "window_sync", "window_summary_readback", "load_acceptance", "replay_intent_readback", "replay_execution", "cursor_persist", "replay_queue_prune", } ALLOWED_ERROR_TYPES = { "CalledProcessError", "FileNotFoundError", "KeyboardInterrupt", "OSError", "RuntimeError", "ValueError", } ISO_UTC_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") SHA256_RE = re.compile(r"^[0-9a-f]{64}$") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Build an allowlisted, PII-free evidence artifact for the Prod Lead incremental sync." ) parser.add_argument("--catchup-summary", required=True) parser.add_argument("--output", required=True) parser.add_argument("--sync-outcome", required=True) parser.add_argument("--mode", choices=("write", "validate-only"), required=True) parser.add_argument("--state-before-sha256", required=True) parser.add_argument("--state-after-sha256", required=True) parser.add_argument("--replay-before-sha256", required=True) parser.add_argument("--replay-after-sha256", required=True) parser.add_argument("--head-sha", default="") parser.add_argument("--run-id", default="") parser.add_argument("--run-attempt", default="") return parser.parse_args() def load_object(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"{path} must contain a JSON object") return payload def safe_int(value: Any) -> int: try: return max(0, int(value or 0)) except (TypeError, ValueError): return 0 def safe_timestamp(value: Any) -> str: text = str(value or "").strip() return text if ISO_UTC_RE.fullmatch(text) else "" def safe_hash(value: Any) -> str: text = str(value or "").strip().lower() return text if text == "missing" or SHA256_RE.fullmatch(text) else "invalid" def hashes_match(before: Any, after: Any) -> bool: safe_before = safe_hash(before) safe_after = safe_hash(after) return safe_before != "invalid" and safe_before == safe_after def safe_status(value: Any, allowed: set[str]) -> str: text = str(value or "").strip() return text if text in allowed else "unknown" def safe_object(value: Any) -> str: text = str(value or "").strip() return text if text in ALLOWED_OBJECTS else "unexpected" def sanitize_window_summary(payload: dict[str, Any]) -> dict[str, Any]: export = payload.get("export") if isinstance(payload.get("export"), dict) else {} raw_record_counts = export.get("record_counts") record_counts = raw_record_counts if isinstance(raw_record_counts, dict) else {} raw_window_record_counts = export.get("window_record_counts") window_record_counts = ( raw_window_record_counts if isinstance(raw_window_record_counts, dict) else {} ) raw_prerequisite_parent_counts = export.get("prerequisite_parent_counts") prerequisite_parent_counts = ( raw_prerequisite_parent_counts if isinstance(raw_prerequisite_parent_counts, dict) else {} ) identity = ( payload.get("person_identity_resolution") if isinstance(payload.get("person_identity_resolution"), dict) else {} ) reference_hydration = ( payload.get("lead_cross_provider_reference_hydration") if isinstance(payload.get("lead_cross_provider_reference_hydration"), dict) else {} ) dedupe = ( payload.get("lead_cross_provider_dedupe") if isinstance(payload.get("lead_cross_provider_dedupe"), dict) else {} ) parent_pull_forward = ( payload.get("lead_cross_window_parent_pull_forward") if isinstance(payload.get("lead_cross_window_parent_pull_forward"), dict) else {} ) replay_intent = ( payload.get("lead_assignment_replay_intent") if isinstance(payload.get("lead_assignment_replay_intent"), dict) else {} ) load = payload.get("load") if isinstance(payload.get("load"), dict) else {} raw_tables = load.get("tables") if isinstance(load.get("tables"), list) else [] selected_objects = sorted( { safe_object(item) for item in payload.get("selected_objects", []) if safe_object(item) != "unexpected" } ) tables = [] for item in raw_tables: if not isinstance(item, dict): continue object_name = safe_object(item.get("object")) if object_name == "unexpected": continue tables.append( { "object": object_name, "status": safe_status(item.get("status"), ALLOWED_LOAD_STATUSES), "row_count": safe_int(item.get("row_count")), "load_column_count": safe_int(item.get("load_column_count")), "chunk_count": safe_int(item.get("chunk_count")), } ) return { "selected_objects": selected_objects, "export_record_counts": { object_name: safe_int(record_counts.get(object_name)) for object_name in sorted(ALLOWED_OBJECTS) if object_name in record_counts }, "window_record_counts": { object_name: safe_int(window_record_counts.get(object_name)) for object_name in sorted(ALLOWED_OBJECTS) if object_name in window_record_counts }, "prerequisite_parent_counts": { object_name: safe_int(prerequisite_parent_counts.get(object_name)) for object_name in sorted(ALLOWED_OBJECTS) if object_name in prerequisite_parent_counts }, "person_identity_resolution": { "status": safe_status(identity.get("status"), ALLOWED_IDENTITY_STATUSES), "object_count": safe_int(identity.get("object_count")), "remapped_count": safe_int(identity.get("remapped_count")), "deferred_reference_count": safe_int( identity.get("deferred_reference_count") ), }, "lead_cross_provider_reference_hydration": { "status": safe_status( reference_hydration.get("status"), ALLOWED_REFERENCE_HYDRATION_STATUSES, ), "reference_only_source_count": safe_int( reference_hydration.get("reference_only_source_count") ), "hydrated_source_count": safe_int( reference_hydration.get("hydrated_source_count") ), "missing_source_count": safe_int( reference_hydration.get("missing_source_count") ), }, "lead_cross_provider_dedupe": { "status": safe_status(dedupe.get("status"), ALLOWED_DEDUPE_STATUSES), "remapped_count": safe_int(dedupe.get("remapped_count")), "conflict_count": safe_int(dedupe.get("conflict_count")), "unmatched_count": safe_int(dedupe.get("unmatched_count")), "create_candidate_count": safe_int(dedupe.get("create_candidate_count")), "unmatched_reference_count": safe_int( dedupe.get("unmatched_reference_count") ), "reference_only_source_count": safe_int( dedupe.get("reference_only_source_count") ), "require_existing_target_match": dedupe.get("require_existing_target_match") is True, }, "lead_cross_window_parent_pull_forward": { "status": safe_status( parent_pull_forward.get("status"), ALLOWED_PARENT_PULL_FORWARD_STATUSES, ), "requested_parent_count": safe_int( parent_pull_forward.get("requested_parent_count") ), "prerequisite_parent_count": safe_int( parent_pull_forward.get("prerequisite_parent_count") ), "missing_source_count": safe_int( parent_pull_forward.get("missing_source_count") ), "deleted_source_count": safe_int( parent_pull_forward.get("deleted_source_count") ), }, "lead_assignment_replay_intent": { "status": safe_status( replay_intent.get("status"), ALLOWED_REPLAY_STATUSES ), "candidate_count": safe_int(replay_intent.get("candidate_count")), "persisted_before_load": replay_intent.get("persisted_before_load") is True, }, "load_tables": sorted(tables, key=lambda item: item["object"]), } def build_evidence( *, catchup_summary_path: Path, sync_outcome: str, mode: str, state_before_sha256: str, state_after_sha256: str, replay_before_sha256: str, replay_after_sha256: str, head_sha: str, run_id: str, run_attempt: str, ) -> dict[str, Any]: evidence: dict[str, Any] = { "evidence_version": 1, "run": { "head_sha": head_sha if re.fullmatch(r"[0-9a-fA-F]{40}", head_sha) else "", "run_id": run_id if run_id.isdigit() else "", "run_attempt": run_attempt if run_attempt.isdigit() else "", "sync_outcome": safe_status(sync_outcome, ALLOWED_SYNC_OUTCOMES), "mode": mode if mode in {"write", "validate-only"} else "unknown", }, "durable_state": { "state_before_sha256": safe_hash(state_before_sha256), "state_after_sha256": safe_hash(state_after_sha256), "state_unchanged": hashes_match(state_before_sha256, state_after_sha256), "replay_before_sha256": safe_hash(replay_before_sha256), "replay_after_sha256": safe_hash(replay_after_sha256), "replay_unchanged": hashes_match(replay_before_sha256, replay_after_sha256), }, "catchup_summary_found": catchup_summary_path.is_file(), "catchup": None, "windows": [], } if not catchup_summary_path.is_file(): return evidence catchup = load_object(catchup_summary_path) evidence["catchup"] = { "effective_start": safe_timestamp(catchup.get("effective_start")), "effective_until": safe_timestamp(catchup.get("effective_until")), "planned_window_count": safe_int(catchup.get("planned_window_count")), "validate_only": catchup.get("validate_only") is True, "replay_queue_disabled": catchup.get("replay_queue_disabled") is True, } raw_windows = catchup.get("windows") if isinstance(catchup.get("windows"), list) else [] for raw_window in raw_windows: if not isinstance(raw_window, dict): continue raw_cursor_persistence = raw_window.get("cursor_persistence") cursor_persistence = ( raw_cursor_persistence if isinstance(raw_cursor_persistence, dict) else {} ) window = { "window_start": safe_timestamp(raw_window.get("window_start")), "window_end": safe_timestamp(raw_window.get("window_end")), "status": safe_status(raw_window.get("status"), ALLOWED_WINDOW_STATUSES), "replay_candidate_count": safe_int( raw_window.get("replay_candidate_count") ), "replay_status": safe_status( raw_window.get("replay_status"), ALLOWED_REPLAY_STATUSES ), "replay_attempt_count": safe_int( raw_window.get("replay_attempt_count") ), "replay_consecutive_no_progress_count": safe_int( raw_window.get("replay_consecutive_no_progress_count") ), "replay_remaining_candidate_count": safe_int( raw_window.get("replay_remaining_candidate_count") ), "replay_error_category_counts": sanitize_error_category_counts( raw_window.get("replay_error_category_counts") ), "cursor_persistence": { "status": safe_status( cursor_persistence.get("status"), ALLOWED_CURSOR_PERSISTENCE_STATUSES, ), "next_window_start": safe_timestamp( cursor_persistence.get("next_window_start") ), }, "failure_stage": safe_status( raw_window.get("failure_stage"), ALLOWED_FAILURE_STAGES ), "error_type": safe_status( raw_window.get("error_type"), ALLOWED_ERROR_TYPES ), "exit_code": safe_int(raw_window.get("exit_code")), "details_found": False, "details": None, } raw_summary_path = str(raw_window.get("summary_path") or "").strip() if raw_summary_path: detail_path = Path(raw_summary_path) if detail_path.is_file(): window["details_found"] = True window["details"] = sanitize_window_summary(load_object(detail_path)) evidence["windows"].append(window) return evidence def main() -> int: args = parse_args() evidence = build_evidence( catchup_summary_path=Path(args.catchup_summary).expanduser().resolve(), sync_outcome=args.sync_outcome, mode=args.mode, state_before_sha256=args.state_before_sha256, state_after_sha256=args.state_after_sha256, replay_before_sha256=args.replay_before_sha256, replay_after_sha256=args.replay_after_sha256, head_sha=args.head_sha, run_id=args.run_id, run_attempt=args.run_attempt, ) output_path = Path(args.output).expanduser().resolve() output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8") print(f"[ok] redacted_evidence={output_path}") return 0 if __name__ == "__main__": raise SystemExit(main())