#!/usr/bin/env python3
"""Content-addressed shared-state contract for ordinary Prod promotion.

The contract describes L2-owned shared-state inputs and the effects allowed for
one Staging-proved candidate. Ordinary Prod defaults to the metadata-first v3
policy: L2 input changes proved by a reset-based Formal Staging run may enter
the default Catalog migration, while unproved changes remain blocked. The
executor still forbids deploy seed/reference-data and other non-Catalog Prod
effects. Runtime metadata compatibility is owned and enforced by L0; this
contract governs only the L2 release operation. Missing or unverifiable
authority remains blocked.
"""

from __future__ import annotations

import hashlib
import json
import subprocess
from pathlib import Path
from typing import Any

try:
    from .lib.final_pack_contract import (
        FinalPackContractError,
        final_pack_path_from_git as contract_final_pack_path_from_git,
        metadata_tree_sha256_from_git,
    )
except ImportError:  # pragma: no cover - direct script execution
    from lib.final_pack_contract import (
        FinalPackContractError,
        final_pack_path_from_git as contract_final_pack_path_from_git,
        metadata_tree_sha256_from_git,
    )


CONTRACT_VERSION = "ordinary-prod-promotion/v3"
ENFORCEMENT_STATUS = "enforced"
RUNTIME_ONLY_SHARED_STATE_POLICY = "runtime-only-compatible-baseline"
METADATA_FIRST_SHARED_STATE_POLICY = "metadata-first-catalog-baseline"
BASELINE_COMPARISON_VERSION = "ordinary-prod-baseline-comparison/v4"
BASELINE_ENFORCEMENT_STATUS = "enforced"
STAGING_PROVED_SHARED_STATE_MISMATCH = (
    "staging-proved-shared-state-mismatch"
)
ACTIVE_BASELINE_UNAVAILABLE_STATUSES = {
    "no-active-release",
    "missing-active-contract",
    "invalid-active-contract",
    "unavailable-active-contract",
}

ALLOWED_PROD_EFFECTS = (
    "candidate-local-files",
    "candidate-process-lifecycle",
    "runtime-readiness-verification",
    "runtime-standby-health",
    "active-slot-stop-at-cutover",
    "runtime-activation-restart",
    "runtime-service-definition",
    "nginx-route-activation",
    "redacted-release-evidence",
)

FORBIDDEN_PROD_EFFECTS = (
    "setup-or-pack-lifecycle-apply",
    "release-actor-shared-state-write",
    "deploy-seed-or-reference-data-apply",
    "connector-replay-or-admission-pause",
    "external-system-configuration-repair",
    "backup-branch-reset",
    "destructive-or-non-backward-compatible-shared-state-migration",
)

METADATA_FIRST_ALLOWED_PROD_EFFECTS = (
    "candidate-local-files",
    "candidate-process-lifecycle",
    "runtime-readiness-verification",
    "runtime-standby-health",
    "setup-or-pack-lifecycle-apply",
    "release-actor-shared-state-write",
    "connector-replay-or-admission-pause",
    "active-slot-stop-at-cutover",
    "runtime-activation-restart",
    "runtime-service-definition",
    "nginx-route-activation",
    "redacted-release-evidence",
)

METADATA_FIRST_FORBIDDEN_PROD_EFFECTS = (
    "deploy-seed-or-reference-data-apply",
    "external-system-configuration-repair",
    "backup-branch-reset",
    "destructive-or-non-backward-compatible-shared-state-migration",
)

# These are L2-owned inputs that currently feed production mutations outside
# the final merged pack.  Hashing the whole owned surface is intentionally
# conservative: an ordinary promotion must block for a separate migration when
# any of these inputs changes without reset-based Formal Staging proof.
SHARED_STATE_INPUTS = (
    "stack.lock.json",
    "packs/PingCAP-HubSpot-Companion",
    "deployments/offline/runtime/seeds",
    "deployments/offline/runtime/reference-data",
    "deployments/offline/runtime/reconcile",
    "deployments/offline/runtime/reconcile_hubspot_owner_directory.py",
    "scripts/apply_deploy_seed.py",
    "scripts/apply_pack_reference_data.py",
    "scripts/apply_pack_reference_data.sh",
)

TOP_LEVEL_KEYS = {
    "schema_version",
    "enforcement_status",
    "shared_state_policy",
    "allowed_prod_effects",
    "forbidden_prod_effects",
    "components",
    "shared_state_digest",
    "contract_sha256",
}

