"""``helix`` CLI entrypoint — argparse-based dispatcher.

Stdlib-only by design. Click / Typer would be a heavy dependency for
a CLI whose surface area is fewer than 30 commands and whose users
mostly invoke it through scripts or a single ``helix mcp call``. The
brief explicitly requires we justify any new dependency, so we don't
add one.

Top-level structure:

* ``helix [--base-url URL] [--json] <command>``
* ``config`` subgroup: ``set-url``, ``show``
* ``auth`` subgroup: ``token set``, ``token clear``
* ``mcp`` subgroup: ``initialize``, ``tools``, ``call``
* ``plugin`` subgroup: ``install``, ``update``, ``status`` — the
  Claude Code plugin distribution channel (REST, not MCP: it moves a
  zip artifact; see :mod:`helix_cli.plugin_dist`)
* convenience commands: ``whoami``, ``assets list``, ``software …``,
  ``cloud servers``, ``notifications {summary, list}``,
  ``approvals list``, ``health``, ``ready``.

Convenience commands are thin wrappers around ``mcp call <tool>`` so
adding a new shortcut is a one-line dispatch entry.
"""
from __future__ import annotations

import argparse
import os
import sys
import traceback
from collections.abc import Callable
from typing import Any, NoReturn

from helix_cli import config as cli_config
from helix_cli import http_client, output
from helix_cli.http_client import CliRestError
from helix_cli.mcp_client import McpClient

# Exit codes (kept in sync with the contract documented in the task
# brief and in :mod:`helix_cli.http_client.CliRestError`). Constants 2
# and 3 are surfaced exclusively through :class:`CliRestError` raised
# inside :class:`~helix_cli.mcp_client.McpClient` — they live in the
# shared exception, not here, so we don't need module-level constants
# for them. The constants below are the ones referenced directly by
# this module's code.
EXIT_SUCCESS = 0
EXIT_USAGE = 1
EXIT_REMOTE = 4
EXIT_NETWORK = 5
EXIT_INTERNAL = 5  # local I/O / serialization / unexpected exception


# ── argparse builder ────────────────────────────────────────────────


class _RedactingArgumentParser(argparse.ArgumentParser):
    """ArgumentParser whose usage errors honour the CLI's contracts.

    The stock ``error()`` does two things we can't accept (launch
    audit, 2026-06):

    1. ``sys.exit(2)`` — colliding with the documented exit-code
       contract where 2 means UNAUTHORIZED. Usage errors are
       :data:`EXIT_USAGE` (1).
    2. It echoes the offending argv fragment raw to stderr — a guessed
       ``--token helix_pat_…`` flag would land the PAT unredacted in
       terminal scrollback and CI logs ("unrecognized arguments:
       --token helix_pat_…"). Route the message through the redacting
       ``write_err`` chokepoint instead.

    Subparsers inherit this class automatically: ``add_subparsers()``
    defaults ``parser_class`` to ``type(self)``, so every
    ``add_parser(...)`` below this root produces another
    ``_RedactingArgumentParser`` — verified by the usage-error tests
    in ``tests/test_cli/test_cli_argparse.py``.
    """

    def error(self, message: str) -> NoReturn:
        # The usage line is static parser metadata (no user input) —
        # safe to print raw, and argparse users expect to see it.
        self.print_usage(sys.stderr)
        # The message CAN embed user argv → redacting chokepoint.
        output.write_err(f"{self.prog}: error: {message}")
        raise SystemExit(EXIT_USAGE)


