#!/usr/bin/env python3 """Deploy one signed-BOM D3 release to the next Prod slot, then cut over. This is the online slot adapter for the frozen D3 deployment engine. Its ownership boundary is intentionally narrow: one Prod D3 release root, one target port slot, its D3-labelled containers, its companion unit, and the one reviewed Prod nginx file. The previously routed runtime remains live through target Patch/Catalog convergence. After nginx cutover and public target readback, the operator stops the pending-authority-bound predecessor and verifies retirement before publishing current authority. """ from __future__ import annotations import hashlib import json import os import re import shutil import subprocess import sys import tempfile import urllib.parse from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Mapping, Sequence REPO_ROOT = Path(__file__).resolve().parent.parent ONLINE_RUNTIME = REPO_ROOT / "deployments" / "online" / "runtime" for path in (REPO_ROOT, ONLINE_RUNTIME): if str(path) not in sys.path: sys.path.insert(0, str(path)) from scripts import serviceification_d3_engine as impl # noqa: E402 from scripts.prod_nginx_access_log import strip_access_logging, with_access_logging # noqa: E402 from scripts import serviceification_d3_test_patch as d3_patch # noqa: E402 from prod_d3_ingress import ( # noqa: E402 PROD_D3_MARKER, PROD_D3_PUBLIC_PROXY_PORTS, PROD_SLOT_PORTS, PROD_SERVER_NAME, ProdRoute, next_slot, parse_prod_route, slot_for_port, ) from prepare_multi_prod_d3_runtime_env import ( # noqa: E402 BOM_RE, CURRENT_SCHEMA, PrepareError, clear_in_progress, load_env as load_prepared_prod_env, read_in_progress_authority, require_in_progress_authority, update_env as update_prepared_prod_env, ) DeployError = impl.DeployError STATE_SCHEMA = "nexuscrm.serviceification-d3-prod-state/v2" EVIDENCE_SCHEMA = "nexuscrm.serviceification-d3-prod-deployment-evidence/v3" COMPANION_SCHEMA = "nexuscrm.serviceification-d3-prod-workflow-action-adapter/v2" PROD_NGINX_CONF = Path("/etc/nginx/conf.d/nexuscrm-prod.conf") ACTIVE_SLOT = "" REQUESTED_PATCH_DIR = "" EXPECTED_PATCH_TREE_DIGEST = "" LAST_PATCH_REPLAY: dict[str, Any] | None = None _ORIG_LOAD_ACTIVE = impl.load_active_transition_context _ORIG_CONVERGE = impl.converge_deployment _ORIG_ENSURE_INGRESS = impl.ensure_d3_public_ingress _ORIG_VERIFY_SERVING = impl.verify_serving _ORIG_REPLACE_NGINX_SERVER = impl.replace_nginx_server_block _ORIG_APPLY_STOPPED_RUNTIME = impl.apply_stopped_upgrade_runtime _ORIG_APPLY_CATALOG = impl.apply_catalog _ORIG_IDEMPOTENCY = impl.catalog_input_idempotency_key @dataclass(frozen=True) class SlotContract: sfcrm_port: int adapter_port: int integration_port: int runtime_grpc: int runtime_admin: int worker_admin_base: int SLOT_CONTRACT = { "slot-3103": SlotContract(3103, 3204, 3403, 3701, 3702, 3710), "slot-3104": SlotContract(3104, 3205, 3413, 3801, 3802, 3810), "slot-3105": SlotContract(3105, 3206, 3423, 3901, 3902, 3910), "slot-3106": SlotContract(3106, 3207, 3433, 4101, 4102, 4110), } if { slot: contract.sfcrm_port for slot, contract in SLOT_CONTRACT.items() } != PROD_SLOT_PORTS: raise RuntimeError("Prod D3 slot contract disagrees with Prod ingress authority") if { slot: frozenset( {contract.sfcrm_port, contract.adapter_port, contract.integration_port} ) for slot, contract in SLOT_CONTRACT.items() } != PROD_D3_PUBLIC_PROXY_PORTS: raise RuntimeError("Prod D3 public proxy ports disagree with ingress authority") def prod_slot(envv: Mapping[str, str], *, require_declared: bool = True) -> str: env_name = str(envv.get("ENV_NAME", "")).strip() prefix = "multi-prod-d3-" if not env_name.startswith(prefix): raise DeployError("Prod D3 requires ENV_NAME=multi-prod-d3-slot-") slot = env_name.removeprefix(prefix) if slot not in SLOT_CONTRACT: raise DeployError(f"Prod D3 does not accept ENV_NAME={env_name}") declared = os.environ.get("D3_SLOT", "").strip() if require_declared and declared and declared not in {slot, f"prod-{slot}"}: raise DeployError(f"D3_SLOT={declared} disagrees with ENV_NAME={env_name}") return slot def bind_slot(slot: str) -> str: global ACTIVE_SLOT if slot not in SLOT_CONTRACT: raise DeployError(f"unknown Prod D3 slot: {slot}") ACTIVE_SLOT = slot return slot def container_label_value(slot: str) -> str: if slot not in SLOT_CONTRACT: raise DeployError(f"unknown Prod D3 slot: {slot}") return f"prod-{slot}" def _apply_impl_identity(slot: str) -> None: impl.CONTAINER_LABEL_VALUE = container_label_value(slot) impl.RUNTIME_SERVER_NAME = f"runtime-api.d3.prod-{slot}.local" impl.RUNTIME_API_ADMISSION_LIMIT = impl.DEFAULT_RUNTIME_API_ADMISSION_LIMIT impl.STATE_SCHEMA = STATE_SCHEMA impl.EVIDENCE_SCHEMA = EVIDENCE_SCHEMA impl.TEST2_NGINX_CONF = PROD_NGINX_CONF def live_route() -> ProdRoute: if PROD_NGINX_CONF.is_symlink() or not PROD_NGINX_CONF.is_file(): raise DeployError(f"Prod nginx authority is unavailable: {PROD_NGINX_CONF}") try: return parse_prod_route(PROD_NGINX_CONF.read_text(encoding="utf-8")) except ValueError as exc: raise DeployError(str(exc)) from exc def require_target_is_standby( envv: Mapping[str, str], *, env_root: Path | None = None, run_id: str = "" ) -> str: slot = prod_slot(envv) route = live_route() target_port = SLOT_CONTRACT[slot].sfcrm_port if route.backend_port != target_port: expected = next_slot(route) if slot != expected: raise DeployError( f"Prod D3 target {slot} is not the next slot after live {route.slot}" ) return slot if env_root is not None and run_id: state_file = env_root / ".runtime" / "serviceification-d3" / "active.json" if state_file.is_file() and not state_file.is_symlink(): state = impl.read_state(state_file) if str(state.get("runId", "")) == run_id: return slot raise DeployError(f"Prod D3 target {slot} is already the live route") def require_live_target(envv: Mapping[str, str]) -> ProdRoute: slot = prod_slot(envv) route = live_route() if route.backend_port != SLOT_CONTRACT[slot].sfcrm_port or not route.is_d3: raise DeployError( f"Prod D3 cutover did not publish target {slot}; live port={route.backend_port}" ) return route def derive_ports( envv: Mapping[str, str], worker_stems: Sequence[str] | str = impl.LEGACY_WORKER_STEMS, slot: str | None = None, ) -> impl.D3Ports: if isinstance(worker_stems, str): if slot is not None: raise DeployError("Prod D3 port derivation received duplicate slot authority") slot = worker_stems stems = impl.LEGACY_WORKER_STEMS else: stems = tuple(worker_stems) if set(stems) not in impl.READABLE_WORKER_STEM_SETS: raise DeployError("Prod D3 port derivation received unsupported Worker roles") env_name = str(envv.get("ENV_NAME", "")) if slot: resolved = slot elif env_name.startswith("multi-prod-d3-"): resolved = prod_slot(envv) elif ACTIVE_SLOT: # load_active_transition_context temporarily aliases ENV_NAME for the # frozen Test2 loader after binding the real Prod slot. resolved = ACTIVE_SLOT else: raise DeployError("Prod D3 port derivation has no slot authority") resolved = bind_slot(resolved) contract = SLOT_CONTRACT[resolved] backend = impl.require_port(envv.get("BACKEND_PORT", ""), "BACKEND_PORT") adapter = impl.require_port( impl.exp_hubspot_options_adapter_port(dict(envv)), "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT", ) if backend != contract.sfcrm_port or adapter != contract.adapter_port: raise DeployError("Prod D3 runtime env disagrees with its slot port contract") if str(envv.get("SERVER_NAME", "")).strip() != PROD_SERVER_NAME: raise DeployError("Prod D3 SERVER_NAME is not the production public host") workers = { stem: contract.worker_admin_base + index for index, stem in enumerate(stems) } result = impl.D3Ports( sfcrm_http=contract.sfcrm_port, integration_http=contract.integration_port, runtime_grpc=contract.runtime_grpc, runtime_admin=contract.runtime_admin, worker_admin=workers, ) values = set(result.all_ports().values()) values.add(contract.adapter_port) if len(values) != len(result.all_ports()) + 1: raise DeployError(f"Prod D3 {resolved} port allocation collides") for other_slot, other in SLOT_CONTRACT.items(): if other_slot == resolved: continue other_ports = { other.sfcrm_port, other.adapter_port, other.integration_port, other.runtime_grpc, other.runtime_admin, *(other.worker_admin_base + index for index in range(len(impl.SUPPORTED_WORKER_STEMS))), } if values & other_ports: raise DeployError( f"Prod D3 private port allocations overlap: {resolved}/{other_slot}" ) return result def container_name(run_id: str, role: str, *, slot: str = "") -> str: resolved = slot or ACTIVE_SLOT or os.environ.get("D3_SLOT", "").removeprefix("prod-") if resolved not in SLOT_CONTRACT: raise DeployError("Prod D3 container identity requires a physical port slot") normalized = re.sub( r"[^a-z0-9_.-]+", "-", f"nexuscrm-d3-prod-{resolved}-{run_id}-{role}".lower() ) if len(normalized) > 128: suffix = hashlib.sha256(normalized.encode()).hexdigest()[:16] normalized = normalized[:111].rstrip("-.") + "-" + suffix return normalized def load_prod_env_file(path: Path) -> dict[str, str]: """Read the shell-safe Prod env exactly as its systemd runner sources it.""" try: return load_prepared_prod_env(path) except PrepareError as exc: raise DeployError(f"invalid Prod D3 runtime env: {exc}") from exc def require_resolved_bom_matches_reference( resolved_bom: Path, reference: str ) -> None: """Prove the supplied BOM bytes came from the exact Tag-selected OCI digest.""" if not BOM_RE.fullmatch(reference): raise DeployError("Prod D3 runtime env has no exact resolved BOM reference") created = subprocess.run( ["docker", "create", reference, "/__nexuscrm_bom_identity_check__"], capture_output=True, text=True, check=False, ) container = created.stdout.strip() if created.returncode != 0 or not re.fullmatch(r"[0-9a-f]{64}", container): detail = created.stderr.strip() or f"docker create exited {created.returncode}" raise DeployError( "cannot inspect Tag-selected resolved BOM artifact: " + detail ) try: with tempfile.TemporaryDirectory(prefix="nexuscrm-prod-d3-bom-") as temp: embedded = Path(temp) / "resolved-bom.json" copied = subprocess.run( [ "docker", "cp", f"{container}:/nexuscrm/release/resolved-bom.json", str(embedded), ], capture_output=True, text=True, check=False, ) if copied.returncode != 0 or not embedded.is_file(): detail = copied.stderr.strip() or f"docker cp exited {copied.returncode}" raise DeployError( "cannot extract Tag-selected resolved BOM artifact: " + detail ) if embedded.read_bytes() != resolved_bom.read_bytes(): raise DeployError( "supplied Prod D3 BOM differs from the Tag-selected OCI artifact" ) finally: subprocess.run( ["docker", "rm", "-f", container], capture_output=True, text=True, check=False, ) def require_resolved_bom_release_identity( envv: Mapping[str, str], bom: Mapping[str, Any] ) -> None: tag_target_commit = str(envv.get("TAG_TARGET_COMMIT", "")).strip() if not re.fullmatch(r"[0-9a-f]{40}", tag_target_commit): raise DeployError("Prod D3 runtime env has no exact Tag target commit") l2_metadata = bom.get("l2_metadata") if ( not isinstance(l2_metadata, Mapping) or l2_metadata.get("commit") != tag_target_commit ): raise DeployError("resolved Prod D3 BOM L2 commit differs from the release Tag") def load_context(env_file: Path, resolved_bom: Path, run_id: str) -> impl.D3Context: if not impl.RUN_ID_RE.fullmatch(run_id): raise DeployError("Prod D3 run id is not canonical") env_file = env_file.expanduser().resolve() envv = load_prod_env_file(env_file) slot = bind_slot(prod_slot(envv)) _apply_impl_identity(slot) env_root_raw = str(envv.get("ENV_ROOT", "")).strip() if not env_root_raw: raise DeployError("Prod D3 runtime env has no ENV_ROOT") env_root = Path(env_root_raw).expanduser().resolve() expected_env = impl.runtime_paths(env_root)["runtime_env_file"].resolve() if env_file != expected_env: raise DeployError(f"Prod D3 runtime env must be canonical: {expected_env}") require_target_is_standby(envv, env_root=env_root, run_id=run_id) resolved_bom = resolved_bom.expanduser().resolve() require_resolved_bom_matches_reference( resolved_bom, str(envv.get("D3_RESOLVED_BOM_REFERENCE", "")).strip(), ) bom = impl.load_bom(resolved_bom) if impl.is_local_test3_bom(bom): raise DeployError("Prod does not accept unsigned Test3 local deployments") require_resolved_bom_release_identity(envv, bom) deployment_bom = impl.operational_bom(bom) worker_stems, worker_families = impl.worker_roles_from_bom(deployment_bom) state_root = env_root / ".runtime" / "serviceification-d3" return impl.D3Context( env_file=env_file, envv=envv, env_root=env_root, resolved_bom=resolved_bom, bom=deployment_bom, bom_file_digest=impl.sha256_bytes(resolved_bom.read_bytes()), run_id=run_id, run_root=state_root / "runs" / run_id, state_file=state_root / "active.json", ports=derive_ports(envv, worker_stems), images=impl.exact_images(deployment_bom), worker_stems=worker_stems, worker_families=worker_families, ) def load_active_transition_context( env_file: Path, run_id: str, *, allow_legacy_reset_receipt: bool = False, ) -> impl.D3Context: env_file = env_file.expanduser().resolve() envv = load_prod_env_file(env_file) slot = bind_slot(prod_slot(envv, require_declared=False)) _apply_impl_identity(slot) saved_load, saved_read, saved_derive = impl.load_env_file, impl.read_state, impl.derive_ports def aliased_load(path: Path) -> dict[str, str]: loaded = dict(saved_load(path)) loaded["ENV_NAME"] = "test-test2" return loaded def aliased_read(path: Path) -> dict[str, Any]: state = dict(saved_read(path)) if state.get("envName") == envv.get("ENV_NAME"): state["envName"] = "test-test2" return state impl.load_env_file, impl.read_state, impl.derive_ports = aliased_load, aliased_read, derive_ports try: context = _ORIG_LOAD_ACTIVE( env_file, run_id, allow_legacy_reset_receipt=allow_legacy_reset_receipt, ) finally: impl.load_env_file, impl.read_state, impl.derive_ports = saved_load, saved_read, saved_derive return impl.D3Context( env_file=context.env_file, envv=envv, env_root=context.env_root, resolved_bom=context.resolved_bom, bom=context.bom, bom_file_digest=context.bom_file_digest, run_id=context.run_id, run_root=context.run_root, state_file=context.state_file, ports=context.ports, images=context.images, worker_stems=context.worker_stems, worker_families=context.worker_families, ) def target_systemd_services(envv: dict[str, str]) -> tuple[str, str]: slot = prod_slot(envv, require_declared=False) port = SLOT_CONTRACT[slot].sfcrm_port return ( f"nexuscrm-multi-prod-d3-{slot}-hubspot-options-adapter.service", f"nexuscrm-multi-prod-slot-{port}-backend.service", ) def quiesce_target_runtime( envv: dict[str, str], _run_dir: Path, _downstream_build_root: Path | None = None, *, disable_autostart: bool, ) -> None: contract = SLOT_CONTRACT[prod_slot(envv, require_declared=False)] target_port = contract.sfcrm_port route = live_route() if route.backend_port == target_port: raise DeployError("Prod D3 refuses to quiesce the currently routed slot") enabled_states = {"enabled", "enabled-runtime", "linked", "linked-runtime"} for unit in target_systemd_services(envv): state = impl.staging.systemd_unit_state(unit) if state["load_state"] == "not-found": continue try: if state["active_state"] not in {"inactive", "failed", "dead"}: subprocess.run( ["sudo", "-n", "systemctl", "stop", unit], check=True ) if disable_autostart and state["unit_file_state"] in enabled_states: subprocess.run( ["sudo", "-n", "systemctl", "disable", unit], check=True ) except subprocess.CalledProcessError as exc: raise DeployError(f"failed to quiesce Prod D3 target unit: {unit}") from exc for unit in target_systemd_services(envv): state = impl.staging.systemd_unit_state(unit) if state["load_state"] == "not-found": continue if state["active_state"] not in {"inactive", "failed", "dead"}: raise DeployError(f"Prod D3 target unit remains active: {unit}") if disable_autostart and state["unit_file_state"] in enabled_states: raise DeployError(f"Prod D3 target unit remains enabled: {unit}") impl.staging.require_staging_target_port_listener_free( str(contract.adapter_port), "Prod D3 target workflow adapter" ) impl.staging.require_staging_target_port_listener_free( str(contract.sfcrm_port), "Prod D3 target backend" ) def _record_patch_replay(context: impl.D3Context) -> None: """Retain one successful Tag-bound replay for Runtime and Catalog apply.""" global LAST_PATCH_REPLAY if not REQUESTED_PATCH_DIR or LAST_PATCH_REPLAY is not None: return replay = d3_patch.replay_patch_before_catalog( context, requested_patch_dir=REQUESTED_PATCH_DIR, environment="prod", read_state=impl.read_state, ) if replay is None: raise DeployError("Prod D3 patch was not replayed before Runtime/Catalog apply") LAST_PATCH_REPLAY = replay def apply_stopped_upgrade_runtime( context: impl.D3Context, *args: Any, **kwargs: Any ) -> dict[str, Any]: """Match Staging: replay after target quiescence, before pack/Flow apply.""" inner = kwargs.pop("apply_runtime", impl.staging.run_catalog_offline_candidate_phase) def apply_runtime_after_quiesce(*runtime_args: Any, **runtime_kwargs: Any) -> Any: # The engine has re-proved target quiescence before invoking this seam. # Ownership patches must precede source-pack reconciliation, not just # the later Catalog publisher, or customized Flow content is preserved. _record_patch_replay(context) return inner(*runtime_args, **runtime_kwargs) kwargs["apply_runtime"] = apply_runtime_after_quiesce return _ORIG_APPLY_STOPPED_RUNTIME(context, *args, **kwargs) def apply_catalog(context: impl.D3Context, *args: Any, **kwargs: Any) -> dict[str, Any]: """Re-prove target quiescence and retain the pre-Catalog replay fallback.""" quiesce_target_runtime( dict(context.envv), impl.runtime_paths(context.env_root)["run_dir"], disable_autostart=True, ) _record_patch_replay(context) return _ORIG_APPLY_CATALOG(context, *args, **kwargs) def preserve_prod_database(context: impl.D3Context) -> dict[str, Any]: state = impl.require_state(context, expected_phase="artifacts-ready") if impl.owned_containers(): raise DeployError("Prod D3 database adoption refuses target containers") exact_sources = state.get("exactSources") if not isinstance(exact_sources, dict): raise DeployError("Prod D3 database adoption has no exact L2 source") l2_root = Path(str(exact_sources.get("l2Root", ""))).resolve() pack_names = impl.runtime_pack_names(l2_root) if not pack_names: raise DeployError("Prod D3 database adoption resolved no Runtime packs") baseline: dict[str, Any] = { # The frozen engine uses this token for the generic non-reset path. # retainedEnvironment carries the online meaning without widening its # reset-receipt vocabulary. "source": "retained-test2-database", "retainedEnvironment": "prod", "targetSlot": prod_slot(context.envv), "resetPerformed": False, "runtimePacks": sorted(pack_names), } if not impl.catalog_authority_tables_present(context.envv): baseline["installationOrgId"] = impl.installation_org_id() return impl.replace_state( context, expected_phase="artifacts-ready", updates={ "phase": "baseline-reset", "baselineReset": baseline, "catalogStart": None, "catalogBefore": None, "workerOwnershipTableBefore": None, "workerOwnershipBefore": None, "runtimeApply": None, }, ) def reestablish_prod_serving_quiescence( context: impl.D3Context, state: Mapping[str, Any], *, execute: Callable[[Sequence[str]], None], capture_output: Callable[[Sequence[str]], str], ) -> None: """Stop only a partial target retry while the routed predecessor stays live.""" baseline = state.get("baselineReset") if ( not isinstance(baseline, dict) or baseline.get("source") not in impl.RESET_SNAPSHOT_BASELINE_SOURCES ): return expected_images = impl.container_images_by_name(context, state) present = set(impl.owned_containers()) if not present.issubset(expected_images): raise DeployError( "Prod D3 serving retry found a target container outside the exact run" ) for name in reversed(impl.expected_serving_container_order(state)): if name not in present: continue if impl.verify_container_identity( context, name, expected_images[name], capture_output=capture_output ): execute(("docker", "stop", "--time", "30", name)) quiesce_target_runtime( dict(context.envv), impl.runtime_paths(context.env_root)["run_dir"], disable_autostart=True, ) def catalog_input_idempotency_key(*args: Any, **kwargs: Any) -> str: key = _ORIG_IDEMPOTENCY(*args, **kwargs) if key.startswith("d3-test2-"): return "d3-prod-" + key.removeprefix("d3-test2-") return key def skip_prod_inspection_fixtures( _envv: dict[str, str], _l2_root: Path, **_kwargs: Any ) -> dict[str, Any]: return {"status": "skipped", "reason": "prod-does-not-own-test-fixtures"} def skip_prod_inspection_identity( context: impl.D3Context, _core_root: Path, **_kwargs: Any ) -> dict[str, Any]: state = impl.require_state(context) if state.get("testIdentityAligned") is not None: return state return impl.replace_state( context, expected_phase=str(state["phase"]), updates={"testIdentityAligned": {"skipped": True, "environment": "prod"}}, ) def skip_prod_governed_reference_data(context: impl.D3Context) -> dict[str, Any]: state = impl.require_state( context, expected_phase={"catalog-applied", "governed-reference-data-in-progress"}, ) impl.verify_catalog_after(context, state) if state["phase"] == "catalog-applied": state = impl.replace_state( context, expected_phase="catalog-applied", updates={"phase": "governed-reference-data-in-progress"}, ) return impl.replace_state( context, expected_phase="governed-reference-data-in-progress", updates={ "phase": "governed-reference-data-applied", "governedReferenceData": { "schema": "nexuscrm.serviceification-d3-prod-reference-data/v1", "status": "preserved", "mutationCount": 0, }, }, ) def render_d3_nginx_server(context: impl.D3Context, frontend: Path) -> str: slot = prod_slot(context.envv) rendered = impl.render_template( REPO_ROOT / "deployments/offline/nginx/nexuscrm-env.conf.template", { "__NGINX_LISTEN_PORT__": "80", "__NGINX_FORWARDED_PROTO__": ( urllib.parse.urlparse(context.envv["PUBLIC_BASE_URL"]).scheme or "https" ), "__SERVER_NAME__": PROD_SERVER_NAME, "__FRONTEND_DIST__": str(frontend), "__BACKEND_PORT__": str(context.ports.sfcrm_http), "__CONNECTED_MAIL_INGRESS_BLOCK__": impl.connected_mail_ingress_block( context.ports.integration_http, urllib.parse.urlparse(context.envv["PUBLIC_BASE_URL"]).scheme or "https", ), "__WEBHOOK_PORT__": str(context.ports.integration_http), "__HEALTH_PORT__": str(context.ports.sfcrm_http), "__HEALTH_PATH__": "/readyz", "__EXP_HUBSPOT_OPTIONS_ADAPTER_PORT__": str(SLOT_CONTRACT[slot].adapter_port), "__ACCESS_LOG__": f"/var/log/nginx/nexuscrm-prod-d3-{slot}.access.log", "__ERROR_LOG__": f"/var/log/nginx/nexuscrm-prod-d3-{slot}.error.log", }, ) rendered = impl.route_mcp_ingress(rendered, context) if "__" in rendered: raise DeployError("Prod D3 nginx rendering left an unresolved token") return with_access_logging(rendered, slot, context.env_root.name, marker=f"{PROD_D3_MARKER} run={context.run_id}") def replace_prod_nginx_server_block(content: str, server_name: str, replacement: str) -> str: if server_name != PROD_SERVER_NAME: raise DeployError("Prod ingress replacement must target its owned server") return _ORIG_REPLACE_NGINX_SERVER(strip_access_logging(content), server_name, replacement) def capture_previous_nginx( context: impl.D3Context, state: Mapping[str, Any] ) -> dict[str, Any]: existing_path = state.get("previousNginxConfig") existing_digest = state.get("previousNginxConfigDigest") if isinstance(existing_path, str) and isinstance(existing_digest, str): path = Path(existing_path) if path.is_symlink() or not path.is_file() or impl.sha256_bytes(path.read_bytes()) != existing_digest: raise DeployError("Prod D3 lost its nginx rollback authority") return dict(state) if existing_path is not None or existing_digest is not None: raise DeployError("Prod D3 nginx rollback authority is incomplete") content = PROD_NGINX_CONF.read_bytes() backup = context.run_root / "authority" / "nginx-before-d3.conf" backup.parent.mkdir(mode=0o700, parents=True, exist_ok=True) if backup.exists() or backup.is_symlink(): raise DeployError("Prod D3 nginx rollback path already exists") backup.write_bytes(content) os.chmod(backup, 0o400) return impl.replace_state( context, expected_phase=str(state["phase"]), updates={ "previousNginxConfig": str(backup), "previousNginxConfigDigest": impl.sha256_bytes(content), }, ) COMPANION_FILES = ( "deployments/offline/runtime/hubspot_workflow_options_adapter.py", "deployments/offline/runtime/common.py", "scripts/deployment_bundle.py", ) def companion_activation_authority() -> tuple[Path, str, dict[str, str]]: source_root = REPO_ROOT.resolve() commit = impl.default_capture( ("git", "-C", str(source_root), "rev-parse", "HEAD") ) if not re.fullmatch(r"[0-9a-f]{40}", commit): raise DeployError("Prod D3 activation checkout has no exact commit") dirty = impl.default_capture( ( "git", "-C", str(source_root), "status", "--porcelain", "--", *COMPANION_FILES, ) ) if dirty: raise DeployError("Prod D3 activation companion sources are modified") digests: dict[str, str] = {} for relative in COMPANION_FILES: source = source_root / relative if source.is_symlink() or not source.is_file(): raise DeployError(f"Prod D3 companion source is unavailable: {relative}") digests[relative] = impl.sha256_bytes(source.read_bytes()) return source_root, commit, digests def _companion_receipt( context: impl.D3Context, state: Mapping[str, Any] ) -> dict[str, Any]: source_root, expected_commit, expected_digests = ( companion_activation_authority() ) _, client_id, client_secret = impl.signed_l2_programmatic_oauth_authority( context, state, label="Prod D3 workflow-action adapter" ) if not str(context.envv.get("EXP_HUBSPOT_ACTION_OPTIONS_TOKEN", "")).strip(): raise DeployError("Prod D3 companion is missing EXP_HUBSPOT_ACTION_OPTIONS_TOKEN") companion_root = context.run_root / "companions" / "workflow-action-adapter" companion_root.mkdir(mode=0o700, parents=True, exist_ok=True) for relative in COMPANION_FILES: source = source_root / relative destination = companion_root / relative destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) if not destination.exists(): shutil.copyfile(source, destination) os.chmod(destination, 0o400) if ( destination.is_symlink() or destination.read_bytes() != source.read_bytes() ): raise DeployError(f"Prod D3 companion source changed: {relative}") if impl.sha256_bytes(destination.read_bytes()) != expected_digests[relative]: raise DeployError(f"Prod D3 companion digest changed: {relative}") adapter = impl.staging.deploy_offline_env adapter_env = dict(context.envv) adapter_updates = { "EXP_NEXUS_CLIENT_ID": client_id, "EXP_NEXUS_CLIENT_SECRET": client_secret, "EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_URL": ( impl.workflow_action_adapter_replay_url(context.ports) ), } adapter_env.update(adapter_updates) update_prepared_prod_env(context.env_file, adapter_updates) paths = impl.runtime_paths(context.env_root) adapter.write_hubspot_options_adapter_runner( adapter_env, context.env_file, context.env_root, companion_root / "deployments/offline/runtime", ) adapter.install_hubspot_options_adapter_unit( adapter_env, source_root / "deployments/offline", companion_root, context.env_file, context.env_root, ) adapter.start_hubspot_options_adapter( adapter_env, context.env_root, paths["log_dir"], paths["run_dir"] ) port = SLOT_CONTRACT[prod_slot(context.envv)].adapter_port adapter.wait_for_health( f"http://127.0.0.1:{port}/hubspot/workflow-actions/options/health", attempts=30, ) return { "schema": COMPANION_SCHEMA, "service": target_systemd_services(dict(context.envv))[0], "port": port, "sourceAuthority": "reviewed-activation-checkout", "sourceCommit": expected_commit, "sourceDigests": expected_digests, "oauthClientId": client_id, "oauthClientSecretSha256": impl.sha256_bytes(client_secret.encode()), "oauthL2Commit": str(context.bom["l2_metadata"]["commit"]), "replayUrl": adapter_env["EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_URL"], } def converge_companion(context: impl.D3Context) -> dict[str, Any]: state = impl.require_state( context, expected_phase={"serving-start-in-progress", "serving"} ) receipt = state.get("prodWorkflowActionAdapter") if not isinstance(receipt, Mapping): receipt = _companion_receipt(context, state) previous = dict(state.get("previousSystemdUnits", {})) previous.pop(str(receipt["service"]), None) state = impl.replace_state( context, expected_phase=str(state["phase"]), updates={ "prodWorkflowActionAdapter": dict(receipt), "previousSystemdUnits": previous, }, ) _, activation_commit, activation_digests = companion_activation_authority() service = str(receipt.get("service", "")) port = receipt.get("port") # The receipt was created from the verified exact L2 tree. Retained checks # must not depend on the Actions checkout surviving the next deployment. client_id = str(receipt.get("oauthClientId", "")) secret_digest = str(receipt.get("oauthClientSecretSha256", "")) persisted_env = load_prod_env_file(context.env_file) client_secret = persisted_env.get("EXP_NEXUS_CLIENT_SECRET", "") replay_url = impl.workflow_action_adapter_replay_url(context.ports) if ( receipt.get("schema") != COMPANION_SCHEMA or receipt.get("sourceAuthority") != "reviewed-activation-checkout" or receipt.get("sourceCommit") != activation_commit or receipt.get("sourceDigests") != activation_digests or service != target_systemd_services(dict(context.envv))[0] or port != SLOT_CONTRACT[prod_slot(context.envv)].adapter_port or not client_id or not client_secret or re.fullmatch(r"sha256:[0-9a-f]{64}", secret_digest) is None or secret_digest != impl.sha256_bytes(client_secret.encode()) or receipt.get("oauthL2Commit") != context.bom["l2_metadata"]["commit"] or persisted_env.get("EXP_NEXUS_CLIENT_ID") != client_id or receipt.get("replayUrl") != replay_url or persisted_env.get("EXP_HUBSPOT_WORKFLOW_ACTION_REPLAY_URL") != replay_url ): raise DeployError( "Prod D3 companion receipt disagrees with activation authority " "or target slot" ) if impl.staging.systemd_unit_state(service).get("active_state") != "active": raise DeployError("Prod D3 companion service is not active") return state def ensure_prod_ingress(context: impl.D3Context) -> dict[str, Any]: converge_companion(context) # Shared serving recovery restarts SFCRM before reaching this ingress gate. impl.verify_workflow_action_adapter_options( SLOT_CONTRACT[prod_slot(context.envv)].adapter_port, str(context.envv["EXP_HUBSPOT_ACTION_OPTIONS_TOKEN"]), ) return _ORIG_ENSURE_INGRESS(context) def verify_prod_serving( context: impl.D3Context, *args: Any, **kwargs: Any ) -> dict[str, Any]: state = converge_companion(context) evidence = _ORIG_VERIFY_SERVING(context, *args, **kwargs) receipt = state.get("prodWorkflowActionAdapter") if not isinstance(receipt, Mapping): raise DeployError("verified Prod D3 state lost its companion receipt") result = {**evidence, "companionProcessCount": 1, "workflowActionAdapter": dict(receipt)} impl.replace_state( context, expected_phase="serving", updates={"deploymentEvidence": result}, ) return result def archive_target_generation(context: impl.D3Context) -> None: if context.state_file.exists() or context.state_file.is_symlink(): active = impl.read_state(context.state_file) if str(active.get("runId", "")) == context.run_id: return for name in impl.owned_containers(): impl.default_execute(("docker", "rm", "-f", name)) remaining = impl.owned_containers() if remaining: raise DeployError( "Prod D3 target leftovers could not be removed: " + ", ".join(remaining) ) if not (context.state_file.exists() or context.state_file.is_symlink()): return active = impl.read_state(context.state_file) archived = context.state_file.parent / f"retired-{active.get('runId', 'unknown')}-{hashlib.sha256(context.run_id.encode()).hexdigest()[:12]}.json" if archived.exists() or archived.is_symlink(): raise DeployError(f"Prod D3 archive already exists: {archived}") os.replace(context.state_file, archived) os.chmod(archived, 0o600) def publish_current(context: impl.D3Context, route: ProdRoute) -> None: d3_root = context.env_root.parent.parent if d3_root.name != "d3" or context.env_root.parent.name != "releases": raise DeployError("Prod D3 release root is outside multi_prod/d3/releases") state = impl.require_state( context, expected_phase="serving", allow_legacy_reset_receipt=True ) retirement = state.get("previousRuntimeRetirement") if ( not isinstance(retirement, dict) or retirement.get("schema") != "nexuscrm.prod-predecessor-retirement/v1" or retirement.get("verified") is not True or str(retirement.get("backendPort")) != context.envv.get("PROD_D3_PREVIOUS_BACKEND_PORT") or retirement.get("kind") != context.envv.get("PROD_D3_PREVIOUS_ROUTE_KIND") or retirement.get("targetBackendPort") != route.backend_port or retirement.get("backendPort") == route.backend_port ): raise DeployError("Prod current publication requires verified predecessor retirement") current = d3_root / "current" if current.exists() and not current.is_symlink(): raise DeployError("Prod D3 current pointer is not a symlink") marker = { "schema": CURRENT_SCHEMA, "releaseTag": context.envv.get("RELEASE_TAG"), "slot": prod_slot(context.envv), "backendPort": route.backend_port, "deploymentRunId": context.run_id, "resolvedBomReference": context.envv.get("D3_RESOLVED_BOM_REFERENCE"), "desiredBomDigest": context.bom["desired_bom_sha256"], "releaseRoot": str(context.env_root), "publishedNginxConfigDigest": state.get("publishedNginxConfigDigest"), } marker_path = d3_root / "current.json" pending_marker = d3_root / f".current.{context.run_id}.json.pending" if pending_marker.is_symlink(): raise DeployError("Prod D3 pending current marker must not be a symlink") if pending_marker.exists(): pending_marker.unlink() with pending_marker.open("x", encoding="utf-8") as stream: os.chmod(pending_marker, 0o600) stream.write(json.dumps(marker, sort_keys=True) + "\n") stream.flush() os.fsync(stream.fileno()) # Publish the self-contained marker before its convenience symlink. If the # second replace fails, durable in-progress authority remains and an exact # retry can repeat this closeout from the marker-bound release root. os.replace(pending_marker, marker_path) pending_link = d3_root / f".current.{context.run_id}.pending" if pending_link.exists() or pending_link.is_symlink(): pending_link.unlink() pending_link.symlink_to(context.env_root) os.replace(pending_link, current) descriptor = os.open( d3_root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) ) try: os.fsync(descriptor) finally: os.close(descriptor) try: clear_in_progress( d3_root=d3_root, release_tag=str(context.envv.get("RELEASE_TAG", "")), tag_target_commit=str(context.envv.get("TAG_TARGET_COMMIT", "")), resolved_bom_reference=str( context.envv.get("D3_RESOLVED_BOM_REFERENCE", "") ), target_slot=prod_slot(context.envv), target_backend_port=route.backend_port, release_root=context.env_root, ) except PrepareError as exc: raise DeployError(str(exc)) from exc def verify_public_target(route: ProdRoute) -> dict[str, Any]: """Prove the switched public route before committing the current marker.""" public_url = f"https://{PROD_SERVER_NAME}" headers = impl.default_capture( ( "curl", "--fail", "--silent", "--show-error", "--dump-header", "-", "--output", "/dev/null", "--max-time", "20", f"{public_url}/health", ) ) if not re.search( rf"(?im)^x-nexuscrm-backend-port:\s*{route.backend_port}\s*$", headers.replace("\r", ""), ): raise DeployError( "public Prod health does not report the switched D3 backend port" ) impl.default_execute( ( "curl", "--fail", "--silent", "--show-error", "--output", "/dev/null", "--max-time", "20", f"{public_url}/index.html", ) ) return { "backendPort": route.backend_port, "healthUrl": f"{public_url}/health", "frontendUrl": f"{public_url}/index.html", } def require_prod_in_progress(context: impl.D3Context) -> dict[str, Any]: """Reconcile the runner with durable target authority before mutation.""" d3_root = context.env_root.parent.parent if d3_root.name != "d3" or context.env_root.parent.name != "releases": raise DeployError("Prod D3 release root is outside durable D3 authority") envv = context.envv try: pending = require_in_progress_authority( read_in_progress_authority(d3_root / "in-progress.json"), release_tag=str(envv.get("RELEASE_TAG", "")).strip(), tag_target_commit=str(envv.get("TAG_TARGET_COMMIT", "")).strip(), resolved_bom_reference=str( envv.get("D3_RESOLVED_BOM_REFERENCE", "") ).strip(), release_root=context.env_root, ) except PrepareError as exc: raise DeployError(str(exc)) from exc target_slot = prod_slot(envv) target_port = SLOT_CONTRACT[target_slot].sfcrm_port previous_port = impl.require_port( envv.get("PROD_D3_PREVIOUS_BACKEND_PORT", ""), "PROD_D3_PREVIOUS_BACKEND_PORT", ) previous_kind = str(envv.get("PROD_D3_PREVIOUS_ROUTE_KIND", "")).strip() if ( pending.get("targetSlot") != target_slot or pending.get("targetBackendPort") != target_port or pending.get("previousBackendPort") != previous_port or pending.get("previousRouteKind") != previous_kind ): raise DeployError("Prod D3 runtime env disagrees with durable target authority") route = live_route() if route.backend_port not in {previous_port, target_port}: raise DeployError("live Prod route escaped durable target authority") if route.backend_port == previous_port and route.is_d3 != (previous_kind == "d3"): raise DeployError("live Prod predecessor kind disagrees with durable authority") if route.backend_port == target_port and not route.is_d3: raise DeployError("live Prod target lacks D3 ingress ownership") return pending def previous_runtime_units(route: ProdRoute) -> tuple[str, ...]: """Exact slot units, plus the reviewed global aliases for Classic handoff.""" if route.is_d3: return ( f"nexuscrm-multi-prod-d3-{route.slot}-hubspot-options-adapter.service", f"nexuscrm-multi-prod-slot-{route.backend_port}-backend.service", ) return ( f"nexuscrm-multi-prod-slot-{route.backend_port}-hubspot-options-adapter.service", "nexuscrm-multi-prod-hubspot-options-adapter.service", f"nexuscrm-multi-prod-slot-{route.backend_port}-backend.service", "nexuscrm-multi-prod-backend.service", ) def previous_runtime_containers(route: ProdRoute) -> list[dict[str, Any]]: """Read only identity, state and restart policy; never capture container env.""" if not route.is_d3: return [] owner = container_label_value(route.slot) identifiers = impl.default_capture(( "docker", "ps", "-a", "--no-trunc", "--filter", f"label={impl.CONTAINER_LABEL_KEY}={owner}", "--format", "{{.ID}}", )).splitlines() projection = ( '{"id":{{json .Id}},"name":{{json .Name}},' '"owner":{{json (index .Config.Labels "' + impl.CONTAINER_LABEL_KEY + '")}},' '"running":{{json .State.Running}},"restarting":{{json .State.Restarting}},' '"restartPolicy":{{json .HostConfig.RestartPolicy.Name}}}' ) result = [] for identifier in sorted(set(identifiers)): if not re.fullmatch(r"[0-9a-f]{64}", identifier): raise DeployError("previous Prod container inventory has an invalid ID") try: item = json.loads(impl.default_capture(( "docker", "inspect", "--format", projection, identifier, ))) except json.JSONDecodeError as exc: raise DeployError("previous Prod container inspection is malformed") from exc if ( not isinstance(item, dict) or item.get("id") != identifier or item.get("owner") != owner or not re.fullmatch( rf"/nexuscrm-d3-prod-{re.escape(route.slot)}-[a-z0-9][a-z0-9_.-]*", str(item.get("name", "")), ) or type(item.get("running")) is not bool or type(item.get("restarting")) is not bool or item.get("restartPolicy") not in {"no", "always", "unless-stopped", "on-failure"} ): raise DeployError("previous Prod container identity/state is not slot-owned") result.append(item) return result def retire_previous_runtime( context: impl.D3Context, pending: Mapping[str, Any], target: ProdRoute, ) -> dict[str, Any]: """Stop only the durable predecessor after successful public target readback. Repeat on an exact-Tag retry, including when an earlier stop was partial. Keep stopped containers and release files for diagnostics/retention; never kill an unknown listener or derive the predecessor from the switched route. """ if require_prod_in_progress(context) != pending: raise DeployError("Prod predecessor authority changed before retirement") if require_live_target(context.envv) != target: raise DeployError("Prod target route changed before predecessor retirement") previous_port = int(pending["previousBackendPort"]) previous_kind = str(pending["previousRouteKind"]) if previous_port == target.backend_port or previous_kind not in {"d3", "legacy"}: raise DeployError("Prod predecessor retirement has an invalid slot binding") previous = ProdRoute( backend_port=previous_port, slot=slot_for_port(previous_port), is_d3=previous_kind == "d3", ) enabled_states = {"enabled", "enabled-runtime", "linked", "linked-runtime"} stopped_states = {"inactive", "failed", "dead"} units = previous_runtime_units(previous) try: # Resolve the entire exact inventory before the first stop. Docker # mutations use immutable IDs, never a host-wide or name-only target. containers = previous_runtime_containers(previous) unit_states = {unit: impl.staging.systemd_unit_state(unit) for unit in units} for unit, state in unit_states.items(): if state["load_state"] == "not-found": continue if state["unit_file_state"] in {"", "unknown"}: raise DeployError(f"cannot prove previous Prod unit autostart state: {unit}") for unit, state in unit_states.items(): if state["load_state"] == "not-found": continue if state["unit_file_state"] in enabled_states: impl.default_execute(("sudo", "-n", "systemctl", "disable", unit)) if state["active_state"] not in stopped_states: impl.default_execute(("sudo", "-n", "systemctl", "stop", unit)) for item in containers: if item["restartPolicy"] != "no": impl.default_execute(("docker", "update", "--restart=no", item["id"])) if item["running"] or item["restarting"]: impl.default_execute(("docker", "stop", "--time", "30", item["id"])) # Always re-read, even if a previous process already recorded success. for item in previous_runtime_containers(previous): if item["running"] or item["restarting"] or item["restartPolicy"] != "no": raise DeployError("previous Prod D3 container remains active or restartable") for unit in units: state = impl.staging.systemd_unit_state(unit) if state["load_state"] == "not-found": continue if state["active_state"] not in stopped_states: raise DeployError(f"previous Prod runtime unit remains active: {unit}") if state["unit_file_state"] in enabled_states | {"", "unknown"}: raise DeployError(f"previous Prod runtime unit remains enabled or unknown: {unit}") contract = SLOT_CONTRACT[previous.slot] ports = [contract.sfcrm_port, contract.adapter_port] if previous.is_d3: ports += [contract.integration_port, contract.runtime_grpc, contract.runtime_admin] ports += [ contract.worker_admin_base + index for index in range(len(impl.SUPPORTED_WORKER_STEMS)) ] for port in ports: impl.staging.require_staging_target_port_listener_free( str(port), "previous Prod runtime", ) except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc: raise DeployError("failed to stop or verify the previous Prod runtime") from exc return { "schema": "nexuscrm.prod-predecessor-retirement/v1", "phase": "post-cutover-retirement", "kind": previous_kind, "backendPort": previous_port, "slot": previous.slot, "targetBackendPort": target.backend_port, "containerIds": [item["id"] for item in containers], "retiredUnits": list(units), "verifiedListenerPorts": ports, "verified": True, } def preflight_companion_authority( context: impl.D3Context, *args: Any, **kwargs: Any ) -> None: """Admit the signed OAuth authority before any target generation mutation.""" if args: raise DeployError("Prod D3 convergence roots must be passed by name") try: l2_root = Path(kwargs["l2_root"]) except (KeyError, TypeError) as exc: raise DeployError("Prod D3 convergence lost its exact L2 source root") from exc metadata = context.bom.get("l2_metadata") l2_commit = metadata.get("commit", "") if isinstance(metadata, Mapping) else "" # Artifact preparation has not yet persisted exactSources. Admit the # supplied checkout against the signed BOM now; activation resolves again # from the persisted exactSources before materializing the companion env. impl.signed_l2_programmatic_oauth_authority( context, {"exactSources": {"l2Root": str(l2_root), "l2Commit": l2_commit}}, label="Prod HubSpot companion preflight", ) def prepare_target_for_convergence( context: impl.D3Context, *args: Any, **kwargs: Any ) -> dict[str, Any]: """Complete target-local preparation while the predecessor remains live.""" if args: raise DeployError("Prod D3 convergence roots must be passed by name") try: app_packs_root = Path(kwargs["app_packs_root"]) core_root = Path(kwargs["core_root"]) l2_root = Path(kwargs["l2_root"]) except (KeyError, TypeError) as exc: raise DeployError("Prod D3 convergence lost its exact source roots") from exc if not (context.state_file.exists() or context.state_file.is_symlink()): impl.capture(context) pre_mutation = { "captured", "catalog-successor-captured", "runtime-successor-captured", "artifacts-ready", "baseline-reset", } for _ in range(8): state = impl.require_state(context) phase = str(state.get("phase", "")) if phase not in pre_mutation: return state if phase in { "captured", "catalog-successor-captured", "runtime-successor-captured", }: impl.prepare_artifacts( context, app_packs_root=app_packs_root, core_root=core_root, l2_root=l2_root, ) elif phase == "artifacts-ready": impl.preserve_current_test2_database(context) else: # Quiesce only the selected target slot here. The currently routed # predecessor remains live throughout convergence and cutover. quiesce_target_runtime( context.envv, context.env_root / "run", disable_autostart=True, ) return state raise DeployError("Prod D3 target preparation exceeded its phase bound") def converge_target_with_bounded_resume( context: impl.D3Context, *args: Any, **kwargs: Any ) -> tuple[dict[str, Any], int]: """Use the engine's exact-run resumability once before requiring intervention.""" for attempt in range(1, 3): try: return _ORIG_CONVERGE(context, *args, **kwargs), attempt except DeployError as exc: if attempt == 2: raise DeployError( "Prod D3 target did not converge after the bounded exact-Tag resume" ) from exc raise AssertionError("bounded Prod D3 convergence loop did not return") def converge_prod( context: impl.D3Context, *args: Any, **kwargs: Any ) -> dict[str, Any]: global LAST_PATCH_REPLAY pending = require_prod_in_progress(context) target_slot = require_target_is_standby( context.envv, env_root=context.env_root, run_id=context.run_id ) if target_slot != pending["targetSlot"]: raise DeployError("live Prod target disagrees with durable target authority") preflight_companion_authority(context, *args, **kwargs) if REQUESTED_PATCH_DIR: d3_patch.require_patch_bundle_environment( REQUESTED_PATCH_DIR, environment="prod", repo_root=Path(kwargs["l2_root"]), ) completed_replay = d3_patch.require_active_run_can_replay_patch( context, REQUESTED_PATCH_DIR, read_state=impl.read_state ) d3_patch.preflight_and_seal_patch( context, REQUESTED_PATCH_DIR, source_root=Path(kwargs["l2_root"]), expected_tree_digest=EXPECTED_PATCH_TREE_DIGEST, environment="prod", ) if completed_replay is not None: LAST_PATCH_REPLAY = completed_replay archive_target_generation(context) prepare_target_for_convergence(context, *args, **kwargs) result, convergence_attempts = converge_target_with_bounded_resume( context, *args, **kwargs ) target_route = require_live_target(context.envv) public_readback = verify_public_target(target_route) retirement = retire_previous_runtime(context, pending, target_route) # Retirement cannot certify the target. Re-prove public traffic before # closing the release, and retain pending authority if that readback fails. target_route = require_live_target(context.envv) public_readback = verify_public_target(target_route) if "mcp_worker" in result.get("images", {}): from verify_hosted_mcp import probe mcp_readback = probe(context.envv["PUBLIC_BASE_URL"]) if not mcp_readback["entryPassed"]: raise DeployError("public MCP entry failed after predecessor retirement") result["mcpEntry"] = mcp_readback final = { **result, "convergenceAttempts": convergence_attempts, "prodPublicTargetReadback": public_readback, "previousRuntimeRetirement": retirement, } if LAST_PATCH_REPLAY is not None: final["patchReplay"] = LAST_PATCH_REPLAY impl.replace_state( context, expected_phase="serving", updates={ "deploymentEvidence": final, "convergenceAttempts": convergence_attempts, "prodPublicTargetReadback": public_readback, "previousRuntimeRetirement": retirement, }, ) publish_current(context, target_route) return final def _install() -> None: impl.load_env_file = load_prod_env_file impl.load_context = load_context impl.load_active_transition_context = load_active_transition_context impl.derive_ports = derive_ports impl.container_name = container_name impl.render_d3_nginx_server = render_d3_nginx_server impl.replace_nginx_server_block = replace_prod_nginx_server_block impl.capture_previous_nginx = capture_previous_nginx impl.ensure_d3_public_ingress = ensure_prod_ingress impl.verify_serving = verify_prod_serving impl.preserve_current_test2_database = preserve_prod_database impl.apply_test2_inspection_catalog_fixtures = skip_prod_inspection_fixtures impl.ensure_test2_inspection_identity = skip_prod_inspection_identity impl.apply_governed_reference_data = skip_prod_governed_reference_data impl.apply_stopped_upgrade_runtime = apply_stopped_upgrade_runtime impl.apply_catalog = apply_catalog impl.catalog_input_idempotency_key = catalog_input_idempotency_key impl.converge_deployment = converge_prod impl.staging.staging_target_systemd_services = target_systemd_services impl.staging.quiesce_staging_target_runtime = quiesce_target_runtime impl.reestablish_prod_clone_serving_quiescence = ( reestablish_prod_serving_quiescence ) impl.test_deploy.require_test_clone_scrub_target_identity = lambda *_args, **_kwargs: None impl.test_deploy.scrub_test_cloned_provider_credential_secrets = lambda *_args, **_kwargs: None impl.TEST2_NGINX_CONF = PROD_NGINX_CONF def main() -> int: global REQUESTED_PATCH_DIR, EXPECTED_PATCH_TREE_DIGEST, LAST_PATCH_REPLAY command = sys.argv[1] if len(sys.argv) > 1 else "" if command == "deploy" and "--preserve-current-database" not in sys.argv: print( "Prod D3 deploy requires --preserve-current-database and never resets Prod", file=sys.stderr, ) return 1 REQUESTED_PATCH_DIR = os.environ.get("D3_PATCH_DIR", "").strip() EXPECTED_PATCH_TREE_DIGEST = os.environ.get( "D3_PATCH_TREE_DIGEST", "" ).strip() LAST_PATCH_REPLAY = None if bool(REQUESTED_PATCH_DIR) != bool(EXPECTED_PATCH_TREE_DIGEST): print("Prod D3 patch path and digest must be supplied together", file=sys.stderr) return 1 _install() return impl.main() if __name__ == "__main__": raise SystemExit(main())