iris_x_hermes/gateway-plugin/setup.py

"""Interactive setup, passive config probes, env-driven auto-configuration.

``interactive_setup`` is the ``hermes gateway setup`` flow (token, host,
port, push backend, pairing QR, device removal). ``check_requirements`` /
``validate_config`` / ``is_connected`` are the passive probes the platform
registry calls from status displays. ``_env_enablement`` seeds
``PlatformConfig.extra`` from env vars before adapter construction.
"""

import logging
import os
import re
import time
from typing import Any

from hermes_constants import get_hermes_home

from . import qr
from .channels import get_directory
from .defaults import (
    DEFAULT_HOME_CHANNEL_NAME,
    DEFAULT_HOST,
    DEFAULT_HTTP_PORT,
    DEFAULT_PORT,
    DEFAULT_PUSH_BACKEND,
)
from .pairing import (
    DeviceRegistry,
    advertise_host,
    generate_token,
    pairing_url,
    qr_payload,
)
from .secrets import _get_scoped_secret

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Passive / config probes (called from status displays -- no side effects)
# ---------------------------------------------------------------------------


def check_requirements() -> bool:
    """PASSIVE dependency probe: token set.

    Must be side-effect free (called from ``hermes setup`` / ``status`` /
    dashboard readiness). Never installs. The HTTP transport is stdlib-only,
    so there is no extra dependency to probe.
    """
    return bool(_get_scoped_secret("IRIS_TOKEN"))


def validate_config(config) -> bool:
    """Given a PlatformConfig, is the platform properly configured?"""
    extra = getattr(config, "extra", {}) or {}
    token = _get_scoped_secret("IRIS_TOKEN") or extra.get("token", "")
    return bool(token)


def is_connected(config) -> bool:
    """Is the platform configured (env or config.yaml)?"""
    return validate_config(config)


# ---------------------------------------------------------------------------
# Env-driven auto-configuration (seeds PlatformConfig.extra pre-adapter)
# ---------------------------------------------------------------------------


def _env_enablement() -> dict | None:
    """Seed ``PlatformConfig.extra`` from env vars during gateway config load.

    Called by the platform registry's env-enablement hook BEFORE adapter
    construction, so ``gateway status`` and ``get_connected_platforms()``
    reflect env-only configuration without instantiating the adapter.
    Returns ``None`` when the platform isn't minimally configured (no token);
    the caller then skips auto-enabling.

    The special ``home_channel`` key in the returned dict is handled by the
    core hook -- it becomes a proper ``HomeChannel`` dataclass on the
    ``PlatformConfig`` rather than being merged into ``extra``.
    """
    token = _get_scoped_secret("IRIS_TOKEN", "")
    if not token:
        return None

    # Seed ONLY explicitly-set env vars: the core commits this seed on top of
    # config.yaml (``extra.update(seed)``), so default values here would
    # clobber user YAML. Unset keys fall through to config.yaml / adapter
    # defaults.
    seed: dict[str, Any] = {}
    host = os.getenv("IRIS_WS_HOST", "").strip()
    if host:
        seed["host"] = host
    http_port_raw = os.getenv("IRIS_HTTP_PORT", "").strip()
    if http_port_raw:
        seed["http_port"] = _parse_port(http_port_raw)
    push = os.getenv("IRIS_PUSH_BACKEND", "").strip().lower()
    if push:
        seed["push_backend"] = push
    home = os.getenv("IRIS_HOME_CHANNEL", "").strip()
    if home:
        seed["home_channel"] = {
            "chat_id": home,
            "name": os.getenv("IRIS_HOME_CHANNEL_NAME", "").strip() or DEFAULT_HOME_CHANNEL_NAME,
        }
    return seed


def _parse_port(raw: str) -> int:
    try:
        return int((raw or "").strip())
    except (ValueError, TypeError):
        return DEFAULT_PORT


# ---------------------------------------------------------------------------
# Target parsing: "<chat_id>[:<thread>]" (platform prefix stripped by core)
# ---------------------------------------------------------------------------


