"""Chatto Platform Adapter for Hermes Agent.

A plugin-based gateway adapter that connects to a Chatto server
(self-hosted team chat) and relays messages to/from the Hermes agent.


The adapter uses the chattolib library for all Chatto API interactions,
including both outbound messaging and realtime WebSocket connections.

"""

from __future__ import annotations

import json
import random

# Put the vendored dependencies for THIS platform on sys.path before importing
# anything from chattolib. Imported relatively as part of the plugin package and
# absolutely when this module is loaded standalone (e.g. by the tests).
try:
    from .vendor_path import setup_vendor_path
except ImportError:  # pragma: no cover - depends on how the module is loaded
    from vendor_path import setup_vendor_path

setup_vendor_path()

import asyncio
import hashlib
import logging
import mimetypes
import os
import re
import tempfile
import time
from collections import deque
from collections.abc import AsyncIterator, Callable, Iterable
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import UTC, datetime
from difflib import SequenceMatcher
from enum import StrEnum
from pathlib import Path
from typing import Any, Final, NamedTuple, TypeVar
from urllib.parse import unquote, urlsplit

import httpx

logger = logging.getLogger(__name__)

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
    BasePlatformAdapter,
    ExecApprovalPrompt,
    MessageEvent,
    MessageType,
    ProcessingOutcome,
    SendResult,
    cache_media_bytes,
    get_inbound_media_max_bytes,
    validate_inbound_media_size,
)
from gateway.session import SessionSource

# Chattolib imports (vendored)
# Using vendored chattolib from vendor/chattolib/
# See vendor_chattolib.sh for how to update the vendored copy

# Absolute imports — vendor/ is on sys.path (see above) and chattolib's own
# modules import each other absolutely. Mixing in relative ".vendor.chattolib"
# imports would load a second, distinct copy of every module, so isinstance()
# checks across the two copies would silently fail.
try:
    from chattolib._pb.chatto.api.v1.presence_pb2 import (
        PresenceStatus as ProtoPresenceStatus,  # type: ignore[import-untyped]
    )
    from chattolib._pb.chatto.realtime.v1 import (  # noqa: F401 — importability probe
        realtime_pb2,
    )
    from chattolib._pb.chatto.realtime.v1.events_pb2 import (
        MessageMention,  # type: ignore[import-untyped]
        MessagePostedEvent,  # type: ignore[import-untyped]
    )
    from chattolib.client import (
        ChattoClient,
    )
    from chattolib.exceptions import (
        ChattoAuthError,
        ChattoConnectError,
        ChattoError,
    )
    from chattolib.realtime import (
        ChattoRealtimeCloseError,
        ChattoRealtimeError,
        RealtimeEvent,
        stream_events,
    )
    from chattolib.types import (
        Message,
        MessageAttachment,
        MessageSearchGroupBy,
        MessageSearchOrder,
        MessageSearchResult,
        MessageSearchScope,
        MessageSearchState,
        PresenceStatus,
        Room,
        RoomKind,
        RoomWithViewerState,
        User,
        format_datetime,
    )

except ImportError as e:
    # Fail loudly: continuing here only defers the failure to a confusing
    # NameError somewhere deep in the adapter.
    logger.error("Chatto: failed to import vendored chattolib: %s", e)
    raise


try:
    from .platform_config import (
        ChattoConfiguration,
        ChattoConstants,
    )
except ImportError:  # pragma: no cover - loaded as a top-level module (tests)
    from platform_config import (
        ChattoConfiguration,
        ChattoConstants,
    )


# --------------------------------------------------------------------------- #
# Chat types
# --------------------------------------------------------------------------- #


class HermesChatType(StrEnum):
    """The ``chat_type`` vocabulary the Hermes gateway understands.

    Declared in ``gateway/session.py:161`` as ``"dm", "group", "channel",
    "thread"`` and consumed as a bare string all over the gateway:
    ``SessionSource.description`` (session.py:239) and the PII-redacting
    description in ``build_session_context_prompt`` (session.py:537) both
    branch on these exact values and fall back to a nameless generic case for
    anything else, and ``build_session_key`` puts the value straight into the
    session key.  Passing a chattolib ``RoomKind`` (``"ROOM_KIND_CHANNEL"``)
    therefore does not fail loudly — it just quietly degrades what the agent is
    told about where it is.

    A StrEnum so it stays a drop-in ``str`` at every one of those call sites.

    GROUP vs CHANNEL
    ----------------
    There is no strict contract between the two, and the adapters disagree in
    practice: Slack labels every non-DM conversation ``"group"`` (including real
    channels), Discord uses both, and Telegram reserves ``"channel"`` for actual
    broadcast channels.  The intended reading is ``group`` = ordinary
    multi-participant chat, ``channel`` = broadcast surface.

    The distinction only changes behaviour in three places:

    1. Authorization (``gateway/authz_mixin.py``) — the only security-relevant
       one.  The group-scoped env allowlists apply to ``{"group", "forum"}``
       ONLY, never to ``"channel"``: ``{PLATFORM}_GROUP_ALLOWED_USERS`` /
       ``_GROUP_ALLOWED_CHATS`` (:616), the chat-id allowlist (:708) and the
       Telegram legacy shim (:724).  The adapter-delegation paths in turn treat
       all three alike (:461, :649, :674, :694), where the value only picks
       ``group_allow_from`` over ``allow_from`` from ``config.extra``.
       For Chatto both choices are equivalent today: those group env maps hold
       Telegram and QQBot only (:535-541), and our own allowlist runs through
       ``CHATTO_ALLOWED_USERS``, which is chat_type-independent.
    2. What the agent is told — ``SessionSource.description`` renders
       ``"group: Name"`` vs ``"channel: Name"`` (session.py:239-246), likewise
       the PII-redacted variant (session.py:537-544).
    3. The session key, which embeds the literal (session.py:1192).  Changing
       the value for a room re-buckets its existing sessions.

    Explicitly NOT affected: ``is_shared_multi_user_session`` (session.py:1063)
    only looks at ``"dm"`` and ``thread_id``, so sender prefixes, the multi-user
    prompt line and ``group_sessions_per_user`` treat group and channel
    identically.
    """

    DM = "dm"
    GROUP = "group"
    CHANNEL = "channel"
    # Emitted by adapters whose thread events are their own chat type (Slack,
    # Discord). We don't: a Chatto thread keeps its room's chat_type and is
    # identified by ``thread_id`` on the source instead. Listed for the record,
    # because build_session_key rewrites the slot to "thread" itself
    # (session.py:1190).
    THREAD = "thread"
    # Not declared in session.py:161 but real: Telegram forum topics travel as
    # "forum", and the authz group allowlists above accept it alongside "group".
    # Chatto has no equivalent, so we never emit it.


# Chatto only distinguishes DMs from channels. UNSPECIFIED means the server
# sent a kind this vendored chattolib doesn't know: map it to the generic
# multi-user bucket rather than guessing "channel", and never to "dm" — that
# value drives session isolation (is_shared_multi_user_session, session.py:1063)
# and would silently turn a room into a private conversation.
#
# CHANNEL for RoomKind.CHANNEL is the descriptive choice and carries no
# behavioural cost (see the GROUP vs CHANNEL note above). Switching to GROUP for
# Slack parity would be this one line — plus the re-bucketing of existing
# sessions that point 3 of that note describes.
_ROOM_KIND_TO_CHAT_TYPE: dict[RoomKind, HermesChatType] = {
    RoomKind.DM: HermesChatType.DM,
    RoomKind.CHANNEL: HermesChatType.CHANNEL,
    RoomKind.UNSPECIFIED: HermesChatType.GROUP,
}


def chat_type_for_room_kind(kind: RoomKind | None) -> HermesChatType:
    """Map a chattolib RoomKind onto the gateway's chat_type vocabulary.

    An unknown or missing kind becomes ``GROUP`` — see ``_ROOM_KIND_TO_CHAT_TYPE``.
    """
    if kind is None:
        return HermesChatType.GROUP
    return _ROOM_KIND_TO_CHAT_TYPE.get(kind, HermesChatType.GROUP)


class RoomPolicy(StrEnum):
    """How a channel-kind room treats an inbound message.

    Derived per room from ``CHATTO_REQUIRE_MENTION_ROOMS`` /
    ``CHATTO_OPTIONAL_MENTION_ROOMS`` — see ``_room_policy``. A StrEnum
    so the values log readably without a formatting dance.
    """

    # Listed for unaddressed answers: every message is dispatched, and one
    # aimed at a named colleague gets a 🫥 acknowledgement instead of a reply.
    OPEN = "open"
    # Listed for addressed-only participation: dispatches only messages that
    # mention the bot (@name or a broadcast handle); others are dropped.
    REQUIRE_MENTION = "require_mention"
    # Listed in neither config: the room is silent — not dispatched at all,
    # read-only like any unlisted membership.
    SILENT = "silent"


# Blocking filesystem/network helpers. The adapter runs on the shared gateway
# event loop, so file reads and HTTP downloads are pushed to a worker thread
# via asyncio.to_thread — a slow disk or dead image URL must not stall every
# platform's message processing.


def _read_file_bytes(path: str) -> bytes:
    """Read a whole file synchronously (run via asyncio.to_thread)."""
    with open(path, "rb") as f:
        return f.read()


def _write_file_bytes(path: str, data: bytes) -> None:
    """Write bytes to a file synchronously (run via asyncio.to_thread)."""
    with open(path, "wb") as f:
        f.write(data)


def _normalise_outbound_text(content: str) -> str:
    """Normalise outgoing text for Chatto.

    Chatto renders Markdown natively, so there is nothing to escape or
    translate — the only transformations here are the ones that measurably
    render wrong: CRLF line endings (which show up as stray blank lines)
    and runs of more than two blank lines. Shared by the adapter's
    ``format_message`` and the standalone cron sender, so both paths render
    identically.
    """
    if not content:
        return content
    normalised = content.replace("\r\n", "\n").replace("\r", "\n")
    while "\n\n\n\n" in normalised:
        normalised = normalised.replace("\n\n\n\n", "\n\n\n")
    return normalised


def _oldest_first(messages: list[Message]) -> list[Message]:
    """Sort messages by the timestamp they carry, oldest first.

    The server's ordering is not a contract anything here builds on — it is a
    page, and a page's order depends on which end it was fetched from — so
    every timeline is read the way a conversation is spoken. A message with no
    timestamp sorts first: it is the one whose position we know least about,
    and putting it at the end would claim an order the data does not support.

    Shared by the thread seeding and the read tools so the block an agent reads
    in a session and the block it gets from ``chatto_get_history`` are ordered
    by one rule.
    """
    return sorted(
        messages, key=lambda m: m.created_at or datetime.min.replace(tzinfo=UTC)
    )


# --------------------------------------------------------------------------- #
# Adapter
# --------------------------------------------------------------------------- #

# Presence as the roster block spells it. UNSPECIFIED gets no label: a server
# that never tracked presence should not have the roster claim anything.
_PRESENCE_LABELS = {
    PresenceStatus.ONLINE: "online",
    PresenceStatus.AWAY: "away",
    PresenceStatus.DO_NOT_DISTURB: "do not disturb",
    PresenceStatus.OFFLINE: "offline",
}

# Proto enum name (``PRESENCE_STATUS_ONLINE``) → chattolib member, built once
# from the enum's own values. The realtime wire carries the presence status as
# a plain int (protobuf runtime without native enum wrappers), so every
# presence change needs this lookup — a linear scan per event would be O(n) on
# a hot path for a five-member enum.
_PRESENCE_BY_PROTO_NAME = {member.value: member for member in PresenceStatus}


@dataclass(frozen=True)
class _Mentions:
    """Who one inbound message addresses.

    Both admission gates read this and nothing else, so "addressed" has a
    single definition. What fills it differs by path, not what counts:

    - :meth:`from_payload` reads the targets the server resolved when the
      message was posted. It is authoritative and the only source the posted
      path uses.
    - ``ChattoAdapter._mentions_from_body`` re-derives them from the text,
      because an edit event carries no targets — see
      ``_dispatch_message_edited``.

    Broadcast and role targets arrive unexpanded (one ``all``/``here``/``role``
    entry, never one per recipient), which is why the payload path needs no
    room roster to answer "was this meant for me".
    """

    me: bool = False
    someone_else: bool = False

    @classmethod
    def from_payload(
        cls, mentions: Iterable[MessageMention], my_user_id: str
    ) -> _Mentions:
        """Read the server-resolved targets of a ``message_posted`` event.

        ``includes_viewer`` is the server's own answer to "did this target
        include me", decided when the message was posted and unaffected by
        membership since. A direct target naming somebody else is the only
        kind that means "not for us": broadcasts and roles address the room,
        and we are in it.
        """
        me = False
        someone_else = False
        for mention in mentions:
            if mention.includes_viewer:
                me = True
            elif mention.WhichOneof("cause") == "direct" and (
                mention.direct.user_id != my_user_id
            ):
                someone_else = True
        return cls(me=me, someone_else=someone_else)


@dataclass
class _RoomRoster:
    """One announced room's roster projection.

    A miniature of the server's room membership: which user IDs belong to
    the room (the users themselves live in the shared ``_user_cache``) and
    how many directory entries were beyond the fetch limit, rendered as
    ``… and N more``. Presence changes patch the cached users in place;
    only a membership change or reconnect discards this and refetches.
    """

    member_ids: set[str] = field(default_factory=set)
    unfetched: int = 0


@dataclass
class _ClarifyPrompt:
    """State for a reaction-driven clarify prompt posted to a Chatto room."""

    clarify_id: str
    choices: list[str]
    session_key: str
    chat_id: str
    message_id: str
    resolved: bool = False
    awaiting_text: bool = False
    resolved_by: str | None = None
    resolved_choice: str | None = None
    expires_at: float = 0.0


@dataclass
class _ApprovalPrompt:
    """State for a reaction-driven exec-approval prompt posted to a Chatto room."""

    session_key: str
    chat_id: str
    message_id: str
    choices_map: dict[str, str]  # emoji -> choice
    resolved: bool = False
    resolved_by: str | None = None
    resolved_choice: str | None = None
    expires_at: float = 0.0


INTERACTIVE_PROMPT_TIMEOUT_SECONDS = 300


_UNSEEDED: Final = object()
"""Sentinel: nothing is known yet about how far this thread's session reaches.

The in-memory map starts every thread here. The session's own copy starts here
too — a session carrying no cursor is one whose frontier cannot be established
(see _thread_context_for), which is not the same as a thread this process has
not seeded.
"""


_CURSOR_META_KEY: Final = "chatto.thread_seed_cursor"
"""Session-metadata key holding the last message ID a thread's session reached.

One opaque ID per thread, on the session the platform already owns: nothing of
the thread's content is mirrored anywhere, and the value dies with the session.
"""


_T = TypeVar("_T")


def hermes_adapter_factory(config: PlatformConfig):
    """Construct a ChattoAdapter from a PlatformConfig."""
    return ChattoAdapter(config)


