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

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

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}$")
PATCH_BUNDLE_MANIFEST_NAMES = (
    "patch-bundle.json",
    "patch-manifest.json",
)
PUBLIC_OAUTH_PKCE_ENV_KEY = "NEXUSCRM_PUBLIC_OAUTH_PKCE_ENABLED"
PACK_PAYLOAD_HASH_V1_RE = re.compile(r"^nexus-pack-sha256-v1:([0-9a-f]{64})$")
DEFAULT_SETUP_COMMAND_TIMEOUT_SECONDS = 1800
SETUP_COMMAND_DIAGNOSTIC_GRACE_SECONDS = 5
SETUP_COMMAND_OUTPUT_TAIL_LINES = 200
PROD_BACKEND_PROXY_RE = re.compile(r"proxy_pass\s+http://127\.0\.0\.1:(\d+)")
SS_LISTENER_PID_RE = re.compile(r"pid=(\d+)")
RELEASE_SETUP_GOOS = "linux"
RELEASE_SETUP_GOARCH = "amd64"
RELEASE_SETUP_GOAMD64 = "v1"
RELEASE_SETUP_CGO_ENABLED = "0"
RELEASE_SETUP_GOPROXY = "https://proxy.golang.org,direct"
RELEASE_SETUP_GOSUMDB = "sum.golang.org"
RELEASE_SETUP_PRIVATE_MODULE_PATTERN = "github.com/pingcap-inc/*"
RELEASE_SETUP_GIT_BINARY = "/usr/bin/git"
RELEASE_SETUP_GO_BINARY = "/usr/local/go/bin/go"


