""" 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. Configuration in config.yaml:: gateway: platforms: chatto: enabled: true extra: url: https://chat.lacy.casa channels: # room IDs to watch (empty = all joined) - REljMv5Pgolo6Y9 home_channel: REljMv5Pgolo6Y9 require_mention: true # only respond to @mentions in rooms allowed_users: [] # empty = allow all allow_all_users: true Or via environment variables (overrides config.yaml): CHATTO_URL, CHATTO_LOGIN, CHATTO_PASSWORD (secrets in ~/.hermes/.env), CHATTO_CHANNELS, CHATTO_HOME_CHANNEL, CHATTO_REQUIRE_MENTION, CHATTO_ALLOWED_USERS, CHATTO_ALLOW_ALL_USERS """ from __future__ import annotations import asyncio import hashlib import logging import mimetypes import os from collections import OrderedDict from datetime import datetime, timezone from typing import Any, Dict, List, Optional from urllib.parse import urlsplit, urlunsplit logger = logging.getLogger(__name__) from gateway.platforms.base import ( BasePlatformAdapter, SendResult, MessageEvent, MessageType, ProcessingOutcome, ) from gateway.config import Platform # Chattolib imports from chattolib import ( ChattoClient, ChattoError, ChattoAuthError, ChattoConnectError, ChattoRealtimeError, ChattoRealtimeCloseError, RealtimeConnection, RealtimeEvent, ServerHello, stream_events, ) from chattolib.types import ( RoomKind, PresenceStatus, RoomWithViewerState, User, Message, ) # --------------------------------------------------------------------------- # # Constants # --------------------------------------------------------------------------- # _MAX_MESSAGE_LENGTH = 10000 _SEEN_CAP = 500 # WebSocket / realtime protocol _WS_PATH = "/api/realtime" _WS_AUTH_TIMEOUT = 20.0 _WS_MAX_MESSAGE_BYTES = 4_000_000 _WS_PING_INTERVAL = 30.0 _WS_RECONNECT_INITIAL_BACKOFF = 1.0 _WS_RECONNECT_MAX_BACKOFF = 30.0 # Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji) # Emoji shortcode mapping (Chatto uses shortcode names, not unicode emoji) _EMOJI_TO_SHORTCODE: Dict[str, str] = { "šŸ‘": "thumbsup", "šŸ‘Ž": "thumbsdown", "ā¤ļø": "heart", "ā¤": "heart", "āœ…": "white_check_mark", "āŒ": "x", "šŸ‘€": "eyes", "šŸŽ‰": "tada", "šŸ˜‚": "joy", "šŸš€": "rocket", "šŸ”„": "fire", "šŸ’Æ": "100", "šŸ¤”": "thinking", "šŸ‘": "clap", "šŸ™": "pray", "šŸ˜…": "sweat_smile", "šŸ˜“": "sleeping", "ā³": "hourglass", } # Chunk size for asset uploads (256 KB) _UPLOAD_CHUNK_SIZE = 256 * 1024 # --------------------------------------------------------------------------- # # Adapter # --------------------------------------------------------------------------- # class ChattoAdapter(BasePlatformAdapter): """Chatto platform adapter ā€” receives messages via WebSocket realtime, sends via ConnectRPC REST.""" MAX_MESSAGE_LENGTH = 10000 _SPLIT_THRESHOLD = 9900 splits_long_messages = True def __init__(self, config, **kwargs): platform = Platform("chatto") super().__init__(config=config, platform=platform) extra = getattr(config, "extra", {}) or {} # --- Configuration (env > config.yaml extra) --- self._base_url = ( os.getenv("CHATTO_URL", "").strip() or str(extra.get("url", "")).strip() ) self._login = os.getenv("CHATTO_LOGIN", "").strip() self._password = os.getenv("CHATTO_PASSWORD", "").strip() raw_channels = os.getenv("CHATTO_CHANNELS", "").strip() if raw_channels: self._channel_ids = [c.strip() for c in raw_channels.split(",") if c.strip()] elif isinstance(extra.get("channels"), list): self._channel_ids = [str(c) for c in extra["channels"]] else: self._channel_ids = [] self._home_channel = ( os.getenv("CHATTO_HOME_CHANNEL", "").strip() or str(extra.get("home_channel", "")).strip() ) self._require_mention = os.getenv("CHATTO_REQUIRE_MENTION", "").strip().lower() if self._require_mention: self._require_mention = self._require_mention in ("true", "1", "yes") else: self._require_mention = bool(extra.get("require_mention", True)) # free_response_channels: room IDs where the bot responds without being tagged fr_env = os.getenv("CHATTO_FREE_RESPONSE_CHANNELS", "").strip() if fr_env: self._free_response_channels = set(c.strip() for c in fr_env.split(",") if c.strip()) else: self._free_response_channels = set( str(c) for c in extra.get("free_response_channels", []) if str(c).strip() ) # --- Chattolib client --- self._chatto_client: Optional[ChattoClient] = None # --- Runtime state --- self._token: Optional[str] = None self._user_id: str = "" self._user_login: str = "" self._user_display: str = "" self._room_names: Dict[str, str] = {} self._room_kinds: Dict[str, str] = {} self._our_thread_roots: set = set() # thread root event IDs we created self._our_message_ids: set = set() # message IDs we sent (for thread root detection) self._seen: Dict[str, OrderedDict] = {} # room_id -> OrderedDict(event_id -> None) self._resume_cursor: Optional[str] = None self._watch_room_ids: List[str] = [] self._ws_task: Optional[asyncio.Task] = None self._ws_ready: Optional[asyncio.Event] = None self._ws_active = False self._ws_ref = None # reference to open websocket for dynamic resubscribe # Persistent typing indicator loops per room self._typing_tasks: Dict[str, asyncio.Task] = {} # Liveness probe (REST health check) self._liveness_interval_seconds = 60.0 self._liveness_failure_threshold = 3 self._liveness_task: Optional[asyncio.Task] = None # Member directory cache: user_id -> user info dict self._user_cache: Dict[str, dict] = {} # ------------------------------------------------------------------ # # Auth # ------------------------------------------------------------------ # async def _get_chatto_client(self) -> Optional[ChattoClient]: """Get or create a ChattoClient instance.""" if self._chatto_client is not None: return self._chatto_client if not self._base_url or not self._login or not self._password: logger.error("Chatto: missing configuration (URL, login, or password)") return None try: self._chatto_client = await ChattoClient.login( self._login, self._password, base_url=self._base_url, ) logger.info("Chatto: logged in as %s via chattolib", self._login) return self._chatto_client except ChattoAuthError as e: logger.error("Chatto: authentication failed: %s", e) self._set_fatal_error("auth_failed", str(e), retryable=True) return None except Exception as e: logger.error("Chatto: failed to create client: %s", e) self._set_fatal_error("client_error", str(e), retryable=True) return None async def _ensure_token(self) -> bool: """Login via chattolib.""" if self._token: return True client = await self._get_chatto_client() if client is not None: self._token = client.token return True return False async def _relogin(self) -> bool: """Force re-login (token expired).""" self._token = None self._chatto_client = None # Also clear chattolib client return await self._ensure_token() # ------------------------------------------------------------------ # # Connection # ------------------------------------------------------------------ # async def connect(self, *, is_reconnect: bool = False) -> bool: """Login, discover rooms, start WebSocket realtime connection.""" if not await self._ensure_token(): return False client = self._chatto_client # Get our own user info try: me = await client.me() self._user_id = str(me.id) self._user_login = str(me.login) self._user_display = str(me.display_name or "") logger.info("Chatto: got user info: %s", self._user_login) except Exception as e: logger.error("Chatto: failed to get user info: %s", e) self._set_fatal_error("connect_failed", str(e), retryable=True) return False # Discover rooms try: rooms_list = await client.list_rooms() rooms = [] for room_with_state in rooms_list: entry = { "room": { "id": str(room_with_state.room.id), "name": str(room_with_state.room.name), "kind": str(room_with_state.room.kind.value) if room_with_state.room.kind else "", }, "viewerState": { "isMember": room_with_state.viewer_state.is_member if room_with_state.viewer_state else False, } } rooms.append(entry) logger.info("Chatto: got %d rooms", len(rooms)) except Exception as e: logger.error("Chatto: failed to list rooms: %s", e) self._set_fatal_error("connect_failed", str(e), retryable=True) return False all_room_ids = [] for entry in rooms: room = entry.get("room", {}) rid = str(room.get("id", "")) if not rid: continue name = str(room.get("name", rid)) kind = str(room.get("kind", "")) self._room_names[rid] = name self._room_kinds[rid] = kind viewer = entry.get("viewerState", {}) is_member = viewer.get("isMember", False) # If user-specified channels, only watch those; otherwise watch all joined rooms if self._channel_ids: if rid in self._channel_ids and not is_member: await self._join_room(rid) all_room_ids.append(rid) elif is_member: all_room_ids.append(rid) if self._channel_ids: watch = list(self._channel_ids) else: watch = all_room_ids if not watch: logger.error("Chatto: no rooms to watch (join a room or set CHATTO_CHANNELS)") self._set_fatal_error("config_missing", "no Chatto rooms to watch", retryable=False) return False # Ensure we're a member of each watched room for rid in watch: if self._room_kinds.get(rid) != "ROOM_KIND_DM": await self._join_room(rid) # Pick home channel if not self._home_channel: self._home_channel = watch[0] self._watch_room_ids = watch # Initialize seen for each room ā€” seed from REST to avoid replaying history for rid in watch: self._seen[rid] = OrderedDict() await self._seed_room(rid) # Start WebSocket realtime connection if not await self._start_chattolib_realtime(): self._set_fatal_error( "ws_connect_failed", "Chatto WebSocket realtime connection failed", retryable=True, ) return False self._mark_connected() self._start_liveness_probe() logger.info( "Chatto: connected to %s as %s, watching %d room(s) via WebSocket", self._base_url, self._user_display or self._user_login, len(watch), ) # Broadcast online presence so the bot appears online in the member list try: await self.set_presence("online") except Exception: logger.debug("Chatto: set_presence(online) failed on connect", exc_info=True) return True async def disconnect(self) -> None: """Stop WebSocket, liveness probe, typing tasks, and clear state.""" # Broadcast away presence before tearing down try: await self.set_presence("away") except Exception: logger.debug("Chatto: set_presence(away) failed on disconnect", exc_info=True) self._mark_disconnected() self._ws_active = False # Cancel liveness probe await self._cancel_liveness_task() # 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): pass self._ws_task = None self._token = None # ------------------------------------------------------------------ # # Liveness probe # ------------------------------------------------------------------ # def _start_liveness_probe(self) -> None: """Start the periodic REST health probe.""" if ( self._liveness_interval_seconds <= 0 or self._liveness_failure_threshold <= 0 ): return if self._liveness_task and not self._liveness_task.done(): return self._liveness_task = asyncio.create_task(self._liveness_loop()) async def _cancel_liveness_task(self) -> None: """Cancel the liveness probe task.""" task = self._liveness_task self._liveness_task = None if task and not task.done(): task.cancel() try: await task except (asyncio.CancelledError, Exception): pass async def _liveness_loop(self) -> None: """Periodically check if the REST API is alive via ViewerService/GetViewer. Also refreshes presence status on each successful probe so the bot stays showing as online ā€” Chatto's presence expires if not refreshed. On ``threshold`` consecutive failures, set a fatal error with ``retryable=True`` so the gateway runner rebuilds the adapter. """ interval = self._liveness_interval_seconds threshold = self._liveness_failure_threshold failures = 0 while self._running: try: await asyncio.sleep(interval) except asyncio.CancelledError: return if not self._running: return try: await self._chatto_client.get_viewer() failures = 0 # Refresh presence to keep showing as online try: await self.set_presence("online") except Exception: logger.debug("Chatto: presence refresh failed", exc_info=True) continue except asyncio.CancelledError: return except Exception as e: reason = str(e) failures += 1 logger.warning( "Chatto: liveness probe failed (%s, %d/%d)", reason, failures, threshold, ) if failures < threshold: continue # Threshold exceeded ā€” force reconnect logger.error( "Chatto: liveness probe failed %d times consecutively; forcing reconnect", failures, ) self._set_fatal_error( "chatto_liveness_failed", f"Chatto REST API liveness check failed: {reason}", retryable=True, ) # Cancel the WebSocket to trigger reconnect if self._ws_task and not self._ws_task.done(): self._ws_task.cancel() return async def _join_room(self, room_id: str) -> None: """Join a room if not already a member.""" try: await self._chatto_client.join_room(room_id=room_id) logger.debug("Chatto: joined room %s (%s)", room_id, self._room_names.get(room_id, room_id)) except ChattoError as e: if "permission_denied" in str(e).lower() or "403" in str(e): logger.debug("Chatto: already a member of %s or cannot join", room_id) else: logger.debug("Chatto: join room %s failed: %s", room_id, e) 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: from chattolib._pb.chatto.api.v1 import room_service_pb2 from chattolib._transport import pb_to_dict resp = await self._chatto_client.services.rooms.get_room_events( room_service_pb2.GetRoomEventsRequest(room_id=room_id), headers=self._chatto_client._headers(), ) data = pb_to_dict(resp) events = data.get("page", {}).get("events", []) for ev in events: ev_id = str(ev.get("id", "")) if ev_id: self._mark_seen(room_id, ev_id) logger.debug("Chatto: seeded room %s with %d events", room_id, len(events)) except Exception as e: logger.debug("Chatto: get room events failed for %s: %s", room_id, e) def _mark_seen(self, room_id: str, event_id: str) -> None: seen = self._seen.setdefault(room_id, OrderedDict()) seen[event_id] = None while len(seen) > _SEEN_CAP: seen.popitem(last=False) def _is_seen(self, room_id: str, event_id: str) -> bool: return event_id in self._seen.get(room_id, {}) # ------------------------------------------------------------------ # # WebSocket Realtime Transport # ------------------------------------------------------------------ # def _websocket_url(self) -> str: """Build the WebSocket URL from the base HTTP URL.""" parsed = urlsplit(self._base_url.strip()) scheme = {"http": "ws", "https": "wss"}.get(parsed.scheme, parsed.scheme) if scheme not in ("ws", "wss") or not parsed.netloc: raise ValueError(f"Chatto URL must use http(s) or ws(s), got {parsed.scheme}") path = parsed.path.rstrip("/") + _WS_PATH return urlunsplit((scheme, parsed.netloc, path, parsed.query, "")) async def _start_chattolib_realtime(self) -> bool: """Start realtime connection using chattolib's stream_events.""" self._ws_ready = asyncio.Event() self._ws_task = asyncio.create_task(self._chattolib_event_loop()) try: await asyncio.wait_for(self._ws_ready.wait(), timeout=_WS_AUTH_TIMEOUT + 10) except (asyncio.TimeoutError, TimeoutError): logger.warning("Chatto: chattolib realtime did not connect in time") self._ws_active = False if self._ws_task and not self._ws_task.done(): self._ws_task.cancel() try: await self._ws_task except asyncio.CancelledError: pass self._ws_task = None return False return True 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. """ client = self._chatto_client backoff = _WS_RECONNECT_INITIAL_BACKOFF try: while True: try: logger.info("Chatto: starting chattolib event stream with %d rooms", len(self._watch_room_ids)) # Start streaming events async for event in stream_events( client, resume_cursor=self._resume_cursor, retained_room_ids=self._watch_room_ids, ): # Signal that we're connected and ready if not self._ws_ready.is_set(): self._ws_active = True self._ws_ready.set() backoff = _WS_RECONNECT_INITIAL_BACKOFF # Handle different event kinds if event.kind == "projection_event": # Convert chattolib RealtimeEvent to our format # event.payload is the RealtimeProjectionEvent protobuf try: # Extract the raw bytes for compatibility with existing handler # For now, we'll use the existing _handle_projection_event # which expects bytes. We need to convert. # # Actually, let's create a new handler that works with # chattolib's event objects directly. await self._handle_chattolib_projection_event(event) except Exception as e: logger.warning("Chatto: failed to handle projection event: %s", e) elif event.kind == "caught_up": # Update resume cursor if hasattr(event.payload, 'cursor'): self._resume_cursor = event.payload.cursor logger.debug("Chatto: caught_up received, cursor=%s", self._resume_cursor or "(none)") elif event.kind in ("message_posted", "mention_notification", "new_direct_message_notification", "user_joined_room", "room_created", "user_left_room", "message_edited", "message_retracted", "session_terminated"): # Transient events - convert to envelope format for existing handler await self._handle_chattolib_transient_event(event) elif event.kind in ("heartbeat", "pong", "subscribed"): # Ignore these logger.debug("Chatto: %s event received", event.kind) elif event.kind == "error": logger.warning("Chatto: server error event: %s", event.payload) elif event.kind == "close": logger.info("Chatto: server sent close event") raise ConnectionError("Server closed connection") else: logger.debug("Chatto: unknown event kind: %s", event.kind) except ChattoRealtimeCloseError as e: logger.warning("Chatto: realtime closed by server: %s (reconnect=%s)", e.message, e.reconnect) if e.reconnect: self._ws_active = False await asyncio.sleep(backoff) backoff = min(backoff * 2, _WS_RECONNECT_MAX_BACKOFF) continue raise except ChattoRealtimeError as e: logger.warning("Chatto: realtime error: %s (fatal=%s)", e.message, e.fatal) if e.fatal: raise except (ConnectionError, asyncio.CancelledError): raise except Exception as e: self._ws_active = False logger.warning("Chatto: event stream error: %s, retrying in %.1fs", e, backoff) await asyncio.sleep(backoff) backoff = min(backoff * 2, _WS_RECONNECT_MAX_BACKOFF) finally: self._ws_active = False async def _handle_chattolib_projection_event(self, event: RealtimeEvent) -> None: """Handle a chattolib RealtimeEvent with kind='projection_event'. This is a wrapper that converts chattolib's event to the format expected by _handle_projection_event. """ try: # event.payload is a RealtimeProjectionEvent protobuf message # We need to convert it to the dict format that _handle_projection_event expects from chattolib._transport import pb_to_dict pe_dict = pb_to_dict(event.payload) # Extract operations from the projection event operations = [] if "operations" in pe_dict: for op_pb in event.payload.operations: op_dict = pb_to_dict(op_pb) operations.append(op_dict) # Build the event dict in the format expected by _handle_projection_event event_data = { "id": pe_dict.get("id", ""), "created_at": pe_dict.get("createdAt", ""), "actor_id": pe_dict.get("actorId", ""), "resume_cursor": pe_dict.get("resumeCursor", ""), "operations": operations, } await self._handle_projection_event_from_dict(event_data) except Exception as e: logger.warning("Chatto: failed to convert chattolib projection event: %s", e) async def _handle_projection_event_from_dict(self, event: dict) -> None: """Handle a projection event from a dict (used by chattolib wrapper).""" # Update resume cursor if provided cursor = event.get("resume_cursor") or event.get("resumeCursor") if cursor: self._resume_cursor = cursor operations = event.get("operations", []) for op in operations: op_type = op.get("type", "") if op_type == "room_timeline_event_upsert": await self._handle_timeline_event_upsert_from_dict(op) async def _handle_timeline_event_upsert_from_dict(self, op: dict) -> None: """Handle a room_timeline_event_upsert operation from dict.""" room_id = op.get("room_id", "") or op.get("roomId", "") event_data = op.get("event", {}) if not event_data: return ev_id = str(event_data.get("id", "")) if not ev_id: return # De-dupe if self._is_seen(room_id, ev_id): return self._mark_seen(room_id, ev_id) # Only handle messagePosted events posted = event_data.get("messagePosted", {}) if not posted: return msg = posted.get("message", {}) if not msg: return await self._dispatch_message(msg, room_id) async def _handle_chattolib_transient_event(self, event: RealtimeEvent) -> None: """Handle a chattolib RealtimeEvent with transient event kinds. This converts chattolib's event to the envelope format expected by _handle_transient_event. """ try: from chattolib._transport import pb_to_dict # Build envelope dict based on event kind envelope = { "type": event.kind, "actorId": event.actor_id or "", "id": event.id or "", "data": pb_to_dict(event.payload) if event.payload else {}, } # Convert data field names to match expected format data = envelope["data"] if event.kind == "message_posted": data["roomId"] = data.get("roomId", "") data["messageEventId"] = data.get("eventId", data.get("id", "")) data["threadRootEventId"] = data.get("threadRootEventId", "") elif event.kind == "mention_notification": data["roomId"] = data.get("roomId", "") data["eventId"] = data.get("eventId", data.get("id", "")) elif event.kind == "new_direct_message_notification": data["roomId"] = data.get("roomId", "") data["eventId"] = data.get("eventId", data.get("id", "")) elif event.kind in ("user_joined_room", "room_created", "user_left_room"): data["roomId"] = data.get("roomId", data.get("room_id", "")) data["actorId"] = data.get("actorId", data.get("actor_id", "")) elif event.kind == "message_edited": data["roomId"] = data.get("roomId", "") data["messageEventId"] = data.get("eventId", data.get("id", "")) elif event.kind == "message_retracted": data["roomId"] = data.get("roomId", "") data["messageEventId"] = data.get("messageEventId", data.get("eventId", "")) data["reason"] = data.get("reason", "") elif event.kind == "session_terminated": data["reason"] = data.get("reason", "") await self._handle_transient_event_from_dict(envelope) except Exception as e: logger.warning("Chatto: failed to handle chattolib transient event: %s", e) async def _handle_transient_event_from_dict(self, envelope: dict) -> None: """Handle a transient event from a dict (used by chattolib wrapper).""" event_type = envelope.get("type", "unknown") event_data = envelope.get("data", {}) actor_id = envelope.get("actorId", "") if event_type == "message_posted": room_id = event_data.get("roomId", "") event_id = event_data.get("messageEventId", "") thread_root = event_data.get("threadRootEventId", "") logger.info("Chatto WS: message_posted in room %s, event %s (thread=%s)", room_id, event_id, thread_root or "none") if room_id and event_id and not self._is_seen(room_id, event_id): await self._fetch_and_dispatch_event(room_id, event_id, thread_root) elif event_type == "mention_notification": room_id = event_data.get("roomId", "") event_id = event_data.get("eventId", "") logger.info("Chatto WS: mention notification in room %s for event %s", room_id, event_id) if room_id and event_id and not self._is_seen(room_id, event_id): await self._fetch_and_dispatch_event(room_id, event_id) elif event_type == "new_direct_message_notification": room_id = event_data.get("roomId", "") event_id = event_data.get("eventId", "") logger.info("Chatto WS: new DM notification in room %s for event %s", room_id, event_id) if room_id and event_id and not self._is_seen(room_id, event_id): await self._fetch_and_dispatch_event(room_id, event_id) elif event_type == "user_joined_room": room_id = event_data.get("roomId", "") actor_id = envelope.get("actorId", "") logger.info("Chatto WS: user_joined_room room=%s actor=%s", room_id, actor_id) if room_id and room_id not in self._watch_room_ids: await self._refresh_rooms() elif event_type == "room_created": room_id = event_data.get("roomId", "") logger.info("Chatto WS: room_created room=%s", room_id) if room_id and room_id not in self._watch_room_ids: await self._refresh_rooms() elif event_type == "user_left_room": room_id = event_data.get("roomId", "") actor_id = envelope.get("actorId", "") logger.info("Chatto WS: user_left_room room=%s actor=%s", room_id, actor_id) if room_id and actor_id == self._user_id and room_id in self._watch_room_ids: self._watch_room_ids.remove(room_id) logger.info("Chatto WS: stopped watching room %s (we left)", room_id) elif event_type == "message_edited": room_id = event_data.get("roomId", "") event_id = event_data.get("messageEventId", "") logger.info("Chatto WS: message_edited in room %s, event %s", room_id, event_id) elif event_type == "message_retracted": room_id = event_data.get("roomId", "") event_id = event_data.get("messageEventId", "") reason = event_data.get("reason", "") logger.info("Chatto WS: message_retracted in room %s, event %s (reason=%s)", room_id, event_id, reason or "none") if room_id and event_id: self._mark_seen(room_id, event_id) elif event_type == "session_terminated": reason = event_data.get("reason", "") logger.warning("Chatto WS: session terminated by server (reason=%s) ā€” forcing reconnect", reason or "none") # We can't close _ws_ref here since we're using chattolib # The reconnect will happen automatically in _chattolib_event_loop else: logger.debug("Chatto WS: unknown transient event type: %s", event_type) # Update resume cursor if provided cursor = event.get("resume_cursor") if cursor: self._resume_cursor = cursor operations = event.get("operations", []) for op in operations: if op.get("type") == "room_timeline_event_upsert": await self._handle_timeline_event_upsert(op) # Other operation types (room_upsert, room_member_upsert, etc.) are # not relevant to message delivery ā€” ignore them. async def _handle_timeline_event_upsert(self, op: dict) -> None: """Handle a room_timeline_event_upsert operation.""" room_id = op.get("room_id", "") event = op.get("event", {}) if not event: return ev_id = str(event.get("id", "")) if not ev_id: return # De-dupe: skip events we've already seen if self._is_seen(room_id, ev_id): return self._mark_seen(room_id, ev_id) # Only handle messagePosted events posted = event.get("messagePosted") if not posted: return msg = posted.get("message", {}) if not msg: return await self._dispatch_message(msg, room_id) async def _handle_transient_event(self, data: bytes) -> None: """Handle a transient RealtimeEventEnvelope (message_posted, mentions, DMs). These are signal-only events ā€” they contain room_id and event_id but NOT the message body. We fetch the actual message via REST as a fallback. """ if not data: return try: envelope = _decode_event_envelope(data) except (ValueError, IndexError) as e: logger.warning("Chatto WS: failed to decode transient event: %s", e) return event_type = envelope.get("type", "unknown") event_data = envelope.get("data", {}) if event_type == "message_posted": room_id = event_data.get("roomId", "") event_id = event_data.get("messageEventId", "") thread_root = event_data.get("threadRootEventId", "") logger.info("Chatto WS: message_posted in room %s, event %s (thread=%s)", room_id, event_id, thread_root or "none") if room_id and event_id and not self._is_seen(room_id, event_id): await self._fetch_and_dispatch_event(room_id, event_id, thread_root) elif event_type == "mention_notification": room_id = event_data.get("roomId", "") event_id = event_data.get("eventId", "") logger.info("Chatto WS: mention notification in room %s for event %s", room_id, event_id) if room_id and event_id and not self._is_seen(room_id, event_id): await self._fetch_and_dispatch_event(room_id, event_id) elif event_type == "new_direct_message_notification": room_id = event_data.get("roomId", "") event_id = event_data.get("eventId", "") logger.info("Chatto WS: new DM notification in room %s for event %s", room_id, event_id) if room_id and event_id and not self._is_seen(room_id, event_id): await self._fetch_and_dispatch_event(room_id, event_id) elif event_type == "user_joined_room": room_id = event_data.get("roomId", "") actor_id = envelope.get("actorId", "") logger.info("Chatto WS: user_joined_room room=%s actor=%s", room_id, actor_id) # If WE joined a room (or someone else joined and we should watch it), # refresh room list and resubscribe if room_id and room_id not in self._watch_room_ids: await self._refresh_rooms() elif event_type == "room_created": room_id = event_data.get("roomId", "") logger.info("Chatto WS: room_created room=%s", room_id) # A new room was created ā€” check if we should join/watch it if room_id and room_id not in self._watch_room_ids: await self._refresh_rooms() elif event_type == "user_left_room": room_id = event_data.get("roomId", "") actor_id = envelope.get("actorId", "") logger.info("Chatto WS: user_left_room room=%s actor=%s", room_id, actor_id) # If WE left a room, stop watching it if room_id and actor_id == self._user_id and room_id in self._watch_room_ids: self._watch_room_ids.remove(room_id) logger.info("Chatto WS: stopped watching room %s (we left)", room_id) elif event_type == "message_edited": room_id = event_data.get("roomId", "") event_id = event_data.get("messageEventId", "") logger.info("Chatto WS: message_edited in room %s, event %s", room_id, event_id) # Log edit ā€” could re-fetch for context if needed in the future elif event_type == "message_retracted": room_id = event_data.get("roomId", "") event_id = event_data.get("messageEventId", "") reason = event_data.get("reason", "") logger.info("Chatto WS: message_retracted in room %s, event %s (reason=%s)", room_id, event_id, reason or "none") # Mark the message as seen so we don't try to dispatch it later if room_id and event_id: self._mark_seen(room_id, event_id) elif event_type == "session_terminated": reason = event_data.get("reason", "") logger.warning("Chatto WS: session terminated by server (reason=%s) ā€” forcing reconnect", reason or "none") # Close the websocket to trigger reconnect with backoff if self._ws_ref: try: await self._ws_ref.close() except Exception: pass else: logger.debug("Chatto WS: unknown transient event type: %s", event_type) async def _fetch_and_dispatch_event(self, room_id: str, event_id: str, thread_root_event_id: str = "") -> None: """Fetch a single event by ID via REST and dispatch it. Used as a fallback when the projection_event for a transient notification (mention/DM) hasn't arrived yet. When thread_root_event_id is set, fetches from the thread timeline instead of the room timeline. """ self._mark_seen(room_id, event_id) try: if thread_root_event_id: # Thread reply ā€” use GetThreadEvents from chattolib._pb.chatto.api.v1 import thread_service_pb2 from chattolib._transport import pb_to_dict resp = await self._chatto_client.services.threads.get_thread_events( thread_service_pb2.GetThreadEventsRequest( room_id=room_id, thread_root_event_id=thread_root_event_id, ), headers=self._chatto_client._headers(), ) data = pb_to_dict(resp) else: # Regular room message ā€” use GetRoomEvents from chattolib._pb.chatto.api.v1 import room_service_pb2 from chattolib._transport import pb_to_dict resp = await self._chatto_client.services.rooms.get_room_events( room_service_pb2.GetRoomEventsRequest(room_id=room_id), headers=self._chatto_client._headers(), ) data = pb_to_dict(resp) events = data.get("page", {}).get("events", []) for ev in events: ev_id = str(ev.get("id", "")) if ev_id == event_id: posted = ev.get("messagePosted") if posted: msg = posted.get("message", {}) if msg: # Ensure thread info is set on the message so # _dispatch_message can extract the thread root ID. if thread_root_event_id and not msg.get("thread"): msg["thread"] = {"threadRootEventId": thread_root_event_id} logger.info("Chatto WS: dispatching event %s via REST fallback (thread=%s)", event_id, thread_root_event_id or "none") await self._dispatch_message(msg, room_id) return logger.warning("Chatto WS: event %s not found in room %s events (thread=%s)", event_id, room_id, thread_root_event_id or "none") except Exception as e: logger.warning("Chatto WS: REST fallback fetch failed for event %s: %s", event_id, e) async def _refresh_rooms(self) -> None: """Re-list rooms and subscribe to any new ones dynamically. Called when a room_created or user_joined_room event arrives. This avoids requiring a gateway restart to pick up new rooms. With chattolib, we just need to update our room list and let the event stream handle subscription automatically. """ try: rooms_list = await self._chatto_client.list_rooms() new_room_ids = [] for room_with_state in rooms_list: rid = str(room_with_state.room.id) name = str(room_with_state.room.name) kind = str(room_with_state.room.kind.value) if room_with_state.room.kind else "" is_member = room_with_state.viewer_state.is_member if room_with_state.viewer_state else False self._room_names[rid] = name self._room_kinds[rid] = kind # If we're a member and not already watching, add it if is_member and rid not in self._watch_room_ids: new_room_ids.append(rid) if not new_room_ids: return logger.info("Chatto WS: discovered %d new room(s): %s", len(new_room_ids), new_room_ids) # Join and seed each new room for rid in new_room_ids: if self._room_kinds.get(rid) != "ROOM_KIND_DM": await self._join_room(rid) self._seen[rid] = OrderedDict() await self._seed_room(rid) self._watch_room_ids.append(rid) # With chattolib, the event stream handles subscription automatically logger.info("Chatto WS: updated watch list with %d room(s)", len(self._watch_room_ids)) except Exception: logger.warning("Chatto WS: _refresh_rooms failed", exc_info=True) async def _dispatch_message(self, msg: dict, room_id: str) -> None: """Build a MessageEvent and hand it to the base class handler. This method is identical to the polling version ā€” it receives a message dict (decoded from protobuf) and dispatches it through the standard Hermes message pipeline. """ if not self._message_handler: return actor_id = str(msg.get("actorId", "")) # Skip our own messages if actor_id == self._user_id: return # Best-effort: cache the sender's display name for richer message context if actor_id and actor_id not in self._user_cache: try: await self.get_user(actor_id) except Exception: logger.debug("Chatto: get_user(%s) failed during dispatch", actor_id, exc_info=True) body = str(msg.get("body", "")) if not body: return msg_id = str(msg.get("id", "")) chat_type = "dm" if self._room_kinds.get(room_id) == "ROOM_KIND_DM" else "group" # Mention detection is_dm = chat_type == "dm" mentioned = False if self._user_login: mentioned = f"@{self._user_login}" in body if self._user_display: mentioned = mentioned or f"@{self._user_display}" in body if self._require_mention and not is_dm and not mentioned: # Allow free-response rooms (like Discord's free_response_channels) if room_id not in self._free_response_channels: return # For DMs, always respond. For rooms with require_mention, only respond when mentioned. # Strip the mention from the text for the agent text = body if mentioned and not is_dm: # Remove mention prefix if present if self._user_login and text.startswith(f"@{self._user_login}"): text = text[len(f"@{self._user_login}"):].lstrip() elif self._user_display and text.startswith(f"@{self._user_display}"): text = text[len(f"@{self._user_display}"):].lstrip() # Resolve user display name from actorLogin or actorDisplayName user_name = str(msg.get("actorLogin", "")) or str(msg.get("actorDisplayName", actor_id)) thread_id = None thread_info = msg.get("thread", {}) if thread_info and str(thread_info.get("threadRootEventId", "")) != msg_id: thread_id = str(thread_info.get("threadRootEventId", "")) # Hermes SDK: Propagate thread context if thread_id is set try: # Hermes injects the SDK into the plugin context as self.sdk propagate_context_to_thread = self.sdk.thread_context.propagate_context_to_thread propagate_context_to_thread(thread_id) except AttributeError: logger.debug("Hermes SDK thread_context not available in plugin context") except Exception as e: logger.warning("Failed to propagate thread context: %s", e, exc_info=True) source = self.build_source( chat_id=room_id, chat_name=self._room_names.get(room_id, room_id), chat_type=chat_type, user_id=actor_id, user_name=user_name, thread_id=thread_id, ) created_at_str = str(msg.get("createdAt", "")) try: timestamp = datetime.fromisoformat(created_at_str.replace("Z", "+00:00")) if created_at_str else datetime.now() except (ValueError, TypeError): timestamp = datetime.now() event = MessageEvent( text=text, message_type=MessageType.TEXT, source=source, message_id=msg_id, timestamp=timestamp, raw_message=msg, ) await self.handle_message(event) # ------------------------------------------------------------------ # # Read state & notification dismissal (best-effort, Chatto-unique) # ------------------------------------------------------------------ # try: await self.mark_room_as_read(room_id) except Exception: logger.debug("Chatto: mark_room_as_read failed for %s", room_id, exc_info=True) try: await self.dismiss_all_notifications() except Exception: logger.debug("Chatto: dismiss_all_notifications failed", exc_info=True) # ------------------------------------------------------------------ # # Sending (REST ā€” unchanged from polling version) # ------------------------------------------------------------------ # async def send( self, chat_id: str, content: str, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = 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. """ if not content: return SendResult(success=False, error="Empty message") formatted = self.format_message(content) if hasattr(self, "format_message") else content chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH) # Thread support ā€” resolve thread_id once # DM rooms don't support threads, so skip threading for DMs thread_id = (metadata or {}).get("thread_id") # Only use reply_to as thread_id if auto_thread is enabled. # When auto_thread=false, responses go directly in the room # without threading under the incoming message. auto_thread_env = os.getenv("CHATTO_AUTO_THREAD", "").strip().lower() auto_thread_setting = auto_thread_env in ("true", "1", "yes") if auto_thread_env else True if reply_to and auto_thread_setting: # reply_to might be the incoming message ID. If we already have # thread_id from metadata, keep it (it's the thread root). # Only use reply_to as thread_id if we don't already have one. if not thread_id: thread_id = reply_to # Check if this is a DM room ā€” DMs don't support threads room_kind = self._room_kinds.get(str(chat_id), "") is_dm = room_kind == "ROOM_KIND_DM" or room_kind == "dm" if is_dm: thread_id = None # 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. auto_thread_enabled = os.getenv("CHATTO_AUTO_THREAD", "").strip().lower() if auto_thread_enabled: auto_thread_enabled = auto_thread_enabled in ("true", "1", "yes") else: auto_thread_enabled = True # default: enabled use_auto_thread = auto_thread_enabled and not thread_id and not is_dm message_ids: List[str] = [] last_resp: Optional[dict] = None last_error: Optional[str] = None retryable = False for i, chunk in enumerate(chunks): try: msg_obj = await self._chatto_client.post_message( room_id=str(chat_id), body=chunk, thread_root_event_id=str(thread_id) if thread_id else "", ) msg_id = str(msg_obj.id) last_resp = {"message": {"id": msg_id}} except ChattoError as e: last_error = str(e) retryable = True break except Exception as e: last_error = str(e) retryable = True break if msg_id: self._mark_seen(str(chat_id), msg_id) message_ids.append(msg_id) self._our_message_ids.add(msg_id) # If we sent a message WITHOUT a thread_id, this message could # become a thread root if someone replies to it if not thread_id: self._our_thread_roots.add(msg_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_id if last_error and not message_ids: return SendResult(success=False, error=last_error, retryable=retryable) first_id = message_ids[0] if message_ids else "" # ------------------------------------------------------------------ # # Thread following (best-effort, Chatto-unique) # ------------------------------------------------------------------ # if thread_id and message_ids: try: await self._follow_thread(str(chat_id), str(thread_id)) except Exception: logger.debug("Chatto: _follow_thread failed for room=%s thread=%s", chat_id, thread_id, exc_info=True) return SendResult(success=True, message_id=first_id, raw_response=last_resp) 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. """ if chat_id in self._typing_tasks: return # already running async def _typing_loop() -> None: try: while True: try: await self._chatto_client.update_typing_indicator(room_id=str(chat_id)) except asyncio.CancelledError: return except Exception: pass 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.""" task = self._typing_tasks.pop(chat_id, None) if task: task.cancel() try: await task except (asyncio.CancelledError, Exception): pass async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: """Get information about a chat/room.""" name = self._room_names.get(chat_id, chat_id) kind = self._room_kinds.get(chat_id, "") chat_type = "dm" if kind == "ROOM_KIND_DM" else "group" return { "name": name, "type": chat_type, } # ------------------------------------------------------------------ # # 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 = _EMOJI_TO_SHORTCODE.get(emoji) if shortcode: return shortcode # Already a shortcode like "thumbsup" ā€” return as-is return emoji async def send_reaction(self, chat_id: str, message_id: str, emoji: str) -> bool: """Add a reaction to a message via MessageService/AddReaction.""" shortcode = self._emoji_to_shortcode(emoji) try: result = await self._chatto_client.add_reaction( room_id=str(chat_id), message_event_id=str(message_id), emoji=shortcode, ) return result except ChattoError as e: logger.debug("Chatto: AddReaction failed: %s", e) return False except Exception as e: logger.debug("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.""" shortcode = self._emoji_to_shortcode(emoji) try: result = await self._chatto_client.remove_reaction( room_id=str(chat_id), message_event_id=str(message_id), emoji=shortcode, ) return result except ChattoError as e: logger.debug("Chatto: RemoveReaction failed: %s", e) return False except Exception as e: logger.debug("Chatto: RemoveReaction error: %s", e) return False # ------------------------------------------------------------------ # # Read state management (Chatto-unique) # ------------------------------------------------------------------ # async def mark_room_as_read(self, room_id: str) -> bool: """Mark a room as read via RoomService/MarkRoomAsRead.""" try: await self._chatto_client.mark_room_as_read(room_id=str(room_id)) return True except ChattoError as e: logger.debug("Chatto: MarkRoomAsRead failed: %s", e) return False except Exception as e: logger.debug("Chatto: MarkRoomAsRead error: %s", e) return False async def mark_thread_as_read(self, room_id: str, thread_root_event_id: str) -> bool: """Mark a thread as read via ThreadService/MarkThreadAsRead.""" try: await self._chatto_client.mark_thread_as_read( room_id=str(room_id), thread_root_event_id=str(thread_root_event_id) ) return True except ChattoError as e: logger.debug("Chatto: MarkThreadAsRead failed: %s", e) return False except Exception as e: logger.debug("Chatto: MarkThreadAsRead error: %s", e) return False # ------------------------------------------------------------------ # # DM initiation (Chatto-unique) # ------------------------------------------------------------------ # async def start_dm(self, user_id: str) -> Optional[str]: """Start a direct message with a user via RoomService/StartDM. Returns the room ID on success, or None on failure. """ if not user_id: return None try: room = await self._chatto_client.start_dm(participant_ids=[str(user_id)]) rid = str(room.id) if room else "" if rid: self._room_names[rid] = self._room_names.get(rid, "") self._room_kinds[rid] = "ROOM_KIND_DM" return rid logger.debug("Chatto: StartDM returned no room id") return None 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 # ------------------------------------------------------------------ # # Thread following (Chatto-unique) # ------------------------------------------------------------------ # async def _follow_thread(self, room_id: str, thread_root_event_id: str) -> None: """Best-effort: follow a thread via ThreadService/FollowThread.""" try: await self._chatto_client.follow_thread( room_id=str(room_id), thread_root_event_id=str(thread_root_event_id) ) except ChattoError as e: logger.debug("Chatto: FollowThread failed: %s", e) except Exception as e: logger.debug("Chatto: FollowThread error: %s", e) # ------------------------------------------------------------------ # # Room creation (Chatto-unique) # ------------------------------------------------------------------ # async def create_room( self, name: str, description: str = "", group_id: str = "", universal: bool = True, ) -> Optional[str]: """Create an ad-hoc room via RoomService/CreateRoom. Returns the room ID on success, or None on failure. """ try: room = await self._chatto_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] = name self._room_kinds[rid] = "ROOM_KIND_GROUP" 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 # ------------------------------------------------------------------ # # Notification dismissal (Chatto-unique) # ------------------------------------------------------------------ # async def dismiss_all_notifications(self) -> bool: """Dismiss all notifications via NotificationService/DismissAllNotifications.""" try: await self._chatto_client.dismiss_all_notifications() return True except ChattoError as e: logger.debug("Chatto: DismissAllNotifications failed: %s", e) return False except Exception as e: logger.debug("Chatto: DismissAllNotifications error: %s", e) return False async def dismiss_notification(self, notification_id: str) -> bool: """Dismiss a single notification via NotificationService/DismissNotification.""" try: await self._chatto_client.dismiss_notification(notification_id=str(notification_id)) return True except ChattoError as e: logger.debug("Chatto: DismissNotification failed: %s", e) return False except Exception as e: logger.debug("Chatto: DismissNotification error: %s", e) return False # ------------------------------------------------------------------ # # Message editing and deletion # ------------------------------------------------------------------ # async def edit_message( self, chat_id: str, message_id: str, new_content: str, metadata: Optional[Dict[str, Any]] = None, ) -> bool: """Edit a previously sent message via MessageService/UpdateMessage.""" try: await self._chatto_client.update_message( room_id=str(chat_id), event_id=str(message_id), body=new_content, ) return True except ChattoError as e: logger.debug("Chatto: UpdateMessage failed: %s", e) return False except Exception as e: logger.debug("Chatto: UpdateMessage error: %s", e) return False async def delete_message( self, chat_id: str, message_id: str, metadata: Optional[Dict[str, Any]] = None, ) -> bool: """Delete a previously sent message via MessageService/DeleteMessage.""" try: result = await self._chatto_client.delete_message( room_id=str(chat_id), event_id=str(message_id), ) return result except ChattoError as e: logger.debug("Chatto: DeleteMessage failed: %s", e) return False except Exception as e: logger.debug("Chatto: DeleteMessage error: %s", e) return False # ------------------------------------------------------------------ # # Processing lifecycle hooks (reactions-based, like Discord) # ------------------------------------------------------------------ # def _reactions_enabled(self) -> bool: """Check if processing reactions are enabled.""" return os.getenv("CHATTO_REACTIONS", "true").lower() not in {"false", "0", "no"} def _event_room_and_message_id(self, event: MessageEvent) -> Tuple[str, str]: """Extract room_id and message_id from a MessageEvent.""" chat_id = "" message_id = str(event.message_id or "") source = event.source if source: chat_id = str(getattr(source, "chat_id", "") or "") # Fallback: try raw_message dict if not chat_id or not message_id: raw = event.raw_message if isinstance(raw, dict): if not chat_id: chat_id = str(raw.get("roomId", "") or "") if not message_id: message_id = str(raw.get("id", "") or "") return chat_id, message_id async def on_processing_start(self, event: MessageEvent) -> None: """Add an šŸ‘€ (eyes) reaction to the incoming message.""" if not self._reactions_enabled(): return chat_id, message_id = self._event_room_and_message_id(event) if not chat_id or not message_id: return await self.send_reaction(chat_id, message_id, "šŸ‘€") async def on_processing_complete( self, event: MessageEvent, outcome: ProcessingOutcome ) -> None: """Swap the šŸ‘€ reaction for āœ… (success) or āŒ (failure).""" if not self._reactions_enabled(): 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.send_reaction(chat_id, message_id, "āœ…") elif outcome == ProcessingOutcome.FAILURE: await self.send_reaction(chat_id, message_id, "āŒ") # ------------------------------------------------------------------ # # Asset upload (chunked) # ------------------------------------------------------------------ # async def _upload_asset(self, room_id: str, file_path: str) -> Optional[str]: """Upload a file via the chunked AssetUploadService. Returns the asset ID on success, or None on failure. """ try: with open(file_path, "rb") as f: file_data = f.read() 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() try: # Step 1: Create upload session upload = await self._chatto_client.create_upload( room_id=room_id, filename=file_name, size=file_size, sha256=sha256_hash, content_type=mime_type, ) upload_id = str(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 + _UPLOAD_CHUNK_SIZE] chunk_sha256 = hashlib.sha256(chunk).hexdigest() await self._chatto_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 self._chatto_client.complete_upload(upload_id=upload_id) if not asset: logger.error("Chatto: CompleteUpload returned no asset") return None asset_id = str(asset.id) 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 send_image_file( self, chat_id: str, file_path: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Send a local image file via the chunked upload API.""" # Validate the path is safe safe_path = self.validate_media_delivery_path(file_path) if not safe_path: logger.warning("Chatto: send_image_file ā€” unsafe path %s", file_path) text = "āš ļø Couldn't deliver the image attachment." if caption: text = f"{caption}\n{text}" 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: # Fallback to a notice text = "āš ļø Couldn't deliver the image attachment." if caption: text = f"{caption}\n{text}" return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata) thread_id = (metadata or {}).get("thread_id") if reply_to: thread_id = reply_to try: msg = await self._chatto_client.post_message( room_id=str(chat_id), body=caption or "", attachment_asset_ids=[asset_id], thread_root_event_id=str(thread_id) if thread_id else "", ) msg_id = str(msg.id) if msg_id: self._mark_seen(str(chat_id), msg_id) return SendResult(success=True, message_id=msg_id, raw_response=msg) except ChattoError as e: 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_image( self, chat_id: str, image_url: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: """Send an image to a Chatto room. Tries to download the image from the URL and upload it as a native attachment. Falls back to sending the URL as a link (Chatto renders link previews) if the download fails. """ # Try downloading and uploading as attachment try: import tempfile import urllib.request as _urllib_request # Download to a temp file parsed = urlsplit(image_url) url_path = parsed.path ext = os.path.splitext(url_path)[1] or ".png" tmp_fd, tmp_path = tempfile.mkstemp(suffix=ext, prefix="chatto_img_") try: os.close(tmp_fd) req = _urllib_request.Request(image_url, headers={"User-Agent": "Hermes/1.0"}) ctx = _ssl_context() with _urllib_request.urlopen(req, timeout=_HTTP_TIMEOUT, context=ctx) as resp: with open(tmp_path, "wb") as f: f.write(resp.read()) # Upload as attachment result = await self.send_image_file( chat_id, tmp_path, caption=caption, reply_to=reply_to, metadata=metadata, ) if result.success: return result finally: try: os.unlink(tmp_path) except OSError: pass except Exception as e: logger.debug("Chatto: send_image download/upload failed, falling back to link: %s", e) # Fallback: send as link (Chatto renders link previews) text = image_url if caption: text = f"{caption}\n{image_url}" return await self.send(chat_id, text, reply_to=reply_to, metadata=metadata) # ------------------------------------------------------------------ # # Platform properties # ------------------------------------------------------------------ # @property def platform_name(self) -> str: return "chatto" @property def supports_markdown(self) -> bool: return True @property def supports_reactions(self) -> bool: return True # ------------------------------------------------------------------ # # Member directory ā€” user lookup and mention resolution (Chatto-unique) # ------------------------------------------------------------------ # async def list_users(self) -> list: """List all server members via UserService/ListUsers. Returns a list of user dicts. Each dict typically contains ``id``, ``login``, and ``displayName`` keys. """ try: members, _ = await self._chatto_client.list_users() users = [] # Cache all returned users and convert to dict format for member in members: if member and member.user: user_dict = { "id": str(member.user.id), "login": str(member.user.login), "displayName": str(member.user.display_name or ""), } uid = user_dict["id"] if uid: self._user_cache[uid] = user_dict users.append(user_dict) return users except ChattoError as e: logger.debug("Chatto: ListUsers failed: %s", e) return [] except Exception as e: logger.debug("Chatto: ListUsers error: %s", e) return [] async def get_user(self, user_id: str) -> Optional[dict]: """Get a single user by ID via UserService/GetUser. Returns the user dict (containing ``id``, ``login``, ``displayName``) or ``None`` on failure. Results are cached in ``self._user_cache``. """ if not user_id: return None # Return cached entry if available if user_id in self._user_cache: return self._user_cache[user_id] try: user_obj = await self._chatto_client.get_user(user_id=str(user_id)) if user_obj and user_obj.user: user_dict = { "id": str(user_obj.user.id), "login": str(user_obj.user.login), "displayName": str(user_obj.user.display_name or ""), } uid = user_dict["id"] if uid: self._user_cache[uid] = user_dict return user_dict return None except ChattoError as e: logger.debug("Chatto: GetUser failed: %s", e) return None except Exception as e: logger.debug("Chatto: GetUser error: %s", e) return None async def batch_get_users(self, user_ids: list) -> list: """Batch-fetch multiple users via UserService/BatchGetUsers. Returns a list of user dicts. Cached entries are reused and only uncached IDs are fetched from the server. """ if not user_ids: return [] # Separate cached from uncached cached: list = [] uncached_ids: list = [] for uid in user_ids: uid_str = str(uid) if uid_str in self._user_cache: cached.append(self._user_cache[uid_str]) else: uncached_ids.append(uid_str) if not uncached_ids: return cached try: members = await self._chatto_client.batch_get_users(user_ids=uncached_ids) fetched = [] for member in members: if member and member.user: user_dict = { "id": str(member.user.id), "login": str(member.user.login), "displayName": str(member.user.display_name or ""), } uid = user_dict["id"] if uid: self._user_cache[uid] = user_dict fetched.append(user_dict) return cached + fetched except ChattoError as e: logger.debug("Chatto: BatchGetUsers failed: %s", e) return cached except Exception as e: logger.debug("Chatto: BatchGetUsers error: %s", e) return cached # ------------------------------------------------------------------ # # Presence broadcasting (Chatto-unique) # ------------------------------------------------------------------ # async def set_presence(self, status: str) -> bool: """Update the bot's presence status via MyAccountService/UpdatePresence. Accepts string values ``"online"``, ``"away"``, ``"dnd"`` (or ``"do_not_disturb"``) and maps them to Chatto's PresenceStatus enum. Returns ``True`` on success. """ # Map string status to chattolib PresenceStatus enum status_map = { "online": PresenceStatus.ONLINE, "away": PresenceStatus.AWAY, "dnd": PresenceStatus.DO_NOT_DISTURB, "do_not_disturb": PresenceStatus.DO_NOT_DISTURB, } status_lower = status.lower().strip() presence_status = status_map.get(status_lower) if presence_status is None: logger.warning("Chatto: unknown presence status %r", status) return False try: await self._chatto_client.update_presence(status=presence_status) logger.debug("Chatto: presence set to %s", status_lower) return True except ChattoError as e: logger.debug("Chatto: UpdatePresence failed: %s", e) return False except Exception as e: logger.debug("Chatto: UpdatePresence error: %s", e) return False # ------------------------------------------------------------------ # # Custom status messages (Chatto-unique) # ------------------------------------------------------------------ # async def set_custom_status(self, text: str) -> bool: """Set a custom status message via MyAccountService/UpdateCustomStatus. The status text is a plain string (max ~100 chars). Useful for indicating long-running operations, e.g. ``"Processing..."``. Returns ``True`` on success. """ if not text: return False # Truncate to a reasonable length status_text = text.strip()[:100] if not status_text: return False try: await self._chatto_client.update_custom_status(emoji="", text=status_text) logger.debug("Chatto: custom status set to %r", status_text) return True except ChattoError as e: logger.debug("Chatto: UpdateCustomStatus failed: %s", e) return False except Exception as e: logger.debug("Chatto: UpdateCustomStatus error: %s", e) return False async def clear_custom_status(self) -> bool: """Clear the custom status message via MyAccountService/DeleteCustomStatus. Returns ``True`` on success. """ try: await self._chatto_client.delete_custom_status() logger.debug("Chatto: custom status cleared") return True except ChattoError as e: logger.debug("Chatto: DeleteCustomStatus failed: %s", e) return False except Exception as e: logger.debug("Chatto: DeleteCustomStatus error: %s", e) return False @property def supports_threads(self) -> bool: return True # --------------------------------------------------------------------------- # # Plugin registration # --------------------------------------------------------------------------- # def check_requirements() -> bool: """Check if Chatto is configured.""" return bool( os.getenv("CHATTO_URL", "").strip() and os.getenv("CHATTO_LOGIN", "").strip() and os.getenv("CHATTO_PASSWORD", "").strip() ) def validate_config(config) -> bool: """Validate that the platform config has enough info to connect.""" extra = getattr(config, "extra", {}) or {} url = os.getenv("CHATTO_URL") or str(extra.get("url", "")) login = os.getenv("CHATTO_LOGIN", "").strip() password = os.getenv("CHATTO_PASSWORD", "").strip() return bool(url and login and password) def is_connected(config) -> bool: """Check whether Chatto is configured.""" return validate_config(config) def _apply_yaml_config(yaml_cfg: dict, chatto_cfg: dict) -> Optional[dict]: """Translate config.yaml chatto.extra keys into CHATTO_* env vars.""" extra = chatto_cfg.get("extra") if isinstance(chatto_cfg.get("extra"), dict) else {} mapping = { "url": "CHATTO_URL", "home_channel": "CHATTO_HOME_CHANNEL", "require_mention": "CHATTO_REQUIRE_MENTION", "free_response_channels": "CHATTO_FREE_RESPONSE_CHANNELS", "auto_thread": "CHATTO_AUTO_THREAD", } for yaml_key, env_key in mapping.items(): val = extra.get(yaml_key) if val is not None and not os.getenv(env_key): if isinstance(val, bool): os.environ[env_key] = str(val).lower() elif isinstance(val, list): os.environ[env_key] = ",".join(str(v) for v in val) else: os.environ[env_key] = str(val) channels = extra.get("channels") if isinstance(channels, list) and not os.getenv("CHATTO_CHANNELS"): os.environ["CHATTO_CHANNELS"] = ",".join(str(c) for c in channels) allowed = extra.get("allowed_users") if isinstance(allowed, list) and not os.getenv("CHATTO_ALLOWED_USERS"): os.environ["CHATTO_ALLOWED_USERS"] = ",".join(str(u) for u in allowed) if "allow_all_users" in extra and not os.getenv("CHATTO_ALLOW_ALL_USERS"): os.environ["CHATTO_ALLOW_ALL_USERS"] = str(extra["allow_all_users"]).lower() # Return nothing to merge ā€” all config flows through env return None def _env_enablement() -> Optional[dict]: """Seed PlatformConfig.extra from env vars for env-only setups.""" url = os.getenv("CHATTO_URL", "").strip() if not url: return None extra = {"url": url} home = os.getenv("CHATTO_HOME_CHANNEL", "").strip() if home: extra["home_channel"] = home channels = os.getenv("CHATTO_CHANNELS", "").strip() if channels: extra["channels"] = [c.strip() for c in channels.split(",") if c.strip()] rm = os.getenv("CHATTO_REQUIRE_MENTION", "").strip().lower() if rm: extra["require_mention"] = rm in ("true", "1", "yes") home_dict = {"home_channel": home} if home else None return {"extra": extra, "home_channel": home_dict} async def _standalone_send( base_url: str, login: str, password: str, room_id: str, content: str, thread_id: Optional[str] = None, ) -> dict: """Out-of-process send for cron delivery (no live adapter needed).""" try: # Create a temporary client for standalone sending from chattolib import ChattoClient client = ChattoClient(base_url=base_url) await client.login(login=login, password=password) msg = await client.post_message( room_id=room_id, body=content, thread_root_event_id=thread_id or "", ) return {"success": True, "response": {"id": str(msg.id)}, "message_id": str(msg.id)} except Exception as e: return {"success": False, "error": str(e)} def interactive_setup() -> None: """Interactive setup wizard for Chatto.""" from hermes_cli.gateway import prompt_env, set_env_var url = prompt_env("Chatto server URL (e.g. https://chat.example.com):") if url: set_env_var("CHATTO_URL", url) login = prompt_env("Chatto login (username):") if login: set_env_var("CHATTO_LOGIN", login) password = prompt_env("Chatto password:", password=True) if password: set_env_var("CHATTO_PASSWORD", password) channels = prompt_env("Room IDs to watch (comma-separated, or empty for all):") if channels: set_env_var("CHATTO_CHANNELS", channels) home = prompt_env("Home room ID for notifications (or empty):") if home: set_env_var("CHATTO_HOME_CHANNEL", home) allow_all = prompt_env("Allow all users? (true/false):") if allow_all: set_env_var("CHATTO_ALLOW_ALL_USERS", allow_all) print("\nāœ“ Chatto configured. Restart the gateway to activate.") def register(ctx) -> None: """Plugin entry point ā€” called by the Hermes plugin system.""" ctx.register_platform( name="chatto", label="Chatto", adapter_factory=lambda cfg: ChattoAdapter(cfg), check_fn=check_requirements, validate_config=validate_config, is_connected=is_connected, required_env=["CHATTO_URL", "CHATTO_LOGIN", "CHATTO_PASSWORD"], install_hint="Requires a Chatto server. See https://docs.chatto.run", setup_fn=interactive_setup, apply_yaml_config_fn=_apply_yaml_config, cron_deliver_env_var="CHATTO_HOME_CHANNEL", standalone_sender_fn=_standalone_send, allowed_users_env="CHATTO_ALLOWED_USERS", allow_all_env="CHATTO_ALLOW_ALL_USERS", max_message_length=_MAX_MESSAGE_LENGTH, emoji="šŸ’¬", allow_update_command=True, pii_safe=False, platform_hint=( "You are chatting in Chatto (a self-hosted team chat server). " "Markdown IS supported. Users address you by @-mentioning your name " "in rooms; direct messages reach you without a mention. " "Keep responses conversational." ), )