class ChattoAdapter(BasePlatformAdapter):
    """Chatto platform adapter.

    Receives messages via WebSocket realtime, sends via ConnectRPC.
    """

    # Read by BasePlatformAdapter.max_message_length_for_chat(), which the
    # gateway and the stream consumer use to chunk outgoing messages. Without
    # it they fall back to 4096 and split Chatto messages far earlier than
    # necessary — send() itself already truncates at SPLIT_THRESHOLD.
    MAX_MESSAGE_LENGTH = ChattoConstants.MAX_MESSAGE_LENGTH
    splits_long_messages = True
    supports_code_blocks: bool = True
    supports_status_text: bool = True  # client.update_custom_status

    def __init__(self, pconfig: PlatformConfig):
        """Signature needs to be compatible with BasePlatformAdapter.__init__."""
        super().__init__(
            config=pconfig, platform=Platform(ChattoConstants.PLATFORM_NAME)
        )
        # "extra" has been pre-populated by Hermes from config.yaml's extra block.

        # --- Configuration from our configuration data class with some logic ---
        self.chatto_config: ChattoConfiguration = ChattoConfiguration(pconfig)

        # ------ State -------
        # Our own user, filled in by connect(). Events arriving before connect()
        # completes must not blow up on an undefined attribute.
        self.me: User | None = None

        # --- Runtime state ---
        self._room_names: dict[str, str] = {}
        self._room_kinds: dict[str, RoomKind] = {}
        # Last realtime cursor whose event was fully applied. Passed back to
        # stream_events on reconnect so the server resumes after it instead of
        # starting at the current boundary and silently skipping everything
        # that arrived while we were disconnected. None until the first event
        # carries one (chattolib leaves it unset for unreplayable events).
        self._realtime_cursor: str | None = None
        # Event IDs already processed — chattolib may redeliver events across
        # reconnects, so every inbound event is checked against this list.
        # Bounded deques: appending past the cap drops the oldest ID on its own.
        self._seen: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
        # Message IDs this adapter has handed to the gateway (posted or
        # edit-re-dispatched). Edits of anything on this list never start a
        # fresh turn — that is the lock against re-answering settled
        # conversations by editing old messages.
        self._dispatched_ids: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
        # Rooms whose member roster has been projected from the directory —
        # the roster is room-scoped and rides on every channel turn (threads
        # hold isolated sessions, so each turn needs its own copy). Together
        # with _rosters this is our miniature projection of server state:
        # presence changes patch the cached users in place; membership events
        # and reconnects discard a room so its next turn refetches once.
        self._roster_announced: deque[str] = deque(maxlen=ChattoConstants.SEEN_CAP)
        # Announced rooms' membership projection (see _RoomRoster).
        self._rosters: dict[str, _RoomRoster] = {}
        # Thread roots whose history has been handed to their session, mapped to
        # the last message ID that was delivered. Thread sessions start empty, so
        # the first turn into a thread seeds what came before it — a cron message
        # whose echo we filtered as a self-event, or chatter that predates the
        # bot joining.
        #
        # The cursor is what makes later turns safe to seed again: the agent's
        # own messages never reach a session (their echo is filtered as a
        # self-event), so a reply it posts into a live thread via
        # chatto_send_room is invisible to that thread forever unless the tail
        # after the cursor is delivered. Re-delivering everything instead would
        # duplicate context on every turn. Insertion-ordered and capped like the
        # seen-ring; unlike the roster this is deliberately NOT discarded on
        # reconnect, since thread history is immutable.
        #
        # Keyed by thread root, which assumes one session per thread. With
        # thread_sessions_per_user turned on, each participant gets their own
        # session while this map stays shared, so one participant's turn moves
        # the cursor past the other's messages and that session is seeded once
        # and never again. The flag is off by default and this is the price of
        # switching it on; keying the map by session key would fix it.
        self._seeded_threads: dict[str, str] = {}
        # Threads whose cursor is on its way to the session but has not landed:
        # a thread's first turn has no session yet, and the second turn retries.
        self._unpersisted_threads: set[str] = set()
        # session_key -> message ID currently being processed there. Written
        # by on_processing_start, cleared by on_processing_complete; an edit
        # landing on the recorded ID is a mid-run correction.
        self._processing: dict[str, str] = {}
        self._joined_room_ids: list[str] = []
        # Rooms the server force-joined everyone into (Room.universal) — used
        # only for [universal] tags in the joined-rooms log line, never for
        # gating.
        self._universal_room_ids: set[str] = set()
        # DM room ID -> chat partner's login, resolved once via the member
        # directory so the joined-rooms summary can name who a DM is with.
        self._dm_partners: dict[str, str] = {}
        # One-shot guard for the unjoined-home-channel warning in _refresh_rooms.
        self._home_warning_logged = False
        self._ws_task: asyncio.Task | None = None
        self._presence_task: asyncio.Task | None = None

        # Persistent typing indicator loops per room
        self._typing_tasks: dict[str, asyncio.Task] = {}

        # Member directory cache: user_id -> user info dict
        self._user_cache: dict[str, User] = {}

        # Interactive prompt registries (clarify + exec-approval), keyed by
        # message_id. Reverse lookup _clarify_id_to_message supports
        # retire_clarify_card.
        self._clarify_prompts: dict[str, _ClarifyPrompt] = {}
        self._approval_prompts: dict[str, _ApprovalPrompt] = {}
        self._clarify_id_to_message: dict[str, str] = {}

        # Chattolib client cache and lock for async access.
        self._chatto_client: ChattoClient | None = None
        self._chatto_client_lock: asyncio.Lock = asyncio.Lock()

    # ------------------------------------------------------------------ #
    # Auth
    # ------------------------------------------------------------------ #

    async def _get_chatto_client(self: ChattoAdapter) -> ChattoClient | None:
        """Return the shared ChattoClient, creating and logging in on first use.

        The default way to get a client. The fast path is a plain attribute
        read; first creation runs under a lock so concurrent callers log in
        exactly once. Returns ``None`` when no client exists yet or creation
        failed (bad credentials, unreachable server) — a normal state during
        startup, shutdown and reconnects, not an exceptional one. Callers
        decide what "no client" means for them, as a guard clause:

            client = await self._get_chatto_client()
            if client is None:
                logger.warning("Chatto: dropping X - no client available")
                return
        """
        if self._chatto_client is not None:
            return self._chatto_client

        async with self._chatto_client_lock:
            if self._chatto_client is not None:
                return self._chatto_client

            try:
                client = await self._open_client(
                    base_url=self.chatto_config.base_url.value,
                    token=self.chatto_config.token.value,
                )
                self._chatto_client = client
                logger.info(
                    "Chatto: authenticated via token to '%s'",
                    self.chatto_config.base_url.value,
                )
                return client
            except ChattoAuthError as e:
                logger.error("Chatto: authentication failed: %s", e)
                return None
            except (ChattoError, ValueError) as e:
                logger.error("Chatto: failed to create client: %s", e)
                return None

    async def _require_chatto_client(self) -> ChattoClient:
        """Return a ChattoClient or raise RuntimeError if unavailable.

        The exception-flavoured variant of :meth:`_get_chatto_client`, for
        callers whose surrounding machinery already routes exceptions.
        Currently that is only the realtime event loop, whose except chain
        turns the raise into a logged warning plus a backed-off reconnect —
        no special "no client" branch needed there. Everywhere else, prefer
        the ``is None`` guard shown in :meth:`_get_chatto_client`.
        """
        client = await self._get_chatto_client()
        if client is None:
            raise RuntimeError("Chatto client unavailable")
        return client

    # ------------------------------------------------------------------ #
    # Connection
    # ------------------------------------------------------------------ #
    async def _open_client(
        self,
        *,
        base_url: str,
        token: str | None = None,
    ) -> ChattoClient:
        """Return a connected ``ChattoClient`` using a bearer token.

        Raises ValueError when no token is configured — caught upstream as a
        normal creation failure, so misconfiguration reads as a logged error
        instead of a crash.
        """
        if not token:
            raise ValueError("Chatto: CHATTO_TOKEN not configured")
        return ChattoClient(token=token, base_url=base_url)

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        """Connect to Chatto and start the realtime event stream.

        BasePlatformAdapter override
        """
        logger.info("Chatto: connecting...")

        client = await self._get_chatto_client()
        if client is None:
            self._set_fatal_error(
                "connect_failed", "Chatto client not available", retryable=True
            )
            return False

        # Get our first own user info
        try:
            self.me = await client.me()
        except Exception as exc:
            logger.error("Chatto: failed to get user info: %s", exc)
            self._set_fatal_error(
                "chatto_auth_failed",
                f"Chatto auth failed: {exc}",
                retryable=False,
            )
            try:
                if self._chatto_client is not None:
                    await self._chatto_client.close()
            finally:
                self._chatto_client = None
            return False

        # Announce online presence so the bot appears online in the member list.
        # The server treats this as a TTL, so _presence_refresh_loop below has to
        # keep re-announcing it — a single call here lapses back to offline.
        await self._announce_online()

        self._closing = False
        # A cursor persisted by a clean shutdown lets this process pick up where
        # the previous one stopped, so a restart inside the window answers what
        # arrived while it was down instead of starting at the live boundary.
        self._realtime_cursor = _restore_realtime_cursor(self.chatto_config)
        if self._realtime_cursor:
            logger.info("Chatto: resuming the realtime stream from the stored cursor")
        # Start background realtime WS event stream loop.
        self._ws_task = asyncio.create_task(
            self._chattolib_event_loop(),
            name="chatto-event-stream",
        )
        self._presence_task = asyncio.create_task(
            self._presence_refresh_loop(),
            name="chatto-presence-refresh",
        )
        self._mark_connected()

        logger.info(
            "Chatto: connected to %s as '%s' (display_name: '%s' id: '%s')",
            self.chatto_config.base_url.value,
            self.me.login,
            self.me.display_name,
            self.me.id,
        )

        return True

    async def _announce_online(self) -> bool:
        """Tell the server we are online. Returns whether the call got through.

        Logged at warning level on failure: a silently dropped presence call is
        indistinguishable from a bot that is simply not running.
        """
        client = await self._get_chatto_client()
        if client is None:
            logger.warning(
                "Chatto: presence refresh failed, bot may appear offline: no client"
            )
            return False
        try:
            await client.set_presence(status=PresenceStatus.ONLINE)
            return True
        except Exception as exc:
            logger.warning(
                "Chatto: presence refresh failed, bot may appear offline: %s", exc
            )
            return False

    async def _presence_refresh_loop(self) -> None:
        """Re-announce ONLINE until disconnect, since presence expires server-side.

        Failures are not fatal — the next tick tries again, so a blip in the
        presence endpoint costs at most one interval of visible offline time.
        """
        while not self._closing:
            await self._sleep_interruptible(ChattoConstants.PRESENCE_REFRESH_INTERVAL)
            if self._closing:
                return
            await self._announce_online()

    async def disconnect(self) -> None:
        """Stop WebSocket, presence refresh, typing tasks, and clear state.

        BasePlatformAdapter override
        """
        # No explicit offline broadcast: chattolib rejects OFFLINE outright
        # ("stop refreshing to go offline"), so cancelling the refresh loop
        # below is what actually takes the bot offline.
        self._closing = True

        # Store the stream position before anything is torn down: this is the
        # one moment where a restart can resume instead of starting at the live
        # boundary, and it costs one small write.
        _persist_realtime_cursor(self.chatto_config, self._realtime_cursor)

        # Cancel all typing tasks
        for chat_id in list(self._typing_tasks.keys()):
            await self.stop_typing(chat_id)

        if self._ws_task and not self._ws_task.done():
            self._ws_task.cancel()
            try:
                await self._ws_task
            except (asyncio.CancelledError, Exception):
                logger.debug(
                    "Chatto: websocket task ended during disconnect", exc_info=True
                )
            self._ws_task = None

        if self._presence_task and not self._presence_task.done():
            self._presence_task.cancel()
            try:
                await self._presence_task
            except (asyncio.CancelledError, Exception):
                logger.debug(
                    "Chatto: presence task ended during disconnect", exc_info=True
                )
        self._presence_task = None

        if self._chatto_client:
            try:
                await self._chatto_client.close()
            except Exception:
                logger.exception("Chatto: error closing client")
            finally:
                self._chatto_client = None

        logger.info("Chatto: disconnected")
        self._mark_disconnected()

    async def _seed_room(self, room_id: str) -> None:
        """Seed high-water mark from the newest events so a restart doesn't replay history."""
        try:
            client = await self._get_chatto_client()
            if client is None:
                logger.debug(
                    "Chatto: _seed_room aborted - no client available for %s", room_id
                )
                return
            timeline_page = await client.get_room_events(room_id)

            for ev in timeline_page.events:
                if ev.id:
                    self._mark_seen(ev.id)
            logger.debug(
                "Chatto: seeded room %s with %d events",
                room_id,
                len(timeline_page.events),
            )
        except Exception as e:
            logger.warning("Chatto: get room events failed for %s: %s", room_id, e)

    # ------------------------------------------------------------------ #
    # Realtime Event List
    # ------------------------------------------------------------------ #

    def _mark_seen(self, event_id: str) -> None:
        # The deque's maxlen evicts the oldest ID — no manual trimming.
        self._seen.append(event_id)

    def _is_seen(self, event_id: str) -> bool:
        return event_id in self._seen

    # ------------------------------------------------------------------ #
    # WebSocket Realtime Transport
    # ------------------------------------------------------------------ #

    def _own_handles(self) -> set[str]:
        """The handles that address this bot, lowercased for comparison.

        Chatto resolves mentions case-insensitively (FDR-006), so ``@Hermes_Bot``
        and ``@hermes_bot`` are the same handle everywhere in our gates.
        """
        if not self.me:
            return set()
        return {
            handle.lower() for handle in (self.me.login, self.me.display_name) if handle
        }

    def _mention_candidates(self, body: str) -> list[str]:
        """Candidate @-handles in a message body, in order of appearance.

        Mirrors the Chatto web frontend's extraction (apps/frontend/src/lib/
        mentions.ts upstream): candidates come from ``ChattoConstants.
        MENTION_RE`` outside code regions. Mentions inside fenced code blocks
        and inline code spans do not resolve upstream either, so a ``@bob``
        quoted in a snippet must not gate our behaviour.
        """
        without_fences = re.sub(r"(?s)(```|~~~).*?(\1|$)", " ", body)
        without_code = re.sub(r"`[^`\n]*`", " ", without_fences)
        return ChattoConstants.MENTION_RE.findall(without_code)

    def _mentions_from_body(self, body: str) -> _Mentions:
        """Re-derive the addressed parties from message text.

        Needed only for edited messages: an edit event carries no resolved
        targets, and re-gating on the new text is a promise we make on purpose
        (a mention added after posting still reaches the agent) — Chatto itself
        would not re-resolve, but it also has no turn for an edit to open. So
        this is the best available reading, not the authoritative one, and it
        inherits the regex's blind spots: a ``@bob`` inside a URL or blockquote
        reads as a mention here. The posted path does not have this problem —
        it reads the server's own resolution.

        Handles no user holds are not mentions: someone writing *about*
        mentioning ("per @-mention", "@nonexistent") is talking to us, and
        staying silent on a false positive is worse than answering one.
        """
        own = self._own_handles()
        me = False
        someone_else = False
        for handle in self._mention_candidates(body):
            lowered = handle.lower()
            if lowered in ChattoConstants.BROADCAST_MENTIONS or lowered in own:
                me = True
            elif self._handle_belongs_to_a_user(lowered):
                someone_else = True
        return _Mentions(me=me, someone_else=someone_else)

    def _handle_belongs_to_a_user(self, lowered_handle: str) -> bool:
        """Whether ``lowered_handle`` is the login of a real Chatto user.

        Resolved strictly from local knowledge — the member cache the adapter
        already keeps — never from the admin directory: ``admin_get_member``
        needs admin permissions a bot account must not be assumed to hold,
        and in the field its ``permission_denied`` failed every lookup, which
        fail-open let the bot answer every @colleague message. The cache is
        filled by the admission pipeline itself (room rosters are fetched for
        every channel turn), presence patching and the edited-message path.
        """
        return any(
            user is not None and (user.login or "").lower() == lowered_handle
            for user in self._user_cache.values()
        )

    def _check_auth(self, user: User) -> bool:
        """Whether this Chatto user may talk to the agent.

        Deliberately our own gate instead of the gateway's authz_mixin: its
        group allowlists key on chat_type and per-platform env vars
        (``{PLATFORM}_GROUP_ALLOWED_USERS``), none of which fit Chatto's one
        flat member directory. ``CHATTO_ALLOWED_USERS`` matches user IDs,
        ``CHATTO_ALLOW_ALL_USERS`` overrides it — the gate stays
        chat_type-independent by design.
        """
        if self._is_authorized(user.id):
            return True
        logger.warning(
            "Chatto: rejecting message from unauthorized user '%s' (%s)",
            user.login,
            user.id,
        )
        return False

    # ------------------------------------------------------------------ #
    # Room management over DM (/join, /leave)
    # ------------------------------------------------------------------ #

    _DM_COMMANDS = ("/join", "/leave")

    async def _handle_dm_command(self, room_id: str, body: str) -> bool:
        """Run a ``/join`` or ``/leave`` admin command sent as a direct message.

        Returns True when ``body`` is one of the commands — whether it
        succeeded or not — so the caller keeps it out of the agent pipeline.
        Membership lives on the Chatto server: a joined room reappears in
        every future ``list_rooms()`` and therefore survives restarts.
        """
        verb, _, argument = body.strip().partition(" ")
        if verb.lower() not in self._DM_COMMANDS:
            return False

        client = await self._get_chatto_client()
        if client is None:
            await self.send(chat_id=room_id, content="Chatto client is not connected.")
            return True

        argument = argument.strip()
        if not argument:
            await self.send(
                chat_id=room_id,
                content="Usage: /join <room-id or #name> | /leave <room-id or #name>",
            )
            return True

        error, target = await self._resolve_room_target(client, argument)
        if error or target is None:
            await self.send(chat_id=room_id, content=error or "Room lookup failed.")
            return True

        if verb.lower() == "/join":
            reply = await self._run_join(client, target)
        else:
            reply = await self._run_leave(client, target)
        await self.send(chat_id=room_id, content=reply)
        return True

    async def _resolve_room_target(
        self,
        client: ChattoClient,
        argument: str,
    ) -> tuple[str | None, RoomWithViewerState | None]:
        """Resolve a ``/join`` or ``/leave`` argument to a room.

        ``#name`` is looked up case-insensitively in a fresh directory scan
        (which also refreshes our name/kind caches); anything else is treated
        as a room ID and verified via GetRoom. An ambiguous name comes back as
        an error naming the candidates, so the admin can retry with an ID.
        """
        if not argument.startswith("#"):
            state = await client.get_room(argument)
            if state is None or state.room is None:
                return f"No room with ID '{argument}'.", None
            return None, state

        wanted = argument[1:].strip().casefold()
        # (state, room) pairs: a listed match's room is already narrowed here,
        # so the candidate listing below needs no fresh Optional dance.
        matches: list[tuple[RoomWithViewerState, Room]] = []
        for state in await client.list_rooms() or []:
            room_obj = state.room if state else None
            if room_obj and (room_obj.name or "").strip().casefold() == wanted:
                matches.append((state, room_obj))
                self._room_names[room_obj.id] = room_obj.name
                self._room_kinds[room_obj.id] = room_obj.kind
        if not matches:
            return f"No room named '{argument}'.", None
        if len(matches) > 1:
            candidates = "\n".join(f"• {room.name} ({room.id})" for _, room in matches)
            return (
                f"Several rooms are named '{argument}' — pick one by ID:\n{candidates}"
            ), None
        return None, matches[0][0]

    async def _run_join(self, client: ChattoClient, state: RoomWithViewerState) -> str:
        """Join a room via RoomService/JoinRoom and track it as joined.

        An account that already holds membership (invited natively in Chatto)
        needs no JoinRoom call — it only gets seeded and added to the list.
        Channel-kind rooms additionally report their mention-list status,
        because a channel on neither list stays silent and this reply is
        where users copy the room ID from (see ``_room_join_hint``).
        """
        room_obj = state.room
        if room_obj is None:
            # Unreachable via _resolve_room_target: both of its paths only
            # return states whose room they already inspected.
            return "Chatto returned an empty room record — try again."
        label = f"'{room_obj.name}' ({room_obj.id})"
        joined_room = room_obj
        if not state.viewer_state.is_member:
            try:
                joined_room = await client.join_room(room_obj.id) or room_obj
            except ChattoError as exc:
                logger.warning("Chatto: /join failed for %s (%s)", room_obj.id, exc)
                return f"Could not join {label}: {exc}"
        self._room_names[joined_room.id] = joined_room.name
        self._room_kinds[joined_room.id] = joined_room.kind
        if joined_room.id not in self._joined_room_ids:
            # Same rule as _refresh_rooms: silent rooms are joined read-only
            # — seeding history nothing will ever answer would be waste.
            if self._answers_in_room(joined_room.id):
                await self._seed_room(joined_room.id)
            else:
                logger.info(
                    "Chatto: %s is on neither mention list - joined read-only",
                    label,
                )
            self._joined_room_ids.append(joined_room.id)
        if room_obj.kind != RoomKind.DM:
            head = (
                f"Already a member of {label}."
                if state.viewer_state.is_member
                else f"Joined {label}."
            )
            return f"{head}\n{self._room_join_hint(room_obj.id)}"
        if state.viewer_state.is_member:
            return f"Already a member of {label} — listening there."
        return f"Joined {label}."

    async def _run_leave(self, client: ChattoClient, state: RoomWithViewerState) -> str:
        """Leave a room via RoomService/LeaveRoom and drop it from the joined list.

        Two rooms are refused: a DM conversation cannot be left, and leaving
        the configured home channel would silently break cron/notification
        delivery, which posts there through the standalone sender.
        """
        room_obj = state.room
        if room_obj is None:
            # Same invariant as _run_join: _resolve_room_target pre-inspects.
            return "Chatto returned an empty room record — try again."
        label = f"'{room_obj.name}' ({room_obj.id})"
        if room_obj.kind == RoomKind.DM:
            return "Direct messages cannot be left."
        home_id = (self.chatto_config.home_channel.value or "").strip()
        if home_id == room_obj.id:
            return (
                f"{label} is the configured home channel "
                "(CHATTO_HOME_CHANNEL); leaving it would break cron and "
                "notification delivery. Point CHATTO_HOME_CHANNEL elsewhere first."
            )
        try:
            left = await client.leave_room(room_obj.id)
        except ChattoError as exc:
            logger.warning("Chatto: /leave failed for %s (%s)", room_obj.id, exc)
            return f"Could not leave {label}: {exc}"
        if not left:
            return f"Chatto refused to leave {label}."
        if room_obj.id in self._joined_room_ids:
            self._joined_room_ids.remove(room_obj.id)
        # We are no longer part of this audience; keep no roster for it.
        self._evict_roster(room_obj.id)
        return f"Left {label}."

    def _room_join_hint(self, room_id: str) -> str:
        """The mention-list status appended to a channel-kind /join reply.

        Users do not know their room IDs by heart — this reply is where they
        copy them from, so a silent channel names both env vars verbatim,
        ready to paste into ~/.hermes/.env. Configuration resolves once at
        gateway startup, hence the restart note.
        """
        policy = self._room_policy(room_id)
        if policy == RoomPolicy.REQUIRE_MENTION:
            return (
                "This channel answers only @mentions "
                "(listed in CHATTO_REQUIRE_MENTION_ROOMS)."
            )
        if policy == RoomPolicy.OPEN:
            return (
                "This channel answers every message "
                "(listed in CHATTO_OPTIONAL_MENTION_ROOMS)."
            )
        return (
            "This channel stays silent until you list its ID in"
            " ~/.hermes/.env (then restart the gateway):\n"
            f"  CHATTO_REQUIRE_MENTION_ROOMS={room_id}   <- answer only @mentions\n"
            f"  CHATTO_OPTIONAL_MENTION_ROOMS={room_id}  <- answer every message"
        )

    # ------------------------------------------------------------------ #
    # Inbound attachments
    # ------------------------------------------------------------------ #

    async def _download_attachment_bytes(self, url: str) -> bytes:
        """Download an attachment, refusing to buffer more than the gateway cap.

        The Content-Length header is checked first so an oversized asset is
        rejected before a single chunk is read; the running total is re-checked
        as chunks arrive, because a missing or lying header must not smuggle an
        unbounded body past the cap.
        """
        max_bytes = get_inbound_media_max_bytes()
        chunks: list[bytes] = []
        total = 0
        async with (
            httpx.AsyncClient(
                timeout=ChattoConstants.HTTP_TIMEOUT,
                follow_redirects=True,
            ) as http,
            http.stream("GET", url) as resp,
        ):
            resp.raise_for_status()
            declared = resp.headers.get("content-length")
            if declared:
                try:
                    declared_size = int(declared)
                except ValueError:
                    logger.debug("Chatto: ignoring invalid Content-Length %r", declared)
                else:
                    validate_inbound_media_size(
                        declared_size,
                        media_type="attachment",
                        max_bytes=max_bytes,
                    )
            async for chunk in resp.aiter_bytes():
                total += len(chunk)
                validate_inbound_media_size(
                    total,
                    media_type="attachment",
                    max_bytes=max_bytes,
                )
                chunks.append(chunk)
        return b"".join(chunks)

    async def _cache_attachments(
        self,
        attachments: list[MessageAttachment],
    ) -> tuple[list[str], list[str], list[str]]:
        """Download message attachments into the gateway media cache.

        Returns ``(media_urls, media_types, media_kinds)`` — the paths are
        agent-visible cache paths, exactly what ``cache_media_bytes`` yields for
        every other platform.  A failing attachment is logged and skipped: the
        message itself still reaches the agent.
        """
        media_urls: list[str] = []
        media_types: list[str] = []
        media_kinds: list[str] = []

        for att in attachments:
            url = att.asset_url.url if att.asset_url else ""
            filename = att.filename
            content_type = att.content_type
            if not url:
                # Videos are announced before transcoding finishes, so the
                # signed URL can legitimately be missing on arrival.
                logger.debug(
                    "Chatto: attachment '%s' has no asset URL yet, skipping",
                    filename,
                )
                continue
            try:
                data = await self._download_attachment_bytes(url)
                cached = cache_media_bytes(
                    data,
                    filename=filename,
                    mime_type=content_type,
                )
            except Exception as e:
                logger.warning(
                    "Chatto: failed to cache attachment '%s' (%s): %s",
                    filename,
                    content_type,
                    e,
                )
                continue
            if cached is None:
                logger.warning(
                    "Chatto: attachment '%s' (%s) could not be cached, skipping",
                    filename,
                    content_type,
                )
                continue
            media_urls.append(cached.path)
            media_types.append(cached.media_type)
            media_kinds.append(cached.kind)

        return media_urls, media_types, media_kinds

    @staticmethod
    def _message_type_for_media_kinds(media_kinds: list[str]) -> MessageType:
        """Pick the MessageType for a set of cached attachment kinds."""
        if "document" in media_kinds:
            return MessageType.DOCUMENT
        if "image" in media_kinds:
            return MessageType.PHOTO
        if "video" in media_kinds:
            return MessageType.VIDEO
        if "audio" in media_kinds:
            return MessageType.AUDIO
        return MessageType.TEXT

    async def _room_kind_for(
        self, client: ChattoClient, room_id: str
    ) -> RoomKind | None:
        """The room's kind, from cache or a fresh GetRoom lookup.

        Returns ``None`` when the room cannot be resolved — the caller treats
        that as "not dispatchable" rather than guessing a kind.
        """
        kind = self._room_kinds.get(room_id)
        if kind is not None:
            return kind
        room_viewer_state = await client.get_room(room_id)
        if room_viewer_state is None or room_viewer_state.room is None:
            return None
        kind = room_viewer_state.room.kind or RoomKind.UNSPECIFIED
        self._room_kinds[room_id] = kind
        return kind

    def _room_policy(self, room_id: str) -> RoomPolicy:
        """Which of the mention lists a channel-kind room is on.

        The two lists are mutually exclusive (enforced by
        ``hermes_validate_config``), so membership decides: optional beats
        require in the face of contradictory runtime config, and a room on
        neither list stays silent. Callers that only need yes/no ask
        ``_answers_in_room``, which delegates to the shared ``_room_answers_in``
        the send tools use.
        """
        if room_id in self.chatto_config.optional_mention_rooms.value:
            return RoomPolicy.OPEN
        if room_id in self.chatto_config.require_mention_rooms.value:
            return RoomPolicy.REQUIRE_MENTION
        return RoomPolicy.SILENT

    def _answers_in_room(self, room_id: str) -> bool:
        """Whether inbound messages from this room reach the agent pipeline.

        Delegates to the shared ``_room_answers_in``, which the send tools gate
        on too — see there for the rule and why it exists only once. A room
        that answers is never read-only (its policy is not SILENT), and an
        unknown kind falls through to the mention lists rather than being
        mistaken for a DM.
        """
        return _room_answers_in(
            self.chatto_config, room_id, self._room_kinds.get(room_id)
        )

    def toolsets_for_source(self, source: SessionSource) -> list[str] | None:
        """Restrict which toolsets a turn in a listed room may use.

        ``CHATTO_ROOM_TOOLSETS`` holds ``room_id:toolset`` entries. A room with
        at least one entry *replaces* the platform's toolset list for turns in
        that room — Hermes' contract for this hook — so the entries have to name
        every toolset the room should keep. A room without an entry returns
        None, leaving ``platform_toolsets`` in charge. DM rooms are looked up
        like any other room, and a thread turn carries its room's ID, so a room
        entry covers its threads too.

        BasePlatformAdapter override
        """
        room_id = str(getattr(source, "chat_id", "") or "")
        if not room_id:
            return None
        by_room, _malformed = _parse_room_toolsets(
            self.chatto_config.room_toolsets.value
        )
        return by_room.get(room_id) or None

    async def _room_roster_context(
        self, client: ChattoClient, room_id: str
    ) -> _RoomRoster | None:
        """Fetch the room's member roster as a fresh projection.

        The agent only ever sees its prompt: without this block it cannot know
        who else is in a channel, because unaddressed messages are dropped by
        mention gating long before they could teach it a name. Fetched users go
        into _user_cache so presence patching and mention resolution share one
        store. Returns ``None`` on failure or when nothing renderable remains —
        best-effort by design, a directory hiccup must never cost the turn.
        """
        try:
            # list_room_members() in 0.5.0b4 returns member IDs only;
            # batch_get_room_members() hydrates them into DirectoryMember.
            member_id_list, page = await client.list_room_members(
                room_id, limit=ChattoConstants.ROSTER_MEMBER_LIMIT
            )
            if not member_id_list:
                return None
            members = await client.batch_get_room_members(room_id, member_id_list)
        except Exception:
            logger.debug(
                "Chatto: could not list members of room %s", room_id, exc_info=True
            )
            return None
        logger.debug(
            "Chatto: roster for room %s: %d members fetched, total_count=%d",
            room_id,
            len(members),
            page.total_count,
        )

        member_ids: set[str] = set()
        users: list[User] = []
        for member in members:
            user = member.user
            if user is None or user.deleted:
                continue
            member_ids.add(user.id)
            self._user_cache[user.id] = user
            users.append(user)
        entries = [
            entry for entry in (self._roster_entry(user) for user in users) if entry
        ]
        if not entries:
            logger.debug(
                "Chatto: roster for room %s empty after filtering "
                "(%d fetched, all deleted/self/without user)",
                room_id,
                len(members),
            )
            return None
        logger.debug("Chatto: roster for room %s: %d entries", room_id, len(entries))
        unfetched = max(0, page.total_count - len(members))
        return _RoomRoster(member_ids=member_ids, unfetched=unfetched)

    def _roster_entry(self, user: User) -> str:
        """One roster entry, e.g. ``@bob (Bob Example, online)``.

        Skips deleted users and our own account (the agent knows itself);
        display name and presence are optional parts of the parenthetical.
        """
        if user.deleted:
            return ""
        if self.me is not None and user.id == self.me.id:
            return ""
        entry = f"@{user.login}"
        details = [
            part
            for part in (
                user.display_name,
                _PRESENCE_LABELS.get(user.presence_status, ""),
            )
            if part
        ]
        if details:
            entry += f" ({', '.join(details)})"
        return entry

    def _roster_line(self, room_id: str) -> str:
        """Render the room's roster line from its cached projection."""
        roster = self._rosters.get(room_id)
        if roster is None:
            return ""
        users = [self._user_cache.get(member_id) for member_id in roster.member_ids]
        entries = [
            entry
            for entry in (
                self._roster_entry(user) for user in users if user is not None
            )
            if entry
        ]
        more = f" … and {roster.unfetched} more" if roster.unfetched else ""
        return ", ".join(entries) + more

    def _announce_roster(self, room_id: str, roster: _RoomRoster) -> None:
        """Record a freshly fetched roster as the room's projection.

        Bounded like _dispatched_ids: appending past the cap drops the oldest
        room's projection alongside its deque entry.
        """
        if room_id not in self._roster_announced:
            oldest = (
                self._roster_announced.popleft()
                if len(self._roster_announced) == self._roster_announced.maxlen
                else None
            )
            self._roster_announced.append(room_id)
            if oldest is not None:
                self._rosters.pop(oldest, None)
        self._rosters[room_id] = roster

    async def _thread_context_for(
        self,
        client: ChattoClient,
        *,
        room_id: str,
        thread_root_event_id: str,
        current_message_id: str,
        source: Any,
    ) -> str:
        """The thread's earlier messages, delivered once per thread.

        A Hermes thread session is keyed on the thread root and starts empty,
        so the turn that opens a thread is answered with nothing but the reply
        itself. The reported case: a cron job posts into a channel, its echo
        is filtered as a self-event and never reaches a session, and a human
        replies — rooting a thread whose root the agent has never seen. Without
        this block the agent answers a message it cannot read.

        Gated on ``thread_root_event_id``, not on the resolved ``thread_id``:
        a fresh room message is anchored to itself (see _admit_and_build), and
        that thread does not exist yet, so there is nothing to fetch. DMs never
        thread and never seed.

        The cursor is kept twice: in memory for this process, and on the session
        for the next one. Only the second one survives a restart, and it is not
        a redundant copy — a session's frontier is *not* derivable from the
        thread's messages, because the bot's own messages belong to no session
        while the previous turn's trigger looks exactly like one the session
        holds. Rebuilding a partition from the fetch therefore lands on the
        wrong message and skips the very output this exists to deliver.

        Best-effort by design, like the roster: a failed lookup leaves the
        thread unseeded so the next turn retries.
        """
        if not thread_root_event_id:
            return ""

        session_key = self._session_key_for(source)

        # The cursor is resolved *before* the fetch, because it anchors it. That
        # is what makes a long thread work: the server returns the oldest messages
        # past the anchor rather than the newest ones in the thread, so a thread
        # that grew past the fetch limit still answers with what the session is
        # missing instead of leaving the cursor outside the window.
        cursor = self._seeded_threads.get(thread_root_event_id, _UNSEEDED)
        if cursor is _UNSEEDED:
            resumed = self._persisted_cursor(session_key, _UNSEEDED)
            if isinstance(resumed, str):
                # A cursor an earlier process left behind: it outlived that
                # process with the session, so the thread goes on delivering its
                # tail.
                self._record_thread_seeded(thread_root_event_id, resumed)
                cursor = resumed
        anchored = cursor if isinstance(cursor, str) else ""

        history = await self._thread_history(
            client,
            room_id=room_id,
            thread_root_event_id=thread_root_event_id,
            after=anchored,
        )
        if history is None:
            logger.debug(
                "Chatto: dispatching into thread %s without history (lookup failed)",
                thread_root_event_id,
            )
            return ""

        if anchored:
            # Everything the anchored page holds landed after the cursor, so it is
            # the tail — no cutting point to find inside the window.
            return self._seed_thread_tail(
                thread_root_event_id,
                history,
                current_message_id,
                session_key=session_key,
            )

        if self._thread_session_exists(source):
            # A session that survived a restart but carries no cursor: a thread
            # whose first turn had nothing to seed wrote no cursor (there was no
            # session to write it to) and never got a second turn to retry on.
            #
            # Its frontier is not derivable from the message list, so this
            # delivers the one thing that is provably at risk — the bot's own
            # messages, which belong to no session — and leaves the people's
            # messages alone, since those became turns. Duplicating the former
            # is possible (a first seed that did carry them, same lost second
            # turn) and is bounded to that one block; losing it is the common
            # case and permanent. The cursor is written either way, so the thread
            # is blind once, not for the rest of its life.
            own = [m for m in history if m.id != current_message_id and self._is_own(m)]
            logger.debug(
                "Chatto: thread %s has a session without a seed cursor; "
                "re-delivering %d own message(s)",
                thread_root_event_id,
                len(own),
            )
            established = self._seeded_through(history, own, current_message_id)
            if established is not None:
                self._record_thread_seeded(thread_root_event_id, established)
                self._persist_thread_cursor(
                    session_key, thread_root_event_id, established
                )
            return self._thread_context_block(own)

        # First turn into this thread: the session starts empty, so hand it the
        # whole thread before the reply.
        earlier = [m for m in history if m.id != current_message_id]
        self._remember_thread_cursor(
            session_key,
            thread_root_event_id,
            self._seeded_through(history, earlier, current_message_id),
            delivered=earlier,
        )
        return self._thread_context_block(earlier)

    @staticmethod
    def _seeded_through(
        history: list[Message], delivered: list[Message], current_message_id: str
    ) -> str:
        """How far this turn's session now reaches.

        The trigger is excluded from the *block* — it is the message text — but
        the session holds it, so the cursor passes it too. Leaving it behind
        would hand the next turn the previous question a second time, as an
        `[Earlier in this thread]` line beside the conversation that already
        contains it.

        The trigger is also the fallback when the fetched window does not list
        it: the session holds this turn's own message either way. That is safe
        because the fetch is anchored *at* the cursor rather than sliced around
        it — a cursor names a boundary, not a position inside the window, so one
        the window does not contain is still a usable cursor. That is what lets
        this be a plain ``str`` instead of "maybe unknowable".
        """
        if any(m.id == current_message_id for m in history):
            return current_message_id
        return delivered[-1].id if delivered else current_message_id

    def _seed_thread_tail(
        self,
        thread_root_event_id: str,
        history: list[Message],
        current_message_id: str,
        *,
        session_key: str,
    ) -> str:
        """Deliver what landed in an already-seeded thread since the cursor.

        This is the agent's own output the session never saw: a message it posted
        into the thread through a tool, whose realtime echo is filtered as a
        self-event. Without it the next human message in that thread is answered
        by an agent that does not know what it said.

        ``history`` is anchored at the cursor, so every message in it is by
        definition one the session does not hold — there is no cut point to find
        inside the window, and no case where the window cannot answer.
        """
        unseen = [m for m in history if m.id != current_message_id]
        # The cursor moves even when the trigger was all that arrived: it is in
        # the session, and a cursor left behind it would hand the next turn the
        # previous question a second time.
        self._remember_thread_cursor(
            session_key,
            thread_root_event_id,
            self._seeded_through(history, unseen, current_message_id),
            delivered=unseen,
        )
        return self._thread_context_block(unseen)

    async def _thread_history(
        self,
        client: ChattoClient,
        *,
        room_id: str,
        thread_root_event_id: str,
        after: str = "",
    ) -> list[Message] | None:
        """Fetch a thread's messages oldest-first, or ``None`` if the lookup failed.

        ``after`` anchors the page at a cursor instead of at the thread's start,
        and that is the whole point: the server then returns the *oldest* messages
        past that cursor rather than the newest ones in the thread. A thread that
        has grown past the fetch limit therefore still answers with exactly what
        the session is missing, where an unanchored page would have left the
        cursor outside the window and the turn with nothing to deliver.

        Two things follow from the anchor, both from the server's own handler: the
        root is not part of an anchored page (``IncludeRoot`` is false whenever
        ``after`` is set), and it is not needed either — a cursor means
        everything up to and including it is already in the session. So the extra
        root fetch is skipped, which is one RPC fewer per turn than the unanchored
        path.

        When the gap is larger than the limit the page holds only its oldest
        slice and the cursor stops at the last message delivered; the next turn
        continues from there. A backlog drains over turns rather than in one, and
        nothing is skipped either way.
        """
        try:
            page = await client.get_thread_events(
                room_id,
                thread_root_event_id,
                limit=ChattoConstants.THREAD_CONTEXT_MESSAGE_LIMIT,
                after=after or None,
            )
        except Exception:
            logger.debug(
                "Chatto: get thread events failed for %s/%s",
                room_id,
                thread_root_event_id,
                exc_info=True,
            )
            return None

        messages = [ev.message for ev in page.events if ev.message is not None]
        if not after and not any(m.id == thread_root_event_id for m in messages):
            # The thread root is fetched explicitly when the unanchored timeline
            # omits it: a cron message is a plain room message until someone
            # replies to it, and whether the server lists it back as the thread's
            # first event is not something the client signature promises.
            messages = (
                await self._thread_root_message(
                    client, room_id=room_id, thread_root_event_id=thread_root_event_id
                )
                + messages
            )
        return _oldest_first(messages)

    async def _thread_root_message(
        self, client: ChattoClient, *, room_id: str, thread_root_event_id: str
    ) -> list[Message]:
        """The thread root as a list, so a miss concatenates cleanly.

        Best-effort: an unknown root costs nothing but the missing root line.
        """
        try:
            root = await client.get_message(
                room_id=room_id, event_id=thread_root_event_id
            )
        except Exception:
            logger.debug(
                "Chatto: get thread root failed for %s/%s",
                room_id,
                thread_root_event_id,
                exc_info=True,
            )
            return []
        return [root] if root is not None and not root.deleted_at else []

    def _remember_thread_cursor(
        self,
        session_key: str,
        thread_root_event_id: str,
        cursor: str,
        *,
        delivered: list[Message],
    ) -> None:
        """Record how far a thread's session reaches, and mirror it if it must.

        The in-memory map is authoritative while this process lives. The session
        copy exists for one thing — a restart, where the map is empty and the
        session survived — and it is written when either of two things holds:

        - the turn put the bot's own output into the session. That is the one
          thing a later turn cannot recover from the message list, so it is the
          one cursor that has to outlive us. Every other turn's frontier is the
          newest message that is not ours, and a restart that loses such a cursor
          delivers nothing, which is right because that turn delivered nothing
          worth losing.
        - an earlier turn's cursor is still waiting for a session to land on. A
          thread's first turn has none — and its seed is exactly the case above,
          a cron message delivered to a session that did not exist yet. By the
          second turn the session exists.

        Writing unconditionally would cost a full routing-table rewrite per turn
        (see _persist_thread_cursor) to save a case that costs nothing.
        """
        self._record_thread_seeded(thread_root_event_id, cursor)
        unrecoverable = any(self._is_own(message) for message in delivered)
        if unrecoverable or thread_root_event_id in self._unpersisted_threads:
            self._persist_thread_cursor(session_key, thread_root_event_id, cursor)

    def _persist_thread_cursor(
        self, session_key: str, thread_root_event_id: str, cursor: str
    ) -> None:
        """Mirror the cursor onto the session, remembering a refusal.

        ``set_session_metadata`` rewrites the whole routing index on every call,
        which is why the caller restricts this to the turns that need it. A
        refusal costs nothing here — the in-memory cursor carries the thread
        until the next process — so it is remembered and retried on the next turn
        rather than logged away, because a cursor that never lands is exactly the
        cursor a restart needs.
        """
        store = getattr(self, "_session_store", None)
        writer = getattr(store, "set_session_metadata", None)
        if not callable(writer):
            return
        try:
            landed = bool(writer(session_key, _CURSOR_META_KEY, cursor))
        except Exception:
            logger.debug(
                "Chatto: persisting the seed cursor for %s failed",
                session_key,
                exc_info=True,
            )
            return
        if landed:
            self._unpersisted_threads.discard(thread_root_event_id)
        else:
            # No session under that key yet, which is the expected shape of a
            # thread's first turn.
            self._unpersisted_threads.add(thread_root_event_id)

    def _persisted_cursor(self, session_key: str, default: _T) -> str | _T:
        """The cursor a previous process left on this session, or ``default``.

        Best-effort bookkeeping, so every failure answers ``default``: a missing
        store, a store without metadata support, a rejected read. That degrades
        to "this thread's frontier is unknown", never to a failed turn.
        """
        store = getattr(self, "_session_store", None)
        reader = getattr(store, "get_session_metadata", None)
        if not callable(reader):
            return default
        try:
            # Duck-typed: the store is optional and may not carry metadata at
            # all, so its value comes back as whatever it is.
            value: Any = reader(session_key, _CURSOR_META_KEY, None)
        except Exception:
            logger.debug(
                "Chatto: reading the seed cursor for %s failed",
                session_key,
                exc_info=True,
            )
            return default
        return value if value is not None else default

    async def _reply_target(
        self, client: ChattoClient, room_id: str, in_reply_to: str
    ) -> Message | None:
        """The message a reply points at, or ``None`` when it cannot be read.

        Best-effort on purpose: a deleted or unreadable target costs the
        context line, not the turn — an unanswered reply is worse than one
        answered without its quote.
        """
        try:
            target = await client.get_message(room_id=room_id, event_id=in_reply_to)
        except Exception:
            logger.debug(
                "Chatto: reply target %s/%s could not be read",
                room_id,
                in_reply_to,
                exc_info=True,
            )
            return None
        if target is None or target.deleted_at:
            return None
        return target

    def _is_own(self, message: Message) -> bool:
        """Whether this message is one the bot itself posted."""
        return self.me is not None and message.actor_id == self.me.id

    def _record_thread_seeded(
        self, thread_root_event_id: str, last_delivered_id: str
    ) -> None:
        """Note that this thread's session holds its history up to that message.

        Re-inserting keeps the map insertion-ordered, which is what the cap below
        evicts from: the least recently touched thread.
        """
        self._seeded_threads.pop(thread_root_event_id, None)
        self._seeded_threads[thread_root_event_id] = last_delivered_id
        while len(self._seeded_threads) > ChattoConstants.SEEN_CAP:
            self._seeded_threads.pop(next(iter(self._seeded_threads)))

    def _thread_session_exists(self, source: Any) -> bool:
        """Whether the gateway already has a session for this source.

        Kept as the thread path's own name: it answers the same question as
        ``_session_exists`` and delegates to it, so the two paths cannot
        disagree about what "a session exists" means.
        """
        return self._session_exists(source)

    async def _room_context_for(
        self,
        client: ChattoClient,
        *,
        room_id: str,
        trigger: Message,
        source: Any,
    ) -> str:
        """The room's earlier messages, delivered when a room session starts.

        A fresh room message roots its own session (see _admit_and_build), so
        the first turn into a room is answered with nothing but that message —
        even when the room has been talking for hours and even when the bot
        itself posted there. The reported case: a human replies to the bot's
        message without opening a thread, and the agent answers a conversation
        it has never seen.

        The block is bounded by the same caps as the thread seed, and it is
        skipped as soon as the session holds a turn: the goal is to open a room
        conversation with its recent context, not to re-deliver the room's
        history on every message.

        Seeding is anchored on the trigger, not on a per-room cursor: every
        fresh room message opens a new session, so a cursor written for one
        session's frontier would hand the next message's session a block
        written for the first one's. The trigger-anchored fetch makes the
        block "what the room said before this message", which is the question
        a session start actually asks.
        """
        if self._session_exists(source):
            return ""

        try:
            page, _anchor_index = await client.get_room_events_around(
                room_id, trigger.id, limit=ChattoConstants.THREAD_CONTEXT_MESSAGE_LIMIT
            )
        except Exception:
            logger.debug(
                "Chatto: room context lookup failed for %s/%s",
                room_id,
                trigger.id,
                exc_info=True,
            )
            return ""
        messages = [ev.message for ev in page.events if ev.message is not None]
        # The page carries its own user directory, which answers the author
        # names in the block without a member-list call — the same hydration
        # the read tools use for a fetched timeline.
        for user_id, user in page.users_by_id.items():
            self._user_cache.setdefault(user_id, user)
        # ``around`` serves both directions of the trigger, and a session
        # start is answered with the past, not the future: newer messages are
        # not "earlier", and including one would hand the agent a message the
        # room may still be typing around. Age is compared by time rather
        # than by position, because the page's order is display order, not a
        # promise about either side of the anchor.
        trigger_time = trigger.created_at or datetime.min.replace(tzinfo=UTC)
        earlier = [
            m
            for m in _oldest_first(messages)
            if m.id != trigger.id
            and not m.deleted_at
            and (m.created_at or datetime.min.replace(tzinfo=UTC)) <= trigger_time
        ]
        return self._earlier_context_block(earlier, "[Earlier in this room]")

    def _session_exists(self, source: Any) -> bool:
        """Whether the gateway already has a session for this source.

        The gateway's own routing index is the authority on "does this
        conversation already exist" — it is persistent (state.db), so unlike
        the in-memory dedup state it answers correctly after a restart.
        Returns False when no store is wired (tests, or a startup path before
        ``set_session_store`` runs) so we fail open and seed.
        """
        store = getattr(self, "_session_store", None)
        if store is None:
            return False
        return store.peek_session_id(self._session_key_for(source)) is not None

    def _thread_context_block(self, messages: list[Message]) -> str:
        """Render earlier thread messages as the agent's reference block.

        Empty when there is nothing to say — a brand-new thread has no earlier
        messages, and that is not worth a header. Over budget, the oldest
        messages go first: the reply's immediate predecessors are what the
        turn is about.
        """
        return self._earlier_context_block(messages, "[Earlier in this thread]")

    def _earlier_context_block(self, messages: list[Message], header: str) -> str:
        """Render earlier messages as one reference block under ``header``.

        Shared by the thread and the room seed: both answer the same question
        ("what did this conversation say before the trigger"), so both get the
        same rendering, the same caps and the same own/verified labelling — two
        copies of one renderer would drift until a thread and a room disagreed
        about what a quoted message looks like.
        """
        entries: list[str] = []
        for message in messages:
            entry = self._thread_message_entry(message)
            if entry:
                entries.append(entry)

        if not entries:
            return ""
        dropped = 0
        while (
            entries
            and len("\n".join(entries)) > ChattoConstants.THREAD_CONTEXT_CHAR_LIMIT
        ):
            entries.pop(0)
            dropped += 1
        if not entries:
            return ""
        body = "\n".join(entries)
        omitted = f"… {dropped} earlier message(s) omitted\n" if dropped else ""
        return f"{omitted}{header}\n{body}"

    def _thread_message_entry(self, message: Message) -> str:
        """One rendered message for an earlier-context block, or ``""``.

        Attachment-only messages render empty: the block is text context, and
        inventing a placeholder for a picture the agent cannot see here would
        be noise.
        """
        body = (message.body or "").strip()
        if not body:
            return ""
        me = self.me
        user = self._user_cache.get(message.actor_id)
        if me is not None and message.actor_id == me.id:
            # The agent's own output — often a cron message it never saw
            # arrive. Labelling it keeps "I said this" distinguishable from a
            # third party talking.
            return f"{me.login} (this bot): {body}"
        author = f"@{user.login}" if user else f"@{message.actor_id}"
        if not self._is_authorized(message.actor_id):
            # Auth gates the trigger message only; the rest of a shared thread
            # is reference text from people who may not be on the allowlist,
            # so it is marked as unverified rather than presented as trusted
            # input. An unknown author lands here too — fail closed.
            return f"{author} (unverified): {body}"
        return f"{author}: {body}"

    def _is_authorized(self, actor_id: str) -> bool:
        """Whether a Chatto user may talk to the agent, by user ID.

        The allowlist test itself, without the logging _check_auth adds: a
        rendering path asks the same question about someone quoted into the
        prompt, where denial is a normal sight rather than an event worth a
        warning. IDs only — a login entry would pass this intake but fail the
        gateway's fresh-role recheck, which matches raw IDs, and the message
        would be silently dropped between the two gates.
        """
        if self.chatto_config.allow_all_users.value:
            return True
        return actor_id in self.chatto_config.allowed_users.value

    def _interactive_prompt_expired(self, prompt: Any) -> bool:
        """Whether an interactive prompt has passed its expiry deadline."""
        try:
            return time.monotonic() > float(prompt.expires_at)
        except (TypeError, ValueError, AttributeError):
            return False

    async def _try_handle_interactive_reaction(
        self,
        room_id: Any,
        message_id: Any,
        emoji: Any,
        actor_id: Any,
    ) -> bool:
        """Route a reaction on a tracked interactive prompt.

        Returns True when the message is a tracked prompt (consumed), False
        otherwise so the caller forwards normally.
        """
        if not room_id or not message_id or not emoji or not actor_id:
            return False
        room_id = str(room_id)
        message_id = str(message_id)
        emoji = str(emoji)
        actor_id = str(actor_id)

        is_clarify = message_id in self._clarify_prompts
        is_approval = message_id in self._approval_prompts
        if not is_clarify and not is_approval:
            return False

        if is_clarify:
            prompt = self._clarify_prompts.get(message_id)
        else:
            prompt = self._approval_prompts.get(message_id)
        if prompt is None:
            return False

        # Room must match.
        if str(getattr(prompt, "chat_id", "")) != room_id:
            return True

        # Expiry handling.
        if self._interactive_prompt_expired(prompt):
            try:
                await self.edit_message(
                    room_id, message_id, "⏰ This prompt has expired."
                )
            except Exception:
                logger.debug("Chatto: expired prompt edit failed", exc_info=True)
            if is_clarify:
                self._clarify_prompts.pop(message_id, None)
                cid = getattr(prompt, "clarify_id", None)
                if cid is not None:
                    self._clarify_id_to_message.pop(cid, None)
            else:
                self._approval_prompts.pop(message_id, None)
            return True

        # Authorization (fail-closed).
        if not self._is_authorized(actor_id):
            return True

        # Double-click guard.
        if getattr(prompt, "resolved", False):
            return True
        if is_clarify and getattr(prompt, "awaiting_text", False):
            return True

        user = self._user_cache.get(actor_id)
        user_label = (
            getattr(user, "name", None) or getattr(user, "login", None) or actor_id
        )

        if is_clarify:
            from tools import clarify_gateway  # type: ignore[import-not-found]

            number_emojis = ["1️⃣", "2️⃣", "3️⃣", "4️⃣", "5️⃣", "6️⃣", "7️⃣", "8️⃣", "9️⃣"]
            other_emoji = "✏️"
            choices = list(getattr(prompt, "choices", []) or [])

            if emoji == other_emoji:
                prompt.awaiting_text = True  # type: ignore[assignment]
                try:
                    clarify_gateway.mark_awaiting_text(prompt.clarify_id)  # type: ignore[attr-defined]
                except Exception:
                    logger.debug("Chatto: mark_awaiting_text failed", exc_info=True)
                try:
                    await self.edit_message(
                        room_id, message_id, f"awaiting text from {user_label}"
                    )
                except Exception:
                    logger.debug("Chatto: prompt edit failed", exc_info=True)
                return True

            if emoji in number_emojis:
                idx = number_emojis.index(emoji)
                if 0 <= idx < len(choices):
                    choice = choices[idx]
                    prompt.resolved = True
                    self._clarify_prompts.pop(message_id, None)
                    cid = getattr(prompt, "clarify_id", None)  # type: ignore[arg-type]
                    if cid is not None:
                        self._clarify_id_to_message.pop(cid, None)
                    try:
                        clarify_gateway.resolve_gateway_clarify(str(cid), choice)
                    except Exception:
                        logger.exception("Chatto: resolve_gateway_clarify failed")
                    try:
                        await self.edit_message(
                            room_id,
                            message_id,
                            f"✅ {user_label} chose: {choice}",
                        )
                    except Exception:
                        logger.debug("Chatto: prompt edit failed", exc_info=True)
                    return True

            # Invalid emoji feedback.
            try:
                await self.send(
                    room_id, "That reaction isn't a valid option for this prompt."
                )
            except Exception:
                logger.debug("Chatto: invalid reaction feedback failed", exc_info=True)
            return True

        # Approval prompt.
        choices_map = getattr(prompt, "choices_map", {}) or {}
        if emoji in choices_map:
            choice = choices_map[emoji]
            prompt.resolved = True
            self._approval_prompts.pop(message_id, None)
            try:
                import tools.approval as approval_mod

                approval_mod.resolve_gateway_approval(
                    str(getattr(prompt, "session_key", "")), choice
                )
            except Exception:
                logger.exception("Chatto: resolve_gateway_approval failed")
            try:
                await self.edit_message(
                    room_id, message_id, f"✅ {user_label}: {choice}"
                )
            except Exception:
                logger.debug("Chatto: prompt edit failed", exc_info=True)
            return True

        # Invalid emoji feedback.
        try:
            await self.send(
                room_id, "That reaction isn't a valid option for this prompt."
            )
        except Exception:
            logger.debug("Chatto: invalid reaction feedback failed", exc_info=True)
        return True

    def _evict_roster(self, room_id: str) -> None:
        """Drop all roster state for a room.

        Called on membership changes (user_joined/left_room), when we leave a
        room ourselves, and on reconnect. The subscribe snapshot carries the
        room and user directory but no presence, and the adapter does not
        request it — so a discarded cache is what forces one honest refetch.
        """
        try:
            self._roster_announced.remove(room_id)
        except ValueError:
            pass
        self._rosters.pop(room_id, None)

    async def _roster_for_room(self, client: ChattoClient, *, room_id: str) -> str:
        """The room's roster line for this turn — delivered on every turn.

        Threads hold isolated sessions and the agent only sees
        channel_context per dispatch, so each channel turn carries the current
        audience rather than deduplicating it. The projection exists to spare
        the directory, not the prompt: presence patches land in _user_cache
        in place, membership events or a reconnect evict the room so its next
        turn refetches once. An unprojected room retries on its next turn
        until a lookup succeeds.
        """
        if room_id not in self._roster_announced:
            fetched = await self._room_roster_context(client, room_id)
            if fetched is None:
                return ""
            self._announce_roster(room_id, fetched)
        return self._roster_line(room_id)

    async def _dispatch_message_posted(
        self, payload: MessagePostedEvent, event_id: str
    ) -> None:
        # Respond-room gate first: read-only memberships must not cost a
        # single API call, so this runs before fetch_message and _get_chatto_client.
        if not self._answers_in_room(payload.room_id):
            logger.debug(
                "Chatto: message from read-only room %s ignored", payload.room_id
            )
            return

        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto: dropping message - no client available")
            return
        logger.debug("Chatto WS: 'message_posted' payload:%s", payload)

        message = await client.get_message(room_id=payload.room_id, event_id=event_id)
        if message is None or message.deleted_at:
            return
        message_body = message.body or ""
        logger.debug("message: %s", message)

        # A message carrying only an image/PDF has an empty body — dropping it
        # here is what made attachments sent to Hermes disappear silently.
        if not message_body and not message.attachments:
            return

        event = await self._admit_and_build(
            client,
            room_id=payload.room_id,
            message=message,
            source_message_id=event_id,
            thread_root_event_id=payload.thread_root_event_id or None,
            mentions=_Mentions.from_payload(
                payload.mentions, self.me.id if self.me else ""
            ),
        )
        if event is None:
            return

        self._remember_dispatched(event.message_id or "")
        logger.info("Chatto: dispatching message to Hermes")
        await self.handle_message(event)
        return

    def _remember_dispatched(self, message_id: str) -> None:
        """Record a message ID as handed to the gateway, capped like _seen."""
        if not message_id:
            return
        self._dispatched_ids.append(message_id)

    def _edit_is_fresh(self, message: Message) -> bool:
        """Whether this edit is young enough to still be processed.

        Age is measured against the posting time, so an edit to an hours-old
        message cannot resurrect a settled conversation even when it arrives
        right now.
        """
        if message.created_at is None or message.updated_at is None:
            return True
        age_seconds = (message.updated_at - message.created_at).total_seconds()
        return age_seconds <= self.chatto_config.edit_window.value

    def _session_key_for(self, source) -> str:
        """The gateway's own session key for this source.

        Delegates to the base class rather than calling ``build_session_key``
        itself. A hand-built key has to be right about two things this one was
        not: the routing identity has to be pinned first (``_canonicalize``), and
        the key has to carry the serving profile's namespace. Without the second,
        a multiplexed gateway keys every bot into ``agent:main:`` and the derived
        key names a session nobody is running — which made
        ``_thread_session_exists`` answer False for a live thread session, and
        would make ``cancel_session_processing`` a no-op on an edited message.
        """
        return self._source_session_key(source)

    async def _dispatch_message_edited(self, payload: Any) -> None:
        """Route an inbound edit according to the edit-dispatch contract.

        Three outcomes for the edited message:

        - currently being processed → cancel that turn and re-dispatch with
          the corrected text (the cancelled turn reports 🚫 via its
          CANCELLED outcome hook),
        - never dispatched (e.g. a forgotten @mention added later) → re-run
          the admission gates against the new body and answer for real,
        - already answered → stay answered.

        Edits whose text parses as a DM membership command are dropped: the
        command ran when the message was posted and must not run again.
        """
        if not self.chatto_config.edit_dispatch.value:
            return
        # Read-only memberships cost no API call, mirroring the posted path.
        if not self._answers_in_room(payload.room_id):
            logger.debug("Chatto: edit from read-only room %s ignored", payload.room_id)
            return

        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto: dropping edit - no client available")
            return
        logger.debug("Chatto WS: 'message_edited' payload:%s", payload)

        message = await client.get_message(
            room_id=payload.room_id, event_id=payload.message_event_id
        )
        if message is None or message.deleted_at:
            return
        message_body = message.body or ""
        if not message_body and not message.attachments:
            return
        if not self._edit_is_fresh(message):
            logger.debug(
                "Chatto: edit of msg %s outside the edit window",
                payload.message_event_id,
            )
            return

        # Cheap triage before the admission pipeline: an edit to an
        # already-settled message (dispatched, neither running nor queued)
        # must not cost get_room/media calls or re-fire acknowledgements.
        was_dispatched = payload.message_event_id in self._dispatched_ids
        if (
            was_dispatched
            and payload.message_event_id not in self._processing.values()
            and not any(
                pending.message_id == payload.message_event_id
                for pending in self._pending_messages.values()
            )
        ):
            logger.debug(
                "Chatto: edit of already-answered msg %s ignored",
                payload.message_event_id,
            )
            return

        event = await self._admit_and_build(
            client,
            room_id=payload.room_id,
            message=message,
            source_message_id=payload.message_event_id,
            thread_root_event_id=message.thread_root_event_id or None,
            allow_dm_commands=False,
        )
        if event is None:
            return

        session_key = self._session_key_for(event.source)
        if self._processing.get(session_key) == payload.message_event_id:
            logger.info(
                "Chatto: msg %s edited mid-run - restarting the turn",
                payload.message_event_id,
            )
            # The cancelled task's completion hook clears _processing and
            # reports 🚫 before this coroutine moves on, because cancel
            # awaits the task. Queued follow-ups must survive.
            await self.cancel_session_processing(
                session_key, release_guard=True, discard_pending=False
            )
        elif not was_dispatched:
            logger.info(
                "Chatto: msg %s was never dispatched - edit starts a fresh turn",
                payload.message_event_id,
            )
        else:
            pending = self._pending_messages.get(session_key)
            if pending is not None and pending.message_id == payload.message_event_id:
                # Still queued behind the running turn: correct it in place
                # instead of answering stale wording later.
                pending.text = event.text
                logger.info(
                    "Chatto: queued msg %s updated to its edited text",
                    payload.message_event_id,
                )
                return
            logger.debug(
                "Chatto: edit of already-answered msg %s ignored",
                payload.message_event_id,
            )
            return

        self._remember_dispatched(event.message_id or "")
        logger.info("Chatto: dispatching edited message to Hermes")
        await self.handle_message(event)

    async def _admit_and_build(
        self,
        client: ChattoClient,
        *,
        room_id: str,
        message: Message,
        source_message_id: str,
        thread_root_event_id: str | None,
        mentions: _Mentions | None = None,
        allow_dm_commands: bool = True,
    ) -> MessageEvent | None:
        """Run one hydrated inbound message through the admission pipeline.

        Shared by the posted and the edited path: user resolution, auth,
        mention gates, thread anchoring and media caching all behave
        identically for both. Returns ``None`` for anything that must not
        reach the agent. DM membership commands are executed here (side
        effect) unless ``allow_dm_commands`` is False — edits pass False so
        a corrected command line neither runs twice nor leaks to the agent.
        A dispatch that opens a channel thread additionally carries the
        member roster in ``channel_context`` (side effect: once the roster
        was fetched, the thread root is recorded as announced).

        ``mentions`` is the already-resolved answer to "who does this
        address"; passing None derives it from the message body instead, which
        is what an edit does because its event carries no targets. The
        derivation happens down here rather than at the call site because
        resolving a handle needs the member cache, and the roster fetch below
        is what fills it — read any earlier, every handle looks unknown.

        The caller owns dispatching: a non-None result still needs
        ``handle_message()``.
        """
        user = self._user_cache.get(message.actor_id)
        if user is None:
            logger.debug("Chatto: resolving user %s", message.actor_id)
            member = await client.get_room_member(
                room_id=message.room_id, user_id=message.actor_id
            )
            if member is None or member.user is None:
                logger.debug(
                    "Chatto: dropping message from unknown user %s",
                    message.actor_id,
                )
                return None
            user = member.user
            self._user_cache[user.id] = user

        if not self._check_auth(user):
            logger.debug(
                "Chatto: dropping message from user %s (auth denied)",
                message.actor_id,
            )
            return None

        room_kind = await self._room_kind_for(client, message.room_id)
        if room_kind is None:
            logger.debug(
                "Chatto: dropping message from room %s (room_kind lookup failed)",
                message.room_id,
            )
            return None

        message_body = message.body or ""

        logger.debug("message_body: %s room_kind: %s", message_body, room_kind)

        # Membership commands ride in over DMs only: they change what the bot
        # listens to and must never reach the agent pipeline or the mention
        # gates.
        if room_kind == RoomKind.DM:
            if allow_dm_commands:
                if await self._handle_dm_command(room_id, message_body):
                    return None
            elif message_body.startswith("/"):
                logger.debug("Chatto: edited DM command %r not re-run", message_body)
                return None

        # Mention gating deliberately covers everything but DMs: in a channel
        # the bot is one of many listeners and must be addressed, whereas a DM
        # is already addressed at it. Chatto has no group rooms — any
        # multi-participant surface is a channel, and a room whose kind the
        # server never set counts as one too (see _answers_in_room). Both gates
        # read one resolved `mentions`, so "addressed" is defined once and
        # broadcast/role targets count for us on either path.
        if room_kind != RoomKind.DM:
            policy = self._room_policy(room_id)

            # Handle resolution on the body-derived path reads the member
            # cache, which the roster projection fills. Fetch it before deriving
            # anything, or every handle looks unknown — the same projection is
            # then reused for this turn's channel_context further down, one
            # fetch per room.
            if policy == RoomPolicy.OPEN:
                await self._roster_for_room(client, room_id=room_id)

        if mentions is None:
            mentions = self._mentions_from_body(message_body)

        if room_kind != RoomKind.DM:
            policy = self._room_policy(room_id)
            if policy == RoomPolicy.REQUIRE_MENTION and not mentions.me:
                logger.debug(
                    "Chatto: dropping unaddressed message from %s (policy %s)",
                    room_id,
                    policy.value,
                )
                return None

            # In an open channel we see every message, including ones plainly
            # aimed at a named colleague. Answering those would be barging in,
            # so acknowledge that we read it and stay quiet. Checked after the
            # bot-mention test above, so a message naming us *and* someone
            # else still counts as ours.
            if policy == RoomPolicy.OPEN and not mentions.me and mentions.someone_else:
                logger.info(
                    "Chatto: message addresses someone else, acknowledging only"
                )
                if self.chatto_config.reactions.value:
                    await self.add_reaction(message.room_id, message.id, "🫥")
                return None

        # Thread anchoring — if the incoming message is inside a Chatto thread, we
        # keep that thread by default; otherwise leave thread_id unset so
        # replies land at the root.
        thread_id = (
            thread_root_event_id or None
        )  # we could also take the room id but then, we're in a thread already.
        if not thread_id and room_kind != RoomKind.DM:
            thread_id = message.id
        logger.debug(
            "Chatto: thread anchoring for msg %s: incoming thread_root_event_id=%r, "
            "room_kind=%s, anchored thread_id=%r",
            message.id,
            message.thread_root_event_id,
            room_kind.name,
            thread_id,
        )

        # Every channel turn carries the room's roster: threads hold isolated
        # sessions and the agent only sees channel_context per dispatch, so a
        # repeat is not deduplication but the point (see _roster_for_room).
        # The gateway prepends channel_context above the message text, so the
        # roster never mingles with what the user actually wrote.
        source = self.build_source(
            chat_id=room_id,
            chat_name=self._room_names.get(message.room_id),
            chat_type=chat_type_for_room_kind(room_kind),
            user_id=message.actor_id,
            user_name=user.login,  # use login, because display_name is changeable by anyone.
            thread_id=thread_id,
            message_id=source_message_id,
            role_authorized=True,
        )

        # Channel context rides ahead of the message text as read-only
        # reference: the roster names the audience, the thread block carries
        # what the thread already said. DMs get neither — there is no audience
        # to name and no thread to catch up on. Built after the source, because
        # the thread block asks the session store whether this thread's session
        # already has the history (see _thread_context_for).
        context_blocks: list[str] = []
        if room_kind != RoomKind.DM:
            logger.debug(
                "Chatto: fetching roster for room %s",
                room_id,
            )
            roster = await self._roster_for_room(client, room_id=room_id)
            if roster:
                context_blocks.append(roster)
            else:
                logger.debug(
                    "Chatto: dispatching channel message without roster "
                    "(lookup failed or empty)"
                )
            thread_context = await self._thread_context_for(
                client,
                room_id=room_id,
                thread_root_event_id=thread_root_event_id or "",
                current_message_id=message.id,
                source=source,
            )
            if thread_context:
                context_blocks.append(thread_context)

            # A fresh room message roots its own session (the anchoring above),
            # so it starts empty even when the room talked for hours. The reply
            # pointer above carries what this message answers; this block adds
            # what the room said before it — including the bot's own output,
            # whose echo is filtered as a self-event and reaches no session
            # any other way.
            if not thread_root_event_id:
                room_context = await self._room_context_for(
                    client,
                    room_id=room_id,
                    trigger=message,
                    source=source,
                )
                if room_context:
                    context_blocks.append(room_context)

        # A reply must not read as a fresh conversation: the gateway renders
        # ``reply_to_text`` into the turn as "[Replying to …]", but only when the
        # adapter resolved it. Signal does this for quotes; a Chatto reply's
        # target is one get_message away, so a session that starts here (every
        # fresh room message roots its own — see the anchoring above) at least
        # knows what the reply answers. DMs get the same treatment: a reply is
        # the one pointer a session start can carry there too.
        reply_to_text: str | None = None
        reply_to_is_own = False
        if message.in_reply_to:
            target = await self._reply_target(client, room_id, message.in_reply_to)
            if target is not None:
                reply_to_text = target.body or ""
                reply_to_is_own = self._is_own(target)

        # prepare a MessageEvent
        message_event = MessageEvent(
            text=message_body,
            source=source,
            message_id=message.id,
            timestamp=message.created_at or datetime.now(UTC),
            raw_message=message,
            reply_to_message_id=message.in_reply_to,
            reply_to_text=reply_to_text,
            reply_to_is_own_message=reply_to_is_own,
            channel_context="\n\n".join(context_blocks) or None,
        )
        if message_event.is_command():
            message_event.message_type = MessageType.COMMAND

        # Attachments — download and hand the local cache paths to the gateway,
        # which runs vision enrichment / document extraction off media_urls.
        (
            message_event.media_urls,
            message_event.media_types,
            media_kinds,
        ) = await self._cache_attachments(list(message.attachments))
        if media_kinds:
            # Same precedence as the Teams/Signal adapters: document-context
            # injection gates strictly on DOCUMENT, image handling keys off the
            # per-path image/* MIME regardless of message_type.
            message_event.message_type = self._message_type_for_media_kinds(media_kinds)
        else:
            message_event.message_type = MessageType.TEXT

        logger.debug("Chatto: MessageEvent: %s", message_event)
        return message_event

    async def _forward_reaction(
        self,
        event: RealtimeEvent,
        payload: Any,
        *,
        removed: bool,
    ) -> None:
        """Forward a human reaction to the gateway's reaction hook surface.

        The handler is registered by the gateway via ``set_reaction_handler``
        and fans out as ``reaction:added`` / ``reaction:removed`` through the
        HookRegistry.  The dict shape mirrors the Slack adapter's — hook
        consumers are written against that contract, not against a per-platform
        one.  Our own lifecycle reactions (👀/✅/❌) are dropped: forwarding them
        would feed the agent its own markers.
        """
        actor_id = event.actor_id
        if actor_id and self.me and actor_id == self.me.id:
            return
        if not payload.room_id or not payload.message_event_id or not actor_id:
            return

        handler = self._reaction_handler
        if handler is None:
            return

        action = "removed" if removed else "added"
        try:
            await handler(
                {
                    "platform": ChattoConstants.PLATFORM_NAME,
                    "event_name": f"reaction:{action}",
                    "reaction": payload.emoji,
                    "user_id": actor_id,
                    "item_user_id": None,
                    "item_type": "message",
                    "channel_id": payload.room_id,
                    "message_ts": payload.message_event_id,
                    "event_ts": event.id,
                    "raw_event": event,
                },
            )
        except Exception:  # pragma: no cover - the hook contract is non-blocking
            logger.debug("Chatto: reaction hook forwarding failed", exc_info=True)

    async def _handle_realtime_event(self, event: RealtimeEvent) -> None:
        if self._is_seen(event.id):
            return

        if event.actor_id is None:
            return

        logger.debug("EVENT happened: '%s' from %s", event.kind, event.actor_id)

        # Self-event filter — the actor_id on the envelope is authoritative.
        actor_id = event.actor_id
        if actor_id and self.me and actor_id == self.me.id:
            return

        if event.kind == "message_posted":
            await self._dispatch_message_posted(event.payload, event.id)

        elif event.kind == "message_edited":
            await self._dispatch_message_edited(event.payload)

        elif event.kind in ("reaction_added", "reaction_removed"):
            # Interactive prompts (clarify / exec-approval) are driven by
            # reactions. Route a reaction on a tracked prompt before
            # forwarding — the prompt is consumed either way, but the reaction
            # still reaches the gateway hook surface afterward.
            if event.kind == "reaction_added":
                try:
                    await self._try_handle_interactive_reaction(
                        getattr(event.payload, "room_id", None),
                        getattr(event.payload, "message_event_id", None),
                        getattr(event.payload, "emoji", None),
                        actor_id,
                    )
                except Exception:
                    logger.debug(
                        "Chatto: interactive reaction routing failed", exc_info=True
                    )
            await self._forward_reaction(
                event,
                event.payload,
                removed=event.kind == "reaction_removed",
            )

        elif event.kind == "presence_changed":
            # A participant's presence changed — patch the cached user in place.
            # The identified user is the event's own actor: the realtime
            # PresenceChangedEvent carries only a status field, and a
            # payload.user_id read here raised AttributeError on every
            # presence change, which the event loop mistook for a broken
            # stream and answered with a reconnect backoff — dropping
            # inbound messages that arrived inside the gap.
            changed_user_id = event.actor_id
            if (
                changed_user_id
                and (user := self._user_cache.get(changed_user_id)) is not None
            ):
                # The realtime payload carries the proto enum as a plain int
                # (protobuf runtime without native enum wrappers); chattolib's
                # own bot.py coerces the same field the same way. Read the
                # wire shape, never ``.name`` directly, which raised
                # AttributeError on every presence change and dropped inbound
                # messages inside the reconnect gaps that followed.
                raw_status = event.payload.status
                if isinstance(raw_status, PresenceStatus):
                    status = raw_status
                else:
                    proto_name = ProtoPresenceStatus.Name(int(raw_status))
                    status = _PRESENCE_BY_PROTO_NAME.get(
                        proto_name, PresenceStatus.UNSPECIFIED
                    )
                user.presence_status = status
                logger.debug(
                    "Chatto: patched presence of %s to %s",
                    changed_user_id,
                    status.name,
                )

        elif event.kind in ("user_joined_room", "user_left_room"):
            # Membership moved — the room's roster projection is now wrong, so
            # discard it; the next turn there refetches once and re-delivers.
            if event.payload.room_id in self._roster_announced:
                logger.debug(
                    "Chatto: %s invalidated the roster of room %s",
                    event.kind,
                    event.payload.room_id,
                )
                self._evict_roster(event.payload.room_id)

        elif event.kind == "room_updated":
            # Room metadata changed (name, description) — patch the cache in
            # place so the next turn sees the fresh values instead of stale
            # names from a long-lived _room_names dict.
            payload = event.payload
            room_id = getattr(payload, "room_id", None)
            if room_id and room_id in self._joined_room_ids:
                name = getattr(payload, "name", None)
                if name is not None:
                    self._room_names[room_id] = name
                logger.debug(
                    "Chatto: room %s updated: name=%s",
                    room_id,
                    name,
                )

        elif event.kind == "room_groups_updated":
            # Room-group membership moved — discard the cached projection so
            # the next turn that needs room groups refetches once.
            logger.debug("Chatto: room_groups_updated — discarding cached group state")

        elif event.kind == "message_retracted":
            # A message was retracted — no-op for now.  Interactive prompts
            # are keyed by their own message IDs (not the original event_id),
            # and the prompt timeout handles stale cards anyway.
            payload = event.payload
            logger.debug(
                "Chatto: message retracted: %s in %s",
                getattr(payload, "message_event_id", None),
                getattr(payload, "room_id", None),
            )

        elif event.kind == "thread_created":
            # A human created a thread — no adapter action needed.  The thread
            # will be seeded on first message because _seeded_threads starts
            # empty and seeds on first turn into an unknown root.
            payload = event.payload
            logger.debug(
                "Chatto: thread created: room=%s root=%s",
                getattr(payload, "room_id", None),
                getattr(payload, "thread_root_event_id", None),
            )

        elif event.kind == "user_typing":
            # Presence-like ephemeral signal: a user is typing in a room.
            # The adapter never surfaces this to the agent — it is only useful
            # for a client UI cursor, which is not what a bot does.
            logger.debug("Chatto: user_typing event received")

        # confirmed:
        # NOTE: "projection_event" (and "caught_up") belong to realtime protocol
        # v2 on unreleased Chatto main — chattolib speaks v1 and can never
        # deliver them here. Revisit when vendored chattolib gains v2 typing.
        elif event.kind in (
            "mention_notification",
            "notification_dismissed",
            "room_marked_as_read",
            "user_typing",
            "notification_created",
            "new_direct_message_notification",
            "message_retracted",
            "thread_created",
            "thread_follow_changed",
        ):
            logger.debug(
                "Chatto: '%s' event received. Not yet implemented or not needed.",
                event.kind,
            )
        else:
            logger.warning("Chatto: unknown event kind: '%s'", event.kind)

    async def _chattolib_event_loop(self) -> None:
        """Event loop using chattolib's stream_events.

        This replaces the manual WebSocket loop with chattolib's high-level
        stream_events() which provides pre-decoded RealtimeEvent objects.
        """
        delay = ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF
        while not self._closing:
            try:
                client = await self._require_chatto_client()

                await self._refresh_rooms()
                logger.info(
                    "Chatto: starting chattolib event stream with %d rooms",
                    len(self._joined_room_ids),
                )

                async for event in stream_events(
                    client, resume_cursor=self._realtime_cursor
                ):
                    if self._closing:
                        return
                    if isinstance(event, RealtimeEvent):
                        try:
                            await self._handle_realtime_event(event)
                        except (AttributeError, TypeError) as exc:
                            # A handler reading a payload field that does not
                            # exist or is not the type it assumed is a handler
                            # bug, not wire damage: it failed one event, so it
                            # must not cost the stream. Reconnecting here is
                            # what turned the presence crashes into 30s
                            # backoff gaps that silently dropped every
                            # message_posted arriving inside them (see
                            # AGENTS.md, "A handler bug is not a broken
                            # stream"). The cursor is NOT advanced: an event
                            # the adapter never applied is not safe to resume
                            # past.
                            logger.warning(
                                "Chatto: handler for '%s' failed (%s); skipping event %s",
                                event.kind,
                                exc,
                                event.id,
                            )
                        except Exception:
                            # Not a wire-shape bug — loud traceback, same skip
                            # semantics: one event failed, the stream stays up.
                            logger.exception(
                                "Chatto: handler for '%s' crashed on event %s; skipping it",
                                event.kind,
                                event.id,
                            )
                        else:
                            # chattolib marks a cursor as safe only once its event
                            # has been applied, and leaves it None when the event
                            # cannot be replayed. Keeping the last good one is what
                            # turns a reconnect into a resume rather than a gap.
                            if event.cursor:
                                self._realtime_cursor = event.cursor
                # Iterator exited cleanly — treat as a normal close and reconnect
                # with the local backoff (no server hint available).
                logger.info("Chatto: realtime stream ended, reconnecting")
            except asyncio.CancelledError:
                return
            except ChattoRealtimeCloseError as exc:
                if not exc.reconnect:
                    logger.error(
                        "Chatto: realtime closed by server (%s: %s), not reconnecting",
                        exc.code,
                        exc.message,
                    )
                    return
                wait = max(
                    exc.retry_after_ms / 1000.0,
                    ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF,
                )
                logger.warning(
                    "Chatto: realtime closed by server (%s), reconnecting in %.1fs",
                    exc.code,
                    wait,
                )
                delay = (
                    ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF
                )  # server hint supersedes local backoff
                await self._sleep_interruptible(wait)
                continue
            except ChattoRealtimeError as exc:
                if exc.fatal:
                    logger.error(
                        "Chatto: fatal realtime error (%s): %s", exc.code, exc.message
                    )
                    return
                logger.warning(
                    "Chatto: realtime error (%s: %s), reconnecting in %.1fs",
                    exc.code,
                    exc.message,
                    delay,
                )
            except Exception as exc:
                cause = exc.__cause__
                if cause is not None and isinstance(cause, TypeError):
                    logger.warning(
                        "Chatto: realtime wire corruption (%s), reconnecting in %.1fs",
                        cause,
                        delay,
                    )
                else:
                    logger.warning(
                        "Chatto: unexpected realtime error: %s, reconnecting in %.1fs",
                        exc,
                        delay,
                    )

            if self._closing:
                return

            jitter = delay * 0.2 * random.random()
            await self._sleep_interruptible(delay + jitter)
            delay = min(delay * 2, ChattoConstants.WS_RECONNECT_MAX_BACKOFF)

    async def _sleep_interruptible(self, seconds: float) -> None:
        """Sleep in short slices so disconnect() cancels promptly."""
        end = asyncio.get_running_loop().time() + seconds
        while not self._closing:
            remaining = end - asyncio.get_running_loop().time()
            if remaining <= 0:
                return
            await asyncio.sleep(min(remaining, 0.5))

    def _warn_if_home_channel_unjoined(self, member_ids: set[str]) -> None:
        """Warn once when CHATTO_HOME_CHANNEL names a room the bot is not in.

        Standalone cron delivery posts straight into that room with a fresh
        client and no join logic of its own — without server-side membership
        every proactive send fails there.
        """
        home_id = (self.chatto_config.home_channel.value or "").strip()
        if (
            not home_id
            or home_id in member_ids
            or home_id in self._joined_room_ids
            or self._home_warning_logged
        ):
            return
        self._home_warning_logged = True
        logger.warning(
            "Chatto: CHATTO_HOME_CHANNEL '%s' is not a joined room - cron and "
            "notification delivery will fail until the bot joins it (invite "
            "the account natively in Chatto, or DM it '/join').",
            home_id,
        )

    async def _resolve_dm_partner(self, client: ChattoClient, room_id: str) -> None:
        """Cache the chat partner's login for a DM room, best-effort.

        DM rooms carry no usable name of their own, so the joined-rooms
        summary names the other side instead. Resolved once per room and
        session; without our own user (pre-connect) or on a directory error
        the raw room name stays in place.
        """
        if room_id in self._dm_partners or self.me is None:
            return
        try:
            member_ids, _page = await client.list_room_members(room_id)
            if not member_ids:
                return
            members = await client.batch_get_room_members(room_id, member_ids)
        except Exception:
            logger.debug(
                "Chatto: could not list members of DM %s", room_id, exc_info=True
            )
            return
        for member in members:
            user = member.user
            if user and user.id != self.me.id:
                self._dm_partners[room_id] = user.login or user.display_name
                return

    def _joined_room_label(self, room_id: str) -> str:
        """One entry for the joined-rooms summary, with how-we-answer tags.

        The tag names the room's participation mode at a glance: DMs answer
        unconditionally, listed channels answer on mentions or every message,
        and an unlisted channel is read-only. A DM is labelled with its chat
        partner's login rather than its (empty) room name, and names are
        quoted so empty strings and spaces stay visible.
        """
        name = self._room_names.get(room_id, room_id)
        if self._room_kinds.get(room_id) == RoomKind.DM:
            tags = "[dm]"
            name = self._dm_partners.get(room_id) or name
        else:
            policy = self._room_policy(room_id)
            if policy == RoomPolicy.REQUIRE_MENTION:
                tags = "[on-mention]"
            elif policy == RoomPolicy.OPEN:
                tags = "[every-message]"
            else:
                tags = "[read-only]"
        if room_id in self._universal_room_ids:
            tags += " [universal]"
        return f'"{name}" ({room_id}) {tags}'

    def _log_joined_rooms(self) -> None:
        """Log the joined rooms and how the bot answers in each of them."""
        labels = [self._joined_room_label(rid) for rid in self._joined_room_ids]
        logger.info(
            "Chatto WS: currently joined in %d room(s): %s",
            len(labels),
            ", ".join(labels),
        )

    async def _refresh_rooms(self) -> None:
        """Refresh room list via ConnectRPC, join and seed any newly discovered rooms.

        Runs once per (re)connect, so this is also the bootstrap point for the
        roster projection: nothing on the wire reports presence at connect
        time (the subscribe snapshot has no presence field; presence arrives
        as its own events), so cached presence could only heal on the next
        random change. Discarding it makes each announced room's first turn
        refetch fresh data — lazily, so quiet rooms stay free of API calls.
        """
        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto WS: _refresh_rooms aborted - no client available")
            return

        for announced in list(self._roster_announced):
            self._evict_roster(announced)

        try:
            rooms_list = await client.list_rooms()
            member_ids: set[str] = set()
            new_room_ids: list[str] = []

            for room_with_state in rooms_list:
                if not room_with_state:
                    continue

                room_obj = room_with_state.room or None
                if not room_obj:
                    continue

                self._room_names[room_obj.id] = room_obj.name
                self._room_kinds[room_obj.id] = room_obj.kind
                if room_obj.kind == RoomKind.DM:
                    await self._resolve_dm_partner(client, room_obj.id)
                if room_obj.universal:
                    self._universal_room_ids.add(room_obj.id)

                if not room_with_state.viewer_state.is_member:
                    continue
                member_ids.add(room_obj.id)
                if room_obj.id not in self._joined_room_ids:
                    new_room_ids.append(room_obj.id)

            # Left via /leave, kicked, deleted: rooms we no longer belong
            # to drop out here — otherwise the next refresh would quietly
            # re-add what /leave just removed.
            stale_room_ids = [
                rid for rid in self._joined_room_ids if rid not in member_ids
            ]
            for rid in stale_room_ids:
                self._joined_room_ids.remove(rid)
                self._evict_roster(rid)
            if stale_room_ids:
                logger.info(
                    "Chatto WS: no longer a member of %d room(s): %s",
                    len(stale_room_ids),
                    stale_room_ids,
                )

            self._warn_if_home_channel_unjoined(member_ids)

            if new_room_ids:
                logger.info(
                    "Chatto WS: discovered %d new room(s): %s",
                    len(new_room_ids),
                    new_room_ids,
                )

                for rid in new_room_ids:
                    # Membership came straight from the directory scan
                    # (viewer_state.is_member); natively invited rooms need no
                    # JoinRoom call — same rule as _run_join.
                    if self._answers_in_room(rid):
                        await self._seed_room(rid)
                    else:
                        logger.info(
                            "Chatto WS: %s (%s) is on neither mention list -"
                            " joined read-only",
                            self._room_names.get(rid, rid),
                            rid,
                        )
                    self._joined_room_ids.append(rid)

            self._log_joined_rooms()

        except Exception:
            logger.warning("Chatto WS: _refresh_rooms failed", exc_info=True)

        # ------------------------------------------------------------------ #
        # Read state (best-effort, Chatto-unique)
        # ------------------------------------------------------------------ #
        # Deliberately outside the try block above: this sweep runs on every
        # refresh — including refreshes that discovered no new rooms. The
        # old early return starved it to "only when something changed",
        # leaving silent rooms unmarked for whole connection lifetimes.
        # Best-effort: mark all joined rooms as read (room_id may be undefined here)
        for _rid in list(self._joined_room_ids):
            try:
                await client.mark_room_as_read(room_id=_rid)
            except Exception:
                logger.debug(
                    "Chatto: mark_room_as_read failed for %s", _rid, exc_info=True
                )

    # ------------------------------------------------------------------ #
    # Sending (ConnectRPC — unchanged from polling version)
    # ------------------------------------------------------------------ #

    def _resolve_outbound_thread(
        self,
        chat_id: str,
        reply_to: str | None,
        metadata: dict[str, Any] | None,
    ) -> str | None:
        """The thread an outbound message belongs in, or ``None`` for the room.

        One definition for every outbound path (text and attachments):
        ``metadata["thread_id"]`` wins — it names the thread root to stay in;
        otherwise ``reply_to`` anchors a thread under the incoming message,
        but only with ``auto_thread`` enabled. DMs never thread.

        Deliberately does NOT open an auto-thread of its own — only the
        chunked text path in :meth:`send` does that, because only there do
        further chunks follow into the freshly created thread.
        """
        thread_id = (metadata or {}).get("thread_id")
        if reply_to and self.chatto_config.auto_thread.value and not thread_id:
            thread_id = reply_to
        if self._room_kinds.get(chat_id) == RoomKind.DM:
            return None
        return str(thread_id) if thread_id else None

    async def send(
        self,
        chat_id: str,
        content: str,
        reply_to: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> SendResult:
        """Send a message to a Chatto room.

        Long messages are split into chunks via ``truncate_message`` and
        each chunk is sent as a separate CreateMessage call.  The first
        chunk's message ID is returned as ``message_id``.

        When ``auto_thread`` is enabled and the incoming message was a
        regular room message (not already in a thread), the first chunk is
        sent as a room message and its ID becomes the thread root. Subsequent
        chunks are sent in that thread. This mirrors Discord's auto_thread
        behavior.

        BasePlatformAdapter override
        """
        if not content:
            return SendResult(success=False, error="Empty message")

        thread_id = self._resolve_outbound_thread(chat_id, reply_to, metadata)
        room_kind = self._room_kinds.get(chat_id)
        is_dm = room_kind == RoomKind.DM

        # Auto-thread: by default, Chatto creates a thread for replies to room
        # messages (not DMs, not already in a thread). This keeps conversations
        # organized in the room. Can be disabled via extra.auto_thread=false.
        use_auto_thread = (
            self.chatto_config.auto_thread.value and not thread_id and not is_dm
        )

        client = await self._get_chatto_client()
        if client is None:
            return SendResult(
                success=False, error="Chatto client not available", retryable=True
            )

        # The split, the routing and the stop-at-first-failure all live in
        # _deliver_chunks, shared with the plugin tools: a long answer must
        # reach the room the same way whoever sent it.
        delivery = await _deliver_chunks(
            client,
            chat_id,
            content,
            thread_root_event_id=str(thread_id) if thread_id else "",
            create_thread=use_auto_thread,
            on_posted=self._mark_seen,
        )
        message_ids = delivery.message_ids
        if use_auto_thread and message_ids and not thread_id:
            thread_id = delivery.root_event_id

        # Nothing got through at all — report the failure instead of a phantom success.
        if not message_ids:
            # Only server-side failures count as retryable; a bug in our own
            # code must not read as a transient network blip.
            return SendResult(
                success=False,
                error=(
                    str(delivery.error)
                    if delivery.error
                    else "Chatto: message could not be sent"
                ),
                retryable=isinstance(delivery.error, ChattoError),
            )

        first_id = message_ids[0]

        # ------------------------------------------------------------------ #
        # Thread following (best-effort, Chatto-unique)
        # ------------------------------------------------------------------ #
        if thread_id:
            try:
                await client.follow_thread(chat_id, thread_id)
            except Exception:
                logger.debug(
                    "Chatto: follow_thread failed for %s/%s",
                    chat_id,
                    thread_id,
                    exc_info=True,
                )

        # A later chunk failed after earlier ones went out: partial delivery.
        if delivery.error is not None:
            logger.warning(
                "Chatto: sent %d/%d chunk(s) to %s before failing: %s",
                len(message_ids),
                delivery.chunks_total,
                chat_id,
                delivery.error,
            )

        # raw_response stays unset (dict-shaped per the SendResult contract):
        # gateway consumers such as the cron scheduler call .get() on it, so a
        # chattolib Message here would crash delivery bookkeeping *after* the
        # send already succeeded — the job then falls back to the standalone
        # path and the room sees the message twice.
        return SendResult(success=True, message_id=first_id)

    def format_message(self, content: str) -> str:
        """Normalise outgoing text for Chatto.

        The transformations live in :func:`_normalise_outbound_text`, shared
        with the standalone cron sender.

        BasePlatformAdapter override
        """
        return _normalise_outbound_text(content)

    async def send_clarify(
        self,
        chat_id: str,
        question: str,
        choices: list[str] | None,
        clarify_id: str,
        session_key: str,
        metadata: dict[str, Any] | None = None,
    ) -> SendResult:
        """Post a reaction-driven clarify prompt.

        When ``choices`` is empty/None or exceeds 9 entries (the number emojis
        only cover 1-9), delegate to the base-class text fallback. Otherwise
        post a numbered message, seed one reaction per option plus ``✏️``
        (Other), register the prompt in both registries, and return.

        BasePlatformAdapter override
        """
        if not choices or len(choices) > 9:
            return await super().send_clarify(
                chat_id, question, choices, clarify_id, session_key, metadata=metadata
            )

        client = await self._get_chatto_client()
        if client is None:
            return SendResult(
                success=False, error="Chatto client not available", retryable=True
            )

        number_emojis = ["1️⃣", "2️⃣", "3️⃣", "4️⃣", "5️⃣", "6️⃣", "7️⃣", "8️⃣", "9️⃣"]
        emojis = [number_emojis[i] for i in range(len(choices))]
        other_emoji = "✏️"

        lines = [f"❓ {question}", ""]
        for i, choice in enumerate(choices):
            lines.append(f"{emojis[i]} {choice}")
        lines.append(f"{other_emoji} Other (type your answer)")
        content = "\n".join(lines)

        try:
            msg = await client.post_message(room_id=str(chat_id), body=content)
        except ChattoError as e:
            logger.warning("Chatto: send_clarify post failed: %s", e)
            return SendResult(success=False, error=str(e), retryable=True)
        except Exception as e:
            logger.warning("Chatto: send_clarify post error: %s", e)
            return SendResult(success=False, error=str(e), retryable=False)

        message_id = str(msg.id)
        if message_id:
            self._mark_seen(message_id)

        for emoji in [*emojis, other_emoji]:
            try:
                await client.add_reaction(
                    room_id=str(chat_id),
                    message_event_id=message_id,
                    emoji=emoji,
                )
            except Exception:
                logger.debug("Chatto: seed reaction %s failed", emoji, exc_info=True)

        prompt = _ClarifyPrompt(
            clarify_id=clarify_id,
            choices=list(choices),
            session_key=session_key,
            chat_id=str(chat_id),
            message_id=message_id,
            expires_at=time.monotonic() + INTERACTIVE_PROMPT_TIMEOUT_SECONDS,
        )
        self._clarify_prompts[message_id] = prompt
        self._clarify_id_to_message[clarify_id] = message_id

        return SendResult(success=True, message_id=message_id)

    async def retire_clarify_card(self, clarify_id: str, notice: str) -> None:
        """Rewrite a still-live clarify card into a terminal, button-less state.

        The gateway calls this whenever it ends a clarify without a click — the
        wait timed out, the session was reset, or unmatched free prose superseded
        the prompt — so the card stops advertising an answer path the released
        clarify can no longer accept.  Keyed by ``clarify_id``, so a late call
        cannot touch a newer prompt; no-op once resolved.

        BasePlatformAdapter override
        """
        message_id = self._clarify_id_to_message.pop(clarify_id, None)
        if message_id is None:
            return
        prompt = self._clarify_prompts.get(message_id)
        if prompt is None:
            return
        prompt.resolved = True
        terminal = f"❌ {notice}"
        try:
            await self.edit_message(prompt.chat_id, message_id, terminal)
        except Exception:
            logger.debug("Chatto: retire_clarify_card edit failed", exc_info=True)
        self._clarify_prompts.pop(message_id, None)

    async def _send_exec_approval_prompt(
        self, prompt: ExecApprovalPrompt
    ) -> SendResult:
        """Post a reaction-driven exec-approval prompt.

        BasePlatformAdapter override
        """
        emoji_map = {
            "once": "✅",
            "session": "🌀",
            "always": "♾️",
            "deny": "❌",
        }
        number_emojis = ["1️⃣", "2️⃣", "3️⃣", "4️⃣", "5️⃣", "6️⃣", "7️⃣", "8️⃣", "9️⃣"]

        choices = list(prompt.choices)
        choices_map: dict[str, str] = {}
        emojis: list[str] = []
        for i, choice in enumerate(choices):
            emoji = emoji_map.get(choice)
            if emoji is None:
                emoji = number_emojis[i] if i < len(number_emojis) else "🔢"
            emojis.append(emoji)
            choices_map[emoji] = choice

        client = await self._get_chatto_client()
        if client is None:
            return SendResult(
                success=False, error="Chatto client not available", retryable=True
            )

        smart_denied = bool(prompt.smart_denied)
        tier = "once"
        if not smart_denied:
            if "always" in choices:
                tier = "always"
            elif "session" in choices:
                tier = "session"
        typed_hints = {
            "once": "React with ✅ to approve once.",
            "session": "React with 🌀 to approve for this session.",
            "always": "React with ♾️ to approve always.",
        }
        typed_hint = typed_hints.get(tier, "")

        lines = [str(prompt.text), ""]
        if typed_hint:
            lines.append(typed_hint)
            lines.append("")
        lines.append("Legend:")
        for emoji, choice in choices_map.items():
            lines.append(f"{emoji} {choice}")
        content = "\n".join(lines)

        try:
            msg = await client.post_message(room_id=str(prompt.chat_id), body=content)
        except ChattoError as e:
            logger.warning("Chatto: send_exec_approval post failed: %s", e)
            return SendResult(success=False, error=str(e), retryable=True)
        except Exception as e:
            logger.warning("Chatto: send_exec_approval post error: %s", e)
            return SendResult(success=False, error=str(e), retryable=False)

        message_id = str(msg.id)
        if message_id:
            self._mark_seen(message_id)

        for emoji in emojis:
            try:
                await client.add_reaction(
                    room_id=str(prompt.chat_id),
                    message_event_id=message_id,
                    emoji=emoji,
                )
            except Exception:
                logger.debug("Chatto: seed reaction %s failed", emoji, exc_info=True)

        approval_prompt = _ApprovalPrompt(
            session_key=str(prompt.session_key),
            chat_id=str(prompt.chat_id),
            message_id=message_id,
            choices_map=choices_map,
            expires_at=time.monotonic() + INTERACTIVE_PROMPT_TIMEOUT_SECONDS,
        )
        self._approval_prompts[message_id] = approval_prompt

        return SendResult(success=True, message_id=message_id)

    async def edit_message(
        self,
        chat_id: str,
        message_id: str,
        content: str,
        *,
        finalize: bool = False,
    ) -> SendResult:
        """Edit a message we previously sent, via MessageService/UpdateMessage.

        The stream consumer drives streaming replies through this: without the
        override the base class reports "Not supported" and every incremental
        update arrives as a *new* message.

        ``finalize`` is a no-op for Chatto — an edit is an edit here, there is
        no in-progress card state to close out (hence no
        ``REQUIRES_EDIT_FINALIZE``).

        Content that exceeds the per-message limit is refused rather than
        silently truncated, so the caller falls back to ``send()``, which
        splits across messages.

        BasePlatformAdapter override
        """
        if not message_id:
            return SendResult(success=False, error="Chatto: no message id to edit")
        if not content:
            return SendResult(success=False, error="Empty message")

        formatted = self.format_message(content)
        if len(formatted) > ChattoConstants.MAX_MESSAGE_LENGTH:
            # Refuse instead of truncating: the caller's fallback path splits.
            return SendResult(
                success=False,
                error=(
                    f"Chatto: edit exceeds {ChattoConstants.MAX_MESSAGE_LENGTH} "
                    f"chars ({len(formatted)})"
                ),
            )

        client = await self._get_chatto_client()
        if client is None:
            return SendResult(
                success=False, error="Chatto client not available", retryable=True
            )

        try:
            msg = await client.update_message(
                room_id=str(chat_id),
                event_id=str(message_id),
                body=formatted,
            )
        except ChattoError as e:
            logger.warning("Chatto: UpdateMessage failed for %s: %s", message_id, e)
            return SendResult(success=False, error=str(e), retryable=True)
        except Exception as e:
            logger.warning("Chatto: UpdateMessage error for %s: %s", message_id, e)
            return SendResult(success=False, error=str(e), retryable=False)

        # Our own edit comes back as a message_edited event; mark it seen so it
        # is never mistaken for inbound traffic.
        edited_id = msg.id or str(message_id)
        self._mark_seen(edited_id)
        return SendResult(success=True, message_id=edited_id)

    async def delete_message(self, chat_id: str, message_id: str) -> bool:
        """Delete a message via MessageService/DeleteMessage.

        Used by the stream consumer's fresh-final cleanup (removing a preview
        message once the completed reply has been sent) and by the ephemeral
        reply TTL.

        BasePlatformAdapter override
        """
        if not chat_id or not message_id:
            return False
        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto: DeleteMessage — client unavailable")
            return False
        try:
            return bool(
                await client.delete_message(
                    room_id=str(chat_id),
                    event_id=str(message_id),
                )
            )
        except ChattoError as e:
            logger.warning("Chatto: DeleteMessage failed for %s: %s", message_id, e)
            return False
        except Exception as e:
            logger.warning("Chatto: DeleteMessage error for %s: %s", message_id, e)
            return False

    async def create_handoff_thread(
        self,
        parent_chat_id: str,
        name: str,
    ) -> str | None:
        """Anchor a session handoff in a fresh thread under *parent_chat_id*.

        Chatto threads hang off a message, not off the room, so we post a seed
        message and hand its ID back as the thread root — the same shape the
        Slack adapter uses.  DMs don't support threads, so they get ``None``
        and the watcher keeps delivering into the DM itself.

        BasePlatformAdapter override
        """
        if not parent_chat_id:
            return None
        if self._room_kinds.get(parent_chat_id) == RoomKind.DM:
            logger.debug("Chatto: handoff thread skipped — %s is a DM", parent_chat_id)
            return None

        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto: handoff thread — client unavailable")
            return None

        try:
            msg = await client.post_message(
                room_id=str(parent_chat_id),
                body=f"🧵 Hermes handoff — **{(name or 'session').strip()[:80]}**",
            )
        except Exception as e:
            logger.warning(
                "Chatto: handoff thread seed-post failed for room %s: %s",
                parent_chat_id,
                e,
            )
            return None

        seed_id = msg.id
        if not seed_id:
            logger.warning("Chatto: handoff thread seed-post returned no message id")
            return None

        self._mark_seen(seed_id)
        try:
            await client.follow_thread(str(parent_chat_id), seed_id)
        except Exception:
            logger.debug(
                "Chatto: follow_thread failed for handoff %s/%s",
                parent_chat_id,
                seed_id,
                exc_info=True,
            )
        return seed_id

    # Overridden from BaseAdapter:
    async def send_typing(self, chat_id: str, metadata=None) -> None:
        """Start a persistent typing indicator for a room.

        Sends a typing ping every 10 seconds (Chatto's indicator likely
        lasts ~8-10s).  The background loop runs until ``stop_typing()``
        is called or the task is cancelled.

        BasePlatformAdapter override
        """
        if chat_id in self._typing_tasks:
            return  # already running

        async def _typing_loop() -> None:
            try:
                while True:
                    try:
                        client = await self._get_chatto_client()
                        if client is None:
                            return
                        await client.refresh_typing_indicator(room_id=str(chat_id))
                    except asyncio.CancelledError:
                        return
                    except Exception:
                        logger.debug(
                            "Chatto: typing indicator refresh failed for %s",
                            chat_id,
                            exc_info=True,
                        )
                    await asyncio.sleep(10)
            except asyncio.CancelledError:
                pass
            finally:
                self._typing_tasks.pop(chat_id, None)

        self._typing_tasks[chat_id] = asyncio.create_task(_typing_loop())

    async def stop_typing(self, chat_id: str) -> None:
        """Stop the persistent typing indicator for a room.

        BasePlatformAdapter override
        """
        task = self._typing_tasks.pop(chat_id, None)
        if task:
            task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                logger.debug("Chatto: typing task ended for %s", chat_id, exc_info=True)

    async def get_chat_info(self, chat_id: str) -> dict[str, Any]:
        """Get information about a chat/room.

        BasePlatformAdapter override
        """
        name = self._room_names.get(chat_id, chat_id)
        kind = self._room_kinds.get(chat_id)
        return {
            "name": name,
            "type": chat_type_for_room_kind(kind).value,
        }

    # ------------------------------------------------------------------ #
    # Reactions
    # ------------------------------------------------------------------ #

    @staticmethod
    def _emoji_to_shortcode(emoji: str) -> str:
        """Convert a unicode emoji to a Chatto shortcode name.

        If the emoji is already a shortcode (no unicode mapping found),
        return it as-is.
        """
        shortcode = ChattoConstants.EMOJI_TO_SHORTCODE.get(emoji)
        if shortcode:
            return shortcode
        # Already a shortcode like "thumbsup" — return as-is
        return emoji

    async def add_reaction(self, room_id: str, message_id: str, emoji: str) -> bool:
        """Add a reaction to a message via MessageService/AddReaction.

        BasePlatformAdapter override
        """
        shortcode = self._emoji_to_shortcode(emoji)
        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto: AddReaction — client unavailable")
            return False
        try:
            result = await client.add_reaction(
                room_id=room_id,
                message_event_id=message_id,
                emoji=shortcode,
            )
            return result
        except ChattoError as e:
            logger.warning("Chatto: AddReaction failed: %s", e)
            return False
        except Exception as e:
            logger.warning("Chatto: AddReaction error: %s", e)
            return False

    async def remove_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool:
        """Remove a reaction from a message via MessageService/RemoveReaction.

        BasePlatformAdapter override
        """
        shortcode = self._emoji_to_shortcode(emoji)
        client = await self._get_chatto_client()
        if client is None:
            logger.warning("Chatto: RemoveReaction — client unavailable")
            return False
        try:
            result = await client.remove_reaction(
                room_id=str(chat_id),
                message_event_id=str(message_id),
                emoji=shortcode,
            )
            return result
        except ChattoError as e:
            logger.warning("Chatto: RemoveReaction failed: %s", e)
            return False
        except Exception as e:
            logger.warning("Chatto: RemoveReaction error: %s", e)
            return False

    # NOTE: there is deliberately no start_dm(). On Chatto only a person can
    # start a DM ("Only people can start a DM... A bot cannot start a DM, even
    # with message.post", dev-docs/guides/operations/permissions), and nothing
    # in Hermes calls such a method — BasePlatformAdapter does not even declare
    # one. An earlier version had it and advertised "DM initiation" in the
    # capability banner, but it could only ever fail: the vendored chattolib's
    # start_dm() calls a service stub that does not exist, so the AttributeError
    # was swallowed into a silent None. DMs become usable the moment the other
    # person writes to the bot; /join and the chatto_send_dm tool both say so.

    # ------------------------------------------------------------------ #
    # Room creation (Chatto-unique)
    # ------------------------------------------------------------------ #

    async def create_room(
        self,
        name: str,
        description: str = "",
        group_id: str = "",
        universal: bool = True,
    ) -> str | None:
        """Create an ad-hoc room via RoomService/CreateRoom.

        Returns the room ID on success, or None on failure.

        BasePlatformAdapter override
        """
        client = await self._get_chatto_client()
        if client is None:
            return None
        try:
            room = await client.create_room(
                name=name,
                group_id=group_id or "",
                description=description,
                universal=universal,
            )
            rid = str(room.id) if room else ""
            if rid:
                self._room_names[rid] = room.name
                self._room_kinds[rid] = room.kind
                return rid
            logger.debug("Chatto: CreateRoom returned no room id")
            return None
        except ChattoError as e:
            logger.debug("Chatto: CreateRoom failed: %s", e)
            return None
        except Exception as e:
            logger.debug("Chatto: CreateRoom error: %s", e)
            return None

    # ------------------------------------------------------------------ #
    # Processing lifecycle hooks (reactions-based, like Discord)
    # ------------------------------------------------------------------ #

    def _event_room_and_message_id(self, event: MessageEvent) -> tuple[str, str]:
        """Extract room_id and message_id from a MessageEvent."""
        message_id = event.message_id or ""
        return event.source.chat_id, message_id

    async def on_processing_start(self, event: MessageEvent) -> None:
        """Record the turn as open, then add an 👀 (eyes) reaction.

        The record is what lets an edit of this very message be recognized as
        a mid-run correction. It must be maintained even when reactions are
        disabled — tracking and decorating are independent concerns.

        BasePlatformAdapter override
        """
        session_key = (
            self._session_key_for(event.source) if event.source is not None else ""
        )
        if session_key and event.message_id:
            self._processing[session_key] = str(event.message_id)

        if not self.chatto_config.reactions.value:
            return

        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            # Routine, not a fault: the gateway runs agent-initiated turns
            # (heartbeat polls, goal continuations) through the same pipeline
            # with message_id=None, and there is no inbound message to mark.
            logger.debug(
                "Chatto: nothing to react to (chat_id=%r, message_id=%r)",
                chat_id,
                message_id,
            )
            return
        await self.add_reaction(chat_id, message_id, "👀")

    async def on_processing_complete(
        self,
        event: MessageEvent,
        outcome: ProcessingOutcome,
    ) -> None:
        """Close the turn's record, then swap 👀 for ✅/❌/🚫.

        Fires for every outcome, including CANCELLED (mid-run edit
        correction) — this is what re-arms `_processing` before the corrected
        turn is dispatched.

        BasePlatformAdapter override
        """
        session_key = (
            self._session_key_for(event.source) if event.source is not None else ""
        )
        if session_key and event.message_id:
            self._processing.pop(session_key, None)

        if not self.chatto_config.reactions.value:
            return
        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            return
        # Remove the processing eyes reaction
        await self.remove_reaction(chat_id, message_id, "👀")
        # Add the outcome reaction
        if outcome == ProcessingOutcome.SUCCESS:
            await self.add_reaction(chat_id, message_id, "✅")
        elif outcome == ProcessingOutcome.FAILURE:
            await self.add_reaction(chat_id, message_id, "❌")
        elif outcome == ProcessingOutcome.CANCELLED:
            await self.add_reaction(chat_id, message_id, "🚫")

    # ------------------------------------------------------------------ #
    # Asset upload (chunked)
    # ------------------------------------------------------------------ #

    async def _upload_asset(self, room_id: str, file_path: str) -> str | None:
        """Upload a file via the chunked AssetUploadService.

        Returns the asset ID on success, or None on failure.
        """
        try:
            file_data = await asyncio.to_thread(_read_file_bytes, file_path)
        except Exception as e:
            logger.error("Chatto: failed to read file %s — %s", file_path, e)
            return None

        if not file_data:
            logger.error("Chatto: file %s is empty", file_path)
            return None

        file_size = len(file_data)
        file_name = os.path.basename(file_path)
        mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
        sha256_hash = hashlib.sha256(file_data).hexdigest()

        client = await self._get_chatto_client()
        if client is None:
            logger.error("Chatto: upload aborted - no client available")
            return None
        try:
            # Step 1: Create upload session
            upload = await client.create_upload(
                room_id=room_id,
                filename=file_name,
                size=file_size,
                sha256=sha256_hash,
                content_type=mime_type,
            )
            # AssetUpload names this upload_id, not id — reading it through an
            # untyped getattr default is what let the mismatch reach production.
            upload_id = upload.upload_id
            if not upload_id:
                logger.error("Chatto: CreateUpload returned no upload ID")
                return None

            # Step 2: Upload chunks
            offset = 0
            while offset < file_size:
                chunk = file_data[offset : offset + ChattoConstants.UPLOAD_CHUNK_SIZE]
                chunk_sha256 = hashlib.sha256(chunk).hexdigest()
                await client.upload_chunk(
                    upload_id=upload_id,
                    offset=offset,
                    content=chunk,
                    chunk_sha256=chunk_sha256,
                )
                offset += len(chunk)

            # Step 3: Complete upload
            upload, asset = await client.complete_upload(upload_id=upload_id)
            if not asset:
                logger.error("Chatto: CompleteUpload returned no asset")
                return None

            logger.info(
                "Chatto: uploaded %s as asset %s (%d bytes)",
                file_name,
                asset.id,
                file_size,
            )
            return asset.id

        except ChattoError as e:
            logger.error("Chatto: upload failed: %s", e)
            return None
        except Exception as e:
            logger.error("Chatto: upload error: %s", e)
            return None

    async def _post_attachment_message(
        self,
        chat_id: str,
        asset_ids: list[str],
        caption: str | None,
        reply_to: str | None,
        metadata: dict[str, Any] | None,
    ) -> SendResult:
        """Post one message carrying already-uploaded assets.

        Threading follows the same rules as a text send
        (:meth:`_resolve_outbound_thread`), so with ``auto_thread`` disabled an
        attachment reply lands in the room like its text counterpart instead of
        quietly opening a thread.
        """
        thread_id = self._resolve_outbound_thread(chat_id, reply_to, metadata)

        client = await self._get_chatto_client()
        if client is None:
            return SendResult(
                success=False, error="Chatto client not available", retryable=True
            )
        try:
            msg = await client.post_message(
                room_id=str(chat_id),
                body=self.format_message(caption) if caption else "",
                attachment_asset_ids=asset_ids,
                thread_root_event_id=thread_id or "",
            )
            self._mark_seen(msg.id)
            return SendResult(success=True, message_id=msg.id)
        except ChattoError as e:
            # Same classification as the chunked text path: server-side
            # failures retry, anything else is ours and must not loop.
            return SendResult(success=False, error=str(e), retryable=True)
        except Exception as e:
            return SendResult(success=False, error=str(e), retryable=False)

    async def _send_local_file_as_attachment(
        self,
        chat_id: str,
        file_path: str,
        caption: str | None,
        reply_to: str | None,
        metadata: dict[str, Any] | None,
        *,
        kind: str,
    ) -> SendResult:
        """Upload a local file and post it as a native Chatto attachment.

        Shared by ``send_image_file``/``send_document``/``send_video``/
        ``send_voice`` — the upload mechanics are identical, only the wording of
        the failure notice differs.  On failure we send that notice as text and
        never the host path (it leaks the Hermes home layout).
        """
        notice = f"⚠️ Couldn't deliver the {kind} attachment."

        safe_path = self.validate_media_delivery_path(file_path)
        if not safe_path:
            logger.warning(
                "[%s] send %s: unsafe path %s",
                self.name,
                kind,
                file_path,
            )
            text = f"{caption}\n{notice}" if caption else notice
            return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)

        asset_id = await self._upload_asset(str(chat_id), safe_path)
        if not asset_id:
            logger.warning(
                "[%s] send %s: upload failed for %s",
                self.name,
                kind,
                safe_path,
            )
            text = f"{caption}\n{notice}" if caption else notice
            return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata)

        return await self._post_attachment_message(
            chat_id,
            [asset_id],
            caption,
            reply_to,
            metadata,
        )

    async def send_image_file(
        self,
        chat_id: str,
        image_path: str,
        caption: str | None = None,
        reply_to: str | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> SendResult:
        """Send a local image file via the chunked upload API.

        The parameter is ``image_path``, not ``file_path``: every caller passes
        it by keyword (``gateway/run.py:22354``, ``:22470``, and the base class's
        own ``send_multiple_images`` file:// branch), so a renamed parameter
        makes each of those raise TypeError and silently degrade to a text
        notice.

        BasePlatformAdapter override
        """
        return await self._send_local_file_as_attachment(
            chat_id,
            image_path,
            caption,
            reply_to,
            metadata,
            kind="image",
        )

    async def send_document(
        self,
        chat_id: str,
        file_path: str,
        caption: str | None = None,
        file_name: str | None = None,
        reply_to: str | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> SendResult:
        """Send a local file as a native Chatto attachment.

        ``file_name`` exists in the base-class signature and is accepted for
        compatibility, but Chatto takes the recipient-visible filename from
        the upload session (derived from the local path); failures are logged
        and noticed by ``_send_local_file_as_attachment`` itself.

        BasePlatformAdapter override
        """
        return await self._send_local_file_as_attachment(
            chat_id,
            file_path,
            caption,
            reply_to,
            metadata,
            kind="file",
        )

    async def send_video(
        self,
        chat_id: str,
        video_path: str,
        caption: str | None = None,
        reply_to: str | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> SendResult:
        """Send a local video as a native Chatto attachment.

        Chatto transcodes and plays it inline.

        BasePlatformAdapter override
        """
        return await self._send_local_file_as_attachment(
            chat_id,
            video_path,
            caption,
            reply_to,
            metadata,
            kind="video",
        )

    async def send_voice(
        self,
        chat_id: str,
        audio_path: str,
        caption: str | None = None,
        reply_to: str | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> SendResult:
        """Send a local audio file as a native Chatto attachment.

        Chatto has no dedicated voice-bubble type, so this is an ordinary audio
        attachment — still far better than the base class's text notice.

        BasePlatformAdapter override
        """
        return await self._send_local_file_as_attachment(
            chat_id,
            audio_path,
            caption,
            reply_to,
            metadata,
            kind="audio",
        )

    async def send_image(
        self,
        chat_id: str,
        image_url: str,
        caption: str | None = None,
        reply_to: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> SendResult:
        """Send an image to a Chatto room.

        Materialises the URL (size-capped download, like every inbound
        attachment) and uploads it as a native attachment. Falls back to the
        plain URL as a link — Chatto renders link previews — when the URL
        cannot be materialised or the post after a successful upload fails.
        An upload failure needs no fallback on top: the text notice of
        ``_send_local_file_as_attachment`` has already gone out.

        BasePlatformAdapter override
        """
        link_text = f"{caption}\n{image_url}" if caption else image_url
        path, is_temp = await self._materialise_image(image_url)
        if path is not None:
            try:
                result = await self._send_local_file_as_attachment(
                    chat_id, path, caption, reply_to, metadata, kind="image"
                )
            finally:
                if is_temp:
                    try:
                        os.unlink(path)
                    except OSError:
                        pass
            if result.success:
                return result

        return await self.send(chat_id, link_text, reply_to=reply_to, metadata=metadata)

    async def _materialise_image(self, image_url: str) -> tuple[str | None, bool]:
        """Resolve one ``send_multiple_images`` entry to a local file path.

        Accepts ``http(s)://`` URLs (downloaded to a temp file), ``file://``
        URIs and bare paths.  Returns ``(path, is_temp)`` — the caller unlinks
        when ``is_temp``.          ``(None, False)`` means the entry is unusable.
        """
        if image_url.startswith(("http://", "https://")):
            ext = os.path.splitext(urlsplit(image_url).path)[1] or ".png"
            tmp_fd, tmp_path = tempfile.mkstemp(suffix=ext, prefix="chatto_img_")
            os.close(tmp_fd)
            try:
                data = await self._download_attachment_bytes(image_url)
                await asyncio.to_thread(_write_file_bytes, tmp_path, data)
            except Exception as e:
                logger.warning("Chatto: image download failed for %s: %s", image_url, e)
                try:
                    os.unlink(tmp_path)
                except OSError:
                    pass
                return None, False
            return tmp_path, True

        local = image_url
        if local.startswith("file://"):
            local = unquote(urlsplit(local).path)
        return self.validate_media_delivery_path(local), False

    async def send_multiple_images(
        self,
        chat_id: str,
        images: list[tuple[str, str]],
        metadata: dict[str, Any] | None = None,
        human_delay: float = 0.0,
    ) -> SendResult:
        """Send a batch of images as ONE message with several attachments.

        The base implementation posts each image separately; a Chatto message
        carries a list of attachment assets, so a batch belongs in a single
        message (and a single notification).

        ``human_delay`` is ignored deliberately — there is only one outbound
        call to pace.  Entries that can't be fetched are dropped with a warning;
        if nothing survives, we fall back to the base class so the user still
        gets the links.

        BasePlatformAdapter override
        """
        if len(images or []) < 2:
            return await super().send_multiple_images(
                chat_id,
                images,
                metadata=metadata,
                human_delay=human_delay,
            )

        asset_ids: list[str] = []
        captions: list[str] = []
        for image_url, alt_text in images:
            path, is_temp = await self._materialise_image(image_url)
            if not path:
                logger.warning("Chatto: skipping unusable image %s", image_url)
                continue
            try:
                asset_id = await self._upload_asset(str(chat_id), path)
            finally:
                if is_temp:
                    try:
                        os.unlink(path)
                    except OSError:
                        pass
            if not asset_id:
                logger.warning("Chatto: upload failed for image %s", image_url)
                continue
            asset_ids.append(asset_id)
            if alt_text:
                captions.append(alt_text)

        if not asset_ids:
            logger.warning(
                "Chatto: no image survived upload, falling back to per-image delivery",
            )
            return await super().send_multiple_images(
                chat_id,
                images,
                metadata=metadata,
                human_delay=human_delay,
            )

        if len(asset_ids) < len(images):
            logger.warning(
                "Chatto: sending %d of %d images — the rest could not be uploaded",
                len(asset_ids),
                len(images),
            )

        return await self._post_attachment_message(
            chat_id,
            asset_ids,
            "\n".join(captions) or None,
            None,
            metadata,
        )


# ---------------------------------------------------------------------------
# Cron / out-of-process delivery
# ---------------------------------------------------------------------------


async def hermes_standalone_sender_fn(
    pconfig: PlatformConfig,
    chat_id: str,
    message: str,
    *,
    thread_id=None,
    media_files=None,
    force_document=False,
) -> SendResult:
    """Deliver a message to Chatto without a running gateway adapter. Do not modify signature.

    Used by cron / scheduled routines that run out-of-process. Creates a
    short-lived chattolib client, posts, and closes. Long messages are
    normalised and split like ``send()`` does, so cron output cannot die on
    the server's per-message limit.
    """
    chatto_config: ChattoConfiguration = ChattoConfiguration(pconfig=pconfig)

    # Create a temporary client for standalone sending — needs a base URL and token.
    has_token = bool(chatto_config.token.value)
    if not chatto_config.base_url.value or not has_token:
        return SendResult(success=False, error="Chatto: base URL or token missing")

    client = ChattoClient(
        base_url=chatto_config.base_url.value, token=chatto_config.token.value
    )

    try:
        kwargs: dict[str, Any] = {}
        if chatto_config.auto_thread.value and thread_id:
            kwargs["thread_root_event_id"] = thread_id
        if media_files and media_files.get("attachment_asset_ids"):
            kwargs["attachment_asset_ids"] = list(media_files["attachment_asset_ids"])

        formatted = _normalise_outbound_text(message)
        chunks = BasePlatformAdapter.truncate_message(
            formatted, ChattoConstants.SPLIT_THRESHOLD
        )
        message_ids: list[str] = []
        try:
            for chunk in chunks:
                posted = await client.post_message(chat_id, chunk, **kwargs)
                message_ids.append(posted.id)
        except Exception as exc:
            if message_ids:
                logger.warning(
                    "Chatto standalone: sent %d/%d chunk(s) to %s before failing: %s",
                    len(message_ids),
                    len(chunks),
                    chat_id,
                    exc,
                )
            # Same classification as the adapter's send paths: only
            # server-side failures read as retryable.
            return SendResult(
                success=False,
                error=str(exc),
                retryable=isinstance(exc, ChattoError),
            )
        return SendResult(success=True, message_id=message_ids[0])
    finally:
        try:
            await client.close()
        except Exception as exc:
            logger.warning(
                "Chatto standalone: error closing short-lived client (perhaps already closed): %s",
                exc,
            )


# Resolved Chatto configuration — populated by ``hermes_validate_config`` so
# plugin tools can access it without re-reading env vars or duplicating
# ``ChattoConfiguration`` logic.
#
# Keyed by the profile the snapshot was validated for, not stored in one slot:
# Hermes multiplexes several profiles through a single gateway process and runs
# every profile's own validate_config (run_adapters.py, one _create_adapter per
# profile), so a single global would be whichever profile validated last — and a
# routed cron fire or turn for another profile would then send with the wrong
# profile's bot token. The key is the same home key the platform registry scopes
# its own entries by, so both agree on what "the serving profile" means.
_resolved_chatto_configs: dict[str, ChattoConfiguration] = {}


def _profile_scope_key() -> str:
    """Registry key of the profile being served right now.

    Falls back to a constant when ``hermes_constants`` is unavailable (a bare
    standalone sender process): one profile there, so one key loses nothing.
    """
    try:
        from hermes_constants import hermes_home_key

        return hermes_home_key()
    except Exception:
        logger.debug("Chatto: no profile scope key available", exc_info=True)
        return ""


def _realtime_cursor_file() -> Path | None:
    """Where a restart-resumable realtime cursor lives, or None without Hermes.

    Under the profile's own home so two profiles never read each other's
    position; ``None`` (a bare process with no ``hermes_constants``) simply
    means nothing is persisted, which is the same behaviour as no config.
    """
    try:
        from hermes_constants import get_hermes_home
    except ImportError:
        return None
    return Path(get_hermes_home()) / "state" / "chatto_realtime_cursor.json"


def _restore_realtime_cursor(cfg: ChattoConfiguration) -> str | None:
    """The stored realtime cursor, if it is this profile's and still fresh.

    Three ways a stored value is refused instead of trusted: a different
    profile wrote it, it carries a shape this build does not understand, or it
    is older than ``resume_window``. In every case the stream starts at the
    live boundary, which is what the plugin did before the cursor was stored at
    all — so a bad file costs the gap it was meant to close, never a crash.
    """
    window = cfg.resume_window.value
    if window <= 0:
        return None
    path = _realtime_cursor_file()
    if path is None:
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(payload, dict):
        return None
    if payload.get("scope") != _profile_scope_key():
        return None
    cursor = payload.get("cursor")
    saved_at = payload.get("saved_at")
    if not isinstance(cursor, str) or not cursor or not isinstance(saved_at, str):
        return None
    try:
        saved = datetime.fromisoformat(saved_at)
    except ValueError:
        return None
    if saved.tzinfo is None:
        saved = saved.replace(tzinfo=UTC)
    if (datetime.now(UTC) - saved).total_seconds() > window:
        return None
    return cursor


def _persist_realtime_cursor(cfg: ChattoConfiguration, cursor: str | None) -> None:
    """Store the cursor of the last applied event, best-effort.

    Written only on a clean stop, not per event: a deploy or restart should not
    lose the gap, while a crash is allowed to — the alternative is a file write
    on every presence change. A failed write is logged and otherwise ignored,
    because losing the resume point is not a reason to fail a shutdown.
    """
    if not cursor or cfg.resume_window.value <= 0:
        return
    path = _realtime_cursor_file()
    if path is None:
        return
    payload = {
        "scope": _profile_scope_key(),
        "cursor": cursor,
        "saved_at": datetime.now(UTC).isoformat(),
    }
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(payload), encoding="utf-8")
    except OSError as exc:
        logger.debug("Chatto: realtime cursor not persisted: %s", exc)


# How similar an ``extra`` key must be to a declared config_key before the
# unknown-key check reads it as a likely typo or renamed field.
_NEAR_MISS_RATIO = 0.75


def _warn_on_unknown_extra_keys(config: PlatformConfig) -> None:
    """Warn about ``extra`` keys that look like misspelled config fields.

    A typo'd key silently resolves to the default otherwise — the warning is
    the only thing telling the user their ``require_mention_channles`` never
    reached us. Only near-matches against our declared config_keys are
    flagged (difflib similarity): keys that resemble nothing of ours are
    either placed there by the Hermes gateway itself (the shared-key loop in
    load_gateway_config() bridges reply_in_thread & co. into every platform's
    extra, hardcoded inline upstream — nothing to import) or are deliberate
    pass-throughs, and flagging those would just train users to ignore us.
    Internal markers (``_enabled_explicit``) are skipped outright.
    """
    extra: dict[str, Any] = getattr(config, "extra", None) or {}
    known = [field.config_key for field in ChattoConfiguration.fields()]
    for key in sorted(extra):
        if key.startswith("_") or key in known:
            continue
        best = max(known, key=lambda k: SequenceMatcher(None, key, k).ratio())
        if SequenceMatcher(None, key, best).ratio() >= _NEAR_MISS_RATIO:
            logger.warning(
                "Chatto: 'extra' key '%s' matches no config field — did you mean '%s'?",
                key,
                best,
            )


def hermes_validate_config(config: PlatformConfig) -> bool:
    """Check whether Chatto Plugin is configured.

    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register(). Do not change signature.
    Takes ``config``. Compare to hermes_is_connected().
    """
    resolved = ChattoConfiguration(pconfig=config)
    _warn_on_unknown_extra_keys(config)

    if len(resolved.allowed_users.value) > 0 and resolved.allow_all_users.value:
        logger.error(
            "Chatto: Conflicting configuration. Either use 'allowed_users' or 'allow_all_users' but not both."
        )
        return False

    require_rooms = set(resolved.require_mention_rooms.value)
    optional_rooms = set(resolved.optional_mention_rooms.value)
    overlap = sorted(require_rooms & optional_rooms)
    if overlap:
        logger.error(
            "Chatto: Conflicting configuration. Room(s) %s are on both "
            "'require_mention_rooms' and 'optional_mention_rooms' — "
            "each room must appear in at most one of them.",
            ", ".join(overlap),
        )
        return False

    if resolved.resume_window.value < 0:
        logger.error(
            "Chatto: 'resume_window' must be 0 (disabled) or a positive number "
            "of seconds, not %s.",
            resolved.resume_window.value,
        )
        return False

    _by_room, malformed_toolsets = _parse_room_toolsets(resolved.room_toolsets.value)
    if malformed_toolsets:
        logger.error(
            "Chatto: 'room_toolsets' entries must look like 'room_id:toolset' — "
            "these could not be read: %s",
            ", ".join(repr(entry) for entry in malformed_toolsets),
        )
        return False

    # base_url always resolves (it defaults to ChattoHQ), so the only real
    # question is whether a token came in.
    if resolved.token.value is None:
        logger.error("Chatto: CHATTO_TOKEN must be set.")
        return False

    # Publish for the plugin tools only now: a config we just rejected is not
    # one they should send with, and a rejected call must not overwrite a
    # snapshot an earlier, accepted call left behind.
    _resolved_chatto_configs[_profile_scope_key()] = resolved
    return True


def hermes_check_fn() -> bool:
    """Report whether the vendored chattolib dependency can be imported."""
    try:
        from chattolib import (
            client,  # noqa: F401 — vendored dependency probe
        )

        return True
    except ImportError:
        return False


# ---------------------------------------------------------------------------
# is_connected probe
# ---------------------------------------------------------------------------


def hermes_is_connected(config: PlatformConfig) -> bool:
    """Report whether the Chatto platform is configured and enabled.

    The name is fixed by the register() contract, but despite what it
    suggests this does not open a connection — it validates configuration
    only (see :func:`hermes_validate_config`); the gateway probes real
    connectivity through ``connect()``.
    """
    return bool(hermes_validate_config(config) and config.enabled)


def hermes_setup_fn() -> None:
    """Interactive setup wizard for Chatto. Is called by and only works in Hermes CLI context.

    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register().
    """
    from hermes_cli.setup import (
        print_success,
        prompt,
        prompt_yes_no,
        save_env_value,
    )

    url = prompt(
        "Chatto server URL (e.g. https://chat.example.com) or leave blank for default ChattoHQ on chat.chatto.run:"
    )
    if url:
        save_env_value(ChattoConfiguration.base_url.env_name, url)

    token = prompt("Chatto bot token (cht_BK_...):", password=True)
    if token:
        save_env_value(ChattoConfiguration.token.env_name, token)

    home = prompt("Home room ID for notifications (or empty):")
    if home:
        save_env_value(ChattoConfiguration.home_channel.env_name, home)

    allow_all = prompt_yes_no("Allow all users to talk? (true/false):")
    if allow_all:
        save_env_value(ChattoConfiguration.allow_all_users.env_name, str(allow_all))
    print_success("\n✓ Chatto configured. Restart the gateway to activate.")


def hermes_env_enablement_fn() -> dict:
    """Seed PlatformConfig.extra from the CHATTO_* environment variables.

    Returns a dict compatible with the PlatformConfig merge hook, holding
    exactly the values that are actually set via env — never fabricated
    defaults. Called by the platform registry during load_gateway_config(),
    which commits this dict onto the platform's ``extra`` verbatim; seeding a
    value that was not configured would overwrite what the user set in
    config.yaml (e.g. a YAML base_url clobbered by the ChattoHQ default).
    "Nothing set" therefore stays the adapter's business: every ConfigField
    falls back to its declared default (base_url to ChattoHQ) on its own.

    Seed keys are the config.yaml "extra" keys (config_key), NOT the env var
    names — ChattoConfiguration reads extra[config_key]. The special
    'home_channel' key is extracted by the gateway and becomes a proper
    HomeChannel dataclass on the PlatformConfig; every other key is merged
    into PlatformConfig.extra.

    Function name should be the same as register argument name with "hermes_" prefix, so we
    know that it is needed for plugin register().
    """
    seed: dict[str, Any] = {}
    for config_field in ChattoConfiguration.fields():
        env_value = os.getenv(config_field.env_name)
        if env_value:
            seed[config_field.config_key] = env_value.strip()

    logger.debug("seed: %s", {k: v for k, v in seed.items() if k != "token"})

    return seed


# ---------------------------------------------------------------------------
# Plugin registration entry point
# ---------------------------------------------------------------------------

# What each capability is called in the startup banner, keyed by the method
# that implements it. Derived from real overrides rather than hard-coded, so
# dropping a method drops its claim from the log instead of leaving a lie.
_CAPABILITY_LABELS = {
    "send": "text",
    "send_image_file": "images",
    "send_multiple_images": "image batches (bundled into one message)",
    "send_video": "video",
    "send_voice": "voice messages",
    "send_document": "documents",
    "add_reaction": "reactions",
    "edit_message": "message editing",
    "delete_message": "message deletion",
    "send_typing": "typing indicators",
    "create_handoff_thread": "threads",
    "create_room": "room creation",
    "toolsets_for_source": "per-room toolsets (CHATTO_ROOM_TOOLSETS)",
}


def _capabilities() -> list[str]:
    """Name the things this adapter genuinely implements itself.

    A capability counts only when ChattoAdapter overrides the base method —
    inheriting BasePlatformAdapter's fallback means the feature is not
    natively supported, and announcing it would mislead.
    """
    found = [
        label
        for name, label in _CAPABILITY_LABELS.items()
        if getattr(ChattoAdapter, name, None)
        is not getattr(BasePlatformAdapter, name, None)
    ]
    if ChattoAdapter.supports_code_blocks:
        found.append("code blocks")
    if ChattoAdapter.supports_status_text:
        found.append("custom status text")
    found.append("presence (refreshed while connected)")
    return found


# --------------------------------------------------------------------------- #
# Plugin tools: send DM to a user, or send to a room
# --------------------------------------------------------------------------- #


class _ToolRefusal(Exception):
    """A tool step that ends the call with a result for the model.

    The rendered JSON travels on the exception, so every early exit reads the
    same to Hermes and none of them escapes as an unhandled error.
    """

    def __init__(self, result_json: str) -> None:
        super().__init__(result_json)
        self.result_json = result_json


def _refuse(error: str, **extra: Any) -> _ToolRefusal:
    """Build a refusal result, so every early exit agrees on the shape."""
    return _ToolRefusal(json.dumps({"success": False, "error": error, **extra}))


def _int_arg(args: dict, key: str, default: int, *, low: int, high: int) -> int:
    """An integer tool argument, validated — a refusal rather than a guess.

    The value comes from a model, so "twenty" and a float are both reachable,
    and an unvalidated ``int()`` would raise out of the handler instead of
    answering. An absent argument takes the default; a *wrong* one is refused,
    which also keeps a falsy ``0`` from silently becoming the default and
    quietly widening the answer.
    """
    raw = args.get(key)
    if raw is None or raw == "":
        return default
    not_a_number = f"{key} must be a whole number between {low} and {high}"
    try:
        value = int(raw)
    except (TypeError, ValueError) as exc:
        raise _refuse(not_a_number) from exc
    if isinstance(raw, float) and not raw.is_integer():
        # int() would truncate 3.5 to 3, answering a question nobody asked.
        raise _refuse(not_a_number)
    if value < low or value > high:
        raise _refuse(f"{key} must be between {low} and {high}")
    return value


def _timestamp_arg(args: dict, key: str) -> datetime | None:
    """An ISO-8601 tool argument as a datetime, or a refusal.

    Search bounds are the first date arguments these tools take, so they get
    the same treatment as numbers: an absent value means "no bound", and a
    wrong one is refused rather than silently dropped — a caller who mistyped
    a date is better served by hearing it than by getting an unbounded search.
    A date without a time is midnight, and a value without an offset is read as
    UTC, matching how everything else here treats a bare timestamp.
    """
    raw = args.get(key)
    if raw is None or raw == "":
        return None
    if isinstance(raw, datetime):
        return raw if raw.tzinfo is not None else raw.replace(tzinfo=UTC)
    try:
        parsed = datetime.fromisoformat(str(raw).strip())
    except ValueError as exc:
        raise _refuse(
            f"{key} must be an ISO-8601 date or timestamp, e.g. '2026-01-31' "
            "or '2026-01-31T12:00:00Z'"
        ) from exc
    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)


def _enum_arg(args: dict, key: str, mapping: dict[str, Any], default: Any) -> Any:
    """A named tool argument resolved through ``mapping``, or a refusal.

    The choices are spelled out in the schema's enum, so an unknown name means
    the caller invented one; picking a default instead would answer a
    different question than the one asked.
    """
    raw = args.get(key)
    if raw is None or raw == "":
        return default
    value = mapping.get(str(raw).strip().lower())
    if value is None:
        raise _refuse(f"{key} must be one of: {', '.join(sorted(mapping))}")
    return value


@asynccontextmanager
async def _tool_client(cfg: ChattoConfiguration | None) -> AsyncIterator[ChattoClient]:
    """A short-lived client for one tool call, closed however the call ends.

    Credentials come from the resolved configuration and from nowhere else. A
    tool argument carrying ``base_url`` would send the configured bot token to
    whatever host the caller named, and every message the bot reads is
    something a caller could talk it into repeating.
    """
    if cfg is None:
        raise _refuse("Chatto configuration unavailable")
    token = cfg.token.value or ""
    if not token:
        raise _refuse("Chatto token not configured")
    try:
        client = ChattoClient(
            base_url=cfg.base_url.value or ChattoClient.DEFAULT_BASE_URL, token=token
        )
    except Exception as exc:
        raise _refuse(f"Failed to create Chatto client: {exc}") from exc
    try:
        yield client
    finally:
        try:
            await client.close()
        except Exception as exc:  # pragma: no cover — close is best-effort
            logger.debug("Chatto tools: client close failed: %s", exc)


# Read-tool bounds, named so the schemas and the validation cannot disagree.
# A chain is walked hop by hop, so its ceiling is a cost bound: each hop is a
# round trip, and a caller asking for the whole conversation has
# chatto_get_history to page it. The page size is bounded for the opposite
# reason — one read must stay inside a model context.
FOLLOW_REPLY_CHAIN_MAX = 10
HISTORY_PAGE_SIZE = 20
HISTORY_PAGE_MAX = 100

# Read-tool bounds for pinned messages and attachments — one call per page,
# kept small enough that a model can digest the output without choking.
PINNED_PAGE_SIZE = 20
ATTACHMENT_PAGE_SIZE = 40
# Search pages are bounded the same way, and for the same reason: a hit is
# something to read, not a dump. The server's own ceiling is 100, and a page
# may come back short while `next_cursor` is set — stale index hits are dropped
# after the provider matches them, which is not an error.
SEARCH_PAGE_SIZE = 20
SEARCH_PAGE_MAX = 100

# The model names a search order the way a person would; the proto spells it
# out. One table maps the two, so the schema's enum and the lookup cannot
# disagree about what "newest" means.
_SEARCH_ORDERS = {
    "relevance": MessageSearchOrder.RELEVANCE,
    "newest": MessageSearchOrder.NEWEST,
    "thread_activity": MessageSearchOrder.THREAD_ACTIVITY,
    "thread_created": MessageSearchOrder.THREAD_CREATED,
}
_SEARCH_SCOPES = {"followed_threads": MessageSearchScope.FOLLOWED_THREADS}
_SEARCH_GROUPS = {"thread": MessageSearchGroupBy.THREAD}

# Why search cannot answer right now, in terms someone can act on. The server
# returns FAILED_PRECONDITION for "the operator turned this off" and
# UNAVAILABLE for "the provider is not ready yet" — the second one resolves
# itself, the first one never does, and the caller should not have to guess
# which it hit.
_SEARCH_STATE_HINTS = {
    MessageSearchState.DISABLED: (
        "message search is turned off on this Chatto server; the operator has "
        "to enable it (`search.enabled = true`) and restart"
    ),
    MessageSearchState.STARTING: (
        "the search provider is starting up; try again shortly"
    ),
    MessageSearchState.INDEXING: (
        "the search provider is still building its index from the message "
        "history, so recent results may be missing; try again later"
    ),
    MessageSearchState.DEGRADED: (
        "the search provider is degraded and answering only partially"
    ),
    MessageSearchState.UNAVAILABLE: ("no search provider is reachable for this server"),
}
# A self-deleting notice: long enough to read, short enough not to linger.
# The floor keeps a TTL from deleting a message before anyone sees it; the
# ceiling keeps a "transient" notice from becoming a day-long fixture.
EPHEMERAL_TTL_DEFAULT = 60
EPHEMERAL_TTL_MIN = 5
EPHEMERAL_TTL_MAX = 3600


CHATTO_LIST_ROOMS_SCHEMA = {
    "name": "chatto_list_rooms",
    "description": (
        "List the Chatto rooms visible to the bot (channels and DM rooms), with "
        "their room IDs and whether the bot is a member. Check `is_member` before "
        "posting: the bot can see rooms it has not joined, and cannot post in "
        "them. Use this to resolve a channel name to its room ID — e.g. to find "
        "the 'platform:chat_id' address for a cron job's `deliver` target. "
        "Optional `name` filters case-insensitively by substring."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "name": {
                "type": "string",
                "description": (
                    "Optional: case-insensitive substring filter on the room name "
                    "(e.g. 'liv' matches 'Liv' and 'Liv-Dev'). Empty lists all rooms."
                ),
            },
        },
        "required": [],
        "additionalProperties": False,
    },
}

CHATTO_LIST_ROOM_ATTACHMENTS_SCHEMA = {
    "name": "chatto_list_room_attachments",
    "description": (
        "List the file attachments in a Chatto room (images, documents, etc.). "
        "Returns each attachment's ID, filename, MIME type, size and an optional "
        "thumbnail URL. Optional ``limit`` (default 40) and ``offset`` allow "
        "paginating through large attachment lists."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": "The Chatto room ID to list attachments in.",
            },
            "room_name": {
                "type": "string",
                "description": (
                    "Optional: human-readable room name — one of ``room_id`` or "
                    "``room_name`` is required."
                ),
            },
            "limit": {
                "type": "integer",
                "description": (
                    f"Maximum attachments to return (default {ATTACHMENT_PAGE_SIZE})."
                ),
            },
            "offset": {
                "type": "integer",
                "description": "Skip this many attachments (for pagination).",
            },
        },
        "required": [],
        "additionalProperties": False,
    },
}