COMPONENT_KEYS = {
    "final_pack",
    "official_app_packs_source",
    "l2_shared_state_inputs",
}

BASELINE_RECEIPT_KEYS = {
    "schema_version",
    "enforcement_status",
    "status",
    "baseline_match",
    "candidate",
    "active",
    "next_action",
    "receipt_sha256",
}
BASELINE_CANDIDATE_KEYS = {
    "release_tag",
    "tag_target_commit",
    "contract_sha256",
    "shared_state_digest",
}
BASELINE_ACTIVE_KEYS = {
    "release_tag",
    "tag_target_commit",
    "contract_sha256",
    "shared_state_digest",
    "contract_source",
}


def _shared_state_projection(components: dict[str, Any]) -> dict[str, Any]:
    # The exact official app-packs commit remains candidate identity, but it is
    # not itself an L2 shared-state mutation signal.  Its selected pack payload
    # is independently covered by source-backed payload admission.  Treating
    # every unrelated L1 commit as a database migration would make the ordinary
    # release contract needlessly conservative.
    return {
        "final_pack": components["final_pack"],
        "l2_shared_state_inputs": components["l2_shared_state_inputs"],
    }


def _prod_effects_for_policy(
    shared_state_policy: str,
) -> tuple[tuple[str, ...], tuple[str, ...]]:
    if shared_state_policy == RUNTIME_ONLY_SHARED_STATE_POLICY:
        return ALLOWED_PROD_EFFECTS, FORBIDDEN_PROD_EFFECTS
    if shared_state_policy == METADATA_FIRST_SHARED_STATE_POLICY:
        return (
            METADATA_FIRST_ALLOWED_PROD_EFFECTS,
            METADATA_FIRST_FORBIDDEN_PROD_EFFECTS,
        )
    raise ValueError("Prod promotion contract shared_state_policy is invalid")


def _canonical_json(value: object) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")


