setup_wizard.py

"""Interactive setup wizard for the Inkbox Hermes plugin."""

from __future__ import annotations

import asyncio
import getpass
import importlib
import importlib.metadata
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import time
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlencode

try:
    from .config import (
        INKBOX_BASE_URL_DEFAULT,
        InkboxPluginConfig,
        VoiceStack,
        inkbox_base_url_kwargs,
        inkbox_client_kwargs,
        public_call_ws_url,
        resolve_voice_stack,
    )
except ImportError:  # pragma: no cover - direct local import/test fallback
    from config import (
        INKBOX_BASE_URL_DEFAULT,
        InkboxPluginConfig,
        VoiceStack,
        inkbox_base_url_kwargs,
        inkbox_client_kwargs,
        public_call_ws_url,
        resolve_voice_stack,
    )

try:
    from hermes_cli.colors import Colors, color
except Exception:  # pragma: no cover - local tests without Hermes
    class Colors:
        CYAN = ""
        DIM = ""
        GREEN = ""
        RED = ""
        YELLOW = ""
        BOLD = ""

    def color(text: str, *_args: Any) -> str:
        return text

try:
    from hermes_cli.cli_output import print_error, print_info, print_success, print_warning
except Exception:  # pragma: no cover - local tests without Hermes
    def print_error(message: str) -> None:
        print(message)

    def print_info(message: str) -> None:
        print(message)

    def print_success(message: str) -> None:
        print(message)

    def print_warning(message: str) -> None:
        print(message)

try:
    from hermes_cli.secret_prompt import masked_secret_prompt
except Exception:  # pragma: no cover - local tests without Hermes
    masked_secret_prompt = None


INKBOX_MIN_VERSION = "0.7.11"
INKBOX_REQUIREMENTS = (
    f"inkbox>={INKBOX_MIN_VERSION},<1.0.0", "aiohttp>=3.9", "segno>=1.5",
    "audioop-lts>=0.2.1; python_version >= '3.13'",
)
_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~")
_AVATAR_PATH = Path(__file__).resolve().parent / "assets" / "hermes_with_iphone.png"
_RAW_AVATAR_BASE_URL_DEFAULT = "https://inkbox.ai"
OPENAI_REALTIME_TEST_MODEL = "gpt-realtime-2"
OPENAI_REALTIME_TEST_URL = "wss://api.openai.com/v1/realtime"
VOICE_STACK_CHOICES = [
    (
        VoiceStack.INKBOX_VOICE_AI,
        "Inkbox Voice AI — Inkbox handles calls on behalf of your agent; "
        "Hermes is notified when the call ends.",
    ),
    (
        VoiceStack.OPENAI_REALTIME,
        "OpenAI Realtime API — Bring your own API key; the realtime agent can "
        "consult Hermes for complex tasks.",
    ),
    (
        VoiceStack.INKBOX_TTS_STT,
        "Inkbox TTS/STT — Hermes talks through the Inkbox voice stack with "
        "increased latency.",
    ),
]
# A restart drains in-flight agent runs before coming back up, so give it more
# room than the CLI's own 90s service-restart timeout.
GATEWAY_COMMAND_TIMEOUT = 180
# A gateway only becomes visible to the CLI once it has taken its runtime lock
# and written its PID record, which happens partway through its own startup.
GATEWAY_START_CONFIRM_TIMEOUT = 15.0


def print_header(title: str) -> None:
    print()
    print(color(f"* {title}", Colors.CYAN, Colors.BOLD))


def _show_qr(data: str) -> bool:
    stdout = getattr(sys, "stdout", None)
    if stdout is not None and hasattr(stdout, "isatty") and not stdout.isatty():
        return False
    try:
        import segno
    except ImportError:
        return False
    try:
        segno.make(data).terminal(compact=True)
        return True
    except Exception:
        return False


def _sanitize_pasted_input(value: str) -> str:
    if not isinstance(value, str) or not value:
        return value
    return _BRACKETED_PASTE_PATTERN.sub("", value)


def _is_interactive_stdin() -> bool:
    stdin = getattr(sys, "stdin", None)
    if stdin is None:
        return False
    try:
        return bool(stdin.isatty())
    except Exception:
        return False


def prompt(question: str, default: str | None = None, *, password: bool = False) -> str:
    display = f"{question} [{default}]: " if default else f"{question}: "
    try:
        if password:
            if masked_secret_prompt is not None:
                value = masked_secret_prompt(color(display, Colors.YELLOW))
            else:
                value = getpass.getpass(display)
        else:
            value = input(color(display, Colors.YELLOW))
    except (KeyboardInterrupt, EOFError):
        print()
        raise SystemExit(1)
    cleaned = _sanitize_pasted_input(value)
    return cleaned.strip() or default or ""


def prompt_yes_no(question: str, default: bool = True) -> bool:
    default_word = "yes" if default else "no"
    while True:
        try:
            value = input(color(f"{question} [y/n] (default: {default_word}): ", Colors.YELLOW)).strip().lower()
        except (KeyboardInterrupt, EOFError):
            print()
            raise SystemExit(1)
        if not value:
            return default
        if value in {"y", "yes"}:
            return True
        if value in {"n", "no"}:
            return False
        print_error("Please enter 'y' or 'n'.")


def prompt_enter_to_continue(message: str) -> None:
    if not _is_interactive_stdin():
        return
    try:
        input(color(f"{message}... ", Colors.YELLOW))
    except (KeyboardInterrupt, EOFError):
        print()
        raise SystemExit(1)


def prompt_choice(
    question: str,
    choices: list[str],
    default: int = 0,
    *,
    description: str | None = None,
) -> int:
    try:
        from hermes_cli.curses_ui import curses_radiolist
    except Exception:
        curses_radiolist = None

    if curses_radiolist is not None:
        selected = curses_radiolist(
            question,
            choices,
            selected=default,
            cancel_returns=default,
            description=description,
        )
        print()
        return selected

    print(color(question, Colors.YELLOW))
    if description:
        for line in description.splitlines():
            print_info(f"  {line}")
    for idx, choice in enumerate(choices, start=1):
        marker = "*" if idx - 1 == default else " "
        print(f"  {marker} {idx}. {choice}")
    while True:
        try:
            value = input(color(f"  Select [1-{len(choices)}] ({default + 1}): ", Colors.DIM)).strip()
        except (KeyboardInterrupt, EOFError):
            print()
            raise SystemExit(1)
        if not value:
            return default
        try:
            selected = int(value) - 1
        except ValueError:
            print_error("Please enter a number.")
            continue
        if 0 <= selected < len(choices):
            return selected
        print_error(f"Please enter a number between 1 and {len(choices)}.")


def _save(name: str, value: str) -> None:
    if value == "":
        return
    from hermes_cli.config import save_env_value

    save_env_value(name, value)


def _env(name: str) -> str:
    try:
        from hermes_cli.config import get_env_value

        return os.getenv(name) or get_env_value(name) or ""
    except Exception:
        return os.getenv(name, "")


def _config_realtime_api_key() -> str:
    try:
        from hermes_cli.config import load_config

        cfg = load_config() or {}
    except Exception:
        return ""

    platforms = cfg.get("platforms") if isinstance(cfg, dict) else {}
    inkbox = platforms.get("inkbox") if isinstance(platforms, dict) else {}
    realtime = inkbox.get("realtime") if isinstance(inkbox, dict) else {}
    api_key = realtime.get("api_key") if isinstance(realtime, dict) else ""
    return str(api_key or "").strip()


def _hermes_openai_api_key() -> tuple[str, str] | None:
    try:
        from hermes_cli.auth import has_usable_secret, resolve_api_key_provider_credentials

        creds = resolve_api_key_provider_credentials("openai-api")
    except Exception:
        return None

    api_key = str(creds.get("api_key") or "").strip()
    if not api_key or not has_usable_secret(api_key):
        return None
    source = str(creds.get("source") or "openai-api").strip() or "openai-api"
    return source, api_key


def _detect_openai_realtime_key() -> tuple[str, str] | None:
    config_key = _config_realtime_api_key()
    if config_key:
        return "platforms.inkbox.realtime.api_key", config_key
    realtime_key = _env("INKBOX_REALTIME_API_KEY").strip()
    if realtime_key:
        return "INKBOX_REALTIME_API_KEY", realtime_key
    hermes_key = _hermes_openai_api_key()
    if hermes_key is not None:
        return hermes_key
    openai_key = _env("OPENAI_API_KEY").strip()
    if openai_key:
        return "OPENAI_API_KEY", openai_key
    return None


def _install_commands() -> list[list[list[str]]]:
    plans: list[list[list[str]]] = []
    uv = shutil.which("uv")
    if uv:
        plans.append([[uv, "pip", "install", "--python", sys.executable, *INKBOX_REQUIREMENTS]])
    plans.append([[sys.executable, "-m", "pip", "install", *INKBOX_REQUIREMENTS]])
    plans.append(
        [
            [sys.executable, "-m", "ensurepip", "--upgrade"],
            [sys.executable, "-m", "pip", "install", *INKBOX_REQUIREMENTS],
        ]
    )
    return plans


def _install_command_text() -> str:
    if sys.platform == "win32":
        return "; ".join(
            "& " + " ".join("'" + arg.replace("'", "''") + "'" for arg in command)
            for command in _install_commands()[0]
        )
    return " && ".join(shlex.join(command) for command in _install_commands()[0])


def _run_install_plan() -> bool:
    last_exc: Exception | None = None
    for plan in _install_commands():
        try:
            for command in plan:
                subprocess.check_call(command)
            return True
        except Exception as exc:
            last_exc = exc
    if last_exc is not None:
        print_error(f"Install failed: {last_exc}")
    return False


def _purge_inkbox_modules() -> None:
    for name in list(sys.modules):
        if name == "inkbox" or name.startswith("inkbox."):
            sys.modules.pop(name, None)


def _load_inkbox_symbols() -> dict[str, Any]:
    from inkbox import Inkbox
    from inkbox.exceptions import InkboxAPIError
    from inkbox.identities.types import IdentityPhoneNumberCreateOptions
    from inkbox.whoami.types import (
        AUTH_SUBTYPE_API_KEY_ADMIN_SCOPED,
        AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_CLAIMED,
        AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_UNCLAIMED,
        WhoamiApiKeyResponse,
    )

    return {
        "Inkbox": Inkbox,
        "InkboxAPIError": InkboxAPIError,
        "IdentityPhoneNumberCreateOptions": IdentityPhoneNumberCreateOptions,
        "WhoamiApiKeyResponse": WhoamiApiKeyResponse,
        "ADMIN_SCOPED": AUTH_SUBTYPE_API_KEY_ADMIN_SCOPED,
        "AGENT_CLAIMED": AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_CLAIMED,
        "AGENT_UNCLAIMED": AUTH_SUBTYPE_API_KEY_AGENT_SCOPED_UNCLAIMED,
    }