CHATTO_LIST_PINNED_MESSAGES_SCHEMA = {
    "name": "chatto_list_pinned_messages",
    "description": (
        "List pinned messages in a Chatto room. Returns each message's event ID, "
        "body, author (user ID and login), and the pin timestamp. Optional "
        "``limit`` (default 20) and ``offset`` allow paginating through long "
        "pin lists."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": "The Chatto room ID to list pinned messages in.",
            },
            "room_name": {
                "type": "string",
                "description": (
                    "Optional: human-readable room name — one of ``room_id`` or "
                    "``room_name`` is required."
                ),
            },
            "limit": {
                "type": "integer",
                "description": f"Maximum pinned messages to return (default {PINNED_PAGE_SIZE}).",
            },
            "offset": {
                "type": "integer",
                "description": "Skip this many pinned messages (for pagination).",
            },
        },
        "required": [],
        "additionalProperties": False,
    },
}

CHATTO_PIN_MESSAGE_SCHEMA = {
    "name": "chatto_pin_message",
    "description": (
        "Pin or unpin a message in a Chatto room. Use ``pin=true`` to pin a "
        "message (it becomes visible via ``chatto_list_pinned_messages``), or "
        "``pin=false`` to remove an existing pin. Requires the ``event_id`` of "
        "the message to pin."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": "The Chatto room ID.",
            },
            "room_name": {
                "type": "string",
                "description": (
                    "Optional: human-readable room name — one of ``room_id`` or "
                    "``room_name`` is required."
                ),
            },
            "event_id": {
                "type": "string",
                "description": "The message event ID to pin or unpin.",
            },
            "pin": {
                "type": "boolean",
                "description": ("True to pin, false to unpin the message."),
            },
        },
        "required": ["event_id", "pin"],
        "additionalProperties": False,
    },
}