def _build_parser() -> argparse.ArgumentParser:
    parser = _RedactingArgumentParser(
        prog="helix",
        description=(
            "Helix CLI — a thin client over the Helix MCP server. "
            "Inherits your Web UI permissions via a personal access "
            "token (helix_pat_…)."
        ),
    )
    parser.add_argument(
        "--base-url",
        default=None,
        help="Override the Helix base URL (default from $HELIX_BASE_URL or config).",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Emit raw JSON instead of human-readable output.",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=http_client.DEFAULT_TIMEOUT_SECONDS,
        help="Per-request timeout in seconds (default 30).",
    )

    sub = parser.add_subparsers(dest="command", metavar="<command>")
    sub.required = False  # ``helix`` alone prints help, exits 0

    # ── config ────────────────────────────────────────────────────────
    p_config = sub.add_parser("config", help="Show or update CLI configuration.")
    s_config = p_config.add_subparsers(dest="action", metavar="<action>")
    s_config.required = True

    p_set_url = s_config.add_parser("set-url", help="Persist a Helix base URL.")
    p_set_url.add_argument("url", help="e.g. https://helix.pingcap.net")

    s_config.add_parser("show", help="Print resolved config (token redacted).")

    # ── auth ──────────────────────────────────────────────────────────
    p_auth = sub.add_parser("auth", help="Manage the persisted PAT.")
    s_auth = p_auth.add_subparsers(dest="action", metavar="<action>")
    s_auth.required = True

    p_token = s_auth.add_parser("token", help="PAT CRUD.")
    s_token = p_token.add_subparsers(dest="token_action", metavar="<action>")
    s_token.required = True

    p_token_set = s_token.add_parser(
        "set",
        help="Save a PAT to the config file. Reads from stdin (--token-stdin).",
    )
    p_token_set.add_argument(
        "--token-stdin",
        action="store_true",
        required=True,
        help="Read the PAT from stdin (the only supported input mode — "
             "argv tokens leak via shell history and ps(1)).",
    )

    s_token.add_parser("clear", help="Remove the persisted PAT.")

    # ── mcp ───────────────────────────────────────────────────────────
    p_mcp = sub.add_parser("mcp", help="Talk to the Helix MCP server directly.")
    s_mcp = p_mcp.add_subparsers(dest="action", metavar="<action>")
    s_mcp.required = True

    s_mcp.add_parser("initialize", help="Send the JSON-RPC handshake.")
    s_mcp.add_parser("tools", help="List registered MCP tools.")

    p_call = s_mcp.add_parser("call", help="Invoke a single MCP tool.")
    p_call.add_argument("tool", help="Tool name (see `helix mcp tools`).")
    p_call.add_argument(
        "--arg",
        action="append",
        default=[],
        metavar="key=value",
        help="Scalar argument; pass --arg multiple times for multiple keys.",
    )
    p_call.add_argument(
        "--json-arg",
        default=None,
        metavar="JSON",
        help="Full JSON object of arguments. Overrides --arg if both given.",
    )
    p_call.add_argument(
        "--yes", "-y",
        action="store_true",
        help=(
            "Skip the interactive confirmation prompt for destructive "
            "tools (revoke / delete / cancel / reveal). Scriptable use."
        ),
    )

    # ── convenience commands ─────────────────────────────────────────
    sub.add_parser("whoami", help="Show the calling user (MCP: whoami).")

    p_assets = sub.add_parser("assets", help="Asset shortcuts.")
    s_assets = p_assets.add_subparsers(dest="action", metavar="<action>")
    s_assets.required = True
    s_assets.add_parser("list", help="List your assets (MCP: list_my_assets).")

    p_sw = sub.add_parser("software", help="Software shortcuts.")
    s_sw = p_sw.add_subparsers(dest="action", metavar="<action>")
    s_sw.required = True
    s_sw.add_parser("catalog", help="Software catalog (MCP: list_software_catalog).")
    s_sw.add_parser("seats", help="Your seats (MCP: list_my_software_seats).")

    p_cloud = sub.add_parser("cloud", help="CMDB / cloud-resource shortcuts.")
    s_cloud = p_cloud.add_subparsers(dest="action", metavar="<action>")
    s_cloud.required = True
    s_cloud.add_parser("servers", help="Cloud servers (MCP: list_cloud_servers).")

    p_notif = sub.add_parser("notifications", help="Inbox shortcuts.")
    s_notif = p_notif.add_subparsers(dest="action", metavar="<action>")
    s_notif.required = True
    s_notif.add_parser(
        "summary",
        help="Inbox totals (MCP: get_my_notifications_summary).",
    )
    s_notif.add_parser(
        "list",
        help="Inbox items (MCP: list_my_notifications).",
    )

    p_appr = sub.add_parser("approvals", help="Approval shortcuts.")
    s_appr = p_appr.add_subparsers(dest="action", metavar="<action>")
    s_appr.required = True
    s_appr.add_parser(
        "list",
        help="Your approval requests (MCP: list_my_approval_requests).",
    )

    p_plugin = sub.add_parser(
        "plugin",
        help="Install / update the official Helix Claude Code plugin.",
    )
    s_plugin = p_plugin.add_subparsers(dest="action", metavar="<action>")
    s_plugin.required = True
    for verb, help_text in (
        ("install", "Download, verify and register the plugin with Claude Code."),
        ("update", "Refresh to the latest published plugin release."),
    ):
        p_verb = s_plugin.add_parser(verb, help=help_text)
        p_verb.add_argument(
            "--dir",
            dest="install_dir",
            default=None,
            help="Install base directory (default: ~/.helix/claude-plugin).",
        )
        p_verb.add_argument(
            "--no-register",
            action="store_true",
            help="Skip invoking the `claude` binary; print the manual commands.",
        )
    p_pl_status = s_plugin.add_parser(
        "status", help="Installed plugin version vs the latest published one."
    )
    p_pl_status.add_argument(
        "--dir",
        dest="install_dir",
        default=None,
        help="Install base directory (default: ~/.helix/claude-plugin).",
    )

    sub.add_parser("health", help="GET /api/v1/health (no auth needed).")
    sub.add_parser("ready", help="GET /api/v1/ready (no auth needed).")

    sub.add_parser(
        "version",
        help="Show this CLI's version and check it against the server.",
    )

    return parser