def _sha256(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _require_sha256(value: object, label: str) -> str:
    normalized = str(value).strip().lower()
    if len(normalized) != 64 or any(ch not in "0123456789abcdef" for ch in normalized):
        raise ValueError(f"{label} must be a lowercase SHA-256")
    return normalized


def _require_git_commit(value: object, label: str) -> str:
    normalized = str(value).strip().lower()
    if len(normalized) != 40 or any(ch not in "0123456789abcdef" for ch in normalized):
        raise ValueError(f"{label} must be a lowercase 40-character git commit")
    return normalized


def _hash_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _path_entries(repo_root: Path, relative_path: str) -> list[dict[str, str]]:
    path = repo_root / relative_path
    if not path.exists():
        raise ValueError(f"Prod promotion input is missing: {relative_path}")
    if path.is_symlink():
        raise ValueError(f"Prod promotion input must not be a symlink: {relative_path}")
    if path.is_file():
        return [{"path": relative_path, "sha256": _hash_file(path)}]
    if not path.is_dir():
        raise ValueError(f"Prod promotion input is not a file or directory: {relative_path}")

    entries: list[dict[str, str]] = []
    for child in sorted(path.rglob("*"), key=lambda candidate: candidate.as_posix()):
        child_relative = child.relative_to(repo_root).as_posix()
        if child.is_symlink():
            raise ValueError(
                f"Prod promotion input must not contain symlinks: {child_relative}"
            )
        if child.is_file():
            entries.append({"path": child_relative, "sha256": _hash_file(child)})
    if not entries:
        raise ValueError(f"Prod promotion input directory is empty: {relative_path}")
    return entries


def l2_shared_state_input_manifest(repo_root: Path) -> dict[str, Any]:
    root = repo_root.expanduser().resolve()
    entries: list[dict[str, str]] = []
    for relative_path in SHARED_STATE_INPUTS:
        entries.extend(_path_entries(root, relative_path))
    entries.sort(key=lambda entry: entry["path"])
    return {
        "paths": list(SHARED_STATE_INPUTS),
        "file_count": len(entries),
        "digest": _sha256(_canonical_json(entries)),
    }


def l2_shared_state_input_manifest_from_source_tree(
    repo_root: Path,
) -> dict[str, Any]:
    """Hash a materialized exact source tree, preserving governed absences."""

    root = repo_root.expanduser().resolve()
    entries: list[dict[str, str]] = []
    for relative_path in SHARED_STATE_INPUTS:
        if not (root / relative_path).exists():
            entries.append({"path": relative_path, "state": "absent"})
            continue
        entries.extend(_path_entries(root, relative_path))
    entries.sort(key=lambda entry: entry["path"])
    return {
        "paths": list(SHARED_STATE_INPUTS),
        "file_count": len(entries),
        "digest": _sha256(_canonical_json(entries)),
    }


def l2_shared_state_input_manifest_from_git(
    repo_root: Path,
    revision: str,
) -> dict[str, Any]:
    """Hash governed inputs from an exact git tree without checking it out."""

    root = repo_root.expanduser().resolve()
    resolved_revision = subprocess.run(
        ["git", "-C", str(root), "rev-parse", "--verify", f"{revision}^{{commit}}"],
        check=True,
        capture_output=True,
        text=True,
    ).stdout.strip()
    _require_git_commit(resolved_revision, "active tag target commit")

    entries_by_path: dict[str, dict[str, str]] = {}
    for relative_path in SHARED_STATE_INPUTS:
        listing = subprocess.run(
            [
                "git",
                "-C",
                str(root),
                "ls-tree",
                "-r",
                "-z",
                "--full-tree",
                resolved_revision,
                "--",
                relative_path,
            ],
            check=True,
            capture_output=True,
        ).stdout
        if not listing:
            # A legacy active tag can predate a governed input. Preserve that
            # absence as content-addressed state so it becomes a deterministic
            # mismatch (migration required), not an unverifiable baseline.
            entries_by_path[f"absent:{relative_path}"] = {
                "path": relative_path,
                "state": "absent",
            }
            continue
        for raw_entry in listing.split(b"\0"):
            if not raw_entry:
                continue
            metadata, separator, raw_path = raw_entry.partition(b"\t")
            if not separator:
                raise ValueError("active git tree contains a malformed entry")
            mode, object_type, object_id = metadata.decode("ascii").split()
            path = raw_path.decode("utf-8")
            if mode == "120000":
                raise ValueError(
                    f"Prod promotion input must not be a symlink: {path}"
                )
            if object_type != "blob" or not mode.startswith("100"):
                raise ValueError(
                    f"Prod promotion input is not a regular file: {path}"
                )
            content = subprocess.run(
                ["git", "-C", str(root), "cat-file", "blob", object_id],
                check=True,
                capture_output=True,
            ).stdout
            entry = {"path": path, "sha256": _sha256(content)}
            existing = entries_by_path.get(path)
            if existing is not None and existing != entry:
                raise ValueError(
                    f"active git tree produced conflicting content for {path}"
                )
            entries_by_path[path] = entry

    entries = sorted(entries_by_path.values(), key=lambda entry: entry["path"])
    return {
        "paths": list(SHARED_STATE_INPUTS),
        "file_count": len(entries),
        "digest": _sha256(_canonical_json(entries)),
    }


def _load_git_json(
    repo_root: Path,
    revision: str,
    path: str,
) -> dict[str, Any]:
    root = repo_root.expanduser().resolve()
    raw = subprocess.run(
        ["git", "-C", str(root), "show", f"{revision}:{path}"],
        check=True,
        capture_output=True,
        text=True,
    ).stdout
    value = json.loads(raw)
    if not isinstance(value, dict):
        raise ValueError(f"{path} in exact Git tree must be a JSON object")
    return value


def _load_source_json(repo_root: Path, path: str) -> dict[str, Any]:
    source = repo_root / path
    if source.is_symlink() or not source.is_file():
        raise ValueError(f"{path} in candidate source must be a regular file")
    try:
        value = json.loads(source.read_text())
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"{path} in candidate source is invalid: {exc}") from exc
    if not isinstance(value, dict):
        raise ValueError(f"{path} in candidate source must be a JSON object")
    return value


def _final_pack_path_from_source_tree(repo_root: Path) -> str:
    stack_lock = _load_source_json(repo_root, "stack.lock.json")
    pack_build = stack_lock.get("pack_build")
    if not isinstance(pack_build, dict):
        raise ValueError("stack.lock.json pack_build is missing in candidate source")
    declared_path = str(
        pack_build.get("final_pack_path")
        or "packs/NOT-EDIT-final-merged-packs"
    ).strip()
    declared = Path(declared_path)
    if (
        not declared_path
        or declared.is_absolute()
        or ".." in declared.parts
        or declared_path.endswith("/")
    ):
        raise ValueError("stack.lock.json final_pack_path is invalid in candidate source")
    return declared.as_posix()


def _final_pack_path_from_git(repo_root: Path, revision: str) -> str:
    try:
        return contract_final_pack_path_from_git(repo_root, revision)
    except FinalPackContractError as exc:
        raise ValueError(str(exc)) from exc


