#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import subprocess from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from incremental_sync_state import ( atomic_write_json, persist_replay_intent, sanitize_error_category_counts, ) DEFAULT_SYNC_CONFIG = "configs/hourly_delta_sync.json" DEFAULT_LOAD_PLAN_CONFIG = "configs/tidb_load_business_phase1.json" DEFAULT_SUMMARY_NAME = "hourly_delta_sync.json" DEFAULT_RUN_LABEL_PREFIX = "lead-campaign-incremental" DEFAULT_WINDOW_MINUTES = 60 DEFAULT_MAX_WINDOWS_PER_RUN = 4 DEFAULT_BOOTSTRAP_CATCHUP_LOOKBACK_MINUTES = 360 DEFAULT_REPLAY_QUEUE_FILE = "~/.cache/nexuscrm/salesforce/hourly_delta_replay_queue.json" def parse_args() -> argparse.Namespace: script_dir = Path(__file__).resolve().parent parser = argparse.ArgumentParser( description="Run catch-up hourly Salesforce delta sync windows with durable cursor updates." ) parser.add_argument("--config", required=True, help="Path to readonly Salesforce config file.") parser.add_argument( "--sync-config", default=str(script_dir / DEFAULT_SYNC_CONFIG), help="Path to the hourly delta sync config.", ) parser.add_argument( "--load-plan-config", default=str(script_dir / DEFAULT_LOAD_PLAN_CONFIG), help="Path to the hourly delta load plan config.", ) parser.add_argument("--output-root", required=True, help="Root output directory for window summaries.") parser.add_argument("--state-file", required=True, help="Durable sync cursor state file.") parser.add_argument("--replay-queue-file", default=DEFAULT_REPLAY_QUEUE_FILE, help="Durable replay queue state file.") parser.add_argument("--target-base-url", help="NexusCRM base URL for cursor-coupled Lead assignment replay.") parser.add_argument("--target-username", help="NexusCRM username for cursor-coupled Lead assignment replay.") parser.add_argument("--target-password", help="NexusCRM password for cursor-coupled Lead assignment replay.") parser.add_argument( "--target-password-env", help="Environment variable containing the NexusCRM replay password.", ) parser.add_argument("--summary-path", help="Optional batch summary output path.") parser.add_argument("--since", help="Optional explicit catch-up start in UTC ISO-8601.") parser.add_argument("--until", help="Optional explicit catch-up end in UTC ISO-8601.") parser.add_argument("--run-label-prefix", default=DEFAULT_RUN_LABEL_PREFIX, help="Prefix for per-window run labels.") parser.add_argument("--window-minutes", type=int, help="Window size in minutes.") parser.add_argument("--max-windows", type=int, help="Maximum number of windows to process this run.") parser.add_argument("--dry-run", action="store_true", help="Plan windows without executing them.") parser.add_argument( "--validate-only", action="store_true", help=( "Export each window and validate its DB-backed identity/write plan " "without loading, replaying, or advancing durable state." ), ) parser.add_argument( "--disable-replay-queue", action="store_true", help="Compute replay candidates in each window summary but do not write durable replay queue entries.", ) return parser.parse_args() def load_json(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"{path} must be a JSON object") return payload def write_json(path: Path, payload: dict[str, Any]) -> None: atomic_write_json(path, payload) def utc_now() -> datetime: return datetime.now(timezone.utc) def parse_utc_timestamp(value: str | None) -> datetime | None: if not value: return None raw = value.strip() if not raw: return None normalized = raw.replace("Z", "+00:00") dt = datetime.fromisoformat(normalized) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc) def format_sf_timestamp(value: datetime) -> str: return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def floor_to_window(value: datetime, window_minutes: int) -> datetime: if window_minutes <= 0: raise ValueError("window_minutes must be positive") value = value.astimezone(timezone.utc).replace(second=0, microsecond=0) total_minutes = value.hour * 60 + value.minute floored_minutes = total_minutes - (total_minutes % window_minutes) hour = floored_minutes // 60 minute = floored_minutes % 60 return value.replace(hour=hour, minute=minute) def build_windows(start: datetime, end: datetime, window_minutes: int, max_windows: int) -> list[tuple[datetime, datetime]]: if end <= start: return [] if max_windows <= 0: raise ValueError("max_windows must be positive") step = timedelta(minutes=window_minutes) windows: list[tuple[datetime, datetime]] = [] cursor = start while cursor < end and len(windows) < max_windows: next_cursor = min(cursor + step, end) windows.append((cursor, next_cursor)) cursor = next_cursor return windows def load_sync_state(state_file: Path) -> dict[str, Any]: if not state_file.exists(): return {"version": 1} return load_json(state_file) def resolve_window_parameters(args: argparse.Namespace, sync_config: dict[str, Any], state_file: Path) -> tuple[int, int]: window_minutes = int(args.window_minutes or sync_config.get("catchup_window_minutes", DEFAULT_WINDOW_MINUTES)) max_windows = int(args.max_windows or sync_config.get("max_catchup_windows_per_run", DEFAULT_MAX_WINDOWS_PER_RUN)) if window_minutes <= 0: raise ValueError("window_minutes must be positive") if max_windows <= 0: raise ValueError("max_windows must be positive") return window_minutes, max_windows def infer_start(args: argparse.Namespace, sync_config: dict[str, Any], state: dict[str, Any], window_minutes: int) -> datetime: explicit_since = parse_utc_timestamp(args.since) if explicit_since: return floor_to_window(explicit_since, window_minutes) next_window_start = parse_utc_timestamp(str(state.get("next_window_start", "")).strip()) if next_window_start: return floor_to_window(next_window_start, window_minutes) bootstrap_lookback_minutes = int( sync_config.get("bootstrap_catchup_lookback_minutes", DEFAULT_BOOTSTRAP_CATCHUP_LOOKBACK_MINUTES) ) if bootstrap_lookback_minutes < 0: raise ValueError("bootstrap_catchup_lookback_minutes must be non-negative") return floor_to_window(utc_now() - timedelta(minutes=bootstrap_lookback_minutes), window_minutes) def infer_until(args: argparse.Namespace, window_minutes: int) -> datetime: explicit_until = parse_utc_timestamp(args.until) if explicit_until: return floor_to_window(explicit_until, window_minutes) return floor_to_window(utc_now(), window_minutes) def load_replay_queue(queue_file: Path) -> dict[str, Any]: if not queue_file.exists(): return {"version": 1, "entries": []} payload = load_json(queue_file) entries = payload.get("entries") if not isinstance(entries, list): payload["entries"] = [] return payload def upsert_replay_queue_entry( queue_file: Path, *, window_start: datetime, window_end: datetime, description_field: str, lead_ids: list[str], ) -> dict[str, Any]: entry_id = f"{format_sf_timestamp(window_start)}__{format_sf_timestamp(window_end)}" now_str = format_sf_timestamp(utc_now()) return persist_replay_intent( queue_file, entry_id=entry_id, window_start=format_sf_timestamp(window_start), window_end=format_sf_timestamp(window_end), description_field=description_field, lead_ids=lead_ids, updated_at=now_str, ) def load_replay_candidates_before_cursor( replay_prepare: Any, ) -> tuple[list[str], str]: if not isinstance(replay_prepare, dict) or replay_prepare.get("enabled") is not True: raise ValueError("Lead assignment replay preparation is missing or disabled") try: candidate_count = int(replay_prepare["candidate_count"]) except (KeyError, TypeError, ValueError) as exc: raise ValueError("Lead assignment replay candidate_count is invalid") from exc if candidate_count < 0: raise ValueError("Lead assignment replay candidate_count must be non-negative") candidate_path_raw = str(replay_prepare.get("candidate_ids_path", "")).strip() if not candidate_path_raw: raise ValueError("Lead assignment replay candidate file path is missing") candidate_path = Path(candidate_path_raw) if not candidate_path.is_file(): raise ValueError("Lead assignment replay candidate file is missing") lead_ids = [ line.strip() for line in candidate_path.read_text(encoding="utf-8").splitlines() if line.strip() ] if len(lead_ids) != len(set(lead_ids)): raise ValueError("Lead assignment replay candidate file contains duplicate IDs") if len(lead_ids) != candidate_count: raise ValueError( "Lead assignment replay candidate count mismatch: " f"summary={candidate_count} file={len(lead_ids)}" ) description_field = str( replay_prepare.get("target_description_field", "Description") ).strip() or "Description" return lead_ids, description_field def verify_replay_queue_entry( queue_file: Path, *, entry_id: str, lead_ids: list[str], ) -> None: payload = load_replay_queue(queue_file) matching = [ entry for entry in payload.get("entries", []) if isinstance(entry, dict) and str(entry.get("id", "")) == entry_id ] if len(matching) != 1: raise ValueError("Lead assignment replay queue entry readback failed") entry = matching[0] if str(entry.get("status", "")) != "pending": raise ValueError("Lead assignment replay queue entry is not pending") queued_ids = [str(item).strip() for item in entry.get("lead_ids", [])] if queued_ids != lead_ids: raise ValueError("Lead assignment replay queue entry ID readback mismatch") def find_replay_queue_entry(queue_file: Path, entry_id: str) -> dict[str, Any] | None: if not queue_file.is_file(): return None payload = load_replay_queue(queue_file) matching = [ entry for entry in payload.get("entries", []) if isinstance(entry, dict) and str(entry.get("id", "")) == entry_id ] if len(matching) > 1: raise ValueError("Lead assignment replay queue contains duplicate window entries") return matching[0] if matching else None def run_replay_queue_entry_before_cursor( *, script_dir: Path, queue_file: Path, output_root: Path, entry_id: str, target_base_url: str, target_username: str, target_password: str, target_password_env: str, ) -> dict[str, Any]: password_env = target_password_env.strip() password_available = bool(target_password) or bool( password_env and os.environ.get(password_env, "") ) if not target_base_url or not target_username or not password_available: raise ValueError( "Lead assignment replay requires target base URL, username, and password before cursor advancement" ) summary_path = ( output_root / "_cursor_coupled_replay" / f"{entry_id.replace(':', '_')}.json" ) replay_command = [ "python3", str(script_dir / "run_hourly_delta_replay_queue.py"), "--replay-queue-file", str(queue_file), "--entry-id", entry_id, "--base-url", target_base_url, "--username", target_username, "--output-root", str(output_root / "_cursor_coupled_replay"), "--summary-path", str(summary_path), ] if password_env: replay_command.extend(["--password-env", password_env]) else: replay_command.extend(["--password", target_password]) subprocess.run( replay_command, check=True, ) entry = find_replay_queue_entry(queue_file, entry_id) if entry is None or str(entry.get("status", "")) != "success": status = str((entry or {}).get("status", "missing")) error = ValueError( "Lead assignment replay did not reach success before cursor advancement; " f"status={status}" ) error.replay_status = status error.replay_attempt_count = int( (entry or {}).get("attempt_count", 0) or 0 ) error.replay_consecutive_no_progress_count = int( (entry or {}).get("consecutive_no_progress_count", 0) or 0 ) error.replay_remaining_candidate_count = len( { str(item).strip() for item in (entry or {}).get("lead_ids", []) if str(item).strip() } ) error.replay_error_category_counts = sanitize_error_category_counts( (entry or {}).get("last_error_category_counts") ) raise error return { "entry_id": entry_id, "status": "success", "attempt_count": int(entry.get("attempt_count", 0) or 0), "consecutive_no_progress_count": int( entry.get("consecutive_no_progress_count", 0) or 0 ), "remaining_candidate_count": len( { str(item).strip() for item in entry.get("lead_ids", []) if str(item).strip() } ), "error_category_counts": sanitize_error_category_counts( entry.get("last_error_category_counts") ), "summary_path": str(summary_path), } def prune_replay_queue(queue_file: Path, keep_success_entries: int) -> None: payload = load_replay_queue(queue_file) entries = [entry for entry in payload.get("entries", []) if isinstance(entry, dict)] success_entries = [entry for entry in entries if str(entry.get("status", "")) == "success"] success_entries.sort(key=lambda entry: str(entry.get("updated_at", "")), reverse=True) keep_ids = {str(entry.get("id", "")) for entry in success_entries[:keep_success_entries]} payload["entries"] = [ entry for entry in entries if str(entry.get("status", "")) != "success" or str(entry.get("id", "")) in keep_ids ] write_json(queue_file, payload) def build_summary_path(output_root: Path, start: datetime, end: datetime) -> Path: return output_root / "_catchup_manifests" / ( f"hourly_delta_catchup_{start.strftime('%Y%m%dT%H%M%SZ')}_{end.strftime('%Y%m%dT%H%M%SZ')}.json" ) def build_failure_window( exc: BaseException, *, window_start: datetime, window_end: datetime, run_label_prefix: str, ) -> dict[str, Any]: cancelled = isinstance(exc, KeyboardInterrupt) def nonnegative_int(attribute: str) -> int: try: return max(0, int(getattr(exc, attribute, 0) or 0)) except (TypeError, ValueError): return 0 failure = { "window_start": format_sf_timestamp(window_start), "window_end": format_sf_timestamp(window_end), "snapshot_date": window_start.date().isoformat(), "run_label": f"{run_label_prefix}-{window_start.strftime('%Y%m%dT%H%MZ')}", "summary_path": str(getattr(exc, "catchup_window_summary_path", "")), "status": "cancelled" if cancelled else "failed", "failure_stage": str(getattr(exc, "catchup_failure_stage", "unknown")), "error_type": type(exc).__name__, "replay_status": str(getattr(exc, "replay_status", "")), "replay_attempt_count": nonnegative_int("replay_attempt_count"), "replay_consecutive_no_progress_count": nonnegative_int( "replay_consecutive_no_progress_count" ), "replay_remaining_candidate_count": nonnegative_int( "replay_remaining_candidate_count" ), "replay_error_category_counts": sanitize_error_category_counts( getattr(exc, "replay_error_category_counts", {}) ), } if isinstance(exc, subprocess.CalledProcessError): failure["exit_code"] = int(exc.returncode) return failure def validate_window_load_before_cursor(window_summary: dict[str, Any]) -> dict[str, Any]: raw_selected_objects = window_summary.get("selected_objects") if not isinstance(raw_selected_objects, list): raise ValueError("window summary selected_objects must be an array") selected_objects = { str(item).strip() for item in raw_selected_objects if str(item).strip() } if not selected_objects: raise ValueError("window summary has no selected_objects") load_summary = window_summary.get("load") if not isinstance(load_summary, dict): raise ValueError("window summary has no completed load summary") raw_tables = load_summary.get("tables") if not isinstance(raw_tables, list): raise ValueError("window load summary has no tables array") tables_by_object: dict[str, dict[str, Any]] = {} duplicate_summaries: set[str] = set() for item in raw_tables: if not isinstance(item, dict): continue object_name = str(item.get("object", "")).strip() if not object_name: continue if object_name in tables_by_object: duplicate_summaries.add(object_name) tables_by_object[object_name] = item if duplicate_summaries: raise ValueError( "load summary has duplicate selected object(s): " + ", ".join(sorted(duplicate_summaries)) ) missing_summaries = sorted(selected_objects - set(tables_by_object)) if missing_summaries: raise ValueError( "load summary missing selected object(s): " + ", ".join(missing_summaries) ) export_summary = window_summary.get("export") if not isinstance(export_summary, dict): raise ValueError("window summary has no export summary") export_record_counts = export_summary.get("record_counts") if not isinstance(export_record_counts, dict): raise ValueError("window export summary has no record_counts object") missing_export_counts = sorted(selected_objects - set(export_record_counts)) if missing_export_counts: raise ValueError( "export summary missing selected object count(s): " + ", ".join(missing_export_counts) ) rejected: list[dict[str, Any]] = [] accepted: list[dict[str, Any]] = [] for object_name in sorted(selected_objects): table = tables_by_object[object_name] status = str(table.get("status", "")).strip() try: expected_rows = int(export_record_counts[object_name]) loaded_rows = int(table["row_count"]) load_column_count = int(table["load_column_count"]) except (KeyError, TypeError, ValueError) as exc: raise ValueError( f"load acceptance has invalid counts for {object_name}" ) from exc if expected_rows < 0 or loaded_rows < 0 or load_column_count < 0: raise ValueError(f"load acceptance has negative counts for {object_name}") item = { "object": object_name, "status": status, "expected_rows": expected_rows, "loaded_rows": loaded_rows, "load_column_count": load_column_count, } if ( status != "loaded" or load_column_count <= 0 or loaded_rows != expected_rows ): rejected.append(item) else: accepted.append(item) if rejected: raise ValueError( "load acceptance failed before cursor update: " + json.dumps(rejected, sort_keys=True) ) return { "status": "accepted", "selected_object_count": len(selected_objects), "objects": accepted, } def execute_window( *, args: argparse.Namespace, script_dir: Path, sync_script: Path, output_root: Path, state_file: Path, replay_queue_file: Path, window_start: datetime, window_end: datetime, window_minutes: int, replay_required: bool, keep_success_entries: int, ) -> dict[str, Any]: stage = "window_sync" run_label = f"{args.run_label_prefix}-{window_start.strftime('%Y%m%dT%H%MZ')}" window_summary_path = ( output_root / window_start.date().isoformat() / run_label / DEFAULT_SUMMARY_NAME ) try: cmd = [ "python3", str(sync_script), "--config", args.config, "--sync-config", args.sync_config, "--load-plan-config", args.load_plan_config, "--output-root", str(output_root), "--state-file", str(state_file), "--snapshot-date", window_start.date().isoformat(), "--run-label", run_label, "--since", format_sf_timestamp(window_start), "--until", format_sf_timestamp(window_end), "--skip-state-update", "--skip-replay", ] if args.validate_only: cmd.extend(["--dry-run-load", "--skip-check"]) elif not args.disable_replay_queue: cmd.extend(["--replay-queue-file", str(replay_queue_file)]) subprocess.run(cmd, check=True) stage = "window_summary_readback" window_summary = load_json(window_summary_path) load_acceptance = None if not args.validate_only: stage = "load_acceptance" load_acceptance = validate_window_load_before_cursor(window_summary) replay_prepare = window_summary.get("lead_assignment_replay_prepare") or {} replay_entry = None replay_execution = None if not args.validate_only and replay_required: stage = "replay_intent_readback" lead_ids, description_field = load_replay_candidates_before_cursor( replay_prepare ) if args.disable_replay_queue and lead_ids: raise ValueError( "Lead assignment replay queue cannot be disabled when new Lead candidates exist" ) if lead_ids: replay_entry = upsert_replay_queue_entry( replay_queue_file, window_start=window_start, window_end=window_end, description_field=description_field, lead_ids=lead_ids, ) verify_replay_queue_entry( replay_queue_file, entry_id=str(replay_entry.get("id", "")), lead_ids=[ str(item).strip() for item in replay_entry.get("lead_ids", []) if str(item).strip() ], ) entry_id = ( f"{format_sf_timestamp(window_start)}__" f"{format_sf_timestamp(window_end)}" ) replay_entry = replay_entry or find_replay_queue_entry( replay_queue_file, entry_id ) if replay_entry is not None and str(replay_entry.get("status", "")) != "success": if args.disable_replay_queue: raise ValueError( "Lead assignment replay queue cannot be disabled while a window replay is pending" ) stage = "replay_execution" replay_execution = run_replay_queue_entry_before_cursor( script_dir=script_dir, queue_file=replay_queue_file, output_root=output_root, entry_id=entry_id, target_base_url=str(args.target_base_url or "").strip(), target_username=str(args.target_username or "").strip(), target_password=str(args.target_password or ""), target_password_env=str(args.target_password_env or ""), ) cursor_persistence = { "status": "skipped_validate_only", "next_window_start": "", } if not args.validate_only: stage = "cursor_persist" current_state = load_sync_state(state_file) current_next_window_start = parse_utc_timestamp( str(current_state.get("next_window_start", "")).strip() ) if ( current_next_window_start is not None and current_next_window_start >= window_end ): cursor_persistence = { "status": ( "preserved_current_cursor" if current_next_window_start == window_end else "preserved_newer_cursor" ), "next_window_start": format_sf_timestamp( current_next_window_start ), } else: state_payload = { "version": 1, "window_minutes": window_minutes, "next_window_start": format_sf_timestamp(window_end), "last_successful_window_start": format_sf_timestamp(window_start), "last_successful_window_end": format_sf_timestamp(window_end), "last_summary_path": str(window_summary_path), "updated_at": format_sf_timestamp(utc_now()), } write_json(state_file, state_payload) cursor_persistence = { "status": "advanced", "next_window_start": format_sf_timestamp(window_end), } result = { "window_start": format_sf_timestamp(window_start), "window_end": format_sf_timestamp(window_end), "snapshot_date": window_start.date().isoformat(), "run_label": run_label, "summary_path": str(window_summary_path), "replay_candidate_count": int(replay_prepare.get("candidate_count", 0) or 0), "replay_queue_entry_id": str(replay_entry.get("id", "")) if replay_entry else "", "replay_status": str( (replay_execution or replay_entry or {}).get( "status", "not_required" if replay_required else "disabled" ) ), "replay_attempt_count": int( (replay_execution or replay_entry or {}).get("attempt_count", 0) or 0 ), "replay_consecutive_no_progress_count": int( (replay_execution or replay_entry or {}).get( "consecutive_no_progress_count", 0 ) or 0 ), "replay_remaining_candidate_count": int( (replay_execution or replay_entry or {}).get( "remaining_candidate_count", (replay_execution or replay_entry or {}).get( "candidate_count", 0 ), ) or 0 ), "replay_error_category_counts": sanitize_error_category_counts( (replay_execution or replay_entry or {}).get( "error_category_counts", (replay_execution or replay_entry or {}).get( "last_error_category_counts", {} ), ) ), "cursor_persistence": cursor_persistence, "load_acceptance": load_acceptance, "status": "validated" if args.validate_only else "completed", } if not args.validate_only: stage = "replay_queue_prune" prune_replay_queue( replay_queue_file, keep_success_entries=keep_success_entries, ) return result except BaseException as exc: try: setattr(exc, "catchup_failure_stage", stage) setattr(exc, "catchup_window_summary_path", str(window_summary_path)) except Exception: pass raise def main() -> int: args = parse_args() script_dir = Path(__file__).resolve().parent sync_script = script_dir / "run_hourly_delta_sync.py" sync_config = load_json(Path(args.sync_config).expanduser().resolve()) state_file = Path(args.state_file).expanduser().resolve() replay_queue_file = Path(args.replay_queue_file).expanduser().resolve() output_root = Path(args.output_root).expanduser().resolve() state = load_sync_state(state_file) window_minutes, max_windows = resolve_window_parameters(args, sync_config, state_file) start = infer_start(args, sync_config, state, window_minutes) until = infer_until(args, window_minutes) windows = build_windows(start, until, window_minutes, max_windows) summary_path = Path(args.summary_path).expanduser().resolve() if args.summary_path else build_summary_path(output_root, start, until) summary: dict[str, Any] = { "manifest_version": 1, "run_type": "hourly_delta_catchup", "state_file": str(state_file), "replay_queue_file": str(replay_queue_file), "window_minutes": window_minutes, "max_windows": max_windows, "effective_start": format_sf_timestamp(start), "effective_until": format_sf_timestamp(until), "planned_window_count": len(windows), "replay_queue_disabled": args.disable_replay_queue, "dry_run": args.dry_run, "validate_only": args.validate_only, "windows": [], } if args.dry_run: summary["windows"] = [ { "window_start": format_sf_timestamp(window_start), "window_end": format_sf_timestamp(window_end), "snapshot_date": window_start.date().isoformat(), "run_label": f"{args.run_label_prefix}-{window_start.strftime('%Y%m%dT%H%MZ')}", "status": "planned", } for window_start, window_end in windows ] write_json(summary_path, summary) print(f"[ok] catchup_summary={summary_path}") return 0 keep_success_entries = int(sync_config.get("replay_success_retention_entries", 48)) replay_config = sync_config.get("lead_assignment_replay") replay_required = isinstance(replay_config, dict) and replay_config.get("enabled") is True for window_start, window_end in windows: try: window_result = execute_window( args=args, script_dir=script_dir, sync_script=sync_script, output_root=output_root, state_file=state_file, replay_queue_file=replay_queue_file, window_start=window_start, window_end=window_end, window_minutes=window_minutes, replay_required=replay_required, keep_success_entries=keep_success_entries, ) except BaseException as exc: cancelled = isinstance(exc, KeyboardInterrupt) summary["terminal_status"] = "cancelled" if cancelled else "failed" failure = build_failure_window( exc, window_start=window_start, window_end=window_end, run_label_prefix=args.run_label_prefix, ) summary["windows"].append(failure) write_json(summary_path, summary) raise summary["windows"].append(window_result) summary["terminal_status"] = "completed" write_json(summary_path, summary) print(f"[ok] catchup_summary={summary_path}") if args.validate_only: print("[ok] validate_only=true state_unchanged=true replay_queue_unchanged=true") else: print(f"[ok] state_file={state_file}") print(f"[ok] replay_queue_file={replay_queue_file}") return 0 if __name__ == "__main__": raise SystemExit(main())