""" 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 import threading from collections import OrderedDict from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple, cast from urllib.parse import urlsplit, urlunsplit from plugins.plugin_utils import lazy_singleton import utils logger = logging.getLogger(__name__) from gateway.platforms.base import ( BasePlatformAdapter, SendResult, MessageEvent, MessageType, ProcessingOutcome, ) from gateway.config import Platform, PlatformConfig # Chattolib imports (vendored) # Using vendored chattolib from vendor/chattolib/ # See vendor_chattolib.sh for how to update the vendored copy try: # Try vendored chattolib first from vendor.chattolib.client import ( ChattoClient, ChattoError, ChattoAuthError, ) from vendor.chattolib.exceptions import ( ChattoConnectError, ) from vendor.chattolib.realtime import ( ChattoRealtimeError, ChattoRealtimeCloseError, RealtimeConnection, RealtimeEvent, ServerHello, stream_events, ) from vendor.chattolib.types import ( RoomKind, PresenceStatus, RoomWithViewerState, User, Message, ) from vendor.chattolib._pb.chatto.api.v1 import ( rooms_pb2 as room_service_pb2, threads_pb2 as thread_service_pb2 ) from vendor.chattolib._transport import pb_to_dict except ImportError as e: logger.error("Chatto: failed to import vendored chattolib: %s", e) from .platform_config import ChattoConfig class AsyncSingletonSlot: """Thread-safe async singleton helper (extends SingletonSlot pattern for async factories).""" def __init__(self): self._instance = None self._lock = threading.Lock() self._future = None async def get(self, factory): """Get or create instance using async factory. Thread-safe with double-checked locking.""" if self._instance is not None: return self._instance with self._lock: if self._instance is None: if self._future is None: self._future = asyncio.ensure_future(factory()) return await self._future return self._instance def reset(self): """Reset the singleton instance.""" with self._lock: self._instance = None self._future = None # --------------------------------------------------------------------------- # # 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 # HTTP timeout used for outbound URL fetches _HTTP_TIMEOUT = 30 # 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 def _decode_event_envelope(data: bytes) -> dict: """Decode a transient event envelope bytes into a dict. Tries protobuf parsing via the vendored realtime pb, falls back to JSON. """ try: from vendor.chattolib._pb.chatto.realtime.v1 import realtime_pb2 frame = cast(Any, realtime_pb2).RealtimeServerFrame() frame.ParseFromString(data) if frame.WhichOneof("frame") == "event": ev = frame.event try: return pb_to_dict(ev) except Exception: # Fall through to manual conversion out = {} # best-effort: copy simple fields if hasattr(ev, "room_id"): out["roomId"] = getattr(ev, "room_id") return out except Exception: pass try: import json return json.loads(data.decode("utf-8")) if data else {} except Exception: return {} # --------------------------------------------------------------------------- # # 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 _chatto_client: ChattoClient def __init__(self, config: PlatformConfig, **kwargs): """Signature needs to be compatible with BasePlatformAdapter.__init__ """ # "extra" has been pre-processed by Hermes-Framework to be a dict of extra config values chatto_config: ChattoConfig = ChattoConfig(config.extra) super().__init__(config=config, platform=Platform(chatto_config._PLATFORM)) # --- Configuration from our configuration data class with some logic --- self.chatto_config = chatto_config # ------ State ------- # SDK runtime handle (injected by Hermes); annotate for Pylance self.sdk: Any = getattr(self, "sdk", None) # --- Runtime state --- self._user_id: 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: ChattoAdapter) -> Optional[ChattoClient]: """Get or create a ChattoClient instance using thread-safe async singleton.""" # Fast path: return cached client if available if self._chatto_client is not None: return self._chatto_client if not self.chatto_config._base_url or not self.chatto_config._login or not self.chatto_config._password: logger.error("Chatto: missing configuration (URL, login, or password)") return None try: client = await self._chatto_client.get( lambda: ChattoClient.login( self.chatto_config._login, self.chatto_config._password, base_url=self.chatto_config._base_url, ) ) self._chatto_client = client # Cache for direct access logger.info("Chatto: logged in as %s via chattolib", self._login) return client except ChattoAuthError as e: logger.error("Chatto: authentication failed: %s", e) return None except Exception as e: logger.error("Chatto: failed to create client: %s", e) return None async def _require_client(self) -> Any: """Return a ChattoClient or raise RuntimeError if unavailable. Use this helper when the caller expects a client to exist and wants a single canonical failure path. Methods that prefer a soft-fail can catch RuntimeError and return gracefully. """ client = await self._get_chatto_client() if client is None: raise RuntimeError("Chatto client unavailable") return cast(Any, client) async def _client(self) -> Optional[ChattoClient]: """Helper to get the chattolib client. Thread-safe via AsyncSingletonSlot.""" return await self._get_chatto_client() async def _ensure_token(self) -> bool: """Login via chattolib.""" if self.chatto_config._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 return await self._ensure_token() # ------------------------------------------------------------------ # # Connection # ------------------------------------------------------------------ # async def _open_client( *, base_url: str, login: str, password: str, token: Optional[str] = None, ) -> Any: """Return a connected ``ChattoClient`` using token or login/password.""" if token: return ChattoClient(token=token, base_url=base_url) return await ChattoClient.login(login, password, base_url=base_url) async def connect(self, *, is_reconnect: bool = False) -> bool: """Log into Chatto and start the realtime event stream. Codemancer""" if not self.chatto_config._base_url: logger.error("Chatto: CHATTO_BASE_URL not configured") return False if not self.chatto_config._login and not self.chatto_config._password: logger.error( "Chatto: CHATTO_LOGIN/PASSWORD is not set" ) return False self._closing = False try: self._client = await self._open_client( base_url=self.chatto_config._base_url, login=self.chatto_config._login, password=self.chatto_config._password, token=self.chatto_config._token, ) except Exception as exc: logger.error("Chatto: failed to construct client: %s", exc) return False # Best-effort presence broadcast (non-fatal on failure). try: await self._client.update_presence(_PRESENCE_ONLINE) except Exception: logger.debug("Chatto: update_presence not supported / failed (non-fatal)") self._stream_task = asyncio.create_task( self._run_event_stream(), name="chatto-event-stream", ) self._mark_connected() return True async def connect_too_long(self, *, is_reconnect: bool = False) -> bool: """Login, discover rooms, start WebSocket realtime connection.""" if not await self._ensure_token(): return False try: client = await self._require_client() except RuntimeError: self._set_fatal_error("connect_failed", "Chatto client not available", retryable=True) return False # 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) return False # Discover rooms try: rooms_list = await client.list_rooms() rooms = [] for room_with_state in rooms_list: # Defensive checks: vendored types may be partially populated if not room_with_state: continue room_obj = getattr(room_with_state, "room", None) if not room_obj: continue entry = { "room": { "id": str(getattr(room_obj, "id", "")), "name": str(getattr(room_obj, "name", "")), "kind": str(getattr(getattr(room_obj, "kind", None), "value", "")) if getattr(room_obj, "kind", None) else "", }, "viewerState": { "isMember": getattr(getattr(room_with_state, "viewer_state", None), "is_member", 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) 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.chatto_config._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.chatto_config._channel_ids: watch = list(self.chatto_config._channel_ids) else: watch = all_room_ids if not watch: logger.error("Chatto: no rooms to watch (join a room or set CHATTO_CHANNELS)") 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(): 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 self._chatto_client = None self._client_slot.reset() # ------------------------------------------------------------------ # # 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: try: client = await self._require_client() except RuntimeError: reason = "no_client" failures += 1 logger.warning( "Chatto: liveness probe aborted - no client available (%s, %d/%d)", reason, failures, threshold, ) continue await 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: try: client = await self._require_client() except RuntimeError: logger.debug("Chatto: _join_room aborted - no client available for %s", room_id) return await 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: # These should be available from check_chatto_requirements() try: client = await self._require_client() except RuntimeError: logger.debug("Chatto: _seed_room aborted - no client available for %s", room_id) return resp = await client.services.rooms.get_room_events( cast(Any, room_service_pb2).GetRoomEventsRequest(room_id=room_id), headers=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.chatto_config._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. """ try: client = await self._require_client() except RuntimeError: logger.warning("Chatto: chattolib event loop aborted - no client available") return # Ensure ws_ready is available for synchronization with starter assert self._ws_ready is not None 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): # 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 # pb_to_dict should be available from check_chatto_requirements() 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: # pb_to_dict should be available from check_chatto_requirements() # 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 = envelope.get("resume_cursor") or envelope.get("resumeCursor") if cursor: self._resume_cursor = cursor operations = envelope.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: ws = cast(Any, self._ws_ref) await ws.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: # Ensure we have a client try: client = await self._require_client() except RuntimeError: logger.warning("Chatto WS: REST fallback fetch aborted - no client available for event %s", event_id) return # Fetch events either from the thread timeline or the room timeline if thread_root_event_id: resp = await client.services.threads.get_thread_events( cast(Any, thread_service_pb2).GetThreadEventsRequest( room_id=room_id, thread_root_event_id=thread_root_event_id, ), headers=client._headers(), ) else: resp = await client.services.rooms.get_room_events( cast(Any, room_service_pb2).GetRoomEventsRequest(room_id=room_id), headers=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) # If the REST fallback didn't find the event, refresh known rooms # by delegating to a dedicated helper. This avoids duplicate logic # across transient/projection event handlers. await self._refresh_rooms() 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.chatto_config._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.chatto_config._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) async def _refresh_rooms(self) -> None: """Refresh room list via REST, join and seed any newly discovered rooms.""" try: try: client = await self._require_client() except RuntimeError: logger.warning("Chatto WS: _refresh_rooms aborted - no client available") return rooms_list = await client.list_rooms() new_room_ids: List[str] = [] for room_with_state in rooms_list: if not room_with_state: continue room_obj = getattr(room_with_state, "room", None) if not room_obj: continue rid = str(getattr(room_obj, "id", "")) name = str(getattr(room_obj, "name", "")) kind = str(getattr(getattr(room_obj, "kind", None), "value", "")) if getattr(room_obj, "kind", None) else "" is_member = getattr(getattr(room_with_state, "viewer_state", None), "is_member", False) self._room_names[rid] = name self._room_kinds[rid] = kind 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) 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) 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) # ------------------------------------------------------------------ # # Read state & notification dismissal (best-effort, Chatto-unique) # ------------------------------------------------------------------ # # Best-effort: mark all watched rooms as read (room_id may be undefined here) for _rid in list(self._watch_room_ids): try: await self.mark_room_as_read(_rid) except Exception: logger.debug("Chatto: mark_room_as_read failed for %s", _rid, 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. if reply_to and self.chatto_config._auto_thread: # 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 try: client = await self._require_client() except RuntimeError: 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=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: try: client = await self._require_client() except RuntimeError: return await 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: try: client = await self._require_client() except RuntimeError: return False result = await 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: try: client = await self._require_client() except RuntimeError: return False 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.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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return None room = await client.start_dm(participant_ids=[str(user_id)]) rid = str(getattr(cast(Any, 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: try: client = await self._require_client() except RuntimeError: return await 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: try: client = await self._require_client() except RuntimeError: return None 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] = 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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return False result = await 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: try: client = await self._require_client() except RuntimeError: logger.error("Chatto: upload aborted - no client available") return None # 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, ) upload_id = str(getattr(cast(Any, 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 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 asset_id = str(getattr(cast(Any, 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: try: client = await self._require_client() except RuntimeError: return SendResult(success=False, error="Chatto client not available", retryable=True) msg = await 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"}) try: import ssl ctx = ssl.create_default_context() except Exception: ctx = None 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: try: client = await self._require_client() except RuntimeError: return [] members, _ = await 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: try: client = await self._require_client() except RuntimeError: return None user_obj = await 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: try: client = await self._require_client() except RuntimeError: return cached members = await 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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return False await 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: try: client = await self._require_client() except RuntimeError: return False await 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 and dependencies are available.""" # Then check configuration return bool( os.getenv("CHATTO_URL", "").strip() and os.getenv("CHATTO_LOGIN", "").strip() and os.getenv("CHATTO_PASSWORD", "").strip() ) def validate_config(config: PlatformConfig) -> 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.""" if not isinstance(chatto_cfg, dict): chatto_cfg = {} extra = chatto_cfg.get("extra", {}) or {} if not isinstance(extra, dict): extra = {} 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): env_val: str = str(val).lower() elif isinstance(val, list): env_val = ",".join(str(v) for v in val) else: env_val = str(val) os.environ[env_key] = env_val channels = extra.get("channels") if isinstance(channels, list) and not os.getenv("CHATTO_CHANNELS"): channels_str: str = ",".join(str(c) for c in channels) os.environ["CHATTO_CHANNELS"] = channels_str allowed = extra.get("allowed_users") if isinstance(allowed, list) and not os.getenv("CHATTO_ALLOWED_USERS"): allowed_str: str = ",".join(str(u) for u in allowed) os.environ["CHATTO_ALLOWED_USERS"] = allowed_str if "allow_all_users" in extra and not os.getenv("CHATTO_ALLOW_ALL_USERS"): allow_all_val: str = str(extra["allow_all_users"]).lower() os.environ["CHATTO_ALLOW_ALL_USERS"] = allow_all_val # 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. Returns a dict compatible with the PlatformConfig merge hook (or None when no env-provided values are present). """ extra_dict: dict = {} env_url = os.getenv("CHATTO_URL", "").strip() if env_url: extra_dict["url"] = env_url env_channels = os.getenv("CHATTO_CHANNELS", "").strip() if env_channels: extra_dict["channels"] = [c.strip() for c in env_channels.split(",") if c.strip()] env_require_mention_str = str(utils.is_truthy_value(os.getenv("CHATTO_REQUIRE_MENTION", ""))) extra_dict["require_mention"] = env_require_mention_str env_home_channel = os.getenv("CHATTO_HOME_CHANNEL", "").strip() if env_home_channel: home_dict = {"home_channel": env_home_channel} return {**extra_dict, **home_dict} async def _standalone_send( pconfig, chat_id, message, *, 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. """ pconfig.getattr("extra", {}) chatto_config = ChattoConfig(extra=pconfig.extra) # Create a temporary client for standalone sending client: ChattoClient token = chatto_config._token login = chatto_config._login password = chatto_config._password if not chatto_config._base_url or (not (login or not password) or not token): return SendResult(success=False, error="Chatto: base URL or credentials missing") try: if token: client = ChattoClient(base_url=chatto_config._base_url, token=token) else: client = await ChattoClient.login( base_url=chatto_config._base_url, login=chatto_config._login, password=chatto_config._password, ) except Exception as exc: return SendResult(success=False, error=f"Chatto login failed: {exc}") try: kwargs: Dict[str, Any] = {} if chatto_config._auto_thread and thread_id: kwargs["in_reply_to"] = thread_id if media_files and media_files.get("attachment_asset_ids"): kwargs["attachment_asset_ids"] = list(media_files["attachment_asset_ids"]) try: posted = await client.post_message(chat_id, message, **kwargs) except Exception as exc: return SendResult(success=False, error=str(exc)) return SendResult(success=True, message_id=getattr(posted, "id", "") or None) finally: try: await client.close() except Exception: logger.error("Chatto standalone: error closing short-lived client") await client.login(login=login, password=password) def interactive_setup() -> None: """Interactive setup wizard for Chatto.""" try: from hermes_cli.gateway import prompt_env, set_env_var # type: ignore except Exception: # Fallback simple prompt implementations for environments where # hermes_cli.gateway helpers are not available (tests / limited shells). def prompt_env(prompt_text: str, password: bool = False): if password: import getpass return getpass.getpass(prompt_text) return input(prompt_text + " ") def set_env_var(k: str, v: str) -> None: os.environ[k] = v 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 Chat Server", 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", env_enablement_fn=_env_enablement, 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", auto_thread_env="CHATTO_AUTO_THREAD", 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." ), )