CHATTO_SET_CUSTOM_STATUS_SCHEMA = {
    "name": "chatto_set_custom_status",
    "description": (
        "Set the bot's custom status in Chatto. Shows an emoji and text next "
        "to the bot name in the client (e.g. 🛠️ processing, ⏳ awaiting review). "
        "Optional ``expires_in_seconds`` auto-clears the status; omit or 0 for "
        "manual clearing."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "emoji": {
                "type": "string",
                "description": (
                    "A single emoji to show next to the status text (e.g. 🛠️, ⏳, 🔴)."
                ),
            },
            "text": {
                "type": "string",
                "description": (
                    "The status text to display (e.g. 'processing', 'awaiting review')."
                ),
            },
            "expires_in_seconds": {
                "type": "integer",
                "description": (
                    "Optional: auto-clear the status after this many seconds. "
                    "Omit or 0 to keep the status until cleared."
                ),
            },
        },
        "required": ["emoji", "text"],
        "additionalProperties": False,
    },
}

CHATTO_DELETE_CUSTOM_STATUS_SCHEMA = {
    "name": "chatto_delete_custom_status",
    "description": ("Clear the bot's custom status, reverting to no status display."),
    "parameters": {
        "type": "object",
        "properties": {},
        "required": [],
        "additionalProperties": False,
    },
}


