#!/usr/bin/env python3 """Converge the release-owned CRM application shell in a TiDB runtime. This release-configuration adapter owns the shell definition and visibility. Navigation is seeded once through the published neutral initializer; existing documents and their application owner survive repeated deployment. Staging, offline adapters, and D3 use the same contract. """ from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import shutil import re import sys import subprocess import tempfile from typing import Any class AppShellError(RuntimeError): pass def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Apply and verify a release-owned CRM app shell.") parser.add_argument("--config", required=True, help="Path to the app-shell JSON configuration.") parser.add_argument("--output", help="Optional JSON evidence output path.") parser.add_argument("--assert-only", action="store_true", help="Verify existing metadata without mutation.") parser.add_argument("--expected-config-sha256", help="Fail unless the config matches this release hash.") parser.add_argument("--db-host", default=os.environ.get("TIDB_HOST", "")) parser.add_argument("--db-port", type=int, default=int(os.environ.get("TIDB_PORT", "4000"))) parser.add_argument("--db-user", default=os.environ.get("TIDB_USER", "")) parser.add_argument("--db-password", default=os.environ.get("TIDB_PASSWORD", "")) parser.add_argument("--db-database", default=os.environ.get("TIDB_DATABASE", "")) parser.add_argument("--db-tls", default=os.environ.get("TIDB_TLS", "true")) parser.add_argument("--controlplane-image", "--navigation-catalog-image", default=os.environ.get("NEXUSCRM_CONTROLPLANE_IMAGE", ""), help="Exact Controlplane image digest selected by the release.") parser.add_argument("--controlplane-bin", help="Explicit local Controlplane executable for local verification.") parser.add_argument("--local-candidate", type=Path, help="Explicit host-only unsigned candidate selecting the local image.") parser.add_argument("--local-controlplane-source-revision", help="Exact source commit for an explicit host-only local Controlplane image override.") parser.add_argument("--navigation-pages", type=Path, help="Sealed L1 page projection for legacy navigation migration.") return parser.parse_args() def required_string(payload: dict[str, Any], key: str) -> str: value = payload.get(key) if not isinstance(value, str) or not value.strip(): raise AppShellError(f"app-shell config missing non-empty {key}") return value.strip() def navigation_permission_tabs(config: dict[str, Any]) -> list[str]: tabs = config.get("tabPermissions") if not isinstance(tabs, list) or not all(isinstance(tab, str) and tab.strip() for tab in tabs): raise AppShellError("app-shell requires an explicit tabPermissions list") if len({tab.casefold() for tab in tabs}) != len(tabs): raise AppShellError("tabPermissions contains duplicate identities") return tabs def navigation_tabs(items: list[dict[str, Any]]) -> list[str]: tabs: list[str] = [] seen: set[str] = set() for item in items: item_type = str(item.get("type") or "").strip() tab = "" if item_type == "object": tab = str(item.get("objectApiName") or "").strip() if not tab: raise AppShellError(f"object navigation item {item.get('id')!r} is missing objectApiName") elif item_type == "page": tab = str(item.get("objectApiName") or item.get("label") or item.get("id") or "").strip() else: continue if not tab: raise AppShellError(f"navigation item {item.get('id')!r} is missing a tab identity") lowered = tab.lower() if lowered not in seen: seen.add(lowered) tabs.append(tab) return tabs def load_app_shell(path: Path) -> tuple[dict[str, Any], str]: raw = path.read_bytes() try: payload = json.loads(raw) except json.JSONDecodeError as exc: raise AppShellError(f"invalid app-shell JSON: {path}") from exc if not isinstance(payload, dict): raise AppShellError("app-shell config must be a JSON object") for key in ("id", "label", "frontendAppId", "appMenuItemId", "startUrl"): required_string(payload, key) if not isinstance(payload.get("isDefault"), bool): raise AppShellError("app-shell config isDefault must be boolean") items = payload.get("navigationItems") if not isinstance(items, list): raise AppShellError("app-shell config navigationItems must be an explicit list") seen_ids: set[str] = set() normalized_items: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): raise AppShellError("each navigation item must be a JSON object") item_id = required_string(item, "id") if item_id in seen_ids: raise AppShellError(f"duplicate navigation item id: {item_id}") seen_ids.add(item_id) target = item.get("target") if target is None: required_string(item, "type") elif not isinstance(target, dict) or len(target) != 1 or not set(target) <= {"page", "resource", "url"}: raise AppShellError("navigation template requires a single neutral target") required_string(item, "label") normalized_items.append(item) if "tabPermissions" in payload: navigation_permission_tabs(payload) else: navigation_tabs(items) return payload, hashlib.sha256(raw).hexdigest() def sql_string(value: str) -> str: return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'" def translate_navigation(items: Any, pages: dict[str, str]) -> list[dict[str, Any]]: """Translate legacy L2 targets using the selected L1 page identities.""" if not isinstance(items, list): raise AppShellError("legacy navigation must be an array") translated = [] for item in items: if not isinstance(item, dict): raise AppShellError("legacy navigation item must be an object") if "target" in item: translated.append(item) # The neutral initializer validates typed items. continue result = {key: item[key] for key in ("id", "label", "icon") if key in item} kind = item.get("type") if kind == "object": target = {"resource": {"resource_type": required_string(item, "objectApiName")}} elif kind == "page": route = required_string(item, "pageUrl") identities = [key for key, value in pages.items() if value == route] if len(identities) != 1: raise AppShellError(f"legacy page route has no unique L1 identity: {route!r}") target = {"page": {"id": identities[0]}} elif kind == "dashboard": target = {"resource": {"resource_type": "Dashboard", "resource_id": required_string(item, "dashboardId")}} elif kind == "web": target = {"url": required_string(item, "pageUrl")} else: raise AppShellError(f"unsupported legacy navigation target: {kind!r}") result["target"] = target translated.append(result) return translated def prepare_navigation_initialization(args: argparse.Namespace, config: dict[str, Any]) -> dict[str, Any]: """Validate the exact seed/migration input before any shell mutation.""" pages = json.loads(args.navigation_pages.read_text()) if getattr(args, "navigation_pages", None) else {} if not isinstance(pages, dict) or not all( isinstance(key, str) and isinstance(value, str) for key, value in pages.items() ): raise AppShellError("L1 navigation page projection must map identities to routes") request = { "schema": "nexuscrm.application-navigation-initialization/v1", "owner_id": required_string(config, "frontendAppId"), "application_id": required_string(config, "id"), "template": translate_navigation(config["navigationItems"], pages), } rows = mysql_query( args, "SELECT FrontendAppId, NavigationRevision, COALESCE(HEX(NavigationItems), '') " f"FROM `_System_App` WHERE Id={sql_string(request['application_id'])}", ) observed = expect_single_row(rows, 3, "navigation initialization source") if rows.strip() else None if observed is not None and observed[0] != request["owner_id"]: raise AppShellError("navigation owner disagrees with sealed L2 configuration") if observed is not None and int(observed[1]) == 0 and observed[2]: raw = bytes.fromhex(observed[2]) current = json.loads(raw) if current is not None: request["migration"] = { "source_digest": hashlib.sha256(raw).hexdigest(), "items": translate_navigation(current, pages), } run_navigation_initializer(args, request, validate_only=True) if observed is not None and int(observed[1]) > 0: # Validate a preserved current document before changing its shell too. run_navigation_initializer(args, request, assert_only=True) return request def run_navigation_initializer(args: argparse.Namespace, request: dict[str, Any], *, validate_only: bool = False, assert_only: bool = False) -> dict[str, Any]: payload = request binary = getattr(args, "controlplane_bin", None) image = getattr(args, "controlplane_image", "") output_root = Path(getattr(args, "output", None)).expanduser().resolve().parent if getattr(args, "output", None) else Path.cwd() output_root.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix=".navigation-init-", dir=output_root) as temporary: directory = Path(temporary) (directory / "input.json").write_text(json.dumps(payload), encoding="utf-8") environment = os.environ.copy() environment.update({"TIDB_HOST": args.db_host, "TIDB_PORT": str(args.db_port), "TIDB_USER": args.db_user, "TIDB_PASSWORD": args.db_password, "TIDB_DATABASE": args.db_database, "TIDB_TLS_DISABLE": str(str(args.db_tls).lower() not in {"1", "true", "yes", "on"}).lower()}) if binary: command = [str(Path(binary).expanduser().resolve()), "application-navigation", "-input", str(directory / "input.json"), "-output", str(directory / "output.json")] else: command = ["docker", "run", "--rm", "--pull", "never", "--network", "host", "--user", f"{os.getuid()}:{os.getgid()}", "--mount", f"type=bind,src={directory},dst=/navigation"] for key in ("TIDB_HOST", "TIDB_PORT", "TIDB_USER", "TIDB_PASSWORD", "TIDB_DATABASE", "TIDB_TLS_DISABLE"): command.extend(["--env", key]) command.extend([image, "application-navigation", "-input", "/navigation/input.json", "-output", "/navigation/output.json"]) if validate_only: command.append("-validate-only") elif assert_only: command.append("-assert-only") result = subprocess.run(command, env=environment, capture_output=True, text=True, timeout=90, check=False) if result.returncode: raise AppShellError("Controlplane navigation initialization failed: " + (result.stderr or result.stdout)[-2000:]) if validate_only: return {} try: receipt = json.loads((directory / "output.json").read_text()) except (OSError, json.JSONDecodeError) as exc: raise AppShellError("Controlplane did not return navigation evidence") from exc if receipt.get("schema") != payload["schema"] or receipt.get("disposition") not in {"initialized", "preserved", "migrated"}: raise AppShellError("invalid Controlplane navigation evidence") return receipt def initialize_navigation(args: argparse.Namespace, config: dict[str, Any], *, request: dict[str, Any] | None = None, validate_only: bool = False) -> dict[str, Any]: if not getattr(args, "controlplane_image", None): args.controlplane_image = getattr(args, "navigation_catalog_image", "") validate_controlplane(args) if request is None: request = prepare_navigation_initialization(args, config) if validate_only: return {} result = run_navigation_initializer(args, request, assert_only=args.assert_only) result["revision"] = result["document"]["revision"] return result def build_apply_sql(config: dict[str, Any], *, navigation_seeded_by_initializer: bool = False) -> str: app_id = required_string(config, "id") app_name = str(config.get("name") or app_id).strip() or app_id label = required_string(config, "label") frontend_app_id = required_string(config, "frontendAppId") description = str(config.get("description") or "") icon = str(config.get("icon") or "") color = str(config.get("color") or "") app_menu_item_id = required_string(config, "appMenuItemId") start_url = required_string(config, "startUrl") is_default = 1 if config["isDefault"] else 0 items = config["navigationItems"] assert isinstance(items, list) tabs = navigation_permission_tabs(config) if "tabPermissions" in config else navigation_tabs(items) nav_json = json.dumps(items, separators=(",", ":"), ensure_ascii=True) navigation_seed = "CAST('null' AS JSON)" if navigation_seeded_by_initializer or any("target" in item for item in items) else f"CAST({sql_string(nav_json)} AS JSON)" tab_rows = "\n UNION ALL ".join(f"SELECT {sql_string(tab)} AS TabName" for tab in tabs) if not tabs: tab_rows = "SELECT CAST(NULL AS CHAR) AS TabName WHERE FALSE" return f""" SET @now = NOW(); SET @app_id = {sql_string(app_id)}; SET @nav = {navigation_seed}; INSERT INTO `_System_App` (Id, Name, Label, Description, NamespacePrefix, Icon, Color, FrontendAppId, NavigationItems, UtilityItems, AssignedProfiles, IsDefault, CreatedDate, LastModifiedDate) VALUES (@app_id, {sql_string(app_name)}, {sql_string(label)}, {sql_string(description)}, NULL, {sql_string(icon)}, {sql_string(color)}, {sql_string(frontend_app_id)}, @nav, JSON_ARRAY(), JSON_ARRAY(), {is_default}, @now, @now) ON DUPLICATE KEY UPDATE Label = VALUES(Label), Description = VALUES(Description), Icon = VALUES(Icon), Color = VALUES(Color), UtilityItems = VALUES(UtilityItems), AssignedProfiles = VALUES(AssignedProfiles), IsDefault = VALUES(IsDefault), LastModifiedDate = VALUES(LastModifiedDate); INSERT INTO `AppMenuItem` (Id, Label, Description, StartUrl, Type, IsVisible, SortOrder, AppId, Icon, Color, IsDeleted, CreatedDate, LastModifiedDate) VALUES ({sql_string(app_menu_item_id)}, {sql_string(label)}, {sql_string(description)}, {sql_string(start_url)}, 'App', 1, 0, @app_id, {sql_string(icon)}, {sql_string(color)}, 0, @now, @now) ON DUPLICATE KEY UPDATE Label = VALUES(Label), Description = VALUES(Description), StartUrl = VALUES(StartUrl), Type = VALUES(Type), IsVisible = VALUES(IsVisible), SortOrder = VALUES(SortOrder), AppId = VALUES(AppId), Icon = VALUES(Icon), Color = VALUES(Color), IsDeleted = VALUES(IsDeleted), LastModifiedDate = VALUES(LastModifiedDate); -- The release-owned CRM shell controls this app's visibility for active -- profiles. Recreate those rows so legacy rows cannot preserve a hidden or -- non-default state through a warm deploy. DELETE pav FROM `ProfileAppVisibility` pav JOIN `Profile` p ON BINARY p.Id = BINARY pav.ProfileId WHERE BINARY pav.AppId = BINARY @app_id AND p.IsActive = 1 AND p.IsDeleted = 0; INSERT INTO `ProfileAppVisibility` (Id, ProfileId, AppId, IsVisible, IsDefault, CreatedDate, LastModifiedDate) SELECT CONCAT('pav_', REPLACE(@app_id, '-', ''), '_', REPLACE(p.Id, '-', '')), p.Id, @app_id, 1, {is_default}, @now, @now FROM `Profile` p WHERE p.IsActive = 1 AND p.IsDeleted = 0 ON DUPLICATE KEY UPDATE IsVisible = VALUES(IsVisible), IsDefault = IF(VALUES(IsVisible) = 1, VALUES(IsDefault), 0), LastModifiedDate = VALUES(LastModifiedDate); -- ProfileTabSetting has no app identifier. The CRM shell owns only its -- configured tabs for active profiles, so recreate exactly that slice rather -- than leaving legacy rows with a conflicting visibility behind. DELETE pts FROM `ProfileTabSetting` pts JOIN `Profile` p ON BINARY p.Id = BINARY pts.ProfileId JOIN ( {tab_rows} ) t ON BINARY LOWER(pts.TabName) = BINARY LOWER(t.TabName) WHERE p.IsActive = 1 AND p.IsDeleted = 0; INSERT INTO `ProfileTabSetting` (Id, ProfileId, TabName, Visibility, CreatedDate, LastModifiedDate) SELECT CONCAT('pts_', SHA2(CONCAT(@app_id, ':', p.Id, ':', t.TabName), 256)), p.Id, t.TabName, 'DefaultOn', @now, @now FROM `Profile` p JOIN ( {tab_rows} ) t WHERE p.IsActive = 1 AND p.IsDeleted = 0; """.strip() def mysql_base_args(args: argparse.Namespace) -> list[str]: missing = [ name for name, value in (("TIDB_HOST", args.db_host), ("TIDB_USER", args.db_user), ("TIDB_DATABASE", args.db_database)) if not str(value or "").strip() ] if missing: raise AppShellError(f"missing database settings: {', '.join(missing)}") client = shutil.which("mariadb") or shutil.which("mysql") if not client: raise AppShellError("mysql or mariadb CLI is required") command = [client, "--batch", "--raw", "--skip-column-names", "--connect-timeout=10"] if str(args.db_tls).strip().lower() in {"", "1", "true", "yes", "on"}: command.append("--ssl") command.extend(["-h", args.db_host, "-P", str(args.db_port), "-u", args.db_user, args.db_database]) return command def mysql_query(args: argparse.Namespace, sql: str) -> str: env = os.environ.copy() env["MYSQL_PWD"] = args.db_password result = subprocess.run( [*mysql_base_args(args), "-e", sql], check=False, capture_output=True, env=env, input=None, text=True, ) if result.returncode != 0: detail = (result.stderr or result.stdout).strip() raise AppShellError(f"app-shell database command failed: {detail}") return result.stdout def mysql_apply(args: argparse.Namespace, sql: str) -> None: env = os.environ.copy() env["MYSQL_PWD"] = args.db_password result = subprocess.run( mysql_base_args(args), check=False, capture_output=True, env=env, input="START TRANSACTION;\n" + sql.rstrip().rstrip(";") + ";\nCOMMIT;\n", text=True, ) if result.returncode != 0: detail = (result.stderr or result.stdout).strip() raise AppShellError(f"app-shell database apply failed: {detail}") def expect_single_row(output: str, columns: int, label: str) -> list[str]: rows = [line for line in output.splitlines() if line] if len(rows) != 1: raise AppShellError(f"{label} expected one row, got {len(rows)}") values = rows[0].split("\t") if len(values) != columns: raise AppShellError(f"{label} expected {columns} columns, got {len(values)}") return values def assert_app_shell(args: argparse.Namespace, config: dict[str, Any], config_hash: str, navigation: dict[str, Any] | None = None) -> dict[str, Any]: app_id = required_string(config, "id") items = config["navigationItems"] assert isinstance(items, list) frontend_app_id = required_string(config, "frontendAppId") expected_default = "1" if config["isDefault"] else "0" app_row = expect_single_row( mysql_query( args, "SELECT COALESCE(FrontendAppId, ''), CAST(IsDefault AS CHAR), CAST(NavigationItems AS CHAR) " + (", NavigationRevision " if navigation is not None else "") + f"FROM `_System_App` WHERE Id={sql_string(app_id)}", ), 4 if navigation is not None else 3, "_System_App", ) if app_row[0] != frontend_app_id or app_row[1] != expected_default: raise AppShellError("_System_App frontend/default metadata does not match app-shell config") try: actual_navigation = json.loads(app_row[2]) except json.JSONDecodeError as exc: raise AppShellError("_System_App NavigationItems is not valid JSON") from exc if not isinstance(actual_navigation, list): raise AppShellError("_System_App NavigationItems must be an array") revision = 0 if navigation is not None: document = navigation["document"] revision = int(app_row[3]) if document.get("application_id") != app_id or revision < document["revision"]: raise AppShellError("current navigation predates the committed Controlplane result") if revision == document["revision"] and actual_navigation != document["items"]: raise AppShellError("current navigation changed without advancing its revision") menu_row = expect_single_row( mysql_query( args, "SELECT Label, StartUrl, AppId, CAST(IsVisible AS CHAR), CAST(IsDeleted AS CHAR) " f"FROM `AppMenuItem` WHERE Id={sql_string(required_string(config, 'appMenuItemId'))}", ), 5, "AppMenuItem", ) expected_menu = [required_string(config, "label"), required_string(config, "startUrl"), app_id, "1", "0"] if menu_row != expected_menu: raise AppShellError("AppMenuItem does not match app-shell config") profile_row = expect_single_row( mysql_query( args, "SELECT COUNT(*), " "COUNT(DISTINCT CASE WHEN pav.IsVisible=1 THEN p.Id END), " "COUNT(DISTINCT CASE WHEN pav.IsVisible=1 AND pav.IsDefault=1 THEN p.Id END) " "FROM `Profile` p LEFT JOIN `ProfileAppVisibility` pav " f"ON BINARY pav.ProfileId=BINARY p.Id AND BINARY pav.AppId=BINARY {sql_string(app_id)} " "WHERE p.IsActive=1 AND p.IsDeleted=0", ), 3, "ProfileAppVisibility", ) active_profiles, visible_profiles, default_profiles = (int(value or "0") for value in profile_row) expected_default_profiles = active_profiles if config["isDefault"] else 0 if visible_profiles != active_profiles or default_profiles != expected_default_profiles: raise AppShellError("ProfileAppVisibility does not converge for active profiles") tabs = navigation_permission_tabs(config) if "tabPermissions" in config else navigation_tabs(items) if tabs: tab_sql = ",".join(sql_string(tab) for tab in tabs) tab_rows = mysql_query( args, "SELECT settings.TabName, COUNT(DISTINCT settings.ProfileId) " "FROM `ProfileTabSetting` settings " "JOIN `Profile` p ON BINARY p.Id=BINARY settings.ProfileId " f"WHERE settings.TabName IN ({tab_sql}) AND settings.Visibility='DefaultOn' " "AND p.IsActive=1 AND p.IsDeleted=0 " "GROUP BY settings.TabName ORDER BY settings.TabName", ) tab_counts = { name.casefold(): int(count) for name, count in (line.split("\t", 1) for line in tab_rows.splitlines() if line) } missing_tabs = [tab for tab in tabs if tab_counts.get(tab.casefold()) != active_profiles] if missing_tabs: observed = ", ".join(f"{name}={count}" for name, count in sorted(tab_counts.items())) or "none" raise AppShellError( "ProfileTabSetting did not converge: " f"missing={', '.join(missing_tabs)}; active_profiles={active_profiles}; " f"observed_default_on={observed}" ) return { "schema_version": 1, "contract": "runtime-app-shell", "config_sha256": config_hash, "app_id": app_id, "frontend_app_id": frontend_app_id, "is_default": bool(config["isDefault"]), "navigation_item_ids": [str(item["id"]) for item in actual_navigation], "navigation_revision": revision, "active_profile_count": active_profiles, "default_profile_count": default_profiles, "status": "passed", } def validate_controlplane(args: argparse.Namespace) -> None: """Reject bad selection before database mutation; D3 owns publication admission.""" binary = getattr(args, "controlplane_bin", None) image = getattr(args, "controlplane_image", "") candidate = getattr(args, "local_candidate", None) local_revision = getattr(args, "local_controlplane_source_revision", None) if local_revision and (binary or candidate or not re.fullmatch(r"[0-9a-f]{40}", local_revision)): raise AppShellError("local Controlplane override requires an exact source revision and unambiguous image selection") if bool(binary) == bool(image): raise AppShellError("select exactly one Controlplane executable or immutable image") if binary: executable = Path(binary).expanduser().resolve() if candidate or not executable.is_file() or not os.access(executable, os.X_OK): raise AppShellError("local Controlplane executable is unavailable or selection is ambiguous") return if local_revision: if os.environ.get("GITHUB_ACTIONS", "").lower() == "true": raise AppShellError("unsigned local deployment is host-only") sys.path.insert(0, str(Path(__file__).resolve().parent)) import serviceification_d3_engine as d3 args.controlplane_image = d3.verify_local_role_image( "catalog_reconcile", image, local_revision, "pingcap-inc/nexuscrm-controlplane", capture_output=d3.default_capture, ) elif candidate: if os.environ.get("GITHUB_ACTIONS", "").lower() == "true": raise AppShellError("unsigned local deployment is host-only") # Reuse the existing local candidate contract and image inspection. sys.path.insert(0, str(Path(__file__).resolve().parent)) import serviceification_d3_engine as d3 bom = d3.load_bom(candidate.expanduser().resolve()) if not d3.is_local_test3_bom(bom) or d3.validate_deployment_bom_view(bom): raise AppShellError("invalid unsigned local candidate") if d3.exact_images(bom)["catalog_reconcile"] != image: raise AppShellError("Controlplane image does not match local candidate") d3.verify_local_test3_images(bom, roles=("catalog_reconcile",), capture_output=d3.default_capture) elif not re.fullmatch(r"ghcr\.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@sha256:[0-9a-f]{64}", image): raise AppShellError("Controlplane image requires the owner repository and immutable digest") else: inspected = subprocess.run(["docker", "image", "inspect", image], capture_output=True, text=True, timeout=30) if inspected.returncode: raise AppShellError("release-selected Controlplane image is unavailable locally") def write_output(path: Path | None, payload: dict[str, Any]) -> None: rendered = json.dumps(payload, indent=2, sort_keys=True) + "\n" if path: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(rendered, encoding="utf-8") else: print(rendered, end="") def main() -> int: args = parse_args() config_path = Path(args.config).expanduser().resolve() config, config_hash = load_app_shell(config_path) expected_hash = str(args.expected_config_sha256 or "").strip() if expected_hash and expected_hash != config_hash: raise AppShellError("app-shell config hash does not match the release contract") navigation_enabled = bool(args.controlplane_image or args.controlplane_bin) if args.navigation_pages and not navigation_enabled: raise AppShellError("navigation page projection requires a Controlplane image") if not navigation_enabled and any("target" in item for item in config["navigationItems"]): raise AppShellError("neutral navigation requires the release-selected Controlplane image") if navigation_enabled: validate_controlplane(args) navigation_request = prepare_navigation_initialization(args, config) if navigation_enabled else None if not args.assert_only: mysql_apply(args, build_apply_sql(config, navigation_seeded_by_initializer=navigation_enabled)) navigation_result = initialize_navigation(args, config, request=navigation_request) if navigation_enabled else None summary = assert_app_shell(args, config, config_hash, navigation_result) if navigation_result is not None: summary["navigation_initialization"] = navigation_result summary["mode"] = "assert" if args.assert_only else "apply-and-assert" write_output(Path(args.output).expanduser().resolve() if args.output else None, summary) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except AppShellError as exc: raise SystemExit(f"app-shell: {exc}") from exc