#!/usr/bin/env python3
from __future__ import annotations

from collections import deque
import base64
import hashlib
import hmac
import json
import os
import queue
import re
import secrets
import shlex
import shutil
import signal
import subprocess
import sys
import tempfile
import textwrap
import threading
import time
from pathlib import Path, PurePosixPath
from typing import Callable, Iterable
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode, urlparse
from urllib.request import Request, urlopen

REPO_ROOT = Path(__file__).resolve().parents[3]
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from scripts import deployment_bundle

DEFAULT_PATH_PREFIX = "/usr/local/go/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
os.environ["PATH"] = f"{DEFAULT_PATH_PREFIX}:{os.environ.get('PATH', '')}"
RELEASE_TAG_JSON_MARKER = "--- release-metadata.json ---"
GIT_COMMIT_RE = re.compile(r"^[0-9a-f]{40}$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
HUBSPOT_SIGNATURE_VERIFIER_PATH = (
    Path("hubspot-crm-connector")
    / "inbound_handler_configs"
    / "signature_verifier.json"
)
STALE_HUBSPOT_SIGNATURE_IDENTITY_FIELDS = (
    "deliveryIdentityEventArrayPath",
    "deliveryIdentityEventKeyPaths",
)
PATCH_BUNDLE_MANIFEST_NAMES = (
    "patch-bundle.json",
    "patch-manifest.json",
)
PATCH_BUNDLE_ENVIRONMENTS = frozenset(
    {"prod", "staging", "test1", "test2", "test3"}
)
DEFAULT_RUNTIME_ROUTE_PROBE_OBJECTS = (
    "Countries_Region__c",
    "Country_Dictionary__c",
)
MYSQL_SSL_MODE_SUPPORT_CACHE: dict[str, bool] = {}
LEGACY_FILE_STORAGE_ROOT = "backend/data/object-storage"
LEGACY_FILE_STORAGE_LEGACY_UPLOAD_ROOT = "backend/data/uploads"
DEFAULT_SETUP_COMMAND_TIMEOUT_SECONDS = 1800
SETUP_COMMAND_DIAGNOSTIC_GRACE_SECONDS = 5
SETUP_COMMAND_OUTPUT_TAIL_LINES = 200
HUBSPOT_WEBHOOK_SECRET_NAMES = ("webhook", "workflow-action")
class DeployError(RuntimeError):
    pass


class SetupCommandError(DeployError):
    def __init__(self, message: str, *, output_tail: str) -> None:
        super().__init__(message)
        self.output_tail = output_tail


SETUP_TRANSIENT_CONNECTION_MARKERS = (
    "error 2013",
    "error 2026",
    "unexpected eof",
    "invalid connection",
    "lost connection",
    "server has gone away",
)


def is_transient_setup_connection_error(error: SetupCommandError) -> bool:
    lowered = error.output_tail.lower()
    return any(marker in lowered for marker in SETUP_TRANSIENT_CONNECTION_MARKERS)


SENSITIVE_ARG_KEYS = {
    "--password",
    "--db-password",
    "--target-password",
    "--source-password",
    "--app-secret",
    "--client-secret",
    "--secret",
    "--token",
    "--private-key",
}
SENSITIVE_IDENTITY_ARG_KEYS = {
    "-u",
    "--user",
    "--db-user",
    "--target-user",
    "--source-user",
}
SENSITIVE_ENV_NAMES = (
    "PASSWORD",
    "SECRET",
    "TOKEN",
    "PRIVATE_KEY",
)


def log(message: str) -> None:
    print(f"[deploy] {message}", flush=True)


def _is_sensitive_env_key(key: str) -> bool:
    upper = key.upper()
    return any(marker in upper for marker in SENSITIVE_ENV_NAMES)


def _redact_value(value: str) -> str:
    return "<redacted>" if value else value


def redact_command_args(args: list[str]) -> list[str]:
    redacted: list[str] = []
    redact_next = False
    for arg in args:
        if redact_next:
            redacted.append("<redacted>")
            redact_next = False
            continue
        if arg in SENSITIVE_ARG_KEYS or arg in SENSITIVE_IDENTITY_ARG_KEYS:
            redacted.append(arg)
            redact_next = True
            continue
        if any(arg.startswith(f"{key}=") for key in SENSITIVE_ARG_KEYS | SENSITIVE_IDENTITY_ARG_KEYS):
            key, _ = arg.split("=", 1)
            redacted.append(f"{key}=<redacted>")
            continue
        if arg.startswith("-p") and len(arg) > 2:
            redacted.append("-p<redacted>")
            continue
        if arg.startswith("-u") and len(arg) > 2:
            redacted.append("-u<redacted>")
            continue
        redacted.append(arg)
    return redacted


def _sensitive_values_from_args(args: list[str]) -> set[str]:
    values: set[str] = set()
    capture_next = False
    for arg in args:
        if capture_next:
            if arg:
                values.add(arg)
            capture_next = False
            continue
        if arg in SENSITIVE_ARG_KEYS or arg in SENSITIVE_IDENTITY_ARG_KEYS:
            capture_next = True
            continue
        if any(arg.startswith(f"{key}=") for key in SENSITIVE_ARG_KEYS | SENSITIVE_IDENTITY_ARG_KEYS):
            _, value = arg.split("=", 1)
            if value:
                values.add(value)
            continue
        if arg.startswith("-p") and len(arg) > 2:
            values.add(arg[2:])
            continue
        if arg.startswith("-u") and len(arg) > 2:
            values.add(arg[2:])
            continue
    return values


def redact_sensitive_text(
    text: str,
    env: dict[str, str] | None = None,
    args: list[str] | None = None,
    explicit_sensitive_values: Iterable[str] | None = None,
) -> str:
    redacted = text or ""
    candidates: set[str] = _sensitive_values_from_args(args or [])
    candidates.update(value for value in explicit_sensitive_values or () if value)
    for source in (os.environ, env or {}):
        for key, value in source.items():
            if _is_sensitive_env_key(key) and value:
                candidates.add(value)
    for value in sorted(candidates, key=len, reverse=True):
        if len(value) < 4:
            continue
        redacted = redacted.replace(value, _redact_value(value))
    redacted = re.sub(r"(-p)([^ \n\t]+)", r"\1<redacted>", redacted)
    redacted = re.sub(r"(-u)([^ \n\t]+)", r"\1<redacted>", redacted)
    redacted = re.sub(r"(--(?:db-)?password(?:=|\s+))([^ \n\t]+)", r"\1<redacted>", redacted, flags=re.IGNORECASE)
    return redacted


def named_credential_master_key_from_env() -> str:
    value = os.environ.get("NEXUSCRM_NC_MASTER_KEY", "").strip()
    if value:
        try:
            decoded = base64.b64decode(value, validate=True)
        except Exception as exc:
            raise DeployError("NEXUSCRM_NC_MASTER_KEY must be standard base64") from exc
        if len(decoded) != 32:
            raise DeployError(
                f"NEXUSCRM_NC_MASTER_KEY must decode to 32 bytes, got {len(decoded)}"
            )
        return value
    return base64.b64encode(secrets.token_bytes(32)).decode("ascii")


def named_credential_master_key_sha256(master_key_b64: str) -> str:
    try:
        decoded = base64.b64decode(master_key_b64.strip(), validate=True)
    except Exception as exc:
        raise DeployError("NEXUSCRM_NC_MASTER_KEY must be standard base64") from exc
    if len(decoded) != 32:
        raise DeployError(
            f"NEXUSCRM_NC_MASTER_KEY must decode to 32 bytes, got {len(decoded)}"
        )
    return hashlib.sha256(decoded).hexdigest()


def run(
    args: list[str],
    *,
    cwd: Path | None = None,
    env: dict[str, str] | None = None,
    inherit_env: bool = True,
    capture: bool = False,
    stdin: str | None = None,
    allow_exit_codes: set[int] | None = None,
) -> str:
    merged_env = os.environ.copy() if inherit_env else {}
    if env:
        merged_env.update(env)
    result = subprocess.run(
        args,
        cwd=str(cwd) if cwd else None,
        env=merged_env,
        text=True,
        input=stdin,
        capture_output=capture,
        check=False,
    )
    allowed = allow_exit_codes or {0}
    if result.returncode not in allowed:
        if capture:
            sys.stderr.write(redact_sensitive_text(result.stdout, merged_env, args))
            sys.stderr.write(redact_sensitive_text(result.stderr, merged_env, args))
        command = " ".join(redact_command_args(args))
        raise DeployError(f"command failed ({result.returncode}): {command}")
    return result.stdout if capture else ""


def require_commands(*names: str) -> None:
    missing = [name for name in names if not shutil.which(name)]
    if missing:
        raise DeployError(f"missing required commands: {', '.join(missing)}")


def shell_quote_env_value(value: str) -> str:
    return shlex.quote(value)


def shell_unquote_env_value(value: str) -> str:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
        return value[1:-1]
    return value


def load_env_file(path: Path) -> dict[str, str]:
    if not path.is_file():
        raise DeployError(f"env file not found: {path}")
    values: dict[str, str] = {}
    for raw_line in path.read_text().splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        if line.startswith("export "):
            line = line[len("export "):].strip()
        key, value = line.split("=", 1)
        values[key] = shell_unquote_env_value(value)
    return values


def require_json_object(value: object, label: str) -> dict[str, object]:
    if not isinstance(value, dict):
        raise DeployError(f"invalid {label}: expected json object")
    return value


def final_pack_tree_hash(root: Path) -> str:
    """Return the canonical served-metadata digest without its algorithm prefix."""
    try:
        return deployment_bundle.tree_sha256(root)
    except deployment_bundle.BundleError as exc:
        raise DeployError(f"cannot hash final pack {root}: {exc}") from exc


def runtime_deploy_pack_entries(lock_data: dict[str, object]) -> list[str]:
    pack_entries = lock_data.get("packs")
    if isinstance(pack_entries, list):
        normalized = [str(entry).strip() for entry in pack_entries if str(entry).strip()]
        if normalized:
            return normalized

    pack_build = require_json_object(lock_data.get("pack_build", {}), "stack.lock.json pack_build")
    final_pack_path = str(pack_build.get("final_pack_path") or "").strip()
    return [final_pack_path or "packs/NOT-EDIT-final-merged-packs"]


def runtime_final_pack_entry(lock_data: dict[str, object]) -> str:
    pack_build = require_json_object(lock_data.get("pack_build", {}), "stack.lock.json pack_build")
    return str(pack_build.get("final_pack_path") or "packs/NOT-EDIT-final-merged-packs").strip()


def final_pack_lineage_path(downstream_root: Path) -> Path:
    stack_lock = downstream_root / "stack.lock.json"
    if not stack_lock.is_file():
        raise DeployError(f"stack.lock.json not found in downstream repo: {downstream_root}")
    lock_data = require_json_object(json.loads(stack_lock.read_text()), str(stack_lock))
    final_pack_path = runtime_final_pack_entry(lock_data)
    if not final_pack_path:
        raise DeployError(f"stack.lock.json pack_build.final_pack_path is empty: {stack_lock}")
    return downstream_root / final_pack_path / ".build-lineage.json"


def pack_repo_commit_runtime_refs(downstream_root: Path) -> dict[str, str]:
    lineage_path = final_pack_lineage_path(downstream_root)
    if not lineage_path.is_file():
        raise DeployError(f"final pack build lineage not found: {lineage_path}")
    lineage = require_json_object(json.loads(lineage_path.read_text()), str(lineage_path))
    upstream = require_json_object(lineage.get("upstreamRuntime", {}), f"{lineage_path} upstreamRuntime")
    official_pack = require_json_object(lineage.get("officialPack", {}), f"{lineage_path} officialPack")
    upstream_ref = str(upstream.get("ref", "")).strip()
    official_app_packs_ref = str(official_pack.get("ref", "")).strip()
    missing = []
    if not upstream_ref:
        missing.append("upstreamRuntime.ref")
    if not official_app_packs_ref:
        missing.append("officialPack.ref")
    if missing:
        raise DeployError(f"final pack build lineage missing required refs: {', '.join(missing)}")
    return {
        "UPSTREAM_REF": upstream_ref,
        "OFFICIAL_APP_PACKS_REF": official_app_packs_ref,
        "LINEAGE_PATH": str(lineage_path),
    }


def resolve_pack_repo_commit_runtime_refs(
    downstream_root: Path,
    upstream_root: Path,
    app_packs_root: Path,
) -> dict[str, str]:
    refs = pack_repo_commit_runtime_refs(downstream_root)
    resolved_upstream_ref = resolve_repo_ref(upstream_root, refs["UPSTREAM_REF"])
    resolved_official_app_packs_ref = resolve_repo_ref(app_packs_root, refs["OFFICIAL_APP_PACKS_REF"])
    return {
        "UPSTREAM_REF": resolved_upstream_ref,
        "OFFICIAL_APP_PACKS_REF": resolved_official_app_packs_ref,
        "RAW_UPSTREAM_REF": refs["UPSTREAM_REF"],
        "RAW_OFFICIAL_APP_PACKS_REF": refs["OFFICIAL_APP_PACKS_REF"],
        "LINEAGE_PATH": refs["LINEAGE_PATH"],
    }


def require_resolved_ref_matches_authority(
    *,
    label: str,
    requested_ref: str,
    requested_resolved_ref: str,
    authority_ref: str,
    authority_source: str,
) -> None:
    if requested_resolved_ref == authority_ref:
        return
    raise DeployError(
        f"{label} ref is not allowed for staging deploys: "
        f"{requested_ref} resolved to {requested_resolved_ref}, "
        f"but {authority_source} requires {authority_ref}"
    )


def require_downstream_ref_resolves_to_main(target_dir: Path, requested_ref: str) -> str:
    raw_requested_ref = str(requested_ref or "").strip() or "main"
    resolved_ref = resolve_repo_ref(target_dir, raw_requested_ref)
    main_ref = resolve_repo_ref(target_dir, "main")
    if resolved_ref != main_ref:
        raise DeployError(
            "staging deploys are only allowed from downstream main: "
            f"{raw_requested_ref} resolved to {resolved_ref}, origin/main is {main_ref}"
        )
    return resolved_ref


def require_immutable_downstream_commit(target_dir: Path, requested_ref: str) -> str:
    raw_requested_ref = str(requested_ref or "").strip()
    if not GIT_COMMIT_RE.fullmatch(raw_requested_ref):
        raise DeployError(
            "staging downstream source must be a frozen lowercase commit SHA: "
            f"{raw_requested_ref!r}"
        )
    resolved_ref = resolve_repo_ref(target_dir, raw_requested_ref)
    if resolved_ref != raw_requested_ref:
        raise DeployError(
            "staging downstream source did not resolve to its frozen commit: "
            f"{raw_requested_ref} resolved to {resolved_ref}"
        )
    return resolved_ref


def runtime_deploy_local_pack_paths(lock_data: dict[str, object]) -> list[str]:
    return [
        entry
        for entry in runtime_deploy_pack_entries(lock_data)
        if not entry.startswith("official_repo:")
    ]


def runtime_deploy_pack_paths(lock_data: dict[str, object]) -> list[str]:
    return runtime_deploy_local_pack_paths(lock_data)


def runtime_pack_entry_name_and_source(downstream_build_root: Path, pack_entry: str) -> tuple[str, str, Path | None]:
    if pack_entry.startswith("official_repo:"):
        pack_name = pack_entry.removeprefix("official_repo:").strip()
        if not pack_name:
            raise DeployError(f"invalid official pack entry in stack.lock.json: {pack_entry!r}")
        return pack_name, pack_entry, None

    pack_dir = downstream_build_root / pack_entry
    pack_json = pack_dir / "pack.json"
    if not pack_json.is_file():
        raise DeployError(f"runtime pack source is missing pack.json: {pack_json}")
    pack_name = str(json.loads(pack_json.read_text()).get("name") or "").strip()
    if not pack_name:
        raise DeployError(f"runtime pack source is missing pack name: {pack_json}")
    return pack_name, f"local_path:{pack_dir}", pack_dir


def runtime_pack_entry_release_source(
    downstream_build_root: Path,
    pack_entry: str,
    downstream_revision: str,
    *,
    downstream_source_root: Path | None = None,
) -> tuple[str, dict[str, str]]:
    pack_name, _local_source, pack_dir = runtime_pack_entry_name_and_source(
        downstream_build_root,
        pack_entry,
    )
    if pack_dir is None:
        raise DeployError(f"official pack {pack_name} does not use release_repo authority")
    revision = str(downstream_revision).strip()
    if not GIT_COMMIT_RE.fullmatch(revision):
        raise DeployError("release_repo source authority requires an exact downstream commit")
    checkout_root = downstream_build_root.resolve()
    resolved_pack_dir = pack_dir.resolve()
    if not resolved_pack_dir.is_relative_to(checkout_root):
        raise DeployError(f"release pack escapes the exact downstream checkout: {pack_entry}")
    source_root = (
        downstream_source_root.expanduser().resolve()
        if downstream_source_root is not None
        else checkout_root
    )
    if exact_git_source_revision(source_root) != revision:
        raise DeployError(
            "release_repo source authority requires an exact downstream Git checkout"
        )
    source_pack_dir = (source_root / pack_entry).resolve()
    if not source_pack_dir.is_relative_to(source_root):
        raise DeployError(f"release pack escapes the exact downstream source: {pack_entry}")
    try:
        snapshot_hash = deployment_bundle.tree_sha256(resolved_pack_dir)
        source_hash = deployment_bundle.tree_sha256(source_pack_dir)
    except deployment_bundle.BundleError as exc:
        raise DeployError(str(exc)) from exc
    if snapshot_hash != source_hash:
        raise DeployError(
            "sealed downstream release pack does not match its exact Git source: "
            + pack_entry
        )
    uri = (
        "urn:nexuscrm:release-pack:pingcap-inc:nexuscrm-pingcap-pack:"
        + quote(pack_entry.strip(), safe="")
    )
    return f"release_repo:{uri}", {
        "uri": uri,
        "pack_root": str(source_pack_dir),
        "revision": revision,
    }


def runtime_pack_entry_source_dir(
    downstream_build_root: Path,
    app_packs_root: Path,
    pack_entry: str,
) -> tuple[str, Path]:
    pack_name, _, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
    return pack_name, pack_dir or (app_packs_root / pack_name)


def require_env(values: dict[str, str], keys: list[str]) -> None:
    missing = [key for key in keys if not values.get(key)]
    if missing:
        raise DeployError(f"missing required env keys: {', '.join(missing)}")


def bool_env(values: dict[str, str], key: str, default: bool = False) -> bool:
    raw = values.get(key)
    if raw is None or raw == "":
        return default
    return raw.lower() in {"1", "true", "yes", "on"}


def build_run_label() -> str:
    run_id = os.environ.get("GITHUB_RUN_ID", "").strip()
    if run_id:
        attempt = os.environ.get("GITHUB_RUN_ATTEMPT", "").strip()
        return f"run-{run_id}-{attempt}" if attempt else f"run-{run_id}"
    return "local-" + time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())


def resolve_backend_log_dir(envv: dict[str, str], env_root: Path) -> Path:
    configured = (envv.get("BACKEND_LOG_DIR") or "").strip()
    if configured:
        return Path(configured)
    return env_root / "logs"


def safe_tcp_port(value: str, label: str) -> int:
    port = str(value).strip()
    if not port.isdigit():
        raise DeployError(f"unsafe {label} value: {port}")
    parsed = int(port)
    if parsed < 1 or parsed > 65535:
        raise DeployError(f"unsafe {label} value: {port}")
    return parsed


def exp_hubspot_options_adapter_port(envv: dict[str, str]) -> str:
    configured = str(envv.get("EXP_HUBSPOT_OPTIONS_ADAPTER_PORT", "")).strip()
    if configured:
        return str(safe_tcp_port(configured, "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT"))
    backend_port = safe_tcp_port(str(envv.get("BACKEND_PORT", "")), "BACKEND_PORT")
    derived = backend_port + 101
    if derived > 65535:
        raise DeployError(f"derived EXP_HUBSPOT_OPTIONS_ADAPTER_PORT is out of range: {derived}")
    return str(derived)


def first_nonempty(*values: str | None) -> str:
    for value in values:
        if value is not None and value != "":
            return value
    return ""


def runtime_frontend_url(envv: dict[str, str]) -> str:
    return first_nonempty(
        envv.get("FRONTEND_URL"),
        envv.get("REACT_APP_FRONTEND_URL"),
        envv.get("PUBLIC_FRONTEND_URL"),
        envv.get("PUBLIC_BASE_URL"),
    )


def runtime_rate_limit_env(envv: dict[str, str]) -> dict[str, str]:
    hourly = first_nonempty(
        envv.get("NEXUS_API_HOURLY_LIMIT"),
        envv.get("CONNECTED_APP_HOURLY_API_LIMIT"),
    )
    daily = first_nonempty(
        envv.get("NEXUS_API_DAILY_LIMIT"),
        envv.get("CONNECTED_APP_DAILY_API_LIMIT"),
    )
    forwarded: dict[str, str] = {}
    if hourly:
        # Export both names so old and new upstream refs see the same override.
        forwarded["NEXUS_API_HOURLY_LIMIT"] = hourly
        forwarded["CONNECTED_APP_HOURLY_API_LIMIT"] = hourly
    if daily:
        forwarded["NEXUS_API_DAILY_LIMIT"] = daily
        forwarded["CONNECTED_APP_DAILY_API_LIMIT"] = daily
    return forwarded


def is_transient_tidb_branch_error(message: str) -> bool:
    lowered = message.lower()
    return (
        "your branch is in the process of restoring" in lowered
        or "access denied for user" in lowered
        or "failed to ping database" in lowered
        or "can't connect" in lowered
        or "could not connect" in lowered
        or "connection refused" in lowered
    )


def optional_process_env(envv: dict[str, str]) -> dict[str, str]:
    forwarded: dict[str, str] = {}
    for key in [
        "TMPDIR",
        "GOCACHE",
        "GOMODCACHE",
        "GOTMPDIR",
        "GOFLAGS",
        "CGO_ENABLED",
        "CC",
        "CXX",
    ]:
        value = envv.get(key)
        if value:
            forwarded[key] = value
    return forwarded


def _append_go_private_pattern(raw_value: str | None, pattern: str) -> str:
    values = [value.strip() for value in str(raw_value or "").split(",") if value.strip()]
    if pattern not in values:
        values.append(pattern)
    return ",".join(values)


def private_go_module_token(envv: dict[str, str]) -> str:
    return str(envv.get("GH_TOKEN") or os.environ.get("GH_TOKEN") or "").strip()


def private_go_module_env(envv: dict[str, str]) -> dict[str, str]:
    """Allow Go to fetch private PingCAP modules from non-interactive deploy runners."""
    pattern = "github.com/pingcap-inc/*"
    forwarded = {
        "GOPRIVATE": _append_go_private_pattern(envv.get("GOPRIVATE") or os.environ.get("GOPRIVATE"), pattern),
        "GONOSUMDB": _append_go_private_pattern(envv.get("GONOSUMDB") or os.environ.get("GONOSUMDB"), pattern),
    }
    token = private_go_module_token(envv)
    if token:
        forwarded.update(
            {
                "GIT_CONFIG_COUNT": "1",
                "GIT_CONFIG_KEY_0": f"url.https://x-access-token:{token}@github.com/pingcap-inc/.insteadOf",
                "GIT_CONFIG_VALUE_0": "https://github.com/pingcap-inc/",
            }
        )
    return forwarded


def resolve_go_module_origin_identity(
    *,
    module_root: Path,
    module_path: str,
    expected_origin_url: str,
    envv: dict[str, str],
) -> dict[str, str]:
    """Resolve an immutable module commit from the selected version in go.mod."""
    go_mod = module_root / "go.mod"
    if not go_mod.is_file():
        raise DeployError(f"Go module root is missing go.mod: {go_mod}")

    module_env = private_go_module_env(envv)
    module_env["GOWORK"] = "off"
    with tempfile.TemporaryDirectory(prefix="nexuscrm-go-module-origin-") as cache_root:
        module_env["GOMODCACHE"] = str(Path(cache_root) / "mod")
        try:
            selected = json.loads(
                run(
                    ["go", "list", "-m", "-json", module_path],
                    cwd=module_root,
                    env=module_env,
                    capture=True,
                )
            )
        except json.JSONDecodeError as exc:
            raise DeployError(f"selected Go module identity is not valid JSON: {module_path}: {exc}") from exc
        if str(selected.get("Path", "")).strip() != module_path:
            raise DeployError(
                f"selected Go module path mismatch: expected {module_path}, got {selected.get('Path')}"
            )
        if selected.get("Replace") is not None:
            raise DeployError(f"selected Go module must not use a replace directive: {module_path}")
        version = str(selected.get("Version", "")).strip()
        if not version:
            raise DeployError(f"selected Go module has no immutable version: {module_path}")

        try:
            downloaded = json.loads(
                run(
                    ["go", "mod", "download", "-json", f"{module_path}@{version}"],
                    cwd=module_root,
                    env=module_env,
                    capture=True,
                )
            )
        except json.JSONDecodeError as exc:
            raise DeployError(
                f"downloaded Go module identity is not valid JSON: {module_path}@{version}: {exc}"
            ) from exc
    if downloaded.get("Error"):
        raise DeployError(f"failed to resolve Go module origin: {module_path}@{version}")
    if str(downloaded.get("Path", "")).strip() != module_path:
        raise DeployError(
            f"downloaded Go module path mismatch: expected {module_path}, got {downloaded.get('Path')}"
        )
    if str(downloaded.get("Version", "")).strip() != version:
        raise DeployError(
            f"downloaded Go module version mismatch: expected {version}, got {downloaded.get('Version')}"
        )

    origin = downloaded.get("Origin")
    if not isinstance(origin, dict):
        raise DeployError(f"downloaded Go module has no VCS origin: {module_path}@{version}")
    if str(origin.get("VCS", "")).strip() != "git":
        raise DeployError(f"downloaded Go module origin is not git: {module_path}@{version}")
    origin_url = str(origin.get("URL", "")).strip().rstrip("/")
    normalized_origin_url = origin_url.removesuffix(".git")
    normalized_expected_url = expected_origin_url.strip().rstrip("/").removesuffix(".git")
    if normalized_origin_url != normalized_expected_url:
        raise DeployError(
            "downloaded Go module origin URL mismatch: "
            f"expected {normalized_expected_url}, got {normalized_origin_url or '<missing>'}"
        )
    origin_hash = str(origin.get("Hash", "")).strip()
    if not GIT_COMMIT_RE.fullmatch(origin_hash):
        raise DeployError(f"downloaded Go module origin hash is not an immutable commit: {module_path}@{version}")
    origin_ref = str(origin.get("Ref", "")).strip()

    return {
        "module_path": module_path,
        "module_version": version,
        "origin_url": normalized_origin_url,
        "origin_ref": origin_ref,
        "origin_hash": origin_hash,
        "module_root": str(module_root.resolve()),
    }


def runner_toolchain_passthrough_env() -> dict[str, str]:
    """Runner-level process variables a subprocess legitimately needs but that are
    NOT deploy parameters: PATH/HOME to locate and run the Go/git toolchain. Deploy
    parameters and secrets must come from runtime.env, never from os.environ, so the
    rest of the ambient process environment is intentionally dropped."""
    passthrough: dict[str, str] = {}
    for key in ("PATH", "HOME"):
        value = os.environ.get(key, "")
        if value:
            passthrough[key] = value
    return passthrough


def runtime_mutable_env_defaults(env_root: Path) -> dict[str, str]:
    shared_root = env_root.expanduser().resolve() / "shared"
    return {
        "FILE_STORAGE_ROOT": str(shared_root / "data" / "object-storage"),
        "FILE_STORAGE_LEGACY_UPLOAD_ROOT": str(shared_root / "data" / "uploads"),
        "TMPDIR": str(shared_root / "tmp"),
        "GOCACHE": str(shared_root / "cache" / "go-build"),
        "GOMODCACHE": str(shared_root / "cache" / "gomod"),
        "GOTMPDIR": str(shared_root / "tmp" / "go-build"),
    }


def normalize_runtime_mutable_env(envv: dict[str, str], env_root: Path) -> dict[str, str]:
    """Keep process-writable runtime paths outside source/release worktrees."""
    defaults = runtime_mutable_env_defaults(env_root)
    legacy_storage_defaults = {
        "FILE_STORAGE_ROOT": LEGACY_FILE_STORAGE_ROOT,
        "FILE_STORAGE_LEGACY_UPLOAD_ROOT": LEGACY_FILE_STORAGE_LEGACY_UPLOAD_ROOT,
    }
    normalized = dict(envv)

    for key, default_value in defaults.items():
        raw_value = str(normalized.get(key, "")).strip()
        if not raw_value or raw_value == legacy_storage_defaults.get(key):
            normalized[key] = default_value
            continue
        if not Path(raw_value).expanduser().is_absolute():
            raise DeployError(
                f"{key} must be an absolute path outside source worktrees; "
                f"got {raw_value!r}"
            )
        normalized[key] = str(Path(raw_value).expanduser())

    return normalized


def ensure_runtime_mutable_dirs(envv: dict[str, str]) -> None:
    for key in [
        "FILE_STORAGE_ROOT",
        "FILE_STORAGE_LEGACY_UPLOAD_ROOT",
        "TMPDIR",
        "GOCACHE",
        "GOMODCACHE",
        "GOTMPDIR",
    ]:
        value = str(envv.get(key, "")).strip()
        if value:
            Path(value).expanduser().mkdir(parents=True, exist_ok=True)


def repo_url_with_token(url: str, token: str) -> str:
    if url.startswith("https://github.com/"):
        return f"https://x-access-token:{token}@{url[len('https://'):]}"
    return url


def clone_or_update_repo(repo_url: str, target_dir: Path, token: str) -> None:
    remote_url = repo_url_with_token(repo_url, token)
    if not (target_dir / ".git").is_dir():
        target_dir.parent.mkdir(parents=True, exist_ok=True)
        run(["git", "clone", remote_url, str(target_dir)])
    run(["git", "-C", str(target_dir), "remote", "set-url", "origin", remote_url])
    run(["git", "-C", str(target_dir), "fetch", "--all", "--tags", "--prune"])
    run(["git", "-C", str(target_dir), "reset", "--hard", "HEAD"])
    run(["git", "-C", str(target_dir), "clean", "-fd"])


def load_release_tag_payload(repo_root: Path, tag_name: str) -> dict[str, object]:
    ref = f"refs/tags/{tag_name}"
    probe = subprocess.run(
        ["git", "-C", str(repo_root), "show-ref", "--verify", "--quiet", ref],
        text=True,
        capture_output=True,
        check=False,
    )
    if probe.returncode != 0:
        raise DeployError(f"release tag not found: {tag_name}")

    message = run(
        ["git", "-C", str(repo_root), "for-each-ref", "--format=%(contents)", ref],
        capture=True,
    )
    payload_text = message.strip()
    if not payload_text:
        raise DeployError(f"release tag has empty message: {tag_name}")
    if RELEASE_TAG_JSON_MARKER in payload_text:
        payload_text = payload_text.split(RELEASE_TAG_JSON_MARKER, 1)[1].strip()
    try:
        payload = json.loads(payload_text)
    except json.JSONDecodeError as exc:
        raise DeployError(f"invalid release tag json payload: {tag_name}: {exc}") from exc
    release_tag = require_json_object(payload, f"release tag {tag_name}")
    release_tag.setdefault("tag_name", tag_name)
    return release_tag


def normalize_release_tag_patch_dir(patch_dir: str, repo_root: Path) -> str:
    raw = patch_dir.strip()
    if not raw:
        return ""
    direct = Path(raw).expanduser()
    if not direct.is_absolute():
        candidate = (repo_root / direct).resolve()
        if candidate.is_dir():
            return str(candidate)
        raise DeployError(f"release tag patch dir not found under repo root: {raw}")
    if direct.is_dir():
        return str(direct.resolve())
    for anchor in ("docs", "deployments", "scripts"):
        parts = direct.parts
        if anchor not in parts:
            continue
        candidate = (repo_root / Path(*parts[parts.index(anchor) :])).resolve()
        if candidate.is_dir():
            return str(candidate)
    raise DeployError(f"release tag patch dir not found: {raw}")


def resolve_patch_dir(patch_dir: str, *, repo_root: Path) -> Path:
    raw = patch_dir.strip()
    if not raw:
        raise DeployError("--patch-dir is required")
    path = Path(raw).expanduser()
    if not path.is_absolute():
        path = (repo_root / path).resolve()
    else:
        path = path.resolve()
    if not path.is_dir():
        raise DeployError(f"patch dir not found: {path}")
    return path