CHATTO_SEND_DM_SCHEMA = {
    "name": "chatto_send_dm",
    "description": (
        "Send a text message as a **Direct Message** to another Chatto user. "
        "Looks up the user by `user_login` or `user_id` and posts into the DM "
        "room you already have with them. Returns success with the room_id "
        "and message_ids, or an error."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "user_login": {
                "type": "string",
                "description": (
                    "The Chatto login name of the recipient "
                    "(e.g. 'nickk'). Exactly one of `user_login` or `user_id` is required."
                ),
            },
            "user_id": {
                "type": "string",
                "description": (
                    "The Chatto user ID of the recipient "
                    "(e.g. 'Uxxxxxxxx'). Exactly one of `user_login` or `user_id` is required."
                ),
            },
            "message": {
                "type": "string",
                "description": "The message text to send (markdown supported).",
            },
        },
        "required": ["message"],
        "additionalProperties": False,
    },
}

CHATTO_FETCH_LINK_PREVIEW_SCHEMA = {
    "name": "chatto_fetch_link_preview",
    "description": (
        "Fetch an OpenGraph link preview for a URL. Returns title, description, "
        "image URL and other metadata that Chatto's server would show when "
        "rendering the link in a message. Passes through whatever metadata the "
        "target site provides."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "url": {
                "type": "string",
                "description": (
                    "The URL to fetch a preview for (e.g. 'https://example.com')."
                ),
            },
        },
        "required": ["url"],
        "additionalProperties": False,
    },
}


