Byted Escloud

                                 Apache License
                           Version 2.0, January 2004
                        http://www.apache.org/licenses/

   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION

   1. Definitions.

      "License" shall mean the terms and conditions for use, reproduction,
      and distribution as defined by Sections 1 through 9 of this document.

      "Licensor" shall mean the copyright owner or entity authorized by
      the copyright owner that is granting the License.

      "Legal Entity" shall mean the union of the acting entity and all
      other entities that control, are controlled by, or are under common
      control with that entity. For the purposes of this definition,
      "control" means (i) the power, direct or indirect, to cause the
      direction or management of such entity, whether by contract or
      otherwise, or (ii) ownership of fifty percent (50%) or more of the
      outstanding shares, or (iii) beneficial ownership of such entity.

      "You" (or "Your") shall mean an individual or Legal Entity
      exercising permissions granted by this License.

      "Source" form shall mean the preferred form for making modifications,
      including but not limited to software source code, documentation
      source, and configuration files.

      "Object" form shall mean any form resulting from mechanical
      transformation or translation of a Source form, including but
      not limited to compiled object code, generated documentation,
      and conversions to other media types.

      "Work" shall mean the work of authorship, whether in Source or
      Object form, made available under the License, as indicated by a
      copyright notice that is included in or attached to the work
      (an example is provided in the Appendix below).

      "Derivative Works" shall mean any work, whether in Source or Object
      form, that is based on (or derived from) the Work and for which the
      editorial revisions, annotations, elaborations, or other modifications
      represent, as a whole, an original work of authorship. For the purposes
      of this License, Derivative Works shall not include works that remain
      separable from, or merely link (or bind by name) to the interfaces of,
      the Work and Derivative Works thereof.

      "Contribution" shall mean any work of authorship, including
      the original version of the Work and any modifications or additions
      to that Work or Derivative Works thereof, that is intentionally
      submitted to Licensor for inclusion in the Work by the copyright owner
      or by an individual or Legal Entity authorized to submit on behalf of
      the copyright owner. For the purposes of this definition, "submitted"
      means any form of electronic, verbal, or written communication sent
      to the Licensor or its representatives, including but not limited to
      communication on electronic mailing lists, source code control systems,
      and issue tracking systems that are managed by, or on behalf of, the
      Licensor for the purpose of discussing and improving the Work, but
      excluding communication that is conspicuously marked or otherwise
      designated in writing by the copyright owner as "Not a Contribution."

      "Contributor" shall mean Licensor and any individual or Legal Entity
      on behalf of whom a Contribution has been received by Licensor and
      subsequently incorporated within the Work.

   2. Grant of Copyright License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      copyright license to reproduce, prepare Derivative Works of,
      publicly display, publicly perform, sublicense, and distribute the
      Work and such Derivative Works in Source or Object form.

   3. Grant of Patent License. Subject to the terms and conditions of
      this License, each Contributor hereby grants to You a perpetual,
      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
      (except as stated in this section) patent license to make, have made,
      use, offer to sell, sell, import, and otherwise transfer the Work,
      where such license applies only to those patent claims licensable
      by such Contributor that are necessarily infringed by their
      Contribution(s) alone or by combination of their Contribution(s)
      with the Work to which such Contribution(s) was submitted. If You
      institute patent litigation against any entity (including a
      cross-claim or counterclaim in a lawsuit) alleging that the Work
      or a Contribution incorporated within the Work constitutes direct
      or contributory patent infringement, then any patent licenses
      granted to You under this License for that Work shall terminate
      as of the date such litigation is filed.

   4. Redistribution. You may reproduce and distribute copies of the
      Work or Derivative Works thereof in any medium, with or without
      modifications, and in Source or Object form, provided that You
      meet the following conditions:

      (a) You must give any other recipients of the Work or
          Derivative Works a copy of this License; and

      (b) You must cause any modified files to carry prominent notices
          stating that You changed the files; and

      (c) You must retain, in the Source form of any Derivative Works
          that You distribute, all copyright, patent, trademark, and
          attribution notices from the Source form of the Work,
          excluding those notices that do not pertain to any part of
          the Derivative Works; and

      (d) If the Work includes a "NOTICE" text file as part of its
          distribution, then any Derivative Works that You distribute must
          include a readable copy of the attribution notices contained
          within such NOTICE file, excluding those notices that do not
          pertain to any part of the Derivative Works, in at least one
          of the following places: within a NOTICE text file distributed
          as part of the Derivative Works; within the Source form or
          documentation, if provided along with the Derivative Works; or,
          within a display generated by the Derivative Works, if and
          wherever such third-party notices normally appear. The contents
          of the NOTICE file are for informational purposes only and
          do not modify the License. You may add Your own attribution
          notices within Derivative Works that You distribute, alongside
          or as an addendum to the NOTICE text from the Work, provided
          that such additional attribution notices cannot be construed
          as modifying the License.

      You may add Your own copyright statement to Your modifications and
      may provide additional or different license terms and conditions
      for use, reproduction, or distribution of Your modifications, or
      for any such Derivative Works as a whole, provided Your use,
      reproduction, and distribution of the Work otherwise complies with
      the conditions stated in this License.

   5. Submission of Contributions. Unless You explicitly state otherwise,
      any Contribution intentionally submitted for inclusion in the Work
      by You to the Licensor shall be under the terms and conditions of
      this License, without any additional terms or conditions.
      Notwithstanding the above, nothing herein shall supersede or modify
      the terms of any separate license agreement you may have executed
      with Licensor regarding such Contributions.

   6. Trademarks. This License does not grant permission to use the trade
      names, trademarks, service marks, or product names of the Licensor,
      except as required for reasonable and customary use in describing the
      origin of the Work and reproducing the content of the NOTICE file.

   7. Disclaimer of Warranty. Unless required by applicable law or
      agreed to in writing, Licensor provides the Work (and each
      Contributor provides its Contributions) on an "AS IS" BASIS,
      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
      implied, including, without limitation, any warranties or conditions
      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
      PARTICULAR PURPOSE. You are solely responsible for determining the
      appropriateness of using or redistributing the Work and assume any
      risks associated with Your exercise of permissions under this License.

   8. Limitation of Liability. In no event and under no legal theory,
      whether in tort (including negligence), contract, or otherwise,
      unless required by applicable law (such as deliberate and grossly
      negligent acts) or agreed to in writing, shall any Contributor be
      liable to You for damages, including any direct, indirect, special,
      incidental, or consequential damages of any character arising as a
      result of this License or out of the use or inability to use the
      Work (including but not limited to damages for loss of goodwill,
      work stoppage, computer failure or malfunction, or any and all
      other commercial damages or losses), even if such Contributor
      has been advised of the possibility of such damages.

   9. Accepting Warranty or Additional Liability. While redistributing
      the Work or Derivative Works thereof, You may choose to offer,
      and charge a fee for, acceptance of support, warranty, indemnity,
      or other liability obligations and/or rights consistent with this
      License. However, in accepting such obligations, You may act only
      on Your own behalf and on Your sole responsibility, not on behalf
      of any other Contributor, and only if You agree to indemnify,
      defend, and hold each Contributor harmless for any liability
      incurred by, or claims asserted against, such Contributor by reason
      of your accepting any such warranty or additional liability.

volcengine-python-sdk==5.0.19
opensearch-py>=2.0.0
requests==2.32.5

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from abc import ABC, abstractmethod
from typing import Any


class ApiError(Exception):
    pass