def load_patch_bundle_definition(patch_dir: Path) -> dict[str, object]:
    manifest_path: Path | None = None
    for candidate in PATCH_BUNDLE_MANIFEST_NAMES:
        candidate_path = patch_dir / candidate
        if candidate_path.is_file():
            manifest_path = candidate_path
            break
    if manifest_path is None:
        expected = ", ".join(PATCH_BUNDLE_MANIFEST_NAMES)
        raise DeployError(f"patch bundle manifest not found in {patch_dir}; expected one of: {expected}")

    try:
        payload = json.loads(manifest_path.read_text())
    except json.JSONDecodeError as exc:
        raise DeployError(f"invalid patch bundle manifest json: {manifest_path}: {exc}") from exc
    manifest = require_json_object(payload, f"patch bundle manifest {manifest_path}")

    created_at = str(manifest.get("created_at", "")).strip()
    if not created_at:
        raise DeployError(f"patch bundle manifest missing created_at: {manifest_path}")

    title = str(manifest.get("title", "") or manifest.get("description", "")).strip()
    raw_allowed_environments = manifest.get("allowed_environments")
    allowed_environments: list[str] | None = None
    if raw_allowed_environments is not None:
        if (
            not isinstance(raw_allowed_environments, list)
            or not raw_allowed_environments
            or any(
                not isinstance(item, str)
                or item not in PATCH_BUNDLE_ENVIRONMENTS
                for item in raw_allowed_environments
            )
            or len(set(raw_allowed_environments)) != len(raw_allowed_environments)
        ):
            raise DeployError(
                "patch bundle manifest allowed_environments must be a non-empty "
                f"unique array of known deployment environments: {manifest_path}"
            )
        allowed_environments = list(raw_allowed_environments)
    raw_sql_files = manifest.get("sql_files")
    if raw_sql_files is None:
        raw_sql_files = manifest.get("sqlFiles")
    sql_files: list[str] | None = None
    if raw_sql_files is not None:
        if not isinstance(raw_sql_files, list) or not raw_sql_files:
            raise DeployError(f"patch bundle manifest sql_files must be a non-empty array: {manifest_path}")
        normalized_sql_files: list[str] = []
        for item in raw_sql_files:
            name = str(item).strip()
            if not name:
                raise DeployError(f"patch bundle manifest contains empty sql_files entry: {manifest_path}")
            normalized_sql_files.append(name)
        sql_files = normalized_sql_files

    refs_payload = manifest.get("refs")
    refs: dict[str, str] | None = None
    if refs_payload is not None:
        refs_object = require_json_object(refs_payload, f"patch bundle manifest refs {manifest_path}")
        refs = {
            "downstream_ref": str(refs_object.get("downstream_ref", "")).strip(),
            "upstream_ref": str(refs_object.get("upstream_ref", "")).strip(),
            "official_app_packs_ref": str(refs_object.get("official_app_packs_ref", "")).strip(),
        }

    return {
        "manifest_path": str(manifest_path),
        "title": title,
        "created_at": created_at,
        "allowed_environments": allowed_environments,
        "sql_files": sql_files,
        "refs": refs,
    }


def canonical_patch_bundle_environment(value: str) -> str:
    """Map runtime slot names to the reviewed Patch Bundle environment axis."""
    environment = str(value or "").strip().lower()
    if environment in PATCH_BUNDLE_ENVIRONMENTS:
        return environment
    if re.fullmatch(r"staging-(?:blue|green)", environment):
        return "staging"
    test_match = re.fullmatch(r"test-(test[123])", environment)
    if test_match is not None:
        return test_match.group(1)
    if re.fullmatch(r"multi-prod-d3-slot-[0-9]+", environment):
        return "prod"
    raise DeployError(
        f"cannot derive a Patch Bundle environment from ENV_NAME={value!r}"
    )


def require_patch_bundle_environment(
    definition: dict[str, object],
    *,
    environment: str | None,
    patch_dir: Path,
) -> None:
    """Reject a scoped Patch Bundle outside its canonical target environment."""
    allowed = definition.get("allowed_environments")
    if allowed is None:
        return
    if not isinstance(allowed, list) or not allowed:
        raise DeployError(
            f"patch bundle has a malformed environment scope: {patch_dir}"
        )
    if environment is None:
        raise DeployError(
            f"environment-scoped Patch Bundle has no target environment: {patch_dir}"
        )
    target = canonical_patch_bundle_environment(environment)
    if target not in allowed:
        raise DeployError(
            f"Patch Bundle is not reviewed for {target}: {patch_dir}"
        )


def resolve_patch_sql_files(patch_dir: Path, definition: dict[str, object]) -> list[Path]:
    sql_files_by_name = {
        path.name: path
        for path in patch_dir.iterdir()
        if path.is_file() and path.suffix.lower() == ".sql"
    }
    manifest_sql_files = definition.get("sql_files")
    if manifest_sql_files:
        assert isinstance(manifest_sql_files, list)
        missing_sql_files = [name for name in manifest_sql_files if name not in sql_files_by_name]
        if missing_sql_files:
            raise DeployError(
                f"patch bundle manifest references missing sql files in {patch_dir}: {', '.join(missing_sql_files)}"
            )
        return [sql_files_by_name[name] for name in manifest_sql_files]

    sql_files = [sql_files_by_name[name] for name in sorted(sql_files_by_name)]
    if not sql_files:
        raise DeployError(f"no .sql files found in patch dir: {patch_dir}")
    return sql_files


def resolve_release_tag_runtime_config(
    *,
    repo_url: str,
    repo_root: Path,
    gh_token: str,
    tag_name: str,
    expected_kind: str,
    expected_environment: str | None = None,
) -> dict[str, object]:
    if gh_token:
        clone_or_update_repo(repo_url, repo_root, gh_token)
    elif not (repo_root / ".git").exists():
        raise DeployError("missing required env GH_TOKEN for release tag resolution")
    else:
        run(["git", "-C", str(repo_root), "fetch", "--tags", "--force", "origin"])
    payload = load_release_tag_payload(repo_root, tag_name)
    kind = str(payload.get("kind", "")).strip()
    if kind != expected_kind:
        raise DeployError(f"unexpected release tag kind for {tag_name}: {kind!r}")
    environment = str(payload.get("environment", "")).strip()
    if expected_environment and environment != expected_environment:
        raise DeployError(
            f"release tag environment mismatch for {tag_name}: expected {expected_environment!r}, got {environment!r}"
        )
    patch = require_json_object(payload.get("patch") or {}, f"release tag patch {tag_name}")
    return {
        "release_tag": str(payload.get("tag_name", tag_name)).strip() or tag_name,
        "environment": environment,
        "downstream_ref": str(payload.get("downstream_ref", "")).strip(),
        "upstream_ref": str(payload.get("upstream_ref", "")).strip(),
        "official_app_packs_ref": str(payload.get("official_app_packs_ref", "")).strip(),
        "patch_dir": normalize_release_tag_patch_dir(str(patch.get("patch_dir", "")).strip(), repo_root),
        "skip_bootstrap": bool(payload.get("skip_bootstrap", False)),
        "reset_branch": bool(payload.get("reset_branch", False)),
        "source_run_id": str(payload.get("source_run_id", "")).strip(),
    }


def cleanup_legacy_runtime_layout(env_root: Path) -> None:
    legacy_names = [
        "NexusCRM",
        "nexuscrm-pingcap-pack",
        "nexuscrm-app-packs",
    ]
    for name in legacy_names:
        legacy_path = env_root / name
        if legacy_path.exists():
            log(f"removing legacy runtime path: {legacy_path}")
            if legacy_path.is_dir() and not legacy_path.is_symlink():
                shutil.rmtree(legacy_path)
            else:
                legacy_path.unlink()

    legacy_runner = env_root / "run_backend.sh"
    if legacy_runner.exists():
        existing = legacy_runner.read_text(errors="ignore")
        if "legacy/lib/stack.sh" in existing or "stack_install_locked_packs" in existing:
            log(f"removing legacy backend runner: {legacy_runner}")
            legacy_runner.unlink()


def write_text_atomic(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = path.parent / f".{path.name}.tmp"
    tmp_path.write_text(content)
    os.replace(tmp_path, path)


def remove_path(path: Path) -> None:
    if not path.exists() and not path.is_symlink():
        return
    if path.is_symlink() or path.is_file():
        path.unlink()
        return
    shutil.rmtree(path)


def normalized_backend_go_build_tags(raw_tags: str | None) -> str:
    return " ".join(tag for tag in str(raw_tags or "").replace(",", " ").split() if tag)


def supported_backend_go_build_tags(raw_tags: str | None) -> str:
    normalized = normalized_backend_go_build_tags(raw_tags)
    if normalized not in ("", "crm"):
        raise DeployError(f"BACKEND_GO_BUILD_TAGS supports only the crm distribution: {normalized!r}")
    return normalized


def ensure_distribution_app_packs_link(nexuscrm_worktree: Path, app_packs_root: Path) -> None:
    source = app_packs_root.resolve()
    if not source.is_dir():
        raise DeployError(f"official app-packs root not found for distribution build: {source}")
    target = nexuscrm_worktree / "nexuscrm-app-packs"
    if target.exists() or target.is_symlink():
        try:
            if target.resolve() == source:
                return
        except OSError:
            pass
        if target.is_symlink() or target.is_file():
            target.unlink()
        else:
            raise DeployError(
                "distribution build requires nexuscrm-app-packs next to the NexusCRM worktree, "
                f"but {target} already exists and points elsewhere"
            )
    target.symlink_to(source, target_is_directory=True)




def backend_distribution_root(nexuscrm_worktree: Path) -> Path | None:
    root = nexuscrm_worktree / "backend" / "distributions"
    if root.is_dir():
        return root
    return None


def backend_setup_command(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    args: list[str],
) -> tuple[Path, list[str], dict[str, str]]:
    build_tags = supported_backend_go_build_tags(envv.get("BACKEND_GO_BUILD_TAGS"))
    if not build_tags:
        return nexuscrm_worktree / "backend", ["go", "run", "./cmd/setup", *args], {}

    distribution_root = backend_distribution_root(nexuscrm_worktree)
    if distribution_root is None:
        return (
            nexuscrm_worktree / "backend",
            ["go", "run", "-tags", build_tags, "./cmd/setup", *args],
            {},
        )

    ensure_distribution_app_packs_link(nexuscrm_worktree, app_packs_root)
    return (
        distribution_root,
        ["go", "run", "-tags", build_tags, "./cmd/setup", *args],
        {"GOWORK": "off"},
    )


def backend_setup_binary_build_command(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    setup_bin: Path,
) -> tuple[Path, list[str], dict[str, str]]:
    build_tags = supported_backend_go_build_tags(envv.get("BACKEND_GO_BUILD_TAGS"))
    if not build_tags:
        return (
            nexuscrm_worktree / "backend",
            [
                "go",
                "build",
                "-mod=readonly",
                "-trimpath",
                "-buildvcs=true",
                "-o",
                str(setup_bin),
                "./cmd/setup",
            ],
            {},
        )

    distribution_root = backend_distribution_root(nexuscrm_worktree)
    if distribution_root is None:
        return (
            nexuscrm_worktree / "backend",
            [
                "go",
                "build",
                "-mod=readonly",
                "-trimpath",
                "-buildvcs=true",
                "-tags",
                build_tags,
                "-o",
                str(setup_bin),
                "./cmd/setup",
            ],
            {},
        )

    ensure_distribution_app_packs_link(nexuscrm_worktree, app_packs_root)
    return (
        distribution_root,
        [
            "go",
            "build",
            "-mod=readonly",
            "-trimpath",
            "-buildvcs=true",
            "-tags",
            build_tags,
            "-o",
            str(setup_bin),
            "./cmd/setup",
        ],
        {"GOWORK": "off"},
    )


def backend_setup_workdir(envv: dict[str, str], nexuscrm_worktree: Path, app_packs_root: Path) -> Path:
    if not supported_backend_go_build_tags(envv.get("BACKEND_GO_BUILD_TAGS")):
        return nexuscrm_worktree / "backend"
    distribution_root = backend_distribution_root(nexuscrm_worktree)
    if distribution_root is None:
        return nexuscrm_worktree / "backend"
    ensure_distribution_app_packs_link(nexuscrm_worktree, app_packs_root)
    return distribution_root


def split_sql_statements(sql_text: str) -> list[str]:
    statements: list[str] = []
    current: list[str] = []
    in_single = False
    in_double = False
    in_backtick = False
    in_line_comment = False
    in_block_comment = False
    index = 0
    while index < len(sql_text):
        ch = sql_text[index]
        nxt = sql_text[index + 1] if index + 1 < len(sql_text) else ""
        if in_line_comment:
            if ch == "\n":
                in_line_comment = False
                current.append(ch)
            index += 1
            continue
        if in_block_comment:
            if ch == "*" and nxt == "/":
                in_block_comment = False
                index += 2
                continue
            index += 1
            continue
        if not (in_single or in_double or in_backtick):
            if ch == "-" and nxt == "-":
                in_line_comment = True
                index += 2
                continue
            if ch == "/" and nxt == "*":
                in_block_comment = True
                index += 2
                continue
        if ch == "'" and not (in_double or in_backtick):
            in_single = not in_single
            current.append(ch)
            index += 1
            continue
        if ch == '"' and not (in_single or in_backtick):
            in_double = not in_double
            current.append(ch)
            index += 1
            continue
        if ch == "`" and not (in_single or in_double):
            in_backtick = not in_backtick
            current.append(ch)
            index += 1
            continue
        if ch == ";" and not (in_single or in_double or in_backtick):
            statement = "".join(current).strip()
            if statement:
                statements.append(statement)
            current = []
            index += 1
            continue
        current.append(ch)
        index += 1
    tail = "".join(current).strip()
    if tail:
        statements.append(tail)
    return statements


def validate_patch_sql_idempotency(sql_file: Path) -> None:
    statements = split_sql_statements(sql_file.read_text())
    errors: list[str] = []
    for idx, statement in enumerate(statements, start=1):
        normalized = re.sub(r"\s+", " ", statement.strip()).upper()
        if not normalized or normalized.startswith(("START TRANSACTION", "COMMIT", "ROLLBACK", "SET ")):
            continue
        if normalized.startswith("INSERT INTO "):
            if " ON DUPLICATE KEY UPDATE " not in normalized and not normalized.startswith("INSERT IGNORE INTO "):
                errors.append(f"statement {idx}: INSERT must use INSERT IGNORE or ON DUPLICATE KEY UPDATE")
        if normalized.startswith("REPLACE INTO "):
            errors.append(f"statement {idx}: REPLACE INTO is not allowed in idempotent patch sql")
        if normalized.startswith("TRUNCATE "):
            errors.append(f"statement {idx}: TRUNCATE is not allowed in idempotent patch sql")
        if normalized.startswith("CREATE TABLE ") and " IF NOT EXISTS " not in normalized:
            errors.append(f"statement {idx}: CREATE TABLE must use IF NOT EXISTS")
        if normalized.startswith("CREATE DATABASE ") and " IF NOT EXISTS " not in normalized:
            errors.append(f"statement {idx}: CREATE DATABASE must use IF NOT EXISTS")
        if normalized.startswith("DROP TABLE ") and " IF EXISTS " not in normalized:
            errors.append(f"statement {idx}: DROP TABLE must use IF EXISTS")
        if normalized.startswith("DROP DATABASE ") and " IF EXISTS " not in normalized:
            errors.append(f"statement {idx}: DROP DATABASE must use IF EXISTS")
        if normalized.startswith("CREATE INDEX ") and " IF NOT EXISTS " not in normalized:
            errors.append(f"statement {idx}: CREATE INDEX must use IF NOT EXISTS")
        if normalized.startswith("ALTER TABLE "):
            if " ADD COLUMN " in normalized and " IF NOT EXISTS " not in normalized:
                errors.append(f"statement {idx}: ALTER TABLE ... ADD COLUMN must use IF NOT EXISTS")
            if " DROP COLUMN " in normalized and " IF EXISTS " not in normalized:
                errors.append(f"statement {idx}: ALTER TABLE ... DROP COLUMN must use IF EXISTS")
            if re.search(r"\bADD\s+(UNIQUE\s+)?(INDEX|KEY)\b", normalized) and " IF NOT EXISTS " not in normalized:
                errors.append(f"statement {idx}: ALTER TABLE ... ADD INDEX/KEY must use IF NOT EXISTS")
            if re.search(r"\bDROP\s+(INDEX|KEY)\b", normalized) and " IF EXISTS " not in normalized:
                errors.append(f"statement {idx}: ALTER TABLE ... DROP INDEX/KEY must use IF EXISTS")
            if re.search(r"\b(MODIFY|CHANGE|RENAME COLUMN|DROP PRIMARY KEY|ADD PRIMARY KEY)\b", normalized):
                errors.append(f"statement {idx}: ALTER TABLE shape rewrite is not allowed in idempotent patch sql")
    if errors:
        raise DeployError(f"non-idempotent patch sql detected in {sql_file}: " + "; ".join(errors))


def resolve_repo_ref(target_dir: Path, requested_ref: str) -> str:
    if requested_ref.startswith("refs/"):
        run(["git", "-C", str(target_dir), "fetch", "origin", requested_ref])
        return run(
            ["git", "-C", str(target_dir), "rev-parse", "FETCH_HEAD^{commit}"],
            capture=True,
        ).strip()

    for candidate in [f"origin/{requested_ref}", requested_ref]:
        probe = subprocess.run(
            ["git", "-C", str(target_dir), "rev-parse", "--verify", f"{candidate}^{{commit}}"],
            text=True,
            capture_output=True,
            check=False,
        )
        if probe.returncode == 0:
            return probe.stdout.strip()

    fetch = subprocess.run(
        ["git", "-C", str(target_dir), "fetch", "--no-tags", "--depth=1", "origin", requested_ref],
        text=True,
        capture_output=True,
        check=False,
    )
    if fetch.returncode == 0:
        return run(
            ["git", "-C", str(target_dir), "rev-parse", "--verify", "FETCH_HEAD^{commit}"],
            capture=True,
        ).strip()
    raise DeployError(f"unable to resolve git ref '{requested_ref}' in {target_dir}")


def checkout_resolved_ref(target_dir: Path, resolved_ref: str) -> None:
    run(["git", "-C", str(target_dir), "checkout", "--detach", resolved_ref])
    run(["git", "-C", str(target_dir), "reset", "--hard", resolved_ref])


def sync_local_branch_with_origin(target_dir: Path, branch: str) -> None:
    if "/" in branch or branch.startswith("refs/"):
        return
    probe = subprocess.run(
        ["git", "-C", str(target_dir), "rev-parse", "--verify", f"origin/{branch}^{{commit}}"],
        text=True,
        capture_output=True,
        check=False,
    )
    if probe.returncode != 0:
        return
    subprocess.run(
        ["git", "-C", str(target_dir), "branch", "-f", branch, f"origin/{branch}"],
        text=True,
        capture_output=True,
        check=False,
    )


def use_tidb_tls(envv: dict[str, str]) -> bool:
    enabled = envv.get("TIDB_TLS")
    disabled = envv.get("TIDB_TLS_DISABLE")
    if disabled not in {None, ""}:
        tls = not bool_env(envv, "TIDB_TLS_DISABLE")
        if enabled not in {None, ""} and bool_env(envv, "TIDB_TLS") != tls:
            raise DeployError("TIDB_TLS and TIDB_TLS_DISABLE disagree")
        return tls
    if enabled in {None, ""}:
        return True
    return bool_env(envv, "TIDB_TLS")


def mysql_supports_ssl_mode(command: str) -> bool:
    cached = MYSQL_SSL_MODE_SUPPORT_CACHE.get(command)
    if cached is not None:
        return cached
    result = subprocess.run(
        [command, "--help"],
        text=True,
        capture_output=True,
        check=False,
    )
    supported = "--ssl-mode" in f"{result.stdout}\n{result.stderr}"
    MYSQL_SSL_MODE_SUPPORT_CACHE[command] = supported
    return supported


def mysql_tls_args(command: str, tls: bool) -> list[str]:
    if not tls:
        return []
    if mysql_supports_ssl_mode(command):
        return ["--ssl-mode", "REQUIRED"]
    return ["--ssl"]


def mysql_client(envv: dict[str, str]) -> list[str]:
    tls = use_tidb_tls(envv)
    if shutil.which("mariadb"):
        command = "mariadb"
        return [command, *mysql_tls_args(command, tls)]
    if shutil.which("mysql"):
        command = "mysql"
        return [command, *mysql_tls_args(command, tls)]
    raise DeployError("missing required command: mysql or mariadb")


def db_exec(envv: dict[str, str], sql: str, *, database: str | None = None) -> None:
    env = {}
    args = mysql_client(envv) + [
        f"-h{envv['TIDB_HOST']}",
        f"-P{envv['TIDB_PORT']}",
        f"-u{envv['TIDB_USER']}",
    ]
    if envv.get("TIDB_PASSWORD"):
        env["MYSQL_PWD"] = envv["TIDB_PASSWORD"]
    if database:
        args.append(database)
    run(args, env=env, stdin=sql + "\n")


def db_query_scalar(envv: dict[str, str], sql: str, *, database: str | None = None) -> str:
    env = {}
    args = mysql_client(envv) + [
        f"-h{envv['TIDB_HOST']}",
        f"-P{envv['TIDB_PORT']}",
        f"-u{envv['TIDB_USER']}",
        "-N",
        "-B",
        "-r",
    ]
    if envv.get("TIDB_PASSWORD"):
        env["MYSQL_PWD"] = envv["TIDB_PASSWORD"]
    if database:
        args.append(database)
    args += ["-e", sql]
    return run(args, env=env, capture=True).strip()


def ensure_tidb_database(envv: dict[str, str]) -> None:
    database = envv["TIDB_DATABASE"]
    if not database.replace("_", "").isalnum():
        raise DeployError(f"unsafe TIDB_DATABASE value: {database}")
    db_exec(envv, f"CREATE DATABASE IF NOT EXISTS `{database}`;")


def tidb_database_table_count(envv: dict[str, str]) -> int:
    database = envv["TIDB_DATABASE"].replace("'", "''")
    count = db_query_scalar(
        envv,
        f"SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='{database}';",
        database="information_schema",
    )
    try:
        return int(count)
    except ValueError as exc:
        raise DeployError(f"unexpected table count for database {envv['TIDB_DATABASE']}: {count}") from exc


def installed_pack_table_exists(envv: dict[str, str]) -> bool:
    return tidb_table_exists(envv, "_System_InstalledPack")


def tidb_table_exists(envv: dict[str, str], table_name: str) -> bool:
    database = envv["TIDB_DATABASE"].replace("'", "''")
    table = table_name.replace("'", "''")
    count = db_query_scalar(
        envv,
        f"SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='{database}' AND table_name='{table}';",
        database="information_schema",
    )
    return count == "1"


def installed_pack_column_exists(envv: dict[str, str], column_name: str) -> bool:
    database = envv["TIDB_DATABASE"].replace("'", "''")
    column = column_name.replace("'", "''")
    count = db_query_scalar(
        envv,
        (
            "SELECT COUNT(*) FROM information_schema.columns "
            f"WHERE table_schema='{database}' "
            "AND table_name='_System_InstalledPack' "
            f"AND column_name='{column}';"
        ),
        database="information_schema",
    )
    return count == "1"


def tidb_cloud_api(
    envv: dict[str, str],
    method: str,
    path: str,
    body: dict | None = None,
    *,
    allow_statuses: set[int] | None = None,
) -> dict:
    """
    TiDB Cloud API wrapper.

    Do not use `curl -f` here: staging reset may return HTTP 409 when the branch is
    temporarily busy. We must be able to inspect the status code and retry.
    """
    if allow_statuses is None:
        allow_statuses = {200, 201, 202, 204}
    url = f"{envv.get('TIDB_CLOUD_API_BASE_URL', 'https://serverless.tidbapi.com/v1beta1')}/clusters/{envv['TIDB_CLOUD_CLUSTER_ID']}{path}"
    def curl_config_value(value: str) -> str:
        if any(ord(character) < 32 and character not in "\t\r\n" for character in value):
            raise DeployError("TiDB Cloud API credential input contains a control character")
        escaped = (
            value.replace("\\", "\\\\")
            .replace('"', '\\"')
            .replace("\t", "\\t")
            .replace("\r", "\\r")
            .replace("\n", "\\n")
        )
        return f'"{escaped}"'

    # Keep the provider identity, branch URL, and request body out of argv.
    # On a shared self-hosted runner, argv is visible more broadly than stdin.
    curl_config = [
        f"url = {curl_config_value(url)}",
        "user = "
        + curl_config_value(
            f"{envv['TIDB_CLOUD_PUBLIC_KEY']}:{envv['TIDB_CLOUD_PRIVATE_KEY']}"
        ),
    ]
    if body is not None:
        curl_config.append(
            f"data = {curl_config_value(json.dumps(body, separators=(',', ':')))}"
        )
    curl_args = [
        "curl",
        "-sS",
        "--digest",
        "--config",
        "-",
        "-X",
        method,
        "-H",
        "Content-Type: application/json",
        "-w",
        "\n__HTTP_STATUS__:%{http_code}\n",
    ]

    # Don't use `run()` because we need stdout even when HTTP != 2xx.
    result = subprocess.run(
        curl_args,
        input="\n".join(curl_config) + "\n",
        text=True,
        capture_output=True,
        check=False,
    )
    safe_path = re.sub(r"/branches/[^/:]+", "/branches/<redacted>", path)
    output = (result.stdout or "").strip()
    if "__HTTP_STATUS__:" not in output:
        raise DeployError(
            f"TiDB Cloud API {method} {safe_path} transport failed "
            f"(curl exit {result.returncode})"
        )
    body_text, status_part = output.rsplit("__HTTP_STATUS__:", 1)
    status_str = status_part.strip().splitlines()[-1].strip()
    try:
        status = int(status_str)
    except ValueError:
        raise DeployError(f"unexpected TiDB Cloud API response for {method} {safe_path}")

    body_text = body_text.strip()
    payload: dict = {}
    if body_text:
        try:
            payload = json.loads(body_text)
        except json.JSONDecodeError:
            payload = {"_raw": body_text}
    payload["_http_status"] = status

    if status not in allow_statuses:
        raise DeployError(f"TiDB Cloud API {method} {safe_path} returned http {status}")
    return payload


def wait_for_tidb_branch_active(envv: dict[str, str], branch_id: str) -> dict:
    last_state = ""
    # Reset/restore on TiDB Cloud can take longer than 10 minutes depending on branch size/load.
    # We default to ~30 minutes to avoid flaky workflow failures.
    for _ in range(360):
        try:
            payload = tidb_cloud_api(envv, "GET", f"/branches/{branch_id}")
        except Exception:
            time.sleep(5)
            continue
        last_state = payload.get("state", "") or payload.get("data", {}).get("state", "")
        if last_state == "ACTIVE":
            return payload
        time.sleep(5)
    branch_sha256 = hashlib.sha256(branch_id.encode("utf-8")).hexdigest()
    raise DeployError(
        "branch sha256="
        f"{branch_sha256} did not become ACTIVE after reset (last state: {last_state})"
    )


def tidb_branch_parent_timestamp(branch: dict) -> str:
    value = branch.get("parentTimestamp")
    if value in (None, ""):
        data = branch.get("data")
        value = data.get("parentTimestamp") if isinstance(data, dict) else None
    parent_timestamp = str(value or "").strip()
    if not parent_timestamp:
        raise DeployError("tidb branch has no parentTimestamp reset epoch")
    return parent_timestamp


def reset_tidb_branch_if_requested(envv: dict[str, str]) -> dict[str, object] | None:
    if not bool_env(envv, "RESET_TIDB_BRANCH"):
        return None
    require_env(
        envv,
        [
            "TIDB_CLOUD_PUBLIC_KEY",
            "TIDB_CLOUD_PRIVATE_KEY",
            "TIDB_CLOUD_CLUSTER_ID",
            "TIDB_CLOUD_BRANCH_ID",
        ],
    )
    branch_id = envv["TIDB_CLOUD_BRANCH_ID"]
    branch_sha256 = hashlib.sha256(branch_id.encode("utf-8")).hexdigest()
    log(f"resetting TiDB Cloud branch sha256={branch_sha256}")
    before_reset = tidb_cloud_api(envv, "GET", f"/branches/{branch_id}")
    accepted_previous_parent_timestamp = tidb_branch_parent_timestamp(before_reset)
    # TiDB Cloud may return HTTP 409 when the branch is already resetting / busy.
    # Treat it as transient: wait for ACTIVE then retry reset once.
    payload = tidb_cloud_api(envv, "POST", f"/branches/{branch_id}:reset", allow_statuses={200, 202, 409})
    reset_status = payload.get("_http_status")
    if reset_status not in {200, 202, 409}:
        raise DeployError(f"tidb branch reset returned an invalid HTTP status: {reset_status!r}")
    completed_branch = wait_for_tidb_branch_active(envv, branch_id)
    if reset_status == 409:
        accepted_previous_parent_timestamp = tidb_branch_parent_timestamp(completed_branch)
        log("tidb branch reset returned 409 (branch busy); retrying after ACTIVE")
        retry_payload = tidb_cloud_api(
            envv,
            "POST",
            f"/branches/{branch_id}:reset",
            allow_statuses={200, 202},
        )
        retry_status = retry_payload.get("_http_status")
        if retry_status not in {200, 202}:
            raise DeployError(
                "tidb branch reset remained busy after the bounded retry "
                f"(HTTP {retry_status!r})"
            )
        completed_branch = wait_for_tidb_branch_active(envv, branch_id)
    completed_parent_timestamp = tidb_branch_parent_timestamp(completed_branch)
    if completed_parent_timestamp == accepted_previous_parent_timestamp:
        raise DeployError(
            "tidb branch reset did not advance parentTimestamp from the state "
            "immediately before the accepted reset"
        )
    final_branch = completed_branch
    if envv.get("TIDB_PASSWORD"):
        tidb_cloud_api(
            envv,
            "PATCH",
            f"/branches/{branch_id}/sqlUsers/root",
            {"password": envv["TIDB_PASSWORD"]},
        )
        after_password = wait_for_tidb_branch_active(envv, branch_id)
        if tidb_branch_parent_timestamp(after_password) != completed_parent_timestamp:
            raise DeployError("tidb branch reset epoch changed during password rotation")
        final_branch = after_password
    return {
        "branch": final_branch,
        "reset_http_status": reset_status,
        "reset_retry_http_status": retry_status if reset_status == 409 else None,
        "previous_parent_timestamp": accepted_previous_parent_timestamp,
        "parent_timestamp": completed_parent_timestamp,
    }


def reset_database_if_requested(envv: dict[str, str]) -> None:
    if not bool_env(envv, "RESET_DB"):
        return
    database = envv["TIDB_DATABASE"]
    db_exec(envv, f"DROP DATABASE IF EXISTS `{database}`; CREATE DATABASE `{database}`;")


def prune_git_worktrees(repo_root: Path) -> None:
    run(["git", "-C", str(repo_root), "worktree", "prune", "--expire", "now"])


def forget_missing_git_worktree(repo_root: Path, worktree: Path) -> None:
    if worktree.exists():
        return
    subprocess.run(
        ["git", "-C", str(repo_root), "worktree", "remove", "--force", str(worktree)],
        text=True,
        check=False,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    prune_git_worktrees(repo_root)


def prepare_downstream_build_root(
    envv: dict[str, str],
    downstream_root: Path,
    runtime_dir: Path,
) -> Path:
    downstream_ref = envv["DOWNSTREAM_REF"]
    safe_ref = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in downstream_ref)
    worktree = runtime_dir / f"downstream-{safe_ref}"
    runtime_dir.mkdir(parents=True, exist_ok=True)
    prune_git_worktrees(downstream_root)
    if worktree.is_dir():
        current = run(["git", "-C", str(worktree), "rev-parse", "HEAD"], capture=True).strip()
        if current != downstream_ref:
            shutil.rmtree(worktree)
    forget_missing_git_worktree(downstream_root, worktree)
    if not worktree.is_dir():
        run(["git", "-C", str(downstream_root), "worktree", "add", "--detach", str(worktree), downstream_ref])
    run(["git", "-C", str(worktree), "reset", "--hard", downstream_ref])
    run(["git", "-C", str(worktree), "clean", "-fd"])
    return worktree


def prepare_nexuscrm_runtime(
    envv: dict[str, str],
    downstream_build_root: Path,
    upstream_root: Path,
    runtime_dir: Path,
) -> Path:
    stack_lock = downstream_build_root / "stack.lock.json"
    lock_data = json.loads(stack_lock.read_text())
    app_shell_config = lock_data.get("app_shell_config")
    upstream_ref = envv["UPSTREAM_REF"]
    safe_ref = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in upstream_ref)
    worktree = runtime_dir / f"nexuscrm-{safe_ref}"
    runtime_dir.mkdir(parents=True, exist_ok=True)
    prune_git_worktrees(upstream_root)
    if worktree.is_dir():
        current = run(["git", "-C", str(worktree), "rev-parse", "HEAD"], capture=True).strip()
        if current != upstream_ref:
            shutil.rmtree(worktree)
    forget_missing_git_worktree(upstream_root, worktree)
    if not worktree.is_dir():
        run(["git", "-C", str(upstream_root), "worktree", "add", "--detach", str(worktree), upstream_ref])
    run(["git", "-C", str(worktree), "reset", "--hard", upstream_ref])
    run(["git", "-C", str(worktree), "clean", "-fd"])

    for rel_path in ["infra/backend.Dockerfile", "infra/entrypoint.sh"]:
        overlay_source = None
        if (downstream_build_root / rel_path).is_file():
            overlay_source = downstream_build_root / rel_path
        elif (upstream_root / rel_path).is_file():
            overlay_source = upstream_root / rel_path
        if overlay_source:
            target = worktree / rel_path
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(overlay_source, target)

    if app_shell_config:
        src_config = downstream_build_root / app_shell_config
        if src_config.is_file():
            target = worktree / "infra/stack-app-shell.json"
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(src_config, target)

    for pack_path in runtime_deploy_local_pack_paths(lock_data):
        src_dir = downstream_build_root / pack_path
        target_dir = worktree / pack_path
        if target_dir.exists():
            shutil.rmtree(target_dir)
        target_dir.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(src_dir, target_dir)

    return worktree