CHATTO_SEND_ROOM_SCHEMA = {
    "name": "chatto_send_room",
    "description": (
        "Send a text message to a Chatto **room** (channel or DM room) "
        "by its room ID or room name. Channels the bot is configured to "
        "answer in only — a room on neither mention list stays read-only. "
        "Returns success with the message_ids, or an error."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": (
                    "The Chatto room ID of the destination "
                    "(e.g. 'RbmwiHk4vs9a3Ql'). Exactly one of `room_id` or `room_name` is required."
                ),
            },
            "room_name": {
                "type": "string",
                "description": (
                    "The display name of the room (case-insensitive). Exactly one "
                    "of `room_id` or `room_name` is required. If several rooms "
                    "share the name, the call returns their IDs instead of guessing."
                ),
            },
            "message": {
                "type": "string",
                "description": "The message text to send (markdown supported).",
            },
            "thread_root_event_id": {
                "type": "string",
                "description": (
                    "Optional: event ID of the thread-root message. When given, "
                    "this message is posted INSIDE that thread, which has to be "
                    "a thread root of the destination room. Mutually exclusive "
                    "with `create_thread`."
                ),
            },
            "create_thread": {
                "type": "boolean",
                "description": (
                    "Optional: post this message as the ROOT of a new thread "
                    "(default false = normal message). Only for channel rooms. "
                    "Mutually exclusive with `thread_root_event_id`."
                ),
            },
        },
        "required": ["message"],
        "additionalProperties": False,
    },
}

CHATTO_SEND_EPHEMERAL_SCHEMA = {
    "name": "chatto_send_ephemeral",
    "description": (
        "Send a text notice to a Chatto room that deletes itself after a TTL. "
        "Use it for progress or status updates that should not stay in the "
        "timeline. Same room rules as chatto_send_room: a channel the bot is "
        "configured to answer in only. Returns the message_ids and the TTL, or "
        "an error. Deletion is best-effort: a gateway restart before the TTL "
        "expires leaves the notice in the room."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": (
                    "The Chatto room ID of the destination. Exactly one of "
                    "`room_id` or `room_name` is required."
                ),
            },
            "room_name": {
                "type": "string",
                "description": (
                    "The display name of the room (case-insensitive). Exactly one "
                    "of `room_id` or `room_name` is required. If several rooms "
                    "share the name, the call returns their IDs instead of guessing."
                ),
            },
            "message": {
                "type": "string",
                "description": "The notice text (markdown supported).",
            },
            "ttl_seconds": {
                "type": "integer",
                "description": (
                    f"Seconds until the notice is deleted (default "
                    f"{EPHEMERAL_TTL_DEFAULT}, min {EPHEMERAL_TTL_MIN}, "
                    f"max {EPHEMERAL_TTL_MAX})."
                ),
            },
        },
        "required": ["message"],
        "additionalProperties": False,
    },
}

CHATTO_MARK_THREAD_AS_READ_SCHEMA = {
    "name": "chatto_mark_thread_as_read",
    "description": (
        "Mark a thread as read for the bot. This updates the server's read "
        "marker so Chatto clients show the thread as read. Useful after the bot "
        "has processed a thread's full context."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": "The Chatto room ID of the thread.",
            },
            "room_name": {
                "type": "string",
                "description": (
                    "Optional: room name — one of ``room_id`` or ``room_name`` is required."
                ),
            },
            "thread_root_event_id": {
                "type": "string",
                "description": ("The event ID of the thread root message."),
            },
        },
        "required": ["thread_root_event_id"],
        "additionalProperties": False,
    },
}


CHATTO_GET_MESSAGE_SCHEMA = {
    "name": "chatto_get_message",
    "description": (
        "Read one or more Chatto messages by event ID, with all their data: "
        "body, author (user ID + login), timestamps, reply/thread references "
        "(in_reply_to, thread_root_event_id), attachments, reactions, pinned "
        "and deleted status. Use it to resolve a `reply_to=`/`in_reply_to` "
        "reference or any other event ID mentioned in a message. "
        "`follow_reply_chain` (0-10) additionally walks the in_reply_to chain "
        "backwards and returns the ancestor messages oldest-first; it applies to "
        "a single `event_id` and is refused with `event_ids`. `event_ids` "
        "fetches several messages in one call, answered in the order asked for "
        "(then `event_id` is refused)."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": (
                    "The Chatto room ID the message lives in "
                    "(e.g. 'RbmwiHk4vs9a3Ql'). Exactly one of `room_id` or "
                    "`room_name` is required."
                ),
            },
            "room_name": {
                "type": "string",
                "description": (
                    "The room's display name (case-insensitive). Exactly one of "
                    "`room_id` or `room_name` is required; ambiguous names are "
                    "refused with the candidate IDs."
                ),
            },
            "event_id": {
                "type": "string",
                "description": (
                    "The event ID of the message to read (e.g. 'EM0D7tFdvd14EXS'). "
                    "Exactly one of `event_id` or `event_ids` is required."
                ),
            },
            "event_ids": {
                "type": "array",
                "items": {"type": "string"},
                "description": (
                    "Several event IDs to read in one call. Mutually exclusive "
                    "with `event_id`."
                ),
            },
            "follow_reply_chain": {
                "type": "integer",
                "description": (
                    "How many in_reply_to ancestors to resolve, 0-10 "
                    "(default 0 = do not follow). Single `event_id` only."
                ),
            },
        },
        "required": [],
        "additionalProperties": False,
    },
}

CHATTO_LIST_ROOM_GROUPS_SCHEMA = {
    "name": "chatto_list_room_groups",
    "description": (
        "List room groups as configured on the Chatto server. Each group has an "
        "ID, name, description, list of rooms, and optional sidebar links. "
        "Useful for browsing the server's room organisation."
    ),
    "parameters": {
        "type": "object",
        "properties": {},
        "required": [],
        "additionalProperties": False,
    },
}


CHATTO_GET_HISTORY_SCHEMA = {
    "name": "chatto_get_history",
    "description": (
        "Read the most recent messages of a Chatto room, or of one thread in "
        "it, oldest-first. Use it for context: what was said before a message "
        "the bot is about to answer or reference. With `thread_root_event_id` "
        "the thread's events are read instead of the room timeline; the root "
        "has to belong to the room and BE a root, like in chatto_send_room. "
        "`limit` (1-100, default 20) bounds the page; `has_older` says whether "
        "older history exists. To read further back, pass the previous "
        "result's `start_cursor` as `before` — one call reads one page."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "room_id": {
                "type": "string",
                "description": (
                    "The Chatto room ID to read (e.g. 'RbmwiHk4vs9a3Ql'). "
                    "Exactly one of `room_id` or `room_name` is required."
                ),
            },
            "room_name": {
                "type": "string",
                "description": (
                    "The room's display name (case-insensitive). Exactly one of "
                    "`room_id` or `room_name` is required; ambiguous names are "
                    "refused with the candidate IDs."
                ),
            },
            "thread_root_event_id": {
                "type": "string",
                "description": (
                    "Optional: read this thread's events instead of the room "
                    "timeline. The ID must be a thread ROOT in this room."
                ),
            },
            "limit": {
                "type": "integer",
                "description": (
                    "Maximum number of messages to return, 1-100 (default 20)."
                ),
            },
            "before": {
                "type": "string",
                "description": (
                    "Optional: page backwards. Pass the `start_cursor` of the "
                    "previous result — cursors are opaque positions, not event "
                    "IDs. Mutually exclusive with `after`."
                ),
            },
            "after": {
                "type": "string",
                "description": (
                    "Optional: page forwards. Pass the `end_cursor` of the "
                    "previous result. Mutually exclusive with `before`."
                ),
            },
        },
        "required": [],
        "additionalProperties": False,
    },
}


CHATTO_SEARCH_MESSAGES_SCHEMA = {
    "name": "chatto_search_messages",
    "description": (
        "Full-text search over Chatto messages the bot can read. Use it to find "
        "something said earlier when you do not know which room, thread or day: "
        "`chatto_get_history` reads a window, this finds a needle. Search is a "
        "server feature the operator has to enable; when it is off or still "
        "building its index, the answer says so instead of reporting no "
        'matches. Query syntax: words are required terms, "quoted text" is an '
        "exact phrase, `AND` separates terms, and `in:`, `from:`, `before:`, "
        "`after:`, `has:attachment` are filters. Scope it with `room_id`/"
        "`room_name`, `author_id` or the time bounds; `group_by: thread` "
        "answers one row per matching thread instead of per message."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": (
                    "Required search expression. Words are required terms "
                    '(all must match), "a quoted phrase" matches exactly, and '
                    "`AND` may join terms. Filters: `in:room-name`, "
                    "`from:user-login`, `before:2026-03-01`, "
                    "`after:2026-01-01`, `has:attachment`. A filter may be used "
                    "without a word or phrase."
                ),
            },
            "room_id": {
                "type": "string",
                "description": (
                    "Optional: search only this room (e.g. 'RbmwiHk4vs9a3Ql'). "
                    "Mutually exclusive with `room_name`. Without either, the "
                    "search covers every room the bot is a member of."
                ),
            },
            "room_name": {
                "type": "string",
                "description": (
                    "Optional: search only the room with this display name "
                    "(case-insensitive). Mutually exclusive with `room_id`; "
                    "ambiguous names are refused with the candidate IDs."
                ),
            },
            "author_id": {
                "type": "string",
                "description": (
                    "Optional: only messages posted by this Chatto user ID. To "
                    "filter by login instead, put `from:<login>` inside "
                    "`query`."
                ),
            },
            "created_after": {
                "type": "string",
                "description": (
                    "Optional ISO-8601 lower bound on posting time, exclusive "
                    "(e.g. '2026-01-31' or '2026-01-31T12:00:00Z'). When "
                    "`query` also carries `after:`, the stricter bound wins."
                ),
            },
            "created_before": {
                "type": "string",
                "description": (
                    "Optional ISO-8601 upper bound on posting time, exclusive. "
                    "When `query` also carries `before:`, the stricter bound "
                    "wins."
                ),
            },
            "has_attachments": {
                "type": "boolean",
                "description": (
                    "Optional: true returns only messages that currently have "
                    "attachments."
                ),
            },
            "order": {
                "type": "string",
                "enum": [
                    "relevance",
                    "newest",
                    "thread_activity",
                    "thread_created",
                ],
                "description": (
                    "Result order (default relevance). `thread_activity` and "
                    "`thread_created` require `group_by: thread`."
                ),
            },
            "group_by": {
                "type": "string",
                "enum": ["thread"],
                "description": (
                    "Optional: with `thread`, return one row per matching "
                    "thread instead of one per message, with the thread's root, "
                    "reply count and latest reply."
                ),
            },
            "scope": {
                "type": "string",
                "enum": ["followed_threads"],
                "description": (
                    "Optional: with `followed_threads`, search only threads the "
                    "bot follows. Combine with `group_by: thread` for a thread "
                    "list."
                ),
            },
            "limit": {
                "type": "integer",
                "description": (
                    "Maximum results per page, 1-100 (default 20). The page is "
                    "counted in messages, or in threads when `group_by` is "
                    "`thread`."
                ),
            },
            "cursor": {
                "type": "string",
                "description": (
                    "Opaque `next_cursor` from a previous call, for the next "
                    "page. Bound to this caller and to every other query field, "
                    "so repeat the same query with it. Results come from a live "
                    "index: a page may repeat or omit a hit that moved while "
                    "paging."
                ),
            },
        },
        "required": ["query"],
        "additionalProperties": False,
    },
}


