#!/usr/bin/env python3 """Validate the deploy Seed catalog and emit deterministic read-only artifacts.""" from __future__ import annotations import argparse import ast import copy import hashlib import json import os from pathlib import Path, PurePosixPath import re import shlex import sys import tempfile from typing import Any CATALOG_SCHEMA_VERSION = 3 ARTIFACT_SCHEMA_VERSION = 3 SOURCE_FORMAT = "deploy-seed-source-v1" PROFILES = ("local", "ci", "test", "exp", "staging", "production") TOP_LEVEL_KEYS = { "schema_version", "catalog_version", "modules", "profiles", "expected_callsites", "excluded_mutations", } MODULE_KEYS = { "description", "dataset_class", "execution_owner", "dispatcher_adapter", "data_sources", "adapter_paths", "authoritative_paths", "selection_schema", "declared_stats", } INVOCATION_KEYS = { "invocation_id", "module_id", "phase", "order", "condition", "selection", } CALLSITE_KEYS = { "callsite_id", "path", "marker", "expected_count", "module_id", "exclusion_id", "invocation_ids", } CALLSITE_OPTIONAL_KEYS = { "expected_python_calls", "required_after_invocation_id", "shell_command", } EXCLUSION_KEYS = { "exclusion_id", "description", "owner", "reason", "source_paths", } EXECUTION_OWNERS = {"seed_dispatcher", "deploy_pipeline"} STAGE_CONTRACTS = { "staging": ("deployments/offline/runtime/deploy_staging_env.py", "STAGING_STAGES"), } RUNTIME_MODE_CONTRACTS = { "staging": {"runtime-env"}, } IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9_]*$") QUALIFIED_IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9_.-]*$") class SeedPlanError(RuntimeError): """Raised when the catalog, source directory, or artifact contract is invalid.""" def _expect_exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None: actual = set(value) if actual != expected: missing = sorted(expected - actual) unknown = sorted(actual - expected) raise SeedPlanError(f"{label} keys mismatch: missing={missing} unknown={unknown}") def _expect_nonempty_string(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise SeedPlanError(f"{label} must be a non-empty string") return value def _expect_identifier(value: Any, label: str, *, qualified: bool = False) -> str: identifier = _expect_nonempty_string(value, label) pattern = QUALIFIED_IDENTIFIER_RE if qualified else IDENTIFIER_RE if pattern.fullmatch(identifier) is None: raise SeedPlanError(f"{label} is not a canonical identifier: {identifier!r}") return identifier def _reject_duplicate_object_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise SeedPlanError(f"JSON contains duplicate key: {key!r}") result[key] = value return result def validate_relpath(value: Any) -> str: """Return one canonical source-relative POSIX file path or fail closed.""" if not isinstance(value, str) or not value: raise SeedPlanError("source-relative path must be a non-empty string") if any(ord(character) < 32 or ord(character) == 127 for character in value): raise SeedPlanError(f"source-relative path contains a control character: {value!r}") if "\\" in value or ":" in value: raise SeedPlanError(f"source-relative path uses a forbidden character: {value!r}") if value.startswith("/") or value.endswith("/") or "//" in value: raise SeedPlanError(f"source-relative path is not canonical: {value!r}") parts = value.split("/") if any(part in {"", ".", ".."} for part in parts): raise SeedPlanError(f"source-relative path contains a forbidden segment: {value!r}") path = PurePosixPath(value) if path.is_absolute() or path.as_posix() != value: raise SeedPlanError(f"source-relative path is not canonical POSIX syntax: {value!r}") return value class DirectorySource: """Read and fingerprint files from one caller-prepared directory namespace.""" def __init__(self, root: Path | str) -> None: try: resolved = Path(root).expanduser().resolve(strict=True) except OSError as exc: raise SeedPlanError(f"source root does not exist: {root}") from exc if not resolved.is_dir(): raise SeedPlanError(f"source root is not a directory: {root}") self.root = resolved self._raw_bytes: dict[str, bytes] = {} self._entries: dict[str, dict[str, str]] = {} def read_bytes(self, relpath: Any) -> bytes: declared_path = validate_relpath(relpath) if declared_path in self._raw_bytes: return self._raw_bytes[declared_path] candidate = self.root.joinpath(*declared_path.split("/")) try: resolved = candidate.resolve(strict=True) except OSError as exc: raise SeedPlanError(f"source file does not exist: {declared_path}") from exc try: resolved_path = resolved.relative_to(self.root).as_posix() except ValueError as exc: raise SeedPlanError(f"source file escapes source root: {declared_path}") from exc if not resolved.is_file(): raise SeedPlanError(f"source path is not a regular file: {declared_path}") try: raw = resolved.read_bytes() except OSError as exc: raise SeedPlanError(f"unable to read source file: {declared_path}") from exc self._raw_bytes[declared_path] = raw self._entries[declared_path] = { "declared_path": declared_path, "resolved_path": resolved_path, "content_sha256": hashlib.sha256(raw).hexdigest(), } return raw def read_text(self, relpath: Any) -> str: declared_path = validate_relpath(relpath) try: return self.read_bytes(declared_path).decode("utf-8") except UnicodeDecodeError as exc: raise SeedPlanError(f"source file is not UTF-8 text: {declared_path}") from exc def evidence(self) -> dict[str, Any]: entries = [copy.deepcopy(self._entries[path]) for path in sorted(self._entries)] canonical_payload = { "format": SOURCE_FORMAT, "entries": entries, } canonical_bytes = json.dumps( canonical_payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return { **canonical_payload, "source_fingerprint_sha256": hashlib.sha256(canonical_bytes).hexdigest(), } def _load_json_object(source: DirectorySource, relpath: str) -> dict[str, Any]: try: value = json.loads( source.read_text(relpath), object_pairs_hook=_reject_duplicate_object_pairs, ) except json.JSONDecodeError as exc: raise SeedPlanError(f"invalid JSON in {relpath}: {exc}") from exc if not isinstance(value, dict): raise SeedPlanError(f"{relpath} must contain a JSON object") return value def _validate_path_list(value: Any, label: str, source: DirectorySource) -> list[str]: if not isinstance(value, list): raise SeedPlanError(f"{label} must be an array") paths: list[str] = [] for index, item in enumerate(value): path = validate_relpath(item) if path in paths: raise SeedPlanError(f"{label} contains duplicate path: {path}") source.read_bytes(path) paths.append(path) return paths def _validate_selection_schema(value: Any, label: str) -> dict[str, dict[str, Any]]: if not isinstance(value, dict): raise SeedPlanError(f"{label} must be an object") result: dict[str, dict[str, Any]] = {} for field, raw_schema in value.items(): _expect_identifier(field, f"{label} field") if not isinstance(raw_schema, dict): raise SeedPlanError(f"{label}.{field} must be an object") _expect_exact_keys(raw_schema, {"type", "allowlist"}, f"{label}.{field}") if raw_schema["type"] != "string_list": raise SeedPlanError(f"{label}.{field}.type must be string_list") allowlist = raw_schema["allowlist"] if not isinstance(allowlist, list) or not allowlist: raise SeedPlanError(f"{label}.{field}.allowlist must be a non-empty array") if not all(isinstance(item, str) and item for item in allowlist): raise SeedPlanError(f"{label}.{field}.allowlist must contain non-empty strings") if len(set(allowlist)) != len(allowlist): raise SeedPlanError(f"{label}.{field}.allowlist contains duplicates") result[field] = copy.deepcopy(raw_schema) return result def _callable_name(value: ast.expr) -> str | None: if isinstance(value, ast.Name): return value.id if isinstance(value, ast.Attribute): return value.attr return None def _assigned_names(value: ast.expr) -> set[str]: if isinstance(value, ast.Name): return {value.id} if isinstance(value, (ast.List, ast.Tuple)): names: set[str] = set() for item in value.elts: names.update(_assigned_names(item)) return names return set() def _count_python_calls(tree: ast.Module, target_symbol: str) -> int: aliases = {target_symbol} assignments: list[tuple[set[str], str]] = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): for imported in node.names: if imported.name == target_symbol: aliases.add(imported.asname or imported.name) continue if isinstance(node, ast.Assign): source_name = _callable_name(node.value) if source_name is not None: target_names: set[str] = set() for target in node.targets: target_names.update(_assigned_names(target)) assignments.append((target_names, source_name)) continue if isinstance(node, ast.AnnAssign) and node.value is not None: source_name = _callable_name(node.value) if source_name is not None: assignments.append((_assigned_names(node.target), source_name)) changed = True while changed: changed = False for target_names, source_name in assignments: if source_name not in aliases: continue new_aliases = target_names - aliases if new_aliases: aliases.update(new_aliases) changed = True count = 0 for node in ast.walk(tree): if not isinstance(node, ast.Call): continue if isinstance(node.func, ast.Name) and node.func.id in aliases: count += 1 elif isinstance(node.func, ast.Attribute) and node.func.attr == target_symbol: count += 1 return count def _python_call_symbol(path: str, marker: str) -> str | None: if not path.endswith(".py"): return None matches = re.findall(r"([A-Za-z_][A-Za-z0-9_]*)\s*\(", marker) return matches[-1] if matches else None def _parse_python_source(source: DirectorySource, path: str) -> ast.Module: try: return ast.parse(source.read_text(path), filename=path) except SyntaxError as exc: raise SeedPlanError(f"cannot parse Python callsite source {path}: {exc}") from exc def _blank_shell_heredoc_bodies(text: str) -> str: result = list(text) quote = "" heredocs: list[tuple[str, bool]] = [] offset = 0 for line in text.splitlines(keepends=True): body = line[:-1] if line.endswith("\n") else line if heredocs: delimiter, strip_tabs = heredocs[0] candidate = body.lstrip("\t") if strip_tabs else body if candidate == delimiter: heredocs.pop(0) for index, character in enumerate(line): if character != "\n": result[offset + index] = " " offset += len(line) continue index = 0 while index < len(line): character = line[index] if quote: if character == quote: quote = "" elif quote == '"' and character == "\\" and index + 1 < len(line): index += 1 index += 1 continue if character in {"'", '"'}: quote = character index += 1 continue if character == "\\" and index + 1 < len(line): index += 2 continue if character == "#" and ( index == 0 or line[index - 1].isspace() or line[index - 1] in ";|&()" ): break if line.startswith("<<<", index): index += 3 continue if line.startswith("<<", index): cursor = index + 2 strip_tabs = False if cursor < len(line) and line[cursor] == "-": strip_tabs = True cursor += 1 while cursor < len(line) and line[cursor] in " \t": cursor += 1 delimiter = "" if cursor < len(line) and line[cursor] in {"'", '"'}: delimiter_quote = line[cursor] cursor += 1 start = cursor while cursor < len(line) and line[cursor] != delimiter_quote: cursor += 1 delimiter = line[start:cursor] else: start = cursor while cursor < len(line) and not line[cursor].isspace() and line[cursor] not in ";|&()<>": cursor += 1 delimiter = line[start:cursor].replace("\\", "") if delimiter: heredocs.append((delimiter, strip_tabs)) index += 1 offset += len(line) return "".join(result) def _collapse_shell_line_continuations(text: str) -> str: result: list[str] = [] for line in text.splitlines(keepends=True): if not line.endswith("\\\n"): result.append(line) continue quote = "" comment = False index = 0 content = line[:-2] while index < len(content): character = content[index] if comment: break if quote: if character == quote: quote = "" index += 1 continue if quote == '"' and character == "\\" and index + 1 < len(content): index += 2 continue index += 1 continue if character in {"'", '"'}: quote = character elif character == "#" and ( index == 0 or content[index - 1].isspace() or content[index - 1] in ";|&()" ): comment = True index += 1 if comment or quote == "'": result.append(line) else: result.append(line[:-2] + " ") return "".join(result) def _expand_shell_punctuation(token: str) -> list[str]: if not token or any(character not in "();<>|&\n" for character in token): return [token] operators = ( ";;&", "&>>", "<<<", ";;", ";&", "&&", "||", "|&", ">>", "<<", "<>", ">&", "<&", "&>", ">|", "(", ")", ";", "|", "&", "<", ">", ) result: list[str] = [] index = 0 while index < len(token): if token[index] == "\n": result.append(";") index += 1 continue for operator in operators: if token.startswith(operator, index): result.append(operator) index += len(operator) break else: result.append(token[index]) index += 1 return result def _shell_tokens(text: str) -> list[str]: lexer = shlex.shlex( _collapse_shell_line_continuations(_blank_shell_heredoc_bodies(text)), posix=True, punctuation_chars="();<>|&\n", ) lexer.whitespace = " \t\r" lexer.whitespace_split = True lexer.commenters = "#" tokens: list[str] = [] try: for token in lexer: tokens.extend(_expand_shell_punctuation(token)) except ValueError as exc: raise SeedPlanError(f"cannot tokenize shell callsite source: {exc}") from exc return tokens def _is_shell_assignment(token: str) -> bool: return re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*=.*", token) is not None def _normalize_shell_command(segment: list[str]) -> list[str]: words = list(segment) if len(words) >= 3 and words[1:3] == ["(", ")"]: words = words[4:] if len(words) > 3 and words[3] == "{" else words[3:] elif words[:1] == ["function"]: try: body_start = words.index("{") + 1 except ValueError: return [] words = words[body_start:] reserved_prefixes = {"!", "{", "elif", "else", "if", "then", "until", "while", "do"} while words and words[0] in reserved_prefixes: words.pop(0) if not words or words[0] in {"case", "done", "esac", "fi", "for", "in", "select", "}"}: return [] redirections = {"<", ">", ">>", "<<", "<<<", "<>", ">&", "<&", "&>", "&>>", ">|"} command_words: list[str] = [] skip_redirection_target = False for word in words: if skip_redirection_target: skip_redirection_target = False continue if word in redirections: skip_redirection_target = True continue command_words.append(word) while command_words and _is_shell_assignment(command_words[0]): command_words.pop(0) if command_words[:1] in (["command"], ["builtin"], ["exec"]): command_words.pop(0) while command_words and command_words[0].startswith("-"): command_words.pop(0) if command_words[:1] == ["env"]: command_words.pop(0) while command_words and ( command_words[0].startswith("-") or _is_shell_assignment(command_words[0]) ): command_words.pop(0) return command_words def _shell_simple_commands(text: str) -> list[list[str]]: delimiters = {";", ";;", ";&", ";;&", "&", "&&", "|", "||", "|&", "(", ")"} commands: list[list[str]] = [] segment: list[str] = [] array_depth = 0 def flush() -> None: command = _normalize_shell_command(segment) if command: commands.append(command) segment.clear() tokens = _shell_tokens(text) index = 0 while index < len(tokens): token = tokens[index] if array_depth: if token == "(": array_depth += 1 elif token == ")": array_depth -= 1 index += 1 continue if ( not segment and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", token) and tokens[index + 1:index + 3] == ["(", ")"] ): index += 3 if index < len(tokens) and tokens[index] == "{": index += 1 continue if token == "(" and segment and _is_shell_assignment(segment[-1]) and segment[-1].endswith("="): array_depth = 1 index += 1 continue if token in delimiters: flush() index += 1 continue segment.append(token) index += 1 flush() return commands def _shell_command_matches(command: list[str], expected: list[str]) -> bool: if not command or not expected or command[0] != expected[0]: return False expected_index = 0 for word in command: if word == expected[expected_index]: expected_index += 1 if expected_index == len(expected): return True return False def _count_shell_commands(text: str, expected: list[str]) -> int: return sum( _shell_command_matches(command, expected) for command in _shell_simple_commands(text) ) def _load_stage_contracts(source: DirectorySource) -> dict[str, set[str]]: contracts: dict[str, set[str]] = {} for profile_name, (path, constant_name) in STAGE_CONTRACTS.items(): try: tree = ast.parse(source.read_text(path), filename=path) except SyntaxError as exc: raise SeedPlanError(f"cannot parse stage contract {path}: {exc}") from exc stage_names: Any = None for node in tree.body: if not isinstance(node, ast.Assign): continue if any(isinstance(target, ast.Name) and target.id == constant_name for target in node.targets): try: stage_names = ast.literal_eval(node.value) except (ValueError, TypeError, SyntaxError) as exc: raise SeedPlanError( f"stage contract {path}:{constant_name} must be a literal string collection" ) from exc break if not isinstance(stage_names, (set, frozenset, list, tuple)) or not stage_names: raise SeedPlanError(f"stage contract {path}:{constant_name} is missing or empty") if not all(isinstance(item, str) and item for item in stage_names): raise SeedPlanError(f"stage contract {path}:{constant_name} must contain strings") if len(set(stage_names)) != len(stage_names): raise SeedPlanError(f"stage contract {path}:{constant_name} contains duplicates") contracts[profile_name] = set(stage_names) return contracts def _validate_condition( value: Any, label: str, profile_name: str, stage_contracts: dict[str, set[str]], ) -> None: if not isinstance(value, dict): raise SeedPlanError(f"{label} must be an object") condition_type = value.get("type") if condition_type == "always": _expect_exact_keys(value, {"type"}, label) return if condition_type in {"env_flag", "required_env_flag", "argument_flag"}: _expect_exact_keys(value, {"type", "name", "default", "equals"}, label) _expect_nonempty_string(value["name"], f"{label}.name") if not isinstance(value["default"], bool) or not isinstance(value["equals"], bool): raise SeedPlanError(f"{label} default and equals must be booleans") return if condition_type == "selected_stage": _expect_exact_keys(value, {"type", "name"}, label) stage_name = _expect_nonempty_string(value["name"], f"{label}.name") allowed_stages = stage_contracts.get(profile_name) if allowed_stages is None or stage_name not in allowed_stages: raise SeedPlanError( f"{label}.name is not a declared {profile_name} stage: {stage_name!r}" ) return if condition_type == "runtime_mode": _expect_exact_keys(value, {"type", "name"}, label) mode_name = _expect_nonempty_string(value["name"], f"{label}.name") allowed_modes = RUNTIME_MODE_CONTRACTS.get(profile_name, set()) if mode_name not in allowed_modes: raise SeedPlanError( f"{label}.name is not a declared {profile_name} runtime mode: {mode_name!r}" ) return if condition_type == "all": _expect_exact_keys(value, {"type", "conditions"}, label) conditions = value["conditions"] if not isinstance(conditions, list) or len(conditions) < 2: raise SeedPlanError(f"{label}.conditions must contain at least two conditions") for index, condition in enumerate(conditions): _validate_condition( condition, f"{label}.conditions[{index}]", profile_name, stage_contracts, ) return raise SeedPlanError(f"{label}.type is unsupported: {condition_type!r}") def _validate_selection( value: Any, schema: dict[str, dict[str, Any]], label: str, ) -> None: if not isinstance(value, dict): raise SeedPlanError(f"{label} must be an object") if set(value) != set(schema): raise SeedPlanError( f"{label} fields mismatch: expected={sorted(schema)} actual={sorted(value)}" ) for field, field_schema in schema.items(): selected = value[field] if not isinstance(selected, list) or not selected: raise SeedPlanError(f"{label}.{field} must be a non-empty array") if not all(isinstance(item, str) and item for item in selected): raise SeedPlanError(f"{label}.{field} must contain non-empty strings") if len(set(selected)) != len(selected): raise SeedPlanError(f"{label}.{field} contains duplicates") allowlist = field_schema["allowlist"] unknown = sorted(set(selected) - set(allowlist)) if unknown: raise SeedPlanError(f"{label}.{field} is not allowlisted: {unknown}") canonical = [item for item in allowlist if item in set(selected)] if selected != canonical: raise SeedPlanError(f"{label}.{field} must follow allowlist order") def validate_catalog(catalog: dict[str, Any], source: DirectorySource) -> dict[str, Any]: _expect_exact_keys(catalog, TOP_LEVEL_KEYS, "catalog") if catalog["schema_version"] != CATALOG_SCHEMA_VERSION: raise SeedPlanError( f"catalog schema_version must be {CATALOG_SCHEMA_VERSION}, got {catalog['schema_version']!r}" ) _expect_nonempty_string(catalog["catalog_version"], "catalog.catalog_version") modules = catalog["modules"] if not isinstance(modules, dict) or not modules: raise SeedPlanError("catalog.modules must be a non-empty object") for module_id, module in modules.items(): _expect_identifier(module_id, "module id") if not isinstance(module, dict): raise SeedPlanError(f"module {module_id} must be an object") _expect_exact_keys(module, MODULE_KEYS, f"module {module_id}") _expect_nonempty_string(module["description"], f"module {module_id}.description") _expect_identifier(module["dataset_class"], f"module {module_id}.dataset_class") if module["execution_owner"] not in EXECUTION_OWNERS: raise SeedPlanError(f"module {module_id} has invalid execution_owner") dispatcher_adapter = module["dispatcher_adapter"] if module["execution_owner"] == "seed_dispatcher": if dispatcher_adapter is None: raise SeedPlanError( f"module {module_id} seed_dispatcher ownership requires dispatcher_adapter" ) _expect_identifier( dispatcher_adapter, f"module {module_id}.dispatcher_adapter", ) elif dispatcher_adapter is not None: raise SeedPlanError( f"module {module_id}.dispatcher_adapter requires seed_dispatcher ownership" ) _validate_path_list(module["data_sources"], f"module {module_id}.data_sources", source) adapter_paths = _validate_path_list( module["adapter_paths"], f"module {module_id}.adapter_paths", source ) if not adapter_paths: raise SeedPlanError(f"module {module_id} must declare at least one adapter path") _validate_path_list( module["authoritative_paths"], f"module {module_id}.authoritative_paths", source ) module["selection_schema"] = _validate_selection_schema( module["selection_schema"], f"module {module_id}.selection_schema" ) stats = module["declared_stats"] if not isinstance(stats, dict): raise SeedPlanError(f"module {module_id}.declared_stats must be an object") for stat_name, stat_value in stats.items(): _expect_identifier(stat_name, f"module {module_id} stat") if isinstance(stat_value, bool) or not isinstance(stat_value, int) or stat_value < 0: raise SeedPlanError( f"module {module_id}.declared_stats.{stat_name} must be a non-negative integer" ) profiles = catalog["profiles"] if not isinstance(profiles, dict) or set(profiles) != set(PROFILES): raise SeedPlanError( f"catalog profiles must be exactly {list(PROFILES)}, got {sorted(profiles) if isinstance(profiles, dict) else type(profiles).__name__}" ) stage_contracts = _load_stage_contracts(source) invocation_by_id: dict[str, dict[str, Any]] = {} for profile_name in PROFILES: profile = profiles[profile_name] if not isinstance(profile, dict): raise SeedPlanError(f"profile {profile_name} must be an object") _expect_exact_keys(profile, {"invocations"}, f"profile {profile_name}") invocations = profile["invocations"] if not isinstance(invocations, list): raise SeedPlanError(f"profile {profile_name}.invocations must be an array") previous_order = -1 for index, invocation in enumerate(invocations): label = f"profile {profile_name}.invocations[{index}]" if not isinstance(invocation, dict): raise SeedPlanError(f"{label} must be an object") _expect_exact_keys(invocation, INVOCATION_KEYS, label) invocation_id = _expect_identifier( invocation["invocation_id"], f"{label}.invocation_id", qualified=True ) if not invocation_id.startswith(f"{profile_name}."): raise SeedPlanError(f"{label}.invocation_id must start with {profile_name}.") if invocation_id in invocation_by_id: raise SeedPlanError(f"duplicate invocation_id: {invocation_id}") module_id = _expect_identifier(invocation["module_id"], f"{label}.module_id") if module_id not in modules: raise SeedPlanError(f"{label} references unknown module: {module_id}") _expect_identifier(invocation["phase"], f"{label}.phase") order = invocation["order"] if isinstance(order, bool) or not isinstance(order, int) or order < 0: raise SeedPlanError(f"{label}.order must be a non-negative integer") if order <= previous_order: raise SeedPlanError(f"profile {profile_name} invocation order must be strictly increasing") previous_order = order _validate_condition( invocation["condition"], f"{label}.condition", profile_name, stage_contracts, ) _validate_selection( invocation["selection"], modules[module_id]["selection_schema"], f"{label}.selection", ) invocation_by_id[invocation_id] = invocation exclusions = catalog["excluded_mutations"] if not isinstance(exclusions, list): raise SeedPlanError("catalog.excluded_mutations must be an array") exclusion_by_id: dict[str, dict[str, Any]] = {} for index, exclusion in enumerate(exclusions): label = f"excluded_mutations[{index}]" if not isinstance(exclusion, dict): raise SeedPlanError(f"{label} must be an object") _expect_exact_keys(exclusion, EXCLUSION_KEYS, label) exclusion_id = _expect_identifier(exclusion["exclusion_id"], f"{label}.exclusion_id") if exclusion_id in exclusion_by_id: raise SeedPlanError(f"duplicate exclusion_id: {exclusion_id}") _expect_nonempty_string(exclusion["description"], f"{label}.description") _expect_identifier(exclusion["owner"], f"{label}.owner") _expect_nonempty_string(exclusion["reason"], f"{label}.reason") source_paths = _validate_path_list(exclusion["source_paths"], f"{label}.source_paths", source) if not source_paths: raise SeedPlanError(f"{label}.source_paths must not be empty") exclusion_by_id[exclusion_id] = exclusion callsites = catalog["expected_callsites"] if not isinstance(callsites, list) or not callsites: raise SeedPlanError("catalog.expected_callsites must be a non-empty array") callsite_ids: set[str] = set() mapped_invocations: dict[str, str] = {} parsed_python_sources: dict[str, ast.Module] = {} for index, callsite in enumerate(callsites): label = f"expected_callsites[{index}]" if not isinstance(callsite, dict): raise SeedPlanError(f"{label} must be an object") callsite_keys = set(callsite) if not CALLSITE_KEYS.issubset(callsite_keys) or not callsite_keys.issubset( CALLSITE_KEYS | CALLSITE_OPTIONAL_KEYS ): missing = sorted(CALLSITE_KEYS - callsite_keys) unknown = sorted(callsite_keys - CALLSITE_KEYS - CALLSITE_OPTIONAL_KEYS) raise SeedPlanError(f"{label} keys mismatch: missing={missing} unknown={unknown}") callsite_id = _expect_identifier(callsite["callsite_id"], f"{label}.callsite_id", qualified=True) if callsite_id in callsite_ids: raise SeedPlanError(f"duplicate callsite_id: {callsite_id}") callsite_ids.add(callsite_id) path = validate_relpath(callsite["path"]) marker = _expect_nonempty_string(callsite["marker"], f"{label}.marker") expected_count = callsite["expected_count"] if isinstance(expected_count, bool) or not isinstance(expected_count, int) or expected_count < 1: raise SeedPlanError(f"{label}.expected_count must be a positive integer") shell_command = callsite.get("shell_command") if path.endswith(".sh"): if not isinstance(shell_command, list) or not shell_command or not all( isinstance(word, str) and word for word in shell_command ): raise SeedPlanError(f"{label}.shell_command must be a non-empty string array") elif shell_command is not None: raise SeedPlanError(f"{label}.shell_command is only valid for shell callsites") module_id = callsite["module_id"] exclusion_id = callsite["exclusion_id"] if not isinstance(module_id, str) or not isinstance(exclusion_id, str): raise SeedPlanError(f"{label} module_id and exclusion_id must be strings") if bool(module_id) == bool(exclusion_id): raise SeedPlanError(f"{label} must declare exactly one module_id or exclusion_id") if module_id and module_id not in modules: raise SeedPlanError(f"{label} references unknown module: {module_id}") if exclusion_id and exclusion_id not in exclusion_by_id: raise SeedPlanError(f"{label} references unknown exclusion: {exclusion_id}") invocation_ids = callsite["invocation_ids"] if not isinstance(invocation_ids, list) or not all( isinstance(item, str) for item in invocation_ids ): raise SeedPlanError(f"{label}.invocation_ids must be an array of strings") if exclusion_id and invocation_ids: raise SeedPlanError(f"{label} exclusion callsite cannot map invocations") required_after_invocation_id = callsite.get("required_after_invocation_id") if required_after_invocation_id is not None: if not exclusion_id: raise SeedPlanError( f"{label}.required_after_invocation_id is only valid for exclusion callsites" ) required_after_invocation_id = _expect_identifier( required_after_invocation_id, f"{label}.required_after_invocation_id", qualified=True, ) if required_after_invocation_id not in invocation_by_id: raise SeedPlanError( f"{label}.required_after_invocation_id references unknown invocation: " f"{required_after_invocation_id}" ) callsite_profile = callsite_id.split(".", 1)[0] if not required_after_invocation_id.startswith(f"{callsite_profile}."): raise SeedPlanError( f"{label}.required_after_invocation_id must use the {callsite_profile} profile" ) invocation_profiles = [ invocation_id.split(".", 1)[0] for invocation_id in invocation_ids ] multiple_profile_mappings = expected_count == 1 and len(invocation_ids) > 1 if ( module_id and multiple_profile_mappings and len(set(invocation_profiles)) != len(invocation_profiles) ): raise SeedPlanError( f"{label} shared direct call must map distinct environment profiles" ) shared_profile_callsite = ( multiple_profile_mappings and len(set(invocation_profiles)) == len(invocation_profiles) ) if module_id and not ( len(invocation_ids) == expected_count or shared_profile_callsite ): raise SeedPlanError( f"{label} must map one invocation per expected direct call or " "multiple profiles to one shared direct call: " f"expected={expected_count} mapped={len(invocation_ids)}" ) for invocation_id in invocation_ids: invocation = invocation_by_id.get(invocation_id) if invocation is None: raise SeedPlanError(f"{label} references unknown invocation: {invocation_id}") if invocation["module_id"] != module_id: raise SeedPlanError(f"{label} module does not match invocation {invocation_id}") if invocation_id in mapped_invocations: raise SeedPlanError( f"invocation {invocation_id} is mapped by multiple callsites: {mapped_invocations[invocation_id]}, {callsite_id}" ) mapped_invocations[invocation_id] = callsite_id source_text = source.read_text(path) marker_count = source_text.count(marker) python_symbol = _python_call_symbol(path, marker) executable_count: int | None = None shell_executable_count: int | None = None expected_python_calls = callsite.get("expected_python_calls", expected_count) if ( isinstance(expected_python_calls, bool) or not isinstance(expected_python_calls, int) or expected_python_calls < 1 ): raise SeedPlanError(f"{label}.expected_python_calls must be a positive integer") if python_symbol is None and "expected_python_calls" in callsite: raise SeedPlanError(f"{label}.expected_python_calls requires a Python call marker") if python_symbol is not None: if path not in parsed_python_sources: parsed_python_sources[path] = _parse_python_source(source, path) executable_count = _count_python_calls(parsed_python_sources[path], python_symbol) elif path.endswith(".sh"): assert isinstance(shell_command, list) shell_executable_count = _count_shell_commands(source_text, shell_command) if marker_count != expected_count or ( executable_count is not None and executable_count != expected_python_calls ) or ( shell_executable_count is not None and shell_executable_count != expected_count ): evidence = f"expected_marker={expected_count} marker={marker_count}" if executable_count is not None: evidence += ( f" expected_executable={expected_python_calls}" f" executable={executable_count} symbol={python_symbol}" ) if shell_executable_count is not None: evidence += ( f" expected_shell_executable={expected_count}" f" shell_executable={shell_executable_count}" ) raise SeedPlanError( f"callsite {callsite_id} count mismatch: {evidence}" ) missing_callsites = sorted(set(invocation_by_id) - set(mapped_invocations)) if missing_callsites: raise SeedPlanError(f"invocations missing expected callsite mapping: {missing_callsites}") return catalog def load_and_validate_catalog( source: DirectorySource, catalog_relpath: str, ) -> dict[str, Any]: catalog_path = validate_relpath(catalog_relpath) catalog = _load_json_object(source, catalog_path) return validate_catalog(catalog, source) def build_plan( catalog: dict[str, Any], profile_name: str, source_evidence: dict[str, Any], ) -> dict[str, Any]: if profile_name not in PROFILES: raise SeedPlanError(f"unsupported profile: {profile_name}") invocations = copy.deepcopy(catalog["profiles"][profile_name]["invocations"]) selected_module_ids = sorted({item["module_id"] for item in invocations}) modules = { module_id: copy.deepcopy(catalog["modules"][module_id]) for module_id in selected_module_ids } return { "schema_version": ARTIFACT_SCHEMA_VERSION, "catalog_version": catalog["catalog_version"], "profile": profile_name, "summary": { "invocations": len(invocations), "modules": len(selected_module_ids), "seed_dispatcher_invocations": sum( modules[item["module_id"]]["execution_owner"] == "seed_dispatcher" for item in invocations ), "deploy_pipeline_invocations": sum( modules[item["module_id"]]["execution_owner"] == "deploy_pipeline" for item in invocations ), }, "modules": modules, "invocations": invocations, "source": copy.deepcopy(source_evidence), } def build_inventory( catalog: dict[str, Any], source_evidence: dict[str, Any], ) -> dict[str, Any]: module_rows = [] for module_id in sorted(catalog["modules"]): module = catalog["modules"][module_id] module_rows.append( { "module_id": module_id, "dataset_class": module["dataset_class"], "execution_owner": module["execution_owner"], "dispatcher_adapter": module["dispatcher_adapter"], "data_sources": copy.deepcopy(module["data_sources"]), "adapter_paths": copy.deepcopy(module["adapter_paths"]), "authoritative_paths": copy.deepcopy(module["authoritative_paths"]), "declared_stats": copy.deepcopy(module["declared_stats"]), } ) profile_rows = [] for profile_name in PROFILES: invocations = catalog["profiles"][profile_name]["invocations"] profile_rows.append( { "profile": profile_name, "invocation_count": len(invocations), "seed_dispatcher_invocation_count": sum( catalog["modules"][item["module_id"]]["execution_owner"] == "seed_dispatcher" for item in invocations ), "deploy_pipeline_invocation_count": sum( catalog["modules"][item["module_id"]]["execution_owner"] == "deploy_pipeline" for item in invocations ), "invocation_ids": [item["invocation_id"] for item in invocations], } ) return { "schema_version": ARTIFACT_SCHEMA_VERSION, "catalog_version": catalog["catalog_version"], "summary": { "modules": len(module_rows), "profiles": len(profile_rows), "invocations": sum(row["invocation_count"] for row in profile_rows), "expected_callsites": len(catalog["expected_callsites"]), "excluded_mutations": len(catalog["excluded_mutations"]), }, "modules": module_rows, "profiles": profile_rows, "expected_callsites": copy.deepcopy(catalog["expected_callsites"]), "excluded_mutations": copy.deepcopy(catalog["excluded_mutations"]), "source": copy.deepcopy(source_evidence), } def render_json(value: dict[str, Any]) -> bytes: return (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8") def _resolved_output_path(raw_path: str, source_root: Path, label: str) -> Path: source_root = source_root.resolve(strict=True) 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 SeedPlanError(f"{label} must be outside source root") if resolved.exists() and not resolved.is_file(): raise SeedPlanError(f"{label} is not a regular file path: {resolved}") return resolved def _write_temp_file(parent: Path, prefix: str, payload: bytes) -> Path: handle = tempfile.NamedTemporaryFile(prefix=prefix, dir=parent, delete=False) path = Path(handle.name) try: with handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) except Exception: path.unlink(missing_ok=True) raise return path def _restore_output(path: Path, previous: bytes | None) -> None: if previous is None: path.unlink(missing_ok=True) return restore_path = _write_temp_file(path.parent, f".{path.name}.restore-", previous) os.replace(restore_path, path) def write_artifact_pair( plan_output: str, inventory_output: str, source_root: Path, plan_payload: bytes, inventory_payload: bytes, ) -> tuple[Path, Path]: plan_path = _resolved_output_path(plan_output, source_root, "plan output") inventory_path = _resolved_output_path(inventory_output, source_root, "inventory output") if plan_path == inventory_path: raise SeedPlanError("plan and inventory output paths must be different") if plan_path.parent != inventory_path.parent: raise SeedPlanError("plan and inventory outputs must use the same parent directory") plan_path.parent.mkdir(parents=True, exist_ok=True) plan_path = _resolved_output_path(str(plan_path), source_root, "plan output") inventory_path = _resolved_output_path(str(inventory_path), source_root, "inventory output") previous_plan = plan_path.read_bytes() if plan_path.is_file() else None previous_inventory = inventory_path.read_bytes() if inventory_path.is_file() else None plan_temp = _write_temp_file(plan_path.parent, f".{plan_path.name}.tmp-", plan_payload) inventory_temp = _write_temp_file( inventory_path.parent, f".{inventory_path.name}.tmp-", inventory_payload ) plan_replaced = False inventory_replaced = False try: os.replace(plan_temp, plan_path) plan_replaced = True os.replace(inventory_temp, inventory_path) inventory_replaced = True except Exception as exc: plan_temp.unlink(missing_ok=True) inventory_temp.unlink(missing_ok=True) if plan_replaced: _restore_output(plan_path, previous_plan) if inventory_replaced: _restore_output(inventory_path, previous_inventory) raise SeedPlanError(f"failed to write Seed plan artifact pair: {exc}") from exc return plan_path, inventory_path def plan_deploy_seed( *, source_root: Path | str, profile_name: str, catalog_relpath: str, ) -> tuple[dict[str, Any], dict[str, Any], DirectorySource]: source = DirectorySource(source_root) catalog = load_and_validate_catalog(source, catalog_relpath) source_evidence = source.evidence() plan = build_plan(catalog, profile_name, source_evidence) inventory = build_inventory(catalog, source_evidence) return plan, inventory, source def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate the deploy Seed catalog and emit read-only plan artifacts." ) parser.add_argument("--source-root", required=True) parser.add_argument("--profile", required=True, choices=PROFILES) parser.add_argument( "--catalog-relpath", default="deployments/offline/runtime/seeds/deploy-seed-catalog.json", ) parser.add_argument("--plan-output", required=True) parser.add_argument("--inventory-output", required=True) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: plan, inventory, source = plan_deploy_seed( source_root=args.source_root, profile_name=args.profile, catalog_relpath=args.catalog_relpath, ) plan_path, inventory_path = write_artifact_pair( args.plan_output, args.inventory_output, source.root, render_json(plan), render_json(inventory), ) except SeedPlanError as exc: print(f"deploy Seed plan failed: {exc}", file=sys.stderr) return 2 print( json.dumps( { "profile": args.profile, "source_fingerprint_sha256": plan["source"]["source_fingerprint_sha256"], "plan_output": str(plan_path), "inventory_output": str(inventory_path), }, sort_keys=True, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())