build_runtime_switch.py

#!/usr/bin/env python3
"""Build a transient runtime target for a manual strategy switch."""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any

SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))
ROOT = SCRIPT_DIR.parents[1]

from runtime_settings import (  # noqa: E402
    LIVE_CONTINUITY_STATES,
    SUPPORTED_PLATFORMS,
    compact_json,
    env_string,
    platform_repository,
    runtime_target_fingerprint,
    select_service_target_entry_index,
    validate_target,
)


DEFAULT_ARTIFACT_BUCKET_URI = "gs://qsl-runtime-logs-shared"
PLATFORM_CONFIG_PATH = ROOT / "platform-config.json"
IBIT_ZSCORE_EXIT_STRATEGY_PROFILE = "ibit_smart_dca"
PLATFORM_DRY_RUN_VARIABLES = {
    "schwab": "SCHWAB_DRY_RUN_ONLY",
    "longbridge": "LONGBRIDGE_DRY_RUN_ONLY",
    "ibkr": "IBKR_DRY_RUN_ONLY",
    "firstrade": "FIRSTRADE_DRY_RUN_ONLY",
    "qmt": "QMT_DRY_RUN_ONLY",
    "binance": "BINANCE_DRY_RUN",
}
PLATFORM_RESERVED_CASH_RATIO_VARIABLES = {
    "schwab": "SCHWAB_RESERVED_CASH_RATIO",
    "longbridge": "LONGBRIDGE_RESERVED_CASH_RATIO",
    "ibkr": "IBKR_RESERVED_CASH_RATIO",
    "firstrade": "FIRSTRADE_RESERVED_CASH_RATIO",
}
PLATFORM_MIN_RESERVED_CASH_VARIABLES = {
    "schwab": "SCHWAB_MIN_RESERVED_CASH_USD",
    "longbridge": "LONGBRIDGE_MIN_RESERVED_CASH_USD",
    "ibkr": "IBKR_MIN_RESERVED_CASH_USD",
    "firstrade": "FIRSTRADE_MIN_RESERVED_CASH_USD",
}
PLATFORM_CASH_ONLY_EXECUTION_VARIABLES = {
    "schwab": "SCHWAB_CASH_ONLY_EXECUTION",
    "longbridge": "LONGBRIDGE_CASH_ONLY_EXECUTION",
    "ibkr": "IBKR_CASH_ONLY_EXECUTION",
    "firstrade": "FIRSTRADE_CASH_ONLY_EXECUTION",
}
PLATFORM_FEATURE_SNAPSHOT_VARIABLES = {
    "schwab": ("SCHWAB_FEATURE_SNAPSHOT_PATH", "SCHWAB_FEATURE_SNAPSHOT_MANIFEST_PATH"),
    "longbridge": (
        "LONGBRIDGE_FEATURE_SNAPSHOT_PATH",
        "LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH",
    ),
    "ibkr": ("IBKR_FEATURE_SNAPSHOT_PATH", "IBKR_FEATURE_SNAPSHOT_MANIFEST_PATH"),
    "firstrade": (
        "FIRSTRADE_FEATURE_SNAPSHOT_PATH",
        "FIRSTRADE_FEATURE_SNAPSHOT_MANIFEST_PATH",
    ),
}
INCOME_LAYER_VARIABLES = (
    "INCOME_LAYER_ENABLED",
    "INCOME_LAYER_START_USD",
    "INCOME_LAYER_MAX_RATIO",
)
MARKET_SIGNAL_RUNTIME_SUFFIXES = (
    "MARKET_SIGNAL_HANDOFF_INDEX_URI",
    "MARKET_SIGNAL_HANDOFF_MANIFEST_URI",
    "MARKET_SIGNAL_CONSUMPTION_AUDIT_URI",
    "MARKET_SIGNAL_CACHE_DIR",
    "MARKET_SIGNAL_REQUIRED",
    "MARKET_SIGNAL_FALLBACK_MODE",
    "MARKET_SIGNAL_MAX_STALE_DAYS",
)
PLATFORM_MARKET_SIGNAL_PREFIXES = {
    "schwab": "SCHWAB",
    "longbridge": "LONGBRIDGE",
    "ibkr": "IBKR",
    "firstrade": "FIRSTRADE",
    "qmt": "QMT",
}
MARKET_SIGNAL_RUNTIME_VARIABLES = tuple(MARKET_SIGNAL_RUNTIME_SUFFIXES) + tuple(
    f"{prefix}_{suffix}"
    for prefix in PLATFORM_MARKET_SIGNAL_PREFIXES.values()
    for suffix in MARKET_SIGNAL_RUNTIME_SUFFIXES
)
CASH_ONLY_EXECUTION_VARIABLE = "CASH_ONLY_EXECUTION"
CASH_ONLY_EXECUTION_MODES = frozenset({"current", "enabled", "disabled"})
CASH_ONLY_EXECUTION_CONTROL_FIELD = "cash_only_execution_mode"
LEGACY_INCOME_LAYER_VARIABLES = (
    "INCOME_THRESHOLD_USD",
    "QQQI_INCOME_RATIO",
    "INCOME_LAYER_QQQI_WEIGHT",
    "INCOME_LAYER_SPYI_WEIGHT",
)
LEGACY_INCOME_LAYER_CONTROL_FIELDS = (
    "income_threshold_usd",
    "qqqi_income_ratio",
    "income_layer_qqqi_weight",
    "income_layer_spyi_weight",
)
OPTION_OVERLAY_CONTROL_FIELDS = (
    "option_overlay_enabled",
    "option_growth_overlay_enabled",
    "option_growth_overlay_recipe",
    "option_growth_overlay_start_usd",
    "option_growth_overlay_nav_budget_ratio",
    "option_income_overlay_enabled",
    "option_income_overlay_recipe",
    "option_income_overlay_start_usd",
    "option_income_overlay_nav_risk_ratio",
)
OPTION_OVERLAY_VARIABLES = tuple(field.upper() for field in OPTION_OVERLAY_CONTROL_FIELDS)
OPTION_OVERLAY_MODES = frozenset({"current", "enabled", "disabled"})
OPTION_OVERLAY_PROFILE_PATH = ROOT / "web" / "strategy-switch-console" / "strategy-profiles.example.json"
RUNTIME_TARGET_VARIABLES = ("RUNTIME_TARGET_ENABLED",)
DCA_PROFILES = frozenset(
    {
        "nasdaq_sp500_smart_dca",
        "ibit_smart_dca",
    }
)
DCA_SUPPORTED_PLATFORMS = frozenset({"longbridge", "ibkr", "schwab", "firstrade"})
DCA_MODES = frozenset({"fixed", "smart"})
DCA_MODE_VARIABLE = "DCA_MODE"
DCA_BASE_INVESTMENT_VARIABLE = "DCA_BASE_INVESTMENT_USD"
DCA_RUNTIME_VARIABLES = (
    DCA_MODE_VARIABLE,
    DCA_BASE_INVESTMENT_VARIABLE,
)
DCA_MODE_CONTROL_FIELD = "dca_mode"
DCA_BASE_INVESTMENT_CONTROL_FIELD = "dca_base_investment_usd"
IBIT_ZSCORE_EXIT_ENABLED_VARIABLE = "IBIT_ZSCORE_EXIT_ENABLED"
IBIT_ZSCORE_EXIT_MODE_VARIABLE = "IBIT_ZSCORE_EXIT_MODE"
IBIT_ZSCORE_EXIT_PARKING_SYMBOL_VARIABLE = "IBIT_ZSCORE_EXIT_PARKING_SYMBOL"
IBIT_ZSCORE_EXIT_RISK_REDUCED_EXPOSURE_VARIABLE = "IBIT_ZSCORE_EXIT_RISK_REDUCED_EXPOSURE"
IBIT_ZSCORE_EXIT_RISK_OFF_EXPOSURE_VARIABLE = "IBIT_ZSCORE_EXIT_RISK_OFF_EXPOSURE"
IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_WINDOW_VARIABLE = "IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW"
IBIT_ZSCORE_EXIT_RUNTIME_VARIABLES = (
    IBIT_ZSCORE_EXIT_ENABLED_VARIABLE,
    IBIT_ZSCORE_EXIT_MODE_VARIABLE,
    IBIT_ZSCORE_EXIT_PARKING_SYMBOL_VARIABLE,
    IBIT_ZSCORE_EXIT_RISK_REDUCED_EXPOSURE_VARIABLE,
    IBIT_ZSCORE_EXIT_RISK_OFF_EXPOSURE_VARIABLE,
    IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_WINDOW_VARIABLE,
)
IBIT_ZSCORE_EXIT_CONTROL_FIELDS = (
    "ibit_zscore_exit_mode",
    "ibit_zscore_exit_parking_symbol",
    "ibit_zscore_exit_risk_reduced_exposure",
    "ibit_zscore_exit_risk_off_exposure",
    "ibit_zscore_exit_allow_outside_execution_window",
)
DEFAULT_VARIABLE_SCOPE = {
    "longbridge": "environment",
    "ibkr": "repository",
    "schwab": "repository",
    "firstrade": "repository",
    "qmt": "repository",
    "binance": "repository",
}
REPOSITORY_SCOPED_SERVICE_INVENTORY_PLATFORMS = frozenset(
    {"longbridge", "schwab", "firstrade"}
)
DEFAULT_SERVICE_NAME = {
    "schwab": "charles-schwab-quant-service",
    "firstrade": "firstrade-quant-service",
    "qmt": "qmt-quant-service",
    "binance": "binance-platform",
}
PLATFORM_ALIASES = {
    "firsttrade": "firstrade",
}