def _chatto_config_for_tools() -> ChattoConfiguration | None:
    """Resolve the Chatto configuration for plugin tool handlers.

    Two tiers, most specific first:
    1. The snapshot ``hermes_validate_config`` published for *this* profile — the
       authoritative one in a running gateway.
    2. A fresh read of config.yaml plus the serving profile's own secrets, for
       the cron and standalone sessions where the gateway load hooks never ran
       and the dotenv residue has been stripped from ``os.environ``. Mirrors how
       the built-in send_message tool resolves its pconfig
       (tools/send_message_tool.py).

    Both tiers are keyed by the serving profile, so a multiplexed gateway hands
    each profile its own credentials rather than whichever one was served last.

    Returns ``None`` when neither tier yields a configuration; the caller
    reports that rather than guessing at credentials.
    """
    scope_key = _profile_scope_key()
    snapshot = _resolved_chatto_configs.get(scope_key)
    if snapshot is not None:
        return snapshot

    # config.yaml's own ``extra`` block, so a deployment configured there — and
    # nowhere in .env — is honoured outside the gateway process too.
    try:
        from gateway.config import Platform as GatewayPlatform
        from gateway.config import load_gateway_config

        pconfig = load_gateway_config().platforms.get(
            GatewayPlatform(ChattoConstants.PLATFORM_NAME)
        )
        resolved = ChattoConfiguration(pconfig=_pconfig_or_empty(pconfig))
    except Exception as exc:
        logger.warning("Chatto tools: could not read gateway config: %s", exc)
        return None

    # In a fresh cron scope the dotenv residue is stripped from os.environ, so
    # ChattoConfiguration's env layer sees nothing. The secrets then come from
    # the scope Hermes installed for the profile being served when there is one —
    # that mapping is that profile's authoritative secret set, and rebuilding
    # one from the launch home would answer with another profile's credentials.
    # With no scope installed (CLI, bare standalone sender) the profile's own
    # .env is the right source, and build_profile_secret_scope reads it
    # directly in gateway, cron and CLI contexts alike.
    try:
        from agent.secret_scope import current_secret_scope

        profile_secrets = current_secret_scope()
        if profile_secrets is None:
            from agent.secret_scope import build_profile_secret_scope
            from hermes_constants import get_hermes_home

            profile_secrets = dict(build_profile_secret_scope(get_hermes_home()))
    except Exception as exc:
        logger.warning("Chatto tools: profile secret scope build failed: %s", exc)
        profile_secrets = {}

    resolved.overlay_profile_secrets(profile_secrets)
    # Cache only a configuration that can actually send. Caching a tokenless one
    # would pin this failure for the life of the process: the next call in a
    # properly scoped context would be served the degraded config and keep
    # reporting "token not configured" long after the scope came back.
    if resolved.token.value:
        _resolved_chatto_configs[scope_key] = resolved
    return resolved


def _parse_room_toolsets(
    entries: Iterable[str],
) -> tuple[dict[str, list[str]], list[str]]:
    """Split ``room_id:toolset`` entries into per-room toolset lists.

    Returns the mapping plus the entries that could not be read. Several
    entries may name the same room; they accumulate in the order given, which
    is what lets a room keep more than one toolset without inventing a nested
    separator inside a comma-separated list. Malformed entries are reported
    rather than raised: startup validation refuses the config, and a running
    gateway must not die on a typo it can ignore.
    """
    by_room: dict[str, list[str]] = {}
    malformed: list[str] = []
    for raw in entries:
        entry = str(raw).strip()
        room_id, _, toolset = entry.partition(":")
        room_id, toolset = room_id.strip(), toolset.strip()
        if not room_id or not toolset:
            malformed.append(entry)
            continue
        toolsets = by_room.setdefault(room_id, [])
        if toolset not in toolsets:
            toolsets.append(toolset)
    return by_room, malformed


def _room_answers_in(
    cfg: ChattoConfiguration | None, room_id: str, kind: RoomKind | None
) -> bool:
    """Whether the bot takes part in this room: answers there, and may post.

    The single definition of that question, on purpose. It answers for two
    callers whose disagreement would be a bug the user cannot see:

    - ``ChattoAdapter._answers_in_room``, the *inbound* gate — whether a
      message here reaches the agent at all
    - ``_room_is_writable``, the *outbound* gate on the send tools, so a tool
      cannot post into a room that would never have answered

    Chatto knows only DMs and channels, so the split is ``kind == DM``: a DM
    always answers (``/join`` must stay reachable), and every other kind —
    channel, an unknown kind (which is how servers that never set ``kind`` show
    up), or a room the adapter has not resolved yet, so ``kind`` is None —
    opts in through one of the two mention lists. The lists are mutually
    exclusive by validation; should contradictory config reach a running adapter
    anyway, optional beats require, because answering a room is safer than
    muting it. A ``cfg`` of None means no configuration could be resolved at
    all, and then nothing but a DM qualifies.
    """
    if kind == RoomKind.DM:
        return True
    if cfg is None:
        return False
    if room_id in cfg.optional_mention_rooms.value:
        return True
    return room_id in cfg.require_mention_rooms.value


def _pconfig_or_empty(pconfig: PlatformConfig | None) -> PlatformConfig:
    """The Chatto PlatformConfig from config.yaml, or an empty stand-in."""
    if pconfig is not None:
        return pconfig
    from gateway.config import PlatformConfig as GatewayPlatformConfig

    return GatewayPlatformConfig(enabled=True, extra={})


def _room_is_writable(cfg: ChattoConfiguration | None, room: Room) -> bool:
    """Whether the bot may post into this room on its own initiative.

    Delegates to ``_room_answers_in`` so the outbound gate is the *same*
    function as the inbound one: a room that would never be answered must not
    be postable, and two copies of that rule would drift apart until the send
    tool became a way around the read-only promise.
    """
    return _room_answers_in(cfg, room.id, room.kind)


def _not_joined_refusal(room: Room, action: str) -> _ToolRefusal:
    """The refusal for a room the bot can see but has not joined.

    ``action`` names what the caller wanted to do there — "post there", "read
    its history there" — because the remedy is identical for both while the
    symptom is not, and a read that reports a sending problem sends the reader
    looking in the wrong place. The room kinds differ in the remedy: giving a
    channel the DM remedy sends the user down a path that cannot work, since a
    direct message to the bot is not how you get it into a channel.

    The test is on ``DM`` rather than on ``CHANNEL`` on purpose: a server that
    never sets ``kind`` reports ``UNSPECIFIED``, and this plugin counts an unknown
    kind as a channel everywhere else (``_room_answers_in``, ``_refresh_rooms``).
    Keying on ``CHANNEL`` would hand exactly the advice this function exists to
    avoid to every self-hosted server on an older or looser build.
    """
    remedy = (
        "Have someone start a DM with the bot account"
        if room.kind == RoomKind.DM
        else f"Invite the bot account in Chatto, or DM it `/join {room.id}`"
    )
    return _refuse(
        f"Room '{room.name or room.id}' ({room.id}) is not one the bot has "
        f"joined, so it cannot {action}. {remedy}, then try again."
    )


def _read_only_room_refusal(room_label: str) -> _ToolRefusal:
    """The refusal a read-only room earns, naming where to opt it in.

    This reply is where an operator copies the room ID from, so it spells both
    env vars out verbatim.
    """
    return _refuse(
        f"Room '{room_label}' is configured read-only: the bot neither answers "
        "there nor posts to it on its own initiative. To allow that, add the "
        f"room ID to {ChattoConfiguration.require_mention_rooms.env_name} (only "
        f"answers when addressed) or {ChattoConfiguration.optional_mention_rooms.env_name} "
        "(answers everything) in ~/.hermes/.env, and restart the gateway."
    )


class _ChunkDelivery(NamedTuple):
    """What a split send actually put into a room.

    ``error`` is the exception that stopped the batch, if any: the chunks in
    ``message_ids`` did go out and must not be resent. Each caller shapes this
    into its own result contract — a ``SendResult`` for the adapter, a refusal
    for the tools — because those are genuinely different contracts; the
    *sending* is not. ``chunks_total`` counts what the split produced, so a
    caller can report "got 1 of 3 out" without re-deriving the split.
    """

    message_ids: list[str]
    root_event_id: str
    error: Exception | None
    chunks_total: int


async def _deliver_chunks(
    client: ChattoClient,
    room_id: str,
    message: str,
    *,
    thread_root_event_id: str = "",
    create_thread: bool = False,
    on_posted: Callable[[str], None] | None = None,
) -> _ChunkDelivery:
    """Post a message split at the server limit, stopping at the first failure.

    One implementation for both send paths, because the decisions are the same
    and a copy of each drifts: split at ``SPLIT_THRESHOLD`` like every other
    send, and route a split answer as *one* answer — chunk 1 opens a thread
    (``create_thread``) or joins the one passed in, and every later chunk
    carries that root, so a long reply does not scatter across the timeline.

    ``on_posted`` is the adapter's ``_mark_seen``: the tools have no dedup state
    to keep, the adapter must not treat its own output as unread input. The
    first failure stops the batch and is returned rather than raised — the
    caller knows better than this function what a partial send means to it, and
    what to call the exception (``ChattoError`` is a server failure, anything
    else is a bug here).
    """
    normalised = _normalise_outbound_text(message)
    chunks = BasePlatformAdapter.truncate_message(
        normalised, ChattoConstants.SPLIT_THRESHOLD
    )
    message_ids: list[str] = []
    root_id = thread_root_event_id
    for chunk in chunks:
        try:
            posted = await client.post_message(
                room_id=room_id, body=chunk, thread_root_event_id=root_id
            )
        except Exception as exc:
            return _ChunkDelivery(message_ids, root_id, exc, len(chunks))
        message_ids.append(posted.id)
        if on_posted is not None:
            on_posted(posted.id)
        if create_thread and not root_id:
            root_id = posted.id
    return _ChunkDelivery(message_ids, root_id, None, len(chunks))


async def _post_chunks(
    client: ChattoClient,
    room_id: str,
    message: str,
    *,
    thread_root_event_id: str = "",
    create_thread: bool = False,
) -> tuple[list[str], str]:
    """Post a message through ``_deliver_chunks`` as a tool result. ``(ids, root)``.

    A chunk that fails leaves the earlier ones delivered, so this refuses with
    exactly what went out: calling a half-delivered send a plain failure
    invites an agent to retry and duplicate what the room already has.
    """
    delivery = await _deliver_chunks(
        client,
        room_id,
        message,
        thread_root_event_id=thread_root_event_id,
        create_thread=create_thread,
    )
    if delivery.error is not None:
        # Blame the server only for a server error. chattolib maps every
        # transport failure to ChattoError, so anything else is a bug on our
        # side — calling that "Chatto API error" would send an operator hunting
        # the wrong component.
        cause = (
            "Chatto API error"
            if isinstance(delivery.error, ChattoError)
            else "Internal plugin error"
        )
        raise _refuse(
            f"{cause} on chunk {len(delivery.message_ids) + 1}/"
            f"{delivery.chunks_total}: {delivery.error}",
            # partial marks a half-delivered send; the caller must not
            # resend it from the top without knowing about these chunks.
            partial=bool(delivery.message_ids),
            room_id=room_id,
            message_ids=delivery.message_ids,
            chunks_sent=len(delivery.message_ids),
            chunks_total=delivery.chunks_total,
        ) from delivery.error
    return delivery.message_ids, delivery.root_event_id


async def _find_user_by_login(client: ChattoClient, user_login: str) -> tuple[str, str]:
    """Find a user among the members of our DM rooms, ``(user_id, login)``.

    Both are empty when no room yields the login. Only a person can start a DM,
    so a DM room is the only place a deliverable recipient can be — scanning
    the channels the bot sits in would cost one member RPC per room and could
    resolve a login whose DM does not exist yet. A bot cannot read the
    server-wide user directory either: that needs admin rights.

    Every DM room is searched, including one whose membership the directory
    left unreported — that field is optional, and skipping such a room would
    make its login unfindable while the DM room sits right there.
    """
    for rws in await client.list_rooms():
        room = rws.room
        if room is None or room.kind != RoomKind.DM:
            continue
        try:
            member_ids, _ = await client.list_room_members(room.id, search=user_login)
        except ChattoError as exc:
            # A room we cannot enumerate must not abort the search over the
            # remaining ones.
            logger.debug("Chatto tools: member search in %s failed: %s", room.id, exc)
            continue
        if not member_ids:
            continue
        for member in await client.batch_get_room_members(room.id, member_ids):
            user = member.user
            if (
                user is not None
                and user.login
                and user.login.casefold() == user_login.casefold()
            ):
                return user.id, user.login
    return "", ""


async def _find_dm_room(
    client: ChattoClient, our_id: str, other_id: str
) -> Room | None:
    """The DM room between two users, or None when there is none yet.

    Membership is read from the room object the directory already returned, and
    only falls back to a ListMembers RPC when that came back empty:
    ``member_user_ids`` is not among the fields the ListRooms documentation
    lists for ``RoomWithViewerState``, so a server is free to leave it
    unpopulated — and a DM room we can see plainly exists, whatever the
    directory said about who is in it. The fallback costs one RPC per DM room
    and only on servers that need it.
    """
    for rws in await client.list_rooms():
        room = rws.room
        if room is None or room.kind != RoomKind.DM:
            continue
        if rws.member_user_ids:
            if other_id in rws.member_user_ids and our_id in rws.member_user_ids:
                return room
            continue
        try:
            member_ids, _page = await client.list_room_members(room.id)
        except ChattoError as exc:
            # An unreadable room cannot be the one we want, but it must not
            # abort the search over the remaining ones.
            logger.debug("Chatto tools: members of DM %s unreadable: %s", room.id, exc)
            continue
        if other_id in member_ids and our_id in member_ids:
            return room
    return None


class _ResolvedRoom(NamedTuple):
    """A room the send path resolved, plus whether the bot belongs to it.

    Membership travels with the room because the directory answers a broader
    question than the send path can use: a room the bot can *see* is not one it
    can *post in* (ListRooms returns visible channels, membership or room.list
    permission). Splitting that into two lookups would be a second chance to
    disagree with the state the first one already carried.
    """

    room: Room
    is_member: bool


async def _resolve_room(
    client: ChattoClient, room_id: str, room_name: str
) -> _ResolvedRoom:
    """The destination room, by ID or by name.

    A name is not unique on Chatto — two room groups may both hold an "Ops" —
    and delivering to whichever the directory listed first is not a choice
    anyone made, so an ambiguous name asks for the ID instead.
    """
    if room_id:
        room_viewer_state = await client.get_room(room_id)
        if room_viewer_state is None or room_viewer_state.room is None:
            raise _refuse(
                f"Room {room_id} was not found, or the bot cannot see it. "
                "Use chatto_list_rooms to look up a room ID."
            )
        return _ResolvedRoom(
            room_viewer_state.room, room_viewer_state.viewer_state.is_member
        )

    matches = [
        _ResolvedRoom(rws.room, rws.viewer_state.is_member)
        for rws in await client.list_rooms()
        if rws.room is not None
        and (rws.room.name or "").casefold() == room_name.casefold()
    ]
    if not matches:
        raise _refuse(
            f"Room not found: {room_name}. Use chatto_list_rooms to see the available names."
        )
    if len(matches) > 1:
        candidates = ", ".join(
            f"{match.room.id} ('{match.room.name}')" for match in matches
        )
        raise _refuse(
            f"Room name '{room_name}' is ambiguous — {len(matches)} rooms carry it: "
            f"{candidates}. Pass the room_id instead."
        )
    return matches[0]


async def _resolve_thread_root(
    client: ChattoClient, room: Room, event_id: str
) -> Message:
    """The thread root a tool was pointed at, or a refusal saying why not.

    Both tools that take a ``thread_root_event_id`` — posting into one and
    reading one — resolve it here, because two copies of "is this a root?" is
    one too many: they drift, and the drift shows up as two tools disagreeing
    about the same event ID.

    Two things have to hold. The message has to exist in this room (a root from
    another room would attach a thread to the wrong conversation), and it has
    to BE a root (a reply's ID names no thread, so following it lands nowhere).
    """
    root = await client.get_message(room.id, event_id)
    if root is None:
        raise _refuse(
            f"No message {event_id} in room {room.id}. It has to be an event ID "
            f"from this room — chatto_get_history lists them."
        )
    if root.thread_root_event_id:
        raise _refuse(
            f"Message {event_id} is a reply inside thread "
            f"{root.thread_root_event_id}, not a thread root. Pass that "
            "thread's root ID instead."
        )
    return root


async def _send_dm(args: dict) -> str:
    """Resolve the recipient, find the DM room, post. See _handle_chatto_send_dm."""
    user_login = args.get("user_login") or ""
    user_id = args.get("user_id") or ""
    message = args.get("message") or ""

    if not message:
        raise _refuse("message is required")
    if not user_login and not user_id:
        raise _refuse("Exactly one of user_login or user_id is required")

    async with _tool_client(_chatto_config_for_tools()) as client:
        # A bot cannot read the server-wide user directory (that needs admin
        # rights), so a login is resolved among the members of shared rooms.
        target_user_id = user_id
        target_login = user_login
        if not target_user_id:
            target_user_id, target_login = await _find_user_by_login(client, user_login)
            if not target_user_id:
                raise _refuse(f"User not found: {user_login} shares no DM with the bot")

        # Only people may start a DM (Chatto permissions guide), so a recipient
        # who has never written to the bot has no room to post into — say so
        # instead of guessing at a server rejection later.
        dm_room = await _find_dm_room(client, (await client.me()).id, target_user_id)
        if dm_room is None:
            raise _refuse(
                f"No DM with {user_login or user_id} exists yet: on Chatto only a "
                "person can start a DM, so they have to send the bot a message "
                "once before it can message them"
            )

        message_ids, _root_id = await _post_chunks(client, dm_room.id, message)
        return json.dumps(
            {
                "success": True,
                "dm_room_id": dm_room.id,
                "recipient": target_login,
                "recipient_user_id": target_user_id,
                "message_ids": message_ids,
                "chunks_sent": len(message_ids),
            }
        )


async def _send_room(args: dict) -> str:
    """Resolve the room, check the gates, post. See _handle_chatto_send_room."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    message = args.get("message") or ""
    thread_root_event_id = args.get("thread_root_event_id") or ""
    create_thread = bool(args.get("create_thread", False))

    if not message:
        raise _refuse("message is required")
    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")
    if thread_root_event_id and create_thread:
        raise _refuse("thread_root_event_id and create_thread are mutually exclusive")

    cfg = _chatto_config_for_tools()
    async with _tool_client(cfg) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        # Posting into a room the plugin treats as read-only would break a
        # promise the README makes about unlisted rooms staying silent.
        if not _room_is_writable(cfg, room):
            raise _read_only_room_refusal(room.name or room.id)

        if create_thread and room.kind == RoomKind.DM:
            raise _refuse(
                "A DM has no threads: drop create_thread and send it as a normal message."
            )

        # The directory lists rooms the bot can see but has not joined, and
        # posting into one fails server-side with an error that names neither
        # the room nor the fix.
        if not resolved.is_member:
            raise _not_joined_refusal(room, "post there")

        # A thread root has to belong to this room and has to BE a root, or
        # the message attaches to a thread somewhere else entirely — or to
        # none that can be followed. One shared check, so this tool and
        # chatto_get_history cannot answer differently about the same ID.
        if thread_root_event_id:
            await _resolve_thread_root(client, room, thread_root_event_id)

        message_ids, root_event_id = await _post_chunks(
            client,
            room.id,
            message,
            thread_root_event_id=thread_root_event_id,
            create_thread=create_thread,
        )

        # Any thread the bot posts into is followed, not only one it opened:
        # Chatto delivers a thread's replies to an account that follows it (or
        # was mentioned in it directly), so replying into someone else's thread
        # without following it can end the conversation the agent just joined.
        # Mirrors the adapter's own send(), which follows every thread it uses.
        if root_event_id:
            try:
                await client.follow_thread(room.id, root_event_id)
            except Exception as exc:
                # Follow is best-effort: the thread exists and the message was
                # delivered, so a hiccup here is not a failed send.
                logger.debug(
                    "Chatto tools: follow_thread(%s) failed: %s", root_event_id, exc
                )

        return json.dumps(
            {
                "success": True,
                "room_id": room.id,
                "room_name": room.name,
                "message_ids": message_ids,
                "chunks_sent": len(message_ids),
                "thread_root_event_id": root_event_id,
                "created_thread": bool(create_thread),
            }
        )


_EPHEMERAL_DELETION_TASKS: set[asyncio.Task[None]] = set()


async def _send_ephemeral(args: dict) -> str:
    """Post a notice that deletes itself. See _handle_chatto_send_ephemeral."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    message = args.get("message") or ""
    ttl_seconds = _int_arg(
        args,
        "ttl_seconds",
        EPHEMERAL_TTL_DEFAULT,
        low=EPHEMERAL_TTL_MIN,
        high=EPHEMERAL_TTL_MAX,
    )

    if not message:
        raise _refuse("message is required")
    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")

    cfg = _chatto_config_for_tools()
    if cfg is None:
        raise _refuse("Chatto configuration unavailable")
    async with _tool_client(cfg) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        # Same gates as chatto_send_room: this tool posts, so it must not be a
        # way into a room the plugin keeps read-only.
        if not _room_is_writable(cfg, room):
            raise _read_only_room_refusal(room.name or room.id)
        if not resolved.is_member:
            raise _not_joined_refusal(room, "post there")

        message_ids, _root = await _post_chunks(client, room.id, message)

    _schedule_ephemeral_deletion(cfg, room.id, message_ids, ttl_seconds)
    return json.dumps(
        {
            "success": True,
            "room_id": room.id,
            "message_ids": message_ids,
            "chunks_sent": len(message_ids),
            "expires_in_seconds": ttl_seconds,
        }
    )


def _schedule_ephemeral_deletion(
    cfg: ChattoConfiguration,
    room_id: str,
    event_ids: list[str],
    ttl_seconds: int,
) -> None:
    """Delete a delivered notice once its TTL has passed, best-effort.

    The task is held in a module-level set because asyncio keeps only a weak
    reference: an unreferenced task can be collected before it ever runs.
    Failure to delete is logged, never raised — the message was delivered, and
    a process restart simply leaves it behind.
    """
    task = asyncio.create_task(_delete_after_ttl(cfg, room_id, event_ids, ttl_seconds))
    _EPHEMERAL_DELETION_TASKS.add(task)
    task.add_done_callback(_EPHEMERAL_DELETION_TASKS.discard)


async def _delete_after_ttl(
    cfg: ChattoConfiguration,
    room_id: str,
    event_ids: list[str],
    ttl_seconds: int,
) -> None:
    """Sleep out the TTL, then delete each chunk it covered."""
    await asyncio.sleep(ttl_seconds)
    try:
        async with _tool_client(cfg) as client:
            for event_id in event_ids:
                try:
                    await client.delete_message(room_id, event_id)
                except ChattoError as exc:
                    logger.debug(
                        "Chatto tools: ephemeral message %s in %s not deleted: %s",
                        event_id,
                        room_id,
                        exc,
                    )
    except Exception as exc:
        # No caller to raise to: the notice was delivered and this task exists
        # only to make it disappear again. A transport oddity that chattolib
        # did not map to ChattoError must not surface as an unretrieved task
        # exception, so it is logged like any other cleanup failure.
        logger.debug(
            "Chatto tools: ephemeral cleanup of %s in %s failed: %s",
            event_ids,
            room_id,
            exc,
            exc_info=True,
        )


async def _list_rooms(args: dict) -> str:
    """List the rooms visible to the bot. See _handle_chatto_list_rooms."""
    name_filter = (args.get("name") or "").strip().casefold()

    async with _tool_client(_chatto_config_for_tools()) as client:
        rooms_out = []
        for rws in await client.list_rooms():
            room = rws.room
            if room is None:
                continue
            # A filter implies named rooms: skip DMs (empty name) and
            # non-matching names alike.
            if name_filter and not (room.name and name_filter in room.name.casefold()):
                continue
            rooms_out.append(
                {
                    "room_id": room.id,
                    "name": room.name or "",
                    "kind": room.kind.value,
                    "threading_mode": room.threading_mode.value,
                    "archived": room.archived,
                    "is_member": rws.viewer_state.is_member,
                    # DM rooms carry no name, so they are labelled by their ID
                    # instead of leaving the model with a blank to match on.
                    "label": room.name or f"(DM room {room.id})",
                }
            )
        return json.dumps(
            {
                "success": True,
                "count": len(rooms_out),
                "rooms": rooms_out,
                "hint": (
                    "Use room_id in chatto_send_room, or as the chat_id part of a "
                    "cron deliver target like 'chatto-platform:<room_id>'."
                ),
            }
        )


async def _list_room_attachments(args: dict) -> str:
    """List file attachments in a Chatto room. See _handle_chatto_list_room_attachments."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    limit = _int_arg(
        args, "limit", ATTACHMENT_PAGE_SIZE, low=1, high=ATTACHMENT_PAGE_SIZE * 5
    )
    offset = _int_arg(args, "offset", 0, low=0, high=100_000)

    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")

    async with _tool_client(_chatto_config_for_tools()) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        attachments, page = await client.list_room_attachments(
            room.id, limit=limit, offset=offset
        )
        return json.dumps(
            {
                "success": True,
                "room_id": room.id,
                "room_name": room.name,
                "count": len(attachments),
                "offset": offset,
                "has_more": page.has_more,
                "total_count": page.total_count,
                "attachments": attachments,
            }
        )


async def _list_pinned_messages(args: dict) -> str:
    """List pinned messages in a Chatto room. See _handle_chatto_list_pinned_messages."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    limit = _int_arg(args, "limit", PINNED_PAGE_SIZE, low=1, high=PINNED_PAGE_SIZE * 5)
    offset = _int_arg(args, "offset", 0, low=0, high=100_000)

    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")

    async with _tool_client(_chatto_config_for_tools()) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        page = await client.list_pinned_messages(room.id, limit=limit, offset=offset)
        messages = [pm.message for pm in page.pinned_messages if pm.message is not None]
        author_logins = await _batch_author_logins(
            client, room.id, [msg.actor_id for msg in messages if msg.actor_id]
        )
        pinned_out = [
            {
                "event_id": msg.id,
                "body": msg.body,
                "author_id": msg.actor_id,
                "author_login": author_logins.get(msg.actor_id) or "",
                "posted_at": _iso_or_none(msg.created_at),
            }
            for msg in messages
        ]
        return json.dumps(
            {
                "success": True,
                "room_id": room.id,
                "room_name": room.name,
                "count": len(pinned_out),
                "has_more": page.page.has_more,
                "total_count": page.page.total_count,
                "pinned_messages": pinned_out,
                "latest_pin_marker": page.latest_pin_marker,
            }
        )