def _parse_version(value: str) -> tuple[int, ...]:
    # Best-effort numeric parse of "X.Y.Z" so we can compare without packaging.
    parts: list[int] = []
    for chunk in value.split("."):
        digits = ""
        for char in chunk:
            if char.isdigit():
                digits += char
            else:
                break
        if digits == "":
            break
        parts.append(int(digits))
    return tuple(parts)


def _inkbox_version_ok() -> bool:
    try:
        installed = importlib.metadata.version("inkbox")
    except Exception:
        return False
    try:
        from packaging.version import Version

        return Version(installed) >= Version(INKBOX_MIN_VERSION)
    except Exception:
        # Fall back to a simple parsed-tuple comparison when packaging is unavailable.
        return _parse_version(installed) >= _parse_version(INKBOX_MIN_VERSION)


def _ensure_inkbox_sdk() -> dict[str, Any] | None:
    try:
        symbols = _load_inkbox_symbols()
        if _inkbox_version_ok():
            return symbols
        first_error = (
            f"inkbox SDK is older than {INKBOX_MIN_VERSION}; an upgrade is required."
        )
    except Exception as exc:
        first_error = exc

    print_warning("The Python Inkbox SDK is not available in the Hermes environment.")
    print_info("The setup command is running under:")
    print_info(f"  {sys.executable}")
    print_info("Install or upgrade the SDK in that exact environment with:")
    print_info(f"  {_install_command_text()}")
    print_info(f"Import error: {first_error}")

    if not _is_interactive_stdin():
        return None
    if not prompt_yes_no("Install/upgrade Inkbox SDK in this Hermes environment now?", True):
        return None

    if not _run_install_plan():
        print_info("Run this command manually, then rerun setup:")
        print_info(f"  {_install_command_text()}")
        return None

    importlib.invalidate_caches()
    _purge_inkbox_modules()
    try:
        return _load_inkbox_symbols()
    except Exception as retry_exc:
        print_error(f"Inkbox SDK still cannot be imported: {retry_exc}")
        print_info("Run this command manually, then rerun setup:")
        print_info(f"  {_install_command_text()}")
        return None


def _error_status(exc: Exception) -> Any:
    return getattr(exc, "status_code", "?")


def _error_detail(exc: Exception) -> str:
    return str(getattr(exc, "detail", "") or exc)


def _enum_value(value: Any) -> str:
    raw = getattr(value, "value", value)
    return str(raw or "")


def _seed_identity_state(identity: Any) -> None:
    try:
        from hermes_cli.config import get_hermes_home

        mailbox = getattr(identity, "mailbox", None)
        phone = getattr(identity, "phone_number", None)
        tunnel = getattr(identity, "tunnel", None)
        state = {
            "handle": getattr(identity, "agent_handle", None),
            "email_address": (
                getattr(identity, "email_address", None)
                or (getattr(mailbox, "email_address", None) if mailbox else None)
            ),
            "phone_number": getattr(phone, "number", None) if phone else None,
            "phone_number_id": str(getattr(phone, "id", "")) if phone else None,
            "imessage_enabled": bool(getattr(identity, "imessage_enabled", False)),
            "tunnel_public_host": getattr(tunnel, "public_host", None) if tunnel else None,
        }
        path = Path(get_hermes_home()) / "inkbox_identity_state.json"
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_suffix(path.suffix + ".tmp")
        tmp.write_text(json.dumps(state, indent=2) + "\n")
        os.replace(tmp, path)
    except Exception as exc:
        print_warning(f"  Could not seed inkbox_identity_state.json: {exc}")
        print_info("  Start the gateway and it will populate the file on connect.")


def _redact_key_source(name: str) -> str:
    if name == "platforms.inkbox.realtime.api_key":
        return "platforms.inkbox.realtime.api_key"
    if name == "INKBOX_REALTIME_API_KEY":
        return "INKBOX_REALTIME_API_KEY"
    if name == "OPENAI_API_KEY":
        return "OPENAI_API_KEY"
    if name == "credential_pool:openai-api":
        return "Hermes credential pool (openai-api)"
    if name == "openai-api":
        return "Hermes OpenAI API credentials"
    return "the configured OpenAI API key"


async def _test_openai_realtime_api_key_async(api_key: str, model: str) -> tuple[bool, str]:
    try:
        import aiohttp
    except Exception as exc:
        return False, f"aiohttp is not available in this Hermes environment: {exc}"

    url = f"{OPENAI_REALTIME_TEST_URL}?{urlencode({'model': model})}"
    headers = {"Authorization": f"Bearer {api_key}"}
    timeout = aiohttp.ClientTimeout(total=12)
    session_update = {
        "type": "session.update",
        "session": {
            "type": "realtime",
            "model": model,
            "instructions": "Validation probe. Do not speak unless audio is provided.",
            "output_modalities": ["audio"],
            "audio": {
                "input": {
                    "format": {"type": "audio/pcmu"},
                    "noise_reduction": None,
                    "transcription": {"model": "gpt-4o-mini-transcribe"},
                    "turn_detection": {
                        "type": "server_vad",
                        "threshold": 0.5,
                        "prefix_padding_ms": 300,
                        "silence_duration_ms": 500,
                        "create_response": True,
                        "interrupt_response": True,
                    },
                },
                "output": {
                    "format": {"type": "audio/pcmu"},
                    "voice": "cedar",
                },
            },
        },
    }

    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.ws_connect(url, headers=headers, heartbeat=10) as ws:
                await ws.send_str(json.dumps(session_update))
                loop = asyncio.get_running_loop()
                deadline = loop.time() + 8.0
                saw_session_created = False
                while True:
                    remaining = deadline - loop.time()
                    if remaining <= 0:
                        if saw_session_created:
                            return True, "OpenAI Realtime websocket accepted the key."
                        return False, "Timed out waiting for an OpenAI Realtime session response."
                    msg = await asyncio.wait_for(ws.receive(), timeout=remaining)
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        try:
                            event = json.loads(msg.data)
                        except Exception:
                            continue
                        event_type = str(event.get("type") or "")
                        if event_type == "session.updated":
                            return True, "OpenAI Realtime session update succeeded."
                        if event_type == "session.created":
                            saw_session_created = True
                            continue
                        if event_type == "error":
                            error = event.get("error") if isinstance(event.get("error"), dict) else event
                            message = str(error.get("message") or event).strip()
                            code = str(error.get("code") or "").strip()
                            prefix = f"{code}: " if code else ""
                            return False, f"{prefix}{message}"
                    if msg.type in {aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR}:
                        detail = str(getattr(ws, "exception", lambda: None)() or "websocket closed")
                        return False, detail
    except aiohttp.WSServerHandshakeError as exc:
        if exc.status in {401, 403}:
            return False, f"OpenAI rejected the key or Realtime permission: HTTP {exc.status}"
        return False, f"OpenAI Realtime websocket handshake failed: HTTP {exc.status} {exc.message}"
    except asyncio.TimeoutError:
        return False, "Timed out connecting to OpenAI Realtime."
    except Exception as exc:
        return False, str(exc)


def _test_openai_realtime_api_key(api_key: str, model: str = OPENAI_REALTIME_TEST_MODEL) -> tuple[bool, str]:
    try:
        return asyncio.run(_test_openai_realtime_api_key_async(api_key, model))
    except RuntimeError as exc:
        return False, f"Could not run Realtime validation from this setup process: {exc}"


def _voice_stack_default_index(detected_realtime_key: str) -> int:
    stack, _ = resolve_voice_stack(
        _env("INKBOX_VOICE_STACK"),
        realtime_enabled=_env("INKBOX_REALTIME_ENABLED") or None,
        realtime_api_key=detected_realtime_key,
    )
    return next(
        (index for index, (candidate, _) in enumerate(VOICE_STACK_CHOICES) if candidate is stack),
        2,
    )


def _setup_call_ws_url(identity: Any) -> str:
    cfg = InkboxPluginConfig(
        identity=str(getattr(identity, "agent_handle", "") or "").strip(),
        public_url=_env("INKBOX_PUBLIC_URL").strip(),
        tunnel_name=_env("INKBOX_TUNNEL_NAME").strip(),
    )
    return public_call_ws_url(cfg, identity)


def _incoming_call_values(config: Any) -> dict[str, Any]:
    return {
        "incoming_call_action": _enum_value(getattr(config, "incoming_call_action", "")),
        "client_websocket_url": getattr(config, "client_websocket_url", None),
        "incoming_call_webhook_url": getattr(config, "incoming_call_webhook_url", None),
    }


def _set_local_incoming_call_action(identity: Any) -> None:
    setter = getattr(identity, "set_incoming_call_action", None)
    if not callable(setter):
        raise RuntimeError("The installed Inkbox SDK cannot configure identity-scoped incoming calls.")
    ws_url = _setup_call_ws_url(identity)
    if not ws_url:
        raise RuntimeError("Could not resolve the public call WebSocket URL.")
    setter(
        incoming_call_action="auto_accept",
        client_websocket_url=ws_url,
        incoming_call_webhook_url=None,
    )


def _load_admin_identity(
    identity: Any,
    *,
    base_url: str,
    Inkbox: Any,
    InkboxAPIError: Any,
    WhoamiApiKeyResponse: Any,
    ADMIN_SCOPED: Any,
) -> Any | None:
    admin_key = prompt(
        "  Paste an admin-scoped Inkbox API key for this authority change",
        password=True,
    ).strip()
    if not admin_key:
        print_warning("  No admin-scoped key entered. Returning to the voice-stack choices.")
        return None

    try:
        client = Inkbox(**inkbox_client_kwargs(admin_key, base_url))
        info = client.whoami()
        if WhoamiApiKeyResponse is not None and not isinstance(info, WhoamiApiKeyResponse):
            print_error("  This authority change requires an admin-scoped API key.")
            return None
        if _enum_value(getattr(info, "auth_subtype", "")) != _enum_value(ADMIN_SCOPED):
            print_error("  This authority change requires an admin-scoped API key.")
            return None
        return client.get_identity(identity.agent_handle)
    except InkboxAPIError as exc:
        print_error(f"  Admin key validation failed: HTTP {_error_status(exc)} {_error_detail(exc)}")
    except Exception as exc:
        print_error(f"  Admin key validation failed: {exc}")
    return None


def _restore_voice_ai_config(
    identity: Any,
    *,
    incoming_before: Any,
    hosted_before: Any,
    authority_before: str,
    authority_identity: Any | None,
    incoming_changed: bool,
    authority_changed: bool,
    hosted_changed: bool,
) -> None:
    failures: list[str] = []
    if incoming_changed:
        try:
            identity.set_incoming_call_action(**_incoming_call_values(incoming_before))
        except Exception as exc:
            failures.append(f"incoming-call action ({exc})")
    if authority_changed and authority_identity is not None:
        try:
            authority_identity.set_hosted_agent_authority_mode(authority_before)
        except Exception as exc:
            failures.append(f"authority mode ({exc})")
    if hosted_changed:
        try:
            identity.set_hosted_agent_config(
                voice=getattr(hosted_before, "voice", None),
                model=getattr(hosted_before, "model", None),
                instructions=getattr(hosted_before, "instructions", None),
            )
        except Exception as exc:
            failures.append(f"Voice AI config ({exc})")
    if failures:
        print_warning("  Could not fully restore the prior call configuration:")
        for failure in failures:
            print_warning(f"    {failure}")


