#!/usr/bin/env python3 """Apply one Test1 AppRelease and prepare the canonical L0 serving state. This is a thin L2 adapter around the existing Controlplane pack lifecycle and Catalog publisher. It never starts, stops, replaces, or restarts an L0 process. The separate L0 management workflow owns activation. """ from __future__ import annotations import argparse import hashlib import json import os import re import secrets import subprocess import sys import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping, Sequence ROOT = Path(__file__).resolve().parents[1] RUNTIME_TOOLS = ROOT / "deployments" / "offline" / "runtime" for import_root in (ROOT / "scripts", RUNTIME_TOOLS): sys.path.insert(0, str(import_root)) from common import DeployError, load_env_file # noqa: E402 from standalone_apprelease_publisher import prepare_upgrade_publisher # noqa: E402 from standalone_apprelease_catalog import ( # noqa: E402 idempotency_key, operation_facts as catalog_operation_facts, read_catalog_start, ) from deploy_staging_env import ( # noqa: E402 require_staging_target_runtime_quiesced, run_catalog_offline_candidate_phase, ) from deploy_offline_env import apply_runtime_app_shell # noqa: E402 from deployment_runtime_state import runtime_paths # noqa: E402 from prepare_test1_l0_state import ( # noqa: E402 StatePreparationError, prepare as prepare_l0_state, ) from render_serviceification_test_stack import ( # noqa: E402 ValidationFailure, load_execution_bom, load_json, render_catalog_config, render_serving_inputs, render_worker_role_inputs, render_worker_activation_inputs, runtime_worker_release_identity, sha256_digest, validate_apprelease_compatibility, validate_catalog_output_for_deployment, ) from serviceification_d3_engine import ( # noqa: E402 CatalogAuthority, read_optional_catalog_authority, read_worker_ownership, ) from serviceification_deployment_tools import ( # noqa: E402 FILES as DEPLOYMENT_TOOL_FILES, ToolError, verify_tools, ) from lib.final_pack_contract import metadata_tree_sha256 # noqa: E402 from standalone_l0_workers import ( # noqa: E402 L0_WORKER_STEMS, RETIRED_L0_WORKER_STEMS, ) COMMIT_RE = re.compile(r"[0-9a-f]{40}") IMAGE_RE = re.compile( r"ghcr\.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@" r"sha256:[0-9a-f]{64}" ) DATABASE_KEYS = ( "TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_PASSWORD", "TIDB_DATABASE", "TIDB_TLS_DISABLE", ) L0_CONTAINER = "nexuscrm-l0-test1-runtime-api" class AppReleaseError(RuntimeError): pass @dataclass(frozen=True) class CatalogStart: kind: str org_id: str app_installation_id: str catalog_revision: int security_epoch: int schema_generation: int engine_generation: int refresh_active_release: bool runtime_audience: str @dataclass(frozen=True) class SelectedL0WorkerRelease: repository: str digest: str source_commit: str def run(args: Sequence[str]) -> None: subprocess.run(list(args), check=True) def capture(args: Sequence[str]) -> str: return subprocess.run( list(args), check=True, text=True, capture_output=True, ).stdout.strip() def canonical_bytes(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() def write_private(path: Path, raw: bytes) -> None: path.parent.mkdir(parents=True, mode=0o700, exist_ok=True) pending = path.with_name(f".{path.name}.pending") if pending.exists() or pending.is_symlink(): raise AppReleaseError(f"stale output exists: {pending}") pending.write_bytes(raw) os.chmod(pending, 0o400) os.replace(pending, path) def require_no_active_d3( envv: Mapping[str, str], *, process_inventory: str | None = None, ) -> None: env_root = Path(str(envv.get("ENV_ROOT", ""))).expanduser() if not str(envv.get("ENV_ROOT", "")).strip(): raise AppReleaseError("database environment has no ENV_ROOT") active = env_root.resolve() / ".runtime/serviceification-d3/active.json" if active.exists() or active.is_symlink(): raise AppReleaseError("Test1 is owned by an active full-D3 deployment") if process_inventory is None: process_inventory = capture(( "docker", "ps", "-aq", "--filter", "label=io.nexuscrm.serviceification-d3=test1", )) if process_inventory.strip(): raise AppReleaseError("Test1 has a full-D3 process inventory") def provision_token(path: Path) -> None: if path.exists(): if path.is_symlink() or not path.is_file(): raise AppReleaseError(f"invalid existing worker token: {path}") return write_private(path, (secrets.token_hex(32) + "\n").encode()) def selected_l0_worker_release( worker_stems: Sequence[str], container: str = L0_CONTAINER, ) -> SelectedL0WorkerRelease: runtime_image = capture(( "docker", "container", "inspect", "--format", "{{.Config.Image}}", container, )) runtime_source = capture(( "docker", "image", "inspect", "--format", '{{index .Config.Labels "org.opencontainers.image.revision"}}', runtime_image, )) worker_images = { capture(( "docker", "container", "inspect", "--format", "{{.Config.Image}}", f"nexuscrm-l0-test1-worker-{stem}", )) for stem in worker_stems } if len(worker_images) != 1: raise AppReleaseError("deployed L0 Worker image closure is inconsistent") worker_image = next(iter(worker_images)) prefix = "ghcr.io/pingcap-inc/nexuscrm-runtime-worker@" worker_source = capture(( "docker", "image", "inspect", "--format", '{{index .Config.Labels "org.opencontainers.image.revision"}}', worker_image, )) selected = SelectedL0WorkerRelease( repository=prefix.removesuffix("@"), digest=worker_image.removeprefix(prefix), source_commit=worker_source, ) if ( not re.fullmatch( r"ghcr\.io/pingcap-inc/nexuscrm-runtime-api@sha256:[0-9a-f]{64}", runtime_image, ) or selected.repository != "ghcr.io/pingcap-inc/nexuscrm-runtime-worker" or not re.fullmatch(r"sha256:[0-9a-f]{64}", selected.digest) or not COMMIT_RE.fullmatch(selected.source_commit) or selected.source_commit != runtime_source ): raise AppReleaseError("deployed L0 API and Worker identities are inconsistent") return selected def prepare_worker_state( bom: dict[str, Any], catalog_output: dict[str, Any], *, worker_release: dict[str, str], catalog_image: str, database_env: Path, envv: Mapping[str, str], state_root: Path, operation_id: str, ) -> None: ownership = read_worker_ownership(envv) requests: dict[str, dict[str, Any]] = {} outputs: dict[str, dict[str, Any]] = {} request_root = state_root / "config/workers/activation" output_root = state_root / "data/worker-activation" role_root = state_root / "config/workers" token_root = state_root / "secrets/workers" output_root.mkdir(parents=True, mode=0o700, exist_ok=True) for role in bom["runtime_worker_roles"]: family = role["work_family"] stem = family.removeprefix("runtime.").removesuffix(".v1").replace(".", "-") current = ownership.get(family) expected_epoch = 0 if current is None else current.ownership_epoch rendered = render_worker_activation_inputs( bom, deployment_id_prefix="test1-runtime", expected_ownership_epoch=expected_epoch, worker_release=worker_release, ) request = rendered[stem] # Bind the retry identity to the AppRelease operation and exact epoch. request["operationId"] = ( f"{operation_id}-worker-{stem}-e{expected_epoch + 1}" ) request_path = request_root / f"{stem}.json" output_path = output_root / f"{stem}.json" write_private(request_path, canonical_bytes(request)) if output_path.exists() or output_path.is_symlink(): output_path.unlink() run(( "docker", "run", "--rm", "--network", "host", "--user", f"{os.getuid()}:{os.getgid()}", "--env-file", str(database_env), "--mount", f"type=bind,src={request_path},dst=/config.json,readonly", "--mount", f"type=bind,src={output_root},dst=/output", catalog_image, "work-ownership", "-config", "/config.json", "-output", f"/output/{stem}.json", )) requests[stem] = request outputs[stem] = load_json(output_path, f"Worker ownership output {stem}") os.chmod(output_path, 0o400) provision_token(token_root / f"{stem}.token") roles = render_worker_role_inputs( bom, catalog_output, requests, outputs, org_id=str(catalog_output["orgId"]), workload_id=str(bom["catalog_consumers"]["handler_workload_id"]), catalog_store_root="/var/lib/nexuscrm/catalog", worker_release=worker_release, ) for stem, role in roles.items(): provision_token(token_root / f"{stem}.token") write_private(role_root / f"{stem}.json", canonical_bytes(role)) write_private( state_root / "config/catalog-controlplane-image", (catalog_image + "\n").encode(), ) def invalidate_worker_state(state_root: Path) -> None: """Make an interrupted AppRelease preparation impossible to activate.""" for stem in (*L0_WORKER_STEMS, *RETIRED_L0_WORKER_STEMS): path = state_root / f"config/workers/{stem}.json" if path.is_symlink() or path.is_file(): path.unlink() elif path.exists(): raise AppReleaseError(f"invalid Worker state path: {path}") for stem in RETIRED_L0_WORKER_STEMS: token = state_root / f"secrets/workers/{stem}.token" if token.is_symlink() or token.is_file(): token.unlink() elif token.exists(): raise AppReleaseError(f"invalid Worker token path: {token}") for name in ( "layout-authoring-catalog.json", "layout-authoring-approval-publication.json", ): path = state_root / f"config/{name}" if path.is_symlink() or path.is_file(): path.unlink() elif path.exists(): raise AppReleaseError(f"invalid Layout publication state path: {path}") def exact_checkout(root: Path, expected: str, label: str) -> Path: if root.is_symlink(): raise AppReleaseError(f"{label} checkout is incomplete") root = root.resolve() if not COMMIT_RE.fullmatch(expected) or not root.is_dir(): raise AppReleaseError(f"{label} checkout is incomplete") actual = capture(("git", "-C", str(root), "rev-parse", "HEAD")) if actual != expected: raise AppReleaseError(f"{label} checkout is {actual}, expected {expected}") return root def select_catalog_start( operation: str, authority: CatalogAuthority | None, *, release_digest: str, installation_org_id: str, ) -> CatalogStart: if operation == "install": if not installation_org_id: raise AppReleaseError("install requires D3_INSTALLATION_ORG_ID") if authority is None: return CatalogStart( "PREPARE", installation_org_id, "", 0, 0, 1, 1, False, "nexuscrm-runtime-api", ) if ( authority.org_id != installation_org_id or authority.active_release_digest != release_digest ): raise AppReleaseError("install found a different AppInstallation; use upgrade") # A prior install may have published Catalog before state preparation # failed. Re-enter through the normal same-release upgrade path. operation = "upgrade" if operation != "upgrade": raise AppReleaseError("operation must be install or upgrade") if authority is None: raise AppReleaseError("upgrade found no AppInstallation; use install") if authority.runtime_audience != "nexuscrm-runtime-api": raise AppReleaseError("existing AppInstallation has the wrong Runtime audience") return CatalogStart( "UPGRADE", authority.org_id, authority.app_installation_id, authority.catalog_revision, authority.security_epoch, authority.schema_generation, authority.engine_generation, authority.active_release_digest == release_digest, authority.runtime_audience, ) def deterministic_uuid7(seed: bytes) -> str: raw = bytearray(hashlib.sha256(seed).digest()[:16]) raw[6] = (raw[6] & 0x0F) | 0x70 raw[8] = (raw[8] & 0x3F) | 0x80 value = raw.hex() return f"{value[:8]}-{value[8:12]}-{value[12:16]}-{value[16:20]}-{value[20:]}" def require_published_authority( authority: CatalogAuthority | None, output: Mapping[str, Any], release_digest: str, start: CatalogStart, ) -> None: expected_output = { "orgId": start.org_id, "catalogRevision": start.catalog_revision + 1, "securityEpoch": start.security_epoch + 1, "schemaGeneration": start.schema_generation, "engineGeneration": start.engine_generation, } if start.kind == "UPGRADE": expected_output["appInstallationId"] = start.app_installation_id wrong_output = [ field for field, value in expected_output.items() if output.get(field) != value ] if wrong_output: raise AppReleaseError( "Catalog output does not advance the exact starting authority: " + ", ".join(wrong_output) ) if authority is None: raise AppReleaseError("Catalog publication created no active authority") expected = { "app_installation_id": output.get("appInstallationId"), "org_id": output.get("orgId"), "active_release_digest": release_digest, "catalog_revision": output.get("catalogRevision"), "security_epoch": output.get("securityEpoch"), "schema_generation": output.get("schemaGeneration"), "engine_generation": output.get("engineGeneration"), "runtime_audience": start.runtime_audience, } mismatches = [ field for field, value in expected.items() if getattr(authority, field) != value ] if mismatches: raise AppReleaseError( "Catalog database authority disagrees with its output: " + ", ".join(mismatches) ) def write_database_env(path: Path, envv: Mapping[str, str]) -> None: missing = [key for key in DATABASE_KEYS[:5] if not str(envv.get(key, ""))] if missing: raise AppReleaseError("database environment lacks " + ", ".join(missing)) values = {key: str(envv[key]) for key in DATABASE_KEYS if key in envv} if any("\n" in value or "\r" in value or "\x00" in value for value in values.values()): raise AppReleaseError("database environment contains an invalid value") write_private( path, ("\n".join(f"{key}={value}" for key, value in values.items()) + "\n").encode(), ) def copy_image_file(image: str, source: str, destination: Path) -> None: if not IMAGE_RE.fullmatch(image): raise AppReleaseError("Catalog Controlplane image must be an immutable digest") run(("docker", "pull", "--quiet", image)) container = "nexuscrm-test1-catalog-contract-" + hashlib.sha256( image.encode(), ).hexdigest()[:16] subprocess.run( ("docker", "rm", "--force", container), check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: run(("docker", "create", "--name", container, image, "/bin/true")) destination.parent.mkdir(parents=True, mode=0o700, exist_ok=True) run(("docker", "cp", f"{container}:{source}", str(destination))) finally: subprocess.run( ("docker", "rm", "--force", container), check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) if destination.is_symlink() or not destination.is_file(): raise AppReleaseError(f"Catalog image lacks {source}") def require_l0_stopped_or_bootstrap(container: str = L0_CONTAINER) -> None: inspected = subprocess.run( ( "docker", "container", "inspect", "--format", "{{.State.Running}}\t{{json .Config.Cmd}}", container, ), text=True, capture_output=True, check=False, ) if inspected.returncode != 0: detail = inspected.stderr.strip() if "No such object" in detail or "No such container" in detail: raise AppReleaseError( "standalone L0 is not deployed; deploy it in bootstrap mode first" ) raise AppReleaseError( f"cannot inspect standalone L0 process: {detail or 'docker inspect failed'}" ) running_raw, separator, command_raw = inspected.stdout.strip().partition("\t") if separator != "\t" or running_raw not in {"true", "false"}: raise AppReleaseError("standalone L0 process has an unknown state") try: command = json.loads(command_raw) except json.JSONDecodeError as exc: raise AppReleaseError("standalone L0 process has an unknown command") from exc if running_raw == "true" and command != ["bootstrap"]: raise AppReleaseError( "standalone L0 Runtime is serving; stop it through Manage Test1 L0 " "before changing the AppRelease" ) for stem in (*L0_WORKER_STEMS, *RETIRED_L0_WORKER_STEMS): worker = f"nexuscrm-l0-test1-worker-{stem}" inspected = subprocess.run( ("docker", "container", "inspect", "--format", "{{.State.Running}}", worker), text=True, capture_output=True, check=False, ) if inspected.returncode != 0: detail = inspected.stderr.strip() if "No such object" in detail or "No such container" in detail: continue raise AppReleaseError( f"cannot inspect standalone L0 Worker {stem}: " f"{detail or 'docker inspect failed'}" ) if inspected.stdout.strip() == "true": raise AppReleaseError( f"standalone L0 Worker {stem} is running; stop L0 through " "Manage Test1 L0 before changing the AppRelease" ) def apply_apprelease(args: argparse.Namespace) -> dict[str, Any]: bom = load_execution_bom(args.resolved_bom.resolve()) envv = load_env_file(args.env_file.resolve()) require_no_active_d3(envv) if os.environ.get("D3_INSTALLATION_ORG_ID"): envv["D3_INSTALLATION_ORG_ID"] = os.environ["D3_INSTALLATION_ORG_ID"] app = bom["app_installation"] metadata = bom["l2_metadata"] app_root = exact_checkout(args.app_packs_root, app["artifact"]["commit"], "AppRelease") l2_root = exact_checkout(args.l2_root, metadata["commit"], "L2") try: tools = verify_tools( root=args.deployment_tools.resolve(), resolved_bom=args.resolved_bom.resolve(), ) except ToolError as exc: raise AppReleaseError(f"Controlplane CLI bundle is invalid: {exc}") from exc core_root = exact_checkout(args.core_root, tools["coreRevision"], "Controlplane Core") pack_root = l2_root / metadata["path"] if metadata_tree_sha256(pack_root) != metadata["tree_sha256"]: raise AppReleaseError("L2 pack tree disagrees with the signed release") setup = args.deployment_tools.resolve() / DEPLOYMENT_TOOL_FILES["setup"] pack_metadata = args.deployment_tools.resolve() / DEPLOYMENT_TOOL_FILES["pack_runtime_metadata"] for path in (setup, pack_metadata): if path.is_symlink() or not path.is_file() or not os.access(path, os.X_OK): raise AppReleaseError(f"Controlplane CLI is unavailable: {path.name}") def read_start(): return read_catalog_start( envv, environment="test1", operation=args.operation, release_digest=app["app_release_digest"], l2_tree_digest=metadata["tree_sha256"], ) before = read_start() start = select_catalog_start( args.operation, before, release_digest=app["app_release_digest"], installation_org_id=str(envv.get("D3_INSTALLATION_ORG_ID", "")), ) operation_facts = catalog_operation_facts( args.operation, app["app_release_digest"], metadata["tree_sha256"], start.catalog_revision, ) operation_raw = canonical_bytes(operation_facts) operation_id = idempotency_key("test1", operation_facts) state_root = args.l0_state_root.absolute() if state_root.is_symlink() or not state_root.is_dir(): raise AppReleaseError("canonical L0 state root is unavailable") runtime_contract = state_root / "config/host-compatibility.json" if runtime_contract.is_symlink() or not runtime_contract.is_file(): raise AppReleaseError("L0 state lacks its host compatibility contract") selected_worker = selected_l0_worker_release(L0_WORKER_STEMS) worker_release = runtime_worker_release_identity(bom, { "digest": selected_worker.digest, "source_commit": selected_worker.source_commit, }) catalog_store = state_root / "data/catalog" if catalog_store.is_symlink() or ( catalog_store.exists() and not catalog_store.is_dir() ): raise AppReleaseError("canonical L0 Catalog store is invalid") catalog_store.mkdir(parents=True, mode=0o700, exist_ok=True) with tempfile.TemporaryDirectory(prefix="test1-apprelease-") as evidence_raw: evidence = Path(evidence_raw) catalog_contract = evidence / "catalog-host-compatibility.json" copy_image_file( args.catalog_image, "/etc/nexuscrm/host-compatibility.json", catalog_contract, ) approval = args.approval_publication.resolve() approval_document = load_json(approval, "Approval publication") if approval_document.get("sourceDigest") != metadata["tree_sha256"]: raise AppReleaseError("Approval publication disagrees with the L2 tree") approval_digest = "sha256:" + hashlib.sha256(approval.read_bytes()).hexdigest() mail_actions = pack_root / "actions/invocable_standard_actions.json" _, frontend_receipt, _ = validate_apprelease_compatibility( bom, app_root, runtime_contract, catalog_contract, ) write_private(evidence / "frontend-receipt.json", canonical_bytes(frontend_receipt)) database_env = evidence / "database.env" write_database_env(database_env, envv) apply_env = dict(envv) apply_env.update({ "NEXUSCRM_OFFICIAL_APP_PACKS_ROOT": str(app_root), "NEXUSCRM_SETUP_BIN": str(setup), "NEXUSCRM_PACK_RUNTIME_METADATA_BIN": str(pack_metadata), "DOWNSTREAM_REF": metadata["commit"], "OFFICIAL_APP_PACKS_REF": app["artifact"]["commit"], "DEPLOY_ID": operation_id, }) require_l0_stopped_or_bootstrap() env_root = Path(str(apply_env.get("ENV_ROOT", ""))).expanduser() if not str(apply_env.get("ENV_ROOT", "")).strip(): raise AppReleaseError("database environment has no ENV_ROOT") require_staging_target_runtime_quiesced( apply_env, runtime_paths(env_root.resolve())["run_dir"], ) run_catalog_offline_candidate_phase( apply_env, l2_root, core_root, app_root, downstream_source_root=l2_root, ) if read_start() != before: raise AppReleaseError("pack apply unexpectedly changed Catalog authority") application_owner_id = "" if any(g["capability_id"] == "runtime.application-navigation.v1" for g in bom["catalog_consumers"]["runtime_capability_grants"]): app_shell = apply_runtime_app_shell( apply_env, l2_root, controller_root=ROOT, output_path=evidence / "app-shell.json", navigation_catalog_image=args.catalog_image, navigation_pages=app_root / "appservices/sfcrm/navigation_pages.json", ) application_owner_id = app_shell["frontend_app_id"] # Integration bindings name immutable pack generations created by the # apply above. Reading them before apply makes first install impossible # and makes upgrade bind the previous generation. bindings_path = evidence / "integration-bindings.json" run(( "python3", str(ROOT / "scripts/render_serviceification_d3_integration_bindings.py"), "--stack-lock", str(l2_root / "stack.lock.json"), "--runtime-env", str(args.env_file.resolve()), "--app-packs-root", str(app_root), "--l2-root", str(l2_root), "--output", str(bindings_path), )) bindings_document = load_json(bindings_path, "integration bindings") bindings = bindings_document.get("integrationBindings") if not isinstance(bindings, list): raise AppReleaseError("integration bindings are incomplete") catalog_config = render_catalog_config( bom, app_root, runtime_contract, catalog_contract, org_id=start.org_id, workload_id=bom["catalog_consumers"]["handler_workload_id"], runtime_audience=start.runtime_audience, package_mount_path=f"/release/{app['artifact']['path']}", attestation_mount_path=f"/release/{app['publisher_attestation']['path']}", artifact_store_root="/var/lib/nexuscrm/catalog", idempotency_key=operation_id, operation_kind=start.kind, app_installation_id=start.app_installation_id, expected_catalog_revision=start.catalog_revision, expected_security_epoch=start.security_epoch, publisher_valid_from="2026-01-01T00:00:00Z", publisher_valid_until="2036-01-01T00:00:00Z", schema_generation=start.schema_generation, engine_generation=start.engine_generation, integration_bindings=bindings, outbound_mail_actions_path=mail_actions if mail_actions.is_file() else None, phase_one_publication_command_id=deterministic_uuid7(operation_raw), approval_publication_bundle_path="/etc/nexuscrm/approval-publication.json", approval_publication_bundle_digest=approval_digest, refresh_active_release=start.refresh_active_release, ) catalog_config_path = evidence / "catalog-request.json" write_private(catalog_config_path, canonical_bytes(catalog_config)) if before is not None: prepare_upgrade_publisher( envv=envv, catalog_config=catalog_config, catalog_image=args.catalog_image, database_env=database_env, evidence=evidence, execute=run, ) catalog_output_path = evidence / "catalog-output.json" run(( "docker", "run", "--rm", "--network", "host", "--user", f"{os.getuid()}:{os.getgid()}", "--env-file", str(database_env), "--mount", f"type=bind,src={catalog_config_path},dst=/etc/nexuscrm/catalog-config.json,readonly", "--mount", f"type=bind,src={approval},dst=/etc/nexuscrm/approval-publication.json,readonly", "--mount", f"type=bind,src={app_root},dst=/release,readonly", "--mount", f"type=bind,src={catalog_store},dst=/var/lib/nexuscrm/catalog", "--mount", f"type=bind,src={evidence},dst=/output", args.catalog_image, "-config", "/etc/nexuscrm/catalog-config.json", "-output", "/output/catalog-output.json", )) output = load_json(catalog_output_path, "Catalog output") validate_catalog_output_for_deployment( bom, output, org_id=start.org_id, workload_id=bom["catalog_consumers"]["handler_workload_id"], ) require_published_authority( read_optional_catalog_authority(envv), output, app["app_release_digest"], start, ) serving = render_serving_inputs( bom, output, frontend_receipt, application_owner_id=application_owner_id, replica_id="test1-runtime", org_id=start.org_id, workload_id=bom["catalog_consumers"]["handler_workload_id"], catalog_store_root="/var/lib/nexuscrm/catalog", ) serving_path = evidence / "serving.json" write_private(serving_path, canonical_bytes(serving)) # L0 activation already requires every role and token. Remove the old # closure first so any interrupted preparation remains fail-closed # instead of looking like a complete mixture of two AppReleases. invalidate_worker_state(state_root) prepare_l0_state(serving_path, state_root, args.env_file.absolute()) prepare_worker_state( bom, output, worker_release=worker_release, catalog_image=args.catalog_image, database_env=database_env, envv=envv, state_root=state_root, operation_id=operation_id, ) return { "schema": "nexuscrm.test1-apprelease-installation/v1", **operation_facts, "appInstallationId": output["appInstallationId"], "catalogRevision": output["catalogRevision"], "securityEpoch": output["securityEpoch"], "catalogSnapshotDigest": output["catalogSnapshotDigest"], "runtimeServingConfigDigest": sha256_digest(serving), } def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser(description=__doc__) value.add_argument("operation", choices=("install", "upgrade")) value.add_argument("--env-file", type=Path, required=True) value.add_argument("--resolved-bom", type=Path, required=True) value.add_argument("--app-packs-root", type=Path, required=True) value.add_argument("--core-root", type=Path, required=True) value.add_argument("--l2-root", type=Path, required=True) value.add_argument("--deployment-tools", type=Path, required=True) value.add_argument("--approval-publication", type=Path, required=True) value.add_argument("--catalog-image", required=True) value.add_argument("--l0-state-root", type=Path, required=True) return value def main() -> int: args = parser().parse_args() try: print(json.dumps(apply_apprelease(args), sort_keys=True, separators=(",", ":"))) return 0 except ( AppReleaseError, DeployError, OSError, subprocess.CalledProcessError, StatePreparationError, ToolError, ValidationFailure, ) as exc: print(f"manage-test1-apprelease: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())