async def _pin_message(args: dict) -> str:
    """Pin or unpin a message in a Chatto room. See _handle_chatto_pin_message."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    event_id = args.get("event_id") or ""
    pin = args.get("pin")

    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")
    if not event_id:
        raise _refuse("event_id is required")
    if pin is None:
        raise _refuse("pin (true/false) is required")

    cfg = _chatto_config_for_tools()
    async with _tool_client(cfg) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        # Pinning changes what the room shows everyone, so it is a write:
        # a room on neither mention list stays read-only, exactly as it does
        # for chatto_send_room.
        if not _room_is_writable(cfg, room):
            raise _read_only_room_refusal(room.name or room.id)
        if not resolved.is_member:
            raise _not_joined_refusal(room, "pin a message there")

        if pin:
            await client.pin_message(room.id, event_id)
            return json.dumps(
                {
                    "success": True,
                    "action": "pinned",
                    "room_id": room.id,
                    "event_id": event_id,
                }
            )
        unpinned = await client.unpin_message(room.id, event_id)
        return json.dumps(
            {
                "success": True,
                "action": "unpinned",
                "room_id": room.id,
                "event_id": event_id,
                "unpinned": unpinned,
            }
        )


def _iso_or_none(moment: datetime | None) -> str | None:
    """A Message timestamp as ISO-8601, or None when the server sent none.

    chattolib's own formatter does the work: it pins a naive timestamp to UTC
    instead of letting it print without an offset, so two timestamps from
    different sources compare the way a reader expects.
    """
    if moment is None:
        return None
    return format_datetime(moment)


async def _batch_author_logins(
    client: ChattoClient, room_id: str, actor_ids: list[str]
) -> dict[str, str]:
    """Map actor IDs to logins, using the room directory the bot may read.

    ``batch_get_room_members`` needs no admin rights and answers member
    profiles for the caller's own rooms, so it is the one sanctioned way to
    turn an ``actor_id`` into a name here. Members who left (or were never
    members) come back missing — their IDs stay as their only label, which
    is still unambiguous for the caller. A directory that refuses the call
    costs the names, not the messages: the logins are enrichment, and an
    unreadable roster must not make a readable history unreadable.
    """
    unique_ids = list({actor_id for actor_id in actor_ids if actor_id})
    if not unique_ids:
        return {}
    try:
        members = await client.batch_get_room_members(room_id, unique_ids)
    except ChattoError as exc:
        logger.debug("Chatto tools: members of room %s unreadable: %s", room_id, exc)
        return {}
    logins: dict[str, str] = {}
    for member in members:
        if member.user is not None and member.user.id:
            logins[member.user.id] = member.user.login
    return logins


def _attachment_out(attachment: MessageAttachment) -> dict[str, Any]:
    """One attachment as the tool's JSON shape."""
    out: dict[str, Any] = {
        "id": attachment.id,
        "filename": attachment.filename,
        "content_type": attachment.content_type,
        "width": attachment.width,
        "height": attachment.height,
        "description": attachment.description,
    }
    if attachment.asset_url is not None:
        out["asset_url"] = attachment.asset_url.url
    if attachment.thumbnail_asset_url is not None:
        out["thumbnail_asset_url"] = attachment.thumbnail_asset_url.url
    if attachment.video_processing is not None:
        # A video carries its own dimensions, duration and transcoded
        # variants, and its own URL only once the server has produced one —
        # without this block the tool reports a video as an attachment with no
        # size, no length and nothing to fetch.
        video = attachment.video_processing
        out["video"] = {
            "duration_ms": video.duration_ms,
            "status": video.status.name,
            "thumbnail_url": (
                video.thumbnail_asset_url.url
                if video.thumbnail_asset_url is not None
                else None
            ),
            "variants": [
                {
                    "quality": variant.quality,
                    "width": variant.width,
                    "height": variant.height,
                    "size": variant.size,
                    "url": (
                        variant.asset_url.url if variant.asset_url is not None else None
                    ),
                }
                for variant in video.variants
            ],
        }
    return out


def _message_out(message: Message, author_logins: dict[str, str]) -> dict[str, Any]:
    """A Message as the tool's JSON shape — every field the caller can act on."""
    return {
        "event_id": message.id,
        "room_id": message.room_id,
        "author_user_id": message.actor_id,
        "author_login": author_logins.get(message.actor_id),
        "created_at": _iso_or_none(message.created_at),
        "updated_at": _iso_or_none(message.updated_at),
        "deleted_at": _iso_or_none(message.deleted_at),
        "body": message.body,
        "in_reply_to": message.in_reply_to or None,
        "thread_root_event_id": message.thread_root_event_id or None,
        "pinned": message.pinned,
        "attachments": [
            _attachment_out(attachment) for attachment in message.attachments
        ],
        "reactions": [
            {
                "emoji": reaction.emoji,
                "count": reaction.count,
                "has_reacted": reaction.has_reacted,
            }
            for reaction in message.reactions
        ],
        "link_preview": (
            {"url": message.link_preview.url, "title": message.link_preview.title}
            if message.link_preview is not None
            else None
        ),
        "thread_summary": (
            {
                "thread_root_event_id": message.thread.thread_root_event_id,
                "reply_count": message.thread.reply_count,
                "participant_count": message.thread.participant_count,
            }
            if message.thread is not None
            else None
        ),
    }


async def _get_message(args: dict) -> str:
    """Resolve the room, fetch the message(s). See _handle_chatto_get_message."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    event_id = args.get("event_id") or ""
    event_ids = [e for e in (args.get("event_ids") or []) if e]
    follow_reply_chain = _int_arg(
        args, "follow_reply_chain", 0, low=0, high=FOLLOW_REPLY_CHAIN_MAX
    )

    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")
    if not event_id and not event_ids:
        raise _refuse("Exactly one of event_id or event_ids is required")
    if event_id and event_ids:
        raise _refuse("event_id and event_ids are mutually exclusive")
    if event_ids and follow_reply_chain:
        # A chain belongs to one message. With several requested there is no
        # single answer, and picking one would hand back a conversation the
        # caller did not ask for.
        raise _refuse(
            "follow_reply_chain applies to a single event_id — read one message "
            "per call to follow its chain"
        )

    async with _tool_client(_chatto_config_for_tools()) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        # Every read RPC the plugin uses documents room membership as a
        # precondition (GetRoomEvents, GetMessage, BatchGetMessages), so an
        # unjoined room is unreadable whatever the mention lists say. Saying so
        # here beats a PERMISSION_DENIED that names neither the room nor the fix.
        if not resolved.is_member:
            raise _not_joined_refusal(room, "read it there")

        if event_id:
            # One message is one lookup: the single-message RPC is the
            # authoritative one, so there is no batch to fall back from.
            single = await client.get_message(room.id, event_id)
            found = {single.id: single} if single is not None else {}
            missing = [] if single is not None else [event_id]
        else:
            fetched = await client.batch_get_messages(room.id, event_ids)
            by_id = {m.id: m for m in fetched if m.id}
            # Reported in the order they were asked for, not the order the
            # server happened to answer in.
            found = {eid: by_id[eid] for eid in event_ids if eid in by_id}
            missing = [eid for eid in event_ids if eid not in found]

        if not found:
            raise _refuse(
                f"No message {', '.join(missing)} found in room {room.id}. "
                "Check the event ID and the room."
            )

        # The reply chain walks in_reply_to backwards, oldest ancestor first,
        # one hop per call — a chain is a path, so it cannot be batched. A hop
        # the server cannot resolve ends the chain: the ancestors above it are
        # unreachable, and stopping beats reporting a hole as the top.
        chain: list[Message] = []
        if follow_reply_chain > 0:
            frontier_id = next(iter(found.values())).in_reply_to
            hops = follow_reply_chain
            while frontier_id and hops > 0:
                ancestor = await client.get_message(room.id, frontier_id)
                if ancestor is None:
                    break
                chain.append(ancestor)
                frontier_id = ancestor.in_reply_to
                hops -= 1
            chain.reverse()

        author_ids = [m.actor_id for m in found.values()] + [m.actor_id for m in chain]
        author_logins = await _batch_author_logins(client, room.id, author_ids)

        return json.dumps(
            {
                "success": True,
                "room_id": room.id,
                "room_name": room.name,
                "messages": [_message_out(m, author_logins) for m in found.values()],
                "missing_event_ids": missing,
                "reply_chain": [_message_out(m, author_logins) for m in chain],
            }
        )


async def _get_history(args: dict) -> str:
    """Resolve the room, read the timeline. See _handle_chatto_get_history."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    thread_root_event_id = args.get("thread_root_event_id") or ""
    limit = _int_arg(args, "limit", HISTORY_PAGE_SIZE, low=1, high=HISTORY_PAGE_MAX)
    before = args.get("before") or ""
    after = args.get("after") or ""

    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")
    if before and after:
        # The client sends `before` OR `after` and drops the second without a
        # word, so both would answer a question nobody asked.
        raise _refuse(
            "before and after are mutually exclusive — pick the direction you "
            "want to page in"
        )

    async with _tool_client(_chatto_config_for_tools()) as client:
        resolved = await _resolve_room(client, room_id, room_name)
        room = resolved.room

        # A read of an unjoined room fails server-side whatever the mention
        # lists say; saying so here names the room and the way in.
        if not resolved.is_member:
            raise _not_joined_refusal(room, "read its history there")

        root: Message | None = None
        if thread_root_event_id:
            root = await _resolve_thread_root(client, room, thread_root_event_id)
            page = await client.get_thread_events(
                room.id,
                thread_root_event_id,
                limit=limit,
                before=before or None,
                after=after or None,
            )
        else:
            page = await client.get_room_events(
                room.id, limit=limit, before=before or None, after=after or None
            )

        messages = [ev.message for ev in page.events if ev.message is not None]
        if root is not None and all(m.id != root.id for m in messages):
            # Whether the server lists the root back as the thread's first
            # event is not something the client signature promises, and a
            # thread read without its root is a conversation whose opening
            # line is missing. The root is already in hand, so putting it back
            # costs nothing.
            messages.insert(0, root)
        messages = _oldest_first(messages)

        # The timeline page carries its own user directory (``users_by_id``),
        # which answers author names without a member-list call. Fall back to
        # the room directory for actors the page did not include.
        author_logins = {
            user_id: user.login
            for user_id, user in page.users_by_id.items()
            if user.login
        }
        unresolved = [
            m.actor_id
            for m in messages
            if m.actor_id and m.actor_id not in author_logins
        ]
        if unresolved:
            author_logins.update(
                await _batch_author_logins(client, room.id, unresolved)
            )

        return json.dumps(
            {
                "success": True,
                "room_id": room.id,
                "room_name": room.name,
                "thread_root_event_id": thread_root_event_id or None,
                "count": len(messages),
                "has_older": page.has_older,
                "has_newer": page.has_newer,
                # Paging is the caller's next call, so this answer has to carry
                # the positions that next call takes as its arguments —
                # otherwise the model would have to guess which message to
                # anchor on.
                "start_cursor": page.start_cursor,
                "end_cursor": page.end_cursor,
                "messages": [_message_out(m, author_logins) for m in messages],
            }
        )


def _search_result_out(
    result: MessageSearchResult, author_logins: dict[str, str]
) -> dict[str, Any]:
    """One search hit as the tool's JSON shape.

    A hit is a pointer, not a record: ``chatto_get_message`` is what reads one
    in full, so this carries what decides *which* hit to read — where it is,
    who said it, when, what it says — and leaves the reaction and attachment
    detail to that call.
    """
    message = result.message
    out: dict[str, Any] = {"relevance": round(result.relevance_score, 4)}
    if message is not None:
        out.update(
            {
                "event_id": message.id,
                "room_id": message.room_id,
                "author_user_id": message.actor_id,
                "author_login": author_logins.get(message.actor_id),
                "created_at": _iso_or_none(message.created_at),
                "body": message.body,
                "in_reply_to": message.in_reply_to or None,
                "thread_root_event_id": message.thread_root_event_id or None,
            }
        )

    context = result.thread_context
    if context is not None:
        # Only THREAD grouping asks for this: the row stands for a whole
        # conversation, so the caller needs its opening line and how far it got,
        # not just the message the provider matched.
        root = context.root_message
        latest = context.latest_reply
        thread = context.thread
        out["thread"] = {
            "root_event_id": (
                thread.thread_root_event_id
                if thread is not None
                else (root.id if root is not None else None)
            ),
            "root_body": root.body if root is not None else None,
            "room_id": context.room.id if context.room is not None else None,
            "room_name": context.room.name if context.room is not None else None,
            "reply_count": thread.reply_count if thread is not None else None,
            "participant_count": (
                thread.participant_count if thread is not None else None
            ),
            "last_reply_at": (
                _iso_or_none(thread.last_reply_at) if thread is not None else None
            ),
            "latest_reply_event_id": latest.id if latest is not None else None,
            "latest_reply_body": latest.body if latest is not None else None,
        }
    return out


async def _search_unavailable_reason(
    client: ChattoClient, exc: ChattoConnectError
) -> str:
    """Why search cannot answer, in terms the caller can act on.

    The server reports one of two codes for two very different situations:
    the operator turned search off, or the provider is not ready yet. The first
    never resolves on its own, the second does — saying which one it is changes
    what the caller should do next. ``GetStatus`` stays available while search
    is disabled, so it is the one call that can tell them apart.
    """
    try:
        status = await client.get_search_status()
    except ChattoError as status_exc:
        logger.debug("Chatto tools: search status unreadable: %s", status_exc)
        return f"Message search is unavailable on this Chatto server ({exc.code})"

    hint = _SEARCH_STATE_HINTS.get(status.state)
    if hint is None:
        return f"Message search is unavailable on this Chatto server ({exc.code})"
    reason = f"Message search is unavailable: {hint}"
    if status.retry_after_seconds:
        reason += f" (retry in about {status.retry_after_seconds}s)"
    return reason


async def _search_messages(args: dict) -> str:
    """Search message bodies. See _handle_chatto_search_messages."""
    query = str(args.get("query") or "").strip()
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    author_id = args.get("author_id") or ""
    cursor = args.get("cursor") or ""
    limit = _int_arg(args, "limit", SEARCH_PAGE_SIZE, low=1, high=SEARCH_PAGE_MAX)
    created_after = _timestamp_arg(args, "created_after")
    created_before = _timestamp_arg(args, "created_before")
    order = _enum_arg(args, "order", _SEARCH_ORDERS, MessageSearchOrder.RELEVANCE)
    group_by = _enum_arg(
        args, "group_by", _SEARCH_GROUPS, MessageSearchGroupBy.UNSPECIFIED
    )
    scope = _enum_arg(args, "scope", _SEARCH_SCOPES, MessageSearchScope.UNSPECIFIED)
    has_attachments = args.get("has_attachments")

    if not query:
        raise _refuse("query is required and must not be empty")
    if room_id and room_name:
        raise _refuse("room_id and room_name are mutually exclusive")
    if has_attachments is not None and not isinstance(has_attachments, bool):
        raise _refuse("has_attachments must be true or false")
    if order is MessageSearchOrder.THREAD_ACTIVITY and (
        group_by is not MessageSearchGroupBy.THREAD
    ):
        # The server refuses this combination; saying which two arguments
        # collide is more useful than relaying INVALID_ARGUMENT.
        raise _refuse("order 'thread_activity' requires group_by 'thread'")
    if created_after and created_before and created_after >= created_before:
        raise _refuse("created_after must be earlier than created_before")

    async with _tool_client(_chatto_config_for_tools()) as client:
        if room_id or room_name:
            room_id = (await _resolve_room(client, room_id, room_name)).room.id

        try:
            page = await client.search_messages(
                query,
                room_id=room_id or None,
                author_id=author_id or None,
                created_after=created_after,
                created_before=created_before,
                has_attachments=has_attachments,
                order=order,
                page_size=limit,
                cursor=cursor or None,
                scope=scope,
                group_by=group_by,
            )
        except ChattoConnectError as exc:
            if exc.code not in {"failed_precondition", "unavailable"}:
                raise
            raise _refuse(await _search_unavailable_reason(client, exc)) from exc

        results = page.results

        # The page carries its own user directory, which answers author names
        # without a member-list call. A cross-room search can still miss an
        # author, so the remaining ones are resolved per room — the directory
        # call is per room, and a hit names the room it came from.
        author_logins = {
            user_id: user.login
            for user_id, user in page.users_by_id.items()
            if user.login
        }
        unresolved_by_room: dict[str, list[str]] = {}
        for result in results:
            message = result.message
            if message is None or not message.actor_id:
                continue
            if message.actor_id in author_logins:
                continue
            unresolved_by_room.setdefault(message.room_id, []).append(message.actor_id)
        for hit_room_id, actor_ids in unresolved_by_room.items():
            if hit_room_id:
                author_logins.update(
                    await _batch_author_logins(client, hit_room_id, actor_ids)
                )

        return json.dumps(
            {
                "success": True,
                "query": query,
                "room_id": room_id or None,
                "count": len(results),
                "grouped_by_thread": group_by is MessageSearchGroupBy.THREAD,
                "thread_total_count": page.thread_total_count,
                "next_cursor": page.next_cursor or None,
                "results": [
                    _search_result_out(result, author_logins) for result in results
                ],
            }
        )


async def _handle_chatto_search_messages(args: dict) -> str:
    """Tool handler for chatto_search_messages.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _search_messages(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_send_dm(args: dict) -> str:
    """Tool handler for chatto_send_dm.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _send_dm(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_send_room(args: dict) -> str:
    """Tool handler for chatto_send_room.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _send_room(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_send_ephemeral(args: dict) -> str:
    """Tool handler for chatto_send_ephemeral.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _send_ephemeral(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_list_rooms(args: dict) -> str:
    """Tool handler for chatto_list_rooms.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type".
    """
    try:
        return await _list_rooms(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_set_custom_status(args: dict) -> str:
    """Tool handler for chatto_set_custom_status.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _set_custom_status(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_delete_custom_status(args: dict) -> str:
    """Tool handler for chatto_delete_custom_status.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _delete_custom_status(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _set_custom_status(args: dict) -> str:
    """Set the bot's custom status. See _handle_chatto_set_custom_status."""
    emoji = (args.get("emoji") or "").strip()
    text = (args.get("text") or "").strip()
    expires_in = args.get("expires_in_seconds")
    if expires_in is not None:
        expires_in = int(expires_in)
        if expires_in < 0:
            raise _refuse("expires_in_seconds must be non-negative")

    if not emoji or not text:
        raise _refuse("emoji and text are required")

    async with _tool_client(_chatto_config_for_tools()) as client:
        expires_at = None
        if expires_in and expires_in > 0:
            from datetime import UTC, datetime, timedelta

            expires_at = datetime.now(UTC) + timedelta(seconds=expires_in)
        await client.set_custom_status(emoji, text, expires_at=expires_at)
        return json.dumps(
            {
                "success": True,
                "emoji": emoji,
                "text": text,
                "expires_at": (_iso_or_none(expires_at) if expires_at else None),
            }
        )


async def _delete_custom_status(args: dict) -> str:
    """Clear the bot's custom status. See _handle_chatto_delete_custom_status."""
    async with _tool_client(_chatto_config_for_tools()) as client:
        await client.delete_custom_status()
        return json.dumps({"success": True, "message": "Custom status cleared"})


async def _fetch_link_preview(args: dict) -> str:
    """Fetch an OpenGraph link preview. See _handle_chatto_fetch_link_preview."""
    url = (args.get("url") or "").strip()
    if not url:
        raise _refuse("url is required")

    async with _tool_client(_chatto_config_for_tools()) as client:
        preview, _ = await client.fetch_link_preview(url)
        return json.dumps(
            {
                "success": True,
                "url": url,
                "title": preview.title if preview else None,
                "description": preview.description if preview else None,
                "image_url": preview.image_url if preview else None,
                "site_name": preview.site_name if preview else None,
                "embed_type": preview.embed_type if preview else None,
            }
        )


async def _mark_thread_as_read(args: dict) -> str:
    """Mark a thread as read. See _handle_chatto_mark_thread_as_read."""
    room_id = args.get("room_id") or ""
    room_name = args.get("room_name") or ""
    thread_root_event_id = args.get("thread_root_event_id") or ""

    if not room_id and not room_name:
        raise _refuse("Exactly one of room_id or room_name is required")
    if not thread_root_event_id:
        raise _refuse("thread_root_event_id is required")

    async with _tool_client(_chatto_config_for_tools()) as client:
        if not room_id:
            resolved = await _resolve_room(client, room_id, room_name)
            room_id = resolved.room.id
        prev_last_read = await client.mark_thread_as_read(room_id, thread_root_event_id)
        return json.dumps(
            {
                "success": True,
                "room_id": room_id,
                "thread_root_event_id": thread_root_event_id,
                "previous_last_read_at": (
                    _iso_or_none(prev_last_read) if prev_last_read else None
                ),
            }
        )


async def _list_room_groups(args: dict) -> str:
    """List room groups. See _handle_chatto_list_room_groups."""
    async with _tool_client(_chatto_config_for_tools()) as client:
        groups = await client.list_room_groups()
        results: list[dict] = []
        for g in groups:
            results.append(
                {
                    "id": g.id,
                    "name": g.name,
                    "description": g.description or "",
                    "room_count": len(g.rooms),
                    "room_ids": [rws.room.id for rws in g.rooms if rws.room],
                    "sidebar_links": [
                        {"id": l.id, "label": l.label, "url": l.url}
                        for l in g.sidebar_links
                    ],
                }
            )
        return json.dumps({"success": True, "groups": results})


async def _handle_chatto_fetch_link_preview(args: dict) -> str:
    """Tool handler for chatto_fetch_link_preview.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _fetch_link_preview(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_mark_thread_as_read(args: dict) -> str:
    """Tool handler for chatto_mark_thread_as_read.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _mark_thread_as_read(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_get_message(args: dict) -> str:
    """Tool handler for chatto_get_message.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _get_message(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_get_history(args: dict) -> str:
    """Tool handler for chatto_get_history.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _get_history(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_list_room_groups(args: dict) -> str:
    """Tool handler for chatto_list_room_groups.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _list_room_groups(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_list_room_attachments(args: dict) -> str:
    """Tool handler for chatto_list_room_attachments.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _list_room_attachments(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_list_pinned_messages(args: dict) -> str:
    """Tool handler for chatto_list_pinned_messages.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _list_pinned_messages(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


async def _handle_chatto_pin_message(args: dict) -> str:
    """Tool handler for chatto_pin_message.

    Returns a JSON string: Hermes' tool contract accepts no other type, and a
    dict (or an unflagged coroutine) fails at dispatch with "unsupported result
    type". Every early exit arrives here as a _ToolRefusal and leaves as that
    refusal's own JSON, so no path escapes as an unhandled error.
    """
    try:
        return await _pin_message(args)
    except _ToolRefusal as refusal:
        return refusal.result_json
    except ChattoAuthError as exc:
        return json.dumps({"success": False, "error": f"Authentication error: {exc}"})
    except ChattoError as exc:
        return json.dumps({"success": False, "error": f"Chatto API error: {exc}"})


def _register_bundled_skills(ctx) -> None:
    """Register the skills shipped in ``skills/``, each with its own description.

    ``register_skill`` stores the description verbatim and never reads the
    SKILL.md frontmatter, so passing none leaves the skill listed in
    ``<available_skills>`` with nothing explaining when to load it — which is
    the only place an agent learns it exists.
    """
    from pathlib import Path

    from agent.skill_utils import parse_frontmatter

    skills_dir = Path(__file__).parent / "skills"
    if not skills_dir.is_dir():
        logger.debug("Chatto: no bundled skills directory at %s", skills_dir)
        return
    for child in sorted(skills_dir.iterdir()):
        skill_md = child / "SKILL.md"
        if not (child.is_dir() and skill_md.exists()):
            continue
        frontmatter, _ = parse_frontmatter(skill_md.read_text(encoding="utf-8"))
        ctx.register_skill(
            child.name,
            str(skill_md),
            description=str(frontmatter.get("description") or ""),
            frontmatter=frontmatter,
        )
        logger.info("Chatto plugin skill registered: %s", child.name)


# The plugin's outbound tools: (tool name, JSON schema, handler). One list, so
# the two registration entry points below cannot disagree about what exists.
_PLUGIN_TOOLS = (
    ("chatto_send_dm", CHATTO_SEND_DM_SCHEMA, _handle_chatto_send_dm),
    ("chatto_send_room", CHATTO_SEND_ROOM_SCHEMA, _handle_chatto_send_room),
    (
        "chatto_send_ephemeral",
        CHATTO_SEND_EPHEMERAL_SCHEMA,
        _handle_chatto_send_ephemeral,
    ),
    ("chatto_list_rooms", CHATTO_LIST_ROOMS_SCHEMA, _handle_chatto_list_rooms),
    ("chatto_get_message", CHATTO_GET_MESSAGE_SCHEMA, _handle_chatto_get_message),
    ("chatto_get_history", CHATTO_GET_HISTORY_SCHEMA, _handle_chatto_get_history),
    (
        "chatto_search_messages",
        CHATTO_SEARCH_MESSAGES_SCHEMA,
        _handle_chatto_search_messages,
    ),
    (
        "chatto_list_room_attachments",
        CHATTO_LIST_ROOM_ATTACHMENTS_SCHEMA,
        _handle_chatto_list_room_attachments,
    ),
    (
        "chatto_list_pinned_messages",
        CHATTO_LIST_PINNED_MESSAGES_SCHEMA,
        _handle_chatto_list_pinned_messages,
    ),
    (
        "chatto_pin_message",
        CHATTO_PIN_MESSAGE_SCHEMA,
        _handle_chatto_pin_message,
    ),
    (
        "chatto_set_custom_status",
        CHATTO_SET_CUSTOM_STATUS_SCHEMA,
        _handle_chatto_set_custom_status,
    ),
    (
        "chatto_delete_custom_status",
        CHATTO_DELETE_CUSTOM_STATUS_SCHEMA,
        _handle_chatto_delete_custom_status,
    ),
    (
        "chatto_fetch_link_preview",
        CHATTO_FETCH_LINK_PREVIEW_SCHEMA,
        _handle_chatto_fetch_link_preview,
    ),
    (
        "chatto_mark_thread_as_read",
        CHATTO_MARK_THREAD_AS_READ_SCHEMA,
        _handle_chatto_mark_thread_as_read,
    ),
    (
        "chatto_list_room_groups",
        CHATTO_LIST_ROOM_GROUPS_SCHEMA,
        _handle_chatto_list_room_groups,
    ),
)


def register_plugin_tools(ctx) -> None:
    """Register the plugin's outbound tools in the ``chatto-platform`` toolset.

    Reached through two doors: ``register()`` in a gateway/web process, which
    materializes the platform, and ``tools.register_tools`` in a CLI/TUI one,
    which never does. A tool list written out twice would mean tools that exist
    in only half the sessions, so both doors call this.
    """
    for name, schema, handler in _PLUGIN_TOOLS:
        ctx.register_tool(
            name=name,
            toolset="chatto-platform",
            schema=schema,
            handler=handler,
            is_async=True,
            description=schema["description"],
        )


def register(ctx) -> None:
    """Plugin entry point — called by the Hermes plugin system."""
    logger.info("Registering Chatto platform plugin on Hermes Agent")

    # --- Plugin tools ---
    register_plugin_tools(ctx)

    # --- Bundled skills ---
    # Ship the plugin's own usage/development knowledge with the code, so any
    # agent using this plugin knows when/how to use its tools and how to
    # extend it. Loaded via skill_view("chatto-platform:chatto-platform-plugin").
    _register_bundled_skills(ctx)

    for capability in _capabilities():
        logger.info("Chatto capability: %s", capability)

    ctx.register_platform(
        name=ChattoConstants.PLATFORM_NAME,  # this will be the config.yaml key.
        label=ChattoConstants.PLATFORM_LABEL,
        adapter_factory=hermes_adapter_factory,
        check_fn=hermes_check_fn,
        validate_config=hermes_validate_config,
        is_connected=hermes_is_connected,
        install_hint=ChattoConstants.INSTALL_HINT,
        env_enablement_fn=hermes_env_enablement_fn,
        setup_fn=hermes_setup_fn,
        cron_deliver_env_var=ChattoConfiguration.home_channel.env_name,
        standalone_sender_fn=hermes_standalone_sender_fn,
        allowed_users_env=ChattoConfiguration.allowed_users.env_name,
        allow_all_env=ChattoConfiguration.allow_all_users.env_name,
        max_message_length=ChattoConstants.MAX_MESSAGE_LENGTH,
        emoji="😺",
        allow_update_command=True,
        pii_safe=False,
        platform_hint=(
            "Using the 'Hermes Chatto Platform Plugin' you connect to a Chatto server. "
            "Authorized admins and users will contact you and call you his Hermes Agent. They are "
            "natural persons and thus responsible for what they do in terms of rights. You compute "
            "on their behalf. They _may_ address you by @-mentioning your name. If configured, "
            "you also react without a @-mention. Direct messages reach you without a mention."
            "Keep responses conversational. Markdown is supported. "
            "Include MEDIA:/absolute/path/to/file in your response to refer to our local files. Images "
            "(.png, .jpg, .gif, .webp) arrive as inline pictures, videos (.mp4, .mov, .webm) as "
            "video attachments, audio as a voice bubble, anything else as a downloadable document. "
            "Do NOT use markdown image syntax for local files. Local files always go through MEDIA:. "
            "Several images in one response are bundled into a single message. "
            "You can also talk to Chatto unprompted: the plugin tools chatto_list_rooms, "
            "chatto_send_room and chatto_send_dm post to a channel or a person's DM on your "
            "behalf. Only send into a channel that the bot answers in, and reach a person by DM "
            "only once they have written to the bot themselves. "
            "chatto_get_history and chatto_get_message read instead of posting: use them to "
            "look up what was said in a room or thread, and to resolve an event ID you were given. "
            "chatto_search_messages finds messages by their words when you do not know the room, "
            "thread or day — it only answers when the server has search enabled, and says so when "
            "it has not. "
        ),
    )