def _configure_voice_ai(
    identity: Any,
    *,
    base_url: str,
    Inkbox: Any,
    InkboxAPIError: Any,
    WhoamiApiKeyResponse: Any,
    ADMIN_SCOPED: Any,
    authority_identity: Any | None,
) -> tuple[bool, Any | None, str]:
    required_methods = (
        "get_hosted_agent_config",
        "set_hosted_agent_config",
        "get_incoming_call_action",
        "set_incoming_call_action",
    )
    if any(not callable(getattr(identity, method, None)) for method in required_methods):
        print_error("  Inkbox Voice AI setup requires a newer Inkbox SDK.")
        print_info("  Upgrade the plugin environment, then rerun setup.")
        return False, authority_identity, ""

    try:
        hosted_before = identity.get_hosted_agent_config()
        incoming_before = identity.get_incoming_call_action()
    except Exception as exc:
        print_error(f"  Could not read the current phone-call configuration: {exc}")
        return False, authority_identity, ""

    authority_before = _enum_value(
        getattr(hosted_before, "authority_mode", "contact_scoped")
    ) or "contact_scoped"
    authority_default = 1 if authority_before == "yolo" else 0
    authority_index = prompt_choice(
        "  How much authority should Inkbox Voice AI have?",
        [
            "Contact-scoped — tools are limited to the current caller and conversation.",
            "YOLO mode — tools can use the identity's wider authorized capabilities.",
        ],
        authority_default,
    )
    authority_mode = "yolo" if authority_index == 1 else "contact_scoped"

    selected_authority_identity = authority_identity
    if authority_mode != authority_before:
        if selected_authority_identity is None:
            selected_authority_identity = _load_admin_identity(
                identity,
                base_url=base_url,
                Inkbox=Inkbox,
                InkboxAPIError=InkboxAPIError,
                WhoamiApiKeyResponse=WhoamiApiKeyResponse,
                ADMIN_SCOPED=ADMIN_SCOPED,
            )
        if selected_authority_identity is None:
            return False, authority_identity, ""
        if not callable(
            getattr(selected_authority_identity, "set_hosted_agent_authority_mode", None)
        ):
            print_error("  The installed Inkbox SDK cannot update Voice AI authority.")
            return False, authority_identity, ""

    hosted_changed = False
    authority_changed = False
    incoming_changed = False
    try:
        hosted_changed = True
        identity.set_hosted_agent_config(voice=None, model=None, instructions=None)
        if authority_mode != authority_before:
            authority_changed = True
            selected_authority_identity.set_hosted_agent_authority_mode(authority_mode)
        incoming_changed = True
        identity.set_incoming_call_action(
            incoming_call_action="hosted_agent",
            client_websocket_url=None,
            incoming_call_webhook_url=None,
        )
    except Exception as exc:
        print_error(f"  Inkbox Voice AI configuration failed: {exc}")
        _restore_voice_ai_config(
            identity,
            incoming_before=incoming_before,
            hosted_before=hosted_before,
            authority_before=authority_before,
            authority_identity=selected_authority_identity,
            incoming_changed=incoming_changed,
            authority_changed=authority_changed,
            hosted_changed=hosted_changed,
        )
        print_info("  The previous local voice-stack selection remains active.")
        return False, selected_authority_identity, ""

    return True, selected_authority_identity, authority_mode


def _save_voice_stack(
    stack: VoiceStack,
    *,
    realtime_api_key: str = "",
    authority_mode: str = "",
) -> None:
    if stack is VoiceStack.OPENAI_REALTIME:
        _save("INKBOX_REALTIME_API_KEY", realtime_api_key)
        _save("INKBOX_REALTIME_MODEL", OPENAI_REALTIME_TEST_MODEL)
        _save("INKBOX_REALTIME_ENABLED", "true")
    else:
        if authority_mode:
            _save("INKBOX_VOICE_AI_AUTHORITY_MODE", authority_mode)
        _save("INKBOX_REALTIME_ENABLED", "false")
    _save("INKBOX_VOICE_STACK", stack.value)


def _configure_phone_call_voice_stack(
    identity: Any,
    *,
    base_url: str,
    Inkbox: Any,
    InkboxAPIError: Any,
    WhoamiApiKeyResponse: Any,
    ADMIN_SCOPED: Any,
    imessage_enabled: bool = False,
    authority_identity: Any | None = None,
) -> None:
    phone = getattr(identity, "phone_number", None)
    if phone is None and not imessage_enabled:
        return

    prompt_enter_to_continue("  Press Enter to continue and set up phone call handling")
    print()
    print(color("  --- Phone call voice stack ---", Colors.CYAN))
    detected = _detect_openai_realtime_key()
    detected_key = detected[1] if detected is not None else ""
    default_index = _voice_stack_default_index(detected_key)
    prompt_for_realtime_key = False

    while True:
        selected_index = prompt_choice(
            "  Choose how this agent should handle phone calls:",
            [description for _, description in VOICE_STACK_CHOICES],
            default_index,
        )
        stack = VOICE_STACK_CHOICES[selected_index][0]
        default_index = selected_index

        if stack is VoiceStack.INKBOX_VOICE_AI:
            ok, authority_identity, authority_mode = _configure_voice_ai(
                identity,
                base_url=base_url,
                Inkbox=Inkbox,
                InkboxAPIError=InkboxAPIError,
                WhoamiApiKeyResponse=WhoamiApiKeyResponse,
                ADMIN_SCOPED=ADMIN_SCOPED,
                authority_identity=authority_identity,
            )
            if not ok:
                continue
            _save_voice_stack(stack, authority_mode=authority_mode)
            print_success("  Inkbox Voice AI is configured for phone calls.")
            print_info("  Hermes will be notified when each call ends.")
            return

        if stack is VoiceStack.OPENAI_REALTIME:
            if prompt_for_realtime_key or not detected_key:
                realtime_api_key = prompt(
                    "  Paste your OpenAI API key for Realtime calls",
                    password=True,
                ).strip()
            else:
                realtime_api_key = detected_key
                if detected is not None:
                    print_success(
                        f"  Found existing OpenAI API key in {_redact_key_source(detected[0])}."
                    )
            if not realtime_api_key:
                print_warning("  No OpenAI API key entered. Returning to the voice-stack choices.")
                prompt_for_realtime_key = True
                continue

            print_info(f"  Testing OpenAI Realtime access with {OPENAI_REALTIME_TEST_MODEL}...")
            ok, detail = _test_openai_realtime_api_key(
                realtime_api_key,
                OPENAI_REALTIME_TEST_MODEL,
            )
            if not ok:
                print_error("  OpenAI Realtime validation failed.")
                print_info(f"  {detail}")
                print_info("  Choose a voice stack again. Realtime was not enabled.")
                detected_key = ""
                prompt_for_realtime_key = True
                continue
            try:
                _set_local_incoming_call_action(identity)
            except Exception as exc:
                print_error(f"  Could not configure incoming Realtime calls: {exc}")
                print_info("  Choose a voice stack again. Realtime was not enabled.")
                detected_key = realtime_api_key
                prompt_for_realtime_key = False
                continue

            _save_voice_stack(stack, realtime_api_key=realtime_api_key)
            print_success("  OpenAI Realtime validation succeeded.")
            print_info("  OpenAI Realtime API is configured for phone calls.")
            return

        try:
            _set_local_incoming_call_action(identity)
        except Exception as exc:
            print_error(f"  Could not configure incoming Inkbox TTS/STT calls: {exc}")
            print_info("  Choose a voice stack again. The previous selection remains active.")
            continue
        _save_voice_stack(stack)
        print_success("  Inkbox TTS/STT is configured for phone calls.")
        return


def _setup_signing_key(api_key: str, base_url: str, Inkbox: Any) -> None:
    print()
    print(color("  --- Webhook signing key ---", Colors.CYAN))
    print_info("  Inkbox signs outbound webhooks with an HMAC over the body.")
    print_info("  Without the matching key, the gateway cannot verify inbound Inkbox traffic.")

    print_info("  A signing key is required to continue.")

    has_key = prompt_yes_no("  Do you already have an Inkbox signing key?", False)
    if has_key:
        key = prompt("  Paste your Inkbox signing key", password=True).strip()
        if key:
            _save("INKBOX_SIGNING_KEY", key)
            _save("INKBOX_REQUIRE_SIGNATURE", "true")
            print_success("  Saved signing key. Signature verification enabled.")
            return
        # An empty paste can't satisfy the requirement — fall through to mint one.
        print_warning("  No key entered; a signing key is required, so we'll mint one now.")

    print_info("  Minting a new key here rotates any existing key for your org.")
    print_info("  Any other gateway using the old key will fail verification until updated.")
    if not prompt_yes_no("  Generate a new signing key now?", True):
        print_error("  A signing key is required; cannot complete setup without one.")
        print_info("  Re-run setup and paste an existing key, or allow key generation.")
        raise SystemExit(1)

    try:
        new_key = Inkbox(**inkbox_client_kwargs(api_key, base_url)).create_signing_key()
    except Exception as exc:
        print_error(f"  Failed to create signing key: {exc}")
        print_error("  A signing key is required; aborting setup. Retry, or paste an existing key.")
        raise SystemExit(1)

    signing_key = str(getattr(new_key, "signing_key", "") or "")
    if not signing_key:
        print_error("  Signing-key response did not include signing_key.")
        print_error("  A signing key is required; aborting setup.")
        raise SystemExit(1)
    _save("INKBOX_SIGNING_KEY", signing_key)
    _save("INKBOX_REQUIRE_SIGNATURE", "true")
    created_at = getattr(new_key, "created_at", None)
    if created_at is not None and hasattr(created_at, "isoformat"):
        print_success(f"  Generated and saved signing key (created at {created_at.isoformat()}).")
    else:
        print_success("  Generated and saved signing key.")
    print_info("  Signature verification enabled.")