def final_pack_identity_from_git(
    repo_root: Path,
    revision: str,
) -> tuple[str, str]:
    """Read final-pack name/version from the exact candidate Git tree."""

    final_pack_path = _final_pack_path_from_git(repo_root, revision)
    pack = _load_git_json(
        repo_root,
        revision,
        f"{final_pack_path}/pack.json",
    )
    pack_name = str(pack.get("name", "")).strip()
    pack_version = str(pack.get("version", "")).strip()
    if not pack_name or not pack_version:
        raise ValueError("final pack identity is incomplete in exact Git tree")
    return pack_name, pack_version


def official_app_packs_commit_from_git(
    repo_root: Path,
    revision: str,
) -> str:
    """Read the exact official app-packs source from final-pack lineage."""

    final_pack_path = _final_pack_path_from_git(repo_root, revision)
    lineage = _load_git_json(
        repo_root,
        revision,
        f"{final_pack_path}/.build-lineage.json",
    )
    official_pack = lineage.get("officialPack")
    if not isinstance(official_pack, dict):
        raise ValueError("final pack lineage officialPack is missing in exact Git tree")
    return _require_git_commit(
        official_pack.get("ref"),
        "final pack lineage officialPack.ref",
    )


def final_pack_payload_hash_from_git(
    repo_root: Path,
    revision: str,
) -> str:
    """Reproduce the release final-pack hash from an exact Git tree.

    The release hash is the SHA-256 of the sorted ``sha256sum`` manifest for
    every final-pack file except ``.build-lineage.json``.  Reading blobs through
    Git keeps the result bound to the selected commit rather than a mutable
    checkout or runtime environment file.
    """

    root = repo_root.expanduser().resolve()
    final_pack_path = _final_pack_path_from_git(root, revision)
    try:
        return metadata_tree_sha256_from_git(
            root,
            revision,
            final_pack_path,
        ).removeprefix("sha256:")
    except FinalPackContractError as exc:
        raise ValueError(str(exc)) from exc


def final_pack_payload_hash_from_source_tree(repo_root: Path) -> str:
    """Reproduce the release final-pack hash from materialized candidate bytes."""

    root = repo_root.expanduser().resolve()
    final_pack_path = _final_pack_path_from_source_tree(root)
    final_pack_root = root / final_pack_path
    if final_pack_root.is_symlink() or not final_pack_root.is_dir():
        raise ValueError("candidate source final pack is missing or invalid")
    manifest_lines: list[bytes] = []
    for path in sorted(
        final_pack_root.rglob("*"),
        key=lambda candidate: candidate.relative_to(final_pack_root).as_posix(),
    ):
        if path.is_dir():
            continue
        relative_path = path.relative_to(final_pack_root).as_posix()
        if path.is_symlink() or not path.is_file():
            raise ValueError(
                f"candidate source final pack contains an unsupported path: {relative_path}"
            )
        if path.name == ".build-lineage.json":
            continue
        if "\n" in relative_path or "\\" in relative_path:
            raise ValueError(
                "candidate source final pack contains an unsupported file name"
            )
        manifest_lines.append(
            f"{_hash_file(path)}  ./{relative_path}\n".encode("utf-8")
        )
    if not manifest_lines:
        raise ValueError("candidate source final pack has no governed payload files")
    return _sha256(b"".join(manifest_lines))


def _build_prod_promotion_contract(
    *,
    l2_inputs: dict[str, Any],
    final_pack_hash: str,
    final_pack_name: str,
    final_pack_version: str,
    official_app_packs_commit: str,
    shared_state_policy: str = METADATA_FIRST_SHARED_STATE_POLICY,
) -> dict[str, Any]:
    pack_hash = _require_sha256(final_pack_hash, "final_pack_hash")
    pack_name = final_pack_name.strip()
    pack_version = final_pack_version.strip()
    if not pack_name or not pack_version:
        raise ValueError("final pack name and version are required")
    official_commit = _require_git_commit(
        official_app_packs_commit,
        "official_app_packs_commit",
    )
    components = {
        "final_pack": {
            "name": pack_name,
            "version": pack_version,
            "payload_sha256": pack_hash,
        },
        "official_app_packs_source": {
            "commit": official_commit,
        },
        "l2_shared_state_inputs": l2_inputs,
    }
    shared_state_digest = _sha256(
        _canonical_json(_shared_state_projection(components))
    )
    allowed_prod_effects, forbidden_prod_effects = _prod_effects_for_policy(
        shared_state_policy
    )
    contract: dict[str, Any] = {
        "schema_version": CONTRACT_VERSION,
        "enforcement_status": ENFORCEMENT_STATUS,
        "shared_state_policy": shared_state_policy,
        "allowed_prod_effects": list(allowed_prod_effects),
        "forbidden_prod_effects": list(forbidden_prod_effects),
        "components": components,
        "shared_state_digest": shared_state_digest,
    }
    contract["contract_sha256"] = _sha256(_canonical_json(contract))
    return contract