# ── argument helpers ────────────────────────────────────────────────


def _parse_arg_pairs(pairs: list[str]) -> dict[str, Any]:
    """Parse ``--arg key=value`` pairs.

    Values are best-effort coerced — ``"true"``/``"false"``/``"null"``
    map to their JSON equivalents, integers to ``int``, floats to
    ``float``, bare strings stay as strings. Anything else passes
    through. This matches what users expect from a pragmatic CLI;
    when the literal string ``"true"`` is needed they can switch to
    ``--json-arg`` for full fidelity.
    """
    out: dict[str, Any] = {}
    for raw in pairs:
        if "=" not in raw:
            raise CliRestError(
                f"--arg must be 'key=value' (got {raw!r})",
                exit_code=EXIT_USAGE,
                code="USAGE",
            )
        key, _, value = raw.partition("=")
        key = key.strip()
        if not key:
            raise CliRestError(
                f"--arg key must not be empty (got {raw!r})",
                exit_code=EXIT_USAGE,
                code="USAGE",
            )
        out[key] = _coerce_scalar(value)
    return out


def _coerce_scalar(value: str) -> Any:
    if value == "":
        return ""
    lowered = value.lower()
    if lowered == "true":
        return True
    if lowered == "false":
        return False
    if lowered == "null":
        return None
    try:
        if value.startswith("-") or value[0].isdigit():
            if "." in value or "e" in lowered:
                return float(value)
            # Preserve zero-padded strings as-is: badge numbers, Aliyun
            # account ids starting with ``0``, phone numbers, and
            # approval-code-shaped strings would otherwise lose their
            # leading zero on int() coercion (``--arg badge=00123``
            # silently became 123). A standalone ``"0"`` is still an
            # int — only a leading zero followed by more digits is
            # treated as a stringy id. (R1 code review finding.)
            if (
                len(value) > 1
                and value[0] == "0"
                and value[1].isdigit()
            ):
                return value
            return int(value)
    except (ValueError, IndexError):
        pass
    return value


def _resolve_args(args: argparse.Namespace) -> dict[str, Any]:
    if getattr(args, "json_arg", None):
        import json as _json
        try:
            parsed = _json.loads(args.json_arg)
        except _json.JSONDecodeError as exc:
            raise CliRestError(
                f"--json-arg is not valid JSON: {exc.msg}",
                exit_code=EXIT_USAGE,
                code="USAGE",
            ) from exc
        if not isinstance(parsed, dict):
            raise CliRestError(
                "--json-arg must be a JSON object",
                exit_code=EXIT_USAGE,
                code="USAGE",
            )
        return parsed
    return _parse_arg_pairs(getattr(args, "arg", []) or [])


# ── output helpers ────────────────────────────────────────────────


def _emit(args: argparse.Namespace, data: Any, *, redact_secrets: bool = True) -> None:
    if args.json:
        output.emit_json(data, redact_secrets=redact_secrets)
    else:
        output.emit_human(data, redact_secrets=redact_secrets)


# ── command handlers ────────────────────────────────────────────────


def _load_cli_config(args: argparse.Namespace) -> cli_config.CliConfig:
    """``load_config`` with user-input ``ValueError`` mapped to usage.

    ``load_config`` raises ``ValueError`` when an env- or file-supplied
    base URL fails validation ("Invalid base URL from env:HELIX_BASE_URL:
    …"). That is user input, not a broken response — letting it escape
    into ``main()``'s generic ``except (OSError, ValueError)`` handler
    would mislabel it ``[BAD_RESPONSE] local error`` with exit 5.
    """
    try:
        return cli_config.load_config(cli_base_url=args.base_url)
    except ValueError as exc:
        raise CliRestError(str(exc), exit_code=EXIT_USAGE, code="USAGE") from exc


