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

from dataclasses import dataclass
from datetime import datetime, timezone
import json
import os
import shutil
import subprocess
import tempfile
import textwrap
import time
import uuid
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

import deployment_runtime_state
from deployment_plan import hubspot_credential_admission_policy
from common import (
    use_tidb_tls,
    DeployError,
    audit_final_pack,
    bool_env,
    build_final_pack,
    build_frontend,
    checkout_resolved_ref,
    cleanup_stale_runtime_dirs,
    cleanup_legacy_runtime_layout,
    clone_or_update_repo,
    compute_source_backed_pack_runtime_metadata,
    db_exec,
    db_query_scalar,
    ensure_runtime_packs_installed,
    ensure_runtime_mutable_dirs,
    ensure_tidb_database,
    exp_hubspot_options_adapter_port,
    install_nginx_conf,
    load_installed_pack_runtime_boundary,
    load_env_file,
    log,
    normalize_installed_pack_runtime_state as common_normalize_installed_pack_runtime_state,
    normalize_runtime_mutable_env,
    oauth_password_grant,
    prepare_downstream_build_root,
    prepare_nexuscrm_runtime,
    reload_nginx_service,
    rebind_installed_pack_source_local_path as common_rebind_installed_pack_source_local_path,
    rebind_retryable_installed_pack_source_identity as common_rebind_retryable_installed_pack_source_identity,
    refresh_runtime_metadata_cache,
    refresh_setup_derived_state,
    release_tcp_port,
    remove_path,
    require_commands,
    require_env,
    render_template,
    reset_database_if_requested,
    resolve_frontend_dist_target,
    resolve_repo_ref,
    runtime_deploy_pack_paths,
    run,
    run_backend_setup,
    assert_owner_directory_lifecycle_active_flows_for_runtime_packs,
    assert_active_pack_source_bound,
    assert_l1_post_activation_job_active_flow_matches_source,
    l1_post_activation_job_already_ran,
    l1_post_activation_jobs_for_runtime_packs,
    record_l1_post_activation_job_run,
    source_fingerprint_for_l1_post_activation_job,
    sql_nullable_string_literal,
    sql_string_literal,
    sync_installed_pack_runtime_metadata as common_sync_installed_pack_runtime_metadata,
    sync_installed_pack_runtime_metadata_without_marking_applied as common_sync_installed_pack_runtime_metadata_without_marking_applied,
    stop_backend,
    sudo_write_text,
    tidb_database_table_count,
    verify_nginx_backend_route_alignment,
    wait_for_health,
)

PACK_PAYLOAD_HASH_V1_PREFIX = "nexus-pack-sha256-v1:"
EXP_HUBSPOT_DEFAULT_OWNER_SENTINEL = "INTEGRATION_DEFAULT"
HUBSPOT_WEBHOOK_SECRET_ALGORITHM = "HMAC-SHA256"
HUBSPOT_WEBHOOK_SECRET_NAMES = ("webhook", "workflow-action")
PINGCAP_REGION_OWNER_RULE_COUNT = 6
PINGCAP_REGION_OWNER_LEAD_PROBES = (
    {"rule_region": "APAC-IN", "lead_region": "APAC", "country": "India"},
)
DEPLOY_SEED_SAFE_PROCESS_ENV_KEYS = (
    "LANG",
    "LC_ALL",
    "SSL_CERT_DIR",
    "SSL_CERT_FILE",
    "TMPDIR",
    "PYTHONDONTWRITEBYTECODE",
)
DEPLOY_SEED_CONDITION_ENV_KEYS = (
    "APPLY_CONNECTED_APPS",
    "APPLY_EXP_CREDITS_APPROVAL_SEED",
    "APPLY_EXP_CREDITS_PERMISSIONS_SEED",
    "APPLY_EXP_QUOTE_APPROVAL_SEED",
    "APPLY_IDENTITY_SYNC_POLICY",
    "APPLY_PINGCAP_REGION_OWNER_RULES",
    "APPLY_SHARED_DEPLOY_BASELINE",
    "APPLY_STAGING_ACCOUNT_NOTIFY_SEED",
    "APPLY_STAGING_CREDITS_APPROVAL_SEED",
    "APPLY_STAGING_CREDITS_PERMISSIONS_SEED",
    "APPLY_STAGING_QUOTE_APPROVAL_SEED",
    "APPLY_TEST_CREDITS_APPROVAL_SEED",
    "APPLY_TEST_CREDITS_PERMISSIONS_SEED",
    "APPLY_TEST_QUOTE_APPROVAL_SEED",
    "DEPLOY_PACK_REFERENCE_DATA_APPLY",
    "SKIP_BACKEND_START",
    "START_SERVICES",
)
DEPLOY_SEED_ADAPTER_ENV_KEYS = (
    "BACKEND_PORT",
    "CONNECTED_APPS_TONY_CONSUMER_SECRET",
    "CONNECTED_APPS_ZENO_CONSUMER_SECRET",
    "EXP_BULK_BASELINE_MAX_WORKERS",
    "EXP_BULK_BASELINE_OBJECTS",
    "NEXUS_ADMIN_PASSWORD",
    "PACK_REFERENCE_DATA_MYSQL_CLIENT",
    "PYTHON_SEED_DEPS_DIR",
    "STAGING_INSPECTION_OAUTH_CLIENT_ID",
    "STAGING_INSPECTION_OAUTH_CLIENT_SECRET",
    "TIDB_DATABASE",
    "TIDB_HOST",
    "TIDB_PASSWORD",
    "TIDB_PORT",
    "TIDB_TLS",
    "TIDB_USER",
    "VERIFY_PINGCAP_REGION_OWNER_RULES",
    "VERIFY_USERNAME",
)
# Every connected_apps run-as ConsumerSecret, sourced at deploy time (never a
# tracked literal). Each app in connected-apps.seed.yaml references its own
# {{ env.* }} key; all must be present on staging/EXP or the prerequisite fails.
CONNECTED_APPS_CONSUMER_SECRET_ENV_KEYS = (
    "CONNECTED_APPS_TONY_CONSUMER_SECRET",
    "CONNECTED_APPS_ZENO_CONSUMER_SECRET",
)
PROGRAMMATIC_OAUTH_CLIENT_ID_ENV = "STAGING_INSPECTION_OAUTH_CLIENT_ID"
PROGRAMMATIC_OAUTH_CLIENT_SECRET_ENV = "STAGING_INSPECTION_OAUTH_CLIENT_SECRET"
DEPLOY_SEED_PROFILE_ENV_KEYS = {
    "exp": ("EXP_HUBSPOT_ACCESS_TOKEN",),
}
DEPLOY_SEED_WORKTREE_ENV_KEY = "NEXUSCRM_WORKTREE"
DEPLOY_SEED_EVIDENCE_KEYS = {
    "catalog_version",
    "execution_mode",
    "failure",
    "format",
    "inventory_file",
    "invocations",
    "phase",
    "plan_file",
    "profile",
    "schema_version",
    "source_fingerprint_sha256",
    "status",
}
DEPLOY_SEED_INVOCATION_EVIDENCE_KEYS = {
    "adapter_id",
    "adapter_output_dir",
    "condition",
    "invocation_id",
    "module_id",
    "order",
    "selection",
    "status",
    "summary",
}