def configured_backend_setup_binary(envv: dict[str, str]) -> Path | None:
    raw_path = (envv.get("NEXUSCRM_SETUP_BIN") or os.environ.get("NEXUSCRM_SETUP_BIN", "")).strip()
    if not raw_path:
        return None
    setup_bin = Path(raw_path).expanduser()
    if not setup_bin.is_absolute():
        raise DeployError(f"NEXUSCRM_SETUP_BIN must be an absolute path: {raw_path}")
    return setup_bin


def hydrate_local_private_go_sums(nexuscrm_worktree: Path, source_go_sum: Path) -> int:
    source_paths = [source_go_sum]
    source_paths.extend(sorted(nexuscrm_worktree.glob("*/go.sum")))
    private_lines = sorted(
        {
            line
            for path in dict.fromkeys(path.resolve() for path in source_paths)
            if path.is_file()
            for line in path.read_text(encoding="utf-8").splitlines()
            if line.startswith("github.com/pingcap-inc/")
        }
    )
    if not private_lines:
        return 0

    target_paths = [source_go_sum]
    target_paths.extend(go_mod.with_name("go.sum") for go_mod in nexuscrm_worktree.glob("*/go.mod"))
    updated = 0
    for target_go_sum in dict.fromkeys(path.resolve() for path in target_paths):
        existing_text = target_go_sum.read_text(encoding="utf-8") if target_go_sum.is_file() else ""
        existing_lines = set(existing_text.splitlines())
        missing_lines = [line for line in private_lines if line not in existing_lines]
        if not missing_lines:
            continue
        prefix = existing_text
        if prefix and not prefix.endswith("\n"):
            prefix += "\n"
        target_go_sum.write_text(prefix + "\n".join(missing_lines) + "\n", encoding="utf-8")
        updated += 1
    return updated


def materialize_release_setup_source_clone(
    *,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    expected_revision: str,
    build_root: Path,
) -> Path:
    expected_revision = str(expected_revision).strip()
    if not GIT_COMMIT_RE.fullmatch(expected_revision):
        raise DeployError("release setup build requires an exact UPSTREAM_REF commit")

    source_root = build_root / "NexusCRM"
    git_env = {
        "PATH": DEFAULT_PATH_PREFIX,
        "HOME": str(build_root),
        "LANG": "C",
        "LC_ALL": "C",
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": "/dev/null",
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_ASKPASS": "/bin/false",
        "GIT_OPTIONAL_LOCKS": "0",
    }
    run(
        [
            "git",
            "clone",
            "--local",
            "--no-hardlinks",
            "--no-checkout",
            str(nexuscrm_worktree),
            str(source_root),
        ],
        env=git_env,
        inherit_env=False,
    )
    run(
        ["git", "-C", str(source_root), "checkout", "--detach", expected_revision],
        env=git_env,
        inherit_env=False,
    )

    source_shallow_raw = run(
        ["git", "-C", str(nexuscrm_worktree), "rev-parse", "--git-path", "shallow"],
        env=git_env,
        inherit_env=False,
        capture=True,
    ).strip()
    source_shallow = Path(source_shallow_raw)
    if not source_shallow.is_absolute():
        source_shallow = nexuscrm_worktree / source_shallow
    if source_shallow.is_file():
        clone_shallow_raw = run(
            ["git", "-C", str(source_root), "rev-parse", "--git-path", "shallow"],
            env=git_env,
            inherit_env=False,
            capture=True,
        ).strip()
        clone_shallow = Path(clone_shallow_raw)
        if not clone_shallow.is_absolute():
            clone_shallow = source_root / clone_shallow
        shutil.copyfile(source_shallow, clone_shallow)

    actual_revision = run(
        ["git", "-C", str(source_root), "rev-parse", "HEAD"],
        env=git_env,
        inherit_env=False,
        capture=True,
    ).strip()
    if actual_revision != expected_revision:
        raise DeployError("release setup source clone did not resolve the exact UPSTREAM_REF")

    exclude_path = source_root / ".git" / "info" / "exclude"
    existing_excludes = exclude_path.read_text(encoding="utf-8") if exclude_path.is_file() else ""
    app_packs_exclude = "/nexuscrm-app-packs"
    if app_packs_exclude not in existing_excludes.splitlines():
        prefix = existing_excludes
        if prefix and not prefix.endswith("\n"):
            prefix += "\n"
        exclude_path.parent.mkdir(parents=True, exist_ok=True)
        exclude_path.write_text(prefix + app_packs_exclude + "\n", encoding="utf-8")
    ensure_distribution_app_packs_link(source_root, app_packs_root)

    status = run(
        ["git", "-C", str(source_root), "status", "--porcelain", "--untracked-files=all"],
        env=git_env,
        inherit_env=False,
        capture=True,
    ).strip()
    if status:
        raise DeployError("release setup source clone is not clean after source binding")
    return source_root


def release_setup_build_env(envv: dict[str, str], build_root: Path) -> dict[str, str]:
    defaults = {
        "HOME": build_root / "home",
        "TMPDIR": build_root / "tmp",
        "GOTMPDIR": build_root / "go-tmp",
        "GOCACHE": build_root / "go-build-cache",
        "GOMODCACHE": build_root / "go-module-cache",
    }
    resolved_paths: dict[str, str] = {}
    for key, default_path in defaults.items():
        configured = str(envv.get(key, "")).strip()
        path = Path(configured).expanduser() if configured else default_path
        path.mkdir(parents=True, exist_ok=True)
        resolved_paths[key] = str(path.resolve())

    build_env = {
        "PATH": DEFAULT_PATH_PREFIX,
        "LANG": "C",
        "LC_ALL": "C",
        **resolved_paths,
        "CGO_ENABLED": str(envv.get("CGO_ENABLED", "0")).strip() or "0",
        "GO111MODULE": "on",
        "GOAUTH": "off",
        "GOENV": "off",
        "GOFLAGS": "",
        "GOTOOLCHAIN": "local",
        "GOWORK": "off",
        "GOPROXY": "https://proxy.golang.org,direct",
        "GOSUMDB": "sum.golang.org",
        "GOINSECURE": "",
        "GOVCS": "public:git,private:git",
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": "/dev/null",
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_ASKPASS": "/bin/false",
        "GIT_ALLOW_PROTOCOL": "https",
        "GIT_OPTIONAL_LOCKS": "0",
    }
    private_env = private_go_module_env(envv)
    private_env.update(
        {
            "GOPRIVATE": "github.com/pingcap-inc/*",
            "GONOPROXY": "github.com/pingcap-inc/*",
            "GONOSUMDB": "github.com/pingcap-inc/*",
        }
    )
    build_env.update(private_env)
    return build_env


def configure_release_setup_controlplane_workspace(
    envv: dict[str, str],
    *,
    build_root: Path,
    source_root: Path,
    build_env: dict[str, str],
) -> Path | None:
    """Bind setup to one exact Controlplane source without dirtying Core.

    The stopped-upgrade setup binary owns L0 schema reconciliation.  A caller
    that supplies an exact Controlplane source must compile that command from
    the same source revision as the separately published Catalog owner, not
    from whichever published module version happens to be pinned by Core.
    """
    raw_root = str(envv.get("NEXUSCRM_SETUP_CONTROLPLANE_ROOT", "")).strip()
    revision = str(envv.get("NEXUSCRM_SETUP_CONTROLPLANE_REF", "")).strip()
    if not raw_root and not revision:
        return None
    if not raw_root or not GIT_COMMIT_RE.fullmatch(revision):
        raise DeployError(
            "release setup exact Controlplane source requires root and commit"
        )

    controlplane_root = Path(raw_root).expanduser()
    if controlplane_root.is_symlink() or not controlplane_root.is_dir():
        raise DeployError("release setup exact Controlplane source is unavailable")
    controlplane_root = controlplane_root.resolve()
    top = Path(
        run(
            ["git", "-C", str(controlplane_root), "rev-parse", "--show-toplevel"],
            env=build_env,
            inherit_env=False,
            capture=True,
        ).strip()
    ).resolve()
    if top != controlplane_root:
        raise DeployError("release setup Controlplane root is not an exact checkout root")
    actual_revision = run(
        ["git", "-C", str(controlplane_root), "rev-parse", "HEAD"],
        env=build_env,
        inherit_env=False,
        capture=True,
    ).strip()
    if actual_revision != revision:
        raise DeployError("release setup Controlplane checkout revision mismatch")
    status = run(
        [
            "git", "-C", str(controlplane_root), "status", "--porcelain=v1",
            "--untracked-files=all",
        ],
        env=build_env,
        inherit_env=False,
        capture=True,
    ).strip()
    if status:
        raise DeployError("release setup Controlplane checkout is not pristine")

    distribution_root = backend_distribution_root(source_root)
    if distribution_root is None:
        raise DeployError(
            "release setup exact Controlplane binding requires a Core distribution module"
        )
    workspace = build_root / "go.work"
    completed = subprocess.run(
        ["go", "work", "init", str(distribution_root), str(controlplane_root)],
        cwd=str(build_root),
        env={**build_env, "GOWORK": "off"},
        text=True,
        capture_output=True,
        check=False,
    )
    if completed.returncode != 0:
        detail = (completed.stderr or completed.stdout).strip()
        raise DeployError(
            "release setup exact Controlplane workspace initialization failed"
            + (f": {detail}" if detail else "")
        )
    if not workspace.is_file():
        raise DeployError("release setup exact Controlplane workspace was not created")

    # Both exact modules may replace the same neutral Core module: the Core
    # distribution points at its local checkout while Controlplane's standalone
    # module points at the published Core pseudo-version.  The candidate already
    # binds one exact Core source, so resolve every local distribution replace at
    # the workspace layer.  Derive the set from the live go.mod instead of
    # copying the module inventory into deployment code.
    module_edit = subprocess.run(
        ["go", "mod", "edit", "-json"],
        cwd=str(distribution_root),
        env={**build_env, "GOWORK": "off"},
        text=True,
        capture_output=True,
        check=False,
    )
    if module_edit.returncode != 0:
        detail = (module_edit.stderr or module_edit.stdout).strip()
        raise DeployError(
            "release setup Core module replacement inventory failed"
            + (f": {detail}" if detail else "")
        )
    try:
        replacements = json.loads(module_edit.stdout).get("Replace", [])
    except (json.JSONDecodeError, AttributeError) as exc:
        raise DeployError("release setup Core module replacement inventory is invalid") from exc
    for replacement in replacements:
        old = replacement.get("Old", {})
        new = replacement.get("New", {})
        old_path = str(old.get("Path", "")).strip()
        old_version = str(old.get("Version", "")).strip()
        new_path = str(new.get("Path", "")).strip()
        new_version = str(new.get("Version", "")).strip()
        if not old_path or not new_path or new_version:
            continue
        local_target = Path(new_path)
        if not local_target.is_absolute():
            local_target = distribution_root / local_target
        local_target = local_target.resolve()
        if not local_target.is_dir():
            raise DeployError(
                f"release setup local Core module replacement is unavailable: {old_path}"
            )
        old_spec = old_path + (f"@{old_version}" if old_version else "")
        edited = subprocess.run(
            ["go", "work", "edit", f"-replace={old_spec}={local_target}"],
            cwd=str(build_root),
            env={**build_env, "GOWORK": str(workspace.resolve())},
            text=True,
            capture_output=True,
            check=False,
        )
        if edited.returncode != 0:
            detail = (edited.stderr or edited.stdout).strip()
            raise DeployError(
                f"release setup workspace replacement failed for {old_path}"
                + (f": {detail}" if detail else "")
            )
    return workspace.resolve()


def verify_release_build_vcs_stamp_input(
    source_root: Path,
    expected_revision: str,
    build_env: dict[str, str],
) -> None:
    status = run(
        ["git", "status", "--porcelain", "--untracked-files=all"],
        cwd=source_root,
        env=build_env,
        inherit_env=False,
        capture=True,
    ).strip()
    if status:
        raise DeployError("release build VCS stamp source is not clean")

    actual_revision = run(
        ["git", "-c", "log.showsignature=false", "log", "-1", "--format=%H"],
        cwd=source_root,
        env=build_env,
        inherit_env=False,
        capture=True,
    ).strip()
    if actual_revision != expected_revision:
        raise DeployError(
            "release build VCS stamp source revision does not match UPSTREAM_REF"
        )


def prepare_backend_setup_binary(envv: dict[str, str], nexuscrm_worktree: Path, app_packs_root: Path) -> Path | None:
    setup_bin = configured_backend_setup_binary(envv)
    if setup_bin is None:
        return None
    setup_bin.parent.mkdir(parents=True, exist_ok=True)
    expected_revision = str(envv.get("UPSTREAM_REF", "")).strip()
    with tempfile.TemporaryDirectory(
        prefix=f".{setup_bin.name}.source-", dir=setup_bin.parent
    ) as raw_build_root:
        build_root = Path(raw_build_root)
        source_root = materialize_release_setup_source_clone(
            nexuscrm_worktree=nexuscrm_worktree,
            app_packs_root=app_packs_root,
            expected_revision=expected_revision,
            build_root=build_root,
        )
        temporary_bin = build_root / "setup"
        setup_cwd, cmd, extra_env = backend_setup_binary_build_command(
            envv, source_root, app_packs_root, temporary_bin
        )
        build_env = release_setup_build_env(envv, build_root)
        build_env.update(extra_env)
        controlplane_workspace = configure_release_setup_controlplane_workspace(
            envv,
            build_root=build_root,
            source_root=source_root,
            build_env=build_env,
        )
        if controlplane_workspace is not None:
            build_env["GOWORK"] = str(controlplane_workspace)
        verify_release_build_vcs_stamp_input(
            source_root,
            expected_revision,
            build_env,
        )
        build_sensitive_values = (private_go_module_token(envv),)
        log(f"building run-scoped backend setup binary from clean source clone: {setup_bin}")
        result = subprocess.run(
            cmd,
            cwd=str(setup_cwd),
            env=build_env,
            text=True,
            capture_output=True,
            check=False,
        )
        if result.returncode != 0:
            if result.stdout:
                sys.stdout.write(
                    redact_sensitive_text(
                        result.stdout,
                        build_env,
                        cmd,
                        build_sensitive_values,
                    )
                )
            if result.stderr:
                sys.stderr.write(
                    redact_sensitive_text(
                        result.stderr,
                        build_env,
                        cmd,
                        build_sensitive_values,
                    )
                )
            raise DeployError(f"command failed ({result.returncode}): {' '.join(cmd)}")
        if not temporary_bin.is_file() or not os.access(temporary_bin, os.X_OK):
            raise DeployError("built setup binary is not executable")
        os.replace(temporary_bin, setup_bin)
    envv["NEXUSCRM_SETUP_BIN"] = str(setup_bin)
    return setup_bin


def stale_runtime_dir_candidates(runtime_dir: Path, keep_paths: list[Path]) -> list[Path]:
    runtime_dir = runtime_dir.resolve()
    keep_resolved = {path.resolve() for path in keep_paths}
    managed_prefixes = ("downstream-", "nexuscrm-")
    candidates: list[Path] = []

    for child in runtime_dir.iterdir():
        if child.is_symlink() or not child.is_dir():
            continue
        if not child.name.startswith(managed_prefixes):
            continue
        if child.resolve() in keep_resolved:
            continue
        candidates.append(child)
    return candidates


def managed_runtime_dir_removal_issues(path: Path, *, expected_uid: int | None = None) -> list[str]:
    expected_uid = os.geteuid() if expected_uid is None else expected_uid
    issues: list[str] = []
    try:
        root = path.resolve()
    except OSError as exc:
        return [f"{path}: cannot resolve path: {exc}"]

    try:
        parent = root.parent
        if not os.access(parent, os.W_OK | os.X_OK):
            issues.append(f"{parent}: parent directory is not writable/searchable")
    except OSError as exc:
        issues.append(f"{root}: cannot inspect parent directory: {exc}")

    for dirpath, dirnames, filenames in os.walk(root, topdown=True, followlinks=False):
        current = Path(dirpath)
        entries = [current]
        entries.extend(current / name for name in dirnames)
        entries.extend(current / name for name in filenames)
        for entry in entries:
            try:
                stat_result = entry.lstat()
            except OSError as exc:
                issues.append(f"{entry}: cannot stat: {exc}")
                continue
            if expected_uid is not None and stat_result.st_uid != expected_uid:
                issues.append(
                    f"{entry}: owner uid {stat_result.st_uid} != deploy uid {expected_uid}"
                )
            if entry.is_dir() and not entry.is_symlink() and not os.access(entry, os.W_OK | os.X_OK):
                issues.append(f"{entry}: directory is not writable/searchable")
        if issues:
            break
    return issues


def cleanup_stale_runtime_dirs(runtime_dir: Path, keep_paths: list[Path]) -> None:
    candidates = stale_runtime_dir_candidates(runtime_dir, keep_paths)
    drift: list[str] = []
    for child in candidates:
        drift.extend(managed_runtime_dir_removal_issues(child))
    if drift:
        preview = "\n".join(f"- {issue}" for issue in drift[:20])
        if len(drift) > 20:
            preview += f"\n- ... {len(drift) - 20} more issue(s)"
        raise DeployError(
            "stale runtime directory ownership preflight failed; "
            "repair managed runtime ownership before retention cleanup:\n"
            f"{preview}"
        )

    for child in candidates:
        log(f"removing stale runtime directory: {child}")
        shutil.rmtree(child)


def build_final_pack(
    *,
    framework_dir: Path,
    downstream_build_root: Path,
    upstream_root: Path,
    app_packs_root: Path,
    official_app_packs_ref: str,
    upstream_ref: str,
) -> None:
    validate_hubspot_signature_verifier_payload(app_packs_root, official_app_packs_ref)
    envv = {
        "PACK_REPO_ROOT": str(downstream_build_root),
        "OFFICIAL_APP_PACKS_SOURCE_REPO": str(app_packs_root),
        "OFFICIAL_APP_PACKS_REF": official_app_packs_ref,
        "NEXUSCRM_SOURCE_REPO": str(upstream_root),
        "NEXUSCRM_REF": upstream_ref,
    }
    envv.update(optional_process_env(os.environ))
    run(
        [
            "bash",
            str(framework_dir / "runtime/tools/build_final_pack.sh"),
            "--reset",
        ],
        cwd=downstream_build_root,
        env=envv,
    )


def validate_hubspot_signature_verifier_payload(app_packs_root: Path, official_app_packs_ref: str) -> None:
    signature_verifier = app_packs_root / HUBSPOT_SIGNATURE_VERIFIER_PATH
    if not signature_verifier.is_file():
        log(
            "HubSpot signature verifier guard skipped: "
            f"{signature_verifier} not present at official app packs ref {official_app_packs_ref}"
        )
        return

    raw = signature_verifier.read_text(encoding="utf-8")
    stale_fields = [field for field in STALE_HUBSPOT_SIGNATURE_IDENTITY_FIELDS if field in raw]
    if stale_fields:
        raise DeployError(
            "stale HubSpot signature verifier payload at resolved official app packs ref "
            f"{official_app_packs_ref}: {signature_verifier} contains unsupported field(s) "
            f"{', '.join(stale_fields)}. Use deliveryIdentityCollectionPath and "
            "deliveryIdentityCollectionKeyPaths; do not add L0 aliases for the legacy fields."
        )

    try:
        payload = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise DeployError(f"invalid HubSpot signature verifier JSON at {signature_verifier}: {exc}") from exc

    endpoint_configs = (
        payload.get("config", {}).get("endpointConfigs", {})
        if isinstance(payload, dict)
        else {}
    )
    webhook_config = endpoint_configs.get("webhook", {}) if isinstance(endpoint_configs, dict) else {}
    if isinstance(webhook_config, dict):
        log(
            "HubSpot signature verifier guard passed: "
            f"ref={official_app_packs_ref} path={signature_verifier} "
            f"collection_path={webhook_config.get('deliveryIdentityCollectionPath', '')}"
        )
    else:
        log(f"HubSpot signature verifier guard passed: ref={official_app_packs_ref} path={signature_verifier}")


def audit_final_pack(framework_dir: Path, downstream_build_root: Path) -> None:
    run(
        ["python3", str(framework_dir / "runtime/tools/audit_final_pack_metadata.py")],
        cwd=downstream_build_root,
        env={"PACK_REPO_ROOT": str(downstream_build_root)},
    )


def verify_final_pack_snapshot(framework_dir: Path, downstream_build_root: Path) -> None:
    merged_env = os.environ.copy()
    merged_env["PACK_REPO_ROOT"] = str(downstream_build_root)
    result = subprocess.run(
        ["python3", str(framework_dir / "runtime/tools/verify_final_pack_snapshot.py")],
        cwd=str(downstream_build_root),
        env=merged_env,
        text=True,
        capture_output=True,
        check=False,
    )
    if result.returncode == 0:
        if result.stdout:
            sys.stdout.write(result.stdout)
        if result.stderr:
            sys.stderr.write(result.stderr)
        return
    if result.stdout:
        sys.stdout.write(result.stdout)
    if result.stderr:
        sys.stderr.write(result.stderr)
    if "pack.json drift classification: order-only" in result.stderr:
        log("final pack snapshot drift is order-only; continuing remote gate")
        return
    raise DeployError(
        f"command failed ({result.returncode}): python3 {framework_dir / 'runtime/tools/verify_final_pack_snapshot.py'}"
    )


# --- nexuscrm-connect download artifacts (AI connect page) -------------------
#
# The app-blind L0 download endpoint (GET /downloads/{kind}/latest) reads its
# kind -> {key, filename} catalog from NEXUSCRM_DOWNLOAD_ARTIFACTS. The tenant
# owns the concrete artifact identity: keys embed the agent-tools package
# version so every release publishes to an immutable object-store key, while
# the public /downloads/{kind}/latest URL stays stable.
#
# The single source of truth for that version is the SELECTED release checkout
# (downstream_build_root) — the same tree we build and publish from — not the
# deploy script's own checkout. The prepare step seeds the catalog env as a
# best-effort default; the deploy step reconciles both the env and the object
# store to the selected checkout so a release-tag / downstream-ref override can
# never leave the served catalog pointing at keys the release did not publish.

CONNECT_DOWNLOAD_ARTIFACT_KEY_PREFIX = "release-artifact/nexuscrm-connect"
CONNECT_DOWNLOAD_ARTIFACT_BASENAMES = {"cli": "cli.tgz", "plugin": "plugin.dxt"}
CONNECT_DOWNLOAD_ARTIFACT_FILENAMES = {
    "cli": "nexuscrm-connect-cli.tgz",
    "plugin": "nexuscrm-connect-plugin.dxt",
}
CONNECT_VERSION_RE = re.compile(r"^\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?$")
CONNECT_BUILD_SCRIPT_RELPATH = Path("agent-tools") / "build_artifacts.py"

def connect_package_version(repo_root: Path) -> str:
    package_json = repo_root / "agent-tools" / "nexuscrm-connect" / "package.json"
    try:
        manifest = json.loads(package_json.read_text())
    except FileNotFoundError as exc:
        raise DeployError(f"missing nexuscrm-connect package manifest: {package_json}") from exc
    except json.JSONDecodeError as exc:
        raise DeployError(f"invalid nexuscrm-connect package manifest {package_json}: {exc}") from exc
    version = str(manifest.get("version", "")).strip()
    if not CONNECT_VERSION_RE.match(version):
        raise DeployError(
            f"nexuscrm-connect package version must be semver, got {version!r}: {package_json}"
        )
    return version


def connect_download_artifacts_catalog(version: str) -> dict[str, dict[str, str]]:
    if not CONNECT_VERSION_RE.match(str(version)):
        raise DeployError(f"connect download artifact version must be semver, got {version!r}")
    return {
        kind: {
            "key": f"{CONNECT_DOWNLOAD_ARTIFACT_KEY_PREFIX}/{version}/{basename}",
            "filename": CONNECT_DOWNLOAD_ARTIFACT_FILENAMES[kind],
        }
        for kind, basename in CONNECT_DOWNLOAD_ARTIFACT_BASENAMES.items()
    }


def connect_download_artifacts_env_value(version: str) -> str:
    return json.dumps(connect_download_artifacts_catalog(version), sort_keys=True, separators=(",", ":"))


def download_artifacts_env_json(envv: dict[str, str]) -> str:
    """NEXUSCRM_DOWNLOAD_ARTIFACTS with any residual shell quoting stripped."""
    return shell_unquote_env_value(str(envv.get("NEXUSCRM_DOWNLOAD_ARTIFACTS", "")).strip())


def preflight_connect_download_artifacts(downstream_build_root: Path) -> str:
    """Fail fast, before any database mutation, if the selected release checkout
    cannot produce the download artifacts.

    Returns the selected checkout's connect package version so callers can log or
    reconcile it. Must be called while the build root is available and before
    setup apply, so a stale/override ref that lacks the build script or a valid
    version aborts the release before it partially mutates the deployment.
    """
    build_script = downstream_build_root / CONNECT_BUILD_SCRIPT_RELPATH
    if not build_script.is_file():
        raise DeployError(f"missing connect artifact build script: {build_script}")
    return connect_package_version(downstream_build_root)


def connect_mcp_url_from_env(envv: dict[str, str]) -> str:
    """The deploying environment's own hosted ``/mcp`` URL.

    Passed to the artifact build as ``--mcp-url``. Only the local-stdio
    ``mcp/mcp.json`` registration snippet is stamped with it (as the base URL,
    minus the ``/mcp`` suffix) so an artifact published by a test/staging deploy
    configures clients for THAT environment, never production. The plugin manifest
    declares no MCP server and is not stamped. Derived from the environment's
    public base URL.
    """
    base = str(envv.get("PUBLIC_BASE_URL", "")).strip().rstrip("/")
    if not base.startswith("https://"):
        raise DeployError(
            f"PUBLIC_BASE_URL must be an https base url to build connect artifacts, got {base!r}"
        )
    return f"{base}/mcp"


def build_connect_download_artifacts(
    envv: dict[str, str],
    downstream_build_root: Path,
    *,
    mcp_url: str | None = None,
    output_dir_name: str = "connect-download-artifacts",
) -> dict[str, object]:
    """Build the nexuscrm-connect cli/plugin artifacts from the SELECTED checkout.

    Meant to run BEFORE any database mutation: it does all the failure-prone
    work (preflight, packaging, version cross-check) so a stale/override ref or a
    broken build aborts the release before it mutates state — never after.

    The version — and therefore the immutable object-store keys — is derived from
    downstream_build_root, the exact tree the build runs against, so the published
    keys can never diverge from the deployed source. The shipped local-stdio
    ``mcp/mcp.json`` snippet is stamped with THIS environment's base URL so a
    test/staging artifact never configures clients for production; the plugin
    manifest declares no MCP server and is not stamped. ``envv`` is reconciled in place
    to the matching catalog (so callers persist the value the backend serves and
    the later publish forwards the right keys). Returns the build summary (paths,
    sha256, version) to hand to publish_connect_download_artifacts.
    """
    version = preflight_connect_download_artifacts(downstream_build_root)
    build_script = downstream_build_root / CONNECT_BUILD_SCRIPT_RELPATH
    mcp_url = str(mcp_url or connect_mcp_url_from_env(envv)).strip()
    if not mcp_url.startswith("https://") or not mcp_url.endswith("/mcp"):
        raise DeployError(f"connect artifact MCP URL must be an https /mcp URL, got {mcp_url!r}")
    if not re.fullmatch(r"[a-z][a-z0-9-]*", output_dir_name):
        raise DeployError(f"invalid connect artifact output directory name: {output_dir_name!r}")
    out_dir = Path(envv["DEPLOY_ARTIFACT_DIR"]).expanduser().resolve() / output_dir_name
    result = subprocess.run(
        ["python3", str(build_script), "--out-dir", str(out_dir), "--mcp-url", mcp_url],
        text=True,
        capture_output=True,
        check=False,
    )
    if result.stderr:
        sys.stderr.write(result.stderr)
    if result.returncode != 0:
        raise DeployError(f"connect artifact build failed ({result.returncode}): {build_script}")
    try:
        summary = json.loads(result.stdout)
    except json.JSONDecodeError as exc:
        raise DeployError(f"connect artifact build emitted invalid JSON summary: {exc}") from exc
    built_version = str(summary.get("version", "")).strip()
    if built_version != version:
        raise DeployError(
            "connect artifact build version does not match the selected checkout manifest: "
            f"built={built_version!r} manifest={version!r}"
        )
    for kind in CONNECT_DOWNLOAD_ARTIFACT_BASENAMES:
        artifact = summary.get(kind, {})
        artifact_path = Path(str(artifact.get("path", "")))
        if not artifact_path.is_file():
            raise DeployError(f"connect artifact build produced no {kind} file: {artifact!r}")

    # Reconcile the catalog env to the selected checkout so the running backend
    # serves exactly the keys this release publishes. Stored unquoted here;
    # callers persist it to their env file with the writer they own.
    envv["NEXUSCRM_DOWNLOAD_ARTIFACTS"] = connect_download_artifacts_env_value(version)
    return summary


def publish_connect_download_artifacts(
    envv: dict[str, str],
    summary: dict[str, object],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
) -> dict[str, object]:
    """Upload the already-built cli/plugin artifacts to the backend object store.

    Consumes the summary from build_connect_download_artifacts (which already
    validated the build and reconciled the catalog env) and uploads each kind via
    the backend's ``publish-artifact`` setup subcommand, so GET
    /downloads/{kind}/latest serves exactly what this release built. Kept separate
    from the build so the upload — which needs the running backend and therefore
    can only happen after setup apply — is the sole post-mutation step, and it
    only ships pre-validated files to immutable versioned keys (idempotent and
    safe to re-run).
    """
    version = str(summary.get("version", "")).strip()
    if not version:
        raise DeployError(
            "connect artifact build summary is missing a version; rebuild before publishing"
        )
    catalog = connect_download_artifacts_catalog(version)
    # Keep the served catalog pinned to the built version so publish-artifact
    # resolves the same immutable keys the backend serves (idempotent no-op when
    # build_connect_download_artifacts already reconciled envv).
    envv["NEXUSCRM_DOWNLOAD_ARTIFACTS"] = connect_download_artifacts_env_value(version)
    for kind in CONNECT_DOWNLOAD_ARTIFACT_BASENAMES:
        artifact = summary.get(kind, {})
        artifact_path = Path(str(artifact.get("path", "")))
        if not artifact_path.is_file():
            raise DeployError(f"connect artifact to publish is missing for {kind}: {artifact!r}")
        log(
            f"publishing connect download artifact kind={kind} version={version} "
            f"sha256={artifact.get('sha256', '')} key={catalog[kind]['key']}"
        )
        run_backend_setup(
            envv,
            nexuscrm_worktree,
            app_packs_root,
            [
                "publish-artifact",
                "--kind",
                kind,
                "--file",
                str(artifact_path),
                "--filename",
                catalog[kind]["filename"],
            ],
        )
    return summary