def _wait_for_sms_opt_in(api_key: str, base_url: str, phone: Any, Inkbox: Any) -> None:
    if phone is None or getattr(phone, "type", None) != "local":
        return
    phone_id = getattr(phone, "id", None)
    if phone_id is None:
        return

    def find_start(texts: Any) -> Any | None:
        for text in texts:
            direction = (getattr(text, "direction", "") or "").lower()
            body = (getattr(text, "text", "") or "").strip().upper()
            if direction == "inbound" and body == "START":
                return text
        return None

    try:
        client = Inkbox(**inkbox_client_kwargs(api_key, base_url))
    except Exception:
        return

    print()
    print(color("  --- Waiting for your START text ---", Colors.YELLOW))
    print_info(f"  Polling every 3s for an inbound START to {phone.number}.")
    print_info("  Without it, the agent cannot send outbound SMS to that phone later.")
    print_info("  Press Ctrl+C to skip; you can text START anytime.")

    spinner = "|/-\\"
    idx = 0
    next_poll_at = time.monotonic()
    clear_line = "\r" + " " * 72 + "\r"

    try:
        while True:
            now = time.monotonic()
            if now >= next_poll_at:
                try:
                    texts = client.texts.list(phone_id, limit=20)
                except Exception:
                    texts = []
                match = find_start(texts)
                if match is not None:
                    remote = getattr(match, "remote_phone_number", "")
                    sys.stdout.write(clear_line)
                    sys.stdout.flush()
                    print_success(f"  Got it. SMS opt-in confirmed from {remote}")
                    return
                next_poll_at = now + 3.0
            sys.stdout.write(f"\r  {spinner[idx]} Listening for START...  ")
            sys.stdout.flush()
            idx = (idx + 1) % len(spinner)
            time.sleep(0.25)
    except KeyboardInterrupt:
        sys.stdout.write(clear_line)
        sys.stdout.flush()
        print()
        print_warning(f"  Skipped. Text START to {phone.number} anytime to enable outbound SMS.")


def _avatar_base_url(base_url: str) -> str:
    return (base_url or _RAW_AVATAR_BASE_URL_DEFAULT).rstrip("/")


async def _identity_has_avatar_async(base_url: str, api_key: str, handle: str) -> bool | None:
    """Check whether an identity already has a contact-card avatar."""
    import aiohttp

    url = f"{_avatar_base_url(base_url)}/api/v1/identities/{handle}/avatar"
    timeout = aiohttp.ClientTimeout(total=10)
    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.get(url, headers={"X-API-Key": api_key}) as resp:
                if resp.status == 200:
                    return True
                if resp.status == 404:
                    return False
                return None
    except Exception:
        return None


async def _upload_avatar_async(
    base_url: str, api_key: str, handle: str, image: bytes
) -> tuple[bool, str]:
    """PUT the Hermes avatar image to the identity's avatar endpoint."""
    import aiohttp

    url = f"{_avatar_base_url(base_url)}/api/v1/identities/{handle}/avatar"
    timeout = aiohttp.ClientTimeout(total=30)
    form = aiohttp.FormData()
    form.add_field("file", image, filename="hermes_with_iphone.png", content_type="image/png")
    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.put(url, headers={"X-API-Key": api_key}, data=form) as resp:
                if resp.status in (200, 201):
                    return True, "ok"
                return False, f"HTTP {resp.status} {(await resp.text())[:200]}"
    except Exception as exc:
        return False, str(exc)


def _identity_has_avatar(base_url: str, api_key: str, handle: str) -> bool | None:
    try:
        return asyncio.run(_identity_has_avatar_async(base_url, api_key, handle))
    except RuntimeError:
        return None


def _upload_avatar(base_url: str, api_key: str, handle: str, image: bytes) -> tuple[bool, str]:
    try:
        return asyncio.run(_upload_avatar_async(base_url, api_key, handle, image))
    except RuntimeError as exc:
        return False, f"could not run avatar upload from this setup process: {exc}"


def _configure_avatar(base_url: str, api_key: str, identity: Any, *, is_signup: bool) -> None:
    """Attach the bundled Hermes avatar to the agent's Inkbox contact card."""
    handle = getattr(identity, "agent_handle", "") or ""
    if not handle or not _AVATAR_PATH.exists():
        return

    if not is_signup:
        if _identity_has_avatar(base_url, api_key, handle) is True:
            return
        print()
        print(color("  --- Agent avatar ---", Colors.CYAN))
        print_info("  This agent has no avatar on its Inkbox contact card.")
        if not prompt_yes_no("  Add the Hermes avatar?", True):
            print_info("  Skipped. You can set an avatar later in the Inkbox console.")
            return

    try:
        image = _AVATAR_PATH.read_bytes()
    except Exception as exc:
        print_warning(f"  Could not read the bundled avatar: {exc}")
        return

    ok, detail = _upload_avatar(base_url, api_key, handle, image)
    if ok:
        print_success("  Attached the Hermes avatar to this agent.")
    else:
        print_warning(f"  Could not attach the avatar: {detail}")
        print_info("  You can set one later in the Inkbox console.")


def _configure_imessage(api_key: str, base_url: str, handle: str, Inkbox: Any) -> bool:
    """Offer to enable iMessage for the agent and walk through connecting.

    Args:
        api_key (str): The agent-scoped Inkbox API key the wizard saved.
        base_url (str): Inkbox API base URL.
        handle (str): Agent identity handle being configured.
        Inkbox (Any): The Inkbox SDK client class.

    Returns:
        bool: True when iMessage ended up enabled (newly or already), so the
        caller can gate iMessage-dependent steps like realtime calling.
    """
    print()
    print(color("  --- iMessage ---", Colors.CYAN))
    print_info("  Inkbox can make this agent reachable over iMessage from your iPhone.")
    print_info("  No number to provision — you connect through the Inkbox iMessage router.")
    print_info("  Once connected, the agent can also make and take voice calls with you")
    print_info("  over that same shared iMessage line.")

    try:
        client = Inkbox(**inkbox_client_kwargs(api_key, base_url))
        identity = client.get_identity(handle)
    except Exception as exc:
        print_warning(f"  Could not load the identity for iMessage setup: {exc}")
        return False

    # Old SDKs predate iMessage entirely — detect by surface, not version.
    if not hasattr(client, "imessages") or not hasattr(identity, "imessage_enabled"):
        print_warning("  The installed Inkbox SDK does not support iMessage yet.")
        print_info("  Upgrade it and rerun setup:")
        print_info(f"    {_install_command_text()}")
        return False

    if identity.imessage_enabled:
        print_success("  iMessage is already enabled for this agent.")
    else:
        if not prompt_yes_no("  Enable iMessage for this agent?", True):
            print_info("  Skipped. Rerun `hermes inkbox setup` anytime to enable iMessage.")
            return False
        try:
            identity.update(imessage_enabled=True)
        except Exception as exc:
            print_error(f"  Could not enable iMessage: {exc}")
            print_info("  You can enable it later from the Inkbox console and rerun setup.")
            return False
        print_success("  iMessage enabled for this agent.")
        try:
            # Re-fetch so the local object reflects the new flag (the SDK
            # gates its iMessage helpers on it).
            identity = client.get_identity(handle)
        except Exception as exc:
            print_warning(f"  Could not refresh the identity after enabling: {exc}")
            return True

    # Surface phones already connected through the router so reruns don't
    # read like a first-time setup, and default the walkthrough off when a
    # connection already exists (connecting another phone is the rare case).
    connected = []
    list_assignments = getattr(identity, "list_imessage_assignments", None)
    if callable(list_assignments):
        try:
            connected = list(list_assignments(limit=5))
        except Exception:
            connected = []
        if connected:
            numbers = ", ".join(
                str(getattr(a, "remote_number", "") or "") for a in connected
            )
            print_success(f"  Already connected: {numbers}")

    question = (
        "  Connect another iPhone to this agent now?"
        if connected
        else "  Connect your iPhone to this agent now?"
    )
    if not prompt_yes_no(question, not connected):
        print_info("  You can connect anytime — rerun `hermes inkbox setup` for the walkthrough.")
        return True
    _wait_for_imessage_first_message(client, identity, handle)
    return True


def _wait_for_imessage_first_message(client: Any, identity: Any, handle: str) -> None:
    """Walk the user through the iMessage connect flow and greet them back.

    Args:
        client (Any): Authenticated Inkbox SDK client (agent-scoped key).
        identity (Any): The iMessage-enabled agent identity object.
        handle (str): Agent identity handle, used in the welcome message.

    Returns:
        None: Polls until the first inbound iMessage arrives (Ctrl+C skips),
        then sends the channel-introduction reply into that conversation.
    """
    from datetime import datetime, timezone

    try:
        triage = client.imessages.get_triage_number()
    except Exception as exc:
        print_warning(f"  Could not fetch the iMessage router number: {exc}")
        print_info("  Rerun `hermes inkbox setup` later to finish connecting.")
        return

    connect_command = str(getattr(triage, "connect_command", "") or "").strip()
    if not connect_command or "your-handle" in connect_command:
        connect_command = f"connect @{handle}"

    print()
    print_info("  From your iPhone, in the Messages app:")
    print(color(f"      1. Text \"{connect_command}\" to {triage.number}", Colors.BOLD))
    print_info("    2. Inkbox texts you back from the number now assigned to this agent.")
    print_info("    3. Send any first message (e.g. \"hi\") in that NEW thread.")
    print_info("  The agent can only message you after you message it first.")

    sms_link = str(getattr(triage, "sms_link", "") or "").strip()
    if not sms_link or "your-handle" in sms_link:
        sms_link = f"sms:{triage.number}?&body={quote(connect_command)}"

    qr_payload = f"SMSTO:{triage.number}:{connect_command}"
    print()
    print_info("  Or just scan this with your iPhone camera to do step 1 in one tap:")
    print()
    if not _show_qr(qr_payload):
        print_info(f"    (install 'segno' to show a scannable QR here: {sms_link})")

    print()
    print(color("  --- Waiting for your first iMessage ---", Colors.YELLOW))
    print_info("  Polling every 3s for an inbound iMessage to this agent.")
    print_info("  Press Ctrl+C to skip; you can connect anytime.")

    started_at = datetime.now(timezone.utc)

    def find_first_inbound(messages: Any) -> Any | None:
        for message in messages:
            direction = (_enum_value(getattr(message, "direction", "")) or "").lower()
            if direction != "inbound":
                continue
            created_at = getattr(message, "created_at", None)
            # Ignore traffic from a connection that predates this run.
            # Naive timestamps can't be compared to the aware cutoff —
            # accept those rather than crash the poll loop.
            if (
                created_at is not None
                and created_at.tzinfo is not None
                and created_at < started_at
            ):
                continue
            return message
        return None

    spinner = "|/-\\"
    idx = 0
    next_poll_at = time.monotonic()
    clear_line = "\r" + " " * 72 + "\r"
    match = None

    try:
        while match is None:
            now = time.monotonic()
            if now >= next_poll_at:
                try:
                    messages = identity.list_imessages(limit=10)
                except Exception:
                    messages = []
                match = find_first_inbound(messages)
                next_poll_at = now + 3.0
            if match is None:
                sys.stdout.write(f"\r  {spinner[idx]} Listening for your first iMessage...  ")
                sys.stdout.flush()
                idx = (idx + 1) % len(spinner)
                time.sleep(0.25)
    except KeyboardInterrupt:
        sys.stdout.write(clear_line)
        sys.stdout.flush()
        print()
        print_warning("  Skipped. The agent replies over iMessage once you connect and message it.")
        return

    sys.stdout.write(clear_line)
    sys.stdout.flush()
    remote = getattr(match, "remote_number", "") or "your phone"
    print_success(f"  Got it. First iMessage received from {remote}.")

    conversation_id = getattr(match, "conversation_id", None)
    welcome = (
        f"You're connected! This is your iMessage channel to your Hermes agent "
        f"@{handle}. Anything you send here goes straight to the agent, and its "
        f"replies will show up right in this thread."
    )
    try:
        identity.send_imessage(conversation_id=conversation_id, text=welcome)
        print_success("  Sent a welcome message back on that thread.")
    except Exception as exc:
        print_warning(f"  Could not send the welcome message: {exc}")
    try:
        # Clear the unread flag the walkthrough message left behind.
        identity.mark_imessage_conversation_read(conversation_id)
    except Exception:
        pass
    # The wizard's closing step offers to start or restart the gateway, so
    # point at that rather than handing out commands mid-flow.
    print_info("  Keep chatting in that thread. Setup finishes by bringing the")
    print_info("  gateway up so it picks up this new iMessage connection.")