def _cmd_config_set_url(args: argparse.Namespace) -> int:
    try:
        path = cli_config.set_base_url(args.url)
    except ValueError as exc:
        # ``set_base_url`` raises ValueError for every URL the CLI
        # refuses ("must start with http:// or https://", credentials
        # embedded, internal IP literal, …). That's the user's input
        # being wrong, not a local I/O fault — map it to EXIT_USAGE
        # with the validator's clean message instead of letting the
        # generic handler print "[BAD_RESPONSE] local error" (exit 5).
        raise CliRestError(str(exc), exit_code=EXIT_USAGE, code="USAGE") from exc
    output.write_out(f"saved base URL to {path}")
    return EXIT_SUCCESS


def _cmd_config_show(args: argparse.Namespace) -> int:
    cfg = _load_cli_config(args)
    path = cli_config.config_path()
    info = {
        "config_path": str(path),
        "config_path_mode": cli_config.file_mode_octal(path),
        "base_url": cfg.base_url,
        "base_url_source": cfg.base_url_source,
        "token_present": cfg.token is not None,
        "token_display": cli_config.redact_pat(cfg.token),
        "token_source": cfg.token_source,
    }
    _emit(args, info)
    return EXIT_SUCCESS


def _cmd_auth_token_set(args: argparse.Namespace) -> int:
    if not args.token_stdin:
        raise CliRestError(
            "must pass --token-stdin (the only supported input mode)",
            exit_code=EXIT_USAGE,
            code="USAGE",
        )
    # TTY → use ``getpass`` so the typed PAT is NOT echoed back to the
    # terminal (and therefore not captured by scrollback, screen-shares,
    # asciinema / iTerm recordings, or ``script(1)`` sessions). Pipe →
    # fall back to a raw stdin read so ``echo $PAT | helix auth token
    # set --token-stdin`` keeps working in scripts.
    import getpass
    if sys.stdin.isatty():
        try:
            raw = getpass.getpass(
                "Helix PAT (input hidden): ", stream=sys.stderr,
            ).strip()
        except (EOFError, KeyboardInterrupt) as exc:
            raise CliRestError(
                "no token provided",
                exit_code=EXIT_USAGE, code="USAGE",
            ) from exc
    else:
        raw = sys.stdin.read().strip()
    if not raw:
        raise CliRestError("stdin produced no token", exit_code=EXIT_USAGE, code="USAGE")
    try:
        path = cli_config.set_token(raw)
    except ValueError as exc:
        raise CliRestError(str(exc), exit_code=EXIT_USAGE, code="USAGE") from exc
    # Don't echo the token back, even redacted, on success — the user
    # just typed it and reading it back to them is noise. State the
    # storage location and the visible prefix only.
    output.write_out(
        f"saved token ({cli_config.redact_pat(raw)}) to {path}"
    )
    return EXIT_SUCCESS


def _cmd_auth_token_clear(args: argparse.Namespace) -> int:
    # ``clear_token`` reports what actually happened so we never print
    # "cleared token" when nothing was written (the old behaviour
    # no-opped on a corrupt config file — leaving the plaintext PAT on
    # disk — yet still claimed success). See config.clear_token.
    path, outcome = cli_config.clear_token()
    if outcome == "cleared":
        output.write_out(f"cleared token from {path}")
    elif outcome == "rewrote-corrupt":
        output.write_out(
            f"config file at {path} was corrupt JSON — rewrote it without "
            f"the token so the credential is off disk"
        )
    else:  # "absent"
        output.write_out(f"no token stored at {path} — nothing to clear")
    return EXIT_SUCCESS


def _cmd_mcp_initialize(args: argparse.Namespace) -> int:
    cfg = _load_cli_config(args)
    client = McpClient(cfg, timeout=args.timeout)
    result = client.initialize()
    _emit(args, result)
    return EXIT_SUCCESS


def _cmd_mcp_tools(args: argparse.Namespace) -> int:
    cfg = _load_cli_config(args)
    client = McpClient(cfg, timeout=args.timeout)
    tools = client.list_tools()
    if args.json:
        output.emit_json({"items": tools, "count": len(tools)})
    else:
        # Compact view — most users only care about "what is the name
        # I have to type into ``helix mcp call``." Description is
        # included one-per-row for orientation; full schemas live in
        # ``--json``.
        lines = [f"{len(tools)} tools registered:"]
        for t in sorted(tools, key=lambda x: x.get("name", "")):
            name = t.get("name", "?")
            desc = (t.get("description") or "").strip().splitlines()
            head = desc[0] if desc else ""
            lines.append(f"  {name:38s} {head}")
        output.write_out("\n".join(lines))
    return EXIT_SUCCESS