def _normalize_platform(value: str) -> str:
    platform = str(value or "").strip().lower()
    platform = PLATFORM_ALIASES.get(platform, platform)
    if platform not in SUPPORTED_PLATFORMS:
        supported = ", ".join(sorted(SUPPORTED_PLATFORMS))
        raise ValueError(f"unsupported platform {value!r}; supported: {supported}")
    return platform


def _normalize_target_name(value: str) -> str:
    text = str(value or "").strip()
    if not text:
        raise ValueError("target_name is required")
    return re.sub(r"[^A-Za-z0-9._=-]+", "-", text).strip("-")


def _deployment_selector_default(platform: str, target_name: str) -> str:
    if platform in {"firstrade", "qmt"}:
        return platform
    return target_name.upper() if target_name.lower() in {"sg", "hk", "paper"} else target_name


def _account_scope_default(platform: str, deployment_selector: str) -> str:
    if platform == "firstrade":
        return "US"
    if platform == "qmt":
        return "CN"
    return deployment_selector


def _account_selector_default(platform: str, account_scope: str) -> list[str]:
    if platform in {"firstrade", "qmt"}:
        return [platform]
    return [account_scope]


def _default_service_name(platform: str, target_name: str) -> str:
    if platform in DEFAULT_SERVICE_NAME:
        return DEFAULT_SERVICE_NAME[platform]
    normalized = target_name.lower()
    if platform == "longbridge":
        return f"longbridge-quant-{normalized}-service"
    if platform == "ibkr":
        return f"interactive-brokers-{normalized}-service"
    raise ValueError(f"no default service_name for platform {platform!r}")


def _default_github_environment(platform: str, target_name: str, variable_scope: str) -> str | None:
    if variable_scope != "environment":
        return None
    if platform == "longbridge":
        return f"longbridge-{target_name.lower()}"
    return target_name


def _split_csv(value: str | None) -> list[str]:
    if not value:
        return []
    return [item.strip() for item in value.replace(";", ",").split(",") if item.strip()]


def _load_json_object(value: str, *, field_name: str) -> dict[str, Any]:
    text = str(value or "").strip()
    if not text:
        return {}
    try:
        payload = json.loads(text)
    except json.JSONDecodeError as exc:
        raise ValueError(f"{field_name} must be valid JSON") from exc
    if not isinstance(payload, dict):
        raise ValueError(f"{field_name} must decode to an object")
    return payload


