hermes-chatto-plugin/adapter.py

"""
Chatto Platform Adapter for Hermes Agent.

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


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

"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import mimetypes
import os
from collections import OrderedDict
from datetime import datetime
import time
from typing import Any, Dict, List, Optional, Tuple, cast
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, 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,
    )
    from vendor.chattolib.exceptions import (
        ChattoAuthError,
        ChattoError,
    )
    from vendor.chattolib.realtime import (
        ChattoRealtimeError,
        ChattoRealtimeCloseError,
        RealtimeEvent,
        stream_events,
    )
    from vendor.chattolib.types import (
        PresenceStatus,
    )
    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 (
        ChattoConfiguration, ChattoConstants,
    )






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."""

    
    _SPLIT_THRESHOLD = 9900
    splits_long_messages = True
    supports_code_blocks: bool = True

    def __init__(self, pconfig: PlatformConfig, **kwargs):
        """Signature needs to be compatible with BasePlatformAdapter.__init__ """
        super().__init__(config=pconfig, platform=Platform(ChattoConstants.PLATFORM_ID))
        # "extra" has been pre-processed by Hermes-Framework to be a dict of extra config values insode PlatformConfig.

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

        # ------ 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] = {}

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

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

    async def _get_chatto_client(self: ChattoAdapter) -> Optional[ChattoClient]:
        """Get or create a ChattoClient instance."""

        if self._chatto_client is not None:
            return self._chatto_client

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

            try:
                assert(self.chatto_config.login.value) # now we can assume, _login is available.

                client = await self._open_client(
                    base_url=self.chatto_config.base_url.value,
                    login=self.chatto_config.base_url.value,
                    password=self.chatto_config.password.value,
                    token=self.chatto_config.token.value,
                )
                self._chatto_client = client
                self._token = client.token
                logger.info("Chatto: logged in as %s via chattolib", self.chatto_config.login.value)
                return client
            except ChattoAuthError as e:
                logger.error("Chatto: authentication failed: %s", e)
                return None
            except (ChattoError, ValueError) as e:
                logger.error("Chatto: failed to create client: %s", e)
                return None

    async def _require_client(self) -> ChattoClient:
        """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 client

    async def _ensure_token(self) -> bool:
        """Ensure we have a logged-in Chatto client and token."""
        if self.chatto_config.token.value and self._chatto_client is not None:
            return True

        client = await self._get_chatto_client()
        return client is not None

    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,
    ) -> ChattoClient:
        """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:
        """Connect to Chatto and start the realtime event stream."""

        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 = me.id
            self._user_login = me.login
            self._user_display = 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.chatto_config.channels_list.value:
                if rid in self.chatto_config.channels_list.value 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.channels_list.value:
            watch = list(self.chatto_config.channels_list.value)
        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.chatto_config.base_url.value,
            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(PresenceStatus.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(PresenceStatus.OFFLINE)
        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

    # ------------------------------------------------------------------ #
    # 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(PresenceStatus.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:
            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) > ChattoConstants.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
    # ------------------------------------------------------------------ #

    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=ChattoConstants.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 = ChattoConstants.WS_RECONNECT_INITIAL_BACKOFF
    
        logger.info("Chatto: starting chattolib event stream with %d rooms", len(self._watch_room_ids))
        while True:
            try:
                
                # 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 = ChattoConstants.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, ChattoConstants.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 (ChattoError, 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, ChattoConstants.WS_RECONNECT_MAX_BACKOFF)

            time.sleep(1)


    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:            
            # 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.value 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_list.value:
                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, ChattoConstants.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.value:
            # 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.
        use_auto_thread = self.chatto_config.auto_thread.value 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 = ChattoConstants.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 _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.chatto_config.reactions.value:
            return
        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            return
        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.chatto_config.reactions.value:
            return
        chat_id, message_id = self._event_room_and_message_id(event)
        if not chat_id or not message_id:
            return
        # Remove the processing eyes reaction
        await self.remove_reaction(chat_id, message_id, "👀")
        # Add the outcome reaction
        if outcome == ProcessingOutcome.SUCCESS:
            await self.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 + ChattoConstants.UPLOAD_CHUNK_SIZE]
                chunk_sha256 = hashlib.sha256(chunk).hexdigest()
                await client.upload_chunk(
                    upload_id=upload_id,
                    offset=offset,
                    content=chunk,
                    chunk_sha256=chunk_sha256,
                )
                offset += len(chunk)

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

            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. Do not change signature."""
        # 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=ChattoConstants.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)




    # ------------------------------------------------------------------ #
    # 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, Exception) as e:
            logger.debug("Chatto: ListUsers failed: %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, presence_status: PresenceStatus) -> bool:
        """Update the bot's presence status via MyAccountService/UpdatePresence.
        Using Chattolibs PresenceStatus Enum.
        Returns ``True`` on success.
        """
        client = await self._require_client()
        await client.update_presence(status=presence_status)


        try:
            assert self._chatto_client
            pstatus = await self._chatto_client.update_presence(status=presence_status)
            logger.debug("Chatto: presence set to %s", presence_status)
            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


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

async def hermes_standalone_sender_fn(
    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.
    """

    chatto_config: ChattoConfiguration = ChattoConfiguration(pconfig=pconfig)
    
    # Create a temporary client for standalone sending
    client: ChattoClient

    if not chatto_config.base_url.value or (not (chatto_config.login.value or not chatto_config.password.value) or not chatto_config.token.value):
        return SendResult(success=False, error="Chatto: base URL or credentials missing")
    
    try: 
        if chatto_config.token.value:
            client = ChattoClient(base_url=chatto_config.base_url.value, token=chatto_config.token.value)
        else:
            if chatto_config.login.value and chatto_config.password.value:
                client = await ChattoClient.login(
                    base_url=chatto_config.base_url.value, login=chatto_config.login.value, password=chatto_config.password.value,
                )
    except (Exception, ValueError) as exc:
        return SendResult(success=False, error=f"Chatto login failed: {exc}")
    finally:
        logger.debug("Chatto standalone client: {client}")

    try:
        kwargs: Dict[str, Any] = {}
        if chatto_config.auto_thread.value 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 as exc:
            logger.error("Chatto standalone: error closing short-lived client. Perhaps already closed. {exc}")