class DeployError(RuntimeError):
    pass


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


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(result.stdout)
            sys.stderr.write(result.stderr)
        raise DeployError(f"command failed ({result.returncode}): {' '.join(args)}")
    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 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
        key, value = line.split("=", 1)
        values[key] = 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 _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_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 = str(envv.get("GH_TOKEN") or os.environ.get("GH_TOKEN") or "").strip()
    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 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 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_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 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 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 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 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 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 release_tag_target_commit(repo_root: Path, tag_name: str) -> str:
    """Resolve an exact, syntactically valid release tag to its commit."""

    normalized_tag = tag_name.strip()
    ref = f"refs/tags/{normalized_tag}"
    if (
        not normalized_tag
        or subprocess.run(
            ["git", "check-ref-format", ref],
            text=True,
            capture_output=True,
            check=False,
        ).returncode
        != 0
    ):
        raise DeployError(f"invalid release tag name: {tag_name!r}")
    resolved = subprocess.run(
        [
            "git",
            "-C",
            str(repo_root.expanduser().resolve()),
            "rev-parse",
            "--verify",
            f"{ref}^{{commit}}",
        ],
        text=True,
        capture_output=True,
        check=False,
    )
    target_commit = resolved.stdout.strip()
    if resolved.returncode != 0 or not GIT_COMMIT_RE.fullmatch(target_commit):
        raise DeployError(f"release tag did not resolve to a commit SHA: {normalized_tag}")
    return target_commit


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", "")).strip()
    raw_sql_files = manifest.get("sql_files")
    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,
        "sql_files": sql_files,
        "refs": refs,
    }


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 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_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], {}

    ensure_distribution_app_packs_link(nexuscrm_worktree, app_packs_root)
    return (
        nexuscrm_worktree / "backend" / "distributions",
        ["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",
            ],
            {},
        )

    ensure_distribution_app_packs_link(nexuscrm_worktree, app_packs_root)
    return (
        nexuscrm_worktree / "backend" / "distributions",
        [
            "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"
    ensure_distribution_app_packs_link(nexuscrm_worktree, app_packs_root)
    return nexuscrm_worktree / "backend" / "distributions"


def configured_backend_setup_binary(envv: dict[str, str]) -> Path | None:
    raw_path = str(envv.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 release_setup_dependency_verification_commands(
    envv: dict[str, str],
) -> list[list[str]]:
    build_tags = supported_backend_go_build_tags(envv.get("BACKEND_GO_BUILD_TAGS"))
    list_command = ["go", "list", "-mod=readonly"]
    if build_tags:
        list_command.extend(["-tags", build_tags])
    list_command.extend(["-deps", "./cmd/setup"])
    return [list_command, ["go", "mod", "verify"]]


def release_setup_verification_env() -> dict[str, str]:
    return {
        "PATH": DEFAULT_PATH_PREFIX,
        "HOME": "/nonexistent",
        "XDG_CONFIG_HOME": "/nonexistent",
        "LANG": "C",
        "LC_ALL": "C",
        "GO111MODULE": "on",
        "GOENV": "off",
        "GOFLAGS": "",
        "GOTOOLCHAIN": "local",
        "GOWORK": "off",
        "GOAUTH": "off",
        "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",
    }


def release_setup_build_env(envv: dict[str, str], build_root: Path) -> dict[str, str]:
    home = build_root / "home"
    tmp = build_root / "tmp"
    go_cache = build_root / "go-build-cache"
    go_mod_cache = build_root / "go-module-cache"
    xdg_config = build_root / "xdg-config"
    for path in (home, tmp, go_cache, go_mod_cache, xdg_config):
        path.mkdir(mode=0o700)

    build_env = {
        "PATH": DEFAULT_PATH_PREFIX,
        "HOME": str(home),
        "TMPDIR": str(tmp),
        "XDG_CONFIG_HOME": str(xdg_config),
        "GO111MODULE": "on",
        "GOENV": "off",
        "GOFLAGS": "",
        "GOTOOLCHAIN": "local",
        "GOWORK": "off",
        "GOCACHE": str(go_cache),
        "GOMODCACHE": str(go_mod_cache),
        "GOPROXY": RELEASE_SETUP_GOPROXY,
        "GOSUMDB": RELEASE_SETUP_GOSUMDB,
        "GOPRIVATE": RELEASE_SETUP_PRIVATE_MODULE_PATTERN,
        "GONOPROXY": RELEASE_SETUP_PRIVATE_MODULE_PATTERN,
        "GONOSUMDB": RELEASE_SETUP_PRIVATE_MODULE_PATTERN,
        "GOINSECURE": "",
        "GOVCS": "public:git,private:git",
        "GOAUTH": "off",
        "GOOS": RELEASE_SETUP_GOOS,
        "GOARCH": RELEASE_SETUP_GOARCH,
        "GOAMD64": RELEASE_SETUP_GOAMD64,
        "CGO_ENABLED": RELEASE_SETUP_CGO_ENABLED,
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": "/dev/null",
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_ASKPASS": "/bin/false",
        "GIT_ALLOW_PROTOCOL": "https",
    }
    token = str(envv.get("GH_TOKEN") or "").strip()
    if token:
        build_env.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 build_env


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)
    temporary_bin = setup_bin.with_name(f".{setup_bin.name}.tmp-{os.getpid()}")
    if temporary_bin.exists() or temporary_bin.is_symlink():
        temporary_bin.unlink()
    setup_cwd, cmd, extra_env = backend_setup_binary_build_command(
        envv, nexuscrm_worktree, app_packs_root, temporary_bin
    )
    log(f"building release-scoped backend setup binary: {setup_bin}")
    try:
        with tempfile.TemporaryDirectory(
            prefix=f".{setup_bin.name}.build-", dir=setup_bin.parent
        ) as raw_build_root:
            build_env = release_setup_build_env(envv, Path(raw_build_root))
            build_env.update(extra_env)
            for verification_command in release_setup_dependency_verification_commands(
                envv
            ):
                verification = subprocess.run(
                    verification_command,
                    cwd=str(setup_cwd),
                    env=build_env,
                    text=True,
                    capture_output=True,
                    check=False,
                )
                if verification.returncode != 0:
                    raise DeployError(
                        "backend setup binary dependency verification failed "
                        f"({verification.returncode})"
                    )
            result = subprocess.run(
                cmd,
                cwd=str(setup_cwd),
                env=build_env,
                text=True,
                capture_output=True,
                check=False,
            )
            if result.returncode != 0:
                raise DeployError(
                    f"backend setup binary build failed ({result.returncode})"
                )
            if not temporary_bin.is_file() or not os.access(temporary_bin, os.X_OK):
                raise DeployError("built backend setup binary is not executable")
            os.replace(temporary_bin, setup_bin)
    finally:
        if temporary_bin.exists() or temporary_bin.is_symlink():
            temporary_bin.unlink()
    envv["NEXUSCRM_SETUP_BIN"] = str(setup_bin)
    return setup_bin


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:
    raw = envv.get("TIDB_TLS")
    if raw is None or raw == "":
        return True
    return raw.lower() in {"1", "true", "yes", "on"}


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


def db_exec(envv: dict[str, str], sql: str, *, database: str | None = None) -> None:
    env: dict[str, str] = {}
    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)
    # Stream SQL over stdin (and pass the password via MYSQL_PWD) instead of
    # argv: runtime pack metadata sync embeds large JSON declaration blobs that
    # overflow ARG_MAX on the client argv. Mirrors deployments/offline/runtime.
    run(args, env=env, stdin=sql + "\n")


def db_query_scalar(envv: dict[str, str], sql: str, *, database: str | None = None) -> str:
    env: dict[str, str] = {}
    args = mysql_client(envv) + [
        f"-h{envv['TIDB_HOST']}",
        f"-P{envv['TIDB_PORT']}",
        f"-u{envv['TIDB_USER']}",
        "-N",
        "-B",
    ]
    if envv.get("TIDB_PASSWORD"):
        env["MYSQL_PWD"] = envv["TIDB_PASSWORD"]
    if database:
        args.append(database)
    return run(args, capture=True, env=env, stdin=sql + "\n").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 installed_pack_table_exists(envv: dict[str, str]) -> bool:
    database = envv["TIDB_DATABASE"].replace("'", "''")
    count = db_query_scalar(
        envv,
        f"SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='{database}' AND table_name='_System_InstalledPack';",
        database="information_schema",
    )
    return count == "1"


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:
    curl_args = [
        "curl",
        "-sS",
        "--digest",
        "-u",
        f"{envv['TIDB_CLOUD_PUBLIC_KEY']}:{envv['TIDB_CLOUD_PRIVATE_KEY']}",
        "-X",
        method,
        f"{envv.get('TIDB_CLOUD_API_BASE_URL', 'https://serverless.tidbapi.com/v1beta1')}/clusters/{envv['TIDB_CLOUD_CLUSTER_ID']}{path}",
        "-H",
        "Content-Type: application/json",
        "-w",
        "\n%{http_code}",
    ]
    if body is not None:
        curl_args += ["-d", json.dumps(body)]
    result = subprocess.run(
        curl_args,
        text=True,
        capture_output=True,
        check=False,
    )
    safe_path = re.sub(r"/branches/[^/:]+", "/branches/<redacted>", path)
    if result.returncode != 0:
        raise DeployError(
            f"TiDB Cloud API {method} {safe_path} transport failed "
            f"(curl exit {result.returncode})"
        )
    output = result.stdout or ""
    if "\n" not in output:
        raise DeployError(f"unexpected TiDB Cloud API response for {method} {safe_path}")
    payload, status_text = output.rsplit("\n", 1)
    status = int(status_text.strip())
    accepted = allow_statuses or {200}
    if status not in accepted:
        raise DeployError(f"TiDB Cloud API {method} {safe_path} returned http {status}")
    if not payload.strip():
        return {"_http_status": status}
    try:
        parsed = json.loads(payload)
    except json.JSONDecodeError as exc:
        raise DeployError(f"TiDB Cloud API {method} {safe_path} returned invalid JSON") from exc
    if not isinstance(parsed, dict):
        raise DeployError(f"TiDB Cloud API {method} {safe_path} returned a non-object response")
    parsed["_http_status"] = status
    return parsed


def wait_for_tidb_branch_active(envv: dict[str, str], branch_id: str) -> dict:
    last_state = ""
    for _ in range(120):
        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 reset_tidb_branch_if_requested(envv: dict[str, str]) -> None:
    if not bool_env(envv, "RESET_TIDB_BRANCH"):
        return
    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}")
    tidb_cloud_api(envv, "POST", f"/branches/{branch_id}:reset", allow_statuses={200, 202, 409})
    wait_for_tidb_branch_active(envv, branch_id)
    if envv.get("TIDB_PASSWORD"):
        tidb_cloud_api(
            envv,
            "PATCH",
            f"/branches/{branch_id}/sqlUsers/root",
            {"password": envv["TIDB_PASSWORD"]},
        )
        wait_for_tidb_branch_active(envv, branch_id)


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 ensure_valid_git_worktree(worktree: Path, expected_ref: str) -> bool:
    if not worktree.is_dir():
        return False
    probe = subprocess.run(
        ["git", "-C", str(worktree), "rev-parse", "HEAD"],
        text=True,
        capture_output=True,
        check=False,
    )
    if probe.returncode != 0:
        shutil.rmtree(worktree)
        return False
    current = probe.stdout.strip()
    if current != expected_ref:
        shutil.rmtree(worktree)
        return False
    return True


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)
    ensure_valid_git_worktree(worktree, downstream_ref)
    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", "-fdx"])
    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)
    ensure_valid_git_worktree(worktree, upstream_ref)
    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_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 cleanup_stale_runtime_dirs(runtime_dir: Path, keep_paths: list[Path]) -> None:
    runtime_dir = runtime_dir.resolve()
    keep_resolved = {path.resolve() for path in keep_paths}
    managed_prefixes = ("downstream-", "nexuscrm-")

    for child in runtime_dir.iterdir():
        if not child.is_dir():
            continue
        if not child.name.startswith(managed_prefixes):
            continue
        if child.resolve() in keep_resolved:
            continue
        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:
    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,
    }
    run(
        [
            "bash",
            str(framework_dir / "runtime/tools/build_final_pack.sh"),
            "--reset",
        ],
        cwd=downstream_build_root,
        env=envv,
    )


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 shell quoting stripped.

    The prod runtime env file is sourced by bash runners, so the JSON value is
    stored single-quoted; the prod env loader keeps values verbatim.
    """
    raw = str(envv.get("NEXUSCRM_DOWNLOAD_ARTIFACTS", "")).strip()
    if len(raw) >= 2 and raw[0] == raw[-1] and raw[0] in {"'", '"'}:
        raw = raw[1:-1]
    return raw


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,
) -> 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 = connect_mcp_url_from_env(envv)
    out_dir = Path(envv["DEPLOY_ARTIFACT_DIR"]).expanduser().resolve() / "connect-download-artifacts"
    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 _file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def materialize_connect_download_artifacts_for_runtime(
    envv: dict[str, str],
    summary: dict[str, object],
    nexuscrm_worktree: Path,
    *,
    artifact_dir: Path,
) -> dict[str, object]:
    """Install built immutable downloads into this candidate's filesystem store."""
    version = str(summary.get("version", "")).strip()
    if not version:
        raise DeployError("connect artifact build summary is missing a version")
    catalog = connect_download_artifacts_catalog(version)
    envv["NEXUSCRM_DOWNLOAD_ARTIFACTS"] = connect_download_artifacts_env_value(version)

    storage_root_raw = str(
        envv.get("NEXUSCRM_DOWNLOAD_ARTIFACT_STORAGE_ROOT", "")
    ).strip()
    if not storage_root_raw:
        raise DeployError(
            "NEXUSCRM_DOWNLOAD_ARTIFACT_STORAGE_ROOT is required to materialize connect artifacts"
        )
    storage_root = Path(storage_root_raw).expanduser()
    candidate_root_raw = str(envv.get("ENV_ROOT", "")).strip()
    if not candidate_root_raw:
        raise DeployError("ENV_ROOT is required to materialize connect artifacts")
    candidate_root = Path(candidate_root_raw).expanduser().resolve()
    if not storage_root.is_absolute():
        storage_root = candidate_root / storage_root
    storage_root = storage_root.resolve()
    try:
        storage_root.relative_to(candidate_root)
    except ValueError as exc:
        raise DeployError(
            "ordinary Prod connect artifact storage must be candidate-local"
        ) from exc

    artifacts: dict[str, dict[str, str]] = {}
    for kind in CONNECT_DOWNLOAD_ARTIFACT_BASENAMES:
        artifact = summary.get(kind, {})
        source = Path(str(artifact.get("path", ""))).expanduser()
        expected_sha256 = str(artifact.get("sha256", "")).strip()
        if not source.is_file() or not SHA256_RE.fullmatch(expected_sha256):
            raise DeployError(f"invalid built connect artifact for {kind}: {artifact!r}")
        if _file_sha256(source) != expected_sha256:
            raise DeployError(f"built connect artifact sha256 mismatch for {kind}")

        key = catalog[kind]["key"]
        target = (storage_root / key).resolve()
        try:
            target.relative_to(storage_root)
        except ValueError as exc:
            raise DeployError(f"connect artifact key escapes filesystem storage root: {key}") from exc
        target.parent.mkdir(parents=True, exist_ok=True)
        file_descriptor, temporary_name = tempfile.mkstemp(
            prefix=".connect-artifact-",
            dir=str(target.parent),
        )
        os.close(file_descriptor)
        temporary_path = Path(temporary_name)
        try:
            shutil.copyfile(source, temporary_path)
            if _file_sha256(temporary_path) != expected_sha256:
                raise DeployError(f"materialized connect artifact sha256 mismatch for {kind}")
            os.replace(temporary_path, target)
        finally:
            temporary_path.unlink(missing_ok=True)

        artifacts[kind] = {
            "key": key,
            "path": str(target),
            "sha256": expected_sha256,
        }
        log(
            f"materialized connect download artifact kind={kind} version={version} "
            f"sha256={expected_sha256} key={key}"
        )

    receipt: dict[str, object] = {
        "version": version,
        "storage_root": str(storage_root),
        "artifacts": artifacts,
    }
    artifact_dir.mkdir(parents=True, exist_ok=True)
    (artifact_dir / "connect-download-artifacts-runtime.json").write_text(
        json.dumps(receipt, indent=2, sort_keys=True) + "\n"
    )
    return receipt