def write_env_file(path: Path, values: dict[str, str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [f"{key}={value}" for key, value in values.items()]
    path.write_text("\n".join(lines) + "\n")


def materialize_exp_live_frontend(frontend_dist: Path) -> Path:
    release_dir = resolve_frontend_dist_target(frontend_dist)
    materialized_dir = frontend_dist.parent / f".{frontend_dist.name}.materialized"
    remove_path(materialized_dir)
    shutil.copytree(release_dir, materialized_dir)
    if frontend_dist.is_symlink():
        frontend_dist.unlink()
    elif frontend_dist.exists():
        remove_path(frontend_dist)
    os.replace(materialized_dir, frontend_dist)
    return frontend_dist


def materialize_runtime_env(env_root: Path, envv: dict[str, str], *, source_env_file: Path | None = None) -> Path:
    target = deployment_runtime_state.runtime_paths(env_root)["runtime_env_file"]
    runtime_values = dict(envv)
    runtime_values["ENV_ROOT"] = str(env_root)
    runtime_values["EXP_HUBSPOT_OPTIONS_ADAPTER_PORT"] = exp_hubspot_options_adapter_port(runtime_values)
    if source_env_file is not None:
        runtime_values["RUNTIME_ENV_SOURCE_FILE"] = str(source_env_file)
    write_env_file(target, runtime_values)
    return target


def exp_hubspot_expected_webhook_url(envv: dict[str, str]) -> str:
    public_base_url = envv.get("PUBLIC_BASE_URL", "").strip().rstrip("/")
    if not public_base_url:
        return ""
    return f"{public_base_url}/webhooks/hubspot-crm-connector/webhook"


def validate_exp_hubspot_public_config(envv: dict[str, str]) -> None:
    configured_url = envv.get("EXP_HUBSPOT_WEBHOOK_URL", "").strip().rstrip("/")
    if not configured_url:
        return
    expected_url = exp_hubspot_expected_webhook_url(envv).rstrip("/")
    if expected_url and configured_url != expected_url:
        raise DeployError(
            "EXP_HUBSPOT_WEBHOOK_URL does not match the deployed EXP webhook route: "
            f"configured={configured_url} expected={expected_url}"
        )


def connected_apps_missing_consumer_secrets(envv: dict[str, str]) -> tuple[str, ...]:
    return tuple(
        name
        for name in CONNECTED_APPS_CONSUMER_SECRET_ENV_KEYS
        if not str(envv.get(name) or os.environ.get(name) or "").strip()
    )


def validate_connected_apps_prerequisite(
    envv: dict[str, str],
    *,
    environment_name: str,
) -> None:
    if not bool_env(envv, "APPLY_CONNECTED_APPS", True):
        return
    missing = connected_apps_missing_consumer_secrets(envv)
    if missing:
        raise DeployError(
            f"missing {environment_name} runner prerequisite "
            + ", ".join(missing)
        )


def validate_exp_connected_apps_prerequisite(envv: dict[str, str]) -> None:
    validate_connected_apps_prerequisite(envv, environment_name="EXP")


def api_json_request(
    base_url: str,
    path: str,
    *,
    access_token: str,
    method: str = "GET",
    payload: dict[str, object] | None = None,
) -> dict[str, object]:
    data = None
    headers = {"Authorization": f"Bearer {access_token}"}
    if payload is not None:
        data = json.dumps(payload).encode()
        headers["Content-Type"] = "application/json"
    request = Request(
        f"{base_url.rstrip('/')}{path}",
        data=data,
        headers=headers,
        method=method,
    )
    try:
        with urlopen(request, timeout=30) as response:
            body = response.read().decode()
    except HTTPError as exc:
        body = exc.read().decode(errors="replace")
        raise DeployError(f"{method} {path} failed: http {exc.code}: {body[:500]}") from exc
    except URLError as exc:
        raise DeployError(f"{method} {path} failed: {exc}") from exc
    return json.loads(body) if body.strip() else {}


def hubspot_json_request(
    path: str,
    *,
    access_token: str,
) -> dict[str, object]:
    request = Request(
        f"https://api.hubapi.com{path}",
        headers={"Authorization": f"Bearer {access_token}"},
        method="GET",
    )
    try:
        with urlopen(request, timeout=30) as response:
            body = response.read().decode()
    except HTTPError as exc:
        body = exc.read().decode(errors="replace")
        raise DeployError(f"GET HubSpot {path} failed: http {exc.code}: {body[:500]}") from exc
    except URLError as exc:
        raise DeployError(f"GET HubSpot {path} failed: {exc}") from exc
    return json.loads(body) if body.strip() else {}


def hubspot_webhook_secret_id(payload: dict[str, object], secret_name: str = "webhook") -> str:
    secret_name = secret_name.strip()
    if not secret_name:
        raise DeployError("hubspot webhook secret name is required")
    raw_items = payload.get("webhookSecrets")
    if not isinstance(raw_items, list):
        raise DeployError("webhook secret list response missing webhookSecrets array")
    matches: list[str] = []
    for raw_item in raw_items:
        if not isinstance(raw_item, dict):
            continue
        owner_pack = str(raw_item.get("ownerPack") or "").strip()
        name = str(raw_item.get("name") or "").strip()
        secret_id = str(raw_item.get("id") or "").strip()
        if owner_pack == "hubspot-crm-connector" and name == secret_name and secret_id:
            matches.append(secret_id)
    if len(matches) != 1:
        raise DeployError(
            f"expected exactly one hubspot-crm-connector/{secret_name} secret definition; "
            f"found {len(matches)}"
        )
    return matches[0]


def _hubspot_webhook_secret_names(payload: dict[str, object], owner_pack: str = "hubspot-crm-connector") -> set[str]:
    raw_items = payload.get("webhookSecrets")
    if not isinstance(raw_items, list):
        raise DeployError("webhook secret list response missing webhookSecrets array")
    wanted_owner_pack = owner_pack.strip().lower()
    found: set[str] = set()
    for raw_item in raw_items:
        if not isinstance(raw_item, dict):
            continue
        raw_owner_pack = str(raw_item.get("ownerPack") or "").strip().lower()
        raw_name = str(raw_item.get("name") or "").strip().lower()
        if raw_owner_pack == wanted_owner_pack and raw_name:
            found.add(raw_name)
    return found


def ensure_hubspot_webhook_secret_definitions(
    envv: dict[str, str],
    *,
    base_url: str,
    access_token: str,
    owner_pack: str = "hubspot-crm-connector",
) -> dict[str, object]:
    owner_pack = owner_pack.strip().lower()
    secrets_payload = api_json_request(
        base_url,
        f"/services/data/v63.0/connector-webhooks/secrets?namespace={quote(owner_pack, safe='')}",
        access_token=access_token,
    )
    present = _hubspot_webhook_secret_names(secrets_payload, owner_pack)
    missing = [name for name in HUBSPOT_WEBHOOK_SECRET_NAMES if name not in present]
    if not missing:
        return secrets_payload

    log(
        "repairing missing HubSpot webhook secret definitions: "
        + ", ".join(f"{owner_pack}/{name}" for name in missing)
    )
    rows_sql = ",\n".join(
        (
            "("
            f"{sql_string_literal(uuid.uuid4().hex)}, "
            f"{sql_string_literal(owner_pack)}, "
            f"{sql_string_literal(name)}, "
            f"{sql_string_literal(HUBSPOT_WEBHOOK_SECRET_ALGORITHM)}, "
            "NULL, 0, NOW(), NOW()"
            ")"
        )
        for name in missing
    )
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            INSERT INTO _System_WebhookSecret
              (Id, OwnerPack, Name, Algorithm, Description, IsDeleted, CreatedDate, LastModifiedDate)
            VALUES
            {rows_sql}
            ON DUPLICATE KEY UPDATE
              Name = VALUES(Name),
              Algorithm = VALUES(Algorithm),
              Description = VALUES(Description),
              IsDeleted = 0,
              LastModifiedDate = NOW();
            """
        ).strip(),
        database=envv["TIDB_DATABASE"],
    )

    repaired_payload = api_json_request(
        base_url,
        f"/services/data/v63.0/connector-webhooks/secrets?namespace={quote(owner_pack, safe='')}",
        access_token=access_token,
    )
    repaired_present = _hubspot_webhook_secret_names(repaired_payload, owner_pack)
    still_missing = [name for name in HUBSPOT_WEBHOOK_SECRET_NAMES if name not in repaired_present]
    if still_missing:
        raise DeployError(
            "failed to materialize HubSpot webhook secret definitions: "
            + ", ".join(f"{owner_pack}/{name}" for name in still_missing)
        )
    return repaired_payload


def verify_exp_hubspot_access_token_identity(envv: dict[str, str]) -> dict[str, str] | None:
    """Verify the EXP token before it can be written to the runtime database."""
    hubspot_access_token = envv.get("EXP_HUBSPOT_ACCESS_TOKEN", "").strip()
    if not hubspot_access_token:
        return None
    policy = hubspot_credential_admission_policy("exp", error_type=DeployError)
    expected_identity = {
        "portalId": policy["expected_portal_id"],
        "accountType": policy["expected_account_type"],
    }
    try:
        payload = hubspot_json_request(
            "/account-info/v3/details",
            access_token=hubspot_access_token,
        )
    except (DeployError, json.JSONDecodeError) as exc:
        raise DeployError("unable to verify HubSpot EXP access-token identity") from exc
    if not isinstance(payload, dict):
        raise DeployError("HubSpot EXP access-token identity response must be an object")
    identity = {
        "portalId": str(payload.get("portalId") or "").strip(),
        "accountType": str(payload.get("accountType") or "").strip(),
    }
    if identity != expected_identity:
        raise DeployError("HubSpot EXP access-token identity does not match the committed policy")
    log(
        "verified HubSpot EXP access-token identity: "
        f"portalId={identity['portalId']} accountType={identity['accountType']}"
    )
    return identity


def provision_exp_hubspot_runtime_config(envv: dict[str, str]) -> None:
    validate_exp_hubspot_public_config(envv)
    hubspot_access_token = envv.get("EXP_HUBSPOT_ACCESS_TOKEN", "").strip()
    hubspot_client_secret = envv.get("EXP_HUBSPOT_CLIENT_SECRET", "").strip()
    if not hubspot_access_token and not hubspot_client_secret:
        log("skipping HubSpot EXP runtime secret provisioning; no HubSpot EXP secrets configured")
        return
    missing = [
        key
        for key, value in [
            ("EXP_HUBSPOT_ACCESS_TOKEN", hubspot_access_token),
            ("EXP_HUBSPOT_CLIENT_SECRET", hubspot_client_secret),
            ("EXP_HUBSPOT_WEBHOOK_URL", envv.get("EXP_HUBSPOT_WEBHOOK_URL", "").strip()),
        ]
        if not value
    ]
    if missing:
        raise DeployError(f"missing required HubSpot EXP config keys: {', '.join(missing)}")
    verify_exp_hubspot_access_token_identity(envv)

    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 provision HubSpot EXP runtime secrets: missing NEXUS_ADMIN_PASSWORD")

    token_json = oauth_password_grant(base_url, username=username, password=password)
    access_token = str(token_json.get("access_token") or "").strip()
    if not access_token:
        raise DeployError("HubSpot EXP runtime secret provisioning oauth response missing access_token")

    log("provisioning HubSpot EXP NamedCredential token")
    api_json_request(
        base_url,
        "/api/named-credentials/HubSpotPrivateApp/secrets/Token",
        access_token=access_token,
        method="POST",
        payload={"secretValue": hubspot_access_token},
    )

    log("rotating HubSpot EXP webhook secrets")
    secrets_payload = ensure_hubspot_webhook_secret_definitions(
        envv,
        base_url=base_url,
        access_token=access_token,
    )
    for secret_name in HUBSPOT_WEBHOOK_SECRET_NAMES:
        secret_id = hubspot_webhook_secret_id(secrets_payload, secret_name)
        api_json_request(
            base_url,
            f"/services/data/v63.0/connector-webhooks/secrets/{quote(secret_id, safe='')}/rotate",
            access_token=access_token,
            method="POST",
            payload={"secretValue": hubspot_client_secret},
        )
    log("HubSpot EXP runtime secrets provisioned")


def request_exp_local_access_token(envv: dict[str, str]) -> str:
    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("missing NEXUS_ADMIN_PASSWORD for local deploy bootstrap")
    client_id = str(envv.get(PROGRAMMATIC_OAUTH_CLIENT_ID_ENV, "")).strip()
    client_secret = str(envv.get(PROGRAMMATIC_OAUTH_CLIENT_SECRET_ENV, "")).strip()
    if bool(client_id) != bool(client_secret):
        raise DeployError("programmatic OAuth client binding is incomplete")
    client_kwargs = (
        {"client_id": client_id, "client_secret": client_secret}
        if client_id
        else {}
    )
    token_json = oauth_password_grant(
        base_url,
        username=username,
        password=password,
        **client_kwargs,
    )
    access_token = str(token_json.get("access_token") or "").strip()
    if not access_token:
        raise DeployError("local deploy bootstrap oauth response missing access_token")
    return access_token


def exp_soql_total_size(base_url: str, access_token: str, soql: str) -> int:
    payload = api_json_request(
        base_url,
        f"/services/data/v63.0/query?q={quote(soql, safe='')}",
        access_token=access_token,
    )
    if not isinstance(payload, dict):
        raise DeployError(f"SOQL query returned non-object payload for {soql!r}")
    total_size = payload.get("totalSize")
    if not isinstance(total_size, int):
        raise DeployError(f"SOQL query missing integer totalSize for {soql!r}")
    return total_size


def exp_soql_records(base_url: str, access_token: str, soql: str) -> list[dict[str, object]]:
    payload = api_json_request(
        base_url,
        f"/services/data/v63.0/query?q={quote(soql, safe='')}",
        access_token=access_token,
    )
    if not isinstance(payload, dict):
        raise DeployError(f"SOQL query returned non-object payload for {soql!r}")
    records = payload.get("records")
    if not isinstance(records, list) or not all(isinstance(record, dict) for record in records):
        raise DeployError(f"SOQL query missing record list for {soql!r}")
    return records


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


def normalized_exp_hubspot_owner_binding_fields(record: dict[str, object]) -> dict[str, str]:
    return {
        "HubSpotOwnerId__c": str(record.get("HubSpotOwnerId__c") or "").strip(),
        "NormalizedEmail__c": str(record.get("NormalizedEmail__c") or "").strip().lower(),
        "CRMPrincipalType__c": str(record.get("CRMPrincipalType__c") or "").strip(),
        "CRMPrincipalId__c": str(record.get("CRMPrincipalId__c") or "").strip(),
        "BindingStatus__c": str(record.get("BindingStatus__c") or "").strip(),
    }


def load_exp_hubspot_owner_binding_bootstrap(runtime_script_dir: Path) -> list[dict[str, str]]:
    mapping_path = runtime_script_dir / "seeds" / "exp-hubspot-owner-bindings.json"
    if not mapping_path.is_file():
        raise DeployError(f"missing EXP HubSpot owner bootstrap mapping: {mapping_path}")
    raw = json.loads(mapping_path.read_text())
    if not isinstance(raw, list) or not raw:
        raise DeployError("EXP HubSpot owner bootstrap mapping must be a non-empty array")

    mappings: list[dict[str, str]] = []
    seen_owner_ids: set[str] = set()
    seen_emails: set[str] = set()
    for index, item in enumerate(raw, start=1):
        if not isinstance(item, dict):
            raise DeployError(f"EXP HubSpot owner bootstrap row {index} must be an object")
        owner_id = str(item.get("hubspot_owner_id") or "").strip()
        email = str(item.get("normalized_email") or "").strip().lower()
        principal_type = str(item.get("crm_principal_type") or "User").strip() or "User"
        binding_status = str(item.get("binding_status") or "bound").strip() or "bound"
        if not owner_id or not email:
            raise DeployError(f"EXP HubSpot owner bootstrap row {index} missing owner id or email")
        if owner_id in seen_owner_ids:
            raise DeployError(f"duplicate HubSpot owner id in EXP bootstrap mapping: {owner_id}")
        if email in seen_emails:
            raise DeployError(f"duplicate normalized email in EXP bootstrap mapping: {email}")
        seen_owner_ids.add(owner_id)
        seen_emails.add(email)
        mappings.append(
            {
                "hubspot_owner_id": owner_id,
                "normalized_email": email,
                "crm_principal_type": principal_type,
                "binding_status": binding_status,
            }
        )
    return mappings


def resolve_exp_hubspot_owner_binding_payload(
    base_url: str,
    access_token: str,
    mapping: dict[str, str],
    *,
    portal_id: str = "",
) -> dict[str, str] | None:
    owner_id = mapping["hubspot_owner_id"]
    email = mapping["normalized_email"]
    user_records = exp_soql_records(
        base_url,
        access_token,
        (
            "SELECT Id FROM User "
            f"WHERE Email = '{soql_string_literal(email)}' "
            "AND IsActive = true"
        ),
    )
    if len(user_records) != 1:
        return None
    user_id = str(user_records[0].get("Id") or "").strip()
    if not user_id:
        raise DeployError(f"EXP HubSpot owner bootstrap user lookup missing Id for {email}")

    payload = {
        "HubSpotOwnerId__c": owner_id,
        "NormalizedEmail__c": email,
        "CRMPrincipalType__c": mapping["crm_principal_type"],
        "CRMPrincipalId__c": user_id,
        "BindingStatus__c": mapping["binding_status"],
        "SourceOfTruth__c": "auto_email",
    }
    if portal_id:
        payload["PortalId__c"] = portal_id
    return payload


def upsert_exp_hubspot_owner_binding(
    base_url: str,
    access_token: str,
    payload: dict[str, str],
) -> str:
    owner_id = payload["HubSpotOwnerId__c"]
    binding_records = exp_soql_records(
        base_url,
        access_token,
        (
            "SELECT Id, ManualLock__c, HubSpotOwnerId__c, NormalizedEmail__c, "
            "CRMPrincipalType__c, CRMPrincipalId__c, BindingStatus__c "
            "FROM HubSpotOwnerBinding__c "
            f"WHERE HubSpotOwnerId__c = '{soql_string_literal(owner_id)}'"
        ),
    )
    if len(binding_records) > 1:
        raise DeployError(
            "EXP HubSpot owner bootstrap found duplicate binding rows for "
            f"HubSpot owner {owner_id}: {len(binding_records)}"
        )
    if binding_records:
        binding_id = str(binding_records[0].get("Id") or "").strip()
        if not binding_id:
            raise DeployError(f"EXP HubSpot owner bootstrap binding lookup missing Id for {owner_id}")
        if bool(binding_records[0].get("ManualLock__c")):
            actual_fields = normalized_exp_hubspot_owner_binding_fields(binding_records[0])
            expected_fields = normalized_exp_hubspot_owner_binding_fields(payload)
            if actual_fields != expected_fields:
                raise DeployError(
                    "locked HubSpot owner binding does not match expected EXP bootstrap mapping for "
                    f"{owner_id}: expected {json.dumps(expected_fields, sort_keys=True)} "
                    f"found {json.dumps(actual_fields, sort_keys=True)}"
                )
            log(f"verified locked HubSpot owner binding during EXP bootstrap: owner={owner_id}")
            return "verified_locked"
        api_json_request(
            base_url,
            f"/services/data/v63.0/sobjects/HubSpotOwnerBinding__c/{quote(binding_id, safe='')}",
            access_token=access_token,
            method="PATCH",
            payload=payload,
        )
        return "updated"

    api_json_request(
        base_url,
        "/services/data/v63.0/sobjects/HubSpotOwnerBinding__c",
        access_token=access_token,
        method="POST",
        payload=payload,
    )
    return "created"


def hubspot_owner_directory_records(hubspot_access_token: str) -> list[dict[str, object]]:
    owners: list[dict[str, object]] = []
    after = ""
    while True:
        path = "/crm/v3/owners/?limit=500&archived=false"
        if after:
            path += f"&after={quote(after, safe='')}"
        payload = hubspot_json_request(path, access_token=hubspot_access_token)
        raw_results = payload.get("results")
        if not isinstance(raw_results, list):
            raise DeployError("HubSpot owner directory response missing results array")
        owners.extend(record for record in raw_results if isinstance(record, dict))
        paging = payload.get("paging")
        if not isinstance(paging, dict):
            break
        next_page = paging.get("next")
        if not isinstance(next_page, dict):
            break
        after = str(next_page.get("after") or "").strip()
        if not after:
            break
    return owners


def hubspot_portal_id(hubspot_access_token: str) -> str:
    payload = hubspot_json_request(
        "/account-info/v3/details",
        access_token=hubspot_access_token,
    )
    portal_id = str(payload.get("portalId") or "").strip()
    if not portal_id or not portal_id.isdigit():
        raise DeployError("HubSpot account details response missing numeric portalId")
    return portal_id


def verify_hubspot_account_identity(
    hubspot_access_token: str,
    *,
    expected_portal_id: str,
    expected_account_type: str,
) -> dict[str, str]:
    token = hubspot_access_token.strip()
    if not token:
        raise DeployError("HubSpot account identity verification requires an access token")
    expected_portal_id = expected_portal_id.strip()
    expected_account_type = expected_account_type.strip()
    if not expected_portal_id or not expected_account_type:
        raise DeployError("HubSpot account identity verification requires an exact portal and account type")
    try:
        payload = hubspot_json_request(
            "/account-info/v3/details",
            access_token=token,
        )
    except (DeployError, json.JSONDecodeError) as exc:
        # The upstream response body is not useful evidence for this gate and
        # could contain tenant-specific data. Keep the failure fail-closed and
        # token-free.
        raise DeployError("HubSpot account-info identity verification request failed") from exc

    if not isinstance(payload, dict):
        raise DeployError("HubSpot account-info identity verification returned a non-object response")
    portal_id = str(payload.get("portalId") or "").strip()
    account_type = str(payload.get("accountType") or "").strip()
    if portal_id != expected_portal_id or account_type != expected_account_type:
        raise DeployError(
            "HubSpot account identity mismatch: "
            f"expected portalId={expected_portal_id} accountType={expected_account_type}; "
            f"got portalId={portal_id or '<missing>'} accountType={account_type or '<missing>'}"
        )
    return {
        "portalId": portal_id,
        "accountType": account_type,
    }


def owner_directory_mapping_from_hubspot_owner(owner: dict[str, object]) -> dict[str, str] | None:
    owner_id = str(owner.get("id") or "").strip()
    email = str(owner.get("email") or "").strip().lower()
    owner_type = str(owner.get("type") or "").strip()
    archived = bool(owner.get("archived"))
    if not owner_id or not email or archived:
        return None
    if owner_type and owner_type != "PERSON":
        return None
    return {
        "hubspot_owner_id": owner_id,
        "normalized_email": email,
        "crm_principal_type": "User",
        "binding_status": "bound",
    }


def retire_stale_auto_email_owner_bindings(
    base_url: str,
    access_token: str,
    *,
    portal_id: str,
    resolved_owner_ids: set[str],
) -> int:
    query = (
        "SELECT Id, HubSpotOwnerId__c, PortalId__c, ManualLock__c "
        "FROM HubSpotOwnerBinding__c "
        "WHERE SourceOfTruth__c = 'auto_email' AND BindingStatus__c = 'bound'"
    )
    binding_records = exp_soql_records(base_url, access_token, query)
    retired = 0
    for record in binding_records:
        if bool(record.get("ManualLock__c")):
            continue
        owner_id = str(record.get("HubSpotOwnerId__c") or "").strip()
        if owner_id in resolved_owner_ids:
            continue
        binding_id = str(record.get("Id") or "").strip()
        if not binding_id:
            raise DeployError("auto-email HubSpot owner binding lookup returned a row without Id")
        api_json_request(
            base_url,
            f"/services/data/v63.0/sobjects/HubSpotOwnerBinding__c/{quote(binding_id, safe='')}",
            access_token=access_token,
            method="PATCH",
            payload={"BindingStatus__c": "inactive"},
        )
        retired += 1

    remaining = exp_soql_records(base_url, access_token, query)
    invalid: list[str] = []
    for record in remaining:
        if bool(record.get("ManualLock__c")):
            continue
        owner_id = str(record.get("HubSpotOwnerId__c") or "").strip()
        record_portal_id = str(record.get("PortalId__c") or "").strip()
        if owner_id not in resolved_owner_ids or record_portal_id != portal_id:
            invalid.append(owner_id or "<missing-owner-id>")
    if invalid:
        raise DeployError(
            "current HubSpot portal owner reconciliation left invalid unlocked auto-email bindings: "
            + ", ".join(invalid[:10])
        )
    return retired


def sync_hubspot_owner_directory_bindings_from_api(
    envv: dict[str, str],
    base_url: str,
    access_token: str,
) -> int:
    hubspot_access_token = envv.get("EXP_HUBSPOT_ACCESS_TOKEN", "").strip()
    if not hubspot_access_token:
        log("skipping direct HubSpot owner directory sync; EXP_HUBSPOT_ACCESS_TOKEN is not set")
        return 0

    portal_id = hubspot_portal_id(hubspot_access_token)
    owners = hubspot_owner_directory_records(hubspot_access_token)
    created = 0
    updated = 0
    verified_locked = 0
    unresolved = 0
    skipped = 0
    resolved_payloads: list[dict[str, str]] = []
    for owner in owners:
        mapping = owner_directory_mapping_from_hubspot_owner(owner)
        if mapping is None:
            skipped += 1
            continue
        payload = resolve_exp_hubspot_owner_binding_payload(
            base_url,
            access_token,
            mapping,
            portal_id=portal_id,
        )
        if payload is None:
            unresolved += 1
            continue
        resolved_payloads.append(payload)

    for payload in resolved_payloads:
        result = upsert_exp_hubspot_owner_binding(base_url, access_token, payload)
        if result == "created":
            created += 1
        elif result == "updated":
            updated += 1
        elif result == "verified_locked":
            verified_locked += 1

    resolved_owner_ids = {payload["HubSpotOwnerId__c"] for payload in resolved_payloads}
    retired = retire_stale_auto_email_owner_bindings(
        base_url,
        access_token,
        portal_id=portal_id,
        resolved_owner_ids=resolved_owner_ids,
    )

    applied = created + updated + verified_locked
    log(
        "direct HubSpot owner directory sync completed "
        f"owners={len(owners)} created={created} updated={updated} "
        f"verified_locked={verified_locked} unresolved={unresolved} skipped={skipped} "
        f"retired={retired} portal_id={portal_id}"
    )
    return applied


def verify_hubspot_owner_directory_lifecycle_effect(
    base_url: str,
    access_token: str,
) -> tuple[int, int]:
    synced_binding_count = exp_soql_total_size(
        base_url,
        access_token,
        (
            "SELECT Id FROM HubSpotOwnerBinding__c "
            f"WHERE HubSpotOwnerId__c != '{EXP_HUBSPOT_DEFAULT_OWNER_SENTINEL}'"
        ),
    )
    exception_count = exp_soql_total_size(
        base_url,
        access_token,
        "SELECT Id FROM HubSpotOwnerBindingException__c",
    )
    if synced_binding_count == 0:
        raise DeployError(
            "HubSpot owner directory lifecycle produced no non-default bindings: "
            f"exceptions={exception_count}"
        )
    log(
        "verified HubSpot owner directory lifecycle "
        f"bindings={synced_binding_count} exceptions={exception_count}"
    )
    return synced_binding_count, exception_count


def verify_exp_hubspot_owner_bootstrap_counts(
    base_url: str,
    access_token: str,
    *,
    expected_bindings: list[dict[str, str]],
    require_directory_sync: bool,
) -> None:
    expected_by_owner_id = {
        payload["HubSpotOwnerId__c"]: normalized_exp_hubspot_owner_binding_fields(payload)
        for payload in expected_bindings
    }
    default_count = exp_soql_total_size(
        base_url,
        access_token,
        "SELECT Id FROM HubSpotOwnerBinding__c WHERE BindingStatus__c = 'default'",
    )
    if default_count != 1:
        raise DeployError(
            "expected exactly one default HubSpot owner binding after EXP bootstrap; "
            f"found {default_count}"
        )
    if expected_by_owner_id:
        owner_id_literals = ", ".join(
            f"'{soql_string_literal(owner_id)}'"
            for owner_id in expected_by_owner_id
        )
        actual_records = exp_soql_records(
            base_url,
            access_token,
            (
                "SELECT HubSpotOwnerId__c, NormalizedEmail__c, CRMPrincipalType__c, "
                "CRMPrincipalId__c, BindingStatus__c FROM HubSpotOwnerBinding__c "
                f"WHERE HubSpotOwnerId__c IN ({owner_id_literals})"
            ),
        )
        actual_by_owner_id: dict[str, dict[str, str]] = {}
        for record in actual_records:
            normalized_record = normalized_exp_hubspot_owner_binding_fields(record)
            owner_id = normalized_record["HubSpotOwnerId__c"]
            if owner_id in actual_by_owner_id:
                raise DeployError(
                    "duplicate HubSpot owner binding rows returned during EXP bootstrap verification for "
                    f"{owner_id}"
                )
            actual_by_owner_id[owner_id] = normalized_record
        if set(actual_by_owner_id) != set(expected_by_owner_id):
            raise DeployError(
                "deploy-time HubSpot owner bootstrap verified the wrong owner id set: "
                f"expected {sorted(expected_by_owner_id)} found {sorted(actual_by_owner_id)}"
            )
        for owner_id, expected_fields in expected_by_owner_id.items():
            actual_fields = actual_by_owner_id[owner_id]
            if actual_fields != expected_fields:
                raise DeployError(
                    "deploy-time HubSpot owner bootstrap verification mismatch for "
                    f"{owner_id}: expected {json.dumps(expected_fields, sort_keys=True)} "
                    f"found {json.dumps(actual_fields, sort_keys=True)}"
                )

    if require_directory_sync:
        synced_binding_count = exp_soql_total_size(
            base_url,
            access_token,
            (
                "SELECT Id FROM HubSpotOwnerBinding__c "
                f"WHERE HubSpotOwnerId__c != '{EXP_HUBSPOT_DEFAULT_OWNER_SENTINEL}'"
            ),
        )
        exception_count = exp_soql_total_size(
            base_url,
            access_token,
            "SELECT Id FROM HubSpotOwnerBindingException__c",
        )
        if synced_binding_count == 0 and exception_count == 0:
            log(
                "deploy-time HubSpot owner directory sync produced no non-default bindings or exceptions; "
                "continuing because EXP bootstrap may legitimately retain only the default binding"
            )
        log(
            "verified EXP HubSpot owner bootstrap "
            f"default=1 expected_bound={len(expected_bindings)} synced={synced_binding_count} exceptions={exception_count}"
        )
        return

    bound_binding_count = exp_soql_total_size(
        base_url,
        access_token,
        (
            "SELECT Id FROM HubSpotOwnerBinding__c "
            f"WHERE HubSpotOwnerId__c != '{EXP_HUBSPOT_DEFAULT_OWNER_SENTINEL}' "
            "AND BindingStatus__c = 'bound'"
        ),
    )
    expected_bound_bindings = len(expected_bindings)
    if bound_binding_count != expected_bound_bindings:
        raise DeployError(
            "deploy-time HubSpot owner bootstrap produced an unexpected bound binding count: "
            f"expected {expected_bound_bindings}, found {bound_binding_count}"
        )
    log(
        "verified EXP HubSpot owner bootstrap "
        f"default=1 bound={bound_binding_count}"
    )


def bootstrap_exp_hubspot_owner_bindings(
    envv: dict[str, str],
    runtime_script_dir: Path,
    downstream_build_root: Path | None = None,
) -> dict[str, int]:
    base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    access_token = request_exp_local_access_token(envv)
    should_run_directory_sync = bool(envv.get("EXP_HUBSPOT_ACCESS_TOKEN", "").strip())
    mappings = load_exp_hubspot_owner_binding_bootstrap(runtime_script_dir)

    created = 0
    updated = 0
    verified_locked = 0
    unresolved_mappings: list[str] = []
    expected_bindings: list[dict[str, str]] = []
    for mapping in mappings:
        payload = resolve_exp_hubspot_owner_binding_payload(base_url, access_token, mapping)
        if payload is None:
            unresolved_mappings.append(mapping["normalized_email"])
            continue
        expected_bindings.append(payload)
        result = upsert_exp_hubspot_owner_binding(base_url, access_token, payload)
        if result == "created":
            created += 1
        elif result == "updated":
            updated += 1
        elif result == "verified_locked":
            verified_locked += 1
    log(
        "applied EXP HubSpot owner bootstrap mapping "
        f"rows={len(mappings)} created={created} updated={updated} "
        f"verified_locked={verified_locked} unresolved={len(unresolved_mappings)}"
    )
    if unresolved_mappings:
        preview = ", ".join(unresolved_mappings[:5])
        suffix = "" if len(unresolved_mappings) <= 5 else ", ..."
        log(
            "skipping direct EXP HubSpot owner bootstrap for mappings without exactly one active local user: "
            f"{preview}{suffix}"
        )

    run_deploy_time_hubspot_owner_directory_lifecycle_sync(
        envv,
        access_token=access_token,
        downstream_build_root=downstream_build_root,
    )

    verify_exp_hubspot_owner_bootstrap_counts(
        base_url,
        access_token,
        expected_bindings=expected_bindings,
        require_directory_sync=should_run_directory_sync,
    )
    return {
        "resolved_bindings": len(expected_bindings),
        "unresolved_mappings": len(unresolved_mappings),
    }


def run_deploy_time_hubspot_owner_directory_lifecycle_sync(
    envv: dict[str, str],
    *,
    access_token: str | None = None,
    downstream_build_root: Path | None = None,
) -> bool:
    if downstream_build_root is not None:
        jobs = l1_post_activation_jobs_for_runtime_packs(
            envv,
            downstream_build_root,
            phase="postBackendHealth",
        )
        if not jobs:
            log("no L1 post-activation jobs to run for phase=postBackendHealth")
            return False
        ran = run_l1_post_activation_jobs(
            envv,
            downstream_build_root,
            phase="postBackendHealth",
            access_token=access_token,
        )
        if ran == 0:
            if access_token is None:
                access_token = request_exp_local_access_token(envv)
            base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
            log("reconciling HubSpot owner directory against the current portal")
            sync_hubspot_owner_directory_bindings_from_api(
                envv,
                base_url,
                access_token,
            )
            verify_hubspot_owner_directory_lifecycle_effect(
                base_url,
                access_token,
            )
        return True

    if not bool(envv.get("EXP_HUBSPOT_ACCESS_TOKEN", "").strip()):
        log("skipping deploy-time HubSpot owner directory sync; EXP_HUBSPOT_ACCESS_TOKEN is not set")
        return False

    base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    if access_token is None:
        access_token = request_exp_local_access_token(envv)

    return run_hubspot_owner_directory_reconcile_job(
        envv,
        base_url,
        access_token,
    )


def run_hubspot_owner_directory_reconcile_job(
    envv: dict[str, str],
    base_url: str,
    access_token: str,
) -> bool:
    log("running verified HubSpot owner-directory reconciliation")
    sync_hubspot_owner_directory_bindings_from_api(
        envv,
        base_url,
        access_token,
    )
    verify_hubspot_owner_directory_lifecycle_effect(
        base_url,
        access_token,
    )
    return True


def run_l1_post_activation_jobs(
    envv: dict[str, str],
    downstream_build_root: Path,
    *,
    phase: str,
    access_token: str | None = None,
) -> int:
    jobs = l1_post_activation_jobs_for_runtime_packs(envv, downstream_build_root, phase=phase)
    if not jobs:
        log(f"no L1 post-activation jobs to run for phase={phase}")
        return 0

    if access_token is None:
        access_token = request_exp_local_access_token(envv)
    base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    ran = 0
    for job in jobs:
        pack_name = str(job["_packName"])
        pack_dir = Path(str(job["_packDir"]))
        assert_active_pack_source_bound(
            envv, pack_name, downstream_build_root, pack_dir,
        )
        assert_l1_post_activation_job_active_flow_matches_source(envv, pack_name, pack_dir, job)
        source_fingerprint = source_fingerprint_for_l1_post_activation_job(job)
        if l1_post_activation_job_already_ran(envv, job, source_fingerprint):
            log(f"skipping L1 post-activation job already run for current source: {job['id']}")
            continue
        if invoke_l1_post_activation_job(
            envv,
            downstream_build_root,
            base_url,
            access_token,
            job,
        ):
            record_l1_post_activation_job_run(envv, job, source_fingerprint)
            ran += 1
    return ran


def invoke_l1_post_activation_salesforce_flow_job(
    base_url: str,
    access_token: str,
    job: dict[str, object],
) -> bool:
    action = job.get("action")
    if not isinstance(action, dict) or action.get("type") != "salesforceFlow":
        raise DeployError(f"unsupported L1 post-activation job action: {job.get('id')}")
    flow_api_name = str(action.get("flowApiName", "")).strip()
    inputs = action.get("inputs", [{}])
    if not flow_api_name:
        raise DeployError(f"L1 post-activation salesforceFlow job is missing flowApiName: {job.get('id')}")
    if not isinstance(inputs, list):
        raise DeployError(f"L1 post-activation salesforceFlow job inputs must be an array: {job.get('id')}")

    log(f"running L1 post-activation Salesforce Flow job: {job.get('id')}")
    try:
        payload = api_json_request(
            base_url,
            f"/services/data/v63.0/actions/custom/flow/{quote(flow_api_name, safe='')}",
            access_token=access_token,
            method="POST",
            payload={"inputs": inputs},
        )
    except DeployError as exc:
        message = str(exc)
        missing_flow_action = str(action.get("missingFlowAction", "")).strip()
        if (
            missing_flow_action == "skip"
            and "actions/custom/flow" in message
            and "http 404" in message
            and '"message":"Flow"' in message
        ):
            log(
                "L1 post-activation Salesforce Flow job is not invocable via actions/custom/flow; "
                f"continuing without direct flow execution: {job.get('id')}"
            )
            return False
        raise

    if not isinstance(payload, list) or len(payload) != 1 or not isinstance(payload[0], dict):
        raise DeployError(f"L1 post-activation Salesforce Flow job returned malformed response: {job.get('id')}")
    first_result = payload[0]
    if not bool(first_result.get("isSuccess")):
        errors = first_result.get("errors")
        error_message = json.dumps(errors, ensure_ascii=True)[:500]
        raise DeployError(
            "L1 post-activation Salesforce Flow job failed: "
            f"job={job.get('id')} {error_message or json.dumps(first_result, ensure_ascii=True)[:500]}"
        )
    return True


def invoke_l1_post_activation_job(
    envv: dict[str, str],
    downstream_build_root: Path,
    base_url: str,
    access_token: str,
    job: dict[str, object],
) -> bool:
    action = job.get("action")
    action_type = (
        str(action.get("type", "")).strip()
        if isinstance(action, dict)
        else ""
    )
    if action_type == "salesforceFlow":
        return invoke_l1_post_activation_salesforce_flow_job(
            base_url,
            access_token,
            job,
        )
    if action_type == "hubspotOwnerDirectoryReconcile":
        return run_hubspot_owner_directory_reconcile_job(
            envv,
            base_url,
            access_token,
        )
    raise DeployError(
        f"unsupported L1 post-activation job action: {job.get('id')}"
    )


def git_head_metadata(repo_root: Path) -> dict[str, str]:
    raw = run(
        ["git", "log", "-1", "--format=%cI|%s"],
        cwd=repo_root,
        capture=True,
    ).strip()
    if "|" not in raw:
        raise DeployError(f"unexpected git head metadata format for {repo_root}: {raw}")
    commit_time, subject = raw.split("|", 1)
    try:
        normalized_commit_time = commit_time[:-1] + "+00:00" if commit_time.endswith("Z") else commit_time
        commit_time_utc = datetime.fromisoformat(normalized_commit_time).astimezone(timezone.utc).isoformat().replace(
            "+00:00", "Z"
        )
    except ValueError as exc:
        raise DeployError(f"unexpected git head commit time for {repo_root}: {commit_time}") from exc
    return {
        "commit_time": commit_time,
        "commit_time_utc": commit_time_utc,
        "subject": subject,
    }


def load_exp_deploy_env(env_file: Path) -> tuple[dict[str, str], Path]:
    required_keys = [
        "ENV_NAME",
        "ENV_ROOT",
        "DOWNSTREAM_REPO_URL",
        "DOWNSTREAM_REF",
        "UPSTREAM_REPO_URL",
        "UPSTREAM_REF",
        "OFFICIAL_APP_PACKS_REPO_URL",
        "OFFICIAL_APP_PACKS_REF",
        "BACKEND_PORT",
        "PUBLIC_BASE_URL",
        "SERVER_NAME",
        "TIDB_HOST",
        "TIDB_PORT",
        "TIDB_USER",
        "TIDB_DATABASE",
        "JWT_SECRET",
        "NEXUS_ADMIN_PASSWORD",
        "GH_TOKEN",
        "DEPLOY_ARTIFACT_DIR",
        "RESET_DB",
    ]
    optional_keys = [
        "INSTALL_SYSTEMD",
        "INSTALL_NGINX",
        "TIDB_PASSWORD",
        "TIDB_TLS",
        "RESET_TIDB_BRANCH",
        "GOOGLE_CLIENT_ID",
        "GOOGLE_CLIENT_SECRET",
        "REACT_APP_GOOGLE_CLIENT_ID",
        "NEXUS_LOGIN_METHODS",
        "EXP_HUBSPOT_APP_NAME",
        "EXP_HUBSPOT_APP_ID",
        "EXP_HUBSPOT_CLIENT_ID",
        "EXP_HUBSPOT_PROJECT_ID",
        "EXP_HUBSPOT_AUTH_TYPE",
        "EXP_HUBSPOT_SUBSCRIPTION_COUNT",
        "EXP_HUBSPOT_WEBHOOK_URL",
        "EXP_HUBSPOT_CLIENT_SECRET",
        "EXP_HUBSPOT_ACCESS_TOKEN",
        "EXP_HUBSPOT_ACTION_OPTIONS_TOKEN",
        "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT",
        "FILE_STORAGE_PROVIDER",
        "FILE_STORAGE_ROOT",
        "FILE_STORAGE_LEGACY_UPLOAD_ROOT",
        "DEPLOY_ID",
        "DEPLOY_ARTIFACT_DIR",
        "AUDIT_FINAL_PACK",
        "SKIP_BACKEND_START",
        "APPLY_SHARED_DEPLOY_BASELINE",
        "SYSTEMD_USER",
        "BACKEND_GO_BUILD_TAGS",
    ]
    source_env_file = env_file.expanduser().resolve()
    envv = load_env_file(source_env_file)
    require_env(envv, required_keys)
    for key in optional_keys:
        envv.setdefault(key, "")
    return envv, source_env_file


def ensure_exp_database_ready(envv: dict[str, str]) -> None:
    ensure_tidb_database(envv)
    if bool_env(envv, "RESET_DB"):
        reset_database_if_requested(envv)
        return
    table_count = tidb_database_table_count(envv)
    if table_count != 0:
        raise DeployError(
            "exp runtime with RESET_DB=0 requires an empty case-scoped database; "
            f"found {table_count} tables in {envv['TIDB_DATABASE']}"
        )


rebind_installed_pack_source_local_path = common_rebind_installed_pack_source_local_path
normalize_installed_pack_runtime_state = common_normalize_installed_pack_runtime_state
rebind_retryable_installed_pack_source_identity = common_rebind_retryable_installed_pack_source_identity
sync_installed_pack_runtime_metadata = common_sync_installed_pack_runtime_metadata
sync_installed_pack_runtime_metadata_without_marking_applied = (
    common_sync_installed_pack_runtime_metadata_without_marking_applied
)


def load_installed_pack_applied_payload_hash(envv: dict[str, str], pack_name: str) -> str:
    escaped_name = pack_name.replace("'", "''")
    row = db_query_scalar(
        envv,
        textwrap.dedent(
            f"""
            SELECT JSON_OBJECT(
                '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("applied_payload_hash", "")).strip()


def is_valid_pack_payload_hash_v1(value: str) -> bool:
    value = value.strip()
    if not value.startswith(PACK_PAYLOAD_HASH_V1_PREFIX):
        return False
    digest = value[len(PACK_PAYLOAD_HASH_V1_PREFIX) :]
    return len(digest) == 64 and all(char in "0123456789abcdef" for char in digest)


def derive_stale_pack_payload_hash(current_hash: str) -> str:
    current_hash = current_hash.strip()
    if not is_valid_pack_payload_hash_v1(current_hash):
        raise DeployError(f"unexpected pack payload hash format: {current_hash!r}")
    digest = current_hash[len(PACK_PAYLOAD_HASH_V1_PREFIX) :]
    replacement = "0" if digest[-1] != "0" else "1"
    return f"{PACK_PAYLOAD_HASH_V1_PREFIX}{digest[:-1]}{replacement}"


def exp_pre_apply_applied_payload_hash_sql(envv: dict[str, str], pack_name: str, current_hash: str) -> str:
    applied_payload_hash = load_installed_pack_applied_payload_hash(envv, pack_name)
    if is_valid_pack_payload_hash_v1(applied_payload_hash) and applied_payload_hash != current_hash:
        return "AppliedPayloadHash"
    return sql_string_literal(derive_stale_pack_payload_hash(current_hash))


def sync_installed_pack_runtime_metadata_preserving_pre_apply_drift(
    envv: dict[str, str],
    pack_name: str,
    pack_dir: Path,
    nexuscrm_worktree: Path,
) -> None:
    common_sync_installed_pack_runtime_metadata_without_marking_applied(
        envv, pack_name, pack_dir, nexuscrm_worktree
    )
    pack_category, stored_manifest_format_version, _ = load_installed_pack_runtime_boundary(envv, pack_name)
    metadata = compute_source_backed_pack_runtime_metadata(
        nexuscrm_worktree,
        pack_dir,
        envv=envv,
        pack_name=pack_name,
        pack_category=pack_category,
        stored_manifest_format_version=stored_manifest_format_version,
        stored_source_revision=envv.get("DOWNSTREAM_REF", "").strip(),
        backend_go_build_tags=envv.get("BACKEND_GO_BUILD_TAGS"),
    )
    applied_payload_hash_sql = exp_pre_apply_applied_payload_hash_sql(
        envv,
        pack_name,
        str(metadata["payload_hash"]),
    )
    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"]),
    )
    escaped_name = pack_name.replace("'", "''")
    # Keep Desired != Applied before setup apply so new pack-owned projections replay.
    db_exec(
        envv,
        textwrap.dedent(
            f"""
            UPDATE _System_InstalledPack
            SET Status='active',
                PackVersion={sql_string_literal(str(metadata["pack_version"]))},
                ManifestFormatVersion={sql_string_literal(str(metadata["manifest_format_version"]))},
                PlatformCompatibility={sql_nullable_string_literal(
                    str(metadata["platform_compatibility"]),
                    bool(metadata["platform_compatibility_valid"]),
                )},
                DesiredPayloadHash={sql_string_literal(str(metadata["payload_hash"]))},
                AppliedPayloadHash={applied_payload_hash_sql},
                PayloadHashAlgorithm={sql_string_literal(str(metadata["payload_hash_algorithm"]))},
                SourceKind={source_kind},
                SourceURI={source_uri},
                SourceRevision={sql_nullable_string_literal(
                    str(metadata["source_revision"]),
                    bool(metadata["source_revision_valid"]),
                )},
                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 rebind_runtime_packs(envv: dict[str, str], downstream_build_root: Path) -> None:
    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    for pack_path in runtime_deploy_pack_paths(stack_lock):
        pack_dir = downstream_build_root / pack_path
        pack_json = pack_dir / "pack.json"
        if not pack_json.is_file():
            continue
        pack_name = json.loads(pack_json.read_text()).get("name")
        if not pack_name:
            continue
        log(f"rebinding installed pack source: {pack_name} -> {pack_dir}")
        rebind_installed_pack_source_local_path(envv, pack_name, pack_dir)
        log(f"normalizing installed pack runtime state: {pack_name}")
        normalize_installed_pack_runtime_state(envv, pack_name, pack_dir)


def finalize_runtime_packs(envv: dict[str, str], downstream_build_root: Path) -> None:
    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    for pack_path in runtime_deploy_pack_paths(stack_lock):
        pack_dir = downstream_build_root / pack_path
        pack_json = pack_dir / "pack.json"
        if not pack_json.is_file():
            continue
        pack_name = json.loads(pack_json.read_text()).get("name")
        if not pack_name:
            continue
        log(f"finalizing installed pack runtime state: {pack_name}")
        normalize_installed_pack_runtime_state(envv, pack_name, pack_dir)


def sync_runtime_pack_metadata(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    *,
    include_runtime_declarations: bool = True,
) -> None:
    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    for pack_path in runtime_deploy_pack_paths(stack_lock):
        pack_dir = downstream_build_root / pack_path
        pack_json = pack_dir / "pack.json"
        if not pack_json.is_file():
            continue
        pack_name = json.loads(pack_json.read_text()).get("name")
        if not pack_name:
            continue
        log(f"syncing installed pack runtime metadata: {pack_name}")
        sync_installed_pack_runtime_metadata(
            envv,
            pack_name,
            pack_dir,
            nexuscrm_worktree,
            include_runtime_declarations=include_runtime_declarations,
        )


def sync_runtime_pack_metadata_before_setup_apply(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
) -> None:
    stack_lock = json.loads((downstream_build_root / "stack.lock.json").read_text())
    for pack_path in runtime_deploy_pack_paths(stack_lock):
        pack_dir = downstream_build_root / pack_path
        pack_json = pack_dir / "pack.json"
        if not pack_json.is_file():
            continue
        pack_name = json.loads(pack_json.read_text()).get("name")
        if not pack_name:
            continue
        log(f"syncing installed pack runtime metadata before setup apply: {pack_name}")
        sync_installed_pack_runtime_metadata_preserving_pre_apply_drift(
            envv,
            pack_name,
            pack_dir,
            nexuscrm_worktree,
        )


def apply_runtime_and_finalize(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
    app_packs_root: Path,
    *,
    reason: str,
    run_setup_apply: bool = True,
) -> None:
    if run_setup_apply:
        log("synchronizing installed pack payload hashes before setup apply")
        sync_runtime_pack_metadata_before_setup_apply(envv, downstream_build_root, nexuscrm_worktree)
        attempts = int(envv.get("SETUP_RETRY_ATTEMPTS", "1"))
        delay_seconds = int(envv.get("SETUP_RETRY_DELAY_SECONDS", "15"))
        last_error: Exception | None = None
        for attempt in range(1, attempts + 1):
            log(f"setup apply attempt={attempt}/{attempts}")
            try:
                run_backend_setup(envv, nexuscrm_worktree, app_packs_root, ["apply"])
                last_error = None
                break
            except Exception as exc:
                last_error = exc
                if attempt == attempts:
                    break
                time.sleep(delay_seconds)
        if last_error is not None:
            raise DeployError(str(last_error))
        log("synchronizing installed pack payload hashes after setup apply")
    else:
        log("skipping full setup apply after explicit runtime pack install")
        log("synchronizing installed pack payload hashes after runtime pack install")
    sync_runtime_pack_metadata(
        envv,
        downstream_build_root,
        nexuscrm_worktree,
        include_runtime_declarations=not run_setup_apply,
    )
    log("finalizing installed pack runtime state after runtime apply")
    finalize_runtime_packs(envv, downstream_build_root)
    assert_owner_directory_lifecycle_active_flows_for_runtime_packs(envv, downstream_build_root)
    refresh_setup_derived_state(
        envv,
        nexuscrm_worktree,
        app_packs_root,
        reason=reason,
    )


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


def exp_seed_apply_password(envv: dict[str, str]) -> str:
    # Temporary seed-only password to bypass "same as current user password" validation.
    return f"{envv['NEXUS_ADMIN_PASSWORD']}-seed"


@dataclass(frozen=True)
class BulkSeedResult:
    state_file: Path | None
    summary: dict[str, object]

    @property
    def objects(self) -> list[str]:
        return [
            str(item)
            for item in self.summary.get("bulk_objects", [])
            if str(item).strip()
        ]

    @property
    def coverage_complete(self) -> bool:
        return bool(self.state_file is not None and self.objects and self.summary.get("coverage_complete"))

    @property
    def records(self) -> int:
        return int(self.summary.get("bulk_record_count", 0) or 0)

    @property
    def aliases(self) -> int:
        return int(self.summary.get("bulk_alias_count", 0) or 0)

    @property
    def fallback_aliases(self) -> int:
        return int(self.summary.get("fallback_alias_count", 0) or 0)


def _regular_file_within(root: Path, relative_path: Path, *, label: str) -> Path:
    resolved_root = root.expanduser().resolve()
    candidate = resolved_root / relative_path
    try:
        resolved = candidate.resolve(strict=True)
    except OSError as exc:
        raise DeployError(f"{label} not found: {candidate}") from exc
    try:
        resolved.relative_to(resolved_root)
    except ValueError as exc:
        raise DeployError(f"{label} escapes its authority root: {relative_path}") from exc
    if not resolved.is_file():
        raise DeployError(f"{label} is not a regular file: {resolved}")
    return resolved


def runtime_app_shell_config(downstream_build_root: Path) -> Path:
    stack_lock_path = _regular_file_within(
        downstream_build_root,
        Path("stack.lock.json"),
        label="stack lock",
    )
    stack_lock = json.loads(stack_lock_path.read_text())
    config_rel = stack_lock.get("app_shell_config")
    if not isinstance(config_rel, str) or not config_rel.strip():
        raise DeployError("stack.lock.json must provide app_shell_config for runtime app-shell apply")
    if "\\" in config_rel or "\x00" in config_rel:
        raise DeployError("stack.lock.json app_shell_config must be a canonical relative path")
    relative_path = Path(config_rel)
    if relative_path.is_absolute() or any(part in {"", ".", ".."} for part in relative_path.parts):
        raise DeployError("stack.lock.json app_shell_config must be a canonical relative path")
    return _regular_file_within(
        downstream_build_root,
        relative_path,
        label="app shell config",
    )


def apply_runtime_app_shell(
    envv: dict[str, str],
    downstream_build_root: Path,
    *,
    controller_root: Path | None = None,
    output_path: Path | None = None,
    expected_config_sha256: str = "",
    navigation_catalog_image: str = "",
    navigation_pages: Path | None = None,
    controlplane_image: str = "",
    local_candidate: Path | None = None,
    local_controlplane_source_revision: str = "",
) -> dict[str, object]:
    config_file = runtime_app_shell_config(downstream_build_root)
    executable_root = (controller_root or downstream_build_root).expanduser().resolve()
    apply_script = _regular_file_within(
        executable_root,
        Path("scripts/apply_runtime_app_shell.py"),
        label="runtime app-shell controller",
    )
    target_output = output_path or Path(envv["DEPLOY_ARTIFACT_DIR"]) / "app-shell.json"
    command = [
        os.environ.get("PYTHON", "python3"),
        str(apply_script),
        "--config",
        str(config_file),
        "--output",
        str(target_output),
    ]
    if expected_config_sha256:
        command.extend(["--expected-config-sha256", expected_config_sha256])
    if controlplane_image:
        command.extend(["--controlplane-image", controlplane_image])
    if local_candidate is not None:
        command.extend(["--local-candidate", str(local_candidate)])
    if local_controlplane_source_revision:
        command.extend(["--local-controlplane-source-revision", local_controlplane_source_revision])
    if navigation_catalog_image:
        if navigation_pages is None:
            raise DeployError("navigation initialization requires the sealed L1 page projection")
        command.extend(["--navigation-catalog-image", navigation_catalog_image,
                        "--navigation-pages", str(navigation_pages)])
    run(
        command,
        cwd=executable_root,
        env={
            "TIDB_HOST": envv["TIDB_HOST"],
            "TIDB_PORT": envv["TIDB_PORT"],
            "TIDB_USER": envv["TIDB_USER"],
            "TIDB_PASSWORD": envv.get("TIDB_PASSWORD", ""),
            "TIDB_DATABASE": envv["TIDB_DATABASE"],
            "TIDB_TLS": str(use_tidb_tls(envv)).lower(),
        },
    )
    try:
        result = json.loads(target_output.read_text())
    except (OSError, json.JSONDecodeError) as exc:
        raise DeployError("runtime app-shell apply did not write valid evidence") from exc
    if result.get("status") != "passed":
        raise DeployError("runtime app-shell apply did not pass its metadata assertions")
    return result


def seed_common_env(envv: dict[str, str]) -> dict[str, str]:
    backend_base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    return {
        "WATCHDOG_BASE_URL": backend_base_url,
        "WATCHDOG_BACKEND_BASE_URL": backend_base_url,
        "WATCHDOG_ADMIN_USERNAME": "admin@example.invalid",
        "WATCHDOG_ADMIN_PASSWORD": envv["NEXUS_ADMIN_PASSWORD"],
        "EXP_SEED_APPLY_PASSWORD": exp_seed_apply_password(envv),
    }


def seed_bulk_env(envv: dict[str, str], nexuscrm_worktree: Path) -> dict[str, str]:
    return {
        **seed_common_env(envv),
        "TIDB_HOST": envv["TIDB_HOST"],
        "TIDB_PORT": envv["TIDB_PORT"],
        "TIDB_USER": envv["TIDB_USER"],
        "TIDB_PASSWORD": envv.get("TIDB_PASSWORD", ""),
        "TIDB_DATABASE": envv["TIDB_DATABASE"],
        "TIDB_TLS": envv.get("TIDB_TLS", "true"),
        "EXP_BULK_HASH_USER_PASSWORDS": "0",
        "NEXUSCRM_WORKTREE": str(nexuscrm_worktree),
        "NEXUSCRM_BACKEND_DIR": str(nexuscrm_worktree / "backend"),
    }


def apply_bulk_seed_file(
    envv: dict[str, str],
    downstream_build_root: Path,
    seed_file: Path,
    nexuscrm_worktree: Path,
) -> BulkSeedResult:
    bulk_script = downstream_build_root / "scripts" / "apply_exp_bulk_baseline.sh"
    if not bulk_script.is_file():
        raise DeployError(f"missing EXP bulk seed script: {bulk_script}")

    output_dir = Path(tempfile.mkdtemp(prefix="exp-bulk-baseline-"))
    state_file = output_dir / "runtime-state.json"
    summary_file = output_dir / "summary.json"
    run(
        [
            "bash",
            str(bulk_script),
            "--seed-file",
            str(seed_file),
            "--output-dir",
            str(output_dir),
            "--state-file",
            str(state_file),
            "--summary-file",
            str(summary_file),
        ],
        cwd=downstream_build_root,
        env=seed_bulk_env(envv, nexuscrm_worktree),
    )
    summary = json.loads(summary_file.read_text())
    return BulkSeedResult(state_file=state_file, summary=summary)


def payload_from_bulk_state(state_file: Path) -> dict[str, object]:
    bulk_state = json.loads(state_file.read_text())
    return {
        "applied_records": [],
        "failed_records": [],
        "hydrated_skipped_records": [
            {
                "key": key,
                "object": state.get("object"),
                "action": "hydrated_from_base_state",
                "id": state.get("id"),
            }
            for key, state in (bulk_state.get("records") or {}).items()
        ],
    }


def apply_seed_with_bulk(
    envv: dict[str, str],
    downstream_build_root: Path,
    seed_file: Path,
    nexuscrm_worktree: Path,
    *,
    label: str,
) -> tuple[dict[str, object], BulkSeedResult]:
    bulk_result = apply_bulk_seed_file(
        envv,
        downstream_build_root,
        seed_file,
        nexuscrm_worktree,
    )

    bulk_objects = ",".join(bulk_result.objects)
    log(
        f"{label} bulk seed "
        f"objects={bulk_objects or 'none'} "
        f"records={bulk_result.records} "
        f"aliases={bulk_result.aliases} "
        f"fallback_aliases={bulk_result.fallback_aliases} "
        f"duration_ms={bulk_result.summary.get('duration_ms', 'unknown')} "
        f"coverage_complete={bulk_result.coverage_complete}"
    )
    if bulk_result.summary.get("unsupported"):
        log(f"{label} bulk seed unsupported={bulk_result.summary.get('unsupported')}")
    if bulk_result.summary.get("fallback_objects"):
        log(f"{label} bulk seed fallback_objects={bulk_result.summary.get('fallback_objects')}")

    if not bulk_result.coverage_complete or bulk_result.state_file is None:
        raise DeployError(
            f"{label} bulk seed coverage is incomplete; REST fallback is retired"
        )
    return payload_from_bulk_state(bulk_result.state_file), bulk_result


def apply_exp_baseline_seed(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
) -> tuple[dict[str, object], BulkSeedResult]:
    seed_file = downstream_build_root / "deployments" / "offline" / "runtime" / "seeds" / "exp-baseline.seed.yaml"
    payload, bulk_result = apply_seed_with_bulk(
        envv,
        downstream_build_root,
        seed_file,
        nexuscrm_worktree,
        label="EXP baseline",
    )

    return payload, bulk_result


def apply_shared_deploy_baseline_seed(
    envv: dict[str, str],
    downstream_build_root: Path,
    nexuscrm_worktree: Path,
) -> tuple[dict[str, object], BulkSeedResult]:
    seed_file = downstream_build_root / "deployments" / "offline" / "runtime" / "seeds" / "deploy-baseline.seed.yaml"
    payload, bulk_result = apply_seed_with_bulk(
        envv,
        downstream_build_root,
        seed_file,
        nexuscrm_worktree,
        label="shared deploy baseline",
    )
    applied_count = len(payload.get("applied_records", []))
    hydrated_count = len(payload.get("hydrated_skipped_records", []) or [])
    log(f"shared deploy baseline materialized records={applied_count + hydrated_count}")
    return payload, bulk_result


def deploy_seed_dispatch_env(
    envv: dict[str, str],
    *,
    profile: str,
    nexuscrm_worktree: Path | None,
) -> dict[str, str]:
    allowed_keys = (
        *DEPLOY_SEED_SAFE_PROCESS_ENV_KEYS,
        *DEPLOY_SEED_CONDITION_ENV_KEYS,
        *DEPLOY_SEED_ADAPTER_ENV_KEYS,
        *DEPLOY_SEED_PROFILE_ENV_KEYS.get(profile, ()),
    )
    dispatch_env = {
        "PATH": str(envv.get("PATH") or os.environ.get("PATH") or os.defpath)
    }
    for name in allowed_keys:
        if name in envv:
            dispatch_env[name] = str(envv[name])
    dispatch_env["PYTHONDONTWRITEBYTECODE"] = "1"
    if nexuscrm_worktree is not None:
        dispatch_env[DEPLOY_SEED_WORKTREE_ENV_KEY] = str(nexuscrm_worktree)
    # Each connected_apps run-as ConsumerSecret is a deploy-time input. Source
    # each from the curated runtime env, else the runner process environment.
    # Do not silently disable the default module when one is absent: the adapter
    # resolves all env references before its first write and fails closed there,
    # and staging/EXP also check the runner prerequisite at their deployment
    # entrypoints before mutation.
    for name in CONNECTED_APPS_CONSUMER_SECRET_ENV_KEYS:
        value = str(envv.get(name) or os.environ.get(name) or "").strip()
        if value:
            dispatch_env[name] = value
    return dispatch_env


def load_deploy_seed_evidence(
    evidence_path: Path,
    *,
    profile: str,
    phase: str,
    execution_mode: str,
) -> dict[str, object]:
    try:
        evidence = json.loads(evidence_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise DeployError(
            f"deploy Seed Dispatcher did not emit valid evidence: {evidence_path}"
        ) from exc
    if not isinstance(evidence, dict) or set(evidence) != DEPLOY_SEED_EVIDENCE_KEYS:
        raise DeployError(
            f"deploy Seed Dispatcher evidence fields do not match the contract: {evidence_path}"
        )
    if evidence.get("schema_version") != 1 or evidence.get("format") != "deploy-seed-apply-v1":
        raise DeployError(
            f"deploy Seed Dispatcher evidence format does not match the contract: {evidence_path}"
        )
    if evidence.get("profile") != profile:
        raise DeployError(
            f"deploy Seed Dispatcher evidence profile does not match {profile}: {evidence_path}"
        )
    if evidence.get("phase") != phase:
        raise DeployError(
            f"deploy Seed Dispatcher evidence phase does not match {phase}: {evidence_path}"
        )
    if evidence.get("execution_mode") != execution_mode:
        raise DeployError(
            "deploy Seed Dispatcher evidence execution mode does not match "
            f"{execution_mode}: {evidence_path}"
        )
    if evidence.get("status") != "passed" or evidence.get("failure") is not None:
        raise DeployError(
            f"deploy Seed Dispatcher evidence did not pass: {evidence_path}"
        )
    if not isinstance(evidence.get("catalog_version"), str) or not str(
        evidence["catalog_version"]
    ).strip():
        raise DeployError(
            f"deploy Seed Dispatcher evidence has no catalog version: {evidence_path}"
        )
    fingerprint = evidence.get("source_fingerprint_sha256")
    if (
        not isinstance(fingerprint, str)
        or len(fingerprint) != 64
        or any(character not in "0123456789abcdef" for character in fingerprint)
    ):
        raise DeployError(
            f"deploy Seed Dispatcher evidence has an invalid source fingerprint: {evidence_path}"
        )
    if evidence.get("plan_file") != "seed-plan.json" or evidence.get(
        "inventory_file"
    ) != "deploy-seed-inventory.json":
        raise DeployError(
            f"deploy Seed Dispatcher evidence artifact names do not match the contract: {evidence_path}"
        )

    invocations = evidence.get("invocations")
    if not isinstance(invocations, list) or not invocations:
        raise DeployError(
            f"deploy Seed Dispatcher evidence has no invocations: {evidence_path}"
        )
    expected_status = "planned" if execution_mode == "dry-run" else "applied"
    invocation_ids: set[str] = set()
    for invocation in invocations:
        if (
            not isinstance(invocation, dict)
            or set(invocation) != DEPLOY_SEED_INVOCATION_EVIDENCE_KEYS
        ):
            raise DeployError(
                f"deploy Seed Dispatcher invocation evidence fields do not match the contract: {evidence_path}"
            )
        invocation_id = invocation.get("invocation_id")
        if (
            not isinstance(invocation_id, str)
            or not invocation_id
            or invocation_id in invocation_ids
        ):
            raise DeployError(
                f"deploy Seed Dispatcher invocation identity is invalid: {evidence_path}"
            )
        invocation_ids.add(invocation_id)
        if not isinstance(invocation.get("module_id"), str) or not invocation["module_id"]:
            raise DeployError(
                f"deploy Seed Dispatcher invocation module is invalid: {evidence_path}"
            )
        if (
            isinstance(invocation.get("order"), bool)
            or not isinstance(invocation.get("order"), int)
            or invocation["order"] < 0
        ):
            raise DeployError(
                f"deploy Seed Dispatcher invocation order is invalid: {evidence_path}"
            )
        if not isinstance(invocation.get("selection"), dict) or not isinstance(
            invocation.get("condition"), dict
        ):
            raise DeployError(
                f"deploy Seed Dispatcher invocation selection or condition is invalid: {evidence_path}"
            )
        if not isinstance(invocation.get("adapter_id"), str) or not invocation["adapter_id"]:
            raise DeployError(
                f"deploy Seed Dispatcher invocation adapter is invalid: {evidence_path}"
            )
        status = invocation.get("status")
        if status == "skipped":
            if invocation.get("adapter_output_dir") is not None or invocation.get("summary") is not None:
                raise DeployError(
                    f"deploy Seed Dispatcher skipped invocation evidence is invalid: {evidence_path}"
                )
            continue
        if (
            status != expected_status
            or not isinstance(invocation.get("adapter_output_dir"), str)
            or not invocation["adapter_output_dir"]
            or not isinstance(invocation.get("summary"), dict)
        ):
            raise DeployError(
                f"deploy Seed Dispatcher completed invocation evidence is invalid: {evidence_path}"
            )
    return evidence


def apply_deploy_seed_phase(
    envv: dict[str, str],
    downstream_build_root: Path,
    *,
    profile: str,
    phase: str,
    nexuscrm_worktree: Path | None = None,
    selected_stage: str | None = None,
    runtime_mode: str | None = None,
) -> dict[str, object]:
    if selected_stage is not None and runtime_mode is not None:
        raise DeployError("deploy Seed phase cannot select both a stage and a runtime mode")
    output_scope = f"{profile}-{phase}"
    if selected_stage is not None:
        output_scope += f"-{selected_stage}"
    if runtime_mode is not None:
        output_scope += f"-{runtime_mode}"
    output_root = (
        Path(envv["DEPLOY_ARTIFACT_DIR"]).expanduser().resolve()
        / "seed-dispatch"
    )
    output_root.mkdir(parents=True, exist_ok=True)
    output_dir = Path(tempfile.mkdtemp(prefix=f"{output_scope}-", dir=output_root))
    log(f"applying deploy Seed profile={profile} phase={phase}")
    dispatch_env = deploy_seed_dispatch_env(
        envv,
        profile=profile,
        nexuscrm_worktree=nexuscrm_worktree,
    )
    command = [
        "python3",
        str(downstream_build_root / "scripts" / "apply_deploy_seed.py"),
        "--source-root",
        str(downstream_build_root),
        "--profile",
        profile,
        "--phase",
        phase,
        "--execution-mode",
        "apply",
        "--output-dir",
        str(output_dir),
    ]
    if selected_stage is not None:
        command.extend(("--selected-stage", selected_stage))
    if runtime_mode is not None:
        command.extend(("--runtime-mode", runtime_mode))
    run(
        command,
        cwd=downstream_build_root,
        env=dispatch_env,
        inherit_env=False,
    )
    evidence_path = output_dir / "seed-apply-evidence.json"
    return load_deploy_seed_evidence(
        evidence_path,
        profile=profile,
        phase=phase,
        execution_mode="apply",
    )


def pingcap_region_owner_rules_seed_file(downstream_build_root: Path) -> Path:
    return downstream_build_root / "deployments" / "offline" / "runtime" / "seeds" / "pingcap-region-owner-rules.seed.yaml"


def parse_seed_scalar(value: str) -> object:
    normalized = value.strip().strip('"').strip("'")
    lowered = normalized.lower()
    if lowered == "true":
        return True
    if lowered == "false":
        return False
    return normalized


def load_simple_seed_sections(seed_file: Path, section_names: set[str]) -> dict[str, list[dict[str, object]]]:
    if not seed_file.is_file():
        raise DeployError(f"missing PingCAP region owner rule seed file: {seed_file}")
    sections = {name: [] for name in section_names}
    current_section_name = ""
    current: dict[str, object] | None = None
    current_record_section = ""

    def flush_current() -> None:
        nonlocal current
        if current_section_name in sections and current is not None:
            sections[current_section_name].append(current)
        current = None

    for raw_line in seed_file.read_text().splitlines():
        stripped = raw_line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if not raw_line.startswith(" "):
            flush_current()
            section_name = stripped[:-1] if stripped.endswith(":") else ""
            current_section_name = section_name if section_name in sections else ""
            current_record_section = ""
            continue
        if current_section_name not in sections:
            continue
        if raw_line.startswith("  - key:"):
            flush_current()
            current = {
                "key": raw_line.split(":", 1)[1].strip(),
                "match": {},
                "fields": {},
            }
            current_record_section = ""
            continue
        if current is None:
            continue
        if stripped in {"match:", "fields:"}:
            current_record_section = stripped[:-1]
            continue
        if ":" not in stripped:
            continue
        key, value = stripped.split(":", 1)
        key = key.strip()
        parsed_value = parse_seed_scalar(value)
        if current_record_section in {"match", "fields"}:
            section = current[current_record_section]
            assert isinstance(section, dict)
            section[key] = parsed_value
        elif current_record_section == "":
            current[key] = parsed_value
    flush_current()
    return sections


def lookup_key_from_seed_reference(owner_ref: str, *, row_index: int) -> str:
    expression = owner_ref.strip()
    if not expression.startswith("{{") or not expression.endswith("}}"):
        raise DeployError(
            f"PingCAP region owner seed row {row_index} must resolve Owner_Id__c from a lookup, not {owner_ref!r}"
        )
    expression = expression.removeprefix("{{").removesuffix("}}").strip()
    if not expression.startswith("lookup.") or not expression.endswith(".id"):
        raise DeployError(
            f"PingCAP region owner seed row {row_index} must resolve Owner_Id__c from a lookup, not {owner_ref!r}"
        )
    lookup_key = expression.removeprefix("lookup.").removesuffix(".id").strip()
    if not lookup_key:
        raise DeployError(f"PingCAP region owner seed row {row_index} has an empty Owner_Id__c lookup")
    return lookup_key


def load_pingcap_region_owner_rules(seed_file: Path) -> list[dict[str, object]]:
    sections = load_simple_seed_sections(seed_file, {"lookup_records", "baseline_records"})
    records = sections["baseline_records"]
    lookups_by_key = {
        str(record.get("key") or "").strip(): record
        for record in sections["lookup_records"]
        if str(record.get("key") or "").strip()
    }

    if not records:
        raise DeployError(f"PingCAP region owner rule seed has no baseline_records: {seed_file}")

    rules: list[dict[str, object]] = []
    seen_regions: set[str] = set()
    for index, record in enumerate(records, start=1):
        if not isinstance(record, dict):
            raise DeployError(f"PingCAP region owner seed row {index} must be an object")
        if record.get("object") != "Partner_Region_Owner_Rule__c":
            continue
        match = record.get("match")
        fields = record.get("fields")
        if not isinstance(match, dict) or not isinstance(fields, dict):
            raise DeployError(f"PingCAP region owner seed row {index} missing match/fields")
        region = str(match.get("Region__c") or fields.get("Region__c") or "").strip()
        owner_name = str(fields.get("Owner_Name__c") or "").strip()
        owner_ref = str(fields.get("Owner_Id__c") or "").strip()
        if not region or not owner_name:
            raise DeployError(f"PingCAP region owner seed row {index} missing region or owner name")
        lookup_key = lookup_key_from_seed_reference(owner_ref, row_index=index)
        lookup = lookups_by_key.get(lookup_key)
        if lookup is None:
            raise DeployError(f"PingCAP region owner seed row {index} references missing lookup {lookup_key!r}")
        if lookup.get("object") != "User":
            raise DeployError(f"PingCAP region owner lookup {lookup_key!r} must target User")
        lookup_match = lookup.get("match")
        if not isinstance(lookup_match, dict) or not lookup_match:
            raise DeployError(f"PingCAP region owner lookup {lookup_key!r} must define a non-empty match")
        if region in seen_regions:
            raise DeployError(f"duplicate PingCAP region owner seed region: {region}")
        seen_regions.add(region)
        rules.append(
            {
                "region": region,
                "owner_name": owner_name,
                "owner_lookup_key": lookup_key,
                "owner_lookup_match": dict(lookup_match),
            }
        )

    if len(rules) != PINGCAP_REGION_OWNER_RULE_COUNT:
        raise DeployError(
            "PingCAP region owner seed must define exactly "
            f"{PINGCAP_REGION_OWNER_RULE_COUNT} Partner_Region_Owner_Rule__c rows; found {len(rules)}"
        )
    if len({rule["owner_name"] for rule in rules}) != len(rules):
        raise DeployError("PingCAP region owner seed must map each region to a distinct owner name")

    return rules


def pingcap_region_owner_lead_probes(rules: list[dict[str, str]]) -> list[dict[str, str]]:
    special = {probe["rule_region"]: probe for probe in PINGCAP_REGION_OWNER_LEAD_PROBES}
    probes: list[dict[str, str]] = []
    for rule in rules:
        region = rule["region"]
        probes.append(
            special.get(
                region,
                {
                    "rule_region": region,
                    "lead_region": region,
                    "country": "United States",
                },
            )
        )
    return probes


def soql_literal(value: object) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    return f"'{soql_string_literal(str(value))}'"


def expected_pingcap_region_owner_user_ids(
    base_url: str,
    access_token: str,
    rules: list[dict[str, object]],
) -> dict[str, str]:
    expected: dict[str, str] = {}
    for rule in rules:
        owner_name = str(rule["owner_name"])
        user_match = rule.get("owner_lookup_match")
        if not isinstance(user_match, dict) or not user_match:
            raise DeployError(f"PingCAP region owner lookup match missing for {owner_name!r}")
        where_clause = " AND ".join(
            f"{field} = {soql_literal(value)}"
            for field, value in user_match.items()
        )
        records = exp_soql_records(
            base_url,
            access_token,
            f"SELECT Id FROM User WHERE {where_clause}",
        )
        if len(records) != 1:
            raise DeployError(
                "PingCAP region owner rule bootstrap requires exactly one User matching "
                f"{owner_name!r} via {user_match}; found {len(records)}"
            )
        user_id = str(records[0].get("Id") or "").strip()
        if not user_id:
            raise DeployError(f"PingCAP region owner User lookup missing Id for {owner_name!r}")
        expected[owner_name] = user_id
    return expected


def upsert_pingcap_region_owner_rule(
    base_url: str,
    access_token: str,
    *,
    region: str,
    owner_name: str,
    owner_id: str,
) -> dict[str, str]:
    records = exp_soql_records(
        base_url,
        access_token,
        (
            "SELECT Id, Region__c, Owner_Name__c, Owner_Id__c "
            "FROM Partner_Region_Owner_Rule__c "
            f"WHERE Region__c = '{soql_string_literal(region)}'"
        ),
    )
    if len(records) > 1:
        raise DeployError(
            "PingCAP region owner rule bootstrap found duplicate rows for "
            f"region {region!r}: {len(records)}"
        )

    payload = {
        "Region__c": region,
        "Owner_Name__c": owner_name,
        "Owner_Id__c": owner_id,
    }
    if records:
        record_id = str(records[0].get("Id") or "").strip()
        if not record_id:
            raise DeployError(f"PingCAP region owner rule lookup missing Id for region {region!r}")
        api_json_request(
            base_url,
            f"/services/data/v63.0/sobjects/Partner_Region_Owner_Rule__c/{quote(record_id, safe='')}",
            access_token=access_token,
            method="PATCH",
            payload=payload,
        )
        return {"id": record_id, "action": "updated"}

    created = api_json_request(
        base_url,
        "/services/data/v63.0/sobjects/Partner_Region_Owner_Rule__c",
        access_token=access_token,
        method="POST",
        payload=payload,
    )
    record_id = str(created.get("id") or created.get("Id") or "").strip()
    if not record_id:
        raise DeployError(f"PingCAP region owner rule create missing id for region {region!r}")
    return {"id": record_id, "action": "created"}


def apply_pingcap_region_owner_rules_via_api(
    base_url: str,
    access_token: str,
    *,
    seed_file: Path,
) -> dict[str, object]:
    rules = load_pingcap_region_owner_rules(seed_file)
    expected_user_ids = expected_pingcap_region_owner_user_ids(base_url, access_token, rules)
    applied_records: list[dict[str, object]] = []
    resolved_lookups: list[dict[str, object]] = []

    for rule in rules:
        region = str(rule["region"])
        owner_name = str(rule["owner_name"])
        owner_id = expected_user_ids[owner_name]
        resolved_lookups.append(
            {
                "object": "User",
                "match": rule["owner_lookup_match"],
                "id": owner_id,
            }
        )
        result = upsert_pingcap_region_owner_rule(
            base_url,
            access_token,
            region=region,
            owner_name=owner_name,
            owner_id=owner_id,
        )
        applied_records.append(
            {
                "object": "Partner_Region_Owner_Rule__c",
                "match": {"Region__c": region},
                "id": result["id"],
                "action": result["action"],
            }
        )

    return {
        "applied_records": applied_records,
        "hydrated_skipped_records": [],
        "resolved_lookups": resolved_lookups,
    }


def verify_pingcap_region_owner_lead_assignment(
    base_url: str,
    access_token: str,
    expected_user_ids_by_region: dict[str, str],
    rules: list[dict[str, object]],
) -> None:
    verified = 0
    for probe in pingcap_region_owner_lead_probes(rules):
        rule_region = probe["rule_region"]
        lead_id = ""
        email = f"deploy-region-owner-{rule_region.lower().replace('-', '')}-{uuid.uuid4().hex[:12]}@example.invalid"
        try:
            created = api_json_request(
                base_url,
                "/services/data/v63.0/sobjects/Lead",
                access_token=access_token,
                method="POST",
                payload={
                    "LastName": f"RegionOwner{rule_region.replace('-', '')}",
                    "Company": "PingCAP Region Owner Deploy Probe",
                    "Email": email,
                    "Region__c": probe["lead_region"],
                    "Country": probe["country"],
                    "Description": "Temporary deploy probe for PingCAP region owner routing.",
                },
            )
            lead_id = str(created.get("id") or created.get("Id") or "").strip()
            if not lead_id:
                raise DeployError(f"PingCAP region owner Lead probe create missing id for {rule_region}")
            readback = api_json_request(
                base_url,
                (
                    f"/services/data/v63.0/sobjects/Lead/{quote(lead_id, safe='')}"
                    "?fields=Id,OwnerId,Region__c,Country,Email"
                ),
                access_token=access_token,
            )
            actual_owner_id = str(readback.get("OwnerId") or "").strip()
            expected_owner_id = expected_user_ids_by_region[rule_region]
            if actual_owner_id != expected_owner_id:
                raise DeployError(
                    "PingCAP region owner Lead probe mismatch for "
                    f"{rule_region}: expected OwnerId={expected_owner_id}, found {actual_owner_id}"
                )
            verified += 1
        finally:
            if lead_id:
                api_json_request(
                    base_url,
                    f"/services/data/v63.0/sobjects/Lead/{quote(lead_id, safe='')}",
                    access_token=access_token,
                    method="DELETE",
                )
    log(f"verified PingCAP region owner Lead assignment probes regions={verified}")


def verify_pingcap_region_owner_rules(
    envv: dict[str, str],
    *,
    seed_file: Path,
    require_lead_assignment_verify: bool,
) -> None:
    base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    access_token = request_exp_local_access_token(envv)
    rules = load_pingcap_region_owner_rules(seed_file)
    expected_user_ids = expected_pingcap_region_owner_user_ids(base_url, access_token, rules)
    expected_user_ids_by_region = {
        str(rule["region"]): expected_user_ids[str(rule["owner_name"])]
        for rule in rules
    }

    verified = 0
    for rule in rules:
        region = str(rule["region"])
        owner_name = str(rule["owner_name"])
        records = exp_soql_records(
            base_url,
            access_token,
            (
                "SELECT Id, Region__c, Owner_Name__c, Owner_Id__c "
                "FROM Partner_Region_Owner_Rule__c "
                f"WHERE Region__c = '{soql_string_literal(region)}'"
            ),
        )
        if len(records) != 1:
            raise DeployError(
                "PingCAP region owner rule bootstrap requires exactly one rule row for "
                f"region {region!r}; found {len(records)}"
            )
        record = records[0]
        actual_owner_name = str(record.get("Owner_Name__c") or "").strip()
        actual_owner_id = str(record.get("Owner_Id__c") or "").strip()
        expected_owner_id = expected_user_ids[owner_name]
        if actual_owner_name != owner_name or actual_owner_id != expected_owner_id:
            raise DeployError(
                "PingCAP region owner rule mismatch for "
                f"{region}: expected owner={owner_name!r}/{expected_owner_id}, "
                f"found owner={actual_owner_name!r}/{actual_owner_id}"
            )
        verified += 1
    log(f"verified PingCAP region owner rules regions={verified}")
    if require_lead_assignment_verify:
        verify_pingcap_region_owner_lead_assignment(
            base_url,
            access_token,
            expected_user_ids_by_region,
            rules,
        )
    else:
        log(
            "skipping PingCAP region owner Lead assignment probes "
            "(not required for this deploy path)"
        )


def hubspot_options_adapter_service_name(envv: dict[str, str]) -> str:
    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,
    runtime_script_dir: Path,
) -> Path:
    runner = env_root / "run_hubspot_options_adapter.sh"
    adapter_script = runtime_script_dir / "hubspot_workflow_options_adapter.py"
    lines = [
        "#!/usr/bin/env bash",
        "set -euo pipefail",
        "set -a",
        f'source "{env_file}"',
        "set +a",
        "",
        'export PATH="/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin:$PATH"',
        f'export EXP_HUBSPOT_OPTIONS_ADAPTER_PORT="{exp_hubspot_options_adapter_port(envv)}"',
        f'exec python3 -B "{adapter_script}"',
        "",
    ]
    runner.write_text("\n".join(lines))
    runner.chmod(0o755)
    return runner


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_user = envv.get("SYSTEMD_USER", "").strip()
    if not systemd_user:
        raise DeployError("missing required env keys: SYSTEMD_USER")
    systemd_service = hubspot_options_adapter_service_name(envv)
    rendered = render_template(
        framework_dir / "systemd/nexuscrm-hubspot-options-adapter.service.template",
        {
            "__ENV_NAME__": envv["ENV_NAME"],
            "__SYSTEMD_USER__": systemd_user,
            "__DOWNSTREAM_ROOT__": str(downstream_root),
            "__ENV_FILE__": str(env_file),
            "__ENV_ROOT__": str(env_root),
        },
    )
    sudo_write_text(Path("/etc/systemd/system") / systemd_service, rendered)
    run(["sudo", "systemctl", "daemon-reload"])
    run(["sudo", "systemctl", "enable", systemd_service])


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


def start_hubspot_options_adapter(envv: dict[str, str], env_root: Path, log_dir: Path, run_dir: Path) -> 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_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", "start", hubspot_options_adapter_service_name(envv)])
    wait_for_health(
        f"http://127.0.0.1:{exp_hubspot_options_adapter_port(envv)}/hubspot/workflow-actions/options/health",
        attempts=30,
    )


if __name__ == "__main__":
    raise SystemExit("Classic monolithic deployment is retired. Use the serviceified D3 deployment controller. This module only supplies shared deployment helpers.")