def build_prod_promotion_contract(
    repo_root: Path,
    *,
    final_pack_hash: str,
    final_pack_name: str,
    final_pack_version: str,
    official_app_packs_commit: str,
    shared_state_policy: str = METADATA_FIRST_SHARED_STATE_POLICY,
) -> dict[str, Any]:
    return _build_prod_promotion_contract(
        l2_inputs=l2_shared_state_input_manifest(repo_root),
        final_pack_hash=final_pack_hash,
        final_pack_name=final_pack_name,
        final_pack_version=final_pack_version,
        official_app_packs_commit=official_app_packs_commit,
        shared_state_policy=shared_state_policy,
    )


def build_prod_promotion_contract_from_source_tree(
    repo_root: Path,
    *,
    shared_state_policy: str = METADATA_FIRST_SHARED_STATE_POLICY,
) -> dict[str, Any]:
    """Derive the contract from exact materialized source without Git metadata."""

    root = repo_root.expanduser().resolve()
    final_pack_path = _final_pack_path_from_source_tree(root)
    pack = _load_source_json(root, f"{final_pack_path}/pack.json")
    pack_name = str(pack.get("name", "")).strip()
    pack_version = str(pack.get("version", "")).strip()
    if not pack_name or not pack_version:
        raise ValueError("final pack identity is incomplete in candidate source")
    lineage = _load_source_json(
        root,
        f"{final_pack_path}/.build-lineage.json",
    )
    official_pack = lineage.get("officialPack")
    if not isinstance(official_pack, dict):
        raise ValueError(
            "final pack lineage officialPack is missing in candidate source"
        )
    return _build_prod_promotion_contract(
        l2_inputs=l2_shared_state_input_manifest_from_source_tree(root),
        final_pack_hash=final_pack_payload_hash_from_source_tree(root),
        final_pack_name=pack_name,
        final_pack_version=pack_version,
        official_app_packs_commit=official_pack.get("ref"),
        shared_state_policy=shared_state_policy,
    )


def build_prod_promotion_contract_from_git(
    repo_root: Path,
    *,
    revision: str,
    final_pack_hash: str | None = None,
    final_pack_name: str | None = None,
    final_pack_version: str | None = None,
    official_app_packs_commit: str | None = None,
    shared_state_policy: str = METADATA_FIRST_SHARED_STATE_POLICY,
) -> dict[str, Any]:
    exact_pack_name, exact_pack_version = final_pack_identity_from_git(
        repo_root,
        revision,
    )
    if final_pack_name is not None and final_pack_name.strip() != exact_pack_name:
        raise ValueError(
            "declared final pack name does not match the exact Git tree"
        )
    if (
        final_pack_version is not None
        and final_pack_version.strip() != exact_pack_version
    ):
        raise ValueError(
            "declared final pack version does not match the exact Git tree"
        )
    exact_pack_hash = final_pack_payload_hash_from_git(repo_root, revision)
    if (
        final_pack_hash is not None
        and _require_sha256(final_pack_hash, "final_pack_hash")
        != exact_pack_hash
    ):
        raise ValueError(
            "declared final pack hash does not match the exact Git tree"
        )
    exact_official_commit = official_app_packs_commit_from_git(
        repo_root,
        revision,
    )
    if (
        official_app_packs_commit is not None
        and _require_git_commit(
            official_app_packs_commit,
            "official_app_packs_commit",
        )
        != exact_official_commit
    ):
        raise ValueError(
            "declared official app-packs commit does not match the exact Git tree"
        )
    return _build_prod_promotion_contract(
        l2_inputs=l2_shared_state_input_manifest_from_git(repo_root, revision),
        final_pack_hash=exact_pack_hash,
        final_pack_name=exact_pack_name,
        final_pack_version=exact_pack_version,
        official_app_packs_commit=exact_official_commit,
        shared_state_policy=shared_state_policy,
    )