def verify_connect_download_artifacts_for_runtime(
    envv: dict[str, str],
    *,
    runtime_worktree: Path,
    artifact_dir: Path,
) -> dict[str, object]:
    """Rehash the exact candidate-local downloads before retained publication."""
    receipt_path = artifact_dir / "connect-download-artifacts-runtime.json"
    try:
        receipt = json.loads(receipt_path.read_text())
    except (FileNotFoundError, json.JSONDecodeError) as exc:
        raise DeployError(f"invalid connect artifact runtime receipt: {receipt_path}") from exc
    if not isinstance(receipt, dict):
        raise DeployError(f"connect artifact runtime receipt must be an object: {receipt_path}")

    version = str(receipt.get("version", "")).strip()
    catalog = connect_download_artifacts_catalog(version)
    if download_artifacts_env_json(envv) != connect_download_artifacts_env_value(version):
        raise DeployError("connect artifact runtime receipt does not match the candidate catalog")

    candidate_root_raw = str(envv.get("ENV_ROOT", "")).strip()
    if not candidate_root_raw:
        raise DeployError("ENV_ROOT is required to verify connect artifacts")
    candidate_root = Path(candidate_root_raw).expanduser().resolve()
    storage_root_raw = str(
        envv.get("NEXUSCRM_DOWNLOAD_ARTIFACT_STORAGE_ROOT", "")
    ).strip()
    if not storage_root_raw:
        raise DeployError(
            "NEXUSCRM_DOWNLOAD_ARTIFACT_STORAGE_ROOT is required to verify connect artifacts"
        )
    expected_storage_root = Path(storage_root_raw).expanduser()
    if not expected_storage_root.is_absolute():
        expected_storage_root = candidate_root / expected_storage_root
    expected_storage_root = expected_storage_root.resolve()
    try:
        expected_storage_root.relative_to(candidate_root)
    except ValueError as exc:
        raise DeployError(
            "ordinary Prod connect artifact storage must be candidate-local"
        ) from exc

    storage_root = Path(str(receipt.get("storage_root", ""))).expanduser().resolve()
    if storage_root != expected_storage_root:
        raise DeployError(
            "connect artifact runtime receipt artifact storage root does not match "
            "NEXUSCRM_DOWNLOAD_ARTIFACT_STORAGE_ROOT"
        )

    artifacts = receipt.get("artifacts")
    if not isinstance(artifacts, dict):
        raise DeployError("connect artifact runtime receipt is missing artifacts")
    for kind in CONNECT_DOWNLOAD_ARTIFACT_BASENAMES:
        artifact = artifacts.get(kind)
        if not isinstance(artifact, dict):
            raise DeployError(f"connect artifact runtime receipt is missing {kind}")
        key = str(artifact.get("key", "")).strip()
        if key != catalog[kind]["key"]:
            raise DeployError(f"connect artifact runtime receipt key mismatch for {kind}")
        expected_path = (storage_root / key).resolve()
        receipt_target = Path(str(artifact.get("path", ""))).expanduser().resolve()
        if receipt_target != expected_path:
            raise DeployError(f"connect artifact runtime receipt path mismatch for {kind}")
        expected_sha256 = str(artifact.get("sha256", "")).strip()
        if not SHA256_RE.fullmatch(expected_sha256) or not expected_path.is_file():
            raise DeployError(f"connect artifact runtime receipt is invalid for {kind}")
        if _file_sha256(expected_path) != expected_sha256:
            raise DeployError(f"connect artifact runtime sha256 mismatch for {kind}")

    log(f"verified candidate-local connect download artifacts from receipt: {receipt_path}")
    return receipt


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:
    skip_bootstrap = bool_env(envv, "SKIP_BOOTSTRAP")
    setup_env = os.environ.copy()
    setup_env.update(
        {
            "NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": str(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": envv["PUBLIC_BASE_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"),
            "NEXUS_ENV": "dev" if skip_bootstrap else "production",
            "NODE_ENV": "production",
            "LOG_FORMAT": "json",
        }
    )
    if envv.get("TIDB_PASSWORD"):
        setup_env["TIDB_PASSWORD"] = envv["TIDB_PASSWORD"]
    download_artifacts_catalog = download_artifacts_env_json(envv)
    if download_artifacts_catalog:
        # publish-artifact resolves its object-store key from the same catalog
        # the backend serves downloads from.
        setup_env["NEXUSCRM_DOWNLOAD_ARTIFACTS"] = download_artifacts_catalog
    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}")
        setup_cwd = backend_setup_workdir(envv, nexuscrm_worktree, app_packs_root)
        cmd = [str(setup_bin), *args]
    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
    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

    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
    # Keep prod aligned with staging: `go run` wraps child exit code 3 as
    # process exit 1 plus `exit status 3` on stderr for skip-backfill apply.
    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 DeployError(f"command failed ({result_code}): {' '.join(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 oauth_password_grant(
    base_url: str,
    *,
    username: str,
    password: str,
) -> dict[str, object]:
    payload = urlencode(
        {
            "grant_type": "password",
            "client_id": "nexuscrm-ui",
            "client_secret": "nexuscrm-ui-internal",
            "username": username,
            "password": password,
        }
    ).encode()
    request = Request(
        f"{base_url.rstrip('/')}/services/oauth2/token",
        data=payload,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    with urlopen(request, timeout=30) as response:
        return json.loads(response.read().decode())


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 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}',
                AppliedPayloadHash=CASE
                    WHEN COALESCE(TRIM(DesiredPayloadHash), '') <> '' THEN DesiredPayloadHash
                    ELSE AppliedPayloadHash
                END,
                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 normalize_installed_pack_runtime_state_before_setup_apply(
    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 load_installed_pack_payload_hash_state(envv: dict[str, str], pack_name: str) -> tuple[str, str]:
    escaped_name = pack_name.replace("'", "''")
    row = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT JSON_OBJECT(
                'desired_payload_hash', COALESCE(DesiredPayloadHash, ''),
                'applied_payload_hash', COALESCE(AppliedPayloadHash, '')
            )
            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("desired_payload_hash", "")).strip(),
        str(payload.get("applied_payload_hash", "")).strip(),
    )