def _self_signup_flow(base_url: str, Inkbox: Any, InkboxAPIError: Any) -> tuple[Any | None, str, bool]:
    print()
    print_info("No problem. We will create a fresh agent identity for you.")
    print_info("You will get an Inkbox-hosted mailbox plus an API key.")
    print_info("A short verification email goes to you to claim full capabilities.")
    print()

    note = "Setting up a Hermes agent on Inkbox."
    human_email = ""
    handle = ""

    while True:
        if not human_email:
            human_email = prompt("  Your email address (for the verification step)").strip()
            if not human_email or "@" not in human_email:
                print_error("  A valid email address is required for signup.")
                return None, "", False

        if not handle:
            handle = prompt(
                "  Desired agent handle (e.g. on-call-agent, recruiting-agent) - "
                "globally unique, also becomes the mailbox local part"
            ).strip()
            if not handle:
                print_error("  Agent handle is required.")
                return None, "", False

        print()
        print_info("Calling agent-signup...")
        try:
            resp = Inkbox.signup(
                human_email=human_email,
                note_to_human=note,
                agent_handle=handle,
                harness="hermes",
                **inkbox_base_url_kwargs(base_url),
            )
            break
        except InkboxAPIError as exc:
            detail = _error_detail(exc)
            detail_lc = detail.lower()
            status = _error_status(exc)
            print_error(f"  Signup failed: HTTP {status} {detail}")

            if status == 429 and "unclaimed agents" in detail_lc:
                ctx = (
                    f"Signup blocked: HTTP 429 - {detail}\n\n"
                    "Inkbox caps unclaimed agents per human email. To free a slot,\n"
                    "verify one of your existing unclaimed agents in the Inkbox console,\n"
                    "or use a different email below.\n\n"
                    f"Email tried: {human_email}"
                )
                if _retry_or_abort("Try a different email", error_context=ctx):
                    human_email = ""
                    continue
                return None, "", False

            if status == 409 or (status == 422 and ("handle" in detail_lc or "unavailable" in detail_lc)):
                ctx = (
                    f"Signup blocked: HTTP {status} - {detail}\n\n"
                    "Agent handles are globally unique. Pick another.\n\n"
                    f"Handle tried: {handle}"
                )
                if _retry_or_abort("Pick a different handle", error_context=ctx):
                    handle = ""
                    continue
                return None, "", False

            ctx = (
                f"Signup failed: HTTP {status} - {detail}\n\n"
                "Inputs tried:\n"
                f"  Email:  {human_email}\n"
                f"  Handle: {handle}"
            )
            if _retry_or_abort("Re-enter all details and try again", error_context=ctx):
                human_email = handle = ""
                continue
            return None, "", False
        except Exception as exc:
            print_error(f"  Signup failed: {exc}")
            return None, "", False

    print_success(f"Agent created - mailbox: {resp.email_address}")
    print_info(f"  Handle: {resp.agent_handle}")
    print_info(f"  A verification email was sent to {human_email}.")
    print_info("  Enter the 6-digit code from that email to claim the agent.")
    print()

    max_attempts = 3
    attempts_used = 0
    while True:
        attempts_left = max_attempts - attempts_used
        if attempts_left <= 0:
            prompt_text = "  Type 'resend' for a new code (Ctrl+C to abort)"
        else:
            prompt_text = f"  Verification code, or 'resend' for a new email ({attempts_left}/{max_attempts} attempts left)"

        entry = prompt(prompt_text).strip()
        if entry.lower() in {"resend", "r"}:
            if _try_resend(Inkbox, InkboxAPIError, resp.api_key, base_url, human_email):
                attempts_used = 0
            continue
        if not entry:
            print_warning("  Type the 6-digit code, or 'resend' for a fresh email.")
            continue
        if attempts_left <= 0:
            print_warning("  This code is dead. Type 'resend' before trying another code.")
            continue
        try:
            verify = Inkbox.verify_signup(
                api_key=resp.api_key,
                verification_code=entry,
                **inkbox_base_url_kwargs(base_url),
            )
            print_success(f"  Verified - claim status: {verify.claim_status}")
            break
        except InkboxAPIError as exc:
            attempts_used += 1
            print_error(
                f"  Wrong code: HTTP {_error_status(exc)} {_error_detail(exc)} "
                f"({attempts_used}/{max_attempts} attempts used)"
            )
            if attempts_used >= max_attempts:
                print_warning("  This code is now dead. Type 'resend' for a fresh one.")
        except Exception as exc:
            print_error(f"  Verification failed: {exc}")

    # Phone provisioning is decoupled from signup: the wizard offers a
    # dedicated number as a standalone step AFTER iMessage setup (see
    # ``interactive_setup``), so a fresh identity starts with no number here.
    class MailboxShim:
        email_address = resp.email_address
        display_name = None

    class SignupIdentityShim:
        agent_handle = resp.agent_handle
        email_address = resp.email_address
        mailbox = MailboxShim()
        phone_number = None

    return SignupIdentityShim(), resp.api_key, False


def _retry_or_abort(retry_label: str, *, error_context: str = "") -> bool:
    print()
    choice = prompt_choice(
        "  What now?",
        [retry_label, "Abort - keep existing Inkbox configuration unchanged"],
        0,
        description=error_context or None,
    )
    if choice == 0:
        return True
    print_warning("  Aborted. No credentials saved.")
    print_info("  Your existing INKBOX_IDENTITY in .env is unchanged.")
    return False


def _try_resend(Inkbox: Any, InkboxAPIError: Any, api_key: str, base_url: str, human_email: str) -> bool:
    try:
        Inkbox.resend_signup_verification(api_key=api_key, **inkbox_base_url_kwargs(base_url))
        print_success(f"  Resent. Check {human_email}.")
        return True
    except InkboxAPIError as exc:
        print_warning(f"  Resend failed: HTTP {_error_status(exc)} {_error_detail(exc)}")
        if _error_status(exc) == 429:
            print_info("  Wait out the cooldown before trying again.")
        return False
    except Exception as exc:
        print_warning(f"  Resend failed: {exc}")
        return False


def _api_key_flow(
    base_url: str,
    Inkbox: Any,
    InkboxAPIError: Any,
    WhoamiApiKeyResponse: Any,
    ADMIN_SCOPED: Any,
    AGENT_CLAIMED: Any,
    AGENT_UNCLAIMED: Any,
    IdentityPhoneNumberCreateOptions: Any,
) -> tuple[Any | None, str, bool, Any | None]:
    print()
    api_key = prompt("  Paste your Inkbox API key (ApiKey_...)", password=True).strip()
    if not api_key:
        print_error("  No key provided.")
        return None, "", False, None

    try:
        client = Inkbox(**inkbox_client_kwargs(api_key, base_url))
        info = client.whoami()
    except InkboxAPIError as exc:
        print_error(f"  whoami failed: HTTP {_error_status(exc)} {_error_detail(exc)}")
        print_info("  Double-check the key and the environment it was issued in.")
        return None, "", False, None
    except Exception as exc:
        print_error(f"  whoami failed: {exc}")
        return None, "", False, None

    if WhoamiApiKeyResponse is not None and not isinstance(info, WhoamiApiKeyResponse):
        print_error("  This wizard requires an API key, but the credential is a JWT.")
        return None, "", False, None

    subtype = _enum_value(getattr(info, "auth_subtype", ""))
    org_id = getattr(info, "organization_id", "")
    print_success(f"  Key validated - org {org_id}, scope: {subtype or 'unknown'}")

    if subtype in {_enum_value(AGENT_CLAIMED), _enum_value(AGENT_UNCLAIMED)}:
        identity, agent_key, did_provision_phone = _pick_agent_scoped(client, api_key)
        return identity, agent_key, did_provision_phone, None
    if subtype == _enum_value(ADMIN_SCOPED):
        identity, agent_key, did_provision_phone = _pick_admin_scoped(
            client,
            api_key,
            IdentityPhoneNumberCreateOptions,
            InkboxAPIError,
        )
        return identity, agent_key, did_provision_phone, identity

    print_error(f"  Unsupported API-key subtype: {subtype!r}.")
    print_info("  Use an admin-scoped or agent-scoped Inkbox API key.")
    return None, "", False, None


def _pick_agent_scoped(client: Any, api_key: str) -> tuple[Any | None, str, bool]:
    try:
        identities = list(client.list_identities())
    except Exception as exc:
        print_error(f"  list_identities failed: {exc}")
        return None, "", False

    if not identities:
        print_error("  Agent-scoped key but no identity returned.")
        return None, "", False
    if len(identities) > 1:
        print_warning(f"  Agent-scoped key returned {len(identities)} identities; using the first.")

    summary = identities[0]
    try:
        identity = client.get_identity(summary.agent_handle)
    except Exception as exc:
        print_error(f"  get_identity failed: {exc}")
        return None, "", False

    print()
    print_info(f"  This API key is bound to identity: {identity.agent_handle}")
    return identity, api_key, False


def _mint_agent_scoped_key(client: Any, identity: Any, InkboxAPIError: Any) -> str | None:
    try:
        created = client.api_keys.create(
            label=f"Hermes gateway - {identity.agent_handle}",
            description=(
                "Auto-minted by hermes inkbox setup. Scoped to one agent "
                "identity so the gateway never stores the admin key."
            ),
            scoped_identity_id=identity.id,
        )
    except InkboxAPIError as exc:
        print_error(f"  Could not mint agent-scoped key: HTTP {_error_status(exc)} {_error_detail(exc)}")
        return None
    except Exception as exc:
        print_error(f"  Could not mint agent-scoped key: {exc}")
        return None
    return str(getattr(created, "api_key", "") or "") or None