class EsCloudApi(ABC):
    @abstractmethod
    def _call(self, method: str, action: str, body: Any) -> Any:
        pass

    def describe_instances(self, body: Any) -> Any:
        return self._call("POST", "DescribeInstances", body)

    def describe_instance(self, body: Any) -> Any:
        return self._call("POST", "DescribeInstance", body)

    def describe_zones(self, body: Any) -> Any:
        return self._call("POST", "DescribeZones", body)

    def describe_node_available_specs(self, body: Any) -> Any:
        return self._call("POST", "DescribeNodeAvailableSpecs", body)

    def create_instance_in_one_step(self, body: Any) -> Any:
        return self._call("POST", "CreateInstanceInOneStep", body)

    def modify_node_spec_in_one_step(self, body: Any) -> Any:
        return self._call("POST", "ModifyNodeSpecInOneStep", body)

    def release_instance(self, body: Any) -> Any:
        return self._call("POST", "ReleaseInstance", body)

    def describe_ip_allow_list(self, body: Any) -> Any:
        return self._call("POST", "DescribeIpAllowList", body)

    def modify_ip_allow_list_v2(self, body: Any) -> Any:
        return self._call("POST", "ModifyIpAllowListV2", body)

    def reset_admin_password(self, body: Any) -> Any:
        return self._call("POST", "ResetAdminPassword", body)

    def describe_instance_nodes(self, body: Any) -> Any:
        return self._call("POST", "DescribeInstanceNodes", body)

    def describe_instance_plugins(self, body: Any) -> Any:
        return self._call("POST", "DescribeInstancePlugins", body)

    def rename_instance(self, body: Any) -> Any:
        return self._call("POST", "RenameInstance", body)

    def modify_maintenance_setting(self, body: Any) -> Any:
        return self._call("POST", "ModifyMaintenanceSetting", body)

    def modify_deletion_protection(self, body: Any) -> Any:
        return self._call("POST", "ModifyDeletionProtection", body)

    def restart_node(self, body: Any) -> Any:
        return self._call("POST", "RestartNode", body)

    def create_public_address(self, body: Any) -> Any:
        return self._call("POST", "CreatePublicAddress", body)

    def release_public_address(self, body: Any) -> Any:
        return self._call("POST", "ReleasePublicAddress", body)


class VpcApi(ABC):
    @abstractmethod
    def _call(self, method: str, action: str, body: Any) -> Any:
        pass

    def describe_vpcs(self, body: Any) -> Any:
        return self._call("GET", "DescribeVpcs", body)

    def describe_subnets(self, body: Any) -> Any:
        return self._call("GET", "DescribeSubnets", body)

    def describe_eip_addresses(self, body: Any) -> Any:
        return self._call("GET", "DescribeEipAddresses", body)

    def allocate_eip_address(self, body: Any) -> Any:
        return self._call("GET", "AllocateEipAddress", body)

    def release_eip_address(self, body: Any) -> Any:
        return self._call("GET", "ReleaseEipAddress", body)

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import os
import sys
import urllib.parse
import urllib.request
from typing import Any, Dict


from api import ApiError, EsCloudApi, VpcApi


class _SimpleResponse:
    def __init__(self, payload: Dict[str, Any]):
        self._payload = payload

    def to_dict(self) -> Dict[str, Any]:
        return self._payload


api_host = os.environ.get("ARK_SKILL_API_BASE")
api_key = os.environ.get("ARK_SKILL_API_KEY")


def check_is_ark_env() -> bool:
    return bool(api_host and api_key)


def _do_http_call(
    service_name: str,
    action: str,
    version: str,
    method: str,
    body_dict: Dict[str, Any] | None = None,
) -> Dict[str, Any]:
    if not check_is_ark_env():
        raise ApiError(
            "ARK_SKILL_API_BASE and ARK_SKILL_API_KEY must be set for ark_shim"
        )

    url = f"{api_host.rstrip('/')}/?Action={action}&Version={version}"
    headers = {
        "ServiceName": service_name,
        "Authorization": f"Bearer {api_key}",
        "Accept": "application/json",
    }

    if method.upper() == "GET":
        if body_dict:
            query = urllib.parse.urlencode(body_dict, doseq=True)
            url = f"{url}&{query}"
        data = None
    else:
        headers["Content-Type"] = "application/json"
        data = json.dumps(body_dict or {}).encode("utf-8")

    req = urllib.request.Request(url, data=data, headers=headers, method=method.upper())
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8")
        raise ApiError(f"HTTP {exc.code}: {exc.reason} - {error_body}")
    except Exception as exc:
        raise ApiError(f"Network error: {str(exc)}")


class ESCloudHttpShim(EsCloudApi):
    def _call(self, method: str, action: str, body: Any) -> Any:
        payload = body.to_dict() if hasattr(body, "to_dict") else (body or {})
        return _SimpleResponse(
            _do_http_call("ESCloud", action, "2023-01-01", method, payload)
        )


class VpcHttpShim(VpcApi):
    def _call(self, method: str, action: str, body: Any) -> Any:
        payload = body.to_dict() if hasattr(body, "to_dict") else (body or {})
        return _SimpleResponse(
            _do_http_call("vpc", action, "2020-04-01", method, payload)
        )


def get_clients():
    try:
        region = os.environ.get("VOLCENGINE_REGION", "cn-beijing")
        return ESCloudHttpShim(), VpcHttpShim(), None, region
    except ApiError as exc:
        print(
            json.dumps(
                {"error": "Initialization Error", "details": str(exc)},
                ensure_ascii=False,
            )
        )
        sys.exit(1)

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import json
import sys
from typing import Any, Dict, List, Optional, Set, Tuple

from api import ApiError, EsCloudApi, VpcApi


def print_result(data: Any) -> None:
    print(json.dumps({"status": "success", "data": data}, ensure_ascii=False))


def print_error(msg: str, details: Optional[str] = None) -> None:
    err: Dict[str, Any] = {"error": msg}
    if details:
        err["details"] = details
    print(json.dumps(err, ensure_ascii=False))
    sys.exit(1)


def api_call(fn):
    try:
        response = fn()
        try:
            result = response.to_dict()
        except AttributeError:
            result = str(response)
        print_result(result)
    except ApiError as e:
        msg = str(e)
        instr = ""
        if "TaskIsRunning" in msg:
            instr = "An operation is already in progress. Wait a few minutes and retry, or check status via 'detail'."
        elif any(
            k in msg
            for k in ["BadRequest", "InvalidParameter", "NotFound", "Unauthorized"]
        ):
            instr = "Verify IDs/specs and permissions. Use 'vpc', 'subnet', and 'node_specs' to fetch valid options."
        details = f"{msg}\n\nInstruction: {instr}" if instr else msg
        print_error("API Error", details)
    except Exception as e:
        print_error(
            "Unexpected Error",
            f"{str(e)}\n\nInstruction: Check network connectivity and credentials.",
        )


def str_to_bool(v: str) -> bool:
    if v.lower() in ("true", "1", "yes", "y"):
        return True
    if v.lower() in ("false", "0", "no", "n"):
        return False
    raise argparse.ArgumentTypeError(f"Boolean value expected, got '{v}'")


def _get_clients() -> Tuple[EsCloudApi, VpcApi, Any, str]:
    import ark_shim

    if ark_shim.check_is_ark_env():
        return ark_shim.get_clients()

    import sdk_shim

    return sdk_shim.get_clients()


def get_zone_id_by_subnet(vpc_api: VpcApi, subnet_id: str) -> str:
    resp = vpc_api.describe_subnets({"SubnetIds": [subnet_id]})
    resp_dict = (
        resp.to_dict()
        if hasattr(resp, "to_dict")
        else (resp if isinstance(resp, dict) else {})
    )
    result = resp_dict.get("Result") or resp_dict.get("result") or resp_dict
    subnets = result.get("Subnets") or result.get("subnets") or []
    if not subnets:
        print_error(
            "Subnet Not Found",
            f"Subnet ID '{subnet_id}' does not exist or is not in the current region. Instruction: Run 'vpc' and 'subnet --vpc-id <ID>' to list valid options.",
        )
    first = subnets[0] if isinstance(subnets, list) and subnets else {}
    zone_id = first.get("ZoneId") or first.get("zone_id") or ""
    if not zone_id:
        print_error(
            "Zone Not Found",
            "Failed to derive zone_id from the subnet. Try a different subnet.",
        )
    return zone_id


def _resolve_eip_billing_type(val):
    """
    EIP billing type constants.
    """
    EIP_BILL_TYPES = {
        "PostPaidByBandwidth": 1,
        "PostPaidByTraffic": 2,
        "PrePaid": 3,
    }
    return EIP_BILL_TYPES.get(val, val)