def _parse_target_ref(target_ref: str) -> tuple | None:  # noqa: PLR0911
    """Parse a raw target string into ``(chat_id, thread_id)`` or ``None``.

    The core strips the platform prefix before calling us, so the native
    syntax is simply ``<chat_id>[:<thread>]`` (e.g. ``chan_7`` or
    ``chan_7:t_31``); the home channel is ``default``. Chat ids are direct
    (no embedded platform prefix), so a cron delivery reads
    ``iris:chan_7`` end to end. A bare friendly name (e.g. ``Cron Reports``)
    is resolved against the channel directory so cron / ``send_message`` can
    target a channel by name immediately, without waiting for the core
    directory's refresh timer. Returns ``None`` for anything unrecognised so
    the target proceeds to the core channel-directory resolution.
    """
    if not target_ref:
        return None
    t = target_ref.strip()
    if not t:
        return None

    thread_id: str | None = None
    if ":" in t:
        head, tail = t.rsplit(":", 1)
        if head and tail.startswith("t_"):
            thread_id = tail
            t = head
        else:
            # Not a <chat>:<thread> pair -- treat the whole string as a name.
            t = target_ref.strip()
    if not t:
        return None

    # Native chat id (default / chan_<n>) or any id known to the directory
    # (covers custom IRIS_HOME_CHANNEL values).
    try:
        known = get_directory().get(t) is not None
    except Exception:
        known = False
    if t == "default" or re.fullmatch(r"chan_\d+", t) or known:
        return (t, thread_id)

    # Bare friendly name -> resolve via the channel directory. A thread resolves
    # to its session lane (parent_chat_id + thread_id); a channel/default to
    # its chat_id.
    try:
        entry = get_directory().resolve_entry(t)
    except Exception:
        entry = None
    if entry is not None:
        if entry["kind"] == "thread":
            return (entry["parent_chat_id"], entry["chat_id"])
        return (entry["chat_id"], None)
    return None


# ---------------------------------------------------------------------------
# Standalone (out-of-process) send -- best-effort, stretch for v1
# ---------------------------------------------------------------------------


async def _standalone_send(  # noqa: PLR0913
    pconfig,
    chat_id: str,
    message: str,
    *,
    thread_id: str | None = None,
    media_files: list[str] | None = None,
    force_document: bool = False,
) -> dict[str, Any]:
    """Out-of-process delivery for cron jobs that run separately from the
    gateway.

    The outbox is served by the *running* gateway, so standalone delivery
    while the gateway process is fully down is best-effort only (see
    ``docs/00-overview.md`` "Out of scope"). For M1 this is a stub that
    reports the gateway is required; the real implementation lands with the
    outbox (M3/M5).
    """
    return {
        "error": (
            "iris standalone send: the running gateway is required to serve "
            "the outbox (standalone delivery is best-effort only)"
        )
    }


# ---------------------------------------------------------------------------
# Verbose tool progress (full args on the progress line)
# ---------------------------------------------------------------------------


def _ensure_verbose_tool_progress() -> None:
    """Ensure the iris platform renders tool progress in ``verbose`` mode.

    Verbose mode makes the gateway's tool-progress line carry the FULL
    argument JSON (not just a ~40-char preview), which the adapter parses
    into the ``tool.start`` frame's ``args`` field; the app then decides how
    much to show (Settings → Tool detail). The tool *output* is captured
    separately via the ``post_tool_call`` hook (verbose mode does not stream
    it).

    Best-effort and idempotent: writes
    ``display.platforms.iris.tool_progress: verbose`` to config.yaml only
    when it isn't already set. The gateway's config cache is mtime-keyed, so
    the write takes effect on the next turn without a restart. Never raises.
    """
    try:
        from hermes_cli.config import load_config_readonly

        cfg = load_config_readonly() or {}
        display = cfg.get("display") or {}
        platforms = display.get("platforms") or {}
        iris_cfg = platforms.get("iris") or {}
        if iris_cfg.get("tool_progress") == "verbose":
            return  # already set
        from utils import atomic_roundtrip_yaml_update

        atomic_roundtrip_yaml_update(
            get_hermes_home() / "config.yaml",
            "display.platforms.iris.tool_progress",
            "verbose",
        )
        logger.info("iris: set display.platforms.iris.tool_progress=verbose")
    except Exception:
        logger.debug("iris: could not ensure verbose tool_progress", exc_info=True)


# ---------------------------------------------------------------------------
# Interactive setup (hermes gateway setup flow)
# ---------------------------------------------------------------------------


def _offer_device_removal() -> None:
    """Setup-flow device management (docs/09 §9.3): if devices are already
    paired, offer to revoke one. Revocation is server-side — no access to
    the device is needed: its per-device token is deleted and its id is
    denylisted, so even the shared token no longer authenticates it.

    Flow: ask (default No) → numbered select menu (last option = exit the
    removal loop, NOT the setup) → confirmation → back to the menu, so
    several devices can be removed in a row.
    """
    try:
        from hermes_cli.cli_output import (
            print_info,
            print_success,
            prompt,
            prompt_yes_no,
        )
    except Exception:
        return

    try:
        reg = DeviceRegistry(get_hermes_home() / "iris" / "devices.db")
    except Exception:
        return
    try:
        devices = reg.list()
        if not devices:
            return
        if not prompt_yes_no("Remove a paired device?", default=False):
            return
        while True:
            print_info("Paired devices:")
            for i, d in enumerate(devices, 1):
                last_seen = time.strftime("%Y-%m-%d %H:%M", time.localtime(d["last_seen"]))
                print_info(f"  {i}. {d['name']}  ({d['device_id']})  last seen {last_seen}")
            exit_idx = len(devices) + 1
            print_info(f"  {exit_idx}. Exit")
            # Default = exit: pressing Enter leaves the removal loop (and
            # continues the setup) without removing anything.
            choice = prompt("Select a device to remove", default=str(exit_idx))
            idx = int(choice) if choice.isdigit() else exit_idx
            if idx < 1 or idx >= exit_idx:
                return
            target = devices[idx - 1]
            if not prompt_yes_no(
                f"Remove {target['name']} ({target['device_id']})? It will no longer "
                "be able to connect (shared token included).",
                default=False,
            ):
                continue  # back to the select menu
            reg.revoke(target["device_id"])
            devices = [d for d in devices if d["device_id"] != target["device_id"]]
            print_success(f"Removed {target['device_id']} \u2014 it can no longer connect.")
            if not devices:
                print_info("No paired devices left.")
                return
    finally:
        reg.close()