def _pick_admin_scoped(
    client: Any,
    api_key: str,
    IdentityPhoneNumberCreateOptions: Any,
    InkboxAPIError: Any,
) -> tuple[Any | None, str, bool]:
    try:
        identities = list(client.list_identities())
    except Exception as exc:
        print_error(f"  list_identities failed: {exc}")
        return None, "", False

    print()
    if identities:
        print_info(f"  Found {len(identities)} identity(ies). Fetching mailbox and phone details.")
        full_records: list[Any | None] = []
        for summary in identities:
            try:
                full_records.append(client.get_identity(summary.agent_handle))
            except Exception as exc:
                print_warning(f"    {summary.agent_handle}: details unavailable ({exc})")
                full_records.append(None)

        choices = []
        for summary, full in zip(identities, full_records):
            mailbox_str = (getattr(full, "email_address", None) if full else None) or getattr(summary, "email_address", None) or "no mailbox"
            phone_str = "no phone"
            phone = getattr(full, "phone_number", None) if full else None
            if phone is not None:
                phone_str = getattr(phone, "number", "phone")
            choices.append(f"{summary.agent_handle}  -  {mailbox_str}  -  {phone_str}")
        choices.append("Create a new identity")

        idx = prompt_choice("  Select the identity this Hermes gateway should run as:", choices, 0)
        if idx < len(identities):
            identity = full_records[idx]
            if identity is None:
                try:
                    identity = client.get_identity(identities[idx].agent_handle)
                except Exception as exc:
                    print_error(f"  get_identity failed: {exc}")
                    return None, "", False
            agent_key = _mint_agent_scoped_key(client, identity, InkboxAPIError)
            if agent_key is None:
                return None, "", False
            return identity, agent_key, False
    else:
        print_info("  No identities exist yet under this org. Let's create the first one.")

    identity, _, did_provision_phone = _create_identity(
        client,
        api_key,
        IdentityPhoneNumberCreateOptions,
        InkboxAPIError,
    )
    if identity is None:
        return None, "", False
    agent_key = _mint_agent_scoped_key(client, identity, InkboxAPIError)
    if agent_key is None:
        return None, "", False
    return identity, agent_key, did_provision_phone


def _create_identity(
    client: Any,
    api_key: str,
    IdentityPhoneNumberCreateOptions: Any,
    InkboxAPIError: Any,
) -> tuple[Any | None, str, bool]:
    del api_key
    try:
        from inkbox.identities.exceptions import HandleUnavailableError
    except Exception:  # pragma: no cover - old SDK fallback
        HandleUnavailableError = ()

    print()
    print_header("Create new agent identity")

    while True:
        handle = prompt(
            "  Agent handle (e.g. on-call-agent, recruiting-agent) - "
            "globally unique, also the mailbox local part"
        ).strip()
        if handle:
            break
        print_error("  Handle is required.")

    display_name = prompt("  Display name for the identity (shown to recipients, optional)").strip()

    # Phone provisioning is decoupled from creation: the wizard offers a
    # dedicated number as a standalone step AFTER iMessage setup (see
    # ``interactive_setup``).  ``IdentityPhoneNumberCreateOptions`` is kept in
    # the signature for call-site compatibility but no longer used here.
    del IdentityPhoneNumberCreateOptions

    print()
    print_info("Creating identity...")
    while True:
        try:
            identity = client.create_identity(
                handle,
                display_name=display_name or None,
                phone_number=None,
            )
            break
        except HandleUnavailableError as exc:
            namespace = getattr(exc, "blocking_namespace", None) or "the global namespace"
            print_error(f"  Handle '{handle}' is unavailable in {namespace}.")
            handle = prompt("  Pick a different handle").strip()
            if not handle:
                return None, "", False
        except InkboxAPIError as exc:
            print_error(f"  Creation failed: HTTP {_error_status(exc)} {_error_detail(exc)}")
            return None, "", False
        except Exception as exc:
            print_error(f"  Creation failed: {exc}")
            return None, "", False

    print_success(f"  Created identity '{identity.agent_handle}'")
    return identity, "", False


def _offer_dedicated_number(client: Any, identity: Any) -> tuple[Any, bool]:
    """Offer to provision a dedicated phone number (SMS + voice).

    Runs as a standalone step AFTER iMessage setup so the wizard walks
    channels in a natural order: connect over iMessage first, then add a
    dedicated number.  Returns ``(possibly-refreshed identity, provisioned?)``;
    a no-op when the identity already has a number.
    """
    existing = getattr(identity, "phone_number", None)
    if existing is not None:
        # Say so instead of silently skipping — otherwise the step looks lost.
        print()
        print(color("  --- Dedicated phone number ---", Colors.CYAN))
        print_success(f"  Already provisioned: {existing.number}")
        return identity, False

    print()
    print(color("  --- Dedicated phone number ---", Colors.CYAN))
    print_info("  A local US number gives this agent its own line for SMS and voice.")
    if not prompt_yes_no("  Provision a dedicated phone number now?", True):
        print_info("  Skipped. Rerun `hermes inkbox setup` anytime to add a number.")
        return identity, False

    try:
        provisioned = client.phone_numbers.provision(agent_handle=identity.agent_handle, type="local")
        print_success(f"  Provisioned: {provisioned.number}")
    except Exception as exc:
        # Graceful fallback — most rejections here are plan gating. Point at
        # pricing and keep the wizard moving; nothing downstream needs a number.
        print_info("  Dedicated phone numbers are available on Inkbox paid tiers —")
        print_info("  see https://inkbox.ai/pricing for details.")
        print_info(f"  (provisioning response: {exc})")
        return identity, False

    try:
        return client.get_identity(identity.agent_handle), True
    except Exception:
        class PhoneShim:
            def __init__(self, phone: Any):
                self.number = phone.number
                self.type = getattr(phone, "type", "local")
                self.sms_status = getattr(phone, "sms_status", None)
                self.id = getattr(phone, "id", None)

        identity.phone_number = PhoneShim(provisioned)
        return identity, True


def _print_agent_summary(identity: Any) -> None:
    print()
    print(color("Inkbox configured", Colors.GREEN, Colors.BOLD))
    print()
    print(color(f"  Handle:   {identity.agent_handle}", Colors.GREEN, Colors.BOLD))

    mailbox = getattr(identity, "mailbox", None)
    email = getattr(identity, "email_address", None) or (getattr(mailbox, "email_address", None) if mailbox else None)
    if email:
        print(color(f"  Mailbox:  {email}", Colors.GREEN, Colors.BOLD))
    else:
        print_info("  Mailbox:  (none - set up later in the Inkbox console)")

    phone = getattr(identity, "phone_number", None)
    if phone is not None:
        sms_status = getattr(phone, "sms_status", None)
        sms_value = getattr(sms_status, "value", sms_status)
        sms_str = f" - SMS: {sms_value}" if sms_value else ""
        print(color(f"  Phone:    {phone.number} ({phone.type}){sms_str}", Colors.GREEN, Colors.BOLD))
        if sms_value == "pending":
            print_info("            (carrier propagation can take a few minutes)")
    else:
        print_info("  Phone:    (none - provision later in the Inkbox console)")

    print()
    # The gateway (re)start is offered as the wizard's closing step, so don't
    # tell the operator to run it here as well.
    print_info("  Wrote INKBOX_API_KEY and INKBOX_IDENTITY to .env.")

    if phone is not None and getattr(phone, "type", None) == "local":
        print()
        print(color("  --- SMS opt-in ---", Colors.YELLOW))
        print_info(f"  Text START to {phone.number} to enable SMS from this agent")
        print_info("  to your phone. Do this from every phone you want to message it from.")

        qr_payload = f"SMSTO:{phone.number}:START"
        sms_link = f"sms:{phone.number}?&body=START"
        print()
        print_info("  Or just scan this with your phone camera to draft that text in one tap:")
        print()
        if not _show_qr(qr_payload):
            print_info(f"    (install 'segno' to show a scannable QR here: {sms_link})")

    print()
    print(color("  --- Reachability rules ---", Colors.CYAN))
    print_info("  Open the Inkbox console to control who can reach this agent:")
    print_info("    https://inkbox.ai/console/contact-rules")
    print_info("  You can allow or block specific contacts, phone numbers, email addresses, and domains.")


def _gateway_runtime_state() -> tuple[bool | None, bool]:
    """Detect whether a Hermes gateway is live for the current profile.

    Returns:
        tuple[bool | None, bool]: ``(running, service_installed)``. ``running``
        is None when neither liveness source is importable (plugin loaded
        outside a Hermes install), in which case ``service_installed`` is False.
    """
    # Preferred: the CLI's unified snapshot — covers systemd/launchd/s6 service
    # slots as well as a bare `hermes gateway run` process.
    try:
        from hermes_cli.gateway import get_gateway_runtime_snapshot

        snapshot = get_gateway_runtime_snapshot()
        return bool(snapshot.running), bool(snapshot.service_installed)
    except Exception:
        pass
    return None, False


def _run_gateway_command(action: str) -> bool:
    """Run ``hermes gateway <action>`` in a subprocess.

    Args:
        action (str): Gateway subcommand to run - "restart", "start" or
            "install".

    Returns:
        bool: True when the command exited 0. The Hermes profile carries over
        via the inherited HERMES_HOME the CLI exported for this process.
    """
    hermes = shutil.which("hermes")
    command = (
        [hermes, "gateway", action]
        if hermes
        else [sys.executable, "-m", "hermes_cli.main", "gateway", action]
    )
    # `install` asks its own questions on this terminal — don't put a clock on
    # the operator. The unattended actions keep the bound.
    timeout = None if action == "install" else GATEWAY_COMMAND_TIMEOUT
    try:
        subprocess.check_call(command, timeout=timeout)
        return True
    except Exception as exc:
        print_error(f"  `hermes gateway {action}` failed: {exc}")
        return False


def _print_ready_banner(handle: str) -> None:
    """Print the closing banner for a setup that ended with a live gateway.

    Args:
        handle (str): Inkbox agent identity the gateway is now running as.

    Returns:
        None: Replaces the next-steps list - there is nothing left to run.
    """
    rows = (("Inkbox identity", handle), ("Check its health", "hermes inkbox doctor"))
    label_width = max(len(label) for label, _ in rows) + 1  # +1 for the colon
    body = [
        "Your Hermes agent is set up and running on Inkbox.",
        "",
        *(f"  {(label + ':').ljust(label_width)}  {value}" for label, value in rows),
    ]
    width = max(len(line) for line in body) + 4
    print()
    print(color("╭" + "─" * (width - 2) + "╮", Colors.GREEN))
    for line in body:
        print(
            color("│ ", Colors.GREEN)
            + color(line.ljust(width - 4), Colors.GREEN, Colors.BOLD)
            + color(" │", Colors.GREEN)
        )
    print(color("╰" + "─" * (width - 2) + "╯", Colors.GREEN))


def _running_gateway_pids() -> tuple[int, ...]:
    """Return the PIDs of every gateway process visible for this profile.

    Returns:
        tuple[int, ...]: Gateway PIDs, empty when none are visible or the CLI
        cannot be asked. The snapshot deduplicates, so one gateway is one PID.
    """
    try:
        from hermes_cli.gateway import get_gateway_runtime_snapshot

        return tuple(get_gateway_runtime_snapshot().gateway_pids)
    except Exception:
        return ()