def require_prod_promotion_contract(
    value: object,
    *,
    expected_final_pack_hash: str | None = None,
    expected_official_app_packs_commit: str | None = None,
) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError("Prod promotion contract must be a JSON object")
    if set(value) != TOP_LEVEL_KEYS:
        raise ValueError("Prod promotion contract has unexpected or missing fields")
    if value.get("schema_version") != CONTRACT_VERSION:
        raise ValueError("Prod promotion contract schema_version is unsupported")
    if value.get("enforcement_status") != ENFORCEMENT_STATUS:
        raise ValueError("Prod promotion contract enforcement_status is unsupported")
    shared_state_policy = str(value.get("shared_state_policy") or "")
    allowed_prod_effects, forbidden_prod_effects = _prod_effects_for_policy(
        shared_state_policy
    )
    if value.get("allowed_prod_effects") != list(allowed_prod_effects):
        raise ValueError("Prod promotion contract allowed_prod_effects changed")
    if value.get("forbidden_prod_effects") != list(forbidden_prod_effects):
        raise ValueError("Prod promotion contract forbidden_prod_effects changed")

    components = value.get("components")
    if not isinstance(components, dict) or set(components) != COMPONENT_KEYS:
        raise ValueError("Prod promotion contract components are invalid")
    final_pack = components.get("final_pack")
    if not isinstance(final_pack, dict) or set(final_pack) != {
        "name",
        "version",
        "payload_sha256",
    }:
        raise ValueError("Prod promotion contract final_pack component is invalid")
    if not str(final_pack.get("name", "")).strip() or not str(
        final_pack.get("version", "")
    ).strip():
        raise ValueError("Prod promotion contract final_pack identity is incomplete")
    final_pack_payload_sha256 = _require_sha256(
        final_pack.get("payload_sha256"),
        "final_pack.payload_sha256",
    )
    if (
        expected_final_pack_hash is not None
        and final_pack_payload_sha256
        != _require_sha256(expected_final_pack_hash, "expected_final_pack_hash")
    ):
        raise ValueError(
            "Prod promotion contract final pack does not match the candidate"
        )

    official = components.get("official_app_packs_source")
    if not isinstance(official, dict) or set(official) != {"commit"}:
        raise ValueError("Prod promotion contract official source component is invalid")
    official_commit = _require_git_commit(
        official.get("commit"),
        "official_app_packs_source.commit",
    )
    if (
        expected_official_app_packs_commit is not None
        and official_commit
        != _require_git_commit(
            expected_official_app_packs_commit,
            "expected_official_app_packs_commit",
        )
    ):
        raise ValueError(
            "Prod promotion contract official app-packs source does not match the candidate"
        )

    l2_inputs = components.get("l2_shared_state_inputs")
    if not isinstance(l2_inputs, dict) or set(l2_inputs) != {
        "paths",
        "file_count",
        "digest",
    }:
        raise ValueError("Prod promotion contract L2 input component is invalid")
    if l2_inputs.get("paths") != list(SHARED_STATE_INPUTS):
        raise ValueError("Prod promotion contract L2 input paths changed")
    if not isinstance(l2_inputs.get("file_count"), int) or l2_inputs["file_count"] <= 0:
        raise ValueError("Prod promotion contract L2 input file_count is invalid")
    _require_sha256(l2_inputs.get("digest"), "l2_shared_state_inputs.digest")

    expected_shared_state_digest = _sha256(
        _canonical_json(_shared_state_projection(components))
    )
    if value.get("shared_state_digest") != expected_shared_state_digest:
        raise ValueError("Prod promotion contract shared_state_digest mismatch")
    unsigned = {key: item for key, item in value.items() if key != "contract_sha256"}
    if value.get("contract_sha256") != _sha256(_canonical_json(unsigned)):
        raise ValueError("Prod promotion contract contract_sha256 mismatch")
    return json.loads(json.dumps(value))


