#!/usr/bin/env python3 """Apply the EXP baseline through its existing bulk/REST engine.""" from __future__ import annotations import argparse import json import os from pathlib import Path import subprocess import sys import time from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent REPO_ROOT = SCRIPT_DIR.parent RUNTIME_DIR = REPO_ROOT / "deployments" / "offline" / "runtime" if str(RUNTIME_DIR) not in sys.path: sys.path.insert(0, str(RUNTIME_DIR)) import deploy_offline_env # noqa: E402 SUMMARY_FILENAME = "summary.json" class ExpBaselineAdapterError(RuntimeError): """Raised when the reviewed EXP baseline adapter contract is violated.""" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Apply the EXP baseline Seed module.") parser.add_argument("--source-root", required=True) parser.add_argument("--seed-file", required=True) parser.add_argument("--output-dir", required=True) parser.add_argument("--dry-run", action="store_true") return parser.parse_args(argv) def _read_json(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ExpBaselineAdapterError(f"invalid bulk baseline summary: {path}") from exc if not isinstance(payload, dict): raise ExpBaselineAdapterError("bulk baseline summary must be an object") return payload def _write_json(path: Path, payload: dict[str, Any]) -> None: path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _baseline_objects(bulk_summary: dict[str, Any]) -> list[str]: objects: list[str] = [] for value in bulk_summary.get("bulk_objects") or []: if isinstance(value, str) and value and value not in objects: objects.append(value) for bucket in ("fallback_objects", "unsupported"): for value in bulk_summary.get(bucket) or []: object_api_name = value.get("object") if isinstance(value, dict) else None if ( isinstance(object_api_name, str) and object_api_name and object_api_name not in objects ): objects.append(object_api_name) return objects def normalize_summary( bulk_summary: dict[str, Any], *, dry_run: bool, payload: dict[str, object] | None, bulk_record_count: int, duration_ms: int, ) -> dict[str, Any]: configured_alias_count = int(bulk_summary.get("total_alias_count", 0)) collapsed_alias_count = int(bulk_summary.get("collapsed_duplicate_records", 0)) physical_record_count = configured_alias_count - collapsed_alias_count fallback_record_count = physical_record_count - bulk_record_count if physical_record_count < 0 or fallback_record_count != 0: raise ExpBaselineAdapterError("bulk baseline counts are inconsistent") if bulk_summary.get("coverage_complete") is not True: raise ExpBaselineAdapterError( "bulk baseline coverage is incomplete; REST fallback is retired" ) materialized_alias_count = 0 failed_alias_count = 0 if payload is not None: materialized_alias_count = len(payload.get("applied_records", []) or []) + len( payload.get("hydrated_skipped_records", []) or [] ) failed_alias_count = len(payload.get("failed_records", []) or []) coverage_complete = ( bulk_summary.get("coverage_complete") is True if dry_run else failed_alias_count == 0 and materialized_alias_count == configured_alias_count ) return { "app_shell_reference_count": int( bulk_summary.get("app_shell_reference_count", 0) ), "apply_engine": "bulk-plan" if dry_run else "bulk", "baseline_objects": _baseline_objects(bulk_summary), "bulk_record_count": bulk_record_count, "collapsed_alias_count": collapsed_alias_count, "configured_alias_count": configured_alias_count, "conflicting_duplicate_match_group_count": int( bulk_summary.get("conflicting_duplicate_match_group_count", 0) ), "coverage_complete": coverage_complete, "dry_run": dry_run, "duplicate_match_group_count": int( bulk_summary.get("duplicate_match_group_count", 0) ), "duration_ms": duration_ms, "fallback_record_count": fallback_record_count, "fallback_used": False, "lookup_record_count": int(bulk_summary.get("lookup_record_count", 0)), "materialized_alias_count": materialized_alias_count, "physical_record_count": physical_record_count, } def _dry_run_bulk(source_root: Path, seed_file: Path, output_dir: Path) -> dict[str, Any]: bulk_output_dir = output_dir / "bulk" state_file = bulk_output_dir / "runtime-state.json" summary_file = bulk_output_dir / "summary.json" completed = subprocess.run( [ "bash", str(source_root / "scripts" / "apply_exp_bulk_baseline.sh"), "--seed-file", str(seed_file), "--output-dir", str(bulk_output_dir), "--state-file", str(state_file), "--summary-file", str(summary_file), "--dry-run", ], cwd=source_root, env={**os.environ, "EXP_BULK_HASH_USER_PASSWORDS": "0"}, check=False, ) if completed.returncode != 0: raise ExpBaselineAdapterError( f"bulk baseline dry-run failed: exit={completed.returncode}" ) return _read_json(summary_file) def run_adapter(args: argparse.Namespace) -> tuple[dict[str, Any], Path]: started = time.perf_counter() source_root = Path(args.source_root).expanduser().resolve() seed_file = Path(args.seed_file).expanduser().resolve() expected_seed_file = ( source_root / "deployments" / "offline" / "runtime" / "seeds" / "exp-baseline.seed.yaml" ) if seed_file != expected_seed_file or not seed_file.is_file(): raise ExpBaselineAdapterError("seed file is not the reviewed EXP baseline source") output_dir = Path(args.output_dir).expanduser().resolve() output_dir.mkdir(parents=True, exist_ok=False) environment = dict(os.environ) payload: dict[str, object] | None = None if args.dry_run: bulk_summary = _dry_run_bulk(source_root, seed_file, output_dir) applied_bulk_record_count = int(bulk_summary.get("bulk_record_count", 0)) else: nexuscrm_worktree_raw = environment.get("NEXUSCRM_WORKTREE", "").strip() if not nexuscrm_worktree_raw: raise ExpBaselineAdapterError("NEXUSCRM_WORKTREE is required in apply mode") nexuscrm_worktree = Path(nexuscrm_worktree_raw).expanduser().resolve() if not nexuscrm_worktree.is_dir(): raise ExpBaselineAdapterError( f"NEXUSCRM_WORKTREE is not a directory: {nexuscrm_worktree}" ) payload, bulk_result = deploy_offline_env.apply_exp_baseline_seed( environment, source_root, nexuscrm_worktree, ) bulk_summary = bulk_result.summary applied_bulk_record_count = bulk_result.records if not bulk_summary.get("total_alias_count"): bulk_summary = _dry_run_bulk(source_root, seed_file, output_dir) summary = normalize_summary( bulk_summary, dry_run=args.dry_run, payload=payload, bulk_record_count=applied_bulk_record_count, duration_ms=round((time.perf_counter() - started) * 1000), ) summary_path = output_dir / SUMMARY_FILENAME _write_json(summary_path, summary) print(json.dumps(summary, indent=2, sort_keys=True)) return summary, summary_path def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: run_adapter(args) except ExpBaselineAdapterError as exc: print(f"EXP baseline adapter failed: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())