#: MCP tools that mutate or revoke state. ``helix mcp call <name>``
#: against any of these prompts for confirmation on a TTY (bypass with
#: ``--yes``). The list is the canonical defense against a "wrong env
#: + good muscle memory" accident — the user typed the tool name once
#: when they meant to test locally; the second prompt makes them
#: re-type it before it ships against prod. Sourced from the MCP tool
#: registry. Names use the literal tool name as registered with
#: ``@mcp_server.tool(name=...)`` so a typo here fails silently to
#: "no confirmation needed" rather than crashing.
_DESTRUCTIVE_MCP_TOOLS = frozenset({
    # API key revocation / mutation
    "revoke_api_key",
    "update_api_key_assignment",
    # Asset hand-off / state-change (admin-only)
    "claim_asset", "return_asset", "borrow_asset",
    "return_borrowed_asset",
    "mark_asset_held_by_leaver", "recover_asset_from_leaver",
    # Asset transfer lifecycle (initiator cancel, recipient reject,
    # admin cancel). Each ends a transfer flow + may reopen blockers.
    "cancel_asset_transfer", "reject_asset_transfer",
    "admin_cancel_asset_transfer",
    # Email group / alias mutations
    "delete_my_email_alias", "delete_email_group",
    "reset_my_email_password", "transfer_email_group_owner",
    # Seat / software cancellations
    "cancel_my_seat_next_month", "admin_change_seat_plan",
    # Cloud-grant / scope mutations
    "revoke_my_cloud_grant", "transfer_cloud_scope_owner",
    "withdraw_my_cloud_scope_request",
    # DNS / SSO / VPN / domain / announcement / AI-tool cancellations
    "cancel_my_dns_request", "cancel_my_sso_grant",
    "cancel_my_domain_registration", "cancel_my_vpn_application",
    "cancel_my_announcement_request", "cancel_my_ai_tool_application",
    # Catalog mutations
    "delete_sso_app", "create_sso_app", "import_sso_apps_csv",
    # Offboarding / leaver flows
    "mark_person_leaver_manual", "revoke_offboarding_blocker",
    "leaver_blocked_force_bypass_grace",
    # Onboarding flow mutations
    "cancel_onboarding", "skip_onboarding_blocker",
    # Cloud / CMDB destructive lifecycle — server destroy is the single
    # most irreversible tool on the platform; the transfer tools hand
    # YOUR resources to someone else; bypass_4_eyes skips a control.
    "server_destroy", "transfer_my_dns_records", "transfer_my_servers",
    "revoke_domain_registration", "bypass_4_eyes_dns_request",
    # Asset office lifecycle (admin)
    "recycle_asset_to_office", "release_asset_from_office",
    # API key membership / release / transfer lifecycle
    "remove_api_key_member", "cancel_my_api_key_release",
    "cancel_my_api_key_transfer",
    # Provisioning execution — runs real downstream actions against
    # external systems (Google / Feishu / cloud), not Helix-local state.
    "execute_provisioning_task", "apply_provisioning_bundle_to_person",
    # Orphan cleanup
    "cleanup_cloud_orphan_assignments",
    # Reveal — not technically destructive but elevates a credential
    # into shoulder-surfing range, treat like a destructive action.
    "reveal_api_key", "reveal_my_contact",
})

#: Tools whose RESULT payload is, by design, a secret the user asked to
#: see. ``helix mcp call reveal_api_key`` exists to hand the credential
#: over — running the result through the credential-shape redaction
#: (``_SECRET_KEY_RE`` / JWT / PAT patterns) would replace the answer
#: with ``<redacted>`` and make the flagship reveal flow return garbage
#: with no hint why. For these tools (and ONLY these tools) the result
#: payload printed to stdout skips secret redaction; it still passes
#: the terminal-control strip (see ``output.write_out``). Errors,
#: warnings, debug traces, and every other tool keep full redaction.
_REVEAL_OUTPUT_TOOLS = frozenset({"reveal_api_key", "reveal_my_contact"})