def build_and_publish_connect_download_artifacts(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
) -> dict[str, object]:
    """Build then publish the connect artifacts in one call.

    Convenience wrapper for callers that build and publish adjacently. Release
    deploys instead call build_connect_download_artifacts BEFORE any database
    mutation and publish_connect_download_artifacts after setup apply, so a build
    failure can never leave the deployment mutated without published artifacts.
    """
    summary = build_connect_download_artifacts(envv, downstream_build_root)
    return publish_connect_download_artifacts(envv, summary, nexuscrm_worktree, app_packs_root)


def run_backend_setup(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    args: list[str],
    *,
    allow_exit_codes: set[int] | None = None,
) -> int:
    frontend_url = runtime_frontend_url(envv)
    skip_bootstrap = bool_env(envv, "SKIP_BOOTSTRAP")
    nexus_env = (envv.get("NEXUS_ENV") or "").strip()
    if nexus_env:
        if nexus_env != "production":
            raise DeployError("NEXUS_ENV must be production for remote deployment")
    else:
        nexus_env = "dev" if skip_bootstrap else "production"
    # The setup subprocess must be reproducible from runtime.env alone: build its
    # environment from the parsed runtime env (envv) plus only the runner-level
    # variables the Go toolchain needs (PATH/HOME). Deploy parameters and secrets
    # (NEXUSCRM_NC_MASTER_KEY, JWT_SECRET, HubSpot, ...) come from runtime.env, not
    # the ambient process environment, so a local run with just runtime.env behaves
    # identically to CI.
    setup_env = runner_toolchain_passthrough_env()
    setup_env.update(envv)
    setup_env.update(
        {
            "NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": str(
                envv.get("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT") or app_packs_root
            ),
            "TIDB_HOST": envv["TIDB_HOST"],
            "TIDB_PORT": envv["TIDB_PORT"],
            "TIDB_USER": envv["TIDB_USER"],
            "TIDB_DATABASE": envv["TIDB_DATABASE"],
            "JWT_SECRET": envv["JWT_SECRET"],
            "NEXUS_ADMIN_PASSWORD": envv["NEXUS_ADMIN_PASSWORD"],
            "PORT": envv["BACKEND_PORT"],
            "API_BASE_URL": envv["PUBLIC_BASE_URL"],
            "FRONTEND_URL": frontend_url,
            "FILE_STORAGE_PROVIDER": envv.get("FILE_STORAGE_PROVIDER", "filesystem"),
            "FILE_STORAGE_ROOT": envv.get("FILE_STORAGE_ROOT", "backend/data/object-storage"),
            "FILE_STORAGE_LEGACY_UPLOAD_ROOT": envv.get("FILE_STORAGE_LEGACY_UPLOAD_ROOT", "backend/data/uploads"),
            # Bootstrap ownership and deployment security profile are separate
            # facts. Preserve the release default; only the explicitly named
            # disposable local clean install may select a weaker profile.
            "NEXUS_ENV": nexus_env,
            "NODE_ENV": "production",
            "LOG_FORMAT": "json",
            "AUTONUMBER_SEED_TOLERATE_LEGACY": envv.get("AUTONUMBER_SEED_TOLERATE_LEGACY", "true"),
            "ENSURE_INDEXES_WORKERS": envv.get("ENSURE_INDEXES_WORKERS", "2"),
        }
    )
    if envv.get("TIDB_PASSWORD"):
        setup_env["TIDB_PASSWORD"] = envv["TIDB_PASSWORD"]
    if envv.get("CORS_ALLOWED_ORIGINS"):
        setup_env["CORS_ALLOWED_ORIGINS"] = envv["CORS_ALLOWED_ORIGINS"]
    setup_env.update(runtime_rate_limit_env(envv))
    setup_env.update(optional_process_env(envv))
    setup_env.update(private_go_module_env(envv))
    setup_bin = configured_backend_setup_binary(envv)
    if setup_bin is not None:
        if not setup_bin.is_file() or not os.access(setup_bin, os.X_OK):
            raise DeployError(f"NEXUSCRM_SETUP_BIN is not executable: {setup_bin}")
        cmd = [str(setup_bin), *args]
        distribution_root = backend_distribution_root(nexuscrm_worktree)
        setup_cwd = distribution_root or nexuscrm_worktree / "backend"
    else:
        setup_cwd, cmd, setup_extra_env = backend_setup_command(envv, nexuscrm_worktree, app_packs_root, args)
        setup_env.update(setup_extra_env)
    timeout_seconds = setup_command_timeout_seconds(envv)
    log(f"running setup command with live output (timeout={timeout_seconds}s): {' '.join(cmd)}")
    process = subprocess.Popen(
        cmd,
        cwd=str(setup_cwd),
        env=setup_env,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
        start_new_session=True,
    )
    output_queue: queue.Queue[str | None] = queue.Queue()
    output_tail: deque[str] = deque(maxlen=SETUP_COMMAND_OUTPUT_TAIL_LINES)

    def forward_output() -> None:
        if process.stdout is not None:
            for line in process.stdout:
                output_queue.put(line)
        output_queue.put(None)

    reader = threading.Thread(target=forward_output, name="nexuscrm-setup-output", daemon=True)
    reader.start()
    deadline = time.monotonic() + timeout_seconds
    stream_closed = False
    try:
        while True:
            remaining_seconds = deadline - time.monotonic()
            if remaining_seconds <= 0:
                request_setup_timeout_diagnostics(process)
                terminate_setup_process_group(process)
                close_setup_output_stream(process, reader)
                while True:
                    try:
                        diagnostic_line = output_queue.get_nowait()
                    except queue.Empty:
                        break
                    if diagnostic_line is None:
                        continue
                    output_tail.append(diagnostic_line)
                    sys.stdout.write(diagnostic_line)
                    sys.stdout.flush()
                raise DeployError(f"setup command timed out after {timeout_seconds}s: {' '.join(cmd)}")
            try:
                line = output_queue.get(timeout=min(1.0, remaining_seconds))
            except queue.Empty:
                if stream_closed and process.poll() is not None:
                    break
                continue
            if line is None:
                stream_closed = True
            else:
                output_tail.append(line)
                sys.stdout.write(line)
                sys.stdout.flush()
            if stream_closed and process.poll() is not None:
                break
    except BaseException:
        if process.poll() is None:
            terminate_setup_process_group(process)
        close_setup_output_stream(process, reader)
        raise

    close_setup_output_stream(process, reader)
    process.wait()
    result_code = process.returncode
    if result_code == 0:
        return 0
    allowed = allow_exit_codes or set()
    if result_code in allowed:
        return result_code
    wrapped_go_run_exit = re.search(r"exit status (\d+)", "".join(output_tail))
    if wrapped_go_run_exit:
        wrapped_code = int(wrapped_go_run_exit.group(1))
        if wrapped_code in allowed:
            return wrapped_code
    raise SetupCommandError(
        f"command failed ({result_code}): {' '.join(cmd)}",
        output_tail=redact_sensitive_text("".join(output_tail), setup_env, cmd),
    )


def setup_command_timeout_seconds(envv: dict[str, str]) -> int:
    raw_value = str(envv.get("SETUP_COMMAND_TIMEOUT_SECONDS", DEFAULT_SETUP_COMMAND_TIMEOUT_SECONDS)).strip()
    try:
        timeout_seconds = int(raw_value)
    except ValueError as exc:
        raise DeployError(f"SETUP_COMMAND_TIMEOUT_SECONDS must be a positive integer, got {raw_value!r}") from exc
    if timeout_seconds <= 0:
        raise DeployError(f"SETUP_COMMAND_TIMEOUT_SECONDS must be a positive integer, got {raw_value!r}")
    return timeout_seconds


def request_setup_timeout_diagnostics(process: subprocess.Popen[str]) -> None:
    """Ask a stuck Go setup process group to emit goroutine stacks before termination."""
    if os.name != "posix" or process.poll() is not None:
        return
    log("setup command deadline exceeded; requesting SIGQUIT stack diagnostics")
    try:
        os.killpg(process.pid, signal.SIGQUIT)
    except ProcessLookupError:
        return
    try:
        process.wait(timeout=SETUP_COMMAND_DIAGNOSTIC_GRACE_SECONDS)
    except subprocess.TimeoutExpired:
        pass


def terminate_setup_process_group(process: subprocess.Popen[str]) -> None:
    """Terminate setup and any go-run child together after a deploy deadline."""
    try:
        if os.name == "posix":
            os.killpg(process.pid, signal.SIGTERM)
        else:
            process.terminate()
    except ProcessLookupError:
        return
    try:
        process.wait(timeout=10)
        return
    except subprocess.TimeoutExpired:
        pass
    try:
        if os.name == "posix":
            os.killpg(process.pid, signal.SIGKILL)
        else:
            process.kill()
    except ProcessLookupError:
        return
    process.wait()


def close_setup_output_stream(process: subprocess.Popen[str], reader: threading.Thread) -> None:
    reader.join(timeout=1)
    if process.stdout is not None:
        process.stdout.close()


def refresh_setup_derived_state(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    *,
    reason: str,
) -> None:
    log(f"refreshing setup-derived views and caches ({reason})")
    run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["rebuild"])


def backend_auth_import_path(backend_root: Path) -> str:
    if (backend_root / "pkg" / "auth").is_dir():
        return "github.com/nexuscrm/backend/pkg/auth"
    return "github.com/nexuscrm/kernel/pkg/auth"


def bcrypt_hash_helper_source(backend_root: Path) -> str:
    return """package main

import (
    "fmt"
    "os"

    authpkg "__AUTH_IMPORT_PATH__"
)

func main() {
    if len(os.Args) != 2 {
        fmt.Fprintln(os.Stderr, "usage: bcrypt_hash <password>")
        os.Exit(2)
    }
    hashed, err := authpkg.HashPassword(os.Args[1])
    if err != nil {
        fmt.Fprintln(os.Stderr, err)
        os.Exit(1)
    }
    fmt.Print(hashed)
}
""".replace("__AUTH_IMPORT_PATH__", backend_auth_import_path(backend_root))


def prepare_bcrypt_hash_binary(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    target: Path,
) -> Path:
    backend_root = nexuscrm_worktree / "backend"
    resolved_target = target.expanduser().resolve()
    resolved_target.parent.mkdir(parents=True, exist_ok=True)
    temporary_binary = resolved_target.with_name(f".{resolved_target.name}.tmp")
    remove_path(temporary_binary)
    tool_env = os.environ.copy()
    tool_env.update(optional_process_env(envv))
    tool_env.update(private_go_module_env(envv))
    with tempfile.TemporaryDirectory(prefix="nexuscrm-bcrypt-hash-build-") as temp_dir:
        tool_path = Path(temp_dir) / "main.go"
        tool_path.write_text(bcrypt_hash_helper_source(backend_root), encoding="utf-8")
        command = [
            "go",
            "build",
            "-mod=readonly",
            "-o",
            str(temporary_binary),
            str(tool_path),
        ]
        result = subprocess.run(
            command,
            cwd=backend_root,
            text=True,
            capture_output=True,
            check=False,
            env=tool_env,
        )
    if result.returncode != 0:
        remove_path(temporary_binary)
        stderr = redact_sensitive_text(result.stderr.strip(), tool_env, command)
        raise DeployError(f"failed to build bcrypt password hash helper: {stderr}")
    temporary_binary.chmod(0o755)
    os.replace(temporary_binary, resolved_target)
    return resolved_target


def bcrypt_hash(password: str, nexuscrm_worktree: Path, *, envv: dict[str, str] | None = None) -> str:
    backend_root = nexuscrm_worktree / "backend"
    tool_env = os.environ.copy()
    tool_env.update(optional_process_env(envv or {}))
    tool_env.update(private_go_module_env(envv or {}))
    configured_helper = str((envv or {}).get("NEXUSCRM_BCRYPT_HASH_BIN", "")).strip()
    if configured_helper:
        helper = Path(configured_helper).expanduser()
        if (
            not helper.is_absolute()
            or helper.is_symlink()
            or not helper.is_file()
            or not os.access(helper, os.X_OK)
        ):
            raise DeployError(
                f"NEXUSCRM_BCRYPT_HASH_BIN is not an absolute executable file: {helper}"
            )
        command = [str(helper), password]
        result = subprocess.run(
            command,
            text=True,
            capture_output=True,
            check=False,
            env=tool_env,
        )
    else:
        with tempfile.TemporaryDirectory(prefix="nexuscrm-bcrypt-hash-") as temp_dir:
            tool_path = Path(temp_dir) / "main.go"
            tool_path.write_text(bcrypt_hash_helper_source(backend_root), encoding="utf-8")
            command = ["go", "run", str(tool_path), password]
            result = subprocess.run(
                command,
                cwd=backend_root,
                text=True,
                capture_output=True,
                check=False,
                env=tool_env,
            )
    if result.returncode != 0:
        stderr = redact_sensitive_text(result.stderr.strip(), tool_env, command)
        raise DeployError(f"failed to generate bcrypt password hash: {stderr}")
    hashed = (result.stdout or "").strip()
    if not hashed.startswith("$2"):
        raise DeployError("generated password hash is not bcrypt formatted")
    return hashed


