#!/usr/bin/env python3 """Apply one Catalog phase through reviewed deploy Seed adapters.""" from __future__ import annotations import argparse import copy import json import os from pathlib import Path import re import subprocess import sys import tempfile from typing import Any, Callable, Mapping SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) from plan_deploy_seed import ( # noqa: E402 PROFILES, DirectorySource, SeedPlanError, plan_deploy_seed, render_json, write_artifact_pair, ) APPLY_EVIDENCE_SCHEMA_VERSION = 1 APPLY_EVIDENCE_FORMAT = "deploy-seed-apply-v1" CATALOG_RELPATH = "deployments/offline/runtime/seeds/deploy-seed-catalog.json" PLAN_FILENAME = "seed-plan.json" INVENTORY_FILENAME = "deploy-seed-inventory.json" EVIDENCE_FILENAME = "seed-apply-evidence.json" TRUE_VALUES = {"1", "true", "yes", "on"} FALSE_VALUES = {"0", "false", "no", "off"} FLAG_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") SCRIPTED_ENGINE_RE = re.compile(r"^[a-z][a-z0-9_-]*$") SCRIPTED_ADAPTER_ID = "scripted_seed_v1" PARTITIONED_SCRIPTED_ADAPTER_ID = "scripted_seed_partition_v1" SCRIPTED_ADAPTER_IDS = {SCRIPTED_ADAPTER_ID, PARTITIONED_SCRIPTED_ADAPTER_ID} SCRIPTED_SUMMARY_SCHEMA_VERSION = 1 SCRIPTED_SUMMARY_FORMAT = "deploy-seed-script-v1" PARTITIONED_SCRIPTED_SUMMARY_FORMAT = "deploy-seed-script-partition-v1" SAFE_PROCESS_ENV_KEYS = ( "LANG", "LC_ALL", "SSL_CERT_DIR", "SSL_CERT_FILE", "TMPDIR", "PYTHONDONTWRITEBYTECODE", ) CONDITION_ENV_KEYS = ( "APPLY_CONNECTED_APPS", "APPLY_EXP_CREDITS_APPROVAL_SEED", "APPLY_EXP_CREDITS_PERMISSIONS_SEED", "APPLY_EXP_QUOTE_APPROVAL_SEED", "APPLY_PINGCAP_REGION_OWNER_RULES", "APPLY_SHARED_DEPLOY_BASELINE", "APPLY_STAGING_ACCOUNT_NOTIFY_SEED", "APPLY_STAGING_CREDITS_APPROVAL_SEED", "APPLY_STAGING_CREDITS_PERMISSIONS_SEED", "APPLY_STAGING_QUOTE_APPROVAL_SEED", "APPLY_TEST_CREDITS_APPROVAL_SEED", "APPLY_TEST_CREDITS_PERMISSIONS_SEED", "APPLY_TEST_QUOTE_APPROVAL_SEED", "APPLY_IDENTITY_SYNC_POLICY", "DEPLOY_PACK_REFERENCE_DATA_APPLY", "SKIP_BACKEND_START", "START_SERVICES", ) PROFILE_ENV_KEYS = { "exp": ("EXP_HUBSPOT_ACCESS_TOKEN",), } PROFILE_ONLY_ENV_KEYS = { name for allowed_keys in PROFILE_ENV_KEYS.values() for name in allowed_keys } DATABASE_ENV_KEYS = ( "TIDB_DATABASE", "TIDB_HOST", "TIDB_PASSWORD", "TIDB_PORT", "TIDB_TLS", "TIDB_USER", ) RUNTIME_API_ENV_KEYS = ( "BACKEND_PORT", "NEXUS_ADMIN_PASSWORD", "VERIFY_USERNAME", ) BULK_ENV_KEYS = ( "EXP_BULK_BASELINE_MAX_WORKERS", "EXP_BULK_BASELINE_OBJECTS", "NEXUSCRM_WORKTREE", "PYTHON_SEED_DEPS_DIR", ) ADAPTER_ENV_KEYS = { "account_approval_notify": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, "SKIP_BACKEND_START", ), "connected_apps": ( *RUNTIME_API_ENV_KEYS, "CONNECTED_APPS_TONY_CONSUMER_SECRET", "CONNECTED_APPS_ZENO_CONSUMER_SECRET", "STAGING_INSPECTION_OAUTH_CLIENT_ID", "STAGING_INSPECTION_OAUTH_CLIENT_SECRET", ), "credits_approval_seed": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, "SKIP_BACKEND_START", ), "credits_permissions_seed": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, "SKIP_BACKEND_START", ), "deploy_baseline": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, *BULK_ENV_KEYS, ), "exp_baseline": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, *BULK_ENV_KEYS, ), "exp_hubspot_owner_bindings": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, "EXP_HUBSPOT_ACCESS_TOKEN", ), "identity_sync_policy": ( *DATABASE_ENV_KEYS, *RUNTIME_API_ENV_KEYS, "STAGING_INSPECTION_OAUTH_CLIENT_ID", "STAGING_INSPECTION_OAUTH_CLIENT_SECRET", ), "pack_reference_data": ( *DATABASE_ENV_KEYS, "PACK_REFERENCE_DATA_MYSQL_CLIENT", "VERIFY_USERNAME", ), "pingcap_region_owner_rules": ( *RUNTIME_API_ENV_KEYS, "VERIFY_PINGCAP_REGION_OWNER_RULES", ), } ADAPTERS = { "deploy_baseline_v1": { "module_id": "deploy_baseline", "entrypoint": "scripts/apply_deploy_baseline.py", "seed_file": "deployments/offline/runtime/seeds/deploy-baseline.seed.yaml", "objects": ( "Countries_Region__c", "Country_Dictionary__c", "UserRole", "Profile", "User", "HubSpotOwnerBinding__c", ), "invocation_ids": ( "exp.deploy_baseline.post_exp_baseline", "staging.deploy_baseline.post_reference_data", "test.deploy_baseline.post_backend_health", ), "invocation_targets": { "exp.deploy_baseline.post_exp_baseline": "exp", "staging.deploy_baseline.post_reference_data": "staging", "test.deploy_baseline.post_backend_health": "test", }, }, "exp_baseline_v1": { "module_id": "exp_baseline", "entrypoint": "scripts/apply_exp_baseline.py", "seed_file": "deployments/offline/runtime/seeds/exp-baseline.seed.yaml", "objects": ( "Pricebook2", "Product2", "User", "_System_ConnectedApp", "PricebookEntry", "HubSpotOwnerBinding__c", ), "invocation_ids": ("exp.exp_baseline.post_runtime_start",), }, "pack_reference_data_v1": { "module_id": "pack_reference_data", "entrypoint": "scripts/apply_pack_reference_data.sh", "manifest": "deployments/offline/runtime/reference-data/manifest.json", "invocation_selections": { "exp.pack_reference_data.governed_reference_data": { "objects": ( "Industry_Vertical__c", "Countries_Region__c", "Country_Dictionary__c", "Dated_Conversion_Rate__c", ) }, "staging.pack_reference_data.post_backend_health": { "objects": ( "Industry_Vertical__c", "ReportFolder", "_System_EmailTemplate", "Dated_Conversion_Rate__c", ) }, "test.pack_reference_data.governed_reference_data": { "objects": ("Industry_Vertical__c", "Dated_Conversion_Rate__c") }, }, } } BASELINE_SUMMARY_KEYS = { "dry_run", "apply_engine", "baseline_objects", "baseline_record_count", "baseline_alias_count", "collapsed_duplicate_records", "bulk_record_count", "fallback_record_count", "fallback_used", "materialized_record_count", "coverage_complete", "duration_ms", } EXP_BASELINE_SUMMARY_KEYS = { "app_shell_reference_count", "apply_engine", "baseline_objects", "bulk_record_count", "collapsed_alias_count", "configured_alias_count", "conflicting_duplicate_match_group_count", "coverage_complete", "dry_run", "duplicate_match_group_count", "duration_ms", "fallback_record_count", "fallback_used", "lookup_record_count", "materialized_alias_count", "physical_record_count", } REFERENCE_SUMMARY_KEYS = { "dry_run", "apply_engine", "reference_data_objects", "reference_data_record_count", "reference_data_alias_count", "collapsed_duplicate_records", "layers", "unsupported", "coverage_complete", "apply_results", "duration_ms", } REFERENCE_APPLY_RESULT_KEYS = {"object", "rows", "aliases", "batches", "duration_ms"} SCRIPTED_SUMMARY_KEYS = { "schema_version", "format", "module_id", "dry_run", "apply_engine", "planned_counts", "applied_counts", "coverage_complete", "verification_status", "duration_ms", } PARTITIONED_SCRIPTED_SUMMARY_KEYS = SCRIPTED_SUMMARY_KEYS | {"outcome_counts"} REQUIRED_APPLY_VERIFICATION_MODULES = {"connected_apps"} class SeedDispatchError(RuntimeError): """Raised when dispatch cannot safely select or apply a Seed invocation.""" Runner = Callable[..., Any] def _adapter_environment( module_id: str, environment: Mapping[str, str], ) -> dict[str, str]: allowed_keys = ADAPTER_ENV_KEYS.get(module_id) if allowed_keys is None: raise SeedDispatchError( f"module {module_id} has no reviewed adapter environment contract" ) filtered = { "PATH": str(environment.get("PATH") or os.environ.get("PATH") or os.defpath) } for name in (*SAFE_PROCESS_ENV_KEYS, *allowed_keys): if name in environment: filtered[name] = str(environment[name]) return filtered def _dispatcher_environment( profile: str, environment: Mapping[str, str], ) -> dict[str, str]: profile_keys = set(PROFILE_ENV_KEYS.get(profile, ())) adapter_keys = { name for allowed_keys in ADAPTER_ENV_KEYS.values() for name in allowed_keys if name not in PROFILE_ONLY_ENV_KEYS } filtered = { "PATH": str(environment.get("PATH") or os.environ.get("PATH") or os.defpath) } for name in ( *SAFE_PROCESS_ENV_KEYS, *CONDITION_ENV_KEYS, *sorted(adapter_keys | profile_keys), ): if name in environment: filtered[name] = str(environment[name]) filtered["PYTHONDONTWRITEBYTECODE"] = "1" return filtered def _parse_bool(value: Any, label: str) -> bool: if isinstance(value, bool): return value normalized = str(value).strip().lower() if normalized in TRUE_VALUES: return True if normalized in FALSE_VALUES: return False raise SeedDispatchError(f"{label} must be a boolean value") def parse_argument_flags(values: list[str]) -> dict[str, bool]: flags: dict[str, bool] = {} for raw in values: name, separator, value = raw.partition("=") if not separator or FLAG_NAME_RE.fullmatch(name) is None: raise SeedDispatchError( f"argument flag must use canonical name=true|false syntax: {raw!r}" ) if name in flags: raise SeedDispatchError(f"duplicate argument flag: {name}") flags[name] = _parse_bool(value, f"argument flag {name}") return flags def evaluate_condition( condition: dict[str, Any], *, environment: Mapping[str, str], argument_flags: Mapping[str, bool], selected_stage: str | None, runtime_mode: str | None, ) -> tuple[bool, dict[str, Any]]: condition_type = condition["type"] if condition_type == "always": return True, {"type": "always", "matched": True} if condition_type in {"env_flag", "required_env_flag", "argument_flag"}: name = condition["name"] default = condition["default"] if condition_type in {"env_flag", "required_env_flag"}: raw_value = environment.get(name, "") actual = ( default if not raw_value.strip() else _parse_bool(raw_value, f"environment flag {name}") ) else: actual = argument_flags.get(name, default) matched = actual == condition["equals"] if condition_type == "required_env_flag" and not matched: raise SeedDispatchError( f"required environment flag {name} must be {condition['equals']}" ) return matched, { "type": condition_type, "name": name, "actual": actual, "expected": condition["equals"], "matched": matched, } if condition_type == "selected_stage": if selected_stage is None: raise SeedDispatchError("--selected-stage is required by the selected phase") matched = selected_stage == condition["name"] return matched, { "type": condition_type, "actual": selected_stage, "expected": condition["name"], "matched": matched, } if condition_type == "runtime_mode": if runtime_mode is None: raise SeedDispatchError("--runtime-mode is required by the selected phase") matched = runtime_mode == condition["name"] return matched, { "type": condition_type, "actual": runtime_mode, "expected": condition["name"], "matched": matched, } if condition_type == "all": results = [ evaluate_condition( item, environment=environment, argument_flags=argument_flags, selected_stage=selected_stage, runtime_mode=runtime_mode, ) for item in condition["conditions"] ] matched = all(result[0] for result in results) return matched, { "type": "all", "matched": matched, "conditions": [result[1] for result in results], } raise SeedDispatchError(f"unsupported condition type: {condition_type!r}") def _argument_flag_names(condition: dict[str, Any]) -> set[str]: if condition["type"] == "argument_flag": return {condition["name"]} if condition["type"] == "all": return set().union(*(_argument_flag_names(item) for item in condition["conditions"])) return set() def _resolve_empty_output_dir(raw_path: str, source_root: Path) -> Path: path = Path(raw_path).expanduser() if not path.is_absolute(): path = Path.cwd() / path resolved = path.resolve(strict=False) if resolved == source_root or source_root in resolved.parents: raise SeedDispatchError("output directory must be outside source root") if resolved.exists() and not resolved.is_dir(): raise SeedDispatchError(f"output directory is not a directory: {resolved}") resolved.mkdir(parents=True, exist_ok=True) resolved = resolved.resolve(strict=True) if resolved == source_root or source_root in resolved.parents: raise SeedDispatchError("output directory must be outside source root") if any(resolved.iterdir()): raise SeedDispatchError(f"output directory must be empty: {resolved}") return resolved def _write_json_atomic(path: Path, payload: dict[str, Any]) -> None: rendered = render_json(payload) handle = tempfile.NamedTemporaryFile(prefix=f".{path.name}.", dir=path.parent, delete=False) temp_path = Path(handle.name) try: with handle: handle.write(rendered) handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, path) except Exception: temp_path.unlink(missing_ok=True) raise def _verify_source_evidence(source_root: Path, expected: dict[str, Any]) -> None: try: current = DirectorySource(source_root) for entry in expected["entries"]: current.read_bytes(entry["declared_path"]) except SeedPlanError as exc: raise SeedDispatchError("source fingerprint changed after planning") from exc if current.evidence() != expected: raise SeedDispatchError("source fingerprint changed after planning") def _adapter_command( *, source_root: Path, output_dir: Path, invocation_id: str, module_id: str, module: dict[str, Any], selection: dict[str, Any], execution_mode: str, ) -> tuple[str, list[str]]: adapter_id = module.get("dispatcher_adapter") if not isinstance(adapter_id, str) or not adapter_id: raise SeedDispatchError(f"module {module_id} has no reviewed dispatcher adapter") if adapter_id in SCRIPTED_ADAPTER_IDS: adapter_paths = module["adapter_paths"] data_sources = module["data_sources"] if not adapter_paths: raise SeedDispatchError(f"module {module_id} has no scripted Seed entrypoint") entrypoint = adapter_paths[0] if not entrypoint.endswith(".py") or not entrypoint.startswith( ("scripts/", "deployments/offline/runtime/") ): raise SeedDispatchError( f"module {module_id} scripted Seed entrypoint is not reviewed: {entrypoint}" ) if not data_sources: raise SeedDispatchError(f"module {module_id} scripted Seed has no data source") command = [ "python3", str(source_root / entrypoint), "--source-root", str(source_root), "--module-id", module_id, "--invocation-id", invocation_id, "--output-dir", str(output_dir), "--selection-json", json.dumps(selection, sort_keys=True, separators=(",", ":")), ] for data_source in data_sources: command.extend(("--data-source", str(source_root / data_source))) if execution_mode == "dry-run": command.append("--dry-run") return adapter_id, command spec = ADAPTERS.get(adapter_id) if spec is None: raise SeedDispatchError(f"module {module_id} uses unsupported adapter: {adapter_id}") if spec["module_id"] != module_id: raise SeedDispatchError( f"adapter {adapter_id} is not registered for module {module_id}" ) entrypoint = spec["entrypoint"] if adapter_id == "deploy_baseline_v1": seed_file = spec["seed_file"] if entrypoint not in module["adapter_paths"]: raise SeedDispatchError( f"adapter {adapter_id} entrypoint is not declared by module {module_id}" ) if seed_file not in module["data_sources"]: raise SeedDispatchError( f"adapter {adapter_id} seed file is not declared by module {module_id}" ) if invocation_id not in spec["invocation_ids"]: raise SeedDispatchError( f"adapter {adapter_id} has no reviewed scope for invocation {invocation_id}" ) if selection != {}: raise SeedDispatchError(f"adapter {adapter_id} does not accept a selection") command = [ "python3", str(source_root / entrypoint), "--source-root", str(source_root), "--seed-file", str(source_root / seed_file), "--output-dir", str(output_dir), "--target", spec["invocation_targets"][invocation_id], ] if execution_mode == "dry-run": command.append("--dry-run") return adapter_id, command if adapter_id == "exp_baseline_v1": seed_file = spec["seed_file"] if entrypoint not in module["adapter_paths"]: raise SeedDispatchError( f"adapter {adapter_id} entrypoint is not declared by module {module_id}" ) if seed_file not in module["data_sources"]: raise SeedDispatchError( f"adapter {adapter_id} seed file is not declared by module {module_id}" ) if invocation_id not in spec["invocation_ids"]: raise SeedDispatchError( f"adapter {adapter_id} has no reviewed scope for invocation {invocation_id}" ) if selection != {}: raise SeedDispatchError(f"adapter {adapter_id} does not accept a selection") command = [ "python3", str(source_root / entrypoint), "--source-root", str(source_root), "--seed-file", str(source_root / seed_file), "--output-dir", str(output_dir), ] if execution_mode == "dry-run": command.append("--dry-run") return adapter_id, command manifest = spec["manifest"] if entrypoint not in module["adapter_paths"]: raise SeedDispatchError( f"adapter {adapter_id} entrypoint is not declared by module {module_id}" ) if manifest not in module["authoritative_paths"]: raise SeedDispatchError( f"adapter {adapter_id} manifest is not declared by module {module_id}" ) if set(selection) != {"objects"}: raise SeedDispatchError(f"adapter {adapter_id} requires an objects selection") reviewed_selection = spec["invocation_selections"].get(invocation_id) if reviewed_selection is None: raise SeedDispatchError( f"adapter {adapter_id} has no reviewed scope for invocation {invocation_id}" ) if selection["objects"] != list(reviewed_selection["objects"]): raise SeedDispatchError( f"adapter {adapter_id} invocation {invocation_id} selection does not match " "the reviewed scope" ) command = [ "bash", str(source_root / entrypoint), "--manifest-file", str(source_root / manifest), "--output-dir", str(output_dir), "--objects", ",".join(selection["objects"]), ] if execution_mode == "dry-run": command.append("--dry-run") return adapter_id, command def _load_reference_data_summary( path: Path, *, expected_objects: list[str], execution_mode: str, ) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise SeedDispatchError("adapter did not emit a valid summary") from exc if not isinstance(payload, dict): raise SeedDispatchError("adapter summary must be an object") if set(payload) != REFERENCE_SUMMARY_KEYS: raise SeedDispatchError("adapter summary fields do not match the reviewed contract") if payload.get("reference_data_objects") != expected_objects: raise SeedDispatchError("adapter summary object selection does not match the plan") if payload.get("dry_run") is not (execution_mode == "dry-run"): raise SeedDispatchError("adapter summary execution mode does not match the dispatcher") if payload.get("coverage_complete") is not True or payload.get("unsupported") != []: raise SeedDispatchError("adapter summary does not prove complete supported coverage") if payload.get("apply_engine") != "mysql-cli": raise SeedDispatchError("adapter summary apply engine is not reviewed") for name in ( "reference_data_record_count", "reference_data_alias_count", "collapsed_duplicate_records", "duration_ms", ): value = payload[name] if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise SeedDispatchError(f"adapter summary {name} must be a non-negative integer") if ( payload["reference_data_alias_count"] - payload["collapsed_duplicate_records"] != payload["reference_data_record_count"] ): raise SeedDispatchError( "adapter summary record, alias, and collapsed counts are inconsistent" ) layers = payload["layers"] if not isinstance(layers, list) or not all( isinstance(layer, list) and all(isinstance(item, str) and item for item in layer) for layer in layers ): raise SeedDispatchError("adapter summary layers are invalid") layered_objects = [item for layer in layers for item in layer] if ( len(layered_objects) != len(set(layered_objects)) or set(layered_objects) != set(expected_objects) ): raise SeedDispatchError("adapter summary layers do not cover the object selection") apply_results = payload["apply_results"] if not isinstance(apply_results, list) or not all( isinstance(result, dict) and set(result) == REFERENCE_APPLY_RESULT_KEYS for result in apply_results ): raise SeedDispatchError("adapter summary apply results are invalid") if execution_mode == "dry-run" and apply_results: raise SeedDispatchError("dry-run adapter summary must not contain apply results") applied_objects: list[str] = [] for result in apply_results: if not isinstance(result["object"], str): raise SeedDispatchError("adapter summary apply result object is invalid") applied_objects.append(result["object"]) if result["object"] not in expected_objects: raise SeedDispatchError("adapter summary apply result is outside the selection") for name in ("rows", "aliases", "batches", "duration_ms"): value = result[name] if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise SeedDispatchError( f"adapter summary apply result {name} must be a non-negative integer" ) if execution_mode == "apply" and ( len(applied_objects) != len(set(applied_objects)) or set(applied_objects) != set(expected_objects) ): raise SeedDispatchError("adapter summary apply results do not cover the selection") if execution_mode == "apply" and sum( result["rows"] for result in apply_results ) != payload["reference_data_record_count"]: raise SeedDispatchError( "adapter summary apply row count does not match the record count" ) if execution_mode == "apply" and sum( result["aliases"] for result in apply_results ) != payload["reference_data_alias_count"]: raise SeedDispatchError( "adapter summary apply alias count does not match the alias count" ) return {key: copy.deepcopy(payload[key]) for key in sorted(REFERENCE_SUMMARY_KEYS)} def _load_deploy_baseline_summary( path: Path, *, module: dict[str, Any], expected_objects: tuple[str, ...], execution_mode: str, ) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise SeedDispatchError("adapter did not emit a valid summary") from exc if not isinstance(payload, dict): raise SeedDispatchError("adapter summary must be an object") if set(payload) != BASELINE_SUMMARY_KEYS: raise SeedDispatchError("adapter summary fields do not match the reviewed contract") if payload.get("dry_run") is not (execution_mode == "dry-run"): raise SeedDispatchError("adapter summary execution mode does not match the dispatcher") if payload.get("coverage_complete") is not True: raise SeedDispatchError("adapter summary does not prove complete baseline coverage") if payload.get("fallback_used") is not False: raise SeedDispatchError("adapter summary must not use retired REST fallback") expected_engine = "bulk-plan" if execution_mode == "dry-run" else "bulk" if payload.get("apply_engine") != expected_engine: raise SeedDispatchError("adapter summary apply engine is inconsistent") objects = payload.get("baseline_objects") expected_object_count = module["declared_stats"]["objects"] if ( not isinstance(objects, list) or objects != list(expected_objects) or len(objects) != expected_object_count or len(objects) != len(set(objects)) or not all(isinstance(item, str) and item for item in objects) ): raise SeedDispatchError("adapter summary baseline objects are invalid") for name in ( "baseline_record_count", "baseline_alias_count", "collapsed_duplicate_records", "bulk_record_count", "fallback_record_count", "materialized_record_count", "duration_ms", ): value = payload[name] if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise SeedDispatchError(f"adapter summary {name} must be a non-negative integer") if payload["baseline_record_count"] != module["declared_stats"]["baseline_records"]: raise SeedDispatchError("adapter summary baseline record count does not match the Catalog") if ( payload["baseline_alias_count"] - payload["collapsed_duplicate_records"] != payload["baseline_record_count"] ): raise SeedDispatchError( "adapter summary record, alias, and collapsed counts are inconsistent" ) if ( payload["bulk_record_count"] + payload["fallback_record_count"] != payload["baseline_record_count"] ): raise SeedDispatchError("adapter summary bulk and fallback counts are inconsistent") if payload["fallback_record_count"] != 0: raise SeedDispatchError("adapter summary must have complete bulk coverage") expected_materialized = 0 if execution_mode == "dry-run" else payload["baseline_record_count"] if payload["materialized_record_count"] != expected_materialized: raise SeedDispatchError("adapter summary materialized record count is inconsistent") return {key: copy.deepcopy(payload[key]) for key in sorted(BASELINE_SUMMARY_KEYS)} def _load_exp_baseline_summary( path: Path, *, module: dict[str, Any], expected_objects: tuple[str, ...], execution_mode: str, ) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise SeedDispatchError("adapter did not emit a valid summary") from exc if not isinstance(payload, dict): raise SeedDispatchError("adapter summary must be an object") if set(payload) != EXP_BASELINE_SUMMARY_KEYS: raise SeedDispatchError( "EXP baseline adapter summary fields do not match the reviewed contract" ) if payload.get("dry_run") is not (execution_mode == "dry-run"): raise SeedDispatchError( "EXP baseline adapter summary execution mode does not match the dispatcher" ) if payload.get("coverage_complete") is not True: raise SeedDispatchError( "EXP baseline adapter summary does not prove complete coverage" ) if payload.get("fallback_used") is not False: raise SeedDispatchError( "EXP baseline adapter must not use retired REST fallback" ) expected_engine = "bulk-plan" if execution_mode == "dry-run" else "bulk" if payload.get("apply_engine") != expected_engine: raise SeedDispatchError("EXP baseline adapter summary apply engine is inconsistent") objects = payload.get("baseline_objects") if ( not isinstance(objects, list) or objects != list(expected_objects) or len(objects) != module["declared_stats"]["objects"] or len(objects) != len(set(objects)) or not all(isinstance(item, str) and item for item in objects) ): raise SeedDispatchError("EXP baseline adapter summary objects are invalid") integer_fields = ( "app_shell_reference_count", "bulk_record_count", "collapsed_alias_count", "configured_alias_count", "conflicting_duplicate_match_group_count", "duplicate_match_group_count", "duration_ms", "fallback_record_count", "lookup_record_count", "materialized_alias_count", "physical_record_count", ) for name in integer_fields: value = payload[name] if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise SeedDispatchError( f"EXP baseline adapter summary {name} must be a non-negative integer" ) expected_counts = { "app_shell_reference_count": module["declared_stats"]["app_shells"], "collapsed_alias_count": module["declared_stats"]["collapsed_aliases"], "configured_alias_count": module["declared_stats"]["baseline_records"], "conflicting_duplicate_match_group_count": module["declared_stats"][ "conflicting_duplicate_match_groups" ], "duplicate_match_group_count": module["declared_stats"][ "duplicate_match_groups" ], "lookup_record_count": module["declared_stats"]["lookup_records"], "physical_record_count": module["declared_stats"]["physical_records"], } for name, expected in expected_counts.items(): if payload[name] != expected: raise SeedDispatchError( f"EXP baseline adapter summary {name} does not match the Catalog" ) if ( payload["configured_alias_count"] - payload["collapsed_alias_count"] != payload["physical_record_count"] ): raise SeedDispatchError( "EXP baseline adapter summary alias and physical counts are inconsistent" ) if ( payload["bulk_record_count"] + payload["fallback_record_count"] != payload["physical_record_count"] ): raise SeedDispatchError( "EXP baseline adapter summary bulk and fallback counts are inconsistent" ) if ( payload["conflicting_duplicate_match_group_count"] > payload["duplicate_match_group_count"] or payload["duplicate_match_group_count"] > payload["collapsed_alias_count"] ): raise SeedDispatchError( "EXP baseline adapter summary duplicate counts are inconsistent" ) if payload["fallback_record_count"] != 0: raise SeedDispatchError( "EXP baseline adapter summary must have complete bulk coverage" ) expected_materialized_aliases = ( 0 if execution_mode == "dry-run" else payload["configured_alias_count"] ) if payload["materialized_alias_count"] != expected_materialized_aliases: raise SeedDispatchError( "EXP baseline adapter summary materialized alias count is inconsistent" ) return { key: copy.deepcopy(payload[key]) for key in sorted(EXP_BASELINE_SUMMARY_KEYS) } def _load_scripted_seed_summary( path: Path, *, module_id: str, module: dict[str, Any], execution_mode: str, partitioned: bool = False, ) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise SeedDispatchError("adapter did not emit a valid summary") from exc if not isinstance(payload, dict): raise SeedDispatchError("adapter summary must be an object") summary_keys = ( PARTITIONED_SCRIPTED_SUMMARY_KEYS if partitioned else SCRIPTED_SUMMARY_KEYS ) if set(payload) != summary_keys: raise SeedDispatchError("scripted adapter summary fields do not match the reviewed contract") if payload.get("schema_version") != SCRIPTED_SUMMARY_SCHEMA_VERSION: raise SeedDispatchError("scripted adapter summary schema version is unsupported") expected_format = ( PARTITIONED_SCRIPTED_SUMMARY_FORMAT if partitioned else SCRIPTED_SUMMARY_FORMAT ) if payload.get("format") != expected_format: raise SeedDispatchError("scripted adapter summary format is unsupported") if payload.get("module_id") != module_id: raise SeedDispatchError("scripted adapter summary module does not match the plan") dry_run = execution_mode == "dry-run" if payload.get("dry_run") is not dry_run: raise SeedDispatchError("scripted adapter summary execution mode does not match the dispatcher") apply_engine = payload.get("apply_engine") if not isinstance(apply_engine, str) or SCRIPTED_ENGINE_RE.fullmatch(apply_engine) is None: raise SeedDispatchError("scripted adapter summary apply engine is invalid") planned_counts = payload.get("planned_counts") if planned_counts != module["declared_stats"]: raise SeedDispatchError("scripted adapter planned counts do not match the Catalog") applied_counts = payload.get("applied_counts") expected_applied_counts = ( {name: 0 for name in planned_counts} if dry_run else planned_counts ) if applied_counts != expected_applied_counts: raise SeedDispatchError("scripted adapter applied counts are inconsistent") for label, counts in (("planned", planned_counts), ("applied", applied_counts)): if not isinstance(counts, dict) or any( isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in counts.values() ): raise SeedDispatchError( f"scripted adapter summary {label} counts must be non-negative integers" ) if partitioned: outcome_counts = payload.get("outcome_counts") if len(planned_counts) != 1: raise SeedDispatchError( "partitioned scripted adapter requires one declared count" ) if not isinstance(outcome_counts, dict) or not outcome_counts or any( not isinstance(name, str) or not name or isinstance(value, bool) or not isinstance(value, int) or value < 0 for name, value in outcome_counts.items() ): raise SeedDispatchError( "partitioned scripted adapter outcome counts must be non-negative integers" ) expected_outcome_total = 0 if dry_run else sum(applied_counts.values()) if sum(outcome_counts.values()) != expected_outcome_total: raise SeedDispatchError( "partitioned scripted adapter outcome counts do not cover applied counts" ) if payload.get("coverage_complete") is not True: raise SeedDispatchError("scripted adapter summary does not prove complete coverage") if dry_run: expected_verification = {"planned"} elif module_id in REQUIRED_APPLY_VERIFICATION_MODULES: expected_verification = {"passed"} else: expected_verification = {"passed", "skipped"} if payload.get("verification_status") not in expected_verification: raise SeedDispatchError("scripted adapter summary verification status is invalid") duration_ms = payload.get("duration_ms") if isinstance(duration_ms, bool) or not isinstance(duration_ms, int) or duration_ms < 0: raise SeedDispatchError("scripted adapter summary duration_ms must be non-negative") return {key: copy.deepcopy(payload[key]) for key in sorted(summary_keys)} def _load_adapter_summary( path: Path, *, adapter_id: str, module_id: str, module: dict[str, Any], selection: dict[str, Any], execution_mode: str, ) -> dict[str, Any]: if adapter_id in SCRIPTED_ADAPTER_IDS: return _load_scripted_seed_summary( path, module_id=module_id, module=module, execution_mode=execution_mode, partitioned=adapter_id == PARTITIONED_SCRIPTED_ADAPTER_ID, ) if adapter_id == "deploy_baseline_v1": return _load_deploy_baseline_summary( path, module=module, expected_objects=ADAPTERS[adapter_id]["objects"], execution_mode=execution_mode, ) if adapter_id == "exp_baseline_v1": return _load_exp_baseline_summary( path, module=module, expected_objects=ADAPTERS[adapter_id]["objects"], execution_mode=execution_mode, ) if adapter_id == "pack_reference_data_v1": return _load_reference_data_summary( path, expected_objects=selection["objects"], execution_mode=execution_mode, ) raise SeedDispatchError(f"unsupported adapter summary contract: {adapter_id}") def dispatch_deploy_seed( *, source_root: Path | str, profile: str, phase: str, execution_mode: str, output_dir: str, catalog_relpath: str = CATALOG_RELPATH, environment: Mapping[str, str] | None = None, argument_flags: Mapping[str, bool] | None = None, selected_stage: str | None = None, runtime_mode: str | None = None, runner: Runner = subprocess.run, ) -> tuple[dict[str, Any], Path]: if execution_mode not in {"dry-run", "apply"}: raise SeedDispatchError(f"unsupported execution mode: {execution_mode!r}") environment = os.environ if environment is None else environment argument_flags = {} if argument_flags is None else argument_flags for name, value in argument_flags.items(): if FLAG_NAME_RE.fullmatch(name) is None or not isinstance(value, bool): raise SeedDispatchError("argument flags must map canonical names to booleans") try: plan, inventory, source = plan_deploy_seed( source_root=source_root, profile_name=profile, catalog_relpath=catalog_relpath, ) except SeedPlanError as exc: raise SeedDispatchError(str(exc)) from exc resolved_output_dir = _resolve_empty_output_dir(output_dir, source.root) invocations = [ invocation for invocation in plan["invocations"] if invocation["phase"] == phase and plan["modules"][invocation["module_id"]]["execution_owner"] == "seed_dispatcher" ] if not invocations: raise SeedDispatchError( f"profile {profile} phase {phase} has no seed_dispatcher invocations" ) declared_argument_flags = set().union( *(_argument_flag_names(invocation["condition"]) for invocation in invocations) ) unknown_argument_flags = sorted(set(argument_flags) - declared_argument_flags) if unknown_argument_flags: raise SeedDispatchError( f"argument flags are not declared by profile {profile} phase {phase}: " f"{unknown_argument_flags}" ) evidence_rows: list[dict[str, Any]] = [] prepared: list[tuple[dict[str, Any], dict[str, Any], Path, str, list[str]]] = [] for invocation in invocations: module = plan["modules"][invocation["module_id"]] matched, condition_evidence = evaluate_condition( invocation["condition"], environment=environment, argument_flags=argument_flags, selected_stage=selected_stage, runtime_mode=runtime_mode, ) row = { "invocation_id": invocation["invocation_id"], "module_id": invocation["module_id"], "order": invocation["order"], "selection": copy.deepcopy(invocation["selection"]), "condition": condition_evidence, "adapter_id": module.get("dispatcher_adapter"), "adapter_output_dir": None, "status": "skipped", "summary": None, } evidence_rows.append(row) if not matched: continue adapter_relpath = f"adapters/{invocation['invocation_id']}" adapter_output_dir = resolved_output_dir / adapter_relpath adapter_id, command = _adapter_command( source_root=source.root, output_dir=adapter_output_dir, invocation_id=invocation["invocation_id"], module_id=invocation["module_id"], module=module, selection=invocation["selection"], execution_mode=execution_mode, ) row["adapter_id"] = adapter_id row["adapter_output_dir"] = adapter_relpath prepared.append((row, module, adapter_output_dir, adapter_id, command)) write_artifact_pair( str(resolved_output_dir / PLAN_FILENAME), str(resolved_output_dir / INVENTORY_FILENAME), source.root, render_json(plan), render_json(inventory), ) evidence = { "schema_version": APPLY_EVIDENCE_SCHEMA_VERSION, "format": APPLY_EVIDENCE_FORMAT, "catalog_version": plan["catalog_version"], "profile": profile, "phase": phase, "execution_mode": execution_mode, "status": "passed", "source_fingerprint_sha256": plan["source"]["source_fingerprint_sha256"], "plan_file": PLAN_FILENAME, "inventory_file": INVENTORY_FILENAME, "invocations": evidence_rows, "failure": None, } evidence_path = resolved_output_dir / EVIDENCE_FILENAME try: _verify_source_evidence(source.root, plan["source"]) except SeedDispatchError: evidence["status"] = "failed" evidence["failure"] = {"code": "source_drift_before_execution"} _write_json_atomic(evidence_path, evidence) raise for row, module, adapter_output_dir, adapter_id, command in prepared: try: _verify_source_evidence(source.root, plan["source"]) except SeedDispatchError: row["status"] = "failed" evidence["status"] = "failed" evidence["failure"] = { "code": "source_drift_before_execution", "invocation_id": row["invocation_id"], } _write_json_atomic(evidence_path, evidence) raise try: completed = runner( command, cwd=source.root, env=_adapter_environment(row["module_id"], environment), check=False, ) except OSError as exc: row["status"] = "failed" evidence["status"] = "failed" evidence["failure"] = { "code": "adapter_start", "invocation_id": row["invocation_id"], } _write_json_atomic(evidence_path, evidence) raise SeedDispatchError( f"unable to start adapter for {row['invocation_id']}" ) from exc if completed.returncode != 0: row["status"] = "failed" evidence["status"] = "failed" evidence["failure"] = { "code": "adapter_exit", "invocation_id": row["invocation_id"], "exit_code": completed.returncode, } _write_json_atomic(evidence_path, evidence) raise SeedDispatchError( f"adapter failed for {row['invocation_id']}: exit={completed.returncode}" ) try: summary = _load_adapter_summary( adapter_output_dir / "summary.json", adapter_id=adapter_id, module_id=row["module_id"], module=module, selection=row["selection"], execution_mode=execution_mode, ) except SeedDispatchError: row["status"] = "failed" evidence["status"] = "failed" evidence["failure"] = { "code": "invalid_adapter_summary", "invocation_id": row["invocation_id"], } _write_json_atomic(evidence_path, evidence) raise row["summary"] = summary row["status"] = "planned" if execution_mode == "dry-run" else "applied" try: _verify_source_evidence(source.root, plan["source"]) except SeedDispatchError: evidence["status"] = "failed" evidence["failure"] = {"code": "source_drift_after_execution"} _write_json_atomic(evidence_path, evidence) raise _write_json_atomic(evidence_path, evidence) return evidence, evidence_path def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Plan and apply one deploy Seed Catalog phase through reviewed adapters." ) parser.add_argument("--source-root", required=True) parser.add_argument("--profile", required=True, choices=PROFILES) parser.add_argument("--phase", required=True) parser.add_argument("--execution-mode", required=True, choices=("dry-run", "apply")) parser.add_argument("--output-dir", required=True) parser.add_argument("--catalog-relpath", default=CATALOG_RELPATH) parser.add_argument("--argument-flag", action="append", default=[]) parser.add_argument("--selected-stage") parser.add_argument("--runtime-mode") return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: flags = parse_argument_flags(args.argument_flag) environment = _dispatcher_environment(args.profile, os.environ) os.environ.clear() os.environ.update(environment) evidence, evidence_path = dispatch_deploy_seed( source_root=args.source_root, profile=args.profile, phase=args.phase, execution_mode=args.execution_mode, output_dir=args.output_dir, catalog_relpath=args.catalog_relpath, environment=environment, argument_flags=flags, selected_stage=args.selected_stage, runtime_mode=args.runtime_mode, ) except SeedDispatchError as exc: print(f"deploy Seed dispatch failed: {exc}", file=sys.stderr) return 2 print( json.dumps( { "profile": evidence["profile"], "phase": evidence["phase"], "execution_mode": evidence["execution_mode"], "status": evidence["status"], "source_fingerprint_sha256": evidence["source_fingerprint_sha256"], "evidence_file": str(evidence_path), }, sort_keys=True, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())