"""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 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 from collections import deque from datetime import UTC, datetime from difflib import SequenceMatcher from enum import StrEnum from typing import Any, cast from urllib.parse import unquote, urlsplit import httpx logger = logging.getLogger(__name__) from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( BasePlatformAdapter, MessageEvent, MessageType, ProcessingOutcome, SendResult, cache_media_bytes, get_inbound_media_max_bytes, validate_inbound_media_size, ) from gateway.session import build_session_key # 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.client import ( ChattoClient, ) from chattolib.exceptions import ( ChattoAuthError, ChattoError, ) from chattolib.realtime import ( ChattoRealtimeCloseError, ChattoRealtimeError, RealtimeEvent, stream_events, ) from chattolib.realtime_types import ( MessageEditedPayload, MessagePostedPayload, ReactionPayload, ) from chattolib.types import ( Message, MessageAttachment, PresenceStatus, Room, RoomKind, RoomWithViewerState, User, ) 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 # --------------------------------------------------------------------------- # # Adapter # --------------------------------------------------------------------------- # 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] = {} # 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) # 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] = {} # Handle -> does a user hold it. Cached both ways; see _mentions_someone_else. self._known_handles: dict[str, bool] = {} # 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, login=self.chatto_config.login.value, password=self.chatto_config.password.value, token=self.chatto_config.token.value, ) self._chatto_client = client logger.info( "Chatto: logged in as '%s' via chattolib", self.chatto_config.login.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, login: str, password: str, token: str | None = None, ) -> ChattoClient: """Return a connected ``ChattoClient`` using token or login/password. Token wins when both are configured. Raises ValueError when neither a token nor login+password is set — caught upstream as a normal creation failure, so misconfiguration reads as a logged error instead of a crash. """ if token: return ChattoClient(token=token, base_url=base_url) if not login or not password: raise ValueError("Chatto: neither token nor login/password configured") return await ChattoClient.login(login, password, 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 # 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 and authentiated to %s using login:'%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.update_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 # 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_me(self, body: str) -> bool: """Whether the message addresses this bot. By login, by display name, or by a broadcast handle — ``@here`` speaks to everyone present and the bot is one of them, so naming a colleague alongside it does not take the bot out of the audience. Matching is case-insensitive, like every mention resolution in Chatto. """ own = self._own_handles() for handle in self._mention_candidates(body): lowered = handle.lower() if lowered in own or lowered in ChattoConstants.BROADCAST_MENTIONS: return True return False async def _handle_belongs_to_a_user(self, handle: str) -> bool: """Whether ``handle`` is the login of a real Chatto user. The API carries no mention entities — ``mention_confirmation_token`` is reserved in the message descriptor — so an @-token is only a candidate until the directory confirms it. Results are cached both ways under the lowercased handle (the mention namespace is case-insensitive per FDR-006), since the same handles recur and a miss is as reusable as a hit. Only logins are looked up: matching another user's display name, as the web frontend does against its room member list, has no directory equivalent here. """ cache_key = handle.lower() known = self._known_handles.get(cache_key) if known is not None: return known client = await self._get_chatto_client() if client is None: # Unresolved means "not confirmed", so the message goes through. return False try: member = await client.get_user(login=handle) except Exception as exc: logger.debug("Chatto: could not resolve handle @%s: %s", handle, exc) return False exists = member is not None and member.user is not None self._known_handles[cache_key] = exists return exists async def _mentions_someone_else(self, body: str) -> bool: """Whether the message @-mentions a person who is not this bot. Broadcast handles are not a person — they address everyone present, the bot included, so they do not count as someone else. A handle no user holds is not a mention at all: 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() for handle in self._mention_candidates(body): lowered = handle.lower() if lowered in ChattoConstants.BROADCAST_MENTIONS: continue if lowered in own: continue if await self._handle_belongs_to_a_user(handle): return True return False 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 login and id, ``CHATTO_ALLOW_ALL_USERS`` overrides both — the gate stays chat_type-independent by design. """ if self.chatto_config.allow_all_users.value: return True if user.login in self.chatto_config.allowed_users.value: return True if user.id in self.chatto_config.allowed_users.value: 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 | /leave ", ) 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) 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. """ 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. Chatto knows only DMs and channels, so the split is ``kind == DM``: a DM always answers (``/join`` must stay reachable), every other room — channel-kind or an unknown kind, which is how servers that never set ``kind`` show up — opts in through the mention lists. A room whose policy is SILENT stays read-only (marked as read, never seeded or answered). """ if self._room_kinds.get(room_id) == RoomKind.DM: return True return self._room_policy(room_id) != RoomPolicy.SILENT async def _dispatch_message_posted(self, payload: MessagePostedPayload) -> 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 payload.fetch_message(client=client) 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=payload.message_event_id, thread_root_event_id=payload.thread_root_event_id or None, ) 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. Built with exactly the inputs ``handle_message`` uses, so lookups in ``_processing`` and calls to ``cancel_session_processing`` hit the same session the gateway is running. """ return build_session_key( source, group_sessions_per_user=self.config.extra.get( "group_sessions_per_user", True ), thread_sessions_per_user=self.config.extra.get( "thread_sessions_per_user", False ), ) async def _dispatch_message_edited(self, payload: MessageEditedPayload) -> 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 payload.fetch_message(client=client) 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, 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. The caller owns dispatching: a non-None result still needs ``handle_message()``. """ user = self._user_cache.get(message.actor_id) if user is None: directory_member = await client.get_user(user_id=message.actor_id) if directory_member is None or directory_member.user is None: return None user = directory_member.user self._user_cache[user.id] = user if not self._check_auth(user): return None room_kind = await self._room_kind_for(client, message.room_id) if room_kind is None: 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). The # shared _mentions_me gate keeps this path and the someone-else check # below on one definition of "addressed", broadcast handles included. if room_kind != RoomKind.DM: policy = self._room_policy(room_id) if policy == RoomPolicy.REQUIRE_MENTION and not self._mentions_me( message_body ): 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 self._mentions_me(message_body) and await self._mentions_someone_else(message_body) ): 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 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, ) # 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, ) 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: ReactionPayload, *, 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 = getattr(self, "_reaction_handler", None) 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) if (event_payload := event.get("message_posted")) is not None: # Self-event filter — the actor_id on the envelope is authoritative # (chattolib does NOT filter this itself; see chatto-bridge notes). actor_id = event.actor_id if actor_id and self.me and actor_id == self.me.id: return await self._dispatch_message_posted(event_payload) elif (edited_payload := event.get("message_edited")) is not None: # Same self-event filter: our own streaming edits echo back here. # Redeliveries of edits we made are also caught by _mark_seen in # edit_message(). actor_id = event.actor_id if actor_id and self.me and actor_id == self.me.id: return await self._dispatch_message_edited( cast(MessageEditedPayload, edited_payload) ) elif event.kind in ("reaction_added", "reaction_removed"): reaction_payload = event.get(event.kind) if reaction_payload is not None: await self._forward_reaction( event, cast(ReactionPayload, reaction_payload), removed=event.kind == "reaction_removed", ) # confirmed: elif event.kind in ( "mention_notification", "projection_event", "presence_changed", "notification_dismissed", "room_marked_as_read", "user_typing", "notification_created", "new_direct_message_notification", "message_retracted", "user_joined_room", "user_left_room", "thread_created", "thread_follow_changed", "room_updated", "room_groups_updated", ): 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): if self._closing: return await self._handle_realtime_event(event) # 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: 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: members, _page = await client.list_room_members(room_id) 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.""" client = await self._get_chatto_client() if client is None: logger.warning("Chatto WS: _refresh_rooms aborted - no client available") return 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) 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 & notification dismissal (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 ) # Dismissal is server-global — once after the per-room sweep. try: await client.dismiss_all_notifications() except Exception: logger.debug("Chatto: dismiss_all_notifications failed", 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") formatted = _normalise_outbound_text(content) chunks = self.truncate_message(formatted, ChattoConstants.SPLIT_THRESHOLD) 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 ) message_ids: list[str] = [] last_error: str | None = None retryable = False client = await self._get_chatto_client() if client is None: return SendResult( success=False, error="Chatto client not available", retryable=True ) for i, chunk in enumerate(chunks): try: msg_obj = await client.post_message( room_id=chat_id, body=chunk, thread_root_event_id=str(thread_id) if thread_id else "", ) except Exception as e: # ChattoError included: both read as "this chunk did not go # out" and stop the batch — the SendResult carries the reason. # Only server-side failures count as retryable; a bug in our # own code must not read as a transient network blip. last_error = str(e) retryable = isinstance(e, ChattoError) break self._mark_seen(msg_obj.id) message_ids.append(msg_obj.id) # Auto-thread: first chunk becomes the thread root, # subsequent chunks go in the thread if use_auto_thread and i == 0 and not thread_id: thread_id = msg_obj.id # Nothing got through at all — report the failure instead of a phantom success. if not message_ids: return SendResult( success=False, error=last_error or "Chatto: message could not be sent", retryable=retryable, ) 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 last_error: logger.warning( "Chatto: sent %d/%d chunk(s) to %s before failing: %s", len(message_ids), len(chunks), chat_id, last_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 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.update_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 # ------------------------------------------------------------------ # # DM initiation (Chatto-unique) # ------------------------------------------------------------------ # async def start_dm(self, user_id: str) -> str | None: """Start a direct message with a user via RoomService/StartDM. Returns the room ID on success, or None on failure. BasePlatformAdapter override """ if not user_id: return None client = await self._get_chatto_client() if client is None: return None try: room = await client.start_dm(participant_ids=[str(user_id)]) self._room_names[room.id] = room.name self._room_kinds[room.id] = room.kind return room.id except ChattoError as e: logger.debug("Chatto: StartDM failed: %s", e) return None except Exception as e: logger.debug("Chatto: StartDM error: %s", e) return None # ------------------------------------------------------------------ # # 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, ) -> None: """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: await super().send_multiple_images( chat_id, images, metadata=metadata, human_delay=human_delay, ) return 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", ) await super().send_multiple_images( chat_id, images, metadata=metadata, human_delay=human_delay, ) return 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), ) 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 — we need a base URL plus # either a token or a full login/password pair. has_credentials = bool( chatto_config.token.value or (chatto_config.login.value and chatto_config.password.value), ) if not chatto_config.base_url.value or not has_credentials: return SendResult( success=False, error="Chatto: base URL or credentials missing" ) client: ChattoClient try: if chatto_config.token.value: client = ChattoClient( base_url=chatto_config.base_url.value, token=chatto_config.token.value ) else: client = await ChattoClient.login( base_url=chatto_config.base_url.value, login=chatto_config.login.value, password=chatto_config.password.value, ) except Exception as exc: return SendResult(success=False, error=f"Chatto login failed: {exc}") 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, ) # 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(). """ chatto_config = ChattoConfiguration(pconfig=config) _warn_on_unknown_extra_keys(config) if ( len(chatto_config.allowed_users.value) > 0 and chatto_config.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(chatto_config.require_mention_rooms.value) optional_rooms = set(chatto_config.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 # base_url always resolves (it defaults to ChattoHQ), so the only real # question is whether any credentials came in. if (chatto_config.token.value is not None) or ( chatto_config.login.value and chatto_config.password.value ): return True logger.error("Chatto: Minimally, either token or login/password must be set.") return False 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) login = prompt("Chatto login (username):") if login: save_env_value(ChattoConfiguration.login.env_name, login) password = prompt("Chatto password:", password=True) if password: save_env_value(ChattoConfiguration.password.env_name, password) 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 field in ChattoConfiguration.fields(): env_value = os.getenv(field.env_name) if env_value: seed[field.config_key] = env_value.strip() logger.debug( "seed: %s", {k: v for k, v in seed.items() if k not in ("token", "password")} ) 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", "start_dm": "direct messages", "create_room": "room creation", } 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 def register(ctx) -> None: """Plugin entry point — called by the Hermes plugin system.""" logger.info("Registering Chatto platform plugin on Hermes Agent") 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. " ), )