def align_runtime_admin_password(envv: dict[str, str], nexuscrm_worktree: Path) -> None:
    password = envv.get("NEXUS_ADMIN_PASSWORD", "").strip()
    username = envv.get("VERIFY_USERNAME", "").strip() or "admin@example.invalid"
    if not password:
        raise DeployError("missing required env keys: NEXUS_ADMIN_PASSWORD")
    hashed = bcrypt_hash(password, nexuscrm_worktree, envv=envv)
    escaped_hash = hashed.replace("'", "''")
    escaped_username = username.replace("'", "''")
    sql = f"""
UPDATE `User`
SET
  `Username` = '{escaped_username}',
  `Email` = '{escaped_username}',
  `Password` = '{escaped_hash}',
  `ProfileId` = 'system_admin',
  `IsActive` = 1,
  `FailedLoginAttempts` = 0,
  `LockedUntil` = NULL,
  `LastModifiedDate` = CURRENT_TIMESTAMP
WHERE `Username` = '{escaped_username}'
   OR `Email` = '{escaped_username}';
"""
    db_exec(envv, sql.strip(), database=envv["TIDB_DATABASE"])
    matched = db_query_scalar(
        envv,
        f"""
SELECT COUNT(*)
FROM `User`
WHERE `Email` = '{escaped_username}'
  AND `Username` = '{escaped_username}'
  AND `Password` = '{escaped_hash}'
  AND `ProfileId` = 'system_admin'
  AND `IsActive` = 1
""".strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()
    if matched != "1":
        raise DeployError(f"failed to align runtime admin identity for {username}: matched rows={matched or '0'}")


def refresh_runtime_metadata_cache(envv: dict[str, str], *, reason: str) -> dict[str, object]:
    if bool_env(envv, "SKIP_BACKEND_START"):
        log(f"skipping live metadata cache refresh ({reason}; SKIP_BACKEND_START=1)")
        return {}

    base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    username = envv.get("VERIFY_USERNAME", "").strip() or "admin@example.invalid"
    password = envv.get("NEXUS_ADMIN_PASSWORD", "").strip()
    if not password:
        raise DeployError("cannot refresh live metadata cache: missing NEXUS_ADMIN_PASSWORD")

    log(f"refreshing live backend metadata cache ({reason})")
    token_json = oauth_password_grant(base_url, username=username, password=password)
    access_token = str(token_json.get("access_token", "")).strip()
    if not access_token:
        raise DeployError("metadata cache refresh oauth response missing access_token")

    request = Request(
        f"{base_url}/services/data/v63.0/metadata/refresh-cache",
        headers={"Authorization": f"Bearer {access_token}"},
        method="POST",
    )
    # The full refresh-cache call rebuilds object/flow/rule caches (~250ms) and
    # the autonumber metadata snapshot. The autonumber Refresh holds a process
    # mutex; on a freshly reset TiDB branch (reset_branch deploys) the standby
    # finalize call can collide with apply-stage autonumber seed/verify on the
    # same lock + cold-scan window and block for minutes, while the steady-state
    # call is sub-second (measured 0.46s). The collision window is short and
    # transient, so retry on timeout/connection failure rather than betting the
    # whole deploy on a single long-timeout shot. Each attempt waits up to 240s
    # for the cold-branch case; transient failures back off and retry.
    attempts = 4
    backoff = 5.0
    last_exc: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            with urlopen(request, timeout=240) as response:
                body = response.read().decode()
                status = response.status
        except HTTPError as exc:
            # A non-2xx response is a definitive server answer, not a transient
            # network stall; surface it immediately.
            body = exc.read().decode(errors="replace")
            raise DeployError(f"metadata cache refresh failed: http {exc.code}: {body[:500]}") from exc
        except (URLError, TimeoutError, OSError) as exc:
            # socket.timeout (TimeoutError/OSError) and URLError-wrapped timeouts
            # are the transient lock/cold-scan collision; retry with backoff.
            last_exc = exc
            if attempt == attempts:
                break
            wait = backoff * attempt
            log(
                f"metadata cache refresh transient failure "
                f"(attempt {attempt}/{attempts}): {exc}; retrying in {wait:.0f}s"
            )
            time.sleep(wait)
            continue

        if status != 200:
            raise DeployError(f"metadata cache refresh failed: http {status}: {body[:500]}")
        payload = json.loads(body) if body.strip() else {}
        if isinstance(payload, dict) and payload.get("success") is False:
            raise DeployError(f"metadata cache refresh failed: {body[:500]}")
        log("live backend metadata cache refreshed")
        return payload if isinstance(payload, dict) else {}

    raise DeployError(
        f"metadata cache refresh failed after {attempts} attempt(s): {last_exc}"
    ) from last_exc


def rebind_installed_pack_source_local_path(envv: dict[str, str], pack_name: str, pack_dir: Path) -> None:
    if not installed_pack_table_exists(envv):
        return
    escaped_name = pack_name.replace("'", "''")
    escaped_path = str(pack_dir.resolve()).replace("'", "''")
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_InstalledPack
            SET SourceKind='local_path',
                SourceURI='{escaped_path}',
                LastModifiedDate=UTC_TIMESTAMP()
            WHERE Name='{escaped_name}'
              AND Status='active'
              AND COALESCE(IsDeleted, 0) = 0;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def git_head_revision(repo_root: Path) -> str:
    result = subprocess.run(
        ["git", "-C", str(repo_root), "rev-parse", "HEAD"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return ""
    return (result.stdout or "").strip()


def exact_git_source_revision(source_root: Path) -> str:
    source_root = source_root.expanduser().resolve()
    result = subprocess.run(
        ["git", "-C", str(source_root), "rev-parse", "--show-toplevel", "HEAD"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return ""
    lines = (result.stdout or "").splitlines()
    if len(lines) != 2:
        return ""
    try:
        git_root = Path(lines[0]).expanduser().resolve()
    except OSError:
        return ""
    revision = lines[1].strip()
    if git_root != source_root or not GIT_COMMIT_RE.fullmatch(revision):
        return ""
    return revision


def exact_downstream_source_revision(downstream_root: Path) -> str:
    """Return the exact Git or sealed-bundle revision for a downstream source root.

    A materialized deployment bundle is an ordinary file tree, not a Git
    checkout. Never let ``git -C`` walk through that tree into an unrelated
    ancestor checkout; its validated materialization receipt is the portable
    source authority.
    """
    downstream_root = downstream_root.expanduser().resolve()
    revision = exact_git_source_revision(downstream_root)
    if revision:
        return revision

    bundle_root = downstream_root.parent
    receipt_path = bundle_root / deployment_bundle.MATERIALIZED_RECEIPT_NAME
    if not receipt_path.is_file() or receipt_path.is_symlink():
        return ""
    try:
        receipt = require_json_object(
            json.loads(receipt_path.read_text()), str(receipt_path)
        )
    except (OSError, json.JSONDecodeError) as exc:
        raise DeployError("materialized deployment bundle receipt is invalid") from exc
    if receipt.get("contract_version") != deployment_bundle.BUNDLE_RECEIPT_VERSION:
        raise DeployError("materialized deployment bundle receipt version is invalid")
    for field in ("archive_sha256", "manifest_sha256"):
        if not SHA256_RE.fullmatch(str(receipt.get(field, ""))):
            raise DeployError(
                f"materialized deployment bundle receipt {field} is invalid"
            )
    downstream_revision = str(receipt.get("downstream_ref", "")).strip()
    if not GIT_COMMIT_RE.fullmatch(downstream_revision):
        raise DeployError(
            "materialized deployment bundle receipt downstream_ref is invalid"
        )
    downstream_relative_path = PurePosixPath(
        str(receipt.get("downstream_relative_path", ""))
    )
    if (
        downstream_relative_path.is_absolute()
        or ".." in downstream_relative_path.parts
        or not downstream_relative_path.parts
        or (bundle_root / Path(*downstream_relative_path.parts)).resolve()
        != downstream_root
    ):
        raise DeployError(
            "materialized deployment bundle receipt downstream path is invalid"
        )
    return downstream_revision


def rebind_installed_pack_source_official_repo(envv: dict[str, str], pack_name: str, app_packs_root: Path) -> None:
    if not installed_pack_table_exists(envv):
        return
    escaped_name = pack_name.replace("'", "''")
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_InstalledPack
            SET SourceKind='official_repo',
                SourceURI='{escaped_name}',
                LastModifiedDate=UTC_TIMESTAMP()
            WHERE Name='{escaped_name}'
              AND Status='active'
              AND COALESCE(IsDeleted, 0) = 0;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def normalize_installed_pack_runtime_state(envv: dict[str, str], pack_name: str, pack_dir: Path) -> None:
    if not installed_pack_table_exists(envv):
        return
    escaped_name = pack_name.replace("'", "''")
    escaped_path = str(pack_dir.resolve()).replace("'", "''")
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_InstalledPack
            SET Status='active',
                SourceKind='local_path',
                SourceURI='{escaped_path}',
                LastError=NULL,
                LastModifiedDate=UTC_TIMESTAMP()
            WHERE Name='{escaped_name}'
              AND Status IN ('active', 'installing', 'failed')
              AND COALESCE(IsDeleted, 0) = 0;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def rebind_active_stack_pack_sources(
    envv: dict[str, str],
    downstream_build_root: Path,
    app_packs_root: Path,
    pack_entries: list[str],
) -> None:
    for pack_entry in pack_entries:
        pack_name, _, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        if installed_pack_status(envv, pack_name) != "active":
            continue
        if pack_dir is None:
            log(f"rebind active official pack source before setup repair: {pack_name}")
            rebind_installed_pack_source_official_repo(envv, pack_name, app_packs_root)
        else:
            log(f"rebind active local pack source before setup repair: {pack_name}")
            rebind_installed_pack_source_local_path(envv, pack_name, pack_dir)


def installed_pack_status(envv: dict[str, str], pack_name: str) -> str:
    if not installed_pack_table_exists(envv):
        return ""
    escaped_name = pack_name.replace("\\", "\\\\").replace("'", "''")
    return db_query_scalar(
        envv,
        f"SELECT COALESCE(Status, '') FROM _System_InstalledPack WHERE Name='{escaped_name}' AND COALESCE(IsDeleted, 0)=0 LIMIT 1;",
        database=envv["TIDB_DATABASE"],
    )


def load_active_installed_pack_source_revision(envv: dict[str, str], pack_name: str) -> str:
    if not installed_pack_table_exists(envv):
        return ""
    escaped_name = pack_name.replace("\\", "\\\\").replace("'", "''")
    return db_query_scalar(
        envv,
        f"SELECT COALESCE(SourceRevision, '') FROM _System_InstalledPack WHERE Name='{escaped_name}' AND Status='active' AND COALESCE(IsDeleted, 0)=0 LIMIT 1;",
        database=envv["TIDB_DATABASE"],
    ).strip()


def load_active_installed_pack_source_state(envv: dict[str, str], pack_name: str) -> tuple[str, str]:
    if not installed_pack_table_exists(envv):
        return "", ""
    escaped_name = pack_name.replace("\\", "\\\\").replace("'", "''")
    row = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT JSON_OBJECT(
                'source_kind', COALESCE(SourceKind, ''),
                'source_uri', COALESCE(SourceURI, '')
            )
            FROM _System_InstalledPack
            WHERE Name='{escaped_name}'
              AND Status='active'
              AND COALESCE(IsDeleted, 0)=0
            LIMIT 1;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    if not row:
        return "", ""
    payload = json.loads(row)
    if not isinstance(payload, dict):
        return "", ""
    return (
        str(payload.get("source_kind", "")).strip(),
        str(payload.get("source_uri", "")).strip(),
    )


def load_installed_pack_convergence_state(
    envv: dict[str, str],
    pack_name: str,
) -> tuple[str, str, str, str]:
    """Read one authoritative source tuple for active or retryable pack state."""
    if not installed_pack_table_exists(envv):
        return "", "", "", ""
    escaped_name = pack_name.replace("\\", "\\\\").replace("'", "''")
    row = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT JSON_OBJECT(
                'status', COALESCE(Status, ''),
                'source_kind', COALESCE(SourceKind, ''),
                'source_uri', COALESCE(SourceURI, ''),
                'source_revision', COALESCE(SourceRevision, '')
            )
            FROM _System_InstalledPack
            WHERE Name='{escaped_name}'
              AND Status IN ('active', 'installing', 'failed')
              AND COALESCE(IsDeleted, 0)=0
            LIMIT 1;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    if not row:
        return "", "", "", ""
    payload = json.loads(row)
    if not isinstance(payload, dict):
        return "", "", "", ""
    return (
        str(payload.get("status", "")).strip(),
        str(payload.get("source_kind", "")).strip(),
        str(payload.get("source_uri", "")).strip(),
        str(payload.get("source_revision", "")).strip(),
    )


def assert_active_pack_source_bound(
    envv: dict[str, str],
    pack_name: str,
    downstream_build_root: Path,
    pack_dir: Path,
) -> None:
    source_kind, source_uri = load_active_installed_pack_source_state(envv, pack_name)
    if not source_kind and not source_uri:
        return
    if source_kind == "local_path":
        expected_uri = str(pack_dir.resolve())
        actual_uri = str(Path(source_uri).expanduser().resolve()) if source_uri else ""
        if actual_uri == expected_uri:
            return
    elif source_kind == "release_repo":
        root = downstream_build_root.resolve()
        source_root = Path(
            str(envv.get("NEXUSCRM_RECOVERY_RELEASE_PACK_ROOT", "")).strip()
            or str(root)
        ).expanduser().resolve()
        resolved_pack_dir = pack_dir.resolve()
        try:
            pack_entry = resolved_pack_dir.relative_to(root).as_posix()
        except ValueError as exc:
            raise DeployError(
                f"active release pack source escapes the exact deploy checkout: name={pack_name}"
            ) from exc
        checkout_revision = exact_downstream_source_revision(root)
        expected_source, _receipt = runtime_pack_entry_release_source(
            root,
            pack_entry,
            checkout_revision,
            downstream_source_root=source_root,
        )
        expected_uri = expected_source.removeprefix("release_repo:")
        active_revision = load_active_installed_pack_source_revision(envv, pack_name)
        if source_uri == expected_uri and active_revision == checkout_revision:
            return
    else:
        expected_uri = "local_path or release_repo bound to the exact deploy checkout"
    active_revision = load_active_installed_pack_source_revision(envv, pack_name)
    raise DeployError(
        "active pack source is not bound to the exact deploy checkout before "
        "post-activation lifecycle sync: "
        f"name={pack_name} source_kind={source_kind or '<empty>'} "
        f"source_uri={source_uri or '<empty>'} "
        f"source_revision={active_revision or '<empty>'} "
        f"expected_source_uri={expected_uri}."
    )


L1_POST_ACTIVATION_JOBS_PATH = Path("deploy/post_activation_jobs.json")
L1_POST_ACTIVATION_RUN_TABLE = "_PingCAP_PostActivationJobRun"


def should_run_l1_post_activation_job(envv: dict[str, str], job: dict[str, object]) -> bool:
    credential_env = str(job.get("credentialEnv", "")).strip()
    if credential_env and not envv.get(credential_env, "").strip():
        return False
    return True


def load_active_flow_metadata_state(envv: dict[str, str], developer_name: str) -> dict[str, object]:
    escaped_name = developer_name.replace("\\", "\\\\").replace("'", "''")
    row = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT JSON_OBJECT(
                'definition_id', fd.Id,
                'active_version_id', COALESCE(fd.ActiveVersionId, ''),
                'latest_version_id', COALESCE(fd.LatestVersionId, ''),
                'owner_pack', COALESCE(fd.OwnerPack, ''),
                'is_customized', COALESCE(fd.IsCustomized, 0),
                'flow_status', COALESCE(f.Status, ''),
                'flow_source', COALESCE(f.Source, ''),
                'metadata_hex', COALESCE(HEX(CAST(f.Metadata AS CHAR)), '')
            )
            FROM FlowDefinition fd
            LEFT JOIN Flow f
              ON f.Id = fd.ActiveVersionId
             AND COALESCE(f.IsDeleted, 0) = 0
            WHERE fd.DeveloperName = '{escaped_name}'
              AND COALESCE(fd.IsDeleted, 0) = 0
            LIMIT 1;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    if not row:
        return {}
    payload = json.loads(row)
    if not isinstance(payload, dict):
        return {}
    metadata_hex = str(payload.pop("metadata_hex", "") or "")
    if metadata_hex:
        try:
            payload["metadata"] = bytes.fromhex(metadata_hex).decode("utf-8")
        except (ValueError, UnicodeDecodeError) as exc:
            raise DeployError(
                f"owner-directory lifecycle flow active metadata is not valid hex-encoded UTF-8: "
                f"developer_name={developer_name}"
            ) from exc
    else:
        payload["metadata"] = ""
    return payload


def assert_l1_post_activation_job_active_flow_matches_source(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    job: dict[str, object],
) -> None:
    source_verification = require_json_object(
        job.get("sourceVerification", {}),
        f"post-activation job {job_id(job)} sourceVerification",
    )
    if source_verification.get("type") != "activeFlowMatchesPackSource":
        return
    developer_name = str(source_verification.get("flowApiName", "")).strip()
    relative_path = str(source_verification.get("path", "")).strip()
    if not developer_name or not relative_path:
        raise DeployError(f"post-activation job {job_id(job)} has incomplete active Flow source verification")
    source_path = pack_dir / relative_path
    if not source_path.is_file():
        raise DeployError(
            f"post-activation job flow source is missing for {pack_name}: "
            f"job={job_id(job)} developer_name={developer_name} path={source_path}"
        )
    source_metadata = json.loads(source_path.read_text())
    active_state = load_active_flow_metadata_state(envv, developer_name)
    if not active_state:
        raise DeployError(
            f"post-activation job flow is not active in runtime DB before execution: "
            f"pack={pack_name} job={job_id(job)} developer_name={developer_name}. "
            "Run the deploy reapply path before executing lifecycle jobs."
        )
    if str(active_state.get("flow_status", "")).strip() != "Active":
        raise DeployError(
            f"post-activation job flow active version is not Active before execution: "
            f"pack={pack_name} job={job_id(job)} developer_name={developer_name} "
            f"active_version_id={active_state.get('active_version_id', '')} "
            f"flow_status={active_state.get('flow_status', '')}."
        )
    active_metadata_raw = str(active_state.get("metadata", "")).strip()
    if not active_metadata_raw:
        raise DeployError(
            f"post-activation job flow active metadata is empty before execution: "
            f"pack={pack_name} job={job_id(job)} developer_name={developer_name} "
            f"active_version_id={active_state.get('active_version_id', '')}."
        )
    active_metadata = json.loads(active_metadata_raw)
    if active_metadata != source_metadata:
        raise DeployError(
            f"post-activation job flow active metadata is stale before execution: "
            f"pack={pack_name} job={job_id(job)} developer_name={developer_name} "
            f"active_version_id={active_state.get('active_version_id', '')} "
            f"latest_version_id={active_state.get('latest_version_id', '')} "
            f"source_path={source_path}. "
            "Run reapply-pack from the current DOWNSTREAM_BUILD_ROOT and verify "
            "FlowDefinition.ActiveVersionId points at the current pack metadata before "
            "executing lifecycle jobs."
        )


def assert_owner_directory_lifecycle_active_flow_matches_source(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
) -> None:
    for job in l1_post_activation_jobs_for_pack(pack_name, pack_dir, phase=None):
        assert_l1_post_activation_job_active_flow_matches_source(envv, pack_name, pack_dir, job)


def job_id(job: dict[str, object]) -> str:
    return str(job.get("id", "")).strip()


def l1_post_activation_jobs_for_pack(
    pack_name: str,
    pack_dir: Path,
    *,
    phase: str | None,
) -> list[dict[str, object]]:
    manifest_path = pack_dir / L1_POST_ACTIVATION_JOBS_PATH
    if not manifest_path.is_file():
        return []
    manifest = require_json_object(json.loads(manifest_path.read_text()), str(manifest_path))
    if manifest.get("schemaVersion") != 1:
        raise DeployError(f"unsupported L1 post-activation job schema in {manifest_path}")
    raw_jobs = manifest.get("jobs")
    if not isinstance(raw_jobs, list):
        raise DeployError(f"invalid L1 post-activation job manifest {manifest_path}: jobs must be an array")
    jobs: list[dict[str, object]] = []
    seen_ids: set[str] = set()
    for raw_job in raw_jobs:
        job = require_json_object(raw_job, f"L1 post-activation job in {manifest_path}")
        current_job_id = job_id(job)
        if not current_job_id:
            raise DeployError(f"invalid L1 post-activation job in {manifest_path}: id is required")
        if current_job_id in seen_ids:
            raise DeployError(f"duplicate L1 post-activation job id in {manifest_path}: {current_job_id}")
        seen_ids.add(current_job_id)
        job_phase = str(job.get("phase", "")).strip()
        if not job_phase:
            raise DeployError(f"invalid L1 post-activation job {current_job_id}: phase is required")
        if phase is not None and job_phase != phase:
            continue
        action = require_json_object(job.get("action", {}), f"L1 post-activation job {current_job_id} action")
        action_type = str(action.get("type", "")).strip()
        if action_type not in {
            "salesforceFlow",
            "hubspotOwnerDirectoryReconcile",
        }:
            raise DeployError(
                "unsupported L1 post-activation job action for "
                f"{current_job_id}: {action_type}"
            )
        if not str(action.get("flowApiName", "")).strip():
            raise DeployError(f"invalid L1 post-activation job {current_job_id}: action.flowApiName is required")
        if (
            action_type == "hubspotOwnerDirectoryReconcile"
            and str(action.get("flowApiName", "")).strip()
            != "HubSpot_Owner_Directory_Sync"
        ):
            raise DeployError(
                "hubspotOwnerDirectoryReconcile is restricted to "
                f"HubSpot_Owner_Directory_Sync: {current_job_id}"
            )
        job = dict(job)
        job["_packName"] = pack_name
        job["_packDir"] = str(pack_dir)
        jobs.append(job)
    return jobs


def l1_post_activation_jobs_for_runtime_packs(
    envv: dict[str, str],
    downstream_build_root: Path,
    *,
    phase: str | None,
) -> list[dict[str, object]]:
    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    jobs: list[dict[str, object]] = []
    for pack_entry in runtime_deploy_pack_entries(stack_lock):
        pack_name, _, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        if pack_dir is None:
            continue
        if installed_pack_status(envv, pack_name) != "active":
            continue
        for job in l1_post_activation_jobs_for_pack(pack_name, pack_dir, phase=phase):
            if should_run_l1_post_activation_job(envv, job):
                jobs.append(job)
    return jobs


def source_fingerprint_for_l1_post_activation_job(job: dict[str, object]) -> str:
    pack_dir = Path(str(job["_packDir"]))
    action = require_json_object(job.get("action", {}), f"post-activation job {job_id(job)} action")
    source_verification = require_json_object(
        job.get("sourceVerification", {}),
        f"post-activation job {job_id(job)} sourceVerification",
    )
    source_path = pack_dir / str(source_verification.get("path", "")).strip()
    source_payload = json.loads(source_path.read_text()) if source_path.is_file() else {}
    payload = {
        "jobId": job_id(job),
        "phase": str(job.get("phase", "")).strip(),
        "packName": str(job.get("_packName", "")).strip(),
        "actionType": str(action.get("type", "")).strip(),
        "flowApiName": str(action.get("flowApiName", "")).strip(),
        "inputs": action.get("inputs", []),
        "missingFlowAction": str(action.get("missingFlowAction", "")).strip(),
        "sourceVerification": {
            "type": str(source_verification.get("type", "")).strip(),
            "flowApiName": str(source_verification.get("flowApiName", "")).strip(),
            "path": str(source_verification.get("path", "")).strip(),
        },
        "sourcePayload": source_payload,
    }
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def l1_post_activation_job_state_key(job: dict[str, object]) -> str:
    idempotency = require_json_object(job.get("idempotency", {}), f"post-activation job {job_id(job)} idempotency")
    key = str(idempotency.get("key") or job_id(job)).strip()
    return f"{job.get('_packName', '')}:{job.get('phase', '')}:{key}"


def ensure_l1_post_activation_job_run_table(envv: dict[str, str]) -> None:
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            CREATE TABLE IF NOT EXISTS {L1_POST_ACTIVATION_RUN_TABLE} (
                JobKey VARCHAR(191) NOT NULL PRIMARY KEY,
                JobId VARCHAR(191) NOT NULL,
                PackName VARCHAR(191) NOT NULL,
                Phase VARCHAR(64) NOT NULL,
                SourceFingerprint CHAR(64) NOT NULL,
                LastRunAt TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                LastModifiedDate TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
            );
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def l1_post_activation_job_already_ran(envv: dict[str, str], job: dict[str, object], source_fingerprint: str) -> bool:
    idempotency = require_json_object(job.get("idempotency", {}), f"post-activation job {job_id(job)} idempotency")
    if idempotency.get("mode") != "sourceFingerprint":
        return False
    ensure_l1_post_activation_job_run_table(envv)
    row = db_query_scalar(
        envv,
        (
            f"SELECT SourceFingerprint FROM {L1_POST_ACTIVATION_RUN_TABLE} "
            f"WHERE JobKey={sql_string_literal(l1_post_activation_job_state_key(job))} LIMIT 1;"
        ),
        database=envv["TIDB_DATABASE"],
    ).strip()
    return row == source_fingerprint


def record_l1_post_activation_job_run(envv: dict[str, str], job: dict[str, object], source_fingerprint: str) -> None:
    idempotency = require_json_object(job.get("idempotency", {}), f"post-activation job {job_id(job)} idempotency")
    if idempotency.get("mode") != "sourceFingerprint":
        return
    ensure_l1_post_activation_job_run_table(envv)
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            INSERT INTO {L1_POST_ACTIVATION_RUN_TABLE}
                (JobKey, JobId, PackName, Phase, SourceFingerprint, LastRunAt, LastModifiedDate)
            VALUES
                ({sql_string_literal(l1_post_activation_job_state_key(job))},
                 {sql_string_literal(job_id(job))},
                 {sql_string_literal(str(job.get("_packName", "")).strip())},
                 {sql_string_literal(str(job.get("phase", "")).strip())},
                 {sql_string_literal(source_fingerprint)},
                 CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            ON DUPLICATE KEY UPDATE
                JobId=VALUES(JobId),
                PackName=VALUES(PackName),
                Phase=VALUES(Phase),
                SourceFingerprint=VALUES(SourceFingerprint),
                LastRunAt=VALUES(LastRunAt),
                LastModifiedDate=VALUES(LastModifiedDate);
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def assert_owner_directory_lifecycle_active_flows_for_runtime_packs(
    envv: dict[str, str],
    downstream_build_root: Path,
) -> None:
    for job in l1_post_activation_jobs_for_runtime_packs(envv, downstream_build_root, phase=None):
        pack_name = str(job["_packName"])
        pack_dir = Path(str(job["_packDir"]))
        assert_active_pack_source_bound(
            envv, pack_name, downstream_build_root, pack_dir,
        )
        assert_l1_post_activation_job_active_flow_matches_source(envv, pack_name, pack_dir, job)


def apply_stack_pack_named_credential_runbooks(
    envv: dict[str, str],
    downstream_build_root: Path,
    app_packs_root: Path,
    pack_entries: list[str],
) -> None:
    for pack_entry in pack_entries:
        pack_name, source_dir = runtime_pack_entry_source_dir(downstream_build_root, app_packs_root, pack_entry)
        sql_file = source_dir / "runbooks" / "provision_named_credential.sql"
        if not sql_file.is_file():
            continue
        log(f"apply named credential provisioning runbook: {pack_name}")
        for statement in split_sql_statements(sql_file.read_text()):
            db_exec(envv, statement, database=envv["TIDB_DATABASE"])


def named_credential_secret_masked_hint(plaintext: str) -> str:
    if not plaintext:
        return ""
    if len(plaintext) <= 4:
        return "\u2026"
    return "\u2026" + plaintext[-4:]


def encrypt_named_credential_secret(plaintext: str, master_key_b64: str) -> str:
    script = r"""
const crypto = require("crypto");
const fs = require("fs");
const input = JSON.parse(fs.readFileSync(0, "utf8"));
const key = Buffer.from(input.masterKey, "base64");
if (key.length !== 32) {
  throw new Error(`NEXUSCRM_NC_MASTER_KEY must decode to 32 bytes, got ${key.length}`);
}
const nonce = crypto.randomBytes(12);
const cipher = crypto.createCipheriv("aes-256-gcm", key, nonce);
const encrypted = Buffer.concat([cipher.update(input.plaintext, "utf8"), cipher.final()]);
const tag = cipher.getAuthTag();
process.stdout.write(Buffer.concat([Buffer.from([1]), nonce, encrypted, tag]).toString("base64"));
"""
    return run(
        ["node", "-e", script],
        capture=True,
        stdin=json.dumps({"masterKey": master_key_b64, "plaintext": plaintext}),
    ).strip()


def decrypt_named_credential_secret(ciphertext_b64: str, master_key_b64: str) -> str:
    script = r"""
const crypto = require("crypto");
const fs = require("fs");
try {
const input = JSON.parse(fs.readFileSync(0, "utf8"));
const key = Buffer.from(input.masterKey, "base64");
if (key.length !== 32) {
  throw new Error(`NEXUSCRM_NC_MASTER_KEY must decode to 32 bytes, got ${key.length}`);
}
const payload = Buffer.from(input.ciphertext, "base64");
if (payload.length < 30 || payload[0] !== 1) {
  throw new Error("unsupported named credential ciphertext");
}
const nonce = payload.subarray(1, 13);
const encrypted = payload.subarray(13, payload.length - 16);
const tag = payload.subarray(payload.length - 16);
const decipher = crypto.createDecipheriv("aes-256-gcm", key, nonce);
decipher.setAuthTag(tag);
const plaintext = Buffer.concat([decipher.update(encrypted), decipher.final()]);
process.stdout.write(plaintext.toString("utf8"));
} catch {
  process.stderr.write("named credential decryption failed\n");
  process.exit(1);
}
"""
    return run(
        ["node", "-e", script],
        capture=True,
        stdin=json.dumps(
            {
                "masterKey": master_key_b64,
                "ciphertext": ciphertext_b64,
            }
        ),
    ).strip()


def provision_hubspot_private_app_named_credential_token(
    envv: dict[str, str],
    *,
    query=db_query_scalar,
    execute=db_exec,
    encrypt=encrypt_named_credential_secret,
) -> None:
    access_token = envv.get("EXP_HUBSPOT_ACCESS_TOKEN", "").strip()
    if not access_token:
        return
    master_key = envv.get("NEXUSCRM_NC_MASTER_KEY", "").strip()
    if not master_key:
        raise DeployError("cannot provision HubSpot NamedCredential token: missing NEXUSCRM_NC_MASTER_KEY")
    credential_id = query(
        envv,
        "SELECT Id FROM _System_NamedCredential "
        "WHERE DeveloperName='HubSpotPrivateApp' "
        "AND OwnerPack='hubspot-crm-connector' LIMIT 2;",
        database=envv["TIDB_DATABASE"],
    )
    if not credential_id or "\n" in credential_id:
        raise DeployError(
            "cannot provision HubSpot NamedCredential token: exactly one "
            "hubspot-crm-connector/HubSpotPrivateApp definition is required"
        )
    secret_id = "nc_secret_hubspot_crm_connector_private_app_token"
    reserved_binding = query(
        envv,
        "SELECT CONCAT_WS('\\t', NamedCredentialId, SecretKind) "
        "FROM _System_NamedCredentialSecret "
        f"WHERE Id={sql_string_literal(secret_id)} LIMIT 1;",
        database=envv["TIDB_DATABASE"],
    )
    if reserved_binding and reserved_binding != f"{credential_id}\tToken":
        raise DeployError(
            "cannot provision HubSpot NamedCredential token: the reserved secret ID "
            "is bound to another credential"
        )
    ciphertext = encrypt(access_token, master_key)
    masked_hint = named_credential_secret_masked_hint(access_token)
    execute(
        envv,
        textwrap.dedent(
            f"""
            INSERT INTO _System_NamedCredentialSecret
                (Id, NamedCredentialId, SecretKind, SecretValue, MaskedHint,
                 RotatedAt, CreatedDate, LastModifiedDate)
            VALUES
                ({sql_string_literal(secret_id)},
                 {sql_string_literal(credential_id)}, 'Token',
                 {sql_string_literal(ciphertext)}, {sql_string_literal(masked_hint)},
                 CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            ON DUPLICATE KEY UPDATE
                NamedCredentialId = VALUES(NamedCredentialId),
                SecretKind = VALUES(SecretKind),
                SecretValue = VALUES(SecretValue),
                MaskedHint = VALUES(MaskedHint),
                RotatedAt = VALUES(RotatedAt),
                LastModifiedDate = VALUES(LastModifiedDate);
            """
        ),
        database=envv["TIDB_DATABASE"],
    )
    log("provisioned HubSpot NamedCredential token secret")


def provision_hubspot_webhook_secret_values(
    envv: dict[str, str],
    client_secret: str,
    *,
    query=db_query_scalar,
    execute=db_exec,
    encrypt=encrypt_named_credential_secret,
    decrypt=decrypt_named_credential_secret,
) -> int:
    """Atomically install and verify the HubSpot connector signing secrets."""
    client_secret = client_secret.strip()
    if len(client_secret) < 16 or any(char in client_secret for char in "\n\r\0"):
        raise DeployError("HubSpot webhook client secret must contain at least 16 valid characters")
    database = envv.get("TIDB_DATABASE", "").strip()
    master_key = envv.get("NEXUSCRM_NC_MASTER_KEY", "").strip()
    if not database or not master_key:
        raise DeployError(
            "HubSpot webhook secret provisioning requires TIDB_DATABASE and NEXUSCRM_NC_MASTER_KEY"
        )

    definitions: dict[str, str] = {}
    for secret_name in HUBSPOT_WEBHOOK_SECRET_NAMES:
        definition_filter = (
            "FROM _System_WebhookSecret "
            "WHERE OwnerPack = 'hubspot-crm-connector' "
            f"AND Name = {sql_string_literal(secret_name)} "
            "AND COALESCE(IsDeleted, 0) = 0"
        )
        definition_count = query(
            envv,
            f"SELECT COUNT(*) {definition_filter};",
            database=database,
        )
        if definition_count != "1":
            raise DeployError(
                "HubSpot webhook secret provisioning requires exactly one "
                f"hubspot-crm-connector/{secret_name} definition"
            )
        secret_id = query(
            envv,
            f"SELECT Id {definition_filter} LIMIT 1;",
            database=database,
        )
        if not secret_id or "\n" in secret_id:
            raise DeployError(
                "HubSpot webhook secret definition has no unique Id: "
                f"hubspot-crm-connector/{secret_name}"
            )
        definitions[secret_name] = secret_id

    transaction_statements = ["START TRANSACTION;"]
    for secret_name, secret_id in definitions.items():
        value_id = (
            "webhook_secret_value_hubspot_crm_connector_"
            + secret_name.replace("-", "_")
        )
        reserved_binding = query(
            envv,
            "SELECT WebhookSecretId FROM _System_WebhookSecretValue "
            f"WHERE Id={sql_string_literal(value_id)} LIMIT 1;",
            database=database,
        )
        if reserved_binding and reserved_binding != secret_id:
            raise DeployError(
                "cannot provision HubSpot webhook secret: the reserved value ID "
                f"for {secret_name} is bound to another definition"
            )
        ciphertext = encrypt(client_secret, master_key)
        transaction_statements.append(
            "UPDATE _System_WebhookSecretValue "
            "SET IsActive = 0, LastModifiedDate = CURRENT_TIMESTAMP "
            f"WHERE WebhookSecretId = {sql_string_literal(secret_id)};"
        )
        transaction_statements.append(
            textwrap.dedent(
                f"""
                INSERT INTO _System_WebhookSecretValue
                  (Id, WebhookSecretId, SecretValue, IsActive, RotatedAt,
                   CreatedDate, LastModifiedDate)
                VALUES
                  ({sql_string_literal(value_id)},
                   {sql_string_literal(secret_id)},
                   {sql_string_literal(ciphertext)}, 1, CURRENT_TIMESTAMP,
                   CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                ON DUPLICATE KEY UPDATE
                  WebhookSecretId = VALUES(WebhookSecretId),
                  SecretValue = VALUES(SecretValue),
                  IsActive = 1,
                  RotatedAt = VALUES(RotatedAt),
                  LastModifiedDate = VALUES(LastModifiedDate);
                """
            ).strip()
        )
    transaction_statements.append("COMMIT;")
    execute(envv, "\n".join(transaction_statements), database=database)

    verified = 0
    for secret_name, secret_id in definitions.items():
        active_value_filter = (
            "FROM _System_WebhookSecretValue "
            f"WHERE WebhookSecretId = {sql_string_literal(secret_id)} "
            "AND IsActive = 1"
        )
        active_value_count = query(
            envv,
            f"SELECT COUNT(*) {active_value_filter};",
            database=database,
        )
        if active_value_count != "1":
            raise DeployError(
                "HubSpot webhook secret readback requires exactly one active value: "
                f"hubspot-crm-connector/{secret_name}"
            )
        stored_ciphertext = query(
            envv,
            f"SELECT SecretValue {active_value_filter};",
            database=database,
        )
        if not stored_ciphertext:
            raise DeployError(
                "HubSpot webhook secret readback is empty: "
                f"hubspot-crm-connector/{secret_name}"
            )
        try:
            plaintext = decrypt(stored_ciphertext, master_key)
        except DeployError as exc:
            raise DeployError(
                "HubSpot webhook secret readback cannot be decrypted: "
                f"hubspot-crm-connector/{secret_name}"
            ) from exc
        if not hmac.compare_digest(
            plaintext.encode("utf-8"), client_secret.encode("utf-8")
        ):
            raise DeployError(
                "HubSpot webhook secret readback differs from the admitted value: "
                f"hubspot-crm-connector/{secret_name}"
            )
        verified += 1
    log(f"HubSpot webhook secrets provisioned and verified: count={verified}")
    return verified


def reapply_active_local_stack_packs(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    pack_entries: list[str],
    *,
    final_pack_entry: str = "packs/NOT-EDIT-final-merged-packs",
) -> None:
    ordered_entries = [entry for entry in pack_entries if entry != final_pack_entry]
    if final_pack_entry in pack_entries:
        ordered_entries.append(final_pack_entry)
    for pack_entry in ordered_entries:
        pack_name, source, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        if installed_pack_status(envv, pack_name) != "active":
            continue
        log(f"reapplying active stack pack before runtime pack install checks: {pack_name}")
        run_backend_setup(
            envv,
            nexuscrm_worktree,
            app_packs_root,
            active_stack_pack_reapply_args(envv, pack_name, source, pack_dir, app_packs_root),
        )


def active_stack_pack_reapply_args(
    envv: dict[str, str],
    pack_name: str,
    source: str,
    pack_dir: Path | None,
    app_packs_root: Path,
    *,
    verify_pack: bool = True,
) -> list[str]:
    args = ["reapply-pack", "--source", source]
    requires_verify = verify_pack
    if pack_dir is None:
        from_revision = load_active_installed_pack_source_revision(envv, pack_name)
        to_revision = git_head_revision(app_packs_root)
        if from_revision and to_revision and from_revision != to_revision:
            args.extend(
                [
                    "--allow-source-revision-advance",
                    "--from-revision",
                    from_revision,
                    "--to-revision",
                    to_revision,
                ]
            )
            requires_verify = True
    if requires_verify:
        args.append("--verify-pack")
    return args


def ensure_runtime_packs_installed(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    *,
    allow_preinstalled_active_packs: bool = True,
) -> None:
    # Deployment owns the empty-db bootstrap contract: materialize the built
    # final pack into installed-pack state before full setup apply runs.
    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    pack_entries = runtime_deploy_pack_entries(stack_lock)
    final_pack_entry = runtime_final_pack_entry(stack_lock)
    log("setup apply-metadata")
    run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["apply-metadata"])
    if allow_preinstalled_active_packs:
        rebind_active_stack_pack_sources(envv, downstream_build_root, app_packs_root, pack_entries)
    log("setup repair")
    run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["repair"])
    log("setup apply-packs initial")
    run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["apply-packs"])
    if allow_preinstalled_active_packs:
        reapply_active_local_stack_packs(
            envv,
            downstream_build_root,
            nexuscrm_worktree,
            app_packs_root,
            pack_entries,
            final_pack_entry=final_pack_entry,
        )
        log("synchronizing active local pack runtime metadata before runtime pack install checks")
        sync_stack_local_pack_metadata(envv, downstream_build_root, nexuscrm_worktree, pack_entries)
        assert_owner_directory_lifecycle_active_flows_for_runtime_packs(envv, downstream_build_root)

    for pack_entry in pack_entries:
        pack_name, source, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        log(f"validate-pack {pack_name}")
        run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["validate-pack", "--source", source])
        status = installed_pack_status(envv, pack_name)
        if status == "active":
            if not allow_preinstalled_active_packs:
                raise DeployError(
                    f"preinstalled active pack is not supported for this runtime bootstrap flow: {pack_name}"
                )
            log(f"pack already active, skipping install: {pack_name}")
            apply_stack_pack_named_credential_runbooks(envv, downstream_build_root, app_packs_root, [pack_entry])
            if pack_name == "hubspot-crm-connector":
                provision_hubspot_private_app_named_credential_token(envv)
            continue
        if status in {"installing", "failed"}:
            if not allow_preinstalled_active_packs:
                raise DeployError(
                    f"preinstalled {status} pack is not supported for this runtime bootstrap flow: {pack_name}"
                )
            if pack_dir is not None:
                log(f"pack has {status} runtime state, rebind source identity before install retry: {pack_name}")
                rebind_retryable_installed_pack_source_identity(envv, pack_name, pack_dir, nexuscrm_worktree)
        log(f"install-pack {pack_name}")
        run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["install-pack", "--source", source])
        apply_stack_pack_named_credential_runbooks(envv, downstream_build_root, app_packs_root, [pack_entry])
        if pack_name == "hubspot-crm-connector":
            provision_hubspot_private_app_named_credential_token(envv)
    assert_owner_directory_lifecycle_active_flows_for_runtime_packs(envv, downstream_build_root)


def sql_string_literal(value: str) -> str:
    return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'"


def sql_nullable_string_literal(value: str | None, valid: bool) -> str:
    if not valid or value is None:
        return "NULL"
    return sql_string_literal(value)


def installed_pack_dependency_row_id(dependent_pack_name: str, depends_on_pack_name: str) -> str:
    digest = hashlib.sha256(
        (dependent_pack_name.strip() + "\x00" + depends_on_pack_name.strip()).encode("utf-8")
    ).hexdigest()
    return "ipd_" + digest


def _runtime_pack_declared_dependencies(pack_manifest: dict[str, object]) -> list[tuple[str, str | None]]:
    dependencies = pack_manifest.get("dependencies")
    if not isinstance(dependencies, list):
        return []
    declared: list[tuple[str, str | None]] = []
    seen: set[str] = set()
    for dep in dependencies:
        dep_name = ""
        version_constraint: str | None = None
        if isinstance(dep, str):
            dep_name = dep.strip()
        elif isinstance(dep, dict):
            raw_name = dep.get("name")
            dep_name = raw_name.strip() if isinstance(raw_name, str) else ""
            raw_constraint = dep.get("versionConstraint")
            if isinstance(raw_constraint, str) and raw_constraint.strip():
                version_constraint = raw_constraint.strip()
        if not dep_name or dep_name in seen:
            continue
        seen.add(dep_name)
        declared.append((dep_name, version_constraint))
    return declared


def sync_runtime_pack_dependency_declarations(envv: dict[str, str], downstream_build_root: Path) -> None:
    if not tidb_table_exists(envv, "_System_InstalledPackDependency"):
        log("skipping runtime pack dependency sync: _System_InstalledPackDependency table missing")
        return
    if not installed_pack_table_exists(envv):
        log("skipping runtime pack dependency sync: _System_InstalledPack table missing")
        return

    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    for pack_path in runtime_deploy_pack_paths(stack_lock):
        pack_dir = downstream_build_root / pack_path
        pack_json = pack_dir / "pack.json"
        if not pack_json.is_file():
            continue
        pack_manifest = json.loads(pack_json.read_text())
        pack_name_raw = pack_manifest.get("name")
        if not isinstance(pack_name_raw, str) or not pack_name_raw.strip():
            continue
        pack_name = pack_name_raw.strip()
        declared_deps = _runtime_pack_declared_dependencies(pack_manifest)

        pack_install_id = db_query_scalar(
            envv,
            textwrap.dedent(
                f"""
                SELECT COALESCE(PackInstallId, '')
                  FROM _System_InstalledPack
                 WHERE Name={sql_string_literal(pack_name)}
                   AND Status IN ('active', 'installing', 'failed')
                   AND COALESCE(IsDeleted, 0) = 0
                 LIMIT 1;
                """
            ).strip(),
            database=envv["TIDB_DATABASE"],
        ).strip()
        if not pack_install_id:
            raise DeployError(f"sync runtime pack dependencies for {pack_name}: active installed pack row missing")

        if declared_deps:
            values = []
            for dep_name, version_constraint in declared_deps:
                row_id = installed_pack_dependency_row_id(pack_name, dep_name)
                values.append(
                    "("
                    + ", ".join(
                        [
                            sql_string_literal(row_id),
                            sql_string_literal(pack_name),
                            sql_string_literal(pack_install_id),
                            sql_string_literal(dep_name),
                            sql_nullable_string_literal(version_constraint, bool(version_constraint)),
                            "0",
                            "UTC_TIMESTAMP()",
                            "UTC_TIMESTAMP()",
                        ]
                    )
                    + ")"
                )
            db_exec(
                envv,
                textwrap.dedent(
                    f"""
                    INSERT INTO _System_InstalledPackDependency
                      (Id, DependentPackName, DependentPackInstallId, DependsOnPackName, VersionConstraint, IsDeleted, CreatedDate, LastModifiedDate)
                    VALUES
                      {", ".join(values)}
                    ON DUPLICATE KEY UPDATE
                      DependentPackInstallId = VALUES(DependentPackInstallId),
                      VersionConstraint      = VALUES(VersionConstraint),
                      IsDeleted              = 0,
                      LastModifiedDate       = VALUES(LastModifiedDate);
                    """
                ).strip(),
                database=envv["TIDB_DATABASE"],
            )
            dep_list = ", ".join(sql_string_literal(dep_name) for dep_name, _ in declared_deps)
            stale_predicate = f"AND DependsOnPackName NOT IN ({dep_list})"
        else:
            stale_predicate = ""

        db_exec(
            envv,
            textwrap.dedent(
                f"""
                UPDATE _System_InstalledPackDependency
                   SET IsDeleted = 1,
                       LastModifiedDate = UTC_TIMESTAMP()
                 WHERE DependentPackName = {sql_string_literal(pack_name)}
                   AND COALESCE(IsDeleted, 0) = 0
                   {stale_predicate};
                """
            ).strip(),
            database=envv["TIDB_DATABASE"],
        )
        log(f"synced runtime pack dependency declarations: {pack_name} deps={len(declared_deps)}")


def connector_inbound_nonterminal_summary(envv: dict[str, str], pack_name_predicate: str = "") -> str:
    if not installed_pack_table_exists(envv) or not tidb_table_exists(envv, "_System_ConnectorEvent"):
        return ""
    return db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT COALESCE(
                GROUP_CONCAT(
                    CONCAT(pack_name, '[', pack_category, '/', status, ']=', row_count)
                    ORDER BY pack_name, pack_category, status
                    SEPARATOR '; '
                ),
                ''
            )
            FROM (
                SELECT p.Name AS pack_name,
                       p.Category AS pack_category,
                       e.Status AS status,
                       COUNT(*) AS row_count
                  FROM _System_InstalledPack p
                  JOIN _System_ConnectorEvent e
                    ON (
                        (p.Category = 'connector-pack' AND e.InstalledPackId = p.PackInstallId)
                        OR (p.Category = 'app-pack' AND COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId)
                    )
                 WHERE 1 = 1
                   {pack_name_predicate}
                   AND p.Status IN ('active', 'installing', 'removing', 'failed')
                   AND COALESCE(p.IsDeleted, 0) = 0
                   AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
                   AND e.Status IN ('pending', 'failed', 'processing')
                 GROUP BY p.Name, p.Category, e.Status
            ) pending_connector_events;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()


def connector_inbound_nonterminal_summary_for_installed_packs(envv: dict[str, str]) -> str:
    return connector_inbound_nonterminal_summary(envv)


def connector_inbound_nonterminal_summary_for_packs(envv: dict[str, str], pack_names: Iterable[str]) -> str:
    names = sorted({str(name).strip() for name in pack_names if str(name).strip()})
    if not names:
        return ""
    if not installed_pack_table_exists(envv) or not tidb_table_exists(envv, "_System_ConnectorEvent"):
        return ""
    pack_name_list = ", ".join(sql_string_literal(name) for name in names)
    if not tidb_table_exists(envv, "_System_ConnectorWorkItem"):
        return connector_inbound_legacy_event_summary_for_packs(envv, pack_name_list)

    # WorkItem owns executable queue lifecycle once the durable queue table is
    # present. Settled source projections intentionally retain older Event facts
    # without one WorkItem per Event, so Event.Status cannot be the drain SSoT.
    work_item_summary = connector_inbound_workitem_summary_for_packs(envv, pack_name_list)
    if work_item_summary:
        return work_item_summary
    return connector_inbound_legacy_event_gap_summary_for_packs(envv, pack_name_list)


def connector_inbound_legacy_event_summary_for_packs(envv: dict[str, str], pack_name_list: str) -> str:
    return db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT COALESCE(
                GROUP_CONCAT(
                    CONCAT(pack_name, '[', pack_category, '/', status, ']=', row_count)
                    ORDER BY pack_name, pack_category, status
                    SEPARATOR '; '
                ),
                ''
            )
            FROM (
                SELECT p.Name AS pack_name,
                       p.Category AS pack_category,
                       e.Status AS status,
                       COUNT(*) AS row_count
                  FROM _System_InstalledPack p
                  JOIN _System_ConnectorEvent e
                    ON (
                        (p.Category = 'connector-pack' AND e.InstalledPackId = p.PackInstallId)
                        OR (p.Category = 'app-pack' AND COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId)
                    )
                 WHERE p.Name IN ({pack_name_list})
                   AND p.Status IN ('active', 'installing', 'removing', 'failed')
                   AND COALESCE(p.IsDeleted, 0) = 0
                   AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
                   AND e.Status IN ('pending', 'failed', 'processing')
                 GROUP BY p.Name, p.Category, e.Status
            ) pending_connector_events;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()


def connector_inbound_workitem_summary_for_packs(envv: dict[str, str], pack_name_list: str) -> str:
    # Prove the global executable queue is empty before the pack-scoped join.
    # The caller still checks uncovered legacy Events after an empty summary.
    pending = db_query_scalar(
        envv,
        textwrap.dedent(
            """
            SELECT EXISTS (
                SELECT 1
                  FROM _System_ConnectorWorkItem w
                 WHERE w.WorkType IN ('connector_event', 'workflow_action_apply')
                   AND COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')
                 LIMIT 1
            );
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()
    if pending == "0":
        return ""
    if pending != "1":
        raise DeployError("connector WorkItem existence query returned an invalid result")
    return db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT COALESCE(
                GROUP_CONCAT(
                    CONCAT(
                        pack_name,
                        '[',
                        pack_category,
                        '/work_item_',
                        work_item_state,
                        ']=',
                        row_count
                    )
                    ORDER BY pack_name, pack_category, work_item_state
                    SEPARATOR '; '
                ),
                ''
            )
            FROM (
                SELECT p.Name AS pack_name,
                       p.Category AS pack_category,
                       COALESCE(NULLIF(TRIM(w.State), ''), 'unknown') AS work_item_state,
                       COUNT(DISTINCT w.WorkId) AS row_count
                  FROM _System_ConnectorWorkItem w
                  LEFT JOIN _System_ConnectorEvent e
                    ON e.Id = w.EventId
                  JOIN _System_InstalledPack p
                    ON (
                        (p.Category = 'connector-pack' AND w.InstalledPackId = p.PackInstallId)
                        OR (p.Category = 'app-pack' AND COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId)
                    )
                 WHERE p.Name IN ({pack_name_list})
                   AND p.Status IN ('active', 'installing', 'removing', 'failed')
                   AND COALESCE(p.IsDeleted, 0) = 0
                   AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
                   AND w.WorkType IN ('connector_event', 'workflow_action_apply')
                   AND COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')
                 GROUP BY p.Name,
                          p.Category,
                          COALESCE(NULLIF(TRIM(w.State), ''), 'unknown')
            ) pending_connector_work_items;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()


def connector_workitem_only_execution_attested(envv: dict[str, str]) -> bool:
    """Read the immutable Connectors WorkItem-only execution handoff receipt.

    This chooses the queue owner, not whether historical Events were correctly
    processed. The canonical Controlplane apply/recovery path still runs
    VerifyConnectorExecutionProtocols before activation, including its stronger
    historical ownership checks. Do not replace that gate with this receipt.
    """
    if not tidb_table_exists(envv, "_System_Migration"):
        return False
    result = db_query_scalar(
        envv,
        textwrap.dedent(
            """
            SELECT CASE
                     WHEN Version = 2 AND BINARY Description = BINARY
                       'Atomically hand off connector execution to WorkItems after terminalizing Event-owned backlog while ingress is durably quiesced (NexusCRM #3980).'
                     THEN 'attested'
                     ELSE 'invalid'
                   END
              FROM _System_Migration
             WHERE Id = '20260724_verify_connector_workitem_only_execution_cutover_v2';
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()
    if not result:
        return False
    if result != "attested":
        raise DeployError("connector WorkItem-only execution handoff receipt is invalid")
    return True


def connector_inbound_legacy_event_gap_summary_for_packs(
    envv: dict[str, str], pack_name_list: str
) -> str:
    """Check legacy Event-owned work only before the one-way executor handoff.

    The caller has already checked executable WorkItems. After the attested
    handoff, Event.Status is admission history, not a second execution queue.
    Missing receipts retain the pre-cutover guard; malformed receipts fail
    closed. Historical data ownership remains the Controlplane release gate.
    """
    if connector_workitem_only_execution_attested(envv):
        return ""
    return db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT COALESCE(
                CONCAT(
                    pack_name,
                    '[',
                    pack_category,
                    '/legacy_event_without_work_item]=1+'
                ),
                ''
            )
            FROM (
                SELECT p.Name AS pack_name,
                       p.Category AS pack_category
                  FROM _System_InstalledPack p
                  JOIN _System_ConnectorEvent e
                    ON (
                        (p.Category = 'connector-pack' AND e.InstalledPackId = p.PackInstallId)
                        OR (p.Category = 'app-pack' AND COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId)
                    )
                  LEFT JOIN _System_ConnectorWorkItem direct_work
                    ON direct_work.EventId = e.Id
                   AND direct_work.WorkType IN ('connector_event', 'workflow_action_apply')
                 WHERE p.Name IN ({pack_name_list})
                   AND p.Status IN ('active', 'installing', 'removing', 'failed')
                   AND COALESCE(p.IsDeleted, 0) = 0
                   AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
                   AND e.Status NOT IN ('processed', 'skipped', 'dead')
                   AND direct_work.WorkId IS NULL
                   AND NOT EXISTS (
                       SELECT 1
                         FROM _System_ConnectorWorkItem projection
                         JOIN _System_ConnectorEvent selected
                           ON selected.Id = projection.EventId
                        WHERE projection.InstalledPackId = e.InstalledPackId
                          AND projection.WorkType IN ('connector_event', 'workflow_action_apply')
                          AND (
                              NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NOT NULL
                              OR (
                                  NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NULL
                                  AND projection.WorkId LIKE 'csp:%'
                                  AND projection.State NOT IN ('processed', 'skipped', 'dead')
                              )
                          )
                          AND COALESCE(projection.SourceLaneKey, '') = COALESCE(e.SourceLaneKey, '')
                          AND selected.InstalledPackId = e.InstalledPackId
                          AND LOWER(TRIM(selected.EventKind)) = LOWER(TRIM(e.EventKind))
                          AND COALESCE(selected.RouteVersion, 0) = COALESCE(e.RouteVersion, 0)
                          AND COALESCE(selected.CompanionPackInstallId, '') = COALESCE(e.CompanionPackInstallId, '')
                          AND COALESCE(selected.CompanionFlowVersionId, '') = COALESCE(e.CompanionFlowVersionId, '')
                          AND (
                              (
                                  e.EnqueueOrdinal <> 0
                                  AND selected.EnqueueOrdinal <> 0
                                  AND (
                                      e.EnqueueOrdinal < selected.EnqueueOrdinal
                                      OR (
                                          e.EnqueueOrdinal = selected.EnqueueOrdinal
                                          AND e.Id <= selected.Id
                                      )
                                  )
                              )
                              OR (
                                  (e.EnqueueOrdinal = 0 OR selected.EnqueueOrdinal = 0)
                                  AND (
                                      e.CreatedDate < selected.CreatedDate
                                      OR (
                                          e.CreatedDate = selected.CreatedDate
                                          AND (
                                              e.DeliverySequence < selected.DeliverySequence
                                              OR (
                                                  e.DeliverySequence = selected.DeliverySequence
                                                  AND e.Id <= selected.Id
                                              )
                                          )
                                      )
                                  )
                              )
                          )
                   )
                 ORDER BY p.Name, p.Category
                 LIMIT 1
            ) legacy_event_gap;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()


def reset_snapshot_connector_queue_nonterminal_counts(envv: dict[str, str]) -> dict[str, int]:
    """Return every inherited delivery row that must never run in a reset target.

    A production snapshot can contain an Event or WorkItem whose historic pack
    identity has already been deleted, is blank, or otherwise cannot join a
    current installed pack.  Those rows are precisely the ones a pack-scoped
    query misses, yet they must not become staging/test delivery work.
    """
    if not tidb_table_exists(envv, "_System_ConnectorEvent") or not tidb_table_exists(
        envv, "_System_ConnectorWorkItem"
    ):
        raise DeployError(
            "reset snapshot connector queue hygiene requires _System_ConnectorEvent "
            "and _System_ConnectorWorkItem"
        )
    event_count = int(
        db_query_scalar(
            envv,
            textwrap.dedent(
                """
                SELECT COUNT(*)
                  FROM _System_ConnectorEvent e
                 WHERE COALESCE(e.Status, '') NOT IN ('processed', 'skipped', 'dead');
                """
            ).strip(),
            database=envv["TIDB_DATABASE"],
        )
        or "0"
    )
    work_item_count = int(
        db_query_scalar(
            envv,
            textwrap.dedent(
                """
                SELECT COUNT(*)
                  FROM _System_ConnectorWorkItem w
                 WHERE COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead');
                """
            ).strip(),
            database=envv["TIDB_DATABASE"],
        )
        or "0"
    )
    return {
        "connector_events": event_count,
        "connector_work_items": work_item_count,
    }


def require_connector_delivery_queue_terminal_before_installed_pack_mutation(
    envv: dict[str, str],
    *,
    context: str,
) -> dict[str, int]:
    """Fail closed unless every connector delivery row is terminal.

    This is deliberately independent of `_System_InstalledPack`: a missing,
    deleted, or mismatched historic owner must block before a rebind can change
    pack identity. Callers that have an explicit reset-snapshot quarantine
    policy may use that fenced operation first; warm paths must only block.
    """

    counts = reset_snapshot_connector_queue_nonterminal_counts(envv)
    if any(counts.values()):
        raise DeployError(
            f"{context}: global nonterminal connector delivery state blocks installed-pack mutation: "
            f"connector_events={counts['connector_events']}, "
            f"connector_work_items={counts['connector_work_items']}"
        )
    return counts


RESET_SNAPSHOT_CONNECTOR_QUEUE_PURGE_PLAN_VERSION = 1
RESET_SNAPSHOT_CONNECTOR_QUEUE_PURGE_TABLES: tuple[tuple[str, str], ...] = (
    # WorkItems are the queue authority and may reference an Event. Clear the
    # dependent execution table first, then its immutable source projection.
    ("_System_ConnectorWorkItem", "connector_work_items"),
    ("_System_ConnectorEvent", "connector_events"),
)


def reset_snapshot_connector_queue_row_counts(envv: dict[str, str]) -> dict[str, int]:
    """Count all inherited queue rows selected by the reset-target purge plan."""
    counts: dict[str, int] = {}
    for table_name, count_key in RESET_SNAPSHOT_CONNECTOR_QUEUE_PURGE_TABLES:
        if not tidb_table_exists(envv, table_name):
            raise DeployError(f"reset snapshot connector queue purge requires {table_name}")
        raw_count = db_query_scalar(
            envv,
            f"SELECT COUNT(*) FROM `{table_name}`;",
            database=envv["TIDB_DATABASE"],
        )
        try:
            counts[count_key] = int(raw_count or "0")
        except ValueError as exc:
            raise DeployError(
                f"invalid reset snapshot connector queue row count for {table_name}: {raw_count!r}"
            ) from exc
    return counts


def require_reset_snapshot_connector_queue_truncate_safe(envv: dict[str, str]) -> None:
    """Fail closed when a purge table has an undeclared inbound foreign key.

    TiDB implements TRUNCATE as DDL.  The reset-target plan intentionally
    contains only two data-only queue tables, so an inbound FK would mean the
    checked-in dependency order is incomplete rather than something to bypass
    with a best-effort DELETE.
    """
    database = envv["TIDB_DATABASE"]
    for table_name, _ in RESET_SNAPSHOT_CONNECTOR_QUEUE_PURGE_TABLES:
        raw_count = db_query_scalar(
            envv,
            textwrap.dedent(
                f"""
                SELECT COUNT(*)
                  FROM information_schema.KEY_COLUMN_USAGE
                 WHERE REFERENCED_TABLE_SCHEMA = {sql_string_literal(database)}
                   AND REFERENCED_TABLE_NAME = {sql_string_literal(table_name)};
                """
            ).strip(),
            database=database,
        )
        try:
            inbound_foreign_keys = int(raw_count or "0")
        except ValueError as exc:
            raise DeployError(
                f"invalid inbound foreign-key count for reset snapshot queue table {table_name}: "
                f"{raw_count!r}"
            ) from exc
        if inbound_foreign_keys:
            raise DeployError(
                "reset snapshot connector queue purge refuses undeclared inbound foreign keys: "
                f"table={table_name} inbound_foreign_keys={inbound_foreign_keys}"
            )


def purge_reset_snapshot_connector_queue(envv: dict[str, str]) -> dict[str, int]:
    """Remove inherited queue state before a reset target runtime can start.

    Callers are restricted to staging/test branch-reset paths after
    target quiescence.  This is neither a production repair nor a replay path.
    The allowlisted DDL avoids retaining production delivery/audit history in a
    Staging target and avoids an unbounded history-wide UPDATE transaction.
    """
    counts = reset_snapshot_connector_queue_row_counts(envv)
    if not any(counts.values()):
        return counts

    require_reset_snapshot_connector_queue_truncate_safe(envv)
    for table_name, count_key in RESET_SNAPSHOT_CONNECTOR_QUEUE_PURGE_TABLES:
        if counts[count_key]:
            db_exec(
                envv,
                f"TRUNCATE TABLE `{table_name}`;",
                database=envv["TIDB_DATABASE"],
            )

    remaining = reset_snapshot_connector_queue_row_counts(envv)
    if any(remaining.values()):
        raise DeployError(
            "reset snapshot connector queue purge did not reach an empty state: "
            f"connector_events={remaining['connector_events']}, "
            f"connector_work_items={remaining['connector_work_items']}"
        )
    return counts


def quarantine_connector_inbound_nonterminal_rows_for_installed_packs(envv: dict[str, str], reason: str) -> int:
    if not installed_pack_table_exists(envv) or not tidb_table_exists(envv, "_System_ConnectorEvent"):
        return 0
    count_sql = textwrap.dedent(
        """
        SELECT COUNT(DISTINCT e.Id)
          FROM _System_InstalledPack p
          JOIN _System_ConnectorEvent e
            ON (
                (p.Category = 'connector-pack' AND e.InstalledPackId = p.PackInstallId)
                OR (p.Category = 'app-pack' AND COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId)
            )
         WHERE p.Status IN ('active', 'installing', 'removing', 'failed')
           AND COALESCE(p.IsDeleted, 0) = 0
           AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
           AND e.Status IN ('pending', 'failed', 'processing');
        """
    ).strip()
    count = int(db_query_scalar(envv, count_sql, database=envv["TIDB_DATABASE"]) or "0")
    if count == 0:
        return 0
    error_message = reason.strip()[:1024] or "quarantined by test branch reset"
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_ConnectorEvent e
              JOIN _System_InstalledPack p
                ON (
                    (p.Category = 'connector-pack' AND e.InstalledPackId = p.PackInstallId)
                    OR (p.Category = 'app-pack' AND COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId)
                )
               SET e.Status = 'dead',
                   e.LastError = {sql_string_literal(error_message)},
                   e.LastErrorAt = UTC_TIMESTAMP(),
                   e.NextAttemptAt = NULL,
                   e.ClaimedBy = NULL,
                   e.ClaimedAt = NULL,
                   e.ClaimToken = NULL,
                   e.LastModifiedDate = UTC_TIMESTAMP()
             WHERE p.Status IN ('active', 'installing', 'removing', 'failed')
               AND COALESCE(p.IsDeleted, 0) = 0
               AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
               AND e.Status IN ('pending', 'failed', 'processing');
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    return count


def connector_inbound_nonterminal_diagnostics_for_packs(envv: dict[str, str], pack_names: Iterable[str]) -> str:
    # Diagnostics must use the same lifecycle authority as the drain decision;
    # raw Event.Status is intentionally stale for settled source projections.
    return connector_inbound_nonterminal_summary_for_packs(envv, pack_names)


def reset_drained_workflow_action_syncstate_for_packs(envv: dict[str, str], pack_names: Iterable[str]) -> None:
    names = sorted({str(name).strip() for name in pack_names if str(name).strip()})
    if not names:
        return
    if (
        not installed_pack_table_exists(envv)
        or not tidb_table_exists(envv, "_System_ConnectorEvent")
        or not tidb_table_exists(envv, "_System_SyncState")
    ):
        return

    pack_name_list = ", ".join(sql_string_literal(name) for name in names)
    if connector_inbound_nonterminal_summary_for_packs(envv, names):
        return

    stale_syncstate_count = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT COUNT(*)
              FROM _System_SyncState s
             WHERE s.Object = 'workflow-actions'
               AND s.Direction = 'inbound'
               AND COALESCE(s.RunningErrorCount, 0) <> 0
               AND s.InstalledPackId IN (
                   SELECT p.PackInstallId
                     FROM _System_InstalledPack p
                    WHERE p.Name IN ({pack_name_list})
                      AND p.Status IN ('active', 'installing', 'removing', 'failed')
                      AND COALESCE(p.IsDeleted, 0) = 0
                      AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
               );
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()
    if int(stale_syncstate_count or "0") == 0:
        return

    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_SyncState
               SET RunningErrorCount = 0,
                   LastErrorAt = NULL,
                   LastErrorMessage = NULL,
                   LastModifiedDate = UTC_TIMESTAMP()
             WHERE Object = 'workflow-actions'
               AND Direction = 'inbound'
               AND COALESCE(RunningErrorCount, 0) <> 0
               AND InstalledPackId IN (
                   SELECT p.PackInstallId
                     FROM _System_InstalledPack p
                    WHERE p.Name IN ({pack_name_list})
                      AND p.Status IN ('active', 'installing', 'removing', 'failed')
                      AND COALESCE(p.IsDeleted, 0) = 0
                      AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
               );
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    log(f"reset workflow-actions inbound SyncState after connector queue drain: {stale_syncstate_count}")


def reset_snapshot_workflow_action_syncstate(envv: dict[str, str]) -> int:
    """Clear cloned workflow-action error counters after the global queue purge."""
    if not tidb_table_exists(envv, "_System_SyncState"):
        return 0

    predicate = textwrap.dedent(
        """
        Object = 'workflow-actions'
          AND Direction = 'inbound'
          AND COALESCE(RunningErrorCount, 0) <> 0
        """
    ).strip()
    stale_count = int(
        db_query_scalar(
            envv,
            f"SELECT COUNT(*) FROM _System_SyncState WHERE {predicate};",
            database=envv["TIDB_DATABASE"],
        )
        or "0"
    )
    if stale_count == 0:
        return 0

    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_SyncState
               SET RunningErrorCount = 0,
                   LastErrorAt = NULL,
                   LastErrorMessage = NULL,
                   LastModifiedDate = UTC_TIMESTAMP()
             WHERE {predicate};
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    remaining = int(
        db_query_scalar(
            envv,
            f"SELECT COUNT(*) FROM _System_SyncState WHERE {predicate};",
            database=envv["TIDB_DATABASE"],
        )
        or "0"
    )
    if remaining:
        raise DeployError(
            "reset snapshot workflow-action SyncState cleanup did not converge: "
            f"remaining={remaining}"
        )
    log(f"reset cloned workflow-actions inbound SyncState: {stale_count}")
    return stale_count


def runtime_pack_names(downstream_build_root: Path) -> list[str]:
    lock_path = downstream_build_root / "stack.lock.json"
    if not lock_path.is_file():
        return []
    lock_data = require_json_object(json.loads(lock_path.read_text()), str(lock_path))
    pack_names: list[str] = []
    for pack_entry in runtime_deploy_pack_entries(lock_data):
        pack_name, _, _ = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        if pack_name:
            pack_names.append(pack_name)
    return pack_names


def wait_for_connector_inbound_drain_for_runtime_packs(
    envv: dict[str, str],
    downstream_build_root: Path,
    *,
    completion_context: str = "Catalog candidate convergence",
) -> None:
    timeout_seconds = int(str(envv.get("DEPLOY_CONNECTOR_INBOUND_DRAIN_TIMEOUT_SECONDS", "360")).strip() or "360")
    poll_seconds = max(1, int(str(envv.get("DEPLOY_CONNECTOR_INBOUND_DRAIN_POLL_SECONDS", "5")).strip() or "5"))
    pack_names = runtime_pack_names(downstream_build_root)
    if not pack_names:
        return
    deadline = time.monotonic() + max(0, timeout_seconds)
    last_summary = ""
    while True:
        summary = connector_inbound_nonterminal_summary_for_packs(envv, pack_names)
        if not summary:
            if last_summary:
                log(f"connector inbound queue drained before {completion_context}")
            reset_drained_workflow_action_syncstate_for_packs(envv, pack_names)
            return
        last_summary = summary
        if time.monotonic() >= deadline:
            diagnostics = connector_inbound_nonterminal_diagnostics_for_packs(envv, pack_names)
            details = diagnostics or summary
            raise DeployError(
                f"connector inbound queue has non-terminal rows before {completion_context}: "
                f"{details}. Wait for the queue to drain, or classify this as a test environment/backlog blocker."
            )
        log(f"waiting for connector inbound queue to drain before {completion_context}: {summary}")
        time.sleep(min(poll_seconds, max(0.0, deadline - time.monotonic())))


def sync_stack_local_pack_metadata(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    pack_entries: list[str],
) -> None:
    for pack_entry in pack_entries:
        pack_name, _, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        if pack_dir is None:
            continue
        if installed_pack_status(envv, pack_name) != "active":
            continue
        log(f"syncing active local pack runtime metadata: {pack_name}")
        sync_installed_pack_runtime_metadata(envv, pack_name, pack_dir, nexuscrm_worktree)


def runtime_metadata_tool_env(nexuscrm_worktree: Path) -> dict[str, str]:
    official_root = os.environ.get("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT", "").strip()
    if official_root:
        return {"NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": official_root}

    resolved_worktree = nexuscrm_worktree.resolve()
    candidate_roots = [
        resolved_worktree / "nexuscrm-app-packs",
        resolved_worktree.parent / "nexuscrm-app-packs",
        resolved_worktree.parent.parent / "repos" / "nexuscrm-app-packs",
    ]
    for candidate_root in candidate_roots:
        if candidate_root.is_dir():
            return {"NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": str(candidate_root.resolve())}

    legacy_root = os.environ.get("OFFICIAL_APP_PACKS_ROOT", "").strip()
    if legacy_root:
        return {"NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": legacy_root}

    return {}


def bind_source_backed_pack_runtime_metadata_revision(
    metadata: dict[str, object],
    source_revision: str,
) -> dict[str, object]:
    """Bind persisted metadata to a caller-owned exact source revision."""
    source_revision = str(source_revision).strip()
    if not GIT_COMMIT_RE.fullmatch(source_revision):
        raise DeployError(
            "source-backed pack runtime metadata requires an exact lowercase "
            "git source revision"
        )
    bound_metadata = dict(metadata)
    bound_metadata["source_revision"] = source_revision
    bound_metadata["source_revision_valid"] = True
    return bound_metadata


def _source_backed_pack_runtime_metadata_tool(
    nexuscrm_worktree: Path,
    pack_dir: Path,
    *,
    envv: dict[str, str] | None = None,
    pack_name: str | None = None,
    pack_category: str | None = None,
    stored_manifest_format_version: str | None = None,
    stored_source_revision: str | None = None,
    backend_go_build_tags: str | None = None,
    build_target: Path | None = None,
) -> dict[str, object] | Path:
    backend_root = nexuscrm_worktree / "backend"
    build_tags = supported_backend_go_build_tags(
        backend_go_build_tags if backend_go_build_tags is not None else os.environ.get("BACKEND_GO_BUILD_TAGS", "")
    )
    tool_root = backend_root
    tool_env = runtime_metadata_tool_env(nexuscrm_worktree)
    tool_env.update(optional_process_env(envv or {}))
    configured_official_root = str(
        (envv or {}).get("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT", "")
    ).strip()
    if configured_official_root:
        tool_env["NEXUSCRM_OFFICIAL_APP_PACKS_ROOT"] = configured_official_root
    tool_env.update(private_go_module_env(envv or {}))
    configured_helper = str(
        (envv or {}).get("NEXUSCRM_PACK_RUNTIME_METADATA_BIN", "")
    ).strip()
    if configured_helper and build_target is None:
        helper = Path(configured_helper).expanduser()
        if (
            not helper.is_absolute()
            or helper.is_symlink()
            or not helper.is_file()
            or not os.access(helper, os.X_OK)
        ):
            raise DeployError(
                "NEXUSCRM_PACK_RUNTIME_METADATA_BIN is not an absolute "
                f"executable file: {helper}"
            )
        command = [
            str(helper),
            str(pack_dir.resolve()),
            (pack_name or "").strip(),
            (pack_category or "").strip(),
            (stored_manifest_format_version or "").strip(),
            (stored_source_revision or "").strip(),
        ]
        raw = run(
            command,
            cwd=backend_root,
            env=tool_env,
            capture=True,
        )
        return json.loads(raw)
    go_args = ["go", "run", "-mod=readonly", "./.runtime-tools/pack_runtime_metadata"]
    packs_import_path = "github.com/pingcap-inc/nexuscrm-controlplane/packs"
    if (backend_root / "internal" / "controlplane" / "packs").is_dir():
        packs_import_path = "github.com/nexuscrm/backend/internal/controlplane/packs"
    register_distribution = False
    distribution_root = backend_distribution_root(nexuscrm_worktree)
    if build_tags and distribution_root is not None:
        official_root = tool_env.get("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT", "").strip()
        if not official_root:
            raise DeployError("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT is required for distribution metadata helper")
        ensure_distribution_app_packs_link(nexuscrm_worktree, Path(official_root))
        tool_root = distribution_root
        go_args = [
            "go",
            "run",
            "-mod=readonly",
            "-tags",
            build_tags,
            "./.runtime-tools/pack_runtime_metadata",
        ]
        tool_env["GOWORK"] = "off"
        register_distribution = True

    tool_dir = tool_root / ".runtime-tools" / "pack_runtime_metadata"
    tool_path = tool_dir / "main.go"
    distribution_registration = ""
    if register_distribution:
        distribution_registration = textwrap.dedent(
            """
            if err := registerProductDistribution(); err != nil {
                fmt.Fprintln(os.Stderr, err)
                os.Exit(1)
            }
            """
        )
    write_text_atomic(
        tool_path,
        textwrap.dedent(
            """
            package main

            import (
                "encoding/json"
                "fmt"
                "os"
                "strings"

                "__PACKS_IMPORT_PATH__"
            )

            var installedPackRegistrationSet *packs.ComponentKindRegistrationSet

            type payload struct {
                Name                        string `json:"name"`
                PayloadHash                 string `json:"payload_hash"`
                PackVersion                 string `json:"pack_version"`
                ManifestFormatVersion       string `json:"manifest_format_version"`
                PlatformCompatibility       string `json:"platform_compatibility"`
                PlatformCompatibilityValid  bool   `json:"platform_compatibility_valid"`
                PayloadHashAlgorithm        string `json:"payload_hash_algorithm"`
                SourceKind                  string `json:"source_kind"`
                SourceKindValid             bool   `json:"source_kind_valid"`
                SourceURI                   string `json:"source_uri"`
                SourceURIValid              bool   `json:"source_uri_valid"`
                SourceRevision              string `json:"source_revision"`
                SourceRevisionValid         bool   `json:"source_revision_valid"`
                RuntimeObjectSpecDecl       string `json:"runtime_object_spec_declarations"`
                RuntimeRelatedListRules     string `json:"runtime_related_list_rules"`
                RuntimeSearchScopeOrder     string `json:"runtime_search_scope_order"`
                RuntimeSearchAllDefaultFields string `json:"runtime_search_all_default_fields"`
                RuntimeFormulaFunctionBundles string `json:"runtime_formula_function_bundles"`
                RuntimeStartupShallowDecl   string `json:"runtime_startup_shallow_declarations"`
            }

            type searchAllDefaultFieldsJSONProvider interface {
                SearchAllDefaultFieldsJSON() string
            }

            func searchAllDefaultFieldsJSON(declaration any) string {
                provider, ok := declaration.(searchAllDefaultFieldsJSONProvider)
                if !ok {
                    return "[]"
                }
                value := strings.TrimSpace(provider.SearchAllDefaultFieldsJSON())
                if value == "" {
                    return "[]"
                }
                return value
            }

            func main() {
                if len(os.Args) != 6 {
                    fmt.Fprintln(os.Stderr, "usage: pack_runtime_metadata <pack_dir> <pack_name> <pack_category> <stored_manifest_format_version> <stored_source_revision>")
                    os.Exit(2)
                }
                __DISTRIBUTION_REGISTRATION__
                packDir := os.Args[1]
                packName := strings.TrimSpace(os.Args[2])
                packCategory := strings.TrimSpace(os.Args[3])
                storedManifestFormatVersion := strings.TrimSpace(os.Args[4])
                storedSourceRevision := strings.TrimSpace(os.Args[5])
                var pack packs.Pack
                if packName != "" && packCategory != "" && storedManifestFormatVersion != "" {
                    resolved, err := packs.ResolveInstalledPack(
                        packs.InstalledPackResolverInput{
                            Name:                  packName,
                            Category:              packs.Category(packCategory),
                            ManifestFormatVersion: storedManifestFormatVersion,
                            SourceKind:            string(packs.PackSourceKindLocalPath),
                            SourceURI:             packDir,
                            SourceRevision:        storedSourceRevision,
                        },
                        packs.InstalledPackResolverOptions{
                            AllowBundledFallback:         false,
                            ComponentKindRegistrationSet: installedPackRegistrationSet,
                        },
                    )
                    if err != nil {
                        fmt.Fprintln(os.Stderr, err)
                        os.Exit(1)
                    }
                    pack = resolved
                } else {
                    sourceOpts := packs.DefaultPackSourceOptions()
                    source, diagnostics := packs.ResolvePackSourceWithOptions("local_path:"+packDir, sourceOpts)
                    if len(diagnostics) > 0 {
                        for _, diagnostic := range diagnostics {
                            fmt.Fprintln(os.Stderr, diagnostic.String())
                        }
                        os.Exit(1)
                    }
                    if source == nil || source.Manifest == nil {
                        fmt.Fprintln(os.Stderr, "resolved pack source is empty")
                        os.Exit(1)
                    }
                    packOpts := []packs.SourceBackedPackOption{
                        packs.WithFlowActionCallRuntimeFromPackSourceOptions(sourceOpts),
                        packs.WithConnectorPushValidationSchemaResolverForSource(source, sourceOpts),
                    }
                    if installedPackRegistrationSet != nil {
                        packOpts = append(packOpts, packs.WithComponentKindRegistrationSet(*installedPackRegistrationSet))
                    }
                    sourceBackedPack, err := packs.NewSourceBackedPack(source, packOpts...)
                    if err != nil {
                        fmt.Fprintln(os.Stderr, err)
                        os.Exit(1)
                    }
                    pack = sourceBackedPack
                }
                payloadHash, err := packs.PackPayloadHash(pack)
                if err != nil {
                    fmt.Fprintln(os.Stderr, err)
                    os.Exit(1)
                }
                declaration, err := packs.BuildPersistedRuntimeDeclaration(pack)
                if err != nil {
                    fmt.Fprintln(os.Stderr, err)
                    os.Exit(1)
                }
                objectSpecDeclarationsJSON := declaration.ObjectPayloadsJSON()
                meta := packs.PackHashMetadataForPack(pack)
                if err := json.NewEncoder(os.Stdout).Encode(payload{
                    Name:                       pack.Name(),
                    PayloadHash:                payloadHash,
                    PackVersion:                meta.PackVersion,
                    ManifestFormatVersion:      meta.ManifestFormatVersion,
                    PlatformCompatibility:      meta.PlatformCompatibility,
                    PlatformCompatibilityValid: meta.PlatformCompatibilityValid,
                    PayloadHashAlgorithm:       meta.PayloadHashAlgorithm,
                    SourceKind:                 meta.SourceKind,
                    SourceKindValid:            meta.SourceKindValid,
                    SourceURI:                  meta.SourceURI,
                    SourceURIValid:             meta.SourceURIValid,
                    SourceRevision:             meta.SourceRevision,
                    SourceRevisionValid:        meta.SourceRevisionValid,
                    RuntimeObjectSpecDecl:      objectSpecDeclarationsJSON,
                    RuntimeRelatedListRules:    declaration.RelatedListRulesJSON(),
                    RuntimeSearchScopeOrder:    declaration.SearchScopeOrderJSON(),
                    RuntimeSearchAllDefaultFields: searchAllDefaultFieldsJSON(declaration),
                    RuntimeFormulaFunctionBundles: declaration.FormulaFunctionBundlesJSON(),
                    RuntimeStartupShallowDecl:  declaration.StartupShallowDeclaredRowsJSON(),
                }); err != nil {
                    fmt.Fprintln(os.Stderr, err)
                    os.Exit(1)
                }
            }
            """
        )
        .replace("__PACKS_IMPORT_PATH__", packs_import_path)
        .replace("__DISTRIBUTION_REGISTRATION__", distribution_registration.rstrip())
        .strip()
        + "\n",
    )
    if register_distribution:
        write_text_atomic(
            tool_dir / "distribution_default.go",
            textwrap.dedent(
                """
                //go:build !crm

                package main

                func registerProductDistribution() error {
                    return nil
                }
                """
            ).strip()
            + "\n",
        )
        write_text_atomic(
            tool_dir / "distribution_crm.go",
            textwrap.dedent(
                """
                //go:build crm

                package main

                import (
                    crmdistribution "github.com/nexuscrm/backend/distributions/crm"
                    "github.com/nexuscrm/backend/pkg/serverruntime"
                )

                func registerProductDistribution() error {
                    hooks, err := serverruntime.RegisterProductDistributionCapabilities(
                        serverruntime.NewProductDistribution(crmdistribution.ProductHooks()),
                    )
                    if err != nil {
                        return err
                    }
                    set := hooks.ComponentKindRegistrationSet()
                    installedPackRegistrationSet = &set
                    return nil
                }
                """
            ).strip()
            + "\n",
        )
    try:
        if build_target is not None:
            resolved_target = build_target.expanduser().resolve()
            resolved_target.parent.mkdir(parents=True, exist_ok=True)
            temporary_binary = resolved_target.with_name(
                f".{resolved_target.name}.tmp"
            )
            remove_path(temporary_binary)
            build_args = ["go", "build", "-mod=readonly"]
            if register_distribution:
                build_args.extend(["-tags", build_tags])
            build_args.extend(
                [
                    "-o",
                    str(temporary_binary),
                    "./.runtime-tools/pack_runtime_metadata",
                ]
            )
            try:
                run(build_args, cwd=tool_root, env=tool_env, capture=True)
                temporary_binary.chmod(0o755)
                os.replace(temporary_binary, resolved_target)
            except BaseException:
                remove_path(temporary_binary)
                raise
            result: dict[str, object] | Path = resolved_target
        else:
            raw = run(
                [
                    *go_args,
                    str(pack_dir.resolve()),
                    (pack_name or "").strip(),
                    (pack_category or "").strip(),
                    (stored_manifest_format_version or "").strip(),
                    (stored_source_revision or "").strip(),
                ],
                cwd=tool_root,
                env=tool_env,
                capture=True,
            )
            result = json.loads(raw)
    finally:
        shutil.rmtree(tool_dir, ignore_errors=True)
        try:
            tool_dir.parent.rmdir()
        except OSError:
            pass
    return result


def compute_source_backed_pack_runtime_metadata(
    nexuscrm_worktree: Path,
    pack_dir: Path,
    *,
    envv: dict[str, str] | None = None,
    pack_name: str | None = None,
    pack_category: str | None = None,
    stored_manifest_format_version: str | None = None,
    stored_source_revision: str | None = None,
    backend_go_build_tags: str | None = None,
) -> dict[str, object]:
    result = _source_backed_pack_runtime_metadata_tool(
        nexuscrm_worktree,
        pack_dir,
        envv=envv,
        pack_name=pack_name,
        pack_category=pack_category,
        stored_manifest_format_version=stored_manifest_format_version,
        stored_source_revision=stored_source_revision,
        backend_go_build_tags=backend_go_build_tags,
    )
    assert isinstance(result, dict)
    return result


def prepare_pack_runtime_metadata_binary(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    target: Path,
) -> Path:
    build_env = dict(envv)
    build_env["NEXUSCRM_OFFICIAL_APP_PACKS_ROOT"] = str(
        app_packs_root.expanduser().resolve()
    )
    result = _source_backed_pack_runtime_metadata_tool(
        nexuscrm_worktree,
        app_packs_root,
        envv=build_env,
        backend_go_build_tags=envv.get("BACKEND_GO_BUILD_TAGS"),
        build_target=target,
    )
    assert isinstance(result, Path)
    return result


def rebind_retryable_installed_pack_source_identity(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    nexuscrm_worktree: Path,
) -> None:
    if not installed_pack_table_exists(envv):
        return
    metadata = compute_source_backed_pack_runtime_metadata(
        nexuscrm_worktree,
        pack_dir,
        envv=envv,
        backend_go_build_tags=envv.get("BACKEND_GO_BUILD_TAGS"),
    )
    source_kind = sql_nullable_string_literal(
        str(metadata["source_kind"]),
        bool(metadata["source_kind_valid"]),
    )
    source_uri = sql_nullable_string_literal(
        str(metadata["source_uri"]),
        bool(metadata["source_uri_valid"]),
    )
    source_revision = sql_nullable_string_literal(
        str(metadata["source_revision"]),
        bool(metadata["source_revision_valid"]),
    )
    escaped_name = pack_name.replace("'", "''")
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_InstalledPack
            SET SourceKind={source_kind},
                SourceURI={source_uri},
                SourceRevision={source_revision},
                LastModifiedDate=UTC_TIMESTAMP()
            WHERE Name='{escaped_name}'
              AND Status IN ('installing', 'failed')
              AND COALESCE(IsDeleted, 0) = 0;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def load_installed_pack_runtime_boundary(
    envv: dict[str, str],
    pack_name: str,
) -> tuple[str, str, str]:
    escaped_name = pack_name.replace("'", "''")
    row = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT JSON_OBJECT(
                'category', COALESCE(Category, ''),
                'manifest_format_version', COALESCE(ManifestFormatVersion, ''),
                'source_revision', COALESCE(SourceRevision, '')
            )
            FROM _System_InstalledPack
            WHERE Name = '{escaped_name}'
              AND Status IN ('active', 'installing', 'failed')
              AND COALESCE(IsDeleted, 0) = 0
            LIMIT 1;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )
    if not row:
        return "", "", ""
    payload = json.loads(row)
    if not isinstance(payload, dict):
        return "", "", ""
    return (
        str(payload.get("category", "")).strip(),
        str(payload.get("manifest_format_version", "")).strip(),
        str(payload.get("source_revision", "")).strip(),
    )


def sync_installed_pack_runtime_metadata_with_apply_state(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    nexuscrm_worktree: Path,
    *,
    include_runtime_declarations: bool = True,
    mark_applied: bool,
    source_revision: str | None = None,
) -> None:
    if not installed_pack_table_exists(envv):
        return
    exact_source_revision: str | None = None
    if source_revision is not None:
        exact_source_revision = str(source_revision).strip()
        if not GIT_COMMIT_RE.fullmatch(exact_source_revision):
            raise DeployError(
                "installed-pack metadata sync requires an exact lowercase git "
                "source revision"
            )
    normalize_installed_pack_runtime_state(envv, pack_name, pack_dir)
    pack_category, stored_manifest_format_version, stored_source_revision = load_installed_pack_runtime_boundary(
        envv, pack_name
    )
    log(
        "installed pack runtime metadata boundary: "
        f"name={pack_name} category={pack_category or '<empty>'} "
        f"manifest={stored_manifest_format_version or '<empty>'} "
        f"source_revision={stored_source_revision or '<empty>'}"
    )
    metadata = compute_source_backed_pack_runtime_metadata(
        nexuscrm_worktree,
        pack_dir,
        envv=envv,
        pack_name=pack_name,
        pack_category=pack_category,
        stored_manifest_format_version=stored_manifest_format_version,
        # Warm deploy repair advances to the current source payload. A caller
        # with a frozen source plan may additionally bind that exact revision;
        # otherwise the mutable stored row must not retain authority.
        stored_source_revision=exact_source_revision or "",
        backend_go_build_tags=envv.get("BACKEND_GO_BUILD_TAGS"),
    )
    if exact_source_revision is not None:
        metadata = bind_source_backed_pack_runtime_metadata_revision(
            metadata,
            exact_source_revision,
        )
    log(
        "computed installed pack runtime metadata: "
        f"name={pack_name} hash={metadata['payload_hash']} "
        f"manifest={metadata['manifest_format_version']} "
        f"source_revision={metadata['source_revision']}"
    )
    source_kind = sql_nullable_string_literal(
        str(metadata["source_kind"]),
        bool(metadata["source_kind_valid"]),
    )
    source_uri = sql_nullable_string_literal(
        str(metadata["source_uri"]),
        bool(metadata["source_uri_valid"]),
    )
    runtime_decl_updates: list[str] = []
    if include_runtime_declarations:
        runtime_decl_updates.append(
            f"RuntimeObjectSpecDeclarations={sql_string_literal(str(metadata['runtime_object_spec_declarations']))}"
        )
        runtime_decl_updates.append(
            f"RuntimeRelatedListRules={sql_string_literal(str(metadata['runtime_related_list_rules']))}"
        )
        runtime_decl_updates.append(
            f"RuntimeSearchScopeOrder={sql_string_literal(str(metadata['runtime_search_scope_order']))}"
        )
        if installed_pack_column_exists(envv, "RuntimeSearchAllDefaultFields"):
            runtime_decl_updates.append(
                "RuntimeSearchAllDefaultFields="
                f"{sql_string_literal(str(metadata['runtime_search_all_default_fields']))}"
            )
        else:
            log("skip RuntimeSearchAllDefaultFields sync; _System_InstalledPack column is absent")
        if installed_pack_column_exists(envv, "RuntimeFormulaFunctionBundles"):
            runtime_decl_updates.append(
                "RuntimeFormulaFunctionBundles="
                f"{sql_string_literal(str(metadata['runtime_formula_function_bundles']))}"
            )
        else:
            log("skip RuntimeFormulaFunctionBundles sync; _System_InstalledPack column is absent")
        runtime_decl_updates.append(
            "RuntimeStartupShallowDeclarations="
            f"{sql_string_literal(str(metadata['runtime_startup_shallow_declarations']))}"
        )
    else:
        log(
            "skip-bootstrap runtime metadata sync uses legacy installed-pack update "
            "without runtime declaration columns"
        )
    runtime_decl_update_sql = "".join(f"\n                {update}," for update in runtime_decl_updates)
    escaped_name = pack_name.replace("'", "''")
    applied_payload_hash_sql = (
        sql_string_literal(str(metadata["payload_hash"]))
        if mark_applied
        else "AppliedPayloadHash"
    )
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_InstalledPack
            SET Status='active',
                PackVersion={sql_string_literal(str(metadata["pack_version"]))},
                ManifestFormatVersion={sql_string_literal(str(metadata["manifest_format_version"]))},
                PlatformCompatibility={sql_nullable_string_literal(
                    str(metadata["platform_compatibility"]),
                    bool(metadata["platform_compatibility_valid"]),
                )},
                DesiredPayloadHash={sql_string_literal(str(metadata["payload_hash"]))},
                AppliedPayloadHash={applied_payload_hash_sql},
                PayloadHashAlgorithm={sql_string_literal(str(metadata["payload_hash_algorithm"]))},
                SourceKind={source_kind},
                SourceURI={source_uri},
                SourceRevision={sql_nullable_string_literal(
                    str(metadata["source_revision"]),
                    bool(metadata["source_revision_valid"]),
                )},{runtime_decl_update_sql}
                LastError=NULL,
                LastAppliedDate=UTC_TIMESTAMP(),
                LastModifiedDate=UTC_TIMESTAMP()
            WHERE Name='{escaped_name}'
              AND Status IN ('active', 'installing', 'failed')
              AND COALESCE(IsDeleted, 0) = 0;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )


def sync_installed_pack_runtime_metadata(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    nexuscrm_worktree: Path,
    *,
    include_runtime_declarations: bool = True,
    source_revision: str | None = None,
) -> None:
    sync_installed_pack_runtime_metadata_with_apply_state(
        envv,
        pack_name,
        pack_dir,
        nexuscrm_worktree,
        include_runtime_declarations=include_runtime_declarations,
        mark_applied=True,
        source_revision=source_revision,
    )


def sync_installed_pack_runtime_metadata_without_marking_applied(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    nexuscrm_worktree: Path,
    *,
    include_runtime_declarations: bool = True,
    source_revision: str | None = None,
) -> None:
    # Catalog candidate convergence must see DesiredPayloadHash !=
    # AppliedPayloadHash when a warm target database still needs pack repair.
    # Backend reconcile uses that installed-pack hash drift to enter the repair
    # path. If this deploy script stamps AppliedPayloadHash to the new payload
    # before Catalog runs, backend wrongly believes the pack is already applied
    # and dies in fatal verify drift instead of repairing schema/data.
    sync_installed_pack_runtime_metadata_with_apply_state(
        envv,
        pack_name,
        pack_dir,
        nexuscrm_worktree,
        include_runtime_declarations=include_runtime_declarations,
        mark_applied=False,
        source_revision=source_revision,
    )


def run_catalog_offline_candidate(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    *,
    downstream_source_root: Path | None = None,
) -> None:
    operation_id = str(envv.get("DEPLOY_ID", "")).strip()
    if not operation_id or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", operation_id):
        raise DeployError("Catalog offline candidate requires a safe DEPLOY_ID")
    if bool_env(envv, "SETUP_REPAIR_LEGACY_UNIQUE_DUPLICATES"):
        raise DeployError(
            "Catalog offline candidate does not support SETUP_REPAIR_LEGACY_UNIQUE_DUPLICATES=true"
        )
    stack_lock_path = downstream_build_root / "stack.lock.json"
    lock_data = require_json_object(json.loads(stack_lock_path.read_text()), str(stack_lock_path))
    downstream_revision = str(envv.get("DOWNSTREAM_REF", "")).strip()
    checkout_revision = exact_downstream_source_revision(downstream_build_root)
    if not GIT_COMMIT_RE.fullmatch(downstream_revision) or checkout_revision != downstream_revision:
        raise DeployError(
            "Catalog offline candidate requires the exact downstream checkout: "
            f"expected {downstream_revision or '<missing>'}, got {checkout_revision or '<missing>'}"
        )
    args = ["apply-offline-candidate", "--operation-id", operation_id]
    for pack_entry in runtime_deploy_pack_entries(lock_data):
        _, source, pack_dir = runtime_pack_entry_name_and_source(downstream_build_root, pack_entry)
        if pack_dir is not None:
            source, receipt = runtime_pack_entry_release_source(
                downstream_build_root,
                pack_entry,
                downstream_revision,
                downstream_source_root=downstream_source_root,
            )
            args.extend(
                [
                    "--release-source-receipt",
                    json.dumps(receipt, sort_keys=True, separators=(",", ":")),
                ]
            )
        args.extend(["--source", source])
    run_catalog_setup_args_with_retry(
        envv,
        nexuscrm_worktree,
        app_packs_root,
        args,
        description="complete Catalog-owned offline candidate",
    )


def run_catalog_official_source_set(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    pack_names: list[str],
) -> None:
    """Replace selected installed official packs under stopped-source authority."""
    operation_id = str(envv.get("DEPLOY_ID", "")).strip()
    if not operation_id or not re.fullmatch(
        r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", operation_id
    ):
        raise DeployError("Catalog official source-set apply requires a safe DEPLOY_ID")
    official_revision = str(envv.get("OFFICIAL_APP_PACKS_REF", "")).strip()
    if not GIT_COMMIT_RE.fullmatch(official_revision):
        raise DeployError(
            "Catalog official source-set apply requires an exact OFFICIAL_APP_PACKS_REF"
        )
    checkout_revision = exact_git_source_revision(app_packs_root)
    if checkout_revision != official_revision:
        raise DeployError(
            "Catalog official source-set apply requires the exact AppPacks checkout: "
            f"expected {official_revision}, got {checkout_revision or '<missing>'}"
        )
    normalized_names = sorted({str(name).strip() for name in pack_names})
    if not normalized_names or any(not name for name in normalized_names):
        raise DeployError(
            "Catalog official source-set apply requires at least one pack name"
        )

    args = ["apply-source-set", "--operation-id", operation_id]
    for pack_name in normalized_names:
        if "/" in pack_name or "\\" in pack_name or pack_name in {".", ".."}:
            raise DeployError(
                f"Catalog official source-set pack name is invalid: {pack_name!r}"
            )
        pack_json = app_packs_root / pack_name / "pack.json"
        if not pack_json.is_file() or pack_json.is_symlink():
            raise DeployError(
                f"Catalog official source-set source is missing pack.json: {pack_json}"
            )
        manifest = require_json_object(
            json.loads(pack_json.read_text()),
            str(pack_json),
        )
        actual_name = str(manifest.get("name") or "").strip()
        if actual_name != pack_name:
            raise DeployError(
                "Catalog official source-set pack name does not match its manifest: "
                f"expected {pack_name}, got {actual_name or '<missing>'}"
            )
        source = f"official_repo:{pack_name}"
        args.extend(
            [
                "--official-source-revision",
                json.dumps(
                    {"name": pack_name, "revision": official_revision},
                    sort_keys=True,
                    separators=(",", ":"),
                ),
                "--source",
                source,
            ]
        )
    run_backend_setup(envv, nexuscrm_worktree, app_packs_root, args)


def run_catalog_source_set(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    *,
    downstream_source_root: Path | None = None,
) -> None:
    """Preconverge one complete active BOM set; the full candidate still follows."""
    operation_id = str(envv.get("DEPLOY_ID", "")).strip()
    if not operation_id or not re.fullmatch(
        r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", operation_id
    ):
        raise DeployError("Catalog source-set apply requires a safe DEPLOY_ID")
    stack_lock_path = downstream_build_root / "stack.lock.json"
    lock_data = require_json_object(
        json.loads(stack_lock_path.read_text()), str(stack_lock_path)
    )
    downstream_revision = str(envv.get("DOWNSTREAM_REF", "")).strip()
    checkout_revision = exact_downstream_source_revision(downstream_build_root)
    if (
        not GIT_COMMIT_RE.fullmatch(downstream_revision)
        or checkout_revision != downstream_revision
    ):
        raise DeployError(
            "Catalog source-set apply requires the exact downstream checkout: "
            f"expected {downstream_revision or '<missing>'}, "
            f"got {checkout_revision or '<missing>'}"
        )
    official_revision = str(envv.get("OFFICIAL_APP_PACKS_REF", "")).strip()
    if not GIT_COMMIT_RE.fullmatch(official_revision):
        raise DeployError(
            "Catalog source-set apply requires an exact OFFICIAL_APP_PACKS_REF"
        )
    if exact_git_source_revision(app_packs_root) != official_revision:
        raise DeployError(
            "Catalog source-set apply requires the exact AppPacks checkout"
        )

    args = ["apply-source-set", "--operation-id", operation_id]
    for pack_entry in runtime_deploy_pack_entries(lock_data):
        pack_name, source, pack_dir = runtime_pack_entry_name_and_source(
            downstream_build_root, pack_entry
        )
        if pack_dir is None:
            args.extend(
                [
                    "--official-source-revision",
                    json.dumps(
                        {"name": pack_name, "revision": official_revision},
                        sort_keys=True,
                        separators=(",", ":"),
                    ),
                ]
            )
        else:
            source, release_source = runtime_pack_entry_release_source(
                downstream_build_root,
                pack_entry,
                downstream_revision,
                downstream_source_root=downstream_source_root,
            )
            args.extend(
                [
                    "--release-source",
                    json.dumps(
                        release_source, sort_keys=True, separators=(",", ":")
                    ),
                ]
            )
        args.extend(["--source", source])
    run_catalog_setup_args_with_retry(
        envv,
        nexuscrm_worktree,
        app_packs_root,
        args,
        description="Catalog source-set preconvergence",
    )


def runtime_dependency_source_revision_advances(
    pack_source_roots: dict[str, Path],
    installed_revisions: dict[str, str],
    target_revisions: dict[str, str],
) -> list[tuple[str, str, str, str]]:
    """Return candidate dependency edges whose dependency revision will change."""
    advances: list[tuple[str, str, str, str]] = []
    for dependent_name, pack_source_root in pack_source_roots.items():
        pack_json = pack_source_root / "pack.json"
        if not pack_json.is_file() or pack_json.is_symlink():
            raise DeployError(
                "Catalog dependency preconvergence source is missing pack.json: "
                f"{pack_json}"
            )
        manifest = require_json_object(json.loads(pack_json.read_text()), str(pack_json))
        manifest_name = str(manifest.get("name") or "").strip()
        if manifest_name != dependent_name:
            raise DeployError(
                "Catalog dependency preconvergence pack name does not match its manifest: "
                f"expected {dependent_name}, got {manifest_name or '<missing>'}"
            )
        for dependency_name, _version_constraint in _runtime_pack_declared_dependencies(
            manifest
        ):
            target_revision = target_revisions.get(dependency_name)
            if target_revision is None:
                continue
            installed_revision = installed_revisions.get(dependency_name, "")
            if installed_revision != target_revision:
                advances.append(
                    (
                        dependent_name,
                        dependency_name,
                        installed_revision,
                        target_revision,
                    )
                )
    return advances


def run_catalog_mutable_source_preconvergence(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    *,
    downstream_source_root: Path | None = None,
) -> None:
    """Preconverge dependency revision advances and resume mutable-source failures."""
    operation_id = str(envv.get("DEPLOY_ID", "")).strip()
    if not operation_id or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", operation_id):
        raise DeployError("Catalog mutable-source preconvergence requires a safe DEPLOY_ID")
    if bool_env(envv, "SETUP_REPAIR_LEGACY_UNIQUE_DUPLICATES"):
        raise DeployError(
            "Catalog mutable-source preconvergence does not support "
            "SETUP_REPAIR_LEGACY_UNIQUE_DUPLICATES=true"
        )
    stack_lock_path = downstream_build_root / "stack.lock.json"
    lock_data = require_json_object(json.loads(stack_lock_path.read_text()), str(stack_lock_path))
    downstream_revision = str(envv.get("DOWNSTREAM_REF", "")).strip()
    checkout_revision = exact_downstream_source_revision(downstream_build_root)
    if not GIT_COMMIT_RE.fullmatch(downstream_revision) or checkout_revision != downstream_revision:
        raise DeployError(
            "Catalog mutable-source preconvergence requires the exact downstream checkout: "
            f"expected {downstream_revision or '<missing>'}, got {checkout_revision or '<missing>'}"
        )

    derived_operation_id = "mutable-preconverge-" + hashlib.sha256(
        operation_id.encode("utf-8")
    ).hexdigest()[:24]
    args = ["apply-offline-candidate", "--operation-id", derived_operation_id]
    needs_recovery_candidate = False
    local_source_adoptions: list[tuple[str, str, str]] = []
    complete_active_set = True
    official_revision = str(envv.get("OFFICIAL_APP_PACKS_REF", "")).strip()
    pack_source_roots: dict[str, Path] = {}
    installed_revisions: dict[str, str] = {}
    target_revisions: dict[str, str] = {}
    for pack_entry in runtime_deploy_pack_entries(lock_data):
        pack_name, source, pack_dir = runtime_pack_entry_name_and_source(
            downstream_build_root,
            pack_entry,
        )
        status, source_kind, source_uri, source_revision = (
            load_installed_pack_convergence_state(envv, pack_name)
        )
        complete_active_set = complete_active_set and status == "active"
        pack_source_roots[pack_name] = pack_dir or (app_packs_root / pack_name)
        installed_revisions[pack_name] = source_revision
        target_revisions[pack_name] = (
            downstream_revision if pack_dir is not None else official_revision
        )
        if pack_dir is not None:
            if status in {"installing", "failed"} and source_kind == "local_path":
                # A prior mutable-source attempt already wrote the exact
                # candidate identity before failing. Retry that convergence;
                # do not relabel the row or pretend it is active.
                needs_recovery_candidate = True
            elif status == "active" and source_kind == "local_path":
                # A production-clone baseline can retain a mutable L2 pack at
                # any source revision.  The immutable Candidate may adopt
                # local_path as release_repo only when revision, desired hash,
                # and applied hash already match exactly.  This reader does not
                # own the target payload hash, so it must not infer equality
                # from revision alone: converge the complete stopped source set
                # before asking Controlplane to relabel it.
                local_source_adoptions.append(
                    (pack_name, source_revision, downstream_revision)
                )
            elif not status or (
                status in {"active", "installing", "failed"}
                and source_kind == "release_repo"
            ):
                # Retryable immutable rows are owned by Controlplane's exact
                # source-revision/payload admission. Supply the sealed receipt
                # and let that owner accept or reject the retry.
                source, receipt = runtime_pack_entry_release_source(
                    downstream_build_root,
                    pack_entry,
                    downstream_revision,
                    downstream_source_root=downstream_source_root,
                )
                args.extend(
                    [
                        "--release-source-receipt",
                        json.dumps(receipt, sort_keys=True, separators=(",", ":")),
                    ]
                )
            elif not (status == "active" and source_kind in {"", "local_path"}):
                raise DeployError(
                    "Catalog mutable-source preconvergence cannot classify installed pack: "
                    f"name={pack_name} status={status or '<missing>'} "
                    f"source_kind={source_kind or '<empty>'} "
                    f"source_uri={source_uri or '<empty>'} "
                    f"source_revision={source_revision or '<empty>'}"
                )
        args.extend(["--source", source])

    dependency_advances: list[tuple[str, str, str, str]] = []
    if complete_active_set:
        dependency_advances = runtime_dependency_source_revision_advances(
            pack_source_roots,
            installed_revisions,
            target_revisions,
        )
    if (
        not dependency_advances
        and not local_source_adoptions
        and not needs_recovery_candidate
    ):
        log("Catalog dependency-order preconvergence is not required")
        return
    if not complete_active_set:
        # Resume the incomplete Candidate before considering any active mutable
        # row for complete source-set adoption. The incomplete row may be an
        # official pack or a local pack; the source-set path requires every
        # installed row to be active.
        run_catalog_setup_args_with_retry(
            envv,
            nexuscrm_worktree,
            app_packs_root,
            args,
            description="Catalog mutable-source preconvergence candidate",
        )
        return
    if dependency_advances or local_source_adoptions:
        formatted_dependency_advances = ", ".join(
            f"{dependent}->{dependency} "
            f"{installed_revision or '<empty>'}->{target_revision or '<empty>'}"
            for dependent, dependency, installed_revision, target_revision in (
                dependency_advances
            )
        )
        formatted_local_adoptions = ", ".join(
            f"local_path adoption {name} "
            f"{installed_revision or '<empty>'}->{target_revision or '<empty>'}"
            for name, installed_revision, target_revision in (
                local_source_adoptions
            )
        )
        formatted_advances = "; ".join(
            value
            for value in (
                formatted_dependency_advances,
                formatted_local_adoptions,
            )
            if value
        )
        log(
            "Catalog source transition requires source-set "
            f"preconvergence: {formatted_advances}"
        )
        # This is a stopped-Test mutation, not a second Candidate lifecycle.
        # Controlplane orders the complete set; the caller's full Candidate
        # remains the only final verification and activation boundary.
        run_catalog_source_set(
            envv,
            downstream_build_root,
            nexuscrm_worktree,
            app_packs_root,
            downstream_source_root=downstream_source_root,
        )
        return
    raise DeployError("Catalog mutable-source preconvergence has no classified action")


def run_catalog_setup_args_with_retry(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    args: list[str],
    *,
    description: str,
) -> None:
    setup_env = dict(envv)
    setup_env.pop("SETUP_SKIP_POLYMORPHIC_BACKFILL", None)
    setup_env.pop("SETUP_REPAIR_LEGACY_UNIQUE_DUPLICATES", None)
    attempts, delay_seconds = setup_retry_policy(envv)
    for attempt in range(1, attempts + 1):
        log(f"submitting {description} attempt={attempt}/{attempts}")
        try:
            run_backend_setup(
                setup_env,
                nexuscrm_worktree,
                app_packs_root,
                args,
            )
            return
        except SetupCommandError as exc:
            if attempt == attempts or not is_transient_setup_connection_error(exc):
                raise
            log(
                "Catalog setup transient TiDB connection loss "
                f"(attempt {attempt}/{attempts}); retrying in {delay_seconds}s"
            )
            time.sleep(delay_seconds)
    raise DeployError(f"{description} failed without an attempt")


def setup_retry_policy(envv: dict[str, str]) -> tuple[int, int]:
    raw_attempts = str(envv.get("SETUP_RETRY_ATTEMPTS", "1")).strip()
    raw_delay_seconds = str(envv.get("SETUP_RETRY_DELAY_SECONDS", "15")).strip()
    try:
        attempts = int(raw_attempts)
    except ValueError as exc:
        raise DeployError(f"SETUP_RETRY_ATTEMPTS must be a positive integer, got {raw_attempts!r}") from exc
    if attempts <= 0:
        raise DeployError(f"SETUP_RETRY_ATTEMPTS must be a positive integer, got {raw_attempts!r}")
    try:
        delay_seconds = int(raw_delay_seconds)
    except ValueError as exc:
        raise DeployError(
            f"SETUP_RETRY_DELAY_SECONDS must be a non-negative integer, got {raw_delay_seconds!r}"
        ) from exc
    if delay_seconds < 0:
        raise DeployError(
            f"SETUP_RETRY_DELAY_SECONDS must be a non-negative integer, got {raw_delay_seconds!r}"
        )
    return attempts, delay_seconds


def supports_synthetic_frontend_app_packs_source(
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
) -> bool:
    stack_lock_path = downstream_build_root / "stack.lock.json"
    try:
        stack_lock = json.loads(stack_lock_path.read_text())
    except (FileNotFoundError, json.JSONDecodeError) as exc:
        raise DeployError(f"invalid stack lock for frontend source selection: {stack_lock_path}: {exc}") from exc
    if not isinstance(stack_lock, dict):
        raise DeployError(f"invalid stack lock for frontend source selection: {stack_lock_path}")
    pack_build = stack_lock.get("pack_build")
    if not isinstance(pack_build, dict):
        pack_build = {}
    final_pack_relative = Path(str(pack_build.get("final_pack_path") or "packs/NOT-EDIT-final-merged-packs"))
    if final_pack_relative.is_absolute() or ".." in final_pack_relative.parts:
        raise DeployError(f"unsafe final pack path for frontend source selection: {final_pack_relative}")
    final_pack_dir = (downstream_build_root / final_pack_relative).resolve()
    try:
        final_pack_dir.relative_to(downstream_build_root.resolve())
    except ValueError as exc:
        raise DeployError(f"final pack path escapes downstream build root: {final_pack_relative}") from exc
    if not (final_pack_dir / "frontend" / "source-manifest.json").is_file():
        return False

    provisioner = nexuscrm_worktree / "frontend" / "scripts" / "provision-frontend-app-packages.mjs"
    if not provisioner.is_file():
        raise DeployError(
            "final pack supplies frontend source but selected NexusCRM runtime has no frontend app-pack provisioner"
        )
    provisioner_source = provisioner.read_text()
    required_env_names = (
        "NEXUSCRM_FRONTEND_APP_PACKS_ROOT",
        "NEXUSCRM_FRONTEND_APP_PACKS_REF",
    )
    missing = [name for name in required_env_names if name not in provisioner_source]
    if missing:
        raise DeployError(
            "final pack supplies frontend source but selected NexusCRM provisioner lacks "
            + ", ".join(missing)
        )
    frontend_package_path = nexuscrm_worktree / "frontend" / "package.json"
    try:
        frontend_package = json.loads(frontend_package_path.read_text())
    except (FileNotFoundError, json.JSONDecodeError) as exc:
        raise DeployError(f"invalid selected NexusCRM frontend package: {frontend_package_path}: {exc}") from exc
    scripts = frontend_package.get("scripts") if isinstance(frontend_package, dict) else None
    prebuild = str(scripts.get("prebuild", "")) if isinstance(scripts, dict) else ""
    provision_command = (
        str(scripts.get("provision:frontend-app-packages", "")) if isinstance(scripts, dict) else ""
    )
    if "provision:frontend-app-packages" not in prebuild or "provision-frontend-app-packages.mjs" not in provision_command:
        raise DeployError("selected NexusCRM frontend build does not wire frontend app-pack provisioning into prebuild")
    return True


def frontend_app_packs_source_materializer(downstream_build_root: Path) -> Path:
    candidates = [
        Path(__file__).resolve().parents[3] / "scripts" / "materialize_frontend_app_packs_source.py",
        downstream_build_root / "scripts" / "materialize_frontend_app_packs_source.py",
    ]
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    raise DeployError(f"frontend app-pack source materializer not found: {candidates}")


def materialize_frontend_app_packs_source(
    downstream_build_root: Path,
    app_packs_root: Path,
    nexuscrm_worktree: Path,
) -> tuple[Path, str]:
    materializer = frontend_app_packs_source_materializer(downstream_build_root)
    target_root = nexuscrm_worktree.parent / f"{nexuscrm_worktree.name}-frontend-app-packs-source"
    output = run(
        [
            sys.executable,
            str(materializer),
            "--downstream-root",
            str(downstream_build_root),
            "--official-app-packs-root",
            str(app_packs_root),
            "--target-root",
            str(target_root),
        ],
        capture=True,
    )
    try:
        payload = json.loads(output)
    except json.JSONDecodeError as exc:
        raise DeployError(f"invalid frontend app-pack source materializer output: {exc}") from exc
    if not isinstance(payload, dict):
        raise DeployError("invalid frontend app-pack source materializer output: expected object")
    resolved_root = Path(str(payload.get("root", ""))).expanduser().resolve()
    resolved_ref = str(payload.get("ref", "")).strip()
    if resolved_root != target_root.resolve() or not GIT_COMMIT_RE.fullmatch(resolved_ref):
        raise DeployError(
            "invalid frontend app-pack source materializer result: "
            f"root={resolved_root} ref={resolved_ref!r}"
        )
    run(["git", "-C", str(resolved_root), "cat-file", "-e", f"{resolved_ref}^{{commit}}"])
    return resolved_root, resolved_ref


def prepare_frontend_build(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    *,
    downstream_build_root: Path,
    app_packs_root: Path,
) -> tuple[Path, dict[str, str], tuple[Path, str] | None] | None:
    if bool_env(envv, "SKIP_FRONTEND"):
        return None
    frontend_app_packs_source: tuple[Path, str] | None = None
    if supports_synthetic_frontend_app_packs_source(downstream_build_root, nexuscrm_worktree):
        frontend_app_packs_source = materialize_frontend_app_packs_source(
            downstream_build_root,
            app_packs_root,
            nexuscrm_worktree,
        )
        frontend_app_packs_root, frontend_app_packs_ref = frontend_app_packs_source
        envv["NEXUSCRM_FRONTEND_APP_PACKS_ROOT"] = str(frontend_app_packs_root)
        envv["NEXUSCRM_FRONTEND_APP_PACKS_REF"] = frontend_app_packs_ref
    else:
        log("selected final pack predates frontend source manifests; using legacy frontend build inputs")
    frontend_dir = nexuscrm_worktree / "frontend"
    build_env = {
        "REACT_APP_BACKEND_URL": "",
        "REACT_APP_FRONTEND_URL": "",
        "API_BASE_URL": "",
        "FRONTEND_URL": "",
    }
    build_env.update(optional_process_env(envv))
    if frontend_app_packs_source:
        frontend_app_packs_root, frontend_app_packs_ref = frontend_app_packs_source
        build_env["NEXUSCRM_FRONTEND_APP_PACKS_ROOT"] = str(frontend_app_packs_root)
        build_env["NEXUSCRM_FRONTEND_APP_PACKS_REF"] = frontend_app_packs_ref
    else:
        official_app_packs_ref = envv.get("OFFICIAL_APP_PACKS_REF", "").strip()
        if official_app_packs_ref:
            build_env["NEXUSCRM_FRONTEND_APP_PACKS_REF"] = official_app_packs_ref
    google_client_id = envv.get("REACT_APP_GOOGLE_CLIENT_ID") or envv.get("GOOGLE_CLIENT_ID")
    if google_client_id:
        build_env["REACT_APP_GOOGLE_CLIENT_ID"] = google_client_id
    return frontend_dir, build_env, frontend_app_packs_source


def build_frontend(
    envv: dict[str, str],
    nexuscrm_worktree: Path,
    frontend_build_dist: Path,
    frontend_dist: Path,
    *,
    downstream_build_root: Path,
    app_packs_root: Path,
) -> tuple[Path, str] | None:
    prepared = prepare_frontend_build(
        envv,
        nexuscrm_worktree,
        downstream_build_root=downstream_build_root,
        app_packs_root=app_packs_root,
    )
    if prepared is None:
        return None
    frontend_dir, build_env, frontend_app_packs_source = prepared
    source_status = run(
        [
            "git",
            "-C",
            str(nexuscrm_worktree),
            "status",
            "--porcelain=v1",
            "--untracked-files=normal",
            "--",
            "frontend",
        ],
        capture=True,
    ).strip()
    if source_status:
        raise DeployError("frontend build source tree is not clean")
    version_path = frontend_dir / "public" / "version.json"
    if version_path.is_symlink() or (
        version_path.exists() and not version_path.is_file()
    ):
        raise DeployError(
            f"frontend build version source must be a regular file: {version_path}"
        )
    version_before = version_path.read_bytes() if version_path.is_file() else None
    try:
        if (frontend_dir / "package-lock.json").is_file():
            run(
                ["npm", "ci", "--include=dev"],
                cwd=frontend_dir,
                env={"NODE_ENV": "development"},
            )
        else:
            run(
                ["npm", "install", "--include=dev"],
                cwd=frontend_dir,
                env={"NODE_ENV": "development"},
            )
        run(["npm", "run", "build"], cwd=frontend_dir, env=build_env)
    finally:
        if version_before is None:
            if version_path.is_symlink() or (
                version_path.exists() and not version_path.is_file()
            ):
                raise DeployError(
                    "frontend build created an invalid version source path"
                )
            version_path.unlink(missing_ok=True)
        else:
            version_path.parent.mkdir(parents=True, exist_ok=True)
            with tempfile.NamedTemporaryFile(
                dir=version_path.parent,
                prefix=f".{version_path.name}.restore-",
                delete=False,
            ) as handle:
                restore_path = Path(handle.name)
                handle.write(version_before)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(restore_path, version_path)
    source_status = run(
        [
            "git",
            "-C",
            str(nexuscrm_worktree),
            "status",
            "--porcelain=v1",
            "--untracked-files=normal",
            "--",
            "frontend",
        ],
        capture=True,
    ).strip()
    if source_status:
        raise DeployError("frontend build mutated the frontend source tree")
    publish_frontend_dist(
        frontend_dir / "dist",
        frontend_build_dist,
        frontend_dist,
        nexuscrm_worktree.name,
    )
    return frontend_app_packs_source


def publish_frontend_dist(source_dist: Path, frontend_build_dist: Path, frontend_dist: Path, release_name: str) -> Path:
    if not source_dist.is_dir():
        raise DeployError(f"frontend build output not found: {source_dist}")
    frontend_build_dist.mkdir(parents=True, exist_ok=True)
    release_dir = frontend_build_dist / release_name
    if release_dir.exists():
        shutil.rmtree(release_dir)
    shutil.copytree(source_dist, release_dir)
    verify_frontend_dist(release_dir)
    publish_symlink(frontend_dist, release_dir)
    assert_frontend_target_matches_release(frontend_dist, frontend_build_dist, release_name)
    prune_old_frontend_releases(frontend_build_dist, keep=3, active_release=release_dir)
    return release_dir


def publish_symlink(link_path: Path, target_path: Path) -> None:
    link_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_link = link_path.parent / f".{link_path.name}.tmp"
    if tmp_link.exists() or tmp_link.is_symlink():
        tmp_link.unlink()
    relative_target = os.path.relpath(target_path, start=link_path.parent)
    tmp_link.symlink_to(relative_target)
    if link_path.exists() and not link_path.is_symlink():
        if link_path.is_dir():
            shutil.rmtree(link_path)
        else:
            link_path.unlink()
    elif link_path.is_symlink():
        link_path.unlink()
    os.replace(tmp_link, link_path)


def prune_old_frontend_releases(frontend_build_dist: Path, keep: int, active_release: Path) -> None:
    builds = sorted(
        (
            path
            for path in frontend_build_dist.iterdir()
            if path.is_dir()
        ),
        key=lambda path: path.name,
    )
    stale_builds = [path for path in builds if path != active_release]
    for stale_path in stale_builds[:-keep]:
        shutil.rmtree(stale_path)


def require_frontend_entrypoint(frontend_dist: Path) -> None:
    index_path = frontend_dist / "index.html"
    if not frontend_dist.is_dir():
        raise DeployError(f"frontend dist not found: {frontend_dist}")
    if not index_path.is_file():
        raise DeployError(f"frontend entrypoint not found: {index_path}")


def verify_frontend_dist_layout(frontend_dist: Path) -> None:
    require_frontend_entrypoint(frontend_dist)
    version_path = frontend_dist / "version.json"
    if not version_path.is_file():
        raise DeployError(f"frontend version manifest not found: {version_path}")


def verify_frontend_dist(frontend_dist: Path) -> None:
    verify_frontend_dist_layout(frontend_dist)
    forbidden_origins = (
        b"https://nexuscrm-new-staging.pingcap.com",
        b"https://nexuscrm-new-prod.pingcap.com",
    )
    for path in frontend_dist.rglob("*"):
        if not path.is_file() or path.suffix not in {".html", ".js", ".css"}:
            continue
        payload = path.read_bytes()
        if any(origin in payload for origin in forbidden_origins):
            raise DeployError(
                "frontend dist contains an environment-bound NexusCRM origin: "
                f"{path.relative_to(frontend_dist)}"
            )


def resolve_frontend_dist_target(frontend_dist: Path) -> Path:
    if not frontend_dist.is_symlink():
        raise DeployError(f"frontend-dist must be a symlink after publish: {frontend_dist}")
    target = frontend_dist.resolve()
    verify_frontend_dist_layout(target)
    return target


def assert_frontend_target_matches_release(
    frontend_dist: Path,
    frontend_build_dist: Path,
    release_name: str,
) -> Path:
    target = resolve_frontend_dist_target(frontend_dist)
    expected = (frontend_build_dist / release_name).resolve()
    if target != expected:
        raise DeployError(
            f"frontend-dist target mismatch: {frontend_dist} -> {target}, expected {expected}"
        )
    return target


def require_backend_runner(env_root: Path) -> Path:
    runner = env_root / "run_backend.sh"
    if not runner.is_file():
        raise DeployError(f"backend runner not found: {runner}")
    if not os.access(runner, os.X_OK):
        raise DeployError(f"backend runner is not executable: {runner}")
    return runner




def render_template(template_path: Path, replacements: dict[str, str]) -> str:
    content = template_path.read_text()
    for key, value in replacements.items():
        content = content.replace(key, value)
    return content


def legacy_hubspot_ingress_nginx_mode(envv: dict[str, str]) -> str:
    override = envv.get("HUBSPOT_INGRESS_NGINX_BLOCK", "").strip().lower()
    if override:
        return "block" if override in {"1", "true", "yes", "on"} else "proxy"
    return "proxy"


def hubspot_ingress_nginx_modes(envv: dict[str, str]) -> tuple[str, str]:
    legacy_mode = legacy_hubspot_ingress_nginx_mode(envv)
    ordinary_mode = envv.get("HUBSPOT_ORDINARY_WEBHOOK_MODE", "").strip().lower() or legacy_mode
    workflow_mode = envv.get("HUBSPOT_WORKFLOW_EXECUTE_MODE", "").strip().lower() or legacy_mode
    if ordinary_mode not in {"block", "proxy", "ack_drop"}:
        raise DeployError(
            "HUBSPOT_ORDINARY_WEBHOOK_MODE must be one of block, proxy, ack_drop"
        )
    if workflow_mode not in {"block", "proxy"}:
        raise DeployError("HUBSPOT_WORKFLOW_EXECUTE_MODE must be one of block, proxy")
    return ordinary_mode, workflow_mode


def apply_hubspot_ingress_nginx_modes(envv: dict[str, str], rendered: str) -> str:
    def remove_location_block(content: str, header_pattern: str) -> str:
        pattern = re.compile(r"\n?    location " + header_pattern + r" \{\n(?:.*?\n)*?    \}\n", re.MULTILINE)
        return pattern.sub("\n", content)

    ordinary_mode, workflow_mode = hubspot_ingress_nginx_modes(envv)
    rendered = remove_location_block(rendered, r"\^~ /webhooks/hubspot-crm-connector/")
    if ordinary_mode != "proxy":
        ordinary_status = 204 if ordinary_mode == "ack_drop" else 403
        webhook_block = f"""    location ^~ /webhooks/hubspot-crm-connector/ {{
        return {ordinary_status};
    }}

"""
        webhook_needle = "    location /webhooks/ {\n"
        if webhook_needle not in rendered:
            raise DeployError("nginx template missing generic /webhooks/ location")
        rendered = rendered.replace(webhook_needle, webhook_block + webhook_needle, 1)

    if workflow_mode == "block":
        rendered = remove_location_block(rendered, r"= /hubspot/workflow-actions/execute")
        execute_block = """    location = /hubspot/workflow-actions/execute {
        return 403;
    }

"""
        health_needle = "    location /health {\n"
        if health_needle not in rendered:
            raise DeployError("nginx template missing /health location")
        rendered = rendered.replace(health_needle, execute_block + health_needle, 1)
    return rendered


def apply_hubspot_ingress_nginx_block(envv: dict[str, str], rendered: str) -> str:
    return apply_hubspot_ingress_nginx_modes(envv, rendered)


def sudo_write_text(target_path: Path, content: str) -> None:
    run(["sudo", "tee", str(target_path)], stdin=content)


def sudo_write_text_atomic(
    target_path: Path,
    content: str,
    *,
    mode: str | None = None,
    execute: Callable[..., None] | None = None,
) -> None:
    execute = execute or run
    target_path = target_path.resolve(strict=False)
    tmp_path = target_path.parent / f".{target_path.name}.tmp"
    execute(["sudo", "mkdir", "-p", str(target_path.parent)])
    execute(["sudo", "tee", str(tmp_path)], stdin=content)
    if mode is not None:
        execute(["sudo", "chmod", mode, str(tmp_path)])
    execute(["sudo", "mv", str(tmp_path), str(target_path)])




def install_nginx_conf(envv: dict[str, str], framework_dir: Path, frontend_dist: Path) -> None:
    if not bool_env(envv, "INSTALL_NGINX", True):
        return
    nginx_path = Path("/etc/nginx/conf.d") / f"nexuscrm-{envv['ENV_NAME']}.conf"
    log_basename = envv.get("NGINX_LOG_BASENAME", envv["ENV_NAME"])
    rendered = apply_hubspot_ingress_nginx_modes(
        envv,
        render_template(
        framework_dir / "nginx/nexuscrm-env.conf.template",
        {
            "__NGINX_LISTEN_PORT__": envv.get("NGINX_LISTEN_PORT", "80"),
            "__NGINX_FORWARDED_PROTO__": envv.get("NGINX_FORWARDED_PROTO") or urlparse(envv["PUBLIC_BASE_URL"]).scheme or "http",
            "__SERVER_NAME__": envv["SERVER_NAME"],
            "__FRONTEND_DIST__": str(frontend_dist),
            "__BACKEND_PORT__": envv["BACKEND_PORT"],
            "__CONNECTED_MAIL_INGRESS_BLOCK__": "",
            "__WEBHOOK_PORT__": envv["BACKEND_PORT"],
            "__HEALTH_PORT__": envv["BACKEND_PORT"],
            "__HEALTH_PATH__": "",
            "__EXP_HUBSPOT_OPTIONS_ADAPTER_PORT__": exp_hubspot_options_adapter_port(envv),
            "__ACCESS_LOG__": f"/var/log/nginx/nexuscrm-{log_basename}.access.log",
            "__ERROR_LOG__": f"/var/log/nginx/nexuscrm-{log_basename}.error.log",
        },
        ),
    )
    activate_nginx_conf(
        nginx_path,
        rendered,
        healthcheck_url=f"http://127.0.0.1:{envv.get('NGINX_LISTEN_PORT', '80')}/health",
        healthcheck_headers={"Host": envv.get("HEALTHCHECK_HOST", envv["SERVER_NAME"].split()[0])},
    )


def activate_nginx_conf(
    nginx_path: Path,
    rendered: str,
    *,
    healthcheck_url: str | None = None,
    healthcheck_headers: dict[str, str] | None = None,
    verify_urls: list[tuple[str, dict[str, str] | None]] | None = None,
    execute: Callable[..., None] | None = None,
    reload_nginx: Callable[[], None] | None = None,
) -> None:
    # Nginx configuration is public routing data; the non-root operator must
    # still be able to verify it when its private authority files use umask 077.
    sudo_write_text_atomic(nginx_path, rendered, mode="0644", execute=execute)
    (reload_nginx or reload_nginx_service)()
    if healthcheck_url:
        wait_for_health(healthcheck_url, headers=healthcheck_headers)
    if verify_urls:
        for url, headers in verify_urls:
            wait_for_http_ok(url, headers=headers)


def reload_nginx_service() -> None:
    run(["sudo", "nginx", "-t"])
    active = subprocess.run(["sudo", "systemctl", "is-active", "--quiet", "nginx"], check=False)
    if active.returncode == 0:
        run(["sudo", "systemctl", "reload", "nginx"])
    else:
        run(["sudo", "systemctl", "start", "nginx"])


def runtime_route_probe_objects(envv: dict[str, str]) -> list[str]:
    raw = envv.get("RUNTIME_ROUTE_PROBE_OBJECTS", "").strip()
    if not raw:
        return list(DEFAULT_RUNTIME_ROUTE_PROBE_OBJECTS)
    objects = [item.strip() for item in raw.split(",") if item.strip()]
    if not objects:
        raise DeployError("RUNTIME_ROUTE_PROBE_OBJECTS is set but empty after parsing")
    return objects


def oauth_password_grant(
    base_url: str,
    *,
    username: str,
    password: str,
    client_id: str = "nexuscrm-ui",
    client_secret: str = "nexuscrm-ui-internal",
    host_header: str | None = None,
) -> dict[str, object]:
    payload = urlencode(
        {
            "grant_type": "password",
            "client_id": client_id,
            "client_secret": client_secret,
            "username": username,
            "password": password,
        }
    ).encode()
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    if host_header:
        headers["Host"] = host_header
    request = Request(
        f"{base_url.rstrip('/')}/services/oauth2/token",
        data=payload,
        headers=headers,
        method="POST",
    )
    try:
        with urlopen(request, timeout=30) as response:
            return json.loads(response.read().decode())
    except HTTPError as exc:
        body = exc.read().decode(errors="replace")
        raise DeployError(f"oauth password grant failed for {username}: http {exc.code}: {body[:500]}") from exc
    except URLError as exc:
        raise DeployError(f"oauth password grant failed for {username}: {exc}") from exc


def describe_object(
    base_url: str,
    *,
    object_api_name: str,
    access_token: str,
    host_header: str | None = None,
) -> dict[str, object]:
    headers = {"Authorization": f"Bearer {access_token}"}
    if host_header:
        headers["Host"] = host_header
    request = Request(
        f"{base_url.rstrip('/')}/services/data/v63.0/sobjects/{object_api_name}/describe",
        headers=headers,
        method="GET",
    )
    with urlopen(request, timeout=30) as response:
        return json.loads(response.read().decode())


def describe_fingerprint(payload: dict[str, object]) -> dict[str, object]:
    raw_fields = payload.get("fields")
    fields = raw_fields if isinstance(raw_fields, list) else []
    field_names: list[str] = []
    for item in fields:
        if not isinstance(item, dict):
            continue
        name = item.get("name")
        if isinstance(name, str) and name:
            field_names.append(name)
    raw_name_fields = payload.get("nameFields")
    name_fields = raw_name_fields if isinstance(raw_name_fields, list) else []
    return {
        "fieldCount": len(field_names),
        "fieldNames": field_names,
        "displayNameField": payload.get("displayNameField"),
        "nameFields": name_fields,
    }


def verify_nginx_backend_route_alignment(envv: dict[str, str]) -> None:
    host_header = envv.get("HEALTHCHECK_HOST", "").strip() or envv["SERVER_NAME"].split()[0]
    backend_base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    nginx_base_url = f"http://127.0.0.1:{envv.get('NGINX_LISTEN_PORT', '80')}"
    verify_username = envv.get("VERIFY_USERNAME", "").strip() or "admin@example.invalid"
    verify_password = envv["NEXUS_ADMIN_PASSWORD"]

    backend_token = oauth_password_grant(
        backend_base_url,
        username=verify_username,
        password=verify_password,
        host_header=host_header,
    )
    nginx_token = oauth_password_grant(
        nginx_base_url,
        username=verify_username,
        password=verify_password,
        host_header=host_header,
    )
    backend_access_token = str(backend_token.get("access_token", "")).strip()
    nginx_access_token = str(nginx_token.get("access_token", "")).strip()
    if not backend_access_token:
        raise DeployError("backend route alignment check missing backend access_token")
    if not nginx_access_token:
        raise DeployError("backend route alignment check missing nginx access_token")

    mismatches: list[str] = []
    for object_api_name in runtime_route_probe_objects(envv):
        backend_fingerprint = describe_fingerprint(
            describe_object(
                backend_base_url,
                object_api_name=object_api_name,
                access_token=backend_access_token,
                host_header=host_header,
            )
        )
        nginx_fingerprint = describe_fingerprint(
            describe_object(
                nginx_base_url,
                object_api_name=object_api_name,
                access_token=nginx_access_token,
                host_header=host_header,
            )
        )
        if backend_fingerprint != nginx_fingerprint:
            mismatches.append(
                f"{object_api_name}: backend={json.dumps(backend_fingerprint, sort_keys=True)} "
                f"nginx={json.dumps(nginx_fingerprint, sort_keys=True)}"
            )
    if mismatches:
        raise DeployError("nginx/backend route alignment mismatch:\n" + "\n".join(mismatches))




def tcp_listener_pids(port: str, label: str) -> list[str]:
    port = str(port).strip()
    if not port:
        return []
    if not port.isdigit():
        raise DeployError(f"unsafe {label} value: {port}")

    pids: set[str] = set()
    if shutil.which("lsof"):
        result = subprocess.run(
            ["sudo", "lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"],
            capture_output=True,
            text=True,
            check=False,
        )
        for line in result.stdout.splitlines():
            pid = line.strip()
            if pid.isdigit():
                pids.add(pid)
    if not pids and shutil.which("ss"):
        result = subprocess.run(
            ["sudo", "ss", "-ltnp"],
            capture_output=True,
            text=True,
            check=False,
        )
        for line in result.stdout.splitlines():
            if "LISTEN" not in line or f":{port}" not in line:
                continue
            for pid in re.findall(r"pid=(\d+)", line):
                pids.add(pid)
    return sorted(pids, key=int)


def backend_port_listener_pids(envv: dict[str, str]) -> list[str]:
    return tcp_listener_pids(str(envv.get("BACKEND_PORT", "")).strip(), "BACKEND_PORT")


def release_tcp_port(port: str, label: str) -> None:
    pids = tcp_listener_pids(port, label)
    if not pids:
        return
    log(f"releasing stale {label} listener(s) on port {port}: {', '.join(pids)}")
    subprocess.run(["sudo", "kill", *pids], check=False)
    time.sleep(1)
    remaining = tcp_listener_pids(port, label)
    if remaining:
        subprocess.run(["sudo", "kill", "-9", *remaining], check=False)
        time.sleep(1)


def release_backend_port(envv: dict[str, str]) -> None:
    release_tcp_port(str(envv.get("BACKEND_PORT", "")).strip(), "backend")


def stop_backend(envv: dict[str, str], run_dir: Path) -> None:
    """Stop the env's backend process, idempotent and best-effort.

    Used before RESET_DB so the database is not reset under a live backend.
    Quiesces retained predecessor processes: respects SKIP_BACKEND_START, handles both
    the non-systemd (pid-file) and systemd branches.  All stop operations are
    non-fatal — the service or process may legitimately be absent on a fresh host.
    """
    if bool_env(envv, "SKIP_BACKEND_START"):
        return
    if not bool_env(envv, "INSTALL_SYSTEMD", True):
        pid_file = run_dir / "backend.pid"
        if pid_file.is_file():
            old_pid = pid_file.read_text().strip()
            if old_pid:
                subprocess.run(["kill", old_pid], check=False)
            remove_path(pid_file)
        release_backend_port(envv)
        return
    # systemd branch: non-fatal — the unit may not exist yet on a fresh host.
    svc = f"nexuscrm-{envv['ENV_NAME']}-backend.service"
    subprocess.run(["sudo", "systemctl", "stop", svc], check=False)
    release_backend_port(envv)


def wait_for_health(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    attempts: int = 60,
    timeout_seconds: float | None = None,
    request_timeout_seconds: float = 10.0,
    failure_check: Callable[[], str | None] | None = None,
) -> None:
    timeout_budget_seconds = (
        attempts * 2.0 if timeout_seconds is None else timeout_seconds
    )
    deadline = time.monotonic() + max(0.1, timeout_budget_seconds)
    for _ in range(attempts):
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            break
        if failure_check is not None:
            failure = failure_check()
            if failure:
                raise DeployError(failure)
        request_timeout = min(max(0.1, request_timeout_seconds), remaining_seconds)
        curl_args = [
            "curl",
            "-fsS",
            "--connect-timeout",
            f"{min(5.0, request_timeout):.3f}",
            "--max-time",
            f"{request_timeout:.3f}",
        ]
        if headers:
            for key, value in headers.items():
                curl_args += ["-H", f"{key}: {value}"]
        curl_args.append(url)
        try:
            result = subprocess.run(
                curl_args,
                capture_output=True,
                text=True,
                check=False,
                timeout=request_timeout,
            )
        except subprocess.TimeoutExpired:
            result = None
        except OSError as exc:
            raise DeployError(f"healthcheck command failed: {url}: {exc}") from exc
        if result is not None and result.returncode == 0:
            if failure_check is not None:
                failure = failure_check()
                if failure:
                    raise DeployError(failure)
            return
        if failure_check is not None:
            failure = failure_check()
            if failure:
                raise DeployError(failure)
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            break
        time.sleep(min(2.0, remaining_seconds))
    raise DeployError(f"healthcheck failed: {url}")


def parse_systemd_show_properties(output: str) -> dict[str, str]:
    properties: dict[str, str] = {}
    for raw_line in output.splitlines():
        if "=" not in raw_line:
            continue
        key, value = raw_line.split("=", 1)
        key = key.strip()
        if key:
            properties[key] = value.strip()
    return properties


def backend_systemd_start_failure(envv: dict[str, str]) -> str | None:
    if not bool_env(envv, "INSTALL_SYSTEMD", True):
        return None
    service = f"nexuscrm-{envv['ENV_NAME']}-backend.service"
    property_names = (
        "ActiveState",
        "SubState",
        "Result",
        "ExecMainCode",
        "ExecMainStatus",
        "NRestarts",
    )
    args = ["sudo", "systemctl", "show", "--no-pager"]
    args.extend(f"--property={name}" for name in property_names)
    args.append(service)
    try:
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
            check=False,
            timeout=2,
        )
    except subprocess.TimeoutExpired:
        return f"backend service {service} state inspection timed out during startup"
    except OSError:
        return f"backend service {service} state inspection command failed during startup"
    if result.returncode != 0:
        return (
            f"backend service {service} state inspection failed during startup: "
            f"exit={result.returncode}"
        )

    properties = parse_systemd_show_properties(result.stdout)
    missing = [name for name in property_names if name not in properties]
    if missing:
        return f"backend service {service} state inspection omitted: {', '.join(missing)}"
    try:
        restart_count = int(properties["NRestarts"])
    except ValueError:
        return (
            f"backend service {service} reported invalid "
            f"NRestarts={properties['NRestarts']!r}"
        )

    active_state = properties["ActiveState"]
    sub_state = properties["SubState"]
    result_state = properties["Result"]
    exec_main_status = properties["ExecMainStatus"]
    failed = (
        restart_count > 0
        or active_state in {"failed", "inactive"}
        or sub_state in {"auto-restart", "dead", "failed"}
        or result_state not in {"", "success"}
        or exec_main_status not in {"", "0"}
    )
    if not failed:
        return None
    detail = ", ".join(f"{name}={properties[name]}" for name in property_names)
    return f"backend service {service} exited during startup: {detail}"


