#!/usr/bin/env python3 """Single new-only Test deployment entrypoint over the shared D3 engine. Owns Test slot identity, signed-BOM deploy/reset composition, ports, containers, and D3-only /health mapping. Legacy Test2 workflows are retired. """ from __future__ import annotations import argparse import hashlib import json import os import re import shutil import subprocess import sys import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Mapping, Sequence _REPO_ROOT = Path(__file__).resolve().parent.parent if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from scripts import serviceification_d3_engine as impl from scripts import serviceification_d3_test_patch as d3_patch from scripts import validate_serviceification_bom as bom_contract from scripts.verify_serviceification_d1_metadata_authority import VerificationFailure, verify_metadata_authority DeployError = impl.DeployError nginx_server_block_bounds = impl.nginx_server_block_bounds TEST_NGINX_CONF = impl.TEST2_NGINX_CONF FROZEN_STATE_SCHEMA = impl.STATE_SCHEMA FROZEN_EVIDENCE_SCHEMA = impl.EVIDENCE_SCHEMA STATE_SCHEMA = "nexuscrm.serviceification-d3-test-state/v1" EVIDENCE_SCHEMA = "nexuscrm.serviceification-d3-test-deployment-evidence/v1" DATABASE_PRODUCTION_REFRESH_EVIDENCE_SCHEMA = ( "nexuscrm.test-database-production-refresh/v1" ) _ORIG_LOAD_ACTIVE = impl.load_active_transition_context _ORIG_FIXTURES = impl.apply_test2_inspection_catalog_fixtures _ORIG_IDENTITY = impl.ensure_test2_inspection_identity _ORIG_IDEMPOTENCY = impl.catalog_input_idempotency_key _ORIG_CONVERGE = impl.converge_deployment _ORIG_APPLY_STOPPED_RUNTIME = impl.apply_stopped_upgrade_runtime _ORIG_APPLY_CATALOG = impl.apply_catalog REQUESTED_PATCH_DIR = "" LAST_PATCH_REPLAY: dict[str, Any] | None = None LOCAL_UNSIGNED = False LOCAL_COMMANDS = frozenset({ "prepare-local", "preflight", "capture", "prepare-artifacts", "quiesce", "apply-stopped-upgrade-runtime", "prepare-catalog", "apply-catalog", "prepare-serving-inputs", "activate-worker-ownership", "start-serving", "deploy", "verify-serving", "status", "status-active", }) def consume_local_mode_argv(argv: list[str]) -> tuple[list[str], bool]: if argv.count("--local-unsigned") > 1: raise DeployError("--local-unsigned was supplied more than once") return [arg for arg in argv if arg != "--local-unsigned"], "--local-unsigned" in argv def require_local_test3_mode( envv: Mapping[str, str], command: str, *, preserve_current_database: bool = False, ) -> None: if not LOCAL_UNSIGNED or test_slot(envv) != "test3": raise DeployError("local unsigned deployment requires explicit --local-unsigned on Test3") if os.environ.get("GITHUB_ACTIONS", "").lower() == "true": raise DeployError("local unsigned deployment is host-only, not a GitHub Actions input") if command not in LOCAL_COMMANDS: raise DeployError(f"{command} is unavailable for local unsigned deployment; use full deploy") if command == "deploy" and not preserve_current_database: raise DeployError("local unsigned deploy requires --preserve-current-database") def prepare_local_deployment(args: argparse.Namespace) -> Path: envv = impl.load_env_file(args.env_file.expanduser().resolve()) require_local_test3_mode(envv, "prepare-local") desired, excluded = local_preparation_bom(args.desired_bom) if excluded: print("Host-only Test3 default selects the six local image roles; " "excluded formal-only services: " + ", ".join(excluded), file=sys.stderr) # Freeze the actual local selection in the candidate, without editing the # formal request or weakening explicit desired-BOM admission. with tempfile.TemporaryDirectory(prefix="d3-local-selection-") as temporary: selected = Path(temporary) / "desired-bom.json" selected.write_bytes(bom_contract.canonical_json_bytes(desired)) local_args = argparse.Namespace(**vars(args)) local_args.desired_bom = selected return impl.prepare_local_image_candidate(local_args) def local_preparation_bom(explicit_path: Path | None) -> tuple[dict[str, Any], list[str]]: source = explicit_path or _REPO_ROOT / "release/serviceification-d3-bom.json" desired = bom_contract.load_bom(source.expanduser().resolve()) errors = bom_contract.validate_bom(desired, require_desired=True) if errors: raise DeployError("invalid desired BOM: " + "; ".join(errors)) excluded = [] if "mcp_worker" in desired["services"]: if explicit_path is not None: raise DeployError("explicit desired BOM selects MCP; use signed deployment or " "supply an explicit six-role local BOM without MCP") del desired["services"]["mcp_worker"] desired["execution"]["topology"] = desired["execution"]["topology"].removesuffix(";mcp_worker>runtime_api") excluded.append("mcp_worker") return desired, excluded def parse_local_preparation_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Prepare a host-only, unsigned Test3 deployment") parser.add_argument("command", choices=["prepare-local"]) for name in ("env-file", "output-dir", "core-root", "app-packs-root", "l2-root", "deployment-tools-root", "approval-publication"): parser.add_argument("--" + name, type=Path, required=True) parser.add_argument("--desired-bom", type=Path, help="Exact six-role local request; default derives the six local roles from the formal BOM and reports excluded MCP") for role in ("runtime-api", "runtime-worker", "integration-service", "sfcrm-app-service", "catalog-reconcile", "frontend"): parser.add_argument("--" + role + "-image", required=True) return parser.parse_args(argv) def _cli_prepare_local() -> int: args = parse_local_preparation_args() result = prepare_local_deployment(args) impl.print_json({"releaseClass": "test-only-unsigned-non-promotable", "candidate": str(result)}) return 0 RETIRED_TEST2_COMMANDS = { "abandon-before-catalog", "recover-active-before-catalog", "recover-completed-reset-for-successor", "rollback-to-legacy", "resume-d3-after-rollback", } @dataclass(frozen=True) class SlotContract: backend_port: int integration_port: int runtime_grpc: int runtime_admin: int worker_admin_base: int server_name: str runtime_api_admission_limit: int SLOT_CONTRACT = { # Test2 and Test3 coexist on the same deployment host. Keep Test3 in its # migrated private block: the former backend-derived allocation overlapped # Test2 Runtime admin and Worker listeners. "test1": SlotContract( 3301, 3402, 3311, 3312, 3321, "nexuscrm-new-test1.pingcap.com", 16, ), "test2": SlotContract( 3302, 3403, 3312, 3313, 3322, "nexuscrm-new-test2.pingcap.com", 32, ), "test3": SlotContract( 3303, 3404, 13313, 13314, 13323, "nexuscrm-new-test3.pingcap.com", 16, ), } ACTIVE_SLOT = "" def test_slot(envv: Mapping[str, str]) -> str: """Return the Test slot implied by ENV_NAME, checked against D3_SLOT when set.""" env_name = str(envv.get("ENV_NAME", "")).strip() prefix = "test-" if not env_name.startswith(prefix): raise DeployError("D3 test runner requires ENV_NAME=test-") slot = env_name[len(prefix):] if slot not in SLOT_CONTRACT: raise DeployError(f"D3 test runner does not accept ENV_NAME={env_name}") declared = os.environ.get("D3_SLOT", "").strip() if declared and declared != 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 D3 test slot: {slot}") ACTIVE_SLOT = slot return slot def deployment_slot(envv: Mapping[str, str]) -> str: """D3 deploys Test1/Test3; Test2 remains a database-maintenance target.""" slot = test_slot(envv) if slot == "test2": raise DeployError("Test2 D3 deployment is retired; use standalone Test2 workflows") return slot def _apply_impl_identity(slot: str) -> None: contract = SLOT_CONTRACT[slot] impl.CONTAINER_LABEL_VALUE = slot impl.RUNTIME_SERVER_NAME = f"runtime-api.d3.{slot}.local" impl.RUNTIME_API_ADMISSION_LIMIT = contract.runtime_api_admission_limit if slot == "test2": impl.STATE_SCHEMA = FROZEN_STATE_SCHEMA impl.EVIDENCE_SCHEMA = FROZEN_EVIDENCE_SCHEMA else: impl.STATE_SCHEMA = STATE_SCHEMA impl.EVIDENCE_SCHEMA = EVIDENCE_SCHEMA def derive_ports( envv: Mapping[str, str], worker_stems: Sequence[str] | str = impl.LEGACY_WORKER_STEMS, slot: str | None = None, ) -> impl.D3Ports: """Reuse committed nginx ports; reject drifted runtime.env values.""" if isinstance(worker_stems, str): if slot is not None: raise DeployError("D3 Test 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("D3 Test port derivation received unsupported Worker roles") resolved = slot or ACTIVE_SLOT or test_slot(envv) bind_slot(resolved) contract = SLOT_CONTRACT[resolved] backend = impl.require_port(envv.get("BACKEND_PORT", ""), "BACKEND_PORT") if backend != contract.backend_port: raise DeployError( f"{resolved} BACKEND_PORT must be {contract.backend_port}, not {backend}" ) # The host workflow-options adapter is independent runtime authority. It # may share the sealed Integration port on an older slot or use a separate # port (Test3 currently does). Validate its shape, but never project it # into the D3 container topology. impl.require_port( impl.exp_hubspot_options_adapter_port(dict(envv)), "EXP_HUBSPOT_OPTIONS_ADAPTER_PORT", ) server_name = str(envv.get("SERVER_NAME", "")).split()[0] if server_name != contract.server_name: raise DeployError( f"{resolved} SERVER_NAME must be {contract.server_name}, not {server_name!r}" ) workers = { stem: contract.worker_admin_base + index for index, stem in enumerate(stems) } result = impl.D3Ports( contract.backend_port, contract.integration_port, contract.runtime_grpc, contract.runtime_admin, workers, ) values = result.all_ports() if len(set(values.values())) != len(values): duplicates = sorted( port for port in set(values.values()) if list(values.values()).count(port) > 1 ) raise DeployError( f"D3 {resolved} port allocation collides: {duplicates}" ) return result def container_name(run_id: str, role: str, *, slot: str = "") -> str: resolved = slot or ACTIVE_SLOT or os.environ.get("D3_SLOT", "").strip() if resolved not in SLOT_CONTRACT: raise DeployError("D3 container names require a Test slot") normalized = re.sub( r"[^a-z0-9_.-]+", "-", f"nexuscrm-d3-{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_context(env_file: Path, resolved_bom: Path, run_id: str) -> impl.D3Context: if not impl.RUN_ID_RE.fullmatch(run_id): raise DeployError("D3 run id must be one bounded canonical identifier") env_file = env_file.expanduser().resolve() envv = impl.load_env_file(env_file) slot = bind_slot(deployment_slot(envv)) _apply_impl_identity(slot) if not impl.bool_env(envv, "INSTALL_SYSTEMD", True): raise DeployError("D3 stopped upgrade requires the slot systemd deployment") env_root_text = envv.get("ENV_ROOT", "").strip() if not env_root_text: raise DeployError("test runtime env has no ENV_ROOT") env_root = Path(env_root_text).expanduser().resolve() expected_env_file = impl.runtime_paths(env_root)["runtime_env_file"].resolve() if env_file != expected_env_file: raise DeployError( f"D3 runtime env must be the active slot authority file: {expected_env_file}" ) resolved_bom = resolved_bom.expanduser().resolve() bom = impl.load_bom(resolved_bom) if bom.get("schema_version") == bom_contract.LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: require_local_test3_mode(envv, "preflight") errors = bom_contract.validate_local_test3_deployment(bom) if errors: raise DeployError("invalid local deployment: " + "; ".join(errors)) impl.require_exact_checkout(_REPO_ROOT, bom["controller_commit"], "local deployment controller") elif LOCAL_UNSIGNED: raise DeployError("--local-unsigned requires a local deployment manifest, not a signed BOM") elif bom.get("schema_version") != impl.RESOLVED_SCHEMA_VERSION_V10: raise DeployError( "D3 test deployment accepts only the new-architecture v10 L0 Engine BOM" ) serialized = resolved_bom.read_bytes() state_root = env_root / ".runtime" / "serviceification-d3" run_root = state_root / "runs" / run_id deployment_bom = impl.operational_bom(bom) worker_stems, worker_families = impl.worker_roles_from_bom(deployment_bom) 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(serialized), run_id=run_id, run_root=run_root, 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 = impl.load_env_file(env_file) slot = bind_slot(deployment_slot(envv)) _apply_impl_identity(slot) state_file = Path(envv["ENV_ROOT"]) / ".runtime/serviceification-d3/active.json" active = impl.read_state(state_file) if active.get("resolvedBomSchema") == bom_contract.LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: require_local_test3_mode(envv, "status-active") saved_load = impl.load_env_file saved_read = impl.read_state saved_derive = impl.derive_ports def aliased_load(path: Path) -> dict[str, str]: loaded = dict(saved_load(path)) loaded["ENV_NAME"] = "test-test2" loaded["EXP_HUBSPOT_OPTIONS_ADAPTER_PORT"] = str( SLOT_CONTRACT[slot].integration_port ) 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 = aliased_load impl.read_state = aliased_read impl.derive_ports = derive_ports try: context = _ORIG_LOAD_ACTIVE( env_file, run_id, allow_legacy_reset_receipt=allow_legacy_reset_receipt, ) finally: impl.load_env_file = saved_load impl.read_state = saved_read impl.derive_ports = saved_derive if context.bom.get("schema_version") == bom_contract.LOCAL_TEST3_DEPLOYMENT_SCHEMA_VERSION: require_local_test3_mode(envv, "status-active") 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 require_fixed_nginx_authority(content: str, slot: str) -> None: """Require the slot's committed ports; /health may already be /readyz.""" contract = SLOT_CONTRACT[slot] start, end = nginx_server_block_bounds(content, contract.server_name) block = content[start:end] backend = f"http://127.0.0.1:{contract.backend_port}" integration = f"http://127.0.0.1:{contract.integration_port}" if ( f"proxy_pass {backend};" not in block and f"proxy_pass {backend}/readyz;" not in block ): raise DeployError( f"live test nginx for {slot} does not proxy to {backend}" ) if f"proxy_pass {integration};" not in block: raise DeployError( f"live test nginx for {slot} does not proxy HubSpot/integration to {integration}" ) def map_slot_health_proxy(content: str, slot: str, *, readyz: bool) -> str: """Rewrite only this slot's /health proxy_pass; leave other slots untouched.""" contract = SLOT_CONTRACT[slot] start, end = nginx_server_block_bounds(content, contract.server_name) block = content[start:end] backend = f"http://127.0.0.1:{contract.backend_port}" wanted = f"{backend}/readyz" if readyz else backend match = re.search( r"(location /health \{\n\s+proxy_pass )" + re.escape(backend) + r"(?:/readyz)?;", block, ) if match is None: raise DeployError(f"live test nginx for {slot} /health does not proxy to {backend}") replacement = f"{match.group(1)}{wanted};" if match.group(0) == replacement: return content return content[:start] + block[:match.start()] + replacement + block[match.end():] + content[end:] def map_slot_mcp_proxy(content: str, context: impl.D3Context, *, enabled: bool) -> str: """Converge only this Test slot, including re-entry and MCP removal.""" slot = test_slot(context.envv) start, end = nginx_server_block_bounds(content, SLOT_CONTRACT[slot].server_name) block = content[start:end] if enabled: block = impl.route_mcp_ingress(block, context) else: # A preceding signed MCP deployment used this same slot-owned port. # Old configurations without the role need no MCP route changes. mcp_port = context.ports.runtime_admin + 1 for location in ("/mcp", "/.well-known/oauth-protected-resource"): pattern = r"(location " + re.escape(location) + r" \{\s*proxy_pass http://127\.0\.0\.1:)" + str(mcp_port) + ";" block = re.sub(pattern, lambda m: m[1] + str(context.ports.sfcrm_http) + ";", block) return content[:start] + block + content[end:] def _activate_slot_health(context: impl.D3Context, *, readyz: bool) -> None: slot = test_slot(context.envv) mapped = map_slot_health_proxy(TEST_NGINX_CONF.read_text(), slot, readyz=readyz) mapped = map_slot_mcp_proxy(mapped, context, enabled=readyz and "mcp_worker" in context.images) extra: dict[str, Any] = {} if readyz: extra = { "healthcheck_url": ( f"http://127.0.0.1:{context.envv.get('NGINX_LISTEN_PORT', '80')}/health" ), "healthcheck_headers": {"Host": SLOT_CONTRACT[slot].server_name}, } impl.activate_nginx_conf(TEST_NGINX_CONF, mapped, **extra) def ensure_d3_public_ingress(context: impl.D3Context) -> dict[str, Any]: """Keep committed nginx ports, then map this slot's /health to sfcrm /readyz.""" state = impl.require_state( context, expected_phase={"serving-start-in-progress", "serving"}, ) if TEST_NGINX_CONF.is_symlink() or not TEST_NGINX_CONF.is_file(): raise DeployError(f"live test nginx authority is unavailable: {TEST_NGINX_CONF}") require_fixed_nginx_authority(TEST_NGINX_CONF.read_text(), test_slot(context.envv)) _activate_slot_health(context, readyz=True) source = Path(str(state["prepared"]["frontendDist"])).resolve() frontend, frontend_digest = impl.publish_readable_frontend(context, source) return impl.replace_state( context, expected_phase=str(state["phase"]), updates={ "publishedFrontendTarget": str(frontend), "publishedFrontendDigest": frontend_digest, "publishedNginxConfigDigest": impl.sha256_bytes(TEST_NGINX_CONF.read_bytes()), }, ) def restore_previous_nginx(context: impl.D3Context, state: Mapping[str, Any]) -> None: """Restore this slot's /health to the monolith backend path; leave others.""" if TEST_NGINX_CONF.is_file() and not TEST_NGINX_CONF.is_symlink(): _activate_slot_health(context, readyz=False) def apply_test2_inspection_catalog_fixtures( envv: dict[str, str], l2_root: Path, **kwargs: Any, ) -> dict[str, Any]: """Converge the shared inspection Catalog authority for every Test slot.""" test_slot(envv) return _ORIG_FIXTURES(envv, l2_root, **kwargs) def ensure_test2_inspection_identity( context: impl.D3Context, core_root: Path, **kwargs: Any, ) -> dict[str, Any]: if test_slot(context.envv) != "test2": 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, "slot": test_slot(context.envv), }, }, ) return _ORIG_IDENTITY(context, core_root, **kwargs) def catalog_input_idempotency_key(*args: Any, **kwargs: Any) -> str: key = _ORIG_IDEMPOTENCY(*args, **kwargs) slot = ACTIVE_SLOT or os.environ.get("D3_SLOT", "").strip() if slot and slot != "test2" and key.startswith("d3-test2-"): return f"d3-{slot}-" + key.removeprefix("d3-test2-") return key def bind_requested_patch_dir(raw: str) -> str: global REQUESTED_PATCH_DIR, LAST_PATCH_REPLAY REQUESTED_PATCH_DIR = str(raw or "").strip() LAST_PATCH_REPLAY = None return REQUESTED_PATCH_DIR def consume_adapter_argv(argv: list[str]) -> tuple[list[str], str]: """Remove generic-only deploy inputs before invoking the engine CLI.""" if not argv: return [], "" remaining = [argv[0]] patch_dir = "" index = 1 while index < len(argv): arg = argv[index] if arg == "--patch-dir": if index + 1 >= len(argv): raise DeployError("--patch-dir requires a directory") if patch_dir: raise DeployError("--patch-dir was supplied more than once") patch_dir = argv[index + 1] index += 2 continue if arg.startswith("--patch-dir="): if patch_dir: raise DeployError("--patch-dir was supplied more than once") patch_dir = arg.split("=", 1)[1] index += 1 continue remaining.append(arg) index += 1 return remaining, patch_dir resolve_operator_patch_dir = d3_patch.resolve_operator_patch_dir seal_patch_bundle = d3_patch.seal_patch_bundle apply_sealed_patch_bundle = d3_patch.apply_sealed_patch_bundle preflight_and_seal_patch = d3_patch.preflight_and_seal_patch patch_seal_dir = d3_patch.patch_seal_dir def replay_patch_before_catalog(context: impl.D3Context) -> dict[str, Any] | None: return d3_patch.replay_patch_before_catalog( context, requested_patch_dir=REQUESTED_PATCH_DIR, environment=test_slot(context.envv), apply_bundle=apply_sealed_patch_bundle, read_state=read_production_refresh_active_state, ) def _record_patch_replay(context: impl.D3Context, *, required: bool) -> None: global LAST_PATCH_REPLAY if LAST_PATCH_REPLAY is not None: return replay = replay_patch_before_catalog(context) if required and replay is None: raise DeployError("requested --patch-dir was not replayed before Catalog apply") if replay is not None: LAST_PATCH_REPLAY = replay state = read_production_refresh_active_state(context.state_file) if state.get("patchReplay") != replay: impl.replace_state( context, expected_phase=str(state.get("phase", "")), updates={"patchReplay": replay}, ) def apply_stopped_upgrade_runtime( context: impl.D3Context, *args: Any, **kwargs: Any ) -> dict[str, Any]: """Replay a sealed patch after frozen quiescence, before Catalog 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: _record_patch_replay(context, required=bool(REQUESTED_PATCH_DIR.strip())) 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]: """Replay a sealed patch at catalog-ready only if runtime apply did not.""" _record_patch_replay(context, required=False) return _ORIG_APPLY_CATALOG(context, *args, **kwargs) def require_no_uncertain_database_production_refresh( env_root: Path, *, subject: str ) -> None: refresh_root = ( env_root / ".runtime" / "serviceification-d3" / "database-production-refreshes" ) # One-time cutover safety: old successful receipts are not a read path and # the old command is gone, but an unfinished provider operation from the # pre-rename program must remain a hard stop until an operator resolves it. retired_reset_root = ( env_root / ".runtime" / "serviceification-d3" / "database-resets" ) uncertain = sorted( [ *refresh_root.glob("*.json.pending"), *retired_reset_root.glob("*.json.pending"), ] ) if uncertain: raise DeployError( f"{subject} found an uncertain production database refresh receipt: " f"{uncertain[0]}" ) def resolve_production_refresh_operation( refresh_root: Path, requested_operation_id: str, ) -> tuple[str, Path, Path]: """Resume the sole pending refresh before accepting a new operation. GitHub assigns a new run ID to a retry dispatched after the owning workflow has changed. The provider-side reset may already have committed, so a new operation must finish the durable pending receipt instead of issuing a second reset or leaving deploy permanently blocked. """ requested_receipt = refresh_root / f"{requested_operation_id}.json" requested_pending = refresh_root / f"{requested_operation_id}.json.pending" retired_reset_root = refresh_root.parent / "database-resets" legacy_pending = sorted(retired_reset_root.glob("*.json.pending")) if legacy_pending: raise DeployError( "production database refresh found an uncertain legacy reset receipt: " f"{legacy_pending[0]}" ) pending = sorted(refresh_root.glob("*.json.pending")) if len(pending) > 1: raise DeployError( "production database refresh found multiple pending operation receipts: " + ", ".join(str(path) for path in pending) ) if not pending or pending[0] == requested_pending: return requested_operation_id, requested_receipt, requested_pending pending_receipt = pending[0] suffix = ".json.pending" operation_id = pending_receipt.name.removesuffix(suffix) if ( operation_id == pending_receipt.name or not impl.RUN_ID_RE.fullmatch(operation_id) ): raise DeployError( "pending production database refresh receipt has a malformed operation id" ) receipt_path = refresh_root / f"{operation_id}.json" if receipt_path.exists() or receipt_path.is_symlink(): raise DeployError( "production database refresh found both completed and pending receipts " f"for operation {operation_id}" ) return operation_id, receipt_path, pending_receipt def read_production_refresh_active_state(path: Path) -> dict[str, Any]: """Read leftover D3 state that a production refresh will archive. Historical Test transitions may still use the original Test2 state schema. Refresh only archives that file; it must not reinterpret it as current state. """ try: value = json.loads(path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise DeployError(f"cannot read D3 state {path}: {exc}") from exc if not isinstance(value, dict) or value.get("schema") not in { STATE_SCHEMA, FROZEN_STATE_SCHEMA, }: raise DeployError(f"D3 state has an unknown schema: {path}") return value def _converge_deployment(context: impl.D3Context, *args: Any, **kwargs: Any) -> dict[str, Any]: global LAST_PATCH_REPLAY # Deployment is an idempotent upgrade over the slot's current database. # A production-clone refresh is a separate # refresh-database-from-production operation with # its own receipt; deployment never invokes or retries that destructive act. require_no_uncertain_database_production_refresh( context.env_root, subject=f"D3 {test_slot(context.envv)} deployment", ) state_file = getattr( context, "state_file", context.env_root / ".runtime" / "serviceification-d3" / "active.json", ) if state_file.is_symlink(): raise DeployError("D3 deployment refuses a symlinked active state") if state_file.exists(): active = read_production_refresh_active_state(state_file) if isinstance(active.get("productionCloneResetRequest"), dict): raise DeployError( "ordinary deploy cannot resume a legacy production-clone reset request; " "use the independent refresh-database-from-production operation" ) completed_replay = d3_patch.require_active_run_can_replay_patch( context, REQUESTED_PATCH_DIR, read_state=read_production_refresh_active_state, ) # Resolve the deployment-owned single-org identity before state capture, # quiescence, or Catalog mutation. Missing configuration cannot leave # partial state. impl.installation_org_id() preflight_and_seal_patch( context, REQUESTED_PATCH_DIR, environment=test_slot(context.envv), ) if completed_replay is not None: LAST_PATCH_REPLAY = completed_replay # The frozen engine still exposes a historical optional choice here. The # new adapter owns a single non-destructive deploy contract, so callers # cannot select the old implicit production-clone path by omitting a flag. kwargs["preserve_current_database"] = True result = _ORIG_CONVERGE(context, *args, **kwargs) if REQUESTED_PATCH_DIR.strip() and LAST_PATCH_REPLAY is None: raise DeployError( "requested --patch-dir was not replayed before D3 reported success" ) additions: dict[str, Any] = {} if LAST_PATCH_REPLAY is not None: additions["patchReplay"] = LAST_PATCH_REPLAY return {**result, **additions} def preserve_current_test_database(context: impl.D3Context) -> dict[str, Any]: """Retain the slot database, but consume a production-refresh hygiene marker. Frozen preserve_current_test2_database records retained-test2-database and skips the clone-only connector-queue purge. After refresh-database-from-production, that would let inherited production connector events execute. When the provider left TEST_RESET_SNAPSHOT_CONNECTOR_QUEUE_HYGIENE_PENDING=1, this wrapper purges those queues against the later deploy's exact L2 pack set, then records the same prod-tidb-reset-snapshot receipt the frozen in-deploy reset uses. """ state = impl.require_state(context, expected_phase="artifacts-ready") slot = test_slot(context.envv) impl.test_deploy.require_test_clone_scrub_target_identity( dict(context.envv), subject=f"D3 retained {slot} database", ) if impl.owned_containers(): raise DeployError("current-database adoption refuses active D3 containers") exact_sources = state.get("exactSources") if not isinstance(exact_sources, dict): raise DeployError("current-database adoption has no exact L2 source") l2_root = Path(str(exact_sources.get("l2Root", ""))).resolve() if not l2_root.is_dir(): raise DeployError("current-database adoption lost the exact L2 source") pack_names = impl.runtime_pack_names(l2_root) if not pack_names: raise DeployError("current-database adoption resolved no Runtime packs") independent_reset_pending = impl.bool_env( context.envv, impl.test_deploy.RESET_SNAPSHOT_CONNECTOR_QUEUE_HYGIENE_PENDING, ) reset_sanitation: dict[str, object] = {} if independent_reset_pending: purged = impl.purge_reset_snapshot_connector_queue(context.envv) impl.reset_drained_workflow_action_syncstate_for_packs( context.envv, pack_names, ) impl.update_env_file( context.env_file, {impl.test_deploy.RESET_SNAPSHOT_CONNECTOR_QUEUE_HYGIENE_PENDING: "0"}, ) reset_sanitation = impl.reset_snapshot_connector_queue_purge_summary(purged) baseline: dict[str, Any] = { "source": ( "prod-tidb-reset-snapshot" if independent_reset_pending else "retained-test2-database" ), "targetSlot": slot, "resetPerformed": independent_reset_pending, "runtimePacks": sorted(pack_names), **reset_sanitation, } 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 slot_owned_containers( slot: str, *, inspect: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, ) -> list[str]: """List current D3-owned containers, excluding handed-over process names.""" if slot not in SLOT_CONTRACT: raise DeployError( f"production database refresh rejects unknown test slot: {slot!r}" ) result = inspect( [ "docker", "ps", "-a", "--filter", f"label={impl.CONTAINER_LABEL_KEY}={slot}", "--format", "{{.Names}}", ], check=True, capture_output=True, text=True, ) names = sorted(name for name in result.stdout.splitlines() if name) if len(names) != len(set(names)) or any( not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", name) for name in names ): raise DeployError( "production database refresh found a malformed D3 container identity" ) # The retained Integration container still has its immutable D3 creator # label after handover. Its canonical standalone name now owns lifecycle. return [name for name in names if not re.match(rf"^nexuscrm-l[01]-{slot}-", name)] def remove_slot_container( name: str, *, execute: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run, ) -> None: if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", name): raise DeployError( "production database refresh refuses a malformed container identity" ) execute(["docker", "rm", "--force", name], check=True) def require_standalone_processes_stopped( slot: str, *, inspect: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, ) -> None: """Database maintenance must not take over independent process owners.""" if slot not in SLOT_CONTRACT: raise DeployError(f"unknown Test slot: {slot!r}") result = inspect( ["docker", "ps", "--filter", f"name=^/nexuscrm-l[01]-{slot}-", "--format", "{{.Names}}"], check=True, capture_output=True, text=True, timeout=30, ) if result.stdout.strip(): raise DeployError( f"production database refresh requires stopped standalone {slot} processes; " "stop L0, L1 and retained Integration through their own controllers first: " + ", ".join(result.stdout.splitlines()) ) def refresh_test_database_from_production( env_file: Path, operation_id: str, *, list_containers: Callable[[str], list[str]] = slot_owned_containers, remove_container: Callable[[str], None] = remove_slot_container, reconcile_connection: Callable[[Path], Mapping[str, str]] | None = None, ) -> dict[str, Any]: """Refresh one Test database from production without deploying a BOM. The operation stops all slot writers, refreshes and sanitizes the clone, records the provider refresh receipt, and archives any prior D3 transition. It deliberately leaves the slot stopped. """ if not impl.RUN_ID_RE.fullmatch(operation_id): raise DeployError("production database refresh operation id is not canonical") env_file = env_file.expanduser().resolve() envv = impl.load_env_file(env_file) slot = bind_slot(test_slot(envv)) _apply_impl_identity(slot) env_root_text = str(envv.get("ENV_ROOT", "")).strip() if not env_root_text: raise DeployError( f"production database refresh requires the {slot} environment root" ) if not impl.bool_env(envv, "INSTALL_SYSTEMD", True): raise DeployError( f"production database refresh requires the {slot} systemd authority" ) env_root = Path(env_root_text).expanduser().resolve() expected_env_file = impl.runtime_paths(env_root)["runtime_env_file"].resolve() if env_file != expected_env_file: raise DeployError( f"production database refresh requires the canonical {slot} runtime env: " f"{expected_env_file}" ) impl.test_deploy.require_test_clone_scrub_target_identity( dict(envv), subject=f"{slot} production database refresh", ) impl.test_deploy.require_env( dict(envv), [ "TIDB_CLOUD_PUBLIC_KEY", "TIDB_CLOUD_PRIVATE_KEY", "TIDB_CLOUD_CLUSTER_ID", "TIDB_CLOUD_BRANCH_ID", ], ) impl.require_commands("docker", "systemctl", "curl") require_standalone_processes_stopped(slot) state_root = env_root / ".runtime" / "serviceification-d3" refresh_root = state_root / "database-production-refreshes" if refresh_root.is_symlink(): raise DeployError( "production database refresh refuses a symlinked receipt directory" ) refresh_root.mkdir(parents=True, mode=0o700, exist_ok=True) if refresh_root.is_symlink() or not refresh_root.is_dir(): raise DeployError( "production database refresh requires a real receipt directory" ) operation_id, receipt_path, pending_receipt = resolve_production_refresh_operation( refresh_root, operation_id, ) if receipt_path.is_symlink() or pending_receipt.is_symlink(): raise DeployError( "production database refresh refuses a symlinked operation receipt" ) if receipt_path.exists(): try: completed = json.loads(receipt_path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise DeployError( "completed production database refresh receipt is unreadable" ) from exc if ( not isinstance(completed, dict) or completed.get("schema") != DATABASE_PRODUCTION_REFRESH_EVIDENCE_SCHEMA or completed.get("stage") != "database-refreshed-from-production" or completed.get("operationId") != operation_id or completed.get("targetSlot") != slot or completed.get("deploymentStarted") is not False or not isinstance(completed.get("providerReset"), dict) ): raise DeployError( "completed production database refresh receipt disagrees with this operation" ) return completed state_file = state_root / "active.json" pending_state = state_file.with_suffix(".json.pending") if pending_state.exists() or pending_state.is_symlink(): raise DeployError( "production database refresh found an uncertain D3 state publication: " f"{pending_state}" ) if pending_receipt.exists(): try: pending_payload = json.loads(pending_receipt.read_text()) except (OSError, json.JSONDecodeError) as exc: raise DeployError( "pending production database refresh receipt is unreadable" ) from exc if ( not isinstance(pending_payload, dict) or pending_payload.get("schema") != DATABASE_PRODUCTION_REFRESH_EVIDENCE_SCHEMA or pending_payload.get("stage") != "database-refresh-from-production-in-progress" or pending_payload.get("operationId") != operation_id or pending_payload.get("targetSlot") != slot or not isinstance(pending_payload.get("providerEpochBefore"), str) or not pending_payload["providerEpochBefore"] ): raise DeployError( "pending production database refresh receipt disagrees with this operation" ) else: require_no_uncertain_database_production_refresh( env_root, subject="production database refresh" ) active: dict[str, Any] | None = None active_bytes: bytes | None = None archived_state: Path | None = None if state_file.is_symlink(): raise DeployError( "production database refresh refuses a symlinked D3 active state" ) if state_file.exists(): active_bytes = state_file.read_bytes() active = read_production_refresh_active_state(state_file) if ( active.get("envName") != envv["ENV_NAME"] or active.get("envRoot") != str(env_root) ): raise DeployError( "production database refresh found a D3 state owned by another slot" ) active_run_id = str(active.get("runId", "")) if not impl.RUN_ID_RE.fullmatch(active_run_id): raise DeployError( "production database refresh found a malformed active D3 run id" ) archive_suffix = hashlib.sha256(operation_id.encode()).hexdigest()[:12] archived_state = state_root / ( f"database-production-refresh-{active_run_id}-{archive_suffix}.json" ) if archived_state.exists() or archived_state.is_symlink(): raise DeployError( "production database refresh archive already exists: " f"{archived_state}" ) containers = list_containers(slot) if len(containers) != len(set(containers)): raise DeployError( "production database refresh found duplicate D3 container identities" ) branch_id = envv["TIDB_CLOUD_BRANCH_ID"] settled_before = impl.wait_for_tidb_branch_active(dict(envv), branch_id) provider_epoch_before = impl.tidb_branch_parent_timestamp(settled_before) pending_payload = { "schema": DATABASE_PRODUCTION_REFRESH_EVIDENCE_SCHEMA, "stage": "database-refresh-from-production-in-progress", "operationId": operation_id, "targetSlot": slot, "activeRunId": str(active.get("runId", "")) if active else "", "activePhase": str(active.get("phase", "")) if active else "", "activeStateDigest": impl.sha256_bytes(active_bytes) if active_bytes else "", "archivedActiveState": str(archived_state) if archived_state else "", "providerEpochBefore": provider_epoch_before, "d3Containers": containers, } pending_receipt.write_bytes(impl.canonical_bytes(pending_payload)) os.chmod(pending_receipt, 0o600) active_digest = str(pending_payload.get("activeStateDigest", "")) active_run_id = str(pending_payload.get("activeRunId", "")) archived_text = str(pending_payload.get("archivedActiveState", "")) archived_state = Path(archived_text) if archived_text else None if active_digest: archive_suffix = hashlib.sha256(operation_id.encode()).hexdigest()[:12] expected_archive = state_root / ( f"database-production-refresh-{active_run_id}-{archive_suffix}.json" ) if ( not impl.SHA256_RE.fullmatch(active_digest) or not impl.RUN_ID_RE.fullmatch(active_run_id) or archived_state is None or archived_state != expected_archive ): raise DeployError( "production database refresh pending state boundary is malformed" ) if archived_state.is_symlink(): raise DeployError( "production database refresh refuses a symlinked state archive" ) if state_file.exists(): if state_file.is_symlink() or impl.sha256_bytes(state_file.read_bytes()) != active_digest: raise DeployError( "production database refresh active state changed after intent publication" ) elif not archived_state.is_file() or impl.sha256_bytes(archived_state.read_bytes()) != active_digest: raise DeployError( "production database refresh lost both active and archived state authority" ) elif active_run_id or archived_text: raise DeployError( "production database refresh pending receipt has an unbound state archive" ) # runtime.env is live authority. Publish rotatable connection material only # after every immutable target, credential, command, receipt, and active-state # guard above has passed. The reconciler commits the complete file atomically. reconcile = ( reconcile_connection if reconcile_connection is not None else impl.test_deploy.reconcile_test_runtime_connection_binding ) reconcile(env_file) envv = impl.load_env_file(env_file) if test_slot(envv) != slot: raise DeployError("production database refresh changed slot during reconciliation") impl.test_deploy.require_test_clone_scrub_target_identity( dict(envv), subject=f"{slot} reconciled production database refresh", ) containers = list_containers(slot) if len(containers) != len(set(containers)): raise DeployError( "production database refresh found duplicate D3 container identities" ) for container in containers: remove_container(container) remaining = list_containers(slot) if remaining: raise DeployError( "production database refresh could not stop all D3 containers: " + ", ".join(remaining) ) impl.staging.quiesce_staging_target_runtime( dict(envv), impl.runtime_paths(env_root)["run_dir"], disable_autostart=True, ) branch_id = envv["TIDB_CLOUD_BRANCH_ID"] settled = impl.wait_for_tidb_branch_active(dict(envv), branch_id) provider_epoch_before = str(pending_payload["providerEpochBefore"]) provider_epoch = impl.tidb_branch_parent_timestamp(settled) if provider_epoch == provider_epoch_before: provider_reset = impl.test_deploy.run_test_reset_branch_snapshot_stage( env_file, slot, ) reset_epoch = provider_reset.get("provider_reset_epoch") if ( not isinstance(reset_epoch, dict) or reset_epoch.get("previous") != provider_epoch_before or not isinstance(reset_epoch.get("value"), str) or reset_epoch["value"] == provider_epoch_before ): raise DeployError( "production database refresh provider receipt disagrees with sealed intent" ) else: # The provider epoch is the external commit record. Finish the # deterministic post-reset work without issuing another reset. fresh_env = impl.load_env_file(env_file) impl.test_deploy.require_test_clone_scrub_target_identity( fresh_env, subject=f"{slot} production database refresh recovery", ) impl.ensure_tidb_database(fresh_env) secret_scrub = impl.test_deploy.scrub_test_cloned_provider_credential_secrets( fresh_env, ) impl.update_env_file( env_file, { "RESET_TIDB_BRANCH": "0", impl.test_deploy.RESET_SNAPSHOT_CONNECTOR_QUEUE_HYGIENE_PENDING: "1", }, ) provider_reset = { "target_slot": slot, "provider_reset_epoch": { "field": "parentTimestamp", "previous": provider_epoch_before, "value": provider_epoch, }, "provider_credential_secret_scrub": secret_scrub, "connector_queue_hygiene_pending": True, "recovered_from_provider_epoch": True, "status": "passed", } fresh_env = impl.load_env_file(env_file) impl.test_deploy.require_test_clone_scrub_target_identity( fresh_env, subject=f"{slot} production database refresh sanitation", ) purged = impl.purge_reset_snapshot_connector_queue(fresh_env) syncstate_reset_count = impl.reset_snapshot_workflow_action_syncstate(fresh_env) impl.update_env_file( env_file, { impl.test_deploy.RESET_SNAPSHOT_CONNECTOR_QUEUE_HYGIENE_PENDING: "0" }, ) provider_reset = { **provider_reset, "connector_queue_hygiene_pending": False, } reset_sanitation = { **impl.reset_snapshot_connector_queue_purge_summary(purged), "workflowActionSyncStatesReset": syncstate_reset_count, "connectorQueueHygienePending": False, } if active_digest and state_file.exists(): if archived_state is None: raise DeployError( "production database refresh lost its D3 archive target" ) os.replace(state_file, archived_state) os.chmod(archived_state, 0o600) result = { **pending_payload, "stage": "database-refreshed-from-production", "providerReset": provider_reset, "resetSanitation": reset_sanitation, "archivedActiveState": str(archived_state) if archived_state else "", "deploymentStarted": False, } pending_receipt.write_bytes(impl.canonical_bytes(result)) os.chmod(pending_receipt, 0o600) os.replace(pending_receipt, receipt_path) return result def reestablish_prod_clone_serving_quiescence( context: impl.D3Context, state: Mapping[str, Any], *, execute: Callable[[Sequence[str]], None], capture_output: Callable[[Sequence[str]], str], ) -> None: """Stop this slot's leftover writers. Ignore other host environments.""" 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("D3 serving retry found a container outside the exact run inventory") 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)) impl.staging.quiesce_staging_target_runtime( dict(context.envv), impl.runtime_paths(context.env_root)["run_dir"], disable_autostart=True, ) units = subprocess.run( ( "systemctl", "list-units", "--type=service", "--state=running", "--no-legend", "--plain", "nexuscrm*.service", ), check=False, capture_output=True, text=True, ) if units.returncode != 0: raise DeployError( "cannot prove zero legacy slot systemd processes: " + (units.stderr.strip() or f"systemctl exit {units.returncode}") ) owned_units = set(impl.staging.staging_target_systemd_services(dict(context.envv))) active_units = sorted({ line.split()[0] for line in units.stdout.splitlines() if line.split() and line.split()[0] in owned_units }) if active_units: raise DeployError( "legacy slot systemd processes remain active after target quiescence: " + ", ".join(active_units) ) docker_processes = subprocess.run( ( "docker", "ps", "--filter", f"label={impl.CONTAINER_LABEL_KEY}={impl.CONTAINER_LABEL_VALUE}", "--format", "{{.Names}}", ), check=False, capture_output=True, text=True, ) if docker_processes.returncode != 0: raise DeployError( "cannot prove zero slot D3 Docker processes: " + (docker_processes.stderr.strip() or f"docker ps exit {docker_processes.returncode}") ) running = sorted(name for name in docker_processes.stdout.splitlines() if name) if running: raise DeployError( "legacy slot D3 Docker processes remain active after D3 tuple stop: " + ", ".join(running) ) def _cli_refresh_database_from_production() -> int: parser = argparse.ArgumentParser() parser.add_argument("command") parser.add_argument("--env-file", required=True) parser.add_argument("--run-id", required=True) parser.add_argument("--resolved-bom") args, extra = parser.parse_known_args() try: if args.resolved_bom or extra: raise DeployError( "production database refresh does not accept a BOM or deploy inputs" ) impl.print_json( refresh_test_database_from_production(Path(args.env_file), args.run_id), ) return 0 except (DeployError, OSError, subprocess.CalledProcessError) as exc: print( f"D3 test-slot refresh-database-from-production failed: {exc}", file=sys.stderr, ) return 1 def _install() -> None: 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.ensure_d3_public_ingress = ensure_d3_public_ingress impl.restore_previous_nginx = restore_previous_nginx impl.apply_test2_inspection_catalog_fixtures = apply_test2_inspection_catalog_fixtures impl.ensure_test2_inspection_identity = ensure_test2_inspection_identity impl.catalog_input_idempotency_key = catalog_input_idempotency_key impl.converge_deployment = _converge_deployment impl.apply_stopped_upgrade_runtime = apply_stopped_upgrade_runtime impl.apply_catalog = apply_catalog impl.preserve_current_test2_database = preserve_current_test_database impl.reestablish_prod_clone_serving_quiescence = ( reestablish_prod_clone_serving_quiescence ) def main() -> int: global LOCAL_UNSIGNED try: remaining, LOCAL_UNSIGNED = consume_local_mode_argv(sys.argv) sys.argv[:] = remaining remaining, patch_dir = consume_adapter_argv(sys.argv) sys.argv[:] = remaining bind_requested_patch_dir(patch_dir) command = sys.argv[1] if len(sys.argv) >= 2 else "" if command == "prepare-local": if patch_dir: raise DeployError("prepare-local does not accept a patch bundle") return _cli_prepare_local() if LOCAL_UNSIGNED: local_parser = argparse.ArgumentParser(add_help=False) local_parser.add_argument("--env-file", required=True, type=Path) local_parser.add_argument("--preserve-current-database", action="store_true") local_args, _ = local_parser.parse_known_args(sys.argv[2:]) require_local_test3_mode( impl.load_env_file(local_args.env_file), command, preserve_current_database=local_args.preserve_current_database, ) if command in RETIRED_TEST2_COMMANDS: raise DeployError( f"{command} was retired with the legacy Test2 deployment; " "use deploy or the independent refresh-database-from-production " "operation" ) if command == "refresh-database-from-production": if patch_dir.strip(): raise DeployError( "production database refresh does not accept deploy-only inputs" ) if patch_dir.strip() and command != "deploy": raise DeployError("--patch-dir is valid only for deploy") if command == "deploy": admission = argparse.ArgumentParser(add_help=False) admission.add_argument("--env-file", required=True, type=Path) admission.add_argument("--resolved-bom", required=True, type=Path) admission.add_argument("--run-id", required=True) requested, _ = admission.parse_known_args(sys.argv[2:]) env_file = requested.env_file.expanduser().resolve() # Reject a trust-mode or slot mismatch before even reconciling credentials. load_context(env_file, requested.resolved_bom, requested.run_id) impl.test_deploy.reconcile_test_runtime_connection_binding(env_file) if command == "refresh-database-from-production": _install() return _cli_refresh_database_from_production() except (DeployError, VerificationFailure, bom_contract.ValidationFailure, OSError, ValueError, TypeError, subprocess.CalledProcessError) as exc: print(f"D3 test-slot deploy failed: {exc}", file=sys.stderr) return 1 _install() return impl.main() if __name__ == "__main__": raise SystemExit(main())