def _confirm_destructive(tool: str, target_url: str, *, assume_yes: bool) -> None:
    """Block destructive MCP tool calls behind a TTY confirmation prompt.

    Bypassed when:
    - ``--yes`` was passed (scriptable use).
    - stdin is not a TTY (piped / batched — the operator is expected
      to have already made the decision when wiring the pipeline).

    Prompt + input both go through stderr so a user piping stdout to
    a file (``helix mcp call ... > out.json``) still sees the prompt
    and isn't left staring at a hung process.
    """
    if tool not in _DESTRUCTIVE_MCP_TOOLS:
        return
    if assume_yes:
        return
    if not sys.stdin.isatty():
        return
    output.write_err(
        f"about to call destructive tool '{tool}' against {target_url}.\n"
        f"  type the tool name to confirm (Ctrl-C to abort):"
    )
    try:
        # Write the inline prompt to stderr (matches the warning above)
        # so a user redirecting stdout still sees it.
        sys.stderr.write("  > ")
        sys.stderr.flush()
        line = sys.stdin.readline().rstrip("\n").strip()
    except (EOFError, KeyboardInterrupt) as exc:
        raise CliRestError(
            "aborted by user",
            exit_code=EXIT_USAGE, code="ABORTED",
        ) from exc
    if line != tool:
        raise CliRestError(
            f"confirmation mismatch — expected {tool!r}, got {line!r}",
            exit_code=EXIT_USAGE, code="ABORTED",
        )


def _cmd_mcp_call(args: argparse.Namespace) -> int:
    cfg = _load_cli_config(args)
    _confirm_destructive(
        args.tool, target_url=cfg.base_url,
        assume_yes=getattr(args, "yes", False),
    )
    client = McpClient(cfg, timeout=args.timeout)
    arguments = _resolve_args(args)
    result = client.call_tool(args.tool, arguments)
    # Reveal tools: the user explicitly asked for the credential, so
    # the result payload skips credential-shape redaction (it would
    # print ``<redacted>`` instead of the answer). Terminal-control
    # stripping still applies inside ``write_out``. Tool errors raise
    # in ``call_tool`` and flow through the fully-redacting error path.
    _emit(
        args, result.payload,
        redact_secrets=args.tool not in _REVEAL_OUTPUT_TOOLS,
    )
    return EXIT_SUCCESS


def _convenience_runner(tool: str) -> Callable[[argparse.Namespace], int]:
    """Return a handler that calls ``tool`` with no arguments.

    All convenience commands today call read-only MCP tools; if a
    future shortcut needs arguments it should be promoted to a
    proper subparser with explicit options rather than going
    through this helper.
    """
    def _run(args: argparse.Namespace) -> int:
        cfg = _load_cli_config(args)
        client = McpClient(cfg, timeout=args.timeout)
        result = client.call_tool(tool, {})
        _emit(args, result.payload)
        return EXIT_SUCCESS
    _run.__name__ = f"_cmd_{tool}"
    return _run


def _cmd_health(args: argparse.Namespace) -> int:
    cfg = _load_cli_config(args)
    resp = http_client.get(cfg, "/api/v1/health", timeout=args.timeout)
    if resp.status_code == 200 and isinstance(resp.body, dict) and resp.body.get("success"):
        _emit(args, resp.body.get("data") or {"status": "healthy"})
        return EXIT_SUCCESS
    detail = ""
    if isinstance(resp.body, dict):
        errors = resp.body.get("errors") or []
        if errors and isinstance(errors[0], dict):
            msg = errors[0].get("message")
            if msg:
                detail = f": {msg}"
    raise CliRestError(
        f"health check failed (status {resp.status_code}){detail}",
        exit_code=EXIT_REMOTE,
        code="UNHEALTHY",
    )


def _cmd_ready(args: argparse.Namespace) -> int:
    cfg = _load_cli_config(args)
    resp = http_client.get(cfg, "/api/v1/ready", timeout=args.timeout)
    if isinstance(resp.body, dict) and resp.body.get("success") and resp.status_code == 200:
        # The server's /ready (app/main.py) ALWAYS replies 200 +
        # success:true — degradation is carried inside ``data`` as
        # ``{"status": "ready"|"degraded", "checks": {"database":
        # "ok"|"error", "redis": ...}}``, never as an HTTP 503. So a
        # bare status-code check would happily exit 0 while the DB is
        # down. Inspect the payload: any non-"ok" check (or an
        # unexpected status value) is NOT_READY.
        data = resp.body.get("data")
        data = data if isinstance(data, dict) else {}
        checks = data.get("checks")
        checks = checks if isinstance(checks, dict) else {}
        failing = sorted(k for k, v in checks.items() if v != "ok")
        status = data.get("status")
        # ``status`` is "ready" today; tolerate "ok" (the generic
        # convention) and a missing field (older server) as healthy so
        # the CLI doesn't false-alarm across server versions.
        degraded = bool(failing) or status not in (None, "ok", "ready")
        if not degraded:
            _emit(args, data)
            return EXIT_SUCCESS
        # We intentionally do NOT emit the body before raising —
        # emitting both the data and a separate error line would break
        # --json mode (two documents on stdout) and confuse human
        # readers. The error message names the failing checks instead.
        detail = (
            ", ".join(f"{k}={checks.get(k)}" for k in failing)
            or f"status={status}"
        )
        raise CliRestError(
            f"server not ready ({detail})",
            exit_code=EXIT_REMOTE,
            code="NOT_READY",
        )
    # Defense in depth: a 503 + envelope-shaped error (load balancer,
    # older server, reverse proxy) is still a structured "not ready",
    # not a network failure. Same no-body-before-raise rule as above.
    detail = ""
    if isinstance(resp.body, dict):
        errors = resp.body.get("errors") or []
        if errors and isinstance(errors[0], dict):
            msg = errors[0].get("message")
            if msg:
                detail = f": {msg}"
    raise CliRestError(
        f"readiness check failed (status {resp.status_code}){detail}",
        exit_code=EXIT_REMOTE,
        code="NOT_READY",
    )