def stop_systemd_backend_after_failed_health(envv: dict[str, str]) -> str | None:
    if not bool_env(envv, "INSTALL_SYSTEMD", True):
        return None
    service = f"nexuscrm-{envv['ENV_NAME']}-backend.service"
    try:
        result = subprocess.run(
            ["sudo", "systemctl", "stop", service],
            capture_output=True,
            text=True,
            check=False,
            timeout=20,
        )
    except subprocess.TimeoutExpired:
        return f"failed to stop rejected backend service {service}: command timed out"
    except OSError as exc:
        return f"failed to stop rejected backend service {service}: {exc}"
    if result.returncode != 0:
        detail = result.stderr.strip() or result.stdout.strip() or "no command output"
        return (
            f"failed to stop rejected backend service {service}: "
            f"exit={result.returncode}: {detail}"
        )
    return None


def tail_text_file(path: Path, *, max_bytes: int = 20_000) -> str:
    try:
        if not path.is_file():
            return ""
        with path.open("rb") as fh:
            try:
                fh.seek(0, os.SEEK_END)
                size = fh.tell()
                fh.seek(max(0, size - max_bytes), os.SEEK_SET)
            except OSError:
                fh.seek(0)
            return fh.read().decode("utf-8", errors="replace")
    except OSError:
        return ""


