"""Bot-relay JSON-RPC handlers — the gateway side of cross-connection A2A. Connections ARE the
peer set: the Desktop owns every gateway socket and relays between them via four doors on EACH
gateway: ``roster.sync`` (push OTHER connections' agents so ``message_agent`` resolves them),
``outbox.drain`` (collect envelopes queued here for other connections), ``deliver`` (one-turn Bot
Chat delivery on the TARGET gateway, returns the reply), ``reply`` (write the reply/error back on
the SENDER gateway for its waiter). Plumbing: ``tools/bot_relay.py``; handlers are rebound onto
server.py's globals (method_ctx.py) and reference ``_ok``/``_err`` bare."""

import logging
from pathlib import Path

from .method_ctx import HandlerRegistry

_registry = HandlerRegistry()
method = _registry.method


def _relay_root() -> Path:
    """Install root shared by every profile (relay state is install-wide). Same formula as the
    writers (``tools/bot_relay``, ``tools/bot_mode_dm``): both ends of the mailbox must agree for
    every HERMES_HOME, including non-``profiles/`` subdirs of ``~/.hermes``."""
    from tools.bot_mode_probe import _default_home, _hermes_root
    return _hermes_root(Path(_default_home()))


# Desktop deadline mirror; no retry runs here. The owner gateway retries a transiently failed
# admission once (``gateway.session_bot._maybe_retry``). Remove with the renderer relay deadline migration.
TURN_MAX_ATTEMPTS = 2  # first admission + the owner's one transient-failure retry (gateway.session_bot)


@method("bot_relay.roster.sync")
def _(rid, params: dict, _root=_relay_root) -> dict:
    """Replace this gateway's view of agents on OTHER connections → ``{count}`` accepted rows
    (``agents`` rows ``{profile, handle, connection_id, ...}``; invalid rows are dropped).
    A roster that changes is logged with its rows: a peer connection listed with this machine's
    own profiles, or two publishers alternating, shows up as a flapping line (the pushing window
    is in the Desktop's desktop.log ``[bot-relay win=…]`` lines at the same time)."""
    try:
        from tools.bot_relay import read_remote_roster, write_remote_roster

        def trace() -> str:
            rows = read_remote_roster(_root())
            return f"n={len(rows)} [" + " ".join(
                f"{r.get('connection_id')}/{r.get('profile')}={r.get('title')!r}" for r in rows) + "]"

        before = trace()
        count = write_remote_roster(_root(), params.get("agents"))
        after = trace()
        if after != before:
            logging.getLogger(__name__).info("bot_relay roster changed: %s -> %s", before, after)
        return _ok(rid, {"count": count})
    except Exception as e:
        return _err(rid, 5090, str(e))


@method("bot_relay.outbox.drain")
def _(rid, params: dict, _root=_relay_root) -> dict:
    """Claim every pending cross-connection envelope queued here → ``{envelopes}``; claimed
    envelopes move to ``claimed/`` atomically so concurrent drains can't double-deliver."""
    try:
        from tools.bot_relay import claim_pending_envelopes
        return _ok(rid, {"envelopes": claim_pending_envelopes(_root())})
    except Exception as e:
        return _err(rid, 5091, str(e))