def _warn_if_multiple_gateways() -> bool:
    """Warn when more than one gateway is alive for this profile.

    Only one process can hold the Inkbox platform lock and the adapter's listen
    port, so the extras start with Inkbox dead - and the survivor may predate
    the config this wizard just wrote.

    Returns:
        bool: True when a warning was printed.
    """
    pids = _running_gateway_pids()
    if len(pids) < 2:
        return False
    print()
    print_warning(f"  Found {len(pids)} running gateway processes: {', '.join(str(p) for p in pids)}")
    print_info("  Only one can own this Inkbox identity - the rest start with")
    print_info("  Inkbox dead (platform lock held, or listen port already in use),")
    print_info("  and the one that wins may predate the config just written.")
    print_info("  Clear them out with: hermes gateway stop --all")
    return True


def _wait_for_gateway_running(timeout: float = GATEWAY_START_CONFIRM_TIMEOUT) -> bool:
    """Poll for gateway liveness after a start or install.

    Args:
        timeout (float): Seconds to keep polling before giving up.

    Returns:
        bool: True once a gateway reports running. A single immediate probe
        races a gateway that is still coming up, so poll rather than ask once.
    """
    deadline = time.monotonic() + timeout
    while True:
        running, _ = _gateway_runtime_state()
        if running:
            return True
        if time.monotonic() >= deadline:
            return False
        time.sleep(1.0)


def _offer_gateway_restart() -> bool:
    """Offer to restart (or start) the gateway so the new config takes effect.

    Returns:
        bool: True when the closing next-steps list should NOT tell the
        operator to start a gateway - either one is confirmed running, or a
        start/install we ran reported success and a second one would duplicate
        it.
    """
    print()
    print(color("  --- Hermes gateway ---", Colors.CYAN))

    # Hermes refuses a self-targeting restart (it would loop against the
    # supervisor's KeepAlive), so don't offer one we can't actually run.
    if os.getenv("_HERMES_GATEWAY") == "1":
        print_info("  Setup is running inside the gateway process.")
        print_info("  Restart it from a shell: hermes gateway restart")
        return True

    running, service_installed = _gateway_runtime_state()

    if running is None:
        print_info("  Could not tell whether a gateway is running.")
        print_info("  Restart a running one with `hermes gateway restart` so it")
        print_info("  picks up this config, or start one with `hermes gateway run`.")
        return False

    if running:
        print_success("  Detected a running Hermes gateway.")
        print_info("  It is still on the old config - Inkbox only takes effect")
        print_info("  once the gateway restarts.")
        _warn_if_multiple_gateways()
        if not prompt_yes_no("  Restart the gateway now?", True):
            print_warning("  Skipped. Run `hermes gateway restart` before using Inkbox.")
            return True
        print_info("  Restarting...")
        if not _run_gateway_command("restart"):
            print_info("  Restart it manually: hermes gateway restart")
            return True
        if not _warn_if_multiple_gateways():
            print_success("  Gateway restarted with the new Inkbox config.")
        return True

    print_warning("  Did not detect a running Hermes gateway - Inkbox is")
    print_warning("  configured, but nothing is listening for it yet.")

    if service_installed:
        if not prompt_yes_no("  Launch the gateway now?", True):
            print_info("  Skipped. Run `hermes gateway start` when you're ready.")
            return False
        print_info("  Starting...")
        if not _run_gateway_command("start"):
            print_info("  Start it manually: hermes gateway start")
            return False
        # Exit 0 only means the command ran. Confirm a gateway is actually up,
        # and that a stale sibling is not holding the Inkbox lock against it.
        if _wait_for_gateway_running():
            if not _warn_if_multiple_gateways():
                print_success("  Gateway started with the new Inkbox config.")
            return True
        print_success("  Gateway start completed.")
        print_info("  Could not confirm the gateway came up within "
                   f"{int(GATEWAY_START_CONFIRM_TIMEOUT)}s.")
        print_info("  Check it with: hermes gateway status")
        return True

    # No service slot yet: `install` sets one up under systemd/launchd/Windows
    # and asks its own start-now question, so it covers install-and-launch in
    # one step. It exits non-zero on platforms with no service manager (Termux,
    # bare WSL) — fall back to the foreground command there.
    print_info("  Installing the background service keeps it running across reboots.")
    if not prompt_yes_no("  Install and launch the gateway service now?", True):
        print_info("  Skipped. Run `hermes gateway install`, or `hermes gateway run`")
        print_info("  to start one in the foreground.")
        return False
    print_info("  Installing...")
    if not _run_gateway_command("install"):
        print_info("  Start one in the foreground instead: hermes gateway run")
        return False

    # `install` brings the gateway up in whatever way the platform allows - a
    # service slot, or a plain background process when the service manager
    # refuses the bootstrap. Either way it needs a moment to become visible.
    if _wait_for_gateway_running():
        if not _warn_if_multiple_gateways():
            print_success("  Gateway installed and running with the new Inkbox config.")
        return True

    # Install reported success, so never answer it with "now go start one":
    # if it did bring a gateway up, a second start would duplicate it.
    print_success("  Gateway install completed.")
    print_info("  Could not confirm the gateway came up within "
               f"{int(GATEWAY_START_CONFIRM_TIMEOUT)}s.")
    print_info("  Check it with: hermes gateway status")
    return True


def _configure_channel_display() -> None:
    """Keep terminal reasoning out of channel replies unless explicitly enabled."""
    from hermes_cli.config import load_config, save_config

    config = load_config()
    display = config.setdefault("display", {})
    platforms = display.setdefault("platforms", {})
    inkbox_display = platforms.setdefault("inkbox", {})
    if "show_reasoning" not in inkbox_display:
        inkbox_display["show_reasoning"] = False
        save_config(config)


def _slack_credential_error(exc: Exception) -> str | None:
    """Translate credential rejections without printing response bodies."""
    status = _error_status(exc)
    detail = getattr(exc, "detail", None)
    if status == 422:
        return "Slack rejected the app-configuration token pair. Check the token types, expiry, and that both belong together."
    if status == 409:
        if isinstance(detail, dict) and detail.get("code") == "credentials_required":
            return "Slack needs a new app-configuration access and refresh token pair."
        if detail == "Generate a separate app-configuration token pair for this organization.":
            return "Generate a separate app-configuration token pair for this Inkbox organization."
        if detail == "Use app-configuration tokens from the Slack user who created these apps.":
            return "Use app-configuration tokens from the Slack user who created this workspace's apps."
    return None


def _print_slack_setup_error(stage: str, exc: Exception) -> None:
    """Identify the failed step without exposing tokens or installation URLs."""
    status = _error_status(exc)
    suffix = f" (HTTP {status})" if isinstance(status, int) and 100 <= status <= 599 else ""
    print_warning(f"  Could not finish Slack setup while {stage}{suffix}.")
    print_info("  The Slack connection was not confirmed. Rerun `hermes inkbox setup` to continue.")


def _wait_for_slack(
    resource: Any, identity_id: Any, *, deadline: float,
    workspace_id: str | None = None,
) -> Any | None:
    """Wait for app readiness or the selected workspace's connection."""
    while time.monotonic() < deadline:
        snapshot = resource.list_connections(identity_id)
        if workspace_id is not None:
            for connection in snapshot.connections:
                if connection.status == "connected" and connection.workspace_id == workspace_id:
                    print_success(f"  Slack connected: {connection.workspace_name}")
                    print_info("  Add the bot to the channels where you want to use it.")
                    return snapshot
        else:
            status = getattr(getattr(snapshot, "setup", None), "status", None)
            if status == "ready":
                if snapshot.installation_available:
                    return snapshot
                print_warning("  Slack installation is currently unavailable. Check the identity in the Inkbox console.")
                return None
            if status == "needs_credentials":
                print_warning("  Slack workspace credentials need updating.")
                return snapshot
            if status in {"failed", "unavailable"}:
                print_warning("  Slack app preparation needs attention. Check Slack in the Inkbox console.")
                return None
            if status not in {"pending", "not_started"}:
                print_warning("  Slack app preparation is not supported by this API/SDK. Upgrade and rerun setup.")
                return None
        time.sleep(min(4.0, max(0.0, deadline - time.monotonic())))
    print_warning("  Still waiting for Slack. Rerun `hermes inkbox setup` to continue.")
    return None


def _select_slack_workspace(resource: Any, snapshot: Any, *, renew_credentials: bool = False) -> Any | None:
    """Reuse the app's workspace or save a masked, write-only configuration pair."""
    bound = getattr(snapshot, "provisioning_workspace", None)
    bound_id = getattr(bound, "id", None) or getattr(snapshot.setup, "provisioning_workspace_id", None)
    if snapshot.application_created and bound_id is None:
        print_warning("  This app has no available workspace binding. Check Slack in the Inkbox console.")
        return None
    workspace = bound
    if workspace is None:
        workspaces = resource.list_provisioning_workspaces()
        if bound_id is not None:
            workspace = next((w for w in workspaces if str(w.id) == str(bound_id)), None)
            if workspace is None:
                print_warning("  The app's saved workspace is unavailable. Check Slack in the Inkbox console.")
                return None
        elif workspaces:
            choices = [
                f"{w.workspace_name} ({w.workspace_id})"
                + (" — update credentials" if w.status != "ready" else "")
                for w in workspaces
            ] + ["Add a workspace"]
            choice = prompt_choice("  Choose the workspace for this agent's Slack app:", choices, 0)
            workspace = workspaces[choice] if choice < len(workspaces) else None
        else:
            print_info("  No saved Slack workspaces were found for this Inkbox organization.")
    if workspace is not None and workspace.status == "ready" and not renew_credentials:
        return workspace
    if workspace is not None:
        print_info(f"  Update app-configuration credentials for {workspace.workspace_name} ({workspace.workspace_id}).")
    print_info("  Open https://api.slack.com/apps and choose your workspace under Your App Configuration Tokens.")
    print_info("  Generate an access token and refresh token. These are not bot tokens.")
    print_info("  Inkbox saves the pair for your organization; this bridge never saves either token locally.")
    while True:
        access = prompt("  Slack app-configuration access token (leave blank to skip)", password=True).strip()
        if not access:
            return None
        refresh = prompt("  Slack app-configuration refresh token (leave blank to skip)", password=True).strip()
        if not refresh:
            return None
        try:
            saved = resource.save_provisioning_workspace(access_token=access, refresh_token=refresh)
        except Exception as exc:
            problem = _slack_credential_error(exc)
            if problem is None:
                _print_slack_setup_error("saving workspace credentials", exc)
                return None
        else:
            if bound_id is not None and str(saved.id) != str(bound_id):
                problem = "These tokens belong to a different workspace. This app cannot move; use its original workspace's tokens."
            elif saved.status != "ready":
                problem = "Workspace credentials still need attention. Generate a new app-configuration token pair."
            else:
                print_success(f"  Saved Slack workspace credentials: {saved.workspace_name}")
                return saved
        print_warning(f"  {problem}")
        if not prompt_yes_no("  Enter another app-configuration token pair?", True):
            return None