def is_valid_pack_payload_hash_v1(value: str) -> bool:
    return bool(PACK_PAYLOAD_HASH_V1_RE.fullmatch(value.strip()))


def derive_stale_pack_payload_hash(current_hash: str) -> str:
    match = PACK_PAYLOAD_HASH_V1_RE.fullmatch(current_hash.strip())
    if match is None:
        raise DeployError(f"unexpected pack payload hash format: {current_hash!r}")
    digest = match.group(1)
    replacement = "0" if digest[-1] != "0" else "1"
    return f"nexus-pack-sha256-v1:{digest[:-1]}{replacement}"


def pre_apply_applied_payload_hash_sql(envv: dict[str, str], pack_name: str, current_hash: str) -> str:
    desired_payload_hash, applied_payload_hash = load_installed_pack_payload_hash_state(envv, pack_name)
    if (
        applied_payload_hash == desired_payload_hash
        and is_valid_pack_payload_hash_v1(applied_payload_hash)
        and applied_payload_hash != current_hash
    ):
        return "AppliedPayloadHash"
    # Warm prod retries must deterministically restore installed-pack drift.
    # If a failed run flattened AppliedPayloadHash to the current release hash,
    # backend verify will skip source-backed reapply on the next retry.
    return sql_string_literal(derive_stale_pack_payload_hash(current_hash))


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_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"],
    ).strip()


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:
    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_inbound_live_workitem_summary_for_packs(envv: dict[str, str], pack_names: Iterable[str]) -> str:
    """Summarize only WorkItems that still have a valid execution lease.

    Admission pause prevents new claims but must not deadlock an upgrade behind
    historical pending, blocked, retry, or lease-free processing rows. Those
    rows have no live executor and are recovered by the candidate runtime after
    installed-pack metadata is coherent. A pack mutation waits only for work
    still fenced by an unexpired owner/token lease.
    """
    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"):
        # A legacy runtime has no WorkItem lease authority. Retain the existing
        # Event drain rather than treating a live Event executor as empty.
        return connector_inbound_legacy_event_summary_for_packs(envv, pack_name_list)
    return db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT COALESCE(
                GROUP_CONCAT(
                    CONCAT(
                        pack_name,
                        '[',
                        pack_category,
                        '/work_item_processing]=',
                        row_count
                    )
                    ORDER BY pack_name, pack_category
                    SEPARATOR '; '
                ),
                ''
            )
            FROM (
                SELECT p.Name AS pack_name,
                       p.Category AS pack_category,
                       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, '') = 'processing'
                   AND TRIM(COALESCE(w.LeaseOwner, '')) <> ''
                   AND TRIM(COALESCE(w.LeaseToken, '')) <> ''
                   AND w.LeaseUntil IS NOT NULL
                   AND w.LeaseUntil > UTC_TIMESTAMP(6)
                 GROUP BY p.Name, p.Category
            ) active_connector_work_items;
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    ).strip()


