#!/usr/bin/env python3 """Immutable plan and timed-rehearsal contract for Prod Catalog migration.""" from __future__ import annotations import hashlib import json import re from pathlib import Path, PurePosixPath from typing import Any from deploy.online.runtime.common import compute_source_backed_pack_runtime_metadata from scripts.deployment_bundle import ( BundleError, canonical_json_bytes, materialized_layout, sha256_bytes, validate_manifest, verify_materialized_files, ) from scripts.prod_promotion_contract import ( METADATA_FIRST_SHARED_STATE_POLICY, build_prod_promotion_contract_from_source_tree, ) PLAN_SCHEMA_VERSION = "prod-catalog-migration-plan/v1" REHEARSAL_SCHEMA_VERSION = "prod-catalog-timed-rehearsal/v1" STAGING_REHEARSAL_OBSERVATION_SCHEMA_VERSION = ( "prod-catalog-staging-rehearsal-observation/v1" ) MAINTENANCE_SCHEMA_VERSION = "prod-catalog-maintenance-evidence/v1" GIT_COMMIT_RE = re.compile(r"^[0-9a-f]{40}$") SHA256_RE = re.compile(r"^[0-9a-f]{64}$") IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") PAYLOAD_HASH_RE = re.compile(r"^nexus-pack-sha256-v1:[0-9a-f]{64}$") COMMAND_ID_RE = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" ) PLAN_KEYS = { "schema_version", "candidate", "operation_id", "packs", "plan_sha256", } CANDIDATE_KEYS = { "release_tag", "tag_target_commit", "upstream_ref", "official_app_packs_ref", "contract_sha256", "shared_state_digest", "final_pack_hash", "bundle_manifest_sha256", } CANDIDATE_SOURCE_KEYS = CANDIDATE_KEYS - {"release_tag"} PLANNED_PACK_KEYS = { "name", "entry", "source_kind", "source_uri", "source_revision", "payload_hash", } REHEARSAL_KEYS = { "schema_version", "status", "candidate", "plan_sha256", "started_at_unix", "completed_at_unix", "duration_seconds", "installed_packs", "evidence_sha256", } INSTALLED_PACK_KEYS = { "name", "status", "source_kind", "source_uri", "source_revision", "desired_payload_hash", "applied_payload_hash", "last_error", } STAGING_REHEARSAL_OBSERVATION_KEYS = { "schema_version", "status", "candidate_source", "operation_id", "packs", "started_at_unix", "completed_at_unix", "duration_seconds", "installed_packs", "observation_sha256", } MAINTENANCE_KEYS = { "schema_version", "status", "candidate", "plan_sha256", "rehearsal_evidence_sha256", "maintenance_mode", "started_at_unix", "completed_at_unix", "duration_seconds", "write_maintenance", "source_reader_stop", "zero_nonterminal_workitems", "evidence_sha256", } WRITE_MAINTENANCE_KEYS = {"status", "observed_at_unix"} SOURCE_READER_STOP_KEYS = { "command_id", "status", "completed_at_unix", "readers", } SOURCE_READER_KEYS = {"name", "state", "listener_count"} ZERO_NONTERMINAL_WORKITEM_KEYS = { "command_id", "status", "drained_at_unix", "drained_nonterminal_count", "observed_at_unix", "scope", "table", "nonterminal_count", } def _canonical_json(value: object) -> bytes: return json.dumps( value, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") def _digest(value: object) -> str: return hashlib.sha256(_canonical_json(value)).hexdigest() def _require_object( value: object, expected_keys: set[str], label: str, ) -> dict[str, Any]: if not isinstance(value, dict) or set(value) != expected_keys: raise ValueError(f"{label} has unexpected or missing fields") return value def _require_match(value: object, pattern: re.Pattern[str], label: str) -> str: normalized = str(value).strip() if not pattern.fullmatch(normalized): raise ValueError(f"{label} is invalid") return normalized def _require_candidate(value: object) -> dict[str, str]: candidate = _require_object(value, CANDIDATE_KEYS, "Catalog candidate") release_tag = _require_match( candidate.get("release_tag"), IDENTIFIER_RE, "Catalog candidate release_tag", ) return { "release_tag": release_tag, "tag_target_commit": _require_match( candidate.get("tag_target_commit"), GIT_COMMIT_RE, "Catalog candidate tag_target_commit", ), "upstream_ref": _require_match( candidate.get("upstream_ref"), GIT_COMMIT_RE, "Catalog candidate upstream_ref", ), "official_app_packs_ref": _require_match( candidate.get("official_app_packs_ref"), GIT_COMMIT_RE, "Catalog candidate official_app_packs_ref", ), "contract_sha256": _require_match( candidate.get("contract_sha256"), SHA256_RE, "Catalog candidate contract_sha256", ), "shared_state_digest": _require_match( candidate.get("shared_state_digest"), SHA256_RE, "Catalog candidate shared_state_digest", ), "final_pack_hash": _require_match( candidate.get("final_pack_hash"), SHA256_RE, "Catalog candidate final_pack_hash", ), "bundle_manifest_sha256": _require_match( candidate.get("bundle_manifest_sha256"), SHA256_RE, "Catalog candidate bundle_manifest_sha256", ), } def _require_candidate_source(value: object) -> dict[str, str]: source = _require_object( value, CANDIDATE_SOURCE_KEYS, "Catalog candidate source", ) candidate = _require_candidate( {"release_tag": "staging-rehearsal", **source} ) candidate.pop("release_tag") return candidate def _require_pack_entry(value: object) -> str: entry = str(value).strip() if entry.startswith("official_repo:"): _require_match( entry.removeprefix("official_repo:"), IDENTIFIER_RE, "Catalog migration official pack entry", ) else: path = PurePosixPath(entry) if ( not entry or ":" in entry or path.is_absolute() or ".." in path.parts or path.as_posix() != entry ): raise ValueError("Catalog migration local pack entry is invalid") return entry def _require_rehearsal_source_uri( value: object, *, expected_kind: str, expected_uri: str, ) -> str: source_uri = str(value).strip() if expected_kind == "official_repo": if source_uri != expected_uri: raise ValueError( "Prod Catalog rehearsal pack source does not match the plan" ) return source_uri path = PurePosixPath(source_uri) expected_suffix = PurePosixPath(expected_uri) if ( not source_uri or not path.is_absolute() or ".." in path.parts or len(path.parts) < len(expected_suffix.parts) or path.parts[-len(expected_suffix.parts) :] != expected_suffix.parts ): raise ValueError( "Prod Catalog rehearsal pack source does not match the plan" ) return source_uri def _require_planned_packs( value: object, *, candidate: dict[str, str], ) -> list[dict[str, str]]: if not isinstance(value, list) or not value: raise ValueError("Catalog migration packs must be a non-empty list") packs: list[dict[str, str]] = [] for index, raw_pack in enumerate(value): pack = _require_object( raw_pack, PLANNED_PACK_KEYS, f"Catalog migration packs[{index}]", ) entry = _require_pack_entry(pack.get("entry")) source_revision = _require_match( pack.get("source_revision"), GIT_COMMIT_RE, f"Catalog migration packs[{index}].source_revision", ) expected_revision = ( candidate["official_app_packs_ref"] if entry.startswith("official_repo:") else candidate["tag_target_commit"] ) expected_source_kind = ( "official_repo" if entry.startswith("official_repo:") else "local_path" ) expected_source_uri = ( entry.removeprefix("official_repo:") if entry.startswith("official_repo:") else entry ) if source_revision != expected_revision: raise ValueError( f"Catalog migration packs[{index}] source revision is invalid" ) packs.append( { "name": _require_match( pack.get("name"), IDENTIFIER_RE, f"Catalog migration packs[{index}].name", ), "entry": entry, "source_kind": _require_match( pack.get("source_kind"), IDENTIFIER_RE, f"Catalog migration packs[{index}].source_kind", ), "source_uri": _require_pack_entry(pack.get("source_uri")), "source_revision": source_revision, "payload_hash": _require_match( pack.get("payload_hash"), PAYLOAD_HASH_RE, f"Catalog migration packs[{index}].payload_hash", ), } ) if ( packs[-1]["source_kind"] != expected_source_kind or packs[-1]["source_uri"] != expected_source_uri ): raise ValueError( f"Catalog migration packs[{index}] source authority is invalid" ) names = [pack["name"] for pack in packs] entries = [pack["entry"] for pack in packs] if len(names) != len(set(names)) or len(entries) != len(set(entries)): raise ValueError("Catalog migration pack identities must be unique") return packs def _planned_packs_from_source_tree( candidate_source_root: Path, *, candidate: dict[str, str], ) -> list[dict[str, str]]: source_root = candidate_source_root.expanduser() if source_root.is_symlink() or not source_root.is_dir(): raise ValueError("Catalog migration candidate source root is invalid") source_root = source_root.resolve() stack_lock_path = source_root / "stack.lock.json" if stack_lock_path.is_symlink() or not stack_lock_path.is_file(): raise ValueError("Catalog migration candidate stack.lock.json is invalid") try: stack_lock = json.loads(stack_lock_path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise ValueError( "Catalog migration candidate stack.lock.json is invalid" ) from exc if not isinstance(stack_lock, dict): raise ValueError("Catalog migration candidate stack.lock.json is invalid") pack_build = stack_lock.get("pack_build") if not isinstance(pack_build, dict): raise ValueError("Catalog migration candidate pack_build is invalid") final_pack_entry = _require_pack_entry( pack_build.get("final_pack_path") or "packs/NOT-EDIT-final-merged-packs" ) raw_entries = stack_lock.get("packs") if isinstance(raw_entries, list) and raw_entries: entries = [str(entry).strip() for entry in raw_entries] else: entries = [final_pack_entry] normalized_entries = [_require_pack_entry(entry) for entry in entries] if final_pack_entry not in normalized_entries: raise ValueError( "Catalog migration candidate final pack is missing from stack packs" ) packs: list[dict[str, str]] = [] for normalized_entry in normalized_entries: if normalized_entry.startswith("official_repo:"): pack_name = normalized_entry.removeprefix("official_repo:") source_revision = candidate["official_app_packs_ref"] else: pack_json = source_root / normalized_entry / "pack.json" try: if ( pack_json.is_symlink() or not pack_json.is_file() or not pack_json.resolve().is_relative_to(source_root) ): raise ValueError pack_manifest = json.loads(pack_json.read_text()) except (OSError, json.JSONDecodeError, ValueError) as exc: raise ValueError( f"Catalog migration pack manifest is invalid: {normalized_entry}" ) from exc if not isinstance(pack_manifest, dict): raise ValueError( f"Catalog migration pack manifest is invalid: {normalized_entry}" ) pack_name = _require_match( pack_manifest.get("name"), IDENTIFIER_RE, f"Catalog migration pack name for {normalized_entry}", ) source_revision = candidate["tag_target_commit"] packs.append( { "name": pack_name, "entry": normalized_entry, "source_kind": ( "official_repo" if normalized_entry.startswith("official_repo:") else "local_path" ), "source_uri": ( normalized_entry.removeprefix("official_repo:") if normalized_entry.startswith("official_repo:") else normalized_entry ), "source_revision": source_revision, } ) names = [pack["name"] for pack in packs] entries = [pack["entry"] for pack in packs] if len(names) != len(set(names)) or len(entries) != len(set(entries)): raise ValueError("Catalog migration pack identities must be unique") return packs def _require_candidate_bundle_authority( candidate: dict[str, str], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], ) -> tuple[Path, Path, Path, str]: bundle_root = candidate_bundle_root.expanduser() if bundle_root.is_symlink() or not bundle_root.is_dir(): raise ValueError("Catalog migration candidate bundle root is invalid") bundle_root = bundle_root.resolve() try: manifest, entries = validate_manifest(candidate_bundle_manifest) verify_materialized_files(bundle_root, entries) layout = materialized_layout(bundle_root, manifest) except BundleError as exc: raise ValueError("Catalog migration candidate bundle manifest is invalid") from exc manifest_sha256 = sha256_bytes(canonical_json_bytes(manifest)) if candidate["bundle_manifest_sha256"] != manifest_sha256: raise ValueError( "Catalog migration candidate bundle_manifest_sha256 does not match " "exact source" ) source = manifest.get("source") if not isinstance(source, dict): raise ValueError("Catalog migration candidate bundle source is invalid") for candidate_key, source_key in ( ("tag_target_commit", "downstream_ref"), ("upstream_ref", "upstream_ref"), ("official_app_packs_ref", "official_app_packs_ref"), ("final_pack_hash", "final_pack_hash"), ): if candidate[candidate_key] != str(source.get(source_key, "")).strip(): raise ValueError( f"Catalog migration candidate {candidate_key} does not match " "exact source" ) downstream_root = Path(str(layout["downstream_root"])) upstream_root = Path(str(layout["upstream_root"])) official_root = Path(str(layout["official_app_packs_root"])) for root in (downstream_root, upstream_root, official_root): if root.is_symlink() or not root.is_dir(): raise ValueError("Catalog migration candidate bundle source is invalid") try: stack_lock_raw = (downstream_root / "stack.lock.json").read_bytes() stack_lock = json.loads(stack_lock_raw) pack_build = stack_lock["pack_build"] final_pack_entry = _require_pack_entry( pack_build.get("final_pack_path") or "packs/NOT-EDIT-final-merged-packs" ) lineage = json.loads( ( downstream_root / final_pack_entry / ".build-lineage.json" ).read_text() ) lineage_refs = { "upstream_ref": lineage["upstreamRuntime"]["ref"], "official_app_packs_ref": lineage["officialPack"]["ref"], } lineage_stack_lock_sha256 = lineage["downstream"]["stackLockSha256"] except (KeyError, OSError, TypeError, json.JSONDecodeError) as exc: raise ValueError("Catalog migration candidate lineage is invalid") from exc source_final_pack = _require_pack_entry( source.get("final_pack_relative_path") ) if ( source_final_pack.startswith("official_repo:") or source_final_pack != final_pack_entry or Path(str(layout["final_pack_dir"])).resolve() != (downstream_root / final_pack_entry).resolve() ): raise ValueError( "Catalog migration candidate final pack layout does not match exact source" ) if lineage_stack_lock_sha256 != hashlib.sha256(stack_lock_raw).hexdigest(): raise ValueError( "Catalog migration candidate stack lock does not match exact lineage" ) for key, expected in lineage_refs.items(): if candidate[key] != expected: raise ValueError( f"Catalog migration candidate {key} does not match exact source" ) promotion_contract = build_prod_promotion_contract_from_source_tree( downstream_root, shared_state_policy=METADATA_FIRST_SHARED_STATE_POLICY, ) for key, expected in ( ("contract_sha256", promotion_contract["contract_sha256"]), ("shared_state_digest", promotion_contract["shared_state_digest"]), ( "final_pack_hash", promotion_contract["components"]["final_pack"]["payload_sha256"], ), ): if candidate[key] != expected: raise ValueError( f"Catalog migration candidate {key} does not match exact source" ) return ( downstream_root, upstream_root, official_root, str(source.get("backend_go_build_tags", "")).strip(), ) def _source_backed_planned_packs( source_packs: list[dict[str, str]], *, downstream_root: Path, upstream_root: Path, official_root: Path, backend_go_build_tags: str, runtime_metadata_env: dict[str, str] | None, candidate: dict[str, str], ) -> list[dict[str, str]]: metadata_env = dict(runtime_metadata_env or {}) metadata_env["NEXUSCRM_OFFICIAL_APP_PACKS_ROOT"] = str(official_root) packs: list[dict[str, str]] = [] for pack in source_packs: pack_root = ( official_root if pack["entry"].startswith("official_repo:") else downstream_root ) pack_dir = ( official_root / pack["name"] if pack["entry"].startswith("official_repo:") else downstream_root / pack["entry"] ) if ( pack_dir.is_symlink() or not pack_dir.is_dir() or not pack_dir.resolve().is_relative_to(pack_root) ): raise ValueError( f"Catalog migration pack source is outside candidate: {pack['entry']}" ) pack_dir = pack_dir.resolve() metadata = compute_source_backed_pack_runtime_metadata( upstream_root, pack_dir, envv=metadata_env, backend_go_build_tags=backend_go_build_tags, ) if str(metadata.get("name", "")).strip() != pack["name"]: raise ValueError( "Catalog migration runtime metadata name does not match " f"{pack['entry']}" ) packs.append( { **pack, "payload_hash": _require_match( metadata.get("payload_hash"), PAYLOAD_HASH_RE, f"Catalog migration runtime payload hash for {pack['name']}", ), } ) return _require_planned_packs(packs, candidate=candidate) def build_prod_catalog_migration_plan( *, candidate: dict[str, object], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: normalized_candidate = _require_candidate(candidate) downstream_root, upstream_root, official_root, build_tags = ( _require_candidate_bundle_authority( normalized_candidate, candidate_bundle_root, candidate_bundle_manifest, ) ) source_packs = _planned_packs_from_source_tree( downstream_root, candidate=normalized_candidate, ) plan: dict[str, Any] = { "schema_version": PLAN_SCHEMA_VERSION, "candidate": normalized_candidate, "operation_id": ( "prod-catalog-" + normalized_candidate["tag_target_commit"] ), "packs": _source_backed_planned_packs( source_packs, downstream_root=downstream_root, upstream_root=upstream_root, official_root=official_root, backend_go_build_tags=build_tags, runtime_metadata_env=runtime_metadata_env, candidate=normalized_candidate, ), } plan["plan_sha256"] = _digest(plan) return require_prod_catalog_migration_plan( plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) def require_prod_catalog_migration_plan( value: object, *, candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: plan = _require_object(value, PLAN_KEYS, "Prod Catalog migration plan") if plan.get("schema_version") != PLAN_SCHEMA_VERSION: raise ValueError("Prod Catalog migration plan schema_version is unsupported") candidate = _require_candidate(plan.get("candidate")) downstream_root, upstream_root, official_root, build_tags = ( _require_candidate_bundle_authority( candidate, candidate_bundle_root, candidate_bundle_manifest, ) ) expected_operation_id = "prod-catalog-" + candidate["tag_target_commit"] if plan.get("operation_id") != expected_operation_id: raise ValueError("Prod Catalog migration operation_id is not deterministic") packs = _require_planned_packs( plan.get("packs"), candidate=candidate, ) source_packs = _planned_packs_from_source_tree( downstream_root, candidate=candidate, ) expected_packs = _source_backed_planned_packs( source_packs, downstream_root=downstream_root, upstream_root=upstream_root, official_root=official_root, backend_go_build_tags=build_tags, runtime_metadata_env=runtime_metadata_env, candidate=candidate, ) if packs != expected_packs: raise ValueError( "Prod Catalog migration plan does not match the exact candidate source" ) unsigned = {key: item for key, item in plan.items() if key != "plan_sha256"} if plan.get("plan_sha256") != _digest(unsigned): raise ValueError("Prod Catalog migration plan_sha256 mismatch") return json.loads(json.dumps(plan)) def build_prod_catalog_rehearsal_evidence( *, plan: dict[str, object], started_at_unix: int, completed_at_unix: int, installed_packs: list[dict[str, str]], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: normalized_plan = require_prod_catalog_migration_plan( plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) evidence: dict[str, Any] = { "schema_version": REHEARSAL_SCHEMA_VERSION, "status": "PASS", "candidate": normalized_plan["candidate"], "plan_sha256": normalized_plan["plan_sha256"], "started_at_unix": started_at_unix, "completed_at_unix": completed_at_unix, "duration_seconds": completed_at_unix - started_at_unix, "installed_packs": installed_packs, } evidence["evidence_sha256"] = _digest(evidence) return require_prod_catalog_rehearsal_evidence( evidence, plan=normalized_plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) def require_prod_catalog_rehearsal_evidence( value: object, *, plan: dict[str, object], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: evidence = _require_object( value, REHEARSAL_KEYS, "Prod Catalog rehearsal evidence", ) normalized_plan = require_prod_catalog_migration_plan( plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) if evidence.get("schema_version") != REHEARSAL_SCHEMA_VERSION: raise ValueError("Prod Catalog rehearsal schema_version is unsupported") if evidence.get("status") != "PASS": raise ValueError("Prod Catalog rehearsal did not pass") if ( _require_candidate(evidence.get("candidate")) != normalized_plan["candidate"] or evidence.get("plan_sha256") != normalized_plan["plan_sha256"] ): raise ValueError("Prod Catalog rehearsal is not bound to the exact plan") started = evidence.get("started_at_unix") completed = evidence.get("completed_at_unix") duration = evidence.get("duration_seconds") if ( type(started) is not int or type(completed) is not int or type(duration) is not int or started < 0 or completed <= started or duration != completed - started ): raise ValueError("Prod Catalog rehearsal timing is invalid") raw_packs = evidence.get("installed_packs") if not isinstance(raw_packs, list) or not raw_packs: raise ValueError("Prod Catalog rehearsal installed_packs must be non-empty") pack_names: list[str] = [] expected_packs = { pack["name"]: pack for pack in normalized_plan["packs"] } for index, raw_pack in enumerate(raw_packs): pack = _require_object( raw_pack, INSTALLED_PACK_KEYS, f"Prod Catalog rehearsal installed_packs[{index}]", ) pack_name = _require_match( pack.get("name"), IDENTIFIER_RE, f"Prod Catalog rehearsal installed_packs[{index}].name", ) pack_names.append(pack_name) if pack_name not in expected_packs: raise ValueError( "Prod Catalog rehearsal pack set does not match the plan" ) if pack.get("status") != "active" or pack.get("last_error") != "": raise ValueError( "Prod Catalog rehearsal pack is not active and error-free" ) source_kind = _require_match( pack.get("source_kind"), IDENTIFIER_RE, f"Prod Catalog rehearsal installed_packs[{index}].source_kind", ) expected_pack = expected_packs[pack_name] if source_kind != expected_pack["source_kind"]: raise ValueError( "Prod Catalog rehearsal pack source does not match the plan" ) _require_rehearsal_source_uri( pack.get("source_uri"), expected_kind=source_kind, expected_uri=expected_pack["source_uri"], ) source_revision = _require_match( pack.get("source_revision"), GIT_COMMIT_RE, f"Prod Catalog rehearsal installed_packs[{index}].source_revision", ) if source_revision != expected_packs[pack_name]["source_revision"]: raise ValueError( "Prod Catalog rehearsal pack source revision does not match the plan" ) desired = _require_match( pack.get("desired_payload_hash"), PAYLOAD_HASH_RE, f"Prod Catalog rehearsal installed_packs[{index}].desired_payload_hash", ) applied = _require_match( pack.get("applied_payload_hash"), PAYLOAD_HASH_RE, f"Prod Catalog rehearsal installed_packs[{index}].applied_payload_hash", ) if desired != expected_packs[pack_name]["payload_hash"] or applied != desired: raise ValueError( "Prod Catalog rehearsal pack payload hashes do not match the plan" ) if pack_names != sorted(pack["name"] for pack in normalized_plan["packs"]): raise ValueError("Prod Catalog rehearsal pack set does not match the plan") unsigned = { key: item for key, item in evidence.items() if key != "evidence_sha256" } if evidence.get("evidence_sha256") != _digest(unsigned): raise ValueError("Prod Catalog rehearsal evidence_sha256 mismatch") return json.loads(json.dumps(evidence)) def build_prod_catalog_staging_rehearsal_observation( *, candidate_source: dict[str, object], started_at_unix: int, completed_at_unix: int, installed_packs: list[dict[str, str]], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: normalized_source = _require_candidate_source(candidate_source) synthetic_plan = build_prod_catalog_migration_plan( candidate={ "release_tag": "staging-rehearsal", **normalized_source, }, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) synthetic_rehearsal = build_prod_catalog_rehearsal_evidence( plan=synthetic_plan, started_at_unix=started_at_unix, completed_at_unix=completed_at_unix, installed_packs=installed_packs, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) observation: dict[str, Any] = { "schema_version": STAGING_REHEARSAL_OBSERVATION_SCHEMA_VERSION, "status": "PASS", "candidate_source": normalized_source, "operation_id": synthetic_plan["operation_id"], "packs": synthetic_plan["packs"], "started_at_unix": synthetic_rehearsal["started_at_unix"], "completed_at_unix": synthetic_rehearsal["completed_at_unix"], "duration_seconds": synthetic_rehearsal["duration_seconds"], "installed_packs": synthetic_rehearsal["installed_packs"], } observation["observation_sha256"] = _digest(observation) return require_prod_catalog_staging_rehearsal_observation( observation, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) def require_prod_catalog_staging_rehearsal_observation( value: object, *, candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: observation = _require_object( value, STAGING_REHEARSAL_OBSERVATION_KEYS, "Prod Catalog Staging rehearsal observation", ) if ( observation.get("schema_version") != STAGING_REHEARSAL_OBSERVATION_SCHEMA_VERSION ): raise ValueError( "Prod Catalog Staging rehearsal observation schema_version is unsupported" ) if observation.get("status") != "PASS": raise ValueError("Prod Catalog Staging rehearsal observation did not pass") candidate_source = _require_candidate_source( observation.get("candidate_source") ) synthetic_plan = build_prod_catalog_migration_plan( candidate={ "release_tag": "staging-rehearsal", **candidate_source, }, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) if ( observation.get("operation_id") != synthetic_plan["operation_id"] or observation.get("packs") != synthetic_plan["packs"] ): raise ValueError( "Prod Catalog Staging rehearsal observation does not match exact source" ) build_prod_catalog_rehearsal_evidence( plan=synthetic_plan, started_at_unix=observation.get("started_at_unix"), completed_at_unix=observation.get("completed_at_unix"), installed_packs=observation.get("installed_packs"), candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) if ( observation.get("duration_seconds") != observation.get("completed_at_unix") - observation.get("started_at_unix") ): raise ValueError("Prod Catalog Staging rehearsal observation timing is invalid") unsigned = { key: item for key, item in observation.items() if key != "observation_sha256" } if observation.get("observation_sha256") != _digest(unsigned): raise ValueError( "Prod Catalog Staging rehearsal observation_sha256 mismatch" ) return json.loads(json.dumps(observation)) def build_prod_catalog_rehearsal_evidence_from_observation( *, observation: dict[str, object], plan: dict[str, object], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: normalized_plan = require_prod_catalog_migration_plan( plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) normalized_observation = ( require_prod_catalog_staging_rehearsal_observation( observation, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) ) candidate_source = { key: value for key, value in normalized_plan["candidate"].items() if key != "release_tag" } if ( normalized_observation["candidate_source"] != candidate_source or normalized_observation["operation_id"] != normalized_plan["operation_id"] or normalized_observation["packs"] != normalized_plan["packs"] ): raise ValueError( "Prod Catalog Staging rehearsal observation is not bound to exact plan" ) return build_prod_catalog_rehearsal_evidence( plan=normalized_plan, started_at_unix=normalized_observation["started_at_unix"], completed_at_unix=normalized_observation["completed_at_unix"], installed_packs=normalized_observation["installed_packs"], candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) def build_prod_catalog_maintenance_evidence( *, plan: dict[str, object], rehearsal_evidence: dict[str, object], expected_source_readers: list[str], source_readers: list[dict[str, str]], maintenance_started_at_unix: int, write_maintenance_observed_at_unix: int, source_reader_stop_command_id: str, source_reader_stop_completed_at_unix: int, zero_nonterminal_workitems_command_id: str, workitems_drained_at_unix: int, drained_nonterminal_workitem_count: int, workitems_observed_at_unix: int, nonterminal_workitem_count: int, completed_at_unix: int, candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: normalized_plan = require_prod_catalog_migration_plan( plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) normalized_rehearsal = require_prod_catalog_rehearsal_evidence( rehearsal_evidence, plan=normalized_plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) evidence: dict[str, Any] = { "schema_version": MAINTENANCE_SCHEMA_VERSION, "status": "PASS", "candidate": normalized_plan["candidate"], "plan_sha256": normalized_plan["plan_sha256"], "rehearsal_evidence_sha256": normalized_rehearsal["evidence_sha256"], "maintenance_mode": "global-write-maintenance", "started_at_unix": maintenance_started_at_unix, "completed_at_unix": completed_at_unix, "duration_seconds": completed_at_unix - maintenance_started_at_unix, "write_maintenance": { "status": "active", "observed_at_unix": write_maintenance_observed_at_unix, }, "source_reader_stop": { "command_id": source_reader_stop_command_id, "status": "PASS", "completed_at_unix": source_reader_stop_completed_at_unix, "readers": sorted(source_readers, key=lambda item: item["name"]), }, "zero_nonterminal_workitems": { "command_id": zero_nonterminal_workitems_command_id, "status": "PASS", "drained_at_unix": workitems_drained_at_unix, "drained_nonterminal_count": drained_nonterminal_workitem_count, "observed_at_unix": workitems_observed_at_unix, "scope": "global", "table": "_System_ConnectorWorkItem", "nonterminal_count": nonterminal_workitem_count, }, } evidence["evidence_sha256"] = _digest(evidence) return require_prod_catalog_maintenance_evidence( evidence, plan=normalized_plan, rehearsal_evidence=normalized_rehearsal, expected_source_readers=expected_source_readers, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) def require_prod_catalog_maintenance_evidence( value: object, *, plan: dict[str, object], rehearsal_evidence: dict[str, object], expected_source_readers: list[str], candidate_bundle_root: Path, candidate_bundle_manifest: dict[str, object], runtime_metadata_env: dict[str, str] | None = None, ) -> dict[str, Any]: evidence = _require_object( value, MAINTENANCE_KEYS, "Prod Catalog maintenance evidence", ) normalized_plan = require_prod_catalog_migration_plan( plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) normalized_rehearsal = require_prod_catalog_rehearsal_evidence( rehearsal_evidence, plan=normalized_plan, candidate_bundle_root=candidate_bundle_root, candidate_bundle_manifest=candidate_bundle_manifest, runtime_metadata_env=runtime_metadata_env, ) if evidence.get("schema_version") != MAINTENANCE_SCHEMA_VERSION: raise ValueError("Prod Catalog maintenance schema_version is unsupported") if evidence.get("status") != "PASS": raise ValueError("Prod Catalog maintenance evidence did not pass") if ( _require_candidate(evidence.get("candidate")) != normalized_plan["candidate"] or evidence.get("plan_sha256") != normalized_plan["plan_sha256"] or evidence.get("rehearsal_evidence_sha256") != normalized_rehearsal["evidence_sha256"] ): raise ValueError( "Prod Catalog maintenance evidence is not bound to the exact rehearsal" ) if evidence.get("maintenance_mode") != "global-write-maintenance": raise ValueError("Prod Catalog maintenance mode is invalid") write_maintenance = _require_object( evidence.get("write_maintenance"), WRITE_MAINTENANCE_KEYS, "Prod Catalog write maintenance evidence", ) if write_maintenance.get("status") != "active": raise ValueError("Prod Catalog global write maintenance is not active") expected_readers = [ _require_match(reader, IDENTIFIER_RE, "expected Prod source reader") for reader in expected_source_readers ] if not expected_readers or expected_readers != sorted(set(expected_readers)): raise ValueError("expected Prod source readers must be non-empty and sorted") reader_stop = _require_object( evidence.get("source_reader_stop"), SOURCE_READER_STOP_KEYS, "Prod Catalog source-reader-stop evidence", ) reader_stop_command_id = _require_match( reader_stop.get("command_id"), COMMAND_ID_RE, "Prod Catalog source-reader-stop command_id", ) if reader_stop.get("status") != "PASS": raise ValueError("Prod Catalog source-reader-stop command did not pass") raw_readers = reader_stop.get("readers") if not isinstance(raw_readers, list): raise ValueError("Prod Catalog source readers must be a list") reader_names: list[str] = [] for index, raw_reader in enumerate(raw_readers): reader = _require_object( raw_reader, SOURCE_READER_KEYS, f"Prod Catalog source readers[{index}]", ) reader_names.append( _require_match( reader.get("name"), IDENTIFIER_RE, f"Prod Catalog source readers[{index}].name", ) ) if reader.get("state") != "inactive": raise ValueError("Prod Catalog source reader is not inactive") if ( type(reader.get("listener_count")) is not int or reader["listener_count"] != 0 ): raise ValueError("Prod Catalog source reader listener count is not zero") if reader_names != expected_readers: raise ValueError("Prod Catalog source reader set is incomplete") zero_workitems = _require_object( evidence.get("zero_nonterminal_workitems"), ZERO_NONTERMINAL_WORKITEM_KEYS, "Prod Catalog zero-nonterminal-WorkItem evidence", ) zero_workitems_command_id = _require_match( zero_workitems.get("command_id"), COMMAND_ID_RE, "Prod Catalog zero-nonterminal-WorkItem command_id", ) if zero_workitems_command_id == reader_stop_command_id: raise ValueError("Prod Catalog maintenance command IDs must be distinct") if zero_workitems.get("status") != "PASS": raise ValueError("Prod Catalog zero-nonterminal-WorkItem command did not pass") if ( zero_workitems.get("scope") != "global" or zero_workitems.get("table") != "_System_ConnectorWorkItem" ): raise ValueError("Prod Catalog WorkItem evidence scope is invalid") if any( type(zero_workitems.get(key)) is not int or zero_workitems[key] != 0 for key in ("drained_nonterminal_count", "nonterminal_count") ): raise ValueError("Prod Catalog nonterminal WorkItem count is not zero") started = evidence.get("started_at_unix") completed = evidence.get("completed_at_unix") duration = evidence.get("duration_seconds") write_observed = write_maintenance.get("observed_at_unix") workitems_drained = zero_workitems.get("drained_at_unix") reader_stop_completed = reader_stop.get("completed_at_unix") workitems_observed = zero_workitems.get("observed_at_unix") if ( any( type(timestamp) is not int for timestamp in ( started, completed, duration, write_observed, workitems_drained, reader_stop_completed, workitems_observed, ) ) or started < 0 or completed <= started or duration != completed - started or not ( started <= write_observed < workitems_drained < reader_stop_completed < workitems_observed <= completed ) ): raise ValueError("Prod Catalog maintenance timing is invalid") if started <= normalized_rehearsal["completed_at_unix"]: raise ValueError( "Prod Catalog rehearsal did not complete before maintenance" ) unsigned = { key: item for key, item in evidence.items() if key != "evidence_sha256" } if evidence.get("evidence_sha256") != _digest(unsigned): raise ValueError("Prod Catalog maintenance evidence_sha256 mismatch") return json.loads(json.dumps(evidence))