def run_diagnostic_command(args: list[str], *, timeout: int = 20) -> str:
    command = f"$ {' '.join(args)}"
    try:
        result = subprocess.run(args, capture_output=True, text=True, check=False, timeout=timeout)
    except subprocess.TimeoutExpired:
        return f"{command}\n(diagnostics command timed out after {timeout}s)"
    except OSError as exc:
        return f"{command}\n(diagnostics command failed: {exc})"
    output = "\n".join(part for part in [result.stdout.strip(), result.stderr.strip()] if part)
    if result.returncode != 0:
        return f"{command}\n(exit {result.returncode})\n{output}".strip()
    return f"{command}\n{output}".strip()


def systemd_unit_log_paths(service: str) -> list[Path]:
    try:
        result = subprocess.run(
            ["sudo", "systemctl", "cat", service, "--no-pager"],
            capture_output=True,
            text=True,
            check=False,
            timeout=20,
        )
    except (OSError, subprocess.TimeoutExpired):
        return []
    if result.returncode != 0:
        return []
    paths: list[Path] = []
    seen: set[Path] = set()
    for raw_line in result.stdout.splitlines():
        line = raw_line.strip()
        if not (line.startswith("StandardOutput=") or line.startswith("StandardError=")):
            continue
        value = line.split("=", 1)[1].strip().strip("\"'")
        for prefix in ("append:", "file:", "truncate:"):
            if value.startswith(prefix):
                value = value[len(prefix) :].strip().strip("\"'")
                break
        if not value.startswith("/"):
            continue
        path = Path(value)
        if path in seen:
            continue
        seen.add(path)
        paths.append(path)
    return paths