def build_active_baseline_comparison(
    candidate_contract: object,
    *,
    candidate_release_tag: str,
    candidate_tag_target_commit: str,
    active_release_tag: str,
    active_tag_target_commit: str,
    active_contract: object | None,
    active_contract_status: str = "available",
    active_contract_source: str = "embedded-tag-contract",
    catalog_migration: bool = False,
    staging_proved_l2_migration: bool = False,
) -> dict[str, Any]:
    candidate = require_prod_promotion_contract(candidate_contract)
    if staging_proved_l2_migration and not catalog_migration:
        raise ValueError(
            "reset-based Formal Staging proof requires Catalog migration mode"
        )
    if (
        catalog_migration
        and candidate["shared_state_policy"]
        != METADATA_FIRST_SHARED_STATE_POLICY
    ):
        raise ValueError(
            "Catalog migration requires a metadata-first Prod promotion contract"
        )
    normalized_candidate_tag = candidate_release_tag.strip()
    if not normalized_candidate_tag:
        raise ValueError("candidate release tag is required")
    normalized_candidate_target = _require_git_commit(
        candidate_tag_target_commit,
        "candidate tag target commit",
    )
    normalized_active_tag = active_release_tag.strip()
    if active_contract_status == "available":
        if not normalized_active_tag:
            raise ValueError(
                "active release tag is required when the active contract is available"
            )
        normalized_active_target = _require_git_commit(
            active_tag_target_commit,
            "active tag target commit",
        )
        active = require_prod_promotion_contract(active_contract)
        if active["shared_state_digest"] == candidate["shared_state_digest"]:
            status = "match"
        elif (
            catalog_migration
            and active["components"]["l2_shared_state_inputs"]["digest"]
            != candidate["components"]["l2_shared_state_inputs"]["digest"]
        ):
            status = (
                STAGING_PROVED_SHARED_STATE_MISMATCH
                if staging_proved_l2_migration
                else "non-catalog-shared-state-mismatch"
            )
        else:
            status = "mismatch"
        active_identity: dict[str, object] = {
            "release_tag": normalized_active_tag,
            "tag_target_commit": normalized_active_target,
            "contract_sha256": active["contract_sha256"],
            "shared_state_digest": active["shared_state_digest"],
            "contract_source": active_contract_source,
        }
    else:
        if active_contract_status not in ACTIVE_BASELINE_UNAVAILABLE_STATUSES:
            raise ValueError(
                "unsupported active runtime baseline contract status"
            )
        if active_contract is not None:
            raise ValueError(
                "unavailable active runtime baseline must not include a contract"
            )
        status = active_contract_status
        active_identity = {
            "release_tag": normalized_active_tag,
            "tag_target_commit": "",
            "contract_sha256": "",
            "shared_state_digest": "",
            "contract_source": "",
        }

    receipt: dict[str, Any] = {
        "schema_version": BASELINE_COMPARISON_VERSION,
        "enforcement_status": BASELINE_ENFORCEMENT_STATUS,
        "status": status,
        "baseline_match": status == "match",
        "candidate": {
            "release_tag": normalized_candidate_tag,
            "tag_target_commit": normalized_candidate_target,
            "contract_sha256": candidate["contract_sha256"],
            "shared_state_digest": candidate["shared_state_digest"],
        },
        "active": active_identity,
        "next_action": (
            "run-default-catalog-migration-and-cutover"
            if catalog_migration
            and status
            in {
                "match",
                "mismatch",
                STAGING_PROVED_SHARED_STATE_MISMATCH,
            }
            else (
                "run-governed-non-catalog-shared-state-migration"
                if catalog_migration
                and status == "non-catalog-shared-state-mismatch"
                else "run-runtime-only-blue-green-cutover"
                if status == "match"
                else "run-separate-governed-migration-and-restage"
                if status == "mismatch"
                else "establish-active-baseline-authority"
            )
        ),
    }
    receipt["receipt_sha256"] = _sha256(_canonical_json(receipt))
    return receipt