def interactive_setup() -> None:
    """Prompt for the pairing token / host / port / push backend.

    M1: token generation, host/port/push prompts, and the pairing QR payload
    (``iris://pair?...``) + app URL printed for the Connect screen.
    """
    try:
        from hermes_cli.cli_output import (
            print_info,
            print_success,
            print_warning,
            prompt,
        )
        from hermes_cli.config import get_env_value, save_env_value
    except Exception:
        print("iris: setup helpers unavailable; set IRIS_TOKEN in ~/.hermes/.env")
        return

    print_info("📱 Android / Desktop (Iris x Hermes)")
    token = get_env_value("IRIS_TOKEN") or ""
    if not token:
        generated = generate_token()
        save_env_value("IRIS_TOKEN", generated)
        print_success(f"Generated pairing token: {generated}")
        print_warning("Keep this secret -- the app presents it on connect.")
    else:
        print_info("Existing IRIS_TOKEN found (not shown).")

    # Device management (docs/09 §9.3): on an existing setup, offer to cut
    # off a lost/compromised device before continuing with the config.
    _offer_device_removal()

    host = prompt("Bind host", default=get_env_value("IRIS_WS_HOST") or DEFAULT_HOST)
    save_env_value("IRIS_WS_HOST", host or DEFAULT_HOST)
    # _parse_port falls back to DEFAULT_PORT (8790) for empty input, so the
    # HTTP default must be applied explicitly (docs/19: 8791).
    http_port_raw = (get_env_value("IRIS_HTTP_PORT") or "").strip()
    port = prompt(
        "HTTP port",
        default=str(int(http_port_raw) if http_port_raw.isdigit() else DEFAULT_HTTP_PORT),
    )
    save_env_value("IRIS_HTTP_PORT", str(_parse_port(port)))
    backend = prompt(
        "Push backend (ntfy/fcm)",
        default=get_env_value("IRIS_PUSH_BACKEND") or DEFAULT_PUSH_BACKEND,
    )
    backend = (backend or DEFAULT_PUSH_BACKEND).strip().lower()
    save_env_value("IRIS_PUSH_BACKEND", backend)
    if backend == "fcm":
        print_warning(
            "FCM push metadata (notification title, device token) is routed "
            "through Google's servers. For truly private communication use "
            "ntfy (self-hosted) instead."
        )

    # Pairing payload for the app's Connect screen (manual entry + QR scan).
    # Advertise a routable host: a bind wildcard (0.0.0.0/127.0.0.1) is
    # replaced by the default-route LAN IP so the QR points somewhere a phone
    # can actually reach (the user can still override the Server URL in-app).
    advertised = advertise_host(host or DEFAULT_HOST)
    url = pairing_url(advertised, _parse_port(port))
    pairing = qr_payload(advertised, _parse_port(port), token)
    print_info("Pair your device (enter this on the app's Connect screen):")
    print_info(f"Pairing URL: {pairing}")
    print_info(f"Server URL: {url}")
    if advertised != (host or DEFAULT_HOST):
        print_info(
            f"QR points to {advertised} (your default LAN address). If your "
            "phone is on a different network, change the Server URL in the app."
        )

    # Scannable QR (docs/20): the same payload as a terminal QR. The URL text
    # lines stay — the QR is a convenience, not a replacement (non-UTF-8
    # terminals still work, and the text is copy-pasteable). render_qr returns
    # '' (not an exception) when the payload is too long to encode.
    qr_block = qr.render_qr(pairing)
    if qr_block:
        print_info("Scan with the Iris app (Connect → Scan QR) or any camera app:")
        print(qr_block)
    else:
        print_warning("QR too large to render; use the pairing URL above.")

    # Always render tool progress verbosely so the app receives the full tool
    # call args (it decides how much to show via Settings → Tool detail).
    _ensure_verbose_tool_progress()

    print_success("Iris configuration saved to ~/.hermes/.env")
    print_info("Restart the gateway for changes to take effect: hermes gateway restart")