def backend_diagnostics_tail_bytes(envv: dict[str, str]) -> int:
    raw = str(envv.get("DEPLOY_BACKEND_DIAGNOSTICS_TAIL_BYTES", "80000")).strip()
    if not raw:
        return 80000
    try:
        return max(1, int(raw))
    except ValueError:
        return 80000


def backend_start_diagnostics(envv: dict[str, str], env_root: Path, log_dir: Path) -> str:
    sections: list[str] = []
    log_dir = resolve_backend_log_dir(envv, env_root)
    tail_bytes = backend_diagnostics_tail_bytes(envv)
    seen_log_paths: set[Path] = set()

    def append_log_tail(path: Path) -> None:
        path_key = path.resolve(strict=False)
        if path_key in seen_log_paths:
            return
        seen_log_paths.add(path_key)
        content = tail_text_file(path, max_bytes=tail_bytes)
        if content:
            sections.append(f"$ tail {path}\n{content.strip()}")

    if bool_env(envv, "INSTALL_SYSTEMD", True):
        service = f"nexuscrm-{envv['ENV_NAME']}-backend.service"
        sections.append(run_diagnostic_command(["sudo", "systemctl", "status", "--no-pager", "--full", service]))
        sections.append(run_diagnostic_command(["sudo", "journalctl", "-u", service, "--no-pager", "-n", "240"]))
        sections.append(run_diagnostic_command(["sudo", "systemctl", "cat", service, "--no-pager"]))
        for path in systemd_unit_log_paths(service):
            append_log_tail(path)

    for name in ["backend.stderr.log", "backend.stdout.log", "backend.log"]:
        append_log_tail(log_dir / name)

    runner = env_root / "run_backend.sh"
    if runner.is_file():
        sections.append(f"backend runner: {runner}")

    return "\n\n".join(section for section in sections if section.strip())


def wait_for_backend_health(
    envv: dict[str, str],
    env_root: Path,
    log_dir: Path,
    *,
    attempts: int = 60,
) -> None:
    url = f"http://127.0.0.1:{envv['BACKEND_PORT']}/health"
    try:
        wait_for_health(
            url,
            attempts=attempts,
            request_timeout_seconds=2.0,
            failure_check=lambda: backend_systemd_start_failure(envv),
        )
    except DeployError as exc:
        stop_failure = stop_systemd_backend_after_failed_health(envv)
        try:
            diagnostics = backend_start_diagnostics(envv, env_root, log_dir)
        except Exception as diagnostics_error:
            diagnostics = f"backend startup diagnostics unavailable: {diagnostics_error}"
        details = [str(exc)]
        if stop_failure:
            details.append(stop_failure)
        if diagnostics:
            details.append(f"backend startup diagnostics:\n{diagnostics}")
        raise DeployError("\n\n".join(details)) from exc


def wait_for_http_ok(url: str, *, headers: dict[str, str] | None = None, attempts: int = 60) -> None:
    curl_args = ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}"]
    if headers:
        for key, value in headers.items():
            curl_args += ["-H", f"{key}: {value}"]
    curl_args.append(url)
    for _ in range(attempts):
        result = subprocess.run(curl_args, capture_output=True, text=True, check=False)
        if result.returncode == 0 and result.stdout.strip() == "200":
            return
        time.sleep(2)
    raise DeployError(f"http check failed: {url}")