def connector_inbound_legacy_event_gap_summary_for_packs(
    envv: dict[str, str], pack_name_list: str
) -> str:
    """Report a true Event-only executor gap after authoritative WorkItems drain.

    This mirrors the L0 WorkItem readiness fence. A source-projection Event at or
    before its selected projection boundary is covered even when the selected
    WorkItem is already terminal; an unrelated non-terminal Event is not.
    """
    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 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,
    *,
    active_leases_only: bool = False,
    keep_admission_paused_on_timeout: bool = False,
    drain_completion_context: str = "source-backed pack reapply",
) -> None:
    timeout_seconds = int(str(envv.get("DEPLOY_CONNECTOR_INBOUND_DRAIN_TIMEOUT_SECONDS", "1800")).strip() or "1800")
    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_live_workitem_summary_for_packs(envv, pack_names)
            if active_leases_only
            else connector_inbound_nonterminal_summary_for_packs(envv, pack_names)
        )
        if not summary:
            if last_summary:
                log(f"connector inbound execution drained before {drain_completion_context}")
            return
        last_summary = summary
        if time.monotonic() >= deadline:
            drain_scope = "active leased WorkItems" if active_leases_only else "non-terminal rows"
            timeout_action = (
                "Candidate WorkItem recovery is incomplete; keeping webhook admission paused "
                "and aborting target promotion."
                if keep_admission_paused_on_timeout
                else (
                    "Keep webhook admission paused only if an operator is actively investigating; "
                    "otherwise resume admission and fix the drain blocker."
                )
            )
            raise DeployError(
                f"connector inbound queue still has {drain_scope} after connector admission gate pause: "
                f"{summary}. {timeout_action}"
            )
        log(f"waiting for connector inbound execution to drain during connector admission gate pause: {summary}")
        time.sleep(min(poll_seconds, max(0.0, deadline - time.monotonic())))


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 provision_hubspot_private_app_named_credential_token(envv: dict[str, str]) -> 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 = db_query_scalar(
        envv,
        "SELECT Id FROM _System_NamedCredential WHERE DeveloperName='HubSpotPrivateApp' LIMIT 1;",
        database=envv["TIDB_DATABASE"],
    )
    if not credential_id:
        raise DeployError("cannot provision HubSpot NamedCredential token: HubSpotPrivateApp definition is missing")
    ciphertext = encrypt_named_credential_secret(access_token, master_key)
    masked_hint = named_credential_secret_masked_hint(access_token)
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            INSERT INTO _System_NamedCredentialSecret
                (Id, NamedCredentialId, SecretKind, SecretValue, MaskedHint,
                 RotatedAt, CreatedDate, LastModifiedDate)
            VALUES
                ('nc_secret_hubspot_crm_connector_private_app_token',
                 {sql_string_literal(credential_id)}, 'Token',
                 {sql_string_literal(ciphertext)}, {sql_string_literal(masked_hint)},
                 UTC_TIMESTAMP(), UTC_TIMESTAMP(), UTC_TIMESTAMP())
            ON DUPLICATE KEY UPDATE
                SecretValue = VALUES(SecretValue),
                MaskedHint = VALUES(MaskedHint),
                RotatedAt = VALUES(RotatedAt),
                LastModifiedDate = VALUES(LastModifiedDate);
            """
        ),
        database=envv["TIDB_DATABASE"],
    )
    log("provisioned HubSpot NamedCredential token secret")


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 assert_active_local_pack_source_bound(envv: dict[str, str], pack_name: str, pack_dir: Path) -> None:
    source_kind, source_uri = load_active_installed_pack_source_state(envv, pack_name)
    assert_local_pack_source_kind(pack_name, source_kind)
    expected_uri = str(pack_dir.resolve())
    actual_uri = str(Path(source_uri).expanduser().resolve()) if source_uri else ""
    if actual_uri != expected_uri:
        raise DeployError(
            "active local pack source is not bound to this deploy build root before owner-directory lifecycle sync: "
            f"name={pack_name} source_uri={source_uri or '<empty>'} expected_source_uri={expected_uri}. "
            "Run the deploy rebind/reapply path from DOWNSTREAM_BUILD_ROOT before executing lifecycle sync."
        )


def assert_local_pack_source_kind(pack_name: str, source_kind: str) -> None:
    if source_kind != "local_path":
        raise DeployError(
            "active pack source is not a local path before owner-directory lifecycle sync: "
            f"name={pack_name} source_kind={source_kind or '<empty>'}. "
            "Run the deploy rebind/reapply path from DOWNSTREAM_BUILD_ROOT before executing lifecycle sync."
        )


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(
                "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 job_id(job: dict[str, object]) -> str:
    return str(job.get("id", "")).strip()


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(
            "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(
            "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(
            "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(
            "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 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_local_pack_source_bound(envv, pack_name, pack_dir)
        assert_l1_post_activation_job_active_flow_matches_source(envv, pack_name, pack_dir, job)


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 runtime_metadata_tool_env(
    nexuscrm_worktree: Path,
    envv: dict[str, str] | None = None,
) -> dict[str, str]:
    candidate_root = str(
        (envv or {}).get("NEXUSCRM_OFFICIAL_APP_PACKS_ROOT", "")
    ).strip()
    if candidate_root:
        return {"NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": candidate_root}

    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.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)}

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

    return {}


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]:
    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, envv)
    tool_env.update(private_go_module_env(envv or {}))
    go_args = ["go", "run", "./.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
    app_packs_link = nexuscrm_worktree / "nexuscrm-app-packs"
    generated_app_packs_link = False
    if build_tags:
        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")
        app_packs_link_was_present = (
            app_packs_link.exists() or app_packs_link.is_symlink()
        )
        ensure_distribution_app_packs_link(nexuscrm_worktree, Path(official_root))
        generated_app_packs_link = not app_packs_link_was_present
        tool_root = backend_root / "distributions"
        go_args = ["go", "run", "-tags", build_tags, "./.runtime-tools/pack_runtime_metadata"]
        tool_env["GOWORK"] = "off"
        register_distribution = True

    tool_dir = tool_root / ".runtime-tools" / "pack_runtime_metadata"
    generated_tool_dir = not (tool_dir.exists() or tool_dir.is_symlink())
    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:
        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,
        )
    finally:
        if generated_tool_dir and tool_dir.exists():
            if tool_dir.is_symlink():
                raise DeployError(
                    f"runtime metadata helper directory became a symlink: {tool_dir}"
                )
            shutil.rmtree(tool_dir)
            try:
                tool_dir.parent.rmdir()
            except OSError:
                pass
        if generated_app_packs_link:
            if app_packs_link.is_symlink():
                app_packs_link.unlink()
            elif app_packs_link.exists():
                raise DeployError(
                    f"runtime metadata app-packs helper became a non-symlink: {app_packs_link}"
                )
    return json.loads(raw)


def installed_pack_source_revision(
    metadata: dict[str, object],
    exact_source_revision: str | None,
) -> tuple[str, bool]:
    if exact_source_revision is not None:
        revision = exact_source_revision.strip()
        return revision, bool(revision)
    return (
        str(metadata["source_revision"]),
        bool(metadata["source_revision_valid"]),
    )


def sync_installed_pack_runtime_metadata_with_apply_state(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    nexuscrm_worktree: Path,
    *,
    mark_applied: bool,
    source_revision: str | None = None,
) -> None:
    if not installed_pack_table_exists(envv):
        return
    if mark_applied:
        normalize_installed_pack_runtime_state(envv, pack_name, pack_dir)
    else:
        # Retrying a failed deploy must keep the pre-existing AppliedPayloadHash
        # in a drifted state. If a prior failed run already flattened the row to
        # the current release hash, the next retry must re-create a valid stale
        # hash so backend source-backed reapply runs again instead of dying in
        # verify-only drift checks.
        normalize_installed_pack_runtime_state_before_setup_apply(envv, pack_name, pack_dir)
    metadata = compute_source_backed_pack_runtime_metadata(
        nexuscrm_worktree,
        pack_dir,
        envv=envv,
        backend_go_build_tags=envv.get("BACKEND_GO_BUILD_TAGS"),
    )
    resolved_source_revision, resolved_source_revision_valid = (
        installed_pack_source_revision(metadata, source_revision)
    )
    escaped_name = pack_name.replace("'", "''")
    applied_payload_hash_sql = (
        sql_string_literal(str(metadata["payload_hash"]))
        if mark_applied
        else pre_apply_applied_payload_hash_sql(envv, pack_name, str(metadata["payload_hash"]))
    )
    runtime_decl_updates: list[str] = [
        f"RuntimeObjectSpecDeclarations={sql_string_literal(str(metadata['runtime_object_spec_declarations']))}",
        f"RuntimeRelatedListRules={sql_string_literal(str(metadata['runtime_related_list_rules']))}",
        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']))}"
    )
    runtime_decl_update_sql = "".join(f"\n                {update}," for update in runtime_decl_updates)
    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"]))},
                SourceRevision={sql_nullable_string_literal(
                    resolved_source_revision,
                    resolved_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,
    *,
    source_revision: str | None = None,
) -> None:
    sync_installed_pack_runtime_metadata_with_apply_state(
        envv,
        pack_name,
        pack_dir,
        nexuscrm_worktree,
        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,
    *,
    source_revision: str | None = None,
) -> None:
    # setup apply must see DesiredPayloadHash != AppliedPayloadHash when a warm
    # prod-mirror database still needs source-backed reapply. Backend reconcile
    # uses that installed-pack hash drift to enter the reapply path. If this
    # deploy script stamps AppliedPayloadHash to the new payload before setup
    # apply runs, backend wrongly believes the pack is already applied and skips
    # reapply, then 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,
        mark_applied=False,
        source_revision=source_revision,
    )


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 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:
    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"
    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"})
    build_env = {
        "REACT_APP_BACKEND_URL": envv["PUBLIC_BASE_URL"],
        "REACT_APP_FRONTEND_URL": envv["PUBLIC_BASE_URL"],
        "API_BASE_URL": envv["PUBLIC_BASE_URL"],
        "FRONTEND_URL": envv["PUBLIC_BASE_URL"],
    }
    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
    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
    run(["npm", "run", "build"], cwd=frontend_dir, env=build_env)
    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)
    verify_frontend_dist(frontend_dist)
    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(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 systemd_service_name(envv: dict[str, str]) -> str:
    override = envv.get("SYSTEMD_SERVICE_NAME", "").strip()
    if override:
        return override
    return f"nexuscrm-{envv['ENV_NAME']}-backend.service"


def strict_systemd_service_state(service: str) -> dict[str, str]:
    if not shutil.which("systemctl"):
        raise DeployError("cannot prove prod backend service state: systemctl is unavailable")
    try:
        inspected = subprocess.run(
            [
                "systemctl",
                "show",
                "--no-pager",
                "--property=LoadState",
                "--property=ActiveState",
                "--property=MainPID",
                "--property=ControlGroup",
                "--property=EnvironmentFiles",
                service,
            ],
            check=False,
            capture_output=True,
            text=True,
        )
    except OSError as exc:
        raise DeployError(f"failed to inspect prod backend service {service}: {exc}") from exc
    if inspected.returncode != 0:
        raise DeployError(
            f"failed to inspect prod backend service {service}: exit {inspected.returncode}"
        )
    values: dict[str, str] = {}
    for line in inspected.stdout.splitlines():
        key, separator, value = line.partition("=")
        if separator:
            values[key] = value.strip()
    load_state = values.get("LoadState", "")
    active_state = values.get("ActiveState", "")
    main_pid = values.get("MainPID", "")
    if not load_state or not active_state or not main_pid.isdigit():
        raise DeployError(f"incomplete prod backend service inspection for {service}")
    if load_state not in {"loaded", "masked", "not-found"}:
        raise DeployError(f"unusable prod backend service load state for {service}: {load_state}")
    return {
        "load_state": load_state,
        "active_state": active_state,
        "main_pid": main_pid,
        "control_group": values.get("ControlGroup", ""),
        "environment_files": values.get("EnvironmentFiles", ""),
    }


def strict_tcp_listener_state(port: str) -> tuple[list[str], set[str]]:
    normalized = str(port).strip()
    if not normalized.isdigit() or not 1 <= int(normalized) <= 65535:
        raise DeployError(f"invalid prod backend port slot: {normalized}")
    if not shutil.which("ss") or not shutil.which("sudo"):
        raise DeployError("cannot prove prod backend listener state: ss or sudo is unavailable")
    try:
        inspected = subprocess.run(
            ["sudo", "-n", "ss", "-H", "-ltnp", f"sport = :{normalized}"],
            check=False,
            capture_output=True,
            text=True,
        )
    except OSError as exc:
        raise DeployError(
            f"failed to inspect prod backend listener on port {normalized}: {exc}"
        ) from exc
    if inspected.returncode != 0:
        raise DeployError(
            f"failed to inspect prod backend listener on port {normalized}: exit {inspected.returncode}"
        )
    lines = [line for line in inspected.stdout.splitlines() if line.strip()]
    pids = {
        match.group(1)
        for line in lines
        for match in SS_LISTENER_PID_RE.finditer(line)
    }
    return lines, pids


def pid_belongs_to_systemd_control_group(pid: str, control_group: str) -> bool:
    normalized_pid = str(pid).strip()
    normalized_group = str(control_group).strip().rstrip("/")
    if not normalized_pid.isdigit() or int(normalized_pid) <= 1:
        raise DeployError(f"invalid prod backend listener pid: {normalized_pid}")
    if not normalized_group.startswith("/"):
        raise DeployError("invalid prod candidate systemd control group")
    try:
        cgroup_lines = (Path("/proc") / normalized_pid / "cgroup").read_text().splitlines()
    except OSError as exc:
        raise DeployError(
            f"cannot prove prod backend listener pid {normalized_pid} control group: {exc}"
        ) from exc
    for line in cgroup_lines:
        _, separator, cgroup_path = line.rpartition(":")
        if not separator:
            continue
        if cgroup_path == normalized_group or cgroup_path.startswith(normalized_group + "/"):
            return True
    return False


def prod_reset_source_live_signals(
    env_root: Path,
    nginx_conf_path: Path,
    backend_ports: Iterable[str],
    *,
    ignored_candidate: tuple[str, Path] | None = None,
) -> list[str]:
    env_root = env_root.expanduser().resolve()
    nginx_conf_path = nginx_conf_path.expanduser()
    if not nginx_conf_path.is_absolute():
        raise DeployError("prod markerless source nginx config path must be absolute")
    nginx_conf_path = nginx_conf_path.resolve()
    signals: list[str] = []
    active_runtime = env_root / "current" / "env" / "runtime.env"
    active_metadata = env_root / "current" / "metadata" / "release.json"
    if active_runtime.exists() or active_runtime.is_symlink():
        signals.append("current runtime env")
    if active_metadata.exists() or active_metadata.is_symlink():
        signals.append("current release metadata")
    if nginx_conf_path.exists() or nginx_conf_path.is_symlink():
        if not nginx_conf_path.is_file():
            signals.append("unreadable prod nginx config")
        else:
            try:
                nginx_content = nginx_conf_path.read_text()
            except OSError as exc:
                raise DeployError(f"cannot inspect prod nginx config {nginx_conf_path}: {exc}") from exc
            if PROD_BACKEND_PROXY_RE.search(nginx_content):
                signals.append("prod nginx backend route")
    if signals:
        return signals

    ports = [str(port).strip() for port in backend_ports]
    if not ports:
        raise DeployError("prod markerless source inspection has no backend port slots")
    if len(set(ports)) != len(ports):
        raise DeployError("duplicate prod backend port slot in markerless source inspection")
    if any(not port.isdigit() or not 1 <= int(port) <= 65535 for port in ports):
        raise DeployError("invalid prod backend port slot in markerless source inspection")
    candidate_port = ""
    candidate_runtime: Path | None = None
    if ignored_candidate is not None:
        candidate_port = str(ignored_candidate[0]).strip()
        candidate_runtime = ignored_candidate[1].expanduser().resolve()
        if candidate_port not in ports:
            raise DeployError("ignored prod candidate port is outside the configured backend slots")
        if not candidate_runtime.is_file():
            raise DeployError(f"ignored prod candidate runtime env is missing: {candidate_runtime}")

    for backend_port in ports:
        service = f"nexuscrm-multi-prod-slot-{backend_port}-backend.service"
        service_state = strict_systemd_service_state(service)
        running = (
            service_state["active_state"] != "inactive"
            or service_state["main_pid"] != "0"
        )
        verified_candidate_control_group = ""
        if running:
            if (
                candidate_runtime is not None
                and backend_port == candidate_port
                and service_state["active_state"] == "active"
                and service_state["main_pid"] != "0"
                and service_state["control_group"].startswith("/")
                and str(candidate_runtime) in service_state["environment_files"].split()
            ):
                verified_candidate_control_group = service_state["control_group"]
            else:
                signals.append(f"active backend service on port {backend_port}")

        listener_lines, listener_pids = strict_tcp_listener_state(backend_port)
        if not listener_lines:
            continue
        listener_is_verified_candidate = (
            bool(verified_candidate_control_group)
            and bool(listener_pids)
            and all(SS_LISTENER_PID_RE.search(line) for line in listener_lines)
            and all(
                pid_belongs_to_systemd_control_group(pid, verified_candidate_control_group)
                for pid in listener_pids
            )
        )
        if not listener_is_verified_candidate:
            signals.append(f"unverified backend listener on port {backend_port}")
    return signals


def hubspot_options_adapter_service_name(envv: dict[str, str]) -> str:
    override = envv.get("HUBSPOT_OPTIONS_ADAPTER_SERVICE_NAME", "").strip()
    if override:
        return override
    return f"nexuscrm-{envv['ENV_NAME']}-hubspot-options-adapter.service"


def hubspot_options_adapter_enabled(envv: dict[str, str]) -> bool:
    return bool(envv.get("EXP_HUBSPOT_ACTION_OPTIONS_TOKEN", "").strip())


def write_hubspot_options_adapter_runner(
    envv: dict[str, str],
    env_file: Path,
    env_root: Path,
    downstream_root: Path,
) -> Path:
    runner = env_root / "run_hubspot_options_adapter.sh"
    adapter_script = downstream_root / "deployments" / "offline" / "runtime" / "hubspot_workflow_options_adapter.py"
    content = textwrap.dedent(
        f"""\
        #!/usr/bin/env bash
        set -euo pipefail
        set -a
        source "{env_file}"
        set +a

        export PATH="/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin:$PATH"
        export EXP_HUBSPOT_OPTIONS_ADAPTER_PORT="{hubspot_options_adapter_port(envv)}"
        exec python3 "{adapter_script}"
        """
    )
    if not content.startswith("#!/usr/bin/env bash\n"):
        raise DeployError("generated HubSpot options adapter runner must start with a bash shebang")
    runner.write_text(content)
    runner.chmod(0o755)
    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 resolve_template_path(framework_dir: Path, relative_path: str) -> Path:
    exact_framework_dir = framework_dir.expanduser().resolve()
    candidate = exact_framework_dir / relative_path
    if candidate.is_file():
        return candidate
    if (
        exact_framework_dir.name != "ec2"
        or exact_framework_dir.parent.name != "ops"
    ):
        raise DeployError(f"template not found in exact framework: {candidate}")
    deployment_checkout = exact_framework_dir.parents[1]
    fallback = deployment_checkout / "deployments" / "offline" / relative_path
    if fallback.is_file():
        return fallback
    raise DeployError(
        "template not found in exact deployment checkout: "
        f"{candidate} (fallback: {fallback})"
    )


def nginx_conf_name(envv: dict[str, str]) -> str:
    return envv.get("NGINX_CONF_NAME", envv["ENV_NAME"]).strip() or envv["ENV_NAME"]


def nginx_log_basename(envv: dict[str, str]) -> str:
    for key in ("NGINX_LOG_BASENAME", "NGINX_CONF_NAME", "ENV_NAME"):
        value = envv.get(key, "").strip()
        if value:
            return value
    return envv["ENV_NAME"]


def render_nginx_conf(envv: dict[str, str], framework_dir: Path, frontend_dist: Path) -> str:
    listen_port = envv.get("NGINX_LISTEN_PORT", "80").strip() or "80"
    listen_suffix = envv.get("NGINX_LISTEN_SUFFIX", "").strip()
    listen_directive = " ".join(part for part in [listen_port, listen_suffix] if part).strip()
    server_names = " ".join(
        part
        for part in [
            envv["SERVER_NAME"].strip(),
            envv.get("NGINX_SERVER_ALIASES", "").strip(),
        ]
        if part
    ).strip()
    log_basename = nginx_log_basename(envv)
    return apply_hubspot_ingress_nginx_modes(
        envv,
        render_template(
        resolve_template_path(framework_dir, "nginx/nexuscrm-env.conf.template"),
        {
            "__NGINX_LISTEN_DIRECTIVE__": listen_directive,
            "__NGINX_LISTEN_PORT__": listen_port,
            "__NGINX_FORWARDED_PROTO__": envv.get("NGINX_FORWARDED_PROTO")
            or urlparse(envv["PUBLIC_BASE_URL"]).scheme
            or "http",
            "__SERVER_NAMES__": server_names,
            "__SERVER_NAME__": envv["SERVER_NAME"].strip(),
            "__FRONTEND_DIST__": str(frontend_dist),
            "__BACKEND_PORT__": envv["BACKEND_PORT"],
            "__CONNECTED_MAIL_INGRESS_BLOCK__": "",
            "__EXP_HUBSPOT_OPTIONS_ADAPTER_PORT__": 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",
        },
        ),
    )


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




def install_hubspot_options_adapter_unit(
    envv: dict[str, str],
    framework_dir: Path,
    downstream_root: Path,
    env_file: Path,
    env_root: Path,
) -> None:
    if not hubspot_options_adapter_enabled(envv) or not bool_env(envv, "INSTALL_SYSTEMD", True):
        return
    systemd_service = hubspot_options_adapter_service_name(envv)
    systemd_path = Path("/etc/systemd/system") / systemd_service
    rendered = render_template(
        resolve_template_path(framework_dir, "systemd/nexuscrm-hubspot-options-adapter.service.template"),
        {
            "__ENV_NAME__": envv["ENV_NAME"],
            "__SYSTEMD_USER__": envv.get("SYSTEMD_USER", os.environ.get("USER", "ec2-user")),
            "__DOWNSTREAM_ROOT__": str(downstream_root),
            "__ENV_FILE__": str(env_file),
            "__ENV_ROOT__": str(env_root),
            "__BACKEND_SERVICE__": systemd_service_name(envv),
        },
    )
    sudo_write_text(systemd_path, rendered)
    run(["sudo", "systemctl", "daemon-reload"])
    run(["sudo", "systemctl", "enable", systemd_service])


def install_rendered_nginx_conf(
    envv: dict[str, str],
    rendered: str,
) -> None:
    if not bool_env(envv, "INSTALL_NGINX", True):
        return
    nginx_path = Path("/etc/nginx/conf.d") / f"nexuscrm-{nginx_conf_name(envv)}.conf"
    sudo_write_text(nginx_path, rendered)
    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 install_nginx_conf(envv: dict[str, str], framework_dir: Path, frontend_dist: Path) -> None:
    if not bool_env(envv, "INSTALL_NGINX", True):
        return
    rendered = render_nginx_conf(envv, framework_dir, frontend_dist)
    install_rendered_nginx_conf(envv, rendered)




def stop_backend(envv: dict[str, str], run_dir: Path) -> None:
    if bool_env(envv, "SKIP_BACKEND_START"):
        return
    if bool_env(envv, "INSTALL_SYSTEMD", True):
        run(["sudo", "systemctl", "stop", systemd_service_name(envv)])
        return
    pid_file = run_dir / "backend.pid"
    if not pid_file.is_file():
        return
    pid = pid_file.read_text().strip()
    if pid:
        subprocess.run(["kill", pid], check=False)
    pid_file.unlink(missing_ok=True)


def stop_hubspot_options_adapter(envv: dict[str, str], run_dir: Path) -> None:
    if not hubspot_options_adapter_enabled(envv):
        return
    if bool_env(envv, "INSTALL_SYSTEMD", True):
        subprocess.run(["sudo", "systemctl", "stop", hubspot_options_adapter_service_name(envv)], check=False)
        return
    pid_file = run_dir / "hubspot-options-adapter.pid"
    if not pid_file.is_file():
        return
    pid = pid_file.read_text().strip()
    if pid:
        subprocess.run(["kill", pid], check=False)
    pid_file.unlink(missing_ok=True)


def start_hubspot_options_adapter(
    envv: dict[str, str],
    env_root: Path,
    log_dir: Path,
    run_dir: Path,
    *,
    timeout_seconds: float = 120.0,
) -> None:
    if not hubspot_options_adapter_enabled(envv):
        log("skipping HubSpot workflow options adapter; EXP_HUBSPOT_ACTION_OPTIONS_TOKEN is not set")
        return
    stop_hubspot_options_adapter(envv, run_dir)
    if not bool_env(envv, "INSTALL_SYSTEMD", True):
        log_dir.mkdir(parents=True, exist_ok=True)
        run_dir.mkdir(parents=True, exist_ok=True)
        log_file = log_dir / "hubspot-options-adapter.log"
        pid_file = run_dir / "hubspot-options-adapter.pid"
        with log_file.open("a") as fh:
            process = subprocess.Popen([str(env_root / "run_hubspot_options_adapter.sh")], stdout=fh, stderr=subprocess.STDOUT)
        pid_file.write_text(str(process.pid))
    else:
        run(["sudo", "systemctl", "restart", hubspot_options_adapter_service_name(envv)])
    wait_for_health(
        f"http://127.0.0.1:{hubspot_options_adapter_port(envv)}/hubspot/workflow-actions/options/health",
        attempts=30,
        timeout_seconds=timeout_seconds,
    )


def wait_for_health(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    attempts: int = 60,
    timeout_seconds: float = 120.0,
) -> None:
    deadline = time.monotonic() + max(0.1, timeout_seconds)
    for _ in range(attempts):
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            break
        curl_args = [
            "curl",
            "-fsS",
            "--connect-timeout",
            "5",
            "--max-time",
            f"{min(10.0, remaining_seconds):.3f}",
        ]
        if headers:
            for key, value in headers.items():
                curl_args += ["-H", f"{key}: {value}"]
        curl_args.append(url)
        result = subprocess.run(curl_args, capture_output=True, text=True, check=False)
        if result.returncode == 0:
            return
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            break
        time.sleep(min(2.0, remaining_seconds))
    raise DeployError(f"healthcheck failed: {url}")


def wait_for_http_ok(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    attempts: int = 60,
    timeout_seconds: float = 120.0,
) -> None:
    deadline = time.monotonic() + max(0.1, timeout_seconds)
    for _ in range(attempts):
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            break
        curl_args = [
            "curl",
            "-sS",
            "--connect-timeout",
            "5",
            "--max-time",
            f"{min(10.0, remaining_seconds):.3f}",
            "-o",
            "/dev/null",
            "-w",
            "%{http_code}",
        ]
        if headers:
            for key, value in headers.items():
                curl_args += ["-H", f"{key}: {value}"]
        curl_args.append(url)
        result = subprocess.run(curl_args, capture_output=True, text=True, check=False)
        if result.returncode == 0 and result.stdout.strip() == "200":
            return
        remaining_seconds = deadline - time.monotonic()
        if remaining_seconds <= 0:
            break
        time.sleep(min(2.0, remaining_seconds))
    raise DeployError(f"http verification failed: {url}")