def _cmd_plugin_install(args: argparse.Namespace) -> int:
    """``helix plugin install`` / ``helix plugin update`` — same
    idempotent operation; `update` exists so intent reads naturally."""
    from pathlib import Path

    from helix_cli import plugin_dist

    cfg = _load_cli_config(args)
    base = Path(args.install_dir).expanduser() if args.install_dir else None
    res = plugin_dist.install(
        cfg,
        timeout=args.timeout,
        base=base,
        register=not args.no_register,
    )
    payload: dict[str, Any] = {
        "version": res.version,
        "plugin_version": res.plugin_version,
        "path": res.path,
        "sha256": res.sha256,
        "size": res.size,
        "first_install": res.first_install,
        "registration": res.registration,
    }
    if not res.registration.get("registered") and res.registration.get("manual"):
        payload["next_steps"] = res.registration["manual"]
    _emit(args, payload)
    return EXIT_SUCCESS


def _cmd_plugin_status(args: argparse.Namespace) -> int:
    from pathlib import Path

    from helix_cli import plugin_dist

    cfg = _load_cli_config(args)
    base = Path(args.install_dir).expanduser() if args.install_dir else None
    _emit(args, plugin_dist.status(cfg, timeout=args.timeout, base=base))
    return EXIT_SUCCESS


def _cmd_version(args: argparse.Namespace) -> int:
    """Show the local CLI version and, best-effort, compare it to the
    server's published latest + minimum-supported versions."""
    from helix_cli import __version__, version_check

    cfg = _load_cli_config(args)
    rep = version_check.fetch_report(cfg, timeout=args.timeout)
    status_note = {
        "up_to_date": "up to date",
        "behind": "a newer version is available — update via Help → AI 助手集成 → helix CLI",
        "below_min": "BELOW the minimum supported version — update now (old builds may have known bugs/security issues)",
        "unknown": "could not reach the server to check (showing local version only)",
    }.get(rep["status"], "unknown")
    _emit(args, {
        "version": __version__,
        "server_latest": rep["latest"],
        "server_min_supported": rep["min_supported"],
        "status": rep["status"],
        "note": status_note,
    })
    # Below-min is the one case worth a non-zero exit so scripts/CI can
    # gate on "my CLI is too old" without parsing the payload.
    return EXIT_REMOTE if rep["status"] == "below_min" else EXIT_SUCCESS


#: Commands that talk to the authenticated API — the passive
#: version-nudge runs only for these (offline/probe commands skip it).
_NETWORK_COMMANDS = frozenset({
    "mcp", "whoami", "assets", "software", "cloud", "notifications", "approvals",
    "plugin",
})


# ── dispatcher ──────────────────────────────────────────────────────