def _configure_slack(api_key: str, base_url: str, handle: str, Inkbox: Any) -> bool:
    """Opt into Slack, prepare a workspace-bound app, and confirm installation."""
    print_header("Slack")
    print_info("  Connect a Slack workspace to chat with this agent.")
    enabled = _env("INKBOX_SLACK_ENABLED").lower() in {"1", "true", "yes", "on"}
    if not prompt_yes_no("  Enable Slack integration in this bridge?", enabled):
        _save("INKBOX_SLACK_ENABLED", "false")
        print_info("  Slack is off in this bridge. Existing workspace connections are unchanged.")
        return False

    stage = "checking the Inkbox API key"
    try:
        with Inkbox(**inkbox_client_kwargs(api_key, base_url)) as client:
            info = client.whoami()
            stage = "loading the agent identity"
            identity = client.get_identity(handle)
            subtype = _enum_value(getattr(info, "auth_subtype", ""))
            is_admin = subtype == "api_key.admin_scoped"
            if not is_admin and (
                subtype != "api_key.agent_scoped.claimed"
                or getattr(info, "scope", None) != f"agent_identity:{identity.id}"
            ):
                print_warning("  Slack requires a claimed identity and a key for this agent. Claim it in the Inkbox console first.")
                return enabled
            resource = getattr(client, "slack", None)
            if not all(callable(getattr(resource, name, None)) for name in (
                "start_setup", "list_provisioning_workspaces", "save_provisioning_workspace", "start_installation",
            )):
                print_warning("  Slack setup requires an Inkbox SDK with provisioning-workspace support.")
                print_info(f"  Upgrade the SDK and rerun setup: {_install_command_text()}")
                return enabled

            _save("INKBOX_SLACK_ENABLED", "true")
            enabled = True

            stage = "checking existing Slack connections"
            snapshot = resource.list_connections(identity.id)
            connected = [c for c in snapshot.connections if c.status == "connected"]
            if connected:
                print_success("  Already connected: " + ", ".join(c.workspace_name for c in connected))
                print_info("  Each identity's app stays in one workspace. Manage reconnection and permissions in the Inkbox console.")
                return True
            if not prompt_yes_no("  Connect Slack now?", True):
                print_info("  Slack is enabled in this bridge. Rerun `hermes inkbox setup` anytime to connect.")
                return True
            setup = getattr(snapshot, "setup", None)
            if setup is None or setup.status == "unavailable":
                print_warning("  Slack setup is unavailable. Check the identity and API in the Inkbox console.")
                return True
            if getattr(setup, "error_code", None) == "outcome_unknown":
                print_warning("  Slack app creation could not be confirmed. Contact support before trying again.")
                return True
            if setup.status not in {"ready", "pending"}:
                stage = "loading saved Slack workspaces"
                workspace = _select_slack_workspace(resource, snapshot)
                if workspace is None:
                    print_info("  Workspace setup skipped. Rerun `hermes inkbox setup` to continue.")
                    return True
                stage = "requesting Slack app preparation"
                resource.start_setup(identity.id, workspace.id)
            print_info("  Preparing Slack… (Ctrl+C skips; other setup steps will continue.)")
            deadline = time.monotonic() + 300.0
            while True:
                stage = "waiting for Slack app preparation"
                ready = _wait_for_slack(resource, identity.id, deadline=deadline)
                if ready is None:
                    return True
                if ready.setup.status == "needs_credentials":
                    if not prompt_yes_no("  Update the workspace credentials and retry Slack setup?", True):
                        return True
                    stage = "loading saved Slack workspaces"
                    workspace = _select_slack_workspace(resource, ready, renew_credentials=True)
                    if workspace is None:
                        return True
                    stage = "requesting Slack app preparation"
                    resource.start_setup(identity.id, workspace.id)
                    deadline = time.monotonic() + 300.0
                    continue
                workspace = getattr(ready, "provisioning_workspace", None)
                if workspace is None:
                    print_warning("  The app's workspace could not be confirmed. Check Slack in the Inkbox console.")
                    return True
                try:
                    stage = "creating the Slack installation link"
                    installation = resource.start_installation(identity.id, workspace_id=workspace.workspace_id)
                    break
                except Exception as exc:
                    detail = getattr(exc, "detail", None)
                    if not (
                        getattr(exc, "status_code", None) == 409
                        and isinstance(detail, dict)
                        and detail.get("code") == "slack_setup_pending"
                    ):
                        raise
                    # Only the explicit pending response proves no link was created.
                    time.sleep(min(4.0, max(0.0, deadline - time.monotonic())))
            if not installation.authorization_url:
                print_warning("  No Slack installation link was returned. Rerun setup later.")
                return True
            print_info("  Open this installation link in your browser and approve Add to Slack:")
            print_info(f"  {installation.authorization_url}")
            print_info(f"  Expires: {installation.expires_at}. Keep the link private and finish in the same browser.")
            print_info("  Waiting for a successful connection… (Ctrl+C skips.)")
            stage = "waiting for Slack installation"
            _wait_for_slack(
                resource, identity.id, deadline=time.monotonic() + 300.0,
                workspace_id=workspace.workspace_id,
            )
    except KeyboardInterrupt:
        print_info("\n  Slack waiting skipped. Rerun `hermes inkbox setup` to continue.")
    except Exception as exc:
        _print_slack_setup_error(stage, exc)
    return enabled



def _configure_response_modes() -> None:
    """Configure independent group addressing and Companion sender gates."""
    for name, values, labels, prompt in (
        ("INKBOX_GROUP_REPLY_MODE", ("auto", "mention"),
         ("Automatic replies", "Require @agent or @handle"), "  Group replies:"),
        ("INKBOX_COMPANION_RESPONSE_MODE", ("safe", "relaxed"),
         ("Safe: directly admitted senders", "Relaxed: all delivered senders"), "  Companion replies:"),
    ):
        current = (_env(name) or values[0]).strip().lower()
        default = values.index(current) if current in values else 0
        selected = prompt_choice(prompt, list(labels), default)
        _save(name, values[selected])


def interactive_setup() -> None:
    print_header("Inkbox")
    print_info("API-first email + SMS + voice + identity for AI agents.")
    print_info("Inkbox is the recommended way to give your Hermes agent")
    print_info("its own real mailbox, phone number, and contact graph.")

    symbols = _ensure_inkbox_sdk()
    if symbols is None:
        return

    Inkbox = symbols["Inkbox"]
    InkboxAPIError = symbols["InkboxAPIError"]
    IdentityPhoneNumberCreateOptions = symbols["IdentityPhoneNumberCreateOptions"]
    WhoamiApiKeyResponse = symbols["WhoamiApiKeyResponse"]
    ADMIN_SCOPED = symbols["ADMIN_SCOPED"]
    AGENT_CLAIMED = symbols["AGENT_CLAIMED"]
    AGENT_UNCLAIMED = symbols["AGENT_UNCLAIMED"]

    existing_key = _env("INKBOX_API_KEY")
    existing_identity = _env("INKBOX_IDENTITY")
    if existing_key and existing_identity:
        print()
        print_success(f"Inkbox is already configured for identity '{existing_identity}'.")
        if not prompt_yes_no("  Reconfigure Inkbox?", False):
            return

    base_url = os.getenv("INKBOX_BASE_URL") or _env("INKBOX_BASE_URL") or INKBOX_BASE_URL_DEFAULT

    print()
    print_info("If you do not have an Inkbox API key yet, that is fine.")
    print_info("We can create a fresh agent identity for you via self-signup.")
    has_key = prompt_yes_no("  Do you already have an Inkbox API key?", False)
    authority_identity = None

    if not has_key:
        identity, api_key, _ = _self_signup_flow(base_url, Inkbox, InkboxAPIError)
        if identity is None:
            return
    else:
        identity, api_key, _, authority_identity = _api_key_flow(
            base_url,
            Inkbox,
            InkboxAPIError,
            WhoamiApiKeyResponse,
            ADMIN_SCOPED,
            AGENT_CLAIMED,
            AGENT_UNCLAIMED,
            IdentityPhoneNumberCreateOptions,
        )
        if identity is None:
            return

    _save("INKBOX_API_KEY", api_key)
    _save("INKBOX_IDENTITY", identity.agent_handle)
    if base_url != INKBOX_BASE_URL_DEFAULT or _env("INKBOX_BASE_URL"):
        _save("INKBOX_BASE_URL", base_url)

    _configure_channel_display()
    _configure_response_modes()
    _configure_avatar(base_url, api_key, identity, is_signup=not has_key)

    _save("INKBOX_ALLOW_ALL_USERS", "true")
    print()
    print_info("Inkbox authorization lives server-side via contact rules:")
    print_info("  https://console.inkbox.ai -> Mailboxes / Phone Numbers -> Contact Rules")
    print_info("Anyone Inkbox lets through reaches the agent. No second allowlist to maintain.")

    # Channels, in the order we want operators to think about them: connect
    # over iMessage FIRST (no number to provision — you reach the agent through
    # the shared Inkbox iMessage router), THEN offer a dedicated phone number
    # for SMS + voice.  Provisioning is decoupled from identity creation so this
    # ordering holds across every entry path (signup, admin, agent-scoped).
    imessage_on = _configure_imessage(api_key, base_url, identity.agent_handle, Inkbox)

    did_provision_phone = False
    try:
        dedicated_client = Inkbox(**inkbox_client_kwargs(api_key, base_url))
        identity, did_provision_phone = _offer_dedicated_number(dedicated_client, identity)
    except Exception as exc:
        print_warning(f"  Skipping dedicated-number setup: {exc}")

    _seed_identity_state(identity)
    _print_agent_summary(identity)

    # Block on the START text right after the number + QR are shown, before
    # moving on to voice setup — otherwise the "text START" prompt and its
    # blocking wait get split by the voice questions and it looks skipped.
    if did_provision_phone:
        _wait_for_sms_opt_in(api_key, base_url, getattr(identity, "phone_number", None), Inkbox)

    _configure_phone_call_voice_stack(
        identity,
        base_url=base_url,
        Inkbox=Inkbox,
        InkboxAPIError=InkboxAPIError,
        WhoamiApiKeyResponse=WhoamiApiKeyResponse,
        ADMIN_SCOPED=ADMIN_SCOPED,
        imessage_enabled=imessage_on,
        authority_identity=authority_identity,
    )

    _configure_slack(api_key, base_url, identity.agent_handle, Inkbox)
    _setup_signing_key(api_key, base_url, Inkbox)

    # Last, once every value is on disk: nothing above reaches the agent until
    # the gateway reloads .env, so offer to do that here instead of leaving it
    # as a printed instruction.
    gateway_live = _offer_gateway_restart()

    # A live gateway means setup finished the job, so close on that rather than
    # a to-do list. Only when nothing is listening is there a next step left.
    if gateway_live:
        _print_ready_banner(identity.agent_handle)
        return

    print()
    print("Next steps:")
    print("  hermes inkbox doctor")
    print("  hermes gateway run")