def require_prod_baseline_receipt(
    value: object,
    *,
    catalog_migration: bool = False,
) -> dict[str, Any]:
    """Validate the signed shape of one active baseline comparison."""

    if not isinstance(value, dict) or set(value) != BASELINE_RECEIPT_KEYS:
        raise ValueError("ordinary Prod baseline receipt has unexpected or missing fields")
    if value.get("schema_version") != BASELINE_COMPARISON_VERSION:
        raise ValueError("ordinary Prod baseline receipt schema_version is unsupported")
    if value.get("enforcement_status") != BASELINE_ENFORCEMENT_STATUS:
        raise ValueError("ordinary Prod baseline receipt is not enforced")
    unsigned = {key: item for key, item in value.items() if key != "receipt_sha256"}
    if value.get("receipt_sha256") != _sha256(_canonical_json(unsigned)):
        raise ValueError("ordinary Prod baseline receipt_sha256 mismatch")
    status = value.get("status")
    if status not in {
        "match",
        "mismatch",
        STAGING_PROVED_SHARED_STATE_MISMATCH,
    }:
        raise ValueError(
            "Prod candidate has no exact active baseline for promotion or migration"
        )
    if (
        status == STAGING_PROVED_SHARED_STATE_MISMATCH
        and not catalog_migration
    ):
        raise ValueError(
            "reset-based Formal Staging proof requires Catalog migration mode"
        )
    expected_match = status == "match"
    if value.get("baseline_match") is not expected_match:
        raise ValueError("ordinary Prod baseline receipt match flag is invalid")
    active = value.get("active")
    candidate = value.get("candidate")
    if (
        not isinstance(active, dict)
        or set(active) != BASELINE_ACTIVE_KEYS
        or not isinstance(candidate, dict)
        or set(candidate) != BASELINE_CANDIDATE_KEYS
    ):
        raise ValueError("ordinary Prod baseline receipt identities are invalid")
    if not str(candidate.get("release_tag", "")).strip():
        raise ValueError("ordinary Prod candidate release tag is missing")
    _require_git_commit(
        candidate.get("tag_target_commit"),
        "ordinary Prod candidate tag target commit",
    )
    if not str(active.get("release_tag", "")).strip():
        raise ValueError("ordinary Prod active release tag is missing")
    _require_git_commit(
        active.get("tag_target_commit"),
        "ordinary Prod active tag target commit",
    )
    shared_state_matches = (
        active.get("shared_state_digest") == candidate.get("shared_state_digest")
    )
    if shared_state_matches is not expected_match:
        raise ValueError(
            "ordinary Prod baseline receipt shared-state relation is invalid"
        )
    expected_next_action = (
        "run-default-catalog-migration-and-cutover"
        if catalog_migration
        else "run-runtime-only-blue-green-cutover"
        if expected_match
        else "run-separate-governed-migration-and-restage"
    )
    if value.get("next_action") != expected_next_action:
        raise ValueError("ordinary Prod baseline receipt next_action is invalid")
    if active.get("contract_source") not in {
        "embedded-tag-contract",
        "derived-exact-active-tag",
    }:
        raise ValueError("ordinary Prod active baseline authority is invalid")
    _require_sha256(active.get("contract_sha256"), "active contract_sha256")
    _require_sha256(candidate.get("contract_sha256"), "candidate contract_sha256")
    _require_sha256(active.get("shared_state_digest"), "active shared_state_digest")
    _require_sha256(
        candidate.get("shared_state_digest"),
        "candidate shared_state_digest",
    )
    return json.loads(json.dumps(value))


def require_prod_baseline_transition(
    value: object,
    *,
    catalog_migration: bool = False,
    candidate_contract: object | None = None,
    active_contract: object | None = None,
    staging_proved_l2_migration: bool = False,
) -> dict[str, Any]:
    """Validate an exact candidate and known active baseline comparison."""

    normalized = require_prod_baseline_receipt(
        value,
        catalog_migration=catalog_migration,
    )
    if catalog_migration:
        if candidate_contract is None or active_contract is None:
            raise ValueError(
                "Catalog migration requires exact candidate and active contracts"
            )
        candidate_authority = require_prod_promotion_contract(candidate_contract)
        active_authority = require_prod_promotion_contract(active_contract)
        if (
            candidate_authority["shared_state_policy"]
            != METADATA_FIRST_SHARED_STATE_POLICY
        ):
            raise ValueError(
                "Catalog migration requires a metadata-first Prod promotion contract"
            )
        if (
            normalized["candidate"].get("contract_sha256")
            != candidate_authority["contract_sha256"]
            or normalized["candidate"].get("shared_state_digest")
            != candidate_authority["shared_state_digest"]
            or normalized["active"].get("contract_sha256")
            != active_authority["contract_sha256"]
            or normalized["active"].get("shared_state_digest")
            != active_authority["shared_state_digest"]
        ):
            raise ValueError(
                "Catalog migration baseline receipt is not bound to the exact contracts"
            )
        l2_shared_state_changed = (
            active_authority["components"]["l2_shared_state_inputs"]["digest"]
            != candidate_authority["components"]["l2_shared_state_inputs"][
                "digest"
            ]
        )
        if l2_shared_state_changed:
            if normalized["status"] != STAGING_PROVED_SHARED_STATE_MISMATCH:
                raise ValueError(
                    "Prod shared-state change is outside the default Catalog migration scope"
                )
            if not staging_proved_l2_migration:
                raise ValueError(
                    "Prod shared-state migration requires reset-based Formal Staging proof"
                )
        elif normalized["status"] == STAGING_PROVED_SHARED_STATE_MISMATCH:
            raise ValueError(
                "Prod baseline receipt claims reset-based Formal Staging proof "
                "without an L2 shared-state change"
            )
    return normalized