def _dispatch(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int:
    cmd = args.command
    if cmd is None:
        parser.print_help()
        return EXIT_SUCCESS

    if cmd == "config":
        if args.action == "set-url":
            return _cmd_config_set_url(args)
        if args.action == "show":
            return _cmd_config_show(args)
    elif cmd == "auth":
        if args.action == "token":
            if args.token_action == "set":
                return _cmd_auth_token_set(args)
            if args.token_action == "clear":
                return _cmd_auth_token_clear(args)
    elif cmd == "mcp":
        if args.action == "initialize":
            return _cmd_mcp_initialize(args)
        if args.action == "tools":
            return _cmd_mcp_tools(args)
        if args.action == "call":
            return _cmd_mcp_call(args)
    elif cmd == "whoami":
        return _convenience_runner("whoami")(args)
    elif cmd == "assets" and args.action == "list":
        return _convenience_runner("list_my_assets")(args)
    elif cmd == "software" and args.action == "catalog":
        return _convenience_runner("list_software_catalog")(args)
    elif cmd == "software" and args.action == "seats":
        return _convenience_runner("list_my_software_seats")(args)
    elif cmd == "cloud" and args.action == "servers":
        return _convenience_runner("list_cloud_servers")(args)
    elif cmd == "notifications" and args.action == "summary":
        return _convenience_runner("get_my_notifications_summary")(args)
    elif cmd == "notifications" and args.action == "list":
        return _convenience_runner("list_my_notifications")(args)
    elif cmd == "approvals" and args.action == "list":
        return _convenience_runner("list_my_approval_requests")(args)
    elif cmd == "plugin":
        if args.action in ("install", "update"):
            return _cmd_plugin_install(args)
        if args.action == "status":
            return _cmd_plugin_status(args)
    elif cmd == "health":
        return _cmd_health(args)
    elif cmd == "ready":
        return _cmd_ready(args)
    elif cmd == "version":
        return _cmd_version(args)

    parser.print_help()
    return EXIT_USAGE


def main(argv: list[str] | None = None) -> int:
    """Process-level entrypoint.

    Returns an integer exit code so callers can use either
    ``raise SystemExit(main())`` (the ``__main__`` shim) or
    ``[project.scripts] helix = "helix_cli.main:main"`` (which Python's
    setuptools/hatchling shim treats as ``sys.exit(main())``).
    """
    parser = _build_parser()
    args = parser.parse_args(argv)
    debug = os.environ.get("HELIX_DEBUG") == "1"

    # Passive version nudge: for commands that already talk to the API,
    # warn (best-effort, cached) if this build is behind / below the
    # supported floor — BEFORE the server-side 426 hard-stop bites.
    # Never blocks or fails the real command.
    if getattr(args, "command", None) in _NETWORK_COMMANDS:
        try:
            from helix_cli import version_check
            version_check.warn_if_outdated(_load_cli_config(args), timeout=args.timeout)
        except Exception:  # noqa: BLE001
            pass

    try:
        return _dispatch(args, parser)
    except CliRestError as exc:
        output.emit_error(str(exc), code=exc.code)
        if debug:
            # Even in debug, redaction in ``write_err`` strips secrets
            # before the trace lands on the terminal.
            output.write_err(traceback.format_exc())
        return exc.exit_code
    except KeyboardInterrupt:
        output.emit_error("interrupted")
        return 130  # standard SIGINT exit code, matches bash/coreutils
    except BrokenPipeError:
        # Downstream closed the pipe (``helix mcp tools | head -1``).
        # Standard Unix tools treat downstream-close as success — head
        # got everything it wanted — so we exit 0 and print NOTHING;
        # mapping this to "[LOCAL_IO] local error" + exit 5 (the old
        # behaviour: BrokenPipeError is an OSError) made every piped
        # invocation look broken. MUST be caught before the generic
        # OSError handler below for exactly that subclass reason.
        # Standard Python recipe (docs: "note on SIGPIPE"): point
        # stdout at devnull so the interpreter's implicit flush at
        # shutdown doesn't print "Exception ignored ..." noise.
        try:
            devnull = os.open(os.devnull, os.O_WRONLY)
            os.dup2(devnull, sys.stdout.fileno())
        except (OSError, ValueError):
            # stdout may not be a real file descriptor (tests patch in
            # a StringIO; embedded use) — nothing to silence then.
            pass
        return EXIT_SUCCESS
    except (OSError, ValueError) as exc:
        # Local I/O failures (disk full, permission denied, NotADirectoryError
        # if HELIX_CONFIG points at a path under a regular file) and
        # serialization failures (circular refs, un-encodable types).
        # These are NOT network errors — calling them "EXIT_NETWORK"
        # sends the user chasing connectivity ghosts. Keep the same
        # exit code 5 (no separate INTERNAL code in the contract) but
        # use a category-accurate ``code=`` so scripts can match on it.
        output.emit_error(
            f"local error: {type(exc).__name__}: {exc}",
            code="LOCAL_IO" if isinstance(exc, OSError) else "BAD_RESPONSE",
        )
        if debug:
            output.write_err(traceback.format_exc())
        return EXIT_INTERNAL
    except Exception as exc:  # noqa: BLE001
        # Unknown errors: never leak the message text raw — it might
        # contain a user-supplied PAT we passed through. ``redact``
        # applies in ``write_err``.
        output.emit_error(
            f"unexpected error: {type(exc).__name__}: {exc}",
            code="INTERNAL",
        )
        if debug:
            output.write_err(traceback.format_exc())
        return EXIT_INTERNAL


__all__ = ["main"]