def cmd_list(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {"PageNumber": args.page_number, "PageSize": args.page_size}
    api_call(lambda: es_api.describe_instances(body))


def cmd_detail(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {"InstanceId": args.id}
    api_call(lambda: es_api.describe_instance(body))


def cmd_zones(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {}
    api_call(lambda: es_api.describe_zones(body))


def cmd_node_specs(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {}
    api_call(lambda: es_api.describe_node_available_specs(body))


def cmd_vpc(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {}
    api_call(lambda: vpc_api.describe_vpcs(body))


def cmd_subnet(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {"VpcId": args.vpc_id}
    api_call(lambda: vpc_api.describe_subnets(body))


def cmd_eip_list(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {}
    if args.status:
        body["Status"] = args.status
    api_call(lambda: vpc_api.describe_eip_addresses(body))


def cmd_eip_allocate(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    billing_type = (
        _resolve_eip_billing_type(args.billing_type) or 2
    )  # Default to PostPaidByTraffic
    body = {
        "Bandwidth": args.bandwidth,
        "BillingType": billing_type,
        "Name": args.name,
        "Description": args.description,
        "ISP": args.isp,
    }
    api_call(lambda: vpc_api.allocate_eip_address(body))


def cmd_eip_release(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {"AllocationId": args.allocation_id}
    api_call(lambda: vpc_api.release_eip_address(body))


def cmd_create(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    zone_id = get_zone_id_by_subnet(vpc_api, args.subnet_id)

    node_specs = []
    if args.master_spec:
        node_specs.append(
            {
                "Type": "Master",
                "Number": args.master_count,
                "ResourceSpecName": args.master_spec,
                "StorageSpecName": args.master_storage_spec or args.hot_storage_spec,
                "StorageSize": args.master_storage_size or 20,
            }
        )
    node_specs.append(
        {
            "Type": "Hot",
            "Number": args.hot_count,
            "ResourceSpecName": args.hot_spec,
            "StorageSpecName": args.hot_storage_spec,
            "StorageSize": args.hot_storage_size,
        }
    )
    if args.kibana_spec:
        node_specs.append(
            {
                "Type": "Kibana",
                "Number": args.kibana_count,
                "ResourceSpecName": args.kibana_spec,
            }
        )

    instance_conf = {
        "InstanceName": args.name,
        "Version": args.version,
        "AdminPassword": args.admin_password,
        "ChargeType": args.charge_type,
        "DeletionProtection": args.deletion_protection,
        "EnableHttps": args.https,
        "EnablePureMaster": args.pure_master
        if args.pure_master is not None
        else (True if args.master_spec else False),
        "ProjectName": "default",
        "RegionId": region,
        "ZoneId": zone_id,
        "VPC": {"VpcId": args.vpc_id},
        "Subnet": {"SubnetId": args.subnet_id},
        "NodeSpecsAssigns": node_specs,
    }

    body = {"InstanceConfiguration": instance_conf}
    api_call(lambda: es_api.create_instance_in_one_step(body))


def cmd_scale(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    assign_kwargs = {
        "Type": args.node_type,
        "Number": args.count,
        "ResourceSpecName": args.spec_name,
    }
    if args.storage_spec_name:
        assign_kwargs["StorageSpecName"] = args.storage_spec_name
    if args.storage_size is not None:
        assign_kwargs["StorageSize"] = args.storage_size

    assign = dict(assign_kwargs)

    body = {
        "InstanceId": args.id,
        "NodeSpecsAssigns": [assign],
    }
    api_call(lambda: es_api.modify_node_spec_in_one_step(body))


def cmd_delete(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    if not getattr(args, "confirm", False):
        print_error(
            "Confirmation Required",
            "Refusing to delete without --confirm. Ask the user to explicitly confirm, then rerun with: control_tools.py delete --id <instance-id> --confirm",
        )
    body = {"InstanceId": args.id}
    api_call(lambda: es_api.release_instance(body))


def cmd_ip_allowlist_get(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {"InstanceId": args.id}
    api_call(lambda: es_api.describe_ip_allow_list(body))


def parse_json_array(raw: str, name: str) -> list:
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as e:
        print_error("Invalid JSON", f"Invalid JSON in {name}: {str(e)}")
    if not isinstance(data, list):
        print_error(
            "Invalid JSON", f"{name} must be a JSON array, e.g. '[\"1.2.3.4/32\"]'"
        )
    return data


def extract_supported_versions_from_node_specs(
    node_specs_dict: Dict[str, Any],
) -> List[str]:
    versions: Set[str] = set()

    def walk(obj: Any) -> None:
        if isinstance(obj, dict):
            for k, v in obj.items():
                if isinstance(v, str) and k.lower() in (
                    "version",
                    "versionid",
                    "version_id",
                ):
                    versions.add(v)
                walk(v)
        elif isinstance(obj, list):
            for item in obj:
                walk(item)

    walk(node_specs_dict)

    def sort_key(s: str) -> Tuple[int, str]:
        # Best-effort sorting: prefer known prefixes, then lexical.
        if s.startswith("V"):
            return (0, s)
        if s.startswith("OPEN_SEARCH_"):
            return (1, s)
        return (2, s)

    return sorted(versions, key=sort_key)


def cmd_versions(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    # ESCloud does not expose a dedicated "describe_versions" API in this SDK.
    # Derive versions from DescribeNodeAvailableSpecs output (best-effort), so results stay current.
    body = {}

    def call():
        return es_api.describe_node_available_specs(body)

    try:
        resp = call()
        resp_dict = resp.to_dict() if hasattr(resp, "to_dict") else {}
        versions = extract_supported_versions_from_node_specs(resp_dict)
        if not versions:
            print_result(
                {
                    "versions": [],
                    "note": "No versions found in DescribeNodeAvailableSpecs response. Use 'node_specs' to inspect raw output or consult the console.",
                }
            )
            return
        print_result({"versions": versions})
    except ApiError as e:
        print_error("API Error", str(e))
    except Exception as e:
        print_error("Unexpected Error", str(e))


def cmd_ip_allowlist_set(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    ips = parse_json_array(args.ips, "--ips")
    allowlist_type = (args.type or "PRIVATE_ES").strip().upper()
    group = {
        "Name": args.group_name,
        "AllowList": ",".join(ips),
    }
    allowlist = {
        "Groups": [group],
        "Type": allowlist_type,
        "AllowList": "",
    }
    body = {
        "InstanceId": args.id,
        "EsIpAllowList": allowlist,
    }
    api_call(lambda: es_api.modify_ip_allow_list_v2(body))


def cmd_public_network_set(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {
        "InstanceId": args.id,
    }
    if args.enable:
        if not args.eip_id:
            print_error(
                "Missing Parameters", "--eip-id is required when enabling public access"
            )
        body["EsEip"] = {"IsOpen": True, "EipId": args.eip_id}
        api_call(lambda: es_api.create_public_address(body))
    else:
        body["EsEip"] = {"IsOpen": False}
        api_call(lambda: es_api.release_public_address(body))


def cmd_reset_password(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {
        "InstanceId": args.id,
        "NewPassword": args.admin_password,
    }
    api_call(lambda: es_api.reset_admin_password(body))


def cmd_nodes(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {"InstanceId": args.id}
    api_call(lambda: es_api.describe_instance_nodes(body))


def cmd_plugins(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {"InstanceId": args.id}
    api_call(lambda: es_api.describe_instance_plugins(body))


def cmd_rename(args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str):
    body = {"InstanceId": args.id, "NewName": args.name}
    api_call(lambda: es_api.rename_instance(body))


def cmd_maintenance_set(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    # Keep a stable CLI surface even if upstream docs vary:
    # allow either JSON object {"MaintenanceDay":[...],"MaintenanceTime":"..."} or
    # explicit flags --day/--time.
    maintenance_day = args.day
    maintenance_time = args.time
    if args.setting:
        setting = json.loads(args.setting)
        if not isinstance(setting, dict):
            print_error("Invalid --setting", "--setting must be a JSON object")
        if "MaintenanceDay" in setting and not maintenance_day:
            maintenance_day = setting.get("MaintenanceDay")
        if "MaintenanceTime" in setting and not maintenance_time:
            maintenance_time = setting.get("MaintenanceTime")
        if "maintenance_day" in setting and not maintenance_day:
            maintenance_day = setting.get("maintenance_day")
        if "maintenance_time" in setting and not maintenance_time:
            maintenance_time = setting.get("maintenance_time")
    if not maintenance_time:
        print_error(
            "Missing Parameters",
            "Provide --time or --setting containing MaintenanceTime.",
        )
    if isinstance(maintenance_day, str) and maintenance_day:
        # Accept common user inputs (Mon/Tue, Monday/Tuesday, MONDAY/TUESDAY) and normalize to
        # full uppercase day names required by the API.
        day_map = {
            "MON": "MONDAY",
            "MONDAY": "MONDAY",
            "TUE": "TUESDAY",
            "TUES": "TUESDAY",
            "TUESDAY": "TUESDAY",
            "WED": "WEDNESDAY",
            "WEDNESDAY": "WEDNESDAY",
            "THU": "THURSDAY",
            "THUR": "THURSDAY",
            "THURS": "THURSDAY",
            "THURSDAY": "THURSDAY",
            "FRI": "FRIDAY",
            "FRIDAY": "FRIDAY",
            "SAT": "SATURDAY",
            "SATURDAY": "SATURDAY",
            "SUN": "SUNDAY",
            "SUNDAY": "SUNDAY",
        }
        raw_days = [d.strip().upper() for d in maintenance_day.split(",") if d.strip()]
        maintenance_day = [day_map.get(d, d) for d in raw_days]

    body = {
        "InstanceId": args.id,
        "MaintenanceDay": maintenance_day,
        "MaintenanceTime": maintenance_time,
    }
    api_call(lambda: es_api.modify_maintenance_setting(body))


def cmd_deletion_protection_set(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {
        "InstanceId": args.id,
        "DeletionProtection": args.enabled,
    }
    api_call(lambda: es_api.modify_deletion_protection(body))


def cmd_restart_node(
    args, es_api: EsCloudApi, vpc_api: VpcApi, configuration, region: str
):
    body = {
        "InstanceId": args.id,
        "NodeName": args.node_id,
        "Force": args.force,
    }
    api_call(lambda: es_api.restart_node(body))


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Volcano Engine ESCloud control tools CLI"
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    p = subparsers.add_parser("list", help="List ESCloud instances")
    p.add_argument("--page-number", type=int, default=1)
    p.add_argument("--page-size", type=int, default=10)
    p.set_defaults(func=cmd_list)

    p = subparsers.add_parser("detail", help="Get instance details")
    p.add_argument("--id", required=True, help="Instance ID")
    p.set_defaults(func=cmd_detail)

    p = subparsers.add_parser("create", help="Create a new instance (one step)")
    p.add_argument("--name", required=True, help="Instance name")
    p.add_argument(
        "--version",
        required=True,
        help="Version (e.g. V7_10, V8_18, OPEN_SEARCH_2_9, OPEN_SEARCH_3_3)",
    )
    p.add_argument("--vpc-id", required=True, help="VPC ID")
    p.add_argument("--subnet-id", required=True, help="Subnet ID (used to derive zone)")
    p.add_argument("--admin-password", required=True, help="Admin password")
    p.add_argument(
        "--charge-type",
        default="PostPaid",
        help="Charge type (PostPaid or PrePaid). Default: PostPaid",
    )
    p.add_argument(
        "--https", type=str_to_bool, default=True, help="Enable HTTPS (default: true)"
    )
    p.add_argument(
        "--deletion-protection",
        type=str_to_bool,
        default=True,
        help="Deletion protection (default: true)",
    )

    p.add_argument(
        "--master-spec",
        default="",
        help="Optional master node resource spec name. If provided, pure master nodes are enabled.",
    )
    p.add_argument(
        "--master-count", type=int, default=3, help="Master node count (default: 3)"
    )
    p.add_argument(
        "--master-storage-spec",
        default="",
        help="Optional master node storage spec name.",
    )
    p.add_argument(
        "--master-storage-size",
        type=int,
        default=None,
        help="Optional master node storage size in GiB.",
    )
    p.add_argument(
        "--pure-master",
        type=str_to_bool,
        default=None,
        help="Explicitly enable/disable pure master nodes.",
    )

    p.add_argument("--hot-spec", required=True, help="Hot node resource spec name")
    p.add_argument(
        "--hot-count", type=int, default=2, help="Hot node count (default: 2)"
    )
    p.add_argument(
        "--hot-storage-spec", required=True, help="Hot node storage spec name"
    )
    p.add_argument(
        "--hot-storage-size",
        type=int,
        required=True,
        help="Hot node storage size in GiB",
    )

    p.add_argument(
        "--kibana-spec", default="", help="Optional Kibana node resource spec name"
    )
    p.add_argument(
        "--kibana-count", type=int, default=1, help="Kibana node count (default: 1)"
    )
    p.set_defaults(func=cmd_create)

    p = subparsers.add_parser("scale", help="Scale node spec/count (one step)")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--node-type",
        required=True,
        help="Node type (Master, Hot, Warm, Cold, Coordinator, Kibana, Other)",
    )
    p.add_argument("--spec-name", required=True, help="Target resource spec name")
    p.add_argument("--count", type=int, required=True, help="Target node count")
    p.add_argument(
        "--storage-spec-name", default="", help="Optional target storage spec name"
    )
    p.add_argument(
        "--storage-size",
        type=int,
        default=None,
        help="Optional target storage size in GiB",
    )
    p.set_defaults(func=cmd_scale)

    p = subparsers.add_parser("delete", help="Delete an instance")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--confirm", action="store_true", help="Required safety flag for deletion"
    )
    p.set_defaults(func=cmd_delete)

    p = subparsers.add_parser("vpc", help="List VPCs")
    p.set_defaults(func=cmd_vpc)

    p = subparsers.add_parser("subnet", help="List subnets in a VPC")
    p.add_argument("--vpc-id", required=True, help="VPC ID")
    p.set_defaults(func=cmd_subnet)

    p = subparsers.add_parser("zones", help="List available zones")
    p.set_defaults(func=cmd_zones)

    p = subparsers.add_parser("node_specs", help="List node available specs")
    p.set_defaults(func=cmd_node_specs)

    p = subparsers.add_parser(
        "versions",
        help="Best-effort list of supported versions (derived from node_specs)",
    )
    p.set_defaults(func=cmd_versions)

    p = subparsers.add_parser("ip_allowlist_get", help="Get IP allowlist")
    p.add_argument("--id", required=True, help="Instance ID")
    p.set_defaults(func=cmd_ip_allowlist_get)

    p = subparsers.add_parser("ip_allowlist_set", help="Set IP allowlist")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--group-name", required=True, help="Group name")
    p.add_argument("--ips", required=True, help="JSON array of CIDRs/IPs")
    p.add_argument(
        "--type",
        default="PRIVATE_ES",
        help="Allowlist type (default: PRIVATE_ES). Use PUBLIC_ES when managing a public endpoint allowlist, if supported by your instance.",
    )
    p.set_defaults(func=cmd_ip_allowlist_set)

    p = subparsers.add_parser("reset_password", help="Reset admin password")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--admin-password", required=True, help="New admin password")
    p.set_defaults(func=cmd_reset_password)

    p = subparsers.add_parser("nodes", help="Describe nodes in an instance")
    p.add_argument("--id", required=True, help="Instance ID")
    p.set_defaults(func=cmd_nodes)

    p = subparsers.add_parser("plugins", help="Describe installed plugins")
    p.add_argument("--id", required=True, help="Instance ID")
    p.set_defaults(func=cmd_plugins)

    p = subparsers.add_parser("rename", help="Rename instance display name")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--name", required=True, help="New instance name")
    p.set_defaults(func=cmd_rename)

    p = subparsers.add_parser(
        "maintenance_set", help="Set maintenance window/setting (JSON pass-through)"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--day", default=None, help="Maintenance days (comma-separated) e.g. Mon,Tue"
    )
    p.add_argument(
        "--time",
        default="",
        help="Maintenance time window (required if --setting omitted)",
    )
    p.add_argument(
        "--setting",
        default="",
        help="Optional JSON object containing MaintenanceDay/MaintenanceTime",
    )
    p.set_defaults(func=cmd_maintenance_set)

    p = subparsers.add_parser(
        "deletion_protection_set", help="Enable/disable deletion protection"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--enabled", type=str_to_bool, required=True, help="true or false")
    p.set_defaults(func=cmd_deletion_protection_set)

    p = subparsers.add_parser("restart_node", help="Restart a specific node")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--node-id", required=True, help="Node name")
    p.add_argument(
        "--force",
        type=str_to_bool,
        default=False,
        help="Force restart (default: false)",
    )
    p.set_defaults(func=cmd_restart_node)

    p = subparsers.add_parser(
        "public_network", help="Enable/disable public network access (EIP)"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--enable",
        type=str_to_bool,
        default=True,
        help="Enable or disable public access (default: true)",
    )
    p.add_argument(
        "--eip-id", default="", help="EIP allocation ID (required when enabling)"
    )

    p.set_defaults(func=cmd_public_network_set)

    p = subparsers.add_parser("eip_list", help="List EIP addresses")
    p.add_argument(
        "--status",
        choices=["Available", "Attaching", "Attached", "Detaching", "Releasing"],
        help="Optional filter by status",
    )
    p.set_defaults(func=cmd_eip_list)

    p = subparsers.add_parser("eip_allocate", help="Allocate (create) a new EIP")
    p.add_argument(
        "--bandwidth", type=int, default=1, help="Bandwidth in Mbps (default: 1)"
    )
    p.add_argument(
        "--billing-type",
        default="PostPaidByTraffic",
        help="Billing type (PostPaidByBandwidth, PostPaidByTraffic, PrePaid). Default: PostPaidByTraffic",
    )
    p.add_argument("--name", default="", help="Optional EIP name")
    p.add_argument("--description", default="", help="Optional EIP description")
    p.add_argument("--isp", default="BGP", help="ISP (BGP, etc.). Default: BGP")
    p.set_defaults(func=cmd_eip_allocate)

    p = subparsers.add_parser("eip_release", help="Release (delete) an EIP")
    p.add_argument("--allocation-id", required=True, help="EIP Allocation ID")
    p.set_defaults(func=cmd_eip_release)

    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()
    es_api, vpc_api, configuration, region = _get_clients()
    args.func(args, es_api, vpc_api, configuration, region)


if __name__ == "__main__":
    main()

# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import json
import sys
import time
from typing import Any, Callable, Tuple
from api import ApiError, EsCloudApi, VpcApi

# EIP Billing Types mapping (human strings to API integers)
EIP_BILLING_TYPES = {
    "PrePaid": 1,
    "PostPaid": 2,  # Postpaid by Bandwidth
    "PostPaidByTraffic": 3,  # Postpaid by Traffic (Recommended)
}


class WorkflowError(Exception):
    def __init__(
        self,
        error: str,
        details: str = "",
        data: dict[str, Any] | None = None,
        steps_completed: list[str] | None = None,
    ):
        super().__init__(details or error)
        self.error = error
        self.details = details
        self.data = data or {}
        self.steps_completed = list(steps_completed or [])


def emit(payload: dict[str, Any], exit_code: int = 0) -> None:
    print(json.dumps(payload, default=str, ensure_ascii=False))
    sys.exit(exit_code)


def first_present(obj: Any, keys: list[str]) -> Any:
    if isinstance(obj, dict):
        for key in keys:
            if key in obj and obj[key] not in (None, ""):
                return obj[key]
        for value in obj.values():
            found = first_present(value, keys)
            if found not in (None, ""):
                return found
    elif isinstance(obj, list):
        for item in obj:
            found = first_present(item, keys)
            if found not in (None, ""):
                return found
    return None


def get_list(obj: dict[str, Any], *keys: str) -> list[Any]:
    for key in keys:
        value = obj.get(key)
        if isinstance(value, list):
            return value
    result = obj.get("Result") or obj.get("result")
    if isinstance(result, dict):
        for key in keys:
            value = result.get(key)
            if isinstance(value, list):
                return value
    return []


def normalize_detail(detail: dict[str, Any]) -> dict[str, Any]:
    result = detail.get("Result") or detail.get("result") or detail
    inst_id = first_present(result, ["InstanceId", "instance_id", "Id", "id"])

    inst_conf = result.get("InstanceConfiguration", {})
    name = first_present(inst_conf, ["InstanceName", "instance_name"])

    status = first_present(result, ["Status", "status"])
    return {
        "id": inst_id,
        "name": name,
        "status": status,
        "raw": detail,
    }


def str_to_bool(v: str) -> bool:
    if isinstance(v, bool):
        return v
    if v.lower() in ("true", "1", "yes", "y"):
        return True
    if v.lower() in ("false", "0", "no", "n"):
        return False
    raise argparse.ArgumentTypeError(f"Boolean value expected, got '{v}'")


def add_eip_alloc_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--eip-id", default="", help="Existing EIP allocation id")
    parser.add_argument(
        "--eip-bandwidth",
        type=int,
        default=10,
        help="EIP bandwidth for auto-allocation",
    )
    parser.add_argument(
        "--eip-billing-type",
        type=str,
        default="PostPaidByTraffic",
        choices=list(EIP_BILLING_TYPES.keys()),
        help=f"EIP billing type for auto-allocation (Allowed: {', '.join(EIP_BILLING_TYPES.keys())})",
    )
    parser.add_argument("--eip-isp", type=str, default="BGP", help="Optional EIP ISP")
    parser.add_argument(
        "--eip-auto-reuse",
        type=str_to_bool,
        default=True,
        help="If true, reuse an existing Available EIP before allocating a new one",
    )


def _get_clients() -> Tuple[EsCloudApi, VpcApi, Any, str]:
    import ark_shim

    if ark_shim.check_is_ark_env():
        return ark_shim.get_clients()
    import sdk_shim

    return sdk_shim.get_clients()


class ControlPlane:
    def __init__(self, debug: bool = False):
        self.es_api, self.vpc_api, self.configuration, self.region = _get_clients()
        self.debug = debug

    def _run_api(
        self, fn: Callable[[], Any], context_data: dict[str, Any] | None = None
    ) -> Any:
        try:
            resp = fn()

            # Normalize to dict for inspection
            data = resp
            try:
                if hasattr(resp, "to_dict"):
                    data = resp.to_dict()
            except Exception:
                pass

            # check for nested errors even if no exception was raised
            if isinstance(data, dict):
                error_msg = self._extract_error(data)
                if error_msg:
                    if not self.debug:
                        # Strip duplicated Error blocks from the raw response to keep output concise
                        if "Result" in data and isinstance(data["Result"], dict):
                            data["Result"].pop("Error", None)
                        if "ResponseMetadata" in data and isinstance(
                            data["ResponseMetadata"], dict
                        ):
                            data["ResponseMetadata"].pop("Error", None)
                    raise WorkflowError("API Error", error_msg, {"response": data})

            return data
        except WorkflowError:
            raise
        except ApiError as exc:
            msg = str(exc)
            if not self.debug:
                # Try to extract a more precise message from the exception string if it contains JSON
                extracted = self._extract_error(msg)
                msg = extracted if extracted else msg

            instruction = ""
            if "TaskIsRunning" in msg:
                instruction = (
                    "An operation is already in progress for this instance. "
                    "Wait for the instance to return to Running and retry."
                )
            elif any(
                key in msg
                for key in [
                    "BadRequest",
                    "InvalidParameter",
                    "NotFound",
                    "Unauthorized",
                    "InvalidAction",
                ]
            ):
                instruction = (
                    "Verify that all IDs and parameters are valid and that the current credentials "
                    "have access to the target VPC, subnet, and instance."
                )
            details = f"{msg}\n\nInstruction: {instruction}" if instruction else msg
            error_data = context_data if self.debug else None
            raise WorkflowError("API Error", details, error_data)
        except Exception as exc:
            error_data = context_data if self.debug else None
            raise WorkflowError(
                "Unexpected Error",
                f"{exc}\n\nInstruction: Internal script error. Check credentials, network, and parameters.",
                error_data,
            )

    def _extract_error(self, source: Any) -> str | None:
        """Parses API response or error string for a precise root cause message."""
        if not source:
            return None

        data = source
        if isinstance(source, str):
            try:
                # Some ApiError messages contain the JSON response
                start = source.find("{")
                end = source.rfind("}")
                if start != -1 and end != -1:
                    data = json.loads(source[start : end + 1])
                else:
                    return None
            except Exception:
                return None

        if not isinstance(data, dict):
            return None

        # Target locations for Volcengine ESCloud errors
        # 1. Result.Error
        # 2. ResponseMetadata.Error
        res_err = (
            data.get("Result", {}).get("Error", {})
            if isinstance(data.get("Result"), dict)
            else {}
        )
        meta_err = (
            data.get("ResponseMetadata", {}).get("Error", {})
            if isinstance(data.get("ResponseMetadata"), dict)
            else {}
        )

        raw_msg = res_err.get("Message") or meta_err.get("Message")
        if not raw_msg:
            return None

        # Keep original message without cropping details
        return raw_msg

    def _success(
        self, goal: str, data: dict[str, Any], steps_completed: list[str]
    ) -> dict[str, Any]:
        return {
            "status": "success",
            "goal": goal,
            "data": data,
            "steps_completed": steps_completed,
        }

    def _error(
        self,
        goal: str,
        error: str,
        details: str,
        steps_completed: list[str],
        data: dict[str, Any] | None = None,
        status: str = "error",
    ) -> dict[str, Any]:
        error_label = error
        details_label = details
        if error_label in ("API Error", "Unexpected Error") and details_label:
            error_label = details_label
            details_label = ""

        payload = {
            "status": status,
            "goal": goal,
            "error": error_label,
        }
        if details_label:
            payload["details"] = details_label

        payload["data"] = data or {}
        payload["steps_completed"] = steps_completed
        return payload

    def _fetch_detail(self, instance_id: str) -> dict[str, Any]:
        body = {"InstanceId": instance_id}
        return self._run_api(
            lambda: self.es_api.describe_instance(body), {"instance_id": instance_id}
        )

    def _wait_for_condition(
        self,
        fetch_fn: Callable[[], dict[str, Any]],
        condition_fn: Callable[[dict[str, Any]], bool],
        poll_interval: int,
        timeout: int,
        consecutive_successes: int = 1,
    ) -> tuple[bool, dict[str, Any]]:
        deadline = time.time() + timeout
        last_detail: dict[str, Any] = {}
        success_count = 0
        while True:
            last_detail = fetch_fn()

            if condition_fn(last_detail):
                success_count += 1
                if success_count >= consecutive_successes:
                    return True, last_detail
            else:
                success_count = 0
            if time.time() >= deadline:
                return False, last_detail
            time.sleep(poll_interval)

    def detail(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        detail = self._fetch_detail(args.id)
        steps.append("fetch_detail")
        normalized = normalize_detail(detail)
        return self._success(
            "detail",
            {
                "instance_id": normalized["id"],
                "instance_name": normalized["name"],
                "status": normalized["status"],
                "detail": detail,
            },
            steps,
        )

    def list(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        body = {"PageNumber": args.page_number, "PageSize": args.page_size}
        instances = self._run_api(
            lambda: self.es_api.describe_instances(body), {"request": body}
        )
        steps.append("fetch_instances")
        return self._success("list", instances, steps)

    def provision_info(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        vpcs_resp = self._run_api(lambda: self.vpc_api.describe_vpcs({}))
        steps.append("fetch_vpcs")

        specs_resp = self._run_api(
            lambda: self.es_api.describe_node_available_specs({})
        )
        steps.append("fetch_specs")

        subnets_by_vpc: dict[str, Any] = {}
        for vpc in get_list(vpcs_resp, "Vpcs", "vpcs"):
            vpc_id = vpc.get("VpcId") or vpc.get("vpc_id")
            if not vpc_id:
                continue
            subnet_resp = self._run_api(
                lambda: self.vpc_api.describe_subnets({"VpcId": vpc_id}),
                {"vpc_id": vpc_id},
            )
            subnets_by_vpc[vpc_id] = subnet_resp
        steps.append("fetch_subnets")

        zones_resp = self._run_api(lambda: self.es_api.describe_zones({}))
        steps.append("fetch_zones")

        return self._success(
            "provision-info",
            {
                "region": self.region,
                "vpcs": vpcs_resp,
                "subnets_by_vpc": subnets_by_vpc,
                "specs": specs_resp,
                "zones": zones_resp,
            },
            steps,
        )

    def provision(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        subnet_resp = self._run_api(
            lambda: self.vpc_api.describe_subnets({"SubnetIds": [args.subnet_id]}),
            {"subnet_id": args.subnet_id},
        )
        subnets = get_list(subnet_resp, "Subnets", "subnets")
        if not subnets:
            raise WorkflowError(
                "Subnet Not Found",
                f"Subnet ID '{args.subnet_id}' does not exist or is not in the current region.",
                steps_completed=steps,
            )
        zone_id = subnets[0].get("ZoneId") or subnets[0].get("zone_id")
        if not zone_id:
            raise WorkflowError(
                "Zone Not Found",
                "Failed to derive zone_id from the subnet.",
                steps_completed=steps,
            )
        steps.append("validate_subnet")

        node_specs = []
        if args.master_spec:
            node_specs.append(
                {
                    "Type": "Master",
                    "Number": args.master_count,
                    "ResourceSpecName": args.master_spec,
                    "StorageSpecName": args.master_storage_spec
                    or args.hot_storage_spec,
                    "StorageSize": args.master_storage_size or 20,
                }
            )

        node_specs.append(
            {
                "Type": "Hot",
                "Number": args.hot_count,
                "ResourceSpecName": args.hot_spec,
                "StorageSpecName": args.hot_storage_spec,
                "StorageSize": args.hot_storage_size,
            }
        )

        if getattr(args, "kibana_spec", ""):
            node_specs.append(
                {
                    "Type": "Kibana",
                    "Number": getattr(args, "kibana_count", 1),
                    "ResourceSpecName": getattr(args, "kibana_spec", ""),
                }
            )

        instance_conf = {
            "InstanceName": args.name,
            "Version": args.version,
            "AdminPassword": args.admin_password,
            "ChargeType": args.charge_type,
            "DeletionProtection": True,
            "EnableHttps": True,
            "EnablePureMaster": True if args.master_spec else False,
            "ProjectName": "default",
            "RegionId": self.region,
            "ZoneId": zone_id,
            "VPC": {"VpcId": args.vpc_id},
            "Subnet": {"SubnetId": args.subnet_id},
            "NodeSpecsAssigns": node_specs,
        }

        body = {"InstanceConfiguration": instance_conf}
        create_resp = self._run_api(
            lambda: self.es_api.create_instance_in_one_step(body), {"request": body}
        )
        steps.append("create_instance")

        instance_id = first_present(create_resp, ["InstanceId", "instance_id", "Id"])
        if not instance_id:
            # This should normally be caught by _run_api now, but keeping a fallback
            raise WorkflowError(
                "API Error",
                "CreateInstanceInOneStep returned no instance identifier.",
                {"create_response": create_resp},
                steps_completed=steps,
            )

        ok, detail = self._wait_for_condition(
            fetch_fn=lambda: self._fetch_detail(instance_id),
            condition_fn=lambda d: normalize_detail(d)["status"] == "Running",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
            consecutive_successes=2,
        )
        steps.append("poll_status")

        data = {
            "instance_id": instance_id,
            "create_response": create_resp,
            "final_detail": detail,
        }
        if ok:
            return self._success("provision", data, steps)
        return self._error(
            "provision",
            "Timeout",
            f"Instance '{instance_id}' did not reach Running within {args.timeout} seconds.",
            steps,
            data,
            status="timeout",
        )

    def deprovision(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        if args.confirm != args.id:
            raise WorkflowError(
                "Confirmation required",
                f"--confirm must exactly match the instance id {args.id!r}.",
                steps_completed=steps,
            )

        detail = self._fetch_detail(args.id)
        steps.append("fetch_detail")

        inst_info = detail.get("Result", detail).get("InstanceInfo", {})
        if inst_info.get("DeletionProtection"):
            if not getattr(args, "force", False):
                raise WorkflowError(
                    "Deletion Protection Enabled",
                    "Instance has deletion protection. Use --force to disable it and delete.",
                    steps_completed=steps,
                )
            self._run_api(
                lambda: self.es_api.modify_deletion_protection(
                    {"InstanceId": args.id, "DeletionProtection": False}
                ),
                {"instance_id": args.id},
            )
            steps.append("disable_deletion_protection")

        release_resp = self._run_api(
            lambda: self.es_api.release_instance({"InstanceId": args.id}),
            {"instance_id": args.id},
        )
        steps.append("release_instance")

        def _fetch_safe():
            try:
                return self._fetch_detail(args.id)
            except WorkflowError as e:
                if (
                    "NotFound" in str(e)
                    or "not exist" in str(e).lower()
                    or "InvalidParameter" in str(e)
                ):
                    return {"is_deleted": True}
                raise e

        ok, detail_after = self._wait_for_condition(
            fetch_fn=_fetch_safe,
            condition_fn=lambda d: d.get("is_deleted")
            or normalize_detail(d).get("status") == "Deleted",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
        )
        steps.append("poll_deletion")

        data = {
            "instance_id": args.id,
            "release_response": release_resp,
            "final_detail": detail_after,
        }
        if ok:
            return self._success("deprovision", data, steps)
        return self._error(
            "deprovision", "Timeout", "Instance was not deleted in time.", steps, data
        )

    def scale(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        current_detail = self._fetch_detail(args.id)
        steps.append("fetch_detail")

        current_status = normalize_detail(current_detail).get("status")
        if current_status != "Running":
            raise WorkflowError(
                "Invalid Instance State",
                f"Instance '{args.id}' must be Running before scaling. Current status: {current_status!r}.",
                {"detail": current_detail},
                steps_completed=steps,
            )
        steps.append("validate_running")

        assign_kwargs = {
            "Type": args.node_type,
            "Number": args.count,
            "ResourceSpecName": args.spec_name,
        }
        if args.storage_spec_name:
            assign_kwargs["StorageSpecName"] = args.storage_spec_name
        if args.storage_size is not None:
            assign_kwargs["StorageSize"] = args.storage_size

        body = {"InstanceId": args.id, "NodeSpecsAssigns": [assign_kwargs]}
        scale_resp = self._run_api(
            lambda: self.es_api.modify_node_spec_in_one_step(body), {"request": body}
        )
        steps.append("scale_instance")

        ok, detail = self._wait_for_condition(
            fetch_fn=lambda: self._fetch_detail(args.id),
            condition_fn=lambda d: normalize_detail(d).get("status") == "Running",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
            consecutive_successes=2,
        )
        steps.append("poll_status")

        data = {
            "instance_id": args.id,
            "scale_response": scale_resp,
            "final_detail": detail,
        }
        if ok:
            return self._success("scale", data, steps)
        return self._error(
            "scale",
            "Timeout",
            f"Instance '{args.id}' did not return to Running within {args.timeout} seconds.",
            steps,
            data,
            status="timeout",
        )

    def public_access(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        detail_before = self._fetch_detail(args.id)
        steps.append("fetch_detail")

        current_status = normalize_detail(detail_before).get("status")
        if current_status != "Running":
            raise WorkflowError(
                "Invalid Instance State",
                f"Instance '{args.id}' must be Running before managing public endpoint.",
                {"detail": detail_before},
                steps_completed=steps,
            )

        body = {
            "InstanceId": args.id,
        }

        if args.enable:
            eip_id = self._ensure_eip(args)
            body["EsEip"] = {"IsOpen": True, "EipId": eip_id}
            steps.append("ensure_eip")
            modify_resp = self._run_api(
                lambda: self.es_api.create_public_address(body), {"request": body}
            )
            steps.append("create_public_address")
        else:
            body["EsEip"] = {"IsOpen": False}
            modify_resp = self._run_api(
                lambda: self.es_api.release_public_address(body), {"request": body}
            )
            steps.append("release_public_address")

        ok, detail_after = self._wait_for_condition(
            fetch_fn=lambda: self._fetch_detail(args.id),
            condition_fn=lambda d: normalize_detail(d).get("status") == "Running",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
            consecutive_successes=2,
        )
        steps.append("poll_endpoint")

        data = {
            "instance_id": args.id,
            "enable": args.enable,
            "modify_response": modify_resp,
            "detail_before": detail_before,
            "detail_after": detail_after,
        }
        if ok:
            return self._success("public-access", data, steps)
        return self._error(
            "public-access",
            "Timeout",
            "Endpoint change did not complete in time.",
            steps,
            data,
            status="timeout",
        )

    def allowlist(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        detail_before = self._fetch_detail(args.id)
        steps.append("fetch_detail")

        current_status = normalize_detail(detail_before).get("status")
        if current_status != "Running":
            raise WorkflowError(
                "Invalid Instance State",
                f"Instance '{args.id}' must be Running before modifying allowlist.",
                {"detail": detail_before},
                steps_completed=steps,
            )

        ips_list = args.ips.split(",") if args.ips else []
        ips_list = [ip.strip() for ip in ips_list if ip.strip()]

        allowlist_type = (args.type or "PRIVATE_ES").strip().upper()
        group = {
            "Name": args.group_name,
            "AllowList": ",".join(ips_list),
        }
        allowlist_conf = {
            "Groups": [group],
            "Type": allowlist_type,
            "AllowList": "",
        }
        body = {
            "InstanceId": args.id,
            "EsIpAllowList": allowlist_conf,
        }

        modify_resp = self._run_api(
            lambda: self.es_api.modify_ip_allow_list_v2(body), {"request": body}
        )
        steps.append("modify_allowlist")

        ok, detail_after = self._wait_for_condition(
            fetch_fn=lambda: self._fetch_detail(args.id),
            condition_fn=lambda d: normalize_detail(d).get("status") == "Running",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
            consecutive_successes=2,
        )
        steps.append("poll_status")

        data = {
            "instance_id": args.id,
            "modify_response": modify_resp,
            "detail_after": detail_after,
        }
        if ok:
            return self._success("allowlist", data, steps)
        return self._error(
            "allowlist",
            "Timeout",
            "Allowlist change did not complete in time.",
            steps,
            data,
            status="timeout",
        )

    def reset_password(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        detail_before = self._fetch_detail(args.id)
        steps.append("fetch_detail")

        current_status = normalize_detail(detail_before).get("status")
        if current_status != "Running":
            raise WorkflowError(
                "Invalid Instance State",
                f"Instance '{args.id}' must be Running before resetting password.",
                {"detail": detail_before},
                steps_completed=steps,
            )

        body = {
            "InstanceId": args.id,
            "NewPassword": args.admin_password,
        }

        modify_resp = self._run_api(
            lambda: self.es_api.reset_admin_password(body), {"request": body}
        )
        steps.append("reset_password")

        ok, detail_after = self._wait_for_condition(
            fetch_fn=lambda: self._fetch_detail(args.id),
            condition_fn=lambda d: normalize_detail(d).get("status") == "Running",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
            consecutive_successes=2,
        )
        steps.append("poll_status")

        data = {
            "instance_id": args.id,
            "modify_response": modify_resp,
            "detail_after": detail_after,
        }
        if ok:
            return self._success("reset-password", data, steps)
        return self._error(
            "reset-password",
            "Timeout",
            "Password reset did not complete in time.",
            steps,
            data,
            status="timeout",
        )

    def _ensure_eip(self, args: argparse.Namespace) -> str:
        if args.eip_id:
            return args.eip_id

        if args.eip_auto_reuse:
            try:
                resp = self._run_api(
                    lambda: self.vpc_api.describe_eip_addresses({"Status": "Available"})
                )
                eips = get_list(resp, "EipAddresses", "eip_addresses")
                for eip in eips:
                    alloc_id = eip.get("AllocationId") or eip.get("allocation_id")
                    if alloc_id and eip.get("Status") == "Available":
                        return alloc_id
            except Exception:
                pass

        # Allocate new EIP
        billing_type = EIP_BILLING_TYPES.get(args.eip_billing_type, 3)
        body = {
            "Bandwidth": args.eip_bandwidth,
            "BillingType": billing_type,
            "ISP": args.eip_isp or "BGP",
        }
        alloc_resp = self._run_api(
            lambda: self.vpc_api.allocate_eip_address(body), {"request": body}
        )
        eip_id = first_present(alloc_resp, ["AllocationId", "allocation_id"])
        if not eip_id:
            raise WorkflowError(
                "API Error",
                f"AllocateEipAddress returned no AllocationId: {alloc_resp!r}",
            )
        return eip_id

    def maintenance(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []

        maintenance_day = args.day
        maintenance_time = args.time
        if not maintenance_time:
            raise WorkflowError(
                "Missing Parameters", "Provide --time.", steps_completed=steps
            )

        day_map = {
            "MON": "MONDAY",
            "MONDAY": "MONDAY",
            "TUE": "TUESDAY",
            "TUES": "TUESDAY",
            "TUESDAY": "TUESDAY",
            "WED": "WEDNESDAY",
            "WEDNESDAY": "WEDNESDAY",
            "THU": "THURSDAY",
            "THUR": "THURSDAY",
            "THURS": "THURSDAY",
            "THURSDAY": "THURSDAY",
            "FRI": "FRIDAY",
            "FRIDAY": "FRIDAY",
            "SAT": "SATURDAY",
            "SATURDAY": "SATURDAY",
            "SUN": "SUNDAY",
            "SUNDAY": "SUNDAY",
        }
        raw_days = [d.strip().upper() for d in maintenance_day.split(",") if d.strip()]
        maintenance_day_parsed = [day_map.get(d, d) for d in raw_days]

        body = {
            "InstanceId": args.id,
            "MaintenanceDay": maintenance_day_parsed,
            "MaintenanceTime": maintenance_time,
        }
        modify_resp = self._run_api(
            lambda: self.es_api.modify_maintenance_setting(body), {"request": body}
        )
        steps.append("modify_maintenance")

        data = {"instance_id": args.id, "modify_response": modify_resp}
        return self._success("maintenance", data, steps)

    def rename(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        body = {"InstanceId": args.id, "NewName": args.name}
        modify_resp = self._run_api(
            lambda: self.es_api.rename_instance(body), {"request": body}
        )
        steps.append("rename_instance")

        data = {"instance_id": args.id, "modify_response": modify_resp}
        return self._success("rename", data, steps)

    def restart_node(self, args: argparse.Namespace) -> dict[str, Any]:
        steps: list[str] = []
        detail_before = self._fetch_detail(args.id)
        steps.append("fetch_detail")

        current_status = normalize_detail(detail_before).get("status")
        if current_status != "Running":
            raise WorkflowError(
                "Invalid Instance State",
                f"Instance '{args.id}' must be Running before restarting nodes.",
                {"detail": detail_before},
                steps_completed=steps,
            )

        body = {
            "InstanceId": args.id,
            "NodeName": args.node_id,
            "Force": args.force,
        }
        modify_resp = self._run_api(
            lambda: self.es_api.restart_node(body), {"request": body}
        )
        steps.append("restart_node")

        ok, detail_after = self._wait_for_condition(
            fetch_fn=lambda: self._fetch_detail(args.id),
            condition_fn=lambda d: normalize_detail(d).get("status") == "Running",
            poll_interval=args.poll_interval,
            timeout=args.timeout,
            consecutive_successes=2,
        )
        steps.append("poll_status")

        data = {
            "instance_id": args.id,
            "modify_response": modify_resp,
            "detail_after": detail_after,
        }
        if ok:
            return self._success("restart-node", data, steps)
        return self._error(
            "restart-node",
            "Timeout",
            "Restart did not complete in time.",
            steps,
            data,
            status="timeout",
        )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Volcano Engine ESCloud goal-based control plane CLI"
    )
    parser.add_argument(
        "--debug", action="store_true", help="Include context data in error payloads"
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    p = subparsers.add_parser(
        "provision-info",
        help="Collect necessary info for provisioning (VPCs, subnets, specs, zones)",
    )
    p.set_defaults(goal="provision-info")

    p = subparsers.add_parser("list", help="List ESCloud instances")
    p.add_argument("--page-number", type=int, default=1)
    p.add_argument("--page-size", type=int, default=10)
    p.set_defaults(goal="list")

    p = subparsers.add_parser(
        "provision", help="Create an ESCloud instance and wait until Running"
    )
    p.add_argument("--name", required=True, help="Instance name")
    p.add_argument("--vpc-id", required=True, help="VPC ID")
    p.add_argument("--subnet-id", required=True, help="Subnet ID")
    p.add_argument("--admin-password", required=True, help="Admin password")
    p.add_argument("--version", required=True, help="ESCloud version (e.g. V7_10)")
    p.add_argument(
        "--charge-type", default="PostPaid", help="Charge type (PostPaid or PrePaid)"
    )

    p.add_argument("--master-spec", default="", help="Master node resource spec name")
    p.add_argument("--master-count", type=int, default=3, help="Master node count")
    p.add_argument(
        "--master-storage-spec", default="", help="Master node storage spec name"
    )
    p.add_argument(
        "--master-storage-size",
        type=int,
        default=None,
        help="Master node storage size in GiB",
    )

    p.add_argument("--hot-spec", required=True, help="Hot node resource spec name")
    p.add_argument("--hot-count", type=int, default=2, help="Hot node count")
    p.add_argument(
        "--hot-storage-spec", required=True, help="Hot node storage spec name"
    )
    p.add_argument(
        "--hot-storage-size",
        type=int,
        required=True,
        help="Hot node storage size in GiB",
    )

    p.add_argument("--kibana-spec", default="", help="Kibana node resource spec name")
    p.add_argument("--kibana-count", type=int, default=1, help="Kibana node count")

    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="provision")

    p = subparsers.add_parser(
        "deprovision", help="Delete an ESCloud instance and wait until destroyed"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--confirm", required=True, help="Must exactly match the instance ID to confirm"
    )
    p.add_argument(
        "--force",
        action="store_true",
        help="Force deletion by disabling deletion protection if enabled",
    )
    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="deprovision")

    p = subparsers.add_parser(
        "detail", help="Fetch status and raw details for an ESCloud instance"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.set_defaults(goal="detail")

    p = subparsers.add_parser(
        "scale", help="Scale ESCloud node spec or count and wait until Running"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--node-type", required=True, help="Node type (Master, Hot, Kibana, etc.)"
    )
    p.add_argument("--spec-name", required=True, help="Target resource spec name")
    p.add_argument("--count", type=int, required=True, help="Target node count")
    p.add_argument(
        "--storage-spec-name", default="", help="Optional target storage spec name"
    )
    p.add_argument(
        "--storage-size",
        type=int,
        default=None,
        help="Optional target storage size in GiB",
    )
    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="scale")

    p = subparsers.add_parser(
        "public-access", help="Toggle public access for ESCloud endpoint"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--enable", type=str_to_bool, required=True, help="true/false")
    add_eip_alloc_args(p)
    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="public-access")

    p = subparsers.add_parser(
        "allowlist", help="Set IP allowlist and wait until Running"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--group-name", default="default", help="Group name")
    p.add_argument("--ips", required=True, help="Comma-separated CIDRs/IPs")
    p.add_argument(
        "--type",
        default="PRIVATE_ES",
        help="Allowlist type (e.g. PRIVATE_ES, PUBLIC_ES)",
    )
    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="allowlist")

    p = subparsers.add_parser(
        "reset-password", help="Reset admin password and wait until Running"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--admin-password", required=True, help="New admin password")
    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="reset-password")

    p = subparsers.add_parser("maintenance", help="Configure maintenance window")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument(
        "--day", required=True, help="Maintenance days (comma-separated, e.g. Mon,Tue)"
    )
    p.add_argument(
        "--time", required=True, help="Maintenance time window (e.g. 02:00-06:00)"
    )
    p.set_defaults(goal="maintenance")

    p = subparsers.add_parser("rename", help="Rename instance display name")
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--name", required=True, help="New instance name")
    p.set_defaults(goal="rename")

    p = subparsers.add_parser(
        "restart-node", help="Restart a specific node and wait until Running"
    )
    p.add_argument("--id", required=True, help="Instance ID")
    p.add_argument("--node-id", required=True, help="Node name to restart")
    p.add_argument("--force", action="store_true", help="Force restart")
    p.add_argument(
        "--poll-interval", type=int, default=5, help="Polling interval in seconds"
    )
    p.add_argument(
        "--timeout", type=int, default=600, help="Polling timeout in seconds"
    )
    p.set_defaults(goal="restart-node")

    return parser


def dispatch(control_plane: ControlPlane, args: argparse.Namespace) -> dict[str, Any]:
    goal = args.goal
    if goal == "provision-info":
        return control_plane.provision_info(args)
    elif goal == "list":
        return control_plane.list(args)
    elif goal == "provision":
        return control_plane.provision(args)
    elif goal == "deprovision":
        return control_plane.deprovision(args)
    elif goal == "detail":
        return control_plane.detail(args)
    elif goal == "scale":
        return control_plane.scale(args)
    elif goal == "public-access":
        return control_plane.public_access(args)
    elif goal == "allowlist":
        return control_plane.allowlist(args)
    elif goal == "reset-password":
        return control_plane.reset_password(args)
    elif goal == "maintenance":
        return control_plane.maintenance(args)
    elif goal == "rename":
        return control_plane.rename(args)
    elif goal == "restart-node":
        return control_plane.restart_node(args)
    raise WorkflowError("Invalid Command", f"Unsupported goal {args.goal!r}.")


def main() -> None:
    try:
        parser = build_parser()
        args = parser.parse_args()

        control_plane = ControlPlane(debug=args.debug)

        resp = dispatch(control_plane, args)
        emit(resp)
    except WorkflowError as exc:
        error_label = exc.error
        details_label = exc.details or str(exc)
        if error_label in ("API Error", "Unexpected Error") and details_label:
            error_label = details_label
            details_label = ""

        payload = {
            "status": "error",
            "goal": getattr(args, "goal", ""),
            "error": error_label,
        }
        if details_label:
            payload["details"] = details_label

        payload["data"] = exc.data
        payload["steps_completed"] = exc.steps_completed
        emit(payload, exit_code=1)
    except Exception as exc:
        import traceback

        print(
            f"FATAL: Unhandled exception: {type(exc).__name__}: {exc}", file=sys.stderr
        )
        traceback.print_exc(file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()