@method("bot_relay.deliver")
def _(rid, params: dict, _root=_relay_root) -> dict:
    """Legacy transport bridge only: canonical admission or an explicit refusal."""
    from tools.bot_live_delivery import _delivery_id, authority_delivery
    from tools.bot_relay import _HANDLE_RE
    try:
        _delivery_id(params.get('id'))
        profile = params.get('profile')
        if not isinstance(profile, str) or not _HANDLE_RE.fullmatch(profile):
            raise ValueError('invalid profile')
    except ValueError:
        return _err(rid, 4090, 'invalid_params', data={'reason': 'invalid_params'})
    from tools.bot_relay import delivery_turn_author, relaying_principal_author
    from tui_gateway.methods_browser_control import _is_authenticated_identity, _principal_digest
    sender_fields = ("from_profile", "from_handle", "from_connection")
    identity = getattr(current_transport(), "auth_identity", None)
    if _is_authenticated_identity(identity):
        # A logged-in client's sender fields are NOT trusted — but the delivery is not refused
        # either: the Desktop is itself a logged-in client on every gateway that requires sign-in
        # (it mints a ws-ticket carrying the signed-in {user_id, provider} —
        # hermes_cli/dashboard_auth/routes.py), so refusing took cross-connection relay offline
        # for exactly the auth-gated gateways it serves; only ``?internal=`` callers are
        # identity-exempt and the Desktop cannot present one. Nor is the author dropped: an
        # unattributed turn is the HUMAN's to the recipient's memory, so a bot DM must stay
        # bot-authored. The author is derived from the caller's minted identity instead — stable,
        # unspoofable, and ``is_bot`` — whether or not the client named a sender.
        author = relaying_principal_author(_principal_digest(identity))
    else:
        author = delivery_turn_author(*(params.get(key) for key in sender_fields))
    forwarded = {key: value for key, value in params.items() if key not in sender_fields}
    if author:
        forwarded["author"] = author
    resolved = 'default' if profile.lower() == 'hermes' else profile
    root = _root()
    # Same identity predicate as `profile list`: infra dirs and tombstones under profiles/ are
    # not teammates (#99392), so a DM never targets one.
    from tools.bot_mode_probe import _roster
    home = dict(_roster(root)).get(resolved)
    if home is None:
        return _err(rid, 4092, f"no profile '{profile}' on this gateway", data={'reason': 'unknown_profile'})
    if isinstance(forwarded.get("message"), str):
        # The sender stamped itself with its bare @handle; a relayed "@hermes" is ANOTHER machine's
        # default, so re-stamp it with the form this gateway can reply to (#103731).
        from tools.bot_mode_probe import local_taken_forms
        from tools.bot_relay import qualify_sender_stamp, read_remote_roster
        forwarded["message"] = qualify_sender_stamp(
            forwarded["message"], params.get("from_handle"), params.get("from_connection"),
            read_remote_roster(root), local_taken_forms(root))
    try:
        return _ok(rid, authority_delivery(home, {**forwarded, 'profile': resolved}))
    except Exception as exc:
        # Boundary: every way the target authority can be unreachable (discovery, ticket, WS,
        # RPC refusal) is the same typed ``runtime_unavailable`` refusal the sender retries on.
        logging.getLogger(__name__).debug("bot_relay.deliver to %s refused", resolved, exc_info=True)
        return _err(rid, 5094, str(exc), data={'reason': 'runtime_unavailable'})


@method("bot_relay.reply")
def _(rid, params: dict, _root=_relay_root) -> dict:
    """Write a relayed ``reply`` and/or ``error`` (+ optional typed ``reason``, see
    ``tools.bot_failure_reasons``) for envelope ``id`` so the sender-side waiter picks it up."""
    envelope_id = str(params.get("id") or "").strip()
    if not envelope_id:
        return _err(rid, 4093, "id required")
    try:
        from tools.bot_relay import write_reply
        write_reply(_root(), envelope_id, reply=str(params.get("reply") or ""),
                    error=str(params.get("error") or ""), reason=str(params.get("reason") or ""))
        return _ok(rid, {"ok": True})
    except ValueError as e:
        return _err(rid, 4094, str(e))
    except Exception as e:
        return _err(rid, 5095, str(e))


def register(server) -> None:
    _registry.install(server)
    from . import methods_groups
    server._LONG_HANDLERS = server._LONG_HANDLERS | methods_groups.LONG_HANDLERS
    for name in (
        "get_hosted_room_service", "_WORKER_UNAVAILABLE", "_profile_name", "_requested_profile",
        "_api_server_key", "_room_link_run_storage_durable"):
        setattr(server, name, getattr(methods_groups, name))
    methods_groups.bind_server(server)
    methods_groups.register(server)