def _load_json_from_file(path: str | None, *, field_name: str) -> dict[str, Any] | list[Any]:
    if not path:
        return {}
    try:
        payload = json.loads(Path(path).read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValueError(f"{field_name} must be valid JSON") from exc
    if not isinstance(payload, (dict, list)):
        raise ValueError(f"{field_name} must decode to an object or array")
    return payload


def _parse_extra_variables(pairs: list[str], raw_json: str) -> dict[str, Any]:
    extras = _load_json_object(raw_json, field_name="extra_variables_json")
    for pair in pairs:
        name, sep, value = pair.partition("=")
        if not sep or not name.strip():
            raise ValueError(f"extra variable must be NAME=VALUE, got: {pair!r}")
        extras[name.strip()] = value
    return extras


def _normalize_dca_mode(value: str) -> str:
    mode = str(value or "").strip().lower()
    aliases = {
        "ordinary": "fixed",
        "ordinary_dca": "fixed",
        "fixed_dca": "fixed",
        "smart_dca": "smart",
    }
    mode = aliases.get(mode, mode)
    if mode not in DCA_MODES:
        raise ValueError("dca_mode must be fixed or smart")
    return mode


def _validate_dca_platform(platform: str, strategy_profile: str) -> None:
    profile = str(strategy_profile or "").strip().lower()
    if profile in DCA_PROFILES and platform not in DCA_SUPPORTED_PLATFORMS:
        supported = ", ".join(sorted(DCA_SUPPORTED_PLATFORMS))
        raise ValueError(
            f"DCA strategy profiles are only supported on {supported}; "
            f"got platform={platform!r}, strategy_profile={profile!r}"
        )


def _normalize_positive_decimal(value: str, *, field_name: str) -> str:
    text = str(value or "").strip()
    if not text or not re.fullmatch(r"(?:\d+|\d*\.\d+)", text):
        raise ValueError(f"{field_name} must be a positive decimal number")
    numeric = float(text)
    if not numeric > 0:
        raise ValueError(f"{field_name} must be greater than 0")
    return text


def _normalize_nonnegative_decimal(value: str, *, field_name: str) -> str:
    text = str(value or "").strip()
    if not text or not re.fullmatch(r"(?:\d+|\d*\.\d+)", text):
        raise ValueError(f"{field_name} must be a non-negative decimal number")
    numeric = float(text)
    if numeric < 0:
        raise ValueError(f"{field_name} must be non-negative")
    return text


def _normalize_ratio_decimal(value: str, *, field_name: str) -> str:
    text = str(value or "").strip()
    if not text or not re.fullmatch(r"(?:\d+|\d*\.\d+)", text):
        raise ValueError(f"{field_name} must be a decimal number")
    numeric = float(text)
    if numeric < 0 or numeric > 1:
        raise ValueError(f"{field_name} must be between 0 and 1")
    return text


def _normalize_optional_bool_text(value: str, *, field_name: str) -> str:
    text = str(value if value is not None else "").strip().lower()
    if text in {"1", "true", "yes", "y", "on"}:
        return "true"
    if text in {"0", "false", "no", "n", "off"}:
        return "false"
    raise ValueError(f"{field_name} must be true or false")


def _normalize_option_overlay_mode(value: str) -> str:
    mode = str(value or "current").strip().lower()
    if mode not in OPTION_OVERLAY_MODES:
        raise ValueError("option_overlay_mode must be current, enabled, or disabled")
    return mode


def _normalize_option_recipe(value: str, *, field_name: str) -> str:
    text = str(value or "").strip()
    if not text or not re.fullmatch(r"[A-Za-z0-9._=-]{1,120}", text):
        raise ValueError(f"{field_name} must be a recipe slug")
    return text


def _normalize_symbol_text(value: str, *, field_name: str) -> str:
    text = str(value or "").strip().upper().removesuffix(".US")
    if not text or not re.fullmatch(r"[A-Z0-9.-]{1,12}", text):
        raise ValueError(f"{field_name} must be a symbol")
    return text


def _cash_only_extra_variables(args: argparse.Namespace, platform: str) -> dict[str, str]:
    mode = str(getattr(args, "cash_only_execution_mode", None) or "current").strip().lower()
    if mode not in CASH_ONLY_EXECUTION_MODES:
        raise ValueError("cash_only_execution_mode must be current, enabled, or disabled")
    if mode == "current":
        return {}
    variable = PLATFORM_CASH_ONLY_EXECUTION_VARIABLES.get(platform)
    if not variable:
        return {}
    return {variable: env_string(mode == "enabled")}


def _extract_cash_only_control_fields(extra_variables: dict[str, Any]) -> dict[str, Any]:
    controls: dict[str, Any] = {}
    if CASH_ONLY_EXECUTION_CONTROL_FIELD in extra_variables:
        controls[CASH_ONLY_EXECUTION_CONTROL_FIELD] = extra_variables.pop(CASH_ONLY_EXECUTION_CONTROL_FIELD)
    return controls


def _extract_dca_control_fields(extra_variables: dict[str, Any]) -> dict[str, Any]:
    controls: dict[str, Any] = {}
    for field_name in (DCA_MODE_CONTROL_FIELD, DCA_BASE_INVESTMENT_CONTROL_FIELD):
        if field_name in extra_variables:
            controls[field_name] = extra_variables.pop(field_name)
    return controls


def _extract_ibit_zscore_exit_control_fields(extra_variables: dict[str, Any]) -> dict[str, Any]:
    controls: dict[str, Any] = {}
    for field_name in IBIT_ZSCORE_EXIT_CONTROL_FIELDS:
        if field_name in extra_variables:
            controls[field_name] = extra_variables.pop(field_name)
    return controls


def _disabled_option_overlay_extra_variables() -> dict[str, str]:
    values = {variable: "" for variable in OPTION_OVERLAY_VARIABLES}
    values["OPTION_OVERLAY_ENABLED"] = "false"
    values["OPTION_GROWTH_OVERLAY_ENABLED"] = "false"
    values["OPTION_INCOME_OVERLAY_ENABLED"] = "false"
    return values


def _profile_bool(item: dict[str, Any], field_name: str, *, default: bool = False) -> bool:
    if item.get(field_name) is None or str(item.get(field_name)).strip() == "":
        return default
    return _normalize_optional_bool_text(item[field_name], field_name=field_name) == "true"


def _option_family_defaults(item: dict[str, Any], family: str) -> dict[str, str]:
    control_prefix = f"option_{family}_overlay"
    env_prefix = f"OPTION_{family.upper()}_OVERLAY"
    enabled = _profile_bool(item, f"{control_prefix}_enabled", default=False)
    values = {
        f"{env_prefix}_ENABLED": "true" if enabled else "false",
        f"{env_prefix}_RECIPE": "",
        f"{env_prefix}_START_USD": "",
    }
    if family == "growth":
        ratio_field = "option_growth_overlay_nav_budget_ratio"
        ratio_variable = "OPTION_GROWTH_OVERLAY_NAV_BUDGET_RATIO"
    else:
        ratio_field = "option_income_overlay_nav_risk_ratio"
        ratio_variable = "OPTION_INCOME_OVERLAY_NAV_RISK_RATIO"
    values[ratio_variable] = ""
    if not enabled:
        return values

    values[f"{env_prefix}_RECIPE"] = _normalize_option_recipe(
        item.get(f"{control_prefix}_recipe"),
        field_name=f"{control_prefix}_recipe",
    )
    values[f"{env_prefix}_START_USD"] = _normalize_nonnegative_decimal(
        item.get(f"{control_prefix}_start_usd"),
        field_name=f"{control_prefix}_start_usd",
    )
    values[ratio_variable] = _normalize_ratio_decimal(item.get(ratio_field), field_name=ratio_field)
    return values


def _load_option_overlay_profile_defaults() -> dict[str, dict[str, str]]:
    try:
        payload = json.loads(OPTION_OVERLAY_PROFILE_PATH.read_text(encoding="utf-8"))
    except OSError as exc:
        raise ValueError(f"cannot read {OPTION_OVERLAY_PROFILE_PATH}") from exc
    except json.JSONDecodeError as exc:
        raise ValueError(f"{OPTION_OVERLAY_PROFILE_PATH} must be valid JSON") from exc
    if not isinstance(payload, list):
        raise ValueError(f"{OPTION_OVERLAY_PROFILE_PATH} must contain a strategy profile list")

    defaults: dict[str, dict[str, str]] = {}
    for item in payload:
        if not isinstance(item, dict):
            continue
        profile = str(item.get("profile") or item.get("strategy_profile") or "").strip().lower()
        if not profile:
            continue
        if not _profile_bool(item, "option_overlay_enabled", default=False):
            continue
        values = _disabled_option_overlay_extra_variables()
        values["OPTION_OVERLAY_ENABLED"] = "true"
        values.update(_option_family_defaults(item, "growth"))
        values.update(_option_family_defaults(item, "income"))
        if values["OPTION_GROWTH_OVERLAY_ENABLED"] != "true" and values["OPTION_INCOME_OVERLAY_ENABLED"] != "true":
            raise ValueError(f"{profile} option overlay is enabled without a growth or income family")
        defaults[profile] = values
    return defaults


def _option_overlay_extra_variables(args: argparse.Namespace, strategy_profile: str) -> dict[str, str]:
    mode = _normalize_option_overlay_mode(getattr(args, "option_overlay_mode", "current"))
    if mode == "current":
        return {}
    if mode == "disabled":
        return _disabled_option_overlay_extra_variables()

    defaults = _load_option_overlay_profile_defaults().get(strategy_profile)
    if not defaults:
        raise ValueError("option_overlay_mode enabled is only supported for strategies with option overlay defaults")
    return dict(defaults)


def _dca_extra_variables(
    args: argparse.Namespace,
    strategy_profile: str,
    controls: dict[str, Any] | None = None,
) -> dict[str, Any]:
    controls = dict(controls or {})
    is_dca_profile = strategy_profile in DCA_PROFILES
    dca_mode = args.dca_mode if str(args.dca_mode or "").strip() else controls.get(DCA_MODE_CONTROL_FIELD, "")
    dca_base_investment_usd = (
        args.dca_base_investment_usd
        if str(args.dca_base_investment_usd or "").strip()
        else controls.get(DCA_BASE_INVESTMENT_CONTROL_FIELD, "")
    )
    has_dca_mode = bool(str(dca_mode or "").strip())
    has_dca_base = bool(str(dca_base_investment_usd or "").strip())
    if not is_dca_profile:
        if has_dca_mode or has_dca_base:
            raise ValueError("DCA settings are only supported for DCA strategy profiles")
        return {variable: "" for variable in DCA_RUNTIME_VARIABLES}

    extra_variables: dict[str, Any] = {}
    if has_dca_mode:
        extra_variables[DCA_MODE_VARIABLE] = _normalize_dca_mode(dca_mode)
    if has_dca_base:
        extra_variables[DCA_BASE_INVESTMENT_VARIABLE] = _normalize_positive_decimal(
            dca_base_investment_usd,
            field_name="dca_base_investment_usd",
        )
    return extra_variables


def _effective_dca_mode(
    args: argparse.Namespace,
    strategy_profile: str,
    controls: dict[str, Any] | None = None,
) -> str:
    if strategy_profile not in DCA_PROFILES:
        return ""
    controls = dict(controls or {})
    raw_mode = args.dca_mode if str(args.dca_mode or "").strip() else controls.get(DCA_MODE_CONTROL_FIELD, "")
    return _normalize_dca_mode(raw_mode) if str(raw_mode or "").strip() else "fixed"


def _reject_direct_dca_extra_variables(extra_variables: dict[str, Any]) -> None:
    provided = [
        variable
        for variable in DCA_RUNTIME_VARIABLES
        if variable in extra_variables and str(extra_variables.get(variable) or "").strip()
    ]
    if provided:
        names = ", ".join(provided)
        raise ValueError(
            f"use dca_mode and dca_base_investment_usd control fields instead of extra_variables_json for {names}"
        )


def _reject_direct_ibit_zscore_exit_extra_variables(extra_variables: dict[str, Any]) -> None:
    provided = [
        variable
        for variable in IBIT_ZSCORE_EXIT_RUNTIME_VARIABLES
        if variable in extra_variables and str(extra_variables.get(variable) or "").strip()
    ]
    if provided:
        names = ", ".join(provided)
        raise ValueError(
            "IBIT_ZSCORE_EXIT variables are derived from ibit_smart_dca smart DCA mode; "
            f"do not set them directly: {names}"
        )


def _reject_research_only_extra_variables(extra_variables: dict[str, Any]) -> None:
    blocked = [
        name
        for name in (
            *OPTION_OVERLAY_CONTROL_FIELDS,
            *OPTION_OVERLAY_VARIABLES,
            *LEGACY_INCOME_LAYER_CONTROL_FIELDS,
            *LEGACY_INCOME_LAYER_VARIABLES,
        )
        if name in extra_variables
    ]
    if blocked:
        names = ", ".join(blocked)
        raise ValueError(
            "direct option overlay settings and legacy income controls are research-only "
            f"and are not supported by live strategy switch settings: {names}"
        )


def _ibit_zscore_exit_extra_variables(
    args: argparse.Namespace,
    strategy_profile: str,
    plugin_mode: str,
    dca_mode: str,
) -> dict[str, Any]:
    is_ibit_profile = strategy_profile == IBIT_ZSCORE_EXIT_STRATEGY_PROFILE
    if not is_ibit_profile:
        return {variable: "" for variable in IBIT_ZSCORE_EXIT_RUNTIME_VARIABLES}

    mode = "live" if plugin_mode != "none" and dca_mode == "smart" else "disabled"
    return {
        IBIT_ZSCORE_EXIT_ENABLED_VARIABLE: "true" if mode != "disabled" else "false",
        IBIT_ZSCORE_EXIT_MODE_VARIABLE: "paper" if mode == "disabled" else mode,
        IBIT_ZSCORE_EXIT_PARKING_SYMBOL_VARIABLE: _normalize_symbol_text(
            "BOXX",
            field_name="ibit_zscore_exit_parking_symbol",
        ),
        IBIT_ZSCORE_EXIT_RISK_REDUCED_EXPOSURE_VARIABLE: _normalize_ratio_decimal(
            "0.50",
            field_name="ibit_zscore_exit_risk_reduced_exposure",
        ),
        IBIT_ZSCORE_EXIT_RISK_OFF_EXPOSURE_VARIABLE: _normalize_ratio_decimal(
            "0.25",
            field_name="ibit_zscore_exit_risk_off_exposure",
        ),
        IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_WINDOW_VARIABLE: _normalize_optional_bool_text(
            "true",
            field_name="ibit_zscore_exit_allow_outside_execution_window",
        ),
    }


def _auto_plugin_mounts(strategy_profile: str, artifact_bucket_uri: str, dca_mode: str = "") -> list[dict[str, Any]]:
    """Retire name-based plugin selection without expanding runtime authority."""
    del strategy_profile, artifact_bucket_uri, dca_mode
    return []


def _current_plugin_mounts(args: argparse.Namespace, strategy_profile: str) -> list[dict[str, Any]]:
    path = str(getattr(args, "current_plugin_mounts_json_file", "") or "").strip()
    if not path:
        raise ValueError("plugin_mode=current requires current_plugin_mounts_json_file")
    payload = _load_json_from_file(path, field_name="current_plugin_mounts_json_file")
    if not isinstance(payload, dict):
        raise ValueError("current_plugin_mounts_json_file must contain an object")
    mounts = payload.get("strategy_plugins")
    if not isinstance(mounts, list) or any(not isinstance(item, dict) for item in mounts):
        raise ValueError("current_plugin_mounts_json_file.strategy_plugins must be an array of objects")
    unexpected = {
        str(item.get("strategy") or "").strip()
        for item in mounts
        if str(item.get("strategy") or "").strip() != strategy_profile
    }
    if unexpected:
        raise ValueError(
            "plugin_mode=current only preserves mounts for the selected strategy; "
            f"found: {', '.join(sorted(unexpected))}"
        )
    return [dict(item) for item in mounts]


def _plugin_mounts(args: argparse.Namespace, strategy_profile: str, dca_mode: str = "") -> list[dict[str, Any]]:
    mode = str(args.plugin_mode or "none").strip().lower()
    if mode == "none":
        return []
    if mode == "auto":
        return _auto_plugin_mounts(strategy_profile, args.artifact_bucket_uri, dca_mode)
    if mode == "current":
        return _current_plugin_mounts(args, strategy_profile)
    if mode == "custom":
        raise ValueError(
            "legacy custom plugin mounts are retired; a P1/P2/P3-bound strategy_plugin_signal.v2 adapter is required"
        )
    raise ValueError(f"unsupported plugin_mode {args.plugin_mode!r}")


def _execution_mode_and_dry_run(raw_mode: str) -> tuple[str, bool]:
    mode = str(raw_mode or "").strip().lower()
    if mode == "live":
        return "live", False
    if mode in {"paper", "dry_run", "dry-run"}:
        # Platform sync adapters currently consume this legacy no-order
        # envelope. Policy validation maps it back to the canonical dry_run
        # control mode; do not alter deployed payload compatibility here.
        return "paper", True
    raise ValueError("execution_mode must be live, paper, or dry_run")


def _validate_requested_execution_mode(strategy_profile: str, raw_mode: str) -> None:
    """Keep legacy ``paper`` requests within their original strategy scope.

    The emitted no-order envelope is intentionally compatible with deployed
    platform parsers. A new ``dry_run`` request may use it for any strategy;
    an explicit old ``paper`` request must still be listed in the strategy
    profile so it cannot broaden a historical mode allowance.
    """
    if str(raw_mode or "").strip().lower() != "paper":
        return
    config = _load_platform_config()
    strategies = config.get("strategies")
    strategy = strategies.get(strategy_profile) if isinstance(strategies, dict) else None
    if not isinstance(strategy, dict):
        return
    allowed_modes = strategy.get("allowed_execution_modes")
    if isinstance(allowed_modes, str):
        allowed = {part.strip().lower() for part in re.split(r"[,\s/|]+", allowed_modes) if part.strip()}
    elif isinstance(allowed_modes, (list, tuple, set)):
        allowed = {str(part or "").strip().lower() for part in allowed_modes if str(part or "").strip()}
    else:
        allowed = set()
    if allowed and "paper" not in allowed:
        raise ValueError(f"strategy {strategy_profile!r} does not allow paper execution; use dry_run")


def _load_platform_config() -> dict[str, Any]:
    try:
        payload = json.loads(PLATFORM_CONFIG_PATH.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"unable to load platform config {PLATFORM_CONFIG_PATH}: {exc}") from exc
    if not isinstance(payload, dict):
        raise ValueError("platform-config.json must contain an object")
    return payload


def _scheduler_plan_for_strategy(
    strategy_profile: str,
    plugin_mounts: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (),
) -> dict[str, str]:
    profile = str(strategy_profile or "").strip().lower()
    config = _load_platform_config()
    scheduling = config.get("scheduling")
    strategies = config.get("strategies")
    domains = config.get("domains")
    if not isinstance(scheduling, dict) or not isinstance(strategies, dict) or not isinstance(domains, dict):
        raise ValueError("platform-config.json must define scheduling, strategies, and domains")
    scheduler_profiles = scheduling.get("profiles")
    strategy = strategies.get(profile)
    if not isinstance(scheduler_profiles, dict) or not isinstance(strategy, dict):
        raise ValueError(f"strategy {profile!r} is missing from the scheduling catalog")

    domain = domains.get(strategy.get("domain"))
    if not isinstance(domain, dict):
        raise ValueError(f"strategy {profile!r} has no configured scheduling domain")
    scheduler_profile = strategy.get("scheduler_profile") or domain.get("scheduler_profile")
    plugin_overrides = strategy.get("scheduler_profile_by_plugin") or {}
    if not isinstance(plugin_overrides, dict):
        raise ValueError(f"strategy {profile!r} scheduler_profile_by_plugin must be an object")
    for mount in plugin_mounts:
        if (
            not isinstance(mount, dict)
            or mount.get("strategy") != profile
            or mount.get("enabled") is not True
        ):
            continue
        plugin = str(mount.get("plugin") or "").strip()
        if plugin in plugin_overrides:
            scheduler_profile = plugin_overrides[plugin]

    scheduler = scheduler_profiles.get(scheduler_profile)
    if not isinstance(scheduler, dict):
        raise ValueError(
            f"strategy {profile!r} references unknown scheduler profile {scheduler_profile!r}"
        )
    return {str(key): str(value) for key, value in scheduler.items()}


def _market_plan_for_strategy(strategy_profile: str) -> dict[str, str]:
    profile = str(strategy_profile or "").strip().lower()
    config = _load_platform_config()
    strategies = config.get("strategies")
    domains = config.get("domains")
    strategy = strategies.get(profile) if isinstance(strategies, dict) else None
    if not isinstance(strategy, dict) or not isinstance(domains, dict):
        raise ValueError(f"strategy {profile!r} is missing from the market catalog")
    domain = domains.get(strategy.get("domain"))
    if not isinstance(domain, dict):
        raise ValueError(f"strategy {profile!r} has no configured market domain")
    market = {
        field: str(domain.get(field) or "").strip()
        for field in ("market", "market_calendar", "market_timezone")
    }
    missing = [field for field, value in market.items() if not value]
    if missing:
        raise ValueError(
            f"strategy {profile!r} market domain is missing {', '.join(missing)}"
        )
    return market


def _feature_snapshot_extra_variables(
    platform: str,
    strategy_profile: str,
    extra_variables: dict[str, Any],
) -> dict[str, str]:
    variable_names = PLATFORM_FEATURE_SNAPSHOT_VARIABLES.get(platform)
    if not variable_names:
        return {}
    snapshot_variable, manifest_variable = variable_names

    config = _load_platform_config()
    strategies = config.get("strategies")
    strategy = strategies.get(strategy_profile) if isinstance(strategies, dict) else None
    if not isinstance(strategy, dict):
        raise ValueError(f"strategy {strategy_profile!r} is missing from the runtime artifact catalog")
    runtime_artifacts = strategy.get("runtime_artifacts") or {}
    if not isinstance(runtime_artifacts, dict):
        raise ValueError(f"strategy {strategy_profile!r} runtime_artifacts must be an object")
    feature_snapshot = runtime_artifacts.get("feature_snapshot") or {}
    if not isinstance(feature_snapshot, dict):
        raise ValueError(
            f"strategy {strategy_profile!r} runtime_artifacts.feature_snapshot must be an object"
        )

    explicit = snapshot_variable in extra_variables or manifest_variable in extra_variables
    if explicit:
        snapshot_path = extra_variables.get(snapshot_variable)
        manifest_path = extra_variables.get(manifest_variable)
    else:
        snapshot_path = feature_snapshot.get("path")
        manifest_path = feature_snapshot.get("manifest_path")
    snapshot_path = snapshot_path.strip() if isinstance(snapshot_path, str) else ""
    manifest_path = manifest_path.strip() if isinstance(manifest_path, str) else ""
    if explicit and not feature_snapshot and (snapshot_path or manifest_path):
        raise ValueError(
            f"strategy {strategy_profile!r} does not accept feature snapshot artifacts"
        )
    if bool(snapshot_path) != bool(manifest_path):
        raise ValueError(
            f"strategy {strategy_profile!r} feature snapshot path and manifest path must be configured together"
        )
    if feature_snapshot.get("required") is True and not (snapshot_path and manifest_path):
        raise ValueError(
            f"strategy {strategy_profile!r} requires feature snapshot path and manifest path"
        )
    return {
        snapshot_variable: snapshot_path,
        manifest_variable: manifest_path,
    }


def _build_runtime_target(args: argparse.Namespace) -> dict[str, Any]:
    platform = _normalize_platform(args.platform)
    target_name = _normalize_target_name(args.target_name)
    execution_mode, dry_run_only = _execution_mode_and_dry_run(args.execution_mode)
    deployment_selector = (
        args.deployment_selector.strip()
        if args.deployment_selector
        else _deployment_selector_default(platform, target_name)
    )
    account_scope = (
        args.account_scope.strip() if args.account_scope else _account_scope_default(platform, deployment_selector)
    )
    account_selector = _split_csv(args.account_selector) or _account_selector_default(platform, account_scope)
    service_name = args.service_name.strip() if args.service_name else _default_service_name(platform, target_name)
    strategy_profile = args.strategy_profile.strip().lower()
    _validate_requested_execution_mode(strategy_profile, args.execution_mode)
    _validate_dca_platform(platform, strategy_profile)
    runtime_target: dict[str, Any] = {
        "platform_id": platform,
        "strategy_profile": strategy_profile,
        "dry_run_only": dry_run_only,
        "deployment_selector": deployment_selector,
        "account_selector": account_selector,
        "account_scope": account_scope,
        "service_name": service_name,
        "execution_mode": execution_mode,
        "scheduler": _scheduler_plan_for_strategy(strategy_profile),
        **_market_plan_for_strategy(strategy_profile),
    }
    execution_windows = _load_json_object(args.execution_windows_json, field_name="execution_windows_json")
    if execution_windows:
        runtime_target["execution_windows"] = execution_windows
    _apply_live_continuity(runtime_target, args)
    return runtime_target


def _apply_live_continuity(runtime_target: dict[str, Any], args: argparse.Namespace) -> None:
    """Attach a frozen legacy baseline without granting new live authority."""

    state = str(getattr(args, "live_continuity_state", "") or "").strip().upper()
    if not state or state == "NONE":
        return
    if state not in LIVE_CONTINUITY_STATES:
        raise ValueError(f"live_continuity_state must be one of NONE, {', '.join(sorted(LIVE_CONTINUITY_STATES))}")
    if runtime_target.get("execution_mode") != "live" or runtime_target.get("dry_run_only") is not False:
        raise ValueError("live_continuity is only valid for an execution_mode=live, non-dry-run target")
    baseline_id = str(getattr(args, "live_continuity_baseline_id", "") or "").strip()
    captured_at = str(getattr(args, "live_continuity_captured_at", "") or "").strip()
    if not baseline_id or not captured_at:
        raise ValueError(
            "live_continuity_baseline_id and live_continuity_captured_at are required when live_continuity_state is set"
        )
    runtime_target["live_continuity"] = {
        "state": state,
        "baseline_kind": "legacy_authorized",
        "baseline_id": baseline_id,
        "baseline_target_sha256": runtime_target_fingerprint(runtime_target),
        "captured_at": captured_at,
    }


def _build_target_entry(
    *,
    platform: str,
    runtime_target: dict[str, Any],
    mounts_variable: str,
    mounts: list[dict[str, Any]],
    extra_variables: dict[str, Any],
) -> dict[str, Any]:
    service_name = str(runtime_target["service_name"])
    entry: dict[str, Any] = {
        "service": service_name,
        "runtime_target": dict(runtime_target),
    }
    if platform == "ibkr":
        entry["ACCOUNT_GROUP"] = runtime_target["account_scope"]
    dry_run_variable = PLATFORM_DRY_RUN_VARIABLES.get(platform)
    if dry_run_variable:
        entry[dry_run_variable] = env_string(runtime_target["dry_run_only"])
    if mounts_variable:
        entry[mounts_variable] = {"strategy_plugins": mounts}
    entry.update(extra_variables)
    return entry


def _preserve_reserved_cash_fields(
    *,
    platform: str,
    current_entry: dict[str, Any],
    replacement: dict[str, Any],
) -> None:
    for variable in (
        PLATFORM_MIN_RESERVED_CASH_VARIABLES.get(platform),
        PLATFORM_RESERVED_CASH_RATIO_VARIABLES.get(platform),
        PLATFORM_CASH_ONLY_EXECUTION_VARIABLES.get(platform),
        CASH_ONLY_EXECUTION_VARIABLE,
        *INCOME_LAYER_VARIABLES,
        *OPTION_OVERLAY_VARIABLES,
        *RUNTIME_TARGET_VARIABLES,
        *DCA_RUNTIME_VARIABLES,
        *IBIT_ZSCORE_EXIT_RUNTIME_VARIABLES,
        *MARKET_SIGNAL_RUNTIME_VARIABLES,
    ):
        if variable and variable not in replacement and variable in current_entry:
            replacement[variable] = current_entry[variable]


def _patch_service_targets(
    *,
    current_payload: dict[str, Any] | list[Any],
    platform: str,
    runtime_target: dict[str, Any],
    mounts_variable: str,
    mounts: list[dict[str, Any]],
    extra_variables: dict[str, Any],
    allow_create: bool,
) -> dict[str, Any] | list[dict[str, Any]]:
    payload = dict(current_payload) if isinstance(current_payload, dict) else None
    if payload is not None:
        raw_entries = payload.get("targets", [])
        if not isinstance(raw_entries, list):
            raise ValueError("service targets must be an array")
    else:
        raw_entries = current_payload
    if any(not isinstance(item, dict) for item in raw_entries):
        raise ValueError("service target entries must be objects")
    entries = [dict(item) for item in raw_entries]
    replacement = _build_target_entry(
        platform=platform,
        runtime_target=runtime_target,
        mounts_variable=mounts_variable,
        mounts=mounts,
        extra_variables=extra_variables,
    )

    matched_index = select_service_target_entry_index(
        runtime_target,
        entries,
        allow_account_scope_fallback=not allow_create,
    )
    if matched_index is not None:
        current_entry = entries[matched_index]
        _preserve_reserved_cash_fields(
            platform=platform,
            current_entry=current_entry,
            replacement=replacement,
        )
        entries[matched_index] = {**current_entry, **replacement}
    elif not allow_create:
        platform_label = "IBKR " if platform == "ibkr" else ""
        raise ValueError(
            f"existing {platform_label}service target was not found; "
            "use --allow-create-service-target to append a new target"
        )
    else:
        entries.append(replacement)
    if payload is None:
        return entries
    payload["targets"] = entries
    return payload


def build_switch_target(args: argparse.Namespace) -> dict[str, Any]:
    platform = _normalize_platform(args.platform)
    target_name = _normalize_target_name(args.target_name)
    variable_scope = args.variable_scope or DEFAULT_VARIABLE_SCOPE[platform]
    if variable_scope not in {"repository", "environment"}:
        raise ValueError("variable_scope must be repository or environment")
    github_environment = args.github_environment or _default_github_environment(platform, target_name, variable_scope)
    runtime_target = _build_runtime_target(args)
    extra_variables = _parse_extra_variables(args.extra_variable, args.extra_variables_json)
    cash_only_controls = _extract_cash_only_control_fields(extra_variables)
    dca_controls = _extract_dca_control_fields(extra_variables)
    _extract_ibit_zscore_exit_control_fields(extra_variables)
    effective_dca_mode = _effective_dca_mode(args, runtime_target["strategy_profile"], dca_controls)
    mounts = _plugin_mounts(args, runtime_target["strategy_profile"], effective_dca_mode)
    runtime_target["scheduler"] = _scheduler_plan_for_strategy(runtime_target["strategy_profile"], mounts)
    mounts_variable = f"{SUPPORTED_PLATFORMS[platform]['plugin_mounts_prefix']}STRATEGY_PLUGIN_MOUNTS_JSON"
    if cash_only_controls.get(CASH_ONLY_EXECUTION_CONTROL_FIELD):
        args.cash_only_execution_mode = str(cash_only_controls[CASH_ONLY_EXECUTION_CONTROL_FIELD]).strip().lower()
    _reject_direct_dca_extra_variables(extra_variables)
    _reject_direct_ibit_zscore_exit_extra_variables(extra_variables)
    _reject_research_only_extra_variables(extra_variables)
    extra_variables.update(
        _feature_snapshot_extra_variables(
            platform,
            runtime_target["strategy_profile"],
            extra_variables,
        )
    )

    if args.set_platform_dry_run_variable:
        extra_variables[PLATFORM_DRY_RUN_VARIABLES[platform]] = env_string(runtime_target["dry_run_only"])
    if args.reserved_cash_ratio:
        extra_variables[PLATFORM_RESERVED_CASH_RATIO_VARIABLES[platform]] = args.reserved_cash_ratio
    if args.min_reserved_cash_usd:
        extra_variables[PLATFORM_MIN_RESERVED_CASH_VARIABLES[platform]] = args.min_reserved_cash_usd
    if args.income_layer_start_usd:
        extra_variables["INCOME_LAYER_START_USD"] = args.income_layer_start_usd
    if args.income_layer_max_ratio:
        extra_variables["INCOME_LAYER_MAX_RATIO"] = args.income_layer_max_ratio
    extra_variables.update(_cash_only_extra_variables(args, platform))
    extra_variables.update(_option_overlay_extra_variables(args, runtime_target["strategy_profile"]))
    extra_variables.update(_dca_extra_variables(args, runtime_target["strategy_profile"], dca_controls))
    effective_plugin_mode = str(args.plugin_mode or "none").strip().lower()
    if effective_plugin_mode == "auto":
        effective_plugin_mode = "none"
    extra_variables.update(
        _ibit_zscore_exit_extra_variables(
            args,
            runtime_target["strategy_profile"],
            effective_plugin_mode,
            effective_dca_mode,
        )
    )

    top_level_mounts = mounts
    plugin_mounts_variable: str | None = mounts_variable
    repository_variables: dict[str, Any] = {}
    if args.existing_service_targets_json_file:
        service_targets = _load_json_from_file(
            args.existing_service_targets_json_file,
            field_name="existing_service_targets_json_file",
        )
        patched_service_targets = _patch_service_targets(
            current_payload=service_targets,
            platform=platform,
            runtime_target=runtime_target,
            mounts_variable=mounts_variable,
            mounts=mounts,
            extra_variables=extra_variables,
            allow_create=args.allow_create_service_target,
        )
        if (
            variable_scope == "environment"
            and platform in REPOSITORY_SCOPED_SERVICE_INVENTORY_PLATFORMS
        ):
            repository_variables = {"CLOUD_RUN_SERVICE_TARGETS_JSON": patched_service_targets}
        else:
            extra_variables = {"CLOUD_RUN_SERVICE_TARGETS_JSON": patched_service_targets}
            top_level_mounts = []
            plugin_mounts_variable = None

    target: dict[str, Any] = {
        "target_id": f"{platform}/{target_name}",
        "description": "Generated by build_runtime_switch.py for manual workflow dispatch.",
        "github": {
            "repository": platform_repository(platform),
            "variable_scope": variable_scope,
        },
        "runtime_target": runtime_target,
        "extra_variables": extra_variables,
    }
    if github_environment:
        target["github"]["environment"] = github_environment
    if plugin_mounts_variable:
        target["plugin_mounts_variable"] = plugin_mounts_variable
        target["plugin_mounts"] = top_level_mounts
    if repository_variables:
        target["repository_variables"] = repository_variables
    errors = validate_target(target)
    if errors:
        raise ValueError("; ".join(errors))
    return target


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--platform", required=True, choices=sorted((*SUPPORTED_PLATFORMS, *PLATFORM_ALIASES)))
    parser.add_argument("--target-name", required=True)
    parser.add_argument("--strategy-profile", required=True)
    parser.add_argument("--execution-mode", choices=("live", "paper", "dry_run"), default="live")
    parser.add_argument("--variable-scope", choices=("repository", "environment"))
    parser.add_argument("--github-environment", default="")
    parser.add_argument("--deployment-selector", default="")
    parser.add_argument("--account-selector", default="")
    parser.add_argument("--account-scope", default="")
    parser.add_argument("--service-name", default="")
    parser.add_argument("--execution-windows-json", default="")
    parser.add_argument(
        "--live-continuity-state",
        choices=("NONE", *sorted(LIVE_CONTINUITY_STATES)),
        default="NONE",
        help="Freeze an already authorised legacy baseline; does not create new live authority.",
    )
    parser.add_argument("--live-continuity-baseline-id", default="")
    parser.add_argument("--live-continuity-captured-at", default="")
    parser.add_argument("--plugin-mode", choices=("auto", "current", "none", "custom"), default="none")
    parser.add_argument("--current-plugin-mounts-json-file", default="")
    parser.add_argument("--custom-plugin-mounts-json", default="")
    parser.add_argument("--artifact-bucket-uri", default=DEFAULT_ARTIFACT_BUCKET_URI)
    parser.add_argument("--extra-variables-json", default="", help="JSON object of non-secret extra variables")
    parser.add_argument("--extra-variable", action="append", default=[], help="NAME=VALUE non-secret extra variable")
    parser.add_argument("--reserved-cash-ratio", default="")
    parser.add_argument("--min-reserved-cash-usd", default="")
    parser.add_argument(
        "--cash-only-execution-mode",
        choices=sorted(CASH_ONLY_EXECUTION_MODES),
        default="current",
    )
    parser.add_argument("--income-layer-start-usd", default="")
    parser.add_argument("--income-layer-max-ratio", default="")
    parser.add_argument("--option-overlay-mode", choices=sorted(OPTION_OVERLAY_MODES), default="current")
    parser.add_argument("--dca-mode", default="")
    parser.add_argument("--dca-base-investment-usd", default="")
    parser.add_argument("--existing-service-targets-json-file", default="")
    parser.add_argument("--allow-create-service-target", action="store_true")
    parser.add_argument("--no-platform-dry-run-variable", dest="set_platform_dry_run_variable", action="store_false")
    parser.set_defaults(set_platform_dry_run_variable=True)
    parser.add_argument("--output", default="-", help="output path, or '-' for stdout")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        target = build_switch_target(args)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    payload = compact_json(target)
    if args.output == "-":
        print(payload)
    else:
        Path(args.output).write_text(payload + "\n", encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())